fix(subagents): drain abandoned capacity waiters before cancellation (#7900)

* fix(subagents): drain abandoned capacity waiters before cancellation

Continue the unmerged queued-waiter cleanup approach from #7571, with deterministic cancellation and timeout controls.

Co-authored-by: 熊小雨 <1178327328@qq.com>

* test(subagents): gate capacity timeout cleanup deterministically

---------

Co-authored-by: linkangjun <331227811+linkangjun@users.noreply.github.com>
Co-authored-by: 熊小雨 <1178327328@qq.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2963,6 +2963,11 @@ for authorization, cancellation, recovery and result semantics.
 
 Ordinary `task` delegation and explicit durable `batch_task` execution share the startup-scoped `subagent_runtime` process capacity. Batch mode keeps large independent item sets in SQL with separate total, live, and running limits, restart recovery, bounded results, and a thread-scoped Web UI panel. The panel pages through bounded previews on demand and opens a selected item’s stored report and captured evidence for review. Citations open the matching saved excerpt without contacting the provider, and an open result refreshes until the item finishes and when the item list observes a later retry. Closed result dialogs do not fetch full details. Bulk stored results remain available through the owner-scoped JSONL export, while internal execution and authorization context never enters owner-facing responses. If the batch worker is later stopped or disabled, threads with persisted batches retain read-only item inspection and JSONL export; execution controls remain disabled until the worker is running again. See `config.example.yaml` and [the implementation contract](docs/plans/2026-08-24-subagent-batch-capacity-implementation.md) for limits and recovery semantics.
 
+Queued requests finish removing their queue entry and returning any transferred
+slot before cancellation propagates. The cleanup waits for the capacity lock
+without its own deadline. Cancellation during timeout cleanup takes precedence
+over the queue timeout.
+
 While durable batch tools are available, a later owner turn can explicitly inspect selected results with `read_batch_result(batch_id, position=0, offset=0, max_chars=4000)`. The reader is restricted to the current user and thread, including in explicitly bound SDK graphs. It returns one window of a JSON document containing the stored report, execution/error state and separate acceptance criteria/verdict, without starting or waiting for work. Concatenate `content` windows before parsing: pass `next_offset` and the returned `revision` as `expected_revision` until `next_offset` is null. A changed item returns `restart_required`; discard prior windows and start at zero. Use `next_position` for a selected next item within `batch_status.total_items`. The maximum window is 8,192 characters; metadata shares the window budget, and JSON response framing adds overhead. `result_truncated` means storage already capped the report; continuation cannot recover discarded content. A succeeded item is not an accepted task, and unchecked criteria remain UNVERIFIED. No automatic result ingestion, waiting loop or UI change is added.
 
 The complete escaped response also fits within 10,000 characters and the bound host's positive `tool_output` externalization/fallback limits, so normal budgeting preserves continuation JSON. Smaller limits can shorten a window; `budget_too_small` means the operator must raise the budget instead of retrying a zero-length page. Limits too small even for that error envelope cannot provide a usable structured response. Explicit SDK graphs with a custom output-budget middleware should supply the same `AppConfig` through `SubagentRuntime`; default graphs use default reader limits.

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

@@ -1,6 +1,11 @@
 ### Subagent System (`packages/harness/deerflow/subagents/`)
 
 Drain cleanup; preserve terminal outcomes on cancel.
+Queued capacity abandonment drains waiter removal and transferred-slot release
+across repeated caller cancellation, just like an acquired slot's final release.
+The drain waits for the capacity lock without its own deadline. Cancellation
+during timeout cleanup takes precedence over the queue timeout.
+Regression: `tests/test_subagent_capacity.py`.
 
 Optional batch dependencies and one-shot follow-up reuse batch claims and AgentRuns.
 See [behavior and ownership](../../../../docs/BATCH_CONTINUATION.md): keep one execution lease and check conversation state before admission; never persist live capabilities.

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

@@ -5,9 +5,10 @@ from __future__ import annotations
 import asyncio
 import threading
 from collections import deque
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Coroutine
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
+from typing import Any
 
 from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig
 
@@ -76,19 +77,23 @@ class SubagentExecutionCapacity:
                 timeout=self._config.queue_timeout_seconds,
             )
         except (TimeoutError, asyncio.CancelledError) as exc:
-            async with self._lock:
-                try:
-                    self._waiters.remove(waiter)
-                except ValueError:
-                    # A release transferred the slot just as the timeout fired.
-                    # If the future completed, this caller owns that transfer and
-                    # must release it before reporting the timeout.
-                    if waiter.done() and not waiter.cancelled():
-                        self._release_locked()
+            # Caller cancellation during timeout cleanup takes precedence over
+            # reporting a queue timeout, once waiter and slot ownership settle.
+            await self._run_cancellation_safe(self._remove_abandoned_waiter(waiter))
             if isinstance(exc, asyncio.CancelledError):
                 raise
             raise SubagentCapacityTimeout(f"Timed out after {self._config.queue_timeout_seconds}s waiting for a subagent execution slot") from exc
 
+    async def _remove_abandoned_waiter(self, waiter: asyncio.Future[None]) -> None:
+        async with self._lock:
+            try:
+                self._waiters.remove(waiter)
+            except ValueError:
+                # A release transferred the slot just as abandonment began.
+                # A completed, non-cancelled future owns the transferred slot.
+                if waiter.done() and not waiter.cancelled():
+                    self._release_locked()
+
     def _release_locked(self) -> None:
         while self._waiters:
             waiter = self._waiters.popleft()
@@ -107,22 +112,32 @@ class SubagentExecutionCapacity:
 
     async def _release_cancellation_safe(self) -> None:
         """Release an acquired slot before propagating repeated cancellation."""
-        release_task = asyncio.create_task(self._release())
+        await self._run_cancellation_safe(self._release())
+
+    @staticmethod
+    async def _run_cancellation_safe(operation: Coroutine[Any, Any, None]) -> None:
+        """Drain owned capacity bookkeeping before propagating cancellation.
+
+        The drain waits for the capacity lock without a separate deadline.
+        Further cancellations cannot interrupt it or replace the first one
+        received during the drain.
+        """
+        cleanup_task = asyncio.create_task(operation)
         cancellation: asyncio.CancelledError | None = None
-        while not release_task.done():
+        while not cleanup_task.done():
             try:
-                await asyncio.shield(release_task)
+                await asyncio.shield(cleanup_task)
             except asyncio.CancelledError as exc:
                 if cancellation is None:
                     cancellation = exc
 
         if cancellation is not None:
             try:
-                release_task.result()
+                cleanup_task.result()
             except Exception as exc:
                 raise cancellation from exc
             raise cancellation
-        release_task.result()
+        cleanup_task.result()
 
     @asynccontextmanager
     async def slot(self, *, wait=True) -> AsyncIterator[None]:

backend/tests/test_subagent_capacity.py

@@ -97,3 +97,67 @@ async def test_repeated_cancellation_cannot_leak_running_slot(monkeypatch):
     async with asyncio.timeout(0.5):
         async with capacity.slot():
             pass
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("owner_released", [False, True])
+@pytest.mark.parametrize("abandonment", ["cancel", "timeout"])
+async def test_abandoned_waiter_drains_cleanup_after_repeated_cancellation(monkeypatch, owner_released, abandonment):
+    """Once cleanup starts, more cancellations cannot orphan a transferred slot."""
+    waiter_ready = asyncio.Event()
+    expire_waiter = asyncio.Event()
+    cleanup_started = asyncio.Event()
+
+    class ObservedLock(asyncio.Lock):
+        async def acquire(self):
+            if self.locked():
+                cleanup_started.set()
+            return await super().acquire()
+
+    capacity = SubagentExecutionCapacity(SubagentRuntimeConfig(max_running=1, max_queued=1))
+    capacity._lock = ObservedLock()
+    real_wait_for = asyncio.wait_for
+
+    async def wait_for_slot(awaitable, timeout):
+        if capacity._waiters and awaitable is capacity._waiters[0]:
+            waiter_ready.set()
+            if abandonment == "timeout":
+                # Exercise asyncio's timeout only once cleanup will contend for the lock.
+                await expire_waiter.wait()
+                timeout = 0
+        return await real_wait_for(awaitable, timeout)
+
+    monkeypatch.setattr(asyncio, "wait_for", wait_for_slot)
+    await capacity._acquire()
+    queued = asyncio.create_task(capacity._acquire())
+    try:
+        await real_wait_for(waiter_ready.wait(), 3)
+        async with capacity._lock:
+            if abandonment == "cancel":
+                if owner_released:
+                    capacity._release_locked()
+                queued.cancel()
+            else:
+                expire_waiter.set()
+            await real_wait_for(cleanup_started.wait(), 3)
+            if owner_released and abandonment == "timeout":
+                # A timed-out waiter's future is cancelled, so release skips it.
+                capacity._release_locked()
+            queued.cancel("cancelled while draining")
+            await asyncio.sleep(0)
+            queued.cancel("cancelled again")
+
+        async with asyncio.timeout(3):
+            with pytest.raises(asyncio.CancelledError, match="cancelled while draining"):
+                await queued
+    finally:
+        if not queued.done():
+            queued.cancel()
+            await asyncio.gather(queued, return_exceptions=True)
+    if not owner_released:
+        await capacity._release()
+    snapshot = capacity.snapshot()
+    assert (snapshot.running, snapshot.queued) == (0, 0)
+    async with asyncio.timeout(0.5):
+        async with capacity.slot():
+            pass