From cdcb6d14084fe0e64c6019cc45eb8c1aec229359 Mon Sep 17 00:00:00 2001 From: shangzh0 <97216392+shangzh0@users.noreply.github.com> Date: Wed, 28 Oct 2026 11:12:48 +0000 Subject: [PATCH] fix(usage): reject session rebinding in Codex rollouts Signed-off-by: shangzh0 <97216392+shangzh0@users.noreply.github.com> --- docs/status-data-contract.md | 7 +- .../quota/codex_session_usage.py | 22 ++++- .../references/repair-patterns.md | 1 + .../control_plane/test_codex_session_usage.py | 90 +++++++++++++++++++ 4 files changed, 115 insertions(+), 5 deletions(-) diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index f63b9d74f2..107234f182 100644 --- a/docs/status-data-contract.md +++ b/docs/status-data-contract.md @@ -2421,7 +2421,12 @@ context without new usage cannot relabel that snapshot. A snapshot without a preceding model context is rejected. This binding does not allocate cumulative session usage across models. The session must be bound explicitly; there is no automatic session discovery, because guessing a concurrent session risks -attributing one session's spend to another run. A host that measures usage +attributing one session's spend to another run. The first `session_meta` must +identify the session before any cumulative usage is accepted. Missing ids or +conflicting later ids refuse the booking instead of rebinding earlier counters. +Repeated metadata for the same id preserves the first session clock, so it +cannot change an already-booked snapshot's duration on replay. +A host that measures usage itself can instead pass one finished per-run measurement with `--usage-json`. Without either flag, usage stays unknown. diff --git a/loopx/control_plane/quota/codex_session_usage.py b/loopx/control_plane/quota/codex_session_usage.py index a06924331b..090e8aafc3 100644 --- a/loopx/control_plane/quota/codex_session_usage.py +++ b/loopx/control_plane/quota/codex_session_usage.py @@ -145,12 +145,22 @@ def _read_codex_session_usage( raw_payload = item.get("payload") payload = raw_payload if isinstance(raw_payload, dict) else {} if kind == "session_meta": - session_id = str( + metadata_session_id = str( payload.get("session_id") or payload.get("id") or "" ).strip() - session_started_at = _parse_timestamp( - payload.get("timestamp") or item.get("timestamp") - ) + if not metadata_session_id: + raise CodexSessionUsageError( + f"codex session rollout has no session_meta id: {path}" + ) + if session_id and metadata_session_id != session_id: + raise CodexSessionUsageError( + f"codex session rollout has conflicting session_meta ids: {path}" + ) + if not session_id: + session_id = metadata_session_id + session_started_at = _parse_timestamp( + payload.get("timestamp") or item.get("timestamp") + ) elif kind == "turn_context": model_text = str(payload.get("model") or "").strip() if model_text: @@ -162,6 +172,10 @@ def _read_codex_session_usage( info = raw_info if isinstance(raw_info, dict) else {} totals = info.get("total_token_usage") if isinstance(totals, Mapping): + if not session_id: + raise CodexSessionUsageError( + f"codex session rollout has usage before session_meta: {path}" + ) last_totals = totals last_totals_at = str(item.get("timestamp") or "").strip() # The model labels this cumulative snapshot only if it was diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index 76147d599b..d812ee520d 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -5,6 +5,7 @@ teaches a reusable control-plane lesson. | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair | | --- | --- | --- | --- | --- | +| `codex_usage_session_rebinding` | Cumulative Codex token usage moves to a different session after another `session_meta`, or replaying the same snapshot loses its recorded duration after repeated metadata. | The explicitly bound synthetic rollout's ordered session ids, first clock, usage snapshot identity, and before/after run-index bytes and aggregate counters. | The usage reader overwrites identity and start time while scanning, allowing later metadata to reinterpret earlier counters. | Require identified metadata before usage, reject conflicting session ids, and retain the first clock for repeated same-id headers. Preserve streaming and the existing booking lock. Cover File/SQLite refusal without an append and repaired same-snapshot zero-delta recovery; never discover another session or rewrite historical usage. | | `implicit_clock_precedes_state_lock` | Concurrent state mutations reject a healthy request as a backward timestamp even though a fresh retry succeeds. | The exact operation, persisted `updated_at`, and a real-clock reproduction where the first caller pauses before acquiring the lock and a later caller commits first. | Implicit time was sampled before lock acquisition, so request order and commit order diverged. | Sample implicit mutation time inside the existing locked state transition. Preserve explicit backward-time rejection and observation-specific out-of-order rules. Cover every adjacent writer with reversed lock acquisition, unchanged-state negative cases, and the real CLI/file-lock path; do not clamp explicit input, weaken the guard, or add retries. | | `stride_inline_outcome_basis_ordering` | A batched milestone dry-run succeeds, but execution returns `stride_review_basis_mismatch` after an outcome refinement has committed. | Exact Turn/obligation/context, the returned `goal_outcome_update` status and operation, current acceptance and run/spend receipts; compare due Stride with ordinary batching. | Sequential refinement changes the context that the decision references; a preview using pre-update facts hides that change. | Feed the typed acceptance owner's prospective contract to dry-run planning. For a changed due-review milestone, refine before a fresh same-Turn guard, read the new facts and author the complete decision; unchanged work needs no update. On partial failure retain applied/replayed receipts and original IDs, reassess fresh context, then settle once. Never assume rollback, blindly replace a context token, lower the gate or replay work. Cover real File/SQLite preview/refusal/recovery, ordinary batching and single debit. | | `native_code_cell_source_normalization` | An optional native context adapter always falls back to full after an output-budget pragma, despite complete retained input. | The same live model call, native cell source, child identity and complete model-visible capture bytes; compare the executable source separately from native framing. | Native removes the documented first-line execution pragma before recording the cell, while the adapter compares that body with the unprocessed model call. | Normalize only that first-line JSON-object pragma and retain exact remaining source, live call/cell ordering and complete-byte verification. Cover unchanged framing, altered bodies, malformed/extra headers and stale or partial input, plus real native resume and recovery. Preserve automatic same-capture full fallback; do not add an admission or owner question. | diff --git a/tests/control_plane/test_codex_session_usage.py b/tests/control_plane/test_codex_session_usage.py index 5085bf9b3b..539717515b 100644 --- a/tests/control_plane/test_codex_session_usage.py +++ b/tests/control_plane/test_codex_session_usage.py @@ -138,6 +138,54 @@ def test_read_codex_session_usage_extracts_cumulative_totals(tmp_path: Path) -> assert "cost_usd" not in observation # unmeasured stays unknown, not zero +@pytest.mark.parametrize("position", [0, 2, 3]) +def test_read_codex_session_usage_rejects_conflicting_session_metadata( + tmp_path: Path, position: int, +) -> None: + lines = _fixture_rollout(tmp_path).read_text(encoding="utf-8").splitlines() + lines.insert(position, _rollout_header(OTHER_SESSION_ID)[0]) + rollout = _write_rollout(tmp_path / "rollout-mixed-sessions.jsonl", lines) + + with pytest.raises(CodexSessionUsageError, match="conflicting session_meta ids"): + read_codex_session_usage(rollout) + + +@pytest.mark.parametrize("id_key", ["id", "session_id"]) +def test_read_codex_session_usage_retains_first_same_session_clock( + tmp_path: Path, id_key: str, +) -> None: + rollout = _fixture_rollout(tmp_path) + original = read_codex_session_usage(rollout) + lines = rollout.read_text(encoding="utf-8").splitlines() + repeated = json.loads(_rollout_header()[0]) + repeated["payload"].pop("session_id") + repeated["payload"][id_key] = SESSION_ID + repeated["payload"]["timestamp"] = "2026-08-26T01:06:00.000Z" + lines.append(json.dumps(repeated)) + _write_rollout(rollout, lines) + + assert read_codex_session_usage(rollout) == original + + +def test_read_codex_session_usage_rejects_unidentified_metadata(tmp_path: Path) -> None: + lines = _fixture_rollout(tmp_path).read_text(encoding="utf-8").splitlines() + lines.insert(0, json.dumps({"type": "session_meta", "payload": {}})) + rollout = _write_rollout(tmp_path / "rollout-unidentified-session.jsonl", lines) + + with pytest.raises(CodexSessionUsageError, match="no session_meta id"): + read_codex_session_usage(rollout) + + +def test_read_codex_session_usage_requires_metadata_before_counters(tmp_path: Path) -> None: + lines = _fixture_rollout(tmp_path).read_text(encoding="utf-8").splitlines() + rollout = _write_rollout( + tmp_path / "rollout-late-session.jsonl", [lines[1], lines[2], lines[0]] + ) + + with pytest.raises(CodexSessionUsageError, match="usage before session_meta"): + read_codex_session_usage(rollout) + + def test_read_codex_session_usage_binds_model_when_token_snapshot_is_observed( tmp_path: Path, ) -> None: @@ -599,6 +647,48 @@ def test_refresh_state_codex_session_produces_non_zero_usage_summary( assert totals["duration_ms_24h"] == 300_000 +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_refresh_state_rejects_session_rebinding_and_recovers_without_rebooking( + tmp_path: Path, provider: str, +) -> None: + registry_path, project, runtime_root = _goal_fixture(tmp_path, provider=provider) + rollout = _fixture_rollout(tmp_path) + lines = rollout.read_text(encoding="utf-8").splitlines() + _refresh( + registry_path=registry_path, runtime_root=runtime_root, + project=project, usage_codex_session=rollout, + ) + index_path = runtime_root / "goals" / GOAL_ID / "runs" / "index.jsonl" + before = index_path.read_bytes() + totals = _summary_totals(runtime_root) + _write_rollout(rollout, [*lines, _rollout_header(OTHER_SESSION_ID)[0]]) + + with pytest.raises(CodexSessionUsageError, match="conflicting session_meta ids"): + _refresh( + registry_path=registry_path, runtime_root=runtime_root, + project=project, usage_codex_session=rollout, + ) + + assert index_path.read_bytes() == before + assert _summary_totals(runtime_root) == totals + assert session_usage_baseline(index_path, OTHER_SESSION_ID) is None + + repeated = json.loads(_rollout_header()[0]) + repeated["payload"]["timestamp"] = "2026-08-26T01:06:00.000Z" + _write_rollout(rollout, [*lines, json.dumps(repeated)]) + result = _refresh( + registry_path=registry_path, runtime_root=runtime_root, + project=project, usage_codex_session=rollout, + ) + + assert result["ok"] is True + assert result["usage"]["measurement_kind"] == "delta" + for field in ("input_tokens", "output_tokens", "cache_tokens", "duration_ms"): + assert result["usage"][field] == 0 + assert len(_index_runs(runtime_root)) == 2 + assert _summary_totals(runtime_root)["input_tokens_24h"] == totals["input_tokens_24h"] + + def test_refresh_state_replaying_same_snapshot_does_not_double_count( tmp_path: Path, ) -> None: