diff --git a/contributors/emails/alazhouhang@qq.com b/contributors/emails/alazhouhang@qq.com new file mode 100644 index 0000000000000..8c3b33f915d84 --- /dev/null +++ b/contributors/emails/alazhouhang@qq.com @@ -0,0 +1 @@ +09473ZH diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py index f72bd1cb7911a..871792d693876 100644 --- a/gateway/platforms/webhook.py +++ b/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: diff --git a/gateway/platforms/webhook_lifecycle.py b/gateway/platforms/webhook_lifecycle.py new file mode 100644 index 0000000000000..619ae501a6db2 --- /dev/null +++ b/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), + ) diff --git a/gateway/run.py b/gateway/run.py index 53496b970a406..3888a4e619016 100644 --- a/gateway/run.py +++ b/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 diff --git a/tests/gateway/test_webhook_lifecycle.py b/tests/gateway/test_webhook_lifecycle.py new file mode 100644 index 0000000000000..4baa2af23b72a --- /dev/null +++ b/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()