feat(cli): TUI full-screen attach to cli-agent sessions (U14)

Suspend-and-handoff passthrough from the Ink TUI to a cli-agent PTY session:
mint an attach ticket, open the cli-session WebSocket, enter the alternate
screen + raw mode, stream WS scrollback/data frames to stdout, frame stdin
bytes into input messages, propagate resizes, and ACK consumed bytes for flow
control. Detach chord Ctrl-] restores the terminal and remounts Ink; a dropped
WS surfaces the error and restores the terminal cleanly.

Untrusted terminal output is neutralized through the same hardening filter the
dashboard WS bridge uses (re-exported from @fusion/dashboard) so OSC 52, non-
http(s) OSC 8 links, and device-status queries are stripped before reaching the
host TTY — the riskiest leg, since the host terminal honors more sequences than
xterm.js. CJK/double-width bytes pass through verbatim.

- packages/cli/src/commands/dashboard-tui/terminal-attach.ts (passthrough loop +
  injectable WS transport for tests)
- controller.openTerminalAttach() Ink integration (unmount/run/remount)
- adds `ws` runtime dep to packages/cli
- re-exports neutralizeTerminalOutput/flushTerminalOutput from @fusion/dashboard

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

.changeset/cli-agent-tui-attach.md

@@ -0,0 +1,13 @@
+---
+"@runfusion/fusion": minor
+---
+
+Add full-screen TUI attach to cli-agent sessions (U14). The Ink dashboard TUI
+can hand the terminal to a CLI agent session as a raw passthrough: it enters the
+alternate screen, streams WebSocket terminal bytes to stdout and stdin keystrokes
+back as input frames, propagates resizes, and ACKs consumed bytes for flow
+control. The detach chord (Ctrl-]) restores the TUI cleanly, and a dropped
+connection surfaces an error and restores the terminal. Untrusted terminal output
+is neutralized through the same hardening filter the dashboard WS bridge uses
+(OSC 52 clipboard writes, non-http(s) OSC 8 links, and device-status queries are
+stripped before reaching the host TTY).

packages/cli/package.json

@@ -68,7 +68,8 @@
     "multer": "^2.1.1",
     "node-pty": "npm:@homebridge/node-pty-prebuilt-multiarch@^0.13.1",
     "react": "^19.2.0",
-    "react-i18next": "^17.0.8"
+    "react-i18next": "^17.0.8",
+    "ws": "^8.18.0"
   },
   "peerDependencies": {
     "@earendil-works/pi-ai": "*",
@@ -95,6 +96,7 @@
     "@fusion/pi-llama-cpp": "workspace:*",
     "@types/node": "^22.0.0",
     "@types/react": "^19.2.0",
+    "@types/ws": "^8.5.0",
     "@vitest/coverage-v8": "^3.1.0",
     "cross-env": "^7.0.0",
     "esbuild": "^0.25.12",

packages/cli/src/commands/dashboard-tui/__tests__/terminal-attach.test.ts

@@ -0,0 +1,407 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  attachTerminalSession,
+  buildWsUrl,
+  fetchAttachTicket,
+  DETACH_CHORD_BYTE,
+  ALT_SCREEN_ENTER,
+  ALT_SCREEN_LEAVE,
+  WS_OPEN,
+  type TerminalWebSocket,
+  type AttachStdin,
+  type AttachStdout,
+} from "../terminal-attach.js";
+
+// ── Fakes ────────────────────────────────────────────────────────────────────
+
+/** In-memory WS-like transport. Never opens a real socket / never port 4040. */
+class FakeWs implements TerminalWebSocket {
+  readyState = WS_OPEN;
+  sent: string[] = [];
+  private handlers = new Map<string, ((...args: unknown[]) => void)[]>();
+  closed = false;
+
+  on(event: string, listener: (...args: unknown[]) => void): void {
+    const list = this.handlers.get(event) ?? [];
+    list.push(listener);
+    this.handlers.set(event, list);
+  }
+  send(data: string): void {
+    this.sent.push(data);
+  }
+  close(): void {
+    if (this.closed) return;
+    this.closed = true;
+    this.readyState = 3; // CLOSED
+    this.emit("close");
+  }
+  emit(event: string, ...args: unknown[]): void {
+    for (const l of this.handlers.get(event) ?? []) l(...args);
+  }
+  /** Simulate the server delivering a (JSON) message frame. */
+  deliver(frame: unknown): void {
+    this.emit("message", Buffer.from(JSON.stringify(frame), "utf8"));
+  }
+  /** Parsed client→server frames. */
+  parsedSent(): Array<Record<string, unknown>> {
+    return this.sent.map((s) => JSON.parse(s));
+  }
+}
+
+class FakeStdin implements AttachStdin {
+  isTTY = true;
+  isRaw = false;
+  private listeners: ((chunk: Buffer | string) => void)[] = [];
+  rawCalls: boolean[] = [];
+  resumed = false;
+  on(_event: "data", listener: (chunk: Buffer | string) => void): void {
+    this.listeners.push(listener);
+  }
+  off(_event: "data", listener: (chunk: Buffer | string) => void): void {
+    this.listeners = this.listeners.filter((l) => l !== listener);
+  }
+  setRawMode(mode: boolean): void {
+    this.rawCalls.push(mode);
+    this.isRaw = mode;
+  }
+  resume(): void {
+    this.resumed = true;
+  }
+  /** Simulate a user keystroke chunk. */
+  feed(chunk: Buffer | string): void {
+    for (const l of [...this.listeners]) l(chunk);
+  }
+  listenerCount(): number {
+    return this.listeners.length;
+  }
+}
+
+class FakeStdout implements AttachStdout {
+  columns = 80;
+  rows = 24;
+  writes: string[] = [];
+  private resizeListeners: (() => void)[] = [];
+  write(chunk: string): void {
+    this.writes.push(chunk);
+  }
+  on(_event: "resize", listener: () => void): void {
+    this.resizeListeners.push(listener);
+  }
+  off(_event: "resize", listener: () => void): void {
+    this.resizeListeners = this.resizeListeners.filter((l) => l !== listener);
+  }
+  fireResize(cols: number, rows: number): void {
+    this.columns = cols;
+    this.rows = rows;
+    for (const l of [...this.resizeListeners]) l();
+  }
+  resizeListenerCount(): number {
+    return this.resizeListeners.length;
+  }
+  all(): string {
+    return this.writes.join("");
+  }
+}
+
+/** A fetchImpl that always returns a ticket. */
+function okTicketFetch(ticket = "TICKET-1"): typeof fetch {
+  return vi.fn(async () =>
+    new Response(JSON.stringify({ ticket, expiresAt: new Date().toISOString(), readOnly: false }), {
+      status: 200,
+      headers: { "content-type": "application/json" },
+    }),
+  ) as unknown as typeof fetch;
+}
+
+function b64(s: string): string {
+  return Buffer.from(s, "utf8").toString("base64");
+}
+
+interface Harness {
+  ws: FakeWs;
+  stdin: FakeStdin;
+  stdout: FakeStdout;
+  onDetach: ReturnType<typeof vi.fn>;
+}
+
+/**
+ * Start an attach, drive the WS `open`, and return the harness + handle.
+ * `await tick()` lets the async ticket fetch resolve.
+ */
+async function startAttach(
+  overrides: Partial<Parameters<typeof attachTerminalSession>[0]> = {},
+): Promise<Harness & { handle: ReturnType<typeof attachTerminalSession> }> {
+  const ws = new FakeWs();
+  const stdin = new FakeStdin();
+  const stdout = new FakeStdout();
+  const onDetach = vi.fn();
+
+  const handle = attachTerminalSession({
+    baseUrl: "http://127.0.0.1:9999",
+    token: "daemon-tok",
+    sessionId: "sess-1",
+    stdin,
+    stdout,
+    onDetach,
+    fetchImpl: okTicketFetch(),
+    wsFactory: () => ws,
+    ackThresholdBytes: 64,
+    ...overrides,
+  });
+
+  // Let the ticket fetch resolve, then open the socket.
+  await tick();
+  ws.emit("open");
+
+  return { ws, stdin, stdout, onDetach, handle };
+}
+
+const tick = () => new Promise((r) => setTimeout(r, 0));
+
+afterEach(() => {
+  vi.restoreAllMocks();
+});
+
+// ── URL / ticket helpers ──────────────────────────────────────────────────────
+
+describe("buildWsUrl", () => {
+  it("derives ws:// from http:// and sets sessionId + ticket", () => {
+    const url = buildWsUrl({ baseUrl: "http://127.0.0.1:4040", sessionId: "s1", ticket: "t1" });
+    expect(url).toBe("ws://127.0.0.1:4040/api/cli-sessions/ws?sessionId=s1&ticket=t1");
+  });
+  it("derives wss:// from https://", () => {
+    const url = buildWsUrl({ baseUrl: "https://host", sessionId: "s", ticket: "t" });
+    expect(url.startsWith("wss://host/api/cli-sessions/ws")).toBe(true);
+  });
+});
+
+describe("fetchAttachTicket", () => {
+  it("POSTs to the attach-ticket route with bearer auth and returns the ticket", async () => {
+    const fetchImpl = vi.fn(async () =>
+      new Response(JSON.stringify({ ticket: "TK", readOnly: false }), { status: 200 }),
+    ) as unknown as typeof fetch;
+    const res = await fetchAttachTicket({
+      baseUrl: "http://h",
+      token: "tok",
+      sessionId: "s 1",
+      fetchImpl,
+    });
+    expect(res.ticket).toBe("TK");
+    const [url, init] = (fetchImpl as unknown as { mock: { calls: unknown[][] } }).mock.calls[0];
+    expect(url).toBe("http://h/api/cli-sessions/s%201/attach-ticket");
+    expect((init as RequestInit).method).toBe("POST");
+    expect((init as RequestInit).headers).toMatchObject({ authorization: "Bearer tok" });
+  });
+
+  it("throws on non-2xx", async () => {
+    const fetchImpl = vi.fn(async () => new Response("nope", { status: 404, statusText: "Not Found" })) as unknown as typeof fetch;
+    await expect(
+      fetchAttachTicket({ baseUrl: "http://h", sessionId: "s", fetchImpl }),
+    ).rejects.toThrow(/HTTP 404/);
+  });
+});
+
+// ── Passthrough loop ──────────────────────────────────────────────────────────
+
+describe("attachTerminalSession passthrough", () => {
+  it("enters alt-screen + raw mode on open and sends an initial resize", async () => {
+    const { stdin, stdout, ws } = await startAttach();
+    expect(stdout.all()).toContain(ALT_SCREEN_ENTER);
+    expect(stdin.rawCalls).toContain(true);
+    expect(stdin.resumed).toBe(true);
+    const resize = ws.parsedSent().find((f) => f.type === "resize");
+    expect(resize).toMatchObject({ type: "resize", cols: 80, rows: 24 });
+  });
+
+  it("frames stdin bytes into input messages (base64)", async () => {
+    const { stdin, ws } = await startAttach();
+    stdin.feed(Buffer.from("ls -la\r", "utf8"));
+    const input = ws.parsedSent().find((f) => f.type === "input");
+    expect(input).toBeDefined();
+    expect(Buffer.from(input!.data as string, "base64").toString("utf8")).toBe("ls -la\r");
+  });
+
+  it("writes data frames to stdout verbatim", async () => {
+    const { stdout, ws } = await startAttach();
+    const payload = "hello \x1b[31mworld\x1b[0m\n";
+    ws.deliver({ type: "data", data: b64(payload) });
+    expect(stdout.all()).toContain(payload);
+  });
+
+  it("passes CJK / double-width bytes through verbatim", async () => {
+    const { stdout, ws } = await startAttach();
+    const payload = "日本語 ❤ 한국어";
+    ws.deliver({ type: "data", data: b64(payload) });
+    expect(stdout.all()).toContain(payload);
+  });
+
+  it("writes scrollback frames to stdout", async () => {
+    const { stdout, ws } = await startAttach();
+    ws.deliver({ type: "scrollback", data: b64("prior output\n") });
+    expect(stdout.all()).toContain("prior output\n");
+  });
+
+  it("propagates host resize as a resize frame", async () => {
+    const { stdout, ws } = await startAttach();
+    stdout.fireResize(120, 40);
+    const resizes = ws.parsedSent().filter((f) => f.type === "resize");
+    expect(resizes.at(-1)).toMatchObject({ cols: 120, rows: 40 });
+  });
+});
+
+// ── Detach chord ──────────────────────────────────────────────────────────────
+
+describe("detach chord (Ctrl-])", () => {
+  it("restores state: leaves alt-screen, restores raw mode, closes WS, calls onDetach", async () => {
+    const { stdin, stdout, ws, onDetach, handle } = await startAttach();
+    stdin.feed(Buffer.from([DETACH_CHORD_BYTE]));
+    await handle.done;
+
+    expect(stdout.all()).toContain(ALT_SCREEN_LEAVE);
+    expect(stdin.rawCalls.at(-1)).toBe(false); // restored to prior (false)
+    expect(ws.closed).toBe(true);
+    expect(onDetach).toHaveBeenCalledTimes(1);
+    expect(onDetach).toHaveBeenCalledWith(undefined);
+    // Listeners removed (refcount back to baseline).
+    expect(stdin.listenerCount()).toBe(0);
+    expect(stdout.resizeListenerCount()).toBe(0);
+  });
+
+  it("flushes bytes before the chord, then detaches", async () => {
+    const { stdin, ws, handle } = await startAttach();
+    stdin.feed(Buffer.from([0x61, 0x62, DETACH_CHORD_BYTE, 0x63])); // "ab" Ctrl-] "c"
+    await handle.done;
+    const inputs = ws.parsedSent().filter((f) => f.type === "input");
+    expect(inputs).toHaveLength(1);
+    expect(Buffer.from(inputs[0].data as string, "base64").toString("utf8")).toBe("ab");
+  });
+
+  it("is idempotent — detach() after a chord does not re-fire onDetach", async () => {
+    const { stdin, onDetach, handle } = await startAttach();
+    stdin.feed(Buffer.from([DETACH_CHORD_BYTE]));
+    await handle.done;
+    handle.detach();
+    expect(onDetach).toHaveBeenCalledTimes(1);
+  });
+});
+
+// ── Error / drop paths ─────────────────────────────────────────────────────────
+
+describe("WS close mid-attach surfaces error", () => {
+  it("close before exit → onDetach(error) and terminal restored", async () => {
+    const { ws, stdout, stdin, onDetach, handle } = await startAttach();
+    ws.close();
+    await handle.done;
+    expect(onDetach).toHaveBeenCalledTimes(1);
+    expect(onDetach.mock.calls[0][0]).toBeInstanceOf(Error);
+    expect(stdout.all()).toContain(ALT_SCREEN_LEAVE);
+    expect(stdin.rawCalls.at(-1)).toBe(false);
+  });
+
+  it("WS error → onDetach(error) and clean restore", async () => {
+    const { ws, stdout, onDetach, handle } = await startAttach();
+    ws.emit("error", new Error("boom"));
+    await handle.done;
+    expect(onDetach).toHaveBeenCalledTimes(1);
+    expect((onDetach.mock.calls[0][0] as Error).message).toBe("boom");
+    expect(stdout.all()).toContain(ALT_SCREEN_LEAVE);
+  });
+
+  it("server `exit` frame ends the attach cleanly (no error)", async () => {
+    const { ws, onDetach, handle } = await startAttach();
+    ws.deliver({ type: "exit" });
+    await handle.done;
+    expect(onDetach).toHaveBeenCalledWith(undefined);
+  });
+
+  it("failed ticket mint surfaces the error without opening the WS", async () => {
+    const onDetach = vi.fn();
+    const wsFactory = vi.fn();
+    const handle = attachTerminalSession({
+      baseUrl: "http://127.0.0.1:9999",
+      sessionId: "s",
+      stdin: new FakeStdin(),
+      stdout: new FakeStdout(),
+      onDetach,
+      fetchImpl: vi.fn(async () => new Response("x", { status: 500, statusText: "Err" })) as unknown as typeof fetch,
+      wsFactory: wsFactory as never,
+    });
+    await handle.done;
+    expect(wsFactory).not.toHaveBeenCalled();
+    expect(onDetach).toHaveBeenCalledTimes(1);
+    expect(onDetach.mock.calls[0][0]).toBeInstanceOf(Error);
+  });
+});
+
+// ── Output neutralization (full U10 parity set) ─────────────────────────────────
+
+describe("output neutralization before stdout", () => {
+  it("strips OSC 52 clipboard-write", async () => {
+    const { stdout, ws } = await startAttach();
+    const hostile = `before\x1b]52;c;${Buffer.from("stolen").toString("base64")}\x07after`;
+    ws.deliver({ type: "data", data: b64(hostile) });
+    const out = stdout.all();
+    expect(out).toContain("before");
+    expect(out).toContain("after");
+    expect(out).not.toContain("52;c;");
+  });
+
+  it("strips a non-http(s) (javascript:) OSC 8 link URI but keeps the text", async () => {
+    const { stdout, ws } = await startAttach();
+    const hostile = `\x1b]8;;javascript:alert(1)\x07click me\x1b]8;;\x07`;
+    ws.deliver({ type: "data", data: b64(hostile) });
+    const out = stdout.all();
+    expect(out).toContain("click me");
+    expect(out).not.toContain("javascript:alert(1)");
+  });
+
+  it("passes an http(s) OSC 8 link through", async () => {
+    const { stdout, ws } = await startAttach();
+    const safe = `\x1b]8;;https://example.com\x07link\x1b]8;;\x07`;
+    ws.deliver({ type: "data", data: b64(safe) });
+    expect(stdout.all()).toContain("https://example.com");
+  });
+
+  it("strips a DSR device-status query (would forge input)", async () => {
+    const { stdout, ws } = await startAttach();
+    const hostile = "x\x1b[6ny"; // DSR cursor-position report
+    ws.deliver({ type: "data", data: b64(hostile) });
+    const out = stdout.all();
+    expect(out).toContain("x");
+    expect(out).toContain("y");
+    expect(out).not.toContain("\x1b[6n");
+  });
+
+  it("neutralizes a sequence split across two frames", async () => {
+    const { stdout, ws } = await startAttach();
+    // Split an OSC 52 across two data frames.
+    const part1 = `safe\x1b]52;c;${Buffer.from("secret").toString("base64")}`;
+    const part2 = `\x07tail`;
+    ws.deliver({ type: "data", data: b64(part1) });
+    ws.deliver({ type: "data", data: b64(part2) });
+    const out = stdout.all();
+    expect(out).toContain("safe");
+    expect(out).toContain("tail");
+    expect(out).not.toContain("52;c;");
+  });
+});
+
+// ── ACK flow control ───────────────────────────────────────────────────────────
+
+describe("ACK flow control", () => {
+  it("emits an ACK after the threshold bytes are written", async () => {
+    const { stdout, ws } = await startAttach({ ackThresholdBytes: 64 });
+    void stdout;
+    // 100 bytes of benign output → crosses the 64-byte threshold once.
+    ws.deliver({ type: "data", data: b64("a".repeat(100)) });
+    const acks = ws.parsedSent().filter((f) => f.type === "ack");
+    expect(acks).toHaveLength(1);
+    expect(acks[0].bytes).toBeGreaterThanOrEqual(64);
+  });
+
+  it("does not ACK below the threshold", async () => {
+    const { ws } = await startAttach({ ackThresholdBytes: 1024 });
+    ws.deliver({ type: "data", data: b64("short") });
+    expect(ws.parsedSent().filter((f) => f.type === "ack")).toHaveLength(0);
+  });
+});

packages/cli/src/commands/dashboard-tui/controller.ts

@@ -134,6 +134,14 @@ export class DashboardTUI {
     waitUntilExit: () => Promise<unknown>;
     clear?: () => void;
   } & Record<string, unknown> | null = null;
+  // Captured at start() so a full-screen terminal attach (U14) can unmount Ink,
+  // hand the TTY to the passthrough loop, then remount the same app on detach.
+  private renderApp: (() => unknown) | null = null;
+  private inkRender: ((node: unknown) => typeof this.inkInstance) | null = null;
+  // True while a full-screen terminal attach owns the TTY; suppresses Ink
+  // re-render/resize work that would corrupt the passthrough surface.
+  private terminalAttachActive = false;
+
   // Resize listener attached at start(), detached at stop().
   private resizeListener: (() => void) | null = null;
   // Debounce timer for resize handling — coalesces tmux/ssh resize bursts.
@@ -705,6 +713,12 @@ export class DashboardTUI {
       process.stdout.write("\x1b[?1049h\x1b[H");
     }
 
+    // Capture the app element factory + render fn so openTerminalAttach() can
+    // remount the identical tree after a full-screen passthrough detaches.
+    this.renderApp = () =>
+      createElement(I18nextProvider, { i18n }, createElement(DashboardApp, { controller: this }));
+    this.inkRender = (node: unknown) => render(node as Parameters<typeof render>[0]);
+
     this.inkInstance = render(
       createElement(I18nextProvider, { i18n }, createElement(DashboardApp, { controller: this })),
     );
@@ -899,6 +913,82 @@ export class DashboardTUI {
     }
   }
 
+  /**
+   * Open a CLI-agent session as a full-screen passthrough (U14). Suspend-and-
+   * handoff: unmount Ink (releasing its raw-mode / stdin grip), let
+   * `attachTerminalSession` own the alt-screen + raw mode for the passthrough
+   * loop, then remount the same Ink app once the user detaches (Ctrl-]) or the
+   * session ends. Resolves after the TUI has been remounted.
+   *
+   * No-op (resolves immediately) when there's no session info / not running.
+   */
+  async openTerminalAttach(sessionId: string, projectId?: string): Promise<void> {
+    if (!this.isRunning || this.terminalAttachActive) return;
+    if (!this.renderApp || !this.inkRender) return;
+    const baseUrl = this.systemInfo?.baseUrl;
+    if (!baseUrl) return;
+    const token = this.systemInfo?.authToken;
+
+    const { attachTerminalSession } = await import("./terminal-attach.js");
+
+    this.terminalAttachActive = true;
+
+    // Unmount Ink so it relinquishes raw mode + the stdin 'data' grip; the
+    // passthrough loop installs its own listeners on the bare stdin/stdout.
+    // Also drop our mouse listener so wheel reports don't leak into the PTY.
+    this.uninstallMouseListener();
+    if (this.inkInstance) {
+      try {
+        this.inkInstance.unmount();
+      } catch {
+        /* ignore */
+      }
+      this.inkInstance = null;
+    }
+    // Leave Ink's alt-screen; the passthrough enters its own.
+    if (process.stdout?.isTTY && typeof process.stdout.write === "function") {
+      try {
+        process.stdout.write("\x1b[?1049l");
+      } catch {
+        /* ignore */
+      }
+    }
+
+    await new Promise<void>((resolve) => {
+      const handle = attachTerminalSession({
+        baseUrl,
+        token,
+        sessionId,
+        projectId,
+        stdin: process.stdin as unknown as import("./terminal-attach.js").AttachStdin,
+        stdout: process.stdout as unknown as import("./terminal-attach.js").AttachStdout,
+        onDetach: (error) => {
+          if (error) {
+            this.error(`Terminal session detached: ${error.message}`, "cli-agent");
+          }
+        },
+      });
+      void handle.done.finally(() => resolve());
+    });
+
+    // Remount Ink on a clean alt-screen.
+    this.terminalAttachActive = false;
+    if (!this.isRunning) return; // stopped while attached — leave the terminal as-is
+    if (process.stdout?.isTTY && typeof process.stdout.write === "function") {
+      try {
+        process.stdout.write("\x1b[?1049h\x1b[H");
+      } catch {
+        /* ignore */
+      }
+    }
+    try {
+      this.inkInstance = this.inkRender(this.renderApp());
+    } catch {
+      /* ignore — remount best-effort */
+    }
+    this.notify();
+  }
+
   // ── Private helpers ────────────────────────────────────────────────────────
 
   // Attach a parallel `data` listener that decodes xterm SGR mouse

packages/cli/src/commands/dashboard-tui/terminal-attach.ts

@@ -0,0 +1,497 @@
+/**
+ * terminal-attach — full-screen passthrough attach to a CLI agent session
+ * from the Ink TUI (CLI Agent Executor, U14).
+ *
+ * Model: SUSPEND-AND-HANDOFF, not embedding. The caller pauses Ink rendering,
+ * then `attachTerminalSession` takes over the real TTY:
+ *   - enter the alternate screen (`\x1b[?1049h`) and put stdin in raw mode,
+ *   - WS `scrollback`/`data` frames → neutralize (U10 filter) → write to stdout,
+ *   - stdin bytes → WS `input` frames (base64),
+ *   - SIGWINCH / stdout resize → WS `resize` frames,
+ *   - ACK `bytes` consumed after every ~32KB written, for flow control,
+ *   - detach chord Ctrl-] (0x1d) → leave alt-screen, restore raw mode, close WS,
+ *   - WS close/error mid-attach → restore the terminal cleanly + surface via
+ *     `onDetach(error)`.
+ *
+ * SECURITY (the riskiest leg): the byte stream is UNTRUSTED. The host terminal
+ * honors more escape sequences than xterm.js, so every byte written to the host
+ * TTY is passed through `neutralizeTerminalOutput` FIRST — the identical filter
+ * the dashboard WS bridge uses (re-exported from `@fusion/dashboard`). OSC 52
+ * clipboard writes, OSC 8 non-http(s) links, and device-status queries (whose
+ * auto-responses forge input) are stripped before they ever reach the terminal.
+ *
+ * CJK / double-width: bytes pass through verbatim — no width math is needed in a
+ * passthrough (the host terminal does the width handling).
+ *
+ * The WS transport is injectable (`wsFactory`) so tests drive the loop with an
+ * in-memory WS-like object and NEVER open a real socket (and never touch port
+ * 4040). The default factory uses the `ws` Node client.
+ */
+
+import { WebSocket } from "ws";
+import { neutralizeTerminalOutput, flushTerminalOutput } from "@fusion/dashboard";
+
+/** Detach chord: Ctrl-] (GS, 0x1d). Documented + shown in the status hint. */
+export const DETACH_CHORD_BYTE = 0x1d;
+/** Human-readable label for the detach chord (status hint). */
+export const DETACH_CHORD_LABEL = "Ctrl-]";
+
+/** Enter / leave the alternate screen buffer. */
+export const ALT_SCREEN_ENTER = "\x1b[?1049h";
+export const ALT_SCREEN_LEAVE = "\x1b[?1049l";
+
+/** Emit an ACK after roughly this many bytes are written to stdout. */
+export const DEFAULT_ACK_THRESHOLD_BYTES = 32 * 1024;
+
+// ── Frame shapes (mirror packages/dashboard/src/cli-session-ws.ts) ──────────
+
+/** Server → client frames. */
+type ServerFrame =
+  | { type: "scrollback"; data?: string }
+  | { type: "data"; data?: string }
+  | { type: "state"; [k: string]: unknown }
+  | { type: "error"; message?: string; code?: string }
+  | { type: "exit" };
+
+/** Client → server frames. */
+type ClientFrame =
+  | { type: "input"; data: string }
+  | { type: "resize"; cols: number; rows: number }
+  | { type: "ack"; bytes: number };
+
+/**
+ * The minimal WebSocket surface the passthrough loop uses. The real `ws` client
+ * satisfies this; tests provide an in-memory implementation.
+ */
+export interface TerminalWebSocket {
+  /** Register an event handler. */
+  on(event: "open", listener: () => void): void;
+  on(event: "message", listener: (data: unknown) => void): void;
+  on(event: "close", listener: (code?: number, reason?: unknown) => void): void;
+  on(event: "error", listener: (err: Error) => void): void;
+  /** Send a (string) frame. */
+  send(data: string): void;
+  /** Close the socket. */
+  close(code?: number, reason?: string): void;
+  /** Current ready state; OPEN === 1 (matches the ws/WHATWG constant). */
+  readyState: number;
+}
+
+/** Ready-state constant matching the `ws` client / WHATWG WebSocket. */
+export const WS_OPEN = 1;
+
+/** Factory that opens a WS connection to `url` with the given headers. */
+export type TerminalWebSocketFactory = (
+  url: string,
+  headers: Record<string, string>,
+) => TerminalWebSocket;
+
+/** Minimal readable stdin surface (a TTY ReadStream satisfies this). */
+export interface AttachStdin {
+  on(event: "data", listener: (chunk: Buffer | string) => void): void;
+  off(event: "data", listener: (chunk: Buffer | string) => void): void;
+  setRawMode?: (mode: boolean) => void;
+  isRaw?: boolean;
+  isTTY?: boolean;
+  resume?: () => void;
+  pause?: () => void;
+}
+
+/** Minimal writable stdout surface (a TTY WriteStream satisfies this). */
+export interface AttachStdout {
+  write(chunk: string): void;
+  columns?: number;
+  rows?: number;
+  on(event: "resize", listener: () => void): void;
+  off(event: "resize", listener: () => void): void;
+}
+
+export interface AttachTerminalSessionOptions {
+  /** Dashboard base URL, e.g. `http://127.0.0.1:4040`. */
+  baseUrl: string;
+  /** Daemon token (Authorization: Bearer …). Optional when auth is disabled. */
+  token?: string;
+  /** Session id to attach to. */
+  sessionId: string;
+  /** Project id (scopes the attach-ticket mint), if known. */
+  projectId?: string;
+  stdin: AttachStdin;
+  stdout: AttachStdout;
+  /**
+   * Called exactly once when the attach ends — cleanly (no arg) or with an error
+   * (WS drop / failed ticket). The caller resumes Ink rendering here.
+   */
+  onDetach: (error?: Error) => void;
+  /** Injectable WS factory (default: the `ws` Node client). */
+  wsFactory?: TerminalWebSocketFactory;
+  /** Injectable fetch (default: global fetch) for the attach-ticket POST. */
+  fetchImpl?: typeof fetch;
+  /** ACK threshold override (bytes). */
+  ackThresholdBytes?: number;
+  /** Print a one-line detach hint before entering the alt-screen. */
+  printHint?: boolean;
+}
+
+/** Handle returned by `attachTerminalSession`; lets the caller force-detach. */
+export interface AttachHandle {
+  /** Resolves when the attach fully ends (after terminal restore + onDetach). */
+  done: Promise<void>;
+  /** Force a clean detach (e.g. the TUI is quitting). Idempotent. */
+  detach(): void;
+}
+
+interface AttachTicketResponse {
+  ticket: string;
+  expiresAt?: string;
+  readOnly?: boolean;
+}
+
+/**
+ * Fetch a single-use attach ticket for the session. Throws on non-2xx so the
+ * caller surfaces a clean error and never opens the WS.
+ */
+export async function fetchAttachTicket(opts: {
+  baseUrl: string;
+  token?: string;
+  sessionId: string;
+  projectId?: string;
+  fetchImpl?: typeof fetch;
+}): Promise<AttachTicketResponse> {
+  const fetchFn = opts.fetchImpl ?? fetch;
+  const url = `${opts.baseUrl.replace(/\/$/, "")}/api/cli-sessions/${encodeURIComponent(
+    opts.sessionId,
+  )}/attach-ticket`;
+  const headers: Record<string, string> = { "content-type": "application/json" };
+  if (opts.token) headers.authorization = `Bearer ${opts.token}`;
+  const res = await fetchFn(url, {
+    method: "POST",
+    headers,
+    body: JSON.stringify(opts.projectId ? { projectId: opts.projectId } : {}),
+  });
+  if (!res.ok) {
+    throw new Error(
+      `Failed to mint attach ticket (HTTP ${res.status} ${res.statusText})`,
+    );
+  }
+  const body = (await res.json()) as AttachTicketResponse;
+  if (!body || typeof body.ticket !== "string" || body.ticket.length === 0) {
+    throw new Error("Attach-ticket response missing `ticket`");
+  }
+  return body;
+}
+
+/** Build the cli-session WS URL with sessionId + ticket query params. */
+export function buildWsUrl(opts: {
+  baseUrl: string;
+  sessionId: string;
+  ticket: string;
+}): string {
+  const u = new URL(`${opts.baseUrl.replace(/\/$/, "")}/api/cli-sessions/ws`);
+  // ws(s):// scheme — derive from http(s).
+  u.protocol = u.protocol === "https:" ? "wss:" : "ws:";
+  u.searchParams.set("sessionId", opts.sessionId);
+  u.searchParams.set("ticket", opts.ticket);
+  return u.toString();
+}
+
+function defaultWsFactory(): TerminalWebSocketFactory {
+  return (url, headers) => {
+    const ws = new WebSocket(url, { headers });
+    return ws as unknown as TerminalWebSocket;
+  };
+}
+
+/**
+ * Decode a server `data`/`scrollback` frame's base64 payload to a UTF-8 string.
+ */
+function decodeFrameData(data: string | undefined): string {
+  if (typeof data !== "string" || data.length === 0) return "";
+  return Buffer.from(data, "base64").toString("utf8");
+}
+
+/**
+ * Run the full-screen passthrough attach. Returns once the attach has fully
+ * ended and the terminal has been restored (the same point `onDetach` fires).
+ *
+ * Lifecycle is single-shot: every termination path (detach chord, WS close, WS
+ * error, ticket failure, force `detach()`) funnels through one idempotent
+ * teardown that restores raw mode, leaves the alt-screen, closes the WS, and
+ * fires `onDetach` exactly once.
+ */
+export function attachTerminalSession(
+  opts: AttachTerminalSessionOptions,
+): AttachHandle {
+  const {
+    stdin,
+    stdout,
+    onDetach,
+    ackThresholdBytes = DEFAULT_ACK_THRESHOLD_BYTES,
+  } = opts;
+  const wsFactory = opts.wsFactory ?? defaultWsFactory();
+
+  let settled = false;
+  let resolveDone: () => void;
+  const done = new Promise<void>((resolve) => {
+    resolveDone = resolve;
+  });
+
+  // Terminal state we must restore on teardown.
+  const priorRaw = stdin.isRaw ?? false;
+  let enteredAltScreen = false;
+  let rawModeSet = false;
+
+  // Live wiring (set once the WS opens).
+  let ws: TerminalWebSocket | null = null;
+  let onStdinData: ((chunk: Buffer | string) => void) | null = null;
+  let onResize: (() => void) | null = null;
+
+  // Outbound neutralization carry (threaded across data frames so a sequence
+  // split across two frames is still caught).
+  let carry = "";
+  // Flow control: bytes written since the last ACK.
+  let bytesSinceAck = 0;
+
+  const sendFrame = (frame: ClientFrame): void => {
+    if (!ws || ws.readyState !== WS_OPEN) return;
+    try {
+      ws.send(JSON.stringify(frame));
+    } catch {
+      /* socket closing */
+    }
+  };
+
+  const ackConsumed = (n: number): void => {
+    bytesSinceAck += n;
+    if (bytesSinceAck >= ackThresholdBytes) {
+      sendFrame({ type: "ack", bytes: bytesSinceAck });
+      bytesSinceAck = 0;
+    }
+  };
+
+  // Write a (possibly partial) untrusted chunk to the host TTY through the U10
+  // neutralizer. `isSnapshot` flushes the carry (scrollback is a complete unit).
+  const writeNeutralized = (text: string, isSnapshot: boolean): void => {
+    const result = neutralizeTerminalOutput(text, carry);
+    let out = result.output;
+    if (isSnapshot) {
+      out += flushTerminalOutput(result.carry);
+      carry = "";
+    } else {
+      carry = result.carry;
+    }
+    if (out.length === 0) return;
+    try {
+      stdout.write(out);
+    } catch {
+      /* stdout closing */
+    }
+    ackConsumed(Buffer.byteLength(out, "utf8"));
+  };
+
+  const teardown = (error?: Error): void => {
+    if (settled) return;
+    settled = true;
+
+    // Detach stdin/resize listeners first so no late bytes race the restore.
+    if (onStdinData) {
+      try {
+        stdin.off("data", onStdinData);
+      } catch {
+        /* ignore */
+      }
+      onStdinData = null;
+    }
+    if (onResize) {
+      try {
+        stdout.off("resize", onResize);
+      } catch {
+        /* ignore */
+      }
+      onResize = null;
+    }
+
+    // Restore raw mode to its prior state (only if we changed it).
+    if (rawModeSet && stdin.setRawMode) {
+      try {
+        stdin.setRawMode(priorRaw);
+      } catch {
+        /* ignore */
+      }
+    }
+
+    // Leave the alt-screen so the caller's shell / Ink scrollback is restored.
+    if (enteredAltScreen) {
+      try {
+        stdout.write(ALT_SCREEN_LEAVE);
+      } catch {
+        /* ignore */
+      }
+    }
+
+    // Close the WS (never throws upward).
+    if (ws) {
+      try {
+        ws.close();
+      } catch {
+        /* ignore */
+      }
+    }
+
+    try {
+      onDetach(error);
+    } finally {
+      resolveDone();
+    }
+  };
+
+  const handleServerFrame = (frame: ServerFrame): void => {
+    switch (frame.type) {
+      case "scrollback":
+        writeNeutralized(decodeFrameData(frame.data), true);
+        break;
+      case "data":
+        writeNeutralized(decodeFrameData(frame.data), false);
+        break;
+      case "exit":
+        teardown();
+        break;
+      case "error":
+        // A server error frame (e.g. read-only) is informational; surface it on
+        // stdout but don't tear down — the stream may continue.
+        if (frame.message) {
+          try {
+            stdout.write(`\r\n[session] ${frame.message}\r\n`);
+          } catch {
+            /* ignore */
+          }
+        }
+        break;
+      case "state":
+      default:
+        break;
+    }
+  };
+
+  // Enter the alt-screen + raw mode, then wire the loop. We do this BEFORE the
+  // WS opens so the first scrollback frame lands on a clean alt-screen.
+  const enterPassthrough = (): void => {
+    if (opts.printHint !== false) {
+      try {
+        stdout.write(
+          `Attached to session ${opts.sessionId}. Press ${DETACH_CHORD_LABEL} to detach.\r\n`,
+        );
+      } catch {
+        /* ignore */
+      }
+    }
+    try {
+      stdout.write(ALT_SCREEN_ENTER);
+      enteredAltScreen = true;
+    } catch {
+      /* ignore */
+    }
+    if (stdin.setRawMode) {
+      try {
+        stdin.setRawMode(true);
+        rawModeSet = true;
+      } catch {
+        /* ignore */
+      }
+    }
+    stdin.resume?.();
+
+    // stdin → input frames; detach chord intercepted.
+    onStdinData = (chunk: Buffer | string): void => {
+      const buf = typeof chunk === "string" ? Buffer.from(chunk, "utf8") : chunk;
+      // Detach chord: if Ctrl-] appears, send any bytes before it, then detach.
+      const idx = buf.indexOf(DETACH_CHORD_BYTE);
+      if (idx !== -1) {
+        if (idx > 0) {
+          sendFrame({ type: "input", data: buf.subarray(0, idx).toString("base64") });
+        }
+        teardown();
+        return;
+      }
+      sendFrame({ type: "input", data: buf.toString("base64") });
+    };
+    stdin.on("data", onStdinData);
+
+    // stdout resize → resize frames.
+    onResize = (): void => {
+      const cols = stdout.columns;
+      const rows = stdout.rows;
+      if (typeof cols === "number" && typeof rows === "number") {
+        sendFrame({ type: "resize", cols, rows });
+      }
+    };
+    stdout.on("resize", onResize);
+
+    // Send the initial size so the PTY matches the host TTY immediately.
+    onResize();
+  };
+
+  // ── Kick off: mint ticket, open WS, run the loop ──
+  (async () => {
+    let ticket: AttachTicketResponse;
+    try {
+      ticket = await fetchAttachTicket({
+        baseUrl: opts.baseUrl,
+        token: opts.token,
+        sessionId: opts.sessionId,
+        projectId: opts.projectId,
+        fetchImpl: opts.fetchImpl,
+      });
+    } catch (err) {
+      teardown(err instanceof Error ? err : new Error(String(err)));
+      return;
+    }
+
+    const url = buildWsUrl({
+      baseUrl: opts.baseUrl,
+      sessionId: opts.sessionId,
+      ticket: ticket.ticket,
+    });
+    const headers: Record<string, string> = {};
+    if (opts.token) headers.authorization = `Bearer ${opts.token}`;
+
+    try {
+      ws = wsFactory(url, headers);
+    } catch (err) {
+      teardown(err instanceof Error ? err : new Error(String(err)));
+      return;
+    }
+
+    ws.on("open", () => {
+      enterPassthrough();
+    });
+    ws.on("message", (data: unknown) => {
+      let text: string;
+      if (typeof data === "string") text = data;
+      else if (Buffer.isBuffer(data)) text = data.toString("utf8");
+      else if (data instanceof Uint8Array) text = Buffer.from(data).toString("utf8");
+      else text = String(data);
+      let frame: ServerFrame;
+      try {
+        frame = JSON.parse(text) as ServerFrame;
+      } catch {
+        return; // ignore malformed
+      }
+      handleServerFrame(frame);
+    });
+    ws.on("close", () => {
+      // A close before any deliberate detach is treated as a clean end if the
+      // server sent `exit` (already torn down), otherwise as a mid-attach drop.
+      if (!settled) {
+        teardown(new Error("Connection closed"));
+      }
+    });
+    ws.on("error", (err: Error) => {
+      teardown(err instanceof Error ? err : new Error(String(err)));
+    });
+  })();
+
+  return {
+    done,
+    detach: () => teardown(),
+  };
+}

packages/dashboard/src/index.ts

@@ -66,3 +66,14 @@ export {
 } from "./badge-pubsub.js";
 
 export * from "./plugins/index.js";
+
+// CLI-session terminal-output hardening — re-exported so the TUI passthrough
+// (packages/cli, U14) applies the IDENTICAL neutralization set as the dashboard
+// WS bridge (U10). The host TTY honors more escape sequences than xterm.js, so
+// the TUI MUST reuse this single implementation rather than fork it.
+export {
+  neutralizeTerminalOutput,
+  flushTerminalOutput,
+  MAX_CARRY_LENGTH,
+  type NeutralizeResult,
+} from "./cli-session-output-filter.js";

pnpm-lock.yaml

@@ -83,6 +83,9 @@ importers:
       react-i18next:
         specifier: ^17.0.8
         version: 17.0.8(i18next@26.3.1(typescript@5.9.3))(react-dom@19.2.4(react@19.2.4))(react@19.2.4)(typescript@5.9.3)
+      ws:
+        specifier: ^8.18.0
+        version: 8.20.0
     devDependencies:
       '@fusion/core':
         specifier: workspace:*
@@ -108,6 +111,9 @@ importers:
       '@types/react':
         specifier: ^19.2.0
         version: 19.2.14
+      '@types/ws':
+        specifier: ^8.5.0
+        version: 8.18.1
       '@vitest/coverage-v8':
         specifier: ^3.1.0
         version: 3.2.4(vitest@3.2.4(@types/debug@4.1.13)(@types/node@25.5.2)(happy-dom@20.10.1)(jiti@2.7.0)(jsdom@29.0.1)(tsx@4.21.0)(yaml@2.8.3))