feat(awooop): surface remediation evidence in run lists

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

@@ -51,6 +51,7 @@ class RunItem(BaseModel):
     step_count: int
     created_at: datetime
     timeout_at: datetime | None
+    remediation_summary: dict[str, Any] | None = None
 
 
 class ListRunsResponse(BaseModel):
@@ -66,6 +67,7 @@ class ApprovalItem(BaseModel):
     agent_id: str
     created_at: datetime
     timeout_at: datetime | None
+    remediation_summary: dict[str, Any] | None = None
 
 
 class ListApprovalsResponse(BaseModel):

apps/api/src/services/platform_operator_service.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import re
 import uuid
+from collections import defaultdict
 from datetime import UTC, datetime
 from typing import Any
 from uuid import UUID
@@ -40,6 +41,7 @@ _DEFAULT_PER_PAGE = 50
 _MAX_PER_PAGE = 200
 _MAX_EVENTS = 100
 _MAX_TIMELINE_ITEMS = 100
+_MAX_LIST_CONTEXT_ROWS = 500
 _MAX_STEP_SUMMARY_CHARS = 128
 _REMEDIATION_HISTORY_LIMIT = 20
 _INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b")
@@ -151,6 +153,14 @@ async def list_runs(
         result = await db.execute(stmt)
         rows = list(result.scalars().all())
 
+        inbound_by_run, outbound_by_run = await _load_run_message_context(db, rows)
+
+    remediation_summaries = await _build_run_remediation_summaries(
+        runs=rows,
+        inbound_by_run=inbound_by_run,
+        outbound_by_run=outbound_by_run,
+    )
+
     runs = [
         {
             "run_id": r.run_id,
@@ -162,6 +172,7 @@ async def list_runs(
             "step_count": r.step_count,
             "created_at": r.created_at,
             "timeout_at": r.timeout_at,
+            "remediation_summary": remediation_summaries.get(r.run_id),
         }
         for r in rows
     ]
@@ -309,6 +320,67 @@ def _collect_run_incident_ids(
     return incident_ids
 
 
+async def _load_run_message_context(
+    db: Any,
+    runs: list[AwoooPRunState],
+) -> tuple[
+    dict[UUID, list[AwoooPConversationEvent]],
+    dict[UUID, list[AwoooPOutboundMessage]],
+]:
+    """Load list-page sidecar events needed to link runs back to incidents."""
+    if not runs:
+        return {}, {}
+
+    run_ids = [run.run_id for run in runs]
+    run_ids_set = set(run_ids)
+    trigger_refs = [str(run.trigger_ref) for run in runs if run.trigger_ref]
+    trigger_ref_to_run = {
+        str(run.trigger_ref): run.run_id
+        for run in runs
+        if run.trigger_ref
+    }
+    trigger_event_ids: list[UUID] = []
+    for trigger_ref in trigger_refs:
+        try:
+            trigger_event_ids.append(uuid.UUID(trigger_ref))
+        except ValueError:
+            continue
+
+    inbound_filters = [AwoooPConversationEvent.run_id.in_(run_ids)]
+    if trigger_refs:
+        inbound_filters.append(AwoooPConversationEvent.provider_event_id.in_(trigger_refs))
+    if trigger_event_ids:
+        inbound_filters.append(AwoooPConversationEvent.event_id.in_(trigger_event_ids))
+
+    inbound_result = await db.execute(
+        select(AwoooPConversationEvent)
+        .where(sa_or(*inbound_filters))
+        .order_by(AwoooPConversationEvent.received_at.desc())
+        .limit(_MAX_LIST_CONTEXT_ROWS)
+    )
+    inbound_by_run: dict[UUID, list[AwoooPConversationEvent]] = defaultdict(list)
+    for event in inbound_result.scalars().all():
+        target_run_id = event.run_id if event.run_id in run_ids_set else None
+        if target_run_id is None:
+            target_run_id = trigger_ref_to_run.get(str(event.provider_event_id))
+        if target_run_id is None:
+            target_run_id = trigger_ref_to_run.get(str(event.event_id))
+        if target_run_id is not None:
+            inbound_by_run[target_run_id].append(event)
+
+    outbound_result = await db.execute(
+        select(AwoooPOutboundMessage)
+        .where(AwoooPOutboundMessage.run_id.in_(run_ids))
+        .order_by(AwoooPOutboundMessage.queued_at.desc())
+        .limit(_MAX_LIST_CONTEXT_ROWS)
+    )
+    outbound_by_run: dict[UUID, list[AwoooPOutboundMessage]] = defaultdict(list)
+    for message in outbound_result.scalars().all():
+        outbound_by_run[message.run_id].append(message)
+
+    return dict(inbound_by_run), dict(outbound_by_run)
+
+
 def _route_label_from_remediation(item: dict[str, Any]) -> str:
     """Render remediation MCP route consistently with Telegram / Work Items."""
     return "/".join(
@@ -341,6 +413,132 @@ def _remediation_timeline_summary(item: dict[str, Any]) -> str:
     )[:500]
 
 
+def _run_remediation_list_summary(
+    *,
+    run: AwoooPRunState,
+    incident_ids: list[str],
+    items: list[dict[str, Any]],
+    errors: list[dict[str, str]] | None = None,
+) -> dict[str, Any]:
+    """Summarize durable ADR-100 dry-run evidence for list-level UX."""
+    sorted_items = sorted(
+        (item for item in items if isinstance(item, dict)),
+        key=lambda item: str(item.get("created_at") or ""),
+        reverse=True,
+    )
+    latest = sorted_items[0] if sorted_items else {}
+    writes_incident = latest.get("writes_incident_state")
+    writes_auto_repair = latest.get("writes_auto_repair_result")
+    route = _route_label_from_remediation(latest) if latest else "--"
+    write_observed = writes_incident is True or writes_auto_repair is True
+    is_read_only = (
+        bool(latest)
+        and latest.get("required_scope") == "read"
+        and writes_incident is False
+        and writes_auto_repair is False
+    )
+
+    if not sorted_items:
+        status_value = "no_evidence"
+    elif latest.get("success") is False or latest.get("allowed") is False:
+        status_value = "blocked"
+    elif write_observed:
+        status_value = "write_observed"
+    elif is_read_only:
+        status_value = "read_only_dry_run"
+    else:
+        status_value = "observed"
+
+    return {
+        "schema_version": "awooop_run_remediation_summary_v1",
+        "source": "alert_operation_log",
+        "incident_ids": incident_ids,
+        "total": len(sorted_items),
+        "status": status_value,
+        "has_dry_run": bool(sorted_items),
+        "is_read_only": is_read_only,
+        "human_gate_open": run.state == "waiting_approval",
+        "latest_at": latest.get("created_at"),
+        "latest_preview": latest.get("verification_result_preview"),
+        "latest_mode": latest.get("mode"),
+        "latest_route": route,
+        "latest_agent_id": latest.get("agent_id"),
+        "latest_tool_name": latest.get("tool_name"),
+        "latest_required_scope": latest.get("required_scope"),
+        "writes_incident_state": writes_incident,
+        "writes_auto_repair_result": writes_auto_repair,
+        "errors": errors or [],
+    }
+
+
+async def _build_run_remediation_summaries(
+    *,
+    runs: list[AwoooPRunState],
+    inbound_by_run: dict[UUID, list[AwoooPConversationEvent]],
+    outbound_by_run: dict[UUID, list[AwoooPOutboundMessage]],
+) -> dict[UUID, dict[str, Any]]:
+    """Build remediation summaries for list endpoints without writing state."""
+    if not runs:
+        return {}
+
+    incident_ids_by_run: dict[UUID, list[str]] = {}
+    all_incident_ids: list[str] = []
+    for run in runs:
+        incident_ids = _collect_run_incident_ids(
+            run=run,
+            inbound_events=inbound_by_run.get(run.run_id, []),
+            outbound_messages=outbound_by_run.get(run.run_id, []),
+        )
+        incident_ids_by_run[run.run_id] = incident_ids
+        for incident_id in incident_ids:
+            _append_unique(all_incident_ids, incident_id)
+
+    histories_by_incident: dict[str, list[dict[str, Any]]] = {}
+    errors_by_incident: dict[str, dict[str, str]] = {}
+    if all_incident_ids:
+        from src.services.adr100_remediation_service import Adr100RemediationService
+
+        service = Adr100RemediationService(record_history=False)
+        for incident_id in all_incident_ids:
+            try:
+                history = await service.history(
+                    limit=_REMEDIATION_HISTORY_LIMIT,
+                    incident_id=incident_id,
+                )
+                histories_by_incident[incident_id] = [
+                    item
+                    for item in history.get("items", [])
+                    if isinstance(item, dict)
+                ]
+            except Exception as exc:
+                logger.warning(
+                    "run_list_remediation_history_fetch_failed",
+                    incident_id=incident_id,
+                    error=str(exc),
+                )
+                errors_by_incident[incident_id] = {
+                    "incident_id": incident_id,
+                    "error": str(exc),
+                }
+
+    summaries: dict[UUID, dict[str, Any]] = {}
+    for run in runs:
+        incident_ids = incident_ids_by_run.get(run.run_id, [])
+        items: list[dict[str, Any]] = []
+        errors: list[dict[str, str]] = []
+        for incident_id in incident_ids:
+            items.extend(histories_by_incident.get(incident_id, []))
+            if incident_id in errors_by_incident:
+                errors.append(errors_by_incident[incident_id])
+        summaries[run.run_id] = _run_remediation_list_summary(
+            run=run,
+            incident_ids=incident_ids,
+            items=items,
+            errors=errors,
+        )
+    return summaries
+
+
 def _timeline_sort_key(item: dict[str, Any], fallback_ts: Any) -> str:
     """Normalize mixed DB datetime / ISO string timestamps for timeline sorting."""
     value = item.get("ts") or fallback_ts
@@ -816,6 +1014,14 @@ async def list_approvals(
         result = await db.execute(stmt)
         rows = list(result.scalars().all())
 
+        inbound_by_run, outbound_by_run = await _load_run_message_context(db, rows)
+
+    remediation_summaries = await _build_run_remediation_summaries(
+        runs=rows,
+        inbound_by_run=inbound_by_run,
+        outbound_by_run=outbound_by_run,
+    )
+
     items = [
         {
             "run_id": r.run_id,
@@ -823,6 +1029,7 @@ async def list_approvals(
             "agent_id": r.agent_id,
             "created_at": r.created_at,
             "timeout_at": r.timeout_at,
+            "remediation_summary": remediation_summaries.get(r.run_id),
         }
         for r in rows
     ]