diff --git a/apps/api/src/services/agent99_telegram_lifecycle.py b/apps/api/src/services/agent99_telegram_lifecycle.py index 4bfd7e0dd..c227b9b37 100644 --- a/apps/api/src/services/agent99_telegram_lifecycle.py +++ b/apps/api/src/services/agent99_telegram_lifecycle.py @@ -261,7 +261,10 @@ async def deliver_agent99_telegram_lifecycle( if isinstance(response, dict) else "" ) - if delivery_status != "pending_unknown": + if delivery_status not in { + "pending_unknown", + "provider_sent_durable_receipt_pending", + }: break await asyncio.sleep(delay_seconds) reconciliation_attempts += 1 diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 631c340ba..a92f66dcf 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -6470,13 +6470,19 @@ class TelegramGateway: sent_receipt["destination_binding"] = destination_binding return sent_receipt if existing_row.get("send_status") == "pending": - return { - "status": "reserved", + pending_receipt: dict[str, object] = { + "status": "pending_unknown", "run_id": str(delivery_run_id), "message_id": str(existing_row["message_id"]), "durable_reservation_committed": True, - "reused_pending_reservation": True, + "provider_retry_blocked": True, } + destination_binding = str( + existing_row.get("destination_binding") or "" + ) + if destination_binding: + pending_receipt["destination_binding"] = destination_binding + return pending_receipt return { "status": "pending_unknown", "run_id": str(delivery_run_id), @@ -6593,7 +6599,12 @@ class TelegramGateway: true ), '{canonical_route_receipt,destination_binding}', - to_jsonb(:expected_destination_binding::text), + to_jsonb( + CAST( + :expected_destination_binding + AS text + ) + ), true ) END diff --git a/apps/api/tests/test_agent99_telegram_lifecycle_api.py b/apps/api/tests/test_agent99_telegram_lifecycle_api.py index 461530ca4..172dba1a0 100644 --- a/apps/api/tests/test_agent99_telegram_lifecycle_api.py +++ b/apps/api/tests/test_agent99_telegram_lifecycle_api.py @@ -769,6 +769,67 @@ async def test_lifecycle_reconciles_pending_send_without_duplicate_provider_mess } +@pytest.mark.asyncio +async def test_lifecycle_reconciles_provider_sent_pending_ack_without_resend( + monkeypatch, +) -> None: + send_calls: list[dict] = [] + reconcile_calls: list[dict] = [] + + class Gateway: + async def send_agent99_lifecycle_receipt(self, **kwargs): + send_calls.append(kwargs) + return { + "ok": True, + "result": {"message_id": 9129}, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": ( + "provider_sent_durable_receipt_pending" + ), + "_awooop_provider_send_performed": True, + "_awooop_delivery_context": { + "destination_binding_verified": True, + "destination_binding": "a" * 64, + }, + } + + async def reconcile_agent99_lifecycle_receipt(self, **kwargs): + reconcile_calls.append(kwargs) + return { + "ok": True, + "result": {"message_id": 9129}, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "sent_reused", + "_awooop_provider_send_performed": False, + "_awooop_delivery_context": { + "destination_binding_verified": True, + }, + } + + async def no_sleep(_delay: float) -> None: + return None + + monkeypatch.setattr( + lifecycle_service, + "get_telegram_gateway", + lambda: Gateway(), + ) + monkeypatch.setattr(lifecycle_service.asyncio, "sleep", no_sleep) + + receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle( + Agent99TelegramLifecycleRequest.model_validate(payload()) + ) + + assert receipt["ok"] is True + assert receipt["delivery_status"] == "sent_reused" + assert receipt["provider_send_performed"] is True + assert receipt["durable_reconciliation_attempts"] == 1 + assert len(send_calls) == 1 + assert len(reconcile_calls) == 1 + assert reconcile_calls[0]["expected_provider_message_id"] == "9129" + assert reconcile_calls[0]["expected_destination_binding"] == "a" * 64 + + @pytest.mark.asyncio async def test_lifecycle_follower_only_reads_inflight_durable_delivery( monkeypatch, diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index 852447fcf..945484d01 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -738,7 +738,7 @@ async def test_projection_stays_open_until_durable_telegram_readback( @pytest.mark.asyncio -async def test_controlled_result_pending_reservation_retries_provider_send( +async def test_controlled_result_pending_reservation_blocks_duplicate_provider_send( monkeypatch: pytest.MonkeyPatch, ) -> None: class _Response: @@ -779,11 +779,11 @@ async def test_controlled_result_pending_reservation_retries_provider_send( "message_id": "00000000-0000-0000-0000-000000000202", }, { - "status": "reserved", + "status": "pending_unknown", "run_id": "00000000-0000-0000-0000-000000000201", "message_id": "00000000-0000-0000-0000-000000000202", "durable_reservation_committed": True, - "reused_pending_reservation": True, + "provider_retry_blocked": True, }, ] ) @@ -808,15 +808,13 @@ async def test_controlled_result_pending_reservation_retries_provider_send( _controlled_result_payload(), ) - assert client.post_count == 2 + assert client.post_count == 1 assert first["_awooop_provider_send_performed"] is True assert first["_awooop_outbound_mirror_acknowledged"] is False - assert second["ok"] is True - assert second["_awooop_provider_send_performed"] is True - assert second["_awooop_delivery_status"] == ( - "provider_sent_durable_receipt_pending" - ) - assert finalize.await_count == 2 + assert second["ok"] is False + assert second["_awooop_provider_send_performed"] is False + assert second["_awooop_delivery_status"] == "pending_unknown" + assert finalize.await_count == 1 @pytest.mark.asyncio @@ -925,6 +923,10 @@ async def test_controlled_result_finalize_backfills_missing_destination_binding( assert "{canonical_route_receipt,destination_binding}" in update_sql assert "COALESCE(" in update_sql assert "IN ('', :expected_destination_binding)" in update_sql + assert "CAST(" in update_sql + assert ":expected_destination_binding\n" in update_sql + assert "AS text" in update_sql + assert ":expected_destination_binding::text" not in update_sql assert db.parameters[1]["expected_destination_binding"] == "a" * 64 @@ -1300,6 +1302,8 @@ def test_controlled_result_uses_durable_reservation_and_single_reconcile_entry() assert "digest_window_minutes" in reserve_source assert "notification_suppressed_deduplicated" in reserve_source assert "durable_reservation_committed" in reserve_source + assert "provider_retry_blocked" in reserve_source + assert "reused_pending_reservation" not in reserve_source assert "reservation_readback_failed" not in reserve_source assert "UPDATE awooop_outbound_message" in finalize_source assert "_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS" in finalize_source