feat(governance): batch queue stale km reviews

apps/api/src/services/governance_km_stale_review_service.py

@@ -29,6 +29,10 @@ from src.db.models import (
 )
 from src.models.governance import (
     KnowledgeReviewDraftStaleRatioSnapshot,
+    KnowledgeStaleCandidate,
+    KnowledgeStaleOwnerReviewBatchItem,
+    KnowledgeStaleOwnerReviewBatchQueueRequest,
+    KnowledgeStaleOwnerReviewBatchQueueResponse,
     KnowledgeStaleOwnerReviewCompleteRequest,
     KnowledgeStaleOwnerReviewCompleteResponse,
     KnowledgeStaleOwnerReviewRequest,
@@ -36,14 +40,19 @@ from src.models.governance import (
 )
 from src.models.knowledge import EntryStatus
 from src.services.governance_agent import KM_STALE_DAYS, KM_STALE_RATIO
-from src.services.governance_query_service import _build_km_stale_candidate
+from src.services.governance_query_service import (
+    _build_km_stale_candidate,
+    query_km_stale_candidates,
+)
 from src.utils.timezone import now_taipei
 
 logger = structlog.get_logger(__name__)
 
 _EXECUTOR_TYPE = "hermes_km_stale_owner_review"
+_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"})
 
 
 class KmStaleOwnerReviewError(Exception):
@@ -158,6 +167,487 @@ async def queue_km_stale_owner_review(
     )
 
 
+async def batch_queue_km_stale_owner_reviews(
+    *,
+    request: KnowledgeStaleOwnerReviewBatchQueueRequest,
+) -> KnowledgeStaleOwnerReviewBatchQueueResponse:
+    """Queue a bounded P0/P1 stale-KM batch for owner review without KM writes."""
+    selected = await _select_batch_stale_candidates(request)
+    items = _build_batch_plan_items(selected, dry_run=request.dry_run)
+    snapshot = await _load_current_km_stale_ratio_snapshot()
+    fingerprint = _build_batch_queue_plan_fingerprint(
+        request=request,
+        items=items,
+        snapshot=snapshot,
+    )
+    queueable = [item for item in items if item.status == "would_queue"]
+
+    if request.dry_run:
+        return _build_batch_queue_response(
+            request=request,
+            status="dry_run",
+            items=items,
+            stale_ratio_snapshot=snapshot,
+            dry_run_plan_fingerprint=fingerprint,
+            writes_governance_audit=False,
+        )
+
+    if not request.dry_run_plan_fingerprint:
+        raise KmStaleOwnerReviewError(
+            403,
+            "dry_run_plan_fingerprint from a dry-run preview is required before queueing a stale KM batch",
+        )
+    if request.dry_run_plan_fingerprint != fingerprint:
+        raise KmStaleOwnerReviewError(
+            409,
+            "dry_run_plan_fingerprint does not match the latest stale KM batch queue plan",
+        )
+    if not queueable:
+        return _build_batch_queue_response(
+            request=request,
+            status="noop_already_queued",
+            items=items,
+            stale_ratio_snapshot=snapshot,
+            dry_run_plan_fingerprint=fingerprint,
+            writes_governance_audit=False,
+        )
+
+    write_result = await _write_batch_owner_review_dispatches(
+        request=request,
+        candidates=selected,
+        plan_items=items,
+        stale_ratio_snapshot=snapshot,
+        plan_fingerprint=fingerprint,
+    )
+    return _build_batch_queue_response(
+        request=request,
+        status="queued",
+        items=write_result["items"],
+        stale_ratio_snapshot=snapshot,
+        dry_run_plan_fingerprint=fingerprint,
+        writes_governance_audit=True,
+        batch_governance_event_id=write_result["batch_governance_event_id"],
+        batch_dispatch_id=write_result["batch_dispatch_id"],
+    )
+
+
+async def _select_batch_stale_candidates(
+    request: KnowledgeStaleOwnerReviewBatchQueueRequest,
+) -> list[KnowledgeStaleCandidate]:
+    """Load a bounded priority batch from the read model instead of duplicating scoring logic."""
+    fetch_limit = min(100, max(request.limit * 4, request.limit))
+    response = await query_km_stale_candidates(
+        project_id=request.project_id,
+        limit=fetch_limit,
+    )
+    wanted_tiers = set(request.priority_tiers)
+    selected: list[KnowledgeStaleCandidate] = []
+    for candidate in response.items:
+        if candidate.priority_tier not in wanted_tiers:
+            continue
+        selected.append(candidate)
+        if len(selected) >= request.limit:
+            break
+    return selected
+
+
+def _build_batch_plan_items(
+    candidates: list[KnowledgeStaleCandidate],
+    *,
+    dry_run: bool,
+) -> list[KnowledgeStaleOwnerReviewBatchItem]:
+    items: list[KnowledgeStaleOwnerReviewBatchItem] = []
+    for candidate in candidates:
+        owner_status = str(candidate.owner_review_status or "")
+        if owner_status in _ACTIVE_DISPATCH_STATUSES:
+            status: Literal["would_queue", "queued", "already_queued", "skipped"] = "already_queued"
+            reason = "active_owner_review_exists"
+            workflow_stage = candidate.owner_review_stage or "waiting_owner_review"
+        elif owner_status == "succeeded":
+            status = "skipped"
+            reason = "already_reviewed_or_completed"
+            workflow_stage = candidate.owner_review_stage or "km_candidate_reviewed"
+        else:
+            status = "would_queue" if dry_run else "would_queue"
+            reason = None
+            workflow_stage = "waiting_owner_review"
+        items.append(
+            KnowledgeStaleOwnerReviewBatchItem(
+                entry_id=candidate.entry_id,
+                title=candidate.title,
+                priority_tier=candidate.priority_tier,
+                recommended_action=candidate.recommended_action,
+                status=status,
+                reason=reason,
+                dispatch_id=candidate.owner_review_dispatch_id,
+                workflow_stage=workflow_stage,
+            )
+        )
+    return items
+
+
+def _build_batch_queue_plan_fingerprint(
+    *,
+    request: KnowledgeStaleOwnerReviewBatchQueueRequest,
+    items: list[KnowledgeStaleOwnerReviewBatchItem],
+    snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
+) -> str:
+    payload = {
+        "schema_version": "km_stale_owner_review_batch_plan_v1",
+        "project_id": request.project_id,
+        "priority_tiers": list(dict.fromkeys(request.priority_tiers)),
+        "limit": request.limit,
+        "owner": request.owner,
+        "owner_note_sha256": (
+            hashlib.sha256(request.owner_note.encode("utf-8")).hexdigest()
+            if request.owner_note
+            else None
+        ),
+        "stale_ratio_snapshot": snapshot.model_dump(),
+        "items": [
+            {
+                "entry_id": item.entry_id,
+                "priority_tier": item.priority_tier,
+                "recommended_action": item.recommended_action,
+                "status": item.status,
+                "dispatch_id": item.dispatch_id,
+            }
+            for item in items
+        ],
+    }
+    encoded = json.dumps(
+        payload,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return "sha256:" + hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+async def _write_batch_owner_review_dispatches(
+    *,
+    request: KnowledgeStaleOwnerReviewBatchQueueRequest,
+    candidates: list[KnowledgeStaleCandidate],
+    plan_items: list[KnowledgeStaleOwnerReviewBatchItem],
+    stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
+    plan_fingerprint: str,
+) -> dict[str, Any]:
+    now = taipei_now()
+    batch_event_id = generate_uuid()
+    batch_dispatch_id = generate_uuid()
+    candidate_by_id = {candidate.entry_id: candidate for candidate in candidates}
+    queued_ids = {item.entry_id for item in plan_items if item.status == "would_queue"}
+    queued_item_ids: dict[str, tuple[str, str]] = {}
+
+    async with get_db_context() as db:
+        batch_event = AiGovernanceEvent(
+            id=batch_event_id,
+            event_type="knowledge_degradation",
+            triggered_at=now,
+            details=_build_batch_owner_review_event_details(
+                request=request,
+                items=plan_items,
+                stale_ratio_snapshot=stale_ratio_snapshot,
+            ),
+            resolved=True,
+            resolved_at=now,
+        )
+        batch_dispatch = GovernanceRemediationDispatch(
+            id=batch_dispatch_id,
+            governance_event_id=batch_event_id,
+            event_type="knowledge_degradation",
+            dispatch_status="succeeded",
+            decision_context=_build_batch_owner_review_decision_context(
+                batch_governance_event_id=batch_event_id,
+                batch_dispatch_id=batch_dispatch_id,
+                request=request,
+                items=plan_items,
+                stale_ratio_snapshot=stale_ratio_snapshot,
+                plan_fingerprint=plan_fingerprint,
+            ),
+            executor_type=_BATCH_EXECUTOR_TYPE,
+            attempt_count=0,
+            max_attempts=1,
+            dispatched_at=now,
+            started_at=now,
+            completed_at=now,
+            created_by=request.owner[:100],
+        )
+        db.add(batch_event)
+        db.add(batch_dispatch)
+
+        for item in plan_items:
+            if item.entry_id not in queued_ids:
+                continue
+            candidate = candidate_by_id[item.entry_id]
+            event_id = generate_uuid()
+            dispatch_id = generate_uuid()
+            candidate_payload = candidate.model_dump(mode="json")
+            event_details = _build_stale_owner_review_event_details(
+                entry_id=item.entry_id,
+                candidate=candidate_payload,
+                owner=request.owner,
+                owner_note=request.owner_note,
+            )
+            event_details["batch"] = {
+                "batch_governance_event_id": batch_event_id,
+                "batch_dispatch_id": batch_dispatch_id,
+                "dry_run_plan_fingerprint": plan_fingerprint,
+            }
+            decision_context = _build_stale_owner_review_decision_context(
+                governance_event_id=event_id,
+                entry_id=item.entry_id,
+                candidate=candidate_payload,
+                owner=request.owner,
+                owner_note=request.owner_note,
+            )
+            decision_context = _attach_batch_to_owner_review_context(
+                decision_context,
+                batch_governance_event_id=batch_event_id,
+                batch_dispatch_id=batch_dispatch_id,
+                plan_fingerprint=plan_fingerprint,
+            )
+            db.add(
+                AiGovernanceEvent(
+                    id=event_id,
+                    event_type="knowledge_degradation",
+                    triggered_at=now,
+                    details=event_details,
+                    resolved=False,
+                )
+            )
+            db.add(
+                GovernanceRemediationDispatch(
+                    id=dispatch_id,
+                    governance_event_id=event_id,
+                    event_type="knowledge_degradation",
+                    dispatch_status="pending",
+                    playbook_id=None,
+                    incident_id=None,
+                    approval_id=None,
+                    decision_context=decision_context,
+                    executor_type=_EXECUTOR_TYPE,
+                    attempt_count=0,
+                    max_attempts=1,
+                    dispatched_at=now,
+                    created_by=request.owner[:100],
+                )
+            )
+            queued_item_ids[item.entry_id] = (event_id, dispatch_id)
+        await db.flush()
+
+    updated_items: list[KnowledgeStaleOwnerReviewBatchItem] = []
+    for item in plan_items:
+        ids = queued_item_ids.get(item.entry_id)
+        if ids is None:
+            updated_items.append(item)
+            continue
+        event_id, dispatch_id = ids
+        updated_items.append(
+            item.model_copy(update={
+                "status": "queued",
+                "governance_event_id": event_id,
+                "dispatch_id": dispatch_id,
+                "workflow_stage": "waiting_owner_review",
+            })
+        )
+
+    logger.info(
+        "km_stale_owner_review_batch_queued",
+        project_id=request.project_id,
+        batch_governance_event_id=batch_event_id,
+        batch_dispatch_id=batch_dispatch_id,
+        queued_count=len(queued_item_ids),
+        candidate_count=len(plan_items),
+    )
+    return {
+        "batch_governance_event_id": batch_event_id,
+        "batch_dispatch_id": batch_dispatch_id,
+        "items": updated_items,
+    }
+
+
+def _attach_batch_to_owner_review_context(
+    context: dict[str, Any],
+    *,
+    batch_governance_event_id: str,
+    batch_dispatch_id: str,
+    plan_fingerprint: str,
+) -> dict[str, Any]:
+    merged = dict(context)
+    workflow = dict(merged.get("workflow") if isinstance(merged.get("workflow"), dict) else {})
+    workflow.update({
+        "batch_governance_event_id": batch_governance_event_id,
+        "batch_dispatch_id": batch_dispatch_id,
+        "batch_plan_fingerprint": plan_fingerprint,
+    })
+    merged["workflow"] = workflow
+    merged["batch"] = {
+        "batch_governance_event_id": batch_governance_event_id,
+        "batch_dispatch_id": batch_dispatch_id,
+        "dry_run_plan_fingerprint": plan_fingerprint,
+    }
+    return merged
+
+
+def _build_batch_owner_review_event_details(
+    *,
+    request: KnowledgeStaleOwnerReviewBatchQueueRequest,
+    items: list[KnowledgeStaleOwnerReviewBatchItem],
+    stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
+) -> dict[str, Any]:
+    return {
+        "schema_version": "km_stale_owner_review_batch_event_v1",
+        "trigger_source": "stale_km_priority_batch_queue",
+        "next_action": "owner_review_stale_km_batch",
+        "impact": {
+            "status": "batch_owner_review_queued",
+            "project_id": request.project_id,
+            "priority_tiers": list(dict.fromkeys(request.priority_tiers)),
+            "requested_limit": request.limit,
+            "candidate_count": len(items),
+            "queued_count": _count_batch_items(items, "would_queue"),
+            "already_queued_count": _count_batch_items(items, "already_queued"),
+            "skipped_count": _count_batch_items(items, "skipped"),
+            "stale_ratio": stale_ratio_snapshot.stale_ratio,
+            "threshold": stale_ratio_snapshot.threshold,
+        },
+        "remediation": {
+            "next_action": "owner_review_stale_km_batch",
+            "items": [
+                "review_p0_p1_stale_km_candidates",
+                "complete_each_candidate_after_owner_approval",
+                "run_stale_ratio_recheck_after_writeback",
+            ],
+        },
+        "ownership": _stale_owner_review_ownership(),
+        "owner": request.owner,
+        "owner_note": request.owner_note,
+        "stale_ratio_snapshot": stale_ratio_snapshot.model_dump(),
+        "items": [item.model_dump() for item in items],
+    }
+
+
+def _build_batch_owner_review_decision_context(
+    *,
+    batch_governance_event_id: str,
+    batch_dispatch_id: str,
+    request: KnowledgeStaleOwnerReviewBatchQueueRequest,
+    items: list[KnowledgeStaleOwnerReviewBatchItem],
+    stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
+    plan_fingerprint: str,
+) -> dict[str, Any]:
+    queued_count = _count_batch_items(items, "would_queue")
+    return {
+        "schema_version": "km_stale_owner_review_batch_dispatch_v1",
+        "version": "v1",
+        "trigger_source": "stale_km_priority_batch_queue",
+        "triggered_metric": "knowledge_degradation",
+        "metric_value": stale_ratio_snapshot.stale_ratio,
+        "threshold": stale_ratio_snapshot.threshold,
+        "suggested_action": "owner_review_stale_km_batch",
+        "next_action": "owner_review_stale_km_batch",
+        "decision_path": "batch_owner_review_queued",
+        "ownership": _stale_owner_review_ownership(),
+        "workflow": {
+            "work_item_id": (
+                "governance:knowledge_degradation:"
+                f"{batch_governance_event_id}:km_stale_owner_review_batch"
+            ),
+            "work_kind": "km_stale_owner_review_batch",
+            "current_stage": "batch_owner_review_queued",
+            "project_id": request.project_id,
+            "priority_tiers": list(dict.fromkeys(request.priority_tiers)),
+            "requested_limit": request.limit,
+            "batch_governance_event_id": batch_governance_event_id,
+            "batch_dispatch_id": batch_dispatch_id,
+            "steps": [
+                "detected",
+                "prioritized_stale_candidate",
+                "batch_owner_review_queued",
+                "waiting_owner_review",
+                "owner_updates_or_archives_km",
+                "stale_ratio_recheck",
+            ],
+            "stage_by_dispatch_status": {
+                "pending": "batch_owner_review_queued",
+                "dispatched": "batch_owner_review_queued",
+                "executing": "batch_owner_review_queued",
+                "succeeded": "batch_owner_review_queued",
+                "failed": "needs_manual_km_triage",
+                "skipped": "needs_manual_km_triage",
+                "cancelled": "cancelled",
+            },
+            "next_action": "owner_review_stale_km_batch",
+            "needs_human_review": True,
+            "writes_km_without_approval": False,
+            "writes_km": False,
+            "writes_governance_audit": True,
+            "dry_run_plan_fingerprint": plan_fingerprint,
+            "stale_ratio_snapshot": stale_ratio_snapshot.model_dump(),
+        },
+        "worker_result": {
+            "status": "batch_owner_review_queued",
+            "candidate_count": len(items),
+            "queued_count": queued_count,
+            "already_queued_count": _count_batch_items(items, "already_queued"),
+            "skipped_count": _count_batch_items(items, "skipped"),
+            "writes_km": False,
+        },
+        "owner": request.owner,
+        "owner_note": request.owner_note,
+        "items": [item.model_dump() for item in items],
+    }
+
+
+def _build_batch_queue_response(
+    *,
+    request: KnowledgeStaleOwnerReviewBatchQueueRequest,
+    status: Literal["dry_run", "queued", "noop_already_queued"],
+    items: list[KnowledgeStaleOwnerReviewBatchItem],
+    stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
+    dry_run_plan_fingerprint: str,
+    writes_governance_audit: bool,
+    batch_governance_event_id: str | None = None,
+    batch_dispatch_id: str | None = None,
+) -> KnowledgeStaleOwnerReviewBatchQueueResponse:
+    queued_status = "would_queue" if status == "dry_run" else "queued"
+    workflow_stage = {
+        "dry_run": "batch_owner_review_previewed",
+        "queued": "batch_owner_review_queued",
+        "noop_already_queued": "batch_noop_already_queued",
+    }[status]
+    return KnowledgeStaleOwnerReviewBatchQueueResponse(
+        project_id=request.project_id,
+        status=status,
+        owner=request.owner,
+        owner_note=request.owner_note,
+        dry_run=request.dry_run,
+        priority_tiers=list(dict.fromkeys(request.priority_tiers)),
+        requested_limit=request.limit,
+        candidate_count=len(items),
+        queued_count=_count_batch_items(items, queued_status),
+        already_queued_count=_count_batch_items(items, "already_queued"),
+        skipped_count=_count_batch_items(items, "skipped"),
+        batch_governance_event_id=batch_governance_event_id,
+        batch_dispatch_id=batch_dispatch_id,
+        workflow_stage=workflow_stage,
+        writes_km=False,
+        writes_governance_audit=writes_governance_audit,
+        stale_ratio_snapshot=stale_ratio_snapshot,
+        dry_run_plan_fingerprint=dry_run_plan_fingerprint,
+        items=items,
+        generated_at=now_taipei(),
+    )
+
+
+def _count_batch_items(
+    items: list[KnowledgeStaleOwnerReviewBatchItem],
+    status: str,
+) -> int:
+    return sum(1 for item in items if item.status == status)
+
+
 async def complete_km_stale_owner_review(
     *,
     entry_id: str,