fix(usage): reject session rebinding in Codex rollouts

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.
 

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

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. |

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: