feat(awooop): persist inbound source envelopes

apps/api/src/services/awooop_truth_chain_service.py

@@ -1177,6 +1177,7 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
             if incident_fingerprints
             else ""
         )
+        fingerprint_value = incident_fingerprints[0] if incident_fingerprints else ""
         inbound_rows = await _fetch_all(
             db,
             """
@@ -1192,6 +1193,9 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
                 content_type,
                 content_hash,
                 content_preview,
+                content_redacted,
+                redaction_version,
+                source_envelope,
                 attachment_sha256,
                 is_duplicate,
                 provider_ts,
@@ -1202,11 +1206,18 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
                 run_id::text = :source_id
                 OR provider_event_id = :source_id
                 OR content_preview ILIKE :source_needle
+                OR coalesce(source_envelope #> '{source_refs,event_ids}', '[]'::jsonb) ? :source_id
+                OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id
+                OR coalesce(source_envelope #> '{source_refs,approval_ids}', '[]'::jsonb) ? :source_id
+                OR coalesce(source_envelope #> '{source_refs,alert_ids}', '[]'::jsonb) ? :source_id
+                OR coalesce(source_envelope #> '{source_refs,sentry_issue_ids}', '[]'::jsonb) ? :source_id
+                OR coalesce(source_envelope #> '{source_refs,signoz_alerts}', '[]'::jsonb) ? :source_id
                 OR (
                   :fingerprint_needle != ''
                   AND (
                     provider_event_id ILIKE :fingerprint_needle
                     OR content_preview ILIKE :fingerprint_needle
+                    OR coalesce(source_envelope #> '{source_refs,fingerprints}', '[]'::jsonb) ? :fingerprint_value
                   )
                 )
               )
@@ -1218,6 +1229,7 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
                 "project_id": project_id,
                 "source_needle": f"%{source_id}%",
                 "fingerprint_needle": fingerprint_needle,
+                "fingerprint_value": fingerprint_value,
                 "limit": _MAX_ROWS,
             },
         )
@@ -1312,7 +1324,14 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
         "project_id": project_id,
         "source_id": source_id,
         "source_type": source_type,
-        "found": incident is not None or drift is not None or bool(runs) or bool(gateway_mcp_rows),
+        "found": (
+            incident is not None
+            or drift is not None
+            or bool(runs)
+            or bool(gateway_mcp_rows)
+            or bool(inbound_rows)
+            or bool(outbound_rows)
+        ),
         "truth_status": truth_status,
         "linked_ids": {
             "incident_id": incident.get("incident_id") if incident else None,
@@ -1321,6 +1340,7 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
             "drift_report_id": drift.get("report_id") if drift else None,
             "operation_ids": _operation_ids(automation_ops),
             "auto_repair_execution_ids": _auto_repair_ids(auto_repair_executions),
+            "conversation_event_ids": [row["event_id"] for row in inbound_rows],
         },
         "incident": incident,
         "drift": {

apps/api/src/services/channel_hub.py

@@ -30,6 +30,7 @@ import asyncio
 import hashlib
 import html
 import json
+import re
 from datetime import UTC, datetime
 from typing import Any
 from uuid import NAMESPACE_URL, UUID, uuid5
@@ -46,7 +47,9 @@ logger = structlog.get_logger(__name__)
 
 # Progressive Feedback Policy：等待超過此秒數才發 interim 訊息
 _INTERIM_WAIT_SECONDS = 30
+_INBOUND_REDACTION_VERSION = "audit_sink_v1"
 _OUTBOUND_REDACTION_VERSION = "audit_sink_v1"
+_INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{6}\b")
 
 
 def _db_timestamp_now() -> datetime:
@@ -54,6 +57,75 @@ def _db_timestamp_now() -> datetime:
     return datetime.now(UTC).replace(tzinfo=None)
 
 
+def _compact_unique(values: list[str | None], *, limit: int = 20) -> list[str]:
+    """Return stable non-empty values without leaking duplicate source refs."""
+    return sorted({str(value).strip() for value in values if str(value or "").strip()})[:limit]
+
+
+def build_inbound_source_envelope(
+    *,
+    provider: str,
+    stage: str,
+    provider_event_id: str,
+    raw_event_id: str | None = None,
+    raw_content: str | None = None,
+    alertname: str | None = None,
+    severity: str | None = None,
+    namespace: str | None = None,
+    target_resource: str | None = None,
+    fingerprint: str | None = None,
+    incident_id: str | None = None,
+    approval_id: str | None = None,
+    source_url: str | None = None,
+    labels: dict[str, Any] | None = None,
+    annotations: dict[str, Any] | None = None,
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build a redaction-friendly inbound replay envelope for truth-chain use."""
+    content_sha256 = hashlib.sha256(raw_content.encode()).hexdigest() if raw_content else None
+    text_refs = _INCIDENT_ID_RE.findall(raw_content or "")
+    provider_name = str(provider or "unknown").strip().lower() or "unknown"
+    source_refs = {
+        "event_ids": _compact_unique([raw_event_id]),
+        "incident_ids": _compact_unique([incident_id, *text_refs]),
+        "approval_ids": _compact_unique([approval_id]),
+        "alert_ids": _compact_unique([provider_event_id, raw_event_id]),
+        "fingerprints": _compact_unique([fingerprint]),
+        "sentry_issue_ids": _compact_unique(
+            [raw_event_id, provider_event_id] if provider_name == "sentry" else []
+        ),
+        "signoz_alerts": _compact_unique(
+            [raw_event_id, alertname] if provider_name == "signoz" else []
+        ),
+    }
+    envelope: dict[str, Any] = {
+        "schema_version": "inbound_source_envelope_v1",
+        "redaction_version": _INBOUND_REDACTION_VERSION,
+        "adapter": f"{provider_name}_webhook",
+        "provider": provider_name,
+        "stage": stage,
+        "provider_event_id": provider_event_id,
+        "source_url": source_url,
+        "content_sha256": content_sha256,
+        "content_length": len(raw_content) if raw_content is not None else 0,
+        "source_refs": source_refs,
+        "log_correlation": {
+            "alertname": alertname,
+            "severity": severity,
+            "namespace": namespace,
+            "target_resource": target_resource,
+            "fingerprint": fingerprint,
+        },
+        "labels": labels or {},
+        "annotations": annotations or {},
+    }
+    if extra:
+        envelope["extra"] = extra
+    sanitized = sanitize(envelope)
+    sanitized["content_sha256"] = content_sha256
+    return sanitized
+
+
 def _input_sha256(input_payload: dict[str, Any] | None) -> str | None:
     """計算 Run input 的穩定 hash，讓 mirror run 也能保留最小完整性證據。"""
     if not input_payload:
@@ -140,6 +212,19 @@ def build_alertmanager_run_id(project_id: str, provider_event_id: str) -> UUID:
     return uuid5(NAMESPACE_URL, f"awooop:alertmanager:{project_id}:{provider_event_id}")
 
 
+def build_external_alert_provider_event_id(provider: str, event_id: str, stage: str) -> str:
+    """建立 Sentry/SignOz 等外部告警 inbound event 的冪等 provider_event_id。"""
+    safe_provider = str(provider).strip().lower()[:32] or "external"
+    safe_event_id = str(event_id).strip()[:96] or "unknown"
+    safe_stage = str(stage).strip()[:32] or "received"
+    return f"{safe_provider}:{safe_stage}:{safe_event_id}"
+
+
+def build_external_alert_run_id(project_id: str, provider_event_id: str) -> UUID:
+    """為外部告警 inbound mirror 建立穩定 shadow run_id。"""
+    return uuid5(NAMESPACE_URL, f"awooop:external-alert:{project_id}:{provider_event_id}")
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # 入站事件記錄
 # ─────────────────────────────────────────────────────────────────────────────
@@ -155,6 +240,7 @@ async def mirror_inbound_event(
     channel_chat_id: str | None = None,
     content_type: str = "text",
     raw_content: str | None = None,
+    source_envelope: dict[str, Any] | None = None,
     attachment_sha256: str | None = None,
     provider_ts: datetime | None = None,
     run_id: UUID | None = None,
@@ -168,12 +254,32 @@ async def mirror_inbound_event(
     """
     content_hash: str | None = None
     content_preview: str | None = None
+    content_redacted: str | None = None
 
     if raw_content is not None:
         content_hash = hashlib.sha256(raw_content.encode()).hexdigest()
         # preview：redact 後截取前 256 字元
-        redacted = _redact_string(raw_content)
-        content_preview = redacted[:256] if len(redacted) > 256 else redacted
+        content_redacted = _redact_string(raw_content)
+        content_preview = (
+            content_redacted[:256] if len(content_redacted) > 256 else content_redacted
+        )
+
+    if source_envelope and source_envelope.get("schema_version") == "inbound_source_envelope_v1":
+        original_content_sha256 = source_envelope.get("content_sha256")
+        envelope = sanitize(source_envelope)
+        envelope.setdefault("redaction_version", _INBOUND_REDACTION_VERSION)
+        envelope["content_sha256"] = content_hash or original_content_sha256
+        envelope.setdefault("content_length", len(raw_content) if raw_content is not None else 0)
+    else:
+        envelope = build_inbound_source_envelope(
+            provider=channel_type,
+            stage="received",
+            provider_event_id=provider_event_id,
+            raw_event_id=provider_event_id,
+            raw_content=raw_content,
+            extra=source_envelope,
+        )
+    source_envelope_json = json.dumps(envelope, ensure_ascii=False, default=str)
 
     result = await db.execute(
         text("""
@@ -181,16 +287,28 @@ async def mirror_inbound_event(
                 project_id, channel_type, provider_event_id,
                 platform_subject_id, channel_user_id, channel_chat_id,
                 run_id, content_type, content_hash, content_preview,
+                content_redacted, redaction_version, source_envelope,
                 attachment_sha256, is_duplicate, provider_ts, received_at
             ) VALUES (
                 :project_id, :channel_type, :provider_event_id,
                 :platform_subject_id, :channel_user_id, :channel_chat_id,
                 :run_id, :content_type, :content_hash, :content_preview,
+                :content_redacted, :redaction_version, CAST(:source_envelope AS jsonb),
                 :attachment_sha256, :is_duplicate, :provider_ts, NOW()
             )
             ON CONFLICT (project_id, channel_type, provider_event_id) DO UPDATE SET
                 is_duplicate = TRUE,
-                run_id = COALESCE(EXCLUDED.run_id, awooop_conversation_event.run_id)
+                run_id = COALESCE(EXCLUDED.run_id, awooop_conversation_event.run_id),
+                content_redacted = COALESCE(
+                    awooop_conversation_event.content_redacted,
+                    EXCLUDED.content_redacted
+                ),
+                redaction_version = EXCLUDED.redaction_version,
+                source_envelope = CASE
+                    WHEN awooop_conversation_event.source_envelope = '{}'::jsonb
+                    THEN EXCLUDED.source_envelope
+                    ELSE awooop_conversation_event.source_envelope
+                END
             RETURNING event_id
         """),
         {
@@ -204,6 +322,9 @@ async def mirror_inbound_event(
             "content_type": content_type,
             "content_hash": content_hash,
             "content_preview": content_preview,
+            "content_redacted": content_redacted,
+            "redaction_version": _INBOUND_REDACTION_VERSION,
+            "source_envelope": source_envelope_json,
             "attachment_sha256": attachment_sha256,
             "is_duplicate": is_duplicate,
             "provider_ts": provider_ts,
@@ -514,6 +635,10 @@ async def record_alertmanager_event(
     approval_id: str | None = None,
     repeat_count: int | None = None,
     is_duplicate: bool = False,
+    source_url: str | None = None,
+    labels: dict[str, Any] | None = None,
+    annotations: dict[str, Any] | None = None,
+    source_extra: dict[str, Any] | None = None,
 ) -> UUID | None:
     """
     將 Alertmanager inbound alert 鏡像到 AwoooP conversation_event。
@@ -546,6 +671,29 @@ async def record_alertmanager_event(
             approval_id=approval_ref,
             repeat_count=repeat_count,
         )
+        source_envelope = build_inbound_source_envelope(
+            provider="alertmanager",
+            stage=stage,
+            provider_event_id=provider_event_id,
+            raw_event_id=alert_id,
+            raw_content=content,
+            alertname=alertname,
+            severity=severity,
+            namespace=namespace,
+            target_resource=target_resource,
+            fingerprint=fingerprint,
+            incident_id=incident_ref,
+            approval_id=approval_ref,
+            source_url=source_url,
+            labels=labels,
+            annotations=annotations,
+            extra={
+                "notification_type": notification_type,
+                "alert_category": alert_category,
+                "repeat_count": repeat_count,
+                **(source_extra or {}),
+            },
+        )
 
         async with get_db_context(project_id) as db:
             run_id = build_alertmanager_run_id(project_id, provider_event_id)
@@ -581,6 +729,7 @@ async def record_alertmanager_event(
                 channel_chat_id=f"alertmanager:{namespace or 'default'}",
                 content_type="text",
                 raw_content=content,
+                source_envelope=source_envelope,
                 provider_ts=_db_timestamp_now(),
                 run_id=run_id,
                 is_duplicate=is_duplicate,
@@ -608,6 +757,129 @@ async def record_alertmanager_event(
         return None
 
 
+async def record_external_alert_event(
+    *,
+    project_id: str,
+    provider: str,
+    event_id: str,
+    stage: str,
+    title: str,
+    severity: str,
+    namespace: str | None = None,
+    target_resource: str | None = None,
+    fingerprint: str | None = None,
+    incident_id: str | None = None,
+    approval_id: str | None = None,
+    source_url: str | None = None,
+    labels: dict[str, Any] | None = None,
+    annotations: dict[str, Any] | None = None,
+    payload: dict[str, Any] | None = None,
+    is_duplicate: bool = False,
+) -> UUID | None:
+    """
+    將 Sentry / SignOz 等非 Alertmanager 告警鏡像到 conversation_event。
+
+    這是 truth-chain 的最低共用入口：只寫 redacted content + source_envelope，
+    不改變原本 webhook 的通知、審批或自動化行為。
+    """
+    provider_name = str(provider or "external").strip().lower() or "external"
+    provider_event_id = build_external_alert_provider_event_id(provider_name, event_id, stage)
+    content = "\n".join([
+        f"{provider_name} inbound {stage}",
+        f"Event ID: {event_id}",
+        f"Title: {title}",
+        f"Severity: {severity}",
+        f"Namespace: {namespace or '-'}",
+        f"Target: {target_resource or '-'}",
+        f"Fingerprint: {fingerprint or '-'}",
+        f"Incident: {incident_id or '-'}",
+        f"Approval: {approval_id or '-'}",
+        f"Source URL: {source_url or '-'}",
+    ])
+    source_envelope = build_inbound_source_envelope(
+        provider=provider_name,
+        stage=stage,
+        provider_event_id=provider_event_id,
+        raw_event_id=event_id,
+        raw_content=content,
+        alertname=title,
+        severity=severity,
+        namespace=namespace,
+        target_resource=target_resource,
+        fingerprint=fingerprint,
+        incident_id=str(incident_id) if incident_id else None,
+        approval_id=str(approval_id) if approval_id else None,
+        source_url=source_url,
+        labels=labels,
+        annotations=annotations,
+        extra={
+            "payload": payload or {},
+        },
+    )
+
+    try:
+        from src.db.base import get_db_context
+
+        async with get_db_context(project_id) as db:
+            run_id = build_external_alert_run_id(project_id, provider_event_id)
+            await ensure_completed_shadow_run(
+                db,
+                project_id=project_id,
+                run_id=run_id,
+                agent_id=f"legacy-{provider_name}-webhook",
+                trigger_type=f"{provider_name}_inbound",
+                trigger_ref=provider_event_id,
+                input_payload={
+                    "provider": provider_name,
+                    "event_id": event_id,
+                    "stage": stage,
+                    "severity": severity,
+                    "namespace": namespace,
+                    "target_resource": target_resource,
+                    "fingerprint": fingerprint,
+                    "incident_id": str(incident_id) if incident_id else None,
+                    "approval_id": str(approval_id) if approval_id else None,
+                },
+            )
+            event_uuid = await mirror_inbound_event(
+                db,
+                project_id=project_id,
+                channel_type="internal",
+                provider_event_id=provider_event_id,
+                platform_subject_id=provider_name,
+                channel_user_id=provider_name,
+                channel_chat_id=f"{provider_name}:{namespace or 'default'}",
+                content_type="text",
+                raw_content=content,
+                source_envelope=source_envelope,
+                provider_ts=_db_timestamp_now(),
+                run_id=run_id,
+                is_duplicate=is_duplicate,
+            )
+
+        logger.info(
+            "external_alert_event_recorded",
+            project_id=project_id,
+            provider=provider_name,
+            event_id=event_id,
+            stage=stage,
+            conversation_event_id=str(event_uuid),
+            incident_id=str(incident_id) if incident_id else None,
+            approval_id=str(approval_id) if approval_id else None,
+        )
+        return event_uuid
+    except Exception as exc:
+        logger.warning(
+            "external_alert_event_record_failed",
+            project_id=project_id,
+            provider=provider_name,
+            event_id=event_id,
+            stage=stage,
+            error=str(exc),
+        )
+        return None
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # 出站訊息記錄
 # ─────────────────────────────────────────────────────────────────────────────