fix(subagents): preserve batch stop cleanup across cancellation

backend/packages/harness/deerflow/subagents/AGENTS.md

@@ -1,4 +1,4 @@
 ### Subagent System (`packages/harness/deerflow/subagents/`)
 
 Result-reader acceptance runs on SQLite and, with TEST_POSTGRES_URI, an isolated
 PostgreSQL schema; both exercise submission, completion, reopen and ToolNode
@@ -32,7 +32,7 @@
 
 `task` preserves the host incarnation, including null/missing/invalid values; never recapture it.
 
-**Direct runtime shutdown**: `SubagentRuntime.stop()` holds its lifecycle lock until service shutdown finishes. Drain without a timeout so owned work cannot detach; propagate the first caller cancellation, chaining any service failure. Cover terminal outcomes and repeated cancellation in `tests/test_subagent_runtime.py`.
+**Direct runtime shutdown**: `SubagentRuntime.stop()` holds its lifecycle lock through an unbounded service drain; propagate first caller cancellation, chaining service failure. Test terminal outcomes/repeated cancellation in `tests/test_subagent_runtime.py`. Batch stops shield/join one cleanup generation. Restart resets its owner only after success; failed cleanup keeps startup fenced.
 
 **Interrupted usage**: After stream cleanup, publish the collector's final
 snapshot on cancellation/timeout; callbacks can precede values chunks.
@@ -56,7 +56,7 @@
 **Upload-state boundary**: Ordinary `task` delegation snapshots a valid parent `ThreadState.uploaded_files` list at dispatch, deep-copies it across the isolated-loop boundary, seeds it into the child's fresh state, and only then makes `list_uploaded_files` eligible for normal tool-policy filtering. An explicit empty list is valid and must be preserved because it means every upload in the thread is historical for this run. Missing or malformed state fails closed with the tool disabled. Durable `batch_task` execution intentionally keeps the tool disabled: delayed and recovered items have no valid parent-run upload boundary, and supporting that case requires a separate persisted-state contract.
 **Knowledge-scope boundary**: `task` and `batch_task` inherit the parent's canonical execution scope, never display data. Batch specs persist it across lease recovery; malformed scope fails. Model input cannot set or broaden it, and shared middleware/retrieval enforce `disabled`.
 **Date context (#4781)**: Every built-in subagent execution registers `SubagentDateContextMiddleware` immediately before `SystemMessageCoalescingMiddleware`. Its one-time `before_agent` hook adds a hidden framework-owned `SystemMessage` containing only `<current_date>` before the first model call; it does not read `AppConfig.memory`, call the memory manager, rewrite the task `HumanMessage`, or inherit the lead agent's frozen-conversation/midnight lifecycle. The coalescer merges that reminder with the subagent's static prompt so strict providers still receive exactly one leading `SystemMessage`. The lead-only `DynamicContextMiddleware` registration and its date, optional-memory, and midnight-update behavior remain unchanged.
-**Execution**: Ordinary and durable-batch native subagents submit coroutines directly to one persistent isolated event loop. Shutdown keeps loop/thread ownership until the worker is confirmed stopped; a bounded join is not proof of termination, and new submissions stay fenced while cleanup is pending. Once the retained worker has exited, the next submission reaps and closes that loop under the shutdown lifecycle lock before creating a replacement. Gateway/embedded startup installs one process-wide async FIFO admission controller (default 3 running, bounded queue). Direct `create_deerflow_agent` callers can instead pass a caller-owned `SubagentRuntime`; reuse the same instance across graphs so its bound `task`, optional batch tools/service, middleware limits, and `SubagentExecutor` all share one controller without reading global YAML. An owned batch service must be started before graph construction and stopped at application shutdown. Waiters hold no scheduler thread, and cancellation/timeout release queue/slot ownership.
+**Execution**: Ordinary and durable-batch native subagents share one persistent isolated event loop. Retain loop/thread ownership until the worker stops; bounded joins do not prove termination. Fence submissions during cleanup. After exit, the next submission reaps/closes the old loop under the shutdown lifecycle lock before replacement. Gateway/embedded startup installs one process-wide async FIFO admission controller (default 3 running, bounded queue). Direct `create_deerflow_agent` callers may pass a caller-owned `SubagentRuntime`; reuse it across graphs so bound `task`, optional batch tools/service, middleware limits and `SubagentExecutor` share one controller without global YAML. Start owned batch services before graph construction; stop them at application shutdown. Waiters hold no scheduler thread; cancellation/timeout release queue/slot ownership.
 **Shared sandbox execution lifecycle** (#5128): every admitted subagent run carries a stable task-derived `sandbox_lease_owner_id` and matching `sandbox_command_scope_id` in its runtime context. Sandbox middleware retains that execution against the lead thread's active provider client, so one child finishing cannot close the sandbox while siblings still run; the final holder performs any pending provider release. A rollback/fork-restored child reusing the parent's live client binds a non-releasing holder: it fences parent cleanup and owns its command scope without requesting a park itself; a parent's earlier park request waits for the child, while a missing inherited client falls through to a normal fresh acquire. On AIO, the command scope selects one explicit persistent shell session per subagent, allowing independent scopes to run concurrently while preserving in-order shell state within one child. Sync sandbox tool bodies offloaded with `asyncio.to_thread` are shielded and drained across repeated cancellation before the outer execution can clean its holder; a cancelled worker can therefore neither re-admit an already-released owner nor run after subagent terminalization. `SubagentExecutor` invokes active LangGraph stream cleanup before releasing its sandbox lease or notifying task stop. Slow cooperative cleanup keeps the result non-terminal and retains its sandbox lease and capacity slot; its warning is diagnostic, not a safe hard-timeout boundary. Middleware performs the normal release, and `SubagentExecutor` repeats it idempotently in `finally` so exceptions, cooperative cancellation, and timeout unwind paths cannot leak a lease or scoped session.
 **Concurrency and total delegation cap**: Ordinary `task` concurrency is resolved once as the minimum of the per-run request, the startup-frozen `subagent_runtime.max_running`, and the schema safety ceiling (1-64), then shared by the lead prompt and `SubagentLimitMiddleware`. Hot reloads must not make either layer advertise more capacity than the already-created process controller; a changed startup-only value takes effect only after restart. The same middleware separately enforces `subagents.max_total_per_run` (default 6, config schema 1-50, runtime override `max_total_subagents` clamped to the same range) against current-run entries in the durable delegation ledger, so a long lead-agent run cannot bypass concurrency limits by launching repeated legal-sized batches at each planning checkpoint, but historical delegations from previous runs in the same thread do not consume the new run's budget. Explicit `batch_task` work does not consume or relax that ordinary-run ledger: its persisted total/live/running limits live under `subagent_batches`. Gateway `run_agent()` and embedded `DeerFlowClient.stream()` both provide a per-invocation `run_id` in runtime context; `DeerFlowClient.stream()` also tags its input `HumanMessage` with that same id so durable-context capture can identify the current request boundary. Gateway resume paths may not append a new `HumanMessage`, so the worker also exposes the pre-run checkpoint's message ids in runtime context; durable-context capture uses that as the current-run boundary and never re-tags older task calls as the resumed run. When no delegation slots remain, task calls are stripped, provider raw tool-call metadata is synced, `finish_reason` is forced to `stop`, and a visible "subagent delegation limit" note is appended so the agent can synthesize already-collected results. Default subagent timeout `subagents.timeout_seconds=1800` (30 min) and built-in `general-purpose` `max_turns=150`.
 **Flow**: `task()` → `SubagentExecutor` → shared slot → polling/SSE. Explicit `batch_task()` → durable rows → lease-based batch service (`subagents/batch_service.py`, Gateway or direct runtime) → same executor/slots → bounded result and owner-scoped API/JSONL. Batch mode is explicit, never inferred from prompt size. Queue rejection/timeout releases the lease without an item attempt; execution failure/expired leases consume one. User cancellation terminalizes nonterminal items and clears leases, fencing stale workers. Resolve background cancellation under `_background_tasks_lock`, then call `Future.cancel()` outside it because callbacks may reacquire the lock. `task_started` includes the effective model. `SubagentTokenCollector` publishes cumulative usage after each LLM response; `task_running` updates cards without double-counting. Terminal ToolMessage metadata (`subagent_model_name`, `subagent_token_usage`) and `subagent.end` preserve model/usage after reload; absent usage stays absent. Resolve one `AppConfig` snapshot (explicit or `get_app_config()`) for assembly, deferred setup, and receipt harvesting; `verification.receipts_enabled=false` holds on both construction paths. Harvest terminal receipts before `try_set_terminal` under the result lock so polling cannot see terminal status without metadata. Each values chunk updates the latest harvest state and publishes receipts before cooperative cancellation. Cancellation/failure scans current ToolMessages; completion uses the bounded ledger snapshot attached to returned assistant text, including max-turn partials. Missing/malformed completed snapshots yield no receipts, preventing omitted receipts from validating citations. Cancellation/timeouts retain the latest streamed evidence.

backend/packages/harness/deerflow/subagents/batch_service.py

@@ -45,6 +45,7 @@ class SubagentBatchService:
     # Also used by focused shutdown tests constructing an instance via __new__.
     _stopping: bool = False
     _stop_drains: int = 0
+    _stop_cleanup_task: asyncio.Task[None] | None = None
 
     def __init__(
         self,
@@ -69,58 +70,72 @@ def __init__(
         self._poller: asyncio.Task[None] | None = None
         self._stopping = False
         self._stop_drains = 0
+        self._stop_cleanup_task: asyncio.Task[None] | None = None
         self._executions: dict[str, asyncio.Task[None]] = {}
         self._execution_ids: dict[str, str] = {}
         self._item_batches: dict[str, str] = {}
 
     async def start(self) -> None:
         if self._stopping:
             raise RuntimeError("cannot start subagent batch poller before stop completes")
+        cleanup_task = getattr(self, "_stop_cleanup_task", None)
+        if cleanup_task is not None:
+            cleanup_task.result()
+            self._stop_cleanup_task = None
         if self._poller is not None:
             return
         self._stop.clear()
         self._poller = asyncio.create_task(self._run(), name="subagent-batch-poller")
 
     async def stop(self) -> None:
-        # Keep poller ownership visible until the entire drain finishes.
-        # Otherwise start() could create a fresh poller during this await.
-        self._stop_drains += 1
-        self._stopping = True
-        self._stop.set()
-        poller = self._poller
+        cleanup_task = getattr(self, "_stop_cleanup_task", None)
+        if cleanup_task is None:
+            # Fence start() before taking the owned-work snapshot. All stop
+            # callers then join this one cleanup generation instead of issuing
+            # competing cancellations and clearing one another's state.
+            self._stopping = True
+            self._stop.set()
+            poller = self._poller
 
-        # Issue every owned-work cancellation before the first await. The
-        # Gateway wraps this stop hook in a deadline; if poller teardown is
-        # slow, cancellation of this coroutine must not prevent native/item
-        # cancellation from being requested.
-        execution_ids = list(self._execution_ids.values())
-        for execution_id in execution_ids:
-            request_cancel_background_task(execution_id)
-        tasks = list(self._executions.values())
-
-        if poller is not None:
-            poller.cancel()
-        for task in tasks:
-            task.cancel()
+            # Issue every owned-work cancellation before the first await. The
+            # Gateway wraps this stop hook in a deadline; if poller teardown is
+            # slow, cancellation of this caller must not prevent native/item
+            # cancellation from being requested.
+            execution_ids = list(self._execution_ids.values())
+            for execution_id in execution_ids:
+                request_cancel_background_task(execution_id)
+            tasks = list(self._executions.values())
 
-        completed = False
-        try:
             if poller is not None:
-                await asyncio.gather(poller, return_exceptions=True)
-            if tasks:
-                await asyncio.gather(*tasks, return_exceptions=True)
-            self._executions.clear()
-            self._execution_ids.clear()
-            self._item_batches.clear()
-            if self._poller is poller:
-                self._poller = None
-            completed = True
-        finally:
-            # A cancelled caller cannot leave a phantom drain count. Keep
-            # start fenced until a subsequent successful stop retries cleanup.
-            self._stop_drains -= 1
-            if completed and self._stop_drains == 0:
-                self._stopping = False
+                poller.cancel()
+            for task in tasks:
+                task.cancel()
+
+            self._stop_drains = 1
+            cleanup_task = asyncio.create_task(
+                self._drain_stop(poller, tasks),
+                name="subagent-batch-stop-cleanup",
+            )
+            self._stop_cleanup_task = cleanup_task
+
+        await asyncio.shield(cleanup_task)
+
+    async def _drain_stop(
+        self,
+        poller: asyncio.Task[None] | None,
+        tasks: list[asyncio.Task[None]],
+    ) -> None:
+        if poller is not None:
+            await asyncio.gather(poller, return_exceptions=True)
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        self._executions.clear()
+        self._execution_ids.clear()
+        self._item_batches.clear()
+        if self._poller is poller:
+            self._poller = None
+        self._stop_drains = 0
+        self._stopping = False
 
     async def _run(self) -> None:
         while not self._stop.is_set():