From 14abd9a28eb035f277aa56098caab1b0600c6e8d Mon Sep 17 00:00:00 2001 From: Xiao Hou Date: Mon, 2 Nov 2026 21:36:59 +0800 Subject: [PATCH] fix(subagents): terminalize a background cancel that beat the loop (#7788) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(subagents): terminalize a background cancel that beat the loop request_cancel_background_task() calls Future.cancel() on the future handed back by _submit_to_isolated_loop_in_context(). That call only wins while the coroutine has not been picked up yet; when it does win, run_with_timeout never starts, so no one writes a terminal status and no one opens the cleanup fence. The registry entry then stayed PENDING forever: task_tool polled its full budget for a result that could not arrive, and cleanup_background_task() refused to drop a non-terminal entry. Publish CANCELLED from the future's own completion callback when the coroutine body never ran. A cancellation that did reach a running coroutine keeps its existing owner -- the future reports cancelled() for that case as well, which is why the callback checks a start marker instead of cancelled() alone. * fix(subagents): publish a pre-start cancel from the isolated loop `Future.cancel()` on the future returned by `_submit_to_isolated_loop_in_context()` fires `forget_future` in the calling thread, but the loop keeps owning the coroutine until its first step runs. A cancel that lands in that window read the start marker as clear and published CANCELLED — opening the cleanup fence — while the worker could still take its capacity slot. Hand the decision to the isolated loop and re-check the marker there, which is the only place serialized against that first step. A loop that is gone cannot start the coroutine, so the inline fallback keeps the original guarantee. * test(subagents): accept both terminal owners in the task-created cancel window Task.__init__ queues the first step inside the task factory, so the step can run before the deferred pre-start decision that was handed to the loop. When it does, the coroutine's own CancelledError branch legitimately publishes "Cancelled by user"; the harness had pinned the pre-start message instead. Assert the invariants both owners share -- terminal cancelled, fence opened only after the loop settled the body, cleanup removes the entry -- and let either error message pass. * fix(subagents): refuse to start a cancelled execution before taking a slot A cancel that wins the race against the loop is decided on the loop, one turn after the submission callback queues the wrapper's first step. That turn is not a guarantee: the wrapper could still start, create the execution task, and take a capacity slot after the terminal status and the cleanup fence had already been published from that decision. request_cancel_background_task sets cancel_event before it cancels the future, so the wrapper can treat that event as the startup half of the handshake and decline to create the execution task. The fence then cannot be outrun by a worker that acquires capacity afterwards. * test(subagents): hold the cancel window open with an event barrier The factory blocked the loop for a fixed 200 ms and then resumed on its own. A caller delayed longer than that -- a loaded CI worker is enough -- let execution start before the cancel landed, and `assert not execution_ran.is_set()` failed even though cancellation was handled correctly. Block the factory on an event instead and release it from the test's `finally` after the cancel request and the closed-fence assertions, so the window stays open exactly as long as the test needs it. --------- Co-authored-by: gouyu-hou <213296172+gouyu-hou@users.noreply.github.com> --- .../harness/deerflow/subagents/executor.py | 79 ++++- backend/tests/test_subagent_executor.py | 289 ++++++++++++++++++ 2 files changed, 367 insertions(+), 1 deletion(-) diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index c377f2b5a3..6b726cb9e2 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -772,6 +772,27 @@ def _get_isolated_subagent_loop() -> asyncio.AbstractEventLoop: return _isolated_subagent_loop +def _run_on_isolated_loop_if_live(callback: Callable[[], None]) -> None: + """Run ``callback`` on the isolated loop, or inline when it cannot run. + + Used to re-order a decision against a coroutine that was already submitted + to the loop: only the loop itself is serialized with that coroutine's first + step, so a caller thread cannot tell which of the two landed first. + + A loop that is missing, stopped, or closed can no longer start the + coroutine, so running inline is then equivalent to running on it. + """ + loop = _isolated_subagent_loop + if loop is None or loop.is_closed() or not loop.is_running(): + callback() + return + try: + loop.call_soon_threadsafe(callback) + except RuntimeError: + # Closed between the lookup and the hand-off; nothing can start on it. + callback() + + def _submit_to_isolated_loop_in_context( context: Context, coro_factory: Callable[[], Coroutine[Any, Any, SubagentResult]], @@ -2280,7 +2301,30 @@ class SubagentExecutor: with _background_tasks_lock: _background_tasks[execution_id] = result + # Set once the coroutine body actually runs. A ``concurrent.futures`` + # future reports ``cancelled()`` both when the cancel beat the loop and + # when a running coroutine was cancelled, and only the first case leaves + # the result without an owner able to publish a terminal status. + execution_started = threading.Event() + async def run_with_timeout() -> SubagentResult: + execution_started.set() + # Startup half of the pre-start cancellation handshake. The + # canceller sets ``cancel_event`` before it decides anything, so a + # body that starts after that cancel must decline to create the + # execution task: a cancel published from the caller thread may + # already have opened the cleanup fence, and creating the task here + # would then take a capacity slot the fence promised was gone. + # Reading the event first makes "fence opened before this body + # started" safe instead of a race. + if result.cancel_event.is_set(): + result.try_set_terminal( + SubagentStatus.CANCELLED, + error="Cancelled before the execution started", + tool_receipts=result.snapshot_tool_receipts(), + ) + _mark_background_execution_done(execution_id, result) + return result execution_task = asyncio.create_task( self._aexecute(task, result), name=f"subagent-{self.config.name}-execution", @@ -2358,8 +2402,41 @@ class SubagentExecutor: with _background_tasks_lock: _background_futures[execution_id] = execution_future + def publish_pre_start_cancel() -> None: + # The chained future reports ``cancelled()`` for two different + # outcomes: the cancel beat the loop, or it reached a started + # coroutine that is still winding down. Only the first leaves the + # result without an owner able to publish a terminal status, and a + # task that exists but has not taken its first step still runs its + # body. Deciding on the loop, after that step, keeps the cleanup + # fence closed while the worker can still hold its capacity slot. + # + # A body that starts after this point cannot take a slot either: + # ``result.cancel_event`` is always set before this callback runs, + # and ``run_with_timeout`` declines to create the execution task + # once it observes it. So "fence opened" here means "no worker will + # acquire capacity", not merely "no worker has yet". + if execution_started.is_set(): + return + result.try_set_terminal( + SubagentStatus.CANCELLED, + error="Cancelled before the execution started", + tool_receipts=result.snapshot_tool_receipts(), + ) + _mark_background_execution_done(execution_id, result) + def forget_future(completed_future: Future[SubagentResult]) -> None: - if not completed_future.cancelled(): + if completed_future.cancelled(): + # ``Future.cancel()`` only wins while the loop has not picked the + # coroutine up yet. In that window ``run_with_timeout`` never runs, + # so nothing writes a terminal status and nothing opens the cleanup + # fence: the entry stayed PENDING and ``cleanup_background_task`` + # refused to remove it, while the poller waited out its full budget + # for a result that could never arrive. Publish the cancellation + # here; a cancellation that reached a running coroutine is handled + # by its own ``CancelledError`` branch, which owns the terminal write. + _run_on_isolated_loop_if_live(publish_pre_start_cancel) + else: completed_future.exception() with _background_tasks_lock: _background_futures.pop(execution_id, None) diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index f28450002a..26234513b1 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -4435,6 +4435,295 @@ class TestCooperativeCancellation: """Test that requesting cancellation on a nonexistent task does not raise.""" executor_module.request_cancel_background_task("nonexistent-task") + def test_cancel_before_the_coroutine_starts_leaves_no_pending_entry(self, executor_module, classes, base_config): + """A cancel that wins the race against the loop must still terminalize. + + ``_submit_to_isolated_loop_in_context`` hands back a Future, and while the + loop has not picked the coroutine up ``Future.cancel()`` succeeds — so + ``run_with_timeout`` never runs, nothing writes a terminal status, and + nothing opens the cleanup fence. The entry stayed PENDING: the poller burned + its whole budget waiting, and ``cleanup_background_task`` refused to remove + it because it only drops terminal entries. + + The coroutine is submitted to the real isolated loop, so ``cancelled()`` + here comes from the same chained future ``run_coroutine_threadsafe`` hands + back in production rather than from a stand-in that was never scheduled. + """ + SubagentExecutor = classes["SubagentExecutor"] + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + body_ran = threading.Event() + + async def never_started() -> None: + body_ran.set() + + loop = executor_module._get_isolated_subagent_loop() + hold = threading.Event() + + def hold_loop() -> None: + hold.wait(5) + + # Keep the loop inside a callback so the submission is still queued when + # the cancel lands: the coroutine has been created but cannot have run. + loop.call_soon_threadsafe(hold_loop) + try: + execution_id = executor.execute_async("Task") + result = executor_module._background_tasks[execution_id] + assert result.status.value == "pending" + assert not executor_module._background_futures[execution_id].done() + + executor_module.request_cancel_background_task(execution_id) + finally: + hold.set() + + try: + deadline = time.monotonic() + 5 + while time.monotonic() < deadline and not result.status.is_terminal: + time.sleep(0.01) + + assert not body_ran.is_set() + assert result.status.value == "cancelled" + assert result.error == "Cancelled before the execution started" + + deadline = time.monotonic() + 5 + while time.monotonic() < deadline and not result.execution_done_event.is_set(): + time.sleep(0.01) + assert result.execution_done_event.is_set() + + executor_module.cleanup_background_task(execution_id) + assert execution_id not in executor_module._background_tasks + finally: + executor_module._background_tasks.pop(execution_id, None) + executor_module._background_futures.pop(execution_id, None) + + def test_cancel_before_the_submission_callback_never_fences_a_live_slot( + self, + executor_module, + classes, + base_config, + monkeypatch, + ): + """A cancel that beats the submission callback must not admit a worker. + + Holding the loop in an earlier callback keeps the submission queued, so + ``Future.cancel()`` wins before ``run_coroutine_threadsafe`` has even + created the asyncio task. The done-callback then publishes the terminal + status and opens the cleanup fence on the loop, one turn *before* that + deferred task takes its first step — so the wrapper could still reach + ``_aexecute``, take the only capacity slot, and have its registry entry + removed while the slot was live. + + The fix is a startup handshake rather than better ordering: + ``request_cancel_background_task`` sets ``cancel_event`` before it + cancels anything, and a wrapper that starts afterwards declines to + create the execution task. The invariant asserted here is the one that + matters — the fence never opens while a worker holds a slot. + """ + from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig + from deerflow.subagents.capacity import SubagentExecutionCapacity + + SubagentExecutor = classes["SubagentExecutor"] + capacity = SubagentExecutionCapacity(SubagentRuntimeConfig(max_running=1)) + executor = SubagentExecutor( + config=base_config, + tools=[], + thread_id="test-thread", + execution_capacity=capacity, + ) + + admitted = threading.Event() + admission_release = asyncio.Event() + loop = executor_module._get_isolated_subagent_loop() + + async def hold_admission(*_args, **_kwargs): + admitted.set() + await admission_release.wait() + return {}, [], None + + monkeypatch.setattr(executor, "_build_initial_state", hold_admission) + + hold = threading.Event() + release = threading.Event() + + def hold_loop() -> None: + hold.set() + release.wait(5) + + loop.call_soon_threadsafe(hold_loop) + assert hold.wait(5) + execution_id = None + try: + execution_id = executor.execute_async("Task") + result = executor_module._background_tasks[execution_id] + # The loop is still inside ``hold_loop``, so the submission callback + # that would create the asyncio task has not run yet. + assert result.status.value == "pending" + assert executor_module._background_futures[execution_id].cancel() + + executor_module.request_cancel_background_task(execution_id) + finally: + release.set() + + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and not (admitted.is_set() or result.execution_done_event.is_set()): + time.sleep(0.01) + + if admitted.is_set(): + # A live worker owns the only slot; handing the entry to cleanup + # now would drop it while the release is still unfinished. + assert capacity.snapshot().running == 1 + assert not result.execution_done_event.is_set() + loop.call_soon_threadsafe(admission_release.set) + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and not result.execution_done_event.is_set(): + time.sleep(0.01) + + assert result.execution_done_event.is_set() + assert result.status.value == "cancelled" + assert capacity.snapshot().running == 0 + # The handshake means the wrapper never creates the execution task, + # so nothing can acquire a slot after the fence opened. + assert not admitted.is_set() + + executor_module.cleanup_background_task(execution_id) + assert execution_id not in executor_module._background_tasks + finally: + loop.call_soon_threadsafe(admission_release.set) + if execution_id is not None: + executor_module._background_tasks.pop(execution_id, None) + executor_module._background_futures.pop(execution_id, None) + + def test_cancel_after_the_task_exists_keeps_the_fence_closed_until_teardown( + self, + executor_module, + classes, + base_config, + ): + """A cancel in the task-created window must not fence a live worker off. + + ``Future.cancel()`` on the chained future succeeds while the asyncio task + exists but has not taken its first step, so a start-marker snapshot taken + from the cancelling thread reads "not started" even though the loop still + owns the coroutine. Deciding there published a terminal status and opened + the cleanup fence from the caller's thread; the decision now waits for the + loop, which is the only place serialized with that first step. + + Which owner ends up publishing is a loop-queue coin toss from here: + ``Task.__init__`` queues the first step inside the factory, so the step + can precede the deferred decision and the coroutine's own + ``CancelledError`` branch wins ("Cancelled by user"); when the step loses + that race the deferred decision publishes instead. The invariants that + matter hold either way: the status is terminal ``cancelled``, the fence + opens only after the loop has settled the body, and cleanup removes the + entry. + """ + SubagentExecutor = classes["SubagentExecutor"] + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + task_created = threading.Event() + execution_ran = threading.Event() + factory_release = threading.Event() + loop = executor_module._get_isolated_subagent_loop() + previous_factory = loop.get_task_factory() + + async def slow_execution(_task, result_holder=None): + execution_ran.set() + await asyncio.sleep(30) + return result_holder + + def hold_after_task_creation(loop_arg, coro, context=None): + task = asyncio.Task(coro, loop=loop_arg, context=context) + if not task_created.is_set(): + # The task is queued but cannot step while the loop sits here, + # which is exactly the window the caller-side snapshot misreads. + # Block on an event rather than a timer: on a loaded worker the + # caller may need longer than any fixed delay to request the + # cancel, and the body would then start first and invalidate the + # window this test is about. + task_created.set() + factory_release.wait(5) + return task + + loop.set_task_factory(hold_after_task_creation) + try: + with patch.object(executor, "_aexecute", new=slow_execution): + execution_id = executor.execute_async("Task") + assert task_created.wait(5) + + result = executor_module._background_tasks[execution_id] + executor_module.request_cancel_background_task(execution_id) + + # The loop is still holding the task, so nothing can have finished + # teardown: publishing here would hand the entry to cleanup while + # the worker could still take its slot. + assert not result.status.is_terminal + assert not result.execution_done_event.is_set() + finally: + factory_release.set() + loop.set_task_factory(previous_factory) + + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and not result.execution_done_event.is_set(): + time.sleep(0.01) + + # Either the deferred decision publishes (the step lost the race) or + # the body started and its own CancelledError branch does; both own + # the terminal write legitimately. + assert not execution_ran.is_set() + assert result.status.value == "cancelled" + assert result.error in ("Cancelled before the execution started", "Cancelled by user") + assert result.execution_done_event.is_set() + + executor_module.cleanup_background_task(execution_id) + assert execution_id not in executor_module._background_tasks + finally: + executor_module._background_tasks.pop(execution_id, None) + executor_module._background_futures.pop(execution_id, None) + + def test_cancel_of_a_running_body_keeps_its_own_terminal_write( + self, + executor_module, + classes, + base_config, + ): + """A cancelled running body still owns the terminal write. + + The deferred pre-start decision runs after the coroutine's first step, so a + body that already started must win: it reports "Cancelled by user" after its + own cleanup, not the pre-start message. + """ + SubagentExecutor = classes["SubagentExecutor"] + executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread") + execution_ran = threading.Event() + + async def slow_execution(_task, result_holder=None): + execution_ran.set() + await asyncio.sleep(30) + return result_holder + + with patch.object(executor, "_aexecute", new=slow_execution): + execution_id = executor.execute_async("Task") + assert execution_ran.wait(5) + result = executor_module._background_tasks[execution_id] + + executor_module.request_cancel_background_task(execution_id) + + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and not result.execution_done_event.is_set(): + time.sleep(0.01) + + assert result.status.value == "cancelled" + assert result.error == "Cancelled by user" + assert result.execution_done_event.is_set() + + executor_module.cleanup_background_task(execution_id) + assert execution_id not in executor_module._background_tasks + finally: + executor_module._background_tasks.pop(execution_id, None) + executor_module._background_futures.pop(execution_id, None) + def test_execute_async_runs_without_calling_execute(self, executor_module, classes, base_config): """Regression: execute_async should not route through execute()/asyncio.run().""" import concurrent.futures