fix(gateway): retry a parked profile's ingress, never ack and drop it

When a secondary's startup Bot recovery fails, round 2 parks it and keeps
starting shared ingress. The parked profile then looked like an invalid
route:

- Relay: a frame routed to it was refused as `profile_mismatch`, which
  relay treats as permanent, so the frame was ACKed with nothing admitted.
- Coalesced webhook: answered 202 `coalesced`, lost on flush, and the
  provider's retry was answered `duplicate`.
- `deliver_only` / `cron_job` webhook: ran for the parked profile and
  recorded the delivery id.

Before round 2 a Bot-recovery failure aborted the start, so no ingress
connected.

A parked profile comes back on a rescan, so its traffic is now temporarily
unavailable (`profile_parked`):

- `producer_scope` refuses a delivery routed to a parked profile with the
  retryable `runtime_draining`. The relay ACK is withheld and the connector
  redelivers. A primary route's rejection now carries its target profile
  (`ProfileRouteRejected.profile`, stashed on the source as
  `_rejected_route_profile`) so the parked case can be told apart. A route
  to a profile this gateway does not serve stays `profile_mismatch`.
- A webhook to a parked `/p/<name>/` gets a retryable 503 with
  `Retry-After` right after authentication, before any delivery id is
  recorded or any route mode runs (`parked_profile_refusal`).
- A coalesced batch whose admission fails releases every delivery id it
  holds (the dispatched one and those it superseded), so a resend is
  processed, not answered as a duplicate.

Refs #134889 (GW-12)
Refs #140018

gateway/platforms/webhook.py

@@ -657,6 +657,9 @@ class WebhookAdapter(BasePlatformAdapter):
         route_name, route_config, profile, error_response = self._resolve_route(request)
         if error_response is None:
             raw_body, error_response = await self._read_authenticated_body(request, route_name, route_config)
+        if error_response is None:  # a parked profile is retried, before any delivery id is recorded
+            from gateway.platforms.webhook_ingress import parked_profile_refusal
+            error_response = parked_profile_refusal(self.gateway_runner, profile)
         if error_response is not None:
             return error_response
         # Rate limiting (after auth)
@@ -735,15 +738,18 @@ class WebhookAdapter(BasePlatformAdapter):
                                   "delivery_id": delivery_id}, status=202)
 
     def _spawn_agent_run(self, payload: Any, prompt: str, delivery_id: str, now: float, *, route_config: dict,
-                         route_name: str, profile, event_type: str) -> asyncio.Task:
+                         route_name: str, profile, event_type: str, superseded=()) -> asyncio.Task:
         """Coalesced-path dispatch (settled timer / disconnect flush): admit the merged event as a task
         so ``WebhookCoalescer.flush`` can await the hand-off. The HTTP ack for these deliveries was
-        already ``coalesced``; an admission failure is logged, never retried by the producer."""
+        already ``coalesced``: an admission failure is logged and gives back the batch's delivery ids
+        (*superseded* too), so a resend of any of them is processed, not answered as a duplicate."""
         async def _admit_or_log():
             outcome = await self._admit_agent_run(payload, prompt, delivery_id, now, route_config=route_config,
                                                   route_name=route_name, profile=profile, event_type=event_type)
             if outcome is None:
                 logger.error("[webhook] Coalesced delivery %s on route %s was not admitted", delivery_id, route_name)
+                for released in (*superseded, delivery_id):
+                    self._seen_deliveries.pop(_WebhookDeliveryIdentity.from_parts(profile, route_name, released), None)
         task = asyncio.create_task(_admit_or_log())
         self._background_tasks.add(task)
         task.add_done_callback(self._background_tasks.discard)

gateway/platforms/webhook_coalesce.py

@@ -54,6 +54,7 @@ class PendingEvent:
     first_at: float
     count: int
     dispatch_kwargs: dict[str, Any]
+    superseded: tuple[str, ...] = ()  # delivery ids of the earlier events this one replaced
 
     def prompt_with_note(self) -> str:
         if self.count <= 1:
@@ -101,8 +102,10 @@ class WebhookCoalescer:
         first_at = existing.first_at if existing else now
         count = existing.count + 1 if existing else 1
         dispatch_kwargs = {**dispatch_kwargs, "route_name": route_name, "event_type": event_type}
+        superseded = (*existing.superseded, existing.delivery_id) if existing else ()
         self._pending[group_key] = PendingEvent(payload=payload, prompt=prompt, delivery_id=delivery_id,
-                                                first_at=first_at, count=count, dispatch_kwargs=dispatch_kwargs)
+                                                first_at=first_at, count=count, dispatch_kwargs=dispatch_kwargs,
+                                                superseded=superseded)
         if existing is not None:
             logger.info("[webhook] coalesced delivery %s superseded by %s (group=%s, %d events)",
                         existing.delivery_id, delivery_id, group_key, count)
@@ -129,7 +132,7 @@ class WebhookCoalescer:
         logger.info("[webhook] coalesce settled group=%s events=%d delivery=%s", group_key, entry.count,
                     entry.delivery_id)
         return self._dispatch(entry.payload, entry.prompt_with_note(), entry.delivery_id, time.time(),
-                              **entry.dispatch_kwargs)
+                              **entry.dispatch_kwargs, superseded=entry.superseded)
 
     async def flush(self) -> None:
         """Dispatch every pending group now and wait until each run is handed to the runner, so an adapter

gateway/platforms/webhook_ingress.py

@@ -27,18 +27,42 @@ async def producer_scope(adapter, event):
     else:
         runtime_home = runner._admit_primary_source(source, home)
         if runtime_home is None:
+            _refuse_parked(runner, getattr(source, '_rejected_route_profile', None))
             source.profile_route_rejected = True
             runtime_home = home
     source._authorization_profile_home = home
     # The routed profile's own ledger admits the delivery; an unserved route is refused.
     authority = authority_for_home(runner, runtime_home)
     if authority is None:
+        _refuse_parked(runner, getattr(source, 'profile', None) or profile)
         raise RuntimeStoreError('profile_mismatch')
     with native_callback(runner, event, home, profile):
         async with _async_profile_runtime_scope(runtime_home):
             yield authority
 
 
+def _refuse_parked(runner, profile):
+    """A delivery routed to a parked profile is retried, never refused for good: a rescan serves the
+    profile again, so it is ``runtime_draining`` (no relay ACK; the connector redelivers), not the
+    permanent ``profile_mismatch`` of a route to a profile this gateway does not serve."""
+    from gateway.run_runtime import profile_parked
+    if profile_parked(runner, profile):
+        raise RuntimeStoreError('runtime_draining')
+
+
+def parked_profile_refusal(runner, profile):
+    """A retryable 503 for a webhook to a parked ``/p/<profile>/``, else None. Answered before the
+    delivery id is recorded: coalesced, ``deliver_only`` and ``cron_job`` routes act before (or
+    without) an admission, so the in-memory id is their only replay guard, and a recorded id would
+    answer the provider's retry as a duplicate once the profile is served again."""
+    from gateway.run_runtime import profile_parked
+    if not profile_parked(runner, profile):
+        return None
+    from aiohttp import web
+    return web.json_response({'error': f'Profile {profile} is not being served; retry this delivery'},
+                             status=503, headers={'Retry-After': '30'})
+
+
 async def admit_producer(adapter, event):
     """Commit a producer delivery; ``None`` when a plugin skipped or consumed it.
 

gateway/profile_routing.py

@@ -49,7 +49,14 @@ def _whatsapp_user_chat_ids_match(platform: str, left: Optional[str], right: Opt
 
 
 class ProfileRouteRejected(RuntimeError):
-    """An explicit route matched a profile this gateway does not serve."""
+    """An explicit route matched a profile this gateway does not serve (``profile``, when known)."""
+
+    @property
+    def profile(self):
+        return self.args[1] if len(self.args) > 1 else None
+
+    def __str__(self):
+        return str(self.args[0]) if self.args else ''
 
 
 @dataclass(frozen=True)

gateway/run.py

@@ -4205,7 +4205,7 @@ class GatewayRunner(
                 logger.warning(
                     "Rejecting profile route %r: target profile %r is not served",
                     matched.name, matched.profile)
-                raise ProfileRouteRejected(matched.name)
+                raise ProfileRouteRejected(matched.name, matched.profile)
             return matched.profile
         logger.debug(
             "No profile route matched: platform=%s chat_id=%s thread_id=%s parent_chat_id=%s",

gateway/run_runtime.py

@@ -81,6 +81,13 @@ def unpark_profile(runner, name):
         _record_parked_profiles(parked)
 
 
+def profile_parked(runner, name):
+    """True when *name* is parked: not served now, but a rescan serves it again, so its ingress is
+    temporarily unavailable (retry it), never an invalid route (refuse it for good)."""
+    descriptor = getattr(runner, 'session_runtime_descriptor', None) or {}
+    return bool(name) and name in (descriptor.get('parked_profiles') or {})
+
+
 def parked_profile_map(runner):
     descriptor = getattr(runner, 'session_runtime_descriptor', None)
     if descriptor is None:

gateway/session_identity.py

@@ -98,7 +98,7 @@ def clear_identity(source: Any) -> None:
     guild's cached voice source is shared by every speaker, and ``profile_routes[].user_id`` routes
     per speaker). ``source.profile`` is the previous event's routing *result*; left in place it
     short-circuits the route and the new speaker runs as the old one."""
-    for attr in (_IDENTITY_ATTR, "profile_route_rejected", "_authorization_profile_home"):
+    for attr in (_IDENTITY_ATTR, "profile_route_rejected", "_authorization_profile_home", "_rejected_route_profile"):
         with suppress(AttributeError):
             delattr(source, attr)
     with suppress(AttributeError):
@@ -257,6 +257,7 @@ def resolve_identity(
                 else runner._profile_name_for_source(source, adapter_profile=adapter_profile))
         except ProfileRouteRejected as exc:
             source.profile_route_rejected = True
+            source._rejected_route_profile = exc.profile  # a parked target is retried (producer_scope)
             raise IdentityUnresolved(f"{where}: {exc}") from exc
         source.profile = routed or adapter_profile
 

tests/gateway/test_parked_profile_ingress.py

@@ -0,0 +1,258 @@
+"""Ingress for a parked secondary is never acknowledged and then dropped.
+
+A parked profile (here: its startup Bot recovery failed) comes back through a rescan, so its
+traffic is temporarily unavailable, not an invalid route. A relay frame is not acknowledged, so the
+connector redelivers it. A webhook gets a retryable 503 before its delivery id is recorded, so the
+provider's retry is processed, not answered as a duplicate.
+"""
+import asyncio
+import json
+import sqlite3
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.run import GatewayRunner
+from gateway.runtime_ownership import process_ownership
+from tests.waiting import async_wait_until
+
+
+def _homes(tmp_path, monkeypatch):
+    root = tmp_path / '.hermes'
+    root.mkdir(mode=0o700)
+    monkeypatch.setattr(Path, 'home', lambda: tmp_path)
+    monkeypatch.setenv('HERMES_HOME', str(root))
+    import hermes_state
+    monkeypatch.setattr(hermes_state, 'DEFAULT_DB_PATH', hermes_state._IMPORT_DEFAULT_DB_PATH)
+    beta = root / 'profiles' / 'beta'
+    beta.mkdir(parents=True, mode=0o700)
+    (beta / 'config.yaml').write_text('{}')
+    return [('default', root), ('beta', beta)]
+
+
+def _admissions(home):
+    with sqlite3.connect(Path(home) / 'state.db') as conn:
+        return conn.execute('SELECT COUNT(*) FROM session_admissions').fetchone()[0]
+
+
+async def _runner_with_beta_parked(tmp_path, monkeypatch, homes, **config_fields):
+    """Multiplexed runner whose secondary ``beta`` fails startup Bot recovery once and is parked."""
+    from gateway import run
+    from gateway import session_bot
+    from gateway.run_runtime import initialize_gateway_runtime, parked_profile_map, start_gateway_runtime_api
+    monkeypatch.setattr(run, '_load_gateway_config', lambda *a: {'platform_toolsets': {'cli': []}})
+    config = GatewayConfig(sessions_dir=tmp_path / 'sessions')
+    config.multiplex_profiles = True
+    config._runtime_profile_homes = tuple(homes)
+    for key, value in config_fields.items():
+        setattr(config, key, value)
+    runner = GatewayRunner(config)
+    beta, real, failed = homes[1][1].resolve(), session_bot.recover_bot_deliveries, []
+
+    async def recover_once_failing(authority):
+        if Path(authority.profile_id).resolve() == beta and not failed:
+            failed.append(authority)
+            raise OSError('mailbox write failed')
+        return await real(authority)
+
+    async def api(_runner):
+        return SimpleNamespace(api_origin='http://127.0.0.1:1')
+
+    monkeypatch.setattr(session_bot, 'recover_bot_deliveries', recover_once_failing)
+    monkeypatch.setattr('gateway.run_api.start_gateway_api', api)
+    process_ownership.reserve([home for _, home in homes])
+    await initialize_gateway_runtime(runner)
+    await start_gateway_runtime_api(runner)
+    assert 'beta' in parked_profile_map(runner)
+    return runner
+
+
+async def _serve_beta_again(runner, beta):
+    """The control-socket rescan serves beta again and lifts its park."""
+    from gateway.run_runtime import unpark_profile
+    assert await runner._serve_profile_runtime('beta', beta) is None
+    unpark_profile(runner, 'beta')
+
+
+def _close(runner, homes):
+    for authority in list(getattr(runner, 'session_authorities', None) or []):
+        if authority is not None:
+            authority.db.close()
+    for _, home in homes:
+        process_ownership.release(home)
+
+
+@pytest.mark.asyncio
+async def test_a_relay_frame_for_a_parked_profile_is_redelivered_not_acked(tmp_path, monkeypatch):
+    websockets = pytest.importorskip('websockets')
+    from gateway.profile_routing import ProfileRoute
+    from gateway.relay.adapter import RelayAdapter
+    from gateway.relay.descriptor import CapabilityDescriptor
+    from gateway.relay.ws_transport import WebSocketRelayTransport
+    homes = _homes(tmp_path, monkeypatch)
+    runner = await _runner_with_beta_parked(
+        tmp_path, monkeypatch, homes,
+        profile_routes=[ProfileRoute(name='to-beta', platform='telegram', profile='beta', chat_id='C')])
+    descriptor = CapabilityDescriptor(1, 'telegram', 'Fixture', 4096, False, True, False, 'plain', 'chars')
+    outgoing, ready, adapter = [], asyncio.Queue(), None
+    frame = json.dumps({'type': 'inbound', 'bufferId': 'buf-1', 'event': {
+        'text': 'please do this', 'message_id': 'm1',
+        'source': {'platform': 'telegram', 'chat_id': 'C', 'chat_type': 'group', 'user_id': 'U'}}}) + '\n'
+
+    async def connector(ws):
+        await ws.recv()  # hello
+        await ws.send(json.dumps({'type': 'descriptor', 'descriptor': json.loads(descriptor.to_json())}) + '\n')
+        await ready.put(ws)
+        async for line in ws:
+            outgoing.append(json.loads(line))
+
+    from gateway.platforms import webhook_ingress
+    real_admit, attempts = webhook_ingress.admit_producer, []
+
+    async def recorded(adapter_, event):
+        try:
+            receipt = await real_admit(adapter_, event)
+        except Exception as exc:
+            attempts.append(exc)
+            raise
+        attempts.append(receipt)
+        return receipt
+
+    monkeypatch.setattr(webhook_ingress, 'admit_producer', recorded)
+    acks = lambda: [f for f in outgoing if f.get('type') == 'inbound_ack' and f.get('bufferId') == 'buf-1']
+    try:
+        async with websockets.serve(connector, '127.0.0.1', 0) as relay:
+            transport = WebSocketRelayTransport(
+                f'ws://127.0.0.1:{relay.sockets[0].getsockname()[1]}/relay', 'telegram', 'fixture-bot',
+                gateway_id='fixture-gateway', upgrade_secret='fixture-secret', reconnect=True,
+                reconnect_backoff_s=.05)
+            adapter = RelayAdapter(PlatformConfig(enabled=True), descriptor, transport)
+            adapter.gateway_runner = runner
+            runner.adapters[Platform.RELAY] = adapter
+            runner._wire_adapter_handlers(adapter)
+            assert await adapter.connect()
+            ws = await asyncio.wait_for(ready.get(), 5)
+            await ws.send(frame)
+            await async_wait_until(lambda: attempts, what='the parked delivery reached admission')
+            from gateway.relay.adapter import _PERMANENT_REFUSALS
+            refused = attempts[0]
+            assert getattr(refused, 'reason', None) not in _PERMANENT_REFUSALS, (
+                f'a parked profile was refused for good (the frame is acknowledged and lost): {refused!r}')
+            # The profile is served again and the connector redelivers the unacknowledged frame.
+            await _serve_beta_again(runner, homes[1][1].resolve())
+            await ws.send(frame)
+            await async_wait_until(acks, what='buf-1 acknowledged after the redelivery')
+            assert len(acks()) == 1, 'the parked delivery was acknowledged too'
+            assert _admissions(homes[1][1]) == 1
+    finally:
+        if adapter is not None:
+            await adapter.disconnect()
+        _close(runner, homes)
+
+
+def _webhook_request(delivery_id, body, route='events'):
+    raw = json.dumps(body).encode()
+    request = MagicMock()
+    request.headers = {'X-GitHub-Delivery': delivery_id}
+    request.content_length = len(raw)
+    request.match_info = {'route_name': route, 'profile': 'beta'}
+    request.method = 'POST'
+
+    async def read():
+        return raw
+
+    request.read = read
+    return request
+
+
+def _webhook(runner, routes):
+    from gateway.platforms.webhook import WebhookAdapter
+    adapter = WebhookAdapter(PlatformConfig(enabled=True, extra={
+        'host': '127.0.0.1', 'port': 0, 'routes': routes, 'rate_limit': 100}))
+    adapter.gateway_runner = runner
+    runner.adapters[Platform.WEBHOOK] = adapter
+    runner._wire_adapter_handlers(adapter)
+    return adapter
+
+
+def _cancel(adapter):
+    for task in list(adapter._background_tasks):
+        task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_a_coalesced_webhook_for_a_parked_profile_is_retried_not_lost(tmp_path, monkeypatch):
+    from gateway.platforms.webhook import _INSECURE_NO_AUTH
+    homes = _homes(tmp_path, monkeypatch)
+    beta = homes[1][1].resolve()
+    runner = await _runner_with_beta_parked(tmp_path, monkeypatch, homes)
+    adapter = _webhook(runner, {'events': {'secret': _INSECURE_NO_AUTH, 'prompt': 'event {id}', 'profile': 'beta',
+                                           'coalesce': {'key': 'id', 'window_seconds': 0.1}}})
+    settled = lambda: not adapter._coalescer.pending and not adapter._background_tasks
+    try:
+        first = await adapter._handle_webhook(_webhook_request('delivery-1', {'id': 17}))
+        assert first.status == 503 and first.headers.get('Retry-After'), (first.status, first.text)
+        await _serve_beta_again(runner, beta)
+        retry = await adapter._handle_webhook(_webhook_request('delivery-1', {'id': 17}))
+        await async_wait_until(settled, what='the retried batch dispatched')
+        assert json.loads(retry.text)['status'] == 'coalesced', retry.text
+        assert _admissions(beta) == 1, 'delivery-1 was never admitted'
+    finally:
+        _cancel(adapter)
+        _close(runner, homes)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('mode', ['deliver_only', 'cron_job'])
+async def test_a_non_durable_webhook_for_a_parked_profile_is_refused_before_its_id_is_recorded(
+        tmp_path, monkeypatch, mode):
+    """``deliver_only`` and ``cron_job`` act without an admission: the in-memory delivery id is their
+    only replay guard, so a parked profile must refuse before recording it."""
+    from gateway.platforms.webhook import _INSECURE_NO_AUTH
+    homes = _homes(tmp_path, monkeypatch)
+    runner = await _runner_with_beta_parked(tmp_path, monkeypatch, homes)
+    route = {'secret': _INSECURE_NO_AUTH, 'prompt': 'event {id}', 'profile': 'beta', mode: True}
+    if mode == 'deliver_only':
+        route['deliver'] = 'log'
+    adapter = _webhook(runner, {'events': route})
+    try:
+        refused = await adapter._handle_webhook(_webhook_request('delivery-1', {'id': 17}))
+        assert refused.status == 503 and refused.headers.get('Retry-After'), (refused.status, refused.text)
+        assert adapter._seen_deliveries == {}, 'the refused delivery id was recorded'
+    finally:
+        _cancel(adapter)
+        _close(runner, homes)
+
+
+@pytest.mark.asyncio
+async def test_a_coalesced_batch_whose_admission_fails_releases_its_delivery_ids(tmp_path, monkeypatch):
+    """The provider got 202 ``coalesced`` for every delivery of the batch. If the flush cannot admit
+    it, a resend of any of them must be processed, not answered as a duplicate."""
+    from gateway.platforms import webhook_ingress
+    from gateway.platforms.webhook import _INSECURE_NO_AUTH, _WebhookDeliveryIdentity
+    homes = _homes(tmp_path, monkeypatch)
+    runner = await _runner_with_beta_parked(tmp_path, monkeypatch, homes)
+    await _serve_beta_again(runner, homes[1][1].resolve())
+    adapter = _webhook(runner, {'events': {'secret': _INSECURE_NO_AUTH, 'prompt': 'event {id}', 'profile': 'beta',
+                                           'coalesce': {'key': 'id', 'window_seconds': 0.1}}})
+
+    async def store_busy(adapter_, event):
+        raise sqlite3.OperationalError('database is locked')
+
+    monkeypatch.setattr(webhook_ingress, 'admit_producer', store_busy)
+    settled = lambda: not adapter._coalescer.pending and not adapter._background_tasks
+    try:
+        for delivery in ('delivery-1', 'delivery-2'):  # one entity: delivery-2 supersedes delivery-1
+            accepted = await adapter._handle_webhook(_webhook_request(delivery, {'id': 17}))
+            assert json.loads(accepted.text)['status'] == 'coalesced', accepted.text
+        await async_wait_until(settled, what='the batch flushed')
+        held = [identity for identity in adapter._seen_deliveries
+                if identity in {_WebhookDeliveryIdentity.from_parts('beta', 'events', d)
+                                for d in ('delivery-1', 'delivery-2')}]
+        assert held == [], f'a failed batch still holds its delivery ids: {held}'
+    finally:
+        _cancel(adapter)
+        _close(runner, homes)