Merge remote-tracking branch 'origin/main' into codex/sre-typed-automation-20260715

apps/api/src/services/mcp_control_plane_service.py

@@ -497,6 +497,68 @@ def _validate_catalog(payload: dict[str, Any], label: str) -> None:
             raise ValueError(
                 f"{label}: external capability {row.get('capability_id')} cannot deploy without immutable receipts"
             )
+        if row.get("compatibility_state") != "protocol_schema_compatibility_verified":
+            continue
+        if row.get("replay_status") != "completed_protocol_schema_replay_verified":
+            raise ValueError(
+                f"{label}: verified compatibility requires a completed replay verifier receipt"
+            )
+        if row.get("replay_independent_verifier") is not True:
+            raise ValueError(
+                f"{label}: verified compatibility requires an independent verifier"
+            )
+        if not isinstance(row.get("replay_tool_count"), int) or int(
+            row["replay_tool_count"]
+        ) <= 0:
+            raise ValueError(
+                f"{label}: verified compatibility requires a positive replay tool count"
+            )
+        if row.get("replay_process_exit_code") != 0 or row.get(
+            "replay_shutdown_mode"
+        ) != "stdin_eof_clean_exit":
+            raise ValueError(
+                f"{label}: verified compatibility requires a clean MCP process terminal"
+            )
+        if row.get("replay_ephemeral_container_removed") is not True:
+            raise ValueError(
+                f"{label}: verified compatibility requires ephemeral container removal"
+            )
+        no_effect_fields = (
+            "replay_browser_started",
+            "replay_tool_call_performed",
+            "replay_navigation_performed",
+            "replay_external_rag_write_performed",
+            "replay_production_write_performed",
+            "replay_promotion_allowed",
+        )
+        if any(row.get(field) is not False for field in no_effect_fields):
+            raise ValueError(
+                f"{label}: compatibility replay must remain no-browser, no-tool, no-write, and no-promotion"
+            )
+        if row.get("deployment_allowed") is not False:
+            raise ValueError(
+                f"{label}: compatibility verification alone cannot enable deployment"
+            )
+        fixed_hex_fields = {
+            "replay_audit_commit": 40,
+            "replay_artifact_audit_commit": 40,
+            "artifact_latest_revalidation_audit_commit": 40,
+            "replay_controller_receipt_sha256": 64,
+            "replay_verifier_receipt_sha256": 64,
+            "replay_tool_name_set_sha256": 64,
+            "replay_tool_schema_manifest_sha256": 64,
+        }
+        if any(
+            not _is_lower_hex(row.get(field), length)
+            for field, length in fixed_hex_fields.items()
+        ):
+            raise ValueError(
+                f"{label}: verified compatibility requires immutable audit and replay hashes"
+            )
+        if "compatibility_replay_not_executed" in _strings(row.get("blockers")):
+            raise ValueError(
+                f"{label}: verified compatibility cannot retain replay-not-executed blocker"
+            )
 
     products = _dict_rows(payload.get("products"))
     product_ids = {str(row.get("product_id") or "") for row in products}
@@ -529,6 +591,14 @@ def _capability_summary(row: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _is_lower_hex(value: Any, length: int) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) == length
+        and all(character in "0123456789abcdef" for character in value)
+    )
+
+
 def _dict_rows(value: Any) -> list[dict[str, Any]]:
     if not isinstance(value, list):
         return []