fix(gateway): one parked-profile notice per chat per minute

A relay reconnect drains its per-instance backlog in order; each frame
for a parked profile is acknowledged and refused with an in-channel
notice, so a backlog of N frames posted N identical notices in the same
chat. The notice is now sent once per (profile, platform, chat) per 60 s
window; later refusals in the window are logged at INFO and still admit
nothing. Another chat, or the same chat after the window, is told again.

Review finding by @kshitijk4poor on #140967
Refs #134889 (GW-17)
Refs #140719

gateway/parked_route_refusal.py

@@ -20,6 +20,7 @@ the delivery yet: it runs that dispatch in :func:`retryable_ingress` and answers
 import asyncio
 import contextvars
 import logging
+import time
 from contextlib import contextmanager
 
 from hermes_state_runtime import RuntimeStoreError
@@ -70,6 +71,28 @@ def retryable_ingress():
 # Worded as an outage notice Discord's backfill recognises (``_is_down_notice_content``): the reply
 # must not mark the never-admitted message answered, so backfill picks it up once it is served again.
 PARKED_ROUTE_NOTICE = 'The agent is currently unavailable; please resend your message in a moment.'
+# One notice per (profile, platform, chat) per window: a relay reconnect drains its whole backlog
+# for a parked profile at once, and N frames must not post N notices in the same chat.
+PARKED_NOTICE_WINDOW_S = 60.0
+_monotonic = time.monotonic
+
+
+def _notice_due(adapter, profile, platform, chat_id):
+    """True (and the window restarts) when this chat was not told about *profile* in the window."""
+    sent = getattr(adapter, '_parked_notice_sent_at', None)
+    if not isinstance(sent, dict):
+        sent = {}
+        try:
+            adapter._parked_notice_sent_at = sent
+        except AttributeError:
+            return True
+    now, key = _monotonic(), (profile, platform, chat_id)
+    for stale in [k for k, at in sent.items() if now - at >= PARKED_NOTICE_WINDOW_S]:
+        del sent[stale]
+    if key in sent:
+        return False
+    sent[key] = now
+    return True
 
 
 def drop_rejected_route(adapter, source):
@@ -100,6 +123,12 @@ def refuse_parked_route(adapter, source, profile, reply_to=None):
             '[%s] Refusing inbound event for %s: routed profile %r on %s is parked (reason=profile_parked); '
             'nothing was admitted, the delivery is answered retryably', name, chat_id, profile, platform)
         return
+    if not _notice_due(adapter, profile, platform, chat_id):
+        logger.info(
+            '[%s] Refusing inbound event for %s: routed profile %r on %s is parked (reason=profile_parked); '
+            'nothing was admitted, the chat was already told within %.0fs', name, chat_id, profile, platform,
+            PARKED_NOTICE_WINDOW_S)
+        return
     logger.warning(
         '[%s] Refusing inbound event for %s: routed profile %r on %s is parked (reason=profile_parked); '
         'nothing was admitted, a resend notice to the sender is scheduled', name, chat_id, profile, platform)

tests/gateway/test_parked_route_refusal_send.py

@@ -60,3 +60,22 @@ async def test_a_successful_refusal_send_logs_no_failure(caplog):
         await _refuse(bot)
     assert len(bot.sent) == 1
     assert not [r for r in caplog.records if 'not delivered' in r.getMessage()]
+
+
+@pytest.mark.asyncio
+async def test_a_backlog_for_a_parked_profile_tells_each_chat_once_per_window(monkeypatch):
+    """A relay reconnect drains a backlog of N frames for a parked profile; the chat is told once
+    per (profile, platform, chat) and window, not N times. Another chat is still told."""
+    from gateway import parked_route_refusal
+    clock = [1000.0]
+    monkeypatch.setattr(parked_route_refusal, '_monotonic', lambda: clock[0])
+    bot = _Bot(SendResult(success=True))
+    for _ in range(5):
+        await _refuse(bot)
+    other = SimpleNamespace(platform=Platform.SLACK, chat_id='D', message_id='m2', _rejected_route_profile='beta')
+    drop_rejected_route(bot, other)
+    await asyncio.gather(*list(bot._background_tasks))
+    assert [chat for chat, _text, _reply in bot.sent] == ['C', 'D']
+    clock[0] += parked_route_refusal.PARKED_NOTICE_WINDOW_S
+    await _refuse(bot)
+    assert [chat for chat, _text, _reply in bot.sent] == ['C', 'D', 'C']