"""
Governance Query Service — /governance 頁面 DB 查詢邏輯
======================================================
封裝 3 個 governance endpoint 的資料庫查詢。
Router 層禁直接存取 DB（leWOOOgo 積木化鐵律）。

函式清單:
  query_governance_events(...)  → GovernanceEventsResponse
  query_governance_queue(...)   → GovernanceQueueResponse
  query_governance_summary(...) → GovernanceSummaryResponse

Graceful fallback 規則:
  queue endpoint — governance_remediation_dispatch 表可能尚未建立（Track D 進行中）。
  捕捉 sqlalchemy.exc.ProgrammingError（表不存在）後回傳 table_pending=True 的空列表，
  確保 API 在表建立前不拋 500。

2026-05-02 ogt + Claude Sonnet 4.6 Asia/Taipei
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Any

import structlog
from sqlalchemy import String, bindparam, func, or_, select, text
from sqlalchemy.exc import ProgrammingError

from src.db.base import get_db_context
from src.db.models import AiGovernanceEvent, KnowledgeEntryRecord
from src.models.governance import (
    DailyCount,
    DispatchItem,
    GovernanceEvent,
    GovernanceEventsResponse,
    GovernanceQueueResponse,
    GovernanceSummaryResponse,
    KnowledgeReviewDraftDedupeGroup,
    KnowledgeReviewDraftDedupeResponse,
    KnowledgeStaleCandidate,
    KnowledgeStaleCandidatesResponse,
    map_severity,
)
from src.models.knowledge import EntryStatus, EntryType
from src.utils.timezone import now_taipei

logger = structlog.get_logger(__name__)

# =============================================================================
# 常數
# =============================================================================

_TAIPEI = timezone(timedelta(hours=8))
_KM_STALE_DAYS = 7


# =============================================================================
# helpers
# =============================================================================

def _extract_impact(details: dict) -> str:
    """
    從 details 抽摘要字串，≤80 字。

    優先讀 details["impact"]（dict），取 status + 主要 metric 欄位。
    fallback 到 details 頂層常見欄位。
    """
    impact_block = details.get("impact")
    if isinstance(impact_block, dict):
        parts: list[str] = []
        if "status" in impact_block:
            parts.append(str(impact_block["status"]))
        # 主要 metric 欄位優先順序
        for key in ("metric", "value", "rate", "ratio", "score", "count"):
            if key in impact_block:
                parts.append(f"{key}={impact_block[key]}")
                break
        summary = " ".join(parts)
        return summary[:80] if summary else ""

    # fallback: 頂層常見欄位
    for key in ("message", "reason", "summary", "description"):
        val = details.get(key)
        if isinstance(val, str) and val:
            return val[:80]

    # 最後 fallback: 把 details 第一個 string value 截取
    for val in details.values():
        if isinstance(val, str) and val:
            return val[:80]

    return ""


def _extract_remediation(details: dict) -> str | None:
    """
    將治理事件 details.remediation 正規化為前端可顯示的短字串。

    Production 事件已出現 dict 形態（例如 {"items": [...]}），API response
    schema 則是字串。這裡做 read-side normalization，避免歷史資料讓
    /governance events 變成 500。
    """
    remediation = details.get("remediation")
    if remediation is None:
        return None
    if isinstance(remediation, str):
        return remediation[:160]
    if isinstance(remediation, dict):
        for key in ("summary", "message", "reason", "action"):
            value = remediation.get(key)
            if isinstance(value, str) and value:
                return value[:160]
        items = remediation.get("items")
        if isinstance(items, list):
            normalized = [str(item) for item in items if item is not None]
            if normalized:
                return "；".join(normalized[:3])[:160]
        return str(remediation)[:160]
    if isinstance(remediation, list):
        normalized = [str(item) for item in remediation if item is not None]
        return "；".join(normalized[:3])[:160] if normalized else None
    return str(remediation)[:160]


def _to_governance_event(
    row: AiGovernanceEvent,
    *,
    dispatch_ids: list[str] | None = None,
) -> GovernanceEvent:
    details = row.details if isinstance(row.details, dict) else {}
    return GovernanceEvent(
        id=row.id,
        event_type=row.event_type,
        severity=map_severity(row.event_type),
        triggered_at=row.triggered_at,
        resolved=row.resolved,
        resolved_at=row.resolved_at,
        impact=_extract_impact(details),
        details=details,
        remediation=_extract_remediation(details),
        dispatch_ids=_merge_dispatch_ids(dispatch_ids or [], details.get("dispatch_ids")),
    )


def _merge_dispatch_ids(
    db_dispatch_ids: list[str],
    legacy_dispatch_ids: Any,
) -> list[str]:
    """合併 DB dispatch trail 與 legacy payload ids，DB truth-first。"""
    merged: list[str] = []
    for raw in [*db_dispatch_ids, *(legacy_dispatch_ids if isinstance(legacy_dispatch_ids, list) else [])]:
        if raw is None:
            continue
        value = str(raw)
        if value and value not in merged:
            merged.append(value)
    return merged


async def _load_dispatch_ids_for_events(event_ids: list[str]) -> dict[str, list[str]]:
    """從 governance_remediation_dispatch 讀取事件對應 dispatch ids。

    events endpoint 必須能在 dispatch 表尚未建立的環境 graceful fallback，
    因此這裡捕捉 ProgrammingError 並回空 dict。
    """
    if not event_ids:
        return {}

    sql = text("""
        SELECT
            d.governance_event_id,
            d.id
        FROM governance_remediation_dispatch d
        WHERE d.governance_event_id IN :event_ids
        ORDER BY d.dispatched_at DESC
    """).bindparams(bindparam("event_ids", expanding=True))

    try:
        async with get_db_context() as db:
            result = await db.execute(sql, {"event_ids": event_ids})
            rows = result.fetchall()
    except ProgrammingError as exc:
        logger.warning(
            "governance_dispatch_ids_table_not_ready",
            error=str(exc),
        )
        return {}

    dispatch_ids_by_event: dict[str, list[str]] = {}
    for row in rows:
        dispatch_ids_by_event.setdefault(str(row.governance_event_id), []).append(str(row.id))
    return dispatch_ids_by_event


# =============================================================================
# Endpoint 1: events
# =============================================================================

async def query_governance_events(
    *,
    event_ids: list[str] | None = None,
    event_types: list[str] | None = None,
    from_dt: datetime | None = None,
    to_dt: datetime | None = None,
    status: str | None = None,         # "resolved" | "unresolved"
    severity: str | None = None,       # "critical" | "warning" | "info"
    page: int = 1,
    size: int = 20,
) -> GovernanceEventsResponse:
    """
    查詢 ai_governance_events 表，支援多維度過濾與分頁。

    severity 過濾在 Python 層完成（event_type 映射）；
    其他過濾在 SQL 層完成（效能優先）。
    """
    async with get_db_context() as db:
        stmt = select(AiGovernanceEvent)

        normalized_event_ids = [
            event_id.strip()
            for event_id in (event_ids or [])
            if isinstance(event_id, str) and event_id.strip()
        ]
        if normalized_event_ids:
            stmt = stmt.where(AiGovernanceEvent.id.in_(normalized_event_ids))

        if event_types:
            stmt = stmt.where(AiGovernanceEvent.event_type.in_(event_types))

        if from_dt is not None:
            stmt = stmt.where(AiGovernanceEvent.triggered_at >= from_dt)

        if to_dt is not None:
            stmt = stmt.where(AiGovernanceEvent.triggered_at <= to_dt)

        if status == "resolved":
            stmt = stmt.where(AiGovernanceEvent.resolved.is_(True))
        elif status == "unresolved":
            stmt = stmt.where(AiGovernanceEvent.resolved.is_(False))

        stmt = stmt.order_by(AiGovernanceEvent.triggered_at.desc())

        # 取全部結果，severity 在 Python 層過濾（避免 DB 不認識 mapping 邏輯）
        result = await db.execute(stmt)
        all_rows = result.scalars().all()

    event_rows_by_id = {str(r.id): r for r in all_rows}
    events = [_to_governance_event(r) for r in all_rows]

    # severity 過濾（Python 層）
    if severity:
        from src.models.governance import _CRITICAL_TYPES, _WARNING_TYPES

        if severity == "critical":
            events = [e for e in events if e.event_type in _CRITICAL_TYPES]
        elif severity == "warning":
            events = [e for e in events if e.event_type in _WARNING_TYPES]
        elif severity == "info":
            events = [
                e for e in events
                if e.event_type not in _CRITICAL_TYPES and e.event_type not in _WARNING_TYPES
            ]

    total = len(events)
    offset = (page - 1) * size
    page_items = events[offset: offset + size]
    dispatch_ids_by_event = await _load_dispatch_ids_for_events([e.id for e in page_items])
    if dispatch_ids_by_event:
        page_items = [
            _to_governance_event(
                event_rows_by_id[item.id],
                dispatch_ids=dispatch_ids_by_event.get(item.id, []),
            )
            for item in page_items
        ]

    return GovernanceEventsResponse(
        items=page_items,
        total=total,
        page=page,
        size=size,
    )


# =============================================================================
# Endpoint 2: queue
# =============================================================================

async def query_governance_queue(
    *,
    dispatch_status: str = "pending",
    event_types: list[str] | None = None,
    page: int = 1,
    size: int = 20,
) -> GovernanceQueueResponse:
    """
    查詢 governance_remediation_dispatch 表。

    Track D 進行中，表可能尚未建立。
    捕捉 ProgrammingError → 回傳 table_pending=True 的空 response。

    proposed_action 從 decision_context JSONB 抽取（Track D 完成後可改為真實 join）。
    """
    try:
        return await _query_dispatch_table(
            dispatch_status=dispatch_status,
            event_types=event_types,
            page=page,
            size=size,
        )
    except ProgrammingError as exc:
        logger.warning(
            "governance_dispatch_table_not_ready",
            error=str(exc),
        )
        return GovernanceQueueResponse(
            items=[],
            total=0,
            page=page,
            size=size,
            table_pending=True,
        )
    except ImportError as exc:
        logger.warning(
            "governance_dispatch_model_not_ready",
            error=str(exc),
        )
        return GovernanceQueueResponse(
            items=[],
            total=0,
            page=page,
            size=size,
            table_pending=True,
        )


async def _query_dispatch_table(
    *,
    dispatch_status: str,
    event_types: list[str] | None,
    page: int,
    size: int,
) -> GovernanceQueueResponse:
    """實際查詢 governance_remediation_dispatch 表（不含 graceful fallback）."""
    # 動態 import：Track D 完成前 ORM class 可能不存在
    # 使用 raw SQL 降低 ORM 模型缺失的耦合風險
    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 {status_filter}
          AND {event_type_filter}
        ORDER BY d.dispatched_at DESC
    """)

    count_sql = text(f"""
        SELECT count(*) AS cnt
        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, params)
        total = int(count_row.scalar_one_or_none() or 0)

        rows = await db.execute(sql, params)
        all_rows = rows.fetchall()

    offset = (page - 1) * size
    page_rows = all_rows[offset: offset + size]

    items: list[DispatchItem] = []
    for row in page_rows:
        decision_ctx: dict = (row.decision_context or {}) if hasattr(row, "decision_context") else {}
        items.append(_to_dispatch_item(row, decision_ctx))

    return GovernanceQueueResponse(
        items=items,
        total=total,
        page=page,
        size=size,
        table_pending=False,
    )


def _to_dispatch_item(row: Any, decision_ctx: dict) -> DispatchItem:
    """把 governance_remediation_dispatch SQL row 轉成 Work Items read model。"""
    proposed_action = _extract_proposed_action(decision_ctx)

    # playbook_trust: Track D 完成後改為 JOIN playbooks 表取 trust_score
    # 現階段從 decision_context 取 mock 值
    playbook_trust_raw = decision_ctx.get("playbook_trust")
    try:
        playbook_trust = float(playbook_trust_raw) if playbook_trust_raw is not None else None
    except (TypeError, ValueError):
        playbook_trust = None

    return DispatchItem(
        id=str(row.id),
        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),
        kb_draft_entry_id=_extract_kb_draft_entry_id(decision_ctx),
        worker_status=_extract_worker_status(decision_ctx),
        dry_run_plan_fingerprint=_extract_dry_run_plan_fingerprint(decision_ctx),
        archived_count=_extract_archived_count(decision_ctx),
        stale_ratio_snapshot=_extract_stale_ratio_snapshot(decision_ctx),
    )


def _extract_proposed_action(decision_ctx: dict) -> str:
    """
    從 decision_context JSONB 抽取 proposed_action，≤120 字。

    Track D 完成後此函式可改為從真實欄位讀取。
    """
    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


def _extract_kb_draft_entry_id(decision_ctx: dict) -> str | None:
    """Expose Hermes KM review draft id for Work Items owner review."""
    workflow = decision_ctx.get("workflow")
    if isinstance(workflow, dict):
        val = workflow.get("kb_draft_entry_id")
        if isinstance(val, str) and val:
            return val[:120]

    worker_result = decision_ctx.get("worker_result")
    if isinstance(worker_result, dict):
        val = worker_result.get("km_draft_entry_id")
        if isinstance(val, str) and val:
            return val[:120]

    return None


def _extract_worker_status(decision_ctx: dict) -> str | None:
    worker_result = decision_ctx.get("worker_result")
    if not isinstance(worker_result, dict):
        return None
    val = worker_result.get("status")
    return val[:80] if isinstance(val, str) and val else None


def _extract_dry_run_plan_fingerprint(decision_ctx: dict) -> str | None:
    for source in (
        decision_ctx,
        decision_ctx.get("workflow"),
    ):
        if not isinstance(source, dict):
            continue
        val = source.get("dry_run_plan_fingerprint")
        if isinstance(val, str) and val:
            return val[:80]
    return None


def _extract_archived_count(decision_ctx: dict) -> int | None:
    raw = decision_ctx.get("archived_count")
    if isinstance(raw, int):
        return max(raw, 0)
    archived = decision_ctx.get("archived_entry_ids")
    if isinstance(archived, list):
        return len(archived)
    workflow = decision_ctx.get("workflow")
    if isinstance(workflow, dict):
        archived = workflow.get("archived_entry_ids")
        if isinstance(archived, list):
            return len(archived)
    return None


def _extract_stale_ratio_snapshot(decision_ctx: dict) -> dict | None:
    for source in (
        decision_ctx,
        decision_ctx.get("workflow"),
    ):
        if not isinstance(source, dict):
            continue
        snapshot = source.get("stale_ratio_snapshot")
        if isinstance(snapshot, dict):
            return {
                key: snapshot.get(key)
                for key in (
                    "stale_count",
                    "total_count",
                    "stale_ratio",
                    "threshold",
                    "stale_days",
                )
                if key in snapshot
            }
    return None


# =============================================================================
# Endpoint 2B: KM review draft dedupe
# =============================================================================

async def query_km_review_draft_dedupe(
    *,
    limit: int = 100,
) -> KnowledgeReviewDraftDedupeResponse:
    """產生 Hermes KM healthcheck review drafts 的 read-only 去重計畫。"""
    rows = await _load_km_healthcheck_review_drafts(limit=limit)
    event_ids = [
        event_id
        for row in rows
        if (event_id := _extract_governance_event_id_from_tags(row.get("tags")))
    ]
    preferred = await _load_preferred_km_draft_ids_by_event(event_ids)
    archive_history = await _load_km_archive_history_by_event(event_ids)
    groups = _build_km_review_draft_dedupe_groups(rows, preferred, archive_history)

    return KnowledgeReviewDraftDedupeResponse(
        total_review_drafts=len(rows),
        event_group_total=len(groups),
        duplicate_draft_total=sum(group.duplicate_count for group in groups),
        groups=groups,
        generated_at=now_taipei(),
    )


async def _load_km_healthcheck_review_drafts(limit: int) -> list[dict[str, Any]]:
    """讀取 Hermes 產生、等待 owner review 的 KM healthcheck 草稿。"""
    async with get_db_context() as db:
        stmt = (
            select(KnowledgeEntryRecord)
            .where(
                KnowledgeEntryRecord.entry_type == EntryType.AUTO_RUNBOOK,
                KnowledgeEntryRecord.status == EntryStatus.REVIEW,
                or_(
                    KnowledgeEntryRecord.title.ilike("%KM healthcheck%"),
                    KnowledgeEntryRecord.content.ilike("%KM healthcheck%"),
                    KnowledgeEntryRecord.tags.cast(String).ilike("%workflow:kb_growth_healthcheck%"),
                ),
            )
            .order_by(KnowledgeEntryRecord.updated_at.desc())
            .limit(limit)
        )
        result = await db.execute(stmt)
        records = result.scalars().all()

    return [
        {
            "id": str(record.id),
            "title": str(record.title),
            "status": str(record.status.value if hasattr(record.status, "value") else record.status),
            "tags": list(record.tags or []),
            "created_by": record.created_by,
            "created_at": record.created_at,
            "updated_at": record.updated_at,
        }
        for record in records
    ]


async def _load_preferred_km_draft_ids_by_event(
    event_ids: list[str],
) -> dict[str, str]:
    """從 dispatch worker_result 讀取 event 對應的 canonical KM draft id。"""
    if not event_ids:
        return {}

    unique_event_ids = list(dict.fromkeys(event_ids))
    sql = text("""
        SELECT
            d.governance_event_id,
            d.decision_context
        FROM governance_remediation_dispatch d
        WHERE d.governance_event_id IN :event_ids
          AND d.executor_type = 'hermes_kb_growth_healthcheck'
          AND d.dispatch_status::text = 'succeeded'
        ORDER BY d.completed_at DESC NULLS LAST, d.dispatched_at DESC
    """).bindparams(bindparam("event_ids", expanding=True))

    try:
        async with get_db_context() as db:
            result = await db.execute(sql, {"event_ids": unique_event_ids})
            rows = result.fetchall()
    except ProgrammingError as exc:
        logger.warning(
            "km_review_dedupe_dispatch_table_not_ready",
            error=str(exc),
        )
        return {}

    preferred: dict[str, str] = {}
    for row in rows:
        event_id = str(row.governance_event_id)
        if event_id in preferred:
            continue
        decision_ctx = row.decision_context or {}
        if not isinstance(decision_ctx, dict):
            continue
        draft_id = _extract_kb_draft_entry_id(decision_ctx)
        if draft_id:
            preferred[event_id] = draft_id
    return preferred


async def _load_km_archive_history_by_event(
    event_ids: list[str],
) -> dict[str, list[DispatchItem]]:
    """讀取 KM duplicate archive / stale ratio recheck 的 terminal audit trail。"""
    if not event_ids:
        return {}

    unique_event_ids = list(dict.fromkeys(event_ids))
    sql = text("""
        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.governance_event_id IN :event_ids
          AND d.executor_type IN (
            'hermes_km_review_dedupe_owner_archive',
            'hermes_km_stale_ratio_recheck'
          )
        ORDER BY
            d.governance_event_id,
            d.completed_at DESC NULLS LAST,
            d.started_at DESC NULLS LAST,
            d.dispatched_at DESC
    """).bindparams(bindparam("event_ids", expanding=True))

    try:
        async with get_db_context() as db:
            result = await db.execute(sql, {"event_ids": unique_event_ids})
            rows = result.fetchall()
    except ProgrammingError as exc:
        logger.warning(
            "km_review_dedupe_archive_history_table_not_ready",
            error=str(exc),
        )
        return {}

    history: dict[str, list[DispatchItem]] = {}
    for row in rows:
        event_id = str(row.governance_event_id)
        bucket = history.setdefault(event_id, [])
        if len(bucket) >= 3:
            continue
        decision_ctx: dict = (row.decision_context or {}) if hasattr(row, "decision_context") else {}
        bucket.append(_to_dispatch_item(row, decision_ctx))

    return history


def _extract_governance_event_id_from_tags(tags: Any) -> str | None:
    if not isinstance(tags, list):
        return None
    for raw in tags:
        tag = str(raw)
        if tag.startswith("governance_event:"):
            event_id = tag.replace("governance_event:", "", 1).strip()
            return event_id or None
    return None


def _build_km_review_draft_dedupe_groups(
    rows: list[dict[str, Any]],
    preferred_draft_ids_by_event: dict[str, str] | None = None,
    archive_history_by_event: dict[str, list[DispatchItem]] | None = None,
) -> list[KnowledgeReviewDraftDedupeGroup]:
    """把 KM review drafts 依 governance_event tag 分組並產生 owner action。"""
    preferred_draft_ids_by_event = preferred_draft_ids_by_event or {}
    archive_history_by_event = archive_history_by_event or {}
    grouped: dict[str, list[dict[str, Any]]] = {}
    for row in rows:
        event_id = _extract_governance_event_id_from_tags(row.get("tags"))
        if not event_id:
            continue
        grouped.setdefault(event_id, []).append(row)

    groups: list[KnowledgeReviewDraftDedupeGroup] = []
    for event_id, entries in grouped.items():
        sorted_entries = sorted(
            entries,
            key=lambda item: str(item.get("updated_at") or item.get("created_at") or ""),
            reverse=True,
        )
        preferred_id = preferred_draft_ids_by_event.get(event_id)
        canonical = next(
            (entry for entry in sorted_entries if entry.get("id") == preferred_id),
            sorted_entries[0],
        )
        duplicate_ids = [
            str(entry["id"])
            for entry in sorted_entries
            if entry.get("id") != canonical.get("id")
        ]
        preferred_source = (
            "dispatch_context"
            if preferred_id and canonical.get("id") == preferred_id
            else "latest_review_draft"
        )
        groups.append(KnowledgeReviewDraftDedupeGroup(
            governance_event_id=event_id,
            canonical_entry_id=str(canonical["id"]),
            canonical_title=str(canonical.get("title") or ""),
            canonical_updated_at=canonical.get("updated_at"),
            preferred_source=preferred_source,
            duplicate_entry_ids=duplicate_ids,
            duplicate_count=len(duplicate_ids),
            total_entries=len(sorted_entries),
            suggested_action="owner_review_canonical_then_archive_duplicates",
            owner_action="review_canonical_and_archive_duplicate_drafts",
            writes_on_read=False,
            can_archive_without_owner_approval=False,
            archive_history=archive_history_by_event.get(event_id, []),
        ))

    return sorted(
        groups,
        key=lambda group: (
            group.duplicate_count,
            group.canonical_updated_at.isoformat() if group.canonical_updated_at else "",
        ),
        reverse=True,
    )


# =============================================================================
# Endpoint 2C: KM stale candidates
# =============================================================================

async def query_km_stale_candidates(
    *,
    project_id: str = "awoooi",
    limit: int = 20,
    threshold_days: int = _KM_STALE_DAYS,
) -> KnowledgeStaleCandidatesResponse:
    """
    產生 stale KM 的 read-only 優先處理清單。

    這個 endpoint 只讀 knowledge_entries，將已陳舊的 KM 依 incident /
    approval / playbook 反查鏈、Sentry / SigNoz 線索、view_count 與陳舊天數排序。
    它不自動改寫 KM，避免把錯誤知識固化到 production。
    """
    cutoff = now_taipei() - timedelta(days=threshold_days)

    async with get_db_context() as db:
        stmt = (
            select(KnowledgeEntryRecord)
            .where(
                KnowledgeEntryRecord.project_id == project_id,
                KnowledgeEntryRecord.status != EntryStatus.ARCHIVED,
                KnowledgeEntryRecord.updated_at < cutoff,
            )
            .order_by(KnowledgeEntryRecord.updated_at.asc())
        )
        result = await db.execute(stmt)
        records = result.scalars().all()

    generated_at = now_taipei()
    candidates = [
        _build_km_stale_candidate(
            record,
            now=generated_at,
            threshold_days=threshold_days,
        )
        for record in records
    ]
    candidates.sort(
        key=lambda item: (
            item.priority_score,
            item.stale_days,
            item.view_count,
            item.updated_at.isoformat() if item.updated_at else "",
        ),
        reverse=True,
    )
    limited = candidates[:limit]
    owner_review_state = await _load_km_stale_owner_review_state_by_entry(
        [candidate.entry_id for candidate in limited]
    )
    limited = [
        candidate.model_copy(update=owner_review_state.get(candidate.entry_id, {}))
        for candidate in limited
    ]

    return KnowledgeStaleCandidatesResponse(
        project_id=project_id,
        total_stale=len(candidates),
        returned=len(limited),
        threshold_days=threshold_days,
        items=limited,
        generated_at=generated_at,
    )


def _build_km_stale_candidate(
    record: KnowledgeEntryRecord,
    *,
    now: datetime,
    threshold_days: int = _KM_STALE_DAYS,
) -> KnowledgeStaleCandidate:
    """將一筆 KnowledgeEntryRecord 轉成 owner 可處理的 stale candidate。"""
    updated_at = record.updated_at
    stale_days = threshold_days
    if updated_at is not None:
        comparable_updated_at = updated_at
        if comparable_updated_at.tzinfo is None:
            comparable_updated_at = comparable_updated_at.replace(tzinfo=_TAIPEI)
        stale_days = max((now - comparable_updated_at).days, threshold_days)

    entry_type = _enum_value(record.entry_type)
    status = _enum_value(record.status)
    source = _enum_value(record.source)
    tags = [str(tag) for tag in (record.tags or []) if tag is not None]
    evidence_text = " ".join([
        record.title or "",
        record.content or "",
        " ".join(tags),
    ]).lower()

    reasons: list[str] = []
    correlation_sources: list[str] = []
    score = stale_days

    if record.related_incident_id:
        score += 80
        reasons.append("linked_incident")
        correlation_sources.append("incident")
    if record.related_approval_id:
        score += 70
        reasons.append("linked_approval")
        correlation_sources.append("approval")
    if record.related_playbook_id:
        score += 70
        reasons.append("linked_playbook")
        correlation_sources.append("playbook")
    if "sentry" in evidence_text:
        score += 30
        reasons.append("sentry_context")
        correlation_sources.append("sentry")
    if "signoz" in evidence_text:
        score += 30
        reasons.append("signoz_context")
        correlation_sources.append("signoz")
    if entry_type == EntryType.ANTI_PATTERN.value:
        score += 45
        reasons.append("anti_pattern_priority")
    if entry_type == EntryType.AUTO_RUNBOOK.value:
        score += 25
        reasons.append("auto_runbook_review_needed")
    if source == "ai_extracted":
        score += 20
        reasons.append("ai_extracted_needs_owner_check")
    if status == EntryStatus.REVIEW.value:
        score += 20
        reasons.append("already_waiting_review")

    view_count = int(record.view_count or 0)
    if view_count > 0:
        score += min(view_count, 50)
        reasons.append("viewed_by_operator")
    if stale_days >= 30:
        score += 25
        reasons.append("older_than_30_days")

    if not reasons:
        reasons.append("stale_by_age")

    priority_tier = _km_priority_tier(score, record, stale_days)
    recommended_action = _km_recommended_action(record, stale_days, view_count)

    return KnowledgeStaleCandidate(
        entry_id=str(record.id),
        project_id=str(record.project_id),
        title=str(record.title),
        entry_type=entry_type,
        category=str(record.category) if record.category else None,
        status=status,
        source=source,
        updated_at=updated_at,
        stale_days=stale_days,
        view_count=view_count,
        priority_score=score,
        priority_tier=priority_tier,
        recommended_action=recommended_action,
        reasons=list(dict.fromkeys(reasons)),
        correlation_sources=list(dict.fromkeys(correlation_sources)),
        related_incident_id=record.related_incident_id,
        related_playbook_id=record.related_playbook_id,
        related_approval_id=record.related_approval_id,
        tags=tags,
    )


async def _load_km_stale_owner_review_state_by_entry(
    entry_ids: list[str],
) -> dict[str, dict[str, str]]:
    """讀取 stale KM owner-review / completion 的最新狀態，供前端接續處理。"""
    if not entry_ids:
        return {}

    unique_entry_ids = list(dict.fromkeys(entry_ids))
    sql = text("""
        SELECT DISTINCT ON (entry_id)
            entry_id,
            d.id,
            d.dispatch_status,
            d.decision_context
        FROM (
            SELECT
                d.*,
                COALESCE(
                    d.decision_context -> 'workflow' ->> 'entry_id',
                    d.decision_context ->> 'entry_id'
                ) AS entry_id
            FROM governance_remediation_dispatch d
            WHERE d.executor_type IN (
                'hermes_km_stale_owner_review',
                'hermes_km_stale_owner_review_complete'
            )
        ) d
        WHERE entry_id IN :entry_ids
        ORDER BY entry_id, d.dispatched_at DESC
    """).bindparams(bindparam("entry_ids", expanding=True))

    try:
        async with get_db_context() as db:
            result = await db.execute(sql, {"entry_ids": unique_entry_ids})
            rows = result.fetchall()
    except ProgrammingError as exc:
        logger.warning(
            "km_stale_owner_review_state_table_not_ready",
            error=str(exc),
        )
        return {}

    state: dict[str, dict[str, str]] = {}
    for row in rows:
        decision_ctx: dict = row.decision_context if isinstance(row.decision_context, dict) else {}
        dispatch_status = str(row.dispatch_status)
        state[str(row.entry_id)] = {
            "owner_review_dispatch_id": str(row.id),
            "owner_review_status": dispatch_status,
            "owner_review_stage": _extract_workflow_stage(decision_ctx, dispatch_status) or "",
            "owner_review_next_action": _extract_next_action(decision_ctx) or "",
        }
    return state


def _km_priority_tier(
    score: int,
    record: KnowledgeEntryRecord,
    stale_days: int,
) -> str:
    """把排序分數壓成 owner 易懂的 P0/P1/P2 分層。"""
    if score >= 160:
        return "P0"
    if record.related_incident_id and (
        record.related_approval_id or record.related_playbook_id or stale_days >= 30
    ):
        return "P0"
    if score >= 90:
        return "P1"
    return "P2"


def _km_recommended_action(
    record: KnowledgeEntryRecord,
    stale_days: int,
    view_count: int,
) -> str:
    """決定 owner 下一步：刷新、審核、或封存/合併。"""
    status = _enum_value(record.status)
    if record.related_incident_id or record.related_playbook_id or record.related_approval_id:
        return "refresh_with_evidence"
    if status == EntryStatus.REVIEW.value or _enum_value(record.source) == "ai_extracted":
        return "owner_review"
    if stale_days >= 30 and view_count == 0:
        return "archive_or_supersede"
    return "owner_review"


def _enum_value(value: Any) -> str:
    """將 SQLAlchemy enum / plain string 正規化為 API 字串。"""
    if value is None:
        return ""
    if hasattr(value, "value"):
        return str(value.value)
    return str(value)


# =============================================================================
# Endpoint 3: summary
# =============================================================================

async def query_governance_summary(*, days: int = 30) -> GovernanceSummaryResponse:
    """
    過去 N 天 SLO 違反時序統計 + compliance_rate。

    compliance_rate = 1 - unresolved / total（total=0 時回 1.0）
    """
    since = now_taipei() - timedelta(days=days)

    async with get_db_context() as db:
        # 總數 & 未解決數
        count_stmt = select(
            func.count().label("total"),
            func.count().filter(AiGovernanceEvent.resolved.is_(False)).label("unresolved"),
        ).where(AiGovernanceEvent.triggered_at >= since)

        count_row = await db.execute(count_stmt)
        counts = count_row.one()
        total_events = int(counts.total)
        unresolved_count = int(counts.unresolved)

        # 每日計數（DATE_TRUNC 在 Postgres 端執行）
        daily_sql = text("""
            SELECT
                DATE_TRUNC('day', triggered_at AT TIME ZONE 'Asia/Taipei')::date AS day,
                event_type,
                count(*) AS cnt
            FROM ai_governance_events
            WHERE triggered_at >= :since
            GROUP BY day, event_type
            ORDER BY day ASC
        """)
        daily_result = await db.execute(daily_sql, {"since": since})
        daily_rows = daily_result.fetchall()

    # 彙整每日資料
    daily_map: dict[str, dict[str, int]] = {}
    for row in daily_rows:
        day_str = row.day.strftime("%Y-%m-%d") if hasattr(row.day, "strftime") else str(row.day)
        if day_str not in daily_map:
            daily_map[day_str] = {}
        daily_map[day_str][row.event_type] = int(row.cnt)

    daily_counts = [
        DailyCount(
            date=day_str,
            total=sum(by_type.values()),
            by_type=by_type,
        )
        for day_str, by_type in sorted(daily_map.items())
    ]

    if total_events == 0:
        compliance_rate = 1.0
    else:
        compliance_rate = round(1.0 - unresolved_count / total_events, 4)

    return GovernanceSummaryResponse(
        compliance_rate=compliance_rate,
        total_events=total_events,
        unresolved_count=unresolved_count,
        daily_counts=daily_counts,
    )
