fix(telegram): support matrix source contracts in container

apps/api/src/services/telegram_alert_ai_automation_matrix.py

@@ -355,29 +355,29 @@ def _require_safe_boundaries(*payloads: dict[str, Any]) -> None:
 def _inspect_source_contract(repo_root: Path) -> dict[str, int | bool]:
     source_checks = {
         "telegram_gateway_mirror_present": (
-            repo_root / "apps/api/src/services/telegram_gateway.py",
+            "apps/api/src/services/telegram_gateway.py",
             "_mirror_outbound_message",
         ),
         "telegram_gateway_record_outbound_present": (
-            repo_root / "apps/api/src/services/telegram_gateway.py",
+            "apps/api/src/services/telegram_gateway.py",
             "record_outbound_message",
         ),
         "ai_alert_card_metadata_present": (
-            repo_root / "apps/api/src/services/telegram_gateway.py",
+            "apps/api/src/services/telegram_gateway.py",
             "ai_automation_alert_card_mirror_v1",
         ),
         "outbound_message_model_present": (
-            repo_root / "apps/api/src/db/awooop_models.py",
+            "apps/api/src/db/awooop_models.py",
             "__tablename__ = \"awooop_outbound_message\"",
         ),
         "ai_alert_card_delivery_readback_endpoint_present": (
-            repo_root / "apps/api/src/api/v1/platform/operator_runs.py",
+            "apps/api/src/api/v1/platform/operator_runs.py",
             "/runs/ai-alert-cards",
         ),
     }
     result: dict[str, int | bool] = {}
-    for key, (path, marker) in source_checks.items():
-        text = path.read_text(encoding="utf-8", errors="replace") if path.exists() else ""
+    for key, (relative_path, marker) in source_checks.items():
+        text = _read_source_contract_text(repo_root, relative_path)
         present = marker in text
         result[key] = present
         if key == "ai_alert_card_delivery_readback_endpoint_present":
@@ -389,6 +389,18 @@ def _inspect_source_contract(repo_root: Path) -> dict[str, int | bool]:
     return result
 
 
+def _read_source_contract_text(repo_root: Path, relative_path: str) -> str:
+    candidates = [repo_root / relative_path]
+    api_prefix = "apps/api/"
+    if relative_path.startswith(api_prefix):
+        candidates.append(repo_root / relative_path.removeprefix(api_prefix))
+
+    for path in candidates:
+        if path.exists():
+            return path.read_text(encoding="utf-8", errors="replace")
+    return ""
+
+
 def _next_priority_action(
     *,
     api_direct_gap_count: int,

apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py

@@ -7,6 +7,7 @@ from fastapi.testclient import TestClient
 
 from src.api.v1.agents import router
 from src.services.telegram_alert_ai_automation_matrix import (
+    _inspect_source_contract,
     load_latest_telegram_alert_ai_automation_matrix,
 )
 
@@ -90,6 +91,35 @@ def test_telegram_alert_ai_automation_matrix_loader_returns_gap_matrix():
         assert marker not in serialized
 
 
+def test_telegram_matrix_source_contract_supports_container_layout(tmp_path):
+    (tmp_path / "src/services").mkdir(parents=True)
+    (tmp_path / "src/db").mkdir(parents=True)
+    (tmp_path / "src/api/v1/platform").mkdir(parents=True)
+    (tmp_path / "src/services/telegram_gateway.py").write_text(
+        "_mirror_outbound_message\n"
+        "record_outbound_message\n"
+        "ai_automation_alert_card_mirror_v1\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "src/db/awooop_models.py").write_text(
+        '__tablename__ = "awooop_outbound_message"\n',
+        encoding="utf-8",
+    )
+    (tmp_path / "src/api/v1/platform/operator_runs.py").write_text(
+        '"/runs/ai-alert-cards"\n',
+        encoding="utf-8",
+    )
+
+    proof = _inspect_source_contract(tmp_path)
+
+    assert proof["telegram_gateway_mirror_present"] is True
+    assert proof["telegram_gateway_record_outbound_present"] is True
+    assert proof["ai_alert_card_metadata_present"] is True
+    assert proof["outbound_message_model_present"] is True
+    assert proof["ai_alert_card_delivery_readback_endpoint_present"] is True
+    assert proof["ai_alert_card_delivery_readback_endpoint_count"] == 1
+
+
 def test_telegram_alert_ai_automation_matrix_endpoint_returns_readback():
     app = FastAPI()
     app.include_router(router, prefix="/api/v1")

docs/LOGBOOK.md

@@ -1,3 +1,17 @@
+## 2026-07-02 — 19:06 CIR-P0-TG-001 production matrix source-contract fallback
+
+**完成內容**：
+- 正式 `/api/v1/agents/telegram-alert-ai-automation-matrix` 在 deploy marker `3cf54f3bf chore(cd): deploy fe83181 [skip ci]` 後回 `500`；priority work-order API 已回新狀態，代表 production image 含本輪 Telegram patch，但 matrix source-contract 檢查仍只看 repo layout `apps/api/src/...`。
+- `telegram_alert_ai_automation_matrix._inspect_source_contract()` 已改為同時支援 repo layout `apps/api/src/...` 與 production container layout `/app/src/...`，避免本地 loader OK、container 找不到 marker 而 500。
+- 新增 regression test，建立只有 `src/services`、`src/db`、`src/api/v1/platform` 的 container-like tmp layout，確認 TelegramGateway mirror、outbound model 與 `/runs/ai-alert-cards` marker 都能讀回。
+
+**驗證**：
+- `DATABASE_URL=postgresql+asyncpg://test:test@localhost/test PYTHONPATH=apps/api python3.11 -m pytest apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py -q -p no:cacheprovider`：`3 passed`。
+- `python3.11 -m py_compile apps/api/src/services/telegram_alert_ai_automation_matrix.py apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py`、`git diff --check`：通過。
+
+**仍維持**：
+- 未送 Telegram、未呼叫 Bot API、未讀 secret / token / `.env` / raw sessions / SQLite / auth；未寫 production DB、未改 receiver route、未 workflow_dispatch、未重啟主機 / VM / Docker / Nginx / K3s / DB / firewall。
+
 ## 2026-07-02 — 18:53 CIR-P0-TG-001 channel_hub direct Telegram sender 收斂
 
 **完成內容**：