diff --git a/apps/api/src/jobs/ai_slo_watchdog_job.py b/apps/api/src/jobs/ai_slo_watchdog_job.py index 223b8ea3..0af6664d 100644 --- a/apps/api/src/jobs/ai_slo_watchdog_job.py +++ b/apps/api/src/jobs/ai_slo_watchdog_job.py @@ -204,17 +204,18 @@ async def _check_once() -> None: # W-6: Trust Drift 偵測(Playbook 信任度漂移) # 2026-05-02 ogt + Claude Sonnet 4.6(亞太): 整併雙寫路徑 - # 2026-05-04 ogt + Claude: 修復重複告警 — 改為直接用 trust_drift_detector 純統計 - # 背景:原本呼叫 governance_agent.check_trust_drift() 會觸發 Telegram 告警 - # 但 governance_agent.run_self_check() 每 1h 也會呼叫同一方法 → 雙重 Telegram - # 修正:watchdog 只取統計數字,不觸發 Telegram;告警由 governance_agent 獨家負責 + # 2026-05-05 Codex: Watchdog 仍透過 governance_agent 單一入口, + # 但用 emit_alert=False 只取統計,避免與 hourly self-check 發出雙重 Telegram。 try: - from src.services.trust_drift_detector import get_trust_drift_detector - dist = await get_trust_drift_detector().detect() - if dist.drift_detected: + from src.services.governance_agent import get_governance_agent + trust_result = await get_governance_agent().check_trust_drift(emit_alert=False) + drifted = trust_result.get("drifted", 0) + if drifted > 0: + auto_deprecated = trust_result.get("auto_deprecated", 0) + kept = trust_result.get("kept", 0) violations.append( - f"Trust Drift 偵測到 {dist.low_count} 個 Playbook 信任度低落" - f"(low_ratio: {dist.low_ratio:.1%},mean_trust: {dist.mean_trust:.2f})" + f"Trust Drift 偵測到 {drifted} 個 Playbook 信任度低落" + f"(auto-deprecated: {auto_deprecated},待人工審核: {kept})" ) # 2026-05-05 ogt W6 修復:移除動態 low_count,避免 count 微變繞過 dedup violation_codes.append("W6:trust_drift") @@ -247,7 +248,7 @@ async def _check_once() -> None: system_impact = "\n".join( [ f"檢出 {len(violations)} 項 KPI 異常(W-1~W-6)", - f"關鍵影響:飛輪自動化能力可能降級", + "關鍵影響:飛輪自動化能力可能降級", *violation_lines, ] ) diff --git a/apps/api/src/services/governance_agent.py b/apps/api/src/services/governance_agent.py index 677f7920..cf16b40c 100644 --- a/apps/api/src/services/governance_agent.py +++ b/apps/api/src/services/governance_agent.py @@ -72,13 +72,15 @@ class GovernanceAgent: # 1. Playbook 信任度漂移 # ========================================================================= - async def check_trust_drift(self) -> dict[str, Any]: + async def check_trust_drift(self, emit_alert: bool = True) -> dict[str, Any]: """Playbook trust_score < 0.2 → 告警建議廢棄;30 天沒用過的直接 auto-deprecate 2026-04-26 P2.2 by Claude 2026-05-02 ogt + Claude Sonnet 4.6: 加 auto_deprecate_low_trust_unused 自治路徑 守衛條件:trust < 0.2 AND (last_used_at < 30 天前 OR 從未使用且創建超過 30 天) → status 改 'deprecated',alert 改報「N 個告警 + M 個 auto-deprecated」 + 2026-05-05 Codex: emit_alert=False 供 W-6 watchdog 查詢統計,維持 + governance_agent 單一入口,但避免與 hourly self-check 發出雙重 Telegram。 """ async with get_db_context() as db: result = await db.execute( @@ -118,7 +120,7 @@ class GovernanceAgent: ids=auto_deprecated_ids[:10], ) - if drifted: + if drifted and emit_alert: drift_ratio = len(drifted) / total if total > 0 else 0.0 await self._alert( "trust_drift",