feat: add bounded Kubernetes runtime verification

.gitea/workflows/module-matrix.yml

@@ -46,6 +46,9 @@ jobs:
       - name: Install WebUI release dependencies with test scripts
         working-directory: govoplan
         run: bash tools/release/install-webui-release-dependencies.sh ../govoplan-core/webui
+      - name: Validate platform endpoint surface declarations
+        working-directory: govoplan
+        run: .venv/bin/python tools/inventory/platform-interface-inventory.py --strict
       - name: Validate Search against PostgreSQL
         working-directory: govoplan
         env:

docs/PLATFORM_CONTROL_PLANE.md

@@ -45,6 +45,28 @@ The command writes:
 - `audit-reports/platform-inventory/platform-interface-inventory.json`
 - `audit-reports/platform-inventory/platform-interface-inventory.md`
 
+Use `--strict` in CI. In addition to translation coverage, strict mode requires
+every backend endpoint without a statically visible WebUI path to have an exact
+entry in
+`tools/inventory/endpoint-surface-declarations.json`. The registry is keyed by
+repository, HTTP method, and canonical version-independent path. It accepts:
+
+- `ui_reachable`: a mounted router, generic action, or provider path hides the
+  reference from static extraction;
+- `intentionally_headless`: a capability/API is deliberately consumed without
+  its own UI;
+- `public_integration`: a documented public or interoperability endpoint;
+- `worker_internal`: a worker, scheduler, reconciliation, or monitoring path;
+- `compatibility`: a retained transition endpoint with a current replacement;
+- `missing_ui`: a real UI gap, which must include a Gitea tracking issue;
+- `removable`: a reviewed dead endpoint pending removal.
+
+Strict mode also rejects declarations that no longer match source. When an
+endpoint is added, changed, or removed, update its declaration in the same
+change. Do not classify an endpoint from a string mismatch alone: first check
+mounted prefixes, dynamic action paths, public clients, worker use, and
+capability consumers.
+
 It combines:
 
 1. loaded module manifests

docs/SCALING_AND_MULTI_HOST_DEPLOYMENT.md

@@ -150,8 +150,29 @@ versioning, and lifecycle controls.
 
 ## Capacity
 
-- Scale API replicas only within the PostgreSQL connection budget.
+- Set `GOVOPLAN_DB_CONNECTION_LIMIT` to the PostgreSQL role's effective
+  connection limit. The Kubernetes export reserves
+  `GOVOPLAN_DB_CONNECTION_RESERVE` connections and rejects a topology whose
+  calculated rolling-update peak would exceed the remainder. The calculation
+  includes API pools, every Celery parent and prefork child, the scheduler,
+  migration, and one surge replica per deployment. Role-specific pool and
+  overflow values are emitted into each workload rather than inherited from one
+  unconstrained global default.
 - Scale workers by queue, with upper bounds based on external provider limits.
+  `GOVOPLAN_WORKER_POOLS` may contain a JSON list of exact queue owners, for
+  example:
+
+  ```json
+  [
+    {"name":"delivery","queues":["send_email","append_sent"],"replicas":2,"concurrency":2},
+    {"name":"platform","queues":["events","workflow","default"],"replicas":2,"concurrency":2}
+  ]
+  ```
+
+  Pool replica totals must equal `replicas.worker`, and the pools must cover
+  `CELERY_QUEUES` exactly without duplicate ownership. Each pool receives its
+  own Deployment, disruption budget, topology-spread selector, runtime identity,
+  and declared concurrency.
 - Keep one fenced scheduler rather than load-balancing schedulers.
 - Increase WebUI replicas for asset/proxy capacity.
 - Measure request latency, database query time and locks, active connections,
@@ -181,3 +202,28 @@ availability, drill replica loss, rolling replacement, session continuity, job
 redelivery, scheduler failover, migration exclusion, object-store outage, and a
 coordinated database/object/key restore. Recovery rules and evidence are
 defined in [Recovery And Rollback Guarantees](RECOVERY_AND_ROLLBACK_GUARANTEES.md).
+
+## Live Multi-Host Evidence
+
+After deploying a pinned release on at least two Kubernetes nodes, create an API
+key with Ops read scope and run:
+
+```bash
+export GOVOPLAN_OPS_API_KEY='...'
+python tools/deployment/govoplan-deploy.py verify-kubernetes \
+  --directory /srv/govoplan/installation \
+  --namespace govoplan
+```
+
+The command fails unless API and WebUI pods are ready on at least two nodes,
+all rendered deployments are available, Ops reports a consistent release and
+module composition, every declared worker queue is served, and the calculated
+database peak remains below its budget. It writes a private, sanitized JSON
+record under the installation evidence directory and never retains the API key.
+
+Use `--exercise-api-pod-loss` in an approved drill window to delete one API pod,
+observe the public readiness path continuously, and record its replacement.
+This proves the bounded stateless-node-loss slice only. Session continuity,
+accepted-job redelivery, state-service failover, and coordinated restore remain
+separate target exercises whose signed evidence is governed by
+`docs/TARGET_MATURITY_EVIDENCE_RUNBOOK.md` and GovOPlaN #37.

docs/SECURITY_AUDIT.md

@@ -163,10 +163,20 @@ important scanner counts in the tracker issue.
 
 The jscpd step is intentionally scoped to application and test source. It
 excludes documentation snippets, package manifests, generated translations,
-public SVG assets, workflow YAML, and declarative backend schema JSON because
+public SVG assets and catalog output, workflow YAML, declarative backend schema
-those reports produce metadata or asset repetition rather than actionable source
+JSON, the generated migration baseline, and mirrored development migration
-duplication. Keep exclusions narrow and create child issues for source-code
+directories because those reports produce metadata or generated-source
-clusters that cross module ownership or make behavior harder to change safely.
+repetition rather than actionable source duplication. Keep exclusions narrow
+and create child issues for source-code clusters that cross module ownership or
+make behavior harder to change safely.
+
+The 2026-08-02 full-workspace baseline covered 64 repositories and reported
+1.82% duplicated lines before those generated-source exclusions. The reviewed
+high-value clusters were catalog acceptance persistence, release publication
+result assembly, and local WebUI JSON mutation wrappers. Similar Dataflow and
+Workflow graph/governance code remains independently owned until its shared
+contract is stable enough for Core; a raw similarity score is not grounds for a
+module-to-module dependency.
 
 ## Image Freshness
 

tests/test_deployment_installer.py

@@ -31,6 +31,9 @@ from govoplan_deploy.bundle import (  # noqa: E402
 )
 from govoplan_deploy.cli import _receipt_uses_direct_web_port, main  # noqa: E402
 import govoplan_deploy.cli as deployment_cli  # noqa: E402
+from govoplan_deploy.cluster_evidence import (  # noqa: E402
+    collect_kubernetes_evidence,
+)
 from govoplan_deploy.kubernetes import render_kubernetes  # noqa: E402
 from govoplan_deploy.model import (  # noqa: E402
     SpecError,
@@ -55,7 +58,104 @@ def run_cli(arguments: list[str]) -> tuple[int, str, str]:
     return result, stdout.getvalue(), stderr.getvalue()
 
 
+def _kubernetes_test_pod(name: str, component: str, node: str) -> dict:
+    return {
+        "metadata": {
+            "name": name,
+            "uid": f"uid-{name}",
+            "labels": {"app.kubernetes.io/component": component},
+        },
+        "spec": {"nodeName": node},
+        "status": {"conditions": [{"type": "Ready", "status": "True"}]},
+    }
+
+
+def _kubernetes_test_deployment(component: str, replicas: int) -> dict:
+    return {
+        "metadata": {
+            "name": f"govoplan-cluster-{component}",
+            "labels": {"app.kubernetes.io/component": component},
+        },
+        "spec": {"replicas": replicas},
+        "status": {"availableReplicas": replicas, "updatedReplicas": replicas},
+    }
+
+
 class DeploymentInstallerTests(unittest.TestCase):
+    def test_kubernetes_evidence_requires_two_node_spread_and_safe_runtime(
+        self,
+    ) -> None:
+        resources = {
+            "nodes": {
+                "items": [
+                    {
+                        "metadata": {"name": name},
+                        "spec": {},
+                        "status": {"conditions": [{"type": "Ready", "status": "True"}]},
+                    }
+                    for name in ("node-a", "node-b")
+                ]
+            },
+            "pods": {
+                "items": [
+                    _kubernetes_test_pod("api-a", "api", "node-a"),
+                    _kubernetes_test_pod("api-b", "api", "node-b"),
+                    _kubernetes_test_pod("web-a", "web", "node-a"),
+                    _kubernetes_test_pod("web-b", "web", "node-b"),
+                    _kubernetes_test_pod("worker-a", "worker", "node-a"),
+                ]
+            },
+            "deployments": {
+                "items": [
+                    _kubernetes_test_deployment("api", 2),
+                    _kubernetes_test_deployment("web", 2),
+                    _kubernetes_test_deployment("worker", 1),
+                ]
+            },
+        }
+
+        def run(arguments):
+            return resources[
+                "nodes"
+                if "nodes" in arguments
+                else "deployments"
+                if "deployments" in arguments
+                else "pods"
+            ]
+
+        evidence = collect_kubernetes_evidence(
+            installation_id="govoplan-cluster",
+            namespace="govoplan",
+            ops_url="https://govoplan.example.test/api/v1/ops/status",
+            api_key="not-retained",
+            command_runner=run,
+            json_fetcher=lambda _url, _key: {
+                "readiness": {"ready": True},
+                "runtime_cluster": {
+                    "expected": {"api": 2, "worker": 1},
+                    "active": {"api": 2, "worker": 1},
+                    "composition": {"skewed": False},
+                    "software_versions": {"skewed": False},
+                    "queues": {"missing": []},
+                },
+                "checks": [
+                    {
+                        "id": "database_capacity",
+                        "state": "ok",
+                        "detail": "Within budget",
+                        "metrics": {"peak": 30, "available": 90},
+                    }
+                ],
+            },
+        )
+
+        self.assertEqual("passed", evidence["result"]["state"])
+        self.assertNotIn("not-retained", json.dumps(evidence))
+        self.assertEqual(
+            ["node-a", "node-b"],
+            evidence["snapshot"]["ready_node_names"],
+        )
+
     def test_pre_migration_failure_restores_checksum_verified_applied_bundle(
         self,
     ) -> None:
@@ -212,6 +312,7 @@ class DeploymentInstallerTests(unittest.TestCase):
                 "FILE_STORAGE_S3_ACCESS_KEY_ID": "object-key",
                 "FILE_STORAGE_S3_SECRET_ACCESS_KEY": "object-secret",
                 "FILE_STORAGE_S3_BUCKET": "govoplan",
+                "GOVOPLAN_DB_CONNECTION_LIMIT": "100",
             },
         )
 
@@ -257,6 +358,103 @@ class DeploymentInstallerTests(unittest.TestCase):
                 "containers"
             ][0]["readinessProbe"]["httpGet"]["httpHeaders"],
         )
+        worker_command = deployments["govoplan-cluster-worker"]["spec"]["template"][
+            "spec"
+        ]["containers"][0]["command"]
+        self.assertIn("--concurrency", worker_command)
+        self.assertEqual(
+            "62",
+            manifest["metadata"]["annotations"][
+                "govoplan.add-ideas.de/database-connection-peak"
+            ],
+        )
+
+    def test_kubernetes_export_splits_worker_queues_and_rejects_capacity_overrun(
+        self,
+    ) -> None:
+        spec = default_spec(
+            installation_id="govoplan-cluster",
+            postgres_mode="external",
+            redis_mode="external",
+            storage_mode="s3",
+            api_replicas=2,
+            web_replicas=2,
+            worker_replicas=3,
+            api_image="registry.example.test/govoplan-api@sha256:" + "a" * 64,
+            web_image="registry.example.test/govoplan-web@sha256:" + "b" * 64,
+        )
+        environment = initial_secrets(
+            spec,
+            supplied={
+                "DATABASE_URL": "postgresql+psycopg://user:db-secret@postgres.example.test/govoplan",
+                "GOVOPLAN_DATABASE_URL_PGTOOLS": "postgresql://user:db-secret@postgres.example.test/govoplan",
+                "REDIS_URL": "rediss://:redis-secret@redis.example.test/0",
+                "FILE_STORAGE_S3_ENDPOINT_URL": "https://s3.example.test",
+                "FILE_STORAGE_S3_REGION": "eu-test-1",
+                "FILE_STORAGE_S3_ACCESS_KEY_ID": "object-key",
+                "FILE_STORAGE_S3_SECRET_ACCESS_KEY": "object-secret",
+                "FILE_STORAGE_S3_BUCKET": "govoplan",
+                "GOVOPLAN_DB_CONNECTION_LIMIT": "100",
+            },
+        )
+        queues = environment["CELERY_QUEUES"].split(",")
+        environment["GOVOPLAN_WORKER_POOLS"] = json.dumps(
+            [
+                {
+                    "name": "delivery",
+                    "queues": queues[:3],
+                    "replicas": 2,
+                    "concurrency": 2,
+                },
+                {
+                    "name": "platform",
+                    "queues": queues[3:],
+                    "replicas": 1,
+                    "concurrency": 1,
+                },
+            ]
+        )
+
+        manifest = render_kubernetes(spec, environment)
+        workers = [
+            item
+            for item in manifest["items"]
+            if item["kind"] == "Deployment"
+            and item["metadata"]["labels"].get("app.kubernetes.io/component")
+            == "worker"
+        ]
+
+        self.assertEqual(2, len(workers))
+        self.assertEqual(
+            {"delivery", "platform"},
+            {
+                item["metadata"]["labels"]["govoplan.add-ideas.de/worker-pool"]
+                for item in workers
+            },
+        )
+        for deployment in workers:
+            container = deployment["spec"]["template"]["spec"]["containers"][0]
+            explicit_environment = {
+                item["name"]: item.get("value")
+                for item in container["env"]
+                if "value" in item
+            }
+            self.assertEqual(
+                deployment["metadata"]["labels"]["govoplan.add-ideas.de/worker-pool"],
+                explicit_environment["GOVOPLAN_WORKER_POOL"],
+            )
+            self.assertEqual(
+                set(
+                    container["command"][
+                        container["command"].index("--queues") + 1
+                    ].split(",")
+                ),
+                set(explicit_environment["CELERY_QUEUES"].split(",")),
+            )
+
+        environment["GOVOPLAN_DB_CONNECTION_LIMIT"] = "40"
+        with self.assertRaisesRegex(ValueError, "connection peak"):
+            render_kubernetes(spec, environment)
 
     def test_kubernetes_export_rejects_mutable_release_images(self) -> None:
         spec = default_spec(

tests/test_platform_interface_inventory.py

@@ -2,7 +2,9 @@ from __future__ import annotations
 
 import ast
 import importlib.util
+import json
 from pathlib import Path
+import tempfile
 import unittest
 
 
@@ -30,6 +32,92 @@ class PlatformInterfaceInventoryTests(unittest.TestCase):
             inventory.canonical_api_path("/api/v2/campaigns/{campaign_id}"),
             "/campaigns/{}",
         )
+        self.assertEqual(
+            inventory.canonical_api_path("/api/v1/calendar/events/delta${querySuffix}"),
+            "/calendar/events/delta",
+        )
+        self.assertEqual(
+            inventory.canonical_api_path("/api/v1/calendar/events/${eventId}"),
+            "/calendar/events/{}",
+        )
+
+    def test_endpoint_declarations_are_exact_and_require_missing_ui_issue(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "endpoints.json"
+            path.write_text(
+                json.dumps(
+                    {
+                        "schema_version": 1,
+                        "endpoints": [
+                            {
+                                "repository": "govoplan-example",
+                                "method": "GET",
+                                "path": "/example/items/{}",
+                                "category": "public_integration",
+                                "rationale": "Published integration API.",
+                            }
+                        ],
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            declarations = inventory._load_endpoint_declarations(path)
+
+            self.assertIn(
+                ("govoplan-example", "GET", "/example/items/{}"),
+                declarations,
+            )
+            payload = json.loads(path.read_text(encoding="utf-8"))
+            payload["endpoints"][0]["category"] = "missing_ui"
+            path.write_text(json.dumps(payload), encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "tracking_issue"):
+                inventory._load_endpoint_declarations(path)
+
+    def test_inventory_reports_unclassified_and_stale_endpoint_declarations(
+        self,
+    ) -> None:
+        webui = {
+            "frontendApiReferences": [],
+            "translationUsages": [],
+            "translationCatalog": {"en": {}, "de": {}},
+            "fields": [],
+            "labels": [],
+            "visibleText": [],
+            "routes": [],
+            "navigation": [],
+            "uiCapabilities": [],
+            "dynamicTranslationUsages": [],
+        }
+        endpoint = {
+            "repository": "govoplan-example",
+            "method": "GET",
+            "path": "/api/v1/example/items",
+            "file": "src/example.py",
+            "line": 1,
+            "handler": "items",
+            "router": "router",
+        }
+        stale = {
+            "repository": "govoplan-example",
+            "method": "GET",
+            "path": "/example/removed",
+            "category": "removable",
+            "rationale": "Removal is pending.",
+        }
+
+        result = inventory._assemble_inventory(
+            webui=webui,
+            backend_endpoints=[endpoint],
+            manifests=[],
+            endpoint_declarations={
+                ("govoplan-example", "GET", "/example/removed"): stale,
+            },
+        )
+
+        self.assertEqual(1, result["summary"]["unclassified_backend_endpoints"])
+        self.assertEqual(1, result["summary"]["stale_endpoint_declarations"])
+        self.assertIsNone(result["api"]["backend_endpoints"][0]["surface"])
 
     def test_fastapi_route_scanner_includes_router_prefix(self) -> None:
         tree = ast.parse(

tools/checks/check-security-audit.sh

@@ -851,6 +851,9 @@ run_jscpd() {
     "package.json"
     "**/generatedTranslations.ts"
     "**/docs/gitea-labels.json"
+    "**/docs/migration-release-baselines.json"
+    "**/public/catalogs/**"
+    "**/alembic/dev_versions/**"
     "**/*.md"
     "**/*.md:*"
     "*.md"

tools/deployment/govoplan_deploy/bundle.py

@@ -40,6 +40,18 @@ RUNTIME_ENV_KEYS = (
     "ENABLED_MODULES",
     "CELERY_ENABLED",
     "CELERY_QUEUES",
+    "CELERY_WORKER_CONCURRENCY",
+    "GOVOPLAN_DB_CONNECTION_LIMIT",
+    "GOVOPLAN_DB_CONNECTION_RESERVE",
+    "GOVOPLAN_API_DB_POOL_SIZE",
+    "GOVOPLAN_API_DB_MAX_OVERFLOW",
+    "GOVOPLAN_WORKER_DB_POOL_SIZE",
+    "GOVOPLAN_WORKER_DB_MAX_OVERFLOW",
+    "GOVOPLAN_SCHEDULER_DB_POOL_SIZE",
+    "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW",
+    "GOVOPLAN_MIGRATION_DB_POOL_SIZE",
+    "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW",
+    "GOVOPLAN_WORKER_POOLS",
     "REDIS_URL",
     "CORS_ORIGINS",
     "GOVOPLAN_TRUSTED_HOSTS",
@@ -193,6 +205,46 @@ def reconcile_runtime_environment(
             "CELERY_QUEUES": (
                 "send_email,append_sent,notifications,mail,calendar,dataflow,workflow,postbox,events,idm,default"
             ),
+            "CELERY_WORKER_CONCURRENCY": values.get(
+                "CELERY_WORKER_CONCURRENCY",
+                "2",
+            ),
+            "GOVOPLAN_DB_CONNECTION_RESERVE": values.get(
+                "GOVOPLAN_DB_CONNECTION_RESERVE",
+                "10",
+            ),
+            "GOVOPLAN_API_DB_POOL_SIZE": values.get(
+                "GOVOPLAN_API_DB_POOL_SIZE",
+                "5",
+            ),
+            "GOVOPLAN_API_DB_MAX_OVERFLOW": values.get(
+                "GOVOPLAN_API_DB_MAX_OVERFLOW",
+                "2",
+            ),
+            "GOVOPLAN_WORKER_DB_POOL_SIZE": values.get(
+                "GOVOPLAN_WORKER_DB_POOL_SIZE",
+                "1",
+            ),
+            "GOVOPLAN_WORKER_DB_MAX_OVERFLOW": values.get(
+                "GOVOPLAN_WORKER_DB_MAX_OVERFLOW",
+                "1",
+            ),
+            "GOVOPLAN_SCHEDULER_DB_POOL_SIZE": values.get(
+                "GOVOPLAN_SCHEDULER_DB_POOL_SIZE",
+                "1",
+            ),
+            "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW": values.get(
+                "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW",
+                "0",
+            ),
+            "GOVOPLAN_MIGRATION_DB_POOL_SIZE": values.get(
+                "GOVOPLAN_MIGRATION_DB_POOL_SIZE",
+                "2",
+            ),
+            "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW": values.get(
+                "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW",
+                "0",
+            ),
             "CORS_ORIGINS": spec.public_url,
             "GOVOPLAN_TRUSTED_HOSTS": public.hostname or "",
             "FORWARDED_ALLOW_IPS": spec.network_subnet,
@@ -212,6 +264,8 @@ def reconcile_runtime_environment(
         values["GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE"] = "true"
     else:
         values["GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE"] = "false"
+    if postgres.mode == "managed":
+        values.setdefault("GOVOPLAN_DB_CONNECTION_LIMIT", "100")
 
     storage = spec.components.storage
     if storage.mode == "local":

tools/deployment/govoplan_deploy/cli.py

@@ -35,6 +35,7 @@ from .bundle import (
     service_names,
     write_env,
 )
+from .cluster_evidence import collect_kubernetes_evidence
 from .model import (
     ComponentConfig,
     DEFAULT_GARAGE_IMAGE,
@@ -141,6 +142,37 @@ def build_parser() -> argparse.ArgumentParser:
         help="Output JSON path; defaults to <directory>/kubernetes.json.",
     )
 
+    verify_kubernetes = subparsers.add_parser(
+        "verify-kubernetes",
+        help="Collect sanitized readiness evidence from a live multi-host profile.",
+    )
+    _directory_argument(verify_kubernetes)
+    verify_kubernetes.add_argument("--namespace", default="govoplan")
+    verify_kubernetes.add_argument(
+        "--ops-url",
+        help="Ops status URL; defaults to <public-url>/api/v1/ops/status.",
+    )
+    verify_kubernetes.add_argument(
+        "--api-key-env",
+        default="GOVOPLAN_OPS_API_KEY",
+        help="Environment variable containing an API key with Ops read scope.",
+    )
+    verify_kubernetes.add_argument(
+        "--exercise-api-pod-loss",
+        action="store_true",
+        help="Delete one ready API pod and prove health-aware replacement.",
+    )
+    verify_kubernetes.add_argument(
+        "--timeout-seconds",
+        type=float,
+        default=180.0,
+    )
+    verify_kubernetes.add_argument(
+        "--output",
+        type=Path,
+        help="Private evidence output path.",
+    )
+
     operations = subparsers.add_parser(
         "operations",
         help="List durable deployment and recovery operations.",
@@ -282,6 +314,8 @@ def main(argv: Sequence[str] | None = None) -> int:
             return _status(args)
         if args.command == "render-kubernetes":
             return _render_kubernetes(args)
+        if args.command == "verify-kubernetes":
+            return _verify_kubernetes(args)
         if args.command == "operations":
             return _operations(args)
         if args.command == "recover":
@@ -598,9 +632,7 @@ def _render_kubernetes(args: argparse.Namespace) -> int:
     output = (args.output or (paths.root / "kubernetes.json")).expanduser().resolve()
     atomic_write(output, canonical_json(manifest), mode=0o600)
     print(f"Wrote stateless Kubernetes runtime manifest to {output}")
-    print(
+    print("Required Secret keys: " + ", ".join(kubernetes_secret_contract()))
-        "Required Secret keys: " + ", ".join(kubernetes_secret_contract())
-    )
     print(
         write_secret_creation_hint(
             paths.env,
@@ -611,6 +643,41 @@ def _render_kubernetes(args: argparse.Namespace) -> int:
     return 0
 
 
+def _verify_kubernetes(args: argparse.Namespace) -> int:
+    paths = bundle_paths(args.directory)
+    spec = load_spec(paths.spec)
+    api_key = str(os.environ.get(args.api_key_env) or "").strip()
+    if not api_key:
+        raise ValueError(
+            f"{args.api_key_env} must contain an API key with Ops read scope"
+        )
+    ops_url = args.ops_url or (spec.public_url.rstrip("/") + "/api/v1/ops/status")
+    evidence = collect_kubernetes_evidence(
+        installation_id=spec.installation_id,
+        namespace=args.namespace,
+        ops_url=ops_url,
+        api_key=api_key,
+        exercise_api_pod_loss=args.exercise_api_pod_loss,
+        timeout_seconds=args.timeout_seconds,
+    )
+    timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
+    output = (
+        (args.output or paths.root / "evidence" / f"kubernetes-{timestamp}.json")
+        .expanduser()
+        .resolve()
+    )
+    ensure_private_directory(output.parent)
+    atomic_write(output, canonical_json(evidence), mode=0o600)
+    print(f"Wrote Kubernetes evidence to {output}")
+    if evidence["result"]["state"] != "passed":
+        print(
+            "Failed checks: " + ", ".join(evidence["result"]["failed_checks"]),
+            file=sys.stderr,
+        )
+        return 1
+    return 0
+
+
 def _operations(args: argparse.Namespace) -> int:
     paths = bundle_paths(args.directory)
     payload = list_operations(paths)

tools/deployment/govoplan_deploy/cluster_evidence.py

@@ -0,0 +1,380 @@
+"""Collect sanitized evidence for the Kubernetes multi-host profile."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping, Sequence
+from datetime import UTC, datetime
+import json
+import shutil
+import subprocess
+import time
+from typing import Any
+from urllib.request import Request, urlopen
+
+
+JsonObject = dict[str, Any]
+CommandRunner = Callable[[Sequence[str]], JsonObject]
+JsonFetcher = Callable[[str, str], JsonObject]
+
+
+def collect_kubernetes_evidence(
+    *,
+    installation_id: str,
+    namespace: str,
+    ops_url: str,
+    api_key: str,
+    exercise_api_pod_loss: bool = False,
+    timeout_seconds: float = 180.0,
+    command_runner: CommandRunner | None = None,
+    json_fetcher: JsonFetcher | None = None,
+) -> JsonObject:
+    """Inspect a live cluster and optionally exercise one API pod replacement."""
+
+    run_json = command_runner or _kubectl_json
+    fetch_json = json_fetcher or _fetch_json
+    nodes = run_json(("get", "nodes", "-o", "json"))
+    pods = run_json(
+        (
+            "-n",
+            namespace,
+            "get",
+            "pods",
+            "-l",
+            f"app.kubernetes.io/instance={installation_id}",
+            "-o",
+            "json",
+        )
+    )
+    deployments = run_json(
+        (
+            "-n",
+            namespace,
+            "get",
+            "deployments",
+            "-l",
+            f"app.kubernetes.io/instance={installation_id}",
+            "-o",
+            "json",
+        )
+    )
+    ops = fetch_json(ops_url, api_key)
+    snapshot = _evaluate_snapshot(
+        installation_id=installation_id,
+        namespace=namespace,
+        nodes=nodes,
+        pods=pods,
+        deployments=deployments,
+        ops=ops,
+    )
+    replacement: JsonObject | None = None
+    if exercise_api_pod_loss:
+        replacement = _exercise_api_pod_loss(
+            installation_id=installation_id,
+            namespace=namespace,
+            ops_url=ops_url,
+            api_key=api_key,
+            initial_pods=pods,
+            timeout_seconds=timeout_seconds,
+            run_json=run_json,
+            fetch_json=fetch_json,
+        )
+    evidence = {
+        "schema_version": 1,
+        "evidence_kind": "govoplan.kubernetes-multi-host",
+        "collected_at": datetime.now(UTC).isoformat(),
+        "installation_id": installation_id,
+        "namespace": namespace,
+        "snapshot": snapshot,
+        "api_pod_loss": replacement,
+    }
+    failures = [
+        check["id"] for check in snapshot["checks"] if check["state"] != "passed"
+    ]
+    if replacement and replacement["state"] != "passed":
+        failures.append("api_pod_loss")
+    evidence["result"] = {
+        "state": "passed" if not failures else "failed",
+        "failed_checks": failures,
+    }
+    return evidence
+
+
+def _evaluate_snapshot(
+    *,
+    installation_id: str,
+    namespace: str,
+    nodes: Mapping[str, Any],
+    pods: Mapping[str, Any],
+    deployments: Mapping[str, Any],
+    ops: Mapping[str, Any],
+) -> JsonObject:
+    ready_nodes = [
+        item
+        for item in _items(nodes)
+        if not bool(item.get("spec", {}).get("unschedulable"))
+        and _condition(item, "Ready") == "True"
+    ]
+    pod_rows = [_pod_summary(item) for item in _items(pods)]
+    deployment_rows = [_deployment_summary(item) for item in _items(deployments)]
+    ready_api = [
+        item for item in pod_rows if item["component"] == "api" and item["ready"]
+    ]
+    ready_web = [
+        item for item in pod_rows if item["component"] == "web" and item["ready"]
+    ]
+    runtime = ops.get("runtime_cluster")
+    runtime = runtime if isinstance(runtime, Mapping) else {}
+    checks = ops.get("checks")
+    check_rows = checks if isinstance(checks, list) else []
+    check_by_id = {
+        str(item.get("id")): item for item in check_rows if isinstance(item, Mapping)
+    }
+    ops_readiness = ops.get("readiness")
+    ops_readiness = ops_readiness if isinstance(ops_readiness, Mapping) else {}
+    snapshot_checks = [
+        _check(
+            "ready_nodes",
+            len(ready_nodes) >= 2,
+            f"{len(ready_nodes)} ready schedulable node(s)",
+        ),
+        _check(
+            "api_node_spread",
+            len(ready_api) >= 2 and len({item["node"] for item in ready_api}) >= 2,
+            f"{len(ready_api)} ready API pod(s) on {len({item['node'] for item in ready_api})} node(s)",
+        ),
+        _check(
+            "web_node_spread",
+            len(ready_web) >= 2 and len({item["node"] for item in ready_web}) >= 2,
+            f"{len(ready_web)} ready WebUI pod(s) on {len({item['node'] for item in ready_web})} node(s)",
+        ),
+        _check(
+            "deployment_availability",
+            bool(deployment_rows)
+            and all(item["available"] >= item["desired"] for item in deployment_rows),
+            f"{len(deployment_rows)} deployment(s) inspected",
+        ),
+        _check(
+            "ops_readiness",
+            bool(ops_readiness.get("ready")),
+            str(ops_readiness.get("detail") or "Ops readiness response inspected"),
+        ),
+        _check(
+            "runtime_composition",
+            not bool(runtime.get("composition", {}).get("skewed")),
+            "Active runtime composition matches the loaded module graph",
+        ),
+        _check(
+            "runtime_versions",
+            not bool(runtime.get("software_versions", {}).get("skewed")),
+            "Active runtime software versions are consistent",
+        ),
+        _check(
+            "worker_queue_coverage",
+            not bool(runtime.get("queues", {}).get("missing")),
+            "Every configured queue has an active worker",
+        ),
+        _check(
+            "database_connection_budget",
+            str(check_by_id.get("database_capacity", {}).get("state")) == "ok",
+            str(
+                check_by_id.get("database_capacity", {}).get("detail")
+                or "Database capacity check is unavailable"
+            ),
+        ),
+    ]
+    return {
+        "installation_id": installation_id,
+        "namespace": namespace,
+        "ready_node_names": sorted(_name(item) for item in ready_nodes),
+        "pods": sorted(pod_rows, key=lambda item: item["name"]),
+        "deployments": sorted(deployment_rows, key=lambda item: item["name"]),
+        "runtime": {
+            "expected": dict(runtime.get("expected") or {}),
+            "active": dict(runtime.get("active") or {}),
+            "composition": dict(runtime.get("composition") or {}),
+            "software_versions": dict(runtime.get("software_versions") or {}),
+            "queues": dict(runtime.get("queues") or {}),
+        },
+        "database_capacity": dict(
+            check_by_id.get("database_capacity", {}).get("metrics") or {}
+        ),
+        "checks": snapshot_checks,
+    }
+
+
+def _exercise_api_pod_loss(
+    *,
+    installation_id: str,
+    namespace: str,
+    ops_url: str,
+    api_key: str,
+    initial_pods: Mapping[str, Any],
+    timeout_seconds: float,
+    run_json: CommandRunner,
+    fetch_json: JsonFetcher,
+) -> JsonObject:
+    candidates = [
+        _pod_summary(item)
+        for item in _items(initial_pods)
+        if item.get("metadata", {}).get("labels", {}).get("app.kubernetes.io/component")
+        == "api"
+        and _condition(item, "Ready") == "True"
+    ]
+    if len(candidates) < 2:
+        return {
+            "state": "failed",
+            "detail": "At least two ready API pods are required for the loss drill.",
+        }
+    victim = sorted(candidates, key=lambda item: item["name"])[0]
+    initial_uids = {item["uid"] for item in candidates}
+    desired_ready = len(candidates)
+    run_json(
+        (
+            "-n",
+            namespace,
+            "delete",
+            "pod",
+            victim["name"],
+            "--wait=false",
+            "-o",
+            "json",
+        )
+    )
+    deadline = time.monotonic() + timeout_seconds
+    health_failures = 0
+    replacement: JsonObject | None = None
+    while time.monotonic() < deadline:
+        try:
+            current_ops = fetch_json(ops_url, api_key)
+            if not bool(current_ops.get("readiness", {}).get("ready")):
+                health_failures += 1
+        except (OSError, ValueError):
+            health_failures += 1
+        current = run_json(
+            (
+                "-n",
+                namespace,
+                "get",
+                "pods",
+                "-l",
+                f"app.kubernetes.io/instance={installation_id},app.kubernetes.io/component=api",
+                "-o",
+                "json",
+            )
+        )
+        ready = [
+            _pod_summary(item)
+            for item in _items(current)
+            if _condition(item, "Ready") == "True"
+        ]
+        new_pods = [item for item in ready if item["uid"] not in initial_uids]
+        if len(ready) >= desired_ready and new_pods:
+            replacement = sorted(new_pods, key=lambda item: item["name"])[-1]
+            break
+        time.sleep(2.0)
+    passed = replacement is not None and health_failures == 0
+    return {
+        "state": "passed" if passed else "failed",
+        "victim": victim,
+        "replacement": replacement,
+        "readiness_failures": health_failures,
+        "detail": (
+            "API pod was replaced without an observed readiness outage."
+            if passed
+            else "API replacement timed out or readiness failed during replacement."
+        ),
+    }
+
+
+def _kubectl_json(arguments: Sequence[str]) -> JsonObject:
+    kubectl = shutil.which("kubectl")
+    if kubectl is None:
+        raise ValueError("kubectl is required for Kubernetes evidence collection")
+    result = subprocess.run(
+        (kubectl, *arguments),
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+    if result.returncode:
+        detail = result.stderr.strip() or result.stdout.strip()
+        raise ValueError(f"kubectl failed: {detail}")
+    try:
+        payload = json.loads(result.stdout)
+    except json.JSONDecodeError as exc:
+        raise ValueError("kubectl did not return JSON") from exc
+    if not isinstance(payload, dict):
+        raise ValueError("kubectl returned a non-object JSON payload")
+    return payload
+
+
+def _fetch_json(url: str, api_key: str) -> JsonObject:
+    request = Request(
+        url,
+        headers={"Accept": "application/json", "X-API-Key": api_key},
+    )
+    with urlopen(request, timeout=15) as response:
+        payload = json.load(response)
+    if not isinstance(payload, dict):
+        raise ValueError("Ops API returned a non-object JSON payload")
+    return payload
+
+
+def _items(payload: Mapping[str, Any]) -> list[Mapping[str, Any]]:
+    items = payload.get("items")
+    if not isinstance(items, list):
+        raise ValueError("Kubernetes list response has no items")
+    return [item for item in items if isinstance(item, Mapping)]
+
+
+def _condition(item: Mapping[str, Any], condition_type: str) -> str | None:
+    conditions = item.get("status", {}).get("conditions", [])
+    for condition in conditions if isinstance(conditions, list) else []:
+        if isinstance(condition, Mapping) and condition.get("type") == condition_type:
+            return str(condition.get("status"))
+    return None
+
+
+def _name(item: Mapping[str, Any]) -> str:
+    return str(item.get("metadata", {}).get("name") or "")
+
+
+def _pod_summary(item: Mapping[str, Any]) -> JsonObject:
+    metadata = item.get("metadata", {})
+    labels = metadata.get("labels", {})
+    return {
+        "name": _name(item),
+        "uid": str(metadata.get("uid") or ""),
+        "component": str(labels.get("app.kubernetes.io/component") or ""),
+        "worker_pool": labels.get("govoplan.add-ideas.de/worker-pool"),
+        "node": str(item.get("spec", {}).get("nodeName") or ""),
+        "ready": _condition(item, "Ready") == "True",
+    }
+
+
+def _deployment_summary(item: Mapping[str, Any]) -> JsonObject:
+    status = item.get("status", {})
+    return {
+        "name": _name(item),
+        "component": str(
+            item.get("metadata", {})
+            .get("labels", {})
+            .get("app.kubernetes.io/component")
+            or ""
+        ),
+        "desired": int(item.get("spec", {}).get("replicas") or 0),
+        "available": int(status.get("availableReplicas") or 0),
+        "updated": int(status.get("updatedReplicas") or 0),
+    }
+
+
+def _check(check_id: str, passed: bool, detail: str) -> JsonObject:
+    return {
+        "id": check_id,
+        "state": "passed" if passed else "failed",
+        "detail": detail,
+    }
+
+
+__all__ = ["collect_kubernetes_evidence"]

tools/deployment/govoplan_deploy/kubernetes.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from hashlib import sha256
+import json
 from pathlib import Path
 import re
 from typing import Any, Mapping
@@ -26,6 +28,17 @@ _CONFIG_KEYS = (
     "ENABLED_MODULES",
     "CELERY_ENABLED",
     "CELERY_QUEUES",
+    "GOVOPLAN_DB_CONNECTION_LIMIT",
+    "GOVOPLAN_DB_CONNECTION_RESERVE",
+    "GOVOPLAN_API_DB_POOL_SIZE",
+    "GOVOPLAN_API_DB_MAX_OVERFLOW",
+    "GOVOPLAN_WORKER_DB_POOL_SIZE",
+    "GOVOPLAN_WORKER_DB_MAX_OVERFLOW",
+    "GOVOPLAN_SCHEDULER_DB_POOL_SIZE",
+    "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW",
+    "GOVOPLAN_MIGRATION_DB_POOL_SIZE",
+    "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW",
+    "GOVOPLAN_WORKER_POOLS",
     "CORS_ORIGINS",
     "GOVOPLAN_TRUSTED_HOSTS",
     "FORWARDED_ALLOW_IPS",
@@ -43,6 +56,15 @@ _CONFIG_KEYS = (
     "FILE_STORAGE_S3_DEPLOYMENT_MANAGED",
     "FILE_STORAGE_S3_ENDPOINT_TRUSTED",
 )
+_QUEUE_NAME = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$")
+
+
+@dataclass(frozen=True, slots=True)
+class _WorkerPool:
+    name: str
+    queues: tuple[str, ...]
+    replicas: int
+    concurrency: int
 
 
 def render_kubernetes(
@@ -57,6 +79,8 @@ def render_kubernetes(
     """Render runtime roles only; shared state services stay externally managed."""
 
     _validate_cluster_profile(spec, environment, namespace, secret_name)
+    worker_pools = _worker_pools(spec, environment)
+    database_capacity = _database_capacity(spec, environment, worker_pools)
     name = _resource_name(spec.installation_id)
     public_host = urlsplit(spec.public_url).hostname or "localhost"
     labels = {"app.kubernetes.io/name": "govoplan", "app.kubernetes.io/instance": name}
@@ -75,6 +99,8 @@ def render_kubernetes(
             "GOVOPLAN_RUNTIME_STALE_AFTER_SECONDS": "60",
             "GOVOPLAN_EXPECTED_API_REPLICAS": str(spec.replicas.api),
             "GOVOPLAN_EXPECTED_WORKER_REPLICAS": str(spec.replicas.worker),
+            "GOVOPLAN_DB_CONNECTION_PEAK": str(database_capacity["peak"]),
+            "GOVOPLAN_DB_CONNECTION_AVAILABLE": str(database_capacity["available"]),
             "FILE_STORAGE_BACKEND": "s3",
             "FILE_STORAGE_S3_DEPLOYMENT_MANAGED": "false",
             "FILE_STORAGE_S3_ENDPOINT_TRUSTED": "true",
@@ -130,6 +156,7 @@ def render_kubernetes(
             readiness_path="/health/ready",
             liveness_path="/health",
             probe_host=public_host,
+            extra_environment=_role_database_environment(environment, "API"),
         ),
         _service(
             name=f"{name}-api",
@@ -168,16 +195,25 @@ def render_kubernetes(
             config_name=config_name,
             secret_name=secret_name,
             service_account=service_account,
+            database_environment=_role_database_environment(
+                environment,
+                "MIGRATION",
+            ),
         ),
     ]
-    if spec.replicas.worker:
+    for pool in worker_pools:
+        worker_name = (
+            f"{name}-worker"
+            if len(worker_pools) == 1 and pool.name == "default"
+            else _name_with_suffix(name, f"worker-{pool.name}")
+        )
         items.append(
             _deployment(
-                name=f"{name}-worker",
+                name=worker_name,
                 namespace=namespace,
                 labels=labels,
                 role="worker",
-                replicas=spec.replicas.worker,
+                replicas=pool.replicas,
                 image=spec.release.api_image,
                 command=(
                     "python",
@@ -187,15 +223,39 @@ def render_kubernetes(
                     "govoplan_core.celery_app:celery",
                     "worker",
                     "--queues",
-                    str(environment["CELERY_QUEUES"]),
+                    ",".join(pool.queues),
+                    "--concurrency",
+                    str(pool.concurrency),
                     "--loglevel",
                     "INFO",
                 ),
                 config_name=config_name,
                 secret_name=secret_name,
                 service_account=service_account,
+                extra_environment={
+                    **_role_database_environment(environment, "WORKER"),
+                    "CELERY_QUEUES": ",".join(pool.queues),
+                    "CELERY_WORKER_CONCURRENCY": str(pool.concurrency),
+                    "GOVOPLAN_WORKER_POOL": pool.name,
+                },
+                selector_labels={
+                    "govoplan.add-ideas.de/worker-pool": pool.name,
+                },
             )
         )
+        if pool.replicas > 1:
+            items.append(
+                _pod_disruption_budget(
+                    worker_name,
+                    namespace,
+                    labels,
+                    "worker",
+                    selector_labels={
+                        "govoplan.add-ideas.de/worker-pool": pool.name,
+                    },
+                )
+            )
+    if worker_pools:
         items.append(
             _deployment(
                 name=f"{name}-scheduler",
@@ -225,6 +285,10 @@ def render_kubernetes(
                 config_name=config_name,
                 secret_name=secret_name,
                 service_account=service_account,
+                extra_environment=_role_database_environment(
+                    environment,
+                    "SCHEDULER",
+                ),
             )
         )
     if spec.replicas.api > 1:
@@ -248,6 +312,15 @@ def render_kubernetes(
             "annotations": {
                 "govoplan.add-ideas.de/profile": "stateless-shared-state",
                 "govoplan.add-ideas.de/secret-contract": ",".join(_SECRET_KEYS),
+                "govoplan.add-ideas.de/database-connection-peak": str(
+                    database_capacity["peak"]
+                ),
+                "govoplan.add-ideas.de/database-connection-available": str(
+                    database_capacity["available"]
+                ),
+                "govoplan.add-ideas.de/worker-pools": ",".join(
+                    pool.name for pool in worker_pools
+                ),
             }
         },
         "items": items,
@@ -290,7 +363,13 @@ def _validate_cluster_profile(
             + ", ".join(unpinned_images)
         )
     missing = [
-        key for key in (*_SECRET_KEYS, "CELERY_QUEUES") if not environment.get(key)
+        key
+        for key in (
+            *_SECRET_KEYS,
+            "CELERY_QUEUES",
+            "GOVOPLAN_DB_CONNECTION_LIMIT",
+        )
+        if not environment.get(key)
     ]
     if missing:
         raise ValueError(
@@ -298,6 +377,245 @@ def _validate_cluster_profile(
         )
 
 
+def _worker_pools(
+    spec: InstallationSpec,
+    environment: Mapping[str, str],
+) -> tuple[_WorkerPool, ...]:
+    if spec.replicas.worker == 0:
+        if str(environment.get("GOVOPLAN_WORKER_POOLS") or "").strip():
+            raise ValueError("Worker pools require at least one worker replica")
+        return ()
+    configured_queues = tuple(
+        item.strip()
+        for item in str(environment.get("CELERY_QUEUES") or "").split(",")
+        if item.strip()
+    )
+    if not configured_queues or any(
+        _QUEUE_NAME.fullmatch(item) is None for item in configured_queues
+    ):
+        raise ValueError("CELERY_QUEUES must contain canonical queue names")
+    raw_pools = str(environment.get("GOVOPLAN_WORKER_POOLS") or "").strip()
+    if not raw_pools:
+        return (
+            _WorkerPool(
+                name="default",
+                queues=configured_queues,
+                replicas=spec.replicas.worker,
+                concurrency=_bounded_environment_integer(
+                    environment,
+                    "CELERY_WORKER_CONCURRENCY",
+                    default=2,
+                    minimum=1,
+                    maximum=64,
+                ),
+            ),
+        )
+    try:
+        payload = json.loads(raw_pools)
+    except json.JSONDecodeError as exc:
+        raise ValueError("GOVOPLAN_WORKER_POOLS must be valid JSON") from exc
+    if not isinstance(payload, list) or not payload:
+        raise ValueError("GOVOPLAN_WORKER_POOLS must be a non-empty JSON list")
+    pools: list[_WorkerPool] = []
+    seen_names: set[str] = set()
+    seen_queues: set[str] = set()
+    for index, item in enumerate(payload):
+        if not isinstance(item, dict):
+            raise ValueError(f"Worker pool {index} must be an object")
+        unknown = set(item) - {"name", "queues", "replicas", "concurrency"}
+        if unknown:
+            raise ValueError(
+                f"Worker pool {index} has unsupported keys: "
+                + ", ".join(sorted(unknown))
+            )
+        pool_name = item.get("name")
+        queues = item.get("queues")
+        replicas = item.get("replicas")
+        concurrency = item.get("concurrency")
+        if not isinstance(pool_name, str) or _DNS_LABEL.fullmatch(pool_name) is None:
+            raise ValueError(f"Worker pool {index} has an invalid name")
+        if pool_name in seen_names:
+            raise ValueError(f"Worker pool name is duplicated: {pool_name}")
+        if (
+            not isinstance(queues, list)
+            or not queues
+            or any(
+                not isinstance(queue, str) or _QUEUE_NAME.fullmatch(queue) is None
+                for queue in queues
+            )
+        ):
+            raise ValueError(f"Worker pool {pool_name} has invalid queues")
+        duplicate_queues = seen_queues.intersection(queues)
+        if duplicate_queues:
+            raise ValueError(
+                "Worker queues must have one owning pool: "
+                + ", ".join(sorted(duplicate_queues))
+            )
+        if not isinstance(replicas, int) or not 1 <= replicas <= 128:
+            raise ValueError(
+                f"Worker pool {pool_name} replicas must be between 1 and 128"
+            )
+        if not isinstance(concurrency, int) or not 1 <= concurrency <= 64:
+            raise ValueError(
+                f"Worker pool {pool_name} concurrency must be between 1 and 64"
+            )
+        pools.append(
+            _WorkerPool(
+                name=pool_name,
+                queues=tuple(queues),
+                replicas=replicas,
+                concurrency=concurrency,
+            )
+        )
+        seen_names.add(pool_name)
+        seen_queues.update(queues)
+    if sum(pool.replicas for pool in pools) != spec.replicas.worker:
+        raise ValueError("Worker pool replicas must equal installation replicas.worker")
+    configured_queue_set = set(configured_queues)
+    if seen_queues != configured_queue_set:
+        missing = configured_queue_set - seen_queues
+        unknown = seen_queues - configured_queue_set
+        detail = []
+        if missing:
+            detail.append("missing " + ", ".join(sorted(missing)))
+        if unknown:
+            detail.append("unknown " + ", ".join(sorted(unknown)))
+        raise ValueError(
+            "Worker pools must cover CELERY_QUEUES exactly: " + "; ".join(detail)
+        )
+    return tuple(sorted(pools, key=lambda pool: pool.name))
+
+
+def _database_capacity(
+    spec: InstallationSpec,
+    environment: Mapping[str, str],
+    worker_pools: tuple[_WorkerPool, ...],
+) -> dict[str, int]:
+    limit = _bounded_environment_integer(
+        environment,
+        "GOVOPLAN_DB_CONNECTION_LIMIT",
+        minimum=10,
+        maximum=1_000_000,
+    )
+    reserve = _bounded_environment_integer(
+        environment,
+        "GOVOPLAN_DB_CONNECTION_RESERVE",
+        default=10,
+        minimum=1,
+        maximum=999_999,
+    )
+    if reserve >= limit:
+        raise ValueError("Database connection reserve must be below the limit")
+    api_ceiling = _role_database_ceiling(environment, "API", 5, 2)
+    worker_ceiling = _role_database_ceiling(environment, "WORKER", 1, 1)
+    scheduler_ceiling = _role_database_ceiling(environment, "SCHEDULER", 1, 0)
+    migration_ceiling = _role_database_ceiling(environment, "MIGRATION", 2, 0)
+    worker_processes = sum(
+        pool.replicas * (pool.concurrency + 1) for pool in worker_pools
+    )
+    steady = (
+        spec.replicas.api * api_ceiling
+        + worker_processes * worker_ceiling
+        + (scheduler_ceiling if worker_pools else 0)
+    )
+    rollout_surge = api_ceiling + migration_ceiling
+    if worker_pools:
+        rollout_surge += scheduler_ceiling
+        rollout_surge += sum(
+            (pool.concurrency + 1) * worker_ceiling for pool in worker_pools
+        )
+    peak = steady + rollout_surge
+    available = limit - reserve
+    if peak > available:
+        raise ValueError(
+            "Kubernetes database connection peak exceeds the configured budget: "
+            f"peak={peak}, available={available}, limit={limit}, reserve={reserve}"
+        )
+    return {
+        "limit": limit,
+        "reserve": reserve,
+        "available": available,
+        "steady": steady,
+        "peak": peak,
+    }
+
+
+def _role_database_ceiling(
+    environment: Mapping[str, str],
+    role: str,
+    default_pool_size: int,
+    default_overflow: int,
+) -> int:
+    return _bounded_environment_integer(
+        environment,
+        f"GOVOPLAN_{role}_DB_POOL_SIZE",
+        default=default_pool_size,
+        minimum=1,
+        maximum=100,
+    ) + _bounded_environment_integer(
+        environment,
+        f"GOVOPLAN_{role}_DB_MAX_OVERFLOW",
+        default=default_overflow,
+        minimum=0,
+        maximum=200,
+    )
+
+
+def _role_database_environment(
+    environment: Mapping[str, str],
+    role: str,
+) -> dict[str, str]:
+    defaults = {
+        "API": (5, 2),
+        "WORKER": (1, 1),
+        "SCHEDULER": (1, 0),
+        "MIGRATION": (2, 0),
+    }
+    pool_size, overflow = defaults[role]
+    return {
+        "GOVOPLAN_DB_POOL_SIZE": str(
+            _bounded_environment_integer(
+                environment,
+                f"GOVOPLAN_{role}_DB_POOL_SIZE",
+                default=pool_size,
+                minimum=1,
+                maximum=100,
+            )
+        ),
+        "GOVOPLAN_DB_MAX_OVERFLOW": str(
+            _bounded_environment_integer(
+                environment,
+                f"GOVOPLAN_{role}_DB_MAX_OVERFLOW",
+                default=overflow,
+                minimum=0,
+                maximum=200,
+            )
+        ),
+    }
+
+
+def _bounded_environment_integer(
+    environment: Mapping[str, str],
+    name: str,
+    *,
+    default: int | None = None,
+    minimum: int,
+    maximum: int,
+) -> int:
+    raw = environment.get(name)
+    if raw is None or not str(raw).strip():
+        if default is None:
+            raise ValueError(f"{name} is required")
+        return default
+    try:
+        value = int(str(raw))
+    except ValueError as exc:
+        raise ValueError(f"{name} must be an integer") from exc
+    if not minimum <= value <= maximum:
+        raise ValueError(f"{name} must be between {minimum} and {maximum}")
+    return value
+
+
 def _deployment(
     *,
     name: str,
@@ -315,8 +633,13 @@ def _deployment(
     liveness_path: str | None = None,
     probe_host: str | None = None,
     extra_environment: Mapping[str, str] | None = None,
+    selector_labels: Mapping[str, str] | None = None,
 ) -> dict[str, Any]:
-    role_labels = {**labels, "app.kubernetes.io/component": role}
+    role_labels = {
+        **labels,
+        "app.kubernetes.io/component": role,
+        **dict(selector_labels or {}),
+    }
     environment: list[dict[str, Any]] = [
         {"name": "TMPDIR", "value": "/tmp"},
         {"name": "GOVOPLAN_RUNTIME_ROLE", "value": role},
@@ -400,6 +723,8 @@ def _deployment(
                 "env": [
                     {"name": "TMPDIR", "value": "/tmp"},
                     {"name": "GOVOPLAN_RUNTIME_ROLE", "value": "migration-wait"},
+                    {"name": "GOVOPLAN_DB_POOL_SIZE", "value": "1"},
+                    {"name": "GOVOPLAN_DB_MAX_OVERFLOW", "value": "0"},
                     *_secret_environment(secret_name),
                 ],
                 "securityContext": {
@@ -439,6 +764,7 @@ def _migration_job(
     config_name: str,
     secret_name: str,
     service_account: str,
+    database_environment: Mapping[str, str],
 ) -> dict[str, Any]:
     job_labels = {**labels, "app.kubernetes.io/component": "migration"}
     job_name = _name_with_suffix(name, f"migrate-{release_key}")
@@ -481,6 +807,12 @@ def _migration_job(
                             "env": [
                                 {"name": "TMPDIR", "value": "/tmp"},
                                 {"name": "GOVOPLAN_RUNTIME_ROLE", "value": "migration"},
+                                *(
+                                    {"name": key, "value": value}
+                                    for key, value in sorted(
+                                        database_environment.items()
+                                    )
+                                ),
                                 {
                                     "name": "GOVOPLAN_NODE_ID",
                                     "valueFrom": {
@@ -541,8 +873,14 @@ def _pod_disruption_budget(
     namespace: str,
     labels: dict[str, str],
     role: str,
+    *,
+    selector_labels: Mapping[str, str] | None = None,
 ) -> dict[str, Any]:
-    role_labels = {**labels, "app.kubernetes.io/component": role}
+    role_labels = {
+        **labels,
+        "app.kubernetes.io/component": role,
+        **dict(selector_labels or {}),
+    }
     return {
         "apiVersion": "policy/v1",
         "kind": "PodDisruptionBudget",

tools/inventory/platform-interface-inventory.py

@@ -19,6 +19,18 @@ from typing import Any
 META_ROOT = Path(__file__).resolve().parents[2]
 HTTP_METHODS = {"delete", "get", "head", "options", "patch", "post", "put"}
 PATH_PARAMETER = re.compile(r"\$\{[^{}]*\}|\{[^{}]+\}")
+ENDPOINT_SURFACE_CATEGORIES = {
+    "ui_reachable",
+    "intentionally_headless",
+    "public_integration",
+    "worker_internal",
+    "compatibility",
+    "missing_ui",
+    "removable",
+}
+DEFAULT_ENDPOINT_DECLARATIONS = (
+    META_ROOT / "tools" / "inventory" / "endpoint-surface-declarations.json"
+)
 
 
 def main() -> int:
@@ -31,21 +43,29 @@ def main() -> int:
     parser.add_argument(
         "--strict",
         action="store_true",
-        help="Fail when a used translation key is absent from a generated locale catalog.",
+        help="Fail on missing translations or incomplete endpoint-surface declarations.",
+    )
+    parser.add_argument(
+        "--endpoint-declarations",
+        type=Path,
+        default=DEFAULT_ENDPOINT_DECLARATIONS,
+        help="Versioned endpoint-surface declaration registry.",
     )
     args = parser.parse_args()
 
-    catalog = json.loads(
+    catalog = json.loads((META_ROOT / "repositories.json").read_text(encoding="utf-8"))
-        (META_ROOT / "repositories.json").read_text(encoding="utf-8")
-    )
     workspace_root = Path(catalog["default_parent"]).resolve()
     webui = _extract_webui()
     backend_endpoints = _extract_backend_endpoints(catalog, workspace_root)
     manifests = _extract_manifests(catalog, workspace_root)
+    endpoint_declarations = _load_endpoint_declarations(
+        args.endpoint_declarations.resolve()
+    )
     inventory = _assemble_inventory(
         webui=webui,
         backend_endpoints=backend_endpoints,
         manifests=manifests,
+        endpoint_declarations=endpoint_declarations,
     )
 
     output_dir = args.output_dir.resolve()
@@ -60,9 +80,23 @@ def main() -> int:
     print(f"Platform inventory JSON: {json_path}")
     print(f"Platform inventory summary: {markdown_path}")
 
-    if args.strict and inventory["translation_health"]["missing_catalog_entries"]:
+    if args.strict:
+        failures: list[str] = []
+        if inventory["translation_health"]["missing_catalog_entries"]:
+            failures.append("used translation keys are missing from generated catalogs")
+        if inventory["api"]["unclassified_endpoints"]:
+            failures.append(
+                f"{len(inventory['api']['unclassified_endpoints'])} backend "
+                "endpoints have no WebUI evidence or surface declaration"
+            )
+        if inventory["api"]["stale_endpoint_declarations"]:
+            failures.append(
+                f"{len(inventory['api']['stale_endpoint_declarations'])} "
+                "endpoint declarations do not match a backend endpoint"
+            )
+        if failures:
             print(
-            "Used translation keys are missing from generated catalogs.",
+                "Strict platform inventory failed: " + "; ".join(failures) + ".",
                 file=sys.stderr,
             )
             return 1
@@ -167,7 +201,9 @@ def _endpoint_from_decorator(
     route = _static_string(decorator.args[0])
     if route is None:
         return None
-    owner = decorator.func.value.id if isinstance(decorator.func.value, ast.Name) else ""
+    owner = (
+        decorator.func.value.id if isinstance(decorator.func.value, ast.Name) else ""
+    )
     prefix = prefixes.get(owner, "")
     return {
         "method": method.upper(),
@@ -247,6 +283,7 @@ def _assemble_inventory(
     webui: dict[str, Any],
     backend_endpoints: list[dict[str, Any]],
     manifests: list[dict[str, Any]],
+    endpoint_declarations: dict[tuple[str, str, str], dict[str, Any]],
 ) -> dict[str, Any]:
     frontend_refs = webui["frontendApiReferences"]
     frontend_paths = {
@@ -254,25 +291,61 @@ def _assemble_inventory(
         for reference in frontend_refs
         if canonical_api_path(reference["path"])
     }
+    endpoint_keys = {endpoint_key(endpoint) for endpoint in backend_endpoints}
+    classified_endpoints: list[dict[str, Any]] = []
+    for endpoint in backend_endpoints:
+        key = endpoint_key(endpoint)
+        canonical_path = key[2]
+        static_webui_reference = canonical_path in frontend_paths
+        declaration = endpoint_declarations.get(key)
+        surface = (
+            {
+                "category": "ui_reachable",
+                "rationale": "A canonical API path reference exists in WebUI source.",
+                "source": "static_webui_scan",
+            }
+            if static_webui_reference
+            else (
+                {**declaration, "source": "declaration"}
+                if declaration is not None
+                else None
+            )
+        )
+        classified_endpoints.append(
+            {
+                **endpoint,
+                "canonical_path": canonical_path,
+                "static_webui_reference": static_webui_reference,
+                "surface": surface,
+            }
+        )
     unreferenced = [
         endpoint
-        for endpoint in backend_endpoints
+        for endpoint in classified_endpoints
-        if canonical_api_path(endpoint["path"]) not in frontend_paths
+        if not endpoint["static_webui_reference"]
     ]
+    unclassified = [
+        endpoint for endpoint in unreferenced if endpoint["surface"] is None
+    ]
+    stale_declarations = [
+        declaration
+        for key, declaration in endpoint_declarations.items()
+        if key not in endpoint_keys
+    ]
+    classification_counts = Counter(
+        endpoint["surface"]["category"]
+        for endpoint in classified_endpoints
+        if endpoint["surface"] is not None
+    )
     usages = {item["key"] for item in webui["translationUsages"]}
     catalogs = webui["translationCatalog"]
-    catalog_keys = {
+    catalog_keys = {locale: set(entries) for locale, entries in catalogs.items()}
-        locale: set(entries)
-        for locale, entries in catalogs.items()
-    }
     expected_locales = sorted(catalog_keys)
     missing_catalog_entries = [
         {
             "key": key,
             "missing_locales": [
-                locale
+                locale for locale in expected_locales if key not in catalog_keys[locale]
-                for locale in expected_locales
-                if key not in catalog_keys[locale]
             ],
         }
         for key in sorted(usages)
@@ -311,9 +384,12 @@ def _assemble_inventory(
             "missing_catalog_entries": missing_catalog_entries,
         },
         "api": {
-            "backend_endpoints": backend_endpoints,
+            "backend_endpoints": classified_endpoints,
             "frontend_references": frontend_refs,
             "unreferenced_by_static_webui_scan": unreferenced,
+            "unclassified_endpoints": unclassified,
+            "stale_endpoint_declarations": stale_declarations,
+            "classification_counts": dict(sorted(classification_counts.items())),
         },
         "summary": {
             "modules": len(manifests),
@@ -326,6 +402,8 @@ def _assemble_inventory(
             "backend_endpoints": len(backend_endpoints),
             "frontend_api_references": len(frontend_refs),
             "backend_endpoints_without_static_webui_reference": len(unreferenced),
+            "unclassified_backend_endpoints": len(unclassified),
+            "stale_endpoint_declarations": len(stale_declarations),
         },
     }
 
@@ -340,6 +418,7 @@ def _render_markdown(inventory: dict[str, Any]) -> str:
         item["repository"]
         for item in inventory["api"]["unreferenced_by_static_webui_scan"]
     )
+    classification_counts = inventory["api"]["classification_counts"]
     lines = [
         "# GovOPlaN Platform Interface Inventory",
         "",
@@ -360,6 +439,8 @@ def _render_markdown(inventory: dict[str, Any]) -> str:
             "- Backend endpoints without a static WebUI reference: "
             f"{summary['backend_endpoints_without_static_webui_reference']}"
         ),
+        f"- Unclassified backend endpoints: {summary['unclassified_backend_endpoints']}",
+        f"- Stale endpoint declarations: {summary['stale_endpoint_declarations']}",
         f"- Used translation keys missing from a locale catalog: {len(missing)}",
         "",
         "## Help Review Candidates",
@@ -388,6 +469,19 @@ def _render_markdown(inventory: dict[str, Any]) -> str:
         f"| `{repository}` | {count} |"
         for repository, count in sorted(endpoint_by_repository.items())
     )
+    lines.extend(
+        [
+            "",
+            "## Endpoint Surface Classifications",
+            "",
+            "| Classification | Endpoints |",
+            "| --- | ---: |",
+        ]
+    )
+    lines.extend(
+        f"| `{category}` | {count} |"
+        for category, count in sorted(classification_counts.items())
+    )
     lines.extend(
         [
             "",
@@ -408,11 +502,89 @@ def canonical_api_path(value: str) -> str:
     if "/api/" in path:
         path = path[path.index("/api/") :]
     path = re.sub(r"^/api/v\d+", "", path)
+    path = re.sub(r"(?<=[^/])\$\{[^{}]*\}$", "", path)
     path = PATH_PARAMETER.sub("{}", path)
     path = re.sub(r"/+", "/", path)
     return path.rstrip("/") or "/"
 
 
+def endpoint_key(endpoint: dict[str, Any]) -> tuple[str, str, str]:
+    return (
+        str(endpoint["repository"]),
+        str(endpoint["method"]).upper(),
+        canonical_api_path(str(endpoint["path"])),
+    )
+
+
+def _load_endpoint_declarations(
+    path: Path,
+) -> dict[tuple[str, str, str], dict[str, Any]]:
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise ValueError(
+            f"Endpoint declaration registry does not exist: {path}"
+        ) from exc
+    except json.JSONDecodeError as exc:
+        raise ValueError(
+            f"Endpoint declaration registry is invalid JSON: {exc}"
+        ) from exc
+    if not isinstance(payload, dict) or payload.get("schema_version") != 1:
+        raise ValueError("Endpoint declaration registry must use schema_version 1.")
+    entries = payload.get("endpoints")
+    if not isinstance(entries, list):
+        raise ValueError(
+            "Endpoint declaration registry must contain an endpoints list."
+        )
+    declarations: dict[tuple[str, str, str], dict[str, Any]] = {}
+    for index, entry in enumerate(entries):
+        if not isinstance(entry, dict):
+            raise ValueError(f"Endpoint declaration {index} must be an object.")
+        repository = entry.get("repository")
+        method = entry.get("method")
+        raw_path = entry.get("path")
+        category = entry.get("category")
+        rationale = entry.get("rationale")
+        if not isinstance(repository, str) or not repository.strip():
+            raise ValueError(f"Endpoint declaration {index} has no repository.")
+        if not isinstance(method, str) or method.lower() not in HTTP_METHODS:
+            raise ValueError(f"Endpoint declaration {index} has an invalid method.")
+        if not isinstance(raw_path, str) or not raw_path.startswith("/"):
+            raise ValueError(f"Endpoint declaration {index} has an invalid path.")
+        canonical_path = canonical_api_path(raw_path)
+        if raw_path != canonical_path:
+            raise ValueError(
+                f"Endpoint declaration {index} path must be canonical: {canonical_path}"
+            )
+        if category not in ENDPOINT_SURFACE_CATEGORIES:
+            raise ValueError(f"Endpoint declaration {index} has an invalid category.")
+        if not isinstance(rationale, str) or not rationale.strip():
+            raise ValueError(f"Endpoint declaration {index} has no rationale.")
+        tracking_issue = entry.get("tracking_issue")
+        if category == "missing_ui" and (
+            not isinstance(tracking_issue, str) or not tracking_issue.strip()
+        ):
+            raise ValueError(
+                f"Endpoint declaration {index} requires a tracking_issue for missing_ui."
+            )
+        key = (repository, method.upper(), canonical_path)
+        if key in declarations:
+            raise ValueError(f"Duplicate endpoint declaration: {key!r}.")
+        declarations[key] = {
+            "repository": repository,
+            "method": method.upper(),
+            "path": canonical_path,
+            "category": category,
+            "rationale": rationale.strip(),
+            **(
+                {"tracking_issue": tracking_issue.strip()}
+                if isinstance(tracking_issue, str) and tracking_issue.strip()
+                else {}
+            ),
+        }
+    return declarations
+
+
 def _join_route(prefix: str, route: str) -> str:
     return f"/{prefix.strip('/')}/{route.strip('/')}".replace("//", "/")
 
@@ -444,17 +616,11 @@ def _call_name(node: ast.AST) -> str:
 
 def _plain_value(value: Any) -> Any:
     if is_dataclass(value):
-        return {
+        return {key: _plain_value(item) for key, item in asdict(value).items()}
-            key: _plain_value(item)
-            for key, item in asdict(value).items()
-        }
     if isinstance(value, tuple):
         return [_plain_value(item) for item in value]
     if isinstance(value, dict):
-        return {
+        return {str(key): _plain_value(item) for key, item in value.items()}
-            str(key): _plain_value(item)
-            for key, item in value.items()
-        }
     return value
 
 

tools/release/govoplan_release/publisher.py

@@ -484,31 +484,35 @@ def publish_catalog_candidate(
             )
         )
 
+    result_fields = {
+        "candidate_root": candidate_root,
+        "candidate_catalog": candidate_catalog,
+        "candidate_keyring": candidate_keyring,
+        "resolved_web_root": resolved_web_root,
+        "target_catalog": target_catalog,
+        "target_keyring": target_keyring,
+        "channel": channel,
+        "branch": effective_branch,
+        "tag_name": effective_tag_name,
+        "remote": remote,
+        "validation_valid": validation_valid,
+        "validation_error": validation_error,
+        "validation_warnings": validation_warnings,
+        "candidate_catalog_hash": candidate_catalog_hash,
+        "candidate_keyring_hash": candidate_keyring_hash,
+        "target_catalog_hash_before": target_catalog_hash_before,
+        "target_keyring_hash_before": target_keyring_hash_before,
+        "catalog_changed": catalog_changed,
+        "keyring_changed": keyring_changed,
+    }
+
     if blockers:
         return result(
             status="blocked",
             applied=False,
-            candidate_root=candidate_root,
-            candidate_catalog=candidate_catalog,
-            candidate_keyring=candidate_keyring,
-            resolved_web_root=resolved_web_root,
-            target_catalog=target_catalog,
-            target_keyring=target_keyring,
-            channel=channel,
-            branch=effective_branch,
-            tag_name=effective_tag_name,
-            remote=remote,
-            validation_valid=validation_valid,
-            validation_error=validation_error,
-            validation_warnings=validation_warnings,
-            candidate_catalog_hash=candidate_catalog_hash,
-            candidate_keyring_hash=candidate_keyring_hash,
-            target_catalog_hash_before=target_catalog_hash_before,
-            target_keyring_hash_before=target_keyring_hash_before,
-            catalog_changed=catalog_changed,
-            keyring_changed=keyring_changed,
             steps=tuple(steps),
             notes=tuple([*notes, *blockers]),
+            **result_fields,
         )
 
     if not apply:
@@ -518,27 +522,9 @@ def publish_catalog_candidate(
         return result(
             status="ready",
             applied=False,
-            candidate_root=candidate_root,
-            candidate_catalog=candidate_catalog,
-            candidate_keyring=candidate_keyring,
-            resolved_web_root=resolved_web_root,
-            target_catalog=target_catalog,
-            target_keyring=target_keyring,
-            channel=channel,
-            branch=effective_branch,
-            tag_name=effective_tag_name,
-            remote=remote,
-            validation_valid=validation_valid,
-            validation_error=validation_error,
-            validation_warnings=validation_warnings,
-            candidate_catalog_hash=candidate_catalog_hash,
-            candidate_keyring_hash=candidate_keyring_hash,
-            target_catalog_hash_before=target_catalog_hash_before,
-            target_keyring_hash_before=target_keyring_hash_before,
-            catalog_changed=catalog_changed,
-            keyring_changed=keyring_changed,
             steps=tuple(steps),
             notes=tuple(notes),
+            **result_fields,
         )
 
     directory_payloads = module_directory_payloads(
@@ -615,11 +601,7 @@ def publish_catalog_candidate(
             publication_commit,
         )
         _seal_git_metadata_file(
-            resolved_web_root
+            resolved_web_root / ".git" / "refs" / "tags" / effective_tag_name,
-            / ".git"
-            / "refs"
-            / "tags"
-            / effective_tag_name,
             label="website publication tag reference",
         )
         publication_tag_object = git_text(
@@ -704,30 +686,12 @@ def publish_catalog_candidate(
     return result(
         status="published" if push else "applied",
         applied=True,
-        candidate_root=candidate_root,
-        candidate_catalog=candidate_catalog,
-        candidate_keyring=candidate_keyring,
-        resolved_web_root=resolved_web_root,
-        target_catalog=target_catalog,
-        target_keyring=target_keyring,
-        channel=channel,
-        branch=effective_branch,
-        tag_name=effective_tag_name,
-        remote=remote,
-        validation_valid=validation_valid,
-        validation_error=validation_error,
-        validation_warnings=validation_warnings,
-        candidate_catalog_hash=candidate_catalog_hash,
-        candidate_keyring_hash=candidate_keyring_hash,
-        target_catalog_hash_before=target_catalog_hash_before,
-        target_keyring_hash_before=target_keyring_hash_before,
-        catalog_changed=catalog_changed,
-        keyring_changed=keyring_changed,
         publication_commit_sha=publication_commit,
         publication_tag_object_sha=publication_tag_object,
         publication_tag_commit_sha=publication_tag_commit,
         steps=tuple(completed_steps),
         notes=tuple(notes),
+        **result_fields,
     )