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({