feat(governance): surface stale km completion queue

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

@@ -37,6 +37,7 @@ from src.models.governance import (
     KnowledgeStaleOwnerReviewBurnDownResponse,
     KnowledgeStaleOwnerReviewCompleteRequest,
     KnowledgeStaleOwnerReviewCompleteResponse,
+    KnowledgeStaleOwnerReviewCompletionQueueResponse,
     KnowledgeStaleOwnerReviewInboxResponse,
     KnowledgeStaleOwnerReviewRequest,
     KnowledgeStaleOwnerReviewResponse,
@@ -50,6 +51,7 @@ from src.services.governance_km_stale_review_service import (
     batch_queue_km_stale_owner_reviews,
     complete_km_stale_owner_review,
     query_km_stale_owner_review_burndown,
+    query_km_stale_owner_review_completion_queue,
     query_km_stale_owner_review_inbox,
     queue_km_stale_owner_review,
 )
@@ -304,6 +306,41 @@ async def get_km_stale_owner_review_burndown(
     )
 
 
+# =============================================================================
+# GET /api/v1/ai/governance/km-stale-owner-review-completion-queue
+# =============================================================================
+
+@router.get(
+    "/ai/governance/km-stale-owner-review-completion-queue",
+    response_model=KnowledgeStaleOwnerReviewCompletionQueueResponse,
+)
+async def get_km_stale_owner_review_completion_queue(
+    project_id: Annotated[str, Query(min_length=1, max_length=64)] = "awoooi",
+    status_bucket: Annotated[
+        str,
+        Query(pattern="^(all|ready|blocked|completed|failed|pending)$"),
+    ] = "all",
+    limit: Annotated[int, Query(ge=1, le=100)] = 20,
+) -> KnowledgeStaleOwnerReviewCompletionQueueResponse:
+    """
+    查詢 stale KM owner-review completion 分流。
+
+    這是 read-only queue：把 active / completed / failed dispatch 拆成
+    ready、blocked、completed、failed，讓前端呈現下一步卡點；打開頁面不寫 KM。
+    """
+    logger.debug(
+        "km_stale_owner_review_completion_queue_request",
+        project_id=project_id,
+        status_bucket=status_bucket,
+        limit=limit,
+    )
+    return await query_km_stale_owner_review_completion_queue(
+        project_id=project_id,
+        status_bucket=status_bucket,
+        limit=limit,
+    )
+
+
 # =============================================================================
 # POST /api/v1/ai/governance/km-stale-candidates/batch-queue-review
 # =============================================================================

apps/api/src/models/governance.py

@@ -417,6 +417,68 @@ class KnowledgeStaleOwnerReviewBurnDownResponse(BaseModel):
     generated_at: datetime
 
 
+class KnowledgeStaleOwnerReviewCompletionQueueItem(BaseModel):
+    dispatch_id: str
+    governance_event_id: str
+    entry_id: str
+    project_id: str
+    title: str
+    dispatch_status: str
+    workflow_stage: str
+    readiness: Literal["ready", "blocked", "completed", "failed"]
+    recommended_completion_outcome: Literal[
+        "refresh_with_evidence",
+        "archive",
+        "supersede",
+    ]
+    next_action: str
+    blockers: list[str] = Field(default_factory=list)
+    required_owner_fields: list[str] = Field(default_factory=list)
+    can_preview: bool
+    can_confirm_after_preview: bool
+    writes_km_on_confirm: bool
+    owner: str | None = None
+    owner_note: str | None = None
+    batch_governance_event_id: str | None = None
+    batch_dispatch_id: str | None = None
+    priority_tier: Literal["P0", "P1", "P2"]
+    priority_score: int
+    recommended_action: Literal[
+        "refresh_with_evidence",
+        "owner_review",
+        "archive_or_supersede",
+    ]
+    stale_days: int
+    view_count: int
+    correlation_sources: list[str] = Field(default_factory=list)
+    reasons: list[str] = Field(default_factory=list)
+    related_incident_id: str | None = None
+    related_playbook_id: str | None = None
+    related_approval_id: str | None = None
+    dry_run_plan_fingerprint: str | None = None
+    queued_at: datetime | None = None
+    started_at: datetime | None = None
+    completed_at: datetime | None = None
+
+
+class KnowledgeStaleOwnerReviewCompletionQueueResponse(BaseModel):
+    schema_version: str = "km_stale_owner_review_completion_queue_v1"
+    project_id: str
+    status_bucket: Literal["all", "ready", "blocked", "completed", "failed", "pending"]
+    total: int
+    returned: int
+    pending_count: int
+    ready_count: int
+    blocked_count: int
+    completed_count: int
+    failed_count: int
+    writes_on_read: bool = False
+    manual_review_required: bool = True
+    batch_writes_allowed: bool = False
+    items: list[KnowledgeStaleOwnerReviewCompletionQueueItem] = Field(default_factory=list)
+    generated_at: datetime
+
+
 class KnowledgeStaleOwnerReviewCompleteRequest(BaseModel):
     dispatch_id: str | None = Field(
         default=None,

apps/api/src/services/governance_km_stale_review_service.py

@@ -37,6 +37,8 @@ from src.models.governance import (
     KnowledgeStaleOwnerReviewBurnDownResponse,
     KnowledgeStaleOwnerReviewCompleteRequest,
     KnowledgeStaleOwnerReviewCompleteResponse,
+    KnowledgeStaleOwnerReviewCompletionQueueItem,
+    KnowledgeStaleOwnerReviewCompletionQueueResponse,
     KnowledgeStaleOwnerReviewInboxItem,
     KnowledgeStaleOwnerReviewInboxResponse,
     KnowledgeStaleOwnerReviewRequest,
@@ -57,6 +59,7 @@ _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"})
+_TERMINAL_FAILED_DISPATCH_STATUSES = frozenset({"failed", "cancelled"})
 _OWNER_REVIEW_DISPATCH_STATUSES = frozenset({
     "pending",
     "dispatched",
@@ -88,6 +91,88 @@ async def query_km_stale_owner_review_inbox(
         raise KmStaleOwnerReviewError(422, "unsupported stale KM owner-review dispatch_status")
 
     generated_at = now_taipei()
+    items = await _load_owner_review_inbox_items(
+        project_id=project_id,
+        dispatch_status=dispatch_status,
+        generated_at=generated_at,
+    )
+    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 query_km_stale_owner_review_completion_queue(
+    *,
+    project_id: str = "awoooi",
+    status_bucket: Literal["all", "ready", "blocked", "completed", "failed", "pending"] = "all",
+    limit: int = 20,
+) -> KnowledgeStaleOwnerReviewCompletionQueueResponse:
+    """Read owner-review completion readiness without writing KM content."""
+    generated_at = now_taipei()
+    inbox_items = await _load_owner_review_inbox_items(
+        project_id=project_id,
+        dispatch_status="all",
+        generated_at=generated_at,
+    )
+    queue_items = [
+        _build_completion_queue_item(item)
+        for item in inbox_items
+    ]
+    queue_items.sort(
+        key=lambda item: (
+            item.readiness == "ready",
+            _priority_rank(item.priority_tier),
+            item.priority_score,
+            item.stale_days,
+            item.queued_at.isoformat() if item.queued_at else "",
+        ),
+        reverse=True,
+    )
+
+    filtered_items = [
+        item for item in queue_items
+        if _completion_queue_bucket_matches(item, status_bucket=status_bucket)
+    ]
+    limited = filtered_items[:limit]
+    return KnowledgeStaleOwnerReviewCompletionQueueResponse(
+        project_id=project_id,
+        status_bucket=status_bucket,
+        total=len(filtered_items),
+        returned=len(limited),
+        pending_count=sum(
+            1 for item in queue_items
+            if item.dispatch_status in _ACTIVE_DISPATCH_STATUSES
+        ),
+        ready_count=sum(1 for item in queue_items if item.readiness == "ready"),
+        blocked_count=sum(1 for item in queue_items if item.readiness == "blocked"),
+        completed_count=sum(1 for item in queue_items if item.readiness == "completed"),
+        failed_count=sum(1 for item in queue_items if item.readiness == "failed"),
+        items=limited,
+        generated_at=generated_at,
+    )
+
+
+async def _load_owner_review_inbox_items(
+    *,
+    project_id: str,
+    dispatch_status: str,
+    generated_at: Any,
+) -> list[KnowledgeStaleOwnerReviewInboxItem]:
     status_filter = (
         "TRUE"
         if dispatch_status == "all"
@@ -165,25 +250,7 @@ async def query_km_stale_owner_review_inbox(
                 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,
-    )
+    return items
 
 
 async def query_km_stale_owner_review_burndown(
@@ -850,6 +917,128 @@ def _build_owner_review_inbox_item(
     )
 
 
+def _build_completion_queue_item(
+    item: KnowledgeStaleOwnerReviewInboxItem,
+) -> KnowledgeStaleOwnerReviewCompletionQueueItem:
+    outcome = _completion_outcome_for_owner_review_item(item)
+    blockers = _completion_queue_blockers(item, outcome=outcome)
+    readiness = _completion_queue_readiness(item, blockers=blockers)
+    can_preview = readiness == "ready"
+    return KnowledgeStaleOwnerReviewCompletionQueueItem(
+        dispatch_id=item.dispatch_id,
+        governance_event_id=item.governance_event_id,
+        entry_id=item.entry_id,
+        project_id=item.project_id,
+        title=item.title,
+        dispatch_status=item.dispatch_status,
+        workflow_stage=item.workflow_stage,
+        readiness=readiness,
+        recommended_completion_outcome=outcome,
+        next_action=_completion_queue_next_action(readiness),
+        blockers=blockers,
+        required_owner_fields=_completion_required_owner_fields(outcome),
+        can_preview=can_preview,
+        can_confirm_after_preview=can_preview,
+        writes_km_on_confirm=can_preview,
+        owner=item.owner,
+        owner_note=item.owner_note,
+        batch_governance_event_id=item.batch_governance_event_id,
+        batch_dispatch_id=item.batch_dispatch_id,
+        priority_tier=item.priority_tier,
+        priority_score=item.priority_score,
+        recommended_action=item.recommended_action,
+        stale_days=item.stale_days,
+        view_count=item.view_count,
+        correlation_sources=item.correlation_sources,
+        reasons=item.reasons,
+        related_incident_id=item.related_incident_id,
+        related_playbook_id=item.related_playbook_id,
+        related_approval_id=item.related_approval_id,
+        dry_run_plan_fingerprint=item.dry_run_plan_fingerprint,
+        queued_at=item.queued_at,
+        started_at=item.started_at,
+        completed_at=item.completed_at,
+    )
+
+
+def _completion_queue_bucket_matches(
+    item: KnowledgeStaleOwnerReviewCompletionQueueItem,
+    *,
+    status_bucket: str,
+) -> bool:
+    if status_bucket == "all":
+        return True
+    if status_bucket == "pending":
+        return item.dispatch_status in _ACTIVE_DISPATCH_STATUSES
+    return item.readiness == status_bucket
+
+
+def _completion_outcome_for_owner_review_item(
+    item: KnowledgeStaleOwnerReviewInboxItem,
+) -> Literal["refresh_with_evidence", "archive", "supersede"]:
+    if item.recommended_action == "archive_or_supersede":
+        return "archive"
+    return "refresh_with_evidence"
+
+
+def _completion_required_owner_fields(
+    outcome: Literal["refresh_with_evidence", "archive", "supersede"],
+) -> list[str]:
+    if outcome == "refresh_with_evidence":
+        return ["owner_note_or_updated_content"]
+    if outcome == "supersede":
+        return ["superseded_by_entry_id", "owner_note"]
+    return ["owner_note"]
+
+
+def _completion_queue_blockers(
+    item: KnowledgeStaleOwnerReviewInboxItem,
+    *,
+    outcome: Literal["refresh_with_evidence", "archive", "supersede"],
+) -> list[str]:
+    blockers: list[str] = []
+    if not item.dispatch_id:
+        blockers.append("missing_dispatch_id")
+    if not item.entry_id:
+        blockers.append("missing_entry_id")
+    if item.dispatch_status == "skipped":
+        blockers.append("owner_review_skipped")
+    if item.dispatch_status not in (
+        _ACTIVE_DISPATCH_STATUSES
+        | _TERMINAL_FAILED_DISPATCH_STATUSES
+        | frozenset({"succeeded", "skipped"})
+    ):
+        blockers.append("unsupported_dispatch_status")
+    if outcome == "supersede":
+        blockers.append("superseded_by_entry_id_required")
+    return blockers
+
+
+def _completion_queue_readiness(
+    item: KnowledgeStaleOwnerReviewInboxItem,
+    *,
+    blockers: list[str],
+) -> Literal["ready", "blocked", "completed", "failed"]:
+    if item.dispatch_status == "succeeded":
+        return "completed"
+    if item.dispatch_status in _TERMINAL_FAILED_DISPATCH_STATUSES:
+        return "failed"
+    if item.dispatch_status in _ACTIVE_DISPATCH_STATUSES and not blockers:
+        return "ready"
+    return "blocked"
+
+
+def _completion_queue_next_action(
+    readiness: Literal["ready", "blocked", "completed", "failed"],
+) -> str:
+    return {
+        "ready": "preview_stale_km_review_completion",
+        "blocked": "inspect_owner_review_blocker",
+        "completed": "view_stale_km_completion_audit",
+        "failed": "manual_km_owner_review_triage",
+    }[readiness]
+
+
 async def _load_owner_review_burndown_counts(
     db: AsyncSession,
     *,