fix(governance): keep trust drift watchdog on governance agent
Some checks failed
CD Pipeline / tests (push) Successful in 2m51s
Code Review / ai-code-review (push) Successful in 24s
CD Pipeline / build-and-deploy (push) Has started running
CD Pipeline / post-deploy-checks (push) Has been cancelled

This commit is contained in:
Your Name
2026-05-05 14:00:13 +08:00
parent 2ff0ef3bb6
commit 8e22110030
2 changed files with 15 additions and 12 deletions

View File

@@ -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,
]
)

View File

@@ -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",