From 9c829f965dcd4b2ee536dbafcf555f58b0d110b5 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 24 Aug 2026 00:28:41 +0530 Subject: [PATCH] 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). --- .../desktop/src/plugins/hermes-bots/plugin.js | 46 ++++++++++ .../tests/relay-push-drain.test.mjs | 85 +++++++++++++++++++ tests/tui_gateway/test_change_watcher.py | 41 +++++++++ tui_gateway/server.py | 39 +++++++++ 4 files changed, 211 insertions(+) create mode 100644 apps/desktop/src/plugins/hermes-bots/tests/relay-push-drain.test.mjs diff --git a/apps/desktop/src/plugins/hermes-bots/plugin.js b/apps/desktop/src/plugins/hermes-bots/plugin.js index e93278173a..1090ac6260 100644 --- a/apps/desktop/src/plugins/hermes-bots/plugin.js +++ b/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: diff --git a/apps/desktop/src/plugins/hermes-bots/tests/relay-push-drain.test.mjs b/apps/desktop/src/plugins/hermes-bots/tests/relay-push-drain.test.mjs new file mode 100644 index 0000000000..63a63f3782 --- /dev/null +++ b/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\)/) +}) diff --git a/tests/tui_gateway/test_change_watcher.py b/tests/tui_gateway/test_change_watcher.py index c078528349..a5bff9e01c 100644 --- a/tests/tui_gateway/test_change_watcher.py +++ b/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) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index ef1b98b5f9..63b382c7dc 100644 --- a/tui_gateway/server.py +++ b/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