Deploy explicitly requested DGX fixes with verified code recovery

docs/engineering/codex_dgx_worker.md

@@ -17,7 +17,7 @@ transcript separately. The current Tasks form's assignee picker offers only
 Unassigned and Von; it does not yet offer this worker. The pilot acceptance
 used labelled fixtures through canonical services, not that picker.
 
-The worker checks hourly, starts at most one coding run per check, and sends a
+The worker checks at the operator-configured interval, starts at most one coding run per check, and sends a
 pickup message followed by a result or question through Von direct messages.
 An empty check makes no model request. Each task has its own Git worktree. Each
 execution is a fresh subscription-backed Codex session, using the retained
@@ -32,7 +32,8 @@ messages do not create new coding assignments in this pilot.
 Publishing uses an operator-configured GitHub account and follows each task's
 publication authority. Missing authentication leaves changes retained and the
 task blocked; it must not claim that local code is published. Deployment of
-the public web server is outside this worker's execution profile. The chosen
+the public web server requires an explicit instruction in the initial task and
+the operator-enabled deployment command described below. The chosen
 GitHub account and current authentication status belong in the Jira decision
 surface, not in the task's text.
 
@@ -67,6 +68,14 @@ prevents the worker from changing the task state. Its result still goes to the
 original configured delegator. This is a single-host worker, not a distributed
 claim/lease protocol.
 
+When the initial task explicitly requests deployment, the coding run returns
+the full merged `deploy_commit` after its code, tests and publication finish.
+The controller rechecks the live task authority and instructions, then calls
+only its operator-configured `deployment_command`. The run cannot supply a
+host, checkout, shell command or credentials. An empty deployment request has
+no deployment effect. Interpretation of the initial task belongs to the model;
+there is no keyword matcher or second per-deployment confirmation.
+
 The Codex process defaults to `workspace-write` sandboxing and an explicit non-Sol
 model. Publication installations use the scoped profile below. Its environment
 omits the controller's Mongo credentials and unrelated
@@ -141,16 +150,57 @@ restrictions; preserve their work and migrate them explicitly if publication
 is required. Never grant writes to the live web checkout's shared `.git` as a
 shortcut. See [Codex permissions](https://learn.chatgpt.com/docs/permissions).
 
+## Explicit deployment
+
+Install `codex_von_deploy.py` and its `deploy_local_main.py` dependency in the
+operator-owned directory. A fixed launcher binds a separate JSON configuration
+and accepts only `--commit` and `--receipt` from the controller. Set its path as
+`deployment_command` in worker config. The deployment configuration contains:
+
+```json
+{
+  "primary_root": "/home/mjw/Von",
+  "runtime_root": "/home/mjw/Von-runtime-main",
+  "state_root": "/home/mjw/.codex-von-worker/worker-state",
+  "health_url": "http://127.0.0.1:5000/health",
+  "public_health_url": "https://von.curiouscat.cc/health",
+  "public_headers_file": "/home/mjw/.codex-von-worker/cloudflare-health.json",
+  "health_timeout_seconds": 960
+}
+```
+
+Prepare the dedicated detached runtime worktree with its environment before
+activation. The command uses the canonical launcher, requires the requested
+SHA to remain `origin/main`, and verifies the exact clean revision locally and
+through the public endpoint. Persistent receipts make retries reconcile an
+interrupted deployment. Failed startup restores the previous code revision
+and verifies the recovered service; a recovery failure remains an explicit
+blocked outcome. This is code recovery, not a database rollback. A changed
+`pdm.lock` requires preparation of a recoverable environment before automatic
+deployment. Migrations and infrastructure changes need their own task scope
+and recovery plan.
+
+For a Cloudflare Access-protected site, keep the two service-token headers in
+the private `public_headers_file`. The deployment command injects only the
+service Client ID into `VON_CLOUDFLARE_HEALTH_SERVICE_IDS` for the server.
+Cloudflare must issue the token for the configured application's Service Auth
+policy. Von validates its signature, issuer, audience, expiry and exact
+configured service identity for `GET /health` only; it never creates a Von user
+session. Human login and private routes retain their existing checks. Health
+requests do not follow redirects, so credentials cannot be forwarded to a
+login provider. Token expiry belongs in the operator's installation record.
+See [Cloudflare service tokens](https://developers.cloudflare.com/cloudflare-one/access-controls/service-credentials/service-tokens/).
+
 The DGX operator wrapper reads the existing web runtime's database credential
 in-process, checks that the configured database target still matches the
 public website, binds the worker config, and suppresses event-driven workflow
 launches in that controller process. It does not alter the web process's
 configuration. Store credentials, configuration and operational state outside
 the repository with owner-only permissions. Never log the database URI.
 
-On a host with an active cron service, install an hourly entry invoking the
+On a host with an active cron service, install a five-minute entry invoking the
 wrapper and redirecting stdout/stderr to an owner-only log. An example schedule
-is `17 * * * *`; preserve any other existing crontab entries. Run the exact
+is `*/5 * * * *`; preserve any other existing crontab entries. Run the exact
 entry manually before installation. No Codex app session or SSH connection
 needs to remain open. The controller uses a nonblocking file lock and passes
 it to its child so concurrent invocations do not start overlapping work.

run.sh

@@ -3634,6 +3634,7 @@ deploy_main() {
     "$py" "${ROOT}/scripts/deploy_local_main.py" \
         --primary-root "$ROOT" \
         --runtime-worktree "$runtime_path" \
+        --health-url "http://127.0.0.1:${PORT}/health" \
         --health-timeout-seconds "$HEALTH_TIMEOUT_SEC"
 }
 

scripts/codex_von_deploy.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""Operator-bound deployment and code recovery for the external Codex worker.
+
+Paths and endpoints come only from the operator's config. The coding run may
+request a merged revision, never a command, host, credential or checkout path.
+"""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import json
+import os
+import re
+from pathlib import Path
+from urllib.parse import urlparse
+
+try:
+    from . import deploy_local_main as runtime
+except ImportError:
+    import deploy_local_main as runtime
+
+
+def save(path, value):
+    temporary = path.with_suffix(".tmp")
+    with temporary.open("w") as stream:
+        json.dump(value, stream, indent=2)
+        stream.flush()
+        os.fsync(stream.fileno())
+    temporary.replace(path)
+
+
+def verify(config, commit):
+    timeout = config.get("health_timeout_seconds", 960)
+    local = runtime._wait_for_verified_health(config["health_url"], commit, timeout)
+    headers = None
+    if config.get("public_headers_file"):
+        headers = json.loads(Path(config["public_headers_file"]).read_text())
+        if set(headers) != {"CF-Access-Client-Id", "CF-Access-Client-Secret"}:
+            raise ValueError("Unexpected public health authentication headers")
+    public = runtime._wait_for_verified_health(
+        config["public_health_url"], commit, timeout, headers=headers
+    )
+    return {"local_pid": local["pid"], "public_pid": public["pid"], "commit": commit}
+
+
+def recover(config, previous_commit):
+    primary = Path(config["primary_root"]).resolve()
+    checkout = Path(config["runtime_root"]).resolve()
+    # Stop only the exact deployment port using the canonical launcher. Changing
+    # code while a candidate process is still alive would invalidate read-back.
+    launcher = checkout / "run.sh"
+    port = str(urlparse(config["health_url"]).port)
+    runtime._run(
+        ["bash", launcher, "stop", "-Port", port, "-NoBrowser"],
+        cwd=checkout,
+        capture_output=False,
+    )
+    runtime._prepare_runtime(primary, checkout, previous_commit)
+    runtime._reconcile_startup_seed_materialisations(checkout)
+    runtime._run(
+        [
+            "bash",
+            launcher,
+            "restart",
+            "-Port",
+            port,
+            "-NoBrowser",
+            "-HealthTimeoutSec",
+            str(config.get("health_timeout_seconds", 960)),
+        ],
+        cwd=checkout,
+        capture_output=False,
+    )
+    runtime._verify_workers(checkout)
+    # An older release can predate machine health authentication. Restore and
+    # verify its local service even if that old public route rejects this token.
+    local = runtime._wait_for_verified_health(
+        config["health_url"], previous_commit, config.get("health_timeout_seconds", 960)
+    )
+    try:
+        return verify(config, previous_commit)
+    except runtime.HealthAccessError:
+        return {
+            "commit": previous_commit,
+            "local_pid": local["pid"],
+            "public_verification": "authentication_unavailable_for_previous_release",
+        }
+
+
+def execute(config, commit, receipt_path, *, recover_only=False):
+    if not re.fullmatch(r"[0-9a-f]{40}", commit):
+        raise ValueError("Deployment requires a full Git commit SHA")
+    primary = Path(config["primary_root"]).resolve()
+    checkout = Path(config["runtime_root"]).resolve()
+    runtime._validate_runtime(primary, checkout)
+    receipt_path = Path(receipt_path)
+    receipt_path.parent.mkdir(parents=True, exist_ok=True)
+    with (Path(config["state_root"]) / "deployment.lock").open("a") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        receipt = (
+            json.loads(receipt_path.read_text()) if receipt_path.exists() else None
+        )
+        if receipt:
+            if receipt["requested_commit"] != commit:
+                raise ValueError("Deployment receipt belongs to another revision")
+            if receipt["status"] in {"deployed", "rolled_back", "recovery_failed"}:
+                return receipt
+        else:
+            if recover_only:
+                raise ValueError("Recovery requires an existing deployment receipt")
+            # Exact target and healthy predecessor are preconditions: a stale
+            # request must not stop a healthy server or fast-forward its checkout.
+            runtime._run(["git", "-C", primary, "fetch", "origin", "main"])
+            if runtime._git(primary, "rev-parse", "origin/main") != commit:
+                raise runtime.DeploymentError(
+                    "Requested revision is no longer origin/main"
+                )
+            prior = runtime._read_health(config["health_url"])
+            previous_commit = prior["version_details"]["git_commit"]
+            error = runtime._health_error(prior, previous_commit)
+            if error:
+                raise runtime.DeploymentError("Predecessor is not ready: " + error)
+            if runtime._git(primary, "show", f"{commit}:pdm.lock") != runtime._git(
+                primary, "show", f"{previous_commit}:pdm.lock"
+            ):
+                raise runtime.DeploymentError(
+                    "Dependency lock changed; prepare a recoverable environment before deployment"
+                )
+            receipt = {
+                "status": "started",
+                "requested_commit": commit,
+                "previous_commit": previous_commit,
+            }
+            save(receipt_path, receipt)
+            try:
+                runtime.deploy(
+                    primary_root=primary,
+                    runtime_root=checkout,
+                    health_url=config["health_url"],
+                    health_timeout_seconds=config.get("health_timeout_seconds", 960),
+                    expected_commit=commit,
+                )
+                receipt.update(status="deployed", verification=verify(config, commit))
+                save(receipt_path, receipt)
+                return receipt
+            except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc:
+                receipt["failure_type"] = type(exc).__name__
+                save(receipt_path, receipt)
+        # An interrupted controller never blindly repeats a restart. Reconcile
+        # the target first; if it is not serving, restore the recorded predecessor.
+        try:
+            current = runtime._read_health(config["health_url"])
+            if not runtime._health_error(current, commit):
+                receipt.update(status="deployed", verification=verify(config, commit))
+                save(receipt_path, receipt)
+                return receipt
+        except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc:
+            receipt["reconciliation_failure_type"] = type(exc).__name__
+        try:
+            receipt.update(
+                status="rolled_back",
+                verification=recover(config, receipt["previous_commit"]),
+            )
+        except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc:
+            receipt.update(
+                status="recovery_failed", recovery_failure_type=type(exc).__name__
+            )
+        save(receipt_path, receipt)
+        return receipt
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--config", required=True)
+    parser.add_argument("--commit", required=True)
+    parser.add_argument("--receipt", required=True)
+    parser.add_argument("--recover-only", action="store_true")
+    args = parser.parse_args()
+    os.umask(0o077)
+    config = json.loads(Path(args.config).read_text())
+    if config.get("public_headers_file"):
+        headers = json.loads(Path(config["public_headers_file"]).read_text())
+        os.environ["VON_CLOUDFLARE_HEALTH_SERVICE_IDS"] = headers["CF-Access-Client-Id"]
+    # Controller-only event suppression must not change the production service.
+    for key in (
+        "VON_EVENT_WORKFLOW_INTEGRATION_ENABLE",
+        "VON_DURABLE_WORKFLOWS_ENABLE",
+    ):
+        os.environ.pop(key, None)
+    result = execute(
+        config, args.commit, Path(args.receipt), recover_only=args.recover_only
+    )
+    print(json.dumps(result), flush=True)
+    return 0 if result["status"] == "deployed" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())