diff --git a/apps/server/src/mcp/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index 3505e4e58633b..a3beb510ea8cf 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -244,10 +244,16 @@ it.effect("tells the agent how to fall back when no desktop app can run the snap ); it.effect.each([ - { mode: "default", input: {}, images: true }, - { mode: "explicit image", input: { includeImage: true }, images: true }, - { mode: "text only", input: { includeImage: false }, images: false }, -])("returns fresh $mode snapshots on repeated MCP calls", ({ input, images }) => + { mode: "default", input: {}, images: true, legacyHost: false }, + { mode: "explicit image", input: { includeImage: true }, images: true, legacyHost: false }, + { mode: "text only", input: { includeImage: false }, images: false, legacyHost: false }, + { + mode: "text only from an older host", + input: { includeImage: false }, + images: false, + legacyHost: true, + }, +])("returns fresh $mode snapshots on repeated MCP calls", ({ input, images, legacyHost }) => Effect.scoped( Effect.gen(function* () { const server = yield* McpServer.McpServer; @@ -287,7 +293,8 @@ it.effect.each([ tabId: alternateTabId, threadId, }); - expect(event.request.input).toEqual({}); + const capture = requests > 6 || images; + expect(event.request.input).toEqual(capture ? {} : { includeImage: false }); return broker.respond({ clientId: "mcp-image-option-client", connectionId: event.connectionId, @@ -296,7 +303,8 @@ it.effect.each([ result: { ...page, title: `Snapshot ${requests}`, - screenshot: { ...screenshot, data: png }, + // Older hosts ignore the capture preference and still return a full PNG. + ...(capture || legacyHost ? { screenshot: { ...screenshot, data: png } } : {}), }, }); }).pipe(Effect.forkScoped); @@ -312,7 +320,11 @@ it.effect.each([ Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), Effect.provideService(McpSchema.McpServerClient, client), ); - const metadata = { ...page, title: `Snapshot ${call}`, screenshot }; + const metadata = { + ...page, + title: `Snapshot ${call}`, + ...(images || legacyHost ? { screenshot } : {}), + }; const { accessibilityTree: _tree, ...boundedMetadata } = metadata; expect(snapshot.isError).toBe(false); expect(snapshot.structuredContent).toEqual({ @@ -443,6 +455,8 @@ it.effect("saves the snapshot PNG on request and reports its path", () => const [only, ...others] = pathOnly.content; expect(others).toEqual([]); expect(only?.type === "text" ? decodeJsonText(only.text) : null).toEqual(saved); + // Both saved variants request a full capture even when the caller hides the image. + expect(inputs).toEqual([{}, {}, {}]); }), ).pipe(Effect.provide(layerTest)), ); @@ -854,6 +868,9 @@ it.effect("registers annotated tools and preserves authenticated request context expect(snapshotTool?.tool.annotations?.readOnlyHint).toBe(true); expect(snapshotTool?.tool.annotations?.idempotentHint).toBe(true); expect(snapshotTool?.tool.annotations?.openWorldHint).toBe(true); + // This formatter advertises no required PNG output schema to MCP clients. + // PreviewAutomationSnapshot validates the host result inside this server. + expect(snapshotTool?.tool.outputSchema).toBeUndefined(); const clickTool = server.tools.find(({ tool }) => tool.name === "preview_click"); expect(clickTool?.tool.annotations?.readOnlyHint).toBe(false); diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index d1904f1dbe0d3..6c99299258c2e 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -492,7 +492,7 @@ const registerPreviewSnapshot = Effect.fn("McpHttpServer.registerPreviewSnapshot Effect.gen(function* () { const snapshot = encodedResult as SnapshotMetadata & { readonly url: string; - readonly screenshot: { + readonly screenshot?: { readonly mimeType: "image/png"; readonly data: string; readonly width: number; @@ -500,9 +500,14 @@ const registerPreviewSnapshot = Effect.fn("McpHttpServer.registerPreviewSnapshot }; }; const { screenshot, ...page } = snapshot; - const png = new Uint8Array(Buffer.from(screenshot.data, "base64")); + const png = + screenshot === undefined + ? undefined + : new Uint8Array(Buffer.from(screenshot.data, "base64")); const screenshotPath = - payload?.save === true ? yield* saveScreenshot(snapshot.url, png) : undefined; + payload?.save === true && png !== undefined + ? yield* saveScreenshot(snapshot.url, png) + : undefined; if (screenshotPath !== undefined && payload?.includeImage === false) { // The agent only wants a file to show the user. The url keeps the site icon on the tool row. const saved = { @@ -517,11 +522,15 @@ const registerPreviewSnapshot = Effect.fn("McpHttpServer.registerPreviewSnapshot } const metadata = { ...page, - screenshot: { - mimeType: screenshot.mimeType, - width: screenshot.width, - height: screenshot.height, - }, + ...(screenshot === undefined + ? {} + : { + screenshot: { + mimeType: screenshot.mimeType, + width: screenshot.width, + height: screenshot.height, + }, + }), ...(screenshotPath === undefined ? {} : { screenshotPath }), }; const bounded = boundSnapshotMetadata(metadata); @@ -548,7 +557,9 @@ const registerPreviewSnapshot = Effect.fn("McpHttpServer.registerPreviewSnapshot text: `Snapshot text was bounded. Omitted: ${bounded.omitted.join("; ")}.`, }, ]), - ...(payload?.includeImage === false + ...(payload?.includeImage === false || + screenshot === undefined || + png === undefined ? [] : [{ type: "image" as const, data: png, mimeType: screenshot.mimeType }]), ], diff --git a/apps/server/src/mcp/PreviewAutomationBroker.test.ts b/apps/server/src/mcp/PreviewAutomationBroker.test.ts index c8b02aa549e34..0a69322aa7bc6 100644 --- a/apps/server/src/mcp/PreviewAutomationBroker.test.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.test.ts @@ -7,6 +7,7 @@ import { PreviewAutomationMalformedResponseError, PreviewAutomationNoAvailableHostError, PreviewAutomationTargetNotEditableError, + PreviewAutomationTimeoutError, PreviewTabId, ProviderInstanceId, ThreadId, @@ -1428,6 +1429,330 @@ it.effect("rejects a routed action when its generation is evicted before deliver ), ); +it.effect.each(["snapshot", "evaluate", "navigate"] as const)( + "retains an opted-in host, pending open, and session tabs after a %s timeout", + (operation) => + Effect.scoped( + Effect.gen(function* () { + const broker = yield* makeBroker; + const connected = yield* Deferred.make(); + const timedOutReceived = yield* Deferred.make(); + const openReceived = yield* Deferred.make(); + const firstTabId = PreviewTabId.make("tab-first-session"); + const secondTabId = PreviewTabId.make("tab-second-session"); + const openedTabId = PreviewTabId.make("tab-opened-second-session"); + const lateTabId = PreviewTabId.make("tab-late-response"); + const otherScope = { + ...scope, + thread: { + ...scope.thread, + threadId: ThreadId.make("thread-2"), + providerSessionId: "provider-session-2", + }, + }; + const routedRequests: RoutedRequest[] = []; + // The policy is a registration option, independent of the client id or preference. + const events = yield* broker.connect(makeHost(), { disconnectOnTimeout: false }); + const consumer = yield* Stream.runForEach(events, (event) => { + if (event.type === "connected") return Deferred.succeed(connected, undefined); + const request = { ...event.request, connectionId: event.connectionId }; + routedRequests.push(request); + if (request.operation === operation) return Deferred.succeed(timedOutReceived, request); + if (request.operation === "open") return Deferred.succeed(openReceived, request); + return broker.respond({ + clientId: "client-1", + connectionId: request.connectionId, + requestId: request.requestId, + ok: true, + result: { host: "client-1", targetTabId: request.tabId }, + }); + }).pipe(Effect.forkScoped); + yield* Deferred.await(connected); + yield* broker.invoke({ scope, operation: "status", input: {}, tabId: firstTabId }); + yield* broker.invoke({ + scope: otherScope, + operation: "status", + input: {}, + tabId: secondTabId, + }); + + const timedOut = yield* broker + .invoke({ scope, operation, input: {}, timeoutMs: 1_000 }) + .pipe(Effect.flip, Effect.forkScoped); + const timedOutRequest = yield* Deferred.await(timedOutReceived); + const opening = yield* broker + .invoke({ scope: otherScope, operation: "open", input: {}, timeoutMs: 10_000 }) + .pipe(Effect.forkScoped); + const openRequest = yield* Deferred.await(openReceived); + + // A new host wins new sessions, exposing any accidental loss of existing leases. + const otherConnected = yield* Deferred.make(); + const otherEvents = yield* broker.connect(makeHost({ clientId: "client-newer" })); + yield* Stream.runForEach(otherEvents, (event) => + event.type === "connected" + ? Deferred.succeed(otherConnected, undefined) + : broker.respond({ + clientId: "client-newer", + connectionId: event.connectionId, + requestId: event.request.requestId, + ok: true, + result: { host: "client-newer" }, + }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(otherConnected); + expect( + yield* broker.invoke({ + scope: { + ...scope, + thread: { ...scope.thread, providerSessionId: "provider-session-new" }, + }, + operation: "status", + input: {}, + }), + ).toEqual({ host: "client-newer" }); + + yield* TestClock.adjust(1_000); + const error = yield* Fiber.join(timedOut); + expect(error).toBeInstanceOf(PreviewAutomationTimeoutError); + expect(error).toMatchObject({ + operation, + timeoutMs: 1_000, + requestId: timedOutRequest.requestId, + clientId: "client-1", + connectionId: timedOutRequest.connectionId, + threadId: scope.thread.threadId, + providerSessionId: scope.thread.providerSessionId, + tabId: firstTabId, + }); + expect(opening.pollUnsafe()).toBeUndefined(); + expect(consumer.pollUnsafe()).toBeUndefined(); + + let lateErrorRead = false; + yield* broker.respond({ + clientId: "client-1", + connectionId: timedOutRequest.connectionId, + requestId: timedOutRequest.requestId, + ok: false, + error: { + _tag: "PreviewAutomationExecutionError", + get message() { + lateErrorRead = true; + return "Late operation failed"; + }, + }, + }); + expect(lateErrorRead).toBe(false); + yield* broker.respond({ + clientId: "client-1", + connectionId: timedOutRequest.connectionId, + requestId: timedOutRequest.requestId, + ok: true, + result: { tabId: lateTabId }, + }); + expect(yield* broker.invoke({ scope, operation: "status", input: {} })).toEqual({ + host: "client-1", + targetTabId: firstTabId, + }); + expect(yield* broker.invoke({ scope: otherScope, operation: "status", input: {} })).toEqual( + { host: "client-1", targetTabId: secondTabId }, + ); + expect(opening.pollUnsafe()).toBeUndefined(); + expect(openRequest.connectionId).toBe(timedOutRequest.connectionId); + expect(openRequest.agentSessionId).not.toBe(timedOutRequest.agentSessionId); + + yield* broker.respond({ + clientId: "client-1", + connectionId: openRequest.connectionId, + requestId: openRequest.requestId, + ok: true, + result: { tabId: openedTabId }, + }); + expect(yield* Fiber.join(opening)).toEqual({ tabId: openedTabId }); + expect(yield* broker.invoke({ scope: otherScope, operation: "status", input: {} })).toEqual( + { host: "client-1", targetTabId: openedTabId }, + ); + expect(routedRequests.filter((request) => request.operation === operation)).toEqual([ + timedOutRequest, + ]); + expect(routedRequests.filter((request) => request.operation === "open")).toEqual([ + openRequest, + ]); + expect(consumer.pollUnsafe()).toBeUndefined(); + }), + ), +); + +it.effect.each([ + { label: "default", clientId: "client-1", options: undefined }, + { + label: "explicit eviction", + clientId: "client-1", + options: { disconnectOnTimeout: true }, + }, + { + label: "preferred server-browser id without opt-in", + clientId: SERVER_BROWSER_AUTOMATION_CLIENT_ID, + options: { preferred: true }, + }, +])("evicts an unanswered host and its unrelated open ($label)", ({ clientId, options }) => + Effect.scoped( + Effect.gen(function* () { + const broker = yield* makeBroker; + const connected = yield* Deferred.make(); + const snapshotReceived = yield* Deferred.make(); + const openReceived = yield* Deferred.make(); + const operations: string[] = []; + const consumer = yield* Stream.runForEach( + yield* broker.connect(makeHost({ clientId }), options), + (event) => { + if (event.type === "connected") return Deferred.succeed(connected, undefined); + operations.push(event.request.operation); + const request = { ...event.request, connectionId: event.connectionId }; + return Deferred.succeed( + event.request.operation === "open" ? openReceived : snapshotReceived, + request, + ); + }, + ).pipe(Effect.forkScoped); + yield* Deferred.await(connected); + const snapshot = yield* broker + .invoke({ scope, operation: "snapshot", input: {}, timeoutMs: 1_000 }) + .pipe(Effect.flip, Effect.forkScoped); + const snapshotRequest = yield* Deferred.await(snapshotReceived); + const opening = yield* broker + .invoke({ + scope: { + ...scope, + thread: { ...scope.thread, providerSessionId: "provider-session-other" }, + }, + operation: "open", + input: {}, + timeoutMs: 10_000, + }) + .pipe(Effect.flip, Effect.forkScoped); + const openRequest = yield* Deferred.await(openReceived); + yield* TestClock.adjust(1_000); + + expect(yield* Fiber.join(snapshot)).toBeInstanceOf(PreviewAutomationTimeoutError); + expect(yield* Fiber.join(opening)).toMatchObject({ + _tag: "PreviewAutomationClientDisconnectedError", + operation: "open", + requestId: openRequest.requestId, + connectionId: snapshotRequest.connectionId, + }); + expect(Exit.isSuccess(yield* Fiber.await(consumer))).toBe(true); + expect( + yield* broker.invoke({ scope, operation: "status", input: {} }).pipe(Effect.flip), + ).toBeInstanceOf(PreviewAutomationNoAvailableHostError); + yield* broker.respond({ + clientId, + connectionId: snapshotRequest.connectionId, + requestId: snapshotRequest.requestId, + ok: true, + result: "late", + }); + expect(operations).toEqual(["snapshot", "open"]); + }), + ), +); + +it.effect.each([true, false])( + "cleans up an interrupted invocation without eviction or replay (before delivery: %s)", + (beforeDelivery) => + Effect.scoped( + Effect.gen(function* () { + const broker = yield* makeBroker; + const connected = yield* Deferred.make(); + const received = yield* Deferred.make(); + const routed = yield* Deferred.make(); + const tabId = PreviewTabId.make("tab-before-interruption"); + const requests: RoutedRequest[] = []; + const events = yield* broker.connect(makeHost(), { disconnectOnTimeout: false }); + const consumer = yield* Stream.runForEach(events, (event) => { + if (event.type === "connected") return Deferred.succeed(connected, event.connectionId); + const request = { ...event.request, connectionId: event.connectionId }; + requests.push(request); + return request.operation === "evaluate" + ? Deferred.succeed(received, request) + : broker.respond({ + clientId: "client-1", + connectionId: request.connectionId, + requestId: request.requestId, + ok: true, + result: { requestId: request.requestId, targetTabId: request.tabId }, + }); + }).pipe(Effect.forkScoped); + const connectionId = yield* Deferred.await(connected); + expect(yield* broker.invoke({ scope, operation: "status", input: {}, tabId })).toEqual({ + requestId: "preview-0", + targetTabId: tabId, + }); + + const tasks: Array<() => void> = []; + let paused = false; + const dispatcher: Scheduler.SchedulerDispatcher = { + scheduleTask: (task) => tasks.push(task), + flush: () => { + let task: (() => void) | undefined; + while ((task = tasks.shift()) !== undefined) task(); + }, + }; + const scheduler: Scheduler.Scheduler = { + executionMode: "async", + makeDispatcher: () => dispatcher, + shouldYield: () => paused, + }; + const invocation = yield* broker + .invoke({ + scope, + operation: "evaluate", + input: {}, + timeoutMs: 1_000, + onTargetTab: () => { + if (!beforeDelivery) return; + paused = true; + Deferred.doneUnsafe(routed, Effect.void); + }, + }) + .pipe(Effect.provideService(Scheduler.Scheduler, scheduler), Effect.forkScoped); + const request = beforeDelivery + ? yield* Deferred.await(routed).pipe(Effect.as({ requestId: "preview-1", connectionId })) + : yield* Deferred.await(received); + // Signal interruption while its dispatcher may be paused, then let cleanup run. + yield* Effect.sync(() => invocation.interruptUnsafe()); + paused = false; + dispatcher.flush(); + expect(Exit.hasInterrupts(yield* Fiber.await(invocation))).toBe(true); + + // Ignoring a retired response includes ignoring its error payload entirely. + let lateErrorRead = false; + yield* broker.respond({ + clientId: "client-1", + connectionId: request.connectionId, + requestId: request.requestId, + ok: false, + error: { + _tag: "PreviewAutomationExecutionError", + get message() { + lateErrorRead = true; + return "Late evaluation failed"; + }, + }, + }); + expect(lateErrorRead).toBe(false); + yield* TestClock.adjust(1_000); + expect(yield* broker.invoke({ scope, operation: "status", input: {} })).toEqual({ + requestId: "preview-2", + targetTabId: tabId, + }); + expect(requests.map((request) => request.operation)).toEqual( + beforeDelivery ? ["status", "status"] : ["status", "evaluate", "status"], + ); + expect(consumer.pollUnsafe()).toBeUndefined(); + }), + ), +); + it.effect("keeps a host that responds with an operation timeout", () => Effect.scoped( Effect.gen(function* () { @@ -1494,3 +1819,247 @@ it.effect("keeps the host connected when a background status read times out", () }), ), ); + +// Leave transport delivery separate from execution so these tests can hold +// a host event without starting a fresh timeout at dequeue. +const makeGuardedRequest = Effect.gen(function* () { + const broker = yield* makeBroker; + const connected = yield* Deferred.make(); + const received = yield* Deferred.make(); + const consumer = yield* Stream.runForEach( + yield* broker.connect(makeHost(), { disconnectOnTimeout: false }), + (event) => + event.type === "connected" + ? Deferred.succeed(connected, undefined) + : Deferred.succeed(received, { ...event.request, connectionId: event.connectionId }), + ).pipe(Effect.forkScoped); + yield* Deferred.await(connected); + const caller = yield* broker + .invoke({ scope, operation: "click", input: {}, timeoutMs: 1_000 }) + .pipe(Effect.forkScoped); + const request = yield* Deferred.await(received); + const identity = { + clientId: "client-1", + connectionId: request.connectionId, + requestId: request.requestId, + }; + return { broker, caller, consumer, identity }; +}); + +it.effect.each(["timeout", "interrupt"] as const)( + "skips host events held until after caller %s and ignores their late responses", + (retirement) => + Effect.scoped( + Effect.gen(function* () { + const { broker, caller, identity } = yield* makeGuardedRequest; + if (retirement === "timeout") { + yield* TestClock.adjust(1_000); + expect(yield* Fiber.join(caller).pipe(Effect.flip)).toBeInstanceOf( + PreviewAutomationTimeoutError, + ); + } else { + yield* Fiber.interrupt(caller); + expect(Exit.hasInterrupts(yield* Fiber.await(caller))).toBe(true); + } + let executions = 0; + yield* broker.runRequest(identity, () => + Effect.sync(() => executions++).pipe(Effect.asVoid), + ); + expect(executions).toBe(0); + let lateErrorRead = false; + yield* broker.respond({ + ...identity, + ok: false, + error: { + _tag: "PreviewAutomationExecutionError", + get message() { + lateErrorRead = true; + return "Late action failed"; + }, + }, + }); + yield* broker.respond({ ...identity, ok: true, result: undefined }); + expect(lateErrorRead).toBe(false); + yield* broker.runRequest(identity, () => + Effect.sync(() => executions++).pipe(Effect.asVoid), + ); + expect(executions).toBe(0); + }), + ), +); + +it.effect("passes only the original request's remaining budget after host dispatch delay", () => + Effect.scoped( + Effect.gen(function* () { + const { broker, caller, identity } = yield* makeGuardedRequest; + yield* TestClock.adjust(400); + const budgetReceived = yield* Deferred.make(); + const guarded = yield* broker + .runRequest(identity, (remainingTimeoutMs) => + Deferred.succeed(budgetReceived, remainingTimeoutMs).pipe(Effect.andThen(Effect.never)), + ) + .pipe(Effect.forkScoped); + expect(yield* Deferred.await(budgetReceived)).toBe(600); + yield* TestClock.adjust(600); + expect(yield* Fiber.join(caller).pipe(Effect.flip)).toBeInstanceOf( + PreviewAutomationTimeoutError, + ); + expect(Exit.isSuccess(yield* Fiber.await(guarded))).toBe(true); + }), + ), +); + +it.effect.each(["timeout", "interrupt", "success", "error", "disconnect"] as const)( + "aborts active host effects and revokes queued actions on caller %s", + (retirement) => + Effect.scoped( + Effect.gen(function* () { + const { broker, caller, consumer, identity } = yield* makeGuardedRequest; + const started = yield* Deferred.make(); + const aborted = yield* Deferred.make(); + const queuedActions: Array<() => void> = []; + let executions = 0; + const guarded = yield* broker + .runRequest(identity, () => + Effect.tryPromise({ + try: (signal) => + new Promise(() => { + // The host's dispatcher starts this only after it acquires its turn. + queuedActions.push(() => { + if (!signal.aborted) executions++; + }); + signal.addEventListener( + "abort", + () => Deferred.doneUnsafe(aborted, Effect.void), + { + once: true, + }, + ); + Deferred.doneUnsafe(started, Effect.succeed(signal)); + }), + catch: () => ({ _tag: "GuardedHostError" as const }), + }), + ) + .pipe(Effect.forkScoped); + const signal = yield* Deferred.await(started); + if (retirement === "timeout") { + yield* TestClock.adjust(1_000); + expect(yield* Fiber.join(caller).pipe(Effect.flip)).toBeInstanceOf( + PreviewAutomationTimeoutError, + ); + } else if (retirement === "interrupt") { + yield* Fiber.interrupt(caller); + expect(Exit.hasInterrupts(yield* Fiber.await(caller))).toBe(true); + } else if (retirement === "disconnect") { + yield* Fiber.interrupt(consumer); + expect(yield* Fiber.join(caller).pipe(Effect.flip)).toBeInstanceOf( + PreviewAutomationClientDisconnectedError, + ); + } else if (retirement === "success") { + yield* broker.respond({ ...identity, ok: true, result: undefined }); + yield* Fiber.join(caller); + } else { + yield* broker.respond({ + ...identity, + ok: false, + error: { _tag: "PreviewAutomationExecutionError", message: "Action failed" }, + }); + expect(yield* Fiber.join(caller).pipe(Effect.flip)).toMatchObject({ + _tag: "PreviewAutomationExecutionError", + }); + } + yield* Deferred.await(aborted); + expect(Exit.isSuccess(yield* Fiber.await(guarded))).toBe(true); + expect(signal.aborted).toBe(true); + queuedActions.forEach((action) => action()); + expect(executions).toBe(0); + yield* broker.respond({ ...identity, ok: true, result: undefined }); + yield* broker.runRequest(identity, () => + Effect.sync(() => executions++).pipe(Effect.asVoid), + ); + expect(executions).toBe(0); + }), + ), +); + +it.effect("rejects wrong client and connection guards without retiring the valid invocation", () => + Effect.scoped( + Effect.gen(function* () { + const { broker, caller, identity } = yield* makeGuardedRequest; + const wrongIdentities = [ + { ...identity, clientId: "client-other" }, + { ...identity, connectionId: "retired-connection" }, + { ...identity, requestId: "retired-request" }, + ]; + let executions = 0; + for (const wrongIdentity of wrongIdentities) { + yield* broker.runRequest(wrongIdentity, () => + Effect.sync(() => executions++).pipe(Effect.asVoid), + ); + yield* broker.respond({ ...wrongIdentity, ok: true, result: "wrong response" }); + } + expect(executions).toBe(0); + expect(caller.pollUnsafe()).toBeUndefined(); + yield* broker.runRequest(identity, () => + Effect.sync(() => executions++).pipe( + Effect.andThen(broker.respond({ ...identity, ok: true, result: undefined })), + ), + ); + yield* Fiber.join(caller); + expect(executions).toBe(1); + yield* broker.runRequest(identity, () => Effect.sync(() => executions++).pipe(Effect.asVoid)); + expect(executions).toBe(1); + }), + ), +); + +it.effect("preserves the guarded host effect's error channel", () => + Effect.scoped( + Effect.gen(function* () { + const { broker, caller, identity } = yield* makeGuardedRequest; + const failure = { _tag: "GuardedHostError" as const }; + expect(yield* broker.runRequest(identity, () => Effect.fail(failure)).pipe(Effect.flip)).toBe( + failure, + ); + yield* Fiber.interrupt(caller); + }), + ), +); + +it.effect("guards host work directly at delivery and permits its response", () => + Effect.scoped( + Effect.gen(function* () { + const broker = yield* makeBroker; + const connected = yield* Deferred.make(); + let executions = 0; + yield* Stream.runForEach( + yield* broker.connect(makeHost(), { disconnectOnTimeout: false }), + (event) => { + if (event.type === "connected") return Deferred.succeed(connected, undefined); + const identity = { + clientId: "client-1", + connectionId: event.connectionId, + requestId: event.request.requestId, + }; + return broker.runRequest(identity, (remainingTimeoutMs) => + Effect.gen(function* () { + expect(remainingTimeoutMs).toBe(1_000); + executions++; + yield* broker.respond({ ...identity, ok: true, result: "executed" }); + }), + ); + }, + ).pipe(Effect.forkScoped); + yield* Deferred.await(connected); + expect( + yield* broker.invoke({ + scope, + operation: "click", + input: {}, + timeoutMs: 1_000, + }), + ).toBe("executed"); + expect(executions).toBe(1); + }), + ), +); diff --git a/apps/server/src/mcp/PreviewAutomationBroker.ts b/apps/server/src/mcp/PreviewAutomationBroker.ts index 7be02956eb371..8ea5a9c2cf866 100644 --- a/apps/server/src/mcp/PreviewAutomationBroker.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.ts @@ -27,6 +27,7 @@ import { type PreviewAutomationResponse, type PreviewAutomationStreamEvent, } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import type * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; @@ -61,6 +62,12 @@ export interface PreviewAutomationConnectOptions { * browsing when every desktop disconnects. */ readonly preferred?: boolean; + /** + * Evict an unanswered foreground request's host (default: true). The trusted + * in-process browser disables this so one slow operation cannot disconnect + * its shared host and abort work in other sessions. + */ + readonly disconnectOnTimeout?: boolean; } export class PreviewAutomationBroker extends Context.Service< @@ -74,6 +81,15 @@ export class PreviewAutomationBroker extends Context.Service< readonly respond: ( response: PreviewAutomationResponse, ) => Effect.Effect; + /** Run in-process host work only while its invocation is still live. */ + readonly runRequest: ( + request: { + readonly clientId: string; + readonly connectionId: string; + readonly requestId: string; + }, + execute: (remainingTimeoutMs: number) => Effect.Effect, + ) => Effect.Effect; readonly invoke: ( request: PreviewAutomationInvokeInput, ) => Effect.Effect; @@ -87,6 +103,7 @@ interface ClientConnection { readonly supportedOperations: ReadonlySet; readonly focused: boolean; readonly preferred: boolean; + readonly disconnectOnTimeout: boolean; readonly liveTabs: NonNullable; readonly focusOrder: number; readonly queue: Queue.Queue; @@ -95,6 +112,8 @@ interface ClientConnection { interface PendingRequest { readonly queue: ClientConnection["queue"]; readonly deferred: Deferred.Deferred; + readonly retired: Deferred.Deferred; + readonly deadlineMs?: number; readonly context: PreviewAutomationRequestErrorContext; } @@ -410,6 +429,7 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() { supportedOperations: new Set(host.supportedOperations ?? PREVIEW_AUTOMATION_V1_OPERATIONS), focused: false, preferred: options?.preferred ?? false, + disconnectOnTimeout: options?.disconnectOnTimeout ?? true, liveTabs: [], focusOrder: 0, queue, @@ -503,11 +523,50 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() { } }); + const runRequest: PreviewAutomationBroker["Service"]["runRequest"] = Effect.fn( + "PreviewAutomationBroker.runRequest", + )(function* ( + request: { + readonly clientId: string; + readonly connectionId: string; + readonly requestId: string; + }, + execute: (remainingTimeoutMs: number) => Effect.Effect, + ): Effect.fn.Return { + // Queue delivery can wake the host before dispatch commits the deadline. + // Read under the same lock as dispatch so the guard sees the complete entry. + const current = yield* SynchronizedRef.modify(state, (current) => [current, current] as const); + const pending = current.pending.get(request.requestId); + if ( + !pending || + pending.context.clientId !== request.clientId || + pending.context.connectionId !== request.connectionId || + current.clients.get(request.clientId)?.queue !== pending.queue || + pending.deadlineMs === undefined + ) { + return; + } + const deadlineMs = pending.deadlineMs; + // Retirement interrupts host work, while execution failures keep their + // error channel. Interruption aborts the host's native signal so its + // dispatcher can revoke actions that have not started yet. + yield* Effect.raceFirst( + Deferred.await(pending.retired), + Effect.gen(function* () { + const live = yield* SynchronizedRef.modify(state, (current) => [current, current] as const); + const remainingTimeoutMs = deadlineMs - (yield* Clock.currentTimeMillis); + if (live.pending.get(request.requestId) !== pending || remainingTimeoutMs <= 0) return; + yield* execute(remainingTimeoutMs); + }), + ); + }); + const invoke = Effect.fn("PreviewAutomationBroker.invoke")(function* ( input: Parameters[0], ): Effect.fn.Return { const deferred = yield* Deferred.make(); - const route = yield* SynchronizedRef.modify(state, (current) => { + const retired = yield* Deferred.make(); + const acquireRoute = SynchronizedRef.modify(state, (current) => { const assignments = new Map( Array.from(current.assignments).filter(([, assignment]) => { const connection = current.clients.get(assignment.clientId); @@ -595,111 +654,140 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() { ...selectorDiagnostics, }; const pending = new Map(current.pending); - pending.set(requestId, { queue: connection.queue, deferred, context }); + pending.set(requestId, { queue: connection.queue, deferred, retired, context }); return [ { connection, requestId, requestContext: context, requestSequence }, { ...current, assignments, pending, requestSequence: current.requestSequence + 1 }, ] as const; }); - if (!route) { - return yield* new PreviewAutomationNoAvailableHostError({ - operation: input.operation, - environmentId: input.scope.environmentId, - threadId: input.scope.thread.threadId, - providerSessionId: input.scope.thread.providerSessionId, - providerInstanceId: input.scope.thread.providerInstanceId, - }); - } - const { connection, requestId, requestContext, requestSequence } = route; - const { timeoutMs } = requestContext; - input.onTargetTab?.(requestContext.tabId); - const removePending = SynchronizedRef.update(state, (next) => { - if (!next.pending.has(requestId)) return next; - const pending = new Map(next.pending); - pending.delete(requestId); - return { ...next, pending }; - }); - const awaitResponse = Effect.fn("PreviewAutomationBroker.awaitResponse")(function* () { - const offered = yield* SynchronizedRef.modifyEffect(state, (current) => { - // A route can outlive its generation while another request evicts it. - // Serialize the live-generation check and offer with queue closure. - if ( - current.clients.get(connection.clientId)?.queue !== connection.queue || - !current.pending.has(requestId) - ) { - return Effect.succeed([false, current] as const); - } - return Queue.offer(connection.queue, { - type: "request", - connectionId: connection.connectionId, - request: { - requestId, - threadId: input.scope.thread.threadId, - tabId: requestContext.tabId, - tabIdExplicit: input.tabId !== undefined, - agentSessionId: hostAssignmentKey(input.scope), - operation: input.operation, - input: input.input, - timeoutMs, - }, - }).pipe(Effect.map((offered) => [offered, current] as const)); - }); - if (!offered) { - const completion = yield* Deferred.poll(deferred); - if (Option.isSome(completion)) { - return (yield* completion.value) as A; - } - return yield* new PreviewAutomationRequestQueueClosedError(requestContext); - } - const result = yield* Deferred.await(deferred).pipe(Effect.timeoutOption(timeoutMs)); - return yield* Option.match(result, { - onNone: () => - Effect.gen(function* () { - // An unanswered request invalidates this connection. Do not replay - // actions: the client may have applied them before 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) { - yield* disconnect(connection.clientId, connection.queue, true); + // Install cleanup with registration, including interruption before delivery. + return yield* Effect.acquireUseRelease( + acquireRoute, + (route) => + Effect.gen(function* () { + if (!route) { + return yield* new PreviewAutomationNoAvailableHostError({ + operation: input.operation, + environmentId: input.scope.environmentId, + threadId: input.scope.thread.threadId, + providerSessionId: input.scope.thread.providerSessionId, + providerInstanceId: input.scope.thread.providerInstanceId, + }); + } + const { connection, requestId, requestContext, requestSequence } = route; + const { timeoutMs } = requestContext; + input.onTargetTab?.(requestContext.tabId); + const awaitResponse = Effect.fn("PreviewAutomationBroker.awaitResponse")(function* () { + const deadlineMs = yield* SynchronizedRef.modifyEffect(state, (current) => { + // A route can outlive its generation while another request evicts it. + // Serialize the live-generation check and offer with queue closure. + if ( + current.clients.get(connection.clientId)?.queue !== connection.queue || + !current.pending.has(requestId) + ) { + return Effect.succeed([undefined, current] as const); + } + return Effect.gen(function* () { + const deadlineMs = (yield* Clock.currentTimeMillis) + timeoutMs; + const pending = new Map(current.pending); + pending.set(requestId, { + queue: connection.queue, + deferred, + retired, + context: requestContext, + deadlineMs, + }); + const offered = yield* Queue.offer(connection.queue, { + type: "request", + connectionId: connection.connectionId, + request: { + requestId, + threadId: input.scope.thread.threadId, + tabId: requestContext.tabId, + tabIdExplicit: input.tabId !== undefined, + agentSessionId: hostAssignmentKey(input.scope), + operation: input.operation, + input: input.input, + timeoutMs, + }, + }); + return [offered ? deadlineMs : undefined, { ...current, pending }] as const; + }); + }); + if (deadlineMs === undefined) { + const completion = yield* Deferred.poll(deferred); + if (Option.isSome(completion)) { + return (yield* completion.value) as A; + } + return yield* new PreviewAutomationRequestQueueClosedError(requestContext); } - return yield* new PreviewAutomationTimeoutError(requestContext); - }), - onSome: (value) => Effect.succeed(value as A), - }); - }); - const result = yield* awaitResponse().pipe(Effect.ensuring(removePending)); - if (input.updateCurrentTab === false) return result; - const responseTabId = readResultTabId(result); - const resultTabId = responseTabId === undefined ? input.tabId : responseTabId; - if (resultTabId === undefined) return result; - const assignmentKey = hostAssignmentKey(input.scope); - yield* SynchronizedRef.update(state, (current) => { - const assignment = current.assignments.get(assignmentKey); - if ( - !assignment || - assignment.connectionId !== connection.connectionId || - assignment.queue !== connection.queue || - (assignment.tabSequence ?? -1) > requestSequence - ) { - return current; - } - const assignments = new Map(current.assignments); - if (resultTabId === null) { - const { tabId: _tabId, ...withoutTabId } = assignment; - assignments.set(assignmentKey, { ...withoutTabId, tabSequence: requestSequence }); - } else { - assignments.set(assignmentKey, { - ...assignment, - ...(resultTabId === undefined ? {} : { tabId: resultTabId }), - tabSequence: requestSequence, - }); - } - return { ...current, assignments }; - }); - return result; + const remainingTimeoutMs = Math.max(0, deadlineMs - (yield* Clock.currentTimeMillis)); + const result = yield* Deferred.await(deferred).pipe( + Effect.timeoutOption(remainingTimeoutMs), + ); + return yield* Option.match(result, { + onNone: () => + Effect.gen(function* () { + // Remote silence can mean the connection is gone. An in-process + // host stays registered so a slow operation cannot abort other work. + // Never replay actions whose effects may already have been applied. + // 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 (connection.disconnectOnTimeout && input.updateCurrentTab !== false) { + yield* disconnect(connection.clientId, connection.queue, true); + } + return yield* new PreviewAutomationTimeoutError(requestContext); + }), + onSome: (value) => Effect.succeed(value as A), + }); + }); + const result = yield* awaitResponse(); + if (input.updateCurrentTab === false) return result; + const responseTabId = readResultTabId(result); + const resultTabId = responseTabId === undefined ? input.tabId : responseTabId; + if (resultTabId === undefined) return result; + const assignmentKey = hostAssignmentKey(input.scope); + yield* SynchronizedRef.update(state, (current) => { + const assignment = current.assignments.get(assignmentKey); + if ( + !assignment || + assignment.connectionId !== connection.connectionId || + assignment.queue !== connection.queue || + (assignment.tabSequence ?? -1) > requestSequence + ) { + return current; + } + const assignments = new Map(current.assignments); + if (resultTabId === null) { + const { tabId: _tabId, ...withoutTabId } = assignment; + assignments.set(assignmentKey, { ...withoutTabId, tabSequence: requestSequence }); + } else { + assignments.set(assignmentKey, { + ...assignment, + ...(resultTabId === undefined ? {} : { tabId: resultTabId }), + tabSequence: requestSequence, + }); + } + return { ...current, assignments }; + }); + return result; + }), + (route) => + Effect.gen(function* () { + yield* SynchronizedRef.update(state, (current) => { + if (!route || !current.pending.has(route.requestId)) return current; + const pending = new Map(current.pending); + pending.delete(route.requestId); + return { ...current, pending }; + }); + // respond/disconnect may have removed the map entry already. The + // invocation still owns retirement on every exit, including interruption. + yield* Deferred.succeed(retired, undefined); + }), + ); }); - return PreviewAutomationBroker.of({ connect, focusHost, respond, invoke }); + return PreviewAutomationBroker.of({ connect, focusHost, respond, runRequest, invoke }); }).pipe(Effect.withSpan("PreviewAutomationBroker.make")); export const layer = Layer.effect(PreviewAutomationBroker, make); diff --git a/apps/server/src/mcp/toolkits/preview/handlers.ts b/apps/server/src/mcp/toolkits/preview/handlers.ts index 70b20677a502e..850e1ccb65b10 100644 --- a/apps/server/src/mcp/toolkits/preview/handlers.ts +++ b/apps/server/src/mcp/toolkits/preview/handlers.ts @@ -209,9 +209,12 @@ const handlers = { invokeTargeted("setColorScheme", input), ), preview_snapshot: McpToolAccess.readsAsCaller((input) => { - // Output selection and saving are MCP-only; the browser still produces a complete snapshot. - const { includeImage: _includeImage, save: _save, ...operationInput } = input ?? {}; - return invokeTargeted("snapshot", operationInput); + const { includeImage, save, ...operationInput } = input ?? {}; + return invokeTargeted("snapshot", { + ...operationInput, + // Saving still needs a capture even when the response contains only its path. + ...(includeImage === false && save !== true ? { includeImage: false } : {}), + }); }), preview_click: McpToolAccess.actsAsCaller((input) => invokeTargeted("click", input, input.timeoutMs), diff --git a/apps/server/src/mcp/toolkits/preview/tools.ts b/apps/server/src/mcp/toolkits/preview/tools.ts index 9c94a1d446c6d..5acf1416f0e9e 100644 --- a/apps/server/src/mcp/toolkits/preview/tools.ts +++ b/apps/server/src/mcp/toolkits/preview/tools.ts @@ -142,7 +142,7 @@ const PreviewSetAppearanceTool = safeBrowserTool( export const PreviewSnapshotTool = readonlyBrowserTool( Tool.make("preview_snapshot", { description: - "Inspect a page before interacting. Pass tabId to inspect a specific tab; omit it to use this agent session's current tab. Returns page state, semantic elements, diagnostics, action history, and a PNG screenshot. Server snapshots include an accessibilityTree with refs; pass locator=aria-ref= to target one exact element, including inside frames. Refresh refs after navigation, another snapshot, or human takeover. The text is capped near 20 KB and lists what it omitted; use preview_evaluate to read more. Set includeImage=false for text-only output with the same page metadata. Set save=true to also write the PNG to disk and get screenshotPath back; with includeImage=false, save=true returns only the url and screenshotPath. Embed that path in your reply as ![alt](screenshotPath) so the user sees it. This is the only way to show the user a screenshot; the image in the tool result is not saved anywhere.", + "Inspect a page before interacting. Pass tabId to inspect a specific tab; omit it to use this agent session's current tab. Returns page state, semantic elements, diagnostics, action history, and a PNG screenshot. Server snapshots include an accessibilityTree with refs; pass locator=aria-ref= to target one exact element, including inside frames. Refresh refs after navigation, another snapshot, or human takeover. The text is capped near 20 KB and lists what it omitted; use preview_evaluate to read more. Set includeImage=false for text-only output; server tabs skip screenshot capture unless save=true. Set save=true to also write the PNG to disk and get screenshotPath back; with includeImage=false, save=true returns only the url and screenshotPath. Embed that path in your reply as ![alt](screenshotPath) so the user sees it. This is the only way to show the user a screenshot; the image in the tool result is not saved anywhere.", parameters: Schema.Struct({ ...PreviewAutomationTabTargetInput.fields, includeImage: Schema.optional( diff --git a/apps/server/src/preview/ServerBrowser.test.ts b/apps/server/src/preview/ServerBrowser.test.ts index dce0c23b96cf8..a42f36e4543c2 100644 --- a/apps/server/src/preview/ServerBrowser.test.ts +++ b/apps/server/src/preview/ServerBrowser.test.ts @@ -11,6 +11,7 @@ import { type PreviewAutomationStatus, } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; +import * as TestClock from "effect/testing/TestClock"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; @@ -1090,3 +1091,309 @@ it.live("a desktop page the desktop takes back reconnects instead of closing", ( }), ).pipe(Effect.provide(layer)), ); + +it.effect("a stalled snapshot releases its tab and leaves other sessions usable", () => + Effect.scoped( + Effect.gen(function* () { + const { browser, broker, tabId } = yield* ready; + const session = contexts[0]!.sessions[0]!; + const captureStarted = Promise.withResolvers(); + const send = session.send.getMockImplementation()!; + session.send.mockImplementation(async (method, input) => { + if (method === "Page.captureScreenshot") { + captureStarted.resolve(); + return new Promise>(() => {}); + } + return send(method, input); + }); + const snapshot = yield* broker + .invoke({ + scope, + tabId, + operation: "snapshot", + input: {}, + timeoutMs: 1_000, + }) + .pipe(Effect.flip, Effect.forkScoped); + yield* Effect.promise(() => captureStarted.promise); + yield* TestClock.adjust(900); + const failed = yield* Fiber.join(snapshot); + expect(failed._tag).toBe("PreviewAutomationTimeoutError"); + const other = yield* broker.invoke({ + scope: asSession("other-agent"), + operation: "open", + input: { show: false }, + }); + expect(other.available).toBe(true); + expect(other.tabId).not.toBe(tabId); + expect( + yield* broker.invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "document.title" }, + }), + ).toBe("evaluated"); + const viewer = yield* browser.attachViewer(viewerInput(tabId, true)); + yield* viewer.input({ type: "takeControl" }); + yield* viewer.input({ type: "releaseControl" }); + const revealed = yield* broker.invoke({ + scope, + tabId, + operation: "open", + input: { open: true }, + }); + expect(revealed.available).toBe(true); + }), + ).pipe(Effect.provide(layer)), +); + +it.effect("an evaluation that never settles times out without poisoning later work", () => + Effect.scoped( + Effect.gen(function* () { + const { broker, tabId } = yield* ready; + const session = contexts[0]!.sessions[0]!; + const started = Promise.withResolvers(); + const send = session.send.getMockImplementation()!; + session.send.mockImplementation(async (method, input) => { + if ( + method === "Runtime.evaluate" && + (input as { expression?: string }).expression === "pending()" + ) { + started.resolve(); + return new Promise>(() => {}); + } + return send(method, input); + }); + const evaluation = yield* broker + .invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "pending()" }, + timeoutMs: 1_000, + }) + .pipe(Effect.flip, Effect.forkScoped); + yield* Effect.promise(() => started.promise); + yield* TestClock.adjust(900); + expect((yield* Fiber.join(evaluation))._tag).toBe("PreviewAutomationTimeoutError"); + expect( + yield* broker.invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "document.title" }, + }), + ).toBe("evaluated"); + }), + ).pipe(Effect.provide(layer)), +); + +it.live("text-only snapshots stay usable when screenshot capture is unavailable", () => + Effect.scoped( + Effect.gen(function* () { + const { browser, broker, tabId } = yield* ready; + yield* browser.attachViewer(viewerInput(tabId, false)); + const viewerSession = contexts[0]!.sessions.at(-1)!; + viewerSession.send.mockClear(); + const session = contexts[0]!.sessions[0]!; + const send = session.send.getMockImplementation()!; + session.send.mockImplementation(async (method, input) => { + if (method === "Page.captureScreenshot") + throw new Error("Screenshot capture is unavailable"); + return send(method, input); + }); + const result = yield* broker.invoke({ + scope, + tabId, + operation: "snapshot", + input: { includeImage: false }, + }); + expect(result.visibleText).toBe("delete"); + expect(result.accessibilityTree).toContain('button "delete"'); + expect(result.screenshot).toBeUndefined(); + expect(viewerSession.send.mock.calls.map(([method]) => method)).not.toContain( + "Page.stopScreencast", + ); + expect(session.send.mock.calls.some(([method]) => method === "Page.captureScreenshot")).toBe( + false, + ); + }), + ).pipe(Effect.provide(layer)), +); + +it.effect( + "a queued click whose deadline expires is never dispatched after a stalled read releases", + () => + Effect.scoped( + Effect.gen(function* () { + const { broker, tabId } = yield* ready; + const session = contexts[0]!.sessions[0]!; + const started = Promise.withResolvers(); + const send = session.send.getMockImplementation()!; + session.send.mockImplementation(async (method, input) => { + if ( + method === "Runtime.evaluate" && + (input as { expression?: string }).expression === "pending()" + ) { + started.resolve(); + return new Promise>(() => {}); + } + return send(method, input); + }); + const evaluation = yield* broker + .invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "pending()" }, + timeoutMs: 1_000, + }) + .pipe(Effect.flip, Effect.forkScoped); + yield* Effect.promise(() => started.promise); + const click = yield* broker + .invoke({ + scope, + tabId, + operation: "click", + input: { selector: "button" }, + timeoutMs: 200, + }) + .pipe(Effect.flip, Effect.forkScoped); + // Status bypasses the tab queue and acknowledges that the host consumed + // the preceding click before the test advances its execution deadline. + yield* Effect.yieldNow; + yield* broker.invoke({ + scope, + tabId, + operation: "status", + input: {}, + }); + yield* TestClock.adjust(180); + expect((yield* Fiber.join(click))._tag).toBe("PreviewAutomationTimeoutError"); + yield* TestClock.adjust(720); + expect((yield* Fiber.join(evaluation))._tag).toBe("PreviewAutomationTimeoutError"); + expect( + yield* broker.invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "document.title" }, + }), + ).toBe("evaluated"); + expect(contexts[0]!.page.locator).not.toHaveBeenCalled(); + }), + ).pipe(Effect.provide(layer)), +); + +it.effect.each([0, 400])( + "passes the remaining execution budget to CDP after %sms queued", + (queuedMs) => + Effect.scoped( + Effect.gen(function* () { + const { broker, tabId } = yield* ready; + const session = contexts[0]!.sessions[0]!; + const started = Promise.withResolvers(); + const release = Promise.withResolvers(); + yield* Effect.addFinalizer(() => Effect.sync(() => release.resolve())); + const send = session.send.getMockImplementation()!; + session.send.mockImplementation(async (method, input) => { + if ( + method === "Runtime.evaluate" && + (input as { expression?: string }).expression === "blocking()" + ) { + started.resolve(); + await release.promise; + } + return send(method, input); + }); + const first = yield* broker + .invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "blocking()" }, + }) + .pipe(Effect.forkScoped); + yield* Effect.promise(() => started.promise); + const second = yield* broker + .invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "document.title" }, + timeoutMs: 1_000, + }) + .pipe(Effect.forkScoped); + yield* Effect.yieldNow; + yield* broker.invoke({ scope, tabId, operation: "status", input: {} }); + yield* TestClock.adjust(queuedMs); + release.resolve(); + yield* Fiber.join(first); + expect(yield* Fiber.join(second)).toBe("evaluated"); + const evaluation = session.send.mock.calls.find( + ([method, input]) => + method === "Runtime.evaluate" && + (input as { expression?: string }).expression === "document.title", + ); + expect(evaluation).toBeDefined(); + const timeout = (evaluation![1] as { timeout: number }).timeout; + expect(timeout).toBeGreaterThan(0); + expect(timeout).toBeLessThanOrEqual(900 - queuedMs); + }), + ).pipe(Effect.provide(layer)), +); + +it.effect("cancelling a caller revokes its queued click before the tab is released", () => + Effect.scoped( + Effect.gen(function* () { + const { broker, tabId } = yield* ready; + const session = contexts[0]!.sessions[0]!; + const started = Promise.withResolvers(); + const release = Promise.withResolvers(); + yield* Effect.addFinalizer(() => Effect.sync(() => release.resolve())); + const send = session.send.getMockImplementation()!; + session.send.mockImplementation(async (method, input) => { + if ( + method === "Runtime.evaluate" && + (input as { expression?: string }).expression === "blocking()" + ) { + started.resolve(); + await release.promise; + } + return send(method, input); + }); + const evaluation = yield* broker + .invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "blocking()" }, + }) + .pipe(Effect.forkScoped); + yield* Effect.promise(() => started.promise); + const click = yield* broker + .invoke({ + scope, + tabId, + operation: "click", + input: { selector: "button" }, + }) + .pipe(Effect.forkScoped); + yield* Effect.yieldNow; + yield* broker.invoke({ scope, tabId, operation: "status", input: {} }); + yield* Fiber.interrupt(click); + release.resolve(); + yield* Fiber.join(evaluation); + expect( + yield* broker.invoke({ + scope, + tabId, + operation: "evaluate", + input: { expression: "document.title" }, + }), + ).toBe("evaluated"); + expect(contexts[0]!.page.locator).not.toHaveBeenCalled(); + }), + ).pipe(Effect.provide(layer)), +); diff --git a/apps/server/src/preview/ServerBrowser.ts b/apps/server/src/preview/ServerBrowser.ts index 42dc916abb61c..386b93c72444d 100644 --- a/apps/server/src/preview/ServerBrowser.ts +++ b/apps/server/src/preview/ServerBrowser.ts @@ -42,6 +42,7 @@ import { resolvePreviewViewport } from "@t3tools/shared/previewViewport"; import * as NodeCrypto from "node:crypto"; import * as NodeFSP from "node:fs/promises"; import * as NodePath from "node:path"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import { constVoid } from "effect/Function"; @@ -1457,7 +1458,12 @@ const make = Effect.gen(function* () { } }; - const runOperation = async (request: PreviewAutomationRequest): Promise => { + const runOperation = async ( + request: PreviewAutomationRequest, + signal: AbortSignal, + remainingTimeoutMs: () => number, + ): Promise => { + signal.throwIfAborted(); const input = request.input; switch (request.operation) { case "status": @@ -1498,6 +1504,7 @@ const make = Effect.gen(function* () { ), ) : undefined; + signal.throwIfAborted(); const navigationTimeout = Math.min(request.timeoutMs, NAVIGATION_TIMEOUT_MS); if (!existing) { closeIdleAgentTabs(); @@ -1521,35 +1528,40 @@ const make = Effect.gen(function* () { "A browser dialog is pending. Read preview_status and use preview_dialog first.", "dialogPending", ); - return tab.control.agent(request.agentSessionId, async () => { - if (tab.dialog) - throw new BrowserControlInterrupted( - "A browser dialog is pending. Read preview_status and use preview_dialog first.", - "dialogPending", - ); - if (existing) { - if (url) await navigate(tab, url, "load", navigationTimeout); - } else { - // Await the original navigation failure even though background creation keeps the tab. - await tab.initialNavigation; - } - const reveal = open.open ?? open.show; - if (reveal !== false) { - await Effect.runPromise( - manager.requestReveal({ - threadId: tab.threadId, - tabId: tab.tabId, - force: reveal === true, - }), - ); - } - if (!existing && url) { - await tab.page - .waitForLoadState("load", { timeout: navigationTimeout }) - .catch(constVoid); - } - return statusWithTitle(tab, request.agentSessionId); - }); + return tab.control.agent( + request.agentSessionId, + async () => { + if (tab.dialog) + throw new BrowserControlInterrupted( + "A browser dialog is pending. Read preview_status and use preview_dialog first.", + "dialogPending", + ); + if (existing) { + if (url) await navigate(tab, url, "load", navigationTimeout); + } else { + // Await the original navigation failure even though background creation keeps the tab. + await tab.initialNavigation; + } + signal.throwIfAborted(); + const reveal = open.open ?? open.show; + if (reveal !== false) { + await Effect.runPromise( + manager.requestReveal({ + threadId: tab.threadId, + tabId: tab.tabId, + force: reveal === true, + }), + ); + } + if (!existing && url) { + await tab.page + .waitForLoadState("load", { timeout: navigationTimeout }) + .catch(constVoid); + } + return statusWithTitle(tab, request.agentSessionId); + }, + signal, + ); } case "recordingStop": { if ( @@ -1581,7 +1593,7 @@ const make = Effect.gen(function* () { "No recording is active for this thread.", ); } - return tab.control.agent(request.agentSessionId ?? "", () => stopRecording(tab)); + return tab.control.agent(request.agentSessionId ?? "", () => stopRecording(tab), signal); } } const tab = await requireTab(request); @@ -1602,22 +1614,31 @@ const make = Effect.gen(function* () { await resolveDialog(tab, input as PreviewAutomationDialogInput); return statusWithTitle(tab, request.agentSessionId); } - return tab.control.agent(request.agentSessionId!, async () => { - if (tab.dialog) - throw new BrowserControlInterrupted( - "A browser dialog is pending. Read preview_status and use preview_dialog first.", - "dialogPending", - ); - const generation = tab.control.generation; - try { - return await executeTabOperation(tab, request); - } finally { - if (generation !== tab.control.generation) ServerBrowserPage.invalidateRefs(tab.page); - } - }); + return tab.control.agent( + request.agentSessionId!, + async () => { + if (tab.dialog) + throw new BrowserControlInterrupted( + "A browser dialog is pending. Read preview_status and use preview_dialog first.", + "dialogPending", + ); + const generation = tab.control.generation; + try { + return await executeTabOperation(tab, request, signal, remainingTimeoutMs); + } finally { + if (generation !== tab.control.generation) ServerBrowserPage.invalidateRefs(tab.page); + } + }, + signal, + ); }; - const executeTabOperation = async (tab: ServerTab, request: PreviewAutomationRequest) => { + const executeTabOperation = async ( + tab: ServerTab, + request: PreviewAutomationRequest, + signal: AbortSignal, + remainingTimeoutMs: () => number, + ) => { const input = request.input; switch (request.operation) { case "navigate": { @@ -1653,9 +1674,17 @@ const make = Effect.gen(function* () { return { tabId: tab.tabId, colorScheme }; } case "snapshot": { - return withScreencastsPaused(tab, () => - ServerBrowserPage.snapshot({ ...tab, renderScale: RENDER_SCALE }), - ); + const includeImage = (input as { readonly includeImage?: boolean }).includeImage !== false; + const capture = () => + ServerBrowserPage.snapshot({ + ...tab, + renderScale: RENDER_SCALE, + includeImage, + signal, + timeoutMs: remainingTimeoutMs(), + }); + // Text reads do not repaint the page and must not wait on capture transport. + return includeImage ? withScreencastsPaused(tab, capture) : capture(); } case "click": { const clickInput = input as PreviewAutomationClickInput; @@ -1693,7 +1722,10 @@ const make = Effect.gen(function* () { ServerBrowserPage.scroll(tab.page, input as PreviewAutomationScrollInput), ); case "evaluate": - return ServerBrowserPage.evaluate(tab.cdp, input as PreviewAutomationEvaluateInput); + return ServerBrowserPage.evaluate(tab.cdp, input as PreviewAutomationEvaluateInput, { + signal, + timeoutMs: remainingTimeoutMs(), + }); case "waitFor": return ServerBrowserPage.waitFor(tab.page, input as PreviewAutomationWaitForInput); case "recordingStart": { @@ -1704,31 +1736,50 @@ const make = Effect.gen(function* () { }; const handleRequest = (connectionId: string, request: PreviewAutomationRequest) => - Effect.tryPromise({ - try: () => runOperation(request).finally(() => markUsed(request)), - catch: ServerBrowserPage.toOperationError, - }).pipe( - Effect.match({ - onSuccess: (result) => ({ ok: true as const, result }), - onFailure: (error) => ({ - ok: false as const, - error: { - _tag: error.tag, - message: error.message, - ...(error.detail === undefined ? {} : { detail: error.detail }), - }, + Clock.clockWith((clock) => { + const timeoutMs = Math.max(1, request.timeoutMs - Math.min(500, request.timeoutMs / 10)); + const started = clock.monotonicTimeNanosUnsafe(); + const remainingTimeoutMs = () => + Math.max(1, timeoutMs - Number(clock.monotonicTimeNanosUnsafe() - started) / 1_000_000); + return Effect.tryPromise({ + try: (signal) => + runOperation(request, signal, remainingTimeoutMs).finally(() => markUsed(request)), + catch: ServerBrowserPage.toOperationError, + }).pipe( + // Leave time to return the error before the broker deadline. The signal also + // revokes queued work so a timed-out click cannot execute after the queue drains. + Effect.timeout(timeoutMs), + Effect.catchTags({ + TimeoutError: () => + Effect.fail( + new ServerBrowserPage.ServerBrowserOperationError( + "PreviewAutomationTimeoutError", + `Preview automation ${request.operation} exceeded its execution deadline.`, + ), + ), }), - }), - Effect.flatMap((outcome) => - broker.respond({ - clientId: SERVER_HOST_CLIENT_ID, - connectionId, - requestId: request.requestId, - ...outcome, + Effect.match({ + onSuccess: (result) => ({ ok: true as const, result }), + onFailure: (error) => ({ + ok: false as const, + error: { + _tag: error.tag, + message: error.message, + ...(error.detail === undefined ? {} : { detail: error.detail }), + }, + }), }), - ), - Effect.ignore, - ); + Effect.flatMap((outcome) => + broker.respond({ + clientId: SERVER_HOST_CLIENT_ID, + connectionId, + requestId: request.requestId, + ...outcome, + }), + ), + Effect.ignore, + ); + }); const mirrorManagerEvent = (event: PreviewEvent) => Effect.promise(async () => { @@ -2154,7 +2205,7 @@ const make = Effect.gen(function* () { environmentId, supportedOperations: [...PREVIEW_AUTOMATION_SERVER_OPERATIONS], }, - { preferred: true }, + { preferred: true, disconnectOnTimeout: false }, ) .pipe( Effect.flatMap((events) => @@ -2164,15 +2215,25 @@ const make = Effect.gen(function* () { hostConnectionId = event.connectionId; return Effect.sync(reportLiveTabs); } - return handleRequest(event.connectionId, event.request).pipe( - Effect.forkScoped, - Effect.asVoid, - ); + return broker + .runRequest( + { + clientId: SERVER_HOST_CLIENT_ID, + connectionId: event.connectionId, + requestId: event.request.requestId, + }, + (remainingTimeoutMs) => + handleRequest(event.connectionId, { + ...event.request, + timeoutMs: remainingTimeoutMs, + }), + ) + .pipe(Effect.forkScoped, Effect.asVoid); }), ), ), ); - // The broker disconnects timed-out hosts, including slow first installs. Reconnect. + // Re-register after transport loss; individual request deadlines keep this host alive. yield* hostSession.pipe( Effect.exit, Effect.andThen(Effect.sleep(HOST_RECONNECT_DELAY)), diff --git a/apps/server/src/preview/ServerBrowserPage.reads.test.ts b/apps/server/src/preview/ServerBrowserPage.reads.test.ts new file mode 100644 index 0000000000000..171c1e37c8c98 --- /dev/null +++ b/apps/server/src/preview/ServerBrowserPage.reads.test.ts @@ -0,0 +1,406 @@ +import type { CDPSession, Page } from "playwright-core"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +import * as ServerBrowserPage from "./ServerBrowserPage.ts"; + +const deferred = () => { + let resolve!: (value: T) => void; + let reject!: (cause: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +}; + +const metadata = { + url: "https://example.test/", + title: "Test page", + loading: false, + visibleText: "continue", + interactiveElements: [], +}; +const layout = { cssVisualViewport: { pageX: 10, pageY: 20 } }; +const makeBrowser = () => { + const pageEvaluate = vi.fn().mockResolvedValue(metadata); + const ariaSnapshot = vi.fn().mockResolvedValue('- button "continue" [ref=e1]'); + const send = vi.fn<(method: string, params?: unknown) => Promise>((method) => { + switch (method) { + case "Page.getLayoutMetrics": + return Promise.resolve(layout); + case "Page.captureScreenshot": + return Promise.resolve({ data: "image" }); + case "Runtime.evaluate": + return Promise.resolve({ result: { value: 42 } }); + default: + return Promise.reject(new Error(`Unexpected CDP method: ${method}`)); + } + }); + const page = { + viewportSize: () => ({ width: 2560, height: 1600 }), + on: vi.fn(), + evaluate: pageEvaluate, + ariaSnapshot, + } as unknown as Page; + const cdp = { send } as unknown as CDPSession; + const snapshot = ( + options: { timeoutMs?: number; includeImage?: boolean; signal?: AbortSignal } = {}, + ) => + ServerBrowserPage.snapshot({ + page, + cdp, + renderScale: 1, + consoleEntries: [], + networkEntries: [], + actionTimeline: [], + ...options, + }); + return { page, cdp, send, pageEvaluate, ariaSnapshot, snapshot }; +}; + +const observeAbort = () => { + const controller = new AbortController(); + const added = vi.spyOn(controller.signal, "addEventListener"); + const removed = vi.spyOn(controller.signal, "removeEventListener"); + const expectClean = () => { + for (const [event, listener] of added.mock.calls) { + expect(removed).toHaveBeenCalledWith(event, listener); + } + expect(vi.getTimerCount()).toBe(0); + }; + return { controller, added, expectClean }; +}; + +describe("bounded server browser reads", () => { + beforeEach(() => vi.useFakeTimers({ now: 0 })); + afterEach(() => { + vi.clearAllTimers(); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it.each([ + "snapshot metadata", + "accessibility tree", + "Page.getLayoutMetrics", + "Page.captureScreenshot", + ])("times out a stalled %s stage and permits a fresh snapshot", async (stage) => { + const browser = makeBrowser(); + const stalled = deferred(); + const abort = observeAbort(); + if (stage === "snapshot metadata") browser.pageEvaluate.mockReturnValueOnce(stalled.promise); + else if (stage === "accessibility tree") + browser.ariaSnapshot.mockReturnValueOnce(stalled.promise); + else if (stage === "Page.getLayoutMetrics") browser.send.mockReturnValueOnce(stalled.promise); + else browser.send.mockResolvedValueOnce(layout).mockReturnValueOnce(stalled.promise); + const failed = expect( + browser.snapshot({ timeoutMs: 100, signal: abort.controller.signal }), + ).rejects.toMatchObject({ + tag: "PreviewAutomationTimeoutError", + message: expect.stringContaining(stage), + detail: { stage, timeoutMs: 100 }, + }); + await vi.advanceTimersByTimeAsync(100); + await failed; + abort.expectClean(); + if (stage === "Page.getLayoutMetrics") + expect(browser.send).not.toHaveBeenCalledWith("Page.captureScreenshot", expect.anything()); + const fresh = await browser.snapshot({ signal: abort.controller.signal }); + expect(fresh).toMatchObject({ title: "Test page", screenshot: { data: "image" } }); + expect(String(fresh.accessibilityTree)).toContain("[ref=t3-"); + abort.expectClean(); + // The timed-out protocol response can reject after the replacement read finishes. + stalled.reject(new Error("Late browser response")); + await vi.advanceTimersByTimeAsync(0); + abort.expectClean(); + }); + + it.each(["Page.getLayoutMetrics", "Page.captureScreenshot"])( + "bounds a standalone stalled %s and permits another capture", + async (stage) => { + const browser = makeBrowser(); + const stalled = deferred(); + const abort = observeAbort(); + if (stage === "Page.getLayoutMetrics") browser.send.mockReturnValueOnce(stalled.promise); + else browser.send.mockResolvedValueOnce(layout).mockReturnValueOnce(stalled.promise); + const options = { + format: "jpeg", + quality: 70, + scale: 0.5, + timeoutMs: 100, + signal: abort.controller.signal, + } as const; + const failed = expect( + ServerBrowserPage.captureViewport(browser.page, browser.cdp, options), + ).rejects.toMatchObject({ + tag: "PreviewAutomationTimeoutError", + detail: { stage, timeoutMs: 100 }, + }); + await vi.advanceTimersByTimeAsync(100); + await failed; + abort.expectClean(); + expect(await ServerBrowserPage.captureViewport(browser.page, browser.cdp, options)).toBe( + "image", + ); + abort.expectClean(); + stalled.reject(new Error("Late capture failure")); + await vi.advanceTimersByTimeAsync(0); + }, + ); + + it("shares the total capture budget across layout and screenshot", async () => { + const browser = makeBrowser(); + const metrics = deferred(); + browser.send + .mockReturnValueOnce(metrics.promise) + .mockReturnValueOnce(new Promise(() => {})); + const failed = expect( + ServerBrowserPage.captureViewport(browser.page, browser.cdp, { + format: "png", + scale: 0.5, + timeoutMs: 100, + }), + ).rejects.toMatchObject({ + tag: "PreviewAutomationTimeoutError", + detail: { stage: "Page.captureScreenshot", timeoutMs: 100 }, + }); + await vi.advanceTimersByTimeAsync(80); + metrics.resolve(layout); + await vi.advanceTimersByTimeAsync(0); + expect(browser.send).toHaveBeenCalledWith( + "Page.captureScreenshot", + expect.objectContaining({ clip: { x: 10, y: 20, width: 2560, height: 1600, scale: 0.5 } }), + ); + await vi.advanceTimersByTimeAsync(19); + expect(vi.getTimerCount()).toBe(1); + await vi.advanceTimersByTimeAsync(1); + await failed; + expect(vi.getTimerCount()).toBe(0); + }); + + it("shares the snapshot deadline with the screenshot after layout completes", async () => { + const browser = makeBrowser(); + const metrics = deferred(); + browser.send + .mockReturnValueOnce(metrics.promise) + .mockReturnValueOnce(new Promise(() => {})); + const failed = expect(browser.snapshot({ timeoutMs: 100 })).rejects.toMatchObject({ + detail: { stage: "Page.captureScreenshot", timeoutMs: 100 }, + }); + await vi.advanceTimersByTimeAsync(80); + metrics.resolve(layout); + await vi.advanceTimersByTimeAsync(20); + await failed; + expect(vi.getTimerCount()).toBe(0); + }); + + it("skips both capture stages when the snapshot does not include an image", async () => { + const browser = makeBrowser(); + browser.send.mockReturnValue(new Promise(() => {})); + const result = await browser.snapshot({ includeImage: false }); + expect(result).toMatchObject(metadata); + expect(result).not.toHaveProperty("screenshot"); + expect(browser.send).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("cleans up all parallel stages immediately when one fails", async () => { + const browser = makeBrowser(); + const abort = observeAbort(); + const original = new Error("Browser disconnected"); + browser.pageEvaluate.mockRejectedValueOnce(original); + browser.ariaSnapshot.mockReturnValueOnce(new Promise(() => {})); + browser.send.mockReturnValueOnce(new Promise(() => {})); + await expect(browser.snapshot({ signal: abort.controller.signal })).rejects.toBe(original); + abort.expectClean(); + }); + + it("aborts every pending snapshot stage and never starts capture after delayed layout", async () => { + const browser = makeBrowser(); + const abort = observeAbort(); + const metrics = deferred(); + browser.pageEvaluate.mockReturnValueOnce(new Promise(() => {})); + browser.ariaSnapshot.mockReturnValueOnce(new Promise(() => {})); + browser.send.mockReturnValueOnce(metrics.promise); + const failed = expect( + browser.snapshot({ signal: abort.controller.signal }), + ).rejects.toMatchObject({ tag: "PreviewAutomationControlInterruptedError" }); + abort.controller.abort(); + await failed; + abort.expectClean(); + metrics.resolve(layout); + await vi.advanceTimersByTimeAsync(0); + expect(browser.send.mock.calls).toEqual([["Page.getLayoutMetrics"]]); + await expect(browser.snapshot()).resolves.toMatchObject({ title: "Test page" }); + abort.expectClean(); + }); + + it.each(["snapshot", "capture", "evaluate"])( + "does not dispatch a pre-aborted %s read", + async (operation) => { + const browser = makeBrowser(); + const abort = observeAbort(); + abort.controller.abort(); + const result = + operation === "snapshot" + ? browser.snapshot({ signal: abort.controller.signal }) + : operation === "capture" + ? ServerBrowserPage.captureViewport(browser.page, browser.cdp, { + format: "png", + scale: 0.5, + signal: abort.controller.signal, + }) + : ServerBrowserPage.evaluate( + browser.cdp, + { expression: "42" }, + { signal: abort.controller.signal }, + ); + await expect(result).rejects.toMatchObject({ + tag: "PreviewAutomationControlInterruptedError", + }); + expect(browser.pageEvaluate).not.toHaveBeenCalled(); + expect(browser.ariaSnapshot).not.toHaveBeenCalled(); + expect(browser.send).not.toHaveBeenCalled(); + expect(abort.added).not.toHaveBeenCalled(); + abort.expectClean(); + }, + ); + + it("bounds an unanswered Runtime.evaluate and preserves a later read", async () => { + const browser = makeBrowser(); + const stalled = deferred(); + const abort = observeAbort(); + browser.send.mockReturnValueOnce(stalled.promise); + const failed = expect( + ServerBrowserPage.evaluate( + browser.cdp, + { expression: "new Promise(() => {})" }, + { timeoutMs: 100, signal: abort.controller.signal }, + ), + ).rejects.toMatchObject({ + tag: "PreviewAutomationTimeoutError", + detail: { stage: "Runtime.evaluate", timeoutMs: 100 }, + }); + expect(browser.send).toHaveBeenCalledWith("Runtime.evaluate", { + expression: "new Promise(() => {})", + awaitPromise: true, + returnByValue: true, + timeout: 100, + }); + await vi.advanceTimersByTimeAsync(100); + await failed; + abort.expectClean(); + expect( + await ServerBrowserPage.evaluate( + browser.cdp, + { expression: "42" }, + { signal: abort.controller.signal }, + ), + ).toBe(42); + abort.expectClean(); + stalled.reject(new Error("Late evaluation failure")); + await vi.advanceTimersByTimeAsync(0); + expect(browser.send.mock.calls.map(([method]) => method)).toEqual([ + "Runtime.evaluate", + "Runtime.evaluate", + ]); + }); + + it.each(["capture", "evaluate"])("cleans up an aborted %s read", async (operation) => { + const browser = makeBrowser(); + const abort = observeAbort(); + const stalled = deferred(); + browser.send.mockReturnValueOnce(stalled.promise); + const failed = expect( + operation === "capture" + ? ServerBrowserPage.captureViewport(browser.page, browser.cdp, { + format: "png", + scale: 0.5, + signal: abort.controller.signal, + }) + : ServerBrowserPage.evaluate( + browser.cdp, + { expression: "42" }, + { signal: abort.controller.signal }, + ), + ).rejects.toMatchObject({ tag: "PreviewAutomationControlInterruptedError" }); + abort.controller.abort(); + await failed; + abort.expectClean(); + stalled.reject(new Error("Late aborted read failure")); + await vi.advanceTimersByTimeAsync(0); + expect(await ServerBrowserPage.evaluate(browser.cdp, { expression: "42" })).toBe(42); + abort.expectClean(); + }); + + it.each(["capture", "evaluate"])("uses the default %s budget", async (operation) => { + const browser = makeBrowser(); + browser.send.mockReturnValueOnce(new Promise(() => {})); + const timeoutMs = operation === "capture" ? 3_000 : 10_000; + const failed = expect( + operation === "capture" + ? ServerBrowserPage.captureViewport(browser.page, browser.cdp, { format: "png", scale: 1 }) + : ServerBrowserPage.evaluate(browser.cdp, { expression: "42" }), + ).rejects.toMatchObject({ detail: { timeoutMs } }); + await vi.advanceTimersByTimeAsync(timeoutMs); + await failed; + expect(vi.getTimerCount()).toBe(0); + }); + + it("keeps oversized configured read budgets below the broker deadline", async () => { + const browser = makeBrowser(); + browser.pageEvaluate.mockReturnValueOnce(new Promise(() => {})); + const failed = expect( + browser.snapshot({ includeImage: false, timeoutMs: 60_000 }), + ).rejects.toMatchObject({ detail: { timeoutMs: 10_000 } }); + await vi.advanceTimersByTimeAsync(10_000); + await failed; + expect(vi.getTimerCount()).toBe(0); + }); + + it.each(["response", "protocol error"])( + "rejects a late evaluation %s even before the timer callback runs", + async (completion) => { + const browser = makeBrowser(); + const reply = deferred<{ result: { value: number } }>(); + const abort = observeAbort(); + browser.send.mockReturnValueOnce(reply.promise); + const failed = expect( + ServerBrowserPage.evaluate( + browser.cdp, + { expression: "42" }, + { + timeoutMs: 100, + signal: abort.controller.signal, + }, + ), + ).rejects.toMatchObject({ + tag: "PreviewAutomationTimeoutError", + detail: { stage: "Runtime.evaluate", timeoutMs: 100 }, + }); + // Move the deadline clock without running timers, like a delayed Node callback. + vi.setSystemTime(100); + if (completion === "response") reply.resolve({ result: { value: 42 } }); + else reply.reject(new Error("Protocol error (Runtime.evaluate): Execution was terminated")); + await failed; + abort.expectClean(); + }, + ); + + it("cleans up a synchronously thrown protocol failure", async () => { + const browser = makeBrowser(); + const abort = observeAbort(); + const failure = new Error("Session detached"); + browser.send.mockImplementationOnce(() => { + throw failure; + }); + await expect( + ServerBrowserPage.evaluate( + browser.cdp, + { expression: "42" }, + { signal: abort.controller.signal }, + ), + ).rejects.toBe(failure); + abort.expectClean(); + }); +}); diff --git a/apps/server/src/preview/ServerBrowserPage.test.ts b/apps/server/src/preview/ServerBrowserPage.test.ts index 3ab9d6dc35b8f..5d8cb8fbc12dd 100644 --- a/apps/server/src/preview/ServerBrowserPage.test.ts +++ b/apps/server/src/preview/ServerBrowserPage.test.ts @@ -56,6 +56,62 @@ describe("server browser element refs", () => { }; const repeatedRows = `
    ${Array.from({ length: 5 }, (_, i) => `
  • row ${i + 1}
    • `).join("")}
`; + it("captures usable element refs without an image", async () => { + await page.setContent(""); + const result = await ServerBrowserPage.snapshot({ + page, + cdp, + renderScale: 1, + consoleEntries: [], + networkEntries: [], + actionTimeline: [], + includeImage: false, + }); + expect(result.screenshot).toBeUndefined(); + expect(result.visibleText).toBe("continue"); + const controller = new AbortController(); + controller.abort(); + await expect( + ServerBrowserPage.snapshot({ + page, + cdp, + renderScale: 1, + consoleEntries: [], + networkEntries: [], + actionTimeline: [], + signal: controller.signal, + }), + ).rejects.toMatchObject({ tag: "PreviewAutomationControlInterruptedError" }); + await ServerBrowserPage.click(page, { + locator: buttonLocator(result.accessibilityTree, "continue"), + }); + expect(await page.locator("button").textContent()).toBe("clicked"); + }); + + it("bounds unresolved page promises and permits a later evaluation", async () => { + await expect( + ServerBrowserPage.evaluate(cdp, { expression: "new Promise(() => {})" }, { timeoutMs: 100 }), + ).rejects.toMatchObject({ + tag: "PreviewAutomationTimeoutError", + detail: { stage: "Runtime.evaluate", timeoutMs: 100 }, + }); + expect(await ServerBrowserPage.evaluate(cdp, { expression: "1 + 1" })).toBe(2); + }); + + it("bounds synchronous page execution and permits a later evaluation", async () => { + await expect( + ServerBrowserPage.evaluate( + cdp, + { expression: "(() => { while (true) {} })()" }, + { timeoutMs: 100 }, + ), + ).rejects.toMatchObject({ + tag: "PreviewAutomationTimeoutError", + detail: { stage: "Runtime.evaluate", timeoutMs: 100 }, + }); + expect(await ServerBrowserPage.evaluate(cdp, { expression: "1 + 1" })).toBe(2); + }); + it("clicks the fifth repeated delete control without touching row one", async () => { await page.setContent(repeatedRows); const result = await takeSnapshot(); diff --git a/apps/server/src/preview/ServerBrowserPage.ts b/apps/server/src/preview/ServerBrowserPage.ts index b6a397e293a68..7c97529d8359f 100644 --- a/apps/server/src/preview/ServerBrowserPage.ts +++ b/apps/server/src/preview/ServerBrowserPage.ts @@ -64,6 +64,97 @@ export const toOperationError = (cause: unknown): ServerBrowserOperationError => }; const DEFAULT_TIMEOUT_MS = 15_000; +const DEFAULT_READ_TIMEOUT_MS = 10_000; +const DEFAULT_CAPTURE_TIMEOUT_MS = 3_000; + +type ReadOptions = { + readonly timeoutMs?: number | undefined; + readonly signal?: AbortSignal | undefined; +}; +type ReadStage = (stage: string, operation: (timeoutMs: number) => Promise) => Promise; + +/** One deadline across parallel and sequential stages, below the broker's 15s deadline. */ +const withReadBudget = async ( + options: ReadOptions, + operation: (read: ReadStage) => Promise, +): Promise => { + const timeoutMs = + options.timeoutMs !== undefined && Number.isFinite(options.timeoutMs) + ? Math.max(1, Math.min(options.timeoutMs, DEFAULT_READ_TIMEOUT_MS)) + : DEFAULT_READ_TIMEOUT_MS; + const interrupted = (stage?: string) => + options.signal?.reason instanceof BrowserControlInterrupted + ? options.signal.reason + : new ServerBrowserOperationError( + "PreviewAutomationControlInterruptedError", + stage === undefined + ? "Browser read was interrupted." + : `Browser read was interrupted during ${stage}.`, + stage === undefined ? undefined : { stage }, + ); + // Reject before snapshot setup can revoke the last usable element refs. + if (options.signal?.aborted) throw interrupted(); + const deadline = Date.now() + timeoutMs; + const pending = new Set<() => void>(); + let closed = false; + const read: ReadStage = (stage, start) => + new Promise((resolve, reject) => { + const timedOut = () => + new ServerBrowserOperationError( + "PreviewAutomationTimeoutError", + `Browser read timed out after ${timeoutMs}ms during ${stage}.`, + { stage, timeoutMs }, + ); + let timer: ReturnType | undefined; + let settled = false; + const finish = (complete: () => void) => { + if (settled) return; + settled = true; + clearTimeout(timer); + options.signal?.removeEventListener("abort", onAbort); + pending.delete(onAbort); + complete(); + }; + const onAbort = () => finish(() => reject(interrupted(stage))); + if (closed || options.signal?.aborted) { + onAbort(); + return; + } + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) { + finish(() => reject(timedOut())); + return; + } + pending.add(onAbort); + options.signal?.addEventListener("abort", onAbort, { once: true }); + timer = setTimeout(() => finish(() => reject(timedOut())), remainingMs); + const fail = (cause: unknown) => + finish(() => + reject( + Date.now() >= deadline || (cause instanceof Error && cause.name === "TimeoutError") + ? timedOut() + : cause, + ), + ); + try { + // CDP and Playwright cannot cancel an individual in-flight call. Keep both + // handlers attached after expiry, without terminating the shared page. + start(remainingMs).then( + (value) => finish(() => (Date.now() >= deadline ? reject(timedOut()) : resolve(value))), + fail, + ); + } catch (cause) { + fail(cause); + } + }); + try { + return await operation(read); + } finally { + // A failed parallel stage must also release its siblings' timers/listeners. + closed = true; + for (const cancel of pending) cancel(); + } +}; const pageRefs = new WeakMap }>(); // A compact runtime namespace prevents old refs from aliasing after a server restart. @@ -150,81 +241,111 @@ const SNAPSHOT_SCRIPT = `(() => { }; })()`; +type CaptureOptions = ReadOptions & { + readonly format: "png" | "jpeg"; + readonly quality?: number; + readonly scale: number; +}; + // Scaled captures repaint live screencasts, so callers pause them. Clips use document offsets. -export const captureViewport = async ( +const captureViewportWithinBudget = async ( page: Page, cdp: CDPSession, - options: { readonly format: "png" | "jpeg"; readonly quality?: number; readonly scale: number }, + options: CaptureOptions, + read: ReadStage, ) => { - let clip; - if (options.scale < 1) { - const viewport = page.viewportSize() ?? { width: 1280, height: 800 }; - const { cssVisualViewport } = await cdp.send("Page.getLayoutMetrics"); - clip = { - x: cssVisualViewport.pageX, - y: cssVisualViewport.pageY, - ...viewport, - scale: options.scale, - }; - } - const { data } = await cdp.send("Page.captureScreenshot", { - format: options.format, - ...(options.quality === undefined ? {} : { quality: options.quality }), - ...(clip ? { clip } : {}), - }); + const metrics = + options.scale < 1 + ? await read("Page.getLayoutMetrics", () => cdp.send("Page.getLayoutMetrics")) + : undefined; + const clip = metrics && { + x: metrics.cssVisualViewport.pageX, + y: metrics.cssVisualViewport.pageY, + ...(page.viewportSize() ?? { width: 1280, height: 800 }), + scale: options.scale, + }; + const { data } = await read("Page.captureScreenshot", () => + cdp.send("Page.captureScreenshot", { + format: options.format, + ...(options.quality === undefined ? {} : { quality: options.quality }), + ...(clip ? { clip } : {}), + }), + ); return data; }; -export const snapshot = async (input: { +export const captureViewport = (page: Page, cdp: CDPSession, options: CaptureOptions) => + withReadBudget( + { ...options, timeoutMs: options.timeoutMs ?? DEFAULT_CAPTURE_TIMEOUT_MS }, + (read) => captureViewportWithinBudget(page, cdp, options, read), + ); + +export const snapshot = (input: { readonly page: Page; readonly cdp: CDPSession; readonly renderScale: number; readonly consoleEntries: ReadonlyArray; readonly networkEntries: ReadonlyArray; readonly actionTimeline: PreviewAutomationSnapshot["actionTimeline"]; -}): Promise => { - const viewport = input.page.viewportSize() ?? { width: 1280, height: 800 }; - const scale = Math.min(1, MAX_SCREENSHOT_WIDTH / (viewport.width * input.renderScale)); - const state = refsFor(input.page); - invalidateRefs(input.page); - const generation = state.generation; - const [page, tree, data] = await Promise.all([ - input.page.evaluate(SNAPSHOT_SCRIPT) as Promise< - Pick< - PreviewAutomationSnapshot, - "url" | "title" | "loading" | "visibleText" | "interactiveElements" - > - >, - input.page.ariaSnapshot({ mode: "ai", boxes: true, timeout: DEFAULT_TIMEOUT_MS }), - captureViewport(input.page, input.cdp, { format: "png", scale }), - ]); - if (state.generation !== generation) { - throw new ServerBrowserOperationError( - "PreviewAutomationExecutionError", - "The page changed while capturing its snapshot. Take another snapshot.", - ); - } - const accessibilityTree = tree - .slice(0, MAX_VISIBLE_TEXT_LENGTH) - .replace(/\[ref=((?:f\d+)?e\d+)\]/g, (_match, nativeRef: string) => { - const ref = `t3-${generation}-${nativeRef}`; - state.refs.set(ref, nativeRef); - return `[ref=${ref}]`; - }); - return { - ...page, - accessibilityTree, - consoleEntries: [...input.consoleEntries], - networkEntries: [...input.networkEntries], - actionTimeline: [...input.actionTimeline], - screenshot: { - mimeType: "image/png", - data, - width: Math.round(viewport.width * input.renderScale * scale), - height: Math.round(viewport.height * input.renderScale * scale), - }, - }; -}; + readonly timeoutMs?: number | undefined; + readonly includeImage?: boolean | undefined; + readonly signal?: AbortSignal | undefined; +}): Promise => + withReadBudget(input, async (read) => { + const viewport = input.page.viewportSize() ?? { width: 1280, height: 800 }; + const scale = Math.min(1, MAX_SCREENSHOT_WIDTH / (viewport.width * input.renderScale)); + const state = refsFor(input.page); + invalidateRefs(input.page); + const generation = state.generation; + const [page, tree, data] = await Promise.all([ + read( + "snapshot metadata", + () => + input.page.evaluate(SNAPSHOT_SCRIPT) as Promise< + Pick< + PreviewAutomationSnapshot, + "url" | "title" | "loading" | "visibleText" | "interactiveElements" + > + >, + ), + read("accessibility tree", (timeoutMs) => + input.page.ariaSnapshot({ mode: "ai", boxes: true, timeout: timeoutMs }), + ), + input.includeImage === false + ? undefined + : captureViewportWithinBudget(input.page, input.cdp, { format: "png", scale }, read), + ]); + if (state.generation !== generation) { + throw new ServerBrowserOperationError( + "PreviewAutomationExecutionError", + "The page changed while capturing its snapshot. Take another snapshot.", + ); + } + const accessibilityTree = tree + .slice(0, MAX_VISIBLE_TEXT_LENGTH) + .replace(/\[ref=((?:f\d+)?e\d+)\]/g, (_match, nativeRef: string) => { + const ref = `t3-${generation}-${nativeRef}`; + state.refs.set(ref, nativeRef); + return `[ref=${ref}]`; + }); + return { + ...page, + accessibilityTree, + consoleEntries: [...input.consoleEntries], + networkEntries: [...input.networkEntries], + actionTimeline: [...input.actionTimeline], + ...(data === undefined + ? {} + : { + screenshot: { + mimeType: "image/png" as const, + data, + width: Math.round(viewport.width * input.renderScale * scale), + height: Math.round(viewport.height * input.renderScale * scale), + }, + }), + }; + }); /** * A click whose handler opens a dialog does not finish until the dialog is @@ -410,12 +531,21 @@ export const scroll = async (page: Page, input: PreviewAutomationScrollInput) => await locator.evaluate((element, [x, y]) => element.scrollBy(x, y), delta); }; -export const evaluate = async (cdp: CDPSession, input: PreviewAutomationEvaluateInput) => { - const result = await cdp.send("Runtime.evaluate", { - expression: input.expression, - awaitPromise: input.awaitPromise ?? true, - returnByValue: input.returnByValue ?? true, - }); +export const evaluate = async ( + cdp: CDPSession, + input: PreviewAutomationEvaluateInput, + options: ReadOptions = {}, +) => { + const result = await withReadBudget(options, (read) => + read("Runtime.evaluate", (timeoutMs) => + cdp.send("Runtime.evaluate", { + expression: input.expression, + awaitPromise: input.awaitPromise ?? true, + returnByValue: input.returnByValue ?? true, + timeout: timeoutMs, + }), + ), + ); if (result.exceptionDetails) { throw new ServerBrowserOperationError( "PreviewAutomationExecutionError", diff --git a/apps/server/src/preview/SessionControl.test.ts b/apps/server/src/preview/SessionControl.test.ts index 73d64cee10196..5c7dffd5ed6b3 100644 --- a/apps/server/src/preview/SessionControl.test.ts +++ b/apps/server/src/preview/SessionControl.test.ts @@ -3,6 +3,45 @@ import { describe, expect, it } from "vite-plus/test"; import { BrowserControlInterrupted, SessionControl } from "./SessionControl.ts"; describe("SessionControl", () => { + it("discards an expired queued action without replaying it after the active action drains", async () => { + const control = new SessionControl("agent"); + const started = Promise.withResolvers(); + const finish = Promise.withResolvers(); + const events: string[] = []; + const running = control.agent("agent", async () => { + started.resolve(); + await finish.promise; + events.push("finished"); + }); + await started.promise; + const expired = new AbortController(); + const queued = expect( + control.agent("agent", async () => events.push("late click"), expired.signal), + ).rejects.toThrow("request expired"); + expired.abort(new Error("request expired")); + const next = control.agent("agent", async () => events.push("fresh action")); + finish.resolve(); + await Promise.all([running, queued, next]); + expect(events).toEqual(["finished", "fresh action"]); + }); + + it("rejects an already cancelled action before it reaches the queue", async () => { + const control = new SessionControl("agent"); + const cancelled = AbortSignal.abort(new Error("request cancelled")); + let executed = false; + await expect( + control.agent( + "agent", + async () => { + executed = true; + }, + cancelled, + ), + ).rejects.toThrow("request cancelled"); + expect(executed).toBe(false); + await expect(control.agent("agent", async () => "ready")).resolves.toBe("ready"); + }); + it("returns an action result before tracked navigation but drains it before already queued actions", async () => { const control = new SessionControl("agent"); const committed = Promise.withResolvers(); diff --git a/apps/server/src/preview/SessionControl.ts b/apps/server/src/preview/SessionControl.ts index a361cd1a467c0..62da0e1b429c5 100644 --- a/apps/server/src/preview/SessionControl.ts +++ b/apps/server/src/preview/SessionControl.ts @@ -62,18 +62,20 @@ export class SessionControl { this.onGenerationChange(); } - private async action(allowed: () => boolean, run: () => Promise) { + private async action(allowed: () => boolean, run: () => Promise, signal?: AbortSignal) { + signal?.throwIfAborted(); this.assertOpen(); if (!allowed()) throw new BrowserControlInterrupted("You do not control this browser tab."); const epoch = this.epoch; return this.enqueue(async () => { + signal?.throwIfAborted(); this.assertOpen(); if (epoch !== this.epoch || !allowed()) throw new BrowserControlInterrupted(); return run(); }); } - agent(agentId: string, run: () => Promise) { + agent(agentId: string, run: () => Promise, signal?: AbortSignal) { if (this.agentId !== agentId) return Promise.reject( new BrowserControlInterrupted("This tab belongs to another agent.", "agentMismatch"), @@ -82,7 +84,7 @@ export class SessionControl { return Promise.reject( new BrowserControlInterrupted("A human controls this tab.", "humanControl"), ); - return this.action(() => this.agentId === agentId && this.owner === null, run); + return this.action(() => this.agentId === agentId && this.owner === null, run, signal); } /** diff --git a/docs/fork-differences.md b/docs/fork-differences.md index 49f217c1a832f..9ce2c193371f2 100644 --- a/docs/fork-differences.md +++ b/docs/fork-differences.md @@ -15,6 +15,20 @@ git log --oneline --no-merges upstream/main..HEAD Agents update this page in the same change that adds, changes, or removes a difference. The rule is in [AGENTS.md](../AGENTS.md#fork-differences). +## Browser automation survives request timeouts + +The server's built-in browser stays available when an individual automation request times out, +preserving other sessions and their tabs. Expired or cancelled queued actions cannot execute +later. Snapshot and evaluation reads have bounded deadlines, and `preview_snapshot` with +`includeImage: false` skips capture unless `save: true` needs a screenshot file. + +This ports upstream [PR #16941](https://github.com/pingdotgg/t3code/pull/16941) ahead of its merge. +External browser hosts keep upstream's disconnect-on-timeout behavior. + +Code: [PreviewAutomationBroker.ts](../apps/server/src/mcp/PreviewAutomationBroker.ts), +[ServerBrowser.ts](../apps/server/src/preview/ServerBrowser.ts), and +[ServerBrowserPage.ts](../apps/server/src/preview/ServerBrowserPage.ts). + ## Pause threads on update Updating the desktop app or a connected server pauses running threads before diff --git a/packages/contracts/src/previewAutomation.ts b/packages/contracts/src/previewAutomation.ts index fa16511bd3e81..4947591520086 100644 --- a/packages/contracts/src/previewAutomation.ts +++ b/packages/contracts/src/previewAutomation.ts @@ -703,12 +703,14 @@ export const PreviewAutomationSnapshot = Schema.Struct({ consoleEntries: Schema.Array(PreviewAutomationConsoleEntry), networkEntries: Schema.Array(PreviewAutomationNetworkEntry), actionTimeline: Schema.Array(PreviewAutomationActionEvent), - screenshot: Schema.Struct({ - mimeType: Schema.Literal("image/png"), - data: Schema.String, - width: Schema.Int, - height: Schema.Int, - }), + screenshot: Schema.optional( + Schema.Struct({ + mimeType: Schema.Literal("image/png"), + data: Schema.String, + width: Schema.Int, + height: Schema.Int, + }), + ), }); export type PreviewAutomationSnapshot = typeof PreviewAutomationSnapshot.Type;