feat(governance): surface stale km burndown

apps/api/src/services/governance_km_stale_review_service.py

@@ -33,6 +33,8 @@ from src.models.governance import (
     KnowledgeStaleOwnerReviewBatchItem,
     KnowledgeStaleOwnerReviewBatchQueueRequest,
     KnowledgeStaleOwnerReviewBatchQueueResponse,
+    KnowledgeStaleOwnerReviewBurnDownItem,
+    KnowledgeStaleOwnerReviewBurnDownResponse,
     KnowledgeStaleOwnerReviewCompleteRequest,
     KnowledgeStaleOwnerReviewCompleteResponse,
     KnowledgeStaleOwnerReviewInboxItem,
@@ -184,6 +186,41 @@ async def query_km_stale_owner_review_inbox(
     )
 
 
+async def query_km_stale_owner_review_burndown(
+    *,
+    project_id: str = "awoooi",
+    limit: int = 20,
+) -> KnowledgeStaleOwnerReviewBurnDownResponse:
+    """Read stale KM owner-review completion progress and current burn-down status."""
+    generated_at = now_taipei()
+    async with get_db_context() as db:
+        current_snapshot = await _compute_km_stale_ratio_snapshot(db)
+        counts = await _load_owner_review_burndown_counts(db, project_id=project_id)
+        completion_rows = await _load_owner_review_completion_rows(
+            db,
+            project_id=project_id,
+            limit=limit,
+        )
+
+    items = _build_owner_review_burndown_items(completion_rows, project_id=project_id)
+    latest = items[0] if items else None
+    return KnowledgeStaleOwnerReviewBurnDownResponse(
+        project_id=project_id,
+        burn_down_status=_stale_burndown_status(current_snapshot),
+        current_snapshot=current_snapshot,
+        entries_to_threshold=_entries_to_stale_threshold(current_snapshot),
+        pending_owner_reviews=counts["pending_owner_reviews"],
+        completed_owner_reviews=counts["completed_owner_reviews"],
+        completion_audit_total=counts["completion_audit_total"],
+        stale_ratio_recheck_total=counts["stale_ratio_recheck_total"],
+        latest_stale_count_delta=latest.stale_count_delta if latest else None,
+        latest_stale_ratio_delta=latest.stale_ratio_delta if latest else None,
+        returned=len(items),
+        items=items,
+        generated_at=generated_at,
+    )
+
+
 async def queue_km_stale_owner_review(
     *,
     entry_id: str,
@@ -813,6 +850,188 @@ def _build_owner_review_inbox_item(
     )
 
 
+async def _load_owner_review_burndown_counts(
+    db: AsyncSession,
+    *,
+    project_id: str,
+) -> dict[str, int]:
+    sql = text("""
+        SELECT
+            COUNT(*) FILTER (
+                WHERE d.executor_type = :owner_executor
+                  AND d.dispatch_status::text IN ('pending', 'dispatched', 'executing')
+            ) AS pending_owner_reviews,
+            COUNT(*) FILTER (
+                WHERE d.executor_type = :owner_executor
+                  AND d.dispatch_status::text = 'succeeded'
+            ) AS completed_owner_reviews,
+            COUNT(*) FILTER (
+                WHERE d.executor_type = :complete_executor
+                  AND d.dispatch_status::text = 'succeeded'
+            ) AS completion_audit_total,
+            COUNT(*) FILTER (
+                WHERE d.executor_type = :complete_executor
+                  AND d.dispatch_status::text = 'succeeded'
+                  AND d.decision_context -> 'stale_ratio_recheck' ->> 'dispatch_id' IS NOT NULL
+            ) AS stale_ratio_recheck_total
+        FROM governance_remediation_dispatch d
+        WHERE d.event_type = 'knowledge_degradation'
+          AND d.executor_type IN (:owner_executor, :complete_executor)
+          AND COALESCE(
+                d.decision_context -> 'workflow' ->> 'project_id',
+                d.decision_context ->> 'project_id',
+                d.decision_context -> 'candidate' ->> 'project_id'
+              ) = :project_id
+    """)
+    result = await db.execute(
+        sql,
+        {
+            "owner_executor": _EXECUTOR_TYPE,
+            "complete_executor": _COMPLETE_EXECUTOR_TYPE,
+            "project_id": project_id,
+        },
+    )
+    row = result.first()
+    return {
+        "pending_owner_reviews": int(row.pending_owner_reviews or 0) if row else 0,
+        "completed_owner_reviews": int(row.completed_owner_reviews or 0) if row else 0,
+        "completion_audit_total": int(row.completion_audit_total or 0) if row else 0,
+        "stale_ratio_recheck_total": int(row.stale_ratio_recheck_total or 0) if row else 0,
+    }
+
+
+async def _load_owner_review_completion_rows(
+    db: AsyncSession,
+    *,
+    project_id: str,
+    limit: int,
+) -> list[Any]:
+    sql = text("""
+        SELECT
+            d.id,
+            d.governance_event_id,
+            d.dispatch_status,
+            d.decision_context,
+            d.dispatched_at,
+            d.started_at,
+            d.completed_at
+        FROM governance_remediation_dispatch d
+        WHERE d.executor_type = :complete_executor
+          AND d.dispatch_status::text = 'succeeded'
+          AND COALESCE(
+                d.decision_context -> 'workflow' ->> 'project_id',
+                d.decision_context ->> 'project_id'
+              ) = :project_id
+        ORDER BY d.completed_at DESC NULLS LAST, d.dispatched_at DESC
+        LIMIT :limit
+    """)
+    result = await db.execute(
+        sql,
+        {
+            "complete_executor": _COMPLETE_EXECUTOR_TYPE,
+            "project_id": project_id,
+            "limit": limit,
+        },
+    )
+    return list(result.fetchall())
+
+
+def _build_owner_review_burndown_items(
+    rows: list[Any],
+    *,
+    project_id: str,
+) -> list[KnowledgeStaleOwnerReviewBurnDownItem]:
+    chronological: list[KnowledgeStaleOwnerReviewBurnDownItem] = []
+    previous_snapshot: KnowledgeReviewDraftStaleRatioSnapshot | None = None
+    for row in reversed(rows):
+        context = row.decision_context if isinstance(row.decision_context, dict) else {}
+        snapshot = _snapshot_from_context(context)
+        stale_count_delta: int | None = None
+        stale_ratio_delta: float | None = None
+        if snapshot is not None and previous_snapshot is not None:
+            stale_count_delta = snapshot.stale_count - previous_snapshot.stale_count
+            stale_ratio_delta = round(snapshot.stale_ratio - previous_snapshot.stale_ratio, 3)
+        if snapshot is not None:
+            previous_snapshot = snapshot
+
+        chronological.append(
+            _build_owner_review_burndown_item(
+                row=row,
+                context=context,
+                project_id=project_id,
+                stale_count_delta=stale_count_delta,
+                stale_ratio_delta=stale_ratio_delta,
+            )
+        )
+    return list(reversed(chronological))
+
+
+def _build_owner_review_burndown_item(
+    *,
+    row: Any,
+    context: dict[str, Any],
+    project_id: str,
+    stale_count_delta: int | None,
+    stale_ratio_delta: float | None,
+) -> KnowledgeStaleOwnerReviewBurnDownItem:
+    workflow = context.get("workflow") if isinstance(context.get("workflow"), dict) else {}
+    worker_result = context.get("worker_result") if isinstance(context.get("worker_result"), dict) else {}
+    snapshot = _snapshot_from_context(context)
+    dispatch_status = str(row.dispatch_status)
+    return KnowledgeStaleOwnerReviewBurnDownItem(
+        completion_dispatch_id=str(row.id),
+        governance_event_id=str(row.governance_event_id),
+        source_dispatch_id=_first_non_empty_string(
+            workflow.get("source_dispatch_id"),
+            context.get("source_dispatch_id"),
+        ),
+        recheck_dispatch_id=_extract_recheck_dispatch_id(context),
+        entry_id=_first_non_empty_string(
+            workflow.get("entry_id"),
+            context.get("entry_id"),
+            worker_result.get("entry_id"),
+        ),
+        project_id=_first_non_empty_string(
+            workflow.get("project_id"),
+            context.get("project_id"),
+        ) or project_id,
+        dispatch_status=dispatch_status,
+        workflow_stage=_extract_complete_workflow_stage(context, dispatch_status),
+        review_outcome=_normalize_optional_review_outcome(
+            workflow.get("review_outcome") or context.get("review_outcome") or worker_result.get("review_outcome")
+        ),
+        owner=_first_non_empty_string(context.get("owner")),
+        completed_at=row.completed_at,
+        stale_ratio_snapshot=snapshot,
+        stale_count_delta=stale_count_delta,
+        stale_ratio_delta=stale_ratio_delta,
+        above_threshold=(
+            snapshot.stale_ratio > snapshot.threshold
+            if snapshot is not None
+            else None
+        ),
+    )
+
+
+def _stale_burndown_status(
+    snapshot: KnowledgeReviewDraftStaleRatioSnapshot | None,
+) -> Literal["above_threshold", "at_or_below_threshold", "no_data"]:
+    if snapshot is None or snapshot.total_count <= 0:
+        return "no_data"
+    if snapshot.stale_ratio > snapshot.threshold:
+        return "above_threshold"
+    return "at_or_below_threshold"
+
+
+def _entries_to_stale_threshold(
+    snapshot: KnowledgeReviewDraftStaleRatioSnapshot | None,
+) -> int:
+    if snapshot is None or snapshot.total_count <= 0:
+        return 0
+    allowed_stale = int(snapshot.total_count * snapshot.threshold)
+    return max(0, snapshot.stale_count - allowed_stale)
+
+
 def _priority_rank(priority_tier: str) -> int:
     return {"P0": 3, "P1": 2, "P2": 1}.get(priority_tier, 0)
 
@@ -1246,6 +1465,7 @@ async def _complete_owner_review_and_write_audit(
                 request=request,
                 stale_ratio_snapshot=stale_ratio_snapshot,
                 plan_fingerprint=plan_fingerprint,
+                project_id=str(record.project_id),
             ),
             executor_type=_RECHECK_EXECUTOR_TYPE,
             attempt_count=0,
@@ -1473,10 +1693,13 @@ def _build_stale_ratio_recheck_context(
     request: KnowledgeStaleOwnerReviewCompleteRequest,
     stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
     plan_fingerprint: str,
+    project_id: str | None = None,
 ) -> dict[str, Any]:
+    resolved_project_id = project_id or "awoooi"
     return {
         "schema_version": "km_stale_owner_review_recheck_v1",
         "version": "v1",
+        "project_id": resolved_project_id,
         "trigger_source": "km_stale_owner_review_complete",
         "triggered_metric": "knowledge_degradation",
         "metric_value": stale_ratio_snapshot.stale_ratio,
@@ -1493,6 +1716,7 @@ def _build_stale_ratio_recheck_context(
             "work_kind": "km_stale_ratio_recheck",
             "current_stage": "stale_ratio_recheck",
             "entry_id": entry_id,
+            "project_id": resolved_project_id,
             "review_outcome": request.review_outcome,
             "steps": [
                 "detected",
@@ -1579,6 +1803,16 @@ def _normalize_review_outcome(value: Any) -> Literal[
     return "refresh_with_evidence"
 
 
+def _normalize_optional_review_outcome(value: Any) -> Literal[
+    "refresh_with_evidence",
+    "archive",
+    "supersede",
+] | None:
+    if value in ("archive", "supersede", "refresh_with_evidence"):
+        return value
+    return None
+
+
 def _extract_recheck_dispatch_id(context: dict[str, Any]) -> str | None:
     recheck = context.get("stale_ratio_recheck")
     if isinstance(recheck, dict):