Persist coding task execution and completion timing

docs/engineering/codex_dgx_worker.md

@@ -726,3 +726,77 @@ changes and source instruction. Preserve deliberate choices and ambiguous cases.
 Apply only reviewed entries after the eligibility fix is installed, through
 canonical task updates, and read back the effective ontology default. Do not use
 a blanket rewrite based solely on the current field matching the delegator.
+
+## Canonical execution timing
+
+The task service exposes `execution_timing` through `get_task`, task listings
+and the existing HTTP task-detail response. This is task-visible lifecycle
+metadata, using the controller's existing attempt ID; it does not grant access
+to private run archives. The supported writers are the trusted controllers,
+through `task_execution_timing_service.start` and `finish` (also available as
+`Von.start_execution` and `Von.finish_execution`). No new task consumer is added.
+
+- DGX records `execution_started_at` after its subprocess exists and before
+  supplying the task prompt. Queue pickup, checkout preparation and archive
+  registration are distinct; the previous local `started_at` was preparation
+  time and must not be imported as execution evidence.
+- The interactive Mac bridge records execution at its authorised `begin`
+  action, when the existing consumer takes the task. `select` only reserves it.
+  `resume` reconciles the same attempt and retains its original start.
+- An observed process exit supplies DGX `ended_at`; the Mac bridge supplies
+  the first retained `finish` time. Missing exit evidence leaves the end unknown.
+  Terminal attempts retain their outcome separately from successful completion.
+- Canonical `finish` retains the terminal observation, reconciles the existing
+  status path, then records `result_accepted_at` and, only for success,
+  `completion_accepted_at`. Repeated delivery preserves those timestamps.
+  Interrupted acceptance can resume after the status write without another run.
+- `started_at` is the earliest recorded execution start. `completed_at` is the
+  accepted completion for the current lifecycle, or null when uncompleted or
+  unknown. Reopening keeps `completion_history` and all prior attempts. A
+  previously accepted attempt cannot complete a reopened task. An external
+  completion is not attributed to an unfinished coding attempt.
+- Durations are **elapsed wall-clock seconds, including waits**, not active
+  coding effort. Attempt duration ends at the observed execution end; task
+  duration ends at accepted completion, so reporting/archive waits can differ.
+  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.
+
+### Operator-owned VS Code bridge integration
+
+The repository cannot activate or directly edit the operator-owned Mac bridge.
+[The bounded patch](patches/codex_vscode_task_timing.patch) adds the shared
+start/resume/finish calls to its existing `task-server.py`, preserves its
+assignment/input checks and lock, and includes canonical timing in its result
+message. It applies to the source inspected on 17 September 2026 with SHA-256
+`a665516dfe918b6e77046867721ad471c787535bf5e8a92d7aa677c57f19df5b`.
+If that source changed, review/rebase the patch against the new source rather
+than replacing the bridge with an old copy. Existing attempts lacking the new
+execution observation keep their historical state and are not assigned upgrade
+timestamps. No secrets or operator configuration belong in the patch.
+
+The isolated replay test reads the operator source named by
+`CODING_TIMING_MAC_BRIDGE_SOURCE`, applies the patch only to a temporary copy,
+and executes its function definitions with fixture state and canonical services
+backed by a mock database. It never executes the host bootstrap or messaging
+helper. Run with PDM:
+
+```sh
+VON_USE_MOCK_DB=1 CODING_TIMING_MAC_BRIDGE_SOURCE=/operator/path/task-server.py \
+  pdm run pytest tests/backend/test_task_execution_timing.py -q
+```
+
+Without the source path, only the optional operator replay is skipped; canonical
+service and DGX supported-path tests still run. This evidence proves the patched
+code path, not an installed Mac consumer. Apply the reviewed patch and select the
+matching backend/worker release at an idle boundary under the existing lock,
+with the prior bridge and release retained for rollback. Read back a supported
+begin/resume/finish attempt on each installed adapter before claiming activation.
+Preserve the current schedule, identities, active consumers and federation/import
+state. Public web deployment is a separate authorised action.

docs/engineering/patches/codex_vscode_task_timing.patch

@@ -0,0 +1,54 @@
+--- a/task-server.py
++++ b/task-server.py
+@@ -162,4 +162,6 @@
+                 or not request.get('checkpoint', '').strip()):
+             return dict(state, may_execute=False)
++        if state.get('execution_started_at'):
++            api.start_execution(state, source='codex_vscode')
+         state.setdefault('recovery_checkpoints', []).append(dict(at=now(), evidence=request['checkpoint']))
+         save(path, state)
+@@ -196,5 +198,8 @@
+         state.pop('pending_admission', None)
+         state.update(phase='running', started_at=now(), consumer_thread=request['thread_id'])
++        state['execution_started_at'] = state['started_at']
++        state['execution_source'] = 'codex_vscode'
+         save(path, state)
++        api.start_execution(state, source='codex_vscode')
+         api.tasks.update_task_status(state['task_id'], 'in_progress', actor_concept_id=CONFIG['agent_id'])
+         assert api.task(state['task_id'])['status'] == 'in_progress'
+@@ -218,4 +223,5 @@
+         raise ValueError('Retained result differs; reconcile it first')
+     state.update(phase='reporting', result=result)
++    state.setdefault('finished_at', now())
+     blocker = retry_policy.retain_blocker(CONFIG, state)
+     if 'result_content' not in state:
+@@ -229,4 +235,22 @@
+         state['waiting_explanation'] = retry_policy.waiting_text(blocker)
+         state['result_content'] += '\n\n' + state['waiting_explanation']
++    _, timing_inputs = task_inputs(api, task)
++    if (state.get('execution_started_at') and state['target_status']
++            and worker.authorised_task(task, CONFIG) and api.native_writer(task)
++            and task['status'] in {'in_progress', state['target_status']}
++            and (task['status'] != 'completed' or any(
++                a['attempt_id'] == attempt and a.get('outcome') == 'completed'
++                for a in task.get('execution_timing', {}).get('attempts', [])))
++            and worker.fingerprint(timing_inputs) == snapshot_hash(state, worker)):
++        save(path, state)
++        if valid(api, worker, task):
++            api.record_supervisor_repair(task, result)
++        timed_task = api.finish_execution(state)
++        if timed_task['status'] != state['target_status']:
++            state['target_status'] = None
++            state['status_update_withheld'] = 'Accepted attempt belongs to an earlier lifecycle'
++        if not state.get('timing_in_result_content'):
++            state['result_content'] += '\n\n' + worker.execution_timing_text(state['execution_timing'], attempt)
++            state['timing_in_result_content'] = True
+     save(path, state)
+     # Canonical read-back precedes terminal bookkeeping; retry uses the same intent.
+@@ -248,5 +272,5 @@
+         raise RuntimeError('Task status read-back failed')
+     state.update(phase='done' if readback['status'] == 'completed' else 'waiting',
+-                 finished_at=now(), task_status_readback=readback['status'])
++                 task_status_readback=readback['status'])
+     save(path, state)
+     return state

scripts/codex_von_worker.py

@@ -822,6 +822,34 @@ def send(self, task, key, content):
             raise RuntimeError("Direct message canonical read-back failed")
         return message_id
 
+    def start_execution(self, state, *, source="codex_dgx"):
+        from src.backend.services import task_execution_timing_service as timing
+
+        state["execution_source"] = source
+        task = timing.start(
+            state["task_id"],
+            attempt_id=state["attempt"],
+            actor_concept_id=self.config["agent_id"],
+            started_at=state["execution_started_at"],
+            source=source,
+        )
+        state["execution_timing"] = task["execution_timing"]
+        return task
+
+    def finish_execution(self, state):
+        from src.backend.services import task_execution_timing_service as timing
+
+        self.start_execution(state, source=state.get("execution_source", "codex_dgx"))
+        task = timing.finish(
+            state["task_id"],
+            attempt_id=state["attempt"],
+            actor_concept_id=self.config["agent_id"],
+            outcome=state["result"]["status"],
+            ended_at=state.get("finished_at"),
+        )
+        state["execution_timing"] = task["execution_timing"]
+        return task
+
     def finish(self, state):
         try:
             task = self.task(state["task_id"])
@@ -837,12 +865,38 @@ def finish(self, state):
             and task.get("status") in ACTIVE
             and self.native_writer(task)
         )
+        timing_receipt = None
+        if state.get("execution_started_at") and (
+            scope_valid
+            or (
+                authorised_task(task, self.config)
+                and self.native_writer(task)
+                and task.get("status") == "completed"
+                and result["status"] == "completed"
+                and any(
+                    a["attempt_id"] == state["attempt"] and a.get("outcome") == "completed"
+                    for a in task.get("execution_timing", {}).get("attempts", [])
+                )
+            )
+        ):
+            # Supervisor receipts require the still-active repair assignment.
+            if scope_valid:
+                self.record_supervisor_repair(task, result)
+            # Canonical acceptance precedes its completion message. Replaying
+            # this attempt cannot complete a reopened task.
+            timing_receipt = self.finish_execution(state)["execution_timing"]
+            task = self.task(state["task_id"])
+            scope_valid = task["status"] == (
+                "completed" if result["status"] == "completed" else "blocked"
+            )
         content = (
             f"{result['summary']}\n\n{result['evidence']}\n\n"
             f"{result['question']}\n\nTask: {state['task_id']}\n"
             f"DGX worktree: {state.get('worktree', 'not started')}\n"
             f"Run: {state['attempt']}"
         ).strip()
+        if timing_receipt:
+            content += "\n\n" + execution_timing_text(timing_receipt, state["attempt"])
         if blocker:
             content += "\n\n" + retry_policy.waiting_text(blocker)
         if state.get("execution_settings"):
@@ -874,7 +928,8 @@ def finish(self, state):
             state["report_task"], state["attempt"] + ":result", state["report_content"]
         )
         if scope_valid:
-            self.record_supervisor_repair(task, result)
+            if timing_receipt is None:
+                self.record_supervisor_repair(task, result)
             evidence_addition = (
                 f"{result['evidence']}\nVon result: {message_id}\n"
                 f"Execution archive: {capture.get('archive_url', 'unavailable')}\n"
@@ -927,6 +982,17 @@ def finish(self, state):
         )
 
 
+def execution_timing_text(timing, attempt_id):
+    attempt = next(a for a in timing["attempts"] if a["attempt_id"] == attempt_id)
+    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."
+    )
+
+
 def command(args, **kwargs):
     return subprocess.run(
         args, check=True, text=True, capture_output=True, **kwargs
@@ -1017,14 +1083,18 @@ def launch(config, state, inputs, conversation, state_path, lock_fd):
         run_dir=str(run_dir),
         input_hash=input_fingerprint(inputs),
         input_snapshot=inputs,
-        started_at=datetime.now(UTC).isoformat(),
+        prepared_at=datetime.now(UTC).isoformat(),
         worker_revision=config.get("runtime_evidence", {}).get("commit"),
     )
     for key in (
         "capture",
         "capture_provenance",
         "archive_pending_reported",
         "finished_at",
+        "started_at",
+        "execution_started_at",
+        "execution_timing",
+        "execution_source",
     ):
         state.pop(key, None)
     state.pop("result", None)
@@ -1146,6 +1216,14 @@ def launch(config, state, inputs, conversation, state_path, lock_fd):
             stderr=errors,
             pass_fds=(lock_fd,),
         ) as process:
+            # The child is alive and waiting for its authorised input. Queue,
+            # checkout preparation and reservation are not execution time.
+            state["started_at"] = state["execution_started_at"] = datetime.now(
+                UTC
+            ).isoformat()
+            write_json(state_path, state)
+            api.start_execution(state)
+            write_json(state_path, state)
             process.stdin.write(prompt)
             process.stdin.close()
             while True:
@@ -1168,7 +1246,10 @@ def recover_result(state):
     """Only a successful process plus valid structured output is completion."""
     run_dir = Path(state["run_dir"])
     try:
-        if json.loads((run_dir / "exit.json").read_text())["returncode"] != 0:
+        exit_receipt = json.loads((run_dir / "exit.json").read_text())
+        if exit_receipt.get("finished_at"):
+            state["finished_at"] = exit_receipt["finished_at"]
+        if exit_receipt["returncode"] != 0:
             raise ValueError("Codex exited unsuccessfully")
         result = validate_result(json.loads((run_dir / "result.json").read_text()))
     except (OSError, ValueError, KeyError) as exc: