pi-remote-control/extensions/remote-control/tmux/control.ts

/**
 * tmux control-mode client — per-session.
 *
 * Spawns `tmux -C attach -t <session>`, parses `%output` notifications,
 * decodes octal-escaped bytes, and broadcasts raw ANSI to subscribers.
 *
 * Design:
 *  - One ControlClient instance per tmux session (per-session, not per-server).
 *  - Subscribers register a callback; each raw Buffer chunk is broadcast.
 *  - On %exit or process close, all subscribers are notified and removed.
 *
 * Reference: feat/spike-tmux-cc / spike-cc.ts (Phase 0.5 PoC).
 *
 * Risk mitigations:
 *  R4: streaming line-parser, per-line decode, no full-buffer copies.
 *
 * Owner: T-1.1
 */

import { type ChildProcess, spawn } from "node:child_process";
import { createInterface } from "node:readline";

export type OutputCallback = (chunk: Buffer) => void;
export type CloseCallback = (reason: string) => void;

export interface ControlClientOptions {
  session: string;
  onClose?: CloseCallback;
}

export class ControlClient {
  readonly session: string;
  private proc: ChildProcess | null = null;
  private subscribers = new Map<symbol, OutputCallback>();
  private closed = false;
  private onClose?: CloseCallback;

  constructor(opts: ControlClientOptions) {
    this.session = opts.session;
    this.onClose = opts.onClose;
  }

  // ---------------------------------------------------------------------------
  // Lifecycle
  // ---------------------------------------------------------------------------

  start(): void {
    if (this.proc) return;
    this.closed = false;

    // -C = control mode; tmux sends %output events for pane data (do NOT use -CC which bypasses %output)
    this.proc = spawn("tmux", ["-C", "attach", "-t", this.session], {
      stdio: ["pipe", "pipe", "pipe"],
    });

    const rl = createInterface({
      // biome-ignore lint/style/noNonNullAssertion: stdout is always set when stdio includes 'pipe'
      input: this.proc.stdout!,
      crlfDelay: Number.POSITIVE_INFINITY,
    });

    rl.on("line", (line: string) => {
      this.parseLine(line);
    });

    this.proc.stderr?.on("data", (_d: Buffer) => {
      // Ignore tmux stderr (status messages). Can log at debug level if needed.
    });

    this.proc.on("close", (code: number | null) => {
      this.closed = true;
      this.subscribers.clear();
      this.onClose?.(`tmux process exited (code=${code})`);
    });
  }

  stop(): void {
    if (this.proc && !this.closed) {
      this.proc.kill("SIGTERM");
    }
  }

  get isRunning(): boolean {
    return !this.closed && this.proc !== null;
  }

  // ---------------------------------------------------------------------------
  // Subscriptions
  // ---------------------------------------------------------------------------

  subscribe(cb: OutputCallback): () => void {
    const key = Symbol();
    this.subscribers.set(key, cb);
    return () => this.subscribers.delete(key);
  }

  // ---------------------------------------------------------------------------
  // Parsing
  // ---------------------------------------------------------------------------

  /**
   * Parse one line of tmux control-mode output.
   *
   * Control-mode lines that matter:
   *   %output %<pane-id> <octal-escaped-bytes>
   *   %exit [reason]
   * Everything else is ignored.
   */
  private parseLine(line: string): void {
    if (!line.startsWith("%")) return;

    const spaceIdx = line.indexOf(" ");
    const type = spaceIdx === -1 ? line.slice(1) : line.slice(1, spaceIdx);
    const rest = spaceIdx === -1 ? "" : line.slice(spaceIdx + 1);

    if (type === "output") {
      this.handleOutput(rest);
    } else if (type === "exit") {
      this.closed = true;
      this.subscribers.clear();
      this.onClose?.(`%exit ${rest}`);
    }
    // layout-change, window-add, etc. are ignored
  }

  /**
   * Handle a %output notification.
   * Format: %<pane-id> <octal-escaped-value>
   */
  private handleOutput(data: string): void {
    const spaceIdx = data.indexOf(" ");
    if (spaceIdx === -1) return; // malformed, skip

    const escapedValue = data.slice(spaceIdx + 1);
    const decoded = decodeOctalEscapes(escapedValue);
    if (decoded.length === 0) return;

    for (const cb of this.subscribers.values()) {
      cb(decoded);
    }
  }
}

// ---------------------------------------------------------------------------
// Octal-escape decoder (from spike-cc.ts, adapted)
// ---------------------------------------------------------------------------

/**
 * Decode tmux's octal-escaped output format.
 * "hello\\012world" → Buffer containing "hello\nworld"
 */
export function decodeOctalEscapes(input: string): Buffer {
  // Fast-path: nothing to decode
  if (!input.includes("\\")) return Buffer.from(input, "binary");

  const bytes: number[] = [];
  let i = 0;
  while (i < input.length) {
    if (input[i] === "\\" && i + 3 < input.length) {
      const oct = input.slice(i + 1, i + 4);
      if (/^[0-7]{3}$/.test(oct)) {
        bytes.push(parseInt(oct, 8));
        i += 4;
        continue;
      }
    }
    bytes.push(input.charCodeAt(i));
    i++;
  }
  return Buffer.from(bytes);
}