feat(governance): preview stale km completion batches

apps/api/src/services/governance_km_stale_review_service.py

@@ -37,6 +37,8 @@ from src.models.governance import (
     KnowledgeStaleOwnerReviewBurnDownResponse,
     KnowledgeStaleOwnerReviewCompleteRequest,
     KnowledgeStaleOwnerReviewCompleteResponse,
+    KnowledgeStaleOwnerReviewCompletionBatchPreviewRequest,
+    KnowledgeStaleOwnerReviewCompletionBatchPreviewResponse,
     KnowledgeStaleOwnerReviewCompletionQueueItem,
     KnowledgeStaleOwnerReviewCompletionQueueResponse,
     KnowledgeStaleOwnerReviewInboxItem,
@@ -120,9 +122,15 @@ async def query_km_stale_owner_review_completion_queue(
     *,
     project_id: str = "awoooi",
     status_bucket: Literal["all", "ready", "blocked", "completed", "failed", "pending"] = "all",
+    priority_tiers: list[str] | None = None,
+    recommended_completion_outcome: str = "all",
+    batch_governance_event_id: str | None = None,
+    can_preview: bool | None = None,
     limit: int = 20,
 ) -> KnowledgeStaleOwnerReviewCompletionQueueResponse:
     """Read owner-review completion readiness without writing KM content."""
+    normalized_priority_tiers = _normalize_priority_tiers(priority_tiers)
+    normalized_outcome = _normalize_completion_outcome(recommended_completion_outcome)
     generated_at = now_taipei()
     inbox_items = await _load_owner_review_inbox_items(
         project_id=project_id,
@@ -148,10 +156,21 @@ async def query_km_stale_owner_review_completion_queue(
         item for item in queue_items
         if _completion_queue_bucket_matches(item, status_bucket=status_bucket)
     ]
+    filtered_items = _filter_completion_queue_items(
+        filtered_items,
+        priority_tiers=normalized_priority_tiers,
+        recommended_completion_outcome=normalized_outcome,
+        batch_governance_event_id=batch_governance_event_id,
+        can_preview=can_preview,
+    )
     limited = filtered_items[:limit]
     return KnowledgeStaleOwnerReviewCompletionQueueResponse(
         project_id=project_id,
         status_bucket=status_bucket,
+        priority_tiers=normalized_priority_tiers,
+        recommended_completion_outcome=normalized_outcome,
+        batch_governance_event_id=batch_governance_event_id,
+        can_preview=can_preview,
         total=len(filtered_items),
         returned=len(limited),
         pending_count=sum(
@@ -167,6 +186,47 @@ async def query_km_stale_owner_review_completion_queue(
     )
 
 
+async def preview_km_stale_owner_review_completion_batch(
+    *,
+    request: KnowledgeStaleOwnerReviewCompletionBatchPreviewRequest,
+) -> KnowledgeStaleOwnerReviewCompletionBatchPreviewResponse:
+    """Preview a bounded completion batch without writing KM or governance audit rows."""
+    generated_at = now_taipei()
+    queue = await query_km_stale_owner_review_completion_queue(
+        project_id=request.project_id,
+        status_bucket=request.status_bucket,
+        priority_tiers=list(request.priority_tiers),
+        recommended_completion_outcome=request.recommended_completion_outcome,
+        batch_governance_event_id=request.batch_governance_event_id,
+        can_preview=None,
+        limit=request.limit,
+    )
+    items = queue.items
+    previewable_items = [item for item in items if item.can_preview]
+    fingerprint = _build_completion_batch_preview_fingerprint(
+        request=request,
+        items=items,
+    )
+    return KnowledgeStaleOwnerReviewCompletionBatchPreviewResponse(
+        project_id=request.project_id,
+        owner=request.owner,
+        owner_note=request.owner_note,
+        status_bucket=request.status_bucket,
+        priority_tiers=list(dict.fromkeys(request.priority_tiers)),
+        recommended_completion_outcome=request.recommended_completion_outcome,
+        batch_governance_event_id=request.batch_governance_event_id,
+        requested_limit=request.limit,
+        candidate_count=len(items),
+        previewable_count=len(previewable_items),
+        blocked_count=sum(1 for item in items if item.readiness == "blocked"),
+        completed_count=sum(1 for item in items if item.readiness == "completed"),
+        failed_count=sum(1 for item in items if item.readiness == "failed"),
+        dry_run_plan_fingerprint=fingerprint,
+        items=items,
+        generated_at=generated_at,
+    )
+
+
 async def _load_owner_review_inbox_items(
     *,
     project_id: str,
@@ -973,6 +1033,107 @@ def _completion_queue_bucket_matches(
     return item.readiness == status_bucket
 
 
+def _normalize_priority_tiers(priority_tiers: list[str] | None) -> list[str]:
+    if not priority_tiers:
+        return []
+    allowed = {"P0", "P1", "P2"}
+    normalized = list(dict.fromkeys(str(tier).upper() for tier in priority_tiers))
+    unsupported = [tier for tier in normalized if tier not in allowed]
+    if unsupported:
+        raise KmStaleOwnerReviewError(
+            422,
+            "unsupported stale KM owner-review completion priority_tier",
+        )
+    return normalized
+
+
+def _normalize_completion_outcome(outcome: str) -> Literal[
+    "all",
+    "refresh_with_evidence",
+    "archive",
+    "supersede",
+]:
+    if outcome == "all":
+        return "all"
+    if outcome == "refresh_with_evidence":
+        return "refresh_with_evidence"
+    if outcome == "archive":
+        return "archive"
+    if outcome == "supersede":
+        return "supersede"
+    raise KmStaleOwnerReviewError(
+        422,
+        "unsupported stale KM owner-review completion outcome",
+    )
+
+
+def _filter_completion_queue_items(
+    items: list[KnowledgeStaleOwnerReviewCompletionQueueItem],
+    *,
+    priority_tiers: list[str],
+    recommended_completion_outcome: str,
+    batch_governance_event_id: str | None,
+    can_preview: bool | None,
+) -> list[KnowledgeStaleOwnerReviewCompletionQueueItem]:
+    filtered = items
+    if priority_tiers:
+        wanted = set(priority_tiers)
+        filtered = [item for item in filtered if item.priority_tier in wanted]
+    if recommended_completion_outcome != "all":
+        filtered = [
+            item for item in filtered
+            if item.recommended_completion_outcome == recommended_completion_outcome
+        ]
+    if batch_governance_event_id:
+        filtered = [
+            item for item in filtered
+            if item.batch_governance_event_id == batch_governance_event_id
+        ]
+    if can_preview is not None:
+        filtered = [item for item in filtered if item.can_preview is can_preview]
+    return filtered
+
+
+def _build_completion_batch_preview_fingerprint(
+    *,
+    request: KnowledgeStaleOwnerReviewCompletionBatchPreviewRequest,
+    items: list[KnowledgeStaleOwnerReviewCompletionQueueItem],
+) -> str:
+    payload = {
+        "schema_version": "km_stale_owner_review_completion_batch_preview_v1",
+        "project_id": request.project_id,
+        "status_bucket": request.status_bucket,
+        "priority_tiers": list(dict.fromkeys(request.priority_tiers)),
+        "recommended_completion_outcome": request.recommended_completion_outcome,
+        "batch_governance_event_id": request.batch_governance_event_id,
+        "limit": request.limit,
+        "owner": request.owner,
+        "owner_note_sha256": (
+            hashlib.sha256(request.owner_note.encode("utf-8")).hexdigest()
+            if request.owner_note
+            else None
+        ),
+        "items": [
+            {
+                "dispatch_id": item.dispatch_id,
+                "entry_id": item.entry_id,
+                "readiness": item.readiness,
+                "recommended_completion_outcome": item.recommended_completion_outcome,
+                "can_preview": item.can_preview,
+                "workflow_stage": item.workflow_stage,
+            }
+            for item in items
+        ],
+    }
+    encoded = json.dumps(
+        payload,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return "sha256:" + hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
 def _completion_outcome_for_owner_review_item(
     item: KnowledgeStaleOwnerReviewInboxItem,
 ) -> Literal["refresh_with_evidence", "archive", "supersede"]: