fix(agent-server): bound native async runs with max_concurrent_runs

[Shailendra005]

HUMAN:

Picked this up because a limit that only covers the fallback path isn't really a limit. Kept the change to the admission mechanism itself and left the queueing/status questions for you to weigh in on.

---

AGENT:

Why

max_concurrent_runs is documented as the cap on conversations executing agent steps concurrently, but it only sizes the ThreadPoolExecutor behind the synchronous conversation.run() fallback. EventService.run() prefers native conversation.arun() whenever the agent overrides astep(), and that branch takes no permit, so the path normal conversations actually use is unbounded. Setting OH_MAX_CONCURRENT_RUNS=N has no effect there, which is what #4063 reports along with the memory figures behind it.

Summary

• Added a server-wide asyncio.Semaphore(max_concurrent_runs) created next to the existing thread pool in ConversationService.__aenter__ and injected into each EventService alongside _run_executor.
• _run_and_publish now holds a permit across both branches, so one shared pool bounds the total rather than the two paths each having their own ceiling.
• Used async with so the permit is returned on exception and on cancellation; a leaked permit would be permanent and would throttle the server toward zero over time.
• _run_semaphore is None for a standalone EventService, which keeps current unlimited behaviour for embedders and existing tests.
• Cleared in the shutdown path beside the executor.

async with self._run_semaphore or nullcontext():
    if has_native_arun:
        await conversation.arun()
    else:
        await loop.run_in_executor(self._run_executor, conversation.run)

Issue Number

Closes #4063

How to Test

tests/agent_server/test_event_service.py::TestRunAdmissionLimit implements the shared active-counter check described in the issue: three EventService instances share one Semaphore(1), each is driven through the real await service.run() with an agent whose arun() blocks, and the test asserts the peak observed concurrency never exceeds the limit. A second test asserts the permit is released when a run raises.

$ OPENHANDS_SUPPRESS_BANNER=1 uv run pytest tests/agent_server/test_event_service.py -k RunAdmission -q
2 passed

Reverting just the async with line and re-running gives the reported behaviour:

AssertionError: 3 conversations ran concurrently under a limit of 1

Full suite and checks:

$ OPENHANDS_SUPPRESS_BANNER=1 uv run pytest tests/agent_server -q
1918 passed, 13 deselected

$ uv run ruff check <changed files>      # All checks passed!
$ uv run ruff format --check <changed>   # 3 files already formatted
$ uv run pyright openhands-agent-server/openhands/agent_server/event_service.py
0 errors, 0 warnings, 0 informations

Two notes for anyone extending these tests: has_native_arun inspects type(conversation).arun and type(conversation.agent).astep, so the methods have to be defined on mock subclasses — instance attributes silently fail the check and route the run down the sync branch instead. And _run_task is cleared when a run finishes, so task handles need capturing before the blocked runs are released.

Scope

This covers the first and last bullets of the expected behaviour: a shared admission mechanism across both paths, with permits released on exception and cancellation. Two of your bullets are deliberately not in this PR, since they look like design calls rather than mechanics:

• Queued conversations still report as running. Making a waiting conversation surface as queued rather than executing changes user-visible status semantics, so I would rather agree the intended states with you first.
• No bounded queue or explicit backpressure. Today excess work waits on the semaphore indefinitely. A bounded queue that rejects or defers beyond a threshold is a policy decision, and it interacts with perf: SDK and agent-server performance and correctness issues from profiling investigation #3153.

Happy to take either as a follow-up, or to fold one in here if you would prefer it in a single change.

I also did not add a separate sync-agent test: that path was already bounded by the executor, and it now shares the same permit pool, which the exception test exercises. Glad to add an explicit one if you want the coverage stated.

[all-hands-bot]

🚦 CI is currently failing on this PR's latest commit.

Please fix the failing checks before OpenHands reviews it - this is re-checked automatically once you push a new commit. (A maintainer can also request @all-hands-bot as a reviewer to have it reviewed regardless of CI status.)

This is an automated check - no AI was used to generate this comment.

[Shailendra005]

@VascoSch92 all three addressed in 284bb71.

Pause/interrupt while queued. You were right, and it was a real hole: a queued run has nothing to cancel, so the request was dropped and the run started anyway once the permit arrived. The run now captures _explicit_interrupt_generation before waiting and bails out if it changed while queued, matching the generation checks already used in run() and the rerun path. New test test_pause_while_queued_is_not_overwritten fails without it with a run paused while queued still executed.

Teardown handing out _run_semaphore=None. Also right. Unlike the executor the semaphore owns no resources, so there is nothing to clean up and clearing it only creates that window. I removed the assignment and left a comment saying why, so it does not get "tidied" back in.

Sync path untested. Added test_sync_fallback_runs_share_the_same_limit, which routes through run_in_executor by setting arun = None and asserts the same peak-concurrency bound. It fails without the change with 3 sync runs executed concurrently under a limit of 1.

All three new tests fail on main and pass here. Full suite 1920 passed, ruff and pyright clean.

One thing I did not touch: bounded queueing / explicit backpressure from the issue's expected-behaviour list is still out of scope here, since it looks like a policy decision that interacts with #3153. Happy to take it as a follow-up.