feat(governance): surface km completion callback evidence

apps/api/src/api/v1/platform/operator_runs.py

@@ -95,6 +95,7 @@ class CallbackReplyItem(BaseModel):
     run_created_at: datetime | None = None
     callback_reply: dict[str, Any]
     awooop_status_chain: dict[str, Any] | None = None
+    km_stale_completion_summary: dict[str, Any] | None = None
     run_detail_href: str | None = None
 
 

apps/api/src/services/platform_operator_service.py

@@ -42,6 +42,9 @@ from src.services.awooop_truth_chain_service import (
     _summarize_mcp,
     fetch_truth_chain,
 )
+from src.services.governance_km_stale_review_service import (
+    query_km_stale_owner_review_completion_queue,
+)
 from src.services.ollama_endpoint_resolver import (
     OllamaEndpointSelection,
     OllamaWorkloadType,
@@ -100,6 +103,9 @@ _SOURCE_CORRELATION_PROVIDERS = ("sentry", "signoz")
 _SOURCE_CORRELATION_EVENT_LIMIT = 200
 _SOURCE_CORRELATION_LOOKBACK_DAYS = 7
 _SOURCE_CORRELATION_PRE_WINDOW_HOURS = 2
+_KM_STALE_COMPLETION_CALLBACK_SCHEMA_VERSION = (
+    "km_stale_owner_review_completion_callback_summary_v1"
+)
 
 # =============================================================================
 # Tenants
@@ -387,32 +393,53 @@ async def list_callback_replies(
 
     items = [_callback_reply_event_item(row) for row in rows]
     status_chain_cache: dict[tuple[str, str], dict[str, Any]] = {}
+    km_completion_queue_cache: dict[str, Any] = {}
+    km_completion_summary_cache: dict[tuple[str, str | None], dict[str, Any]] = {}
     for item in items:
         incident = item.get("incident_id")
+        item_project_id = str(item.get("project_id") or project_id or "awoooi")
         if not incident:
             item["awooop_status_chain"] = _build_awooop_status_chain(
                 incident_ids=[],
                 source_id=None,
             )
+            item["km_stale_completion_summary"] = (
+                _empty_km_stale_completion_summary(
+                    project_id=item_project_id,
+                    incident_id=None,
+                    status_value="no_incident",
+                    reason="callback_reply_missing_incident_id",
+                )
+            )
             continue
         incident_id = str(incident)
-        item_project_id = str(item.get("project_id") or project_id or "awoooi")
         cache_key = (item_project_id, incident_id)
         cached = status_chain_cache.get(cache_key)
         if cached is not None:
             item["awooop_status_chain"] = cached
-            continue
-        remediation_history = await _fetch_run_remediation_history(
-            [incident_id],
-            limit=5,
-        )
-        chain = await _fetch_awooop_status_chain(
-            incident_ids=[incident_id],
-            project_id=item_project_id,
-            remediation_history=remediation_history,
-        )
-        status_chain_cache[cache_key] = chain
-        item["awooop_status_chain"] = chain
+        else:
+            remediation_history = await _fetch_run_remediation_history(
+                [incident_id],
+                limit=5,
+            )
+            chain = await _fetch_awooop_status_chain(
+                incident_ids=[incident_id],
+                project_id=item_project_id,
+                remediation_history=remediation_history,
+            )
+            status_chain_cache[cache_key] = chain
+            item["awooop_status_chain"] = chain
+
+        summary_cache_key = (item_project_id, incident_id)
+        summary = km_completion_summary_cache.get(summary_cache_key)
+        if summary is None:
+            summary = await _fetch_km_stale_completion_summary_for_incident(
+                project_id=item_project_id,
+                incident_id=incident_id,
+                queue_cache=km_completion_queue_cache,
+            )
+            km_completion_summary_cache[summary_cache_key] = summary
+        item["km_stale_completion_summary"] = summary
 
     return {
         "items": items,
@@ -422,6 +449,54 @@ async def list_callback_replies(
     }
 
 
+async def _fetch_km_stale_completion_summary_for_incident(
+    *,
+    project_id: str,
+    incident_id: str | None,
+    queue_cache: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Fetch read-only KM owner-review completion context for callback evidence."""
+    normalized_project_id = project_id or "awoooi"
+    normalized_incident_id = str(incident_id or "").strip() or None
+    if not normalized_incident_id:
+        return _empty_km_stale_completion_summary(
+            project_id=normalized_project_id,
+            incident_id=None,
+            status_value="no_incident",
+            reason="callback_reply_missing_incident_id",
+        )
+
+    cache = queue_cache if queue_cache is not None else {}
+    queue = cache.get(normalized_project_id)
+    if queue is None:
+        try:
+            queue = await query_km_stale_owner_review_completion_queue(
+                project_id=normalized_project_id,
+                status_bucket="all",
+                limit=100,
+            )
+        except Exception as exc:
+            logger.warning(
+                "operator_km_stale_completion_summary_fetch_failed",
+                project_id=normalized_project_id,
+                incident_id=normalized_incident_id,
+                error=str(exc),
+            )
+            return _empty_km_stale_completion_summary(
+                project_id=normalized_project_id,
+                incident_id=normalized_incident_id,
+                status_value="fetch_failed",
+                reason="km_stale_completion_queue_fetch_failed",
+            )
+        cache[normalized_project_id] = queue
+
+    return _build_km_stale_completion_summary(
+        queue=queue,
+        project_id=normalized_project_id,
+        incident_id=normalized_incident_id,
+    )
+
+
 async def list_cicd_events(
     *,
     project_id: str | None,
@@ -985,6 +1060,106 @@ def _callback_reply_event_item(row: Mapping[str, Any]) -> dict[str, Any]:
     }
 
 
+def _empty_km_stale_completion_summary(
+    *,
+    project_id: str,
+    incident_id: str | None,
+    status_value: str,
+    reason: str | None = None,
+) -> dict[str, Any]:
+    """Build the nullable KM owner-review summary shape for callback evidence."""
+    return {
+        "schema_version": _KM_STALE_COMPLETION_CALLBACK_SCHEMA_VERSION,
+        "project_id": project_id,
+        "incident_id": incident_id,
+        "status": status_value,
+        "missing_reason": reason,
+        "total": 0,
+        "returned": 0,
+        "pending_count": 0,
+        "ready_count": 0,
+        "blocked_count": 0,
+        "completed_count": 0,
+        "failed_count": 0,
+        "writes_on_read": False,
+        "manual_review_required": True,
+        "batch_writes_allowed": False,
+        "items_truncated": False,
+        "related_total": 0,
+        "related_items": [],
+    }
+
+
+def _object_field(payload: Any, name: str, default: Any = None) -> Any:
+    if isinstance(payload, Mapping):
+        return payload.get(name, default)
+    return getattr(payload, name, default)
+
+
+def _object_int_field(payload: Any, name: str) -> int:
+    try:
+        return int(_object_field(payload, name, 0) or 0)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _build_km_stale_completion_summary(
+    *,
+    queue: Any,
+    project_id: str,
+    incident_id: str,
+) -> dict[str, Any]:
+    """Summarize KM owner-review completion queue state for one incident."""
+    related_items: list[dict[str, Any]] = []
+    for item in list(_object_field(queue, "items", []) or []):
+        if str(_object_field(item, "related_incident_id") or "").strip() != incident_id:
+            continue
+        related_items.append({
+            "entry_id": _object_field(item, "entry_id"),
+            "title": _object_field(item, "title"),
+            "dispatch_id": _object_field(item, "dispatch_id"),
+            "governance_event_id": _object_field(item, "governance_event_id"),
+            "readiness": _object_field(item, "readiness"),
+            "workflow_stage": _object_field(item, "workflow_stage"),
+            "next_action": _object_field(item, "next_action"),
+            "priority_tier": _object_field(item, "priority_tier"),
+            "recommended_completion_outcome": _object_field(
+                item,
+                "recommended_completion_outcome",
+            ),
+            "can_preview": bool(_object_field(item, "can_preview", False)),
+        })
+
+    total = _object_int_field(queue, "total")
+    returned = _object_int_field(queue, "returned")
+    return {
+        "schema_version": _KM_STALE_COMPLETION_CALLBACK_SCHEMA_VERSION,
+        "project_id": project_id,
+        "incident_id": incident_id,
+        "status": "matched_owner_review"
+        if related_items
+        else "no_related_owner_review",
+        "missing_reason": None if related_items else "no_matching_completion_item",
+        "total": total,
+        "returned": returned,
+        "pending_count": _object_int_field(queue, "pending_count"),
+        "ready_count": _object_int_field(queue, "ready_count"),
+        "blocked_count": _object_int_field(queue, "blocked_count"),
+        "completed_count": _object_int_field(queue, "completed_count"),
+        "failed_count": _object_int_field(queue, "failed_count"),
+        "writes_on_read": bool(_object_field(queue, "writes_on_read", False)),
+        "manual_review_required": bool(
+            _object_field(queue, "manual_review_required", True)
+        ),
+        "batch_writes_allowed": bool(
+            _object_field(queue, "batch_writes_allowed", False)
+        ),
+        "items_truncated": total > returned,
+        "related_total": len(related_items),
+        "related_items": related_items[:3],
+    }
+
+
 def _outbound_timeline_status(
     send_status: str,
     callback_reply: dict[str, Any] | None,

apps/api/tests/test_awooop_operator_timeline_labels.py

@@ -421,6 +421,28 @@ def test_list_callback_replies_response_preserves_callback_evidence() -> None:
                     "repair_state": "read_only_dry_run",
                     "needs_human": True,
                 },
+                "km_stale_completion_summary": {
+                    "schema_version": (
+                        "km_stale_owner_review_completion_callback_summary_v1"
+                    ),
+                    "project_id": "awoooi",
+                    "incident_id": "INC-20260513-79ED5E",
+                    "status": "matched_owner_review",
+                    "ready_count": 3,
+                    "blocked_count": 1,
+                    "completed_count": 2,
+                    "failed_count": 0,
+                    "batch_writes_allowed": False,
+                    "manual_review_required": True,
+                    "related_total": 1,
+                    "related_items": [
+                        {
+                            "entry_id": "km-1",
+                            "readiness": "ready",
+                            "next_action": "preview_stale_km_review_completion",
+                        }
+                    ],
+                },
                 "run_detail_href": (
                     "/awooop/runs/5c0306e0-591a-5445-9a33-80f499426b38"
                     "?project_id=awoooi"
@@ -438,9 +460,77 @@ def test_list_callback_replies_response_preserves_callback_evidence() -> None:
     assert dumped["items"][0]["awooop_status_chain"]["repair_state"] == (
         "read_only_dry_run"
     )
+    assert dumped["items"][0]["km_stale_completion_summary"]["ready_count"] == 3
+    assert dumped["items"][0]["km_stale_completion_summary"]["related_total"] == 1
     assert dumped["items"][0]["run_detail_href"].endswith("project_id=awoooi")
 
 
+@pytest.mark.asyncio
+async def test_km_stale_completion_summary_matches_callback_incident(
+    monkeypatch,
+) -> None:
+    async def fake_query_km_stale_completion_queue(**kwargs):
+        assert kwargs["project_id"] == "awoooi"
+        assert kwargs["status_bucket"] == "all"
+        assert kwargs["limit"] == 100
+        return SimpleNamespace(
+            project_id="awoooi",
+            total=2,
+            returned=2,
+            pending_count=2,
+            ready_count=1,
+            blocked_count=1,
+            completed_count=0,
+            failed_count=0,
+            writes_on_read=False,
+            manual_review_required=True,
+            batch_writes_allowed=False,
+            items=[
+                SimpleNamespace(
+                    entry_id="km-1",
+                    title="Bitan pharmacy status drift",
+                    dispatch_id="dispatch-1",
+                    governance_event_id="event-1",
+                    readiness="ready",
+                    workflow_stage="waiting_owner_review",
+                    next_action="preview_stale_km_review_completion",
+                    priority_tier="P0",
+                    recommended_completion_outcome="refresh_with_evidence",
+                    can_preview=True,
+                    related_incident_id="INC-20260513-79ED5E",
+                ),
+                SimpleNamespace(
+                    entry_id="km-2",
+                    title="Other stale KM",
+                    related_incident_id="INC-20260513-OTHER",
+                    readiness="blocked",
+                ),
+            ],
+        )
+
+    monkeypatch.setattr(
+        platform_operator_service,
+        "query_km_stale_owner_review_completion_queue",
+        fake_query_km_stale_completion_queue,
+    )
+
+    summary = await platform_operator_service._fetch_km_stale_completion_summary_for_incident(
+        project_id="awoooi",
+        incident_id="INC-20260513-79ED5E",
+        queue_cache={},
+    )
+
+    assert summary["schema_version"] == (
+        "km_stale_owner_review_completion_callback_summary_v1"
+    )
+    assert summary["status"] == "matched_owner_review"
+    assert summary["ready_count"] == 1
+    assert summary["blocked_count"] == 1
+    assert summary["batch_writes_allowed"] is False
+    assert summary["related_items"][0]["entry_id"] == "km-1"
+    assert summary["related_items"][0]["can_preview"] is True
+
+
 def test_list_approvals_response_preserves_status_chain() -> None:
     run_id = UUID("5c0306e0-591a-5445-9a33-80f499426b38")
     response = ListApprovalsResponse.model_validate({

apps/web/messages/en.json

@@ -2782,7 +2782,23 @@
         "sendStatus": "Send status: {status}",
         "providerMessage": "Message: {messageId}",
         "previewEmpty": "No preview",
-        "openRun": "Open Run"
+        "openRun": "Open Run",
+        "kmCompletion": {
+          "title": "KM Owner Review",
+          "status": "Status: {status}",
+          "counts": "ready {ready} / blocked {blocked} / completed {completed} / failed {failed}",
+          "guardrail": "Guardrail: writes_on_read={writesOnRead}; batch_writes_allowed={batchWrite}; manual_review_required={manualReview}",
+          "related": "{entryId} · {readiness} · {nextAction}",
+          "noRelated": "This incident has no matching owner-review completion item yet.",
+          "fetchFailed": "KM owner-review summary failed to load: {reason}",
+          "statuses": {
+            "matched_owner_review": "Matched owner review",
+            "no_related_owner_review": "No matched owner review",
+            "fetch_failed": "Fetch failed",
+            "no_incident": "Missing incident",
+            "observed": "Recorded"
+          }
+        }
       }
     },
     "aiRouteStatus": {

apps/web/messages/zh-TW.json

@@ -2783,7 +2783,23 @@
         "sendStatus": "送訊狀態：{status}",
         "providerMessage": "Message：{messageId}",
         "previewEmpty": "無摘要",
-        "openRun": "開啟 Run"
+        "openRun": "開啟 Run",
+        "kmCompletion": {
+          "title": "KM Owner Review",
+          "status": "狀態：{status}",
+          "counts": "ready {ready} / blocked {blocked} / completed {completed} / failed {failed}",
+          "guardrail": "Guardrail：writes_on_read={writesOnRead}；batch_writes_allowed={batchWrite}；manual_review_required={manualReview}",
+          "related": "{entryId} · {readiness} · {nextAction}",
+          "noRelated": "本 Incident 尚未對到 owner-review completion item。",
+          "fetchFailed": "KM owner-review 摘要讀取失敗：{reason}",
+          "statuses": {
+            "matched_owner_review": "已匹配 owner review",
+            "no_related_owner_review": "未匹配 owner review",
+            "fetch_failed": "讀取失敗",
+            "no_incident": "缺少 Incident",
+            "observed": "已記錄"
+          }
+        }
       }
     },
     "aiRouteStatus": {

apps/web/src/app/[locale]/awooop/runs/page.tsx

@@ -281,6 +281,40 @@ interface EventRecurrenceResponse {
   items: EventRecurrenceItem[];
 }
 
+interface KmStaleCompletionSummaryItem {
+  entry_id?: string | null;
+  title?: string | null;
+  dispatch_id?: string | null;
+  governance_event_id?: string | null;
+  readiness?: string | null;
+  workflow_stage?: string | null;
+  next_action?: string | null;
+  priority_tier?: string | null;
+  recommended_completion_outcome?: string | null;
+  can_preview?: boolean | null;
+}
+
+interface KmStaleCompletionSummary {
+  schema_version?: string;
+  project_id?: string | null;
+  incident_id?: string | null;
+  status?: string | null;
+  missing_reason?: string | null;
+  total?: number;
+  returned?: number;
+  pending_count?: number;
+  ready_count?: number;
+  blocked_count?: number;
+  completed_count?: number;
+  failed_count?: number;
+  writes_on_read?: boolean | null;
+  manual_review_required?: boolean | null;
+  batch_writes_allowed?: boolean | null;
+  items_truncated?: boolean | null;
+  related_total?: number;
+  related_items?: KmStaleCompletionSummaryItem[];
+}
+
 interface CallbackReplyEvent {
   message_id: string;
   run_id: string;
@@ -301,6 +335,7 @@ interface CallbackReplyEvent {
   agent_id?: string | null;
   run_detail_href?: string | null;
   awooop_status_chain?: AwoooPStatusChain | null;
+  km_stale_completion_summary?: KmStaleCompletionSummary | null;
 }
 
 interface CallbackRepliesResponse {
@@ -660,6 +695,18 @@ function normalizeCallbackReplyEventStatus(statusValue?: string | null): Callbac
   return "observed";
 }
 
+function normalizeKmCompletionStatus(statusValue?: string | null) {
+  if (
+    statusValue === "matched_owner_review" ||
+    statusValue === "no_related_owner_review" ||
+    statusValue === "fetch_failed" ||
+    statusValue === "no_incident"
+  ) {
+    return statusValue;
+  }
+  return "observed";
+}
+
 function RemediationEvidenceCell({ summary }: { summary?: RemediationSummary | null }) {
   const t = useTranslations("awooop.listEvidence");
   const status = normalizeRemediationStatus(summary);
@@ -1409,6 +1456,66 @@ function GroupedAlertEventsPanel({ events }: { events: PlatformEvent[] }) {
   );
 }
 
+function CallbackKmCompletionSummary({
+  summary,
+}: {
+  summary?: KmStaleCompletionSummary | null;
+}) {
+  const t = useTranslations("awooop.callbackReply.events.kmCompletion");
+  if (!summary) return null;
+
+  const statusKey = normalizeKmCompletionStatus(summary.status);
+  const related = Array.isArray(summary.related_items)
+    ? summary.related_items[0]
+    : null;
+
+  return (
+    <div className="mt-3 border-t border-[#e0ddd4] pt-3">
+      <div className="flex items-center gap-2">
+        <ShieldCheck className="h-3.5 w-3.5 text-[#17602a]" aria-hidden="true" />
+        <p className="text-xs font-semibold text-[#141413]">{t("title")}</p>
+      </div>
+      <div className="mt-2 grid gap-1 text-xs leading-5 text-[#5f5b52]">
+        <p>
+          {t("status", {
+            status: t(`statuses.${statusKey}` as never),
+          })}
+        </p>
+        <p>
+          {t("counts", {
+            ready: summary.ready_count ?? 0,
+            blocked: summary.blocked_count ?? 0,
+            completed: summary.completed_count ?? 0,
+            failed: summary.failed_count ?? 0,
+          })}
+        </p>
+        <p>
+          {t("guardrail", {
+            writesOnRead: String(summary.writes_on_read ?? false),
+            batchWrite: String(summary.batch_writes_allowed ?? false),
+            manualReview: String(summary.manual_review_required ?? true),
+          })}
+        </p>
+        {related ? (
+          <p className="truncate font-mono text-[#77736a]">
+            {t("related", {
+              entryId: related.entry_id ?? "--",
+              readiness: related.readiness ?? "--",
+              nextAction: related.next_action ?? "--",
+            })}
+          </p>
+        ) : statusKey === "fetch_failed" ? (
+          <p className="text-[#9f2f25]">
+            {t("fetchFailed", { reason: summary.missing_reason ?? "--" })}
+          </p>
+        ) : (
+          <p className="text-[#77736a]">{t("noRelated")}</p>
+        )}
+      </div>
+    </div>
+  );
+}
+
 function CallbackReplyEvidencePanel({
   events,
   total,
@@ -1499,6 +1606,9 @@ function CallbackReplyEvidencePanel({
                   compact
                   className="mt-3"
                 />
+                <CallbackKmCompletionSummary
+                  summary={event.km_stale_completion_summary}
+                />
                 <Link
                   href={runHref as never}
                   className="mt-3 inline-flex items-center gap-1.5 border border-[#d8d3c7] bg-[#faf9f3] px-2 py-1 text-xs font-semibold text-[#2e2b26] hover:border-[#1f6feb] hover:bg-[#edf4ff] hover:text-[#0f4fa8]"