From 990e43d216be7967452aa3a5d776a830c516ca8c Mon Sep 17 00:00:00 2001 From: Foad Kesheh Date: Sat, 5 Sep 2026 08:30:53 -0300 Subject: [PATCH] fix(dashboard): blank terminal for agent-spawned sessions, inflated bridge count, and Centrifugo reconnect flapping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three user-visible problems, one shared cause in how the bridge and the dashboard model sessions and connections. Blank terminal pane. The bridge served terminal output only for sessions whose PTY was attached in the current process, checked via an in-memory map. A session alive in tmux but not attached in-process — the normal state right after another agent spawns or re-runs one — failed that check, so the browser's watch request was dropped and nothing was published, on every device. The liveness check now also counts live tmux sessions, and the first watcher reattaches to the tmux session (reusing the existing resurrection path, read-only, no init keystroke) so output streams. A missing bridgeId on a session row can no longer blank the pane: the UI preserves the known bridgeId across merges. "N bridges" too high. The bridge list was keyed by Centrifugo connection id, so one bridge that reconnected showed as several. It now tracks all connections per bridgeId and shows one entry per bridge (newest connection wins), and a bridge disappears only when its last connection leaves — no transient "0 bridges" that would disable the + button. Reconnect flapping (the source of the duplicate connections). The bridge subscribed to channels it only publishes to, so Centrifugo echoed its whole output stream back; the application-level ping then queued behind that backlog and the client declared "no ping" and reconnected, roughly once a minute. The publish-only subscriptions are removed (publishing is allowed by allow_publish_for_client, verified), a maxServerPingDelay is set, the reconnect re-sync is throttled and skips finished sessions, the tmux liveness probe is cached, and the Solo hub config carries explicit ping settings. Verified in an isolated Docker container running this branch: an agent-spawned session (created via the local API, not the panel) rendered its terminal and echoed a marker in a real browser, proving both the reattach fix and that terminal output still publishes after the subscriptions were removed; no flapping and no publish errors over the run. 772 bridge tests and the UI hook tests pass. Co-Authored-By: Claude Fable 5.1 --- bridge/package-lock.json | 4 +- bridge/package.json | 2 +- bridge/src/centrifugo-client.ts | 44 +++++-- .../direct-transport/publish-router.test.ts | 99 +++++++++++++- bridge/src/direct-transport/publish-router.ts | 34 ++++- bridge/src/index.ts | 78 ++++++++++- bridge/src/tmux.test.ts | 75 +++++++++++ bridge/src/tmux.ts | 60 ++++++++- ui/src/hooks/useBridges.test.ts | 123 ++++++++++++++++++ ui/src/hooks/useBridges.ts | 90 +++++++++++-- ui/src/hooks/useSessions.test.ts | 74 +++++++++++ ui/src/hooks/useSessions.ts | 13 +- 12 files changed, 657 insertions(+), 39 deletions(-) create mode 100644 bridge/src/tmux.test.ts create mode 100644 ui/src/hooks/useBridges.test.ts create mode 100644 ui/src/hooks/useSessions.test.ts diff --git a/bridge/package-lock.json b/bridge/package-lock.json index ed86196..6b6ed0a 100644 --- a/bridge/package-lock.json +++ b/bridge/package-lock.json @@ -1,12 +1,12 @@ { "name": "ftown-bridge", - "version": "0.19.24", + "version": "0.19.25", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "ftown-bridge", - "version": "0.19.24", + "version": "0.19.25", "license": "MIT", "dependencies": { "@xterm/addon-serialize": "^0.14.0", diff --git a/bridge/package.json b/bridge/package.json index b6d1a53..c894386 100644 --- a/bridge/package.json +++ b/bridge/package.json @@ -1,6 +1,6 @@ { "name": "ftown-bridge", - "version": "0.19.24", + "version": "0.19.25", "description": "CLI bridge for ftown — generic PTY-over-Centrifugo relay", "type": "module", "main": "dist/index.js", diff --git a/bridge/src/centrifugo-client.ts b/bridge/src/centrifugo-client.ts index 3711c04..e93a787 100644 --- a/bridge/src/centrifugo-client.ts +++ b/bridge/src/centrifugo-client.ts @@ -90,6 +90,15 @@ export class CentrifugoClient { token, getToken, websocket: WebSocket, + // Server sends an app-level ping every ping_interval (10s in + // centrifugo/config.json). centrifuge-js treats the connection as dead if + // no server ping arrives within maxServerPingDelay of the previous one. + // The library default (10s) leaves ZERO slack: a single inbound-backlog + // stall (e.g. a websocket_compression + reconnect re-sync storm delaying + // the ping frame past 10s) trips a false "no ping" disconnect and starts + // the reconnect flapping. 20s = one full ping_interval of tolerance, so a + // ping must be missed entirely (not merely delayed) before we reconnect. + maxServerPingDelay: 20_000, }); this.client.on('connecting', (ctx) => { @@ -140,11 +149,18 @@ export class CentrifugoClient { this.client.disconnect(); } - subscribeToSessions(userId: string): void { - const channel = `sessions:updates#${userId}`; - const sub = this.client.newSubscription(channel); - sub.subscribe(); - this.subscriptions.set(channel, sub); + // The bridge only PUBLISHES session updates (see publishSessionUpdate); it has + // no publication handler for `sessions:updates#{userId}` — the UI is the sole + // consumer. The `sessions` namespace has `allow_publish_for_client: true` + // (centrifugo/config.json), so publishing needs no subscription. The old + // subscribe here served only to satisfy a permission the config already + // grants, at the cost of Centrifugo echoing every one of our own publications + // back over the wire — dead inbound traffic that, during a reconnect re-sync + // storm, delayed the app-level ping and tripped false "no ping" reconnects. + // Removed. Kept as a no-op so the integrator's call site (index.ts) is + // unaffected — this file does not own index.ts. + subscribeToSessions(_userId: string): void { + // intentionally empty — publish-only channel, no subscription needed. } async publishSessionUpdate(userId: string, session: Session): Promise { @@ -191,14 +207,16 @@ export class CentrifugoClient { async publishTerminalData(userId: string, sessionId: string, data: string): Promise { const channel = `terminal:${sessionId}#${userId}`; - if (!this.subscriptions.has(channel)) { - const sub = this.client.newSubscription(channel); - this.subscriptions.set(channel, sub); - await new Promise((resolve) => { - sub.on('subscribed', () => resolve()); - sub.subscribe(); - }); - } + // The bridge only PUBLISHES terminal output; the UI is the sole subscriber. + // The `terminal` namespace has `allow_publish_for_client: true` + // (centrifugo/config.json), so client publishing needs no subscription. The + // old subscribe-before-publish here existed only to satisfy a permission the + // config already grants — and it was the worst offender for the reconnect + // flapping: subscribing to our OWN high-volume output channel made Centrifugo + // echo every keystroke of terminal output straight back to the bridge (dead + // inbound traffic, no handler), and `terminal` has `force_recovery: true` + // with a 10000-entry history, so each reconnect replayed that backlog and + // starved the app-level ping. Removed — we publish directly. try { await this.client.publish(channel, truncateData({ type: 'output', data })); } catch (err) { diff --git a/bridge/src/direct-transport/publish-router.test.ts b/bridge/src/direct-transport/publish-router.test.ts index 813b085..08d1084 100644 --- a/bridge/src/direct-transport/publish-router.test.ts +++ b/bridge/src/direct-transport/publish-router.test.ts @@ -11,7 +11,7 @@ import assert from 'node:assert/strict'; import { PublishRouter } from './publish-router.js'; import type { CentrifugoPublisher } from './publish-router.js'; -import type { WatchRegistry } from './watch-registry.js'; +import { WatchRegistry } from './watch-registry.js'; import type { DirectPeerManager } from './peer-manager.js'; import type { SignalType } from './contract.js'; @@ -96,14 +96,16 @@ function makeRouter(isKnownSession?: (sessionId: string) => boolean) { const watchRegistry = new FakeWatchRegistry(); const peerManager = new FakePeerManager(); const centrifugo = new FakeCentrifugoClient(); + const warnings: string[] = []; const router = new PublishRouter({ registry: watchRegistry as unknown as WatchRegistry, peerManager: peerManager as unknown as DirectPeerManager, centrifugo, userId: USER_ID, isKnownSession, + warn: (message) => { warnings.push(message); }, }); - return { router, watchRegistry, peerManager, centrifugo }; + return { router, watchRegistry, peerManager, centrifugo, warnings }; } describe('PublishRouter.publishTerminalData', () => { @@ -353,3 +355,96 @@ describe('PublishRouter loopback fan-out (addendum)', () => { assert.deepStrictEqual(peerManager.sendScreenCalls, [['sess-1', 'SCREEN']]); }); }); + +/** + * Sessions alive in tmux but with no PTY client in this bridge process (agent + * spawned via the local API, a re-run, or a session adopted after a restart). + * `isKnownSession` is wired in index.ts as + * `runner.isRunning(sid) || terminalManager.has(sid) || runner.hasTmuxSession(sid)`; + * these tests pin the router half of that contract — the third arm must be able + * to admit a watch on its own, and a genuinely unknown session must still be + * dropped, now with exactly one log line naming it. + */ +describe('PublishRouter tmux-only sessions and unknown-watch logging', () => { + /** Stand-in for the index.ts predicate: nothing in-process, alive in tmux. */ + const tmuxOnly = (alive: string) => (sessionId: string) => sessionId === alive; + + it('accepts terminal_watch for a session known only through the tmux arm of the predicate', () => { + const { router, watchRegistry, warnings } = makeRouter(tmuxOnly('tmux-sess')); + + router.handleCommand({ type: 'terminal_watch', sessionId: 'tmux-sess', clientId: 'client-1' }); + + assert.deepStrictEqual(watchRegistry.watchCalls, [['tmux-sess', 'client-1']]); + assert.strictEqual(watchRegistry.hasWatchers('tmux-sess'), true); + assert.deepStrictEqual(warnings, []); + }); + + it('a tmux-only watch fires onNewWatcher on the real WatchRegistry (the screen-dump trigger)', () => { + const registry = new WatchRegistry({ sweepIntervalMs: 0 }); + const newWatchers: string[] = []; + registry.onNewWatcher((sessionId) => { newWatchers.push(sessionId); }); + const router = new PublishRouter({ + registry, + peerManager: new FakePeerManager() as unknown as DirectPeerManager, + centrifugo: new FakeCentrifugoClient(), + userId: USER_ID, + isKnownSession: tmuxOnly('tmux-sess'), + }); + + router.handleCommand({ type: 'terminal_watch', sessionId: 'tmux-sess', clientId: 'client-1' }); + + assert.deepStrictEqual(newWatchers, ['tmux-sess']); + assert.strictEqual(registry.hasWatchers('tmux-sess'), true); + registry.dispose(); + }); + + it('R2 output for a tmux-only session reaches Centrifugo once its watch is accepted', async () => { + const { router, centrifugo } = makeRouter(tmuxOnly('tmux-sess')); + + await router.publishTerminalData('tmux-sess', 'pre-watch'); + assert.deepStrictEqual(centrifugo.calls, []); + + router.handleCommand({ type: 'terminal_watch', sessionId: 'tmux-sess', clientId: 'client-1' }); + await router.publishTerminalData('tmux-sess', 'post-watch'); + + assert.deepStrictEqual( + centrifugo.calls.map((c) => [c.kind, c.payload]), + [['data', 'post-watch']], + ); + }); + + it('still drops terminal_watch for an unknown session, and logs it once with the sessionId', () => { + const { router, watchRegistry, warnings } = makeRouter(() => false); + + // Watchers re-send terminal_watch every WATCH_HEARTBEAT_MS; only the first logs. + router.handleCommand({ type: 'terminal_watch', sessionId: 'foreign-sess', clientId: 'client-1' }); + router.handleCommand({ type: 'terminal_watch', sessionId: 'foreign-sess', clientId: 'client-1' }); + router.handleCommand({ type: 'terminal_watch', sessionId: 'foreign-sess', clientId: 'client-2' }); + + assert.deepStrictEqual(watchRegistry.watchCalls, []); + assert.strictEqual(warnings.length, 1); + assert.match(warnings[0], /terminal_watch/); + assert.match(warnings[0], /foreign-sess/); + }); + + it('logs each distinct unknown session once, so one noisy session cannot mask another', () => { + const { router, warnings } = makeRouter(() => false); + + router.handleCommand({ type: 'terminal_watch', sessionId: 'foreign-a', clientId: 'client-1' }); + router.handleCommand({ type: 'terminal_watch', sessionId: 'foreign-b', clientId: 'client-1' }); + router.handleCommand({ type: 'terminal_watch', sessionId: 'foreign-a', clientId: 'client-1' }); + + assert.strictEqual(warnings.length, 2); + assert.ok(warnings.some((w) => w.includes('foreign-a'))); + assert.ok(warnings.some((w) => w.includes('foreign-b'))); + }); + + it('does not log for accepted watches or for terminal_unwatch of an unknown session', () => { + const { router, warnings } = makeRouter((sessionId) => sessionId === 'mine'); + + router.handleCommand({ type: 'terminal_watch', sessionId: 'mine', clientId: 'client-1' }); + router.handleCommand({ type: 'terminal_unwatch', sessionId: 'foreign-sess', clientId: 'client-1' }); + + assert.deepStrictEqual(warnings, []); + }); +}); diff --git a/bridge/src/direct-transport/publish-router.ts b/bridge/src/direct-transport/publish-router.ts index e860991..1d06976 100644 --- a/bridge/src/direct-transport/publish-router.ts +++ b/bridge/src/direct-transport/publish-router.ts @@ -28,8 +28,17 @@ export interface PublishRouterOptions { * to accepting all sessionIds. */ isKnownSession?: (sessionId: string) => boolean; + /** Injectable log sink for dropped watches (tests); defaults to console.warn. */ + warn?: (message: string) => void; } +/** + * Cap on remembered "already logged" sessionIds. Watch messages fan out to + * every bridge on commands:rpc, so a long-lived bridge on a busy account would + * otherwise accumulate one entry per foreign session forever. + */ +const UNKNOWN_LOG_CAP = 500; + /** * Implements R2: terminal output/screen always fan out to direct-attached peers; * they also go to Centrifugo iff the session has an unexpired remote watcher. @@ -42,6 +51,9 @@ export class PublishRouter { private readonly userId: string; private readonly loopback?: LoopbackPeerServerLike; private readonly isKnownSession: (sessionId: string) => boolean; + private readonly warn: (message: string) => void; + /** sessionIds already logged as unknown; keeps heartbeats from spamming. */ + private readonly unknownLogged = new Set(); constructor(options: PublishRouterOptions) { this.registry = options.registry; @@ -50,6 +62,7 @@ export class PublishRouter { this.userId = options.userId; this.loopback = options.loopback; this.isKnownSession = options.isKnownSession ?? (() => true); + this.warn = options.warn ?? ((message) => console.warn(message)); } /** R2 gating: a session is direct-attached if EITHER local rung has a peer. */ @@ -91,7 +104,10 @@ export class PublishRouter { if (typeof msg.sessionId !== 'string' || msg.sessionId === '') return; if (typeof msg.clientId !== 'string' || msg.clientId === '') return; if (msg.type === 'terminal_watch') { - if (!this.isKnownSession(msg.sessionId)) return; + if (!this.isKnownSession(msg.sessionId)) { + this.logUnknownWatch(msg.sessionId); + return; + } this.registry.watch(msg.sessionId, msg.clientId); } else { this.registry.unwatch(msg.sessionId, msg.clientId); @@ -101,4 +117,20 @@ export class PublishRouter { console.error('[DirectTransport] Failed to handle direct command:', err); } } + + /** + * A dropped watch is normal for another bridge's session, but for a session + * the user IS looking at it renders a permanently blank pane — and used to be + * silent. Log once per sessionId: watchers re-send terminal_watch every + * WATCH_HEARTBEAT_MS, so an unconditional line would repeat every 20s forever. + */ + private logUnknownWatch(sessionId: string): void { + if (this.unknownLogged.has(sessionId)) return; + if (this.unknownLogged.size >= UNKNOWN_LOG_CAP) this.unknownLogged.clear(); + this.unknownLogged.add(sessionId); + this.warn( + `[DirectTransport] Dropped terminal_watch for unknown session ${sessionId} ` + + '(no running process, no terminal buffer, no tmux session on this bridge)', + ); + } } diff --git a/bridge/src/index.ts b/bridge/src/index.ts index 9257219..26fd91a 100644 --- a/bridge/src/index.ts +++ b/bridge/src/index.ts @@ -489,16 +489,37 @@ program // not re-request its list on reconnect, so without this its session list // goes stale/empty after a Centrifugo blip until a page reload. onReconnect: async () => { + // Re-publishing EVERY session on each reconnect is a publish storm that + // (before the flapping fix) fed the command-echo backlog. Terminal + // records (completed/error) don't change and were already in the UI's + // list before the blip, so only live sessions need a re-push. Publish in + // bounded batches with a macrotask yield between them so a large account + // can't monopolize the event loop. A single publish's payload is + // unchanged. + const RESYNC_CHUNK = 10; const sessions = await store.listSessions(); - for (const session of sessions) { - await centrifugo.publishSessionUpdate(userId, session); + const live = sessions.filter( + (session) => session.status === 'running' || session.status === 'pending', + ); + for (let i = 0; i < live.length; i += RESYNC_CHUNK) { + await Promise.all( + live.slice(i, i + RESYNC_CHUNK).map((session) => + centrifugo.publishSessionUpdate(userId, session), + ), + ); + await new Promise((resolve) => setTimeout(resolve, 0)); } const loops = listLoops(); - for (const loop of loops) { - await centrifugo.publishLoopUpdate(userId, loop); + for (let i = 0; i < loops.length; i += RESYNC_CHUNK) { + await Promise.all( + loops.slice(i, i + RESYNC_CHUNK).map((loop) => + centrifugo.publishLoopUpdate(userId, loop), + ), + ); + await new Promise((resolve) => setTimeout(resolve, 0)); } console.log( - `[Bridge] Re-synced ${sessions.length} session(s) and ${loops.length} loop(s) after Centrifugo reconnect`, + `[Bridge] Re-synced ${live.length} session(s) and ${loops.length} loop(s) after Centrifugo reconnect`, ); }, }); @@ -540,13 +561,56 @@ program userId, // Watch messages fan out to every bridge on commands:rpc; only register // watchers for sessions this bridge actually serves terminal data for. - isKnownSession: (sid) => runner.isRunning(sid) || terminalManager.has(sid), + // hasTmuxSession is the third arm: a session can be alive in tmux with no + // PTY client in THIS process (agent-spawned re-run, adopted after a bridge + // restart, resurrection deferred). Those used to fail the guard, so the + // watch was dropped and every device rendered a blank pane. The tmux probe + // is a subprocess, so it runs last — the two in-memory checks short-circuit + // for every session this bridge already serves. + isKnownSession: (sid) => runner.isRunning(sid) || terminalManager.has(sid) || runner.hasTmuxSession(sid), }); + + // Accepting the watch is not enough on its own to make output flow. + // publishScreenDump serves entirely from terminalManager, which is fed by + // TerminalPump from runner 'data' events — there is no tmux capture-pane + // path — so a session alive in tmux with no PTY client here dumps an empty + // screen and then streams nothing. Attach one client through the SAME adopt + // path session resurrection uses (runner.reattach); tmux redraws the full + // screen to a joining client, so live output resumes within milliseconds and + // rides the already-attached pump. No new pump, no new tmux plumbing. + const watchReattachInFlight = new Set(); + const ensureWatchedSessionAttached = (sid: string): void => { + if (runner.isRunning(sid) || watchReattachInFlight.has(sid)) return; + if (!runner.hasTmuxSession(sid)) return; + watchReattachInFlight.add(sid); + void store.loadSession(sid) + .then((session) => { + // Only adopt what this store still considers live: a tmux session with + // no live record may belong to another bridge on this machine. + if (!session || (session.status !== 'running' && session.status !== 'pending')) return; + if (runner.isRunning(sid)) return; + if (!runner.reattach(sid, { + workingDir: session.workingDir, + parentSessionId: session.parentSessionId, + })) return; + // Idempotent: subscribeToTerminalInput no-ops on an existing channel. + wireTerminalInput(sid); + console.log(`[Bridge] Reattached tmux session ${sid} for a new terminal watcher`); + }) + .catch((err) => { + console.error(`[Bridge] Failed to reattach ${sid} for a terminal watcher:`, err); + }) + .finally(() => { watchReattachInFlight.delete(sid); }); + }; + // Every NEW remote watcher (first, each additional distinct client, or a // post-expiry re-registration) ⇒ push a full screen resync so the joining // Centrifugo-fallback client renders before incremental output (R1). The // dump is channel-wide; existing viewers re-render idempotently. - watchRegistry.onNewWatcher((sid) => publishScreenDump(sid)); + watchRegistry.onNewWatcher((sid) => { + ensureWatchedSessionAttached(sid); + publishScreenDump(sid); + }); // Bind the loopback WS upgrade handler onto the already-listening server. const loopbackHttpServer = localApiServer.getHttpServer(); diff --git a/bridge/src/tmux.test.ts b/bridge/src/tmux.test.ts new file mode 100644 index 0000000..5e9e009 --- /dev/null +++ b/bridge/src/tmux.test.ts @@ -0,0 +1,75 @@ +/** + * Cache behavior for hasTmuxSession: the has-session probe is a synchronous + * tmux subprocess run on every terminal_watch (and re-sent every ~20s by each + * watcher), so it must be served from a short-TTL cache — both true and false + * outcomes — rather than spawning tmux each time. Uses the injected test probe/ + * clock seams so no real tmux process is spawned. + */ +import { afterEach, describe, it } from 'node:test'; +import assert from 'node:assert/strict'; + +import { + hasTmuxSession, + invalidateTmuxSessionCache, + __setTmuxProbeForTest, + __resetTmuxProbeForTest, +} from './tmux.js'; + +afterEach(() => { + __resetTmuxProbeForTest(); +}); + +describe('hasTmuxSession negative/positive cache', () => { + it('probes tmux only once for repeated calls within the TTL', () => { + let calls = 0; + let clock = 1000; + __setTmuxProbeForTest({ probe: () => { calls += 1; return true; }, now: () => clock }); + + assert.strictEqual(hasTmuxSession('s1'), true); + assert.strictEqual(hasTmuxSession('s1'), true); + clock += 1999; // still inside the 2000ms window + assert.strictEqual(hasTmuxSession('s1'), true); + assert.strictEqual(calls, 1); + }); + + it('caches a false result too (foreign/dead session does not re-spawn tmux)', () => { + let calls = 0; + let clock = 1000; + __setTmuxProbeForTest({ probe: () => { calls += 1; return false; }, now: () => clock }); + + assert.strictEqual(hasTmuxSession('gone'), false); + assert.strictEqual(hasTmuxSession('gone'), false); + assert.strictEqual(calls, 1); + }); + + it('re-probes after the TTL expires', () => { + let calls = 0; + let clock = 1000; + __setTmuxProbeForTest({ probe: () => { calls += 1; return true; }, now: () => clock }); + + assert.strictEqual(hasTmuxSession('s1'), true); + clock += 2001; // past the 2000ms TTL + assert.strictEqual(hasTmuxSession('s1'), true); + assert.strictEqual(calls, 2); + }); + + it('keys the cache per sessionId', () => { + let calls = 0; + __setTmuxProbeForTest({ probe: () => { calls += 1; return true; }, now: () => 1000 }); + + hasTmuxSession('a'); + hasTmuxSession('b'); + hasTmuxSession('a'); + assert.strictEqual(calls, 2); + }); + + it('invalidateTmuxSessionCache forces the next call to re-probe', () => { + let calls = 0; + __setTmuxProbeForTest({ probe: () => { calls += 1; return true; }, now: () => 1000 }); + + hasTmuxSession('s1'); + invalidateTmuxSessionCache('s1'); + hasTmuxSession('s1'); + assert.strictEqual(calls, 2); + }); +}); diff --git a/bridge/src/tmux.ts b/bridge/src/tmux.ts index 5321f0d..2d022c8 100644 --- a/bridge/src/tmux.ts +++ b/bridge/src/tmux.ts @@ -147,9 +147,34 @@ export async function createTmuxSession(options: CreateTmuxSessionOptions): Prom args.push(`/bin/zsh -l -c ${shellQuote(inner)}`); await execFileAsync('tmux', args, { env: options.env }); + invalidateTmuxSessionCache(options.sessionId); } -export function hasTmuxSession(sessionId: string): boolean { +/** + * Negative + positive cache for the has-session probe. `isKnownSession` runs it + * on every terminal_watch, and watchers re-send terminal_watch every ~20s: a + * foreign session (alive on another bridge, or gone) misses both in-memory arms + * and would otherwise spawn `tmux has-session` — a synchronous, event-loop + * blocking subprocess — on every heartbeat. Caching both outcomes for a short + * TTL collapses that to at most one probe per session per window. Create/attach + * and kill invalidate the entry so liveness transitions are seen immediately; + * the TTL is the backstop for transitions that bypass those paths. + */ +const HAS_SESSION_CACHE_TTL_MS = 2000; + +interface HasSessionCacheEntry { + value: boolean; + expiresAt: number; +} + +const hasSessionCache = new Map(); + +// Test seams (production uses the defaults). `probe` performs the real tmux +// subprocess; `now` is the clock the TTL is measured against. +let hasSessionProbe: (sessionId: string) => boolean = probeTmuxSession; +let hasSessionNow: () => number = () => Date.now(); + +function probeTmuxSession(sessionId: string): boolean { try { execFileSync( 'tmux', @@ -162,6 +187,38 @@ export function hasTmuxSession(sessionId: string): boolean { } } +export function hasTmuxSession(sessionId: string): boolean { + const now = hasSessionNow(); + const cached = hasSessionCache.get(sessionId); + if (cached && cached.expiresAt > now) { + return cached.value; + } + const value = hasSessionProbe(sessionId); + hasSessionCache.set(sessionId, { value, expiresAt: now + HAS_SESSION_CACHE_TTL_MS }); + return value; +} + +/** Drop the cached liveness for a session (call on create/attach/kill). */ +export function invalidateTmuxSessionCache(sessionId: string): void { + hasSessionCache.delete(sessionId); +} + +/** Test-only: override the probe/clock and clear the cache. */ +export function __setTmuxProbeForTest( + hooks: { probe?: (sessionId: string) => boolean; now?: () => number } = {}, +): void { + if (hooks.probe) hasSessionProbe = hooks.probe; + if (hooks.now) hasSessionNow = hooks.now; + hasSessionCache.clear(); +} + +/** Test-only: restore production probe/clock and clear the cache. */ +export function __resetTmuxProbeForTest(): void { + hasSessionProbe = probeTmuxSession; + hasSessionNow = () => Date.now(); + hasSessionCache.clear(); +} + /** Session ids of all live ftown-* sessions on the dedicated socket. */ export function listFtownTmuxSessions(): string[] { try { @@ -180,6 +237,7 @@ export function listFtownTmuxSessions(): string[] { } export async function killTmuxSession(sessionId: string): Promise { + invalidateTmuxSessionCache(sessionId); try { await execFileAsync('tmux', [ '-L', TMUX_SOCKET_NAME, diff --git a/ui/src/hooks/useBridges.test.ts b/ui/src/hooks/useBridges.test.ts new file mode 100644 index 0000000..242e986 --- /dev/null +++ b/ui/src/hooks/useBridges.test.ts @@ -0,0 +1,123 @@ +import { describe, it, expect } from "vitest"; +import { dedupeBridges, applyBridgeJoin, applyBridgeLeave, type BridgeInfo } from "./useBridges"; + +const bridge = (overrides: Partial = {}): BridgeInfo => ({ + clientId: "client-1", + bridgeId: "bridge-1", + hostname: "host-1", + connectedAt: "2026-09-05T10:00:00.000Z", + ...overrides, +}); + +describe("dedupeBridges", () => { + it("collapses two clients sharing a bridgeId down to the one with the newest connectedAt", () => { + const older = bridge({ clientId: "old-client", connectedAt: "2026-09-05T09:00:00.000Z" }); + const newer = bridge({ clientId: "new-client", connectedAt: "2026-09-05T10:00:00.000Z" }); + + expect(dedupeBridges([older, newer])).toEqual([newer]); + // Order in the input shouldn't matter. + expect(dedupeBridges([newer, older])).toEqual([newer]); + }); + + it("keeps entries for different bridgeIds", () => { + const a = bridge({ clientId: "a", bridgeId: "bridge-a", connectedAt: "2026-09-05T09:00:00.000Z" }); + const b = bridge({ clientId: "b", bridgeId: "bridge-b", connectedAt: "2026-09-05T09:00:00.000Z" }); + + expect(dedupeBridges([b, a])).toEqual([a, b]); // sorted by bridgeId + }); + + it("falls back to the most-recently-seen entry when connectedAt is missing on both", () => { + const first = bridge({ clientId: "first", connectedAt: "" }); + const second = bridge({ clientId: "second", connectedAt: "" }); + + expect(dedupeBridges([first, second])).toEqual([second]); + }); + + it("falls back to the most-recently-seen entry when connectedAt ties exactly", () => { + const first = bridge({ clientId: "first", connectedAt: "2026-09-05T10:00:00.000Z" }); + const second = bridge({ clientId: "second", connectedAt: "2026-09-05T10:00:00.000Z" }); + + expect(dedupeBridges([first, second])).toEqual([second]); + }); + + it("prefers a present connectedAt over a missing one, regardless of order", () => { + const missing = bridge({ clientId: "missing", connectedAt: "" }); + const present = bridge({ clientId: "present", connectedAt: "2026-09-05T09:00:00.000Z" }); + + expect(dedupeBridges([present, missing])).toEqual([present]); + }); +}); + +describe("applyBridgeJoin", () => { + it("adds a new client that isn't known yet", () => { + const prev = [bridge({ bridgeId: "bridge-a", clientId: "a" })]; + const incoming = bridge({ bridgeId: "bridge-b", clientId: "b" }); + + expect(applyBridgeJoin(prev, incoming)).toEqual([prev[0], incoming]); + }); + + it("adds a second client for the same bridgeId rather than replacing the first (all known clients are kept)", () => { + const stale = bridge({ clientId: "old-client", connectedAt: "2026-09-05T09:00:00.000Z" }); + const fresh = bridge({ clientId: "new-client", connectedAt: "2026-09-05T10:00:00.000Z" }); + + const result = applyBridgeJoin([stale], fresh); + expect(result).toHaveLength(2); + expect(result).toEqual(expect.arrayContaining([stale, fresh])); + }); + + it("updates the entry in place when the same clientId re-announces", () => { + const original = bridge({ clientId: "client-1", hostname: "host-1" }); + const updated = bridge({ clientId: "client-1", hostname: "host-1-renamed" }); + + const result = applyBridgeJoin([original], updated); + expect(result).toEqual([updated]); + }); +}); + +describe("applyBridgeLeave", () => { + it("removes the entry whose clientId matches the leaving client", () => { + const entry = bridge({ clientId: "client-1" }); + expect(applyBridgeLeave([entry], "client-1")).toEqual([]); + }); + + it("leaves other known clients (including duplicates of the same bridgeId) untouched", () => { + const stale = bridge({ clientId: "old-client", connectedAt: "2026-09-05T09:00:00.000Z" }); + const fresh = bridge({ clientId: "new-client", connectedAt: "2026-09-05T10:00:00.000Z" }); + + expect(applyBridgeLeave([stale, fresh], "new-client")).toEqual([stale]); + }); + + it("leaves other bridges untouched", () => { + const a = bridge({ bridgeId: "bridge-a", clientId: "a" }); + const b = bridge({ bridgeId: "bridge-b", clientId: "b" }); + + expect(applyBridgeLeave([a, b], "a")).toEqual([b]); + }); +}); + +describe("presence lifecycle (applyBridgeJoin/applyBridgeLeave over the known-client set, exposed via dedupeBridges)", () => { + it("winner leaves while a stale duplicate remains: the bridge is still listed, via the remaining client", () => { + let allClients: BridgeInfo[] = []; + const stale = bridge({ clientId: "old-client", connectedAt: "2026-09-05T09:00:00.000Z" }); + const fresh = bridge({ clientId: "new-client", connectedAt: "2026-09-05T10:00:00.000Z" }); + + allClients = applyBridgeJoin(allClients, stale); + allClients = applyBridgeJoin(allClients, fresh); + expect(dedupeBridges(allClients)).toEqual([fresh]); // the winner is exposed + + // The winning connection's own leave arrives before the stale duplicate's + // presence timeout — the bridge must NOT vanish from the exposed list. + allClients = applyBridgeLeave(allClients, "new-client"); + expect(dedupeBridges(allClients)).toEqual([stale]); + }); + + it("last known client leaves: the bridge disappears from the exposed list", () => { + let allClients: BridgeInfo[] = []; + const stale = bridge({ clientId: "old-client" }); + + allClients = applyBridgeJoin(allClients, stale); + allClients = applyBridgeLeave(allClients, "old-client"); + + expect(dedupeBridges(allClients)).toEqual([]); + }); +}); diff --git a/ui/src/hooks/useBridges.ts b/ui/src/hooks/useBridges.ts index ab7c041..d1ccf24 100644 --- a/ui/src/hooks/useBridges.ts +++ b/ui/src/hooks/useBridges.ts @@ -15,9 +15,80 @@ interface UseBridgesResult { hasBridges: boolean; } +/** + * True when `candidate` should win over `current` for the same bridgeId: + * a strictly newer connectedAt wins outright, and a tie (equal or both + * missing/unparseable) falls back to "most recently seen" — i.e. whichever + * entry is being applied later, which is what callers pass as `candidate`. + * ISO 8601 timestamps compare correctly as strings, so no Date parsing is + * needed for the common case; anything else falls through to the tie rule. + */ +function winsOver(candidate: BridgeInfo, current: BridgeInfo): boolean { + const a = candidate.connectedAt || ""; + const b = current.connectedAt || ""; + return a >= b; +} + +/** + * Collapses the full set of KNOWN clients (every clientId currently present + * for every bridgeId, including stale duplicates from a reconnect whose old + * connection hasn't expired yet) down to at most one exposed entry per + * bridgeId — the winner per `winsOver`. This is a pure read-side projection: + * the underlying client set is never mutated by it, only the value shown to + * consumers (e.g. the dashboard header count) is. + * + * Pure and exported so it can be unit-tested without a Centrifuge client. + */ +export function dedupeBridges(entries: BridgeInfo[]): BridgeInfo[] { + const byBridgeId = new Map(); + for (const entry of entries) { + const existing = byBridgeId.get(entry.bridgeId); + if (!existing || winsOver(entry, existing)) { + byBridgeId.set(entry.bridgeId, entry); + } + } + return Array.from(byBridgeId.values()).sort((a, b) => a.bridgeId.localeCompare(b.bridgeId)); +} + +/** + * Pure reducer for a Centrifugo "join" event over the full known-client set: + * upserts `bridge` by clientId (adds it if new, updates it in place if the + * same clientId rejoins/re-announces). Deliberately does NOT dedupe by + * bridgeId here — a reconnecting bridge's stale duplicate must stay in the + * known-client set until its own "leave" arrives, so the bridge doesn't + * disappear from the exposed list (see `dedupeBridges`) if the winning + * client leaves first. Callers derive the exposed list by running the + * result through `dedupeBridges`. + */ +export function applyBridgeJoin(allClients: BridgeInfo[], bridge: BridgeInfo): BridgeInfo[] { + const existingIndex = allClients.findIndex((b) => b.clientId === bridge.clientId); + if (existingIndex === -1) { + return [...allClients, bridge]; + } + const next = [...allClients]; + next[existingIndex] = bridge; + return next; +} + +/** + * Pure reducer for a Centrifugo "leave" event over the full known-client + * set: removes only the entry whose clientId matches `clientId`. Other + * clients for the same bridgeId (e.g. a still-live duplicate connection) + * are left untouched, so the bridge keeps appearing in the exposed + * (`dedupeBridges`) list until its LAST known client leaves. + */ +export function applyBridgeLeave(allClients: BridgeInfo[], clientId: string): BridgeInfo[] { + return allClients.filter((b) => b.clientId !== clientId); +} + export function useBridges(client: Centrifuge | null, userId: string | null): UseBridgesResult { const [bridges, setBridges] = useState([]); const subRef = useRef(null); + // Every known client per bridgeId (not deduped) — the source of truth fed + // to dedupeBridges to produce the exposed `bridges` list. Kept in a ref + // (not state) since it's an internal accumulator; only the derived, + // deduped projection needs to trigger a re-render. + const allClientsRef = useRef([]); const fetchPresence = useCallback(async (sub: Subscription) => { try { @@ -33,16 +104,19 @@ export function useBridges(client: Centrifuge | null, userId: string | null): Us hostname: data.hostname ?? "unknown", connectedAt: data.connectedAt ?? "", }; - }) - .sort((a, b) => a.bridgeId.localeCompare(b.bridgeId)); - setBridges(bridgeList); + }); + // The presence snapshot is authoritative: replace the full known-client set. + allClientsRef.current = bridgeList; + setBridges(dedupeBridges(bridgeList)); } catch { + allClientsRef.current = []; setBridges([]); } }, []); useEffect(() => { if (!client || !userId) { + allClientsRef.current = []; setBridges([]); return; } @@ -71,14 +145,13 @@ export function useBridges(client: Centrifuge | null, userId: string | null): Us hostname: data.hostname ?? "unknown", connectedAt: data.connectedAt ?? "", }; - setBridges((prev) => { - if (prev.some((b) => b.clientId === bridge.clientId)) return prev; - return [...prev, bridge].sort((a, b) => a.bridgeId.localeCompare(b.bridgeId)); - }); + allClientsRef.current = applyBridgeJoin(allClientsRef.current, bridge); + setBridges(dedupeBridges(allClientsRef.current)); }); sub.on("leave", (ctx) => { - setBridges((prev) => prev.filter((b) => b.clientId !== ctx.info.client)); + allClientsRef.current = applyBridgeLeave(allClientsRef.current, ctx.info.client); + setBridges(dedupeBridges(allClientsRef.current)); }); sub.subscribe(); @@ -94,6 +167,7 @@ export function useBridges(client: Centrifuge | null, userId: string | null): Us sub.unsubscribe(); client.removeSubscription(sub); subRef.current = null; + allClientsRef.current = []; setBridges([]); }; }, [client, userId, fetchPresence]); diff --git a/ui/src/hooks/useSessions.test.ts b/ui/src/hooks/useSessions.test.ts new file mode 100644 index 0000000..abf1b2a --- /dev/null +++ b/ui/src/hooks/useSessions.test.ts @@ -0,0 +1,74 @@ +import { describe, it, expect } from "vitest"; +import { mergeSessionSnapshot } from "./useSessions"; +import type { Session } from "@/types"; + +function makeSession(overrides: Partial = {}): Session { + return { + id: "s1", + name: "session", + status: "running", + bridgeId: "bridge-1", + createdAt: "2024-01-01T00:00:00Z", + updatedAt: "2024-01-01T00:00:00Z", + ...overrides, + }; +} + +describe("mergeSessionSnapshot", () => { + it("keeps the existing bridgeId when the incoming row omits it", () => { + const current = makeSession({ id: "s1", bridgeId: "bridge-1" }); + const incoming = makeSession({ + id: "s1", + bridgeId: undefined, + } as unknown as Partial) as Session; + + const result = mergeSessionSnapshot(current, incoming); + expect(result.bridgeId).toBe("bridge-1"); + }); + + it("replaces bridgeId when the incoming row provides a new one", () => { + const current = makeSession({ id: "s1", bridgeId: "bridge-1" }); + const incoming = makeSession({ id: "s1", bridgeId: "bridge-2" }); + + const result = mergeSessionSnapshot(current, incoming); + expect(result.bridgeId).toBe("bridge-2"); + }); + + it("leaves a brand-new session without a bridgeId when there is no existing row to merge against", () => { + const incoming = makeSession({ + id: "s1", + bridgeId: undefined, + } as unknown as Partial) as Session; + + const result = mergeSessionSnapshot(undefined, incoming); + expect(result.bridgeId).toBeFalsy(); + }); + + it("preserves bridgeId across a full-list snapshot merge", () => { + const prev = [makeSession({ id: "s1", bridgeId: "bridge-1" })]; + const incomingList = [ + makeSession({ + id: "s1", + bridgeId: undefined, + } as unknown as Partial) as Session, + ]; + + const merged = new Map(prev.map((s) => [s.id, s])); + for (const s of incomingList) { + merged.set(s.id, mergeSessionSnapshot(merged.get(s.id), s)); + } + + expect(merged.get("s1")!.bridgeId).toBe("bridge-1"); + }); + + it("returns the incoming session unchanged when current is undefined", () => { + const incoming = makeSession({ + id: "s2", + name: "new-session", + bridgeId: "bridge-new", + }); + + const result = mergeSessionSnapshot(undefined, incoming); + expect(result).toEqual(incoming); + }); +}); diff --git a/ui/src/hooks/useSessions.ts b/ui/src/hooks/useSessions.ts index eac4513..7e22982 100644 --- a/ui/src/hooks/useSessions.ts +++ b/ui/src/hooks/useSessions.ts @@ -44,16 +44,21 @@ function isSessionUsage(value: unknown): value is SessionUsage { && typeof usage.collectedAt === "string"; } -function mergeSessionSnapshot(current: Session | undefined, incoming: Session): Session { +export function mergeSessionSnapshot(current: Session | undefined, incoming: Session): Session { + if (!current) return incoming; + let merged = incoming; if ( - current?.status === "running" + current.status === "running" && incoming.status === "running" && current.usage && (!incoming.usage || incoming.usage.collectedAt <= current.usage.collectedAt) ) { - return { ...incoming, usage: current.usage }; + merged = { ...merged, usage: current.usage }; } - return incoming; + if (!incoming.bridgeId && current.bridgeId) { + merged = { ...merged, bridgeId: current.bridgeId }; + } + return merged; } interface SessionUpdateMessage {