move the pty sidecar under src/servers/sidecar
it was the only sidecar living outside src/servers/sidecar/ — it sat in api/terminal/ next to the bridge that talks to it, which is the one place a reader looking for "the sidecars" would not check. now src/servers/sidecar/pty/index.mjs, matching every peer, with a note on the pm2 entry about why this one is node and .mjs (node-pty is a native addon) rather than bun and typescript like the rest. the templates/ directory went to api/users/, next to provision.ts:seedShellConfigs, which is now its only consumer — the sidecar's duplicate seeder went with the sandbox branch in the previous commit. api/terminal/ is left holding exactly one thing: the websocket bridge. no behaviour change. the pm2 entry's script path changed, so `pm2 restart officer-pty` is not enough — pm2 remembers the old path until the entry is deleted and started again. commands are in SIDECAR_WORK_LOG.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,257 @@
|
||||
// Graceful shutdown
|
||||
process.on('SIGINT', () => {
|
||||
console.log('[pty-sidecar] shutting down...');
|
||||
for (const [id, session] of sessions) {
|
||||
try { session.term.kill(); } catch { /* ignore */ }
|
||||
}
|
||||
sessions.clear();
|
||||
if (ws) { try { ws.close(); } catch { /* ignore */ } }
|
||||
process.exit(0);
|
||||
});
|
||||
process.on('SIGTERM', () => process.emit('SIGINT'));
|
||||
|
||||
import { join } from 'node:path';
|
||||
import WebSocket from 'ws';
|
||||
import * as pty from 'node-pty';
|
||||
|
||||
import 'dotenv/config';
|
||||
|
||||
const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`;
|
||||
const REGISTER_URL = `${API_URL}/api/sidecar/register`;
|
||||
|
||||
const BUFFER_MAX = 50 * 1024;
|
||||
const RECONNECT_DELAYS = [200, 500, 1000, 2000, 4000, 8000, 15000];
|
||||
|
||||
// ── What kind of shell this process runs ──
|
||||
//
|
||||
// These used to arrive inside every pty:init, which meant officer chose the owner's shell and read the
|
||||
// owner's HOME to do it. They are this process's business: it is the one that spawns the thing.
|
||||
//
|
||||
// HOME_DIR mirrors `data-path.ts:getOwnerHomeDir` — on a host where the owner's real login home differs
|
||||
// from this process's HOME, the shell should open in the former, like every other host-executing surface.
|
||||
const HOME_DIR = process.env.HOME_DIR ?? process.env.HOME ?? process.cwd();
|
||||
const SHELL = { command: process.env.SHELL ?? '/bin/zsh', args: ['-i'] };
|
||||
// A terminal is always a plain host shell: the owner is the only account and it is their own machine
|
||||
// (`platform/CLAUDE.md` — do not add a jail without being asked). There used to be a second branch here
|
||||
// for a bwrap sandbox, selected by `config.host`, which officer hardcoded to true. Nothing ever built the
|
||||
// bwrap command it expected, on either side, so it could not have run — it is in git history if the
|
||||
// decision is ever revisited, and the `ensureUserFiles` half of it duplicated
|
||||
// `api/users/provision.ts:seedShellConfigs`, which is the live seeder of those templates.
|
||||
|
||||
const resolveCwd = (cwd) => {
|
||||
if (!cwd || cwd === '~') return HOME_DIR;
|
||||
if (cwd.startsWith('~/')) return join(HOME_DIR, cwd.slice(2));
|
||||
if (cwd.startsWith('/')) return cwd;
|
||||
// Relative paths have no meaning here — this process's cwd is the repo, not the user's folder.
|
||||
return HOME_DIR;
|
||||
};
|
||||
|
||||
/** @type {Map<string, { term: import('node-pty').IPty, buffer: string, cols: number, rows: number }>} */
|
||||
const sessions = new Map();
|
||||
|
||||
// ── Helpers ──
|
||||
|
||||
// Always resolve the CURRENT registration socket, never one captured in a closure.
|
||||
//
|
||||
// `term.onData` used to close over the socket that was live when the session was created. Officer is a
|
||||
// PM2 peer that restarts often, and each restart gives this process a brand new socket — so every
|
||||
// pre-existing session went on writing to a closed one, where the readyState check below dropped it
|
||||
// silently. The shell stayed alive and kept accepting input (that arrives on the new socket), but its
|
||||
// output never came back: the terminal looked frozen until you closed the panel. Reading the module
|
||||
// variable at send time is the whole fix.
|
||||
const sendJson = (msg) => {
|
||||
try {
|
||||
if (ws && ws.readyState === WebSocket.OPEN) {
|
||||
ws.send(JSON.stringify(msg));
|
||||
}
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
};
|
||||
|
||||
const appendBuffer = (session, data) => {
|
||||
session.buffer += data;
|
||||
if (session.buffer.length > BUFFER_MAX) {
|
||||
session.buffer = session.buffer.slice(-BUFFER_MAX);
|
||||
}
|
||||
};
|
||||
|
||||
// ── Command handler ──
|
||||
|
||||
async function handleCommand(msg) {
|
||||
switch (msg.type) {
|
||||
case 'pty:init': {
|
||||
const { sessionId, config } = msg;
|
||||
if (!sessionId) return;
|
||||
|
||||
const existing = sessions.get(sessionId);
|
||||
console.log(`[pty-sidecar] init sessionId=${sessionId} existing=${!!existing} total=${sessions.size}`);
|
||||
|
||||
if (existing) {
|
||||
// Re-attach. The scrollback goes out as `pty:replay`, not as ordinary output, because the client
|
||||
// may already be showing some of it: after an officer restart the browser keeps its terminal and
|
||||
// reconnects, so replaying blind appended a second copy of everything on screen. Marked as
|
||||
// history, the client can reset and rebuild from it instead.
|
||||
if (existing.buffer.length > 0) {
|
||||
sendJson({ type: 'pty:replay', sessionId, data: existing.buffer });
|
||||
}
|
||||
|
||||
// Resize PTY to new client dimensions
|
||||
const cols = config.cols ?? existing.cols;
|
||||
const rows = config.rows ?? existing.rows;
|
||||
if (cols > 0 && rows > 0 && (cols !== existing.cols || rows !== existing.rows)) {
|
||||
existing.cols = cols;
|
||||
existing.rows = rows;
|
||||
try {
|
||||
existing.term.resize(cols, rows);
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
|
||||
sendJson({ type: 'pty:ready', id: msg.id, sessionId });
|
||||
return;
|
||||
}
|
||||
|
||||
// New session — spawn PTY
|
||||
const cwd = resolveCwd(config.cwd);
|
||||
const cols = config.cols ?? 80;
|
||||
const rows = config.rows ?? 24;
|
||||
|
||||
let term;
|
||||
try {
|
||||
term = pty.spawn(SHELL.command, SHELL.args, {
|
||||
name: 'xterm-256color',
|
||||
cols,
|
||||
rows,
|
||||
cwd,
|
||||
env: { ...process.env, TERM: 'xterm-256color' },
|
||||
});
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : 'Failed to start terminal';
|
||||
sendJson({ type: 'pty:output', sessionId, data: `\r\n[Terminal error] ${message}\r\n` });
|
||||
sendJson({ type: 'pty:exit', sessionId, exitCode: 1 });
|
||||
return;
|
||||
}
|
||||
|
||||
const session = { term, buffer: '', cols, rows };
|
||||
sessions.set(sessionId, session);
|
||||
|
||||
term.onData((output) => {
|
||||
appendBuffer(session, output);
|
||||
sendJson({ type: 'pty:output', sessionId, data: output });
|
||||
});
|
||||
|
||||
term.onExit(({ exitCode, signal }) => {
|
||||
console.log(`[pty-sidecar] session ${sessionId} exited code=${exitCode} signal=${signal}`);
|
||||
sendJson({ type: 'pty:exit', sessionId, exitCode, signal });
|
||||
sessions.delete(sessionId);
|
||||
});
|
||||
|
||||
sendJson({ type: 'pty:ready', id: msg.id, sessionId });
|
||||
return;
|
||||
}
|
||||
|
||||
case 'pty:input': {
|
||||
const session = sessions.get(msg.sessionId);
|
||||
if (session) {
|
||||
session.term.write(msg.data ?? '');
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'pty:resize': {
|
||||
const session = sessions.get(msg.sessionId);
|
||||
if (session && msg.cols > 0 && msg.rows > 0) {
|
||||
session.cols = msg.cols;
|
||||
session.rows = msg.rows;
|
||||
try {
|
||||
session.term.resize(msg.cols, msg.rows);
|
||||
} catch {
|
||||
// PTY may have already exited
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'pty:close': {
|
||||
const session = sessions.get(msg.sessionId);
|
||||
if (session) {
|
||||
try {
|
||||
session.term.kill();
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
sessions.delete(msg.sessionId);
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Connect to API server with reconnect ──
|
||||
|
||||
let ws = null;
|
||||
let reconnectAttempt = 0;
|
||||
let reconnectTimer = null;
|
||||
|
||||
console.log(`[pty-sidecar] starting, connecting to ${REGISTER_URL}`);
|
||||
|
||||
function connect() {
|
||||
if (ws && (ws.readyState === WebSocket.CONNECTING || ws.readyState === WebSocket.OPEN)) return;
|
||||
|
||||
try {
|
||||
ws = new WebSocket(REGISTER_URL);
|
||||
} catch (err) {
|
||||
console.error(`[pty-sidecar] failed to create WebSocket:`, err.message ?? err);
|
||||
scheduleReconnect();
|
||||
return;
|
||||
}
|
||||
|
||||
ws.on('open', () => {
|
||||
reconnectAttempt = 0;
|
||||
console.log('[pty-sidecar] connected, sending registration...');
|
||||
sendJson({ type: 'register', name: 'pty', capabilities: ['terminal'] });
|
||||
});
|
||||
|
||||
ws.on('message', (data) => {
|
||||
try {
|
||||
const msg = JSON.parse(typeof data === 'string' ? data : data.toString());
|
||||
|
||||
if (msg.type === 'registered') {
|
||||
console.log(`[pty-sidecar] registered with API server (id=${msg.id})`);
|
||||
return;
|
||||
}
|
||||
|
||||
handleCommand(msg);
|
||||
} catch {
|
||||
// skip malformed messages
|
||||
}
|
||||
});
|
||||
|
||||
ws.on('close', () => {
|
||||
console.log('[pty-sidecar] disconnected from API server');
|
||||
ws = null;
|
||||
scheduleReconnect();
|
||||
});
|
||||
|
||||
ws.on('error', (err) => {
|
||||
if (reconnectAttempt <= 1) {
|
||||
console.error(`[pty-sidecar] connection error: ${err.message ?? err}`);
|
||||
}
|
||||
// onclose will fire after this
|
||||
});
|
||||
}
|
||||
|
||||
function scheduleReconnect() {
|
||||
if (reconnectTimer) return;
|
||||
const delay = RECONNECT_DELAYS[Math.min(reconnectAttempt, RECONNECT_DELAYS.length - 1)];
|
||||
reconnectAttempt++;
|
||||
reconnectTimer = setTimeout(() => {
|
||||
reconnectTimer = null;
|
||||
connect();
|
||||
}, delay);
|
||||
}
|
||||
|
||||
// Start connecting
|
||||
connect();
|
||||
@@ -0,0 +1,134 @@
|
||||
import { describe, test, expect, afterAll } from 'bun:test';
|
||||
import type { ServerWebSocket, Subprocess } from 'bun';
|
||||
|
||||
// Integration test for the one thing about the pty sidecar that cannot be reasoned about from the code
|
||||
// alone: what happens to a live shell when officer goes away and comes back. It stands up a fake
|
||||
// registration socket, runs the real sidecar against it, then kills the socket and rebinds the same port
|
||||
// the way `pm2 restart officer` does.
|
||||
//
|
||||
// Nothing here touches the running officer — the sidecar dials API_URL, which is overridden per spawn.
|
||||
|
||||
const SIDECAR = 'src/servers/sidecar/pty/index.mjs';
|
||||
|
||||
type Frame = Record<string, any>;
|
||||
|
||||
// One fake officer. `stop()` drops the socket; a new instance on the same port is the restart.
|
||||
function fakeOfficer(port?: number) {
|
||||
let socket: ServerWebSocket<unknown> | null = null;
|
||||
const frames: Frame[] = [];
|
||||
const waiters: Array<{ match: (f: Frame) => boolean; resolve: (f: Frame) => void }> = [];
|
||||
|
||||
const server = Bun.serve({
|
||||
port: port ?? 0,
|
||||
fetch(req, srv) {
|
||||
if (new URL(req.url).pathname === '/api/sidecar/register' && srv.upgrade(req)) return undefined;
|
||||
return new Response('no', { status: 404 });
|
||||
},
|
||||
websocket: {
|
||||
open(ws) {
|
||||
socket = ws;
|
||||
},
|
||||
message(_ws, raw) {
|
||||
const frame = JSON.parse(String(raw)) as Frame;
|
||||
frames.push(frame);
|
||||
for (let i = waiters.length - 1; i >= 0; i--) {
|
||||
if (waiters[i]!.match(frame)) waiters.splice(i, 1)[0]!.resolve(frame);
|
||||
}
|
||||
},
|
||||
close() {
|
||||
socket = null;
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
return {
|
||||
port: server.port,
|
||||
send: (msg: Frame) => socket?.send(JSON.stringify(msg)),
|
||||
/** Resolve on the first frame matching `match`, including ones already received. */
|
||||
await: (match: (f: Frame) => boolean, timeoutMs = 15_000) =>
|
||||
new Promise<Frame>((resolve, reject) => {
|
||||
const seen = frames.find(match);
|
||||
if (seen) return resolve(seen);
|
||||
const timer = setTimeout(() => reject(new Error(`timed out waiting for a frame`)), timeoutMs);
|
||||
waiters.push({
|
||||
match,
|
||||
resolve: (f) => {
|
||||
clearTimeout(timer);
|
||||
resolve(f);
|
||||
},
|
||||
});
|
||||
}),
|
||||
/** Output frames for one session, concatenated — the terminal's visible text. */
|
||||
outputFor: (sessionId: string) =>
|
||||
frames
|
||||
.filter((f) => f.type === 'pty:output' && f.sessionId === sessionId)
|
||||
.map((f) => f.data as string)
|
||||
.join(''),
|
||||
stop: () => server.stop(true),
|
||||
};
|
||||
}
|
||||
|
||||
const isOutput = (sessionId: string, needle: string) => (f: Frame) =>
|
||||
f.type === 'pty:output' && f.sessionId === sessionId && String(f.data).includes(needle);
|
||||
|
||||
let child: Subprocess | null = null;
|
||||
afterAll(() => child?.kill());
|
||||
|
||||
describe('pty sidecar', () => {
|
||||
test('a shell keeps streaming output after officer restarts under it', async () => {
|
||||
const sessionId = 'test-restart';
|
||||
let officer = fakeOfficer();
|
||||
const port = officer.port;
|
||||
|
||||
child = Bun.spawn(['node', SIDECAR], {
|
||||
env: {
|
||||
...process.env,
|
||||
API_URL: `ws://127.0.0.1:${port}`,
|
||||
// The sidecar now chooses the shell and the home itself, so the test pins both rather than
|
||||
// spawning the owner's interactive zsh (which would read their rc files and their history).
|
||||
SHELL: '/bin/sh',
|
||||
HOME_DIR: '/tmp',
|
||||
ENV: '/dev/null',
|
||||
},
|
||||
stdout: 'ignore',
|
||||
stderr: 'ignore',
|
||||
});
|
||||
|
||||
await officer.await((f) => f.type === 'register' && f.capabilities?.includes('terminal'));
|
||||
|
||||
// `~` is resolved by the sidecar against its own HOME_DIR, not by officer.
|
||||
officer.send({ type: 'pty:init', id: 'i1', sessionId, config: { sessionId, cwd: '~', cols: 80, rows: 24 } });
|
||||
await officer.await((f) => f.type === 'pty:ready' && f.sessionId === sessionId);
|
||||
|
||||
officer.send({ type: 'pty:input', id: 'in0', sessionId, data: 'pwd\n' });
|
||||
await officer.await(isOutput(sessionId, '/tmp'));
|
||||
|
||||
officer.send({ type: 'pty:input', id: 'in1', sessionId, data: 'echo before-restart\n' });
|
||||
await officer.await(isOutput(sessionId, 'before-restart'));
|
||||
|
||||
// ── the restart ──
|
||||
officer.stop();
|
||||
officer = fakeOfficer(port);
|
||||
await officer.await((f) => f.type === 'register');
|
||||
|
||||
// The shell is the same process; only officer changed. Before the sendJson fix this input was
|
||||
// accepted and executed, but its output went to the socket captured at init time and vanished.
|
||||
officer.send({ type: 'pty:input', id: 'in2', sessionId, data: 'echo after-restart\n' });
|
||||
await officer.await(isOutput(sessionId, 'after-restart'));
|
||||
|
||||
// Re-attaching replays the scrollback, marked as history so the client can rebuild rather than
|
||||
// append — and it contains what happened on both sides of the restart.
|
||||
officer.send({ type: 'pty:init', id: 'i2', sessionId, config: { sessionId, cols: 80, rows: 24 } });
|
||||
const replay = await officer.await((f) => f.type === 'pty:replay' && f.sessionId === sessionId);
|
||||
expect(String(replay.data)).toContain('before-restart');
|
||||
expect(String(replay.data)).toContain('after-restart');
|
||||
|
||||
// Re-attach must not spawn a second shell, and must not replay as ordinary output.
|
||||
await officer.await((f) => f.type === 'pty:ready' && f.id === 'i2');
|
||||
expect(officer.outputFor(sessionId)).not.toContain('before-restart');
|
||||
|
||||
officer.send({ type: 'pty:close', id: 'c1', sessionId });
|
||||
await officer.await((f) => f.type === 'pty:exit' && f.sessionId === sessionId);
|
||||
officer.stop();
|
||||
}, 30_000);
|
||||
});
|
||||
Reference in New Issue
Block a user