From b2cdf7adabc6a22415e6a8174dcd66cdbaa95acd Mon Sep 17 00:00:00 2001 From: Chirag Arora <76108151+ChiragArora31@users.noreply.github.com> Date: Fri, 1 May 2026 16:15:43 +0530 Subject: [PATCH] fix(web): scope terminal tmux resolution by project (#1551) * fix(web): scope terminal resolution by project * fix(web): avoid suffix false-positives in project-scoped tmux key match Made-with: Cursor * fix(web): scope fullscreen terminal resize by project Made-with: Cursor --- .../web/server/__tests__/tmux-utils.test.ts | 60 ++++++ packages/web/server/mux-websocket.ts | 116 ++++++----- packages/web/server/tmux-utils.ts | 19 +- .../web/src/components/DirectTerminal.tsx | 5 +- packages/web/src/components/SessionDetail.tsx | 1 + .../__tests__/DirectTerminal.render.test.tsx | 21 ++ .../terminal/useFullscreenResize.ts | 5 +- .../components/terminal/useXtermTerminal.ts | 90 ++++---- packages/web/src/lib/mux-protocol.ts | 18 +- packages/web/src/providers/MuxProvider.tsx | 110 ++++++---- .../providers/__tests__/MuxProvider.test.tsx | 193 +++++++++++++----- 11 files changed, 452 insertions(+), 186 deletions(-) diff --git a/packages/web/server/__tests__/tmux-utils.test.ts b/packages/web/server/__tests__/tmux-utils.test.ts index 58cf92f31..5a863eaf7 100644 --- a/packages/web/server/__tests__/tmux-utils.test.ts +++ b/packages/web/server/__tests__/tmux-utils.test.ts @@ -775,6 +775,66 @@ describe("resolveTmuxSession", () => { ); }); + it("prefers the project-scoped storageKey when session IDs collide", () => { + const fs = { + readdir: () => [ + "aaaaaaaaaaaa-alpha", + "bbbbbbbbbbbb-beta", + ], + exists: (p: string) => + p.endsWith("/aaaaaaaaaaaa-alpha/sessions/app-1") || + p.endsWith("/bbbbbbbbbbbb-beta/sessions/app-1"), + homedir: () => "/home/user", + }; + const mockExec = vi.fn() + .mockImplementationOnce(() => { + throw new Error("session not found"); // exact match fails + }) + .mockImplementationOnce(() => { + return ""; // project-scoped candidate succeeds + }); + + const result = resolveTmuxSession("app-1", TMUX, mockExec, fs, "beta"); + + expect(result).toBe("bbbbbbbbbbbb-beta-app-1"); + expect(mockExec).toHaveBeenNthCalledWith( + 2, + TMUX, + ["has-session", "-t", "=bbbbbbbbbbbb-beta-app-1"], + { timeout: 5000 }, + ); + }); + + it("does not treat suffix project names as exact project matches", () => { + const fs = { + readdir: () => [ + "aaaaaaaaaaaa-my-app", + "bbbbbbbbbbbb-app", + ], + exists: (p: string) => + p.endsWith("/aaaaaaaaaaaa-my-app/sessions/app-1") || + p.endsWith("/bbbbbbbbbbbb-app/sessions/app-1"), + homedir: () => "/home/user", + }; + const mockExec = vi.fn() + .mockImplementationOnce(() => { + throw new Error("session not found"); // exact match fails + }) + .mockImplementationOnce(() => { + return ""; // correctly probes exact project match first + }); + + const result = resolveTmuxSession("app-1", TMUX, mockExec, fs, "app"); + + expect(result).toBe("bbbbbbbbbbbb-app-app-1"); + expect(mockExec).toHaveBeenNthCalledWith( + 2, + TMUX, + ["has-session", "-t", "=bbbbbbbbbbbb-app-app-1"], + { timeout: 5000 }, + ); + }); + it("accepts wrapped storageKeys with spaces/unicode in the project name", () => { // Legacy storageKeys use basename(projectPath), which has no character // restrictions on-disk. Regexes that reject spaces or unicode would diff --git a/packages/web/server/mux-websocket.ts b/packages/web/server/mux-websocket.ts index dfec567b3..ee18764a8 100644 --- a/packages/web/server/mux-websocket.ts +++ b/packages/web/server/mux-websocket.ts @@ -17,33 +17,26 @@ import { findTmux, resolveTmuxSession, validateSessionId } from "./tmux-utils.js // ── Client → Server ── type ClientMessage = - | { ch: "terminal"; id: string; type: "data"; data: string } - | { ch: "terminal"; id: string; type: "resize"; cols: number; rows: number } - | { ch: "terminal"; id: string; type: "open"; tmuxName?: string } - | { ch: "terminal"; id: string; type: "close" } + | { ch: "terminal"; id: string; type: "data"; data: string; projectId?: string } + | { ch: "terminal"; id: string; type: "resize"; cols: number; rows: number; projectId?: string } + | { ch: "terminal"; id: string; type: "open"; projectId?: string; tmuxName?: string } + | { ch: "terminal"; id: string; type: "close"; projectId?: string } | { ch: "system"; type: "ping" } - | { ch: "subscribe"; topics: ("sessions")[] }; + | { ch: "subscribe"; topics: "sessions"[] }; // ── Server → Client ── type ServerMessage = - | { ch: "terminal"; id: string; type: "data"; data: string } - | { ch: "terminal"; id: string; type: "exited"; code: number } - | { ch: "terminal"; id: string; type: "opened" } - | { ch: "terminal"; id: string; type: "error"; message: string } + | { ch: "terminal"; id: string; type: "data"; data: string; projectId?: string } + | { ch: "terminal"; id: string; type: "exited"; code: number; projectId?: string } + | { ch: "terminal"; id: string; type: "opened"; projectId?: string } + | { ch: "terminal"; id: string; type: "error"; message: string; projectId?: string } | { ch: "sessions"; type: "snapshot"; sessions: SessionPatch[] } | { ch: "sessions"; type: "error"; error: string } | { ch: "system"; type: "pong" } | { ch: "system"; type: "error"; message: string }; // Mirrors AttentionLevel in src/lib/types.ts — keep in sync. -type AttentionLevel = - | "merge" - | "action" - | "respond" - | "review" - | "pending" - | "working" - | "done"; +type AttentionLevel = "merge" | "action" | "respond" | "review" | "pending" | "working" | "done"; interface SessionPatch { id: string; @@ -211,25 +204,33 @@ class TerminalManager { this.TMUX = tmuxPath ?? findTmux(); } + private terminalKey(id: string, projectId?: string): string { + return projectId ? `${projectId}:${id}` : id; + } + /** * Open/attach to a terminal. If already open, just return. * If has subscribers but PTY crashed, re-attach. */ - open(id: string, tmuxName?: string): string { + open(id: string, projectId?: string, tmuxName?: string): string { // Validate and resolve if (!validateSessionId(id)) { throw new Error(`Invalid session ID: ${id}`); } // Use provided tmuxName, or reuse from existing terminal entry, or resolve - const existing = this.terminals.get(id); - const tmuxSessionId = tmuxName ?? existing?.tmuxSessionId ?? resolveTmuxSession(id, this.TMUX); + const key = this.terminalKey(id, projectId); + const existing = this.terminals.get(key); + const tmuxSessionId = + tmuxName ?? + existing?.tmuxSessionId ?? + resolveTmuxSession(id, this.TMUX, undefined, undefined, projectId); if (!tmuxSessionId) { throw new Error(`Session not found: ${id}`); } // Get or create terminal entry - let terminal = this.terminals.get(id); + let terminal = this.terminals.get(key); if (!terminal) { terminal = { id, @@ -241,7 +242,7 @@ class TerminalManager { bufferBytes: 0, reattachAttempts: 0, }; - this.terminals.set(id, terminal); + this.terminals.set(key, terminal); } // If PTY is already attached, we're done @@ -322,9 +323,11 @@ class TerminalManager { // after every attach (e.g. resource exhaustion or a broken tmux session). if (terminal.subscribers.size > 0 && terminal.reattachAttempts < MAX_REATTACH_ATTEMPTS) { terminal.reattachAttempts += 1; - console.log(`[MuxServer] Re-attaching to ${id} (attempt ${terminal.reattachAttempts}/${MAX_REATTACH_ATTEMPTS})`); + console.log( + `[MuxServer] Re-attaching to ${id} (attempt ${terminal.reattachAttempts}/${MAX_REATTACH_ATTEMPTS})`, + ); try { - this.open(id); + this.open(id, projectId); terminal.reattachAttempts = 0; // reset on successful attach return; // re-attached — don't notify exit } catch (err) { @@ -347,8 +350,8 @@ class TerminalManager { /** * Write data to the PTY if attached */ - write(id: string, data: string): void { - const terminal = this.terminals.get(id); + write(id: string, data: string, projectId?: string): void { + const terminal = this.terminals.get(this.terminalKey(id, projectId)); if (terminal?.pty) { terminal.pty.write(data); } @@ -357,8 +360,8 @@ class TerminalManager { /** * Resize the PTY if attached */ - resize(id: string, cols: number, rows: number): void { - const terminal = this.terminals.get(id); + resize(id: string, cols: number, rows: number, projectId?: string): void { + const terminal = this.terminals.get(this.terminalKey(id, projectId)); if (terminal?.pty) { terminal.pty.resize(cols, rows); } @@ -369,10 +372,16 @@ class TerminalManager { * Automatically opens the terminal if needed. * @param onExit - called when the PTY exits and cannot be re-attached */ - subscribe(id: string, callback: (data: string) => void, onExit?: (exitCode: number) => void): () => void { + subscribe( + id: string, + projectId: string | undefined, + callback: (data: string) => void, + onExit?: (exitCode: number) => void, + ): () => void { // Ensure terminal is open - this.open(id); - const terminal = this.terminals.get(id); + this.open(id, projectId); + const key = this.terminalKey(id, projectId); + const terminal = this.terminals.get(key); if (!terminal) { throw new Error(`Failed to open terminal: ${id}`); } @@ -391,7 +400,7 @@ class TerminalManager { terminal.pty.kill(); terminal.pty = null; } - this.terminals.delete(id); + this.terminals.delete(key); } }; } @@ -399,12 +408,11 @@ class TerminalManager { /** * Get buffered data for a terminal */ - getBuffer(id: string): string { - const terminal = this.terminals.get(id); + getBuffer(id: string, projectId?: string): string { + const terminal = this.terminals.get(this.terminalKey(id, projectId)); if (!terminal) return ""; return terminal.buffer.join(""); } - } /** @@ -455,7 +463,6 @@ export function createMuxWebSocket(tmuxPath?: string): WebSocketServer | null { * Handle incoming messages */ ws.on("message", (data) => { - try { const msg = JSON.parse(data.toString("utf8")) as ClientMessage; @@ -466,64 +473,80 @@ export function createMuxWebSocket(tmuxPath?: string): WebSocketServer | null { } } else if (msg.ch === "terminal") { const { id, type } = msg; + const projectId = "projectId" in msg ? msg.projectId : undefined; + const subscriptionKey = projectId ? `${projectId}:${id}` : id; try { if (type === "open") { // Validate session exists - terminalManager.open(id, "tmuxName" in msg ? msg.tmuxName : undefined); + terminalManager.open(id, projectId, "tmuxName" in msg ? msg.tmuxName : undefined); // Send opened confirmation (idempotent — safe to send on re-open) - const openedMsg: ServerMessage = { ch: "terminal", id, type: "opened" }; + const openedMsg: ServerMessage = { + ch: "terminal", + id, + type: "opened", + ...(projectId && { projectId }), + }; ws.send(JSON.stringify(openedMsg)); // Subscribe and send history buffer only for new subscribers. // Skipping the buffer on re-open prevents duplicate output when // MuxProvider re-sends open for all terminals on reconnect. - if (!subscriptions.has(id)) { + if (!subscriptions.has(subscriptionKey)) { // Send buffered history to catch up the new subscriber - const buffer = terminalManager.getBuffer(id); + const buffer = terminalManager.getBuffer(id, projectId); if (buffer) { const bufferMsg: ServerMessage = { ch: "terminal", id, type: "data", data: buffer, + ...(projectId && { projectId }), }; ws.send(JSON.stringify(bufferMsg)); } const unsub = terminalManager.subscribe( id, + projectId, (data) => { const dataMsg: ServerMessage = { ch: "terminal", id, type: "data", data, + ...(projectId && { projectId }), }; if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify(dataMsg)); } }, (exitCode) => { - const exitedMsg: ServerMessage = { ch: "terminal", id, type: "exited", code: exitCode }; + const exitedMsg: ServerMessage = { + ch: "terminal", + id, + type: "exited", + code: exitCode, + ...(projectId && { projectId }), + }; if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify(exitedMsg)); } }, ); - subscriptions.set(id, unsub); + subscriptions.set(subscriptionKey, unsub); } } else if (type === "data" && "data" in msg) { - terminalManager.write(id, msg.data); + terminalManager.write(id, msg.data, projectId); } else if (type === "resize" && "cols" in msg && "rows" in msg) { - terminalManager.resize(id, msg.cols, msg.rows); + terminalManager.resize(id, msg.cols, msg.rows, projectId); } else if (type === "close") { // Unsubscribe this client only — TerminalManager is shared across // all mux connections so we must not kill the PTY here. - const unsub = subscriptions.get(id); + const unsub = subscriptions.get(subscriptionKey); if (unsub) { unsub(); - subscriptions.delete(id); + subscriptions.delete(subscriptionKey); } } } catch (err) { @@ -533,6 +556,7 @@ export function createMuxWebSocket(tmuxPath?: string): WebSocketServer | null { id, type: "error", message: err instanceof Error ? err.message : String(err), + ...(projectId && { projectId }), }; ws.send(JSON.stringify(errorMsg)); } diff --git a/packages/web/server/tmux-utils.ts b/packages/web/server/tmux-utils.ts index 0403a9a3a..039e30244 100644 --- a/packages/web/server/tmux-utils.ts +++ b/packages/web/server/tmux-utils.ts @@ -60,7 +60,11 @@ const defaultFs: FsAdapter = { * it finds a live tmux session — otherwise a stale metadata dir from * one project could shadow the live session of another. */ -function findStorageKeysForSession(sessionId: string, fs: FsAdapter): string[] { +function findStorageKeysForSession( + sessionId: string, + fs: FsAdapter, + projectId?: string, +): string[] { const aoBase = join(fs.homedir(), ".agent-orchestrator"); let entries: string[]; try { @@ -70,14 +74,20 @@ function findStorageKeysForSession(sessionId: string, fs: FsAdapter): string[] { } const matches: string[] = []; + const projectMatches: string[] = []; for (const entry of entries) { if (!STORAGE_KEY_PATTERN.test(entry)) continue; const sessionFile = join(aoBase, entry, "sessions", sessionId); if (fs.exists(sessionFile)) { - matches.push(entry); + const unwrappedProjectId = entry.slice(13); // Strip "{hash}-" prefix when present. + if (projectId && (entry === projectId || unwrappedProjectId === projectId)) { + projectMatches.push(entry); + } else { + matches.push(entry); + } } } - return matches; + return [...projectMatches, ...matches]; } /** @@ -146,6 +156,7 @@ export function resolveTmuxSession( tmuxPath: string, execFn: typeof execFileSync = execFileSync, fs: FsAdapter = defaultFs, + projectId?: string, ): string | null { // Try exact match first using = prefix for exact matching (e.g., "ao-orchestrator") // Without =, tmux uses prefix matching: "ao-1" would match "ao-15" @@ -161,7 +172,7 @@ export function resolveTmuxSession( // even when the storageKey is wrapped (`{hash}-{projectName}`). Walk // every candidate so a stale metadata dir in one project can't shadow // the live session of another project with the same sessionId. - for (const storageKey of findStorageKeysForSession(sessionId, fs)) { + for (const storageKey of findStorageKeysForSession(sessionId, fs, projectId)) { const tmuxName = `${storageKey}-${sessionId}`; try { execFn(tmuxPath, ["has-session", "-t", `=${tmuxName}`], { timeout: 5000 }); diff --git a/packages/web/src/components/DirectTerminal.tsx b/packages/web/src/components/DirectTerminal.tsx index 521f3b08c..42649b949 100644 --- a/packages/web/src/components/DirectTerminal.tsx +++ b/packages/web/src/components/DirectTerminal.tsx @@ -16,6 +16,7 @@ export { resolveMonoFontFamily } from "./terminal/terminal-font"; interface DirectTerminalProps { sessionId: string; + projectId?: string; /** Actual tmux session name. When provided, the terminal server uses it directly instead of resolving from sessionId. */ tmuxName?: string; startFullscreen?: boolean; @@ -38,6 +39,7 @@ interface DirectTerminalProps { */ export function DirectTerminal({ sessionId, + projectId, tmuxName, startFullscreen = false, variant = "agent", @@ -69,10 +71,11 @@ export function DirectTerminal({ variant, fontSize, autoFocus, + projectId, tmuxName, }); - useFullscreenResize(fullscreen, sessionId, terminalInstance, fitAddon, terminalRef); + useFullscreenResize(fullscreen, sessionId, projectId, terminalInstance, fitAddon, terminalRef); // Sync fullscreen to URL query param useEffect(() => { diff --git a/packages/web/src/components/SessionDetail.tsx b/packages/web/src/components/SessionDetail.tsx index df4b6e292..bb96b8a47 100644 --- a/packages/web/src/components/SessionDetail.tsx +++ b/packages/web/src/components/SessionDetail.tsx @@ -208,6 +208,7 @@ export function SessionDetail({ ) : ( ({ + useFullscreenResizeMock: vi.fn(), +})); vi.mock("next/navigation", () => ({ useRouter: () => ({ replace: replaceMock }), @@ -15,6 +18,10 @@ vi.mock("next-themes", () => ({ useTheme: () => ({ resolvedTheme: "dark" }), })); +vi.mock("../terminal/useFullscreenResize", () => ({ + useFullscreenResize: useFullscreenResizeMock, +})); + class MockTerminal { options: Record; parser = { @@ -105,6 +112,7 @@ describe("DirectTerminal render", () => { beforeEach(() => { searchParams = new URLSearchParams(); replaceMock.mockReset(); + useFullscreenResizeMock.mockReset(); MockWebSocket.instances = []; Object.defineProperty(document, "fonts", { configurable: true, @@ -186,4 +194,17 @@ describe("DirectTerminal render", () => { expect(terminalShell).toHaveClass("relative"); expect(terminalShell).not.toHaveClass("fixed"); }); + + it("passes projectId to fullscreen resize hook for scoped mux resize", () => { + render(); + + expect(useFullscreenResizeMock).toHaveBeenCalledWith( + false, + "app-1", + "project-a", + expect.any(Object), + expect.any(Object), + expect.any(Object), + ); + }); }); diff --git a/packages/web/src/components/terminal/useFullscreenResize.ts b/packages/web/src/components/terminal/useFullscreenResize.ts index b913ca800..734f222ba 100644 --- a/packages/web/src/components/terminal/useFullscreenResize.ts +++ b/packages/web/src/components/terminal/useFullscreenResize.ts @@ -16,6 +16,7 @@ import { useMux } from "@/hooks/useMux"; export function useFullscreenResize( fullscreen: boolean, sessionId: string, + projectId: string | undefined, terminalInstance: RefObject, fitAddon: RefObject, containerRef: RefObject, @@ -56,7 +57,7 @@ export function useFullscreenResize( fit.fit(); terminal.refresh(0, terminal.rows - 1); - resizeTerminalMux(sessionId, terminal.cols, terminal.rows); + resizeTerminalMux(sessionId, terminal.cols, terminal.rows, projectId); }; rafId = requestAnimationFrame(resizeTerminal); @@ -96,5 +97,5 @@ export function useFullscreenResize( clearTimeout(timer1); clearTimeout(timer2); }; - }, [fullscreen, sessionId, resizeTerminalMux, containerRef, fitAddon, terminalInstance]); + }, [fullscreen, sessionId, projectId, resizeTerminalMux, containerRef, fitAddon, terminalInstance]); } diff --git a/packages/web/src/components/terminal/useXtermTerminal.ts b/packages/web/src/components/terminal/useXtermTerminal.ts index 93b27c5c0..60dc8278a 100644 --- a/packages/web/src/components/terminal/useXtermTerminal.ts +++ b/packages/web/src/components/terminal/useXtermTerminal.ts @@ -23,6 +23,7 @@ export interface UseXtermTerminalOptions { variant: TerminalVariant; fontSize: number; autoFocus: boolean; + projectId?: string; /** Actual tmux session name. When provided, the terminal server uses it directly instead of resolving from sessionId. */ tmuxName?: string; } @@ -48,7 +49,7 @@ export function useXtermTerminal( sessionId: string, options: UseXtermTerminalOptions, ): UseXtermTerminalResult { - const { appearance, variant, fontSize, autoFocus, tmuxName } = options; + const { appearance, variant, fontSize, autoFocus, projectId, tmuxName } = options; const { resolvedTheme } = useTheme(); const terminalThemes = useMemo(() => buildTerminalThemes(variant), [variant]); const { @@ -131,7 +132,7 @@ export function useXtermTerminal( if (mounted && fitAddon.current) { try { fitAddon.current.fit(); - resizeTerminalMux(sessionId, terminal.cols, terminal.rows); + resizeTerminalMux(sessionId, terminal.cols, terminal.rows, projectId); } catch { // Ignore fit errors } @@ -148,14 +149,18 @@ export function useXtermTerminal( terminalInstance.current.options.fontFamily = resolveMonoFontFamily(); terminalInstance.current.clearTextureAtlas?.(); fitAddon.current.fit(); - resizeTerminalMux(sessionId, terminalInstance.current.cols, terminalInstance.current.rows); + resizeTerminalMux( + sessionId, + terminalInstance.current.cols, + terminalInstance.current.rows, + projectId, + ); } catch { // Ignore fit errors } }; // Feature-detect addEventListener — jsdom's document.fonts mock lacks it. - const fontsFace = - typeof document !== "undefined" ? document.fonts : undefined; + const fontsFace = typeof document !== "undefined" ? document.fonts : undefined; const fontsListenerAttached = !!fontsFace && typeof fontsFace.addEventListener === "function"; if (fontsListenerAttached) { @@ -168,12 +173,18 @@ export function useXtermTerminal( // In alternate buffer (tmux) there is no way to detect when the user // returned to the live tail, so the jump-to-latest button stays visible // until clicked. - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const cleanupTouchScroll = attachTouchScroll(terminal as any, (data) => { - writeTerminal(sessionId, data); - }, { - onScrollAway: () => { followOutputRef.current = false; setFollowOutput(false); }, - }); + const cleanupTouchScroll = attachTouchScroll( + terminal, + (data) => { + writeTerminal(sessionId, data, projectId); + }, + { + onScrollAway: () => { + followOutputRef.current = false; + setFollowOutput(false); + }, + }, + ); let resizeObserver: ResizeObserver | null = null; if (terminalRef.current) { @@ -181,7 +192,7 @@ export function useXtermTerminal( if (mounted && fitAddon.current) { try { fitAddon.current.fit(); - resizeTerminalMux(sessionId, terminal.cols, terminal.rows); + resizeTerminalMux(sessionId, terminal.cols, terminal.rows, projectId); } catch { // Ignore fit errors } @@ -234,24 +245,28 @@ export function useXtermTerminal( // hazard if subscribeTerminal ever fires synchronously. let programmaticScroll = false; - openTerminal(sessionId, tmuxName); + openTerminal(sessionId, projectId, tmuxName); - unsubscribe = subscribeTerminal(sessionId, (data) => { - if (selectionActive) { - writeBuffer.push(data); - bufferBytes += data.length; - if (bufferBytes > MAX_BUFFER_BYTES) { - selectionActive = false; - flushWriteBuffer(); + unsubscribe = subscribeTerminal( + sessionId, + (data) => { + if (selectionActive) { + writeBuffer.push(data); + bufferBytes += data.length; + if (bufferBytes > MAX_BUFFER_BYTES) { + selectionActive = false; + flushWriteBuffer(); + } + } else { + terminal.write(data); + if (followOutputRef.current) { + programmaticScroll = true; + terminal.scrollToBottom(); + } } - } else { - terminal.write(data); - if (followOutputRef.current) { - programmaticScroll = true; - terminal.scrollToBottom(); - } - } - }); + }, + projectId, + ); // Use xterm's onScroll event (fires with new viewportY) instead of a DOM // scroll listener — xterm v6 may update scrollTop via RAF, making DOM @@ -275,16 +290,16 @@ export function useXtermTerminal( const handleResize = () => { if (fit) { fit.fit(); - resizeTerminalMux(sessionId, terminal.cols, terminal.rows); + resizeTerminalMux(sessionId, terminal.cols, terminal.rows, projectId); } }; window.addEventListener("resize", handleResize); inputDisposable = terminal.onData((data) => { - writeTerminal(sessionId, data); + writeTerminal(sessionId, data, projectId); }); - resizeTerminalMux(sessionId, terminal.cols, terminal.rows); + resizeTerminalMux(sessionId, terminal.cols, terminal.rows, projectId); cleanup = () => { clearTimeout(deferredFitTimeout); @@ -300,7 +315,7 @@ export function useXtermTerminal( inputDisposable?.dispose(); inputDisposable = null; unsubscribe?.(); - closeTerminal(sessionId); + closeTerminal(sessionId, projectId); terminal.dispose(); }; }) @@ -320,6 +335,7 @@ export function useXtermTerminal( }, [ appearance, sessionId, + projectId, tmuxName, variant, resolvedTheme, @@ -339,8 +355,8 @@ export function useXtermTerminal( const terminal = terminalInstance.current; if (!fit || !terminal) return; fit.fit(); - resizeTerminalMux(sessionId, terminal.cols, terminal.rows); - }, [muxStatus, sessionId, resizeTerminalMux]); + resizeTerminalMux(sessionId, terminal.cols, terminal.rows, projectId); + }, [muxStatus, sessionId, projectId, resizeTerminalMux]); // Live theme switching without terminal recreation useEffect(() => { @@ -364,8 +380,8 @@ export function useXtermTerminal( // localStorage might be unavailable } fit.fit(); - resizeTerminalMux(sessionId, terminal.cols, terminal.rows); - }, [fontSize, sessionId, resizeTerminalMux]); + resizeTerminalMux(sessionId, terminal.cols, terminal.rows, projectId); + }, [fontSize, sessionId, projectId, resizeTerminalMux]); const scrollToLatest = () => { const t = terminalInstance.current; @@ -377,7 +393,7 @@ export function useXtermTerminal( // Alternate buffer (tmux/vim): xterm has no scrollback to scroll // to. The user is in tmux copy-mode (entered by attachTouchScroll // on swipe). Send 'q' to exit copy-mode and return to live tail. - writeTerminal(sessionId, "q"); + writeTerminal(sessionId, "q", projectId); } } followOutputRef.current = true; diff --git a/packages/web/src/lib/mux-protocol.ts b/packages/web/src/lib/mux-protocol.ts index c406c6af2..64e81b3e2 100644 --- a/packages/web/src/lib/mux-protocol.ts +++ b/packages/web/src/lib/mux-protocol.ts @@ -3,20 +3,20 @@ import type { AttentionLevel } from "./types"; // ── Client → Server ── export type ClientMessage = - | { ch: "terminal"; id: string; type: "data"; data: string } - | { ch: "terminal"; id: string; type: "resize"; cols: number; rows: number } - | { ch: "terminal"; id: string; type: "open"; tmuxName?: string } - | { ch: "terminal"; id: string; type: "close" } + | { ch: "terminal"; id: string; type: "data"; data: string; projectId?: string } + | { ch: "terminal"; id: string; type: "resize"; cols: number; rows: number; projectId?: string } + | { ch: "terminal"; id: string; type: "open"; projectId?: string; tmuxName?: string } + | { ch: "terminal"; id: string; type: "close"; projectId?: string } | { ch: "system"; type: "ping" } - | { ch: "subscribe"; topics: ("sessions")[] }; + | { ch: "subscribe"; topics: "sessions"[] }; // ── Server → Client ── export type ServerMessage = - | { ch: "terminal"; id: string; type: "data"; data: string } - | { ch: "terminal"; id: string; type: "exited"; code: number } - | { ch: "terminal"; id: string; type: "opened" } - | { ch: "terminal"; id: string; type: "error"; message: string } + | { ch: "terminal"; id: string; type: "data"; data: string; projectId?: string } + | { ch: "terminal"; id: string; type: "exited"; code: number; projectId?: string } + | { ch: "terminal"; id: string; type: "opened"; projectId?: string } + | { ch: "terminal"; id: string; type: "error"; message: string; projectId?: string } | { ch: "sessions"; type: "snapshot"; sessions: SessionPatch[] } | { ch: "sessions"; type: "error"; error: string } | { ch: "system"; type: "pong" } diff --git a/packages/web/src/providers/MuxProvider.tsx b/packages/web/src/providers/MuxProvider.tsx index 43c77f867..8449a36c3 100644 --- a/packages/web/src/providers/MuxProvider.tsx +++ b/packages/web/src/providers/MuxProvider.tsx @@ -4,11 +4,15 @@ import React, { useEffect, useRef, useState, useMemo, useCallback, type ReactNod import type { ClientMessage, ServerMessage, SessionPatch } from "@/lib/mux-protocol"; interface MuxContextValue { - subscribeTerminal: (id: string, callback: (data: string) => void) => () => void; - writeTerminal: (id: string, data: string) => void; - openTerminal: (id: string, tmuxName?: string) => void; - closeTerminal: (id: string) => void; - resizeTerminal: (id: string, cols: number, rows: number) => void; + subscribeTerminal: ( + id: string, + callback: (data: string) => void, + projectId?: string, + ) => () => void; + writeTerminal: (id: string, data: string, projectId?: string) => void; + openTerminal: (id: string, projectId?: string, tmuxName?: string) => void; + closeTerminal: (id: string, projectId?: string) => void; + resizeTerminal: (id: string, cols: number, rows: number, projectId?: string) => void; status: "connecting" | "connected" | "reconnecting" | "disconnected"; sessions: SessionPatch[]; /** Last session-fetch error from the server, null when healthy. */ @@ -49,7 +53,10 @@ function normalizePathValue(value: unknown): string | undefined { return trimmed; } -function buildMuxWsUrl(runtimeConfig: { directTerminalPort?: string; proxyWsPath?: string }): string { +function buildMuxWsUrl(runtimeConfig: { + directTerminalPort?: string; + proxyWsPath?: string; +}): string { const loc = window.location; const protocol = loc.protocol === "https:" ? "wss:" : "ws:"; @@ -66,17 +73,24 @@ function buildMuxWsUrl(runtimeConfig: { directTerminalPort?: string; proxyWsPath } // Direct port connection — prefer runtime-configured port, fall back to env/default - const port = runtimeConfig.directTerminalPort ?? process.env.NEXT_PUBLIC_DIRECT_TERMINAL_PORT ?? "14801"; + const port = + runtimeConfig.directTerminalPort ?? process.env.NEXT_PUBLIC_DIRECT_TERMINAL_PORT ?? "14801"; return `${protocol}//${loc.hostname}:${port}/mux`; } +function terminalKey(id: string, projectId?: string): string { + return projectId ? `${projectId}:${id}` : id; +} + export function MuxProvider({ children }: { children: ReactNode }) { const wsRef = useRef(null); const subscribersRef = useRef(new Map void>>()); - const openedTerminalsRef = useRef(new Map()); - const [status, setStatus] = useState<"connecting" | "connected" | "reconnecting" | "disconnected">( - "connecting", + const openedTerminalsRef = useRef( + new Map(), ); + const [status, setStatus] = useState< + "connecting" | "connected" | "reconnecting" | "disconnected" + >("connecting"); const [sessions, setSessions] = useState([]); const [lastError, setLastError] = useState(null); const reconnectAttempt = useRef(0); @@ -108,12 +122,13 @@ export function MuxProvider({ children }: { children: ReactNode }) { reconnectAttempt.current = 0; // Re-open previously opened terminals - for (const [terminalId, tmuxName] of openedTerminalsRef.current) { + for (const terminal of openedTerminalsRef.current.values()) { const openMsg: ClientMessage = { ch: "terminal", - id: terminalId, + id: terminal.id, type: "open", - ...(tmuxName && { tmuxName }), + ...(terminal.projectId && { projectId: terminal.projectId }), + ...(terminal.tmuxName && { tmuxName: terminal.tmuxName }), }; ws.send(JSON.stringify(openMsg)); } @@ -131,9 +146,10 @@ export function MuxProvider({ children }: { children: ReactNode }) { const msg = JSON.parse(event.data as string) as ServerMessage; if (msg.ch === "terminal") { + const key = terminalKey(msg.id, "projectId" in msg ? msg.projectId : undefined); if (msg.type === "data") { // Push to subscribers - const subs = subscribersRef.current.get(msg.id); + const subs = subscribersRef.current.get(key); if (subs) { for (const callback of subs) { callback(msg.data); @@ -141,14 +157,17 @@ export function MuxProvider({ children }: { children: ReactNode }) { } } else if (msg.type === "opened") { // Terminal opened successfully — preserve existing tmuxName - if (!openedTerminalsRef.current.has(msg.id)) { - openedTerminalsRef.current.set(msg.id, undefined); + if (!openedTerminalsRef.current.has(key)) { + openedTerminalsRef.current.set(key, { + id: msg.id, + ...("projectId" in msg && msg.projectId ? { projectId: msg.projectId } : {}), + }); } } else if (msg.type === "exited") { // PTY exited and could not be re-attached — remove so it isn't // re-opened on reconnect, and surface a terminal-level error chunk - openedTerminalsRef.current.delete(msg.id); - const subs = subscribersRef.current.get(msg.id); + openedTerminalsRef.current.delete(key); + const subs = subscribersRef.current.get(key); if (subs) { const notice = `\r\n\x1b[31m[Terminal exited with code ${msg.code}]\x1b[0m\r\n`; for (const callback of subs) { @@ -235,32 +254,34 @@ export function MuxProvider({ children }: { children: ReactNode }) { }, [connect]); const subscribeTerminal = useCallback( - (id: string, callback: (data: string) => void): (() => void) => { + (id: string, callback: (data: string) => void, projectId?: string): (() => void) => { + const key = terminalKey(id, projectId); // Add to subscribers - let subs = subscribersRef.current.get(id); + let subs = subscribersRef.current.get(key); if (!subs) { subs = new Set(); - subscribersRef.current.set(id, subs); + subscribersRef.current.set(key, subs); } subs.add(callback); // Request open if not already open - if (!openedTerminalsRef.current.has(id) && wsRef.current?.readyState === WebSocket.OPEN) { + if (!openedTerminalsRef.current.has(key) && wsRef.current?.readyState === WebSocket.OPEN) { const openMsg: ClientMessage = { ch: "terminal", id, type: "open", + ...(projectId && { projectId }), }; wsRef.current.send(JSON.stringify(openMsg)); } // Return unsubscribe function return () => { - const subs = subscribersRef.current.get(id); + const subs = subscribersRef.current.get(key); if (subs) { subs.delete(callback); if (subs.size === 0) { - subscribersRef.current.delete(id); + subscribersRef.current.delete(key); } } }; @@ -268,55 +289,62 @@ export function MuxProvider({ children }: { children: ReactNode }) { [], ); - const writeTerminal = useCallback((id: string, data: string) => { + const writeTerminal = useCallback((id: string, data: string, projectId?: string) => { if (wsRef.current?.readyState === WebSocket.OPEN) { const msg: ClientMessage = { ch: "terminal", id, type: "data", data, + ...(projectId && { projectId }), }; wsRef.current.send(JSON.stringify(msg)); } }, []); - const openTerminal = useCallback((id: string, tmuxName?: string) => { - openedTerminalsRef.current.set(id, tmuxName); + const openTerminal = useCallback((id: string, projectId?: string, tmuxName?: string) => { + openedTerminalsRef.current.set(terminalKey(id, projectId), { id, projectId, tmuxName }); if (wsRef.current?.readyState === WebSocket.OPEN) { const msg: ClientMessage = { ch: "terminal", id, type: "open", + ...(projectId && { projectId }), ...(tmuxName && { tmuxName }), }; wsRef.current.send(JSON.stringify(msg)); } }, []); - const closeTerminal = useCallback((id: string) => { - openedTerminalsRef.current.delete(id); + const closeTerminal = useCallback((id: string, projectId?: string) => { + openedTerminalsRef.current.delete(terminalKey(id, projectId)); if (wsRef.current?.readyState === WebSocket.OPEN) { const msg: ClientMessage = { ch: "terminal", id, type: "close", + ...(projectId && { projectId }), }; wsRef.current.send(JSON.stringify(msg)); } }, []); - const resizeTerminal = useCallback((id: string, cols: number, rows: number) => { - if (wsRef.current?.readyState === WebSocket.OPEN) { - const msg: ClientMessage = { - ch: "terminal", - id, - type: "resize", - cols, - rows, - }; - wsRef.current.send(JSON.stringify(msg)); - } - }, []); + const resizeTerminal = useCallback( + (id: string, cols: number, rows: number, projectId?: string) => { + if (wsRef.current?.readyState === WebSocket.OPEN) { + const msg: ClientMessage = { + ch: "terminal", + id, + type: "resize", + cols, + rows, + ...(projectId && { projectId }), + }; + wsRef.current.send(JSON.stringify(msg)); + } + }, + [], + ); const contextValue: MuxContextValue = useMemo( () => ({ diff --git a/packages/web/src/providers/__tests__/MuxProvider.test.tsx b/packages/web/src/providers/__tests__/MuxProvider.test.tsx index 9086f29d4..737418f91 100644 --- a/packages/web/src/providers/__tests__/MuxProvider.test.tsx +++ b/packages/web/src/providers/__tests__/MuxProvider.test.tsx @@ -244,7 +244,12 @@ describe("MuxProvider connection lifecycle", () => { }); it("falls back gracefully when fetch fails", async () => { - vi.stubGlobal("fetch", vi.fn(async () => { throw new Error("network"); })); + vi.stubGlobal( + "fetch", + vi.fn(async () => { + throw new Error("network"); + }), + ); renderHook(() => useMux(), { wrapper }); await flushInit(); // Still creates a WebSocket (using defaults) @@ -252,8 +257,16 @@ describe("MuxProvider connection lifecycle", () => { }); it("sets disconnected status when WebSocket constructor throws", async () => { - vi.stubGlobal("WebSocket", vi.fn(() => { throw new Error("unavailable"); })); - vi.stubGlobal("fetch", vi.fn(async () => ({ ok: false }))); + vi.stubGlobal( + "WebSocket", + vi.fn(() => { + throw new Error("unavailable"); + }), + ); + vi.stubGlobal( + "fetch", + vi.fn(async () => ({ ok: false })), + ); const { result } = renderHook(() => useMux(), { wrapper }); await flushInit(); expect(result.current.status).toBe("disconnected"); @@ -271,10 +284,12 @@ describe("MuxProvider connection lifecycle", () => { act(() => result.current.openTerminal("session-abc")); // Confirm open message sent on ws1 - expect(ws1.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.type === "open" && p.id === "session-abc"; - })).toBe(true); + expect( + ws1.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.type === "open" && p.id === "session-abc"; + }), + ).toBe(true); // Disconnect + reconnect act(() => ws1.simulateClose()); @@ -286,10 +301,12 @@ describe("MuxProvider connection lifecycle", () => { act(() => ws2.simulateOpen()); // session-abc should be re-opened on ws2 - expect(ws2.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.type === "open" && p.id === "session-abc"; - })).toBe(true); + expect( + ws2.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.type === "open" && p.id === "session-abc"; + }), + ).toBe(true); } finally { vi.useRealTimers(); } @@ -329,7 +346,9 @@ describe("MuxProvider message handling", () => { const { result, ws } = await setupConnected(); const received: string[] = []; - act(() => { result.current.subscribeTerminal("s1", (d) => received.push(d)); }); + act(() => { + result.current.subscribeTerminal("s1", (d) => received.push(d)); + }); act(() => ws.simulateMessage({ ch: "terminal", id: "s1", type: "data", data: "hello" })); expect(received).toContain("hello"); @@ -354,10 +373,12 @@ describe("MuxProvider message handling", () => { const ws2 = MockWebSocket.instances[MockWebSocket.instances.length - 1]; act(() => ws2.simulateOpen()); - expect(ws2.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.id === "s1"; - })).toBe(true); + expect( + ws2.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.id === "s1"; + }), + ).toBe(true); } finally { vi.useRealTimers(); } @@ -367,7 +388,9 @@ describe("MuxProvider message handling", () => { const { result, ws } = await setupConnected(); const received: string[] = []; - act(() => { result.current.subscribeTerminal("s1", (d) => received.push(d)); }); + act(() => { + result.current.subscribeTerminal("s1", (d) => received.push(d)); + }); act(() => ws.simulateMessage({ ch: "terminal", id: "s1", type: "exited", code: 1 })); expect(received.some((m) => m.includes("exited"))).toBe(true); @@ -407,7 +430,9 @@ describe("MuxProvider message handling", () => { const spy = vi.spyOn(console, "error").mockImplementation(() => {}); const { result, ws } = await setupConnected(); - act(() => ws.simulateMessage({ ch: "terminal", id: "s1", type: "error", message: "PTY failed" })); + act(() => + ws.simulateMessage({ ch: "terminal", id: "s1", type: "error", message: "PTY failed" }), + ); expect(spy).toHaveBeenCalledWith(expect.stringContaining("Terminal error"), expect.any(String)); expect(result.current.status).toBe("connected"); spy.mockRestore(); @@ -416,11 +441,21 @@ describe("MuxProvider message handling", () => { it("updates sessions on snapshot message", async () => { const { result, ws } = await setupConnected(); - act(() => ws.simulateMessage({ - ch: "sessions", - type: "snapshot", - sessions: [{ id: "s1", status: "working", activity: null, attentionLevel: "working", lastActivityAt: "" }], - })); + act(() => + ws.simulateMessage({ + ch: "sessions", + type: "snapshot", + sessions: [ + { + id: "s1", + status: "working", + activity: null, + attentionLevel: "working", + lastActivityAt: "", + }, + ], + }), + ); expect(result.current.sessions.length).toBe(1); expect(result.current.sessions[0].id).toBe("s1"); @@ -458,10 +493,12 @@ describe("MuxProvider terminal operations", () => { it("writeTerminal sends data message", async () => { const { result, ws } = await setupConnected(); act(() => result.current.writeTerminal("s1", "hello\n")); - expect(ws.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.type === "data" && p.data === "hello\n"; - })).toBe(true); + expect( + ws.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.type === "data" && p.data === "hello\n"; + }), + ).toBe(true); }); it("writeTerminal is no-op when WebSocket is not open", async () => { @@ -475,38 +512,48 @@ describe("MuxProvider terminal operations", () => { it("openTerminal sends open message when connected", async () => { const { result, ws } = await setupConnected(); act(() => result.current.openTerminal("session-abc")); - expect(ws.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.type === "open" && p.id === "session-abc"; - })).toBe(true); + expect( + ws.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.type === "open" && p.id === "session-abc"; + }), + ).toBe(true); }); it("closeTerminal sends close message", async () => { const { result, ws } = await setupConnected(); act(() => result.current.openTerminal("session-abc")); act(() => result.current.closeTerminal("session-abc")); - expect(ws.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.type === "close" && p.id === "session-abc"; - })).toBe(true); + expect( + ws.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.type === "close" && p.id === "session-abc"; + }), + ).toBe(true); }); it("resizeTerminal sends resize message with cols and rows", async () => { const { result, ws } = await setupConnected(); act(() => result.current.resizeTerminal("session-abc", 120, 40)); - expect(ws.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.type === "resize" && p.cols === 120 && p.rows === 40; - })).toBe(true); + expect( + ws.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.type === "resize" && p.cols === 120 && p.rows === 40; + }), + ).toBe(true); }); it("subscribeTerminal sends open for untracked terminal", async () => { const { result, ws } = await setupConnected(); - act(() => { result.current.subscribeTerminal("session-new", () => {}); }); - expect(ws.sentMessages.some((m) => { - const p = JSON.parse(m) as Record; - return p.ch === "terminal" && p.type === "open" && p.id === "session-new"; - })).toBe(true); + act(() => { + result.current.subscribeTerminal("session-new", () => {}); + }); + expect( + ws.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return p.ch === "terminal" && p.type === "open" && p.id === "session-new"; + }), + ).toBe(true); }); it("subscribeTerminal unsubscribe stops data delivery", async () => { @@ -514,7 +561,9 @@ describe("MuxProvider terminal operations", () => { const received: string[] = []; let unsub!: () => void; - act(() => { unsub = result.current.subscribeTerminal("s1", (d) => received.push(d)); }); + act(() => { + unsub = result.current.subscribeTerminal("s1", (d) => received.push(d)); + }); act(() => ws.simulateMessage({ ch: "terminal", id: "s1", type: "data", data: "before" })); act(() => unsub()); act(() => ws.simulateMessage({ ch: "terminal", id: "s1", type: "data", data: "after" })); @@ -522,6 +571,55 @@ describe("MuxProvider terminal operations", () => { expect(received).toContain("before"); expect(received).not.toContain("after"); }); + + it("keeps terminal subscribers scoped by project id", async () => { + const { result, ws } = await setupConnected(); + const alpha: string[] = []; + const beta: string[] = []; + + act(() => { + result.current.subscribeTerminal("app-1", (data) => alpha.push(data), "alpha"); + result.current.subscribeTerminal("app-1", (data) => beta.push(data), "beta"); + }); + + act(() => + ws.simulateMessage({ + ch: "terminal", + id: "app-1", + projectId: "alpha", + type: "data", + data: "a", + }), + ); + act(() => + ws.simulateMessage({ + ch: "terminal", + id: "app-1", + projectId: "beta", + type: "data", + data: "b", + }), + ); + + expect(alpha).toEqual(["a"]); + expect(beta).toEqual(["b"]); + expect( + ws.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return ( + p.ch === "terminal" && p.type === "open" && p.id === "app-1" && p.projectId === "alpha" + ); + }), + ).toBe(true); + expect( + ws.sentMessages.some((m) => { + const p = JSON.parse(m) as Record; + return ( + p.ch === "terminal" && p.type === "open" && p.id === "app-1" && p.projectId === "beta" + ); + }), + ).toBe(true); + }); }); // --------------------------------------------------------------------------- @@ -548,7 +646,10 @@ describe("buildMuxWsUrl", () => { }); it("uses path-based URL when port is empty (reverse proxy)", async () => { - vi.stubGlobal("fetch", vi.fn(async () => ({ ok: false }))); + vi.stubGlobal( + "fetch", + vi.fn(async () => ({ ok: false })), + ); Object.defineProperty(window, "location", { writable: true, value: { protocol: "http:", host: "localhost", hostname: "localhost", port: "" },