feat(governance): complete stale km owner review

apps/api/src/services/governance_km_stale_review_service.py

@@ -10,11 +10,14 @@ Owner-review intake for stale KM priority candidates.
 
 from __future__ import annotations
 
+import hashlib
+import json
 from datetime import timedelta
 from typing import Any, Literal
 
 import structlog
-from sqlalchemy import select, text
+from sqlalchemy import func, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.db.base import get_db_context
 from src.db.models import (
@@ -25,6 +28,9 @@ from src.db.models import (
     taipei_now,
 )
 from src.models.governance import (
+    KnowledgeReviewDraftStaleRatioSnapshot,
+    KnowledgeStaleOwnerReviewCompleteRequest,
+    KnowledgeStaleOwnerReviewCompleteResponse,
     KnowledgeStaleOwnerReviewRequest,
     KnowledgeStaleOwnerReviewResponse,
 )
@@ -36,6 +42,8 @@ from src.utils.timezone import now_taipei
 logger = structlog.get_logger(__name__)
 
 _EXECUTOR_TYPE = "hermes_km_stale_owner_review"
+_COMPLETE_EXECUTOR_TYPE = "hermes_km_stale_owner_review_complete"
+_RECHECK_EXECUTOR_TYPE = "hermes_km_stale_ratio_recheck"
 
 
 class KmStaleOwnerReviewError(Exception):
@@ -150,6 +158,110 @@ async def queue_km_stale_owner_review(
     )
 
 
+async def complete_km_stale_owner_review(
+    *,
+    entry_id: str,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+) -> KnowledgeStaleOwnerReviewCompleteResponse:
+    """Complete stale KM owner review after explicit owner approval."""
+    completed = await _load_completed_owner_review_dispatch(entry_id)
+    if completed is not None and not request.dry_run:
+        decision_context = completed.decision_context if isinstance(completed.decision_context, dict) else {}
+        workflow = decision_context.get("workflow") if isinstance(decision_context, dict) else {}
+        worker_result = decision_context.get("worker_result") if isinstance(decision_context, dict) else {}
+        return _build_complete_response(
+            entry_id=entry_id,
+            project_id=str(workflow.get("project_id") or decision_context.get("project_id") or "awoooi"),
+            status="already_completed",
+            review_outcome=_normalize_review_outcome(worker_result.get("review_outcome")),
+            governance_event_id=str(completed.governance_event_id),
+            dispatch_id=str(workflow.get("source_dispatch_id") or request.dispatch_id or ""),
+            audit_dispatch_id=str(completed.id),
+            stale_ratio_recheck_dispatch_id=_extract_recheck_dispatch_id(decision_context),
+            workflow_stage=_extract_complete_workflow_stage(decision_context, "succeeded"),
+            owner=request.owner,
+            owner_approved=request.owner_approved,
+            dry_run=request.dry_run,
+            writes_km=False,
+            writes_governance_audit=False,
+            stale_ratio_snapshot=_snapshot_from_context(decision_context),
+            dry_run_plan_fingerprint=_extract_plan_fingerprint(decision_context),
+        )
+
+    record = await _load_stale_or_reviewed_record(entry_id)
+    dispatch = await _load_owner_review_dispatch_for_completion(
+        entry_id=entry_id,
+        dispatch_id=request.dispatch_id,
+    )
+    _validate_completion_request(record, request)
+    superseded_by = await _load_superseded_by_record(request.superseded_by_entry_id)
+    fingerprint = _build_completion_plan_fingerprint(
+        record=record,
+        dispatch=dispatch,
+        request=request,
+        superseded_by=superseded_by,
+    )
+
+    if request.dry_run:
+        return _build_complete_response(
+            entry_id=entry_id,
+            project_id=str(record.project_id),
+            status="dry_run",
+            review_outcome=request.review_outcome,
+            governance_event_id=str(dispatch.governance_event_id),
+            dispatch_id=str(dispatch.id),
+            workflow_stage=_completion_stage_for_outcome(request.review_outcome),
+            owner=request.owner,
+            owner_approved=request.owner_approved,
+            dry_run=True,
+            writes_km=False,
+            writes_governance_audit=False,
+            stale_ratio_snapshot=await _load_current_km_stale_ratio_snapshot(),
+            dry_run_plan_fingerprint=fingerprint,
+        )
+
+    if not request.owner_approved:
+        raise KmStaleOwnerReviewError(
+            403,
+            "owner_approved=true is required before completing stale KM owner review",
+        )
+    if not request.dry_run_plan_fingerprint:
+        raise KmStaleOwnerReviewError(
+            403,
+            "dry_run_plan_fingerprint from a dry-run preview is required before writing KM",
+        )
+    if request.dry_run_plan_fingerprint != fingerprint:
+        raise KmStaleOwnerReviewError(
+            409,
+            "dry_run_plan_fingerprint does not match the latest stale KM review plan",
+        )
+
+    result = await _complete_owner_review_and_write_audit(
+        entry_id=entry_id,
+        request=request,
+        resolved_dispatch_id=str(dispatch.id),
+        plan_fingerprint=fingerprint,
+    )
+    return _build_complete_response(
+        entry_id=entry_id,
+        project_id=result["project_id"],
+        status="completed",
+        review_outcome=request.review_outcome,
+        governance_event_id=result["governance_event_id"],
+        dispatch_id=result["dispatch_id"],
+        audit_dispatch_id=result["audit_dispatch_id"],
+        stale_ratio_recheck_dispatch_id=result["recheck_dispatch_id"],
+        workflow_stage=_completion_stage_for_outcome(request.review_outcome),
+        owner=request.owner,
+        owner_approved=True,
+        dry_run=False,
+        writes_km=True,
+        writes_governance_audit=True,
+        stale_ratio_snapshot=result["stale_ratio_snapshot"],
+        dry_run_plan_fingerprint=fingerprint,
+    )
+
+
 async def _load_stale_candidate_record(entry_id: str) -> KnowledgeEntryRecord:
     cutoff = now_taipei() - timedelta(days=KM_STALE_DAYS)
     async with get_db_context() as db:
@@ -193,6 +305,657 @@ async def _load_active_owner_review_dispatch(
         return result.scalar_one_or_none()
 
 
+async def _load_owner_review_dispatch_for_completion(
+    *,
+    entry_id: str,
+    dispatch_id: str | None,
+) -> GovernanceRemediationDispatch:
+    """Load the active owner-review dispatch that guards KM writes."""
+    id_filter = "AND d.id = :dispatch_id" if dispatch_id else ""
+    sql = text(f"""
+        SELECT *
+        FROM governance_remediation_dispatch d
+        WHERE d.executor_type = :executor_type
+          AND d.dispatch_status::text IN ('pending', 'dispatched', 'executing')
+          AND (
+            d.decision_context -> 'workflow' ->> 'entry_id' = :entry_id
+            OR d.decision_context ->> 'entry_id' = :entry_id
+          )
+          {id_filter}
+        ORDER BY d.dispatched_at DESC
+        LIMIT 1
+    """)
+    params = {
+        "executor_type": _EXECUTOR_TYPE,
+        "entry_id": entry_id,
+    }
+    if dispatch_id:
+        params["dispatch_id"] = dispatch_id
+
+    async with get_db_context() as db:
+        result = await db.execute(
+            select(GovernanceRemediationDispatch).from_statement(sql),
+            params,
+        )
+        dispatch = result.scalar_one_or_none()
+
+    if dispatch is None:
+        raise KmStaleOwnerReviewError(
+            409,
+            "active stale KM owner-review dispatch not found; queue review first",
+        )
+    return dispatch
+
+
+async def _load_completed_owner_review_dispatch(
+    entry_id: str,
+) -> GovernanceRemediationDispatch | None:
+    sql = text("""
+        SELECT *
+        FROM governance_remediation_dispatch d
+        WHERE d.executor_type = :executor_type
+          AND d.dispatch_status::text = 'succeeded'
+          AND (
+            d.decision_context -> 'workflow' ->> 'entry_id' = :entry_id
+            OR d.decision_context ->> 'entry_id' = :entry_id
+          )
+        ORDER BY d.completed_at DESC NULLS LAST, d.dispatched_at DESC
+        LIMIT 1
+    """)
+    async with get_db_context() as db:
+        result = await db.execute(
+            select(GovernanceRemediationDispatch).from_statement(sql),
+            {"executor_type": _COMPLETE_EXECUTOR_TYPE, "entry_id": entry_id},
+        )
+        return result.scalar_one_or_none()
+
+
+async def _load_stale_or_reviewed_record(entry_id: str) -> KnowledgeEntryRecord:
+    async with get_db_context() as db:
+        result = await db.execute(
+            select(KnowledgeEntryRecord).where(KnowledgeEntryRecord.id == entry_id)
+        )
+        record = result.scalar_one_or_none()
+
+    if record is None:
+        raise KmStaleOwnerReviewError(404, "KM entry not found")
+    if _enum_value(record.status) == EntryStatus.ARCHIVED.value:
+        raise KmStaleOwnerReviewError(409, "archived KM entries cannot be completed as stale review")
+    return record
+
+
+async def _load_superseded_by_record(
+    superseded_by_entry_id: str | None,
+) -> KnowledgeEntryRecord | None:
+    if not superseded_by_entry_id:
+        return None
+    async with get_db_context() as db:
+        result = await db.execute(
+            select(KnowledgeEntryRecord).where(
+                KnowledgeEntryRecord.id == superseded_by_entry_id
+            )
+        )
+        record = result.scalar_one_or_none()
+    if record is None:
+        raise KmStaleOwnerReviewError(404, "superseded_by_entry_id KM entry not found")
+    if _enum_value(record.status) == EntryStatus.ARCHIVED.value:
+        raise KmStaleOwnerReviewError(409, "superseded_by_entry_id cannot be archived")
+    return record
+
+
+def _validate_completion_request(
+    record: KnowledgeEntryRecord,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+) -> None:
+    """Validate owner intent before any KM write is allowed."""
+    if request.review_outcome == "supersede":
+        if not request.superseded_by_entry_id:
+            raise KmStaleOwnerReviewError(
+                422,
+                "superseded_by_entry_id is required when review_outcome=supersede",
+            )
+        if request.superseded_by_entry_id == str(record.id):
+            raise KmStaleOwnerReviewError(409, "KM entry cannot supersede itself")
+    if request.review_outcome == "refresh_with_evidence" and not (
+        request.owner_note or request.updated_title or request.updated_content
+    ):
+        raise KmStaleOwnerReviewError(
+            422,
+            "owner_note or updated KM content is required for refresh_with_evidence",
+        )
+
+
+def _build_completion_plan_fingerprint(
+    *,
+    record: KnowledgeEntryRecord,
+    dispatch: GovernanceRemediationDispatch,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+    superseded_by: KnowledgeEntryRecord | None,
+) -> str:
+    """Fingerprint the exact owner-reviewed plan that may write KM."""
+    payload = {
+        "schema_version": "km_stale_owner_review_completion_plan_v1",
+        "entry_id": str(record.id),
+        "dispatch_id": str(dispatch.id),
+        "governance_event_id": str(dispatch.governance_event_id),
+        "review_outcome": request.review_outcome,
+        "current_status": _enum_value(record.status),
+        "current_updated_at": record.updated_at.isoformat() if record.updated_at else None,
+        "updated_title": request.updated_title,
+        "updated_content_sha256": (
+            hashlib.sha256(request.updated_content.encode("utf-8")).hexdigest()
+            if request.updated_content
+            else None
+        ),
+        "owner_note_sha256": (
+            hashlib.sha256(request.owner_note.encode("utf-8")).hexdigest()
+            if request.owner_note
+            else None
+        ),
+        "superseded_by_entry_id": str(superseded_by.id) if superseded_by else None,
+        "superseded_by_updated_at": (
+            superseded_by.updated_at.isoformat()
+            if superseded_by and superseded_by.updated_at
+            else None
+        ),
+    }
+    encoded = json.dumps(
+        payload,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return "sha256:" + hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+async def _complete_owner_review_and_write_audit(
+    *,
+    entry_id: str,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+    resolved_dispatch_id: str,
+    plan_fingerprint: str,
+) -> dict[str, Any]:
+    """Apply owner-approved KM outcome and write terminal dispatch trail."""
+    now = now_taipei()
+    async with get_db_context() as db:
+        record_result = await db.execute(
+            select(KnowledgeEntryRecord)
+            .where(KnowledgeEntryRecord.id == entry_id)
+            .with_for_update()
+        )
+        record = record_result.scalar_one_or_none()
+        if record is None:
+            raise KmStaleOwnerReviewError(404, "KM entry not found")
+        if _enum_value(record.status) == EntryStatus.ARCHIVED.value:
+            raise KmStaleOwnerReviewError(409, "KM entry was archived before completion")
+
+        dispatch_result = await db.execute(
+            select(GovernanceRemediationDispatch)
+            .where(GovernanceRemediationDispatch.id == resolved_dispatch_id)
+            .with_for_update()
+        )
+        dispatch = dispatch_result.scalar_one_or_none()
+        if dispatch is None:
+            raise KmStaleOwnerReviewError(
+                409,
+                "active stale KM owner-review dispatch not found during write",
+            )
+        if str(dispatch.dispatch_status) not in {"pending", "dispatched", "executing"}:
+            raise KmStaleOwnerReviewError(
+                409,
+                "stale KM owner-review dispatch is no longer active",
+            )
+        if str(dispatch.id) != (request.dispatch_id or str(dispatch.id)):
+            raise KmStaleOwnerReviewError(409, "dispatch_id does not match active owner review")
+        dispatch_entry_id = _dispatch_entry_id(dispatch)
+        if dispatch_entry_id != entry_id:
+            raise KmStaleOwnerReviewError(409, "dispatch_id does not match KM entry")
+
+        _apply_owner_review_outcome(record, request, now=now)
+        dispatch.dispatch_status = "succeeded"
+        dispatch.started_at = dispatch.started_at or taipei_now()
+        dispatch.completed_at = taipei_now()
+        dispatch.decision_context = _merge_owner_review_completion_into_context(
+            dispatch.decision_context if isinstance(dispatch.decision_context, dict) else {},
+            request=request,
+            plan_fingerprint=plan_fingerprint,
+            completed_at=now.isoformat(),
+        )
+
+        await db.flush()
+        stale_ratio_snapshot = await _compute_km_stale_ratio_snapshot(db)
+        recheck = GovernanceRemediationDispatch(
+            id=generate_uuid(),
+            governance_event_id=str(dispatch.governance_event_id),
+            event_type="knowledge_degradation",
+            dispatch_status="succeeded",
+            decision_context=_build_stale_ratio_recheck_context(
+                governance_event_id=str(dispatch.governance_event_id),
+                entry_id=entry_id,
+                request=request,
+                stale_ratio_snapshot=stale_ratio_snapshot,
+                plan_fingerprint=plan_fingerprint,
+            ),
+            executor_type=_RECHECK_EXECUTOR_TYPE,
+            attempt_count=0,
+            max_attempts=1,
+            dispatched_at=taipei_now(),
+            started_at=taipei_now(),
+            completed_at=taipei_now(),
+            created_by=request.owner[:100],
+        )
+        audit = GovernanceRemediationDispatch(
+            id=generate_uuid(),
+            governance_event_id=str(dispatch.governance_event_id),
+            event_type="knowledge_degradation",
+            dispatch_status="succeeded",
+            decision_context=_build_owner_review_completion_audit_context(
+                governance_event_id=str(dispatch.governance_event_id),
+                source_dispatch_id=str(dispatch.id),
+                entry=record,
+                request=request,
+                stale_ratio_snapshot=stale_ratio_snapshot,
+                recheck_dispatch_id=str(recheck.id),
+                plan_fingerprint=plan_fingerprint,
+                completed_at=now.isoformat(),
+            ),
+            executor_type=_COMPLETE_EXECUTOR_TYPE,
+            attempt_count=0,
+            max_attempts=1,
+            dispatched_at=taipei_now(),
+            started_at=taipei_now(),
+            completed_at=taipei_now(),
+            created_by=request.owner[:100],
+        )
+        db.add(recheck)
+        db.add(audit)
+        await db.flush()
+
+        return {
+            "project_id": str(record.project_id),
+            "governance_event_id": str(dispatch.governance_event_id),
+            "dispatch_id": str(dispatch.id),
+            "audit_dispatch_id": str(audit.id),
+            "recheck_dispatch_id": str(recheck.id),
+            "stale_ratio_snapshot": stale_ratio_snapshot,
+        }
+
+
+def _dispatch_entry_id(dispatch: GovernanceRemediationDispatch) -> str | None:
+    context = dispatch.decision_context if isinstance(dispatch.decision_context, dict) else {}
+    workflow = context.get("workflow") if isinstance(context.get("workflow"), dict) else {}
+    for source in (workflow, context):
+        value = source.get("entry_id") if isinstance(source, dict) else None
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
+def _apply_owner_review_outcome(
+    record: KnowledgeEntryRecord,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+    *,
+    now,
+) -> None:
+    if request.updated_title:
+        record.title = request.updated_title
+    if request.updated_content:
+        record.content = request.updated_content
+    if request.review_outcome in ("archive", "supersede"):
+        record.status = EntryStatus.ARCHIVED
+    record.tags = _append_owner_review_tags(
+        [str(tag) for tag in (record.tags or [])],
+        request=request,
+        reviewed_at=now.isoformat(),
+    )
+    record.updated_at = now
+
+
+def _append_owner_review_tags(
+    tags: list[str],
+    *,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+    reviewed_at: str,
+) -> list[str]:
+    additions = [
+        "reviewed_by:km_stale_owner_review",
+        f"stale_review_outcome:{request.review_outcome}",
+        f"stale_review_owner:{request.owner[:80]}",
+        f"stale_reviewed_at:{reviewed_at}",
+    ]
+    if request.superseded_by_entry_id:
+        additions.append(f"superseded_by:{request.superseded_by_entry_id}")
+    merged = list(tags)
+    for tag in additions:
+        if tag not in merged:
+            merged.append(tag)
+    return merged
+
+
+def _merge_owner_review_completion_into_context(
+    context: dict[str, Any],
+    *,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+    plan_fingerprint: str,
+    completed_at: str,
+) -> dict[str, Any]:
+    merged = dict(context)
+    workflow = dict(merged.get("workflow") if isinstance(merged.get("workflow"), dict) else {})
+    workflow.update({
+        "current_stage": _completion_stage_for_outcome(request.review_outcome),
+        "next_action": "stale_ratio_recheck",
+        "review_outcome": request.review_outcome,
+        "owner_approved": request.owner_approved,
+        "dry_run_plan_fingerprint": plan_fingerprint,
+        "completed_at": completed_at,
+        "writes_km": True,
+        "writes_governance_audit": True,
+    })
+    workflow["steps"] = list(dict.fromkeys([
+        *[str(step) for step in workflow.get("steps", []) if step is not None],
+        "owner_updates_or_archives_km",
+        _completion_stage_for_outcome(request.review_outcome),
+        "stale_ratio_recheck",
+    ]))
+    workflow["stage_by_dispatch_status"] = {
+        **(
+            workflow.get("stage_by_dispatch_status")
+            if isinstance(workflow.get("stage_by_dispatch_status"), dict)
+            else {}
+        ),
+        "succeeded": _completion_stage_for_outcome(request.review_outcome),
+        "failed": "needs_manual_km_triage",
+    }
+    merged["workflow"] = workflow
+    merged["next_action"] = "stale_ratio_recheck"
+    merged["decision_path"] = f"owner_approved_{request.review_outcome}"
+    merged["worker_result"] = {
+        "status": "owner_review_completed",
+        "review_outcome": request.review_outcome,
+        "writes_km": True,
+        "completed_at": completed_at,
+    }
+    return merged
+
+
+def _build_owner_review_completion_audit_context(
+    *,
+    governance_event_id: str,
+    source_dispatch_id: str,
+    entry: KnowledgeEntryRecord,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+    stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
+    recheck_dispatch_id: str,
+    plan_fingerprint: str,
+    completed_at: str,
+) -> dict[str, Any]:
+    stage = _completion_stage_for_outcome(request.review_outcome)
+    return {
+        "schema_version": "km_stale_owner_review_complete_audit_v1",
+        "version": "v1",
+        "project_id": str(entry.project_id),
+        "decision_path": f"owner_approved_{request.review_outcome}",
+        "next_action": "stale_ratio_recheck",
+        "ownership": _stale_owner_review_ownership(),
+        "workflow": {
+            "work_item_id": (
+                "governance:knowledge_degradation:"
+                f"{governance_event_id}:km_stale_owner_review_complete:{entry.id}"
+            ),
+            "work_kind": "km_stale_owner_review_complete",
+            "current_stage": stage,
+            "entry_id": str(entry.id),
+            "project_id": str(entry.project_id),
+            "source_dispatch_id": source_dispatch_id,
+            "steps": [
+                "detected",
+                "prioritized_stale_candidate",
+                "waiting_owner_review",
+                "owner_updates_or_archives_km",
+                stage,
+                "stale_ratio_recheck",
+            ],
+            "stage_by_dispatch_status": {
+                "succeeded": stage,
+                "failed": "needs_manual_km_triage",
+                "cancelled": "cancelled",
+            },
+            "next_action": "stale_ratio_recheck",
+            "writes_km_without_approval": False,
+            "writes_km": True,
+            "writes_governance_audit": True,
+            "dry_run_plan_fingerprint": plan_fingerprint,
+            "stale_ratio_snapshot": stale_ratio_snapshot.model_dump(),
+        },
+        "worker_result": {
+            "status": "owner_review_completed",
+            "entry_id": str(entry.id),
+            "review_outcome": request.review_outcome,
+            "owner_approved": request.owner_approved,
+            "writes_km": True,
+            "stale_ratio": stale_ratio_snapshot.stale_ratio,
+            "threshold": stale_ratio_snapshot.threshold,
+            "above_threshold": stale_ratio_snapshot.stale_ratio > stale_ratio_snapshot.threshold,
+        },
+        "owner": request.owner,
+        "owner_note": request.owner_note,
+        "entry_id": str(entry.id),
+        "review_outcome": request.review_outcome,
+        "source_dispatch_id": source_dispatch_id,
+        "superseded_by_entry_id": request.superseded_by_entry_id,
+        "updated_title": request.updated_title,
+        "updated_content_provided": bool(request.updated_content),
+        "dry_run_plan_fingerprint": plan_fingerprint,
+        "stale_ratio_recheck": {
+            "status": "completed",
+            "dispatch_id": recheck_dispatch_id,
+            "executor_type": _RECHECK_EXECUTOR_TYPE,
+        },
+        "completed_at": completed_at,
+    }
+
+
+def _build_stale_ratio_recheck_context(
+    *,
+    governance_event_id: str,
+    entry_id: str,
+    request: KnowledgeStaleOwnerReviewCompleteRequest,
+    stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot,
+    plan_fingerprint: str,
+) -> dict[str, Any]:
+    return {
+        "schema_version": "km_stale_owner_review_recheck_v1",
+        "version": "v1",
+        "trigger_source": "km_stale_owner_review_complete",
+        "triggered_metric": "knowledge_degradation",
+        "metric_value": stale_ratio_snapshot.stale_ratio,
+        "threshold": stale_ratio_snapshot.threshold,
+        "suggested_action": "run_stale_ratio_recheck",
+        "next_action": "run_stale_ratio_recheck",
+        "decision_path": f"stale_ratio_recheck_after_{request.review_outcome}",
+        "ownership": _stale_owner_review_ownership(),
+        "workflow": {
+            "work_item_id": (
+                "governance:knowledge_degradation:"
+                f"{governance_event_id}:stale_ratio_recheck:{entry_id}"
+            ),
+            "work_kind": "km_stale_ratio_recheck",
+            "current_stage": "stale_ratio_recheck",
+            "entry_id": entry_id,
+            "review_outcome": request.review_outcome,
+            "steps": [
+                "detected",
+                "waiting_owner_review",
+                "owner_updates_or_archives_km",
+                "stale_ratio_recheck",
+                "km_governance_close_or_continue",
+            ],
+            "stage_by_dispatch_status": {
+                "pending": "stale_ratio_recheck",
+                "dispatched": "stale_ratio_recheck",
+                "executing": "stale_ratio_recheck",
+                "succeeded": "km_governance_rechecked",
+                "failed": "needs_manual_km_triage",
+                "skipped": "needs_manual_km_triage",
+                "cancelled": "cancelled",
+            },
+            "next_action": "km_governance_close_or_continue",
+            "writes_km_without_approval": False,
+            "writes_km": False,
+            "dry_run_plan_fingerprint": plan_fingerprint,
+            "stale_ratio_snapshot": stale_ratio_snapshot.model_dump(),
+        },
+        "worker_result": {
+            "status": "stale_ratio_rechecked",
+            "entry_id": entry_id,
+            "review_outcome": request.review_outcome,
+            "stale_ratio": stale_ratio_snapshot.stale_ratio,
+            "threshold": stale_ratio_snapshot.threshold,
+            "above_threshold": stale_ratio_snapshot.stale_ratio > stale_ratio_snapshot.threshold,
+        },
+    }
+
+
+async def _load_current_km_stale_ratio_snapshot() -> KnowledgeReviewDraftStaleRatioSnapshot:
+    async with get_db_context() as db:
+        return await _compute_km_stale_ratio_snapshot(db)
+
+
+async def _compute_km_stale_ratio_snapshot(
+    db: AsyncSession,
+) -> KnowledgeReviewDraftStaleRatioSnapshot:
+    stale_cutoff = now_taipei() - timedelta(days=KM_STALE_DAYS)
+    total_result = await db.execute(
+        select(func.count()).select_from(KnowledgeEntryRecord).where(
+            KnowledgeEntryRecord.status != EntryStatus.ARCHIVED,
+        )
+    )
+    total = int(total_result.scalar() or 0)
+    stale_result = await db.execute(
+        select(func.count()).select_from(KnowledgeEntryRecord).where(
+            KnowledgeEntryRecord.status != EntryStatus.ARCHIVED,
+            KnowledgeEntryRecord.updated_at < stale_cutoff,
+        )
+    )
+    stale = int(stale_result.scalar() or 0)
+    ratio = round(stale / total, 3) if total > 0 else 0.0
+    return KnowledgeReviewDraftStaleRatioSnapshot(
+        stale_count=stale,
+        total_count=total,
+        stale_ratio=ratio,
+        threshold=KM_STALE_RATIO,
+        stale_days=KM_STALE_DAYS,
+    )
+
+
+def _completion_stage_for_outcome(
+    outcome: Literal["refresh_with_evidence", "archive", "supersede"],
+) -> str:
+    return {
+        "refresh_with_evidence": "km_writeback_after_approval",
+        "archive": "km_archive_after_approval",
+        "supersede": "km_supersede_after_approval",
+    }[outcome]
+
+
+def _normalize_review_outcome(value: Any) -> Literal[
+    "refresh_with_evidence",
+    "archive",
+    "supersede",
+]:
+    if value in ("archive", "supersede", "refresh_with_evidence"):
+        return value
+    return "refresh_with_evidence"
+
+
+def _extract_recheck_dispatch_id(context: dict[str, Any]) -> str | None:
+    recheck = context.get("stale_ratio_recheck")
+    if isinstance(recheck, dict):
+        value = recheck.get("dispatch_id")
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
+def _extract_complete_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 "km_writeback_after_approval"
+
+
+def _snapshot_from_context(
+    context: dict[str, Any],
+) -> KnowledgeReviewDraftStaleRatioSnapshot | None:
+    for source in (context, context.get("workflow")):
+        if not isinstance(source, dict):
+            continue
+        raw = source.get("stale_ratio_snapshot")
+        if isinstance(raw, dict):
+            try:
+                return KnowledgeReviewDraftStaleRatioSnapshot(**raw)
+            except Exception:
+                return None
+    return None
+
+
+def _extract_plan_fingerprint(context: dict[str, Any]) -> str | None:
+    for source in (context, context.get("workflow")):
+        if not isinstance(source, dict):
+            continue
+        value = source.get("dry_run_plan_fingerprint")
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
+def _build_complete_response(
+    *,
+    entry_id: str,
+    project_id: str,
+    status: Literal["dry_run", "completed", "already_completed"],
+    review_outcome: Literal["refresh_with_evidence", "archive", "supersede"],
+    governance_event_id: str,
+    dispatch_id: str,
+    workflow_stage: str,
+    owner: str,
+    owner_approved: bool,
+    dry_run: bool,
+    writes_km: bool,
+    writes_governance_audit: bool,
+    audit_dispatch_id: str | None = None,
+    stale_ratio_recheck_dispatch_id: str | None = None,
+    stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot | None = None,
+    dry_run_plan_fingerprint: str | None = None,
+) -> KnowledgeStaleOwnerReviewCompleteResponse:
+    return KnowledgeStaleOwnerReviewCompleteResponse(
+        entry_id=entry_id,
+        project_id=project_id,
+        status=status,
+        review_outcome=review_outcome,
+        governance_event_id=governance_event_id,
+        dispatch_id=dispatch_id,
+        audit_dispatch_id=audit_dispatch_id,
+        stale_ratio_recheck_dispatch_id=stale_ratio_recheck_dispatch_id,
+        workflow_stage=workflow_stage,
+        owner=owner,
+        owner_approved=owner_approved,
+        dry_run=dry_run,
+        writes_km=writes_km,
+        writes_governance_audit=writes_governance_audit,
+        stale_ratio_snapshot=stale_ratio_snapshot,
+        dry_run_plan_fingerprint=dry_run_plan_fingerprint,
+        generated_at=now_taipei(),
+    )
+
+
 def _build_stale_owner_review_event_details(
     *,
     entry_id: str,

apps/api/src/services/governance_query_service.py

@@ -923,6 +923,13 @@ async def query_km_stale_candidates(
         reverse=True,
     )
     limited = candidates[:limit]
+    owner_review_state = await _load_km_stale_owner_review_state_by_entry(
+        [candidate.entry_id for candidate in limited]
+    )
+    limited = [
+        candidate.model_copy(update=owner_review_state.get(candidate.entry_id, {}))
+        for candidate in limited
+    ]
 
     return KnowledgeStaleCandidatesResponse(
         project_id=project_id,
@@ -1033,6 +1040,61 @@ def _build_km_stale_candidate(
     )
 
 
+async def _load_km_stale_owner_review_state_by_entry(
+    entry_ids: list[str],
+) -> dict[str, dict[str, str]]:
+    """讀取 stale KM owner-review / completion 的最新狀態，供前端接續處理。"""
+    if not entry_ids:
+        return {}
+
+    unique_entry_ids = list(dict.fromkeys(entry_ids))
+    sql = text("""
+        SELECT DISTINCT ON (entry_id)
+            entry_id,
+            d.id,
+            d.dispatch_status,
+            d.decision_context
+        FROM (
+            SELECT
+                d.*,
+                COALESCE(
+                    d.decision_context -> 'workflow' ->> 'entry_id',
+                    d.decision_context ->> 'entry_id'
+                ) AS entry_id
+            FROM governance_remediation_dispatch d
+            WHERE d.executor_type IN (
+                'hermes_km_stale_owner_review',
+                'hermes_km_stale_owner_review_complete'
+            )
+        ) d
+        WHERE entry_id IN :entry_ids
+        ORDER BY entry_id, d.dispatched_at DESC
+    """).bindparams(bindparam("entry_ids", expanding=True))
+
+    try:
+        async with get_db_context() as db:
+            result = await db.execute(sql, {"entry_ids": unique_entry_ids})
+            rows = result.fetchall()
+    except ProgrammingError as exc:
+        logger.warning(
+            "km_stale_owner_review_state_table_not_ready",
+            error=str(exc),
+        )
+        return {}
+
+    state: dict[str, dict[str, str]] = {}
+    for row in rows:
+        decision_ctx: dict = row.decision_context if isinstance(row.decision_context, dict) else {}
+        dispatch_status = str(row.dispatch_status)
+        state[str(row.entry_id)] = {
+            "owner_review_dispatch_id": str(row.id),
+            "owner_review_status": dispatch_status,
+            "owner_review_stage": _extract_workflow_stage(decision_ctx, dispatch_status) or "",
+            "owner_review_next_action": _extract_next_action(decision_ctx) or "",
+        }
+    return state
+
+
 def _km_priority_tier(
     score: int,
     record: KnowledgeEntryRecord,