feat(awooop): search callback reply evidence

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

@@ -31,6 +31,9 @@ from src.services.platform_operator_service import (
 from src.services.platform_operator_service import (
     list_approvals as list_approvals_svc,
 )
+from src.services.platform_operator_service import (
+    list_callback_replies as list_callback_replies_svc,
+)
 from src.services.platform_operator_service import (
     list_runs as list_runs_svc,
 )
@@ -62,6 +65,36 @@ class ListRunsResponse(BaseModel):
     per_page: int
 
 
+class CallbackReplyItem(BaseModel):
+    message_id: UUID
+    run_id: UUID
+    project_id: str
+    status: str
+    needs_human: bool
+    action: str | None = None
+    incident_id: str | None = None
+    event_at: datetime | None = None
+    channel_type: str
+    message_type: str
+    send_status: str
+    send_error: str | None = None
+    provider_message_id: str | None = None
+    triggered_by_state: str | None = None
+    content_preview: str | None = None
+    run_state: str | None = None
+    agent_id: str | None = None
+    run_created_at: datetime | None = None
+    callback_reply: dict[str, Any]
+    run_detail_href: str | None = None
+
+
+class ListCallbackRepliesResponse(BaseModel):
+    items: list[CallbackReplyItem]
+    total: int
+    page: int
+    per_page: int
+
+
 class ApprovalItem(BaseModel):
     run_id: UUID
     project_id: str
@@ -130,6 +163,36 @@ async def list_runs(
     )
 
 
+@router.get(
+    "/runs/callback-replies",
+    response_model=ListCallbackRepliesResponse,
+    summary="列出 Telegram Callback Reply Evidence",
+    description=(
+        "從 AwoooP outbound mirror 查詢 Telegram 詳情 / 歷史 callback reply 的"
+        "送達、fallback、救援與失敗證據；只讀，不修改 incident、run 或 Telegram 狀態。"
+    ),
+)
+async def list_callback_replies(
+    project_id: str | None = Query(None, description="租戶 ID（可選）"),
+    callback_reply_status: str | None = Query(
+        None,
+        description="Telegram callback reply 狀態 filter（sent/fallback_sent/rescue_sent/failed/observed/no_callback）",
+    ),
+    action: str | None = Query(None, description="Callback action filter（例如 detail/history）"),
+    incident_id: str | None = Query(None, description="關聯 Incident ID filter（可選）"),
+    page: int = Query(1, ge=1, description="頁碼，從 1 開始"),
+    per_page: int = Query(20, ge=1, le=_MAX_PER_PAGE, description="每頁筆數"),
+) -> dict[str, Any]:
+    return await list_callback_replies_svc(
+        project_id=project_id,
+        callback_reply_status=callback_reply_status,
+        action=action,
+        incident_id=incident_id,
+        page=page,
+        per_page=per_page,
+    )
+
+
 @router.get(
     "/runs/{run_id}/detail",
     summary="查詢 Run 詳細時間線",

apps/api/src/services/platform_operator_service.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import re
 import uuid
+from collections.abc import Mapping
 from collections import defaultdict
 from datetime import UTC, datetime
 from typing import Any
@@ -65,6 +66,13 @@ _CALLBACK_REPLY_STATUS_FILTERS = {
     "failed",
     "observed",
 }
+_CALLBACK_REPLY_RAW_STATUS_BY_FILTER = {
+    "sent": "callback_reply_sent",
+    "fallback_sent": "callback_reply_fallback_sent",
+    "rescue_sent": "callback_reply_rescue_sent",
+    "failed": "callback_reply_failed",
+}
+_CALLBACK_REPLY_ACTION_RE = re.compile(r"^[a-z0-9_:-]{1,64}$", re.IGNORECASE)
 
 # =============================================================================
 # Tenants
@@ -246,6 +254,116 @@ async def list_runs(
     return {"runs": runs, "total": total, "page": page, "per_page": per_page}
 
 
+async def list_callback_replies(
+    project_id: str | None,
+    callback_reply_status: str | None,
+    action: str | None,
+    incident_id: str | None,
+    page: int,
+    per_page: int,
+) -> dict[str, Any]:
+    """列出 Telegram detail/history callback reply evidence，不改 runtime 狀態。"""
+    _validate_callback_reply_status_filter(callback_reply_status)
+    callback_action = _validate_callback_reply_action_filter(action)
+    _validate_incident_id_filter(incident_id)
+
+    if callback_reply_status == "no_callback":
+        return {
+            "items": [],
+            "total": 0,
+            "page": page,
+            "per_page": per_page,
+        }
+
+    where_clauses = [
+        "m.source_envelope ? 'callback_reply'",
+        "(:project_id IS NULL OR m.project_id = :project_id)",
+    ]
+    params: dict[str, Any] = {
+        "project_id": project_id,
+        "limit": per_page,
+        "offset": (page - 1) * per_page,
+    }
+
+    raw_status = _CALLBACK_REPLY_RAW_STATUS_BY_FILTER.get(
+        str(callback_reply_status or "")
+    )
+    if raw_status:
+        where_clauses.append(
+            "m.source_envelope #>> '{callback_reply,status}' = :raw_status"
+        )
+        params["raw_status"] = raw_status
+    elif callback_reply_status == "observed":
+        where_clauses.append(
+            """
+            COALESCE(m.source_envelope #>> '{callback_reply,status}', '')
+            NOT IN (
+                'callback_reply_sent',
+                'callback_reply_fallback_sent',
+                'callback_reply_rescue_sent',
+                'callback_reply_failed'
+            )
+            """
+        )
+
+    if callback_action:
+        where_clauses.append(
+            "LOWER(m.source_envelope #>> '{callback_reply,action}') = :callback_action"
+        )
+        params["callback_action"] = callback_action
+    if incident_id:
+        where_clauses.append(
+            "m.source_envelope #>> '{callback_reply,incident_id}' = :incident_id"
+        )
+        params["incident_id"] = incident_id
+
+    where_sql = " AND ".join(where_clauses)
+    count_sql = text(f"""
+        SELECT COUNT(*) AS total
+        FROM awooop_outbound_message m
+        WHERE {where_sql}
+    """)
+    list_sql = text(f"""
+        SELECT
+            m.message_id,
+            m.project_id,
+            m.run_id,
+            m.channel_type,
+            m.message_type,
+            m.content_preview,
+            m.provider_message_id,
+            m.send_status,
+            m.send_error,
+            m.queued_at,
+            m.sent_at,
+            m.triggered_by_state,
+            m.source_envelope -> 'callback_reply' AS callback_reply,
+            r.agent_id,
+            r.state AS run_state,
+            r.created_at AS run_created_at
+        FROM awooop_outbound_message m
+        LEFT JOIN awooop_run_state r
+            ON r.project_id = m.project_id
+           AND r.run_id = m.run_id
+        WHERE {where_sql}
+        ORDER BY COALESCE(m.sent_at, m.queued_at) DESC, m.message_id DESC
+        LIMIT :limit OFFSET :offset
+    """)
+
+    async with get_db_context(project_id or "awoooi") as db:
+        count_result = await db.execute(count_sql, params)
+        total = count_result.scalar_one()
+        rows_result = await db.execute(list_sql, params)
+        rows = list(rows_result.mappings().all())
+
+    return {
+        "items": [_callback_reply_event_item(row) for row in rows],
+        "total": total,
+        "page": page,
+        "per_page": per_page,
+    }
+
+
 def _timeline_item(
     *,
     ts: Any,
@@ -346,6 +464,55 @@ def _outbound_callback_reply(source_envelope: Any) -> dict[str, Any] | None:
     return callback_reply if isinstance(callback_reply, dict) else None
 
 
+def _callback_reply_public_status(callback_reply: dict[str, Any]) -> str:
+    """Map raw Telegram callback reply result into the Operator Console filter value."""
+    raw_status = str(callback_reply.get("status") or "")
+    return {
+        "callback_reply_sent": "sent",
+        "callback_reply_fallback_sent": "fallback_sent",
+        "callback_reply_rescue_sent": "rescue_sent",
+        "callback_reply_failed": "failed",
+    }.get(raw_status, "observed")
+
+
+def _callback_reply_event_item(row: Mapping[str, Any]) -> dict[str, Any]:
+    """Convert one callback reply outbound row into a read-only evidence item."""
+    callback_reply = _as_dict(row.get("callback_reply"))
+    action = str(callback_reply.get("action") or "").strip() or None
+    incident_id = str(callback_reply.get("incident_id") or "").strip() or None
+    project_id = str(row.get("project_id") or "")
+    run_id = row.get("run_id")
+    status_value = _callback_reply_public_status(callback_reply)
+    event_at = row.get("sent_at") or row.get("queued_at")
+
+    return {
+        "message_id": row.get("message_id"),
+        "run_id": run_id,
+        "project_id": project_id,
+        "status": status_value,
+        "needs_human": status_value == "failed",
+        "action": action,
+        "incident_id": incident_id,
+        "event_at": event_at,
+        "channel_type": row.get("channel_type"),
+        "message_type": row.get("message_type"),
+        "send_status": row.get("send_status"),
+        "send_error": row.get("send_error"),
+        "provider_message_id": row.get("provider_message_id"),
+        "triggered_by_state": row.get("triggered_by_state"),
+        "content_preview": row.get("content_preview"),
+        "run_state": row.get("run_state"),
+        "agent_id": row.get("agent_id"),
+        "run_created_at": row.get("run_created_at"),
+        "callback_reply": callback_reply,
+        "run_detail_href": (
+            f"/awooop/runs/{run_id}?project_id={project_id}"
+            if run_id and project_id
+            else None
+        ),
+    }
+
+
 def _outbound_timeline_status(
     send_status: str,
     callback_reply: dict[str, Any] | None,
@@ -444,12 +611,7 @@ def _run_callback_reply_summary(
     ]
     failed = statuses.count("callback_reply_failed")
     latest_status = str(latest_callback.get("status") or "")
-    summary_status = {
-        "callback_reply_sent": "sent",
-        "callback_reply_fallback_sent": "fallback_sent",
-        "callback_reply_rescue_sent": "rescue_sent",
-        "callback_reply_failed": "failed",
-    }.get(latest_status, "observed")
+    summary_status = _callback_reply_public_status(latest_callback)
 
     return {
         "schema_version": "awooop_run_callback_reply_summary_v1",
@@ -769,6 +931,20 @@ def _validate_callback_reply_status_filter(value: str | None) -> None:
         )
 
 
+def _validate_callback_reply_action_filter(value: str | None) -> str | None:
+    if value is None:
+        return None
+    normalized = value.strip().lower()
+    if not normalized:
+        return None
+    if not _CALLBACK_REPLY_ACTION_RE.fullmatch(normalized):
+        raise HTTPException(
+            status_code=422,
+            detail="callback action 格式錯誤，僅允許 a-z、0-9、底線、冒號與短橫線",
+        )
+    return normalized
+
+
 def _validate_incident_id_filter(value: str | None) -> None:
     if value is None:
         return