fix(webhook): retry failed direct deliveries

gateway/platforms/webhook.py

@@ -220,6 +220,10 @@ def __init__(self, config: PlatformConfig):
         # Idempotency: TTL cache of recently processed delivery IDs.
         # Prevents duplicate agent runs when webhook providers retry.
         self._seen_deliveries: Dict[str, float] = {}
+        # Direct deliveries stay reserved independently of the TTL until their
+        # downstream send finishes. Successful sends enter _seen_deliveries;
+        # failed or cancelled sends release the reservation for provider retry.
+        self._deliveries_in_flight: set[str] = set()
         self._idempotency_ttl: int = 3600  # 1 hour
         self._seen_deliveries_next_prune_at: float = 0.0
 
@@ -448,18 +452,35 @@ def _record_rate_limit_hit(self, route_name: str, now: float) -> bool:
         window.append(now)
         return True
 
-    def _record_delivery_id(self, delivery_id: str, now: float) -> bool:
-        """Return True when this delivery should be processed."""
+    def _record_delivery_id(
+        self, delivery_id: str, now: float, *, reserve_until_complete: bool = False
+    ) -> bool:
+        """Claim an unseen delivery ID, optionally until downstream completion."""
+        if delivery_id in self._deliveries_in_flight:
+            return False
         seen_at = self._seen_deliveries.get(delivery_id)
         if seen_at is not None and now - seen_at < self._idempotency_ttl:
             return False
         if seen_at is not None:
             self._seen_deliveries.pop(delivery_id, None)
-        self._seen_deliveries[delivery_id] = now
+        if reserve_until_complete:
+            self._deliveries_in_flight.add(delivery_id)
+        else:
+            self._seen_deliveries[delivery_id] = now
         if len(self._seen_deliveries) > max(self._rate_limit * 2, 128):
             self._prune_seen_deliveries(now)
         return True
 
+    def _complete_delivery_id(self, delivery_id: str, now: float) -> None:
+        """Commit a successful direct-delivery claim to the TTL cache."""
+        self._deliveries_in_flight.discard(delivery_id)
+        self._seen_deliveries[delivery_id] = now
+
+    def _forget_delivery_id(self, delivery_id: str) -> None:
+        """Release a failed claim so a provider retry can process it."""
+        self._deliveries_in_flight.discard(delivery_id)
+        self._seen_deliveries.pop(delivery_id, None)
+
     async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
         return {"name": chat_id, "type": "webhook"}
 
@@ -802,7 +823,10 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         # ── Idempotency ─────────────────────────────────────────
         # Skip duplicate deliveries (webhook retries).
         now = time.time()
-        if not self._record_delivery_id(delivery_id, now):
+        deliver_only = bool(route_config.get("deliver_only"))
+        if not self._record_delivery_id(
+            delivery_id, now, reserve_until_complete=deliver_only
+        ):
             logger.info(
                 "[webhook] Skipping duplicate delivery %s", delivery_id
             )
@@ -817,36 +841,41 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         # cron jobs, other agents) that need to push a plain notification
         # to a user's chat with zero LLM cost.  Reuses the same HMAC auth,
         # rate limiting, idempotency, and template rendering as agent mode.
-        if route_config.get("deliver_only"):
-            delivery = {
-                "deliver": route_config.get("deliver", "log"),
-                "deliver_extra": self._render_delivery_extra(
-                    route_config.get("deliver_extra", {}), payload
-                ),
-                "payload": payload,
-            }
-            logger.info(
-                "[webhook] direct-deliver event=%s route=%s target=%s msg_len=%d delivery=%s",
-                event_type,
-                route_name,
-                delivery["deliver"],
-                len(prompt),
-                delivery_id,
-            )
+        if deliver_only:
             try:
+                delivery = {
+                    "deliver": route_config.get("deliver", "log"),
+                    "deliver_extra": self._render_delivery_extra(
+                        route_config.get("deliver_extra", {}), payload
+                    ),
+                    "payload": payload,
+                }
+                logger.info(
+                    "[webhook] direct-deliver event=%s route=%s target=%s msg_len=%d delivery=%s",
+                    event_type,
+                    route_name,
+                    delivery["deliver"],
+                    len(prompt),
+                    delivery_id,
+                )
                 result = await self._direct_deliver(prompt, delivery)
+            except asyncio.CancelledError:
+                self._forget_delivery_id(delivery_id)
+                raise
             except Exception:
                 logger.exception(
                     "[webhook] direct-deliver failed route=%s delivery=%s",
                     route_name,
                     delivery_id,
                 )
+                self._forget_delivery_id(delivery_id)
                 return web.json_response(
                     {"status": "error", "error": "Delivery failed", "delivery_id": delivery_id},
                     status=502,
                 )
 
             if result.success:
+                self._complete_delivery_id(delivery_id, time.time())
                 return web.json_response(
                     {
                         "status": "delivered",
@@ -864,6 +893,7 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                 delivery["deliver"],
                 result.error,
             )
+            self._forget_delivery_id(delivery_id)
             return web.json_response(
                 {"status": "error", "error": "Delivery failed", "delivery_id": delivery_id},
                 status=502,

tests/gateway/test_webhook_deliver_only.py

@@ -19,7 +19,7 @@
 
 import pytest
 from aiohttp import web
-from aiohttp.test_utils import TestClient, TestServer
+from aiohttp.test_utils import TestClient, TestServer, make_mocked_request
 
 from gateway.config import Platform, PlatformConfig
 from gateway.platforms.base import MessageEvent, SendResult
@@ -224,6 +224,149 @@ async def test_hmac_still_enforced(self):
         # Target never called
         mock_target.send.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    async def test_failed_delivery_can_retry_with_same_delivery_id(self):
+        """A failed direct delivery releases its idempotency claim."""
+        routes = {
+            "r": {
+                "secret": _INSECURE_NO_AUTH,
+                "deliver": "telegram",
+                "deliver_only": True,
+                "deliver_extra": {"chat_id": "c-1"},
+                "prompt": "hi",
+            }
+        }
+        adapter = _make_adapter(routes)
+        mock_target = _wire_mock_target(adapter)
+        mock_target.send = AsyncMock(
+            side_effect=[
+                SendResult(success=False, error="temporary outage"),
+                SendResult(success=True),
+            ]
+        )
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            headers = {"X-GitHub-Delivery": "retry-1"}
+            first = await cli.post("/webhooks/r", json={}, headers=headers)
+            second = await cli.post("/webhooks/r", json={}, headers=headers)
+
+            assert first.status == 502
+            assert second.status == 200
+            assert (await second.json())["status"] == "delivered"
+
+        assert mock_target.send.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_delivery_exception_can_retry_with_same_delivery_id(self):
+        routes = {
+            "r": {
+                "secret": _INSECURE_NO_AUTH,
+                "deliver": "telegram",
+                "deliver_only": True,
+                "deliver_extra": {"chat_id": "c-1"},
+                "prompt": "hi",
+            }
+        }
+        adapter = _make_adapter(routes)
+        mock_target = _wire_mock_target(adapter)
+        mock_target.send = AsyncMock(
+            side_effect=[RuntimeError("temporary outage"), SendResult(success=True)]
+        )
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            headers = {"X-GitHub-Delivery": "retry-exception-1"}
+            first = await cli.post("/webhooks/r", json={}, headers=headers)
+            second = await cli.post("/webhooks/r", json={}, headers=headers)
+
+            assert first.status == 502
+            assert second.status == 200
+            assert (await second.json())["status"] == "delivered"
+
+        assert mock_target.send.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_cancelled_delivery_releases_claim_for_retry(self):
+        routes = {
+            "r": {
+                "secret": _INSECURE_NO_AUTH,
+                "deliver": "telegram",
+                "deliver_only": True,
+                "deliver_extra": {"chat_id": "c-1"},
+                "prompt": "hi",
+            }
+        }
+        adapter = _make_adapter(routes)
+        mock_target = _wire_mock_target(adapter)
+        mock_target.send = AsyncMock(
+            side_effect=[asyncio.CancelledError(), SendResult(success=True)]
+        )
+        request = make_mocked_request(
+            "POST",
+            "/webhooks/r",
+            headers={"X-GitHub-Delivery": "retry-cancel-1"},
+            match_info={"route_name": "r"},
+        )
+        request.read = AsyncMock(return_value=b"{}")
+
+        with pytest.raises(asyncio.CancelledError):
+            await adapter._handle_webhook(request)
+
+        retry = await adapter._handle_webhook(request)
+        assert retry.status == 200
+        assert json.loads(retry.text)["status"] == "delivered"
+        assert mock_target.send.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_concurrent_duplicate_delivery_id_sends_once(self):
+        """The in-flight claim still suppresses concurrent duplicates."""
+        routes = {
+            "r": {
+                "secret": _INSECURE_NO_AUTH,
+                "deliver": "telegram",
+                "deliver_only": True,
+                "deliver_extra": {"chat_id": "c-1"},
+                "prompt": "hi",
+            }
+        }
+        adapter = _make_adapter(routes)
+        mock_target = _wire_mock_target(adapter)
+        entered = asyncio.Event()
+        release = asyncio.Event()
+
+        async def _send(*_args, **_kwargs):
+            entered.set()
+            await release.wait()
+            return SendResult(success=True)
+
+        mock_target.send = AsyncMock(side_effect=_send)
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            headers = {"X-GitHub-Delivery": "same-direct-id"}
+            first = asyncio.create_task(cli.post("/webhooks/r", json={}, headers=headers))
+            await asyncio.wait_for(entered.wait(), timeout=1)
+
+            # Even a send that outlives the configured TTL remains reserved
+            # until completion; an expired timestamp must not admit a second.
+            adapter._idempotency_ttl = 0
+            second = await cli.post("/webhooks/r", json={}, headers=headers)
+            assert second.status == 200
+            assert (await second.json())["status"] == "duplicate"
+
+            adapter._idempotency_ttl = 3600
+            release.set()
+            first_response = await first
+            assert first_response.status == 200
+            assert (await first_response.json())["status"] == "delivered"
+
+            retry = await cli.post("/webhooks/r", json={}, headers=headers)
+            assert retry.status == 200
+            assert (await retry.json())["status"] == "duplicate"
+
+        assert mock_target.send.await_count == 1
+
 
 # ===================================================================
 # Unit: _direct_deliver dispatch