feat(governance): expose km draft dedupe plan

apps/api/src/api/v1/ai_governance.py

@@ -28,8 +28,10 @@ from src.models.governance import (
     GovernanceEventsResponse,
     GovernanceQueueResponse,
     GovernanceSummaryResponse,
+    KnowledgeReviewDraftDedupeResponse,
 )
 from src.services.governance_query_service import (
+    query_km_review_draft_dedupe,
     query_governance_events,
     query_governance_queue,
     query_governance_summary,
@@ -124,6 +126,27 @@ async def get_governance_queue(
     )
 
 
+# =============================================================================
+# GET /api/v1/ai/governance/km-review-drafts/dedupe
+# =============================================================================
+
+@router.get(
+    "/ai/governance/km-review-drafts/dedupe",
+    response_model=KnowledgeReviewDraftDedupeResponse,
+)
+async def get_km_review_draft_dedupe(
+    limit: Annotated[int, Query(ge=10, le=200)] = 100,
+) -> KnowledgeReviewDraftDedupeResponse:
+    """
+    查詢 Hermes KM healthcheck review drafts 的去重 read model。
+
+    這是 read-only owner review surface：只回傳 canonical / duplicate /
+    owner_action，不自動 archive、不自動 approve/publish KM。
+    """
+    logger.debug("km_review_draft_dedupe_request", limit=limit)
+    return await query_km_review_draft_dedupe(limit=limit)
+
+
 # =============================================================================
 # GET /api/v1/ai/governance/summary
 # =============================================================================

apps/api/src/models/governance.py

@@ -118,6 +118,34 @@ class GovernanceQueueResponse(BaseModel):
     )
 
 
+# =============================================================================
+# Endpoint 2B: KM review draft dedupe
+# =============================================================================
+
+class KnowledgeReviewDraftDedupeGroup(BaseModel):
+    governance_event_id: str
+    canonical_entry_id: str
+    canonical_title: str
+    canonical_updated_at: datetime | None = None
+    preferred_source: Literal["dispatch_context", "latest_review_draft"]
+    duplicate_entry_ids: list[str] = Field(default_factory=list)
+    duplicate_count: int
+    total_entries: int
+    suggested_action: str
+    owner_action: str
+    writes_on_read: bool = False
+    can_archive_without_owner_approval: bool = False
+
+
+class KnowledgeReviewDraftDedupeResponse(BaseModel):
+    schema_version: str = "km_review_draft_dedupe_v1"
+    total_review_drafts: int
+    event_group_total: int
+    duplicate_draft_total: int
+    groups: list[KnowledgeReviewDraftDedupeGroup]
+    generated_at: datetime
+
+
 # =============================================================================
 # Endpoint 3: summary
 # =============================================================================

apps/api/src/services/governance_query_service.py

@@ -23,11 +23,11 @@ from datetime import datetime, timedelta, timezone
 from typing import Any
 
 import structlog
-from sqlalchemy import bindparam, func, select, text
+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
+from src.db.models import AiGovernanceEvent, KnowledgeEntryRecord
 from src.models.governance import (
     DailyCount,
     DispatchItem,
@@ -35,8 +35,11 @@ from src.models.governance import (
     GovernanceEventsResponse,
     GovernanceQueueResponse,
     GovernanceSummaryResponse,
+    KnowledgeReviewDraftDedupeGroup,
+    KnowledgeReviewDraftDedupeResponse,
     map_severity,
 )
+from src.models.knowledge import EntryStatus, EntryType
 from src.utils.timezone import now_taipei
 
 logger = structlog.get_logger(__name__)
@@ -559,6 +562,181 @@ def _extract_worker_status(decision_ctx: dict) -> str | None:
     return val[:80] if isinstance(val, str) and val else 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)
+    preferred = await _load_preferred_km_draft_ids_by_event([
+        event_id
+        for row in rows
+        if (event_id := _extract_governance_event_id_from_tags(row.get("tags")))
+    ])
+    groups = _build_km_review_draft_dedupe_groups(rows, preferred)
+
+    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
+
+
+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,
+) -> list[KnowledgeReviewDraftDedupeGroup]:
+    """把 KM review drafts 依 governance_event tag 分組並產生 owner action。"""
+    preferred_draft_ids_by_event = preferred_draft_ids_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,
+        ))
+
+    return sorted(
+        groups,
+        key=lambda group: (
+            group.duplicate_count,
+            group.canonical_updated_at.isoformat() if group.canonical_updated_at else "",
+        ),
+        reverse=True,
+    )
+
+
 # =============================================================================
 # Endpoint 3: summary
 # =============================================================================