Record attributable coding usage with explicit unknown cost

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
 

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)

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