diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 03246adf..59b5ed46 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -304,8 +304,12 @@ mutes INFO logging while it owns the screen; pair it with The dashboard footer also carries a live token total: completed tasks' trusted telemetry plus every running rollout's live usage (ACP session counters reconciled with the sandbox gateway's live capture), so spend is -visible mid-run. Cost stays completed-tasks-only — `$` comes from the -gateway log imported at scoring time. +visible mid-run. The live figure is a lower bound — it trails the gateway +log by however much the capture has yet to read — and if that tail ever +stops advancing altogether, the run logs one `Live token counter has +stalled` warning so a stale number is never passed off as a current one. +Cost stays completed-tasks-only — `$` comes from the gateway log imported +at scoring time. After the run, each failed task gets one dim `✗ task: reason` line — verifier error first, else a compact reward/metric breakdown, else the diff --git a/src/benchflow/cli/_live_progress.py b/src/benchflow/cli/_live_progress.py index f3d32357..8e560077 100644 --- a/src/benchflow/cli/_live_progress.py +++ b/src/benchflow/cli/_live_progress.py @@ -135,13 +135,21 @@ def _fmt_tokens(n: int) -> str: # while the session counters are unavailable — before the agent session # exists and after it is torn down — so sandbox create, agent install, and # the verifier read as work, not as a hang. +# +# The labels must be monotonic across a run: the row is a progress indicator, +# so stepping back to an earlier stage reads as the run having restarted. +# "executed" maps to the agent stage, not the verifier: since verify() marks +# "verifying" at its own entry, the only stretch that renders under "executed" +# is the inside of disconnect() (session already dropped, agent process being +# killed) — labelling that "verifying…" made the row show "verifying…", then +# "running agent…" (disconnect's "installed"), then "verifying…" again. _PHASE_LABELS = { "created": "creating sandbox…", "setup": "creating sandbox…", "started": "installing agent…", "installed": "running agent…", "connected": "running agent…", - "executed": "verifying…", + "executed": "running agent…", "verifying": "verifying…", "verified": "cleaning up…", "cleaned": "cleaning up…", diff --git a/src/benchflow/providers/litellm_runtime.py b/src/benchflow/providers/litellm_runtime.py index 1b14e953..bde3485b 100644 --- a/src/benchflow/providers/litellm_runtime.py +++ b/src/benchflow/providers/litellm_runtime.py @@ -74,8 +74,34 @@ _PROXY_DOCS_DISABLE_ENV = {"DOCS_URL": "", "NO_DOCS": "true"} _SKILL_CATALOG_GATE_AGENT_ENV = "BENCHFLOW_SKILL_CATALOG_GATE_AGENT" _REQUIRED_SKILL_NAMES_ENV = "BENCHFLOW_REQUIRED_SKILL_NAMES_JSON" -_LIVE_CAPTURE_CHUNK_BYTES = 24 * 1024 +# Live callback-log capture budgets. The reader tails the gateway's +# callback.jsonl by byte range, one command per read, and each read on the +# sandbox path costs a full transient-exec round trip (Daytona: create +# session, run, poll on a 1s interval, fetch logs, delete) — so the per-read +# *yield* is what sets the tail's throughput ceiling, not the poll cadence. +# The shipped 24KB-per-`dd bs=1` read put that ceiling near the rate a +# full-message provider log grows at (#965 follow-up: DeepSeek writes the whole +# conversation per record), and any single failed read silently cost the whole +# tick — which is how the dashboard's token figure froze at a plausible number +# for 20 minutes of a live run. 512KB per read (~700KB of base64 on the wire) +# over 8 reads/tick raises the ceiling by ~20x per round trip while cutting +# round trips per tick from 64 to 8. +_LIVE_CAPTURE_CHUNK_BYTES = 512 * 1024 +_LIVE_CAPTURE_MAX_READS_PER_TICK = 8 +# dd block size inside the sandbox: the range is selected with +# iflag=skip_bytes,count_bytes so the read is a seek plus a handful of +# full-size reads, instead of `bs=1`'s two syscalls per byte. +_LIVE_CAPTURE_READ_BLOCK_BYTES = 64 * 1024 +# Per-read sandbox deadline. A read that trips it advances nothing, so it must +# clear the transport comfortably: Daytona polls session commands on a 1s +# interval and this payload is ~700KB of base64. +_LIVE_CAPTURE_READ_TIMEOUT_SEC = 20 _LIVE_CAPTURE_INTERVAL_SEC = 1.0 +# Consecutive zero-progress polls (~1s each) before the stall is escalated from +# debug to a single WARNING. A capture that is behind but still advancing is +# normal backlog; one that has stopped advancing is the state that renders a +# frozen-but-plausible token count. +_LIVE_CAPTURE_STALL_WARN_TICKS = 30 # Agents that cannot make model calls through LiteLLM. ``oracle`` has no model # at all. Gemini is routable through LiteLLM's native Google GenerateContent @@ -95,6 +121,27 @@ class LiteLLMEndpoint: local_base_url: str +@dataclass(frozen=True) +class _CallbackChunk: + """One byte-ranged read of the gateway's callback log. + + ``size`` is the log's size observed by the same command that produced + ``data``, or None when the read failed. It carries the two facts the + previous bare-``bytes`` return could not express, both of which the live + token counter depends on: + + * an empty read is "caught up" only when the reader can see it is standing + at EOF (``offset >= size``). Without the size, a *failed* read looks + identical to a drained log — which is how a broken tail kept reporting a + stale total as if it were current. + * how far behind the tail is, in bytes, so the lag is a number in the log + rather than an invisible condition. + """ + + data: bytes + size: int | None = None + + class LiteLLMProcess: """Common interface for a running LiteLLM proxy.""" @@ -106,6 +153,12 @@ class LiteLLMProcess: # attribute access even before start_live_capture ever runs). _live_usage_total_tokens: int = 0 _live_usage_seen: bool = False + # Live-capture lag bookkeeping. ``_live_capture_lag_bytes`` is 0 when the + # tail reached EOF on the last poll, a byte count when it is behind, and + # None when it is behind by an unknown amount (the read itself failed). + _live_capture_lag_bytes: int | None = 0 + _live_capture_stall_ticks: int = 0 + _live_capture_stall_warned: bool = False @property def base_url(self) -> str: @@ -137,6 +190,9 @@ def start_live_capture(self, path: Path) -> None: # only observe None (fresh capture) — never a transient "0 tokens". self._live_usage_seen = False self._live_usage_total_tokens = 0 + self._live_capture_lag_bytes = 0 + self._live_capture_stall_ticks = 0 + self._live_capture_stall_warned = False self._live_capture_task = asyncio.create_task(self._live_capture_loop()) def live_usage_tokens(self) -> int | None: @@ -157,7 +213,7 @@ def live_usage_tokens(self) -> int | None: return None return self._live_usage_total_tokens - async def _read_callback_chunk(self, offset: int, limit: int) -> bytes: + async def _read_callback_chunk(self, offset: int, limit: int) -> _CallbackChunk: raise NotImplementedError async def _capture_live_records(self) -> None: @@ -166,14 +222,37 @@ async def _capture_live_records(self) -> None: if trajectory is None or writer is None: return + start_offset = int(getattr(self, "_live_callback_offset", 0)) changed = False - for _ in range(64): + caught_up = False + log_size: int | None = None + for _ in range(_LIVE_CAPTURE_MAX_READS_PER_TICK): offset = int(getattr(self, "_live_callback_offset", 0)) - chunk = await self._read_callback_chunk(offset, _LIVE_CAPTURE_CHUNK_BYTES) - if not chunk: + try: + chunk = await self._read_callback_chunk( + offset, _LIVE_CAPTURE_CHUNK_BYTES + ) + except asyncio.CancelledError: + raise + except Exception as exc: + # A raising read (the sandbox's exec poll hitting its deadline, + # a teardown race) is a lag like any other failed read. Catch it + # HERE rather than letting the loop's blanket handler swallow + # it, so the tick still reaches the lag accounting below instead + # of stalling invisibly — the exact shape of the frozen counter + # this fix exists to surface. + logger.debug("Live LLM callback read failed: %s", exc) + chunk = _CallbackChunk(b"", None) + if chunk.size is not None: + log_size = chunk.size + if not chunk.data: + # Empty means "drained" ONLY when the reader could see the + # log's size and is standing at it. A failed read (size None) + # is a lag, and must not be recorded as having caught up. + caught_up = chunk.size is not None and offset >= chunk.size break - self._live_callback_offset = offset + len(chunk) - data = getattr(self, "_live_callback_remainder", b"") + chunk + self._live_callback_offset = offset + len(chunk.data) + data = getattr(self, "_live_callback_remainder", b"") + chunk.data lines = data.split(b"\n") self._live_callback_remainder = lines.pop() for raw_line in lines: @@ -200,10 +279,78 @@ async def _capture_live_records(self) -> None: ) self._live_usage_seen = True changed = True - if len(chunk) < _LIVE_CAPTURE_CHUNK_BYTES: + if len(chunk.data) < _LIVE_CAPTURE_CHUNK_BYTES: + caught_up = True break if changed: writer.write(trajectory) + self._note_live_capture_lag( + start_offset=start_offset, caught_up=caught_up, log_size=log_size + ) + + def _note_live_capture_lag( + self, *, start_offset: int, caught_up: bool, log_size: int | None + ) -> None: + """Record — and, once it persists, surface — a tail that is behind. + + A poll that ends without reaching EOF is exactly the state that made + the dashboard's token figure lie: the cell keeps rendering a plausible + cumulative number that has silently stopped tracking the run. Every + behind poll is logged at debug with the byte lag; a tail that is behind + AND has stopped advancing altogether earns one WARNING (latched — a + stalled capture must not turn into a per-second log flood), because a + frozen-but-plausible spend figure is worse than a visibly absent one: + users read it to decide whether to kill a run. + + The counter itself is deliberately left alone. It is a cumulative lower + bound at all times (the tail always trails the log by some amount), so + blanking it on a stall would trade a slightly stale number for no + number, and the dashboard already reconciles it as + ``max(acp_snapshot, gateway_live)`` — a None here would just hand the + cell to the other, equally lagging signal. + """ + offset = int(getattr(self, "_live_callback_offset", 0)) + if caught_up: + self._live_capture_lag_bytes = 0 + self._live_capture_stall_ticks = 0 + return + + lag = max(log_size - offset, 0) if log_size is not None else None + self._live_capture_lag_bytes = lag + lag_text = "an unknown number of" if lag is None else str(lag) + if offset > start_offset: + # Behind but still draining: normal backlog after a burst. + self._live_capture_stall_ticks = 0 + logger.debug( + "Live LLM capture is %s bytes behind the gateway log " + "(read through byte %d this poll).", + lag_text, + offset, + ) + return + + self._live_capture_stall_ticks += 1 + logger.debug( + "Live LLM capture made no progress at byte %d (%s bytes behind, " + "%d consecutive stalled polls).", + offset, + lag_text, + self._live_capture_stall_ticks, + ) + if ( + self._live_capture_stall_ticks >= _LIVE_CAPTURE_STALL_WARN_TICKS + and not self._live_capture_stall_warned + ): + self._live_capture_stall_warned = True + logger.warning( + "Live token counter has stalled: the model-gateway log tail " + "has not advanced past byte %d in %d polls (%s bytes behind). " + "The live token/cost figures now UNDERCOUNT this run; the " + "final totals, imported from the full log, are unaffected.", + offset, + self._live_capture_stall_ticks, + lag_text, + ) async def _live_capture_loop(self) -> None: while True: @@ -297,14 +444,19 @@ def _log_size(self) -> int: except OSError: return -1 - async def _read_callback_chunk(self, offset: int, limit: int) -> bytes: - def _read() -> bytes: + async def _read_callback_chunk(self, offset: int, limit: int) -> _CallbackChunk: + def _read() -> _CallbackChunk: try: with self.log_path.open("rb") as handle: + size = os.fstat(handle.fileno()).st_size handle.seek(offset) - return handle.read(limit) + return _CallbackChunk(handle.read(limit), size) + except FileNotFoundError: + # The proxy has not written its first record yet: nothing to + # read, and nothing behind — not a failed read. + return _CallbackChunk(b"", 0) except OSError: - return b"" + return _CallbackChunk(b"", None) return await asyncio.to_thread(_read) @@ -401,9 +553,29 @@ async def _remote_log_size(self) -> int: return int((result.stdout or "-1").strip() or -1) return -1 - async def _read_callback_chunk(self, offset: int, limit: int) -> bytes: + async def _read_callback_chunk(self, offset: int, limit: int) -> _CallbackChunk: + # One command per read, emitting ``\n``: the log + # size rides along for free so the caller can tell "drained" from + # "the read failed" and can quantify the lag without a second exec. + # + # The range is selected with iflag=skip_bytes,count_bytes so dd seeks + # and then reads in 64KB blocks. The previous `bs=1 skip=` + # form made dd issue two syscalls per byte transferred, which — on top + # of a full transient-exec round trip per read — is what held the tail + # to ~24KB per second-plus and let a fast-growing provider log outrun + # it permanently. + # + # Userland floor: `base64 -w 0` on this same line already requires GNU + # coreutils (BusyBox base64 has no -w), so the only new constraint is + # the version — skip_bytes/count_bytes landed in coreutils 8.11 (2011). + # An image that somehow lacks them now fails *visibly*: dd emits + # nothing, the size line still arrives, and the read is classified as a + # lag that escalates to a warning instead of freezing the counter. + path = shlex.quote(self.log_path) command = ( - f"dd if={shlex.quote(self.log_path)} bs=1 skip={offset} count={limit} " + f"stat -c %s {path} 2>/dev/null || echo 0; " + f"dd if={path} iflag=skip_bytes,count_bytes,fullblock " + f"bs={_LIVE_CAPTURE_READ_BLOCK_BYTES} skip={offset} count={limit} " "2>/dev/null | base64 -w 0" ) # Daytona otherwise retains one session wrapper shell for every @@ -412,11 +584,24 @@ async def _read_callback_chunk(self, offset: int, limit: int) -> bytes: # terminal calls. Providers without a transient-exec path keep the # existing behavior. execute = getattr(self.sandbox, "exec_transient", self.sandbox.exec) - result = await execute(command, timeout_sec=10) - encoded = (result.stdout or "").strip() - if result.return_code != 0 or not encoded: - return b"" - return base64.b64decode(encoded, validate=True) + result = await execute(command, timeout_sec=_LIVE_CAPTURE_READ_TIMEOUT_SEC) + if result.return_code != 0: + return _CallbackChunk(b"", None) + head, _, payload = (result.stdout or "").strip().partition("\n") + try: + size = int(head) + except ValueError: + return _CallbackChunk(b"", None) + # Join before validating: the transport may re-wrap a long line, but a + # payload that is genuinely not base64 must stay a failed read rather + # than decode to silent garbage that would desync the byte offset. + encoded = "".join(payload.split()) + if not encoded: + return _CallbackChunk(b"", size) + try: + return _CallbackChunk(base64.b64decode(encoded, validate=True), size) + except ValueError: + return _CallbackChunk(b"", None) async def _load_callback_log(self) -> None: text = "" diff --git a/src/benchflow/rollout/__init__.py b/src/benchflow/rollout/__init__.py index 64e67410..be130b9a 100644 --- a/src/benchflow/rollout/__init__.py +++ b/src/benchflow/rollout/__init__.py @@ -228,6 +228,12 @@ logger = logging.getLogger(__name__) _SETUP_COMMAND_LOCK_SAFE_RE = re.compile(r"[^A-Za-z0-9._-]+") +# Lifecycle phases from verify() onward. The agent will not run again in this +# rollout once one of these is set, so nothing may rewind ``_phase`` out of +# them — the live dashboard renders the phase as a label and a backwards step +# reads as the run having restarted (see disconnect()). +_TERMINAL_PHASES = frozenset({"verifying", "verified", "cleaned"}) + _MCP_TRANSPORT_TO_ACP_TYPE = { "stdio": "stdio", @@ -1384,7 +1390,16 @@ async def disconnect(self) -> None: self._active_role = None self._session_tool_count = 0 self._session_traj_count = 0 - self._phase = "installed" + # Rewinding the phase to "installed" is right for the between-scenes + # disconnect (another agent turn follows, and connect_as() will mark + # "connected"), but disconnect() is ALSO called from cleanup(), after + # verify() has already moved the rollout into its terminal phases. Left + # unguarded, that rewind made the live dashboard walk backwards — + # "verifying…" and then "running agent…" again for the whole teardown + # stretch — and briefly blanked ``Rollout.result``, which is gated on + # the same terminal phases. + if getattr(self, "_phase", None) not in _TERMINAL_PHASES: + self._phase = "installed" def on_ask_user(self, handler: Any) -> None: """Register the agent-initiated ``session/request_permission`` handler. diff --git a/tests/test_cli_live_progress.py b/tests/test_cli_live_progress.py index d3705abe..9f72a3e4 100644 --- a/tests/test_cli_live_progress.py +++ b/tests/test_cli_live_progress.py @@ -434,6 +434,75 @@ async def run() -> None: assert rollout._phase == "verifying" +def test_phase_labels_never_walk_backwards_through_a_run(): + # Fresh-user dogfood follow-up: the row showed "verifying…" and then + # "running agent…" again for the last ~90s of a 26-minute run. The label is + # a progress indicator — an earlier stage reappearing reads as the run + # having restarted, so the label sequence a rollout can produce must be + # monotonic. + from benchflow.cli._live_progress import _PHASE_LABELS + + order = [ + "creating sandbox…", + "installing agent…", + "running agent…", + "verifying…", + "cleaning up…", + ] + lifecycle = [ + "created", + "setup", + "started", + "installed", + "connected", + "executed", + "verifying", + "verified", + "cleaned", + ] + ranks = [order.index(_PHASE_LABELS[phase]) for phase in lifecycle] + assert ranks == sorted(ranks) + # The specific inversion that shipped: verify() marks "verifying" at entry, + # so "executed" only ever renders inside disconnect() — agent teardown, not + # verification. + assert _PHASE_LABELS["executed"] == "running agent…" + + +def test_rollout_disconnect_does_not_rewind_a_terminal_phase(): + # cleanup() calls disconnect() *after* verify(), and the unguarded rewind + # to "installed" relabelled the whole teardown stretch "running agent…" + # (and transiently blanked Rollout.result, which is gated on the same + # phases). Between scenes the rewind is still correct: another agent turn + # follows, and connect_as() re-marks "connected". + import asyncio + + from benchflow.rollout import Rollout + + def _rollout(phase: str) -> SimpleNamespace: + return SimpleNamespace( + _phase=phase, + _is_session_factory=False, + _capture_partial_acp_trajectory=lambda: None, + _acp_client=None, + _session=None, + _session_adapter=None, + _agent_launch="", + _env=None, + _active_role=None, + _session_tool_count=0, + _session_traj_count=0, + ) + + mid_run = _rollout("executed") + asyncio.run(Rollout.disconnect(mid_run)) + assert mid_run._phase == "installed" + + for terminal in ("verifying", "verified", "cleaned"): + done = _rollout(terminal) + asyncio.run(Rollout.disconnect(done)) + assert done._phase == terminal + + def test_progress_enabled_respects_tty_and_optout(monkeypatch): tty = SimpleNamespace(is_terminal=True) notty = SimpleNamespace(is_terminal=False) diff --git a/tests/test_live_llm_trajectory.py b/tests/test_live_llm_trajectory.py index 47fb6c35..7ae3419c 100644 --- a/tests/test_live_llm_trajectory.py +++ b/tests/test_live_llm_trajectory.py @@ -1,6 +1,8 @@ from __future__ import annotations +import contextlib import json +import logging import re from datetime import datetime from types import SimpleNamespace @@ -156,17 +158,28 @@ async def test_host_proxy_mirrors_callback_before_stop(tmp_path, monkeypatch): class _SandboxWithCallbackLog: + """A sandbox whose exec channel serves byte ranges of a canned log. + + Mirrors the wire shape the real reader parses: ``\\n``, + where ``size`` is the whole log (what ``stat -c %s`` reports) and the + payload is only the requested window. + """ + def __init__(self, data: bytes) -> None: self.data = data + self.reads: list[tuple[int, int]] = [] async def exec(self, command: str, timeout_sec: int): del timeout_sec skip = int(re.search(r"skip=(\d+)", command).group(1)) count = int(re.search(r"count=(\d+)", command).group(1)) + self.reads.append((skip, count)) import base64 encoded = base64.b64encode(self.data[skip : skip + count]).decode() - return SimpleNamespace(return_code=0, stdout=encoded) + return SimpleNamespace( + return_code=0, stdout=f"{len(self.data)}\n{encoded}", stderr="" + ) class _TransientSandboxWithCallbackLog(_SandboxWithCallbackLog): @@ -433,5 +446,216 @@ async def test_daytona_proxy_uses_transient_exec_for_callback_poll(tmp_path): chunk = await process._read_callback_chunk(0, 24 * 1024) - assert chunk + assert chunk.data assert sandbox.transient_calls == 1 + + +# --------------------------------------------------------------------------- # +# Capture throughput: the tail must not fall permanently behind a fast log. +# --------------------------------------------------------------------------- # + + +class _RecordingSandbox(_SandboxWithCallbackLog): + """Records the raw command so the read *shape* can be asserted.""" + + def __init__(self, data: bytes) -> None: + super().__init__(data) + self.commands: list[str] = [] + + async def exec_transient(self, command: str, timeout_sec: int): + self.commands.append(command) + return await super().exec(command, timeout_sec) + + +async def _arm_capture(process, path) -> None: + """Arm the live capture but suppress its poller, so a test steps ticks. + + ``start_live_capture`` creates the poll task and returns without awaiting, + so cancelling it here guarantees it never runs a tick — each subsequent + ``_capture_live_records()`` is then exactly one poll, with no timing race. + """ + process.start_live_capture(path) + task = process._live_capture_task + task.cancel() + with contextlib.suppress(runtime_mod.asyncio.CancelledError): + await task + process._live_capture_task = None + + +@pytest.mark.asyncio +async def test_callback_read_is_byte_ranged_with_a_block_sized_transfer(tmp_path): + """The read must select its range by *seeking*, not by copying byte by byte. + + ``dd bs=1 skip= count=`` — the shipped form — issues two + syscalls per byte transferred, and each read costs a whole sandbox exec + round trip on top. That capped the tail's throughput at one small window + per round trip, which a full-message provider log outgrows. Pin the shape: + a byte-ranged transfer with a block size in the KB range, reporting the + log's size alongside the requested window. + """ + payload = bytes(range(256)) * 64 # 16KB of every byte value + sandbox = _RecordingSandbox(payload) + process = _sandbox_process(sandbox) + + chunk = await process._read_callback_chunk(4096, 2048) + + assert chunk.data == payload[4096:6144] + assert chunk.size == len(payload) + command = sandbox.commands[0] + assert "bs=1 " not in command # the per-byte form must be gone + assert "iflag=skip_bytes,count_bytes" in command + assert f"bs={runtime_mod._LIVE_CAPTURE_READ_BLOCK_BYTES}" in command + assert runtime_mod._LIVE_CAPTURE_READ_BLOCK_BYTES >= 4096 + + +@pytest.mark.asyncio +async def test_live_counter_keeps_advancing_when_log_outgrows_a_tick( + tmp_path, monkeypatch +): + """The #965 flatline regression, unit-shaped. + + The dogfood that motivated this fix watched the footer climb to 56.8k + tokens and then hold that exact value for 20 minutes of a live run whose + final total was 1.70M — 3.3% of the truth, rendered as if it were current. + The mechanism: the gateway log grew faster than one poll's read budget + drained it. Drive that state directly — a log growing at ~2x the per-tick + budget — and require the counter to keep advancing every tick and to + converge on the true total once the log stops growing. + """ + monkeypatch.setattr(runtime_mod, "_LIVE_CAPTURE_CHUNK_BYTES", 256) + monkeypatch.setattr(runtime_mod, "_LIVE_CAPTURE_MAX_READS_PER_TICK", 4) + tick_budget = 256 * 4 + + record = _callback_line( + content="x" * 300, + usage={"prompt_tokens": 900, "completion_tokens": 100, "total_tokens": 1000}, + ).encode() + assert len(record) * 3 > tick_budget # the log really does outrun a tick + + sandbox = _RecordingSandbox(record) + process = _sandbox_process(sandbox) + await _arm_capture(process, tmp_path / "trajectory" / "llm_trajectory.jsonl") + + growth_ticks, total_records = 8, 1 + seen: list[int | None] = [] + for tick in range(48): + await process._capture_live_records() + seen.append(process.live_usage_tokens()) + if tick < growth_ticks: + sandbox.data += record * 3 + total_records += 3 + + advancing = [t for t in seen[:growth_ticks] if t is not None] + assert advancing == sorted(advancing) + # The counter must not sit on one value while the log grows — that is the + # frozen-but-plausible figure users read to decide whether to kill a run. + assert len(set(advancing)) > 1 + assert advancing[-1] > advancing[0] + # ...and once growth stops it converges on the log's true total. + assert seen[-1] == total_records * 1000 + # Each round trip carried a full window: draining N bytes must not cost + # more round trips than N/window (+1 for the poll that finds EOF per tick). + max_reads = -(-len(sandbox.data) // 256) + len(seen) + assert len(sandbox.reads) <= max_reads + assert all(count == 256 for _, count in sandbox.reads) + + +class _TimingOutSandbox(_SandboxWithCallbackLog): + """Reads fail the way `timeout ...` kills one: non-zero, no payload.""" + + def __init__(self, data: bytes) -> None: + super().__init__(data) + self.failing = False + + async def exec_transient(self, command: str, timeout_sec: int): + if self.failing: + return SimpleNamespace(return_code=124, stdout="", stderr="") + return await super().exec(command, timeout_sec) + + +class _RaisingSandbox(_TimingOutSandbox): + """The other half of the same failure: the exec poll's own deadline.""" + + async def exec_transient(self, command: str, timeout_sec: int): + if self.failing: + raise RuntimeError("Command timed out after 20 seconds") + return await _SandboxWithCallbackLog.exec(self, command, timeout_sec) + + +@pytest.mark.parametrize("sandbox_cls", [_TimingOutSandbox, _RaisingSandbox]) +@pytest.mark.asyncio +async def test_stalled_reader_is_reported_as_lag_not_as_a_drained_log( + sandbox_cls, tmp_path, monkeypatch, caplog +): + """A read that fails must register as *behind*, loudly — never as caught up. + + The shipped reader returned bare bytes, so a failed read (the sandbox + ``timeout`` killing the command, a truncated transport) was indistinguishable + from a drained log: the poll ended, the offset stayed put, and the token + figure kept rendering its last value with nothing in the logs above debug. + A read that *raised* was worse still — the loop's blanket handler swallowed + the whole tick before any lag was recorded. Now both shapes are a known lag, + and a lag that stops advancing entirely escalates once — and only once — to + a warning. + """ + monkeypatch.setattr(runtime_mod, "_LIVE_CAPTURE_STALL_WARN_TICKS", 3) + + sandbox = sandbox_cls( + _callback_line( + usage={"prompt_tokens": 100, "completion_tokens": 20, "total_tokens": 120} + ).encode() + ) + process = _sandbox_process(sandbox) + await _arm_capture(process, tmp_path / "llm_trajectory.jsonl") + + await process._capture_live_records() + assert process.live_usage_tokens() == 120 + assert process._live_capture_lag_bytes == 0 # drained + + # The log grows; every read now fails. The counter holds its stale value — + # that is unavoidable — but the state is no longer silent. + sandbox.data += _callback_line( + content="second", + usage={"prompt_tokens": 60, "completion_tokens": 20, "total_tokens": 80}, + ).encode() + sandbox.failing = True + with caplog.at_level(logging.WARNING, logger=runtime_mod.logger.name): + for _ in range(6): + await process._capture_live_records() + + assert process.live_usage_tokens() == 120 # frozen, as observed in #965 + assert process._live_capture_lag_bytes is None # behind by an unknown amount + assert process._live_capture_stall_ticks == 6 + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 # latched: a stall must not flood the log + assert "stalled" in warnings[0].getMessage() + + # Recovery clears the signal and the counter catches up. + sandbox.failing = False + await process._capture_live_records() + assert process.live_usage_tokens() == 200 + assert process._live_capture_lag_bytes == 0 + assert process._live_capture_stall_ticks == 0 + + +@pytest.mark.asyncio +async def test_partial_read_of_a_growing_log_is_recorded_as_measured_lag( + tmp_path, monkeypatch +): + """Behind-but-advancing must be quantified, and must not latch a warning.""" + monkeypatch.setattr(runtime_mod, "_LIVE_CAPTURE_CHUNK_BYTES", 128) + monkeypatch.setattr(runtime_mod, "_LIVE_CAPTURE_MAX_READS_PER_TICK", 1) + + record = _callback_line( + content="y" * 400, + usage={"prompt_tokens": 10, "completion_tokens": 10, "total_tokens": 20}, + ).encode() + sandbox = _RecordingSandbox(record) + process = _sandbox_process(sandbox) + await _arm_capture(process, tmp_path / "llm_trajectory.jsonl") + + await process._capture_live_records() + + assert process._live_capture_lag_bytes == len(record) - 128 + assert process._live_capture_stall_ticks == 0 # advancing, so not a stall + assert process._live_capture_stall_warned is False