diff --git a/docs/protocol-compatibility.md b/docs/protocol-compatibility.md index c0c318d963..da73c2ebe1 100644 --- a/docs/protocol-compatibility.md +++ b/docs/protocol-compatibility.md @@ -56,7 +56,9 @@ without requesting streams. Application caches and which agents are visible rema `terminal_chunked_replay` is SDK-owned delivery behavior. The daemon sends new binary opcodes only to a source advertising the capability. The SDK keeps the established subscribe request against older -ANSI hosts and normalizes their single Restore frame into begin/data/end. JSON snapshot +ANSI hosts and normalizes their single Restore frame into begin/data/end. `observeTerminal` defaults +to an explicit full snapshot: omitting restore on an ANSI-era host selects its discarded JSON path. +Explicit caller restore modes and history counts pass through unchanged. JSON snapshot wire declarations remain accepted, but JSON snapshot frames have no rendering runtime. Terminals require `terminal-restore-modes`; older hosts show an update message without attaching. diff --git a/docs/terminal-performance.md b/docs/terminal-performance.md index a5faa617b6..b50d982515 100644 --- a/docs/terminal-performance.md +++ b/docs/terminal-performance.md @@ -25,7 +25,7 @@ Terminal frames share the daemon main event loop with all agent traffic. The `ev - **Replay delivery is isolated per viewer.** Live worker output never waits for app delivery. Each daemon stream owns its bounded live tail and revision deduplication; a slow or failed replay cannot pause another viewer of the same terminal. - **Replay geometry belongs to the parser queue.** Native measurements and web fits must leave the captured dimensions intact through every chunk's commit. Apply the latest measured size after replay, before live bytes; clear, replacement and disposal must release that geometry ownership. A layout effect must not resize the renderer directly during replay. - **Coalesced output carries the LAST chunk's revision.** Snapshot replay dedup (`replayTerminalOutputAfterSnapshot`) skips buffered output with `revision <= replayRevision`; a merged batch with a lower revision would be wrongly skipped (lost output). -- **The input-mode tracker runs once per process boundary, not per hop.** The worker owns the authoritative tracker; the daemon caches the replay preamble from replay responses and `snapshotReady` messages. Do not reintroduce a per-chunk `feed()` on the daemon main loop. +- **The input-mode tracker runs once per process boundary, not per hop.** The worker owns the authoritative tracker; the daemon caches current modes from ordered worker events and adoption. A replay response carries capture-time modes for its viewer; it must not overwrite the shared cache after newer live output. Apply that viewer's preamble before its buffered live bytes. Do not reintroduce a per-chunk `feed()` on the daemon main loop. - **Snapshot catch-up is backpressure-gated.** A stream falls back to a full snapshot only when `outputBytesSinceSnapshot > MAX_TERMINAL_OUTPUT_FRAME_BYTES` (256KB) **and** the client transport reports `bufferedAmount > MAX_CLIENT_BUFFERED_BYTES` (4MB). A client that keeps draining streams continuously, no matter how much output is produced. Before this gate existed, every 256KB of build output dropped a frame and forced a full JSON cell-grid snapshot (~200k objects across IPC) — the historical source of spiky lag and GC hitches. - **Plugin daemon sessions report IPC queue bytes.** Their virtual socket increments `bufferedAmount` before `process.send` and decrements it only in the send callback. Text and binary frames share that ordered queue, so the normal snapshot catch-up and physical high-water gates remain valid for server-side plugin SDK traffic. - **Client output writes are not serialized per frame.** The emulator runtime drains contiguous plain writes straight into xterm (which buffers internally). Only barrier ops (`clear`, `snapshot`, `suppressInput` writes) wait — behind a zero-length sentinel write — so resets can't interleave with in-flight output. diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 7d7d3a1c97..ba7e9e5f28 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -8486,3 +8486,89 @@ test("closes before admission when a protected-capable daemon is reached in lega expect(daemonReceived).toHaveLength(1); expect(JSON.parse(daemonReceived[0] as string)).toMatchObject({ type: "hello" }); }); + +test.each([false, true])( + "default terminal observation requests ANSI replay (chunked=%s)", + async (chunked) => { + const transport = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_unit_test", + logger: createMockLogger(), + reconnect: { enabled: false }, + transportFactory: () => transport.transport, + }); + clients.push(client); + const connecting = client.connect(); + transport.triggerOpen({ + features: { ownedSubscriptions: true, "terminal-restore-modes": true }, + }); + await connecting; + const events: import("./terminal-stream-router").TerminalStreamEvent[] = []; + const observation = client.observeTerminal("default-terminal", (event) => events.push(event)); + await vi.waitFor(() => + expect(parseSentFrame(transport.sent.at(-1)).type).toBe("subscribe_terminal_request"), + ); + const request = parseSentFrame(transport.sent.at(-1)); + expect(request.restore).toEqual({ mode: "full-snapshot" }); + transport.triggerMessage( + wrapSessionMessage({ + type: "subscribe_terminal_response", + payload: { + terminalId: "default-terminal", + requestId: request.requestId, + subscriptionId: "default-subscription", + slot: 2, + error: null, + }, + }), + ); + await observation.ready; + const data = new TextEncoder().encode("\x1b[31mretained 界🙂\x1b[0m"); + const metadata = { rows: 24, cols: 80, retainedHistoryLines: 1000 }; + if (chunked) { + transport.triggerMessage( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.ReplayBegin, + slot: 2, + payload: JSON.stringify(metadata), + }), + ); + transport.triggerMessage( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.ReplayData, + slot: 2, + payload: data, + }), + ); + transport.triggerMessage( + encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.ReplayEnd, slot: 2 }), + ); + } else { + // An ANSI-era host selects its existing Restore path for this explicit request. + transport.triggerMessage( + encodeTerminalStreamFrame({ opcode: TerminalStreamOpcode.Restore, slot: 2, payload: data }), + ); + } + expect(events).toEqual([ + { + terminalId: "default-terminal", + subscriptionId: "default-subscription", + type: "replay", + replay: chunked ? { type: "begin", ...metadata } : { type: "begin" }, + }, + { + terminalId: "default-terminal", + subscriptionId: "default-subscription", + type: "replay", + replay: { type: "data", data }, + }, + { + terminalId: "default-terminal", + subscriptionId: "default-subscription", + type: "replay", + replay: { type: "end" }, + }, + ]); + }, +); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 5a02b85a7a..e3eb826f78 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -6414,7 +6414,8 @@ export class DaemonClient { } /** - * Replay begin/data/end and live output arrive in receive. Subscribe to the returned handle + * Defaults to a full ANSI replay. Replay begin/data/end and live output arrive in receive. + * Subscribe to the returned handle * for terminal_stream_exit: payload.error means observation failure, not PTY exit. * Either outcome detaches this slot and releases the handle automatically. */ @@ -6433,7 +6434,7 @@ export class DaemonClient { { type: "subscribe_terminal_request", terminalId, - ...(options?.restore ? { restore: options.restore } : {}), + restore: options?.restore ?? { mode: "full-snapshot" }, }, options, ); 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 11156cd127..eed9ebfa0f 100644 --- a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts @@ -1931,3 +1931,67 @@ test.each([400, 3000])( }, 30000, ); + +test.each([false, true])( + "default SDK observation replays a terminal without waiting for output (quietScreen=%s)", + async (quietScreen) => { + const cwd = tmpCwd(); + tempDirs.push(cwd); + const created = await createTerminalInWorkspace(ctx.client, { + cwd, + options: { + command: process.execPath, + args: [ + "-e", + `${quietScreen ? 'process.stdout.write("QUIET-SCREEN\\r\\n");' : ""} process.stdin.on("data", data => process.stdout.write("LIVE:" + data)); setInterval(() => {}, 1000);`, + ], + }, + }); + const id = created.terminal!.id; + if (quietScreen) + await waitForCondition( + async () => (await ctx.client.captureTerminal(id)).lines.join("").includes("QUIET-SCREEN"), + 10000, + ); + const events: import("@getpaseo/client").TerminalStreamEvent[] = []; + const observation = ctx.client.observeTerminal(id, (event) => events.push(event)); + try { + await observation.ready; + await waitForCondition( + () => events.some((event) => event.type === "replay" && event.replay.type === "end"), + 10000, + ); + const replay = events.flatMap((event) => (event.type === "replay" ? [event.replay] : [])); + const begin = replay[0]; + if (begin?.type !== "begin" || !begin.rows || !begin.cols) + throw new Error("Missing replay begin geometry"); + expect(replay.at(-1)).toEqual({ type: "end" }); + const terminal = new HeadlessTerminal({ + rows: begin.rows, + cols: begin.cols, + allowProposedApi: true, + }); + try { + for (const event of replay) + if (event.type === "data") + await new Promise((resolve) => terminal.write(event.data, resolve)); + const text = extractStateText(readHeadlessState(terminal)); + expect(text).toBe(quietScreen ? "QUIET-SCREEN" : ""); + } finally { + terminal.dispose(); + } + ctx.client.sendTerminalInput(id, { type: "input", data: "next\r" }); + await waitForCondition( + () => + events.some( + (event) => + event.type === "output" && new TextDecoder().decode(event.data).includes("LIVE:next"), + ), + 10000, + ); + } finally { + await observation.release(); + } + }, + 30000, +); diff --git a/packages/server/src/terminal/terminal-session-controller.test.ts b/packages/server/src/terminal/terminal-session-controller.test.ts index 346a1f5c78..16c0ee5c12 100644 --- a/packages/server/src/terminal/terminal-session-controller.test.ts +++ b/packages/server/src/terminal/terminal-session-controller.test.ts @@ -23,10 +23,10 @@ interface ReplayFixture { data: Uint8Array; metadata: TerminalReplayMetadata; } -function terminalReplay(text: string, revision: number): ReplayFixture { +function terminalReplay(text: string, revision: number, replayPreamble = ""): ReplayFixture { return { data: Buffer.from(text), - metadata: { rows: 1, cols: 80, retainedHistoryLines: 0, revision, replayPreamble: "" }, + metadata: { rows: 1, cols: 80, retainedHistoryLines: 0, revision, replayPreamble }, }; } function replayFixture( @@ -168,7 +168,7 @@ describe("terminal-session-controller restore", () => { expect(binaryFrames).toEqual([]); terminalListener?.({ type: "output", data: "restore-after\n", revision: 2 }); - snapshot.resolve(terminalReplay("restore-before", 1)); + snapshot.resolve(terminalReplay("restore-before", 1, "\x1b[?1h\x1b[?2004h")); await snapshot.promise; await new Promise((resolve) => setTimeout(resolve, 20)); diff --git a/packages/server/src/terminal/terminal-session-controller.ts b/packages/server/src/terminal/terminal-session-controller.ts index eb60b14884..b1f680fa96 100644 --- a/packages/server/src/terminal/terminal-session-controller.ts +++ b/packages/server/src/terminal/terminal-session-controller.ts @@ -34,6 +34,7 @@ import { resolveTerminalSubscriptionSnapshotMode, type TerminalRestoreOptions, } from "./terminal-restore.js"; +import type { TerminalReplayMetadata } from "./terminal-replay.js"; import type { TerminalSession } from "./terminal.js"; import type { TerminalManager, TerminalsChangedEvent } from "./terminal-manager.js"; import { applyTerminalSize } from "./terminal-size-ownership.js"; @@ -76,10 +77,9 @@ interface ActiveTerminalStream { outputCoalescer: TerminalOutputCoalescer; } -interface SnapshotSendResult { - shouldContinue: boolean; - replayRevision?: number; -} +type SnapshotSendResult = + | { shouldContinue: false } + | { shouldContinue: true; replay: TerminalReplayMetadata | null }; export interface TerminalSessionControllerOptions { terminalManager: TerminalManager | null; @@ -1041,7 +1041,7 @@ export class TerminalSessionController { return; } activeStream.snapshotOutput = undefined; - this.replayTerminalOutputAfterSnapshot(activeStream, terminal, snapshotResult.replayRevision); + this.replayTerminalOutputAfterSnapshot(activeStream, terminal, snapshotResult.replay); activeStream.needsSnapshot = false; activeStream.outputBytesSinceSnapshot = 0; } catch (error) { @@ -1090,7 +1090,7 @@ export class TerminalSessionController { ? { scrollbackLines: restore.scrollbackLines } : resolveTerminalRestoreSnapshotOptions(restore); if (snapshotOptions === null) { - return { shouldContinue: true }; + return { shouldContinue: true, replay: null }; } const deliver = createTerminalRestoreDelivery({ @@ -1118,15 +1118,17 @@ export class TerminalSessionController { // The restore frame went out-of-band; keep the replay that follows on the // coalescer's trailing path so it doesn't flush back-to-back with it. activeStream.outputCoalescer.markFlushed(); - return { shouldContinue: true, replayRevision: snapshot.revision }; + return { shouldContinue: true, replay: snapshot }; } private replayTerminalOutputAfterSnapshot( activeStream: ActiveTerminalStream, terminal: TerminalSession, - replayRevision: number | undefined, + replay: TerminalReplayMetadata | null, ): void { - const replayPreamble = terminal.getReplayPreamble(); + // A captured viewer must start from its cut, then apply the buffered mode changes. + // Live-only subscriptions have no capture and start from the current shared modes. + const replayPreamble = replay === null ? terminal.getReplayPreamble() : replay.replayPreamble; if (replayPreamble.length > 0) { activeStream.outputCoalescer.handle(replayPreamble); } @@ -1137,11 +1139,7 @@ export class TerminalSessionController { activeStream.bufferedOutputs.length, ); for (const output of bufferedOutputs) { - if ( - replayRevision !== undefined && - output.revision !== undefined && - output.revision <= replayRevision - ) { + if (replay !== null && output.revision !== undefined && output.revision <= replay.revision) { continue; } activeStream.outputCoalescer.handle(output.data); diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index 7a22caaa02..a37ccd46db 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -475,6 +475,14 @@ function runWorker(argv: string[]): void { terminal: toTerminalInfo(session), state, }); + // Startup output/readiness may have arrived while recording the program, + // before the daemon knew this terminal. Publish its current modes after creation. + sendToController({ + type: "terminalMessage", + terminalId: session.id, + message: { type: "snapshotReady" }, + replayPreamble: session.getReplayPreamble(), + }); sendToController({ type: "response", requestId: message.requestId, diff --git a/packages/server/src/terminal/worker-terminal-manager.test.ts b/packages/server/src/terminal/worker-terminal-manager.test.ts index c06f64e868..c3c55309d8 100644 --- a/packages/server/src/terminal/worker-terminal-manager.test.ts +++ b/packages/server/src/terminal/worker-terminal-manager.test.ts @@ -1,3 +1,6 @@ +import { TerminalInputModeTracker } from "@getpaseo/protocol/terminal-input-mode"; +import { encodeTerminalKeyInput } from "@getpaseo/protocol/terminal-key-input"; +import type { SubscribeTerminalRequest } from "@getpaseo/protocol/messages"; import { TerminalSessionController } from "./terminal-session-controller.js"; import { SessionDelivery } from "../server/session/owned-subscriptions/index.js"; import { @@ -385,27 +388,36 @@ it.each(["current", "draining"] as const)( "keeps the peer usable when the %s worker dies", async (target) => { const paseoHome = createTemporaryHome(); - await launchDetachedWorkerWithTerminal(paseoHome); - const old = await launchDetachedWorkerWithTerminal(paseoHome); + const seeds = [ + await launchDetachedWorkerWithTerminal(paseoHome), + await launchDetachedWorkerWithTerminal(paseoHome), + ]; + // Exercise an existing seed adopted as current; launch age does not imply draining ownership. + const [firstAdopted] = await workerRecords(paseoHome); + const seed = seeds.find((candidate) => candidate.workerPid === firstAdopted!.pid)!; manager = await startRealWorkerManager({ paseoHome }); const added = await manager.createTerminal({ workspaceId: "ws-peer", - cwd: old.cwd, + cwd: seed.cwd, ...nodeTerminalCommand("setInterval(() => {}, 1000)"), }); - const current = (await workerRecords(paseoHome)).find( - (record) => record.pid !== old.workerPid, + const records = await workerRecords(paseoHome); + const current = records.find((record) => + record.terminals.some((terminal) => terminal.id === added.id), )!; - const deadId = target === "current" ? added.id : old.terminalId; - const healthyId = target === "current" ? old.terminalId : added.id; + const draining = records.find((record) => record.pid !== current.pid)!; + const drainingId = draining.terminals[0]!.id; + const deadId = target === "current" ? added.id : drainingId; + const healthyId = target === "current" ? drainingId : added.id; const healthy = manager.getTerminal(healthyId); - process.kill(target === "current" ? current.pid : old.workerPid, "SIGKILL"); + expect(healthy).toBeDefined(); + process.kill(target === "current" ? current.pid : draining.pid, "SIGKILL"); await waitForCondition(() => !manager!.getTerminal(deadId), 10000); expect(manager.getTerminal(healthyId)).toBe(healthy); expect(await readReplay(manager, healthyId)).not.toBeNull(); await manager.createTerminal({ workspaceId: "ws-peer", - cwd: old.cwd, + cwd: seed.cwd, ...nodeTerminalCommand("setInterval(() => {}, 1000)"), }); expect((await workerRecords(paseoHome)).length).toBeLessThanOrEqual(2); @@ -546,7 +558,7 @@ it("pulls fresh terminal state from the worker authority", async () => { // preamble-caching contract is verified on Linux/macOS; the daemon's input-mode // handling runs identically on every platform once the escape is observed. it.skipIf(isPlatform("win32"))( - "caches the input-mode replay preamble from the worker after getTerminalState", + "caches input modes from worker output independently of replay capture", async () => { const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-preamble-")); temporaryDirs.push(cwd); @@ -564,11 +576,10 @@ it.skipIf(isPlatform("win32"))( }), ); - await waitForCondition(async () => { - const snapshot = await readReplay(manager!, session.id); - return snapshot !== null && session.getReplayPreamble() === "\x1b[=1;1u"; - }, 10000); - + await waitForCondition(() => session.getReplayPreamble() === "\x1b[=1;1u", 10000); + // Initial shared modes must be ready without a replay response repairing the cache. + expect(session.getReplayPreamble()).toBe("\x1b[=1;1u"); + expect(await readReplay(manager!, session.id)).not.toBeNull(); expect(session.getReplayPreamble()).toBe("\x1b[=1;1u"); }, ); @@ -855,6 +866,7 @@ it("never adopts a live worker's terminals after a kill cleanup it could not fin // the shim; the PATH is restored as soon as it is up. async function startWorkerWithSlowProcessTable( delaySeconds: number, + options: { paseoHome?: string; requestTimeoutMs?: number } = {}, ): Promise { const realPs = (process.env.PATH ?? "") .split(delimiter) @@ -868,7 +880,7 @@ async function startWorkerWithSlowProcessTable( const originalPath = process.env.PATH; process.env.PATH = `${shimDir}${delimiter}${originalPath ?? ""}`; try { - return await startRealWorkerManager(); + return await startRealWorkerManager(options); } finally { process.env.PATH = originalPath; } @@ -1982,13 +1994,20 @@ function terminalViewers( }, started: () => started, finish, - async attach(source: object, requestId: string) { + async attach( + source: object, + requestId: string, + restore: NonNullable = { + mode: "visible-snapshot", + scrollbackLines: 1000, + }, + ) { delivery.attach(source, true); const request = { type: "subscribe_terminal_request" as const, terminalId, requestId, - restore: { mode: "visible-snapshot" as const, scrollbackLines: 1000 }, + restore, }; await delivery.request(source, request, () => controller.dispatch(request, delivery)); }, @@ -2113,20 +2132,28 @@ it("keeps concurrent viewer replays and live output ordered behind slow app deli it.each([true, false])( "queues replay independently of app delivery past the worker deadline (chunked=%s)", async (chunked) => { - manager = await startRealWorkerManager({ requestTimeoutMs: 800 }); - const session = trackTerminal( - await manager.createTerminal({ - cwd: createTemporaryHome(), - workspaceId: "slow-delivery", - ...nodeTerminalCommand( - 'process.stdout.write("READY\\r\\n"); process.stdin.on("data", data => process.stdout.write("ECHO:" + data)); setInterval(() => {}, 1000);', - ), - }), - ); + const paseoHome = createTemporaryHome(); + // Slow program identification belongs to setup, not the replay request deadline. + // The POSIX shim deterministically exceeds 800 ms; Windows keeps its real OS path. + manager = isPlatform("win32") + ? await startRealWorkerManager({ paseoHome }) + : await startWorkerWithSlowProcessTable(1, { paseoHome }); + const created = await manager.createTerminal({ + cwd: paseoHome, + workspaceId: "slow-delivery", + ...nodeTerminalCommand( + 'process.stdout.write("READY\\r\\n"); process.stdin.on("data", data => process.stdout.write("ECHO:" + data)); setInterval(() => {}, 1000);', + ), + }); await waitForCondition( - async () => (await manager!.captureTerminal(session.id)).lines.join("").includes("READY"), + async () => (await manager!.captureTerminal(created.id)).lines.join("").includes("READY"), 10000, ); + await manager.detach(); + manager = await startRealWorkerManager({ paseoHome, requestTimeoutMs: 800 }); + const session = manager.getTerminal(created.id); + if (!session) throw new Error("Replay fixture did not adopt its prepared terminal"); + trackTerminal(session); const viewers = terminalViewers(manager, session.id, chunked); const phone = {}; viewers.block(phone); @@ -2208,3 +2235,108 @@ it.each(["replacement", "socket-overflow"] as const)( }, 30000, ); + +it.each([true, false])( + "keeps live modes after a replay captured enabled=%s while restoring its own modes first", + async (enabled) => { + const on = "\x1b[?1h\x1b[?2004h"; + const off = "\x1b[?1l\x1b[?2004l"; + const captured = enabled ? on : ""; + const latest = enabled ? "" : on; + const live = (enabled ? off : on) + "MODE-LIVE"; + const worker = new FakeTerminalWorker(); + manager = createWorkerTerminalManager({ worker, requestTimeoutMs: 1000 }); + worker.emitWorkerMessage({ + type: "terminalCreated", + terminal: { + id: "mode-terminal", + name: "Terminal", + cwd: "/tmp", + workspaceId: "modes", + activity: null, + }, + state: createTerminalState(), + }); + worker.emitWorkerMessage({ + type: "terminalMessage", + terminalId: "mode-terminal", + replayPreamble: captured, + message: { type: "snapshotReady", revision: 1 }, + }); + const session = manager.getTerminal("mode-terminal")!; + const viewers = terminalViewers(manager, session.id); + const restoring = {}, + following = {}; + try { + await viewers.attach(restoring, "restoring"); + await waitForCondition( + () => worker.sentMessages.some((message) => message.type === "replayTerminal"), + 1000, + ); + const request = worker.sentMessages.find((message) => message.type === "replayTerminal")!; + const metadata = { + rows: 30, + cols: 120, + retainedHistoryLines: 10, + revision: 1, + replayPreamble: captured, + }; + worker.emitWorkerMessage({ + type: "terminalReplay", + requestId: request.requestId, + event: { type: "begin", metadata }, + }); + worker.emitWorkerMessage({ + type: "terminalReplay", + requestId: request.requestId, + event: { type: "data", data: Buffer.from("CAPTURE-A") }, + }); + worker.emitWorkerMessage({ + type: "terminalMessage", + terminalId: session.id, + replayPreamble: latest, + message: { type: "output", data: live, revision: 2 }, + }); + expect(session.getReplayPreamble()).toBe(latest); + worker.emitWorkerMessage({ + type: "terminalReplay", + requestId: request.requestId, + event: { type: "data", data: Buffer.from("CAPTURE-B") }, + }); + worker.emitWorkerMessage({ + type: "terminalReplay", + requestId: request.requestId, + event: { type: "end" }, + }); + worker.emitWorkerMessage({ + type: "response", + requestId: request.requestId, + ok: true, + result: metadata, + }); + await waitForCondition(() => viewers.text(restoring).includes("MODE-LIVE"), 1000); + expect(session.getReplayPreamble()).toBe(latest); + expect(viewers.text(restoring)).toBe(captured + live); + const sequence = viewers.frames + .filter((row) => row.source === restoring) + .map((row) => row.frame.opcode); + expect(sequence.indexOf(TerminalStreamOpcode.Output)).toBeGreaterThan( + sequence.indexOf(TerminalStreamOpcode.ReplayEnd), + ); + const modes = new TerminalInputModeTracker(); + modes.feed(session.getReplayPreamble()); + expect(encodeTerminalKeyInput({ key: "ArrowUp" }, { inputMode: modes.getState() })).toBe( + enabled ? "\x1b[A" : "\x1bOA", + ); + expect(modes.getState().bracketedPaste).toBe(!enabled); + await viewers.attach(following, "live-following", { mode: "live" }); + await waitForCondition(() => latest === "" || viewers.text(following).length > 0, 1000); + expect(viewers.text(following)).toBe(latest); + expect( + worker.sentMessages.filter((message) => message.type === "replayTerminal"), + ).toHaveLength(1); + } finally { + await viewers.delivery.close(); + } + }, +); diff --git a/packages/server/src/terminal/worker-terminal-manager.ts b/packages/server/src/terminal/worker-terminal-manager.ts index 5f2f7712fb..8e34d56484 100644 --- a/packages/server/src/terminal/worker-terminal-manager.ts +++ b/packages/server/src/terminal/worker-terminal-manager.ts @@ -885,8 +885,6 @@ export function createWorkerTerminalManager( { type: "replayTerminal", terminalId: id, options }, sink, )) as TerminalReplayMetadata | null; - const record = recordsById.get(id); - if (record && metadata) record.replayPreamble = metadata.replayPreamble; return metadata; },