perf(usage): stream session ledger reconstruction

docs/architecture/rfcs/goal-usage-token-cost-v0.md

@@ -19,6 +19,18 @@ slot accounting) but what it spent. This closes the gap left open in
 `docs/quota-allocation.md`, which notes that later versions may "replace slots
 with real runtime, token spend, or cost."
 
+## Current implementation checkpoint
+
+The Codex capture adapter and typed run usage are shipped; the problem and
+smallest-slice sections below retain the original proposal basis. The existing
+booking module now streams its complete retained run index when reconstructing
+the session baseline. This removes whole-index text buffers without adding an
+public interface, cache, ledger or decision owner. The [usage contract](../../status-data-contract.md#usage-summary)
+retains replay, duplicate suppression, corruption refusal and unknown metrics.
+Buffering scales with the largest record; selected-session row identities still
+grow with that session's retained history. Lifetime and runtime coverage remain
+open questions, not inferred totals.
+
 ## Problem
 
 `usage_summary` is explicitly a run-history proxy that excludes token counts.

docs/status-data-contract.md

@@ -2442,6 +2442,13 @@ producer boundary, and the run index append is the single commit point: each
 session's delta basis is reconstructed from its own already-booked rows
 (telescoping absolute + delta sums per session id), not from a second state
 file, so a crash or retry can never leave the basis behind the ledger.
+The basis reader streams the complete retained index instead of buffering its
+whole text and split lines. Its record buffer scales with the largest index
+record; it still retains the selected session's row identities for duplicate
+suppression. Missing indexes mean no prior booking. Corrupt JSON or an IO
+failure, including a failure after an earlier valid row, refuses the booking
+without publishing a partial usage row. This changes neither retention nor
+the booking lock and adds no cache or second ledger.
 Baselines are per session and each observation is bound to
 `usage.source_snapshot_id`, so interleaved sessions never rebase against each
 other: a returning session books only its own increment instead of re-booking

loopx/control_plane/quota/codex_session_usage.py

@@ -279,6 +279,35 @@ def _legacy_model_at_booking(
     return model
 
 
+def _usage_index_records(index_path: Path) -> Iterator[tuple[int, dict[str, Any]]]:
+    try:
+        stream = index_path.open(encoding="utf-8")
+    except FileNotFoundError:
+        return
+    except OSError as exc:
+        raise CodexSessionUsageError(
+            f"cannot read run index for the usage basis: {exc}"
+        ) from exc
+    try:
+        with stream:
+            for line_number, line in enumerate(stream, start=1):
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    row = json.loads(line)
+                except json.JSONDecodeError as exc:
+                    raise CodexSessionUsageError(
+                        f"run index row {line_number} is corrupt: {index_path}"
+                    ) from exc
+                if isinstance(row, dict):
+                    yield line_number, row
+    except OSError as exc:
+        raise CodexSessionUsageError(
+            f"cannot read run index for the usage basis: {exc}"
+        ) from exc
+
+
 def session_usage_baseline(
     index_path: Path, session_id: str
 ) -> dict[str, Any] | None:
@@ -292,38 +321,20 @@ def session_usage_baseline(
     Corrupt index rows fail closed: skipping a booked row would shrink the
     basis and double-book real spend. Callers must hold the usage booking lock
     while reading the basis and appending the row it funds.
+    Buffer one record at a time; retain only the selected session's aggregate
+    metadata and row identities needed for duplicate suppression.
     """
     sid = str(session_id or "").strip()
     if not sid:
         raise CodexSessionUsageError("usage observation has no session id")
     prefix = f"{CODEX_USAGE_PROVIDER}:{sid}:"
-    try:
-        raw = index_path.read_text(encoding="utf-8")
-    except FileNotFoundError:
-        return None
-    except OSError as exc:
-        raise CodexSessionUsageError(
-            f"cannot read run index for the usage basis: {exc}"
-        ) from exc
-
     int_totals: dict[str, int | None] = {field: None for field in _BASELINE_INT_FIELDS}
     cost_total: float | None = None
     last_usage: Mapping[str, Any] | None = None
     snapshot_first_booked_at: Any = None
     binding_recorded = False
     seen_rows: set[tuple[str, str, str]] = set()
-    for line_number, line in enumerate(raw.split("\n"), start=1):
-        line = line.strip()
-        if not line:
-            continue
-        try:
-            row = json.loads(line)
-        except json.JSONDecodeError as exc:
-            raise CodexSessionUsageError(
-                f"run index row {line_number} is corrupt: {index_path}"
-            ) from exc
-        if not isinstance(row, dict):
-            continue
+    for line_number, row in _usage_index_records(index_path):
         raw_usage = row.get("usage")
         if not isinstance(raw_usage, dict):
             continue

tests/control_plane/test_codex_session_usage.py

@@ -11,11 +11,13 @@
 
 import json
 import threading
+import tracemalloc
 from datetime import datetime
 from pathlib import Path
 from typing import Any
 
 import pytest
+from canonical_authority_fixture import promoted_create_fixture
 
 import loopx.state_refresh as state_refresh_module
 from loopx.control_plane.quota.codex_session_usage import (
@@ -324,12 +326,13 @@ def test_session_usage_baseline_missing_index_means_first_observation(
     assert session_usage_baseline(tmp_path / "index.jsonl", SESSION_ID) is None
 
 
+@pytest.mark.parametrize("padding", ["", "\u0085", "\u2028"])
 def test_session_usage_baseline_telescopes_only_its_own_session_rows(
-    tmp_path: Path,
+    tmp_path: Path, padding: str,
 ) -> None:
     index_path = tmp_path / "index.jsonl"
     index_path.write_text(
-        "\n".join(
+        (padding + "\n").join(
             [
                 _index_row(
                     generated_at="2026-08-26T01:05:00",
@@ -387,6 +390,80 @@ def test_session_usage_baseline_fails_closed_on_corrupt_index_row(
         session_usage_baseline(index_path, SESSION_ID)
 
 
+def test_booking_reads_the_complete_ledger_without_buffering_other_history(
+    tmp_path: Path,
+) -> None:
+    index_path = tmp_path / "index.jsonl"
+    with index_path.open("w", encoding="utf-8") as stream:
+        stream.write(_index_row(
+            generated_at="2026-08-26T01:01:00", source_snapshot_id=f"codex:{SESSION_ID}:t1",
+            input_tokens=700, output_tokens=100, cache_tokens=0,
+            measurement_kind="absolute",
+        ) + "\n")
+        unrelated = json.dumps({"recommended_action": "x" * 8192}) + "\n"
+        for _ in range(512):
+            stream.write(unrelated)
+        stream.write(_index_row(
+            generated_at="2026-08-26T01:02:00", source_snapshot_id=f"codex:{SESSION_ID}:t2",
+            input_tokens=100, output_tokens=50, cache_tokens=0,
+            measurement_kind="delta",
+        ))
+    rollout = _fixture_rollout(tmp_path)
+    record: dict[str, Any] = {}
+    index_record: dict[str, Any] = {}
+    tracemalloc.start()
+    try:
+        book_codex_session_usage(record, rollout, index_path, index_record=index_record)
+        _, peak = tracemalloc.get_traced_memory()
+    finally:
+        tracemalloc.stop()
+    assert peak < 1024 * 1024
+    assert record["usage"]["measurement_kind"] == "delta"
+    assert record["usage"]["input_tokens"] == 400
+    assert record["usage"]["output_tokens"] == 150
+    assert record["usage"] == index_record["usage"]
+
+
+def test_index_read_failure_does_not_publish_a_partial_usage_booking(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    index_path = tmp_path / "index.jsonl"
+    index_path.write_text(_index_row(
+        generated_at="2026-08-26T01:01:00", source_snapshot_id=f"codex:{SESSION_ID}:t1",
+        input_tokens=700, output_tokens=100, measurement_kind="absolute",
+    ) + "\n", encoding="utf-8")
+    rollout = _fixture_rollout(tmp_path)
+    original_open = Path.open
+
+    class FailedIndex:
+        def __init__(self, stream):
+            self.stream = stream
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return self.stream.__exit__(*args)
+
+        def read(self, *args):
+            raise PermissionError("fixture index read failed")
+
+        def __iter__(self):
+            yield next(self.stream)
+            raise PermissionError("fixture index read failed")
+
+    def failed_open(path, *args, **kwargs):
+        stream = original_open(path, *args, **kwargs)
+        return FailedIndex(stream) if path == index_path else stream
+
+    monkeypatch.setattr(Path, "open", failed_open)
+    record: dict[str, Any] = {}
+    index_record: dict[str, Any] = {}
+    with pytest.raises(CodexSessionUsageError, match="cannot read run index"):
+        book_codex_session_usage(record, rollout, index_path, index_record=index_record)
+    assert record == index_record == {}
+
+
 @pytest.mark.parametrize(
     ("mutation", "error_match"),
     [
@@ -443,29 +520,11 @@ def test_book_codex_session_usage_fails_closed_on_invalid_baseline_without_appen
     assert "usage" not in index_record
 
 
-def _goal_fixture(tmp_path: Path) -> tuple[Path, Path, Path]:
-    project = tmp_path / "project"
-    state = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md"
-    state.parent.mkdir(parents=True)
-    state.write_text("# Active Goal State\n", encoding="utf-8")
-    registry_path = tmp_path / "registry.json"
-    registry_path.write_text(
-        json.dumps(
-            {
-                "goals": [
-                    {
-                        "id": GOAL_ID,
-                        "status": "active",
-                        "repo": str(project),
-                        "state_file": str(state.relative_to(project)),
-                    }
-                ]
-            }
-        ),
-        encoding="utf-8",
+def _goal_fixture(tmp_path: Path, *, provider: str = "sqlite") -> tuple[Path, Path, Path]:
+    registry_path, runtime_root, _ = promoted_create_fixture(
+        tmp_path, provider=provider, goal_id=GOAL_ID, registered_agents=[],
     )
-    runtime_root = tmp_path / "runtime"
-    return registry_path, project, runtime_root
+    return registry_path, tmp_path / "project", runtime_root
 
 
 def _refresh(
@@ -508,10 +567,11 @@ def _summary_totals(runtime_root: Path) -> dict[str, Any]:
     return summary["totals"]
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_refresh_state_codex_session_produces_non_zero_usage_summary(
-    tmp_path: Path,
+    tmp_path: Path, provider: str,
 ) -> None:
-    registry_path, project, runtime_root = _goal_fixture(tmp_path)
+    registry_path, project, runtime_root = _goal_fixture(tmp_path, provider=provider)
     rollout = _fixture_rollout(tmp_path)
 
     payload = _refresh(