fix(api): fail closed on stale production runtime deploy readback

ops/runner/test_verify_awoooi_non110_cd_closure.py

@@ -46,12 +46,35 @@ def _queue(*, no_matching: bool) -> dict:
     }
 
 
-def _workbench(*, image_current: bool, governance_ready: bool) -> dict:
+def _workbench(
+    *,
+    image_current: bool,
+    governance_ready: bool,
+    runtime_source_current: bool | None = None,
+    runtime_image_current: bool | None = None,
+) -> dict:
+    runtime_source_current = (
+        image_current if runtime_source_current is None else runtime_source_current
+    )
+    runtime_image_current = (
+        image_current if runtime_image_current is None else runtime_image_current
+    )
     return {
         "schema_version": "delivery_closure_workbench_v1",
         "summary": {
             "source_count": 6,
             "production_deploy_image_tag_matches_main": image_current,
+            "production_deploy_runtime_build_matches_committed_source_control_readback": (
+                runtime_source_current
+            ),
+            "production_deploy_runtime_build_matches_committed_production_image_tag": (
+                runtime_image_current
+            ),
+            "production_deploy_runtime_build_readback_status": (
+                "matches_committed_deploy_readback"
+                if runtime_source_current and runtime_image_current
+                else "runtime_build_diverges_from_committed_deploy_readback"
+            ),
             "production_deploy_governance_fields_present": governance_ready,
         },
     }
@@ -191,6 +214,37 @@ def test_closure_verifier_accepts_full_closure_evidence() -> None:
     assert all(step["status"] == "complete" for step in payload["ordered_steps"])
     assert payload["blockers"] == []
     assert payload["readback"]["production_deploy_image_tag_matches_main"] is True
+    assert (
+        payload["readback"][
+            "production_deploy_runtime_build_matches_committed_source_control_readback"
+        ]
+        is True
+    )
+
+
+def test_closure_verifier_blocks_stale_runtime_even_when_desired_tag_matches() -> None:
+    module = _load_module()
+    payload = module.build_closure_verifier(
+        readiness_text=_readiness(ready=True),
+        queue=_queue(no_matching=False),
+        production_workbench=_workbench(
+            image_current=True,
+            governance_ready=True,
+            runtime_source_current=False,
+            runtime_image_current=False,
+        ),
+    )
+    assert payload["status"] == "blocked_production_image_not_current"
+    assert payload["progress"]["ordered_completed_prefix_count"] == 4
+    assert payload["progress"]["next_blocked_step_id"] == "production_image_tag_current"
+    assert "production_image_tag_not_current" in payload["blockers"]
+    assert payload["readback"]["production_deploy_image_tag_matches_main"] is False
+    assert (
+        payload["readback"][
+            "production_deploy_runtime_build_matches_committed_source_control_readback"
+        ]
+        is False
+    )
 
 
 def test_cli_uses_fixture_files_without_live_dispatch(tmp_path: Path) -> None:

ops/runner/verify-awoooi-non110-cd-closure.py

@@ -158,6 +158,20 @@ def _production_summary(workbench: dict[str, Any]) -> dict[str, Any]:
     return summary if isinstance(summary, dict) else {}
 
 
+def _production_image_tag_current(production: dict[str, Any]) -> bool:
+    return (
+        production.get("production_deploy_image_tag_matches_main") is True
+        and production.get(
+            "production_deploy_runtime_build_matches_committed_source_control_readback"
+        )
+        is True
+        and production.get(
+            "production_deploy_runtime_build_matches_committed_production_image_tag"
+        )
+        is True
+    )
+
+
 def _build_ordered_steps(
     *,
     readiness: dict[str, Any],
@@ -292,9 +306,7 @@ def build_closure_verifier(
         queue_readback.get("no_matching_online_runner_visible") is True
     )
     production_workbench_present = bool(production)
-    production_image_tag_matches_main = (
-        production.get("production_deploy_image_tag_matches_main") is True
-    )
+    production_image_tag_matches_main = _production_image_tag_current(production)
     production_governance_fields_present = (
         production.get("production_deploy_governance_fields_present") is True
     )
@@ -375,6 +387,22 @@ def build_closure_verifier(
             "production_deploy_image_tag_matches_main": (
                 production_image_tag_matches_main
             ),
+            "production_deploy_runtime_build_matches_committed_source_control_readback": (
+                production.get(
+                    "production_deploy_runtime_build_matches_committed_source_control_readback"
+                )
+                is True
+            ),
+            "production_deploy_runtime_build_matches_committed_production_image_tag": (
+                production.get(
+                    "production_deploy_runtime_build_matches_committed_production_image_tag"
+                )
+                is True
+            ),
+            "production_deploy_runtime_build_readback_status": str(
+                production.get("production_deploy_runtime_build_readback_status")
+                or ""
+            ),
             "production_deploy_governance_fields_present": (
                 production_governance_fields_present
             ),