feat(eval): report live progress for long headless sessions

A proposer or benchmark session could run for an hour with nothing in the
log between "proposing…" and its final result, so a wedged run looked
exactly like a working one. The last CI failure spent 66 minutes silently
retrying a dead endpoint before saying so.

A session's stdout is evidence and is only written out after redaction,
so it can never be echoed. Add a stdout_observer hook to run_managed that
sees the stream without copying it anywhere, and a SessionProgress
reporter that prints only what can be derived safely: turn counts, tool
names, API retries, and a heartbeat while the session is quiet. API
retries are called out by name because that is the signature of the
gateway wedging.

Progress goes to stdout so the benchmark sweep's lines reach the log
live through the existing echo_stdout passthrough, rather than as a
bounded stderr tail after the fact.

Co-authored-by: Cursor <cursoragent@cursor.com>

eval/tests/test_session_progress.py

@@ -0,0 +1,103 @@
+"""Live progress reporting for long headless sessions.
+
+The session event stream is evidence and is redacted before it is written
+anywhere, so progress may only report metadata derived from it. These tests pin
+that boundary along with the signals that distinguish work from a wedged run.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import time
+
+from workflow_bench.runner_sessions import SessionProgress
+
+
+def _drain_lines(stream: io.StringIO) -> list[str]:
+    return [line for line in stream.getvalue().splitlines() if line.strip()]
+
+
+def test_progress_reports_turns_and_tool_names_but_never_model_content() -> None:
+    stream = io.StringIO()
+    progress = SessionProgress("gen 0 proposer", stream=stream, heartbeat_s=3600)
+    events = [
+        {"type": "system", "subtype": "init"},
+        {
+            "type": "assistant",
+            "message": {
+                "content": [
+                    {"type": "text", "text": "SECRET-REASONING-abc123"},
+                    {"type": "tool_use", "id": "t1", "name": "Grep", "input": {"pattern": "SECRET-INPUT"}},
+                ]
+            },
+        },
+        {"type": "result", "num_turns": 1, "is_error": False, "total_cost_usd": 1.5},
+    ]
+    for event in events:
+        progress.observe((json.dumps(event) + "\n").encode())
+
+    output = stream.getvalue()
+    assert "SECRET-REASONING-abc123" not in output
+    assert "SECRET-INPUT" not in output
+    assert "session initialized" in output
+    assert "turn 1 · Grep" in output
+    assert "finished · 1 turns · ok · $1.50" in output
+
+
+def test_progress_calls_out_api_retries_because_that_is_the_stuck_signature() -> None:
+    stream = io.StringIO()
+    progress = SessionProgress("proposer", stream=stream, heartbeat_s=3600)
+    event = {
+        "type": "system",
+        "subtype": "api_retry",
+        "attempt": 7,
+        "max_retries": 10,
+        "retry_delay_ms": 34199.87,
+        "error": "unknown",
+    }
+    progress.observe((json.dumps(event) + "\n").encode())
+
+    line = _drain_lines(stream)[-1]
+    assert "API retry 7/10 in 34s" in line
+    assert "no response from the model endpoint" in line
+
+
+def test_progress_speaks_up_while_a_session_is_silent() -> None:
+    stream = io.StringIO()
+    with SessionProgress("proposer", stream=stream, heartbeat_s=0.05):
+        time.sleep(0.35)
+
+    heartbeats = [line for line in _drain_lines(stream) if "still running" in line]
+    assert heartbeats, "a silent session must still report that it is alive"
+    assert "0 turns" in heartbeats[0]
+
+
+def test_progress_survives_partial_chunks_garbage_and_unbounded_lines() -> None:
+    stream = io.StringIO()
+    progress = SessionProgress("proposer", stream=stream, heartbeat_s=3600)
+    payload = json.dumps(
+        {"type": "assistant", "message": {"content": [{"type": "tool_use", "id": "t1", "name": "Bash"}]}}
+    ).encode()
+    # An event split across reads, non-JSON noise, and a huge newline-free run.
+    progress.observe(payload[:10])
+    progress.observe(payload[10:] + b"\nnot json at all\n")
+    progress.observe(b"x" * (4 * 1024 * 1024))
+    progress.observe(b'\n{"type":"result","num_turns":2,"is_error":true}\n')
+
+    output = stream.getvalue()
+    assert "turn 1 · Bash" in output
+    assert "finished · 2 turns · error" in output
+
+
+def test_progress_sanitizes_a_hostile_tool_name() -> None:
+    stream = io.StringIO()
+    progress = SessionProgress("proposer", stream=stream, heartbeat_s=3600)
+    event = {
+        "type": "assistant",
+        "message": {"content": [{"type": "tool_use", "id": "t1", "name": "Bash\nFAKE-LOG-LINE injected"}]},
+    }
+    progress.observe((json.dumps(event) + "\n").encode())
+
+    assert "FAKE-LOG-LINE" not in stream.getvalue()
+    assert len(_drain_lines(stream)) == 1

eval/workflow_bench/evolve.py

@@ -577,6 +577,7 @@ def run_proposer(
     proposal_path: Path,
     evidence_bundle: Path,
     bwrap_bin: Path,
+    progress_label: str | None = None,
 ) -> dict[str, Any]:
     """Run one proposer in confinement and copy only validated outputs out."""
 
@@ -626,6 +627,7 @@ def run_proposer(
                     disable_slash_commands=True,
                     transcript_projects=sandbox.transcript_projects,
                     transcript_cwd=Path("/workspace"),
+                    progress_label=progress_label or "proposer",
                 )
             if not record["ok"]:
                 return record
@@ -1202,6 +1204,7 @@ def _run_generations(
                     proposal_path=gen_dir / "proposal.md",
                     evidence_bundle=bundle,
                     bwrap_bin=bwrap_bin,
+                    progress_label=f"gen {generation} proposer",
                 )
             # Redact any API token echoed into the session record (e.g. an
             # error_detail stderr_tail) before it enters the uploaded artifact.

eval/workflow_bench/process_control.py

@@ -15,7 +15,7 @@ import subprocess
 import sys
 import threading
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import BinaryIO, Literal
@@ -126,6 +126,7 @@ def _drain(
     tail: _TailBuffer,
     capture: _BoundedCapture | None = None,
     echo: BinaryIO | None = None,
+    observer: Callable[[bytes], None] | None = None,
 ) -> None:
     try:
         # read1, not read: on a BufferedReader, read(n) blocks until it has all
@@ -146,6 +147,14 @@ def _drain(
                     echo.flush()
                 except (OSError, ValueError):
                     echo = None
+            if observer is not None:
+                # Progress reporting must never be able to break the drain, and
+                # the drain must keep running even if the observer is broken:
+                # a stalled reader is what deadlocks the child.
+                try:
+                    observer(chunk)
+                except Exception:
+                    observer = None
     except (OSError, ValueError):
         # A forced close is part of the reap path. The terminal result records
         # an actual reap failure; a reader seeing the close is not one itself.
@@ -436,6 +445,7 @@ def _run_managed_inner(
     stdin_data: bytes | None = None,
     capture_stdout_bytes: int | None = None,
     echo_stdout: bool = False,
+    stdout_observer: Callable[[bytes], None] | None = None,
     _ownership_slot: list[tuple[subprocess.Popen[bytes], _WindowsJob | None, int | None]],
 ) -> ManagedProcessResult:
     """Implementation registered with an outer post-spawn ownership guard."""
@@ -484,7 +494,11 @@ def _run_managed_inner(
     stdout_capture = _BoundedCapture(capture_stdout_bytes) if capture_stdout_bytes is not None else None
     echo = getattr(sys.stderr, "buffer", None) if echo_stdout else None
     readers = [
-        threading.Thread(target=_drain, args=(process.stdout, stdout, stdout_capture, echo), daemon=True),
+        threading.Thread(
+            target=_drain,
+            args=(process.stdout, stdout, stdout_capture, echo, stdout_observer),
+            daemon=True,
+        ),
         threading.Thread(target=_drain, args=(process.stderr, stderr), daemon=True),
     ]
     for reader in readers:
@@ -702,6 +716,7 @@ def run_managed(
     stdin_data: bytes | None = None,
     capture_stdout_bytes: int | None = None,
     echo_stdout: bool = False,
+    stdout_observer: Callable[[bytes], None] | None = None,
 ) -> ManagedProcessResult:
     """Run one command with bounded output and owned-tree termination.
 
@@ -709,6 +724,11 @@ def run_managed(
     arrives, so a long child (the benchmark sweep) reports progress in the CI
     log instead of surfacing only its bounded tail after it finishes. Use it
     only for children whose stdout is log text.
+
+    `stdout_observer` sees the same chunks without copying them anywhere, so a
+    child whose stdout is *not* printable (a Claude session's evidence stream)
+    can still report derived progress. The observer runs on the reader thread:
+    it must not block, and raising only disables further calls.
     """
 
     ownership_slot: list[tuple[subprocess.Popen[bytes], _WindowsJob | None, int | None]] = []
@@ -725,6 +745,7 @@ def run_managed(
             stdin_data=stdin_data,
             capture_stdout_bytes=capture_stdout_bytes,
             echo_stdout=echo_stdout,
+            stdout_observer=stdout_observer,
             _ownership_slot=ownership_slot,
         )
     except BaseException:

eval/workflow_bench/runner.py

@@ -425,7 +425,9 @@ def run_arm(
     # rely on ANTHROPIC_API_KEY alone (the sandboxed HOME has no OAuth/
     # keychain state to conflict with it).
     bare = arm == "baseline_nomcp"
+    progress_label = transcript_output_prefix or f"{task.get('id', 'task')}-{arm}"
     common = {
+        "progress_label": progress_label,
         "claude_bin": sandbox.claude_bin,
         "timeout": args.timeout,
         "model": args.model,
@@ -460,7 +462,11 @@ def run_arm(
             plan_prompt.format(task=task["prompt"]),
             worktree,
             expected_skill=expected_skills[0],
-            **{**common, "allowed_tools": allowed_agent_tools(implementation=False)},
+            **{
+                **common,
+                "progress_label": f"{progress_label} plan",
+                "allowed_tools": allowed_agent_tools(implementation=False),
+            },
         )
         sessions.append(plan_session)
         if plan_session["ok"]:
@@ -487,7 +493,11 @@ def run_arm(
                     work_prompt.format(plan=plan_doc.relative_to(worktree)),
                     worktree,
                     expected_skill=expected_skills[1],
-                    **{**common, "allowed_tools": allowed_agent_tools(implementation=True)},
+                    **{
+                        **common,
+                        "progress_label": f"{progress_label} work",
+                        "allowed_tools": allowed_agent_tools(implementation=True),
+                    },
                 )
                 _require_implementation_fingerprint(
                     work_session,

eval/workflow_bench/runner_sessions.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+import contextlib
 import hashlib
 import json
 import math
 import os
 import re
 import stat
+import sys
+import threading
 import time
 from collections.abc import Sequence
 from pathlib import Path, PurePosixPath
@@ -45,6 +48,147 @@ SESSION_TIMEOUT_SECONDS = 5400
 # evidence preflight (evolve._transcript_artifact_metadata) validates against
 # this exact value, so producer and consumer stay pinned to one schema.
 PARENT_EVENT_STREAM_SOURCE = "parent-captured-stream-json"
+# Progress reporting only. A session can work quietly for many minutes, so the
+# reporter also speaks up on its own to distinguish "thinking" from "wedged".
+PROGRESS_HEARTBEAT_SECONDS = 60.0
+MAX_PROGRESS_LINE_BYTES = 1024 * 1024
+_SAFE_TOOL_NAME = re.compile(r"[A-Za-z0-9._:-]{1,64}")
+
+
+def _safe_tool_name(value: Any) -> str:
+    """A tool name is an identifier; anything else is treated as content."""
+
+    match = _SAFE_TOOL_NAME.fullmatch(value.strip()) if isinstance(value, str) else None
+    return match.group(0) if match else "tool"
+
+
+class SessionProgress:
+    """Narrate a live Claude session without ever echoing its output.
+
+    A session's stdout is evidence: it is redacted before anything is written
+    out, so it can never be streamed to the log. This reports only what the
+    parent can derive safely — turn counts, tool names, API retries, and how
+    long the session has been quiet — which is what tells a watcher whether a
+    long run is working or stuck.
+    """
+
+    def __init__(
+        self,
+        label: str,
+        *,
+        stream: Any = None,
+        heartbeat_s: float = PROGRESS_HEARTBEAT_SECONDS,
+    ) -> None:
+        self.label = label
+        self.heartbeat_s = heartbeat_s
+        # stdout, not stderr: the benchmark sweep runs as a child of the
+        # evolution loop, which echoes only the child's stdout as it arrives
+        # (run_managed(echo_stdout=True)). Its stderr surfaces as a bounded
+        # tail after the fact, which is exactly the blind spot this closes.
+        self._stream = stream if stream is not None else sys.stdout
+        self._lock = threading.Lock()
+        self._buffer = bytearray()
+        self._started = time.monotonic()
+        self._last_spoke = self._started
+        self._events = 0
+        self._turns = 0
+        self._tools = 0
+        self._last_activity = "starting"
+        self._timer: threading.Thread | None = None
+        self._done = threading.Event()
+
+    def __enter__(self) -> SessionProgress:
+        self._say(f"started (heartbeat every {self.heartbeat_s:g}s)")
+        self._timer = threading.Thread(target=self._heartbeat, daemon=True)
+        self._timer.start()
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self._done.set()
+        timer, self._timer = self._timer, None
+        if timer is not None:
+            timer.join(timeout=2)
+
+    def _elapsed(self) -> str:
+        seconds = int(time.monotonic() - self._started)
+        return f"{seconds // 60}m{seconds % 60:02d}s"
+
+    def _say(self, message: str) -> None:
+        try:
+            print(f"[{self.label} {self._elapsed()}] {message}", file=self._stream, flush=True)
+        except (OSError, ValueError):
+            return
+        self._last_spoke = time.monotonic()
+
+    def _heartbeat(self) -> None:
+        tick = min(1.0, max(self.heartbeat_s / 2, 0.01))
+        while not self._done.wait(tick):
+            with self._lock:
+                quiet = time.monotonic() - self._last_spoke
+                if quiet < self.heartbeat_s:
+                    continue
+                self._say(
+                    f"still running · {self._events} events · {self._turns} turns · "
+                    f"{self._tools} tool calls · last: {self._last_activity}"
+                )
+
+    def observe(self, chunk: bytes) -> None:
+        """Consume one stdout chunk. Never raises; never blocks on I/O."""
+
+        with self._lock:
+            self._buffer.extend(chunk)
+            # Bound the partial line: a single enormous event must not grow the
+            # buffer without limit just because it has no newline yet.
+            if len(self._buffer) > MAX_PROGRESS_LINE_BYTES:
+                del self._buffer[:-MAX_PROGRESS_LINE_BYTES]
+            while (newline := self._buffer.find(b"\n")) >= 0:
+                line = bytes(self._buffer[:newline])
+                del self._buffer[: newline + 1]
+                self._observe_line(line)
+
+    def _observe_line(self, line: bytes) -> None:
+        if not line.strip():
+            return
+        try:
+            event = json.loads(line.decode("utf-8", errors="replace"))
+        except (json.JSONDecodeError, ValueError):
+            return
+        if not isinstance(event, dict):
+            return
+        self._events += 1
+        kind = event.get("type")
+        if kind == "assistant":
+            self._turns += 1
+            names = [
+                _safe_tool_name(block.get("name"))
+                for block in _event_content(event)
+                if isinstance(block, dict) and block.get("type") == "tool_use"
+            ]
+            if names:
+                self._tools += len(names)
+                self._last_activity = ", ".join(names[:4])
+                self._say(f"turn {self._turns} · {self._last_activity}")
+            else:
+                self._last_activity = "model reply"
+        elif kind == "system" and event.get("subtype") == "api_retry":
+            # The signature of the gateway wedging: say it loudly and at once.
+            attempt = event.get("attempt")
+            limit = event.get("max_retries")
+            delay = event.get("retry_delay_ms")
+            wait = f" in {float(delay) / 1000:.0f}s" if isinstance(delay, (int, float)) else ""
+            self._last_activity = f"API retry {attempt}/{limit}"
+            self._say(f"API retry {attempt}/{limit}{wait} — no response from the model endpoint")
+        elif kind == "system" and event.get("subtype") == "init":
+            self._last_activity = "session init"
+            self._say("session initialized")
+        elif kind == "result":
+            cost = event.get("total_cost_usd")
+            self._last_activity = "result"
+            self._say(
+                f"finished · {event.get('num_turns', 0)} turns · "
+                f"{'error' if event.get('is_error') else 'ok'}"
+                + (f" · ${float(cost):.2f}" if isinstance(cost, (int, float)) else "")
+            )
 
 
 def measured_cost(raw: Any) -> float | None:
@@ -374,6 +518,7 @@ def run_claude(
     transcript_output_prefix: str | None = None,
     transcript_secrets: tuple[str, ...] = (),
     plugin_dirs: Sequence[str] = (),
+    progress_label: str | None = None,
 ) -> dict[str, Any]:
     """Run one headless session and return its usage record."""
 
@@ -420,15 +565,18 @@ def run_claude(
         cmd += ["--disallowedTools", tool]
     managed_cmd = [*(command_prefix or []), *cmd]
     started = time.monotonic()
-    proc = run_managed(
-        managed_cmd,
-        cwd=None if command_prefix else cwd,
-        timeout=timeout,
-        env=env,
-        require_pid_namespace=require_pid_namespace,
-        stdin_data=prompt.encode(),
-        capture_stdout_bytes=MAX_TRANSCRIPT_BYTES,
-    )
+    with contextlib.ExitStack() as progress_stack:
+        progress = progress_stack.enter_context(SessionProgress(progress_label)) if progress_label else None
+        proc = run_managed(
+            managed_cmd,
+            cwd=None if command_prefix else cwd,
+            timeout=timeout,
+            env=env,
+            require_pid_namespace=require_pid_namespace,
+            stdin_data=prompt.encode(),
+            capture_stdout_bytes=MAX_TRANSCRIPT_BYTES,
+            stdout_observer=progress.observe if progress is not None else None,
+        )
     wall_s = time.monotonic() - started
     event_stream_error: str | None = None
     events: list[dict[str, Any]] = []