Complete coherent DGX worker release installation (JVNAUTOSCI-2750)

docs/engineering/codex_dgx_worker.md

@@ -167,16 +167,70 @@ that omits them fails visibly before pickup instead of silently choosing default
 Installing only a newer worker script cannot repair an older canonical reader.
 Do not change global model defaults to compensate for a mismatched installation.
 
-For an authorised upgrade, install the reviewed backend and worker bundle as one
-compatible release, retain the previous backend/bundle selection for rollback,
-and update the scheduled wrapper to pass the chosen backend explicitly. Preserve
-the existing worker lock, state directory, identities, memberships and task
-selectors. Let active work finish before switching the scheduled process; do not
-launch a second controller to test pickup. Before re-enabling the schedule, check
-a labelled isolated task through that adapter's point/list reads, `inputs` and
-`resolve_execution_settings`, including exact task-sourced model/reasoning values.
-This installation change is separate from repository publication and public web
-deployment and requires its own activation authority.
+For an authorised upgrade, use the coherent installation below. Preserve the
+existing worker lock, state directory, identities, memberships, task selectors
+and five-minute schedule. A selected release is not evidence of actual pickup.
+Local installation and public deployment remain distinct effects; both need
+applicable task or standing authority.
+
+### Coherent worker/backend releases
+
+`scripts/codex_von_release.py` prepares a complete detached checkout of one
+reviewed revision with independent Git objects. It copies only tracked source,
+including backend, worker, inbox, prompts and deployment helpers. It does not
+copy credentials, environment files or private configuration. Installations
+reside in an operator-owned `worker_release_root/<full SHA>`; `current` is an
+atomically replaced symlink. Never modify or remove a release used by an active
+process. The existing PDM environment remains operator-owned; a changed
+dependency lock still requires recoverable environment preparation.
+
+One-time operator bootstrap (outside a coding sandbox):
+
+1. Retain the existing poll/deployment wrappers and their backend/bundle
+   bindings for rollback. Let the active controller finish. The installer
+   refuses to overlap its existing `state_root/worker.lock`.
+2. Add `worker_release_root` to the existing private deployment configuration;
+   keep its other bindings unchanged. Using the existing PDM Python, run the
+   reviewed `scripts/codex_von_release.py --config <deployment-config> --commit
+   <full-origin-main-SHA> --receipt <private-installation-receipt>`. This selects
+   source only; it does not start a worker, restart the web server or touch Von
+   records. The receipt retains the previous pointer if one exists. At the
+   first bootstrap, rollback also requires the retained pre-bootstrap wrappers.
+3. Retarget the existing `poll-von` wrapper to resolve `current` **once per
+   invocation, before importing any backend modules**. Use that resolved
+   directory for both `scripts/codex_von_worker.py` and `--backend-root`.
+   Retain the wrapper's database-target binding, event suppression, configured
+   identity, subscription launcher, model defaults and state. Retarget the
+   existing fixed deployment launcher to the same selected release's
+   `scripts/codex_von_deploy.py`. It must forward the controller's
+   `--worker-lock-fd`, `--commit`, `--receipt` and optional `--recover-only`;
+   preserve that descriptor through `exec` or explicit `pass_fds` if the wrapper
+   starts a subprocess. The configuration path remains operator-bound. Install
+   this pair together.
+4. Through the same prepared poll wrapper, run the worker with
+   `--check-task <assigned-native-task-id>` under its normal actor context.
+   Permit that diagnostic option explicitly in the wrapper. This checks
+   canonical point/list reads, `inputs` and exact resolved task preferences;
+   it prints only provenance/settings, does not accept conversation invitations,
+   send messages, change task state or invoke a model. A completed task is
+   reported as ineligible, not picked up. Check an isolated native assignment
+   with task-sourced Astra/high and no project/collection requirement.
+5. Retain the diagnostic output and inspect the next existing scheduled
+   invocation's `state_root/runtime.json`: observed time, PID, backend revision,
+   worker path and actual task-service module path must match the selection.
+   The runtime receipt proves controller startup, not task pickup or completion.
+
+With this bootstrap and `worker_release_root` enabled, each authorised public
+deployment prepares the complete worker release before stopping the server,
+verifies local/public health, then selects the worker release for the next
+invocation. The controller passes its existing lock descriptor to deployment;
+standalone operator deployment acquires that same lock. Loaded files remain
+intact. The deployment receipt retains both predecessor code revisions and
+worker selection. Interrupted activation reconciles the pointer; failed
+deployment attempts restore the prior worker selection and web service,
+reporting failures separately. Pointer selection is recorded as
+`selected_for_next_invocation`; verify actual startup and task settings as
+above before claiming local capability activation. No new poller is installed.
 
 Ordinary conversation `task_create` permits explicit assignment to a represented
 same-organisation coding agent using the continuation route's live membership
@@ -259,16 +313,18 @@ 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
+Install `codex_von_deploy.py`, `codex_von_release.py` and `deploy_local_main.py`
+from the same revision, preferably through the coherent release selection above.
+A fixed launcher binds a separate JSON configuration and accepts `--commit`,
+`--receipt`, `--worker-lock-fd` and optional `--recover-only` 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",
+  "worker_release_root": "/home/mjw/.codex-von-worker/releases",
   "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",

scripts/codex_von_deploy.py

@@ -12,13 +12,42 @@
 import json
 import os
 import re
+from contextlib import contextmanager
 from pathlib import Path
 from urllib.parse import urlparse
 
 try:
     from . import deploy_local_main as runtime
+    from . import codex_von_release as releases
 except ImportError:
     import deploy_local_main as runtime
+    import codex_von_release as releases
+
+
+@contextmanager
+def worker_lock(config, inherited_fd=None):
+    """Serialise host activation with the existing controller, including its child."""
+    path = Path(config["state_root"]) / "worker.lock"
+    with path.open("a") as lock:
+        fd = lock.fileno() if inherited_fd is None else inherited_fd
+        if (os.fstat(fd).st_dev, os.fstat(fd).st_ino) != (
+            path.stat().st_dev,
+            path.stat().st_ino,
+        ):
+            raise ValueError("Inherited descriptor is not the configured worker lock")
+        # A passed descriptor shares the controller's lock; never unlock it here.
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        yield
+
+
+def select_worker(config, receipt):
+    if not config.get("worker_release_root"):
+        return
+    root = config["worker_release_root"]
+    if not receipt.get("previous_worker_release"):
+        raise runtime.DeploymentError("Worker activation lacks its rollback selection")
+    target = releases.prepare(config["primary_root"], root, receipt["requested_commit"])
+    receipt["worker_activation"] = releases.activate(root, target)
 
 
 def save(path, value):
@@ -132,6 +161,15 @@ def execute(config, commit, receipt_path, *, recover_only=False):
                 "requested_commit": commit,
                 "previous_commit": previous_commit,
             }
+            if config.get("worker_release_root"):
+                previous_worker = releases.current(config["worker_release_root"])
+                releases.verify(
+                    previous_worker, runtime._git(previous_worker, "rev-parse", "HEAD")
+                )
+                # Persist rollback before the pointer can change. Preparing a new
+                # immutable checkout cannot affect the loaded controller/backend.
+                receipt["previous_worker_release"] = str(previous_worker)
+                releases.prepare(primary, config["worker_release_root"], commit)
             save(receipt_path, receipt)
             try:
                 runtime.deploy(
@@ -141,7 +179,9 @@ def execute(config, commit, receipt_path, *, recover_only=False):
                     health_timeout_seconds=config.get("health_timeout_seconds", 960),
                     expected_commit=commit,
                 )
-                receipt.update(status="deployed", verification=verify(config, commit))
+                verification = verify(config, commit)
+                select_worker(config, receipt)
+                receipt.update(status="deployed", verification=verification)
                 save(receipt_path, receipt)
                 return receipt
             except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc:
@@ -152,11 +192,26 @@ def execute(config, commit, receipt_path, *, recover_only=False):
         try:
             current = runtime._read_health(config["health_url"])
             if not runtime._health_error(current, commit):
-                receipt.update(status="deployed", verification=verify(config, commit))
+                verification = verify(config, commit)
+                if recover_only and config.get("worker_release_root"):
+                    selected = releases.current(config["worker_release_root"])
+                    if runtime._git(selected, "rev-parse", "HEAD") != commit:
+                        raise runtime.DeploymentError(
+                            "Revoked deployment cannot start a new worker activation"
+                        )
+                select_worker(config, receipt)
+                receipt.update(status="deployed", verification=verification)
                 save(receipt_path, receipt)
                 return receipt
         except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc:
             receipt["reconciliation_failure_type"] = type(exc).__name__
+        if receipt.get("previous_worker_release"):
+            try:
+                receipt["worker_activation"] = releases.activate(
+                    config["worker_release_root"], receipt["previous_worker_release"]
+                )
+            except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc:
+                receipt["worker_recovery_failure_type"] = type(exc).__name__
         try:
             receipt.update(
                 status="rolled_back",
@@ -166,6 +221,8 @@ def execute(config, commit, receipt_path, *, recover_only=False):
             receipt.update(
                 status="recovery_failed", recovery_failure_type=type(exc).__name__
             )
+        if receipt.get("worker_recovery_failure_type"):
+            receipt["status"] = "recovery_failed"
         save(receipt_path, receipt)
         return receipt
 
@@ -176,6 +233,7 @@ def main():
     parser.add_argument("--commit", required=True)
     parser.add_argument("--receipt", required=True)
     parser.add_argument("--recover-only", action="store_true")
+    parser.add_argument("--worker-lock-fd", type=int)
     args = parser.parse_args()
     os.umask(0o077)
     config = json.loads(Path(args.config).read_text())
@@ -188,9 +246,10 @@ def main():
         "VON_DURABLE_WORKFLOWS_ENABLE",
     ):
         os.environ.pop(key, None)
-    result = execute(
-        config, args.commit, Path(args.receipt), recover_only=args.recover_only
-    )
+    with worker_lock(config, args.worker_lock_fd):
+        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
 

scripts/codex_von_release.py

@@ -0,0 +1,160 @@
+"""Coherent, versioned backend/worker installations for operator-bound deployment.
+
+The caller owns the existing worker lock. Never overwrite a loaded release;
+only the next invocation resolves the atomically replaced current symlink.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import tempfile
+from pathlib import Path
+
+try:
+    from . import deploy_local_main as runtime
+except ImportError:
+    import deploy_local_main as runtime
+
+
+def current(release_root):
+    pointer = Path(release_root).resolve() / "current"
+    if not pointer.is_symlink():
+        raise runtime.DeploymentError(
+            "Worker release current must be an operator-prepared symlink"
+        )
+    target = pointer.resolve(strict=True)
+    if target.parent != pointer.parent or not target.is_dir():
+        raise runtime.DeploymentError("Worker release must be inside its release root")
+    return target
+
+
+def verify(target, commit):
+    target = Path(target).resolve()
+    if runtime._git(target, "rev-parse", "HEAD") != commit:
+        raise runtime.DeploymentError("Worker release revision mismatch")
+    if runtime._git(target, "status", "--porcelain", "--untracked-files=no"):
+        raise runtime.DeploymentError("Worker release has modified tracked files")
+    for name in (
+        "scripts/codex_von_worker.py",
+        "scripts/codex_von_inbox.py",
+        "scripts/codex_von_worker_prompt.md",
+        "scripts/codex_von_inbox_prompt.md",
+        "scripts/codex_von_deploy.py",
+        "scripts/codex_von_release.py",
+        "scripts/deploy_local_main.py",
+        "src/backend/services/task_management_service.py",
+    ):
+        if not (target / name).is_file():
+            raise runtime.DeploymentError("Incomplete worker release: " + name)
+    return {
+        "commit": commit,
+        "backend_root": str(target),
+        "worker_script": str(target / "scripts/codex_von_worker.py"),
+    }
+
+
+def prepare(primary, release_root, commit):
+    if not re.fullmatch(r"[0-9a-f]{40}", commit):
+        raise ValueError("Worker release requires a full Git SHA")
+    root = Path(release_root).resolve()
+    root.mkdir(parents=True, exist_ok=True)
+    target = root / commit
+    if not target.exists():
+        # Fetch only the exact reviewed commit into independent Git metadata.
+        # No environment, credentials, private config or source alternates are copied.
+        with tempfile.TemporaryDirectory(prefix=".prepare-", dir=root) as staging:
+            checkout = Path(staging) / "checkout"
+            runtime._run(["git", "init", "--quiet", checkout])
+            runtime._run(
+                [
+                    "git",
+                    "-C",
+                    checkout,
+                    "fetch",
+                    "--quiet",
+                    "--depth=1",
+                    Path(primary).resolve(),
+                    commit,
+                ]
+            )
+            runtime._run(
+                ["git", "-C", checkout, "checkout", "--quiet", "--detach", commit]
+            )
+            verify(checkout, commit)
+            checkout.rename(target)
+    verify(target, commit)
+    return target
+
+
+def activate(release_root, target):
+    root = Path(release_root).resolve()
+    target = Path(target).resolve(strict=True)
+    if target.parent != root:
+        raise runtime.DeploymentError(
+            "Worker activation target is outside release root"
+        )
+    commit = runtime._git(target, "rev-parse", "HEAD")
+    evidence = verify(target, commit)
+    # A private temporary directory avoids collisions or following an old link.
+    with tempfile.TemporaryDirectory(prefix=".switch-", dir=root) as staging:
+        link = Path(staging) / "current"
+        link.symlink_to(target)
+        link.replace(root / "current")
+    directory_fd = os.open(root, os.O_RDONLY)
+    try:
+        os.fsync(directory_fd)
+    finally:
+        os.close(directory_fd)
+    if current(root) != target:
+        raise runtime.DeploymentError("Worker release pointer read-back mismatch")
+    return {**evidence, "status": "selected_for_next_invocation"}
+
+
+def main():
+    """Operator bootstrap; no polling, model execution or service restart."""
+    try:
+        from .codex_von_deploy import save, worker_lock
+    except ImportError:
+        from codex_von_deploy import save, worker_lock
+
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--config", required=True, help="Operator deployment config")
+    parser.add_argument("--commit", required=True)
+    parser.add_argument("--receipt", required=True)
+    args = parser.parse_args()
+    os.umask(0o077)
+    config = json.loads(Path(args.config).read_text())
+    path = Path(args.receipt)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with worker_lock(config):
+        primary = Path(config["primary_root"]).resolve()
+        runtime._run(["git", "-C", primary, "fetch", "origin", "main"])
+        if runtime._git(primary, "rev-parse", "origin/main") != args.commit:
+            raise runtime.DeploymentError(
+                "Worker installation requires current origin/main"
+            )
+        root = Path(config["worker_release_root"]).resolve()
+        target = prepare(primary, root, args.commit)
+        if path.exists():
+            receipt = json.loads(path.read_text())
+            if receipt["requested_commit"] != args.commit:
+                raise ValueError("Installation receipt belongs to another revision")
+        else:
+            receipt = {
+                "status": "prepared",
+                "requested_commit": args.commit,
+                "previous_worker_release": (
+                    str(current(root)) if (root / "current").is_symlink() else None
+                ),
+            }
+            save(path, receipt)
+        receipt.update(activate(root, target))
+        save(path, receipt)
+    print(json.dumps(receipt), flush=True)
+
+
+if __name__ == "__main__":
+    main()