feat: add bounded Kubernetes runtime verification

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.

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(

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",