fix(subagents): keep isolated-loop shutdown ownership retryable (#6316)

* fix(subagents): retain isolated-loop shutdown ownership

* fix(subagents): reap stopped isolated loop before replacement

* test(subagents): cover pending-loop recovery after worker exit

* docs(subagents): document pending-loop recovery

* fix(subagents): revalidate pending loop recovery

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

@@ -42,7 +42,7 @@ does not change retry policy.
 **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. 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 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.
 **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/executor.py

@@ -597,7 +597,9 @@ def _harvest_bash_executions(
 _isolated_subagent_loop: asyncio.AbstractEventLoop | None = None
 _isolated_subagent_loop_thread: threading.Thread | None = None
 _isolated_subagent_loop_started: threading.Event | None = None
+_isolated_subagent_loop_shutdown_pending = False
 _isolated_subagent_loop_lock = threading.Lock()
+_isolated_subagent_loop_shutdown_lock = threading.Lock()
 
 
 def _run_isolated_subagent_loop(
@@ -613,38 +615,56 @@ def _run_isolated_subagent_loop(
         started_event.clear()
 
 
-def _shutdown_isolated_subagent_loop() -> None:
+def _shutdown_isolated_subagent_loop(*, only_if_pending: bool = False) -> None:
     """Stop and close the persistent isolated subagent loop."""
-    global _isolated_subagent_loop, _isolated_subagent_loop_thread, _isolated_subagent_loop_started
+    global _isolated_subagent_loop, _isolated_subagent_loop_thread, _isolated_subagent_loop_started, _isolated_subagent_loop_shutdown_pending
 
-    with _isolated_subagent_loop_lock:
-        loop = _isolated_subagent_loop
-        thread = _isolated_subagent_loop_thread
-        _isolated_subagent_loop = None
-        _isolated_subagent_loop_thread = None
-        _isolated_subagent_loop_started = None
+    with _isolated_subagent_loop_shutdown_lock:
+        with _isolated_subagent_loop_lock:
+            # Dispatch recovery can become stale while waiting for this lock.
+            # Recheck the fence before touching the current loop generation.
+            if only_if_pending and not _isolated_subagent_loop_shutdown_pending:
+                return
+            loop = _isolated_subagent_loop
+            thread = _isolated_subagent_loop_thread
+            if loop is None:
+                _isolated_subagent_loop_shutdown_pending = False
+                return
+            _isolated_subagent_loop_shutdown_pending = True
 
-    if loop is None:
-        return
+            # A previous bounded shutdown/startup attempt already requested
+            # loop.stop. Dispatch-side recovery should only probe whether the
+            # retained worker has exited; repeatedly joining here would stall
+            # every caller for up to one second while the worker remains live.
+            if only_if_pending and thread is not None and thread.is_alive():
+                return
 
-    if loop.is_running():
-        loop.call_soon_threadsafe(loop.stop)
+        if loop.is_running():
+            loop.call_soon_threadsafe(loop.stop)
 
-    if thread is not None and thread.is_alive() and thread is not threading.current_thread():
-        thread.join(timeout=1)
+        if thread is not None and thread.is_alive() and thread is not threading.current_thread():
+            thread.join(timeout=1)
 
-    thread_stopped = thread is None or not thread.is_alive()
-    loop_stopped = not loop.is_running()
+        thread_stopped = thread is None or not thread.is_alive()
+        loop_stopped = not loop.is_running()
 
-    if not loop.is_closed():
-        if thread_stopped and loop_stopped:
-            loop.close()
-        else:
+        if not thread_stopped or not loop_stopped:
             logger.warning(
-                "Skipping close of isolated subagent loop because shutdown did not complete within timeout (thread_alive=%s, loop_running=%s)",
+                "Retaining isolated subagent loop ownership because shutdown did not complete within timeout (thread_alive=%s, loop_running=%s)",
                 thread is not None and thread.is_alive(),
                 loop.is_running(),
             )
+            return
+
+        if not loop.is_closed():
+            loop.close()
+
+        with _isolated_subagent_loop_lock:
+            if _isolated_subagent_loop is loop and _isolated_subagent_loop_thread is thread:
+                _isolated_subagent_loop = None
+                _isolated_subagent_loop_thread = None
+                _isolated_subagent_loop_started = None
+                _isolated_subagent_loop_shutdown_pending = False
 
 
 atexit.register(_shutdown_isolated_subagent_loop)
@@ -652,8 +672,19 @@ atexit.register(_shutdown_isolated_subagent_loop)
 
 def _get_isolated_subagent_loop() -> asyncio.AbstractEventLoop:
     """Return the persistent event loop used by isolated subagent executions."""
-    global _isolated_subagent_loop, _isolated_subagent_loop_thread, _isolated_subagent_loop_started
+    global _isolated_subagent_loop, _isolated_subagent_loop_thread, _isolated_subagent_loop_started, _isolated_subagent_loop_shutdown_pending
+
+    # A startup/shutdown timeout retains ownership while the worker is alive.
+    # Ordinary dispatch is also the retry path once that worker has actually
+    # exited: reap the retained loop under the shutdown lifecycle lock before
+    # deciding whether replacement is still fenced.
+    if _isolated_subagent_loop_shutdown_pending:
+        _shutdown_isolated_subagent_loop(only_if_pending=True)
+
     with _isolated_subagent_loop_lock:
+        if _isolated_subagent_loop_shutdown_pending:
+            raise RuntimeError("Isolated subagent event loop shutdown is still pending; retained worker is still exiting")
+
         thread_is_alive = _isolated_subagent_loop_thread is not None and _isolated_subagent_loop_thread.is_alive()
         loop_is_usable = _isolated_subagent_loop is not None and not _isolated_subagent_loop.is_closed() and _isolated_subagent_loop.is_running() and thread_is_alive
 
@@ -670,7 +701,18 @@ def _get_isolated_subagent_loop() -> asyncio.AbstractEventLoop:
             if not started_event.wait(timeout=5):
                 loop.call_soon_threadsafe(loop.stop)
                 thread.join(timeout=1)
-                loop.close()
+                if thread.is_alive() or loop.is_running():
+                    _isolated_subagent_loop = loop
+                    _isolated_subagent_loop_thread = thread
+                    _isolated_subagent_loop_started = started_event
+                    _isolated_subagent_loop_shutdown_pending = True
+                    logger.warning(
+                        "Retaining isolated subagent loop ownership after startup timeout (thread_alive=%s, loop_running=%s)",
+                        thread.is_alive(),
+                        loop.is_running(),
+                    )
+                elif not loop.is_closed():
+                    loop.close()
                 raise RuntimeError("Timed out starting isolated subagent event loop")
             _isolated_subagent_loop = loop
             _isolated_subagent_loop_thread = thread

backend/packages/harness/deerflow/tools/builtins/task_tool.py

@@ -498,8 +498,8 @@ def _schedule_deferred_subagent_cleanup(
     try:
         cleanup_handle = run_on_isolated_subagent_loop(coro)
     except Exception:
-        # Unreachable in practice — the persistent loop backs the subagent
-        # execution itself, so it exists by the time a poller needs cleanup.
+        # The persistent loop can be temporarily fenced while retained
+        # shutdown ownership is still exiting; keep cleanup on the caller loop.
         logger.warning(
             f"[trace={trace_id}] Persistent subagent loop unavailable for deferred cleanup of {execution_id}; falling back to the caller loop",
             exc_info=True,

backend/tests/test_subagent_executor.py

@@ -3134,6 +3134,285 @@ class TestThreadSafety:
         # it returns only once the coroutine ran and the future resolved.
         assert handles[0].result(timeout=10) is None
 
+    def test_shutdown_retains_isolated_loop_ownership_until_worker_exits(self, executor_module):
+        """A bounded join must not detach a still-live persistent loop."""
+
+        class StubbornThread:
+            def __init__(self):
+                self.alive = True
+                self.join_calls = 0
+
+            def is_alive(self):
+                return self.alive
+
+            def join(self, timeout=None):
+                self.join_calls += 1
+
+        class DeferredStopLoop:
+            def __init__(self):
+                self.running = True
+                self.closed = False
+                self.stop_calls = 0
+
+            def is_running(self):
+                return self.running
+
+            def stop(self):
+                self.stop_calls += 1
+
+            def call_soon_threadsafe(self, callback, *args):
+                self.stop_calls += 1
+
+            def is_closed(self):
+                return self.closed
+
+            def close(self):
+                assert not self.running
+                self.closed = True
+
+        loop = DeferredStopLoop()
+        thread = StubbornThread()
+        started = threading.Event()
+        started.set()
+
+        executor_module._isolated_subagent_loop = loop
+        executor_module._isolated_subagent_loop_thread = thread
+        executor_module._isolated_subagent_loop_started = started
+        executor_module._isolated_subagent_loop_shutdown_pending = False
+
+        executor_module._shutdown_isolated_subagent_loop()
+
+        assert executor_module._isolated_subagent_loop is loop
+        assert executor_module._isolated_subagent_loop_thread is thread
+        assert executor_module._isolated_subagent_loop_started is started
+        assert executor_module._isolated_subagent_loop_shutdown_pending is True
+        assert loop.closed is False
+        assert thread.join_calls == 1
+
+        with patch.object(executor_module.asyncio, "new_event_loop") as new_event_loop:
+            with pytest.raises(RuntimeError, match="shutdown is still pending"):
+                executor_module._get_isolated_subagent_loop()
+            new_event_loop.assert_not_called()
+
+        thread.alive = False
+        loop.running = False
+        replacement = executor_module._get_isolated_subagent_loop()
+
+        assert loop.closed is True
+        assert replacement is not loop
+        assert executor_module._isolated_subagent_loop is replacement
+        assert executor_module._isolated_subagent_loop_thread is not thread
+        assert executor_module._isolated_subagent_loop_shutdown_pending is False
+
+        executor_module._shutdown_isolated_subagent_loop()
+        assert executor_module._isolated_subagent_loop is None
+        assert executor_module._isolated_subagent_loop_thread is None
+        assert executor_module._isolated_subagent_loop_started is None
+        assert executor_module._isolated_subagent_loop_shutdown_pending is False
+
+    def test_startup_timeout_retains_ownership_until_worker_exits(self, executor_module, caplog):
+        """A startup timeout must retain the worker and allow later reaping."""
+
+        class StartupEvent:
+            def wait(self, timeout=None):
+                return False
+
+        class StartupThread:
+            def __init__(self, *args, **kwargs):
+                self.alive = False
+                self.join_calls = 0
+
+            def start(self):
+                self.alive = True
+
+            def is_alive(self):
+                return self.alive
+
+            def join(self, timeout=None):
+                self.join_calls += 1
+
+        class StartupLoop:
+            def __init__(self):
+                self.running = True
+                self.closed = False
+
+            def is_running(self):
+                return self.running
+
+            def stop(self):
+                self.running = False
+
+            def call_soon_threadsafe(self, callback, *args):
+                callback(*args)
+
+            def is_closed(self):
+                return self.closed
+
+            def close(self):
+                assert not self.running
+                self.closed = True
+
+        retained = StartupLoop()
+        startup_thread = StartupThread()
+        startup_event = StartupEvent()
+
+        with (
+            caplog.at_level("WARNING"),
+            patch.object(executor_module.asyncio, "new_event_loop", return_value=retained),
+            patch.object(executor_module.threading, "Event", return_value=startup_event),
+            patch.object(executor_module.threading, "Thread", return_value=startup_thread),
+        ):
+            with pytest.raises(RuntimeError, match="Timed out starting isolated subagent event loop"):
+                executor_module._get_isolated_subagent_loop()
+
+        assert executor_module._isolated_subagent_loop is retained
+        assert executor_module._isolated_subagent_loop_thread is startup_thread
+        assert executor_module._isolated_subagent_loop_started is startup_event
+        assert executor_module._isolated_subagent_loop_shutdown_pending is True
+        assert startup_thread.join_calls == 1
+        assert retained.closed is False
+        assert "Retaining isolated subagent loop ownership after startup timeout" in caplog.text
+
+        with patch.object(executor_module.asyncio, "new_event_loop") as new_event_loop:
+            with pytest.raises(RuntimeError, match="retained worker is still exiting"):
+                executor_module._get_isolated_subagent_loop()
+            new_event_loop.assert_not_called()
+
+        # Dispatch-side recovery is only a liveness probe while ownership is
+        # retained; it must not block on another bounded join for every caller.
+        assert startup_thread.join_calls == 1
+
+        startup_thread.alive = False
+        retained.running = False
+        replacement = executor_module._get_isolated_subagent_loop()
+
+        assert retained.closed is True
+        assert replacement is not retained
+        assert executor_module._isolated_subagent_loop is replacement
+        assert executor_module._isolated_subagent_loop_thread is not startup_thread
+        assert executor_module._isolated_subagent_loop_shutdown_pending is False
+
+        executor_module._shutdown_isolated_subagent_loop()
+
+    def test_concurrent_pending_recovery_does_not_stop_replacement(self, executor_module, monkeypatch):
+        """A stale recovery getter must not tear down a replacement loop."""
+
+        class GateLock:
+            def __init__(self):
+                self._lock = threading.Lock()
+                self._count_lock = threading.Lock()
+                self._release = threading.Event()
+                self.both_waiting = threading.Event()
+                self._attempts = 0
+
+            def __enter__(self):
+                with self._count_lock:
+                    self._attempts += 1
+                    gated = self._attempts <= 2
+                    if self._attempts == 2:
+                        self.both_waiting.set()
+                if gated and not self._release.wait(timeout=5):
+                    raise RuntimeError("timed out waiting to release lifecycle contenders")
+                self._lock.acquire()
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                self._lock.release()
+
+            def release_waiters(self):
+                self._release.set()
+
+        class FakeLoop:
+            def __init__(self, *, running):
+                self.running = running
+                self.closed = False
+
+            def is_running(self):
+                return self.running
+
+            def stop(self):
+                self.running = False
+
+            def call_soon_threadsafe(self, callback, *args):
+                callback(*args)
+
+            def is_closed(self):
+                return self.closed
+
+            def close(self):
+                assert not self.running
+                self.closed = True
+
+        class DeadThread:
+            def is_alive(self):
+                return False
+
+            def join(self, timeout=None):
+                return None
+
+        class ReplacementThread:
+            def __init__(self, *, target, args, name, daemon):
+                self.args = args
+                self.alive = False
+
+            def start(self):
+                self.alive = True
+                self.args[1].set()
+
+            def is_alive(self):
+                return self.alive
+
+            def join(self, timeout=None):
+                self.alive = False
+
+        retained = FakeLoop(running=False)
+        retained_thread = DeadThread()
+        replacement = FakeLoop(running=True)
+        started = threading.Event()
+
+        executor_module._isolated_subagent_loop = retained
+        executor_module._isolated_subagent_loop_thread = retained_thread
+        executor_module._isolated_subagent_loop_started = started
+        executor_module._isolated_subagent_loop_shutdown_pending = True
+
+        gate = GateLock()
+        real_thread = threading.Thread
+        monkeypatch.setattr(executor_module, "_isolated_subagent_loop_shutdown_lock", gate)
+        monkeypatch.setattr(executor_module.asyncio, "new_event_loop", MagicMock(return_value=replacement))
+        monkeypatch.setattr(executor_module.threading, "Thread", ReplacementThread)
+
+        results = []
+        errors = []
+
+        def recover():
+            try:
+                results.append(executor_module._get_isolated_subagent_loop())
+            except BaseException as exc:
+                errors.append(exc)
+
+        first = real_thread(target=recover)
+        second = real_thread(target=recover)
+        first.start()
+        second.start()
+
+        assert gate.both_waiting.wait(timeout=5)
+        gate.release_waiters()
+        first.join(timeout=5)
+        second.join(timeout=5)
+
+        assert not first.is_alive()
+        assert not second.is_alive()
+        assert errors == []
+        assert results == [replacement, replacement]
+        assert retained.closed is True
+        assert replacement.closed is False
+        assert replacement.is_running()
+        assert executor_module._isolated_subagent_loop is replacement
+        assert executor_module._isolated_subagent_loop_shutdown_pending is False
+
+        executor_module._shutdown_isolated_subagent_loop()
+        assert replacement.closed is True
+
     def test_multiple_executors_in_parallel(self, classes, base_config, msg):
         """Test multiple executors running in parallel via thread pool."""
         from concurrent.futures import ThreadPoolExecutor, as_completed