1import {
2 PREVIEW_AUTOMATION_V1_OPERATIONS,
3 PreviewAutomationClientDisconnectedError,
4 PreviewAutomationControlInterruptedError,
5 PreviewAutomationControlReason,
6 PreviewAutomationExecutionError,
7 SERVER_BROWSER_AUTOMATION_CLIENT_ID,
8 PreviewAutomationInvalidSelectorError,
9 PreviewAutomationMalformedResponseError,
10 PreviewAutomationNoAvailableHostError,
11 PreviewAutomationRemoteUnavailableError,
12 PreviewAutomationRecordingTransferError,
13 PreviewAutomationRecordingDesktopUpdateRequiredError,
14 PreviewAutomationRecordingTooLargeError,
15 PreviewAutomationRecordingDeadlineExpiredError,
16 PreviewAutomationRequestQueueClosedError,
17 PreviewAutomationResultTooLargeError,
18 PreviewAutomationTabNotFoundError,
19 PreviewAutomationTargetNotEditableError,
20 PreviewAutomationTimeoutError,
21 PreviewAutomationUnsupportedClientError,
22 PreviewTabId,
23 type PreviewAutomationError,
24 type PreviewAutomationOperation,
25 type PreviewAutomationHost,
26 type PreviewAutomationHostFocus,
27 type PreviewAutomationResponse,
28 type PreviewAutomationStreamEvent,
29} from "@t3tools/contracts";
30import * as Context from "effect/Context";
31import type * as Cause from "effect/Cause";
32import * as Crypto from "effect/Crypto";
33import * as Deferred from "effect/Deferred";
34import * as Effect from "effect/Effect";
35import * as Layer from "effect/Layer";
36import * as Option from "effect/Option";
37import * as Queue from "effect/Queue";
38import * as Schema from "effect/Schema";
39import * as Stream from "effect/Stream";
40import * as SynchronizedRef from "effect/SynchronizedRef";
41
42import * as McpInvocationContext from "./McpInvocationContext.ts";
43
44const HOST_RESPONSE_GRACE_MS = 1_000;
45
46export interface PreviewAutomationInvokeInput {
47 /** Preview tabs belong to a thread, so only thread callers reach the broker. */
48 readonly scope: McpInvocationContext.McpThreadInvocationScope;
49 readonly operation: PreviewAutomationOperation;
50 readonly input: unknown;
51 readonly tabId?: PreviewTabId;
52 readonly timeoutMs?: number;
53 /** Background metadata reads must not change the agent's current tab. */
54 readonly updateCurrentTab?: boolean;
55 /** Capture the routed tab before another request changes the current assignment. */
56 readonly onTargetTab?: (tabId: PreviewTabId | undefined) => void;
57}
58
59export interface PreviewAutomationConnectOptions {
60 /**
61 * New agent work goes to a preferred host before any desktop. The server's
62 * own headless browser registers this way so a standalone environment keeps
63 * browsing when every desktop disconnects.
64 */
65 readonly preferred?: boolean;
66}
67
68export class PreviewAutomationBroker extends Context.Service<
69 PreviewAutomationBroker,
70 {
71 readonly connect: (
72 host: PreviewAutomationHost,
73 options?: PreviewAutomationConnectOptions,
74 ) => Effect.Effect<Stream.Stream<PreviewAutomationStreamEvent>>;
75 readonly focusHost: (host: PreviewAutomationHostFocus) => Effect.Effect<void>;
76 readonly respond: (
77 response: PreviewAutomationResponse,
78 ) => Effect.Effect<void, PreviewAutomationError>;
79 readonly invoke: <A = unknown>(
80 request: PreviewAutomationInvokeInput,
81 ) => Effect.Effect<A, PreviewAutomationError>;
82 }
83>()("t3/mcp/PreviewAutomationBroker") {}
84
85interface ClientConnection {
86 readonly clientId: string;
87 readonly connectionId: string;
88 readonly environmentId: PreviewAutomationHost["environmentId"];
89 readonly supportedOperations: ReadonlySet<PreviewAutomationOperation>;
90 readonly focused: boolean;
91 readonly preferred: boolean;
92 readonly liveTabs: NonNullable<PreviewAutomationHostFocus["liveTabs"]>;
93 readonly focusOrder: number;
94 readonly queue: Queue.Queue<PreviewAutomationStreamEvent, Cause.Done>;
95}
96
97interface PendingRequest {
98 readonly queue: ClientConnection["queue"];
99 readonly deferred: Deferred.Deferred<unknown, PreviewAutomationError>;
100 readonly context: PreviewAutomationRequestErrorContext;
101}
102
103/**
104 * A lease pinning one provider session to one desktop runtime. It lives exactly
105 * as long as the connection it names: `connectionId`/`queue` identity is what
106 * makes a lease valid, so a disconnected or replaced host is dropped on the next
107 * lookup. The lease deliberately has no clock of its own — it used to inherit
108 * the MCP credential's expiry, which coupled host stickiness to an unrelated
109 * auth deadline and could migrate a live session to another runtime mid-flow.
110 */
111interface HostAssignment {
112 readonly clientId: ClientConnection["clientId"];
113 readonly connectionId: ClientConnection["connectionId"];
114 readonly queue: ClientConnection["queue"];
115 readonly tabId?: PreviewTabId;
116 readonly tabSequence?: number;
117}
118
119interface PreviewAutomationRequestErrorContext {
120 readonly operation: PreviewAutomationOperation;
121 readonly environmentId: McpInvocationContext.McpInvocationScope["environmentId"];
122 readonly threadId: McpInvocationContext.McpThreadCaller["threadId"];
123 readonly providerSessionId: string;
124 readonly providerInstanceId: McpInvocationContext.McpThreadCaller["providerInstanceId"];
125 readonly clientId: string;
126 readonly connectionId: ClientConnection["connectionId"];
127 readonly requestId: string;
128 readonly tabId?: PreviewTabId;
129 readonly timeoutMs: number;
130 readonly selectorKind?: "locator" | "selector";
131 readonly selectorLength?: number;
132}
133
134interface BrokerState {
135 readonly clients: ReadonlyMap<string, ClientConnection>;
136 readonly assignments: ReadonlyMap<string, HostAssignment>;
137 readonly pending: ReadonlyMap<string, PendingRequest>;
138 readonly requestSequence: number;
139 readonly focusSequence: number;
140}
141
142const removeConnectionFromState = (
143 current: BrokerState,
144 clientId: string,
145 queue: ClientConnection["queue"],
146): { readonly state: BrokerState; readonly disconnected: ReadonlyArray<PendingRequest> } => {
147 const clients = new Map(current.clients);
148 const assignments = new Map(current.assignments);
149 const pending = new Map(current.pending);
150 const disconnected: PendingRequest[] = [];
151 if (current.clients.get(clientId)?.queue === queue) clients.delete(clientId);
152 for (const [assignmentKey, assignment] of assignments) {
153 if (assignment.queue === queue) assignments.delete(assignmentKey);
154 }
155 for (const [requestId, entry] of pending) {
156 if (entry.queue !== queue) continue;
157 pending.delete(requestId);
158 disconnected.push(entry);
159 }
160 return {
161 state: { ...current, clients, assignments, pending },
162 disconnected,
163 };
164};
165
166const selectorDiagnosticsFromInput = (
167 input: unknown,
168): Pick<PreviewAutomationRequestErrorContext, "selectorKind" | "selectorLength"> => {
169 if (typeof input !== "object" || input === null) return {};
170 if ("locator" in input && typeof input.locator === "string") {
171 return { selectorKind: "locator", selectorLength: input.locator.length };
172 }
173 if ("selector" in input && typeof input.selector === "string") {
174 return { selectorKind: "selector", selectorLength: input.selector.length };
175 }
176 return {};
177};
178
179const hostAssignmentKey = (scope: McpInvocationContext.McpThreadInvocationScope): string =>
180 `${scope.environmentId}\u0000${scope.thread.providerSessionId}`;
181
182const isPreviewTabId = Schema.is(PreviewTabId);
183const decodeControlReason = Schema.decodeUnknownOption(PreviewAutomationControlReason);
184
185const readResultTabId = (result: unknown): PreviewTabId | null | undefined => {
186 if (typeof result !== "object" || result === null || !("tabId" in result)) return undefined;
187 const tabId = result.tabId;
188 return tabId === null || isPreviewTabId(tabId) ? tabId : undefined;
189};
190
191const supportsOperation = (
192 connection: ClientConnection,
193 operation: PreviewAutomationOperation,
194): boolean => connection.supportedOperations.has(operation);
195
196type RemoteDetailKind = "null" | "array" | "object" | "string" | "number" | "boolean";
197
198function remoteDetailKind(detail: unknown): RemoteDetailKind {
199 if (detail === null) return "null";
200 if (Array.isArray(detail)) return "array";
201 switch (typeof detail) {
202 case "string":
203 return "string";
204 case "number":
205 return "number";
206 case "boolean":
207 return "boolean";
208 default:
209 return "object";
210 }
211}
212
213/** Enough for a browser's own error line, such as a refused connection and its URL. */
214const MAX_REASON_CHARS = 500;
215
216const classifyResponseError = (
217 context: PreviewAutomationRequestErrorContext,
218 error: NonNullable<PreviewAutomationResponse["error"]>,
219): PreviewAutomationError => {
220 const remoteDiagnostics = {
221 remoteTag: error._tag,
222 remoteMessageLength: error.message.length,
223 ...(error.detail === undefined ? {} : { remoteDetailKind: remoteDetailKind(error.detail) }),
224 cause: error,
225 };
226 switch (error._tag) {
227 case "PreviewAutomationRecordingDesktopUpdateRequiredError":
228 return new PreviewAutomationRecordingDesktopUpdateRequiredError({
229 threadId: context.threadId,
230 cause: error,
231 });
232 case "PreviewAutomationRecordingTooLargeError":
233 return new PreviewAutomationRecordingTooLargeError({
234 threadId: context.threadId,
235 cause: error,
236 });
237 case "PreviewAutomationRecordingDeadlineExpiredError":
238 return new PreviewAutomationRecordingDeadlineExpiredError({
239 threadId: context.threadId,
240 cause: error,
241 });
242 case "PreviewAutomationRecordingTransferError":
243 return new PreviewAutomationRecordingTransferError({
244 threadId: context.threadId,
245 cause: error,
246 });
247 case "PreviewAutomationNoAvailableHostError":
248 return new PreviewAutomationNoAvailableHostError({
249 ...context,
250 ...remoteDiagnostics,
251 });
252 case "PreviewAutomationUnsupportedClientError":
253 return new PreviewAutomationUnsupportedClientError({
254 ...context,
255 ...remoteDiagnostics,
256 });
257 case "PreviewAutomationTabNotFoundError":
258 return new PreviewAutomationTabNotFoundError({
259 ...context,
260 ...remoteDiagnostics,
261 });
262 case "PreviewAutomationTimeoutError":
263 return new PreviewAutomationTimeoutError({
264 ...context,
265 ...remoteDiagnostics,
266 });
267 case "PreviewAutomationControlInterruptedError": {
268 const reason = decodeControlReason(error.detail);
269 return new PreviewAutomationControlInterruptedError({
270 ...context,
271 ...remoteDiagnostics,
272 ...(Option.isSome(reason) ? { reason: reason.value } : {}),
273 });
274 }
275 case "PreviewAutomationInvalidSelectorError": {
276 const staleRef =
277 typeof error.detail === "object" &&
278 error.detail !== null &&
279 "staleRef" in error.detail &&
280 error.detail.staleRef === true;
281 return new PreviewAutomationInvalidSelectorError({
282 ...context,
283 ...remoteDiagnostics,
284 ...(staleRef ? { staleRef } : {}),
285 });
286 }
287 case "PreviewAutomationTargetNotEditableError": {
288 const detail =
289 typeof error.detail === "object" && error.detail !== null ? error.detail : undefined;
290 const remoteSelectorKind =
291 detail &&
292 "selectorKind" in detail &&
293 (detail.selectorKind === "focused-element" ||
294 detail.selectorKind === "locator" ||
295 detail.selectorKind === "selector")
296 ? detail.selectorKind
297 : undefined;
298 const remoteSelectorLength =
299 detail &&
300 "selectorLength" in detail &&
301 typeof detail.selectorLength === "number" &&
302 Number.isInteger(detail.selectorLength) &&
303 detail.selectorLength >= 0
304 ? detail.selectorLength
305 : undefined;
306 return new PreviewAutomationTargetNotEditableError({
307 ...context,
308 ...remoteDiagnostics,
309 ...(remoteSelectorKind === undefined && context.selectorKind === undefined
310 ? {}
311 : { selectorKind: remoteSelectorKind ?? context.selectorKind }),
312 ...(remoteSelectorLength === undefined && context.selectorLength === undefined
313 ? {}
314 : { selectorLength: remoteSelectorLength ?? context.selectorLength }),
315 });
316 }
317 case "PreviewAutomationResultTooLargeError": {
318 const detail =
319 typeof error.detail === "object" && error.detail !== null ? error.detail : undefined;
320 const maximumBytes =
321 detail &&
322 "maximumBytes" in detail &&
323 typeof detail.maximumBytes === "number" &&
324 Number.isInteger(detail.maximumBytes) &&
325 detail.maximumBytes > 0
326 ? detail.maximumBytes
327 : undefined;
328 return new PreviewAutomationResultTooLargeError({
329 ...context,
330 ...remoteDiagnostics,
331 ...(maximumBytes === undefined ? {} : { maximumBytes }),
332 });
333 }
334 case "PreviewAutomationUnavailableError":
335 return new PreviewAutomationRemoteUnavailableError({
336 ...context,
337 ...remoteDiagnostics,
338 });
339 default:
340 return new PreviewAutomationExecutionError({
341 ...context,
342 ...remoteDiagnostics,
343 // The server's own browser writes these; other hosts' text stays out of the agent's context.
344 ...(context.clientId === SERVER_BROWSER_AUTOMATION_CLIENT_ID
345 ? { reason: error.message.slice(0, MAX_REASON_CHARS) }
346 : {}),
347 });
348 }
349};
350
351export const make = Effect.gen(function* PreviewAutomationBrokerMake() {
352 const crypto = yield* Crypto.Crypto;
353 const state = yield* SynchronizedRef.make<BrokerState>({
354 clients: new Map(),
355 assignments: new Map(),
356 pending: new Map(),
357 requestSequence: 0,
358 focusSequence: 0,
359 });
360
361 const closeConnection = Effect.fn("PreviewAutomationBroker.closeConnection")(function* (
362 queue: ClientConnection["queue"],
363 disconnected: ReadonlyArray<PendingRequest>,
364 completeStream = false,
365 ) {
366 if (completeStream) {
367 // Discard this generation's commands and complete the RPC stream so a
368 // responsive desktop can re-register after a timeout eviction.
369 yield* Queue.clear(queue);
370 yield* Queue.end(queue);
371 } else {
372 // Replaced registrations must not reconnect and displace their successor.
373 yield* Queue.shutdown(queue);
374 }
375 yield* Effect.forEach(
376 disconnected,
377 ({ deferred, context }) =>
378 Deferred.fail(deferred, new PreviewAutomationClientDisconnectedError(context)),
379 { discard: true },
380 );
381 });
382
383 const disconnect = Effect.fn("PreviewAutomationBroker.disconnect")(function* (
384 clientId: string,
385 queue: ClientConnection["queue"],
386 completeStream = false,
387 ) {
388 yield* SynchronizedRef.modifyEffect(state, (current) => {
389 // Retired generations were already closed by their replacement or eviction.
390 if (current.clients.get(clientId)?.queue !== queue) {
391 return Effect.succeed([undefined, current] as const);
392 }
393 const removed = removeConnectionFromState(current, clientId, queue);
394 return closeConnection(queue, removed.disconnected, completeStream).pipe(
395 Effect.as([undefined, removed.state] as const),
396 );
397 });
398 });
399
400 const acquireConnection = Effect.fn("PreviewAutomationBroker.acquireConnection")(function* (
401 host: PreviewAutomationHost,
402 options: PreviewAutomationConnectOptions | undefined,
403 ) {
404 const clientId = host.clientId;
405 const queue = yield* Queue.unbounded<PreviewAutomationStreamEvent, Cause.Done>();
406 const connectionId = yield* crypto.randomUUIDv4().pipe(Effect.orDie);
407 yield* Queue.offer(queue, { type: "connected", connectionId });
408 const connection: ClientConnection = {
409 clientId,
410 connectionId,
411 environmentId: host.environmentId,
412 supportedOperations: new Set(host.supportedOperations ?? PREVIEW_AUTOMATION_V1_OPERATIONS),
413 focused: false,
414 preferred: options?.preferred ?? false,
415 liveTabs: [],
416 focusOrder: 0,
417 queue,
418 };
419 const registration = yield* SynchronizedRef.modify(state, (current) => {
420 const previousConnection = current.clients.get(clientId);
421 const removed = previousConnection
422 ? removeConnectionFromState(current, clientId, previousConnection.queue)
423 : { state: current, disconnected: [] };
424 const clients = new Map(removed.state.clients);
425 const focusSequence = removed.state.focusSequence + 1;
426 const registeredConnection = { ...connection, focusOrder: focusSequence };
427 clients.set(clientId, registeredConnection);
428 return [
429 {
430 previousConnection,
431 disconnected: removed.disconnected,
432 registeredConnection,
433 },
434 { ...removed.state, clients, focusSequence },
435 ] as const;
436 });
437 if (registration.previousConnection) {
438 yield* closeConnection(registration.previousConnection.queue, registration.disconnected);
439 }
440 return registration.registeredConnection;
441 });
442
443 const connect: PreviewAutomationBroker["Service"]["connect"] = Effect.fn(
444 "PreviewAutomationBroker.connect",
445 )((host, options) =>
446 Effect.succeed(
447 Stream.unwrap(
448 Effect.acquireRelease(acquireConnection(host, options), (connection) =>
449 disconnect(connection.clientId, connection.queue),
450 ).pipe(Effect.map((connection) => Stream.fromQueue(connection.queue))),
451 ),
452 ),
453 );
454
455 const focusHost: PreviewAutomationBroker["Service"]["focusHost"] = Effect.fn(
456 "PreviewAutomationBroker.focusHost",
457 )(function* (host) {
458 yield* SynchronizedRef.update(state, (current) => {
459 const currentHost = current.clients.get(host.clientId);
460 if (
461 !currentHost ||
462 currentHost.environmentId !== host.environmentId ||
463 currentHost.connectionId !== host.connectionId
464 ) {
465 return current;
466 }
467 const clients = new Map(current.clients);
468 const focusSequence = host.focused ? current.focusSequence + 1 : current.focusSequence;
469 clients.set(host.clientId, {
470 ...currentHost,
471 focused: host.focused,
472 liveTabs: host.liveTabs ?? currentHost.liveTabs,
473 focusOrder: host.focused ? focusSequence : currentHost.focusOrder,
474 });
475 return { ...current, clients, focusSequence };
476 });
477 });
478
479 const respond: PreviewAutomationBroker["Service"]["respond"] = Effect.fn(
480 "PreviewAutomationBroker.respond",
481 )(function* (response) {
482 const pending = yield* SynchronizedRef.modify(state, (current) => {
483 const entry = current.pending.get(response.requestId);
484 if (
485 !entry ||
486 entry.context.clientId !== response.clientId ||
487 entry.context.connectionId !== response.connectionId
488 ) {
489 return [undefined, current] as const;
490 }
491 const next = new Map(current.pending);
492 next.delete(response.requestId);
493 return [entry, { ...current, pending: next }] as const;
494 });
495 if (!pending) return;
496 if (response.ok) {
497 yield* Deferred.succeed(pending.deferred, response.result);
498 } else {
499 yield* Deferred.fail(
500 pending.deferred,
501 response.error
502 ? classifyResponseError(pending.context, response.error)
503 : new PreviewAutomationMalformedResponseError(pending.context),
504 );
505 }
506 });
507
508 const invoke = Effect.fn("PreviewAutomationBroker.invoke")(function* <A = unknown>(
509 input: Parameters<PreviewAutomationBroker["Service"]["invoke"]>[0],
510 ): Effect.fn.Return<A, PreviewAutomationError> {
511 const deferred = yield* Deferred.make<unknown, PreviewAutomationError>();
512 const route = yield* SynchronizedRef.modify(state, (current) => {
513 const assignments = new Map(
514 Array.from(current.assignments).filter(([, assignment]) => {
515 const connection = current.clients.get(assignment.clientId);
516 return (
517 connection?.connectionId === assignment.connectionId &&
518 connection.queue === assignment.queue
519 );
520 }),
521 );
522 const assignmentKey = hostAssignmentKey(input.scope);
523 const assigned = assignments.get(assignmentKey);
524 const assignedConnection = assigned ? current.clients.get(assigned.clientId) : undefined;
525 const hasLiveAssignment = assignedConnection?.environmentId === input.scope.environmentId;
526 // Keep one provider session on one physical desktop runtime so a
527 // multi-step browser interaction cannot jump between independent
528 // Electron cookie/DOM state. A live assignment that predates an
529 // operation is not silently moved to a newer client: the caller gets a
530 // capability failure and can deliberately start a fresh provider
531 // session. A dead lease is pruned above and may fail over.
532 const ownsTargetTab = (host: ClientConnection, visibleOnly = false) =>
533 host.liveTabs.some(
534 (tab) =>
535 tab.threadId === input.scope.thread.threadId &&
536 (!visibleOnly || tab.visible === true) &&
537 (input.tabId === undefined || tab.tabId === input.tabId),
538 );
539 const connection =
540 hasLiveAssignment && supportsOperation(assignedConnection, input.operation)
541 ? assignedConnection
542 : hasLiveAssignment
543 ? undefined
544 : Array.from(current.clients.values())
545 .filter(
546 (host) =>
547 host.environmentId === input.scope.environmentId &&
548 supportsOperation(host, input.operation),
549 )
550 .sort(
551 (left, right) =>
552 Number(input.tabId !== undefined && ownsTargetTab(right)) -
553 Number(input.tabId !== undefined && ownsTargetTab(left)) ||
554 Number(right.preferred) - Number(left.preferred) ||
555 Number(ownsTargetTab(right, true)) - Number(ownsTargetTab(left, true)) ||
556 Number(ownsTargetTab(right)) - Number(ownsTargetTab(left)) ||
557 Number(right.focused) - Number(left.focused) ||
558 right.focusOrder - left.focusOrder,
559 )[0];
560 if (!connection) {
561 if (!hasLiveAssignment) assignments.delete(assignmentKey);
562 return [undefined, { ...current, assignments }] as const;
563 }
564 // The environment host may install Chromium on its first open (up to
565 // ten minutes). Keep that request alive without replaying its effects.
566 const timeoutMs =
567 input.timeoutMs ?? (input.operation === "open" && connection.preferred ? 660_000 : 15_000);
568 const canReuseAssignedTab =
569 assigned !== undefined &&
570 assigned.connectionId === connection.connectionId &&
571 assigned.queue === connection.queue;
572 assignments.set(assignmentKey, {
573 clientId: connection.clientId,
574 connectionId: connection.connectionId,
575 queue: connection.queue,
576 ...(canReuseAssignedTab && assigned.tabId !== undefined ? { tabId: assigned.tabId } : {}),
577 ...(canReuseAssignedTab && assigned.tabSequence !== undefined
578 ? { tabSequence: assigned.tabSequence }
579 : {}),
580 });
581
582 const requestSequence = current.requestSequence;
583 const requestId = `preview-${requestSequence}`;
584 const tabId = input.tabId ?? (canReuseAssignedTab ? assigned.tabId : undefined);
585 const selectorDiagnostics = selectorDiagnosticsFromInput(input.input);
586 const context: PreviewAutomationRequestErrorContext = {
587 operation: input.operation,
588 environmentId: input.scope.environmentId,
589 threadId: input.scope.thread.threadId,
590 providerSessionId: input.scope.thread.providerSessionId,
591 providerInstanceId: input.scope.thread.providerInstanceId,
592 clientId: connection.clientId,
593 connectionId: connection.connectionId,
594 requestId,
595 ...(tabId === undefined ? {} : { tabId }),
596 timeoutMs,
597 ...selectorDiagnostics,
598 };
599 const pending = new Map(current.pending);
600 pending.set(requestId, { queue: connection.queue, deferred, context });
601 return [
602 { connection, requestId, requestContext: context, requestSequence },
603 { ...current, assignments, pending, requestSequence: current.requestSequence + 1 },
604 ] as const;
605 });
606 if (!route) {
607 return yield* new PreviewAutomationNoAvailableHostError({
608 operation: input.operation,
609 environmentId: input.scope.environmentId,
610 threadId: input.scope.thread.threadId,
611 providerSessionId: input.scope.thread.providerSessionId,
612 providerInstanceId: input.scope.thread.providerInstanceId,
613 });
614 }
615 const { connection, requestId, requestContext, requestSequence } = route;
616 const { timeoutMs } = requestContext;
617 input.onTargetTab?.(requestContext.tabId);
618 const removePending = SynchronizedRef.update(state, (next) => {
619 if (!next.pending.has(requestId)) return next;
620 const pending = new Map(next.pending);
621 pending.delete(requestId);
622 return { ...next, pending };
623 });
624 const awaitResponse = Effect.fn("PreviewAutomationBroker.awaitResponse")(function* () {
625 const offered = yield* SynchronizedRef.modifyEffect(state, (current) => {
626 // A route can outlive its generation while another request evicts it.
627 // Serialize the live-generation check and offer with queue closure.
628 if (
629 current.clients.get(connection.clientId)?.queue !== connection.queue ||
630 !current.pending.has(requestId)
631 ) {
632 return Effect.succeed([false, current] as const);
633 }
634 return Queue.offer(connection.queue, {
635 type: "request",
636 connectionId: connection.connectionId,
637 request: {
638 requestId,
639 threadId: input.scope.thread.threadId,
640 tabId: requestContext.tabId,
641 tabIdExplicit: input.tabId !== undefined,
642 agentSessionId: hostAssignmentKey(input.scope),
643 operation: input.operation,
644 input: input.input,
645 timeoutMs,
646 },
647 }).pipe(Effect.map((offered) => [offered, current] as const));
648 });
649 if (!offered) {
650 const completion = yield* Deferred.poll(deferred);
651 if (Option.isSome(completion)) {
652 return (yield* completion.value) as A;
653 }
654 return yield* new PreviewAutomationRequestQueueClosedError(requestContext);
655 }
656 // The browser starts its operation timer after delivery. Allow its timeout
657 // response to arrive before treating the entire host as unresponsive.
658 const responseTimeoutMs =
659 input.updateCurrentTab === false ? timeoutMs : timeoutMs + HOST_RESPONSE_GRACE_MS;
660 const result = yield* Deferred.await(deferred).pipe(Effect.timeoutOption(responseTimeoutMs));
661 return yield* Option.match(result, {
662 onNone: () =>
663 Effect.gen(function* () {
664 // An unanswered request invalidates this connection. Do not replay
665 // actions: the client may have applied them before becoming unreachable.
666 // A background metadata read has a short budget and changes nothing,
667 // so a slow one must not cut the host off from the agent's next call.
668 if (input.updateCurrentTab !== false) {
669 yield* disconnect(connection.clientId, connection.queue, true);
670 }
671 return yield* new PreviewAutomationTimeoutError(requestContext);
672 }),
673 onSome: (value) => Effect.succeed(value as A),
674 });
675 });
676 const result = yield* awaitResponse().pipe(Effect.ensuring(removePending));
677 if (input.updateCurrentTab === false) return result;
678 const responseTabId = readResultTabId(result);
679 const resultTabId = responseTabId === undefined ? input.tabId : responseTabId;
680 if (resultTabId === undefined) return result;
681 const assignmentKey = hostAssignmentKey(input.scope);
682 yield* SynchronizedRef.update(state, (current) => {
683 const assignment = current.assignments.get(assignmentKey);
684 if (
685 !assignment ||
686 assignment.connectionId !== connection.connectionId ||
687 assignment.queue !== connection.queue ||
688 (assignment.tabSequence ?? -1) > requestSequence
689 ) {
690 return current;
691 }
692 const assignments = new Map(current.assignments);
693 if (resultTabId === null) {
694 const { tabId: _tabId, ...withoutTabId } = assignment;
695 assignments.set(assignmentKey, { ...withoutTabId, tabSequence: requestSequence });
696 } else {
697 assignments.set(assignmentKey, {
698 ...assignment,
699 ...(resultTabId === undefined ? {} : { tabId: resultTabId }),
700 tabSequence: requestSequence,
701 });
702 }
703 return { ...current, assignments };
704 });
705 return result;
706 });
707
708 return PreviewAutomationBroker.of({ connect, focusHost, respond, invoke });
709}).pipe(Effect.withSpan("PreviewAutomationBroker.make"));
710
711export const layer = Layer.effect(PreviewAutomationBroker, make);