fix(cd): verify used database connection budget correctly

.gitea/workflows/cd.yaml

@@ -3907,7 +3907,7 @@ jobs:
                       and connection_limit >= required_connection_budget
                       and (
                           available_connection_headroom
-                          >= required_connection_budget
+                          >= representative_probe_count
                       )
                   ),
                   "table_privilege_gap_count": table_gap_count,

apps/api/src/services/executor_trust_boundary_readback.py

@@ -215,8 +215,14 @@ def _build_database_identity_boundary(
             workloads[workload_id]["connection_limit"]
             - workloads[workload_id]["active_role_connection_count"]
         )
+        and isinstance(
+            workloads[workload_id]["representative_connection_probe_count"],
+            int,
+        )
+        and workloads[workload_id]["representative_connection_probe_count"]
+        >= 1
         and workloads[workload_id]["available_connection_headroom"]
-        >= workloads[workload_id]["required_connection_budget"]
+        >= workloads[workload_id]["representative_connection_probe_count"]
         for workload_id in _DB_WORKLOAD_IDS
     )
     verifier_verified = receipt.get("db_identity_verifier") == _DB_IDENTITY_VERIFIER
@@ -315,7 +321,12 @@ def _build_database_identity_boundary(
             or active_role_connections < 0
             or not isinstance(available_headroom, int)
             or available_headroom != connection_limit - active_role_connections
-            or available_headroom < required_budget
+            or not isinstance(
+                workload["representative_connection_probe_count"], int
+            )
+            or workload["representative_connection_probe_count"] < 1
+            or available_headroom
+            < workload["representative_connection_probe_count"]
         ):
             blockers.append(f"{workload_id}_connection_budget_not_verified")
     if not db_identity_isolated:

apps/api/tests/test_executor_trust_boundary_readback.py

@@ -345,8 +345,8 @@ def test_executor_trust_boundary_requires_broker_secret_isolation() -> None:
 
 def test_executor_trust_boundary_requires_live_connection_headroom() -> None:
     receipt = _verified_receipt()
-    receipt["worker_active_role_connection_count"] = "7"
-    receipt["worker_available_connection_headroom"] = "1"
+    receipt["worker_active_role_connection_count"] = "8"
+    receipt["worker_available_connection_headroom"] = "0"
 
     readback = build_executor_trust_boundary_readback(
         receipt,
@@ -361,6 +361,24 @@ def test_executor_trust_boundary_requires_live_connection_headroom() -> None:
     )
 
 
+def test_executor_trust_boundary_does_not_double_count_used_budget() -> None:
+    receipt = _verified_receipt()
+    receipt["worker_active_role_connection_count"] = "4"
+    receipt["worker_available_connection_headroom"] = "4"
+
+    readback = build_executor_trust_boundary_readback(
+        receipt,
+        deployed_source_sha="abc123",
+    )
+
+    database_boundary = readback["database_identity_boundary"]
+    assert readback["full_workload_boundary_verified"] is True
+    assert database_boundary["connection_budget_verified"] is True
+    assert "worker_connection_budget_not_verified" not in (
+        database_boundary["active_blockers"]
+    )
+
+
 def test_executor_trust_boundary_binds_budget_and_verifier_contract() -> None:
     wrong_budget = _verified_receipt()
     wrong_budget["worker_required_connection_budget"] = "4"

apps/api/tests/test_workload_database_identity_projection.py

@@ -234,12 +234,12 @@ def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None:
     assert "representative_connection_probe_count" in script
     assert "active_role_connection_count" in script
     assert "available_connection_headroom" in script
-    assert "connection_headroom_below_required" in script
+    assert "connection_headroom_below_probe_requirement" in script
     assert "FROM pg_stat_activity" in script
     assert "pid <> pg_backend_pid()" in script
     assert "for preflight_attempt in 1 2 3" in script
     assert "preflight_attempts_exhausted=3" in script
-    assert '"connection_headroom_below_required"' in script
+    assert '"connection_headroom_below_probe_requirement"' in script
     assert 'error_code = "workload_database_preflight_failed"' in script
     assert '"error_code": error_code' in script
     assert 'text("SELECT pg_sleep(0.25)")' in script
@@ -261,6 +261,7 @@ def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None:
 
 def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
     workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
+    normalized_workflow = " ".join(workflow.split())
 
     assert "WORKLOAD_DB_ROLLBACK_FILE" in workflow
     assert "WORKLOAD_DB_SECRET_PREEXISTENCE_FILE" in workflow
@@ -268,6 +269,14 @@ def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
     assert "bash -s -- apply" in workflow
     assert "bash -s -- verify" in workflow
     assert workflow.index("bash -s -- apply") < workflow.index("bash -s -- verify")
+    assert (
+        "available_connection_headroom >= representative_probe_count"
+        in normalized_workflow
+    )
+    assert (
+        "available_connection_headroom >= required_connection_budget"
+        not in normalized_workflow
+    )
     assert '"spec": item["spec"]' in workflow
     assert '"data": item' not in workflow
     assert "get secret '${secret_name}' -n awoooi-prod -o jsonpath" not in workflow

scripts/ops/awoooi-workload-db-identity-bootstrap.sh

@@ -382,8 +382,10 @@ spec:
                   raise RuntimeError("connection_limit_mismatch")
               if observed_limit < required_budget:
                   raise RuntimeError("connection_budget_below_required")
-              if available_connection_headroom < required_budget:
-                  raise RuntimeError("connection_headroom_below_required")
+              if available_connection_headroom < representative_probe_count:
+                  raise RuntimeError(
+                      "connection_headroom_below_probe_requirement"
+                  )
               if table_gap_count or sequence_gap_count:
                   raise RuntimeError("database_privilege_gap")
               print(json.dumps({
@@ -414,7 +416,7 @@ spec:
                   "representative_probe_count_invalid",
                   "connection_limit_mismatch",
                   "connection_budget_below_required",
-                  "connection_headroom_below_required",
+                  "connection_headroom_below_probe_requirement",
                   "database_privilege_gap",
               }
               error_code = str(exc)