"""Canonical, durable Telegram lifecycle delivery for Windows Agent99.""" from __future__ import annotations import asyncio import html from typing import Any from src.models.agent99_completion import Agent99TelegramLifecycleRequest from src.services.telegram_gateway import get_telegram_gateway _LIFECYCLE_LABELS = { "detected": "已偵測 / DETECTED", "investigating": "調查中 / INVESTIGATING", "remediating": "修復中 / REMEDIATING", "verifying": "驗證中 / VERIFYING", "recovered": "已恢復 / RECOVERED", "blocked": "需介入 / ACTION REQUIRED", "unknown": "狀態未知 / UNKNOWN", } _SEVERITY_LABELS = { "critical": "重大 / CRITICAL", "warning": "警告 / WARNING", "info": "資訊 / INFO", } _DURABLE_RECONCILE_DELAYS_SECONDS = (0.25, 0.5, 1.0) def _format_lifecycle_card(payload: Agent99TelegramLifecycleRequest) -> str: def esc(value: object) -> str: return html.escape(str(value or "")) return "\n".join( [ f"Agent99 事件|{esc(payload.title)}", ( f"狀態:{esc(_LIFECYCLE_LABELS[payload.lifecycle])}" f" 等級:{esc(_SEVERITY_LABELS[payload.severity])}" ), "────────────────────", f"資產 {esc(payload.target)}", f"影響 {esc(payload.impact)}", f"AI 處置 {esc(payload.action)}", f"驗證 {esc(payload.verification)}", "────────────────────", ( f"事件:{esc(payload.incident_id)} " f"發生:{payload.occurrences} 次 耗時:{esc(payload.duration_text)}" ), f"下一次更新:{esc(payload.next_update)}", f"時間:{esc(payload.occurred_at.isoformat())}", ] )[:4096] def _format_lifecycle_caption(payload: Agent99TelegramLifecycleRequest) -> str: """Keep the visual card caption scannable on mobile Telegram clients.""" def esc(value: object, limit: int) -> str: rendered: list[str] = [] rendered_length = 0 truncated = False for character in str(value or ""): encoded = html.escape(character) if rendered_length + len(encoded) > limit - 3: truncated = True break rendered.append(encoded) rendered_length += len(encoded) return "".join(rendered) + ("..." if truncated else "") return "\n".join( [ ( f"{esc(_LIFECYCLE_LABELS[payload.lifecycle], 48)}|" f"{esc(_SEVERITY_LABELS[payload.severity], 32)}" ), f"{esc(payload.title, 100)}", f"資產:{esc(payload.target, 80)}", f"Agent99:{esc(payload.action, 220)}", f"驗證:{esc(payload.verification, 180)}", ( f"事件 {esc(payload.incident_id, 180)}|" f"發生 {payload.occurrences} 次|{esc(payload.duration_text, 60)}" ), ] ) async def deliver_agent99_telegram_lifecycle( payload: Agent99TelegramLifecycleRequest, ) -> dict[str, Any]: """Send once through the registered AWOOOI SRE route and verify its ack.""" visual_png = payload.visual.png_bytes() if payload.visual is not None else None gateway = get_telegram_gateway() send_kwargs = { "delivery_id": payload.delivery_id, "incident_id": payload.incident_id, "state_key": payload.state_key, "text": ( _format_lifecycle_caption(payload) if visual_png is not None else _format_lifecycle_card(payload) ), "priority": "P0" if payload.severity == "critical" else "P1", "project_id": payload.project_id, "reply_to_message_id": payload.reply_to_message_id, "visual_png": visual_png, "visual_sha256": ( payload.visual.sha256 if payload.visual is not None else None ), } response = await gateway.send_agent99_lifecycle_receipt(**send_kwargs) provider_send_performed = bool( isinstance(response, dict) and response.get("_awooop_provider_send_performed") is True ) reconciliation_attempts = 0 # A provider send can finish before the durable outbox finalizer becomes # readable. Re-entering with the same delivery identity only re-reads the # send-once reservation; it cannot send a second provider message. for delay_seconds in _DURABLE_RECONCILE_DELAYS_SECONDS: delivery_status = ( str(response.get("_awooop_delivery_status") or "") if isinstance(response, dict) else "" ) if delivery_status != "pending_unknown": break await asyncio.sleep(delay_seconds) reconciliation_attempts += 1 response = await gateway.send_agent99_lifecycle_receipt(**send_kwargs) provider_send_performed = bool( provider_send_performed or ( isinstance(response, dict) and response.get("_awooop_provider_send_performed") is True ) ) result = response.get("result") if isinstance(response, dict) else None message_id = str(result.get("message_id") or "") if isinstance(result, dict) else "" delivery_context = ( response.get("_awooop_delivery_context") if isinstance(response, dict) and isinstance(response.get("_awooop_delivery_context"), dict) else {} ) delivery_status = ( str(response.get("_awooop_delivery_status") or "") if isinstance(response, dict) else "" ) durable_ack = bool( isinstance(response, dict) and response.get("_awooop_outbound_mirror_acknowledged") is True ) destination_verified = bool( delivery_status == "sent_reused" or delivery_context.get("destination_binding_verified") is True ) ok = bool( durable_ack and destination_verified and message_id and delivery_status in {"sent", "sent_reused"} ) visual_requested = payload.visual is not None visual_sent = bool( ok and visual_requested and isinstance(response, dict) and response.get("_awooop_visual_delivery_verified") is True ) return { "schema_version": "agent99_telegram_lifecycle_receipt_v1", "ok": ok, "delivery_id": payload.delivery_id, "incident_id": payload.incident_id, "event_type": payload.event_type, "lifecycle": payload.lifecycle, "provider_message_id": message_id or None, "delivery_status": delivery_status or "unknown", "durable_outbound_acknowledged": durable_ack, "destination_binding_verified": destination_verified, "provider_send_performed": provider_send_performed, "durable_reconciliation_attempts": reconciliation_attempts, "visual_requested": visual_requested, "visual_sent": visual_sent, "visual_delivery_status": ( delivery_status if visual_requested else "not_requested" ), "visual_sha256": ( payload.visual.sha256 if payload.visual is not None else None ), "stores_secret": False, "raw_response_stored": False, "raw_visual_stored": False, }