stack/packages/mosaic/src/fleet/promotion-transport.ts

import { readFile } from 'node:fs/promises';
import {
  buildTmuxListPanesCommand,
  getRosterAgent,
  parseTmuxListPanes,
  resolveFleetPaths,
  type CommandResult,
  type CommandRunner,
  type FleetRoster,
  RUNTIME_ACCEPTABLE_COMMANDS,
  socketArgs,
} from '../commands/fleet.js';
import { loadFleetRoster } from './fleet-roster-v1.js';

const PROMOTION_COMMAND = '/mosaic-promote';
const SESSION_ID_PATTERN = /^[a-f0-9]{64}$/;
const TRANSPORT_COMMAND_TIMEOUT_MS = 5_000;

export interface PromotionTarget {
  bundle: string;
  seat: string;
  sessionId: string;
}

export interface PromotionTransport {
  resolve(seat: string): Promise<PromotionTarget>;
  sendPromotion(target: PromotionTarget): Promise<void>;
}

export interface TmuxPromotionTransportOptions {
  environmentReader?: (pid: number) => Promise<string>;
  childrenReader?: (pid: number) => Promise<number[]>;
  mosaicHome: string;
  rosterLoader?: () => Promise<FleetRoster>;
  runner: CommandRunner;
}

// The launcher runs the runtime as a spawnSync CHILD of node(mosaic) (see
// launch.ts:99 — deliberate, so the parent survives to propagate signals), so
// MOSAIC_LEASE_SESSION_ID lives on the claude child, NOT on the tmux pane's root
// pid. Bound the descendant search so a hung/large process tree can't stall it.
const MAX_SUBTREE_PIDS = 128;

/** Local, roster-bound transport for the in-seat promotion command. */
export class TmuxPromotionTransport implements PromotionTransport {
  private readonly environmentReader: (pid: number) => Promise<string>;
  private readonly childrenReader: (pid: number) => Promise<number[]>;
  private readonly rosterLoader: () => Promise<FleetRoster>;

  constructor(private readonly options: TmuxPromotionTransportOptions) {
    this.environmentReader = options.environmentReader ?? readPaneEnvironment;
    this.childrenReader = options.childrenReader ?? readChildPids;
    this.rosterLoader =
      options.rosterLoader ??
      (() => loadFleetRoster(resolveFleetPaths(options.mosaicHome).rosterPath));
  }

  async resolve(seat: string): Promise<PromotionTarget> {
    const roster = await this.rosterLoader();
    const agent = getRosterAgent(roster, seat);
    if (agent.runtime !== 'claude') {
      throw new Error(`Lease promotion is currently available only for Claude seats: ${seat}.`);
    }
    const paneResult = await this.run(
      buildTmuxListPanesCommand(agent.name, roster.tmux.socketName),
    );
    if (paneResult.exitCode !== 0) {
      throw new Error(`Promotion seat is unavailable: ${seat}.`);
    }
    const pane = parseTmuxListPanes(paneResult.stdout);
    const allowedCommands = RUNTIME_ACCEPTABLE_COMMANDS.claude;
    if (
      pane.dead ||
      pane.pid === null ||
      pane.command === null ||
      allowedCommands === undefined ||
      !allowedCommands.includes(pane.command)
    ) {
      throw new Error(`Promotion seat runtime identity mismatch: ${seat}.`);
    }
    const sessionId = await this.resolveLeaseSessionId(pane.pid);
    if (sessionId === null) {
      throw new Error(`Promotion seat has no readable lease session: ${seat}.`);
    }
    return {
      bundle: roster.tmux.socketName || 'default',
      seat: agent.name,
      sessionId,
    };
  }

  /**
   * Find the lease session id in the pane's process subtree. The pane's root pid
   * is node(mosaic), which has no lease env; the id lives on the claude child.
   * BFS from the root, bounded, returning the first descendant that carries a
   * valid MOSAIC_LEASE_SESSION_ID. Fail-closed (null) if none is found.
   */
  private async resolveLeaseSessionId(rootPid: number): Promise<string | null> {
    const queue: number[] = [rootPid];
    const seen = new Set<number>();
    while (queue.length > 0 && seen.size < MAX_SUBTREE_PIDS) {
      const pid = queue.shift()!;
      if (seen.has(pid)) continue;
      seen.add(pid);
      let sessionId: string | null = null;
      try {
        sessionId = parseLeaseSessionId(await this.environmentReader(pid));
      } catch {
        sessionId = null;
      }
      if (sessionId !== null) return sessionId;
      let children: number[] = [];
      try {
        children = await this.childrenReader(pid);
      } catch {
        children = [];
      }
      for (const child of children) {
        if (!seen.has(child)) queue.push(child);
      }
    }
    return null;
  }

  async sendPromotion(target: PromotionTarget): Promise<void> {
    const targetPane = `=${target.seat}:0.0`;
    const socketName = target.bundle === 'default' ? '' : target.bundle;
    // Registered Claude commands must arrive as their exact literal text; the
    // fleet agent sender prepends an identity envelope, so it cannot carry this
    // command without preventing the UserPromptSubmit matcher from recognizing it.
    await this.runPromotionCommand([
      'tmux',
      ...socketArgs(socketName),
      'send-keys',
      '-t',
      targetPane,
      '-l',
      PROMOTION_COMMAND,
    ]);
    await this.runPromotionCommand([
      'tmux',
      ...socketArgs(socketName),
      'send-keys',
      '-t',
      targetPane,
      'Enter',
    ]);
  }

  private async runPromotionCommand(command: string[]): Promise<void> {
    const result = await this.run(command);
    if (result.exitCode !== 0) {
      throw new Error('Promotion command delivery failed.');
    }
  }

  private async run(command: string[]): Promise<CommandResult> {
    const [executable, ...args] = command;
    if (executable === undefined) {
      throw new Error('Promotion transport command is empty.');
    }
    return await withTimeout(this.options.runner(executable, args), TRANSPORT_COMMAND_TIMEOUT_MS);
  }
}

function withTimeout<T>(operation: Promise<T>, timeoutMs: number): Promise<T> {
  return new Promise((resolve, reject) => {
    const timeout = setTimeout(() => {
      reject(new Error(`Promotion transport command timed out after ${timeoutMs}ms.`));
    }, timeoutMs);
    void operation.then(
      (value) => {
        clearTimeout(timeout);
        resolve(value);
      },
      (error: unknown) => {
        clearTimeout(timeout);
        reject(error);
      },
    );
  });
}

async function readPaneEnvironment(pid: number): Promise<string> {
  return readFile(`/proc/${pid}/environ`, 'utf8');
}

async function readChildPids(pid: number): Promise<number[]> {
  // Linux exposes direct children of the main thread here (CONFIG_PROC_CHILDREN).
  try {
    const raw = await readFile(`/proc/${pid}/task/${pid}/children`, 'utf8');
    return raw
      .split(/\s+/)
      .filter(Boolean)
      .map((value) => Number.parseInt(value, 10))
      .filter((value) => Number.isInteger(value) && value > 0);
  } catch {
    return [];
  }
}

function parseLeaseSessionId(environment: string): string | null {
  const value = environment
    .split('\0')
    .find((entry) => entry.startsWith('MOSAIC_LEASE_SESSION_ID='))
    ?.slice('MOSAIC_LEASE_SESSION_ID='.length);
  return value !== undefined && SESSION_ID_PATTERN.test(value) ? value : null;
}