Skip to content

🤖 fix: buffer terminal output until client subscribes #818

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ibetitsmike wants to merge 1 commit into from
Closed

🤖 fix: buffer terminal output until client subscribes #818

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion src/browser/api.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -343,7 +343,11 @@ const webApi: IPCApi = {
onOutput: (sessionId: string, callback: (data: string) => void) => {
// Subscribe to terminal output events via WebSocket
const channel = `terminal:output:${sessionId}`;
return wsManager.on(channel, callback as (data: unknown) => void);
const unsubscribe = wsManager.on(channel, callback as (data: unknown) => void);
// Tell server we're ready to receive output - this flushes any buffered output
// that arrived before we registered our handler (fixes first prompt issue)
void invokeIPC(IPC_CHANNELS.TERMINAL_SUBSCRIBE, sessionId);
return unsubscribe;
},
onExit: (sessionId: string, callback: (exitCode: number) => void) => {
// Subscribe to terminal exit events via WebSocket
Expand Down
1 change: 1 addition & 0 deletions src/common/constants/ipc-constants.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ export const IPC_CHANNELS = {
TERMINAL_CLOSE: "terminal:close",
TERMINAL_RESIZE: "terminal:resize",
TERMINAL_INPUT: "terminal:input",
TERMINAL_SUBSCRIBE: "terminal:subscribe",
TERMINAL_WINDOW_OPEN: "terminal:window:open",
TERMINAL_WINDOW_CLOSE: "terminal:window:close",

Expand Down
2 changes: 2 additions & 0 deletions src/desktop/preload.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,8 @@ const api: IPCApi = {
const channel = `terminal:output:${sessionId}`;
const handler = (_event: unknown, data: string) => callback(data);
ipcRenderer.on(channel, handler);
// Tell server we're ready to receive output - flushes any buffered output
void ipcRenderer.invoke(IPC_CHANNELS.TERMINAL_SUBSCRIBE, sessionId);
return () => ipcRenderer.removeListener(channel, handler);
},
onExit: (sessionId: string, callback: (exitCode: number) => void) => {
Expand Down
12 changes: 12 additions & 0 deletions src/node/services/ipcMain.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1905,6 +1905,18 @@ export class IpcMain {
}
});

// Subscribe to terminal output (flushes buffered output)
// In browser mode, output is buffered until the client subscribes to avoid
// losing initial output (like the shell prompt) during the HTTP round-trip.
ipcMain.handle(IPC_CHANNELS.TERMINAL_SUBSCRIBE, (_event, sessionId: string) => {
try {
this.ptyService.subscribeOutput(sessionId);
} catch (err) {
log.error("Error subscribing to terminal:", err);
throw err;
}
});

ipcMain.handle(IPC_CHANNELS.TERMINAL_WINDOW_OPEN, async (_event, workspaceId: string) => {
console.log(`[BACKEND] TERMINAL_WINDOW_OPEN handler called with: ${workspaceId}`);
try {
Expand Down
137 changes: 93 additions & 44 deletions src/node/services/ptyService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,10 @@ interface SessionData {
runtime: Runtime;
onData: (data: string) => void;
onExit: (exitCode: number) => void;
/** Buffer for output received before client subscribes */
outputBuffer: string[];
/** Whether the client has subscribed to receive output */
clientSubscribed: boolean;
}

/**
Expand Down Expand Up @@ -161,37 +165,55 @@ export class PTYService {
);
}

// Create session data with buffer for pre-subscription output
const sessionData: SessionData = {
pty: ptyProcess,
workspaceId: params.workspaceId,
workspacePath,
runtime,
onData,
onExit,
outputBuffer: [],
clientSubscribed: false,
};
this.sessions.set(sessionId, sessionData);

// Forward PTY data via callback
// Buffer to handle escape sequences split across chunks
let buffer = "";
let escapeBuffer = "";

ptyProcess.onData((data) => {
// Append new data to buffer
buffer += data;
// Append new data to escape sequence buffer
escapeBuffer += data;

// Check if buffer ends with an incomplete escape sequence
// Look for ESC at the end without its complete sequence
let sendUpTo = buffer.length;
let sendUpTo = escapeBuffer.length;

// If buffer ends with ESC or ESC[, hold it back for next chunk
if (buffer.endsWith("\x1b")) {
sendUpTo = buffer.length - 1;
} else if (buffer.endsWith("\x1b[")) {
sendUpTo = buffer.length - 2;
if (escapeBuffer.endsWith("\x1b")) {
sendUpTo = escapeBuffer.length - 1;
} else if (escapeBuffer.endsWith("\x1b[")) {
sendUpTo = escapeBuffer.length - 2;
} else {
// Check if it ends with ESC[ followed by incomplete CSI sequence
// eslint-disable-next-line no-control-regex, @typescript-eslint/prefer-regexp-exec
const match = buffer.match(/\x1b\[[0-9;]*$/);
const match = escapeBuffer.match(/\x1b\[[0-9;]*$/);
if (match) {
sendUpTo = buffer.length - match[0].length;
sendUpTo = escapeBuffer.length - match[0].length;
}
}

// Send complete data
if (sendUpTo > 0) {
const toSend = buffer.substring(0, sendUpTo);
onData(toSend);
buffer = buffer.substring(sendUpTo);
const toSend = escapeBuffer.substring(0, sendUpTo);
// Buffer output until client subscribes (fixes race in browser mode)
if (sessionData.clientSubscribed) {
onData(toSend);
} else {
sessionData.outputBuffer.push(toSend);
}
escapeBuffer = escapeBuffer.substring(sendUpTo);
}
});

Expand All @@ -201,15 +223,6 @@ export class PTYService {
this.sessions.delete(sessionId);
onExit(exitCode);
});

this.sessions.set(sessionId, {
pty: ptyProcess,
workspaceId: params.workspaceId,
workspacePath,
runtime,
onData,
onExit,
});
} else if (runtime instanceof SSHRuntime) {
// SSH: Use node-pty to spawn SSH with local PTY (enables resize support)
const sshConfig = runtime.getConfig();
Expand Down Expand Up @@ -263,29 +276,47 @@ export class PTYService {
);
}

// Create session data with buffer for pre-subscription output
const sessionData: SessionData = {
pty: ptyProcess,
workspaceId: params.workspaceId,
workspacePath,
runtime,
onData,
onExit,
outputBuffer: [],
clientSubscribed: false,
};
this.sessions.set(sessionId, sessionData);

// Handle data (same as local - buffer incomplete escape sequences)
let buffer = "";
let escapeBuffer = "";
ptyProcess.onData((data) => {
buffer += data;
let sendUpTo = buffer.length;
escapeBuffer += data;
let sendUpTo = escapeBuffer.length;

// Hold back incomplete escape sequences
if (buffer.endsWith("\x1b")) {
sendUpTo = buffer.length - 1;
} else if (buffer.endsWith("\x1b[")) {
sendUpTo = buffer.length - 2;
if (escapeBuffer.endsWith("\x1b")) {
sendUpTo = escapeBuffer.length - 1;
} else if (escapeBuffer.endsWith("\x1b[")) {
sendUpTo = escapeBuffer.length - 2;
} else {
// eslint-disable-next-line no-control-regex, @typescript-eslint/prefer-regexp-exec
const match = buffer.match(/\x1b\[[0-9;]*$/);
const match = escapeBuffer.match(/\x1b\[[0-9;]*$/);
if (match) {
sendUpTo = buffer.length - match[0].length;
sendUpTo = escapeBuffer.length - match[0].length;
}
}

if (sendUpTo > 0) {
const toSend = buffer.substring(0, sendUpTo);
onData(toSend);
buffer = buffer.substring(sendUpTo);
const toSend = escapeBuffer.substring(0, sendUpTo);
// Buffer output until client subscribes (fixes race in browser mode)
if (sessionData.clientSubscribed) {
onData(toSend);
} else {
sessionData.outputBuffer.push(toSend);
}
escapeBuffer = escapeBuffer.substring(sendUpTo);
}
});

Expand All @@ -295,16 +326,6 @@ export class PTYService {
this.sessions.delete(sessionId);
onExit(exitCode);
});

// Store PTY (same interface as local)
this.sessions.set(sessionId, {
pty: ptyProcess,
workspaceId: params.workspaceId,
workspacePath,
runtime,
onData,
onExit,
});
} else {
throw new Error(`Unsupported runtime type: ${runtime.constructor.name}`);
}
Expand All @@ -317,6 +338,34 @@ export class PTYService {
};
}

/**
* Subscribe to terminal output (flushes buffered output and starts streaming).
* In browser mode, output is buffered until the client subscribes to avoid
* losing initial output (like the shell prompt) during the HTTP round-trip.
*/
subscribeOutput(sessionId: string): void {
const session = this.sessions.get(sessionId);
if (!session) {
log.info(`Cannot subscribe to session ${sessionId}: not found`);
return;
}

if (session.clientSubscribed) {
log.debug(`Session ${sessionId} already subscribed`);
return;
}

// Mark as subscribed and flush buffered output
session.clientSubscribed = true;
if (session.outputBuffer.length > 0) {
log.debug(`Flushing ${session.outputBuffer.length} buffered chunks for ${sessionId}`);
for (const chunk of session.outputBuffer) {
session.onData(chunk);
}
session.outputBuffer = [];
}
}

/**
* Send input to a terminal session
*/
Expand Down