from __future__ import annotations import os 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 from fastapi import APIRouter, Depends, HTTPException, Request, status from pydantic import BaseModel, Field from sqlalchemy import text from sqlalchemy.exc import SQLAlchemyError from govoplan_core.audit.logging import audit_event from govoplan_core.auth import ApiPrincipal, require_any_scope from govoplan_core.core.maintenance import saved_maintenance_mode from govoplan_core.core.module_installer import ( default_installer_runtime_dir, list_module_installer_runs, read_module_installer_run, ) from govoplan_core.core.operations import ( OperationalCheckProviderRegistration, RuntimeWorkStatusContext, RuntimeWorkStatusProviderRegistration, ) from govoplan_core.core.recovery import ( RecoveryOperation, ) from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.provider_governance import ( ExternalProviderStateContext, collect_external_provider_states, ) from govoplan_core.core.runtime_coordination import ( RuntimeCoordinationError, cancel_runtime_node_drain, list_runtime_nodes, request_runtime_node_drain, runtime_composition_hash, ) from govoplan_core.db.session import get_database from govoplan_core.settings import settings as core_settings from govoplan_ops.backend.manifest import OPS_READ_SCOPES, OPS_RUN_SCOPES from govoplan_ops.backend.infrastructure import ( deployment_capability_status, infrastructure_dependency_inventory, ) router = APIRouter(prefix="/ops", tags=["ops"]) _module_check_cache: dict[str, tuple[float, dict[str, Any]]] = {} _module_check_cache_lock = threading.Lock() _runtime_work_cache: dict[str, tuple[float, dict[str, Any]]] = {} _runtime_work_cache_lock = threading.Lock() class RuntimeDrainRequest(BaseModel): reason: str = Field(default="operator request", min_length=1, max_length=500) @router.get("/status") def ops_status( request: Request, principal: ApiPrincipal = Depends(require_any_scope(*OPS_READ_SCOPES)), ) -> dict[str, Any]: del principal return _ops_status_payload(request) @router.get("/readiness") def ops_readiness( request: Request, principal: ApiPrincipal = Depends(require_any_scope(*OPS_READ_SCOPES)), ) -> dict[str, Any]: del principal payload = _ops_status_payload(request) readiness = payload["readiness"] if not readiness["ready"]: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=readiness ) return readiness @router.post("/checks/run") def run_ops_checks( request: Request, principal: ApiPrincipal = Depends(require_any_scope(*OPS_RUN_SCOPES)), ) -> dict[str, Any]: del principal return _ops_status_payload(request, force_module_checks=True) @router.get("/infrastructure/dependencies") def infrastructure_dependencies( request: Request, principal: ApiPrincipal = Depends(require_any_scope(*OPS_READ_SCOPES)), ) -> dict[str, object]: """Return a fresh, non-secret inventory for host deployment preflight.""" del principal return infrastructure_dependency_inventory( _registry(request), installation_id=core_settings.installation_id, ).to_dict() @router.get("/runtime/nodes") def runtime_nodes( request: Request, principal: ApiPrincipal = Depends(require_any_scope(*OPS_READ_SCOPES)), ) -> dict[str, Any]: del principal registry = _registry(request) return _runtime_cluster_status( expected_composition_hash=runtime_composition_hash( tuple(manifest.id for manifest in registry.manifests()) ) ) @router.post("/runtime/nodes/{node_id}/drain") def drain_runtime_node( node_id: str, body: RuntimeDrainRequest, principal: ApiPrincipal = Depends(require_any_scope(*OPS_RUN_SCOPES)), ) -> dict[str, Any]: try: with get_database().SessionLocal() as session: node = request_runtime_node_drain( session, installation_id=core_settings.installation_id, node_id=node_id, reason=body.reason, ) audit_event( session, tenant_id=None, scope="system", user_id=getattr(principal.user, "id", None), api_key_id=principal.api_key_id, action="ops.runtime_node.drain_requested", object_type="runtime_node", object_id=node.node_id, details={ "installation_id": core_settings.installation_id, "reason": node.drain_reason, "incarnation": node.incarnation, }, ) session.commit() return { "node_id": node.node_id, "state": node.state, "drain_reason": node.drain_reason, } except RuntimeCoordinationError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc @router.delete("/runtime/nodes/{node_id}/drain") def cancel_runtime_drain( node_id: str, principal: ApiPrincipal = Depends(require_any_scope(*OPS_RUN_SCOPES)), ) -> dict[str, Any]: try: with get_database().SessionLocal() as session: node = cancel_runtime_node_drain( session, installation_id=core_settings.installation_id, node_id=node_id, ) audit_event( session, tenant_id=None, scope="system", user_id=getattr(principal.user, "id", None), api_key_id=principal.api_key_id, action="ops.runtime_node.drain_cancelled", object_type="runtime_node", object_id=node.node_id, details={ "installation_id": core_settings.installation_id, "incarnation": node.incarnation, }, ) session.commit() return {"node_id": node.node_id, "state": node.state} except RuntimeCoordinationError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc def _ops_status_payload( request: Request, *, force_module_checks: bool = False, ) -> dict[str, Any]: registry = _registry(request) database = _database_status() maintenance_mode = ( database.get("maintenance_mode") if isinstance(database.get("maintenance_mode"), dict) else {"enabled": False, "message": None} ) current_profile = _current_profile() database_capacity = _database_capacity_check() expected_composition_hash = runtime_composition_hash( tuple(manifest.id for manifest in registry.manifests()) ) runtime_cluster = _runtime_cluster_status( expected_composition_hash=expected_composition_hash ) runtime_work = _runtime_work_statuses( registry, RuntimeWorkStatusContext( profile=current_profile, observed_at=datetime.now(UTC), stale_after_seconds=core_settings.runtime_stale_after_seconds, runtime_nodes=tuple(runtime_cluster.get("nodes", [])), ), force=force_module_checks, ) worker_check = _runtime_work_check(runtime_work, current_profile) redis_check = _runtime_backend_check(runtime_work) module_checks = _module_operational_checks( registry, force=force_module_checks, ) storage_check = _storage_check() backup_check = _backup_restore_check() infrastructure = deployment_capability_status() recovery_metrics = runtime_cluster.get("recovery", {}).get("metrics", {}) checks = [ _check( "module_registry", "Module registry", "ok", f"{len(registry.manifests())} modules enabled.", ), _check( "database", "Database", "ok" if database["ok"] else "error", str(database["detail"]), ), _maintenance_check(maintenance_mode), redis_check, worker_check, database_capacity, _runtime_cluster_check(runtime_cluster), storage_check, backup_check, _deployment_security_check(current_profile), _infrastructure_capability_check(infrastructure), *module_checks, ] readiness = _readiness(checks, maintenance_mode) governance = _governance_inventory( registry, external_provider_states=_external_provider_states(registry), ) return { "summary": { "app_env": core_settings.app_env, "active_profile": current_profile, "module_count": len(registry.manifests()), "celery_enabled": bool(core_settings.celery_enabled), "celery_queues": _celery_queues(), "redis_url": _redact_url(core_settings.redis_url), "maintenance_mode": maintenance_mode, "database_url": _redact_url(core_settings.database_url), "database_connection_peak": core_settings.database_connection_peak, "database_connection_available": ( core_settings.database_connection_available ), "file_storage_backend": core_settings.file_storage_backend, "storage_metrics": storage_check.get("metrics", {}), "backup_state": backup_check.get("state", "unknown"), "worker_metrics": worker_check.get("metrics", {}), "operational_probe_count": len(module_checks), "runtime_node_count": len(runtime_cluster.get("nodes", [])), "recovery_required_count": runtime_cluster.get("recovery", {}).get( "requires_attention", 0 ), "failed_operation_count": int(recovery_metrics.get("failed") or 0), "outcome_unknown_count": int(recovery_metrics.get("outcome_unknown") or 0), "active_operation_count": int(recovery_metrics.get("active") or 0), "infrastructure_capability_count": len( infrastructure.get("capabilities", []) ), "pending_post_install_task_count": len( infrastructure.get("post_install_tasks", []) ), }, "readiness": readiness, "checks": checks, "governance": governance, "deployment_profiles": _deployment_profiles(current_profile), "sizing": _sizing_assumptions(), "runtime_cluster": runtime_cluster, "runtime_work": runtime_work, "infrastructure": infrastructure, } def _infrastructure_capability_check( infrastructure: Mapping[str, object], ) -> dict[str, Any]: if infrastructure.get("available") is True: capabilities = infrastructure.get("capabilities") count = len(capabilities) if isinstance(capabilities, list) else 0 return _check( "infrastructure_capability_receipt", "Infrastructure capabilities", "ok", f"A validated non-secret deployment receipt reports {count} capabilities.", ) if infrastructure.get("configured") is True: return _check( "infrastructure_capability_receipt", "Infrastructure capabilities", "warning", str( infrastructure.get("error") or "The configured deployment capability receipt is unavailable." ), ) return _check( "infrastructure_capability_receipt", "Infrastructure capabilities", "ok", "No deployment capability receipt is mounted in this runtime profile.", ) def _runtime_cluster_status( *, expected_composition_hash: str | None = None, ) -> dict[str, Any]: try: with get_database().SessionLocal() as session: nodes = list_runtime_nodes( session, installation_id=core_settings.installation_id, stale_after_seconds=core_settings.runtime_stale_after_seconds, ) operations = ( session.query(RecoveryOperation) .filter( RecoveryOperation.installation_id == core_settings.installation_id ) .order_by(RecoveryOperation.updated_at.desc()) .limit(50) .all() ) except SQLAlchemyError as exc: return { "available": False, "detail": f"Runtime coordination query failed: {exc}", "state_profile": core_settings.state_profile, "nodes": [], "recovery": { "operations": [], "requires_attention": 0, "metrics": _recovery_metrics([]), }, } active_nodes = [ node for node in nodes if node["state"] == "active" and not node["stale"] ] active_api = sum(node["role"] == "api" for node in active_nodes) active_workers = sum(node["role"] == "worker" for node in active_nodes) runtime_nodes = [node for node in active_nodes if node["role"] in {"api", "worker"}] composition_hashes = sorted( { str(node["composition_hash"]) for node in runtime_nodes if node.get("composition_hash") } ) software_versions = sorted( { str(node["software_version"]) for node in runtime_nodes if node.get("software_version") } ) unexpected_composition_nodes = [ str(node["node_id"]) for node in runtime_nodes if expected_composition_hash and node.get("composition_hash") != expected_composition_hash ] active_worker_nodes = [node for node in active_nodes if node["role"] == "worker"] active_queues = sorted( { str(queue) for node in active_worker_nodes for queue in (node.get("queues") or []) if queue } ) expected_queues = _celery_queues() if core_settings.celery_enabled else [] missing_queues = sorted(set(expected_queues) - set(active_queues)) worker_pools = sorted( { str(node.get("metadata", {}).get("worker_pool") or "default") for node in active_worker_nodes } ) operation_payloads = [ { "id": operation.id, "module_id": operation.module_id, "operation_type": operation.operation_type, "resource_type": operation.resource_type, "resource_id": operation.resource_id, "mode": operation.mode, "status": operation.status, "checkpoint_count": operation.checkpoint_count, "evidence_head_sha256": operation.evidence_head_sha256, "failure_summary": operation.failure_summary, "updated_at": operation.updated_at.isoformat(), } for operation in operations ] recovery_metrics = _recovery_metrics(operation_payloads) return { "available": True, "detail": "Shared runtime directory is available.", "installation_id": core_settings.installation_id, "state_profile": core_settings.state_profile, "expected": { "api": core_settings.runtime_expected_api_replicas, "worker": core_settings.runtime_expected_worker_replicas, }, "active": {"api": active_api, "worker": active_workers}, "composition": { "expected_hash": expected_composition_hash, "active_hashes": composition_hashes, "unexpected_nodes": unexpected_composition_nodes, "skewed": len(composition_hashes) > 1 or bool(unexpected_composition_nodes), }, "software_versions": { "active": software_versions, "skewed": len(software_versions) > 1, }, "queues": { "expected": expected_queues, "active": active_queues, "missing": missing_queues, "worker_pools": worker_pools, }, "nodes": nodes, "recovery": { "operations": operation_payloads, "requires_attention": recovery_metrics["requires_attention"], "metrics": recovery_metrics, }, } def _recovery_metrics(operations: list[dict[str, Any]]) -> dict[str, int]: statuses = [str(operation.get("status") or "") for operation in operations] failed = sum(value in {"failed", "manual_intervention"} for value in statuses) outcome_unknown = statuses.count("outcome_unknown") recovery_required = sum( value in {"recovery_required", "recovering"} for value in statuses ) active = sum( value in {"planned", "prepared", "running", "recovering"} for value in statuses ) return { "failed": failed, "outcome_unknown": outcome_unknown, "recovery_required": recovery_required, "active": active, "requires_attention": failed + outcome_unknown + recovery_required, } def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]: if not cluster.get("available"): return _check( "runtime_cluster", "Runtime cluster", "error", str(cluster.get("detail") or "Runtime directory unavailable."), readiness_critical=True, ) nodes = cluster.get("nodes") or [] stale = [node for node in nodes if node.get("stale")] expected = cluster.get("expected") or {} active = cluster.get("active") or {} missing_api = max(0, int(expected.get("api") or 0) - int(active.get("api") or 0)) missing_workers = max( 0, int(expected.get("worker") or 0) - int(active.get("worker") or 0) ) composition = cluster.get("composition") or {} versions = cluster.get("software_versions") or {} queues = cluster.get("queues") or {} composition_skewed = bool(composition.get("skewed")) version_skewed = bool(versions.get("skewed")) missing_queues = list(queues.get("missing") or []) state = "ok" detail = f"{len(nodes)} runtime node record(s); {len(stale)} stale." readiness_critical = False if missing_api or missing_workers: state = "error" if cluster.get("state_profile") == "shared" else "warning" readiness_critical = state == "error" detail += ( f" Missing expected replicas: api={missing_api}, worker={missing_workers}." ) if composition_skewed or version_skewed or missing_queues: state = "error" if cluster.get("state_profile") == "shared" else "warning" readiness_critical = readiness_critical or state == "error" if composition_skewed: detail += " Active runtime composition differs from the loaded graph." if version_skewed: detail += " Active runtime software versions differ." if missing_queues: detail += ( " Queues without an active worker: " + ", ".join(missing_queues) + "." ) if stale and state == "ok": state = "warning" return _check( "runtime_cluster", "Runtime cluster", state, detail, readiness_critical=readiness_critical, metrics={ "nodes": len(nodes), "stale": len(stale), "active_api": int(active.get("api") or 0), "active_workers": int(active.get("worker") or 0), "composition_skewed": composition_skewed, "version_skewed": version_skewed, "missing_queues": len(missing_queues), "recovery_required": int( cluster.get("recovery", {}).get("requires_attention") or 0 ), }, ) def _database_capacity_check() -> dict[str, Any]: limit = core_settings.database_connection_limit reserve = core_settings.database_connection_reserve peak = core_settings.database_connection_peak available = core_settings.database_connection_available shared = core_settings.state_profile == "shared" if limit is None or peak is None or available is None: return _check( "database_capacity", "Database connection capacity", "error" if shared else "inactive", ( "Shared deployments must declare the PostgreSQL connection " "limit and rendered peak demand." if shared else "No deployment-level PostgreSQL connection budget is configured." ), readiness_critical=shared, ) expected_available = limit - reserve valid = reserve < limit and available == expected_available and peak <= available utilization = peak / available if available else 1.0 if not valid: return _check( "database_capacity", "Database connection capacity", "error", ( f"Rendered peak {peak} does not fit the declared available " f"capacity {available} (limit {limit}, reserve {reserve})." ), readiness_critical=shared, metrics={ "limit": limit, "reserve": reserve, "available": available, "peak": peak, }, ) return _check( "database_capacity", "Database connection capacity", "warning" if utilization >= 0.8 else "ok", ( f"Rendered peak {peak} uses {utilization:.0%} of {available} " f"connections available after the {reserve}-connection reserve." ), metrics={ "limit": limit, "reserve": reserve, "available": available, "peak": peak, "utilization": round(utilization, 4), }, ) def _registry(request: Request) -> PlatformRegistry: registry = getattr(request.app.state, "govoplan_registry", None) if not isinstance(registry, PlatformRegistry): raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="GovOPlaN module registry is not configured", ) return registry def _governance_inventory( registry: PlatformRegistry, *, external_provider_states: Mapping[str, Mapping[str, object]] | None = None, ) -> dict[str, Any]: modules: list[dict[str, Any]] = [] for manifest in registry.manifests(): capability_names = tuple(sorted(manifest.capability_factories)) architecture = ( manifest.architecture.to_dict() if manifest.architecture else None ) external_providers = tuple( { **declaration.to_dict(), "runtime_state": dict( (external_provider_states or {}).get(declaration.id, {}) ) or None, } for declaration in manifest.external_providers ) access_control_count = ( len(manifest.resource_acl_providers) + len(manifest.ownership_providers) + sum( len(providers) for providers in manifest.delete_veto_providers.values() ) + len(manifest.uninstall_guard_providers) ) modules.append( { "module_id": manifest.id, "name": manifest.name, "version": manifest.version, "permission_count": len(manifest.permissions), "role_template_count": len(manifest.role_templates), "capability_count": len(capability_names), "policy_count": sum( name.startswith("policy.") for name in capability_names ), "documentation_count": len(manifest.documentation), "documentation_provider_count": len(manifest.documentation_providers), "access_control_count": access_control_count, "search_provider_count": len(manifest.search_providers) + len(manifest.search_sources), "migration_managed": manifest.migration_spec is not None, "architecture": architecture, "external_provider_count": len(external_providers), "external_providers": external_providers, } ) return { "summary": { "module_count": len(modules), "permission_count": sum(item["permission_count"] for item in modules), "role_template_count": sum(item["role_template_count"] for item in modules), "capability_count": sum(item["capability_count"] for item in modules), "policy_count": sum(item["policy_count"] for item in modules), "documented_module_count": sum( bool( item["documentation_count"] or item["documentation_provider_count"] ) for item in modules ), "access_control_count": sum( item["access_control_count"] for item in modules ), "search_provider_count": sum( item["search_provider_count"] for item in modules ), "migration_module_count": sum( bool(item["migration_managed"]) for item in modules ), "architecture_declared_module_count": sum( bool(item["architecture"]) for item in modules ), "external_provider_count": sum( item["external_provider_count"] for item in modules ), "configured_external_provider_count": sum( bool( provider.get("runtime_state", {}).get("configured") if isinstance(provider.get("runtime_state"), Mapping) else False ) for item in modules for provider in item["external_providers"] ), "provider_attention_count": sum( ( str(provider.get("runtime_state", {}).get("health")) in {"warning", "error", "unknown"} or str(provider.get("runtime_state", {}).get("recovery")) == "attention" ) if isinstance(provider.get("runtime_state"), Mapping) else False for item in modules for provider in item["external_providers"] ), "supported_module_count": sum( bool(item["architecture"]) and item["architecture"]["maturity"] in {"supported", "lts"} for item in modules ), }, "modules": modules, } def _external_provider_states( registry: PlatformRegistry, ) -> dict[str, dict[str, object]]: registrations = registry.external_provider_state_providers() if not registrations: return {} try: with get_database().session() as session: return collect_external_provider_states( registrations, ExternalProviderStateContext(session=session), ) except Exception: # noqa: BLE001 - preserve Ops inventory during DB outages. return collect_external_provider_states( registrations, ExternalProviderStateContext(session=None), ) def _database_status() -> dict[str, Any]: try: with get_database().session() as session: session.execute(text("select 1")) maintenance = saved_maintenance_mode(session).as_dict() return { "ok": True, "detail": "Database session succeeded.", "maintenance_mode": maintenance, } except (RuntimeError, SQLAlchemyError) as exc: return { "ok": False, "detail": f"Database check failed: {exc}", "maintenance_mode": {"enabled": False, "message": None}, } def _maintenance_check(maintenance_mode: dict[str, Any]) -> dict[str, Any]: if maintenance_mode.get("enabled"): message = str(maintenance_mode.get("message") or "Maintenance mode is enabled.") return _check( "maintenance_mode", "Maintenance mode", "warning", message, readiness_critical=True, ) return _check( "maintenance_mode", "Maintenance mode", "ok", "Maintenance mode is off." ) def _runtime_work_statuses( registry: PlatformRegistry, context: RuntimeWorkStatusContext, *, force: bool = False, ) -> list[dict[str, Any]]: registrations: list[RuntimeWorkStatusProviderRegistration] = [] for manifest in registry.manifests(): registrations.extend(manifest.runtime_work_status_providers) results: list[dict[str, Any]] = [] seen: set[str] = set() now = time.monotonic() for registration in registrations: key = f"{registration.module_id}:{registration.provider_id}" if registration.provider_id in seen: results.append(_unavailable_runtime_work(registration, "Duplicate provider id.")) continue seen.add(registration.provider_id) with _runtime_work_cache_lock: cached = _runtime_work_cache.get(key) if ( not force and cached is not None and now - cached[0] < max(1, registration.cache_seconds) ): results.append(dict(cached[1])) continue try: status_result = registration.provider(context) if status_result.provider_id != registration.provider_id: raise ValueError("provider id mismatch") result = status_result.as_dict() except Exception: # noqa: BLE001 - provider failures remain isolated and sanitized. result = _unavailable_runtime_work( registration, "The runtime-work provider failed without usable evidence.", ) with _runtime_work_cache_lock: _runtime_work_cache[key] = (now, result) results.append(dict(result)) return results def _unavailable_runtime_work( registration: RuntimeWorkStatusProviderRegistration, detail: str, ) -> dict[str, Any]: return { "provider_id": registration.provider_id, "label": registration.provider_id, "backend": "unavailable", "enabled": None, "configured": None, "state": "unreachable", "detail": detail, "observed_at": datetime.now(UTC).isoformat(), "active_workers": None, "last_heartbeat_at": None, "queue_depths": {}, "active_work": None, "reserved_work": None, "failures": None, "stale_after_seconds": None, "guidance": "Restore or configure the owning status provider.", } def _runtime_work_check( statuses: list[dict[str, Any]], profile: str, ) -> dict[str, Any]: if not statuses: return _check( "worker_split", "Background workers", "warning", "No runtime-work status provider is installed; worker and queue health is unavailable.", readiness_critical=profile not in {"local-dev", "development"}, metrics={"workers": None, "active_tasks": None, "reserved_tasks": None}, ) development_profile = profile in {"local-dev", "development"} state_rank = { "unreachable": 8, "stale": 7, "degraded": 6, "unconfigured": 5, "starting": 4, "busy": 3, "healthy": 2, "idle": 1, "disabled": 0 if development_profile else 6, } worst = max(statuses, key=lambda item: state_rank.get(str(item.get("state")), 8)) runtime_state = str(worst.get("state") or "unreachable") disabled_in_dev = runtime_state == "disabled" and development_profile check_state = ( "inactive" if disabled_in_dev else "ok" if runtime_state in {"healthy", "idle", "busy"} else "warning" if runtime_state in {"disabled", "starting", "degraded"} else "error" ) queue_depths: dict[str, int | None] = {} for item in statuses: depths = item.get("queue_depths") if isinstance(depths, Mapping): queue_depths.update( { str(queue): int(value) if isinstance(value, int) else None for queue, value in depths.items() } ) return _check( "worker_split", "Background workers", check_state, str(worst.get("detail") or "Runtime-work status is unavailable."), readiness_critical=( not disabled_in_dev and runtime_state in {"disabled", "unconfigured", "degraded", "stale", "unreachable"} ), metrics={ "state": runtime_state, "workers": _sum_known(statuses, "active_workers"), "active_tasks": _sum_known(statuses, "active_work"), "reserved_tasks": _sum_known(statuses, "reserved_work"), "failures": _sum_known(statuses, "failures"), "queue_depths": queue_depths, }, ) def _runtime_backend_check(statuses: list[dict[str, Any]]) -> dict[str, Any]: if not statuses or all(item.get("enabled") is False for item in statuses): return _check( "redis_broker", "Work backend", "inactive", "No enabled runtime-work backend requires a connectivity assertion.", ) if any(item.get("state") == "unreachable" for item in statuses): return _check( "redis_broker", "Work backend", "error", "An enabled runtime-work backend is unreachable.", readiness_critical=True, ) if any(item.get("configured") is not True for item in statuses if item.get("enabled") is True): return _check( "redis_broker", "Work backend", "warning", "An enabled runtime-work backend is not fully configured.", readiness_critical=True, ) return _check( "redis_broker", "Work backend", "ok", "Enabled runtime-work providers returned bounded status evidence.", ) def _sum_known(statuses: list[dict[str, Any]], field: str) -> int | None: values = [item.get(field) for item in statuses] known = [int(value) for value in values if isinstance(value, int)] return sum(known) if known else None def _storage_check() -> dict[str, Any]: backend = str(core_settings.file_storage_backend or "local").lower() if backend == "s3": configured = bool( core_settings.file_storage_s3_endpoint_url and core_settings.file_storage_s3_bucket ) return _check( "file_storage", "File storage", "ok" if configured else "warning", "S3 file storage is configured." if configured else "S3 file storage needs endpoint and bucket settings.", readiness_critical=not configured, metrics={ "backend": "s3", "capacity_observable": False, }, ) root = Path(str(core_settings.file_storage_local_root or "runtime/files")) if root.exists() and root.is_dir(): writable = os.access(root, os.W_OK) state = "ok" if writable else "warning" detail = ( f"Local file storage root: {root}" if writable else f"Local file storage root is not writable: {root}" ) metrics = _local_storage_capacity(root) return _check( "file_storage", "File storage", state, detail, readiness_critical=not writable, metrics=metrics, ) return _check( "file_storage", "File storage", "warning", f"Local file storage root does not exist yet: {root}", readiness_critical=False, metrics={"backend": "local", "capacity_observable": False}, ) def _local_storage_capacity(root: Path) -> dict[str, Any]: try: usage = shutil.disk_usage(root) except OSError: return {"backend": "local", "capacity_observable": False} used_percent = round((usage.used / usage.total) * 100, 1) if usage.total else 0.0 return { "backend": "local", "capacity_observable": True, "capacity_total_bytes": int(usage.total), "capacity_used_bytes": int(usage.used), "capacity_free_bytes": int(usage.free), "capacity_used_percent": used_percent, } def _module_operational_checks( registry: PlatformRegistry, *, force: bool, ) -> list[dict[str, Any]]: registrations: list[OperationalCheckProviderRegistration] = [] for manifest in registry.manifests(): registrations.extend(manifest.operational_check_providers) results: list[dict[str, Any]] = [] seen: set[str] = set() for registration in registrations: cache_key = f"{registration.module_id}:{registration.check_id}" if registration.check_id in seen: results.append( _check( f"ops.duplicate.{registration.check_id}", "Operational check registry", "error", f"Operational check id {registration.check_id!r} is registered more than once.", readiness_critical=True, ) ) continue seen.add(registration.check_id) cached = _cached_module_check( cache_key, max_age_seconds=max(0, registration.cache_seconds), ) if not force and cached is not None: results.append(cached) continue try: check = registration.provider() if check.id != registration.check_id: raise ValueError( f"provider returned {check.id!r}, expected {registration.check_id!r}" ) result = check.as_dict() except Exception as exc: # noqa: BLE001 - one optional module must not hide all Ops status. result = _check( registration.check_id, f"{registration.module_id} operational check", "error", f"The module-owned check failed unexpectedly ({type(exc).__name__}).", readiness_critical=True, ) with _module_check_cache_lock: _module_check_cache[cache_key] = (time.monotonic(), result) results.append(dict(result)) return results def _cached_module_check( cache_key: str, *, max_age_seconds: int, ) -> dict[str, Any] | None: if max_age_seconds <= 0: return None with _module_check_cache_lock: cached = _module_check_cache.get(cache_key) if cached is None or time.monotonic() - cached[0] >= max_age_seconds: return None return dict(cached[1]) 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 for summary in runs: try: record = read_module_installer_run( runtime_dir=runtime_dir, run_id=str(summary["run_id"]), ) except Exception: # noqa: BLE001 - malformed historical evidence is skipped. continue snapshot = record.get("snapshot") backup = ( snapshot.get("database_backup") if isinstance(snapshot, Mapping) else None ) if isinstance(backup, Mapping): latest_evidence = (record, backup) break drill = _restore_drill_evidence(runtime_dir) if latest_evidence is None: return _check( "backup_restore_evidence", "Backup and restore evidence", "warning", "No database backup evidence is present in installer history. External deployment backups may exist, but are not evidenced here.", metrics={"restore_drill_ok": bool(drill and drill.get("ok") is True)}, ) record, backup = latest_evidence restore_check = backup.get("restore_check") restore_check_ok = isinstance(restore_check, Mapping) and ( restore_check.get("return_code") == 0 or str(restore_check.get("result") or "").lower() == "ok" ) drill_ok = bool(drill and drill.get("ok") is True) state = "ok" if restore_check_ok and drill_ok else "warning" missing: list[str] = [] if not restore_check_ok: missing.append("backup restore-readiness check") if not drill_ok: missing.append("recorded rollback drill") detail = f"Installer run {record.get('run_id')} records a {backup.get('type', 'database')} backup." if missing: detail += " Missing evidence: " + ", ".join(missing) + "." else: detail += " The backup check and rollback drill both passed." return _check( "backup_restore_evidence", "Backup and restore evidence", state, detail, metrics={ "run_id": record.get("run_id"), "backup_type": backup.get("type"), "restore_check_ok": restore_check_ok, "restore_drill_ok": drill_ok, "drill_completed_at": drill.get("completed_at") if drill else None, }, ) def _restore_drill_evidence(runtime_dir: Path) -> dict[str, object] | None: configured = os.environ.get("GOVOPLAN_RESTORE_DRILL_EVIDENCE_PATH") path = ( Path(configured).expanduser() if configured else runtime_dir / "restore-drill-evidence.json" ) try: import json payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None return payload if isinstance(payload, dict) else None def _deployment_security_check(current_profile: str) -> dict[str, Any]: app_env = str(core_settings.app_env or "").lower() if app_env in {"dev", "test", "local"}: return _check( "deployment_security", "HTTP and certificates", "inactive", "Local/test profile; HTTPS and secure-cookie checks are deployment-owned.", ) blockers: list[str] = [] if not core_settings.auth_cookie_secure: blockers.append("AUTH_COOKIE_SECURE=false") cors_origins = _cors_origins() if not cors_origins: blockers.append("CORS_ORIGINS is empty") elif any( origin == "*" or origin.startswith("http://localhost") or origin.startswith("http://127.0.0.1") for origin in cors_origins ): blockers.append("CORS_ORIGINS contains local or wildcard origins") if not blockers: return _check( "deployment_security", "HTTP and certificates", "ok", f"{current_profile} uses secure cookies and explicit non-local CORS origins.", ) profile_label = ( "production" if app_env in {"prod", "production"} else app_env or current_profile ) detail = f"{profile_label} profile needs deployment HTTP/certificate hardening: {', '.join(blockers)}." return _check( "deployment_security", "HTTP and certificates", "warning", detail, readiness_critical=app_env in {"prod", "production"}, ) def _check( check_id: str, label: str, state: str, detail: str, *, readiness_critical: bool = False, metrics: Mapping[str, object] | None = None, ) -> dict[str, Any]: return { "id": check_id, "label": label, "state": state, "detail": detail, "readiness_critical": readiness_critical, "metrics": dict(metrics or {}), } def _readiness( checks: list[dict[str, Any]], maintenance_mode: dict[str, Any] ) -> dict[str, Any]: blockers = [ { "id": check["id"], "label": check["label"], "state": check["state"], "detail": check["detail"], } for check in checks if check.get("readiness_critical") and check.get("state") != "ok" ] if maintenance_mode.get("enabled") and not any( item["id"] == "maintenance_mode" for item in blockers ): blockers.append( { "id": "maintenance_mode", "label": "Maintenance mode", "state": "warning", "detail": str( maintenance_mode.get("message") or "Maintenance mode is enabled." ), } ) return { "ready": not blockers, "blockers": blockers, "profile": _current_profile(), } def _current_profile() -> str: app_env = str(core_settings.app_env or "").lower() if app_env in {"dev", "test"} and core_settings.celery_enabled: return "production-like-dev" if app_env in {"dev", "test"}: return "local-dev" if core_settings.celery_enabled: return "split-worker" return "single-process" def _celery_queues() -> list[str]: return [ item.strip() for item in str(core_settings.celery_queues or "").split(",") if item.strip() ] def _cors_origins() -> list[str]: return [ item.strip() for item in str(core_settings.cors_origins or "").split(",") if item.strip() ] def _deployment_profiles(current_profile: str) -> list[dict[str, Any]]: profiles = [ { "id": "local-dev", "name": "Local development", "current": current_profile == "local-dev", "components": [ "FastAPI", "SQLite or local database", "Vite dev server", "local file storage", ], "fit": "Developer workstation, demos, and module integration work.", }, { "id": "production-like-dev", "name": "Production-like development", "current": current_profile == "production-like-dev", "components": [ "FastAPI", "Vite dev server", "PostgreSQL", "Redis broker", "Celery worker", "durable local file storage", ], "fit": "Local validation of the production dependency shape without publishing packages.", }, { "id": "single-process", "name": "Small institution", "current": current_profile == "single-process", "components": [ "FastAPI", "PostgreSQL", "reverse proxy", "scheduled backups", "local or S3-compatible file storage", ], "fit": "Low-volume internal administration without heavy background jobs.", }, { "id": "split-worker", "name": "Institution platform", "current": current_profile == "split-worker", "components": [ "FastAPI", "worker process", "Redis broker", "PostgreSQL", "object storage", "health probes", ], "fit": "Campaigns, imports, exports, scheduled work, and larger tenant counts.", }, ] return profiles def _sizing_assumptions() -> list[dict[str, str]]: return [ { "area": "API", "baseline": "One process behind a reverse proxy.", "scale_trigger": "Sustained request latency or concurrent admin/portal usage.", "operator_note": "Scale horizontally only after session, CSRF, and proxy headers are stable.", }, { "area": "Workers", "baseline": "Disabled in local development; one worker for production imports, mail, and package jobs.", "scale_trigger": "Backlogs in send_email, append_sent, imports, or export queues.", "operator_note": "Keep package installer execution separate from ordinary worker queues.", }, { "area": "Database", "baseline": "PostgreSQL for shared-tenancy production.", "scale_trigger": "Tenant growth, audit volume, DMS/file metadata, or reporting workloads.", "operator_note": "Backups and restore checks are part of the deployment profile, not optional documentation.", }, { "area": "Files", "baseline": "Local storage for development; S3-compatible object storage for production.", "scale_trigger": "Cross-node deployments, DMS volume, or file retention requirements.", "operator_note": "Retention policy and legal-hold behavior should be verified before large imports.", }, ] def _redact_url(value: str) -> str: try: parsed = urlsplit(value) except ValueError: return "" if parsed.username or parsed.password: host = parsed.hostname or "" port = f":{parsed.port}" if parsed.port else "" return f"{parsed.scheme}://@{host}{port}{parsed.path}" return value