diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml
index 6205e9158..8a9a32a77 100644
--- a/.gitea/workflows/cd.yaml
+++ b/.gitea/workflows/cd.yaml
@@ -362,6 +362,16 @@ jobs:
;;
apps/api/src/core/config.py)
;;
+ apps/api/src/api/v1/mcp_control_plane.py)
+ ;;
+ apps/api/src/models/mcp_control_plane.py)
+ ;;
+ apps/api/src/jobs/mcp_version_lifecycle_job.py)
+ ;;
+ apps/api/src/services/mcp_control_plane_service.py)
+ ;;
+ apps/api/src/services/mcp_version_lifecycle_service.py)
+ ;;
apps/api/src/db/base.py)
;;
apps/api/src/services/agent_replay_normalizer.py)
@@ -408,6 +418,8 @@ jobs:
;;
apps/api/migrations/asset_inventory_database_types_2026-07-14_down.sql)
;;
+ apps/api/migrations/mcp_version_lifecycle_controller_2026-07-15.sql)
+ ;;
apps/api/src/services/auto_approve.py)
;;
apps/api/src/services/decision_fusion.py)
@@ -544,6 +556,10 @@ jobs:
;;
apps/api/tests/test_signal_worker_ansible_executor_binding.py)
;;
+ apps/api/tests/test_mcp_control_plane_api.py)
+ ;;
+ apps/api/tests/test_mcp_version_lifecycle_service.py)
+ ;;
apps/api/tests/test_channel_hub_grouped_alert_events.py)
;;
apps/api/tests/test_shadow_auto_approve.py)
@@ -644,6 +660,8 @@ jobs:
;;
ops/runner/test_verify_awoooi_non110_cd_closure.py)
;;
+ docs/operations/mcp-control-plane-catalog.snapshot.json)
+ ;;
ops/runner/awoooi-cd-lane-drain.service)
;;
ops/runner/check-awoooi-110-controlled-cd-lane-readiness.sh)
@@ -915,15 +933,19 @@ jobs:
python3.11 -m py_compile \
src/core/config.py \
src/db/base.py \
+ src/api/v1/mcp_control_plane.py \
src/api/v1/platform/events.py \
src/api/v1/agents.py \
src/api/v1/iwooos.py \
src/api/v1/webhooks.py \
src/jobs/ai_slo_watchdog_job.py \
+ src/jobs/mcp_version_lifecycle_job.py \
src/repositories/knowledge_repository.py \
src/models/knowledge.py \
src/models/playbook.py \
src/services/knowledge_service.py \
+ src/services/mcp_control_plane_service.py \
+ src/services/mcp_version_lifecycle_service.py \
src/services/awoooi_production_deploy_readback_blocker.py \
src/services/agent_replay_normalizer.py \
src/services/ai_agent_log_intelligence_integration_readback.py \
@@ -1067,6 +1089,8 @@ jobs:
tests/test_ai_agent_report_truth_actionability_review_api.py \
tests/test_awooop_truth_chain_service.py \
tests/test_signal_worker_ansible_executor_binding.py \
+ tests/test_mcp_control_plane_api.py \
+ tests/test_mcp_version_lifecycle_service.py \
tests/test_channel_hub_grouped_alert_events.py \
tests/test_shadow_auto_approve.py \
tests/test_destructive_patterns.py \
@@ -2241,6 +2265,126 @@ jobs:
asyncio.run(main())
PY
+ # MCP version discovery must have a durable receipt store before the
+ # signal worker starts its first six-hour lifecycle run. Apply the
+ # additive RLS schema, then verify both schema policy and runtime-role
+ # read access without printing connection details or row content.
+ - name: Apply MCP Version Lifecycle Receipt Schema
+ env:
+ DATABASE_URL: ${{ secrets.DATABASE_URL }}
+ MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}
+ run: |
+ set -euo pipefail
+ if [ -z "${MIGRATION_DATABASE_URL:-}" ]; then
+ echo "::error::MIGRATION_DATABASE_URL secret not set in Gitea"
+ exit 1
+ fi
+ if [ -z "${DATABASE_URL:-}" ]; then
+ echo "::error::DATABASE_URL secret not set in Gitea"
+ exit 1
+ fi
+ docker run --rm --network bridge -i \
+ -e DATABASE_URL \
+ -e MIGRATION_DATABASE_URL \
+ -v "$PWD/apps/api/migrations/mcp_version_lifecycle_controller_2026-07-15.sql:/tmp/mcp_version_lifecycle.sql:ro" \
+ "${HARBOR}/awoooi/api:${{ github.sha }}" python - <<'PY'
+ import asyncio
+ import os
+ from pathlib import Path
+
+ import asyncpg
+
+ MIGRATION = Path("/tmp/mcp_version_lifecycle.sql").read_text(
+ encoding="utf-8"
+ )
+ REQUIRED_TABLES = {
+ "awooop_mcp_version_lifecycle_run",
+ "awooop_mcp_version_observation",
+ }
+
+
+ def normalize_url(value: str) -> str:
+ return value.replace("postgresql+asyncpg://", "postgresql://", 1)
+
+
+ async def apply_and_verify_schema() -> None:
+ connection = await asyncpg.connect(
+ normalize_url(os.environ["MIGRATION_DATABASE_URL"]),
+ timeout=10,
+ )
+ try:
+ await connection.execute(MIGRATION)
+ rows = await connection.fetch(
+ """
+ SELECT relname, relrowsecurity, relforcerowsecurity
+ FROM pg_class
+ WHERE relname = ANY($1::text[])
+ """,
+ sorted(REQUIRED_TABLES),
+ )
+ table_state = {
+ str(row["relname"]): (
+ bool(row["relrowsecurity"]),
+ bool(row["relforcerowsecurity"]),
+ )
+ for row in rows
+ }
+ policy_count = await connection.fetchval(
+ """
+ SELECT COUNT(*)::int
+ FROM pg_policies
+ WHERE tablename = ANY($1::text[])
+ """,
+ sorted(REQUIRED_TABLES),
+ )
+ schema_verified = (
+ set(table_state) == REQUIRED_TABLES
+ and all(state == (True, True) for state in table_state.values())
+ and int(policy_count or 0) == 2
+ )
+ print("mcp_version_lifecycle_migration_applied=true")
+ print(f"mcp_version_lifecycle_schema_verified={str(schema_verified).lower()}")
+ print(f"mcp_version_lifecycle_rls_policy_count={int(policy_count or 0)}")
+ if not schema_verified:
+ raise RuntimeError("mcp_version_lifecycle_schema_verifier_failed")
+ finally:
+ await connection.close()
+
+
+ async def verify_runtime_role() -> None:
+ connection = await asyncpg.connect(
+ normalize_url(os.environ["DATABASE_URL"]),
+ timeout=10,
+ )
+ try:
+ async with connection.transaction():
+ await connection.execute(
+ "SELECT set_config('app.project_id', 'awoooi', TRUE)"
+ )
+ run_count = await connection.fetchval(
+ "SELECT COUNT(*)::int FROM awooop_mcp_version_lifecycle_run"
+ )
+ observation_count = await connection.fetchval(
+ "SELECT COUNT(*)::int FROM awooop_mcp_version_observation"
+ )
+ print("mcp_version_lifecycle_runtime_role_read_verified=true")
+ print(f"mcp_version_lifecycle_existing_run_count={int(run_count or 0)}")
+ print(
+ "mcp_version_lifecycle_existing_observation_count="
+ f"{int(observation_count or 0)}"
+ )
+ finally:
+ await connection.close()
+
+
+ async def main() -> None:
+ await apply_and_verify_schema()
+ await verify_runtime_role()
+
+
+ asyncio.run(main())
+ PY
+
# 2026-03-31 ogt: 移除中間通知
# 2026-03-31 ogt: P0-1 Secrets 自動注入 (ADR-035 強制)
diff --git a/apps/api/migrations/mcp_version_lifecycle_controller_2026-07-15.sql b/apps/api/migrations/mcp_version_lifecycle_controller_2026-07-15.sql
new file mode 100644
index 000000000..958fea9a5
--- /dev/null
+++ b/apps/api/migrations/mcp_version_lifecycle_controller_2026-07-15.sql
@@ -0,0 +1,141 @@
+-- Durable external MCP version lifecycle receipts.
+--
+-- Safety contract:
+-- * catalog metadata only; no package install, tool call, credential, raw page,
+-- request body, response body, or RAG content is stored;
+-- * additive schema only, so application rollback does not delete evidence;
+-- * every observation is bound to one run_id / trace_id / work_item_id;
+-- * tenant RLS remains fail-closed through app.project_id.
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS awooop_mcp_version_lifecycle_run (
+ run_id UUID PRIMARY KEY,
+ trace_id VARCHAR(128) NOT NULL,
+ work_item_id VARCHAR(128) NOT NULL,
+ project_id VARCHAR(64) NOT NULL
+ REFERENCES awooop_projects(project_id) ON DELETE RESTRICT,
+ trigger_kind VARCHAR(32) NOT NULL,
+ status VARCHAR(48) NOT NULL,
+ candidate_count INTEGER NOT NULL DEFAULT 0,
+ observed_count INTEGER NOT NULL DEFAULT 0,
+ baseline_count INTEGER NOT NULL DEFAULT 0,
+ changed_count INTEGER NOT NULL DEFAULT 0,
+ policy_blocked_count INTEGER NOT NULL DEFAULT 0,
+ error_count INTEGER NOT NULL DEFAULT 0,
+ receipt JSONB NOT NULL DEFAULT '{}'::jsonb,
+ started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+ ended_at TIMESTAMPTZ,
+
+ CONSTRAINT uix_mcp_version_lifecycle_trace
+ UNIQUE (project_id, trace_id),
+ CONSTRAINT chk_mcp_version_lifecycle_trigger
+ CHECK (trigger_kind IN ('startup','scheduled','operator_replay','test')),
+ CONSTRAINT chk_mcp_version_lifecycle_status
+ CHECK (status IN (
+ 'running',
+ 'completed_no_change',
+ 'completed_no_write_policy_blocked',
+ 'partial_degraded',
+ 'failed'
+ )),
+ CONSTRAINT chk_mcp_version_lifecycle_counts
+ CHECK (
+ candidate_count >= 0 AND observed_count >= 0 AND baseline_count >= 0
+ AND changed_count >= 0 AND policy_blocked_count >= 0
+ AND error_count >= 0
+ ),
+ CONSTRAINT chk_mcp_version_lifecycle_terminal_time
+ CHECK (
+ (status = 'running' AND ended_at IS NULL)
+ OR (status <> 'running' AND ended_at IS NOT NULL)
+ )
+);
+
+CREATE TABLE IF NOT EXISTS awooop_mcp_version_observation (
+ observation_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+ run_id UUID NOT NULL
+ REFERENCES awooop_mcp_version_lifecycle_run(run_id) ON DELETE RESTRICT,
+ trace_id VARCHAR(128) NOT NULL,
+ work_item_id VARCHAR(128) NOT NULL,
+ project_id VARCHAR(64) NOT NULL
+ REFERENCES awooop_projects(project_id) ON DELETE RESTRICT,
+ candidate_id VARCHAR(160) NOT NULL,
+ catalog_source VARCHAR(32) NOT NULL,
+ source_url TEXT NOT NULL,
+ source_url_sha256 VARCHAR(64) NOT NULL,
+ http_status INTEGER,
+ content_sha256 VARCHAR(64),
+ observed_version VARCHAR(128),
+ previous_version VARCHAR(128),
+ discovery_status VARCHAR(32) NOT NULL,
+ change_state VARCHAR(32) NOT NULL,
+ lifecycle_terminal VARCHAR(48) NOT NULL,
+ policy_decision JSONB NOT NULL DEFAULT '{}'::jsonb,
+ stage_receipts JSONB NOT NULL DEFAULT '{}'::jsonb,
+ error_class VARCHAR(128),
+ observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+
+ CONSTRAINT uix_mcp_version_observation_run_candidate
+ UNIQUE (run_id, candidate_id),
+ CONSTRAINT chk_mcp_version_catalog_source
+ CHECK (catalog_source IN ('smithery','mcp-so')),
+ CONSTRAINT chk_mcp_version_source_url_hash
+ CHECK (source_url_sha256 ~ '^[0-9a-f]{64}$'),
+ CONSTRAINT chk_mcp_version_content_hash
+ CHECK (content_sha256 IS NULL OR content_sha256 ~ '^[0-9a-f]{64}$'),
+ CONSTRAINT chk_mcp_version_discovery_status
+ CHECK (discovery_status IN ('observed','source_error','policy_blocked')),
+ CONSTRAINT chk_mcp_version_change_state
+ CHECK (change_state IN (
+ 'baseline_created','no_change','version_changed',
+ 'content_changed','source_unavailable'
+ )),
+ CONSTRAINT chk_mcp_version_terminal
+ CHECK (lifecycle_terminal IN (
+ 'no_write_baseline','no_write_no_change','no_write_policy_blocked',
+ 'no_write_source_error','replay_passed_canary_pending',
+ 'promoted','rolled_back'
+ ))
+);
+
+CREATE INDEX IF NOT EXISTS idx_mcp_version_lifecycle_recent
+ ON awooop_mcp_version_lifecycle_run (project_id, started_at DESC);
+
+CREATE INDEX IF NOT EXISTS idx_mcp_version_observation_candidate_recent
+ ON awooop_mcp_version_observation
+ (project_id, candidate_id, observed_at DESC);
+
+CREATE INDEX IF NOT EXISTS idx_mcp_version_observation_run
+ ON awooop_mcp_version_observation (project_id, run_id, observed_at);
+
+ALTER TABLE awooop_mcp_version_lifecycle_run ENABLE ROW LEVEL SECURITY;
+ALTER TABLE awooop_mcp_version_lifecycle_run FORCE ROW LEVEL SECURITY;
+ALTER TABLE awooop_mcp_version_observation ENABLE ROW LEVEL SECURITY;
+ALTER TABLE awooop_mcp_version_observation FORCE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS mcp_version_lifecycle_run_tenant
+ ON awooop_mcp_version_lifecycle_run;
+CREATE POLICY mcp_version_lifecycle_run_tenant
+ ON awooop_mcp_version_lifecycle_run
+ USING (project_id = current_setting('app.project_id', TRUE))
+ WITH CHECK (project_id = current_setting('app.project_id', TRUE));
+
+DROP POLICY IF EXISTS mcp_version_observation_tenant
+ ON awooop_mcp_version_observation;
+CREATE POLICY mcp_version_observation_tenant
+ ON awooop_mcp_version_observation
+ USING (project_id = current_setting('app.project_id', TRUE))
+ WITH CHECK (project_id = current_setting('app.project_id', TRUE));
+
+GRANT SELECT, INSERT, UPDATE ON awooop_mcp_version_lifecycle_run TO awooop_app;
+GRANT SELECT, INSERT ON awooop_mcp_version_observation TO awooop_app;
+GRANT SELECT, INSERT, UPDATE ON awooop_mcp_version_lifecycle_run TO awoooi;
+GRANT SELECT, INSERT ON awooop_mcp_version_observation TO awoooi;
+
+COMMENT ON TABLE awooop_mcp_version_lifecycle_run IS
+ 'Durable controlled-apply lifecycle run receipts; additive and rollback-safe.';
+COMMENT ON TABLE awooop_mcp_version_observation IS
+ 'Normalized external MCP catalog observations; raw external content is forbidden.';
+
+COMMIT;
diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py
index 19b8a0818..d718b9da3 100644
--- a/apps/api/src/core/config.py
+++ b/apps/api/src/core/config.py
@@ -831,6 +831,33 @@ class Settings(BaseSettings):
"dedicated signal worker."
),
)
+ ENABLE_MCP_VERSION_LIFECYCLE_WORKER: bool = Field(
+ default=True,
+ description=(
+ "True=run the fail-closed external MCP catalog observation and version "
+ "lifecycle controller in the dedicated signal worker. The controller "
+ "never installs or promotes an external artifact without immutable "
+ "supply-chain and replay receipts."
+ ),
+ )
+ MCP_VERSION_LIFECYCLE_INTERVAL_SECONDS: int = Field(
+ default=21_600,
+ ge=900,
+ le=86_400,
+ description="External MCP discovery cadence; defaults to every six hours.",
+ )
+ MCP_VERSION_LIFECYCLE_STARTUP_SLEEP_SECONDS: int = Field(
+ default=30,
+ ge=0,
+ le=900,
+ description="Delay before the first MCP version lifecycle observation run.",
+ )
+ MCP_VERSION_LIFECYCLE_SOURCE_TIMEOUT_SECONDS: int = Field(
+ default=12,
+ ge=3,
+ le=30,
+ description="Bounded timeout for one allowlisted Smithery or mcp.so read.",
+ )
AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS: int = Field(
default=600,
ge=60,
diff --git a/apps/api/src/jobs/mcp_version_lifecycle_job.py b/apps/api/src/jobs/mcp_version_lifecycle_job.py
new file mode 100644
index 000000000..a92a21afc
--- /dev/null
+++ b/apps/api/src/jobs/mcp_version_lifecycle_job.py
@@ -0,0 +1,102 @@
+"""Scheduled owner for the durable external MCP version lifecycle controller."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from typing import Any
+
+import structlog
+
+from src.core.config import settings
+from src.core.redis_client import get_redis
+from src.services.mcp_control_plane_service import load_mcp_control_plane_catalog
+from src.services.mcp_version_lifecycle_service import (
+ run_mcp_version_lifecycle_once,
+)
+
+logger = structlog.get_logger(__name__)
+
+_LOCK_KEY = "awoooi:mcp:version-lifecycle:leader"
+_LOCK_TTL_SECONDS = 900
+_RELEASE_LOCK_LUA = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+ return redis.call('DEL', KEYS[1])
+end
+return 0
+"""
+
+
+async def run_mcp_version_lifecycle_loop() -> None:
+ """Run once after startup, then every configured six-hour interval."""
+ interval = settings.MCP_VERSION_LIFECYCLE_INTERVAL_SECONDS
+ startup_sleep = settings.MCP_VERSION_LIFECYCLE_STARTUP_SLEEP_SECONDS
+ logger.info(
+ "mcp_version_lifecycle_loop_started",
+ interval_seconds=interval,
+ startup_sleep_seconds=startup_sleep,
+ owner="signal_worker",
+ lock_backend="redis",
+ )
+ await asyncio.sleep(startup_sleep)
+ trigger_kind = "startup"
+ while True:
+ try:
+ await run_mcp_version_lifecycle_if_leader(trigger_kind=trigger_kind)
+ except asyncio.CancelledError:
+ raise
+ except Exception as exc:
+ logger.exception(
+ "mcp_version_lifecycle_loop_error",
+ error_type=type(exc).__name__,
+ external_install_performed=False,
+ external_tool_call_performed=False,
+ external_rag_write_performed=False,
+ )
+ trigger_kind = "scheduled"
+ await asyncio.sleep(interval)
+
+
+async def run_mcp_version_lifecycle_if_leader(
+ *,
+ trigger_kind: str,
+) -> dict[str, Any]:
+ """Acquire one atomic lease so rolling replicas cannot duplicate a run."""
+ redis_client = get_redis()
+ lease_token = str(uuid.uuid4())
+ acquired = await redis_client.set(
+ _LOCK_KEY,
+ lease_token,
+ nx=True,
+ ex=_LOCK_TTL_SECONDS,
+ )
+ if not acquired:
+ logger.info(
+ "mcp_version_lifecycle_run_skipped_existing_leader",
+ trigger_kind=trigger_kind,
+ )
+ return {
+ "status": "skipped_existing_leader",
+ "trigger_kind": trigger_kind,
+ "external_write_performed": False,
+ }
+
+ try:
+ catalog = await asyncio.to_thread(load_mcp_control_plane_catalog)
+ return await run_mcp_version_lifecycle_once(
+ catalog,
+ trigger_kind=trigger_kind,
+ )
+ finally:
+ try:
+ await redis_client.eval(
+ _RELEASE_LOCK_LUA,
+ 1,
+ _LOCK_KEY,
+ lease_token,
+ )
+ except Exception as exc:
+ logger.warning(
+ "mcp_version_lifecycle_lock_release_failed",
+ error_type=type(exc).__name__,
+ )
diff --git a/apps/api/src/services/mcp_control_plane_service.py b/apps/api/src/services/mcp_control_plane_service.py
index f29d79f74..9f664b62f 100644
--- a/apps/api/src/services/mcp_control_plane_service.py
+++ b/apps/api/src/services/mcp_control_plane_service.py
@@ -20,6 +20,9 @@ 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.mcp_version_lifecycle_service import (
+ collect_mcp_version_lifecycle_readback,
+)
from src.services.snapshot_paths import default_operations_dir
logger = structlog.get_logger(__name__)
@@ -81,9 +84,7 @@ def build_mcp_control_plane_overview(
)
capability_rows = _dict_rows(catalog.get("capabilities"))
- capability_index = {
- str(row["capability_id"]): row for row in capability_rows
- }
+ 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"])
@@ -143,6 +144,17 @@ def build_mcp_control_plane_overview(
"sources": 0,
},
}
+ lifecycle_runtime = runtime_payload.get("version_lifecycle")
+ lifecycle_runtime = lifecycle_runtime if isinstance(lifecycle_runtime, dict) else {}
+ version_lifecycle = dict(catalog["version_lifecycle"])
+ version_lifecycle["runtime"] = lifecycle_runtime
+ lifecycle_status = str(lifecycle_runtime.get("status") or "not_requested")
+ if lifecycle_status == "ready" and lifecycle_runtime.get("latest_run"):
+ version_lifecycle["runtime_state"] = "running_fail_closed"
+ elif lifecycle_status == "pending_first_run":
+ version_lifecycle["runtime_state"] = "scheduled_pending_first_receipt"
+ elif lifecycle_status == "degraded":
+ version_lifecycle["runtime_state"] = "degraded_with_safe_next_action"
external_rows = [
row for row in capability_rows if row.get("source_type") == "external"
]
@@ -153,7 +165,10 @@ def build_mcp_control_plane_overview(
1
for row in products
if row["inventory_state"]
- not in {"no_committed_mcp_runtime_detected", "no_federated_mcp_inventory_readback"}
+ not in {
+ "no_committed_mcp_runtime_detected",
+ "no_federated_mcp_inventory_readback",
+ }
)
quarantined_products = sum(
1
@@ -162,7 +177,8 @@ def build_mcp_control_plane_overview(
or "remediation_required" in row["inventory_state"]
)
- active_blockers = set(_strings(catalog["version_lifecycle"].get("active_blockers")))
+ active_blockers = set(_strings(version_lifecycle.get("active_blockers")))
+ active_blockers.update(_strings(lifecycle_runtime.get("blockers")))
active_blockers.update(
{
"cross_product_runtime_federation_receipts_missing",
@@ -171,7 +187,11 @@ def build_mcp_control_plane_overview(
"distributed_gateway_rate_limit_receipt_missing",
}
)
- rag = runtime_payload.get("rag") if isinstance(runtime_payload.get("rag"), dict) else {}
+ 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"}:
@@ -208,7 +228,7 @@ def build_mcp_control_plane_overview(
"architecture": catalog["architecture"],
"security_policy": catalog["security_policy"],
"supply_chain_policy": catalog["supply_chain_policy"],
- "version_lifecycle": catalog["version_lifecycle"],
+ "version_lifecycle": version_lifecycle,
"runtime": runtime_payload,
"catalog_sources": catalog["source_policy"]["catalog_sources"],
"capabilities": capability_rows,
@@ -218,7 +238,7 @@ def build_mcp_control_plane_overview(
"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",
+ "promote_first_internally_mirrored_candidate_through_replay_shadow_and_canary",
"mirror_and_pin_each_approved_external_candidate_before_shadow",
"wire_quarantine_scan_citation_verifier_and_governed_rag_promotion",
],
@@ -227,6 +247,7 @@ def build_mcp_control_plane_overview(
"public_response_redacted_aggregate_only": True,
"mutation_endpoint_exposed": False,
"operator_auth_required_for_future_mutation": True,
+ "normalized_version_receipt_write_allowed": True,
"external_install_allowed": False,
"external_tool_call_allowed": False,
"external_rag_write_allowed": False,
@@ -238,7 +259,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, and RAG stats",
+ "runtime": "live provider registry, MCP tool registry, gateway audit aggregate, RAG stats, and durable version lifecycle receipts",
},
}
@@ -246,6 +267,9 @@ 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()
+ if lifecycle_payload.get("status") == "degraded":
+ blockers.append("version_lifecycle_runtime_readback_degraded")
try:
provider_registry = get_provider_registry()
tool_registry = get_mcp_tool_registry()
@@ -351,6 +375,7 @@ async def collect_mcp_control_plane_runtime() -> dict[str, Any]:
"provider_registry": provider_payload,
"gateway_audit_24h": gateway_payload,
"rag": rag_payload,
+ "version_lifecycle": lifecycle_payload,
"federated_product_runtime_receipt_count": 0,
"federated_product_runtime_receipt_expected": 12,
"blockers": sorted(blockers),
@@ -361,7 +386,10 @@ 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":
+ 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")
@@ -397,7 +425,12 @@ def _validate_catalog(payload: dict[str, Any], label: str) -> None:
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")
+ 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"
diff --git a/apps/api/src/services/mcp_version_lifecycle_service.py b/apps/api/src/services/mcp_version_lifecycle_service.py
new file mode 100644
index 000000000..d1ecf328a
--- /dev/null
+++ b/apps/api/src/services/mcp_version_lifecycle_service.py
@@ -0,0 +1,935 @@
+"""Durable, fail-closed lifecycle receipts for external MCP candidates.
+
+The controller reads only committed Smithery and mcp.so catalog URLs. It stores
+normalized metadata and hashes, never raw external content. A version/content
+change cannot reach replay, canary, rollout, or promotion unless immutable
+supply-chain receipts and a registered internal replay adapter are present.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import html
+import json
+import re
+import uuid
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from html.parser import HTMLParser
+from typing import Any, Protocol
+from urllib.parse import urljoin, urlparse
+
+import httpx
+import structlog
+from sqlalchemy import text
+
+from src.core.config import settings
+from src.core.http_client import get_general_client
+from src.db.base import get_db_context
+
+logger = structlog.get_logger(__name__)
+
+PROJECT_ID = "awoooi"
+SCHEMA_VERSION = "mcp_version_lifecycle_runtime_v1"
+_MAX_SOURCE_BYTES = 2_000_000
+_MAX_REDIRECTS = 2
+_MAX_CONCURRENT_FETCHES = 3
+_SOURCE_HOSTS = {
+ "mcp-so": frozenset({"mcp.so", "www.mcp.so"}),
+ "smithery": frozenset({"smithery.ai", "www.smithery.ai"}),
+}
+_VERSION_PATTERNS = (
+ re.compile(
+ rb"""(?ix)["'](?:version|latestVersion|packageVersion)["']\s*[:=]\s*["']v?"""
+ rb"""([0-9]+\.[0-9]+\.[0-9]+(?:[-+][0-9A-Za-z.-]+)?)["']"""
+ ),
+ re.compile(
+ rb"""(?ix)data-(?:version|package-version)=["']v?"""
+ rb"""([0-9]+\.[0-9]+\.[0-9]+(?:[-+][0-9A-Za-z.-]+)?)["']"""
+ ),
+)
+
+
+@dataclass(frozen=True, slots=True)
+class CatalogFetchReceipt:
+ """Bounded source observation. The raw body is never persisted."""
+
+ status: str
+ source_url: str
+ http_status: int | None = None
+ body: bytes = b""
+ error_class: str | None = None
+ redirect_count: int = 0
+
+
+CatalogFetcher = Callable[[str, str, int], Awaitable[CatalogFetchReceipt]]
+
+
+class _StableMetadataParser(HTMLParser):
+ """Extract stable public metadata while ignoring randomized page sections."""
+
+ def __init__(self) -> None:
+ super().__init__(convert_charrefs=True)
+ self.in_title = False
+ self.title_parts: list[str] = []
+ self.metadata: dict[str, str] = {}
+
+ def handle_starttag(
+ self,
+ tag: str,
+ attrs: list[tuple[str, str | None]],
+ ) -> None:
+ attributes = {key.lower(): value or "" for key, value in attrs}
+ if tag.lower() == "title":
+ self.in_title = True
+ return
+ if tag.lower() == "meta":
+ key = (attributes.get("name") or attributes.get("property") or "").lower()
+ if key in {"description", "og:title", "og:description"}:
+ self.metadata[key] = attributes.get("content", "")[:4_000]
+ return
+ if tag.lower() == "link" and "canonical" in attributes.get("rel", "").lower():
+ self.metadata["canonical"] = attributes.get("href", "")[:2_000]
+
+ def handle_endtag(self, tag: str) -> None:
+ if tag.lower() == "title":
+ self.in_title = False
+
+ def handle_data(self, data: str) -> None:
+ if self.in_title:
+ self.title_parts.append(data)
+
+ def stable_fields(self) -> dict[str, str]:
+ title = _clean_metadata_text(" ".join(self.title_parts))
+ fields = {
+ key: _clean_metadata_text(value)
+ for key, value in self.metadata.items()
+ if _clean_metadata_text(value)
+ }
+ if title:
+ fields["title"] = title
+ return fields
+
+
+class LifecycleRepository(Protocol):
+ """Persistence boundary used by production PostgreSQL and unit-test fakes."""
+
+ async def start_run(self, payload: dict[str, Any]) -> None:
+ ...
+
+ async def latest_observations(self) -> dict[str, dict[str, Any]]:
+ ...
+
+ async def save_observation(self, payload: dict[str, Any]) -> None:
+ ...
+
+ async def finish_run(self, payload: dict[str, Any]) -> None:
+ ...
+
+
+class PostgresLifecycleRepository:
+ """PostgreSQL repository with project-scoped RLS context."""
+
+ async def start_run(self, payload: dict[str, Any]) -> None:
+ async with get_db_context(PROJECT_ID) as db:
+ await db.execute(
+ text(
+ """
+ INSERT INTO awooop_mcp_version_lifecycle_run (
+ run_id, trace_id, work_item_id, project_id, trigger_kind,
+ status, candidate_count, receipt, started_at
+ ) VALUES (
+ :run_id, :trace_id, :work_item_id, :project_id,
+ :trigger_kind, 'running', :candidate_count,
+ CAST(:receipt AS jsonb), :started_at
+ )
+ """
+ ),
+ {
+ **payload,
+ "receipt": _json(payload["receipt"]),
+ },
+ )
+
+ async def latest_observations(self) -> dict[str, dict[str, Any]]:
+ async with get_db_context(PROJECT_ID) as db:
+ result = await db.execute(
+ text(
+ """
+ SELECT DISTINCT ON (candidate_id)
+ candidate_id, observed_version, content_sha256,
+ discovery_status, observed_at
+ FROM awooop_mcp_version_observation
+ WHERE project_id = :project_id
+ AND discovery_status = 'observed'
+ ORDER BY candidate_id, observed_at DESC
+ """
+ ),
+ {"project_id": PROJECT_ID},
+ )
+ return {
+ str(row["candidate_id"]): dict(row) for row in result.mappings().all()
+ }
+
+ async def save_observation(self, payload: dict[str, Any]) -> None:
+ async with get_db_context(PROJECT_ID) as db:
+ await db.execute(
+ text(
+ """
+ INSERT INTO awooop_mcp_version_observation (
+ run_id, trace_id, work_item_id, project_id, candidate_id,
+ catalog_source, source_url, source_url_sha256,
+ http_status, content_sha256, observed_version,
+ previous_version, discovery_status, change_state,
+ lifecycle_terminal, policy_decision, stage_receipts,
+ error_class, observed_at
+ ) VALUES (
+ :run_id, :trace_id, :work_item_id, :project_id,
+ :candidate_id, :catalog_source, :source_url,
+ :source_url_sha256, :http_status, :content_sha256,
+ :observed_version, :previous_version, :discovery_status,
+ :change_state, :lifecycle_terminal,
+ CAST(:policy_decision AS jsonb),
+ CAST(:stage_receipts AS jsonb), :error_class,
+ :observed_at
+ )
+ """
+ ),
+ {
+ **payload,
+ "policy_decision": _json(payload["policy_decision"]),
+ "stage_receipts": _json(payload["stage_receipts"]),
+ },
+ )
+
+ async def finish_run(self, payload: dict[str, Any]) -> None:
+ async with get_db_context(PROJECT_ID) as db:
+ await db.execute(
+ text(
+ """
+ UPDATE awooop_mcp_version_lifecycle_run
+ SET status = :status,
+ observed_count = :observed_count,
+ baseline_count = :baseline_count,
+ changed_count = :changed_count,
+ policy_blocked_count = :policy_blocked_count,
+ error_count = :error_count,
+ receipt = CAST(:receipt AS jsonb),
+ ended_at = :ended_at
+ WHERE run_id = :run_id AND project_id = :project_id
+ """
+ ),
+ {
+ **payload,
+ "receipt": _json(payload["receipt"]),
+ },
+ )
+
+
+async def fetch_catalog_source(
+ source_url: str,
+ catalog_source: str,
+ timeout_seconds: int,
+) -> CatalogFetchReceipt:
+ """Fetch one allowlisted catalog URL without cross-domain redirects."""
+ allowed_hosts = _SOURCE_HOSTS.get(catalog_source)
+ if not allowed_hosts or not _is_allowed_url(source_url, allowed_hosts):
+ return CatalogFetchReceipt(
+ status="policy_blocked",
+ source_url=source_url,
+ error_class="catalog_url_not_allowlisted",
+ )
+
+ client = await get_general_client()
+ current_url = source_url
+ for redirect_count in range(_MAX_REDIRECTS + 1):
+ try:
+ async with client.stream(
+ "GET",
+ current_url,
+ follow_redirects=False,
+ timeout=httpx.Timeout(float(timeout_seconds), connect=5.0),
+ headers={
+ "Accept": "application/json,text/html,text/plain;q=0.9",
+ "User-Agent": "awoooi-mcp-version-lifecycle/1.0",
+ },
+ ) as response:
+ if response.status_code in {301, 302, 303, 307, 308}:
+ location = response.headers.get("location")
+ if not location or redirect_count >= _MAX_REDIRECTS:
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ http_status=response.status_code,
+ error_class="redirect_limit_or_location_missing",
+ redirect_count=redirect_count,
+ )
+ redirected = urljoin(current_url, location)
+ if not _is_allowed_url(redirected, allowed_hosts):
+ return CatalogFetchReceipt(
+ status="policy_blocked",
+ source_url=current_url,
+ http_status=response.status_code,
+ error_class="cross_domain_redirect_blocked",
+ redirect_count=redirect_count,
+ )
+ current_url = redirected
+ continue
+
+ if response.status_code != 200:
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ http_status=response.status_code,
+ error_class=f"http_{response.status_code}",
+ redirect_count=redirect_count,
+ )
+
+ content_type = response.headers.get("content-type", "").lower()
+ if not any(
+ marker in content_type
+ for marker in (
+ "application/json",
+ "application/octet-stream",
+ "text/html",
+ "text/plain",
+ )
+ ):
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ http_status=response.status_code,
+ error_class="unsupported_content_type",
+ redirect_count=redirect_count,
+ )
+
+ content_length = response.headers.get("content-length")
+ if content_length and int(content_length) > _MAX_SOURCE_BYTES:
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ http_status=response.status_code,
+ error_class="source_payload_too_large",
+ redirect_count=redirect_count,
+ )
+
+ body = bytearray()
+ async for chunk in response.aiter_bytes():
+ body.extend(chunk)
+ if len(body) > _MAX_SOURCE_BYTES:
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ http_status=response.status_code,
+ error_class="source_payload_too_large",
+ redirect_count=redirect_count,
+ )
+ return CatalogFetchReceipt(
+ status="ok",
+ source_url=current_url,
+ http_status=response.status_code,
+ body=bytes(body),
+ redirect_count=redirect_count,
+ )
+ except httpx.TimeoutException:
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ error_class="source_timeout",
+ redirect_count=redirect_count,
+ )
+ except (httpx.RequestError, ValueError):
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ error_class="source_network_error",
+ redirect_count=redirect_count,
+ )
+
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=current_url,
+ error_class="redirect_loop",
+ redirect_count=_MAX_REDIRECTS,
+ )
+
+
+async def run_mcp_version_lifecycle_once(
+ catalog: dict[str, Any],
+ *,
+ trigger_kind: str = "scheduled",
+ repository: LifecycleRepository | None = None,
+ fetcher: CatalogFetcher | None = None,
+ generated_at: datetime | None = None,
+) -> dict[str, Any]:
+ """Observe every eligible external candidate and persist terminal receipts."""
+ repository = repository or PostgresLifecycleRepository()
+ fetcher = fetcher or fetch_catalog_source
+ now = generated_at or datetime.now(UTC)
+ run_id = uuid.uuid4()
+ trace_id = f"mcp-version-{run_id}"
+ work_item_id = f"MCP-VERSION-{now:%Y%m%dT%H%M%SZ}"
+ candidates = _eligible_candidates(catalog)
+ start_receipt = {
+ "schema_version": SCHEMA_VERSION,
+ "controller": "signal_worker_scheduled_fail_closed",
+ "raw_external_content_stored": False,
+ "external_install_performed": False,
+ "external_tool_call_performed": False,
+ "external_rag_write_performed": False,
+ "github_used": False,
+ "source_host_allowlist": sorted(
+ host for hosts in _SOURCE_HOSTS.values() for host in hosts
+ ),
+ }
+ await repository.start_run(
+ {
+ "run_id": run_id,
+ "trace_id": trace_id,
+ "work_item_id": work_item_id,
+ "project_id": PROJECT_ID,
+ "trigger_kind": trigger_kind,
+ "candidate_count": len(candidates),
+ "receipt": start_receipt,
+ "started_at": now,
+ }
+ )
+
+ counts = {
+ "observed_count": 0,
+ "baseline_count": 0,
+ "changed_count": 0,
+ "policy_blocked_count": 0,
+ "error_count": 0,
+ }
+ terminals: dict[str, int] = {}
+ try:
+ previous = await repository.latest_observations()
+
+ semaphore = asyncio.Semaphore(_MAX_CONCURRENT_FETCHES)
+
+ async def fetch_candidate(
+ candidate: dict[str, Any],
+ ) -> tuple[dict[str, Any], CatalogFetchReceipt]:
+ source_url = str(candidate["catalog_url"])
+ source_name = str(candidate["catalog_source"])
+ async with semaphore:
+ try:
+ fetched = await fetcher(
+ source_url,
+ source_name,
+ settings.MCP_VERSION_LIFECYCLE_SOURCE_TIMEOUT_SECONDS,
+ )
+ except Exception:
+ fetched = CatalogFetchReceipt(
+ status="error",
+ source_url=source_url,
+ error_class="source_fetcher_error",
+ )
+ return candidate, fetched
+
+ fetched_candidates = await asyncio.gather(
+ *(fetch_candidate(candidate) for candidate in candidates)
+ )
+ for candidate, fetched in fetched_candidates:
+ observation = build_candidate_observation(
+ candidate,
+ fetched=fetched,
+ previous=previous.get(str(candidate["capability_id"])),
+ run_id=run_id,
+ trace_id=trace_id,
+ work_item_id=work_item_id,
+ observed_at=now,
+ )
+ await repository.save_observation(observation)
+ _increment_counts(counts, observation)
+ terminal = str(observation["lifecycle_terminal"])
+ terminals[terminal] = terminals.get(terminal, 0) + 1
+
+ status = _run_status(counts)
+ ended_at = datetime.now(UTC)
+ final_receipt = {
+ **start_receipt,
+ "status": status,
+ "candidate_count": len(candidates),
+ **counts,
+ "terminal_counts": terminals,
+ "independent_post_verifier": {
+ "status": "verified",
+ "mutation_scope": "normalized_metadata_rows_only",
+ "raw_external_content_stored": False,
+ "external_install_performed": False,
+ "external_tool_call_performed": False,
+ "external_rag_write_performed": False,
+ "production_upgrade_performed": False,
+ },
+ "rollback": {
+ "status": "no_write_terminal",
+ "schema_rollback": "additive_tables_preserved",
+ },
+ }
+ await repository.finish_run(
+ {
+ "run_id": run_id,
+ "project_id": PROJECT_ID,
+ "status": status,
+ **counts,
+ "receipt": final_receipt,
+ "ended_at": ended_at,
+ }
+ )
+ logger.info(
+ "mcp_version_lifecycle_run_completed",
+ run_id=str(run_id),
+ status=status,
+ candidate_count=len(candidates),
+ **counts,
+ )
+ return {
+ "schema_version": SCHEMA_VERSION,
+ "run_id": str(run_id),
+ "trace_id": trace_id,
+ "work_item_id": work_item_id,
+ "status": status,
+ "candidate_count": len(candidates),
+ **counts,
+ "terminal_counts": terminals,
+ }
+ except Exception:
+ ended_at = datetime.now(UTC)
+ failure_receipt = {
+ **start_receipt,
+ "status": "failed",
+ "error_class": "controller_execution_error",
+ "independent_post_verifier": {
+ "status": "partial",
+ "external_install_performed": False,
+ "external_tool_call_performed": False,
+ "external_rag_write_performed": False,
+ "production_upgrade_performed": False,
+ },
+ "rollback": {"status": "no_write_terminal"},
+ }
+ try:
+ await repository.finish_run(
+ {
+ "run_id": run_id,
+ "project_id": PROJECT_ID,
+ "status": "failed",
+ **counts,
+ "receipt": failure_receipt,
+ "ended_at": ended_at,
+ }
+ )
+ except Exception:
+ logger.exception(
+ "mcp_version_lifecycle_failure_receipt_write_failed",
+ run_id=str(run_id),
+ )
+ raise
+
+
+def build_candidate_observation(
+ candidate: dict[str, Any],
+ *,
+ fetched: CatalogFetchReceipt,
+ previous: dict[str, Any] | None,
+ run_id: uuid.UUID,
+ trace_id: str,
+ work_item_id: str,
+ observed_at: datetime,
+) -> dict[str, Any]:
+ """Normalize, diff, gate, and terminalize one source observation."""
+ candidate_id = str(candidate["capability_id"])
+ observed_version = _extract_unambiguous_version(fetched.body)
+ content_sha256, fingerprint_scope = _stable_catalog_fingerprint(
+ fetched.body,
+ source_url=fetched.source_url,
+ observed_version=observed_version,
+ )
+ previous_version = (
+ str(previous.get("observed_version"))
+ if previous and previous.get("observed_version")
+ else None
+ )
+ change_state = _change_state(
+ fetched=fetched,
+ content_sha256=content_sha256,
+ observed_version=observed_version,
+ previous=previous,
+ )
+ missing_receipts = _missing_supply_chain_receipts(candidate)
+ supply_chain_ready = not missing_receipts
+
+ if fetched.status != "ok":
+ discovery_status = (
+ "policy_blocked" if fetched.status == "policy_blocked" else "source_error"
+ )
+ terminal = "no_write_source_error"
+ replay_status = "not_started_source_unavailable"
+ elif change_state == "baseline_created":
+ discovery_status = "observed"
+ terminal = "no_write_baseline"
+ replay_status = "not_required_baseline_only"
+ elif change_state == "no_change":
+ discovery_status = "observed"
+ terminal = "no_write_no_change"
+ replay_status = "not_required_no_diff"
+ else:
+ discovery_status = "observed"
+ terminal = "no_write_policy_blocked"
+ replay_status = (
+ "blocked_registered_internal_replay_adapter_missing"
+ if supply_chain_ready
+ else "blocked_immutable_supply_chain_receipts_missing"
+ )
+
+ policy_decision = {
+ "risk_tier": str(candidate.get("risk_tier") or "high"),
+ "deployment_allowed_by_catalog": candidate.get("deployment_allowed") is True,
+ "immutable_supply_chain_ready": supply_chain_ready,
+ "missing_receipts": missing_receipts,
+ "registered_internal_replay_adapter": False,
+ "external_install_allowed": False,
+ "production_upgrade_allowed": False,
+ "rag_write_allowed": False,
+ }
+ stage_receipts = {
+ "sensor_source_receipt": {
+ "status": fetched.status,
+ "http_status": fetched.http_status,
+ "redirect_count": fetched.redirect_count,
+ "content_sha256": content_sha256,
+ "content_fingerprint_scope": fingerprint_scope,
+ "raw_body_stored": False,
+ },
+ "normalized_asset_identity": {
+ "candidate_id": candidate_id,
+ "catalog_source": str(candidate["catalog_source"]),
+ "source_url_sha256": hashlib.sha256(
+ fetched.source_url.encode("utf-8")
+ ).hexdigest(),
+ },
+ "source_of_truth_diff": {
+ "change_state": change_state,
+ "previous_version": previous_version,
+ "observed_version": observed_version,
+ "previous_content_sha256": (
+ str(previous.get("content_sha256"))
+ if previous and previous.get("content_sha256")
+ else None
+ ),
+ "observed_content_sha256": content_sha256,
+ },
+ "decision": {
+ "candidate_action": "observe_only_until_all_gates_pass",
+ "reason": replay_status,
+ },
+ "risk_policy": policy_decision,
+ "check_mode": {
+ "status": replay_status,
+ "external_process_started": False,
+ "artifact_downloaded": False,
+ },
+ "bounded_execution": {
+ "status": "not_executed",
+ "external_tool_call_performed": False,
+ "production_write_performed": False,
+ },
+ "post_verifier_and_rollback": {
+ "status": "verified_no_write_terminal",
+ "terminal": terminal,
+ "rollback_required": False,
+ "rollback_reason": "no_runtime_mutation_performed",
+ },
+ "learning_writeback": {
+ "status": "committed_with_observation_row",
+ "target": "awooop_mcp_version_observation",
+ "rag_or_km_write_performed": False,
+ },
+ }
+ return {
+ "run_id": run_id,
+ "trace_id": trace_id,
+ "work_item_id": work_item_id,
+ "project_id": PROJECT_ID,
+ "candidate_id": candidate_id,
+ "catalog_source": str(candidate["catalog_source"]),
+ "source_url": fetched.source_url,
+ "source_url_sha256": hashlib.sha256(
+ fetched.source_url.encode("utf-8")
+ ).hexdigest(),
+ "http_status": fetched.http_status,
+ "content_sha256": content_sha256,
+ "observed_version": observed_version,
+ "previous_version": previous_version,
+ "discovery_status": discovery_status,
+ "change_state": change_state,
+ "lifecycle_terminal": terminal,
+ "policy_decision": policy_decision,
+ "stage_receipts": stage_receipts,
+ "error_class": fetched.error_class,
+ "observed_at": observed_at,
+ }
+
+
+async def collect_mcp_version_lifecycle_readback() -> dict[str, Any]:
+ """Return the newest durable run and bounded per-candidate terminal states."""
+ try:
+ async with get_db_context(PROJECT_ID) as db:
+ run_result = await db.execute(
+ text(
+ """
+ SELECT
+ run_id, trace_id, work_item_id, trigger_kind, status,
+ candidate_count, observed_count, baseline_count,
+ changed_count, policy_blocked_count, error_count,
+ started_at, ended_at,
+ started_at >= NOW() - make_interval(
+ secs => :freshness_seconds
+ ) AS fresh
+ FROM awooop_mcp_version_lifecycle_run
+ WHERE project_id = :project_id
+ ORDER BY started_at DESC
+ LIMIT 1
+ """
+ ),
+ {
+ "project_id": PROJECT_ID,
+ "freshness_seconds": (
+ settings.MCP_VERSION_LIFECYCLE_INTERVAL_SECONDS * 2
+ ),
+ },
+ )
+ latest = run_result.mappings().one_or_none()
+ if latest is None:
+ return {
+ "status": "pending_first_run",
+ "controller_configured": True,
+ "controller_owner": "signal_worker",
+ "interval_seconds": settings.MCP_VERSION_LIFECYCLE_INTERVAL_SECONDS,
+ "latest_run": None,
+ "observations": [],
+ "blockers": ["first_runtime_receipt_pending"],
+ }
+ observation_result = await db.execute(
+ text(
+ """
+ SELECT candidate_id, catalog_source, observed_version,
+ discovery_status, change_state, lifecycle_terminal,
+ error_class, observed_at
+ FROM awooop_mcp_version_observation
+ WHERE project_id = :project_id AND run_id = :run_id
+ ORDER BY candidate_id
+ LIMIT 64
+ """
+ ),
+ {"project_id": PROJECT_ID, "run_id": latest["run_id"]},
+ )
+ observations = [
+ {
+ "candidate_id": str(row["candidate_id"]),
+ "catalog_source": str(row["catalog_source"]),
+ "observed_version": row["observed_version"],
+ "discovery_status": str(row["discovery_status"]),
+ "change_state": str(row["change_state"]),
+ "lifecycle_terminal": str(row["lifecycle_terminal"]),
+ "error_class": row["error_class"],
+ "observed_at": _iso(row["observed_at"]),
+ }
+ for row in observation_result.mappings().all()
+ ]
+
+ run_payload = {
+ "run_id": str(latest["run_id"]),
+ "trace_id": str(latest["trace_id"]),
+ "work_item_id": str(latest["work_item_id"]),
+ "trigger_kind": str(latest["trigger_kind"]),
+ "status": str(latest["status"]),
+ "candidate_count": int(latest["candidate_count"]),
+ "observed_count": int(latest["observed_count"]),
+ "baseline_count": int(latest["baseline_count"]),
+ "changed_count": int(latest["changed_count"]),
+ "policy_blocked_count": int(latest["policy_blocked_count"]),
+ "error_count": int(latest["error_count"]),
+ "started_at": _iso(latest["started_at"]),
+ "ended_at": _iso(latest["ended_at"]),
+ "fresh": bool(latest["fresh"]),
+ }
+ blockers = []
+ if not run_payload["fresh"]:
+ blockers.append("version_lifecycle_receipt_stale")
+ if run_payload["error_count"]:
+ blockers.append("catalog_source_observation_failures")
+ if run_payload["policy_blocked_count"]:
+ blockers.append("external_candidates_not_promotion_ready")
+ return {
+ "status": "ready" if run_payload["fresh"] else "degraded",
+ "controller_configured": True,
+ "controller_owner": "signal_worker",
+ "lock_backend": "redis_atomic_lease",
+ "interval_seconds": settings.MCP_VERSION_LIFECYCLE_INTERVAL_SECONDS,
+ "latest_run": run_payload,
+ "observations": observations,
+ "blockers": blockers,
+ }
+ except Exception as exc:
+ logger.warning(
+ "mcp_version_lifecycle_readback_failed",
+ error_type=type(exc).__name__,
+ )
+ return {
+ "status": "degraded",
+ "controller_configured": True,
+ "controller_owner": "signal_worker",
+ "interval_seconds": settings.MCP_VERSION_LIFECYCLE_INTERVAL_SECONDS,
+ "latest_run": None,
+ "observations": [],
+ "blockers": ["version_lifecycle_schema_or_database_readback_failed"],
+ }
+
+
+def _eligible_candidates(catalog: dict[str, Any]) -> list[dict[str, Any]]:
+ rows = catalog.get("capabilities")
+ if not isinstance(rows, list):
+ return []
+ candidates = [
+ row
+ for row in rows
+ if isinstance(row, dict)
+ and row.get("source_type") == "external"
+ and row.get("decision") != "rejected"
+ and row.get("catalog_source") in _SOURCE_HOSTS
+ and row.get("catalog_url")
+ ]
+ return sorted(candidates, key=lambda row: str(row["capability_id"]))
+
+
+def _is_allowed_url(url: str, allowed_hosts: frozenset[str]) -> bool:
+ parsed = urlparse(url)
+ try:
+ port = parsed.port
+ except ValueError:
+ return False
+ return (
+ parsed.scheme == "https"
+ and (parsed.hostname or "").lower() in allowed_hosts
+ and parsed.username is None
+ and parsed.password is None
+ and port in {None, 443}
+ )
+
+
+def _extract_unambiguous_version(body: bytes) -> str | None:
+ versions = {
+ match.group(1).decode("ascii")
+ for pattern in _VERSION_PATTERNS
+ for match in pattern.finditer(body)
+ }
+ return next(iter(versions)) if len(versions) == 1 else None
+
+
+def _stable_catalog_fingerprint(
+ body: bytes,
+ *,
+ source_url: str,
+ observed_version: str | None,
+) -> tuple[str | None, str]:
+ """Hash stable identity metadata instead of randomized recommendation HTML."""
+ if not body:
+ return None, "not_available"
+
+ parser = _StableMetadataParser()
+ try:
+ parser.feed(body.decode("utf-8", errors="ignore"))
+ except (UnicodeError, ValueError):
+ stable_fields: dict[str, str] = {}
+ else:
+ stable_fields = parser.stable_fields()
+ if stable_fields:
+ normalized = {
+ "source_url": source_url,
+ "observed_version": observed_version,
+ "metadata": stable_fields,
+ }
+ return hashlib.sha256(_json(normalized).encode("utf-8")).hexdigest(), (
+ "title_description_canonical_version"
+ )
+ return hashlib.sha256(body).hexdigest(), "bounded_raw_fallback"
+
+
+def _clean_metadata_text(value: str) -> str:
+ return " ".join(html.unescape(value).split())[:4_000]
+
+
+def _change_state(
+ *,
+ fetched: CatalogFetchReceipt,
+ content_sha256: str | None,
+ observed_version: str | None,
+ previous: dict[str, Any] | None,
+) -> str:
+ if fetched.status != "ok":
+ return "source_unavailable"
+ if previous is None:
+ return "baseline_created"
+ previous_version = previous.get("observed_version")
+ if previous_version and observed_version and previous_version != observed_version:
+ return "version_changed"
+ if previous.get("content_sha256") != content_sha256:
+ return "content_changed"
+ return "no_change"
+
+
+def _missing_supply_chain_receipts(candidate: dict[str, Any]) -> list[str]:
+ receipts = candidate.get("supply_chain_receipts")
+ receipts = receipts if isinstance(receipts, dict) else {}
+ required = (
+ "exact_version",
+ "immutable_artifact_digest",
+ "internal_mirror_ref",
+ "sbom_ref",
+ "signature_or_provenance_ref",
+ "license_decision",
+ "vulnerability_decision",
+ "rollback_artifact_ref",
+ )
+ missing = [key for key in required if not receipts.get(key)]
+ if candidate.get("deployment_allowed") is not True:
+ missing.append("deployment_policy_not_enabled")
+ return missing
+
+
+def _increment_counts(counts: dict[str, int], observation: dict[str, Any]) -> None:
+ if observation["discovery_status"] == "observed":
+ counts["observed_count"] += 1
+ else:
+ counts["error_count"] += 1
+ if observation["change_state"] == "baseline_created":
+ counts["baseline_count"] += 1
+ if observation["change_state"] in {"version_changed", "content_changed"}:
+ counts["changed_count"] += 1
+ if observation["policy_decision"]["immutable_supply_chain_ready"] is False:
+ counts["policy_blocked_count"] += 1
+
+
+def _run_status(counts: dict[str, int]) -> str:
+ if counts["error_count"]:
+ return "partial_degraded"
+ if counts["policy_blocked_count"]:
+ return "completed_no_write_policy_blocked"
+ if counts["changed_count"] == 0:
+ return "completed_no_change"
+ return "completed_no_write_policy_blocked"
+
+
+def _json(payload: dict[str, Any]) -> str:
+ return json.dumps(
+ payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+ )
+
+
+def _iso(value: Any) -> str | None:
+ return value.isoformat() if hasattr(value, "isoformat") else None
diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py
index cd82cb069..1a16c0a39 100644
--- a/apps/api/src/workers/signal_worker.py
+++ b/apps/api/src/workers/signal_worker.py
@@ -623,6 +623,7 @@ async def _main() -> None:
backup_restore_legacy_backfill_task: asyncio.Task[Any] | None = None
security_maintenance_tasks: list[asyncio.Task[Any]] = []
agent99_controlled_dispatch_reconciler_task: asyncio.Task[Any] | None = None
+ mcp_version_lifecycle_task: asyncio.Task[Any] | None = None
if settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER:
from src.jobs.awooop_ansible_candidate_backfill_job import (
run_awooop_ansible_candidate_backfill_loop,
@@ -688,6 +689,30 @@ async def _main() -> None:
task_count=len(security_maintenance_tasks),
)
+ if settings.ENABLE_MCP_VERSION_LIFECYCLE_WORKER:
+ from src.jobs.mcp_version_lifecycle_job import (
+ run_mcp_version_lifecycle_loop,
+ )
+
+ mcp_version_lifecycle_task = asyncio.create_task(
+ run_mcp_version_lifecycle_loop(),
+ name="run_mcp_version_lifecycle_loop",
+ )
+ logger.info(
+ "signal_worker_mcp_version_lifecycle_started",
+ interval_seconds=settings.MCP_VERSION_LIFECYCLE_INTERVAL_SECONDS,
+ startup_sleep_seconds=(
+ settings.MCP_VERSION_LIFECYCLE_STARTUP_SLEEP_SECONDS
+ ),
+ external_install_allowed=False,
+ external_rag_write_allowed=False,
+ )
+ else:
+ logger.warning(
+ "signal_worker_mcp_version_lifecycle_disabled",
+ owner="signal_worker",
+ )
+
# 2026-07-14 Codex v1.4: Agent99 reconciliation has one runtime owner.
# The API reserves its single DB connection for requests, so durable
# dispatch completion belongs to this standalone worker after the CD-owned
@@ -769,7 +794,16 @@ async def _main() -> None:
await agent99_controlled_dispatch_reconciler_task
except asyncio.CancelledError:
pass
+ if mcp_version_lifecycle_task is not None:
+ mcp_version_lifecycle_task.cancel()
+ try:
+ await mcp_version_lifecycle_task
+ except asyncio.CancelledError:
+ pass
await worker.stop()
+ from src.core.http_client import close_general_client
+
+ await close_general_client()
await close_worker_redis_pool() # 關閉 Worker 專屬連線
await close_redis_pool()
await close_db() # 關閉 PostgreSQL 連線池
diff --git a/apps/api/tests/test_mcp_control_plane_api.py b/apps/api/tests/test_mcp_control_plane_api.py
index 45a9b61f4..c300dd028 100644
--- a/apps/api/tests/test_mcp_control_plane_api.py
+++ b/apps/api/tests/test_mcp_control_plane_api.py
@@ -34,17 +34,24 @@ def test_catalog_contains_exact_12_products_and_explicit_missing_rows() -> None:
assert catalog["source_policy"]["github_allowed"] is False
assert catalog["security_policy"]["external_rag_auto_write_allowed"] is False
- capabilities = {
- row["capability_id"]: row for row in catalog["capabilities"]
- }
+ capabilities = {row["capability_id"]: row for row in catalog["capabilities"]}
assert capabilities["rejected.github"]["decision"] == "rejected"
assert capabilities["quarantine.agent-bounty-mcp"]["decision"] == "quarantine"
- assert capabilities["external.smithery-site-audit-shadow"]["decision"] == "evaluation_shadow_candidate"
- assert capabilities["external.smithery-site-audit-shadow"]["deployment_allowed"] is False
- assert capabilities["rejected.smithery-technical-analysis"]["decision"] == "rejected"
- assert "tool_output_contains_cross_tool_prompt_injection" in capabilities[
- "quarantine.agent-bounty-mcp"
- ]["blockers"]
+ assert (
+ capabilities["external.smithery-site-audit-shadow"]["decision"]
+ == "evaluation_shadow_candidate"
+ )
+ assert (
+ capabilities["external.smithery-site-audit-shadow"]["deployment_allowed"]
+ is False
+ )
+ assert (
+ capabilities["rejected.smithery-technical-analysis"]["decision"] == "rejected"
+ )
+ assert (
+ "tool_output_contains_cross_tool_prompt_injection"
+ in capabilities["quarantine.agent-bounty-mcp"]["blockers"]
+ )
assert not any(
row["deployment_allowed"]
for row in catalog["capabilities"]
@@ -73,7 +80,10 @@ def test_overview_joins_committed_product_matrix_without_runtime_claims() -> Non
products = {row["product_id"]: row for row in payload["products"]}
assert products["ewoooc"]["canonical_repo"] == "wooo/ewoooc"
assert products["momo-pro-system"]["canonical_repo"] == "wooo/momo-pro-system"
- assert products["agent-bounty-protocol"]["inventory_state"] == "source_detected_quarantined"
+ assert (
+ products["agent-bounty-protocol"]["inventory_state"]
+ == "source_detected_quarantined"
+ )
def test_catalog_rejects_external_deployment_without_immutable_receipts(
@@ -102,15 +112,53 @@ def test_overview_endpoint_returns_real_committed_catalog_without_runtime() -> N
app.include_router(router, prefix="/api/v1")
client = TestClient(app)
- response = client.get(
- "/api/v1/mcp-control-plane/overview?include_runtime=false"
- )
+ response = client.get("/api/v1/mcp-control-plane/overview?include_runtime=false")
assert response.status_code == 200
data = response.json()
assert data["summary"]["product_count"] == 12
assert data["runtime"]["status"] == "not_requested"
assert data["operation_boundaries"]["github_allowed"] is False
- assert data["operation_boundaries"]["public_response_redacted_aggregate_only"] is True
+ assert (
+ data["operation_boundaries"]["public_response_redacted_aggregate_only"] is True
+ )
assert data["operation_boundaries"]["mutation_endpoint_exposed"] is False
- assert data["operation_boundaries"]["operator_auth_required_for_future_mutation"] is True
+ assert (
+ data["operation_boundaries"]["operator_auth_required_for_future_mutation"]
+ is True
+ )
+
+
+def test_overview_promotes_durable_lifecycle_receipt_not_static_policy() -> None:
+ catalog = load_mcp_control_plane_catalog(_OPERATIONS_DIR)
+ matrix = load_latest_product_governance_matrix_readback(_OPERATIONS_DIR)
+ runtime = {
+ "status": "ready",
+ "provider_registry": {"provider_count": 9, "tool_count": 39},
+ "gateway_audit_24h": {"call_count": 1},
+ "rag": {"total_chunks": 1},
+ "version_lifecycle": {
+ "status": "ready",
+ "latest_run": {
+ "run_id": "11111111-1111-1111-1111-111111111111",
+ "status": "completed_no_write_policy_blocked",
+ "fresh": True,
+ },
+ "blockers": ["external_candidates_not_promotion_ready"],
+ },
+ }
+
+ payload = build_mcp_control_plane_overview(
+ catalog=catalog,
+ product_matrix=matrix,
+ runtime=runtime,
+ )
+
+ assert payload["version_lifecycle"]["runtime_state"] == "running_fail_closed"
+ assert payload["version_lifecycle"]["runtime"]["status"] == "ready"
+ assert "external_candidates_not_promotion_ready" in payload["active_blockers"]
+ assert (
+ payload["operation_boundaries"]["normalized_version_receipt_write_allowed"]
+ is True
+ )
+ assert payload["operation_boundaries"]["production_upgrade_allowed"] is False
diff --git a/apps/api/tests/test_mcp_version_lifecycle_service.py b/apps/api/tests/test_mcp_version_lifecycle_service.py
new file mode 100644
index 000000000..885e9a3eb
--- /dev/null
+++ b/apps/api/tests/test_mcp_version_lifecycle_service.py
@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+# ruff: noqa: E402
+import hashlib
+import os
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+from src.services.mcp_version_lifecycle_service import (
+ CatalogFetchReceipt,
+ _extract_unambiguous_version,
+ _stable_catalog_fingerprint,
+ build_candidate_observation,
+ run_mcp_version_lifecycle_once,
+)
+
+
+class FakeLifecycleRepository:
+ def __init__(self, previous: dict[str, dict[str, Any]] | None = None) -> None:
+ self.previous = previous or {}
+ self.started: dict[str, Any] | None = None
+ self.observations: list[dict[str, Any]] = []
+ self.finished: dict[str, Any] | None = None
+
+ async def start_run(self, payload: dict[str, Any]) -> None:
+ self.started = payload
+
+ async def latest_observations(self) -> dict[str, dict[str, Any]]:
+ return self.previous
+
+ async def save_observation(self, payload: dict[str, Any]) -> None:
+ self.observations.append(payload)
+
+ async def finish_run(self, payload: dict[str, Any]) -> None:
+ self.finished = payload
+
+
+def _catalog() -> dict[str, Any]:
+ return {
+ "capabilities": [
+ {
+ "capability_id": "external.context7-readonly",
+ "source_type": "external",
+ "catalog_source": "mcp-so",
+ "catalog_url": "https://mcp.so/server/context7",
+ "decision": "development_shadow_candidate",
+ "risk_tier": "medium",
+ "deployment_allowed": False,
+ },
+ {
+ "capability_id": "external.smithery-site-audit-shadow",
+ "source_type": "external",
+ "catalog_source": "smithery",
+ "catalog_url": "https://smithery.ai/servers/example/site-audit",
+ "decision": "evaluation_shadow_candidate",
+ "risk_tier": "high",
+ "deployment_allowed": False,
+ },
+ {
+ "capability_id": "rejected.disallowed-source",
+ "source_type": "external",
+ "catalog_source": "smithery",
+ "catalog_url": "https://example.invalid/rejected",
+ "decision": "rejected",
+ "risk_tier": "high",
+ "deployment_allowed": False,
+ },
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_lifecycle_run_persists_baselines_and_fail_closed_terminals() -> None:
+ repository = FakeLifecycleRepository()
+
+ async def fetcher(
+ url: str,
+ source: str,
+ timeout_seconds: int,
+ ) -> CatalogFetchReceipt:
+ assert source in {"mcp-so", "smithery"}
+ assert timeout_seconds > 0
+ return CatalogFetchReceipt(
+ status="ok",
+ source_url=url,
+ http_status=200,
+ body=b'{"version":"1.2.3","content":"not persisted"}',
+ )
+
+ result = await run_mcp_version_lifecycle_once(
+ _catalog(),
+ trigger_kind="test",
+ repository=repository,
+ fetcher=fetcher,
+ generated_at=datetime(2026, 7, 15, tzinfo=UTC),
+ )
+
+ assert result["status"] == "completed_no_write_policy_blocked"
+ assert result["candidate_count"] == 2
+ assert result["baseline_count"] == 2
+ assert result["observed_count"] == 2
+ assert result["policy_blocked_count"] == 2
+ assert repository.started is not None
+ assert repository.finished is not None
+ assert len(repository.observations) == 2
+ assert {row["lifecycle_terminal"] for row in repository.observations} == {
+ "no_write_baseline"
+ }
+ for observation in repository.observations:
+ assert observation["observed_version"] == "1.2.3"
+ assert observation["policy_decision"]["external_install_allowed"] is False
+ assert observation["policy_decision"]["rag_write_allowed"] is False
+ assert (
+ observation["stage_receipts"]["sensor_source_receipt"]["raw_body_stored"]
+ is False
+ )
+ assert (
+ observation["stage_receipts"]["bounded_execution"]["status"]
+ == "not_executed"
+ )
+
+
+def test_changed_version_stops_before_replay_without_immutable_receipts() -> None:
+ old_body = b'{"version":"1.2.3"}'
+ candidate = _catalog()["capabilities"][0]
+ observation = build_candidate_observation(
+ candidate,
+ fetched=CatalogFetchReceipt(
+ status="ok",
+ source_url=candidate["catalog_url"],
+ http_status=200,
+ body=b'{"version":"1.3.0"}',
+ ),
+ previous={
+ "observed_version": "1.2.3",
+ "content_sha256": hashlib.sha256(old_body).hexdigest(),
+ },
+ run_id=uuid.uuid4(),
+ trace_id="trace-version-change",
+ work_item_id="MCP-VERSION-TEST",
+ observed_at=datetime(2026, 7, 15, tzinfo=UTC),
+ )
+
+ assert observation["change_state"] == "version_changed"
+ assert observation["lifecycle_terminal"] == "no_write_policy_blocked"
+ assert observation["stage_receipts"]["check_mode"]["status"] == (
+ "blocked_immutable_supply_chain_receipts_missing"
+ )
+ assert (
+ observation["stage_receipts"]["bounded_execution"][
+ "external_tool_call_performed"
+ ]
+ is False
+ )
+ assert (
+ observation["stage_receipts"]["post_verifier_and_rollback"]["rollback_required"]
+ is False
+ )
+
+
+@pytest.mark.asyncio
+async def test_source_failure_is_durable_partial_degraded_no_write() -> None:
+ repository = FakeLifecycleRepository()
+
+ async def fetcher(
+ url: str,
+ source: str,
+ timeout_seconds: int,
+ ) -> CatalogFetchReceipt:
+ del source, timeout_seconds
+ return CatalogFetchReceipt(
+ status="error",
+ source_url=url,
+ http_status=503,
+ error_class="http_503",
+ )
+
+ result = await run_mcp_version_lifecycle_once(
+ _catalog(),
+ trigger_kind="test",
+ repository=repository,
+ fetcher=fetcher,
+ )
+
+ assert result["status"] == "partial_degraded"
+ assert result["error_count"] == 2
+ assert all(
+ row["lifecycle_terminal"] == "no_write_source_error"
+ for row in repository.observations
+ )
+ assert repository.finished is not None
+ verifier = repository.finished["receipt"]["independent_post_verifier"]
+ assert verifier["external_install_performed"] is False
+ assert verifier["external_rag_write_performed"] is False
+
+
+def test_version_parser_rejects_ambiguous_page_versions() -> None:
+ assert _extract_unambiguous_version(b'{"version":"1.2.3"}') == "1.2.3"
+ assert (
+ _extract_unambiguous_version(b'{"version":"1.2.3","latestVersion":"2.0.0"}')
+ is None
+ )
+
+
+def test_stable_fingerprint_ignores_randomized_recommendation_sections() -> None:
+ stable_head = b"""
+ Context7 MCP
+
+
+
+ """
+ first, first_scope = _stable_catalog_fingerprint(
+ stable_head + b'Random A',
+ source_url="https://mcp.so/servers/context7-mcp",
+ observed_version=None,
+ )
+ second, second_scope = _stable_catalog_fingerprint(
+ stable_head + b'Random B