From a0a5e850e1ce16ea515c75b0bcabcd9001aab31d Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 22:18:34 +0800 Subject: [PATCH] fix(agent99): reconcile telegram lifecycle ack --- .../services/agent99_telegram_lifecycle.py | 61 +++++++--- .../test_agent99_telegram_lifecycle_api.py | 105 ++++++++++++++++++ 2 files changed, 152 insertions(+), 14 deletions(-) diff --git a/apps/api/src/services/agent99_telegram_lifecycle.py b/apps/api/src/services/agent99_telegram_lifecycle.py index 898d97c49..8df6831e0 100644 --- a/apps/api/src/services/agent99_telegram_lifecycle.py +++ b/apps/api/src/services/agent99_telegram_lifecycle.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import html from typing import Any @@ -24,6 +25,8 @@ _SEVERITY_LABELS = { "info": "資訊 / INFO", } +_DURABLE_RECONCILE_DELAYS_SECONDS = (0.25, 0.5, 1.0) + def _format_lifecycle_card(payload: Agent99TelegramLifecycleRequest) -> str: def esc(value: object) -> str: @@ -92,21 +95,53 @@ async def deliver_agent99_telegram_lifecycle( """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 - response = await get_telegram_gateway().send_agent99_lifecycle_receipt( - delivery_id=payload.delivery_id, - incident_id=payload.incident_id, - state_key=payload.state_key, - text=( + 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), + "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 = ( @@ -152,10 +187,8 @@ async def deliver_agent99_telegram_lifecycle( "delivery_status": delivery_status or "unknown", "durable_outbound_acknowledged": durable_ack, "destination_binding_verified": destination_verified, - "provider_send_performed": bool( - isinstance(response, dict) - and response.get("_awooop_provider_send_performed") is True - ), + "provider_send_performed": provider_send_performed, + "durable_reconciliation_attempts": reconciliation_attempts, "visual_requested": visual_requested, "visual_sent": visual_sent, "visual_delivery_status": ( diff --git a/apps/api/tests/test_agent99_telegram_lifecycle_api.py b/apps/api/tests/test_agent99_telegram_lifecycle_api.py index a1cce3ec0..6800fca14 100644 --- a/apps/api/tests/test_agent99_telegram_lifecycle_api.py +++ b/apps/api/tests/test_agent99_telegram_lifecycle_api.py @@ -257,6 +257,111 @@ async def test_lifecycle_service_uses_durable_canonical_sender(monkeypatch) -> N assert "raw_log" not in calls[0]["text"] +@pytest.mark.asyncio +async def test_lifecycle_reconciles_pending_send_without_duplicate_provider_message( + monkeypatch, +) -> None: + calls: list[dict] = [] + sleeps: list[float] = [] + responses = [ + { + "ok": True, + "result": {"message_id": 9126}, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "pending_unknown", + "_awooop_provider_send_performed": True, + "_awooop_visual_delivery_verified": True, + "_awooop_delivery_context": { + "destination_binding_verified": True, + }, + }, + { + "ok": True, + "result": {"message_id": 9126}, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "sent_reused", + "_awooop_provider_send_performed": False, + "_awooop_visual_delivery_verified": True, + }, + ] + + class Gateway: + async def send_agent99_lifecycle_receipt(self, **kwargs): + calls.append(kwargs) + return responses.pop(0) + + async def no_sleep(delay: float) -> None: + sleeps.append(delay) + + monkeypatch.setattr( + lifecycle_service, + "get_telegram_gateway", + lambda: Gateway(), + ) + monkeypatch.setattr(lifecycle_service.asyncio, "sleep", no_sleep) + request = Agent99TelegramLifecycleRequest.model_validate( + {**payload(), "visual": visual_payload()} + ) + + receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request) + + assert receipt["ok"] is True + assert receipt["delivery_status"] == "sent_reused" + assert receipt["durable_outbound_acknowledged"] is True + assert receipt["provider_message_id"] == "9126" + assert receipt["provider_send_performed"] is True + assert receipt["durable_reconciliation_attempts"] == 1 + assert receipt["visual_sent"] is True + assert sleeps == [0.25] + assert len(calls) == 2 + assert calls[0] == calls[1] + assert calls[0]["delivery_id"] == payload()["delivery_id"] + + +@pytest.mark.asyncio +async def test_lifecycle_reconciliation_fails_closed_after_bounded_readbacks( + monkeypatch, +) -> None: + calls: list[dict] = [] + sleeps: list[float] = [] + + class Gateway: + async def send_agent99_lifecycle_receipt(self, **kwargs): + calls.append(kwargs) + return { + "ok": True, + "result": {"message_id": 9127}, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "pending_unknown", + "_awooop_provider_send_performed": len(calls) == 1, + "_awooop_delivery_context": { + "destination_binding_verified": True, + }, + } + + async def no_sleep(delay: float) -> None: + sleeps.append(delay) + + monkeypatch.setattr( + lifecycle_service, + "get_telegram_gateway", + lambda: Gateway(), + ) + monkeypatch.setattr(lifecycle_service.asyncio, "sleep", no_sleep) + request = Agent99TelegramLifecycleRequest.model_validate(payload()) + + receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request) + + assert receipt["ok"] is False + assert receipt["delivery_status"] == "pending_unknown" + assert receipt["durable_outbound_acknowledged"] is False + assert receipt["provider_send_performed"] is True + assert receipt["durable_reconciliation_attempts"] == 3 + assert sleeps == [0.25, 0.5, 1.0] + assert len(calls) == 4 + assert all(call == calls[0] for call in calls) + + @pytest.mark.asyncio async def test_visual_lifecycle_uses_one_durable_send_photo_receipt(monkeypatch) -> None: calls: list[dict] = []