from __future__ import annotations

from collections import defaultdict
from collections.abc import Iterable, Mapping
from contextlib import AbstractContextManager, closing
from dataclasses import dataclass, field
from datetime import UTC, datetime
from importlib import metadata
import hashlib
import json
import logging
import os
from pathlib import Path
import re
import shlex
import shutil
import sqlite3
import stat
import subprocess  # nosec B404 - installer commands are structured and policy-validated before execution.
import sys
import tomllib
from typing import Any, Literal
import time

from sqlalchemy.engine import make_url
from sqlalchemy.orm import Session

from govoplan_core.core.maintenance import saved_maintenance_mode
from govoplan_core.core.events import current_event_trace
from govoplan_core.core.module_management import (
    PROTECTED_MODULES,
    ModuleInstallPlan,
    ModuleInstallPlanItem,
    desired_modules_after_package_plan,
    module_dependents,
    module_install_plan_commands,
    save_desired_enabled_modules,
    save_module_install_plan,
)
from govoplan_core.core.modules import ModuleManifest, MigrationRetirementPlan
from govoplan_core.core.versioning import compare_versions, format_version_range, version_satisfies_range
from govoplan_core.security.http_fetch import fetch_http, validate_http_url

logger = logging.getLogger(__name__)

IssueSeverity = Literal["blocker", "warning", "info"]
ChecklistStatus = Literal["done", "pending", "blocked", "warning"]
SUPPORTED_MANIFEST_CONTRACT_VERSION = "1"
PACKAGE_TARGET_ACTIONS = {"install", "update"}
MIGRATION_TASK_PHASES = (
    "pre_migration_check",
    "pre_migration_prepare",
    "post_migration_backfill",
    "post_migration_verify",
)
MIGRATION_TASK_MUTATING_PHASES = {"pre_migration_prepare", "post_migration_backfill"}
MIGRATION_TASK_REVIEW_SAFETY = {"requires_review", "forward_only", "destructive"}
MIGRATION_TASK_BLOCKING_SAFETY = {"forward_only", "destructive"}


@dataclass(frozen=True, slots=True)
class ModuleInstallerIssue:
    severity: IssueSeverity
    code: str
    message: str
    module_id: str | None = None

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "severity": self.severity,
            "code": self.code,
            "message": self.message,
        }
        if self.module_id:
            payload["module_id"] = self.module_id
        return payload


@dataclass(frozen=True, slots=True)
class ModuleInstallChecklistItem:
    id: str
    label: str
    status: ChecklistStatus
    detail: str | None = None

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "id": self.id,
            "label": self.label,
            "status": self.status,
        }
        if self.detail:
            payload["detail"] = self.detail
        return payload


@dataclass(frozen=True, slots=True)
class ModuleInstallTargetItem:
    module_id: str
    action: str
    source: str
    current_version: str | None = None
    target_version: str | None = None
    python_package: str | None = None
    python_ref: str | None = None
    webui_package: str | None = None
    webui_ref: str | None = None
    migration_safety: str = "automatic"
    migration_notes: str | None = None
    current_version_min: str | None = None
    current_version_max_exclusive: str | None = None
    bridge_release: bool = False
    bridge_notes: str | None = None
    allow_downgrade: bool = False
    allow_same_version: bool = False
    recovery_tested: bool = False
    recovery_notes: str | None = None
    data_safety_acknowledged: bool = False

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "module_id": self.module_id,
            "action": self.action,
            "source": self.source,
            "migration_safety": self.migration_safety,
            "bridge_release": self.bridge_release,
            "allow_downgrade": self.allow_downgrade,
            "allow_same_version": self.allow_same_version,
            "recovery_tested": self.recovery_tested,
            "data_safety_acknowledged": self.data_safety_acknowledged,
        }
        for key in (
            "current_version",
            "target_version",
            "python_package",
            "python_ref",
            "webui_package",
            "webui_ref",
            "migration_notes",
            "current_version_min",
            "current_version_max_exclusive",
            "bridge_notes",
            "recovery_notes",
        ):
            value = getattr(self, key)
            if value:
                payload[key] = value
        return payload


@dataclass(frozen=True, slots=True)
class ModuleMigrationPlanStep:
    module_id: str
    action: str
    phase: str
    source: str
    has_migration_metadata: bool = False
    metadata_pending: bool = False
    migration_safety: str = "automatic"
    current_version: str | None = None
    target_version: str | None = None
    reason: str | None = None

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "module_id": self.module_id,
            "action": self.action,
            "phase": self.phase,
            "source": self.source,
            "has_migration_metadata": self.has_migration_metadata,
            "metadata_pending": self.metadata_pending,
            "migration_safety": self.migration_safety,
        }
        for key in ("current_version", "target_version", "reason"):
            value = getattr(self, key)
            if value:
                payload[key] = value
        return payload


@dataclass(frozen=True, slots=True)
class ModuleMigrationTaskPlanItem:
    module_id: str
    task_id: str
    phase: str
    summary: str
    task_version: str = "1"
    safety: str = "automatic"
    idempotent: bool = True
    timeout_seconds: int | None = None
    source: str = "manifest"
    has_executor: bool = False
    metadata_pending: bool = False

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "module_id": self.module_id,
            "task_id": self.task_id,
            "phase": self.phase,
            "summary": self.summary,
            "task_version": self.task_version,
            "safety": self.safety,
            "idempotent": self.idempotent,
            "source": self.source,
            "has_executor": self.has_executor,
            "metadata_pending": self.metadata_pending,
        }
        if self.timeout_seconds is not None:
            payload["timeout_seconds"] = self.timeout_seconds
        return payload


@dataclass(frozen=True, slots=True)
class ModuleMigrationExecutionPlan:
    enabled_modules: tuple[str, ...] = ()
    steps: tuple[ModuleMigrationPlanStep, ...] = ()
    tasks: tuple[ModuleMigrationTaskPlanItem, ...] = ()

    @property
    def requires_database_migration(self) -> bool:
        return bool(self.steps or self.tasks)

    def as_dict(self) -> dict[str, object]:
        return {
            "enabled_modules": list(self.enabled_modules),
            "requires_database_migration": self.requires_database_migration,
            "steps": [step.as_dict() for step in self.steps],
            "tasks": [task.as_dict() for task in self.tasks],
        }


@dataclass(frozen=True, slots=True)
class ModuleInstallerPreflight:
    allowed: bool
    maintenance_mode: bool
    frontend_rebuild_required: bool
    restart_required: bool
    commands: tuple[str, ...]
    rollback_commands: tuple[str, ...]
    issues: tuple[ModuleInstallerIssue, ...]
    checklist: tuple[ModuleInstallChecklistItem, ...] = ()
    target_plan: tuple[ModuleInstallTargetItem, ...] = ()
    migration_plan: ModuleMigrationExecutionPlan = field(default_factory=ModuleMigrationExecutionPlan)
    artifact_integrity: tuple[dict[str, object], ...] = ()

    def as_dict(self) -> dict[str, object]:
        return {
            "allowed": self.allowed,
            "maintenance_mode": self.maintenance_mode,
            "frontend_rebuild_required": self.frontend_rebuild_required,
            "restart_required": self.restart_required,
            "commands": list(self.commands),
            "rollback_commands": list(self.rollback_commands),
            "issues": [issue.as_dict() for issue in self.issues],
            "checklist": [item.as_dict() for item in self.checklist],
            "target_plan": [item.as_dict() for item in self.target_plan],
            "migration_plan": self.migration_plan.as_dict(),
            "artifact_integrity": list(self.artifact_integrity),
        }


@dataclass(frozen=True, slots=True)
class ModuleInstallerRunResult:
    run_id: str
    status: Literal["dry-run", "applied", "failed", "rolled-back"]
    record_path: Path
    commands: tuple[str, ...]
    return_code: int = 0
    error: str | None = None
    rollback: dict[str, object] | None = None

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "run_id": self.run_id,
            "status": self.status,
            "record_path": str(self.record_path),
            "commands": list(self.commands),
            "return_code": self.return_code,
        }
        if self.error:
            payload["error"] = self.error
        if self.rollback:
            payload["rollback"] = self.rollback
        return payload


@dataclass(slots=True)
class _ModuleInstallRunState:
    run_id: str
    run_dir: Path
    record_path: Path
    commands: tuple[dict[str, Any], ...]
    result_commands: tuple[str, ...]
    record_redactions: tuple[str, ...]
    record: dict[str, Any]


def default_installer_runtime_dir(database_url: str | None = None, *, cwd: Path | None = None) -> Path:
    root = cwd or Path.cwd()
    if database_url and database_url.startswith("sqlite:///"):
        raw_path = database_url[len("sqlite:///"):]
        if raw_path and raw_path != ":memory:":
            db_path = Path(raw_path).expanduser()
            if not db_path.is_absolute():
                db_path = root / db_path
            return db_path.parent / "module-installer"
    return root / "runtime" / "module-installer"


def module_install_preflight(
    *,
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
    current_enabled: Iterable[str],
    desired_enabled: Iterable[str],
    maintenance_mode: bool,
    session: object | None = None,
    webui_root: Path | None = None,
    runtime_dir: Path | None = None,
) -> ModuleInstallerPreflight:
    issues: list[ModuleInstallerIssue] = []
    current = {item for item in current_enabled}
    desired_sequence = tuple(dict.fromkeys(str(item).strip() for item in desired_enabled if str(item).strip()))
    desired = set(desired_sequence)
    planned_items = tuple(item for item in plan.items if item.status == "planned")
    dependents = module_dependents(available)
    target_plan = _module_install_target_plan(plan, available)
    migration_plan, migration_issues = _module_migration_execution_plan(
        plan=plan,
        available=available,
        desired_enabled=desired_sequence,
    )

    if not planned_items:
        issues.append(ModuleInstallerIssue("warning", "empty_plan", "No planned package changes are present."))
    if not maintenance_mode:
        issues.append(ModuleInstallerIssue("blocker", "maintenance_required", "Package changes require maintenance mode."))
    if os.getenv("GOVOPLAN_STATE_PROFILE", "local").strip().lower() == "shared":
        issues.append(
            ModuleInstallerIssue(
                "blocker",
                "immutable_cluster_release_required",
                (
                    "Shared-state deployments cannot mutate packages on one runtime node. "
                    "Build and roll out one immutable release image across every API, worker, "
                    "and scheduler node."
                ),
            )
        )

    activation_candidates = desired_modules_after_package_plan(desired_sequence, plan)
    issues.extend(module_manifest_compatibility_issues(available, module_ids=activation_candidates))
    issues.extend(_package_catalog_preflight_issues(plan, available))
    issues.extend(migration_issues)

    frontend_rebuild_required = False
    for item in planned_items:
        item_issues, item_frontend_rebuild_required = _planned_item_preflight_issues(
            item,
            current=current,
            desired=desired,
            dependents=dependents,
            available=available,
            session=session,
        )
        issues.extend(item_issues)
        frontend_rebuild_required = frontend_rebuild_required or item_frontend_rebuild_required

    artifact_integrity, artifact_integrity_issues = _verify_artifact_integrity(
        planned_items,
        require_verified=_configured_require_artifact_integrity(),
    )
    issues.extend(artifact_integrity_issues)

    issues.extend(_frontend_rebuild_preflight_issues(frontend_rebuild_required=frontend_rebuild_required, webui_root=webui_root))

    record_dir = runtime_dir or default_installer_runtime_dir()
    rollback_commands = (
        f"govoplan-module-installer --runtime-dir {record_dir} --rollback <run-id>",
    )
    blocker = any(issue.severity == "blocker" for issue in issues)
    checklist = _post_install_checklist(
        planned_items=planned_items,
        maintenance_mode=maintenance_mode,
        frontend_rebuild_required=frontend_rebuild_required,
        migration_plan=migration_plan,
        blocker=blocker,
    )
    return ModuleInstallerPreflight(
        allowed=not blocker and bool(planned_items),
        maintenance_mode=maintenance_mode,
        frontend_rebuild_required=frontend_rebuild_required,
        restart_required=bool(planned_items),
        commands=module_install_plan_commands(planned_items),
        rollback_commands=rollback_commands,
        issues=tuple(issues),
        checklist=checklist,
        target_plan=target_plan,
        migration_plan=migration_plan,
        artifact_integrity=artifact_integrity,
    )


def _planned_item_preflight_issues(
    item: ModuleInstallPlanItem,
    *,
    current: set[str],
    desired: set[str],
    dependents: Mapping[str, tuple[str, ...]],
    available: Mapping[str, ModuleManifest],
    session: object | None,
) -> tuple[tuple[ModuleInstallerIssue, ...], bool]:
    issues: list[ModuleInstallerIssue] = []
    if item.action == "uninstall":
        issues.extend(_uninstall_guard_issues(
            item,
            current=current,
            desired=desired,
            dependents=dependents,
            available=available,
            session=session,
        ))
    if item.action in PACKAGE_TARGET_ACTIONS:
        issues.extend(_package_target_action_preflight_issues(item, available=available))
    return tuple(issues), bool(item.webui_package or item.webui_ref)


def _package_target_action_preflight_issues(
    item: ModuleInstallPlanItem,
    *,
    available: Mapping[str, ModuleManifest],
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    if item.action == "update" and item.module_id not in available:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "update_module_missing",
            "Package updates require the module to be installed; use install for new modules.",
            item.module_id,
        ))
    if item.module_id not in available:
        issues.append(ModuleInstallerIssue(
            "info",
            "compatibility_checked_after_install",
            "Module manifest compatibility can be fully checked after the package is installed and discovered.",
            item.module_id,
        ))
    if item.python_ref and not item.python_package:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "python_package_required",
            "Python installs must include the distribution package name so rollback can uninstall newly added packages.",
            item.module_id,
        ))
    if item.python_ref and not _looks_pinned_dependency_ref(item.python_ref):
        issues.append(ModuleInstallerIssue(
            "blocker",
            "unpinned_python_ref",
            "Python install refs must be pinned to an exact version or tagged git ref.",
            item.module_id,
        ))
    if item.webui_ref and not _looks_pinned_dependency_ref(item.webui_ref):
        issues.append(ModuleInstallerIssue(
            "blocker",
            "unpinned_webui_ref",
            "WebUI install refs must be pinned to an exact version or tagged git ref.",
            item.module_id,
        ))
    return tuple(issues)


def _frontend_rebuild_preflight_issues(
    *,
    frontend_rebuild_required: bool,
    webui_root: Path | None,
) -> tuple[ModuleInstallerIssue, ...]:
    if not frontend_rebuild_required:
        return ()
    if webui_root is None:
        return (ModuleInstallerIssue("blocker", "webui_root_missing", "WebUI package changes need a webui root for npm install/build."),)
    if not (webui_root / "package.json").exists():
        return (ModuleInstallerIssue("blocker", "webui_package_missing", f"WebUI package.json does not exist at {webui_root}."),)
    return ()


def run_module_install_plan(
    *,
    session: Session,
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
    current_enabled: Iterable[str],
    desired_enabled: Iterable[str],
    database_url: str,
    runtime_dir: Path | None = None,
    webui_root: Path | None = None,
    npm_bin: str = "npm",
    build_webui: bool = False,
    migrate_database: bool = False,
    database_backup_command: str | None = None,
    database_restore_command: str | None = None,
    database_restore_check_command: str | None = None,
    activate_installed_modules: bool = True,
    remove_uninstalled_modules_from_desired: bool = True,
    dry_run: bool = False,
    request_context: Mapping[str, object] | None = None,
) -> ModuleInstallerRunResult:
    maintenance_mode = saved_maintenance_mode(session)
    effective_runtime_dir = runtime_dir or default_installer_runtime_dir(database_url)
    preflight = module_install_preflight(
        plan=plan,
        available=available,
        current_enabled=current_enabled,
        desired_enabled=desired_enabled,
        maintenance_mode=maintenance_mode.enabled,
        session=session,
        webui_root=webui_root,
        runtime_dir=effective_runtime_dir,
    )
    if not preflight.allowed:
        raise ModuleInstallerError("Install preflight is blocked: " + "; ".join(issue.message for issue in preflight.issues if issue.severity == "blocker"))

    state = _prepare_module_install_run(
        plan=plan,
        preflight=preflight,
        database_url=database_url,
        effective_runtime_dir=effective_runtime_dir,
        webui_root=webui_root,
        npm_bin=npm_bin,
        build_webui=build_webui,
        migrate_database=migrate_database,
        database_backup_command=database_backup_command,
        database_restore_command=database_restore_command,
        database_restore_check_command=database_restore_check_command,
        activate_installed_modules=activate_installed_modules,
        remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
        dry_run=dry_run,
        request_context=request_context,
    )

    if dry_run:
        return ModuleInstallerRunResult(run_id=state.run_id, status="dry-run", record_path=state.record_path, commands=state.result_commands)

    executed, failed_error = _execute_module_install_run(
        session=session,
        plan=plan,
        available=available,
        effective_runtime_dir=effective_runtime_dir,
        state=state,
    )

    if failed_error is not None:
        return _failed_module_install_run_result(
            state=state,
            plan=plan,
            executed=executed,
            failed_error=failed_error,
            effective_runtime_dir=effective_runtime_dir,
            npm_bin=npm_bin,
            build_webui=build_webui,
            database_restore_command=database_restore_command,
            database_url=database_url,
        )

    return _applied_module_install_run_result(
        session=session,
        plan=plan,
        desired_enabled=desired_enabled,
        activate_installed_modules=activate_installed_modules,
        remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
        executed=executed,
        state=state,
    )


def _prepare_module_install_run(
    *,
    plan: ModuleInstallPlan,
    preflight: ModuleInstallerPreflight,
    database_url: str,
    effective_runtime_dir: Path,
    webui_root: Path | None,
    npm_bin: str,
    build_webui: bool,
    migrate_database: bool,
    database_backup_command: str | None,
    database_restore_command: str | None,
    database_restore_check_command: str | None,
    activate_installed_modules: bool,
    remove_uninstalled_modules_from_desired: bool,
    dry_run: bool,
    request_context: Mapping[str, object] | None,
) -> _ModuleInstallRunState:
    run_id = _run_id()
    run_dir = effective_runtime_dir / "runs" / run_id
    _create_private_installer_run_dir(run_dir)
    commands = structured_install_commands(
        plan,
        webui_root=webui_root,
        npm_bin=npm_bin,
        build_webui=build_webui,
        migrate_database_url=database_url if migrate_database else None,
        migration_enabled_modules=preflight.migration_plan.enabled_modules if migrate_database else (),
        migration_module_order=tuple(step.module_id for step in preflight.migration_plan.steps if step.phase == "upgrade") if migrate_database else (),
        migration_task_record_path=run_dir / "migration-tasks.json" if migrate_database and preflight.migration_plan.tasks else None,
        verify_modules=True,
    )
    record_redactions = _installer_secret_redactions(database_url)
    record = _initial_module_install_record(
        run_id=run_id,
        plan=plan,
        preflight=preflight,
        commands=commands,
        record_redactions=record_redactions,
        snapshot=_snapshot_environment(
            run_dir,
            webui_root=webui_root,
            database_url=database_url,
            include_database=migrate_database or _destructive_retirement_requested(plan),
            database_backup_command=database_backup_command,
            database_restore_command=database_restore_command,
            database_restore_check_command=database_restore_check_command,
        ),
        build_webui=build_webui,
        migrate_database=migrate_database,
        activate_installed_modules=activate_installed_modules,
        remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
        dry_run=dry_run,
        request_context=request_context,
    )
    record_path = run_dir / "record.json"
    _write_json(record_path, record)
    return _ModuleInstallRunState(
        run_id=run_id,
        run_dir=run_dir,
        record_path=record_path,
        commands=commands,
        result_commands=_command_displays(commands, redactions=record_redactions),
        record_redactions=record_redactions,
        record=record,
    )


def _initial_module_install_record(
    *,
    run_id: str,
    plan: ModuleInstallPlan,
    preflight: ModuleInstallerPreflight,
    commands: tuple[dict[str, Any], ...],
    record_redactions: tuple[str, ...],
    snapshot: dict[str, object],
    build_webui: bool,
    migrate_database: bool,
    activate_installed_modules: bool,
    remove_uninstalled_modules_from_desired: bool,
    dry_run: bool,
    request_context: Mapping[str, object] | None,
) -> dict[str, Any]:
    record: dict[str, Any] = {
        "run_id": run_id,
        "started_at": datetime.now(tz=UTC).isoformat(),
        "status": "dry-run" if dry_run else "running",
        "plan": [item.as_dict() for item in plan.items],
        "preflight": preflight.as_dict(),
        "snapshot": snapshot,
        "commands": [_command_record(command, redactions=record_redactions) for command in commands],
        "build_webui": build_webui,
        "migrate_database": migrate_database,
        "activate_installed_modules": activate_installed_modules,
        "remove_uninstalled_modules_from_desired": remove_uninstalled_modules_from_desired,
    }
    if request_context:
        record["request"] = dict(request_context)
    return record


def _execute_module_install_run(
    *,
    session: Session,
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
    effective_runtime_dir: Path,
    state: _ModuleInstallRunState,
) -> tuple[list[dict[str, object]], str | None]:
    executed: list[dict[str, object]] = []
    failed_error: str | None = None
    with _installer_lock(effective_runtime_dir):
        try:
            _execute_module_install_retirements(session=session, plan=plan, available=available, state=state)
            for command in state.commands:
                executed.append(_run_module_install_command(command, state=state))
                state.record["commands"] = executed
                _write_json(state.record_path, state.record)
        except Exception as exc:
            failed_error = _redact_installer_text(str(exc), redactions=state.record_redactions)
            _rollback_session_after_module_install_error(session, exc)
    return executed, failed_error


def _execute_module_install_retirements(
    *,
    session: Session,
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
    state: _ModuleInstallRunState,
) -> None:
    retirements = _execute_destructive_retirements(session=session, plan=plan, available=available)
    if not retirements:
        return
    session.commit()
    state.record["retirements"] = retirements
    _write_json(state.record_path, state.record)


def _run_module_install_command(command: Mapping[str, Any], *, state: _ModuleInstallRunState) -> dict[str, object]:
    result = _run_structured_installer_command(command)
    command_record: dict[str, object] = {
        **_command_record(command, redactions=state.record_redactions),
        "return_code": result.returncode,
        "stdout": _redact_installer_text(result.stdout[-8000:], redactions=state.record_redactions),
        "stderr": _redact_installer_text(result.stderr[-8000:], redactions=state.record_redactions),
    }
    migration_task_record_path = command.get("migration_task_record_path")
    if migration_task_record_path:
        task_records = _read_json_file(Path(str(migration_task_record_path)))
        if isinstance(task_records, list):
            command_record["migration_tasks"] = task_records
            state.record["migration_tasks"] = task_records
    if result.returncode != 0:
        display = _redact_installer_text(str(command["display"]), redactions=state.record_redactions)
        raise ModuleInstallerError(f"Command failed ({result.returncode}): {display}")
    return command_record


def _rollback_session_after_module_install_error(session: Session, exc: Exception) -> None:
    try:
        session.rollback()
    except Exception as rollback_exc:
        logger.debug("Module installer rollback failed after run error: %s", rollback_exc, exc_info=True)


def _failed_module_install_run_result(
    *,
    state: _ModuleInstallRunState,
    plan: ModuleInstallPlan,
    executed: list[dict[str, object]],
    failed_error: str,
    effective_runtime_dir: Path,
    npm_bin: str,
    build_webui: bool,
    database_restore_command: str | None,
    database_url: str,
) -> ModuleInstallerRunResult:
    state.record.update({
        "status": "failed",
        "finished_at": datetime.now(tz=UTC).isoformat(),
        "error": failed_error,
        "commands": executed,
    })
    _write_json(state.record_path, state.record)
    if not _destructive_retirement_requested(plan):
        return ModuleInstallerRunResult(
            run_id=state.run_id,
            status="failed",
            record_path=state.record_path,
            commands=state.result_commands,
            return_code=1,
            error=failed_error,
        )
    rollback = rollback_module_install_run(
        run_id=state.run_id,
        runtime_dir=effective_runtime_dir,
        npm_bin=npm_bin,
        build_webui=build_webui,
        database_restore_command=database_restore_command,
        database_url=database_url,
    )
    _update_run_record(state.record_path, {"destructive_retirement_rollback": rollback.as_dict()})
    return ModuleInstallerRunResult(
        run_id=state.run_id,
        status="rolled-back" if rollback.return_code == 0 else "failed",
        record_path=state.record_path,
        commands=state.result_commands,
        return_code=1,
        error=failed_error,
        rollback=rollback.as_dict(),
    )


def _applied_module_install_run_result(
    *,
    session: Session,
    plan: ModuleInstallPlan,
    desired_enabled: Iterable[str],
    activate_installed_modules: bool,
    remove_uninstalled_modules_from_desired: bool,
    executed: list[dict[str, object]],
    state: _ModuleInstallRunState,
) -> ModuleInstallerRunResult:
    save_module_install_plan(session, tuple(_mark_applied(item) for item in plan.items))
    if activate_installed_modules or remove_uninstalled_modules_from_desired:
        next_desired = desired_modules_after_package_plan(
            desired_enabled,
            plan,
            activate_installed_modules=activate_installed_modules,
            remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
        )
        save_desired_enabled_modules(session, next_desired)
        state.record["desired_enabled_after"] = list(next_desired)
    session.commit()
    state.record.update({
        "status": "applied",
        "finished_at": datetime.now(tz=UTC).isoformat(),
        "commands": executed,
    })
    _write_json(state.record_path, state.record)
    return ModuleInstallerRunResult(run_id=state.run_id, status="applied", record_path=state.record_path, commands=state.result_commands)


def supervise_module_install_plan(
    *,
    session: Session,
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
    current_enabled: Iterable[str],
    desired_enabled: Iterable[str],
    database_url: str,
    runtime_dir: Path | None = None,
    webui_root: Path | None = None,
    npm_bin: str = "npm",
    build_webui: bool = False,
    migrate_database: bool = False,
    database_backup_command: str | None = None,
    database_restore_command: str | None = None,
    database_restore_check_command: str | None = None,
    activate_installed_modules: bool = True,
    remove_uninstalled_modules_from_desired: bool = True,
    restart_command: str | None = None,
    restart_commands: Iterable[str] | None = None,
    health_url: str | None = None,
    health_urls: Iterable[str] | None = None,
    health_timeout_seconds: float = 60.0,
    health_interval_seconds: float = 2.0,
    request_context: Mapping[str, object] | None = None,
) -> ModuleInstallerRunResult:
    effective_runtime_dir = runtime_dir or default_installer_runtime_dir(database_url)
    effective_restart_commands = _command_list(restart_command, restart_commands)
    effective_health_urls = _command_list(health_url, health_urls)
    result = run_module_install_plan(
        session=session,
        plan=plan,
        available=available,
        current_enabled=current_enabled,
        desired_enabled=desired_enabled,
        database_url=database_url,
        runtime_dir=effective_runtime_dir,
        webui_root=webui_root,
        npm_bin=npm_bin,
        build_webui=build_webui,
        migrate_database=migrate_database,
        database_backup_command=database_backup_command,
        database_restore_command=database_restore_command,
        database_restore_check_command=database_restore_check_command,
        activate_installed_modules=activate_installed_modules,
        remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
        dry_run=False,
        request_context=request_context,
    )
    supervisor: dict[str, object] = {
        "started_at": datetime.now(tz=UTC).isoformat(),
        "restart_commands": list(effective_restart_commands),
        "health_urls": list(effective_health_urls),
    }
    _update_run_record(result.record_path, {"supervisor": supervisor})

    if result.return_code != 0:
        if result.rollback is not None:
            return result
        return _rollback_after_supervisor_failure(
            result=result,
            session=session,
            original_plan=plan,
            original_desired_enabled=tuple(desired_enabled),
            runtime_dir=effective_runtime_dir,
            npm_bin=npm_bin,
            restart_commands=effective_restart_commands,
            health_urls=effective_health_urls,
            health_timeout_seconds=health_timeout_seconds,
            health_interval_seconds=health_interval_seconds,
            build_webui=build_webui,
            reason=result.error or "Installer command failed.",
        )

    restart_results = _run_restart_commands(effective_restart_commands)
    if restart_results:
        supervisor["restart_commands"] = restart_results
        _update_run_record(result.record_path, {"supervisor": supervisor})
        if any(item.get("return_code") != 0 for item in restart_results):
            return _rollback_after_supervisor_failure(
                result=result,
                session=session,
                original_plan=plan,
                original_desired_enabled=tuple(desired_enabled),
                runtime_dir=effective_runtime_dir,
                npm_bin=npm_bin,
                restart_commands=effective_restart_commands,
                health_urls=effective_health_urls,
                health_timeout_seconds=health_timeout_seconds,
                health_interval_seconds=health_interval_seconds,
                build_webui=build_webui,
                reason="Restart command failed.",
            )

    if effective_health_urls:
        health = _wait_for_health_urls(
            effective_health_urls,
            timeout_seconds=health_timeout_seconds,
            interval_seconds=health_interval_seconds,
        )
        supervisor["health"] = health
        _update_run_record(result.record_path, {"supervisor": supervisor})
        if not health.get("ok"):
            return _rollback_after_supervisor_failure(
                result=result,
                session=session,
                original_plan=plan,
                original_desired_enabled=tuple(desired_enabled),
                runtime_dir=effective_runtime_dir,
                npm_bin=npm_bin,
                restart_commands=effective_restart_commands,
                health_urls=effective_health_urls,
                health_timeout_seconds=health_timeout_seconds,
                health_interval_seconds=health_interval_seconds,
                build_webui=build_webui,
                reason=str(health.get("error") or "Health check failed."),
            )

    supervisor.update({
        "status": "ok",
        "finished_at": datetime.now(tz=UTC).isoformat(),
    })
    _update_run_record(result.record_path, {"supervisor": supervisor})
    return result


def rollback_module_install_run(
    *,
    run_id: str,
    runtime_dir: Path,
    npm_bin: str = "npm",
    build_webui: bool = False,
    database_restore_command: str | None = None,
    database_url: str | None = None,
) -> ModuleInstallerRunResult:
    run_dir = runtime_dir / "runs" / run_id
    record_path = run_dir / "record.json"
    if not record_path.exists():
        raise ModuleInstallerError(f"Run record does not exist: {record_path}")
    record = json.loads(record_path.read_text(encoding="utf-8"))
    snapshot = record.get("snapshot") if isinstance(record.get("snapshot"), dict) else {}
    raw_database_backup = snapshot.get("database_backup") if isinstance(snapshot.get("database_backup"), Mapping) else {}
    rollback_redactions = _installer_secret_redactions(database_url or _database_url_from_external_snapshot(run_dir, raw_database_backup))
    commands = _rollback_commands_from_snapshot(run_dir=run_dir, record=record, snapshot=snapshot, npm_bin=npm_bin, build_webui=build_webui)

    with _installer_lock(runtime_dir):
        executed, command_failed = _execute_rollback_commands(commands=commands, redactions=rollback_redactions)
        if command_failed:
            record.update({"rollback_status": "failed", "rollback_commands": executed})
            _write_json(record_path, record)
            return ModuleInstallerRunResult(run_id=run_id, status="failed", record_path=record_path, commands=_command_displays(commands, redactions=rollback_redactions), return_code=1)
        database_restore = _restore_database_snapshot(
            run_dir,
            snapshot,
            database_restore_command=database_restore_command,
            database_url=database_url,
        )
        if database_restore is not None and not database_restore.get("restored"):
            record.update({
                "rollback_status": "failed",
                "rollback_commands": executed,
                "rollback_database_restore": database_restore,
            })
            _write_json(record_path, record)
            return ModuleInstallerRunResult(run_id=run_id, status="failed", record_path=record_path, commands=_command_displays(commands, redactions=rollback_redactions), return_code=1, error=str(database_restore.get("error") or "Database restore failed."))

    record.update({
        "rollback_status": "rolled-back",
        "rolled_back_at": datetime.now(tz=UTC).isoformat(),
        "rollback_commands": executed,
    })
    if database_restore is not None:
        record["rollback_database_restore"] = database_restore
    _write_json(record_path, record)
    return ModuleInstallerRunResult(run_id=run_id, status="rolled-back", record_path=record_path, commands=_command_displays(commands, redactions=rollback_redactions))


def _rollback_commands_from_snapshot(
    *,
    run_dir: Path,
    record: Mapping[str, object],
    snapshot: Mapping[str, object],
    npm_bin: str,
    build_webui: bool,
) -> list[dict[str, Any]]:
    commands = _rollback_python_commands(run_dir=run_dir, record=record, snapshot=snapshot)
    commands.extend(_rollback_webui_commands(run_dir=run_dir, record=record, snapshot=snapshot, npm_bin=npm_bin, build_webui=build_webui))
    return commands


def _rollback_python_commands(
    *,
    run_dir: Path,
    record: Mapping[str, object],
    snapshot: Mapping[str, object],
) -> list[dict[str, Any]]:
    commands = [
        _structured_command([sys.executable, "-m", "pip", "uninstall", "-y", package], source="rollback.python")
        for package in _planned_python_install_packages(record)
    ]
    pip_freeze = snapshot.get("pip_freeze")
    if isinstance(pip_freeze, str) and (run_dir / pip_freeze).exists():
        commands.append(_structured_command([sys.executable, "-m", "pip", "install", "-r", str(run_dir / pip_freeze)], source="rollback.python"))
    return commands


def _rollback_webui_commands(
    *,
    run_dir: Path,
    record: Mapping[str, object],
    snapshot: Mapping[str, object],
    npm_bin: str,
    build_webui: bool,
) -> list[dict[str, Any]]:
    webui_root = snapshot.get("webui_root")
    if not isinstance(webui_root, str):
        return []
    root = Path(webui_root)
    _restore_snapshot_file(run_dir, snapshot, "package_json", root / "package.json")
    _restore_snapshot_file(run_dir, snapshot, "package_lock", root / "package-lock.json")
    if not (root / "package.json").exists():
        return []
    commands = [_structured_command([npm_bin, "install"], cwd=root, source="rollback.webui")]
    if build_webui or bool(record.get("build_webui")):
        commands.append(_structured_command([npm_bin, "run", "build"], cwd=root, source="rollback.webui"))
    return commands


def _execute_rollback_commands(
    *,
    commands: list[dict[str, Any]],
    redactions: tuple[str, ...],
) -> tuple[list[dict[str, object]], bool]:
    executed: list[dict[str, object]] = []
    for command in commands:
        result = _run_structured_installer_command(command)
        executed.append({
            **_command_record(command, redactions=redactions),
            "return_code": result.returncode,
            "stdout": _redact_installer_text(result.stdout[-8000:], redactions=redactions),
            "stderr": _redact_installer_text(result.stderr[-8000:], redactions=redactions),
        })
        if result.returncode != 0:
            return executed, True
    return executed, False


def list_module_installer_runs(*, runtime_dir: Path, limit: int = 25) -> tuple[dict[str, object], ...]:
    runs_dir = runtime_dir / "runs"
    if not runs_dir.exists():
        return ()
    records: list[dict[str, object]] = []
    for record_path in sorted(runs_dir.glob("*/record.json"), reverse=True):
        record = _read_run_record(record_path)
        if record is None:
            continue
        records.append(_run_summary(record_path, record))
        if len(records) >= limit:
            break
    return tuple(records)


def read_module_installer_run(*, runtime_dir: Path, run_id: str) -> dict[str, object]:
    if not re.match(r"^[A-Za-z0-9_.:-]+$", run_id):
        raise ModuleInstallerError(f"Invalid installer run id: {run_id!r}")
    record_path = runtime_dir / "runs" / run_id / "record.json"
    record = _read_run_record(record_path)
    if record is None:
        raise ModuleInstallerError(f"Installer run record does not exist: {run_id}")
    record["record_path"] = str(record_path)
    return record


def module_installer_lock_status(*, runtime_dir: Path) -> dict[str, object]:
    lock_path = runtime_dir / "install.lock"
    if not lock_path.exists():
        return {"locked": False, "path": str(lock_path)}
    try:
        payload = json.loads(lock_path.read_text(encoding="utf-8"))
        if not isinstance(payload, dict):
            payload = {}
    except (OSError, json.JSONDecodeError):
        payload = {}
    return {"locked": True, "path": str(lock_path), **payload}


def module_installer_daemon_status(*, runtime_dir: Path) -> dict[str, object]:
    status_path = _daemon_status_path(runtime_dir)
    record = _read_run_record(status_path)
    if record is None:
        return {"running": False, "status": "unknown", "path": str(status_path)}
    pid = record.get("pid")
    running = isinstance(pid, int) and _process_is_running(pid)
    return {**record, "running": running, "path": str(status_path)}


def update_module_installer_daemon_status(*, runtime_dir: Path, patch: Mapping[str, object]) -> dict[str, object]:
    status_path = _daemon_status_path(runtime_dir)
    record = _read_run_record(status_path) or {}
    record.update({
        "pid": os.getpid(),
        "updated_at": datetime.now(tz=UTC).isoformat(),
        **patch,
    })
    _write_json(status_path, record)
    return module_installer_daemon_status(runtime_dir=runtime_dir)


def queue_module_installer_request(
    *,
    runtime_dir: Path,
    options: Mapping[str, object] | None = None,
    requested_by: str | None = None,
    tenant_id: str | None = None,
    retry_of: str | None = None,
    trace: Mapping[str, object] | None = None,
) -> dict[str, object]:
    request_id = _run_id()
    effective_trace = dict(trace) if trace is not None else _current_trace_payload()
    record = {
        "request_id": request_id,
        "status": "queued",
        "created_at": datetime.now(tz=UTC).isoformat(),
        "requested_by": requested_by,
        "tenant_id": tenant_id,
        "options": dict(options or {}),
    }
    if effective_trace:
        record["trace"] = effective_trace
    if retry_of:
        record["retry_of"] = retry_of
    with _request_queue_lock(runtime_dir):
        path = _request_record_path(runtime_dir, request_id)
        _write_json(path, record)
    return {**record, "record_path": str(path)}


def list_module_installer_requests(*, runtime_dir: Path, limit: int = 25) -> tuple[dict[str, object], ...]:
    requests_dir = runtime_dir / "requests"
    if not requests_dir.exists():
        return ()
    records: list[dict[str, object]] = []
    for record_path in sorted(requests_dir.glob("*.json"), reverse=True):
        record = _read_run_record(record_path)
        if record is None:
            continue
        record["record_path"] = str(record_path)
        records.append(record)
        if len(records) >= limit:
            break
    return tuple(records)


def read_module_installer_request(*, runtime_dir: Path, request_id: str) -> dict[str, object]:
    if not re.match(r"^[A-Za-z0-9_.:-]+$", request_id):
        raise ModuleInstallerError(f"Invalid installer request id: {request_id!r}")
    record_path = _request_record_path(runtime_dir, request_id)
    record = _read_run_record(record_path)
    if record is None:
        raise ModuleInstallerError(f"Installer request record does not exist: {request_id}")
    record["record_path"] = str(record_path)
    return record


def cancel_module_installer_request(
    *,
    runtime_dir: Path,
    request_id: str,
    cancelled_by: str | None = None,
) -> dict[str, object]:
    with _request_queue_lock(runtime_dir):
        record = read_module_installer_request(runtime_dir=runtime_dir, request_id=request_id)
        if record.get("status") != "queued":
            raise ModuleInstallerError("Only queued installer requests can be cancelled.")
        record.update({
            "status": "cancelled",
            "cancelled_at": datetime.now(tz=UTC).isoformat(),
            "cancelled_by": cancelled_by,
        })
        path = _request_record_path(runtime_dir, request_id)
        record.pop("record_path", None)
        _write_json(path, record)
    return read_module_installer_request(runtime_dir=runtime_dir, request_id=request_id)


def retry_module_installer_request(
    *,
    runtime_dir: Path,
    request_id: str,
    requested_by: str | None = None,
) -> dict[str, object]:
    record = read_module_installer_request(runtime_dir=runtime_dir, request_id=request_id)
    status = record.get("status")
    if status not in {"failed", "cancelled"}:
        raise ModuleInstallerError("Only failed or cancelled installer requests can be retried.")
    options = record.get("options") if isinstance(record.get("options"), Mapping) else {}
    return queue_module_installer_request(
        runtime_dir=runtime_dir,
        requested_by=requested_by,
        tenant_id=str(record.get("tenant_id")) if record.get("tenant_id") else None,
        options=options,
        retry_of=request_id,
    )


def claim_next_module_installer_request(*, runtime_dir: Path) -> dict[str, object] | None:
    with _request_queue_lock(runtime_dir):
        for record_path in sorted((runtime_dir / "requests").glob("*.json")):
            record = _read_run_record(record_path)
            if record is None or record.get("status") != "queued":
                continue
            record.update({
                "status": "running",
                "started_at": datetime.now(tz=UTC).isoformat(),
                "worker_pid": os.getpid(),
            })
            _write_json(record_path, record)
            return {**record, "record_path": str(record_path)}
    return None


def update_module_installer_request(*, runtime_dir: Path, request_id: str, patch: Mapping[str, object]) -> dict[str, object]:
    with _request_queue_lock(runtime_dir):
        record = read_module_installer_request(runtime_dir=runtime_dir, request_id=request_id)
        record.update(patch)
        path = _request_record_path(runtime_dir, request_id)
        record.pop("record_path", None)
        _write_json(path, record)
    return read_module_installer_request(runtime_dir=runtime_dir, request_id=request_id)


def structured_install_commands(
    plan: ModuleInstallPlan,
    *,
    webui_root: Path | None,
    npm_bin: str = "npm",
    build_webui: bool = False,
    migrate_database_url: str | None = None,
    migration_enabled_modules: Iterable[str] = (),
    migration_module_order: Iterable[str] = (),
    migration_task_record_path: Path | None = None,
    verify_modules: bool = False,
) -> tuple[dict[str, Any], ...]:
    """Build the only structured commands the installer may execute directly.

    Command provenance is intentionally narrow: Python package refs and WebUI refs
    come from the reviewed module install plan, WebUI install/build commands are
    fixed internal templates, and migration/verification commands call core-owned
    modules with argv lists. Operator-provided database backup/restore hooks use
    `_operator_command_argv` instead and are rejected when they contain shell
    operators, substitutions, redirects, or inline environment assignments.
    """
    commands: list[dict[str, Any]] = []
    webui_changed = False
    for item in plan.items:
        if item.status != "planned":
            continue
        item_commands, item_webui_changed = _structured_item_commands(item, webui_root=webui_root, npm_bin=npm_bin)
        commands.extend(item_commands)
        webui_changed = webui_changed or item_webui_changed
    commands.extend(_structured_webui_followup_commands(webui_changed=webui_changed, webui_root=webui_root, npm_bin=npm_bin, build_webui=build_webui))
    verify_command = _module_verify_command(plan) if verify_modules else None
    if verify_command is not None:
        commands.append(verify_command)
    if migrate_database_url is not None:
        commands.append(_structured_migration_command(
            migrate_database_url=migrate_database_url,
            migration_enabled_modules=migration_enabled_modules,
            migration_module_order=migration_module_order,
            migration_task_record_path=migration_task_record_path,
        ))
    return tuple(commands)


def _structured_item_commands(
    item: ModuleInstallPlanItem,
    *,
    webui_root: Path | None,
    npm_bin: str,
) -> tuple[tuple[dict[str, Any], ...], bool]:
    commands: list[dict[str, Any]] = []
    webui_changed = False
    if item.action in PACKAGE_TARGET_ACTIONS:
        if item.python_ref:
            commands.append(_structured_command([sys.executable, "-m", "pip", "install", item.python_ref], source="module-plan.python"))
        if item.webui_package and item.webui_ref and webui_root is not None:
            commands.append(_structured_command([npm_bin, "pkg", "set", f"dependencies.{item.webui_package}={item.webui_ref}"], cwd=webui_root, source="module-plan.webui"))
            webui_changed = True
    elif item.action == "uninstall":
        if item.python_package:
            commands.append(_structured_command([sys.executable, "-m", "pip", "uninstall", "-y", item.python_package], source="module-plan.python"))
        if item.webui_package and webui_root is not None:
            commands.append(_structured_command([npm_bin, "pkg", "delete", f"dependencies.{item.webui_package}"], cwd=webui_root, source="module-plan.webui"))
            webui_changed = True
    return tuple(commands), webui_changed


def _structured_webui_followup_commands(
    *,
    webui_changed: bool,
    webui_root: Path | None,
    npm_bin: str,
    build_webui: bool,
) -> tuple[dict[str, Any], ...]:
    if not webui_changed or webui_root is None:
        return ()
    commands = [_structured_command([npm_bin, "install"], cwd=webui_root, source="webui-dependencies")]
    if build_webui:
        commands.append(_structured_command([npm_bin, "run", "build"], cwd=webui_root, source="webui-build"))
    return tuple(commands)


def _structured_migration_command(
    *,
    migrate_database_url: str,
    migration_enabled_modules: Iterable[str],
    migration_module_order: Iterable[str],
    migration_task_record_path: Path | None,
) -> dict[str, Any]:
    migrate_command = [sys.executable, "-m", "govoplan_core.commands.init_db", "--database-url", migrate_database_url]
    for module_id in dict.fromkeys(str(item).strip() for item in migration_enabled_modules if str(item).strip()):
        migrate_command.extend(["--enabled-module", module_id])
    for module_id in dict.fromkeys(str(item).strip() for item in migration_module_order if str(item).strip()):
        migrate_command.extend(["--migration-module", module_id])
    if migration_task_record_path is None:
        return _structured_command(migrate_command, source="database-migration")
    migrate_command.extend(["--migration-task-record-output", str(migration_task_record_path)])
    command = _structured_command(migrate_command, source="database-migration")
    command["migration_task_record_path"] = str(migration_task_record_path)
    return command


def _destructive_retirement_requested(plan: ModuleInstallPlan) -> bool:
    return any(item.status == "planned" and item.action == "uninstall" and item.destroy_data for item in plan.items)


def _execute_destructive_retirements(
    *,
    session: Session,
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
) -> list[dict[str, object]]:
    return [
        _execute_destructive_retirement_item(session=session, item=item, available=available)
        for item in plan.items
        if item.status == "planned" and item.action == "uninstall" and item.destroy_data
    ]


def _execute_destructive_retirement_item(
    *,
    session: Session,
    item: ModuleInstallPlanItem,
    available: Mapping[str, ModuleManifest],
) -> dict[str, object]:
    manifest = available.get(item.module_id)
    migration = manifest.migration_spec if manifest is not None else None
    provider = migration.retirement_provider if migration is not None else None
    if provider is None:
        raise ModuleInstallerError(f"Module {item.module_id!r} does not provide a destructive retirement provider.")
    retirement_plan = provider(session, item.module_id)
    if not retirement_plan.supported:
        raise ModuleInstallerError(retirement_plan.blocking_reason or retirement_plan.summary or f"Module {item.module_id!r} cannot be retired.")
    if not retirement_plan.destroy_data_supported or retirement_plan.destroy_data_executor is None:
        raise ModuleInstallerError(f"Module {item.module_id!r} does not support destructive data retirement.")
    retirement_plan.destroy_data_executor(session, item.module_id)
    return {
        "module_id": item.module_id,
        "summary": retirement_plan.destroy_data_summary or retirement_plan.summary,
        "warnings": list(retirement_plan.destroy_data_warnings),
    }



class ModuleInstallerError(RuntimeError):
    pass


def _uninstall_guard_issues(
    item: ModuleInstallPlanItem,
    *,
    current: set[str],
    desired: set[str],
    dependents: Mapping[str, tuple[str, ...]],
    available: Mapping[str, ModuleManifest],
    session: object | None,
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    if item.module_id in PROTECTED_MODULES:
        issues.append(ModuleInstallerIssue("blocker", "protected_module", "Protected platform modules cannot be uninstalled.", item.module_id))
    if item.module_id in current:
        issues.append(ModuleInstallerIssue("blocker", "module_active", "Disable the module before uninstalling its package.", item.module_id))
    if item.module_id in desired:
        issues.append(ModuleInstallerIssue("blocker", "module_desired", "Remove the module from saved startup state before uninstalling its package.", item.module_id))
    active_dependents = sorted(module_id for module_id in dependents.get(item.module_id, ()) if module_id in current or module_id in desired)
    if active_dependents:
        issues.append(ModuleInstallerIssue("blocker", "active_dependents", "Disable dependent modules first: " + ", ".join(active_dependents), item.module_id))
    manifest = available.get(item.module_id)
    if manifest is not None:
        migration_issues, destroy_supported = _migration_uninstall_guard_issues(item, manifest=manifest, session=session)
        issues.extend(migration_issues)
        issues.extend(_provider_uninstall_guard_issues(item, manifest=manifest, session=session, destroy_supported=destroy_supported))
    return tuple(issues)


def _migration_uninstall_guard_issues(
    item: ModuleInstallPlanItem,
    *,
    manifest: ModuleManifest,
    session: object | None,
) -> tuple[tuple[ModuleInstallerIssue, ...], bool]:
    migration = manifest.migration_spec
    if migration is None:
        if item.destroy_data:
            return (
                (ModuleInstallerIssue("blocker", "migration_retirement_missing", "This module does not declare a data retirement provider.", item.module_id),),
                False,
            )
        return (), False
    if not migration.retirement_supported:
        return (
            (ModuleInstallerIssue(
                "blocker" if item.destroy_data else "warning",
                "migration_retirement_not_requested",
                "This module owns migrations. Default package uninstall is non-destructive: schema/data remain dormant until reinstall or explicit retirement support is added.",
                item.module_id,
            ),),
            False,
        )
    if migration.retirement_provider is None:
        detail = f" {migration.retirement_notes}" if migration.retirement_notes else ""
        return (
            (ModuleInstallerIssue(
                "blocker" if item.destroy_data else "warning",
                "migration_retirement_manual",
                "This module declares migration retirement support but no automated retirement provider is registered." + detail,
                item.module_id,
            ),),
            False,
        )
    try:
        retirement_plan = migration.retirement_provider(session, item.module_id)
    except Exception as exc:
        return (
            (ModuleInstallerIssue("blocker", "migration_retirement_check_failed", f"Migration retirement check failed: {exc}", item.module_id),),
            False,
        )
    destroy_supported = retirement_plan.destroy_data_supported and retirement_plan.destroy_data_executor is not None
    return _migration_retirement_issues(retirement_plan, item.module_id, destroy_data_requested=item.destroy_data), destroy_supported


def _provider_uninstall_guard_issues(
    item: ModuleInstallPlanItem,
    *,
    manifest: ModuleManifest,
    session: object | None,
    destroy_supported: bool,
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    for provider in manifest.uninstall_guard_providers:
        try:
            for result in provider(session, item.module_id):
                if item.destroy_data and destroy_supported and result.code == "persistent_data_present":
                    continue
                issues.append(ModuleInstallerIssue(result.severity, result.code, result.message, item.module_id))
        except Exception as exc:
            issues.append(ModuleInstallerIssue("blocker", "uninstall_guard_failed", f"Uninstall guard failed: {exc}", item.module_id))
    return tuple(issues)


def _migration_retirement_issues(plan: MigrationRetirementPlan, module_id: str, *, destroy_data_requested: bool = False) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    if not plan.supported:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "migration_retirement_blocked",
            plan.blocking_reason or plan.summary or "Migration retirement provider blocked package uninstall.",
            module_id,
        ))
    elif plan.summary:
        issues.append(ModuleInstallerIssue("info", "migration_retirement_supported", plan.summary, module_id))
    if destroy_data_requested:
        if not plan.destroy_data_supported or plan.destroy_data_executor is None:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "migration_retirement_destroy_unsupported",
                "Destroying module data was requested, but this module does not provide an automated destructive retirement executor.",
                module_id,
            ))
        else:
            issues.append(ModuleInstallerIssue(
                "warning",
                "migration_retirement_destroy_planned",
                plan.destroy_data_summary or "Destructive module data retirement will run before package uninstall.",
                module_id,
            ))
            for warning in plan.destroy_data_warnings:
                issues.append(ModuleInstallerIssue("warning", "migration_retirement_destroy_warning", warning, module_id))
    for warning in plan.warnings:
        issues.append(ModuleInstallerIssue("warning", "migration_retirement_warning", warning, module_id))
    return tuple(issues)


def _module_verify_command(plan: ModuleInstallPlan) -> dict[str, Any] | None:
    installed: list[str] = []
    absent: list[str] = []
    for item in plan.items:
        if item.status != "planned":
            continue
        if item.action in PACKAGE_TARGET_ACTIONS and item.python_ref:
            installed.append(item.module_id)
        elif item.action == "uninstall" and item.python_package:
            absent.append(item.module_id)
    if not installed and not absent:
        return None
    argv = [sys.executable, "-m", "govoplan_core.commands.module_verify"]
    for module_id in dict.fromkeys(installed):
        argv.extend(["--installed", module_id])
    for module_id in dict.fromkeys(absent):
        argv.extend(["--absent", module_id])
    return _structured_command(argv, source="module-verification")


def module_manifest_compatibility_issues(
    available: Mapping[str, ModuleManifest],
    *,
    module_ids: Iterable[str] | None = None,
) -> tuple[ModuleInstallerIssue, ...]:
    current_core_version = _current_core_version()
    issues: list[ModuleInstallerIssue] = []
    manifests = _selected_manifests(available, module_ids)
    for manifest in manifests.values():
        compatibility = manifest.compatibility
        if compatibility.manifest_contract_version != SUPPORTED_MANIFEST_CONTRACT_VERSION:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "unsupported_manifest_contract",
                (
                    f"Module manifest contract {compatibility.manifest_contract_version} is not supported by "
                    f"this core runtime ({SUPPORTED_MANIFEST_CONTRACT_VERSION})."
                ),
                manifest.id,
            ))
        if compatibility.core_version_min and compare_versions(current_core_version, compatibility.core_version_min) < 0:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "core_version_too_low",
                f"Module requires core >= {compatibility.core_version_min}; current core is {current_core_version}.",
                manifest.id,
            ))
        if compatibility.core_version_max and compare_versions(current_core_version, compatibility.core_version_max) > 0:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "core_version_too_high",
                f"Module supports core <= {compatibility.core_version_max}; current core is {current_core_version}.",
                manifest.id,
            ))
    issues.extend(_interface_contract_issues(manifests))
    return tuple(issues)


def _current_core_version() -> str:
    try:
        return metadata.version("govoplan-core")
    except metadata.PackageNotFoundError:
        pyproject_path = Path(__file__).resolve().parents[3] / "pyproject.toml"
        try:
            with pyproject_path.open("rb") as handle:
                project = tomllib.load(handle).get("project", {})
            version = project.get("version")
            if isinstance(version, str) and version:
                return version
        except (OSError, tomllib.TOMLDecodeError):
            return "0.0.0"
        return "0.0.0"


def _selected_manifests(
    available: Mapping[str, ModuleManifest],
    module_ids: Iterable[str] | None,
) -> dict[str, ModuleManifest]:
    if module_ids is None:
        return dict(available)
    selected: dict[str, ModuleManifest] = {}
    for module_id in module_ids:
        manifest = available.get(module_id)
        if manifest is not None:
            selected[module_id] = manifest
    return selected


def _interface_contract_issues(available: Mapping[str, ModuleManifest]) -> tuple[ModuleInstallerIssue, ...]:
    providers: dict[str, list[tuple[str, str]]] = defaultdict(list)
    for manifest in available.values():
        for provided in manifest.provides_interfaces:
            providers[provided.name].append((manifest.id, provided.version))

    issues: list[ModuleInstallerIssue] = []
    for manifest in available.values():
        for requirement in manifest.requires_interfaces:
            available_versions = providers.get(requirement.name, [])
            matching = [
                (module_id, version)
                for module_id, version in available_versions
                if version_satisfies_range(
                    version,
                    version_min=requirement.version_min,
                    version_max_exclusive=requirement.version_max_exclusive,
                )
            ]
            if matching:
                continue
            version_range = format_version_range(
                version_min=requirement.version_min,
                version_max_exclusive=requirement.version_max_exclusive,
            )
            if not available_versions:
                if requirement.optional:
                    continue
                issues.append(ModuleInstallerIssue(
                    "blocker",
                    "required_interface_missing",
                    f"Module requires interface {requirement.name!r} ({version_range}), but no provider is installed.",
                    manifest.id,
                ))
                continue
            provider_summary = ", ".join(f"{module_id}@{version}" for module_id, version in available_versions)
            issues.append(ModuleInstallerIssue(
                "blocker",
                "interface_version_mismatch",
                (
                    f"Module requires interface {requirement.name!r} ({version_range}), "
                    f"but installed providers are {provider_summary}."
                ),
                manifest.id,
            ))
    return tuple(issues)


def _package_catalog_preflight_issues(
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
) -> tuple[ModuleInstallerIssue, ...]:
    package_items = tuple(item for item in plan.items if item.status == "planned" and item.action in PACKAGE_TARGET_ACTIONS)
    if not package_items:
        return ()
    catalog_items = tuple(item for item in package_items if item.source == "catalog")
    try:
        from govoplan_core.core.module_package_catalog import validate_module_package_catalog

        result = validate_module_package_catalog()
    except Exception as exc:
        return _catalog_validation_exception_issues(exc, catalog_items=bool(catalog_items))
    issues = list(_catalog_validation_result_issues(result, catalog_items=bool(catalog_items)))
    if issues or not result.get("configured"):
        return tuple(issues)
    issues.extend(_catalog_warning_issues(result))
    if catalog_items:
        issues.extend(_selected_catalog_interface_issues(catalog_items, result, available))
    return tuple(issues)


def _catalog_validation_exception_issues(exc: Exception, *, catalog_items: bool) -> tuple[ModuleInstallerIssue, ...]:
    severity: IssueSeverity = "blocker" if catalog_items else "warning"
    return (ModuleInstallerIssue(
        severity,
        "catalog_validation_failed",
        f"Package catalog could not be validated for this install plan: {exc}",
    ),)


def _catalog_validation_result_issues(
    result: Mapping[str, object],
    *,
    catalog_items: bool,
) -> tuple[ModuleInstallerIssue, ...]:
    if not result.get("configured"):
        if catalog_items:
            return (ModuleInstallerIssue(
                "blocker",
                "catalog_required",
                "Catalog-sourced package installs require a configured package catalog before activation.",
            ),)
        return ()
    if result.get("valid") is True:
        return ()
    severity: IssueSeverity = "blocker" if catalog_items else "warning"
    return (ModuleInstallerIssue(
        severity,
        "catalog_validation_failed",
        str(result.get("error") or "Module package catalog is invalid."),
    ),)


def _catalog_warning_issues(result: Mapping[str, object]) -> tuple[ModuleInstallerIssue, ...]:
    warnings = result.get("warnings")
    if not isinstance(warnings, list):
        return ()
    return tuple(ModuleInstallerIssue("warning", "catalog_warning", str(warning)) for warning in warnings)


def _module_install_target_plan(
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
) -> tuple[ModuleInstallTargetItem, ...]:
    planned_items = tuple(item for item in plan.items if item.status == "planned")
    if not planned_items:
        return ()
    catalog_modules = _catalog_modules_for_target_plan(planned_items)
    return tuple(
        _module_install_target_item(item, available=available, catalog_modules=catalog_modules)
        for item in planned_items
    )


def _module_install_target_item(
    item: ModuleInstallPlanItem,
    *,
    available: Mapping[str, ModuleManifest],
    catalog_modules: Mapping[str, Mapping[str, object]],
) -> ModuleInstallTargetItem:
    manifest = available.get(item.module_id)
    catalog_module = catalog_modules.get(item.module_id) if item.source == "catalog" else None
    policy = _catalog_target_policy(item, catalog_module)
    return ModuleInstallTargetItem(
        module_id=item.module_id,
        action=item.action,
        source=item.source,
        current_version=manifest.version if manifest is not None else None,
        target_version=None if item.action == "uninstall" else _catalog_optional_string(catalog_module, "version"),
        python_package=item.python_package or _catalog_optional_string(catalog_module, "python_package"),
        python_ref=item.python_ref or _catalog_optional_string(catalog_module, "python_ref"),
        webui_package=item.webui_package or _catalog_optional_string(catalog_module, "webui_package"),
        webui_ref=item.webui_ref or _catalog_optional_string(catalog_module, "webui_ref"),
        migration_safety=str(policy["migration_safety"]),
        migration_notes=policy["migration_notes"] if isinstance(policy["migration_notes"], str) else None,
        current_version_min=policy["current_version_min"] if isinstance(policy["current_version_min"], str) else None,
        current_version_max_exclusive=policy["current_version_max_exclusive"] if isinstance(policy["current_version_max_exclusive"], str) else None,
        bridge_release=policy["bridge_release"] is True,
        bridge_notes=policy["bridge_notes"] if isinstance(policy["bridge_notes"], str) else None,
        allow_downgrade=policy["allow_downgrade"] is True,
        allow_same_version=policy["allow_same_version"] is True,
        recovery_tested=policy["recovery_tested"] is True,
        recovery_notes=policy["recovery_notes"] if isinstance(policy["recovery_notes"], str) else None,
        data_safety_acknowledged=item.data_safety_acknowledged,
    )


def _catalog_target_policy(
    item: ModuleInstallPlanItem,
    catalog_module: Mapping[str, object] | None,
) -> dict[str, object]:
    if catalog_module is None or item.action not in PACKAGE_TARGET_ACTIONS:
        return {
            "migration_safety": "automatic",
            "migration_notes": None,
            "current_version_min": None,
            "current_version_max_exclusive": None,
            "bridge_release": False,
            "bridge_notes": None,
            "allow_downgrade": False,
            "allow_same_version": False,
            "recovery_tested": False,
            "recovery_notes": None,
        }
    return {
        "migration_safety": _catalog_optional_string(catalog_module, "migration_safety") or "automatic",
        "migration_notes": _catalog_optional_string(catalog_module, "migration_notes"),
        "current_version_min": _catalog_optional_string(catalog_module, "current_version_min"),
        "current_version_max_exclusive": _catalog_optional_string(catalog_module, "current_version_max_exclusive"),
        "bridge_release": _catalog_optional_bool(catalog_module, "bridge_release"),
        "bridge_notes": _catalog_optional_string(catalog_module, "bridge_notes"),
        "allow_downgrade": _catalog_optional_bool(catalog_module, "allow_downgrade"),
        "allow_same_version": _catalog_optional_bool(catalog_module, "allow_same_version"),
        "recovery_tested": _catalog_optional_bool(catalog_module, "recovery_tested"),
        "recovery_notes": _catalog_optional_string(catalog_module, "recovery_notes"),
    }


def module_install_catalog_companion_module_ids(
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
    *,
    validation: Mapping[str, object] | None = None,
) -> tuple[str, ...]:
    planned_items = _catalog_companion_planned_items(plan)
    if not planned_items:
        return ()
    validation = _catalog_companion_validation(validation)
    if validation is None:
        return ()
    modules = _catalog_modules_by_id(validation)
    original_ids = {item.module_id for item in planned_items}
    planned_ids = set(original_ids)
    companions: list[str] = []
    for _round in range(20):
        missing = _catalog_companion_missing_ids(planned_ids=planned_ids, modules=modules, available=available)
        if not missing:
            break
        for module_id in sorted(missing):
            planned_ids.add(module_id)
            companions.append(module_id)
    return tuple(module_id for module_id in companions if module_id not in original_ids)


def _catalog_companion_planned_items(plan: ModuleInstallPlan) -> tuple[ModuleInstallPlanItem, ...]:
    return tuple(
        item
        for item in plan.items
        if item.status == "planned" and item.source == "catalog" and item.action in PACKAGE_TARGET_ACTIONS
    )


def _catalog_companion_validation(validation: Mapping[str, object] | None) -> Mapping[str, object] | None:
    if validation is None:
        try:
            from govoplan_core.core.module_package_catalog import validate_module_package_catalog

            validation = validate_module_package_catalog()
        except Exception:
            return None
    return validation if validation.get("valid") is True else None


def _catalog_companion_missing_ids(
    *,
    planned_ids: set[str],
    modules: Mapping[str, Mapping[str, object]],
    available: Mapping[str, ModuleManifest],
) -> set[str]:
    target_provider_versions = _catalog_target_provider_versions_for_ids(
        planned_ids=planned_ids,
        modules=modules,
        available=available,
    )
    missing = set()
    missing.update(_missing_catalog_companions_for_planned_ids(
        planned_ids=planned_ids,
        modules=modules,
        available=available,
        target_provider_versions=target_provider_versions,
    ))
    missing.update(_installed_modules_needing_catalog_companion_updates(
        planned_ids=planned_ids,
        modules=modules,
        available=available,
        target_provider_versions=target_provider_versions,
    ))
    return {module_id for module_id in missing if module_id not in planned_ids and module_id in modules}


def _missing_catalog_companions_for_planned_ids(
    *,
    planned_ids: set[str],
    modules: Mapping[str, Mapping[str, object]],
    available: Mapping[str, ModuleManifest],
    target_provider_versions: Mapping[str, tuple[tuple[str, str], ...]],
) -> set[str]:
    missing: set[str] = set()
    for module_id in sorted(planned_ids):
        module = modules.get(module_id)
        if module is None:
            continue
        missing.update(_missing_catalog_dependency_ids(module, planned_ids=planned_ids, modules=modules, available=available))
        missing.update(_missing_catalog_requirement_provider_ids(module, planned_ids=planned_ids, modules=modules, target_provider_versions=target_provider_versions))
    return missing


def _missing_catalog_dependency_ids(
    module: Mapping[str, object],
    *,
    planned_ids: set[str],
    modules: Mapping[str, Mapping[str, object]],
    available: Mapping[str, ModuleManifest],
) -> set[str]:
    return {
        dependency_id
        for dependency_id in _catalog_string_list(module.get("dependencies"))
        if dependency_id not in planned_ids and dependency_id not in available and dependency_id in modules
    }


def _missing_catalog_requirement_provider_ids(
    module: Mapping[str, object],
    *,
    planned_ids: set[str],
    modules: Mapping[str, Mapping[str, object]],
    target_provider_versions: Mapping[str, tuple[tuple[str, str], ...]],
) -> set[str]:
    missing: set[str] = set()
    for requirement in _catalog_interface_items(module.get("requires_interfaces")):
        if _catalog_requirement_satisfied(requirement, target_provider_versions):
            continue
        name = requirement.get("name") if isinstance(requirement.get("name"), str) else None
        if not name:
            continue
        missing.update(_catalog_provider_suggestions(
            name=name,
            version_min=requirement.get("version_min") if isinstance(requirement.get("version_min"), str) else None,
            version_max_exclusive=(
                requirement.get("version_max_exclusive")
                if isinstance(requirement.get("version_max_exclusive"), str)
                else None
            ),
            catalog_modules=modules,
            planned_ids=planned_ids,
        ))
    return missing


def _installed_modules_needing_catalog_companion_updates(
    *,
    planned_ids: set[str],
    modules: Mapping[str, Mapping[str, object]],
    available: Mapping[str, ModuleManifest],
    target_provider_versions: Mapping[str, tuple[tuple[str, str], ...]],
) -> set[str]:
    missing: set[str] = set()
    for manifest in available.values():
        if manifest.id in planned_ids:
            continue
        if _manifest_requirements_satisfied(manifest, target_provider_versions):
            continue
        candidate_update = modules.get(manifest.id)
        if candidate_update is None:
            continue
        candidate_requirements = _catalog_interface_items(candidate_update.get("requires_interfaces"))
        if _catalog_requirements_satisfied(candidate_requirements, target_provider_versions):
            missing.add(manifest.id)
    return missing


def _manifest_requirements_satisfied(
    manifest: ModuleManifest,
    target_provider_versions: Mapping[str, tuple[tuple[str, str], ...]],
) -> bool:
    requirements = tuple(
        {
            "name": requirement.name,
            "version_min": requirement.version_min,
            "version_max_exclusive": requirement.version_max_exclusive,
            "optional": requirement.optional,
        }
        for requirement in manifest.requires_interfaces
    )
    return all(_catalog_requirement_satisfied(requirement, target_provider_versions) for requirement in requirements)


def _module_migration_execution_plan(
    *,
    plan: ModuleInstallPlan,
    available: Mapping[str, ModuleManifest],
    desired_enabled: Iterable[str],
) -> tuple[ModuleMigrationExecutionPlan, tuple[ModuleInstallerIssue, ...]]:
    planned_items = tuple(item for item in plan.items if item.status == "planned")
    enabled_modules = desired_modules_after_package_plan(desired_enabled, plan)
    catalog_modules = _catalog_modules_for_target_plan(planned_items)
    nodes = _migration_plan_nodes(planned_items=planned_items, available=available, catalog_modules=catalog_modules)
    if not nodes:
        return ModuleMigrationExecutionPlan(enabled_modules=enabled_modules), ()

    issues: list[ModuleInstallerIssue] = []
    target_metadata = _migration_target_metadata_by_id(enabled_modules=enabled_modules, nodes=nodes, available=available, catalog_modules=catalog_modules)
    edges = _migration_execution_edges(
        nodes=nodes,
        target_metadata=target_metadata,
        enabled_modules=enabled_modules,
        available=available,
        issues=issues,
    )
    ordered_ids = _ordered_migration_execution_ids(edges=edges, nodes=nodes, issues=issues)

    steps = tuple(_migration_step_from_node(nodes[module_id]) for module_id in ordered_ids if module_id in nodes)
    tasks = _migration_task_plan_items(ordered_ids, target_metadata, nodes)
    issues.extend(_migration_execution_plan_issues(planned_items=planned_items, steps=steps, tasks=tasks))
    return ModuleMigrationExecutionPlan(enabled_modules=enabled_modules, steps=steps, tasks=tasks), tuple(issues)


def _migration_plan_nodes(
    *,
    planned_items: tuple[ModuleInstallPlanItem, ...],
    available: Mapping[str, ModuleManifest],
    catalog_modules: Mapping[str, Mapping[str, object]],
) -> dict[str, Mapping[str, object]]:
    return {
        item.module_id: _migration_node(
            item=item,
            manifest=available.get(item.module_id),
            catalog_module=catalog_modules.get(item.module_id),
        )
        for item in planned_items
        if _migration_step_relevant(item, available.get(item.module_id), catalog_modules.get(item.module_id))
    }


def _migration_target_metadata_by_id(
    *,
    enabled_modules: tuple[str, ...],
    nodes: Mapping[str, Mapping[str, object]],
    available: Mapping[str, ModuleManifest],
    catalog_modules: Mapping[str, Mapping[str, object]],
) -> dict[str, Mapping[str, object]]:
    return {
        module_id: _migration_target_metadata(
            module_id=module_id,
            manifest=available.get(module_id),
            catalog_module=catalog_modules.get(module_id),
        )
        for module_id in set(enabled_modules) | set(nodes)
    }


def _migration_execution_edges(
    *,
    nodes: Mapping[str, Mapping[str, object]],
    target_metadata: Mapping[str, Mapping[str, object]],
    enabled_modules: tuple[str, ...],
    available: Mapping[str, ModuleManifest],
    issues: list[ModuleInstallerIssue],
) -> dict[str, set[str]]:
    edges: dict[str, set[str]] = {module_id: set() for module_id in nodes}
    provider_modules = _migration_provider_modules(target_metadata)
    target_enabled = set(enabled_modules)
    for module_id, node in nodes.items():
        metadata = target_metadata.get(module_id)
        if metadata is None:
            continue
        _add_explicit_migration_edges(edges, issues, module_id=module_id, metadata=metadata, target_enabled=target_enabled, available=available)
        _add_dependency_migration_edges(edges, module_id=module_id, node=node, metadata=metadata, nodes=nodes)
        _add_interface_migration_edges(edges, module_id=module_id, node=node, metadata=metadata, nodes=nodes, provider_modules=provider_modules)
    return edges


def _add_explicit_migration_edges(
    edges: dict[str, set[str]],
    issues: list[ModuleInstallerIssue],
    *,
    module_id: str,
    metadata: Mapping[str, object],
    target_enabled: set[str],
    available: Mapping[str, ModuleManifest],
) -> None:
    for before_id in metadata["migration_before"]:
        _add_migration_edge(edges, issues, before=module_id, after=before_id, source=module_id, target_enabled=target_enabled, available=available)
    for after_id in metadata["migration_after"]:
        _add_migration_edge(edges, issues, before=after_id, after=module_id, source=module_id, target_enabled=target_enabled, available=available)


def _add_dependency_migration_edges(
    edges: dict[str, set[str]],
    *,
    module_id: str,
    node: Mapping[str, object],
    metadata: Mapping[str, object],
    nodes: Mapping[str, Mapping[str, object]],
) -> None:
    for dependency_id in [*metadata["dependencies"], *metadata["optional_dependencies"]]:
        if dependency_id not in nodes:
            continue
        _add_related_migration_edge(edges, module_id=module_id, related_id=dependency_id, node=node, related_node=nodes[dependency_id])


def _add_interface_migration_edges(
    edges: dict[str, set[str]],
    *,
    module_id: str,
    node: Mapping[str, object],
    metadata: Mapping[str, object],
    nodes: Mapping[str, Mapping[str, object]],
    provider_modules: Mapping[str, tuple[tuple[str, str], ...]],
) -> None:
    for requirement in metadata["requires_interfaces"]:
        for provider_id in _matching_migration_provider_ids(module_id=module_id, requirement=requirement, nodes=nodes, provider_modules=provider_modules):
            _add_related_migration_edge(edges, module_id=module_id, related_id=provider_id, node=node, related_node=nodes[provider_id])


def _matching_migration_provider_ids(
    *,
    module_id: str,
    requirement: Mapping[str, object],
    nodes: Mapping[str, Mapping[str, object]],
    provider_modules: Mapping[str, tuple[tuple[str, str], ...]],
) -> tuple[str, ...]:
    name = requirement.get("name") if isinstance(requirement.get("name"), str) else None
    if not name:
        return ()
    version_min = requirement.get("version_min") if isinstance(requirement.get("version_min"), str) else None
    version_max_exclusive = requirement.get("version_max_exclusive") if isinstance(requirement.get("version_max_exclusive"), str) else None
    return tuple(
        provider_id
        for provider_id, version in provider_modules.get(name, ())
        if provider_id in nodes
        and provider_id != module_id
        and version_satisfies_range(version, version_min=version_min, version_max_exclusive=version_max_exclusive)
    )


def _add_related_migration_edge(
    edges: dict[str, set[str]],
    *,
    module_id: str,
    related_id: str,
    node: Mapping[str, object],
    related_node: Mapping[str, object],
) -> None:
    if node["phase"] == "retirement" and related_node["phase"] == "retirement":
        edges[module_id].add(related_id)
    else:
        edges[related_id].add(module_id)


def _ordered_migration_execution_ids(
    *,
    edges: dict[str, set[str]],
    nodes: Mapping[str, Mapping[str, object]],
    issues: list[ModuleInstallerIssue],
) -> tuple[str, ...]:
    ordered_ids, cycle_ids = _topological_module_order(edges)
    if not cycle_ids:
        return ordered_ids
    cycle_text = ", ".join(cycle_ids)
    issues.append(ModuleInstallerIssue(
        "blocker",
        "migration_graph_cycle",
        f"Module migration ordering has a cycle involving: {cycle_text}. Adjust migration_after/migration_before metadata or publish a bridge release.",
    ))
    return tuple(sorted(nodes))


def _migration_execution_plan_issues(
    *,
    planned_items: tuple[ModuleInstallPlanItem, ...],
    steps: tuple[ModuleMigrationPlanStep, ...],
    tasks: tuple[ModuleMigrationTaskPlanItem, ...],
) -> tuple[ModuleInstallerIssue, ...]:
    issues = list(_migration_task_preflight_issues(tasks, {item.module_id: item for item in planned_items}))
    pending = [step.module_id for step in steps if step.metadata_pending]
    if pending:
        issues.append(ModuleInstallerIssue(
            "info",
            "migration_metadata_pending",
            "Migration metadata for newly installed manual package(s) can only be confirmed after package verification: " + ", ".join(pending) + ".",
        ))
    return tuple(issues)


def _migration_step_relevant(
    item: ModuleInstallPlanItem,
    manifest: ModuleManifest | None,
    catalog_module: Mapping[str, object] | None,
) -> bool:
    if item.action == "uninstall":
        return manifest is not None and manifest.migration_spec is not None
    if item.action not in PACKAGE_TARGET_ACTIONS:
        return False
    if manifest is not None and manifest.migration_spec is not None:
        return True
    if catalog_module is not None:
        safety = _catalog_optional_string(catalog_module, "migration_safety") or "automatic"
        if safety != "automatic":
            return True
        if _catalog_string_list(catalog_module.get("migration_after")) or _catalog_string_list(catalog_module.get("migration_before")):
            return True
        if _catalog_migration_task_items(catalog_module.get("migration_tasks")):
            return True
    return bool(item.python_ref and item.source == "manual")


def _migration_node(
    *,
    item: ModuleInstallPlanItem,
    manifest: ModuleManifest | None,
    catalog_module: Mapping[str, object] | None,
) -> dict[str, object]:
    has_manifest_metadata = manifest is not None and manifest.migration_spec is not None
    has_catalog_metadata = _catalog_module_has_migration_metadata(catalog_module)
    phase = "retirement" if item.action == "uninstall" else "upgrade"
    source = _migration_node_source(has_manifest_metadata=has_manifest_metadata, has_catalog_metadata=has_catalog_metadata)
    return {
        "module_id": item.module_id,
        "action": item.action,
        "phase": phase,
        "source": source,
        "has_migration_metadata": has_manifest_metadata or has_catalog_metadata,
        "metadata_pending": item.action in PACKAGE_TARGET_ACTIONS and source == "pending",
        "migration_safety": _catalog_optional_string(catalog_module, "migration_safety") or "automatic",
        "current_version": manifest.version if manifest is not None else None,
        "target_version": _catalog_optional_string(catalog_module, "version"),
        "reason": _migration_node_reason(item, has_manifest_metadata=has_manifest_metadata, has_catalog_metadata=has_catalog_metadata),
    }


def _catalog_module_has_migration_metadata(catalog_module: Mapping[str, object] | None) -> bool:
    if catalog_module is None:
        return False
    return (
        (_catalog_optional_string(catalog_module, "migration_safety") or "automatic") != "automatic"
        or bool(_catalog_string_list(catalog_module.get("migration_after")))
        or bool(_catalog_string_list(catalog_module.get("migration_before")))
        or bool(_catalog_migration_task_items(catalog_module.get("migration_tasks")))
    )


def _migration_node_source(*, has_manifest_metadata: bool, has_catalog_metadata: bool) -> str:
    if has_manifest_metadata:
        return "manifest"
    if has_catalog_metadata:
        return "catalog"
    return "pending"


def _migration_node_reason(
    item: ModuleInstallPlanItem,
    *,
    has_manifest_metadata: bool,
    has_catalog_metadata: bool,
) -> str:
    if item.action == "uninstall":
        return "Module owns migration metadata and may leave dormant schema behind."
    if has_manifest_metadata:
        return "Installed module owns migration metadata."
    if has_catalog_metadata:
        return "Catalog target declares migration metadata."
    return "Manual backend package may add migration metadata after installation."


def _migration_step_from_node(node: Mapping[str, object]) -> ModuleMigrationPlanStep:
    return ModuleMigrationPlanStep(
        module_id=str(node["module_id"]),
        action=str(node["action"]),
        phase=str(node["phase"]),
        source=str(node["source"]),
        has_migration_metadata=node.get("has_migration_metadata") is True,
        metadata_pending=node.get("metadata_pending") is True,
        migration_safety=str(node.get("migration_safety") or "automatic"),
        current_version=node.get("current_version") if isinstance(node.get("current_version"), str) else None,
        target_version=node.get("target_version") if isinstance(node.get("target_version"), str) else None,
        reason=node.get("reason") if isinstance(node.get("reason"), str) else None,
    )


def _migration_target_metadata(
    *,
    module_id: str,
    manifest: ModuleManifest | None,
    catalog_module: Mapping[str, object] | None,
) -> dict[str, object]:
    metadata = _empty_migration_target_metadata(module_id)
    if manifest is not None:
        metadata.update(_manifest_migration_target_metadata(manifest))
    if catalog_module is not None:
        _apply_catalog_migration_target_metadata(metadata, catalog_module)
    return metadata


def _empty_migration_target_metadata(module_id: str) -> dict[str, object]:
    return {
        "module_id": module_id,
        "dependencies": (),
        "optional_dependencies": (),
        "provides_interfaces": (),
        "requires_interfaces": (),
        "migration_after": (),
        "migration_before": (),
        "migration_tasks": (),
    }


def _manifest_migration_target_metadata(manifest: ModuleManifest) -> dict[str, object]:
    metadata: dict[str, object] = {
        "dependencies": tuple(manifest.dependencies),
        "optional_dependencies": tuple(manifest.optional_dependencies),
        "provides_interfaces": tuple({"name": item.name, "version": item.version} for item in manifest.provides_interfaces),
        "requires_interfaces": tuple(
            {
                "name": item.name,
                "version_min": item.version_min,
                "version_max_exclusive": item.version_max_exclusive,
                "optional": item.optional,
            }
            for item in manifest.requires_interfaces
        ),
    }
    if manifest.migration_spec is not None:
        metadata["migration_after"] = tuple(manifest.migration_spec.migration_after)
        metadata["migration_before"] = tuple(manifest.migration_spec.migration_before)
        metadata["migration_tasks"] = tuple(_manifest_migration_task_metadata(task) for task in manifest.migration_spec.migration_tasks)
    return metadata


def _manifest_migration_task_metadata(task: object) -> dict[str, object]:
    return {
        "task_id": task.task_id,
        "phase": task.phase,
        "summary": task.summary,
        "task_version": task.task_version,
        "safety": task.safety,
        "idempotent": task.idempotent,
        "timeout_seconds": task.timeout_seconds,
        "source": "manifest",
        "has_executor": task.executor is not None,
    }


def _apply_catalog_migration_target_metadata(metadata: dict[str, object], catalog_module: Mapping[str, object]) -> None:
    _apply_catalog_metadata_value(metadata, "dependencies", _catalog_string_list(catalog_module.get("dependencies")))
    _apply_catalog_metadata_value(metadata, "optional_dependencies", _catalog_string_list(catalog_module.get("optional_dependencies")))
    _apply_catalog_metadata_value(metadata, "provides_interfaces", _catalog_interface_items(catalog_module.get("provides_interfaces")))
    _apply_catalog_metadata_value(metadata, "requires_interfaces", _catalog_interface_items(catalog_module.get("requires_interfaces")))
    _apply_catalog_metadata_value(metadata, "migration_after", _catalog_string_list(catalog_module.get("migration_after")))
    _apply_catalog_metadata_value(metadata, "migration_before", _catalog_string_list(catalog_module.get("migration_before")))
    _apply_catalog_metadata_value(metadata, "migration_tasks", _catalog_migration_task_items(catalog_module.get("migration_tasks")))


def _apply_catalog_metadata_value(metadata: dict[str, object], key: str, value: tuple[object, ...]) -> None:
    if value:
        metadata[key] = value


def _migration_task_plan_items(
    ordered_ids: tuple[str, ...],
    target_metadata: Mapping[str, Mapping[str, object]],
    nodes: Mapping[str, Mapping[str, object]],
) -> tuple[ModuleMigrationTaskPlanItem, ...]:
    tasks: list[ModuleMigrationTaskPlanItem] = []
    for module_id in ordered_ids:
        tasks.extend(_migration_task_plan_items_for_module(module_id=module_id, target_metadata=target_metadata, nodes=nodes))
    return tuple(tasks)


def _migration_task_plan_items_for_module(
    *,
    module_id: str,
    target_metadata: Mapping[str, Mapping[str, object]],
    nodes: Mapping[str, Mapping[str, object]],
) -> tuple[ModuleMigrationTaskPlanItem, ...]:
    node = nodes.get(module_id)
    metadata = target_metadata.get(module_id)
    if node is None or metadata is None:
        return ()
    tasks: list[ModuleMigrationTaskPlanItem] = []
    migration_tasks = tuple(task for task in metadata.get("migration_tasks", ()) if isinstance(task, Mapping))
    for phase in MIGRATION_TASK_PHASES:
        for task in migration_tasks:
            if str(task.get("phase") or "") == phase:
                tasks.append(_migration_task_plan_item(module_id=module_id, node=node, task=task, phase=phase))
    return tuple(tasks)


def _migration_task_plan_item(
    *,
    module_id: str,
    node: Mapping[str, object],
    task: Mapping[str, object],
    phase: str,
) -> ModuleMigrationTaskPlanItem:
    source = str(task.get("source") or node.get("source") or "manifest")
    return ModuleMigrationTaskPlanItem(
        module_id=module_id,
        task_id=str(task.get("task_id") or ""),
        phase=phase,
        summary=str(task.get("summary") or ""),
        task_version=str(task.get("task_version") or "1"),
        safety=str(task.get("safety") or "automatic"),
        idempotent=task.get("idempotent") is not False,
        timeout_seconds=task.get("timeout_seconds") if isinstance(task.get("timeout_seconds"), int) else None,
        source=source,
        has_executor=task.get("has_executor") is True,
        metadata_pending=node.get("metadata_pending") is True or source != "manifest",
    )


def _migration_task_preflight_issues(
    tasks: tuple[ModuleMigrationTaskPlanItem, ...],
    planned_by_id: Mapping[str, ModuleInstallPlanItem],
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    for task in tasks:
        if not task.task_id:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "migration_task_id_required",
                "Module migration task is missing a task id.",
                task.module_id,
            ))
        if task.phase not in MIGRATION_TASK_PHASES:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "migration_task_phase_invalid",
                f"Module migration task {task.task_id!r} declares unsupported phase {task.phase!r}.",
                task.module_id,
            ))
        if not task.idempotent:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "migration_task_not_idempotent",
                f"Module migration task {task.task_id!r} is not idempotent; publish an idempotent task before activation.",
                task.module_id,
            ))
        planned_item = planned_by_id.get(task.module_id)
        acknowledged = planned_item.data_safety_acknowledged if planned_item is not None else False
        if task.safety in MIGRATION_TASK_REVIEW_SAFETY and not acknowledged:
            severity: IssueSeverity = "blocker" if task.safety in MIGRATION_TASK_BLOCKING_SAFETY else "warning"
            code = (
                "migration_task_acknowledgement_required"
                if task.safety in MIGRATION_TASK_BLOCKING_SAFETY
                else "migration_task_review_required"
            )
            issues.append(ModuleInstallerIssue(
                severity,
                code,
                f"Module migration task {task.task_id!r} is marked {task.safety}; review and acknowledge the package data-safety step before activation.",
                task.module_id,
            ))
        elif task.safety in MIGRATION_TASK_REVIEW_SAFETY:
            issues.append(ModuleInstallerIssue(
                "warning",
                "migration_task_acknowledged",
                f"Module migration task {task.task_id!r} is marked {task.safety} and has been acknowledged.",
                task.module_id,
            ))
        if task.metadata_pending:
            issues.append(ModuleInstallerIssue(
                "info",
                "migration_task_executor_pending",
                f"Module migration task {task.task_id!r} executor can only be confirmed after package installation.",
                task.module_id,
            ))
        elif not task.has_executor:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "migration_task_executor_missing",
                f"Module migration task {task.task_id!r} is declared in the installed manifest but has no executor.",
                task.module_id,
            ))
    return tuple(issues)


def _migration_provider_modules(
    target_metadata: Mapping[str, Mapping[str, object]],
) -> dict[str, tuple[tuple[str, str], ...]]:
    providers: dict[str, list[tuple[str, str]]] = defaultdict(list)
    for module_id, module_metadata in target_metadata.items():
        for provided in module_metadata.get("provides_interfaces", ()):
            if not isinstance(provided, Mapping):
                continue
            name = provided.get("name") if isinstance(provided.get("name"), str) else None
            version = provided.get("version") if isinstance(provided.get("version"), str) else None
            if name and version:
                providers[name].append((module_id, version))
    return {name: tuple(values) for name, values in providers.items()}


def _add_migration_edge(
    edges: dict[str, set[str]],
    issues: list[ModuleInstallerIssue],
    *,
    before: str,
    after: str,
    source: str,
    target_enabled: set[str],
    available: Mapping[str, ModuleManifest],
) -> None:
    if before == after:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "migration_graph_self_dependency",
            f"Module {source!r} declares migration ordering against itself.",
            source,
        ))
        return
    if before not in edges or after not in edges:
        referenced = before if before not in edges else after
        if referenced not in target_enabled and referenced not in available:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "migration_graph_reference_missing",
                f"Module {source!r} declares migration ordering against absent module {referenced!r}.",
                source,
            ))
        return
    edges[before].add(after)


def _topological_module_order(edges: Mapping[str, set[str]]) -> tuple[tuple[str, ...], tuple[str, ...]]:
    incoming, outgoing = _topological_graph_indexes(edges)
    ready = _topological_ready_nodes(incoming)
    ordered: list[str] = []
    while ready:
        module_id = ready.pop(0)
        ordered.append(module_id)
        _release_topological_node(module_id, incoming=incoming, outgoing=outgoing, ordered=ordered, ready=ready)
    if len(ordered) == len(incoming):
        return tuple(ordered), ()
    return tuple(ordered), _topological_cycle_ids(incoming)


def _topological_graph_indexes(edges: Mapping[str, set[str]]) -> tuple[dict[str, set[str]], dict[str, set[str]]]:
    incoming: dict[str, set[str]] = {module_id: set() for module_id in edges}
    outgoing: dict[str, set[str]] = {module_id: set(values) for module_id, values in edges.items()}
    for before, after_values in edges.items():
        for after in after_values:
            incoming.setdefault(after, set()).add(before)
            outgoing.setdefault(before, set()).add(after)
    return incoming, outgoing


def _topological_ready_nodes(incoming: Mapping[str, set[str]]) -> list[str]:
    return sorted(module_id for module_id, values in incoming.items() if not values)


def _release_topological_node(
    module_id: str,
    *,
    incoming: dict[str, set[str]],
    outgoing: Mapping[str, set[str]],
    ordered: list[str],
    ready: list[str],
) -> None:
    for after in sorted(outgoing.get(module_id, ())):
        incoming[after].discard(module_id)
        if not incoming[after] and after not in ordered and after not in ready:
            ready.append(after)
            ready.sort()


def _topological_cycle_ids(incoming: Mapping[str, set[str]]) -> tuple[str, ...]:
    return tuple(sorted(module_id for module_id, values in incoming.items() if values))


def _catalog_modules_for_target_plan(
    planned_items: tuple[ModuleInstallPlanItem, ...],
) -> dict[str, Mapping[str, object]]:
    if not any(item.source == "catalog" and item.action in PACKAGE_TARGET_ACTIONS for item in planned_items):
        return {}
    try:
        from govoplan_core.core.module_package_catalog import validate_module_package_catalog

        result = validate_module_package_catalog()
    except Exception:
        return {}
    if result.get("valid") is not True:
        return {}
    return _catalog_modules_by_id(result)


def _catalog_optional_string(module: Mapping[str, object] | None, key: str) -> str | None:
    if module is None:
        return None
    value = module.get(key)
    if isinstance(value, str):
        cleaned = value.strip()
        return cleaned or None
    return None


def _catalog_optional_bool(module: Mapping[str, object] | None, key: str) -> bool:
    if module is None:
        return False
    return module.get(key) is True


def _selected_catalog_interface_issues(
    catalog_items: tuple[ModuleInstallPlanItem, ...],
    validation: Mapping[str, object],
    available: Mapping[str, ModuleManifest],
) -> tuple[ModuleInstallerIssue, ...]:
    modules = _catalog_modules_by_id(validation)
    planned_ids = {item.module_id for item in catalog_items}
    target_provider_versions = _catalog_target_provider_versions(
        catalog_items=catalog_items,
        modules=modules,
        available=available,
    )

    issues: list[ModuleInstallerIssue] = []
    for item in catalog_items:
        module = modules.get(item.module_id)
        if module is None:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "catalog_entry_missing",
                f"Catalog-sourced install plan item {item.module_id!r} is no longer present in the package catalog.",
                item.module_id,
            ))
            continue
        issues.extend(_catalog_dependency_issues(
            module_id=item.module_id,
            module=module,
            planned_ids=planned_ids,
            available=available,
            modules=modules,
        ))
        issues.extend(_catalog_update_policy_issues(item=item, module=module, available=available))
        issues.extend(_catalog_data_safety_issues(item=item, module=module))
        issues.extend(_catalog_requirement_issues(
            module_id=item.module_id,
            requirements=_catalog_interface_items(module.get("requires_interfaces")),
            target_provider_versions=target_provider_versions,
            catalog_modules=modules,
            planned_ids=planned_ids,
            code_prefix="catalog",
        ))

    for manifest in available.values():
        if manifest.id in planned_ids:
            continue
        requirements = tuple(
            {
                "name": requirement.name,
                "version_min": requirement.version_min,
                "version_max_exclusive": requirement.version_max_exclusive,
                "optional": requirement.optional,
            }
            for requirement in manifest.requires_interfaces
        )
        requirement_issues = _catalog_requirement_issues(
            module_id=manifest.id,
            requirements=requirements,
            target_provider_versions=target_provider_versions,
            catalog_modules=modules,
            planned_ids=planned_ids,
            code_prefix="target",
        )
        candidate_update = modules.get(manifest.id)
        if (
            any(issue.severity == "blocker" for issue in requirement_issues)
            and candidate_update is not None
            and _catalog_requirements_satisfied(
                _catalog_interface_items(candidate_update.get("requires_interfaces")),
                target_provider_versions,
            )
        ):
            issues.append(ModuleInstallerIssue(
                "blocker",
                "companion_update_required",
                (
                    f"Target package plan would leave installed module {manifest.id!r} "
                    f"incompatible; add the catalog update for {manifest.id!r} to the same plan."
                ),
                manifest.id,
            ))
            continue
        issues.extend(requirement_issues)
    return tuple(issues)


def _catalog_data_safety_issues(
    *,
    item: ModuleInstallPlanItem,
    module: Mapping[str, object],
) -> tuple[ModuleInstallerIssue, ...]:
    raw_safety = module.get("migration_safety")
    safety = raw_safety if isinstance(raw_safety, str) and raw_safety else "automatic"
    raw_notes = module.get("migration_notes")
    notes = raw_notes.strip() if isinstance(raw_notes, str) and raw_notes.strip() else None
    detail = f" Catalog note: {notes}" if notes else ""
    if safety == "automatic":
        return ()
    if safety == "requires_review":
        return _catalog_review_safety_issue(item=item, detail=detail)
    if safety == "forward_only":
        return _catalog_forward_only_safety_issues(item=item, module=module, detail=detail)
    if safety == "destructive":
        return _catalog_destructive_safety_issues(item=item, module=module, notes=notes, detail=detail)
    return (ModuleInstallerIssue(
        "blocker",
        "catalog_migration_safety_invalid",
        f"Catalog entry has unsupported migration_safety value {safety!r}.",
        item.module_id,
    ),)


def _catalog_review_safety_issue(*, item: ModuleInstallPlanItem, detail: str) -> tuple[ModuleInstallerIssue, ...]:
    severity: IssueSeverity = "info" if item.data_safety_acknowledged else "warning"
    code = "migration_review_acknowledged" if item.data_safety_acknowledged else "migration_review_required"
    message = "Catalog marks this package change as requiring migration review before activation." + detail
    return (ModuleInstallerIssue(severity, code, message, item.module_id),)


def _catalog_forward_only_safety_issues(
    *,
    item: ModuleInstallPlanItem,
    module: Mapping[str, object],
    detail: str,
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    recovery_notes = _catalog_optional_string(module, "recovery_notes")
    if not _catalog_optional_bool(module, "recovery_tested"):
        issues.append(ModuleInstallerIssue(
            "blocker",
            "recovery_validation_required",
            "Forward-only catalog updates must declare recovery_tested=true after a verified restore or forward-recovery rehearsal.",
            item.module_id,
        ))
    if not recovery_notes and not item.notes:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "recovery_notes_required",
            "Forward-only catalog updates require recovery notes or operator notes describing the tested recovery path.",
            item.module_id,
        ))
    if not item.data_safety_acknowledged:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "forward_only_migration_acknowledgement_required",
            "Catalog marks this package change as forward-only; acknowledge the data safety review before activation." + detail,
            item.module_id,
        ))
    if issues:
        return tuple(issues)
    return (ModuleInstallerIssue(
        "warning",
        "forward_only_migration_acknowledged",
        "Forward-only package change has been acknowledged by the operator." + detail,
        item.module_id,
    ),)


def _catalog_destructive_safety_issues(
    *,
    item: ModuleInstallPlanItem,
    module: Mapping[str, object],
    notes: str | None,
    detail: str,
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    recovery_notes = _catalog_optional_string(module, "recovery_notes")
    if not _catalog_optional_bool(module, "recovery_tested"):
        issues.append(ModuleInstallerIssue(
            "blocker",
            "recovery_validation_required",
            "Destructive catalog updates must declare recovery_tested=true after a verified restore or forward-recovery rehearsal.",
            item.module_id,
        ))
    if not recovery_notes and not item.notes:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "recovery_notes_required",
            "Destructive catalog updates require recovery notes or operator notes describing the tested recovery path.",
            item.module_id,
        ))
    if not notes and not item.notes:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "destructive_migration_plan_required",
            "Catalog marks this package change as destructive; add catalog migration notes or operator notes describing the cleanup/retirement plan.",
            item.module_id,
        ))
    if not item.data_safety_acknowledged:
        issues.append(ModuleInstallerIssue(
            "blocker",
            "destructive_migration_acknowledgement_required",
            "Catalog marks this package change as destructive; acknowledge the data safety review before activation." + detail,
            item.module_id,
        ))
    if not issues:
        issues.append(ModuleInstallerIssue(
            "warning",
            "destructive_migration_acknowledged",
            "Destructive package change has been acknowledged by the operator." + detail,
            item.module_id,
        ))
    return tuple(issues)


def _catalog_update_policy_issues(
    *,
    item: ModuleInstallPlanItem,
    module: Mapping[str, object],
    available: Mapping[str, ModuleManifest],
) -> tuple[ModuleInstallerIssue, ...]:
    if item.action != "update":
        return ()
    manifest = available.get(item.module_id)
    if manifest is None:
        return ()
    target_version = _catalog_optional_string(module, "version")
    issues: list[ModuleInstallerIssue] = []
    issues.extend(_catalog_target_version_policy_issues(item=item, module=module, manifest=manifest, target_version=target_version))
    issues.extend(_catalog_update_window_issues(item=item, module=module, manifest=manifest, target_version=target_version))
    issues.extend(_catalog_bridge_release_issues(item=item, module=module))
    return tuple(issues)


def _catalog_target_version_policy_issues(
    *,
    item: ModuleInstallPlanItem,
    module: Mapping[str, object],
    manifest: ModuleManifest,
    target_version: str | None,
) -> tuple[ModuleInstallerIssue, ...]:
    if not target_version:
        return ()
    comparison = compare_versions(target_version, manifest.version)
    if comparison < 0 and not _catalog_optional_bool(module, "allow_downgrade"):
        return (ModuleInstallerIssue(
            "blocker",
            "catalog_downgrade_blocked",
            (
                f"Catalog target version {target_version} is older than installed version "
                f"{manifest.version}; set allow_downgrade only for a reviewed rollback path."
            ),
            item.module_id,
        ),)
    if comparison == 0 and not _catalog_optional_bool(module, "allow_same_version"):
        return (ModuleInstallerIssue(
            "blocker",
            "catalog_noop_update_blocked",
            (
                f"Catalog target version {target_version} matches the installed version; "
                "use an install plan only when package refs or metadata intentionally need refreshing."
            ),
            item.module_id,
        ),)
    return ()


def _catalog_update_window_issues(
    *,
    item: ModuleInstallPlanItem,
    module: Mapping[str, object],
    manifest: ModuleManifest,
    target_version: str | None,
) -> tuple[ModuleInstallerIssue, ...]:
    current_version_min = _catalog_optional_string(module, "current_version_min")
    current_version_max_exclusive = _catalog_optional_string(module, "current_version_max_exclusive")
    if current_version_min is None and current_version_max_exclusive is None:
        return ()
    if version_satisfies_range(manifest.version, version_min=current_version_min, version_max_exclusive=current_version_max_exclusive):
        return ()
    version_range = format_version_range(version_min=current_version_min, version_max_exclusive=current_version_max_exclusive)
    bridge_note = _catalog_optional_string(module, "bridge_notes")
    bridge_detail = f" Bridge note: {bridge_note}" if bridge_note else ""
    return (ModuleInstallerIssue(
        "blocker",
        "catalog_update_bridge_required",
        (
            f"Installed version {manifest.version} is outside the supported update window "
            f"for catalog target {target_version or item.module_id} ({version_range})."
            + bridge_detail
        ),
        item.module_id,
    ),)


def _catalog_bridge_release_issues(
    *,
    item: ModuleInstallPlanItem,
    module: Mapping[str, object],
) -> tuple[ModuleInstallerIssue, ...]:
    if not _catalog_optional_bool(module, "bridge_release"):
        return ()
    bridge_note = _catalog_optional_string(module, "bridge_notes")
    detail = f" Bridge note: {bridge_note}" if bridge_note else ""
    return (ModuleInstallerIssue(
        "info",
        "catalog_bridge_release",
        "Catalog marks this target as a bridge release for staged update paths." + detail,
        item.module_id,
    ),)


def _catalog_modules_by_id(validation: Mapping[str, object]) -> dict[str, Mapping[str, object]]:
    return {
        str(item.get("module_id")): item
        for item in validation.get("modules", [])
        if isinstance(item, Mapping) and item.get("module_id")
    }


def _catalog_target_provider_versions(
    *,
    catalog_items: tuple[ModuleInstallPlanItem, ...],
    modules: Mapping[str, Mapping[str, object]],
    available: Mapping[str, ModuleManifest],
) -> dict[str, list[tuple[str, str]]]:
    planned_ids = {item.module_id for item in catalog_items}
    providers: dict[str, list[tuple[str, str]]] = defaultdict(list)
    for manifest in available.values():
        if manifest.id in planned_ids:
            continue
        for provided in manifest.provides_interfaces:
            providers[provided.name].append((manifest.id, provided.version))
    for item in catalog_items:
        module = modules.get(item.module_id)
        if module is None:
            continue
        for provided in _catalog_interface_items(module.get("provides_interfaces")):
            name = provided.get("name") if isinstance(provided.get("name"), str) else None
            version = provided.get("version") if isinstance(provided.get("version"), str) else None
            if name and version:
                providers[name].append((item.module_id, version))
    return providers


def _catalog_target_provider_versions_for_ids(
    *,
    planned_ids: set[str],
    modules: Mapping[str, Mapping[str, object]],
    available: Mapping[str, ModuleManifest],
) -> dict[str, list[tuple[str, str]]]:
    providers: dict[str, list[tuple[str, str]]] = defaultdict(list)
    for manifest in available.values():
        if manifest.id in planned_ids:
            continue
        for provided in manifest.provides_interfaces:
            providers[provided.name].append((manifest.id, provided.version))
    for module_id in planned_ids:
        module = modules.get(module_id)
        if module is None:
            continue
        for provided in _catalog_interface_items(module.get("provides_interfaces")):
            name = provided.get("name") if isinstance(provided.get("name"), str) else None
            version = provided.get("version") if isinstance(provided.get("version"), str) else None
            if name and version:
                providers[name].append((module_id, version))
    return providers


def _catalog_requirement_satisfied(
    requirement: Mapping[str, object],
    target_provider_versions: Mapping[str, list[tuple[str, str]]],
) -> bool:
    name = requirement.get("name") if isinstance(requirement.get("name"), str) else None
    if not name:
        return True
    version_min = requirement.get("version_min") if isinstance(requirement.get("version_min"), str) else None
    version_max_exclusive = (
        requirement.get("version_max_exclusive")
        if isinstance(requirement.get("version_max_exclusive"), str)
        else None
    )
    providers = target_provider_versions.get(name, ())
    matching = [
        version
        for _provider_module_id, version in providers
        if version_satisfies_range(
            version,
            version_min=version_min,
            version_max_exclusive=version_max_exclusive,
        )
    ]
    if matching:
        return True
    return requirement.get("optional") is True and not providers


def _catalog_dependency_issues(
    *,
    module_id: str,
    module: Mapping[str, object],
    planned_ids: set[str],
    available: Mapping[str, ModuleManifest],
    modules: Mapping[str, Mapping[str, object]],
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    for dependency_id in _catalog_string_list(module.get("dependencies")):
        if dependency_id in planned_ids or dependency_id in available:
            continue
        if dependency_id in modules:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "companion_update_required",
                (
                    f"Catalog-sourced module {module_id!r} depends on {dependency_id!r}; "
                    "add that catalog entry to the same package plan."
                ),
                module_id,
            ))
            continue
        issues.append(ModuleInstallerIssue(
            "blocker",
            "catalog_required_dependency_missing",
            (
                f"Catalog-sourced module {module_id!r} depends on {dependency_id!r}, "
                "but it is neither installed nor present in the package catalog."
            ),
            module_id,
        ))
    return tuple(issues)


def _catalog_requirement_issues(
    *,
    module_id: str,
    requirements: tuple[Mapping[str, object], ...],
    target_provider_versions: Mapping[str, list[tuple[str, str]]],
    catalog_modules: Mapping[str, Mapping[str, object]],
    planned_ids: set[str],
    code_prefix: str,
) -> tuple[ModuleInstallerIssue, ...]:
    issues: list[ModuleInstallerIssue] = []
    for required in requirements:
        name = required.get("name") if isinstance(required.get("name"), str) else None
        if not name:
            continue
        version_min = required.get("version_min") if isinstance(required.get("version_min"), str) else None
        version_max_exclusive = (
            required.get("version_max_exclusive")
            if isinstance(required.get("version_max_exclusive"), str)
            else None
        )
        optional = required.get("optional") is True
        providers = target_provider_versions.get(name, ())
        matching = [
            (provider_module_id, version)
            for provider_module_id, version in providers
            if version_satisfies_range(
                version,
                version_min=version_min,
                version_max_exclusive=version_max_exclusive,
            )
        ]
        if matching:
            continue
        version_range = format_version_range(
            version_min=version_min,
            version_max_exclusive=version_max_exclusive,
        )
        suggestions = _catalog_provider_suggestions(
            name=name,
            version_min=version_min,
            version_max_exclusive=version_max_exclusive,
            catalog_modules=catalog_modules,
            planned_ids=planned_ids,
        )
        if suggestions:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "companion_update_required",
                (
                    f"Module {module_id!r} requires interface {name!r} ({version_range}); "
                    f"add companion catalog update(s) to the same plan: {', '.join(suggestions)}."
                ),
                module_id,
            ))
            continue
        if not providers:
            if optional:
                continue
            issues.append(ModuleInstallerIssue(
                "blocker",
                f"{code_prefix}_required_interface_missing",
                (
                    f"Module {module_id!r} requires interface {name!r} ({version_range}), "
                    "but the target package plan has no provider."
                ),
                module_id,
            ))
            continue
        provider_summary = ", ".join(f"{provider_module_id}@{version}" for provider_module_id, version in providers)
        severity: IssueSeverity = "warning" if optional else "blocker"
        issues.append(ModuleInstallerIssue(
            severity,
            f"{code_prefix}_interface_version_mismatch",
            (
                f"Module {module_id!r} requires interface {name!r} ({version_range}), "
                f"but target providers are {provider_summary}."
            ),
            module_id,
        ))
    return tuple(issues)


def _catalog_provider_suggestions(
    *,
    name: str,
    version_min: str | None,
    version_max_exclusive: str | None,
    catalog_modules: Mapping[str, Mapping[str, object]],
    planned_ids: set[str],
) -> tuple[str, ...]:
    suggestions: list[str] = []
    for module_id, module in catalog_modules.items():
        if module_id in planned_ids:
            continue
        for provided in _catalog_interface_items(module.get("provides_interfaces")):
            provided_name = provided.get("name") if isinstance(provided.get("name"), str) else None
            version = provided.get("version") if isinstance(provided.get("version"), str) else None
            if provided_name != name or not version:
                continue
            if version_satisfies_range(
                version,
                version_min=version_min,
                version_max_exclusive=version_max_exclusive,
            ):
                suggestions.append(module_id)
                break
    return tuple(sorted(dict.fromkeys(suggestions)))


def _catalog_requirements_satisfied(
    requirements: tuple[Mapping[str, object], ...],
    target_provider_versions: Mapping[str, list[tuple[str, str]]],
) -> bool:
    for required in requirements:
        name = required.get("name") if isinstance(required.get("name"), str) else None
        if not name:
            continue
        version_min = required.get("version_min") if isinstance(required.get("version_min"), str) else None
        version_max_exclusive = (
            required.get("version_max_exclusive")
            if isinstance(required.get("version_max_exclusive"), str)
            else None
        )
        providers = target_provider_versions.get(name, ())
        matching = [
            version
            for _provider_module_id, version in providers
            if version_satisfies_range(
                version,
                version_min=version_min,
                version_max_exclusive=version_max_exclusive,
            )
        ]
        if matching:
            continue
        if required.get("optional") is True and not providers:
            continue
        return False
    return True


def _catalog_string_list(value: object) -> tuple[str, ...]:
    if not isinstance(value, list):
        return ()
    return tuple(str(item).strip() for item in value if str(item).strip())


def _catalog_interface_items(value: object) -> tuple[Mapping[str, object], ...]:
    if not isinstance(value, list):
        return ()
    return tuple(item for item in value if isinstance(item, Mapping))


def _catalog_migration_task_items(value: object) -> tuple[Mapping[str, object], ...]:
    if not isinstance(value, list):
        return ()
    return tuple({**item, "source": "catalog", "has_executor": False} for item in value if isinstance(item, Mapping))


def _post_install_checklist(
    *,
    planned_items: tuple[ModuleInstallPlanItem, ...],
    maintenance_mode: bool,
    frontend_rebuild_required: bool,
    migration_plan: ModuleMigrationExecutionPlan,
    blocker: bool,
) -> tuple[ModuleInstallChecklistItem, ...]:
    has_package_changes = bool(planned_items)
    preflight_status: ChecklistStatus = "blocked" if blocker else "done"
    package_status: ChecklistStatus = "pending" if has_package_changes else "done"
    restart_status: ChecklistStatus = "pending" if has_package_changes else "done"
    migration_status: ChecklistStatus = "pending" if migration_plan.requires_database_migration else "done"
    if blocker and migration_plan.requires_database_migration:
        migration_status = "blocked"
    webui_status: ChecklistStatus = "pending" if frontend_rebuild_required else "done"
    module_state_status: ChecklistStatus = "pending" if has_package_changes else "done"
    health_status: ChecklistStatus = "pending" if has_package_changes else "done"
    return (
        ModuleInstallChecklistItem(
            id="maintenance-mode",
            label="Maintenance mode",
            status="done" if maintenance_mode else "blocked",
            detail="Enable maintenance mode before changing installed packages.",
        ),
        ModuleInstallChecklistItem(
            id="preflight",
            label="Installer preflight",
            status=preflight_status,
            detail="Compatibility, pinned refs, uninstall guards, and package targets are checked before apply.",
        ),
        ModuleInstallChecklistItem(
            id="package-apply",
            label="Package apply",
            status=package_status,
            detail="The installer snapshots Python and WebUI package state before mutating packages.",
        ),
        ModuleInstallChecklistItem(
            id="migrations",
            label="Module migrations",
            status=migration_status,
            detail=(
                "Run ordered module migrations after backend package changes."
                if migration_plan.requires_database_migration
                else "No module migration step is expected for this plan."
            ),
        ),
        ModuleInstallChecklistItem(
            id="webui-rebuild",
            label="WebUI rebuild",
            status=webui_status,
            detail="Required when WebUI package dependencies changed.",
        ),
        ModuleInstallChecklistItem(
            id="backend-restart",
            label="Backend restart/reload",
            status=restart_status,
            detail="Required after Python package changes so module discovery sees the new package set.",
        ),
        ModuleInstallChecklistItem(
            id="module-state",
            label="Module state activation",
            status=module_state_status,
            detail="Enable or disable the module after package installation/removal has succeeded.",
        ),
        ModuleInstallChecklistItem(
            id="health-check",
            label="Health check",
            status=health_status,
            detail="Supervisor mode polls the health endpoint and rolls back if the restarted server does not come up.",
        ),
    )


def _rollback_after_supervisor_failure(
    *,
    result: ModuleInstallerRunResult,
    session: Session,
    original_plan: ModuleInstallPlan,
    original_desired_enabled: Iterable[str],
    runtime_dir: Path,
    npm_bin: str,
    restart_commands: Iterable[str],
    health_urls: Iterable[str],
    health_timeout_seconds: float,
    health_interval_seconds: float,
    build_webui: bool,
    reason: str,
) -> ModuleInstallerRunResult:
    supervisor: dict[str, object] = {
        "status": "rollback-started",
        "failure_reason": reason,
        "rolled_back_at": datetime.now(tz=UTC).isoformat(),
    }
    try:
        session.rollback()
    except Exception as rollback_exc:
        logger.debug("Module installer rollback failed before rollback plan: %s", rollback_exc, exc_info=True)
    rollback = rollback_module_install_run(run_id=result.run_id, runtime_dir=runtime_dir, npm_bin=npm_bin, build_webui=build_webui)
    rollback_record = _read_run_record(result.record_path) or {}
    if not _database_restore_was_applied(rollback_record):
        save_module_install_plan(session, original_plan.items)
        save_desired_enabled_modules(session, original_desired_enabled)
        session.commit()
    supervisor["rollback"] = rollback.as_dict()

    rollback_restart = _run_restart_commands(restart_commands)
    if rollback_restart:
        supervisor["rollback_restart_commands"] = rollback_restart
    if rollback_restart and all(item.get("return_code") == 0 for item in rollback_restart) and tuple(health_urls):
        supervisor["rollback_health"] = _wait_for_health_urls(
            health_urls,
            timeout_seconds=health_timeout_seconds,
            interval_seconds=health_interval_seconds,
        )
    supervisor["status"] = "rolled-back" if rollback.return_code == 0 else "rollback-failed"
    supervisor["finished_at"] = datetime.now(tz=UTC).isoformat()
    _update_run_record(result.record_path, {"supervisor": supervisor})
    return ModuleInstallerRunResult(
        run_id=result.run_id,
        status="rolled-back" if rollback.return_code == 0 else "failed",
        record_path=result.record_path,
        commands=result.commands,
        return_code=1,
        error=reason,
        rollback=rollback.as_dict(),
    )


def _command_list(first: str | None, rest: Iterable[str] | None) -> tuple[str, ...]:
    commands: list[str] = []
    if first and first.strip():
        commands.append(first.strip())
    if rest:
        for item in rest:
            if item and str(item).strip():
                commands.append(str(item).strip())
    return tuple(dict.fromkeys(commands))


def _run_restart_commands(commands: Iterable[str]) -> list[dict[str, object]]:
    return [_run_restart_command(command) for command in commands]


def _run_restart_command(command: str) -> dict[str, object]:
    if not command.strip():
        return {
            "command": command,
            "return_code": 0,
            "stdout": "",
            "stderr": "",
        }
    result = _run_operator_command(command)
    return {
        "command": command,
        "return_code": result.returncode,
        "stdout": result.stdout[-8000:],
        "stderr": result.stderr[-8000:],
    }


def _wait_for_health_urls(urls: Iterable[str], *, timeout_seconds: float, interval_seconds: float) -> dict[str, object]:
    results: list[dict[str, object]] = []
    ok = True
    errors: list[str] = []
    for url in urls:
        health = _wait_for_health_url(
            url,
            timeout_seconds=timeout_seconds,
            interval_seconds=interval_seconds,
        )
        health["url"] = url
        results.append(health)
        if not health.get("ok"):
            ok = False
            errors.append(f"{url}: {health.get('error') or health.get('status') or 'failed'}")
    payload: dict[str, object] = {"ok": ok, "checks": results}
    if errors:
        payload["error"] = "; ".join(errors)
    return payload


def _planned_python_install_packages(record: Mapping[str, object]) -> tuple[str, ...]:
    raw_plan = record.get("plan")
    if not isinstance(raw_plan, list):
        return ()
    packages: list[str] = []
    for item in raw_plan:
        if not isinstance(item, Mapping):
            continue
        if item.get("action") != "install":
            continue
        package = item.get("python_package")
        if isinstance(package, str) and package.strip():
            packages.append(package.strip())
            continue
        python_ref = item.get("python_ref")
        if isinstance(python_ref, str):
            package_name = _python_package_name_from_ref(python_ref)
            if package_name:
                packages.append(package_name)
    return tuple(dict.fromkeys(packages))


def _python_package_name_from_ref(value: str) -> str | None:
    cleaned = value.strip()
    if not cleaned:
        return None
    if " @ " in cleaned:
        cleaned = cleaned.split(" @ ", 1)[0].strip()
    if "==" in cleaned:
        cleaned = cleaned.split("==", 1)[0].strip()
    if "[" in cleaned:
        cleaned = cleaned.split("[", 1)[0].strip()
    if re.match(r"^[A-Za-z0-9_.-]+$", cleaned):
        return cleaned
    return None


def _wait_for_health_url(url: str, *, timeout_seconds: float, interval_seconds: float) -> dict[str, object]:
    try:
        health_url = validate_http_url(url, label="health URL")
    except ValueError as exc:
        return {"ok": False, "attempts": 0, "error": str(exc)}
    deadline = time.monotonic() + max(timeout_seconds, 0.1)
    attempts = 0
    last_error = ""
    while time.monotonic() < deadline:
        attempts += 1
        try:
            response = fetch_http(health_url, timeout=min(max(interval_seconds, 0.5), 10.0), label="health URL")
            if 200 <= response.status < 400:
                return {"ok": True, "status": response.status, "attempts": attempts}
            last_error = f"HTTP {response.status}"
        except OSError as exc:
            last_error = str(exc)
        time.sleep(max(interval_seconds, 0.1))
    return {"ok": False, "attempts": attempts, "error": last_error or "health timeout"}


def _update_run_record(path: Path, patch: Mapping[str, object]) -> None:
    try:
        record = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return
    if not isinstance(record, dict):
        return
    record.update(patch)
    _write_json(path, record)


def _read_run_record(path: Path) -> dict[str, object] | None:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return None
    return payload if isinstance(payload, dict) else None


def _read_json_file(path: Path) -> object | None:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return None


def _run_summary(path: Path, record: Mapping[str, object]) -> dict[str, object]:
    supervisor = record.get("supervisor") if isinstance(record.get("supervisor"), Mapping) else {}
    commands = record.get("commands") if isinstance(record.get("commands"), list) else []
    plan = record.get("plan") if isinstance(record.get("plan"), list) else []
    request = record.get("request") if isinstance(record.get("request"), Mapping) else {}
    return {
        "run_id": str(record.get("run_id") or path.parent.name),
        "status": str(record.get("status") or "unknown"),
        "started_at": record.get("started_at"),
        "finished_at": record.get("finished_at"),
        "rollback_status": record.get("rollback_status"),
        "supervisor_status": supervisor.get("status") if isinstance(supervisor, Mapping) else None,
        "error": record.get("error"),
        "record_path": str(path),
        "commands_count": len(commands),
        "planned_modules": [str(item.get("module_id")) for item in plan if isinstance(item, Mapping) and item.get("module_id")],
        "request_id": request.get("request_id") if isinstance(request, Mapping) else None,
        "requested_by": request.get("requested_by") if isinstance(request, Mapping) else None,
        "trace": request.get("trace") if isinstance(request.get("trace"), Mapping) else None,
    }


def _current_trace_payload() -> dict[str, object]:
    trace = current_event_trace()
    if trace is None:
        return {}
    payload: dict[str, object] = {"correlation_id": trace.correlation_id}
    if trace.causation_id:
        payload["causation_id"] = trace.causation_id
    return payload


def _database_restore_was_applied(record: Mapping[str, object]) -> bool:
    restore = record.get("rollback_database_restore")
    return isinstance(restore, Mapping) and restore.get("restored") is True


def _looks_pinned_dependency_ref(value: str) -> bool:
    cleaned = value.strip()
    if "==" in cleaned:
        return True
    if re.match(r"^\d+\.\d+\.\d+(?:[-+][A-Za-z0-9_.-]+)?$", cleaned):
        return True
    if cleaned.startswith(("git+", "ssh://", "https://", "http://")):
        return "#v" in cleaned or "@v" in cleaned or "#refs/tags/" in cleaned or "@refs/tags/" in cleaned
    if " @ git+" in cleaned:
        return "@v" in cleaned or "@refs/tags/" in cleaned
    return False


def _configured_require_artifact_integrity() -> bool:
    return os.getenv("GOVOPLAN_MODULE_INSTALLER_REQUIRE_ARTIFACT_INTEGRITY", "").strip().lower() in {"1", "true", "yes", "on"}


def _verify_artifact_integrity(
    planned_items: tuple[ModuleInstallPlanItem, ...],
    *,
    require_verified: bool,
) -> tuple[tuple[dict[str, object], ...], tuple[ModuleInstallerIssue, ...]]:
    records: list[dict[str, object]] = []
    issues: list[ModuleInstallerIssue] = []
    for item in planned_items:
        if item.action not in PACKAGE_TARGET_ACTIONS:
            continue
        for kind, package_name, package_ref in (
            ("python", item.python_package, item.python_ref),
            ("webui", item.webui_package, item.webui_ref),
        ):
            if not package_ref:
                continue
            raw_metadata = _artifact_metadata(item.artifact_integrity, kind)
            if raw_metadata is None:
                if require_verified:
                    issues.append(ModuleInstallerIssue(
                        "blocker",
                        "artifact_integrity_required",
                        f"{kind.capitalize()} artifact integrity metadata is required in production mode.",
                        item.module_id,
                    ))
                continue
            record, record_issues = _verify_artifact_metadata(
                item,
                kind=kind,
                package_name=package_name,
                package_ref=package_ref,
                metadata=raw_metadata,
                require_verified=require_verified,
            )
            records.append(record)
            issues.extend(record_issues)
    return tuple(records), tuple(issues)


def _artifact_metadata(value: Mapping[str, object] | None, kind: str) -> Mapping[str, object] | None:
    if not isinstance(value, Mapping):
        return None
    raw = value.get(kind)
    return raw if isinstance(raw, Mapping) else None


def _verify_artifact_metadata(
    item: ModuleInstallPlanItem,
    *,
    kind: str,
    package_name: str | None,
    package_ref: str,
    metadata: Mapping[str, object],
    require_verified: bool,
) -> tuple[dict[str, object], tuple[ModuleInstallerIssue, ...]]:
    issues: list[ModuleInstallerIssue] = []
    expected_sha256 = _artifact_text(metadata, "sha256")
    artifact_path = _artifact_path(metadata)
    expected_ref = _artifact_text(metadata, "ref") or _artifact_text(metadata, "expected_ref")
    record: dict[str, object] = {
        "module_id": item.module_id,
        "kind": kind,
        "package_ref": package_ref,
        "verified": False,
    }
    if package_name:
        record["package"] = package_name
    for key in ("sha256", "sbom_url", "provenance_url", "registry_identity", "git_ref"):
        value = _artifact_text(metadata, key)
        if value:
            record[key] = value
    if expected_ref:
        record["expected_ref"] = expected_ref
        if expected_ref != package_ref:
            issues.append(ModuleInstallerIssue(
                "blocker",
                "artifact_ref_mismatch",
                f"{kind.capitalize()} artifact metadata expected ref {expected_ref!r} but the install plan uses {package_ref!r}.",
                item.module_id,
            ))
    if not expected_sha256:
        issues.append(ModuleInstallerIssue(
            "blocker" if require_verified else "warning",
            "artifact_digest_missing",
            f"{kind.capitalize()} artifact integrity metadata has no sha256 digest.",
            item.module_id,
        ))
        return record, tuple(issues)
    if artifact_path is None:
        issues.append(ModuleInstallerIssue(
            "blocker" if require_verified else "warning",
            "artifact_not_verified",
            f"{kind.capitalize()} artifact digest is declared but no local artifact path is available for verification.",
            item.module_id,
        ))
        return record, tuple(issues)
    record["artifact_path"] = str(artifact_path)
    if not artifact_path.exists():
        issues.append(ModuleInstallerIssue(
            "blocker",
            "artifact_missing",
            f"{kind.capitalize()} artifact does not exist: {artifact_path}",
            item.module_id,
        ))
        return record, tuple(issues)
    actual_sha256 = _sha256_file(artifact_path)
    record["actual_sha256"] = actual_sha256
    if actual_sha256 != expected_sha256.lower():
        issues.append(ModuleInstallerIssue(
            "blocker",
            "artifact_digest_mismatch",
            f"{kind.capitalize()} artifact digest mismatch for {artifact_path}.",
            item.module_id,
        ))
        return record, tuple(issues)
    record["verified"] = True
    record["verified_at"] = datetime.now(tz=UTC).isoformat()
    return record, tuple(issues)


def _artifact_text(metadata: Mapping[str, object], key: str) -> str | None:
    value = metadata.get(key)
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _artifact_path(metadata: Mapping[str, object]) -> Path | None:
    value = _artifact_text(metadata, "path") or _artifact_text(metadata, "artifact_path")
    if not value:
        return None
    if value.startswith("file://"):
        value = value[len("file://"):]
    path = Path(value).expanduser()
    return path if path.is_absolute() else Path.cwd() / path


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


def _snapshot_environment(
    run_dir: Path,
    *,
    webui_root: Path | None,
    database_url: str | None,
    include_database: bool,
    database_backup_command: str | None,
    database_restore_command: str | None,
    database_restore_check_command: str | None,
) -> dict[str, object]:
    snapshot: dict[str, object] = {}
    freeze_result = _run_installer_subprocess([sys.executable, "-m", "pip", "freeze"])
    if freeze_result.returncode == 0:
        freeze_path = run_dir / "pip-freeze.txt"
        freeze_path.write_text(freeze_result.stdout, encoding="utf-8")
        snapshot["pip_freeze"] = freeze_path.name
    if include_database:
        snapshot["database_backup"] = _snapshot_database(
            run_dir,
            database_url,
            backup_command=database_backup_command,
            restore_command=database_restore_command,
            restore_check_command=database_restore_check_command,
        )
    if webui_root is not None:
        snapshot["webui_root"] = str(webui_root)
        _snapshot_file(webui_root / "package.json", run_dir / "package.json.before", snapshot, "package_json")
        _snapshot_file(webui_root / "package-lock.json", run_dir / "package-lock.json.before", snapshot, "package_lock")
    return snapshot


def _snapshot_database(
    run_dir: Path,
    database_url: str | None,
    *,
    backup_command: str | None,
    restore_command: str | None,
    restore_check_command: str | None,
) -> dict[str, object]:
    if backup_command:
        if not restore_command:
            raise ModuleInstallerError("--database-backup-command requires --database-restore-command so rollback can restore the snapshot.")
        return _snapshot_external_database(
            run_dir,
            database_url,
            backup_command=backup_command,
            restore_command=restore_command,
            restore_check_command=restore_check_command,
        )
    if _sqlite_database_path(database_url) is not None:
        return _snapshot_sqlite_database(run_dir, database_url)
    raise ModuleInstallerError(
        "Automatic database backup for --migrate supports sqlite:/// URLs only. "
        "Provide --database-backup-command and --database-restore-command for this database."
    )


def _snapshot_sqlite_database(run_dir: Path, database_url: str | None) -> dict[str, object]:
    db_path = _sqlite_database_path(database_url)
    if db_path is None:
        raise ModuleInstallerError("Automatic database backup for --migrate currently supports sqlite:/// URLs only.")
    backup_path = run_dir / "database.sqlite.before"
    if db_path.exists():
        with closing(sqlite3.connect(str(db_path))) as source, closing(sqlite3.connect(str(backup_path))) as target:
            source.backup(target)
    else:
        backup_path.touch()
    with closing(sqlite3.connect(str(backup_path))) as candidate:
        row = candidate.execute("PRAGMA integrity_check").fetchone()
    integrity = str(row[0] if row else "missing result")
    if integrity.lower() != "ok":
        raise ModuleInstallerError(
            f"SQLite backup failed its restore-readiness integrity check: {integrity}"
        )
    return {
        "type": "sqlite",
        "source": str(db_path),
        "path": backup_path.name,
        "restore_check": {
            "type": "sqlite_integrity_check",
            "result": integrity,
        },
    }


def _snapshot_external_database(
    run_dir: Path,
    database_url: str | None,
    *,
    backup_command: str,
    restore_command: str,
    restore_check_command: str | None,
) -> dict[str, object]:
    backup_path = run_dir / "database.external.backup"
    metadata_path = run_dir / "database-backup-metadata.json"
    redactions = _installer_secret_redactions(database_url)
    database_url_secret = _write_installer_secret(run_dir / "database-url.secret", database_url) if database_url else None
    result = _run_database_hook(
        backup_command,
        run_dir=run_dir,
        database_url=database_url,
        backup_path=backup_path,
        metadata_path=metadata_path,
    )
    payload: dict[str, object] = {
        "type": "external",
        "database_url": _redact_installer_text(database_url or "", redactions=redactions),
        "backup_command": _redact_installer_text(backup_command, redactions=redactions),
        "restore_command": _redact_installer_text(restore_command, redactions=redactions),
        "restore_check_command": _redact_installer_text(restore_check_command, redactions=redactions) if restore_check_command else None,
        "backup_path": backup_path.name,
        "metadata_path": metadata_path.name,
        "return_code": result.returncode,
        "stdout": _redact_installer_text(result.stdout[-8000:], redactions=redactions),
        "stderr": _redact_installer_text(result.stderr[-8000:], redactions=redactions),
    }
    if database_url_secret:
        payload["database_url_secret"] = database_url_secret
    if result.returncode != 0:
        raise ModuleInstallerError(f"Database backup command failed ({result.returncode}): {_redact_installer_text(backup_command, redactions=redactions)}")
    if restore_check_command:
        restore_check = _run_database_hook(
            restore_check_command,
            run_dir=run_dir,
            database_url=database_url,
            backup_path=backup_path,
            metadata_path=metadata_path,
        )
        payload["restore_check"] = {
            "command": _redact_installer_text(restore_check_command, redactions=redactions),
            "return_code": restore_check.returncode,
            "stdout": _redact_installer_text(restore_check.stdout[-8000:], redactions=redactions),
            "stderr": _redact_installer_text(restore_check.stderr[-8000:], redactions=redactions),
        }
        if restore_check.returncode != 0:
            raise ModuleInstallerError(f"Database restore-check command failed ({restore_check.returncode}): {_redact_installer_text(restore_check_command, redactions=redactions)}")
    if metadata_path.exists():
        try:
            metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
            if isinstance(metadata, dict):
                payload["metadata"] = _redact_installer_value(metadata, redactions=redactions)
        except json.JSONDecodeError as exc:
            raise ModuleInstallerError(f"Database backup metadata is not valid JSON: {metadata_path}") from exc
    return payload


def _restore_database_snapshot(
    run_dir: Path,
    snapshot: Mapping[str, object],
    *,
    database_restore_command: str | None = None,
    database_url: str | None = None,
) -> dict[str, object] | None:
    raw = snapshot.get("database_backup")
    if not isinstance(raw, Mapping):
        return None
    if raw.get("type") == "external":
        return _restore_external_database_snapshot(
            run_dir,
            raw,
            database_restore_command=database_restore_command,
            database_url=database_url,
        )
    if raw.get("type") != "sqlite":
        return {"restored": False, "error": f"Unsupported database backup type: {raw.get('type')}"}
    source_name = raw.get("path")
    target_name = raw.get("source")
    if not isinstance(source_name, str) or not isinstance(target_name, str):
        return {"restored": False, "error": "Database backup metadata is incomplete."}
    backup_path = run_dir / source_name
    target_path = Path(target_name)
    if not backup_path.exists():
        return {"restored": False, "error": f"Database backup does not exist: {backup_path}"}
    target_path.parent.mkdir(parents=True, exist_ok=True)
    if backup_path.stat().st_size == 0:
        target_path.unlink(missing_ok=True)
        return {"restored": True, "type": "sqlite", "source": str(target_path), "empty": True}
    with closing(sqlite3.connect(str(backup_path))) as source, closing(sqlite3.connect(str(target_path))) as target:
        source.backup(target)
    return {"restored": True, "type": "sqlite", "source": str(target_path)}


def _restore_external_database_snapshot(
    run_dir: Path,
    raw: Mapping[str, object],
    *,
    database_restore_command: str | None,
    database_url: str | None,
) -> dict[str, object]:
    command = database_restore_command or raw.get("restore_command")
    if not isinstance(command, str) or not command.strip():
        return {"restored": False, "type": "external", "error": "No database restore command is available for this run."}
    backup_name = raw.get("backup_path")
    metadata_name = raw.get("metadata_path")
    raw_database_url = _database_url_from_external_snapshot(run_dir, raw)
    backup_path = run_dir / backup_name if isinstance(backup_name, str) else run_dir / "database.external.backup"
    metadata_path = run_dir / metadata_name if isinstance(metadata_name, str) else run_dir / "database-backup-metadata.json"
    effective_database_url = database_url or raw_database_url
    redactions = _installer_secret_redactions(effective_database_url)
    result = _run_database_hook(
        command,
        run_dir=run_dir,
        database_url=effective_database_url,
        backup_path=backup_path,
        metadata_path=metadata_path,
    )
    payload: dict[str, object] = {
        "restored": result.returncode == 0,
        "type": "external",
        "command": _redact_installer_text(command, redactions=redactions),
        "return_code": result.returncode,
        "stdout": _redact_installer_text(result.stdout[-8000:], redactions=redactions),
        "stderr": _redact_installer_text(result.stderr[-8000:], redactions=redactions),
    }
    if result.returncode != 0:
        payload["error"] = f"Database restore command failed ({result.returncode})."
    return payload


def _write_installer_secret(path: Path, value: str | None) -> str | None:
    if not value:
        return None
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    flags |= getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(path, flags, 0o600)
    except OSError as exc:
        raise ModuleInstallerError(f"Could not create private installer secret file: {path.name}") from exc
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            handle.write(value)
    except OSError as exc:
        path.unlink(missing_ok=True)
        raise ModuleInstallerError(f"Could not write private installer secret file: {path.name}") from exc

    try:
        mode = stat.S_IMODE(path.stat().st_mode)
    except OSError as exc:
        path.unlink(missing_ok=True)
        raise ModuleInstallerError(f"Could not verify private installer secret file: {path.name}") from exc
    if mode != 0o600:
        path.unlink(missing_ok=True)
        raise ModuleInstallerError(f"Installer secret file permissions are not private: {path.name}")
    return path.name


def _create_private_installer_run_dir(path: Path) -> None:
    path.mkdir(parents=True, mode=0o700, exist_ok=False)
    try:
        # mkdir's requested mode is still reduced by the process umask. Set the
        # exact owner-only mode so the directory remains usable with a strict
        # deployment umask while never retaining group/other access.
        path.chmod(0o700)
        mode = stat.S_IMODE(path.stat().st_mode)
    except OSError as exc:
        path.rmdir()
        raise ModuleInstallerError("Could not secure the installer run directory") from exc
    if mode != 0o700:
        path.rmdir()
        raise ModuleInstallerError("Installer run directory permissions are not exactly owner-only")


def _database_url_from_external_snapshot(run_dir: Path, raw: Mapping[str, object]) -> str | None:
    secret_name = raw.get("database_url_secret")
    if isinstance(secret_name, str) and secret_name:
        secret_path = run_dir / Path(secret_name).name
        try:
            value = secret_path.read_text(encoding="utf-8").strip()
        except OSError:
            value = ""
        if value:
            return value
    value = raw.get("database_url")
    if isinstance(value, str) and "<redacted>" not in value and "***" not in value:
        return value
    return None


def _installer_secret_redactions(database_url: str | None) -> dict[str, str]:
    redactions: dict[str, str] = {}
    for value in (database_url, _database_url_for_pgtools(database_url or "") if database_url else None):
        if not value:
            continue
        redacted = _redact_url_credentials(value)
        redactions[value] = redacted
    return redactions


def _redact_installer_value(value: object, *, redactions: Mapping[str, str]) -> object:
    if isinstance(value, str):
        return _redact_installer_text(value, redactions=redactions)
    if isinstance(value, Mapping):
        return {str(key): _redact_installer_value(item, redactions=redactions) for key, item in value.items()}
    if isinstance(value, list):
        return [_redact_installer_value(item, redactions=redactions) for item in value]
    return value


def _redact_installer_text(value: str, *, redactions: Mapping[str, str]) -> str:
    result = value
    for raw, redacted in sorted(redactions.items(), key=lambda item: len(item[0]), reverse=True):
        result = result.replace(raw, redacted)
    return _redact_url_credentials(result)


def _redact_url_credentials(value: str) -> str:
    try:
        return make_url(value).render_as_string(hide_password=True)
    except Exception:
        return re.sub(r"([A-Za-z][A-Za-z0-9+.-]*://[^\s/@:]+:)([^@\s]+)(@)", r"\1***\3", value)


def _run_structured_installer_command(command: Mapping[str, Any]) -> subprocess.CompletedProcess[str]:
    try:
        argv = _validated_installer_argv(command.get("argv"))
        cwd = _validated_installer_cwd(command.get("cwd"))
    except ValueError as exc:
        return subprocess.CompletedProcess(args=command.get("argv"), returncode=2, stdout="", stderr=str(exc))
    return _run_installer_subprocess(argv, cwd=cwd)


def _run_installer_subprocess(
    argv: Iterable[str],
    *,
    cwd: Path | str | None = None,
    env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    command = _validated_installer_argv(argv)
    effective_cwd = _validated_installer_cwd(cwd)
    try:
        # Audited subprocess choke point for the installer: callers must provide
        # a validated argv sequence and `shell=True` is never used here.
        return subprocess.run(  # noqa: S603 # nosec B603 # nosemgrep: python.lang.security.audit.dangerous-subprocess-use.dangerous-subprocess-use
            command,
            cwd=effective_cwd,
            env=dict(env) if env is not None else None,
            text=True,
            capture_output=True,
            check=False,
        )
    except OSError as exc:
        return subprocess.CompletedProcess(args=command, returncode=127, stdout="", stderr=str(exc))


def _validated_installer_argv(value: object) -> list[str]:
    if not isinstance(value, (list, tuple)):
        raise ValueError("Installer command argv must be a list or tuple")
    if not value:
        raise ValueError("Installer command argv must not be empty")
    argv: list[str] = []
    for raw in value:
        if not isinstance(raw, str):
            raise ValueError("Installer command argv entries must be strings")
        if "\x00" in raw:
            raise ValueError("Installer command argv entries must not contain NUL bytes")
        argv.append(raw)
    return argv


def _validated_installer_cwd(value: object) -> str | None:
    if value is None:
        return None
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, str):
        if "\x00" in value:
            raise ValueError("Installer command cwd must not contain NUL bytes")
        return value
    raise ValueError("Installer command cwd must be a path string")


def _run_database_hook(
    command: str,
    *,
    run_dir: Path,
    database_url: str | None,
    backup_path: Path,
    metadata_path: Path,
) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env.update({
        "GOVOPLAN_INSTALLER_RUN_DIR": str(run_dir),
        "GOVOPLAN_DATABASE_BACKUP_PATH": str(backup_path),
        "GOVOPLAN_DATABASE_BACKUP_METADATA": str(metadata_path),
    })
    if database_url:
        env["GOVOPLAN_DATABASE_URL"] = database_url
        pgtools_url = _database_url_for_pgtools(database_url)
        if pgtools_url:
            env["GOVOPLAN_DATABASE_URL_PGTOOLS"] = pgtools_url
        env.setdefault("DATABASE_URL", database_url)
    return _run_operator_command(command, cwd=run_dir, env=env)


def _run_operator_command(
    command: str,
    *,
    cwd: Path | None = None,
    env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    try:
        argv = _operator_command_argv(command)
    except ValueError as exc:
        return subprocess.CompletedProcess(args=command, returncode=2, stdout="", stderr=str(exc))
    return _run_installer_subprocess(argv, cwd=cwd, env=env)


def _operator_command_argv(command: str) -> tuple[str, ...]:
    try:
        argv = tuple(shlex.split(command))
    except ValueError as exc:
        raise ValueError(f"Invalid command syntax: {exc}") from exc
    if not argv:
        raise ValueError("Command is empty")
    if any(_operator_command_part_uses_shell(part) for part in argv):
        raise ValueError("Shell operators, substitutions, and redirects are not supported in installer hooks")
    if _looks_like_env_assignment(argv[0]):
        raise ValueError("Environment assignments must be configured through installer hook environment, not shell syntax")
    return argv


def _operator_command_part_uses_shell(value: str) -> bool:
    return value in {"|", "||", "&", "&&", ";", "<", ">", ">>", "2>", "2>>"} or "$(" in value or "`" in value


def _looks_like_env_assignment(value: str) -> bool:
    key, separator, _rest = value.partition("=")
    return bool(separator) and bool(re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", key))


def _database_url_for_pgtools(database_url: str) -> str | None:
    if database_url.startswith(("postgresql://", "postgres://")):
        return database_url
    if database_url.startswith("postgresql+"):
        return re.sub(r"^postgresql\+[^:]+://", "postgresql://", database_url)
    return None


def _sqlite_database_path(database_url: str | None) -> Path | None:
    if not database_url or not database_url.startswith("sqlite:///"):
        return None
    raw_path = database_url[len("sqlite:///"):]
    if raw_path in {"", ":memory:"}:
        return None
    return Path(raw_path).expanduser().resolve()


def _snapshot_file(source: Path, target: Path, snapshot: dict[str, object], key: str) -> None:
    if not source.exists():
        return
    shutil.copy2(source, target)
    snapshot[key] = target.name


def _restore_snapshot_file(run_dir: Path, snapshot: Mapping[str, object], key: str, target: Path) -> None:
    value = snapshot.get(key)
    if not isinstance(value, str):
        return
    source = run_dir / value
    if not source.exists():
        return
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(source, target)


def _mark_applied(item: ModuleInstallPlanItem) -> ModuleInstallPlanItem:
    if item.status != "planned":
        return item
    return ModuleInstallPlanItem(
        module_id=item.module_id,
        action=item.action,
        python_package=item.python_package,
        python_ref=item.python_ref,
        webui_package=item.webui_package,
        webui_ref=item.webui_ref,
        artifact_integrity=item.artifact_integrity,
        destroy_data=item.destroy_data,
        status="applied",
        notes=item.notes,
    )


def _structured_command(argv: list[str], *, cwd: Path | None = None, source: str) -> dict[str, Any]:
    command = _validated_installer_argv(argv)
    effective_cwd = _validated_installer_cwd(cwd)
    if not source or "\x00" in source:
        raise ValueError("Installer command source must be a non-empty string")
    return {
        "argv": command,
        "cwd": effective_cwd,
        "display": _display_command(command, cwd=effective_cwd),
        "source": source,
    }


def _display_command(argv: list[str], *, cwd: Path | str | None = None) -> str:
    command = " ".join(_quote_arg(arg) for arg in argv)
    if cwd is not None:
        return f"cd {_quote_arg(str(cwd))} && {command}"
    return command


def _quote_arg(value: str) -> str:
    import shlex

    return shlex.quote(value)


def _command_displays(commands: Iterable[Mapping[str, Any]], *, redactions: Mapping[str, str]) -> tuple[str, ...]:
    return tuple(_redact_installer_text(str(command["display"]), redactions=redactions) for command in commands)


def _command_record(command: Mapping[str, Any], *, redactions: Mapping[str, str] | None = None) -> dict[str, object]:
    effective_redactions = redactions or {}
    payload: dict[str, object] = {
        "display": _redact_installer_text(str(command["display"]), redactions=effective_redactions),
        "argv": [_redact_installer_text(str(item), redactions=effective_redactions) for item in command["argv"]],
        "source": _redact_installer_text(str(command.get("source") or "unknown"), redactions=effective_redactions),
    }
    if command.get("cwd"):
        payload["cwd"] = _redact_installer_text(str(command["cwd"]), redactions=effective_redactions)
    if command.get("migration_task_record_path"):
        payload["migration_task_record_path"] = _redact_installer_text(str(command["migration_task_record_path"]), redactions=effective_redactions)
    return payload


class _InstallerLock(AbstractContextManager[None]):
    def __init__(self, lock_path: Path) -> None:
        self.lock_path = lock_path
        self.fd: int | None = None

    def __enter__(self) -> None:
        self.lock_path.parent.mkdir(parents=True, exist_ok=True)
        try:
            self.fd = os.open(self.lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except FileExistsError as exc:
            raise ModuleInstallerError(f"Module installer lock already exists: {self.lock_path}") from exc
        os.write(self.fd, json.dumps({"pid": os.getpid(), "created_at": datetime.now(tz=UTC).isoformat()}).encode("utf-8"))
        return None

    def __exit__(self, exc_type, exc, traceback) -> bool | None:
        if self.fd is not None:
            os.close(self.fd)
        self.lock_path.unlink(missing_ok=True)
        return None


def _installer_lock(runtime_dir: Path) -> _InstallerLock:
    return _InstallerLock(runtime_dir / "install.lock")


def _request_queue_lock(runtime_dir: Path) -> _InstallerLock:
    return _InstallerLock(runtime_dir / "requests.lock")


def _daemon_status_path(runtime_dir: Path) -> Path:
    return runtime_dir / "daemon.status.json"


def _request_record_path(runtime_dir: Path, request_id: str) -> Path:
    return runtime_dir / "requests" / f"{request_id}.json"


def _run_id() -> str:
    return datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S%fZ")


def _write_json(path: Path, payload: Mapping[str, object]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")


def _process_is_running(pid: int) -> bool:
    if pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True
