diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md index 21176ace..94d30481 100644 --- a/docs/engineering/codex_dgx_worker.md +++ b/docs/engineering/codex_dgx_worker.md @@ -761,12 +761,64 @@ through `task_execution_timing_service.start` and `finish` (also available as Planned start/due dates and generic `updated_at` are not used. No historical timestamps are fabricated during upgrade. -Cost is explicitly `unknown` with a null amount/currency. Codex `turn.completed` -records can contain token usage, but the subscription execution route supplies -no trustworthy per-task monetary receipt or applicable price basis. This slice -does not sum parent/child calls or treat subscription tokens as an API invoice. -Attributable usage/cost needs separate controller-owned backlog work; it is not -a timing acceptance gate. +### Coding-attempt usage and cost + +The DGX controller reads its retained `events.jsonl` after timing acceptance and +uses `task_execution_timing_service.record_usage` to persist the observation on +the existing attempt. `get_task`, task listings and HTTP task detail project it +under `execution_timing.attempts[].usage`, with the task-local observed subtotal +under `execution_timing.usage`. No new consumer, model call or billing credential +is involved. The model-produced result is not a usage source. + +The supported source is a fresh `codex exec --json` invocation with one root +thread, one started turn and one unambiguous completed-turn receipt. The +[official event example](https://learn.chatgpt.com/docs/non-interactive-mode#make-output-machine-readable) +supplies input, cached-input and output counters. Receipts retain the source, +thread ID, provider turn ID when exposed (otherwise null with local ordinal 1), +configured model separately from the unknown observed model/provider, and the +SHA-256 of the retained event bytes. Raw prompts and tool contents stay outside +task metadata. These observations are controller-captured CLI reports, not +independently verified provider billing receipts. + +- Coverage is **partial**, even with valid root counters: child and external + tool coverage is unknown. Nested child/tool records are never added to a root + aggregate. Missing, malformed, multi-thread or multi-turn streams remain + **unknown**; resumed/cumulative thread counters are not guessed attempt deltas. +- Repeated canonical delivery replaces nothing and adds nothing. A new attempt + retains its own receipt; identical aggregates from the same thread count once + in the task subtotal. Conflicting observations of a reused thread exclude that + whole thread from the subtotal and expose `conflicting_threads`. Different + fresh root threads count as separate work. This is a task-local subtotal, not + a cross-task billing ledger. +- `observed_total_tokens` adds input and output only: cached input is a subset + of input. No observations yields null, not zero. Unknown attempts remain + visible alongside observed counters. Historical attempts are not backfilled. +- Usage recording failure retains `usage_recording_error_type` in controller + state and reports the limitation without blocking timing or completion. A + replay of `finish_execution` can retry accounting; there is no additional + automatic accounting retry consumer. Unknown receipts can be enriched by a + later observation; an already observed receipt cannot silently change. + +Cost remains explicitly `unknown` with a null amount/currency. Neither inspected +subscription route supplies a trustworthy per-task monetary receipt or an +applicable price basis. No actual charge or estimated allocation is fabricated +from API prices, configured model names or subscription token counts. + +The operator-owned bridge source inspected on 17 September 2026 (hash below) +binds `consumer_thread` at `begin` and checks it at `resume`, but exposes no token +receipt or task-specific turn interval. With the existing timing patch and the +matching backend, its shared `finish_execution` records **unknown** usage with +that thread ID and source `codex_vscode.consumer_thread`. A thread can contain +other work, so its lifetime counters cannot be attributed to the task. No +private conversation or Codex database is scanned for accounting. + +Acceptance uses the existing controller with a fixture executable emitting +documented events, followed by canonical service/message read-back, repeated +delivery, retry and conflicting-thread cases, and a usage-persistence failure. +The operator patch replay below also verifies explicit unknown usage. Run +`tests/backend/test_coding_execution_usage.py` alongside the timing tests. This +proves repository behaviour against isolated canonical services; activation and +live attempt observations require the operator's normal release boundary. ### Operator-owned VS Code bridge integration diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index 1d6da5bb..54cf64c3 100755 --- a/scripts/codex_von_worker.py +++ b/scripts/codex_von_worker.py @@ -847,6 +847,19 @@ def finish_execution(self, state): outcome=state["result"]["status"], ended_at=state.get("finished_at"), ) + # Optional accounting must never hold up accepted timing/completion. + # Only controller-retained events are read, never model result fields. + try: + receipt = execution_usage(state) + task = timing.record_usage( + state["task_id"], + attempt_id=state["attempt"], + actor_concept_id=self.config["agent_id"], + receipt=receipt, + ) + state.pop("usage_recording_error_type", None) + except Exception as exc: # noqa: BLE001 - accounting cannot gate timing + state["usage_recording_error_type"] = type(exc).__name__ state["execution_timing"] = task["execution_timing"] return task @@ -897,6 +910,10 @@ def finish(self, state): ).strip() if timing_receipt: content += "\n\n" + execution_timing_text(timing_receipt, state["attempt"]) + if state.get("usage_recording_error_type"): + content += ( + "\nUsage receipt could not be recorded; timing acceptance succeeded." + ) if blocker: content += "\n\n" + retry_policy.waiting_text(blocker) if state.get("execution_settings"): @@ -982,14 +999,43 @@ def finish(self, state): ) +def execution_usage(state): + from src.backend.services import coding_execution_usage as usage + + model = (state.get("execution_settings") or {}).get("model") + if state.get("execution_source") == "codex_vscode": + return usage.unknown( + "Interactive bridge has no trusted task-specific token receipt or turn interval", + configured_model=model, + thread_id=state.get("consumer_thread"), + source="codex_vscode.consumer_thread", + ) + if not state.get("run_dir"): + return usage.unknown("No retained exec stream", configured_model=model) + try: + with (Path(state["run_dir"]) / "events.jsonl").open("rb") as stream: + return usage.from_exec_stream(stream, configured_model=model) + except OSError: + return usage.unknown("Exec stream unavailable", configured_model=model) + + def execution_timing_text(timing, attempt_id): attempt = next(a for a in timing["attempts"] if a["attempt_id"] == attempt_id) + usage = attempt.get("usage") or {} + counters = usage.get("tokens") + usage_text = ( + f"Observed root tokens (partial coverage): input {counters['input_tokens']}, " + f"cached input {counters['cached_input_tokens']} (included in input), " + f"output {counters['output_tokens']}." + if counters + else "Attempt token usage: unknown." + ) return ( f"Execution start (UTC): {attempt['started_at']}\n" f"Execution end (UTC): {attempt.get('ended_at') or 'unknown'}\n" f"Task completion accepted (UTC): {attempt.get('completion_accepted_at') or 'not completed'}\n" f"Attempt elapsed wall-clock seconds (includes waits): {attempt.get('elapsed_wall_seconds')}\n" - "Task cost: unknown; subscription usage is not a per-task invoice." + f"{usage_text}\nTask cost: unknown; subscription usage is not a per-task invoice." ) @@ -1095,6 +1141,7 @@ def launch(config, state, inputs, conversation, state_path, lock_fd): "execution_started_at", "execution_timing", "execution_source", + "usage_recording_error_type", ): state.pop(key, None) state.pop("result", None) diff --git a/src/backend/services/coding_execution_usage.py b/src/backend/services/coding_execution_usage.py new file mode 100644 index 00000000..8bf3fd7b --- /dev/null +++ b/src/backend/services/coding_execution_usage.py @@ -0,0 +1,202 @@ +"""Usage observations, never subscription charges or API-price allocations. + +Only a fresh, single-root-turn exec stream is attributable without taking +possibly cumulative thread counters as attempt deltas. Child/tool usage is not +added to the root aggregate. Unsupported streams retain unknown coverage. +""" + +from __future__ import annotations + +import hashlib +import json + +TOKEN_FIELDS = ("input_tokens", "cached_input_tokens", "output_tokens") +SOURCE = "codex.exec.turn.completed" + + +def tokens(value): + if not isinstance(value, dict): + raise ValueError("Missing token counters") # noqa: TRY004 - invalid receipt value + result = {key: value.get(key) for key in TOKEN_FIELDS} + if any(type(n) is not int or n < 0 for n in result.values()): + raise ValueError("Token counters must be non-negative integers") + if result["cached_input_tokens"] > result["input_tokens"]: + raise ValueError("Cached tokens are a subset of input tokens") + return result + + +def unknown(reason, *, configured_model=None, thread_id=None, source=SOURCE): + return { + "coverage": "unknown", + "reason": reason, + "source": source, + "provider": None, + "observed_model": None, + "configured_model": configured_model, + "thread_id": thread_id, + "turn_id": None, + "turn_ordinal": None, + "source_sha256": None, + "tokens": None, + } + + +def from_exec_stream(lines, *, configured_model=None): + """Inspect only top-level lifecycle events from one controller-owned run. + + A digest binds the receipt to the retained bytes without publishing prompts, + tool arguments, paths or raw event contents into task-visible metadata. + """ + receipt = unknown( + "No attributable completed root turn", configured_model=configured_model + ) + digest = hashlib.sha256() + threads, starts, completions = set(), [], [] + invalid = False + for line in lines: + digest.update(line) + try: + event = json.loads(line) + except (ValueError, UnicodeDecodeError): + invalid = True + continue + if not isinstance(event, dict): + invalid = True + continue + kind = event.get("type") + if kind == "thread.started": + if event.get("parent_thread_id") or event.get("parent_turn_id"): + invalid = True + thread = event.get("thread_id") + if isinstance(thread, str) and thread: + threads.add(thread) + else: + invalid = True + elif kind == "turn.started": + if len(threads) != 1: + invalid = True + starts.append(event) + elif kind == "turn.completed": + if len(starts) != 1: + invalid = True + # Repeated identical terminal delivery is one observation. + if event not in completions: + completions.append(event) + elif kind == "turn.failed": + invalid = True + receipt["source_sha256"] = digest.hexdigest() + if len(threads) == 1: + receipt["thread_id"] = next(iter(threads)) + if invalid or len(threads) != 1 or len(starts) != 1 or len(completions) != 1: + receipt["reason"] = "Missing, malformed or ambiguous single-root-turn stream" + return receipt + completed = completions[0] + for event in (starts[0], completed): + if event.get("thread_id", receipt["thread_id"]) != receipt["thread_id"]: + receipt["reason"] = "Event thread differs from the root thread" + return receipt + if event.get("parent_thread_id") or event.get("parent_turn_id"): + receipt["reason"] = "Child usage cannot be added to a root aggregate" + return receipt + turn_id = completed.get("turn_id") or starts[0].get("turn_id") + if turn_id is not None and (not isinstance(turn_id, str) or not turn_id): + return receipt + if ( + starts[0].get("turn_id") + and completed.get("turn_id") + and starts[0]["turn_id"] != completed["turn_id"] + ): + return receipt + try: + counters = tokens(completed.get("usage")) + except ValueError: + receipt["reason"] = "Missing or invalid terminal token counters" + return receipt + receipt.update( + coverage="partial", + reason="Observed root aggregate only; child/tool coverage and provider model are unknown", + tokens=counters, + turn_id=turn_id, + turn_ordinal=1, + ) + return receipt + + +def validate(receipt): + """Canonical writer accepts the bounded observation, never money fields.""" + if not isinstance(receipt, dict) or set(receipt) != set(unknown("")): + raise ValueError("Invalid coding usage receipt fields") + if receipt["source"] not in {SOURCE, "codex_vscode.consumer_thread"} or receipt[ + "coverage" + ] not in {"unknown", "partial"}: + raise ValueError("Unsupported coding usage source or coverage") + if receipt["provider"] is not None or receipt["observed_model"] is not None: + raise ValueError("Exec usage does not attest provider/model identity") + if not isinstance(receipt["reason"], str) or not receipt["reason"]: + raise ValueError("Usage coverage requires a reason") + for key in ("configured_model", "thread_id", "turn_id"): + if receipt[key] is not None and ( + not isinstance(receipt[key], str) or not receipt[key] + ): + raise ValueError("Invalid usage identity") + sha = receipt["source_sha256"] + if sha is not None and ( + not isinstance(sha, str) + or len(sha) != 64 + or any(c not in "0123456789abcdef" for c in sha) + ): + raise ValueError("Invalid usage evidence digest") + if receipt["coverage"] == "partial": + if receipt["source"] != SOURCE: + raise ValueError("Interactive thread IDs do not attest task token usage") + if ( + not receipt["thread_id"] + or not sha + or type(receipt["turn_ordinal"]) is not int + or receipt["turn_ordinal"] != 1 + ): + raise ValueError("Observed usage requires a single root turn and evidence") + if receipt["tokens"] != tokens(receipt["tokens"]): + raise ValueError("Unsupported token fields") + elif receipt["tokens"] is not None or receipt["turn_ordinal"] is not None: + raise ValueError("Unknown usage cannot supply token totals") + + +def project(attempts): + """Task-local observed subtotal; duplicate thread aggregates never add twice. + + Different observations of a reused thread might be cumulative. Exclude the + whole conflicting group instead of guessing a delta or summing snapshots. + """ + by_thread = {} + observed_attempts = 0 + for attempt in attempts: + receipt = attempt.get("usage") or {} + if receipt.get("coverage") != "partial": + continue + observed_attempts += 1 + by_thread.setdefault(receipt["thread_id"], []).append(receipt["tokens"]) + included, conflicts = [], [] + for thread, observations in by_thread.items(): + if all(value == observations[0] for value in observations): + included.append(observations[0]) + else: + conflicts.append(thread) + subtotal = ( + {key: sum(value[key] for value in included) for key in TOKEN_FIELDS} + if included + else None + ) + return { + "coverage": "partial" if subtotal is not None else "unknown", + "observed_tokens": subtotal, + "observed_total_tokens": (subtotal["input_tokens"] + subtotal["output_tokens"]) + if subtotal is not None + else None, + "total_basis": "Observed root aggregates only; cached input is already included in input", + "observed_attempts": observed_attempts, + "unknown_attempts": len(attempts) - observed_attempts, + "distinct_root_threads": len(included), + "conflicting_threads": sorted(conflicts), + "reason": "Task-wide coverage is incomplete; no child totals or monetary charges are inferred", + } diff --git a/src/backend/services/task_execution_timing_service.py b/src/backend/services/task_execution_timing_service.py index 85738c8b..e78f7773 100644 --- a/src/backend/services/task_execution_timing_service.py +++ b/src/backend/services/task_execution_timing_service.py @@ -9,6 +9,7 @@ from copy import deepcopy from datetime import UTC, datetime +from . import coding_execution_usage as usage from .task_project_home_service import guard_task_write KEY = "execution_timing" @@ -65,6 +66,7 @@ def project(metadata, status): "duration_basis": "elapsed wall clock, including waits; not active coding time", "attempts": attempts, "completion_history": completions, + "usage": usage.project(attempts), "cost": { "status": "unknown", "amount": None, @@ -115,6 +117,26 @@ def _actor(task, actor): raise PermissionError("Execution timing must be recorded by the assigned actor") +@guard_task_write +def record_usage(task_concept_id, *, attempt_id, actor_concept_id, receipt): + """Persist controller-observed usage independently of timing acceptance.""" + from . import task_management_service as tasks + + usage.validate(receipt) + _actor(tasks.get_task(task_concept_id), actor_concept_id) + + def apply(timing, doc): + attempt = _attempt(timing, attempt_id) + if not attempt or attempt["actor_concept_id"] != actor_concept_id: + raise ValueError("Usage requires the actor's existing execution attempt") + prior = attempt.get("usage") + if prior and prior != receipt and prior["coverage"] != "unknown": + raise ValueError("Attempt already has a different usage receipt") + attempt["usage"] = deepcopy(receipt) + + return _change(task_concept_id, apply) + + @guard_task_write def start(task_concept_id, *, attempt_id, actor_concept_id, started_at, source): from . import task_management_service as tasks diff --git a/tests/backend/test_coding_execution_usage.py b/tests/backend/test_coding_execution_usage.py new file mode 100644 index 00000000..5636dd77 --- /dev/null +++ b/tests/backend/test_coding_execution_usage.py @@ -0,0 +1,229 @@ +"""Replay the existing controller with fixture events, then read canonical tasks.""" + +import io +import json +from pathlib import Path + +import pytest +from test_codex_von_retry import canonical_controller, config # noqa: F401 + +from scripts import codex_von_worker as worker +from src.backend.services import coding_execution_usage as usage +from src.backend.services import task_execution_timing_service as timing + +# ruff: noqa: F811 + +COUNTERS = {"input_tokens": 100, "cached_input_tokens": 60, "output_tokens": 20} + + +def events(thread="root", counters=None): + return [ + {"type": "thread.started", "thread_id": thread}, + {"type": "turn.started"}, + {"type": "turn.completed", "usage": counters or COUNTERS}, + ] + + +def parse(rows): + return usage.from_exec_stream( + io.BytesIO(b"".join(json.dumps(row).encode() + b"\n" for row in rows)), + configured_model="gpt-6-astra", + ) + + +def record(api, task_id, receipt, attempt="first"): + return timing.record_usage( + task_id, + attempt_id=attempt, + actor_concept_id=api.config["agent_id"], + receipt=receipt, + ) + + +def start(api, task_id, attempt="first"): + return timing.start( + task_id, + attempt_id=attempt, + actor_concept_id=api.config["agent_id"], + started_at="2026-09-17T10:00:00Z", + source="codex_dgx", + ) + + +def test_root_only_duplicate_terminal_and_identity(): + rows = events() + # Nested child counters and tool result text are not independent receipts. + rows.insert( + 2, + { + "type": "item.completed", + "item": { + "type": "collab_tool_call", + "usage": COUNTERS, + "result": events("child"), + }, + }, + ) + rows.append(rows[-1]) + receipt = parse(rows) + usage.validate(receipt) + assert receipt["tokens"] == COUNTERS + assert receipt["coverage"] == "partial" + assert receipt["thread_id"] == "root" + assert receipt["turn_id"] is None + assert receipt["turn_ordinal"] == 1 + assert len(receipt["source_sha256"]) == 64 + assert receipt["configured_model"] == "gpt-6-astra" + assert receipt["observed_model"] is None + assert receipt["provider"] is None + rows = events() + rows[1]["turn_id"] = rows[2]["turn_id"] = "observed-turn" + assert parse(rows)["turn_id"] == "observed-turn" + + +@pytest.mark.parametrize( + "rows", + [ + [], + events()[:-1], + events()[1:], + events() + events("child"), + events() + events()[1:], + events() + + [{"type": "turn.completed", "usage": dict(COUNTERS, output_tokens=21)}], + events(counters=dict(COUNTERS, input_tokens=True)), + events(counters=dict(COUNTERS, cached_input_tokens=101)), + events(counters=dict(COUNTERS, output_tokens=-1)), + events(counters={"input_tokens": 10}), + [dict(events()[0], parent_thread_id="parent"), *events()[1:]], + [*events()[:-1], dict(events()[-1], thread_id="foreign")], + [*events()[:-1], dict(events()[-1], parent_turn_id="parent")], + [*events(), {"type": "turn.failed"}], + [*events(), None], + ], +) +def test_unknown_is_not_zero_or_a_guessed_delta(rows): + receipt = parse(rows) + usage.validate(receipt) + assert receipt["coverage"] == "unknown" + assert receipt["tokens"] is None + + +def test_truncated_stream_stays_unknown(): + raw = b"".join(json.dumps(row).encode() + b"\n" for row in events()) + receipt = usage.from_exec_stream(io.BytesIO(raw + b'{"type":')) + assert receipt["coverage"] == "unknown" + + +def test_canonical_replay_retry_unknown_and_conflicting_thread(canonical_controller): + _, api, create, *_ = canonical_controller + task_id = create() + start(api, task_id) + receipt = parse(events()) + first = record(api, task_id, receipt)["execution_timing"] + assert record(api, task_id, receipt)["execution_timing"] == first + assert api.task(task_id)["execution_timing"] == first + assert ( + first["usage"]["observed_total_tokens"] == 120 + ) # cached input is not added again + assert first["cost"]["amount"] is None + assert first["cost"]["currency"] is None + with pytest.raises(ValueError, match="different usage"): + record(api, task_id, parse(events(counters=dict(COUNTERS, output_tokens=21)))) + # A new attempt replaying the same root must not double count it. + start(api, task_id, "retry") + record(api, task_id, receipt, "retry") + start(api, task_id, "unknown") + record(api, task_id, usage.unknown("No receipt"), "unknown") + readback = api.task(task_id)["execution_timing"]["usage"] + assert readback["observed_total_tokens"] == 120 + assert readback["distinct_root_threads"] == 1 + assert readback["unknown_attempts"] == 1 + assert readback["coverage"] == "partial" + # New root execution is real extra work, not repeated result delivery. + record(api, task_id, parse(events("new-root")), "unknown") + assert ( + api.task(task_id)["execution_timing"]["usage"]["observed_total_tokens"] == 240 + ) + # Conflicting cumulative counters on reused thread cannot safely be summed. + start(api, task_id, "conflict") + record( + api, + task_id, + parse(events(counters=dict(COUNTERS, output_tokens=21))), + "conflict", + ) + readback = api.task(task_id)["execution_timing"]["usage"] + assert readback["observed_total_tokens"] == 120 + assert readback["conflicting_threads"] == ["root"] + + +def test_receipts_cannot_assert_charges_or_unbound_attempts(canonical_controller): + _, api, create, *_ = canonical_controller + task_id = create() + with pytest.raises(ValueError, match="existing execution"): + record(api, task_id, parse(events())) + start(api, task_id) + with pytest.raises(ValueError, match="receipt fields"): + record( + api, task_id, dict(parse(events()), cost={"status": "actual", "amount": 1}) + ) + with pytest.raises(ValueError, match="identity"): + record(api, task_id, dict(parse(events()), observed_model="gpt-6-astra")) + with pytest.raises(PermissionError): + timing.record_usage( + task_id, + attempt_id="first", + actor_concept_id="#V#other", + receipt=parse(events()), + ) + + +@pytest.mark.parametrize("accounting_unavailable", [False, True]) +def test_dgx_supported_delivery_and_timing_independence( + canonical_controller, monkeypatch, accounting_unavailable +): + cfg, api, create, read_state, launches, fd = canonical_controller + runner = Path(cfg["codex_command"]) + # Exercise stdout capture, process exit, result acceptance and messaging. + script = runner.read_text().replace("'status': 'blocked'", "'status': 'completed'") + runner.write_text( + script + + "\n" + + "\n".join("print(" + repr(json.dumps(row)) + ")" for row in events()) + ) + if accounting_unavailable: + + def unavailable(*args, **kwargs): + raise OSError("Fixture usage persistence unavailable") + + monkeypatch.setattr(timing, "record_usage", unavailable) + task_id = create() + worker.tick(cfg, api, fd) + _, state = read_state(task_id) + task = api.task(task_id) + assert task["status"] == "completed" + assert task["execution_timing"]["completed_at"] + receipt = task["execution_timing"] + if accounting_unavailable: + assert receipt["usage"]["observed_tokens"] is None + assert state["usage_recording_error_type"] == "OSError" + else: + attempt = receipt["attempts"][0] + assert attempt["attempt_id"] == state["attempt"] + assert attempt["usage"]["tokens"] == COUNTERS + assert attempt["usage"]["thread_id"] == "root" + assert receipt["usage"]["observed_total_tokens"] == 120 + # Replayed result acceptance (including lost acknowledgements) is idempotent. + assert api.finish_execution(state)["execution_timing"] == receipt + message = api.messages.get_message_for_user( + state["message_id"], cfg["delegator_id"] + ) + text = api.messages.project_direct_message(message)["content"] + assert "subscription usage is not a per-task invoice" in text + assert ( + "partial coverage" if not accounting_unavailable else "could not be recorded" + ) in text + worker.tick(cfg, api, fd) + assert len(launches()) == 1 + assert api.task(task_id)["execution_timing"] == receipt diff --git a/tests/backend/test_task_execution_timing.py b/tests/backend/test_task_execution_timing.py index f7ea330c..856dec86 100644 --- a/tests/backend/test_task_execution_timing.py +++ b/tests/backend/test_task_execution_timing.py @@ -325,6 +325,10 @@ def unavailable(*args, **kwargs): receipt = api.task(task_id)["execution_timing"] assert receipt["completed_at"] assert receipt["attempts"][0]["source"] == "codex_vscode" + assert receipt["attempts"][0]["usage"]["coverage"] == "unknown" + assert receipt["attempts"][0]["usage"]["thread_id"] == "fixture-thread" + assert receipt["attempts"][0]["usage"]["source"] == "codex_vscode.consumer_thread" + assert receipt["usage"]["observed_tokens"] is None assert ( receipt["completed_at"] in messages["vscode-task:" + request["attempt"] + ":result"]