fix(dashboard): blank agent-spawned terminals, inflated bridge count, Centrifugo reconnect flapping

bridge/package.json

@@ -1,6 +1,6 @@
 {
   "name": "ftown-bridge",
-  "version": "0.19.26",
+  "version": "0.19.27",
   "description": "CLI bridge for ftown — generic PTY-over-Centrifugo relay",
   "type": "module",
   "main": "dist/index.js",

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<void> {
@@ -191,14 +207,16 @@ export class CentrifugoClient {
 
   async publishTerminalData(userId: string, sessionId: string, data: string): Promise<void> {
     const channel = `terminal:${sessionId}#${userId}`;
-    if (!this.subscriptions.has(channel)) {
-      const sub = this.client.newSubscription(channel);
-      this.subscriptions.set(channel, sub);
-      await new Promise<void>((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) {

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, []);
+  });
+});

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<string>();
 
   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)',
+    );
+  }
 }

bridge/src/index.ts

@@ -481,16 +481,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`,
         );
       },
     });
@@ -532,13 +553,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<string>();
+    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();