feat(governance): 新增 AI 技術雷達滾動監控

scripts/agents/ai-technology-watch.py

@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+"""Run the AWOOOI read-only AI technology watch."""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[2]
+API_ROOT = ROOT / "apps" / "api"
+SERVICE_PATH = ROOT / "apps" / "api" / "src" / "services" / "ai_technology_watch.py"
+run_ai_technology_watch = None
+
+
+def main() -> int:
+    global run_ai_technology_watch
+    if run_ai_technology_watch is None:
+        run_ai_technology_watch = _load_service()
+
+    parser = argparse.ArgumentParser(description="Run AWOOOI AI technology watch.")
+    parser.add_argument(
+        "--registry",
+        default="docs/ai/ai-technology-watch-sources.v1.json",
+        help="AI technology watch registry JSON",
+    )
+    parser.add_argument("--output", required=True, help="report output JSON")
+    parser.add_argument(
+        "--mode",
+        choices=("offline", "live"),
+        default="live",
+        help="offline validates registry only; live fetches primary sources",
+    )
+    parser.add_argument("--previous-report")
+    parser.add_argument("--timeout-seconds", type=int, default=12)
+    args = parser.parse_args()
+
+    registry = _read_json(Path(args.registry))
+    previous = _read_json(Path(args.previous_report)) if args.previous_report else None
+    report = run_ai_technology_watch(
+        registry,
+        registry_path=args.registry,
+        mode=args.mode,
+        previous_report=previous,
+        timeout_seconds=args.timeout_seconds,
+    )
+    Path(args.output).write_text(
+        json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    print(json.dumps(report["summary"], ensure_ascii=False, sort_keys=True))
+    return 0
+
+
+def _read_json(path: Path) -> dict[str, Any]:
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise SystemExit(f"{path}: expected JSON object")
+    return payload
+
+
+def _load_service() -> Any:
+    api_root = str(API_ROOT)
+    if api_root not in sys.path:
+        sys.path.insert(0, api_root)
+    module_name = "awoooi_ai_technology_watch_service"
+    spec = importlib.util.spec_from_file_location(module_name, SERVICE_PATH)
+    if spec is None or spec.loader is None:
+        raise SystemExit(f"cannot load AI technology watch service from {SERVICE_PATH}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
+    return module.run_ai_technology_watch
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/dev/ai-technology-radar-readback.py

@@ -0,0 +1,434 @@
+#!/usr/bin/env python3
+"""Build the AWOOOI AI technology radar readback artifact."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+def build_radar(
+    *,
+    technology_watch: dict[str, Any],
+    agent_market_radar: dict[str, Any],
+    evidence_commit: str,
+    generated_at: str | None = None,
+) -> dict[str, Any]:
+    """Build a read-only AI technology radar readback from committed evidence."""
+    _require_schema(technology_watch, "ai_technology_watch_report_v1", "technology_watch")
+    _require_schema(agent_market_radar, "ai_agent_market_radar_readback_v1", "agent_market_radar")
+
+    watch_summary = technology_watch.get("summary") or {}
+    agent_summary = agent_market_radar.get("summary") or {}
+    source_count = int(watch_summary.get("source_count", 0))
+    failure_count = int(watch_summary.get("source_failure_count", 0))
+    success_rate = _success_rate(source_count, failure_count)
+
+    return {
+        "schema_version": "ai_technology_radar_readback_v1",
+        "generated_at": generated_at or datetime.now(timezone.utc).isoformat(),
+        "source_scope": {
+            "technology_watch_report": "docs/evaluations/ai_technology_watch_report_2026-06-25.json",
+            "technology_source_registry": "docs/ai/ai-technology-watch-sources.v1.json",
+            "agent_market_radar_readback": "docs/operations/ai-agent-market-radar-readback.snapshot.json",
+            "gitea_main_evidence_basis_commit": evidence_commit,
+            "scope_note": "本讀回只整合已提交的只讀來源監控、AI Agent 市場雷達與治理 gate；不包含 raw chat history、secret、session 或本機工作視窗內容。",
+        },
+        "summary": {
+            "overall_completion_percent": float(agent_summary.get("overall_completion_percent", 42.2)),
+            "ai_technology_radar_completion_percent": success_rate,
+            "technology_count": int(watch_summary.get("technology_count", 0)),
+            "technology_area_count": int(watch_summary.get("technology_area_count", 0)),
+            "source_count": source_count,
+            "changed_technologies": int(watch_summary.get("changed_technologies", 0)),
+            "review_queue_count": int(watch_summary.get("review_queue_count", 0)),
+            "source_failures": failure_count,
+            "high_priority_count": int(watch_summary.get("high_priority_count", 0)),
+            "rolling_update_status": "near_real_time_watch_ready_integration_gated",
+        },
+        "policy": {
+            "read_only": True,
+            "raw_chat_history_synced": False,
+            "sdk_installation_approved": False,
+            "paid_api_calls_approved": False,
+            "production_routing_approved": False,
+            "telegram_send_approved": False,
+            "model_provider_switch_approved": False,
+            "host_write_approved": False,
+            "openclaw_replacement_approved": False,
+        },
+        "technology_area_counts": dict(technology_watch.get("technology_area_counts") or {}),
+        "technology_domains": _technology_domains(technology_watch),
+        "high_priority_review_queue": _high_priority_review_queue(technology_watch),
+        "professional_agent_roles": _professional_agent_roles(),
+        "rolling_update_controls": _rolling_update_controls(technology_watch),
+        "integration_candidates": _integration_candidates(technology_watch),
+        "priority_workplan": _priority_workplan(),
+        "blocked_gates": [
+            "sdk_installation_approved=false",
+            "paid_api_calls_approved=false",
+            "production_routing_approved=false",
+            "telegram_send_approved=false",
+            "model_provider_switch_approved=false",
+            "host_write_approved=false",
+            "openclaw_replacement_approved=false",
+            "replay_shadow_canary_gate_required=true",
+            "cost_and_data_boundary_review_required=true",
+        ],
+        "report_contract": {
+            "api_endpoint": "/api/v1/agents/ai-technology-radar-readback",
+            "frontend_target": "/zh-TW/governance?tab=agent-market",
+            "schedule_workflow": ".gitea/workflows/ai-technology-watch.yaml",
+            "schedule_enabled": True,
+            "schedule_cron_utc": "0 */6 * * *",
+            "near_real_time": "每 6 小時讀取 primary sources，偵測主流 AI 技術版本、文件與 release 變更。",
+            "daily": "每日彙整變更、來源失敗、審核佇列與可自動處理項目。",
+            "weekly": "每週做技術 scorecard，決定 sandbox / replay / adapter design 優先級。",
+            "monthly": "每月進行策略 review，決定納入 roadmap、維持 watch-only 或移出監控。",
+            "human_review_required_for": [
+                "新 SDK / package / MCP server 安裝",
+                "付費 API 或 token 上限變更",
+                "模型 provider / 生產路由切換",
+                "Telegram Bot 即時發送或審批按鈕策略變更",
+                "主機、K8s、workflow、Nginx、secret 或資料層寫入",
+                "OpenClaw 生產決策核心替換、拆分或降級",
+            ],
+            "agent_auto_allowed_for": [
+                "官方來源只讀監控",
+                "版本與文件 hash 比對",
+                "審核佇列分類",
+                "繁中 no-send 報告草稿",
+                "離線 replay fixture 準備",
+                "低風險文件與讀回 snapshot 更新提案",
+            ],
+        },
+    }
+
+
+def render_markdown(payload: dict[str, Any]) -> str:
+    """Render a Traditional Chinese operator report."""
+    summary = payload["summary"]
+    lines = [
+        "# AI 技術雷達與滾動更新讀回",
+        "",
+        f"- 產生時間：`{payload['generated_at']}`",
+        f"- 整體治理完成度：`{summary['overall_completion_percent']}%`",
+        f"- AI 技術雷達來源成功率：`{summary['ai_technology_radar_completion_percent']}%`",
+        f"- 監控技術項目：`{summary['technology_count']}`",
+        f"- 技術領域：`{summary['technology_area_count']}`",
+        f"- 官方 / primary sources：`{summary['source_count']}`",
+        f"- 來源失敗：`{summary['source_failures']}`",
+        f"- 需要審核變更：`{summary['changed_technologies']}`",
+        f"- 高優先級項目：`{summary['high_priority_count']}`",
+        f"- 滾動更新狀態：`{summary['rolling_update_status']}`",
+        "",
+        "## 技術領域覆蓋",
+        "",
+        "| 技術領域 | 技術數 | 高優先級 | 需要審核 | 代表技術 |",
+        "|---|---:|---:|---:|---|",
+    ]
+    for domain in payload["technology_domains"]:
+        lines.append(
+            f"| `{domain['technology_area']}` | `{domain['technology_count']}` | "
+            f"`{domain['high_priority_count']}` | `{domain['changed_count']}` | "
+            f"{', '.join(domain['representative_technologies'])} |"
+        )
+
+    lines.extend([
+        "",
+        "## 高優先級審核佇列",
+        "",
+        "| 技術 | 領域 | 優先級 | Gate | 下一步 |",
+        "|---|---|---|---|---|",
+    ])
+    for item in payload["high_priority_review_queue"]:
+        lines.append(
+            f"| {item['display_name']} | `{item['technology_area']}` | "
+            f"`{item['evaluation_priority']}` | `{item['gate_status']}` | {item['next_gate']} |"
+        )
+
+    lines.extend([
+        "",
+        "## Agent 專業分工",
+        "",
+        "| Agent | 專業角色 | 自動化範圍 | 需要審核的邊界 |",
+        "|---|---|---|---|",
+    ])
+    for role in payload["professional_agent_roles"]:
+        lines.append(
+            f"| {role['agent']} | {role['professional_role']} | "
+            f"{role['auto_scope']} | {role['review_boundary']} |"
+        )
+
+    lines.extend([
+        "",
+        "## 滾動更新控制",
+        "",
+        "| 節奏 | Agent 可自動做什麼 | 輸出 | Gate |",
+        "|---|---|---|---|",
+    ])
+    for control in payload["rolling_update_controls"]:
+        lines.append(
+            f"| {control['cadence']} | {control['agent_auto_action']} | "
+            f"{control['output']} | `{control['gate']}` |"
+        )
+
+    lines.extend([
+        "",
+        "## 優先工作清單",
+        "",
+        "| 順序 | 工作 | 優先級 | 自動化模式 | 完成定義 |",
+        "|---:|---|---|---|---|",
+    ])
+    for item in payload["priority_workplan"]:
+        lines.append(
+            f"| {item['order']} | {item['work_item']} | `{item['priority']}` | "
+            f"`{item['automation_mode']}` | {item['done_definition']} |"
+        )
+
+    lines.extend([
+        "",
+        "## 仍被 Gate 擋下",
+        "",
+    ])
+    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 _success_rate(source_count: int, failure_count: int) -> float:
+    if source_count <= 0:
+        return 0.0
+    return round(((source_count - failure_count) / source_count) * 100, 1)
+
+
+def _technology_domains(report: dict[str, Any]) -> list[dict[str, Any]]:
+    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in report.get("technologies") or []:
+        grouped[str(row.get("technology_area") or "uncategorized")].append(row)
+
+    domains = []
+    for area, rows in sorted(grouped.items()):
+        high_priority = [row for row in rows if row.get("evaluation_priority") in {"p0", "p1"}]
+        changed = [row for row in rows if row.get("changed")]
+        domains.append({
+            "technology_area": area,
+            "technology_count": len(rows),
+            "high_priority_count": len(high_priority),
+            "changed_count": len(changed),
+            "representative_technologies": [
+                str(row.get("display_name") or row.get("technology_id"))
+                for row in rows[:4]
+            ],
+        })
+    return domains
+
+
+def _high_priority_review_queue(report: dict[str, Any]) -> list[dict[str, Any]]:
+    rows = [
+        row
+        for row in report.get("technologies") or []
+        if row.get("evaluation_priority") in {"p0", "p1"} and row.get("changed")
+    ]
+    return [
+        {
+            "technology_id": row.get("technology_id"),
+            "display_name": row.get("display_name"),
+            "technology_area": row.get("technology_area"),
+            "evaluation_priority": row.get("evaluation_priority"),
+            "gate_status": "scorecard_required_before_integration",
+            "next_gate": "刷新 scorecard，若涉及 SDK/API/route/Telegram/host write 則送人工審核。",
+            "requires_cost_approval": bool(row.get("requires_cost_approval")),
+            "requires_dependency_approval": bool(row.get("requires_dependency_approval")),
+            "requires_security_review": bool(row.get("requires_security_review")),
+        }
+        for row in rows[:12]
+    ]
+
+
+def _professional_agent_roles() -> list[dict[str, Any]]:
+    return [
+        {
+            "agent": "OpenClaw",
+            "professional_role": "生產決策仲裁者、風險分級與最後 policy guard",
+            "auto_scope": "維持現有 production baseline、讀取 replay / shadow 評分、拒絕無證據替換",
+            "review_boundary": "任何取代、降級、生產路由切換都必須通過 replay / shadow / canary 與人工批准。",
+        },
+        {
+            "agent": "NemoTron",
+            "professional_role": "離線回放評估者、模型能力比較、合約輸出 smoke gate",
+            "auto_scope": "只讀 request pack、比對候選輸出、產生 replay scorecard 草稿",
+            "review_boundary": "不得自行呼叫外部 NIM/API、不得讀 labels 作答、不得進生產路由。",
+        },
+        {
+            "agent": "Hermes",
+            "professional_role": "知識管理、RAG 整理、報告草稿與長期技能庫維護",
+            "auto_scope": "整理 primary source 摘要、建立 no-send 日週月報、準備人審包",
+            "review_boundary": "不得同步 raw chat history、不得保存 secret、不得直接發 Telegram live report。",
+        },
+        {
+            "agent": "MarketRadar",
+            "professional_role": "AI 技術市場雷達、版本監控、來源失敗偵測",
+            "auto_scope": "每 6 小時只讀 primary sources、產生 freshness / review queue",
+            "review_boundary": "不得自動新增 SDK、不得自動修改 provider route 或 workflow 行為。",
+        },
+        {
+            "agent": "Critic / Reviewer",
+            "professional_role": "獨立審核、反例檢查、整合風險評分",
+            "auto_scope": "檢查政策旗標、來源可靠性、成本與資安風險",
+            "review_boundary": "只能輸出 blocked / candidate / owner_review，不得直接執行寫入。",
+        },
+    ]
+
+
+def _rolling_update_controls(report: dict[str, Any]) -> list[dict[str, Any]]:
+    cadence = report.get("cadence") or {}
+    return [
+        {
+            "cadence": "每 6 小時",
+            "agent_auto_action": "讀取官方文件、PyPI、npm、GitHub release、primary source hash。",
+            "output": "AI 技術 watch report、來源失敗清單、review queue。",
+            "gate": "read_only_only",
+            "cadence_source": cadence.get("near_real_time_watch", ""),
+        },
+        {
+            "cadence": "每日",
+            "agent_auto_action": "依 business applicability、成本、依賴、資安、AWOOOI fit 分類。",
+            "output": "日報摘要與中低風險自動處理建議。",
+            "gate": "no_send_report_until_delivery_gate",
+            "cadence_source": cadence.get("daily_triage", ""),
+        },
+        {
+            "cadence": "每週",
+            "agent_auto_action": "刷新 scorecard，決定 sandbox / replay / adapter design 優先級。",
+            "output": "週報、優先序、候選整合審查包。",
+            "gate": "scorecard_required_before_replay",
+            "cadence_source": cadence.get("weekly_scorecard", ""),
+        },
+        {
+            "cadence": "每月",
+            "agent_auto_action": "彙整趨勢，提出 roadmap / watch-only / retire 建議。",
+            "output": "月報與策略審核包。",
+            "gate": "human_review_for_strategy_or_production_change",
+            "cadence_source": cadence.get("monthly_strategy_review", ""),
+        },
+    ]
+
+
+def _integration_candidates(report: dict[str, Any]) -> list[dict[str, Any]]:
+    return [
+        {
+            "technology_id": row.get("technology_id"),
+            "display_name": row.get("display_name"),
+            "technology_area": row.get("technology_area"),
+            "integration_surface": row.get("integration_surface"),
+            "awoooi_role": row.get("awoooi_role"),
+            "changed": bool(row.get("changed")),
+            "decision": row.get("decision"),
+            "recommended_actions": row.get("recommended_actions") or [],
+        }
+        for row in report.get("technologies") or []
+    ]
+
+
+def _priority_workplan() -> list[dict[str, Any]]:
+    return [
+        {
+            "order": 1,
+            "priority": "P0",
+            "work_item": "AI 技術雷達 primary source 監控產品化",
+            "automation_mode": "agent_auto_read_only",
+            "done_definition": "API、snapshot、Markdown、schema、測試與 production readback 都能顯示技術領域、來源與 Gate。",
+        },
+        {
+            "order": 2,
+            "priority": "P0",
+            "work_item": "近即時版本 / release / docs 變更偵測",
+            "automation_mode": "agent_auto_schedule_read_only",
+            "done_definition": "每 6 小時可跑 watch；失敗來源會進日報，不會自動整合。",
+        },
+        {
+            "order": 3,
+            "priority": "P0",
+            "work_item": "OpenClaw / Hermes / NemoTron / MarketRadar 專業分工與成長紀錄",
+            "automation_mode": "agent_auto_read_model_human_review_for_write",
+            "done_definition": "每個 Agent 的角色、輸出、學習寫回與限制都能被前端讀回。",
+        },
+        {
+            "order": 4,
+            "priority": "P1",
+            "work_item": "AI 技術 scorecard 與 sandbox / replay 優先級",
+            "automation_mode": "agent_propose_owner_review",
+            "done_definition": "高優先級變更先進 scorecard，再進 no-cost/no-write sandbox 或 replay 計畫。",
+        },
+        {
+            "order": 5,
+            "priority": "P1",
+            "work_item": "Telegram Bot 報告與高風險審核橋接",
+            "automation_mode": "blocked_until_telegram_send_gate",
+            "done_definition": "低中風險只告警回報；高風險需 owner approval 後才可發送或執行。",
+        },
+        {
+            "order": 6,
+            "priority": "P2",
+            "work_item": "新 AI 技術探索與 watchlist 擴充",
+            "automation_mode": "agent_auto_discover_human_classify",
+            "done_definition": "GitHub topic / package registry / 官方 blog 可提出候選，但加入正式 watchlist 前需審核。",
+        },
+    ]
+
+
+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 technology radar readback.")
+    parser.add_argument("--technology-watch", required=True)
+    parser.add_argument("--agent-market-radar", required=True)
+    parser.add_argument("--evidence-commit", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--markdown-output", required=True)
+    parser.add_argument("--markdown-alias-output")
+    args = parser.parse_args()
+
+    payload = build_radar(
+        technology_watch=load_json(Path(args.technology_watch)),
+        agent_market_radar=load_json(Path(args.agent_market_radar)),
+        evidence_commit=args.evidence_commit,
+    )
+    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")
+    if args.markdown_alias_output:
+        Path(args.markdown_alias_output).write_text(markdown, encoding="utf-8")
+    print(
+        "AI_TECHNOLOGY_RADAR_READBACK_OK "
+        f"overall={payload['summary']['overall_completion_percent']}% "
+        f"technology={payload['summary']['technology_count']} "
+        f"sources={payload['summary']['source_count']} "
+        f"changed={payload['summary']['changed_technologies']} "
+        f"failures={payload['summary']['source_failures']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())