feat(governance): 新增 AI 技術雷達日週月報讀回

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

@@ -44,6 +44,9 @@ from src.services.ai_agent_market_radar_readback import (
 from src.services.ai_technology_radar_readback import (
     load_latest_ai_technology_radar_readback,
 )
+from src.services.ai_technology_report_cadence_readback import (
+    load_latest_ai_technology_report_cadence_readback,
+)
 from src.services.agent_service import (
     AgentService,
     TaskState,
@@ -750,6 +753,36 @@ async def get_ai_technology_radar_readback() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/ai-technology-report-cadence-readback",
+    response_model=dict[str, Any],
+    summary="取得 AI 技術雷達日週月報與報告後分析讀回",
+    description=(
+        "讀取最新已提交的 AI 技術雷達日報、週報、月報 readback；"
+        "此端點只呈現報告節奏、Agent 工作狀態、圖表化摘要、報告後 AI 分析包、"
+        "低中高風險處理邊界與 Telegram no-send 審核包。"
+        "它不送 Telegram、不寫 receipt、不呼叫 Bot API、不執行低中風險 runtime write、"
+        "不安裝 SDK、不呼叫付費 API、不切換模型、不改主機、不修改 production routing、不替換 OpenClaw。"
+    ),
+)
+async def get_ai_technology_report_cadence_readback() -> dict[str, Any]:
+    """回傳 AI 技術雷達日週月報與報告後分析只讀快照。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_ai_technology_report_cadence_readback)
+        return redact_public_lan_topology(payload)
+    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_technology_report_cadence_readback_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="AI 技術雷達日週月報 readback 無效",
+        ) from exc
+
+
 @router.get(
     "/automation-inventory-snapshot",
     response_model=dict[str, Any],

apps/api/src/services/ai_technology_report_cadence_readback.py

@@ -0,0 +1,108 @@
+"""
+AI technology report cadence readback.
+
+Loads the committed daily / weekly / monthly AI technology report artifact.
+This surface is no-send and no-write: it does not deliver Telegram messages,
+write report receipts, call model providers, install SDKs, modify hosts, or
+change production routing.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_operations_dir
+
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_SNAPSHOT_NAME = "ai-technology-report-cadence-readback.snapshot.json"
+
+
+def load_latest_ai_technology_report_cadence_readback(
+    operations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the committed AI technology report cadence readback snapshot."""
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
+    snapshot_path = directory / _SNAPSHOT_NAME
+    with snapshot_path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if not isinstance(payload, dict):
+        raise ValueError(f"{snapshot_path}: expected JSON object")
+    if payload.get("schema_version") != "ai_technology_report_cadence_readback_v1":
+        raise ValueError(f"{snapshot_path}: unexpected schema_version")
+
+    policy = payload.get("policy") or {}
+    forbidden_true = [
+        key
+        for key in [
+            "raw_chat_history_synced",
+            "raw_report_payload_display_allowed",
+            "report_delivery_enabled",
+            "telegram_send_enabled",
+            "bot_api_call_enabled",
+            "report_receipt_write_enabled",
+            "ai_post_report_analysis_live_run_enabled",
+            "low_medium_runtime_auto_write_enabled",
+            "sdk_installation_approved",
+            "paid_api_calls_approved",
+            "production_routing_approved",
+            "model_provider_switch_approved",
+            "host_write_approved",
+            "openclaw_replacement_approved",
+        ]
+        if policy.get(key) is not False
+    ]
+    if forbidden_true:
+        raise ValueError(f"{snapshot_path}: unsafe policy flags: {forbidden_true}")
+    if policy.get("read_only") is not True:
+        raise ValueError(f"{snapshot_path}: read_only policy must be true")
+    if policy.get("high_risk_owner_review_required") is not True:
+        raise ValueError(f"{snapshot_path}: high risk owner review must remain required")
+
+    summary = payload.get("summary") or {}
+    zero_fields = [
+        "live_delivery_count_24h",
+        "report_receipt_write_count_24h",
+        "auto_optimization_write_count",
+    ]
+    nonzero = [field for field in zero_fields if summary.get(field) != 0]
+    if nonzero:
+        raise ValueError(f"{snapshot_path}: no-write summary fields must stay zero: {nonzero}")
+    if summary.get("telegram_send_enabled") is not False:
+        raise ValueError(f"{snapshot_path}: telegram_send_enabled must stay false")
+
+    cadences = payload.get("report_cadences") or []
+    cadence_ids = {row.get("cadence") for row in cadences}
+    if cadence_ids != {"daily", "weekly", "monthly"}:
+        raise ValueError(f"{snapshot_path}: report cadences must cover daily, weekly, monthly")
+    if len(payload.get("post_report_analysis_packets") or []) != 3:
+        raise ValueError(f"{snapshot_path}: post report analysis packets must cover 3 reports")
+
+    telegram_bridge = payload.get("telegram_report_bridge") or {}
+    if telegram_bridge.get("telegram_send_enabled") is not False:
+        raise ValueError(f"{snapshot_path}: Telegram bridge send must stay false")
+    if telegram_bridge.get("bot_api_call_enabled") is not False:
+        raise ValueError(f"{snapshot_path}: Telegram bridge bot API calls must stay false")
+    if telegram_bridge.get("report_receipt_write_enabled") is not False:
+        raise ValueError(f"{snapshot_path}: Telegram bridge receipt writes must stay false")
+
+    serialized = json.dumps(payload, ensure_ascii=False)
+    forbidden_fragments = [
+        "/Users/",
+        ".claude/projects",
+        ".codex",
+        "192.168.",
+        "auth.json",
+        "conversations",
+        "sessions",
+        "批准！繼續",
+        "My request for Codex",
+        "In app browser",
+    ]
+    leaked = [fragment for fragment in forbidden_fragments if fragment in serialized]
+    if leaked:
+        raise ValueError(f"{snapshot_path}: forbidden local or raw-history fragment: {leaked}")
+
+    return payload