"""Run the learner closed loop only inside a disposable, HEAD-bound stack.

The runner is fail closed.  It never accepts caller-selected runtime ports or a
database URL, constructs every credential and endpoint itself, and tears down
only resources carrying its exact Compose project and sentinel labels.

This script is intentionally not registered in Task Scheduler.  Scheduling is
allowed only after one explicit ``--execute`` run produces a GREEN receipt.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import secrets
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Mapping, Protocol, Sequence
from urllib.parse import urlsplit


REPO_ROOT = Path(__file__).resolve().parents[1]
COMPOSE_FILE = REPO_ROOT / "infra" / "docker-compose.yml"
FIXTURE_SCRIPT = (
    REPO_ROOT
    / "apps"
    / "web"
    / "e2e"
    / "harness"
    / "prepare-returned-practice-db.py"
)
PERIODIC_SPEC = "e2e/periodic-learner-real-closed-loop.spec.ts"
RETURNED_SPEC = "e2e/returned-practice-db-closed-loop.spec.ts"
SCHEMA_VERSION = "vignette.periodic-learner-e2e-receipt.v1"
PLAYWRIGHT_RETRIES = 0
FORBIDDEN_PORTS = frozenset({8001, 55432, 9099})
FORBIDDEN_MARKERS = (
    "vignette.chanpaca.net",
    "api-vignette.chanpaca.net",
    "100.116.83.60",
    "vignette-preview-20260807",
    "vignette-dev-db",
    "/volume1/",
    "\\\\100.116.83.60\\",
)
RUNTIME_ENV_KEYS = frozenset(
    {
        "DATABASE_URL",
        "DATABASE_ADMIN_URL",
        "DB_PORT",
        "POSTGRES_PORT",
        "ENGINE_URL",
        "ENGINE_PORT",
        "HTTP_PORT",
        "HTTPS_PORT",
        "PLAYWRIGHT_BASE_URL",
        "E2E_PUBLIC_BASE_URL",
        "PUBLIC_API_BASE",
        "FRONTEND_BASE_URL",
        "FRONTEND_ORIGIN_MAP",
        "CORS_ORIGINS",
        "AUTH_DEV_LOGIN_EXTRA_ORIGINS",
        "COMPOSE_PROJECT_NAME",
        "COMPOSE_FILE",
        "COMPOSE_PROFILES",
        "DOCKER_CONTEXT",
        "DOCKER_HOST",
    }
)
INHERITED_TARGET_ENV_KEYS = frozenset(
    {
        "DATABASE_URL",
        "DATABASE_ADMIN_URL",
        "DB_PORT",
        "POSTGRES_PORT",
        "PGHOST",
        "PGPORT",
        "ENGINE_URL",
        "ENGINE_PORT",
        "HTTP_PORT",
        "HTTPS_PORT",
        "PLAYWRIGHT_BASE_URL",
        "E2E_PUBLIC_BASE_URL",
        "FRONTEND_BASE_URL",
        "COMPOSE_FILE",
        "COMPOSE_PROJECT_NAME",
    }
)
CRITICAL_UNTRACKED_SUFFIXES = frozenset(
    {".py", ".ts", ".tsx", ".js", ".mjs", ".json", ".yml", ".yaml", ".sql"}
)
PROJECT_RE = re.compile(r"^vignette-periodic-[0-9a-f]{8}-[a-z0-9-]{6,32}$")
HEX_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
SENSITIVE_ENV_KEY_RE = re.compile(
    r"(?:PASSWORD|SECRET|TOKEN|API_KEY)$|^ENGINE_GATEWAY_SHARED_SECRET$"
)
DIAGNOSTIC_TAIL_CHARS = 12_000
EMAIL_RE = re.compile(r"(?i)\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b")
UUID_RE = re.compile(
    r"(?i)\b[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\b"
)
SENSITIVE_QUERY_RE = re.compile(
    r"(?i)([?&](?:code|email|token|secret|session|state)=)[^&\s]+"
)


class GateError(RuntimeError):
    """A bounded, receipt-safe gate failure."""


@dataclass(frozen=True, slots=True)
class CommandResult:
    returncode: int
    stdout: str = ""
    stderr: str = ""
    duration_seconds: float = 0.0


@dataclass(frozen=True, slots=True)
class ProcessHandle:
    pid: int


class Controller(Protocol):
    def run(
        self,
        stage: str,
        argv: Sequence[str],
        *,
        cwd: Path,
        env: Mapping[str, str] | None = None,
        timeout: float,
        check: bool = True,
    ) -> CommandResult: ...

    def start(
        self,
        stage: str,
        argv: Sequence[str],
        *,
        cwd: Path,
        env: Mapping[str, str],
        stdout_path: Path,
        stderr_path: Path,
    ) -> ProcessHandle: ...

    def stop_exact(self, handle: ProcessHandle, *, timeout: float) -> bool: ...

    def http_json(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        timeout: float,
    ) -> dict[str, Any]: ...

    def http_text(self, url: str, *, timeout: float) -> str: ...

    def tcp_listening(self, host: str, port: int, *, timeout: float = 0.25) -> bool: ...

    def sleep(self, seconds: float) -> None: ...


class LocalController:
    def __init__(self) -> None:
        self._processes: dict[int, tuple[subprocess.Popen[bytes], Any, Any]] = {}

    def run(
        self,
        stage: str,
        argv: Sequence[str],
        *,
        cwd: Path,
        env: Mapping[str, str] | None = None,
        timeout: float,
        check: bool = True,
    ) -> CommandResult:
        started = time.monotonic()
        try:
            completed = subprocess.run(
                list(argv),
                cwd=str(cwd),
                env=dict(env) if env is not None else None,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=timeout,
                check=False,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            raise GateError(f"{stage}: command failed to run: {exc}") from exc
        result = CommandResult(
            completed.returncode,
            completed.stdout,
            completed.stderr,
            time.monotonic() - started,
        )
        if check and result.returncode != 0:
            tail = (result.stderr or result.stdout).strip()[-1200:]
            raise GateError(f"{stage}: exit {result.returncode}: {tail}")
        return result

    def start(
        self,
        stage: str,
        argv: Sequence[str],
        *,
        cwd: Path,
        env: Mapping[str, str],
        stdout_path: Path,
        stderr_path: Path,
    ) -> ProcessHandle:
        stdout_handle = stdout_path.open("wb")
        stderr_handle = stderr_path.open("wb")
        try:
            process = subprocess.Popen(
                list(argv),
                cwd=str(cwd),
                env=dict(env),
                stdin=subprocess.DEVNULL,
                stdout=stdout_handle,
                stderr=stderr_handle,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            )
        except OSError as exc:
            stdout_handle.close()
            stderr_handle.close()
            raise GateError(f"{stage}: process failed to start: {exc}") from exc
        self._processes[process.pid] = (process, stdout_handle, stderr_handle)
        return ProcessHandle(process.pid)

    def stop_exact(self, handle: ProcessHandle, *, timeout: float) -> bool:
        entry = self._processes.pop(handle.pid, None)
        if entry is None:
            return False
        process, stdout_handle, stderr_handle = entry
        try:
            if process.poll() is None:
                if os.name == "nt":
                    subprocess.run(
                        ["taskkill.exe", "/PID", str(handle.pid), "/T", "/F"],
                        stdin=subprocess.DEVNULL,
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.DEVNULL,
                        timeout=timeout,
                        check=False,
                    )
                else:
                    process.terminate()
                try:
                    process.wait(timeout=timeout)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait(timeout=timeout)
            return process.poll() is not None
        finally:
            stdout_handle.close()
            stderr_handle.close()

    def http_json(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        timeout: float,
    ) -> dict[str, Any]:
        request = urllib.request.Request(url, headers=dict(headers or {}))
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                payload = json.loads(response.read().decode("utf-8"))
        except (OSError, urllib.error.HTTPError, json.JSONDecodeError) as exc:
            raise GateError(f"HTTP JSON probe failed: {url}: {exc}") from exc
        if not isinstance(payload, dict):
            raise GateError(f"HTTP JSON probe did not return an object: {url}")
        return payload

    def http_text(self, url: str, *, timeout: float) -> str:
        request = urllib.request.Request(url)
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return response.read().decode("utf-8", errors="replace")
        except (OSError, urllib.error.HTTPError) as exc:
            raise GateError(f"HTTP text probe failed: {url}: {exc}") from exc

    def tcp_listening(self, host: str, port: int, *, timeout: float = 0.25) -> bool:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
            probe.settimeout(timeout)
            return probe.connect_ex((host, port)) == 0

    def sleep(self, seconds: float) -> None:
        time.sleep(seconds)


@dataclass(frozen=True, slots=True)
class SourceIdentity:
    head: str
    tree: str


@dataclass(frozen=True, slots=True)
class RuntimeIdentity:
    run_id: str
    project: str
    sentinel: str
    http_port: int
    https_port: int
    database_port: int
    engine_port: int

    @property
    def ports(self) -> tuple[int, ...]:
        return (
            self.http_port,
            self.https_port,
            self.database_port,
            self.engine_port,
        )


@dataclass(frozen=True, slots=True)
class RunnerConfig:
    receipt_path: Path
    python_exe: str
    node_exe: str
    docker_exe: str
    execute: bool
    request_timeout: float = 240.0
    readiness_timeout: float = 600.0


@dataclass(slots=True)
class ExecutionState:
    source: SourceIdentity | None = None
    runtime: RuntimeIdentity | None = None
    temp_dir: Path | None = None
    env_file: Path | None = None
    override_file: Path | None = None
    engine: ProcessHandle | None = None
    docker_context: str | None = None
    stack_attempted: bool = False
    stages: list[dict[str, Any]] = field(default_factory=list)
    proof: dict[str, Any] = field(default_factory=dict)


def _sha256_path(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _now_iso() -> str:
    return datetime.now(UTC).isoformat().replace("+00:00", "Z")


def _safe_error(exc: BaseException) -> str:
    return str(exc).replace(str(REPO_ROOT), "<repo>")[-1600:]


def _sanitize_diagnostic_text(text: str, environ: Mapping[str, str]) -> str:
    sanitized = text.replace(str(REPO_ROOT), "<repo>")
    sensitive_values = sorted(
        {
            value
            for key, value in environ.items()
            if value and SENSITIVE_ENV_KEY_RE.search(key)
        },
        key=len,
        reverse=True,
    )
    for value in sensitive_values:
        sanitized = sanitized.replace(value, "<redacted>")
    sanitized = EMAIL_RE.sub("<redacted-email>", sanitized)
    sanitized = UUID_RE.sub("<redacted-uuid>", sanitized)
    sanitized = SENSITIVE_QUERY_RE.sub(r"\1<redacted>", sanitized)
    if any(value in sanitized for value in sensitive_values):
        raise GateError("failure diagnostics retained a runtime secret")
    return sanitized[-DIAGNOSTIC_TAIL_CHARS:]


def assert_safe_value(label: str, value: str) -> None:
    normalized = value.strip().lower().replace("\\", "/")
    if any(marker.lower().replace("\\", "/") in normalized for marker in FORBIDDEN_MARKERS):
        raise GateError(f"{label}: public/NAS marker is forbidden")
    if normalized.isdigit() and int(normalized) in FORBIDDEN_PORTS:
        raise GateError(f"{label}: protected runtime port is forbidden")
    if re.search(r":(?:8001|55432|9099)(?:\D|$)", normalized):
        raise GateError(f"{label}: protected runtime port is forbidden")
    parsed = urlsplit(value)
    try:
        parsed_port = parsed.port
    except ValueError as exc:
        raise GateError(f"{label}: invalid endpoint") from exc
    if parsed_port in FORBIDDEN_PORTS:
        raise GateError(f"{label}: protected runtime port is forbidden")


def assert_safe_invocation(argv: Sequence[str], environ: Mapping[str, str]) -> None:
    for index, value in enumerate(argv):
        assert_safe_value(f"argv[{index}]", str(value))
    for key in RUNTIME_ENV_KEYS:
        value = environ.get(key)
        if value:
            assert_safe_value(f"env:{key}", value)


def assert_no_inherited_runtime_targets(environ: Mapping[str, str]) -> None:
    present = sorted(key for key in INHERITED_TARGET_ENV_KEYS if environ.get(key))
    if present:
        raise GateError(
            "inherited runtime target variables are forbidden: " + ", ".join(present)
        )


def assert_loopback_url(label: str, value: str, expected_port: int) -> None:
    assert_safe_value(label, value)
    parsed = urlsplit(value)
    if parsed.scheme not in {"http", "postgresql"}:
        raise GateError(f"{label}: unsupported scheme")
    if parsed.hostname not in {"127.0.0.1", "localhost"}:
        raise GateError(f"{label}: endpoint must be loopback")
    if parsed.port != expected_port:
        raise GateError(f"{label}: endpoint is not bound to the allocated port")


def allocate_unique_ports(count: int) -> tuple[int, ...]:
    sockets: list[socket.socket] = []
    try:
        while len(sockets) < count:
            reservation = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            reservation.bind(("127.0.0.1", 0))
            port = int(reservation.getsockname()[1])
            if port in FORBIDDEN_PORTS or any(
                int(item.getsockname()[1]) == port for item in sockets
            ):
                reservation.close()
                continue
            sockets.append(reservation)
        return tuple(int(item.getsockname()[1]) for item in sockets)
    finally:
        for reservation in sockets:
            reservation.close()


def build_runtime_identity(source: SourceIdentity) -> RuntimeIdentity:
    run_id = f"{datetime.now(UTC).strftime('%Y%m%dT%H%M%S')}-{secrets.token_hex(4)}"
    project = f"vignette-periodic-{source.head[:8]}-{run_id[-8:]}"
    if not PROJECT_RE.fullmatch(project):
        raise GateError("generated Compose project is invalid")
    http_port, https_port, database_port, engine_port = allocate_unique_ports(4)
    sentinel = f"periodic:{source.head}:{source.tree}:{run_id}"
    return RuntimeIdentity(
        run_id,
        project,
        sentinel,
        http_port,
        https_port,
        database_port,
        engine_port,
    )


def validate_runtime_identity(runtime: RuntimeIdentity) -> None:
    if not PROJECT_RE.fullmatch(runtime.project):
        raise GateError("Compose project does not match the disposable namespace")
    if len(set(runtime.ports)) != 4:
        raise GateError("runtime ports must be unique")
    if any(port in FORBIDDEN_PORTS or not 1024 < port < 65536 for port in runtime.ports):
        raise GateError("runtime allocated a protected or invalid port")
    for port in runtime.ports:
        assert_safe_value("resolved-port", str(port))


def _clean_child_env(overrides: Mapping[str, str]) -> dict[str, str]:
    env = os.environ.copy()
    for key in RUNTIME_ENV_KEYS:
        env.pop(key, None)
    env.update(overrides)
    env["NO_PROXY"] = "127.0.0.1,localhost,host.docker.internal"
    env["no_proxy"] = env["NO_PROXY"]
    assert_safe_invocation([], env)
    return env


def _secret(prefix: str) -> str:
    return f"{prefix}-{secrets.token_hex(32)}"


def build_stack_environment(runtime: RuntimeIdentity) -> dict[str, str]:
    base_url = f"http://127.0.0.1:{runtime.http_port}"
    engine_url = f"http://host.docker.internal:{runtime.engine_port}"
    values = {
        "ENVIRONMENT": "dev",
        "POSTGRES_USER": "vignette_owner",
        "POSTGRES_PASSWORD": _secret("owner"),
        "POSTGRES_DB": "vignette_periodic",
        "APP_DB_USER": "vignette_app",
        "APP_DB_PASSWORD": _secret("app"),
        "ENGINE_URL": engine_url,
        "ENGINE_MODE": "claude_cli",
        "ENGINE_GATEWAY_SHARED_SECRET": _secret("engine"),
        "OPENAI_API_KEY": _secret("sk-periodic"),
        "SESSION_SECRET": _secret("session"),
        "VIGNETTE_RUPTURE_INTERNAL_TOKEN": _secret("rupture"),
        "VIGNETTE_PRACTICE_INTERNAL_TOKEN": _secret("practice"),
        "VIGNETTE_CALIBRATION_TRANSFER_INTERNAL_TOKEN": _secret("transfer"),
        "VIGNETTE_SUPERVISION_RESEARCH_INTERNAL_TOKEN": _secret("supervision"),
        "VIGNETTE_MULTIMODAL_ALLIANCE_INTERNAL_TOKEN": _secret("alliance"),
        "VIGNETTE_CONTINUOUS_IMPROVEMENT_INTERNAL_TOKEN": _secret("ci"),
        "OAUTH_GOOGLE_CLIENT_ID": "periodic-local-client",
        "OAUTH_GOOGLE_CLIENT_SECRET": _secret("oauth"),
        "OAUTH_REDIRECT_URI": f"{base_url}/api/auth/callback",
        "AUTH_ALLOWED_EMAIL_DOMAINS": json.dumps(["hs.ac.kr"], separators=(",", ":")),
        "AUTH_TEACHER_EMAILS": "[]",
        "AUTH_ADMIN_EMAILS": "[]",
        "AUTH_DEV_LOGIN_ENABLED": "true",
        "AUTH_DEV_LOGIN_EXTRA_ORIGINS": json.dumps([base_url], separators=(",", ":")),
        "AUTO_SEED_PERSONAS": "true",
        "ALLOW_SEED_PERSONA_FALLBACK": "false",
        "FRONTEND_BASE_URL": base_url,
        "FRONTEND_ORIGIN_MAP": json.dumps({"default": base_url}, separators=(",", ":")),
        "CORS_ORIGINS": json.dumps([base_url], separators=(",", ":")),
        "PUBLIC_API_BASE": "/api",
        "HTTP_PORT": str(runtime.http_port),
        "HTTPS_PORT": str(runtime.https_port),
        "DB_HOST_PORT": str(runtime.database_port),
        "PERIODIC_SENTINEL": runtime.sentinel,
        "VIGNETTE_CONTINUOUS_IMPROVEMENT_PRODUCER_ENABLED": "false",
        "VIGNETTE_CONTINUOUS_IMPROVEMENT_DRIFT_TRIGGER_ENABLED": "false",
        "VIGNETTE_CONTINUOUS_IMPROVEMENT_ROLLBACK_EXECUTOR_ENABLED": "false",
        "NOTIFICATION_EMAIL_PROVIDER": "disabled",
    }
    assert_safe_invocation([], values)
    return values


def write_env_file(path: Path, values: Mapping[str, str]) -> None:
    lines = [f"{key}={value}" for key, value in sorted(values.items())]
    path.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n")


def write_compose_override(path: Path) -> None:
    path.write_text(
        """services:
  db:
    ports:
      - \"127.0.0.1:${DB_HOST_PORT}:5432\"
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
  api:
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
  web:
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
  proxy:
    ports: !override
      - \"127.0.0.1:${HTTP_PORT}:80\"
      - \"127.0.0.1:${HTTPS_PORT}:443\"
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
volumes:
  pgdata:
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
  apiuploads:
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
  caddydata:
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
networks:
  vignette:
    labels:
      com.vignette.periodic-e2e-sentinel: \"${PERIODIC_SENTINEL}\"
""",
        encoding="utf-8",
        newline="\n",
    )


class PeriodicRunner:
    def __init__(self, config: RunnerConfig, controller: Controller) -> None:
        self.config = config
        self.controller = controller
        self.state = ExecutionState()

    def _run(
        self,
        stage: str,
        argv: Sequence[str],
        *,
        cwd: Path = REPO_ROOT,
        env: Mapping[str, str] | None = None,
        timeout: float = 120.0,
        check: bool = True,
    ) -> CommandResult:
        assert_safe_invocation(argv, env or {})
        result = self.controller.run(
            stage,
            argv,
            cwd=cwd,
            env=env,
            timeout=timeout,
            check=check,
        )
        self.state.stages.append(
            {
                "stage": stage,
                "exit_code": result.returncode,
                "duration_seconds": round(result.duration_seconds, 3),
                "stdout_sha256": hashlib.sha256(result.stdout.encode("utf-8")).hexdigest(),
            }
        )
        return result

    def source_identity(self, *, require_clean: bool) -> SourceIdentity:
        head = self._run(
            "source_head",
            ["git.exe", "rev-parse", "HEAD"],
            timeout=20,
        ).stdout.strip()
        tree = self._run(
            "source_tree",
            ["git.exe", "rev-parse", "HEAD^{tree}"],
            timeout=20,
        ).stdout.strip()
        if not HEX_SHA_RE.fullmatch(head) or not HEX_SHA_RE.fullmatch(tree):
            raise GateError("git did not return a canonical HEAD/tree")
        if require_clean:
            tracked = self._run(
                "source_tracked_clean",
                ["git.exe", "status", "--porcelain", "--untracked-files=no"],
                timeout=30,
            ).stdout.strip()
            if tracked:
                raise GateError("tracked worktree differs from HEAD")
            untracked = self._run(
                "source_untracked_inventory",
                ["git.exe", "status", "--porcelain", "--untracked-files=all"],
                timeout=30,
            ).stdout.splitlines()
            dangerous: list[str] = []
            for row in untracked:
                if not row.startswith("?? "):
                    continue
                relative = row[3:].strip().replace("\\", "/")
                candidate = Path(relative)
                if (
                    candidate.suffix.lower() in CRITICAL_UNTRACKED_SUFFIXES
                    and relative.startswith(("apps/api/", "apps/web/", "data/", "infra/"))
                ):
                    dangerous.append(relative)
            if dangerous:
                raise GateError("untracked runtime source is present")
        return SourceIdentity(head, tree)

    def _wait_json(
        self,
        label: str,
        url: str,
        predicate: Any,
        *,
        headers: Mapping[str, str] | None = None,
        timeout: float,
    ) -> dict[str, Any]:
        assert_safe_value(label, url)
        deadline = time.monotonic() + timeout
        last_error = "not attempted"
        while time.monotonic() < deadline:
            try:
                payload = self.controller.http_json(url, headers=headers, timeout=10)
                if predicate(payload):
                    return payload
                last_error = "predicate rejected response"
            except GateError as exc:
                last_error = _safe_error(exc)
            self.controller.sleep(1.0)
        raise GateError(f"{label}: readiness timeout: {last_error}")

    def _compose_argv(self, *tail: str) -> list[str]:
        runtime = self._require_runtime()
        if self.state.env_file is None or self.state.override_file is None:
            raise GateError("Compose files are not initialized")
        return [
            *self._docker_prefix(),
            "compose",
            "-p",
            runtime.project,
            "--env-file",
            str(self.state.env_file),
            "-f",
            str(COMPOSE_FILE),
            "-f",
            str(self.state.override_file),
            *tail,
        ]

    def _docker_prefix(self) -> list[str]:
        if not self.state.docker_context:
            raise GateError("local Docker context is not pinned")
        return [self.config.docker_exe, "--context", self.state.docker_context]

    def _pin_local_docker_context(self) -> dict[str, str]:
        context = self._run(
            "docker_context_show",
            [self.config.docker_exe, "context", "show"],
            timeout=30,
        ).stdout.strip()
        if not context or not re.fullmatch(r"[A-Za-z0-9_.-]+", context):
            raise GateError("Docker context name is not canonical")
        inspected = self._run(
            "docker_context_inspect",
            [self.config.docker_exe, "context", "inspect", context],
            timeout=30,
        )
        try:
            payload = json.loads(inspected.stdout)
            host = str(payload[0]["Endpoints"]["docker"]["Host"])
        except (json.JSONDecodeError, IndexError, KeyError, TypeError) as exc:
            raise GateError("Docker context endpoint is unreadable") from exc
        assert_safe_value("Docker context endpoint", host)
        if not host.lower().startswith("npipe://"):
            raise GateError("Docker context must use the local Windows named pipe")
        self.state.docker_context = context
        return {"context": context, "transport": "npipe", "remote": False}

    def _require_runtime(self) -> RuntimeIdentity:
        if self.state.runtime is None:
            raise GateError("runtime identity is not initialized")
        return self.state.runtime

    def _resource_ids(self, kind: str) -> list[str]:
        runtime = self._require_runtime()
        noun = {"container": "ps", "volume": "volume", "network": "network"}[kind]
        argv = [*self._docker_prefix(), noun]
        if kind == "container":
            argv.extend(["-aq"])
        else:
            argv.extend(["ls", "-q"])
        argv.extend(["--filter", f"label=com.docker.compose.project={runtime.project}"])
        result = self._run(f"inventory_{kind}", argv, timeout=30)
        return [line.strip() for line in result.stdout.splitlines() if line.strip()]

    def _verify_resources_up(self) -> dict[str, int]:
        runtime = self._require_runtime()
        containers = self._resource_ids("container")
        volumes = self._resource_ids("volume")
        networks = self._resource_ids("network")
        if len(containers) != 4 or len(volumes) != 3 or len(networks) != 1:
            raise GateError("Compose did not create the exact disposable resource set")
        labels = self._run(
            "sentinel_container_labels",
            [
                *self._docker_prefix(),
                "inspect",
                "--format",
                '{{ index .Config.Labels "com.vignette.periodic-e2e-sentinel" }}',
                *containers,
            ],
            timeout=30,
        ).stdout.splitlines()
        if len(labels) != 4 or any(value.strip() != runtime.sentinel for value in labels):
            raise GateError("container sentinel label mismatch")
        port_output = self._run(
            "resolved_compose_ports",
            self._compose_argv("port", "proxy", "80"),
            timeout=30,
        ).stdout.strip()
        db_port_output = self._run(
            "resolved_database_port",
            self._compose_argv("port", "db", "5432"),
            timeout=30,
        ).stdout.strip()
        for label, output, expected in (
            ("resolved proxy port", port_output, runtime.http_port),
            ("resolved database port", db_port_output, runtime.database_port),
        ):
            assert_safe_value(label, output)
            if not output.endswith(f":{expected}"):
                raise GateError(f"{label} mismatch")
        return {
            "containers": len(containers),
            "volumes": len(volumes),
            "networks": len(networks),
        }

    def _validate_resolved_compose_config(self, stack_env: Mapping[str, str]) -> str:
        runtime = self._require_runtime()
        result = self._run(
            "resolved_compose_config",
            self._compose_argv("config", "--format", "json"),
            timeout=60,
        )
        try:
            payload = json.loads(result.stdout)
        except json.JSONDecodeError as exc:
            raise GateError("resolved Compose config is not JSON") from exc
        services = payload.get("services")
        if not isinstance(services, dict) or set(services) != {"db", "api", "web", "proxy"}:
            raise GateError("resolved Compose services are not the exact expected set")

        expected_bindings = {
            ("db", 5432): runtime.database_port,
            ("proxy", 80): runtime.http_port,
            ("proxy", 443): runtime.https_port,
        }
        actual_bindings: dict[tuple[str, int], int] = {}
        for service_name, service in services.items():
            if not isinstance(service, dict):
                raise GateError("resolved Compose service is not an object")
            labels = service.get("labels") or {}
            if labels.get("com.vignette.periodic-e2e-sentinel") != runtime.sentinel:
                raise GateError("resolved Compose service sentinel mismatch")
            for binding in service.get("ports") or []:
                if not isinstance(binding, dict):
                    raise GateError("resolved Compose port is not structured")
                host_ip = str(binding.get("host_ip") or "")
                if host_ip != "127.0.0.1":
                    raise GateError("resolved Compose port is not loopback-only")
                target = int(binding.get("target") or 0)
                published = int(binding.get("published") or 0)
                assert_safe_value("resolved Compose published port", str(published))
                actual_bindings[(service_name, target)] = published
        if actual_bindings != expected_bindings:
            raise GateError("resolved Compose port set differs from the generated plan")

        api_environment = services["api"].get("environment") or {}
        if not isinstance(api_environment, dict):
            raise GateError("resolved API environment is not structured")
        database_url = str(api_environment.get("DATABASE_URL") or "")
        assert_safe_value("resolved API DATABASE_URL", database_url)
        parsed_database = urlsplit(database_url)
        if (
            parsed_database.scheme != "postgresql"
            or parsed_database.hostname != "db"
            or parsed_database.port != 5432
            or parsed_database.path != f"/{stack_env['POSTGRES_DB']}"
        ):
            raise GateError("resolved API DATABASE_URL escapes the disposable DB service")
        engine_url = str(api_environment.get("ENGINE_URL") or "")
        assert_safe_value("resolved API ENGINE_URL", engine_url)
        parsed_engine = urlsplit(engine_url)
        if (
            parsed_engine.scheme != "http"
            or parsed_engine.hostname != "host.docker.internal"
            or parsed_engine.port != runtime.engine_port
        ):
            raise GateError("resolved API ENGINE_URL escapes the disposable engine")
        for key in ("FRONTEND_BASE_URL", "OAUTH_REDIRECT_URI"):
            value = str(api_environment.get(key) or "")
            assert_safe_value(f"resolved API {key}", value)
            if urlsplit(value).hostname not in {"127.0.0.1", "localhost"}:
                raise GateError(f"resolved API {key} is not loopback")

        volumes = payload.get("volumes") or {}
        networks = payload.get("networks") or {}
        if len(volumes) != 3 or len(networks) != 1:
            raise GateError("resolved Compose resource count is not disposable-exact")
        for collection in (volumes, networks):
            for resource in collection.values():
                labels = resource.get("labels") if isinstance(resource, dict) else None
                if not isinstance(labels, dict) or labels.get(
                    "com.vignette.periodic-e2e-sentinel"
                ) != runtime.sentinel:
                    raise GateError("resolved Compose resource sentinel mismatch")
        return hashlib.sha256(result.stdout.encode("utf-8")).hexdigest()

    def _install_database_sentinel(self, stack_env: Mapping[str, str]) -> str:
        runtime = self._require_runtime()
        sql = (
            "CREATE TABLE IF NOT EXISTS public.periodic_e2e_sentinel ("
            "run_id text PRIMARY KEY, compose_project text NOT NULL, source_head text NOT NULL, "
            "source_tree text NOT NULL); "
            "INSERT INTO public.periodic_e2e_sentinel VALUES ("
            f"'{runtime.run_id}','{runtime.project}','{self.state.source.head}',"
            f"'{self.state.source.tree}') ON CONFLICT (run_id) DO NOTHING; "
            "SELECT run_id || '|' || compose_project || '|' || source_head || '|' || source_tree "
            f"FROM public.periodic_e2e_sentinel WHERE run_id='{runtime.run_id}';"
        )
        result = self._run(
            "database_sentinel",
            self._compose_argv(
                "exec",
                "-T",
                "db",
                "psql",
                "-v",
                "ON_ERROR_STOP=1",
                "-U",
                stack_env["POSTGRES_USER"],
                "-d",
                stack_env["POSTGRES_DB"],
                "-tAc",
                sql,
            ),
            timeout=60,
        )
        expected = (
            f"{runtime.run_id}|{runtime.project}|{self.state.source.head}|"
            f"{self.state.source.tree}"
        )
        if expected not in result.stdout:
            raise GateError("database sentinel round trip mismatch")
        return hashlib.sha256(expected.encode("utf-8")).hexdigest()

    def _browser_env(
        self,
        runtime: RuntimeIdentity,
        fixture: Path,
        result: Path | None = None,
    ) -> dict[str, str]:
        values = {
            "PLAYWRIGHT_BASE_URL": f"http://127.0.0.1:{runtime.http_port}",
            "PLAYWRIGHT_SKIP_WEB_SERVER": "1",
            "E2E_RETURNED_PRACTICE_DB_CLOSED_LOOP": "1",
            "E2E_RETURNED_PRACTICE_FIXTURE": str(fixture),
            "CI": "1",
        }
        if result is not None:
            values["E2E_PERIODIC_LEARNER_REAL_CLOSED_LOOP"] = "1"
            values["E2E_PERIODIC_LEARNER_RESULT"] = str(result)
        return _clean_child_env(values)

    def _prepare_returned_practice_fixture(
        self,
        *,
        stage: str,
        base_url: str,
        database_url: str,
        database_admin_url: str,
        stack_env: Mapping[str, str],
        fixture_path: Path,
    ) -> dict[str, Any]:
        harness_env = _clean_child_env(
            {
                "PYTHONPATH": str(REPO_ROOT / "apps" / "api"),
                "PYTHONUTF8": "1",
            }
        )
        self._run(
            stage,
            [
                self.config.python_exe,
                "-X",
                "utf8",
                str(FIXTURE_SCRIPT),
                "--api-base-url",
                f"{base_url}/api",
                "--database-url",
                database_url,
                "--database-admin-url",
                database_admin_url,
                "--practice-internal-token",
                stack_env["VIGNETTE_PRACTICE_INTERNAL_TOKEN"],
                "--transfer-internal-token",
                stack_env["VIGNETTE_CALIBRATION_TRANSFER_INTERNAL_TOKEN"],
                "--out",
                str(fixture_path),
                "--request-timeout",
                str(self.config.request_timeout),
                "--review-poll-timeout",
                str(self.config.readiness_timeout),
            ],
            env=harness_env,
            timeout=20 * 60,
        )
        fixture = json.loads(fixture_path.read_text(encoding="utf-8"))
        setup = fixture.get("setup_proof") or {}
        expected_setup = {
            "source_review_ready": True,
            "follow_up_review_ready": True,
            "distinct_persona": True,
            "initial_runtime_observation_count": 0,
            "initial_actual_transfer_execution_count": 0,
        }
        if any(setup.get(key) != value for key, value in expected_setup.items()):
            raise GateError("returned-practice fixture setup proof is incomplete")
        return {
            **expected_setup,
            "fixture_sha256": _sha256_path(fixture_path),
        }

    def _validate_periodic_result(self, result_path: Path) -> dict[str, Any]:
        try:
            payload = json.loads(result_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            raise GateError(f"periodic browser result is unreadable: {exc}") from exc
        if payload.get("schema_version") != "vignette.periodic-learner-real-closed-loop.v1":
            raise GateError("periodic browser result schema mismatch")
        required_true = (
            "same_learner",
            "recommendation_verified",
            "session_created_in_browser",
            "sse_turn_verified",
            "review_ready",
        )
        if any(payload.get(key) is not True for key in required_true):
            raise GateError("periodic browser proof is incomplete")
        for goal in ("g4", "g5"):
            value = payload.get(goal)
            if not isinstance(value, dict) or (
                value.get("initial_count"),
                value.get("first_count"),
                value.get("replay_count"),
                value.get("replay_idempotent"),
            ) != (0, 1, 1, True):
                raise GateError(f"{goal} did not prove 0 -> 1 -> 1 idempotency")
        return {
            "same_learner": True,
            "recommendation_verified": True,
            "session_created_in_browser": True,
            "sse_turn_verified": True,
            "review_ready": True,
            "g4_counts": [0, 1, 1],
            "g5_counts": [0, 1, 1],
            "result_sha256": _sha256_path(result_path),
        }

    def _cleanup(self) -> dict[str, Any]:
        runtime = self.state.runtime
        proof: dict[str, Any] = {
            "compose_down_exit_code": None,
            "engine_exact_pid_stopped": self.state.engine is None,
            "temp_removed": self.state.temp_dir is None,
            "container_remainder": None,
            "volume_remainder": None,
            "network_remainder": None,
            "listener_counts": {},
            "errors": [],
        }
        if runtime is None:
            return proof
        if self.state.stack_attempted and self.state.env_file and self.state.override_file:
            try:
                down = self._run(
                    "compose_down",
                    self._compose_argv("down", "--remove-orphans", "--volumes", "--timeout", "30"),
                    timeout=600,
                    check=False,
                )
                proof["compose_down_exit_code"] = down.returncode
            except BaseException as exc:  # cleanup must continue
                proof["errors"].append(_safe_error(exc))
        if self.state.engine is not None:
            try:
                proof["engine_exact_pid_stopped"] = self.controller.stop_exact(
                    self.state.engine, timeout=15
                )
            except BaseException as exc:
                proof["errors"].append(_safe_error(exc))
        for kind in ("container", "volume", "network"):
            try:
                proof[f"{kind}_remainder"] = len(self._resource_ids(kind))
            except BaseException as exc:
                proof["errors"].append(_safe_error(exc))
        for port in runtime.ports:
            proof["listener_counts"][str(port)] = int(
                self.controller.tcp_listening("127.0.0.1", port)
            )
        if self.state.temp_dir is not None:
            try:
                resolved = self.state.temp_dir.resolve()
                expected_parent = Path(tempfile.gettempdir()).resolve()
                if resolved.parent != expected_parent or not resolved.name.startswith(
                    "vignette-periodic-"
                ):
                    raise GateError("refusing to remove a non-runner temp directory")
                shutil.rmtree(resolved)
                proof["temp_removed"] = not resolved.exists()
            except BaseException as exc:
                proof["errors"].append(_safe_error(exc))
        return proof

    def _capture_failure_diagnostics(
        self, stack_env: Mapping[str, str]
    ) -> dict[str, Any]:
        if not self.state.stack_attempted:
            return {"captured": False, "reason": "stack_not_attempted"}
        diagnostics: dict[str, Any] = {
            "captured": True,
            "secret_value_emitted": False,
            "commands": {},
        }
        commands = (
            (
                "compose_ps",
                self._compose_argv("ps", "--format", "json"),
            ),
            (
                "compose_logs",
                self._compose_argv(
                    "logs",
                    "--no-color",
                    "--timestamps",
                    "--tail",
                    "240",
                    "db",
                    "api",
                ),
            ),
        )
        for stage, argv in commands:
            result = self._run(stage, argv, timeout=90, check=False)
            combined = "\n".join(part for part in (result.stdout, result.stderr) if part)
            sanitized = _sanitize_diagnostic_text(combined, stack_env)
            diagnostics["commands"][stage] = {
                "exit_code": result.returncode,
                "duration_seconds": round(result.duration_seconds, 3),
                "raw_sha256": hashlib.sha256(combined.encode("utf-8")).hexdigest(),
                "sanitized_tail": sanitized,
            }
        return diagnostics

    @staticmethod
    def _cleanup_green(cleanup: Mapping[str, Any]) -> bool:
        return (
            cleanup.get("compose_down_exit_code") in {None, 0}
            and cleanup.get("engine_exact_pid_stopped") is True
            and cleanup.get("temp_removed") is True
            and cleanup.get("container_remainder") in {None, 0}
            and cleanup.get("volume_remainder") in {None, 0}
            and cleanup.get("network_remainder") in {None, 0}
            and all(value == 0 for value in cleanup.get("listener_counts", {}).values())
            and not cleanup.get("errors")
        )

    def preflight(self) -> dict[str, Any]:
        source = self.source_identity(require_clean=False)
        runtime = build_runtime_identity(source)
        validate_runtime_identity(runtime)
        self.state.source = source
        self.state.runtime = runtime
        docker_target = self._pin_local_docker_context()
        if any(self.controller.tcp_listening("127.0.0.1", port) for port in runtime.ports):
            raise GateError("an allocated preflight port already has a listener")
        temp_dir = Path(tempfile.mkdtemp(prefix=f"vignette-periodic-{runtime.run_id}-"))
        self.state.temp_dir = temp_dir
        self.state.env_file = temp_dir / "stack.env"
        self.state.override_file = temp_dir / "compose.periodic.yml"
        try:
            stack_env = build_stack_environment(runtime)
            write_env_file(self.state.env_file, stack_env)
            write_compose_override(self.state.override_file)
            resolved_config_sha = self._validate_resolved_compose_config(stack_env)
        finally:
            shutil.rmtree(temp_dir)
            self.state.temp_dir = None
        return {
            "schema_version": SCHEMA_VERSION,
            "status": "PREFLIGHT_ONLY",
            "source": {"head": source.head, "tree": source.tree},
            "runtime": {
                "project": runtime.project,
                "ports": list(runtime.ports),
                "protected_ports_reused": False,
                "public_or_nas_reference": False,
            },
            "resolved_config": {
                "sha256": resolved_config_sha,
                "ports_loopback_only": True,
                "database_service_only": True,
                "sentinel_labels_complete": True,
            },
            "docker_target": docker_target,
            "temp_removed": not temp_dir.exists(),
            "automation_changed": False,
        }

    def execute(self) -> dict[str, Any]:
        started_at = _now_iso()
        status = "FAILED"
        error = ""
        cleanup: dict[str, Any] = {}
        stack_env: dict[str, str] = {}
        try:
            self.state.source = self.source_identity(require_clean=True)
            self.state.runtime = build_runtime_identity(self.state.source)
            runtime = self.state.runtime
            validate_runtime_identity(runtime)
            self.state.proof["docker_target"] = self._pin_local_docker_context()
            if any(self.controller.tcp_listening("127.0.0.1", port) for port in runtime.ports):
                raise GateError("an allocated runtime port already has a listener")

            temp_dir = Path(tempfile.mkdtemp(prefix=f"vignette-periodic-{runtime.run_id}-"))
            self.state.temp_dir = temp_dir
            self.state.env_file = temp_dir / "stack.env"
            self.state.override_file = temp_dir / "compose.periodic.yml"
            periodic_fixture_path = temp_dir / "periodic-learner-fixture.json"
            returned_fixture_path = temp_dir / "returned-regression-fixture.json"
            periodic_result = temp_dir / "periodic-result.json"
            stack_env = build_stack_environment(runtime)
            write_env_file(self.state.env_file, stack_env)
            write_compose_override(self.state.override_file)

            base_url = f"http://127.0.0.1:{runtime.http_port}"
            database_url = (
                f"postgresql://{stack_env['APP_DB_USER']}:{stack_env['APP_DB_PASSWORD']}"
                f"@127.0.0.1:{runtime.database_port}/{stack_env['POSTGRES_DB']}"
            )
            database_admin_url = (
                f"postgresql://{stack_env['POSTGRES_USER']}:{stack_env['POSTGRES_PASSWORD']}"
                f"@127.0.0.1:{runtime.database_port}/{stack_env['POSTGRES_DB']}"
            )
            assert_loopback_url("browser base URL", base_url, runtime.http_port)
            assert_loopback_url("database URL", database_url, runtime.database_port)
            assert_loopback_url("database admin URL", database_admin_url, runtime.database_port)
            self.state.proof["resolved_config"] = {
                "sha256": self._validate_resolved_compose_config(stack_env),
                "ports_loopback_only": True,
                "database_service_only": True,
                "engine_port_isolated": True,
                "sentinel_labels_complete": True,
            }

            engine_env = _clean_child_env(
                {
                    "ENGINE_GATEWAY_SHARED_SECRET": stack_env[
                        "ENGINE_GATEWAY_SHARED_SECRET"
                    ],
                    "ENGINE_CLI_CWD": str(temp_dir / "engine-runtime"),
                    "PYTHONPATH": str(REPO_ROOT / "apps" / "api"),
                    "PYTHONUTF8": "1",
                }
            )
            engine_argv = [
                self.config.python_exe,
                "-X",
                "utf8",
                "-m",
                "uvicorn",
                "engine_gateway.gateway:app",
                "--host",
                "127.0.0.1",
                "--port",
                str(runtime.engine_port),
            ]
            assert_safe_invocation(engine_argv, engine_env)
            self.state.engine = self.controller.start(
                "engine_start",
                engine_argv,
                cwd=REPO_ROOT / "apps" / "api",
                env=engine_env,
                stdout_path=temp_dir / "engine.stdout.log",
                stderr_path=temp_dir / "engine.stderr.log",
            )
            engine_health = self._wait_json(
                "engine_health",
                f"http://127.0.0.1:{runtime.engine_port}/health",
                lambda payload: payload.get("ok") is True,
                timeout=60,
            )
            engine_ready = self._wait_json(
                "engine_ready",
                f"http://127.0.0.1:{runtime.engine_port}/ready?force=true",
                lambda payload: payload.get("ok") is True,
                headers={
                    "X-Vignette-Engine-Token": stack_env[
                        "ENGINE_GATEWAY_SHARED_SECRET"
                    ]
                },
                timeout=self.config.readiness_timeout,
            )
            self.state.proof["engine"] = {
                "health_ok": engine_health.get("ok") is True,
                "ready_ok": engine_ready.get("ok") is True,
                "provider": engine_ready.get("engine"),
            }

            self.state.stack_attempted = True
            self._run(
                "compose_up",
                self._compose_argv(
                    "up",
                    "-d",
                    "--build",
                    "--wait",
                    "--wait-timeout",
                    str(int(self.config.readiness_timeout)),
                ),
                timeout=1800,
            )
            resources = self._verify_resources_up()
            api_health = self._wait_json(
                "api_health",
                f"{base_url}/api/health",
                lambda payload: payload.get("status") == "ok"
                and payload.get("db") is True
                and payload.get("engine") is True,
                timeout=self.config.readiness_timeout,
            )
            web_text = self.controller.http_text(base_url, timeout=30)
            if '<div id="root"></div>' not in web_text:
                raise GateError("web readiness omitted the React root")
            sentinel_sha = self._install_database_sentinel(stack_env)
            self.state.proof["stack"] = {
                **resources,
                "api_db_ready": api_health.get("db") is True,
                "api_engine_ready": api_health.get("engine") is True,
                "web_ready": True,
                "database_sentinel_sha256": sentinel_sha,
            }

            self.state.proof["fixtures"] = {
                "periodic": self._prepare_returned_practice_fixture(
                    stage="prepare_periodic_learner_fixture",
                    base_url=base_url,
                    database_url=database_url,
                    database_admin_url=database_admin_url,
                    stack_env=stack_env,
                    fixture_path=periodic_fixture_path,
                )
            }

            web_dir = REPO_ROOT / "apps" / "web"
            playwright_cli = web_dir / "node_modules" / "@playwright" / "test" / "cli.js"
            periodic_run = self._run(
                "periodic_same_learner_browser",
                [
                    self.config.node_exe,
                    str(playwright_cli),
                    "test",
                    PERIODIC_SPEC,
                    "--project=chromium-single-run",
                    "--workers=1",
                    f"--retries={PLAYWRIGHT_RETRIES}",
                    "--reporter=line",
                ],
                cwd=web_dir,
                env=self._browser_env(
                    runtime,
                    periodic_fixture_path,
                    periodic_result,
                ),
                timeout=25 * 60,
            )
            if not re.search(r"\b1 passed\b", periodic_run.stdout + periodic_run.stderr):
                raise GateError("periodic browser did not report exactly 1 passed")
            self.state.proof["periodic_browser"] = self._validate_periodic_result(
                periodic_result
            )

            self.state.proof["fixtures"]["returned_regression"] = (
                self._prepare_returned_practice_fixture(
                    stage="prepare_returned_regression_fixture",
                    base_url=base_url,
                    database_url=database_url,
                    database_admin_url=database_admin_url,
                    stack_env=stack_env,
                    fixture_path=returned_fixture_path,
                )
            )

            returned_run = self._run(
                "returned_practice_desktop_mobile",
                [
                    self.config.node_exe,
                    str(playwright_cli),
                    "test",
                    RETURNED_SPEC,
                    "--project=chromium-desktop",
                    "--project=chromium-mobile",
                    "--workers=1",
                    f"--retries={PLAYWRIGHT_RETRIES}",
                    "--reporter=line",
                ],
                cwd=web_dir,
                env=self._browser_env(runtime, returned_fixture_path),
                timeout=25 * 60,
            )
            if not re.search(r"\b4 passed\b", returned_run.stdout + returned_run.stderr):
                raise GateError("returned-practice DB gate did not report exactly 4 passed")
            self.state.proof["returned_practice_browser"] = {
                "desktop_passed": 2,
                "mobile_passed": 2,
                "total_passed": 4,
                "route_mock_count": 0,
                "output_sha256": hashlib.sha256(
                    returned_run.stdout.encode("utf-8")
                ).hexdigest(),
            }

            final_source = self.source_identity(require_clean=True)
            if final_source != self.state.source:
                raise GateError("source HEAD/tree changed during the execution")
            status = "GREEN"
        except BaseException as exc:
            error = _safe_error(exc)
            if self.state.stack_attempted:
                try:
                    self.state.proof["failure_diagnostics"] = (
                        self._capture_failure_diagnostics(stack_env)
                    )
                except BaseException as diagnostic_exc:
                    self.state.proof["failure_diagnostics"] = {
                        "captured": False,
                        "reason": _safe_error(diagnostic_exc),
                    }
        finally:
            cleanup = self._cleanup()
            if not self._cleanup_green(cleanup):
                status = "FAILED"
                if not error:
                    error = "exact cleanup proof is not GREEN"

        runtime = self.state.runtime
        receipt = {
            "schema_version": SCHEMA_VERSION,
            "status": status,
            "started_at": started_at,
            "finished_at": _now_iso(),
            "source": (
                {"head": self.state.source.head, "tree": self.state.source.tree}
                if self.state.source
                else None
            ),
            "runtime": (
                {
                    "run_id": runtime.run_id,
                    "compose_project": runtime.project,
                    "sentinel_sha256": hashlib.sha256(
                        runtime.sentinel.encode("utf-8")
                    ).hexdigest(),
                    "ports": {
                        "http": runtime.http_port,
                        "https": runtime.https_port,
                        "database": runtime.database_port,
                        "engine": runtime.engine_port,
                    },
                    "protected_ports_reused": False,
                    "public_contacted": False,
                    "nas_contacted": False,
                    "active_database_contacted": False,
                }
                if runtime
                else None
            ),
            "proof": self.state.proof,
            "stages": self.state.stages,
            "cleanup": cleanup,
            "automation_changed": False,
            "error": error or None,
        }
        self.config.receipt_path.parent.mkdir(parents=True, exist_ok=True)
        temp_receipt = self.config.receipt_path.with_suffix(
            self.config.receipt_path.suffix + ".tmp"
        )
        temp_receipt.write_text(
            json.dumps(receipt, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
            newline="\n",
        )
        temp_receipt.replace(self.config.receipt_path)
        return receipt


def parse_args(argv: Sequence[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    mode = parser.add_mutually_exclusive_group(required=True)
    mode.add_argument("--preflight", action="store_true")
    mode.add_argument("--execute", action="store_true")
    parser.add_argument("--receipt", default="")
    parser.add_argument("--python", default=sys.executable)
    parser.add_argument("--node", default=shutil.which("node.exe") or "node.exe")
    parser.add_argument("--docker", default=shutil.which("docker.exe") or "docker.exe")
    parser.add_argument("--request-timeout", type=float, default=240.0)
    parser.add_argument("--readiness-timeout", type=float, default=600.0)
    args = parser.parse_args(argv)
    if args.execute and not args.receipt:
        parser.error("--execute requires --receipt")
    if args.request_timeout <= 0 or args.readiness_timeout < 60:
        parser.error("timeouts must be positive and readiness must be at least 60 seconds")
    return args


def main(argv: Sequence[str] | None = None) -> int:
    raw_argv = list(sys.argv[1:] if argv is None else argv)
    try:
        assert_safe_invocation(raw_argv, os.environ)
        assert_no_inherited_runtime_targets(os.environ)
        args = parse_args(raw_argv)
        receipt = (
            Path(args.receipt).resolve()
            if args.receipt
            else Path(tempfile.gettempdir()) / "vignette-periodic-preflight.json"
        )
        config = RunnerConfig(
            receipt_path=receipt,
            python_exe=args.python,
            node_exe=args.node,
            docker_exe=args.docker,
            execute=args.execute,
            request_timeout=args.request_timeout,
            readiness_timeout=args.readiness_timeout,
        )
        runner = PeriodicRunner(config, LocalController())
        result = runner.execute() if args.execute else runner.preflight()
        print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
        return 0 if result["status"] in {"GREEN", "PREFLIGHT_ONLY"} else 1
    except GateError as exc:
        print(
            json.dumps(
                {"schema_version": SCHEMA_VERSION, "status": "BLOCKED", "error": _safe_error(exc)},
                ensure_ascii=False,
                separators=(",", ":"),
            )
        )
        return 2


if __name__ == "__main__":
    raise SystemExit(main())