From 5f5afb1eef1834288be502fb0c7c7c9ce46d9d60 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sun, 26 Jul 2026 16:04:01 -0700 Subject: [PATCH] fix(delegation): count streamed tokens as liveness in the stale monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 40 ++++++++++++++++++++++++++-- tools/delegate_tool.py | 27 +++++++++++++------ 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 8ad9623657e0..09ab7801b51f 100644 --- a/tests/tools/test_async_delegation.py +++ b/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 diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 3c724fa3d206..67963b2cecca 100644 --- a/tools/delegate_tool.py +++ b/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)