feat(benchmark): expose measured solver usage in trajectory packs

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.

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)

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)

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[];
+}

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.']};
+}

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",