from __future__ import annotations

import hashlib
import importlib
import base64
import json
import os
import re
import runpy
import shlex
import shutil
import stat
import subprocess
import sys
import tempfile
import tomllib
import unittest
from dataclasses import replace
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from sqlalchemy import Column, Integer, MetaData, Table, create_engine, insert, inspect
from sqlalchemy.orm import Session

# Keep the default app import side effect from bootstrapping a development DB.
_TEST_ROOT = Path(tempfile.mkdtemp(prefix="govoplan-module-tests-"))
os.environ.setdefault("APP_ENV", "test")
os.environ.setdefault("DATABASE_URL", f"sqlite:///{_TEST_ROOT / 'module-test.db'}")
os.environ.setdefault("DEV_BOOTSTRAP_ENABLED", "false")
os.environ.setdefault("CELERY_ENABLED", "false")

from fastapi import APIRouter, FastAPI
from fastapi.testclient import TestClient
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from govoplan_core.core.events import event_context
from govoplan_core.core import module_installer as module_installer_module
from govoplan_core.core.migrations import migration_metadata_plan
from govoplan_core.core.module_management import (
    ModuleInstallPlan,
    ModuleInstallPlanItem,
    ModuleManagementError,
    desired_modules_after_package_plan,
    module_install_plan_commands,
    normalize_module_install_plan_item,
    saved_desired_enabled_modules,
    saved_module_install_plan,
    save_desired_enabled_modules,
    save_module_install_plan,
    plan_desired_enabled_modules,
)
from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, MaintenanceMode, saved_maintenance_mode, save_maintenance_mode
from govoplan_core.core.module_installer import (
    cancel_module_installer_request,
    claim_next_module_installer_request,
    list_module_installer_runs,
    list_module_installer_requests,
    module_installer_daemon_status,
    module_install_preflight,
    module_install_catalog_companion_module_ids,
    module_manifest_compatibility_issues,
    module_installer_lock_status,
    queue_module_installer_request,
    read_module_installer_request,
    read_module_installer_run,
    retry_module_installer_request,
    rollback_module_install_run,
    run_module_install_plan,
    structured_install_commands,
    supervise_module_install_plan,
    update_module_installer_daemon_status,
    update_module_installer_request,
)
from govoplan_core.core.configuration_packages import (
    CONFIGURATION_PROVIDER_CAPABILITY,
    ConfigurationApplyResult,
    ConfigurationExportResult,
    ConfigurationExportSelection,
    ConfigurationPackageFragment,
    ConfigurationPlanItem,
    ConfigurationPreflightContext,
    ConfigurationPreflightResult,
    ConfigurationProvider,
    ConfigurationProviderDescription,
    ConfigurationRequiredData,
    apply_configuration_package,
    configuration_package_catalog,
    dry_run_configuration_package,
    export_configuration_package,
    sign_configuration_package_catalog,
    validate_configuration_package_catalog,
)
from govoplan_core.core.module_license import issue_module_license, module_license_decision, module_license_diagnostics, validate_module_license
from govoplan_core.core.module_package_catalog import (
    module_package_catalog,
    record_module_package_catalog_acceptance,
    sign_module_package_catalog,
    validate_module_package_catalog,
)
from govoplan_core.core.modules import FrontendModule, FrontendRoute, MigrationRetirementPlan, ModuleCompatibility, ModuleMigrationTask, ModuleMigrationTaskContext, ModuleMigrationTaskResult, ModuleUninstallGuardResult, PublicFrontendRoute
from govoplan_core.core.module_guards import drop_table_retirement_provider
from govoplan_core.core.modules import MigrationSpec, ModuleInterfaceProvider, ModuleInterfaceRequirement, ModuleManifest, PermissionDefinition, RoleTemplate
from govoplan_core.core.registry import PlatformRegistry, RegistryError
from govoplan_core.admin.models import SystemSettings
from govoplan_core.db.base import Base
from govoplan_core.db.bootstrap import bootstrap_dev_data
from govoplan_core.db.migrations import run_registered_module_migration_tasks
from govoplan_core.db.session import configure_database as configure_database_handle, get_database, reset_database
from govoplan_core.security.module_permissions import scopes_grant_compatible
from govoplan_core.security.permissions import scope_grants
from govoplan_core.server.app import create_app
from govoplan_core.server.config import GovoplanServerConfig
from govoplan_core.server.platform import create_platform_router
from govoplan_core.server.registry import available_module_manifests, build_platform_registry
from govoplan_core.server.route_validation import RouteCollisionError
from govoplan_core.tenancy.scope import Tenant, create_scope_tables
from govoplan_files.backend.storage.backends import LocalFilesystemStorageBackend, StorageBackendError

_MANIFEST_PROJECT_MODULES = {
    "access": "govoplan_access",
    "admin": "govoplan_admin",
    "tenancy": "govoplan_tenancy",
    "policy": "govoplan_policy",
    "audit": "govoplan_audit",
    "dashboard": "govoplan_dashboard",
    "files": "govoplan_files",
    "mail": "govoplan_mail",
    "campaigns": "govoplan_campaign",
    "docs": "govoplan_docs",
    "ops": "govoplan_ops",
}
_MODULE_ID_RE = re.compile(r"^[a-z][a-z0-9_]*$")
_PACKAGE_NAME_RE = re.compile(r"^@govoplan/[a-z0-9_-]+-webui$")


def configure_database(database_url: str):
    return configure_database_handle(database_url, dispose_previous=True)


def _join_route_path(prefix: str, path: str) -> str:
    if not prefix:
        return path
    if not path:
        return prefix
    return f"{prefix.rstrip('/')}/{path.lstrip('/')}"


def _route_paths(routes_or_app, *, prefix: str = "") -> set[str]:
    routes = getattr(routes_or_app, "routes", routes_or_app)
    paths: set[str] = set()
    for route in routes:
        path = getattr(route, "path", None)
        if path:
            paths.add(_join_route_path(prefix, path))
        include_context = getattr(route, "include_context", None)
        nested_prefix = _join_route_path(prefix, str(getattr(include_context, "prefix", "") or ""))
        nested_router = getattr(route, "original_router", None)
        if nested_router is not None:
            paths.update(_route_paths(nested_router, prefix=nested_prefix))
        nested_routes = getattr(route, "routes", None)
        if nested_routes is not None:
            paths.update(_route_paths(nested_routes, prefix=nested_prefix))
    return paths


def _settings(root: Path) -> SimpleNamespace:
    return SimpleNamespace(
        app_env="test",
        database_url=f"sqlite:///{root / 'test.db'}",
        dev_mailbox_api_enabled=False,
        file_storage_backend="local",
        file_storage_local_root=str(root / "files"),
        file_storage_local_fallback_roots="",
        file_storage_s3_endpoint_url=None,
        file_storage_s3_region=None,
        file_storage_s3_bucket=None,
        file_storage_s3_access_key_id=None,
        file_storage_s3_secret_access_key=None,
        s3_endpoint_url=None,
        s3_region=None,
        s3_bucket=None,
        s3_access_key_id=None,
        s3_secret_access_key=None,
        file_upload_max_bytes=1024 * 1024,
        file_upload_zip_max_bytes=10 * 1024 * 1024,
        celery_enabled=False,
        redis_url="redis://127.0.0.1:6379/15",
        mock_mailbox_dir=str(root / "mock-mailbox"),
    )


class ModuleSystemTests(unittest.TestCase):
    @classmethod
    def tearDownClass(cls) -> None:
        reset_database(dispose=True)
        shutil.rmtree(_TEST_ROOT, ignore_errors=True)

    def _app_for_modules(self, modules: tuple[str, ...]):
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-app-", dir=_TEST_ROOT))
        settings = _settings(root)
        config = GovoplanServerConfig(
            title="GovOPlaN Module Test",
            version="test",
            settings=settings,
            enabled_modules=modules,
            base_routers=(),
            post_module_routers=(),
            extra_routers=(),
            lifespan=None,
            app_configurators=(),
        )
        return create_app(config), settings

    def _source_project_version(self, package_module: str) -> str:
        module = importlib.import_module(package_module)
        module_file = getattr(module, "__file__", None)
        if not module_file:
            self.fail(f"Could not locate source package for {package_module}")
        for path in Path(module_file).resolve().parents:
            pyproject_path = path / "pyproject.toml"
            if pyproject_path.exists():
                with pyproject_path.open("rb") as handle:
                    return str(tomllib.load(handle)["project"]["version"])
        self.fail(f"Could not locate pyproject.toml for {package_module}")

    def test_discovers_installed_core_and_product_module_manifests(self) -> None:
        manifests = available_module_manifests()
        self.assertTrue({"access", "admin", "tenancy", "policy", "audit", "dashboard", "files", "mail", "campaigns", "docs", "ops"}.issubset(manifests))
        self.assertEqual(manifests["campaigns"].dependencies, ())
        self.assertTrue(manifests["campaigns"].required_capabilities)
        self.assertEqual(
            manifests["campaigns"].optional_dependencies,
            ("files", "mail", "notifications", "addresses", "postbox", "approvals"),
        )
        self.assertEqual(manifests["dashboard"].dependencies, ())
        self.assertTrue(manifests["dashboard"].required_capabilities)
        self.assertEqual(manifests["docs"].dependencies, ())
        self.assertTrue(manifests["docs"].required_capabilities)
        self.assertEqual(manifests["ops"].dependencies, ())
        self.assertTrue(manifests["ops"].required_capabilities)

    def test_access_manifest_contributes_admin_frontend_metadata(self) -> None:
        manifests = available_module_manifests()
        access = manifests["access"]

        self.assertEqual(("/admin",), tuple(item.path for item in access.nav_items))
        self.assertIsNotNone(access.frontend)
        assert access.frontend is not None
        self.assertEqual("@govoplan/access-webui", access.frontend.package_name)
        self.assertEqual(("/admin",), tuple(route.path for route in access.frontend.routes))
        self.assertEqual(("/admin",), tuple(item.path for item in access.frontend.nav_items))
        self.assertTrue(access.frontend.routes[0].required_any)
        self.assertEqual("admin", access.frontend.nav_items[0].icon)

    def test_obsolete_access_core_import_aliases_are_removed(self) -> None:
        obsolete_paths = (
            "govoplan_core.security.api_keys",
            "govoplan_core.security.sessions",
            "govoplan_core.security.passwords",
            "govoplan_core.api.v1.auth",
            "govoplan_core.api.v1.admin",
            "govoplan_core.api.v1.admin_schemas",
            "govoplan_core.admin.service",
            "govoplan_core.admin.governance",
        )

        for legacy_path in obsolete_paths:
            with self.subTest(legacy_path=legacy_path):
                self.assertIsNone(importlib.util.find_spec(legacy_path))

    def test_tenant_wildcard_grants_module_tenant_scopes(self) -> None:
        self.assertTrue(scope_grants("tenant:*", "calendar:event:read"))
        self.assertTrue(scopes_grant_compatible(["tenant:*"], "calendar:event:read"))
        self.assertFalse(scope_grants("tenant:*", "system:settings:read"))
        self.assertTrue(scopes_grant_compatible(["access:membership:read"], "admin:users:read"))
        self.assertTrue(scopes_grant_compatible(["admin:users:read"], "access:membership:read"))
        self.assertTrue(scopes_grant_compatible(["access:tenant:read"], "system:tenants:read"))
        self.assertTrue(scopes_grant_compatible(["system:*"], "access:tenant:read"))
        self.assertFalse(scopes_grant_compatible(["tenant:*"], "system:tenants:read"))
        self.assertFalse(scopes_grant_compatible(["tenant:*"], "access:tenant:read"))

    def test_core_webui_retired_legacy_admin_api_surface(self) -> None:
        webui_src = Path(__file__).resolve().parents[1] / "webui" / "src"
        legacy_admin_import_pattern = re.compile(r"api/admin(?=[\"'#?]|$)")
        self.assertFalse((webui_src / "api" / "admin.ts").exists())
        for path in webui_src.rglob("*.ts*"):
            with self.subTest(path=path.relative_to(webui_src)):
                self.assertIsNone(legacy_admin_import_pattern.search(path.read_text(encoding="utf-8")))

    def test_platform_modules_own_live_legacy_model_tables(self) -> None:
        from govoplan_admin.backend.db.models import GovernanceTemplate
        from govoplan_audit.backend.db.models import AuditLog
        from govoplan_access.backend.db.base import AccessBase
        from govoplan_access.backend.db.models import Account, ApiKey, AuthSession, User
        from govoplan_core.admin.models import SystemSettings
        from govoplan_core.db.base import Base
        from govoplan_core.tenancy.scope import scope_registry
        from govoplan_tenancy.backend.db.models import Tenant

        self.assertIs(AccessBase, Base)
        self.assertEqual("access_accounts", Account.__tablename__)
        self.assertEqual("core_scopes", Tenant.__tablename__)
        self.assertEqual("access_users", User.__tablename__)
        self.assertEqual("access_api_keys", ApiKey.__tablename__)
        self.assertEqual("access_auth_sessions", AuthSession.__tablename__)
        self.assertEqual("audit_log", AuditLog.__tablename__)
        self.assertEqual("admin_governance_templates", GovernanceTemplate.__tablename__)
        self.assertEqual("core_system_settings", SystemSettings.__tablename__)
        self.assertIn("access_accounts", Base.metadata.tables)
        self.assertIn("core_scopes", scope_registry.metadata.tables)
        self.assertNotIn("core_scopes", Base.metadata.tables)
        self.assertNotIn("accounts", Base.metadata.tables)
        self.assertNotIn("auth_sessions", Base.metadata.tables)

    def test_module_manifest_versions_match_source_project_versions(self) -> None:
        manifests = available_module_manifests()
        for manifest_id, package_module in _MANIFEST_PROJECT_MODULES.items():
            with self.subTest(manifest_id=manifest_id):
                self.assertIn(manifest_id, manifests)
                self.assertEqual(self._source_project_version(package_module), manifests[manifest_id].version)

    def test_available_module_manifests_follow_contract_shape(self) -> None:
        manifests = available_module_manifests()
        for manifest_id, manifest in manifests.items():
            with self.subTest(manifest_id=manifest_id):
                self.assertRegex(manifest.id, _MODULE_ID_RE)
                self.assertEqual(manifest_id, manifest.id)
                self.assertTrue(manifest.name.strip())
                self.assertTrue(manifest.version.strip())
                self.assertEqual(len(manifest.dependencies), len(set(manifest.dependencies)))
                self.assertEqual(len(manifest.optional_dependencies), len(set(manifest.optional_dependencies)))
                self.assertFalse(set(manifest.dependencies) & set(manifest.optional_dependencies))
                self.assertNotIn(manifest.id, manifest.dependencies)
                self.assertNotIn(manifest.id, manifest.optional_dependencies)
                self.assertEqual(manifest.compatibility.manifest_contract_version, "1")
                if manifest.migration_spec is not None:
                    self.assertEqual(manifest.id, manifest.migration_spec.module_id)
                    self.assertTrue(manifest.migration_spec.metadata is not None or manifest.migration_spec.script_location)
                if manifest.frontend is not None:
                    self.assertEqual(manifest.id, manifest.frontend.module_id)
                    if manifest.frontend.package_name is not None:
                        self.assertRegex(manifest.frontend.package_name, _PACKAGE_NAME_RE)
                    for route in (*manifest.frontend.routes, *manifest.frontend.settings_routes, *manifest.frontend.public_routes):
                        self.assertTrue(route.path.startswith("/"))
                        self.assertTrue(route.component.strip())
                    for item in (*manifest.nav_items, *manifest.frontend.nav_items):
                        self.assertTrue(item.path.startswith("/"))
                        self.assertTrue(item.label.strip())
                        if item.icon is not None:
                            self.assertTrue(item.icon.strip())

    def test_registry_rejects_unsupported_manifest_contract_version(self) -> None:
        registry = PlatformRegistry()
        registry.register(ModuleManifest(
            id="example",
            name="Example",
            version="test",
            compatibility=ModuleCompatibility(manifest_contract_version="2"),
        ))

        with self.assertRaisesRegex(RegistryError, "unsupported manifest contract version"):
            registry.validate()

    def test_manifest_can_preserve_a_legacy_permission_namespace(self) -> None:
        registry = PlatformRegistry()
        registry.register(
            ModuleManifest(
                id="workflow_engine",
                name="Workflow Engine",
                version="test",
                permission_namespace="workflow",
                permissions=(
                    PermissionDefinition(
                        scope="workflow:definition:read",
                        label="Read workflows",
                        description="Read workflow definitions.",
                        category="Workflow",
                        level="tenant",
                        module_id="workflow",
                        resource="definition",
                        action="read",
                    ),
                ),
            )
        )

        registry.validate()

    def test_manifest_rejects_an_undeclared_permission_namespace(self) -> None:
        registry = PlatformRegistry()
        registry.register(
            ModuleManifest(
                id="workflow_engine",
                name="Workflow Engine",
                version="test",
                permissions=(
                    PermissionDefinition(
                        scope="workflow:definition:read",
                        label="Read workflows",
                        description="Read workflow definitions.",
                        category="Workflow",
                        level="tenant",
                        module_id="workflow",
                        resource="definition",
                        action="read",
                    ),
                ),
            )
        )

        with self.assertRaisesRegex(RegistryError, "expected 'workflow_engine'"):
            registry.validate()

    def test_default_authenticated_role_templates_are_managed_tenant_roles(self) -> None:
        for template, message in (
            (
                RoleTemplate(
                    slug="invalid-system-default",
                    name="Invalid",
                    description="Invalid system default.",
                    permissions=("system:*",),
                    level="system",
                    default_authenticated=True,
                ),
                "must be tenant-level",
            ),
            (
                RoleTemplate(
                    slug="invalid-unmanaged-default",
                    name="Invalid",
                    description="Invalid unmanaged default.",
                    permissions=("tenant:*",),
                    managed=False,
                    default_authenticated=True,
                ),
                "must be managed",
            ),
        ):
            with self.subTest(template=template.slug):
                registry = PlatformRegistry()
                registry.register(
                    ModuleManifest(
                        id="example",
                        name="Example",
                        version="test",
                        role_templates=(template,),
                    )
                )
                with self.assertRaisesRegex(RegistryError, message):
                    registry.validate()

    def test_default_authenticated_role_templates_reject_wildcard_permissions(self) -> None:
        registry = PlatformRegistry()
        registry.register(
            ModuleManifest(
                id="example",
                name="Example",
                version="test",
                role_templates=(
                    RoleTemplate(
                        slug="unsafe-default",
                        name="Unsafe default",
                        description="Must not grant broad authority automatically.",
                        permissions=("tenant:*",),
                        default_authenticated=True,
                    ),
                ),
            )
        )

        with self.assertRaisesRegex(RegistryError, "must use explicit permissions"):
            registry.validate()

    def test_role_template_slugs_are_unique_per_level_across_modules(self) -> None:
        registry = PlatformRegistry()
        for module_id in ("first", "second"):
            registry.register(
                ModuleManifest(
                    id=module_id,
                    name=module_id.title(),
                    version="test",
                    role_templates=(
                        RoleTemplate(
                            slug="shared-reader",
                            name="Shared reader",
                            description="A colliding role template.",
                            permissions=(),
                        ),
                    ),
                )
            )

        with self.assertRaisesRegex(RegistryError, "Duplicate tenant role template slug"):
            registry.validate()

    def test_registry_rejects_invalid_frontend_manifest_shape(self) -> None:
        registry = PlatformRegistry()
        registry.register(ModuleManifest(
            id="example",
            name="Example",
            version="test",
            frontend=FrontendModule(
                module_id="example",
                package_name="@govoplan/example-webui",
                routes=(FrontendRoute(path="missing-leading-slash", component="ExamplePage"),),
            ),
        ))

        with self.assertRaisesRegex(RegistryError, "Frontend route"):
            registry.validate()

    def test_public_frontend_routes_are_explicitly_allowlisted(self) -> None:
        registry = PlatformRegistry()
        registry.register(ModuleManifest(
            id="example",
            name="Example",
            version="test",
            frontend=FrontendModule(
                module_id="example",
                package_name="@govoplan/example-webui",
                routes=(FrontendRoute(path="/example", component="ExamplePage"),),
                public_routes=(
                    PublicFrontendRoute(
                        path="/example/public/:token",
                        component="ExamplePublicPage",
                    ),
                ),
            ),
        ))
        registry.register(ModuleManifest(
            id="private",
            name="Private",
            version="test",
            frontend=FrontendModule(
                module_id="private",
                package_name="@govoplan/private-webui",
                routes=(FrontendRoute(path="/private", component="PrivatePage"),),
            ),
        ))
        registry.validate()

        app = FastAPI()
        app.state.govoplan_registry = registry
        app.include_router(create_platform_router(), prefix="/api/v1")
        with TestClient(app) as client:
            response = client.get("/api/v1/platform/public-modules")

        self.assertEqual(response.status_code, 200, response.text)
        self.assertEqual(["example"], [item["id"] for item in response.json()["modules"]])
        public_module = response.json()["modules"][0]
        self.assertNotIn("dependencies", public_module)
        self.assertNotIn("nav", public_module["frontend"])
        self.assertNotIn("routes", public_module["frontend"])
        self.assertEqual(
            ["/example/public/:token"],
            [item["path"] for item in public_module["frontend"]["public_routes"]],
        )

    def test_registry_rejects_duplicate_public_frontend_routes(self) -> None:
        registry = PlatformRegistry()
        for module_id in ("first", "second"):
            registry.register(ModuleManifest(
                id=module_id,
                name=module_id.title(),
                version="test",
                frontend=FrontendModule(
                    module_id=module_id,
                    public_routes=(
                        PublicFrontendRoute(
                            path="/shared/public/:token",
                            component=f"{module_id.title()}PublicPage",
                        ),
                    ),
                ),
            ))

        with self.assertRaisesRegex(RegistryError, "Duplicate public frontend route"):
            registry.validate()

    def test_server_startup_rejects_duplicate_configured_routes(self) -> None:
        first = APIRouter()
        second = APIRouter()

        @first.get("/collision")
        def first_collision() -> dict[str, bool]:
            return {"first": True}

        @second.get("/collision")
        def second_collision() -> dict[str, bool]:
            return {"second": True}

        config = GovoplanServerConfig(
            title="GovOPlaN Route Collision Test",
            version="test",
            settings=_settings(Path(tempfile.mkdtemp(prefix="govoplan-route-collision-", dir=_TEST_ROOT))),
            enabled_modules=(),
            base_routers=(first, second),
            post_module_routers=(),
            extra_routers=(),
            lifespan=None,
            app_configurators=(),
        )

        with self.assertRaisesRegex(RouteCollisionError, "GET .*/collision"):
            create_app(config)

    def test_module_lifecycle_rejects_duplicate_module_routes(self) -> None:
        def route_factory(label: str):
            def factory(_context):
                router = APIRouter()

                @router.get("/collision")
                def collision() -> dict[str, str]:
                    return {"module": label}

                return router

            return factory

        config = GovoplanServerConfig(
            title="GovOPlaN Module Route Collision Test",
            version="test",
            settings=_settings(Path(tempfile.mkdtemp(prefix="govoplan-module-route-collision-", dir=_TEST_ROOT))),
            enabled_modules=("alpha", "beta"),
            base_routers=(),
            post_module_routers=(),
            extra_routers=(),
            lifespan=None,
            app_configurators=(),
            manifest_factories=(
                lambda: ModuleManifest(id="alpha", name="Alpha", version="test", route_factory=route_factory("alpha")),
                lambda: ModuleManifest(id="beta", name="Beta", version="test", route_factory=route_factory("beta")),
            ),
        )

        with self.assertRaisesRegex(RouteCollisionError, "GET .*/collision"):
            create_app(config)

    def test_enabled_module_permutations_register_expected_routes(self) -> None:
        cases = (
            ("core_only", (), set(), set()),
            ("files_only", ("files",), {"access", "files"}, {"/api/v1/files"}),
            ("mail_only", ("mail",), {"access", "mail"}, {"/api/v1/mail"}),
            ("campaign_without_files_or_mail", ("campaigns",), {"access", "campaigns"}, {"/api/v1/campaigns"}),
            ("campaign_without_files", ("campaigns", "mail"), {"access", "campaigns", "mail"}, {"/api/v1/campaigns", "/api/v1/mail"}),
            ("campaign_with_files_but_no_mail", ("campaigns", "files"), {"access", "campaigns", "files"}, {"/api/v1/campaigns", "/api/v1/files"}),
            ("dashboard_only", ("dashboard",), {"access", "dashboard"}, set()),
            ("docs_and_ops", ("docs", "ops"), {"access", "docs", "ops"}, {"/api/v1/docs", "/api/v1/ops"}),
            ("full_product", ("dashboard", "campaigns", "files", "mail", "docs", "ops"), {"access", "dashboard", "campaigns", "files", "mail", "docs", "ops"}, {"/api/v1/campaigns", "/api/v1/files", "/api/v1/mail", "/api/v1/docs", "/api/v1/ops"}),
        )
        for name, enabled_modules, expected_modules, expected_prefixes in cases:
            with self.subTest(name=name):
                app, _settings_obj = self._app_for_modules(enabled_modules)
                route_paths = _route_paths(app)
                self.assertIn("/api/v1/platform/modules", route_paths)
                self.assertEqual(
                    "access" in expected_modules,
                    "/api/v1/auth/login" in route_paths,
                )
                self.assertEqual(
                    "access" in expected_modules,
                    "/api/v1/admin/users" in route_paths,
                )
                for prefix in ("/api/v1/campaigns", "/api/v1/files", "/api/v1/mail", "/api/v1/docs", "/api/v1/ops"):
                    has_prefix = any(path == prefix or path.startswith(f"{prefix}/") for path in route_paths)
                    self.assertEqual(prefix in expected_prefixes, has_prefix, f"{name}: {prefix}")

                with TestClient(app) as client:
                    health = client.get("/health")
                    self.assertEqual(health.status_code, 200, health.text)
                    self.assertEqual({"status": "ok"}, health.json())
                    response = client.get("/api/v1/platform/modules")
                self.assertEqual(
                    401 if "access" in expected_modules else 503,
                    response.status_code,
                    response.text,
                )
                payload_modules = {
                    item.id
                    for item in app.state.govoplan_registry.manifests()
                }
                self.assertEqual(expected_modules, payload_modules)

    def test_governance_template_routes_are_contributed_by_admin_module(self) -> None:
        access_only_app, _settings_obj = self._app_for_modules(("access",))
        access_only_paths = _route_paths(access_only_app)
        self.assertIn("/api/v1/admin/users", access_only_paths)
        self.assertNotIn("/api/v1/admin/system/governance-templates", access_only_paths)

        admin_app, _settings_obj = self._app_for_modules(("admin",))
        admin_paths = _route_paths(admin_app)
        self.assertIn("/api/v1/admin/system/governance-templates", admin_paths)


    def _run_physical_absence_probe(self, *, enabled_modules: tuple[str, ...], blocked_modules: tuple[str, ...]) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-absence-probe-", dir=_TEST_ROOT))
        enabled = ",".join(("tenancy", "access", *enabled_modules)) if enabled_modules else "tenancy,access"
        script = f"""
import importlib.abc
import json
import os
import sys
from pathlib import Path

os.environ["APP_ENV"] = "test"
os.environ["DATABASE_URL"] = {f"sqlite:///{root / 'probe.db'}"!r}
os.environ["DEV_BOOTSTRAP_ENABLED"] = "false"
os.environ["CELERY_ENABLED"] = "false"
os.environ["ENABLED_MODULES"] = {enabled!r}

BLOCKED = {blocked_modules!r}

class Blocker(importlib.abc.MetaPathFinder):
    def find_spec(self, fullname, path=None, target=None):
        for prefix in BLOCKED:
            if fullname == prefix or fullname.startswith(prefix + "."):
                raise ModuleNotFoundError(f"No module named {{prefix!r}}", name=prefix)
        return None

sys.meta_path.insert(0, Blocker())

from govoplan_core.server.app import create_app
from govoplan_core.server.config import GovoplanServerConfig
from govoplan_core.server.registry import build_platform_registry
from govoplan_core.db.session import reset_database

try:
    registry = build_platform_registry({enabled_modules!r})
    config = GovoplanServerConfig(
        title="absence probe",
        version="test",
        enabled_modules={enabled_modules!r},
        base_routers=(),
        post_module_routers=(),
        extra_routers=(),
        lifespan=None,
        app_configurators=(),
    )
    app = create_app(config)
    route_paths = [getattr(route, "path", "") for route in app.routes]
    print(json.dumps({{"modules": [item.id for item in registry.manifests()], "routes": route_paths}}, sort_keys=True))
finally:
    reset_database(dispose=True)
"""
        result = subprocess.run(
            [sys.executable, "-c", script],
            cwd=str(Path(__file__).resolve().parents[1]),
            env=os.environ.copy(),
            text=True,
            capture_output=True,
            check=False,
        )
        if result.returncode != 0:
            self.fail(result.stderr or result.stdout)

    def test_campaign_optional_integrations_follow_enabled_modules(self) -> None:
        cases = (
            (("campaigns",), False, False),
            (("campaigns", "mail"), False, True),
            (("campaigns", "files"), True, False),
            (("campaigns", "files", "mail"), True, True),
        )
        for enabled_modules, files_enabled, mail_enabled in cases:
            with self.subTest(enabled_modules=enabled_modules):
                registry = build_platform_registry(enabled_modules)
                self.assertTrue(registry.has_module("campaigns"))
                self.assertEqual(files_enabled, registry.integration_enabled("campaigns", "files"))
                self.assertEqual(mail_enabled, registry.integration_enabled("campaigns", "mail"))

    def test_tenant_lifecycle_with_product_modules_installed_and_absent(self) -> None:
        cases = (
            ("tenancy_only", ()),
            ("files_only", ("files",)),
            ("mail_only", ("mail",)),
            ("campaign_only", ("campaigns",)),
            ("full_product", ("files", "mail", "campaigns")),
        )
        for name, product_modules in cases:
            with self.subTest(name=name):
                app, _settings_obj = self._app_for_modules(("tenancy", *product_modules))
                database = get_database()
                create_scope_tables(database.engine)
                Base.metadata.create_all(bind=database.engine)
                with database.session() as session:
                    bootstrap_dev_data(session, user_password="test-admin")

                with TestClient(app) as client:
                    login = client.post(
                        "/api/v1/auth/login",
                        json={"email": "admin@example.local", "password": "test-admin"},
                    )
                    self.assertEqual(200, login.status_code, login.text)
                    headers = {"Authorization": f"Bearer {login.json()['access_token']}"}

                    tenancy_switch = client.post(
                        "/api/v1/tenancy/switch-tenant",
                        headers=headers,
                        json={"tenant_id": login.json()["tenant"]["id"]},
                    )
                    self.assertEqual(200, tenancy_switch.status_code, tenancy_switch.text)
                    self.assertEqual(login.json()["tenant"]["id"], tenancy_switch.json()["tenant_id"])

                    created = client.post(
                        "/api/v1/admin/tenants",
                        headers=headers,
                        json={
                            "slug": f"lifecycle-{name}",
                            "name": f"Lifecycle {name}",
                            "settings": {},
                        },
                    )
                    self.assertEqual(201, created.status_code, created.text)
                    tenant_id = created.json()["id"]

                    updated = client.patch(
                        f"/api/v1/admin/tenants/{tenant_id}",
                        headers=headers,
                        json={"name": f"Lifecycle {name} Updated"},
                    )
                    self.assertEqual(200, updated.status_code, updated.text)
                    self.assertEqual(f"Lifecycle {name} Updated", updated.json()["name"])

                    current_tenant_retire = client.request(
                        "DELETE",
                        f"/api/v1/admin/tenants/{login.json()['tenant']['id']}",
                        headers=headers,
                        json={"mode": "retire", "reason": "active tenant guard"},
                    )
                    self.assertEqual(409, current_tenant_retire.status_code, current_tenant_retire.text)

                    plan = client.get(
                        f"/api/v1/admin/tenants/{tenant_id}/deletion-plan",
                        headers=headers,
                    )
                    self.assertEqual(200, plan.status_code, plan.text)
                    self.assertTrue(plan.json()["allowed"])
                    self.assertFalse(plan.json()["destructive_supported"])

                    destructive = client.request(
                        "DELETE",
                        f"/api/v1/admin/tenants/{tenant_id}",
                        headers=headers,
                        json={"mode": "destroy", "reason": "not supported"},
                    )
                    self.assertEqual(409, destructive.status_code, destructive.text)
                    issue_codes = {item["code"] for item in destructive.json()["detail"]["plan"]["issues"]}
                    self.assertIn("tenant_data_present", issue_codes)

                    with database.session() as session:
                        empty_tenant = Tenant(
                            slug=f"empty-destroy-{name}",
                            name=f"Empty Destroy {name}",
                            settings={},
                            is_active=True,
                        )
                        session.add(empty_tenant)
                        session.commit()
                        empty_tenant_id = empty_tenant.id

                    destroyed = client.request(
                        "DELETE",
                        f"/api/v1/admin/tenants/{empty_tenant_id}",
                        headers=headers,
                        json={"mode": "destroy", "reason": "empty tenant cleanup"},
                    )
                    self.assertEqual(200, destroyed.status_code, destroyed.text)
                    self.assertEqual("destroy", destroyed.json()["plan"]["action"])
                    self.assertTrue(destroyed.json()["plan"]["destructive_supported"])

                    retired = client.request(
                        "DELETE",
                        f"/api/v1/admin/tenants/{tenant_id}",
                        headers=headers,
                        json={"mode": "retire", "reason": "module permutation test"},
                    )
                    self.assertEqual(200, retired.status_code, retired.text)
                    retired_item = retired.json()["item"]
                    self.assertFalse(retired_item["is_active"])
                    self.assertEqual("retired", retired_item["settings"]["lifecycle"]["state"])
                    self.assertEqual("module permutation test", retired_item["settings"]["lifecycle"]["reason"])

    def test_registry_rejects_missing_required_capability(self) -> None:
        registry = PlatformRegistry()
        registry.register(ModuleManifest(
            id="needs_auth",
            name="Needs Auth",
            version="test",
            required_capabilities=("auth.principalResolver",),
        ))

        with self.assertRaisesRegex(RegistryError, "requires unavailable capability"):
            registry.validate()

    def test_registry_keeps_optional_auth_modules_access_free(self) -> None:
        for modules in ((), ("poll",), ("evaluation",), ("scheduling",)):
            with self.subTest(modules=modules):
                registry = build_platform_registry(modules)
                self.assertFalse(registry.has_module("access"))

    def test_registry_prefers_selected_auth_provider_over_default(self) -> None:
        provider = ModuleManifest(
            id="alternative_auth",
            name="Alternative auth",
            version="test",
            capability_factories={
                "auth.principalResolver": lambda _context: object(),
            },
        )
        consumer = ModuleManifest(
            id="consumer",
            name="Consumer",
            version="test",
            required_capabilities=("auth.principalResolver",),
        )

        registry = build_platform_registry(
            ("alternative_auth", "consumer"),
            manifest_factories=(lambda: provider, lambda: consumer),
        )

        self.assertFalse(registry.has_module("access"))
        self.assertTrue(registry.has_module("alternative_auth"))
        self.assertTrue(registry.has_module("consumer"))

    def test_registry_resolves_required_interface_ranges(self) -> None:
        registry = PlatformRegistry()
        registry.register(ModuleManifest(
            id="files",
            name="Files",
            version="1.4.0",
            provides_interfaces=(ModuleInterfaceProvider(name="files.spaces", version="1.4.0"),),
        ))
        registry.register(ModuleManifest(
            id="campaigns",
            name="Campaigns",
            version="1.1.0",
            requires_interfaces=(
                ModuleInterfaceRequirement(
                    name="files.spaces",
                    version_min="1.0.0",
                    version_max_exclusive="2.0.0",
                ),
            ),
        ))

        snapshot = registry.validate()

        self.assertEqual({"files", "campaigns"}, {manifest.id for manifest in snapshot.manifests})

    def test_registry_rejects_missing_required_interface(self) -> None:
        registry = PlatformRegistry()
        registry.register(ModuleManifest(
            id="campaigns",
            name="Campaigns",
            version="1.1.0",
            requires_interfaces=(ModuleInterfaceRequirement(name="files.spaces", version_min="1.0.0"),),
        ))

        with self.assertRaisesRegex(RegistryError, "requires unavailable interface"):
            registry.validate()

    def test_registry_rejects_incompatible_optional_interface_provider(self) -> None:
        registry = PlatformRegistry()
        registry.register(ModuleManifest(
            id="files",
            name="Files",
            version="2.0.0",
            provides_interfaces=(ModuleInterfaceProvider(name="files.spaces", version="2.0.0"),),
        ))
        registry.register(ModuleManifest(
            id="campaigns",
            name="Campaigns",
            version="1.1.0",
            requires_interfaces=(
                ModuleInterfaceRequirement(
                    name="files.spaces",
                    version_min="1.0.0",
                    version_max_exclusive="2.0.0",
                    optional=True,
                ),
            ),
        ))

        with self.assertRaisesRegex(RegistryError, "available providers"):
            registry.validate()

    def test_access_and_organizations_do_not_hard_depend_on_tenancy(self) -> None:
        manifests = available_module_manifests()

        self.assertNotIn("tenancy", manifests["access"].dependencies)
        self.assertIn("tenancy", manifests["access"].optional_dependencies)
        self.assertNotIn("tenancy", manifests["organizations"].dependencies)
        self.assertIn("tenancy", manifests["organizations"].optional_dependencies)

    def test_module_management_plan_adds_protected_modules_and_required_dependencies(self) -> None:
        manifests = available_module_manifests()

        plan = plan_desired_enabled_modules(("campaigns",), manifests)

        self.assertEqual(("access", "admin", "campaigns"), plan.enabled_modules)
        self.assertEqual(("access",), plan.added_dependencies)

    def test_module_install_plan_is_saved_alongside_desired_state(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-install-plan-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files @ git+ssh://git.example.invalid/GovOPlaN/govoplan-files.git@v0.1.4",
                "webui_package": "@govoplan/files-webui",
                "webui_ref": "git+ssh://git.example.invalid/GovOPlaN/govoplan-files.git#v0.1.4",
            }])
            save_desired_enabled_modules(session, ("tenancy", "access", "files"))
            session.commit()

        self.assertEqual(("files",), tuple(item.module_id for item in plan.items))
        with database.session() as session:
            saved = saved_module_install_plan(session)

        self.assertEqual(("files",), tuple(item.module_id for item in saved.items))
        self.assertEqual("manual", saved.items[0].source)
        self.assertEqual((
            "python -m pip install 'govoplan-files @ git+ssh://git.example.invalid/GovOPlaN/govoplan-files.git@v0.1.4'",
            "npm pkg set 'dependencies.@govoplan/files-webui=git+ssh://git.example.invalid/GovOPlaN/govoplan-files.git#v0.1.4' && npm install",
        ), module_install_plan_commands(saved))

    def test_module_install_plan_preserves_catalog_source(self) -> None:
        item = normalize_module_install_plan_item({
            "module_id": "files",
            "action": "install",
            "source": "catalog",
            "catalog": {
                "channel": "stable",
                "sequence": 8,
                "key_id": "release",
            },
            "python_package": "govoplan-files",
            "python_ref": "govoplan-files==0.1.4",
        })

        self.assertEqual("catalog", item.source)
        self.assertEqual("catalog", item.as_dict()["source"])
        self.assertEqual("stable", item.catalog["channel"] if item.catalog else None)
        self.assertEqual(8, item.as_dict()["catalog"]["sequence"])

    def test_module_install_plan_update_uses_package_install_commands(self) -> None:
        item = normalize_module_install_plan_item({
            "module_id": "files",
            "action": "update",
            "python_package": "govoplan-files",
            "python_ref": "govoplan-files==0.2.0",
            "webui_package": "@govoplan/files-webui",
            "webui_ref": "git+ssh://git.example.invalid/GovOPlaN/govoplan-files.git#v0.2.0",
            "data_safety_acknowledged": True,
        })
        plan = ModuleInstallPlan(items=(item,))

        self.assertEqual("update", item.action)
        self.assertTrue(item.data_safety_acknowledged)
        self.assertTrue(item.as_dict()["data_safety_acknowledged"])
        self.assertEqual((
            "python -m pip install govoplan-files==0.2.0",
            "npm pkg set 'dependencies.@govoplan/files-webui=git+ssh://git.example.invalid/GovOPlaN/govoplan-files.git#v0.2.0' && npm install",
        ), module_install_plan_commands(plan))

    def test_admin_catalog_install_plan_records_catalog_provenance(self) -> None:
        app, _settings_obj = self._app_for_modules(("tenancy", "admin"))
        database = get_database()
        create_scope_tables(database.engine)
        Base.metadata.create_all(bind=database.engine)
        with database.session() as session:
            bootstrap_dev_data(session, user_password="test-admin")

        root = Path(tempfile.mkdtemp(prefix="govoplan-admin-module-catalog-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 11,
            "generated_at": "2026-07-10T00:00:00Z",
            "modules": [{
                "module_id": "files",
                "name": "Files",
                "version": "0.1.6",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.6",
                "webui_package": "@govoplan/files-webui",
                "webui_ref": "git+ssh://git.example.invalid/GovOPlaN/govoplan-files.git#v0.1.6",
            }],
        }), encoding="utf-8")

        with TestClient(app) as client:
            login = client.post(
                "/api/v1/auth/login",
                json={"email": "admin@example.local", "password": "test-admin"},
            )
            self.assertEqual(200, login.status_code, login.text)
            headers = {"Authorization": f"Bearer {login.json()['access_token']}"}
            with patch.dict(os.environ, {
                "GOVOPLAN_MODULE_PACKAGE_CATALOG": str(catalog_path),
                "GOVOPLAN_MODULE_PACKAGE_CATALOG_REQUIRE_SIGNATURE": "false",
                "GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS": "",
            }):
                planned = client.post("/api/v1/admin/system/modules/install-plan/catalog/files", headers=headers)

        self.assertEqual(200, planned.status_code, planned.text)
        item = planned.json()["items"][0]
        self.assertEqual("files", item["module_id"])
        self.assertEqual("catalog", item["source"])
        self.assertEqual("stable", item["catalog"]["channel"])
        self.assertEqual(11, item["catalog"]["sequence"])
        self.assertEqual(str(catalog_path), item["catalog"]["source"])

    def test_admin_catalog_install_plan_marks_installed_module_as_update(self) -> None:
        app, _settings_obj = self._app_for_modules(("tenancy", "admin", "files"))
        database = get_database()
        create_scope_tables(database.engine)
        Base.metadata.create_all(bind=database.engine)
        with database.session() as session:
            bootstrap_dev_data(session, user_password="test-admin")

        root = Path(tempfile.mkdtemp(prefix="govoplan-admin-module-catalog-update-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 12,
            "generated_at": "2026-07-10T00:00:00Z",
            "modules": [{
                "module_id": "files",
                "name": "Files",
                "version": "0.2.0",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.2.0",
            }],
        }), encoding="utf-8")

        with TestClient(app) as client:
            login = client.post(
                "/api/v1/auth/login",
                json={"email": "admin@example.local", "password": "test-admin"},
            )
            self.assertEqual(200, login.status_code, login.text)
            headers = {"Authorization": f"Bearer {login.json()['access_token']}"}
            with patch.dict(os.environ, {
                "GOVOPLAN_MODULE_PACKAGE_CATALOG": str(catalog_path),
                "GOVOPLAN_MODULE_PACKAGE_CATALOG_REQUIRE_SIGNATURE": "false",
                "GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS": "",
            }):
                planned = client.post("/api/v1/admin/system/modules/install-plan/catalog/files", headers=headers)

        self.assertEqual(200, planned.status_code, planned.text)
        item = planned.json()["items"][0]
        self.assertEqual("files", item["module_id"])
        self.assertEqual("update", item["action"])
        self.assertEqual("catalog", item["source"])

    def test_module_install_plan_rejects_local_dependency_refs(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-install-plan-local-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            with self.assertRaises(ModuleManagementError):
                save_module_install_plan(session, [{
                    "module_id": "files",
                    "action": "install",
                    "python_package": "govoplan-files",
                    "python_ref": "file:../govoplan-files",
                }])

    def test_module_install_plan_allows_uninstall_webui_package_without_ref(self) -> None:
        item = normalize_module_install_plan_item({
            "module_id": "files",
            "action": "uninstall",
            "python_package": "govoplan-files",
            "webui_package": "@govoplan/files-webui",
        })
        plan = ModuleInstallPlan(items=(item,))

        self.assertIsNone(item.webui_ref)
        self.assertEqual((
            "python -m pip uninstall -y govoplan-files",
            "npm pkg delete dependencies.@govoplan/files-webui && npm install",
        ), module_install_plan_commands(plan))

    def test_desired_modules_after_package_plan_applies_default_install_uninstall_state(self) -> None:
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="mail",
                action="install",
                python_package="govoplan-mail",
                python_ref="govoplan-mail==0.1.4",
            ),
            ModuleInstallPlanItem(
                module_id="files",
                action="uninstall",
                python_package="govoplan-files",
            ),
        ))

        self.assertEqual(
            ("tenancy", "access", "mail"),
            desired_modules_after_package_plan(("tenancy", "access", "files"), plan),
        )

    def test_maintenance_mode_round_trips_through_system_settings(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-maintenance-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            saved = save_maintenance_mode(session, MaintenanceMode(enabled=True, message="Tagged module install"))
            session.commit()

        self.assertTrue(saved.enabled)
        with database.session() as session:
            mode = saved_maintenance_mode(session)

        self.assertTrue(mode.enabled)
        self.assertEqual("Tagged module install", mode.message)
        self.assertEqual("system:maintenance:access", MAINTENANCE_ACCESS_SCOPE)

    def test_module_installer_preflight_blocks_active_module_uninstall(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-guard-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "uninstall",
                "python_package": "govoplan-files",
            }])
            session.commit()

        preflight = module_install_preflight(
            plan=plan,
            available=available_module_manifests(),
            current_enabled=("tenancy", "access", "files"),
            desired_enabled=("tenancy", "access", "files"),
            maintenance_mode=True,
        )

        self.assertFalse(preflight.allowed)
        self.assertIn("module_active", {issue.code for issue in preflight.issues})
        self.assertIn("module_desired", {issue.code for issue in preflight.issues})

    def test_module_installer_blocks_node_local_package_mutation_in_shared_profile(
        self,
    ) -> None:
        plan = ModuleInstallPlan(
            items=(
                ModuleInstallPlanItem(
                    module_id="files",
                    action="update",
                    source="manual",
                    python_package="govoplan-files",
                    python_ref="govoplan-files==1.2.3",
                ),
            )
        )
        with patch.dict(os.environ, {"GOVOPLAN_STATE_PROFILE": "shared"}):
            preflight = module_install_preflight(
                plan=plan,
                available={
                    "files": ModuleManifest(
                        id="files",
                        name="Files",
                        version="1.2.2",
                    )
                },
                current_enabled=("files",),
                desired_enabled=("files",),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        self.assertIn(
            "immutable_cluster_release_required",
            {issue.code for issue in preflight.issues},
        )

    def test_module_installer_preflight_blocks_incompatible_manifest(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-compat-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
            }])
            session.commit()

        available = dict(available_module_manifests())
        available["files"] = replace(
            available["files"],
            compatibility=ModuleCompatibility(core_version_min="999.0.0"),
        )

        preflight = module_install_preflight(
            plan=plan,
            available=available,
            current_enabled=("tenancy", "access"),
            desired_enabled=("tenancy", "access"),
            maintenance_mode=True,
        )

        self.assertFalse(preflight.allowed)
        self.assertIn("core_version_too_low", {issue.code for issue in preflight.issues})
        self.assertTrue(preflight.checklist)

    def test_module_installer_preflight_reports_catalog_warnings_before_activation(self) -> None:
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="install",
                python_package="govoplan-files",
                python_ref="govoplan-files==0.1.4",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": ["Catalog providers do not satisfy files.campaign_attachments."],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available={},
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertTrue(preflight.allowed)
        self.assertIn("catalog_warning", {issue.code for issue in preflight.issues})
        self.assertTrue(any("files.campaign_attachments" in issue.message for issue in preflight.issues))

    def test_module_installer_preflight_blocks_invalid_catalog_sourced_install(self) -> None:
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="install",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==0.1.4",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": False,
                "error": "Module package catalog must be signed by a trusted key.",
                "warnings": [],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available={},
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("catalog_validation_failed", blockers)

    def test_module_installer_preflight_warns_for_invalid_catalog_during_manual_install(self) -> None:
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="install",
                python_package="govoplan-files",
                python_ref="govoplan-files==0.1.4",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": False,
                "error": "Module package catalog must be signed by a trusted key.",
                "warnings": [],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available={},
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertTrue(preflight.allowed)
        warnings = {issue.code for issue in preflight.issues if issue.severity == "warning"}
        self.assertIn("catalog_validation_failed", warnings)

    def test_module_installer_preflight_blocks_unsatisfied_catalog_interface_for_selected_install(self) -> None:
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="campaigns",
                action="install",
                source="catalog",
                python_package="govoplan-campaign",
                python_ref="govoplan-campaign==0.1.4",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "campaigns",
                    "requires_interfaces": [{
                        "name": "files.campaign_attachments",
                        "version_min": "1.0.0",
                        "version_max_exclusive": "2.0.0",
                    }],
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available={},
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("catalog_required_interface_missing", blockers)

    def test_module_installer_preflight_requires_companion_catalog_provider_update(self) -> None:
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                provides_interfaces=(ModuleInterfaceProvider(name="files.attachments", version="1.0.0"),),
            ),
            "campaigns": ModuleManifest(id="campaigns", name="Campaigns", version="1.0.0"),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="campaigns",
                action="update",
                source="catalog",
                python_package="govoplan-campaign",
                python_ref="govoplan-campaign==2.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [
                    {
                        "module_id": "files",
                        "provides_interfaces": [{"name": "files.attachments", "version": "2.0.0"}],
                    },
                    {
                        "module_id": "campaigns",
                        "requires_interfaces": [{
                            "name": "files.attachments",
                            "version_min": "2.0.0",
                            "version_max_exclusive": "3.0.0",
                        }],
                    },
                ],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        companion_issues = [issue for issue in preflight.issues if issue.code == "companion_update_required"]
        self.assertTrue(companion_issues)
        self.assertTrue(any("files" in issue.message for issue in companion_issues))

    def test_module_installer_suggests_catalog_companion_provider_updates(self) -> None:
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                provides_interfaces=(ModuleInterfaceProvider(name="files.attachments", version="1.0.0"),),
            ),
            "campaigns": ModuleManifest(id="campaigns", name="Campaigns", version="1.0.0"),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="campaigns",
                action="update",
                source="catalog",
                python_package="govoplan-campaign",
                python_ref="govoplan-campaign==2.0.0",
            ),
        ))
        validation = {
            "configured": True,
            "valid": True,
            "warnings": [],
            "modules": [
                {
                    "module_id": "files",
                    "provides_interfaces": [{"name": "files.attachments", "version": "2.0.0"}],
                },
                {
                    "module_id": "campaigns",
                    "requires_interfaces": [{
                        "name": "files.attachments",
                        "version_min": "2.0.0",
                        "version_max_exclusive": "3.0.0",
                    }],
                },
            ],
        }

        companions = module_install_catalog_companion_module_ids(plan, available, validation=validation)

        self.assertEqual(("files",), companions)

    def test_module_installer_suggests_catalog_companion_consumer_updates(self) -> None:
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                provides_interfaces=(ModuleInterfaceProvider(name="files.attachments", version="1.0.0"),),
            ),
            "campaigns": ModuleManifest(
                id="campaigns",
                name="Campaigns",
                version="1.0.0",
                requires_interfaces=(
                    ModuleInterfaceRequirement(
                        name="files.attachments",
                        version_min="1.0.0",
                        version_max_exclusive="2.0.0",
                    ),
                ),
            ),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))
        validation = {
            "configured": True,
            "valid": True,
            "warnings": [],
            "modules": [
                {
                    "module_id": "files",
                    "provides_interfaces": [{"name": "files.attachments", "version": "2.0.0"}],
                },
                {
                    "module_id": "campaigns",
                    "requires_interfaces": [{
                        "name": "files.attachments",
                        "version_min": "2.0.0",
                        "version_max_exclusive": "3.0.0",
                    }],
                },
            ],
        }

        companions = module_install_catalog_companion_module_ids(plan, available, validation=validation)

        self.assertEqual(("campaigns",), companions)

    def test_module_installer_preflight_blocks_unacknowledged_forward_only_catalog_update(self) -> None:
        available = {"files": ModuleManifest(id="files", name="Files", version="1.0.0")}
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "files",
                    "version": "2.0.0",
                    "migration_safety": "forward_only",
                    "migration_notes": "Database rollback requires restoring the pre-update snapshot.",
                    "recovery_tested": True,
                    "recovery_notes": "Snapshot restore was rehearsed on the staging dataset.",
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("forward_only_migration_acknowledgement_required", blockers)

    def test_module_installer_preflight_requires_recovery_validation_for_forward_only_update(self) -> None:
        available = {"files": ModuleManifest(id="files", name="Files", version="1.0.0")}
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
                data_safety_acknowledged=True,
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "files",
                    "version": "2.0.0",
                    "migration_safety": "forward_only",
                    "migration_notes": "Database rollback requires restoring the pre-update snapshot.",
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("recovery_validation_required", blockers)
        self.assertIn("recovery_notes_required", blockers)

    def test_module_installer_preflight_allows_acknowledged_forward_only_catalog_update(self) -> None:
        available = {"files": ModuleManifest(id="files", name="Files", version="1.0.0")}
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
                data_safety_acknowledged=True,
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "files",
                    "version": "2.0.0",
                    "migration_safety": "forward_only",
                    "migration_notes": "Database rollback requires restoring the pre-update snapshot.",
                    "recovery_tested": True,
                    "recovery_notes": "Snapshot restore was rehearsed on the staging dataset.",
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertTrue(preflight.allowed)
        self.assertEqual(("files",), tuple(item.module_id for item in preflight.target_plan))
        self.assertEqual("1.0.0", preflight.target_plan[0].current_version)
        self.assertEqual("2.0.0", preflight.target_plan[0].target_version)
        self.assertEqual("forward_only", preflight.target_plan[0].migration_safety)
        self.assertTrue(preflight.target_plan[0].recovery_tested)
        self.assertEqual("Snapshot restore was rehearsed on the staging dataset.", preflight.target_plan[0].recovery_notes)
        self.assertTrue(preflight.target_plan[0].data_safety_acknowledged)
        self.assertEqual("2.0.0", preflight.as_dict()["target_plan"][0]["target_version"])
        warnings = {issue.code for issue in preflight.issues if issue.severity == "warning"}
        self.assertIn("forward_only_migration_acknowledged", warnings)

    def test_module_installer_preflight_blocks_catalog_downgrade_and_noop_updates_by_default(self) -> None:
        available = {
            "files": ModuleManifest(id="files", name="Files", version="1.0.0"),
            "mail": ModuleManifest(id="mail", name="Mail", version="1.0.0"),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==0.9.0",
            ),
            ModuleInstallPlanItem(
                module_id="mail",
                action="update",
                source="catalog",
                python_package="govoplan-mail",
                python_ref="govoplan-mail==1.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [
                    {"module_id": "files", "version": "0.9.0"},
                    {"module_id": "mail", "version": "1.0.0"},
                ],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("catalog_downgrade_blocked", blockers)
        self.assertIn("catalog_noop_update_blocked", blockers)

    def test_module_installer_preflight_allows_reviewed_downgrade_and_noop_updates(self) -> None:
        available = {
            "files": ModuleManifest(id="files", name="Files", version="1.0.0"),
            "mail": ModuleManifest(id="mail", name="Mail", version="1.0.0"),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==0.9.0",
            ),
            ModuleInstallPlanItem(
                module_id="mail",
                action="update",
                source="catalog",
                python_package="govoplan-mail",
                python_ref="govoplan-mail==1.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [
                    {"module_id": "files", "version": "0.9.0", "allow_downgrade": True},
                    {"module_id": "mail", "version": "1.0.0", "allow_same_version": True},
                ],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertTrue(preflight.allowed)
        blocker_codes = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertNotIn("catalog_downgrade_blocked", blocker_codes)
        self.assertNotIn("catalog_noop_update_blocked", blocker_codes)

    def test_module_installer_preflight_requires_catalog_update_bridge_when_current_version_is_outside_window(self) -> None:
        available = {"files": ModuleManifest(id="files", name="Files", version="1.0.0")}
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==3.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "files",
                    "version": "3.0.0",
                    "current_version_min": "2.0.0",
                    "current_version_max_exclusive": "3.0.0",
                    "bridge_notes": "Install 2.x bridge before moving to 3.x.",
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("catalog_update_bridge_required", blockers)
        self.assertEqual("2.0.0", preflight.target_plan[0].current_version_min)
        self.assertEqual("3.0.0", preflight.target_plan[0].current_version_max_exclusive)

    def test_module_installer_preflight_reports_catalog_bridge_release_metadata(self) -> None:
        available = {"files": ModuleManifest(id="files", name="Files", version="1.0.0")}
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "files",
                    "version": "2.0.0",
                    "bridge_release": True,
                    "bridge_notes": "Keeps both 1.x and 2.x attachment interfaces available.",
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertTrue(preflight.allowed)
        self.assertIn("catalog_bridge_release", {issue.code for issue in preflight.issues if issue.severity == "info"})
        self.assertTrue(preflight.target_plan[0].bridge_release)
        self.assertEqual("Keeps both 1.x and 2.x attachment interfaces available.", preflight.target_plan[0].bridge_notes)

    def test_module_installer_preflight_requires_destructive_catalog_update_plan(self) -> None:
        available = {"files": ModuleManifest(id="files", name="Files", version="1.0.0")}
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
                data_safety_acknowledged=True,
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "files",
                    "migration_safety": "destructive",
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("destructive_migration_plan_required", blockers)

    def test_module_installer_preflight_allows_acknowledged_destructive_catalog_update_with_plan(self) -> None:
        available = {"files": ModuleManifest(id="files", name="Files", version="1.0.0")}
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
                data_safety_acknowledged=True,
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [{
                    "module_id": "files",
                    "migration_safety": "destructive",
                    "migration_notes": "Drops obsolete cache tables after exporting the retained documents.",
                    "recovery_tested": True,
                    "recovery_notes": "Restore and forward-recovery path verified on staging.",
                }],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertTrue(preflight.allowed)
        warnings = {issue.code for issue in preflight.issues if issue.severity == "warning"}
        self.assertIn("destructive_migration_acknowledged", warnings)

    def test_module_installer_preflight_allows_companion_updates_in_same_target_set(self) -> None:
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                provides_interfaces=(ModuleInterfaceProvider(name="files.attachments", version="1.0.0"),),
            ),
            "campaigns": ModuleManifest(id="campaigns", name="Campaigns", version="1.0.0"),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
            ModuleInstallPlanItem(
                module_id="campaigns",
                action="update",
                source="catalog",
                python_package="govoplan-campaign",
                python_ref="govoplan-campaign==2.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [
                    {
                        "module_id": "files",
                        "provides_interfaces": [{"name": "files.attachments", "version": "2.0.0"}],
                    },
                    {
                        "module_id": "campaigns",
                        "requires_interfaces": [{
                            "name": "files.attachments",
                            "version_min": "2.0.0",
                            "version_max_exclusive": "3.0.0",
                        }],
                    },
                ],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertTrue(preflight.allowed)
        self.assertNotIn("companion_update_required", {issue.code for issue in preflight.issues})

    def test_module_installer_preflight_orders_migration_plan_from_interface_requirements(self) -> None:
        metadata = MetaData()
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                provides_interfaces=(ModuleInterfaceProvider(name="files.attachments", version="1.0.0"),),
                migration_spec=MigrationSpec(module_id="files", metadata=metadata),
            ),
            "campaigns": ModuleManifest(
                id="campaigns",
                name="Campaigns",
                version="1.0.0",
                requires_interfaces=(
                    ModuleInterfaceRequirement(
                        name="files.attachments",
                        version_min="1.0.0",
                        version_max_exclusive="2.0.0",
                    ),
                ),
                migration_spec=MigrationSpec(module_id="campaigns", metadata=metadata),
            ),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="campaigns",
                action="update",
                python_package="govoplan-campaign",
                python_ref="govoplan-campaign==2.0.0",
            ),
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))

        preflight = module_install_preflight(
            plan=plan,
            available=available,
            current_enabled=("files", "campaigns"),
            desired_enabled=("files", "campaigns"),
            maintenance_mode=True,
        )

        self.assertTrue(preflight.allowed)
        self.assertEqual(("files", "campaigns"), tuple(step.module_id for step in preflight.migration_plan.steps))
        self.assertEqual(("files", "campaigns"), tuple(preflight.migration_plan.enabled_modules))
        self.assertEqual(["files", "campaigns"], preflight.as_dict()["migration_plan"]["enabled_modules"])

    def test_module_installer_preflight_blocks_migration_graph_cycles(self) -> None:
        metadata = MetaData()
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                migration_spec=MigrationSpec(module_id="files", metadata=metadata, migration_after=("campaigns",)),
            ),
            "campaigns": ModuleManifest(
                id="campaigns",
                name="Campaigns",
                version="1.0.0",
                migration_spec=MigrationSpec(module_id="campaigns", metadata=metadata, migration_after=("files",)),
            ),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="campaigns",
                action="update",
                python_package="govoplan-campaign",
                python_ref="govoplan-campaign==2.0.0",
            ),
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))

        preflight = module_install_preflight(
            plan=plan,
            available=available,
            current_enabled=("files", "campaigns"),
            desired_enabled=("files", "campaigns"),
            maintenance_mode=True,
        )

        self.assertFalse(preflight.allowed)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("migration_graph_cycle", blockers)

    def test_structured_install_commands_pass_target_modules_and_migration_order(self) -> None:
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="install",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))

        commands = structured_install_commands(
            plan,
            webui_root=None,
            migrate_database_url="sqlite:///example.db",
            migration_enabled_modules=("access", "files"),
            migration_module_order=("files",),
            migration_task_record_path=Path("/tmp/govoplan-migration-tasks.json"),
            verify_modules=True,
        )

        init_db = commands[-1]["argv"]
        self.assertIn("govoplan_core.commands.init_db", init_db)
        self.assertIn("--enabled-module", init_db)
        self.assertIn("access", init_db)
        self.assertIn("files", init_db)
        self.assertIn("--migration-module", init_db)
        self.assertIn("--migration-task-record-output", init_db)
        self.assertEqual("/tmp/govoplan-migration-tasks.json", commands[-1]["migration_task_record_path"])

    def test_module_installer_preflight_reports_and_blocks_unsafe_migration_tasks(self) -> None:
        metadata = MetaData()
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                migration_spec=MigrationSpec(
                    module_id="files",
                    metadata=metadata,
                    migration_tasks=(
                        ModuleMigrationTask(
                            task_id="legacy_path_check",
                            phase="pre_migration_check",
                            summary="Check legacy paths before the schema migration.",
                            executor=lambda _context: ModuleMigrationTaskResult(),
                        ),
                        ModuleMigrationTask(
                            task_id="unsafe_backfill",
                            phase="post_migration_backfill",
                            summary="Backfill legacy file spaces.",
                            safety="forward_only",
                            idempotent=False,
                            executor=lambda _context: ModuleMigrationTaskResult(),
                        ),
                    ),
                ),
            ),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))

        preflight = module_install_preflight(
            plan=plan,
            available=available,
            current_enabled=("files",),
            desired_enabled=("files",),
            maintenance_mode=True,
        )

        self.assertFalse(preflight.allowed)
        task_ids = {task.task_id for task in preflight.migration_plan.tasks}
        self.assertEqual({"legacy_path_check", "unsafe_backfill"}, task_ids)
        blockers = {issue.code for issue in preflight.issues if issue.severity == "blocker"}
        self.assertIn("migration_task_not_idempotent", blockers)
        self.assertIn("migration_task_acknowledgement_required", blockers)

    def test_module_installer_preflight_allows_acknowledged_forward_only_migration_task(self) -> None:
        metadata = MetaData()
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                migration_spec=MigrationSpec(
                    module_id="files",
                    metadata=metadata,
                    migration_tasks=(
                        ModuleMigrationTask(
                            task_id="backfill_spaces",
                            phase="post_migration_backfill",
                            summary="Backfill file spaces.",
                            safety="forward_only",
                            executor=lambda _context: ModuleMigrationTaskResult(status="ok"),
                        ),
                    ),
                ),
            ),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
                data_safety_acknowledged=True,
            ),
        ))

        preflight = module_install_preflight(
            plan=plan,
            available=available,
            current_enabled=("files",),
            desired_enabled=("files",),
            maintenance_mode=True,
        )

        self.assertTrue(preflight.allowed)
        self.assertEqual(("backfill_spaces",), tuple(task.task_id for task in preflight.migration_plan.tasks))
        self.assertIn("tasks", preflight.as_dict()["migration_plan"])

    def test_registered_module_migration_tasks_execute_in_phase_order(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-migration-tasks-", dir=_TEST_ROOT))
        database_url = f"sqlite:///{root / 'tasks.db'}"
        calls: list[tuple[str, str, bool, str | None]] = []

        def executor(context: ModuleMigrationTaskContext) -> ModuleMigrationTaskResult:
            calls.append((context.module_id, context.task_id, context.session is not None, context.target_version))
            return ModuleMigrationTaskResult(status="warning", message=f"ran {context.task_id}", details={"phase": context.phase})

        manifest = ModuleManifest(
            id="taskmod",
            name="Task Module",
            version="2.0.0",
            migration_spec=MigrationSpec(
                module_id="taskmod",
                metadata=MetaData(),
                migration_tasks=(
                    ModuleMigrationTask(
                        task_id="verify",
                        phase="post_migration_verify",
                        summary="Verify migrated state.",
                        executor=executor,
                    ),
                    ModuleMigrationTask(
                        task_id="check",
                        phase="pre_migration_check",
                        summary="Check source state.",
                        executor=executor,
                    ),
                ),
            ),
        )

        records = run_registered_module_migration_tasks(
            database_url=database_url,
            enabled_modules=("taskmod",),
            migration_module_order=("taskmod",),
            phases=("pre_migration_check", "post_migration_verify"),
            manifest_factories=(lambda: manifest,),
        )

        self.assertEqual(
            [("taskmod", "check", True, "2.0.0"), ("taskmod", "verify", True, "2.0.0")],
            calls,
        )
        self.assertEqual(["check", "verify"], [record["task_id"] for record in records])
        self.assertEqual(["warning", "warning"], [record["status"] for record in records])

    def test_registered_module_migration_tasks_dispose_replaced_database(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-migration-dispose-", dir=_TEST_ROOT))
        previous = configure_database(f"sqlite:///{root / 'previous.db'}")
        previous_pool = previous.engine.pool
        manifest = ModuleManifest(id="taskmod", name="Task Module", version="1.0.0")

        try:
            run_registered_module_migration_tasks(
                database_url=f"sqlite:///{root / 'target.db'}",
                enabled_modules=("taskmod",),
                phases=(),
                manifest_factories=(lambda: manifest,),
            )

            self.assertIsNot(previous.engine.pool, previous_pool)
            self.assertEqual(f"sqlite:///{root / 'target.db'}", get_database().database_url)
        finally:
            reset_database(dispose=True)

    def test_module_installer_preflight_blocks_provider_update_that_breaks_installed_consumer(self) -> None:
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="1.0.0",
                provides_interfaces=(ModuleInterfaceProvider(name="files.attachments", version="1.0.0"),),
            ),
            "campaigns": ModuleManifest(
                id="campaigns",
                name="Campaigns",
                version="1.0.0",
                requires_interfaces=(
                    ModuleInterfaceRequirement(
                        name="files.attachments",
                        version_min="1.0.0",
                        version_max_exclusive="2.0.0",
                    ),
                ),
            ),
        }
        plan = ModuleInstallPlan(items=(
            ModuleInstallPlanItem(
                module_id="files",
                action="update",
                source="catalog",
                python_package="govoplan-files",
                python_ref="govoplan-files==2.0.0",
            ),
        ))

        with patch(
            "govoplan_core.core.module_package_catalog.validate_module_package_catalog",
            return_value={
                "configured": True,
                "valid": True,
                "warnings": [],
                "modules": [
                    {
                        "module_id": "files",
                        "provides_interfaces": [{"name": "files.attachments", "version": "2.0.0"}],
                    },
                    {
                        "module_id": "campaigns",
                        "requires_interfaces": [{
                            "name": "files.attachments",
                            "version_min": "2.0.0",
                            "version_max_exclusive": "3.0.0",
                        }],
                    },
                ],
            },
        ):
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=(),
                desired_enabled=(),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        companion_issues = [issue for issue in preflight.issues if issue.code == "companion_update_required"]
        self.assertTrue(companion_issues)
        self.assertTrue(any("campaigns" in issue.message for issue in companion_issues))

    def test_module_installer_reports_interface_contract_issues(self) -> None:
        available = {
            "files": ModuleManifest(
                id="files",
                name="Files",
                version="2.0.0",
                provides_interfaces=(ModuleInterfaceProvider(name="files.spaces", version="2.0.0"),),
            ),
            "campaigns": ModuleManifest(
                id="campaigns",
                name="Campaigns",
                version="1.1.0",
                requires_interfaces=(
                    ModuleInterfaceRequirement(
                        name="files.spaces",
                        version_min="1.0.0",
                        version_max_exclusive="2.0.0",
                    ),
                    ModuleInterfaceRequirement(name="mail.delivery", version_min="1.0.0"),
                ),
            ),
        }

        issues = module_manifest_compatibility_issues(available)

        self.assertIn("interface_version_mismatch", {issue.code for issue in issues})
        self.assertIn("required_interface_missing", {issue.code for issue in issues})

    def test_module_installer_preflight_requires_python_package_for_install_rollback(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-install-rollback-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_ref": "govoplan-files==0.1.4",
            }])
            session.commit()

        preflight = module_install_preflight(
            plan=plan,
            available=available_module_manifests(),
            current_enabled=("tenancy", "access"),
            desired_enabled=("tenancy", "access"),
            maintenance_mode=True,
        )

        self.assertFalse(preflight.allowed)
        self.assertIn("python_package_required", {issue.code for issue in preflight.issues})

    def test_module_installer_preflight_uses_module_uninstall_guards(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-module-guard-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "uninstall",
                "python_package": "govoplan-files",
            }])
            session.commit()

            available = dict(available_module_manifests())
            migration = available["files"].migration_spec
            self.assertIsNotNone(migration)
            assert migration is not None
            available["files"] = replace(
                available["files"],
                migration_spec=replace(migration, retirement_supported=False, retirement_provider=None),
                uninstall_guard_providers=(lambda _session, _module_id: (
                    ModuleUninstallGuardResult("blocker", "stored_data_present", "Stored files remain."),
                ),),
            )

            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                maintenance_mode=True,
                session=session,
            )

        self.assertFalse(preflight.allowed)
        self.assertIn("stored_data_present", {issue.code for issue in preflight.issues})
        self.assertIn("migration_retirement_not_requested", {issue.code for issue in preflight.issues})

    def test_module_installer_allows_non_destructive_uninstall_for_migration_owner(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-nondestructive-uninstall-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "uninstall",
                "python_package": "govoplan-files",
            }])
            session.commit()

            available = dict(available_module_manifests())
            migration = available["files"].migration_spec
            self.assertIsNotNone(migration)
            assert migration is not None
            available["files"] = replace(
                available["files"],
                migration_spec=replace(migration, retirement_supported=False, retirement_provider=None),
                uninstall_guard_providers=(),
            )
            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                maintenance_mode=True,
                session=session,
            )

        self.assertTrue(preflight.allowed)
        issues = {issue.code: issue for issue in preflight.issues}
        self.assertEqual("warning", issues["migration_retirement_not_requested"].severity)

    def test_module_installer_preflight_uses_migration_retirement_provider(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-retirement-provider-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "uninstall",
                "python_package": "govoplan-files",
            }])
            session.commit()

            available = dict(available_module_manifests())
            migration = available["files"].migration_spec
            self.assertIsNotNone(migration)
            assert migration is not None
            available["files"] = replace(
                available["files"],
                migration_spec=replace(
                    migration,
                    retirement_supported=True,
                    retirement_provider=lambda _session, _module_id: MigrationRetirementPlan(
                        supported=True,
                        summary="Retirement provider can preserve historical migration state.",
                        warnings=("Operator must verify archived file storage.",),
                    ),
                ),
            )

            preflight = module_install_preflight(
                plan=plan,
                available=available,
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                maintenance_mode=True,
                session=session,
            )

        codes = {issue.code for issue in preflight.issues}
        self.assertTrue(preflight.allowed)
        self.assertNotIn("migration_retirement_not_requested", codes)
        self.assertIn("migration_retirement_supported", codes)
        self.assertIn("migration_retirement_warning", codes)

    def test_module_installer_destroy_data_retirement_drops_module_tables(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-destroy-data-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        metadata = MetaData()
        table = Table("retirement_example", metadata, Column("id", Integer, primary_key=True))
        metadata.create_all(bind=database.engine)
        example_model = type("ExampleModel", (), {"__table__": table})
        manifest = ModuleManifest(
            id="example",
            name="Example",
            version="0.1.0",
            migration_spec=MigrationSpec(
                module_id="example",
                metadata=metadata,
                retirement_supported=True,
                retirement_provider=drop_table_retirement_provider(example_model, label="Example"),
            ),
        )

        def fake_run(*_args, **_kwargs):
            return SimpleNamespace(returncode=0, stdout="", stderr="")

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "example",
                "action": "uninstall",
                "python_package": "govoplan-example",
                "destroy_data": True,
            }])
            session.commit()

            preflight = module_install_preflight(
                plan=plan,
                available={"example": manifest},
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                maintenance_mode=True,
                session=session,
            )
            self.assertTrue(preflight.allowed)
            self.assertIn("migration_retirement_destroy_planned", {issue.code for issue in preflight.issues})

            with patch("govoplan_core.core.module_installer.subprocess.run", side_effect=fake_run):
                result = run_module_install_plan(
                    session=session,
                    plan=plan,
                    available={"example": manifest},
                    current_enabled=("tenancy", "access"),
                    desired_enabled=("tenancy", "access"),
                    database_url=settings.database_url,
                    runtime_dir=root / "installer",
                )

        self.assertEqual("applied", result.status)
        self.assertFalse(inspect(database.engine).has_table("retirement_example"))
        record = json.loads(result.record_path.read_text(encoding="utf-8"))
        self.assertEqual("example", record["retirements"][0]["module_id"])
        self.assertIn("database_backup", record["snapshot"])

    def test_drop_table_retirement_is_atomic_with_session_writes(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-retirement-transaction-", dir=_TEST_ROOT))
        engine = create_engine(
            f"sqlite:///{root / 'retirement.db'}",
            connect_args={"timeout": 0.1},
        )
        metadata = MetaData()
        audit_table = Table("retirement_audit", metadata, Column("id", Integer, primary_key=True))
        owned_table = Table("retirement_owned", metadata, Column("id", Integer, primary_key=True))
        metadata.create_all(bind=engine)
        owned_model = type("OwnedModel", (), {"__table__": owned_table})

        with Session(engine) as session:
            session.execute(insert(audit_table).values(id=1))
            plan = drop_table_retirement_provider(owned_model, label="Owned")(session, "owned")
            self.assertIsNotNone(plan.destroy_data_executor)
            plan.destroy_data_executor(session, "owned")  # type: ignore[misc]
            self.assertFalse(inspect(session.connection()).has_table("retirement_owned"))
            session.rollback()

        self.assertTrue(inspect(engine).has_table("retirement_owned"))
        with engine.connect() as connection:
            self.assertEqual([], connection.execute(audit_table.select()).all())
        engine.dispose()

    def test_module_installer_dry_run_writes_run_record_without_applying(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-dry-run-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
            }])
            session.commit()

            result = run_module_install_plan(
                session=session,
                plan=plan,
                available=available_module_manifests(),
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                database_url=settings.database_url,
                runtime_dir=root / "installer",
                migrate_database=True,
                dry_run=True,
            )

        self.assertEqual("dry-run", result.status)
        self.assertTrue(result.record_path.exists())
        record_text = result.record_path.read_text(encoding="utf-8")
        self.assertIn("pip", record_text)
        self.assertIn("database_backup", record_text)
        self.assertIn("govoplan_core.commands.module_verify", record_text)
        self.assertIn("govoplan_core.commands.init_db", record_text)
        runs = list_module_installer_runs(runtime_dir=root / "installer")
        self.assertEqual((result.run_id,), tuple(item["run_id"] for item in runs))
        self.assertFalse(module_installer_lock_status(runtime_dir=root / "installer")["locked"])
        record = read_module_installer_run(runtime_dir=root / "installer", run_id=result.run_id)
        self.assertEqual(result.run_id, record["run_id"])

    def test_module_installer_records_verified_local_artifact_integrity(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-artifact-integrity-", dir=_TEST_ROOT))
        artifact_path = root / "govoplan-files-0.1.4-py3-none-any.whl"
        artifact_path.write_bytes(b"package artifact")
        digest = hashlib.sha256(artifact_path.read_bytes()).hexdigest()
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
                "artifact_integrity": {
                    "python": {
                        "ref": "govoplan-files==0.1.4",
                        "path": str(artifact_path),
                        "sha256": digest,
                        "sbom_url": "https://govoplan.example/sbom/files-0.1.4.spdx.json",
                        "provenance_url": "https://govoplan.example/provenance/files-0.1.4.intoto.jsonl",
                    },
                },
            }])
            session.commit()

            preflight = module_install_preflight(
                plan=plan,
                available=available_module_manifests(),
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                maintenance_mode=True,
                runtime_dir=root / "installer",
            )
            self.assertTrue(preflight.allowed)
            self.assertEqual(digest, preflight.as_dict()["artifact_integrity"][0]["actual_sha256"])

            result = run_module_install_plan(
                session=session,
                plan=plan,
                available=available_module_manifests(),
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                database_url=settings.database_url,
                runtime_dir=root / "installer",
                dry_run=True,
            )

        record = read_module_installer_run(runtime_dir=root / "installer", run_id=result.run_id)
        verified = record["preflight"]["artifact_integrity"][0]
        self.assertTrue(verified["verified"])
        self.assertEqual(str(artifact_path), verified["artifact_path"])

    def test_module_installer_blocks_missing_integrity_in_production_mode(self) -> None:
        plan = ModuleInstallPlan(items=(ModuleInstallPlanItem(
            module_id="files",
            action="install",
            python_package="govoplan-files",
            python_ref="govoplan-files==0.1.4",
        ),))

        with patch.dict(os.environ, {"GOVOPLAN_MODULE_INSTALLER_REQUIRE_ARTIFACT_INTEGRITY": "true"}):
            preflight = module_install_preflight(
                plan=plan,
                available=available_module_manifests(),
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                maintenance_mode=True,
            )

        self.assertFalse(preflight.allowed)
        self.assertIn("artifact_integrity_required", {issue.code for issue in preflight.issues})

    def test_supervised_module_install_rollback_restores_desired_modules(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-desired-rollback-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])

        def fake_run(*_args, **kwargs):
            argv = tuple(_args[0]) if _args else ()
            if kwargs.get("shell") or argv == ("restart-fails",):
                return SimpleNamespace(returncode=1, stdout="", stderr="restart failed")
            return SimpleNamespace(returncode=0, stdout="", stderr="")

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "example",
                "action": "install",
                "python_package": "govoplan-example",
                "python_ref": "govoplan-example==0.1.4",
            }])
            save_desired_enabled_modules(session, ("tenancy", "access"))
            session.commit()

            with patch("govoplan_core.core.module_installer.subprocess.run", side_effect=fake_run):
                result = supervise_module_install_plan(
                    session=session,
                    plan=plan,
                    available=available_module_manifests(),
                    current_enabled=("tenancy", "access"),
                    desired_enabled=("tenancy", "access"),
                    database_url=settings.database_url,
                    runtime_dir=root / "installer",
                    restart_command="restart-fails",
                )

            restored_desired = saved_desired_enabled_modules(session, ("tenancy", "access"))
            restored_plan = saved_module_install_plan(session)

        self.assertEqual("rolled-back", result.status)
        self.assertEqual(("tenancy", "access"), restored_desired)
        self.assertEqual(("planned",), tuple(item.status for item in restored_plan.items))

    def test_module_installer_external_database_backup_command_is_recorded(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-external-backup-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        backup_script = (
            "import json, os, pathlib; "
            "pathlib.Path(os.environ['GOVOPLAN_DATABASE_BACKUP_PATH']).write_text('backup', encoding='utf-8'); "
            "pathlib.Path(os.environ['GOVOPLAN_DATABASE_BACKUP_METADATA']).write_text(json.dumps({"
            "'snapshot': 'external', "
            "'pgtools_url': os.environ.get('GOVOPLAN_DATABASE_URL_PGTOOLS')"
            "}), encoding='utf-8'); "
            "print(os.environ.get('GOVOPLAN_DATABASE_URL'))"
        )
        backup_command = f"{sys.executable} -c {shlex.quote(backup_script)}"
        database_url = "postgresql+psycopg://govoplan:db-secret@db.example.invalid/govoplan"

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
            }])
            session.commit()

            result = run_module_install_plan(
                session=session,
                plan=plan,
                available=available_module_manifests(),
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                database_url=database_url,
                runtime_dir=root / "installer",
                migrate_database=True,
                database_backup_command=backup_command,
                database_restore_command="restore-database",
                dry_run=True,
            )

        record_text = result.record_path.read_text(encoding="utf-8")
        self.assertNotIn("db-secret", record_text)
        record = json.loads(record_text)
        database_backup = record["snapshot"]["database_backup"]
        self.assertEqual("external", database_backup["type"])
        self.assertEqual("postgresql+psycopg://govoplan:***@db.example.invalid/govoplan", database_backup["database_url"])
        self.assertEqual("external", database_backup["metadata"]["snapshot"])
        self.assertEqual("postgresql://govoplan:***@db.example.invalid/govoplan", database_backup["metadata"]["pgtools_url"])
        self.assertEqual("postgresql+psycopg://govoplan:***@db.example.invalid/govoplan", database_backup["stdout"].strip())
        secret_path = result.record_path.parent / database_backup["database_url_secret"]
        self.assertEqual(database_url, secret_path.read_text(encoding="utf-8"))
        self.assertEqual(0o700, stat.S_IMODE(result.record_path.parent.stat().st_mode))
        self.assertEqual(0o600, stat.S_IMODE(secret_path.stat().st_mode))
        self.assertEqual("backup", (result.record_path.parent / "database.external.backup").read_text(encoding="utf-8"))

    def test_installer_secret_creation_refuses_an_existing_path(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-secret-existing-", dir=_TEST_ROOT))
        secret_path = root / "database-url.secret"
        secret_path.write_text("existing-value", encoding="utf-8")

        with self.assertRaisesRegex(module_installer_module.ModuleInstallerError, "Could not create private installer secret file"):
            module_installer_module._write_installer_secret(secret_path, "replacement-value")

        self.assertEqual("existing-value", secret_path.read_text(encoding="utf-8"))

    def test_installer_run_directory_remains_usable_with_a_restrictive_umask(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-run-umask-", dir=_TEST_ROOT))
        run_dir = root / "run"
        previous_umask = os.umask(0o777)
        try:
            module_installer_module._create_private_installer_run_dir(run_dir)
        finally:
            os.umask(previous_umask)

        self.assertEqual(0o700, stat.S_IMODE(run_dir.stat().st_mode))

    def test_module_installer_rejects_unsafe_external_database_hook_command(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-unsafe-hook-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        backup_command = f"{sys.executable} -c {shlex.quote('print(1)')} && {sys.executable} -c {shlex.quote('print(2)')}"

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
            }])
            session.commit()

            with self.assertRaisesRegex(module_installer_module.ModuleInstallerError, "Database backup command failed"):
                run_module_install_plan(
                    session=session,
                    plan=plan,
                    available=available_module_manifests(),
                    current_enabled=("tenancy", "access"),
                    desired_enabled=("tenancy", "access"),
                    database_url="postgresql://db.example.invalid/govoplan",
                    runtime_dir=root / "installer",
                    migrate_database=True,
                    database_backup_command=backup_command,
                    database_restore_command="restore-database",
                    dry_run=True,
                )

        rejected = module_installer_module._run_operator_command("DATABASE_URL=postgres://db.example.invalid/govoplan pg_dump")
        self.assertEqual(2, rejected.returncode)
        self.assertIn("Environment assignments", rejected.stderr)

    def test_module_installer_external_database_restore_check_runs_before_migrations(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-restore-check-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        backup_script = (
            "import os, pathlib; "
            "pathlib.Path(os.environ['GOVOPLAN_DATABASE_BACKUP_PATH']).write_text('backup', encoding='utf-8')"
        )
        check_script = (
            "import os, pathlib; "
            "pathlib.Path(os.environ['GOVOPLAN_INSTALLER_RUN_DIR'], 'restore-check.marker').write_text("
            "pathlib.Path(os.environ['GOVOPLAN_DATABASE_BACKUP_PATH']).read_text(encoding='utf-8'), encoding='utf-8')"
        )

        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
            }])
            session.commit()

            result = run_module_install_plan(
                session=session,
                plan=plan,
                available=available_module_manifests(),
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                database_url="postgresql://db.example.invalid/govoplan",
                runtime_dir=root / "installer",
                migrate_database=True,
                database_backup_command=f"{sys.executable} -c {shlex.quote(backup_script)}",
                database_restore_command="restore-database",
                database_restore_check_command=f"{sys.executable} -c {shlex.quote(check_script)}",
                dry_run=True,
            )

        record = json.loads(result.record_path.read_text(encoding="utf-8"))
        database_backup = record["snapshot"]["database_backup"]
        self.assertEqual(0, database_backup["restore_check"]["return_code"])
        self.assertEqual("backup", (result.record_path.parent / "restore-check.marker").read_text(encoding="utf-8"))

    def test_module_installer_external_database_restore_command_runs_on_rollback(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-external-restore-", dir=_TEST_ROOT))
        runtime_dir = root / "installer"
        run_id = "external-restore-test"
        run_dir = runtime_dir / "runs" / run_id
        run_dir.mkdir(parents=True)
        restore_script = (
            "import os, pathlib; "
            "pathlib.Path(os.environ['GOVOPLAN_INSTALLER_RUN_DIR'], 'restore.marker').write_text("
            "os.environ.get('GOVOPLAN_DATABASE_URL', ''), encoding='utf-8')"
        )
        restore_command = f"{sys.executable} -c {shlex.quote(restore_script)}"
        database_url = "postgresql://restore-user:restore-secret@db.example.invalid/govoplan"
        (run_dir / "database.external.backup").write_text("backup", encoding="utf-8")
        (run_dir / "database-url.secret").write_text(database_url, encoding="utf-8")
        (run_dir / "record.json").write_text(json.dumps({
            "run_id": run_id,
            "plan": [],
            "snapshot": {
                "database_backup": {
                    "type": "external",
                    "database_url": "postgresql://restore-user:***@db.example.invalid/govoplan",
                    "database_url_secret": "database-url.secret",
                    "backup_path": "database.external.backup",
                    "metadata_path": "database-backup-metadata.json",
                    "restore_command": restore_command,
                },
            },
        }), encoding="utf-8")

        result = rollback_module_install_run(run_id=run_id, runtime_dir=runtime_dir, npm_bin="npm")

        self.assertEqual("rolled-back", result.status)
        self.assertEqual(database_url, (run_dir / "restore.marker").read_text(encoding="utf-8"))
        record = json.loads((run_dir / "record.json").read_text(encoding="utf-8"))
        self.assertNotIn("restore-secret", json.dumps(record))
        self.assertTrue(record["rollback_database_restore"]["restored"])

    def test_module_installer_request_queue_round_trips(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-request-queue-", dir=_TEST_ROOT))
        runtime_dir = root / "installer"

        with event_context(correlation_id="installer-trace-1"):
            queued = queue_module_installer_request(
                runtime_dir=runtime_dir,
                requested_by="user-1",
                tenant_id="tenant-1",
                options={"migrate_database": True, "health_urls": ["http://127.0.0.1:8000/health"]},
            )
        request_id = str(queued["request_id"])

        self.assertEqual("queued", queued["status"])
        self.assertEqual("tenant-1", queued["tenant_id"])
        self.assertEqual("installer-trace-1", queued["trace"]["correlation_id"])
        self.assertEqual((request_id,), tuple(item["request_id"] for item in list_module_installer_requests(runtime_dir=runtime_dir)))
        claimed = claim_next_module_installer_request(runtime_dir=runtime_dir)
        self.assertIsNotNone(claimed)
        assert claimed is not None
        self.assertEqual(request_id, claimed["request_id"])
        self.assertEqual("running", claimed["status"])
        self.assertEqual("installer-trace-1", claimed["trace"]["correlation_id"])
        updated = update_module_installer_request(
            runtime_dir=runtime_dir,
            request_id=request_id,
            patch={"status": "completed", "result": {"return_code": 0}},
        )
        self.assertEqual("completed", updated["status"])
        self.assertEqual("completed", read_module_installer_request(runtime_dir=runtime_dir, request_id=request_id)["status"])

        cancellable = queue_module_installer_request(runtime_dir=runtime_dir, requested_by="user-1", tenant_id="tenant-1")
        cancelled = cancel_module_installer_request(
            runtime_dir=runtime_dir,
            request_id=str(cancellable["request_id"]),
            cancelled_by="user-2",
        )
        self.assertEqual("cancelled", cancelled["status"])
        self.assertEqual("user-2", cancelled["cancelled_by"])

        retry = retry_module_installer_request(
            runtime_dir=runtime_dir,
            request_id=str(cancellable["request_id"]),
            requested_by="user-3",
        )
        self.assertEqual("queued", retry["status"])
        self.assertEqual(cancellable["request_id"], retry["retry_of"])
        self.assertEqual("tenant-1", retry["tenant_id"])

        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        with database.session() as session:
            save_maintenance_mode(session, MaintenanceMode(enabled=True))
            plan = save_module_install_plan(session, [{
                "module_id": "files",
                "action": "install",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
            }])
            session.commit()
            result = run_module_install_plan(
                session=session,
                plan=plan,
                available=available_module_manifests(),
                current_enabled=("tenancy", "access"),
                desired_enabled=("tenancy", "access"),
                database_url=settings.database_url,
                runtime_dir=runtime_dir,
                dry_run=True,
                request_context={
                    "request_id": request_id,
                    "requested_by": "user-1",
                    "trace": queued["trace"],
                },
            )

        record = json.loads(result.record_path.read_text(encoding="utf-8"))
        self.assertEqual(request_id, record["request"]["request_id"])
        self.assertEqual("installer-trace-1", record["request"]["trace"]["correlation_id"])
        run_summary = list_module_installer_runs(runtime_dir=runtime_dir)[0]
        self.assertEqual(request_id, run_summary["request_id"])
        self.assertEqual("installer-trace-1", run_summary["trace"]["correlation_id"])

    def test_module_installer_daemon_status_reports_heartbeat(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-daemon-status-", dir=_TEST_ROOT))
        runtime_dir = root / "installer"

        initial = module_installer_daemon_status(runtime_dir=runtime_dir)
        self.assertFalse(initial["running"])

        status = update_module_installer_daemon_status(
            runtime_dir=runtime_dir,
            patch={"status": "polling", "current_request_id": None},
        )

        self.assertTrue(status["running"])
        self.assertEqual("polling", status["status"])

    def test_module_package_catalog_loads_json_entries(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "modules": [{
                "module_id": "files",
                "name": "Files",
                "version": "0.1.4",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
                "dependencies": ["access"],
                "optional_dependencies": ["mail"],
                "migration_safety": "forward_only",
                "migration_notes": "Requires a tested backup restore path.",
                "migration_after": ["access"],
                "migration_before": ["campaigns"],
                "migration_tasks": [{
                    "task_id": "backfill_spaces",
                    "phase": "post_migration_backfill",
                    "summary": "Backfill file spaces after schema migration.",
                    "task_version": "2",
                    "safety": "forward_only",
                    "idempotent": True,
                    "timeout_seconds": 30,
                }],
                "current_version_min": "0.1.0",
                "current_version_max_exclusive": "0.2.0",
                "bridge_release": True,
                "bridge_notes": "Keeps 0.1.x and 0.2.x file-space contracts available.",
                "allow_downgrade": True,
                "allow_same_version": True,
                "recovery_tested": True,
                "recovery_notes": "Restore rehearsal completed for the release candidate.",
                "provides_interfaces": [{"name": "files.spaces", "version": "1.2.0"}],
                "requires_interfaces": [{"name": "access.directory", "version_min": "1.0.0", "optional": True}],
                "artifact_integrity": {
                    "python": {
                        "ref": "govoplan-files==0.1.4",
                        "sha256": "0" * 64,
                        "sbom_url": "https://govoplan.example/sbom/files-0.1.4.spdx.json",
                        "provenance_url": "https://govoplan.example/provenance/files-0.1.4.intoto.jsonl",
                    }
                },
                "tags": ["official"],
            }],
        }), encoding="utf-8")

        catalog = module_package_catalog(catalog_path)

        self.assertEqual("files", catalog[0]["module_id"])
        self.assertEqual("install", catalog[0]["action"])
        self.assertEqual(["access"], catalog[0]["dependencies"])
        self.assertEqual(["mail"], catalog[0]["optional_dependencies"])
        self.assertEqual("forward_only", catalog[0]["migration_safety"])
        self.assertEqual("Requires a tested backup restore path.", catalog[0]["migration_notes"])
        self.assertEqual(["access"], catalog[0]["migration_after"])
        self.assertEqual(["campaigns"], catalog[0]["migration_before"])
        self.assertEqual([{
            "task_id": "backfill_spaces",
            "phase": "post_migration_backfill",
            "summary": "Backfill file spaces after schema migration.",
            "task_version": "2",
            "safety": "forward_only",
            "idempotent": True,
            "timeout_seconds": 30,
        }], catalog[0]["migration_tasks"])
        self.assertEqual("0.1.0", catalog[0]["current_version_min"])
        self.assertEqual("0.2.0", catalog[0]["current_version_max_exclusive"])
        self.assertTrue(catalog[0]["bridge_release"])
        self.assertEqual("Keeps 0.1.x and 0.2.x file-space contracts available.", catalog[0]["bridge_notes"])
        self.assertTrue(catalog[0]["allow_downgrade"])
        self.assertTrue(catalog[0]["allow_same_version"])
        self.assertTrue(catalog[0]["recovery_tested"])
        self.assertEqual("Restore rehearsal completed for the release candidate.", catalog[0]["recovery_notes"])
        self.assertEqual(["official"], catalog[0]["tags"])
        self.assertEqual([{"name": "files.spaces", "version": "1.2.0"}], catalog[0]["provides_interfaces"])
        self.assertEqual(
            [{"name": "access.directory", "optional": True, "version_min": "1.0.0"}],
            catalog[0]["requires_interfaces"],
        )
        self.assertEqual("0" * 64, catalog[0]["artifact_integrity"]["python"]["sha256"])

        validation = validate_module_package_catalog(catalog_path)
        self.assertTrue(validation["valid"])
        self.assertEqual("files", validation["modules"][0]["module_id"])

    def test_module_package_catalog_warns_about_interface_range_mismatch(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-interfaces-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "modules": [
                {
                    "module_id": "files",
                    "version": "2.0.0",
                    "python_package": "govoplan-files",
                    "python_ref": "govoplan-files==2.0.0",
                    "provides_interfaces": [{"name": "files.spaces", "version": "2.0.0"}],
                },
                {
                    "module_id": "campaigns",
                    "version": "1.1.0",
                    "python_package": "govoplan-campaign",
                    "python_ref": "govoplan-campaign==1.1.0",
                    "requires_interfaces": [{
                        "name": "files.spaces",
                        "version_min": "1.0.0",
                        "version_max_exclusive": "2.0.0",
                    }],
                },
            ],
        }), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path)

        self.assertTrue(validation["valid"])
        self.assertTrue(any("catalog providers" in warning for warning in validation["warnings"]))

    def test_module_package_catalog_validation_reports_bad_entries(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-invalid-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({"modules": [{"name": "Missing id"}]}), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path)

        self.assertFalse(validation["valid"])
        self.assertIn("module_id", str(validation["error"]))

    def test_module_package_catalog_rejects_invalid_migration_safety(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-invalid-safety-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "modules": [{
                "module_id": "files",
                "migration_safety": "maybe",
            }],
        }), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path)

        self.assertFalse(validation["valid"])
        self.assertIn("migration_safety", str(validation["error"]))

    def test_module_package_catalog_rejects_invalid_current_version_window(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-invalid-current-window-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "modules": [{
                "module_id": "files",
                "current_version_min": "2.0.0",
                "current_version_max_exclusive": "2.0.0",
            }],
        }), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path)

        self.assertFalse(validation["valid"])
        self.assertIn("current-version range", str(validation["error"]))

    def test_module_package_catalog_rejects_invalid_interface_ranges(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-invalid-interface-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "modules": [{
                "module_id": "campaigns",
                "python_package": "govoplan-campaign",
                "python_ref": "govoplan-campaign==1.1.0",
                "requires_interfaces": [{
                    "name": "files.spaces",
                    "version_min": "2.0.0",
                    "version_max_exclusive": "2.0.0",
                }],
            }],
        }), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path)

        self.assertFalse(validation["valid"])
        self.assertIn("invalid interface range", str(validation["error"]))

    def test_module_package_catalog_requires_signature_by_default_in_production(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-production-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 1,
            "modules": [{"module_id": "files"}],
        }), encoding="utf-8")

        with patch.dict(os.environ, {
            "APP_ENV": "production",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_REQUIRE_SIGNATURE": "",
        }):
            validation = validate_module_package_catalog(catalog_path)

        self.assertFalse(validation["valid"])
        self.assertIn("signed", str(validation["error"]))

        with patch.dict(os.environ, {
            "APP_ENV": "production",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_REQUIRE_SIGNATURE": "false",
        }):
            relaxed = validate_module_package_catalog(catalog_path)

        self.assertTrue(relaxed["valid"], relaxed["error"])
        self.assertTrue(any("unsigned" in warning for warning in relaxed["warnings"]))

    def test_module_package_catalog_validates_signed_approved_channel(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-signed-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        private_key_path = root / "catalog-private.pem"
        signed_path = root / "catalog.signed.json"
        private_key = Ed25519PrivateKey.generate()
        private_key_path.write_bytes(private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        ))
        public_key = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "modules": [{
                "module_id": "files",
                "python_package": "govoplan-files",
                "python_ref": "govoplan-files==0.1.4",
            }],
        }), encoding="utf-8")

        sign_module_package_catalog(
            path=catalog_path,
            key_id="release-1",
            private_key_path=private_key_path,
            output_path=signed_path,
        )
        validation = validate_module_package_catalog(
            signed_path,
            require_trusted=True,
            approved_channels=("stable",),
            trusted_keys={"release-1": base64.b64encode(public_key).decode("ascii")},
        )

        self.assertTrue(validation["valid"], validation["error"])
        self.assertTrue(validation["signed"])
        self.assertTrue(validation["trusted"])
        self.assertEqual("stable", validation["channel"])

    def test_release_catalog_generator_reads_manifest_interface_contracts(self) -> None:
        generator = runpy.run_path(
            str(Path(__file__).resolve().parents[2] / "govoplan" / "tools" / "release" / "generate-release-catalog.py"),
            run_name="govoplan_release_catalog_contract_test",
        )
        generated_at = generator["datetime"].now(tz=generator["UTC"])
        catalog = generator["_catalog_payload"](
            version="0.1.9",
            tag="v0.1.9",
            channel="test",
            sequence=1,
            generated_at=generated_at,
            expires_at=generated_at + generator["timedelta"](days=1),
            repository_base="git+ssh://git@example.test/add-ideas",
            public_base_url="https://example.test",
        )
        modules = {item["module_id"]: item for item in catalog["modules"]}

        self.assertIn({"name": "files.campaign_attachments", "version": "0.1.6"}, modules["files"]["provides_interfaces"])
        self.assertIn({
            "name": "campaigns.access",
            "optional": True,
            "version_min": "0.1.0",
            "version_max_exclusive": "0.2.0",
        }, modules["files"]["requires_interfaces"])
        self.assertEqual(
            ["campaigns", "encryption"],
            modules["files"]["optional_dependencies"],
        )
        self.assertIn({"name": "mail.campaign_delivery", "version": "0.2.0"}, modules["mail"]["provides_interfaces"])
        self.assertIn({
            "name": "campaigns.access",
            "optional": True,
            "version_min": "0.1.0",
            "version_max_exclusive": "0.2.0",
        }, modules["mail"]["requires_interfaces"])
        self.assertIn({
            "name": "files.campaign_attachments",
            "optional": True,
            "version_min": "0.1.0",
            "version_max_exclusive": "0.2.0",
        }, modules["campaigns"]["requires_interfaces"])
        self.assertIn({
            "name": "mail.campaign_delivery",
            "optional": True,
            "version_min": "0.2.0",
            "version_max_exclusive": "0.3.0",
        }, modules["campaigns"]["requires_interfaces"])
        self.assertEqual(
            ["files", "mail", "notifications", "addresses", "postbox", "approvals"],
            modules["campaigns"]["optional_dependencies"],
        )
        self.assertEqual("requires_review", modules["files"]["migration_safety"])
        self.assertIn("migration", modules["files"]["migration_notes"].lower())
        self.assertEqual("0.1.9", modules["files"]["version"])
        self.assertIn("@v0.1.9", modules["files"]["python_ref"])

    def test_module_package_catalog_validates_remote_url_and_cache_fallback(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-remote-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        private_key_path = root / "catalog-private.pem"
        signed_path = root / "catalog.signed.json"
        cache_path = root / "catalog-cache.json"
        private_key = Ed25519PrivateKey.generate()
        private_key_path.write_bytes(private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        ))
        public_key = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 42,
            "generated_at": "2026-07-07T00:00:00Z",
            "expires_at": "2030-12-31T23:59:59Z",
            "modules": [{"module_id": "files"}],
        }), encoding="utf-8")
        sign_module_package_catalog(
            path=catalog_path,
            key_id="release-remote",
            private_key_path=private_key_path,
            output_path=signed_path,
        )
        body = signed_path.read_text(encoding="utf-8")

        env = {
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_URL": "https://catalog.example.invalid/stable.json",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_CACHE": str(cache_path),
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_REQUIRE_SIGNATURE": "true",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS": "stable",
        }
        trusted_keys = {"release-remote": base64.b64encode(public_key).decode("ascii")}
        with patch.dict(os.environ, env):
            with patch("govoplan_core.core.module_package_catalog.fetch_http_text", return_value=body):
                fetched = validate_module_package_catalog(trusted_keys=trusted_keys)
            with patch(
                "govoplan_core.core.module_package_catalog.fetch_http_text",
                side_effect=OSError("offline"),
            ):
                cached = validate_module_package_catalog(trusted_keys=trusted_keys)

        self.assertTrue(fetched["valid"], fetched["error"])
        self.assertEqual("url", fetched["source_type"])
        self.assertFalse(fetched["cache_used"])
        self.assertEqual("stable", fetched["channel"])
        self.assertEqual(42, fetched["sequence"])
        self.assertTrue(fetched["trusted"])
        self.assertEqual("release-remote", fetched["key_id"])
        self.assertTrue(cache_path.exists())
        self.assertTrue(cached["valid"], cached["error"])
        self.assertTrue(cached["cache_used"])
        self.assertEqual(str(cache_path), cached["cache_path"])
        self.assertTrue(cached["trusted"])

    def test_module_package_catalog_rejects_unapproved_channel_when_required(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-channel-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({"channel": "nightly", "modules": [{"module_id": "files"}]}), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path, approved_channels=("stable",))

        self.assertFalse(validation["valid"])
        self.assertIn("not approved", str(validation["error"]))

    def test_module_package_catalog_rejects_expired_catalog(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-expired-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 1,
            "generated_at": "2000-01-01T00:00:00Z",
            "expires_at": "2000-01-02T00:00:00Z",
            "modules": [{"module_id": "files"}],
        }), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path)

        self.assertFalse(validation["valid"])
        self.assertIn("expired", str(validation["error"]))

    def test_module_package_catalog_rejects_not_before_catalog(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-not-before-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 1,
            "generated_at": "2026-07-07T00:00:00Z",
            "not_before": "2030-01-01T00:00:00Z",
            "expires_at": "2030-12-31T23:59:59Z",
            "modules": [{"module_id": "files"}],
        }), encoding="utf-8")

        validation = validate_module_package_catalog(catalog_path)

        self.assertFalse(validation["valid"])
        self.assertEqual("2030-01-01T00:00:00Z", validation["not_before"])
        self.assertIn("not valid before", str(validation["error"]))

    def test_module_package_catalog_records_sequence_acceptance(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-sequence-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        older_catalog_path = root / "catalog-older.json"
        state_path = root / "sequence-state.json"
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 7,
            "generated_at": "2026-07-07T00:00:00Z",
            "expires_at": "2030-12-31T23:59:59Z",
            "modules": [{"module_id": "files"}],
        }), encoding="utf-8")
        older_catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 6,
            "generated_at": "2026-07-06T00:00:00Z",
            "expires_at": "2030-12-31T23:59:59Z",
            "modules": [{"module_id": "files"}],
        }), encoding="utf-8")

        with patch.dict(os.environ, {
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_SEQUENCE_STATE": str(state_path),
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_ENFORCE_SEQUENCE": "true",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS": "",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_REQUIRE_SIGNATURE": "",
        }):
            validation = validate_module_package_catalog(catalog_path)
            self.assertTrue(validation["valid"], validation["error"])
            record_module_package_catalog_acceptance(validation)
            replay = validate_module_package_catalog(catalog_path)
            older = validate_module_package_catalog(older_catalog_path)
        with patch.dict(os.environ, {
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_SEQUENCE_STATE": str(state_path),
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_ENFORCE_SEQUENCE": "",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_APPROVED_CHANNELS": "",
            "GOVOPLAN_MODULE_PACKAGE_CATALOG_REQUIRE_SIGNATURE": "",
        }):
            same_non_strict = validate_module_package_catalog(catalog_path)

        self.assertFalse(replay["valid"])
        self.assertIn("already accepted", str(replay["error"]))
        self.assertFalse(older["valid"])
        self.assertIn("older than accepted", str(older["error"]))
        self.assertTrue(same_non_strict["valid"], same_non_strict["error"])

    def test_configuration_package_catalog_validates_signed_entries(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-configuration-package-catalog-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        private_key_path = root / "configuration-private.pem"
        signed_path = root / "catalog.signed.json"
        private_key = Ed25519PrivateKey.generate()
        private_key_path.write_bytes(private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        ))
        public_key = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        catalog_path.write_text(json.dumps({
            "channel": "stable",
            "sequence": 3,
            "generated_at": "2026-07-07T00:00:00Z",
            "expires_at": "2030-12-31T23:59:59Z",
            "packages": [
                {
                    "package_id": "govoplan.application-handling.basic",
                    "name": "Basic application handling",
                    "version": "0.1.0",
                    "publisher": "ADD ideas",
                    "category": "workflow",
                    "artifact_ref": "git+ssh://git.example.invalid/config-packages.git@application-basic-v0.1.0",
                    "artifact_sha256": "a" * 64,
                    "required_modules": [{"module_id": "access", "version": ">=0.1.0"}],
                    "required_capabilities": [CONFIGURATION_PROVIDER_CAPABILITY, "access.configuration"],
                    "fragments": [
                        {
                            "module_id": "access",
                            "fragment_type": "access_roles",
                            "fragment_id": "clerks",
                            "payload": {"access_roles": [{"slug": "application-clerk"}]},
                        }
                    ],
                    "data_requirements": [{"key": "service_name", "label": "Service name"}],
                    "tags": ["official", "example"],
                }
            ],
        }), encoding="utf-8")

        unsigned = validate_configuration_package_catalog(catalog_path)
        self.assertTrue(unsigned["valid"], unsigned["error"])
        self.assertFalse(unsigned["signed"])
        self.assertEqual("govoplan.application-handling.basic", unsigned["packages"][0]["package_id"])
        self.assertEqual("access", unsigned["packages"][0]["required_modules"][0]["module_id"])
        self.assertIn("unsigned", " ".join(unsigned["warnings"]).lower())

        sign_configuration_package_catalog(
            path=catalog_path,
            key_id="configuration-release-1",
            private_key_path=private_key_path,
            output_path=signed_path,
        )
        validation = validate_configuration_package_catalog(
            signed_path,
            require_trusted=True,
            approved_channels=("stable",),
            trusted_keys={"configuration-release-1": base64.b64encode(public_key).decode("ascii")},
        )
        self.assertTrue(validation["valid"], validation["error"])
        self.assertTrue(validation["signed"])
        self.assertTrue(validation["trusted"])
        self.assertEqual("stable", validation["channel"])

        catalog = configuration_package_catalog(
            signed_path,
            require_trusted=True,
            approved_channels=("stable",),
            trusted_keys={"configuration-release-1": base64.b64encode(public_key).decode("ascii")},
        )
        self.assertEqual(("official", "example"), tuple(catalog[0]["tags"]))

    def test_configuration_provider_contract_is_runtime_checkable(self) -> None:
        class AccessConfigurationProvider:
            module_id = "access"

            def describe(self) -> ConfigurationProviderDescription:
                return ConfigurationProviderDescription(module_id=self.module_id, fragment_types=("access_roles",), exported_scopes=("tenant",))

            def preflight(self, fragment: ConfigurationPackageFragment, context: ConfigurationPreflightContext) -> ConfigurationPreflightResult:
                del context
                return ConfigurationPreflightResult(
                    required_data=(ConfigurationRequiredData(key="service_name", label="Service name"),),
                    plan=(ConfigurationPlanItem(action="create", module_id=fragment.module_id, fragment_type=fragment.fragment_type, fragment_id=fragment.fragment_id),),
                )

            def apply(self, fragment: ConfigurationPackageFragment, supplied_data: dict[str, object], context: ConfigurationPreflightContext) -> ConfigurationApplyResult:
                del supplied_data, context
                return ConfigurationApplyResult(created_refs={fragment.fragment_id or fragment.fragment_type: "role:application-clerk"})

            def export(self, selection: ConfigurationExportSelection, context: ConfigurationPreflightContext) -> ConfigurationExportResult:
                del selection, context
                return ConfigurationExportResult(fragments=(ConfigurationPackageFragment(module_id=self.module_id, fragment_type="access_roles", fragment_id="clerks"),))

            def health(self, import_result: ConfigurationApplyResult, context: ConfigurationPreflightContext):
                del import_result, context
                return ()

        provider = AccessConfigurationProvider()
        fragment = ConfigurationPackageFragment(module_id="access", fragment_type="access_roles", fragment_id="clerks")

        self.assertIsInstance(provider, ConfigurationProvider)
        self.assertEqual(("access_roles",), provider.describe().fragment_types)
        self.assertEqual("service_name", provider.preflight(fragment, ConfigurationPreflightContext()).required_data[0].key)
        self.assertEqual("role:application-clerk", provider.apply(fragment, {}, ConfigurationPreflightContext()).created_refs["clerks"])

        package = {
            "package_id": "govoplan.test",
            "name": "Test package",
            "version": "1.0.0",
            "required_modules": [{"module_id": "access", "version": "1.0.0"}],
            "required_capabilities": [CONFIGURATION_PROVIDER_CAPABILITY, "access.configuration"],
            "data_requirements": [{"key": "service_name", "label": "Service name"}],
            "fragments": [{"module_id": "access", "fragment_type": "access_roles", "fragment_id": "clerks", "payload": {"access_roles": []}}],
        }
        missing_context = ConfigurationPreflightContext(
            installed_modules={"access": "1.0.0"},
            capabilities=frozenset({CONFIGURATION_PROVIDER_CAPABILITY, "access.configuration"}),
        )
        missing = dry_run_configuration_package(package, [provider], missing_context)
        self.assertIn("required_data_missing", {item.code for item in missing.diagnostics})
        self.assertEqual("service_name", missing.required_data[0].key)

        ready_context = ConfigurationPreflightContext(
            supplied_data={"service_name": "Application handling"},
            installed_modules={"access": "1.0.0"},
            capabilities=frozenset({CONFIGURATION_PROVIDER_CAPABILITY, "access.configuration"}),
        )
        dry_run = dry_run_configuration_package(package, [provider], ready_context)
        self.assertFalse([item for item in dry_run.diagnostics if item.severity == "blocker"])
        self.assertEqual("create", dry_run.plan[0].action)

        applied = apply_configuration_package(package, [provider], ready_context)
        self.assertEqual("role:application-clerk", applied.created_refs["clerks"])

        exported = export_configuration_package([provider], ConfigurationExportSelection(module_ids=("access",)), ready_context)
        self.assertEqual("access_roles", exported.fragments[0].fragment_type)

    def test_module_license_decision_is_observe_only_unless_enforced(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-license-observe-", dir=_TEST_ROOT))
        missing_license = root / "missing-license.json"

        with patch.dict(os.environ, {
            "GOVOPLAN_LICENSE_FILE": str(missing_license),
            "GOVOPLAN_LICENSE_ENFORCEMENT": "",
        }):
            observe = module_license_decision(("module.mail",))

        with patch.dict(os.environ, {
            "GOVOPLAN_LICENSE_FILE": str(missing_license),
            "GOVOPLAN_LICENSE_ENFORCEMENT": "true",
        }):
            enforced = module_license_decision(("module.mail",))

        self.assertTrue(observe["allowed"])
        self.assertFalse(observe["enforced"])
        self.assertFalse(enforced["allowed"])
        self.assertTrue(enforced["enforced"])

    def test_module_license_validates_signed_feature_license(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-license-signed-", dir=_TEST_ROOT))
        license_path = root / "license.json"
        private_key = Ed25519PrivateKey.generate()
        public_key = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        payload = {
            "license_id": "test-license",
            "subject": "Test installation",
            "features": ["module.mail"],
            "valid_from": "2026-01-01T00:00:00Z",
            "valid_until": "2030-12-31T23:59:59Z",
        }
        signature = private_key.sign(json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8"))
        payload["signature"] = {
            "algorithm": "ed25519",
            "key_id": "license-1",
            "value": base64.b64encode(signature).decode("ascii"),
        }
        license_path.write_text(json.dumps(payload), encoding="utf-8")
        trusted_keys = {"license-1": base64.b64encode(public_key).decode("ascii")}

        validation = validate_module_license(license_path, trusted_keys=trusted_keys, require_trusted=True)

        self.assertTrue(validation["valid"], validation["error"])
        self.assertTrue(validation["trusted"])
        self.assertEqual(["module.mail"], validation["features"])

    def test_module_license_issuer_writes_signed_license(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-license-issuer-", dir=_TEST_ROOT))
        private_key_path = root / "license-private.pem"
        license_path = root / "license.json"
        private_key = Ed25519PrivateKey.generate()
        private_key_path.write_bytes(private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        ))
        public_key = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )

        issue_module_license(
            path=license_path,
            license_id="license-2026",
            subject="GovOPlaN Test",
            features=("module.mail", "support.standard"),
            valid_from="2026-01-01T00:00:00Z",
            valid_until="2030-12-31T23:59:59Z",
            signing_key_id="license-key-1",
            signing_private_key_path=private_key_path,
            issuer="GovOPlaN Release",
        )
        trusted_keys = {"license-key-1": base64.b64encode(public_key).decode("ascii")}

        validation = validate_module_license(license_path, trusted_keys=trusted_keys, require_trusted=True)
        diagnostics = module_license_diagnostics(
            license_path,
            trusted_keys=trusted_keys,
            require_trusted=True,
            required_features=("module.mail",),
        )

        self.assertTrue(validation["valid"], validation["error"])
        self.assertEqual("license-2026", validation["license_id"])
        self.assertTrue(diagnostics["trusted"])
        self.assertEqual([], diagnostics["missing_features"])
        self.assertIn("module.mail", diagnostics["features"])

    def test_module_license_diagnostics_report_expired_unsigned_missing_and_missing_feature(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-license-diagnostics-", dir=_TEST_ROOT))
        missing_path = root / "missing.json"
        expired_path = root / "expired.json"
        unsigned_path = root / "unsigned.json"
        feature_path = root / "feature.json"
        private_key_path = root / "license-private.pem"
        private_key = Ed25519PrivateKey.generate()
        private_key_path.write_bytes(private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        ))
        public_key = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        trusted_keys = {"license-key-1": base64.b64encode(public_key).decode("ascii")}
        expired_path.write_text(json.dumps({
            "license_id": "expired",
            "subject": "Expired",
            "features": ["module.mail"],
            "valid_from": "2020-01-01T00:00:00Z",
            "valid_until": "2020-01-02T00:00:00Z",
        }), encoding="utf-8")
        unsigned_path.write_text(json.dumps({
            "license_id": "unsigned",
            "subject": "Unsigned",
            "features": ["module.mail"],
            "valid_from": "2026-01-01T00:00:00Z",
            "valid_until": "2030-12-31T23:59:59Z",
        }), encoding="utf-8")
        issue_module_license(
            path=feature_path,
            license_id="feature",
            subject="Feature",
            features=("module.files",),
            valid_from="2026-01-01T00:00:00Z",
            valid_until="2030-12-31T23:59:59Z",
            signing_key_id="license-key-1",
            signing_private_key_path=private_key_path,
        )

        with patch.dict(os.environ, {"GOVOPLAN_LICENSE_ENFORCEMENT": "true"}):
            missing = module_license_diagnostics(missing_path, required_features=("module.mail",), require_trusted=True)
            expired = module_license_diagnostics(expired_path, required_features=("module.mail",), require_trusted=False)
            unsigned = module_license_diagnostics(unsigned_path, required_features=("module.mail",), require_trusted=True)
            missing_feature = module_license_diagnostics(
                feature_path,
                trusted_keys=trusted_keys,
                require_trusted=True,
                required_features=("module.mail",),
            )

        self.assertFalse(missing["valid"])
        self.assertFalse(missing["allowed"])
        self.assertFalse(expired["valid"])
        self.assertIn("expired", str(expired["reason"]))
        self.assertFalse(unsigned["valid"])
        self.assertFalse(unsigned["signed"])
        self.assertFalse(unsigned["trusted"])
        self.assertTrue(missing_feature["valid"], missing_feature["reason"])
        self.assertTrue(missing_feature["trusted"])
        self.assertEqual(["module.mail"], missing_feature["missing_features"])
        self.assertFalse(missing_feature["allowed"])

    def test_module_installer_cli_issues_and_validates_license(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-license-cli-", dir=_TEST_ROOT))
        private_key_path = root / "license-private.pem"
        license_path = root / "license.json"
        private_key = Ed25519PrivateKey.generate()
        private_key_path.write_bytes(private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        ))
        public_key = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        trusted_key = base64.b64encode(public_key).decode("ascii")

        issue_result = subprocess.run(
            [
                sys.executable,
                "-m",
                "govoplan_core.commands.module_installer",
                "--issue-license",
                str(license_path),
                "--license-id",
                "cli-license",
                "--license-subject",
                "CLI Test",
                "--license-feature",
                "module.mail",
                "--license-valid-from",
                "2026-01-01T00:00:00Z",
                "--license-valid-until",
                "2030-12-31T23:59:59Z",
                "--license-signing-key-id",
                "license-key-1",
                "--license-signing-private-key",
                str(private_key_path),
                "--format",
                "json",
            ],
            cwd=str(Path(__file__).resolve().parents[1]),
            env=os.environ.copy(),
            text=True,
            capture_output=True,
            check=False,
        )
        self.assertEqual(0, issue_result.returncode, issue_result.stderr)
        self.assertTrue(json.loads(issue_result.stdout)["issued"])

        validate_result = subprocess.run(
            [
                sys.executable,
                "-m",
                "govoplan_core.commands.module_installer",
                "--validate-license",
                str(license_path),
                "--license-trusted-key",
                f"license-key-1={trusted_key}",
                "--require-trusted-license",
                "--license-required-feature",
                "module.mail",
                "--format",
                "json",
            ],
            cwd=str(Path(__file__).resolve().parents[1]),
            env=os.environ.copy(),
            text=True,
            capture_output=True,
            check=False,
        )

        self.assertEqual(0, validate_result.returncode, validate_result.stderr)
        diagnostics = json.loads(validate_result.stdout)
        self.assertTrue(diagnostics["valid"], diagnostics["reason"])
        self.assertTrue(diagnostics["trusted"])
        self.assertEqual([], diagnostics["missing_features"])

    def test_module_installer_validates_package_catalog_from_cli(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-package-catalog-cli-", dir=_TEST_ROOT))
        catalog_path = root / "catalog.json"
        catalog_path.write_text(json.dumps({"modules": [{"module_id": "files"}]}), encoding="utf-8")

        result = subprocess.run(
            [
                sys.executable,
                "-m",
                "govoplan_core.commands.module_installer",
                "--validate-package-catalog",
                str(catalog_path),
                "--format",
                "json",
            ],
            cwd=str(Path(__file__).resolve().parents[1]),
            env=os.environ.copy(),
            text=True,
            capture_output=True,
            check=False,
        )

        self.assertEqual(0, result.returncode, result.stderr)
        self.assertTrue(json.loads(result.stdout)["valid"])

    def test_module_installer_daemon_once_exits_with_empty_queue(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-daemon-once-", dir=_TEST_ROOT))
        result = subprocess.run(
            [
                sys.executable,
                "-m",
                "govoplan_core.commands.module_installer",
                "--daemon-once",
                "--runtime-dir",
                str(root / "installer"),
                "--database-url",
                f"sqlite:///{root / 'daemon.db'}",
            ],
            cwd=str(Path(__file__).resolve().parents[1]),
            env=os.environ.copy(),
            text=True,
            capture_output=True,
            check=False,
        )

        self.assertEqual(0, result.returncode, result.stderr)

    def test_module_installer_rollback_rebuilds_webui_when_original_run_built(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-rollback-webui-", dir=_TEST_ROOT))
        runtime_dir = root / "installer"
        run_id = "rollback-webui-test"
        run_dir = runtime_dir / "runs" / run_id
        webui_root = root / "webui"
        run_dir.mkdir(parents=True)
        webui_root.mkdir()
        (webui_root / "package.json").write_text('{"scripts":{"build":"vite build"}}', encoding="utf-8")
        (run_dir / "package.json.before").write_text('{"scripts":{"build":"vite build"},"dependencies":{}}', encoding="utf-8")
        (run_dir / "package-lock.json.before").write_text('{"lockfileVersion":3}', encoding="utf-8")
        (run_dir / "record.json").write_text(json.dumps({
            "run_id": run_id,
            "build_webui": True,
            "plan": [{
                "module_id": "example",
                "action": "install",
                "python_package": "govoplan-example",
            }],
            "snapshot": {
                "webui_root": str(webui_root),
                "package_json": "package.json.before",
                "package_lock": "package-lock.json.before",
            },
        }), encoding="utf-8")

        with patch("govoplan_core.core.module_installer.subprocess.run") as run_mock:
            run_mock.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
            result = rollback_module_install_run(run_id=run_id, runtime_dir=runtime_dir, npm_bin="npm")

        commands = [call.args[0] for call in run_mock.call_args_list]
        self.assertEqual("rolled-back", result.status)
        self.assertIn([sys.executable, "-m", "pip", "uninstall", "-y", "govoplan-example"], commands)
        self.assertIn(["npm", "install"], commands)
        self.assertIn(["npm", "run", "build"], commands)

    def test_create_app_uses_saved_desired_module_state_on_startup(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-state-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        with database.session() as session:
            session.add(SystemSettings(
                id="global",
                settings={"module_management": {"desired_enabled": ["files"]}},
            ))
            session.commit()

        config = GovoplanServerConfig(
            title="GovOPlaN Module State Test",
            version="test",
            settings=settings,
            enabled_modules=(),
            base_routers=(),
            post_module_routers=(),
            extra_routers=(),
            lifespan=None,
            app_configurators=(),
        )

        app = create_app(config)
        with TestClient(app) as client:
            response = client.get("/api/v1/platform/modules")

        self.assertEqual(response.status_code, 401, response.text)
        payload_modules = {
            item.id for item in app.state.govoplan_registry.manifests()
        }
        self.assertEqual({"access", "files"}, payload_modules)

    def test_create_app_ignores_saved_desired_modules_that_are_not_installed(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-state-stale-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        with database.session() as session:
            session.add(SystemSettings(
                id="global",
                settings={"module_management": {"desired_enabled": ["files", "missing-module"]}},
            ))
            session.commit()

        config = GovoplanServerConfig(
            title="GovOPlaN Module State Test",
            version="test",
            settings=settings,
            enabled_modules=(),
            base_routers=(),
            post_module_routers=(),
            extra_routers=(),
            lifespan=None,
            app_configurators=(),
        )

        app = create_app(config)
        with TestClient(app) as client:
            response = client.get("/api/v1/platform/modules")

        self.assertEqual(response.status_code, 401, response.text)
        payload_modules = {
            item.id for item in app.state.govoplan_registry.manifests()
        }
        self.assertEqual({"access", "files"}, payload_modules)

    def test_create_app_preserves_admin_when_configured_and_saved_state_is_older(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-module-state-admin-", dir=_TEST_ROOT))
        settings = _settings(root)
        configure_database(settings.database_url)
        database = get_database()
        Base.metadata.create_all(bind=database.engine, tables=[SystemSettings.__table__])
        with database.session() as session:
            session.add(SystemSettings(
                id="global",
                settings={"module_management": {"desired_enabled": ["files"]}},
            ))
            session.commit()

        config = GovoplanServerConfig(
            title="GovOPlaN Module State Test",
            version="test",
            settings=settings,
            enabled_modules=("admin",),
            base_routers=(),
            post_module_routers=(),
            extra_routers=(),
            lifespan=None,
            app_configurators=(),
        )

        app = create_app(config)
        with TestClient(app) as client:
            response = client.get("/api/v1/platform/modules")

        self.assertEqual(response.status_code, 401, response.text)
        payload_modules = {
            item.id for item in app.state.govoplan_registry.manifests()
        }
        self.assertEqual({"access", "admin", "files"}, payload_modules)

    def test_module_lifecycle_enables_and_disables_routes_without_restart(self) -> None:
        app, _settings_obj = self._app_for_modules(())
        lifecycle = getattr(app.state, "govoplan_lifecycle", None)
        self.assertIsNotNone(lifecycle)

        with TestClient(app) as client:
            response = client.get("/api/v1/platform/modules")
            self.assertEqual(response.status_code, 503, response.text)
            self.assertEqual((), lifecycle.active_module_ids())
            self.assertFalse("/api/v1/files" in _route_paths(app))

            result = lifecycle.apply_enabled_modules(("files",), migrate=False)
            self.assertEqual(("access", "files"), result.activated_modules)
            response = client.get("/api/v1/platform/modules")
            self.assertEqual(response.status_code, 401, response.text)
            self.assertEqual(("access", "files"), lifecycle.active_module_ids())
            self.assertTrue("/api/v1/files" in _route_paths(app))
            self.assertEqual(401, client.get("/api/v1/files").status_code)

            result = lifecycle.apply_enabled_modules((), migrate=False)
            self.assertEqual(("access", "files"), result.deactivated_modules)
            response = client.get("/api/v1/platform/modules")
            self.assertEqual(response.status_code, 503, response.text)
            self.assertEqual((), lifecycle.active_module_ids())
            disabled_response = client.get("/api/v1/files")
            self.assertEqual(404, disabled_response.status_code)
            self.assertEqual("Module is disabled: files", disabled_response.json()["detail"])

    def test_module_lifecycle_live_apply_defaults_by_environment(self) -> None:
        app, settings = self._app_for_modules(())
        lifecycle = app.state.govoplan_lifecycle

        self.assertTrue(lifecycle.live_apply_enabled())
        settings.app_env = "production"
        self.assertFalse(lifecycle.live_apply_enabled())
        settings.module_live_apply_enabled = True
        self.assertTrue(lifecycle.live_apply_enabled())


    def test_module_permutations_start_when_absent_modules_are_physically_unavailable(self) -> None:
        cases = (
            ((), ("govoplan_tenancy", "govoplan_files", "govoplan_mail", "govoplan_campaign")),
            (("files",), ("govoplan_tenancy", "govoplan_mail", "govoplan_campaign")),
            (("mail",), ("govoplan_tenancy", "govoplan_files", "govoplan_campaign")),
            (("campaigns",), ("govoplan_tenancy", "govoplan_files", "govoplan_mail")),
            (("campaigns", "files"), ("govoplan_tenancy", "govoplan_mail")),
            (("campaigns", "mail"), ("govoplan_tenancy", "govoplan_files")),
            (("campaigns", "files", "mail"), ()),
        )
        for enabled_modules, blocked_modules in cases:
            with self.subTest(enabled_modules=enabled_modules, blocked_modules=blocked_modules):
                self._run_physical_absence_probe(enabled_modules=enabled_modules, blocked_modules=blocked_modules)

    def test_module_route_factories_receive_runtime_settings(self) -> None:
        app, settings = self._app_for_modules(("calendar", "files", "mail"))
        self.assertIsNotNone(app)

        from govoplan_calendar.backend.runtime import get_settings as get_calendar_settings
        from govoplan_files.backend.runtime import get_settings as get_files_settings
        from govoplan_mail.backend.runtime import get_settings as get_mail_settings

        self.assertIs(settings, get_calendar_settings())
        self.assertIs(settings, get_files_settings())
        self.assertIs(settings, get_mail_settings())

    def test_module_migrations_are_registered_only_for_enabled_modules(self) -> None:
        core_plan = migration_metadata_plan(build_platform_registry(()))
        self.assertEqual((), core_plan.script_locations)
        self.assertEqual((), core_plan.metadata)
        self.assertEqual((), core_plan.modules)

        files_plan = migration_metadata_plan(build_platform_registry(("files",)))
        self.assertEqual(2, len(files_plan.script_locations))
        self.assertIn("files", {item.module_id for item in files_plan.modules})
        files_locations = "\n".join(files_plan.script_locations)
        self.assertIn("govoplan_access/backend/migrations/versions", files_locations)
        self.assertIn("govoplan_files/backend/migrations/versions", files_locations)

        full_plan = migration_metadata_plan(build_platform_registry(("campaigns", "files", "mail")))
        locations = "\n".join(full_plan.script_locations)
        self.assertIn("govoplan_access/backend/migrations/versions", locations)
        self.assertIn("govoplan_campaign/backend/migrations/versions", locations)
        self.assertIn("govoplan_files/backend/migrations/versions", locations)
        self.assertIn("govoplan_mail/backend/migrations/versions", locations)

    def test_local_storage_backend_reads_from_fallback_roots_without_copying(self) -> None:
        with tempfile.TemporaryDirectory(prefix="govoplan-storage-fallback-") as directory:
            root = Path(directory) / "primary"
            fallback = Path(directory) / "fallback"
            key = "tenant-a/files/demo.txt"
            (fallback / key).parent.mkdir(parents=True)
            (fallback / key).write_bytes(b"fallback-data")

            backend = LocalFilesystemStorageBackend(root=root, fallback_roots=(fallback,))

            self.assertTrue(backend.exists(key))
            self.assertEqual(b"fallback-data", backend.get_bytes(key))
            self.assertEqual([b"fallback", b"-data"], list(backend.iter_bytes(key, chunk_size=8)))
            self.assertFalse((root / key).exists())

            backend.delete(key)
            self.assertTrue((fallback / key).exists())
            self.assertEqual(b"fallback-data", backend.get_bytes(key))

            with self.assertRaises(StorageBackendError):
                backend.get_bytes("../escape.txt")


if __name__ == "__main__":
    unittest.main()
