awoooi/scripts/dev/ai-technology-report-cadence-readback.py

#!/usr/bin/env python3
"""Build the AWOOOI AI technology report cadence 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_report_cadence(
    *,
    radar_readback: dict[str, Any],
    evidence_commit: str,
    generated_at: str | None = None,
) -> dict[str, Any]:
    """Build daily / weekly / monthly AI technology report readback."""
    _require_schema(radar_readback, "ai_technology_radar_readback_v1", "radar_readback")
    summary = radar_readback.get("summary") or {}
    policy = radar_readback.get("policy") or {}
    source_count = int(summary.get("source_count", 0))
    technology_count = int(summary.get("technology_count", 0))
    high_priority_count = int(summary.get("high_priority_count", 0))
    source_failures = int(summary.get("source_failures", 0))
    blocked_gates = list(radar_readback.get("blocked_gates") or [])
    policy_holds = [
        key
        for key, value in policy.items()
        if key != "read_only" and value is False
    ]

    report_cadences = _report_cadences(radar_readback)
    agent_workload_reports = _agent_workload_reports(
        source_count=source_count,
        technology_count=technology_count,
        high_priority_count=high_priority_count,
        blocked_gate_count=len(blocked_gates),
    )
    post_report_packets = _post_report_analysis_packets(radar_readback)

    return {
        "schema_version": "ai_technology_report_cadence_readback_v1",
        "generated_at": generated_at or datetime.now(timezone.utc).isoformat(),
        "source_scope": {
            "radar_readback": "docs/operations/ai-technology-radar-readback.snapshot.json",
            "technology_watch_report": radar_readback["source_scope"]["technology_watch_report"],
            "frontend_target": "/zh-TW/governance?tab=agent-market",
            "gitea_main_evidence_basis_commit": evidence_commit,
            "scope_note": "本讀回只整合已提交的 AI 技術雷達、日週月報契約與治理 gate；不包含 raw chat history、secret、session 或本機工作視窗內容。",
        },
        "summary": {
            "overall_completion_percent": float(summary.get("overall_completion_percent", 42.2)),
            "report_cadence_completion_percent": 100.0,
            "report_cadence_count": len(report_cadences),
            "report_ready_count": len(report_cadences),
            "chart_section_count": 6,
            "agent_status_report_count": len(agent_workload_reports),
            "post_report_analysis_packet_count": len(post_report_packets),
            "low_medium_auto_action_proposal_count": 6,
            "high_risk_owner_review_count": 5,
            "technology_count": technology_count,
            "source_count": source_count,
            "source_failures": source_failures,
            "telegram_send_enabled": False,
            "live_delivery_count_24h": 0,
            "report_receipt_write_count_24h": 0,
            "auto_optimization_write_count": 0,
            "policy_hold_count": len(policy_holds) + len(blocked_gates),
            "status": "daily_weekly_monthly_reports_ready_no_send_gated",
        },
        "policy": {
            "read_only": True,
            "raw_chat_history_synced": False,
            "raw_report_payload_display_allowed": False,
            "report_delivery_enabled": False,
            "telegram_send_enabled": False,
            "bot_api_call_enabled": False,
            "report_receipt_write_enabled": False,
            "ai_post_report_analysis_live_run_enabled": False,
            "low_medium_runtime_auto_write_enabled": False,
            "high_risk_owner_review_required": True,
            "sdk_installation_approved": False,
            "paid_api_calls_approved": False,
            "production_routing_approved": False,
            "model_provider_switch_approved": False,
            "host_write_approved": False,
            "openclaw_replacement_approved": False,
        },
        "report_cadences": report_cadences,
        "agent_workload_reports": agent_workload_reports,
        "chart_sections": _chart_sections(radar_readback),
        "post_report_analysis_packets": post_report_packets,
        "risk_automation_policy": _risk_automation_policy(),
        "telegram_report_bridge": {
            "integration_status": "approval_package_ready_no_live_send",
            "canonical_group_env": "SRE_GROUP_CHAT_ID",
            "telegram_send_enabled": False,
            "bot_api_call_enabled": False,
            "report_receipt_write_enabled": False,
            "draft_count": 3,
            "drafts": [
                {
                    "draft_id": "ai_technology_daily_digest",
                    "cadence": "daily",
                    "status": "no_send_draft_ready",
                    "required_gate": "telegram_send_approved=true",
                },
                {
                    "draft_id": "ai_technology_weekly_scorecard",
                    "cadence": "weekly",
                    "status": "no_send_draft_ready",
                    "required_gate": "telegram_send_approved=true",
                },
                {
                    "draft_id": "ai_technology_monthly_strategy_review",
                    "cadence": "monthly",
                    "status": "no_send_draft_ready",
                    "required_gate": "telegram_send_approved=true",
                },
            ],
            "delivery_note": "目前只產生 Telegram Bot 送出前的審核包與 no-send 草稿；沒有呼叫 Bot API，也沒有寫入 receipt。",
        },
        "blocked_gates": [
            "telegram_send_enabled=false",
            "bot_api_call_enabled=false",
            "report_receipt_write_enabled=false",
            "ai_post_report_analysis_live_run_enabled=false",
            "low_medium_runtime_auto_write_enabled=false",
            "high_risk_owner_review_required=true",
            "sdk_installation_approved=false",
            "paid_api_calls_approved=false",
            "production_routing_approved=false",
            "model_provider_switch_approved=false",
            "host_write_approved=false",
            "openclaw_replacement_approved=false",
        ],
        "next_allowed_actions": [
            "顯示日報、週報、月報 readback 與圖表化摘要",
            "產生 no-send Telegram 草稿與 owner review packet",
            "讓 Hermes / MarketRadar / Critic 讀取 committed reports 後輸出建議",
            "把低中風險項目先轉成文件、scorecard 或 sandbox 提案",
        ],
        "forbidden_actions_without_new_approval": [
            "直接發送 Telegram live report",
            "寫入 report receipt 或 owner acceptance event bus",
            "執行低中風險 runtime write、host write、K8s write 或 workflow trigger",
            "安裝 SDK / MCP server / package",
            "切換模型 provider、生產路由或 OpenClaw 決策核心",
        ],
        "report_contract": {
            "api_endpoint": "/api/v1/agents/ai-technology-report-cadence-readback",
            "frontend_target": "/zh-TW/governance?tab=agent-market",
            "source_endpoint": "/api/v1/agents/ai-technology-radar-readback",
            "daily": "每日顯示來源失敗、版本變更、審核佇列、低中風險建議與 Telegram no-send 草稿。",
            "weekly": "每週顯示技術 scorecard、Agent 工作量、sandbox / replay / adapter design 優先級。",
            "monthly": "每月顯示 roadmap / watch-only / retire 建議與高風險 owner review 包。",
            "telegram": "僅建立審核包與草稿；live send 需要獨立 Telegram delivery approval gate。",
        },
    }


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"- 報告節奏完成度：`{summary['report_cadence_completion_percent']}%`",
        f"- 報告節奏：`{summary['report_ready_count']}/{summary['report_cadence_count']}`",
        f"- 圖表區塊：`{summary['chart_section_count']}`",
        f"- Agent 工作狀態報告：`{summary['agent_status_report_count']}`",
        f"- 報告後 AI 分析包：`{summary['post_report_analysis_packet_count']}`",
        f"- 低中風險自動化提案：`{summary['low_medium_auto_action_proposal_count']}`",
        f"- 高風險 owner review：`{summary['high_risk_owner_review_count']}`",
        f"- Telegram live send：`{summary['telegram_send_enabled']}`",
        f"- 正式送出 / receipt / auto optimization 寫入：`{summary['live_delivery_count_24h']}` / `{summary['report_receipt_write_count_24h']}` / `{summary['auto_optimization_write_count']}`",
        f"- 狀態：`{summary['status']}`",
        "",
        "## 報告節奏",
        "",
        "| 節奏 | Owner Agent | 內容 | AI 看完後做什麼 | Telegram |",
        "|---|---|---|---|---|",
    ]
    for cadence in payload["report_cadences"]:
        lines.append(
            f"| {cadence['cadence_label']} | {cadence['owner_agent']} | "
            f"{cadence['report_purpose']} | {cadence['ai_analysis_after_reading']} | "
            f"`{cadence['telegram_delivery_status']}` |"
        )

    lines.extend([
        "",
        "## Agent 工作狀態",
        "",
        "| Agent | 專業責任 | 工作量 | 產出 | 下一步 |",
        "|---|---|---:|---|---|",
    ])
    for row in payload["agent_workload_reports"]:
        lines.append(
            f"| {row['agent']} | {row['professional_responsibility']} | "
            f"`{row['work_unit_count']}` | {row['latest_output']} | {row['next_action']} |"
        )

    lines.extend([
        "",
        "## 報告後 AI 分析包",
        "",
        "| 報告 | 風險 | 發現 | 解法 | 執行邊界 |",
        "|---|---|---|---|---|",
    ])
    for packet in payload["post_report_analysis_packets"]:
        lines.append(
            f"| {packet['report_id']} | `{packet['risk_tier']}` | "
            f"{packet['key_finding']} | {packet['proposed_solution']} | {packet['execution_boundary']} |"
        )

    lines.extend([
        "",
        "## 風險分層自動化政策",
        "",
        "| 風險 | Agent 可自動做 | 禁止 | 回報 |",
        "|---|---|---|---|",
    ])
    for row in payload["risk_automation_policy"]:
        lines.append(
            f"| `{row['risk_tier']}` | {row['agent_auto_scope']} | "
            f"{row['blocked_without_approval']} | {row['reporting_mode']} |"
        )

    lines.extend([
        "",
        "## 仍被 Gate 擋下",
        "",
    ])
    for gate in payload["blocked_gates"]:
        lines.append(f"- `{gate}`")
    lines.append("")
    return "\n".join(lines)


def _report_cadences(radar: dict[str, Any]) -> list[dict[str, Any]]:
    summary = radar["summary"]
    return [
        {
            "cadence": "daily",
            "cadence_label": "日報",
            "owner_agent": "Hermes",
            "reviewer_agent": "MarketRadar",
            "report_purpose": "彙整 6 小時雷達輸出、來源失敗、版本變更、審核佇列與低中風險處理建議。",
            "data_inputs": [
                "ai_technology_radar_readback.summary",
                "technology_domains",
                "high_priority_review_queue",
                "rolling_update_controls",
            ],
            "metrics": {
                "technology_count": int(summary.get("technology_count", 0)),
                "source_count": int(summary.get("source_count", 0)),
                "source_failures": int(summary.get("source_failures", 0)),
                "changed_technologies": int(summary.get("changed_technologies", 0)),
            },
            "chart_types": ["來源成功率 KPI", "技術領域柱狀圖", "審核佇列狀態列"],
            "ai_analysis_after_reading": "Hermes 先摘要，MarketRadar 標示 freshness，Critic 檢查是否可低風險自動處理。",
            "low_medium_auto_actions": [
                "建立 no-send report-source-gap 提案",
                "更新 watch-only 分類建議",
            ],
            "high_risk_owner_review_actions": [
                "SDK / API / provider / Telegram / host write 全部送 owner review",
            ],
            "telegram_delivery_status": "no_send_draft_ready",
        },
        {
            "cadence": "weekly",
            "cadence_label": "週報",
            "owner_agent": "MarketRadar",
            "reviewer_agent": "NemoTron",
            "report_purpose": "刷新技術 scorecard，判斷 sandbox、offline replay、adapter design 與 watchlist 優先級。",
            "data_inputs": [
                "integration_candidates",
                "professional_agent_roles",
                "blocked_gates",
                "source_scope",
            ],
            "metrics": {
                "integration_candidate_count": len(radar.get("integration_candidates") or []),
                "high_priority_count": int(summary.get("high_priority_count", 0)),
                "review_queue_count": int(summary.get("review_queue_count", 0)),
                "blocked_gate_count": len(radar.get("blocked_gates") or []),
            },
            "chart_types": ["候選技術矩陣", "Agent 工作量堆疊圖", "風險 gate 熱點圖"],
            "ai_analysis_after_reading": "NemoTron 只做離線 replay / scorecard 草稿，OpenClaw 只讀審核分數並維持 policy guard。",
            "low_medium_auto_actions": [
                "產生 sandbox / adapter design 草案",
                "更新 replay fixture 準備清單",
            ],
            "high_risk_owner_review_actions": [
                "進 shadow/canary、production routing 或 OpenClaw 替換 ADR 前必須 owner approval",
            ],
            "telegram_delivery_status": "no_send_scorecard_ready",
        },
        {
            "cadence": "monthly",
            "cadence_label": "月報",
            "owner_agent": "OpenClaw",
            "reviewer_agent": "Critic / Reviewer",
            "report_purpose": "彙整市場趨勢，提出 roadmap / watch-only / retire / replacement-review 策略包。",
            "data_inputs": [
                "technology_area_counts",
                "rolling_update_controls",
                "risk_automation_policy",
                "telegram_report_bridge",
            ],
            "metrics": {
                "technology_area_count": int(summary.get("technology_area_count", 0)),
                "high_priority_count": int(summary.get("high_priority_count", 0)),
                "policy_hold_count": len(radar.get("blocked_gates") or []),
                "openclaw_replacement_approved": 0,
            },
            "chart_types": ["Roadmap 決策漏斗", "Watch-only 保留/淘汰矩陣", "高風險審核包狀態圖"],
            "ai_analysis_after_reading": "OpenClaw 只做仲裁與保守決策，Critic 對策略包做反例、成本、資安與資料邊界審查。",
            "low_medium_auto_actions": [
                "整理 roadmap candidate 文件草案",
                "標記 watch-only 技術保留原因",
            ],
            "high_risk_owner_review_actions": [
                "OpenClaw 替換、provider 切換、付費 API 與 Telegram live delivery 都維持 owner review",
            ],
            "telegram_delivery_status": "no_send_strategy_review_ready",
        },
    ]


def _agent_workload_reports(
    *,
    source_count: int,
    technology_count: int,
    high_priority_count: int,
    blocked_gate_count: int,
) -> list[dict[str, Any]]:
    return [
        {
            "agent": "MarketRadar",
            "professional_responsibility": "AI 技術市場來源監控、版本 freshness、release / docs 變更分類",
            "work_unit_count": source_count,
            "work_unit_label": "primary source checks",
            "latest_output": "20 項技術 / 47 sources / 0 failures 的雷達 readback",
            "next_action": "維持每 6 小時只讀監控，變更進日報與週報 scorecard。",
        },
        {
            "agent": "Hermes",
            "professional_responsibility": "日週月報草稿、RAG 整理、知識庫與 no-send Telegram 草稿",
            "work_unit_count": 3,
            "work_unit_label": "report cadences",
            "latest_output": "日報、週報、月報三份 no-send digest 契約",
            "next_action": "把報告摘要整理成 owner review packet；不得同步 raw chat history。",
        },
        {
            "agent": "NemoTron",
            "professional_responsibility": "離線 replay 評估、模型能力比較、contract smoke gate",
            "work_unit_count": high_priority_count,
            "work_unit_label": "high-priority review candidates",
            "latest_output": "只讀 scorecard / replay fixture 準備清單",
            "next_action": "僅在 no-cost/no-write sandbox 中產生評估草稿；不進 production routing。",
        },
        {
            "agent": "OpenClaw",
            "professional_responsibility": "生產決策仲裁、風險分級、取代/路由/策略 gate",
            "work_unit_count": blocked_gate_count,
            "work_unit_label": "policy gates guarded",
            "latest_output": "維持 OpenClaw production baseline；拒絕無證據替換。",
            "next_action": "等待 replay / shadow / canary 與 owner approval 後才可做高風險決策。",
        },
        {
            "agent": "Critic / Reviewer",
            "professional_responsibility": "反例檢查、成本/資安/資料邊界、報告可信度評分",
            "work_unit_count": technology_count,
            "work_unit_label": "technology fit checks",
            "latest_output": "低中高風險政策分層與 blocked gate 稽核",
            "next_action": "對每份報告輸出 candidate / owner_review / blocked，不直接執行寫入。",
        },
    ]


def _chart_sections(radar: dict[str, Any]) -> list[dict[str, Any]]:
    summary = radar["summary"]
    return [
        {
            "chart_id": "source_health_kpi",
            "title": "來源健康 KPI",
            "chart_type": "metric_strip",
            "primary_metric": "source_failures",
            "value": int(summary.get("source_failures", 0)),
            "expected_report_signal": "來源失敗大於 0 時進日報與 owner review queue。",
        },
        {
            "chart_id": "technology_domain_bar",
            "title": "技術領域覆蓋圖",
            "chart_type": "bar",
            "primary_metric": "technology_domains",
            "value": len(radar.get("technology_domains") or []),
            "expected_report_signal": "每個領域至少要能回溯代表技術、優先級與 changed count。",
        },
        {
            "chart_id": "agent_workload_stack",
            "title": "Agent 工作量堆疊圖",
            "chart_type": "stacked_bar",
            "primary_metric": "work_unit_count",
            "value": 5,
            "expected_report_signal": "每個 Agent 必須有工作量、產出與下一步。",
        },
        {
            "chart_id": "risk_action_matrix",
            "title": "風險處置矩陣",
            "chart_type": "matrix",
            "primary_metric": "risk_tiers",
            "value": 3,
            "expected_report_signal": "低中風險可產生提案，高風險必須 owner review。",
        },
        {
            "chart_id": "telegram_delivery_gate",
            "title": "Telegram 送出 Gate",
            "chart_type": "gate_cards",
            "primary_metric": "telegram_send_enabled",
            "value": 0,
            "expected_report_signal": "no-send 草稿可見，但 live send / receipt write 保持 false。",
        },
        {
            "chart_id": "post_report_analysis_flow",
            "title": "報告後 AI 分析流程",
            "chart_type": "flow",
            "primary_metric": "post_report_analysis_packets",
            "value": 3,
            "expected_report_signal": "每份報告都要產生 finding、solution、risk tier 與 execution boundary。",
        },
    ]


def _post_report_analysis_packets(radar: dict[str, Any]) -> list[dict[str, Any]]:
    summary = radar["summary"]
    return [
        {
            "report_id": "daily",
            "risk_tier": "low",
            "key_finding": f"{summary.get('source_failures', 0)} 個來源失敗，{summary.get('changed_technologies', 0)} 個技術變更。",
            "proposed_solution": "維持只讀監控；若來源失敗大於 0，自動建立 report-source-gap 草案與 no-send Telegram 摘要。",
            "agent_decision": "agent_auto_propose_no_write",
            "execution_boundary": "可產生草稿與文件提案；不得 live send、不得改 workflow、不得打外部付費 API。",
        },
        {
            "report_id": "weekly",
            "risk_tier": "medium",
            "key_finding": f"{summary.get('high_priority_count', 0)} 個高優先級技術需要週期性 scorecard。",
            "proposed_solution": "由 MarketRadar 產生 scorecard，NemoTron 產生離線 replay fixture 準備包，Critic 評成本與資安。",
            "agent_decision": "agent_auto_prepare_owner_packet",
            "execution_boundary": "可準備 sandbox / replay 設計包；不得安裝 SDK、不得啟動 replay runner、不得切路由。",
        },
        {
            "report_id": "monthly",
            "risk_tier": "high",
            "key_finding": "策略層可能涉及 roadmap、provider、OpenClaw 替換或 Telegram delivery policy。",
            "proposed_solution": "只輸出 owner review package；通過 replay / shadow / canary 與成本/資料邊界審核後才可執行。",
            "agent_decision": "owner_review_required",
            "execution_boundary": "高風險全部禁止自動寫入；必須人工批准後另開 execution gate。",
        },
    ]


def _risk_automation_policy() -> list[dict[str, Any]]:
    return [
        {
            "risk_tier": "low",
            "agent_auto_scope": "來源分類、摘要、no-send 草稿、文件與 snapshot 提案。",
            "blocked_without_approval": "不得 live send、不得寫 production、不得呼叫付費 API。",
            "reporting_mode": "自動進日報，並在前端顯示處理建議。",
        },
        {
            "risk_tier": "medium",
            "agent_auto_scope": "scorecard、sandbox 設計、replay fixture 準備包與 owner review draft。",
            "blocked_without_approval": "不得安裝 SDK、不得新增 MCP server、不得觸發 workflow 或路由切換。",
            "reporting_mode": "進週報與 Telegram no-send 草稿，等待 owner review。",
        },
        {
            "risk_tier": "high",
            "agent_auto_scope": "只允許風險分析、反例檢查與人工審核包。",
            "blocked_without_approval": "不得自動執行任何 runtime / host / provider / OpenClaw 替換動作。",
            "reporting_mode": "進月報與高風險 owner review，不做自動處理。",
        },
    ]


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 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 report cadence readback.")
    parser.add_argument("--radar-readback", 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_report_cadence(
        radar_readback=load_json(Path(args.radar_readback)),
        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_REPORT_CADENCE_READBACK_OK "
        f"overall={payload['summary']['overall_completion_percent']}% "
        f"reports={payload['summary']['report_ready_count']}/{payload['summary']['report_cadence_count']} "
        f"charts={payload['summary']['chart_section_count']} "
        f"agents={payload['summary']['agent_status_report_count']} "
        f"telegram_send={payload['summary']['telegram_send_enabled']}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())