Fence and reconcile module lifecycle effects

src/govoplan_core/commands/module_installer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+from importlib.metadata import PackageNotFoundError, version
 import json
 from pathlib import Path
 import sys
@@ -33,6 +34,10 @@ from govoplan_core.core.module_installer_notifications import (
     installer_notification_priority,
     installer_notification_subject,
 )
+from govoplan_core.core.runtime_coordination import (
+    bind_process_runtime_identity,
+    runtime_identity,
+)
 from govoplan_core.core.module_license import issue_module_license, module_license_diagnostics
 from govoplan_core.core.module_package_catalog import sign_module_package_catalog, validate_module_package_catalog
 from govoplan_core.core.module_management import (
@@ -107,11 +112,27 @@ def _build_parser() -> argparse.ArgumentParser:
 def main() -> int:
     args = _build_parser().parse_args()
     runtime_dir = args.runtime_dir or default_installer_runtime_dir(args.database_url)
+    bind_process_runtime_identity(
+        runtime_identity(
+            settings,
+            software_version=_core_version(),
+            role="installer",
+        )
+    )
     try:
         return _dispatch_command(args=args, runtime_dir=runtime_dir)
     except ModuleInstallerError as exc:
         print(f"error: {exc}", file=sys.stderr)
         return 1
+    finally:
+        bind_process_runtime_identity(None)
+
+
+def _core_version() -> str:
+    try:
+        return version("govoplan-core")
+    except PackageNotFoundError:
+        return "development"
 
 
 def _dispatch_command(*, args: argparse.Namespace, runtime_dir: Path) -> int:

src/govoplan_core/core/lifecycle.py

@@ -7,6 +7,11 @@ from threading import RLock
 from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, status
 
 from govoplan_core.core.module_management import ModuleManagementError, REQUIRED_PLATFORM_MODULES, plan_desired_enabled_modules
+from govoplan_core.core.module_lifecycle_recovery import (
+    ModuleLifecycleRecovery,
+    begin_runtime_graph_recovery,
+    canonical_sha256,
+)
 from govoplan_core.core.modules import ModuleContext, ModuleManifest
 from govoplan_core.core.registry import PlatformRegistry
 from govoplan_core.core.runtime import configure_runtime
@@ -99,30 +104,110 @@ class ModuleLifecycleManager:
             next_set = set(plan.enabled_modules)
             activated = tuple(module_id for module_id in plan.enabled_modules if module_id not in previous_set)
             deactivated = tuple(module_id for module_id in previous if module_id not in next_set)
+            graph_changes = bool(activated or deactivated)
+            recovery: ModuleLifecycleRecovery | None = None
+            if graph_changes or migrate:
+                from govoplan_core.db.session import get_database
 
-            if migrate:
-                self._migrate(plan.enabled_modules)
+                with get_database().session() as recovery_session:
+                    recovery = begin_runtime_graph_recovery(
+                        recovery_session,
+                        previous_modules=previous,
+                        requested_modules=plan.enabled_modules,
+                        migrate=migrate,
+                    )
 
-            mounted = tuple(module_id for module_id in plan.enabled_modules if self._mount_module_router(module_id))
+            old_manifests = {
+                manifest.id: manifest for manifest in self.registry.manifests()
+            }
+            try:
+                if recovery is not None:
+                    recovery.checkpoint(
+                        kind="runtime-graph-effect-started",
+                        summary="Runtime module graph entered its mutation boundary",
+                        evidence={
+                            "activated_sha256": canonical_sha256(activated),
+                            "deactivated_sha256": canonical_sha256(deactivated),
+                            "migrate": migrate,
+                        },
+                        effect_started=True,
+                    )
 
-            old_manifests = {manifest.id: manifest for manifest in self.registry.manifests()}
-            for module_id in deactivated:
-                hook = old_manifests[module_id].on_deactivate
-                if hook is not None:
-                    hook(self.context)
+                if migrate:
+                    self._migrate(plan.enabled_modules)
 
-            self.registry.replace(self.available_modules[module_id] for module_id in plan.enabled_modules)
-            self.configure_runtime()
+                mounted = tuple(module_id for module_id in plan.enabled_modules if self._mount_module_router(module_id))
 
-            for module_id in activated:
-                hook = self.available_modules[module_id].on_activate
-                if hook is not None:
-                    hook(self.context)
+                for module_id in deactivated:
+                    hook = old_manifests[module_id].on_deactivate
+                    if hook is not None:
+                        hook(self.context)
 
-            self.reconcile_workflow_definitions()
+                self.registry.replace(self.available_modules[module_id] for module_id in plan.enabled_modules)
+                self.configure_runtime()
 
-            if self._app is not None:
-                self._app.openapi_schema = None
+                for module_id in activated:
+                    hook = self.available_modules[module_id].on_activate
+                    if hook is not None:
+                        hook(self.context)
+
+                reconciliation = self.reconcile_workflow_definitions()
+
+                if self._app is not None:
+                    self._app.openapi_schema = None
+
+                if recovery is not None:
+                    from govoplan_core.db.session import get_database
+
+                    with get_database().session() as recovery_session:
+                        recovery.succeed(
+                            recovery_session,
+                            evidence={
+                                "active_graph_sha256": canonical_sha256(
+                                    self.active_module_ids()
+                                ),
+                                "mounted_graph_sha256": canonical_sha256(
+                                    self.mounted_module_ids()
+                                ),
+                                "workflow_reconciliation_sha256": canonical_sha256(
+                                    reconciliation
+                                ),
+                            },
+                            commit_projection=False,
+                        )
+            except Exception as exc:
+                self.registry.replace(old_manifests.values())
+                self.configure_runtime()
+                if self._app is not None:
+                    self._app.openapi_schema = None
+                if recovery is not None:
+                    from govoplan_core.db.session import get_database
+
+                    recovery.unresolved(
+                        summary="Runtime graph mutation did not reach verified completion",
+                        evidence={
+                            "error_type": type(exc).__name__,
+                            "previous_graph_sha256": canonical_sha256(previous),
+                            "registry_restored": True,
+                            "migrate": migrate,
+                        },
+                        outcome_unknown=migrate,
+                    )
+                    if not migrate:
+                        with get_database().session() as recovery_session:
+                            recovery.recovered(
+                                recovery_session,
+                                evidence={
+                                    "active_graph_sha256": canonical_sha256(
+                                        self.active_module_ids()
+                                    ),
+                                    "previous_graph_restored": (
+                                        self.active_module_ids() == previous
+                                    ),
+                                },
+                                summary="Previous runtime module graph was restored",
+                            )
+                raise
 
             return ModuleLifecycleResult(
                 enabled_modules=plan.enabled_modules,

src/govoplan_core/core/module_installer.py

@@ -27,6 +27,12 @@ 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_lifecycle_recovery import (
+    ModuleLifecycleRecovery,
+    ModuleLifecycleRecoveryError,
+    begin_module_installer_recovery,
+    canonical_sha256,
+)
 from govoplan_core.core.module_management import (
     PROTECTED_MODULES,
     ModuleInstallPlan,
@@ -268,6 +274,11 @@ class ModuleInstallerRunResult:
     return_code: int = 0
     error: str | None = None
     rollback: dict[str, object] | None = None
+    recovery: ModuleLifecycleRecovery | None = field(
+        default=None,
+        repr=False,
+        compare=False,
+    )
 
     def as_dict(self) -> dict[str, object]:
         payload: dict[str, object] = {
@@ -293,6 +304,7 @@ class _ModuleInstallRunState:
     result_commands: tuple[str, ...]
     record_redactions: tuple[str, ...]
     record: dict[str, Any]
+    recovery: ModuleLifecycleRecovery | None = None
 
 
 def default_installer_runtime_dir(database_url: str | None = None, *, cwd: Path | None = None) -> Path:
@@ -503,6 +515,7 @@ def run_module_install_plan(
     remove_uninstalled_modules_from_desired: bool = True,
     dry_run: bool = False,
     request_context: Mapping[str, object] | None = None,
+    finalize_recovery: bool = True,
 ) -> ModuleInstallerRunResult:
     maintenance_mode = saved_maintenance_mode(session)
     effective_runtime_dir = runtime_dir or default_installer_runtime_dir(database_url)
@@ -520,6 +533,7 @@ def run_module_install_plan(
         raise ModuleInstallerError("Install preflight is blocked: " + "; ".join(issue.message for issue in preflight.issues if issue.severity == "blocker"))
 
     state = _prepare_module_install_run(
+        session=session,
         plan=plan,
         preflight=preflight,
         database_url=database_url,
@@ -550,6 +564,7 @@ def run_module_install_plan(
 
     if failed_error is not None:
         return _failed_module_install_run_result(
+            session=session,
             state=state,
             plan=plan,
             executed=executed,
@@ -569,11 +584,13 @@ def run_module_install_plan(
         remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
         executed=executed,
         state=state,
+        finalize_recovery=finalize_recovery,
     )
 
 
 def _prepare_module_install_run(
     *,
+    session: Session,
     plan: ModuleInstallPlan,
     preflight: ModuleInstallerPreflight,
     database_url: str,
@@ -605,13 +622,39 @@ def _prepare_module_install_run(
         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(
+    recovery: ModuleLifecycleRecovery | None = None
+    if not dry_run:
+        try:
+            recovery = begin_module_installer_recovery(
+                session,
+                run_id=run_id,
+                plan=tuple(item.as_dict() for item in plan.items),
+                command_count=len(commands),
+                migrate_database=migrate_database,
+                destructive_retirement=_destructive_retirement_requested(plan),
+                snapshot_sha256=None,
+                backup_reference=(
+                    f"module-installer:{run_id}:database-backup"
+                    if _destructive_retirement_requested(plan)
+                    else None
+                ),
+                request_context_sha256=canonical_sha256(dict(request_context or {})),
+            )
+            recovery.checkpoint(
+                kind="snapshot-started",
+                summary="Installer environment snapshot started before package effects",
+                evidence={
+                    "run_id": run_id,
+                    "database_backup_expected": bool(
+                        migrate_database or _destructive_retirement_requested(plan)
+                    ),
+                },
+            )
+        except ModuleLifecycleRecoveryError as exc:
+            raise ModuleInstallerError(str(exc)) from exc
+
+    try:
+        snapshot = _snapshot_environment(
             run_dir,
             webui_root=webui_root,
             database_url=database_url,
@@ -619,7 +662,32 @@ def _prepare_module_install_run(
             database_backup_command=database_backup_command,
             database_restore_command=database_restore_command,
             database_restore_check_command=database_restore_check_command,
-        ),
+        )
+    except Exception as exc:
+        if recovery is not None:
+            recovery.unresolved(
+                summary="Installer snapshot preparation failed before package effects",
+                evidence={"snapshot_error_type": type(exc).__name__},
+                outcome_unknown=False,
+            )
+        raise
+
+    if recovery is not None:
+        recovery.checkpoint(
+            kind="snapshot-verified",
+            summary="Installer environment snapshot and backup evidence were verified",
+            evidence={
+                "snapshot_sha256": canonical_sha256(snapshot),
+                **_database_backup_recovery_evidence(snapshot),
+            },
+        )
+    record = _initial_module_install_record(
+        run_id=run_id,
+        plan=plan,
+        preflight=preflight,
+        commands=commands,
+        record_redactions=record_redactions,
+        snapshot=snapshot,
         build_webui=build_webui,
         migrate_database=migrate_database,
         activate_installed_modules=activate_installed_modules,
@@ -627,6 +695,8 @@ def _prepare_module_install_run(
         dry_run=dry_run,
         request_context=request_context,
     )
+    if recovery is not None:
+        record["recovery"] = _module_lifecycle_recovery_record(recovery)
     record_path = run_dir / "record.json"
     _write_json(record_path, record)
     return _ModuleInstallRunState(
@@ -637,6 +707,7 @@ def _prepare_module_install_run(
         result_commands=_command_displays(commands, redactions=record_redactions),
         record_redactions=record_redactions,
         record=record,
+        recovery=recovery,
     )
 
 
@@ -673,6 +744,40 @@ def _initial_module_install_record(
     return record
 
 
+def _module_lifecycle_recovery_record(
+    recovery: ModuleLifecycleRecovery,
+    *,
+    status: str = "running",
+) -> dict[str, object]:
+    return {
+        "operation_id": recovery.operation_id,
+        "operation_type": recovery.operation_type,
+        "mode": recovery.mode.value,
+        "plan_sha256": recovery.plan_sha256,
+        "replayed": recovery.replayed,
+        "status": status,
+    }
+
+
+def _database_backup_recovery_evidence(
+    snapshot: Mapping[str, object],
+) -> dict[str, object]:
+    backup = snapshot.get("database_backup")
+    if not isinstance(backup, Mapping):
+        return {"database_backup_present": False}
+    sha256 = str(backup.get("artifact_sha256") or "").strip()
+    return {
+        "database_backup_present": True,
+        "database_backup_type": str(backup.get("type") or "unknown"),
+        "database_backup_sha256": sha256 or "unavailable",
+        "database_backup_size_bytes": int(backup.get("size_bytes") or 0),
+        "database_backup_reference": (
+            f"sha256:{sha256}" if sha256 else "unavailable"
+        ),
+        "restore_check_sha256": canonical_sha256(backup.get("restore_check")),
+    }
+
+
 def _execute_module_install_run(
     *,
     session: Session,
@@ -685,14 +790,91 @@ def _execute_module_install_run(
     failed_error: str | None = None
     with _installer_lock(effective_runtime_dir):
         try:
+            if state.recovery is not None:
+                state.recovery.checkpoint(
+                    kind="effects-starting",
+                    summary="Installer acquired local and distributed execution fences",
+                    evidence={
+                        "command_count": len(state.commands),
+                        "destructive_retirement": _destructive_retirement_requested(plan),
+                    },
+                )
+                if _destructive_retirement_requested(plan):
+                    state.recovery.checkpoint(
+                        kind="retirement-effect-started",
+                        summary="Destructive module retirement entered its effect boundary",
+                        evidence={
+                            "retirement_plan_sha256": canonical_sha256(
+                                [
+                                    item.as_dict()
+                                    for item in plan.items
+                                    if item.destroy_data
+                                ]
+                            ),
+                        },
+                        effect_started=True,
+                    )
             _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))
+            for index, command in enumerate(state.commands):
+                if state.recovery is not None:
+                    command_record = _command_record(
+                        command,
+                        redactions=state.record_redactions,
+                    )
+                    state.recovery.checkpoint(
+                        kind="command-effect-started",
+                        summary="Installer command entered its effect boundary",
+                        evidence={
+                            "command_index": index,
+                            "command_source": str(command.get("source") or "unknown"),
+                            "command_sha256": canonical_sha256(command_record),
+                        },
+                        effect_started=True,
+                    )
+                command_result = _run_module_install_command(command, state=state)
+                executed.append(command_result)
+                if state.recovery is not None:
+                    state.recovery.checkpoint(
+                        kind="command-result-verified",
+                        summary="Installer command returned a conclusive successful result",
+                        evidence={
+                            "command_index": index,
+                            "return_code": int(command_result["return_code"]),
+                            "result_sha256": canonical_sha256(command_result),
+                        },
+                    )
                 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)
+            if state.recovery is not None:
+                outcome_unknown = not isinstance(exc, ModuleInstallerError)
+                try:
+                    state.recovery.unresolved(
+                        summary="Module installer effects did not reach verified completion",
+                        evidence={
+                            "error_type": type(exc).__name__,
+                            "completed_command_count": len(executed),
+                        },
+                        outcome_unknown=outcome_unknown,
+                    )
+                    state.record["recovery"] = _module_lifecycle_recovery_record(
+                        state.recovery,
+                        status=(
+                            "outcome_unknown"
+                            if outcome_unknown
+                            else "recovery_required"
+                            if state.recovery.effect_started
+                            else "failed"
+                        ),
+                    )
+                except Exception as recovery_exc:
+                    state.record["recovery_error"] = type(recovery_exc).__name__
+                    failed_error = (
+                        f"{failed_error}; recovery ledger transition failed: "
+                        f"{type(recovery_exc).__name__}"
+                    )
     return executed, failed_error
 
 
@@ -740,6 +922,7 @@ def _rollback_session_after_module_install_error(session: Session, exc: Exceptio
 
 def _failed_module_install_run_result(
     *,
+    session: Session,
     state: _ModuleInstallRunState,
     plan: ModuleInstallPlan,
     executed: list[dict[str, object]],
@@ -765,6 +948,7 @@ def _failed_module_install_run_result(
             commands=state.result_commands,
             return_code=1,
             error=failed_error,
+            recovery=state.recovery,
         )
     rollback = rollback_module_install_run(
         run_id=state.run_id,
@@ -775,6 +959,30 @@ def _failed_module_install_run_result(
         database_url=database_url,
     )
     _update_run_record(state.record_path, {"destructive_retirement_rollback": rollback.as_dict()})
+    if rollback.return_code == 0 and state.recovery is not None:
+        try:
+            state.recovery.recovered(
+                session,
+                evidence={
+                    "rollback_return_code": rollback.return_code,
+                    "rollback_sha256": canonical_sha256(rollback.as_dict()),
+                },
+                summary="Verified rollback restored the pre-install module state",
+            )
+            _update_run_record(
+                state.record_path,
+                {
+                    "recovery": _module_lifecycle_recovery_record(
+                        state.recovery,
+                        status="recovered",
+                    )
+                },
+            )
+        except Exception as recovery_exc:
+            _update_run_record(
+                state.record_path,
+                {"recovery_error": type(recovery_exc).__name__},
+            )
     return ModuleInstallerRunResult(
         run_id=state.run_id,
         status="rolled-back" if rollback.return_code == 0 else "failed",
@@ -783,6 +991,7 @@ def _failed_module_install_run_result(
         return_code=1,
         error=failed_error,
         rollback=rollback.as_dict(),
+        recovery=state.recovery,
     )
 
 
@@ -795,6 +1004,7 @@ def _applied_module_install_run_result(
     remove_uninstalled_modules_from_desired: bool,
     executed: list[dict[str, object]],
     state: _ModuleInstallRunState,
+    finalize_recovery: bool,
 ) -> 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:
@@ -806,14 +1016,50 @@ def _applied_module_install_run_result(
         )
         save_desired_enabled_modules(session, next_desired)
         state.record["desired_enabled_after"] = list(next_desired)
-    session.commit()
+    recovery_evidence = {
+        "command_count": len(executed),
+        "command_results_sha256": canonical_sha256(executed),
+        "desired_graph_sha256": canonical_sha256(
+            state.record.get("desired_enabled_after", list(desired_enabled))
+        ),
+        "plan_projection_sha256": canonical_sha256(
+            [item.as_dict() for item in plan.items]
+        ),
+    }
+    if state.recovery is not None and finalize_recovery:
+        state.recovery.succeed(
+            session,
+            evidence=recovery_evidence,
+            commit_projection=True,
+        )
+        recovery_status = "succeeded"
+    else:
+        session.commit()
+        recovery_status = "awaiting_supervisor" if state.recovery is not None else None
+        if state.recovery is not None:
+            state.recovery.checkpoint(
+                kind="local-projection-committed",
+                summary="Package and desired-graph projections await runtime health verification",
+                evidence=recovery_evidence,
+            )
     state.record.update({
         "status": "applied",
         "finished_at": datetime.now(tz=UTC).isoformat(),
         "commands": executed,
     })
+    if state.recovery is not None and recovery_status is not None:
+        state.record["recovery"] = _module_lifecycle_recovery_record(
+            state.recovery,
+            status=recovery_status,
+        )
     _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)
+    return ModuleInstallerRunResult(
+        run_id=state.run_id,
+        status="applied",
+        record_path=state.record_path,
+        commands=state.result_commands,
+        recovery=state.recovery,
+    )
 
 
 def supervise_module_install_plan(
@@ -864,6 +1110,7 @@ def supervise_module_install_plan(
         remove_uninstalled_modules_from_desired=remove_uninstalled_modules_from_desired,
         dry_run=False,
         request_context=request_context,
+        finalize_recovery=False,
     )
     supervisor: dict[str, object] = {
         "started_at": datetime.now(tz=UTC).isoformat(),
@@ -938,6 +1185,27 @@ def supervise_module_install_plan(
         "status": "ok",
         "finished_at": datetime.now(tz=UTC).isoformat(),
     })
+    if result.recovery is not None:
+        result.recovery.succeed(
+            session,
+            evidence={
+                "restart_results_sha256": canonical_sha256(restart_results),
+                "health_results_sha256": canonical_sha256(supervisor.get("health")),
+                "runtime_health_verified": True,
+            },
+            commit_projection=False,
+        )
+        supervisor["recovery_operation_id"] = result.recovery.operation_id
+        supervisor["recovery_status"] = "succeeded"
+        _update_run_record(
+            result.record_path,
+            {
+                "recovery": _module_lifecycle_recovery_record(
+                    result.recovery,
+                    status="succeeded",
+                )
+            },
+        )
     _update_run_record(result.record_path, {"supervisor": supervisor})
     return result
 
@@ -3264,6 +3532,31 @@ def _rollback_after_supervisor_failure(
         session.commit()
     supervisor["rollback"] = rollback.as_dict()
 
+    if rollback.return_code == 0 and result.recovery is not None:
+        try:
+            result.recovery.recovered(
+                session,
+                evidence={
+                    "rollback_sha256": canonical_sha256(rollback.as_dict()),
+                    "desired_graph_restored": True,
+                },
+                summary="Supervisor rollback restored package and desired module state",
+            )
+            supervisor["recovery_operation_id"] = result.recovery.operation_id
+            supervisor["recovery_status"] = "recovered"
+            _update_run_record(
+                result.record_path,
+                {
+                    "recovery": _module_lifecycle_recovery_record(
+                        result.recovery,
+                        status="recovered",
+                    )
+                },
+            )
+        except Exception as recovery_exc:
+            supervisor["recovery_status"] = "reconciliation-failed"
+            supervisor["recovery_error"] = type(recovery_exc).__name__
+
     rollback_restart = _run_restart_commands(restart_commands)
     if rollback_restart:
         supervisor["rollback_restart_commands"] = rollback_restart
@@ -3284,6 +3577,7 @@ def _rollback_after_supervisor_failure(
         return_code=1,
         error=reason,
         rollback=rollback.as_dict(),
+        recovery=result.recovery,
     )
 
 
@@ -3698,10 +3992,13 @@ def _snapshot_sqlite_database(run_dir: Path, database_url: str | None) -> dict[s
         raise ModuleInstallerError(
             f"SQLite backup failed its restore-readiness integrity check: {integrity}"
         )
+    artifact_sha256 = _sha256_file(backup_path)
     return {
         "type": "sqlite",
         "source": str(db_path),
         "path": backup_path.name,
+        "artifact_sha256": artifact_sha256,
+        "size_bytes": backup_path.stat().st_size,
         "restore_check": {
             "type": "sqlite_integrity_check",
             "result": integrity,
@@ -3744,6 +4041,12 @@ def _snapshot_external_database(
         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 not backup_path.is_file() or backup_path.stat().st_size <= 0:
+        raise ModuleInstallerError(
+            "Database backup command did not create a non-empty backup artifact."
+        )
+    payload["artifact_sha256"] = _sha256_file(backup_path)
+    payload["size_bytes"] = backup_path.stat().st_size
     if restore_check_command:
         restore_check = _run_database_hook(
             restore_check_command,

src/govoplan_core/core/module_lifecycle_recovery.py

@@ -0,0 +1,457 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import json
+from typing import Mapping, Sequence
+from uuid import uuid4
+
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryOperation,
+    RecoveryPlan,
+    RecoveryStatus,
+)
+from govoplan_core.core.recovery_runtime import (
+    DurableRecoveryOperation,
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+    claim_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import process_runtime_identity
+
+
+class ModuleLifecycleRecoveryError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class ModuleLifecycleRecoveryDeclaration:
+    operation_type: str
+    mode: RecoveryMode
+    resources: tuple[str, ...]
+    verification: tuple[str, ...]
+
+
+MODULE_LIFECYCLE_RECOVERY_OPERATIONS = (
+    ModuleLifecycleRecoveryDeclaration(
+        operation_type="module-lifecycle.pre-migration",
+        mode=RecoveryMode.COMPENSATION,
+        resources=("postgresql", "package-environment", "webui-bundle", "filesystem"),
+        verification=(
+            "verify the canonical install plan and immutable package references",
+            "verify the package and WebUI snapshots before mutation",
+            "verify the installed manifests and desired module graph",
+        ),
+    ),
+    ModuleLifecycleRecoveryDeclaration(
+        operation_type="module-lifecycle.post-migration",
+        mode=RecoveryMode.FORWARD_RECOVERY,
+        resources=(
+            "postgresql",
+            "package-environment",
+            "webui-bundle",
+            "runtime-nodes",
+        ),
+        verification=(
+            "verify the backup reference and migration execution evidence",
+            "verify migration heads and installed module manifests",
+            "verify the desired graph and runtime health before completion",
+        ),
+    ),
+    ModuleLifecycleRecoveryDeclaration(
+        operation_type="module-retirement.destroy-data",
+        mode=RecoveryMode.SNAPSHOT_RESTORE,
+        resources=("postgresql", "object-storage", "package-environment"),
+        verification=(
+            "verify the pinned backup artifact and restore-readiness evidence",
+            "verify the retirement provider result and remaining migration state",
+            "verify the installed manifests and desired module graph",
+        ),
+    ),
+    ModuleLifecycleRecoveryDeclaration(
+        operation_type="module-runtime.apply-graph",
+        mode=RecoveryMode.COMPENSATION,
+        resources=("postgresql", "runtime-nodes", "module-registry"),
+        verification=(
+            "verify the requested graph against available module contracts",
+            "verify activation and deactivation hooks completed",
+            "verify the active graph and workflow contribution reconciliation",
+        ),
+    ),
+)
+
+
+_DECLARATIONS = {
+    item.operation_type: item for item in MODULE_LIFECYCLE_RECOVERY_OPERATIONS
+}
+
+
+def canonical_sha256(value: object) -> str:
+    encoded = json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+        default=str,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def lifecycle_session_factory(session: Session) -> sessionmaker[Session]:
+    bind = session.get_bind()
+    if bind is None:
+        raise ModuleLifecycleRecoveryError(
+            "Module lifecycle recovery requires a database bind"
+        )
+    return sessionmaker(bind=bind, expire_on_commit=False)
+
+
+@dataclass(slots=True)
+class ModuleLifecycleRecovery:
+    operation: DurableRecoveryOperation | None
+    operation_id: str
+    operation_type: str
+    mode: RecoveryMode
+    plan_sha256: str
+    replayed: bool
+    effect_started: bool = False
+
+    def checkpoint(
+        self,
+        *,
+        kind: str,
+        summary: str,
+        evidence: Mapping[str, object],
+        effect_started: bool = False,
+    ) -> None:
+        if self.operation is None:
+            return
+        self.effect_started = self.effect_started or effect_started
+        self.operation.checkpoint(
+            kind=kind,
+            summary=summary,
+            evidence={
+                **dict(evidence),
+                "effect_started": self.effect_started,
+                "plan_sha256": self.plan_sha256,
+            },
+        )
+
+    def succeed(
+        self,
+        session: Session,
+        *,
+        evidence: Mapping[str, object],
+        commit_projection: bool,
+    ) -> None:
+        if self.operation is None:
+            return
+        terminal = {
+            "verified": True,
+            "checks": {
+                **dict(evidence),
+                "plan_sha256": self.plan_sha256,
+                "effect_started": self.effect_started,
+            },
+        }
+        if commit_projection:
+            self.operation.commit_verified_success(session, evidence=terminal)
+        else:
+            self.operation.succeed(evidence=terminal)
+
+    def unresolved(
+        self,
+        *,
+        summary: str,
+        evidence: Mapping[str, object],
+        outcome_unknown: bool,
+    ) -> None:
+        if self.operation is None:
+            return
+        if not self.effect_started:
+            self.operation.fail(
+                summary=summary,
+                evidence={
+                    "verified": True,
+                    "checks": {
+                        **dict(evidence),
+                        "effect_started": False,
+                    },
+                },
+            )
+            return
+        self.operation.unresolved(
+            status=(
+                RecoveryStatus.OUTCOME_UNKNOWN
+                if outcome_unknown
+                else RecoveryStatus.RECOVERY_REQUIRED
+            ),
+            summary=summary,
+            evidence={
+                **dict(evidence),
+                "effect_started": True,
+                "plan_sha256": self.plan_sha256,
+            },
+            failure_summary=(
+                "Inspect the installer run record and affected state services "
+                "before retrying or restoring"
+            ),
+        )
+
+    def recovered(
+        self,
+        session: Session,
+        *,
+        evidence: Mapping[str, object],
+        summary: str,
+    ) -> None:
+        state = session.get(RecoveryOperation, self.operation_id)
+        if state is None:
+            raise ModuleLifecycleRecoveryError(
+                "Module lifecycle recovery operation is unavailable"
+            )
+        if state.status == RecoveryStatus.RECOVERED.value:
+            return
+        try:
+            handle = claim_durable_recovery_operation(
+                lifecycle_session_factory(session),
+                identity=process_runtime_identity(),
+                operation_id=self.operation_id,
+                lease_ttl_seconds=900,
+            )
+        except RecoveryOperationStateConflict as exc:
+            if exc.status == RecoveryStatus.RECOVERED.value:
+                return
+            raise ModuleLifecycleRecoveryError(
+                f"Module lifecycle recovery is already {exc.status}"
+            ) from exc
+        except (RecoveryOperationBusy, RecoveryGuaranteeError, RuntimeError) as exc:
+            raise ModuleLifecycleRecoveryError(
+                "Module lifecycle recovery authority is unavailable"
+            ) from exc
+        session.expire_all()
+        state = session.get(RecoveryOperation, self.operation_id)
+        if state is None:
+            raise ModuleLifecycleRecoveryError(
+                "Module lifecycle recovery operation is unavailable"
+            )
+        recovery_evidence = {
+            "verified": True,
+            "checks": {
+                **dict(evidence),
+                "plan_sha256": self.plan_sha256,
+            },
+        }
+        if state.status == RecoveryStatus.OUTCOME_UNKNOWN.value:
+            handle.resolve_unknown(
+                effect_occurred=False,
+                evidence=recovery_evidence,
+                summary=summary,
+            )
+        else:
+            handle.compensate(
+                failure_summary=summary,
+                failure_evidence={
+                    "effect_started": self.effect_started,
+                    "plan_sha256": self.plan_sha256,
+                },
+                recovery_evidence=recovery_evidence,
+            )
+
+
+def begin_module_installer_recovery(
+    session: Session,
+    *,
+    run_id: str,
+    plan: Sequence[Mapping[str, object]],
+    command_count: int,
+    migrate_database: bool,
+    destructive_retirement: bool,
+    snapshot_sha256: str | None,
+    backup_reference: str | None,
+    request_context_sha256: str,
+) -> ModuleLifecycleRecovery:
+    operation_type = (
+        "module-retirement.destroy-data"
+        if destructive_retirement
+        else "module-lifecycle.post-migration"
+        if migrate_database
+        else "module-lifecycle.pre-migration"
+    )
+    declaration = _DECLARATIONS[operation_type]
+    plan_sha256 = canonical_sha256([dict(item) for item in plan])
+    recovery_plan = RecoveryPlan(
+        mode=declaration.mode,
+        preconditions=(
+            "maintenance mode and installer preflight are current",
+            "package references and the requested module graph are pinned",
+            "the deployment-wide module lifecycle fence is owned",
+        ),
+        compensation_steps=(
+            "restore the Python and WebUI package snapshots",
+            "restore the prior desired module graph",
+            "verify installed manifests and runtime health",
+        )
+        if declaration.mode == RecoveryMode.COMPENSATION
+        else (),
+        forward_recovery_steps=(
+            "inspect migration task and command evidence",
+            "complete or repair migrations under the same deployment fence",
+            "verify migration heads, manifests, desired graph, and runtime health",
+        )
+        if declaration.mode == RecoveryMode.FORWARD_RECOVERY
+        else (),
+        verification_steps=declaration.verification,
+        backup_reference=(
+            backup_reference
+            if declaration.mode == RecoveryMode.SNAPSHOT_RESTORE
+            else None
+        ),
+    )
+    if declaration.mode == RecoveryMode.SNAPSHOT_RESTORE and not backup_reference:
+        raise ModuleLifecycleRecoveryError(
+            "Destructive module retirement requires verified backup evidence"
+        )
+    session.commit()
+    try:
+        started = begin_durable_recovery_operation(
+            lifecycle_session_factory(session),
+            identity=process_runtime_identity(),
+            module_id="core",
+            operation_type=operation_type,
+            idempotency_key=f"module-installer:{run_id}",
+            request={
+                "run_id": run_id,
+                "plan_sha256": plan_sha256,
+                "command_count": command_count,
+                "migrate_database": migrate_database,
+                "destructive_retirement": destructive_retirement,
+                "snapshot_expected": True,
+                "request_context_sha256": request_context_sha256,
+            },
+            recovery_plan=recovery_plan,
+            precondition_evidence={
+                "plan_sha256": plan_sha256,
+                "snapshot_sha256": snapshot_sha256 or "pending",
+                "request_context_sha256": request_context_sha256,
+                "command_count": command_count,
+                "backup_reference_present": bool(backup_reference),
+            },
+            lease_resource_key="core:module-lifecycle:deployment",
+            lease_ttl_seconds=900,
+            resource_type="module_installer_run",
+            resource_id=run_id,
+            metadata={
+                "resources": list(declaration.resources),
+                "migrate_database": migrate_database,
+                "destructive_retirement": destructive_retirement,
+            },
+            block_unresolved_resource=True,
+        )
+    except RecoveryOperationBusy as exc:
+        raise ModuleLifecycleRecoveryError(
+            "Another runtime owns the deployment module lifecycle fence"
+        ) from exc
+    except RecoveryOperationStateConflict as exc:
+        raise ModuleLifecycleRecoveryError(
+            f"Module installer recovery is already {exc.status}"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError, SQLAlchemyError, ValueError) as exc:
+        raise ModuleLifecycleRecoveryError(
+            "The recovery ledger is unavailable; module mutation did not start"
+        ) from exc
+    return ModuleLifecycleRecovery(
+        operation=started.operation,
+        operation_id=started.operation_id,
+        operation_type=operation_type,
+        mode=declaration.mode,
+        plan_sha256=plan_sha256,
+        replayed=started.replayed,
+    )
+
+
+def begin_runtime_graph_recovery(
+    session: Session,
+    *,
+    previous_modules: Sequence[str],
+    requested_modules: Sequence[str],
+    migrate: bool,
+) -> ModuleLifecycleRecovery:
+    declaration = _DECLARATIONS["module-runtime.apply-graph"]
+    plan = {
+        "previous_modules": sorted(set(previous_modules)),
+        "requested_modules": sorted(set(requested_modules)),
+        "migrate": migrate,
+    }
+    plan_sha256 = canonical_sha256(plan)
+    session.commit()
+    try:
+        started = begin_durable_recovery_operation(
+            lifecycle_session_factory(session),
+            identity=process_runtime_identity(),
+            module_id="core",
+            operation_type=declaration.operation_type,
+            idempotency_key=f"module-runtime:{uuid4()}",
+            request={**plan, "plan_sha256": plan_sha256},
+            recovery_plan=RecoveryPlan(
+                mode=declaration.mode,
+                preconditions=(
+                    "the requested graph passed module contract validation",
+                    "the deployment-wide module lifecycle fence is owned",
+                ),
+                compensation_steps=(
+                    "restore the previous in-process active registry",
+                    "reconfigure capability contexts from the previous graph",
+                ),
+                verification_steps=declaration.verification,
+            ),
+            precondition_evidence={
+                "plan_sha256": plan_sha256,
+                "previous_graph_sha256": canonical_sha256(
+                    sorted(set(previous_modules))
+                ),
+                "requested_graph_sha256": canonical_sha256(
+                    sorted(set(requested_modules))
+                ),
+            },
+            lease_resource_key="core:module-lifecycle:deployment",
+            lease_ttl_seconds=300,
+            resource_type="module_runtime_graph",
+            resource_id=plan_sha256,
+            metadata={"resources": list(declaration.resources)},
+            block_unresolved_resource=True,
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        raise ModuleLifecycleRecoveryError(
+            "Another lifecycle mutation is active or unresolved"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError, SQLAlchemyError, ValueError) as exc:
+        raise ModuleLifecycleRecoveryError(
+            "The recovery ledger is unavailable; the active graph was unchanged"
+        ) from exc
+    return ModuleLifecycleRecovery(
+        operation=started.operation,
+        operation_id=started.operation_id,
+        operation_type=declaration.operation_type,
+        mode=declaration.mode,
+        plan_sha256=plan_sha256,
+        replayed=started.replayed,
+    )
+
+
+__all__ = [
+    "MODULE_LIFECYCLE_RECOVERY_OPERATIONS",
+    "ModuleLifecycleRecovery",
+    "ModuleLifecycleRecoveryDeclaration",
+    "ModuleLifecycleRecoveryError",
+    "begin_module_installer_recovery",
+    "begin_runtime_graph_recovery",
+    "canonical_sha256",
+    "lifecycle_session_factory",
+]

src/govoplan_core/core/recovery_runtime.py

@@ -512,6 +512,7 @@ def begin_durable_recovery_operation(
     resource_type: str | None = None,
     resource_id: str | None = None,
     metadata: dict[str, Any] | None = None,
+    block_unresolved_resource: bool = False,
 ) -> DurableRecoveryStart:
     if lease_ttl_seconds < 1:
         raise ValueError("Recovery lease TTL must be at least one second")
@@ -539,6 +540,31 @@ def begin_durable_recovery_operation(
                 RecoveryOperation.idempotency_key == idempotency_key,
             )
         ).scalar_one_or_none()
+        if block_unresolved_resource:
+            blocking = session.execute(
+                select(RecoveryOperation).where(
+                    RecoveryOperation.installation_id == identity.installation_id,
+                    RecoveryOperation.lease_resource_key == lease_resource_key,
+                    RecoveryOperation.status.not_in(
+                        (
+                            RecoveryStatus.SUCCEEDED.value,
+                            RecoveryStatus.REJECTED.value,
+                            RecoveryStatus.FAILED.value,
+                            RecoveryStatus.RECOVERED.value,
+                            RecoveryStatus.MANUAL_INTERVENTION.value,
+                        )
+                    ),
+                )
+            ).scalars().first()
+            if blocking is not None and (
+                existing is None or blocking.id != existing.id
+            ):
+                release_lease(session, claim)
+                session.commit()
+                raise RecoveryOperationStateConflict(
+                    blocking.id,
+                    blocking.status,
+                )
         operation = plan_recovery_operation(
             session,
             installation_id=identity.installation_id,