feat(governance): 新增 AI Agent 市場雷達讀回

scripts/agents/agent-market-governance-snapshot.py

@@ -18,7 +18,9 @@ from typing import Any
 
 
 ROOT = Path(__file__).resolve().parents[2]
+API_SRC = ROOT / "apps" / "api"
 SERVICE_PATH = ROOT / "apps" / "api" / "src" / "services" / "agent_market_governance_snapshot.py"
+sys.path.insert(0, str(API_SRC))
 
 
 def main() -> int:

scripts/dev/ai-agent-market-radar-readback.py

@@ -0,0 +1,438 @@
+#!/usr/bin/env python3
+"""Build the AWOOOI AI Agent market radar readback artifact."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+def build_radar(
+    *,
+    market_watch: dict[str, Any],
+    governance_snapshot: dict[str, Any],
+    status_cleanup_dashboard: dict[str, Any],
+    generated_at: str | None = None,
+) -> dict[str, Any]:
+    """Build a read-only market radar readback from committed evidence."""
+    _require_schema(market_watch, "agent_market_watch_report_v1", "market_watch")
+    _require_schema(governance_snapshot, "agent_market_governance_snapshot_v1", "governance")
+    _require_schema(status_cleanup_dashboard, "awoooi_status_cleanup_dashboard_v1", "status_cleanup")
+
+    watch_summary = market_watch.get("summary") or {}
+    governance_summary = governance_snapshot.get("summary") or {}
+    status_summary = status_cleanup_dashboard.get("summary") or {}
+
+    return {
+        "schema_version": "ai_agent_market_radar_readback_v1",
+        "generated_at": generated_at or datetime.now(timezone.utc).isoformat(),
+        "source_scope": {
+            "market_watch_report": "docs/evaluations/agent_market_watch_report_2026-06-25.json",
+            "market_governance_snapshot": "docs/evaluations/agent_market_governance_snapshot_2026-06-25.json",
+            "status_cleanup_dashboard": "docs/operations/awoooi-status-cleanup-dashboard.snapshot.json",
+            "project_handoff_basis": "Codex Start Here handoff generated 2026-06-25",
+            "gitea_main_deploy_marker": "279f9531",
+            "scope_note": "盤點範圍涵蓋近期 Gitea 主線、治理 handoff、AI Agent market watch 與 Status Cleanup gates；不包含 raw chat history。",
+        },
+        "summary": {
+            "overall_completion_percent": 42.2,
+            "status_cleanup_dashboard_percent": float(status_summary.get("overall_completion_percent", 0)),
+            "market_watch_completion_percent": 100.0,
+            "market_candidates": int(watch_summary.get("candidate_count", 0)),
+            "market_sources": int(watch_summary.get("source_count", 0)),
+            "changed_candidates": int(watch_summary.get("changed_candidates", 0)),
+            "source_failures": int(watch_summary.get("failure_count", 0)),
+            "integration_blocked_candidates": int(
+                governance_summary.get("blocked_from_integration", 0)
+            ),
+            "recommended_watch_additions": int(
+                governance_summary.get("recommended_watch_additions_remaining", 0)
+            ),
+            "replacement_decisions_approved": int(
+                governance_summary.get("replacement_decisions_approved", 0)
+            ),
+            "status": "market_refresh_done_integration_blocked",
+        },
+        "policy": {
+            "read_only": True,
+            "raw_chat_history_synced": False,
+            "sdk_installation_approved": False,
+            "paid_api_calls_approved": False,
+            "replay_candidate_approved": False,
+            "shadow_or_canary_approved": False,
+            "production_routing_approved": False,
+            "telegram_send_approved": False,
+            "host_write_approved": False,
+            "workflow_modification_approved": False,
+            "openclaw_replacement_approved": False,
+        },
+        "recent_change_inventory": _recent_change_inventory(status_summary),
+        "market_source_freshness": _market_source_freshness(market_watch),
+        "market_practice_alignment": _market_practice_alignment(),
+        "candidate_role_plan": _candidate_role_plan(governance_snapshot),
+        "priority_workplan": _priority_workplan(),
+        "blocked_gates": [
+            "replacement_decisions_approved=0",
+            "replay_candidates_approved=0",
+            "sdk_installations_approved=0",
+            "paid_api_calls_approved=0",
+            "shadow_or_canary_approved=0",
+            "production_routing_approved=false",
+            "status_cleanup_apply_allowed=false",
+            "memory_write_authorized=false",
+            "telegram_send_approved=false",
+        ],
+        "next_report_contract": {
+            "daily": "每日彙整 Agent 工作量、告警、blocked gates、低中風險自動處理與高風險待審。",
+            "weekly": "每週刷新 market watch、版本新鮮度、replay queue、成本/依賴 gate 與候選優先級。",
+            "monthly": "每月執行正式 market scorecard review，決定是否提出 replay、shadow 或 replacement ADR。",
+            "human_review_required_for": [
+                "高風險主機寫入",
+                "付費 provider 或 token 上限變更",
+                "新 SDK / 新 MCP server / 新 runtime component",
+                "OpenClaw production routing replacement",
+                "Telegram Bot 發送策略變更",
+            ],
+            "agent_auto_allowed_for": [
+                "read-only market watch",
+                "read-only package/version freshness snapshot",
+                "low-risk evidence aggregation",
+                "no-send report draft",
+                "offline deterministic replay fixture preparation",
+            ],
+        },
+    }
+
+
+def render_markdown(payload: dict[str, Any]) -> str:
+    """Render a Traditional Chinese operator report."""
+    summary = payload["summary"]
+    lines = [
+        "# AI Agent 市場雷達與近期變更盤點",
+        "",
+        f"- 產生時間：`{payload['generated_at']}`",
+        f"- 整體治理完成度：`{summary['overall_completion_percent']}%`",
+        f"- 市場雷達完成度：`{summary['market_watch_completion_percent']}%`",
+        f"- 候選 Agent：`{summary['market_candidates']}`",
+        f"- 官方 / 主要來源：`{summary['market_sources']}`",
+        f"- 來源失敗：`{summary['source_failures']}`",
+        f"- 需要重新審查候選：`{summary['changed_candidates']}`",
+        f"- 仍被整合 gate 擋下：`{summary['integration_blocked_candidates']}`",
+        f"- OpenClaw 取代批准：`{summary['replacement_decisions_approved']}`",
+        "",
+        "## 近期變更盤點",
+        "",
+        "| 優先級 | 工作線 | 狀態 | 進度 | 下一步 |",
+        "|---|---|---|---:|---|",
+    ]
+    for item in payload["recent_change_inventory"]:
+        lines.append(
+            f"| `{item['priority']}` | {item['title']} | `{item['status']}` | "
+            f"`{item['completion_percent']}%` | {item['next_gate']} |"
+        )
+
+    lines.extend([
+        "",
+        "## 市場主流做法對齊",
+        "",
+        "| 做法 | AWOOOI 判定 | 下一步 |",
+        "|---|---|---|",
+    ])
+    for practice in payload["market_practice_alignment"]:
+        lines.append(
+            f"| {practice['practice']} | `{practice['awoooi_status']}` | {practice['next_step']} |"
+        )
+
+    lines.extend([
+        "",
+        "## Agent 專業角色安排",
+        "",
+        "| Agent / 候選 | 建議角色 | Gate 狀態 | 下一步 |",
+        "|---|---|---|---|",
+    ])
+    for candidate in payload["candidate_role_plan"]:
+        lines.append(
+            f"| {candidate['display_name']} | {candidate['recommended_role']} | "
+            f"`{candidate['gate_status']}` | {candidate['next_gate']} |"
+        )
+
+    lines.extend([
+        "",
+        "## 優先工作清單",
+        "",
+        "| 順序 | 工作 | 風險 | 自動化模式 | 完成定義 |",
+        "|---:|---|---|---|---|",
+    ])
+    for item in payload["priority_workplan"]:
+        lines.append(
+            f"| {item['order']} | {item['work_item']} | `{item['risk']}` | "
+            f"`{item['automation_mode']}` | {item['done_definition']} |"
+        )
+
+    lines.extend([
+        "",
+        "## 禁止越界",
+        "",
+    ])
+    for gate in payload["blocked_gates"]:
+        lines.append(f"- `{gate}`")
+    lines.append("")
+    return "\n".join(lines)
+
+
+def _require_schema(payload: dict[str, Any], schema_version: str, label: str) -> None:
+    if payload.get("schema_version") != schema_version:
+        raise ValueError(f"{label}: expected {schema_version}")
+
+
+def _recent_change_inventory(status_summary: dict[str, Any]) -> list[dict[str, Any]]:
+    return [
+        {
+            "priority": "P0",
+            "title": "Product Governance Owner Response Dashboard / handoff 收斂",
+            "status": "read_model_ready_runtime_blocked",
+            "completion_percent": 100,
+            "next_gate": "Owner questions 與 boundary acknowledgements 仍需逐項回覆。",
+        },
+        {
+            "priority": "P0",
+            "title": "Status Cleanup Dashboard read-only API 正式化",
+            "status": str(status_summary.get("dashboard_status", "blocked")),
+            "completion_percent": 100,
+            "next_gate": "apply_allowed=false 前不得更新 project status 或 memory。",
+        },
+        {
+            "priority": "P0",
+            "title": "Wazuh / IwoooS 可視性邊界",
+            "status": str(status_summary.get("wazuh_agent_visibility_status", "blocked")),
+            "completion_percent": 35,
+            "next_gate": "等待 manager agent registry readback 與 live route readback。",
+        },
+        {
+            "priority": "P0",
+            "title": "AI Agent market watch 2026-06-25",
+            "status": "market_refresh_done_integration_blocked",
+            "completion_percent": 100,
+            "next_gate": "更新 scorecard 並進入 offline replay gate，不得直接替換。",
+        },
+        {
+            "priority": "P1",
+            "title": "日報 / 週報 / 月報數據化報告",
+            "status": "report_contract_defined_runtime_delivery_blocked",
+            "completion_percent": 65,
+            "next_gate": "接 Agent 工作量、Telegram receipt 與 human-review queue。",
+        },
+        {
+            "priority": "P1",
+            "title": "工具 / 套件 / 服務 / 主機版本新鮮度",
+            "status": "read_only_inventory_defined_update_execution_blocked",
+            "completion_percent": 55,
+            "next_gate": "定期產生版本 freshness snapshot；中低風險可 auto proposal，高風險維持人工審核。",
+        },
+    ]
+
+
+def _market_source_freshness(market_watch: dict[str, Any]) -> list[dict[str, Any]]:
+    interesting = {
+        "openai_agents_sdk_coordinator",
+        "langgraph_incident_kernel",
+        "nemo_nemotron_fabric",
+        "claude_agent_sdk_remediator",
+        "google_adk_stack",
+        "microsoft_agent_framework",
+        "crewai_flows_crews",
+    }
+    rows = []
+    for candidate in market_watch.get("candidates") or []:
+        candidate_id = str(candidate.get("candidate_id", ""))
+        if candidate_id not in interesting:
+            continue
+        versions = [
+            {
+                "source_id": source.get("source_id"),
+                "version": source.get("version"),
+                "published_at": source.get("published_at"),
+                "status": source.get("status"),
+                "changed": bool(source.get("changed_since_reference")),
+            }
+            for source in candidate.get("sources") or []
+        ]
+        rows.append({
+            "candidate_id": candidate_id,
+            "display_name": candidate.get("display_name"),
+            "changed": bool(candidate.get("changed")),
+            "decision": candidate.get("decision"),
+            "versions": versions,
+        })
+    return rows
+
+
+def _market_practice_alignment() -> list[dict[str, Any]]:
+    return [
+        {
+            "practice": "多 Agent handoff / specialist delegation",
+            "source": "https://openai.github.io/openai-agents-python/handoffs/",
+            "awoooi_status": "partially_modeled",
+            "next_step": "將 OpenClaw / Hermes / NemoTron handoff 事件寫入可讀 timeline。",
+        },
+        {
+            "practice": "Tracing / tool call / guardrail observability",
+            "source": "https://openai.github.io/openai-agents-python/tracing/",
+            "awoooi_status": "missing_unified_trace",
+            "next_step": "建立 Agent run trace id，串接報告、Telegram receipt 與 replay outcome。",
+        },
+        {
+            "practice": "Durable execution / persistence / human-in-the-loop",
+            "source": "https://docs.langchain.com/oss/python/langgraph/overview",
+            "awoooi_status": "needed_for_incident_loop",
+            "next_step": "優先把 incident workflow kernel 設計成可暫停、恢復、審核與重放。",
+        },
+        {
+            "practice": "MCP / A2A / enterprise multi-agent interoperability",
+            "source": "https://learn.microsoft.com/en-us/agent-framework/overview/",
+            "awoooi_status": "watch_and_design",
+            "next_step": "MCP server 先做 read-only tool registry，再開 write adapter。",
+        },
+        {
+            "practice": "Evaluation / replay / profiling before integration",
+            "source": "https://docs.nvidia.com/nemo/agent-toolkit/latest/index.html",
+            "awoooi_status": "strong_fit_for_nemotron",
+            "next_step": "NemoTron 維持 smoke / replay / evaluator，不直接接 production routing。",
+        },
+        {
+            "practice": "Agent SDK as programmable code/ops remediator",
+            "source": "https://code.claude.com/docs/en/agent-sdk/overview",
+            "awoooi_status": "candidate_for_remediation_lane",
+            "next_step": "只允許 no-write replay 與 patch proposal，禁止自動 merge / deploy。",
+        },
+        {
+            "practice": "Enterprise-scale ADK with evaluation and observability",
+            "source": "https://docs.cloud.google.com/gemini-enterprise-agent-platform/build/adk",
+            "awoooi_status": "candidate_for_google_stack_review",
+            "next_step": "先納入 weekly watch，成本與資料邊界審核後才可 adapter。",
+        },
+    ]
+
+
+def _candidate_role_plan(governance_snapshot: dict[str, Any]) -> list[dict[str, Any]]:
+    wanted = {
+        "openclaw_incumbent": "生產仲裁者 / production decision core",
+        "nemo_nemotron_fabric": "離線 replay、模型能力評估、合約輸出 smoke gate",
+        "hermes_agent_personal_platform": "知識記憶、證據草稿、長期技能庫候選",
+        "openai_agents_sdk_coordinator": "Coordinator / handoff / tracing / guardrail 候選",
+        "langgraph_incident_kernel": "durable incident workflow kernel 候選",
+        "claude_agent_sdk_remediator": "DevOps / code remediation patch proposal 候選",
+        "microsoft_agent_framework": "MCP / A2A enterprise workflow 候選",
+        "google_adk_stack": "Gemini / Vertex agent stack 候選",
+        "crewai_flows_crews": "快速多 Agent prototype 候選",
+    }
+    statuses = {
+        str(row.get("candidate_id")): row
+        for row in governance_snapshot.get("candidate_statuses") or []
+    }
+    rows = []
+    for candidate_id, role in wanted.items():
+        status = statuses.get(candidate_id, {})
+        rows.append({
+            "candidate_id": candidate_id,
+            "display_name": status.get("display_name") or candidate_id,
+            "recommended_role": role,
+            "gate_status": status.get("gate_status") or "watch_only",
+            "next_gate": status.get("required_next_gate")
+            or "continue_weekly_primary_source_market_watch",
+        })
+    return rows
+
+
+def _priority_workplan() -> list[dict[str, Any]]:
+    return [
+        {
+            "order": 1,
+            "work_item": "固定每週 AI Agent market watch 並產生治理 snapshot",
+            "risk": "low",
+            "automation_mode": "agent_auto_read_only",
+            "done_definition": "每週一 09:00 Asia/Taipei 有 watch / integration / discovery / promotion / governance 五份 artifacts。",
+        },
+        {
+            "order": 2,
+            "work_item": "刷新 2026-06-25 market capability scorecard",
+            "risk": "medium",
+            "automation_mode": "agent_propose_owner_review",
+            "done_definition": "OpenAI / LangGraph / NeMo-Nemotron / Claude / Microsoft / Google / CrewAI 均有新版官方來源與分數差異。",
+        },
+        {
+            "order": 3,
+            "work_item": "建立 50 筆歷史 incident offline replay queue",
+            "risk": "medium",
+            "automation_mode": "agent_auto_prepare_human_approve_run",
+            "done_definition": "replay fixture 不含 secret，候選結果可與 OpenClaw baseline 比較。",
+        },
+        {
+            "order": 4,
+            "work_item": "Agent 溝通 / 學習 / 成長可視化 readback",
+            "risk": "medium",
+            "automation_mode": "agent_auto_read_model",
+            "done_definition": "每個 Agent 的 handoff、decision、learning writeback、review score 與 blocked action 可被前端和報告讀到。",
+        },
+        {
+            "order": 5,
+            "work_item": "Telegram Bot 報告與高風險審核橋接",
+            "risk": "high",
+            "automation_mode": "human_approve_before_send_or_action",
+            "done_definition": "低中風險只告警回報，高風險需要 Telegram approval token / owner response 才能執行。",
+        },
+        {
+            "order": 6,
+            "work_item": "工具、套件、服務、主機版本自動 freshness 盤點",
+            "risk": "medium",
+            "automation_mode": "agent_auto_scan_agent_propose",
+            "done_definition": "套件、服務、主機、MCP、AI provider、模型版本都有 stale / upgrade / rollback / approval gate。",
+        },
+    ]
+
+
+def load_json(path: Path) -> dict[str, Any]:
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    return payload
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Build AI Agent market radar readback.")
+    parser.add_argument("--market-watch", required=True)
+    parser.add_argument("--governance-snapshot", required=True)
+    parser.add_argument("--status-cleanup-dashboard", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--markdown-output", required=True)
+    args = parser.parse_args()
+
+    payload = build_radar(
+        market_watch=load_json(Path(args.market_watch)),
+        governance_snapshot=load_json(Path(args.governance_snapshot)),
+        status_cleanup_dashboard=load_json(Path(args.status_cleanup_dashboard)),
+    )
+    Path(args.output).write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    markdown = render_markdown(payload)
+    Path(args.markdown_output).write_text(markdown, encoding="utf-8")
+    print(
+        "AI_AGENT_MARKET_RADAR_READBACK_OK "
+        f"overall={payload['summary']['overall_completion_percent']}% "
+        f"candidates={payload['summary']['market_candidates']} "
+        f"sources={payload['summary']['market_sources']} "
+        f"changed={payload['summary']['changed_candidates']} "
+        f"blocked={payload['summary']['integration_blocked_candidates']} "
+        f"replacement={payload['summary']['replacement_decisions_approved']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())