feat(deploy): add declarative installation workflow

tools/checks/check-focused.sh

@@ -30,6 +30,10 @@ GOVOPLAN_CORE_ROOT="$ROOT" PYTHON="$PYTHON" CHECK_TESTCLIENT_DEPRECATIONS=1 bash
 "$PYTHON" "$META_ROOT/tools/checks/check-contracts.py" --no-impact
 PYTHONDONTWRITEBYTECODE=1 "$PYTHON" "$META_ROOT/tools/checks/check-manifest-shapes.py"
 
+cd "$META_ROOT"
+"$PYTHON" -m unittest tests.test_deployment_installer
+cd "$ROOT"
+
 "$PYTHON" - <<'PY'
 import ast
 import pathlib

tools/deployment/__main__.py

@@ -0,0 +1,6 @@
+"""Executable entry point for the single-file GovOPlaN deployer."""
+
+from govoplan_deploy.cli import main
+
+
+raise SystemExit(main())

tools/deployment/build-deployer-zipapp.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Build the dependency-free GovOPlaN deployer as one executable zipapp."""
+
+from __future__ import annotations
+
+import argparse
+from hashlib import sha256
+from pathlib import Path
+import zipapp
+
+
+ROOT = Path(__file__).resolve().parent
+DEFAULT_OUTPUT = ROOT.parent.parent / "runtime" / "deployment" / "govoplan-deploy.pyz"
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
+    args = parser.parse_args(argv)
+
+    output = args.output.expanduser().resolve()
+    if output == ROOT or ROOT in output.parents:
+        raise SystemExit("output must be outside the deployment source directory")
+    output.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    temporary = output.with_name(f".{output.name}.tmp")
+    if temporary.exists():
+        temporary.unlink()
+    zipapp.create_archive(
+        ROOT,
+        target=temporary,
+        interpreter="/usr/bin/env python3",
+        compressed=True,
+        filter=_include_source,
+    )
+    temporary.chmod(0o755)
+    temporary.replace(output)
+    digest = sha256(output.read_bytes()).hexdigest()
+    print(f"{digest}  {output}")
+    return 0
+
+
+def _include_source(path: Path) -> bool:
+    return (
+        "__pycache__" not in path.parts
+        and path.suffix not in {".pyc", ".pyo"}
+        and path.name != "build-deployer-zipapp.py"
+    )
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/deployment/govoplan-deploy.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+"""GovOPlaN deployment entry point."""
+
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+
+TOOLS_ROOT = Path(__file__).resolve().parent
+if str(TOOLS_ROOT) not in sys.path:
+    sys.path.insert(0, str(TOOLS_ROOT))
+
+from govoplan_deploy.cli import main  # noqa: E402
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/deployment/govoplan_deploy/__init__.py

@@ -0,0 +1,19 @@
+"""Declarative GovOPlaN host deployment tooling."""
+
+from .model import (
+    BASE_MODULES,
+    FULL_MODULES,
+    InstallationSpec,
+    SpecError,
+    default_spec,
+    load_spec,
+)
+
+__all__ = [
+    "BASE_MODULES",
+    "FULL_MODULES",
+    "InstallationSpec",
+    "SpecError",
+    "default_spec",
+    "load_spec",
+]

tools/deployment/govoplan_deploy/bundle.py

@@ -0,0 +1,572 @@
+"""Secret handling and deterministic Compose bundle rendering."""
+
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+from hashlib import sha256
+import hmac
+import json
+import os
+from pathlib import Path
+import secrets
+import stat
+from typing import Mapping
+from urllib.parse import quote, urlsplit
+
+from .model import ENV_NAME_PATTERN, InstallationSpec
+
+
+SPEC_FILENAME = "installation.json"
+ENV_FILENAME = "secrets.env"
+COMPOSE_FILENAME = "compose.json"
+PLAN_FILENAME = "plan.json"
+RECEIPT_FILENAME = "receipt.json"
+LOCK_FILENAME = ".deployment.lock"
+RUNTIME_ENV_KEYS = (
+    "APP_ENV",
+    "GOVOPLAN_INSTALL_PROFILE",
+    "MASTER_KEY_B64",
+    "DATABASE_URL",
+    "GOVOPLAN_DATABASE_URL_PGTOOLS",
+    "ENABLED_MODULES",
+    "CELERY_ENABLED",
+    "CELERY_QUEUES",
+    "REDIS_URL",
+    "CORS_ORIGINS",
+    "GOVOPLAN_TRUSTED_HOSTS",
+    "FORWARDED_ALLOW_IPS",
+    "AUTH_COOKIE_SECURE",
+    "AUTH_COOKIE_SAMESITE",
+    "GOVOPLAN_HTTP_HSTS_SECONDS",
+    "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS",
+    "GOVOPLAN_MIGRATION_TRACK",
+    "DEV_AUTO_MIGRATE_ENABLED",
+    "DEV_BOOTSTRAP_ENABLED",
+    "GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE",
+    "GOVOPLAN_DEPLOYMENT_SPEC_PATH",
+    "FILE_STORAGE_BACKEND",
+    "FILE_STORAGE_LOCAL_ROOT",
+    "FILE_STORAGE_S3_ENDPOINT_URL",
+    "FILE_STORAGE_S3_REGION",
+    "FILE_STORAGE_S3_ACCESS_KEY_ID",
+    "FILE_STORAGE_S3_SECRET_ACCESS_KEY",
+    "FILE_STORAGE_S3_BUCKET",
+)
+
+
+@dataclass(frozen=True, slots=True)
+class BundlePaths:
+    root: Path
+    spec: Path
+    env: Path
+    compose: Path
+    plan: Path
+    receipt: Path
+    lock: Path
+
+
+def bundle_paths(root: Path) -> BundlePaths:
+    expanded = root.expanduser()
+    if expanded.is_symlink():
+        raise ValueError(
+            f"installation root must not be a symbolic link: {expanded}"
+        )
+    resolved = expanded.resolve()
+    return BundlePaths(
+        root=resolved,
+        spec=resolved / SPEC_FILENAME,
+        env=resolved / ENV_FILENAME,
+        compose=resolved / COMPOSE_FILENAME,
+        plan=resolved / PLAN_FILENAME,
+        receipt=resolved / RECEIPT_FILENAME,
+        lock=resolved / LOCK_FILENAME,
+    )
+
+
+def ensure_private_directory(path: Path) -> None:
+    path.mkdir(mode=0o700, parents=True, exist_ok=True)
+    if path.is_symlink() or not path.is_dir():
+        raise ValueError(f"installation root must be a real directory: {path}")
+    current = stat.S_IMODE(path.stat().st_mode)
+    if current & 0o077:
+        path.chmod(0o700)
+
+
+def initial_secrets(
+    spec: InstallationSpec,
+    *,
+    supplied: Mapping[str, str] | None = None,
+) -> dict[str, str]:
+    values = dict(supplied or {})
+    values.setdefault(
+        "MASTER_KEY_B64",
+        base64.urlsafe_b64encode(os.urandom(32)).decode("ascii"),
+    )
+    values.setdefault("POSTGRES_DB", "govoplan")
+    values.setdefault("POSTGRES_USER", "govoplan")
+    values.setdefault("POSTGRES_PASSWORD", secrets.token_urlsafe(36))
+    values.setdefault("REDIS_PASSWORD", secrets.token_urlsafe(36))
+    return reconcile_runtime_environment(spec, values)
+
+
+def reconcile_runtime_environment(
+    spec: InstallationSpec,
+    current: Mapping[str, str],
+) -> dict[str, str]:
+    values = dict(current)
+    postgres = spec.components.postgres
+    if postgres.mode == "managed":
+        database = values.setdefault("POSTGRES_DB", "govoplan")
+        username = values.setdefault("POSTGRES_USER", "govoplan")
+        password = values.setdefault("POSTGRES_PASSWORD", secrets.token_urlsafe(36))
+        encoded_user = quote(username, safe="")
+        encoded_password = quote(password, safe="")
+        encoded_database = quote(database, safe="")
+        values["DATABASE_URL"] = (
+            f"postgresql+psycopg://{encoded_user}:{encoded_password}"
+            f"@postgres:5432/{encoded_database}"
+        )
+        values["GOVOPLAN_DATABASE_URL_PGTOOLS"] = (
+            f"postgresql://{encoded_user}:{encoded_password}"
+            f"@postgres:5432/{encoded_database}"
+        )
+    elif not values.get(postgres.url_env):
+        raise ValueError(
+            f"external PostgreSQL requires {postgres.url_env} in {ENV_FILENAME}"
+        )
+    else:
+        _validate_service_url(
+            values[postgres.url_env],
+            schemes={"postgresql", "postgresql+psycopg"},
+            label="external PostgreSQL URL",
+        )
+
+    redis = spec.components.redis
+    if redis.mode == "managed":
+        password = values.setdefault("REDIS_PASSWORD", secrets.token_urlsafe(36))
+        values["REDIS_URL"] = f"redis://:{quote(password, safe='')}@redis:6379/0"
+    elif redis.mode == "external":
+        if not values.get(redis.url_env):
+            raise ValueError(
+                f"external Redis requires {redis.url_env} in {ENV_FILENAME}"
+            )
+        _validate_service_url(
+            values[redis.url_env],
+            schemes={"redis", "rediss"},
+            label="external Redis URL",
+        )
+    else:
+        values["REDIS_URL"] = ""
+
+    public = urlsplit(spec.public_url)
+    values.update(
+        {
+            "APP_ENV": "production" if spec.profile == "self-hosted" else "staging",
+            "GOVOPLAN_INSTALL_PROFILE": spec.profile,
+            "ENABLED_MODULES": ",".join(spec.enabled_modules),
+            "CELERY_ENABLED": "true" if redis.mode != "disabled" else "false",
+            "CELERY_QUEUES": (
+                "send_email,append_sent,notifications,calendar,dataflow,events,default"
+            ),
+            "CORS_ORIGINS": spec.public_url,
+            "GOVOPLAN_TRUSTED_HOSTS": public.hostname or "",
+            "FORWARDED_ALLOW_IPS": spec.network_subnet,
+            "AUTH_COOKIE_SECURE": "true" if public.scheme == "https" else "false",
+            "AUTH_COOKIE_SAMESITE": "lax",
+            "GOVOPLAN_HTTP_HSTS_SECONDS": (
+                "31536000" if public.scheme == "https" else "0"
+            ),
+            "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false",
+            "GOVOPLAN_MIGRATION_TRACK": "release",
+            "DEV_AUTO_MIGRATE_ENABLED": "false",
+            "DEV_BOOTSTRAP_ENABLED": "false",
+            "GOVOPLAN_DEPLOYMENT_SPEC_PATH": "/etc/govoplan/deployment/installation.json",
+        }
+    )
+    if redis.mode == "disabled":
+        values["GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE"] = "true"
+    else:
+        values["GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE"] = "false"
+
+    storage = spec.components.storage
+    values["FILE_STORAGE_BACKEND"] = storage.mode
+    if storage.mode == "local":
+        values["FILE_STORAGE_LOCAL_ROOT"] = "/var/lib/govoplan/files"
+    else:
+        required = (
+            "FILE_STORAGE_S3_ENDPOINT_URL",
+            "FILE_STORAGE_S3_REGION",
+            "FILE_STORAGE_S3_ACCESS_KEY_ID",
+            "FILE_STORAGE_S3_SECRET_ACCESS_KEY",
+            "FILE_STORAGE_S3_BUCKET",
+        )
+        missing = [name for name in required if not values.get(name)]
+        if missing:
+            raise ValueError(
+                "S3 storage requires values in secrets.env: " + ", ".join(missing)
+            )
+        endpoint = _validate_service_url(
+            values["FILE_STORAGE_S3_ENDPOINT_URL"],
+            schemes={"http", "https"},
+            label="S3 endpoint URL",
+        )
+        if spec.profile == "self-hosted" and endpoint.scheme != "https":
+            raise ValueError("self-hosted S3 endpoint URL must use HTTPS")
+    return dict(sorted(values.items()))
+
+
+def render_compose(spec: InstallationSpec) -> dict[str, object]:
+    runtime_env = {
+        "environment": _environment_references(RUNTIME_ENV_KEYS),
+    }
+    deployment_mount = (
+        f"./{SPEC_FILENAME}:/etc/govoplan/deployment/installation.json:ro"
+    )
+    data_mounts = [deployment_mount]
+    if spec.components.storage.mode == "local":
+        data_mounts.append("files-data:/var/lib/govoplan/files")
+
+    dependency_conditions: dict[str, dict[str, str]] = {}
+    services: dict[str, object] = {}
+    if spec.components.postgres.mode == "managed":
+        services["postgres"] = {
+            "image": spec.components.postgres.image,
+            "restart": "unless-stopped",
+            "environment": _environment_references(
+                ("POSTGRES_DB", "POSTGRES_USER", "POSTGRES_PASSWORD"),
+                required=True,
+            ),
+            "healthcheck": {
+                "test": [
+                    "CMD-SHELL",
+                    'pg_isready -U "$${POSTGRES_USER}" -d "$${POSTGRES_DB}"',
+                ],
+                "interval": "5s",
+                "timeout": "3s",
+                "retries": 30,
+            },
+            "volumes": ["postgres-data:/var/lib/postgresql/data"],
+            "networks": ["internal"],
+        }
+        dependency_conditions["postgres"] = {"condition": "service_healthy"}
+
+    if spec.components.redis.mode == "managed":
+        services["redis"] = {
+            "image": spec.components.redis.image,
+            "restart": "unless-stopped",
+            "environment": _environment_references(
+                ("REDIS_PASSWORD",),
+                required=True,
+            ),
+            "command": [
+                "sh",
+                "-ec",
+                'exec redis-server --appendonly yes --requirepass "$$REDIS_PASSWORD"',
+            ],
+            "healthcheck": {
+                "test": [
+                    "CMD-SHELL",
+                    'redis-cli -a "$${REDIS_PASSWORD}" --no-auth-warning ping',
+                ],
+                "interval": "5s",
+                "timeout": "3s",
+                "retries": 30,
+            },
+            "volumes": ["redis-data:/data"],
+            "networks": ["internal"],
+        }
+        dependency_conditions["redis"] = {"condition": "service_healthy"}
+
+    if spec.components.mail.mode == "test-mail":
+        services["test-mail"] = {
+            "image": spec.components.mail.image,
+            "restart": "unless-stopped",
+            "environment": {
+                "GREENMAIL_OPTS": (
+                    "-Dgreenmail.setup.test.smtp -Dgreenmail.setup.test.imap "
+                    "-Dgreenmail.hostname=0.0.0.0"
+                )
+            },
+            "networks": ["internal"],
+        }
+
+    common_runtime: dict[str, object] = {
+        **runtime_env,
+        "restart": "unless-stopped",
+        "volumes": data_mounts,
+        "networks": ["internal"],
+    }
+    if dependency_conditions:
+        common_runtime["depends_on"] = dependency_conditions
+
+    services["migrate"] = {
+        **runtime_env,
+        "image": spec.release.api_image,
+        "command": ["python", "-m", "govoplan_core.commands.init_db"],
+        "restart": "no",
+        "volumes": data_mounts,
+        "networks": ["internal"],
+        **({"depends_on": dependency_conditions} if dependency_conditions else {}),
+    }
+    services["api"] = {
+        **common_runtime,
+        "image": spec.release.api_image,
+        "command": [
+            "python",
+            "-m",
+            "uvicorn",
+            "govoplan_core.server.app:app",
+            "--host",
+            "0.0.0.0",
+            "--port",
+            "8000",
+            "--proxy-headers",
+        ],
+        "healthcheck": {
+            "test": [
+                "CMD",
+                "python",
+                "-c",
+                (
+                    "import urllib.request;"
+                    "urllib.request.urlopen('http://127.0.0.1:8000/health',timeout=3)"
+                ),
+            ],
+            "interval": "10s",
+            "timeout": "5s",
+            "retries": 30,
+            "start_period": "20s",
+        },
+    }
+    services["web"] = {
+        "image": spec.release.web_image,
+        "restart": "unless-stopped",
+        "environment": {"GOVOPLAN_API_UPSTREAM": "http://api:8000"},
+        "depends_on": {"api": {"condition": "service_healthy"}},
+        "ports": [_published_port(spec.listen.address, spec.listen.port, 8080)],
+        "networks": ["internal"],
+    }
+    if spec.components.redis.mode != "disabled":
+        services["worker"] = {
+            **common_runtime,
+            "image": spec.release.api_image,
+            "command": [
+                "python",
+                "-m",
+                "celery",
+                "-A",
+                "govoplan_core.celery_app:celery",
+                "worker",
+                "--queues",
+                (
+                    "send_email,append_sent,notifications,calendar,"
+                    "dataflow,events,default"
+                ),
+                "--loglevel",
+                "INFO",
+            ],
+        }
+        services["scheduler"] = {
+            **common_runtime,
+            "image": spec.release.api_image,
+            "command": [
+                "python",
+                "-m",
+                "celery",
+                "-A",
+                "govoplan_core.celery_app:celery",
+                "beat",
+                "--loglevel",
+                "INFO",
+            ],
+        }
+    volumes: dict[str, object] = {}
+    if spec.components.postgres.mode == "managed":
+        volumes["postgres-data"] = {}
+    if spec.components.redis.mode == "managed":
+        volumes["redis-data"] = {}
+    if spec.components.storage.mode == "local":
+        volumes["files-data"] = {}
+
+    return {
+        "name": spec.installation_id,
+        "services": services,
+        "volumes": volumes,
+        "networks": {
+            "internal": {
+                "driver": "bridge",
+                "ipam": {"config": [{"subnet": spec.network_subnet}]},
+            }
+        },
+    }
+
+
+def service_names(spec: InstallationSpec) -> tuple[str, ...]:
+    return tuple(render_compose(spec)["services"].keys())
+
+
+def canonical_json(value: object) -> bytes:
+    return (
+        json.dumps(value, indent=2, sort_keys=True, separators=(",", ": "))
+        + "\n"
+    ).encode("utf-8")
+
+
+def digest_json(value: object) -> str:
+    return sha256(canonical_json(value)).hexdigest()
+
+
+def environment_fingerprint(values: Mapping[str, str]) -> str:
+    key = values.get("MASTER_KEY_B64", "").encode("utf-8")
+    if not key:
+        return ""
+    payload = canonical_json(dict(sorted(values.items())))
+    return hmac.new(key, payload, sha256).hexdigest()
+
+
+def read_env(path: Path) -> dict[str, str]:
+    if not path.exists():
+        return {}
+    values: dict[str, str] = {}
+    for line_number, raw_line in enumerate(
+        path.read_text(encoding="utf-8").splitlines(), start=1
+    ):
+        line = raw_line.strip()
+        if not line or line.startswith("#"):
+            continue
+        key, separator, raw_value = line.partition("=")
+        if not separator or not ENV_NAME_PATTERN.fullmatch(key):
+            raise ValueError(f"invalid environment line {line_number} in {path}")
+        if key in values:
+            raise ValueError(
+                f"duplicate environment key {key!r} on line {line_number}"
+            )
+        value = raw_value
+        if value.startswith('"'):
+            try:
+                decoded = json.loads(value)
+            except json.JSONDecodeError as exc:
+                raise ValueError(
+                    f"invalid quoted environment value on line {line_number}"
+                ) from exc
+            if not isinstance(decoded, str):
+                raise ValueError(
+                    f"environment value on line {line_number} must be a string"
+                )
+            value = decoded
+        elif value.startswith("'"):
+            value = _single_quoted_env_value(value, line_number=line_number)
+        values[key] = value
+    return values
+
+
+def write_env(path: Path, values: Mapping[str, str]) -> None:
+    invalid = sorted(key for key in values if not ENV_NAME_PATTERN.fullmatch(key))
+    if invalid:
+        raise ValueError(
+            "invalid environment variable names: " + ", ".join(invalid)
+        )
+    lines = [
+        "# Generated by govoplan-deploy. Keep this file private.",
+        *[f"{key}={_env_value(value)}" for key, value in sorted(values.items())],
+        "",
+    ]
+    atomic_write(path, "\n".join(lines).encode("utf-8"), mode=0o600)
+
+
+def atomic_write(path: Path, payload: bytes, *, mode: int) -> None:
+    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    temporary = path.with_name(
+        f".{path.name}.{os.getpid()}.{secrets.token_hex(8)}.tmp"
+    )
+    descriptor = os.open(
+        temporary,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+        mode,
+    )
+    try:
+        with os.fdopen(descriptor, "wb", closefd=True) as handle:
+            handle.write(payload)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+        path.chmod(mode)
+        directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory_fd)
+        finally:
+            os.close(directory_fd)
+    finally:
+        if temporary.exists():
+            temporary.unlink()
+
+
+def _env_value(value: str) -> str:
+    if "\x00" in value or "\n" in value or "\r" in value:
+        raise ValueError("environment values must not contain NUL or line breaks")
+    if value and all(
+        character.isalnum() or character in "_./:@%+,-"
+        for character in value
+    ):
+        return value
+    escaped = value.replace("\\", "\\\\").replace("'", "\\'")
+    return f"'{escaped}'"
+
+
+def _validate_service_url(
+    value: str,
+    *,
+    schemes: set[str],
+    label: str,
+):
+    parsed = urlsplit(value)
+    try:
+        parsed.port
+    except ValueError as exc:
+        raise ValueError(f"{label} contains an invalid port") from exc
+    if parsed.scheme not in schemes or not parsed.hostname:
+        raise ValueError(
+            f"{label} must use one of {', '.join(sorted(schemes))} and include a host"
+        )
+    if parsed.fragment:
+        raise ValueError(f"{label} must not contain a fragment")
+    return parsed
+
+
+def _single_quoted_env_value(value: str, *, line_number: int) -> str:
+    if len(value) < 2 or not value.endswith("'"):
+        raise ValueError(
+            f"unterminated single-quoted environment value on line {line_number}"
+        )
+    body = value[1:-1]
+    output: list[str] = []
+    index = 0
+    while index < len(body):
+        character = body[index]
+        if character != "\\":
+            output.append(character)
+            index += 1
+            continue
+        index += 1
+        if index >= len(body) or body[index] not in {"\\", "'"}:
+            raise ValueError(
+                f"invalid single-quoted escape on line {line_number}"
+            )
+        output.append(body[index])
+        index += 1
+    return "".join(output)
+
+
+def _published_port(address: str, host_port: int, container_port: int) -> str:
+    host = f"[{address}]" if ":" in address else address
+    return f"{host}:{host_port}:{container_port}"
+
+
+def _environment_references(
+    keys: tuple[str, ...],
+    *,
+    required: bool = False,
+) -> dict[str, str]:
+    suffix = ":?required by GovOPlaN deployment" if required else ":-"
+    return {key: f"${{{key}{suffix}}}" for key in keys}

tools/deployment/govoplan_deploy/cli.py

@@ -0,0 +1,692 @@
+"""Command-line interface for GovOPlaN deployment reconciliation."""
+
+from __future__ import annotations
+
+import argparse
+from contextlib import contextmanager
+from dataclasses import replace
+from datetime import UTC, datetime
+import fcntl
+import getpass
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import time
+from typing import Iterator, Mapping, Sequence
+from urllib.error import URLError
+from urllib.request import urlopen
+
+from .bundle import (
+    atomic_write,
+    bundle_paths,
+    canonical_json,
+    digest_json,
+    environment_fingerprint,
+    ensure_private_directory,
+    initial_secrets,
+    read_env,
+    reconcile_runtime_environment,
+    render_compose,
+    service_names,
+    write_env,
+)
+from .model import (
+    ComponentConfig,
+    InstallationSpec,
+    ListenConfig,
+    SpecError,
+    default_spec,
+    load_spec,
+    parse_spec,
+)
+from .planning import DeploymentPlan, build_plan
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="govoplan-deploy",
+        description=(
+            "Create, validate, and reconcile a declarative GovOPlaN Compose deployment."
+        ),
+    )
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    init = subparsers.add_parser(
+        "init", help="Create a new private installation bundle."
+    )
+    _directory_argument(init)
+    _configuration_arguments(init)
+    init.add_argument(
+        "--non-interactive",
+        action="store_true",
+        help="Use flags/defaults without prompting.",
+    )
+
+    configure = subparsers.add_parser(
+        "configure",
+        help="Change installation choices while preserving generated secrets.",
+    )
+    _directory_argument(configure)
+    _configuration_arguments(configure, defaults=False)
+
+    render = subparsers.add_parser(
+        "render", help="Regenerate Compose and the deployment plan without applying."
+    )
+    _directory_argument(render)
+    render.add_argument("--json", action="store_true", help="Print the plan as JSON.")
+
+    doctor = subparsers.add_parser(
+        "doctor", help="Run specification, secret, host, and provenance checks."
+    )
+    _directory_argument(doctor)
+    doctor.add_argument("--json", action="store_true", help="Print the plan as JSON.")
+
+    apply_parser = subparsers.add_parser(
+        "apply", help="Apply an allowed plan with Docker Compose."
+    )
+    _directory_argument(apply_parser)
+    apply_parser.add_argument(
+        "--allow-unverified-images",
+        action="store_true",
+        help="Allow mutable/unverified images for an evaluation profile only.",
+    )
+    apply_parser.add_argument(
+        "--skip-pull",
+        action="store_true",
+        help="Do not pull images before reconciliation.",
+    )
+    apply_parser.add_argument(
+        "--health-timeout-seconds",
+        type=float,
+        default=120.0,
+        help="Maximum time to wait for the public health endpoint.",
+    )
+
+    status = subparsers.add_parser(
+        "status", help="Show desired state and current Compose process state."
+    )
+    _directory_argument(status)
+    status.add_argument("--json", action="store_true", help="Print JSON.")
+    return parser
+
+
+def _directory_argument(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--directory",
+        type=Path,
+        default=Path.home() / ".local" / "share" / "govoplan" / "installations" / "default",
+        help="Private installation state directory.",
+    )
+
+
+def _configuration_arguments(
+    parser: argparse.ArgumentParser, *, defaults: bool = True
+) -> None:
+    default = (lambda value: value) if defaults else (lambda _value: None)
+    parser.add_argument("--installation-id", default=default("govoplan-local"))
+    parser.add_argument(
+        "--profile",
+        choices=("evaluation", "self-hosted"),
+        default=default("evaluation"),
+    )
+    parser.add_argument("--public-url", default=default("http://127.0.0.1:8080"))
+    parser.add_argument("--listen-address", default=default("127.0.0.1"))
+    parser.add_argument("--listen-port", type=int, default=default(8080))
+    parser.add_argument(
+        "--postgres",
+        choices=("managed", "external"),
+        default=default("managed"),
+    )
+    parser.add_argument("--database-url", help="External PostgreSQL URL; stored privately.")
+    parser.add_argument(
+        "--redis",
+        choices=("managed", "external", "disabled"),
+        default=default("managed"),
+    )
+    parser.add_argument("--redis-url", help="External Redis URL; stored privately.")
+    parser.add_argument(
+        "--mail",
+        choices=("disabled", "external-relay", "test-mail"),
+        default=default("disabled"),
+    )
+    parser.add_argument(
+        "--storage",
+        choices=("local", "s3"),
+        default=default("local"),
+    )
+    parser.add_argument("--s3-endpoint-url", help="S3-compatible endpoint URL.")
+    parser.add_argument("--s3-region", help="S3 region.")
+    parser.add_argument("--s3-access-key-id", help="S3 access key id; stored privately.")
+    parser.add_argument(
+        "--s3-secret-access-key",
+        help="S3 secret access key; stored privately.",
+    )
+    parser.add_argument("--s3-bucket", help="S3 bucket for managed files.")
+    parser.add_argument(
+        "--module-set",
+        choices=("core", "base", "full"),
+        default=default("base"),
+    )
+    parser.add_argument("--api-image", default=default("govoplan-api:unpublished"))
+    parser.add_argument("--web-image", default=default("govoplan-web:unpublished"))
+    parser.add_argument("--release-version", default=default("unpublished"))
+    parser.add_argument("--release-channel", default=default("stable"))
+    parser.add_argument("--manifest-url", default=default(""))
+    parser.add_argument("--manifest-sha256", default=default(""))
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    try:
+        if args.command == "init":
+            return _init(args)
+        if args.command == "configure":
+            return _configure(args)
+        if args.command in {"render", "doctor"}:
+            return _render_or_doctor(args)
+        if args.command == "apply":
+            return _apply(args)
+        if args.command == "status":
+            return _status(args)
+    except (SpecError, ValueError, OSError, subprocess.SubprocessError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+    raise RuntimeError(f"unsupported command: {args.command}")
+
+
+def _init(args: argparse.Namespace) -> int:
+    paths = bundle_paths(args.directory)
+    ensure_private_directory(paths.root)
+    if paths.spec.exists():
+        raise ValueError(
+            f"{paths.spec} already exists; use configure for an existing installation"
+        )
+    if not args.non_interactive and sys.stdin.isatty():
+        _prompt_configuration(args)
+    spec = default_spec(
+        installation_id=args.installation_id,
+        profile=args.profile,
+        public_url=args.public_url,
+        listen_address=args.listen_address,
+        listen_port=args.listen_port,
+        postgres_mode=args.postgres,
+        redis_mode=args.redis,
+        mail_mode=args.mail,
+        storage_mode=args.storage,
+        module_set=args.module_set,
+        api_image=args.api_image,
+        web_image=args.web_image,
+        manifest_url=args.manifest_url,
+        manifest_sha256=args.manifest_sha256,
+        version=args.release_version,
+        channel=args.release_channel,
+    )
+    supplied = _supplied_secret_values(args)
+    secrets = initial_secrets(spec, supplied=supplied)
+    _write_bundle(spec, paths, secrets)
+    plan = build_plan(spec, paths, include_host_checks=False)
+    _write_plan(paths.plan, plan)
+    print(f"Created GovOPlaN installation bundle at {paths.root}")
+    print(f"Edit choices with: govoplan-deploy configure --directory {paths.root}")
+    print(f"Check readiness with: govoplan-deploy doctor --directory {paths.root}")
+    if any(check.level == "error" for check in plan.checks):
+        print("The bundle is not apply-ready yet; run doctor for the blocking checks.")
+    return 0
+
+
+def _configure(args: argparse.Namespace) -> int:
+    paths = bundle_paths(args.directory)
+    ensure_private_directory(paths.root)
+    current = load_spec(paths.spec)
+    if args.installation_id and args.installation_id != current.installation_id:
+        raise ValueError(
+            "installation_id is immutable; create a separate installation "
+            "instead of renaming a Compose project"
+        )
+    spec = _updated_spec(current, args)
+    if (
+        current.components.postgres.mode == "managed"
+        and spec.components.postgres.mode == "external"
+        and not args.database_url
+    ):
+        raise ValueError(
+            "switching PostgreSQL from managed to external requires --database-url"
+        )
+    if (
+        current.components.redis.mode != "external"
+        and spec.components.redis.mode == "external"
+        and not args.redis_url
+    ):
+        raise ValueError(
+            "switching Redis to external requires --redis-url"
+        )
+    secrets = read_env(paths.env)
+    secrets.update(_supplied_secret_values(args))
+    secrets = reconcile_runtime_environment(spec, secrets)
+    _write_bundle(spec, paths, secrets)
+    plan = build_plan(spec, paths, include_host_checks=False)
+    _write_plan(paths.plan, plan)
+    print(f"Updated desired state at {paths.root}")
+    _print_plan(plan)
+    return 0
+
+
+def _render_or_doctor(args: argparse.Namespace) -> int:
+    paths = bundle_paths(args.directory)
+    spec = load_spec(paths.spec)
+    secrets = reconcile_runtime_environment(spec, read_env(paths.env))
+    _write_bundle(spec, paths, secrets)
+    plan = build_plan(
+        spec,
+        paths,
+        include_host_checks=args.command == "doctor",
+    )
+    _write_plan(paths.plan, plan)
+    if args.json:
+        print(json.dumps(plan.to_dict(), indent=2, sort_keys=True))
+    else:
+        _print_plan(plan)
+    return 1 if plan.blocked else 0
+
+
+def _apply(args: argparse.Namespace) -> int:
+    paths = bundle_paths(args.directory)
+    spec = load_spec(paths.spec)
+    if args.allow_unverified_images and spec.profile != "evaluation":
+        raise ValueError(
+            "--allow-unverified-images is restricted to evaluation installations"
+        )
+    ensure_private_directory(paths.root)
+    with _deployment_lock(paths.lock):
+        secrets = reconcile_runtime_environment(spec, read_env(paths.env))
+        _write_bundle(spec, paths, secrets)
+        plan = build_plan(spec, paths, include_host_checks=True)
+        _write_plan(paths.plan, plan)
+        effective_errors = [
+            check
+            for check in plan.checks
+            if check.level == "error"
+            and not (
+                args.allow_unverified_images
+                and check.id.startswith(
+                    (
+                        "release.api_image.",
+                        "release.web_image.",
+                        "components.postgres.image.",
+                        "components.redis.image.",
+                        "components.mail.image.",
+                        "release.manifest",
+                        "modules.image_composition",
+                    )
+                )
+            )
+        ]
+        if effective_errors:
+            _print_plan(plan)
+            raise ValueError("deployment plan is blocked; resolve doctor errors first")
+        docker = shutil.which("docker")
+        if docker is None:
+            raise ValueError("Docker CLI is required for apply")
+        compose = [
+            docker,
+            "compose",
+            "--env-file",
+            str(paths.env),
+            "--project-name",
+            spec.installation_id,
+            "--file",
+            str(paths.compose),
+        ]
+        if not args.skip_pull:
+            _run([*compose, "pull"], cwd=paths.root)
+        dependencies = [
+            name
+            for name in ("postgres", "redis", "test-mail")
+            if name in service_names(spec)
+        ]
+        if dependencies:
+            _run([*compose, "up", "--detach", *dependencies], cwd=paths.root)
+        _run([*compose, "run", "--rm", "migrate"], cwd=paths.root)
+        runtime_services = [
+            name
+            for name in ("api", "web", "worker", "scheduler")
+            if name in service_names(spec)
+        ]
+        _run(
+            [*compose, "up", "--detach", "--remove-orphans", *runtime_services],
+            cwd=paths.root,
+        )
+        _wait_for_health(
+            f"{spec.public_url}/health",
+            timeout_seconds=args.health_timeout_seconds,
+        )
+        receipt = {
+            "schema_version": 1,
+            "installation_id": spec.installation_id,
+            "applied_at": _now(),
+            "spec_sha256": digest_json(spec.to_dict()),
+            "compose_sha256": digest_json(render_compose(spec)),
+            "environment_fingerprint": environment_fingerprint(secrets),
+            "release": {
+                "channel": spec.release.channel,
+                "version": spec.release.version,
+                "manifest_sha256": spec.release.manifest_sha256,
+                "api_image": spec.release.api_image,
+                "web_image": spec.release.web_image,
+            },
+            "services": list(service_names(spec)),
+            "listen": {
+                "address": spec.listen.address,
+                "port": spec.listen.port,
+            },
+            "management": {
+                "mode": "govoplan-deploy",
+                "agent": "cli",
+                "web_updates": False,
+            },
+        }
+        atomic_write(paths.receipt, canonical_json(receipt), mode=0o600)
+        print(f"GovOPlaN is ready at {spec.public_url}")
+        return 0
+
+
+def _status(args: argparse.Namespace) -> int:
+    paths = bundle_paths(args.directory)
+    spec = load_spec(paths.spec)
+    plan = build_plan(spec, paths, include_host_checks=False)
+    docker = shutil.which("docker")
+    processes: object = []
+    process_error = ""
+    if docker:
+        result = subprocess.run(
+            [
+                docker,
+                "compose",
+                "--env-file",
+                str(paths.env),
+                "--project-name",
+                spec.installation_id,
+                "--file",
+                str(paths.compose),
+                "ps",
+                "--format",
+                "json",
+            ],
+            cwd=paths.root,
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=15,
+        )
+        if result.returncode == 0:
+            processes = _parse_compose_ps(result.stdout)
+        else:
+            process_error = result.stderr.strip() or result.stdout.strip()
+    else:
+        process_error = "Docker CLI is unavailable."
+    payload = {
+        "installation_id": spec.installation_id,
+        "public_url": spec.public_url,
+        "plan": plan.to_dict(),
+        "processes": processes,
+        "process_error": process_error,
+    }
+    if args.json:
+        print(json.dumps(payload, indent=2, sort_keys=True))
+    else:
+        _print_plan(plan)
+        if process_error:
+            print(f"Runtime: {process_error}")
+        elif isinstance(processes, list):
+            print(f"Runtime: {len(processes)} Compose process(es) reported.")
+    return 0
+
+
+def _updated_spec(
+    current: InstallationSpec, args: argparse.Namespace
+) -> InstallationSpec:
+    module_set = getattr(args, "module_set", None)
+    if module_set:
+        modules = default_spec(module_set=module_set).enabled_modules
+    else:
+        modules = current.enabled_modules
+    release = replace(
+        current.release,
+        channel=args.release_channel or current.release.channel,
+        version=args.release_version or current.release.version,
+        manifest_url=(
+            args.manifest_url
+            if args.manifest_url is not None
+            else current.release.manifest_url
+        ),
+        manifest_sha256=(
+            args.manifest_sha256
+            if args.manifest_sha256 is not None
+            else current.release.manifest_sha256
+        ),
+        api_image=args.api_image or current.release.api_image,
+        web_image=args.web_image or current.release.web_image,
+    )
+    components = ComponentConfig(
+        postgres=replace(
+            current.components.postgres,
+            mode=args.postgres or current.components.postgres.mode,
+        ),
+        redis=replace(
+            current.components.redis,
+            mode=args.redis or current.components.redis.mode,
+        ),
+        mail=replace(
+            current.components.mail,
+            mode=args.mail or current.components.mail.mode,
+        ),
+        storage=replace(
+            current.components.storage,
+            mode=args.storage or current.components.storage.mode,
+        ),
+    )
+    value = replace(
+        current,
+        installation_id=args.installation_id or current.installation_id,
+        profile=args.profile or current.profile,
+        public_url=args.public_url or current.public_url,
+        listen=ListenConfig(
+            address=args.listen_address or current.listen.address,
+            port=args.listen_port or current.listen.port,
+        ),
+        release=release,
+        components=components,
+        enabled_modules=modules,
+    )
+    return parse_spec(value.to_dict())
+
+
+def _supplied_secret_values(args: argparse.Namespace) -> dict[str, str]:
+    values: dict[str, str] = {}
+    if getattr(args, "database_url", None):
+        values["DATABASE_URL"] = args.database_url
+        values["GOVOPLAN_DATABASE_URL_PGTOOLS"] = _pgtools_url(args.database_url)
+    if getattr(args, "redis_url", None):
+        values["REDIS_URL"] = args.redis_url
+    s3_values = {
+        "FILE_STORAGE_S3_ENDPOINT_URL": getattr(args, "s3_endpoint_url", None),
+        "FILE_STORAGE_S3_REGION": getattr(args, "s3_region", None),
+        "FILE_STORAGE_S3_ACCESS_KEY_ID": getattr(
+            args, "s3_access_key_id", None
+        ),
+        "FILE_STORAGE_S3_SECRET_ACCESS_KEY": getattr(
+            args, "s3_secret_access_key", None
+        ),
+        "FILE_STORAGE_S3_BUCKET": getattr(args, "s3_bucket", None),
+    }
+    values.update({key: value for key, value in s3_values.items() if value})
+    return values
+
+
+def _pgtools_url(database_url: str) -> str:
+    if database_url.startswith("postgresql+psycopg://"):
+        return "postgresql://" + database_url.removeprefix(
+            "postgresql+psycopg://"
+        )
+    return database_url
+
+
+def _write_bundle(
+    spec: InstallationSpec,
+    paths,
+    secrets: Mapping[str, str],
+) -> None:
+    ensure_private_directory(paths.root)
+    atomic_write(paths.spec, canonical_json(spec.to_dict()), mode=0o600)
+    write_env(paths.env, secrets)
+    atomic_write(paths.compose, canonical_json(render_compose(spec)), mode=0o600)
+
+
+def _write_plan(path: Path, plan: DeploymentPlan) -> None:
+    atomic_write(path, canonical_json(plan.to_dict()), mode=0o600)
+
+
+def _print_plan(plan: DeploymentPlan) -> None:
+    state = "BLOCKED" if plan.blocked else "READY"
+    print(f"GovOPlaN deployment plan: {state}")
+    for action in plan.actions:
+        print(f"  {action.action:12} {action.target}: {action.detail}")
+    for check in plan.checks:
+        prefix = {"ok": "OK", "warning": "WARN", "error": "ERROR"}.get(
+            check.level, check.level.upper()
+        )
+        print(f"  [{prefix}] {check.message}")
+        if check.action and check.level != "ok":
+            print(f"          {check.action}")
+
+
+def _prompt_configuration(args: argparse.Namespace) -> None:
+    args.profile = _prompt_choice(
+        "Installation profile", args.profile, ("evaluation", "self-hosted")
+    )
+    if args.profile == "self-hosted" and args.public_url.startswith("http://"):
+        args.public_url = "https://govoplan.example.org"
+    args.public_url = _prompt("Public URL", args.public_url)
+    args.postgres = _prompt_choice(
+        "PostgreSQL", args.postgres, ("managed", "external")
+    )
+    if args.postgres == "external" and not args.database_url:
+        args.database_url = getpass.getpass(
+            "External PostgreSQL URL (input hidden): "
+        ).strip()
+    redis_choices = (
+        ("managed", "external")
+        if args.profile == "self-hosted"
+        else ("managed", "external", "disabled")
+    )
+    args.redis = _prompt_choice(
+        "Redis", args.redis, redis_choices
+    )
+    if args.redis == "external" and not args.redis_url:
+        args.redis_url = getpass.getpass(
+            "External Redis URL (input hidden): "
+        ).strip()
+    mail_choices = (
+        ("disabled", "external-relay")
+        if args.profile == "self-hosted"
+        else ("disabled", "external-relay", "test-mail")
+    )
+    args.mail = _prompt_choice(
+        "Mail integration",
+        args.mail,
+        mail_choices,
+    )
+    args.storage = _prompt_choice("File storage", args.storage, ("local", "s3"))
+    if args.storage == "s3":
+        args.s3_endpoint_url = args.s3_endpoint_url or _prompt(
+            "S3 endpoint URL", "https://s3.example.org"
+        )
+        args.s3_region = args.s3_region or _prompt("S3 region", "eu-central-1")
+        args.s3_access_key_id = args.s3_access_key_id or _prompt(
+            "S3 access key id", ""
+        )
+        args.s3_secret_access_key = (
+            args.s3_secret_access_key
+            or getpass.getpass("S3 secret access key (input hidden): ").strip()
+        )
+        args.s3_bucket = args.s3_bucket or _prompt("S3 bucket", "govoplan-files")
+    args.module_set = _prompt_choice(
+        "Initial module set", args.module_set, ("core", "base", "full")
+    )
+
+
+def _prompt(label: str, default: str) -> str:
+    value = input(f"{label} [{default}]: ").strip()
+    return value or default
+
+
+def _prompt_choice(label: str, default: str, choices: tuple[str, ...]) -> str:
+    while True:
+        value = _prompt(f"{label} ({'/'.join(choices)})", default)
+        if value in choices:
+            return value
+        print(f"Choose one of: {', '.join(choices)}")
+
+
+@contextmanager
+def _deployment_lock(path: Path) -> Iterator[None]:
+    descriptor = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+    try:
+        try:
+            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+            raise ValueError("another deployment operation owns the installation lock") from exc
+        os.ftruncate(descriptor, 0)
+        os.write(descriptor, f"{os.getpid()}\n".encode("ascii"))
+        os.fsync(descriptor)
+        yield
+    finally:
+        fcntl.flock(descriptor, fcntl.LOCK_UN)
+        os.close(descriptor)
+
+
+def _run(argv: Sequence[str], *, cwd: Path) -> None:
+    result = subprocess.run(list(argv), cwd=cwd, check=False)
+    if result.returncode != 0:
+        raise subprocess.CalledProcessError(result.returncode, list(argv))
+
+
+def _wait_for_health(url: str, *, timeout_seconds: float) -> None:
+    deadline = time.monotonic() + max(timeout_seconds, 1.0)
+    last_error = "health endpoint did not answer"
+    while time.monotonic() < deadline:
+        try:
+            with urlopen(url, timeout=3) as response:  # noqa: S310 - URL originates in the validated installation specification.
+                if 200 <= response.status < 300:
+                    return
+                last_error = f"health endpoint returned HTTP {response.status}"
+        except (OSError, URLError) as exc:
+            last_error = str(exc)
+        time.sleep(2)
+    raise ValueError(f"GovOPlaN did not become healthy: {last_error}")
+
+
+def _parse_compose_ps(output: str) -> list[object]:
+    stripped = output.strip()
+    if not stripped:
+        return []
+    try:
+        value = json.loads(stripped)
+    except json.JSONDecodeError:
+        rows = []
+        for line in stripped.splitlines():
+            try:
+                rows.append(json.loads(line))
+            except json.JSONDecodeError:
+                return [{"raw": stripped}]
+        return rows
+    return value if isinstance(value, list) else [value]
+
+
+def _now() -> str:
+    return datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace(
+        "+00:00", "Z"
+    )

tools/deployment/govoplan_deploy/model.py

@@ -0,0 +1,482 @@
+"""Installation intent model and validation."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import ipaddress
+import json
+from pathlib import Path
+import re
+from typing import Any, Mapping
+from urllib.parse import urlsplit
+
+
+SCHEMA_VERSION = 1
+INSTALLATION_ID_PATTERN = re.compile(r"^[a-z][a-z0-9-]{1,47}$")
+ENV_NAME_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$")
+SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
+IMAGE_DIGEST_PATTERN = re.compile(r"^[^@\s]+@sha256:[0-9a-f]{64}$")
+RFC1918_NETWORKS = tuple(
+    ipaddress.ip_network(value)
+    for value in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")
+)
+
+BASE_MODULES = (
+    "tenancy",
+    "organizations",
+    "identity",
+    "idm",
+    "access",
+    "admin",
+    "dashboard",
+    "policy",
+    "audit",
+    "docs",
+    "ops",
+)
+
+FULL_MODULES = (
+    *BASE_MODULES,
+    "addresses",
+    "dist_lists",
+    "files",
+    "mail",
+    "campaigns",
+    "calendar",
+    "poll",
+    "scheduling",
+    "connectors",
+    "datasources",
+    "dataflow",
+    "workflow",
+    "views",
+    "search",
+    "risk_compliance",
+    "postbox",
+    "evaluation",
+)
+
+
+class SpecError(ValueError):
+    """Raised when an installation specification is malformed."""
+
+
+@dataclass(frozen=True, slots=True)
+class ListenConfig:
+    address: str
+    port: int
+
+
+@dataclass(frozen=True, slots=True)
+class ReleaseConfig:
+    channel: str
+    version: str
+    manifest_url: str
+    manifest_sha256: str
+    api_image: str
+    web_image: str
+
+
+@dataclass(frozen=True, slots=True)
+class ServiceConfig:
+    mode: str
+    image: str = ""
+    url_env: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class StorageConfig:
+    mode: str
+
+
+@dataclass(frozen=True, slots=True)
+class ComponentConfig:
+    postgres: ServiceConfig
+    redis: ServiceConfig
+    mail: ServiceConfig
+    storage: StorageConfig
+
+
+@dataclass(frozen=True, slots=True)
+class InstallationSpec:
+    schema_version: int
+    installation_id: str
+    profile: str
+    public_url: str
+    listen: ListenConfig
+    network_subnet: str
+    release: ReleaseConfig
+    components: ComponentConfig
+    enabled_modules: tuple[str, ...]
+
+    def to_dict(self) -> dict[str, Any]:
+        value = asdict(self)
+        value["enabled_modules"] = list(self.enabled_modules)
+        return value
+
+
+def default_spec(
+    *,
+    installation_id: str = "govoplan-local",
+    profile: str = "evaluation",
+    public_url: str = "http://127.0.0.1:8080",
+    listen_address: str = "127.0.0.1",
+    listen_port: int = 8080,
+    postgres_mode: str = "managed",
+    redis_mode: str = "managed",
+    mail_mode: str = "disabled",
+    storage_mode: str = "local",
+    module_set: str = "base",
+    api_image: str = "govoplan-api:unpublished",
+    web_image: str = "govoplan-web:unpublished",
+    manifest_url: str = "",
+    manifest_sha256: str = "",
+    version: str = "unpublished",
+    channel: str = "stable",
+) -> InstallationSpec:
+    modules = {
+        "core": (),
+        "base": BASE_MODULES,
+        "full": FULL_MODULES,
+    }.get(module_set)
+    if modules is None:
+        raise SpecError("module_set must be core, base, or full")
+    subnet_octet = 32 + (sum(installation_id.encode("utf-8")) % 192)
+    raw = {
+        "schema_version": SCHEMA_VERSION,
+        "installation_id": installation_id,
+        "profile": profile,
+        "public_url": public_url,
+        "listen": {"address": listen_address, "port": listen_port},
+        "network_subnet": f"172.30.{subnet_octet}.0/24",
+        "release": {
+            "channel": channel,
+            "version": version,
+            "manifest_url": manifest_url,
+            "manifest_sha256": manifest_sha256,
+            "api_image": api_image,
+            "web_image": web_image,
+        },
+        "components": {
+            "postgres": {
+                "mode": postgres_mode,
+                "image": "postgres:16-alpine",
+                "url_env": "DATABASE_URL",
+            },
+            "redis": {
+                "mode": redis_mode,
+                "image": "redis:7-alpine",
+                "url_env": "REDIS_URL",
+            },
+            "mail": {
+                "mode": mail_mode,
+                "image": "greenmail/standalone:2.1.9",
+                "url_env": "",
+            },
+            "storage": {"mode": storage_mode},
+        },
+        "enabled_modules": list(modules),
+    }
+    return parse_spec(raw)
+
+
+def load_spec(path: Path) -> InstallationSpec:
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise SpecError(f"installation specification does not exist: {path}") from exc
+    except json.JSONDecodeError as exc:
+        raise SpecError(f"installation specification is not valid JSON: {exc}") from exc
+    return parse_spec(raw)
+
+
+def parse_spec(raw: object) -> InstallationSpec:
+    root = _mapping(raw, "installation")
+    _only_keys(
+        root,
+        {
+            "schema_version",
+            "installation_id",
+            "profile",
+            "public_url",
+            "listen",
+            "network_subnet",
+            "release",
+            "components",
+            "enabled_modules",
+        },
+        "installation",
+    )
+    schema_version = _integer(root, "schema_version")
+    if schema_version != SCHEMA_VERSION:
+        raise SpecError(
+            f"schema_version must be {SCHEMA_VERSION}; found {schema_version}"
+        )
+
+    installation_id = _string(root, "installation_id")
+    if not INSTALLATION_ID_PATTERN.fullmatch(installation_id):
+        raise SpecError(
+            "installation_id must start with a lowercase letter and contain only "
+            "lowercase letters, digits, or hyphens (2-48 characters)"
+        )
+
+    profile = _choice(root, "profile", {"evaluation", "self-hosted"})
+    public_url = _http_url(_string(root, "public_url"), "public_url")
+    public_parts = urlsplit(public_url)
+    if profile == "self-hosted" and public_parts.scheme != "https":
+        raise SpecError("self-hosted public_url must use HTTPS")
+
+    listen_raw = _mapping(root.get("listen"), "listen")
+    _only_keys(listen_raw, {"address", "port"}, "listen")
+    listen = ListenConfig(
+        address=_ip_address(_string(listen_raw, "address"), "listen.address"),
+        port=_port(_integer(listen_raw, "port"), "listen.port"),
+    )
+
+    network_subnet = _network(
+        _string(root, "network_subnet"), "network_subnet"
+    )
+    release = _release(root.get("release"))
+    components = _components(root.get("components"), profile=profile)
+
+    enabled_raw = root.get("enabled_modules")
+    if not isinstance(enabled_raw, list):
+        raise SpecError("enabled_modules must be an array")
+    enabled_modules: list[str] = []
+    seen_modules: set[str] = set()
+    for index, value in enumerate(enabled_raw):
+        if not isinstance(value, str) or not re.fullmatch(
+            r"[a-z][a-z0-9_]{1,63}", value
+        ):
+            raise SpecError(
+                f"enabled_modules[{index}] must be a canonical module id"
+            )
+        if value in seen_modules:
+            raise SpecError(f"enabled_modules contains duplicate module {value!r}")
+        enabled_modules.append(value)
+        seen_modules.add(value)
+
+    return InstallationSpec(
+        schema_version=schema_version,
+        installation_id=installation_id,
+        profile=profile,
+        public_url=public_url,
+        listen=listen,
+        network_subnet=network_subnet,
+        release=release,
+        components=components,
+        enabled_modules=tuple(enabled_modules),
+    )
+
+
+def _release(raw: object) -> ReleaseConfig:
+    value = _mapping(raw, "release")
+    _only_keys(
+        value,
+        {
+            "channel",
+            "version",
+            "manifest_url",
+            "manifest_sha256",
+            "api_image",
+            "web_image",
+        },
+        "release",
+    )
+    channel = _string(value, "channel")
+    if not re.fullmatch(r"[a-z][a-z0-9-]{1,31}", channel):
+        raise SpecError("release.channel must be a canonical channel name")
+    version = _string(value, "version")
+    if not version or len(version) > 80 or any(char.isspace() for char in version):
+        raise SpecError("release.version must be a non-empty version token")
+    manifest_url = _optional_string(value, "manifest_url")
+    if manifest_url:
+        manifest_url = _http_url(manifest_url, "release.manifest_url")
+        if urlsplit(manifest_url).scheme != "https":
+            raise SpecError("release.manifest_url must use HTTPS")
+    manifest_sha256 = _optional_string(value, "manifest_sha256").lower()
+    if manifest_sha256 and not SHA256_PATTERN.fullmatch(manifest_sha256):
+        raise SpecError("release.manifest_sha256 must be a lowercase SHA-256 hex digest")
+    api_image = _image(_string(value, "api_image"), "release.api_image")
+    web_image = _image(_string(value, "web_image"), "release.web_image")
+    return ReleaseConfig(
+        channel=channel,
+        version=version,
+        manifest_url=manifest_url,
+        manifest_sha256=manifest_sha256,
+        api_image=api_image,
+        web_image=web_image,
+    )
+
+
+def _components(raw: object, *, profile: str) -> ComponentConfig:
+    value = _mapping(raw, "components")
+    _only_keys(value, {"postgres", "redis", "mail", "storage"}, "components")
+    postgres = _service(
+        value.get("postgres"),
+        "components.postgres",
+        modes={"managed", "external"},
+        image_required_for={"managed"},
+        url_env_required_for={"external"},
+    )
+    redis = _service(
+        value.get("redis"),
+        "components.redis",
+        modes={"managed", "external", "disabled"},
+        image_required_for={"managed"},
+        url_env_required_for={"external"},
+    )
+    mail = _service(
+        value.get("mail"),
+        "components.mail",
+        modes={"disabled", "external-relay", "test-mail"},
+        image_required_for={"test-mail"},
+        url_env_required_for=set(),
+    )
+    storage_raw = _mapping(value.get("storage"), "components.storage")
+    _only_keys(storage_raw, {"mode"}, "components.storage")
+    storage = StorageConfig(mode=_choice(storage_raw, "mode", {"local", "s3"}))
+    if postgres.url_env != "DATABASE_URL":
+        raise SpecError("components.postgres.url_env must be DATABASE_URL")
+    if redis.url_env != "REDIS_URL":
+        raise SpecError("components.redis.url_env must be REDIS_URL")
+    if mail.url_env:
+        raise SpecError("components.mail.url_env must be empty")
+    if profile == "self-hosted" and redis.mode == "disabled":
+        raise SpecError("self-hosted installations require managed or external Redis")
+    if profile == "self-hosted" and mail.mode == "test-mail":
+        raise SpecError("the test-mail component is restricted to evaluation installs")
+    return ComponentConfig(
+        postgres=postgres,
+        redis=redis,
+        mail=mail,
+        storage=storage,
+    )
+
+
+def _service(
+    raw: object,
+    label: str,
+    *,
+    modes: set[str],
+    image_required_for: set[str],
+    url_env_required_for: set[str],
+) -> ServiceConfig:
+    value = _mapping(raw, label)
+    _only_keys(value, {"mode", "image", "url_env"}, label)
+    mode = _choice(value, "mode", modes)
+    image = _optional_string(value, "image")
+    url_env = _optional_string(value, "url_env")
+    if mode in image_required_for:
+        image = _image(image, f"{label}.image")
+    if mode in url_env_required_for:
+        if not ENV_NAME_PATTERN.fullmatch(url_env):
+            raise SpecError(f"{label}.url_env must name an environment variable")
+    return ServiceConfig(mode=mode, image=image, url_env=url_env)
+
+
+def image_is_digest_pinned(value: str) -> bool:
+    return bool(IMAGE_DIGEST_PATTERN.fullmatch(value))
+
+
+def image_is_unpublished(value: str) -> bool:
+    return value.endswith(":unpublished")
+
+
+def _mapping(value: object, label: str) -> Mapping[str, Any]:
+    if not isinstance(value, dict):
+        raise SpecError(f"{label} must be an object")
+    if not all(isinstance(key, str) for key in value):
+        raise SpecError(f"{label} keys must be strings")
+    return value
+
+
+def _only_keys(value: Mapping[str, Any], allowed: set[str], label: str) -> None:
+    unknown = sorted(set(value) - allowed)
+    if unknown:
+        raise SpecError(f"{label} contains unknown fields: {', '.join(unknown)}")
+
+
+def _string(value: Mapping[str, Any], key: str) -> str:
+    result = value.get(key)
+    if not isinstance(result, str):
+        raise SpecError(f"{key} must be a string")
+    return result.strip()
+
+
+def _optional_string(value: Mapping[str, Any], key: str) -> str:
+    result = value.get(key, "")
+    if not isinstance(result, str):
+        raise SpecError(f"{key} must be a string")
+    return result.strip()
+
+
+def _integer(value: Mapping[str, Any], key: str) -> int:
+    result = value.get(key)
+    if isinstance(result, bool) or not isinstance(result, int):
+        raise SpecError(f"{key} must be an integer")
+    return result
+
+
+def _choice(value: Mapping[str, Any], key: str, choices: set[str]) -> str:
+    result = _string(value, key)
+    if result not in choices:
+        raise SpecError(f"{key} must be one of: {', '.join(sorted(choices))}")
+    return result
+
+
+def _http_url(value: str, label: str) -> str:
+    try:
+        parsed = urlsplit(value)
+        parsed.port
+    except ValueError as exc:
+        raise SpecError(f"{label} contains an invalid host or port") from exc
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise SpecError(f"{label} must be an absolute HTTP(S) URL")
+    if parsed.username or parsed.password or parsed.fragment or parsed.query:
+        raise SpecError(
+            f"{label} must not contain credentials, a query, or a fragment"
+        )
+    if label == "public_url" and parsed.path not in {"", "/"}:
+        raise SpecError("public_url must address the site root without a path")
+    return value.rstrip("/")
+
+
+def _image(value: str, label: str) -> str:
+    if (
+        not value
+        or len(value) > 300
+        or any(character.isspace() for character in value)
+        or value.startswith("-")
+    ):
+        raise SpecError(f"{label} must be a valid non-empty OCI image reference")
+    return value
+
+
+def _ip_address(value: str, label: str) -> str:
+    try:
+        return str(ipaddress.ip_address(value))
+    except ValueError as exc:
+        raise SpecError(f"{label} must be an IPv4 or IPv6 address") from exc
+
+
+def _network(value: str, label: str) -> str:
+    try:
+        network = ipaddress.ip_network(value, strict=True)
+    except ValueError as exc:
+        raise SpecError(f"{label} must be a canonical private IP network") from exc
+    if (
+        network.version != 4
+        or not any(network.subnet_of(parent) for parent in RFC1918_NETWORKS)
+        or not 24 <= network.prefixlen <= 28
+    ):
+        raise SpecError(
+            f"{label} must be an RFC1918 IPv4 /24 to /28 network"
+        )
+    return str(network)
+
+
+def _port(value: int, label: str) -> int:
+    if value < 1 or value > 65535:
+        raise SpecError(f"{label} must be between 1 and 65535")
+    return value

tools/deployment/govoplan_deploy/planning.py

@@ -0,0 +1,579 @@
+"""Deployment plan and host preflight checks."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import json
+import os
+from pathlib import Path
+import platform
+import shutil
+import socket
+import stat
+import subprocess
+from typing import Callable, Mapping, Sequence
+from urllib.parse import urlsplit
+
+from .bundle import (
+    BundlePaths,
+    digest_json,
+    environment_fingerprint,
+    read_env,
+    render_compose,
+    service_names,
+)
+from .model import (
+    InstallationSpec,
+    image_is_digest_pinned,
+    image_is_unpublished,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class Check:
+    id: str
+    level: str
+    message: str
+    action: str = ""
+
+    def to_dict(self) -> dict[str, str]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class PlanAction:
+    action: str
+    target: str
+    detail: str
+
+    def to_dict(self) -> dict[str, str]:
+        return asdict(self)
+
+
+@dataclass(frozen=True, slots=True)
+class DeploymentPlan:
+    installation_id: str
+    desired_spec_sha256: str
+    desired_compose_sha256: str
+    desired_environment_fingerprint: str
+    actions: tuple[PlanAction, ...]
+    checks: tuple[Check, ...]
+
+    @property
+    def blocked(self) -> bool:
+        return any(check.level == "error" for check in self.checks)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "installation_id": self.installation_id,
+            "desired_spec_sha256": self.desired_spec_sha256,
+            "desired_compose_sha256": self.desired_compose_sha256,
+            "desired_environment_fingerprint": (
+                self.desired_environment_fingerprint
+            ),
+            "blocked": self.blocked,
+            "actions": [action.to_dict() for action in self.actions],
+            "checks": [check.to_dict() for check in self.checks],
+        }
+
+
+CommandRunner = Callable[[Sequence[str], Path], subprocess.CompletedProcess[str]]
+
+
+def build_plan(
+    spec: InstallationSpec,
+    paths: BundlePaths,
+    *,
+    include_host_checks: bool = True,
+    command_runner: CommandRunner | None = None,
+) -> DeploymentPlan:
+    compose = render_compose(spec)
+    desired_services = set(service_names(spec))
+    previous = _read_receipt(paths.receipt)
+    previous_services = {
+        str(item) for item in previous.get("services", []) if isinstance(item, str)
+    }
+    previous_spec_digest = previous.get("spec_sha256")
+    previous_compose_digest = previous.get("compose_sha256")
+    previous_environment_fingerprint = previous.get("environment_fingerprint")
+    spec_digest = digest_json(spec.to_dict())
+    compose_digest = digest_json(compose)
+    environment_digest = environment_fingerprint(read_env(paths.env))
+
+    actions: list[PlanAction] = []
+    if not previous:
+        actions.append(
+            PlanAction(
+                "create",
+                "installation",
+                "Create the first deployment revision.",
+            )
+        )
+    if previous_spec_digest != spec_digest:
+        actions.append(
+            PlanAction(
+                "reconfigure",
+                "installation",
+                "Reconcile runtime configuration with installation.json.",
+            )
+        )
+    if previous_compose_digest != compose_digest:
+        actions.append(
+            PlanAction(
+                "render",
+                "compose",
+                "Render a new deterministic Compose definition.",
+            )
+        )
+    if previous_environment_fingerprint != environment_digest:
+        actions.append(
+            PlanAction(
+                "reconfigure",
+                "environment",
+                "Reconcile changed secret bindings or runtime environment values.",
+            )
+        )
+    for name in sorted(desired_services - previous_services):
+        actions.append(PlanAction("start", name, "Start the selected service."))
+    for name in sorted(previous_services - desired_services):
+        actions.append(
+            PlanAction(
+                "remove",
+                name,
+                "Remove the service container; retained volumes are not deleted.",
+            )
+        )
+    if (
+        previous
+        and previous_spec_digest == spec_digest
+        and previous_compose_digest == compose_digest
+        and previous_environment_fingerprint == environment_digest
+        and previous_services == desired_services
+    ):
+        actions.append(
+            PlanAction("noop", "installation", "Desired state matches the last receipt.")
+        )
+
+    checks = list(static_checks(spec, paths))
+    if include_host_checks:
+        checks.extend(host_checks(spec, paths, command_runner=command_runner))
+    return DeploymentPlan(
+        installation_id=spec.installation_id,
+        desired_spec_sha256=spec_digest,
+        desired_compose_sha256=compose_digest,
+        desired_environment_fingerprint=environment_digest,
+        actions=tuple(actions),
+        checks=tuple(checks),
+    )
+
+
+def static_checks(spec: InstallationSpec, paths: BundlePaths) -> tuple[Check, ...]:
+    checks: list[Check] = []
+    images = {
+        "release.api_image": spec.release.api_image,
+        "release.web_image": spec.release.web_image,
+    }
+    if spec.components.postgres.mode == "managed":
+        images["components.postgres.image"] = spec.components.postgres.image
+    if spec.components.redis.mode == "managed":
+        images["components.redis.image"] = spec.components.redis.image
+    if spec.components.mail.mode == "test-mail":
+        images["components.mail.image"] = spec.components.mail.image
+
+    for label, image in images.items():
+        if image_is_unpublished(image):
+            checks.append(
+                Check(
+                    f"{label}.published",
+                    "error",
+                    f"{label} still uses the unpublished placeholder.",
+                    "Set an available image reference before apply.",
+                )
+            )
+        elif not image_is_digest_pinned(image):
+            level = "error" if spec.profile == "self-hosted" else "warning"
+            checks.append(
+                Check(
+                    f"{label}.digest",
+                    level,
+                    f"{label} is not pinned by OCI digest.",
+                    "Use an image@sha256:... reference from a verified distribution.",
+                )
+            )
+        else:
+            checks.append(
+                Check(
+                    f"{label}.digest",
+                    "ok",
+                    f"{label} is pinned by OCI digest.",
+                )
+            )
+
+    if spec.release.manifest_url and spec.release.manifest_sha256:
+        checks.append(
+            Check(
+                "release.manifest",
+                "ok",
+                "A distribution manifest URL and expected digest are recorded.",
+            )
+        )
+    else:
+        checks.append(
+            Check(
+                "release.manifest",
+                "error" if spec.profile == "self-hosted" else "warning",
+                "No verified distribution manifest is recorded.",
+                "Use a published signed distribution manifest for self-hosted apply.",
+            )
+        )
+    if spec.profile == "self-hosted":
+        checks.append(
+            Check(
+                "release.signature_verification",
+                "error",
+                "Signed distribution-manifest verification is not implemented in the deployer yet.",
+                "Use the published verifier/bootstrap slice before a production apply.",
+            )
+        )
+    checks.append(
+        Check(
+            "modules.image_composition",
+            "error" if spec.enabled_modules else "warning",
+            "The selected module set is not yet verified against image package contents.",
+            "Use the signed distribution composition evidence before production apply.",
+        )
+    )
+
+    values = read_env(paths.env)
+    required = {"MASTER_KEY_B64", "DATABASE_URL"}
+    if spec.components.redis.mode != "disabled":
+        required.add("REDIS_URL")
+    if spec.components.storage.mode == "s3":
+        required.update(
+            {
+                "FILE_STORAGE_S3_ENDPOINT_URL",
+                "FILE_STORAGE_S3_REGION",
+                "FILE_STORAGE_S3_ACCESS_KEY_ID",
+                "FILE_STORAGE_S3_SECRET_ACCESS_KEY",
+                "FILE_STORAGE_S3_BUCKET",
+            }
+        )
+    missing = sorted(name for name in required if not values.get(name))
+    checks.append(
+        Check(
+            "secrets.required",
+            "error" if missing else "ok",
+            (
+                "Missing required secret values: " + ", ".join(missing)
+                if missing
+                else "Required runtime secret references are populated."
+            ),
+            "Re-run configure with the required external service values." if missing else "",
+        )
+    )
+    if paths.env.exists():
+        mode = stat.S_IMODE(paths.env.stat().st_mode)
+        checks.append(
+            Check(
+                "secrets.permissions",
+                "error" if mode & 0o077 else "ok",
+                (
+                    f"{paths.env.name} has unsafe mode {mode:04o}."
+                    if mode & 0o077
+                    else f"{paths.env.name} is private ({mode:04o})."
+                ),
+                f"Run chmod 600 {paths.env}" if mode & 0o077 else "",
+            )
+        )
+    if spec.profile == "self-hosted":
+        checks.append(
+            Check(
+                "bootstrap.administrator",
+                "warning",
+                "Production first-administrator enrollment is not automated yet.",
+                "Use the controlled one-time administrator procedure until the enrollment slice lands.",
+            )
+        )
+    if spec.components.mail.mode == "external-relay":
+        checks.append(
+            Check(
+                "mail.provisioning",
+                "warning",
+                "The external relay is selected but Mail profile seeding remains an explicit post-install configuration task.",
+                "Create the reusable server and credential profile in Mail after first login.",
+            )
+        )
+    if spec.components.storage.mode == "local" and spec.profile == "self-hosted":
+        checks.append(
+            Check(
+                "storage.local",
+                "warning",
+                "Local file storage is suitable for one host but prevents stateless API scale-out.",
+                "Include files-data in backup/restore drills or configure S3 storage.",
+            )
+        )
+    return tuple(checks)
+
+
+def host_checks(
+    spec: InstallationSpec,
+    paths: BundlePaths,
+    *,
+    command_runner: CommandRunner | None = None,
+) -> tuple[Check, ...]:
+    checks: list[Check] = []
+    machine = platform.machine().lower()
+    supported = machine in {"x86_64", "amd64", "aarch64", "arm64"}
+    checks.append(
+        Check(
+            "host.architecture",
+            "ok" if supported else "error",
+            f"Host architecture is {machine or 'unknown'}.",
+            "Use a supported amd64 or arm64 host." if not supported else "",
+        )
+    )
+
+    memory_bytes = _memory_bytes()
+    if memory_bytes is not None:
+        minimum = 4 * 1024**3
+        checks.append(
+            Check(
+                "host.memory",
+                "ok" if memory_bytes >= minimum else "warning",
+                f"Host memory is approximately {memory_bytes / 1024**3:.1f} GiB.",
+                "Use at least 4 GiB for the base container profile."
+                if memory_bytes < minimum
+                else "",
+            )
+        )
+    cpu_count = os.cpu_count()
+    if cpu_count is not None:
+        checks.append(
+            Check(
+                "host.cpu",
+                "ok" if cpu_count >= 2 else "warning",
+                f"Host reports {cpu_count} logical CPU(s).",
+                "Use at least two logical CPUs for the base container profile."
+                if cpu_count < 2
+                else "",
+            )
+        )
+    entropy = _entropy_available()
+    if entropy is not None:
+        checks.append(
+            Check(
+                "host.entropy",
+                "ok" if entropy >= 256 else "warning",
+                f"Kernel entropy availability is {entropy}.",
+                "Wait for or provide sufficient host entropy before generating production keys."
+                if entropy < 256
+                else "",
+            )
+        )
+    free_bytes = shutil.disk_usage(paths.root).free
+    minimum_free = 10 * 1024**3
+    checks.append(
+        Check(
+            "host.disk",
+            "ok" if free_bytes >= minimum_free else "warning",
+            f"Free installation filesystem space is approximately {free_bytes / 1024**3:.1f} GiB.",
+            "Keep at least 10 GiB free before pulling images and creating data volumes."
+            if free_bytes < minimum_free
+            else "",
+        )
+    )
+
+    docker = shutil.which("docker")
+    if docker is None:
+        checks.append(
+            Check(
+                "host.compose",
+                "error",
+                "Docker CLI is not installed or not on PATH.",
+                "Install Docker Engine with the Compose v2 plugin.",
+            )
+        )
+    else:
+        runner = command_runner or _run_command
+        result = runner((docker, "compose", "version", "--short"), paths.root)
+        checks.append(
+            Check(
+                "host.compose",
+                "ok" if result.returncode == 0 else "error",
+                (
+                    f"Docker Compose is available ({result.stdout.strip()})."
+                    if result.returncode == 0
+                    else "Docker Compose v2 is not available."
+                ),
+                "Install or enable the Docker Compose v2 plugin."
+                if result.returncode != 0
+                else "",
+            )
+        )
+        daemon = runner(
+            (docker, "info", "--format", "{{.ServerVersion}}"),
+            paths.root,
+        )
+        checks.append(
+            Check(
+                "host.container_runtime",
+                "ok" if daemon.returncode == 0 else "error",
+                (
+                    f"Docker daemon is reachable ({daemon.stdout.strip()})."
+                    if daemon.returncode == 0
+                    else "Docker CLI cannot reach the Docker daemon."
+                ),
+                "Start Docker and grant this operator access to the daemon."
+                if daemon.returncode != 0
+                else "",
+            )
+        )
+
+    values = read_env(paths.env)
+    if spec.components.postgres.mode == "external":
+        checks.append(
+            _endpoint_check(
+                "external.postgres",
+                "External PostgreSQL",
+                values.get("DATABASE_URL", ""),
+                default_ports={"postgresql": 5432, "postgresql+psycopg": 5432},
+            )
+        )
+    if spec.components.redis.mode == "external":
+        checks.append(
+            _endpoint_check(
+                "external.redis",
+                "External Redis",
+                values.get("REDIS_URL", ""),
+                default_ports={"redis": 6379, "rediss": 6379},
+            )
+        )
+    if spec.components.storage.mode == "s3":
+        checks.append(
+            _endpoint_check(
+                "external.s3",
+                "S3-compatible storage",
+                values.get("FILE_STORAGE_S3_ENDPOINT_URL", ""),
+                default_ports={"http": 80, "https": 443},
+            )
+        )
+
+    receipt = _read_receipt(paths.receipt)
+    previous_listen = receipt.get("listen")
+    desired_listen = {
+        "address": spec.listen.address,
+        "port": spec.listen.port,
+    }
+    if not receipt or previous_listen != desired_listen:
+        available = _port_available(spec.listen.address, spec.listen.port)
+        checks.append(
+            Check(
+                "host.listen_port",
+                "ok" if available else "error",
+                (
+                    f"Listen endpoint {spec.listen.address}:{spec.listen.port} is available."
+                    if available
+                    else f"Listen endpoint {spec.listen.address}:{spec.listen.port} is already in use."
+                ),
+                "Choose another listen port or stop the conflicting service."
+                if not available
+                else "",
+            )
+        )
+    return tuple(checks)
+
+
+def _read_receipt(path: Path) -> Mapping[str, object]:
+    if not path.exists():
+        return {}
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def _memory_bytes() -> int | None:
+    try:
+        for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines():
+            if line.startswith("MemTotal:"):
+                return int(line.split()[1]) * 1024
+    except (OSError, ValueError, IndexError):
+        return None
+    return None
+
+
+def _entropy_available() -> int | None:
+    try:
+        return int(
+            Path("/proc/sys/kernel/random/entropy_avail")
+            .read_text(encoding="utf-8")
+            .strip()
+        )
+    except (OSError, ValueError):
+        return None
+
+
+def _port_available(address: str, port: int) -> bool:
+    family = socket.AF_INET6 if ":" in address else socket.AF_INET
+    with socket.socket(family, socket.SOCK_STREAM) as handle:
+        try:
+            handle.bind((address, port))
+        except OSError:
+            return False
+    return True
+
+
+def _endpoint_check(
+    check_id: str,
+    label: str,
+    url: str,
+    *,
+    default_ports: Mapping[str, int],
+) -> Check:
+    try:
+        parsed = urlsplit(url)
+        host = parsed.hostname
+        port = parsed.port or default_ports.get(parsed.scheme)
+    except ValueError as exc:
+        return Check(
+            check_id,
+            "error",
+            f"{label} endpoint is invalid: {exc}",
+            "Correct the private endpoint binding and rerun doctor.",
+        )
+    if not host or port is None:
+        return Check(
+            check_id,
+            "error",
+            f"{label} endpoint has no usable host and port.",
+            "Correct the private endpoint binding and rerun doctor.",
+        )
+    try:
+        with socket.create_connection((host, port), timeout=1.5):
+            pass
+    except OSError as exc:
+        return Check(
+            check_id,
+            "error",
+            f"{label} is not reachable at {host}:{port}: {exc}",
+            "Check DNS, routing, firewall, service health, and the selected endpoint.",
+        )
+    return Check(
+        check_id,
+        "ok",
+        f"{label} is reachable at {host}:{port}.",
+    )
+
+
+def _run_command(
+    argv: Sequence[str], cwd: Path
+) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        list(argv),
+        cwd=cwd,
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=10,
+    )