diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md index 16d61eca..21176ace 100644 --- a/docs/engineering/codex_dgx_worker.md +++ b/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. diff --git a/docs/engineering/patches/codex_vscode_task_timing.patch b/docs/engineering/patches/codex_vscode_task_timing.patch new file mode 100644 index 00000000..8a64574d --- /dev/null +++ b/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 diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index dee3db47..1d6da5bb 100755 --- a/scripts/codex_von_worker.py +++ b/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,7 +1083,7 @@ 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 ( @@ -1025,6 +1091,10 @@ def launch(config, state, inputs, conversation, state_path, lock_fd): "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: diff --git a/src/backend/services/task_execution_timing_service.py b/src/backend/services/task_execution_timing_service.py new file mode 100644 index 00000000..85738c8b --- /dev/null +++ b/src/backend/services/task_execution_timing_service.py @@ -0,0 +1,221 @@ +"""Coding execution observations on the canonical task, independent of archives. + +Controllers supply observed process/consumer times, never queue/import times. +Attempt IDs are the existing controller IDs; compare-and-set preserves retries. +""" + +from __future__ import annotations + +from copy import deepcopy +from datetime import UTC, datetime + +from .task_project_home_service import guard_task_write + +KEY = "execution_timing" + + +def timestamp(value): + if value is None: + return None + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + raise ValueError("Execution timestamps require an explicit UTC offset") + return parsed.astimezone(UTC).isoformat() + + +def elapsed(start, end): + if not start or not end: + return None + return (datetime.fromisoformat(end) - datetime.fromisoformat(start)).total_seconds() + + +def project(metadata, status): + timing = deepcopy(metadata.get(KEY) or {}) + attempts = timing.get("attempts", []) + starts = [a["started_at"] for a in attempts if a.get("started_at")] + completions = [ + {"attempt_id": a["attempt_id"], "completed_at": a["completion_accepted_at"]} + for a in attempts + if a.get("completion_accepted_at") + ] + completions.sort(key=lambda row: row["completed_at"]) + started = min(starts) if starts else None + completed = ( + completions[-1]["completed_at"] + if completions + and status in {"completed", "deployed"} + and any( + a.get("completion_accepted_at") == completions[-1]["completed_at"] + and a.get("completion_transition_count") + == ( + metadata.get("task_status_transition_count", 0) - (status == "deployed") + ) + for a in attempts + ) + else None + ) + for attempt in attempts: + attempt["elapsed_wall_seconds"] = elapsed( + attempt.get("started_at"), attempt.get("ended_at") + ) + return { + "started_at": started, + "completed_at": completed, + "elapsed_wall_seconds": elapsed(started, completed), + "duration_basis": "elapsed wall clock, including waits; not active coding time", + "attempts": attempts, + "completion_history": completions, + "cost": { + "status": "unknown", + "amount": None, + "currency": None, + "reason": "Subscription usage is not a per-task invoice; no attributable price receipt.", + }, + } + + +def _change(task_id, change): + from . import task_management_service as tasks + + for _ in range(8): + task_id, doc = tasks._get_task_doc(task_id) + previous = (doc.get("metadata") or {}).get(KEY) + timing = deepcopy(previous or {"attempts": []}) + change(timing, doc) + if timing == previous: + return tasks.get_task(task_id) + result = tasks.ConceptsRepository.update_one( + { + "concept_id": task_id, + f"metadata.{KEY}": previous, + "metadata.task_status_transition_count": ( + doc.get("metadata") or {} + ).get("task_status_transition_count"), + }, + {"$set": {f"metadata.{KEY}": timing}}, + ) + if result.matched_count: + return tasks.get_task(task_id) + raise RuntimeError("Concurrent execution timing update; retry the retained receipt") + + +def _attempt(timing, attempt_id): + return next((a for a in timing["attempts"] if a["attempt_id"] == attempt_id), None) + + +def _actor(task, actor): + from ..security.access_control import get_effective_user_concept_id + + trusted = get_effective_user_concept_id() + if ( + not actor + or (trusted and trusted != actor) + or task.get("assignee_concept_id") != actor + ): + raise PermissionError("Execution timing must be recorded by the assigned actor") + + +@guard_task_write +def start(task_concept_id, *, attempt_id, actor_concept_id, started_at, source): + from . import task_management_service as tasks + + if not isinstance(attempt_id, str) or not attempt_id.strip() or not source: + raise ValueError("An existing controller attempt ID and source are required") + started_at = timestamp(started_at) + if not started_at: + raise ValueError("Observed execution start is required") + task = tasks.get_task(task_concept_id) + _actor(task, actor_concept_id) + + def apply(timing, doc): + existing = _attempt(timing, attempt_id) + if existing: + if ( + existing["actor_concept_id"] != actor_concept_id + or existing["source"] != source + ): + raise ValueError("Attempt already bound to another actor/source") + # Resume/replayed begin must retain the original observed start. + return + if task["status"] not in {"pending", "in_progress", "blocked"}: + raise ValueError("Task is not open for execution") + timing["attempts"].append( + { + "attempt_id": attempt_id, + "actor_concept_id": actor_concept_id, + "source": source, + "started_at": started_at, + } + ) + + return _change(task_concept_id, apply) + + +@guard_task_write +def finish(task_concept_id, *, attempt_id, actor_concept_id, outcome, ended_at=None): + """Accept a retained result, reconcile status, then persist canonical acceptance. + + A crash after status transition is retried with the same attempt. An already + accepted attempt cannot complete a reopened task. Missing exit times stay + unknown; result acceptance is a separate observed timestamp. + """ + from . import task_management_service as tasks + + if outcome not in {"completed", "blocked", "needs_input", "failed"}: + raise ValueError("Invalid execution outcome") + ended_at = timestamp(ended_at) + task = tasks.get_task(task_concept_id) + _actor(task, actor_concept_id) + + def retain(timing, doc): + attempt = _attempt(timing, attempt_id) + if not attempt or attempt["actor_concept_id"] != actor_concept_id: + raise ValueError("Execution attempt start was not recorded") + if attempt.get("outcome"): + if attempt["outcome"] != outcome or attempt.get("ended_at") != ended_at: + raise ValueError("Attempt already has a different terminal receipt") + return + if task["status"] not in {"pending", "in_progress", "blocked"}: + raise ValueError("Task became terminal before this result was retained") + if ended_at and elapsed(attempt["started_at"], ended_at) < 0: + raise ValueError("Execution end precedes start") + attempt.update( + outcome=outcome, + ended_at=ended_at, + result_base_transition_count=(doc.get("metadata") or {}).get( + "task_status_transition_count", 0 + ), + ) + + task = _change(task_concept_id, retain) + attempt = _attempt(task[KEY], attempt_id) + if attempt.get("result_accepted_at"): + return task + target = "completed" if outcome == "completed" else "blocked" + if task["status"] not in {"pending", "in_progress", "blocked", target}: + raise ValueError("Task changed before execution result acceptance") + _, doc = tasks._get_task_doc(task_concept_id) + transitions = (doc.get("metadata") or {}).get("task_status_transition_count", 0) + base = attempt["result_base_transition_count"] + if transitions > base + 1 or (transitions == base + 1 and task["status"] != target): + raise ValueError("Task reopened or changed after the retained execution result") + expected_transitions = transitions + (task["status"] != target) + tasks.update_task_status(task_concept_id, target, actor_concept_id=actor_concept_id) + if tasks.get_task(task_concept_id)["status"] != target: + raise RuntimeError("Execution result status read-back failed") + + def accept(timing, doc): + if (doc.get("metadata") or {}).get( + "task_status_transition_count", 0 + ) != expected_transitions: + raise ValueError("Task status changed before timing acceptance") + attempt = _attempt(timing, attempt_id) + if not attempt.get("result_accepted_at"): + attempt["result_accepted_at"] = datetime.now(UTC).isoformat() + if outcome == "completed": + attempt["completion_accepted_at"] = attempt["result_accepted_at"] + attempt["completion_transition_count"] = ( + doc.get("metadata") or {} + ).get("task_status_transition_count", 0) + + return _change(task_concept_id, accept) diff --git a/src/backend/services/task_management_service.py b/src/backend/services/task_management_service.py index 615d1240..ae5924ed 100644 --- a/src/backend/services/task_management_service.py +++ b/src/backend/services/task_management_service.py @@ -2189,7 +2189,10 @@ def _build_task_response( except Exception: pass + from .task_execution_timing_service import project as project_execution_timing + return { + "execution_timing": project_execution_timing(metadata, status), "task_concept_id": task_concept_id, "title": title or doc.get("name", "Untitled Task"), "description": description, diff --git a/tests/backend/test_codex_von_retry.py b/tests/backend/test_codex_von_retry.py index 034e3973..06b0f656 100644 --- a/tests/backend/test_codex_von_retry.py +++ b/tests/backend/test_codex_von_retry.py @@ -240,6 +240,7 @@ def canonical_controller(config, tmp_path, monkeypatch): } fake.write_text( f"#!{sys.executable}\n" + "import json, sys\nfrom pathlib import Path\n" + "sys.stdin.read()\n" "result = Path(sys.argv[sys.argv.index('--output-last-message') + 1])\n" "context = json.loads((result.parent / 'context.json').read_text())\n" f"with open({str(counter)!r}, 'a') as stream: stream.write(json.dumps(context) + '\\n')\n" diff --git a/tests/backend/test_codex_von_worker.py b/tests/backend/test_codex_von_worker.py index 824271f0..23bac21a 100644 --- a/tests/backend/test_codex_von_worker.py +++ b/tests/backend/test_codex_von_worker.py @@ -434,6 +434,7 @@ def finish(state): def test_interrupted_clone_recovers_with_independent_metadata_and_no_service_secrets( config, tmp_path, monkeypatch ): + monkeypatch.setattr(worker.Von, "start_execution", lambda *a, **kw: None) source = tmp_path / "source" run = worker.command run(["git", "init", "-b", "main", str(source)]) diff --git a/tests/backend/test_task_execution_timing.py b/tests/backend/test_task_execution_timing.py new file mode 100644 index 00000000..f7ea330c --- /dev/null +++ b/tests/backend/test_task_execution_timing.py @@ -0,0 +1,346 @@ +"""Canonical fixture reads and both existing coding-controller boundaries. + +The optional Mac replay loads only function definitions from the operator's +source, patches a temporary copy, and substitutes all host effects. It does not +import bootstrap/configuration, access credentials, or touch the installed bridge. +""" + +import ast +import json +import os +import subprocess +import uuid +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from test_codex_von_retry import canonical_controller, config # noqa: F401 + +from scripts import codex_von_retry as retry +from scripts import codex_von_worker as worker +from src.backend.services import task_execution_timing_service as timing + +# Imported pytest fixtures are intentionally named in test signatures. +# ruff: noqa: F811 + +START = "2026-09-17T10:00:00+00:00" +END = "2026-09-17T10:05:00+00:00" + + +def start(api, task_id, attempt="first", started_at=START): + return timing.start( + task_id, + attempt_id=attempt, + actor_concept_id=api.config["agent_id"], + started_at=started_at, + source="fixture", + ) + + +def finish(api, task_id, attempt="first", outcome="completed", ended_at=END): + return timing.finish( + task_id, + attempt_id=attempt, + actor_concept_id=api.config["agent_id"], + outcome=outcome, + ended_at=ended_at, + ) + + +def test_canonical_start_finish_retry_reopen_and_detail( + canonical_controller, monkeypatch +): + _, api, create, *_ = canonical_controller + task_id = create() + assert api.task(task_id)["execution_timing"]["started_at"] is None + start(api, task_id) + start(api, task_id, started_at="2026-09-17T11:00:00Z") + assert api.task(task_id)["execution_timing"]["started_at"] == START + blocked = finish(api, task_id, outcome="blocked")["execution_timing"] + assert blocked["completed_at"] is None + assert blocked["attempts"][0]["elapsed_wall_seconds"] == 300 + assert blocked["completion_history"] == [] + assert finish(api, task_id, outcome="blocked")["execution_timing"] == blocked + start(api, task_id, "second", "2026-09-17T10:06:00Z") + completed = finish(api, task_id, "second", ended_at="2026-09-17T10:08:00Z")[ + "execution_timing" + ] + assert completed["started_at"] == START + assert completed["completed_at"] + assert completed["attempts"][1]["elapsed_wall_seconds"] == 120 + assert completed["cost"]["amount"] is None + assert completed["cost"]["status"] == "unknown" + assert ( + finish(api, task_id, "second", ended_at="2026-09-17T10:08:00Z")[ + "execution_timing" + ] + == completed + ) + # Existing HTTP detail route projects the same persisted data. + from flask import Flask + + from src.backend.server.routes import task_routes + + monkeypatch.setattr( + task_routes, "list_task_external_resource_actions", lambda _: [] + ) + # Visibility sanitisation retains only resolvable type relationships. + for concept_id in [ + api.tasks.TASK_SPECIFICATION_TYPE_ID, + *api.task(task_id)["task_type_ids"], + ]: + api.tasks.ConceptsRepository.update_one( + {"concept_id": concept_id}, + {"$setOnInsert": {"concept_id": concept_id, "relationships": {}}}, + upsert=True, + ) + app = Flask(__name__) + app.secret_key = "isolated-timing-fixture" + app.register_blueprint(task_routes.task_bp, url_prefix="/api/tasks") + client = app.test_client() + with client.session_transaction() as session: + session["user_concept_id"] = api.config["delegator_id"] + session["organisation_concept_id"] = api.config["organisation_id"] + from src.backend.security.access_control import override_current_actor + + with override_current_actor( + api.config["delegator_id"], api.config["organisation_id"] + ): + response = client.get("/api/tasks/" + task_id.replace("#", "%23")) + assert response.status_code == 200, response.get_json() + assert response.get_json()["execution_timing"] == completed + api.tasks.update_task_status(task_id, "pending") + reopened = api.task(task_id)["execution_timing"] + assert reopened["completed_at"] is None + assert reopened["completion_history"] == completed["completion_history"] + finish(api, task_id, "second", ended_at="2026-09-17T10:08:00Z") + assert api.task(task_id)["status"] == "pending" + start(api, task_id, "third", "2026-09-17T10:09:00Z") + final = finish(api, task_id, "third", ended_at="2026-09-17T10:10:00Z")[ + "execution_timing" + ] + assert len(final["completion_history"]) == 2 + assert len(final["attempts"]) == 3 + assert final["started_at"] == START + + +def test_interrupted_finish_reconciles_and_conflicting_receipt_is_rejected( + canonical_controller, monkeypatch +): + _, api, create, *_ = canonical_controller + task_id = create() + start(api, task_id) + original = api.tasks.update_task_status + + def lost_ack(*a, **kw): + original(*a, **kw) + raise OSError("Lost acknowledgement after status persistence") + + monkeypatch.setattr(api.tasks, "update_task_status", lost_ack) + with pytest.raises(OSError): + finish(api, task_id) + assert api.task(task_id)["execution_timing"]["completed_at"] is None + monkeypatch.setattr(api.tasks, "update_task_status", original) + accepted = finish(api, task_id)["execution_timing"] + assert accepted["completed_at"] + assert finish(api, task_id)["execution_timing"] == accepted + with pytest.raises(ValueError, match="different terminal"): + finish(api, task_id, outcome="blocked") + + +def test_interrupted_finish_cannot_complete_reopened_task( + canonical_controller, monkeypatch +): + _, api, create, *_ = canonical_controller + task_id = create() + start(api, task_id) + original = api.tasks.update_task_status + + def lost_ack(*a, **kw): + original(*a, **kw) + raise OSError("Lost acknowledgement") + + monkeypatch.setattr(api.tasks, "update_task_status", lost_ack) + with pytest.raises(OSError): + finish(api, task_id) + monkeypatch.setattr(api.tasks, "update_task_status", original) + original(task_id, "pending") + with pytest.raises(ValueError, match="reopened"): + finish(api, task_id) + assert api.task(task_id)["status"] == "pending" + + +def test_missing_exit_and_invalid_observations_stay_honest(canonical_controller): + _, api, create, *_ = canonical_controller + task_id = create() + with pytest.raises(ValueError, match="offset"): + start(api, task_id, started_at="2026-09-17T10:00:00") + assert api.task(task_id)["execution_timing"]["attempts"] == [] + start(api, task_id) + with pytest.raises(ValueError, match="precedes"): + finish(api, task_id, ended_at="2026-09-16T10:00:00Z") + unknown = finish(api, task_id, outcome="failed", ended_at=None)["execution_timing"] + assert unknown["completed_at"] is None + assert unknown["attempts"][0]["ended_at"] is None + assert unknown["attempts"][0]["elapsed_wall_seconds"] is None + + +def test_dgx_supported_launch_and_result_readback(canonical_controller, monkeypatch): + cfg, api, create, read_state, launches, fd = canonical_controller + runner = Path(cfg["codex_command"]) + runner.write_text( + runner.read_text().replace("'status': 'blocked'", "'status': 'completed'") + ) + task_id = create() + repair_observations = [] + + def observe_repair(task, result): + assert api.task(task_id)["status"] == "in_progress" + repair_observations.append(task["status"]) + + monkeypatch.setattr(api, "record_supervisor_repair", observe_repair) + worker.tick(cfg, api, fd) + assert repair_observations == ["in_progress"] + _, state = read_state(task_id) + result = api.task(task_id)["execution_timing"] + assert result["started_at"] == state["execution_started_at"] + assert result["attempts"][0]["ended_at"] == state["finished_at"] + assert result["attempts"][0]["source"] == "codex_dgx" + assert result["completed_at"] + assert result["attempts"][0]["elapsed_wall_seconds"] >= 0 + assert state["execution_timing"] == result + worker.write_json( + Path(cfg["state_root"]) / "timing-readback.json", + { + "environment": "isolated canonical-service fixture; no live task or provider", + "task_id": task_id, + "execution_timing": result, + }, + ) + message = api.messages.get_message_for_user( + state["message_id"], cfg["delegator_id"] + ) + content = api.messages.project_direct_message(message)["content"] + assert result["completed_at"] in content + assert "includes waits" in content + worker.tick(cfg, api, fd) + assert len(launches()) == 1 + assert api.task(task_id)["execution_timing"] == result + + +@pytest.mark.parametrize("revise_before_acceptance", [False, True]) +def test_mac_operator_patch_supported_path( + canonical_controller, tmp_path, monkeypatch, revise_before_acceptance +): + source = os.environ.get("CODING_TIMING_MAC_BRIDGE_SOURCE") + if not source: + pytest.skip( + "Operator-owned bridge source is needed for the isolated Mac replay" + ) + cfg, api, create, *_ = canonical_controller + copy = tmp_path / "task-server.py" + copy.write_text(Path(source).read_text()) + patch = ( + Path(__file__).resolve().parents[2] + / "docs/engineering/patches/codex_vscode_task_timing.patch" + ) + subprocess.run( + ["patch", "--batch", "--fuzz=0", str(copy), str(patch)], + check=True, + capture_output=True, + ) + parsed = ast.parse(copy.read_text()) + functions = [ + node + for node in parsed.body + if isinstance(node, ast.FunctionDef) and node.name not in {"main", "report"} + ] + messages = {} + + def report(key, subject, content, task_id): + assert key not in messages or messages[key] == content + messages[key] = content + return {"canonical_readback": True} + + root = tmp_path / "bridge-state" + root.mkdir() + namespace = dict( # noqa: C408 - the execution namespace mirrors module globals + ROOT=root, + CONFIG=cfg, + LIVE={"reserved", "running", "reporting"}, + retry_policy=retry, + json=json, + uuid=uuid, + os=os, + datetime=datetime, + timezone=timezone, + report=report, + ) + exec( # noqa: S102 - only reviewed operator function definitions in an isolated fixture + compile(ast.Module(body=functions, type_ignores=[]), str(copy), "exec"), + namespace, + ) + dispatch = namespace["dispatch"] + task_id = create() + selected = dispatch({"action": "select"}, api, worker) + assert selected["task_id"] == task_id + assert api.task(task_id)["execution_timing"]["started_at"] is None + request = {"attempt": selected["attempt"], "thread_id": "fixture-thread"} + begun = dispatch(dict(request, action="begin"), api, worker) + assert begun["may_execute"] + start_time = api.task(task_id)["execution_timing"]["started_at"] + resumed = dispatch( + dict(request, action="resume", checkpoint="Retained effects reconciled"), + api, + worker, + ) + assert resumed["may_execute"] + assert api.task(task_id)["execution_timing"]["started_at"] == start_time + result = dict( # noqa: C408 + status="completed", + summary="Fixture complete", + evidence="Canonical read-back", + question="", + deploy_commit="", + ) + if revise_before_acceptance: + original = api.finish_execution + + def unavailable(*args, **kwargs): + raise OSError("Canonical acceptance unavailable") + + monkeypatch.setattr(api, "finish_execution", unavailable) + with pytest.raises(OSError): + dispatch(dict(request, action="finish", result=result), api, worker) + api.tasks.update_task_fields( + task_id, fields={"notes": "Revised task constraint"} + ) + monkeypatch.setattr(api, "finish_execution", original) + dispatch(dict(request, action="finish", result=result), api, worker) + assert api.task(task_id)["status"] == "in_progress" + assert api.task(task_id)["execution_timing"]["completed_at"] is None + return + finished = dispatch(dict(request, action="finish", result=result), api, worker) + assert finished["phase"] == "done" + receipt = api.task(task_id)["execution_timing"] + assert receipt["completed_at"] + assert receipt["attempts"][0]["source"] == "codex_vscode" + assert ( + receipt["completed_at"] + in messages["vscode-task:" + request["attempt"] + ":result"] + ) + repeated = dispatch(dict(request, action="finish", result=result), api, worker) + assert repeated == finished + assert api.task(task_id)["execution_timing"] == receipt + + +def test_external_completion_is_not_attributed_to_unfinished_attempt( + canonical_controller, +): + _, api, create, *_ = canonical_controller + task_id = create() + start(api, task_id) + api.tasks.update_task_status(task_id, "completed") + with pytest.raises(ValueError, match="became terminal"): + finish(api, task_id) + assert api.task(task_id)["execution_timing"]["completed_at"] is None diff --git a/tests/backend/test_task_run_archives.py b/tests/backend/test_task_run_archives.py index c9676914..dc5c970e 100644 --- a/tests/backend/test_task_run_archives.py +++ b/tests/backend/test_task_run_archives.py @@ -292,6 +292,8 @@ def test_hash_readback_detects_storage_corruption(fixture): def test_exact_worker_launch_publishes_before_exit_and_downloads_after( fixture, tmp_path, monkeypatch ): + # Task timing persistence is exercised against canonical services separately. + monkeypatch.setattr(worker.Von, "start_execution", lambda *a, **kw: None) import sys from src.backend.server.routes import task_routes as routes