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

#!/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),
        "primary_source_alignment": _primary_source_alignment(),
        "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([
        "",
        "## 主流實務來源證據",
        "",
        "| 實務 | 官方來源 | AWOOOI Gate | Agent 分工 |",
        "|---|---|---|---|",
    ])
    for item in payload["primary_source_alignment"]:
        lines.append(
            f"| {item['practice']} | {item['source']} | `{item['awoooi_gate']}` | "
            f"{item['agent_assignment']} |"
        )

    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 _primary_source_alignment() -> list[dict[str, str]]:
    return [
        {
            "practice": "OpenAI Agents SDK：專家協作、tool execution、approvals、state 由產品掌控",
            "source": "https://developers.openai.com/api/docs/guides/agents",
            "awoooi_gate": "sandbox_orchestration_no_write",
            "agent_assignment": "OpenClaw 負責 policy guard；MarketRadar 追版本；Hermes 產審核包。",
        },
        {
            "practice": "NVIDIA Nemotron 3 Ultra / NeMo：長任務 Agent、profiling、evaluation、MCP / A2A 互通",
            "source": "https://developer.nvidia.com/blog/nvidia-nemotron-3-ultra-powers-faster-more-efficient-reasoning-for-long-running-agents/",
            "awoooi_gate": "nemotron_replay_evaluator_only",
            "agent_assignment": "NemoTron 只做離線 replay / evaluator / smoke gate，不接 production routing。",
        },
        {
            "practice": "LangGraph：durable execution、human-in-the-loop、stateful workflow runtime",
            "source": "https://docs.langchain.com/oss/python/langgraph/overview",
            "awoooi_gate": "incident_workflow_kernel_replay_first",
            "agent_assignment": "OpenClaw 仲裁狀態轉移；Hermes 記錄 replay 證據與交接原因。",
        },
        {
            "practice": "MCP：標準化 agent-to-tool / resource / prompt 連接，且需明確 user consent",
            "source": "https://modelcontextprotocol.io/specification/2025-06-18",
            "awoooi_gate": "read_only_tool_registry_before_write_adapter",
            "agent_assignment": "MarketRadar 監控 SDK / spec；Critic 檢查資料權限與 tool safety。",
        },
        {
            "practice": "A2A：跨框架 Agent 溝通、委派與互通；MCP 處理工具、A2A 處理 Agent 對 Agent",
            "source": "https://a2a-protocol.org/latest/",
            "awoooi_gate": "agent_to_agent_interop_watch_only",
            "agent_assignment": "OpenClaw 設定協作邊界；Hermes 彙整 handoff 記錄；NemoTron 比對輸出。",
        },
        {
            "practice": "OpenTelemetry GenAI：Agent / LLM / MCP trace 語意慣例，支援可觀測與稽核",
            "source": "https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/",
            "awoooi_gate": "trace_semconv_mapping_before_runtime_export",
            "agent_assignment": "Critic 定義稽核欄位；MarketRadar 追語意規範版本；Hermes 產日週月報。",
        },
    ]


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())