fix(delegation): progress-based stale detection for detached async runners

Replace the wall-clock timeout watchdog (from #60234) with progress-based
staleness detection, on by default with zero config:

- The async registry now accepts a progress_fn per dispatch; delegate_task
  wires a sampler over the batch's child agents (api_call_count +
  current_tool from get_activity_summary()).
- A single monitor thread sweeps running delegations: a child whose
  progress token keeps advancing is never touched, no matter how long it
  runs. A frozen token past the stale threshold (450s idle / 1200s
  in-tool, mirroring the sync-path heartbeat monitor) marks the record
  'stalling' and interrupts the child.
- A stalling child that unwinds within the grace window (120s) finalizes
  through the NORMAL path, preserving its partial results. One that never
  returns is force-finalized with a terminal 'stalled' completion event so
  the owning session hears an outcome and the async slot frees.
- Late runner returns after force-finalization are deduped by the
  begin/push/finish finalization split (kept from #60234).

Why not a timeout: delegation.child_timeout_seconds defaults to 0 by
deliberate design (DEFAULT_CHILD_TIMEOUT rationale) — a timeout-based
watchdog never arms for default configs, leaving the reported silent-
profile symptom (#60203) unfixed, and when armed it kills legitimately
slow heavy subagents mid-task. Progress detection distinguishes 'wedged
at first API call' from 'grinding through a 2h review'.

Builds on izumi0uu's finalization-atomicity work from #60234.

tests/tools/test_async_delegation.py

@@ -229,12 +229,21 @@ def test_interrupt_all_signals_running_children():
     assert evt["status"] == "interrupted"
 
 
-def test_async_delegation_timeout_finalizes_stuck_runner():
+def _fast_stale_monitor(monkeypatch, *, idle=0.15, in_tool=0.3, grace=0.15):
+    """Shrink the stale-monitor cadence so tests run in milliseconds."""
+    monkeypatch.setattr(ad, "_STALE_CHECK_INTERVAL", 0.03)
+    monkeypatch.setattr(ad, "_STALE_IDLE_SECONDS", idle)
+    monkeypatch.setattr(ad, "_STALE_IN_TOOL_SECONDS", in_tool)
+    monkeypatch.setattr(ad, "_STALL_GRACE_SECONDS", grace)
+
+
+def test_stalled_runner_is_interrupted_then_finalized(monkeypatch):
+    _fast_stale_monitor(monkeypatch)
     gate = threading.Event()
     interrupted = {"count": 0}
 
     def stuck_runner():
-        gate.wait(timeout=5)
+        gate.wait(timeout=10)
         return {"status": "completed", "summary": "too late"}
 
     def interrupt_fn():
@@ -243,34 +252,108 @@ def test_async_delegation_timeout_finalizes_stuck_runner():
     res = ad.dispatch_async_delegation(
         goal="stuck child", context=None, toolsets=None, role="leaf",
         model="m", session_key="", runner=stuck_runner,
-        interrupt_fn=interrupt_fn, max_async_children=1, timeout_seconds=0.1,
+        interrupt_fn=interrupt_fn, max_async_children=1,
+        # Frozen progress token: the child never advances an API call.
+        progress_fn=lambda: ((0, None), False),
     )
     assert res["status"] == "dispatched"
 
-    evt = _drain_one(timeout=2.0)
+    evt = _drain_for(res["delegation_id"], timeout=5.0)
     try:
         assert evt is not None
         assert evt["type"] == "async_delegation"
-        assert evt["status"] == "timeout"
+        assert evt["status"] == "stalled"
         assert evt["delegation_id"] == res["delegation_id"]
         assert evt["api_calls"] == 0
-        assert "timed out after 0.1s" in evt["error"]
-        assert interrupted["count"] == 1
+        assert "stalled" in evt["error"]
+        # Interrupt was requested BEFORE force-finalization (grace window).
+        assert interrupted["count"] >= 1
         assert ad.active_count() == 0
     finally:
         gate.set()
 
     # If the ignored runner eventually returns, it must not enqueue a second
-    # completion for a delegation the watchdog already finalized.
+    # completion for a delegation the monitor already finalized.
     assert _drain_one(timeout=0.5) is None
 
 
-def test_async_delegation_batch_timeout_finalizes_stuck_runner():
+def test_progressing_runner_is_never_stalled(monkeypatch):
+    """A child that keeps advancing is left alone no matter how long it runs."""
+    _fast_stale_monitor(monkeypatch)
+    gate = threading.Event()
+    ticks = {"n": 0}
+
+    def slow_but_alive_runner():
+        gate.wait(timeout=10)
+        return {"status": "completed", "summary": "done", "api_calls": 7}
+
+    def progress_fn():
+        # Token advances on every sample — simulates a child making steady
+        # API-call progress.
+        ticks["n"] += 1
+        return (ticks["n"], None), False
+
+    res = ad.dispatch_async_delegation(
+        goal="slow child", context=None, toolsets=None, role="leaf",
+        model="m", session_key="", runner=slow_but_alive_runner,
+        max_async_children=1, progress_fn=progress_fn,
+    )
+    assert res["status"] == "dispatched"
+
+    # Run well past the (shrunk) idle threshold — several monitor sweeps.
+    time.sleep(0.6)
+    assert ad.active_count() == 1
+    assert process_registry.completion_queue.empty()
+
+    gate.set()
+    evt = _drain_for(res["delegation_id"], timeout=5.0)
+    assert evt is not None
+    assert evt["status"] == "completed"
+    assert evt["summary"] == "done"
+
+
+def test_stalling_runner_that_honors_interrupt_keeps_its_result(monkeypatch):
+    """Interrupt-responsive children finalize through the NORMAL path.
+
+    The monitor's interrupt gives a wedged-looking child a grace window; if
+    the runner returns during it, the real result (partial work, api_calls)
+    is delivered instead of a synthetic stalled event.
+    """
+    _fast_stale_monitor(monkeypatch, grace=5.0)
+    interrupted = threading.Event()
+
+    def runner():
+        # "Wedged" until interrupted, then unwinds and reports partial work.
+        interrupted.wait(timeout=10)
+        return {
+            "status": "interrupted",
+            "summary": "partial work saved",
+            "api_calls": 3,
+        }
+
+    res = ad.dispatch_async_delegation(
+        goal="responsive child", context=None, toolsets=None, role="leaf",
+        model="m", session_key="", runner=runner,
+        interrupt_fn=interrupted.set, max_async_children=1,
+        progress_fn=lambda: ((3, None), False),
+    )
+    assert res["status"] == "dispatched"
+
+    evt = _drain_for(res["delegation_id"], timeout=5.0)
+    assert evt is not None
+    assert evt["status"] == "interrupted"
+    assert evt["summary"] == "partial work saved"
+    assert evt["api_calls"] == 3
+    assert ad.active_count() == 0
+
+
+def test_stalled_batch_is_interrupted_then_finalized(monkeypatch):
+    _fast_stale_monitor(monkeypatch)
     gate = threading.Event()
     interrupted = {"count": 0}
 
     def stuck_batch():
-        gate.wait(timeout=5)
+        gate.wait(timeout=10)
         return {"results": [{"status": "completed", "summary": "too late"}]}
 
     def interrupt_fn():
@@ -279,20 +362,21 @@ def test_async_delegation_batch_timeout_finalizes_stuck_runner():
     res = ad.dispatch_async_delegation_batch(
         goals=["a", "b"], context="ctx", toolsets=None, role="leaf",
         model="m", session_key="", runner=stuck_batch,
-        interrupt_fn=interrupt_fn, max_async_children=1, timeout_seconds=0.1,
+        interrupt_fn=interrupt_fn, max_async_children=1,
+        progress_fn=lambda: (((0, None), (0, None)), False),
     )
     assert res["status"] == "dispatched"
 
-    evt = _drain_one(timeout=2.0)
+    evt = _drain_for(res["delegation_id"], timeout=5.0)
     try:
         assert evt is not None
         assert evt["type"] == "async_delegation"
-        assert evt["status"] == "timeout"
+        assert evt["status"] == "stalled"
         assert evt["is_batch"] is True
         assert evt["goals"] == ["a", "b"]
         assert evt["results"] == []
-        assert "timed out after 0.1s" in evt["error"]
-        assert interrupted["count"] == 1
+        assert "stalled" in evt["error"]
+        assert interrupted["count"] >= 1
         assert ad.active_count() == 0
     finally:
         gate.set()
@@ -300,7 +384,36 @@ def test_async_delegation_batch_timeout_finalizes_stuck_runner():
     assert _drain_one(timeout=0.5) is None
 
 
-def test_timeout_stays_finalizing_until_durable_persistence(tmp_path, monkeypatch):
+def test_in_tool_stall_uses_higher_threshold(monkeypatch):
+    """A frozen child inside a tool gets the in-tool ceiling, not the idle one."""
+    _fast_stale_monitor(monkeypatch, idle=0.1, in_tool=10.0, grace=0.1)
+    gate = threading.Event()
+
+    def runner():
+        gate.wait(timeout=10)
+        return {"status": "completed", "summary": "long tool finished"}
+
+    res = ad.dispatch_async_delegation(
+        goal="long tool child", context=None, toolsets=None, role="leaf",
+        model="m", session_key="", runner=runner, max_async_children=1,
+        # Frozen token but in_tool=True — a legitimately slow terminal
+        # command / web fetch. Must NOT be stalled at the idle threshold.
+        progress_fn=lambda: ((1, "terminal"), True),
+    )
+    assert res["status"] == "dispatched"
+
+    time.sleep(0.5)  # far past idle threshold, well under in-tool threshold
+    assert ad.active_count() == 1
+    assert process_registry.completion_queue.empty()
+
+    gate.set()
+    evt = _drain_for(res["delegation_id"], timeout=5.0)
+    assert evt is not None
+    assert evt["status"] == "completed"
+
+
+def test_stall_stays_finalizing_until_durable_persistence(tmp_path, monkeypatch):
+    _fast_stale_monitor(monkeypatch)
     gate = threading.Event()
     persist_entered = threading.Event()
     allow_persist = threading.Event()
@@ -312,19 +425,19 @@ def test_timeout_stays_finalizing_until_durable_persistence(tmp_path, monkeypatc
         real_persist(event, result)
 
     def stuck_runner():
-        gate.wait(timeout=5)
+        gate.wait(timeout=10)
         return {"status": "completed", "summary": "too late"}
 
     monkeypatch.setenv("HERMES_HOME", str(tmp_path))
     monkeypatch.setattr(ad, "_persist_completion", blocking_persist)
     dispatched = ad.dispatch_async_delegation(
-        goal="durable timeout", context=None, toolsets=None, role="leaf",
+        goal="durable stall", context=None, toolsets=None, role="leaf",
         model="m", session_key="owner", runner=stuck_runner,
-        max_async_children=1, timeout_seconds=0.05,
+        max_async_children=1, progress_fn=lambda: ((0, None), False),
     )
 
     try:
-        assert persist_entered.wait(timeout=2)
+        assert persist_entered.wait(timeout=5)
         assert ad.active_count() == 1
         record = next(
             item for item in ad.list_async_delegations()
@@ -336,17 +449,17 @@ def test_timeout_stays_finalizing_until_durable_persistence(tmp_path, monkeypatc
         allow_persist.set()
         evt = _drain_for(dispatched["delegation_id"])
         assert evt is not None
-        assert evt["status"] == "timeout"
+        assert evt["status"] == "stalled"
         assert ad.active_count() == 0
         durable = ad.get_durable_delegation(dispatched["delegation_id"])
-        assert durable["state"] == "timeout"
+        assert durable["state"] == "stalled"
         assert durable["delivery_state"] == "pending"
     finally:
         allow_persist.set()
         gate.set()
 
 
-def test_timeout_completion_restores_once_after_process_restart(tmp_path):
+def test_stalled_completion_restores_once_after_process_restart(tmp_path):
     repo = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
     env = {**os.environ, "HERMES_HOME": str(tmp_path), "PYTHONPATH": repo}
     producer = r'''
@@ -354,13 +467,17 @@ import json
 import threading
 import time
 from tools import async_delegation as ad
+ad._STALE_CHECK_INTERVAL = 0.03
+ad._STALE_IDLE_SECONDS = 0.1
+ad._STALL_GRACE_SECONDS = 0.1
 gate = threading.Event()
 r = ad.dispatch_async_delegation(
-    goal="restart timeout", context=None, toolsets=None, role="leaf", model="m",
+    goal="restart stall", context=None, toolsets=None, role="leaf", model="m",
     session_key="owner-session", parent_session_id="durable-parent",
-    runner=lambda: gate.wait(timeout=60), timeout_seconds=.05,
+    runner=lambda: gate.wait(timeout=60),
+    progress_fn=lambda: ((0, None), False),
 )
-deadline = time.time() + 5
+deadline = time.time() + 10
 while ad.active_count() and time.time() < deadline:
     time.sleep(.01)
 row = ad.get_durable_delegation(r["delegation_id"])
@@ -368,11 +485,11 @@ print(json.dumps({"delegation_id": r["delegation_id"], "row": row}, sort_keys=Tr
 '''
     first = subprocess.run(
         [sys.executable, "-c", producer], cwd=repo, env=env,
-        text=True, capture_output=True, timeout=15, check=True,
+        text=True, capture_output=True, timeout=30, check=True,
     )
     produced = json.loads(first.stdout.strip().splitlines()[-1])
     delegation_id = produced["delegation_id"]
-    assert produced["row"]["state"] == "timeout"
+    assert produced["row"]["state"] == "stalled"
     assert produced["row"]["delivery_state"] == "pending"
 
     consumer = r'''
@@ -388,7 +505,7 @@ print(json.dumps({"event": evt, "remaining": process_registry.completion_queue.q
     restored = json.loads(second.stdout.strip().splitlines()[-1])
     assert restored["remaining"] == 0
     assert restored["event"]["delegation_id"] == delegation_id
-    assert restored["event"]["status"] == "timeout"
+    assert restored["event"]["status"] == "stalled"
     assert restored["event"]["restored"] is True
 
     acker = f'''
@@ -944,7 +1061,7 @@ def test_delegate_task_background_batch_runs_as_one_unit(monkeypatch):
     assert _drain_one() is None
 
 
-def test_delegate_task_background_passes_child_timeout_to_async_registry(monkeypatch):
+def test_delegate_task_background_passes_progress_fn_to_async_registry(monkeypatch):
     import json
     from unittest.mock import MagicMock
     import tools.delegate_tool as dt
@@ -959,6 +1076,10 @@ def test_delegate_task_background_passes_child_timeout_to_async_registry(monkeyp
     fake_child = MagicMock()
     fake_child._delegate_role = "leaf"
     fake_child._subagent_id = "s1"
+    fake_child.get_activity_summary.return_value = {
+        "api_call_count": 4,
+        "current_tool": "terminal",
+    }
 
     creds = {
         "model": "m", "provider": None, "base_url": None, "api_key": None,
@@ -968,19 +1089,24 @@ def test_delegate_task_background_passes_child_timeout_to_async_registry(monkeyp
 
     def fake_dispatch(**kwargs):
         captured.update(kwargs)
-        return {"status": "dispatched", "delegation_id": "deleg_timeout"}
+        return {"status": "dispatched", "delegation_id": "deleg_progress"}
 
     monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
     monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds)
-    monkeypatch.setattr(dt, "_get_child_timeout", lambda: 600.0)
     monkeypatch.setattr(ad, "dispatch_async_delegation_batch", fake_dispatch)
 
-    out = dt.delegate_task(goal="background timeout", background=True, parent_agent=parent)
+    out = dt.delegate_task(goal="background stall guard", background=True, parent_agent=parent)
 
     parsed = json.loads(out)
     assert parsed["status"] == "dispatched"
-    assert parsed["delegation_id"] == "deleg_timeout"
-    assert captured["timeout_seconds"] == 600.0
+    assert parsed["delegation_id"] == "deleg_progress"
+    # The dispatch wires a live progress sampler over the child agents so the
+    # async registry's stale monitor can watch the detached batch.
+    progress_fn = captured["progress_fn"]
+    assert callable(progress_fn)
+    token, in_tool = progress_fn()
+    assert token == ((4, "terminal"),)
+    assert in_tool is True
 
 
 def test_model_dispatch_forces_background():

tools/async_delegation.py

@@ -85,6 +85,36 @@ _MAX_DURABLE_PENDING = 1000
 _MAX_DELIVERY_ATTEMPTS = 8
 _DB_LOCK = threading.Lock()
 
+# ---------------------------------------------------------------------------
+# Stale-delegation detection (progress-based, on by default)
+# ---------------------------------------------------------------------------
+# A detached runner that wedges before returning (e.g. stuck inside its first
+# model API call — #60203) never reaches its ``finally`` finalizer, so no
+# completion event is ever published: the delegation shows "dispatched"
+# forever and the owning session looks silent until a process restart. We do
+# NOT fix this with a wall-clock timeout — legitimate heavy subagent work
+# (deep reviews, research fan-outs, slow reasoning models) must never be
+# killed for taking long (see delegate_tool.DEFAULT_CHILD_TIMEOUT rationale).
+# Instead a single monitor thread watches per-dispatch PROGRESS (api-call
+# count + current tool, via an injected ``progress_fn``): a child that is
+# advancing is left alone forever; a child with NO progress past the stale
+# threshold is interrupted, given a grace window to unwind and deliver its
+# partial results through the normal finalize path, and only force-finalized
+# with a terminal ``stalled`` event if it never returns.
+#
+# Thresholds mirror the sync-path heartbeat staleness monitor in
+# delegate_tool: idle (not inside a tool) stays tight so a wedged first API
+# call is caught quickly; in-tool is much higher so legitimately slow tools
+# (long terminal commands, big fetches) get time to finish.
+_STALE_CHECK_INTERVAL = 30.0  # seconds between monitor sweeps
+_STALE_IDLE_SECONDS = 450.0  # no progress, no current tool → stalled
+_STALE_IN_TOOL_SECONDS = 1200.0  # no progress while inside a tool → stalled
+_STALL_GRACE_SECONDS = 120.0  # after interrupt, time for the runner to return
+
+_monitor_lock = threading.Lock()
+_monitor_thread: Optional[threading.Thread] = None
+_monitor_stop = threading.Event()
+
 
 def _db_path():
     return get_hermes_home() / "state.db"
@@ -483,10 +513,6 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]:
     }
 
 
-def _format_timeout_seconds(timeout_seconds: float) -> str:
-    return f"{timeout_seconds:g}"
-
-
 def _get_executor(max_workers: int) -> ThreadPoolExecutor:
     """Lazily create (or grow) the shared daemon executor.
 
@@ -509,7 +535,10 @@ def _get_executor(max_workers: int) -> ThreadPoolExecutor:
 def active_count() -> int:
     """Number of async delegations currently running."""
     with _records_lock:
-        return sum(1 for r in _records.values() if r.get("status") in {"running", "finalizing"})
+        return sum(
+            1 for r in _records.values()
+            if r.get("status") in {"running", "stalling", "finalizing"}
+        )
 
 
 def _new_delegation_id() -> str:
@@ -576,7 +605,7 @@ def dispatch_async_delegation(
     origin_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
-    timeout_seconds: Optional[float] = None,
+    progress_fn: Optional[Callable[[], tuple]] = None,
 ) -> Dict[str, Any]:
     """Spawn ``runner`` on the daemon executor and return a handle immediately.
 
@@ -601,6 +630,14 @@ def dispatch_async_delegation(
     interrupt_fn
         Optional callable to signal the child to stop (used on shutdown /
         explicit cancel).
+    progress_fn
+        Optional zero-arg callable returning ``(token, in_tool)`` where
+        ``token`` is any comparable snapshot of the child's progress (api
+        call count + current tool) and ``in_tool`` says whether the child is
+        currently inside a tool call. Sampled by the stale monitor; a frozen
+        token past the stale threshold marks the delegation stuck (see the
+        stale-detection block at the top of this module). When omitted, the
+        delegation is not monitored.
     max_async_children
         Concurrency cap. When at capacity the dispatch is REJECTED (the caller
         should fall back to sync or tell the user) rather than queued, so a
@@ -629,14 +666,19 @@ def dispatch_async_delegation(
         "dispatched_at": dispatched_at,
         "completed_at": None,
         "interrupt_fn": interrupt_fn,
-        "timeout_seconds": timeout_seconds,
+        "progress_fn": progress_fn,
+        # Stale-monitor bookkeeping (see _stale_monitor_loop).
+        "_progress_token": None,
+        "_progress_ts": dispatched_at,
+        "_interrupted_at": None,
     }
     # Capacity check and record insert under ONE lock hold — checking
     # active_count() separately would let two concurrent dispatches (e.g.
     # from different gateway sessions) both pass the check and exceed the cap.
     with _records_lock:
         running = sum(
-            1 for r in _records.values() if r.get("status") == "running"
+            1 for r in _records.values()
+            if r.get("status") in ("running", "stalling")
         )
         if running >= max_async_children:
             return {
@@ -685,7 +727,8 @@ def dispatch_async_delegation(
             "status": "rejected",
             "error": f"Failed to schedule async delegation: {exc}",
         }
-    _start_timeout_watchdog(delegation_id, timeout_seconds, is_batch=False)
+    if progress_fn is not None:
+        _ensure_stale_monitor()
 
     logger.info(
         "Dispatched async delegation %s (session_key=%s): %s",
@@ -711,7 +754,7 @@ def _begin_finalization(
     """Atomically claim terminal delivery while keeping the record active."""
     with _records_lock:
         record = _records.get(delegation_id)
-        if record is None or record.get("status") != "running":
+        if record is None or record.get("status") not in ("running", "stalling"):
             return
         # Stay active until durable persistence and queue publication finish;
         # otherwise process shutdown can kill this daemon worker in the narrow
@@ -720,11 +763,9 @@ def _begin_finalization(
         record["completed_at"] = time.time()
         interrupt_fn = record.get("interrupt_fn")
         record["interrupt_fn"] = None  # drop the closure; child is done
-        timer = record.pop("_timeout_timer", None)
+        record["progress_fn"] = None  # stop stale-monitor sampling
         event_record = dict(record)
 
-    if timer is not None:
-        timer.cancel()
     return event_record, interrupt_fn
 
 
@@ -810,7 +851,7 @@ def dispatch_async_delegation_batch(
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
     delegation_id: Optional[str] = None,
-    timeout_seconds: Optional[float] = None,
+    progress_fn: Optional[Callable[[], tuple]] = None,
 ) -> Dict[str, Any]:
     """Dispatch a WHOLE fan-out batch as ONE background unit.
 
@@ -856,11 +897,15 @@ def dispatch_async_delegation_batch(
         "completed_at": None,
         "interrupt_fn": interrupt_fn,
         "is_batch": True,
-        "timeout_seconds": timeout_seconds,
+        "progress_fn": progress_fn,
+        "_progress_token": None,
+        "_progress_ts": dispatched_at,
+        "_interrupted_at": None,
     }
     with _records_lock:
         running = sum(
-            1 for r in _records.values() if r.get("status") == "running"
+            1 for r in _records.values()
+            if r.get("status") in ("running", "stalling")
         )
         if running >= max_async_children:
             return {
@@ -913,7 +958,8 @@ def dispatch_async_delegation_batch(
             "status": "rejected",
             "error": f"Failed to schedule async delegation batch: {exc}",
         }
-    _start_timeout_watchdog(delegation_id, timeout_seconds, is_batch=True)
+    if progress_fn is not None:
+        _ensure_stale_monitor()
 
     logger.info(
         "Dispatched async delegation batch %s (%d task(s), session_key=%s)",
@@ -989,65 +1035,136 @@ def _push_batch_completion_event(
         )
 
 
-def _start_timeout_watchdog(
-    delegation_id: str,
-    timeout_seconds: Optional[float],
-    *,
-    is_batch: bool,
-) -> None:
-    """Finalize a detached delegation if its worker never returns.
+def _ensure_stale_monitor() -> None:
+    """Start (once) the module-level stale-delegation monitor thread.
 
-    ``delegate_task(background=true)`` returns a handle immediately; the only
-    user-visible result is the completion event this module enqueues later. If
-    the worker thread wedges before returning, the normal ``finally`` block is
-    never reached and users only see a permanent "dispatched" state. A
-    configured timeout must therefore be enforced by the async registry itself,
-    not only inside ``delegate_tool._run_single_child``.
+    One daemon thread serves every dispatch; it exits on its own when no
+    monitorable records remain, and is restarted by the next dispatch that
+    carries a ``progress_fn``.
     """
-    if timeout_seconds is None or timeout_seconds <= 0:
-        return
-
-    timer = threading.Timer(
-        timeout_seconds,
-        _expire_delegation,
-        args=(delegation_id, float(timeout_seconds)),
-        kwargs={"is_batch": is_batch},
-    )
-    timer.daemon = True
-    with _records_lock:
-        record = _records.get(delegation_id)
-        if record is None or record.get("status") != "running":
+    global _monitor_thread
+    with _monitor_lock:
+        if _monitor_thread is not None and _monitor_thread.is_alive():
             return
-        record["timeout_seconds"] = float(timeout_seconds)
-        record["_timeout_timer"] = timer
-    timer.start()
+        _monitor_stop.clear()
+        _monitor_thread = threading.Thread(
+            target=_stale_monitor_loop,
+            name="async-delegate-stale-monitor",
+            daemon=True,
+        )
+        _monitor_thread.start()
 
 
-def _expire_delegation(
-    delegation_id: str,
-    timeout_seconds: float,
-    *,
-    is_batch: bool,
-) -> None:
-    """Timeout a still-running async delegation and emit one completion."""
+def _stale_monitor_loop() -> None:
+    """Sweep running delegations for stalled progress.
+
+    Per sweep, for every running record with a ``progress_fn``:
+
+    - Sample ``(token, in_tool)``. A changed token refreshes the record's
+      progress timestamp — a child that keeps advancing is never touched, no
+      matter how long it runs.
+    - A frozen token past the idle/in-tool threshold marks the record
+      ``stalling``: we call ``interrupt_fn`` so a responsive-but-slow child
+      can unwind and deliver its (partial) result through the normal
+      ``_finalize`` path with full fidelity.
+    - A ``stalling`` record whose runner still hasn't returned after the
+      grace window is force-finalized with one terminal ``stalled`` event so
+      the owning session hears an outcome and the async slot frees. A late
+      runner return after that is ignored by ``_begin_finalization``.
+    """
+    while not _monitor_stop.wait(_STALE_CHECK_INTERVAL):
+        now = time.time()
+        stalled: List[tuple] = []  # (delegation_id, is_batch, quiet_for, in_tool)
+        expired: List[str] = []  # stalling past grace → force-finalize
+        any_monitorable = False
+        with _records_lock:
+            for record in _records.values():
+                status = record.get("status")
+                if status == "stalling":
+                    any_monitorable = True
+                    interrupted_at = record.get("_interrupted_at") or now
+                    if now - interrupted_at >= _STALL_GRACE_SECONDS:
+                        expired.append(record["delegation_id"])
+                    continue
+                if status != "running":
+                    continue
+                progress_fn = record.get("progress_fn")
+                if progress_fn is None:
+                    continue
+                any_monitorable = True
+                try:
+                    token, in_tool = progress_fn()
+                except Exception:
+                    # An unreadable child must not look permanently healthy —
+                    # keep the last timestamp running instead of refreshing it.
+                    token, in_tool = record.get("_progress_token"), False
+                if token != record.get("_progress_token"):
+                    record["_progress_token"] = token
+                    record["_progress_ts"] = now
+                    continue
+                quiet_for = now - (record.get("_progress_ts") or now)
+                limit = (
+                    _STALE_IN_TOOL_SECONDS if in_tool else _STALE_IDLE_SECONDS
+                )
+                if quiet_for >= limit:
+                    record["status"] = "stalling"
+                    record["_interrupted_at"] = now
+                    stalled.append(
+                        (
+                            record["delegation_id"],
+                            bool(record.get("is_batch")),
+                            quiet_for,
+                            in_tool,
+                        )
+                    )
+        for delegation_id, _is_batch, quiet_for, in_tool in stalled:
+            logger.warning(
+                "Async delegation %s made no progress for %.0fs "
+                "(in_tool=%s) — interrupting; grace window %.0fs",
+                delegation_id, quiet_for, in_tool, _STALL_GRACE_SECONDS,
+            )
+            with _records_lock:
+                record = _records.get(delegation_id)
+                fn = record.get("interrupt_fn") if record else None
+            if callable(fn):
+                try:
+                    fn()
+                except Exception as exc:
+                    logger.debug(
+                        "Async delegation %s stall interrupt failed: %s",
+                        delegation_id, exc,
+                    )
+        for delegation_id in expired:
+            _finalize_stalled(delegation_id)
+        if not any_monitorable:
+            return
+
+
+def _finalize_stalled(delegation_id: str) -> None:
+    """Force-finalize a stalling delegation whose runner never returned."""
     claimed = _begin_finalization(delegation_id)
     if claimed is None:
         return
-    event_record, interrupt_fn = claimed
+    event_record, _interrupt_fn = claimed
 
     completed_at = event_record.get("completed_at") or time.time()
     duration = round(
         completed_at - (event_record.get("dispatched_at") or completed_at),
         2,
     )
-    timeout_text = _format_timeout_seconds(timeout_seconds)
     error = (
-        f"Async delegation {delegation_id} timed out after {timeout_text}s "
-        "without producing a completion event. Hermes requested interruption; "
-        "the detached worker may be stuck before or inside the first model API "
-        "call."
+        f"Async delegation {delegation_id} stalled: the detached subagent "
+        "stopped making progress (no new API calls or tool activity), did "
+        "not respond to interruption, and never produced a completion "
+        "event. The worker may be wedged inside a model API call — this is "
+        "a known failure mode of long-lived gateway processes (#60203). "
+        "Re-dispatch the task if it is still needed."
     )
-    if is_batch:
+    logger.error(
+        "Async delegation %s force-finalized as stalled after %.0fs",
+        delegation_id, duration,
+    )
+    if event_record.get("is_batch"):
         _push_batch_completion_event(
             event_record,
             {
@@ -1055,32 +1172,22 @@ def _expire_delegation(
                 "error": error,
                 "total_duration_seconds": duration,
             },
-            "timeout",
+            "stalled",
         )
     else:
         _push_completion_event(
             event_record,
             {
-                "status": "timeout",
+                "status": "stalled",
                 "summary": None,
                 "error": error,
                 "api_calls": 0,
                 "duration_seconds": duration,
-                "exit_reason": "timeout",
+                "exit_reason": "stalled",
             },
-            "timeout",
+            "stalled",
         )
-    _finish_finalization(delegation_id, "timeout")
-
-    if callable(interrupt_fn):
-        try:
-            interrupt_fn()
-        except Exception as exc:
-            logger.debug(
-                "Async delegation %s timeout interrupt failed: %s",
-                delegation_id,
-                exc,
-            )
+    _finish_finalization(delegation_id, "stalled")
 
 
 def list_async_delegations() -> List[Dict[str, Any]]:
@@ -1093,7 +1200,8 @@ def list_async_delegations() -> List[Dict[str, Any]]:
             {
                 k: v
                 for k, v in r.items()
-                if k not in {"interrupt_fn", "_timeout_timer"}
+                if k not in {"interrupt_fn", "progress_fn"}
+                and not k.startswith("_")
             }
             for r in _records.values()
         ]
@@ -1109,7 +1217,8 @@ def interrupt_all(reason: str = "shutdown") -> int:
     count = 0
     with _records_lock:
         targets = [
-            r for r in _records.values() if r.get("status") == "running"
+            r for r in _records.values()
+            if r.get("status") in ("running", "stalling")
         ]
     for r in targets:
         fn = r.get("interrupt_fn")
@@ -1157,7 +1266,7 @@ def interrupt_for_session(
     with _records_lock:
         targets = [
             r for r in _records.values()
-            if r.get("status") == "running"
+            if r.get("status") in ("running", "stalling")
             and (
                 (origin_ui_session_id and str(r.get("origin_ui_session_id") or "") == origin_ui_session_id)
                 or (session_key and str(r.get("session_key") or "") == session_key)
@@ -1184,19 +1293,18 @@ def interrupt_for_session(
 
 
 def _reset_for_tests() -> None:
-    """Test-only: clear all state and tear down the executor."""
-    global _executor, _executor_max_workers
+    """Test-only: clear all state and tear down the executor + monitor."""
+    global _executor, _executor_max_workers, _monitor_thread
     with _executor_lock:
         if _executor is not None:
             _executor.shutdown(wait=False)
         _executor = None
         _executor_max_workers = 0
+    _monitor_stop.set()
+    with _monitor_lock:
+        thread = _monitor_thread
+        _monitor_thread = None
+    if thread is not None and thread.is_alive():
+        thread.join(timeout=2)
     with _records_lock:
-        timers = [
-            timer
-            for record in _records.values()
-            if (timer := record.get("_timeout_timer")) is not None
-        ]
         _records.clear()
-    for timer in timers:
-        timer.cancel()

tools/delegate_tool.py

@@ -3038,6 +3038,27 @@ def delegate_task(
                 except Exception:
                     pass
 
+        def _batch_progress():
+            # Progress token for the async registry's stale monitor: the
+            # combined (api_call_count, current_tool) of every child. Any
+            # child advancing an iteration or entering/leaving a tool changes
+            # the token; a frozen token past the stale threshold means the
+            # whole detached batch is wedged (e.g. stuck inside the first
+            # model API call — #60203). in_tool=True while ANY child is
+            # inside a tool so legitimately slow tools get the higher
+            # staleness ceiling, mirroring the sync-path heartbeat monitor.
+            parts = []
+            in_tool = False
+            for _c in _child_agents:
+                try:
+                    _summary = _c.get_activity_summary()
+                    _tool = _summary.get("current_tool")
+                    parts.append((_summary.get("api_call_count", 0), _tool))
+                    in_tool = in_tool or bool(_tool)
+                except Exception:
+                    parts.append(None)
+            return tuple(parts), in_tool
+
         _goals = [t["goal"] for t in task_list]
         dispatch = dispatch_async_delegation_batch(
             goals=_goals,
@@ -3057,7 +3078,7 @@ def delegate_task(
             # Reuse the live-transcript directory's id (when created) so the
             # returned delegation_id matches cache/delegation/live/<id>/.
             delegation_id=live_deleg_id,
-            timeout_seconds=_get_child_timeout(),
+            progress_fn=_batch_progress,
         )
 
         if dispatch.get("status") == "dispatched":