fix(acp): reset the pending grace on in-progress tool-call updates; truthful expiry diagnostics

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 "

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:

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

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