From de63b975f534dacb1f991e39c81480ba42356290 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 12 Oct 2026 17:53:58 -0700 Subject: [PATCH 1/6] fix(hooks): decode hook stdin as UTF-8 and tag unreadable payloads (#775) The relay read stdin as text, which on Windows decodes with the locale code page: a UTF-8 transcript path either failed to decode (the whole payload became {}) or came out garbled. Both twins now read the bytes through a shared _read_payload and tag an unreadable payload with payload_error, which HookEvent carries for the wait loop to crumb. --- src/bmad_loop/data/bmad_loop_hook.py | 38 ++++++++++--- src/bmad_loop/events.py | 48 ++++++++++++----- src/bmad_loop/signals.py | 7 +++ tests/test_events.py | 80 ++++++++++++++++++++++++++++ tests/test_signals.py | 22 ++++++++ 5 files changed, 176 insertions(+), 19 deletions(-) diff --git a/src/bmad_loop/data/bmad_loop_hook.py b/src/bmad_loop/data/bmad_loop_hook.py index af769a26..8281153c 100644 --- a/src/bmad_loop/data/bmad_loop_hook.py +++ b/src/bmad_loop/data/bmad_loop_hook.py @@ -358,18 +358,41 @@ def _write_event(events_dir, name, event): os.replace(tmp_path, os.path.join(events_dir, name)) +def _read_payload(stdin): + """The hook payload and why it is unusable: ``(payload, None)`` on success, + else ``({}, tag)``. + + A hook that fires with nothing on stdin, half a JSON document, undecodable + bytes, or a bare list is not an error the operator can act on — the event still + has to be written, because the run's completion signal rides on it. Every + non-dict outcome collapses to ``{}`` and the shaped event carries nulls plus a + ``payload_error`` tag (the exception's class name, or ``not-an-object``) the + orchestrator crumbs, so the degrade stays visible. + + The bytes are read when the stream has a buffer: ``json.loads`` detects + UTF-8/16/32 itself, where a text stdin would decode with the locale code page + (cp1252 on Windows) and garble or reject a non-ASCII transcript path. A + text-only stream (no ``buffer``) is read as text. ``UnicodeDecodeError`` and + ``json.JSONDecodeError`` are both ``ValueError`` subclasses and ride the same + arm; ``OSError`` covers a closed or unreadable descriptor. + """ + buffer = getattr(stdin, "buffer", None) + try: + payload = json.loads(buffer.read()) if buffer is not None else json.load(stdin) + except (ValueError, OSError) as e: + return {}, type(e).__name__ + if not isinstance(payload, dict): + return {}, "not-an-object" + return payload, None + + def main() -> int: run_dir = os.environ.get("BMAD_LOOP_RUN_DIR") task_id = os.environ.get("BMAD_LOOP_TASK_ID") if not run_dir or not task_id: return 0 event_name = sys.argv[1] if len(sys.argv) > 1 else "Unknown" - try: - payload = json.load(sys.stdin) - except (json.JSONDecodeError, ValueError): - payload = {} - if not isinstance(payload, dict): - payload = {} + payload, payload_error = _read_payload(sys.stdin) ts = time.time_ns() event = { @@ -400,6 +423,9 @@ def main() -> int: # a filter — the orchestrator decides what a "mismatch" means. "lineage": _lineage(event_name, ("bmad_loop_hook.py",)), } + # Only when set, so a readable payload's event is unchanged. + if payload_error: + event["payload_error"] = payload_error # The orchestrator's own events dir when it named one, else the legacy # in-tree location this file's older selves are still installed at (see the # module docstring). `or`, not a presence test: an exported-but-empty value diff --git a/src/bmad_loop/events.py b/src/bmad_loop/events.py index 9825e44e..d1821594 100644 --- a/src/bmad_loop/events.py +++ b/src/bmad_loop/events.py @@ -368,13 +368,16 @@ def shape_event( task_id: str, payload: dict[str, Any], lineage: str | None = None, + payload_error: str | None = None, ) -> dict[str, Any]: """The event record the orchestrator consumes, built from one hook payload. ``lineage`` is the relay-side process-lineage tag (DW-507, see :func:`_lineage`); :func:`relay` computes it, since it reads this process's - own parent chain rather than anything in the payload.""" - return { + own parent chain rather than anything in the payload. ``payload_error`` is + :func:`_read_payload`'s tag, written only when set so a readable payload's + event is unchanged.""" + event = { "ts": ts, "event": event_name, "task_id": task_id, @@ -402,24 +405,37 @@ def shape_event( # a filter — the orchestrator decides what a "mismatch" means. "lineage": lineage, } + if payload_error: + event["payload_error"] = payload_error + return event -def _read_payload(stdin: IO[str]) -> dict[str, Any]: - """The hook payload, or an empty dict for anything unreadable. +def _read_payload(stdin): + """The hook payload and why it is unusable: ``(payload, None)`` on success, + else ``({}, tag)``. A hook that fires with nothing on stdin, half a JSON document, undecodable bytes, or a bare list is not an error the operator can act on — the event still has to be written, because the run's completion signal rides on it. Every - non-dict outcome collapses to ``{}`` and the shaped event simply carries nulls. - ``UnicodeDecodeError`` and ``json.JSONDecodeError`` are both ``ValueError`` - subclasses and ride the same arm; ``OSError`` covers a closed or unreadable - descriptor. + non-dict outcome collapses to ``{}`` and the shaped event carries nulls plus a + ``payload_error`` tag (the exception's class name, or ``not-an-object``) the + orchestrator crumbs, so the degrade stays visible. + + The bytes are read when the stream has a buffer: ``json.loads`` detects + UTF-8/16/32 itself, where a text stdin would decode with the locale code page + (cp1252 on Windows) and garble or reject a non-ASCII transcript path. A + text-only stream (no ``buffer``) is read as text. ``UnicodeDecodeError`` and + ``json.JSONDecodeError`` are both ``ValueError`` subclasses and ride the same + arm; ``OSError`` covers a closed or unreadable descriptor. """ + buffer = getattr(stdin, "buffer", None) try: - payload = json.load(stdin) - except (ValueError, OSError): - return {} - return payload if isinstance(payload, dict) else {} + payload = json.loads(buffer.read()) if buffer is not None else json.load(stdin) + except (ValueError, OSError) as e: + return {}, type(e).__name__ + if not isinstance(payload, dict): + return {}, "not-an-object" + return payload, None def relay(event_name: str, stdin: IO[str]) -> int: @@ -443,8 +459,14 @@ def relay(event_name: str, stdin: IO[str]) -> int: if not run_dir or not task_id: return 0 ts = time.time_ns() + payload, payload_error = _read_payload(stdin) event = shape_event( - ts, event_name, task_id, _read_payload(stdin), _lineage(event_name, ("bmad-loop", "relay")) + ts, + event_name, + task_id, + payload, + _lineage(event_name, ("bmad-loop", "relay")), + payload_error, ) # $BMAD_LOOP_EVENTS_DIR when the orchestrator names one (#494 moved the # channel out of the project tree), else the legacy in-tree location — the diff --git a/src/bmad_loop/signals.py b/src/bmad_loop/signals.py index 2e263afe..6a9e9bad 100644 --- a/src/bmad_loop/signals.py +++ b/src/bmad_loop/signals.py @@ -48,6 +48,11 @@ class HookEvent: # that predates the field and for any value outside LINEAGE_TAGS. APPENDED # with a default, like the two fields above. lineage: str | None = None + # Why the relay could not read the hook payload (events._read_payload): the + # exception class name or "not-an-object". None on a readable payload, on an + # older vendored relay that predates the field, and for a non-string value. + # APPENDED with a default, like the fields above. + payload_error: str | None = None # The relays' lineage tags (DW-507, events._lineage). Anything else parses as None. @@ -78,6 +83,7 @@ def _parse_event(entry: Path) -> HookEvent | None: notification_type = data.get("notification_type") source = data.get("source") lineage = data.get("lineage") + payload_error = data.get("payload_error") # Payload values are forwarded from the CLI unvalidated, so a non-string one # reads as absent rather than reaching attribution's set arithmetic (#767). return HookEvent( @@ -90,6 +96,7 @@ def _parse_event(entry: Path) -> HookEvent | None: notification_type=notification_type if isinstance(notification_type, str) else None, source=source if isinstance(source, str) else None, lineage=lineage if isinstance(lineage, str) and lineage in LINEAGE_TAGS else None, + payload_error=payload_error if isinstance(payload_error, str) else None, ) diff --git a/tests/test_events.py b/tests/test_events.py index bf36edc0..62b8a4b9 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -50,6 +50,7 @@ "_is_link_like", "_write_all", "_write_event", + "_read_payload", ) @@ -993,6 +994,85 @@ def test_relay_tolerates_garbage_on_stdin(tmp_path, monkeypatch, capsys, garbage assert json.loads(files[0].read_text())["session_id"] is None +def _run_twin(twin: str, event: str, stdin: bytes, run_dir: Path, **env: str): + """Run one relay twin as a real subprocess, feeding ``stdin`` as raw bytes. + ``twin`` is "hook" (the copied script) or "relay" (``bmad-loop relay``).""" + argv = ( + [sys.executable, str(HOOK), event] + if twin == "hook" + else [sys.executable, "-m", "bmad_loop", "relay", event] + ) + return subprocess.run( + argv, + input=stdin, + # As in test_relay_and_the_hook_shape_the_same_event: `env=` replaces the + # environment, and a Windows child needs SYSTEMROOT to start at all. + env={ + "PATH": os.environ.get("PATH", ""), + **({"SYSTEMROOT": os.environ.get("SYSTEMROOT", "")} if os.name == "nt" else {}), + "BMAD_LOOP_RUN_DIR": str(run_dir), + "BMAD_LOOP_TASK_ID": "t1", + **env, + }, + capture_output=True, + timeout=30, + ) + + +@pytest.mark.parametrize("twin", ["hook", "relay"]) +def test_utf8_payload_survives_cp1252_stdin(tmp_path, twin): + """A Windows host runs the relay without UTF-8 mode, so a text stdin decodes + with the locale code page (cp1252). The CLIs write UTF-8, so a non-ASCII + transcript path either fails to decode (0x81 is undefined in cp1252, so the + whole payload became `{}`) or comes out garbled and names a file that does + not exist. Both twins must read the bytes and recover the exact path. + + Ablation: revert `_read_payload` to `json.load(stdin)` (on either twin) and + that twin's row fails — the event carries a null path.""" + path = str(tmp_path / "Ánna" / "rollout.jsonl") + payload = {"session_id": "s1", "transcript_path": path} + stdin = json.dumps(payload, ensure_ascii=False).encode("utf-8") + assert b"\xc3\x81" in stdin + + proc = _run_twin(twin, "Stop", stdin, tmp_path, PYTHONIOENCODING="cp1252", PYTHONUTF8="0") + + assert proc.returncode == 0, proc.stderr + event = json.loads(next((tmp_path / "events").glob("*.json")).read_text(encoding="utf-8")) + assert event["transcript_path"] == path + assert event["session_id"] == "s1" + assert "payload_error" not in event + + +@pytest.mark.parametrize("twin", ["hook", "relay"]) +@pytest.mark.parametrize( + ("stdin", "expected"), + [ + (b'{"session_id": "s1", "transcript_pa', "JSONDecodeError"), + (b'["s1"]', "not-an-object"), + (b'{"session_id": "s1"}', None), + ], + ids=["truncated", "list", "valid"], +) +def test_unreadable_payload_is_tagged(tmp_path, twin, stdin, expected): + """An unreadable payload still writes the event (the completion signal rides + on it), but the event says WHY it carries nulls, so the orchestrator can crumb + the degrade instead of folding it into an ordinary missing transcript. A + readable payload's event has no `payload_error` key at all. + + Ablation: drop the tag (return `{}, None` from `_read_payload`'s failure + arms) and the truncated/list rows fail on either twin.""" + proc = _run_twin(twin, "Stop", stdin, tmp_path) + + assert proc.returncode == 0, proc.stderr + event = json.loads(next((tmp_path / "events").glob("*.json")).read_text(encoding="utf-8")) + assert event.get("payload_error") == expected + if expected is None: + assert "payload_error" not in event + assert event["session_id"] == "s1" + else: + assert event["session_id"] is None + + class _UnreadableStream: """A stdin `json.load` cannot read. `json.load` calls `fp.read()`, so the failure surfaces from there exactly as a real one would.""" diff --git a/tests/test_signals.py b/tests/test_signals.py index 45264b74..77fab687 100644 --- a/tests/test_signals.py +++ b/tests/test_signals.py @@ -78,6 +78,28 @@ def test_parse_event_keeps_only_a_known_lineage_tag(tmp_path, extra, expected): assert event.lineage == expected +@pytest.mark.parametrize( + ("extra", "expected"), + [ + ({"payload_error": "JSONDecodeError"}, "JSONDecodeError"), + ({"payload_error": "not-an-object"}, "not-an-object"), + ({"payload_error": 1}, None), # a non-string is dropped, not coerced + ({"payload_error": ["JSONDecodeError"]}, None), + ({}, None), # a readable payload (or an older relay) carries no tag + ], +) +def test_parse_event_keeps_string_payload_error_only(tmp_path, extra, expected): + """The relay's unreadable-payload tag lands on the HookEvent (str only), so + the wait loop's crumb never journals a value the relay did not write. + + Ablation: drop the `isinstance(payload_error, str)` check and the non-string + rows fail.""" + watcher = SignalWatcher(tmp_path / "events") + write_event(watcher.events_dir, 1, "t1", "Stop", **extra) + (event,) = watcher.poll() + assert event.payload_error == expected + + @pytest.mark.parametrize("value", [["A", "B"], {"id": "A"}, 3], ids=["list", "dict", "int"]) def test_parse_event_drops_a_non_string_session_id(tmp_path, value): """#767: a non-string id reads as absent, so attribution never hashes it — From 5d46534aeaa029315c07911c482150362f3e1076 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 12 Oct 2026 17:53:58 -0700 Subject: [PATCH 2/6] fix(adapters): resolve a codex rollout by session id when the hook path is unusable (#775) A codex rollout's filename carries the hook session_id, so a missing or unusable hook transcript_path is resolved under $CODEX_HOME/sessions. The resolution sticks against a later unusable hook path, also runs at window death, and is crumbed (transcript-resolved / transcript-unresolved), as is an unreadable hook payload (hook-payload-unreadable). --- CHANGELOG.md | 5 + docs/adapter-authoring-guide.md | 3 + src/bmad_loop/adapters/generic.py | 96 +++++++++- src/bmad_loop/data/profiles/codex.toml | 5 +- src/bmad_loop/tokens.py | 54 +++++- tests/test_generic_tmux.py | 253 +++++++++++++++++++++++++ tests/test_tokens.py | 104 ++++++++++ 7 files changed, 510 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c1f0a141..f7a9a797 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,11 @@ breaking changes may land in a minor release. ### Fixed +- Resolve a codex session's rollout by session id when the hook's transcript path is + missing or unreadable, so its usage is captured and the token budget guard arms + (#775). +- Decode hook stdin as UTF-8 bytes instead of the locale code page, and record an + unreadable hook payload in `session-lifecycle.jsonl` (#775). - Stop failing every opencode-http session start against OpenCode 2.x with a health timeout (#847). - Keep probing an opencode-http session over HTTP after an event-stream reconnect diff --git a/docs/adapter-authoring-guide.md b/docs/adapter-authoring-guide.md index b2dc75f4..4ad680d5 100644 --- a/docs/adapter-authoring-guide.md +++ b/docs/adapter-authoring-guide.md @@ -255,6 +255,9 @@ Turn the report's `token_field_candidates` into a parser in `read_usage`. Decide **per-call vs cumulative** yourself: a `token_count`-style event that carries running totals (codex) is read differently from per-message blocks that are summed (claude, gemini). Re-run the scan; the **parsed counts** self-check should now appear. +At run time the parser reads the file named by the hook payload's `transcript_path`. For +`codex-rollout`, a missing or unusable hook path is resolved from the session id under +`$CODEX_HOME/sessions`. **`none` is a legitimate final answer.** If the report has no `token_field_candidates`, the CLI may not expose usage anywhere a parser can reach. `antigravity` is the example: it diff --git a/src/bmad_loop/adapters/generic.py b/src/bmad_loop/adapters/generic.py index 7f9d7217..3be96c44 100644 --- a/src/bmad_loop/adapters/generic.py +++ b/src/bmad_loop/adapters/generic.py @@ -29,6 +29,7 @@ import enum import hashlib import json +import os import shlex import stat import time @@ -41,7 +42,7 @@ import regex -from .. import devcontract, gates, runs +from .. import devcontract, gates, runs, tokens from ..bmadconfig import ProjectPaths from ..journal import LOGS_DIR, TASK_CYCLE_ARTIFACTS from ..model import TokenUsage @@ -297,6 +298,18 @@ def load_result_document(tasks_dir: Path, task_id: str) -> dict | None: return data +def _is_regular_file(path: str | None) -> bool: + """Whether a hook-reported transcript path names a regular file. Any fault + reads as False: the caller's answer to False is a by-id lookup whose miss is + crumbed, so a metadata fault cannot hide.""" + if not path: + return False + try: + return stat.S_ISREG(Path(path).stat().st_mode) + except (OSError, ValueError): + return False + + class _ResultFileMixin: """Result-file read-back and verdict finalization: acquire the skill-written result dict and fold it into the session's final @@ -990,6 +1003,15 @@ def wait_for_completion(self, handle: SessionHandle, spec: SessionSpec) -> Sessi # (its Stop is dropped), so it is crumbed once instead of staying silent. pin_checked = handle.pinned_session_id is None transcript_path: str | None = None + # A codex-rollout transcript can be found by session id when the hook's + # path is missing or names no file (a Windows relay decoding stdin with + # the locale code page garbles it). `resolved` is that path once found: + # it sticks against a later unusable hook path. `hook_transcript_path` is + # the last raw value a hook sent, for the crumbs. + resolvable = self.profile.usage_parser == "codex-rollout" + resolved: str | None = None + hook_transcript_path: str | None = None + unresolved_crumbed = False nudges_left = self._stop_nudges # Positive grace arms at launch for dev/review sessions, so a CLI that # goes silent before its first Stop cannot burn the full wall timeout. A @@ -1155,6 +1177,37 @@ def sample_transcript(path: str, now: float) -> None: ) scan_failures = 0 + def try_resolve() -> None: + """Resolve the transcript by session id when no readable one is known. + One lookup per call; a hit becomes the loop's transcript (so every + SessionResult and the budget guard see it) and is crumbed once — the + ``resolved is None`` gate makes the hit itself one-shot.""" + nonlocal transcript_path, resolved + if not resolvable or not session_id or resolved is not None: + return + if _is_regular_file(transcript_path): + return + # The pane's env as launch() builds it, over this process's own: the + # CLI reads CODEX_HOME from there. Known limit: a tmux pane inherits + # the tmux SERVER's env, so a server started under another CODEX_HOME + # is missed — and the miss is crumbed at Stop. + hit = tokens.resolve_transcript( + self.profile.usage_parser, + session_id, + {**os.environ, **self.profile.env, **spec.env}, + ) + if hit is None: + return + sample_transcript(str(hit), time.monotonic()) + transcript_path = resolved = str(hit) + self._note_lifecycle( + handle.task_id, + "transcript-resolved", + session_id=session_id, + hook_transcript_path=hook_transcript_path, + path=resolved, + ) + def produced_work() -> bool: # Read at call time, after a final frame sample, so every exit below # reports the loop's final view of the pane rather than the last tick's. @@ -1463,7 +1516,12 @@ def produced_work() -> bool: ) probe_failures = 0 if not alive: - # died without a SessionEnd hook (killed, crashed hard) + # died without a SessionEnd hook (killed, crashed hard). A + # SessionStart that came before its rollout existed, then no + # Stop: resolve once more so the crash still carries the + # transcript. Before the return, not inside `produced_work()`: + # the exit's `transcript_path` argument is read first. + try_resolve() return self._final( handle, spec, @@ -1638,6 +1696,16 @@ def produced_work() -> bool: dropped_so_far=foreign_hook_events, ) continue + if event.payload_error: + # The relay could not read this hook's payload (events._read_payload), + # so the event carries nulls. Completion handling is unchanged; + # the crumb keeps the degrade visible. + self._note_lifecycle( + handle.task_id, + "hook-payload-unreadable", + hook_event=event.event, + payload_error=event.payload_error, + ) if ( event.event == "Stop" and self.profile.subagent_stop_without_transcript @@ -1651,13 +1719,21 @@ def produced_work() -> bool: # real transcript is preserved for usage tallying. continue session_id = event.session_id or session_id - if event.transcript_path and event.transcript_path != transcript_path: + hook_transcript_path = event.transcript_path or hook_transcript_path + hook_path = event.transcript_path + if resolved and hook_path and not _is_regular_file(hook_path): + # A rollout resolved by id sticks: an unusable hook path on a + # later event (the same garbled path a Stop repeats) must not + # overwrite it. + hook_path = None + if hook_path and hook_path != transcript_path: # Take the idle baseline as soon as a hook names a new transcript, # including a later re-point (#680). A write before the next # heartbeat is then a change for the #727 work check rather than # being absorbed into the baseline at exit. - sample_transcript(event.transcript_path, time.monotonic()) - transcript_path = event.transcript_path or transcript_path + sample_transcript(hook_path, time.monotonic()) + transcript_path = hook_path or transcript_path + try_resolve() if event.event == "SessionStart": continue @@ -1686,6 +1762,16 @@ def produced_work() -> bool: # Latched after the subagent filter above, which rejects a stop that # is not the main session's turn-end. Never cleared. stop_seen = True + if resolvable and not unresolved_crumbed and not _is_regular_file(transcript_path): + # Not gated on session_id: a Stop whose payloads all came + # through as `{}` has none, and that miss must show too. + unresolved_crumbed = True + self._note_lifecycle( + handle.task_id, + "transcript-unresolved", + session_id=session_id, + hook_transcript_path=hook_transcript_path, + ) result_json = self._result_json(handle, spec, wait=True) if result_json is not None: if transcript_path: diff --git a/src/bmad_loop/data/profiles/codex.toml b/src/bmad_loop/data/profiles/codex.toml index 426d6ed3..7a9c4ee0 100644 --- a/src/bmad_loop/data/profiles/codex.toml +++ b/src/bmad_loop/data/profiles/codex.toml @@ -2,8 +2,9 @@ # the initial prompt argument; the $skill-name mention syntax attaches the # skill instead (skills live in .agents/skills/) — live-verified 2026-06-11. # "use subagents as needed" is required or Codex runs the skill's parallel -# phases (e.g. review layers) without spawning subagents. Codex has no -# SessionEnd hook event — the adapter's window-death fallback covers crashes. +# phases (e.g. review layers) without spawning subagents. Codex defines +# SessionEnd and SubagentStop hook events, but bmad-loop maps only SessionStart +# and Stop; the adapter's window-death fallback covers crashes. name = "codex" binary = "codex" prompt_template = "Use the ${skill} skill now, and use subagents as needed: {args}" diff --git a/src/bmad_loop/tokens.py b/src/bmad_loop/tokens.py index af7afe08..c6c99f86 100644 --- a/src/bmad_loop/tokens.py +++ b/src/bmad_loop/tokens.py @@ -14,9 +14,14 @@ streams; max is order-independent across files). The dedupe is global because fork subagents copy their parent's message ids. Entries without an id each count. -- codex-rollout: ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl; `token_count` - event payloads carry CUMULATIVE totals, so the last one is - the session usage (sessions here are fresh per task). +- codex-rollout: $CODEX_HOME/sessions/YYYY/MM/DD/rollout--.jsonl + (CODEX_HOME defaults to ~/.codex; day dirs are LOCAL dates + while session_meta timestamps are UTC). The filename UUID is + the hook `session_id`, so when the hook's transcript_path is + missing or unusable the rollout is resolved by id + (`resolve_transcript`). `token_count` event payloads carry + CUMULATIVE totals, so the last one is the session usage + (sessions here are fresh per task). - gemini-chat: ~/.gemini/tmp//chats/session-*.jsonl — a JSONL patch stream: bare message entries (and messages inside "$set" patches) carry a per-API-call `tokens` block @@ -35,6 +40,8 @@ from __future__ import annotations import json +import uuid +from collections.abc import Mapping from pathlib import Path from typing import Any @@ -53,6 +60,47 @@ def read_usage(parser: str, transcript_path: Path) -> TokenUsage | None: return None +def codex_home(env: Mapping[str, str]) -> Path | None: + """Codex's state root: ``CODEX_HOME`` when set and non-empty, else + ``~/.codex``; None when no home directory can be determined.""" + configured = env.get("CODEX_HOME") + if configured: + return Path(configured) + try: + return Path.home() / ".codex" + except RuntimeError: + return None + + +def find_codex_rollout(session_id: str | None, home: Path | None) -> Path | None: + """The rollout file Codex wrote for ``session_id`` under ``home``, or None. + + The id is canonicalised as a UUID before it reaches the glob, so a hostile + or garbled id (``*``, ``../x``) matches nothing. Day directories are local + dates, so every one is searched; on several matches the newest wins.""" + if home is None or session_id is None: + return None + try: + canonical = str(uuid.UUID(session_id)) + except (ValueError, TypeError, AttributeError): + return None + try: + matches = list((home / "sessions").glob(f"*/*/*/rollout-*-{canonical}.jsonl")) + if not matches: + return None + return max(matches, key=lambda p: p.stat().st_mtime) + except OSError: + return None + + +def resolve_transcript(parser: str, session_id: str | None, env: Mapping[str, str]) -> Path | None: + """The transcript ``parser`` reads for ``session_id``, found without the + hook's transcript_path. Only codex-rollout files are resolvable by id.""" + if parser != "codex-rollout": + return None + return find_codex_rollout(session_id, codex_home(env)) + + def _jsonl_entries(transcript_path: Path): if not transcript_path.is_file(): return diff --git a/tests/test_generic_tmux.py b/tests/test_generic_tmux.py index 527f8433..c774dc8c 100644 --- a/tests/test_generic_tmux.py +++ b/tests/test_generic_tmux.py @@ -2287,6 +2287,227 @@ def test_wait_for_completion_keeps_identified_stop_for_stop_only_profile(tmp_pat assert _lifecycle_lines(adapter) == [] +CODEX_SESSION = "019a2b3c-4d5e-7f60-8a9b-0c1d2e3f4a5b" + + +def _codex_home(tmp_path, monkeypatch): + """A CODEX_HOME under tmp_path (HOME/USERPROFILE too, so a resolver that + ignored CODEX_HOME cannot find the operator's real ~/.codex).""" + home = tmp_path / "codex-home" + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + return home + + +def _write_codex_rollout(home, session_id=CODEX_SESSION, input_tokens=500, output_tokens=60): + folder = home / "sessions" / "2026" / "10" / "12" + folder.mkdir(parents=True, exist_ok=True) + path = folder / f"rollout-2026-10-12T00-30-00-{session_id}.jsonl" + lines = [ + {"type": "session_meta", "payload": {"id": session_id}}, + { + "type": "event_msg", + "payload": { + "type": "token_count", + "info": { + "total_token_usage": { + "input_tokens": input_tokens, + "cached_input_tokens": 0, + "output_tokens": output_tokens, + } + }, + }, + }, + ] + path.write_text("".join(json.dumps(line) + "\n" for line in lines), encoding="utf-8") + return path + + +def _codex_dev_adapter(tmp_path, profile_name="codex"): + adapter, impl = make_dev_adapter(tmp_path, profile_name=profile_name) + (impl / "spec-3-1-foo.md").write_text( + "---\nstatus: done\n---\n\n## Auto Run Result\n\nStatus: done\n", encoding="utf-8" + ) + return adapter + + +def test_codex_stop_without_transcript_resolves_rollout(tmp_path, monkeypatch): + """#775's symptom: codex hooks that arrive with no transcript_path (a Windows + relay decoding stdin with cp1252) left the session untracked. The rollout is + found by session id instead, rides the SessionResult, and tallies. + + Ablation: delete the intake `try_resolve()` call and this fails — the result + carries no transcript, and read_usage is None.""" + rollout = _write_codex_rollout(_codex_home(tmp_path, monkeypatch)) + adapter = _codex_dev_adapter(tmp_path) + adapter.watcher = _ScriptedWatcher( + [ + _hook_event("SessionStart", session_id=CODEX_SESSION), + _stop_event("3-1-dev-1", CODEX_SESSION, None), + ] + ) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "completed" + assert Path(result.transcript_path) == rollout + assert adapter.read_usage(result) == TokenUsage(input_tokens=500, output_tokens=60) + (resolved,) = _lifecycle_events(adapter, "transcript-resolved") + assert resolved["session_id"] == CODEX_SESSION + assert resolved["hook_transcript_path"] is None + assert Path(resolved["path"]) == rollout + assert _lifecycle_events(adapter, "transcript-unresolved") == [] + + +def test_codex_garbled_hook_path_is_replaced_and_sticks(tmp_path, monkeypatch): + """A garbled (code-page-mangled) hook path names no file. It is replaced by + the rollout resolved at SessionStart, and the Stop that repeats the SAME + garbled path must not overwrite the resolution. + + Ablation: delete the stickiness guard (`hook_path = None` for an unusable + path once resolved) and this fails — the result carries the garbled path.""" + rollout = _write_codex_rollout(_codex_home(tmp_path, monkeypatch)) + garbled = str(tmp_path / "Ã\x81nna" / "rollout.jsonl") + adapter = _codex_dev_adapter(tmp_path) + adapter.watcher = _ScriptedWatcher( + [ + _hook_event("SessionStart", session_id=CODEX_SESSION, transcript_path=garbled), + _stop_event("3-1-dev-1", CODEX_SESSION, garbled), + ] + ) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "completed" + assert Path(result.transcript_path) == rollout + (resolved,) = _lifecycle_events(adapter, "transcript-resolved") + assert resolved["hook_transcript_path"] == garbled + + +def test_codex_existing_hook_path_wins(tmp_path, monkeypatch): + """A hook path that names a real file is used as-is: the by-id lookup never + runs, so the healthy path (every Linux codex session) costs no glob.""" + rollout = _write_codex_rollout(_codex_home(tmp_path, monkeypatch)) + + def boom(*_a, **_k): + raise AssertionError("resolve_transcript must not run for a readable hook path") + + monkeypatch.setattr(generic.tokens, "resolve_transcript", boom) + adapter = _codex_dev_adapter(tmp_path) + adapter.watcher = _ScriptedWatcher( + [ + _hook_event("SessionStart", session_id=CODEX_SESSION, transcript_path=str(rollout)), + _stop_event("3-1-dev-1", CODEX_SESSION, str(rollout)), + ] + ) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "completed" + assert result.transcript_path == str(rollout) + assert _lifecycle_events(adapter, "transcript-resolved") == [] + assert _lifecycle_events(adapter, "transcript-unresolved") == [] + + +def test_codex_unresolved_stop_is_crumbed_without_session_id(tmp_path, monkeypatch): + """Both payloads collapsed to `{}`: no session id, no path. Nothing can be + resolved, and that must be visible rather than an untracked session with no + record of why. + + Ablation: gate the `transcript-unresolved` crumb on `session_id` and this + fails.""" + _write_codex_rollout(_codex_home(tmp_path, monkeypatch)) + adapter = _codex_dev_adapter(tmp_path) + adapter.watcher = _ScriptedWatcher( + [ + _hook_event("SessionStart", session_id=None), + _stop_event("3-1-dev-1", None, None), + ] + ) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "completed" + assert result.transcript_path is None + (unresolved,) = _lifecycle_events(adapter, "transcript-unresolved") + assert unresolved["session_id"] is None + assert unresolved["hook_transcript_path"] is None + + +def test_codex_window_death_resolves_at_exit(tmp_path, monkeypatch): + """SessionStart fired before the rollout existed and the window died before + any Stop: the window-death exit still resolves, so the crashed result + carries the rollout for usage and salvage. + + Ablation: remove the `try_resolve()` call in the dead-window branch and this + fails.""" + monkeypatch.setattr(generic, "RESULT_GRACE_S", 0.0) + monkeypatch.setattr(generic, "RESULT_POLL_S", 0.0) + home = _codex_home(tmp_path, monkeypatch) + adapter, _ = make_dev_adapter(tmp_path, profile_name="codex") + adapter._window_alive = lambda handle: False + created: list[Path] = [] + + def create_rollout(call_n): + if call_n == 2: + created.append(_write_codex_rollout(home)) + + adapter.watcher = _ScriptedWatcher( + [_hook_event("SessionStart", session_id=CODEX_SESSION)], on_call=create_rollout + ) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "crashed" + assert created + assert Path(result.transcript_path) == created[0] + assert len(_lifecycle_events(adapter, "transcript-resolved")) == 1 + + +def test_non_codex_stop_without_transcript_never_resolves(tmp_path, monkeypatch): + """Scope pin: only codex-rollout transcripts are resolvable by id. A claude + Stop with no path stays untracked and crumbs nothing new, even with a codex + rollout of the same id on disk.""" + _write_codex_rollout(_codex_home(tmp_path, monkeypatch)) + + def boom(*_a, **_k): + raise AssertionError("resolve_transcript must not run for a non-codex profile") + + monkeypatch.setattr(generic.tokens, "resolve_transcript", boom) + adapter = _codex_dev_adapter(tmp_path, profile_name="claude") + adapter.watcher = _ScriptedWatcher([_stop_event("3-1-dev-1", CODEX_SESSION, None)]) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "completed" + assert result.transcript_path is None + assert adapter.read_usage(result) is None + events = {entry["event"] for entry in _lifecycle_lines(adapter)} + assert not events & {"transcript-resolved", "transcript-unresolved"} + + +def test_payload_error_is_crumbed_and_stop_still_completes(tmp_path): + """A Stop whose payload the relay could not read still completes exactly as + before — completion rides on the event, not the payload — and the degrade is + crumbed with the relay's tag. + + Ablation: remove the `hook-payload-unreadable` crumb and this fails.""" + adapter = _codex_dev_adapter(tmp_path, profile_name="claude") + stop = dataclasses.replace( + _stop_event("3-1-dev-1", None, None), payload_error="JSONDecodeError" + ) + adapter.watcher = _ScriptedWatcher([stop]) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "completed" + assert result.stop_seen is True + (crumb,) = _lifecycle_events(adapter, "hook-payload-unreadable") + assert crumb["hook_event"] == "Stop" + assert crumb["payload_error"] == "JSONDecodeError" + + def test_wait_for_completion_keeps_matching_parent_session_end_crash(tmp_path): """The launched session's own SessionEnd (same id as its bound start) is admitted and still crashes the session.""" @@ -3904,6 +4125,38 @@ def spying_tally(parser, path): assert tripped[0]["mode"] == "warn" +def test_budget_guard_arms_after_codex_resolution(tmp_path, monkeypatch): + """A codex session whose hooks carry no transcript path used to leave the + mid-session budget guard blind (it samples the loop's transcript). Resolved + by id at SessionStart, the guard samples the rollout and trips. + + Ablation: delete the intake `try_resolve()` call and this fails — nothing + is sampled, so no trip and budget_weighted is None.""" + home = _codex_home(tmp_path, monkeypatch) + _write_codex_rollout(home, input_tokens=5000, output_tokens=0) + adapter, clock, sent = _budget_adapter(tmp_path, monkeypatch, usage_parser="codex-rollout") + (adapter.tasks_dir / "b-1" / "result.json").write_text('{"ok": true}', encoding="utf-8") + start = HookEvent( + ts=1, + event="SessionStart", + task_id="b-1", + session_id=CODEX_SESSION, + transcript_path=None, + path=Path("x"), + ) + adapter.watcher = _ScriptedWatcher( + [start, None, None, _stop_event("b-1", CODEX_SESSION, None)], + on_call=_advance_31(clock), + ) + + result = adapter.wait_for_completion(_budget_handle(), _budget_spec(tmp_path, mode="warn")) + + assert result.status == "completed" + assert result.budget_weighted == 5000 + tripped = [ln for ln in _lifecycle_lines(adapter, "b-1") if ln["event"] == "budget-tripped"] + assert len(tripped) == 1 + + def test_budget_enforce_nudges_then_terminates_over_budget(tmp_path, monkeypatch): """Enforce mode: BUDGET_NUDGE_TEXT at trip, then grace expiry under a live window ends over_budget WITHOUT accepting the on-disk artifact (#48/#53).""" diff --git a/tests/test_tokens.py b/tests/test_tokens.py index 1f83fbec..a46081c9 100644 --- a/tests/test_tokens.py +++ b/tests/test_tokens.py @@ -1,5 +1,9 @@ import json +import os +import pytest + +from bmad_loop import tokens from bmad_loop.model import TokenUsage from bmad_loop.tokens import ( read_usage, @@ -311,6 +315,106 @@ def test_codex_rollout_without_token_counts_is_none(tmp_path): assert tally_codex_rollout(tmp_path / "nope.jsonl") is None +SESSION = "019a2b3c-4d5e-7f60-8a9b-0c1d2e3f4a5b" + + +def _codex_rollout(home, day, session_id, *, utc="2026-10-11T23:30:00.000Z", output=7): + """A minimal rollout at ``home/sessions//rollout--.jsonl``. + ``day`` is the LOCAL date dir, independent of the UTC session_meta stamp.""" + folder = home / "sessions" / day + folder.mkdir(parents=True, exist_ok=True) + path = folder / f"rollout-{day.replace('/', '-')}T00-30-00-{session_id}.jsonl" + lines = [ + {"timestamp": utc, "type": "session_meta", "payload": {"id": session_id}}, + { + "type": "event_msg", + "payload": { + "type": "token_count", + "info": {"total_token_usage": {"input_tokens": 50, "output_tokens": output}}, + }, + }, + ] + path.write_text("".join(json.dumps(line) + "\n" for line in lines), encoding="utf-8") + return path + + +def test_resolve_codex_rollout_by_session_id(tmp_path): + """The rollout sits in the LOCAL-date dir (10/12) while its session_meta is + stamped the previous UTC day: the lookup globs every day dir rather than + deriving one from a timestamp, and the resolved file tallies. + + Ablation: make `find_codex_rollout` return None and this fails.""" + path = _codex_rollout(tmp_path, "2026/10/12", SESSION, utc="2026-10-11T23:30:00.000Z") + env = {"CODEX_HOME": str(tmp_path)} + + assert tokens.resolve_transcript("codex-rollout", SESSION, env) == path + # Any UUID spelling canonicalises to the filename's. + assert tokens.resolve_transcript("codex-rollout", SESSION.upper(), env) == path + assert read_usage("codex-rollout", path) == TokenUsage(input_tokens=50, output_tokens=7) + + +@pytest.mark.parametrize("session_id", ["*", "../x", "", None, "abc"]) +def test_resolve_rejects_non_uuid_ids(tmp_path, session_id): + """A hook payload is unvalidated, so an id that is not a UUID never reaches + the glob — "*" would otherwise match the sentinel rollout below. + + Ablation: drop the `uuid.UUID` canonicalisation (glob the raw id) and the + "*" row fails.""" + _codex_rollout(tmp_path, "2026/10/12", SESSION) + env = {"CODEX_HOME": str(tmp_path)} + assert tokens.resolve_transcript("codex-rollout", session_id, env) is None + + +def test_resolve_ignores_other_uuids_and_parsers(tmp_path): + """Only the session's own rollout resolves, and only for codex-rollout: a + claude or gemini transcript is never looked up by id. + + Ablation: drop the parser check in `resolve_transcript` and the + claude-jsonl row resolves the codex file.""" + other = "019a2b3c-0000-7000-8000-000000000000" + _codex_rollout(tmp_path, "2026/10/12", other) + path = _codex_rollout(tmp_path, "2026/10/12", SESSION) + env = {"CODEX_HOME": str(tmp_path)} + + assert tokens.resolve_transcript("codex-rollout", SESSION, env) == path + missing = "019a2b3c-ffff-7fff-8fff-ffffffffffff" + assert tokens.resolve_transcript("codex-rollout", missing, env) is None + for parser in ("claude-jsonl", "gemini-chat", "copilot-events", ""): + assert tokens.resolve_transcript(parser, SESSION, env) is None + + +def test_resolve_defaults_to_home_dot_codex(tmp_path, monkeypatch): + """Without CODEX_HOME (or with it empty) the lookup roots at ~/.codex, read + at call time. HOME and USERPROFILE are both patched: Windows' Path.home() + reads USERPROFILE, never HOME. + + Ablation: drop the `Path.home() / ".codex"` default (return None) and this + fails.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("CODEX_HOME", raising=False) + path = _codex_rollout(tmp_path / ".codex", "2026/10/12", SESSION) + + assert tokens.resolve_transcript("codex-rollout", SESSION, {}) == path + assert tokens.resolve_transcript("codex-rollout", SESSION, {"CODEX_HOME": ""}) == path + + +def test_resolve_newest_mtime_wins(tmp_path): + """Two rollouts with one id (a resumed session written into a later day dir) + resolve to the newer by mtime, whatever order the glob returns them in. + + Ablation: take the first glob match instead of the max mtime and one of the + two assertions fails, whichever order the glob yields.""" + older = _codex_rollout(tmp_path, "2026/10/11", SESSION) + newer = _codex_rollout(tmp_path, "2026/10/12", SESSION) + os.utime(older, (2_000_000_000, 2_000_000_000)) + os.utime(newer, (1_000_000_000, 1_000_000_000)) + env = {"CODEX_HOME": str(tmp_path)} + assert tokens.resolve_transcript("codex-rollout", SESSION, env) == older + os.utime(newer, (3_000_000_000, 3_000_000_000)) + assert tokens.resolve_transcript("codex-rollout", SESSION, env) == newer + + def test_gemini_chat_dedupes_reemitted_messages(tmp_path): # shape captured from a real ~/.gemini/tmp//chats/session-*.jsonl # (2026-06-11): a JSONL patch stream where the same message id is From 217dddddebcb3a8771c6d3b00e7ef380b84e2044 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 12 Oct 2026 18:04:14 -0700 Subject: [PATCH 3/6] fix(adapters): count codex subagent rollouts in session usage (#775) Codex writes each subagent to its own rollout and leaves its usage out of the parent's total_token_usage, so the codex tally under-counted any session that spawned subagents. tally_codex_rollout now adds each child's last cumulative totals: children are rollouts in the parent's day dir and later ones (midnight crossings) whose first-line session_meta names the parent as session_id, or whose parent_thread_id chains into the tree. First-line facts are cached per path (bounded, oldest-first), so the budget guard's per-heartbeat tally re-reads only file bodies. --- CHANGELOG.md | 5 + src/bmad_loop/tokens.py | 202 ++++++++++++++++++++++++++++-- tests/test_tokens.py | 271 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 471 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f7a9a797..82e1b494 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -58,6 +58,11 @@ breaking changes may land in a minor release. (#775). - Decode hook stdin as UTF-8 bytes instead of the locale code page, and record an unreadable hook payload in `session-lifecycle.jsonl` (#775). +- Count codex subagent rollouts in session usage. Codex writes each subagent to its + own rollout and leaves its usage out of the parent's totals, so subagent-heavy + sessions were under-counted by half or more. Codex totals and budgets can roughly + double for those sessions, so budgets tuned on old figures may trip differently + (#775). - Stop failing every opencode-http session start against OpenCode 2.x with a health timeout (#847). - Keep probing an opencode-http session over HTTP after an event-stream reconnect diff --git a/src/bmad_loop/tokens.py b/src/bmad_loop/tokens.py index c6c99f86..7d65062c 100644 --- a/src/bmad_loop/tokens.py +++ b/src/bmad_loop/tokens.py @@ -21,7 +21,14 @@ missing or unusable the rollout is resolved by id (`resolve_transcript`). `token_count` event payloads carry CUMULATIVE totals, so the last one is the session usage - (sessions here are fresh per task). + (sessions here are fresh per task). Each subagent writes its + own rollout, and its usage is NOT part of the parent's + totals: a sibling whose first-line session_meta names the + parent as its `session_id` (the root thread, at any depth), + or whose `parent_thread_id` chains into the parent's tree, + adds its own last cumulative totals. Siblings are sought in + the parent's day dir and later ones (a session can cross + midnight), from the parent's filename timestamp on. - gemini-chat: ~/.gemini/tmp//chats/session-*.jsonl — a JSONL patch stream: bare message entries (and messages inside "$set" patches) carry a per-API-call `tokens` block @@ -39,7 +46,10 @@ from __future__ import annotations +import datetime import json +import re +import threading import uuid from collections.abc import Mapping from pathlib import Path @@ -210,7 +220,10 @@ def tally(transcript_path: Path) -> TokenUsage | None: # ----------------------------------------------------------- codex-rollout -def tally_codex_rollout(transcript_path: Path) -> TokenUsage | None: +def _codex_rollout_total(transcript_path: Path) -> dict[str, Any] | None: + """The rollout's last cumulative `token_count` totals, or None when it has + none. A subagent's rollout holds only its own turns, so this rule holds + for children as well as for the parent. OSError/ValueError propagate.""" last: dict[str, Any] | None = None for entry in _jsonl_entries(transcript_path): payload = entry.get("payload") @@ -221,17 +234,192 @@ def tally_codex_rollout(transcript_path: Path) -> TokenUsage | None: continue totals = info.get("total_token_usage") last = totals if isinstance(totals, dict) else info - if last is None: - return None - cached = _int(last.get("cached_input_tokens")) + return last + + +def _codex_usage(totals: dict[str, Any]) -> TokenUsage: + cached = _int(totals.get("cached_input_tokens")) return TokenUsage( # Codex's input_tokens includes the cached portion; split it out. - input_tokens=max(0, _int(last.get("input_tokens")) - cached), - output_tokens=_int(last.get("output_tokens")), + input_tokens=max(0, _int(totals.get("input_tokens")) - cached), + output_tokens=_int(totals.get("output_tokens")), cache_read_tokens=cached, ) +# A session_meta line is ~18-23 KB (it embeds base_instructions). A membership +# test reads at most this much of a sibling, never its body. +_CODEX_META_READ_CAP = 1 << 20 + +# (id, session_id, parent_thread_id) from a rollout's first-line session_meta, +# or None when the file does not open with one. session_meta is written once and +# never changes, so every answer, negative included, is cached by path for +# good; only a first line still being written (no newline yet) is re-read. +# Bounded oldest-first like opencode_http's USAGE_STASH_CAP. A day dir holds a +# few hundred rollouts and a tally lists the parent's day plus the next, so the +# cap sits far above one tally's candidate set: the budget guard's per-heartbeat +# tally never evicts its own siblings, and re-reads only the files' bodies. +CODEX_META_CACHE_CAP = 4096 +_CodexMeta = tuple[str, str | None, str | None] +_codex_meta_cache: dict[str, _CodexMeta | None] = {} +_codex_meta_lock = threading.Lock() + +_ROLLOUT_TS = re.compile(r"rollout-(\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2})") + + +def _read_codex_first_line(path: Path) -> bytes: + with path.open("rb") as f: + return f.readline(_CODEX_META_READ_CAP) + + +def _parse_codex_meta(line: bytes) -> _CodexMeta | None: + try: + entry = json.loads(line) + except ValueError: + return None + if not isinstance(entry, dict) or entry.get("type") != "session_meta": + return None + payload = entry.get("payload") + if not isinstance(payload, dict): + return None + + def text(value: Any) -> str | None: + return value if isinstance(value, str) and value else None + + thread_id = text(payload.get("id")) + if thread_id is None: + return None + parent = text(payload.get("parent_thread_id")) + if parent is None: + # Older rollouts name the parent only in the spawn source. + source = payload.get("source") + subagent = source.get("subagent") if isinstance(source, dict) else None + spawn = subagent.get("thread_spawn") if isinstance(subagent, dict) else None + parent = text(spawn.get("parent_thread_id")) if isinstance(spawn, dict) else None + return thread_id, text(payload.get("session_id")), parent + + +def _codex_meta(path: Path) -> _CodexMeta | None: + """The rollout's session_meta facts, cached (see CODEX_META_CACHE_CAP). + OSError propagates.""" + key = str(path) + with _codex_meta_lock: + if key in _codex_meta_cache: + return _codex_meta_cache[key] + line = _read_codex_first_line(path) + meta = _parse_codex_meta(line) + if line.endswith(b"\n") or len(line) >= _CODEX_META_READ_CAP: + with _codex_meta_lock: + if key not in _codex_meta_cache: + while len(_codex_meta_cache) >= CODEX_META_CACHE_CAP: + del _codex_meta_cache[next(iter(_codex_meta_cache))] + _codex_meta_cache[key] = meta + return meta + + +def _codex_day(folder: Path) -> datetime.date | None: + try: + return datetime.date( + int(folder.parent.parent.name), int(folder.parent.name), int(folder.name) + ) + except ValueError: + return None + + +def _codex_sibling_candidates(transcript_path: Path) -> list[Path]: + """Rollouts that may be the parent's subagents: `rollout-*.jsonl` in the + parent's day dir and each later day dir through the day after the parent's + last write (capped at local today + 1; dirs are local dates and a child + can start after midnight), whose filename timestamp is not earlier than + the parent's. A parent outside a YYYY/MM/DD tree searches its own dir.""" + folder = transcript_path.parent + first = _codex_day(folder) + folders = [folder] + if first is not None: + today = datetime.date.today() + try: + written = datetime.date.fromtimestamp(transcript_path.stat().st_mtime) + except (OSError, ValueError, OverflowError): + written = today + last = max(first, min(written, today)) + datetime.timedelta(days=1) + sessions = folder.parent.parent.parent + folders = [] + day = first + while day <= last: + folders.append(sessions / f"{day.year:04d}" / f"{day.month:02d}" / f"{day.day:02d}") + day += datetime.timedelta(days=1) + stamp = _ROLLOUT_TS.match(transcript_path.name) + floor = stamp.group(1) if stamp else "" + candidates: list[Path] = [] + for each in folders: + for path in sorted(each.glob("rollout-*.jsonl")): + match = _ROLLOUT_TS.match(path.name) + if path != transcript_path and (match.group(1) if match else "") >= floor: + candidates.append(path) + return candidates + + +def tally_codex_rollout(transcript_path: Path) -> TokenUsage | None: + """The session's usage: the parent rollout's last cumulative totals plus + each subagent rollout's own (Codex keeps a child's usage out of the + parent's totals). + + A parent whose first line is not a session_meta tallies alone. A child is + a sibling (see `_codex_sibling_candidates`) whose session_meta names the + parent as its `session_id` (the root thread, at any depth), or whose + `parent_thread_id` chains into the parent's tree (children that predate + `session_id`), iterated to a fixpoint. One thread written to two files + counts once, from the later file. A child's usage makes the session + tracked even when the parent reports none. An unreadable sibling + (OSError/ValueError) is skipped; parent errors propagate as before, and + None still means no usage signal anywhere.""" + parent_totals = _codex_rollout_total(transcript_path) + try: + parent_meta = _codex_meta(transcript_path) + except OSError: + parent_meta = None + totals: list[dict[str, Any]] = [] if parent_totals is None else [parent_totals] + if parent_meta is not None: + root = parent_meta[0] + metas: dict[str, _CodexMeta] = {} + try: + candidates = _codex_sibling_candidates(transcript_path) + except OSError: + candidates = [] + for path in candidates: + try: + meta = _codex_meta(path) + except OSError: + continue + if meta is not None and meta[0] != root: + metas[str(path)] = meta + tree = {root} + members = {path for path, meta in metas.items() if meta[1] == root} + tree.update(metas[path][0] for path in members) + grew = True + while grew: + grew = False + for path, (thread_id, _, parent) in metas.items(): + if path not in members and parent in tree: + members.add(path) + tree.add(thread_id) + grew = True + by_thread = {metas[path][0]: Path(path) for path in sorted(members)} + for child in by_thread.values(): + try: + child_totals = _codex_rollout_total(child) + except (OSError, ValueError): + continue + if child_totals is not None: + totals.append(child_totals) + if not totals: + return None + usage = TokenUsage() + for each in totals: + usage.add(_codex_usage(each)) + return usage + + # ------------------------------------------------------------- gemini-chat diff --git a/tests/test_tokens.py b/tests/test_tokens.py index a46081c9..075c57f7 100644 --- a/tests/test_tokens.py +++ b/tests/test_tokens.py @@ -1,3 +1,4 @@ +import datetime import json import os @@ -415,6 +416,276 @@ def test_resolve_newest_mtime_wins(tmp_path): assert tokens.resolve_transcript("codex-rollout", SESSION, env) == newer +ROOT = "019a0000-0000-7000-8000-000000000001" +CHILD = "019a0000-0000-7000-8000-000000000002" +GRANDCHILD = "019a0000-0000-7000-8000-000000000003" +OTHER = "019a0000-0000-7000-8000-000000000004" + + +def _local_ts(day, hour=12, minute=0): + year, month, dom = (int(part) for part in day.split("/")) + return datetime.datetime(year, month, dom, hour, minute).timestamp() + + +def _codex_thread(home, day, clock, thread_id, *, session_id=None, parent=None, usage=None): + """A rollout at ``home/sessions//rollout-T-.jsonl`` + whose session_meta carries the given tree fields, plus one cumulative + token_count when ``usage`` = (input, cached, output). Its mtime is pinned + to noon of its LOCAL ``day``, which bounds the sibling search window.""" + payload = {"id": thread_id} + if session_id is not None: + payload["session_id"] = session_id + if parent is not None: + payload["parent_thread_id"] = parent + lines = [{"type": "session_meta", "payload": payload}] + if usage is not None: + input_tokens, cached, output = usage + totals = { + "input_tokens": input_tokens, + "cached_input_tokens": cached, + "output_tokens": output, + } + lines.append( + { + "type": "event_msg", + "payload": {"type": "token_count", "info": {"total_token_usage": totals}}, + } + ) + path = home / "sessions" / day / f"rollout-{day.replace('/', '-')}T{clock}-{thread_id}.jsonl" + _write_jsonl(path, lines) + stamp = _local_ts(day) + os.utime(path, (stamp, stamp)) + return path + + +def test_codex_tally_sums_subagent_rollouts(tmp_path): + """Codex keeps each subagent's usage in its own rollout, out of the + parent's totals: a depth-1 and a depth-2 child (both naming the root as + session_id) add their own last cumulative totals; an unrelated rollout + from the same day does not. + + Ablation: drop the `session_id == root` filter (every sibling a member) + and the unrelated rollout is counted.""" + parent = _codex_thread( + tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(100, 40, 10) + ) + _codex_thread( + tmp_path, + "2026/10/01", + "10-05-00", + CHILD, + session_id=ROOT, + parent=ROOT, + usage=(50, 10, 5), + ) + _codex_thread( + tmp_path, + "2026/10/01", + "10-06-00", + GRANDCHILD, + session_id=ROOT, + parent=CHILD, + usage=(20, 0, 2), + ) + _codex_thread( + tmp_path, "2026/10/01", "10-07-00", OTHER, session_id=OTHER, usage=(1000, 0, 1000) + ) + + assert tally_codex_rollout(parent) == TokenUsage( + input_tokens=60 + 40 + 20, output_tokens=17, cache_read_tokens=50 + ) + + +def test_codex_tally_child_across_midnight(tmp_path): + """Day dirs are local dates, so a child spawned after midnight lands in + the next day's dir; the parent's last write (00:20) puts it in the window. + + Ablation: search only the parent's own day dir and this fails.""" + parent = _codex_thread( + tmp_path, "2026/10/01", "23-50-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + after_midnight = _local_ts("2026/10/02", hour=0, minute=20) + os.utime(parent, (after_midnight, after_midnight)) + _codex_thread( + tmp_path, + "2026/10/02", + "00-10-00", + CHILD, + session_id=ROOT, + parent=ROOT, + usage=(7, 0, 3), + ) + + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=17, output_tokens=4) + + +def test_codex_tally_parent_without_session_meta_is_parent_only(tmp_path): + """A parent that does not open with a session_meta has no thread id to + match children against, so it tallies alone — even beside a sibling whose + session_meta carries no tree fields at all. + + Ablation: search siblings with a None root instead of returning the + parent's own total and the bare sibling (session_id None == root) is + counted.""" + parent = tmp_path / "sessions/2026/10/01" / f"rollout-2026-10-01T10-00-00-{ROOT}.jsonl" + _write_jsonl( + parent, + [ + { + "type": "event_msg", + "payload": { + "type": "token_count", + "info": {"total_token_usage": {"input_tokens": 30, "output_tokens": 3}}, + }, + } + ], + ) + stamp = _local_ts("2026/10/01") + os.utime(parent, (stamp, stamp)) + _codex_thread(tmp_path, "2026/10/01", "10-05-00", CHILD, usage=(500, 0, 50)) + + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=30, output_tokens=3) + + +def test_codex_tally_tracked_when_only_a_child_has_usage(tmp_path): + """A parent that reported no token_count yet still reads as tracked when + a child did: None stays reserved for no usage signal anywhere. + + Ablation: return None whenever the parent has no totals and this fails.""" + parent = _codex_thread(tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT) + _codex_thread( + tmp_path, + "2026/10/01", + "10-05-00", + CHILD, + session_id=ROOT, + parent=ROOT, + usage=(9, 0, 1), + ) + + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=9, output_tokens=1) + + +def test_codex_tally_parent_thread_chain_without_session_id(tmp_path): + """Older rollouts carry no session_id; membership then follows the + parent_thread_id chain into the parent's tree, to a fixpoint — the + grandchild's file sorts before its parent's, so one pass would miss it. + A rollout chained to an unknown thread stays out. + + Ablation: run the chain loop a single pass and the grandchild is + missed.""" + parent = _codex_thread(tmp_path, "2026/10/01", "10-00-00", ROOT, usage=(10, 0, 1)) + _codex_thread(tmp_path, "2026/10/01", "10-30-00", CHILD, parent=ROOT, usage=(20, 0, 2)) + _codex_thread(tmp_path, "2026/10/01", "10-20-00", GRANDCHILD, parent=CHILD, usage=(30, 0, 3)) + _codex_thread( + tmp_path, + "2026/10/01", + "10-40-00", + OTHER, + parent="019a0000-0000-7000-8000-0000000000ff", + usage=(900, 0, 9), + ) + + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=60, output_tokens=6) + + +def test_codex_tally_skips_unreadable_sibling(tmp_path): + """A child whose body cannot be decoded is skipped; the readable child and + the parent still count. + + Ablation: let the child's read error propagate and the tally raises.""" + parent = _codex_thread( + tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + _codex_thread( + tmp_path, + "2026/10/01", + "10-05-00", + CHILD, + session_id=ROOT, + parent=ROOT, + usage=(5, 0, 5), + ) + broken = _codex_thread( + tmp_path, "2026/10/01", "10-06-00", GRANDCHILD, session_id=ROOT, parent=ROOT + ) + with broken.open("ab") as f: + f.write(b'{"type": "event_msg", "payload": "\xff\xfe"}\n') + + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=15, output_tokens=6) + + +def test_codex_membership_cache_hit(tmp_path, monkeypatch): + """The budget guard tallies every heartbeat: after the first pass each + sibling's first line is cached (negatives too), so a repeat tally reads + none, and a new sibling costs exactly one first-line read. + + Ablation: skip the cache store and the second tally re-reads every first + line.""" + monkeypatch.setattr(tokens, "_codex_meta_cache", {}) + reads = [] + real_read = tokens._read_codex_first_line + + def counting_read(path): + reads.append(path.name) + return real_read(path) + + monkeypatch.setattr(tokens, "_read_codex_first_line", counting_read) + parent = _codex_thread( + tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + _codex_thread( + tmp_path, + "2026/10/01", + "10-05-00", + CHILD, + session_id=ROOT, + parent=ROOT, + usage=(5, 0, 1), + ) + _codex_thread(tmp_path, "2026/10/01", "10-06-00", OTHER, session_id=OTHER, usage=(9, 0, 9)) + + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=15, output_tokens=2) + assert len(reads) == 3 + reads.clear() + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=15, output_tokens=2) + assert reads == [] + late = _codex_thread( + tmp_path, + "2026/10/01", + "10-09-00", + GRANDCHILD, + session_id=ROOT, + parent=CHILD, + usage=(1, 0, 1), + ) + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=16, output_tokens=3) + assert reads == [late.name] + + +def test_codex_membership_cache_is_bounded(tmp_path, monkeypatch): + """The cache evicts oldest-first at its cap, so a long-lived engine never + grows it without bound. + + Ablation: drop the eviction loop and the cache outgrows the cap.""" + monkeypatch.setattr(tokens, "_codex_meta_cache", {}) + monkeypatch.setattr(tokens, "CODEX_META_CACHE_CAP", 2) + parent = _codex_thread(tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT) + for index, thread in enumerate((CHILD, GRANDCHILD, OTHER)): + _codex_thread( + tmp_path, + "2026/10/01", + f"10-0{index + 1}-00", + thread, + session_id=ROOT, + parent=ROOT, + usage=(1, 0, 1), + ) + + assert tally_codex_rollout(parent) == TokenUsage(input_tokens=3, output_tokens=3) + assert len(tokens._codex_meta_cache) == 2 + + def test_gemini_chat_dedupes_reemitted_messages(tmp_path): # shape captured from a real ~/.gemini/tmp//chats/session-*.jsonl # (2026-06-11): a JSONL patch stream where the same message id is From b4514da5b29b121f17bdbdc1a5ba2add97335af1 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 12 Oct 2026 18:21:30 -0700 Subject: [PATCH 4/6] fix(adapters): journal codex usage sources skipped as unreadable (#775) The codex tally skipped an unreadable subagent rollout, a failed sibling scan, or an unreadable parent first line with no record, so an under-counted session looked like a measured one. TokenUsage now carries the skipped sources (file name and exception class only, excluded from equality, add and to_dict), and the engine journals them on session-end as usage_unread. diagnostics drops the field to a presence flag because a rollout name embeds the session id. --- CHANGELOG.md | 5 ++-- src/bmad_loop/diagnostics.py | 5 ++++ src/bmad_loop/engine.py | 9 +++++-- src/bmad_loop/model.py | 6 +++++ src/bmad_loop/tokens.py | 24 ++++++++++++------ tests/test_engine.py | 27 +++++++++++++++++++++ tests/test_model.py | 13 ++++++++++ tests/test_tokens.py | 47 +++++++++++++++++++++++++++++++++--- 8 files changed, 121 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 82e1b494..3f08c18d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,8 +61,9 @@ breaking changes may land in a minor release. - Count codex subagent rollouts in session usage. Codex writes each subagent to its own rollout and leaves its usage out of the parent's totals, so subagent-heavy sessions were under-counted by half or more. Codex totals and budgets can roughly - double for those sessions, so budgets tuned on old figures may trip differently - (#775). + double for those sessions, so budgets tuned on old figures may trip differently. + A subagent rollout that cannot be read is named in `usage_unread` on the + `session-end` journal entry (#775). - Stop failing every opencode-http session start against OpenCode 2.x with a health timeout (#847). - Keep probing an opencode-http session over HTTP after an event-stream reconnect diff --git a/src/bmad_loop/diagnostics.py b/src/bmad_loop/diagnostics.py index 4020e6fb..22a8eea2 100644 --- a/src/bmad_loop/diagnostics.py +++ b/src/bmad_loop/diagnostics.py @@ -372,6 +372,11 @@ # free-text rule above (it is a `GitError` string quoting git's own stderr), # not this identifier-shape argument — same set, different rationale. "repo", + # The usage sources a parser skipped as unreadable (`session-end`), as + # ": ". A codex rollout's file name embeds the session + # id, so the list reduces to a presence flag here; the operator reads the + # names from the raw journal. + "usage_unread", # An absolute host path naming the folder a sentinel's upstream correction has # to land in (`rearm-upstream-write-unreachable`). Dropped for `repo`'s reason, # not aliased for `spec_file`'s: it is a DIRECTORY, journalled by one kind, and diff --git a/src/bmad_loop/engine.py b/src/bmad_loop/engine.py index a6136f88..b494c123 100644 --- a/src/bmad_loop/engine.py +++ b/src/bmad_loop/engine.py @@ -72,6 +72,7 @@ RunState, SessionRecord, StoryTask, + TokenUsage, VerifyOutcome, env_fault_site_reverifiable, result_mapping, @@ -7988,7 +7989,7 @@ def _base_commit_message(self, task: StoryTask) -> str: # ------------------------------------------------------------- helpers - def _session_end_extras(self, result: SessionResult) -> dict: + def _session_end_extras(self, result: SessionResult, usage: TokenUsage | None = None) -> dict: """Timeout forensics for the session-end entry (#157). ``teardown_s`` is the timeout-fire → journal gap — the window the incident showed could silently stretch for hours. A tripped session-budget guard @@ -8044,6 +8045,10 @@ def _session_end_extras(self, result: SessionResult) -> dict: extras["resultless_verdict"] = result.resultless_verdict if result.resultless_detail: extras["resultless_detail"] = result.resultless_detail + # skipped usage sources: same present-only convention, so a grep finds + # exactly the sessions whose `tokens` is a lower bound. + if usage is not None and usage.unread: + extras["usage_unread"] = list(usage.unread) return extras @staticmethod @@ -8438,7 +8443,7 @@ def _run_session( tokens_weighted=( usage.weighted_total(self.state.cache_read_weight()) if usage else None ), - **self._session_end_extras(result), + **self._session_end_extras(result, usage), ) ended = True finally: diff --git a/src/bmad_loop/model.py b/src/bmad_loop/model.py index 7e3825a9..8a51ed64 100644 --- a/src/bmad_loop/model.py +++ b/src/bmad_loop/model.py @@ -120,6 +120,12 @@ class TokenUsage: output_tokens: int = 0 cache_read_tokens: int = 0 cache_creation_tokens: int = 0 + # Sources the parser skipped as unreadable, so the totals above are a lower + # bound. Each entry is ": ": no directory and no message, + # because a home path or quoted transcript text must not reach the journal. + # Diagnostic only, and left out of equality, `add` and `to_dict`. The engine + # journals it on `session-end` as `usage_unread`. APPENDED. + unread: tuple[str, ...] = field(default=(), compare=False) def add(self, other: "TokenUsage") -> None: self.input_tokens += other.input_tokens diff --git a/src/bmad_loop/tokens.py b/src/bmad_loop/tokens.py index 7d65062c..7717fabb 100644 --- a/src/bmad_loop/tokens.py +++ b/src/bmad_loop/tokens.py @@ -371,25 +371,34 @@ def tally_codex_rollout(transcript_path: Path) -> TokenUsage | None: `session_id`), iterated to a fixpoint. One thread written to two files counts once, from the later file. A child's usage makes the session tracked even when the parent reports none. An unreadable sibling - (OSError/ValueError) is skipped; parent errors propagate as before, and - None still means no usage signal anywhere.""" + (OSError/ValueError) is skipped and named in `TokenUsage.unread`, as are + a failed sibling scan and an unreadable parent first line (both leave the + children uncounted). Parent body errors propagate as before, and None + still means no usage signal anywhere: a skip can only qualify a figure + that exists. The budget guard's live samples ignore `unread`, since a + child mid-write can tear a UTF-8 read; the engine's post-session read + journals it.""" parent_totals = _codex_rollout_total(transcript_path) + unread: list[str] = [] try: parent_meta = _codex_meta(transcript_path) - except OSError: + except OSError as e: parent_meta = None + unread.append(f"{transcript_path.name}: {type(e).__name__}") totals: list[dict[str, Any]] = [] if parent_totals is None else [parent_totals] if parent_meta is not None: root = parent_meta[0] metas: dict[str, _CodexMeta] = {} try: candidates = _codex_sibling_candidates(transcript_path) - except OSError: + except OSError as e: candidates = [] + unread.append(f"{transcript_path.parent.name}/: {type(e).__name__}") for path in candidates: try: meta = _codex_meta(path) - except OSError: + except OSError as e: + unread.append(f"{path.name}: {type(e).__name__}") continue if meta is not None and meta[0] != root: metas[str(path)] = meta @@ -408,13 +417,14 @@ def tally_codex_rollout(transcript_path: Path) -> TokenUsage | None: for child in by_thread.values(): try: child_totals = _codex_rollout_total(child) - except (OSError, ValueError): + except (OSError, ValueError) as e: + unread.append(f"{child.name}: {type(e).__name__}") continue if child_totals is not None: totals.append(child_totals) if not totals: return None - usage = TokenUsage() + usage = TokenUsage(unread=tuple(unread)) for each in totals: usage.add(_codex_usage(each)) return usage diff --git a/tests/test_engine.py b/tests/test_engine.py index c56de02c..a1d4099e 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -15365,6 +15365,33 @@ def test_session_end_extras_omits_resultless_when_unset(project): assert all("resultless_verdict" not in e and "resultless_detail" not in e for e in ends) +def test_session_end_carries_usage_unread(project): + """A usage parser that skipped unreadable sources leaves them on + `session-end`, so a lower-bound `tokens` is visible as one. + + ABLATION: delete the `usage_unread` block in `_session_end_extras` and the + assertion fails.""" + write_sprint(project, {"1-1-a": "ready-for-dev"}) + timeout = SessionResult(status="timeout") + engine, adapter = make_engine(project, [timeout, timeout]) + unread = ("rollout-2026-10-01T10-06-00-x.jsonl: UnicodeDecodeError",) + adapter.usage_per_session = TokenUsage(input_tokens=10, output_tokens=5, unread=unread) + engine.run() + ends = [e for e in engine.journal.entries() if e["kind"] == "session-end"] + assert len(ends) == 2 + assert all(e["usage_unread"] == list(unread) for e in ends) + assert all(e["tokens"] == 15 for e in ends) + + +def test_session_end_omits_usage_unread_when_clean(project): + """Present-only, the `parked` convention.""" + write_sprint(project, {"1-1-a": "ready-for-dev"}) + engine, _ = make_engine(project, [SessionResult(status="timeout")]) + engine.run() + ends = [e for e in engine.journal.entries() if e["kind"] == "session-end"] + assert ends and all("usage_unread" not in e for e in ends) + + def test_engine_attaches_its_journal_to_every_adapter(project): """The engine hands its `Journal` to the adapters it owns (#680), so an adapter-side `session-idle` lands in the same file with the same diff --git a/tests/test_model.py b/tests/test_model.py index baae539c..cc25f9e6 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -337,6 +337,19 @@ def test_attach_session_usage_does_not_double_count_existing_usage(): assert task.tokens.total == 15 +def test_token_usage_unread_is_diagnostic_only(): + """`unread` qualifies a figure without changing it: equality, `add` and the + persisted dict ignore it, so a skip never alters stored totals or `--json`. + + Ablation: drop `compare=False` and the equality assertion fails.""" + usage = TokenUsage(input_tokens=10, unread=("rollout-x.jsonl: OSError",)) + assert usage == TokenUsage(input_tokens=10) + assert "unread" not in usage.to_dict() + total = TokenUsage() + total.add(usage) + assert total.unread == () + + def test_session_record_result_json_round_trips(): record = SessionRecord( task_id="1-1-a-dev-1", diff --git a/tests/test_tokens.py b/tests/test_tokens.py index 075c57f7..89af24d3 100644 --- a/tests/test_tokens.py +++ b/tests/test_tokens.py @@ -590,10 +590,11 @@ def test_codex_tally_parent_thread_chain_without_session_id(tmp_path): def test_codex_tally_skips_unreadable_sibling(tmp_path): - """A child whose body cannot be decoded is skipped; the readable child and - the parent still count. + """A child whose body cannot be decoded is skipped and named in `unread`; + the readable child and the parent still count. - Ablation: let the child's read error propagate and the tally raises.""" + Ablation: let the child's read error propagate and the tally raises; drop + the child's `unread` append and the `unread` assertion fails.""" parent = _codex_thread( tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) ) @@ -612,7 +613,45 @@ def test_codex_tally_skips_unreadable_sibling(tmp_path): with broken.open("ab") as f: f.write(b'{"type": "event_msg", "payload": "\xff\xfe"}\n') - assert tally_codex_rollout(parent) == TokenUsage(input_tokens=15, output_tokens=6) + usage = tally_codex_rollout(parent) + assert usage == TokenUsage(input_tokens=15, output_tokens=6) + assert usage is not None and usage.unread == (f"{broken.name}: UnicodeDecodeError",) + + +def test_codex_tally_names_sibling_whose_first_line_fails(tmp_path, monkeypatch): + """A sibling whose session_meta cannot be read may be a child, so it is + named in `unread` too, by file name only: no directory reaches the journal. + + Ablation: drop the candidate's `unread` append and the assertion fails.""" + monkeypatch.setattr(tokens, "_codex_meta_cache", {}) + parent = _codex_thread( + tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + locked = _codex_thread( + tmp_path, "2026/10/01", "10-05-00", CHILD, session_id=ROOT, usage=(5, 0, 5) + ) + real_read = tokens._read_codex_first_line + + def read(path): + if path == locked: + raise PermissionError(13, "denied", str(path)) + return real_read(path) + + monkeypatch.setattr(tokens, "_read_codex_first_line", read) + + usage = tally_codex_rollout(parent) + assert usage == TokenUsage(input_tokens=10, output_tokens=1) + assert usage is not None and usage.unread == (f"{locked.name}: PermissionError",) + + +def test_codex_clean_tally_has_no_unread(tmp_path): + parent = _codex_thread( + tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + _codex_thread(tmp_path, "2026/10/01", "10-05-00", CHILD, session_id=ROOT, usage=(5, 0, 5)) + + usage = tally_codex_rollout(parent) + assert usage is not None and usage.unread == () def test_codex_membership_cache_hit(tmp_path, monkeypatch): From 2546e9a0523edfd9c5f372322217e50d70123557 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 12 Oct 2026 18:58:25 -0700 Subject: [PATCH 5/6] fix(adapters): close codex resolution and unread gaps from review (#775) - Retry the by-id rollout lookup on each heartbeat, so a rollout created after SessionStart arms the budget guard before Stop. - Resolve on the stall re-probe's window-death exit too. - Root the codex_home fallback at the passed env's HOME/USERPROFILE. - Name a rollout whose complete first line does not parse in `unread` (parent or sibling) instead of reading it as not-a-subagent; a first line still being written stays a silent, uncached miss. - List sibling day dirs directly (Path.glob swallows scan errors) and name an unlistable one in `unread` while still searching the others. --- src/bmad_loop/adapters/generic.py | 10 ++- src/bmad_loop/tokens.py | 65 ++++++++++------ tests/test_generic_tmux.py | 93 +++++++++++++++++++++++ tests/test_tokens.py | 120 ++++++++++++++++++++++++++++++ 4 files changed, 265 insertions(+), 23 deletions(-) diff --git a/src/bmad_loop/adapters/generic.py b/src/bmad_loop/adapters/generic.py index 3be96c44..a21ebd5b 100644 --- a/src/bmad_loop/adapters/generic.py +++ b/src/bmad_loop/adapters/generic.py @@ -1179,7 +1179,8 @@ def sample_transcript(path: str, now: float) -> None: def try_resolve() -> None: """Resolve the transcript by session id when no readable one is known. - One lookup per call; a hit becomes the loop's transcript (so every + One lookup per call — each hook event, each heartbeat, each + window-death exit; a hit becomes the loop's transcript (so every SessionResult and the budget guard see it) and is crumbed once — the ``resolved is None`` gate makes the hit itself one-shot.""" nonlocal transcript_path, resolved @@ -1287,6 +1288,11 @@ def produced_work() -> bool: now = time.monotonic() if last_heartbeat is None or now - last_heartbeat >= HEARTBEAT_INTERVAL_S: last_heartbeat = now + # A rollout created after SessionStart, in a turn with no further + # hook event, is found here rather than at Stop — so the budget + # guard below is not blind for the whole turn. Bounded by the + # heartbeat cadence; a no-op once a readable transcript is known. + try_resolve() # Transcript idle sample (#680), ahead of the heartbeat write so # the payload carries this tick's age. Inert until a hook event # has named the transcript. @@ -1608,6 +1614,8 @@ def produced_work() -> bool: # nothing is waiting on a human once the CLI is gone. The # probe fault is crumbed (`site="stall"`, DW-454). if self._probe_liveness(handle, "stall") is False: + # A window-death exit like the top-of-tick one above. + try_resolve() return self._final( handle, spec, diff --git a/src/bmad_loop/tokens.py b/src/bmad_loop/tokens.py index 7717fabb..8fccd728 100644 --- a/src/bmad_loop/tokens.py +++ b/src/bmad_loop/tokens.py @@ -47,7 +47,9 @@ from __future__ import annotations import datetime +import fnmatch import json +import os import re import threading import uuid @@ -72,10 +74,15 @@ def read_usage(parser: str, transcript_path: Path) -> TokenUsage | None: def codex_home(env: Mapping[str, str]) -> Path | None: """Codex's state root: ``CODEX_HOME`` when set and non-empty, else - ``~/.codex``; None when no home directory can be determined.""" + ``.codex`` under the home ``env`` names (``USERPROFILE`` on Windows, + ``HOME`` elsewhere, as ``Path.home()`` reads them), else under this + process's home; None when no home directory can be determined.""" configured = env.get("CODEX_HOME") if configured: return Path(configured) + home = env.get("USERPROFILE" if os.name == "nt" else "HOME") + if home: + return Path(home) / ".codex" try: return Path.home() / ".codex" except RuntimeError: @@ -254,7 +261,8 @@ def _codex_usage(totals: dict[str, Any]) -> TokenUsage: # (id, session_id, parent_thread_id) from a rollout's first-line session_meta, # or None when the file does not open with one. session_meta is written once and # never changes, so every answer, negative included, is cached by path for -# good; only a first line still being written (no newline yet) is re-read. +# good; only a first line still being written (no newline yet) is re-read, and +# a complete first line that is not JSON raises, uncached, every time. # Bounded oldest-first like opencode_http's USAGE_STASH_CAP. A day dir holds a # few hundred rollouts and a tally lists the parent's day plus the next, so the # cap sits far above one tally's candidate set: the budget guard's per-heartbeat @@ -273,10 +281,9 @@ def _read_codex_first_line(path: Path) -> bytes: def _parse_codex_meta(line: bytes) -> _CodexMeta | None: - try: - entry = json.loads(line) - except ValueError: - return None + """None for a JSON line that is not a usable session_meta; ValueError + (JSONDecodeError, UnicodeDecodeError) for one that is not JSON at all.""" + entry = json.loads(line) if not isinstance(entry, dict) or entry.get("type") != "session_meta": return None payload = entry.get("payload") @@ -301,14 +308,21 @@ def text(value: Any) -> str | None: def _codex_meta(path: Path) -> _CodexMeta | None: """The rollout's session_meta facts, cached (see CODEX_META_CACHE_CAP). - OSError propagates.""" + OSError propagates, and so does ValueError for a complete first line that + does not parse; a first line still being written reads as None.""" key = str(path) with _codex_meta_lock: if key in _codex_meta_cache: return _codex_meta_cache[key] line = _read_codex_first_line(path) - meta = _parse_codex_meta(line) - if line.endswith(b"\n") or len(line) >= _CODEX_META_READ_CAP: + complete = line.endswith(b"\n") or len(line) >= _CODEX_META_READ_CAP + try: + meta = _parse_codex_meta(line) + except ValueError: + if complete: + raise + meta = None + if complete: with _codex_meta_lock: if key not in _codex_meta_cache: while len(_codex_meta_cache) >= CODEX_META_CACHE_CAP: @@ -326,12 +340,15 @@ def _codex_day(folder: Path) -> datetime.date | None: return None -def _codex_sibling_candidates(transcript_path: Path) -> list[Path]: +def _codex_sibling_candidates(transcript_path: Path, unread: list[str]) -> list[Path]: """Rollouts that may be the parent's subagents: `rollout-*.jsonl` in the parent's day dir and each later day dir through the day after the parent's last write (capped at local today + 1; dirs are local dates and a child can start after midnight), whose filename timestamp is not earlier than - the parent's. A parent outside a YYYY/MM/DD tree searches its own dir.""" + the parent's. A parent outside a YYYY/MM/DD tree searches its own dir. + A missing day dir holds no rollouts; one that cannot be listed is named + in ``unread`` (by its date path) and the other dirs are still searched — + listed directly, since ``Path.glob`` swallows a scan's OSError.""" folder = transcript_path.parent first = _codex_day(folder) folders = [folder] @@ -352,7 +369,16 @@ def _codex_sibling_candidates(transcript_path: Path) -> list[Path]: floor = stamp.group(1) if stamp else "" candidates: list[Path] = [] for each in folders: - for path in sorted(each.glob("rollout-*.jsonl")): + try: + paths = sorted(each.iterdir()) + except FileNotFoundError: + continue + except OSError as e: + unread.append(f"{'/'.join(each.parts[-3:])}/: {type(e).__name__}") + continue + for path in paths: + if not fnmatch.fnmatchcase(path.name, "rollout-*.jsonl"): + continue match = _ROLLOUT_TS.match(path.name) if path != transcript_path and (match.group(1) if match else "") >= floor: candidates.append(path) @@ -372,8 +398,8 @@ def tally_codex_rollout(transcript_path: Path) -> TokenUsage | None: counts once, from the later file. A child's usage makes the session tracked even when the parent reports none. An unreadable sibling (OSError/ValueError) is skipped and named in `TokenUsage.unread`, as are - a failed sibling scan and an unreadable parent first line (both leave the - children uncounted). Parent body errors propagate as before, and None + an unlistable day dir and an unreadable or unparseable parent first line + (which leaves the children uncounted). Parent body errors propagate as before, and None still means no usage signal anywhere: a skip can only qualify a figure that exists. The budget guard's live samples ignore `unread`, since a child mid-write can tear a UTF-8 read; the engine's post-session read @@ -382,22 +408,17 @@ def tally_codex_rollout(transcript_path: Path) -> TokenUsage | None: unread: list[str] = [] try: parent_meta = _codex_meta(transcript_path) - except OSError as e: + except (OSError, ValueError) as e: parent_meta = None unread.append(f"{transcript_path.name}: {type(e).__name__}") totals: list[dict[str, Any]] = [] if parent_totals is None else [parent_totals] if parent_meta is not None: root = parent_meta[0] metas: dict[str, _CodexMeta] = {} - try: - candidates = _codex_sibling_candidates(transcript_path) - except OSError as e: - candidates = [] - unread.append(f"{transcript_path.parent.name}/: {type(e).__name__}") - for path in candidates: + for path in _codex_sibling_candidates(transcript_path, unread): try: meta = _codex_meta(path) - except OSError as e: + except (OSError, ValueError) as e: unread.append(f"{path.name}: {type(e).__name__}") continue if meta is not None and meta[0] != root: diff --git a/tests/test_generic_tmux.py b/tests/test_generic_tmux.py index c774dc8c..ef1c86af 100644 --- a/tests/test_generic_tmux.py +++ b/tests/test_generic_tmux.py @@ -2465,6 +2465,61 @@ def create_rollout(call_n): assert len(_lifecycle_events(adapter, "transcript-resolved")) == 1 +def test_codex_stall_reprobe_death_resolves_at_exit(tmp_path, monkeypatch): + """The other window-death exit: alive at the top-of-tick probe, dead at + the stall-expiry re-probe. A rollout that appeared after SessionStart + still rides the crashed result. + + Ablation: remove the `try_resolve()` call before the stall re-probe's + crash return and this fails.""" + monkeypatch.setattr(generic, "RESULT_GRACE_S", 0.0) + monkeypatch.setattr(generic, "RESULT_POLL_S", 0.0) + home = _codex_home(tmp_path, monkeypatch) + adapter, _ = make_dev_adapter(tmp_path, profile_name="codex") + adapter._stall_grace_s = 10.0 + adapter._stall_nudges = 0 + alive_calls = {"n": 0} + + def flaky_alive(handle): + alive_calls["n"] += 1 + return alive_calls["n"] == 1 + + adapter._window_alive = flaky_alive + # heartbeat-free clock: the heartbeat's own resolve must not be what finds it + clock = {"t": 1000.0} + + class _Clock: + monotonic = staticmethod(lambda: clock["t"]) + time = staticmethod(lambda: 0.0) + sleep = staticmethod(lambda *_: None) + time_ns = staticmethod(lambda: 0) + + monkeypatch.setattr(generic, "time", _Clock) + monkeypatch.setattr(generic, "HEARTBEAT_INTERVAL_S", 1e9) + created: list[Path] = [] + + def expire_after_rollout(call_n): + if call_n == 3: + created.append(_write_codex_rollout(home)) + clock["t"] += 11.0 + + adapter.watcher = _ScriptedWatcher( + [ + _hook_event("SessionStart", session_id=CODEX_SESSION), + _stop_event("3-1-dev-1", CODEX_SESSION, None), + ], + on_call=expire_after_rollout, + ) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "crashed" + assert alive_calls["n"] == 2 + assert created + assert Path(result.transcript_path) == created[0] + assert len(_lifecycle_events(adapter, "transcript-resolved")) == 1 + + def test_non_codex_stop_without_transcript_never_resolves(tmp_path, monkeypatch): """Scope pin: only codex-rollout transcripts are resolvable by id. A claude Stop with no path stays untracked and crumbs nothing new, even with a codex @@ -4157,6 +4212,44 @@ def test_budget_guard_arms_after_codex_resolution(tmp_path, monkeypatch): assert len(tripped) == 1 +def test_budget_guard_arms_when_rollout_appears_mid_turn(tmp_path, monkeypatch): + """SessionStart came before its rollout existed, and the turn sends no + further hook event: the heartbeat retries the by-id lookup, so the guard + samples the rollout and trips before Stop. + + Ablation: delete the heartbeat's `try_resolve()` call and this fails — + the rollout is found only at Stop, after the last sample.""" + home = _codex_home(tmp_path, monkeypatch) + adapter, clock, sent = _budget_adapter(tmp_path, monkeypatch, usage_parser="codex-rollout") + (adapter.tasks_dir / "b-1" / "result.json").write_text('{"ok": true}', encoding="utf-8") + start = HookEvent( + ts=1, + event="SessionStart", + task_id="b-1", + session_id=CODEX_SESSION, + transcript_path=None, + path=Path("x"), + ) + advance = _advance_31(clock) + + def advance_and_create(call_n): + advance(call_n) + if call_n == 2: + _write_codex_rollout(home, input_tokens=5000, output_tokens=0) + + adapter.watcher = _ScriptedWatcher( + [start, None, None, _stop_event("b-1", CODEX_SESSION, None)], + on_call=advance_and_create, + ) + + result = adapter.wait_for_completion(_budget_handle(), _budget_spec(tmp_path, mode="warn")) + + assert result.status == "completed" + assert result.budget_weighted == 5000 + tripped = [ln for ln in _lifecycle_lines(adapter, "b-1") if ln["event"] == "budget-tripped"] + assert len(tripped) == 1 + + def test_budget_enforce_nudges_then_terminates_over_budget(tmp_path, monkeypatch): """Enforce mode: BUDGET_NUDGE_TEXT at trip, then grace expiry under a live window ends over_budget WITHOUT accepting the on-disk artifact (#48/#53).""" diff --git a/tests/test_tokens.py b/tests/test_tokens.py index 89af24d3..e2ba0c1a 100644 --- a/tests/test_tokens.py +++ b/tests/test_tokens.py @@ -400,6 +400,20 @@ def test_resolve_defaults_to_home_dot_codex(tmp_path, monkeypatch): assert tokens.resolve_transcript("codex-rollout", SESSION, {"CODEX_HOME": ""}) == path +def test_resolve_default_home_comes_from_the_passed_env(tmp_path, monkeypatch): + """The pane's env decides where Codex writes: a HOME/USERPROFILE override + there must root the fallback, not this process's own home. + + Ablation: read `Path.home()` unconditionally and this fails.""" + monkeypatch.setenv("HOME", str(tmp_path / "controller")) + monkeypatch.setenv("USERPROFILE", str(tmp_path / "controller")) + pane_home = tmp_path / "pane" + path = _codex_rollout(pane_home / ".codex", "2026/10/12", SESSION) + env = {"HOME": str(pane_home), "USERPROFILE": str(pane_home)} + + assert tokens.resolve_transcript("codex-rollout", SESSION, env) == path + + def test_resolve_newest_mtime_wins(tmp_path): """Two rollouts with one id (a resumed session written into a later day dir) resolve to the newer by mtime, whatever order the glob returns them in. @@ -644,6 +658,112 @@ def read(path): assert usage is not None and usage.unread == (f"{locked.name}: PermissionError",) +def test_codex_tally_names_sibling_whose_first_line_does_not_parse(tmp_path, monkeypatch): + """A complete first line that is not JSON (invalid UTF-8 here) may hide a + child, so it is named in `unread` rather than read as "not a subagent". + + Ablation: let `_codex_meta` return None on a parse failure and the + assertion fails.""" + monkeypatch.setattr(tokens, "_codex_meta_cache", {}) + parent = _codex_thread( + tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + garbled = tmp_path / "sessions/2026/10/01" / f"rollout-2026-10-01T10-05-00-{CHILD}.jsonl" + garbled.write_bytes(b'{"type": "session_meta", "payload": "\xff"}\n') + + usage = tally_codex_rollout(parent) + assert usage == TokenUsage(input_tokens=10, output_tokens=1) + assert usage is not None and usage.unread == (f"{garbled.name}: UnicodeDecodeError",) + # failures are not cached: the next tally names it again + usage = tally_codex_rollout(parent) + assert usage is not None and usage.unread == (f"{garbled.name}: UnicodeDecodeError",) + + +def test_codex_tally_names_parent_whose_first_line_does_not_parse(tmp_path, monkeypatch): + """An unparseable parent first line leaves its children uncounted, so the + parent's own total is a lower bound and says so. + + Ablation: catch only OSError around the parent's `_codex_meta` and the + tally raises.""" + monkeypatch.setattr(tokens, "_codex_meta_cache", {}) + parent = tmp_path / "sessions/2026/10/01" / f"rollout-2026-10-01T10-00-00-{ROOT}.jsonl" + _write_jsonl( + parent, + [ + { + "type": "event_msg", + "payload": { + "type": "token_count", + "info": {"total_token_usage": {"input_tokens": 30, "output_tokens": 3}}, + }, + } + ], + ) + body = parent.read_bytes() + parent.write_bytes(b"{not json\n" + body) + + usage = tally_codex_rollout(parent) + assert usage == TokenUsage(input_tokens=30, output_tokens=3) + assert usage is not None and usage.unread == (f"{parent.name}: JSONDecodeError",) + + +def test_codex_tally_skips_sibling_first_line_still_being_written(tmp_path, monkeypatch): + """A first line with no newline yet is a rollout mid-creation, not a + fault: it reads as not-a-child for now, uncached, and names nothing. + + Ablation: raise for an incomplete line too and `unread` is non-empty.""" + monkeypatch.setattr(tokens, "_codex_meta_cache", {}) + parent = _codex_thread( + tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + torn = tmp_path / "sessions/2026/10/01" / f"rollout-2026-10-01T10-05-00-{CHILD}.jsonl" + torn.write_bytes(b'{"type": "session_meta", "payl') + + usage = tally_codex_rollout(parent) + assert usage == TokenUsage(input_tokens=10, output_tokens=1) + assert usage is not None and usage.unread == () + assert str(torn) not in tokens._codex_meta_cache + + +def test_codex_tally_names_unlistable_day_dir(tmp_path, monkeypatch): + """`Path.glob` swallows a scan's OSError, so the dirs are listed directly: + an unlistable day dir is named in `unread` by its date path, and the + other days are still searched. + + Ablation: list the dirs with `glob("rollout-*.jsonl")` again and the + `unread` assertion fails.""" + parent = _codex_thread( + tmp_path, "2026/10/01", "23-50-00", ROOT, session_id=ROOT, usage=(10, 0, 1) + ) + after_midnight = _local_ts("2026/10/02", hour=0, minute=20) + os.utime(parent, (after_midnight, after_midnight)) + _codex_thread( + tmp_path, "2026/10/01", "23-55-00", CHILD, session_id=ROOT, parent=ROOT, usage=(7, 0, 3) + ) + _codex_thread( + tmp_path, + "2026/10/02", + "00-10-00", + GRANDCHILD, + session_id=ROOT, + parent=ROOT, + usage=(500, 0, 50), + ) + locked = tmp_path / "sessions/2026/10/02" + real_iterdir = type(locked).iterdir + + def iterdir(self): + if self == locked: + raise PermissionError(13, "denied", str(self)) + return real_iterdir(self) + + monkeypatch.setattr(type(locked), "iterdir", iterdir) + + usage = tally_codex_rollout(parent) + assert usage == TokenUsage(input_tokens=17, output_tokens=4) + assert usage is not None and usage.unread == ("2026/10/02/: PermissionError",) + + def test_codex_clean_tally_has_no_unread(tmp_path): parent = _codex_thread( tmp_path, "2026/10/01", "10-00-00", ROOT, session_id=ROOT, usage=(10, 0, 1) From 8857192eeaa35d3d6b003cecfec6aedaa90600fa Mon Sep 17 00:00:00 2001 From: t Date: Mon, 12 Oct 2026 19:09:52 -0700 Subject: [PATCH 6/6] fix(adapters): resolve the codex rollout before every non-hook exit (#775) The timeout, both hard-stop polls, and the live-window stall exit returned without retrying the by-id lookup, so a SessionStart that came before its rollout existed, followed by one of those exits before the next heartbeat, left the result untracked. Each now resolves first; the stall resolve moves ahead of the re-probe so the crashed and stalled branches both carry it. --- src/bmad_loop/adapters/generic.py | 14 ++-- tests/test_generic_tmux.py | 102 ++++++++++++++++++++++++++++++ 2 files changed, 112 insertions(+), 4 deletions(-) diff --git a/src/bmad_loop/adapters/generic.py b/src/bmad_loop/adapters/generic.py index a21ebd5b..b61c685c 100644 --- a/src/bmad_loop/adapters/generic.py +++ b/src/bmad_loop/adapters/generic.py @@ -1179,8 +1179,8 @@ def sample_transcript(path: str, now: float) -> None: def try_resolve() -> None: """Resolve the transcript by session id when no readable one is known. - One lookup per call — each hook event, each heartbeat, each - window-death exit; a hit becomes the loop's transcript (so every + One lookup per call — each hook event, each heartbeat, and before + each exit that no hook event just preceded; a hit becomes the loop's transcript (so every SessionResult and the budget guard see it) and is crumbed once — the ``resolved is None`` gate makes the hit itself one-shot.""" nonlocal transcript_path, resolved @@ -1253,6 +1253,10 @@ def produced_work() -> bool: # own Stop may have been read as foreign) foreign_hook_events=foreign_hook_events, ) + # Every exit resolves first (a SessionStart that came before its + # rollout existed, then no further hook event): the exit's + # `transcript_path` argument is read before `produced_work()`. + try_resolve() return SessionResult( status="timeout", session_id=session_id, @@ -1277,6 +1281,7 @@ def produced_work() -> bool: # left on disk for the engine to consume and attribute the stop. if self._hard_stop_requested(): self._note_lifecycle(handle.task_id, "stop-abort-fired") + try_resolve() return SessionResult( status="aborted", session_id=session_id, @@ -1489,6 +1494,7 @@ def produced_work() -> bool: # the pre-#319 outcome, never a worse one. if self._hard_stop_requested(): self._note_lifecycle(handle.task_id, "stop-abort-fired") + try_resolve() return SessionResult( status="aborted", session_id=session_id, @@ -1613,9 +1619,9 @@ def produced_work() -> bool: # persistent failure. A dead window is never labelled parked: # nothing is waiting on a human once the CLI is gone. The # probe fault is crumbed (`site="stall"`, DW-454). + # Both exits below (crashed or stalled) carry the rollout. + try_resolve() if self._probe_liveness(handle, "stall") is False: - # A window-death exit like the top-of-tick one above. - try_resolve() return self._final( handle, spec, diff --git a/tests/test_generic_tmux.py b/tests/test_generic_tmux.py index ef1c86af..2457ff90 100644 --- a/tests/test_generic_tmux.py +++ b/tests/test_generic_tmux.py @@ -2520,6 +2520,108 @@ def expire_after_rollout(call_n): assert len(_lifecycle_events(adapter, "transcript-resolved")) == 1 +@pytest.mark.parametrize("exit_kind", ["timeout", "abort-top", "abort-after-wait"]) +def test_codex_non_hook_exits_resolve(tmp_path, monkeypatch, exit_kind): + """SessionStart came before its rollout existed, and the session then ends + on an exit no hook event precedes (timeout, or a hard stop at either poll) + before the next heartbeat: the exit itself resolves, so the result carries + the rollout. + + Ablation: remove the `try_resolve()` call ahead of that exit's return and + the matching case fails.""" + monkeypatch.setattr(generic, "RESULT_GRACE_S", 0.0) + monkeypatch.setattr(generic, "RESULT_POLL_S", 0.0) + home = _codex_home(tmp_path, monkeypatch) + adapter, _ = make_dev_adapter(tmp_path, profile_name="codex") + clock = {"t": 1000.0} + + class _Clock: + monotonic = staticmethod(lambda: clock["t"]) + time = staticmethod(lambda: 0.0) + sleep = staticmethod(lambda *_: None) + time_ns = staticmethod(lambda: 0) + + monkeypatch.setattr(generic, "time", _Clock) + # heartbeat-free after the first tick: its own resolve must not find it + monkeypatch.setattr(generic, "HEARTBEAT_INTERVAL_S", 1e9) + created: list[Path] = [] + + def alive(handle): + # past the post-wait poll, so the top-of-loop poll is the one to fire + if exit_kind == "abort-top" and created: + _lodge_stop_request(adapter, "hard") + return True + + adapter._window_alive = alive + + def create_then_exit(call_n): + if call_n != 2: + return + created.append(_write_codex_rollout(home)) + if exit_kind == "timeout": + clock["t"] += 1000.0 + elif exit_kind == "abort-after-wait": + _lodge_stop_request(adapter, "hard") + + adapter.watcher = _ScriptedWatcher( + [_hook_event("SessionStart", session_id=CODEX_SESSION)], on_call=create_then_exit + ) + + result = adapter.wait_for_completion(_dev_handle(), _short_spec(tmp_path, timeout_s=100.0)) + + assert result.status == ("timeout" if exit_kind == "timeout" else "aborted") + assert adapter.watcher.calls == 2 + assert created + assert Path(result.transcript_path) == created[0] + assert len(_lifecycle_events(adapter, "transcript-resolved")) == 1 + + +def test_codex_live_stall_resolves_at_exit(tmp_path, monkeypatch): + """The stall exit under a live window: a rollout that appeared after + SessionStart, with no hook event since, still rides the stalled result. + + Ablation: move the stall exit's `try_resolve()` back inside the dead-window + branch and this fails.""" + monkeypatch.setattr(generic, "RESULT_GRACE_S", 0.0) + monkeypatch.setattr(generic, "RESULT_POLL_S", 0.0) + home = _codex_home(tmp_path, monkeypatch) + adapter, _ = make_dev_adapter(tmp_path, profile_name="codex") + adapter._stall_grace_s = 10.0 + adapter._stall_nudges = 0 + adapter._window_alive = lambda handle: True + clock = {"t": 1000.0} + + class _Clock: + monotonic = staticmethod(lambda: clock["t"]) + time = staticmethod(lambda: 0.0) + sleep = staticmethod(lambda *_: None) + time_ns = staticmethod(lambda: 0) + + monkeypatch.setattr(generic, "time", _Clock) + monkeypatch.setattr(generic, "HEARTBEAT_INTERVAL_S", 1e9) + created: list[Path] = [] + + def expire_after_rollout(call_n): + if call_n == 3: + created.append(_write_codex_rollout(home)) + clock["t"] += 11.0 + + adapter.watcher = _ScriptedWatcher( + [ + _hook_event("SessionStart", session_id=CODEX_SESSION), + _stop_event("3-1-dev-1", CODEX_SESSION, None), + ], + on_call=expire_after_rollout, + ) + + result = adapter.wait_for_completion(_dev_handle(), _dev_spec(tmp_path)) + + assert result.status == "stalled" + assert created + assert Path(result.transcript_path) == created[0] + assert len(_lifecycle_events(adapter, "transcript-resolved")) == 1 + + def test_non_codex_stop_without_transcript_never_resolves(tmp_path, monkeypatch): """Scope pin: only codex-rollout transcripts are resolvable by id. A claude Stop with no path stays untracked and crumbs nothing new, even with a codex