diff --git a/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.css b/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.css index fc359252b4..e623c5f33a 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.css @@ -92,6 +92,9 @@ .research-usage-tail { border-inline-start: 2px solid var(--pw-caution); padding-inline-start: 16px; margin-block: 20px; } .research-usage-tail strong { font-weight: 500; color: var(--pw-text); } .research-request-coverage { border-block-start: 1px solid var(--pw-line); padding-block-start: 16px; margin-block: 20px; } +.research-input-share-value { display: flex; align-items: baseline; flex-wrap: wrap; gap: 12px; } +.research-input-share-value strong { font-size: var(--text-title); font-variant-numeric: tabular-nums; } +.research-input-share-value > span { color: var(--pw-muted); } .research-request-values { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 12px 24px; margin-block: 12px; } .research-request-values > div { min-width: 0; } .research-request-values dt { font-size: var(--text-body); color: var(--pw-muted); } diff --git a/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.tsx b/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.tsx index 19756f8b1e..5f8d312588 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/goal-research-progress.tsx @@ -29,6 +29,38 @@ export function verdictLabel(metric: TrajectoryMetric, verdict: AssessmentVerdic return verdict === "yes" ? (zh ? "有依据" : "Supported") : (zh ? "未建立" : "Not established"); } +function LoopXInputShare({usage, zh}: {usage: TrajectorySolverUsage | undefined; zh: boolean}) { + const attribution = usage?.loopx_input; + const percent = (n: number) => new Intl.NumberFormat(zh ? "zh-CN" : "en", {style: "percent", maximumFractionDigits: 1}).format(n); + const measured = attribution && attribution.status !== "unavailable" ? attribution : null; + const estimated = attribution?.method === "tiktoken_o200k_base_input_estimate_v1"; + return
+

{estimated ? (zh ? "LoopX Turn packet · 估计" : "LoopX Turn packet · estimate") : (zh ? "LoopX 输入占比" : "LoopX input share")}

+

{zh ? "同一捕获窗口内，LoopX packet 进入各次模型请求的 token 总量 ÷ 全部请求输入 token 总量，包含缓存输入。重复消费按各次请求计算，同一响应的重复回执只计一次。" : "LoopX packet tokens processed across requests divided by all request input tokens in the same capture, including cached input. Reprocessed context counts in each request; duplicate receipts for one response count once."}

+

{estimated ? (zh ? "分子仅计宿主明确记录的 Turn packet 文本，使用 o200k_base 分词估计；分母是同请求的原生输入计数。尚未核验原生模型分词器，不含其他 LoopX 命令输出、封装或压缩摘要，缓存归属未知；这是已记录部分的估计，不是 provider 实测或净效率损失。" : "Only explicitly recorded Turn packet text is estimated with o200k_base; the denominator is native input for the same requests. The provider model's tokenizer is unverified. Other LoopX tool output, framing and compacted summaries are excluded; cache allocation is unknown. This estimates the recorded component, not provider-measured attribution or net efficiency loss.") : (zh ? "需要宿主提供逐请求 token 归因；包字节、字符数和命令次数不能替代。缺失归因保持未知；上下文压缩也不自动全部归为 LoopX。这个比例衡量结构开销，不能单独证明净效率损失或因果影响。" : "Requires host-supplied per-request token attribution, not packet bytes, characters or command counts. Missing attribution stays unknown; compaction is not automatically all attributed to LoopX. This measures input overhead, not net efficiency loss or causal impact.")}

+
+

{measured?.input_fraction == null ? "—" : percent(measured.input_fraction)} + {measured ? {measured.attributed_requests} / {measured.total_requests} {estimated ? (zh ? "次请求已匹配" : "requests matched") : (zh ? "次请求已归因" : "requests attributed")} : null} +

+ {!attribution ?

{zh ? "未采集 packet 的 token 归因" : "Packet token attribution was not captured"}

+ : attribution.status === "unavailable" ?

{estimated ? (zh ? "输入估计暂不可用，占比未知。" : "Input estimate unavailable; share is unknown.") : (zh ? "归因记录未通过核对，占比未知。" : "Attribution records could not be reconciled; share is unknown.")}

+ : attribution.status === "partial" ?

{estimated ? (zh ? "部分请求已匹配 · 全窗口占比未知" : "Some requests matched · whole-capture share unknown") : (zh ? "部分归因 · 全窗口占比未知" : "Partial attribution · whole-capture share unknown")}

+ : attribution.total_input_tokens === 0 ?

{zh ? "捕获输入为零，比例不适用。" : "Captured input is zero; the ratio is undefined."}

: null} + {attribution ?
{estimated ? (zh ? "估计明细" : "Estimate details") : (zh ? "归因明细" : "Attribution details")} + {measured ?
+
{estimated ? (zh ? "已记录 packet 文本估计" : "Recorded packet text estimate") : (zh ? "已归因 LoopX 输入" : "Attributed LoopX input")}
{measured.loopx_input_tokens.toLocaleString()} tokens
+
{zh ? "全部请求输入" : "All request input"}
{measured.total_input_tokens.toLocaleString()} tokens
+
{zh ? "已覆盖请求输入" : "Input in attributed requests"}
{measured.attributed_input_tokens.toLocaleString()} tokens
+
{zh ? "LoopX 输入中的缓存" : "Cached subset of LoopX input"}
{measured.loopx_cached_input_tokens?.toLocaleString() ?? "—"} tokens
+
{zh ? "占未缓存输入" : "Share of uncached input"}
{measured.uncached_input_fraction == null ? "—" : percent(measured.uncached_input_fraction)}
+ {measured.status === "partial" && measured.input_fraction_bounds ? <>
{zh ? "缺失归因的可能范围" : "Possible range with missing attribution"}
{measured.input_fraction_bounds.map(percent).join(" – ")}

{zh ? "假定未归因请求全非 LoopX / 全为 LoopX；不是置信区间。" : "Unattributed requests treated as none / all LoopX; not a confidence interval."}

: null} +
:

{attribution.status === "unavailable" ? attribution.reason : null}

} + {measured?.estimate_basis ?

{measured.estimate_basis.tokenizer} {measured.estimate_basis.version} · {measured.estimate_basis.encoding}
{measured.estimate_basis.models.join(", ")}

: null} +

{attribution.method}

+
: null} +
; +} + // The assessment owner supplies localized questions. Both readers use that // copy without inventing short titles or translating the original evidence. export const episodeQuestion = (result: TrajectoryAssessment, id: string, episode?: TrajectoryEpisode) => @@ -194,6 +226,7 @@ export function ResearchProgress({pack: p, packs, selectedId, zh, onSelect, onCi

{incomplete.usage ? <>{zh ? "其中末轮" : "Of which, unfinished Turn"} · {zh ? "输入" : "input"} {number(incomplete.usage.input_tokens)} · {zh ? "输出" : "output"} {number(incomplete.usage.output_tokens)} tokens : (zh ? "末轮尚无计数观测，不记零" : "No counter observation for the unfinished Turn; not zero")}

{zh ? "仅含已持久化用量，不能据此判定运行已结束。" : "Persisted usage only; this does not establish termination."}

: null} + {usage ?

{zh ? "请求覆盖" : "Request coverage"}

{zh ? "逐响应去重后，核对轮次与会话累计量、上下文压缩记录及会话计数。以下是独立核对口径，与上方会话计数不相加，也不改写 Goal 账本。" : "Deduplicated responses are reconciled against Turn and session totals, compaction records and session counters. This separate basis is not added to the session counters above and does not rewrite the Goal ledger."}

diff --git a/docs/development/assets/research-input-share/after-desktop.png b/docs/development/assets/research-input-share/after-desktop.png new file mode 100644 index 0000000000..dc95f7cecc Binary files /dev/null and b/docs/development/assets/research-input-share/after-desktop.png differ diff --git a/docs/development/assets/research-input-share/after-detail-narrow.png b/docs/development/assets/research-input-share/after-detail-narrow.png new file mode 100644 index 0000000000..c70a409fe9 Binary files /dev/null and b/docs/development/assets/research-input-share/after-detail-narrow.png differ diff --git a/docs/development/assets/research-input-share/after-expanded-desktop.png b/docs/development/assets/research-input-share/after-expanded-desktop.png new file mode 100644 index 0000000000..d86659b738 Binary files /dev/null and b/docs/development/assets/research-input-share/after-expanded-desktop.png differ diff --git a/docs/development/assets/research-input-share/after-help-narrow.png b/docs/development/assets/research-input-share/after-help-narrow.png new file mode 100644 index 0000000000..547f59672a Binary files /dev/null and b/docs/development/assets/research-input-share/after-help-narrow.png differ diff --git a/docs/development/assets/research-input-share/after-partial-desktop.png b/docs/development/assets/research-input-share/after-partial-desktop.png new file mode 100644 index 0000000000..646cfd6b0b Binary files /dev/null and b/docs/development/assets/research-input-share/after-partial-desktop.png differ diff --git a/docs/development/assets/research-input-share/before-desktop.png b/docs/development/assets/research-input-share/before-desktop.png new file mode 100644 index 0000000000..9d76f0f269 Binary files /dev/null and b/docs/development/assets/research-input-share/before-desktop.png differ diff --git a/docs/development/assets/research-input-share/before-expanded-desktop.png b/docs/development/assets/research-input-share/before-expanded-desktop.png new file mode 100644 index 0000000000..c6c33f5242 Binary files /dev/null and b/docs/development/assets/research-input-share/before-expanded-desktop.png differ diff --git a/docs/development/assets/research-input-share/estimate-detail-narrow.png b/docs/development/assets/research-input-share/estimate-detail-narrow.png new file mode 100644 index 0000000000..8c00150ad3 Binary files /dev/null and b/docs/development/assets/research-input-share/estimate-detail-narrow.png differ diff --git a/docs/development/assets/research-input-share/estimate-expanded-desktop.png b/docs/development/assets/research-input-share/estimate-expanded-desktop.png new file mode 100644 index 0000000000..164bc6125b Binary files /dev/null and b/docs/development/assets/research-input-share/estimate-expanded-desktop.png differ diff --git a/docs/development/assets/research-input-share/estimate-first-desktop.png b/docs/development/assets/research-input-share/estimate-first-desktop.png new file mode 100644 index 0000000000..eeb0fa0dae Binary files /dev/null and b/docs/development/assets/research-input-share/estimate-first-desktop.png differ diff --git a/docs/development/assets/research-input-share/estimate-first-narrow.png b/docs/development/assets/research-input-share/estimate-first-narrow.png new file mode 100644 index 0000000000..a6fb8f38dd Binary files /dev/null and b/docs/development/assets/research-input-share/estimate-first-narrow.png differ diff --git a/docs/development/assets/research-input-share/estimate-help-narrow.png b/docs/development/assets/research-input-share/estimate-help-narrow.png new file mode 100644 index 0000000000..90cc0cf8b9 Binary files /dev/null and b/docs/development/assets/research-input-share/estimate-help-narrow.png differ diff --git a/docs/reference/trajectory-assessment.md b/docs/reference/trajectory-assessment.md index 393364cfca..06c5d6396b 100644 --- a/docs/reference/trajectory-assessment.md +++ b/docs/reference/trajectory-assessment.md @@ -686,3 +686,122 @@ existing TypeScript owner; replacement history and response text are excluded. Registration retains the source hash, and changed or missing sources mark the saved capture historical. Re-registering a source with this additional projection can create a new pack identity; old packs are never rewritten. + +### LoopX packet share of model input + +Research → Sources and usage also exposes **LoopX input share**. For the same +reconciled requests and capture window, the metric is: + +`sum(LoopX packet input tokens) / sum(all request input tokens)` + +Both include cached input. Reprocessed packet context counts in each request; +duplicate receipts and compaction mirrors count once. A compaction request is +not automatically all LoopX input. The separate uncached share requires an +explicit cache attribution, rather than charging every LoopX token as uncached. +This is a measured resource share, not a percentage loss of task effectiveness, +avoidable time, monetary cost or LoopX's causal impact. Compare native outcomes +and complete costs separately; a lower packet share alone is not an improvement. + +The existing frozen usage JSONL import accepts an optional **host-produced +attribution receipt**, bound to a native `token_usage_record` in the same +capture. For example, a synthetic receipt for a request with 100 input tokens: + +```json +{"type":"loopx_input_token_attribution","payload":{ + "method":"provider_input_token_attribution_v1", + "session_id":"synthetic-session","thread_id":"synthetic-session", + "turn_id":"synthetic-turn","root_turn_id":"synthetic-turn", + "response_id":"synthetic-response", + "input_tokens":100,"loopx_input_tokens":20,"loopx_cached_input_tokens":12 +}} +``` + +The host is responsible for classifying the actual delivered packet spans and +counting them on the **same provider token basis** as that request's input. +Count each span once even when several LoopX components describe it. References +or packets emitted but not consumed are not processed input. Do not classify +arbitrary prose or commands as LoopX usage by substring matching, divide byte or +character counts into estimated tokens, or append invented receipts to original +client logs. A recorder may export an immutable annotated capture; preserve the +original log separately. The import validates identities and arithmetic, not +the truth of a host's attribution or its billing completeness. Raw prompts, +packet text and account metadata never cross the counter transport. + +This reader does **not** add instrumentation to Codex or reconstruct attribution +from historical counters. Existing Codex captures without these receipts retain +their exact saved projection and show **uncaptured**, not zero. Hosts capable of +producing attribution use the existing `--usage-jsonl` registration path; no +model call, second accounting store or new configuration is introduced. + +With provider receipts, the TypeScript usage owner adds +`solver_usage.projection.loopx_input`. Every receipt must bind the original session, response, +Turn/root identity and input total. Counts must fit the native input/cache +subsets; conflicts withhold only attribution, leaving valid session and request +counters intact. Complete coverage shows the ratio. Partial coverage retains +the attributed request/input denominators and withholds the whole-window ratio; +its optional bounds allocate missing request input as none/all LoopX, **not a +confidence interval**. Zero input has an undefined ratio. Missing cache +attribution keeps the uncached ratio unknown. Source changes retain historical +values with the existing warning and never borrow another pack's measurement. + +中文：同一请求窗口内，LoopX packet 的实际输入 token 总量除以全部输入 token 总量， +含缓存；重复处理按请求计量，重复回执不重复记账。需要宿主真实归因，现有历史计数 +不能反推 packet token；未采集、部分覆盖与真实零分开。比例是结构开销，不能单独 +当作净效率损失、可避免成本或因果贡献。普通阅读不调用模型。 + +### Recorded Turn packet estimate + +The existing native `codex-cli` Turn host can record its actual injected packet +when the invocation already has `CODEX_ROLLOUT_TRACE_ROOT` enabled. This is a +local, opt-in Codex diagnostic feature; LoopX does not enable it globally or +change a running experiment. The Agent prepares a private, invocation-scoped +trace directory through the existing host environment. Native tracing contains +sensitive prompts and outputs; keep it and `loopx-input-sources/*.json` private. +The host records the exact injection only beside its newly observed session +bundle. It never edits the original client rollout or changes task results, +session authority, permissions, account selection or model execution. + +An Agent can prepare the optional computation provider with +`uv sync --extra input-token-estimates` in a source checkout, or install the +`input-token-estimates` package extra in its existing qualified runtime. This +uses [tiktoken 0.12.0](https://pypi.org/project/tiktoken/0.12.0/) and the explicitly named `o200k_base` encoding. Its first +encoding load can download the public encoding asset; packet text stays local. +It does **not** establish the tokenizer of an unknown provider model. + +Register a frozen host source with the existing usage capture using +`--loopx-input-json ` alongside `--usage-jsonl` and +`--usage-session-id`. The Agent prepares those artifacts within the existing +source root; the owner need not discover ids or copy a command. Registration +reads the native trace and tokenizes locally, without running a model. Ordinary +Research reads use the saved projection and make no tokenizer or model call. + +The same TS request owner binds the recorded source digest to one unambiguous +native input item, then follows that item and content across full requests. +Each native response, Turn, model and all six counters must match the existing +usage capture. Repeated consumption counts per request; duplicate responses +do not increase totals, and copies under another item id are not classified as +extra LoopX input. Changed source files become historical. Missing dependencies, +invalid references and conflicting identities withhold only this observation, +preserving valid native totals. Compressed websocket requests need verified +replay; compaction and uncovered requests cannot supply invented attribution. + +Research labels this **LoopX Turn packet · estimate** with method +`tiktoken_o200k_base_input_estimate_v1`. Its numerator covers only the explicitly +recorded Turn packet text; it excludes other LoopX tool output, model framing +and compacted summaries. It is not the total LoopX footprint. The denominator +is native input, including cache, for the same captured requests. The native +model, encoding and tokenizer version remain available in details. Unknown +cache allocation stays unknown. Partial request coverage withholds the whole +capture ratio; estimates provide no mathematical or confidence bounds. A token +estimate larger than its native input is withheld, never clamped. Existing +provider-measured receipts take precedence; the two methods are not mixed. + +This producer supplies a recorded-component estimate, not a scientific +calibration of model tokenization, a retrospective reconstruction, or proof of +net efficiency loss. Existing uninstrumented experiments remain unknown; do not +launch or retune a frozen solver just to populate this field. + +中文：Agent 可在既有获准调用的私有原生 trace 中记录真实注入的 Turn packet， +并在登记同一会话用量时以 o200k_base 做明确标注的文本估计。只统计已记录的 +Turn packet，不冒充全部 LoopX 输入或 provider 实测；原生模型分词器、缓存归属、 +压缩回放和缺失范围保持各自边界。原始记录不改，读取不调用模型，不重启既有实验。 diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py index ca7cae74e4..1b45d3e547 100644 --- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py +++ b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py @@ -126,6 +126,8 @@ def _usage_capture( session_id: str, *, include_incomplete: bool = False, + loopx_input_json: str | None = None, + input_goal_id: str | None = None, ) -> dict: data = _source_path(source_root, artifact).read_bytes() events = [] @@ -142,6 +144,21 @@ def _usage_capture( payload = {k: payload[k] for k in ("id", "forked_from_id") if k in payload} elif event.get("type") == "token_usage_record": payload = _request_usage_record(payload) + elif event.get("type") == "loopx_input_token_attribution": + # Optional host measurement, not a tokenizer or prose classifier. + # Preserve only binding/counts; raw prompts and packet text stay private. + payload = { + key: payload[key] if isinstance(payload[key], str) else None + for key in ( + "method", "thread_id", "session_id", "turn_id", + "root_turn_id", "response_id", + ) if key in payload + } | { + key: payload[key] if type(payload[key]) is int else None + for key in ( + "input_tokens", "loopx_input_tokens", "loopx_cached_input_tokens", + ) if key in payload + } elif event.get("type") == "compacted": payload = { "compaction_response_id": payload.get("compaction_response_id") @@ -185,12 +202,25 @@ def _usage_capture( "payload": payload, } ) + observation = None + input_artifacts = [] + if loopx_input_json is not None: + from .trajectory_input_capture import load_codex_input_observation + try: + observation = load_codex_input_observation(source_root, loopx_input_json) + input_artifacts = observation.pop("artifacts") + except (ImportError, OSError, ValueError, KeyError, TypeError): + # A supplementary observation cannot erase independently valid + # native counters. The typed owner withholds only input attribution. + observation = {"error": "Input observation or tokenizer is unavailable"} projection = effect_runtime_result( "benchmark.trajectory.codex-usage", { "events": events, "session_id": session_id, "include_incomplete": include_incomplete, + **({"loopx_input_trace": observation, "input_goal_id": input_goal_id} + if loopx_input_json is not None else {}), }, large_local_snapshot=True, ) @@ -198,6 +228,7 @@ def _usage_capture( "artifact": artifact, "artifact_sha256": hashlib.sha256(data).hexdigest(), "projection": projection, + **({"input_artifacts": input_artifacts} if input_artifacts else {}), } @@ -212,6 +243,10 @@ def _captured_view(record: dict, field: str) -> dict | None: ).read_bytes() if hashlib.sha256(data).hexdigest() != captured["artifact_sha256"]: reason = "source_changed" + for artifact in captured.get("input_artifacts", []): + data = _source_path(Path(record["source_root"]), artifact["artifact"]).read_bytes() + if hashlib.sha256(data).hexdigest() != artifact["artifact_sha256"]: + reason = "source_changed" except (OSError, ValueError): reason = "source_unavailable" return { @@ -294,6 +329,7 @@ def register_pack( usage_jsonl: str | None = None, usage_session_id: str | None = None, include_incomplete_usage: bool = False, + loopx_input_json: str | None = None, native_outcomes_json: str | None = None, base_pack_id: str | None = None, ) -> dict: @@ -327,6 +363,8 @@ def register_pack( record["base_pack_id"] = _opaque(base_pack_id) if include_incomplete_usage and usage_jsonl is None: raise ValueError("include-incomplete-usage requires a usage capture") + if loopx_input_json is not None and usage_jsonl is None: + raise ValueError("LoopX input observation requires a usage capture") if (usage_jsonl is None) != (usage_session_id is None): raise ValueError( "Usage capture and expected session identity must be supplied together" @@ -335,6 +373,8 @@ def register_pack( record["solver_usage"] = _usage_capture( source_root, usage_jsonl, usage_session_id, include_incomplete=include_incomplete_usage, + loopx_input_json=loopx_input_json, + input_goal_id=record["input"]["scope"]["goal_id"], ) if native_outcomes_json is not None: data = _source_path(source_root, native_outcomes_json).read_bytes() diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_input_capture.py b/loopx/capabilities/benchmark_toolkit/trajectory_input_capture.py new file mode 100644 index 0000000000..28fe13a15f --- /dev/null +++ b/loopx/capabilities/benchmark_toolkit/trajectory_input_capture.py @@ -0,0 +1,126 @@ +"""Private native-input IO and optional BPE computation; TS owns attribution. + +The host records its own injected packet only when native rollout tracing was +already enabled for that invocation. It never enables tracing globally, edits a +Codex rollout, changes execution, or claims provider token attribution. +""" +from __future__ import annotations + +import hashlib +import json +from importlib.metadata import version +from pathlib import Path +from typing import Any + + +def _read(root: Path, artifact: str) -> tuple[dict[str, Any], bytes]: + path = root / artifact + if Path(artifact).is_absolute() or ".." in Path(artifact).parts or "\\" in artifact: + raise ValueError("Input observation reference escapes its source") + path.resolve(strict=True).relative_to(root.resolve(strict=True)) + raw = path.read_bytes() + value = json.loads(raw) + if not isinstance(value, dict): + raise ValueError("Input observation must contain an object") + return value, raw + + +def load_codex_input_observation(source_root: Path, artifact: str) -> dict[str, Any]: + """Transport native ids/digests and tokenizer observations, never raw text. + + o200k_base is an explicitly named estimator. Its use does not establish the + tokenizer of an unknown/new provider model. Encoding assets may be fetched + by tiktoken during operator registration; ordinary pack reads never tokenize. + """ + import tiktoken # Optional specialized computation provider, not a state owner. + + source, source_bytes = _read(source_root, artifact) + artifacts = [{"artifact": artifact, "artifact_sha256": hashlib.sha256(source_bytes).hexdigest()}] + if source.get("schema_version") != "loopx_codex_input_source_v1": + raise ValueError("Unsupported native input source") + text = source.get("text") + if not isinstance(text, str): + raise ValueError("Native input source text is unavailable") + bundle = source.get("trace_bundle") + if not isinstance(bundle, str) or Path(bundle).name != bundle: + raise ValueError("Invalid native input bundle") + manifest, manifest_bytes = _read(source_root, f"{bundle}/manifest.json") + artifacts.append({"artifact": f"{bundle}/manifest.json", "artifact_sha256": hashlib.sha256(manifest_bytes).hexdigest()}) + events_path = source_root / bundle / "trace.jsonl" + events_path.resolve(strict=True).relative_to(source_root.resolve(strict=True)) + raw_events = events_path.read_bytes() + artifacts.append({"artifact": f"{bundle}/trace.jsonl", "artifact_sha256": hashlib.sha256(raw_events).hexdigest()}) + encoding = tiktoken.get_encoding("o200k_base") + source_digest = hashlib.sha256(text.encode()).hexdigest() + source_tokens = len(encoding.encode_ordinary(text)) + + def part(value: dict[str, Any]) -> dict[str, Any]: + if not isinstance(value, dict): + raise ValueError("Invalid native input content") + content = value.get("text") + if value.get("type") != "input_text" or not isinstance(content, str): + return {"type": value.get("type")} + digest = hashlib.sha256(content.encode()).hexdigest() + return {"type": "input_text", "sha256": digest, + **({"tokens": source_tokens} if digest == source_digest else {})} + + def payload(ref: dict[str, Any]) -> dict[str, Any]: + if not isinstance(ref, dict): + raise ValueError("Invalid native input payload reference") + relative = ref.get("path") + if (not isinstance(relative, str) or len(Path(relative).parts) != 2 + or Path(relative).parts[0] != "payloads"): + raise ValueError("Invalid native input payload reference") + data, raw = _read(source_root, f"{bundle}/{relative}") + artifacts.append({"artifact": f"{bundle}/{relative}", "artifact_sha256": hashlib.sha256(raw).hexdigest()}) + return {"body": data, "sha256": hashlib.sha256(raw).hexdigest()} + + events = [] + for line in raw_events.decode().splitlines(): + row = json.loads(line) + if not isinstance(row, dict) or not isinstance(row.get("payload"), dict): + raise ValueError("Invalid native input event") + event = row["payload"] + observed = {key: event.get(key) for key in ( + "type", "inference_call_id", "thread_id", "codex_turn_id", + "model", "response_id", "upstream_request_id", + )} + if event.get("type") == "inference_started": + read = payload(event["request_payload"]) + body = read["body"] + items = body.get("input") + if not isinstance(items, list) or any(not isinstance(item, dict) for item in items): + raise ValueError("Invalid native request input") + for item in items: + if item.get("content") is not None and not isinstance(item["content"], list): + raise ValueError("Invalid native request content") + observed["request"] = { + "sha256": read["sha256"], "model": body.get("model"), + "previous_response_id": body.get("previous_response_id"), + "input": [{"id": item.get("id"), "type": item.get("type"), + "role": item.get("role"), + "content": [part(p) for p in (item.get("content") or [])] + } for item in items], + } + elif event.get("type") == "inference_completed": + read = payload(event["response_payload"]) + body = read["body"] + observed["response"] = {"sha256": read["sha256"], **{ + key: body.get(key) for key in ("response_id", "upstream_request_id", "token_usage") + }} + events.append({key: row.get(key) for key in ( + "schema_version", "seq", "rollout_id", "thread_id", "codex_turn_id", + )} | {"payload": observed}) + return { + "source": {key: source.get(key) for key in ( + "schema_version", "session_id", "goal_id", "source_sha256", + )} | {"text_sha256": source_digest}, + "source_artifact_sha256": hashlib.sha256(source_bytes).hexdigest(), + "trace_sha256": hashlib.sha256(raw_events).hexdigest(), + "manifest": {key: manifest.get(key) for key in ( + "schema_version", "root_thread_id", "rollout_id", + )}, + "tokenizer": {"name": "tiktoken", "version": version("tiktoken"), "encoding": "o200k_base"}, + "events": events, + "artifacts": artifacts, + } diff --git a/loopx/cli_commands/benchmark_trajectory.py b/loopx/cli_commands/benchmark_trajectory.py index 1306573bbd..5bc655724e 100644 --- a/loopx/cli_commands/benchmark_trajectory.py +++ b/loopx/cli_commands/benchmark_trajectory.py @@ -42,6 +42,10 @@ def register_benchmark_trajectory_commands(subparsers, add_format) -> None: "--usage-session-id", help="Expected captured solver session identity; required with --usage-jsonl.", ) + register.add_argument( + "--loopx-input-json", + help="Optional host-recorded LoopX input source beside its private native trace, relative to source-root; estimates recorded Turn packet text with the optional input-token-estimates provider. Requires usage capture; never starts a model.", + ) register.add_argument( "--include-incomplete-usage", action="store_true", help="Observe persisted counters through an unfinished final Turn; requires usage capture. Does not establish complete cost or solver termination.", @@ -97,6 +101,7 @@ def _handle_benchmark_trajectory_command( usage_jsonl=args.usage_jsonl, usage_session_id=args.usage_session_id, include_incomplete_usage=args.include_incomplete_usage, + loopx_input_json=args.loopx_input_json, native_outcomes_json=args.native_outcomes_json, base_pack_id=args.base_pack_id, ) diff --git a/loopx/control_plane/capabilities/trajectory_assessment_types.ts b/loopx/control_plane/capabilities/trajectory_assessment_types.ts index 1c1ce0e7a0..d62579c217 100644 --- a/loopx/control_plane/capabilities/trajectory_assessment_types.ts +++ b/loopx/control_plane/capabilities/trajectory_assessment_types.ts @@ -74,8 +74,23 @@ export interface TrajectorySolverUsage { limitations: string[]; /** Supplementary reconciliation, never a replacement for the session-counter basis. */ request_coverage?: TrajectoryRequestCoverage; + /** Optional measured attribution or a labeled recorded-packet estimate; not a counter authority. */ + loopx_input?: TrajectoryLoopXInput; } +export type TrajectoryLoopXInput = { + method: 'provider_input_token_attribution_v1' | 'tiktoken_o200k_base_input_estimate_v1'; + estimate_basis?: {tokenizer: string; version: string; encoding: 'o200k_base'; + scope: 'recorded_turn_packets'; models: string[]}; +} & ({status: 'complete' | 'partial'; attributed_requests: number; total_requests: number; + attributed_input_tokens: number; total_input_tokens: number; loopx_input_tokens: number; + loopx_cached_input_tokens: number | null; + /** Whole-capture shares remain unknown until every request has attribution. */ + input_fraction: number | null; uncached_input_fraction: number | null; + /** Possible allocation of missing requests, not a statistical confidence interval. */ + input_fraction_bounds: [number, number] | null; +} | {status: 'unavailable'; reason: string}); + export interface TrajectoryRequestTokens { input_tokens: number; cached_input_tokens: number; cache_write_input_tokens: number; output_tokens: number; reasoning_output_tokens: number; total_tokens: number; diff --git a/loopx/control_plane/capabilities/trajectory_codex_usage.ts b/loopx/control_plane/capabilities/trajectory_codex_usage.ts index 1946534dd4..abe8b2959f 100644 --- a/loopx/control_plane/capabilities/trajectory_codex_usage.ts +++ b/loopx/control_plane/capabilities/trajectory_codex_usage.ts @@ -1,7 +1,8 @@ /** 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'; -import {reconcileTrajectoryRequests} from './trajectory_request_usage.ts'; +import {reconcileTrajectoryRequests, summarizeTrajectoryLoopXInput} from './trajectory_request_usage.ts'; +import {estimateTrajectoryLoopXInput} from './trajectory_input_usage.ts'; const fail = (message: string): never => { throw new EffectRuntimeRequestError(`trajectory usage: ${message}`); }; const counterKeys = ['input_tokens', 'cached_input_tokens', 'output_tokens'] as const; @@ -76,8 +77,11 @@ export function summarizeTrajectoryCodexUsage(params: Record): usage: active.samples && tail ? {...tail, uncached_input_tokens: tail.input_tokens - tail.cached_input_tokens} : null} : null; const start = turns[0]?.start ?? active!.start, end = active ? observedThrough : turns.at(-1)!.end; const requestCoverage = reconcileTrajectoryRequests(rows, params.session_id); + const loopxInput = summarizeTrajectoryLoopXInput(rows, params.session_id, requestCoverage) + ?? estimateTrajectoryLoopXInput(params.loopx_input_trace, params.session_id, params.input_goal_id, rows, requestCoverage); return {session_id: params.session_id, method: observed ? 'codex_observed_session_counters_v1' : 'codex_complete_session_counters_v1', ...(requestCoverage ? {request_coverage: requestCoverage} : {}), + ...(loopxInput ? {loopx_input: loopxInput} : {}), ...(observed ? {incomplete_turn: incomplete} : {}), start, end, completed_turns: turns.length, counter_observations: samples, input_tokens: previous.input_tokens, cached_input_tokens: previous.cached_input_tokens, diff --git a/loopx/control_plane/capabilities/trajectory_input_usage.ts b/loopx/control_plane/capabilities/trajectory_input_usage.ts new file mode 100644 index 0000000000..90d992fb27 --- /dev/null +++ b/loopx/control_plane/capabilities/trajectory_input_usage.ts @@ -0,0 +1,117 @@ +/** Bind optional native input observations to the existing request ledger. + * This estimates only explicitly recorded LoopX Turn packets. It never labels + * arbitrary prose, a quoted packet, tool output or all compaction as LoopX. */ +import type {TrajectoryLoopXInput, TrajectoryRequestCoverage} from './trajectory_assessment_types.ts'; + +const method = 'tiktoken_o200k_base_input_estimate_v1'; +class InputObservationError extends Error {} +const fail = (message: string): never => { throw new InputObservationError(message); }; +function object(value: unknown): Record { + if (!value || typeof value !== 'object' || Array.isArray(value)) return fail('Missing native input observation.'); + return value as Record; +} +function array(value: unknown): unknown[] { + return Array.isArray(value) ? value : fail('Missing native input records.'); +} +function text(value: unknown): string { + return typeof value === 'string' && value.length > 0 ? value : fail('Missing native input identity.'); +} +function count(value: unknown): number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 + ? value : fail('Invalid input token estimate.'); +} +function digest(value: unknown): string { + return typeof value === 'string' && /^[a-f0-9]{64}$/.test(value) ? value : fail('Missing input source digest.'); +} + +export function estimateTrajectoryLoopXInput(value: unknown, session: string, goal: unknown, + rows: Array<{type?: unknown; payload?: unknown}>, coverage: TrajectoryRequestCoverage | undefined): TrajectoryLoopXInput | undefined { + if (value === undefined) return undefined; + try { + if (coverage?.status !== 'verified') return fail('Request coverage must reconcile before estimating input.'); + const input = object(value), source = object(input.source), manifest = object(input.manifest); + if (source.schema_version !== 'loopx_codex_input_source_v1' || source.session_id !== session + || source.goal_id !== text(goal) || manifest.schema_version !== 1 + || manifest.root_thread_id !== session || manifest.rollout_id !== session) return fail('Input source Goal or native session mismatch.'); + const sourceDigest = digest(source.source_sha256); + if (sourceDigest !== digest(source.text_sha256)) return fail('Recorded LoopX source changed.'); + digest(input.source_artifact_sha256); digest(input.trace_sha256); + const tokenizer = object(input.tokenizer); + if (tokenizer.name !== 'tiktoken' || tokenizer.encoding !== 'o200k_base') return fail('Unsupported input estimator.'); + const version = text(tokenizer.version); + const native = new Map(rows.filter(row => row.type === 'token_usage_record').map(row => { + const p = object(row.payload); return [text(p.response_id), p] as const; + })); + if (!native.size || native.size !== coverage.unique_requests) return fail('Native request coverage changed.'); + const started = new Map}>(); + const completed = new Map(); + const models = new Set(); + let sequence = 0, sourceItem: string | null = null, sourcePart = -1; + for (const raw of array(input.events)) { + const row = object(raw), event = object(row.payload); + if (row.schema_version !== 1 || row.rollout_id !== session || row.seq !== ++sequence) return fail('Native trace sequence or session mismatch.'); + if (!['inference_started', 'inference_completed'].includes(event.type as string)) continue; + if (row.thread_id !== session) return fail('Native trace request belongs to another session.'); + const id = text(event.inference_call_id), turn = text(row.codex_turn_id); + if (event.type === 'inference_started') { + if (started.has(id)) return fail('Duplicate native inference identity.'); + const request = object(event.request); + digest(request.sha256); + if (request.model !== event.model || event.thread_id !== session || event.codex_turn_id !== turn) return fail('Native request identity or model mismatch.'); + started.set(id, {turn, request}); + } else { + const start = started.get(id); + if (!start || start.turn !== turn) return fail('Native response has no matching request.'); + const response = object(event.response), responseId = text(event.response_id); + digest(response.sha256); + if (responseId !== response.response_id || event.upstream_request_id !== response.upstream_request_id) return fail('Native response identity mismatch.'); + const original = native.get(responseId); + if (!original || original.turn_id !== turn) return fail('Native input response is outside the usage capture.'); + const usage = object(response.token_usage), expected = object(original.usage); + for (const key of ['input_tokens', 'cached_input_tokens', 'cache_write_input_tokens', + 'output_tokens', 'reasoning_output_tokens', 'total_tokens']) { + if (count(usage[key]) !== expected[key]) return fail('Native response counters disagree with the usage capture.'); + } + // A compressed websocket request is not a full model-visible input. + // It stays uncovered until its previous-response lineage is replayed. + if (start.request.previous_response_id != null) continue; + const items = array(start.request.input).map(object); + if (sourceItem === null) { + const matches: Array<{id: string; part: number}> = []; + for (const item of items) { + if (item.type !== 'message' || item.role !== 'user') continue; + array(item.content).map(object).forEach((part, index) => { + if (part.type === 'input_text' && part.sha256 === sourceDigest) matches.push({id: text(item.id), part: index}); + }); + } + if (matches.length > 1) return fail('Recorded source is ambiguous in the native request.'); + if (!matches.length) continue; + sourceItem = matches[0]!.id; sourcePart = matches[0]!.part; + } + const matches = items.filter(item => item.id === sourceItem); + if (matches.length !== 1 || matches[0]!.type !== 'message' || matches[0]!.role !== 'user') continue; + const part = object(array(matches[0]!.content)[sourcePart]); + if (part.type !== 'input_text' || part.sha256 !== sourceDigest) continue; + const loopx = count(part.tokens), total = count(usage.input_tokens); + if (loopx > total) return fail('Input estimator exceeds the native request total; no clamping is permitted.'); + const previous = completed.get(responseId); + if (previous && (previous.input !== total || previous.loopx !== loopx)) return fail('Conflicting input observations for one response.'); + completed.set(responseId, {input: total, loopx}); models.add(text(start.request.model)); + } + } + let attributed = 0, loopx = 0; + for (const request of completed.values()) { + attributed = count(attributed + request.input); loopx = count(loopx + request.loopx); + } + const complete = completed.size === native.size, total = coverage.all_requests.input_tokens; + return {method, status: complete ? 'complete' : 'partial', + estimate_basis: {tokenizer: 'tiktoken', version, encoding: 'o200k_base', scope: 'recorded_turn_packets', models: [...models].sort()}, + attributed_requests: completed.size, total_requests: native.size, + attributed_input_tokens: attributed, total_input_tokens: total, loopx_input_tokens: loopx, + loopx_cached_input_tokens: null, input_fraction: complete && total > 0 ? loopx / total : null, + uncached_input_fraction: null, input_fraction_bounds: null}; + } catch (error) { + if (!(error instanceof InputObservationError)) throw error; + return {method, status: 'unavailable', reason: error.message}; + } +} diff --git a/loopx/control_plane/capabilities/trajectory_request_usage.ts b/loopx/control_plane/capabilities/trajectory_request_usage.ts index 982e49f58e..60b730ff8d 100644 --- a/loopx/control_plane/capabilities/trajectory_request_usage.ts +++ b/loopx/control_plane/capabilities/trajectory_request_usage.ts @@ -1,5 +1,5 @@ /** Reconcile persisted request counters without changing session-counter accounting. */ -import type {TrajectoryRequestCoverage, TrajectoryRequestTokens} from './trajectory_assessment_types.ts'; +import type {TrajectoryLoopXInput, TrajectoryRequestCoverage, TrajectoryRequestTokens} from './trajectory_assessment_types.ts'; const keys = ['input_tokens', 'cached_input_tokens', 'cache_write_input_tokens', 'output_tokens', 'reasoning_output_tokens', 'total_tokens'] as const; @@ -103,3 +103,53 @@ export function reconcileTrajectoryRequests(rows: Array<{type?: unknown; payload return {method, status: 'unavailable', reason: error.message}; } } + +/** A producer's counted attribution is a separate observation. It cannot repair + * native counters or turn unknown historical packet contents into token counts. */ +export function summarizeTrajectoryLoopXInput(rows: Array<{type?: unknown; payload?: unknown}>, + session: string, coverage: TrajectoryRequestCoverage | undefined): TrajectoryLoopXInput | undefined { + const annotations = rows.filter(e => e.type === 'loopx_input_token_attribution'); + if (!annotations.length) return undefined; + const method = 'provider_input_token_attribution_v1'; + try { + if (coverage?.status !== 'verified') return invalid('Request coverage must reconcile before input attribution.'); + const requests = new Map(rows.filter(e => e.type === 'token_usage_record').map(e => { + const r = record(e.payload, session); return [r.response_id, r] as const; + })); + const counted = new Map(); + const count = (v: unknown): number => typeof v === 'number' && Number.isSafeInteger(v) && v >= 0 + ? v : invalid('Missing or invalid attributed token count.'); + for (const annotation of annotations) { + const p = object(annotation.payload), request = requests.get(p.response_id as string); + if (!request || p.method !== method || p.thread_id !== session || p.session_id !== session + || p.turn_id !== request.turn_id || p.root_turn_id !== request.root_turn_id) return invalid('Input attribution method or request identity mismatch.'); + const input = count(p.input_tokens), loopx = count(p.loopx_input_tokens); + const cached = !Object.hasOwn(p, 'loopx_cached_input_tokens') + ? (request.usage.cached_input_tokens === 0 ? 0 : null) : count(p.loopx_cached_input_tokens); + if (input !== request.usage.input_tokens || loopx > input || (cached !== null + && (cached > loopx || cached > request.usage.cached_input_tokens + || loopx - cached > input - request.usage.cached_input_tokens))) return invalid('Attributed tokens do not fit their request counters.'); + const old = counted.get(request.response_id); + if (old && (old.input !== input || old.loopx !== loopx || old.cached !== cached)) return invalid('Conflicting input attribution for one response.'); + counted.set(request.response_id, {input, loopx, cached}); + } + let input = 0, loopx = 0, cached = 0, cacheKnown = true; + for (const row of counted.values()) { + input = count(input + row.input); loopx = count(loopx + row.loopx); + if (row.cached === null) cacheKnown = false; + else cached = count(cached + row.cached); + } + const total = coverage.all_requests.input_tokens; + const uncached = total - coverage.all_requests.cached_input_tokens; + const complete = counted.size === requests.size; + return {method, status: complete ? 'complete' : 'partial', attributed_requests: counted.size, + total_requests: requests.size, attributed_input_tokens: input, total_input_tokens: total, + loopx_input_tokens: loopx, loopx_cached_input_tokens: cacheKnown ? cached : null, + input_fraction: complete && total > 0 ? loopx / total : null, + uncached_input_fraction: complete && cacheKnown && uncached > 0 ? (loopx - cached) / uncached : null, + input_fraction_bounds: total > 0 ? [loopx / total, (loopx + (total - input)) / total] : null}; + } catch (error) { + if (!(error instanceof ReconciliationError)) throw error; + return {method, status: 'unavailable', reason: error.message}; + } +} diff --git a/loopx/control_plane/turn_driver/codex_cli.py b/loopx/control_plane/turn_driver/codex_cli.py index 9dacc88ad6..45fde68e2c 100644 --- a/loopx/control_plane/turn_driver/codex_cli.py +++ b/loopx/control_plane/turn_driver/codex_cli.py @@ -16,6 +16,7 @@ child_execution_receipts_json_schema, ) from ...extensions.codex_native_child import native_child_observer +from .codex_input_trace import capture_codex_input_source from .codex_sessions import ( CODEX_CLI_SESSION_SCHEMA_VERSION as CODEX_CLI_SESSION_SCHEMA_VERSION, _discard_codex_cli_session, @@ -873,12 +874,14 @@ def observe_stderr(line: str) -> None: events = HostOutputLines(observe_event) diagnostics = HostOutputLines(observe_stderr) - observed = run_host_process(command, project=project, input_text=_prompt(request), - timeout_seconds=timeout_seconds, drain_timeout_seconds=OUTPUT_DRAIN_TIMEOUT_SECONDS, - on_stdout=events.feed, on_stderr=diagnostics.feed, - launch_guard=lambda: launch_scope("codex_cli_host_launch")) - events.finish() - diagnostics.finish() + input_text = _prompt(request) + with capture_codex_input_source(project, input_text, lineage["goal_id"], observed_session): + observed = run_host_process(command, project=project, input_text=input_text, + timeout_seconds=timeout_seconds, drain_timeout_seconds=OUTPUT_DRAIN_TIMEOUT_SECONDS, + on_stdout=events.feed, on_stderr=diagnostics.feed, + launch_guard=lambda: launch_scope("codex_cli_host_launch")) + events.finish() + diagnostics.finish() returncode = observed["returncode"] timed_out = observed["outcome"] == "timeout" output_observation_incomplete = not (observed["output_complete"] and events.complete and diagnostics.complete) diff --git a/loopx/control_plane/turn_driver/codex_input_trace.py b/loopx/control_plane/turn_driver/codex_input_trace.py new file mode 100644 index 0000000000..2f28de6682 --- /dev/null +++ b/loopx/control_plane/turn_driver/codex_input_trace.py @@ -0,0 +1,73 @@ +"""Opt-in native Codex diagnostic IO; no token attribution or authority.""" +from __future__ import annotations + +import hashlib +import json +import os +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path + + +def record_codex_input_source( + trace_root: Path, *, session_id: str, goal_id: str, text: str, + prior_bundles: set[str], +) -> Path | None: + """Record the actual host injection beside its new native diagnostic bundle. + + Called after native session readback, before decoding the task result. A + recording failure cannot change the original Host result or its authority. + Files contain sensitive input and remain inside the opt-in private trace root. + """ + matches = [] + for bundle in trace_root.glob("trace-*"): + if bundle.name in prior_bundles or not bundle.is_dir(): + continue + manifest_path = bundle / "manifest.json" + manifest_path.resolve(strict=True).relative_to(trace_root.resolve(strict=True)) + manifest = json.loads(manifest_path.read_bytes()) + if not isinstance(manifest, dict): + raise ValueError("Invalid native trace manifest") + if (manifest.get("schema_version") == 1 + and manifest.get("root_thread_id") == session_id + and manifest.get("rollout_id") == session_id): + matches.append(bundle.name) + if len(matches) != 1: + return None + digest = hashlib.sha256(text.encode()).hexdigest() + value = {"schema_version": "loopx_codex_input_source_v1", "session_id": session_id, + "goal_id": goal_id, "trace_bundle": matches[0], + "source_sha256": digest, "text": text} + directory = trace_root / "loopx-input-sources" + directory.mkdir(mode=0o700, exist_ok=True) + directory.resolve(strict=True).relative_to(trace_root.resolve(strict=True)) + identity = hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() + path = directory / f"{identity}.json" + data = json.dumps(value, ensure_ascii=False, sort_keys=True).encode() + try: + with path.open("xb") as output: + path.chmod(0o600) + output.write(data) + except FileExistsError: + if path.read_bytes() != data: + raise ValueError("Conflicting native input source") + return path + + +@contextmanager +def capture_codex_input_source(project: Path, text: str, goal_id: str, + session_ids: list[str]) -> Iterator[None]: + trace_value = os.environ.get("CODEX_ROLLOUT_TRACE_ROOT") + try: + root = (project / Path(trace_value).expanduser()).resolve() if trace_value else None + prior = {path.name for path in root.glob("trace-*")} if root else set() + except (OSError, ValueError, RuntimeError): + root, prior = None, set() + yield + if root is not None and len(session_ids) == 1: + try: + record_codex_input_source(root, session_id=session_ids[0], goal_id=goal_id, + text=text, prior_bundles=prior) + except (OSError, ValueError, KeyError, TypeError, RuntimeError): + # Optional diagnostics must not alter native execution or settlement. + pass diff --git a/pyproject.toml b/pyproject.toml index a45fbac94f..13e2e1e421 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,9 @@ Issues = "https://github.com/loopx-project/loopx/issues" Changelog = "https://github.com/loopx-project/loopx/releases" [project.optional-dependencies] +input-token-estimates = [ + "tiktoken==0.12.0", +] ego-source-reader = [ "mcp==1.30.0", ] @@ -43,6 +46,7 @@ test-core = [ "pytest>=9.0.3,<10", ] test = [ + "tiktoken==0.12.0", "distlib>=0.3.9,<1; sys_platform == 'win32'", "docker==7.2.0", "resvg-py==0.5.0", diff --git a/tests/capabilities/test_trajectory_assessment_store.py b/tests/capabilities/test_trajectory_assessment_store.py index ae81a08b79..6b91fbbb61 100644 --- a/tests/capabilities/test_trajectory_assessment_store.py +++ b/tests/capabilities/test_trajectory_assessment_store.py @@ -1116,3 +1116,65 @@ def test_observed_usage_real_cli_without_model_dispatch(pack): assert view['solver_usage']['projection']['incomplete_turn']['usage']['input_tokens'] == 40 assert inspect_pack(**scope, pack_id=record['pack_id'])['job'] is None assert not list(scope['runtime_root'].rglob('receipt.json')) + + +@pytest.mark.parametrize("coverage", ["complete", "partial", "conflicting", "invalid_cache"]) +def test_loopx_input_attribution_real_transport_and_privacy(pack, monkeypatch, coverage): + import loopx.capabilities.benchmark_toolkit.trajectory_assessment_store as store + + scope, _, old, input_path, sources, _ = pack + rows = [json.loads(line) for line in _request_usage_fixture().splitlines()] + annotations = [] + for response, turn, total, loopx, cached in [ + ("one", "a", 100, 20, 12), ("compact", "a", 20, 5, 4), + ("two", "b", 40, 15, 10), + ]: + annotations.append(dict(type="loopx_input_token_attribution", payload=dict( + method="provider_input_token_attribution_v1", response_id=response, + session_id="solver-session", thread_id="solver-session", turn_id=turn, + root_turn_id=turn, input_tokens=total, loopx_input_tokens=loopx, + loopx_cached_input_tokens=cached, + raw_packet="SYNTHETIC-PRIVATE-PACKET-DO-NOT-TRANSPORT"))) + if coverage == "partial": + annotations = annotations[:1] + if coverage == "conflicting": + conflict = json.loads(json.dumps(annotations[0])) + conflict["payload"]["loopx_input_tokens"] = 21 + annotations.append(conflict) + if coverage == "invalid_cache": + annotations[0]["payload"]["loopx_cached_input_tokens"] = True + path = sources / "usage.jsonl" + path.write_text("\n".join(json.dumps(row) for row in [*rows, *annotations]) + "\n") + original = path.read_bytes() + actual = store.effect_runtime_result + transported = [] + + def observe(command, params, **kwargs): + if command == "benchmark.trajectory.codex-usage": + transported.append(params) + return actual(command, params, **kwargs) + + monkeypatch.setattr(store, "effect_runtime_result", observe) + registered = register_pack(**scope, input_path=input_path, source_root=sources, + title="Synthetic attributed input", usage_jsonl="usage.jsonl", usage_session_id="solver-session") + assert registered["analysis_key"] == old["analysis_key"] + view = inspect_pack(**scope, pack_id=registered["pack_id"]) + usage = view["solver_usage"]["projection"] + assert usage["input_tokens"] == 140 + assert usage["request_coverage"]["status"] == "verified" + attribution = usage["loopx_input"] + assert attribution["status"] == ("unavailable" if coverage in {"conflicting", "invalid_cache"} else coverage) + if coverage == "complete": + assert attribution["loopx_input_tokens"] == 40 + assert attribution["total_input_tokens"] == 160 + assert attribution["input_fraction"] == 0.25 + assert attribution["uncached_input_fraction"] == 0.2 + elif coverage == "partial": + assert attribution["input_fraction"] is None + assert attribution["input_fraction_bounds"] == [0.125, 0.5] + else: + assert "input_fraction" not in attribution + assert transported and "SYNTHETIC-PRIVATE-PACKET" not in json.dumps(transported) + assert path.read_bytes() == original + assert view["job"] is None and view["result"] is None + assert "loopx_input" not in inspect_pack(**scope, pack_id=old["pack_id"]).get("solver_usage", {}).get("projection", {}) diff --git a/tests/capabilities/test_trajectory_input_capture.py b/tests/capabilities/test_trajectory_input_capture.py new file mode 100644 index 0000000000..b6d17e894d --- /dev/null +++ b/tests/capabilities/test_trajectory_input_capture.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import hashlib +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from loopx.capabilities.benchmark_toolkit.trajectory_input_capture import ( + load_codex_input_observation, +) +from loopx.control_plane.turn_driver.codex_input_trace import record_codex_input_source +from loopx.capabilities.benchmark_toolkit.trajectory_assessment_store import ( + _captured_view, _usage_capture, +) + + +def native_trace(root: Path) -> tuple[Path, list[dict]]: + bundle = root / "trace-fixture-session" + (bundle / "payloads").mkdir(parents=True) + (bundle / "manifest.json").write_text(json.dumps({"schema_version": 1, + "root_thread_id": "session", "rollout_id": "session"})) + usage = {"input_tokens": 100, "cached_input_tokens": 60, "cache_write_input_tokens": 0, + "output_tokens": 5, "reasoning_output_tokens": 0, "total_tokens": 105} + request = {"model": "native-fixture", "input": [{"id": "packet-item", "type": "message", + "role": "user", "content": [{"type": "input_text", "text": "hello world"}]}]} + response = {"response_id": "response", "upstream_request_id": "request", "token_usage": usage} + for name, data in (("1.json", request), ("2.json", response)): + (bundle / "payloads" / name).write_text(json.dumps(data)) + payloads = [ + {"type": "inference_started", "inference_call_id": "inference", "thread_id": "session", + "codex_turn_id": "turn", "model": "native-fixture", "request_payload": {"path": "payloads/1.json"}}, + {"type": "inference_completed", "inference_call_id": "inference", "response_id": "response", + "upstream_request_id": "request", "response_payload": {"path": "payloads/2.json"}}, + ] + events = [{"schema_version": 1, "seq": i + 1, "rollout_id": "session", "thread_id": "session", + "codex_turn_id": "turn", "payload": p} for i, p in enumerate(payloads)] + (bundle / "trace.jsonl").write_text("\n".join(map(json.dumps, events))) + path = record_codex_input_source(root, session_id="session", goal_id="goal", + text="hello world", prior_bundles=set()) + assert path is not None + rows = [{"type": "session_meta", "payload": {"id": "session"}}, + {"type": "event_msg", "payload": {"type": "task_started", "turn_id": "turn"}}, + {"type": "token_usage_record", "payload": {"session_id": "session", "thread_id": "session", + "turn_id": "turn", "root_turn_id": "turn", "response_id": "response", "usage": usage, + "thread_token_usage": usage, "turn_token_usage": usage}}, + {"type": "event_msg", "payload": {"type": "token_count", "info": {"total_token_usage": usage, "last_token_usage": usage}}}, + {"type": "event_msg", "payload": {"type": "task_complete", "turn_id": "turn"}}] + for i, row in enumerate(rows): + row["timestamp"] = f"2026-01-01T00:00:0{i}Z" + (root / "usage.jsonl").write_text("\n".join(map(json.dumps, rows))) + return path, rows + + +def test_native_input_source_is_idempotent_private_and_never_adopts_old_or_other_sessions(tmp_path): + path, _ = native_trace(tmp_path) + assert record_codex_input_source(tmp_path, session_id="session", goal_id="goal", + text="hello world", prior_bundles=set()) == path + assert record_codex_input_source(tmp_path, session_id="other", goal_id="goal", + text="hello world", prior_bundles=set()) is None + assert record_codex_input_source(tmp_path, session_id="session", goal_id="goal", + text="hello world", prior_bundles={"trace-fixture-session"}) is None + assert path.stat().st_mode & 0o077 == 0 + moved = tmp_path / "original-sources" + path.parent.rename(moved) + outside = tmp_path.parent / "outside-sources" + outside.mkdir() + path.parent.symlink_to(outside, target_is_directory=True) + with pytest.raises(ValueError): + record_codex_input_source(tmp_path, session_id="session", goal_id="goal", + text="hello world", prior_bundles=set()) + assert not list(outside.iterdir()) + + +def test_real_codec_io_typed_estimate_and_source_change_readback(tmp_path): + path, _ = native_trace(tmp_path) + original = {str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in tmp_path.rglob('*') if p.is_file()} + observed = load_codex_input_observation(tmp_path, str(path.relative_to(tmp_path))) + assert observed["tokenizer"]["encoding"] == "o200k_base" + assert observed["events"][0]["payload"]["request"]["input"][0]["content"][0]["tokens"] == 2 + assert "hello world" not in json.dumps(observed) + usage = _usage_capture(tmp_path, "usage.jsonl", "session", input_goal_id="goal", + loopx_input_json=str(path.relative_to(tmp_path))) + result = usage["projection"]["loopx_input"] + assert result["status"] == "complete" and result["input_fraction"] == 0.02 + assert result["method"] == "tiktoken_o200k_base_input_estimate_v1" + assert result["loopx_cached_input_tokens"] is None + assert all(hashlib.sha256(Path(p).read_bytes()).hexdigest() == digest for p, digest in original.items()) + record = {"source_root": str(tmp_path), "solver_usage": usage} + assert _captured_view(record, "solver_usage")["applicability"] == "current" + (tmp_path / "trace-fixture-session/payloads/1.json").write_text("{}") + assert _captured_view(record, "solver_usage")["applicability"] == "historical" + + +def test_invalid_optional_observation_cannot_erase_native_counters_or_escape_root(tmp_path): + path, _ = native_trace(tmp_path) + value = json.loads(path.read_text()) + value["trace_bundle"] = "../elsewhere" + path.write_text(json.dumps(value)) + with pytest.raises(ValueError, match="bundle"): + load_codex_input_observation(tmp_path, str(path.relative_to(tmp_path))) + usage = _usage_capture(tmp_path, "usage.jsonl", "session", input_goal_id="goal", + loopx_input_json=str(path.relative_to(tmp_path))) + assert usage["projection"]["input_tokens"] == 100 + assert usage["projection"]["loopx_input"]["status"] == "unavailable" + + +def test_real_cli_registers_recorded_component_without_model_execution(tmp_path): + from loopx.capabilities.benchmark_toolkit.trajectory_assessment_store import inspect_pack + + source, _ = native_trace(tmp_path) + evidence = "Synthetic fixed negative result." + (tmp_path / "evidence.txt").write_text(evidence) + data = {"schema_version": "trajectory_assessment_input_v1", + "scope": {"goal_id": "goal", "case_id": "case", "run_id": "run", "revision": "fixed", "window": "fixed capture"}, + "objective": "Compare independently verified outcomes and cost", + "selection": {"rule": "complete synthetic population", "population_episodes": 1, "exhaustive": True}, + "sources": [{"id": "source", "artifact": "evidence.txt", "artifact_sha256": hashlib.sha256(evidence.encode()).hexdigest(), + "first_line": 1, "text": evidence}], + "episodes": [{"id": "episode", "question": "Did the fixed test improve the original objective?", + "source_ids": ["source"], "eligible_metrics": ["benefit_test"]}]} + path = tmp_path / "input.json" + path.write_text(json.dumps(data)) + registry = tmp_path / "registry.json" + registry.write_text('{"goals": []}') + runtime = tmp_path / "runtime" + result = subprocess.run([sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json", "benchmark", "trajectory-pack-register", + "--goal-id", "research", "--input-json", str(path), "--source-root", str(tmp_path), + "--title", "Synthetic recorded input", "--usage-jsonl", "usage.jsonl", "--usage-session-id", "session", + "--loopx-input-json", str(source.relative_to(tmp_path))], capture_output=True, text=True, check=True) + registered = json.loads(result.stdout) + view = inspect_pack(runtime_root=runtime, owner_goal_id="research", pack_id=registered["pack_id"]) + assert view["solver_usage"]["projection"]["loopx_input"]["input_fraction"] == 0.02 + assert view["job"] is None and view["result"] is None + assert "hello world" not in json.dumps(view) diff --git a/tests/control_plane_ts/trajectory_input_usage.test.ts b/tests/control_plane_ts/trajectory_input_usage.test.ts new file mode 100644 index 0000000000..390e790ea6 --- /dev/null +++ b/tests/control_plane_ts/trajectory_input_usage.test.ts @@ -0,0 +1,195 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import {summarizeTrajectoryCodexUsage as summarize} from '../../loopx/control_plane/capabilities/trajectory_codex_usage.ts'; + +function nativeEstimate() { + const digest = 'a'.repeat(64), other = 'b'.repeat(64); + const first = tokens(100, 60, 5), second = tokens(300, 200, 5), total = tokens(400, 260, 10); + const rows = [event('session_meta', {id: 'session'}), event('event_msg', {type: 'task_started', turn_id: 'turn'})]; + for (const [id, usage, cumulative] of [['first', first, first], ['second', second, total]] as const) { + rows.push(event('token_usage_record', {response_id: id, session_id: 'session', thread_id: 'session', + turn_id: 'turn', root_turn_id: 'turn', usage, thread_token_usage: cumulative, turn_token_usage: cumulative}), + event('event_msg', {type: 'token_count', info: {total_token_usage: cumulative, last_token_usage: usage}})); + } + rows.push(event('event_msg', {type: 'task_complete', turn_id: 'turn'})); + const trace: any = {source: {schema_version: 'loopx_codex_input_source_v1', session_id: 'session', goal_id: 'goal', + source_sha256: digest, text_sha256: digest}, source_artifact_sha256: digest, trace_sha256: other, + manifest: {schema_version: 1, root_thread_id: 'session', rollout_id: 'session'}, + tokenizer: {name: 'tiktoken', version: '0.12.0', encoding: 'o200k_base'}, events: []}; + for (const [id, usage] of [['first', first], ['second', second]] as const) { + trace.events.push({schema_version: 1, seq: trace.events.length + 1, rollout_id: 'session', thread_id: 'session', + codex_turn_id: 'turn', payload: {type: 'inference_started', inference_call_id: id, thread_id: 'session', + codex_turn_id: 'turn', model: 'native-model', request: {sha256: other, model: 'native-model', + previous_response_id: null, input: [{id: 'source-item', type: 'message', role: 'user', + content: [{type: 'input_text', sha256: digest, tokens: 20}]}]}}}, + {schema_version: 1, seq: trace.events.length + 2, rollout_id: 'session', thread_id: 'session', codex_turn_id: 'turn', + payload: {type: 'inference_completed', inference_call_id: id, response_id: id, upstream_request_id: id, + response: {sha256: other, response_id: id, upstream_request_id: id, token_usage: usage}}}); + } + return {trace, rows: rows as any[]}; +} +const estimate = (fixture = nativeEstimate()) => summarize({events: fixture.rows.map((r, i) => ({...r, line: i + 1, + timestamp: new Date(Date.UTC(2026, 0, 1, 0, 0, i)).toISOString()})), session_id: 'session', + input_goal_id: 'goal', loopx_input_trace: fixture.trace}); + +test('native packet estimate sums repeated consumption against same-request totals and stays labeled', () => { + const result = estimate(), value = result.loopx_input!; + assert.equal(value.method, 'tiktoken_o200k_base_input_estimate_v1'); + assert.equal(value.status, 'complete'); if (value.status === 'unavailable') return; + assert.equal(value.loopx_input_tokens, 40); assert.equal(value.input_fraction, 0.1); + assert.equal(value.total_input_tokens, 400); assert.equal(value.uncached_input_fraction, null); + assert.equal(value.loopx_cached_input_tokens, null); assert.equal(value.input_fraction_bounds, null); + assert.equal(value.estimate_basis?.scope, 'recorded_turn_packets'); + assert.deepEqual(value.estimate_basis?.models, ['native-model']); + assert.equal(result.input_tokens, 400); assert.equal(result.cached_input_tokens, 260); +}); + +test('source copies do not become extra LoopX input; changed ids, text and unreplayed deltas stay partial', () => { + const original = nativeEstimate(); + const copied = structuredClone(original.trace.events[2].payload.request.input[0]); copied.id = 'quoted-copy'; + original.trace.events[2].payload.request.input.push(copied); + assert.equal(estimate(original).loopx_input?.status, 'complete'); + const a = estimate(original).loopx_input!; if (a.status !== 'unavailable') assert.equal(a.loopx_input_tokens, 40); + for (const change of [ + (r: any) => {r.input[0].id = 'different-item';}, + (r: any) => {r.input[0].role = 'assistant';}, + (r: any) => {r.input[0].content[0].sha256 = 'c'.repeat(64);}, + (r: any) => {r.previous_response_id = 'unreplayed';}, + ]) { + const f = nativeEstimate(); change(f.trace.events[2].payload.request); + const value = estimate(f).loopx_input!; assert.equal(value.status, 'partial'); + if (value.status !== 'unavailable') {assert.equal(value.input_fraction, null); assert.equal(value.attributed_requests, 1);} + } +}); + +test('bad native identity, source, sequence, counter and estimator evidence withholds only attribution', () => { + const mutations = [ + (t: any) => {t.source.goal_id = 'foreign';}, (t: any) => {t.source.session_id = 'foreign';}, + (t: any) => {t.source.source_sha256 = 'f'.repeat(64);}, + (t: any) => {t.events[0].seq = 2;}, (t: any) => {t.events[0].thread_id = 'foreign';}, + (t: any) => {t.events[1].codex_turn_id = 'foreign';}, + (t: any) => {t.events[1].payload.response_id = 'foreign';}, + (t: any) => {t.events[1].payload.response.token_usage = tokens(101, 60, 5);}, + (t: any) => {t.events[0].payload.request.model = 'foreign';}, + (t: any) => {t.events[0].payload.request.input[0].content[0].tokens = true;}, + (t: any) => {t.events[0].payload.request.input[0].content[0].tokens = 101;}, + (t: any) => {t.events[0].payload.request.input.push(structuredClone(t.events[0].payload.request.input[0]));}, + (t: any) => {t.tokenizer.encoding = 'character-count';}, + ]; + for (const mutate of mutations) { + const f = nativeEstimate(); mutate(f.trace); const result = estimate(f); + assert.equal(result.loopx_input?.status, 'unavailable'); assert.equal(result.input_tokens, 400); + assert.equal(result.request_coverage?.status, 'verified'); + } +}); + +const tokens = (input: number, cached: number, output: number) => ({input_tokens: input, + cached_input_tokens: cached, cache_write_input_tokens: 0, output_tokens: output, + reasoning_output_tokens: 0, total_tokens: input + output}); +const event = (type: string, payload: object) => ({type, payload}); +function fixture() { + const request = (response_id: string, usage: object, total: object) => event('token_usage_record', { + response_id, session_id: 'session', thread_id: 'session', turn_id: 'turn', root_turn_id: 'turn', + usage, turn_token_usage: total, thread_token_usage: total}); + const counter = (total: object, last: object) => event('event_msg', {type: 'token_count', + info: {total_token_usage: total, last_token_usage: last}}); + const compact = request('compact', tokens(20, 10, 7), tokens(120, 70, 17)); + return [event('session_meta', {id: 'session'}), event('event_msg', {type: 'task_started', turn_id: 'turn'}), + request('one', tokens(100, 60, 10), tokens(100, 60, 10)), counter(tokens(100, 60, 10), tokens(100, 60, 10)), + compact, event('compacted', {compaction_response_id: 'compact', latest_token_usage_record: compact.payload}), + counter(tokens(120, 70, 17), tokens(20, 10, 7)), + request('two', tokens(40, 20, 5), tokens(160, 90, 22)), counter(tokens(160, 90, 22), tokens(40, 20, 5)), + event('event_msg', {type: 'task_complete', turn_id: 'turn'})] as any[]; +} +const attribution = (response_id: string, input_tokens: number, loopx_input_tokens: number, + loopx_cached_input_tokens?: number) => event('loopx_input_token_attribution', { + method: 'provider_input_token_attribution_v1', response_id, session_id: 'session', thread_id: 'session', + turn_id: 'turn', root_turn_id: 'turn', input_tokens, loopx_input_tokens, + ...(loopx_cached_input_tokens === undefined ? {} : {loopx_cached_input_tokens})}); +const run = (rows: any[], observed = false) => summarize({session_id: 'session', include_incomplete: observed, + events: rows.map((row, i) => ({...row, line: i + 1, timestamp: new Date(Date.UTC(2026, 0, 1, 0, 0, i)).toISOString()}))}); +const complete = () => [...fixture(), attribution('one', 100, 20, 12), + attribution('compact', 20, 5, 4), attribution('two', 40, 15, 10)]; + +test('packet share sums same-request tokens including cache, rather than averaging request percentages', () => { + const a = run(complete()).loopx_input!; + assert.equal(a.status, 'complete'); if (a.status === 'unavailable') return; + assert.equal(a.loopx_input_tokens, 40); assert.equal(a.total_input_tokens, 160); + assert.equal(a.input_fraction, 0.25); assert.equal(a.uncached_input_fraction, 0.2); + assert.equal(a.loopx_cached_input_tokens, 26); + assert.equal(a.attributed_requests, 3); assert.equal(a.total_requests, 3); + assert.deepEqual(a.input_fraction_bounds, [0.25, 0.25]); + assert.equal(run(complete()).input_tokens, 160); +}); + +test('partial attribution keeps the whole-capture percentage unknown with honest possible bounds', () => { + const a = run([...fixture(), attribution('one', 100, 20)]).loopx_input!; + assert.equal(a.status, 'partial'); if (a.status === 'unavailable') return; + assert.equal(a.input_fraction, null); assert.equal(a.uncached_input_fraction, null); + assert.equal(a.loopx_cached_input_tokens, null); + assert.equal(a.attributed_requests, 1); assert.equal(a.attributed_input_tokens, 100); + assert.deepEqual(a.input_fraction_bounds, [0.125, 0.5]); +}); + +test('missing capture is unknown, and an explicitly measured zero remains zero', () => { + assert.equal(run(fixture()).loopx_input, undefined); + const zero = run([...fixture(), attribution('one', 100, 0, 0), + attribution('compact', 20, 0, 0), attribution('two', 40, 0, 0)]).loopx_input!; + assert.equal(zero.status, 'complete'); if (zero.status === 'unavailable') return; + assert.equal(zero.input_fraction, 0); assert.equal(zero.loopx_input_tokens, 0); +}); + +test('response mirrors and repeated attribution deduplicate while reprocessed packet input counts in each request', () => { + const rows = complete(); rows.push(structuredClone(rows[2]), structuredClone(rows[10])); + assert.deepEqual(run(rows).loopx_input, run(complete()).loopx_input); +}); + +test('invalid attribution does not overwrite independently reconciled request or session counters', () => { + const cases: Array<(rows: any[]) => void> = [ + rows => {rows[10].payload.response_id = 'missing';}, + rows => {rows[10].payload.session_id = 'foreign';}, + rows => {rows[10].payload.turn_id = 'foreign';}, + rows => {rows[10].payload.root_turn_id = 'foreign';}, + rows => {rows[10].payload.method = 'characters_divided_by_four';}, + rows => {rows[10].payload.input_tokens = 101;}, + rows => {rows[10].payload.loopx_input_tokens = 101;}, + rows => {rows[10].payload.loopx_input_tokens = true;}, + rows => {rows[10].payload.loopx_input_tokens = -1;}, + rows => {rows[10].payload.loopx_input_tokens = 1.5;}, + rows => {rows[10].payload.loopx_cached_input_tokens = 21;}, + rows => {rows[10].payload.loopx_cached_input_tokens = null;}, + rows => {rows[10].payload.loopx_input_tokens = 90; rows[10].payload.loopx_cached_input_tokens = 0;}, + rows => {const other = structuredClone(rows[10]); other.payload.loopx_input_tokens++; rows.push(other);}, + ]; + for (const mutate of cases) { + const rows = complete(); mutate(rows); const a = run(rows); + assert.equal(a.loopx_input?.status, 'unavailable'); + assert.equal(a.request_coverage?.status, 'verified'); assert.equal(a.input_tokens, 160); + assert.equal('input_fraction' in a.loopx_input!, false); + } +}); + +test('request gaps withhold attribution, and an unfinished Turn remains explicitly unfinished', () => { + const rows = complete(); rows[2].payload.thread_token_usage = tokens(101, 60, 10); + assert.equal(run(rows).loopx_input?.status, 'unavailable'); + const open = complete(); open.splice(9, 1); + assert.throws(() => run(open), /completed Turn/); + const a = run(open, true); assert.equal(a.loopx_input?.status, 'complete'); + assert.equal(a.incomplete_turn?.turn_id, 'turn'); +}); + +test('a zero denominator is undefined, and unknown cache attribution cannot become uncached savings', () => { + const rows = fixture().slice(0, 4); + rows[2].payload.usage = tokens(0, 0, 0); + rows[2].payload.thread_token_usage = tokens(0, 0, 0); + rows[2].payload.turn_token_usage = tokens(0, 0, 0); + rows[3].payload.info = {total_token_usage: tokens(0, 0, 0), last_token_usage: tokens(0, 0, 0)}; + rows.push(event('event_msg', {type: 'task_complete', turn_id: 'turn'}), attribution('one', 0, 0)); + const zero = run(rows).loopx_input!; + assert.equal(zero.status, 'complete'); if (zero.status === 'unavailable') return; + assert.equal(zero.input_fraction, null); assert.equal(zero.input_fraction_bounds, null); + const withoutCache = complete(); delete withoutCache[10].payload.loopx_cached_input_tokens; + const a = run(withoutCache).loopx_input!; + assert.equal(a.status, 'complete'); if (a.status === 'unavailable') return; + assert.equal(a.input_fraction, 0.25); assert.equal(a.uncached_input_fraction, null); +}); diff --git a/tests/test_loopx_turn_codex_cli.py b/tests/test_loopx_turn_codex_cli.py index fc389c10cd..23a81648b8 100644 --- a/tests/test_loopx_turn_codex_cli.py +++ b/tests/test_loopx_turn_codex_cli.py @@ -1480,6 +1480,37 @@ def test_codex_cli_host_discards_missing_resume_session( assert codex_cli_session_binding(runtime_root, envelope) is None +def test_native_input_observation_is_opt_in_and_does_not_change_host_result(tmp_path, monkeypatch): + from loopx.control_plane.turn_driver import codex_input_trace as trajectory_input_capture + + executable, log_path = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log_path)) + monkeypatch.delenv("CODEX_ROLLOUT_TRACE_ROOT", raising=False) + project = tmp_path / "project" + project.mkdir() + request = _request() + observed = [] + + def record(root, **fields): + observed.append((root, fields)) + raise OSError("Synthetic diagnostic failure must not fail the Host") + + monkeypatch.setattr(trajectory_input_capture, "record_codex_input_source", record) + baseline = run_codex_cli_host(request, runtime_root=tmp_path / "off", project=project, + codex_bin=str(executable), timeout_seconds=5) + assert observed == [] + trace = tmp_path / "traces" + (trace / "trace-old").mkdir(parents=True) + monkeypatch.setenv("CODEX_ROLLOUT_TRACE_ROOT", str(trace)) + result = run_codex_cli_host(request, runtime_root=tmp_path / "on", project=project, + codex_bin=str(executable), timeout_seconds=5) + assert result == baseline + assert len(observed) == 1 and observed[0][0] == trace + fields = observed[0][1] + assert fields["text"] == _prompt(request) and fields["goal_id"] == "fixture-goal" + assert fields["prior_bundles"] == {"trace-old"} + + def test_public_e2e_smoke_runs_n_transactions_on_one_session() -> None: root = Path(__file__).resolve().parents[1] result = subprocess.run(