wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into codex/p0-obs-002-20260715

Browse Source 
# Conflicts:
#	docs/LOGBOOK.md
This commit is contained in:
ogt
2026-07-15 02:00:02 +08:00
parent 88a457892e 31de3a7098
commit 7d48f28d05
26 changed files with 2804 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
apps/api/src/api/v1/mcp_control_plane.py Normal file
View File

@@ -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

6
apps/api/src/main.py
View File

@@ -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

67
apps/api/src/models/mcp_control_plane.py Normal file
View File

@@ -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]

101
apps/api/src/plugins/mcp/gateway.py
View File

@@ -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(

115
apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py
View File

@@ -11,8 +11,11 @@ prove this route has rolled out.
from __future__ import annotations
import asyncio
import copy
import json
import time
from collections.abc import Mapping
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import text
@@ -51,6 +54,13 @@ _CONSUMER_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
_CONSUMER_DIRECT_QUERY_TIMEOUT_SECONDS = 2.0
_CONSUMER_TELEGRAM_CONTEXT_TIMEOUT_SECONDS = 1.5
_CONSUMER_STATEMENT_TIMEOUT_MS = 1500
_CONSUMER_READBACK_CACHE_TTL_SECONDS = 20.0
_consumer_readback_cache: dict[
tuple[int, int, str], tuple[float, dict[str, Any]]
] = {}
_consumer_readback_inflight: dict[
tuple[int, int, str], asyncio.Task[dict[str, Any]]
] = {}
logger = get_logger(__name__)
@@ -58,7 +68,48 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
*,
project_id: str = DEFAULT_PROJECT_ID,
) -> dict[str, Any]:
"""Return live consumer bindings for LOG controlled writeback receipts."""
"""Return live consumer bindings without duplicating the same DB read burst."""
loop = asyncio.get_running_loop()
cache_key = (id(loop), id(get_db_context), project_id)
cached = _consumer_readback_cache_get(cache_key=cache_key)
if cached is not None:
return cached
task = _consumer_readback_inflight.get(cache_key)
coalesced = task is not None
if task is None:
task = loop.create_task(
_load_latest_ai_agent_log_controlled_writeback_consumer_readback_uncached(
project_id=project_id
)
)
_consumer_readback_inflight[cache_key] = task
try:
payload = await asyncio.shield(task)
payload = _with_consumer_readback_freshness(
payload,
cache_hit=False,
cache_age_seconds=0.0,
inflight_coalesced=coalesced,
)
if _consumer_readback_cacheable(payload):
_consumer_readback_cache[cache_key] = (
time.monotonic(),
copy.deepcopy(payload),
)
return payload
finally:
if task.done() and _consumer_readback_inflight.get(cache_key) is task:
_consumer_readback_inflight.pop(cache_key, None)
async def _load_latest_ai_agent_log_controlled_writeback_consumer_readback_uncached(
*,
project_id: str,
) -> dict[str, Any]:
"""Read current receipts once; the public loader owns cache/coalescing."""
try:
rows, consumer_rows = await _load_consumer_rows_with_retry(project_id=project_id)
@@ -94,6 +145,68 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
)
def _consumer_readback_cache_get(
*,
cache_key: tuple[int, int, str],
) -> dict[str, Any] | None:
now = time.monotonic()
for key, (stored_at, _payload) in list(_consumer_readback_cache.items()):
if now - stored_at > _CONSUMER_READBACK_CACHE_TTL_SECONDS:
_consumer_readback_cache.pop(key, None)
cached = _consumer_readback_cache.get(cache_key)
if cached is None:
return None
stored_at, payload = cached
return _with_consumer_readback_freshness(
payload,
cache_hit=True,
cache_age_seconds=max(0.0, now - stored_at),
inflight_coalesced=False,
)
def _consumer_readback_cacheable(payload: Mapping[str, Any]) -> bool:
boundaries = payload.get("operation_boundaries")
return bool(
isinstance(boundaries, Mapping)
and boundaries.get("metadata_ledger_read_performed") is True
and boundaries.get("consumer_apply_receipt_read_performed") is True
)
def _with_consumer_readback_freshness(
payload: Mapping[str, Any],
*,
cache_hit: bool,
cache_age_seconds: float,
inflight_coalesced: bool,
) -> dict[str, Any]:
result = copy.deepcopy(dict(payload))
prior = result.get("readback_freshness")
observed_at = (
str(prior.get("observed_at") or "")
if isinstance(prior, Mapping)
else ""
)
result["readback_freshness"] = {
"observed_at": observed_at or datetime.now(UTC).isoformat(),
"cache_ttl_seconds": _CONSUMER_READBACK_CACHE_TTL_SECONDS,
"cache_hit": cache_hit,
"cache_age_seconds": round(cache_age_seconds, 3),
"stale": False,
"inflight_coalesced": inflight_coalesced,
}
return result
def _reset_consumer_readback_cache_for_tests() -> None:
"""Reset process-local readback state between focused unit tests."""
_consumer_readback_cache.clear()
_consumer_readback_inflight.clear()
async def _load_consumer_rows_with_retry(
*,
project_id: str,

41
apps/api/src/services/awooop_ansible_check_mode_service.py
View File

@@ -7733,9 +7733,18 @@ async def claim_catalog_drift_failed_check_modes(
{
"project_id": project_id,
"candidate_max_age_hours": max(1, max_age_hours),
"scan_limit": min(100, max(10, limit * 10)),
"scan_limit": 100,
},
)
ranked_rows: list[
tuple[
int,
dict[str, Any],
AnsibleCheckModeClaim,
str,
str,
]
] = []
for row_value in result.mappings().all():
row = dict(row_value)
previous_input = _json_loads(row.get("input"))
@@ -7744,6 +7753,9 @@ async def claim_catalog_drift_failed_check_modes(
or previous_input.get("playbook_path")
or ""
)
previous_catalog_id = str(
previous_input.get("catalog_id") or ""
)
previous_catalog_revision = str(
previous_input.get("catalog_revision") or ""
)
@@ -7768,6 +7780,33 @@ async def claim_catalog_drift_failed_check_modes(
or not (path_changed or revision_changed)
):
continue
catalog_route_changed = bool(
previous_catalog_id
and canonical_claim.catalog_id != previous_catalog_id
)
replay_priority = (
0 if catalog_route_changed else 1 if path_changed else 2
)
ranked_rows.append(
(
replay_priority,
row,
canonical_claim,
previous_check_path,
previous_catalog_revision,
)
)
for (
_,
row,
canonical_claim,
previous_check_path,
previous_catalog_revision,
) in sorted(ranked_rows, key=lambda item: item[0]):
current_catalog_revision = str(
canonical_claim.input_payload.get("catalog_revision") or ""
)
catalog_replay_revision = (
current_catalog_revision
or f"path:{canonical_claim.playbook_path}"

451
apps/api/src/services/mcp_control_plane_service.py Normal file
View File

@@ -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