diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 9f9746c662..24d2c20122 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -415,6 +415,34 @@ Backend-specific networking stays in the benchmark's native launcher. ## Results, timeout and recovery +### Live session observation + +Frequent capacity and liveness polls may use +`benchmark.runtime.session_observation.observe_codex_session_tail`. It reads at +most the final 16 MiB of a live rollout by default, keeps only complete JSONL +records, and returns content-free event counts, the latest cumulative token +snapshot, timestamps, and framing diagnostics. Structured terminal provider +failures reuse the same record framing through `provider_failure.py`; free-text +errors are not classified as rate limits. The observation does not return prompts, +assistant messages, tool output, error text, session ids, model ids, or local +paths. + +An external controller can call the same path without importing benchmark code: + +```sh +python -m benchmark.runtime.session_observation /explicit/session/rollout.jsonl +``` + +The rollout path is selected by the isolated runner/controller and is never +echoed into the observation. The module is not loaded or invoked by ordinary +workers unless a controller explicitly runs this command. + +This bounded tail is an operational signal, not an accounting or integrity +source. Exact usage booking continues to stream the complete explicitly bound +rollout through `read_codex_session_usage`; trajectory conversion and benchmark +integrity continue to use their existing complete artifacts. A missing usage +event in the bounded tail remains unknown rather than becoming zero. + Private per-wake receipts distinguish process success, Turn settlement and native benchmark results. Sessions are collected once per native filename; resume updates that copy instead of counting the old prefix in every wake. diff --git a/benchmark/runtime/provider_failure.py b/benchmark/runtime/provider_failure.py index cc17851800..859f28d3ef 100644 --- a/benchmark/runtime/provider_failure.py +++ b/benchmark/runtime/provider_failure.py @@ -2,10 +2,13 @@ from __future__ import annotations -import json from collections.abc import Iterator, Mapping from pathlib import Path +from benchmark.runtime.session_observation import ( + CodexSessionObservationError, + iter_codex_session_tail_records, +) _DEFAULT_TAIL_BYTES = 2 * 1024 * 1024 _DEFAULT_MAX_FILES = 32 @@ -26,25 +29,9 @@ def _jsonl_tail(path: Path, *, tail_bytes: int) -> Iterator[Mapping[str, object] if tail_bytes <= 0 or path.is_symlink(): return try: - with path.open("rb") as stream: - stream.seek(0, 2) - size = stream.tell() - start = max(0, size - tail_bytes) - stream.seek(start) - data = stream.read(tail_bytes) - except OSError: + yield from iter_codex_session_tail_records(path, max_bytes=tail_bytes) + except (CodexSessionObservationError, ValueError): return - if start: - _, separator, data = data.partition(b"\n") - if not separator: - return - for line in data.splitlines(): - try: - value = json.loads(line) - except (UnicodeDecodeError, ValueError): - continue - if isinstance(value, Mapping): - yield value def _http_status(value: Mapping[str, object]) -> int | None: diff --git a/benchmark/runtime/session_observation.py b/benchmark/runtime/session_observation.py new file mode 100644 index 0000000000..023240ad6d --- /dev/null +++ b/benchmark/runtime/session_observation.py @@ -0,0 +1,221 @@ +"""Content-free, bounded observations from a live Codex session rollout. + +This is an operational signal for liveness and capacity control. Exact usage +accounting still uses the complete streaming reader owned by +``loopx.control_plane.quota.codex_session_usage``. +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +from collections import Counter +from collections.abc import Iterator, Mapping +from datetime import datetime +from pathlib import Path +from typing import Any + +DEFAULT_SESSION_TAIL_BYTES = 16 * 1024 * 1024 +_USAGE_FIELDS = ( + "input_tokens", + "output_tokens", + "cached_input_tokens", + "reasoning_output_tokens", + "total_tokens", +) +_COUNTED_RESPONSE_TYPES = frozenset( + { + "command_execution", + "custom_tool_call", + "function_call", + "function_call_output", + "local_shell_call", + "message", + "reasoning", + "web_search_call", + } +) + + +class CodexSessionObservationError(RuntimeError): + """The selected live-rollout snapshot could not be observed safely.""" + + +def _complete_tail(path: Path, max_bytes: int) -> tuple[bytes, dict[str, Any]]: + if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or max_bytes <= 0: + raise ValueError("max_bytes must be a positive integer") + try: + with path.open("rb") as stream: + size = os.fstat(stream.fileno()).st_size + start = max(0, size - max_bytes) + # Inspect the preceding byte so an exact record boundary does not + # discard an otherwise complete first record. + stream.seek(max(0, start - 1)) + raw = stream.read(size - max(0, start - 1)) + except OSError as exc: + raise CodexSessionObservationError( + f"cannot read Codex session rollout tail: {type(exc).__name__}" + ) from exc + expected = size - max(0, start - 1) + if len(raw) != expected: + raise CodexSessionObservationError( + "Codex session rollout changed while its tail was read" + ) + + leading_discarded = False + data = raw + if start: + starts_at_boundary = raw[:1] == b"\n" + data = raw[1:] + if not starts_at_boundary: + leading_discarded = True + _partial, separator, data = data.partition(b"\n") + if not separator: + data = b"" + + trailing_discarded = bool(data and not data.endswith(b"\n")) + if trailing_discarded: + prefix, separator, _partial = data.rpartition(b"\n") + data = prefix + separator if separator else b"" + + return data, { + "source_size_bytes": size, + "sampled_bytes": min(size, max_bytes), + "tail_truncated": bool(start), + "leading_record_discarded": leading_discarded, + "trailing_record_discarded": trailing_discarded, + } + + +def _compact_usage(value: Any) -> dict[str, int | float] | None: + if not isinstance(value, Mapping): + return None + compact = { + field: item + for field in _USAGE_FIELDS + if isinstance((item := value.get(field)), (int, float)) + and not isinstance(item, bool) + and math.isfinite(item) + and item >= 0 + } + return compact or None + + +def _timestamp(value: Any) -> str | None: + if not isinstance(value, str) or not value: + return None + candidate = value[:-1] + "+00:00" if value.endswith("Z") else value + try: + datetime.fromisoformat(candidate) + except ValueError: + return None + return value + + +def iter_codex_session_tail_records( + rollout_path: Path, + *, + max_bytes: int = DEFAULT_SESSION_TAIL_BYTES, +) -> Iterator[Mapping[str, Any]]: + """Yield valid objects from complete records in one bounded rollout tail.""" + data, _framing = _complete_tail(Path(rollout_path), max_bytes) + for raw in data.splitlines(): + if not raw.strip(): + continue + try: + item = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + continue + if isinstance(item, Mapping): + yield item + + +def observe_codex_session_tail( + rollout_path: Path, + *, + max_bytes: int = DEFAULT_SESSION_TAIL_BYTES, +) -> dict[str, Any]: + """Observe only complete records from a bounded live-rollout tail. + + The result intentionally contains no prompt, completion, tool output, error + message, local path, session id, or model id. It is suitable for a frequent + benchmark-control poll, but not for billing, exact trajectory conversion, or + integrity qualification. + """ + data, framing = _complete_tail(Path(rollout_path), max_bytes) + counts: Counter[str] = Counter() + latest_usage: dict[str, int | float] | None = None + last_event_at: str | None = None + complete_records = 0 + invalid_records = 0 + + for raw in data.splitlines(): + if not raw.strip(): + continue + complete_records += 1 + try: + item = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + invalid_records += 1 + continue + if not isinstance(item, dict): + continue + payload = item.get("payload") + payload = payload if isinstance(payload, dict) else {} + raw_event_type = payload.get("type") + event_type = raw_event_type if isinstance(raw_event_type, str) else "" + timestamp = _timestamp(item.get("timestamp")) + if timestamp is not None and ( + last_event_at is None or timestamp > last_event_at + ): + last_event_at = timestamp + if item.get("type") == "response_item" and event_type: + counts[ + event_type if event_type in _COUNTED_RESPONSE_TYPES else "other" + ] += 1 + if event_type == "token_count": + info = payload.get("info") + info = info if isinstance(info, dict) else {} + observed = _compact_usage(info.get("total_token_usage")) + if observed is not None: + latest_usage = observed + return { + "schema_version": "codex_session_tail_observation_v0", + **framing, + "event_counts_scope": "bounded_tail", + "event_counts": dict(counts), + "complete_record_count": complete_records, + "invalid_complete_record_count": invalid_records, + "usage_cumulative": latest_usage, + "last_event_at": last_event_at, + } + + +def main(argv: list[str] | None = None) -> int: + """Print one bounded observation for an explicitly selected rollout.""" + parser = argparse.ArgumentParser( + description="Read a content-free bounded tail from one live Codex rollout." + ) + parser.add_argument("rollout", type=Path) + parser.add_argument("--max-bytes", type=int, default=DEFAULT_SESSION_TAIL_BYTES) + options = parser.parse_args(argv) + try: + observation = observe_codex_session_tail( + options.rollout, max_bytes=options.max_bytes + ) + except (CodexSessionObservationError, ValueError) as exc: + parser.exit(2, f"session observation unavailable: {type(exc).__name__}\n") + print( + json.dumps( + observation, + separators=(",", ":"), + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmark/tests/test_session_tail_observation.py b/benchmark/tests/test_session_tail_observation.py new file mode 100644 index 0000000000..df49101b46 --- /dev/null +++ b/benchmark/tests/test_session_tail_observation.py @@ -0,0 +1,192 @@ +"""Live observation must not rescan an unbounded Codex rollout.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +from benchmark.runtime.session_observation import observe_codex_session_tail + + +def _row(kind: str, payload: dict, *, timestamp: str) -> bytes: + return ( + json.dumps( + {"type": kind, "timestamp": timestamp, "payload": payload}, + separators=(",", ":"), + ).encode() + + b"\n" + ) + + +def _usage(total: int, *, timestamp: str) -> bytes: + return _row( + "event_msg", + { + "type": "token_count", + "info": { + "total_token_usage": { + "input_tokens": total, + "output_tokens": 7, + "cached_input_tokens": 3, + } + }, + }, + timestamp=timestamp, + ) + + +def test_large_rollout_observes_latest_usage_from_a_bounded_tail( + tmp_path: Path, +) -> None: + rollout = tmp_path / "rollout.jsonl" + private_text = "SYNTHETIC-CONVERSATION-CONTENT" + with rollout.open("wb") as stream: + stream.write(_usage(10, timestamp="2026-10-31T00:00:00Z")) + for index in range(128): + stream.write( + _row( + "response_item", + {"type": "message", "role": "assistant", "content": private_text}, + timestamp=f"2026-10-31T00:01:{index % 60:02d}Z", + ) + ) + stream.write(_usage(20, timestamp="2026-10-31T00:03:00Z")) + + observation = observe_codex_session_tail(rollout, max_bytes=2048) + + assert observation["tail_truncated"] is True + assert observation["sampled_bytes"] <= 2048 + assert observation["usage_cumulative"] == { + "input_tokens": 20, + "output_tokens": 7, + "cached_input_tokens": 3, + } + assert observation["last_event_at"] == "2026-10-31T00:03:00Z" + assert private_text not in json.dumps(observation) + + +def test_module_entrypoint_emits_the_same_content_free_projection( + tmp_path: Path, +) -> None: + rollout = tmp_path / "rollout.jsonl" + rollout.write_bytes(_usage(42, timestamp="2026-10-31T00:00:00Z")) + + result = subprocess.run( + [ + sys.executable, + "-m", + "benchmark.runtime.session_observation", + "--max-bytes", + "4096", + str(rollout), + ], + check=True, + capture_output=True, + text=True, + ) + + assert json.loads(result.stdout) == observe_codex_session_tail( + rollout, max_bytes=4096 + ) + + +def test_module_entrypoint_does_not_echo_an_unreadable_rollout_path( + tmp_path: Path, +) -> None: + missing = tmp_path / "SYNTHETIC-PRIVATE-PATH.jsonl" + + result = subprocess.run( + [sys.executable, "-m", "benchmark.runtime.session_observation", str(missing)], + capture_output=True, + check=False, + text=True, + ) + + assert result.returncode == 2 + assert "CodexSessionObservationError" in result.stderr + assert str(missing) not in result.stderr + + +def test_partial_boundary_records_are_ignored_without_hiding_complete_usage( + tmp_path: Path, +) -> None: + rollout = tmp_path / "rollout.jsonl" + complete = _usage(42, timestamp="2026-10-31T00:02:00Z") + rollout.write_bytes( + b'{"type":"response_item","payload":{"type":"message","text":"' + + b"x" * 512 + + b'"}}\n' + + complete + + b'{"type":"event_msg","payload":' + ) + + observation = observe_codex_session_tail(rollout, max_bytes=len(complete) + 80) + + assert observation["tail_truncated"] is True + assert observation["leading_record_discarded"] is True + assert observation["trailing_record_discarded"] is True + assert observation["usage_cumulative"]["input_tokens"] == 42 + + +def test_tail_starting_at_record_boundary_keeps_the_first_complete_record( + tmp_path: Path, +) -> None: + rollout = tmp_path / "rollout.jsonl" + retained = _usage(42, timestamp="2026-10-31T00:02:00Z") + rollout.write_bytes( + _row( + "response_item", + {"type": "message", "role": "assistant", "content": "discarded"}, + timestamp="2026-10-31T00:01:00Z", + ) + + retained + ) + + observation = observe_codex_session_tail(rollout, max_bytes=len(retained)) + + assert observation["tail_truncated"] is True + assert observation["leading_record_discarded"] is False + assert observation["complete_record_count"] == 1 + assert observation["usage_cumulative"]["input_tokens"] == 42 + + +def test_tail_smaller_than_one_record_reports_unknown_usage(tmp_path: Path) -> None: + rollout = tmp_path / "rollout.jsonl" + rollout.write_bytes(_usage(42, timestamp="2026-10-31T00:00:00Z")) + + observation = observe_codex_session_tail(rollout, max_bytes=16) + + assert observation["usage_cumulative"] is None + assert observation["complete_record_count"] == 0 + assert observation["leading_record_discarded"] is True + + +def test_malformed_fields_cannot_enter_the_content_free_projection( + tmp_path: Path, +) -> None: + rollout = tmp_path / "rollout.jsonl" + private_text = "SYNTHETIC-PRIVATE-FIELD" + rollout.write_bytes( + _row( + "response_item", + {"type": private_text, "message": private_text}, + timestamp=private_text, + ) + + _row( + "event_msg", + { + "type": "token_count", + "info": {"total_token_usage": {"input_tokens": float("nan")}}, + }, + timestamp="2026-10-31T00:03:00Z", + ) + ) + + observation = observe_codex_session_tail(rollout) + + assert observation["event_counts"] == {"other": 1} + assert observation["usage_cumulative"] is None + assert observation["last_event_at"] == "2026-10-31T00:03:00Z" + assert private_text not in json.dumps(observation)