Repository Public
1/**2 * Browser connection ⇄ tmux attach client. One WebSocket = one `node-pty` running3 * `tmux attach-session -t =cr-<uuid>`, sized to the client. Closing the WebSocket kills only4 * that attach client; the tmux session (and the process inside it) keeps running.5 *6 * The browser's scrollback is fed by the runtime, not by tmux (see core-scrollback.ts): the pane7 * history is replayed before the attach starts and kept in step while the client is attached.8 *9 * Terminal bytes are never logged and never parsed.10 */11import fs from 'node:fs';12import path from 'node:path';13import { createRequire } from 'node:module';14import pty from 'node-pty';15import type { IPty } from 'node-pty';16import type { WebSocket, RawData } from 'ws';17import type { FastifyBaseLogger } from 'fastify';18import { TermClientMsg, WS_CLOSE, type TermServerMsg } from '@cr/shared';19import { Backpressure } from './core-backpressure.js';20import { envInt, type CoreContext } from './core-util.js';21import { DEFAULT_REPLAY_LINES, MAX_REPLAY_LINES, ScrollbackMirror } from './core-scrollback.js';22import { paneExitCode } from './tmux.js';2324export const MAX_INBOUND_FRAME_BYTES = 1024 * 1024;25/** Input typed before the attach PTY is up is held (bounded) and replayed in order. */26const MAX_PENDING_INPUT_BYTES = 1024 * 1024;2728/**29 * node-pty's prebuilt `spawn-helper` loses its executable bit with some package managers30 * (pnpm store), which makes every spawn fail with "posix_spawnp failed". Repair it in place.31 */32export function ensurePtyHelperExecutable(): void {33 try {34 const require = createRequire(import.meta.url);35 const root = path.dirname(require.resolve('node-pty/package.json'));36 const candidates = [37 path.join(root, 'prebuilds', `${process.platform}-${process.arch}`, 'spawn-helper'),38 path.join(root, 'build', 'Release', 'spawn-helper'),39 ];40 for (const c of candidates) {41 try {42 const st = fs.statSync(c);43 if ((st.mode & 0o111) !== 0o111) fs.chmodSync(c, st.mode | 0o755);44 } catch {45 /* candidate absent */46 }47 }48 } catch {49 /* node-pty layout unknown: spawn will report the real error */50 }51}5253function toBuffer(data: RawData): Buffer {54 if (Buffer.isBuffer(data)) return data;55 if (Array.isArray(data)) return Buffer.concat(data);56 return Buffer.from(data);57}5859interface Attachment {60 sessionId: string;61 end: (code: number | null) => void;62 shutdown: () => void;63}6465export class AttachManager {66 private readonly bySession = new Map<string, Set<Attachment>>();67 /** Called when a tmux attach client was started or ended (the session watcher re-polls). */68 onClientsChanged: (() => void) | null = null;6970 constructor(71 private readonly core: CoreContext,72 private readonly log: FastifyBaseLogger,73 private readonly saturationMs?: number,74 /** History lines replayed on attach (`CR_REPLAY_LINES`, default 5000, at most 20000). */75 private readonly replayLines: number = Math.min(MAX_REPLAY_LINES, envInt('CR_REPLAY_LINES', DEFAULT_REPLAY_LINES)),76 ) {77 ensurePtyHelperExecutable();78 }7980 get count(): number {81 let n = 0;82 for (const s of this.bySession.values()) n += s.size;83 return n;84 }8586 /** The session's process ended: tell every attached client, then close 4410. */87 notifyExit(sessionId: string, code: number | null): void {88 for (const a of [...(this.bySession.get(sessionId) ?? [])]) a.end(code);89 }9091 /** Runtime shutdown: detach every client. tmux sessions are left running. */92 closeAll(): void {93 for (const set of [...this.bySession.values()]) for (const a of [...set]) a.shutdown();94 }9596 /** Must be called synchronously from the WS route handler so no early frame is lost. */97 handle(ws: WebSocket, sessionId: string, cols: number, rows: number): void {98 const name = `cr-${sessionId}`;99 let term: IPty | null = null;100 let bp: Backpressure | null = null;101 let finished = false;102 let size = { cols, rows };103 let pending: Buffer[] = [];104 let pendingBytes = 0;105106 const send = (msg: TermServerMsg): void => {107 if (ws.readyState === ws.OPEN) ws.send(JSON.stringify(msg));108 };109 const mirror = new ScrollbackMirror({110 tmux: this.core.tmux,111 name,112 replayLines: this.replayLines,113 send: (chunk) => {114 if (!finished && ws.readyState === ws.OPEN) ws.send(chunk, { binary: true });115 },116 size: () => size,117 busy: () => ws.bufferedAmount > this.core.config.termBufferBytes,118 redraw: () => {119 try {120 if (term) process.kill(term.pid, 'SIGWINCH'); // the tmux attach client → full repaint121 } catch {122 /* client already gone */123 }124 },125 onError: (err) => this.log.warn({ err, sessionId }, 'attach: scrollback sync failed'),126 });127 const killPty = (): void => {128 mirror.dispose();129 bp?.dispose();130 bp = null;131 const t = term;132 term = null;133 if (t) {134 try {135 t.kill(); // SIGHUP to the tmux *client* only136 } catch {137 /* already gone */138 }139 this.onClientsChanged?.();140 }141 };142 const attachment: Attachment = {143 sessionId,144 end: (code) => finish(WS_CLOSE.SESSION_ENDED, 'session ended', { t: 'exit', code }),145 shutdown: () => finish(1001, 'runtime shutting down'),146 };147 const finish = (closeCode: number, reason: string, last?: TermServerMsg): void => {148 if (finished) return;149 finished = true;150 const set = this.bySession.get(sessionId);151 set?.delete(attachment);152 if (set && set.size === 0) this.bySession.delete(sessionId);153 killPty();154 if (last) send(last);155 if (ws.readyState === ws.OPEN) ws.close(closeCode, reason);156 };157158 ws.on('close', () => finish(1000, 'client closed'));159 ws.on('error', () => finish(1011, 'socket error'));160 ws.on('message', (data: RawData, isBinary: boolean) => {161 if (finished) return;162 if (isBinary) {163 const buf = toBuffer(data);164 if (buf.length > MAX_INBOUND_FRAME_BYTES) return finish(1009, 'frame too large');165 if (term) term.write(buf);166 else if (pendingBytes + buf.length <= MAX_PENDING_INPUT_BYTES) {167 pending.push(buf);168 pendingBytes += buf.length;169 }170 return;171 }172 let msg: TermClientMsg;173 try {174 msg = TermClientMsg.parse(JSON.parse(toBuffer(data).toString('utf8')));175 } catch {176 return send({ t: 'error', code: 'bad_message', message: 'invalid control message' });177 }178 if (msg.t === 'ping') return send({ t: 'pong', n: msg.n });179 if (msg.cols === size.cols && msg.rows === size.rows) return;180 size = { cols: msg.cols, rows: msg.rows };181 try {182 term?.resize(msg.cols, msg.rows);183 } catch {184 /* pty already closed; the exit path handles it */185 }186 mirror.noteResize();187 });188189 const set = this.bySession.get(sessionId) ?? new Set<Attachment>();190 set.add(attachment);191 this.bySession.set(sessionId, set);192193 void (async () => {194 const row = await this.core.tmux.getSession(name);195 if (finished) return;196 if (!row) {197 return finish(WS_CLOSE.NOT_FOUND, 'no such session', { t: 'error', code: 'not_found', message: 'no such session' });198 }199 send({ t: 'ready', sessionId, cols: size.cols, rows: size.rows });200 if (row.paneDead) return attachment.end(paneExitCode(row));201202 // History first (the browser starts from a blank terminal), then tmux paints the screen.203 await mirror.replay();204 if (finished) return;205 const t = pty.spawn(this.core.tmux.bin, this.core.tmux.attachArgv(name), {206 name: 'xterm-256color',207 cols: size.cols,208 rows: size.rows,209 cwd: this.core.config.dataDir,210 env: { ...this.core.tmux.env, TERM: 'xterm-256color', COLORTERM: 'truecolor' },211 encoding: null, // raw bytes: onData yields Buffers212 });213 term = t;214 const pressure = new Backpressure({215 highWaterBytes: this.core.config.termBufferBytes,216 buffered: () => ws.bufferedAmount,217 pause: () => t.pause(),218 resume: () => t.resume(),219 onSaturated: () => {220 this.log.warn({ sessionId }, 'attach: client saturated for too long, closing');221 finish(WS_CLOSE.BACKPRESSURE, 'client too slow');222 },223 saturationMs: this.saturationMs,224 });225 bp = pressure;226 // With `encoding: null` node-pty emits Buffers although its typings say string.227 (t.onData as unknown as (cb: (d: Buffer | string) => void) => void)((d) => {228 if (finished || ws.readyState !== ws.OPEN) return;229 const chunk = typeof d === 'string' ? Buffer.from(d, 'utf8') : d;230 ws.send(chunk, { binary: true });231 mirror.noteOutput(chunk);232 pressure.check();233 });234 t.onExit(() => {235 if (finished || term !== t) return;236 term = null;237 // The attach client ended on its own: either the session is gone/dead, or tmux dropped us.238 void this.core.tmux.getSession(name).then(239 (r) => {240 if (!r || r.paneDead) attachment.end(r ? paneExitCode(r) : null);241 else finish(1011, 'attach client ended');242 },243 () => finish(1011, 'attach client ended'),244 );245 });246 this.onClientsChanged?.();247 for (const buf of pending) t.write(buf);248 pending = [];249 pendingBytes = 0;250 })().catch((err: unknown) => {251 this.log.error({ err, sessionId }, 'attach failed');252 finish(1011, 'attach failed', { t: 'error', code: 'attach_failed', message: 'could not attach to the session' });253 });254 }255}256