feat(mcp): add governed cross-product control plane

apps/api/src/api/v1/mcp_control_plane.py

@@ -0,0 +1,70 @@
+"""Read-only API for the 12-product MCP control plane cockpit."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import structlog
+from fastapi import APIRouter, HTTPException, Query, status
+
+from src.models.mcp_control_plane import McpControlPlaneOverview
+from src.services.mcp_control_plane_service import (
+    build_mcp_control_plane_overview,
+    collect_mcp_control_plane_runtime,
+    load_mcp_control_plane_catalog,
+)
+from src.services.product_governance_matrix_readback import (
+    load_latest_product_governance_matrix_readback,
+)
+from src.services.public_redaction import redact_public_lan_topology
+
+logger = structlog.get_logger(__name__)
+router = APIRouter(prefix="/mcp-control-plane", tags=["MCP Control Plane"])
+
+
+@router.get(
+    "/overview",
+    response_model=McpControlPlaneOverview,
+    summary="取得 12 產品 MCP 控制面總覽",
+    description=(
+        "整合 Gitea product governance matrix、AWOOOI MCP Provider Registry、"
+        "Gateway aggregate audit 與既有 RAG stats。回應只包含 catalog、狀態與聚合數字；"
+        "不安裝外部 MCP、不呼叫工具、不讀取 secret、不保存 request/response body、"
+        "不寫入 RAG，且 GitHub 全面排除。"
+    ),
+)
+async def get_mcp_control_plane_overview(
+    include_runtime: bool = Query(
+        True,
+        description="是否讀取 provider registry、Gateway 24h aggregate 與 RAG stats",
+    ),
+) -> McpControlPlaneOverview:
+    try:
+        catalog, product_matrix = await asyncio.gather(
+            asyncio.to_thread(load_mcp_control_plane_catalog),
+            asyncio.to_thread(load_latest_product_governance_matrix_readback),
+        )
+        runtime = await collect_mcp_control_plane_runtime() if include_runtime else None
+        payload = build_mcp_control_plane_overview(
+            catalog=catalog,
+            product_matrix=product_matrix,
+            runtime=runtime,
+        )
+        return McpControlPlaneOverview.model_validate(
+            redact_public_lan_topology(payload)
+        )
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="MCP control plane source inventory is not available",
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error(
+            "mcp_control_plane_inventory_invalid",
+            error_type=type(exc).__name__,
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="MCP control plane source inventory is invalid",
+        ) from exc

apps/api/src/main.py

@@ -65,6 +65,7 @@ from src.api.v1 import iwooos as iwooos_v1  # IwoooS security governance API
 from src.api.v1 import knowledge as knowledge_v1  # KB Phase 1: Knowledge Base
 from src.api.v1 import learning as learning_v1  # Phase D-G P0: Learning API
 from src.api.v1 import metrics as metrics_v1  # Phase 7: Gold Metrics (真實血脈)
+from src.api.v1 import mcp_control_plane as mcp_control_plane_v1
 from src.api.v1 import monitoring as monitoring_v1  # 2026-04-03: 監控工具狀態
 from src.api.v1 import notifications as notifications_v1  # 2026-04-10: 通知頻道狀態
 from src.api.v1 import (
@@ -1269,6 +1270,11 @@ app.include_router(
 app.include_router(
     rag_v1.router, prefix="/api/v1", tags=["RAG Knowledge Base"]
 )  # Phase 33 ADR-067: RAG 知識庫
+app.include_router(
+    mcp_control_plane_v1.router,
+    prefix="/api/v1",
+    tags=["MCP Control Plane"],
+)
 app.include_router(
     errors_v1.router, prefix="/api/v1", tags=["Errors"]
 )  # #40: Sentry 錯誤 BFF API

apps/api/src/models/mcp_control_plane.py

@@ -0,0 +1,67 @@
+"""Response contracts for the cross-product MCP control plane."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class McpCapabilityReadback(BaseModel):
+    """One governed internal capability or external discovery candidate."""
+
+    model_config = ConfigDict(extra="allow")
+
+    capability_id: str
+    title: str
+    source_type: str
+    catalog_source: str
+    catalog_url: str | None = None
+    decision: str
+    risk_tier: str
+    scope: str
+    data_boundary: str
+    deployment_allowed: bool
+    version_state: str
+    digest_state: str
+    sbom_state: str
+    signature_state: str
+    rag_policy: str
+    blockers: list[str] = Field(default_factory=list)
+    evidence_refs: list[str] = Field(default_factory=list)
+
+
+class McpProductReadback(BaseModel):
+    """One product joined with Gitea truth and MCP desired/observed state."""
+
+    model_config = ConfigDict(extra="allow")
+
+    product_id: str
+    canonical_repo: str
+    source_control_status: str
+    inventory_state: str
+    existing_capability_ids: list[str]
+    recommended_capability_ids: list[str]
+    rejected_capability_ids: list[str]
+    blockers: list[str]
+
+
+class McpControlPlaneOverview(BaseModel):
+    """Redacted read-only cockpit payload."""
+
+    schema_version: str
+    generated_at: str
+    status: str
+    summary: dict[str, Any]
+    architecture: dict[str, Any]
+    security_policy: dict[str, Any]
+    supply_chain_policy: dict[str, Any]
+    version_lifecycle: dict[str, Any]
+    runtime: dict[str, Any]
+    catalog_sources: list[dict[str, Any]]
+    capabilities: list[McpCapabilityReadback]
+    products: list[McpProductReadback]
+    active_blockers: list[str]
+    next_actions: list[str]
+    operation_boundaries: dict[str, bool]
+    source_refs: dict[str, str]

apps/api/src/plugins/mcp/gateway.py

@@ -35,8 +35,6 @@ AwoooP Phase 5.2: ADR-116 五閘門強制執行
 
 from __future__ import annotations
 
-import hashlib
-import json
 import time
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
@@ -56,6 +54,11 @@ from src.db.awooop_models import (
     AwoooPProject,
 )
 from src.plugins.mcp.interfaces import MCPToolResult
+from src.plugins.mcp.redaction_middleware import (
+    compute_safe_hash,
+    redact_mcp_input,
+    redact_mcp_output,
+)
 from src.plugins.mcp.registry import get_provider_registry
 
 logger = structlog.get_logger(__name__)
@@ -99,6 +102,13 @@ class GateApprovalError(McpGatewayError):
         super().__init__("E-MCP-GATE-005", msg, gate=5)
 
 
+class GateRateLimitError(McpGatewayError):
+    def __init__(self, msg: str = "distributed rate limit exceeded") -> None:
+        # Rate limiting is part of the governed tool/grant boundary (Gate 3),
+        # preserving the existing durable audit schema's 1..5 gate range.
+        super().__init__("E-MCP-RATE-429", msg, gate=3)
+
+
 class CredentialResolutionError(McpGatewayError):
     def __init__(self, msg: str = "credential 解析失敗") -> None:
         super().__init__("E-MCP-GATE-009", msg, gate=0)
@@ -125,6 +135,7 @@ class GateCheckResult:
     gate1_project: bool = False
     gate2_agent: bool = False
     gate3_tool: bool = False
+    distributed_rate_limit: bool = False
     gate4_env: bool = False
     gate5_approval: bool = False
 
@@ -133,6 +144,7 @@ class GateCheckResult:
             "gate1_project": self.gate1_project,
             "gate2_agent": self.gate2_agent,
             "gate3_tool": self.gate3_tool,
+            "distributed_rate_limit": self.distributed_rate_limit,
             "gate4_env": self.gate4_env,
             "gate5_approval": self.gate5_approval,
         }
@@ -143,6 +155,7 @@ class GateCheckResult:
             self.gate1_project,
             self.gate2_agent,
             self.gate3_tool,
+            self.distributed_rate_limit,
             self.gate4_env,
             self.gate5_approval,
         ])
@@ -186,6 +199,10 @@ class McpGateway:
             # Gate 3 — Tool + Grant
             tool_row, grant_row = await self._gate3_tool(ctx, gate_result)
 
+            # Gate 3b — Redis-backed distributed rate limit. A process-local
+            # counter would diverge across replicas and is therefore forbidden.
+            await self._enforce_distributed_rate_limit(ctx, gate_result)
+
             # Gate 4 — Environment（shadow mode 直接放行）
             await self._gate4_environment(ctx, tool_row, gate_result)
 
@@ -320,6 +337,46 @@ class McpGateway:
         gate_result.gate3_tool = True
         return tool_row, grant_row
 
+    async def _enforce_distributed_rate_limit(
+        self,
+        ctx: GatewayContext,
+        gate_result: GateCheckResult,
+    ) -> None:
+        """Apply an atomic per-project/agent/tool Redis fixed-window limit."""
+        limits = {"read": 120, "write": 20, "admin": 5}
+        limit = limits.get(ctx.required_scope, 5)
+        window_seconds = 60
+        key = (
+            "mcp_rate_limit:"
+            f"{ctx.project_id}:{ctx.agent_id}:{ctx.tool_name}:{ctx.required_scope}"
+        )
+        script = """
+        local current = redis.call('INCR', KEYS[1])
+        if current == 1 then
+          redis.call('EXPIRE', KEYS[1], ARGV[1])
+        end
+        return current
+        """
+        try:
+            redis = get_redis()
+            count = int(await redis.eval(script, 1, key, window_seconds))
+        except Exception as exc:
+            logger.warning(
+                "mcp_distributed_rate_limit_unavailable",
+                project_id=ctx.project_id,
+                tool_name=ctx.tool_name,
+                required_scope=ctx.required_scope,
+                error_type=type(exc).__name__,
+            )
+            raise GateRateLimitError(
+                "distributed rate-limit backend unavailable; fail closed"
+            ) from exc
+        if count > limit:
+            raise GateRateLimitError(
+                f"tool '{ctx.tool_name}' exceeded {limit} calls per {window_seconds}s"
+            )
+        gate_result.distributed_rate_limit = True
+
     async def _gate4_environment(
         self,
         ctx: GatewayContext,
@@ -390,7 +447,8 @@ class McpGateway:
         """呼叫底層 MCP provider 執行工具"""
         provider = await self._resolve_provider(ctx, tool_row)
 
-        # 找不到 provider → 回傳 shadow no-op
+        # A missing provider is evidence of an unclosed runtime path. Shadow mode
+        # must not turn that absence into a successful execution receipt.
         if provider is None:
             logger.warning(
                 "mcp_gateway_no_provider",
@@ -398,12 +456,21 @@ class McpGateway:
                 is_shadow=ctx.is_shadow,
             )
             return MCPToolResult(
-                success=True,
-                execution_id=f"shadow-noop-{ctx.tool_name}",
-                output={"shadow": True, "message": "no provider registered, shadow no-op"},
+                success=False,
+                execution_id=f"shadow-provider-missing-{ctx.tool_name}",
+                output={
+                    "shadow": True,
+                    "status": "degraded",
+                    "provider_registered": False,
+                },
+                error="mcp_provider_not_registered",
             )
 
-        audit_params = dict(parameters)
+        # Tool parameters are an execution boundary, not an audit payload. Apply
+        # credential and pattern redaction before they can reach any provider;
+        # the canonical audit context is re-added separately and is stripped by
+        # AuditedMCPToolProvider before the inner provider call.
+        audit_params = redact_mcp_input(parameters)
         existing_audit = (
             parameters.get("_mcp_audit")
             if isinstance(parameters, dict) and isinstance(parameters.get("_mcp_audit"), dict)
@@ -420,7 +487,11 @@ class McpGateway:
             "agent_role": existing_audit.get("agent_role") or ctx.agent_id,
             "gateway_path": "awooop_mcp_gateway",
         }
-        return await provider.execute(ctx.tool_name, audit_params)
+        result = await provider.execute(ctx.tool_name, audit_params)
+        result.output = redact_mcp_output(result.output)
+        if result.data is not None:
+            result.data = result.output
+        return result
 
     async def _resolve_provider(
         self,
@@ -472,15 +543,14 @@ class McpGateway:
     ) -> None:
         """寫 awooop_mcp_gateway_audit — 只寫 hash，不寫明文 input/output"""
         try:
-            input_hash = hashlib.sha256(
-                json.dumps(parameters, sort_keys=True, default=str).encode()
-            ).hexdigest()
+            # Hash only the redacted canonical form. This avoids persisting a
+            # reversible low-entropy hash of credential-shaped input while still
+            # retaining deterministic request correlation.
+            input_hash = compute_safe_hash(redact_mcp_input(parameters))
 
             output_hash: str | None = None
             if result is not None:
-                output_hash = hashlib.sha256(
-                    json.dumps(result.output, sort_keys=True, default=str).encode()
-                ).hexdigest()
+                output_hash = compute_safe_hash(redact_mcp_output(result.output))
 
             gate_payload = {
                 **gate_result.as_dict(),
@@ -489,6 +559,9 @@ class McpGateway:
                 "policy_enforced": True,
                 "is_shadow": ctx.is_shadow,
                 "required_scope": ctx.required_scope,
+                "body_storage": "hash_only",
+                "input_redaction_enforced": True,
+                "output_redaction_enforced": True,
             }
 
             audit = AwoooPMcpGatewayAudit(

apps/api/src/services/mcp_control_plane_service.py

@@ -0,0 +1,451 @@
+"""Cross-product MCP control plane catalog and runtime readback.
+
+The service deliberately separates committed desired/source inventory from live
+runtime evidence. Marketplace directories are discovery inputs only; this code
+never installs packages, resolves credentials, calls a tool, writes RAG data, or
+promotes an external candidate.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import structlog
+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_tool_registry import get_mcp_tool_registry
+from src.services.snapshot_paths import default_operations_dir
+
+logger = structlog.get_logger(__name__)
+
+SCHEMA_VERSION = "mcp_control_plane_overview_v1"
+CATALOG_SCHEMA_VERSION = "mcp_control_plane_catalog_v1"
+TZ_TAIPEI = ZoneInfo("Asia/Taipei")
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_CATALOG_FILE = "mcp-control-plane-catalog.snapshot.json"
+_EXPECTED_PRODUCT_IDS = {
+    "awoooi",
+    "ewoooc",
+    "2026fifa",
+    "agent-bounty-protocol",
+    "awooogo",
+    "stockplatform-v2",
+    "vibework",
+    "momo-pro-system",
+    "tsenyang-website",
+    "vtuber",
+    "bitan-pharmacy",
+    "clawbot-openclaw",
+}
+
+
+def load_mcp_control_plane_catalog(
+    operations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load and validate the committed 12-product MCP catalog."""
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
+    path = directory / _CATALOG_FILE
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    _validate_catalog(payload, str(path))
+    return payload
+
+
+def build_mcp_control_plane_overview(
+    *,
+    catalog: dict[str, Any],
+    product_matrix: dict[str, Any],
+    runtime: dict[str, Any] | None = None,
+    generated_at: str | None = None,
+) -> dict[str, Any]:
+    """Join catalog desired state with source-control and runtime truth."""
+    _validate_catalog(catalog, "catalog")
+    matrix_rows = {
+        str(row.get("product_id")): row
+        for row in _dict_rows(product_matrix.get("products"))
+        if row.get("product_id")
+    }
+    if set(matrix_rows) != _EXPECTED_PRODUCT_IDS:
+        raise ValueError(
+            "product governance matrix must contain the exact 12 governed products: "
+            f"missing={sorted(_EXPECTED_PRODUCT_IDS - set(matrix_rows))}, "
+            f"unexpected={sorted(set(matrix_rows) - _EXPECTED_PRODUCT_IDS)}"
+        )
+
+    capability_rows = _dict_rows(catalog.get("capabilities"))
+    capability_index = {
+        str(row["capability_id"]): row for row in capability_rows
+    }
+    products: list[dict[str, Any]] = []
+    for desired in _dict_rows(catalog.get("products")):
+        product_id = str(desired["product_id"])
+        source = matrix_rows[product_id]
+        existing_ids = _strings(desired.get("existing_capability_ids"))
+        recommended_ids = _strings(desired.get("recommended_capability_ids"))
+        rejected_ids = _strings(desired.get("rejected_capability_ids"))
+        products.append(
+            {
+                "product_id": product_id,
+                "canonical_repo": str(source.get("canonical_repo") or ""),
+                "source_control_status": str(
+                    source.get("source_control_status") or "unknown"
+                ),
+                "prod_branch": str(source.get("prod_branch") or "main"),
+                "prod_sha_short": str(source.get("prod_sha_short") or ""),
+                "visibility_readback": str(
+                    source.get("visibility_readback") or "unknown"
+                ),
+                "inventory_state": str(desired.get("inventory_state") or "unknown"),
+                "existing_capability_ids": existing_ids,
+                "recommended_capability_ids": recommended_ids,
+                "rejected_capability_ids": rejected_ids,
+                "existing_capabilities": [
+                    _capability_summary(capability_index[item]) for item in existing_ids
+                ],
+                "recommended_capabilities": [
+                    _capability_summary(capability_index[item])
+                    for item in recommended_ids
+                ],
+                "rejected_capabilities": [
+                    _capability_summary(capability_index[item]) for item in rejected_ids
+                ],
+                "blockers": _strings(desired.get("blockers")),
+            }
+        )
+
+    runtime_payload = runtime or {
+        "requested": False,
+        "status": "not_requested",
+        "provider_registry": {
+            "status": "not_requested",
+            "provider_count": 0,
+            "provider_names": [],
+            "tool_count": 0,
+        },
+        "gateway_audit_24h": {
+            "status": "not_requested",
+            "call_count": 0,
+            "success_count": 0,
+            "blocked_count": 0,
+            "failed_count": 0,
+        },
+        "rag": {
+            "status": "not_requested",
+            "total_chunks": 0,
+            "sources": 0,
+        },
+    }
+    external_rows = [
+        row for row in capability_rows if row.get("source_type") == "external"
+    ]
+    promotion_ready = sum(
+        1 for row in external_rows if row.get("deployment_allowed") is True
+    )
+    detected_products = sum(
+        1
+        for row in products
+        if row["inventory_state"]
+        not in {"no_committed_mcp_runtime_detected", "no_federated_mcp_inventory_readback"}
+    )
+    quarantined_products = sum(
+        1
+        for row in products
+        if "quarantine" in row["inventory_state"]
+        or "remediation_required" in row["inventory_state"]
+    )
+
+    active_blockers = set(_strings(catalog["version_lifecycle"].get("active_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",
+        }
+    )
+    rag = runtime_payload.get("rag") if isinstance(runtime_payload.get("rag"), dict) else {}
+    if rag.get("total_chunks") == 0:
+        active_blockers.add("internal_rag_index_empty")
+    if runtime_payload.get("status") not in {"ready", "not_requested"}:
+        active_blockers.add("mcp_runtime_readback_degraded")
+
+    source_summary = product_matrix.get("summary")
+    source_summary = source_summary if isinstance(source_summary, dict) else {}
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "generated_at": generated_at
+        or datetime.now(TZ_TAIPEI).isoformat(timespec="seconds"),
+        "status": "partial_degraded_with_safe_next_action",
+        "summary": {
+            "product_count": len(products),
+            "expected_product_count": 12,
+            "source_control_ready_product_count": int(
+                source_summary.get("source_control_ready_product_count") or 0
+            ),
+            "mcp_source_detected_product_count": detected_products,
+            "mcp_source_gap_product_count": len(products) - detected_products,
+            "quarantined_or_remediation_product_count": quarantined_products,
+            "capability_count": len(capability_rows),
+            "external_candidate_count": len(external_rows),
+            "external_promotion_ready_count": promotion_ready,
+            "github_in_scope_count": 0,
+            "active_blocker_count": len(active_blockers),
+            "runtime_provider_count": int(
+                _nested(runtime_payload, "provider_registry", "provider_count") or 0
+            ),
+            "runtime_tool_count": int(
+                _nested(runtime_payload, "provider_registry", "tool_count") or 0
+            ),
+        },
+        "architecture": catalog["architecture"],
+        "security_policy": catalog["security_policy"],
+        "supply_chain_policy": catalog["supply_chain_policy"],
+        "version_lifecycle": catalog["version_lifecycle"],
+        "runtime": runtime_payload,
+        "catalog_sources": catalog["source_policy"]["catalog_sources"],
+        "capabilities": capability_rows,
+        "products": products,
+        "active_blockers": sorted(active_blockers),
+        "next_actions": [
+            "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",
+            "implement_durable_version_discovery_compatibility_canary_and_rollback_controller",
+            "mirror_and_pin_each_approved_external_candidate_before_shadow",
+            "wire_quarantine_scan_citation_verifier_and_governed_rag_promotion",
+        ],
+        "operation_boundaries": {
+            "read_only_api_allowed": True,
+            "public_response_redacted_aggregate_only": True,
+            "mutation_endpoint_exposed": False,
+            "operator_auth_required_for_future_mutation": True,
+            "external_install_allowed": False,
+            "external_tool_call_allowed": False,
+            "external_rag_write_allowed": False,
+            "github_allowed": False,
+            "secret_value_read_allowed": False,
+            "request_or_response_body_storage_allowed": False,
+            "production_upgrade_allowed": False,
+        },
+        "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, and RAG stats",
+        },
+    }
+
+
+async def collect_mcp_control_plane_runtime() -> dict[str, Any]:
+    """Collect aggregate-only runtime evidence without request or response bodies."""
+    blockers: list[str] = []
+    try:
+        provider_registry = get_provider_registry()
+        tool_registry = get_mcp_tool_registry()
+        provider_payload = {
+            "status": "registered",
+            "provider_count": len(provider_registry.names()),
+            "provider_names": sorted(provider_registry.names()),
+            "tool_count": tool_registry.tool_count,
+        }
+    except Exception as exc:
+        logger.warning(
+            "mcp_control_plane_provider_registry_readback_failed",
+            error_type=type(exc).__name__,
+        )
+        blockers.append("provider_registry_readback_failed")
+        provider_payload = {
+            "status": "degraded",
+            "provider_count": 0,
+            "provider_names": [],
+            "tool_count": 0,
+        }
+
+    try:
+        # Control-plane readback is deliberately scoped to the canonical AWOOOI
+        # tenant. Query the governed pgvector table directly so repository
+        # fallback-to-zero cannot disguise a DB/RLS failure as a healthy empty
+        # index. Only aggregate counts leave this service.
+        async with get_db_context("awoooi") as db:
+            result = await db.execute(
+                text(
+                    """
+                    SELECT
+                        COUNT(*)::int AS total_chunks,
+                        COUNT(DISTINCT source)::int AS sources
+                    FROM rag_chunks
+                    """
+                )
+            )
+            rag_stats = result.mappings().one()
+        rag_payload = {
+            "status": "ready",
+            "total_chunks": int(rag_stats.get("total_chunks") or 0),
+            "sources": int(rag_stats.get("sources") or 0),
+            "scope": "project:awoooi",
+        }
+    except Exception as exc:
+        logger.warning(
+            "mcp_control_plane_rag_readback_failed",
+            error_type=type(exc).__name__,
+        )
+        blockers.append("rag_stats_readback_failed")
+        rag_payload = {
+            "status": "degraded",
+            "total_chunks": 0,
+            "sources": 0,
+            "scope": "project:awoooi",
+        }
+
+    try:
+        async with get_db_context("awoooi") as db:
+            result = await db.execute(
+                text(
+                    """
+                    SELECT
+                        COUNT(*)::int AS call_count,
+                        COUNT(*) FILTER (WHERE result_status = 'success')::int AS success_count,
+                        COUNT(*) FILTER (WHERE result_status = 'blocked')::int AS blocked_count,
+                        COUNT(*) FILTER (WHERE result_status IN ('failed', 'timeout'))::int AS failed_count
+                    FROM awooop_mcp_gateway_audit
+                    WHERE created_at >= NOW() - INTERVAL '24 hours'
+                    """
+                )
+            )
+            row = result.mappings().one()
+        gateway_payload = {
+            "status": "ready",
+            "call_count": int(row["call_count"] or 0),
+            "success_count": int(row["success_count"] or 0),
+            "blocked_count": int(row["blocked_count"] or 0),
+            "failed_count": int(row["failed_count"] or 0),
+            "body_storage": "hash_only",
+            "scope": "project:awoooi",
+        }
+    except Exception as exc:
+        logger.warning(
+            "mcp_control_plane_gateway_audit_readback_failed",
+            error_type=type(exc).__name__,
+        )
+        blockers.append("gateway_audit_readback_failed")
+        gateway_payload = {
+            "status": "degraded",
+            "call_count": 0,
+            "success_count": 0,
+            "blocked_count": 0,
+            "failed_count": 0,
+            "body_storage": "hash_only",
+            "scope": "project:awoooi",
+        }
+
+    return {
+        "requested": True,
+        "status": "ready" if not blockers else "degraded",
+        "provider_registry": provider_payload,
+        "gateway_audit_24h": gateway_payload,
+        "rag": rag_payload,
+        "federated_product_runtime_receipt_count": 0,
+        "federated_product_runtime_receipt_expected": 12,
+        "blockers": sorted(blockers),
+    }
+
+
+def _validate_catalog(payload: dict[str, Any], label: str) -> None:
+    if payload.get("schema_version") != CATALOG_SCHEMA_VERSION:
+        raise ValueError(f"{label}: invalid schema_version")
+    architecture = payload.get("architecture") or {}
+    if architecture.get("decision") != "single_control_plane_reuse_existing_gateway_registry_rag":
+        raise ValueError(f"{label}: duplicate gateway/RAG architecture is forbidden")
+    if architecture.get("separate_gateway_allowed") is not False:
+        raise ValueError(f"{label}: separate gateway must remain false")
+    if architecture.get("separate_rag_database_allowed") is not False:
+        raise ValueError(f"{label}: separate RAG database must remain false")
+
+    source_policy = payload.get("source_policy") or {}
+    for key in (
+        "github_allowed",
+        "github_actions_allowed",
+        "github_artifact_sources_allowed",
+    ):
+        if source_policy.get(key) is not False:
+            raise ValueError(f"{label}: {key} must remain false")
+    security = payload.get("security_policy") or {}
+    if security.get("external_rag_auto_write_allowed") is not False:
+        raise ValueError(f"{label}: external RAG auto-write must remain false")
+    if security.get("gateway_request_body_storage") != "forbidden_hash_only":
+        raise ValueError(f"{label}: gateway request body policy must be hash-only")
+
+    capabilities = _dict_rows(payload.get("capabilities"))
+    capability_ids = [str(row.get("capability_id") or "") for row in capabilities]
+    if not all(capability_ids) or len(capability_ids) != len(set(capability_ids)):
+        raise ValueError(f"{label}: capability IDs must be non-empty and unique")
+    capability_id_set = set(capability_ids)
+    github_rows = [
+        row for row in capabilities if row.get("capability_id") == "rejected.github"
+    ]
+    if len(github_rows) != 1 or github_rows[0].get("decision") != "rejected":
+        raise ValueError(f"{label}: GitHub capability must be explicitly rejected")
+    for row in capabilities:
+        if row.get("source_type") != "external":
+            continue
+        if row.get("deployment_allowed") is True and any(
+            row.get(field) in {None, "", "missing", "unresolved"}
+            for field in ("version_state", "digest_state", "sbom_state", "signature_state")
+        ):
+            raise ValueError(
+                f"{label}: external capability {row.get('capability_id')} cannot deploy without immutable receipts"
+            )
+
+    products = _dict_rows(payload.get("products"))
+    product_ids = {str(row.get("product_id") or "") for row in products}
+    if product_ids != _EXPECTED_PRODUCT_IDS or len(products) != 12:
+        raise ValueError(
+            f"{label}: catalog must contain exact 12 products; "
+            f"missing={sorted(_EXPECTED_PRODUCT_IDS - product_ids)}, "
+            f"unexpected={sorted(product_ids - _EXPECTED_PRODUCT_IDS)}"
+        )
+    for row in products:
+        refs = (
+            _strings(row.get("existing_capability_ids"))
+            + _strings(row.get("recommended_capability_ids"))
+            + _strings(row.get("rejected_capability_ids"))
+        )
+        unknown = sorted(set(refs) - capability_id_set)
+        if unknown:
+            raise ValueError(
+                f"{label}: product {row.get('product_id')} references unknown capabilities {unknown}"
+            )
+
+
+def _capability_summary(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "capability_id": row["capability_id"],
+        "title": row["title"],
+        "decision": row["decision"],
+        "risk_tier": row["risk_tier"],
+        "deployment_allowed": row["deployment_allowed"],
+    }
+
+
+def _dict_rows(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list):
+        return []
+    return [row for row in value if isinstance(row, dict)]
+
+
+def _strings(value: Any) -> list[str]:
+    if not isinstance(value, list):
+        return []
+    return [str(item) for item in value if str(item)]
+
+
+def _nested(payload: dict[str, Any], first: str, second: str) -> Any:
+    nested = payload.get(first)
+    return nested.get(second) if isinstance(nested, dict) else None