feat(awooop): surface reconciliation in incident timeline

apps/api/src/api/v1/incidents.py

@@ -134,6 +134,7 @@ class IncidentTimelineResponse(BaseModel):
     timeline: list[IncidentTimelineStage] = Field(default_factory=list)
     events: list[IncidentTimelineEvent] = Field(default_factory=list)
     ascii_timeline: str
+    reconciliation: dict[str, Any] = Field(default_factory=dict)
 
 
 # =============================================================================

apps/api/src/services/awooop_truth_chain_service.py

@@ -97,7 +97,7 @@ def _operation_ids(automation_ops: list[dict[str, Any]]) -> list[str]:
     return [str(row["op_id"]) for row in automation_ops if row.get("op_id")]
 
 
-def _build_reconciliation(
+def build_incident_reconciliation(
     *,
     incident: dict[str, Any] | None,
     approvals: list[dict[str, Any]],
@@ -694,7 +694,7 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
         legacy_mcp_total=legacy_mcp_summary["total"],
         outbound_visible_total=len(outbound_rows),
     )
-    reconciliation = _build_reconciliation(
+    reconciliation = build_incident_reconciliation(
         incident=incident,
         approvals=approvals,
         evidence_rows=evidence_rows,

apps/api/src/services/incident_timeline_service.py

@@ -27,6 +27,7 @@ from src.db.models import (
     KnowledgeEntryRecord,
     TimelineEvent,
 )
+from src.services.awooop_truth_chain_service import build_incident_reconciliation
 
 logger = structlog.get_logger(__name__)
 
@@ -222,6 +223,29 @@ def _automation_summary(row: Any) -> str | None:
     return row.error
 
 
+def _reconciliation_event(reconciliation: dict[str, Any]) -> dict[str, Any] | None:
+    """Render truth-chain reconciliation into the operator timeline."""
+    if not reconciliation.get("applicable"):
+        return None
+    status = str(reconciliation.get("consistency_status") or "unknown")
+    mismatches = reconciliation.get("mismatches") or []
+    if status == "consistent" and not mismatches:
+        return None
+
+    stage_status = "error" if status == "blocked" else "warning"
+    codes = [str(row.get("code")) for row in mismatches if row.get("code")]
+    description = "; ".join(codes) if codes else None
+    return _event(
+        stage="safe",
+        status=stage_status,
+        title=f"Lifecycle reconciliation: {status}",
+        description=description,
+        actor="truth_chain_reconciliation",
+        source_table="truth_chain",
+        data=reconciliation,
+    )
+
+
 async def _fetch_automation_ops(
     db: Any,
     incident_id: str,
@@ -365,6 +389,49 @@ async def fetch_incident_timeline(incident_id: str) -> dict[str, Any] | None:
         automation_ops = await _fetch_automation_ops(db, incident_id, approval_ids)
 
     events: list[dict[str, Any]] = []
+    reconciliation = build_incident_reconciliation(
+        incident={
+            "incident_id": incident.incident_id,
+            "status": _value(incident.status),
+        },
+        approvals=[
+            {
+                "id": str(approval.id),
+                "status": _value(approval.status),
+                "action": approval.action,
+                "resolved_at": _iso(approval.resolved_at),
+            }
+            for approval in sorted(
+                approvals,
+                key=lambda row: row.created_at or datetime.min,
+                reverse=True,
+            )
+        ],
+        evidence_rows=[
+            {
+                "sensors_attempted": evidence.sensors_attempted,
+                "sensors_succeeded": evidence.sensors_succeeded,
+            }
+            for evidence in evidence_records
+        ],
+        automation_ops=[
+            {
+                "status": op.status,
+                "operation_type": op.operation_type,
+                "op_id": op.op_id,
+            }
+            for op in automation_ops
+        ],
+        timeline_events=[
+            {
+                "event_type": event.event_type,
+                "status": event.status,
+            }
+            for event in raw_timeline
+        ],
+    )
+    if reconciliation_event := _reconciliation_event(reconciliation):
+        events.append(reconciliation_event)
 
     alert_name = incident.alertname
     if not alert_name and incident.signals:
@@ -639,6 +706,7 @@ async def fetch_incident_timeline(incident_id: str) -> dict[str, Any] | None:
         "timeline": stage_list,
         "events": events,
         "ascii_timeline": format_ascii_timeline(stage_list),
+        "reconciliation": reconciliation,
     }
     logger.info(
         "incident_timeline_fetched",

apps/api/src/services/telegram_gateway.py

@@ -5045,6 +5045,24 @@ class TelegramGateway:
                     "🧭 <b>處理歷程</b>",
                     f"<code>{html.escape(timeline['ascii_timeline'])}</code>",
                 ]
+                reconciliation = timeline.get("reconciliation") or {}
+                if reconciliation.get("consistency_status") in {"blocked", "degraded"}:
+                    mismatch_codes = [
+                        str(row.get("code"))
+                        for row in reconciliation.get("mismatches", [])
+                        if row.get("code")
+                    ]
+                    lines += [
+                        "",
+                        "🚦 <b>真相鏈狀態</b>",
+                        f"狀態: <code>{html.escape(str(reconciliation.get('consistency_status')))}</code>",
+                        f"下一步: <code>{html.escape(str(reconciliation.get('operator_next_state')))}</code>",
+                    ]
+                    if mismatch_codes:
+                        lines.append(
+                            "矛盾: "
+                            + html.escape(", ".join(mismatch_codes[:4]))
+                        )
 
             await self.send_notification("\n".join(lines))
 

apps/api/tests/test_awooop_truth_chain_service.py

@@ -8,7 +8,7 @@ from src.services.awooop_ansible_audit_service import (
     build_ansible_truth,
 )
 from src.services.awooop_truth_chain_service import (
-    _build_reconciliation,
+    build_incident_reconciliation,
     _clean_row,
     _truth_status,
 )
@@ -162,7 +162,7 @@ def test_drift_repeat_state_counts_matching_fingerprint_only() -> None:
 
 
 def test_reconciliation_blocks_open_incident_after_no_action_approval() -> None:
-    reconciliation = _build_reconciliation(
+    reconciliation = build_incident_reconciliation(
         incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
         approvals=[
             {
@@ -191,7 +191,7 @@ def test_reconciliation_blocks_open_incident_after_no_action_approval() -> None:
 
 
 def test_reconciliation_marks_consistent_resolved_execution() -> None:
-    reconciliation = _build_reconciliation(
+    reconciliation = build_incident_reconciliation(
         incident={"incident_id": "INC-2", "status": "RESOLVED"},
         approvals=[
             {

apps/api/tests/test_incident_timeline_service.py

@@ -1,4 +1,8 @@
-from src.services.incident_timeline_service import STAGE_DEFS, format_ascii_timeline
+from src.services.incident_timeline_service import (
+    STAGE_DEFS,
+    _reconciliation_event,
+    format_ascii_timeline,
+)
 
 
 def _stages(status_by_stage: dict[str, str]) -> list[dict]:
@@ -23,3 +27,30 @@ def test_format_ascii_timeline_skips_unrecorded_stages() -> None:
 
 def test_format_ascii_timeline_has_empty_fallback() -> None:
     assert format_ascii_timeline(_stages({})) == "webhook:skip > ai:skip > executor:skip"
+
+
+def test_reconciliation_event_marks_safe_stage_failed() -> None:
+    event = _reconciliation_event({
+        "applicable": True,
+        "consistency_status": "blocked",
+        "operator_next_state": "manual_required",
+        "mismatches": [
+            {"code": "approval_approved_without_execution_record"},
+            {"code": "evidence_all_sensors_failed"},
+        ],
+    })
+
+    assert event is not None
+    assert event["stage"] == "safe"
+    assert event["status"] == "error"
+    assert event["title"] == "Lifecycle reconciliation: blocked"
+    assert "approval_approved_without_execution_record" in event["description"]
+    assert event["data"]["operator_next_state"] == "manual_required"
+
+
+def test_reconciliation_event_omits_consistent_state() -> None:
+    assert _reconciliation_event({
+        "applicable": True,
+        "consistency_status": "consistent",
+        "mismatches": [],
+    }) is None