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."""