feat(mcp): add durable product federation receipts

apps/api/src/services/mcp_control_plane_service.py

@@ -8,6 +8,7 @@ promotes an external candidate.
 
 from __future__ import annotations
 
+import asyncio
 import json
 from datetime import datetime
 from pathlib import Path
@@ -19,6 +20,7 @@ from sqlalchemy import text
 
 from src.db.base import get_db_context
 from src.plugins.mcp.registry import get_provider_registry
+from src.services.mcp_federation_service import collect_mcp_federation_readback
 from src.services.mcp_tool_registry import get_mcp_tool_registry
 from src.services.mcp_version_lifecycle_service import (
     collect_mcp_version_lifecycle_readback,
@@ -143,7 +145,40 @@ def build_mcp_control_plane_overview(
             "total_chunks": 0,
             "sources": 0,
         },
+        "federated_runtime": {
+            "status": "not_requested",
+            "receipts": [],
+            "verified_fresh_receipt_count": 0,
+            "expected_receipt_count": 2,
+            "blockers": [],
+        },
+        "federated_product_runtime_receipt_count": 0,
+        "federated_product_runtime_receipt_expected": 12,
     }
+    federation_runtime = runtime_payload.get("federated_runtime")
+    federation_runtime = (
+        federation_runtime if isinstance(federation_runtime, dict) else {}
+    )
+    federation_receipt_index = {
+        str(row.get("product_id")): row
+        for row in _dict_rows(federation_runtime.get("receipts"))
+        if row.get("product_id") and row.get("fresh") is True
+    }
+    for product in products:
+        receipt = federation_receipt_index.get(product["product_id"])
+        product["federated_runtime_receipt"] = receipt
+        if receipt is None:
+            continue
+        product["blockers"] = sorted(
+            set(product["blockers"])
+            .difference({"awoooi_federated_runtime_readback_missing"})
+            .union(_strings(receipt.get("blockers")))
+        )
+        product["inventory_state"] = (
+            "federated_runtime_readback_verified"
+            if receipt.get("health_status") == "ready"
+            else "federated_runtime_readback_partial_degraded"
+        )
     lifecycle_runtime = runtime_payload.get("version_lifecycle")
     lifecycle_runtime = lifecycle_runtime if isinstance(lifecycle_runtime, dict) else {}
     version_lifecycle = dict(catalog["version_lifecycle"])
@@ -179,14 +214,27 @@ def build_mcp_control_plane_overview(
 
     active_blockers = set(_strings(version_lifecycle.get("active_blockers")))
     active_blockers.update(_strings(lifecycle_runtime.get("blockers")))
+    active_blockers.update(_strings(federation_runtime.get("blockers")))
+    active_blockers.update(
+        blocker
+        for product in products
+        for blocker in _strings(product.get("blockers"))
+    )
     active_blockers.update(
         {
-            "cross_product_runtime_federation_receipts_missing",
             "external_candidate_immutable_supply_chain_receipts_missing",
             "external_rag_quarantine_promotion_verifier_missing",
             "distributed_gateway_rate_limit_receipt_missing",
         }
     )
+    federated_receipt_count = int(
+        runtime_payload.get("federated_product_runtime_receipt_count") or 0
+    )
+    federated_receipt_expected = int(
+        runtime_payload.get("federated_product_runtime_receipt_expected") or 12
+    )
+    if federated_receipt_count < federated_receipt_expected:
+        active_blockers.add("cross_product_runtime_federation_receipts_missing")
     rag = (
         runtime_payload.get("rag")
         if isinstance(runtime_payload.get("rag"), dict)
@@ -224,6 +272,8 @@ def build_mcp_control_plane_overview(
             "runtime_tool_count": int(
                 _nested(runtime_payload, "provider_registry", "tool_count") or 0
             ),
+            "federated_product_runtime_receipt_count": federated_receipt_count,
+            "federated_product_runtime_receipt_expected": federated_receipt_expected,
         },
         "architecture": catalog["architecture"],
         "security_policy": catalog["security_policy"],
@@ -235,7 +285,11 @@ def build_mcp_control_plane_overview(
         "products": products,
         "active_blockers": sorted(active_blockers),
         "next_actions": [
-            "federate_ewoooc_and_momo_read_only_inventory_health_and_version_receipts",
+            (
+                "resolve_ewoooc_momo_runtime_blockers_and_expand_next_product_federation_receipts"
+                if len(federation_receipt_index) == 2
+                else "federate_ewoooc_and_momo_read_only_inventory_health_and_version_receipts"
+            ),
             "disable_agent_bounty_github_and_prompt_injection_tool_path",
             "remove_bitan_github_external_provider_from_effective_policy",
             "promote_first_internally_mirrored_candidate_through_replay_shadow_and_canary",
@@ -259,7 +313,7 @@ def build_mcp_control_plane_overview(
         "source_refs": {
             "catalog": f"docs/operations/{_CATALOG_FILE}",
             "product_matrix": "GET /api/v1/agents/product-governance-matrix-readback",
-            "runtime": "live provider registry, MCP tool registry, gateway audit aggregate, RAG stats, and durable version lifecycle receipts",
+            "runtime": "live provider registry, MCP tool registry, gateway audit aggregate, RAG stats, durable version lifecycle receipts, and normalized federated product receipts",
         },
     }
 
@@ -267,9 +321,14 @@ def build_mcp_control_plane_overview(
 async def collect_mcp_control_plane_runtime() -> dict[str, Any]:
     """Collect aggregate-only runtime evidence without request or response bodies."""
     blockers: list[str] = []
-    lifecycle_payload = await collect_mcp_version_lifecycle_readback()
+    lifecycle_payload, federation_payload = await asyncio.gather(
+        collect_mcp_version_lifecycle_readback(),
+        collect_mcp_federation_readback(),
+    )
     if lifecycle_payload.get("status") == "degraded":
         blockers.append("version_lifecycle_runtime_readback_degraded")
+    if federation_payload.get("status") not in {"ready", "not_requested"}:
+        blockers.append("federation_runtime_readback_degraded")
     try:
         provider_registry = get_provider_registry()
         tool_registry = get_mcp_tool_registry()
@@ -376,7 +435,10 @@ async def collect_mcp_control_plane_runtime() -> dict[str, Any]:
         "gateway_audit_24h": gateway_payload,
         "rag": rag_payload,
         "version_lifecycle": lifecycle_payload,
-        "federated_product_runtime_receipt_count": 0,
+        "federated_runtime": federation_payload,
+        "federated_product_runtime_receipt_count": int(
+            federation_payload.get("verified_fresh_receipt_count") or 0
+        ),
         "federated_product_runtime_receipt_expected": 12,
         "blockers": sorted(blockers),
     }