feat(governance): surface stale km owner review inbox

apps/api/src/services/governance_km_stale_review_service.py

@@ -35,6 +35,8 @@ from src.models.governance import (
     KnowledgeStaleOwnerReviewBatchQueueResponse,
     KnowledgeStaleOwnerReviewCompleteRequest,
     KnowledgeStaleOwnerReviewCompleteResponse,
+    KnowledgeStaleOwnerReviewInboxItem,
+    KnowledgeStaleOwnerReviewInboxResponse,
     KnowledgeStaleOwnerReviewRequest,
     KnowledgeStaleOwnerReviewResponse,
 )
@@ -53,6 +55,15 @@ _BATCH_EXECUTOR_TYPE = "hermes_km_stale_owner_review_batch"
 _COMPLETE_EXECUTOR_TYPE = "hermes_km_stale_owner_review_complete"
 _RECHECK_EXECUTOR_TYPE = "hermes_km_stale_ratio_recheck"
 _ACTIVE_DISPATCH_STATUSES = frozenset({"pending", "dispatched", "executing"})
+_OWNER_REVIEW_DISPATCH_STATUSES = frozenset({
+    "pending",
+    "dispatched",
+    "executing",
+    "succeeded",
+    "failed",
+    "skipped",
+    "cancelled",
+})
 
 
 class KmStaleOwnerReviewError(Exception):
@@ -64,6 +75,115 @@ class KmStaleOwnerReviewError(Exception):
         self.detail = detail
 
 
+async def query_km_stale_owner_review_inbox(
+    *,
+    project_id: str = "awoooi",
+    dispatch_status: str = "pending",
+    limit: int = 20,
+) -> KnowledgeStaleOwnerReviewInboxResponse:
+    """Read owner-review dispatches with KM priority context for the operator console."""
+    if dispatch_status != "all" and dispatch_status not in _OWNER_REVIEW_DISPATCH_STATUSES:
+        raise KmStaleOwnerReviewError(422, "unsupported stale KM owner-review dispatch_status")
+
+    generated_at = now_taipei()
+    status_filter = (
+        "TRUE"
+        if dispatch_status == "all"
+        else "d.dispatch_status::text = :dispatch_status"
+    )
+    sql = text(f"""
+        SELECT
+            d.id,
+            d.governance_event_id,
+            d.dispatch_status,
+            d.decision_context,
+            d.dispatched_at,
+            d.started_at,
+            d.completed_at,
+            COALESCE(
+                d.decision_context -> 'workflow' ->> 'entry_id',
+                d.decision_context ->> 'entry_id'
+            ) AS entry_id,
+            COALESCE(
+                d.decision_context -> 'workflow' ->> 'project_id',
+                d.decision_context ->> 'project_id',
+                d.decision_context -> 'candidate' ->> 'project_id'
+            ) AS project_id
+        FROM governance_remediation_dispatch d
+        WHERE d.executor_type = :executor_type
+          AND {status_filter}
+          AND COALESCE(
+                d.decision_context -> 'workflow' ->> 'project_id',
+                d.decision_context ->> 'project_id',
+                d.decision_context -> 'candidate' ->> 'project_id'
+              ) = :project_id
+        ORDER BY d.dispatched_at DESC
+    """)
+    params: dict[str, Any] = {
+        "executor_type": _EXECUTOR_TYPE,
+        "project_id": project_id,
+    }
+    if dispatch_status != "all":
+        params["dispatch_status"] = dispatch_status
+
+    async with get_db_context() as db:
+        result = await db.execute(sql, params)
+        rows = result.fetchall()
+        entry_ids = [
+            str(row.entry_id)
+            for row in rows
+            if isinstance(row.entry_id, str) and row.entry_id
+        ]
+        records_by_id: dict[str, KnowledgeEntryRecord] = {}
+        if entry_ids:
+            record_result = await db.execute(
+                select(KnowledgeEntryRecord).where(KnowledgeEntryRecord.id.in_(entry_ids))
+            )
+            records_by_id = {
+                str(record.id): record
+                for record in record_result.scalars().all()
+            }
+
+    items: list[KnowledgeStaleOwnerReviewInboxItem] = []
+    for row in rows:
+        entry_id = str(row.entry_id or "")
+        record = records_by_id.get(entry_id)
+        if record is None:
+            continue
+        candidate = _build_km_stale_candidate(
+            record,
+            now=generated_at,
+            threshold_days=KM_STALE_DAYS,
+        )
+        decision_context = row.decision_context if isinstance(row.decision_context, dict) else {}
+        items.append(
+            _build_owner_review_inbox_item(
+                row=row,
+                candidate=candidate,
+                decision_context=decision_context,
+            )
+        )
+
+    items.sort(
+        key=lambda item: (
+            _priority_rank(item.priority_tier),
+            item.priority_score,
+            item.stale_days,
+            item.queued_at.isoformat() if item.queued_at else "",
+        ),
+        reverse=True,
+    )
+    limited = items[:limit]
+    return KnowledgeStaleOwnerReviewInboxResponse(
+        project_id=project_id,
+        dispatch_status=dispatch_status,
+        total=len(items),
+        returned=len(limited),
+        items=limited,
+        generated_at=generated_at,
+    )
+
+
 async def queue_km_stale_owner_review(
     *,
     entry_id: str,
@@ -648,6 +768,107 @@ def _count_batch_items(
     return sum(1 for item in items if item.status == status)
 
 
+def _build_owner_review_inbox_item(
+    *,
+    row: Any,
+    candidate: KnowledgeStaleCandidate,
+    decision_context: dict[str, Any],
+) -> KnowledgeStaleOwnerReviewInboxItem:
+    dispatch_status = str(row.dispatch_status)
+    workflow = decision_context.get("workflow") if isinstance(decision_context.get("workflow"), dict) else {}
+    batch = decision_context.get("batch") if isinstance(decision_context.get("batch"), dict) else {}
+    return KnowledgeStaleOwnerReviewInboxItem(
+        dispatch_id=str(row.id),
+        governance_event_id=str(row.governance_event_id),
+        entry_id=candidate.entry_id,
+        project_id=candidate.project_id,
+        title=candidate.title,
+        dispatch_status=dispatch_status,
+        workflow_stage=_extract_owner_review_workflow_stage(decision_context, dispatch_status),
+        next_action=_extract_owner_review_next_action(decision_context),
+        owner=_extract_owner_review_owner(decision_context),
+        owner_note=_extract_owner_review_owner_note(decision_context),
+        batch_governance_event_id=_first_non_empty_string(
+            workflow.get("batch_governance_event_id"),
+            batch.get("batch_governance_event_id"),
+        ),
+        batch_dispatch_id=_first_non_empty_string(
+            workflow.get("batch_dispatch_id"),
+            batch.get("batch_dispatch_id"),
+        ),
+        priority_tier=candidate.priority_tier,
+        priority_score=candidate.priority_score,
+        recommended_action=candidate.recommended_action,
+        stale_days=candidate.stale_days,
+        view_count=candidate.view_count,
+        correlation_sources=candidate.correlation_sources,
+        reasons=candidate.reasons,
+        related_incident_id=candidate.related_incident_id,
+        related_playbook_id=candidate.related_playbook_id,
+        related_approval_id=candidate.related_approval_id,
+        dry_run_plan_fingerprint=_extract_plan_fingerprint(decision_context),
+        queued_at=row.dispatched_at,
+        started_at=row.started_at,
+        completed_at=row.completed_at,
+    )
+
+
+def _priority_rank(priority_tier: str) -> int:
+    return {"P0": 3, "P1": 2, "P2": 1}.get(priority_tier, 0)
+
+
+def _extract_owner_review_workflow_stage(
+    context: dict[str, Any],
+    dispatch_status: str,
+) -> str:
+    workflow = context.get("workflow")
+    if isinstance(workflow, dict):
+        stages = workflow.get("stage_by_dispatch_status")
+        if isinstance(stages, dict):
+            value = stages.get(dispatch_status)
+            if isinstance(value, str) and value:
+                return value
+        value = workflow.get("current_stage")
+        if isinstance(value, str) and value:
+            return value
+    return {
+        "pending": "waiting_owner_review",
+        "dispatched": "waiting_owner_review",
+        "executing": "owner_review_in_progress",
+        "succeeded": "km_candidate_reviewed",
+        "failed": "needs_manual_km_triage",
+        "skipped": "waiting_owner_review",
+        "cancelled": "cancelled",
+    }.get(dispatch_status, "unknown")
+
+
+def _extract_owner_review_next_action(context: dict[str, Any]) -> str | None:
+    workflow = context.get("workflow")
+    if isinstance(workflow, dict):
+        value = workflow.get("next_action")
+        if isinstance(value, str) and value:
+            return value
+    value = context.get("next_action")
+    return value if isinstance(value, str) and value else None
+
+
+def _extract_owner_review_owner(context: dict[str, Any]) -> str | None:
+    value = context.get("owner")
+    return value if isinstance(value, str) and value else None
+
+
+def _extract_owner_review_owner_note(context: dict[str, Any]) -> str | None:
+    value = context.get("owner_note")
+    return value if isinstance(value, str) and value else None
+
+
+def _first_non_empty_string(*values: Any) -> str | None:
+    for value in values:
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
 async def complete_km_stale_owner_review(
     *,
     entry_id: str,