fix(api): 在週報顯示報表資料源沉澱

apps/api/src/services/telegram_gateway.py

@@ -28,7 +28,7 @@ import html
 import json
 import os
 import re
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from uuid import NAMESPACE_URL, UUID, uuid5
 
@@ -3888,6 +3888,11 @@ class WeeklyReportMessage:
     git_source_ok: bool = True
     cost_source_ok: bool = False
     all_zero_actionable_anomaly: bool = False
+    report_source_confidence_percent: int = 0
+    report_source_ok_count: int = 0
+    report_source_total_count: int = 0
+    report_source_gap_ids: list[str] = field(default_factory=list)
+    report_asset_state_lines: list[str] = field(default_factory=list)
 
     def format(self) -> str:
         """格式化為 Telegram HTML"""
@@ -3935,6 +3940,22 @@ class WeeklyReportMessage:
             f"├ {html.escape(item)}" if index < len(source_gaps) - 1 else f"└ {html.escape(item)}"
             for index, item in enumerate(source_gaps[:5])
         )
+        source_health_block = ""
+        if self.report_source_total_count > 0:
+            gap_text = ", ".join(self.report_source_gap_ids[:4]) if self.report_source_gap_ids else "none"
+            asset_lines = self.report_asset_state_lines[:5] or ["尚未讀到資產沉澱 read model"]
+            formatted_assets = "\n".join(
+                f"├ {html.escape(item)}" if index < len(asset_lines) - 1 else f"└ {html.escape(item)}"
+                for index, item in enumerate(asset_lines)
+            )
+            source_health_block = (
+                f"━━━━━━━━━━━━━━━━━━━\n"
+                f"🧾 <b>報表資料源 / 沉澱</b>\n"
+                f"├ 來源: <code>{self.report_source_ok_count}/{self.report_source_total_count}</code> "
+                f"| 信心: <code>{self.report_source_confidence_percent}%</code>\n"
+                f"├ 缺口: <code>{html.escape(gap_text)}</code>\n"
+                f"{formatted_assets}\n"
+            )
 
         message = (
             f"═══════════════════════════\n"
@@ -3974,6 +3995,7 @@ class WeeklyReportMessage:
             f"━━━━━━━━━━━━━━━━━━━\n"
             f"🧩 <b>資料缺口 / 下一步</b>\n"
             f"{gap_lines}\n"
+            f"{source_health_block}"
             f"只讀判讀：不自動改排程、不直接發修復、不取代人工批准。\n"
         )
 

apps/api/src/services/weekly_report_service.py

@@ -210,6 +210,35 @@ class WeeklyReportService:
         except Exception as _disp_e:
             logger.warning("weekly_report_disposition_failed", error=str(_disp_e))
 
+        report_source_confidence = 0
+        report_source_ok = 0
+        report_source_total = 0
+        report_source_gap_ids: list[str] = []
+        report_asset_state_lines: list[str] = []
+        try:
+            from src.services.ai_agent_report_source_health import build_ai_agent_report_source_health
+
+            source_health = await build_ai_agent_report_source_health(days=7)
+            source_rollups = source_health.get("rollups") or {}
+            report_source_confidence = int(source_rollups.get("confidence_percent") or 0)
+            report_source_ok = int(source_rollups.get("source_ok_count") or 0)
+            report_source_total = int(source_rollups.get("source_count") or 0)
+            report_source_gap_ids = [
+                str(source.get("work_item_id"))
+                for source in source_health.get("source_health", [])
+                if source.get("work_item_id")
+            ][:5]
+            report_asset_state_lines = [
+                (
+                    f"{asset.get('label')}: {asset.get('state')} "
+                    f"{int(asset.get('done_count') or 0)}/"
+                    f"{int(asset.get('done_count') or 0) + int(asset.get('blocked_count') or 0)}"
+                )
+                for asset in source_health.get("automation_assets", [])
+            ][:5]
+        except Exception as _source_e:
+            logger.warning("weekly_report_source_health_failed", error=str(_source_e))
+
         # 組裝週報
         report = WeeklyReportMessage(
             week_range=week_range,
@@ -246,6 +275,11 @@ class WeeklyReportService:
                 and deploys == 0
                 and disp_total == 0
             ),
+            report_source_confidence_percent=report_source_confidence,
+            report_source_ok_count=report_source_ok,
+            report_source_total_count=report_source_total,
+            report_source_gap_ids=report_source_gap_ids,
+            report_asset_state_lines=report_asset_state_lines,
         )
 
         logger.info(

apps/api/tests/test_telegram_message_templates.py

@@ -453,6 +453,49 @@ def test_weekly_report_keeps_nonzero_source_status_visible() -> None:
     assert "Tokens: <code>1,200</code>" in body
 
 
+def test_weekly_report_includes_report_source_health_assets() -> None:
+    report = WeeklyReportMessage(
+        week_range="2026-W25",
+        report_date="2026-06-18 19:40",
+        alert_total=4,
+        ai_proposal_count=1,
+        commits_count=2,
+        deploy_count=1,
+        stats_source_ok=True,
+        k3s_source_ok=True,
+        git_source_ok=True,
+        cost_source_ok=False,
+        report_source_confidence_percent=40,
+        report_source_ok_count=2,
+        report_source_total_count=5,
+        report_source_gap_ids=[
+            "report-source-gap:incident_summary",
+            "report-source-gap:resolution_stats",
+            "report-source-gap:ai_performance",
+        ],
+        report_asset_state_lines=[
+            "KM: draft_ready 3/6",
+            "PlayBook: draft_required 0/3",
+            "腳本: readback_only 1/1",
+            "排程: no_send_preview 3/3",
+            "Verifier: source_health_ready 1/4",
+        ],
+    )
+
+    body = report.format()
+
+    assert "報表資料源 / 沉澱" in body
+    assert "來源: <code>2/5</code>" in body
+    assert "信心: <code>40%</code>" in body
+    assert "report-source-gap:incident_summary" in body
+    assert "report-source-gap:resolution_stats" in body
+    assert "report-source-gap:ai_performance" in body
+    assert "KM: draft_ready 3/6" in body
+    assert "PlayBook: draft_required 0/3" in body
+    assert "Verifier: source_health_ready 1/4" in body
+    assert "不自動改排程" in body
+
+
 def test_telegram_html_chunks_preserve_complete_lines() -> None:
     """歷史/詳情長訊息不得用 text[:500] 切壞 HTML tag。"""
     lines = [