diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md
index ff313753..3e6a32de 100644
--- a/docs/engineering/codex_dgx_worker.md
+++ b/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,9 +313,10 @@ 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
@@ -269,6 +324,7 @@ and accepts only `--commit` and `--receipt` from the controller. Set its path as
   "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",
diff --git a/scripts/codex_von_deploy.py b/scripts/codex_von_deploy.py
index dcc8f6d9..cca8c8ad 100755
--- a/scripts/codex_von_deploy.py
+++ b/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
 
diff --git a/scripts/codex_von_release.py b/scripts/codex_von_release.py
new file mode 100644
index 00000000..7981bdb7
--- /dev/null
+++ b/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()
diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py
index 236c7b40..d8f69aa1 100755
--- a/scripts/codex_von_worker.py
+++ b/scripts/codex_von_worker.py
@@ -19,6 +19,7 @@
 import sys
 import uuid
 from contextlib import ExitStack
+from datetime import datetime, timezone
 from pathlib import Path
 
 ACTIVE = {"pending", "in_progress", "blocked"}
@@ -123,6 +124,57 @@ def authorised_task(task, config):
     )
 
 
+def runtime_evidence(backend_root, api):
+    root = Path(backend_root).resolve()
+    return {
+        "observed_at": datetime.now(timezone.utc).isoformat(),
+        "pid": os.getpid(),
+        "commit": subprocess.check_output(
+            ["git", "-C", str(root), "rev-parse", "HEAD"], text=True
+        ).strip(),
+        "backend_root": str(root),
+        "worker_script": str(Path(__file__).resolve()),
+        "task_service_module": str(Path(api.tasks.__file__).resolve()),
+    }
+
+
+def check_task(config, api, task_id):
+    """Read assignment/settings through the scheduled adapter without pickup."""
+    task = api.task(task_id)
+    if not task or not authorised_task(task, config) or not api.native_writer(task):
+        raise PermissionError("Task is outside this worker's native assignment scope")
+    # Do not call pending(): that path can report external-writer assignments.
+    offset = 0
+    while True:
+        page = api.tasks.search_tasks(
+            assignee_concept_id=config["agent_id"],
+            created_by_concept_id=config["delegator_id"],
+            organisation_concept_id=config["organisation_id"],
+            statuses=[task["status"]],
+            limit=200,
+            offset=offset,
+        )
+        match = next(
+            (row for row in page["tasks"] if row["task_concept_id"] == task_id), None
+        )
+        if match is not None:
+            require_task_execution_preferences(match)
+            for key in ("requested_model", "requested_reasoning_effort"):
+                if match[key] != task[key]:
+                    raise RuntimeError("Task point/list execution preferences disagree")
+            break
+        offset += len(page["tasks"])
+        if not page["tasks"] or offset >= page["total"]:
+            raise RuntimeError("Assigned task is missing from canonical list read")
+    return {
+        "task_id": task_id,
+        "status": task["status"],
+        "eligible": task["status"] in ACTIVE,
+        "execution_settings": resolve_execution_settings(config, api.inputs(task)),
+        "pickup_performed": False,
+    }
+
+
 def validate_result(value):
     fields = set(RESULT_SCHEMA["required"])
     if not isinstance(value, dict) or set(value) not in (
@@ -651,6 +703,8 @@ def apply_deployment(config, api, state, state_path, lock_fd):
                 commit,
                 "--receipt",
                 str(receipt_path),
+                "--worker-lock-fd",
+                str(lock_fd),
                 *(["--recover-only"] if recover_only else []),
             ],
             check=False,
@@ -673,9 +727,9 @@ def apply_deployment(config, api, state, state_path, lock_fd):
         receipt
     )
     if success:
-        result["summary"] += (
-            f" Deployed and verified the public server at {commit[:12]}."
-        )
+        result[
+            "summary"
+        ] += f" Deployed and verified the public server at {commit[:12]}."
     else:
         result["status"] = "blocked"
         result["summary"] += (
@@ -686,9 +740,9 @@ def apply_deployment(config, api, state, state_path, lock_fd):
         )
     if recover_only:
         result["status"] = "blocked"
-        result["summary"] += (
-            " The task changed; only the interrupted deployment was reconciled."
-        )
+        result[
+            "summary"
+        ] += " The task changed; only the interrupted deployment was reconciled."
     write_json(state_path, state)
 
 
@@ -790,6 +844,9 @@ def main():
         help="Reviewed canonical backend checkout; "
         "required when running a copied worker script bundle",
     )
+    parser.add_argument(
+        "--check-task", help="Read back one assigned task without running work"
+    )
     options = parser.parse_args()
     os.umask(0o077)
     config = json.loads(Path(options.config).read_text())
@@ -802,7 +859,7 @@ def main():
         except BlockingIOError:
             print('{"outcome":"already_running"}')
             return
-        bind_backend_root(options.backend_root)
+        backend_root = bind_backend_root(options.backend_root)
         from src.backend.integrations.internal_mcp.gateway import (
             INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE,
             bind_internal_mcp_actor_context_source,
@@ -835,7 +892,14 @@ def main():
                 raise PermissionError(
                     "The configured worker has no organisation membership"
                 )
-            tick(config, Von(config), lock.fileno())
+            api = Von(config)
+            evidence = runtime_evidence(backend_root, api)
+            if options.check_task:
+                evidence["task_check"] = check_task(config, api, options.check_task)
+                print(json.dumps(evidence), flush=True)
+                return
+            write_json(state_root / "runtime.json", evidence)
+            tick(config, api, lock.fileno())
 
 
 if __name__ == "__main__":
diff --git a/tests/backend/test_codex_von_worker.py b/tests/backend/test_codex_von_worker.py
index 3614e302..bd40457f 100644
--- a/tests/backend/test_codex_von_worker.py
+++ b/tests/backend/test_codex_von_worker.py
@@ -487,6 +487,67 @@ def test_backend_root_accepts_current_canonical_checkout(monkeypatch):
     assert sys.path[0] == str(root)
 
 
+def test_read_only_installed_adapter_check_preserves_exact_task_settings(
+    config, monkeypatch
+):
+    row = task(config, requested_model="gpt-6-astra", requested_reasoning_effort="high")
+    api = worker.Von(config)
+    monkeypatch.setattr(api.tasks, "get_task", lambda _: row)
+    monkeypatch.setattr(
+        api.tasks, "search_tasks", lambda **_: {"tasks": [row], "total": 1}
+    )
+    monkeypatch.setattr(
+        api.tasks, "list_task_comments", lambda *a, **kw: {"comments": [], "total": 0}
+    )
+    monkeypatch.setattr(api.messages, "get_messages_for_user", lambda *a, **kw: [])
+    monkeypatch.setattr(api, "pending", lambda: pytest.fail("pickup path entered"))
+    result = worker.check_task(config, api, row["task_concept_id"])
+    assert result == {
+        "task_id": row["task_concept_id"],
+        "status": "pending",
+        "eligible": True,
+        "pickup_performed": False,
+        "execution_settings": {
+            "model": "gpt-6-astra",
+            "model_source": "task",
+            "reasoning_effort": "high",
+            "reasoning_effort_source": "task",
+        },
+    }
+    evidence = worker.runtime_evidence(Path(__file__).resolve().parents[2], api)
+    assert Path(evidence["task_service_module"]).is_relative_to(
+        evidence["backend_root"]
+    )
+    assert len(evidence["commit"]) == 40
+    row["status"] = "completed"
+    assert worker.check_task(config, api, row["task_concept_id"])["eligible"] is False
+
+
+def test_adapter_check_rejects_cross_scope_before_reading_task_inputs(config):
+    row = task(config, organisation_concept_id="#V#other")
+    api = SimpleNamespace(
+        task=lambda _: row, inputs=lambda _: pytest.fail("read inputs")
+    )
+    with pytest.raises(PermissionError, match="assignment scope"):
+        worker.check_task(config, api, row["task_concept_id"])
+
+
+@pytest.mark.parametrize("missing", [False, True])
+def test_adapter_check_exposes_stale_list_preferences(config, missing):
+    row = task(config, requested_model="gpt-6-astra", requested_reasoning_effort="high")
+    stale = {**row, "requested_reasoning_effort": None}
+    if missing:
+        stale.pop("requested_model")
+    api = SimpleNamespace(
+        task=lambda _: row,
+        native_writer=lambda _: True,
+        tasks=SimpleNamespace(search_tasks=lambda **_: {"tasks": [stale], "total": 1}),
+        inputs=lambda _: pytest.fail("resolved stale preferences"),
+    )
+    with pytest.raises(RuntimeError, match="execution.preference"):
+        worker.check_task(config, api, row["task_concept_id"])
+
+
 @pytest.mark.parametrize("missing", ["requested_model", "requested_reasoning_effort"])
 def test_old_backend_read_cannot_silently_launch_worker_defaults(
     config, monkeypatch, missing
diff --git a/tests/test_codex_von_deploy.py b/tests/test_codex_von_deploy.py
index 72d06401..9d4511bd 100644
--- a/tests/test_codex_von_deploy.py
+++ b/tests/test_codex_von_deploy.py
@@ -2,6 +2,7 @@
 import subprocess
 import sys
 import threading
+from pathlib import Path
 from http.server import BaseHTTPRequestHandler, HTTPServer
 
 import pytest
@@ -181,6 +182,139 @@ def test_recovery_only_cannot_start_a_new_deployment(deployment, monkeypatch):
     assert not path.exists()
 
 
+def test_web_and_worker_release_activation_share_recovery_receipt(
+    deployment, monkeypatch
+):
+    config, previous, target, events, path = deployment
+    root = path.parent / "releases"
+    config["worker_release_root"] = str(root)
+    monkeypatch.setattr(deployer.releases, "current", lambda _: root / previous)
+    monkeypatch.setattr(deployer.releases, "verify", lambda *a: {})
+    monkeypatch.setattr(deployer.releases, "prepare", lambda p, r, c: root / c)
+    monkeypatch.setattr(deployer.runtime, "deploy", lambda **_: events.append("web"))
+    monkeypatch.setattr(
+        deployer, "verify", lambda c, sha: events.append("health") or {"commit": sha}
+    )
+    monkeypatch.setattr(
+        deployer.releases,
+        "activate",
+        lambda r, t: events.append("pointer") or {"commit": target},
+    )
+    receipt = deployer.execute(config, target, path)
+    assert receipt["status"] == "deployed"
+    assert events == ["web", "health", "pointer"]
+    assert receipt["previous_worker_release"] == str(root / previous)
+    assert receipt["worker_activation"]["commit"] == target
+
+
+def test_interrupted_pointer_switch_reconciles_without_web_restart(
+    deployment, monkeypatch
+):
+    config, previous, target, events, path = deployment
+    root = path.parent / "releases"
+    config["worker_release_root"] = str(root)
+    deployer.save(
+        path,
+        {
+            "status": "started",
+            "requested_commit": target,
+            "previous_commit": previous,
+            "previous_worker_release": str(root / previous),
+        },
+    )
+    monkeypatch.setattr(
+        deployer.runtime,
+        "_read_health",
+        lambda _: {"version_details": {"git_commit": target}},
+    )
+    monkeypatch.setattr(
+        deployer.runtime, "deploy", lambda **_: pytest.fail("restarted web")
+    )
+    monkeypatch.setattr(deployer.releases, "prepare", lambda p, r, c: root / c)
+    monkeypatch.setattr(deployer.releases, "activate", lambda r, t: {"commit": t.name})
+    receipt = deployer.execute(config, target, path)
+    assert receipt["status"] == "deployed"
+    assert receipt["worker_activation"]["commit"] == target
+    assert not events
+
+
+def test_worker_activation_failure_restores_both_selections(deployment, monkeypatch):
+    config, previous, target, events, path = deployment
+    root = path.parent / "releases"
+    config["worker_release_root"] = str(root)
+    monkeypatch.setattr(deployer.releases, "current", lambda _: root / previous)
+    monkeypatch.setattr(deployer.releases, "verify", lambda *a: {})
+    monkeypatch.setattr(deployer.releases, "prepare", lambda p, r, c: root / c)
+    monkeypatch.setattr(deployer.runtime, "deploy", lambda **_: None)
+
+    def activate(r, target_path):
+        events.append(("pointer", Path(target_path).name))
+        if Path(target_path).name == target:
+            raise OSError("fixture pointer switch failure")
+        return {"commit": previous}
+
+    monkeypatch.setattr(deployer.releases, "activate", activate)
+    receipt = deployer.execute(config, target, path)
+    assert receipt["status"] == "rolled_back"
+    assert receipt["worker_activation"]["commit"] == previous
+    assert events[-2:] == [("pointer", previous), ("recover", previous)]
+
+
+def test_worker_rollback_failure_does_not_skip_web_recovery(deployment, monkeypatch):
+    config, previous, target, events, path = deployment
+    config["worker_release_root"] = str(path.parent / "releases")
+    deployer.save(
+        path,
+        {
+            "status": "started",
+            "requested_commit": target,
+            "previous_commit": previous,
+            "previous_worker_release": "/fixture/old",
+        },
+    )
+
+    def fail(*a):
+        raise OSError("fixture inaccessible release")
+
+    monkeypatch.setattr(deployer.releases, "activate", fail)
+    receipt = deployer.execute(config, target, path)
+    assert receipt["status"] == "recovery_failed"
+    assert receipt["worker_recovery_failure_type"] == "OSError"
+    assert receipt["verification"]["commit"] == previous
+    assert events == [("recover", previous)]
+
+
+def test_revoked_request_cannot_select_a_new_worker_release(deployment, monkeypatch):
+    config, previous, target, events, path = deployment
+    root = path.parent / "releases"
+    config["worker_release_root"] = str(root)
+    deployer.save(
+        path,
+        {
+            "status": "started",
+            "requested_commit": target,
+            "previous_commit": previous,
+            "previous_worker_release": str(root / previous),
+        },
+    )
+    monkeypatch.setattr(
+        deployer.runtime,
+        "_read_health",
+        lambda _: {"version_details": {"git_commit": target}},
+    )
+    monkeypatch.setattr(deployer.releases, "current", lambda _: root / previous)
+    monkeypatch.setattr(deployer.runtime, "_git", lambda *a: previous)
+    monkeypatch.setattr(
+        deployer.releases, "prepare", lambda *a: pytest.fail("new activation")
+    )
+    monkeypatch.setattr(
+        deployer.releases, "activate", lambda r, p: {"commit": Path(p).name}
+    )
+    receipt = deployer.execute(config, target, path, recover_only=True)
+    assert receipt["status"] == "rolled_back"
+    assert receipt["worker_activation"]["commit"] == previous
+
+
 def test_failed_real_process_restores_previous_git_revision_and_http_service(
     tmp_path, monkeypatch
 ):
diff --git a/tests/test_codex_von_release.py b/tests/test_codex_von_release.py
new file mode 100644
index 00000000..28145175
--- /dev/null
+++ b/tests/test_codex_von_release.py
@@ -0,0 +1,185 @@
+"""Local Git/process fixtures; never touch installed services or start Codex."""
+
+import fcntl
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from scripts import codex_von_deploy as deployer
+from scripts import codex_von_release as releases
+
+
+def git(root, *args):
+    return subprocess.check_output(["git", "-C", str(root), *args], text=True).strip()
+
+
+@pytest.fixture
+def source(tmp_path):
+    source = tmp_path / "source"
+    source.mkdir()
+    git(source, "init", "-q")
+    git(source, "config", "user.email", "fixture@example.invalid")
+    git(source, "config", "user.name", "Fixture")
+    repo = Path(__file__).resolve().parents[1]
+    for name in (
+        "codex_von_worker.py",
+        "codex_von_inbox.py",
+        "codex_von_deploy.py",
+        "codex_von_release.py",
+        "deploy_local_main.py",
+        "codex_von_worker_prompt.md",
+        "codex_von_inbox_prompt.md",
+    ):
+        target = source / "scripts" / name
+        target.parent.mkdir(exist_ok=True)
+        target.write_bytes((repo / "scripts" / name).read_bytes())
+    service = source / "src/backend/services/task_management_service.py"
+    service.parent.mkdir(parents=True)
+    service.write_text("RELEASE = 'old'\n")
+    git(source, "add", ".")
+    git(source, "commit", "-qm", "old fixture")
+    old = git(source, "rev-parse", "HEAD")
+    service.write_text("RELEASE = 'new'\n")
+    git(source, "commit", "-qam", "new fixture")
+    return source, old, git(source, "rev-parse", "HEAD")
+
+
+def test_complete_release_switch_preserves_loaded_files_and_rollback(source, tmp_path):
+    source, old, new = source
+    root = tmp_path / "releases"
+    old_path = releases.prepare(source, root, old)
+    releases.activate(root, old_path)
+    # Untracked operator material is never carried into a prepared release.
+    (source / "private-fixture.txt").write_text("not a release input")
+    new_path = releases.prepare(source, root, new)
+    assert not (new_path / "private-fixture.txt").exists()
+    assert releases.current(root) == old_path
+    old_contents = (
+        old_path / "src/backend/services/task_management_service.py"
+    ).read_text()
+    receipt = releases.activate(root, new_path)
+    assert receipt["status"] == "selected_for_next_invocation"
+    assert receipt["commit"] == new
+    assert releases.current(root) == new_path
+    assert (
+        old_path / "src/backend/services/task_management_service.py"
+    ).read_text() == old_contents
+    # Fresh process resolves both the actual worker and canonical service from
+    # the selected release, without a model call or a database connection.
+    code = """import json, pathlib, sys
+root = pathlib.Path(sys.argv[1]).resolve()
+sys.path.insert(0, str(root / 'scripts'))
+import codex_von_worker as worker
+worker.bind_backend_root(root)
+from src.backend.services import task_management_service as tasks
+print(json.dumps([worker.__file__, tasks.__file__, tasks.RELEASE]))
+"""
+    result = json.loads(
+        subprocess.check_output(
+            [sys.executable, "-c", code, str(root / "current")], cwd=tmp_path, text=True
+        )
+    )
+    assert result == [
+        str(new_path / "scripts/codex_von_worker.py"),
+        str(new_path / "src/backend/services/task_management_service.py"),
+        "new",
+    ]
+    assert releases.prepare(source, root, new) == new_path
+    releases.activate(root, old_path)
+    assert releases.current(root) == old_path
+
+
+def test_modified_release_is_not_reused_or_activated(source, tmp_path):
+    source, old, new = source
+    root = tmp_path / "releases"
+    predecessor = releases.prepare(source, root, old)
+    releases.activate(root, predecessor)
+    target = releases.prepare(source, root, new)
+    (target / "scripts/codex_von_worker.py").write_text("# damaged fixture\n")
+    for operation in (
+        lambda: releases.prepare(source, root, new),
+        lambda: releases.activate(root, target),
+    ):
+        with pytest.raises(deployer.runtime.DeploymentError, match="modified"):
+            operation()
+    assert releases.current(root) == predecessor
+
+
+def test_activation_cannot_redirect_to_an_unrelated_checkout(source, tmp_path):
+    source, _, _ = source
+    root = tmp_path / "releases"
+    root.mkdir()
+    with pytest.raises(deployer.runtime.DeploymentError, match="outside"):
+        releases.activate(root, source)
+
+
+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:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        with pytest.raises(BlockingIOError):
+            with deployer.worker_lock(config):
+                pytest.fail("overlapped active worker")
+        code = """import json, sys
+from scripts.codex_von_deploy import worker_lock
+with worker_lock(json.loads(sys.argv[1]), int(sys.argv[2])):
+ print('controller lock retained')
+"""
+        result = subprocess.check_output(
+            [sys.executable, "-c", code, json.dumps(config), str(lock.fileno())],
+            pass_fds=(lock.fileno(),),
+            text=True,
+        )
+        assert result.strip() == "controller lock retained"
+        with pytest.raises(BlockingIOError):
+            with deployer.worker_lock(config):
+                pytest.fail("child released parent's lock")
+    with deployer.worker_lock(config):
+        pass
+    with (tmp_path / "other.lock").open("a") as other:
+        with pytest.raises(ValueError, match="not the configured"):
+            with deployer.worker_lock(config, other.fileno()):
+                pytest.fail("accepted unrelated lock")
+
+
+def test_operator_bootstrap_cli_selects_only_merged_source_and_keeps_receipt(
+    source, tmp_path
+):
+    source, old, new = source
+    git(source, "branch", "-M", "main")
+    git(source, "remote", "add", "origin", str(source))
+    config = tmp_path / "fixture-config.json"
+    root = tmp_path / "releases"
+    config.write_text(
+        json.dumps(
+            {
+                "primary_root": str(source),
+                "state_root": str(tmp_path),
+                "worker_release_root": str(root),
+            }
+        )
+    )
+    receipt = tmp_path / "installation.json"
+    command = [
+        sys.executable,
+        "scripts/codex_von_release.py",
+        "--config",
+        str(config),
+        "--receipt",
+        str(receipt),
+        "--commit",
+    ]
+    stale = subprocess.run([*command, old], capture_output=True, text=True)
+    assert stale.returncode != 0
+    assert "current origin/main" in stale.stderr
+    assert not receipt.exists()
+    for _ in range(2):
+        output = subprocess.check_output([*command, new], text=True)
+        observed = json.loads(output)
+        assert observed["status"] == "selected_for_next_invocation"
+        assert observed["previous_worker_release"] is None
+        assert observed == json.loads(receipt.read_text())
+        assert releases.current(root) == root / new