pty: the sidecar owns its own transport

Terminals were a set of commands the platform drove. Officer sent pty:init / pty:input /
pty:resize / pty:close / pty:list over the registration socket, subscribed to ONE global
output stream, filtered every frame down to a session and rewrapped it — double
JSON-encoded — on the way out. That is terminal knowledge living in the process whose job
is authentication, and it made officer part of the data path for every keystroke.

The sidecar now serves its own loopback HTTP + WebSocket listener and announces the port
as `pty:server`, like every other HTTP sidecar. Officer authenticates the upgrade and
relays frames without reading them.

Split into three files, because "the sidecar" was one:
- sessions.mjs — the shell store. Spawn, attach, detach, resize, kill, scrollback, the
  OSC-title scrape. Clients are a Set per session, so two panels can watch one shell.
- server.mjs — the listener. /ws speaks the browser's existing contract unchanged
  ({input,resize} in, {output,replay,exit,panel-refresh} out), plus /_officer/sessions,
  DELETE /_officer/sessions/:id and POST /_officer/panel-refresh.
- index.mjs — the registration socket, and nothing else. It carries a port now.

On the platform side /api/terminal/* becomes createSidecarProxy, deleting the hand-rolled
router from two days ago, and websocket.ts drops from a translating bridge to a byte relay
modelled on the vault one. The whole PtyCommand/PtyEvent/PtyInitConfig/PtySessionInfo
vocabulary is gone from protocol.ts, connect.ts and sidecar-registry.ts.

broadcastPanelRefresh is now a POST to the sidecar: officer no longer holds terminal
sockets to loop over. Fire-and-forget — a missed refresh is a stale panel, not a failure.

The frontend did not move. The sidecar speaks what the browser already spoke.

The integration test was rewritten against the new shape, and tests something stronger than
before: officer is stopped mid-session and the shell keeps streaming, because officer is
not in the path at all. It also covers re-attach replay, the session list and kill.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/servers/sidecar/pty/index.test.ts

@@ -1,23 +1,51 @@
 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.
+// Integration test for the things about the pty sidecar that cannot be reasoned about from the code alone:
+// that it serves its own transport, that a shell outlives both its viewer and officer, and that re-attach
+// replays rather than re-spawns.
 //
-// Nothing here touches the running officer — the sidecar dials API_URL, which is overridden per spawn.
+// The fake officer exists only to receive the announced port — officer is not in the data path any more,
+// so the browser side is spoken directly to the sidecar's own listener. Nothing here touches the running
+// officer: the sidecar dials API_URL, 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;
+function collector() {
   const frames: Frame[] = [];
   const waiters: Array<{ match: (f: Frame) => boolean; resolve: (f: Frame) => void }> = [];
 
+  const push = (frame: 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);
+    }
+  };
+
+  const await_ = (match: (f: Frame) => boolean, what: string, 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 ${what}`)), timeoutMs);
+      waiters.push({
+        match,
+        resolve: (f) => {
+          clearTimeout(timer);
+          resolve(f);
+        },
+      });
+    });
+
+  return { frames, push, await_ };
+}
+
+/** A fake officer: just enough registration socket to receive `pty:server`. */
+function fakeOfficer(port?: number) {
+  let socket: ServerWebSocket<unknown> | null = null;
+  const { push, await_ } = collector();
+
   const server = Bun.serve({
     port: port ?? 0,
     fetch(req, srv) {
@@ -27,13 +55,10 @@ function fakeOfficer(port?: number) {
     websocket: {
       open(ws) {
         socket = ws;
+        ws.send(JSON.stringify({ type: 'registered', id: 'sc_test' }));
       },
       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);
-        }
+        push(JSON.parse(String(raw)) as Frame);
       },
       close() {
         socket = null;
@@ -43,49 +68,54 @@ function fakeOfficer(port?: number) {
 
   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(''),
+    await: (match: (f: Frame) => boolean, what: string) => await_(match, what),
     stop: () => server.stop(true),
+    get connected() {
+      return socket !== null;
+    },
   };
 }
 
-const isOutput = (sessionId: string, needle: string) => (f: Frame) =>
-  f.type === 'pty:output' && f.sessionId === sessionId && String(f.data).includes(needle);
+/** A fake browser. The sidecar's socket contract IS the one the browser already speaks. */
+function client(wsBase: string, query: string) {
+  const ws = new WebSocket(`${wsBase}/ws?${query}`);
+  const { frames, push, await_ } = collector();
+
+  ws.addEventListener('message', (ev) => push(JSON.parse(String(ev.data)) as Frame));
+
+  return {
+    open: () =>
+      new Promise<void>((resolve) =>
+        ws.readyState === WebSocket.OPEN ? resolve() : ws.addEventListener('open', () => resolve()),
+      ),
+    send: (msg: Frame) => ws.send(JSON.stringify(msg)),
+    await: (match: (f: Frame) => boolean, what: string) => await_(match, what),
+    outputSoFar: () =>
+      frames
+        .filter((f) => f.type === 'output')
+        .map((f) => String(f.data))
+        .join(''),
+    close: () => ws.close(),
+  };
+}
+
+const sawOutput = (needle: string) => (f: Frame) => f.type === 'output' && 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';
+  test('serves its own transport; a shell outlives both its viewer and officer', async () => {
+    const sessionId = 'test-transport';
     let officer = fakeOfficer();
-    const port = officer.port;
+    const officerPort = 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).
+        API_URL: `ws://127.0.0.1:${officerPort}`,
+        // The sidecar chooses the shell and the home itself, so pin 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',
@@ -94,41 +124,62 @@ describe('pty sidecar', () => {
       stderr: 'ignore',
     });
 
-    await officer.await((f) => f.type === 'register' && f.capabilities?.includes('terminal'));
+    // The only thing that crosses the registration socket now.
+    const announced = await officer.await((f) => f.type === 'pty:server' && typeof f.port === 'number', 'pty:server');
+    const httpBase = `http://127.0.0.1:${announced.port}`;
+    const wsBase = `ws://127.0.0.1:${announced.port}`;
 
-    // `~` 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);
+    const a = client(wsBase, `sessionId=${sessionId}&cwd=~&cols=80&rows=24`);
+    await a.open();
 
-    officer.send({ type: 'pty:input', id: 'in0', sessionId, data: 'pwd\n' });
-    await officer.await(isOutput(sessionId, '/tmp'));
+    // `~` is resolved by the sidecar against its own HOME_DIR.
+    a.send({ type: 'input', data: 'pwd\n' });
+    await a.await(sawOutput('/tmp'), 'pwd output');
 
-    officer.send({ type: 'pty:input', id: 'in1', sessionId, data: 'echo before-restart\n' });
-    await officer.await(isOutput(sessionId, 'before-restart'));
+    a.send({ type: 'input', data: 'echo before-restart\n' });
+    await a.await(sawOutput('before-restart'), 'before-restart');
 
-    // ── the restart ──
+    // ── officer restarts ──
+    // It is not in the data path, so this is a non-event for the shell. Under the old translating bridge
+    // every keystroke and every byte of output crossed the registration socket.
     officer.stop();
-    officer = fakeOfficer(port);
-    await officer.await((f) => f.type === 'register');
+    a.send({ type: 'input', data: 'echo during-outage\n' });
+    await a.await(sawOutput('during-outage'), 'output while officer is down');
 
-    // 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'));
+    officer = fakeOfficer(officerPort);
+    await officer.await((f) => f.type === 'pty:server', 're-announced port');
 
-    // 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);
+    // ── the viewer goes away, a new one attaches ──
+    a.close();
+    const b = client(wsBase, `sessionId=${sessionId}&cols=80&rows=24`);
+    await b.open();
+
+    // Re-attach replays scrollback as history rather than ordinary output, so the client rebuilds instead
+    // of appending a second copy — and it spans everything, including what happened while officer was down.
+    const replay = await b.await((f) => f.type === 'replay', 'replay');
     expect(String(replay.data)).toContain('before-restart');
-    expect(String(replay.data)).toContain('after-restart');
+    expect(String(replay.data)).toContain('during-outage');
+    expect(b.outputSoFar()).not.toContain('before-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');
+    // Same shell, not a second one.
+    b.send({ type: 'input', data: 'echo after-reattach\n' });
+    await b.await(sawOutput('after-reattach'), 'after-reattach');
 
-    officer.send({ type: 'pty:close', id: 'c1', sessionId });
-    await officer.await((f) => f.type === 'pty:exit' && f.sessionId === sessionId);
+    // ── the sidecar's own HTTP surface ──
+    const listed = (await fetch(`${httpBase}/_officer/sessions`).then((r) => r.json())) as { sessions: any[] };
+    const entry = listed.sessions.find((s) => s.sessionId === sessionId);
+    expect(entry).toBeTruthy();
+    expect(entry.pid).toBeGreaterThan(0);
+    expect(entry.clients).toBe(1);
+
+    const killed = await fetch(`${httpBase}/_officer/sessions/${sessionId}`, { method: 'DELETE' });
+    expect(killed.status).toBe(200);
+    await b.await((f) => f.type === 'exit', 'exit after kill');
+
+    const after = (await fetch(`${httpBase}/_officer/sessions`).then((r) => r.json())) as { sessions: any[] };
+    expect(after.sessions.find((s) => s.sessionId === sessionId)).toBeUndefined();
+
+    b.close();
     officer.stop();
   }, 30_000);
 });