feat(awooop): trace non-incident action cards

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

@@ -116,9 +116,11 @@ class CallbackReplyAuditSummary(BaseModel):
     outbound_total: int
     outbound_source_envelope_total: int
     outbound_source_refs_total: int
+    outbound_trace_ref_total: int = 0
     outbound_incident_ref_total: int
     outbound_reply_markup_total: int = 0
     outbound_reply_markup_missing_incident_ref_total: int = 0
+    outbound_reply_markup_missing_trace_ref_total: int = 0
     outbound_reply_markup_missing_incident_ref_top_prefixes: list[
         OutboundReplyMarkupGapPrefix
     ] = Field(default_factory=list)

apps/api/src/services/platform_operator_service.py

@@ -473,6 +473,33 @@ async def _fetch_callback_reply_audit_summary(
     """Summarize Telegram outbound mirror and callback evidence capture coverage."""
     result = await db.execute(
         text("""
+            WITH outbound AS (
+                SELECT
+                    m.*,
+                    EXISTS (
+                        SELECT 1
+                        FROM jsonb_each(
+                            CASE
+                                WHEN jsonb_typeof(
+                                    COALESCE(
+                                        m.source_envelope -> 'source_refs',
+                                        '{}'::jsonb
+                                    )
+                                ) = 'object'
+                                THEN COALESCE(
+                                    m.source_envelope -> 'source_refs',
+                                    '{}'::jsonb
+                                )
+                                ELSE '{}'::jsonb
+                            END
+                        ) AS refs(key, value)
+                        WHERE jsonb_typeof(refs.value) = 'array'
+                          AND refs.value <> '[]'::jsonb
+                    ) AS has_trace_ref
+                FROM awooop_outbound_message m
+                WHERE m.project_id = :project_id
+                  AND m.channel_type = 'telegram'
+            )
             SELECT
                 COUNT(*) AS outbound_total,
                 COUNT(*) FILTER (
@@ -481,6 +508,9 @@ async def _fetch_callback_reply_audit_summary(
                 COUNT(*) FILTER (
                     WHERE source_envelope ? 'source_refs'
                 ) AS outbound_source_refs_total,
+                COUNT(*) FILTER (
+                    WHERE has_trace_ref
+                ) AS outbound_trace_ref_total,
                 COUNT(*) FILTER (
                     WHERE COALESCE(
                         source_envelope #> '{source_refs,incident_ids}',
@@ -497,6 +527,10 @@ async def _fetch_callback_reply_audit_summary(
                         '[]'::jsonb
                       ) = '[]'::jsonb
                 ) AS outbound_reply_markup_missing_incident_ref_total,
+                COUNT(*) FILTER (
+                    WHERE source_envelope #>> '{reply_markup,present}' = 'true'
+                      AND NOT has_trace_ref
+                ) AS outbound_reply_markup_missing_trace_ref_total,
                 COALESCE((
                     SELECT jsonb_agg(
                         jsonb_build_object(
@@ -525,10 +559,8 @@ async def _fetch_callback_reply_audit_summary(
                             ) AS recent_24h_total,
                             MIN(COALESCE(sent_at, queued_at)) AS first_sent_at,
                             MAX(COALESCE(sent_at, queued_at)) AS last_sent_at
-                        FROM awooop_outbound_message
-                        WHERE project_id = :project_id
-                          AND channel_type = 'telegram'
-                          AND source_envelope #>> '{reply_markup,present}' = 'true'
+                        FROM outbound
+                        WHERE source_envelope #>> '{reply_markup,present}' = 'true'
                           AND COALESCE(
                             source_envelope #> '{source_refs,incident_ids}',
                             '[]'::jsonb
@@ -604,9 +636,7 @@ async def _fetch_callback_reply_audit_summary(
                 MAX(COALESCE(sent_at, queued_at)) FILTER (
                     WHERE source_envelope ? 'callback_reply'
                 ) AS latest_callback_at
-            FROM awooop_outbound_message
-            WHERE project_id = :project_id
-              AND channel_type = 'telegram'
+            FROM outbound
         """),
         {"project_id": project_id},
     )
@@ -661,6 +691,7 @@ def _callback_reply_audit_summary_from_row(
         "outbound_source_refs_total": _safe_int(
             row.get("outbound_source_refs_total")
         ),
+        "outbound_trace_ref_total": _safe_int(row.get("outbound_trace_ref_total")),
         "outbound_incident_ref_total": outbound_incident_refs,
         "outbound_reply_markup_total": _safe_int(
             row.get("outbound_reply_markup_total")
@@ -668,6 +699,9 @@ def _callback_reply_audit_summary_from_row(
         "outbound_reply_markup_missing_incident_ref_total": _safe_int(
             row.get("outbound_reply_markup_missing_incident_ref_total")
         ),
+        "outbound_reply_markup_missing_trace_ref_total": _safe_int(
+            row.get("outbound_reply_markup_missing_trace_ref_total")
+        ),
         "outbound_reply_markup_missing_incident_ref_top_prefixes": (
             top_missing_prefixes
         ),

apps/api/src/services/telegram_gateway.py

@@ -70,6 +70,10 @@ POLLING_LEADER_WATCH = 30  # seconds - 非 Leader Pod 每 30s 嘗試接管
 logger = structlog.get_logger(__name__)
 _TELEGRAM_BOT_URL_RE = re.compile(r"(api\.telegram\.org/bot)[^/\s]+")
 _INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b")
+_META_EVENT_ID_RE = re.compile(r"\bMETA-\d{14}\b")
+_AI_ADVISORY_CALLBACK_RE = re.compile(
+    r"^ai_advisory_(?:handled|snooze|view|produce_cmd):([^:]+):(.+)$"
+)
 _CODE_REF_RE = re.compile(r"<code>([0-9a-f]{7,12})</code>", re.IGNORECASE)
 _TELEGRAM_HTML_CHUNK_LIMIT = 3600
 _AWOOOP_WEB_BASE_URL = "https://awoooi.wooo.work"
@@ -969,6 +973,47 @@ def _reply_markup_incident_ids(payload: dict) -> list[str]:
     return sorted(incident_ids)
 
 
+def _append_source_ref(refs: dict[str, list[str]], key: str, value: object) -> None:
+    text = str(value or "").strip()
+    if not text:
+        return
+
+    values = refs.setdefault(key, [])
+    if text not in values:
+        values.append(text)
+
+
+def _reply_markup_domain_source_refs(payload: dict) -> dict[str, list[str]]:
+    """Extract non-secret domain refs from known button schemas."""
+    reply_markup = payload.get("reply_markup")
+    if not isinstance(reply_markup, dict):
+        return {}
+
+    refs: dict[str, list[str]] = {}
+    for row in reply_markup.get("inline_keyboard") or []:
+        if not isinstance(row, list):
+            continue
+        for button in row:
+            if not isinstance(button, dict):
+                continue
+            callback_data = str(button.get("callback_data") or "")
+            match = _AI_ADVISORY_CALLBACK_RE.match(callback_data)
+            if not match:
+                continue
+            advisory_type = match.group(1).strip()
+            advisory_id = match.group(2).strip()
+            if not advisory_type or not advisory_id:
+                continue
+            advisory_ref = f"{advisory_type}:{advisory_id}"
+            advisory_event = f"ai_advisory:{advisory_ref}"
+            _append_source_ref(refs, "event_ids", advisory_event)
+            _append_source_ref(refs, "advisory_ids", advisory_ref)
+            _append_source_ref(refs, "alert_ids", advisory_type)
+            _append_source_ref(refs, "fingerprints", advisory_event)
+
+    return {key: values[:20] for key, values in refs.items() if values}
+
+
 def _outbound_source_envelope(method: str, payload: dict) -> dict[str, object]:
     """Build a redaction-friendly source envelope for Channel Hub replay."""
     text = str(payload.get("text") or payload.get("caption") or "")
@@ -976,7 +1021,22 @@ def _outbound_source_envelope(method: str, payload: dict) -> dict[str, object]:
         set(_INCIDENT_ID_RE.findall(text))
         | set(_reply_markup_incident_ids(payload))
     )
+    domain_refs = _reply_markup_domain_source_refs(payload)
+    event_ids = sorted(
+        set(_META_EVENT_ID_RE.findall(text))
+        | set(domain_refs.get("event_ids", []))
+    )
     code_refs = sorted(set(match.group(1) for match in _CODE_REF_RE.finditer(text)))
+    source_refs: dict[str, list[str]] = {
+        "incident_ids": incident_ids[:20],
+        "event_ids": event_ids[:20],
+        "code_refs": code_refs[:20],
+    }
+    for key, values in domain_refs.items():
+        if key == "event_ids":
+            continue
+        source_refs[key] = values[:20]
+
     return {
         "adapter": "legacy_telegram_gateway",
         "method": method,
@@ -986,10 +1046,7 @@ def _outbound_source_envelope(method: str, payload: dict) -> dict[str, object]:
         "disable_web_page_preview": payload.get("disable_web_page_preview"),
         "has_reply_context": _has_reply_context(payload),
         "reply_markup": _reply_markup_summary(payload),
-        "source_refs": {
-            "incident_ids": incident_ids[:20],
-            "code_refs": code_refs[:20],
-        },
+        "source_refs": source_refs,
     }
 
 
@@ -4463,6 +4520,16 @@ class TelegramGateway:
             ]
         }
 
+        source_refs: dict[str, list[str]] = {
+            "approval_ids": [approval_id],
+            "alert_ids": [alertname],
+            "fingerprints": [f"meta:{alert_category}:{alertname}"],
+        }
+        if _INCIDENT_ID_RE.fullmatch(incident_id):
+            source_refs["incident_ids"] = [incident_id]
+        else:
+            source_refs["event_ids"] = [incident_id]
+
         return await self._send_request(
             "sendMessage",
             {
@@ -4470,6 +4537,9 @@ class TelegramGateway:
                 "text": text,
                 "parse_mode": "HTML",
                 "reply_markup": keyboard,
+                _AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY: {
+                    "source_refs": source_refs,
+                },
             },
         )