fix(delegation): count streamed tokens as liveness in the stale monitor

Include last_activity_ts in the progress token sampled from each child.
_touch_activity ticks on every streamed chunk ('receiving stream
response'), every tool transition, and API-call start/completion — so a
child mid-stream on a long response is alive even though api_call_count
only advances when the call completes. Same liveness signal as the
compaction inactivity budget (PR #71508): if tokens are flowing it never
dies; staleness is measured from the last streamed token / tool activity
/ API call.

tests/tools/test_async_delegation.py

@@ -347,6 +347,39 @@ def test_stalling_runner_that_honors_interrupt_keeps_its_result(monkeypatch):
     assert ad.active_count() == 0
 
 
+def test_streaming_child_counts_as_alive(monkeypatch):
+    """A child mid-stream (api_call_count frozen, last_activity_ts ticking)
+    must never be stalled — streamed chunks tick _touch_activity, and the
+    progress token includes that timestamp (same liveness signal as the
+    compaction inactivity budget, PR #71508)."""
+    _fast_stale_monitor(monkeypatch)
+    gate = threading.Event()
+    now = {"ts": 1000.0}
+
+    def progress_fn():
+        # api_call_count and current_tool frozen (long streaming response in
+        # flight), but the activity timestamp advances with every chunk.
+        now["ts"] += 1.0
+        return ((1, None, now["ts"]),), False
+
+    res = ad.dispatch_async_delegation(
+        goal="streaming child", context=None, toolsets=None, role="leaf",
+        model="m", session_key="", max_async_children=1,
+        runner=lambda: (gate.wait(timeout=10), {"status": "completed", "summary": "streamed"})[1],
+        progress_fn=progress_fn,
+    )
+    assert res["status"] == "dispatched"
+
+    time.sleep(0.6)  # several sweeps past the shrunk idle 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_stalled_batch_is_interrupted_then_finalized(monkeypatch):
     _fast_stale_monitor(monkeypatch)
     gate = threading.Event()
@@ -1079,6 +1112,7 @@ def test_delegate_task_background_passes_progress_fn_to_async_registry(monkeypat
     fake_child.get_activity_summary.return_value = {
         "api_call_count": 4,
         "current_tool": "terminal",
+        "last_activity_ts": 1234.5,
     }
 
     creds = {
@@ -1101,11 +1135,13 @@ def test_delegate_task_background_passes_progress_fn_to_async_registry(monkeypat
     assert parsed["status"] == "dispatched"
     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.
+    # async registry's stale monitor can watch the detached batch. The token
+    # includes last_activity_ts so streamed chunks count as liveness (each
+    # chunk ticks _touch_activity), not just completed API calls.
     progress_fn = captured["progress_fn"]
     assert callable(progress_fn)
     token, in_tool = progress_fn()
-    assert token == ((4, "terminal"),)
+    assert token == ((4, "terminal", 1234.5),)
     assert in_tool is True
 
 

tools/delegate_tool.py

@@ -3040,20 +3040,31 @@ def delegate_task(
 
         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.
+            # combined (api_call_count, current_tool, last_activity_ts) of
+            # every child. last_activity_ts is ticked by _touch_activity on
+            # every streamed chunk ("receiving stream response"), every tool
+            # transition, and every API-call start/completion — so a child
+            # streaming a long response is alive even though api_call_count
+            # only advances when the call completes (same liveness signal as
+            # the compaction inactivity budget, PR #71508). A fully frozen
+            # token past the stale threshold means the 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))
+                    parts.append(
+                        (
+                            _summary.get("api_call_count", 0),
+                            _tool,
+                            _summary.get("last_activity_ts"),
+                        )
+                    )
                     in_tool = in_tool or bool(_tool)
                 except Exception:
                     parts.append(None)