diff --git a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts index aa02d0a9a..24873b13b 100644 --- a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts @@ -457,6 +457,102 @@ describe("daemon E2E terminal", () => { rmSync(cwd, { recursive: true, force: true }); }, 30000); + test("subscribe response is sent before the initial snapshot frame", async () => { + const cwd = tmpCwd(); + const created = await ctx.client.createTerminal(cwd); + const terminalId = created.terminal!.id; + const ws = await connectRawWebSocket(ctx.daemon.port); + + try { + ctx.client.sendTerminalInput(terminalId, { + type: "input", + data: "printf 'hello-ordering\\n'\r", + }); + await new Promise((resolve) => setTimeout(resolve, 300)); + + const observed = await new Promise>((resolve, reject) => { + const events: Array<"response" | "snapshot"> = []; + const timeoutHandle = setTimeout(() => { + cleanup(); + reject(new Error("Timed out waiting for subscribe response and snapshot")); + }, 10000); + + const cleanup = () => { + clearTimeout(timeoutHandle); + ws.off("message", onMessage); + ws.off("error", onError); + }; + + const maybeResolve = () => { + if (!events.includes("response") || !events.includes("snapshot")) { + return; + } + cleanup(); + resolve(events); + }; + + const onMessage = (raw: WebSocket.RawData) => { + const buffer = toWsBuffer(raw); + const text = typeof raw === "string" ? raw : buffer?.toString("utf8"); + if (text) { + try { + const parsedResult = WSOutboundMessageSchema.safeParse(JSON.parse(text)); + if ( + parsedResult.success && + parsedResult.data.type === "session" && + parsedResult.data.message.type === "subscribe_terminal_response" && + parsedResult.data.message.payload.requestId === "sub-ordering" + ) { + events.push("response"); + maybeResolve(); + return; + } + } catch { + // ignore non-session text frames + } + } + + if (!buffer) { + return; + } + const frame = decodeTerminalStreamFrame(new Uint8Array(buffer)); + if (frame?.opcode !== TerminalStreamOpcode.Snapshot) { + return; + } + const state = decodeTerminalSnapshotPayload(frame.payload); + if (!state || !extractStateText(state).includes("hello-ordering")) { + return; + } + events.push("snapshot"); + maybeResolve(); + }; + + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + + ws.on("message", onMessage); + ws.on("error", onError); + ws.send( + JSON.stringify({ + type: "session", + message: { + type: "subscribe_terminal_request", + terminalId, + requestId: "sub-ordering", + }, + }), + ); + }); + + expect(observed).toEqual(["response", "snapshot"]); + } finally { + await closeWebSocket(ws); + rmSync(cwd, { recursive: true, force: true }); + } + }, 30000); + test("client sends input and receives output as raw bytes", async () => { const cwd = tmpCwd(); const created = await ctx.client.createTerminal(cwd); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index d4dbe6867..99196d94c 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -7712,6 +7712,11 @@ export class Session { requestId: msg.requestId, }, }); + + const activeStream = this.activeTerminalStreams.get(slot); + if (activeStream) { + this.trySendTerminalSnapshot(activeStream); + } } private handleUnsubscribeTerminalRequest(msg: UnsubscribeTerminalRequest): void { @@ -7812,7 +7817,6 @@ export class Session { const existingStream = this.activeTerminalStreams.get(existingSlot); if (existingStream) { existingStream.needsSnapshot = true; - this.trySendTerminalSnapshot(existingStream); return existingSlot; } this.terminalIdToSlot.delete(terminal.id); @@ -7860,7 +7864,6 @@ export class Session { this.markAllActiveTerminalStreamsForSnapshot(); } }); - this.trySendTerminalSnapshot(activeStream); return slot; }