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

feat(mcp): run governed version lifecycle receipts

Browse Source
This commit is contained in:
ogt
2026-07-15 02:16:48 +08:00
parent b49a1ec164
commit aed4b044d4
13 changed files with 1831 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
apps/api/src/core/config.py
View File

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

102
apps/api/src/jobs/mcp_version_lifecycle_job.py Normal file
View File

@@ -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__,
)

55
apps/api/src/services/mcp_control_plane_service.py
View File

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

935
apps/api/src/services/mcp_version_lifecycle_service.py Normal file
View File

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

34
apps/api/src/workers/signal_worker.py
View File

@@ -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 連線池