feat(tess): add roster-bound tmux fleet provider (#724)

packages/mosaic/src/fleet/tmux-runtime-transport.test.ts

@@ -0,0 +1,160 @@
+import { describe, expect, it, vi } from 'vitest';
+import type { CommandResult, CommandRunner, FleetRoster } from '../commands/fleet.js';
+import { FleetTmuxRuntimeTransport } from './tmux-runtime-transport.js';
+import type { FleetRuntimeTransportError } from './tmux-runtime-transport.js';
+
+const roster: FleetRoster = {
+  version: 1,
+  transport: 'tmux',
+  tmux: { socketName: 'tess-fleet', holderSession: '_holder' },
+  defaults: { workingDirectory: '~/src' },
+  runtimes: { codex: { resetCommand: '/clear' } },
+  agents: [{ name: 'coder0', runtime: 'codex', className: 'code' }],
+};
+
+function commandResult(stdout = '', exitCode = 0, stderr = ''): CommandResult {
+  return { stdout, stderr, exitCode };
+}
+
+describe('FleetTmuxRuntimeTransport security boundary', (): void => {
+  it('rejects a prefix target before it invokes tmux', async (): Promise<void> => {
+    const runner = vi.fn<CommandRunner>(async (): Promise<CommandResult> => commandResult());
+    const transport = new FleetTmuxRuntimeTransport({
+      rosterLoader: async (): Promise<FleetRoster> => roster,
+      runner,
+      mosaicHome: '/mosaic',
+    });
+
+    await expect(transport.verifySession('coder')).rejects.toMatchObject({
+      code: 'not_found',
+    } satisfies Partial<FleetRuntimeTransportError>);
+    expect(runner).not.toHaveBeenCalled();
+  });
+
+  it('uses the roster socket and exact pane target while verifying peer identity', async (): Promise<void> => {
+    const runner = vi.fn<CommandRunner>(
+      async (): Promise<CommandResult> => commandResult('111 codex 0 0 0 0\n'),
+    );
+    const transport = new FleetTmuxRuntimeTransport({
+      rosterLoader: async (): Promise<FleetRoster> => roster,
+      runner,
+      mosaicHome: '/mosaic',
+    });
+
+    await expect(transport.verifySession('coder0')).resolves.toEqual({
+      id: 'coder0',
+      runtimeId: 'codex',
+      socketName: 'tess-fleet',
+    });
+    expect(runner).toHaveBeenCalledWith('tmux', [
+      '-L',
+      'tess-fleet',
+      'list-panes',
+      '-t',
+      '=coder0:0.0',
+      '-F',
+      '#{pane_pid} #{pane_current_command} #{pane_dead} #{pane_activity} #{window_activity} #{session_activity}',
+    ]);
+  });
+
+  it('denies a live pane whose runtime does not match the roster identity', async (): Promise<void> => {
+    const runner = vi.fn<CommandRunner>(
+      async (): Promise<CommandResult> => commandResult('111 python3 0 0 0 0\n'),
+    );
+    const transport = new FleetTmuxRuntimeTransport({
+      rosterLoader: async (): Promise<FleetRoster> => roster,
+      runner,
+      mosaicHome: '/mosaic',
+    });
+
+    await expect(transport.verifySession('coder0')).rejects.toMatchObject({
+      code: 'forbidden',
+    } satisfies Partial<FleetRuntimeTransportError>);
+    expect(runner).toHaveBeenCalledTimes(1);
+  });
+
+  it('surfaces a runtime identity mismatch instead of hiding it from fleet status', async (): Promise<void> => {
+    const rosterWithMismatchedPeer: FleetRoster = {
+      ...roster,
+      agents: [...roster.agents, { name: 'coder1', runtime: 'codex', className: 'code' }],
+    };
+    const runner = vi.fn<CommandRunner>(
+      async (_command: string, args: string[]): Promise<CommandResult> =>
+        commandResult(
+          args.includes('=coder1:0.0') ? '222 python3 0 0 0 0\n' : '111 codex 0 0 0 0\n',
+        ),
+    );
+    const transport = new FleetTmuxRuntimeTransport({
+      rosterLoader: async (): Promise<FleetRoster> => rosterWithMismatchedPeer,
+      runner,
+      mosaicHome: '/mosaic',
+    });
+
+    await expect(transport.listSessions()).rejects.toMatchObject({
+      code: 'forbidden',
+    } satisfies Partial<FleetRuntimeTransportError>);
+  });
+
+  it('denies sending when the roster socket does not contain the exact target', async (): Promise<void> => {
+    const runner = vi.fn<CommandRunner>(
+      async (): Promise<CommandResult> => commandResult('', 1, "can't find session: coder0"),
+    );
+    const transport = new FleetTmuxRuntimeTransport({
+      rosterLoader: async (): Promise<FleetRoster> => roster,
+      runner,
+      mosaicHome: '/mosaic',
+    });
+
+    await expect(transport.sendMessage('coder0', 'hello', 'tess')).rejects.toMatchObject({
+      code: 'unavailable',
+    } satisfies Partial<FleetRuntimeTransportError>);
+    expect(runner).toHaveBeenCalledTimes(1);
+  });
+
+  it('sends only through the maintained sender after exact target and identity verification', async (): Promise<void> => {
+    const runner = vi
+      .fn<CommandRunner>()
+      .mockResolvedValueOnce(commandResult('111 codex 0 0 0 0\n'))
+      .mockResolvedValueOnce(commandResult());
+    const transport = new FleetTmuxRuntimeTransport({
+      rosterLoader: async (): Promise<FleetRoster> => roster,
+      runner,
+      mosaicHome: '/mosaic',
+    });
+
+    await transport.sendMessage('coder0', 'hello', 'tess');
+
+    expect(runner).toHaveBeenNthCalledWith(2, '/mosaic/tools/tmux/agent-send.sh', [
+      '-L',
+      'tess-fleet',
+      '-S',
+      'tess',
+      '-s',
+      'coder0',
+      '-m',
+      'hello',
+    ]);
+  });
+
+  it('terminates only the exact roster target after identity verification', async (): Promise<void> => {
+    const runner = vi
+      .fn<CommandRunner>()
+      .mockResolvedValueOnce(commandResult('111 codex 0 0 0 0\n'))
+      .mockResolvedValueOnce(commandResult());
+    const transport = new FleetTmuxRuntimeTransport({
+      rosterLoader: async (): Promise<FleetRoster> => roster,
+      runner,
+      mosaicHome: '/mosaic',
+    });
+
+    await transport.terminate('coder0');
+
+    expect(runner).toHaveBeenNthCalledWith(2, 'tmux', [
+      '-L',
+      'tess-fleet',
+      'kill-session',
+      '-t',
+      '=coder0',
+    ]);
+  });
+});

packages/mosaic/src/fleet/tmux-runtime-transport.ts

@@ -0,0 +1,200 @@
+import {
+  buildAgentSendCommand,
+  buildTmuxListPanesCommand,
+  getRosterAgent,
+  loadFleetRoster,
+  parseTmuxListPanes,
+  resolveFleetPaths,
+  RUNTIME_ACCEPTABLE_COMMANDS,
+  socketArgs,
+  type CommandResult,
+  type CommandRunner,
+  type FleetRoster,
+} from '../commands/fleet.js';
+
+export type FleetRuntimeTransportErrorCode =
+  | 'forbidden'
+  | 'invalid_request'
+  | 'not_found'
+  | 'unavailable';
+
+/** A roster-bound, verified tmux target. It never accepts a caller-selected socket. */
+export interface FleetRuntimeTarget {
+  id: string;
+  runtimeId: string;
+  socketName: string;
+}
+
+/** Narrow transport boundary consumed by the runtime provider. */
+export interface FleetRuntimeTransport {
+  verifySession(sessionId: string): Promise<FleetRuntimeTarget>;
+  listSessions(): Promise<FleetRuntimeTarget[]>;
+  sendMessage(sessionId: string, message: string, sourceLabel: string): Promise<void>;
+  terminate(sessionId: string): Promise<void>;
+}
+
+export interface FleetTmuxRuntimeTransportOptions {
+  mosaicHome: string;
+  rosterPath?: string;
+  rosterLoader?: () => Promise<FleetRoster>;
+  runner: CommandRunner;
+}
+
+/** A typed, fail-closed error for roster, target, identity, and tmux failures. */
+export class FleetRuntimeTransportError extends Error {
+  constructor(
+    readonly code: FleetRuntimeTransportErrorCode,
+    message: string,
+  ) {
+    super(message);
+    this.name = FleetRuntimeTransportError.name;
+  }
+}
+
+/**
+ * Roster-bound transport for the local fleet tmux server. Every side-effecting
+ * operation verifies an exact roster name, the roster socket, and the runtime
+ * command in the exact pane before it invokes the maintained sender or tmux.
+ */
+export class FleetTmuxRuntimeTransport implements FleetRuntimeTransport {
+  private readonly rosterLoader: () => Promise<FleetRoster>;
+
+  constructor(private readonly options: FleetTmuxRuntimeTransportOptions) {
+    this.rosterLoader =
+      options.rosterLoader ??
+      (() =>
+        loadFleetRoster(options.rosterPath ?? resolveFleetPaths(options.mosaicHome).rosterPath));
+  }
+
+  async verifySession(sessionId: string): Promise<FleetRuntimeTarget> {
+    const roster = await this.loadRoster();
+    return this.verifyRosterSession(roster, sessionId);
+  }
+
+  async listSessions(): Promise<FleetRuntimeTarget[]> {
+    const roster = await this.loadRoster();
+    const targets = await Promise.all(
+      roster.agents.map(async (agent): Promise<FleetRuntimeTarget | undefined> => {
+        try {
+          return await this.verifyRosterSession(roster, agent.name);
+        } catch (error: unknown) {
+          if (error instanceof FleetRuntimeTransportError && error.code === 'unavailable') {
+            return undefined;
+          }
+          throw error;
+        }
+      }),
+    );
+    return targets.filter((target): target is FleetRuntimeTarget => target !== undefined);
+  }
+
+  async sendMessage(sessionId: string, message: string, sourceLabel: string): Promise<void> {
+    if (message.length === 0) {
+      throw new FleetRuntimeTransportError('invalid_request', 'Fleet message content is required');
+    }
+    if (sourceLabel.length === 0) {
+      throw new FleetRuntimeTransportError(
+        'invalid_request',
+        'Fleet message source label is required',
+      );
+    }
+    const target = await this.verifySession(sessionId);
+    const command = buildAgentSendCommand(
+      resolveFleetPaths(this.options.mosaicHome),
+      target.id,
+      message,
+      target.socketName,
+      sourceLabel,
+    );
+    await this.runChecked(command, 'Fleet message delivery failed');
+  }
+
+  async terminate(sessionId: string): Promise<void> {
+    const target = await this.verifySession(sessionId);
+    await this.runChecked(
+      ['tmux', ...socketArgs(target.socketName), 'kill-session', '-t', `=${target.id}`],
+      'Fleet session termination failed',
+    );
+  }
+
+  private async loadRoster(): Promise<FleetRoster> {
+    try {
+      return await this.rosterLoader();
+    } catch (error: unknown) {
+      const message = error instanceof Error ? error.message : 'Fleet roster is unavailable';
+      throw new FleetRuntimeTransportError(
+        'unavailable',
+        `Fleet roster is unavailable: ${message}`,
+      );
+    }
+  }
+
+  private async verifyRosterSession(
+    roster: FleetRoster,
+    sessionId: string,
+  ): Promise<FleetRuntimeTarget> {
+    const agent = this.exactRosterAgent(roster, sessionId);
+    const socketName = roster.tmux.socketName;
+    const result = await this.run(buildTmuxListPanesCommand(agent.name, socketName));
+    if (result.exitCode !== 0) {
+      throw new FleetRuntimeTransportError(
+        'unavailable',
+        `Fleet session is unavailable: ${agent.name}`,
+      );
+    }
+    const pane = parseTmuxListPanes(result.stdout);
+    if (pane.dead || pane.command === null) {
+      throw new FleetRuntimeTransportError(
+        'unavailable',
+        `Fleet session is unavailable: ${agent.name}`,
+      );
+    }
+    if (!hasExactRuntimeIdentity(agent.runtime, pane.command)) {
+      throw new FleetRuntimeTransportError(
+        'forbidden',
+        `Fleet runtime identity mismatch: ${agent.name}`,
+      );
+    }
+    return { id: agent.name, runtimeId: agent.runtime, socketName };
+  }
+
+  private exactRosterAgent(roster: FleetRoster, sessionId: string): FleetRoster['agents'][number] {
+    try {
+      const agent = getRosterAgent(roster, sessionId);
+      if (agent.name !== sessionId) {
+        throw new FleetRuntimeTransportError('not_found', 'Fleet session target must be exact');
+      }
+      return agent;
+    } catch (error: unknown) {
+      if (error instanceof FleetRuntimeTransportError) {
+        throw error;
+      }
+      throw new FleetRuntimeTransportError('not_found', 'Fleet session target is not roster-bound');
+    }
+  }
+
+  private async run(command: string[]): Promise<CommandResult> {
+    const [executable, ...args] = command;
+    if (executable === undefined) {
+      throw new FleetRuntimeTransportError('invalid_request', 'Fleet command is required');
+    }
+    try {
+      return await this.options.runner(executable, args);
+    } catch (error: unknown) {
+      const message = error instanceof Error ? error.message : 'Fleet command failed';
+      throw new FleetRuntimeTransportError('unavailable', message);
+    }
+  }
+
+  private async runChecked(command: string[], failureMessage: string): Promise<void> {
+    const result = await this.run(command);
+    if (result.exitCode !== 0) {
+      throw new FleetRuntimeTransportError('unavailable', failureMessage);
+    }
+  }
+}
+
+function hasExactRuntimeIdentity(runtimeId: string, paneCommand: string): boolean {
+  const expectedCommands = RUNTIME_ACCEPTABLE_COMMANDS[runtimeId];
+  return expectedCommands !== undefined && expectedCommands.includes(paneCommand);
+}