from __future__ import annotations

import copy
import importlib.util
import sys
from contextlib import nullcontext
from pathlib import Path

import pytest

BROKER = Path(__file__).resolve().parents[2] / "ops/gitea-deploy-broker"


@pytest.fixture
def module(monkeypatch):
    monkeypatch.syspath_prepend(str(BROKER))
    spec = importlib.util.spec_from_file_location(
        "acceptance_quiescence_test", BROKER / "runner_acceptance_quiescence.py"
    )
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    monkeypatch.setattr(module.fleet, "_migration_lock", lambda **kwargs: nullcontext())
    return module


class Driver:
    def __init__(self, module):
        self.module = module
        self.state = {
            "daemon": {
                "legacyAttested": True,
                "engineId": "fixed-private-engine",
                "outerId": "a" * 64,
                "dockerRootDir": module.ROOTLESS_STORAGE,
                "rootless": True,
            },
            "productExclusionVerified": True,
            "dispatchBarrierVerified": True,
            "resources": {
                "attachmentsVerified": True,
                "ownershipVerified": True,
                "networks": ["fixed-network"],
                "volumes": ["preserved-data"],
            },
            "rows": copy.deepcopy(module._fixed_rows()),
            "origins": {},
        }
        for row in self.state["rows"]:
            row["ownerLabelDigest"] = "b" * 64 if row["ownerLabelPresent"] else None
            row["labelDigest"] = module._digest(row["composeLabels"])
        for project, claim in module.ORIGINS.items():
            if claim is None:
                self.state["origins"][project] = None
                continue
            attempt = claim["attempt"] if claim["attempt"] is not None else 0
            self.state["origins"][project] = {
                "repository": "jens/workmesh",
                "run": claim["run"],
                "head": claim["head"],
                "status": "completed",
                "conclusion": "failure",
                "attempt": attempt,
                "task": claim["task"],
                "taskHead": claim["head"],
                "taskAttempt": attempt,
                "jobsTerminal": True,
            }
        self.runners = [
            {
                "id": 9,
                "name": "legacy-linux-1",
                "labels": ["linux", "docker"],
                "busy": False,
                "disabled": False,
            },
            {
                "id": 22,
                "name": "legacy-linux-2",
                "labels": ["linux", "docker"],
                "busy": False,
                "disabled": False,
            },
            {
                "id": 700,
                "name": "windows-preserved",
                "labels": ["windows"],
                "busy": False,
                "disabled": False,
            },
        ]
        self.effects = []
        self.pending = []
        self.timeout = False
        self.after_term = None

    def maintenance(self):
        return nullcontext()

    def capture_guards(self):
        return {"runner": {"synthetic": True}, "host": {"synthetic": True}}

    def install_guards(self, receipt, persist):
        for name in ("runner", "host"):
            if receipt["guardProgress"].get(name) in {
                "installed",
                "restore_intent",
                "restored",
            }:
                continue
            receipt["guardProgress"][name] = "install_intent"
            persist()
            receipt["guardProgress"][name] = "installed"
            persist()

    def restore_guards(self, receipt, persist):
        for name in ("runner", "host"):
            if receipt["guardProgress"].get(name) == "restored":
                continue
            receipt["guardProgress"][name] = "restore_intent"
            persist()
            receipt["guardProgress"][name] = "restored"
            persist()

    def snapshot(self):
        return copy.deepcopy(self.state)

    def registrations(self):
        return copy.deepcopy(self.runners)

    def attest_registrations(self, rows):
        assert {row["id"] for row in rows if "linux" in row["labels"]} == {9, 22}

    def pending_jobs(self):
        return copy.deepcopy(self.pending)

    def pause(self):
        pass

    def set_registration_disabled(self, original, value):
        self.effects.append(("registration", original["id"], value))
        next(row for row in self.runners if row["id"] == original["id"])["disabled"] = (
            value
        )

    def disable_restart(self, cid):
        self.effects.append(("restart", cid))
        next(row for row in self.state["rows"] if row["id"] == cid)["restartPolicy"] = {
            "Name": "no",
            "MaximumRetryCount": 0,
        }

    def term(self, cid):
        row = next(row for row in self.state["rows"] if row["id"] == cid)
        if row["state"]["Running"]:
            self.effects.append(("TERM", cid))
            if not self.timeout:
                row["state"].update(
                    {
                        "Running": False,
                        "Status": "exited",
                        "Pid": 0,
                        "FinishedAt": "2026-10-09T20:00:00Z",
                    }
                )
        if self.after_term:
            self.after_term(self)

    def wait_exited(self, cid):
        return not self.timeout


@pytest.fixture
def proven(module, monkeypatch):
    receipt = {"value": None}
    monkeypatch.setattr(
        module, "_read_receipt", lambda: copy.deepcopy(receipt["value"])
    )

    def save(value, expected):
        current = receipt["value"]
        actual = module._digest(current) if current is not None else None
        module._require(actual == expected, "maintenance_receipt_cas_failed")
        receipt["value"] = copy.deepcopy(value)
        return module._digest(value)

    monkeypatch.setattr(module, "_save_receipt", save)
    return Driver(module), receipt


def test_missing_backup_origin_never_becomes_authority(module, monkeypatch):
    monkeypatch.setitem(module.ORIGINS, "workmesh-backup-exercise-39a1a242e4b9", None)
    monkeypatch.setattr(module, "_read_receipt", lambda: None)
    driver = Driver(module)
    result = module.preflight(revision="d" * 40, projects={}, driver=driver)
    assert result["status"] == "blocked"
    assert result["effectsPerformed"] is False
    assert driver.effects == []


def test_positive_preserves_resources_and_windows_then_idempotent(module, proven):
    driver, receipt = proven
    data_before = copy.deepcopy(driver.state["resources"])
    ids_before = {row["id"] for row in driver.state["rows"]}
    assert (
        module.preflight(revision="d" * 40, projects={}, driver=driver)["status"]
        == "ready"
    )
    result = module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert result["status"] == "quiesced"
    assert result["preservedData"] is True
    assert driver.state["resources"] == data_before
    assert {row["id"] for row in driver.state["rows"]} == ids_before
    assert all(not row["state"]["Running"] for row in driver.state["rows"])
    assert sum(effect[0] == "TERM" for effect in driver.effects) == 12
    assert driver.runners[2]["disabled"] is False
    assert all(
        effect[0] in {"registration", "restart", "TERM"} for effect in driver.effects
    )
    assert receipt["value"]["phase"] == "complete"
    count = len(driver.effects)
    assert (
        module.quiesce(revision="d" * 40, projects={}, driver=driver)["status"]
        == "already_quiesced"
    )
    assert len(driver.effects) == count
    with pytest.raises(module.QuiescenceError, match="maintenance_owner_changed"):
        module.quiesce(revision="e" * 40, projects={}, driver=driver)


@pytest.mark.parametrize("kind", ["queued", "in_progress", "pending"])
def test_initial_new_work_refuses_all_effects(module, proven, kind):
    driver, _ = proven
    driver.pending = [{"id": 99, "status": kind}]
    with pytest.raises(module.QuiescenceError, match="global_ci_work_present"):
        module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert driver.effects == []


@pytest.mark.parametrize("kind", ["queued", "in_progress", "pending"])
def test_new_work_during_execution_blocks_next_container(module, proven, kind):
    driver, receipt = proven
    driver.after_term = lambda self: setattr(
        self, "pending", [{"id": 99, "status": kind}]
    )
    with pytest.raises(module.QuiescenceError, match="global_ci_work_present"):
        module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert sum(e[0] == "restart" for e in driver.effects) == 1
    assert receipt["value"]["phase"] != "complete"
    assert all(row["disabled"] for row in driver.runners[:2])


@pytest.mark.parametrize(
    "change",
    [
        lambda s: s["daemon"].update({"rootless": False}),
        lambda s: s["daemon"].update({"dockerRootDir": "/srv/example/appdata/production"}),
        lambda s: s["rows"][0].update({"id": "f" * 64}),
        lambda s: s["rows"][0].update({"imageId": "sha256:" + "f" * 64}),
        lambda s: s["rows"][0]["composeLabels"].update(
            {"com.docker.compose.config-hash": "f" * 64}
        ),
        lambda s: s["rows"][0]["composeLabels"].update(
            {"com.docker.compose.project.working_dir": "/production"}
        ),
        lambda s: s["rows"][0]["mounts"].append(
            {"Type": "bind", "Source": "/production", "RW": True}
        ),
        lambda s: s["rows"][0]["networks"].update({"foreign": "f" * 64}),
        lambda s: s["resources"].update({"attachmentsVerified": False}),
        lambda s: s["resources"].update({"ownershipVerified": False}),
        lambda s: s.update({"productExclusionVerified": False}),
        lambda s: s.update({"dispatchBarrierVerified": False}),
        lambda s: s["rows"].append(copy.deepcopy(s["rows"][0])),
        lambda s: s["rows"][0].update({"ownerLabelDigest": "e" * 64}),
    ],
)
def test_identity_or_scope_drift_has_no_effect(module, proven, change):
    driver, _ = proven
    change(driver.state)
    with pytest.raises(module.QuiescenceError):
        module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert driver.effects == []


@pytest.mark.parametrize(
    "field,value",
    [
        ("attempt", None),
        ("attempt", 2),
        ("taskAttempt", 3),
        ("task", 999),
        ("head", "e" * 40),
        ("jobsTerminal", False),
    ],
)
def test_origin_drift_refuses_effects(module, proven, field, value):
    driver, _ = proven
    driver.state["origins"]["workmesh-product-21300-7bace5f2edf0d259"][field] = value
    with pytest.raises(module.QuiescenceError):
        module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert driver.effects == []


def test_legacy_attempt_zero_is_actual_sentinel_and_missing_is_not_assumed(
    module, proven
):
    driver, _ = proven
    module.ORIGINS["workmesh-runtime-19421"]["attempt"] = 0
    driver.state["origins"]["workmesh-runtime-19421"].update(attempt=0, taskAttempt=0)
    module.attest_snapshot(driver.snapshot())
    driver.state["origins"]["workmesh-runtime-19421"]["attempt"] = None
    with pytest.raises(module.QuiescenceError, match="attempt_unavailable"):
        module.attest_snapshot(driver.snapshot())


def test_timeout_retains_owned_guard_and_resumes_only_own_transition(module, proven):
    driver, receipt = proven
    driver.timeout = True
    result = module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert result["status"] == "partial" and result["guardHeld"] is True
    assert receipt["value"]["progress"][driver.state["rows"][0]["id"]] == "term_intent"
    driver.timeout = False
    result = module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert result["status"] == "quiesced"


def test_registration_drift_after_partial_never_restores_foreign_identity(
    module, proven
):
    driver, _ = proven
    driver.timeout = True
    module.quiesce(revision="d" * 40, projects={}, driver=driver)
    before = len(driver.effects)
    driver.runners[0]["labels"].append("foreign-project")
    with pytest.raises(module.QuiescenceError, match="legacy_registration_changed"):
        module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert len(driver.effects) == before


def test_no_arbitrary_or_destructive_docker_effect(module, monkeypatch):
    calls = []

    class Result:
        returncode = 0

    monkeypatch.setattr(
        module.fleet, "_run", lambda args, **kwargs: calls.append(args) or Result()
    )
    driver = module.ControllerDriver({})
    monkeypatch.setattr(driver, "assert_barrier", lambda: None)
    monkeypatch.setattr(driver, "_guard_binding", lambda: {})
    monkeypatch.setattr(driver, "_effect_admission", lambda cid, args: None)
    cid = module._fixed_rows()[0]["id"]
    driver.disable_restart(cid)
    assert calls == [
        [
            "docker",
            "exec",
            "-e",
            "DOCKER_HOST=tcp://127.0.0.1:2375",
            "gitea-runner-dind",
            "docker",
            "update",
            "--restart=no",
            cid,
        ]
    ]
    with pytest.raises(module.QuiescenceError, match="container_target_invalid"):
        driver.disable_restart("user-selected")


def test_public_preflight_never_serializes_private_canary(module):
    class BadDriver:
        def maintenance(self):
            return nullcontext()

        def pending_jobs(self):
            raise ValueError("Bearer SECRET-CANARY")

        def proof_gaps(self):
            return ["terminal_origin_unavailable"]

    result = module.preflight(revision="d" * 40, projects={}, driver=BadDriver())
    assert "SECRET-CANARY" not in str(result)
    assert result["effectsPerformed"] is False


@pytest.mark.parametrize("boundary", range(1, 93))
def test_every_durable_write_boundary_resumes_without_adoption(
    module, proven, monkeypatch, boundary
):
    driver, receipt = proven
    save = module._save_receipt
    count = 0

    def crash(value, expected):
        nonlocal count
        count += 1
        result = save(value, expected)
        if count == boundary:
            raise OSError("simulated power loss after durable commit")
        return result

    monkeypatch.setattr(module, "_save_receipt", crash)
    try:
        module.quiesce(revision="d" * 40, projects={}, driver=driver)
    except OSError:
        pass
    monkeypatch.setattr(module, "_save_receipt", save)
    result = module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert result["status"] in {"quiesced", "already_quiesced"}
    assert receipt["value"]["phase"] == "complete"
    assert all(not row["state"]["Running"] for row in driver.state["rows"])
    assert sum(e[0] == "TERM" for e in driver.effects) == 12


@pytest.mark.parametrize(
    "effect", ["set_registration_disabled", "disable_restart", "term"]
)
def test_effect_commit_before_receipt_completion_is_owned_resume(
    module, proven, monkeypatch, effect
):
    driver, _ = proven
    original = getattr(driver, effect)
    crashed = False

    def wrapper(*args):
        nonlocal crashed
        original(*args)
        if not crashed:
            crashed = True
            raise OSError("simulated loss after external effect")

    monkeypatch.setattr(driver, effect, wrapper)
    with pytest.raises(OSError):
        module.quiesce(revision="d" * 40, projects={}, driver=driver)
    monkeypatch.setattr(driver, effect, original)
    assert (
        module.quiesce(revision="d" * 40, projects={}, driver=driver)["status"]
        == "quiesced"
    )


def test_current_fixed_origins_include_reviewed_backup_and_attempts(module):
    assert module.ControllerDriver({}).proof_gaps() == []
    assert module.ORIGINS["workmesh-backup-exercise-39a1a242e4b9"]["run"] == 20184
    assert all(row["attempt"] == 1 for row in module.ORIGINS.values())


def test_malformed_complete_receipt_cannot_reenable_maintenance(module, proven):
    driver, receipt = proven
    module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert module.maintenance_fenced() is False
    receipt["value"]["progress"].pop(next(iter(receipt["value"]["progress"])))
    assert module.maintenance_fenced() is True


def test_failure_projection_never_misreports_partial_effects(module, proven):
    driver, _ = proven
    assert module.failure_projection("d" * 40)["effectsPerformed"] is False
    driver.timeout = True
    module.quiesce(revision="d" * 40, projects={}, driver=driver)
    result = module.failure_projection("d" * 40)
    assert result["status"] == "partial"
    assert result["effectsPerformed"] is True and result["guardHeld"] is True


@pytest.fixture
def native_origins(module, monkeypatch):
    dates = {
        19421: ("2026-09-29T00:53:46Z", "2026-09-29T00:57:44Z", 768),
        19617: ("2026-09-29T07:36:50Z", "2026-09-29T07:41:39Z", 834),
        20184: ("2026-09-29T22:34:11Z", "2026-09-29T22:35:35Z", 941),
        21300: ("2026-10-01T20:53:43Z", "2026-10-01T20:54:47Z", 1128),
    }
    tasks, runs, jobs = [], {}, {}
    for claim in module.ORIGINS.values():
        start, end, number = dates[claim["run"]]
        conclusion = "cancelled" if claim["run"] == 20184 else "failure"
        tasks.append(
            {
                "id": claim["task"],
                "run_number": number,
                "head_sha": claim["head"],
                "workflow_id": "runtime-acceptance.yml",
                "status": conclusion,
                "created_at": start,
                "run_started_at": start,
                "updated_at": end,
                "url": f"http://127.0.0.1:3000/Jens/WorkMesh/actions/runs/{claim['run']}",
            }
        )
        runs[claim["run"]] = {
            "id": claim["run"],
            "run_number": number,
            "head_sha": claim["head"],
            "status": "completed",
            "conclusion": conclusion,
            "run_attempt": 1,
            "run_started_at": None,
            "updated_at": end,
        }
        jobs[claim["run"]] = [
            {
                "id": claim["job"],
                "run_id": claim["run"],
                "head_sha": claim["head"],
                "run_attempt": 1,
                "runner_id": claim["runner"],
                "started_at": start,
                "status": "completed",
                "conclusion": conclusion,
            }
        ]
    # Pagination includes unrelated metadata but never returns it publicly.
    tasks.extend({"id": i} for i in range(50000, 50055))
    calls = []

    def api(path, **kwargs):
        calls.append(path)
        if "/actions/tasks?" in path:
            page = int(path.rsplit("=", 1)[-1])
            return {
                "total_count": len(tasks),
                "workflow_runs": copy.deepcopy(tasks[(page - 1) * 50 : page * 50]),
            }
        run = int(path.split("/runs/")[1].split("/")[0])
        return (
            {"total_count": len(jobs[run]), "jobs": copy.deepcopy(jobs[run])}
            if "/jobs?" in path
            else copy.deepcopy(runs[run])
        )

    monkeypatch.setattr(module.fleet, "_api_json", api)
    return tasks, runs, jobs, calls


def test_native_documented_task_binding_current_attempt_and_null_run_start(
    module, native_origins
):
    result = module.ControllerDriver({})._origins()
    module._attest_origins(result)
    assert all(row["attempt"] == 1 and row["jobsTerminal"] for row in result.values())
    assert len([path for path in native_origins[3] if "/tasks?" in path]) == 2


@pytest.mark.parametrize(
    "change",
    [
        "task-url",
        "task-head",
        "task-workflow",
        "task-time",
        "foreign-job",
        "duplicate-job-start",
        "later-attempt",
        "active-job",
        "missing-job",
        "missing-attempt",
        "terminal-task-changed",
        "duplicate-task",
    ],
)
def test_native_origin_drift_refuses_authority(module, native_origins, change):
    tasks, runs, jobs, _ = native_origins
    claim = module.ORIGINS[next(iter(module.ORIGINS))]
    task = next(row for row in tasks if row["id"] == claim["task"])
    job = jobs[claim["run"]][0]
    if change == "task-url":
        task["url"] = task["url"].replace("127.0.0.1", "foreign.invalid")
    elif change == "task-head":
        task["head_sha"] = "f" * 40
    elif change == "task-workflow":
        task["workflow_id"] = "other.yml"
    elif change == "task-time":
        task["created_at"] = "2026-10-02T00:00:00Z"
    elif change == "foreign-job":
        job["runner_id"] = 800
    elif change == "duplicate-job-start":
        jobs[claim["run"]].append(copy.deepcopy(job))
    elif change == "later-attempt":
        runs[claim["run"]]["run_attempt"] = 2
    elif change == "active-job":
        job.update(status="in_progress", conclusion=None)
    elif change == "missing-job":
        jobs[claim["run"]] = []
    elif change == "missing-attempt":
        runs[claim["run"]].pop("run_attempt")
    elif change == "terminal-task-changed":
        task["status"] = "success"
    elif change == "duplicate-task":
        tasks[-1] = copy.deepcopy(tasks[0])
    with pytest.raises(module.QuiescenceError):
        module._attest_origins(module.ControllerDriver({})._origins())


def test_exact_inspect_inventory_cannot_omit_foreign_volume_attachment(
    module, monkeypatch
):
    driver = module.ControllerDriver({})
    ids = ["a" * 64, "b" * 64]
    monkeypatch.setattr(driver, "_docker", lambda *args: [{"Id": ids[0]}])
    with pytest.raises(module.QuiescenceError, match="legacy_inventory_unavailable"):
        driver._snapshot_from_ids({}, ids)


@pytest.fixture
def product(module, monkeypatch):
    source = "/mnt/cache/appdata/gitea-deploy-runners/sources/workmesh"
    names = [
        "workmesh-postgres-1",
        "workmesh-api-1",
        "workmesh-n8n-1",
        "workmesh-workmesh-1",
    ]
    target = {
        "slug": "workmesh",
        "branch": "main",
        "exact_source": True,
        "source_dir": source,
        "env_source": "/srv/example/appdata/workmesh/.env",
        "compose_files": ["compose.yaml"],
        "compose_project": "workmesh",
        "containers": names,
    }
    rows = {
        name: {
            "Id": str(i + 1) * 64,
            "Name": "/" + name,
            "Image": "sha256:" + str(i + 1) * 64,
            "Created": "2026-09-01T00:00:00Z",
            "State": {"Running": True, "Health": {"Status": "healthy"}},
            "Config": {
                "Labels": {
                    "com.docker.compose.project": "workmesh",
                    "com.docker.compose.service": name.removeprefix(
                        "workmesh-"
                    ).removesuffix("-1"),
                    "com.docker.compose.project.working_dir": source,
                    "com.docker.compose.project.config_files": source + "/compose.yaml",
                }
            },
            "Mounts": [
                {
                    "Type": "volume",
                    "Name": "production_data",
                    "Source": "/var/lib/docker/volumes/production_data/_data",
                    "Destination": "/data",
                    "RW": True,
                }
            ],
            "NetworkSettings": {
                "Networks": {"workmesh_default": {"NetworkID": "e" * 64}}
            },
        }
        for i, name in enumerate(names)
    }
    info = {"ID": "host-engine", "DockerRootDir": "/var/lib/docker"}
    inventory = [row["Id"] for row in rows.values()]

    def run(argv, **kw):
        import json

        if argv[1] == "info":
            return json.dumps(info)
        if argv[1] == "inspect":
            return json.dumps([rows[argv[2]]])
        assert argv[1:3] == ["ps", "-a"]
        return "\n".join(inventory)

    monkeypatch.setattr(module.fleet, "_run", run)
    return target, rows, info, inventory


def test_manual_product_observation_protects_exact_healthy_target_without_fake_receipt(
    module, product
):
    target, _, _, _ = product
    proof = module.ControllerDriver({"jens/workmesh": target})._product_binding(
        {"engineId": "private"}, []
    )
    assert (
        len(proof["rows"]) == 4
        and proof["receiptStatus"] == "legacy-physical-observation"
    )


@pytest.mark.parametrize(
    "change",
    ["source", "engine", "root", "cid", "health", "mount", "network", "extra-product"],
)
def test_product_disjointness_fails_on_changed_identity_or_overlap(
    module, product, change
):
    target, rows, info, inventory = product
    row = next(iter(rows.values()))
    private = [
        {
            "id": "a" * 64,
            "mounts": [{"Source": "/private-data"}],
            "networks": {"private": "b" * 64},
        }
    ]
    if change == "source":
        row["Config"]["Labels"]["com.docker.compose.project.config_files"] = (
            "/foreign/compose.yaml"
        )
    elif change == "engine":
        info["ID"] = "private"
    elif change == "root":
        info["DockerRootDir"] = module.ROOTLESS_STORAGE
    elif change == "cid":
        row["Id"] = "a" * 64
    elif change == "health":
        row["State"]["Health"]["Status"] = "unhealthy"
    elif change == "mount":
        row["Mounts"][0]["Source"] = "/private-data"
    elif change == "network":
        row["NetworkSettings"]["Networks"]["workmesh_default"]["NetworkID"] = "b" * 64
    elif change == "extra-product":
        inventory.append("f" * 64)
    with pytest.raises(module.QuiescenceError):
        module.ControllerDriver({"jens/workmesh": target})._product_binding(
            {"engineId": "private"}, private
        )


@pytest.fixture
def real_guards(module, proven, monkeypatch, tmp_path):
    import hashlib

    driver = module.ControllerDriver({})
    monkeypatch.setattr(driver, "assert_barrier", lambda: None)
    monkeypatch.setattr(
        driver, "registrations", lambda: copy.deepcopy(proven[0].runners)
    )
    monkeypatch.setattr(driver, "attest_registrations", lambda rows: None)
    monkeypatch.setattr(driver, "pending_jobs", lambda: [])
    sources = {}
    for name in ("runner", "host"):
        stanza = (
            "if ! flock -n 8; then\nexit 0\nfi"
            if name == "runner"
            else "flock -n 8 || exit 0"
        )
        raw = (
            '#!/bin/bash\nreadonly MAINTENANCE_LOCK="/srv/example/appdata/gitea-deploy-runners/state/runner-maintenance.lock"\nexec 8>"$MAINTENANCE_LOCK"\n'
            + stanza
            + "\n# private SECRET-CANARY\n"
        ).encode()
        path = tmp_path / name
        path.write_bytes(raw)
        sources[name] = (path, hashlib.sha256(raw).hexdigest())
    monkeypatch.setattr(module, "GUARDS", sources)
    monkeypatch.setattr(
        module.fleet, "_read_source_file", lambda path, **kw: path.read_bytes()
    )

    def write(path, raw, *, mode, expected_sha256):
        module._require(
            hashlib.sha256(path.read_bytes()).hexdigest() == expected_sha256,
            "guard_cas_failed",
        )
        path.write_bytes(raw)

    monkeypatch.setattr(module.fleet, "_atomic_write_owned", write)
    # Only the filesystem-independent binding is exercised on Windows here.
    monkeypatch.setattr(
        module.os, "fstat", lambda fd: type("Info", (), {"st_dev": 1, "st_ino": 2})()
    )
    return driver, sources, proven[1]


def test_actual_guard_wrapper_owned_cas_and_exact_restore(module, real_guards):
    driver, sources, store = real_guards
    captured = driver.capture_guards()
    value = {
        "guards": captured,
        "guardProgress": {},
        "registrations": driver.registrations(),
        "registrationProgress": {},
        "progress": {row["id"]: "quiesced" for row in module._fixed_rows()},
    }
    store["value"] = copy.deepcopy(value)

    def persist():
        store["value"] = copy.deepcopy(value)

    driver.install_guards(value, persist)
    assert all(
        module.GUARD_MARKER.encode() in path.read_bytes()
        for path, _ in sources.values()
    )
    value["registrationProgress"] = {"9": "restored", "22": "restored"}
    persist()
    driver.restore_guards(value, persist)
    assert value["guardProgress"] == {"runner": "restored", "host": "restored"}
    assert all(
        module.hashlib.sha256(path.read_bytes()).hexdigest() == digest
        for path, digest in sources.values()
    )


@pytest.mark.parametrize(
    "stage", [None, "install_intent", "installed", "restore_intent", "restored"]
)
def test_foreign_guard_changes_are_never_adopted_or_overwritten(
    module, real_guards, stage
):
    driver, sources, store = real_guards
    captured = driver.capture_guards()
    store["value"] = {
        "guards": captured,
        "guardProgress": {"runner": stage} if stage else {},
    }
    path = sources["runner"][0]
    path.write_bytes(path.read_bytes() + b"foreign-change")
    before = path.read_bytes()
    with pytest.raises(module.QuiescenceError, match="dispatch_guard_changed"):
        driver._guard_binding()
    assert path.read_bytes() == before


@pytest.mark.parametrize(
    "failure", ["short-write", "temp-fsync", "before-link", "after-link"]
)
def test_initial_receipt_atomic_publication_faults(
    module, monkeypatch, tmp_path, failure
):
    import json

    monkeypatch.setattr(module, "STATE_PATH", tmp_path / "receipt.json")
    monkeypatch.setattr(module.fleet, "_trusted_directory", lambda path: None)
    monkeypatch.setattr(
        module.fleet, "_trusted_private_file", lambda path, **kw: path.read_bytes()
    )
    receipt = {
        "schema": module.SCHEMA,
        "revision": "d" * 40,
        "manifestDigest": module._digest(module.OBSERVATION),
        "phase": "guarding",
        "baseline": {},
        "registrations": [],
        "registrationProgress": {},
        "progress": {},
        "guards": {"runner": {}, "host": {}},
        "guardProgress": {},
    }
    real_open, real_link, real_fsync = module.os.fdopen, module.os.link, module.os.fsync
    if failure == "short-write":

        class Short:
            def __init__(self, fd, mode):
                self.stream = real_open(fd, mode)

            def __enter__(self):
                return self

            def __exit__(self, *args):
                self.stream.close()

            def write(self, raw):
                return self.stream.write(raw[:1])

        monkeypatch.setattr(module.os, "fdopen", Short)
    elif failure == "temp-fsync":
        monkeypatch.setattr(
            module.os, "fsync", lambda fd: (_ for _ in ()).throw(OSError("fault"))
        )
    else:

        def link(src, dst, **kw):
            if failure == "after-link":
                real_link(src, dst, **kw)
            raise OSError("fault")

        monkeypatch.setattr(module.os, "link", link)
    with pytest.raises((OSError, module.QuiescenceError)):
        module._save_receipt(receipt, None)
    if failure == "after-link":
        assert json.loads(module.STATE_PATH.read_bytes()) == receipt
        assert module._read_receipt() == receipt
    else:
        assert not module.STATE_PATH.exists()
    assert list(tmp_path.glob(".acceptance-receipt-*")) == []


def test_deadline_expires_before_native_read_or_effect(module, monkeypatch):
    driver = module.ControllerDriver({})
    driver._deadline = module.time.monotonic() - 1
    monkeypatch.setattr(
        module.fleet, "_run", lambda *a, **kw: pytest.fail("no effect after deadline")
    )
    monkeypatch.setattr(
        module.fleet,
        "_api_json",
        lambda *a, **kw: pytest.fail("no read after deadline"),
    )
    with pytest.raises(module.QuiescenceError, match="maintenance_execution_deadline"):
        driver._run(["docker", "info"])
    with pytest.raises(module.QuiescenceError, match="maintenance_execution_deadline"):
        driver._api_json("/admin/actions/runners")


def test_readonly_poll_reports_real_partial_and_complete_without_resuming(
    module, proven
):
    driver, receipt = proven
    driver.timeout = True
    module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert module.operation_status()["status"] == "partial"
    assert module.operation_status(active=True)["status"] == "in_progress"
    before = len(driver.effects)
    assert module.operation_status()["phase"] == "quiescing"
    assert len(driver.effects) == before
    driver.timeout = False
    module.quiesce(revision="d" * 40, projects={}, driver=driver)
    assert module.operation_status()["status"] == "quiesced"
    assert module.operation_status()["completedContainers"] == 18


def test_external_migration_lock_is_nonblocking_and_readonly_busy(module, proven, monkeypatch):
    def busy(**kwargs):
        assert kwargs == {"blocking": False}
        raise module.fleet.RunnerFleetSpecializeError("runner_fleet_specialization_recovery_busy")
    monkeypatch.setattr(module.fleet, "_migration_lock", busy)
    driver, _ = proven
    result = module.preflight(revision="d"*40, projects={}, driver=driver)
    assert result["status"] == "in_progress" and result["effectsPerformed"] is False
    assert driver.effects == []