diff --git a/docs/reference/trajectory-assessment.md b/docs/reference/trajectory-assessment.md
index 3b35c010af..cbd6c92110 100644
--- a/docs/reference/trajectory-assessment.md
+++ b/docs/reference/trajectory-assessment.md
@@ -189,3 +189,72 @@ and historical states, inline source return and packaged frontend/backend
 readback. Present the key finding and uncertainty first, episode decisions next,
 and citations/method/cost on demand. CLI success alone does not complete that
 journey or qualify the model assessor.
+
+## Measured solver session cost (optional)
+
+A trusted pack curator can bind a **complete captured Codex session** separately
+from the model evidence. The existing register command accepts:
+
+```sh
+loopx benchmark trajectory-pack-register --goal-id OWNER_GOAL \
+  --input-json evidence-pack.json --source-root authorized-capture \
+  --title "Case · observation window" \
+  --usage-jsonl solver-session.jsonl --usage-session-id EXPECTED_SESSION_ID \
+  --format json
+```
+
+The capture path is relative to `source-root`, under the same containment rules.
+The expected session ID must be supplied with it. Registration reads the original
+file, verifies one unforked session with complete measured Turns, and retains its
+hash and a deterministic numeric projection. It never sends the session text to
+the analysis model. It does not execute, resume or change the solver.
+
+`solver_usage.projection` is returned by existing discovery and pack inspection,
+without a model call. It includes actual first/last event times, per-Turn source
+lines, input, cached input, uncached input and output tokens. Codex cumulative
+counter observations are differenced across Turns; summing the cumulative
+`turn.completed` snapshots would charge earlier work repeatedly. Cached input is
+part of input, and reasoning is part of output, not another additive cost.
+
+Complete counter histories use the existing private local snapshot transport
+when they exceed the ordinary 2 MiB RPC wire. Only this explicitly allowed
+method opts in; reduced snapshots retain the 64 MiB ceiling and digest checks.
+Oversized captures fail explicitly rather than dropping earlier Turns.
+
+This adapter deliberately rejects missing prefixes/counter gaps, resets, wrong
+session IDs, forked sessions, unmatched or unmeasured completions and a capture
+ending inside a Turn. Unsupported captures are not zero-cost sessions. The
+current adapter does not combine donor and continuation sessions. A curator must
+verify that the session belongs to the selected run; identity and hash checks do
+not independently establish that scientific association.
+
+The projection's own event window can differ from the curated episode window.
+`elapsed_seconds` spans its first Turn start to its last completion;
+`turn_elapsed_seconds` sums those non-overlapping timestamp intervals, excluding
+between-Turn gaps. Both include model/tool/orchestration time and are neither CPU
+usage nor token-generation time. They deliberately use event timestamps rather
+than mixing in provider-rounded `duration_ms`. Dollar, evaluator and inherited
+study costs remain unknown; no token price or waste fraction is inferred.
+
+Usage facts have their own provenance and applicability. A changed or missing
+capture leaves the captured values visible as historical, while unrelated model
+evidence retains its existing availability. Changing a bound capture produces a
+new pack ID but does not invalidate a reusable assessment of unchanged evidence.
+The source binding and projection are covered by that ID. Omit both options to
+retain the previous registration identity and response shape. To stop using this
+projection, register/select a pack without the optional capture; no solver change
+or destructive removal of historical evidence is necessary.
+
+Ownership stays with `benchmark-toolkit`: this built-in provider adapter produces
+a read-only observation-window diagnostic. The existing quota Codex reader owns
+live cumulative observations and idempotent ledger booking; it intentionally
+accepts partial active sessions and does not validate complete per-Turn windows.
+This diagnostic neither books usage nor replaces that authority. Python performs
+file containment and counter-only transport; TypeScript owns window validation
+and reduction, and the browser-safe assessment types expose the result.
+
+This is the measured-cost producer for the existing research workbench. Its
+HTTP consumer can read the same inspection field; numerical presentation and the
+full packaged user journey remain a companion acceptance. No new renderer path
+or command input, model execution permission, score authority or default analysis
+is introduced.
diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
index aa430d5323..1d9dafcc0d 100644
--- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
+++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
@@ -45,13 +45,14 @@ def _owner(runtime_root: Path, owner_goal_id: str) -> Path:
 
 
 def _pack_id(owner_goal_id: str, record: dict) -> str:
-    return _digest(
-        {
-            "owner_goal_id": owner_goal_id,
-            "analysis_key": record["analysis_key"],
-            "title": record["title"],
-        }
-    )
+    identity = {
+        "owner_goal_id": owner_goal_id,
+        "analysis_key": record["analysis_key"],
+        "title": record["title"],
+    }
+    if "solver_usage" in record:
+        identity["solver_usage"] = record["solver_usage"]
+    return _digest(identity)
 
 
 def _prepared(record: dict) -> dict:
@@ -62,18 +63,8 @@ def _prepared(record: dict) -> dict:
 
 
 def _verify_sources(record: dict) -> None:
-    root = Path(record["source_root"]).resolve(strict=True)
     for source in record["input"]["sources"]:
-        relative = PurePosixPath(source["artifact"])
-        if (
-            relative.is_absolute()
-            or ".." in relative.parts
-            or "\\" in source["artifact"]
-        ):
-            raise PackUnavailable("invalid_source_path")
-        path = (root / str(relative)).resolve(strict=True)
-        if not path.is_relative_to(root) or not path.is_file():
-            raise PackUnavailable("invalid_source_path")
+        path = _source_path(Path(record["source_root"]), source["artifact"])
         data = path.read_bytes()
         if hashlib.sha256(data).hexdigest() != source["artifact_sha256"]:
             raise PackUnavailable("source_changed")
@@ -84,6 +75,97 @@ def _verify_sources(record: dict) -> None:
             raise PackUnavailable("excerpt_mismatch")
 
 
+def _source_path(source_root: Path, artifact: str) -> Path:
+    relative = PurePosixPath(artifact)
+    if relative.is_absolute() or ".." in relative.parts or "\\" in artifact:
+        raise PackUnavailable("invalid_source_path")
+    root = source_root.resolve(strict=True)
+    path = (root / str(relative)).resolve(strict=True)
+    if not path.is_relative_to(root) or not path.is_file():
+        raise PackUnavailable("invalid_source_path")
+    return path
+
+
+def _usage_capture(source_root: Path, artifact: str, session_id: str) -> dict:
+    data = _source_path(source_root, artifact).read_bytes()
+    events = []
+    # Transport only the provider's counter/lifecycle fields, not model text,
+    # account metadata or tool outputs. TS owns all accounting decisions.
+    for line_number, line in enumerate(data.decode("utf-8").splitlines(), 1):
+        event = json.loads(line)
+        if not isinstance(event, dict):
+            raise ValueError("Usage capture contains a non-object event")
+        payload = event.get("payload") or {}
+        if not isinstance(payload, dict):
+            raise ValueError("Usage capture contains a non-object payload")
+        if event.get("type") == "session_meta":
+            payload = {k: payload[k] for k in ("id", "forked_from_id") if k in payload}
+        elif event.get("type") == "event_msg" and payload.get("type") in {
+            "task_started",
+            "task_complete",
+            "token_count",
+        }:
+            payload = {
+                k: payload[k] for k in ("type", "turn_id", "info") if k in payload
+            }
+            if isinstance(payload.get("info"), dict):
+                payload["info"] = {
+                    name: {
+                        key: value[key]
+                        for key in (
+                            "input_tokens",
+                            "cached_input_tokens",
+                            "output_tokens",
+                        )
+                        if key in value
+                    }
+                    if isinstance(value, dict)
+                    else value
+                    for name in ("total_token_usage", "last_token_usage")
+                    if (value := payload["info"].get(name)) is not None
+                }
+        else:
+            continue
+        events.append(
+            {
+                "line": line_number,
+                "type": event["type"],
+                "timestamp": event.get("timestamp"),
+                "payload": payload,
+            }
+        )
+    projection = effect_runtime_result(
+        "benchmark.trajectory.codex-usage",
+        {"events": events, "session_id": session_id},
+        large_local_snapshot=True,
+    )
+    return {
+        "artifact": artifact,
+        "artifact_sha256": hashlib.sha256(data).hexdigest(),
+        "projection": projection,
+    }
+
+
+def _usage_view(record: dict) -> dict | None:
+    captured = record.get("solver_usage")
+    if captured is None:
+        return None
+    reason = None
+    try:
+        data = _source_path(
+            Path(record["source_root"]), captured["artifact"]
+        ).read_bytes()
+        if hashlib.sha256(data).hexdigest() != captured["artifact_sha256"]:
+            reason = "source_changed"
+    except (OSError, ValueError):
+        reason = "source_unavailable"
+    return {
+        **captured,
+        "applicability": "historical" if reason else "current",
+        "reason_code": reason,
+    }
+
+
 def _index(owner: Path) -> dict:
     path = owner / "packs.json"
     if not path.exists():
@@ -141,6 +223,7 @@ def _summary(record: dict, reason: str | None) -> dict:
         "config": record["config"],
         "availability": "unavailable" if reason else "available",
         "reason_code": reason,
+        **({"solver_usage": _usage_view(record)} if "solver_usage" in record else {}),
     }
 
 
@@ -152,6 +235,8 @@ def register_pack(
     source_root: Path,
     title: str,
     config: dict | None = None,
+    usage_jsonl: str | None = None,
+    usage_session_id: str | None = None,
 ) -> dict:
     """Trusted operator only; no HTTP/renderer path-registration endpoint."""
     if not title.strip():
@@ -173,6 +258,14 @@ def register_pack(
         },
     }
     _verify_sources(record)
+    if (usage_jsonl is None) != (usage_session_id is None):
+        raise ValueError(
+            "Usage capture and expected session identity must be supplied together"
+        )
+    if usage_jsonl is not None:
+        record["solver_usage"] = _usage_capture(
+            source_root, usage_jsonl, usage_session_id
+        )
     owner = _owner(runtime_root, owner_goal_id)
     (owner / "packs").mkdir(parents=True, exist_ok=True, mode=0o700)
     pack_id = _pack_id(owner_goal_id, record)
diff --git a/loopx/cli_commands/benchmark_trajectory.py b/loopx/cli_commands/benchmark_trajectory.py
index 27dda857b3..e6d91ea7ff 100644
--- a/loopx/cli_commands/benchmark_trajectory.py
+++ b/loopx/cli_commands/benchmark_trajectory.py
@@ -33,6 +33,14 @@ def register_benchmark_trajectory_commands(subparsers, add_format) -> None:
     register.add_argument("--input-json", type=Path, required=True)
     register.add_argument("--source-root", type=Path, required=True)
     register.add_argument("--title", required=True)
+    register.add_argument(
+        "--usage-jsonl",
+        help="Optional complete Codex session capture, relative to source-root; read-only cost facts, not model input.",
+    )
+    register.add_argument(
+        "--usage-session-id",
+        help="Expected captured solver session identity; required with --usage-jsonl.",
+    )
     add_format(register)
     listing = subparsers.add_parser(
         "trajectory-packs",
@@ -77,6 +85,8 @@ def _handle_benchmark_trajectory_command(
                 input_path=args.input_json,
                 source_root=args.source_root,
                 title=args.title,
+                usage_jsonl=args.usage_jsonl,
+                usage_session_id=args.usage_session_id,
             )
             if args.benchmark_command == "trajectory-pack-register"
             else discover(**scope)
diff --git a/loopx/control_plane/capabilities/trajectory_assessment_types.ts b/loopx/control_plane/capabilities/trajectory_assessment_types.ts
index 31e699a2dd..c78859419e 100644
--- a/loopx/control_plane/capabilities/trajectory_assessment_types.ts
+++ b/loopx/control_plane/capabilities/trajectory_assessment_types.ts
@@ -29,3 +29,15 @@ export interface TrajectoryAssessment {
     assessed_fraction: number | null; missing_data_bounds: [number, number] | null}>;
   limitations: string[];
 }
+
+/** Provider-observed facts, separate from inferred episode judgments. */
+export interface TrajectorySolverUsage {
+  session_id: string; method: 'codex_complete_session_counters_v1';
+  start: string; end: string; completed_turns: number; counter_observations: number;
+  input_tokens: number; cached_input_tokens: number; uncached_input_tokens: number; output_tokens: number;
+  elapsed_seconds: number; turn_elapsed_seconds: number;
+  turns: Array<{turn_id: string; start: string; end: string; first_line: number; last_line: number;
+    elapsed_seconds: number; usage: {input_tokens: number; cached_input_tokens: number;
+      uncached_input_tokens: number; output_tokens: number}}>;
+  limitations: string[];
+}
diff --git a/loopx/control_plane/capabilities/trajectory_codex_usage.ts b/loopx/control_plane/capabilities/trajectory_codex_usage.ts
new file mode 100644
index 0000000000..229b534ee1
--- /dev/null
+++ b/loopx/control_plane/capabilities/trajectory_codex_usage.ts
@@ -0,0 +1,76 @@
+/** Read-only Codex session adapter for benchmark-toolkit, separate from model judgments. */
+import {EffectRuntimeRequestError} from '../effect_runtime_errors.ts';
+import type {TrajectorySolverUsage} from './trajectory_assessment_types.ts';
+
+const fail = (message: string): never => { throw new EffectRuntimeRequestError(`trajectory usage: ${message}`); };
+const counterKeys = ['input_tokens', 'cached_input_tokens', 'output_tokens'] as const;
+type Counters = Record<typeof counterKeys[number], number>;
+function counters(value: unknown): Counters {
+  const x = value as Record<string, unknown> | undefined;
+  if (!x || counterKeys.some(k => typeof x[k] !== 'number' || !Number.isSafeInteger(x[k]) || (x[k] as number) < 0)) return fail('missing or invalid counters');
+  if ((x.cached_input_tokens as number) > (x.input_tokens as number)) return fail('cached input exceeds input');
+  return Object.fromEntries(counterKeys.map(k => [k, x[k]])) as Counters;
+}
+const subtract = (a: Counters, b: Counters): Counters => Object.fromEntries(counterKeys.map(k => [k, a[k] - b[k]])) as Counters;
+const zero = (): Counters => ({input_tokens: 0, cached_input_tokens: 0, output_tokens: 0});
+const time = (v: unknown): number => {
+  if (typeof v !== 'string' || !/^\d{4}-\d\d-\d\dT.*(?:Z|[+-]\d\d:\d\d)$/.test(v) || !Number.isFinite(Date.parse(v))) return fail('timestamp must include timezone');
+  return Date.parse(v);
+};
+export function summarizeTrajectoryCodexUsage(params: Record<string, unknown>): TrajectorySolverUsage {
+  if (!Array.isArray(params.events) || typeof params.session_id !== 'string' || !params.session_id) return fail('session events and expected identity required');
+  const rows = params.events;
+  let sessionSeen = false, previousTime = -Infinity, previous = zero(), samples = 0;
+  let active: {turn_id: string; start: string; start_ms: number; first_line: number; baseline: Counters; samples: number} | null = null;
+  const turns: Array<{turn_id: string; start: string; end: string; first_line: number; last_line: number; elapsed_seconds: number; usage: Counters & {uncached_input_tokens: number}}> = [];
+  let previousLine = 0;
+  for (const event of rows) {
+    if (!event || typeof event !== 'object' || Array.isArray(event)) return fail('invalid event');
+    if (!Number.isSafeInteger(event.line) || event.line <= previousLine) return fail('invalid source line order');
+    previousLine = event.line;
+    const p = event.payload;
+    if (event.type === 'session_meta') {
+      if (event.line !== 1 || sessionSeen || p?.id !== params.session_id || p?.forked_from_id) return fail('expected one complete, unforked session');
+      sessionSeen = true;
+      continue;
+    }
+    if (!sessionSeen) return fail('session header missing');
+    if (event.type !== 'event_msg' || !['task_started', 'task_complete', 'token_count'].includes(p?.type)) continue;
+    const stamp = time(event.timestamp);
+    if (stamp < previousTime) return fail('out-of-order events');
+    previousTime = stamp;
+    if (p.type === 'task_started') {
+      if (active || typeof p.turn_id !== 'string' || !p.turn_id || turns.some(t => t.turn_id === p.turn_id)) return fail('duplicate or overlapping Turn');
+      active = {turn_id: p.turn_id, start: event.timestamp, start_ms: stamp, first_line: event.line, baseline: {...previous}, samples: 0};
+    } else if (p.type === 'token_count' && p.info) {
+      if (!active) return fail('usage outside a captured Turn');
+      const total = counters(p.info.total_token_usage), last = counters(p.info.last_token_usage);
+      const delta = subtract(total, previous);
+      if (samples === 0 && counterKeys.some(k => total[k] !== last[k])) return fail('missing initial usage prefix');
+      if (counterKeys.some(k => delta[k] < 0)) return fail('counter reset; do not silently splice sessions');
+      // Repeated observations are legal. A changed total must represent exactly
+      // the newly observed request, including the first request after zero.
+      if (counterKeys.some(k => delta[k] !== 0) && counterKeys.some(k => delta[k] !== last[k])) return fail('missing usage prefix or counter gap');
+      previous = total; samples++; active.samples++;
+    } else if (p.type === 'task_complete') {
+      if (!active || p.turn_id !== active.turn_id || !active.samples) return fail('unmatched or unmeasured completed Turn');
+      const usage = subtract(previous, active.baseline);
+      if (usage.cached_input_tokens > usage.input_tokens) return fail('invalid delta cache accounting');
+      turns.push({turn_id: active.turn_id, start: active.start, end: event.timestamp, first_line: active.first_line,
+        last_line: event.line, elapsed_seconds: (stamp - active.start_ms) / 1000,
+        usage: {...usage, uncached_input_tokens: usage.input_tokens - usage.cached_input_tokens}});
+      active = null;
+    }
+  }
+  if (active || !turns.length) return fail('capture must end after a completed Turn');
+  return {session_id: params.session_id, method: 'codex_complete_session_counters_v1',
+    start: turns[0].start, end: turns.at(-1)!.end, completed_turns: turns.length, counter_observations: samples,
+    input_tokens: previous.input_tokens, cached_input_tokens: previous.cached_input_tokens,
+    uncached_input_tokens: previous.input_tokens - previous.cached_input_tokens, output_tokens: previous.output_tokens,
+    elapsed_seconds: (time(turns.at(-1)!.end) - time(turns[0].start)) / 1000,
+    turn_elapsed_seconds: turns.reduce((sum, t) => sum + t.elapsed_seconds, 0), turns,
+    limitations: ['Complete captured session only; donor/prefix study costs and evaluator costs are not included.',
+      'Input includes cached input. Output includes any reasoning tokens. No dollar or generation-time estimate.',
+      'Elapsed time includes tools and orchestration; it does not measure CPU time or avoidable overhead.',
+      'Registered operator binding identifies the case; this does not independently prove run provenance.']};
+}
diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py
index 8e0faa6c25..29ec488dd1 100644
--- a/loopx/control_plane/effect_runtime.py
+++ b/loopx/control_plane/effect_runtime.py
@@ -42,6 +42,7 @@
 MAX_REQUEST_BYTES = 2 * 1024 * 1024
 MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024
 LOCAL_SNAPSHOT_METHODS = frozenset({
+    "benchmark.trajectory.codex-usage",
     "coordination.local_authority.todo_source",
     "todo.context.page",
     "work_item.context.project",
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index a055003a97..57bd22f70b 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -281,6 +281,7 @@ export function createEffectRuntimeHandlers(
     ["todo.user_completion.plan", lazyHandler(() => import("./todos/user_completion.ts"), ({evaluateUserCompletion}) => evaluateUserCompletion)],
     ["agent.capability_gate.evaluate", lazyHandler(() => import("./agents/capability_gate.ts"), ({evaluateCapabilityGate}) => evaluateCapabilityGate)],
     ["agent.capability_memory", lazyHandler(() => import("./agents/capability_memory.ts"), ({agentCapabilityMemory}) => agentCapabilityMemory)],
+    ["benchmark.trajectory.codex-usage", lazyHandler(() => import("./capabilities/trajectory_codex_usage.ts"), ({summarizeTrajectoryCodexUsage}) => summarizeTrajectoryCodexUsage)],
     ["benchmark.trajectory.prepare", lazyHandler(() => import("./capabilities/trajectory_assessment.ts"), ({prepareTrajectoryAssessment}) => prepareTrajectoryAssessment)],
     ["benchmark.trajectory.qualify", lazyHandler(() => import("./capabilities/trajectory_assessment.ts"), ({qualifyTrajectoryAssessment}) => qualifyTrajectoryAssessment)],
     ["agent.preferences", lazyHandler(() => import("./capabilities/agent_preferences.ts"), ({agentPreferences}) => agentPreferences)],
diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts
index 5084e58c7d..2ef72aac66 100644
--- a/loopx/control_plane/effect_runtime_server.ts
+++ b/loopx/control_plane/effect_runtime_server.ts
@@ -28,6 +28,7 @@ const MAX_REQUEST_BYTES = 2 * 1024 * 1024;
 const MAX_INLINE_RESPONSE_BYTES = 2 * 1024 * 1024;
 // Explicit opt-in: ordinary effects retain the 2 MiB request/response wire.
 const LOCAL_SNAPSHOT_METHODS = new Set([
+  "benchmark.trajectory.codex-usage",
   "coordination.local_authority.todo_source",
   "todo.context.page",
   "work_item.context.project",
diff --git a/tests/capabilities/test_trajectory_assessment_store.py b/tests/capabilities/test_trajectory_assessment_store.py
index 130fe6eaf4..eef8eac753 100644
--- a/tests/capabilities/test_trajectory_assessment_store.py
+++ b/tests/capabilities/test_trajectory_assessment_store.py
@@ -7,6 +7,7 @@
 import os
 import subprocess
 import sys
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -353,3 +354,248 @@ def upgraded_method(record):
     result_file.write_text("{}", encoding="utf-8")
     with pytest.raises(PackUnavailable, match="historical_result_unverified"):
         inspect_pack(**scope)
+
+
+def _usage_fixture():
+    def counts(i, c, o):
+        return dict(input_tokens=i, cached_input_tokens=c, output_tokens=o)
+
+    events = [
+        dict(type="session_meta", payload=dict(id="solver-session")),
+        dict(type="event_msg", payload=dict(type="task_started", turn_id="a")),
+        dict(
+            type="event_msg",
+            payload=dict(
+                type="token_count",
+                info=dict(
+                    total_token_usage=counts(100, 60, 10),
+                    last_token_usage=counts(100, 60, 10),
+                ),
+            ),
+        ),
+        dict(type="event_msg", payload=dict(type="task_complete", turn_id="a")),
+        dict(type="event_msg", payload=dict(type="task_started", turn_id="b")),
+        dict(
+            type="event_msg",
+            payload=dict(
+                type="token_count",
+                info=dict(
+                    total_token_usage=counts(140, 80, 15),
+                    last_token_usage=counts(40, 20, 5),
+                ),
+            ),
+        ),
+        dict(type="event_msg", payload=dict(type="task_complete", turn_id="b")),
+    ]
+    for i, event in enumerate(events):
+        event["timestamp"] = f"2026-01-01T00:00:0{i}Z"
+    return "\n".join(json.dumps(e) for e in events) + "\n"
+
+
+def test_optional_solver_usage_is_read_only_scoped_and_separate_from_model(pack):
+    scope, _, old, input_path, sources, _ = pack
+    assert "solver_usage" not in inspect_pack(**scope)
+    (sources / "usage.jsonl").write_text(_usage_fixture())
+    args = dict(
+        **scope,
+        input_path=input_path,
+        source_root=sources,
+        title=old["title"],
+        usage_jsonl="usage.jsonl",
+        usage_session_id="solver-session",
+    )
+    registered = register_pack(**args)
+    assert registered["pack_id"] != old["pack_id"]
+    assert registered["analysis_key"] == old["analysis_key"]
+    assert register_pack(**args) == registered
+    view = inspect_pack(**scope)
+    cost = view["solver_usage"]
+    assert cost["projection"]["input_tokens"] == 140
+    assert cost["projection"]["uncached_input_tokens"] == 60
+    assert cost["projection"]["turns"][1]["usage"]["input_tokens"] == 40
+    assert view["job"] is None and view["result"] is None
+    assert discover(**scope)["packs"][-1]["solver_usage"] == cost
+    assert not list(scope["runtime_root"].rglob("receipt.json"))
+    (sources / "usage.jsonl").write_text("changed capture")
+    stale = inspect_pack(**scope)["solver_usage"]
+    assert stale["applicability"] == "historical"
+    assert stale["reason_code"] == "source_changed"
+    assert stale["projection"] == cost["projection"]
+    (sources / "usage.jsonl").unlink()
+    assert inspect_pack(**scope)["solver_usage"]["reason_code"] == "source_unavailable"
+    assert "solver_usage" not in inspect_pack(**scope, pack_id=old["pack_id"])
+
+
+def test_usage_registration_rejects_wrong_identity_paths_gaps_and_tampering(pack):
+    from loopx.control_plane.effect_runtime import EffectRuntimeRejected
+
+    scope, _, old, input_path, sources, _ = pack
+    args = dict(
+        **scope,
+        input_path=input_path,
+        source_root=sources,
+        title=old["title"],
+        usage_jsonl="usage.jsonl",
+        usage_session_id="solver-session",
+    )
+    path = sources / "usage.jsonl"
+    path.write_text(_usage_fixture())
+    with pytest.raises(ValueError, match="supplied together"):
+        register_pack(**{**args, "usage_session_id": None})
+    with pytest.raises(EffectRuntimeRejected, match="session"):
+        register_pack(**{**args, "usage_session_id": "wrong-session"})
+    with pytest.raises(PackUnavailable, match="invalid_source_path"):
+        register_pack(**{**args, "usage_jsonl": "../usage.jsonl"})
+    outside = sources.parent / "outside.jsonl"
+    outside.write_text(_usage_fixture())
+    (sources / "escaping.jsonl").symlink_to(outside)
+    with pytest.raises(PackUnavailable, match="invalid_source_path"):
+        register_pack(**{**args, "usage_jsonl": "escaping.jsonl"})
+    path.write_text(
+        _usage_fixture().replace('"input_tokens": 140', '"input_tokens": 240')
+    )
+    with pytest.raises(EffectRuntimeRejected, match="gap"):
+        register_pack(**args)
+    path.write_text(_usage_fixture())
+    registered = register_pack(**args)
+    stored = next(scope["runtime_root"].rglob(registered["pack_id"] + ".json"))
+    row = json.loads(stored.read_text())
+    row["solver_usage"]["projection"]["input_tokens"] = 0
+    stored.write_text(json.dumps(row))
+    with pytest.raises(PackUnavailable, match="invalid_registration"):
+        inspect_pack(**scope)
+
+
+def test_large_usage_real_cli_register_readback_and_replay(pack, monkeypatch):
+    from loopx.control_plane import effect_runtime
+
+    scope, _, old, input_path, sources, _ = pack
+    start = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    rows = [dict(type="session_meta", payload=dict(id="solver-session"))]
+    for turn in range(1, 4501):
+        payloads = [
+            dict(type="task_started", turn_id=str(turn)),
+            dict(
+                type="token_count",
+                info=dict(
+                    total_token_usage=dict(
+                        input_tokens=100 * turn,
+                        cached_input_tokens=60 * turn,
+                        output_tokens=10 * turn,
+                    ),
+                    last_token_usage=dict(
+                        input_tokens=100, cached_input_tokens=60, output_tokens=10
+                    ),
+                ),
+            ),
+            dict(type="task_complete", turn_id=str(turn)),
+        ]
+        for offset, payload in enumerate(payloads):
+            rows.append(
+                dict(
+                    type="event_msg",
+                    payload=payload,
+                    timestamp=(
+                        start + timedelta(seconds=3 * turn + offset)
+                    ).isoformat(),
+                )
+            )
+    capture = sources / "usage.jsonl"
+    capture.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
+    # This is larger even after the adapter's content reduction, not merely a
+    # large raw transcript made small by discarding model messages.
+    params = dict(
+        session_id="solver-session",
+        events=[
+            dict(line=i + 1, timestamp=None, **row)
+            if "timestamp" not in row
+            else dict(line=i + 1, **row)
+            for i, row in enumerate(rows)
+        ],
+    )
+    assert (
+        len(json.dumps(params, separators=(",", ":")).encode())
+        > effect_runtime.MAX_REQUEST_BYTES
+    )
+    registry = sources.parent / "registry.json"
+    registry.write_text('{"projects":{}}')
+    prefix = [
+        sys.executable,
+        "-m",
+        "loopx.entrypoint",
+        "--registry",
+        str(registry),
+        "--runtime-root",
+        str(scope["runtime_root"]),
+        "--format",
+        "json",
+        "benchmark",
+    ]
+
+    def cli(*args):
+        result = subprocess.run([*prefix, *args], text=True, capture_output=True)
+        assert result.returncode == 0, result.stdout + result.stderr
+        return json.loads(result.stdout)
+
+    args = [
+        "trajectory-pack-register",
+        "--goal-id",
+        scope["owner_goal_id"],
+        "--input-json",
+        str(input_path),
+        "--source-root",
+        str(sources),
+        "--title",
+        old["title"],
+        "--usage-jsonl",
+        "usage.jsonl",
+        "--usage-session-id",
+        "solver-session",
+    ]
+    registered = cli(*args)
+    assert cli(*args) == registered
+    view = inspect_pack(**scope, pack_id=registered["pack_id"])
+    cost = view["solver_usage"]
+    assert cost["artifact_sha256"] == hashlib.sha256(capture.read_bytes()).hexdigest()
+    facts = cost["projection"]
+    assert facts["completed_turns"] == 4500
+    assert facts["input_tokens"] == 450000
+    assert facts["cached_input_tokens"] == 270000
+    assert facts["uncached_input_tokens"] == 180000
+    assert facts["output_tokens"] == 45000
+    assert facts["turns"][0]["first_line"] == 2
+    assert facts["turns"][-1]["last_line"] == 13501
+    assert all(t["usage"]["input_tokens"] == 100 for t in facts["turns"])
+    listing = cli("trajectory-packs", "--goal-id", scope["owner_goal_id"])
+    assert (
+        len(listing["packs"]) == 2
+    )  # original optional-off pack plus one replayed pack
+    assert (
+        next(p for p in listing["packs"] if p["pack_id"] == registered["pack_id"])[
+            "solver_usage"
+        ]
+        == cost
+    )
+    assert "solver_usage" not in inspect_pack(**scope, pack_id=old["pack_id"])
+    assert view["job"] is None and view["result"] is None
+
+    # The method opts into a bounded channel, not a larger ordinary RPC wire.
+    with pytest.raises(
+        effect_runtime.EffectRuntimeRejected, match="request is oversized"
+    ):
+        effect_runtime.effect_runtime_result("benchmark.trajectory.codex-usage", params)
+    monkeypatch.setattr(
+        effect_runtime, "MAX_LOCAL_SNAPSHOT_BYTES", effect_runtime.MAX_REQUEST_BYTES
+    )
+    with pytest.raises(
+        effect_runtime.EffectRuntimeRejected, match="local snapshot is oversized"
+    ):
+        register_pack(
+            **scope,
+            input_path=input_path,
+            source_root=sources,
+            title="over limit",
+            usage_jsonl="usage.jsonl",
+            usage_session_id="solver-session",
+        )
+    assert len(discover(**scope)["packs"]) == 2
diff --git a/tests/control_plane_ts/trajectory_codex_usage.test.ts b/tests/control_plane_ts/trajectory_codex_usage.test.ts
new file mode 100644
index 0000000000..6d7c8ec395
--- /dev/null
+++ b/tests/control_plane_ts/trajectory_codex_usage.test.ts
@@ -0,0 +1,40 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {summarizeTrajectoryCodexUsage as summarize} from '../../loopx/control_plane/capabilities/trajectory_codex_usage.ts';
+const usage = (input: number, cached: number, output: number) => ({input_tokens:input,cached_input_tokens:cached,output_tokens:output});
+function fixture() {
+  return [
+    {type:'session_meta', payload:{id:'session'}},
+    {type:'event_msg',payload:{type:'task_started',turn_id:'a'}},
+    {type:'event_msg',payload:{type:'token_count',info:{total_token_usage:usage(100,60,10),last_token_usage:usage(100,60,10)}}},
+    {type:'event_msg',payload:{type:'task_complete',turn_id:'a'}},
+    {type:'event_msg',payload:{type:'task_started',turn_id:'b'}},
+    {type:'event_msg',payload:{type:'token_count',info:{total_token_usage:usage(140,80,15),last_token_usage:usage(40,20,5)}}},
+    {type:'event_msg',payload:{type:'task_complete',turn_id:'b'}},
+  ].map((e,i)=>({...e,line:i+1,timestamp:`2026-01-01T00:00:0${i}Z`})) as any[];
+}
+const run=(events=fixture())=>summarize({events,session_id:'session'});
+test('cumulative snapshots telescope; cached input is a subset and waiting stays separate',()=>{
+  const x=run();assert.equal(x.input_tokens,140);assert.equal(x.uncached_input_tokens,60);
+  assert.equal(x.turns[1].usage.input_tokens,40);assert.equal(x.output_tokens,15);
+  assert.equal(x.elapsed_seconds,5);assert.equal(x.turn_elapsed_seconds,4);
+  assert.equal(x.turns[1].first_line,5);assert.equal(x.turns[1].last_line,7);
+});
+test('duplicate counter observation does not charge twice',()=>{
+  const es=fixture();es.splice(3,0,structuredClone(es[2]));es.forEach((e,i)=>e.line=i+1);
+  assert.equal(run(es).input_tokens,140);
+});
+test('unknown counters, resets, missing prefix, session mismatch and unfinished work reject',()=>{
+  for(const edit of [
+    (e:any[])=>{e[0].payload.id='other';},
+    (e:any[])=>{e[0].payload.forked_from_id='parent';},
+    (e:any[])=>{e[2].payload.info.total_token_usage.input_tokens=200;},
+    (e:any[])=>{e[5].payload.info.total_token_usage.input_tokens=50;},
+    (e:any[])=>{delete e[5].payload.info.total_token_usage.cached_input_tokens;},
+    (e:any[])=>{e[5].payload.info.total_token_usage.cached_input_tokens=150;},
+    (e:any[])=>{e[6].payload.turn_id='unknown';},
+    (e:any[])=>{e[5].timestamp='2025-01-01T00:00:00Z';},
+    (e:any[])=>{e.pop();},
+    (e:any[])=>{e.splice(5,1);},
+  ]) {const es=fixture();edit(es);assert.throws(()=>run(es));}
+});