fix: persist serialized webhook consumer lifecycle

contributors/emails/alazhouhang@qq.com

@@ -0,0 +1 @@
+09473ZH

gateway/platforms/webhook.py

@@ -197,6 +197,8 @@ def __init__(self, config: PlatformConfig):
         self._dynamic_routes_mtime: float = 0.0
         self._routes: Dict[str, dict] = dict(self._static_routes)
         self._runner = None
+        self._lifecycle = None
+        self._lifecycle_events = {}
         self._serial_groups: dict[tuple[str, str], str] = {}
         self._delivery_groups: dict[str, tuple[str, str]] = {}
         # Routes already warned about legacy V1 body-only signatures
@@ -739,6 +741,44 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                     {"error": "Cannot parse body"}, status=400
                 )
 
+        # Signed control requests use the same route/profile authentication but
+        # never render a prompt, consume model quota or create a session.
+        if route_config.get("serial_event_key") and payload.get("operation"):
+            ledger = self._delivery_lifecycle()
+            operation = payload["operation"]
+            if operation not in {"status", "cancel"}:
+                return web.json_response({"error": "invalid lifecycle operation"}, status=400)
+            identities = payload["deliveries"]
+            if not isinstance(identities, list) or not 1 <= len(identities) <= 100:
+                return web.json_response({"error": "invalid lifecycle batch"}, status=400)
+            results = []
+            for item in identities:
+                identity, scope = item["identity"], item["scope"]
+                if not all(isinstance(v, str) and v.strip() for v in (identity, scope)):
+                    return web.json_response({"error": "invalid lifecycle identity"}, status=400)
+                row = ledger.get(route_name, identity)
+                if row is not None and row["scope"] != scope:
+                    return web.json_response({"error": "lifecycle scope mismatch"}, status=409)
+                if operation == "cancel":
+                    ledger.cancel(route_name, identity, scope)
+                    # Tombstone first: an in-flight POST can never admit this work.
+                    if (row is not None and row["chat"] in self._lifecycle_events
+                            and not ledger.has_other_work(row["chat"], route_name, identity)):
+                        event = self._lifecycle_events[row["chat"]]
+                        from gateway.session import build_session_key
+                        key = build_session_key(event.source,
+                            group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
+                            thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False))
+                        await self.cancel_session_processing(key)
+                    row = ledger.get(route_name, identity)
+                active = self._serial_groups.get((route_name, scope))
+                results.append({"identity": identity, "serial_key": scope,
+                    "status": "unknown" if row is None else row["status"],
+                    "active_chat_id": None if row is None else row["chat"],
+                    "scope_busy": active is not None,
+                    "consumer_running": row is not None and row["chat"] in self._delivery_groups})
+            return web.json_response({"deliveries": results})
+
         # Check event type filter
         event_type = (
             request.headers.get("X-GitHub-Event", "")
@@ -856,6 +896,13 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                 return web.json_response({"error": "serial key must be nonempty"}, status=400)
             if route_config.get("deliver_only"):
                 return web.json_response({"error": "serial key requires agent delivery"}, status=400)
+            if route_config.get("serial_event_key"):
+                previous = self._delivery_lifecycle().get(route_name, delivery_id)
+                if previous is not None:
+                    if previous["scope"] != serial_value:
+                        return web.json_response({"error": "delivery scope mismatch"}, status=409)
+                    return web.json_response({"status": previous["status"],
+                        "serial_key": serial_value, "active_chat_id": previous["chat"]}, status=200)
             serial_group = (route_name, serial_value)
             active = self._serial_groups.get(serial_group)
             if active is not None:
@@ -883,6 +930,7 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                         steered = self.gateway_runner.steer_webhook(event)
                         if steered:
                             seen.add(identity)
+                            self._delivery_lifecycle().admit(route_name, delivery_id, serial_value, active)
                 return web.json_response(
                     {"status": "busy", "serial_key": serial_value,
                      "active_chat_id": active, "steered": steered}, status=202,
@@ -1021,6 +1069,9 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         if serial_group is not None:
             self._serial_groups[serial_group] = session_chat_id
             self._delivery_groups[session_chat_id] = serial_group
+        if serial_group is not None and route_config.get("serial_event_key"):
+            self._delivery_lifecycle().admit(route_name, delivery_id, serial_group[1], session_chat_id)
+            self._lifecycle_events[session_chat_id] = event
         task = asyncio.create_task(self._dispatch_delivery(event))
         self._background_tasks.add(task)
         task.add_done_callback(self._background_tasks.discard)
@@ -1037,14 +1088,35 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
             status=202,
         )
 
+    def sender_manages_resume(self, chat_id: str) -> bool:
+        parts = chat_id.split(":", 2)
+        return (len(parts) == 3 and parts[0] == "webhook"
+                and bool(self._routes.get(parts[1], {}).get("serial_event_key")))
+
+    def _delivery_lifecycle(self):
+        from hermes_constants import get_hermes_home
+        from gateway.platforms.webhook_lifecycle import WebhookLifecycle
+        if self._lifecycle is None:
+            self._lifecycle = WebhookLifecycle(get_hermes_home() / "webhook-lifecycle.sqlite3")
+        return self._lifecycle
+
     async def _dispatch_delivery(self, event: "MessageEvent") -> None:
         try:
+            if event.source.chat_id in self._lifecycle_events:
+                group = self._delivery_groups[event.source.chat_id]
+                identity = event.source.chat_id.removeprefix(f"webhook:{group[0]}:")
+                if self._delivery_lifecycle().get(group[0], identity)["status"] == "cancelled":
+                    self._release_serial_group(event.source.chat_id)
+                    return
             await self.handle_message(event)
         except BaseException:
             self._release_serial_group(event.source.chat_id)
             raise
 
     def _release_serial_group(self, chat_id: str) -> None:
+        if chat_id in self._lifecycle_events:
+            self._delivery_lifecycle().finish(chat_id)
+            self._lifecycle_events.pop(chat_id)
         self._steered_events.pop(chat_id, None)
         group = self._delivery_groups.pop(chat_id, None)
         if group is not None:

gateway/platforms/webhook_lifecycle.py

@@ -0,0 +1,69 @@
+"""Durable admission receipts for signed, serialized webhook work."""
+
+import sqlite3
+from pathlib import Path
+from uuid import uuid4
+
+
+class WebhookLifecycle:
+    def __init__(self, path: Path):
+        self.boot = str(uuid4())
+        self.db = sqlite3.connect(path)
+        self.db.row_factory = sqlite3.Row
+        self.db.execute("""CREATE TABLE IF NOT EXISTS deliveries (
+            route TEXT NOT NULL, identity TEXT NOT NULL, scope TEXT NOT NULL,
+            chat TEXT NOT NULL, status TEXT NOT NULL, boot TEXT NOT NULL,
+            PRIMARY KEY(route,identity))""")
+        self.db.execute(
+            "CREATE INDEX IF NOT EXISTS deliveries_chat ON deliveries(chat,status,boot)"
+        )
+        self.db.commit()
+
+    def get(self, route: str, identity: str) -> dict | None:
+        row = self.db.execute(
+            "SELECT * FROM deliveries WHERE route=? AND identity=?", (route, identity)
+        ).fetchone()
+        if row is None:
+            return None
+        result = dict(row)
+        if result["status"] == "accepted" and result["boot"] != self.boot:
+            result["status"] = "interrupted"
+        return result
+
+    def admit(self, route: str, identity: str, scope: str, chat: str) -> None:
+        with self.db:
+            self.db.execute(
+                "INSERT INTO deliveries VALUES(?,?,?,?,?,?)",
+                (route, identity, scope, chat, "accepted", self.boot),
+            )
+
+    def finish(self, chat: str) -> None:
+        with self.db:
+            self.db.execute(
+                "UPDATE deliveries SET status='completed' "
+                "WHERE chat=? AND status='accepted' AND boot=?",
+                (chat, self.boot),
+            )
+
+    def has_other_work(self, chat: str, route: str, identity: str) -> bool:
+        return (
+            self.db.execute(
+                "SELECT 1 FROM deliveries WHERE chat=? AND status='accepted' "
+                "AND NOT (route=? AND identity=?) LIMIT 1",
+                (chat, route, identity),
+            ).fetchone()
+            is not None
+        )
+
+    def cancel(self, route: str, identity: str, scope: str) -> None:
+        row = self.get(route, identity)
+        if row is not None and row["scope"] != scope:
+            raise ValueError("cancellation scope mismatch")
+        if row is not None and row["status"] == "cancelled":
+            return
+        with self.db:
+            self.db.execute(
+                "INSERT INTO deliveries VALUES(?,?,?,?,?,?) "
+                "ON CONFLICT(route,identity) DO UPDATE SET status='cancelled'",
+                (route, identity, scope, "", "cancelled", self.boot),
+            )

gateway/run.py

@@ -10072,6 +10072,11 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int:
                 )
                 continue
 
+            from gateway.platforms.webhook import WebhookAdapter
+            if isinstance(adapter, WebhookAdapter) and adapter.sender_manages_resume(source.chat_id):
+                # Durable sender reconciliation owns recovery; never run both paths.
+                continue
+
             # Validate the session owner against the current allowlist
             # before auto-resuming. A session created before
             # TELEGRAM_ALLOWED_USERS (or equivalent) was configured, or

tests/gateway/test_webhook_lifecycle.py

@@ -0,0 +1,128 @@
+"""Real HTTP admission/replay/cancellation acceptance for Harness #285."""
+
+import asyncio
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import TestClient, TestServer
+from gateway.config import PlatformConfig
+from gateway.platforms.webhook import WebhookAdapter, _INSECURE_NO_AUTH
+from gateway.platforms.webhook_lifecycle import WebhookLifecycle
+
+
+@pytest.mark.asyncio
+async def test_durable_lifecycle_response_loss_busy_finish_cancel_and_restart(tmp_path):
+    adapter = WebhookAdapter(
+        PlatformConfig(
+            enabled=True,
+            extra={
+                "rate_limit": 3,
+                "routes": {
+                    "test": {
+                        "secret": _INSECURE_NO_AUTH,
+                        "prompt": "Review {id}",
+                        "serial_key": "scope",
+                        "serial_event_key": "id",
+                    }
+                },
+            },
+        )
+    )
+    adapter._lifecycle = WebhookLifecycle(tmp_path / "lifecycle.sqlite")
+    gate = asyncio.Event()
+    started = []
+    starts = asyncio.Queue()
+    finishes = asyncio.Queue()
+    release = adapter._release_serial_group
+
+    def released(chat):
+        release(chat)
+        finishes.put_nowait(chat)
+
+    adapter._release_serial_group = released
+
+    async def handler(event):
+        started.append(event.source.chat_id)
+        starts.put_nowait(event.source.chat_id)
+        await gate.wait()
+        return ""
+
+    adapter._message_handler = handler
+    app = web.Application()
+    app.router.add_post("/webhooks/{route_name}", adapter._handle_webhook)
+    async with TestClient(TestServer(app)) as client:
+
+        async def post(identity, scope="A"):
+            r = await client.post(
+                "/webhooks/test",
+                json={"id": identity, "scope": scope},
+                headers={"X-Request-ID": identity},
+            )
+            assert r.status in (200, 202), await r.text()
+            return await r.json()
+
+        async def control(identity, operation="status", scope="A"):
+            r = await client.post(
+                "/webhooks/test",
+                json={
+                    "operation": operation,
+                    "deliveries": [{"identity": identity, "scope": scope}],
+                },
+            )
+            assert r.status == 200, await r.text()
+            return (await r.json())["deliveries"][0]
+
+        await post("one")  # sender loses this response
+        for _ in range(20):
+            assert (await post("one"))["status"] == "accepted"
+            assert (await control("one"))["status"] == "accepted"
+        assert (await post("two"))["status"] == "busy"
+        assert (await control("two"))["status"] == "unknown"
+        assert (await control("two"))["scope_busy"]
+        await asyncio.wait_for(starts.get(), 2)
+        assert len(started) == 1
+        assert len(adapter._rate_counts["test"]) == 1
+        gate.set()
+        await asyncio.wait_for(finishes.get(), 2)
+        assert (await control("one"))["status"] == "completed"
+        assert (await post("one"))["status"] == "completed"
+        assert not (await control("two"))["scope_busy"]
+        assert (await post("two"))["status"] == "accepted"
+        await asyncio.wait_for(starts.get(), 2)
+        await asyncio.wait_for(finishes.get(), 2)
+        assert len(started) == 2
+        gate.clear()
+        assert (await post("active-cancel", "D"))["status"] == "accepted"
+        await asyncio.wait_for(starts.get(), 2)
+        assert len(started) == 3
+        chat = started[-1]
+        adapter._lifecycle.admit("test", "shared-risk", "D", chat)
+        cancelled = await control("active-cancel", "cancel", "D")
+        assert cancelled["status"] == "cancelled" and cancelled["consumer_running"]
+        assert not gate.is_set() and finishes.empty()
+        # Only the last obligation may stop a shared consumer.
+        assert (await control("shared-risk", "cancel", "D"))["status"] == "cancelled"
+        await asyncio.wait_for(finishes.get(), 2)
+        assert not (await control("active-cancel", "status", "D"))["scope_busy"]
+        assert (await post("active-cancel", "D"))["status"] == "cancelled"
+        # A cancellation preceding a delayed POST persists across receiver restart.
+        assert (await control("late", "cancel", "B"))["status"] == "cancelled"
+        assert (await post("late", "B"))["status"] == "cancelled"
+        adapter._lifecycle.admit("test", "crashed", "C", "old-chat")
+        adapter._lifecycle.db.close()
+        adapter._lifecycle = WebhookLifecycle(tmp_path / "lifecycle.sqlite")
+        assert (await post("crashed", "C"))["status"] == "interrupted"
+        assert (await post("late", "B"))["status"] == "cancelled"
+        assert len(started) == 3
+        adapter._routes["test"]["secret"] = "signed-only"
+        denied = await client.post(
+            "/webhooks/test",
+            json={
+                "operation": "cancel",
+                "deliveries": [{"identity": "forged", "scope": "E"}],
+            },
+        )
+        assert denied.status == 401
+        assert adapter._lifecycle.get("test", "forged") is None
+        assert adapter.sender_manages_resume("webhook:test:crashed")
+        assert not adapter.sender_manages_resume("webhook:other:crashed")
+        adapter._lifecycle.db.close()