feat(governance): surface km healthcheck dispatch
All checks were successful
Code Review / ai-code-review (push) Successful in 9s
Type Sync Check / check-type-sync (push) Successful in 38s
CD Pipeline / tests (push) Successful in 5m51s
CD Pipeline / build-and-deploy (push) Successful in 3m29s
CD Pipeline / post-deploy-checks (push) Successful in 1m20s

This commit is contained in:
Your Name
2026-05-19 21:07:55 +08:00
parent 3b50ff3cc3
commit c99be252d3
11 changed files with 722 additions and 20 deletions

View File

@@ -93,8 +93,9 @@ async def get_governance_events(
async def get_governance_queue(
dispatch_status: Annotated[
str,
Query(pattern="^(pending|dispatched|succeeded|failed)$"),
Query(pattern="^(all|pending|dispatched|executing|succeeded|failed|skipped|cancelled)$"),
] = "pending",
event_type: Annotated[list[str] | None, Query(alias="event_type")] = None,
page: Annotated[int, Query(ge=1)] = 1,
size: Annotated[int, Query(ge=10, le=100)] = 20,
) -> GovernanceQueueResponse:
@@ -104,17 +105,20 @@ async def get_governance_queue(
governance_remediation_dispatch 表由 Track D 建立,尚未完成時
本 endpoint 回傳 { table_pending: true, items: [], total: 0 },不拋 500。
- dispatch_status: pendingdefault/ dispatched / succeeded / failed
- dispatch_status: pendingdefault/ dispatched / executing / succeeded / failed / skipped / cancelled / all
- event_type: 多值過濾(可重複傳)
- page / size: 分頁
"""
logger.debug(
"governance_queue_request",
dispatch_status=dispatch_status,
event_type=event_type,
page=page,
size=size,
)
return await query_governance_queue(
dispatch_status=dispatch_status,
event_types=event_type,
page=page,
size=size,
)

View File

@@ -87,13 +87,22 @@ class DispatchItem(BaseModel):
governance_event_id: str
event_type: str
dispatch_status: str
executor_type: str | None = None
proposed_action: str = Field(description="≤120 字動作摘要")
playbook_id: str | None = None
playbook_trust: float | None = Field(default=None, ge=0.0, le=1.0)
created_at: datetime
dispatched_at: datetime | None = None
started_at: datetime | None = None
completed_at: datetime | None = None
operator_note: str | None = None
decision_path: str | None = None
workflow_stage: str | None = None
workflow_steps: list[str] = Field(default_factory=list)
next_action: str | None = None
lead_agent: str | None = None
support_agents: list[str] = Field(default_factory=list)
human_owner: str | None = None
class GovernanceQueueResponse(BaseModel):

View File

@@ -36,6 +36,7 @@ from src.repositories.governance_remediation_dispatch_repo import (
DispatchAlreadyActive,
create_dispatch,
get_active_for_event,
transition_status,
)
from src.services.decision_fusion_adapter import FusedDecision, get_decision_fusion_adapter
@@ -199,6 +200,8 @@ async def dispatch_governance_event(event: AiGovernanceEvent) -> str | None:
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")
@@ -209,10 +212,8 @@ async def dispatch_governance_event(event: AiGovernanceEvent) -> str | None:
# pending_approval → pending等人工審核
if decision.decision_path == "auto_dispatch":
executor_type = "playbook_executor"
initial_status_note = "auto_dispatch"
else: # pending_approval
executor_type = "manual"
initial_status_note = "pending_approval"
# Step 5: 建構 decision_context JSONB完整三維快照
decision_context = _build_decision_context(event, decision)
@@ -258,6 +259,38 @@ async def dispatch_governance_event(event: AiGovernanceEvent) -> str | None:
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 範圍內的治理事件。
@@ -296,6 +329,11 @@ def _build_decision_context(
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",
@@ -303,6 +341,9 @@ def _build_decision_context(
"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),
@@ -318,11 +359,86 @@ def _build_decision_context(
"affected_resources": [event.event_type],
"extra": {
"event_id": event.id,
"event_details_keys": list((event.details or {}).keys()),
"event_details_keys": list(details.keys()),
"ownership": ownership,
},
}
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 模式)
# =============================================================================

View File

@@ -20,9 +20,10 @@ Graceful fallback 規則:
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import Any
import structlog
from sqlalchemy import func, select, text
from sqlalchemy import bindparam, func, select, text
from sqlalchemy.exc import ProgrammingError
from src.db.base import get_db_context
@@ -209,6 +210,7 @@ async def query_governance_events(
async def query_governance_queue(
*,
dispatch_status: str = "pending",
event_types: list[str] | None = None,
page: int = 1,
size: int = 20,
) -> GovernanceQueueResponse:
@@ -223,6 +225,7 @@ async def query_governance_queue(
try:
return await _query_dispatch_table(
dispatch_status=dispatch_status,
event_types=event_types,
page=page,
size=size,
)
@@ -255,43 +258,67 @@ async def query_governance_queue(
async def _query_dispatch_table(
*,
dispatch_status: str,
event_types: list[str] | None,
page: int,
size: int,
) -> GovernanceQueueResponse:
"""實際查詢 governance_remediation_dispatch 表(不含 graceful fallback."""
# 動態 importTrack D 完成前 ORM class 可能不存在
# 使用 raw SQL 降低 ORM 模型缺失的耦合風險
sql = text("""
status_filter = (
"TRUE"
if dispatch_status == "all"
else "d.dispatch_status = CAST(:dispatch_status AS governance_dispatch_status)"
)
event_type_filter = (
"TRUE"
if not event_types
else "e.event_type::text IN :event_types"
)
params: dict[str, Any] = {}
if dispatch_status != "all":
params["dispatch_status"] = dispatch_status
if event_types:
params["event_types"] = event_types
sql = text(f"""
SELECT
d.id,
d.governance_event_id,
e.event_type,
d.dispatch_status,
d.executor_type,
d.decision_context,
d.playbook_id,
d.dispatched_at AS created_at,
d.dispatched_at,
d.started_at,
d.completed_at,
NULL::text AS operator_note
FROM governance_remediation_dispatch d
JOIN ai_governance_events e ON e.id = d.governance_event_id
WHERE d.dispatch_status = CAST(:dispatch_status AS governance_dispatch_status)
WHERE {status_filter}
AND {event_type_filter}
ORDER BY d.dispatched_at DESC
""")
count_sql = text("""
count_sql = text(f"""
SELECT count(*) AS cnt
FROM governance_remediation_dispatch
WHERE dispatch_status = CAST(:dispatch_status AS governance_dispatch_status)
FROM governance_remediation_dispatch d
JOIN ai_governance_events e ON e.id = d.governance_event_id
WHERE {status_filter}
AND {event_type_filter}
""")
if event_types:
sql = sql.bindparams(bindparam("event_types", expanding=True))
count_sql = count_sql.bindparams(bindparam("event_types", expanding=True))
async with get_db_context() as db:
count_row = await db.execute(count_sql, {"dispatch_status": dispatch_status})
count_row = await db.execute(count_sql, params)
total = int(count_row.scalar_one_or_none() or 0)
rows = await db.execute(
sql.bindparams(dispatch_status=dispatch_status),
)
rows = await db.execute(sql, params)
all_rows = rows.fetchall()
offset = (page - 1) * size
@@ -315,13 +342,22 @@ async def _query_dispatch_table(
governance_event_id=str(row.governance_event_id),
event_type=str(row.event_type),
dispatch_status=str(row.dispatch_status),
executor_type=str(row.executor_type) if row.executor_type else None,
proposed_action=proposed_action,
playbook_id=str(row.playbook_id) if row.playbook_id else None,
playbook_trust=playbook_trust,
created_at=row.created_at,
dispatched_at=row.dispatched_at,
started_at=row.started_at,
completed_at=row.completed_at,
operator_note=row.operator_note,
decision_path=_extract_decision_path(decision_ctx),
workflow_stage=_extract_workflow_stage(decision_ctx, str(row.dispatch_status)),
workflow_steps=_extract_workflow_steps(decision_ctx),
next_action=_extract_next_action(decision_ctx),
lead_agent=_extract_lead_agent(decision_ctx),
support_agents=_extract_support_agents(decision_ctx),
human_owner=_extract_human_owner(decision_ctx),
))
return GovernanceQueueResponse(
@@ -339,13 +375,99 @@ def _extract_proposed_action(decision_ctx: dict) -> str:
Track D 完成後此函式可改為從真實欄位讀取。
"""
for key in ("proposed_action", "action", "suggestion", "description", "summary"):
for key in (
"proposed_action",
"suggested_action",
"next_action",
"action",
"suggestion",
"description",
"summary",
):
val = decision_ctx.get(key)
if isinstance(val, str) and val:
return val[:120]
return "(待補充)"
def _extract_decision_path(decision_ctx: dict) -> str | None:
val = decision_ctx.get("decision_path")
return val[:80] if isinstance(val, str) and val else None
def _extract_next_action(decision_ctx: dict) -> str | None:
for key in ("next_action", "suggested_action", "proposed_action"):
val = decision_ctx.get(key)
if isinstance(val, str) and val:
return val[:120]
workflow = decision_ctx.get("workflow")
if isinstance(workflow, dict):
val = workflow.get("next_action")
if isinstance(val, str) and val:
return val[:120]
return None
def _extract_workflow_stage(decision_ctx: dict, dispatch_status: str) -> str | None:
workflow = decision_ctx.get("workflow")
if isinstance(workflow, dict):
stages = workflow.get("stage_by_dispatch_status")
if isinstance(stages, dict):
stage = stages.get(dispatch_status)
if isinstance(stage, str) and stage:
return stage[:120]
current = workflow.get("current_stage")
if isinstance(current, str) and current:
return current[:120]
return {
"pending": "queued_for_review",
"dispatched": "dispatched",
"executing": "executing",
"succeeded": "completed",
"failed": "failed",
"skipped": "skipped",
"cancelled": "cancelled",
}.get(dispatch_status)
def _extract_workflow_steps(decision_ctx: dict) -> list[str]:
workflow = decision_ctx.get("workflow")
if not isinstance(workflow, dict):
return []
steps = workflow.get("steps")
if not isinstance(steps, list):
return []
return [str(step)[:120] for step in steps if step is not None][:8]
def _extract_ownership(decision_ctx: dict) -> dict:
ownership = decision_ctx.get("ownership")
if isinstance(ownership, dict):
return ownership
extra = decision_ctx.get("extra")
if isinstance(extra, dict) and isinstance(extra.get("ownership"), dict):
return extra["ownership"]
return {}
def _extract_lead_agent(decision_ctx: dict) -> str | None:
val = _extract_ownership(decision_ctx).get("lead_agent")
return val[:80] if isinstance(val, str) and val else None
def _extract_support_agents(decision_ctx: dict) -> list[str]:
raw = _extract_ownership(decision_ctx).get("support_agents")
if not isinstance(raw, list):
return []
return [str(item)[:160] for item in raw if item is not None][:6]
def _extract_human_owner(decision_ctx: dict) -> str | None:
val = _extract_ownership(decision_ctx).get("human_owner")
return val[:120] if isinstance(val, str) and val else None
# =============================================================================
# Endpoint 3: summary
# =============================================================================