feat(awooop): show callback reply states in timeline

apps/api/src/services/platform_operator_service.py

@@ -271,10 +271,26 @@ def _outbound_timeline_title(
     channel_type: str,
     message_type: str,
     content_preview: str | None,
+    callback_reply: dict[str, Any] | None = None,
 ) -> str:
     """將 legacy Telegram outbound 分類成 Operator 看得懂的語義標題。"""
     channel = channel_type.upper()
     preview = content_preview or ""
+    if callback_reply:
+        action = str(callback_reply.get("action") or "").strip()
+        status = str(callback_reply.get("status") or "").strip()
+        action_label = {
+            "detail": "詳情",
+            "history": "歷史",
+        }.get(action, action or "Callback")
+        status_label = {
+            "callback_reply_sent": "已送出",
+            "callback_reply_fallback_sent": " fallback 已送出",
+            "callback_reply_rescue_sent": "救援已送出",
+            "callback_reply_failed": "送出失敗",
+        }.get(status, status or "已記錄")
+        return f"{channel}：{action_label}回覆{status_label}"
+
     if "RUNBOOK REVIEW" in preview:
         return f"{channel}：Runbook 待人工審核"
     if "[AWOOOI CI/CD]" in preview or "AWOOOI CI/CD" in preview:
@@ -299,6 +315,73 @@ def _outbound_timeline_title(
     return f"{channel}：{fallback}"
 
 
+def _outbound_callback_reply(source_envelope: Any) -> dict[str, Any] | None:
+    """Extract Telegram callback reply evidence from outbound source envelope."""
+    if not isinstance(source_envelope, dict):
+        return None
+    callback_reply = source_envelope.get("callback_reply")
+    return callback_reply if isinstance(callback_reply, dict) else None
+
+
+def _outbound_timeline_status(
+    send_status: str,
+    callback_reply: dict[str, Any] | None,
+) -> str:
+    """Prefer callback delivery status when the outbound row records one."""
+    if callback_reply:
+        status = callback_reply.get("status")
+        if isinstance(status, str) and status:
+            return status
+    return send_status
+
+
+def _outbound_timeline_summary(
+    *,
+    content_preview: str | None,
+    send_error: str | None,
+    callback_reply: dict[str, Any] | None,
+) -> str | None:
+    """Summarize callback reply state without forcing operators to inspect raw JSON."""
+    if not callback_reply:
+        return content_preview or send_error
+
+    parts = [
+        f"callback={callback_reply.get('action') or '--'}",
+        f"incident={callback_reply.get('incident_id') or '--'}",
+        f"status={callback_reply.get('status') or '--'}",
+    ]
+    parse_mode = callback_reply.get("parse_mode")
+    if parse_mode:
+        parts.append(f"parse_mode={parse_mode}")
+    error = callback_reply.get("error")
+    if error:
+        parts.append(f"error={error}")
+    if content_preview:
+        parts.append(str(content_preview))
+    return " · ".join(parts)
+
+
+def _outbound_timeline_metadata(
+    row: AwoooPOutboundMessage,
+    callback_reply: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Build compact outbound metadata with callback fields first when present."""
+    metadata: dict[str, Any] = {}
+    if callback_reply:
+        metadata.update({
+            "callback_status": callback_reply.get("status"),
+            "callback_action": callback_reply.get("action"),
+            "callback_incident_id": callback_reply.get("incident_id"),
+            "callback_parse_mode": callback_reply.get("parse_mode"),
+        })
+    metadata.update({
+        "message_type": row.message_type,
+        "provider_message_id": row.provider_message_id,
+        "triggered_by_state": row.triggered_by_state,
+    })
+    return metadata
+
+
 def _mcp_gateway_summary_row(row: AwoooPMcpGatewayAudit) -> dict[str, Any]:
     """Convert SQLAlchemy audit rows into the truth-chain summary shape."""
     return {
@@ -971,8 +1054,10 @@ async def get_run_detail(
         for row in inbound_events
     ]
 
-    outbound_items = [
-        {
+    outbound_items = []
+    for row in outbound_messages:
+        callback_reply = _outbound_callback_reply(row.source_envelope)
+        outbound_items.append({
             "message_id": row.message_id,
             "channel_type": row.channel_type,
             "message_type": row.message_type,
@@ -983,9 +1068,8 @@ async def get_run_detail(
             "queued_at": row.queued_at,
             "sent_at": row.sent_at,
             "triggered_by_state": row.triggered_by_state,
-        }
-        for row in outbound_messages
-    ]
+            "callback_reply": callback_reply,
+        })
 
     def _mcp_item(row: AwoooPMcpGatewayAudit) -> dict[str, Any]:
         gate_result = row.gate_result if isinstance(row.gate_result, dict) else {}
@@ -1136,6 +1220,7 @@ async def get_run_detail(
             )
         )
     for row in outbound_messages:
+        callback_reply = _outbound_callback_reply(row.source_envelope)
         timeline.append(
             _timeline_item(
                 ts=row.sent_at or row.queued_at,
@@ -1144,14 +1229,15 @@ async def get_run_detail(
                     row.channel_type,
                     row.message_type,
                     row.content_preview,
+                    callback_reply,
                 ),
-                status=row.send_status,
-                summary=row.content_preview or row.send_error,
-                metadata={
-                    "message_type": row.message_type,
-                    "provider_message_id": row.provider_message_id,
-                    "triggered_by_state": row.triggered_by_state,
-                },
+                status=_outbound_timeline_status(row.send_status, callback_reply),
+                summary=_outbound_timeline_summary(
+                    content_preview=row.content_preview,
+                    send_error=row.send_error,
+                    callback_reply=callback_reply,
+                ),
+                metadata=_outbound_timeline_metadata(row, callback_reply),
             )
         )
     if run.completed_at: