614 lines
23 KiB
Python
614 lines
23 KiB
Python
"""
|
||
GovernanceDispatcher — 治理事件 → 修復派遣
|
||
============================================
|
||
Poll 模式：每 30s 掃 ai_governance_events 中 resolved=False 且
|
||
無活躍 dispatch 的事件，呼叫 DecisionFusionAdapter 三維融合後
|
||
寫入 governance_remediation_dispatch 表。
|
||
|
||
職責：
|
||
1. Poll unresolved 治理事件（不直接修改 ai_governance_events 表）
|
||
2. 呼叫 DecisionFusionAdapter.fuse_decision → FusedDecision
|
||
3. 依 decision_path 決定是否寫入 dispatch
|
||
4. 不執行 remediation（實際執行由 approval_execution / auto_repair 消費 dispatch 表）
|
||
|
||
Tier 3 鐵線（絕不觸碰）：
|
||
- decision_manager.py / learning_service.py / trust_engine.py
|
||
- 本模組透過 DecisionFusionAdapter（wrapper）間接使用這些能力
|
||
|
||
2026-05-03 ogt + Claude Sonnet 4.6（亞太）: GovernanceDispatcher Wave 2E 實作
|
||
2026-05-04 ogt + Claude Sonnet 4.6（亞太）: skip 路徑無限迴圈修復
|
||
- skip 決策後設 Redis 90min 冷卻，避免重複 LLM 呼叫
|
||
- 超過 2 小時的 stale skip 事件標記 resolved=True（新事件若問題持續會重新產生）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
import structlog
|
||
from sqlalchemy import exists, select, update
|
||
|
||
from src.db.base import get_db_context
|
||
from src.db.models import AiGovernanceEvent, GovernanceRemediationDispatch
|
||
from src.repositories.governance_remediation_dispatch_repo import (
|
||
ACTIVE_STATUSES,
|
||
DispatchAlreadyActive,
|
||
create_dispatch,
|
||
get_active_for_event,
|
||
list_by_event,
|
||
transition_status,
|
||
)
|
||
from src.services.decision_fusion_adapter import FusedDecision, get_decision_fusion_adapter
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# =============================================================================
|
||
# 常數
|
||
# TODO: 移到 settings（ADR-P2E-FUTURE），目前暫時 hardcode
|
||
# =============================================================================
|
||
|
||
# Poll 間隔（秒）
|
||
# TODO: 移到 settings，允許運維不重啟調整 poll 間隔
|
||
_DISPATCHER_INTERVAL_SEC: int = 30
|
||
|
||
# Skip 冷卻時間（秒）：skip 決策後 90 分鐘內不重新評估同一事件
|
||
# 原因：skip = 信心度不足，短期內 playbook trust / MCP 指標不會驟變
|
||
_SKIP_COOLDOWN_SEC: int = 5400 # 90 分鐘
|
||
|
||
# Stale 事件閾值（秒）：超過此時間的 skip 事件直接標 resolved
|
||
# 原因：持久問題會由 governance_agent 重新產生新事件；舊事件繼續留著只是積壓
|
||
_STALE_EVENT_SEC: int = 7200 # 2 小時
|
||
|
||
# 每輪最多處理幾個事件（避免單輪阻塞過長）
|
||
_MAX_EVENTS_PER_CYCLE: int = 10
|
||
|
||
# 允許建立 dispatch 的 event_type（對齊 governance_event_type enum）
|
||
_DISPATCHABLE_EVENT_TYPES: frozenset[str] = frozenset({
|
||
"trust_drift",
|
||
"knowledge_degradation",
|
||
"llm_hallucination",
|
||
"execution_blast_radius",
|
||
"governance_slo_data_gap",
|
||
})
|
||
|
||
|
||
# =============================================================================
|
||
# Redis 冷卻 helpers（防止 skip 事件無限重評迴圈）
|
||
# =============================================================================
|
||
|
||
async def _is_skip_cooldown(event_id: str) -> bool:
|
||
"""確認事件是否在 skip 冷卻期內（90 分鐘）。"""
|
||
try:
|
||
from src.core.redis_client import get_redis
|
||
redis = get_redis()
|
||
return bool(await redis.exists(f"governance:skip:{event_id}"))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
async def _set_skip_cooldown(event_id: str) -> None:
|
||
"""設置 skip 冷卻期（90 分鐘），防止重複 LLM 呼叫。"""
|
||
try:
|
||
from src.core.redis_client import get_redis
|
||
redis = get_redis()
|
||
await redis.setex(f"governance:skip:{event_id}", _SKIP_COOLDOWN_SEC, "1")
|
||
except Exception as exc:
|
||
logger.warning("governance_skip_cooldown_set_failed", event_id=event_id, error=str(exc))
|
||
|
||
|
||
async def _mark_event_resolved(event_id: str, reason: str) -> None:
|
||
"""將 stale skip 事件標記為 resolved（持久問題會由 governance_agent 重新產生新事件）。
|
||
|
||
對齊模型設計：resolved=True 由「下次計算時補填」，
|
||
dispatcher skip = 系統判斷當前無法自動修復，等同一次計算完成。
|
||
"""
|
||
try:
|
||
from src.utils.timezone import now_taipei
|
||
async with get_db_context() as db:
|
||
await db.execute(
|
||
update(AiGovernanceEvent)
|
||
.where(AiGovernanceEvent.id == event_id)
|
||
.where(AiGovernanceEvent.resolved.is_(False))
|
||
.values(resolved=True, resolved_at=now_taipei())
|
||
)
|
||
logger.info(
|
||
"governance_event_stale_resolved",
|
||
event_id=event_id,
|
||
reason=reason,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("governance_event_resolve_failed", event_id=event_id, error=str(exc))
|
||
|
||
|
||
# =============================================================================
|
||
# 核心函數
|
||
# =============================================================================
|
||
|
||
async def dispatch_governance_event(event: AiGovernanceEvent) -> str | None:
|
||
"""處理單一治理事件：決策融合 → 寫 dispatch 記錄。
|
||
|
||
Args:
|
||
event: AiGovernanceEvent ORM 物件（唯讀，不修改）
|
||
|
||
Returns:
|
||
建立的 dispatch_id（str），或 None（skip / 已有活躍 dispatch）
|
||
"""
|
||
event_id = event.id
|
||
event_type = event.event_type
|
||
|
||
# Step 0: 檢查是否已有活躍 dispatch（冪等保護）。
|
||
# 先查 DB 再看 skip cooldown，避免曾經被 skip 的 knowledge_degradation
|
||
# 無法補建 Hermes KB healthcheck intake work item。
|
||
existing = await get_active_for_event(event_id)
|
||
if existing is not None:
|
||
logger.debug(
|
||
"governance_dispatch_skipped_already_active",
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
existing_dispatch_id=existing.id,
|
||
existing_status=existing.dispatch_status,
|
||
)
|
||
return None
|
||
|
||
# knowledge_degradation 的 run_kb_growth_healthcheck 是治理工作項 intake，
|
||
# 不是自動修復執行；先落 pending dispatch，讓既有 unresolved 事件也能在 AwoooP 追蹤。
|
||
if _is_kb_growth_healthcheck_event(event):
|
||
if await _has_kb_growth_review_draft(event_id):
|
||
logger.debug(
|
||
"governance_kb_healthcheck_dispatch_skipped_waiting_review",
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
)
|
||
return None
|
||
return await _record_kb_growth_healthcheck_dispatch(event)
|
||
|
||
# Step 1: Redis skip 冷卻檢查（防止一般 skip 事件每 30s 重新做 LLM 呼叫）。
|
||
if await _is_skip_cooldown(event_id):
|
||
logger.debug(
|
||
"governance_dispatch_skip_cooldown",
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
)
|
||
return None
|
||
|
||
# Step 2: 決策融合（三維：LLM × Playbook × MCP）
|
||
adapter = get_decision_fusion_adapter()
|
||
try:
|
||
decision: FusedDecision = await adapter.fuse_decision(event)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"governance_fusion_failed",
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
error=str(exc),
|
||
)
|
||
# LLM 失敗 fallback：skip + log（不寫 dispatch）
|
||
logger.info(
|
||
"governance_dispatch_fallback_skip",
|
||
event_id=event_id,
|
||
reason="fusion_exception",
|
||
)
|
||
return None
|
||
|
||
# Step 3: 依 decision_path 決定要不要寫 dispatch
|
||
if decision.decision_path == "skip":
|
||
# 2026-05-04 ogt: 修復無限迴圈根因
|
||
# skip 決策後設 90min Redis 冷卻，避免每 30s 重新呼叫 LLM
|
||
# 超過 2h 的 stale 事件直接標 resolved（持久問題由 governance_agent 重新產生新事件）
|
||
await _set_skip_cooldown(event_id)
|
||
|
||
triggered_at_aware = event.triggered_at
|
||
if triggered_at_aware is not None and triggered_at_aware.tzinfo is None:
|
||
triggered_at_aware = triggered_at_aware.replace(tzinfo=timezone.utc)
|
||
event_age_sec = (
|
||
(datetime.now(timezone.utc) - triggered_at_aware).total_seconds()
|
||
if triggered_at_aware is not None else 0
|
||
)
|
||
|
||
logger.info(
|
||
"governance_dispatch_path_skip",
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
confidence=round(decision.confidence, 4),
|
||
event_age_sec=int(event_age_sec),
|
||
stale=event_age_sec > _STALE_EVENT_SEC,
|
||
)
|
||
|
||
await _record_skipped_dispatch(event, decision)
|
||
|
||
if event_age_sec > _STALE_EVENT_SEC:
|
||
await _mark_event_resolved(event_id, reason=f"skip_stale_{int(event_age_sec)}s")
|
||
|
||
return None
|
||
|
||
# Step 4: 決定 executor_type 與 dispatch_status
|
||
# auto_dispatch → dispatched（下游 auto_repair 消費）
|
||
# pending_approval → pending（等人工審核）
|
||
if decision.decision_path == "auto_dispatch":
|
||
executor_type = "playbook_executor"
|
||
else: # pending_approval
|
||
executor_type = "manual"
|
||
|
||
# Step 5: 建構 decision_context JSONB（完整三維快照）
|
||
decision_context = _build_decision_context(event, decision)
|
||
|
||
# Step 6: 寫入 governance_remediation_dispatch（用 repo 函數）
|
||
try:
|
||
dispatch_row = await create_dispatch(
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
executor_type=executor_type,
|
||
playbook_id=decision.matched_playbook_id,
|
||
decision_context=decision_context,
|
||
created_by="governance_dispatcher",
|
||
)
|
||
except DispatchAlreadyActive:
|
||
# 並行 race condition：另一個 worker 先建立了 dispatch
|
||
logger.info(
|
||
"governance_dispatch_race_condition",
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
)
|
||
return None
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"governance_dispatch_create_failed",
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
error=str(exc),
|
||
)
|
||
return None
|
||
|
||
logger.info(
|
||
"governance_dispatched",
|
||
dispatch_id=dispatch_row.id,
|
||
event_id=event_id,
|
||
event_type=event_type,
|
||
decision_path=decision.decision_path,
|
||
confidence=round(decision.confidence, 4),
|
||
executor_type=executor_type,
|
||
playbook_id=decision.matched_playbook_id,
|
||
)
|
||
|
||
return dispatch_row.id
|
||
|
||
|
||
async def _record_kb_growth_healthcheck_dispatch(event: AiGovernanceEvent) -> str | None:
|
||
"""將 knowledge_degradation intake 成 Hermes KB healthcheck work item."""
|
||
try:
|
||
from src.services.governance_agent import (
|
||
_build_knowledge_degradation_dispatch_context,
|
||
)
|
||
|
||
dispatch_row = await create_dispatch(
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
executor_type="hermes_kb_growth_healthcheck",
|
||
decision_context=_build_knowledge_degradation_dispatch_context(
|
||
event.id,
|
||
event.details if isinstance(event.details, dict) else {},
|
||
),
|
||
max_attempts=1,
|
||
created_by="governance_dispatcher_intake",
|
||
)
|
||
except DispatchAlreadyActive:
|
||
logger.info(
|
||
"governance_kb_healthcheck_dispatch_race_condition",
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
)
|
||
return None
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"governance_kb_healthcheck_dispatch_failed",
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
error=str(exc),
|
||
)
|
||
return None
|
||
|
||
logger.info(
|
||
"governance_kb_healthcheck_dispatched",
|
||
dispatch_id=dispatch_row.id,
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
)
|
||
return dispatch_row.id
|
||
|
||
|
||
async def _record_skipped_dispatch(event: AiGovernanceEvent, decision: FusedDecision) -> None:
|
||
"""留下 skipped 派遣 trail，讓治理事件的 AI 判斷不再只存在 log 裡。
|
||
|
||
skip 不代表「事情解決」，而是 DecisionFusion 判斷目前不能自動派遣。
|
||
這筆 terminal dispatch row 讓 AwoooP / Work Items 可顯示：
|
||
AI 已判斷、為何沒有自動處理、下一步需要誰接手。
|
||
"""
|
||
try:
|
||
row = await create_dispatch(
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
executor_type="manual",
|
||
playbook_id=decision.matched_playbook_id,
|
||
decision_context=_build_decision_context(event, decision),
|
||
created_by="governance_dispatcher_skip",
|
||
)
|
||
await transition_status(row.id, "pending", "skipped")
|
||
except DispatchAlreadyActive:
|
||
logger.info(
|
||
"governance_skip_dispatch_race_condition",
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"governance_skip_dispatch_record_failed",
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
error=str(exc),
|
||
)
|
||
|
||
|
||
async def _poll_unresolved_events() -> list[AiGovernanceEvent]:
|
||
"""查詢 unresolved 且 event_type 在 dispatchable 範圍內的治理事件。
|
||
|
||
Returns:
|
||
最多 _MAX_EVENTS_PER_CYCLE 筆 AiGovernanceEvent ORM 物件列表
|
||
"""
|
||
async with get_db_context() as db:
|
||
active_dispatch_exists = exists().where(
|
||
GovernanceRemediationDispatch.governance_event_id == AiGovernanceEvent.id,
|
||
GovernanceRemediationDispatch.dispatch_status.in_(list(ACTIVE_STATUSES)),
|
||
)
|
||
result = await db.execute(
|
||
select(AiGovernanceEvent)
|
||
.where(AiGovernanceEvent.resolved.is_(False))
|
||
.where(AiGovernanceEvent.event_type.in_(list(_DISPATCHABLE_EVENT_TYPES)))
|
||
.where(~active_dispatch_exists)
|
||
.order_by(AiGovernanceEvent.triggered_at.asc())
|
||
.limit(_MAX_EVENTS_PER_CYCLE)
|
||
)
|
||
rows = result.scalars().all()
|
||
|
||
return list(rows)
|
||
|
||
|
||
def _build_decision_context(
|
||
event: AiGovernanceEvent,
|
||
decision: FusedDecision,
|
||
) -> dict[str, Any]:
|
||
"""建構 decision_context JSONB（完整三維輸入快照）。
|
||
|
||
規格對齊 DecisionContextV1（models/governance_dispatch.py），
|
||
但直接建 dict 不依賴 Pydantic model（避免引入額外依賴）。
|
||
|
||
Fields:
|
||
version: schema 版本（v1）
|
||
trigger_source: 觸發來源
|
||
suggested_action: AI 建議的修復動作摘要
|
||
fusion_scores: 三維分數詳情
|
||
llm_reasoning: LLM 原始輸出摘要
|
||
mcp_snapshot: MCP 情報快照
|
||
decision_path: 決策分支
|
||
confidence: 最終融合信心度
|
||
"""
|
||
details = event.details if isinstance(event.details, dict) else {}
|
||
next_action = _extract_event_next_action(details, decision.recommended_action)
|
||
workflow = _build_event_workflow(event, details, decision, next_action)
|
||
ownership = details.get("ownership") if isinstance(details.get("ownership"), dict) else {}
|
||
|
||
return {
|
||
"version": "v1",
|
||
"trigger_source": "governance_dispatcher",
|
||
"triggered_metric": event.event_type,
|
||
"metric_value": decision.confidence,
|
||
"threshold": 0.85, # TODO: 移到 settings
|
||
"suggested_action": decision.recommended_action,
|
||
"next_action": next_action,
|
||
"workflow": workflow,
|
||
"ownership": ownership,
|
||
"fusion_scores": {
|
||
"llm_score": round(decision.llm_score, 4),
|
||
"playbook_score": round(decision.playbook_score, 4),
|
||
"mcp_score": round(decision.mcp_score, 4),
|
||
"confidence": round(decision.confidence, 4),
|
||
"weights": {"llm": 0.4, "playbook": 0.3, "mcp": 0.3}, # TODO: 移到 settings
|
||
},
|
||
"llm_reasoning": decision.llm_reasoning,
|
||
"mcp_snapshot": decision.mcp_snapshot,
|
||
"decision_path": decision.decision_path,
|
||
"matched_playbook_id": decision.matched_playbook_id,
|
||
"playbook_trust": decision.playbook_trust,
|
||
"affected_resources": [event.event_type],
|
||
"extra": {
|
||
"event_id": event.id,
|
||
"event_details_keys": list(details.keys()),
|
||
"ownership": ownership,
|
||
},
|
||
}
|
||
|
||
|
||
def _is_kb_growth_healthcheck_event(event: AiGovernanceEvent) -> bool:
|
||
if event.event_type != "knowledge_degradation":
|
||
return False
|
||
details = event.details if isinstance(event.details, dict) else {}
|
||
remediation = details.get("remediation")
|
||
if isinstance(remediation, dict) and remediation.get("next_action") == "run_kb_growth_healthcheck":
|
||
return True
|
||
return details.get("next_action") == "run_kb_growth_healthcheck"
|
||
|
||
|
||
async def _has_kb_growth_review_draft(event_id: str) -> bool:
|
||
"""同一治理事件已有 Hermes review draft 時，不再重複建立 pending dispatch。"""
|
||
try:
|
||
rows = await list_by_event(event_id)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"governance_kb_healthcheck_history_lookup_failed",
|
||
event_id=event_id,
|
||
error=str(exc),
|
||
)
|
||
return False
|
||
|
||
for row in rows:
|
||
if row.executor_type != "hermes_kb_growth_healthcheck":
|
||
continue
|
||
if row.dispatch_status != "succeeded":
|
||
continue
|
||
context = row.decision_context if isinstance(row.decision_context, dict) else {}
|
||
workflow = context.get("workflow") if isinstance(context.get("workflow"), dict) else {}
|
||
worker_result = (
|
||
context.get("worker_result")
|
||
if isinstance(context.get("worker_result"), dict)
|
||
else {}
|
||
)
|
||
if workflow.get("current_stage") == "waiting_owner_review":
|
||
return True
|
||
if worker_result.get("status") == "draft_created":
|
||
return True
|
||
return False
|
||
|
||
|
||
def _extract_event_next_action(details: dict[str, Any], fallback: str) -> str:
|
||
remediation = details.get("remediation")
|
||
if isinstance(remediation, dict):
|
||
next_action = remediation.get("next_action")
|
||
if isinstance(next_action, str) and next_action:
|
||
return next_action[:160]
|
||
next_action = details.get("next_action")
|
||
if isinstance(next_action, str) and next_action:
|
||
return next_action[:160]
|
||
return fallback[:160]
|
||
|
||
|
||
def _build_event_workflow(
|
||
event: AiGovernanceEvent,
|
||
details: dict[str, Any],
|
||
decision: FusedDecision,
|
||
next_action: str,
|
||
) -> dict[str, Any]:
|
||
"""把治理事件轉成 AwoooP 可讀 workflow metadata。
|
||
|
||
這不是自動修復決策；最終是否執行仍由 DecisionFusion 與 dispatch state 決定。
|
||
"""
|
||
base_steps = ["detected", "ai_analyzed"]
|
||
if next_action == "run_kb_growth_healthcheck":
|
||
steps = [
|
||
*base_steps,
|
||
"queued_kb_healthcheck",
|
||
"draft_km_updates",
|
||
"waiting_owner_review",
|
||
"km_writeback_after_approval",
|
||
"stale_ratio_recheck",
|
||
]
|
||
stage_by_status = {
|
||
"pending": "queued_kb_healthcheck",
|
||
"dispatched": "queued_kb_healthcheck",
|
||
"executing": "draft_km_updates",
|
||
"succeeded": "stale_ratio_recheck",
|
||
"failed": "needs_manual_km_triage",
|
||
"skipped": "waiting_owner_review",
|
||
"cancelled": "cancelled",
|
||
}
|
||
current_stage = (
|
||
"waiting_owner_review"
|
||
if decision.decision_path == "skip"
|
||
else "queued_kb_healthcheck"
|
||
)
|
||
work_kind = "kb_growth_healthcheck"
|
||
else:
|
||
steps = [*base_steps, "dispatch_decision", "operator_review"]
|
||
stage_by_status = {
|
||
"pending": "queued_for_review",
|
||
"dispatched": "dispatched",
|
||
"executing": "executing",
|
||
"succeeded": "completed",
|
||
"failed": "failed",
|
||
"skipped": "skipped",
|
||
"cancelled": "cancelled",
|
||
}
|
||
current_stage = stage_by_status.get("pending", "queued_for_review")
|
||
work_kind = "governance_remediation"
|
||
|
||
return {
|
||
"work_item_id": f"governance:{event.event_type}:{event.id}",
|
||
"work_kind": work_kind,
|
||
"current_stage": current_stage,
|
||
"steps": steps,
|
||
"stage_by_dispatch_status": stage_by_status,
|
||
"next_action": next_action,
|
||
"needs_human_review": decision.decision_path != "auto_dispatch",
|
||
"writes_km_without_approval": False,
|
||
"impact": details.get("impact") if isinstance(details.get("impact"), dict) else {},
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# 排程迴圈（仿 run_governance_loop 模式）
|
||
# =============================================================================
|
||
|
||
async def run_governance_dispatcher_loop(
|
||
interval_seconds: int = _DISPATCHER_INTERVAL_SEC,
|
||
) -> None:
|
||
"""每 30s 掃 unresolved 事件 → dispatch。
|
||
|
||
仿照 governance_agent.run_governance_loop 模式：
|
||
- while True → try/except → sleep
|
||
- 任一事件失敗不阻塞其他事件（獨立 try/except）
|
||
- CancelledError 向上傳播（允許 graceful shutdown）
|
||
|
||
2026-05-03 ogt + Claude Sonnet 4.6（亞太）: Wave 2E 實作
|
||
"""
|
||
logger.info(
|
||
"governance_dispatcher_loop_started",
|
||
interval_seconds=interval_seconds,
|
||
max_events_per_cycle=_MAX_EVENTS_PER_CYCLE,
|
||
)
|
||
|
||
while True:
|
||
try:
|
||
events = await _poll_unresolved_events()
|
||
|
||
if events:
|
||
logger.info(
|
||
"governance_dispatcher_cycle_start",
|
||
event_count=len(events),
|
||
)
|
||
dispatched = 0
|
||
skipped = 0
|
||
for event in events:
|
||
try:
|
||
result = await dispatch_governance_event(event)
|
||
if result is not None:
|
||
dispatched += 1
|
||
else:
|
||
skipped += 1
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"governance_dispatcher_event_error",
|
||
event_id=event.id,
|
||
event_type=event.event_type,
|
||
error=str(exc),
|
||
)
|
||
skipped += 1
|
||
|
||
logger.info(
|
||
"governance_dispatcher_cycle_done",
|
||
total=len(events),
|
||
dispatched=dispatched,
|
||
skipped=skipped,
|
||
)
|
||
else:
|
||
logger.debug("governance_dispatcher_no_events")
|
||
|
||
except asyncio.CancelledError:
|
||
logger.info("governance_dispatcher_loop_cancelled")
|
||
raise
|
||
except Exception as exc:
|
||
logger.warning("governance_dispatcher_loop_error", error=str(exc))
|
||
|
||
await asyncio.sleep(interval_seconds)
|