feat(awooop): filter runs by callback reply state

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

@@ -98,7 +98,8 @@ class DecideApprovalResponse(BaseModel):
     response_model=ListRunsResponse,
     summary="列出 Runs",
     description=(
-        "返回 awooop_run_state 記錄，支援 project_id / state / remediation_status / incident_id filter 與分頁。\n\n"
+        "返回 awooop_run_state 記錄，支援 project_id / state / remediation_status / "
+        "callback_reply_status / incident_id filter 與分頁。\n\n"
         "- 按 created_at DESC 排序\n"
         "- 注意：此路徑為 /runs/list 以避免與 runs.py 的 /runs/{run_id} 衝突"
     ),
@@ -110,6 +111,10 @@ async def list_runs(
         None,
         description="AI 證據狀態 filter（no_evidence/mcp_observed/read_only_dry_run/write_observed/blocked/observed）",
     ),
+    callback_reply_status: str | None = Query(
+        None,
+        description="Telegram callback reply 狀態 filter（no_callback/sent/fallback_sent/rescue_sent/failed/observed）",
+    ),
     incident_id: str | None = Query(None, description="關聯 Incident ID filter（可選）"),
     page: int = Query(1, ge=1, description="頁碼，從 1 開始"),
     per_page: int = Query(_DEFAULT_PER_PAGE, ge=1, le=_MAX_PER_PAGE, description="每頁筆數"),
@@ -118,6 +123,7 @@ async def list_runs(
         project_id=project_id,
         state=state,
         remediation_status=remediation_status,
+        callback_reply_status=callback_reply_status,
         incident_id=incident_id,
         page=page,
         per_page=per_page,

apps/api/src/services/platform_operator_service.py

@@ -57,6 +57,14 @@ _REMEDIATION_STATUS_FILTERS = {
     "blocked",
     "observed",
 }
+_CALLBACK_REPLY_STATUS_FILTERS = {
+    "no_callback",
+    "sent",
+    "fallback_sent",
+    "rescue_sent",
+    "failed",
+    "observed",
+}
 
 # =============================================================================
 # Tenants
@@ -146,12 +154,14 @@ async def list_runs(
     project_id: str | None,
     state: str | None,
     remediation_status: str | None,
+    callback_reply_status: str | None,
     incident_id: str | None,
     page: int,
     per_page: int,
 ) -> dict[str, Any]:
-    """列出 runs，支援 project_id、state、remediation_status、incident_id filter 與分頁。"""
+    """列出 runs，支援 project/state/evidence/callback/incident filter 與分頁。"""
     _validate_remediation_status_filter(remediation_status)
+    _validate_callback_reply_status_filter(callback_reply_status)
     _validate_incident_id_filter(incident_id)
 
     async with get_db_context("awoooi") as db:
@@ -162,7 +172,7 @@ async def list_runs(
             stmt = stmt.where(AwoooPRunState.state == state)
 
         offset = (page - 1) * per_page
-        if remediation_status or incident_id:
+        if remediation_status or incident_id or callback_reply_status:
             result = await db.execute(stmt)
             candidate_rows = list(result.scalars().all())
             context_limit = _list_filter_context_limit(len(candidate_rows))
@@ -176,6 +186,10 @@ async def list_runs(
                 inbound_by_run=inbound_by_run,
                 outbound_by_run=outbound_by_run,
             )
+            callback_reply_summaries = {
+                row.run_id: _run_callback_reply_summary(outbound_by_run.get(row.run_id, []))
+                for row in candidate_rows
+            }
             filtered_rows = [
                 row
                 for row in candidate_rows
@@ -187,6 +201,10 @@ async def list_runs(
                     remediation_summaries.get(row.run_id),
                     incident_id,
                 )
+                and _callback_reply_summary_matches_status(
+                    callback_reply_summaries.get(row.run_id),
+                    callback_reply_status,
+                )
             ]
             total = len(filtered_rows)
             rows = filtered_rows[offset : offset + per_page]
@@ -204,6 +222,10 @@ async def list_runs(
                 inbound_by_run=inbound_by_run,
                 outbound_by_run=outbound_by_run,
             )
+            callback_reply_summaries = {
+                row.run_id: _run_callback_reply_summary(outbound_by_run.get(row.run_id, []))
+                for row in rows
+            }
 
     runs = [
         {
@@ -217,9 +239,7 @@ async def list_runs(
             "created_at": r.created_at,
             "timeout_at": r.timeout_at,
             "remediation_summary": remediation_summaries.get(r.run_id),
-            "callback_reply_summary": _run_callback_reply_summary(
-                outbound_by_run.get(r.run_id, [])
-            ),
+            "callback_reply_summary": callback_reply_summaries.get(r.run_id),
         }
         for r in rows
     ]
@@ -738,6 +758,17 @@ def _validate_remediation_status_filter(value: str | None) -> None:
         )
 
 
+def _validate_callback_reply_status_filter(value: str | None) -> None:
+    if value is None:
+        return
+    if value not in _CALLBACK_REPLY_STATUS_FILTERS:
+        allowed = ", ".join(sorted(_CALLBACK_REPLY_STATUS_FILTERS))
+        raise HTTPException(
+            status_code=422,
+            detail=f"callback_reply_status 必須是: {allowed}",
+        )
+
+
 def _validate_incident_id_filter(value: str | None) -> None:
     if value is None:
         return
@@ -758,6 +789,16 @@ def _remediation_summary_matches_status(
     return status_value == remediation_status
 
 
+def _callback_reply_summary_matches_status(
+    summary: dict[str, Any] | None,
+    callback_reply_status: str | None,
+) -> bool:
+    if callback_reply_status is None:
+        return True
+    status_value = str((summary or {}).get("status") or "no_callback")
+    return status_value == callback_reply_status
+
+
 def _remediation_summary_matches_incident_id(
     summary: dict[str, Any] | None,
     incident_id: str | None,