Per-session token usage + cursor startup fix + E2E suite expansion

.github/workflows/e2e.yml

@@ -2,6 +2,8 @@ name: e2e
 
 on:
   pull_request:
+  push:
+    branches: [main]
   workflow_dispatch:
 
 jobs:
@@ -14,6 +16,18 @@ jobs:
       - uses: actions/setup-node@v4
         with:
           node-version: 24
+          cache: 'npm'
+          cache-dependency-path: |
+            ui/package-lock.json
+            bridge/package-lock.json
+            e2e/package-lock.json
+
+      - name: Cache Playwright browsers
+        id: playwright-cache
+        uses: actions/cache@v4
+        with:
+          path: ~/.cache/ms-playwright
+          key: ${{ runner.os }}-playwright-${{ hashFiles('e2e/package-lock.json') }}
 
       # The bridge spawns sessions with a hardcoded `/bin/zsh -l`; tmux is the
       # preferred PTY runtime (falls back to direct node-pty if absent).
@@ -62,10 +76,16 @@ jobs:
           npm ci
           npm run build
 
-      - name: Install Playwright chromium
+      - name: Install Playwright chromium (browsers + OS deps)
+        if: steps.playwright-cache.outputs.cache-hit != 'true'
         working-directory: e2e
         run: npx playwright install --with-deps chromium
 
+      - name: Install Playwright OS deps only (browsers cached)
+        if: steps.playwright-cache.outputs.cache-hit == 'true'
+        working-directory: e2e
+        run: npx playwright install-deps chromium
+
       - name: Start UI + bridge (recorded PIDs, scratch HOME)
         run: bash e2e/scripts/start-services.sh
 

bridge/src/centrifugo-client.test.ts

@@ -0,0 +1,175 @@
+import { describe, it, beforeEach, afterEach } from 'node:test';
+import assert from 'node:assert/strict';
+
+import { CentrifugoClient } from './centrifugo-client.js';
+import type { Command } from './types.js';
+
+/**
+ * Publisher-identity guard tests for the two inbound client-publish handlers.
+ *
+ * Threat recap: Centrifugo gates SUBSCRIBE but not PUBLISH on user-limited
+ * channels, so any authenticated user can publish to another user's
+ * `commands:rpc#{owner}` / `terminal-input:{sid}#{owner}`. The bridge must act
+ * ONLY on publications whose authenticated publisher (`ctx.info.user`) equals the
+ * channel owner, and fail CLOSED when that identity is missing/empty.
+ */
+
+const OWNER = 'owner@example.com';
+const ATTACKER = 'attacker@evil.com';
+
+interface FakePublicationCtx {
+  data: unknown;
+  info?: { user?: string };
+}
+
+/**
+ * Build a CentrifugoClient whose underlying Centrifuge is replaced by a stub that
+ * hands back a fake subscription. The fake records the 'publication' handler the
+ * client registers so a test can invoke it with a fabricated PublicationContext.
+ */
+function makeClientWithFakeSub(): {
+  client: CentrifugoClient;
+  publish: (ctx: FakePublicationCtx) => void;
+} {
+  const client = new CentrifugoClient('ws://127.0.0.1:0/connection/websocket', 'tok', async () => 'tok');
+
+  let publicationHandler: ((ctx: FakePublicationCtx) => void) | undefined;
+  const fakeSub = {
+    on(event: string, cb: (ctx: FakePublicationCtx) => void) {
+      if (event === 'publication') publicationHandler = cb;
+    },
+    subscribe() {},
+    unsubscribe() {},
+  };
+
+  (client as unknown as { client: { newSubscription: () => unknown } }).client = {
+    newSubscription: () => fakeSub,
+  };
+
+  return {
+    client,
+    publish: (ctx: FakePublicationCtx) => {
+      if (!publicationHandler) throw new Error('publication handler was never registered');
+      publicationHandler(ctx);
+    },
+  };
+}
+
+describe('CentrifugoClient.subscribeToCommands — publisher-identity guard (fail-closed)', () => {
+  let warnCalls = 0;
+  const originalWarn = console.warn;
+
+  beforeEach(() => {
+    warnCalls = 0;
+    console.warn = () => {
+      warnCalls += 1;
+    };
+  });
+
+  afterEach(() => {
+    console.warn = originalWarn;
+  });
+
+  it('dispatches a command published by the channel owner', () => {
+    const { client, publish } = makeClientWithFakeSub();
+    const received: Command[] = [];
+    client.subscribeToCommands(OWNER, (c) => received.push(c));
+
+    publish({ data: { type: 'list_sessions', requestId: 'r1', payload: {} }, info: { user: OWNER } });
+
+    assert.strictEqual(received.length, 1);
+    assert.strictEqual(received[0].requestId, 'r1');
+    assert.strictEqual(warnCalls, 0);
+  });
+
+  it('drops a command published by another user (cross-tenant RCE close)', () => {
+    const { client, publish } = makeClientWithFakeSub();
+    const received: Command[] = [];
+    client.subscribeToCommands(OWNER, (c) => received.push(c));
+
+    publish({ data: { type: 'create_session', requestId: 'r2', payload: {} }, info: { user: ATTACKER } });
+
+    assert.strictEqual(received.length, 0);
+    assert.strictEqual(warnCalls, 1);
+  });
+
+  it('drops a command when publisher info is absent (fail-closed)', () => {
+    const { client, publish } = makeClientWithFakeSub();
+    const received: Command[] = [];
+    client.subscribeToCommands(OWNER, (c) => received.push(c));
+
+    publish({ data: { type: 'create_session', requestId: 'r3', payload: {} } });
+
+    assert.strictEqual(received.length, 0);
+  });
+
+  it('drops a command when publisher info.user is empty (fail-closed)', () => {
+    const { client, publish } = makeClientWithFakeSub();
+    const received: Command[] = [];
+    client.subscribeToCommands(OWNER, (c) => received.push(c));
+
+    publish({ data: { type: 'create_session', requestId: 'r4', payload: {} }, info: { user: '' } });
+
+    assert.strictEqual(received.length, 0);
+  });
+});
+
+describe('CentrifugoClient.subscribeToTerminalInput — publisher-identity guard (fail-closed)', () => {
+  const originalWarn = console.warn;
+
+  beforeEach(() => {
+    console.warn = () => {};
+  });
+
+  afterEach(() => {
+    console.warn = originalWarn;
+  });
+
+  it('delivers input published by the channel owner', () => {
+    const { client, publish } = makeClientWithFakeSub();
+    const inputs: Array<{ sid: string; data: string }> = [];
+    client.subscribeToTerminalInput(
+      OWNER,
+      'sess-1',
+      (sid, data) => inputs.push({ sid, data }),
+      () => {},
+      () => {},
+    );
+
+    publish({ data: { type: 'input', data: 'ls\n' }, info: { user: OWNER } });
+
+    assert.deepStrictEqual(inputs, [{ sid: 'sess-1', data: 'ls\n' }]);
+  });
+
+  it('drops keystroke injection from another user', () => {
+    const { client, publish } = makeClientWithFakeSub();
+    const inputs: Array<{ sid: string; data: string }> = [];
+    client.subscribeToTerminalInput(
+      OWNER,
+      'sess-1',
+      (sid, data) => inputs.push({ sid, data }),
+      () => {},
+      () => {},
+    );
+
+    publish({ data: { type: 'input', data: 'rm -rf ~\n' }, info: { user: ATTACKER } });
+
+    assert.strictEqual(inputs.length, 0);
+  });
+
+  it('drops input when publisher info is absent (fail-closed)', () => {
+    const { client, publish } = makeClientWithFakeSub();
+    const inputs: Array<{ sid: string; data: string }> = [];
+    client.subscribeToTerminalInput(
+      OWNER,
+      'sess-1',
+      (sid, data) => inputs.push({ sid, data }),
+      () => {},
+      () => {},
+    );
+
+    publish({ data: { type: 'input', data: 'whoami\n' } });
+
+    assert.strictEqual(inputs.length, 0);
+  });
+});

bridge/src/centrifugo-client.ts

@@ -22,6 +22,29 @@ type DirectCommandHandler = (msg: DirectCommandMessage) => void;
 
 const MAX_PUBLISH_BYTES = 460_000;
 
+/**
+ * Fail-closed publisher-identity check for inbound client publications on a
+ * user-limited channel `ch#{owner}`.
+ *
+ * Threat: Centrifugo's `allow_user_limited_channels` gates SUBSCRIBE only, not
+ * PUBLISH; with `allow_publish_for_client:true` any authenticated user can
+ * PUBLISH to another user's `commands:rpc#{owner}` / `terminal-input:{sid}#{owner}`.
+ * The bridge executes commands and applies keystrokes from those channels, so an
+ * attacker's publish would be cross-tenant RCE / keystroke injection.
+ *
+ * Defense: Centrifugo attaches the PUBLISHER's authenticated identity to every
+ * client publication as `ctx.info.user` (ClientInfo.user). The bridge and the
+ * owner's UI both connect with token `sub == owner`, so their legitimate
+ * publishes carry `info.user == owner`. We accept ONLY those and drop everything
+ * else — a mismatched publisher, and (fail-closed) any publication whose info is
+ * missing/empty (e.g. a server-API publish with no client context, or an empty
+ * anonymous user). `owner` is always a non-empty email, so the empty-string
+ * anonymous user can never match.
+ */
+function isOwnerPublication(ctx: PublicationContext, owner: string): boolean {
+  return owner.length > 0 && ctx.info?.user === owner;
+}
+
 function byteLen(str: string): number {
   return Buffer.byteLength(str, 'utf8');
 }
@@ -198,6 +221,15 @@ export class CentrifugoClient {
     const sub = this.client.newSubscription(channel);
 
     sub.on('publication', (ctx: PublicationContext) => {
+      // Fail-closed: only the channel owner (info.user == userId) may drive this
+      // terminal. Drop keystroke/resize/init from any other publisher — see
+      // isOwnerPublication for the cross-user-publish threat.
+      if (!isOwnerPublication(ctx, userId)) {
+        console.warn(
+          `[Centrifugo] Dropped foreign publication on ${channel} from user="${ctx.info?.user ?? ''}"`,
+        );
+        return;
+      }
       const msg = ctx.data as { type: string; data?: string; cols?: number; rows?: number };
       if (msg.type === 'input' && msg.data !== undefined) {
         onInput(sessionId, msg.data);
@@ -226,6 +258,18 @@ export class CentrifugoClient {
     const sub = this.client.newSubscription(channel);
 
     sub.on('publication', (ctx: PublicationContext) => {
+      // Fail-closed RCE close: execute commands ONLY from the channel owner
+      // (info.user == userId). The bridge's own command_response/signal echoes
+      // and the owner UI's commands all carry info.user == userId and pass; a
+      // foreign publisher's create_session/kill/bridge_exec is dropped here.
+      // See isOwnerPublication for the cross-user-publish threat. This filters
+      // strictly on publisher identity, never on message type.
+      if (!isOwnerPublication(ctx, userId)) {
+        console.warn(
+          `[Centrifugo] Dropped foreign command publication on ${channel} from user="${ctx.info?.user ?? ''}"`,
+        );
+        return;
+      }
       const data = ctx.data as Record<string, unknown>;
       if (data.type === 'command_response') {
         return;

bridge/src/command-rpc.ts

@@ -16,6 +16,7 @@ import type {
   DeleteLoopPayload,
   GetHistoryPayload,
   GetLoopRunsPayload,
+  GetSessionUsagePayload,
   RemoveSessionPayload,
   RenameSessionPayload,
   RunLoopNowPayload,
@@ -204,6 +205,20 @@ export function createCommandHandler(deps: CommandRpcDeps): (command: Command) =
           break;
         }
 
+        case 'get_session_usage': {
+          const payload = command.payload as GetSessionUsagePayload;
+          if (!payload.sessionId) {
+            response = { requestId: command.requestId, success: false, error: 'Missing sessionId' };
+            break;
+          }
+
+          const result = await sessionController.usage(payload.sessionId);
+          response = result.ok
+            ? { requestId: command.requestId, success: true, data: { usage: result.usage } }
+            : { requestId: command.requestId, success: false, error: result.message };
+          break;
+        }
+
         case 'create_loop': {
           const payload = command.payload as CreateLoopPayload;
           // bridgeId is forced to THIS bridge inside the controller (the routing

bridge/src/create-ftown-session.ts

@@ -4,7 +4,8 @@ import { basename, resolve } from 'node:path';
 
 import { buildSessionCommand } from './agent-commands.js';
 import { ensureCodexWorkdirTrust } from './codex-installer.js';
-import { harnessAcceptsPromptAsCliArg } from './harness-registry.js';
+import { HARNESSES, harnessAcceptsPromptAsCliArg, type HarnessSpec } from './harness-registry.js';
+import { staggerSpawn } from './spawn-stagger.js';
 import { PROVIDER_AUTH_ENV, PROVIDER_RUNTIME_ENV, loadProviderEnv } from './provider-env-store.js';
 import { registerSessionWorkspace } from './session-registry.js';
 
@@ -231,6 +232,23 @@ export function assertProviderAuthAvailable(
   if (missing) throw missing;
 }
 
+/**
+ * Per-harness spawn stagger, applied immediately before every runner.run
+ * (create AND relaunch). Harnesses with spawnStaggerMs set (currently cursor,
+ * whose startup token refresh races a macOS Keychain write under concurrent
+ * spawns) get their spawns serialized at that gap. NOTE: this delays the
+ * create/relaunch response by up to queueDepth × spawnStaggerMs — intended.
+ */
+async function staggerHarnessSpawn(shellType: ShellType | undefined): Promise<void> {
+  if (!shellType) return;
+  // Widen: the `as const` registry narrows away optional fields it doesn't set.
+  const spec: HarnessSpec | undefined = HARNESSES[shellType];
+  const staggerMs = spec?.spawnStaggerMs;
+  if (staggerMs) {
+    await staggerSpawn(`spawn:${shellType}`, staggerMs);
+  }
+}
+
 /** The stored-session fields relaunch derivation needs — satisfied by both live records and tombstones. */
 export type RelaunchCommandSource = Pick<
   Session,
@@ -325,6 +343,8 @@ export async function relaunchFtownSession(
   await deps.store.saveSession(session);
   await deps.centrifugo.publishSessionUpdate(deps.userId, session);
 
+  await staggerHarnessSpawn(session.shellType);
+
   deps.runner.run(session.id, command, {
     workingDir: session.workingDir,
     env: session.env,
@@ -506,6 +526,8 @@ export async function createFtownSession(
     ensureCodexWorkdirTrust(workingDir ?? process.cwd());
   }
 
+  await staggerHarnessSpawn(effectiveInput.shellType);
+
   deps.runner.run(sessionId, launchCommand, {
     workingDir,
     env: sessionEnv,