fix(api): delegate production bootstrap ddl

apps/api/src/core/config.py

@@ -302,7 +302,7 @@ class Settings(BaseSettings):
         default=True,
         description=(
             "Run idempotent schema bootstrap DDL during process startup. "
-            "Production workers disable this after CD migration preflight."
+            "Production workloads disable this after CD migration preflight."
         ),
     )
 

apps/api/src/main.py

@@ -155,6 +155,25 @@ def _close_unscheduled_coroutine(coro: Coroutine[Any, Any, Any]) -> None:
         logger.warning("api_background_task_close_failed", error=str(exc))
 
 
+async def initialize_api_database() -> bool:
+    """Run local/dev bootstrap DDL or emit the production ownership receipt."""
+    if not settings.DATABASE_BOOTSTRAP_DDL_ENABLED:
+        logger.info(
+            "database_bootstrap_ddl_skipped",
+            owner="cd_migration_preflight",
+            workload="api",
+            reason="database_bootstrap_ddl_disabled",
+        )
+        return False
+
+    await init_db()
+    db_url = settings.DATABASE_URL
+    logger.info(
+        "database_initialized", url=db_url.split("@")[-1] if "@" in db_url else db_url
+    )
+    return True
+
+
 def _resolve_request_project_context(request: Request) -> tuple[str | None, str]:
     """Resolve tenant context for RLS while keeping non-webhook routes fail-closed."""
     for candidate in (
@@ -293,12 +312,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
             return None
         return asyncio.create_task(coro, name=task_name)
 
-    # CTO-201: Initialize PostgreSQL database (統帥鐵律: 禁止 SQLite)
-    await init_db()
-    db_url = settings.DATABASE_URL
-    logger.info(
-        "database_initialized", url=db_url.split("@")[-1] if "@" in db_url else db_url
-    )
+    # CTO-201: production DDL belongs to CD migration preflight; local/dev keeps
+    # the default bootstrap path (統帥鐵律: 禁止 SQLite).
+    await initialize_api_database()
 
     # Phase 5: Initialize HTTP Clients (ClickHouse, Ollama)
     # 統帥鐵律: 連線池在啟動時建立，關閉時回收

apps/api/tests/test_runtime_bootstrap_guards.py

@@ -84,6 +84,88 @@ def test_production_worker_delegates_bootstrap_ddl_to_cd() -> None:
     assert 'owner="cd_migration_preflight"' in source
 
 
+def test_production_api_delegates_bootstrap_ddl_to_cd() -> None:
+    from src import main as api_main
+
+    root = Path(__file__).resolve().parents[3]
+    documents = yaml.safe_load_all(
+        (root / "k8s/awoooi-prod/06-deployment-api.yaml").read_text(encoding="utf-8")
+    )
+    manifest = next(
+        document for document in documents if document.get("kind") == "Deployment"
+    )
+    container = manifest["spec"]["template"]["spec"]["containers"][0]
+    env = {item["name"]: item.get("value") for item in container["env"]}
+    source = inspect.getsource(api_main.initialize_api_database)
+
+    assert (
+        type(api_main.settings).model_fields["DATABASE_BOOTSTRAP_DDL_ENABLED"].default
+        is True
+    )
+    assert env["DATABASE_BOOTSTRAP_DDL_ENABLED"] == "false"
+    assert "if not settings.DATABASE_BOOTSTRAP_DDL_ENABLED" in source
+    assert 'owner="cd_migration_preflight"' in source
+    assert 'workload="api"' in source
+
+
+@pytest.mark.asyncio
+async def test_api_database_bootstrap_runs_when_enabled(monkeypatch) -> None:
+    from src import main as api_main
+
+    calls: list[str] = []
+    logs: list[tuple[str, dict[str, object]]] = []
+
+    async def fake_init_db() -> None:
+        calls.append("init_db")
+
+    class _FakeLogger:
+        def info(self, event: str, **fields: object) -> None:
+            logs.append((event, fields))
+
+    monkeypatch.setattr(api_main.settings, "DATABASE_BOOTSTRAP_DDL_ENABLED", True)
+    monkeypatch.setattr(
+        api_main.settings,
+        "DATABASE_URL",
+        "postgresql+asyncpg://test:test@127.0.0.1/test",
+    )
+    monkeypatch.setattr(api_main, "init_db", fake_init_db)
+    monkeypatch.setattr(api_main, "logger", _FakeLogger())
+
+    assert await api_main.initialize_api_database() is True
+    assert calls == ["init_db"]
+    assert logs == [("database_initialized", {"url": "127.0.0.1/test"})]
+
+
+@pytest.mark.asyncio
+async def test_api_database_bootstrap_skips_with_cd_owner_receipt(monkeypatch) -> None:
+    from src import main as api_main
+
+    logs: list[tuple[str, dict[str, object]]] = []
+
+    async def fail_if_called() -> None:
+        raise AssertionError("production API must not execute bootstrap DDL")
+
+    class _FakeLogger:
+        def info(self, event: str, **fields: object) -> None:
+            logs.append((event, fields))
+
+    monkeypatch.setattr(api_main.settings, "DATABASE_BOOTSTRAP_DDL_ENABLED", False)
+    monkeypatch.setattr(api_main, "init_db", fail_if_called)
+    monkeypatch.setattr(api_main, "logger", _FakeLogger())
+
+    assert await api_main.initialize_api_database() is False
+    assert logs == [
+        (
+            "database_bootstrap_ddl_skipped",
+            {
+                "owner": "cd_migration_preflight",
+                "workload": "api",
+                "reason": "database_bootstrap_ddl_disabled",
+            },
+        )
+    ]
+
+
 def test_get_engine_uses_database_pool_budget(monkeypatch):
     from src.db import base as db_base
 

k8s/awoooi-prod/06-deployment-api.yaml

@@ -14,11 +14,14 @@ metadata:
     system: awoooi
     environment: prod
 spec:
-  # 2026-07-01 Codex: temporary post-reboot recovery cap. The production
-  # database role `awoooi` currently has CONNECTION LIMIT 2, so a two-replica
-  # API rollout deadlocks when the old pod still holds DB connections. Restore
-  # replicas=2 only after the DB role limit / pool budget is raised and verified.
-  replicas: 1
+  # 2026-07-14 Codex: keep the bounded two-replica serving topology only with
+  # startup DDL delegated to CD migration preflight below. The initial canary
+  # briefly reached 2/2 Ready and 200m/238m CPU, then failed when both API pods
+  # re-entered init_db and saturated at the 1 CPU limit. The independent
+  # verifier therefore requires the DDL gate, DB CONNECTION LIMIT 12 receipt,
+  # 2/2 Ready, zero restart delta, and unsaturated CPU for a full probe window.
+  # Roll back to replicas=1 if any post-deploy condition fails.
+  replicas: 2
   revisionHistoryLimit: 3
   selector:
     matchLabels:
@@ -174,6 +177,11 @@ spec:
               # 2026-07-11 Codex: release each API connection after use while
               # workload-specific DB identities are being split and verified.
               value: "true"
+            - name: DATABASE_BOOTSTRAP_DDL_ENABLED
+              # 2026-07-14 Codex: production schema DDL has one owner. CD
+              # migration preflight applies/verifies it before API rollout;
+              # API startup emits an explicit no-write ownership receipt.
+              value: "false"
             - name: AWOOOI_API_BACKGROUND_TASKS_ENABLED
               value: "false"
             - name: IWOOOS_WAZUH_READONLY_ENABLED

ops/runner/test_cd_controlled_runtime_profile.py

@@ -124,6 +124,13 @@ def test_prod_db_pool_budget_and_non_overlap_rollouts_are_source_controlled() ->
     assert "value: \"1\"" in api
     assert "DATABASE_MAX_OVERFLOW" in api
     assert "value: \"0\"" in api
+    assert "DATABASE_BOOTSTRAP_DDL_ENABLED" in api
+    assert "CD migration preflight" in api
+    assert "replicas: 2" in api
+    assert "CONNECTION" in api and "LIMIT 12" in api
+    assert "Roll back to replicas=1" in api
+    assert "maxSurge: 0" in api
+    assert "maxUnavailable: 1" in api
 
     assert "DATABASE_POOL_SIZE" in worker
     assert "DATABASE_MAX_OVERFLOW" in worker

ops/runner/test_verify_awoooi_api_liveness_stabilization.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPT = ROOT / "ops" / "runner" / "verify_awoooi_api_liveness_stabilization.py"
+SPEC = importlib.util.spec_from_file_location("api_liveness_verifier", SCRIPT)
+assert SPEC and SPEC.loader
+verifier = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(verifier)
+
+
+def _live_deployment() -> dict:
+    deployment = verifier._source_deployment()
+    deployment["status"] = {
+        "replicas": 2,
+        "updatedReplicas": 2,
+        "availableReplicas": 2,
+        "readyReplicas": 2,
+    }
+    return deployment
+
+
+def _receipt() -> dict[str, str]:
+    return {
+        "connection_budget_verified": "true",
+        "api_database_null_pool": "true",
+        "api_representative_db_preflight_passed": "true",
+        "api_http_concurrency_probe_passed": "true",
+        "api_db_connection_limit": "12",
+        "api_required_connection_budget": "8",
+        "api_available_connection_headroom": "11",
+    }
+
+
+def _pods() -> list[dict]:
+    return [
+        {
+            "metadata": {"name": name},
+            "status": {"containerStatuses": [{"name": "api", "ready": True}]},
+        }
+        for name in ("awoooi-api-a", "awoooi-api-b")
+    ]
+
+
+def test_source_contract_is_bounded_and_rollback_ready() -> None:
+    deployment = verifier._source_deployment()
+    assert verifier.evaluate_source_contract(deployment) == []
+    edge_config = (verifier.ROOT / "ops/nginx/awoooi.wooo.work.conf").read_text()
+    assert verifier.evaluate_routing_contract(
+        verifier._source_service(), edge_config
+    ) == []
+    assert verifier.ROLLBACK_COMMAND.endswith("--replicas=1")
+
+
+def test_live_contract_accepts_two_ready_unsaturated_stable_pods() -> None:
+    pods = _pods()
+    restarts = {pod["metadata"]["name"]: 0 for pod in pods}
+    cpu = {pod["metadata"]["name"]: 410 for pod in pods}
+    assert verifier.evaluate_live_state(
+        _live_deployment(), _receipt(), pods, restarts, restarts, cpu, 900
+    ) == []
+
+
+def test_live_contract_fails_closed_on_budget_restart_or_cpu_regression() -> None:
+    pods = _pods()
+    receipt = _receipt()
+    receipt["connection_budget_verified"] = "false"
+    failures = verifier.evaluate_live_state(
+        _live_deployment(),
+        receipt,
+        pods,
+        {"awoooi-api-a": 0, "awoooi-api-b": 0},
+        {"awoooi-api-a": 1, "awoooi-api-b": 0},
+        {"awoooi-api-a": 1000, "awoooi-api-b": 300},
+        900,
+    )
+    assert "receipt_connection_budget_verified_not_verified" in failures
+    assert "live_restart_delta:1" in failures
+    assert "live_cpu_saturated:awoooi-api-a:1000m" in failures
+
+
+def test_edge_receipt_requires_a_qualified_production_vantage() -> None:
+    assert verifier.evaluate_edge_receipt(
+        "192.168.0.120",
+        {"vip125": 200, "node120": 200, "node121": 200},
+    ) == []
+    assert verifier.evaluate_edge_receipt(
+        None,
+        {"vip125": None, "node120": None, "node121": None},
+    ) == ["edge_production_vantage_unavailable"]
+
+
+def test_local_vip_failure_is_not_used_as_production_edge_receipt() -> None:
+    qualified = {"vip125": 200, "node120": 200, "node121": 200}
+    local_unqualified_status = None
+    assert local_unqualified_status is None
+    assert verifier.evaluate_edge_receipt("192.168.0.110", qualified) == []

ops/runner/verify_awoooi_api_liveness_stabilization.py

@@ -0,0 +1,383 @@
+#!/usr/bin/env python3
+"""Fail-closed verifier for the bounded AWOOOI API two-replica stabilization."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[2]
+MANIFEST = ROOT / "k8s" / "awoooi-prod" / "06-deployment-api.yaml"
+NAMESPACE = "awoooi-prod"
+DEPLOYMENT = "awoooi-api"
+BOUNDARY_RECEIPT = "awoooi-executor-boundary-verification"
+PUBLIC_LIVENESS_URL = "https://awoooi.wooo.work/api/v1/health/live"
+EDGE_VIP_LIVENESS_URL = "http://192.168.0.125:32334/api/v1/health/live"
+EDGE_LIVENESS_URLS = {
+    "vip125": EDGE_VIP_LIVENESS_URL,
+    "node120": "http://192.168.0.120:32334/api/v1/health/live",
+    "node121": "http://192.168.0.121:32334/api/v1/health/live",
+}
+EDGE_VANTAGE_HOSTS = ("192.168.0.120", "192.168.0.110")
+ROLLBACK_COMMAND = (
+    "kubectl -n awoooi-prod scale deployment/awoooi-api --replicas=1"
+)
+
+
+def _source_documents() -> list[dict[str, Any]]:
+    """Render the manifest with kubectl's client parser; never contact the API."""
+    completed = subprocess.run(
+        ["kubectl", "create", "--dry-run=client", "-f", str(MANIFEST), "-o", "json"],
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=20,
+    )
+    decoder = json.JSONDecoder()
+    remaining = completed.stdout
+    documents: list[dict[str, Any]] = []
+    while remaining.strip():
+        document, offset = decoder.raw_decode(remaining.lstrip())
+        remaining = remaining.lstrip()[offset:]
+        documents.append(document)
+    return documents
+
+
+def _source_deployment() -> dict[str, Any]:
+    return next(
+        document
+        for document in _source_documents()
+        if document.get("kind") == "Deployment"
+    )
+
+
+def _source_service() -> dict[str, Any]:
+    return next(
+        document
+        for document in _source_documents()
+        if document.get("kind") == "Service"
+    )
+
+
+def _env(container: dict[str, Any]) -> dict[str, str | None]:
+    return {entry["name"]: entry.get("value") for entry in container.get("env", [])}
+
+
+def evaluate_source_contract(deployment: dict[str, Any]) -> list[str]:
+    failures: list[str] = []
+    spec = deployment.get("spec") or {}
+    strategy = (spec.get("strategy") or {}).get("rollingUpdate") or {}
+    container = spec["template"]["spec"]["containers"][0]
+    env = _env(container)
+
+    expectations = {
+        "replicas": (spec.get("replicas"), 2),
+        "max_surge": (strategy.get("maxSurge"), 0),
+        "max_unavailable": (strategy.get("maxUnavailable"), 1),
+        "cpu_limit": ((container.get("resources") or {}).get("limits", {}).get("cpu"), "1"),
+        "memory_limit": ((container.get("resources") or {}).get("limits", {}).get("memory"), "1Gi"),
+        "database_pool_size": (env.get("DATABASE_POOL_SIZE"), "1"),
+        "database_max_overflow": (env.get("DATABASE_MAX_OVERFLOW"), "0"),
+        "database_null_pool": (env.get("DATABASE_NULL_POOL"), "true"),
+        "database_bootstrap_ddl_enabled": (
+            env.get("DATABASE_BOOTSTRAP_DDL_ENABLED"),
+            "false",
+        ),
+        "api_background_tasks": (env.get("AWOOOI_API_BACKGROUND_TASKS_ENABLED"), "false"),
+    }
+    for name, (actual, expected) in expectations.items():
+        if actual != expected:
+            failures.append(f"source_{name}_mismatch:{actual!r}!={expected!r}")
+
+    probes = {
+        name: container.get(name) or {}
+        for name in ("startupProbe", "livenessProbe", "readinessProbe")
+    }
+    for name, probe in probes.items():
+        expected_path = (
+            "/api/v1/health/ready"
+            if name == "readinessProbe"
+            else "/api/v1/health/live"
+        )
+        if (probe.get("httpGet") or {}).get("path") != expected_path:
+            failures.append(f"source_{name}_not_lightweight")
+    return failures
+
+
+def evaluate_routing_contract(service: dict[str, Any], edge_config: str) -> list[str]:
+    failures: list[str] = []
+    spec = service.get("spec") or {}
+    ports = spec.get("ports") or []
+    api_port = next((item for item in ports if item.get("name") == "http"), {})
+    if service.get("metadata", {}).get("name") != "awoooi-api-svc":
+        failures.append("source_api_service_name_mismatch")
+    if spec.get("type") != "NodePort":
+        failures.append("source_api_service_not_nodeport")
+    if api_port.get("port") != 8000 or api_port.get("targetPort") != 8000:
+        failures.append("source_api_service_port_mismatch")
+    if api_port.get("nodePort") != 32334:
+        failures.append("source_api_nodeport_mismatch")
+    if "server 192.168.0.125:32334;" not in edge_config:
+        failures.append("source_edge_vip_upstream_mismatch")
+    return failures
+
+
+def evaluate_live_state(
+    deployment: dict[str, Any],
+    receipt: dict[str, str],
+    pods: list[dict[str, Any]],
+    restart_before: dict[str, int],
+    restart_after: dict[str, int],
+    cpu_millicores: dict[str, int],
+    max_cpu_millicores: int,
+) -> list[str]:
+    failures = evaluate_source_contract(deployment)
+    status = deployment.get("status") or {}
+    for name, actual in {
+        "desired_replicas": status.get("replicas", 0),
+        "updated_replicas": status.get("updatedReplicas", 0),
+        "available_replicas": status.get("availableReplicas", 0),
+        "ready_replicas": status.get("readyReplicas", 0),
+    }.items():
+        if actual != 2:
+            failures.append(f"live_{name}_not_two:{actual}")
+    if status.get("unavailableReplicas", 0) not in (0, None):
+        failures.append(f"live_unavailable_replicas:{status['unavailableReplicas']}")
+
+    expected_receipt = {
+        "connection_budget_verified": "true",
+        "api_database_null_pool": "true",
+        "api_representative_db_preflight_passed": "true",
+        "api_http_concurrency_probe_passed": "true",
+    }
+    for key, expected in expected_receipt.items():
+        if receipt.get(key) != expected:
+            failures.append(f"receipt_{key}_not_verified")
+    for key, minimum in {
+        "api_db_connection_limit": 12,
+        "api_required_connection_budget": 8,
+        "api_available_connection_headroom": 8,
+    }.items():
+        try:
+            actual = int(receipt.get(key, ""))
+        except ValueError:
+            actual = -1
+        if actual < minimum:
+            failures.append(f"receipt_{key}_below_{minimum}:{actual}")
+
+    ready_pods = 0
+    for pod in pods:
+        statuses = (pod.get("status") or {}).get("containerStatuses") or []
+        api = next((item for item in statuses if item.get("name") == "api"), None)
+        if api and api.get("ready") is True:
+            ready_pods += 1
+    if ready_pods != 2:
+        failures.append(f"live_ready_pod_count_not_two:{ready_pods}")
+
+    all_names = set(restart_before) | set(restart_after)
+    restart_delta = sum(
+        max(0, restart_after.get(name, 0) - restart_before.get(name, 0))
+        for name in all_names
+    )
+    if restart_delta:
+        failures.append(f"live_restart_delta:{restart_delta}")
+    if set(cpu_millicores) != {pod["metadata"]["name"] for pod in pods}:
+        failures.append("live_cpu_sample_incomplete")
+    for pod_name, cpu in cpu_millicores.items():
+        if cpu >= max_cpu_millicores:
+            failures.append(f"live_cpu_saturated:{pod_name}:{cpu}m")
+    return failures
+
+
+def _kubectl_json(*args: str) -> dict[str, Any]:
+    completed = subprocess.run(
+        ["kubectl", *args],
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=20,
+    )
+    return json.loads(completed.stdout)
+
+
+def _restart_counts(pods: list[dict[str, Any]]) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for pod in pods:
+        statuses = (pod.get("status") or {}).get("containerStatuses") or []
+        api = next((item for item in statuses if item.get("name") == "api"), None)
+        counts[pod["metadata"]["name"]] = int((api or {}).get("restartCount", 0))
+    return counts
+
+
+def _cpu_samples() -> dict[str, int]:
+    completed = subprocess.run(
+        [
+            "kubectl", "-n", NAMESPACE, "top", "pod", "-l", f"app={DEPLOYMENT}",
+            "--containers", "--no-headers",
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=20,
+    )
+    samples: dict[str, int] = {}
+    for line in completed.stdout.splitlines():
+        fields = line.split()
+        if len(fields) >= 3 and fields[1] == "api" and fields[2].endswith("m"):
+            samples[fields[0]] = int(fields[2][:-1])
+    return samples
+
+
+def _http_status(url: str) -> int | None:
+    try:
+        with urllib.request.urlopen(url, timeout=8) as response:
+            return int(response.status)
+    except urllib.error.HTTPError as exc:
+        return int(exc.code)
+    except Exception:
+        return None
+
+
+def evaluate_edge_receipt(
+    vantage_host: str | None,
+    statuses: dict[str, int | None],
+) -> list[str]:
+    if not vantage_host:
+        return ["edge_production_vantage_unavailable"]
+    failures: list[str] = []
+    for endpoint in EDGE_LIVENESS_URLS:
+        if statuses.get(endpoint) != 200:
+            failures.append(
+                f"edge_{endpoint}_liveness_unavailable:{statuses.get(endpoint)}"
+            )
+    return failures
+
+
+def _edge_status_from_production_vantage() -> tuple[str | None, dict[str, int | None]]:
+    for host in EDGE_VANTAGE_HOSTS:
+        statuses: dict[str, int | None] = {}
+        vantage_reachable = True
+        for endpoint, url in EDGE_LIVENESS_URLS.items():
+            completed = subprocess.run(
+                [
+                    "ssh",
+                    "-o", "BatchMode=yes",
+                    "-o", "ConnectTimeout=5",
+                    "-o", "ConnectionAttempts=1",
+                    "-o", "ServerAliveInterval=3",
+                    "-o", "ServerAliveCountMax=1",
+                    f"wooo@{host}",
+                    (
+                        "curl -sS -o /dev/null -w '%{http_code}' "
+                        f"--connect-timeout 3 --max-time 6 {url}"
+                    ),
+                ],
+                check=False,
+                capture_output=True,
+                text=True,
+                timeout=12,
+            )
+            if completed.returncode == 255:
+                vantage_reachable = False
+                break
+            try:
+                statuses[endpoint] = int(completed.stdout.strip())
+            except ValueError:
+                statuses[endpoint] = None
+        if vantage_reachable:
+            return host, statuses
+    return None, {}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--live", action="store_true")
+    parser.add_argument("--sample-seconds", type=int, default=35)
+    parser.add_argument("--max-pod-cpu-millicores", type=int, default=900)
+    args = parser.parse_args()
+
+    source = _source_deployment()
+    failures = evaluate_source_contract(source)
+    edge_config = (ROOT / "ops" / "nginx" / "awoooi.wooo.work.conf").read_text(
+        encoding="utf-8"
+    )
+    failures.extend(evaluate_routing_contract(_source_service(), edge_config))
+    result: dict[str, Any] = {
+        "schema_version": "awoooi_api_liveness_stabilization_verifier_v1",
+        "mode": "live" if args.live else "source",
+        "rollback": {"target_replicas": 1, "command": ROLLBACK_COMMAND},
+        "routing_truth": {
+            "kubernetes_service": "awoooi-api-svc:8000",
+            "node_port": 32334,
+            "edge_vip": "192.168.0.125:32334",
+            "public_liveness": PUBLIC_LIVENESS_URL,
+            "edge_probe_vantage": "production_host_120_then_110",
+            "local_edge_probe_classification": "vantage_unqualified",
+            "edge_reachability_is_separate_from_replica_rollback": True,
+        },
+    }
+    try:
+        if args.live:
+            deployment_before = _kubectl_json(
+                "-n", NAMESPACE, "get", "deployment", DEPLOYMENT, "-o", "json"
+            )
+            receipt = _kubectl_json(
+                "-n", NAMESPACE, "get", "configmap", BOUNDARY_RECEIPT, "-o", "json"
+            ).get("data") or {}
+            pods_before = _kubectl_json(
+                "-n", NAMESPACE, "get", "pods", "-l", f"app={DEPLOYMENT}", "-o", "json"
+            ).get("items") or []
+            restart_before = _restart_counts(pods_before)
+            time.sleep(max(0, args.sample_seconds))
+            deployment_after = _kubectl_json(
+                "-n", NAMESPACE, "get", "deployment", DEPLOYMENT, "-o", "json"
+            )
+            pods_after = _kubectl_json(
+                "-n", NAMESPACE, "get", "pods", "-l", f"app={DEPLOYMENT}", "-o", "json"
+            ).get("items") or []
+            failures = evaluate_live_state(
+                deployment_after,
+                receipt,
+                pods_after,
+                restart_before,
+                _restart_counts(pods_after),
+                _cpu_samples(),
+                args.max_pod_cpu_millicores,
+            )
+            result["sample_seconds"] = max(0, args.sample_seconds)
+            desired_before = (deployment_before.get("spec") or {}).get("replicas")
+            desired_after = (deployment_after.get("spec") or {}).get("replicas")
+            result["desired_replicas_before"] = desired_before
+            result["desired_replicas_after"] = desired_after
+            if desired_before != desired_after:
+                failures.append(
+                    f"live_desired_replica_drift:{desired_before}->{desired_after}"
+                )
+            public_status = _http_status(PUBLIC_LIVENESS_URL)
+            local_edge_vip_status = _http_status(EDGE_VIP_LIVENESS_URL)
+            edge_vantage, edge_statuses = _edge_status_from_production_vantage()
+            result["routing_truth"]["public_status"] = public_status
+            result["routing_truth"]["local_edge_vip_status"] = local_edge_vip_status
+            result["routing_truth"]["qualified_vantage_host"] = edge_vantage
+            result["routing_truth"]["qualified_edge_statuses"] = edge_statuses
+            if public_status != 200:
+                failures.append(f"public_liveness_unavailable:{public_status}")
+            failures.extend(evaluate_edge_receipt(edge_vantage, edge_statuses))
+    except Exception as exc:  # fail closed without leaking command output
+        failures.append(f"verifier_error:{type(exc).__name__}")
+
+    result["status"] = "verified" if not failures else "blocked_with_safe_rollback"
+    result["failures"] = failures
+    print(json.dumps(result, ensure_ascii=False, sort_keys=True))
+    return 0 if not failures else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())