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): preserve PTYs across reconnect churn (#1710)

This commit is contained in:
nikhil achale 2026-05-08 15:41:44 +05:30
parent b9b20e19d1
commit 01843d7744
3 changed files with 175 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
packages/web/server/__tests__/direct-terminal-ws.integration.test.ts
View File

@ -196,7 +196,12 @@ describe("mux terminal open", () => {
const ws = await connectMux();
ws.send(
JSON.stringify({ ch: "terminal", id: "test-session", tmuxName: TEST_SESSION, type: "open" }),
JSON.stringify({
ch: "terminal",
id: "test-session",
tmuxName: TEST_SESSION,
type: "open",
}),
);
const msg = await waitForMessage(ws, (m) => m.ch === "terminal" && m.type === "opened");
@ -296,9 +301,10 @@ describe("mux terminal open", () => {
// 2. Wait > REATTACH_RESET_GRACE_MS so the grace timer fires and
// resets reattachAttempts to 0 on the still-attached PTY 1.
// 3. Kill the tmux session — PTY 1 exits, the server burns the full
// MAX_REATTACH_ATTEMPTS=3 budget trying to re-attach, then emits
// "exited". A 2 s window is comfortably more than 3 × ~50 ms.
// 4. Per-test timeout raised to 15 s to accommodate the 5+ s wait.
// MAX_REATTACH_ATTEMPTS=3 budget trying to re-attach with the
// production backoff between attempts, then emits "exited".
// 4. Per-test timeout raised to 20 s to accommodate the 5+ s wait
// and the retry backoff windows.
//
// If the grace timer or its closure guard ever regresses (e.g. is
// unref'd in a way that prevents firing, or the wrong reference is
@ -322,7 +328,7 @@ describe("mux terminal open", () => {
const exitedMsg = await waitForMessage(
ws,
(m) => m.ch === "terminal" && m.type === "exited",
2000,
8000,
);
expect(exitedMsg.id).toBe(RECOVERY_TEST_SESSION);
@ -334,7 +340,7 @@ describe("mux terminal open", () => {
/* already gone */
}
}
}, 15_000);
}, 20_000);
it("bounds re-attach attempts when tmux session dies mid-subscription (issue #1639)", async () => {
// Reproduces the runaway re-attach loop that exhausts the system PTY
@ -364,11 +370,11 @@ describe("mux terminal open", () => {
// to attach to and will exit ~40 ms after each re-attach attempt.
execFileSync(TMUX, ["kill-session", "-t", RUNAWAY_TEST_SESSION], { timeout: 5000 });
// 3 re-attach attempts × ~50 ms each = ~150 ms; allow generous margin.
// 3 re-attach attempts with production backoff; allow generous margin.
const exitedMsg = await waitForMessage(
ws,
(m) => m.ch === "terminal" && m.type === "exited",
2000,
8000,
);
expect(exitedMsg.id).toBe(RUNAWAY_TEST_SESSION);
@ -381,7 +387,7 @@ describe("mux terminal open", () => {
/* already gone */
}
}
});
}, 12_000);
});
// =============================================================================
@ -393,7 +399,12 @@ describe("mux terminal I/O", () => {
const ws = await connectMux();
ws.send(
JSON.stringify({ ch: "terminal", id: "test-session", tmuxName: TEST_SESSION, type: "open" }),
JSON.stringify({
ch: "terminal",
id: `test-session-data-${process.pid}`,
tmuxName: TEST_SESSION,
type: "open",
}),
);
await waitForMessage(ws, (m) => m.ch === "terminal" && m.type === "opened");

89
packages/web/server/__tests__/mux-websocket.test.ts
View File

@ -311,3 +311,92 @@ describe("TerminalManager.open — tmux target args (regression for #1714)", ()
expect(args).toEqual(["attach-session", "-t", "=ao-177"]);
});
});
describe("TerminalManager.subscribe — PTY lifecycle", () => {
type ExitHandler = (event: { exitCode: number }) => void;
let ptys: Array<{ exit: (exitCode?: number) => void; kill: ReturnType<typeof vi.fn> }>;
beforeEach(() => {
vi.useFakeTimers();
mockSpawn.mockReset();
mockPtySpawn.mockReset();
ptys = [];
// spawn() returns an object that emits "error" — we just need .on() to work.
mockSpawn.mockImplementation(() => new EventEmitter());
mockPtySpawn.mockImplementation(() => {
let exitHandler: ExitHandler | undefined;
const pty = {
onData: vi.fn(),
onExit: vi.fn((handler: ExitHandler) => {
exitHandler = handler;
}),
write: vi.fn(),
resize: vi.fn(),
kill: vi.fn(),
};
ptys.push({
exit: (exitCode = 0) => exitHandler?.({ exitCode }),
kill: pty.kill,
});
return pty;
});
});
afterEach(() => {
vi.clearAllTimers();
vi.useRealTimers();
});
it("keeps the PTY alive during the last-subscriber close grace period", async () => {
const mgr = new TerminalManager("/usr/bin/tmux");
const unsubscribe = mgr.subscribe("ao-177", undefined, vi.fn());
unsubscribe();
await vi.advanceTimersByTimeAsync(29_999);
expect(ptys[0]?.kill).not.toHaveBeenCalled();
expect(mockPtySpawn).toHaveBeenCalledTimes(1);
mgr.subscribe("ao-177", undefined, vi.fn());
await vi.advanceTimersByTimeAsync(1);
expect(ptys[0]?.kill).not.toHaveBeenCalled();
expect(mockPtySpawn).toHaveBeenCalledTimes(1);
});
it("kills the PTY after the close grace period when no subscriber reconnects", async () => {
const mgr = new TerminalManager("/usr/bin/tmux");
const unsubscribe = mgr.subscribe("ao-177", undefined, vi.fn());
unsubscribe();
await vi.advanceTimersByTimeAsync(30_000);
expect(ptys[0]?.kill).toHaveBeenCalledTimes(1);
});
it("backs off before re-attaching after a PTY exit with active subscribers", async () => {
const mgr = new TerminalManager("/usr/bin/tmux");
mgr.subscribe("ao-177", undefined, vi.fn(), vi.fn());
ptys[0]?.exit(1);
expect(mockPtySpawn).toHaveBeenCalledTimes(1);
mgr.subscribe("ao-177", undefined, vi.fn(), vi.fn());
expect(mockPtySpawn).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1_999);
expect(mockPtySpawn).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
expect(mockPtySpawn).toHaveBeenCalledTimes(2);
});
});

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

@ -200,11 +200,25 @@ interface ManagedTerminal {
* reachable for up to 5 s after teardown.
*/
resetTimer?: ReturnType<typeof setTimeout>;
/**
* Pending last-subscriber grace timer. When all WebSocket subscribers
* disconnect, keep the PTY alive briefly so reload/sleep/wake reconnects
* can re-use the same macOS PTY slot.
*/
closeTimer?: ReturnType<typeof setTimeout>;
/**
* Pending delayed re-attach after a PTY exits while subscribers are present.
* Tracked so teardown can cancel the retry instead of spawning a fresh PTY
* after the terminal has no listeners.
*/
reattachTimer?: ReturnType<typeof setTimeout>;
}
const RING_BUFFER_MAX = 50 * 1024; // 50KB max per terminal
const WS_BUFFER_HIGH_WATERMARK = 64 * 1024; // 64KB
const MAX_REATTACH_ATTEMPTS = 3;
const TERMINAL_CLOSE_GRACE_MS = 30_000;
const REATTACH_RETRY_DELAY_MS = 2_000;
/**
* Grace period a freshly-attached PTY must survive before its successful
* attach is allowed to reset the re-attach counter. Prevents tight crash
@ -273,6 +287,9 @@ export class TerminalManager {
if (terminal.pty) {
return tmuxSessionId;
}
if (terminal.reattachTimer) {
return tmuxSessionId;
}
// tmux 3.4 only honours the `=` exact-match prefix on has-session and
// attach-session; set-option silently ignores it, so we use the bare id
@ -381,12 +398,26 @@ export class TerminalManager {
console.log(
`[MuxServer] Re-attaching to ${id} (attempt ${terminal.reattachAttempts}/${MAX_REATTACH_ATTEMPTS})`,
);
try {
this.open(id, projectId, tmuxSessionId);
return; // re-attached — don't notify exit
} catch (err) {
console.error(`[MuxServer] Failed to re-attach ${id}:`, err);
if (terminal.reattachTimer) {
clearTimeout(terminal.reattachTimer);
}
terminal.reattachTimer = setTimeout(() => {
terminal.reattachTimer = undefined;
if (terminal.subscribers.size === 0 || terminal.pty) {
return;
}
try {
this.open(id, projectId, tmuxSessionId);
} catch (err) {
console.error(`[MuxServer] Failed to re-attach ${id}:`, err);
for (const cb of terminal.exitCallbacks) {
cb(exitCode);
}
}
}, REATTACH_RETRY_DELAY_MS);
terminal.reattachTimer.unref();
return; // scheduled re-attach — don't notify exit yet
} else if (terminal.reattachAttempts >= MAX_REATTACH_ATTEMPTS) {
console.error(`[MuxServer] Max re-attach attempts reached for ${id}, giving up`);
}
@ -443,22 +474,42 @@ export class TerminalManager {
// Add subscriber
terminal.subscribers.add(callback);
if (onExit) terminal.exitCallbacks.add(onExit);
if (terminal.closeTimer) {
clearTimeout(terminal.closeTimer);
terminal.closeTimer = undefined;
}
// Return unsubscribe function
return () => {
terminal.subscribers.delete(callback);
if (onExit) terminal.exitCallbacks.delete(onExit);
// Kill PTY and clean up when the last subscriber leaves
// Kill PTY and clean up after a short grace period when the last subscriber leaves.
if (terminal.subscribers.size === 0) {
if (terminal.resetTimer) {
clearTimeout(terminal.resetTimer);
terminal.resetTimer = undefined;
if (terminal.closeTimer) {
return;
}
if (terminal.pty) {
terminal.pty.kill();
terminal.pty = null;
}
this.terminals.delete(key);
terminal.closeTimer = setTimeout(() => {
terminal.closeTimer = undefined;
if (terminal.subscribers.size > 0) {
return;
}
if (terminal.resetTimer) {
clearTimeout(terminal.resetTimer);
terminal.resetTimer = undefined;
}
if (terminal.reattachTimer) {
clearTimeout(terminal.reattachTimer);
terminal.reattachTimer = undefined;
}
if (terminal.pty) {
terminal.pty.kill();
terminal.pty = null;
}
this.terminals.delete(key);
}, TERMINAL_CLOSE_GRACE_MS);
terminal.closeTimer.unref();
}
};
}