bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Chirag Arora 2026-05-01 16:15:43 +05:30 committed by GitHub
parent 0a9ba4cd7f
commit b2cdf7adab
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 452 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
packages/web/server/__tests__/tmux-utils.test.ts
View File

@ -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

116
packages/web/server/mux-websocket.ts
View File

@ -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));
}

19
packages/web/server/tmux-utils.ts
View File

@ -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 });

5
packages/web/src/components/DirectTerminal.tsx
View File

@ -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(() => {

1
packages/web/src/components/SessionDetail.tsx
View File

@ -208,6 +208,7 @@ export function SessionDetail({
) : (
<DirectTerminal
sessionId={session.id}
projectId={session.projectId}
tmuxName={session.metadata?.tmuxName}
startFullscreen={startFullscreen}
variant={terminalVariant}

21
packages/web/src/components/__tests__/DirectTerminal.render.test.tsx
View File

@ -4,6 +4,9 @@ import { DirectTerminal } from "../DirectTerminal";
const replaceMock = vi.fn();
let searchParams = new URLSearchParams();
const { useFullscreenResizeMock } = vi.hoisted(() => ({
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<string, unknown>;
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(<DirectTerminal sessionId="app-1" projectId="project-a" />);
expect(useFullscreenResizeMock).toHaveBeenCalledWith(
false,
"app-1",
"project-a",
expect.any(Object),
expect.any(Object),
expect.any(Object),
);
});
});

5
packages/web/src/components/terminal/useFullscreenResize.ts
View File

@ -16,6 +16,7 @@ import { useMux } from "@/hooks/useMux";
export function useFullscreenResize(
fullscreen: boolean,
sessionId: string,
projectId: string | undefined,
terminalInstance: RefObject<TerminalType | null>,
fitAddon: RefObject<FitAddonType | null>,
containerRef: RefObject<HTMLDivElement | null>,
@ -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]);
}

90
packages/web/src/components/terminal/useXtermTerminal.ts
View File

@ -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;

18
packages/web/src/lib/mux-protocol.ts
View File

@ -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" }

110
packages/web/src/providers/MuxProvider.tsx
View File

@ -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<WebSocket | null>(null);
const subscribersRef = useRef(new Map<string, Set<(data: string) => void>>());
const openedTerminalsRef = useRef(new Map<string, string | undefined>());
const [status, setStatus] = useState<"connecting" | "connected" | "reconnecting" | "disconnected">(
"connecting",
const openedTerminalsRef = useRef(
new Map<string, { id: string; projectId?: string; tmuxName?: string }>(),
);
const [status, setStatus] = useState<
"connecting" | "connected" | "reconnecting" | "disconnected"
>("connecting");
const [sessions, setSessions] = useState<SessionPatch[]>([]);
const [lastError, setLastError] = useState<string | null>(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(
() => ({

193
packages/web/src/providers/__tests__/MuxProvider.test.tsx
View File

@ -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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
return p.ch === "terminal" && p.id === "s1";
})).toBe(true);
expect(
ws2.sentMessages.some((m) => {
const p = JSON.parse(m) as Record<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
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: "" },