From 46fbb044391a1f8204c8f13f151b4981bc9bd913 Mon Sep 17 00:00:00 2001 From: JeremyJC67 <56396327+JeremyJC67@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:14:34 -0700 Subject: [PATCH] fix(acp): reset the pending grace on in-progress tool-call updates; truthful expiry diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A single long tool call that streamed in-progress tool_call_update notifications past the grace boundary was killed as idle: those updates mutate ToolCallRecord in place, invisible to both the pending-set snapshot and _activity_count. The watchdog now observes a monotonic ACPSession.tool_call_update_count each poll, and any change restarts the pending grace clock alongside pending-set changes — a call that keeps talking defers as long as it talks, while one silent for the full grace still trips. On grace expiry, the raised message and IdleTimeoutDiagnostic now report the truth via additive fields: which pending calls exceeded the grace, when the pending set last changed, when the last update was observed, and how many updates were seen. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DZggU3vNYzr3hfDZJUE7jX --- src/benchflow/acp/runtime.py | 71 ++++++++++++++++++++-- src/benchflow/acp/session.py | 7 +++ src/benchflow/diagnostics.py | 18 +++++- tests/test_acp.py | 113 +++++++++++++++++++++++++++++++++++ 4 files changed, 203 insertions(+), 6 deletions(-) diff --git a/src/benchflow/acp/runtime.py b/src/benchflow/acp/runtime.py index 2fe71b4d4..9b389c541 100644 --- a/src/benchflow/acp/runtime.py +++ b/src/benchflow/acp/runtime.py @@ -845,9 +845,19 @@ def _activity_count() -> int: # completion update lost in transit (e.g. a half-open PTY websocket frame # drop) leaves the call pending forever and would otherwise disarm the # watchdog for the rest of the wall-clock budget (#1061). + # The grace clock restarts whenever the pending set changes OR a + # tool_call_update is observed: in-progress updates mutate ToolCallRecord + # in place — invisible to both the pending-set snapshot and + # _activity_count — and a call still streaming progress is demonstrably + # alive. Only a call that stays pending AND silent for the full grace + # stops deferring the idle path. pending_grace = idle_timeout * 3 pending_snapshot: tuple[str, ...] = () pending_since = last_progress + seen_update_count = session.tool_call_update_count + # Poll-granular observation times, kept for truthful expiry diagnostics. + pending_set_changed_at: datetime | None = None + last_tool_update_at: datetime | None = None # poll_interval considers BOTH idle_timeout and wall-clock timeout so that # short overall budgets don't overshoot (e.g. timeout=30s with default # poll_interval=30s could overshoot 100%). Cap at 30s, floor at 1s. @@ -869,6 +879,14 @@ def _activity_count() -> int: break now = asyncio.get_event_loop().time() cur_count = _activity_count() + # Observe tool_call_update traffic every poll: an in-progress + # update proves the pending call's transport is alive, so it + # restarts the grace clock (but not the idle clock — with no + # pending call, updates alone are not progress). + if session.tool_call_update_count != seen_update_count: + seen_update_count = session.tool_call_update_count + pending_since = now + last_tool_update_at = datetime.now(UTC) if cur_count > last_count: last_progress = now last_activity_at = datetime.now(UTC) @@ -878,20 +896,23 @@ def _activity_count() -> int: # emit no ACP updates until they return, so a >idle_timeout run would # otherwise false-fire the watchdog and discard real work. Treat a # pending tool call as progress, but only within pending_grace of the - # pending set last changing: a call whose completion update was lost - # in transit stays pending forever, and an unbounded deferral would - # disarm the watchdog for the rest of the wall-clock budget (#1061). - # A genuine model-side hang has no pending tool call (the prior tool - # already completed via tool_call_update), so it trips the idle path. + # last pending-set change or observed update: a call whose completion + # update was lost in transit stays pending forever, and an unbounded + # deferral would disarm the watchdog for the rest of the wall-clock + # budget (#1061). A genuine model-side hang has no pending tool call + # (the prior tool already completed via tool_call_update), so it + # trips the idle path. elif session.pending_tool_call_ids(): snapshot = tuple(sorted(session.pending_tool_call_ids())) if snapshot != pending_snapshot: pending_snapshot = snapshot pending_since = now + pending_set_changed_at = datetime.now(UTC) if now - pending_since < pending_grace: last_progress = now last_activity_at = datetime.now(UTC) if now - last_progress >= idle_timeout: + pending_ids = session.pending_tool_call_ids() diag = IdleTimeoutDiagnostic( idle_timeout_sec=idle_timeout, idle_duration_sec=int(now - last_progress), @@ -900,7 +921,47 @@ def _activity_count() -> int: n_message_chunks=len(session.message_chunks), n_thought_chunks=len(session.thought_chunks), last_activity_at=last_activity_at.isoformat(), + pending_tool_call_ids=list(pending_ids), + pending_grace_sec=pending_grace, + pending_set_last_changed_at=( + pending_set_changed_at.isoformat() + if pending_set_changed_at is not None + else None + ), + last_tool_update_at=( + last_tool_update_at.isoformat() + if last_tool_update_at is not None + else None + ), + n_tool_call_updates=session.tool_call_update_count, ) + if pending_ids: + # A non-empty pending set here means the grace expired + # (an in-grace pending call would have deferred + # last_progress this very poll). Report that — not the + # generic "no new tool call, message, or thought" line, + # which is untrue when updates were flowing earlier. + fired_at = datetime.now(UTC) + set_age = ( + f"{int((fired_at - pending_set_changed_at).total_seconds())}s ago" + if pending_set_changed_at is not None + else "unknown" + ) + update_age = ( + f"{int((fired_at - last_tool_update_at).total_seconds())}s ago" + if last_tool_update_at is not None + else "never" + ) + raise IdleTimeoutError( + f"Agent idle for {idle_timeout}s: " + f"{len(pending_ids)} pending tool call(s) exceeded the " + f"{pending_grace}s pending grace " + f"({', '.join(pending_ids)}; pending set last changed " + f"{set_age}, last tool-call update {update_age}, " + f"{session.tool_call_update_count} updates seen, " + f"{len(session.tool_calls)} tool calls so far)", + diag, + ) raise IdleTimeoutError( f"Agent idle for {idle_timeout}s with no new tool call, " f"message, or thought " diff --git a/src/benchflow/acp/session.py b/src/benchflow/acp/session.py index aa66ccdf1..fcfa8f4a4 100644 --- a/src/benchflow/acp/session.py +++ b/src/benchflow/acp/session.py @@ -202,6 +202,12 @@ def __init__(self, session_id: str): self.thought_chunks: list[str] = [] self.tool_calls: list[ToolCallRecord] = [] self._tool_call_map: dict[str, ToolCallRecord] = {} + # Monotonic count of tool_call_update notifications. In-progress + # updates mutate a ToolCallRecord in place — no list grows and the + # pending set is unchanged — so this counter is the only signal the + # idle watchdog has that a long-running call is still streaming + # progress (PR #1066 review). + self.tool_call_update_count: int = 0 # Distinct display titles across recorded tool calls, maintained # incrementally at record creation so the eval dashboard can detect # single-tool agents (prime-agent funnels everything through one @@ -401,6 +407,7 @@ def handle_update(self, update: dict) -> None: self._record_tool_call(record) elif update_type == "tool_call_update": + self.tool_call_update_count += 1 tc_id = update.get("toolCallId", "") record = self._tool_call_map.get(tc_id) if not record: diff --git a/src/benchflow/diagnostics.py b/src/benchflow/diagnostics.py index 3b73cd1d8..e275148d3 100644 --- a/src/benchflow/diagnostics.py +++ b/src/benchflow/diagnostics.py @@ -132,18 +132,34 @@ class IdleTimeoutDiagnostic(Diagnostic): n_message_chunks: int = 0 n_thought_chunks: int = 0 last_activity_at: str = "" + # Pending-grace truth (PR #1066): when the fire happened because a + # pending tool call exhausted the grace window, these say so instead of + # letting the fields above imply pure silence. Additive — older + # result.json dicts round-trip through format_issue_from_dict with the + # defaults below. + pending_tool_call_ids: list[str] = field(default_factory=list) + pending_grace_sec: int | None = None + pending_set_last_changed_at: str | None = None + last_tool_update_at: str | None = None + n_tool_call_updates: int = 0 field: ClassVar[str] = "idle_timeout_info" category: ClassVar[str | None] = "idle_timeout" summary_description: ClassVar[str] = "hit idle timeout" def format_issue(self, task_name: str) -> str: - return ( + line = ( f"{task_name}: idle timeout after " f"{self.idle_duration_sec}s idle " f"({self.n_tool_calls} tool calls, " f"{self.wall_clock_elapsed_sec}s wall)" ) + if self.pending_tool_call_ids: + line += ( + f" — {len(self.pending_tool_call_ids)} pending tool call(s) " + f"exceeded the {self.pending_grace_sec}s pending grace" + ) + return line @dataclass diff --git a/tests/test_acp.py b/tests/test_acp.py index 0da38aa47..6b21fe752 100644 --- a/tests/test_acp.py +++ b/tests/test_acp.py @@ -932,6 +932,119 @@ async def prompt(self, _prompt: str): timeout=20.0, ) + @pytest.mark.asyncio + async def test_streaming_tool_call_updates_defer_past_grace(self): + """A single long tool call that streams in-progress tool_call_update + notifications past the grace boundary is demonstrably alive and must + NOT be idle-killed: updates mutate the ToolCallRecord in place, so + they are invisible to both the pending-set snapshot and + _activity_count — each observed update must reset the grace clock. + The wall-clock backstop (AgentPromptTimeoutError) still bounds it. + + Runtime: ~7s (wall timeout 7 > grace 3 + idle 1, so an unpatched + grace clock false-fires the idle path first). The outer wait_for is + a loose hang guard, not a timing assertion. + """ + from benchflow.acp.runtime import AgentPromptTimeoutError, execute_prompts + + class StreamingToolClient: + def __init__(self, session: ACPSession): + self._session = session + + async def prompt(self, _prompt: str): + self._session.handle_update( + { + "sessionUpdate": "tool_call", + "toolCallId": "tc_stream", + "title": "long training run", + "kind": "bash", + } + ) + while True: # stream progress until cancelled + await asyncio.sleep(0.4) + self._session.handle_update( + { + "sessionUpdate": "tool_call_update", + "toolCallId": "tc_stream", + "status": "in_progress", + } + ) + + session = ACPSession("streaming-grace-session") + with pytest.raises(AgentPromptTimeoutError): + await asyncio.wait_for( + execute_prompts( + StreamingToolClient(session), # type: ignore[arg-type] + session, + ["solve"], + timeout=7, + idle_timeout=1, + ), + timeout=30.0, + ) + # The call outlived the watchdog: still pending, never idle-killed. + assert session.pending_tool_call_ids() == ["tc_stream"] + + @pytest.mark.asyncio + async def test_grace_expiry_reports_pending_truth(self): + """When the grace genuinely expires (call pending, updates stopped), + the raised message and IdleTimeoutDiagnostic must say so — not claim + 'no new tool call, message, or thought' as if the session were + silent all along. + + Runtime: ~4s (2 quick updates, then silence through grace + idle). + """ + from benchflow.acp.runtime import IdleTimeoutError, execute_prompts + + class StallAfterUpdatesClient: + def __init__(self, session: ACPSession): + self._session = session + + async def prompt(self, _prompt: str): + self._session.handle_update( + { + "sessionUpdate": "tool_call", + "toolCallId": "tc_stall", + "title": "flaky build", + "kind": "bash", + } + ) + for _ in range(2): + await asyncio.sleep(0.2) + self._session.handle_update( + { + "sessionUpdate": "tool_call_update", + "toolCallId": "tc_stall", + "status": "in_progress", + } + ) + await asyncio.Future() # completion update lost forever + + session = ACPSession("grace-expiry-session") + with pytest.raises(IdleTimeoutError) as exc_info: + await asyncio.wait_for( + execute_prompts( + StallAfterUpdatesClient(session), # type: ignore[arg-type] + session, + ["solve"], + timeout=60, + idle_timeout=1, + ), + timeout=30.0, + ) + msg = str(exc_info.value) + assert "pending grace" in msg + assert "tc_stall" in msg + info = exc_info.value.diagnostic.to_dict() + assert info["pending_tool_call_ids"] == ["tc_stall"] + assert info["pending_grace_sec"] == 3 # 3x idle_timeout + assert info["n_tool_call_updates"] == 2 + assert info["pending_set_last_changed_at"] is not None + assert info["last_tool_update_at"] is not None + # Existing fields keep their meaning. + assert info["reason"] == "idle_timeout" + assert info["idle_duration_sec"] >= 1 + class TestIdleTimeoutDiagnostics: """Guards ENG-149: idle timeouts must carry structured diagnostics."""