Expose sanitized backup restore evidence

README.md

@@ -19,6 +19,8 @@ This repository owns:
   deployment-rendered PostgreSQL connection budget
 - audited API and worker drain/cancel controls
 - recovery-operation status and evidence-chain summaries
+- sanitized, expiry-aware coordinated backup and isolated-restore status from
+  the deployment verifier, without backup artifact or key-custody references
 - governance inventory for module-declared permissions, roles, capabilities,
   policies, documentation, access-control hooks, search providers, and
   migration ownership

docs/BACKUP_EVIDENCE_STATUS.md

@@ -0,0 +1,22 @@
+# Backup Evidence Status
+
+The Ops health surface reports the sanitized result of deployment-side backup
+verification. It never reads backup artifacts, provider URLs, credentials,
+encryption-key references, signing keys, or orchestrator APIs.
+
+The status can be:
+
+- `verified`: a signed coordinated PostgreSQL, object, configuration, and key
+  custody recovery point and its isolated restore drill are current;
+- `expired`: the retained receipt is no longer fresh enough to authorize a
+  release-changing migration;
+- `absent`: no signed receipt has been adopted;
+- `invalid`: the deployment verifier rejected, lost, or only partially received
+  the evidence set.
+
+The panel shows the recovery point and drill identifiers, capture/expiry times,
+component count, and measured RPO/RTO. These values help operators find the
+private report in the approved evidence store; they are not themselves a
+backup. Resolve an absent, expired, or invalid state through the deployment
+runbook and `govoplan-deploy verify-backup --adopt`, then reconcile the runtime
+environment. Do not upload provider reports or keys through the application.

src/govoplan_ops/backend/api/v1/routes.py

@@ -5,6 +5,7 @@ import shutil
 import threading
 import time
 from collections.abc import Mapping
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 from urllib.parse import urlsplit
@@ -259,9 +260,7 @@ def _ops_status_payload(
                 "requires_attention", 0
             ),
             "failed_operation_count": int(recovery_metrics.get("failed") or 0),
-            "outcome_unknown_count": int(
+            "outcome_unknown_count": int(recovery_metrics.get("outcome_unknown") or 0),
-                recovery_metrics.get("outcome_unknown") or 0
-            ),
             "active_operation_count": int(recovery_metrics.get("active") or 0),
         },
         "readiness": readiness,
@@ -408,8 +407,7 @@ def _recovery_metrics(operations: list[dict[str, Any]]) -> dict[str, int]:
         value in {"recovery_required", "recovering"} for value in statuses
     )
     active = sum(
-        value in {"planned", "prepared", "running", "recovering"}
+        value in {"planned", "prepared", "running", "recovering"} for value in statuses
-        for value in statuses
     )
     return {
         "failed": failed,
@@ -906,9 +904,7 @@ def _local_storage_capacity(root: Path) -> dict[str, Any]:
         usage = shutil.disk_usage(root)
     except OSError:
         return {"backend": "local", "capacity_observable": False}
-    used_percent = (
+    used_percent = round((usage.used / usage.total) * 100, 1) if usage.total else 0.0
-        round((usage.used / usage.total) * 100, 1) if usage.total else 0.0
-    )
     return {
         "backend": "local",
         "capacity_observable": True,
@@ -987,6 +983,150 @@ def _cached_module_check(
 
 
 def _backup_restore_check() -> dict[str, Any]:
+    projected = _deployer_backup_restore_check()
+    if projected is not None:
+        return projected
+
+    return _legacy_backup_restore_check()
+
+
+def _deployer_backup_restore_check() -> dict[str, Any] | None:
+    state = str(os.environ.get("GOVOPLAN_BACKUP_EVIDENCE_STATE") or "").strip()
+    if not state:
+        return None
+    if state == "absent":
+        return _check(
+            "backup_restore_evidence",
+            "Backup and restore evidence",
+            "warning",
+            "No signed coordinated backup and isolated-restore evidence is adopted.",
+            metrics={"evidence_state": "absent", "restore_drill_ok": False},
+        )
+    if state != "verified":
+        return _check(
+            "backup_restore_evidence",
+            "Backup and restore evidence",
+            "warning",
+            "Deployment backup evidence is incomplete, stale, or failed verification.",
+            metrics={"evidence_state": "invalid", "restore_drill_ok": False},
+        )
+
+    values = {
+        "evidence_id": os.environ.get("GOVOPLAN_BACKUP_EVIDENCE_ID", ""),
+        "recovery_point_id": os.environ.get("GOVOPLAN_BACKUP_RECOVERY_POINT_ID", ""),
+        "restore_drill_id": os.environ.get("GOVOPLAN_BACKUP_RESTORE_DRILL_ID", ""),
+        "evidence_sha256": os.environ.get("GOVOPLAN_BACKUP_EVIDENCE_SHA256", ""),
+        "release_manifest_sha256": os.environ.get(
+            "GOVOPLAN_BACKUP_RELEASE_MANIFEST_SHA256", ""
+        ),
+        "captured_at": os.environ.get("GOVOPLAN_BACKUP_CAPTURED_AT", ""),
+        "expires_at": os.environ.get("GOVOPLAN_BACKUP_EXPIRES_AT", ""),
+        "restore_started_at": os.environ.get("GOVOPLAN_BACKUP_RESTORE_STARTED_AT", ""),
+        "restore_completed_at": os.environ.get(
+            "GOVOPLAN_BACKUP_RESTORE_COMPLETED_AT", ""
+        ),
+        "verified_at": os.environ.get("GOVOPLAN_BACKUP_VERIFIED_AT", ""),
+        "measured_rpo_seconds": os.environ.get(
+            "GOVOPLAN_BACKUP_MEASURED_RPO_SECONDS", ""
+        ),
+        "measured_rto_seconds": os.environ.get(
+            "GOVOPLAN_BACKUP_MEASURED_RTO_SECONDS", ""
+        ),
+        "component_count": os.environ.get("GOVOPLAN_BACKUP_COMPONENT_COUNT", ""),
+    }
+    try:
+        expires_at = _ops_timestamp(values["expires_at"])
+        captured_at = _ops_timestamp(values["captured_at"])
+        restore_started_at = _ops_timestamp(values["restore_started_at"])
+        restore_completed_at = _ops_timestamp(values["restore_completed_at"])
+        verified_at = _ops_timestamp(values["verified_at"])
+        measured_rpo = _ops_nonnegative_integer(values["measured_rpo_seconds"])
+        measured_rto = _ops_nonnegative_integer(values["measured_rto_seconds"])
+        component_count = _ops_nonnegative_integer(values["component_count"])
+        if component_count < 4:
+            raise ValueError("partial component set")
+        if any(
+            len(values[field]) != 64
+            or any(character not in "0123456789abcdef" for character in values[field])
+            for field in ("evidence_sha256", "release_manifest_sha256")
+        ):
+            raise ValueError("invalid digest")
+        if not all(
+            values[field]
+            for field in ("evidence_id", "recovery_point_id", "restore_drill_id")
+        ):
+            raise ValueError("missing identity")
+        if not captured_at <= restore_started_at <= restore_completed_at <= verified_at:
+            raise ValueError("invalid chronology")
+        if (
+            abs(
+                (restore_completed_at - restore_started_at).total_seconds()
+                - measured_rto
+            )
+            > 5
+        ):
+            raise ValueError("RTO does not match chronology")
+    except ValueError:
+        return _check(
+            "backup_restore_evidence",
+            "Backup and restore evidence",
+            "warning",
+            "The sanitized deployment backup receipt is malformed.",
+            metrics={"evidence_state": "invalid", "restore_drill_ok": False},
+        )
+
+    expired = expires_at <= datetime.now(UTC)
+    return _check(
+        "backup_restore_evidence",
+        "Backup and restore evidence",
+        "warning" if expired else "ok",
+        (
+            "Signed backup and restore evidence has expired; adopt a fresh recovery point before migration."
+            if expired
+            else (
+                f"Recovery point {values['recovery_point_id']} and isolated restore "
+                f"drill {values['restore_drill_id']} are deployment-verified."
+            )
+        ),
+        metrics={
+            "evidence_state": "expired" if expired else "verified",
+            "evidence_id": values["evidence_id"],
+            "recovery_point_id": values["recovery_point_id"],
+            "restore_drill_id": values["restore_drill_id"],
+            "captured_at": values["captured_at"],
+            "expires_at": values["expires_at"],
+            "restore_started_at": values["restore_started_at"],
+            "restore_completed_at": values["restore_completed_at"],
+            "measured_rpo_seconds": measured_rpo,
+            "measured_rto_seconds": measured_rto,
+            "component_count": component_count,
+            "restore_drill_ok": not expired,
+        },
+    )
+
+
+def _ops_timestamp(value: str) -> datetime:
+    if not value or len(value) > 64:
+        raise ValueError("invalid timestamp")
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ValueError("invalid timestamp") from exc
+    if parsed.tzinfo is None:
+        raise ValueError("timestamp has no timezone")
+    return parsed.astimezone(UTC)
+
+
+def _ops_nonnegative_integer(value: str) -> int:
+    if not value.isascii() or not value.isdigit() or len(value) > 16:
+        raise ValueError("invalid integer")
+    parsed = int(value)
+    if parsed > 2**63 - 1:
+        raise ValueError("integer out of bounds")
+    return parsed
+
+
+def _legacy_backup_restore_check() -> dict[str, Any]:
     runtime_dir = default_installer_runtime_dir(core_settings.database_url)
     runs = list_module_installer_runs(runtime_dir=runtime_dir, limit=25)
     latest_evidence: tuple[dict[str, object], Mapping[str, object]] | None = None

src/govoplan_ops/backend/manifest.py

@@ -1,8 +1,25 @@
 from __future__ import annotations
 
-from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
+from govoplan_core.core.access import (
-from govoplan_core.core.modules import DocumentationCondition, DocumentationTopic, FrontendModule, FrontendRoute, ModuleContext, ModuleManifest, NavItem, PermissionDefinition, RoleTemplate
+    CAPABILITY_AUTH_PERMISSION_EVALUATOR,
-from govoplan_core.core.provider_governance import ModuleArchitectureDeclaration, ModuleArchitectureDocumentation, ModuleMaturityEvidence
+    CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+)
+from govoplan_core.core.modules import (
+    DocumentationCondition,
+    DocumentationTopic,
+    FrontendModule,
+    FrontendRoute,
+    ModuleContext,
+    ModuleManifest,
+    NavItem,
+    PermissionDefinition,
+    RoleTemplate,
+)
+from govoplan_core.core.provider_governance import (
+    ModuleArchitectureDeclaration,
+    ModuleArchitectureDocumentation,
+    ModuleMaturityEvidence,
+)
 from govoplan_core.core.views import ViewSurface
 
 OPS_READ_SCOPE = "ops:operations:read"
@@ -25,6 +42,11 @@ ARCHITECTURE = ModuleArchitectureDeclaration(
             reference="docs/SCALABILITY_PROFILES.md",
             summary="Documents operational topology and scaling posture.",
         ),
+        ModuleMaturityEvidence(
+            kind="documentation",
+            reference="docs/BACKUP_EVIDENCE_STATUS.md",
+            summary="Documents the sanitized signed backup and restore status projection.",
+        ),
     ),
     known_limits=(
         "Provider health observations depend on module-owned operational probes and may be unavailable until configured.",
@@ -32,8 +54,14 @@ ARCHITECTURE = ModuleArchitectureDeclaration(
     owned_concepts=("operations status projection", "bounded operational probes"),
     non_owned_concepts=("domain repair", "external provider credentials"),
     documentation=ModuleArchitectureDocumentation(
-        recovery=("docs/SCALABILITY_PROFILES.md",),
+        recovery=(
-        operations=("docs/SCALABILITY_PROFILES.md",),
+            "docs/SCALABILITY_PROFILES.md",
+            "docs/BACKUP_EVIDENCE_STATUS.md",
+        ),
+        operations=(
+            "docs/SCALABILITY_PROFILES.md",
+            "docs/BACKUP_EVIDENCE_STATUS.md",
+        ),
     ),
 )
 
@@ -63,7 +91,10 @@ manifest = ModuleManifest(
     id="ops",
     name="Ops",
     version="0.1.8",
-    required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
+    required_capabilities=(
+        CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+        CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    ),
     optional_dependencies=("audit", "docs", "notifications"),
     permissions=(
         _permission(
@@ -108,7 +139,7 @@ manifest = ModuleManifest(
             id="ops.runtime-coordination-and-recovery",
             title="Drain runtime nodes and inspect recovery evidence",
             summary="Ops projects shared runtime heartbeats, replica gaps, drain controls, and recovery states that require operator attention.",
-            body="Use the runtime table to identify stale or composition-skewed API and worker replicas. Drain before replacement so API readiness closes and workers stop taking new queue work; cancellation is available while the node is still draining. The recovery table reports durable Core recovery operations, but a recorded forward-recovery or manual-intervention state is not an automatic database restore.",
+            body="Use the runtime table to identify stale or composition-skewed API and worker replicas. Drain before replacement so API readiness closes and workers stop taking new queue work; cancellation is available while the node is still draining. The recovery table reports durable Core recovery operations. Backup status separately projects only the sanitized deployment verification receipt: a verified status identifies a coordinated recovery point and isolated restore drill, while absent, expired, or invalid evidence blocks a release-changing migration.",
             documentation_types=("admin", "user"),
             audience=("operator", "system_admin"),
             conditions=(
@@ -126,24 +157,55 @@ manifest = ModuleManifest(
                     "Compare active non-stale nodes with the configured API and worker replica expectations.",
                     "Request drain and wait for the node to report draining before replacing it.",
                     "Inspect every recovery-required, outcome-unknown, or manual-intervention operation and follow its recorded recovery mode.",
+                    "Confirm that backup evidence is verified and current before authorizing a release-changing migration.",
                     "Verify replacement composition, readiness, queue consumers, and recovery evidence before closing the operation.",
                 ],
                 "limitations": [
                     "Drain is observed on the runtime heartbeat interval and does not forcibly terminate active work.",
-                    "Ops does not create database backups or make an unsafe post-migration rollback reversible.",
+                    "Ops does not create or restore backups and never receives private artifact or key-custody references.",
+                    "A verified receipt proves the recorded drill; it does not make an unsafe post-migration code rollback reversible.",
                 ],
             },
         ),
     ),
     route_factory=_route_factory,
-    nav_items=(NavItem(path="/ops", label="Ops", icon="activity", required_any=OPS_READ_SCOPES, order=890),),
+    nav_items=(
+        NavItem(
+            path="/ops",
+            label="Ops",
+            icon="activity",
+            required_any=OPS_READ_SCOPES,
+            order=890,
+        ),
+    ),
     frontend=FrontendModule(
         module_id="ops",
         package_name="@govoplan/ops-webui",
-        routes=(FrontendRoute(path="/ops", component="OpsPage", required_any=OPS_READ_SCOPES, order=890),),
+        routes=(
-        nav_items=(NavItem(path="/ops", label="Ops", icon="activity", required_any=OPS_READ_SCOPES, order=890),),
+            FrontendRoute(
+                path="/ops",
+                component="OpsPage",
+                required_any=OPS_READ_SCOPES,
+                order=890,
+            ),
+        ),
+        nav_items=(
+            NavItem(
+                path="/ops",
+                label="Ops",
+                icon="activity",
+                required_any=OPS_READ_SCOPES,
+                order=890,
+            ),
+        ),
         view_surfaces=(
-            ViewSurface(id="ops.widget.health", module_id="ops", kind="section", label="Operations health widget", order=100),
+            ViewSurface(
+                id="ops.widget.health",
+                module_id="ops",
+                kind="section",
+                label="Operations health widget",
+                order=100,
+            ),
         ),
     ),
     architecture=ARCHITECTURE,

tests/test_operational_checks.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from govoplan_core.core.operations import (
@@ -156,3 +157,57 @@ def test_local_storage_capacity_reports_bounded_filesystem_metrics(
     assert metrics["capacity_observable"] is True
     assert metrics["capacity_total_bytes"] > 0
     assert 0 <= metrics["capacity_used_percent"] <= 100
+
+
+def test_deployer_backup_receipt_reports_verified_and_expired_state(
+    monkeypatch,
+) -> None:
+    now = datetime.now(UTC)
+    values = {
+        "GOVOPLAN_BACKUP_EVIDENCE_STATE": "verified",
+        "GOVOPLAN_BACKUP_EVIDENCE_ID": "evidence-1",
+        "GOVOPLAN_BACKUP_RECOVERY_POINT_ID": "recovery-1",
+        "GOVOPLAN_BACKUP_RESTORE_DRILL_ID": "drill-1",
+        "GOVOPLAN_BACKUP_EVIDENCE_SHA256": "a" * 64,
+        "GOVOPLAN_BACKUP_RELEASE_MANIFEST_SHA256": "b" * 64,
+        "GOVOPLAN_BACKUP_CAPTURED_AT": (now - timedelta(hours=2)).isoformat(),
+        "GOVOPLAN_BACKUP_EXPIRES_AT": (now + timedelta(hours=2)).isoformat(),
+        "GOVOPLAN_BACKUP_RESTORE_STARTED_AT": (
+            now - timedelta(hours=1, minutes=5)
+        ).isoformat(),
+        "GOVOPLAN_BACKUP_RESTORE_COMPLETED_AT": (now - timedelta(hours=1)).isoformat(),
+        "GOVOPLAN_BACKUP_VERIFIED_AT": (now - timedelta(minutes=30)).isoformat(),
+        "GOVOPLAN_BACKUP_MEASURED_RPO_SECONDS": "120",
+        "GOVOPLAN_BACKUP_MEASURED_RTO_SECONDS": "300",
+        "GOVOPLAN_BACKUP_COMPONENT_COUNT": "4",
+    }
+    for key, value in values.items():
+        monkeypatch.setenv(key, value)
+
+    verified = routes._backup_restore_check()
+    assert verified["state"] == "ok"
+    assert verified["metrics"]["recovery_point_id"] == "recovery-1"
+    assert verified["metrics"]["measured_rto_seconds"] == 300
+
+    monkeypatch.setenv(
+        "GOVOPLAN_BACKUP_EXPIRES_AT",
+        (now - timedelta(minutes=1)).isoformat(),
+    )
+    expired = routes._backup_restore_check()
+    assert expired["state"] == "warning"
+    assert expired["metrics"]["evidence_state"] == "expired"
+
+
+def test_deployer_backup_receipt_fails_closed_without_exposing_private_evidence(
+    monkeypatch,
+) -> None:
+    monkeypatch.setenv("GOVOPLAN_BACKUP_EVIDENCE_STATE", "invalid")
+    invalid = routes._backup_restore_check()
+
+    assert invalid["state"] == "warning"
+    assert invalid["metrics"] == {
+        "evidence_state": "invalid",
+        "restore_drill_ok": False,
+    }
+    assert "artifact" not in invalid["detail"].lower()
+    assert "key" not in invalid["detail"].lower()