diff --git a/docs/architecture/rfcs/goal-usage-token-cost-v0.md b/docs/architecture/rfcs/goal-usage-token-cost-v0.md index 96a4b81f85..bb4d6000a9 100644 --- a/docs/architecture/rfcs/goal-usage-token-cost-v0.md +++ b/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. diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index f529330596..f63b9d74f2 100644 --- a/docs/status-data-contract.md +++ b/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 diff --git a/loopx/control_plane/quota/codex_session_usage.py b/loopx/control_plane/quota/codex_session_usage.py index 36d0c3679d..a06924331b 100644 --- a/loopx/control_plane/quota/codex_session_usage.py +++ b/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 diff --git a/tests/control_plane/test_codex_session_usage.py b/tests/control_plane/test_codex_session_usage.py index 166051104f..5085bf9b3b 100644 --- a/tests/control_plane/test_codex_session_usage.py +++ b/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(