bug(agent-server): max_concurrent_runs does not limit native async conversations

Problem

Config.max_concurrent_runs is documented as the maximum number of conversations that can execute agent steps concurrently, but it currently only controls the size of the dedicated ThreadPoolExecutor used by the synchronous conversation.run() fallback.

EventService.run() prefers the native async path for normal LocalConversation instances whose agents override astep():

if has_native_arun:
    await conversation.arun()
else:
    await loop.run_in_executor(self._run_executor, conversation.run)

As a result, OH_MAX_CONCURRENT_RUNS=N does not constrain native async conversations. Multiple standard conversations can all enter arun() concurrently regardless of the configured value. The Agent Server therefore has no effective global admission/concurrency limit for its normal execution path.

This is distinct from #3143 / #3169, which isolated synchronous conversation execution in a dedicated thread pool but did not cover the native async path added later.

Impact

A burst of conversations can exhaust memory and cause the operating system to kill the Agent Server. This is particularly easy to trigger when an automation dispatcher or other client launches many conversations together.

Illustrative measurements from one 8 GiB self-hosted Agent Server using the current SDK:

The same server loaded 83 persisted conversations and used approximately 2.21 GiB RSS before a burst. Conversation memory varies substantially because restoring state rebuilds the view and caches parsed event objects. Active LLM, tool, plugin, and MCP state can add further transient memory.

These figures are illustrative rather than a formal benchmark, but they show why a real execution cap is needed.

Reproduction

1. Start Agent Server with OH_MAX_CONCURRENT_RUNS=1.
2. Create two or more ordinary conversations using an agent that implements native astep().
3. Start all conversations concurrently.
4. Instrument the agent's astep() method with a shared active-counter.
5. Observe that the counter exceeds 1 because each conversation runs through arun() without acquiring the configured executor limit.

A regression test can implement this with a blocking async test agent and assert that the maximum observed active count never exceeds the configured limit.

Expected behavior

max_concurrent_runs should bound concurrent agent execution across both paths:

• native async conversation.arun()
• synchronous conversation.run() fallback

A shared async admission mechanism, such as an asyncio.Semaphore, could guard the complete run operation. The design should also ensure:

• permits are released on exceptions, cancellation, pause, and shutdown;
• queued conversations do not misleadingly appear to be actively executing;
• queueing is bounded or excess work receives explicit backpressure;
• the configured limit has focused tests for both native async and sync agents.

Related work

• #3143 / #3169: dedicated thread pool and configurable synchronous run worker count
• #3153: Agent Server performance, resource usage, and backpressure tracking
• #3140: eager loading of persisted conversations at startup

Start at EventService.run and trace how Config.max_concurrent_runs reaches the native async conversation.arun() path and the synchronous conversation.run() fallback. Add focused regression coverage with a blocking async test agent and sync agents, verifying the configured limit across both paths and release on exceptions, cancellation, pause, and shutdown.