feat: expose multi-host runtime readiness

README.md

@@ -15,6 +15,8 @@ This repository owns:
 - operator-facing status APIs
 - runtime-node registration, heartbeat, composition, stale-node, and expected
   replica projection
+- queue-specific worker-pool coverage, release/composition skew, and the
+  deployment-rendered PostgreSQL connection budget
 - audited API and worker drain/cancel controls
 - recovery-operation status and evidence-chain summaries
 - governance inventory for module-declared permissions, roles, capabilities,

docs/SCALABILITY_PROFILES.md

@@ -39,6 +39,9 @@ The Ops API reports:
 - runtime node identity, role, software/module composition, queues, heartbeat,
   stale state, and drain state
 - configured versus active API and worker replica counts
+- active worker-pool names, exact queue coverage, and missing queue owners
+- release/module-composition skew and software-version skew across active nodes
+- rendered PostgreSQL connection peak, declared server limit, and operator reserve
 - recovery operation status, mode, checkpoint count, and last update
 
 These values are intentionally diagnostic. They do not replace deployment
@@ -54,6 +57,13 @@ manual-intervention record means an operator must repair the current release or
 restore a separately verified coordinated backup. Ops does not convert that
 state into a safe rollback.
 
+For the `shared` profile, missing expected replicas, release/composition skew,
+unserved queues, or an invalid PostgreSQL connection budget are readiness
+errors. Stale historical records remain visible, but cannot downgrade one of
+those errors to a warning. A database budget at or above 80 percent is reported
+as a warning so operators retain room for measured bursts and administrative
+connections.
+
 `deployment_security` is inactive for local/test profiles. In staging or pilot
 profiles it warns when secure cookies or CORS origins still look local. In
 production it becomes readiness-critical because TLS certificates, proxy

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

@@ -35,6 +35,7 @@ from govoplan_core.core.runtime_coordination import (
     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
@@ -70,7 +71,9 @@ def ops_readiness(
     payload = _ops_status_payload(request)
     readiness = payload["readiness"]
     if not readiness["ready"]:
-        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=readiness)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=readiness
+        )
     return readiness
 
 
@@ -85,10 +88,16 @@ def run_ops_checks(
 
 @router.get("/runtime/nodes")
 def runtime_nodes(
+    request: Request,
     principal: ApiPrincipal = Depends(require_any_scope(*OPS_READ_SCOPES)),
 ) -> dict[str, Any]:
     del principal
-    return _runtime_cluster_status()
+    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")
@@ -175,21 +184,42 @@ def _ops_status_payload(
 ) -> 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}
+    maintenance_mode = (
+        database.get("maintenance_mode")
+        if isinstance(database.get("maintenance_mode"), dict)
+        else {"enabled": False, "message": None}
+    )
     redis_check = _redis_check()
     current_profile = _current_profile()
     worker_check = _worker_check()
-    runtime_cluster = _runtime_cluster_status()
+    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
+    )
     module_checks = _module_operational_checks(
         registry,
         force=force_module_checks,
     )
     checks = [
-        _check("module_registry", "Module registry", "ok", f"{len(registry.manifests())} modules enabled."),
+        _check(
-        _check("database", "Database", "ok" if database["ok"] else "error", str(database["detail"])),
+            "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_restore_check(),
@@ -211,11 +241,17 @@ def _ops_status_payload(
             "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,
             "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),
+            "recovery_required_count": runtime_cluster.get("recovery", {}).get(
+                "requires_attention", 0
+            ),
         },
         "readiness": readiness,
         "checks": checks,
@@ -226,7 +262,10 @@ def _ops_status_payload(
     }
 
 
-def _runtime_cluster_status() -> dict[str, Any]:
+def _runtime_cluster_status(
+    *,
+    expected_composition_hash: str | None = None,
+) -> dict[str, Any]:
     try:
         with get_database().SessionLocal() as session:
             nodes = list_runtime_nodes(
@@ -237,8 +276,7 @@ def _runtime_cluster_status() -> dict[str, Any]:
             operations = (
                 session.query(RecoveryOperation)
                 .filter(
-                    RecoveryOperation.installation_id
+                    RecoveryOperation.installation_id == core_settings.installation_id
-                    == core_settings.installation_id
                 )
                 .order_by(RecoveryOperation.updated_at.desc())
                 .limit(50)
@@ -253,12 +291,48 @@ def _runtime_cluster_status() -> dict[str, Any]:
             "recovery": {"operations": [], "requires_attention": 0},
         }
     active_nodes = [
-        node
+        node for node in nodes if node["state"] == "active" and not node["stale"]
-        for node in nodes
-        if node["state"] != "stopped" 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,
@@ -295,6 +369,22 @@ def _runtime_cluster_status() -> dict[str, Any]:
             "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,
@@ -317,15 +407,36 @@ def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]:
     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))
+    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}."
+        detail += (
-    elif stale:
+            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",
@@ -338,6 +449,9 @@ def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]:
             "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
             ),
@@ -345,10 +459,70 @@ def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]:
     )
 
 
+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")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="GovOPlaN module registry is not configured",
+        )
     return registry
 
 
@@ -376,26 +550,33 @@ def _governance_inventory(
         access_control_count = (
             len(manifest.resource_acl_providers)
             + len(manifest.ownership_providers)
-            + sum(len(providers) for providers in manifest.delete_veto_providers.values())
+            + sum(
+                len(providers) for providers in manifest.delete_veto_providers.values()
+            )
             + len(manifest.uninstall_guard_providers)
         )
-        modules.append({
+        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),
+                "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),
+                "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),
@@ -404,12 +585,20 @@ def _governance_inventory(
             "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"])
+                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),
+            "access_control_count": sum(
-            "search_provider_count": sum(item["search_provider_count"] for item in modules),
+                item["access_control_count"] for item in modules
-            "migration_module_count": sum(bool(item["migration_managed"]) 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
             ),
@@ -471,29 +660,63 @@ def _database_status() -> dict[str, Any]:
         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}
+        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}}
+        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(
-    return _check("maintenance_mode", "Maintenance mode", "ok", "Maintenance mode is off.")
+            "maintenance_mode",
+            "Maintenance mode",
+            "warning",
+            message,
+            readiness_critical=True,
+        )
+    return _check(
+        "maintenance_mode", "Maintenance mode", "ok", "Maintenance mode is off."
+    )
 
 
 def _redis_check() -> dict[str, Any]:
     if not core_settings.celery_enabled:
-        return _check("redis_broker", "Redis broker", "inactive", "Redis is not required while Celery is disabled.")
+        return _check(
+            "redis_broker",
+            "Redis broker",
+            "inactive",
+            "Redis is not required while Celery is disabled.",
+        )
     try:
         from redis import Redis
 
-        client = Redis.from_url(core_settings.redis_url, socket_connect_timeout=0.75, socket_timeout=0.75)
+        client = Redis.from_url(
+            core_settings.redis_url, socket_connect_timeout=0.75, socket_timeout=0.75
+        )
         client.ping()
-        return _check("redis_broker", "Redis broker", "ok", f"Redis broker reachable at {_redact_url(core_settings.redis_url)}.")
+        return _check(
+            "redis_broker",
+            "Redis broker",
+            "ok",
+            f"Redis broker reachable at {_redact_url(core_settings.redis_url)}.",
+        )
     except Exception as exc:  # noqa: BLE001 - diagnostic endpoint should report the concrete failure.
-        return _check("redis_broker", "Redis broker", "error", f"Redis broker check failed: {exc}", readiness_critical=True)
+        return _check(
+            "redis_broker",
+            "Redis broker",
+            "error",
+            f"Redis broker check failed: {exc}",
+            readiness_critical=True,
+        )
 
 
 def _worker_check() -> dict[str, Any]:
@@ -531,25 +754,29 @@ def _worker_check() -> dict[str, Any]:
         return result
     if replies:
         worker_names = ", ".join(sorted(replies))
-        active_queues = sorted({
+        active_queues = sorted(
+            {
                 str(queue.get("name"))
                 for queues in active_queues_by_worker.values()
                 if isinstance(queues, list)
                 for queue in queues
                 if isinstance(queue, Mapping) and queue.get("name")
-        })
+            }
+        )
         expected_queues = _celery_queues()
         missing_queues = sorted(set(expected_queues) - set(active_queues))
         active_tasks = sum(
-            len(tasks)
+            len(tasks) for tasks in active_by_worker.values() if isinstance(tasks, list)
-            for tasks in active_by_worker.values()
-            if isinstance(tasks, list)
         )
         queue_depths = _queue_depths(expected_queues)
         state = "warning" if missing_queues else "ok"
         detail = f"{len(replies)} worker(s) replied: {worker_names}."
         if missing_queues:
-            detail += " No worker consumes configured queues: " + ", ".join(missing_queues) + "."
+            detail += (
+                " No worker consumes configured queues: "
+                + ", ".join(missing_queues)
+                + "."
+            )
         result = _check(
             "worker_split",
             "Background workers",
@@ -596,24 +823,48 @@ def _queue_depths(queues: list[str]) -> dict[str, int]:
         values = pipeline.execute()
     except Exception:  # noqa: BLE001 - worker heartbeat remains the readiness source.
         return {}
-    return {
+    return {queue: int(value) for queue, value in zip(queues, values, strict=True)}
-        queue: int(value)
-        for queue, value in zip(queues, values, strict=True)
-    }
 
 
 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)
+        configured = bool(
-        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)
+            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,
+        )
     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}"
+        detail = (
-        return _check("file_storage", "File storage", state, detail, readiness_critical=not writable)
+            f"Local file storage root: {root}"
-    return _check("file_storage", "File storage", "warning", f"Local file storage root does not exist yet: {root}", readiness_critical=False)
+            if writable
+            else f"Local file storage root is not writable: {root}"
+        )
+        return _check(
+            "file_storage",
+            "File storage",
+            state,
+            detail,
+            readiness_critical=not writable,
+        )
+    return _check(
+        "file_storage",
+        "File storage",
+        "warning",
+        f"Local file storage root does not exist yet: {root}",
+        readiness_critical=False,
+    )
 
 
 def _module_operational_checks(
@@ -630,13 +881,15 @@ def _module_operational_checks(
     for registration in registrations:
         cache_key = f"{registration.module_id}:{registration.check_id}"
         if registration.check_id in seen:
-            results.append(_check(
+            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(
@@ -694,7 +947,9 @@ def _backup_restore_check() -> dict[str, Any]:
         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
+        backup = (
+            snapshot.get("database_backup") if isinstance(snapshot, Mapping) else None
+        )
         if isinstance(backup, Mapping):
             latest_evidence = (record, backup)
             break
@@ -711,13 +966,10 @@ def _backup_restore_check() -> dict[str, Any]:
 
     record, backup = latest_evidence
     restore_check = backup.get("restore_check")
-    restore_check_ok = (
+    restore_check_ok = isinstance(restore_check, Mapping) and (
-        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] = []
@@ -725,9 +977,7 @@ def _backup_restore_check() -> dict[str, Any]:
         missing.append("backup restore-readiness check")
     if not drill_ok:
         missing.append("recorded rollback drill")
-    detail = (
+    detail = f"Installer run {record.get('run_id')} records a {backup.get('type', 'database')} backup."
-        f"Installer run {record.get('run_id')} records a {backup.get('type', 'database')} backup."
-    )
     if missing:
         detail += " Missing evidence: " + ", ".join(missing) + "."
     else:
@@ -749,7 +999,11 @@ def _backup_restore_check() -> dict[str, Any]:
 
 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"
+    path = (
+        Path(configured).expanduser()
+        if configured
+        else runtime_dir / "restore-drill-evidence.json"
+    )
     try:
         import json
 
@@ -762,7 +1016,12 @@ def _restore_drill_evidence(runtime_dir: Path) -> dict[str, object] | 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.")
+        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:
@@ -770,13 +1029,27 @@ def _deployment_security_check(current_profile: str) -> dict[str, Any]:
     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):
+    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.")
+        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
+    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",
@@ -806,14 +1079,32 @@ def _check(
     }
 
 
-def _readiness(checks: list[dict[str, Any]], maintenance_mode: dict[str, Any]) -> dict[str, Any]:
+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"]}
+        {
+            "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):
+    if maintenance_mode.get("enabled") and not any(
-        blockers.append({"id": "maintenance_mode", "label": "Maintenance mode", "state": "warning", "detail": str(maintenance_mode.get("message") or "Maintenance mode is enabled.")})
+        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,
@@ -833,11 +1124,19 @@ def _current_profile() -> str:
 
 
 def _celery_queues() -> list[str]:
-    return [item.strip() for item in str(core_settings.celery_queues or "").split(",") if item.strip()]
+    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()]
+    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]]:
@@ -846,28 +1145,53 @@ def _deployment_profiles(current_profile: str) -> list[dict[str, Any]]:
             "id": "local-dev",
             "name": "Local development",
             "current": current_profile == "local-dev",
-            "components": ["FastAPI", "SQLite or local database", "Vite dev server", "local file storage"],
+            "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"],
+            "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"],
+            "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"],
+            "components": [
+                "FastAPI",
+                "worker process",
+                "Redis broker",
+                "PostgreSQL",
+                "object storage",
+                "health probes",
+            ],
             "fit": "Campaigns, imports, exports, scheduled work, and larger tenant counts.",
         },
     ]

tests/test_operational_checks.py

@@ -52,11 +52,13 @@ def test_module_operational_check_failure_is_isolated() -> None:
         raise RuntimeError("secret detail")
 
     result = routes._module_operational_checks(
-        _Registry(OperationalCheckProviderRegistration(
+        _Registry(
+            OperationalCheckProviderRegistration(
                 module_id="example",
                 check_id="example.failed",
                 provider=provider,
-        )),
+            )
+        ),
         force=True,
     )[0]
 
@@ -82,3 +84,43 @@ def test_shared_runtime_cluster_missing_replicas_blocks_readiness() -> None:
     assert check["state"] == "error"
     assert check["readiness_critical"] is True
     assert check["metrics"]["recovery_required"] == 1
+
+
+def test_shared_runtime_cluster_skew_or_unserved_queue_blocks_readiness() -> None:
+    check = routes._runtime_cluster_check(
+        {
+            "available": True,
+            "state_profile": "shared",
+            "nodes": [],
+            "expected": {"api": 2, "worker": 2},
+            "active": {"api": 2, "worker": 2},
+            "composition": {"skewed": True},
+            "software_versions": {"skewed": False},
+            "queues": {"missing": ["calendar"]},
+            "recovery": {"requires_attention": 0},
+        }
+    )
+
+    assert check["state"] == "error"
+    assert check["readiness_critical"] is True
+    assert check["metrics"]["composition_skewed"] is True
+    assert check["metrics"]["missing_queues"] == 1
+
+
+def test_database_capacity_check_enforces_shared_rendered_budget(
+    monkeypatch,
+) -> None:
+    monkeypatch.setattr(routes.core_settings, "state_profile", "shared")
+    monkeypatch.setattr(routes.core_settings, "database_connection_limit", 100)
+    monkeypatch.setattr(routes.core_settings, "database_connection_reserve", 10)
+    monkeypatch.setattr(routes.core_settings, "database_connection_available", 90)
+    monkeypatch.setattr(routes.core_settings, "database_connection_peak", 70)
+
+    healthy = routes._database_capacity_check()
+    assert healthy["state"] == "ok"
+    assert healthy["metrics"]["peak"] == 70
+
+    monkeypatch.setattr(routes.core_settings, "database_connection_peak", 91)
+    overrun = routes._database_capacity_check()
+    assert overrun["state"] == "error"
+    assert overrun["readiness_critical"] is True

webui/src/api/ops.ts

@@ -117,6 +117,22 @@ export type OpsRuntimeCluster = {
   state_profile: string;
   expected?: { api: number; worker: number };
   active?: { api: number; worker: number };
+  composition?: {
+    expected_hash?: string | null;
+    active_hashes: string[];
+    unexpected_nodes: string[];
+    skewed: boolean;
+  };
+  software_versions?: {
+    active: string[];
+    skewed: boolean;
+  };
+  queues?: {
+    expected: string[];
+    active: string[];
+    missing: string[];
+    worker_pools: string[];
+  };
   nodes: OpsRuntimeNode[];
   recovery: {
     operations: OpsRecoveryOperation[];
@@ -137,6 +153,8 @@ export type OpsStatus = {
       message?: string | null;
     };
     database_url: string;
+    database_connection_peak?: number | null;
+    database_connection_available?: number | null;
     file_storage_backend: string;
     worker_metrics: {
       workers?: number;

webui/src/features/ops/OpsPage.tsx

@@ -125,6 +125,7 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth:
           <MetricCard label="i18n:govoplan-ops.redis.5eaa1f2f" value={status?.summary.redis_url ? "configured" : "-"} tone={status?.summary.celery_enabled ? "info" : "neutral"} detail={status?.summary.redis_url ?? "-"} />
           <MetricCard label="i18n:govoplan-ops.warnings.1430f976" value={warningCount + errorCount} tone={errorCount ? "danger" : warningCount ? "warning" : "good"} detail="i18n:govoplan-ops.current_health_checks.7830bccf" />
           <MetricCard label="Runtime nodes" value={status?.summary.runtime_node_count ?? 0} tone={status?.runtime_cluster.available ? "good" : "danger"} detail={runtimeNodeMetricDetail(status)} />
+          <MetricCard label="Database capacity" value={databaseCapacityValue(status)} tone={databaseCapacityTone(status)} detail="Peak pooled connections / available connections" />
           <MetricCard label="Recovery" value={status?.summary.recovery_required_count ?? 0} tone={status?.summary.recovery_required_count ? "danger" : "good"} detail="Operations requiring recovery attention" />
           <MetricCard label="Provider bindings" value={status?.governance.summary.configured_external_provider_count ?? 0} tone={status?.governance.summary.provider_attention_count ? "warning" : "good"} detail={`${status?.governance.summary.provider_attention_count ?? 0} requiring attention`} />
         </div>
@@ -480,6 +481,24 @@ function runtimeNodeMetricDetail(status: OpsStatus | null): string {
   return `${active.api}/${expected.api} API · ${active.worker}/${expected.worker} workers`;
 }
 
+function databaseCapacityValue(status: OpsStatus | null): string {
+  const peak = status?.summary.database_connection_peak;
+  const available = status?.summary.database_connection_available;
+  return typeof peak === "number" && typeof available === "number"
+    ? `${peak} / ${available}`
+    : "not declared";
+}
+
+function databaseCapacityTone(status: OpsStatus | null): "good" | "warning" | "danger" {
+  const peak = status?.summary.database_connection_peak;
+  const available = status?.summary.database_connection_available;
+  if (typeof peak !== "number" || typeof available !== "number") {
+    return status?.runtime_cluster.state_profile === "shared" ? "danger" : "warning";
+  }
+  if (peak > available) return "danger";
+  return peak / available >= 0.8 ? "warning" : "good";
+}
+
 function workerMetricDetail(status: OpsStatus | null): string {
   if (!status?.summary.celery_enabled) return "i18n:govoplan-ops.celery_worker_setting.323d7737";
   const metrics = status.summary.worker_metrics;