Allow webhook retries after downstream delivery failure

gateway/platforms/webhook.py

@@ -56,6 +56,7 @@
     BasePlatformAdapter,
     MessageEvent,
     MessageType,
+    ProcessingOutcome,
     SendResult,
 )
 from gateway.platforms.webhook_filters import (
@@ -362,6 +363,17 @@ def _record_delivery_id(self, delivery_id: str, now: float) -> bool:
             self._prune_seen_deliveries(now)
         return True
 
+    def _forget_delivery_id(self, delivery_id: str, session_chat_id: str | None = None) -> None:
+        """Release a failed delivery claim so provider retries can run again."""
+        self._seen_deliveries.pop(delivery_id, None)
+        if not session_chat_id:
+            return
+        self._delivery_info.pop(session_chat_id, None)
+        self._delivery_info_created.pop(session_chat_id, None)
+        self._delivery_info_order = deque(
+            item for item in self._delivery_info_order if item[1] != session_chat_id
+        )
+
     async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
         return {"name": chat_id, "type": "webhook"}
 
@@ -708,6 +720,7 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                     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,
@@ -731,6 +744,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,
@@ -801,7 +815,7 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         )
 
     async def on_processing_complete(
-        self, event: "MessageEvent", outcome: Any
+        self, event: "MessageEvent", outcome: ProcessingOutcome
     ) -> None:
         """Close the per-delivery webhook session once its run finishes.
 
@@ -822,6 +836,8 @@ async def on_processing_complete(
         ``end_session()`` is first-reason-wins and no-ops on an already-ended
         row, so this never clobbers a ``compression``/``agent_close`` reason.
         """
+        if outcome is ProcessingOutcome.FAILURE and event.message_id:
+            self._forget_delivery_id(event.message_id, event.source.chat_id)
         await self._end_webhook_session(event, event.source.chat_id)
 
     async def _end_webhook_session(

tests/gateway/test_webhook_adapter.py

@@ -1049,6 +1049,33 @@ async def test_svix_id_used_as_delivery_id_for_deduplication(self):
             assert data["status"] == "duplicate"
             assert data["delivery_id"] == "msg_duplicate"
 
+    @pytest.mark.asyncio
+    async def test_failed_agent_delivery_allows_retry(self):
+        """A failed real background webhook run should release the delivery ID."""
+        routes = {"idem": {"secret": _INSECURE_NO_AUTH, "prompt": "test"}}
+        adapter = _make_adapter(routes=routes)
+
+        async def _raise_once(*args, **kwargs):
+            raise RuntimeError("temporary agent failure")
+
+        adapter._message_handler = AsyncMock(side_effect=_raise_once)
+
+        app = _create_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            headers = {"X-GitHub-Delivery": "delivery-failure-1"}
+
+            resp1 = await cli.post("/webhooks/idem", json={"a": 1}, headers=headers)
+            assert resp1.status == 202
+            for _ in range(20):
+                if "delivery-failure-1" not in adapter._seen_deliveries:
+                    break
+                await asyncio.sleep(0.05)
+            assert "delivery-failure-1" not in adapter._seen_deliveries
+
+            adapter._message_handler = AsyncMock(return_value=None)
+            resp2 = await cli.post("/webhooks/idem", json={"a": 1}, headers=headers)
+            assert resp2.status == 202
+
 
 # ===================================================================
 # Rate limiting

tests/gateway/test_webhook_deliver_only.py

@@ -399,6 +399,84 @@ async def test_idempotency_still_applies(self):
         # Target was called exactly once
         assert mock_target.send.await_count == 1
 
+    @pytest.mark.asyncio
+    async def test_failed_delivery_can_retry_with_same_delivery_id(self):
+        """A failed direct delivery should not poison the retry cache."""
+        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"}
+
+            r1 = await cli.post("/webhooks/r", json={}, headers=headers)
+            assert r1.status == 502
+
+            r2 = await cli.post("/webhooks/r", json={}, headers=headers)
+            assert r2.status == 200
+            data = await r2.json()
+            assert data["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 idempotency claim blocks concurrent direct 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)
+
+            second = await cli.post("/webhooks/r", json={}, headers=headers)
+            assert second.status == 200
+            second_data = await second.json()
+            assert second_data["status"] == "duplicate"
+
+            release.set()
+            first_resp = await first
+            assert first_resp.status == 200
+            first_data = await first_resp.json()
+            assert first_data["status"] == "delivered"
+
+        assert mock_target.send.await_count == 1
+
     @pytest.mark.asyncio
     async def test_rate_limit_still_applies(self):
         """Route-level rate limit caps deliver_only POSTs too."""