fix(sre): fail closed DB verifier runtime drift

apps/api/src/services/db_bounded_executor.py

@@ -335,6 +335,16 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]:
     drift = bool(
         index_count > 1 or (index_exists and not index_definition_exact_except_validity)
     )
+    verify_ready = bool(
+        row.get("table_exists") is True
+        and row.get("columns_complete") is True
+        and row.get("table_kind_supported") is True
+        and row.get("primary_database") is True
+        and row.get("schema_usage") is True
+        and row.get("table_select") is True
+        and index_shape_exact
+        and not drift
+    )
     apply_ready = bool(
         base_ready
         and not drift
@@ -384,6 +394,7 @@ def evaluate_catalog_snapshot(row: Mapping[str, Any]) -> dict[str, Any]:
         "reindex_ready": reindex_ready,
         "catalog_drift": drift,
         "apply_ready": apply_ready,
+        "verify_ready": verify_ready,
     }
 
 
@@ -522,7 +533,7 @@ async def execute_bounded_db_command(
                     error_code=None if terminal == "check_ready" else "precheck_failed",
                 )
             if mode == "verify":
-                verified = before["index_shape_exact"] and not before["catalog_drift"]
+                verified = before["verify_ready"]
                 return _receipt(
                     mode=mode,
                     trace_id=trace_id,

apps/api/tests/test_db_bounded_executor.py

@@ -255,6 +255,7 @@ def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None:
     assert absent["index_exists"] is False
     assert exact["index_shape_exact"] is True
     assert exact["apply_ready"] is True
+    assert exact["verify_ready"] is True
     assert invalid_exact["index_shape_exact"] is False
     assert invalid_exact["index_definition_exact_except_validity"] is True
     assert invalid_exact["reindex_ready"] is True
@@ -271,6 +272,7 @@ def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None:
     assert non_owner["apply_ready"] is False
     # CREATE INDEX requires table ownership, not a separate SELECT grant.
     assert select_revoked["apply_ready"] is True
+    assert select_revoked["verify_ready"] is False
     assert partitioned_parent["apply_ready"] is False
 
 
@@ -457,6 +459,32 @@ async def test_verify_is_catalog_only_and_never_takes_the_apply_lock() -> None:
     assert all("advisory_lock" not in sql for sql in connection.statements)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "runtime_drift",
+    [
+        {"primary_database": False},
+        {"table_select": False},
+        {"schema_usage": False},
+    ],
+)
+async def test_verify_fails_closed_on_runtime_identity_or_read_drift(
+    runtime_drift: dict,
+) -> None:
+    connection = _Connection([{**_exact_catalog(), **runtime_drift}])
+    engine = _Engine(connection)
+
+    receipt = await execute_bounded_db_command(
+        **_contract("verify"),
+        engine=engine,
+    )
+
+    assert receipt["terminal"] == "verification_failed"
+    assert receipt["writes_performed"] is False
+    assert receipt["catalog"]["verify_ready"] is False
+    assert receipt["error_code"] == "catalog_verification_failed"
+
+
 @pytest.mark.asyncio
 async def test_failed_create_reports_unknown_write_state_without_backend_detail() -> (
     None