feat(ai): 新增 P2-004 供應鏈漂移監控

apps/api/src/api/v1/agents.py

@@ -280,6 +280,9 @@ from src.services.dependency_drift_check_plan import (
 from src.services.dependency_risk_policy import (
     load_latest_dependency_risk_policy,
 )
+from src.services.dependency_supply_chain_drift_monitor import (
+    load_latest_dependency_supply_chain_drift_monitor,
+)
 from src.services.dependency_upgrade_approval_package_template import (
     load_latest_dependency_upgrade_approval_package_template,
 )
@@ -3197,6 +3200,35 @@ async def get_dependency_drift_check_plan() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/dependency-supply-chain-drift-monitor",
+    response_model=dict[str, Any],
+    summary="取得 P2-004 依賴 / 供應鏈漂移監控",
+    description=(
+        "讀取最新已提交的 P2-004 依賴 / 供應鏈漂移監控；"
+        "此端點只回傳 repo-only committed snapshot，不啟用排程、不寫 workflow、"
+        "不呼叫外部 CVE / license / registry / Agent market 來源、不安裝或升級套件、"
+        "不寫 lockfile、不執行 npm audit、不執行 docker build、不 pull image、不推 registry、"
+        "不建立 PR、不送 Telegram、不讀 secret、不做 host probe、不做 production write。"
+    ),
+)
+async def get_dependency_supply_chain_drift_monitor() -> dict[str, Any]:
+    """Return the latest read-only P2-004 dependency / supply-chain drift monitor."""
+    try:
+        return await asyncio.to_thread(load_latest_dependency_supply_chain_drift_monitor)
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error("dependency_supply_chain_drift_monitor_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="P2-004 依賴 / 供應鏈漂移監控快照無效",
+        ) from exc
+
+
 @router.get(
     "/dependency-upgrade-approval-package-template",
     response_model=dict[str, Any],

apps/api/src/services/dependency_supply_chain_drift_monitor.py

@@ -0,0 +1,159 @@
+"""
+P2-004 dependency / supply-chain drift monitor snapshot.
+
+Loads the latest committed, read-only monitor that aggregates existing
+package, JavaScript, Docker, dependency-policy, and version freshness
+snapshots. This monitor does not query external CVE/license/registry sources,
+install packages, write lockfiles, build images, create PRs, send Telegram
+messages, read secrets, or mutate production.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_evaluations_dir
+
+_DEFAULT_EVALUATIONS_DIR = default_evaluations_dir(Path(__file__))
+_SNAPSHOT_PATTERN = "dependency_supply_chain_drift_monitor_*.json"
+_SCHEMA_VERSION = "dependency_supply_chain_drift_monitor_v1"
+
+_ACTION_REQUIRED_STATUSES = {"action_required", "blocked_until_approval"}
+_BLOCKED_BOUNDARY_FLAGS = {
+    "schedule_activation_allowed",
+    "workflow_write_allowed",
+    "external_cve_lookup_allowed",
+    "external_license_lookup_allowed",
+    "registry_lookup_allowed",
+    "agent_market_external_lookup_allowed",
+    "package_installation_allowed",
+    "package_upgrade_allowed",
+    "lockfile_write_allowed",
+    "docker_build_allowed",
+    "image_pull_allowed",
+    "image_rebuild_allowed",
+    "registry_push_allowed",
+    "pr_creation_allowed",
+    "telegram_send_allowed",
+    "production_write_allowed",
+    "paid_external_service_allowed",
+    "secret_read_allowed",
+    "host_probe_allowed",
+    "npm_audit_allowed",
+}
+
+
+def load_latest_dependency_supply_chain_drift_monitor(
+    evaluations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the newest committed P2-004 dependency / supply-chain drift monitor."""
+    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(
+            f"no dependency supply-chain drift monitor snapshots found in {directory}"
+        )
+
+    latest = candidates[-1]
+    with latest.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if not isinstance(payload, dict):
+        raise ValueError(f"{latest}: expected JSON object")
+    _require_schema(payload, _SCHEMA_VERSION, str(latest))
+    _require_read_only_boundaries(payload, str(latest))
+    _require_rollup_consistency(payload, str(latest))
+    _require_candidate_evidence(payload, str(latest))
+    _require_owner_action_links(payload, str(latest))
+    return payload
+
+
+def _require_schema(payload: dict[str, Any], expected: str, label: str) -> None:
+    actual = payload.get("schema_version")
+    if actual != expected:
+        raise ValueError(f"{label}: expected schema_version={expected}, got {actual!r}")
+
+
+def _require_read_only_boundaries(payload: dict[str, Any], label: str) -> None:
+    program_status = payload.get("program_status") or {}
+    if program_status.get("read_only_mode") is not True:
+        raise ValueError(f"{label}: program_status.read_only_mode must be true")
+
+    boundaries = payload.get("monitor_boundaries") or {}
+    if boundaries.get("read_only_repo_monitor_allowed") is not True:
+        raise ValueError(f"{label}: read_only_repo_monitor_allowed must be true")
+
+    allowed = sorted(flag for flag in _BLOCKED_BOUNDARY_FLAGS if boundaries.get(flag) is not False)
+    if allowed:
+        raise ValueError(f"{label}: monitor boundaries must remain false: {allowed}")
+
+
+def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
+    source_snapshots = payload.get("source_snapshot_readbacks") or []
+    monitor_checks = payload.get("monitor_checks") or []
+    drift_candidates = payload.get("drift_candidates") or []
+    owner_actions = payload.get("owner_actions") or []
+    rollups = payload.get("rollups") or {}
+    boundaries = payload.get("monitor_boundaries") or {}
+
+    if rollups.get("source_snapshot_count") != len(source_snapshots):
+        raise ValueError(f"{label}: rollups.source_snapshot_count must match source snapshots")
+    if rollups.get("monitor_check_count") != len(monitor_checks):
+        raise ValueError(f"{label}: rollups.monitor_check_count must match monitor checks")
+    if rollups.get("drift_candidate_count") != len(drift_candidates):
+        raise ValueError(f"{label}: rollups.drift_candidate_count must match drift candidates")
+    if rollups.get("owner_action_count") != len(owner_actions):
+        raise ValueError(f"{label}: rollups.owner_action_count must match owner actions")
+
+    stale_count = sum(
+        1 for snapshot in source_snapshots if snapshot.get("freshness_status") == "stale_action_required"
+    )
+    if rollups.get("stale_source_snapshot_count") != stale_count:
+        raise ValueError(f"{label}: rollups.stale_source_snapshot_count must match stale snapshots")
+
+    action_required_ids = {
+        candidate.get("candidate_id")
+        for candidate in drift_candidates
+        if candidate.get("status") in _ACTION_REQUIRED_STATUSES
+    }
+    if set(rollups.get("action_required_drift_candidate_ids") or []) != action_required_ids:
+        raise ValueError(
+            f"{label}: rollups.action_required_drift_candidate_ids must match candidates"
+        )
+    if rollups.get("action_required_candidate_count") != len(action_required_ids):
+        raise ValueError(f"{label}: rollups.action_required_candidate_count must match candidates")
+
+    by_domain: dict[str, int] = {}
+    for candidate in drift_candidates:
+        domain = candidate.get("domain")
+        by_domain[domain] = by_domain.get(domain, 0) + 1
+    if rollups.get("by_domain") != by_domain:
+        raise ValueError(f"{label}: rollups.by_domain must match drift candidates")
+
+    blocked_count = sum(1 for flag in _BLOCKED_BOUNDARY_FLAGS if boundaries.get(flag) is False)
+    if rollups.get("blocked_operation_count") != blocked_count:
+        raise ValueError(f"{label}: rollups.blocked_operation_count must match false boundaries")
+
+
+def _require_candidate_evidence(payload: dict[str, Any], label: str) -> None:
+    drift_candidates = payload.get("drift_candidates") or []
+    for candidate in drift_candidates:
+        candidate_id = candidate.get("candidate_id") or "<missing>"
+        evidence_refs = candidate.get("evidence_refs") or []
+        if not evidence_refs:
+            raise ValueError(f"{label}: drift candidate {candidate_id} must include evidence_refs")
+        if not candidate.get("next_owner_action"):
+            raise ValueError(f"{label}: drift candidate {candidate_id} must include next_owner_action")
+
+
+def _require_owner_action_links(payload: dict[str, Any], label: str) -> None:
+    owner_action_ids = {action.get("action_id") for action in payload.get("owner_actions") or []}
+    for candidate in payload.get("drift_candidates") or []:
+        action_id = candidate.get("next_owner_action")
+        if action_id not in owner_action_ids:
+            raise ValueError(
+                f"{label}: drift candidate {candidate.get('candidate_id')} references "
+                f"unknown owner action {action_id!r}"
+            )