diff --git a/apps/server/src/mcp/PreviewAutomationBroker.test.ts b/apps/server/src/mcp/PreviewAutomationBroker.test.ts index d3ba6911ad9f..39dd98bf235f 100644 --- a/apps/server/src/mcp/PreviewAutomationBroker.test.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.test.ts @@ -1540,3 +1540,84 @@ it.effect("keeps each thread's current tab when threads share a provider session }), ), ); + +it.effect.each([ + { clientId: SERVER_BROWSER_AUTOMATION_CLIENT_ID, retained: true }, + { clientId: "remote-preferred", retained: false }, +])( + "an unanswered action preserves other threads only on the in-process host ($clientId)", + ({ clientId, retained }) => + Effect.scoped( + Effect.gen(function* () { + const broker = yield* makeBroker; + const otherScope = { + ...scope, + thread: { + ...scope.thread, + threadId: ThreadId.make("thread-2"), + providerSessionId: "provider-session-2", + }, + }; + const connected = yield* Deferred.make(); + const slowReceived = yield* Deferred.make(); + const otherReceived = yield* Deferred.make(); + const requests = requestsFrom( + yield* broker.connect(makeHost({ clientId }), { preferred: true }), + () => Deferred.doneUnsafe(connected, Effect.void), + ); + yield* Stream.runForEach(requests, (request) => { + if (request.operation === "snapshot") return Deferred.succeed(slowReceived, undefined); + if (request.operation === "evaluate") return Deferred.succeed(otherReceived, request); + return broker.respond({ + clientId, + connectionId: request.connectionId, + requestId: request.requestId, + ok: true, + result: { + tabId: request.operation === "open" ? `tab-${request.threadId}` : request.tabId, + }, + }); + }).pipe(Effect.forkScoped); + + yield* Deferred.await(connected); + for (const caller of [scope, otherScope]) + yield* broker.invoke({ scope: caller, operation: "open", input: {} }); + const pending = yield* broker + .invoke({ scope: otherScope, operation: "evaluate", input: {}, timeoutMs: 10_000 }) + .pipe(Effect.result, Effect.forkScoped); + const otherRequest = yield* Deferred.await(otherReceived); + const timedOut = yield* broker + .invoke({ scope, operation: "snapshot", input: {}, timeoutMs: 1_000 }) + .pipe(Effect.flip, Effect.forkScoped); + yield* Deferred.await(slowReceived); + yield* TestClock.adjust(2_000); + expect(yield* Fiber.join(timedOut)).toMatchObject({ + _tag: "PreviewAutomationTimeoutError", + }); + + yield* broker.respond({ + clientId, + connectionId: otherRequest.connectionId, + requestId: otherRequest.requestId, + ok: true, + result: { tabId: otherRequest.tabId }, + }); + const otherResult = yield* Fiber.join(pending); + expect(otherResult).toMatchObject( + retained + ? { _tag: "Success", success: { tabId: "tab-thread-2" } } + : { _tag: "Failure", failure: { _tag: "PreviewAutomationClientDisconnectedError" } }, + ); + for (const caller of [scope, otherScope]) { + const next = yield* broker + .invoke({ scope: caller, operation: "click", input: {} }) + .pipe(Effect.result); + expect(next).toMatchObject( + retained + ? { _tag: "Success", success: { tabId: `tab-${caller.thread.threadId}` } } + : { _tag: "Failure", failure: { _tag: "PreviewAutomationNoAvailableHostError" } }, + ); + } + }), + ), +); diff --git a/apps/server/src/mcp/PreviewAutomationBroker.ts b/apps/server/src/mcp/PreviewAutomationBroker.ts index c1ae21dfd3b4..7cca62108d19 100644 --- a/apps/server/src/mcp/PreviewAutomationBroker.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.ts @@ -654,7 +654,11 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() { // becoming unreachable. A background metadata read has a short budget // and changes nothing, so a slow one must not cut the host off from the // agent's next call. - if (input.updateCurrentTab !== false) { + // A slow in-process page must not drop other threads' requests and tab assignments. + if ( + input.updateCurrentTab !== false && + connection.clientId !== SERVER_BROWSER_AUTOMATION_CLIENT_ID + ) { yield* disconnect(connection.clientId, connection.queue, true); } return yield* new PreviewAutomationTimeoutError(requestContext); diff --git a/apps/server/src/preview/ServerBrowser.test.ts b/apps/server/src/preview/ServerBrowser.test.ts index 861fc063a86e..e8cbce27dc5d 100644 --- a/apps/server/src/preview/ServerBrowser.test.ts +++ b/apps/server/src/preview/ServerBrowser.test.ts @@ -46,7 +46,9 @@ vi.mock("./ServerBrowserContexts.ts", () => ({ return context as unknown as BrowserContext; } async scratchPage() { - return makeContext().page as unknown as Page; + const page = makeContext().page; + scratchPages.push(page); + return page as unknown as Page; } async connectDesktopPage(endpoint: string) { const context = makeContext(); @@ -73,7 +75,7 @@ function makeSession(frame?: ReturnType) { const events = new NodeEvents.EventEmitter(); return { on: vi.fn((name: string, listener: () => void) => events.on(name, listener)), - emit: (name: string) => events.emit(name), + emit: (name: string, ...args: unknown[]) => events.emit(name, ...args), detach: vi.fn(async () => {}), send: vi.fn(async (method: string, input?: unknown): Promise> => { if (method === "Page.getNavigationHistory") return { currentIndex: 0, entries: [{}] }; @@ -154,7 +156,7 @@ function makeContext(onClose?: (context: BrowserContext) => void) { reload: vi.fn(async () => {}), emulateMedia: vi.fn(async () => {}), waitForLoadState: vi.fn(async () => {}), - evaluate: vi.fn(async () => ({ + evaluate: vi.fn(async (): Promise => ({ url, title: "test page", loading: false, @@ -196,6 +198,7 @@ function makeContext(onClose?: (context: BrowserContext) => void) { } const contexts: ReturnType[] = []; +const scratchPages: ReturnType["page"][] = []; const contextRequests: Array<{ profileId: string; isolated: boolean }> = []; let contextGate: PromiseWithResolvers | null = null; let pageCopy = ""; @@ -309,6 +312,7 @@ const viewerInput = (tabId: string, canOperate: boolean) => ({ beforeEach(() => { contexts.length = 0; + scratchPages.length = 0; contextRequests.length = 0; contextGate = null; contextFailure = null; @@ -1797,3 +1801,281 @@ it.live("a desktop page that comes back reconnects without waiting for a viewer" }), ).pipe(Effect.provide(layer)), ); + +it.live.each(["wait", "pause", "capture", "resume"] as const)( + "a stalled snapshot %s releases the tab queue and capture lock without leaking a paused viewer", + (stage) => + Effect.scoped( + Effect.gen(function* () { + const { browser, broker, tabId } = yield* ready; + const viewer = yield* browser.attachViewer(viewerInput(tabId, stage === "wait")); + if (stage === "wait") { + yield* viewer.input({ type: "takeControl" }); + yield* viewer.input({ type: "key", action: "down", key: "Shift", code: "ShiftLeft" }); + yield* viewer.input({ type: "mouse", action: "down", button: "left", x: 10, y: 20 }); + } + // Recovery must reach a viewer even when its output queue is full. + stallViewer(viewer, { _tag: "viewport", width: 1280, height: 800 }); + const { page, sessions } = contexts[0]!; + const session = sessions.at(-1)!; + const blocked = Promise.withResolvers(); + const release = Promise.withResolvers(); + const send = session.send.getMockImplementation()!; + let streaming = true; + let stalled = false; + session.send.mockImplementation(async (method, input) => { + if ( + !stalled && + ((stage === "wait" && method === "Page.captureScreenshot") || + (stage === "pause" && method === "Page.stopScreencast") || + (stage === "resume" && method === "Page.startScreencast")) + ) { + stalled = true; + blocked.resolve(); + if (stage !== "wait") await release.promise; + } + if (stalled && stage === "wait") await release.promise; + const result = await send(method, input); + if (method === "Page.stopScreencast") streaming = false; + if (method === "Page.startScreencast") streaming = true; + return result; + }); + const cdp = sessions[0]!; + if (stage === "wait") { + session.detach.mockImplementation(() => release.promise); + const tabSend = cdp.send.getMockImplementation()!; + cdp.send.mockImplementation(async (method, input) => { + const result = await tabSend(method, input); + if (method.startsWith("Input.dispatch")) await release.promise; + return result; + }); + } + if (stage === "capture") { + page.ariaSnapshot.mockImplementationOnce(async () => { + blocked.resolve(); + await release.promise; + return '- button "delete" [ref=e1]'; + }); + } + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + release.resolve(); + vi.useRealTimers(); + }), + ); + if (stage === "wait") { + session.emit("Page.screencastFrame", { data: "ZnJhbWU=", sessionId: 1 }); + session.emit("Page.screencastFrame", { data: "ZnJhbWU=", sessionId: 2 }); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(200)); + yield* Effect.promise(() => blocked.promise); + } + const snapshot = yield* broker + .invoke({ + // A different agent can observe while the viewer controls the owning agent's tab. + scope: stage === "wait" ? asSession("agent-b") : scope, + tabId, + operation: "snapshot", + input: {}, + }) + .pipe(Effect.flip, Effect.forkScoped({ startImmediately: true })); + yield* Effect.promise(() => blocked.promise); + // The host must answer within the action budget, before the broker timeout. + if (stage === "wait") { + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + // Exercise simultaneous expiry before the queued capture can start. + yield* Effect.sync(() => vi.advanceTimersByTime(15_000)); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + } else { + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(15_000)); + } + expect.soft(snapshot.pollUnsafe()).toMatchObject({ + _tag: "Success", + value: { + _tag: "PreviewAutomationTimeoutError", + remoteTag: "PreviewAutomationTimeoutError", + }, + }); + + if (stage === "wait") { + expect.soft(cdp.send).toHaveBeenCalledWith("Input.dispatchKeyEvent", { + type: "keyUp", + key: "Shift", + code: "ShiftLeft", + }); + expect.soft(cdp.send).toHaveBeenCalledWith("Input.dispatchMouseEvent", { + type: "mouseReleased", + button: "left", + x: 10, + y: 20, + buttons: 0, + clickCount: 1, + }); + // Even the healthy session's release acknowledgements cannot hold the control queue. + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(1_000)); + } + const sendsAtTimeout = session.send.mock.calls.length; + const releasesAtTimeout = cdp.send.mock.calls.filter(([method]) => + method.startsWith("Input.dispatch"), + ).length; + + const evaluation = yield* broker + .invoke({ scope, tabId, operation: "evaluate", input: { expression: "1 + 1" } }) + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect.soft(evaluation.pollUnsafe()).toMatchObject({ _tag: "Success", value: "evaluated" }); + + const recovered = yield* broker + .invoke({ scope, tabId, operation: "snapshot", input: {} }) + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect.soft(recovered.pollUnsafe()).toMatchObject({ + _tag: "Success", + value: { visibleText: "delete", screenshot: { data: "ZnJhbWU=" } }, + }); + // A failed CDP session may be retired, but cannot remain attached and paused. + expect.soft(streaming || session.detach.mock.calls.length > 0).toBe(true); + let replacement: ReturnType | undefined; + if (stage !== "capture") { + expect(session.detach).toHaveBeenCalled(); + expect(yield* Queue.clear(viewer.output)).toContainEqual({ _tag: "reconnect" }); + const reattached = yield* browser.attachViewer(viewerInput(tabId, false)); + replacement = sessions.at(-1)!; + expect(replacement.send).toHaveBeenCalledWith("Page.startScreencast", expect.anything()); + if (stage === "wait") + expect(yield* Queue.clear(reattached.output)).toContainEqual( + expect.objectContaining({ _tag: "control", controller: "agent" }), + ); + } + + release.resolve(); + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect(streaming || session.detach.mock.calls.length > 0).toBe(true); + if (replacement) expect(replacement.detach).not.toHaveBeenCalled(); + if (stage === "wait") { + expect(page.ariaSnapshot).toHaveBeenCalledTimes(1); + expect(session.send).toHaveBeenCalledTimes(sendsAtTimeout); + expect( + cdp.send.mock.calls.filter(([method]) => method.startsWith("Input.dispatch")), + ).toHaveLength(releasesAtTimeout); + expect(yield* Queue.clear(viewer.output)).toEqual([]); + } + }), + ).pipe(Effect.provide(layer)), +); + +it.live( + "a snapshot timeout preserves recorded frames and retires a persistently stalled recording session", + () => + Effect.scoped( + Effect.gen(function* () { + const { broker, tabId } = yield* ready; + yield* broker.invoke({ scope, tabId, operation: "recordingStart", input: {} }); + const session = contexts[0]!.sessions.at(-1)!; + const encoder = scratchPages.at(-1)!; + const blocked = Promise.withResolvers(); + const release = Promise.withResolvers(); + const send = session.send.getMockImplementation()!; + let stops = 0; + session.send.mockImplementation(async (method, input) => { + if (method === "Page.stopScreencast" && ++stops === 2) blocked.resolve(); + await release.promise; + return send(method, input); + }); + // Playwright detach waits for a command on the same stalled CDP session. + session.detach.mockImplementation(() => release.promise); + // Reaching the encoder limit must not tie frame draining to the stalled stop command. + encoder.evaluate.mockResolvedValueOnce(false); + session.emit("Page.screencastFrame", { + data: "ZnJhbWU=", + sessionId: 1, + metadata: { deviceWidth: 1280 }, + }); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + release.resolve(); + vi.useRealTimers(); + }), + ); + const snapshot = yield* broker + .invoke({ scope, tabId, operation: "snapshot", input: {} }) + .pipe(Effect.flip, Effect.forkScoped({ startImmediately: true })); + yield* Effect.promise(() => blocked.promise); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(15_000)); + expect(snapshot.pollUnsafe()).toMatchObject({ + _tag: "Success", + value: { + _tag: "PreviewAutomationTimeoutError", + remoteTag: "PreviewAutomationTimeoutError", + }, + }); + expect(session.detach).toHaveBeenCalledTimes(1); + const sendsAtTimeout = session.send.mock.calls.length; + const framesAtTimeout = encoder.evaluate.mock.calls.length; + session.emit("Page.screencastFrame", { + data: "bGF0ZQ==", + sessionId: 2, + metadata: { deviceWidth: 1280 }, + }); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect.soft(encoder.evaluate).toHaveBeenCalledTimes(framesAtTimeout); + + const recovered = yield* broker + .invoke({ scope, tabId, operation: "snapshot", input: {} }) + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect.soft(recovered.pollUnsafe()).toMatchObject({ + _tag: "Success", + value: { visibleText: "delete" }, + }); + + encoder.evaluate + .mockResolvedValueOnce({ mimeType: "video/webm", count: 1, bytes: 5 }) + .mockResolvedValueOnce("ZnJhbWU="); + const stopping = yield* broker + .invoke<{ path: string; mimeType: string; sizeBytes: number }>({ + scope, + tabId, + operation: "recordingStop", + input: {}, + }) + .pipe(Effect.forkScoped({ startImmediately: true })); + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect(encoder.close).toHaveBeenCalled(); + const stopped = yield* Fiber.join(stopping); + expect(session.send).toHaveBeenCalledTimes(sendsAtTimeout); + expect(session.detach).toHaveBeenCalledTimes(1); + const fs = yield* FileSystem.FileSystem; + expect(stopped).toMatchObject({ mimeType: "video/webm", sizeBytes: 5 }); + expect(yield* fs.readFileString(stopped.path)).toBe("frame"); + yield* broker.invoke({ scope, tabId, operation: "recordingStart", input: {} }); + const replacement = contexts[0]!.sessions.at(-1)!; + expect(replacement).not.toBe(session); + + release.resolve(); + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(0)); + expect( + session.send.mock.calls.filter(([method]) => method === "Page.startScreencast"), + ).toHaveLength(1); + expect(replacement.detach).not.toHaveBeenCalled(); + const next = yield* broker.invoke({ + scope, + tabId, + operation: "snapshot", + input: {}, + }); + expect(next.visibleText).toBe("delete"); + expect( + replacement.send.mock.calls.filter(([method]) => method === "Page.startScreencast"), + ).toHaveLength(2); + }), + ).pipe(Effect.provide(layer)), +); diff --git a/apps/server/src/preview/ServerBrowser.ts b/apps/server/src/preview/ServerBrowser.ts index 0f134133acc1..88dab63daa6e 100644 --- a/apps/server/src/preview/ServerBrowser.ts +++ b/apps/server/src/preview/ServerBrowser.ts @@ -94,6 +94,8 @@ const SCREENCAST_MOTION_QUALITY = 50; const HOST_RECONNECT_DELAY = "1 second"; const DESKTOP_ATTACH_TIMEOUT = "10 seconds"; const VIEWER_OUTPUT_LIMIT = 64; +const SNAPSHOT_TIMEOUT_MS = 15_000; +const VIEWER_INPUT_RELEASE_TIMEOUT_MS = 1_000; const RECORDING_SCREENCAST = { format: "jpeg", quality: 90, everyNthFrame: 1 } as const; const decodeViewportSetting = Schema.decodeUnknownSync(PreviewViewportSettingSchema); const AGENT_CURSOR_MOVE_MS = 160; @@ -250,6 +252,7 @@ interface ViewerState { readonly push: (output: ServerBrowserViewerOutput) => void; readonly pause: () => Promise; readonly resume: () => Promise; + readonly disconnect: () => void; scrolledAt: number; requestedSize: { width: number; height: number; order: number } | null; @@ -266,7 +269,7 @@ interface EncoderWindow { interface Recording { readonly encoder: Page; - readonly session: CDPSession; + session: CDPSession | null; readonly startedAt: string; readonly framesInFlight: Set>; } @@ -1462,7 +1465,14 @@ const make = Effect.gen(function* () { const opened = await tab.page.context().newCDPSession(tab.page); session = opened; const framesInFlight = new Set>(); + const recording: Recording = { + encoder, + session: opened, + startedAt: new Date().toISOString(), + framesInFlight, + }; opened.on("Page.screencastFrame", (frame) => { + if (recording.session !== opened) return; const cssWidth = tab.page.viewportSize()?.width ?? frame.metadata.deviceWidth; const delivered: Promise = encoder .evaluate( @@ -1470,26 +1480,23 @@ const make = Effect.gen(function* () { (globalThis as unknown as EncoderWindow).__t3Recorder.frame(data, width), [frame.data, cssWidth] as const, ) - .then(async (accepted) => { - if (!accepted) await opened.send("Page.stopScreencast"); + .then((accepted) => { + // Draining encoded frames must not wait for a stalled screencast session. + if (!accepted && recording.session === opened) + void opened.send("Page.stopScreencast").catch(constVoid); }) .catch(constVoid) .finally(() => { framesInFlight.delete(delivered); - void opened - .send("Page.screencastFrameAck", { sessionId: frame.sessionId }) - .catch(constVoid); + if (recording.session === opened) + void opened + .send("Page.screencastFrameAck", { sessionId: frame.sessionId }) + .catch(constVoid); }); framesInFlight.add(delivered); }); await opened.send("Page.startScreencast", RECORDING_SCREENCAST); if (tab.closing) throw new Error("The tab closed while the recording started."); - const recording: Recording = { - encoder, - session: opened, - startedAt: new Date().toISOString(), - framesInFlight, - }; tab.recording = recording; return recording; } catch (cause) { @@ -1505,27 +1512,65 @@ const make = Effect.gen(function* () { }; // Scaled captures repaint every screencast; pause them to avoid leaking that frame. - const withScreencastsPaused = (tab: ServerTab, capture: () => Promise): Promise => - withCaptureLock(tab, async () => { + const withScreencastsPaused = (tab: ServerTab, capture: () => Promise): Promise => { + let expired = false; + let capturing = false; + let recording: Recording | null = null; + const endCapture = () => { + if (!capturing) return; + capturing = false; + tab.capturing -= 1; + }; + const timeout = new ServerBrowserPage.ServerBrowserOperationError( + "PreviewAutomationTimeoutError", + "The browser did not finish pausing, capturing, and resuming its screencasts in time.", + ); + let timer: ReturnType | undefined; + const deadline = new Promise((_, reject) => { + timer = setTimeout(() => { + expired = true; + endCapture(); + // Retire the sessions: a late stop must never pause a recovered viewer. + for (const viewer of tab.viewers) viewer.disconnect(); + // Detach itself can stall; retire the session while keeping the encoded frames. + const session = recording?.session; + if (recording) recording.session = null; + void session?.detach().catch(constVoid); + reject(timeout); + }, SNAPSHOT_TIMEOUT_MS); + }); + const locked = withCaptureLock(tab, () => { + // The deadline also covers waiting for a viewer's still capture. + if (expired) throw timeout; + capturing = true; tab.capturing += 1; - const recording = tab.recording; - try { - await Promise.all([ - ...[...tab.viewers].map((viewer) => viewer.pause()), - recording?.session.send("Page.stopScreencast").catch(constVoid), - ]); - return await capture(); - } finally { - tab.capturing -= 1; - - await Promise.all([ - ...[...tab.viewers].map((viewer) => viewer.resume()), - recording && tab.recording === recording - ? recording.session.send("Page.startScreencast", RECORDING_SCREENCAST).catch(constVoid) - : undefined, - ]); - } + recording = tab.recording; + const captured = (async () => { + try { + await Promise.all([ + ...[...tab.viewers].map((viewer) => viewer.pause()), + recording?.session?.send("Page.stopScreencast").catch(constVoid), + ]); + if (expired) throw timeout; + return await capture(); + } finally { + endCapture(); + if (!expired) { + await Promise.all([ + ...[...tab.viewers].map((viewer) => viewer.resume()), + recording && tab.recording === recording + ? recording.session + ?.send("Page.startScreencast", RECORDING_SCREENCAST) + .catch(constVoid) + : undefined, + ]); + } + } + })(); + return Promise.race([captured, deadline]); }); + return Promise.race([locked, deadline]).finally(() => clearTimeout(timer)); + }; const stopRecording = (tab: ServerTab) => withCaptureLock(tab, async () => { @@ -1539,9 +1584,9 @@ const make = Effect.gen(function* () { let mimeType: string | null; const chunks: Array = []; try { - await recording.session.send("Page.stopScreencast").catch(constVoid); + await recording.session?.send("Page.stopScreencast").catch(constVoid); await Promise.all(recording.framesInFlight); - await recording.session.detach().catch(constVoid); + await recording.session?.detach().catch(constVoid); const stopped = await recording.encoder.evaluate(() => (globalThis as unknown as EncoderWindow).__t3Recorder.stop(), ); @@ -2047,24 +2092,29 @@ const make = Effect.gen(function* () { } }); - const releaseViewerInput = async (viewer: ViewerState, session: CDPSession) => { - for (const { key, code } of viewer.pressedKeys.values()) { - await session.send("Input.dispatchKeyEvent", { type: "keyUp", key, code }).catch(constVoid); - } + const releaseViewerInput = (viewer: ViewerState, session: CDPSession) => { + const keys = [...viewer.pressedKeys.values()]; + const buttons = [...viewer.pressedButtons]; viewer.pressedKeys.clear(); - for (const [button, point] of viewer.pressedButtons) { - await session - .send("Input.dispatchMouseEvent", { - type: "mouseReleased", - button, - ...point, - buttons: 0, - clickCount: 1, - }) - .catch(constVoid); - } viewer.pressedButtons.clear(); viewer.requestedSize = null; + // Send every release before awaiting: an abandoned cleanup must not send late input. + return Promise.all([ + ...keys.map(({ key, code }) => + session.send("Input.dispatchKeyEvent", { type: "keyUp", key, code }).catch(constVoid), + ), + ...buttons.map(([button, point]) => + session + .send("Input.dispatchMouseEvent", { + type: "mouseReleased", + button, + ...point, + buttons: 0, + clickCount: 1, + }) + .catch(constVoid), + ), + ]).then(constVoid); }; const dispatchViewerInput = async ( @@ -2212,14 +2262,7 @@ const make = Effect.gen(function* () { try: () => tab.page.context().newCDPSession(tab.page), catch: (cause) => new ServerBrowserLaunchError({ cause }), }), - (session) => - Effect.promise(() => - session - .send("Page.stopScreencast") - .catch(constVoid) - .then(() => session.detach()) - .catch(constVoid), - ), + (session) => Effect.promise(() => session.detach().catch(constVoid)), ); const quality = Math.min(100, Math.max(1, Math.round(input.quality))); // Chromium may drop the final frame during a burst; send a still when it settles. @@ -2227,13 +2270,15 @@ const make = Effect.gen(function* () { let mayHaveDropped = false; let motion = false; const recentFrames: Array = []; + let disconnected = false; let screencastParams = Promise.resolve(); let screencastScale = 1; const startScreencast = (scale: number) => { screencastScale = scale; screencastParams = screencastParams.then(async () => { - if (tab.capturing > 0) return; + if (disconnected || tab.capturing > 0) return; await session.send("Page.stopScreencast").catch(constVoid); + if (disconnected) return; await session .send("Page.startScreencast", { format: "jpeg", @@ -2256,7 +2301,12 @@ const make = Effect.gen(function* () { if (next._tag === "frame") runFork(next.ack); // State a stalled viewer cannot miss replaces its backlog. It runs // synchronously so an older replacement can never land after a newer one. - else if (next._tag === "gone" || next._tag === "control" || next._tag === "fileChooser") { + else if ( + next._tag === "gone" || + next._tag === "reconnect" || + next._tag === "control" || + next._tag === "fileChooser" + ) { const dropped = Effect.runSyncExit(Queue.clear(output)); if (dropped._tag === "Failure") return; Queue.offerUnsafe(output, next); @@ -2268,11 +2318,36 @@ const make = Effect.gen(function* () { }, pause: () => { screencastParams = screencastParams.then(() => - session.send("Page.stopScreencast").then(constVoid, constVoid), + disconnected + ? undefined + : session.send("Page.stopScreencast").then(constVoid, constVoid), ); return screencastParams; }, resume: () => startScreencast(screencastScale), + disconnect: () => { + disconnected = true; + tab.viewers.delete(viewer); + viewer.push({ _tag: "reconnect" }); + void session.detach().catch(constVoid); + runFork( + Effect.promise(async () => { + const releasing = tab.control.disconnect(viewer.id, () => { + // The viewer session is stalled; release held input through the tab instead. + let timer: ReturnType | undefined; + const deadline = new Promise((resolve) => { + timer = setTimeout(resolve, VIEWER_INPUT_RELEASE_TIMEOUT_MS); + }); + return Promise.race([releaseViewerInput(viewer, tab.cdp), deadline]).finally(() => + clearTimeout(timer), + ); + }); + broadcastControl(tab); + reportLiveTabs(); + await releasing; + }), + ); + }, scrolledAt: 0, requestedSize: null, }; @@ -2283,6 +2358,7 @@ const make = Effect.gen(function* () { }), () => Effect.promise(async () => { + disconnected = true; await tab.control.disconnect(viewer.id, () => releaseViewerInput(viewer, session)); tab.viewers.delete(viewer); broadcastControl(tab); @@ -2296,14 +2372,26 @@ const make = Effect.gen(function* () { pushFileChooser(tab); // Full scale: a scaled capture would flash in every other viewer. const pushStill = async () => { - const data = await withCaptureLock(tab, () => - ServerBrowserPage.captureViewport(tab.page, session, { - format: "jpeg", - quality, - scale: 1, - }), - ).catch(() => null); - if (data) + if (disconnected) return; + const data = await withCaptureLock(tab, () => { + if (disconnected) return Promise.resolve(null); + let timer: ReturnType | undefined; + const deadline = new Promise((resolve) => { + timer = setTimeout(() => { + viewer.disconnect(); + resolve(null); + }, SNAPSHOT_TIMEOUT_MS); + }); + return Promise.race([ + ServerBrowserPage.captureViewport(tab.page, session, { + format: "jpeg", + quality, + scale: 1, + }), + deadline, + ]).finally(() => clearTimeout(timer)); + }).catch(() => null); + if (data && !disconnected) viewer.push({ _tag: "frame", data: Buffer.from(data, "base64"), ack: Effect.void }); }; let settleTimer: ReturnType | null = null; @@ -2314,6 +2402,7 @@ const make = Effect.gen(function* () { ); let screencastStarted = false; session.on("Page.screencastFrame", (frame) => { + if (disconnected) return; screencastStarted = true; if (framesInFlight > 0) mayHaveDropped = true; framesInFlight += 1; @@ -2366,7 +2455,7 @@ const make = Effect.gen(function* () { output, input: (raw: unknown) => Effect.promise(async () => { - if (!viewer.canOperate) return; + if (disconnected || !viewer.canOperate) return; const message = asRecord(raw); if (!message) return; try { @@ -2469,7 +2558,7 @@ const make = Effect.gen(function* () { ), ), ); - // The broker disconnects timed-out hosts, including slow first installs. Reconnect. + // The stream ends if the broker replaces or drops this host. Reconnect. yield* hostSession.pipe( Effect.exit, Effect.andThen(Effect.sleep(HOST_RECONNECT_DELAY)),