#!/usr/bin/env python3
"""Authenticated, fail-closed rollback executor for the isolated NAS preview.

The service is disabled unless both ``--enable`` and
``VIGNETTE_NAS_ROLLBACK_EXECUTOR_ENABLED=true`` are present.  It accepts only
the G8 ``runtime``/``release_gate`` rollback command, binds that command to an
offline-provisioned manifest, and recreates only the audited preview services.

The executor never builds, pulls, stops a Compose project, or mutates a
database, volume, or network.  Docker is invoked with argument arrays and
``shell=False``.  An execution that fails after mutation restores the exact
API/Web image IDs observed immediately before the attempt.
"""

from __future__ import annotations

import argparse
import contextlib
import hashlib
import hmac
import json
import os
import re
import shutil
import ssl
import stat
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Protocol
from uuid import UUID


TOKEN_HEADER = "X-Vignette-Rollback-Executor-Token"
TOKEN_ENV = "VIGNETTE_NAS_ROLLBACK_EXECUTOR_TOKEN"
JOURNAL_KEY_ENV = "VIGNETTE_NAS_ROLLBACK_EXECUTOR_JOURNAL_KEY"
ENABLED_ENV = "VIGNETTE_NAS_ROLLBACK_EXECUTOR_ENABLED"
MAX_BODY_BYTES = 65_536
MAX_HEALTH_BODY_BYTES = 65_536

TARGET_PROJECT = "vignette-preview-20260807"
TARGET_ROOT = PurePosixPath("/volume1/docker/vignette-preview-20260807")
TARGET_SERVICES = ("api", "web")
TARGET_HEALTH_URL = "http://127.0.0.1:8088/api/health"
TARGET_DOCKER_SOCKET = "unix:///var/run/docker.sock"
DEFAULT_STATE_DIR = Path(str(TARGET_ROOT / ".rollback-executor"))

REQUEST_SCHEMA = "oas.rollback-executor.v1"
MANIFEST_SCHEMA = "vignette.nas-preview-rollback-manifest.v1"
JOURNAL_SCHEMA = "vignette.nas-preview-rollback-journal.v1"

SHA256_RE = re.compile(r"^[a-f0-9]{64}$")
IMAGE_RE = re.compile(r"^sha256:[a-f0-9]{64}$")
ARTIFACT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$")
EXECUTION_ID_RE = re.compile(r"^nas-g8-[a-f0-9]{24}$")
RELEASE_KEY_RE = re.compile(r"^[a-f0-9]{16}-\d{8}T\d{6}Z-\d+$")
SAFE_REF_PREFIXES = ("https://", "audit://", "db://", "repo://")

REQUEST_FIELDS = frozenset(
    {
        "schema_version",
        "idempotency_key",
        "approval_event_id",
        "rollback_scope",
        "target_kind",
        "target_id",
        "subject_id",
        "rollback_target_id",
        "artifact_record_id",
        "artifact_id",
        "artifact_sha256",
        "artifact_provenance_uri",
        "authorization_evidence_refs",
    }
)


class ExecutorError(RuntimeError):
    """Expected fail-closed error with a non-sensitive external code."""

    def __init__(self, code: str, *, http_status: int = 422):
        self.code = code
        self.http_status = http_status
        super().__init__(code)


class CommandFailure(RuntimeError):
    """A command failed; stdout/stderr are intentionally not retained."""

    def __init__(self, operation: str, returncode: int | None = None):
        self.operation = operation
        self.returncode = returncode
        super().__init__(operation)


def _canonical_bytes(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")


def _sha256(value: Any) -> str:
    return hashlib.sha256(_canonical_bytes(value)).hexdigest()


def _resolved_policy_sha256(value: dict[str, Any]) -> str:
    """Hash resolved policy while images remain a separate exact binding."""

    normalized = json.loads(_canonical_bytes(value))
    services = normalized.get("services")
    if not isinstance(services, dict):
        raise CommandFailure("compose_config_policy")
    for service_name in ("api", "web"):
        service = services.get(service_name)
        if not isinstance(service, dict):
            raise CommandFailure("compose_config_policy")
        service.pop("image", None)
    return _sha256(normalized)


def _write_all(descriptor: int, value: bytes) -> None:
    offset = 0
    while offset < len(value):
        written = os.write(descriptor, value[offset:])
        if written <= 0:
            raise OSError("short write")
        offset += written


def _fsync_directory(path: Path) -> None:
    if os.name == "nt" or not hasattr(os, "O_DIRECTORY"):
        return
    descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _utc_now() -> str:
    return datetime.now(UTC).isoformat(timespec="seconds")


def _canonical_uuid(value: Any, field: str) -> str:
    if not isinstance(value, str):
        raise ExecutorError(f"invalid_{field}")
    try:
        parsed = UUID(value)
    except ValueError as exc:
        raise ExecutorError(f"invalid_{field}") from exc
    if str(parsed) != value:
        raise ExecutorError(f"invalid_{field}")
    return value


def _strict_string(value: Any, field: str, *, maximum: int) -> str:
    if not isinstance(value, str) or not value or len(value) > maximum:
        raise ExecutorError(f"invalid_{field}")
    return value


def _validate_absolute_ref(value: Any, field: str) -> str:
    value = _strict_string(value, field, maximum=500)
    if not value.startswith(SAFE_REF_PREFIXES):
        raise ExecutorError(f"invalid_{field}")
    return value


@dataclass(frozen=True)
class RollbackRequest:
    schema_version: str
    idempotency_key: str
    approval_event_id: str
    rollback_scope: str
    target_kind: str
    target_id: str
    subject_id: str
    rollback_target_id: str
    artifact_record_id: str
    artifact_id: str
    artifact_sha256: str
    artifact_provenance_uri: str
    authorization_evidence_refs: tuple[str, ...]

    @classmethod
    def parse(cls, payload: Any) -> RollbackRequest:
        if not isinstance(payload, dict) or set(payload) != REQUEST_FIELDS:
            raise ExecutorError("invalid_request_shape")
        if payload.get("schema_version") != REQUEST_SCHEMA:
            raise ExecutorError("invalid_request_schema")
        if payload.get("rollback_scope") != "runtime":
            raise ExecutorError("unsupported_rollback_scope")
        if payload.get("target_kind") != "release_gate":
            raise ExecutorError("unsupported_target_kind")
        artifact_id = _strict_string(
            payload.get("artifact_id"), "artifact_id", maximum=240
        )
        if ARTIFACT_ID_RE.fullmatch(artifact_id) is None:
            raise ExecutorError("invalid_artifact_id")
        artifact_sha256 = payload.get("artifact_sha256")
        if (
            not isinstance(artifact_sha256, str)
            or SHA256_RE.fullmatch(artifact_sha256) is None
        ):
            raise ExecutorError("invalid_artifact_sha256")
        raw_refs = payload.get("authorization_evidence_refs")
        if not isinstance(raw_refs, list) or not 1 <= len(raw_refs) <= 100:
            raise ExecutorError("invalid_authorization_evidence_refs")
        refs = tuple(
            _validate_absolute_ref(value, "authorization_evidence_ref")
            for value in raw_refs
        )
        if len(set(refs)) != len(refs):
            raise ExecutorError("duplicate_authorization_evidence_refs")
        return cls(
            schema_version=REQUEST_SCHEMA,
            idempotency_key=_canonical_uuid(
                payload.get("idempotency_key"), "idempotency_key"
            ),
            approval_event_id=_canonical_uuid(
                payload.get("approval_event_id"), "approval_event_id"
            ),
            rollback_scope="runtime",
            target_kind="release_gate",
            target_id=_canonical_uuid(payload.get("target_id"), "target_id"),
            subject_id=_strict_string(
                payload.get("subject_id"), "subject_id", maximum=180
            ),
            rollback_target_id=_strict_string(
                payload.get("rollback_target_id"), "rollback_target_id", maximum=180
            ),
            artifact_record_id=_canonical_uuid(
                payload.get("artifact_record_id"), "artifact_record_id"
            ),
            artifact_id=artifact_id,
            artifact_sha256=artifact_sha256,
            artifact_provenance_uri=_validate_absolute_ref(
                payload.get("artifact_provenance_uri"), "artifact_provenance_uri"
            ),
            authorization_evidence_refs=refs,
        )

    def canonical_payload(self) -> dict[str, Any]:
        payload = asdict(self)
        payload["authorization_evidence_refs"] = list(self.authorization_evidence_refs)
        return payload


def _validate_compose_file(value: Any, artifact_id: str) -> str:
    value = _strict_string(value, "compose_file", maximum=500)
    candidate = PurePosixPath(value)
    if (
        not candidate.is_absolute()
        or ".." in candidate.parts
        or str(candidate) != value
    ):
        raise ExecutorError("invalid_compose_file")
    try:
        relative = candidate.relative_to(TARGET_ROOT)
    except ValueError as exc:
        raise ExecutorError("invalid_compose_file") from exc
    immutable = relative.parts == (
        ".rollback-executor",
        "runtime",
        artifact_id,
        "docker-compose.yml",
    )
    if not immutable:
        raise ExecutorError("invalid_compose_file")
    return value


def _validate_env_file(value: Any, artifact_id: str) -> str:
    value = _strict_string(value, "env_file", maximum=500)
    expected = str(
        TARGET_ROOT / ".rollback-executor" / "runtime" / artifact_id / ".env"
    )
    if value != expected:
        raise ExecutorError("invalid_env_file")
    return value


@dataclass(frozen=True)
class RollbackArtifact:
    idempotency_key: str
    approval_event_id: str
    authorization_evidence_refs: tuple[str, ...]
    artifact_id: str
    artifact_sha256: str
    artifact_record_id: str
    artifact_provenance_uri: str
    target_id: str
    subject_id: str
    rollback_target_id: str
    compose_file: str
    compose_sha256: str
    env_file: str
    env_sha256: str
    resolved_config_sha256: str
    api_image: str
    web_image: str

    @classmethod
    def parse(cls, payload: Any) -> RollbackArtifact:
        expected = {
            "idempotency_key",
            "approval_event_id",
            "authorization_evidence_refs",
            "artifact_id",
            "artifact_sha256",
            "artifact_record_id",
            "artifact_provenance_uri",
            "target_id",
            "subject_id",
            "rollback_target_id",
            "compose_file",
            "compose_sha256",
            "env_file",
            "env_sha256",
            "resolved_config_sha256",
            "api_image",
            "web_image",
        }
        if not isinstance(payload, dict) or set(payload) != expected:
            raise ExecutorError("invalid_manifest_artifact_shape")
        artifact_id = _strict_string(
            payload.get("artifact_id"), "artifact_id", maximum=240
        )
        if ARTIFACT_ID_RE.fullmatch(artifact_id) is None:
            raise ExecutorError("invalid_artifact_id")
        artifact_sha256 = payload.get("artifact_sha256")
        if (
            not isinstance(artifact_sha256, str)
            or SHA256_RE.fullmatch(artifact_sha256) is None
        ):
            raise ExecutorError("invalid_artifact_sha256")
        api_image = payload.get("api_image")
        web_image = payload.get("web_image")
        if not isinstance(api_image, str) or IMAGE_RE.fullmatch(api_image) is None:
            raise ExecutorError("invalid_api_image")
        if not isinstance(web_image, str) or IMAGE_RE.fullmatch(web_image) is None:
            raise ExecutorError("invalid_web_image")
        raw_refs = payload.get("authorization_evidence_refs")
        if not isinstance(raw_refs, list) or not 1 <= len(raw_refs) <= 100:
            raise ExecutorError("invalid_authorization_evidence_refs")
        refs = tuple(
            _validate_absolute_ref(value, "authorization_evidence_ref")
            for value in raw_refs
        )
        if len(set(refs)) != len(refs):
            raise ExecutorError("duplicate_authorization_evidence_refs")
        compose_sha256 = payload.get("compose_sha256")
        env_sha256 = payload.get("env_sha256")
        resolved_config_sha256 = payload.get("resolved_config_sha256")
        if (
            not isinstance(compose_sha256, str)
            or SHA256_RE.fullmatch(compose_sha256) is None
        ):
            raise ExecutorError("invalid_compose_sha256")
        if not isinstance(env_sha256, str) or SHA256_RE.fullmatch(env_sha256) is None:
            raise ExecutorError("invalid_env_sha256")
        if (
            not isinstance(resolved_config_sha256, str)
            or SHA256_RE.fullmatch(resolved_config_sha256) is None
        ):
            raise ExecutorError("invalid_resolved_config_sha256")
        return cls(
            idempotency_key=_canonical_uuid(
                payload.get("idempotency_key"), "idempotency_key"
            ),
            approval_event_id=_canonical_uuid(
                payload.get("approval_event_id"), "approval_event_id"
            ),
            authorization_evidence_refs=refs,
            artifact_id=artifact_id,
            artifact_sha256=artifact_sha256,
            artifact_record_id=_canonical_uuid(
                payload.get("artifact_record_id"), "artifact_record_id"
            ),
            artifact_provenance_uri=_validate_absolute_ref(
                payload.get("artifact_provenance_uri"), "artifact_provenance_uri"
            ),
            target_id=_canonical_uuid(payload.get("target_id"), "target_id"),
            subject_id=_strict_string(
                payload.get("subject_id"), "subject_id", maximum=180
            ),
            rollback_target_id=_strict_string(
                payload.get("rollback_target_id"), "rollback_target_id", maximum=180
            ),
            compose_file=_validate_compose_file(
                payload.get("compose_file"), artifact_id
            ),
            compose_sha256=compose_sha256,
            env_file=_validate_env_file(payload.get("env_file"), artifact_id),
            env_sha256=env_sha256,
            resolved_config_sha256=resolved_config_sha256,
            api_image=api_image,
            web_image=web_image,
        )

    def matches(self, request: RollbackRequest) -> bool:
        return all(
            (
                self.idempotency_key == request.idempotency_key,
                self.approval_event_id == request.approval_event_id,
                self.authorization_evidence_refs == request.authorization_evidence_refs,
                self.artifact_id == request.artifact_id,
                self.artifact_sha256 == request.artifact_sha256,
                self.artifact_record_id == request.artifact_record_id,
                self.artifact_provenance_uri == request.artifact_provenance_uri,
                self.target_id == request.target_id,
                self.subject_id == request.subject_id,
                self.rollback_target_id == request.rollback_target_id,
            )
        )


@dataclass(frozen=True)
class RollbackManifest:
    artifacts: dict[tuple[str, str], RollbackArtifact]
    content_sha256: str

    def bind(self, request: RollbackRequest) -> RollbackArtifact:
        artifact = self.artifacts.get((request.artifact_id, request.artifact_sha256))
        if artifact is None:
            raise ExecutorError("artifact_not_allowlisted", http_status=403)
        if not artifact.matches(request):
            raise ExecutorError("artifact_request_binding_mismatch", http_status=409)
        return artifact


def _read_regular_file(path: Path) -> bytes:
    flags = os.O_RDONLY
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        descriptor = os.open(path, flags)
    except OSError as exc:
        raise ExecutorError("manifest_unreadable", http_status=500) from exc
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode):
            raise ExecutorError("manifest_not_regular", http_status=500)
        if os.name != "nt" and info.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
            raise ExecutorError("manifest_permissions_unsafe", http_status=500)
        if os.name != "nt" and info.st_uid != os.geteuid():
            raise ExecutorError("manifest_owner_unsafe", http_status=500)
        with os.fdopen(descriptor, "rb", closefd=False) as handle:
            return handle.read(MAX_BODY_BYTES + 1)
    finally:
        os.close(descriptor)


def _sha256_regular_file(path: Path, *, maximum_bytes: int = 8_388_608) -> str:
    flags = os.O_RDONLY
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        descriptor = os.open(path, flags)
    except OSError as exc:
        raise CommandFailure("runtime_input_unreadable") from exc
    digest = hashlib.sha256()
    total = 0
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode):
            raise CommandFailure("runtime_input_not_regular")
        if os.name != "nt" and (
            info.st_uid != os.geteuid() or info.st_mode & (stat.S_IWGRP | stat.S_IWOTH)
        ):
            raise CommandFailure("runtime_input_permissions_unsafe")
        while True:
            chunk = os.read(descriptor, 65_536)
            if not chunk:
                break
            total += len(chunk)
            if total > maximum_bytes:
                raise CommandFailure("runtime_input_too_large")
            digest.update(chunk)
    finally:
        os.close(descriptor)
    return digest.hexdigest()


def load_manifest(path: Path, expected_sha256: str) -> RollbackManifest:
    if SHA256_RE.fullmatch(expected_sha256) is None:
        raise ExecutorError("invalid_manifest_sha256", http_status=500)
    raw = _read_regular_file(path)
    if not raw or len(raw) > MAX_BODY_BYTES:
        raise ExecutorError("manifest_size_rejected", http_status=500)
    actual_sha256 = hashlib.sha256(raw).hexdigest()
    if not hmac.compare_digest(actual_sha256, expected_sha256):
        raise ExecutorError("manifest_sha256_mismatch", http_status=500)
    try:
        payload = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ExecutorError("manifest_json_invalid", http_status=500) from exc
    if not isinstance(payload, dict) or set(payload) != {
        "schema_version",
        "target",
        "artifacts",
    }:
        raise ExecutorError("manifest_shape_invalid", http_status=500)
    if payload.get("schema_version") != MANIFEST_SCHEMA:
        raise ExecutorError("manifest_schema_invalid", http_status=500)
    target = payload.get("target")
    if not isinstance(target, dict) or target != {
        "compose_project": TARGET_PROJECT,
        "remote_root": str(TARGET_ROOT),
        "services": list(TARGET_SERVICES),
        "health_url": TARGET_HEALTH_URL,
    }:
        raise ExecutorError("manifest_target_invalid", http_status=500)
    raw_artifacts = payload.get("artifacts")
    if not isinstance(raw_artifacts, list) or not raw_artifacts:
        raise ExecutorError("manifest_artifacts_invalid", http_status=500)
    artifacts: dict[tuple[str, str], RollbackArtifact] = {}
    artifact_ids: set[str] = set()
    idempotency_keys: set[str] = set()
    approval_event_ids: set[str] = set()
    for raw_artifact in raw_artifacts:
        artifact = RollbackArtifact.parse(raw_artifact)
        key = (artifact.artifact_id, artifact.artifact_sha256)
        if (
            key in artifacts
            or artifact.artifact_id in artifact_ids
            or artifact.idempotency_key in idempotency_keys
            or artifact.approval_event_id in approval_event_ids
        ):
            raise ExecutorError("manifest_artifact_duplicate", http_status=500)
        artifacts[key] = artifact
        artifact_ids.add(artifact.artifact_id)
        idempotency_keys.add(artifact.idempotency_key)
        approval_event_ids.add(artifact.approval_event_id)
    return RollbackManifest(artifacts=artifacts, content_sha256=actual_sha256)


@dataclass(frozen=True)
class ImagePair:
    api_image: str
    web_image: str


class RuntimeController(Protocol):
    def snapshot(self) -> ImagePair: ...

    def ensure_images_present(self, images: ImagePair) -> None: ...

    def apply(
        self, artifact: RollbackArtifact, images: ImagePair, execution_id: str
    ) -> None: ...

    def verify(self, images: ImagePair) -> None: ...


class HealthProbe(Protocol):
    def wait_ready(self) -> None: ...


@dataclass(frozen=True)
class ProcessResult:
    returncode: int
    stdout: str


class CommandRunner(Protocol):
    def run(
        self, operation: str, argv: list[str], *, timeout: float
    ) -> ProcessResult: ...


class SubprocessRunner:
    """Execute one fixed argv without logging arguments or command output."""

    def run(self, operation: str, argv: list[str], *, timeout: float) -> ProcessResult:
        if (
            not isinstance(argv, list)
            or not argv
            or any(not isinstance(item, str) for item in argv)
        ):
            raise TypeError("argv must be a non-empty string list")
        try:
            completed = subprocess.run(
                argv,
                check=False,
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=timeout,
                shell=False,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            raise CommandFailure(operation) from exc
        if completed.returncode != 0:
            raise CommandFailure(operation, completed.returncode)
        return ProcessResult(completed.returncode, completed.stdout)


class DockerRuntimeController:
    """Narrow Docker adapter for one exact Compose project and three services."""

    def __init__(
        self,
        runner: CommandRunner,
        *,
        state_dir: Path = DEFAULT_STATE_DIR,
        docker_bin: str = "docker",
    ):
        if not isinstance(docker_bin, str) or not docker_bin:
            raise ValueError("docker binary is required")
        self.runner = runner
        self.state_dir = state_dir
        self.docker_bin = docker_bin

    @property
    def _docker(self) -> list[str]:
        return [self.docker_bin, "--host", TARGET_DOCKER_SOCKET]

    @staticmethod
    def _container(service: str) -> str:
        if service not in {"api", "web"}:
            raise ValueError("service is not inspectable")
        return f"{TARGET_PROJECT}-{service}-1"

    def _inspect(self, service: str, template: str, operation: str) -> str:
        result = self.runner.run(
            operation,
            [
                *self._docker,
                "inspect",
                f"--format={template}",
                self._container(service),
            ],
            timeout=15,
        )
        lines = [line.strip() for line in result.stdout.splitlines() if line.strip()]
        if len(lines) != 1:
            raise CommandFailure(operation)
        return lines[0]

    def _assert_identity(self, service: str) -> None:
        project = self._inspect(
            service,
            '{{ index .Config.Labels "com.docker.compose.project" }}',
            f"inspect_{service}_project",
        )
        compose_service = self._inspect(
            service,
            '{{ index .Config.Labels "com.docker.compose.service" }}',
            f"inspect_{service}_service",
        )
        if project != TARGET_PROJECT or compose_service != service:
            raise CommandFailure(f"identity_{service}")

    def _assert_dependencies(self) -> None:
        network_name = f"{TARGET_PROJECT}_vignette"
        network = self.runner.run(
            "inspect_preview_network",
            [
                *self._docker,
                "network",
                "inspect",
                '--format={{ index .Labels "com.docker.compose.project" }}|'
                '{{ index .Labels "com.docker.compose.network" }}',
                network_name,
            ],
            timeout=15,
        )
        if network.stdout.strip() != f"{TARGET_PROJECT}|vignette":
            raise CommandFailure("preview_network_identity")
        volume_name = f"{TARGET_PROJECT}_apiuploads"
        volume = self.runner.run(
            "inspect_preview_volume",
            [
                *self._docker,
                "volume",
                "inspect",
                '--format={{ index .Labels "com.docker.compose.project" }}|'
                '{{ index .Labels "com.docker.compose.volume" }}',
                volume_name,
            ],
            timeout=15,
        )
        if volume.stdout.strip() != f"{TARGET_PROJECT}|apiuploads":
            raise CommandFailure("preview_volume_identity")

    def snapshot(self) -> ImagePair:
        for service in ("api", "web"):
            self._assert_identity(service)
        pair = ImagePair(
            api_image=self._inspect("api", "{{.Image}}", "inspect_api_image"),
            web_image=self._inspect("web", "{{.Image}}", "inspect_web_image"),
        )
        self._validate_images(pair)
        return pair

    @staticmethod
    def _validate_images(images: ImagePair) -> None:
        if IMAGE_RE.fullmatch(images.api_image) is None:
            raise CommandFailure("invalid_api_image")
        if IMAGE_RE.fullmatch(images.web_image) is None:
            raise CommandFailure("invalid_web_image")

    def ensure_images_present(self, images: ImagePair) -> None:
        self._validate_images(images)
        for service, image in (("api", images.api_image), ("web", images.web_image)):
            result = self.runner.run(
                f"verify_{service}_image_present",
                [
                    *self._docker,
                    "image",
                    "inspect",
                    "--format={{.Id}}",
                    image,
                ],
                timeout=15,
            )
            lines = [
                line.strip() for line in result.stdout.splitlines() if line.strip()
            ]
            if lines != [image]:
                raise CommandFailure(f"verify_{service}_image_present")

    @staticmethod
    def _resolved_networks(service: dict[str, Any]) -> set[str]:
        networks = service.get("networks", {})
        if isinstance(networks, dict):
            return set(networks)
        if isinstance(networks, list) and all(
            isinstance(item, str) for item in networks
        ):
            return set(networks)
        raise CommandFailure("compose_network_policy")

    @staticmethod
    def _is_configured(service: dict[str, Any], key: str) -> bool:
        if key not in service:
            return False
        return service[key] not in (None, False, "", [], {})

    @classmethod
    def _validate_resolved_config(
        cls,
        payload: Any,
        images: ImagePair,
    ) -> None:
        if not isinstance(payload, dict):
            raise CommandFailure("compose_config_policy")
        if set(payload) - {"name", "networks", "services", "volumes"}:
            raise CommandFailure("compose_top_level_policy")
        services = payload.get("services")
        if not isinstance(services, dict) or set(services) != {
            "api",
            "db",
            "proxy",
            "web",
        }:
            raise CommandFailure("compose_service_policy")
        forbidden = {
            "cap_add",
            "cgroup",
            "cgroup_parent",
            "command",
            "configs",
            "container_name",
            "device_cgroup_rules",
            "devices",
            "entrypoint",
            "env_file",
            "extends",
            "ipc",
            "network_mode",
            "pid",
            "ports",
            "privileged",
            "pull_policy",
            "runtime",
            "secrets",
            "security_opt",
            "sysctls",
            "userns_mode",
            "uts",
            "volumes_from",
        }
        for service_name, expected_image in (
            ("api", images.api_image),
            ("web", images.web_image),
        ):
            service = services.get(service_name)
            if not isinstance(service, dict) or service.get("image") != expected_image:
                raise CommandFailure("compose_image_policy")
            if any(cls._is_configured(service, key) for key in forbidden):
                raise CommandFailure("compose_privilege_policy")
            if cls._resolved_networks(service) != {"vignette"}:
                raise CommandFailure("compose_network_policy")
        api_volumes = services["api"].get("volumes", [])
        if not isinstance(api_volumes, list) or len(api_volumes) != 1:
            raise CommandFailure("compose_volume_policy")
        api_upload = api_volumes[0]
        allowed_volume_sources = {
            "apiuploads",
            f"{TARGET_PROJECT}_apiuploads",
        }
        if (
            not isinstance(api_upload, dict)
            or api_upload.get("type") != "volume"
            or api_upload.get("source") not in allowed_volume_sources
            or api_upload.get("target") != "/app/uploads"
            or api_upload.get("read_only") is True
        ):
            raise CommandFailure("compose_volume_policy")
        if services["web"].get("volumes", []) not in (None, []):
            raise CommandFailure("compose_volume_policy")
        networks = payload.get("networks", {})
        if not isinstance(networks, dict) or set(networks) != {"vignette"}:
            raise CommandFailure("compose_network_policy")
        network = networks["vignette"]
        if not isinstance(network, dict):
            raise CommandFailure("compose_network_policy")
        if network.get("external") is True:
            raise CommandFailure("compose_network_policy")
        if network.get("driver", "bridge") != "bridge":
            raise CommandFailure("compose_network_policy")
        if network.get("name") not in (
            None,
            "vignette",
            f"{TARGET_PROJECT}_vignette",
        ):
            raise CommandFailure("compose_network_policy")
        volumes = payload.get("volumes", {})
        if not isinstance(volumes, dict) or set(volumes) != {
            "apiuploads",
            "caddydata",
            "pgdata",
        }:
            raise CommandFailure("compose_volume_policy")
        if any(
            isinstance(config, dict) and config.get("external") is True
            for config in volumes.values()
        ):
            raise CommandFailure("compose_volume_policy")

    def _check_compose_policy(
        self,
        artifact: RollbackArtifact,
        images: ImagePair,
        compose_argv: list[str],
    ) -> None:
        config_result = self.runner.run(
            "compose_policy",
            [*compose_argv, "config", "--format", "json"],
            timeout=30,
        )
        try:
            resolved_config = json.loads(config_result.stdout)
        except json.JSONDecodeError as exc:
            raise CommandFailure("compose_config_policy") from exc
        self._validate_resolved_config(resolved_config, images)
        resolved_sha256 = _resolved_policy_sha256(resolved_config)
        if not hmac.compare_digest(
            resolved_sha256,
            artifact.resolved_config_sha256,
        ):
            raise CommandFailure("resolved_config_sha256_mismatch")

    def _verify_runtime_inputs(self, artifact: RollbackArtifact) -> None:
        actual_compose_sha = _sha256_regular_file(Path(artifact.compose_file))
        actual_env_sha = _sha256_regular_file(Path(artifact.env_file))
        if not hmac.compare_digest(actual_compose_sha, artifact.compose_sha256):
            raise CommandFailure("compose_sha256_mismatch")
        if not hmac.compare_digest(actual_env_sha, artifact.env_sha256):
            raise CommandFailure("env_sha256_mismatch")

    def apply(
        self, artifact: RollbackArtifact, images: ImagePair, execution_id: str
    ) -> None:
        self._validate_images(images)
        self._verify_runtime_inputs(artifact)
        self._assert_dependencies()
        if EXECUTION_ID_RE.fullmatch(execution_id) is None:
            raise CommandFailure("invalid_execution_id")
        override = self.state_dir / f"{execution_id}.override.yml"
        content = (
            "services:\n"
            "  api:\n"
            f"    image: {images.api_image}\n"
            "  web:\n"
            f"    image: {images.web_image}\n"
        )
        descriptor: int | None = None
        try:
            descriptor = os.open(
                override,
                os.O_WRONLY | os.O_CREAT | os.O_EXCL,
                0o600,
            )
            _write_all(descriptor, content.encode("ascii"))
            os.fsync(descriptor)
            os.close(descriptor)
            descriptor = None
            compose_argv = [
                *self._docker,
                "compose",
                "-p",
                TARGET_PROJECT,
                "--env-file",
                artifact.env_file,
                "-f",
                artifact.compose_file,
                "-f",
                str(override),
            ]
            self._check_compose_policy(artifact, images, compose_argv)
            argv = [
                *compose_argv,
                "up",
                "-d",
                "--no-build",
                "--no-deps",
                "--pull",
                "never",
                "--force-recreate",
                *TARGET_SERVICES,
            ]
            self.runner.run("compose_recreate", argv, timeout=300)
            self._verify_runtime_inputs(artifact)
            self._check_compose_policy(artifact, images, compose_argv)
            self._assert_dependencies()
        except OSError as exc:
            raise CommandFailure("write_override") from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
            with contextlib.suppress(OSError):
                override.unlink()

    def verify(self, images: ImagePair) -> None:
        actual = self.snapshot()
        if actual != images:
            raise CommandFailure("image_verification_mismatch")


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(
        self,
        req: urllib.request.Request,
        fp: Any,
        code: int,
        msg: str,
        headers: Any,
        newurl: str,
    ) -> None:
        return None


class BoundedHealthProbe:
    def __init__(
        self,
        *,
        timeout_seconds: float = 60,
        interval_seconds: float = 2,
        request_timeout_seconds: float = 3,
    ):
        if not 1 <= timeout_seconds <= 300:
            raise ValueError("health timeout must be between 1 and 300 seconds")
        if not 0.1 <= interval_seconds <= 10:
            raise ValueError("health interval must be between 0.1 and 10 seconds")
        if not 0.5 <= request_timeout_seconds <= 10:
            raise ValueError(
                "health request timeout must be between 0.5 and 10 seconds"
            )
        self.timeout_seconds = timeout_seconds
        self.interval_seconds = interval_seconds
        self.request_timeout_seconds = request_timeout_seconds
        self._opener = urllib.request.build_opener(_NoRedirect)

    def _ready(self) -> bool:
        request = urllib.request.Request(
            TARGET_HEALTH_URL,
            headers={"Accept": "application/json"},
            method="GET",
        )
        try:
            with self._opener.open(
                request, timeout=self.request_timeout_seconds
            ) as response:
                if response.status != 200:
                    return False
                raw = response.read(MAX_HEALTH_BODY_BYTES + 1)
        except (OSError, urllib.error.URLError, ValueError):
            return False
        if len(raw) > MAX_HEALTH_BODY_BYTES:
            return False
        try:
            payload = json.loads(raw)
        except (UnicodeDecodeError, json.JSONDecodeError):
            return False
        return (
            isinstance(payload, dict)
            and payload.get("status") == "ok"
            and payload.get("db") is True
            and payload.get("engine") is True
        )

    def wait_ready(self) -> None:
        deadline = time.monotonic() + self.timeout_seconds
        while True:
            if self._ready():
                return
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise CommandFailure("health_timeout")
            time.sleep(min(self.interval_seconds, remaining))


class ProcessFileLock:
    """Kernel-backed lock; a crashed process releases it automatically."""

    def __init__(self, path: Path):
        self.path = path
        self._handle: Any = None

    def __enter__(self) -> ProcessFileLock:
        if self.path.is_symlink():
            raise ExecutorError("lock_path_unsafe", http_status=500)
        flags = os.O_RDWR | os.O_CREAT
        if hasattr(os, "O_NOFOLLOW"):
            flags |= os.O_NOFOLLOW
        try:
            descriptor = os.open(self.path, flags, 0o600)
            info = os.fstat(descriptor)
            if not stat.S_ISREG(info.st_mode):
                os.close(descriptor)
                raise ExecutorError("lock_path_unsafe", http_status=500)
            if os.name != "nt" and (
                info.st_uid != os.geteuid()
                or info.st_mode & (stat.S_IRWXG | stat.S_IRWXO)
            ):
                os.close(descriptor)
                raise ExecutorError("lock_permissions_unsafe", http_status=500)
            self._handle = os.fdopen(descriptor, "r+b", closefd=True)
            if os.name == "nt":
                import msvcrt

                self._handle.seek(0)
                if self._handle.read(1) == b"":
                    self._handle.write(b"0")
                    self._handle.flush()
                self._handle.seek(0)
                msvcrt.locking(self._handle.fileno(), msvcrt.LK_NBLCK, 1)
            else:
                import fcntl

                fcntl.flock(self._handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError as exc:
            if self._handle is not None:
                self._handle.close()
            self._handle = None
            raise ExecutorError("executor_busy", http_status=503) from exc
        return self

    def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
        assert self._handle is not None
        try:
            if os.name == "nt":
                import msvcrt

                self._handle.seek(0)
                msvcrt.locking(self._handle.fileno(), msvcrt.LK_UNLCK, 1)
            else:
                import fcntl

                fcntl.flock(self._handle.fileno(), fcntl.LOCK_UN)
        finally:
            self._handle.close()
            self._handle = None


class DurableJournal:
    def __init__(self, state_dir: Path, *, integrity_key: bytes):
        if not isinstance(integrity_key, bytes) or len(integrity_key) < 32:
            raise ValueError("journal integrity key must contain at least 32 bytes")
        self.state_dir = state_dir
        self.path = state_dir / "journal.jsonl"
        self.lock_path = state_dir / "executor.lock"
        self._integrity_key = integrity_key
        self._integrity_key_id = hashlib.sha256(integrity_key).hexdigest()[:16]

    def _record_hash(self, value: dict[str, Any]) -> str:
        return hmac.new(
            self._integrity_key,
            _canonical_bytes(value),
            hashlib.sha256,
        ).hexdigest()

    def lock(self) -> ProcessFileLock:
        return ProcessFileLock(self.lock_path)

    def read(self) -> list[dict[str, Any]]:
        if not self.path.exists():
            return []
        if self.path.is_symlink():
            raise ExecutorError("journal_path_unsafe", http_status=500)
        records: list[dict[str, Any]] = []
        previous_hash = "0" * 64
        try:
            flags = os.O_RDONLY
            if hasattr(os, "O_NOFOLLOW"):
                flags |= os.O_NOFOLLOW
            descriptor = os.open(self.path, flags)
            info = os.fstat(descriptor)
            if not stat.S_ISREG(info.st_mode):
                os.close(descriptor)
                raise ExecutorError("journal_path_unsafe", http_status=500)
            if os.name != "nt" and (
                info.st_uid != os.geteuid()
                or info.st_mode & (stat.S_IRWXG | stat.S_IRWXO)
            ):
                os.close(descriptor)
                raise ExecutorError("journal_permissions_unsafe", http_status=500)
            with os.fdopen(descriptor, "r", encoding="utf-8") as handle:
                for raw_line in handle:
                    if not raw_line.endswith("\n"):
                        raise ExecutorError("journal_truncated", http_status=500)
                    record = json.loads(raw_line)
                    if not isinstance(record, dict):
                        raise ExecutorError("journal_invalid", http_status=500)
                    record_hash = record.get("record_hash")
                    unhashed = dict(record)
                    unhashed.pop("record_hash", None)
                    if (
                        record.get("schema_version") != JOURNAL_SCHEMA
                        or record.get("integrity_key_id") != self._integrity_key_id
                        or record.get("previous_hash") != previous_hash
                        or not isinstance(record_hash, str)
                        or SHA256_RE.fullmatch(record_hash) is None
                        or not hmac.compare_digest(
                            self._record_hash(unhashed), record_hash
                        )
                    ):
                        raise ExecutorError(
                            "journal_integrity_failure", http_status=500
                        )
                    records.append(record)
                    previous_hash = record_hash
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise ExecutorError("journal_unreadable", http_status=500) from exc
        return records

    def append(self, event: dict[str, Any]) -> dict[str, Any]:
        records = self.read()
        previous_hash = records[-1]["record_hash"] if records else "0" * 64
        record = {
            "schema_version": JOURNAL_SCHEMA,
            "recorded_at": _utc_now(),
            "integrity_key_id": self._integrity_key_id,
            "previous_hash": previous_hash,
            **event,
        }
        record["record_hash"] = self._record_hash(record)
        raw = _canonical_bytes(record) + b"\n"
        try:
            if self.path.is_symlink():
                raise ExecutorError("journal_path_unsafe", http_status=500)
            flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT
            if hasattr(os, "O_NOFOLLOW"):
                flags |= os.O_NOFOLLOW
            descriptor = os.open(self.path, flags, 0o600)
            try:
                info = os.fstat(descriptor)
                if not stat.S_ISREG(info.st_mode):
                    raise ExecutorError("journal_path_unsafe", http_status=500)
                if os.name != "nt" and (
                    info.st_uid != os.geteuid()
                    or info.st_mode & (stat.S_IRWXG | stat.S_IRWXO)
                ):
                    raise ExecutorError("journal_permissions_unsafe", http_status=500)
                _write_all(descriptor, raw)
                os.fsync(descriptor)
            finally:
                os.close(descriptor)
            _fsync_directory(self.state_dir)
        except OSError as exc:
            raise ExecutorError("journal_write_failure", http_status=500) from exc
        return record

    def latest_by_key(self, key: str) -> dict[str, Any] | None:
        matches = [
            record for record in self.read() if record.get("idempotency_key") == key
        ]
        return matches[-1] if matches else None

    def latest_records(self) -> dict[str, dict[str, Any]]:
        latest: dict[str, dict[str, Any]] = {}
        for record in self.read():
            key = record.get("idempotency_key")
            if isinstance(key, str):
                latest[key] = record
        return latest


class NasPreviewRollbackExecutor:
    def __init__(
        self,
        *,
        manifest: RollbackManifest,
        runtime: RuntimeController,
        health: HealthProbe,
        journal: DurableJournal,
    ):
        self.manifest = manifest
        self.runtime = runtime
        self.health = health
        self.journal = journal
        self._blocked = False

    @property
    def ready(self) -> bool:
        return not self._blocked

    @staticmethod
    def _execution_id(request_hash: str, idempotency_key: str) -> str:
        digest = hashlib.sha256(
            f"{idempotency_key}|{request_hash}".encode("ascii")
        ).hexdigest()
        return f"nas-g8-{digest[:24]}"

    @staticmethod
    def _receipt(
        request: RollbackRequest,
        execution_id: str,
        evidence_ref: str,
    ) -> dict[str, Any]:
        return {
            "schema_version": REQUEST_SCHEMA,
            "status": "executed",
            "execution_id": execution_id,
            "idempotency_key": request.idempotency_key,
            "rollback_scope": request.rollback_scope,
            "target_kind": request.target_kind,
            "target_id": request.target_id,
            "artifact_record_id": request.artifact_record_id,
            "artifact_sha256": request.artifact_sha256,
            "evidence_refs": [evidence_ref],
        }

    def _restore_previous(
        self,
        artifact: RollbackArtifact,
        previous: ImagePair,
        execution_id: str,
    ) -> None:
        self.runtime.ensure_images_present(previous)
        self.runtime.apply(artifact, previous, execution_id)
        self.runtime.verify(previous)
        self.health.wait_ready()
        self.runtime.verify(previous)

    def recover_incomplete(self) -> None:
        """Conservatively restore pre-mutation images after a process crash."""

        with self.journal.lock():
            for key, record in self.journal.latest_records().items():
                event = record.get("event")
                unresolved_failure = (
                    event == "failed" and record.get("rollback_status") == "unverified"
                )
                if event not in {"started", "prepared"} and not unresolved_failure:
                    continue
                common = {
                    "event": "failed",
                    "execution_id": record.get("execution_id"),
                    "idempotency_key": key,
                    "request_hash": record.get("request_hash"),
                    "failure_code": "recovered_incomplete_execution",
                }
                if event == "started":
                    self.journal.append({**common, "rollback_status": "not_started"})
                    continue
                artifact = self.manifest.artifacts.get(
                    (record.get("artifact_id"), record.get("artifact_sha256"))
                )
                try:
                    if artifact is None:
                        raise CommandFailure("recovery_artifact_missing")
                    if record.get("artifact_binding_sha256") != _sha256(
                        asdict(artifact)
                    ):
                        raise CommandFailure("recovery_artifact_mismatch")
                    api_image = record.get("previous_api_image")
                    web_image = record.get("previous_web_image")
                    if (
                        not isinstance(api_image, str)
                        or IMAGE_RE.fullmatch(api_image) is None
                        or not isinstance(web_image, str)
                        or IMAGE_RE.fullmatch(web_image) is None
                    ):
                        raise CommandFailure("recovery_images_invalid")
                    execution_id = record.get("execution_id")
                    if (
                        not isinstance(execution_id, str)
                        or EXECUTION_ID_RE.fullmatch(execution_id) is None
                    ):
                        raise CommandFailure("recovery_execution_id_invalid")
                    self._restore_previous(
                        artifact,
                        ImagePair(api_image, web_image),
                        execution_id,
                    )
                except Exception:
                    self._blocked = True
                    self.journal.append({**common, "rollback_status": "unverified"})
                    raise ExecutorError(
                        "rollback_unverified", http_status=503
                    ) from None
                self.journal.append({**common, "rollback_status": "verified"})

    def execute(self, request: RollbackRequest) -> dict[str, Any]:
        if self._blocked:
            raise ExecutorError("rollback_unverified", http_status=503)
        artifact = self.manifest.bind(request)
        request_hash = _sha256(request.canonical_payload())
        execution_id = self._execution_id(request_hash, request.idempotency_key)
        evidence_ref = (
            f"audit://vignette-nas-preview/rollback-executions/{execution_id}"
        )
        with self.journal.lock():
            existing = self.journal.latest_by_key(request.idempotency_key)
            if existing is not None:
                if existing.get("request_hash") != request_hash:
                    raise ExecutorError("idempotency_binding_conflict", http_status=409)
                if existing.get("event") == "executed":
                    receipt = existing.get("receipt")
                    if not isinstance(receipt, dict):
                        raise ExecutorError("journal_receipt_invalid", http_status=500)
                    return receipt
                raise ExecutorError("idempotency_terminal_conflict", http_status=409)

            self.journal.append(
                {
                    "event": "started",
                    "execution_id": execution_id,
                    "idempotency_key": request.idempotency_key,
                    "request_hash": request_hash,
                    "manifest_sha256": self.manifest.content_sha256,
                    "artifact_binding_sha256": _sha256(asdict(artifact)),
                }
            )
            previous: ImagePair | None = None
            prepared = False
            failure_code = "execution_failed"
            receipt = self._receipt(request, execution_id, evidence_ref)
            try:
                previous = self.runtime.snapshot()
                desired = ImagePair(artifact.api_image, artifact.web_image)
                self.runtime.ensure_images_present(desired)
                self.journal.append(
                    {
                        "event": "prepared",
                        "execution_id": execution_id,
                        "idempotency_key": request.idempotency_key,
                        "request_hash": request_hash,
                        "artifact_id": artifact.artifact_id,
                        "artifact_sha256": artifact.artifact_sha256,
                        "artifact_binding_sha256": _sha256(asdict(artifact)),
                        "previous_api_image": previous.api_image,
                        "previous_web_image": previous.web_image,
                    }
                )
                prepared = True
                self.runtime.apply(artifact, desired, execution_id)
                self.runtime.verify(desired)
                self.health.wait_ready()
                self.runtime.verify(desired)
                self.journal.append(
                    {
                        "event": "executed",
                        "execution_id": execution_id,
                        "idempotency_key": request.idempotency_key,
                        "request_hash": request_hash,
                        "receipt": receipt,
                    }
                )
            except Exception:
                rollback_status = "not_started"
                if previous is not None and prepared:
                    try:
                        self._restore_previous(artifact, previous, execution_id)
                    except Exception:
                        rollback_status = "unverified"
                        failure_code = "rollback_unverified"
                    else:
                        rollback_status = "verified"
                failure_event: dict[str, Any] = {
                    "event": "failed",
                    "execution_id": execution_id,
                    "idempotency_key": request.idempotency_key,
                    "request_hash": request_hash,
                    "failure_code": failure_code,
                    "rollback_status": rollback_status,
                }
                if previous is not None and prepared:
                    failure_event.update(
                        {
                            "artifact_id": artifact.artifact_id,
                            "artifact_sha256": artifact.artifact_sha256,
                            "artifact_binding_sha256": _sha256(asdict(artifact)),
                            "previous_api_image": previous.api_image,
                            "previous_web_image": previous.web_image,
                        }
                    )
                self.journal.append(failure_event)
                if rollback_status == "unverified":
                    self._blocked = True
                raise ExecutorError(failure_code, http_status=503) from None
            return receipt


def authenticate(presented: str, configured: str) -> bool:
    """Always use the constant-time primitive for token comparison."""

    if not isinstance(presented, str):
        presented = ""
    return hmac.compare_digest(presented.encode("utf-8"), configured.encode("utf-8"))


class QuietThreadingHTTPServer(ThreadingHTTPServer):
    daemon_threads = True
    request_queue_size = 16

    def __init__(self, *args: Any, **kwargs: Any):
        self._request_slots = threading.BoundedSemaphore(16)
        super().__init__(*args, **kwargs)

    def process_request(self, request: Any, client_address: Any) -> None:
        if not self._request_slots.acquire(blocking=False):
            self.shutdown_request(request)
            return
        try:
            super().process_request(request, client_address)
        except Exception:
            self._request_slots.release()
            raise

    def process_request_thread(self, request: Any, client_address: Any) -> None:
        try:
            super().process_request_thread(request, client_address)
        finally:
            self._request_slots.release()

    def handle_error(self, request: Any, client_address: Any) -> None:
        return None


def _handler(
    *, token: str, executor: NasPreviewRollbackExecutor
) -> type[BaseHTTPRequestHandler]:
    class Handler(BaseHTTPRequestHandler):
        server_version = "VignetteRollbackExecutor/1"
        sys_version = ""

        def setup(self) -> None:
            super().setup()
            self.connection.settimeout(5)

        def log_message(self, format: str, *args: object) -> None:
            return None

        def _json(self, status: int, payload: dict[str, Any]) -> None:
            body = _canonical_bytes(payload)
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.send_header("X-Content-Type-Options", "nosniff")
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self) -> None:
            if self.path != "/healthz":
                self._json(404, {"detail": "not_found"})
                return
            if not authenticate(self.headers.get(TOKEN_HEADER, ""), token):
                self._json(403, {"detail": "forbidden"})
                return
            if not executor.ready:
                self._json(503, {"detail": "rollback_unverified"})
                return
            self._json(
                200,
                {
                    "status": "ok",
                    "target": TARGET_PROJECT,
                    "manifest_sha256": executor.manifest.content_sha256,
                },
            )

        def do_POST(self) -> None:
            if self.path != "/rollback":
                self._json(404, {"detail": "not_found"})
                return
            if not authenticate(self.headers.get(TOKEN_HEADER, ""), token):
                self._json(403, {"detail": "forbidden"})
                return
            if self.headers.get("Transfer-Encoding") is not None:
                self._json(400, {"detail": "invalid_transport"})
                return
            try:
                content_length = int(self.headers.get("Content-Length", "0"))
            except ValueError:
                self._json(400, {"detail": "invalid_content_length"})
                return
            if content_length < 1 or content_length > MAX_BODY_BYTES:
                self._json(413, {"detail": "request_size_rejected"})
                return
            try:
                deadline = time.monotonic() + 5
                remaining = content_length
                chunks: list[bytes] = []
                while remaining:
                    timeout = deadline - time.monotonic()
                    if timeout <= 0:
                        raise TimeoutError
                    self.connection.settimeout(timeout)
                    chunk = self.rfile.read1(min(remaining, 65_536))
                    if not chunk:
                        break
                    chunks.append(chunk)
                    remaining -= len(chunk)
                raw = b"".join(chunks)
                if len(raw) != content_length:
                    self._json(400, {"detail": "incomplete_body"})
                    return
                payload = json.loads(raw)
                request = RollbackRequest.parse(payload)
                receipt = executor.execute(request)
            except (TimeoutError, OSError):
                self._json(408, {"detail": "request_timeout"})
                return
            except (UnicodeDecodeError, json.JSONDecodeError):
                self._json(400, {"detail": "invalid_json"})
                return
            except ExecutorError as exc:
                self._json(exc.http_status, {"detail": exc.code})
                return
            self._json(200, receipt)

    return Handler


def _validate_posix_host_path_chain(
    *,
    root: Path,
    state_dir: Path,
    manifest_path: Path,
    runtime_paths: tuple[Path, ...],
    current_euid: int,
    lstat_func: Callable[[Path], os.stat_result],
) -> None:
    expected: dict[Path, str] = {
        root: "directory",
        state_dir: "directory",
        state_dir / "runtime": "directory",
        manifest_path: "file",
    }
    for leaf in (manifest_path, *runtime_paths):
        try:
            relative = leaf.relative_to(root)
        except ValueError as exc:
            raise ExecutorError("host_path_outside_target", http_status=500) from exc
        cursor = root
        for part in relative.parts[:-1]:
            cursor /= part
            expected[cursor] = "directory"
        expected[leaf] = "file"
    for path, expected_kind in sorted(
        expected.items(), key=lambda item: len(item[0].parts)
    ):
        try:
            info = lstat_func(path)
        except OSError as exc:
            raise ExecutorError("host_path_unavailable", http_status=500) from exc
        if stat.S_ISLNK(info.st_mode):
            raise ExecutorError("host_path_symlink_unsafe", http_status=500)
        if info.st_uid != current_euid:
            raise ExecutorError("host_path_owner_unsafe", http_status=500)
        if info.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
            raise ExecutorError("host_path_permissions_unsafe", http_status=500)
        actual_kind_ok = (
            stat.S_ISDIR(info.st_mode)
            if expected_kind == "directory"
            else stat.S_ISREG(info.st_mode)
        )
        if not actual_kind_ok:
            raise ExecutorError("host_path_type_unsafe", http_status=500)


def _require_secure_host_paths(
    manifest_path: Path,
    state_dir: Path,
    *,
    runtime_paths: tuple[Path, ...] = (),
) -> None:
    if state_dir != DEFAULT_STATE_DIR:
        raise ExecutorError("state_dir_not_allowlisted", http_status=500)
    if state_dir.is_symlink():
        raise ExecutorError("state_dir_not_canonical", http_status=500)
    try:
        resolved_root = Path(str(TARGET_ROOT)).resolve(strict=True)
        resolved_manifest = manifest_path.resolve(strict=True)
        resolved_state = state_dir.resolve(strict=True)
        resolved_runtime_paths = tuple(
            path.resolve(strict=True) for path in runtime_paths
        )
    except OSError as exc:
        raise ExecutorError("host_path_unavailable", http_status=500) from exc
    if resolved_root != Path(str(TARGET_ROOT)):
        raise ExecutorError("target_root_not_canonical", http_status=500)
    if resolved_state != state_dir:
        raise ExecutorError("state_dir_not_canonical", http_status=500)
    try:
        resolved_manifest.relative_to(resolved_root)
        resolved_state.relative_to(resolved_root)
    except ValueError as exc:
        raise ExecutorError("host_path_outside_target", http_status=500) from exc
    if any(
        resolved != original
        for resolved, original in zip(resolved_runtime_paths, runtime_paths)
    ):
        raise ExecutorError("runtime_path_not_canonical", http_status=500)
    if os.name != "nt":
        _validate_posix_host_path_chain(
            root=Path(str(TARGET_ROOT)),
            state_dir=state_dir,
            manifest_path=manifest_path,
            runtime_paths=runtime_paths,
            current_euid=os.geteuid(),
            lstat_func=lambda path: path.lstat(),
        )


def _enabled(value: str | None) -> bool:
    return isinstance(value, str) and value.strip().lower() in {
        "1",
        "true",
        "yes",
        "on",
    }


def _configured_token() -> str:
    token = os.environ.get(TOKEN_ENV, "")
    forbidden = {"change-me", "changeme", "placeholder", "secret", "test-token"}
    if len(token) < 32 or token.strip().lower() in forbidden:
        raise ExecutorError("executor_token_invalid", http_status=500)
    return token


def _configured_journal_key() -> bytes:
    value = os.environ.get(JOURNAL_KEY_ENV, "")
    if len(value) < 32:
        raise ExecutorError("journal_key_invalid", http_status=500)
    return hashlib.sha256(value.encode("utf-8")).digest()


def _secure_docker_binary() -> str:
    if any(
        os.environ.get(name)
        for name in ("DOCKER_CONFIG", "DOCKER_CONTEXT", "DOCKER_HOST")
    ):
        raise ExecutorError("docker_context_not_local", http_status=500)
    discovered = shutil.which("docker")
    if discovered is None:
        raise ExecutorError("docker_binary_unavailable", http_status=500)
    try:
        resolved = Path(discovered).resolve(strict=True)
        info = resolved.stat()
    except OSError as exc:
        raise ExecutorError("docker_binary_unavailable", http_status=500) from exc
    allowlist = {
        Path("/usr/bin/docker"),
        Path("/usr/local/bin/docker"),
        Path("/var/packages/ContainerManager/target/usr/bin/docker"),
    }
    if resolved not in allowlist:
        raise ExecutorError("docker_binary_not_allowlisted", http_status=500)
    if info.st_uid != 0 or info.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
        raise ExecutorError("docker_binary_permissions_unsafe", http_status=500)
    socket_path = Path("/var/run/docker.sock")
    try:
        socket_info = socket_path.stat()
    except OSError as exc:
        raise ExecutorError("docker_socket_unavailable", http_status=500) from exc
    if (
        not stat.S_ISSOCK(socket_info.st_mode)
        or socket_info.st_uid != 0
        or socket_info.st_mode & stat.S_IWOTH
    ):
        raise ExecutorError("docker_socket_permissions_unsafe", http_status=500)
    return str(resolved)


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--manifest", type=Path, required=True)
    parser.add_argument("--manifest-sha256", required=True)
    parser.add_argument("--state-dir", type=Path, default=DEFAULT_STATE_DIR)
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=18149)
    parser.add_argument("--health-timeout-seconds", type=float, default=60)
    parser.add_argument("--health-interval-seconds", type=float, default=2)
    parser.add_argument("--health-request-timeout-seconds", type=float, default=3)
    parser.add_argument("--tls-cert", type=Path)
    parser.add_argument("--tls-key", type=Path)
    parser.add_argument("--enable", action="store_true")
    parser.add_argument("--check-config", action="store_true")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _build_parser().parse_args(argv)
    manifest = load_manifest(args.manifest, args.manifest_sha256)
    if args.check_config:
        print(
            json.dumps(
                {
                    "ok": True,
                    "enabled": False,
                    "manifest_sha256": manifest.content_sha256,
                    "artifacts": len(manifest.artifacts),
                    "target": TARGET_PROJECT,
                },
                separators=(",", ":"),
            )
        )
        return 0
    if not args.enable or not _enabled(os.environ.get(ENABLED_ENV)):
        raise SystemExit(
            "executor is disabled; explicit CLI and environment opt-in required"
        )
    token = _configured_token()
    journal_integrity_key = _configured_journal_key()
    docker_bin = _secure_docker_binary()
    runtime_paths = tuple(
        path
        for artifact in manifest.artifacts.values()
        for path in (Path(artifact.compose_file), Path(artifact.env_file))
    )
    _require_secure_host_paths(
        args.manifest,
        args.state_dir,
        runtime_paths=runtime_paths,
    )
    for artifact in manifest.artifacts.values():
        for path in (Path(artifact.compose_file), Path(artifact.env_file)):
            if (
                not path.is_file()
                or path.is_symlink()
                or path.resolve(strict=True) != path
            ):
                raise ExecutorError("runtime_path_unavailable", http_status=500)
    health = BoundedHealthProbe(
        timeout_seconds=args.health_timeout_seconds,
        interval_seconds=args.health_interval_seconds,
        request_timeout_seconds=args.health_request_timeout_seconds,
    )
    journal = DurableJournal(
        args.state_dir,
        integrity_key=journal_integrity_key,
    )
    with journal.lock():
        journal.read()
    executor = NasPreviewRollbackExecutor(
        manifest=manifest,
        runtime=DockerRuntimeController(
            SubprocessRunner(),
            state_dir=args.state_dir,
            docker_bin=docker_bin,
        ),
        health=health,
        journal=journal,
    )
    executor.recover_incomplete()
    if (args.tls_cert is None) != (args.tls_key is None):
        raise SystemExit("--tls-cert and --tls-key must be configured together")
    tls_enabled = args.tls_cert is not None
    if not tls_enabled and args.host not in {"127.0.0.1", "::1", "localhost"}:
        raise SystemExit("non-loopback binding requires TLS")
    server = QuietThreadingHTTPServer(
        (args.host, args.port),
        _handler(token=token, executor=executor),
    )
    if tls_enabled:
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.minimum_version = ssl.TLSVersion.TLSv1_2
        context.load_cert_chain(args.tls_cert, args.tls_key)
        server.socket = context.wrap_socket(server.socket, server_side=True)
    print(
        json.dumps(
            {
                "ready": True,
                "enabled": True,
                "host": args.host,
                "port": args.port,
                "tls": tls_enabled,
                "target": TARGET_PROJECT,
            },
            separators=(",", ":"),
        ),
        flush=True,
    )
    server.serve_forever()
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except ExecutorError as exc:
        print(
            json.dumps({"ok": False, "error": exc.code}, separators=(",", ":")),
            file=sys.stderr,
        )
        raise SystemExit(2) from None