feat(governance): add AI agent version lifecycle proposals

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

@@ -292,6 +292,9 @@ from src.services.ai_agent_tool_adoption_approval_package import (
 from src.services.ai_agent_version_freshness_snapshot import (
     load_latest_ai_agent_version_freshness_snapshot,
 )
+from src.services.ai_agent_version_lifecycle_update_proposal import (
+    load_latest_ai_agent_version_lifecycle_update_proposal,
+)
 from src.services.ai_provider_route_matrix import (
     load_latest_ai_provider_route_matrix,
 )
@@ -3009,6 +3012,35 @@ async def get_agent_proactive_operations_contract() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/agent-version-lifecycle-update-proposal",
+    response_model=dict[str, Any],
+    summary="取得 AI Agent 版本生命週期更新提案佇列",
+    description=(
+        "讀取最新已提交的 AI Agent / 套件 / 工具 / 服務 / 主機版本生命週期更新提案；"
+        "此端點只回傳已脫敏治理資料與批准 gate，不啟用排程、不外查 registry/CVE/市場來源、"
+        "不升級套件、不寫 lockfile、不 pull/build/push image、不操作 host/K3s/stateful、"
+        "不建立 PR、不 auto merge、不送 Telegram、不讀取機密、不切換 provider、"
+        "不替換 OpenClaw、不修改生產路由。"
+    ),
+)
+async def get_agent_version_lifecycle_update_proposal() -> dict[str, Any]:
+    """Return the latest read-only AI Agent version lifecycle update proposal."""
+    try:
+        return await asyncio.to_thread(load_latest_ai_agent_version_lifecycle_update_proposal)
+    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("ai_agent_version_lifecycle_update_proposal_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AI Agent 版本生命週期更新提案無效",
+        ) from exc
+
+
 @router.get(
     "/agent-version-freshness-snapshot",
     response_model=dict[str, Any],

apps/api/src/services/ai_agent_version_lifecycle_update_proposal.py

@@ -0,0 +1,274 @@
+"""
+AI Agent version lifecycle update proposal snapshot.
+
+Loads the latest committed, read-only proposal queue for AI Agents, packages,
+tools, services, hosts, and AI solution updates. This module only exposes
+sanitized governance data: it never activates schedules, queries live external
+sources, upgrades packages, mutates hosts, sends Telegram messages, changes
+provider routes, or replaces OpenClaw.
+"""
+
+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 = "ai_agent_version_lifecycle_update_proposal_*.json"
+_SCHEMA_VERSION = "ai_agent_version_lifecycle_update_proposal_v1"
+_RUNTIME_AUTHORITY = "version_lifecycle_update_proposal_only_no_write_or_upgrade"
+_VALID_AGENTS = {"openclaw", "hermes", "nemotron"}
+_VALID_RISK_TIERS = {"low", "medium", "high", "critical"}
+_VALID_PRIORITIES = {"P0", "P1", "P2", "P3"}
+_BLOCKED_RUNTIME_FLAGS = {
+    "schedule_activation_allowed",
+    "external_market_lookup_allowed",
+    "external_registry_lookup_allowed",
+    "external_cve_lookup_allowed",
+    "package_upgrade_allowed",
+    "lockfile_write_allowed",
+    "host_upgrade_allowed",
+    "os_package_upgrade_allowed",
+    "kernel_upgrade_allowed",
+    "k3s_upgrade_allowed",
+    "kubectl_command_allowed",
+    "node_drain_allowed",
+    "reboot_allowed",
+    "stateful_restart_allowed",
+    "database_migration_allowed",
+    "image_pull_allowed",
+    "docker_build_allowed",
+    "registry_push_allowed",
+    "workflow_write_allowed",
+    "pr_creation_allowed",
+    "auto_merge_allowed",
+    "provider_route_switch_allowed",
+    "openclaw_replacement_allowed",
+    "paid_api_call_allowed",
+    "secret_read_allowed",
+    "telegram_direct_send_allowed",
+    "telegram_gateway_queue_write_allowed",
+    "production_write_allowed",
+    "conversation_transcript_display_allowed",
+}
+_TRANSCRIPT_MARKERS = {
+    "# In app browser",
+    "My request for Codex",
+    "Current URL:",
+    "AGENTS.md instructions",
+    "<environment_context>",
+    "批准！繼續",
+}
+_SECRET_MARKERS = {
+    "auth.json",
+    ".env",
+    "bot token",
+    "telegram token",
+    "secret value",
+    "private key",
+}
+
+
+def load_latest_ai_agent_version_lifecycle_update_proposal(
+    evaluations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the newest committed AI Agent version lifecycle update proposal."""
+    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
+    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
+    if not candidates:
+        raise FileNotFoundError(
+            f"no AI Agent version lifecycle update proposal 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")
+    label = str(latest)
+    _require_schema(payload, _SCHEMA_VERSION, label)
+    _require_read_only_boundaries(payload, label)
+    _require_rollup_consistency(payload, label)
+    _require_domain_and_proposal_integrity(payload, label)
+    _require_telegram_contract(payload, label)
+    _require_no_sensitive_public_content(payload, label)
+    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")
+    if program_status.get("runtime_authority") != _RUNTIME_AUTHORITY:
+        raise ValueError(f"{label}: runtime_authority must stay {_RUNTIME_AUTHORITY}")
+
+    boundaries = payload.get("runtime_boundaries") or {}
+    if boundaries.get("read_only_update_proposal_allowed") is not True:
+        raise ValueError(f"{label}: read_only_update_proposal_allowed must be true")
+
+    opened = sorted(flag for flag in _BLOCKED_RUNTIME_FLAGS if boundaries.get(flag) is not False)
+    if opened:
+        raise ValueError(f"{label}: runtime boundaries must remain false: {opened}")
+
+
+def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
+    domains = payload.get("lifecycle_domains") or []
+    proposals = payload.get("update_proposals") or []
+    cadences = payload.get("cadence_matrix") or []
+    gates = payload.get("approval_gate_matrix") or []
+    boundaries = payload.get("runtime_boundaries") or {}
+    rollups = payload.get("rollups") or {}
+
+    expected_counts = {
+        "domain_count": len(domains),
+        "proposal_count": len(proposals),
+        "cadence_count": len(cadences),
+        "approval_gate_count": len(gates),
+        "read_only_proposal_count": sum(
+            1 for proposal in proposals if proposal.get("direct_update_allowed") is False
+        ),
+        "approval_required_count": sum(
+            1 for proposal in proposals if proposal.get("requires_owner_approval") is True
+        ),
+        "critical_candidate_count": sum(
+            1 for proposal in proposals if proposal.get("risk_tier") == "critical"
+        ),
+        "high_candidate_count": sum(
+            1 for proposal in proposals if proposal.get("risk_tier") == "high"
+        ),
+        "false_runtime_boundary_count": sum(
+            1 for flag in _BLOCKED_RUNTIME_FLAGS if boundaries.get(flag) is False
+        ),
+        "auto_execution_allowed_count": sum(
+            1 for proposal in proposals if proposal.get("auto_execution_allowed") is True
+        ),
+        "telegram_direct_send_count": int(
+            (payload.get("telegram_digest_contract") or {}).get("direct_send_allowed") is True
+        ),
+        "telegram_gateway_queue_write_count": int(
+            (payload.get("telegram_digest_contract") or {}).get("gateway_queue_write_allowed")
+            is True
+        ),
+        "production_write_count": int(boundaries.get("production_write_allowed") is True),
+        "update_allowed_count": sum(
+            1 for flag in _BLOCKED_RUNTIME_FLAGS if flag.endswith("_allowed") and boundaries.get(flag) is True
+        ),
+    }
+    mismatched = {
+        key: {"expected": expected, "actual": rollups.get(key)}
+        for key, expected in expected_counts.items()
+        if rollups.get(key) != expected
+    }
+    if mismatched:
+        raise ValueError(f"{label}: rollup counts must match payload sections: {mismatched}")
+
+    domain_ids = sorted(domain.get("domain_id") for domain in domains)
+    proposal_ids = sorted(proposal.get("proposal_id") for proposal in proposals)
+    if sorted(rollups.get("domain_ids") or []) != domain_ids:
+        raise ValueError(f"{label}: rollups.domain_ids mismatch")
+    if sorted(rollups.get("proposal_ids") or []) != proposal_ids:
+        raise ValueError(f"{label}: rollups.proposal_ids mismatch")
+
+
+def _require_domain_and_proposal_integrity(payload: dict[str, Any], label: str) -> None:
+    domain_ids = {domain.get("domain_id") for domain in payload.get("lifecycle_domains") or []}
+    gate_ids = {gate.get("gate_id") for gate in payload.get("approval_gate_matrix") or []}
+
+    unsafe_domains = [
+        domain.get("domain_id")
+        for domain in payload.get("lifecycle_domains") or []
+        if domain.get("owner_agent") not in _VALID_AGENTS
+        or domain.get("risk_tier") not in _VALID_RISK_TIERS
+        or not domain.get("cadence")
+        or not domain.get("decision_policy")
+    ]
+    if unsafe_domains:
+        raise ValueError(f"{label}: lifecycle domains missing owner/risk/cadence: {unsafe_domains}")
+
+    for proposal in payload.get("update_proposals") or []:
+        proposal_id = proposal.get("proposal_id") or "<missing>"
+        if proposal.get("domain_id") not in domain_ids:
+            raise ValueError(f"{label}: proposal {proposal_id} references unknown domain")
+        if proposal.get("owner_agent") not in _VALID_AGENTS:
+            raise ValueError(f"{label}: proposal {proposal_id} has invalid owner_agent")
+        if proposal.get("risk_tier") not in _VALID_RISK_TIERS:
+            raise ValueError(f"{label}: proposal {proposal_id} has invalid risk_tier")
+        if proposal.get("priority") not in _VALID_PRIORITIES:
+            raise ValueError(f"{label}: proposal {proposal_id} has invalid priority")
+        if proposal.get("approval_gate") not in gate_ids:
+            raise ValueError(f"{label}: proposal {proposal_id} references unknown approval gate")
+        if proposal.get("requires_owner_approval") is not True:
+            raise ValueError(f"{label}: proposal {proposal_id} must require owner approval")
+        if proposal.get("direct_update_allowed") is not False:
+            raise ValueError(f"{label}: proposal {proposal_id} must not allow direct update")
+        if proposal.get("auto_execution_allowed") is not False:
+            raise ValueError(f"{label}: proposal {proposal_id} must not allow auto execution")
+        if not proposal.get("evidence_refs"):
+            raise ValueError(f"{label}: proposal {proposal_id} must include evidence_refs")
+        if not proposal.get("validation_plan"):
+            raise ValueError(f"{label}: proposal {proposal_id} must include validation_plan")
+        if not proposal.get("rollback_plan"):
+            raise ValueError(f"{label}: proposal {proposal_id} must include rollback_plan")
+        if not proposal.get("blocked_runtime_actions"):
+            raise ValueError(f"{label}: proposal {proposal_id} must include blocked_runtime_actions")
+
+    unsafe_gates = [
+        gate.get("gate_id")
+        for gate in payload.get("approval_gate_matrix") or []
+        if gate.get("owner_approval_required") is not True
+        or gate.get("auto_execute_allowed") is not False
+        or not gate.get("required_evidence")
+    ]
+    if unsafe_gates:
+        raise ValueError(f"{label}: approval gates must stay owner-reviewed: {unsafe_gates}")
+
+
+def _require_telegram_contract(payload: dict[str, Any], label: str) -> None:
+    contract = payload.get("telegram_digest_contract") or {}
+    if contract.get("direct_send_allowed") is not False:
+        raise ValueError(f"{label}: Telegram direct send must remain false")
+    if contract.get("gateway_queue_write_allowed") is not False:
+        raise ValueError(f"{label}: Telegram Gateway queue write must remain false")
+    if contract.get("bot_api_call_allowed") is not False:
+        raise ValueError(f"{label}: Telegram Bot API call must remain false")
+    if not contract.get("draft_outputs"):
+        raise ValueError(f"{label}: Telegram draft outputs must be declared")
+
+
+def _require_no_sensitive_public_content(payload: dict[str, Any], label: str) -> None:
+    for path, value in _walk_values(payload):
+        if not isinstance(value, str):
+            continue
+        lowered = value.lower()
+        transcript_hits = sorted(marker for marker in _TRANSCRIPT_MARKERS if marker in value)
+        if transcript_hits:
+            raise ValueError(
+                f"{label}: forbidden work-window conversation content at {path}: {transcript_hits}"
+            )
+        secret_hits = sorted(marker for marker in _SECRET_MARKERS if marker in lowered)
+        if secret_hits:
+            raise ValueError(f"{label}: forbidden secret marker at {path}: {secret_hits}")
+
+
+def _walk_values(value: Any, path: str = "$") -> list[tuple[str, Any]]:
+    if isinstance(value, dict):
+        results: list[tuple[str, Any]] = []
+        for key, child in value.items():
+            results.extend(_walk_values(child, f"{path}.{key}"))
+        return results
+    if isinstance(value, list):
+        results = []
+        for index, child in enumerate(value):
+            results.extend(_walk_values(child, f"{path}[{index}]"))
+        return results
+    return [(path, value)]