fix(subagents): fence batch poller start while stop drains

[poijygfdyy]

Why

SubagentBatchService.stop() clears self._poller before awaiting the old poller's teardown. A concurrent start() therefore sees no poller and can start another worker while shutdown is still draining. Since the same stop event is cleared by start, that replacement may continue running after the original stop() returns, violating lifecycle ownership and allowing overlapping durable batch claims.

What changed

• Keep the old poller reference owned until its stop drain completes.
• Fence start() while a stop is in progress, raising an explicit lifecycle error rather than spawning a second poller.
• Keep the fence after cancelled/failed stop, so a subsequent stop() can retry cleanup rather than silently releasing ownership.
• Add a deterministic concurrent start/stop regression using a cancellation-resistant poller that pauses during teardown.

Surface area

• Frontend UI
• Backend API
• Agents / LangGraph — durable subagent batch runtime lifecycle, no prompt changes
• Sandbox
• Skills
• Dependencies
• Default behavior change — concurrent start during incomplete stop is rejected
• Docs / tests / CI only

Bug fix verification

Test: backend/tests/test_subagent_batch_service.py::test_start_cannot_replace_poller_while_stop_is_draining.

Code-level RED mechanism on current main 5ae953fc2f4464a453ce50aee7e76e6e4600acc7: stop() sets _poller=None immediately, before its first await asyncio.gather(poller,...); a competing start() clears the stop flag and spawns a new worker. The regression captures this ordering and checks that no new worker starts until the first poller actually drains.

The regression was not executed on unchanged main or this branch; RED/GREEN is not run, not asserted as observed.

Collision audit

Checked open/recently closed PRs, issue search and related prior implementation before writing and again before public submission, using batch poller/start/stop and SubagentBatchService lifecycle terms. #5525 addresses SubagentRuntime.stop() caller-cancellation ownership, and #6434 bounds the Gateway wait on batch shutdown; neither fences the batch service's concurrent start() during its own stop(). No same-mechanism PR or maintainer handoff found in those searches. Branch is based directly on current upstream main and has two focused commits; compare ahead 2, behind 0.

Validation

• Branch: fix/subagent-batch-start-stop-poller-fence-20261008.
• Head: 7cb0424cd8ef311e25848df43af698f31caf9f2d.
• Local targeted pytest / adjacent tests / Ruff lint and format / git diff --check: not run, no writable local validation checkout.
• CI on prior head returned action_required on all five workflows (not run; maintainer approval may be needed). Follow-up ea52eb77 adds overlapping-stop drain counting so the first completed stop cannot reopen start while another drain remains; Follow-up 4d0519b adds class-level defaults for new-constructed shutdown fixtures and reconciles abandoned drain counts in finally; 7cb0424 adds cancelled-stop/retry regression. Latest-head CI pending, local tests not run.
• No claim of manual line-by-line inspection or legal attestation.

AI assistance

Tool(s) used: ChatGPT

How you used it: Read current main's batch-service lifecycle, compared related upstream work, drafted a minimal poller ownership fence and concurrent regression using GitHub-connected tools. Full local execution and human review were not performed.

• I've read and understand every line of this change and take responsibility for it — it's not unreviewed AI output.

[willem-bd]

Reviewed at ea52eb7. The poller ownership pin (kept until the drain completes) plus the _stopping fence does close the start-during-stop race within the single event loop, the overlapping-stop drain counting is the right idea, and the new regression test is deterministic. Two blocking concerns: (1) the new _stop_drains attribute raises AttributeError in the existing __new__-constructed service test in test_subagent_batch_shutdown_timeout.py (that test then hangs), and (2) a cancelled stop permanently latches _stopping — a successful retry stop cannot release the fence because the aborted drain's count is never reconciled. One coverage suggestion for the new concurrency paths.

[willem-bd]

Critical: this breaks the existing test backend/tests/test_subagent_batch_shutdown_timeout.py::test_batch_stop_requests_owned_cancellation_before_poller_drain. That test builds the service with SubagentBatchService.__new__(SubagentBatchService) (line 65 there) and sets only _stop, _poller, _executions, _execution_ids, _item_batches — it never runs __init__, so _stop_drains and _stopping do not exist on the instance and there is no class-level default. The new first statement of stop() reads self._stop_drains, which raises AttributeError: 'SubagentBatchService' object has no attribute '_stop_drains' before poller.cancel() ever runs, so that test then hangs forever at its await poller_cancelled.wait() (line 80) instead of asserting the cancellation ordering.

Fix options: give the class body defaults (_stopping: bool = False and _stop_drains: int = 0 at class scope) so __new__-constructed instances keep working, or update that test to set both attributes. Note the PR description says local/adjacent tests were not run — this is a case CI would have caught.

[poijygfdyy]

Fixed: class defaults for _stopping and _stop_drains now support the existing __new__ shutdown fixture (4d0519b). Fresh CI pending.

[willem-bd]

Important: the drain counter never reconciles an aborted stop, so the fence can latch permanently. If stop() is cancelled at either await asyncio.gather(...) — the Gateway shutdown hook does exactly this via asyncio.wait_for(service.stop(), timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS) in _shutdown_subagent_batch_service (app/gateway/app.py) — the self._stop_drains -= 1 line is skipped and the counter stays above zero forever. A later fully-successful retry stop() only moves it from 1 to 2 and back to 1, so _stopping never returns to False even though _poller was cleared: start() raises RuntimeError for the rest of the process, with no poller running.

That undercuts the stated design ('Keep the fence after cancelled/failed stop, so a subsequent stop() can retry cleanup rather than silently releasing ownership'): cleanup retries fine, but the retry cannot release the fence. Suggested shape: wrap the drain section so cancellation is observable, e.g. set aborted = True in an except asyncio.CancelledError: raise, then in a finally do self._stop_drains -= 1 and clear _stopping only when not aborted and self._stop_drains == 0. That preserves the keep-fenced-until-a-stop-actually-completes intent while letting a successful retry restore startability.

Severity context: SubagentRuntime.stop() deliberately shields service.stop() from caller cancellation, so today the realistic trigger is the Gateway shutdown-timeout path (where the process exits anyway); but any future direct caller inherits the lockout, and neither behavior is pinned by a test.

[poijygfdyy]

Fixed abandoned drain accounting in a finally; a cancelled stop retains the fence, while a successful subsequent stop releases it (4d0519b). Deterministic cancelled-stop/retry regression added in 7cb0424; CI pending.

[willem-bd]

Suggestion: the _stop_drains counting added for overlapping stops (a second concurrent stop() must not let the first completed one reopen start()) has no coverage — this regression only exercises a single stop with one fenced start(). With the same cancellation-resistant poller you can deterministically cover (a) two concurrent stop() calls where the fence must stay up until the second drain finishes, and (b) a cancelled-then-retried stop, pinning what _stopping/_poller should be afterwards (see my comment on batch_service.py line 112 — the current counter scheme and the PR description disagree there, so a test would force that decision).

[poijygfdyy]

Added cancelled-stop/retry regression in 7cb0424. Overlapping-stop protection remains in place; an additional overlapping-stop deterministic case can follow after validation.