feat(bot-mode): push-notified relay drain with poll backstop (#93091)

Cross-connection DMs were pure polling: the Desktop drains every gateway's
bot_relay outbox on a 4s interval, so each hop eats up to 4s outbound plus
4s for the reply leg (#92760 'bots reply slowly').

Emission point: the gateway's existing change watcher (_CHANGE_WATCHES in
tui_gateway/server.py). Envelopes are written by the AGENT process
(message_agent -> tools.bot_relay.enqueue_envelope), not the gateway, so no
gateway RPC is on the enqueue path and an in-process emit is impossible.
That is exactly the situation the change watcher already solves for the
pairing store (pairing.changed: 'written by a different process; the files
are the only shared signal') - so a new bot_relay.outbox.pending entry in
the existing watch table is the smallest correct diff: one cheap 1s-interval
stat probe folded into the existing 0.5s watcher tick, no new thread, no new
RPC, and _broadcast_global_event fans it to every connected WS client for
free. The signature is monotone (newest envelope mtime ever seen) so a
drain emptying outbox/ never re-fires the event.

Desktop (hermes-bots plugin): subscribe via the existing host.onEvent tap
(feature-detected - older shells lack it) and run drainRelayOutboxes through
a 250ms trailing debounce so a burst of signals collapses to one drain.
The 4s interval poll is intentionally UNCHANGED as the backstop: the event
tap only hears the active gateway socket, so per-connection push detection
would be complex and wrong to trade the poll against - push simply makes
the common case near-instant while older backends keep working exactly as
before.

Tests: 3 new watcher contracts (fires on enqueue, monotone across drain,
silent with no outbox) and a new relay-push-drain.test.mjs (debounce burst
-> one drain, re-arm after window, disposed no-op, poll backstop intact).

apps/desktop/src/plugins/hermes-bots/plugin.js

@@ -1376,11 +1376,18 @@ function handleSessionsGatewayTransition() {
 // relay degrades to whatever subset of connections supports it.
 const RELAY_ROSTER_INTERVAL_MS = 60_000
 const RELAY_DRAIN_INTERVAL_MS = 4_000
+// Push path (#93091): the gateway broadcasts `bot_relay.outbox.pending` when
+// an envelope lands on disk; a burst of signals inside this window collapses
+// to ONE drain. The interval poll above stays as the backstop for older
+// backends (and connections whose events don't reach the tap).
+const RELAY_PUSH_DEBOUNCE_MS = 250
 let relayDisposed = false
 let relayRosterTimer = null
 let relayDrainTimer = null
 let relayRosterBusy = false
 let relayDrainBusy = false
+let relayPushUnsub = null
+let relayPushDebounceTimer = null
 
 /** One representative route per reachable connection id. */
 async function relayConnections() {
@@ -1578,6 +1585,23 @@ async function drainRelayOutboxes() {
   }
 }
 
+/** Push-notified drain (#93091): collapse a burst of pending signals into
+ *  one drain call ~RELAY_PUSH_DEBOUNCE_MS after the first signal. */
+function scheduleRelayPushDrain() {
+  if (relayDisposed || typeof setTimeout !== 'function') {
+    return
+  }
+
+  if (relayPushDebounceTimer !== null) {
+    return
+  }
+
+  relayPushDebounceTimer = setTimeout(() => {
+    relayPushDebounceTimer = null
+    void drainRelayOutboxes()
+  }, RELAY_PUSH_DEBOUNCE_MS)
+}
+
 function startBotRelay() {
   relayDisposed = false
 
@@ -1595,6 +1619,14 @@ function startBotRelay() {
   if (relayDrainTimer === null) {
     relayDrainTimer = setInterval(() => void drainRelayOutboxes(), RELAY_DRAIN_INTERVAL_MS)
   }
+
+  // Push path: the gateway change watcher broadcasts when an envelope hits
+  // the outbox; drain immediately (debounced) instead of waiting the poll
+  // out. Feature-detected — older shells have no host.onEvent — and the 4s
+  // poll above stays untouched as the backstop either way.
+  if (relayPushUnsub === null && typeof host.onEvent === 'function') {
+    relayPushUnsub = host.onEvent('bot_relay.outbox.pending', () => scheduleRelayPushDrain())
+  }
 }
 
 function stopBotRelay() {
@@ -1609,6 +1641,20 @@ function stopBotRelay() {
     clearInterval(relayDrainTimer)
     relayDrainTimer = null
   }
+
+  if (relayPushDebounceTimer !== null) {
+    clearTimeout(relayPushDebounceTimer)
+    relayPushDebounceTimer = null
+  }
+
+  if (relayPushUnsub !== null) {
+    try {
+      relayPushUnsub()
+    } catch {
+      // Disposer from an older shell shape — never break teardown.
+    }
+    relayPushUnsub = null
+  }
 }
 
 /** Per-bot appearance + display meta, persisted via ctx.storage:

apps/desktop/src/plugins/hermes-bots/tests/relay-push-drain.test.mjs

@@ -0,0 +1,85 @@
+import assert from 'node:assert/strict'
+import { readFileSync } from 'node:fs'
+import test from 'node:test'
+import vm from 'node:vm'
+
+// Push-notified relay drain (#93091, motivated by #92760 slow replies): the
+// gateway broadcasts `bot_relay.outbox.pending` when an envelope lands in the
+// outbox; the plugin drains immediately instead of waiting out the 4s poll.
+// Contracts:
+// - a burst of signals inside RELAY_PUSH_DEBOUNCE_MS collapses to ONE drain;
+// - after the debounce fires, a later signal schedules a fresh drain;
+// - the subscription is feature-detected (host.onEvent) and disposed, and the
+//   4s interval poll remains untouched as the backstop.
+
+const pluginSource = readFileSync(new URL('../plugin.js', import.meta.url), 'utf8')
+
+function loadScheduler() {
+  const start = pluginSource.indexOf('function scheduleRelayPushDrain(')
+  const end = pluginSource.indexOf('function startBotRelay(', start)
+  assert.ok(start > 0, 'plugin defines scheduleRelayPushDrain')
+  assert.ok(end > start, 'scheduler has a stable source boundary')
+
+  const context = {
+    setTimeout,
+    clearTimeout,
+    relayDisposed: false,
+    relayPushDebounceTimer: null,
+    RELAY_PUSH_DEBOUNCE_MS: 50, // shrunk from 250ms to keep the test fast
+    drainCalls: 0
+  }
+  vm.createContext(context)
+  vm.runInContext(
+    `async function drainRelayOutboxes() { drainCalls += 1 }\n${pluginSource.slice(start, end)}\nglobalThis.scheduleRelayPushDrain = scheduleRelayPushDrain`,
+    context
+  )
+  return context
+}
+
+const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))
+
+test('a burst of pending signals collapses to one drain', async () => {
+  const ctx = loadScheduler()
+
+  for (let i = 0; i < 6; i += 1) {
+    ctx.scheduleRelayPushDrain()
+  }
+
+  assert.equal(ctx.drainCalls, 0, 'drain waits for the debounce window')
+  await sleep(120)
+  assert.equal(ctx.drainCalls, 1, 'six signals inside the window → one drain')
+})
+
+test('a signal after the window fires a fresh drain', async () => {
+  const ctx = loadScheduler()
+
+  ctx.scheduleRelayPushDrain()
+  await sleep(120)
+  assert.equal(ctx.drainCalls, 1)
+
+  ctx.scheduleRelayPushDrain()
+  ctx.scheduleRelayPushDrain()
+  await sleep(120)
+  assert.equal(ctx.drainCalls, 2)
+})
+
+test('a disposed relay never schedules a drain', async () => {
+  const ctx = loadScheduler()
+  ctx.relayDisposed = true
+
+  ctx.scheduleRelayPushDrain()
+  await sleep(120)
+  assert.equal(ctx.drainCalls, 0)
+})
+
+test('subscription is feature-detected, disposed, and the poll backstop stays', () => {
+  const start = pluginSource.indexOf("host.onEvent('bot_relay.outbox.pending'")
+  assert.ok(start > 0, 'plugin subscribes to bot_relay.outbox.pending')
+  const block = pluginSource.slice(start - 400, start + 400)
+  assert.match(block, /typeof host\.onEvent === 'function'/)
+  assert.match(pluginSource, /relayPushUnsub\(\)/)
+  // The 4s interval poll is untouched — push is an accelerator, not a
+  // replacement (older backends never emit the event).
+  assert.match(pluginSource, /RELAY_DRAIN_INTERVAL_MS = 4_000/)
+  assert.match(pluginSource, /setInterval\(\(\) => void drainRelayOutboxes\(\), RELAY_DRAIN_INTERVAL_MS\)/)
+})

tests/tui_gateway/test_change_watcher.py

@@ -24,6 +24,7 @@ def watcher_home(tmp_path, monkeypatch):
     monkeypatch.setattr(server, "_change_sigs", {})
     monkeypatch.setattr(server, "_change_checked_at", {})
     monkeypatch.setattr(server, "_change_broadcast_at", {})
+    monkeypatch.setattr(server, "_bot_relay_outbox_seen", 0)
 
     events = []
     monkeypatch.setattr(
@@ -177,6 +178,46 @@ def test_renderable_pet_broadcasts_meta_payload(watcher_home, monkeypatch):
     assert payload["spritesheetRevision"]
 
 
+def test_enqueued_envelope_broadcasts_outbox_pending(watcher_home):
+    """A cross-connection envelope written by the agent process must reach the
+    Desktop's push-triggered drain on its own signal (#93091) — the drain poll
+    is the backstop, not the transport."""
+    home, events = watcher_home
+    outbox = home / "bot_relay" / "outbox"
+    outbox.mkdir(parents=True)
+    server._broadcast_watched_changes(now=0.0)
+
+    (outbox / ("a" * 32 + ".json")).write_text('{"id": "' + "a" * 32 + '"}')
+    server._broadcast_watched_changes(now=10.0)
+
+    assert ("bot_relay.outbox.pending", {}) in events
+
+
+def test_drained_outbox_does_not_rebroadcast_pending(watcher_home):
+    """Signature is monotone: a drain empties outbox/ (rename → claimed/), and
+    that emptying must NOT look like a change — only new envelopes fire."""
+    home, events = watcher_home
+    outbox = home / "bot_relay" / "outbox"
+    outbox.mkdir(parents=True)
+    envelope = outbox / ("b" * 32 + ".json")
+    envelope.write_text("{}")
+    server._broadcast_watched_changes(now=0.0)
+
+    envelope.unlink()  # the Desktop drained it
+    server._broadcast_watched_changes(now=10.0)
+    server._broadcast_watched_changes(now=20.0)
+
+    assert not [e for e in events if e[0] == "bot_relay.outbox.pending"]
+
+
+def test_no_outbox_dir_never_fires_pending(watcher_home):
+    home, events = watcher_home
+    server._broadcast_watched_changes(now=0.0)
+    server._broadcast_watched_changes(now=10.0)
+
+    assert not [e for e in events if e[0] == "bot_relay.outbox.pending"]
+
+
 def test_broken_probe_never_kills_the_pass(watcher_home, monkeypatch):
     home, events = watcher_home
     server._broadcast_watched_changes(now=0.0)

tui_gateway/server.py

@@ -3936,6 +3936,42 @@ def _pairing_sig():
     return sig
 
 
+# Newest outbox-envelope mtime the watcher has EVER seen (monotone). A drain
+# empties the outbox (rename → claimed/), and letting the signature fall back
+# to None on empty would fire a spurious pending event right after every
+# drain — so the signature only moves forward, on genuinely new envelopes.
+_bot_relay_outbox_seen = 0
+
+
+def _bot_relay_outbox_sig():
+    """Newest mtime across pending bot-relay outbox envelopes (monotone).
+
+    Envelopes are written by the AGENT process (``message_agent`` →
+    ``tools.bot_relay.enqueue_envelope``) — a different process that never
+    touches this gateway's transports — so the files are the only shared
+    signal, exactly like the pairing store. The Desktop reacts to
+    ``bot_relay.outbox.pending`` with an immediate (debounced) drain instead
+    of waiting out its poll interval (#93091, motivated by #92760).
+    """
+    global _bot_relay_outbox_seen
+    home = _watcher_home()
+    root = home.parent.parent if home.parent.name == "profiles" else home
+    newest = 0
+    try:
+        for entry in (root / "bot_relay" / "outbox").iterdir():
+            if not entry.name.endswith(".json"):
+                continue
+            try:
+                newest = max(newest, entry.stat().st_mtime_ns)
+            except OSError:
+                continue
+    except OSError:
+        pass
+    if newest > _bot_relay_outbox_seen:
+        _bot_relay_outbox_seen = newest
+    return _bot_relay_outbox_seen or None
+
+
 # Watched change signals: event → (check interval, signature fn, payload fn).
 # Signatures are stat/dict-lookup cheap, same bar as the skin watcher; the
 # check interval keeps the pricier probes (pet resolves the active sheet off
@@ -3946,6 +3982,9 @@ _CHANGE_WATCHES: dict[str, tuple[float, Any, Any]] = {
     "sessions.changed": (0.5, _sessions_sig, lambda: {}),
     "platforms.changed": (2.0, _platforms_sig, lambda: {}),
     "pairing.changed": (2.0, _pairing_sig, lambda: {}),
+    # Cross-connection DM latency: 1s check so a queued envelope reaches the
+    # Desktop's push-triggered drain fast; the Desktop's poll stays backstop.
+    "bot_relay.outbox.pending": (1.0, _bot_relay_outbox_sig, lambda: {}),
 }
 
 # state.db moves on every message append during a streaming turn, and the