diff --git a/apps/server/src/mcp/PreviewAutomationBroker.ts b/apps/server/src/mcp/PreviewAutomationBroker.ts index c1ae21dfd3b4c..66af77067e60d 100644 --- a/apps/server/src/mcp/PreviewAutomationBroker.ts +++ b/apps/server/src/mcp/PreviewAutomationBroker.ts @@ -33,6 +33,8 @@ import * as Crypto from "effect/Crypto"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Fiber from "effect/Fiber"; +import * as Scope from "effect/Scope"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; @@ -69,6 +71,10 @@ export class PreviewAutomationBroker extends Context.Service< readonly respond: ( response: PreviewAutomationResponse, ) => Effect.Effect; + readonly handleRequest: ( + request: Pick, + effect: Effect.Effect, + ) => Effect.Effect; readonly invoke: ( request: PreviewAutomationInvokeInput, ) => Effect.Effect; @@ -88,6 +94,7 @@ interface ClientConnection { } interface PendingRequest { + readonly scope: Scope.Scope; readonly queue: ClientConnection["queue"]; readonly deferred: Deferred.Deferred; readonly context: PreviewAutomationRequestErrorContext; @@ -497,9 +504,26 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() { } }); + const handleRequest: PreviewAutomationBroker["Service"]["handleRequest"] = Effect.fn( + "PreviewAutomationBroker.handleRequest", + )(function* (request, effect) { + const current = yield* SynchronizedRef.get(state); + const pending = current.pending.get(request.requestId); + if ( + !pending || + pending.context.clientId !== request.clientId || + pending.context.connectionId !== request.connectionId + ) + return; + // In-process host work belongs to the caller, including interruption cleanup. + const fiber = yield* Effect.forkIn(effect, pending.scope); + return yield* Fiber.join(fiber); + }); + const invoke = Effect.fn("PreviewAutomationBroker.invoke")(function* ( input: Parameters[0], - ): Effect.fn.Return { + ): Effect.fn.Return { + const requestScope = yield* Effect.scope; const deferred = yield* Deferred.make(); const route = yield* SynchronizedRef.modify(state, (current) => { const assignments = new Map( @@ -586,7 +610,7 @@ 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, context, scope: requestScope }); return [ { connection, requestId, requestContext: context, requestSequence }, { ...current, assignments, pending, requestSequence: current.requestSequence + 1 }, @@ -692,9 +716,9 @@ export const make = Effect.gen(function* PreviewAutomationBrokerMake() { return { ...current, assignments }; }); return result; - }); + }, Effect.scoped); - return PreviewAutomationBroker.of({ connect, focusHost, respond, invoke }); + return PreviewAutomationBroker.of({ connect, focusHost, respond, invoke, handleRequest }); }).pipe(Effect.withSpan("PreviewAutomationBroker.make")); export const layer = Layer.effect(PreviewAutomationBroker, make); diff --git a/apps/server/src/preview/ServerBrowser.test.ts b/apps/server/src/preview/ServerBrowser.test.ts index 961ea352c7c3b..999ac1e5aacbe 100644 --- a/apps/server/src/preview/ServerBrowser.test.ts +++ b/apps/server/src/preview/ServerBrowser.test.ts @@ -44,6 +44,7 @@ vi.mock("./ServerBrowserContexts.ts", () => ({ async contextFor(profileId: string, isolationKey?: string) { contextRequests.push({ profileId, isolated: isolationKey !== undefined }); if (contextFailure) throw contextFailure; + contextStarted?.resolve(); await contextGate?.promise; const context = makeContext(this.onClose); contexts.push(context); @@ -202,6 +203,7 @@ function makeContext(onClose?: (context: BrowserContext) => void) { const contexts: ReturnType[] = []; const contextRequests: Array<{ profileId: string; isolated: boolean }> = []; let contextGate: PromiseWithResolvers | null = null; +let contextStarted: PromiseWithResolvers | null = null; let pageCopy = ""; let keyCopy = ""; let focusedFrame: ReturnType | null = null; @@ -315,6 +317,7 @@ beforeEach(() => { contexts.length = 0; contextRequests.length = 0; contextGate = null; + contextStarted = null; contextFailure = null; pageCopy = ""; keyCopy = ""; @@ -1563,6 +1566,205 @@ it.live("control updates replace a stalled viewer backlog in the order they happ ).pipe(Effect.provide(layer)), ); +it.live.each([ + { budget: "agent", limit: 8 }, + { budget: "server", limit: 32 }, +])("opens and popups reserve the $budget budget before session creation", ({ budget, limit }) => + Effect.scoped( + Effect.gen(function* () { + const { broker, browser, tabId } = yield* ready; + const manager = yield* Manager.PreviewManager; + const open = (target = scope) => + broker.invoke({ + scope: target, + operation: "open", + input: { reuseExistingTab: false, show: false }, + }); + for (let index = 1; index < limit - 3; index++) { + yield* open(budget === "agent" ? scope : asSession(`other-${Math.floor(index / 8)}`)); + } + + const sessionGate = Promise.withResolvers(); + const launchGate = (contextGate = Promise.withResolvers()); + const sessionStarted = Promise.withResolvers(); + const realOpen = manager.open; + const delayedOpen = vi.spyOn(manager, "open").mockImplementation((input) => + Effect.promise(() => { + sessionStarted.resolve(); + return sessionGate.promise; + }).pipe(Effect.andThen(realOpen(input))), + ); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + sessionGate.resolve(); + launchGate.resolve(); + delayedOpen.mockRestore(); + }), + ); + const events = yield* manager.subscribeEvents; + const first = yield* open().pipe(Effect.result, Effect.forkScoped); + yield* Effect.promise(() => sessionStarted.promise); + const popup = makeContext(); + const newCDPSession = popup.newCDPSession; + popup.newCDPSession = async (target) => { + await launchGate.promise; + return newCDPSession(target); + }; + contexts[0]!.page.emit("popup", popup.page); + const last = yield* open().pipe(Effect.result, Effect.forkScoped); + const overflow = yield* open().pipe(Effect.result, Effect.forkScoped); + // The host processes this after the opens, while manager creation is blocked. + const reused = yield* broker.invoke({ + scope, + tabId, + operation: "open", + input: { show: false }, + }); + expect(reused.tabId).toBe(tabId); + expect((yield* manager.list({ threadId: scope.thread.threadId })).sessions).toHaveLength( + limit - 3, + ); + + sessionGate.resolve(); + for (let count = 0; count < 3; count++) { + let event = yield* PubSub.take(events); + while (event.type !== "opened") event = yield* PubSub.take(events); + } + // The same slots must stay reserved after publication, while pages are still opening. + const duringLaunch = yield* open().pipe(Effect.result, Effect.forkScoped); + const rejectedPopup = makeContext(); + contexts[0]!.page.emit("popup", rejectedPopup.page); + yield* broker.invoke({ scope, tabId, operation: "status", input: {} }); + launchGate.resolve(); + + expect((yield* Fiber.join(first))._tag).toBe("Success"); + expect((yield* Fiber.join(last))._tag).toBe("Success"); + for (const fiber of [overflow, duringLaunch]) { + expect(yield* Fiber.join(fiber)).toMatchObject({ + _tag: "Failure", + failure: { _tag: "PreviewAutomationControlInterruptedError", reason: "tabLimit" }, + }); + } + const { sessions } = yield* manager.list({ threadId: scope.thread.threadId }); + for (const session of sessions) + yield* browser.attachViewer(viewerInput(session.tabId, false)); + expect(sessions).toHaveLength(limit); + expect(popup.page.isClosed()).toBe(false); + expect(rejectedPopup.page.isClosed()).toBe(true); + const status = yield* broker.invoke({ + scope, + tabId, + operation: "status", + input: {}, + }); + expect(status.tabs).toHaveLength(limit); + }), + ).pipe(Effect.provide(layer)), +); + +it.live("failed session and browser launches release capacity, as does closing a tab", () => + Effect.scoped( + Effect.gen(function* () { + const { broker, tabId } = yield* ready; + const manager = yield* Manager.PreviewManager; + const open = broker.invoke({ + scope, + operation: "open", + input: { reuseExistingTab: false, show: false }, + }); + for (let count = 1; count < 7; count++) yield* open; + const failSession = vi + .spyOn(manager, "open") + .mockImplementationOnce(() => Effect.die(new Error("session creation failed"))); + yield* Effect.addFinalizer(() => Effect.sync(() => failSession.mockRestore())); + expect(yield* open.pipe(Effect.result)).toMatchObject({ _tag: "Failure" }); + failSession.mockRestore(); + const events = yield* manager.subscribeEvents; + contextFailure = new Error("Chromium launch failed"); + expect(yield* open.pipe(Effect.result)).toMatchObject({ _tag: "Failure" }); + let failedSession = yield* PubSub.take(events); + while (failedSession.type !== "opened") failedSession = yield* PubSub.take(events); + // A failed launch must not leave a session a viewer can relaunch past the budget. + expect((yield* manager.list({ threadId: scope.thread.threadId })).sessions).toHaveLength(7); + contextFailure = null; + expect((yield* open).available).toBe(true); + expect(yield* open.pipe(Effect.flip)).toMatchObject({ reason: "tabLimit" }); + expect( + yield* broker + .invoke({ + scope, + tabId: PreviewTabId.make(failedSession.tabId), + operation: "open", + input: { show: false }, + }) + .pipe(Effect.flip), + ).toMatchObject({ _tag: "PreviewAutomationTabNotFoundError" }); + const browser = yield* ServerBrowser.ServerBrowser; + expect( + yield* browser.attachViewer(viewerInput(failedSession.tabId, false)).pipe(Effect.flip), + ).toMatchObject({ _tag: "ServerBrowserTabNotFoundError" }); + yield* broker.invoke({ scope, tabId, operation: "close", input: {} }); + expect((yield* open).available).toBe(true); + const status = yield* broker.invoke({ + scope, + operation: "status", + input: {}, + }); + expect(status.tabs).toHaveLength(8); + }), + ).pipe(Effect.provide(layer)), +); + +it.live("an interrupted open holds capacity until launch settles and leaves no session", () => + Effect.scoped( + Effect.gen(function* () { + const { broker, browser } = yield* ready; + const manager = yield* Manager.PreviewManager; + const open = broker.invoke({ + scope, + operation: "open", + input: { reuseExistingTab: false, show: false }, + }); + for (let count = 1; count < 7; count++) yield* open; + const gate = (contextGate = Promise.withResolvers()); + const started = (contextStarted = Promise.withResolvers()); + yield* Effect.addFinalizer(() => Effect.sync(() => gate.resolve())); + const events = yield* manager.subscribeEvents; + const opening = yield* open.pipe(Effect.forkScoped); + let pending = yield* PubSub.take(events); + while (pending.type !== "opened") pending = yield* PubSub.take(events); + yield* Effect.promise(() => started.promise); + // This viewer joins the real launch promise before the request is interrupted. + const launchingViewer = yield* browser + .attachViewer(viewerInput(pending.tabId, false)) + .pipe(Effect.result, Effect.forkScoped({ startImmediately: true })); + yield* Fiber.interrupt(opening); + // Only the canceled launch stays gated; an incorrectly admitted retry can finish. + contextGate = null; + const duringLaunch = yield* open.pipe(Effect.result); + gate.resolve(); + expect(yield* Fiber.join(launchingViewer)).toMatchObject({ + _tag: "Failure", + failure: { _tag: "ServerBrowserTabNotFoundError" }, + }); + expect(duringLaunch).toMatchObject({ + _tag: "Failure", + failure: { _tag: "PreviewAutomationControlInterruptedError", reason: "tabLimit" }, + }); + // Attaching must not resurrect the interrupted request's session. + const attached = yield* browser + .attachViewer(viewerInput(pending.tabId, false)) + .pipe(Effect.result); + expect(attached).toMatchObject({ + _tag: "Failure", + failure: { _tag: "ServerBrowserTabNotFoundError" }, + }); + expect((yield* manager.list({ threadId: scope.thread.threadId })).sessions).toHaveLength(7); + expect((yield* open).available).toBe(true); + }), + ).pipe(Effect.provide(layer)), +); + it.live("an agent's tabs stop at the limit until an unwatched idle tab closes", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/preview/ServerBrowser.ts b/apps/server/src/preview/ServerBrowser.ts index d9626784c32d2..1993340ae21c1 100644 --- a/apps/server/src/preview/ServerBrowser.ts +++ b/apps/server/src/preview/ServerBrowser.ts @@ -462,13 +462,17 @@ const make = Effect.gen(function* () { const environment = yield* ServerEnvironment.ServerEnvironment; const previewBrowser = yield* PreviewBrowser.PreviewBrowser; const desktopChannel = yield* DesktopBrowserChannel.DesktopBrowserChannel; - const runFork = Effect.runForkWith(yield* Effect.context()); + const services = yield* Effect.context(); + const runFork = Effect.runForkWith(services); + const runPromise = Effect.runPromiseWith(services); const launchServices = yield* Effect.context(); const setupCommand = yield* resolveRootCliCommand(PreviewBrowserHost.SETUP_SUBCOMMAND); const tabs = new Map(); - const pendingTabs = new Map>(); - + const pendingTabs = new Map< + string | symbol, + { readonly owner: string | null; promise?: Promise } + >(); const closedPendingTabs = new Set(); const adoptedPages = new Map(); const closedSessions = new Set(); @@ -886,6 +890,7 @@ const make = Effect.gen(function* () { page.on("close", () => dropTab(tab, true)); page.on("crash", () => dropTab(tab, true)); tabs.set(key, tab); + pendingTabs.delete(key); reportLiveTabs(); // Existing desktop and popup PDFs can start loading before interception attaches. if ( @@ -1177,8 +1182,7 @@ const make = Effect.gen(function* () { }; const adoptPopup = async (opener: ServerTab, popup: Page) => { - const agentId = opener.control.agentId; - if (opener.closing || (agentId !== null && atTabLimit(agentId))) { + if (opener.closing) { await popup.close().catch(constVoid); return; } @@ -1187,25 +1191,26 @@ const make = Effect.gen(function* () { const controller = [...opener.viewers].find( (viewer) => viewer.id === opener.control.controller, ); - await Effect.runPromise( - manager.open({ - threadId: opener.threadId, - ...(/^https?:/i.test(url) ? { url } : {}), - runtime: "server", - ...(opener.profileId === undefined ? {} : { profileId: opener.profileId }), - ...(opener.control.agentId === null - ? {} - : { automationOwner: opener.control.agentId, reveal: false }), - beforePublish: (snapshot) => - adoptedPages.set(tabKey(snapshot.threadId, snapshot.tabId), { - page: popup, - openerTabId: opener.tabId, - }), - }), + await runPromise( + Effect.scoped( + openTab({ + threadId: opener.threadId, + ...(/^https?:/i.test(url) ? { url } : {}), + runtime: "server", + ...(opener.profileId === undefined ? {} : { profileId: opener.profileId }), + ...(opener.control.agentId === null + ? {} + : { automationOwner: opener.control.agentId, reveal: false }), + beforePublish: (snapshot) => + adoptedPages.set(tabKey(snapshot.threadId, snapshot.tabId), { + page: popup, + openerTabId: opener.tabId, + }), + }), + ), ).then( - (snapshot) => { - if (opener.control.agentId === null) - controller?.push({ _tag: "popup", tabId: snapshot.tabId }); + (tab) => { + if (opener.control.agentId === null) controller?.push({ _tag: "popup", tabId: tab.tabId }); }, async () => { await popup.close().catch(constVoid); @@ -1216,9 +1221,11 @@ const make = Effect.gen(function* () { const ensureTab = (snapshot: PreviewSessionSnapshot): Promise => { const key = tabKey(snapshot.threadId, snapshot.tabId); const pending = pendingTabs.get(key); - if (pending) return pending; + if (pending?.promise) return pending.promise; const existing = tabs.get(key); if (existing) return Promise.resolve(existing); + const launch = pending ?? { owner: snapshot.automationOwner ?? null }; + pendingTabs.set(key, launch); const opening = createTab(snapshot) .catch((cause: unknown) => { runFork( @@ -1230,10 +1237,11 @@ const make = Effect.gen(function* () { throw cause; }) .finally(() => { + if (pendingTabs.get(key) !== launch) return; pendingTabs.delete(key); closedPendingTabs.delete(key); }); - pendingTabs.set(key, opening); + launch.promise = opening; return opening; }; @@ -1253,16 +1261,66 @@ const make = Effect.gen(function* () { const atTabLimit = (agentSessionId: string) => tabs.size + pendingTabs.size >= SERVER_TAB_LIMIT || - [...tabs.values()].filter((tab) => tab.control.agentId === agentSessionId).length >= + [...tabs.values()].filter((tab) => tab.control.agentId === agentSessionId).length + + [...pendingTabs.values()].filter((launch) => launch.owner === agentSessionId).length >= AGENT_TAB_LIMIT; - const assertTabCapacity = (agentSessionId: string) => { - if (atTabLimit(agentSessionId)) - throw new BrowserControlInterrupted( - `Too many server browser tabs are open (${AGENT_TAB_LIMIT} per agent session, ${SERVER_TAB_LIMIT} per server). Close one with t3_preview_close or reuse one from preview_status tabs.`, - "tabLimit", - ); - }; + const openTab = Effect.fnUntraced(function* (input: Parameters[0]) { + const reservation = yield* Effect.acquireRelease( + Effect.gen(function* () { + if (input.automationOwner !== undefined && atTabLimit(input.automationOwner)) + return yield* Effect.fail( + new BrowserControlInterrupted( + `Too many server browser tabs are open (${AGENT_TAB_LIMIT} per agent session, ${SERVER_TAB_LIMIT} per server). Close one with t3_preview_close or reuse one from preview_status tabs.`, + "tabLimit", + ), + ); + const reservation = { + key: Symbol() as string | symbol, + launch: { owner: input.automationOwner ?? null }, + }; + pendingTabs.set(reservation.key, reservation.launch); + return reservation; + }), + (reservation) => + Effect.sync(() => { + const pending = pendingTabs.get(reservation.key); + // An interrupted request still occupies capacity until its launch settles. + if (pending === reservation.launch && !pending.promise) + pendingTabs.delete(reservation.key); + }), + ); + const snapshot = yield* Effect.acquireRelease( + manager.open({ + ...input, + beforePublish: (snapshot) => { + pendingTabs.delete(reservation.key); + reservation.key = tabKey(snapshot.threadId, snapshot.tabId); + pendingTabs.set(reservation.key, reservation.launch); + input.beforePublish?.(snapshot); + }, + }), + (snapshot, exit) => + Exit.isFailure(exit) + ? Effect.gen(function* () { + const key = tabKey(snapshot.threadId, snapshot.tabId); + if (pendingTabs.get(key)?.promise) closedPendingTabs.add(key); + const tab = tabs.get(key); + if (tab) dropTab(tab, false); + yield* manager + .close({ + threadId: ThreadId.make(snapshot.threadId), + tabId: snapshot.tabId, + }) + .pipe(Effect.orDie); + }) + : Effect.void, + ); + return yield* Effect.tryPromise({ + try: () => ensureTab(snapshot), + catch: ServerBrowserPage.toOperationError, + }); + }); // Desktop browsing bypasses streamed viewers and server input tracking, // so these idle signals only apply to headless tabs. @@ -1704,93 +1762,89 @@ const make = Effect.gen(function* () { } }; - const runOperation = async (request: PreviewAutomationRequest): Promise => { - const input = request.input; - switch (request.operation) { - case "status": - return statusWithTitle( - request.tabId === undefined - ? latestThreadTab(request.threadId, request.agentSessionId) - : tabs.get(tabKey(request.threadId, request.tabId)), - request.agentSessionId, - ); - case "open": { - if (!request.agentSessionId) - throw new BrowserControlInterrupted( - "The agent session is missing. Reconnect the provider.", - ); - const open = input as PreviewAutomationOpenInput; - const url = open.url === undefined ? undefined : normalizePreviewUrl(open.url); - const reuse = open.reuseExistingTab ?? true; - if ( - reuse && - !request.tabIdExplicit && - [...tabs.values()].filter( - (tab) => - tab.threadId === request.threadId && tab.control.agentId === request.agentSessionId, - ).length > 1 - ) - throw new BrowserControlInterrupted( - "Multiple tabs are open. Pass tabId or reuseExistingTab=false.", - "tabRequired", - ); + const openAgentTab = Effect.fnUntraced(function* (request: PreviewAutomationRequest) { + if (!request.agentSessionId) + return yield* Effect.fail( + new BrowserControlInterrupted("The agent session is missing. Reconnect the provider."), + ); + const agentSessionId = request.agentSessionId; + const open = request.input as PreviewAutomationOpenInput; + const url = yield* Effect.try({ + try: () => (open.url === undefined ? undefined : normalizePreviewUrl(open.url)), + catch: ServerBrowserPage.toOperationError, + }); + const reuse = open.reuseExistingTab ?? true; + if ( + reuse && + !request.tabIdExplicit && + [...tabs.values()].filter( + (tab) => + tab.threadId === request.threadId && tab.control.agentId === request.agentSessionId, + ).length > 1 + ) + return yield* Effect.fail( + new BrowserControlInterrupted( + "Multiple tabs are open. Pass tabId or reuseExistingTab=false.", + "tabRequired", + ), + ); - const found = - reuse && request.tabId !== undefined - ? await Effect.runPromise( - findTab(request.threadId, request.tabId).pipe( - Effect.catchTags({ - ServerBrowserTabNotFoundError: () => Effect.succeed(undefined), - }), - ), - ) - : undefined; - if (reuse && request.tabIdExplicit && !found) { - const { sessions } = await Effect.runPromise( - manager.list({ threadId: ThreadId.make(request.threadId) }), - ); - const desktop = sessions.some( - (session) => session.tabId === request.tabId && session.runtime !== "server", - ); - throw new ServerBrowserPage.ServerBrowserOperationError( - desktop ? "PreviewAutomationExecutionError" : "PreviewAutomationTabNotFoundError", - desktop - ? "This tab belongs to the desktop runtime. This environment's server browser cannot control desktop tabs. Omit tabId to open a server tab instead." - : "The requested server preview tab does not exist.", - ); - } - // Only an explicit tabId reuses a tab this session did not open; a tab - // it last read (such as the user's) is not taken over implicitly. - const existing = - found && (request.tabIdExplicit || found.control.agentId === request.agentSessionId) - ? found - : undefined; - const navigationTimeout = Math.min(request.timeoutMs, NAVIGATION_TIMEOUT_MS); - const profileId = existing ? undefined : resolveOpenProfile(open.profileId); - if (!existing) { - closeIdleAgentTabs(); - assertTabCapacity(request.agentSessionId); - } - const tab = - existing ?? - (await ensureTab( - await Effect.runPromise( - manager.open({ - threadId: request.threadId, - ...(url ? { url } : {}), - ...(profileId === undefined ? {} : { profileId }), - runtime: "server", - reveal: false, - automationOwner: request.agentSessionId, - }), - ), - )); - if (existing?.dialog) - throw new BrowserControlInterrupted( - "A browser dialog is pending. Read preview_status and use preview_dialog first.", - "dialogPending", - ); - return tab.control.agent(request.agentSessionId, async () => { + const found = + reuse && request.tabId !== undefined + ? yield* findTab(request.threadId, request.tabId).pipe( + Effect.catchTags({ + ServerBrowserTabNotFoundError: () => Effect.succeed(undefined), + }), + ) + : undefined; + if (reuse && request.tabIdExplicit && !found) { + const { sessions } = yield* manager.list({ threadId: ThreadId.make(request.threadId) }); + const desktop = sessions.some( + (session) => session.tabId === request.tabId && session.runtime !== "server", + ); + return yield* Effect.fail( + new ServerBrowserPage.ServerBrowserOperationError( + desktop ? "PreviewAutomationExecutionError" : "PreviewAutomationTabNotFoundError", + desktop + ? "This tab belongs to the desktop runtime. This environment's server browser cannot control desktop tabs. Omit tabId to open a server tab instead." + : "The requested server preview tab does not exist.", + ), + ); + } + // Only an explicit tabId reuses a tab this session did not open; a tab + // it last read (such as the user's) is not taken over implicitly. + const existing = + found && (request.tabIdExplicit || found.control.agentId === request.agentSessionId) + ? found + : undefined; + const navigationTimeout = Math.min(request.timeoutMs, NAVIGATION_TIMEOUT_MS); + const profileId = existing + ? undefined + : yield* Effect.try({ + try: () => resolveOpenProfile(open.profileId), + catch: ServerBrowserPage.toOperationError, + }); + if (!existing) closeIdleAgentTabs(); + const tab = + existing ?? + (yield* openTab({ + threadId: request.threadId, + ...(url ? { url } : {}), + ...(profileId === undefined ? {} : { profileId }), + runtime: "server", + reveal: false, + automationOwner: request.agentSessionId, + })); + if (existing?.dialog) + return yield* Effect.fail( + new BrowserControlInterrupted( + "A browser dialog is pending. Read preview_status and use preview_dialog first.", + "dialogPending", + ), + ); + return yield* Effect.tryPromise({ + try: () => + tab.control.agent(agentSessionId, async () => { if (tab.dialog) throw new BrowserControlInterrupted( "A browser dialog is pending. Read preview_status and use preview_dialog first.", @@ -1803,7 +1857,7 @@ const make = Effect.gen(function* () { } const reveal = open.open ?? open.show; if (reveal !== false) { - await Effect.runPromise( + await runPromise( manager.requestReveal({ threadId: tab.threadId, tabId: tab.tabId, @@ -1817,8 +1871,21 @@ const make = Effect.gen(function* () { .catch(constVoid); } return statusWithTitle(tab, request.agentSessionId); - }); - } + }), + catch: ServerBrowserPage.toOperationError, + }); + }, Effect.scoped); + + const runOperation = async (request: PreviewAutomationRequest): Promise => { + const input = request.input; + switch (request.operation) { + case "status": + return statusWithTitle( + request.tabId === undefined + ? latestThreadTab(request.threadId, request.agentSessionId) + : tabs.get(tabKey(request.threadId, request.tabId)), + request.agentSessionId, + ); case "recordingStop": { if ( !request.tabIdExplicit && @@ -2001,41 +2068,54 @@ 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 }), - }, - }), - }), - Effect.flatMap((outcome) => - broker.respond({ - clientId: SERVER_HOST_CLIENT_ID, - connectionId, - requestId: request.requestId, - ...outcome, + broker.handleRequest( + { clientId: SERVER_HOST_CLIENT_ID, connectionId, requestId: request.requestId }, + (request.operation === "open" + ? openAgentTab(request).pipe( + Effect.catchDefect((cause) => Effect.fail(ServerBrowserPage.toOperationError(cause))), + ) + : Effect.tryPromise({ + try: () => runOperation(request), + catch: ServerBrowserPage.toOperationError, + }) + ).pipe( + Effect.ensuring(Effect.sync(() => markUsed(request))), + Effect.mapError(ServerBrowserPage.toOperationError), + 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.flatMap((outcome) => + broker.respond({ + clientId: SERVER_HOST_CLIENT_ID, + connectionId, + requestId: request.requestId, + ...outcome, + }), + ), + Effect.ignore, ), - Effect.ignore, ); const mirrorManagerEvent = (event: PreviewEvent) => Effect.promise(async () => { if (event.type === "opened" && event.snapshot.runtime === "server") { - await ensureTab(event.snapshot).catch(constVoid); + // Agent opens own their launch and rollback; mirroring must not restart a canceled launch. + if (event.snapshot.automationOwner === undefined) + await ensureTab(event.snapshot).catch(constVoid); return; } const key = tabKey(event.threadId, event.tabId); const tab = tabs.get(key); - if (event.type === "closed" && !tab && pendingTabs.has(key)) closedPendingTabs.add(key); + if (event.type === "closed" && !tab && pendingTabs.get(key)?.promise) + closedPendingTabs.add(key); if (!tab) return; if (event.type === "closed") { closedSessions.add(key);