Publish curated Unraid-AppOps-Gateway source from 21729207a161d87393cfb73e79582bc373aa19cf

PUBLIC_SOURCE_EXPORT.md

@@ -1,3 +1,3 @@
 # Public source export
 
-This source snapshot was generated from the private canonical repository at revision `bc001a6df859f7c9836fa145afb739d1ceebc524`. It contains no private Git history or operational evidence. Changes are published from the canonical repository.
+This source snapshot was generated from the private canonical repository at revision `21729207a161d87393cfb73e79582bc373aa19cf`. It contains no private Git history or operational evidence. Changes are published from the canonical repository.

PUBLIC_SOURCE_MANIFEST.json

@@ -1,6 +1,6 @@
 {
   "schemaVersion": 1,
-  "sourceRevision": "bc001a6df859f7c9836fa145afb739d1ceebc524",
+  "sourceRevision": "21729207a161d87393cfb73e79582bc373aa19cf",
   "files": [
     {
       "path": ".dockerignore",
@@ -969,13 +969,13 @@
     },
     {
       "path": "backend/tests/test_controller_hub_release_volume.py",
-      "sha256": "4fe32bd783e2535724d75ff892e9b842ecba0a3b3ec6b141de45ab31d96bd945",
+      "sha256": "b629dadd5616d5afd5ff8ad33662d93da829424226f0bb36bda71d9c621049cf",
-      "bytes": 118768
+      "bytes": 144585
     },
     {
       "path": "backend/tests/test_controller_root_procedure_contract.py",
-      "sha256": "964b3e21828a9d815d05ffb0e51acf6e9e45efe56af9fa3b70f39200d3a5c336",
+      "sha256": "2e6a8645674c32f16d251b5cdcdc4f75d72759c4253aebc312b58e766b1b6581",
-      "bytes": 5125
+      "bytes": 6445
     },
     {
       "path": "backend/tests/test_controller_runtime_capabilities.py",
@@ -994,8 +994,8 @@
     },
     {
       "path": "backend/tests/test_deploy_broker.py",
-      "sha256": "6b1ad652cc7f06ce94105846e15110741fb9514e833101ae670279d5a1711deb",
+      "sha256": "b0e96b1cd2954dc2a6452f5f7c50c5fbae541dc0b4de1e60af11349654594902",
-      "bytes": 415988
+      "bytes": 438851
     },
     {
       "path": "backend/tests/test_deployment_auto_rollback.py",
@@ -1584,8 +1584,8 @@
     },
     {
       "path": "backend/tests/test_runner_fleet_cutover_safety.py",
-      "sha256": "85f93b1e70114ea1b86bdc4adeb05ea18b91e2886c5ed0dd44474991ee3ddda9",
+      "sha256": "2eb1ec46b706a00a3bb7a5fc6d0d135fccc2d82324ba03296286e3b30a0747a9",
-      "bytes": 15917
+      "bytes": 29912
     },
     {
       "path": "backend/tests/test_runner_fleet_scheduler_recovery.py",

backend/tests/test_controller_hub_release_volume.py

@@ -318,7 +318,10 @@ def _prepare_promotion_doubles(monkeypatch, tmp_path, *, config_valid=True):
             raise module.VolumeTransitionError("staged_config_invalid")
 
     monkeypatch.setattr(module, "_validate_staged_configuration", validate_staged)
-    monkeypatch.setattr(module, "_install_release_leaf", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        module, "_install_release_leaf",
+        lambda revision, *_args, **_kwargs: events.append(f"release_installed:{revision}"),
+    )
     monkeypatch.setattr(module, "_read_root_compose", lambda: (ORIGINAL_COMPOSE, 0o644))
     monkeypatch.setattr(module, "_mirror_snapshot_tar", lambda: b"mirror")
     monkeypatch.setattr(
@@ -329,6 +332,10 @@ def _prepare_promotion_doubles(monkeypatch, tmp_path, *, config_valid=True):
         module, "_verify_compose_image",
         lambda *_args, **_kwargs: events.append("compose_verified"),
     )
+    monkeypatch.setattr(
+        module, "_verify_promotion_source_attestation",
+        lambda *_args, **_kwargs: events.append("source_attested"),
+    )
     monkeypatch.setattr(
         module, "ensure_backup_volume",
         lambda **_kwargs: events.append("backup_volume_prepared"),
@@ -360,6 +367,73 @@ def test_promotion_prepares_backup_volume_after_source_stage_controller_and_conf
     assert events.index("compose_rendered") < events.index("backup_volume_prepared")
     assert events.count("compose_verified") == 2
     assert events.index("compose_verified") < events.index("backup_volume_prepared")
+    assert events.count("source_attested") == 4
+    attestation_after_backup = events.index(
+        "source_attested", events.index("backup_volume_prepared") + 1
+    )
+    assert attestation_after_backup < events.index("release_installed:" + "b" * 40)
+
+
+def test_promotion_rejects_config_leaf_change_during_staged_validation(monkeypatch, tmp_path):
+    events = _prepare_promotion_doubles(monkeypatch, tmp_path)
+    original_config = ({}, {"": 0o755})
+    changed_config = ({"changed.json": (b"changed", 0o600)}, {"": 0o755})
+    reads = 0
+
+    def read_config():
+        nonlocal reads
+        reads += 1
+        if reads == 1:
+            return original_config
+        events.append("config_changed")
+        return changed_config
+
+    monkeypatch.setattr(module, "_read_config_leaf", read_config)
+
+    with pytest.raises(module.VolumeTransitionError, match="controller_volume_config_changed"):
+        module.prepare_promotion(
+            tmp_path / "stage",
+            source_revision="d" * 40,
+            source_tree="e" * 40,
+            controller_version="2.2.0",
+            image_id=IMAGE_ID,
+        )
+
+    assert reads == 2
+    assert events.index("config_validated") < events.index("config_changed")
+    assert not any(event.startswith("release_installed:") for event in events)
+
+
+def test_promotion_rejects_legacy_config_source_change_before_config_install(monkeypatch, tmp_path):
+    events = _prepare_promotion_doubles(monkeypatch, tmp_path)
+    monkeypatch.setattr(module, "_read_config_leaf", lambda: None)
+    monkeypatch.setattr(
+        module, "_config_tree",
+        lambda entries, _directories, **_kwargs: (
+            ({"changed.json": (b"changed", 0o600)}, {"": 0o755})
+            if "changed.json" in entries else ({}, {"": 0o755})
+        ),
+    )
+    identity = {
+        "sourceRevision": "b" * 40,
+        "sourceTree": "c" * 40,
+        "controllerVersion": "2.1.0",
+    }
+    monkeypatch.setattr(
+        module, "_capture_controller_source",
+        lambda *_args: ({"changed.json": (b"changed", 0o600)}, {"": 0o755}, identity),
+    )
+
+    with pytest.raises(module.VolumeTransitionError, match="controller_volume_config_changed"):
+        module.prepare_promotion(
+            tmp_path / "stage",
+            source_revision="d" * 40,
+            source_tree="e" * 40,
+            controller_version="2.2.0",
+            image_id=IMAGE_ID,
+        )
+
+    assert not any(event.startswith("release_installed:") for event in events)
 
 
 def test_promotion_does_not_prepare_backup_volume_when_staged_configuration_fails(
@@ -1226,16 +1300,489 @@ def test_drifted_tag_requires_healthy_current_controller_provenance(compose, mon
     elif damage == "interpolation":
         source_reference = "${CONTROLLER_IMAGE}"
     original = f"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: {source_reference}\n".encode()
-    if damage:
     with pytest.raises(module.VolumeTransitionError, match="image_reference_(changed|invalid)"):
         module.rendered_model(original, REVISION, IMAGE_ID, current_container=current)
-    else:
-        rendered = json.loads(module.rendered_model(original, REVISION, IMAGE_ID, current_container=current))
-        assert rendered["services"][module.SERVICE]["image"] == IMAGE_ID
-        assert rendered["services"][module.SERVICE]["pull_policy"] == "never"
     assert not any("up" in arguments or "pull" in arguments for arguments, _ in calls)
 
 
+def _tag_drift_container():
+    return {
+        "Id": "d" * 64,
+        "Image": IMAGE_ID,
+        "Name": f"/{module.CONTAINER}",
+        "State": {"Running": True, "Health": {"Status": "healthy"}},
+        "Config": {
+            "Image": "docker.gitea.com/runner-images:ubuntu-latest",
+            "Labels": {
+                "com.docker.compose.project": "gitea-deploy-runners",
+                "com.docker.compose.service": "deploy-control",
+                "com.docker.compose.oneoff": "False",
+            },
+        },
+        "Mounts": [
+            {
+                "Type": "bind",
+                "Source": module.CONTROLLER_DIR.as_posix(),
+                "Destination": module.EXECUTABLE_TARGET,
+                "RW": False,
+            },
+            {
+                "Type": "bind",
+                "Source": module._expected_config_path().as_posix(),
+                "Destination": module.CONFIG_TARGET,
+                "RW": False,
+            },
+            {
+                "Type": "volume",
+                "Name": module.VOLUME,
+                "Source": module.MOUNTPOINT.as_posix(),
+                "Destination": module.TARGET,
+                "RW": True,
+            },
+        ],
+    }
+
+
+_DEFAULT_SOURCE_IMAGE_ID = object()
+
+
+def _install_tag_drift_docker_fixture(
+    monkeypatch, *, source_image_id=_DEFAULT_SOURCE_IMAGE_ID, source_reference=None,
+    expected_image_output=IMAGE_ID, mutate=None
+):
+    source_ref = "docker.gitea.com/runner-images:ubuntu-latest"
+    if source_reference is not None:
+        source_ref = source_reference
+    if source_image_id is _DEFAULT_SOURCE_IMAGE_ID:
+        source_image_id = "sha256:" + "c" * 64
+    container = _tag_drift_container()
+    if mutate is not None:
+        mutate(container)
+    compose_model = model()
+    compose_model["services"][module.SERVICE]["image"] = source_ref
+
+    def run(arguments, *, data=None, **_kwargs):
+        if arguments[:2] == ["docker", "inspect"]:
+            return json.dumps([container]).encode("utf-8")
+        if arguments[:3] == ["docker", "image", "inspect"]:
+            reference = arguments[-1]
+            if reference == source_ref:
+                if source_image_id is None:
+                    return b""
+                return (source_image_id + "\n").encode("ascii")
+            if reference == IMAGE_ID:
+                return (expected_image_output + "\n").encode("ascii") if expected_image_output else b""
+            return b"sha256:" + b"c" * 64 + b"\n"
+        if arguments[-2:] == ["--format", "json"]:
+            return json.dumps(compose_model).encode("utf-8")
+        return b""
+
+    monkeypatch.setattr(module, "_run", run)
+    return container, source_ref, source_image_id
+
+
+def test_observed_runner_tag_drift_proof_pins_both_rendered_models(monkeypatch):
+    container, source_ref, source_image_id = _install_tag_drift_docker_fixture(monkeypatch)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+
+    proof = module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+    rollback = json.loads(module.rendered_model(
+        original, REVISION, IMAGE_ID, source_tag_drift_proof=proof
+    ))
+    forward = json.loads(module.rendered_model(
+        original, "c" * 40, IMAGE_ID, source_tag_drift_proof=proof
+    ))
+
+    assert proof.source_reference == source_ref
+    assert proof.source_image_id == source_image_id
+    for rendered in (rollback, forward):
+        assert rendered["services"][module.SERVICE]["image"] == IMAGE_ID
+        assert rendered["services"][module.SERVICE]["pull_policy"] == "never"
+
+
+@pytest.mark.parametrize(
+    ("mutate", "source_image_id", "expected_error"),
+    [
+        (lambda item: item["Config"].__setitem__("Image", "wrong:tag"), None, "source_tag_drift_container_invalid"),
+        (lambda item: item["Config"]["Labels"].__setitem__("com.docker.compose.project", "other"), None, "source_tag_drift_container_invalid"),
+        (lambda item: item["Config"]["Labels"].__setitem__("com.docker.compose.service", "runner"), None, "source_tag_drift_container_invalid"),
+        (lambda item: item["Config"]["Labels"].__setitem__("com.docker.compose.oneoff", "True"), None, "source_tag_drift_container_invalid"),
+        (lambda item: item["State"].__setitem__("Running", False), None, "container_not_running"),
+        (lambda item: item.__setitem__("Name", "/renamed-controller"), None, "container_image_changed"),
+        (lambda item: item["Mounts"].__setitem__(0, {"Type": "bind", "Source": "/unsafe", "Destination": module.EXECUTABLE_TARGET, "RW": False}), None, "executable_mount_invalid"),
+        (lambda item: item["Mounts"][1].__setitem__("RW", True), None, "config_alias_invalid"),
+        (None, None, "image_reference_invalid"),
+        (None, IMAGE_ID, "source_tag_drift_not_observed"),
+        (None, "not-an-image-id", "image_reference_invalid"),
+    ],
+)
+def test_observed_runner_tag_drift_rejects_unattested_or_non_drift_cases(
+    monkeypatch, mutate, source_image_id, expected_error
+):
+    container, _source_ref, _observed_id = _install_tag_drift_docker_fixture(
+        monkeypatch, source_image_id=source_image_id, mutate=mutate
+    )
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+
+    with pytest.raises(module.VolumeTransitionError, match=expected_error):
+        module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+
+
+def test_observed_runner_tag_drift_rejects_wrong_source_tag(monkeypatch):
+    container, _source_ref, _observed_id = _install_tag_drift_docker_fixture(
+        monkeypatch, source_reference="docker.gitea.com/runner-images@sha256:" + "c" * 64
+    )
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images@sha256:" + b"c" * 64 + b"\n"
+
+    with pytest.raises(module.VolumeTransitionError, match="source_tag_drift_reference_invalid"):
+        module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+
+
+def test_observed_runner_tag_drift_rejects_missing_local_expected_image(monkeypatch):
+    container, _source_ref, _observed_id = _install_tag_drift_docker_fixture(
+        monkeypatch, expected_image_output=""
+    )
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+
+    with pytest.raises(module.VolumeTransitionError, match="source_tag_drift_expected_image_missing"):
+        module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+
+
+def test_rendered_model_rejects_altered_drift_proof(monkeypatch):
+    container, _source_ref, _observed_id = _install_tag_drift_docker_fixture(monkeypatch)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+    proof = module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+    altered = proof._replace(compose_sha256="0" * 64)
+
+    with pytest.raises(module.VolumeTransitionError, match="source_tag_drift_proof_invalid"):
+        module.rendered_model(original, REVISION, IMAGE_ID, source_tag_drift_proof=altered)
+
+
+@pytest.mark.parametrize(
+    ("mutate", "expected_error"),
+    [
+        (lambda item: item.__setitem__("Id", "e" * 64), "container_changed_during_tag_drift_check"),
+        (lambda item: item.__setitem__("Image", "sha256:" + "e" * 64), "container_image_changed"),
+        (lambda item: item["Config"].__setitem__("Image", "other:tag"), "source_tag_drift_container_invalid"),
+        (lambda item: item["Config"]["Labels"].__setitem__("com.docker.compose.project", "other"), "source_tag_drift_container_invalid"),
+        (lambda item: item["Mounts"][1].__setitem__("RW", True), "config_alias_invalid"),
+        (lambda item: item["Mounts"].append({"Type": "bind", "Source": "/other", "Destination": "/extra", "RW": False}), "source_tag_drift_proof_invalid"),
+        (lambda item: item["Config"].__setitem__("User", "unexpected"), "source_tag_drift_proof_invalid"),
+    ],
+)
+def test_rendered_model_revalidates_live_container_and_mount_proof(
+    monkeypatch, mutate, expected_error
+):
+    container, _source_ref, _observed_id = _install_tag_drift_docker_fixture(monkeypatch)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+    proof = module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+    mutate(container)
+
+    with pytest.raises(module.VolumeTransitionError, match=expected_error):
+        module.rendered_model(original, REVISION, IMAGE_ID, source_tag_drift_proof=proof)
+
+
+def test_rendered_model_rejects_tag_that_resolves_back_to_expected_image(monkeypatch):
+    container, source_ref, _observed_id = _install_tag_drift_docker_fixture(monkeypatch)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+    proof = module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+    original_run = module._run
+
+    def moved_tag(arguments, **kwargs):
+        if arguments[:3] == ["docker", "image", "inspect"] and arguments[-1] == source_ref:
+            return (IMAGE_ID + "\n").encode("ascii")
+        return original_run(arguments, **kwargs)
+
+    monkeypatch.setattr(module, "_run", moved_tag)
+
+    with pytest.raises(module.VolumeTransitionError, match="source_tag_drift_changed"):
+        module.rendered_model(original, REVISION, IMAGE_ID, source_tag_drift_proof=proof)
+
+
+def test_rendered_model_rejects_missing_expected_image_after_proof(monkeypatch):
+    container, _source_ref, _observed_id = _install_tag_drift_docker_fixture(monkeypatch)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+    proof = module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+    original_run = module._run
+
+    def missing_expected(arguments, **kwargs):
+        if arguments[:3] == ["docker", "image", "inspect"] and arguments[-1] == IMAGE_ID:
+            return b""
+        return original_run(arguments, **kwargs)
+
+    monkeypatch.setattr(module, "_run", missing_expected)
+
+    with pytest.raises(module.VolumeTransitionError, match="source_tag_drift_expected_image_missing"):
+        module.rendered_model(original, REVISION, IMAGE_ID, source_tag_drift_proof=proof)
+
+
+def test_rendered_model_rejects_reuse_with_changed_captured_compose_bytes(monkeypatch):
+    container, _source_ref, _observed_id = _install_tag_drift_docker_fixture(monkeypatch)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+    proof = module._attest_observed_runner_tag_drift(original, container, IMAGE_ID)
+
+    with pytest.raises(module.VolumeTransitionError, match="source_tag_drift_proof_invalid"):
+        module.rendered_model(original + b"# changed captured source\n", REVISION,
+                              IMAGE_ID, source_tag_drift_proof=proof)
+
+
+def test_observed_runner_tag_drift_rechecks_same_container_around_image_inspection(monkeypatch):
+    source_ref = "docker.gitea.com/runner-images:ubuntu-latest"
+    first = _tag_drift_container()
+    second = _tag_drift_container()
+    second["Id"] = "e" * 64
+    compose_model = model()
+    compose_model["services"][module.SERVICE]["image"] = source_ref
+    inspect_count = 0
+
+    def run(arguments, *, data=None, **_kwargs):
+        nonlocal inspect_count
+        if arguments[:2] == ["docker", "inspect"]:
+            inspect_count += 1
+            return json.dumps([first if inspect_count == 1 else second]).encode("utf-8")
+        if arguments[:3] == ["docker", "image", "inspect"]:
+            reference = arguments[-1]
+            return (("sha256:" + "c" * 64) if reference == source_ref else IMAGE_ID).encode("ascii")
+        if arguments[-2:] == ["--format", "json"]:
+            return json.dumps(compose_model).encode("utf-8")
+        return b""
+
+    monkeypatch.setattr(module, "_run", run)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+
+    with pytest.raises(module.VolumeTransitionError, match="container_changed_during_tag_drift_check"):
+        module._attest_observed_runner_tag_drift(original, first, IMAGE_ID)
+
+
+def test_observed_runner_tag_drift_rejects_mount_change_during_tag_inspection(monkeypatch):
+    source_ref = "docker.gitea.com/runner-images:ubuntu-latest"
+    first = _tag_drift_container()
+    second = _tag_drift_container()
+    second["Mounts"][1]["RW"] = True
+    compose_model = model()
+    compose_model["services"][module.SERVICE]["image"] = source_ref
+    inspect_count = 0
+
+    def run(arguments, *, data=None, **_kwargs):
+        nonlocal inspect_count
+        if arguments[:2] == ["docker", "inspect"]:
+            inspect_count += 1
+            return json.dumps([first if inspect_count == 1 else second]).encode("utf-8")
+        if arguments[:3] == ["docker", "image", "inspect"]:
+            reference = arguments[-1]
+            return (("sha256:" + "c" * 64) if reference == source_ref else IMAGE_ID).encode("ascii")
+        if arguments[-2:] == ["--format", "json"]:
+            return json.dumps(compose_model).encode("utf-8")
+        return b""
+
+    monkeypatch.setattr(module, "_run", run)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+
+    with pytest.raises(module.VolumeTransitionError, match="config_alias_invalid"):
+        module._attest_observed_runner_tag_drift(original, first, IMAGE_ID)
+
+
+def test_observed_runner_tag_drift_rejects_config_change_during_tag_inspection(monkeypatch):
+    source_ref = "docker.gitea.com/runner-images:ubuntu-latest"
+    first = _tag_drift_container()
+    second = _tag_drift_container()
+    second["Config"]["User"] = "unexpected"
+    compose_model = model()
+    compose_model["services"][module.SERVICE]["image"] = source_ref
+    inspect_count = 0
+
+    def run(arguments, *, data=None, **_kwargs):
+        nonlocal inspect_count
+        if arguments[:2] == ["docker", "inspect"]:
+            inspect_count += 1
+            return json.dumps([first if inspect_count == 1 else second]).encode("utf-8")
+        if arguments[:3] == ["docker", "image", "inspect"]:
+            reference = arguments[-1]
+            return (("sha256:" + "c" * 64) if reference == source_ref else IMAGE_ID).encode("ascii")
+        if arguments[-2:] == ["--format", "json"]:
+            return json.dumps(compose_model).encode("utf-8")
+        return b""
+
+    monkeypatch.setattr(module, "_run", run)
+    original = b"name: gitea-deploy-runners\nservices:\n  deploy-control:\n    image: docker.gitea.com/runner-images:ubuntu-latest\n"
+
+    with pytest.raises(module.VolumeTransitionError, match="container_changed_during_tag_drift_check"):
+        module._attest_observed_runner_tag_drift(original, first, IMAGE_ID)
+
+
+def test_cli_explicit_tag_drift_switch_is_accepted_only_for_prepare(monkeypatch, tmp_path):
+    calls = []
+    monkeypatch.setattr(sys, "argv", [
+        "prepare_controller_hub_release_volume.py", "prepare",
+        "--stage-root", str(tmp_path), "--source-revision", REVISION,
+        "--source-tree", REVISION, "--controller-version", "2.1.13",
+        "--image-id", IMAGE_ID, "--allow-observed-runner-tag-drift",
+    ])
+    monkeypatch.setattr(
+        module, "prepare_promotion",
+        lambda *args, **kwargs: calls.append(kwargs) or tmp_path / "backup",
+    )
+
+    module.main()
+
+    assert len(calls) == 1
+    assert {key: value for key, value in calls[0].items() if key != "on_phase"} == {
+        "source_revision": REVISION,
+        "source_tree": REVISION,
+        "controller_version": "2.1.13",
+        "image_id": IMAGE_ID,
+        "allow_observed_runner_tag_drift": True,
+    }
+    assert callable(calls[0]["on_phase"])
+
+
+def test_cli_rejects_tag_drift_switch_for_non_prepare_operation(monkeypatch):
+    monkeypatch.setattr(sys, "argv", [
+        "prepare_controller_hub_release_volume.py", "activate",
+        "--backup", "/unused", "--allow-observed-runner-tag-drift",
+    ])
+
+    with pytest.raises(SystemExit, match="source_tag_recovery_operation_invalid"):
+        module.main()
+
+
+@pytest.mark.parametrize("allow_recovery", [False, True])
+def test_promotion_image_proof_failures_precede_persistent_volume_writes(
+    monkeypatch, tmp_path, allow_recovery
+):
+    original = b"captured source compose with exact original bytes\n"
+    current = _tag_drift_container()
+    old_revision = "c" * 40
+    config_tree = ({"projects.json": (b"{}", 0o600)}, {"": 0o755})
+    compose_model = model()
+    compose_model["services"][module.SERVICE]["image"] = "docker.gitea.com/runner-images:ubuntu-latest"
+    proof = module._ObservedRunnerTagDriftProof(
+        current["Id"], IMAGE_ID, module.OBSERVED_RUNNER_IMAGE_REF,
+        "sha256:" + "c" * 64, module._hash(original),
+        module._tag_drift_config_identity(current), (),
+    )
+    stage_validation = []
+    writes = []
+
+    monkeypatch.setattr(module, "_validate_private_stage", lambda *_args: None)
+    monkeypatch.setattr(module, "ensure_volume", lambda **_kwargs: None)
+    monkeypatch.setattr(module, "ensure_backup_volume", lambda **_kwargs: None)
+    monkeypatch.setattr(module, "_controller_image_inspect", lambda *_args, **_kwargs: current)
+    monkeypatch.setattr(module, "_snapshot_attested_controller", lambda *_args: (
+        {"controller-release.json": (b"captured release", 0o644)},
+        {"": 0o755},
+        {"sourceRevision": old_revision, "sourceTree": "d" * 40},
+    ))
+    monkeypatch.setattr(module, "_read_root_compose", lambda: (original, 0o600))
+    monkeypatch.setattr(module, "_attest_observed_runner_tag_drift", lambda *_args: proof)
+    monkeypatch.setattr(module, "_validate_previous_private_tls", lambda *_args: None)
+    monkeypatch.setattr(module, "_release_tree", lambda *_args: ({"deploy.py": (b"old", 0o644)}, {"": 0o755}))
+    monkeypatch.setattr(module, "_config_tree", lambda *_args, **_kwargs: config_tree)
+    monkeypatch.setattr(module, "_read_config_leaf", lambda: None)
+    monkeypatch.setattr(module, "_prepare_merged_tree", lambda *_args: ({"deploy.py": (b"new", 0o644)}, {"": 0o755}))
+    monkeypatch.setattr(module, "_validate_staged_configuration", lambda *_args: stage_validation.append(True))
+    monkeypatch.setattr(module, "_mirror_snapshot_tar", lambda: b"mirror")
+    monkeypatch.setattr(module, "_source_compose_model", lambda _original: (
+        compose_model, compose_model["services"][module.SERVICE]
+    ))
+    if allow_recovery:
+        def invalidate_after_stage(*_args):
+            assert stage_validation
+            raise module.VolumeTransitionError("controller_volume_source_tag_drift_changed")
+
+        monkeypatch.setattr(module, "_verify_tag_drift_proof", invalidate_after_stage)
+    else:
+        monkeypatch.setattr(
+            module, "_verify_source_compose_image",
+            lambda *_args: (_ for _ in ()).throw(
+                module.VolumeTransitionError("controller_volume_image_reference_changed")
+            ),
+        )
+    monkeypatch.setattr(module, "_verify_compose_image", lambda *_args: None)
+    monkeypatch.setattr(module, "_install_config_tree", lambda *_args: writes.append("config"))
+    monkeypatch.setattr(module, "_install_release_leaf", lambda *_args: writes.append("release"))
+
+    with pytest.raises(module.VolumeTransitionError):
+        module.prepare_promotion(
+            tmp_path, source_revision=REVISION, source_tree="e" * 40,
+            controller_version="2.1.13", image_id=IMAGE_ID,
+            allow_observed_runner_tag_drift=allow_recovery,
+        )
+
+    assert stage_validation
+    assert writes == []
+
+
+def test_promotion_revalidates_drift_proof_immediately_before_first_volume_write(
+    monkeypatch, tmp_path
+):
+    original = b"captured source compose with exact original bytes\n"
+    current = _tag_drift_container()
+    old_revision = "c" * 40
+    config_tree = ({"projects.json": (b"{}", 0o600)}, {"": 0o755})
+    stage_validation = []
+    proof = module._ObservedRunnerTagDriftProof(
+        current["Id"], IMAGE_ID, module.OBSERVED_RUNNER_IMAGE_REF,
+        "sha256:" + "c" * 64, module._hash(original),
+        module._tag_drift_config_identity(current), (),
+    )
+    writes = []
+    proof_checks = 0
+    image_checks = 0
+
+    def source_model(_original):
+        compose_model = model()
+        compose_model["services"][module.SERVICE]["image"] = module.OBSERVED_RUNNER_IMAGE_REF
+        return compose_model, compose_model["services"][module.SERVICE]
+
+    monkeypatch.setattr(module, "_validate_private_stage", lambda *_args: None)
+    monkeypatch.setattr(module, "ensure_volume", lambda **_kwargs: None)
+    monkeypatch.setattr(module, "ensure_backup_volume", lambda **_kwargs: None)
+    monkeypatch.setattr(module, "_controller_image_inspect", lambda *_args, **_kwargs: current)
+    monkeypatch.setattr(module, "_snapshot_attested_controller", lambda *_args: (
+        {"controller-release.json": (b"captured release", 0o644)},
+        {"": 0o755},
+        {"sourceRevision": old_revision, "sourceTree": "d" * 40},
+    ))
+    monkeypatch.setattr(module, "_read_root_compose", lambda: (original, 0o600))
+    monkeypatch.setattr(module, "_attest_observed_runner_tag_drift", lambda *_args: proof)
+    monkeypatch.setattr(module, "_validate_previous_private_tls", lambda *_args: None)
+    monkeypatch.setattr(module, "_release_tree", lambda *_args: ({"deploy.py": (b"old", 0o644)}, {"": 0o755}))
+    monkeypatch.setattr(module, "_config_tree", lambda *_args, **_kwargs: config_tree)
+    monkeypatch.setattr(module, "_read_config_leaf", lambda: config_tree)
+    monkeypatch.setattr(module, "_prepare_merged_tree", lambda *_args: ({"deploy.py": (b"new", 0o644)}, {"": 0o755}))
+    monkeypatch.setattr(module, "_validate_staged_configuration", lambda *_args: stage_validation.append(True))
+    monkeypatch.setattr(module, "_mirror_snapshot_tar", lambda: b"mirror")
+    monkeypatch.setattr(module, "_source_compose_model", source_model)
+    monkeypatch.setattr(module, "_compose", lambda *_args, **_kwargs: b"")
+
+    def verify_tag_proof(*_args):
+        nonlocal proof_checks
+        proof_checks += 1
+        assert stage_validation
+        if proof_checks == 4:
+            raise module.VolumeTransitionError("controller_volume_source_tag_drift_changed")
+
+    def verify_image(*_args):
+        nonlocal image_checks
+        image_checks += 1
+
+    monkeypatch.setattr(module, "_verify_tag_drift_proof", verify_tag_proof)
+    monkeypatch.setattr(module, "_verify_compose_image", verify_image)
+    monkeypatch.setattr(module, "_install_config_tree", lambda *_args: writes.append("config"))
+    monkeypatch.setattr(module, "_install_release_leaf", lambda *_args: writes.append("release"))
+
+    with pytest.raises(module.VolumeTransitionError, match="source_tag_drift_changed"):
+        module.prepare_promotion(
+            tmp_path, source_revision=REVISION, source_tree="e" * 40,
+            controller_version="2.1.13", image_id=IMAGE_ID,
+            allow_observed_runner_tag_drift=True,
+        )
+
+    assert proof_checks == 4
+    assert image_checks == 2
+    assert writes == []
+
 def test_config_snapshot_has_exact_allowlist_and_broker_token():
     entries = {
         "projects.json": (b"{}", 0o600),

backend/tests/test_controller_root_procedure_contract.py

@@ -81,6 +81,25 @@ def test_windows_wrapper_supplies_private_archive_and_exact_image_identity():
     assert '[[ "$archive_parent" =~ ^/tmp/itworx-controller-artifact-promoter-' in installer
 
 
+def test_observed_runner_tag_recovery_switch_is_fixed_and_prepare_only():
+    wrapper = (ROOT / "scripts/upgrade_deployment_controller.ps1").read_text(encoding="utf-8")
+    installer = (ROOT / "scripts/upgrade_deployment_controller.sh").read_text(encoding="utf-8")
+    preparer = (ROOT / "scripts/prepare_controller_hub_release_volume.py").read_text(encoding="utf-8")
+
+    assert "[switch]$AllowObservedRunnerTagDrift" in wrapper
+    assert '$recoveryArgument = if ($AllowObservedRunnerTagDrift)' in wrapper
+    assert "--allow-observed-runner-tag-drift" in wrapper
+    assert "OBSERVED_RUNNER_IMAGE_REF = 'docker.gitea.com/runner-images:ubuntu-latest'" in preparer
+    assert "--allow-observed-runner-tag-drift)" in installer
+    assert "recovery_arguments+=(--allow-observed-runner-tag-drift)" in installer
+    prepare_start = installer.index('backup="$(python3 "${work}/${PREPARER_NAME}" prepare')
+    prepare_stop = installer.index('python3 "${work}/${PREPARER_NAME}" stop --backup "$backup"')
+    prepare_command = installer[prepare_start:prepare_stop]
+    assert '"${recovery_arguments[@]}"' in prepare_command
+    assert "parser.add_argument('--allow-observed-runner-tag-drift', action='store_true')" in preparer
+    assert "if args.allow_observed_runner_tag_drift and args.operation != 'prepare':" in preparer
+
+
 def test_state_reconciliation_is_committed_only_after_volume_verification():
     installer = (ROOT / "scripts/upgrade_deployment_controller.sh").read_text(encoding="utf-8")
     verify = installer.index('python3 "${work}/${PREPARER_NAME}" verify --backup "$backup"')

backend/tests/test_deploy_broker.py

@@ -2445,6 +2445,30 @@ def _workmesh_product_project(tmp_path: Path) -> dict[str, object]:
     }
 
 
+def _agenttestlab_project(tmp_path: Path, containers: list[str] | None = None) -> dict[str, object]:
+    return {
+        "slug": "agenttestlab",
+        "branch": "main",
+        "source_dir": "/mnt/cache/appdata/gitea-deploy-runners/sources/agenttestlab",
+        "compose_project": "agenttestlab",
+        "compose_dir": "",
+        "compose_files": ["compose.inspect.yaml"],
+        "containers": list(containers or [
+            "agenttestlab-postgres-1",
+            "agenttestlab-redis-1",
+            "agenttestlab-api-1",
+            "agenttestlab-worker-1",
+            "agenttestlab-agenttestlab-1",
+        ]),
+        "exact_source": True,
+        "env_source": "/srv/example/appdata/agenttestlab/deploy.env",
+        "health_timeout": 300,
+        "verify_oci_revision": False,
+        "_repository_name": "jens/agenttestlab",
+        "_repository_path": str(tmp_path),
+    }
+
+
 def _workmesh_legacy_reattest_setup(tmp_path: Path):
     revision = "a" * 40
     project = _workmesh_product_project(tmp_path)
@@ -3100,6 +3124,17 @@ def test_workmesh_topology_transition_requires_exact_registry_and_recorded_recei
     assert deploy_module.workmesh_rollback_topology(wrong, previous) is wrong
 
 
+def test_snapshot_topology_preserves_workmesh_legacy_container_name_contract(
+    deploy_module, tmp_path
+):
+    project = _workmesh_product_project(tmp_path)
+    snapshot = {"containers": [{"name": name} for name in deploy_module.WORKMESH_LEGACY_CONTAINERS]}
+
+    restored = deploy_module.snapshot_rollback_topology(project, snapshot)
+
+    assert restored["containers"] == deploy_module.WORKMESH_LEGACY_CONTAINERS
+
+
 def test_workmesh_product_deploy_removes_only_reviewed_compose_orphan(
     deploy_module, tmp_path, monkeypatch
 ):
@@ -3272,6 +3307,482 @@ def test_workmesh_product_snapshot_retains_authorized_five_container_rollback(
     assert deploy_module.durable_rollback_snapshot(project, "jens/workmesh", new_revision) == snapshot
 
 
+def test_agenttestlab_deploy_accepts_only_the_reviewed_five_or_six_container_topology(
+    deploy_module, tmp_path, monkeypatch
+):
+    project = _agenttestlab_project(tmp_path, deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS)
+    commands = []
+    monkeypatch.setattr(deploy_module, "deployment_environment", lambda *_: {})
+    monkeypatch.setattr(deploy_module, "compose_command", lambda *_: (tmp_path, ["docker", "compose"]))
+    monkeypatch.setattr(deploy_module, "prepare_compose_migration", lambda *_: [])
+    monkeypatch.setattr(deploy_module, "run", lambda command, **_: commands.append(command))
+    observed = [
+        set(deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS),
+        set(deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS),
+        set(deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS) | {"agenttestlab-unreviewed-1"},
+    ]
+    monkeypatch.setattr(
+        deploy_module, "compose_project_container_names", lambda *_: observed.pop(0)
+    )
+
+    deploy_module.deploy_project(project, "a" * 40, "jens/agenttestlab")
+    deploy_module.deploy_project(project, "b" * 40, "jens/agenttestlab")
+    with pytest.raises(deploy_module.DeployError, match="topology changed"):
+        deploy_module.deploy_project(project, "c" * 40, "jens/agenttestlab")
+
+    assert commands == [
+        ["docker", "compose", "config", "--quiet"],
+        ["docker", "compose", "build"],
+        ["docker", "compose", "up", "--detach", "--no-build"],
+        ["docker", "compose", "config", "--quiet"],
+        ["docker", "compose", "build"],
+        ["docker", "compose", "up", "--detach", "--no-build"],
+        ["docker", "compose", "config", "--quiet"],
+        ["docker", "compose", "build"],
+    ]
+
+
+def test_agenttestlab_transition_preserves_registry_and_accepts_verified_post_deploy_retry(
+    deploy_module, tmp_path, monkeypatch
+):
+    entry = _agenttestlab_project(tmp_path)
+    entry.pop("_repository_name")
+    entry.pop("_repository_path")
+    original = {"jens/agenttestlab": entry, "jens/other": {"slug": "other", "policy": ["unchanged"]}}
+    original_bytes = json.dumps(original).encode()
+    registry = tmp_path / "projects.json"
+    registry.write_bytes(original_bytes)
+    registry.chmod(0o600)
+    revision = "b" * 40
+    protected_head = revision
+    live_revision = "a" * 40
+    tree = "c" * 40
+    state = {
+        "schema_version": 3,
+        "repository": "jens/agenttestlab",
+        "live_revision": live_revision,
+        "live_status": "healthy",
+        "live_receipt": {
+            "build": {"tree": tree},
+            "containers": {name: {"image_id": "sha256:" + "1" * 64}
+                           for name in deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS},
+        },
+        "live_artifacts": {
+            "containers": {name: {"image_id": "sha256:" + "1" * 64}
+                           for name in deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS},
+        },
+    }
+    project = {**entry, "_repository_name": "jens/agenttestlab", "_repository_path": str(tmp_path)}
+    monkeypatch.setattr(deploy_module, "CONFIG_PATH", registry)
+    monkeypatch.setattr(deploy_module, "STATE_ROOT", tmp_path)
+    monkeypatch.setattr(deploy_module, "_root_owned", lambda _path: True)
+    monkeypatch.setattr(deploy_module, "_fsync_directory", lambda _path: None)
+    monkeypatch.setattr(deploy_module, "load_projects", lambda: json.loads(registry.read_text()))
+    monkeypatch.setattr(deploy_module, "validate_commit", lambda _repo, _branch, value: value)
+
+    def git_or_compose_output(command, **_kwargs):
+        if command[-2:] == ["rev-parse", "refs/heads/main"]:
+            return protected_head
+        return "\n".join(deploy_module.AGENTTESTLAB_SERVICES)
+
+    monkeypatch.setattr(deploy_module, "output", git_or_compose_output)
+    monkeypatch.setattr(deploy_module, "read_state", lambda *_: state)
+    monkeypatch.setattr(deploy_module, "state_live_artifacts", lambda _state: state["live_artifacts"])
+    monkeypatch.setattr(
+        deploy_module, "compose_project_container_names",
+        lambda *_: set(deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS),
+    )
+    monkeypatch.setattr(deploy_module, "exact_source_matches_revision", lambda *_: True)
+    monkeypatch.setattr(deploy_module, "compose_command", lambda *_: (tmp_path, ["docker", "compose"]))
+    monkeypatch.setattr(deploy_module, "deployment_environment", lambda *_: {})
+    verified = []
+    monkeypatch.setattr(
+        deploy_module,
+        "verified_current_receipt",
+        lambda _repo, target, _revision, _tree: verified.append(list(target["containers"])) or {"verified": True},
+    )
+
+    result = deploy_module.transition_agenttestlab_registry(
+        original, project, tmp_path, revision
+    )
+
+    expected = copy.deepcopy(original)
+    expected["jens/agenttestlab"]["containers"] = deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS.copy()
+    assert result == "product-topology-registered"
+    assert json.loads(registry.read_bytes()) == expected
+    assert expected["jens/agenttestlab"]["env_source"] == entry["env_source"]
+    assert expected["jens/other"] == original["jens/other"]
+    backup = deploy_module._agenttestlab_registry_backup_path()
+    assert backup.read_bytes() == original_bytes
+    monkeypatch.setattr(
+        deploy_module,
+        "_root_only_backup",
+        lambda path, payload: None
+        if path.read_bytes() == payload
+        else pytest.fail("idempotent transition changed the saved registry backup"),
+    )
+
+    product_project = {**expected["jens/agenttestlab"], "_repository_name": "jens/agenttestlab"}
+    result = deploy_module.transition_agenttestlab_registry(
+        expected, product_project, tmp_path, revision
+    )
+    assert result == "already-registered"
+
+    restored_artifacts = []
+    monkeypatch.setattr(
+        deploy_module,
+        "deployment_artifacts",
+        lambda target: restored_artifacts.append(list(target["containers"])) or {
+            "containers": {name: {"image_id": "sha256:" + "1" * 64}
+                           for name in target["containers"]}
+        },
+    )
+    failed_state = deploy_module.resolve_failed_live_state(
+        state,
+        {"revision": live_revision, "receipt": state["live_receipt"]},
+        product_project,
+        production_started=True,
+        failure_stage="deployment",
+        restore_error=None,
+    )
+    assert restored_artifacts == [deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS]
+    assert set(failed_state["live_artifacts"]["containers"]) == set(deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS)
+    assert json.loads(registry.read_bytes()) == expected
+
+    later_revision = "d" * 40
+    protected_head = later_revision
+    state["live_revision"] = later_revision
+    state["live_receipt"]["containers"] = {
+        name: {"image_id": "sha256:" + "2" * 64}
+        for name in deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS
+    }
+    state["live_artifacts"]["containers"] = copy.deepcopy(state["live_receipt"]["containers"])
+    monkeypatch.setattr(
+        deploy_module, "compose_project_container_names",
+        lambda *_: set(deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS),
+    )
+    result = deploy_module.transition_agenttestlab_registry(
+        expected, product_project, tmp_path, later_revision
+    )
+
+    assert result == "already-registered"
+    assert registry.read_bytes() != original_bytes
+    assert json.loads(registry.read_bytes()) == expected
+    assert verified == [
+        deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS,
+        deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS,
+        deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS,
+    ]
+
+    saved = deploy_module.agenttestlab_registry_rollback_backup(expected)
+    deploy_module.restore_agenttestlab_registry(expected, saved)
+    assert saved == original_bytes
+    assert registry.read_bytes() != original_bytes
+    assert json.loads(registry.read_bytes()) == original
+
+    legacy_project = {
+        **original["jens/agenttestlab"],
+        "_repository_name": "jens/agenttestlab",
+        "_repository_path": str(tmp_path),
+    }
+    state["live_revision"] = live_revision
+    state["live_receipt"]["containers"] = {
+        name: {"image_id": "sha256:" + "1" * 64}
+        for name in deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS
+    }
+    state["live_artifacts"]["containers"] = copy.deepcopy(state["live_receipt"]["containers"])
+    monkeypatch.setattr(
+        deploy_module, "compose_project_container_names",
+        lambda *_: set(deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS),
+    )
+    result = deploy_module.transition_agenttestlab_registry(
+        original, legacy_project, tmp_path, later_revision
+    )
+    assert result == "product-topology-registered"
+    assert json.loads(registry.read_bytes()) == expected
+    assert backup.read_bytes() == original_bytes
+
+
+def test_agenttestlab_transition_rejects_wrong_target_source_and_rendered_services(
+    deploy_module, tmp_path, monkeypatch
+):
+    entry = _agenttestlab_project(tmp_path)
+    entry.pop("_repository_name")
+    entry.pop("_repository_path")
+    projects = {"jens/agenttestlab": entry}
+    original_bytes = (json.dumps(projects, indent=2) + "\n").encode()
+    registry = tmp_path / "projects.json"
+    registry.write_bytes(original_bytes)
+    project = {**entry, "_repository_name": "jens/agenttestlab"}
+    revision = "b" * 40
+    tree = "c" * 40
+    state = {
+        "live_revision": "a" * 40,
+        "live_status": "healthy",
+        "live_receipt": {
+            "build": {"tree": tree},
+            "containers": {name: {"image_id": "sha256:" + "1" * 64}
+                           for name in deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS},
+        },
+        "live_artifacts": {
+            "containers": {name: {"image_id": "sha256:" + "1" * 64}
+                           for name in deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS},
+        },
+    }
+    compose_services = [*deploy_module.AGENTTESTLAB_SERVICES]
+    source_matches = True
+    monkeypatch.setattr(deploy_module, "CONFIG_PATH", registry)
+    monkeypatch.setattr(deploy_module, "STATE_ROOT", tmp_path)
+    monkeypatch.setattr(deploy_module, "_fsync_directory", lambda _path: None)
+    monkeypatch.setattr(deploy_module, "load_projects", lambda: json.loads(registry.read_text()))
+    monkeypatch.setattr(deploy_module, "validate_commit", lambda _repo, _branch, value: value)
+
+    def output_stub(command, **_kwargs):
+        if command[-2:] == ["rev-parse", "refs/heads/main"]:
+            return revision
+        return "\n".join(compose_services)
+
+    monkeypatch.setattr(deploy_module, "output", output_stub)
+    monkeypatch.setattr(deploy_module, "read_state", lambda *_: state)
+    monkeypatch.setattr(deploy_module, "state_live_artifacts", lambda _state: state["live_artifacts"])
+    monkeypatch.setattr(
+        deploy_module, "compose_project_container_names",
+        lambda *_: set(deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS),
+    )
+    monkeypatch.setattr(
+        deploy_module, "exact_source_matches_revision",
+        lambda *_: source_matches,
+    )
+    monkeypatch.setattr(deploy_module, "compose_command", lambda *_: (tmp_path, ["docker", "compose"]))
+    monkeypatch.setattr(deploy_module, "deployment_environment", lambda *_: {})
+    monkeypatch.setattr(deploy_module, "verified_current_receipt", lambda *_: {"verified": True})
+
+    with pytest.raises(deploy_module.DeployError, match="fixed main target"):
+        deploy_module.transition_agenttestlab_registry(
+            projects,
+            {**project, "_repository_name": "other/agenttestlab"},
+            tmp_path,
+            revision,
+        )
+    with pytest.raises(deploy_module.DeployError, match="fixed main target"):
+        deploy_module.transition_agenttestlab_registry(
+            projects,
+            {**project, "source_dir": str(tmp_path)},
+            tmp_path,
+            revision,
+        )
+
+    source_matches = False
+    with pytest.raises(deploy_module.DeployError, match="prepared source"):
+        deploy_module.transition_agenttestlab_registry(projects, project, tmp_path, revision)
+    source_matches = True
+    compose_services.append("unreviewed")
+    with pytest.raises(deploy_module.DeployError, match="six-service topology"):
+        deploy_module.transition_agenttestlab_registry(projects, project, tmp_path, revision)
+    compose_services.pop()
+    state["live_artifacts"]["containers"][deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS[0]]["image_id"] = "sha256:" + "2" * 64
+    with pytest.raises(deploy_module.DeployError, match="artifact identity"):
+        deploy_module.transition_agenttestlab_registry(projects, project, tmp_path, revision)
+
+    assert registry.read_bytes() == original_bytes
+    assert not deploy_module._agenttestlab_registry_backup_path().exists()
+
+
+def test_agenttestlab_automatic_failure_and_explicit_rollback_keep_distinct_registry_semantics(
+    deploy_module, tmp_path, monkeypatch
+):
+    legacy = _agenttestlab_project(tmp_path)
+    product = {**legacy, "containers": deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS.copy()}
+    previous_receipt = {"containers": {
+        name: {"image_id": "sha256:" + "1" * 64}
+        for name in deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS
+    }}
+    assert deploy_module.rollback_topology(product, {"live_receipt": previous_receipt})["containers"] == deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS
+    assert deploy_module.rollback_topology(
+        product,
+        {"live_receipt": {"containers": {
+            name: {"image_id": "sha256:" + "2" * 64}
+            for name in deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS
+        }}},
+    ) is product
+
+    observed = []
+    monkeypatch.setattr(
+        deploy_module,
+        "deployment_artifacts",
+        lambda target: observed.append(list(target["containers"])) or {
+            "containers": {name: {"image_id": "sha256:" + "1" * 64}
+                           for name in target["containers"]}
+        },
+    )
+    restored = deploy_module.resolve_failed_live_state(
+        {"live_revision": "a" * 40, "live_status": "healthy", "live_receipt": previous_receipt},
+        {"revision": "a" * 40, "receipt": previous_receipt},
+        product,
+        production_started=True,
+        failure_stage="deployment",
+        restore_error=None,
+    )
+    assert observed == [deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS]
+    assert set(restored["live_artifacts"]["containers"]) == set(deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS)
+
+
+@pytest.mark.parametrize("drift", [None, "image", "build", "image_ref"])
+def test_agenttestlab_snapshot_restore_allows_only_container_id_recreation(
+    deploy_module, tmp_path, monkeypatch, drift
+):
+    previous_revision = "a" * 40
+    failed_revision = "d" * 40
+    legacy = _agenttestlab_project(tmp_path)
+    legacy["_repository_name"] = "jens/agenttestlab"
+    receipt = _receipt(legacy, previous_revision, tree="c" * 40)
+    project = {**legacy, "containers": deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS.copy()}
+    project["_repository_path"] = str(tmp_path)
+    snapshot = {
+        "revision": previous_revision,
+        "receipt": receipt,
+        "containers": [
+            {
+                "name": name,
+                "container_id": receipt["containers"][name]["container_id"],
+                "image_id": receipt["containers"][name]["image_id"],
+                "image_ref": receipt["containers"][name]["image_ref"],
+                "rollback_ref": f"gitea-deploy-rollback/agenttestlab-{index}:{previous_revision[:12]}",
+            }
+            for index, name in enumerate(deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS)
+        ],
+    }
+    observed_projects = []
+    observed_receipts = []
+    monkeypatch.setattr(deploy_module, "run", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(deploy_module, "inspect_container", lambda _name: None)
+    snapshot_images = {item["image_ref"]: item["image_id"] for item in snapshot["containers"]}
+    monkeypatch.setattr(
+        deploy_module, "inspect_image", lambda reference: {"Id": snapshot_images[reference]}
+    )
+    monkeypatch.setattr(deploy_module, "repository_path_from_project", lambda _project: tmp_path)
+    monkeypatch.setattr(deploy_module, "sync_source", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(deploy_module, "copy_environment", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        deploy_module,
+        "remove_attempt_only_compose_containers",
+        lambda target, _snapshot: observed_projects.append(list(target["containers"])),
+    )
+    monkeypatch.setattr(deploy_module, "deployment_environment", lambda *_args: {})
+    monkeypatch.setattr(
+        deploy_module,
+        "restore_snapshot_runtime",
+        lambda target, *_args, **_kwargs: observed_projects.append(list(target["containers"])),
+    )
+    monkeypatch.setattr(
+        deploy_module,
+        "validate_recorded_receipt",
+        lambda target, saved, *_args: observed_projects.append(list(target["containers"]))
+        or saved["build"],
+    )
+    monkeypatch.setattr(deploy_module, "cleanup_rollback_image_tags", lambda *_args: None)
+
+    def verify(target, _build, *, expected_image_ids):
+        observed_projects.append(list(target["containers"]))
+        observed = copy.deepcopy(receipt)
+        for item in observed["containers"].values():
+            item["container_id"] = "3" * 64
+        if drift == "image":
+            observed["containers"][deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS[0]]["image_id"] = "sha256:" + "9" * 64
+        elif drift == "build":
+            observed["build"]["builtAt"] = "2026-10-02T12:00:00.000Z"
+        elif drift == "image_ref":
+            observed["containers"][deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS[0]]["image_ref"] = "unexpected:production"
+        observed_receipts.append(observed)
+        return observed
+
+    monkeypatch.setattr(deploy_module, "verify_deployment", verify)
+
+    if drift is None:
+        deploy_module.restore_compose_snapshot(project, snapshot, "jens/agenttestlab", failed_revision)
+        assert project["_rollback_receipt"] == observed_receipts[0]
+        assert project["_rollback_receipt"]["containers"][
+            deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS[0]
+        ]["container_id"] == "3" * 64
+        assert observed_projects == [
+            deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS,
+            deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS,
+            deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS,
+            deploy_module.AGENTTESTLAB_LEGACY_CONTAINERS,
+        ]
+    else:
+        with pytest.raises(deploy_module.DeployError, match="Rollback runtime no longer matches"):
+            deploy_module.restore_compose_snapshot(
+                project, snapshot, "jens/agenttestlab", failed_revision
+            )
+        assert "_rollback_receipt" not in project
+
+
+def test_agenttestlab_registry_replacement_failure_leaves_original_bytes_intact(
+    deploy_module, tmp_path, monkeypatch
+):
+    original = {"jens/agenttestlab": _agenttestlab_project(tmp_path)}
+    original["jens/agenttestlab"].pop("_repository_name")
+    original["jens/agenttestlab"].pop("_repository_path")
+    original_bytes = (json.dumps(original, indent=2) + "\n").encode()
+    registry = tmp_path / "projects.json"
+    registry.write_bytes(original_bytes)
+    registry.chmod(0o600)
+    expected = copy.deepcopy(original)
+    expected["jens/agenttestlab"]["containers"] = deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS.copy()
+    replace = os.replace
+    calls = 0
+
+    def fail_registry_replace(source, destination):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise OSError("synthetic atomic replacement failure")
+        replace(source, destination)
+
+    monkeypatch.setattr(deploy_module, "CONFIG_PATH", registry)
+    monkeypatch.setattr(deploy_module, "load_projects", lambda: json.loads(registry.read_text()))
+    monkeypatch.setattr(deploy_module, "_root_owned", lambda _path: True)
+    monkeypatch.setattr(deploy_module.os, "replace", fail_registry_replace)
+
+    with pytest.raises(OSError, match="synthetic atomic replacement failure"):
+        deploy_module._replace_agenttestlab_registry_atomically(
+            original, original_bytes, expected
+        )
+
+    assert registry.read_bytes() == original_bytes
+    assert list(tmp_path.glob(".projects.agenttestlab.*.tmp")) == []
+
+
+def test_agenttestlab_registry_replacement_rejects_group_readable_registry(
+    deploy_module, tmp_path, monkeypatch
+):
+    if os.name == "nt":
+        pytest.skip("POSIX registry permission checks are unavailable on Windows")
+
+    original = {"jens/agenttestlab": _agenttestlab_project(tmp_path)}
+    original["jens/agenttestlab"].pop("_repository_name")
+    original["jens/agenttestlab"].pop("_repository_path")
+    original_bytes = (json.dumps(original, indent=2) + "\n").encode()
+    registry = tmp_path / "projects.json"
+    registry.write_bytes(original_bytes)
+    registry.chmod(0o644)
+    expected = copy.deepcopy(original)
+    expected["jens/agenttestlab"]["containers"] = deploy_module.AGENTTESTLAB_PRODUCT_CONTAINERS.copy()
+
+    monkeypatch.setattr(deploy_module, "CONFIG_PATH", registry)
+    monkeypatch.setattr(deploy_module, "load_projects", lambda: json.loads(registry.read_text()))
+    monkeypatch.setattr(deploy_module, "_root_owned", lambda _path: True)
+
+    with pytest.raises(deploy_module.DeployError, match="private root-owned file"):
+        deploy_module._replace_agenttestlab_registry_atomically(
+            original, original_bytes, expected
+        )
+
+    assert registry.read_bytes() == original_bytes
+    assert registry.stat().st_mode & 0o777 == 0o644
+
+
 def test_deployment_environment_exports_verified_build_date(deploy_module, tmp_path):
     project = _project(tmp_path)
     repository = "jens/unraid-appops-gateway"

backend/tests/test_runner_fleet_cutover_safety.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import importlib.util
+import hashlib
 import json
 import shlex
 import shutil
@@ -19,6 +20,7 @@ SCRIPT_PATH = ROOT / "scripts" / "upgrade_ci_runner_fleet_v2.sh"
 PRIMARY = "itworx-security-runner"
 SECONDARY = "itworx-validation-runner-2"
 LEGACY = "itworx-unraid-deploy"
+WINDOWS_NATIVE = "itworx-windows11-native"
 OTHER_FIXED_RUNNERS = (
     "itworx-validation-runner-3",
     "itworx-docker-runner",
@@ -358,6 +360,280 @@ def _bash_function(name: str) -> str:
     return script[start:end]
 
 
+def _labels_digest(labels: list[object]) -> str:
+    canonical = json.dumps(
+        labels, sort_keys=True, separators=(",", ":"), ensure_ascii=False
+    ).encode("utf-8")
+    return hashlib.sha256(canonical).hexdigest()
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+@pytest.mark.parametrize(
+    ("rows", "total_count", "expected_status", "expected_snapshot"),
+    [
+        (
+            [
+                {
+                    "id": 41,
+                    "name": WINDOWS_NATIVE,
+                    "busy": False,
+                    "disabled": False,
+                    "labels": ["windows-validation"],
+                },
+                {
+                    "id": 42,
+                    "name": "itworx-security-runner",
+                    "busy": False,
+                    "disabled": True,
+                    "labels": [],
+                },
+            ],
+            2,
+            0,
+            f"41:false:{_labels_digest(['windows-validation'])}",
+        ),
+        (
+            [
+                {
+                    "id": 42,
+                    "name": "itworx-security-runner",
+                    "busy": False,
+                    "disabled": True,
+                    "labels": [],
+                }
+            ],
+            1,
+            3,
+            "",
+        ),
+        (
+            [
+                {
+                    "id": 41,
+                    "name": WINDOWS_NATIVE,
+                    "busy": False,
+                    "disabled": False,
+                    "labels": [],
+                }
+            ],
+            2,
+            1,
+            "",
+        ),
+        (
+            [
+                {
+                    "id": 41,
+                    "name": WINDOWS_NATIVE,
+                    "busy": False,
+                    "disabled": False,
+                    "labels": [],
+                },
+                {
+                    "id": 41,
+                    "name": "other-runner",
+                    "busy": False,
+                    "disabled": False,
+                    "labels": [],
+                },
+            ],
+            2,
+            1,
+            "",
+        ),
+        (
+            [
+                {
+                    "id": 41,
+                    "name": WINDOWS_NATIVE,
+                    "busy": False,
+                    "disabled": False,
+                    "labels": [],
+                },
+                {
+                    "id": 42,
+                    "name": WINDOWS_NATIVE,
+                    "busy": False,
+                    "disabled": True,
+                    "labels": [],
+                },
+            ],
+            2,
+            1,
+            "",
+        ),
+        (
+            [
+                {
+                    "id": index + 1,
+                    "name": WINDOWS_NATIVE if index == 0 else f"other-{index}",
+                    "busy": False,
+                    "disabled": False,
+                    "labels": [],
+                }
+                for index in range(100)
+            ],
+            100,
+            1,
+            "",
+        ),
+        (
+            [
+                {
+                    "id": 41,
+                    "name": WINDOWS_NATIVE,
+                    "busy": "false",
+                    "disabled": False,
+                    "labels": [],
+                }
+            ],
+            1,
+            1,
+            "",
+        ),
+    ],
+    ids=(
+        "known-registration",
+        "optional-registration-absent",
+        "incomplete-inventory",
+        "duplicate-id",
+        "duplicate-name",
+        "bounded-inventory-limit",
+        "malformed-inventory-row",
+    ),
+)
+def test_windows_snapshot_requires_a_complete_unique_inventory(
+    tmp_path, rows, total_count, expected_status, expected_snapshot
+):
+    token_file = tmp_path / "token"
+    token_file.write_text("synthetic-token", encoding="utf-8")
+    inventory = json.dumps({"total_count": total_count, "runners": rows})
+    body = "\n".join(
+        [
+            f"ADMIN_TOKEN_FILE={shlex.quote(str(token_file))}",
+            f"WINDOWS_NATIVE={shlex.quote(WINDOWS_NATIVE)}",
+            f"MOCK_INVENTORY={shlex.quote(inventory)}",
+            'curl() { printf "%s" "$MOCK_INVENTORY"; }',
+            _bash_function("windows_native_snapshot"),
+            "windows_native_snapshot",
+        ]
+    )
+    result = subprocess.run(
+        ["bash"], input=body.encode(), capture_output=True, check=False
+    )
+    assert result.returncode == expected_status, result.stderr.decode(errors="replace")
+    assert result.stdout.decode(errors="replace").strip() == expected_snapshot
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+def test_windows_snapshot_fails_closed_on_inventory_api_error(tmp_path):
+    token_file = tmp_path / "token"
+    token_file.write_text("synthetic-token", encoding="utf-8")
+    body = "\n".join(
+        [
+            f"ADMIN_TOKEN_FILE={shlex.quote(str(token_file))}",
+            f"WINDOWS_NATIVE={shlex.quote(WINDOWS_NATIVE)}",
+            'curl() { return 22; }',
+            _bash_function("windows_native_snapshot"),
+            "windows_native_snapshot",
+        ]
+    )
+    result = subprocess.run(
+        ["bash"], input=body.encode(), capture_output=True, check=False
+    )
+    assert result.returncode == 1
+    assert result.stdout == b""
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+@pytest.mark.parametrize(
+    ("prior_disabled", "expected_patches"),
+    [("false", ["patch:41:true", "patch:41:false"]), ("true", [])],
+    ids=("restore-prior-enabled", "preserve-prior-paused"),
+)
+def test_windows_admission_capture_and_restore_preserve_prior_state(
+    prior_disabled, expected_patches
+):
+    labels = _labels_digest(["windows-validation"])
+    body = "\n".join(
+        [
+            f"WINDOWS_NATIVE={shlex.quote(WINDOWS_NATIVE)}",
+            'windows_native_saved=0; windows_native_id=""; windows_native_prior_disabled=""; windows_native_labels_digest=""',
+            'mock_id=41; mock_disabled="' + prior_disabled + '"; mock_labels="' + labels + '"',
+            'windows_native_snapshot() { printf "%s:%s:%s\\n" "$mock_id" "$mock_disabled" "$mock_labels"; }',
+            'set_runner_disabled_by_id() { printf "patch:%s:%s\\n" "$1" "$2"; mock_disabled="$2"; }',
+            _bash_function("capture_windows_native_admission"),
+            _bash_function("restore_windows_native_admission"),
+            "capture_windows_native_admission",
+            '[[ "$windows_native_saved" == 1 ]] || exit 91',
+            '[[ "$mock_disabled" == true ]] || exit 92',
+            "restore_windows_native_admission",
+            'printf "final:%s:saved:%s\\n" "$mock_disabled" "$windows_native_saved"',
+        ]
+    )
+    result = subprocess.run(
+        ["bash"], input=body.encode(), capture_output=True, check=False
+    )
+    assert result.returncode == 0, result.stderr.decode(errors="replace")
+    output = result.stdout.decode(errors="replace").splitlines()
+    assert output[:-1] == expected_patches
+    assert output[-1] == f"final:{prior_disabled}:saved:0"
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+@pytest.mark.parametrize(
+    "snapshot",
+    [
+        "99:true:{digest}",
+        "41:true:{other_digest}",
+    ],
+    ids=("registration-replaced", "labels-drifted"),
+)
+def test_windows_admission_restore_rejects_identity_or_label_drift(snapshot):
+    expected_digest = _labels_digest(["windows-validation"])
+    other_digest = _labels_digest(["changed-label"])
+    rendered = snapshot.format(digest=expected_digest, other_digest=other_digest)
+    body = "\n".join(
+        [
+            'windows_native_saved=1; windows_native_id=41; windows_native_prior_disabled=false',
+            f'windows_native_labels_digest="{expected_digest}"',
+            f'MOCK_SNAPSHOT="{rendered}"',
+            'windows_native_snapshot() { printf "%s\\n" "$MOCK_SNAPSHOT"; }',
+            'set_runner_disabled_by_id() { echo unexpected-patch; }',
+            _bash_function("restore_windows_native_admission"),
+            'if restore_windows_native_admission; then exit 93; else status=$?; fi',
+            'printf "status:%s:saved:%s\\n" "$status" "$windows_native_saved"',
+        ]
+    )
+    result = subprocess.run(
+        ["bash"], input=body.encode(), capture_output=True, check=False
+    )
+    assert result.returncode == 0, result.stderr.decode(errors="replace")
+    assert result.stdout.decode(errors="replace").strip() == "status:1:saved:1"
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+def test_windows_admission_restore_failure_keeps_recovery_intent():
+    labels = _labels_digest(["windows-validation"])
+    body = "\n".join(
+        [
+            'windows_native_saved=1; windows_native_id=41; windows_native_prior_disabled=false',
+            f'windows_native_labels_digest="{labels}"',
+            f'mock_snapshot="41:true:{labels}"',
+            'windows_native_snapshot() { printf "%s\\n" "$mock_snapshot"; }',
+            'set_runner_disabled_by_id() { return 1; }',
+            _bash_function("restore_windows_native_admission"),
+            'if restore_windows_native_admission; then exit 94; else status=$?; fi',
+            'printf "status:%s:saved:%s\\n" "$status" "$windows_native_saved"',
+        ]
+    )
+    result = subprocess.run(
+        ["bash"], input=body.encode(), capture_output=True, check=False
+    )
+    assert result.returncode == 0, result.stderr.decode(errors="replace")
+    assert result.stdout.decode(errors="replace").strip() == "status:1:saved:1"
+
+
 @pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
 @pytest.mark.parametrize(
     ("proof", "executor_status", "allowed"),
@@ -458,20 +734,25 @@ def test_dind_idle_shell_requires_successful_empty_docker_inventory(
     assert (result.returncode == 0) is allowed, result.stderr.decode(errors="replace")
 
 
-@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+def _restore_exit_harness(
-@pytest.mark.parametrize(
+    *,
-    ("cutover_started", "dispatch_started"),
+    cutover_started: int,
-    [(0, 0), (1, 0), (1, 1)],
+    dispatch_started: int,
-    ids=("before-cutover", "cutover-before-dispatch", "after-dispatch"),
+    exit_status: int,
-)
+    fail_first_cp: bool = False,
-def test_failure_restores_only_safe_runner_phase(cutover_started, dispatch_started):
+    fail_windows_patch: bool = False,
+):
+    labels = _labels_digest(["windows-validation"])
     body = "\n".join(
         [
+            "set -Eeuo pipefail",
             "backup=/synthetic-backup",
             "COMPOSE_FILE=/synthetic-compose",
             "PRIMARY_CONFIG=/synthetic-primary",
             "SECONDARY_CONFIG=/synthetic-secondary",
             "ROOT=/synthetic-root",
+            "RUNNER_GUARD_SCRIPT=/synthetic-runner-guard",
+            "HOST_GUARD_SCRIPT=/synthetic-host-guard",
             "guard_changed=0",
             "host_guard_changed=0",
             f"cutover_started={cutover_started}",
@@ -480,30 +761,132 @@ def test_failure_restores_only_safe_runner_phase(cutover_started, dispatch_start
             'registered=("new-registration")',
             'fenced=("itworx-security-runner:false")',
             "created_paths=()",
+            "windows_native_saved=1",
+            "windows_native_id=41",
+            "windows_native_prior_disabled=false",
+            f'windows_native_labels_digest="{labels}"',
+            "mock_windows_disabled=true",
+            f"fail_first_cp={int(fail_first_cp)}",
+            f"fail_windows_patch={int(fail_windows_patch)}",
+            "cp_calls=0",
             "exec 3>&1",
-            'cp() { printf "cp:%s\\n" "$*"; }',
+            'cp() { printf "cp:%s\\n" "$*" >&3; ((cp_calls += 1)); if ((fail_first_cp && cp_calls == 1)); then return 1; fi; }',
             'docker() { printf "docker:%s\\n" "$*" >&3; }',
-            'delete_runner() { printf "delete:%s\\n" "$*"; }',
+            'delete_runner() { printf "delete:%s\\n" "$*" >&3; }',
-            'set_runner_disabled() { printf "restore-fence:%s:%s\\n" "$1" "$2"; }',
+            'set_runner_disabled() { printf "restore-fence:%s:%s\\n" "$1" "$2" >&3; }',
-            'stop_capacity_daemon() { printf "stop-daemon\\n"; }',
+            'set_runner_disabled_by_id() { printf "windows-patch:%s:%s\\n" "$1" "$2" >&3; if ((fail_windows_patch)); then return 1; fi; mock_windows_disabled="$2"; }',
-            'rm() { printf "rm:%s\\n" "$*"; }',
+            'windows_native_snapshot() { printf "windows-snapshot\\n" >&3; printf "41:%s:%s\\n" "$mock_windows_disabled" "$windows_native_labels_digest"; }',
+            'stop_capacity_daemon() { printf "stop-daemon\\n" >&3; }',
+            'rm() { printf "rm:%s\\n" "$*" >&3; }',
+            _bash_function("restore_managed_failure"),
+            _bash_function("restore_windows_native_admission"),
             _bash_function("restore"),
-            "false",
+            "trap restore EXIT",
-            "restore",
+            f"exit {exit_status}",
         ]
     )
-    result = subprocess.run(
+    return subprocess.run(
         ["bash"], input=body.encode(), capture_output=True, check=False
     )
-    output = result.stdout.decode(errors="replace")
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+@pytest.mark.parametrize(
+    ("cutover_started", "dispatch_started"),
+    [(0, 0), (1, 0), (1, 1)],
+    ids=("before-cutover", "cutover-before-dispatch", "after-dispatch"),
+)
+def test_exit_cleanup_restores_windows_after_managed_recovery(
+    cutover_started, dispatch_started
+):
+    result = _restore_exit_harness(
+        cutover_started=cutover_started,
+        dispatch_started=dispatch_started,
+        exit_status=1,
+    )
+    output = result.stdout.decode(errors="replace").splitlines()
     errors = result.stderr.decode(errors="replace")
     assert result.returncode == 1, errors
+    windows_patch_index = output.index("windows-patch:41:false")
+    assert not any(
+        line.startswith("docker:") for line in output[windows_patch_index + 1 :]
+    )
     if dispatch_started:
-        assert output == ""
+        assert output == [
+            "windows-snapshot",
+            "windows-patch:41:false",
+            "windows-snapshot",
+        ]
         assert "runner_fleet_recovery_needed" in errors
     else:
-        assert "cp:" in output
+        assert any(line.startswith("cp:") for line in output)
         assert "delete:new-registration" in output
         assert "restore-fence:itworx-security-runner:false" in output
         assert ("stop-daemon" in output) is bool(cutover_started)
-        assert ("docker:compose " in output) is bool(cutover_started)
+        compose_indexes = [
+            index
+            for index, line in enumerate(output)
+            if line.startswith("docker:compose ")
+        ]
+        assert bool(compose_indexes) is bool(cutover_started)
+        if compose_indexes:
+            assert compose_indexes[-1] < windows_patch_index
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+def test_copy_failure_still_restores_windows_after_managed_cleanup_aborts():
+    result = _restore_exit_harness(
+        cutover_started=1,
+        dispatch_started=0,
+        exit_status=1,
+        fail_first_cp=True,
+    )
+    output = result.stdout.decode(errors="replace").splitlines()
+    errors = result.stderr.decode(errors="replace")
+    assert result.returncode == 75
+    assert output == [
+        "cp:-a -- /synthetic-backup/docker-compose.yml /synthetic-compose",
+        "windows-snapshot",
+        "windows-patch:41:false",
+        "windows-snapshot",
+    ]
+    assert "managed rollback did not complete" in errors
+    assert not any(line.startswith("docker:") for line in output)
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+def test_exit_cleanup_restores_windows_on_success():
+    result = _restore_exit_harness(
+        cutover_started=0,
+        dispatch_started=1,
+        exit_status=0,
+    )
+    assert result.returncode == 0, result.stderr.decode(errors="replace")
+    assert result.stdout.decode(errors="replace").splitlines() == [
+        "windows-snapshot",
+        "windows-patch:41:false",
+        "windows-snapshot",
+    ]
+
+    script = SCRIPT_PATH.read_text(encoding="utf-8")
+    dispatch_index = script.index("dispatch_started=1")
+    restore_index = script.index("if ! restore_windows_native_admission; then", dispatch_index)
+    ready_index = script.index("printf 'ci_runner_fleet_v2_ready", restore_index)
+    trap_removed_index = script.index("trap - EXIT", ready_index)
+    assert restore_index < ready_index < trap_removed_index
+    assert "docker " not in script[restore_index:trap_removed_index]
+
+
+@pytest.mark.skipif(shutil.which("bash") is None, reason="bash is unavailable")
+def test_post_dispatch_windows_restore_failure_marks_recovery_needed():
+    result = _restore_exit_harness(
+        cutover_started=1,
+        dispatch_started=1,
+        exit_status=1,
+        fail_windows_patch=True,
+    )
+    assert result.returncode == 75
+    assert (
+        "runner_fleet_recovery_needed: Windows-native admission state could not be restored"
+        in result.stderr.decode(errors="replace")
+    )