From 5db44213f466de986faa29fcbf6e362ba049a0ae Mon Sep 17 00:00:00 2001 From: witbrock Date: Sun, 13 Sep 2026 17:53:00 -0400 Subject: [PATCH] Retain coding blockers until an authorised retry or verified repair --- docs/engineering/codex_dgx_worker.md | 100 +++- scripts/codex_von_inbox.py | 57 ++- scripts/codex_von_inbox_prompt.md | 9 + scripts/codex_von_release.py | 9 + scripts/codex_von_retry.py | 217 +++++++++ scripts/codex_von_worker.py | 204 ++++++++- scripts/codex_von_worker_prompt.md | 13 + tests/backend/test_codex_von_retry.py | 585 ++++++++++++++++++++++++ tests/backend/test_task_run_archives.py | 2 + tests/test_codex_von_release.py | 23 + 10 files changed, 1203 insertions(+), 16 deletions(-) create mode 100644 scripts/codex_von_retry.py create mode 100644 tests/backend/test_codex_von_retry.py diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md index 0d7a8489..b748bfe1 100644 --- a/docs/engineering/codex_dgx_worker.md +++ b/docs/engineering/codex_dgx_worker.md @@ -165,7 +165,7 @@ separate. The launcher should unset `OPENAI_API_KEY` and `CODEX_API_KEY`, set the dedicated `CODEX_HOME`, and execute the installed official Codex CLI. Validate an actual agent-issued shell command before enabling polling. -Install the worker, `codex_von_inbox.py`, and both adjacent `_prompt.md` files +Install the worker, `codex_von_inbox.py`, `codex_von_retry.py`, and both adjacent `_prompt.md` files from the same reviewed revision. Run them with the project's PDM-managed Python environment. When these scripts are copied into a release bundle outside a complete checkout, pass `--backend-root /path/to/reviewed/Von` to the worker. This selects the @@ -178,6 +178,104 @@ point and list reads. The adapter requires `requested_model` and `requested_reasoning_effort` keys even when their values are null. An older reader that omits them fails visibly before pickup instead of silently choosing defaults. Installing only a newer worker script cannot repair an older canonical reader. + +### Waiting for a coding dependency + +The retry policy in `scripts/codex_von_retry.py` separates task context from +permission to spend another coding attempt. After `blocked` or `needs_input`, +the existing task state retains the result, failed attempt, recovery owner and +expectation. Archive attachments, evidence/checkpoint updates, self-reports, +unrelated comments, CI, a pending status and elapsed time cannot admit a retry. +The task checkpoint and one idempotent message explain the wait. Other eligible +tasks continue, including when one retained report cannot be delivered. + +A coding result may provide an optional `blocker` with `key`, `kind` +(`dependency` or `transient`), `owner`, `recovery`, `scope_json`, +`required_observations_json` and `probe_key` (usually null). The two JSON strings +hold objects describing the relevant candidate/profile and exact observations, +for example `{"authenticated":true,"profile_bound":true}`. The controller binds +the task, coding actor, organisation, failed attempt and observation time. This +describes an external prerequisite; code the assigned agent can safely repair +and independently useful work should be handled before returning a blocker. + +There are three supported continuation routes: + +- The delegator can write a task comment beginning exactly + `/retry-coding `. This is an explicit one-attempt + retry after review, not a claim that the dependency was repaired. Quoted + commands and other authors' comments do not qualify. +- For natural-language repair, answers or changed scope, use the existing inbox + reply route. Its bounded interpretation receives the retained result and + supplied receipt bytes. It must distinguish relevant recovery, explicit retry + and independently useful scope from a status report. The controller binds a + `resume_task` decision to that failed attempt and source message. This path + uses one ordinary inbox interpretation per new message; polling does not + repeatedly ask a model to reinterpret an unchanged failure. +- A delegator-authored canonical task comment may carry `source.coding_retry`, + an operator-verified repair receipt. Required fields are `attempt`, + `blocker_key`, `binding` (`task_id`, `agent_id`, `organisation_id`), `scope` + (matching the blocker's JSON object), `observed_at`, `observations`, + `evidence_reference` and `reason`. Observations must include every required + value with its exact JSON type, and be observed after the failed attempt. + Use the linked file-copy concept and checksum as the evidence reference when + applicable. The trusted producer must actually inspect/read back the repair; + a filename, receipt metadata or a top-level ready flag is insufficient. The + comparator does not discover or execute arbitrary evidence files. It rejects + wrong-task/actor/org, wrong-revision, stale, future and contradictory receipts. + +Older text-only results are retained as **unclassified**, not silently inferred +to have machine-checkable recovery conditions. They can use explicit retry or +the contextual inbox route. Ordinary free-text task comments remain context; +they do not automatically constitute repair. The visible wait supplies the +supported continuation route, so a repair outside Git or with different wording +does not require changing a classifier or a filename convention. Missing local +state for a canonical blocked/in-progress task requires reconciliation instead +of assuming earlier effects never happened. + +For known transient dependencies, the operator can register `retry_probes` in +private controller configuration, keyed by `probe_key`. Each entry contains an +`argv` list for an existing cheap read-only helper, `timeout_seconds`, +`interval_seconds` and `max_interval_seconds`. Choose bounds from that helper's +declared/observed duration with headroom; the subprocess timeout protects the +controller queue's liveness, independently of coding-run duration. The helper +receives the retained blocker JSON on stdin and returns the repair receipt +above on stdout. No task/model-supplied command, URL or executable is run. Failed +or unknown probes back off exponentially to the configured maximum, with the +next check persisted before execution; they never launch a coding model. No +probe is registered by default. Unknown coding-process outcomes always retain +their work/effects for reconciliation, rather than treating them as a cheap +transient failure. + +The source token is consumed in the same durable write that marks the new +attempt running, before the subprocess starts. Restart first reconciles an +existing run/report/archive. A replayed inbox acknowledgement cannot bind the +old message to a newer failure or reopen a task whose retry already ran. Task +cancellation, reassignment, project writer and current authority are rechecked +before the admitted continuation. Model preferences still come from the +canonical task. + +### DGX and Mac retry-policy handoff + +Source merge and installation are distinct. Use the existing coherent release +procedure below at an idle controller boundary; retain the previous release, +state, worktrees, schedule, lock and configured model preferences. Do not run a +second worker to activate or test the upgrade. `--check-task` and the next +scheduled runtime receipt must identify the selected module paths/revision. +This change does not request a public web deployment. + +The Mac bridge is operator-owned and is not source-controlled in this repository. +Its operator must import the same `codex_von_retry` module, map its retained +attempt/result and canonical inputs to `retain_blocker`/`admission`, and persist +`consume` with the running checkpoint before its existing launch call under its +existing lock. Bind config identity locally; never trust a comment's claimed +author or scope instead of the canonical author and current assignment. Keep +the same inbox source authorisation, repair producer verification, current-task +recheck, cheap-probe backoff and result/effect reconciliation boundaries as the +DGX adapter. Run the shared tests plus an isolated replay of that installed +bridge and record canonical wait/recovery read-back and zero duplicate launches. +An import test or a DGX replay alone is not evidence of Mac integration or +activation. Return the Mac adapter diff and both hosts' installation/runtime +receipts through the coordinating operator before claiming both are active. Do not change global model defaults to compensate for a mismatched installation. For an authorised upgrade, use the coherent installation below. Preserve the diff --git a/scripts/codex_von_inbox.py b/scripts/codex_von_inbox.py index e4fa644a..fbbb5810 100644 --- a/scripts/codex_von_inbox.py +++ b/scripts/codex_von_inbox.py @@ -13,8 +13,9 @@ authorised_task, capability_context, fingerprint, - resolve_execution_settings, referenced_tasks, + resolve_execution_settings, + retry_policy, stage_file_copy_evidence, write_json, ) @@ -23,8 +24,9 @@ authorised_task, capability_context, fingerprint, - resolve_execution_settings, referenced_tasks, + resolve_execution_settings, + retry_policy, stage_file_copy_evidence, write_json, ) @@ -139,6 +141,20 @@ def context_for(config, api, message): tasks = [item["task"] for item in lookup["results"] if item["status"] == "found"] for item in lookup["results"]: if item["status"] == "found": + retained = retained_task_state(config, item["task_id"]) + if retained: + item["retained_attempt"] = { + key: retained.get(key) + for key in ( + "attempt", + "phase", + "result", + "blocker", + "worktree", + "finished_at", + "execution_settings", + ) + } try: item["assignment_context"] = api.inputs(item["task"]) except Exception as exc: @@ -163,6 +179,11 @@ def context_for(config, api, message): } +def retained_task_state(config, task_id): + path = Path(config["state_root"]) / "tasks" / (fingerprint(task_id)[:16] + ".json") + return json.loads(path.read_text()) if path.exists() else {} + + def validate_result(value, context): fields = set(SCHEMA["required"]) # Retained pre-upgrade runs have four fields. Keep them recoverable. @@ -323,7 +344,11 @@ def prepare_attachment_inputs(config, context, run): inputs.append(item) context["attachment_inputs"] = inputs if len(references) > 8: - inputs.append({"error": "Only the first eight attachments were prepared in this run; remaining references have not been inspected."}) + inputs.append( + { + "error": "Only the first eight attachments were prepared in this run; remaining references have not been inspected." + } + ) def launch(config, state, lock_fd): @@ -496,11 +521,29 @@ def finish(config, api, state, path): if task_id: if result["action"] == "resume_task": state["effect_stage"] = "task_resume" + retained = retained_task_state(config, task_id) + blocker = ( + retry_policy.retain_blocker(config, retained) if retained else None + ) + # Freeze before reopen/delivery; a lost acknowledgement must never + # bind the old request to a newer failed attempt on the next poll. + if "retry_intent" not in state: + state["retry_intent"] = ( + { + "attempt": blocker["attempt"], + "binding": blocker["binding"], + "reason": result["answer"], + } + if blocker + else None + ) + write_json(path, state) # This exact follow-up is the new coding/deployment authority carrier. followup = { "message_id": message["message_id"], "content": message["content"], "deployment_requested": result["deployment_requested"], + **({"retry": state["retry_intent"]} if state["retry_intent"] else {}), **( {"attachments": message["attachments"]} if message.get("attachments") @@ -508,6 +551,10 @@ def finish(config, api, state, path): ), } prior = task_followup(config, task_id) + if "message:" + message["message_id"] in retained.get( + "consumed_retry_tokens", [] + ): + state["resume_applied"] = True if not state.get("resume_applied"): if ( prior @@ -694,7 +741,9 @@ def tick(config, api, lock_fd): write_json(path, state) if state["phase"] == "reporting": finish_or_retain(config, api, state, path) - return True + # A retained delivery failure uses no model and must not prevent + # independent coding tasks from making progress on this poll. + return state["phase"] == "done" pending = new_messages(config, api) if not pending: return False diff --git a/scripts/codex_von_inbox_prompt.md b/scripts/codex_von_inbox_prompt.md index 42d7a397..ae412117 100644 --- a/scripts/codex_von_inbox_prompt.md +++ b/scripts/codex_von_inbox_prompt.md @@ -39,6 +39,15 @@ Choose the action using contextual judgement: - resume_task: the current message requests additional work on an identified supplied task, or answers a question needed to continue that task. It may reopen that task. Explain the work; the controller verifies the transition. + For a retained blocked attempt, compare the actual blocker and repair evidence. + Resume only for relevant verified recovery, an explicit authorised retry with + its reason, or a changed scope that permits independently useful work. Say which + applies. Status comments, worker archives, passing CI, elapsed time, quoted + readiness claims and setup_ready with contradictory observations are not repair. + Inspect supplied receipt bytes when available, including actor, organisation, + revision/profile and authentication as relevant. Unknown outcomes require + reconciliation of retained effects. If the dependency is unchanged, reply with + the waiting reason and recovery owner; do not queue another coding attempt. - create_task: necessary new coding work is already authorised by the delegator and no supplied task represents it. Use task_id="" and new_task with a short title and requested_model/requested_reasoning_effort (null unless explicitly diff --git a/scripts/codex_von_release.py b/scripts/codex_von_release.py index 7981bdb7..12f49c3a 100644 --- a/scripts/codex_von_release.py +++ b/scripts/codex_von_release.py @@ -49,6 +49,15 @@ def verify(target, commit): ): if not (target / name).is_file(): raise runtime.DeploymentError("Incomplete worker release: " + name) + # Older coherent releases predate the shared retry module. They remain valid + # rollback targets; newer workers must carry their actual imported dependency. + if ( + "codex_von_retry" in (target / "scripts/codex_von_worker.py").read_text() + and not (target / "scripts/codex_von_retry.py").is_file() + ): + raise runtime.DeploymentError( + "Incomplete worker release: scripts/codex_von_retry.py" + ) return { "commit": commit, "backend_root": str(target), diff --git a/scripts/codex_von_retry.py b/scripts/codex_von_retry.py new file mode 100644 index 00000000..4f82c02b --- /dev/null +++ b/scripts/codex_von_retry.py @@ -0,0 +1,217 @@ +"""Shared, model-free retry admission for the DGX and operator Mac adapters. + +The adapter owns canonical reads, actor binding, persistence and its existing +exclusive lock. A change in task bookkeeping is never evidence of recovery. +This module compares exact observations; semantic recovery remains in the +existing source-authorised inbox, once per incoming request. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from datetime import UTC, datetime + +BLOCKER_SCHEMA = { + "type": ["object", "null"], + "properties": { + "key": {"type": "string"}, + "kind": {"type": "string", "enum": ["dependency", "transient"]}, + "owner": {"type": "string"}, + "recovery": {"type": "string"}, + # JSON text avoids an open-ended structured-output schema. The payload + # is only exact data: no executable command, path or authority grant. + "scope_json": {"type": "string"}, + "required_observations_json": {"type": "string"}, + "probe_key": {"type": ["string", "null"]}, + }, + "required": [ + "key", + "kind", + "owner", + "recovery", + "scope_json", + "required_observations_json", + "probe_key", + ], + "additionalProperties": False, +} + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest() + + +def timestamp(value): + try: + parsed = datetime.fromisoformat(str(value)) + return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed + except (ValueError, TypeError): + return None + + +def validate_blocker(value): + if value is None: + return None + if not isinstance(value, dict) or set(value) != set(BLOCKER_SCHEMA["required"]): + raise ValueError("Invalid blocker fields") + for key in set(value) - {"probe_key"}: + if not isinstance(value[key], str) or not value[key].strip(): + raise ValueError("Blocker fields must be non-empty text") + if value["kind"] not in {"dependency", "transient"} or not ( + value["probe_key"] is None or isinstance(value["probe_key"], str) + ): + raise ValueError("Invalid blocker kind or probe key") + for key in ("scope_json", "required_observations_json"): + parsed = json.loads(value[key]) + if not isinstance(parsed, dict): + raise ValueError( # noqa: TRY004 - invalid terminal result + "Blocker observations and scope must be JSON objects" + ) + return value + + +def retain_blocker(config, state): + """Incrementally retain old text failures without pretending to classify them.""" + result = state.get("result") or {} + if result.get("status") not in {"blocked", "needs_input"}: + return None + existing = state.get("blocker") + if existing and existing.get("attempt") == state.get("attempt"): + return existing + supplied = validate_blocker(result.get("blocker")) or {} + blocker = { + "attempt": state.get("attempt", "legacy"), + "key": supplied.get("key") + or digest([result.get("summary"), result.get("question")]), + "kind": supplied.get("kind", "unclassified"), + "reason": result.get("summary", "Prior attempt is blocked"), + "owner": supplied.get("owner", "delegator / operator triage"), + "recovery": supplied.get("recovery") + or result.get("question") + or "Inspect the retained result and supply relevant repair evidence or an explicit retry reason.", + "observed_at": state.get("finished_at") or state.get("started_at"), + "scope": json.loads(supplied.get("scope_json", "{}")), + "required_observations": json.loads( + supplied.get("required_observations_json", "{}") + ), + "probe_key": supplied.get("probe_key"), + "binding": { + "task_id": state["task_id"], + "agent_id": config["agent_id"], + "organisation_id": config["organisation_id"], + }, + } + state["blocker"] = blocker + return blocker + + +def waiting_text(blocker): + return ( + f"Waiting without another coding launch: {blocker['reason']}\n" + f"Last attempt: {blocker['attempt']} ({blocker.get('observed_at') or 'time unavailable'}).\n" + f"Recovery owner: {blocker['owner']}. Required: {blocker['recovery']}\n" + f"Retry after review with a task comment: /retry-coding {blocker['attempt']} \n" + + ( + f"Next cheap check: configured probe {blocker['probe_key']} (with backoff)." + if blocker.get("kind") == "transient" and blocker.get("probe_key") + else "Next check: canonical repair/retry input on the normal poll; no timed coding retry." + ) + ) + + +def verified_repair(blocker, receipt, now): + """A trusted producer must attest exact, fresh, attempt-bound observations. + + No recursive search for a 'ready' boolean: required observations (including + authentication when relevant) must all match, even if setup_ready is true. + A receipt can link a private file copy without this module opening it. + """ + if not isinstance(receipt, dict): + return False + observed = timestamp(receipt.get("observed_at")) + blocked_at = timestamp(blocker.get("observed_at")) + required = blocker["required_observations"] + observations = receipt.get("observations") + return bool( + required + and receipt.get("attempt") == blocker["attempt"] + and receipt.get("blocker_key") == blocker["key"] + and receipt.get("binding") == blocker["binding"] + and receipt.get("scope") == blocker["scope"] + and receipt.get("evidence_reference") + and receipt.get("reason") + and observed + and blocked_at + and blocked_at < observed <= now + and isinstance(observations, dict) + and all( + key in observations + and json.dumps(observations[key], sort_keys=True) + == json.dumps(value, sort_keys=True) + for key, value in required.items() + ) + ) + + +def admission(config, state, inputs, *, now=None): + """Return one recoverable admission, or None. Never infer repair from edits. + + `followup.retry` is written only by the source-authorised inbox adapter. + Comments are canonical task-service rows; source metadata alone is not + trusted authorship. The adapter must recheck task authority before launch. + """ + blocker = retain_blocker(config, state) + if not blocker: + return {"kind": "initial"} + now = now or datetime.now(UTC) + consumed = state.get("consumed_retry_tokens", []) + followup = inputs.get("followup") or {} + retry = followup.get("retry") or {} + if ( + retry.get("attempt") == blocker["attempt"] + and retry.get("binding") == blocker["binding"] + and retry.get("reason") + and followup.get("message_id") + ): + token = "message:" + followup["message_id"] + if token not in consumed: + return { + "kind": "authorised_followup", + "token": token, + "reason": retry["reason"], + } + for row in inputs.get("comments", []): + if row.get("author_concept_id") != config["delegator_id"]: + continue + comment_id = row.get("comment_id") + token = "comment:" + str(comment_id) + if not comment_id or token in consumed: + continue + # A deliberate command in the owner's own comment, never quoted text + # or an arbitrary occurrence of 'retry' in a status report. + match = re.fullmatch( + r"/retry-coding ([^\s]+) ([\s\S]+)", row.get("body", "").strip() + ) + if match and match[1] == blocker["attempt"] and match[2].strip(): + return {"kind": "manual_retry", "token": token, "reason": match[2]} + receipt = (row.get("source") or {}).get("coding_retry") + if verified_repair(blocker, receipt, now): + return { + "kind": "verified_repair", + "token": token, + "reason": receipt["reason"], + "receipt": receipt, + } + return None + + +def consume(state, decision): + """Persist with the new running attempt, BEFORE any coding subprocess.""" + token = decision.get("token") + if token: + state.setdefault("consumed_retry_tokens", []).append(token) + state["launch_admission"] = decision + state.pop("blocker", None) + state.pop("waiting_report", None) diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index 9ebdf8c8..232e8938 100755 --- a/scripts/codex_von_worker.py +++ b/scripts/codex_von_worker.py @@ -23,6 +23,14 @@ from datetime import UTC, datetime from pathlib import Path +try: + from . import codex_von_retry as retry_policy +except ImportError: + try: + import codex_von_retry as retry_policy + except ImportError: + from scripts import codex_von_retry as retry_policy + ACTIVITY_CHECKPOINT_SECONDS = 15 ACTIVE = {"pending", "in_progress", "blocked"} @@ -34,6 +42,7 @@ "evidence": {"type": "string"}, "question": {"type": "string"}, "deploy_commit": {"type": "string"}, + "blocker": retry_policy.BLOCKER_SCHEMA, }, "required": ["status", "summary", "evidence", "question", "deploy_commit"], "additionalProperties": False, @@ -83,6 +92,7 @@ def capability_context(config, *, read_only, worktree=None): "von_mcp": "disabled; canonical reads/effects belong to the controller", "controller_actions": ["reply", "resume_task", "create_task"], "deployment_enabled": bool(config.get("deployment_command")), + "retry_probe_keys": sorted((config.get("retry_probes") or {}).keys()), "limitations": [ "Use available shell tools for non-secret host/repository facts; missing supplied facts do not mean inspection is unavailable.", "Do not access credentials, private databases or live-service mutation routes.", @@ -433,13 +443,14 @@ def check_task(config, api, task_id): def validate_result(value): fields = set(RESULT_SCHEMA["required"]) - if not isinstance(value, dict) or set(value) not in ( + if not isinstance(value, dict) or set(value) - {"blocker"} not in ( fields, fields - {"deploy_commit"}, ): raise ValueError("Codex did not return the required result fields") - if any(not isinstance(v, str) for v in value.values()): + if any(not isinstance(v, str) for k, v in value.items() if k != "blocker"): raise ValueError("Codex result fields must be text") + retry_policy.validate_blocker(value.get("blocker")) if value["status"] not in RESULT_SCHEMA["properties"]["status"]["enum"]: raise ValueError("Unknown Codex outcome") if not value["summary"].strip() or ( @@ -733,6 +744,7 @@ def finish(self, state): "title": state.get("title", state["task_id"]), } result = state["result"] + blocker = retry_policy.retain_blocker(self.config, state) scope_valid = ( authorised_task(task, self.config) and task.get("status") in ACTIVE @@ -744,6 +756,8 @@ def finish(self, state): f"DGX worktree: {state.get('worktree', 'not started')}\n" f"Run: {state['attempt']}" ).strip() + if blocker: + content += "\n\n" + retry_policy.waiting_text(blocker) if state.get("execution_settings"): settings = state["execution_settings"] content += ( @@ -785,8 +799,11 @@ def finish(self, state): report_fields = { "progress_signal": result["summary"], "evidence": evidence, - "next_checkpoint": result["question"] - or "Result available in Von messages.", + "next_checkpoint": ( + retry_policy.waiting_text(blocker) + if blocker + else result["question"] or "Result available in Von messages." + ), } self.tasks.update_task_fields( state["task_id"], @@ -799,6 +816,8 @@ def finish(self, state): ) if self.task(state["task_id"])["status"] != status: raise RuntimeError("Task status canonical read-back failed") + if blocker: + state["waiting_report"] = {"delivered": True, "message_id": message_id} # Reporting changes evidence/checkpoint fields now supplied as context. # Do not mistake this worker's own report for a new instruction. # Retain the launch snapshot so a concurrent user note/comment @@ -877,6 +896,7 @@ def launch(config, state, inputs, conversation, state_path, lock_fd): "assignment": inputs, "conversation": conversation, "prior_result": state.get("result"), + "retry_admission": state.get("pending_admission"), "worker_identity": config["agent_id"], "capabilities": capability_context(config, read_only=False, worktree=worktree), } @@ -901,6 +921,7 @@ def launch(config, state, inputs, conversation, state_path, lock_fd): write_json(context_path, context) prompt = Path(__file__).with_name("codex_von_worker_prompt.md").read_text() prompt += f"\nRead your assigned context from {context_path}.\n" + retry_policy.consume(state, state.pop("pending_admission", {"kind": "initial"})) state.update( phase="running", attempt=attempt, @@ -1067,7 +1088,7 @@ def recover_result(state): "status": "blocked", "summary": "The Codex run did not produce a verified terminal result.", "evidence": f"{type(exc).__name__}. Work and logs remain at {run_dir}.", - "question": "Please set this task back to pending to retry after checking the retained run.", + "question": "Reconcile the retained work and effects, then give an explicit retry reason in the task thread.", } state.update(phase="reporting", result=result) @@ -1208,6 +1229,99 @@ def backfill_run(config, api, selection): return {"task_id": task_id, "attempt": attempt, "capture": state["capture"]} +def probe_recovery(config, state, state_path): + """Run only an operator-registered cheap read probe, never model output code. + + Probe bounds protect controller/queue liveness. The operator selects these + from that probe's declared/observed duration, independently of coding time. + Saving the next check first also backs off an interrupted/unknown probe. + """ + blocker = state.get("blocker") or {} + if blocker.get("kind") != "transient": + return None + probe = (config.get("retry_probes") or {}).get(blocker.get("probe_key")) + if not probe: + return None + now = datetime.now(UTC) + next_check = retry_policy.timestamp(blocker.get("next_check_at")) + if next_check and now < next_check: + return None + from datetime import timedelta + + timeout = float(probe["timeout_seconds"]) + interval = float(probe["interval_seconds"]) + maximum = float(probe["max_interval_seconds"]) + argv = probe["argv"] + if ( + not isinstance(argv, list) + or not argv + or not all(isinstance(arg, str) for arg in argv) + or not 0 < timeout <= interval <= maximum + ): + raise ValueError("Invalid operator retry probe bounds or argv") + failures = min(blocker.get("probe_failures", 0), 16) + blocker["next_check_at"] = ( + now + timedelta(seconds=min(maximum, interval * 2**failures)) + ).isoformat() + write_json(state_path, state) + try: + completed = subprocess.run( + argv, + input=json.dumps(blocker), + text=True, + capture_output=True, + timeout=timeout, + check=True, + ) + receipt = json.loads(completed.stdout) + if retry_policy.verified_repair(blocker, receipt, datetime.now(UTC)): + return { + "kind": "probe_recovery", + "token": "probe:" + retry_policy.digest(receipt), + "reason": receipt["reason"], + "receipt": receipt, + } + blocker["probe_outcome"] = "observations_not_verified" + except (OSError, ValueError, subprocess.SubprocessError) as exc: + blocker["probe_outcome"] = type(exc).__name__ + blocker["probe_failures"] = failures + 1 + write_json(state_path, state) + return None + + +def report_waiting(config, api, task, state, path): + """One stable report per blocker, including legacy state adopted at upgrade.""" + blocker = state["blocker"] + report = state.setdefault( + "waiting_report", + { + "key": fingerprint([blocker["binding"], blocker["attempt"], blocker["key"]]) + + ":waiting", + "content": retry_policy.waiting_text(blocker), + }, + ) + if report.get("delivered"): + return + write_json(path, state) + report["message_id"] = api.send(task, report["key"], report["content"]) + api.tasks.update_task_fields( + state["task_id"], + fields={"next_checkpoint": report["content"]}, + actor_concept_id=config["agent_id"], + ) + api.tasks.update_task_status( + state["task_id"], "blocked", actor_concept_id=config["agent_id"] + ) + current = api.task(state["task_id"]) + if ( + current.get("next_checkpoint") != report["content"] + or current["status"] != "blocked" + ): + raise RuntimeError("Waiting checkpoint canonical read-back failed") + report["delivered"] = True + write_json(path, state) + + def tick(config, api, lock_fd): states_dir = Path(config["state_root"]) / "tasks" states_dir.mkdir(parents=True, exist_ok=True) @@ -1225,8 +1339,12 @@ def tick(config, api, lock_fd): recover_result(state) write_json(path, state) if state["phase"] in {"reporting", "archive_pending"}: - apply_deployment(config, api, state, path, lock_fd) - finish_captured(config, api, state, path) + try: + apply_deployment(config, api, state, path, lock_fd) + finish_captured(config, api, state, path) + except Exception as exc: # noqa: BLE001 - preserve any failed report without wedging other tasks + state["reconciliation_error"] = type(exc).__name__ + write_json(path, state) if config.get("inbox_enabled") and inbox_module().tick(config, api, lock_fd): return for task in api.pending(): @@ -1239,22 +1357,86 @@ def tick(config, api, lock_fd): if path.exists() else {"task_id": task_id, "phase": "new"} ) - if state["phase"] == "archive_pending": + if state["phase"] in {"running", "reporting", "archive_pending"}: continue state["title"] = task["title"] inputs = api.inputs(task) + # A lost controller state is not proof that earlier effects never ran. + # Adopt canonical blocked/in-progress evidence without replaying work. + if (state["phase"] == "new" and task["status"] != "pending") or ( + state["phase"] == "waiting" and not state.get("result") + ): + state.update( + phase="waiting", + attempt="unretained-" + + fingerprint( + [ + task_id, + task.get("updated_at"), + task.get("evidence"), + ] + )[:16], + result={ + "status": "blocked", + "summary": task.get("progress_signal") + or "Canonical task has prior work but its local attempt state is unavailable.", + "evidence": task.get("evidence") or "", + "question": task.get("next_checkpoint") + or "Reconcile canonical results and retained work before an explicit retry.", + }, + ) + if state["phase"] == "waiting": + decision = retry_policy.admission(config, state, inputs) + write_json(path, state) + if decision is None: + try: + decision = probe_recovery(config, state, path) + except (KeyError, TypeError, ValueError) as exc: + state["blocker"]["probe_outcome"] = type(exc).__name__ + if decision is None: + # One task's unavailable reporting must not wedge the queue. + try: + report_waiting(config, api, task, state, path) + except Exception as exc: # noqa: BLE001 - canonical reporting errors stay isolated to this task + state["waiting_report_error"] = type(exc).__name__ + write_json(path, state) + continue + state["pending_admission"] = decision if ( - state["phase"] in {"waiting", "done"} + state["phase"] == "done" and task["status"] != "pending" and state.get("input_hash") == input_fingerprint(inputs) ): continue + if state.get("pending_admission"): + # Probes and evidence reads can outlast a cancellation or reassignment. + # Re-read the supported canonical route before consuming any retry. + current = api.task(task_id) + if not ( + authorised_task(current, config) + and current.get("status") in ACTIVE + and api.native_writer(current) + ): + continue + current_inputs = api.inputs(current) + decision = state["pending_admission"] + if decision["kind"] != "probe_recovery": + checked = retry_policy.admission(config, state, current_inputs) + if not checked or checked.get("token") != decision.get("token"): + continue + task, inputs = current, current_inputs conversation = api.conversation(task) # The model may proceed from adequate task text when a transcript is unavailable. api.send( task, fingerprint([task_id, input_fingerprint(inputs)]) + ":started", - f"I am picking up this task on the DGX: {task['title']}.\nTask: {task_id}\n" + ( + f"Continuing after {state['pending_admission']['kind']}: " + f"{state['pending_admission'].get('reason', '')}\n" + if state.get("pending_admission") + else "" + ) + + f"I am picking up this task on the DGX: {task['title']}.\nTask: {task_id}\n" "I will send the result or a question here. Reply in this task thread or add a task comment.", ) api.tasks.update_task_status( @@ -1276,7 +1458,7 @@ def tick(config, api, lock_fd): ) state["result"][ "question" - ] = "Correct this task's model or reasoning setting, then requeue it." + ] = "Correct this task's model or reasoning setting, then give an explicit retry reason in the task thread." write_json(path, state) apply_deployment(config, api, state, path, lock_fd) finish_captured(config, api, state, path) diff --git a/scripts/codex_von_worker_prompt.md b/scripts/codex_von_worker_prompt.md index eadfd8c2..79199edd 100644 --- a/scripts/codex_von_worker_prompt.md +++ b/scripts/codex_von_worker_prompt.md @@ -71,6 +71,19 @@ be answered and `blocked` for a technical impediment. Include the question in plain English, or an empty string when none is needed. Keep the summary useful to Michael without requiring him to read raw logs. +For an external dependency that prevents further useful authorised work, you may +include `blocker`: key (stable dependency name), kind (dependency or transient), +owner, recovery, scope_json (JSON object of relevant revision/profile/scope), +required_observations_json (JSON object of exact observations that demonstrate +repair), and probe_key (null unless an applicable operator probe is supplied). +Use null or omit blocker otherwise. Report observed facts, not an inferred ready +flag; authentication/profile prerequisites must include their actual observations. +The controller adds the task/actor/organisation and failed-attempt binding. +Do not classify code you can safely repair as a controller dependency. Continue +independently useful work before returning a blocked result. Old text-only +failures remain supported, but need explicit retry or contextual inbox review; +unrelated edits, new archives, status changes and elapsed time will not retry them. + The supplied capabilities describe the controller and this sandbox, with known limitations; configured execution settings are not provider-observed model identity. Use available read-only host tools when relevant facts are absent. diff --git a/tests/backend/test_codex_von_retry.py b/tests/backend/test_codex_von_retry.py new file mode 100644 index 00000000..034e3973 --- /dev/null +++ b/tests/backend/test_codex_von_retry.py @@ -0,0 +1,585 @@ +"""Unchanged-blocker admission and the existing controller/canonical service path. + +All services use an isolated mongomock database. The subprocess is a tiny local +fixture, not Codex: no coding agent, provider, host bridge or live task is used. +""" + +import copy +import json +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from scripts import codex_von_inbox as inbox +from scripts import codex_von_retry as retry +from scripts import codex_von_worker as worker + + +def descriptor(**changes): + return { + "key": "browser_fixture", + "kind": "dependency", + "owner": "operator", + "recovery": "Verify the candidate profile and authenticated fixture actor.", + "scope_json": json.dumps({"candidate_sha": "a" * 40, "profile": "#V#profile"}), + "required_observations_json": json.dumps( + {"authenticated": True, "profile_bound": True} + ), + "probe_key": None, + **changes, + } + + +@pytest.fixture +def config(tmp_path): + return { + "agent_id": "#V#worker", + "delegator_id": "#V#owner", + "organisation_id": "#V#org", + "state_root": str(tmp_path / "state"), + } + + +def blocked_state(config, **changes): + state = { + "task_id": "#V#task", + "attempt": "failed-attempt", + "phase": "waiting", + "finished_at": "2026-09-13T21:12:00Z", + "result": { + "status": "blocked", + "summary": "Fixture prerequisite unavailable", + "evidence": "Observed authenticated:false", + "question": "Repair fixture", + "blocker": descriptor(), + }, + **changes, + } + retry.retain_blocker(config, state) + return state + + +def repair_receipt(state): + blocker = state["blocker"] + return { + "attempt": blocker["attempt"], + "blocker_key": blocker["key"], + "binding": copy.deepcopy(blocker["binding"]), + "scope": copy.deepcopy(blocker["scope"]), + "observed_at": "2026-09-13T21:17:00Z", + "observations": {"authenticated": True, "profile_bound": True}, + "reason": "Operator verified repaired fixture", + "evidence_reference": "#V#computer_file_copy_fixture sha256:" + "b" * 64, + } + + +def repair_comment(config, state): + return { + "comment_id": "repair-comment", + "author_concept_id": config["delegator_id"], + "body": "Verified repair", + "source": {"coding_retry": repair_receipt(state)}, + } + + +@pytest.mark.parametrize( + "change", + [ + "wrong_task", + "wrong_actor", + "wrong_org", + "wrong_attempt", + "wrong_key", + "wrong_revision", + "stale", + "future", + "auth_false", + "auth_integer", + "missing_check", + "missing_evidence", + "untrusted_author", + "worker_author", + "quoted_retry", + "plain_ready", + ], +) +def test_irrelevant_stale_and_contradictory_evidence_never_admits(config, change): + state = blocked_state(config) + comment = repair_comment(config, state) + receipt = comment["source"]["coding_retry"] + if change.startswith("wrong_"): + field = change.removeprefix("wrong_") + if field in {"task", "actor", "org"}: + receipt["binding"][ + {"task": "task_id", "actor": "agent_id", "org": "organisation_id"}[ + field + ] + ] = "wrong" + elif field == "revision": + receipt["scope"]["candidate_sha"] = "c" * 40 + else: + receipt["blocker_key" if field == "key" else field] = "wrong" + elif change in {"stale", "future"}: + receipt["observed_at"] = ( + "2026-09-13T21:00:00Z" if change == "stale" else "2099-01-01T00:00:00Z" + ) + elif change in {"auth_false", "auth_integer"}: + # The observed PR649 failure: top-level readiness contradicted by auth. + receipt["setup_ready"] = True + receipt["observations"]["authenticated"] = ( + False if change == "auth_false" else 1 + ) + elif change == "missing_check": + del receipt["observations"]["profile_bound"] + elif change == "missing_evidence": + del receipt["evidence_reference"] + elif change in {"untrusted_author", "worker_author"}: + comment["author_concept_id"] = ( + "#V#third_party" if change == "untrusted_author" else config["agent_id"] + ) + else: + comment.pop("source") + comment["body"] = ( + "> /retry-coding failed-attempt please retry" + if change == "quoted_retry" + else "Setup ready, CI passed." + ) + assert retry.admission(config, state, {"comments": [comment]}) is None + + +def test_repair_admits_once_and_cannot_apply_to_next_attempt(config): + state = blocked_state(config) + inputs = {"comments": [repair_comment(config, state)]} + decision = retry.admission(config, state, inputs) + assert decision["kind"] == "verified_repair" + retry.consume(state, decision) + assert retry.admission(config, state, inputs) is None + state["attempt"] = "next-failed-attempt" + assert retry.admission(config, state, inputs) is None + + +def test_legacy_text_preserves_reason_and_accepts_explicit_retry(config): + state = blocked_state(config) + del state["result"]["blocker"] + del state["blocker"] + assert ( + retry.admission(config, state, {"status": "pending", "evidence": "New archive"}) + is None + ) + assert state["blocker"]["kind"] == "unclassified" + assert state["blocker"]["reason"] == state["result"]["summary"] + comment = { + "comment_id": "manual", + "author_concept_id": config["delegator_id"], + "body": "/retry-coding failed-attempt Checked retained effects; continue the repaired path.", + } + decision = retry.admission(config, state, {"comments": [comment]}) + assert decision["kind"] == "manual_retry" + retry.consume(state, decision) + assert retry.admission(config, state, {"comments": [comment]}) is None + + +@pytest.fixture +def canonical_controller(config, tmp_path, monkeypatch): + monkeypatch.setenv("VON_USE_MOCK_DB", "1") + monkeypatch.setenv("VON_DB_NAME", "test_coding_retry_" + tmp_path.name) + from src.backend.services import organisation_membership_service as memberships + + monkeypatch.setattr( + memberships, + "get_organisation_members", + lambda _: { + "members": [ + {"user_concept_id": config["agent_id"]}, + {"user_concept_id": config["delegator_id"]}, + ], + }, + ) + api = worker.Von(config) + monkeypatch.setattr( + api.messages, "maybe_launch_direct_message_workflow", lambda **kw: None + ) + for name in ( + "maybe_launch_task_created_workflow", + "maybe_launch_task_status_workflow", + "maybe_launch_effort_unit_completed_workflow", + ): + monkeypatch.setattr(api.tasks, name, lambda **kw: None) + monkeypatch.setattr(api, "conversation", lambda _: {"available": False}) + monkeypatch.setattr(worker, "archive_checkpoint", lambda *a, **kw: True) + source = tmp_path / "source" + worker.command(["git", "init", "-b", "main", str(source)]) + (source / "AGENTS.md").write_text("Isolated controller fixture.\n") + worker.command(["git", "-C", str(source), "add", "AGENTS.md"]) + worker.command( + [ + "git", + "-C", + str(source), + "-c", + "user.name=Fixture", + "-c", + "user.email=fixture@example.invalid", + "commit", + "-m", + "Fixture", + ] + ) + worker.command(["git", "-C", str(source), "remote", "add", "origin", str(source)]) + counter = tmp_path / "launches.jsonl" + fake = tmp_path / "fixture-codex" + result = { + "status": "blocked", + "summary": "Fixture prerequisite unavailable", + "question": "Repair fixture", + "evidence": "Observed authenticated:false", + "deploy_commit": "", + "blocker": descriptor(), + } + fake.write_text( + f"#!{sys.executable}\n" + "import json, sys\nfrom pathlib import Path\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" + f"result.write_text(json.dumps({result!r}))\n" + ) + fake.chmod(0o700) + config.update(source_repo=str(source), codex_command=str(fake)) + + def create(title="Fixture task"): + return api.tasks.create_task( + title, + "Authorised isolated controller acceptance", + assignee_concept_id=config["agent_id"], + created_by_concept_id=config["delegator_id"], + organisation_concept_id=config["organisation_id"], + requested_model="gpt-6-astra", + requested_reasoning_effort="xhigh", + )["task_concept_id"] + + def read_state(task_id): + path = ( + Path(config["state_root"]) + / "tasks" + / (worker.fingerprint(task_id)[:16] + ".json") + ) + return path, json.loads(path.read_text()) + + def launches(): + return ( + [json.loads(line) for line in counter.read_text().splitlines()] + if counter.exists() + else [] + ) + + with (tmp_path / "controller.lock").open("w") as lock: + yield config, api, create, read_state, launches, lock.fileno() + + +def test_actual_controller_skips_bookkeeping_and_continues_other_tasks( + canonical_controller, +): + config, api, create, read_state, launches, fd = canonical_controller + task_id = create() + worker.tick(config, api, fd) + _, state = read_state(task_id) + assert api.task(task_id)["status"] == "blocked" + assert ( + "Waiting without another coding launch" in api.task(task_id)["next_checkpoint"] + ) + result_message = api.messages.get_message_for_user( + state["message_id"], config["delegator_id"] + ) + assert ( + "Recovery owner: operator" + in api.messages.project_direct_message(result_message)["content"] + ) + for i in range(4): + api.tasks.update_task_fields( + task_id, + fields={ + "evidence": f"Verified archive {i}", + "notes": f"CI and closeout {i}", + }, + ) + api.tasks.add_task_comment( + task_id, body=f"Self report {i}", author_concept_id=config["agent_id"] + ) + api.tasks.add_task_comment( + task_id, + body=f"Status note {i}; CI passed", + author_concept_id=config["delegator_id"], + ) + api.tasks.add_task_attachment( + task_id, + filename=f"archive-{i}.zip", + uri=f"https://example.invalid/archive-{i}", + ) + api.tasks.update_task_status(task_id, "pending") + worker.tick(config, api, fd) + assert len(launches()) == 1 + other = create("Other eligible task") + worker.tick(config, api, fd) + assert [row["task_id"] for row in launches()] == [task_id, other] + _, state = read_state(task_id) + receipt = repair_receipt(state) + receipt["observed_at"] = (datetime.now(UTC) + timedelta(microseconds=1)).isoformat() + api.tasks.add_task_comment( + task_id, + body="Repair verified", + author_concept_id=config["delegator_id"], + source={"coding_retry": receipt}, + ) + worker.tick(config, api, fd) + worker.tick(config, api, fd) + assert len(launches()) == 3 + resumed = launches()[-1] + assert resumed["prior_result"]["status"] == "blocked" + assert resumed["retry_admission"]["kind"] == "verified_repair" + assert resumed["execution_settings"]["reasoning_effort"] == "xhigh" + assert ( + resumed["capabilities"]["worktree"] == launches()[0]["capabilities"]["worktree"] + ) + _, resumed_state = read_state(task_id) + assert len(resumed_state["consumed_retry_tokens"]) == 1 + + +def test_lost_result_ack_reconciles_without_another_subprocess( + canonical_controller, monkeypatch +): + config, api, create, read_state, launches, fd = canonical_controller + task_id = create() + original = api.send + failed = False + + def lost_ack(task, key, content): + nonlocal failed + result = original(task, key, content) + if key.endswith(":result") and not failed: + failed = True + raise OSError("simulated lost message acknowledgement") + return result + + monkeypatch.setattr(api, "send", lost_ack) + with pytest.raises(OSError): + worker.tick(config, api, fd) + _path, state = read_state(task_id) + assert state["phase"] == "reporting" + worker.tick(config, api, fd) + worker.tick(config, api, fd) + assert len(launches()) == 1 + assert read_state(task_id)[1]["phase"] == "waiting" + messages = api.messages.get_messages_for_user( + config["delegator_id"], thread_id=task_id + ) + assert len(messages) == 2 # pickup and one canonical result + + +def test_missing_exit_reconciles_unknown_effects_without_replay(canonical_controller): + config, api, create, read_state, launches, fd = canonical_controller + task_id = create() + worker.tick(config, api, fd) + path, state = read_state(task_id) + state["phase"] = "running" + state.pop("result") + state.pop("report_content", None) + (Path(state["run_dir"]) / "exit.json").unlink() + # Use a new attempt identity to model a process lost before any result report. + state["attempt"] = "interrupted-attempt" + worker.write_json(path, state) + worker.tick(config, api, fd) + worker.tick(config, api, fd) + assert len(launches()) == 1 + assert "Reconcile" in read_state(task_id)[1]["result"]["question"] + + +def test_transient_probe_backoff_and_recovery(config, tmp_path): + state = blocked_state(config) + state["result"]["blocker"] = descriptor(kind="transient", probe_key="fixture") + state.pop("blocker") + retry.retain_blocker(config, state) + path = tmp_path / "state.json" + receipt_path = tmp_path / "probe-output.json" + calls = tmp_path / "probe-calls" + probe = tmp_path / "probe.py" + probe.write_text( + "from pathlib import Path\n" + + f"with open({str(calls)!r}, 'a') as f: f.write('check\\n')\n" + + f"print(Path({str(receipt_path)!r}).read_text())\n" + ) + config["retry_probes"] = { + "fixture": { + "argv": [sys.executable, str(probe)], + "timeout_seconds": 1, + "interval_seconds": 10, + "max_interval_seconds": 40, + } + } + bad = repair_receipt(state) + bad["observations"]["authenticated"] = False + worker.write_json(receipt_path, bad) + assert worker.probe_recovery(config, state, path) is None + assert worker.probe_recovery(config, state, path) is None + assert len(calls.read_text().splitlines()) == 1 + state["blocker"]["next_check_at"] = "2000-01-01T00:00:00Z" + worker.write_json(receipt_path, repair_receipt(state)) + decision = worker.probe_recovery(config, state, path) + assert decision["kind"] == "probe_recovery" + retry.consume(state, decision) + assert len(state["consumed_retry_tokens"]) == 1 + assert worker.probe_recovery(config, state, path) is None + + +def test_natural_language_followup_is_bound_to_retained_attempt(canonical_controller): + config, api, create, read_state, _launches, fd = canonical_controller + task_id = create() + worker.tick(config, api, fd) + _, prior = read_state(task_id) + message_doc = api.messages.create_message( + sender_id=config["delegator_id"], + recipient_ids=[config["agent_id"]], + content="The operator repaired the fixture. Continue the same task.", + org_id=config["organisation_id"], + thread_id=task_id, + ) + message = api.messages.project_direct_message(message_doc) + state = { + "phase": "reporting", + "context": {"message": message, "tasks": [api.task(task_id)]}, + "result": { + "action": "resume_task", + "task_id": task_id, + "answer": "Relevant repair verified; continue the retained work.", + "deployment_requested": False, + }, + } + path = inbox.state_path(config, message["message_id"]) + inbox.finish(config, api, state, path) + followup = inbox.task_followup(config, task_id) + assert followup["retry"]["attempt"] == prior["attempt"] + config["inbox_enabled"] = True + # Exercise canonical followup reads without running the semantic inbox fixture. + inputs = api.inputs(api.task(task_id)) + assert retry.admission(config, prior, inputs)["kind"] == "authorised_followup" + decision = retry.admission(config, prior, inputs) + retry.consume(prior, decision) + prior.update(attempt="subsequent-failure", phase="waiting") + worker.write_json(read_state(task_id)[0], prior) + # Replay a lost-ack inbox state: do not reopen/rebind to the subsequent failure. + state.pop("resume_applied") + api.tasks.update_task_status(task_id, "blocked") + inbox.finish(config, api, state, path) + assert api.task(task_id)["status"] == "blocked" + assert ( + inbox.task_followup(config, task_id)["retry"]["attempt"] != "subsequent-failure" + ) + assert retry.admission(config, prior, api.inputs(api.task(task_id))) is None + + +def test_manual_retry_uses_actual_controller_and_one_continuation(canonical_controller): + config, api, create, read_state, launches, fd = canonical_controller + task_id = create() + worker.tick(config, api, fd) + _, state = read_state(task_id) + api.tasks.add_task_comment( + task_id, + author_concept_id=config["delegator_id"], + body=f"/retry-coding {state['attempt']} Reconciled retained effects; retry once.", + ) + worker.tick(config, api, fd) + worker.tick(config, api, fd) + assert len(launches()) == 2 + assert launches()[-1]["retry_admission"]["kind"] == "manual_retry" + + +def test_transient_probe_can_resume_actual_controller( + canonical_controller, monkeypatch +): + config, api, create, read_state, launches, fd = canonical_controller + task_id = create() + worker.tick(config, api, fd) + path, state = read_state(task_id) + state["blocker"].update(kind="transient", probe_key="fixture") + worker.write_json(path, state) + receipt = repair_receipt(state) + receipt["observed_at"] = datetime.now(UTC).isoformat() + config["retry_probes"] = { + "fixture": { + "argv": [sys.executable, "-c", "print(" + repr(json.dumps(receipt)) + ")"], + "timeout_seconds": 1, + "interval_seconds": 10, + "max_interval_seconds": 40, + } + } + worker.tick(config, api, fd) + worker.tick(config, api, fd) + assert len(launches()) == 2 + assert launches()[-1]["retry_admission"]["kind"] == "probe_recovery" + + +@pytest.mark.parametrize("change", ["cancelled", "reassigned"]) +def test_revocation_during_probe_prevents_coding( + canonical_controller, monkeypatch, change +): + config, api, create, _read_state, launches, fd = canonical_controller + task_id = create() + worker.tick(config, api, fd) + + def probe(*args): + if change == "cancelled": + api.tasks.update_task_status(task_id, "cancelled") + else: + api.tasks.assign_task(task_id, "#V#other") + return { + "kind": "probe_recovery", + "token": "probe:fixture", + "reason": "Repaired", + } + + monkeypatch.setattr(worker, "probe_recovery", probe) + worker.tick(config, api, fd) + assert len(launches()) == 1 + + +def test_missing_state_requires_reconciliation_and_reports_once(canonical_controller): + config, api, create, read_state, launches, fd = canonical_controller + task_id = create() + api.tasks.update_task_status(task_id, "in_progress") + worker.tick(config, api, fd) + worker.tick(config, api, fd) + assert not launches() + assert api.task(task_id)["status"] == "blocked" + assert read_state(task_id)[1]["blocker"]["kind"] == "unclassified" + assert ( + len( + api.messages.get_messages_for_user( + config["delegator_id"], thread_id=task_id + ) + ) + == 1 + ) + + +def test_pending_report_failure_does_not_wedge_other_work( + canonical_controller, monkeypatch +): + config, api, create, read_state, launches, fd = canonical_controller + task_id = create() + worker.tick(config, api, fd) + path, state = read_state(task_id) + state["phase"] = "reporting" + worker.write_json(path, state) + original = api.finish + + def finish(state): + if state["task_id"] == task_id: + raise OSError("retained reporting unavailable") + original(state) + + monkeypatch.setattr(api, "finish", finish) + other = create("Another eligible task") + worker.tick(config, api, fd) + assert [row["task_id"] for row in launches()] == [task_id, other] + assert read_state(task_id)[1]["phase"] == "reporting" diff --git a/tests/backend/test_task_run_archives.py b/tests/backend/test_task_run_archives.py index 7fc066c0..c9676914 100644 --- a/tests/backend/test_task_run_archives.py +++ b/tests/backend/test_task_run_archives.py @@ -322,6 +322,8 @@ def test_exact_worker_launch_publishes_before_exit_and_downloads_after( "state_root": str(tmp_path), "agent_id": "#V#worker", "codex_command": str(executable), + "delegator_id": "#V#owner", + "organisation_id": "#V#org", } monkeypatch.setattr(routes, "_get_current_user_concept_id", lambda: actor["id"]) monkeypatch.setattr(routes, "_get_current_org_concept_id", lambda: actor["org"]) diff --git a/tests/test_codex_von_release.py b/tests/test_codex_von_release.py index 5a497f46..e31222d6 100644 --- a/tests/test_codex_von_release.py +++ b/tests/test_codex_von_release.py @@ -27,6 +27,7 @@ def source(tmp_path): for name in ( "codex_von_worker.py", "codex_von_inbox.py", + "codex_von_retry.py", "codex_von_deploy.py", "codex_von_release.py", "deploy_local_main.py", @@ -132,6 +133,28 @@ def test_activation_cannot_redirect_to_an_unrelated_checkout(source, tmp_path): releases.activate(root, source) +def test_retry_dependency_is_required_only_for_workers_that_import_it(source, tmp_path): + source, _, _ = source + (source / "scripts/codex_von_retry.py").unlink() + git(source, "add", "-u") + git(source, "commit", "-qm", "Incomplete new worker fixture") + with pytest.raises(deployer.runtime.DeploymentError, match="codex_von_retry"): + releases.prepare( + source, tmp_path / "releases", git(source, "rev-parse", "HEAD") + ) + (source / "scripts/codex_von_worker.py").write_text( + "# Pre-retry worker rollback fixture\n" + ) + git(source, "commit", "-qam", "Legacy worker fixture") + target = releases.prepare( + source, tmp_path / "releases", git(source, "rev-parse", "HEAD") + ) + assert ( + releases.activate(tmp_path / "releases", target)["status"] + == "selected_for_next_invocation" + ) + + def test_shared_controller_lock_is_inherited_without_unlocking(tmp_path): config = {"state_root": str(tmp_path)} with (tmp_path / "worker.lock").open("a") as lock: