fix(adapters): capture codex usage when the hook path is unusable; count subagent rollouts (#775)

CHANGELOG.md

@@ -53,6 +53,17 @@ 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).
+- 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.
+  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

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

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,38 @@ 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 — 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
+            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.
@@ -1199,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,
@@ -1223,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,
@@ -1234,6 +1293,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.
@@ -1430,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,
@@ -1463,7 +1528,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,
@@ -1549,6 +1619,8 @@ 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:
                         return self._final(
                             handle,
@@ -1638,6 +1710,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 +1733,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 +1776,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:

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

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

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
+        # "<file name>: <ExcType>". 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

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: