1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import * as Layer from "effect/Layer";
4import { expect, it } from "@effect/vitest";
5import {
6 EnvironmentId,
7 PreviewAutomationClientDisconnectedError,
8 PreviewAutomationMalformedResponseError,
9 PreviewAutomationNoAvailableHostError,
10 PreviewTabId,
11 ProviderInstanceId,
12 ThreadId,
13 type PreviewAutomationHost,
14 type PreviewAutomationRequest,
15 type PreviewAutomationStreamEvent,
16 SERVER_BROWSER_AUTOMATION_CLIENT_ID,
17} from "@t3tools/contracts";
18import * as Effect from "effect/Effect";
19import * as Deferred from "effect/Deferred";
20import * as Fiber from "effect/Fiber";
21import * as Result from "effect/Result";
22import * as Stream from "effect/Stream";
23
24import { PreviewBrowserRootError } from "../preview/PreviewBrowserHost.ts";
25import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
26
27const makeBroker = PreviewAutomationBroker.make.pipe(
28 Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)),
29);
30
31const scope = {
32 environmentId: EnvironmentId.make("environment-1"),
33 requestNamespace: "provider-session-1",
34 thread: {
35 threadId: ThreadId.make("thread-1"),
36 providerSessionId: "provider-session-1",
37 providerInstanceId: ProviderInstanceId.make("codex"),
38 },
39 client: undefined,
40 capabilities: new Set(["preview"] as const),
41 issuedAt: 1,
42};
43
44const makeHost = (overrides: Partial<PreviewAutomationHost> = {}): PreviewAutomationHost => ({
45 clientId: "client-1",
46 environmentId: scope.environmentId,
47 ...overrides,
48});
49
50type RoutedRequest = PreviewAutomationRequest & {
51 readonly connectionId: PreviewAutomationStreamEvent["connectionId"];
52};
53
54const requestsFrom = (
55 events: Stream.Stream<PreviewAutomationStreamEvent>,
56 onConnected: (connectionId: PreviewAutomationStreamEvent["connectionId"]) => void = () => {},
57): Stream.Stream<RoutedRequest> =>
58 events.pipe(
59 Stream.filterMap((event) => {
60 if (event.type === "connected") {
61 onConnected(event.connectionId);
62 return Result.failVoid;
63 }
64 return Result.succeed({ ...event.request, connectionId: event.connectionId });
65 }),
66 );
67
68it.effect("atomically registers a connected host and correlates its response", () =>
69 Effect.gen(function* () {
70 const broker = yield* makeBroker;
71 const requests = requestsFrom(yield* broker.connect(makeHost()));
72 yield* Stream.runForEach(requests, (request) =>
73 broker.respond({
74 clientId: "client-1",
75 connectionId: request.connectionId,
76 requestId: request.requestId,
77 ok: true,
78 result: { available: true },
79 }),
80 ).pipe(Effect.forkScoped);
81 yield* Effect.yieldNow;
82
83 const result = yield* broker.invoke<{ available: boolean }>({
84 scope,
85 operation: "open",
86 input: {},
87 });
88
89 expect(result).toEqual({ available: true });
90 }),
91);
92
93it.effect("targets multiple tabs explicitly while retaining a default tab", () =>
94 Effect.gen(function* () {
95 const broker = yield* makeBroker;
96 const appTabId = PreviewTabId.make("tab-web-app");
97 const simulatorTabId = PreviewTabId.make("tab-ios-simulator");
98 const openedTabIds = [appTabId, simulatorTabId];
99 let openIndex = 0;
100 const routedRequests: RoutedRequest[] = [];
101 const requests = requestsFrom(yield* broker.connect(makeHost()));
102 yield* Stream.runForEach(requests, (request) => {
103 routedRequests.push(request);
104 return broker.respond({
105 clientId: "client-1",
106 connectionId: request.connectionId,
107 requestId: request.requestId,
108 ok: true,
109 result:
110 request.operation === "open"
111 ? { available: true, tabId: openedTabIds[openIndex++] }
112 : { url: "http://localhost:3200" },
113 });
114 }).pipe(Effect.forkScoped);
115 yield* Effect.yieldNow;
116
117 yield* broker.invoke({ scope, operation: "open", input: { reuseExistingTab: false } });
118 yield* broker.invoke({ scope, operation: "open", input: { reuseExistingTab: false } });
119 yield* broker.invoke({ scope, operation: "snapshot", input: {} });
120 yield* broker.invoke({ scope, operation: "snapshot", input: {}, tabId: appTabId });
121 yield* broker.invoke({ scope, operation: "snapshot", input: {} });
122
123 expect(routedRequests).toHaveLength(5);
124 expect(routedRequests[0]?.tabId).toBeUndefined();
125 expect(routedRequests[1]?.tabId).toBe(appTabId);
126 expect(routedRequests[2]?.tabId).toBe(simulatorTabId);
127 expect(routedRequests[2]?.tabIdExplicit).toBe(false);
128 expect(routedRequests[3]?.tabId).toBe(appTabId);
129 expect(routedRequests[3]?.tabIdExplicit).toBe(true);
130 expect(routedRequests[4]?.tabId).toBe(appTabId);
131 }),
132);
133
134it.effect.each([true, false])(
135 "keeps an older target stable while a newer explicit tab responds (implicit: %s)",
136 (implicit) =>
137 Effect.gen(function* () {
138 const broker = yield* makeBroker;
139 const olderTabId = PreviewTabId.make("tab-older-request");
140 const newerTabId = PreviewTabId.make("tab-newer-request");
141 const releaseOlderResponse = yield* Deferred.make<void>();
142 const routedRequests: RoutedRequest[] = [];
143 const requests = requestsFrom(yield* broker.connect(makeHost()));
144 yield* Stream.runForEach(requests, (request) => {
145 routedRequests.push(request);
146 const response = Effect.gen(function* () {
147 if (request.tabId === olderTabId && request.operation === "snapshot") {
148 yield* Deferred.await(releaseOlderResponse);
149 }
150 yield* broker.respond({
151 clientId: "client-1",
152 connectionId: request.connectionId,
153 requestId: request.requestId,
154 ok: true,
155 result: { url: "http://localhost:3200" },
156 });
157 if (request.tabId === newerTabId) {
158 yield* Deferred.succeed(releaseOlderResponse, undefined);
159 }
160 });
161 return response.pipe(Effect.forkScoped, Effect.asVoid);
162 }).pipe(Effect.forkScoped);
163 yield* Effect.yieldNow;
164
165 yield* broker.invoke({ scope, operation: "status", input: {}, tabId: olderTabId });
166 let capturedTabId: PreviewTabId | undefined;
167 const older = yield* broker
168 .invoke({
169 scope,
170 operation: "snapshot",
171 input: {},
172 ...(implicit ? {} : { tabId: olderTabId }),
173 onTargetTab: (tabId) => {
174 capturedTabId = tabId;
175 },
176 })
177 .pipe(Effect.forkScoped);
178 yield* Effect.yieldNow;
179 const newer = yield* broker
180 .invoke({ scope, operation: "snapshot", input: {}, tabId: newerTabId })
181 .pipe(Effect.forkScoped);
182 yield* Fiber.join(newer);
183 yield* Fiber.join(older);
184 yield* broker.invoke({
185 scope,
186 operation: "status",
187 input: {},
188 tabId: olderTabId,
189 updateCurrentTab: false,
190 });
191 yield* broker.invoke({ scope, operation: "snapshot", input: {} });
192
193 expect(routedRequests.at(-1)?.tabId).toBe(newerTabId);
194 expect(capturedTabId).toBe(olderTabId);
195 }),
196);
197
198it.effect("does not let a no-tab response suppress an earlier tab decision", () =>
199 Effect.gen(function* () {
200 const broker = yield* makeBroker;
201 const initialTabId = PreviewTabId.make("tab-initial");
202 const openedTabId = PreviewTabId.make("tab-opened-late");
203 const releaseOpenResponse = yield* Deferred.make<void>();
204 const routedRequests: RoutedRequest[] = [];
205 const requests = requestsFrom(yield* broker.connect(makeHost()));
206 yield* Stream.runForEach(requests, (request) => {
207 routedRequests.push(request);
208 const marker =
209 typeof request.input === "object" && request.input !== null && "marker" in request.input
210 ? request.input.marker
211 : undefined;
212 const response = Effect.gen(function* () {
213 if (marker === "older") {
214 yield* Deferred.await(releaseOpenResponse);
215 }
216 yield* broker.respond({
217 clientId: "client-1",
218 connectionId: request.connectionId,
219 requestId: request.requestId,
220 ok: true,
221 result:
222 request.operation === "open"
223 ? { available: true, tabId: marker === "older" ? openedTabId : initialTabId }
224 : { url: "http://localhost:3200" },
225 });
226 if (marker === "newer") {
227 yield* Deferred.succeed(releaseOpenResponse, undefined);
228 }
229 });
230 return response.pipe(Effect.forkScoped, Effect.asVoid);
231 }).pipe(Effect.forkScoped);
232 yield* Effect.yieldNow;
233
234 yield* broker.invoke({ scope, operation: "open", input: {} });
235 const older = yield* broker
236 .invoke({
237 scope,
238 operation: "open",
239 input: { marker: "older", reuseExistingTab: false },
240 })
241 .pipe(Effect.forkScoped);
242 yield* Effect.yieldNow;
243 const newer = yield* broker
244 .invoke({ scope, operation: "snapshot", input: { marker: "newer" } })
245 .pipe(Effect.forkScoped);
246 yield* Fiber.join(newer);
247 yield* Fiber.join(older);
248 yield* broker.invoke({ scope, operation: "snapshot", input: {} });
249
250 expect(routedRequests.at(-1)?.tabId).toBe(openedTabId);
251 }),
252);
253
254it.effect("announces a live replacement stream before delivering requests", () =>
255 Effect.gen(function* () {
256 const broker = yield* makeBroker;
257 const events = yield* broker.connect(makeHost());
258 const receivedTypes: PreviewAutomationStreamEvent["type"][] = [];
259 const consumer = yield* events.pipe(
260 Stream.take(2),
261 Stream.runForEach((event) => {
262 receivedTypes.push(event.type);
263 return event.type === "connected"
264 ? Effect.void
265 : broker.respond({
266 clientId: "client-1",
267 connectionId: event.connectionId,
268 requestId: event.request.requestId,
269 ok: true,
270 result: "ready",
271 });
272 }),
273 Effect.forkScoped,
274 );
275 yield* Effect.yieldNow;
276
277 const result = yield* broker.invoke<string>({ scope, operation: "status", input: {} });
278 yield* Fiber.join(consumer);
279
280 expect(receivedTypes).toEqual(["connected", "request"]);
281 expect(result).toBe("ready");
282 }),
283);
284
285it.effect.each([
286 {
287 clientId: SERVER_BROWSER_AUTOMATION_CLIENT_ID,
288 shown: true,
289 message: "page.goto: net::ERR_CONNECTION_REFUSED at http://localhost:4719/",
290 },
291 {
292 clientId: "client-1",
293 shown: false,
294 message: "page.goto: net::ERR_CONNECTION_REFUSED at http://localhost:4719/",
295 },
296 {
297 clientId: SERVER_BROWSER_AUTOMATION_CLIENT_ID,
298 shown: true,
299 message: new PreviewBrowserRootError().message,
300 },
301])("preserves server browser failure guidance ($clientId)", ({ clientId, shown, message }) =>
302 Effect.gen(function* () {
303 const broker = yield* makeBroker;
304 const requests = requestsFrom(yield* broker.connect(makeHost({ clientId })));
305 yield* Stream.runForEach(requests, (request) =>
306 broker.respond({
307 clientId,
308 connectionId: request.connectionId,
309 requestId: request.requestId,
310 ok: false,
311 error: {
312 _tag: "PreviewAutomationExecutionError",
313 message,
314 },
315 }),
316 ).pipe(Effect.forkScoped);
317 yield* Effect.yieldNow;
318 const error = yield* broker
319 .invoke<void>({ scope, operation: "open", input: {} })
320 .pipe(Effect.flip);
321 expect(error._tag).toBe("PreviewAutomationExecutionError");
322 expect(error.message.includes(message)).toBe(shown);
323 }),
324);
325
326it.effect("names the click target that matched nothing when the server browser times out", () =>
327 Effect.gen(function* () {
328 const broker = yield* makeBroker;
329 const clientId = SERVER_BROWSER_AUTOMATION_CLIENT_ID;
330 const requests = requestsFrom(yield* broker.connect(makeHost({ clientId })));
331 yield* Stream.runForEach(requests, (request) =>
332 broker.respond({
333 clientId,
334 connectionId: request.connectionId,
335 requestId: request.requestId,
336 ok: false,
337 error: {
338 _tag: "PreviewAutomationTimeoutError",
339 message: "No element matches the target.",
340 detail: { reason: 'No element matches ".definitely-no-such-t3-target".' },
341 },
342 }),
343 ).pipe(Effect.forkScoped);
344 yield* Effect.yieldNow;
345
346 const error = yield* broker
347 .invoke<void>({
348 scope,
349 operation: "click",
350 input: { selector: ".definitely-no-such-t3-target", timeoutMs: 15_000 },
351 })
352 .pipe(Effect.flip);
353
354 expect(error._tag).toBe("PreviewAutomationTimeoutError");
355 expect(error.message).toBe(
356 'Preview automation click timed out after 15000ms. No element matches ".definitely-no-such-t3-target".',
357 );
358 }),
359);
360
361it.effect("distinguishes malformed remote failures", () =>
362 Effect.gen(function* () {
363 const broker = yield* makeBroker;
364 const requests = requestsFrom(yield* broker.connect(makeHost()));
365 yield* Stream.runForEach(requests, (request) =>
366 broker.respond({
367 clientId: "client-1",
368 connectionId: request.connectionId,
369 requestId: request.requestId,
370 ok: false,
371 }),
372 ).pipe(Effect.forkScoped);
373 yield* Effect.yieldNow;
374
375 const error = yield* broker
376 .invoke<void>({ scope, operation: "status", input: {}, timeoutMs: 2_000 })
377 .pipe(Effect.flip);
378
379 expect(error).toBeInstanceOf(PreviewAutomationMalformedResponseError);
380 expect(error).toMatchObject({
381 operation: "status",
382 environmentId: scope.environmentId,
383 threadId: scope.thread.threadId,
384 providerSessionId: scope.thread.providerSessionId,
385 providerInstanceId: scope.thread.providerInstanceId,
386 clientId: "client-1",
387 requestId: "preview-0",
388 timeoutMs: 2_000,
389 });
390 }),
391);
392
393it.effect("does not create host state from focus updates without a live stream", () =>
394 Effect.gen(function* () {
395 const broker = yield* makeBroker;
396 yield* broker.focusHost({
397 clientId: "client-1",
398 environmentId: scope.environmentId,
399 connectionId: "connection-missing",
400 focused: true,
401 });
402
403 const error = yield* broker
404 .invoke<void>({ scope, operation: "status", input: {} })
405 .pipe(Effect.flip);
406 expect(error).toBeInstanceOf(PreviewAutomationNoAvailableHostError);
407 }),
408);
409
410it.effect("removes host availability when the authoritative request stream disconnects", () =>
411 Effect.gen(function* () {
412 const broker = yield* makeBroker;
413 const requests = requestsFrom(yield* broker.connect(makeHost()));
414 const beforeAcquisition = yield* broker
415 .invoke<void>({ scope, operation: "status", input: {} })
416 .pipe(Effect.flip);
417 expect(beforeAcquisition).toBeInstanceOf(PreviewAutomationNoAvailableHostError);
418
419 const consumer = yield* Stream.runDrain(requests).pipe(Effect.forkScoped);
420 yield* Effect.yieldNow;
421 yield* Fiber.interrupt(consumer);
422
423 const error = yield* broker
424 .invoke<void>({ scope, operation: "status", input: {} })
425 .pipe(Effect.flip);
426 expect(error).toBeInstanceOf(PreviewAutomationNoAvailableHostError);
427 }),
428);
429
430it.effect("routes requests for background threads through an environment-level host", () =>
431 Effect.gen(function* () {
432 const broker = yield* makeBroker;
433 const backgroundThreadId = ThreadId.make("thread-background");
434 const requests = requestsFrom(yield* broker.connect(makeHost()));
435 let routedThreadId: string | undefined;
436 yield* Stream.runForEach(requests, (request) => {
437 routedThreadId = request.threadId;
438 return broker.respond({
439 clientId: "client-1",
440 connectionId: request.connectionId,
441 requestId: request.requestId,
442 ok: true,
443 result: "background",
444 });
445 }).pipe(Effect.forkScoped);
446 yield* Effect.yieldNow;
447
448 const result = yield* broker.invoke<string>({
449 scope: {
450 ...scope,
451 thread: {
452 ...scope.thread,
453 threadId: backgroundThreadId,
454 providerSessionId: "provider-session-background",
455 },
456 },
457 operation: "status",
458 input: {},
459 });
460
461 expect(result).toBe("background");
462 expect(routedThreadId).toBe(backgroundThreadId);
463 }),
464);
465
466it.effect("never routes a provider session to a host from another environment", () =>
467 Effect.gen(function* () {
468 const broker = yield* makeBroker;
469 const matchingRequests = requestsFrom(
470 yield* broker.connect(makeHost({ clientId: "client-matching" })),
471 );
472 const foreignRequests = requestsFrom(
473 yield* broker.connect(
474 makeHost({
475 clientId: "client-foreign",
476 environmentId: EnvironmentId.make("environment-foreign"),
477 }),
478 ),
479 );
480 yield* Stream.runForEach(matchingRequests, (request) =>
481 broker.respond({
482 clientId: "client-matching",
483 connectionId: request.connectionId,
484 requestId: request.requestId,
485 ok: true,
486 result: "matching",
487 }),
488 ).pipe(Effect.forkScoped);
489 yield* Stream.runForEach(foreignRequests, (request) =>
490 broker.respond({
491 clientId: "client-foreign",
492 connectionId: request.connectionId,
493 requestId: request.requestId,
494 ok: true,
495 result: "foreign",
496 }),
497 ).pipe(Effect.forkScoped);
498 yield* Effect.yieldNow;
499
500 expect(yield* broker.invoke<string>({ scope, operation: "status", input: {} })).toBe(
501 "matching",
502 );
503 }),
504);
505
506it.effect("pins a provider session to its initial host despite later focus changes", () =>
507 Effect.gen(function* () {
508 const broker = yield* makeBroker;
509 let firstConnectionId = "";
510 let secondConnectionId = "";
511 const firstRequests = requestsFrom(
512 yield* broker.connect(makeHost({ clientId: "client-first" })),
513 (connectionId) => {
514 firstConnectionId = connectionId;
515 },
516 );
517 const secondRequests = requestsFrom(
518 yield* broker.connect(makeHost({ clientId: "client-second" })),
519 (connectionId) => {
520 secondConnectionId = connectionId;
521 },
522 );
523 yield* Stream.runForEach(firstRequests, (request) =>
524 broker.respond({
525 clientId: "client-first",
526 connectionId: request.connectionId,
527 requestId: request.requestId,
528 ok: true,
529 result: "first",
530 }),
531 ).pipe(Effect.forkScoped);
532 yield* Stream.runForEach(secondRequests, (request) =>
533 broker.respond({
534 clientId: "client-second",
535 connectionId: request.connectionId,
536 requestId: request.requestId,
537 ok: true,
538 result: "second",
539 }),
540 ).pipe(Effect.forkScoped);
541 yield* Effect.yieldNow;
542
543 yield* broker.focusHost({
544 clientId: "client-first",
545 environmentId: scope.environmentId,
546 connectionId: "connection-stale",
547 focused: true,
548 liveTabs: [{ threadId: scope.thread.threadId, tabId: PreviewTabId.make("stale-tab") }],
549 });
550 expect(yield* broker.invoke<string>({ scope, operation: "status", input: {} })).toBe("second");
551 yield* broker.focusHost({
552 clientId: "client-first",
553 environmentId: scope.environmentId,
554 connectionId: firstConnectionId,
555 focused: true,
556 });
557
558 const firstPinnedScope = {
559 ...scope,
560 thread: { ...scope.thread, providerSessionId: "provider-session-first-pinned" },
561 };
562 expect(
563 yield* broker.invoke<string>({ scope: firstPinnedScope, operation: "status", input: {} }),
564 ).toBe("first");
565
566 yield* broker.focusHost({
567 clientId: "client-second",
568 environmentId: scope.environmentId,
569 connectionId: secondConnectionId,
570 focused: true,
571 });
572
573 expect(
574 yield* broker.invoke<string>({ scope: firstPinnedScope, operation: "status", input: {} }),
575 ).toBe("first");
576 expect(
577 yield* broker.invoke<string>({
578 scope: {
579 ...scope,
580 thread: { ...scope.thread, providerSessionId: "provider-session-second-pinned" },
581 },
582 operation: "status",
583 input: {},
584 }),
585 ).toBe("second");
586 }),
587);
588
589it.effect("prefers the live tab owner for new sessions without moving existing leases", () =>
590 Effect.gen(function* () {
591 const broker = yield* makeBroker;
592 const connections = new Map<string, string>();
593 for (const clientId of ["owner", "other"]) {
594 const requests = requestsFrom(yield* broker.connect(makeHost({ clientId })), (connectionId) =>
595 connections.set(clientId, connectionId),
596 );
597 yield* Stream.runForEach(requests, (request) =>
598 broker.respond({
599 clientId,
600 connectionId: request.connectionId,
601 requestId: request.requestId,
602 ok: true,
603 result: clientId,
604 }),
605 ).pipe(Effect.forkScoped);
606 }
607 yield* Effect.yieldNow;
608 yield* broker.focusHost({
609 clientId: "owner",
610 environmentId: scope.environmentId,
611 connectionId: connections.get("owner")!,
612 focused: false,
613 liveTabs: [
614 { threadId: scope.thread.threadId, tabId: PreviewTabId.make("signed-in"), visible: true },
615 ],
616 });
617 yield* broker.focusHost({
618 clientId: "other",
619 environmentId: scope.environmentId,
620 connectionId: connections.get("other")!,
621 focused: true,
622 liveTabs: [
623 {
624 threadId: scope.thread.threadId,
625 tabId: PreviewTabId.make("signed-in"),
626 visible: false,
627 },
628 {
629 threadId: ThreadId.make("another-thread"),
630 tabId: PreviewTabId.make("different-tab"),
631 visible: true,
632 },
633 ],
634 });
635 expect(yield* broker.invoke<string>({ scope, operation: "evaluate", input: {} })).toBe("owner");
636 expect(
637 yield* broker.invoke<string>({
638 scope: { ...scope, thread: { ...scope.thread, providerSessionId: "explicit-owner" } },
639 tabId: PreviewTabId.make("signed-in"),
640 operation: "snapshot",
641 input: {},
642 }),
643 ).toBe("owner");
644 expect(
645 yield* broker.invoke<string>({
646 scope: { ...scope, thread: { ...scope.thread, providerSessionId: "other-tab" } },
647 tabId: PreviewTabId.make("different-tab"),
648 operation: "evaluate",
649 input: {},
650 }),
651 ).toBe("other");
652
653 yield* broker.focusHost({
654 clientId: "owner",
655 environmentId: scope.environmentId,
656 connectionId: connections.get("owner")!,
657 focused: false,
658 liveTabs: [],
659 });
660 expect(yield* broker.invoke<string>({ scope, operation: "evaluate", input: {} })).toBe("owner");
661 expect(
662 yield* broker.invoke<string>({
663 scope: { ...scope, thread: { ...scope.thread, providerSessionId: "after-tab-closed" } },
664 operation: "evaluate",
665 input: {},
666 }),
667 ).toBe("other");
668 }),
669);
670
671it.effect("routes resize to a capable host instead of a newer legacy connection", () =>
672 Effect.gen(function* () {
673 const broker = yield* makeBroker;
674 const capableRequests = requestsFrom(
675 yield* broker.connect(
676 makeHost({ clientId: "client-capable", supportedOperations: ["resize"] }),
677 ),
678 );
679 const legacyRequests = requestsFrom(
680 yield* broker.connect(makeHost({ clientId: "client-legacy" })),
681 );
682 yield* Stream.runForEach(capableRequests, (request) =>
683 broker.respond({
684 clientId: "client-capable",
685 connectionId: request.connectionId,
686 requestId: request.requestId,
687 ok: true,
688 result: "capable",
689 }),
690 ).pipe(Effect.forkScoped);
691 yield* Stream.runForEach(legacyRequests, (request) =>
692 broker.respond({
693 clientId: "client-legacy",
694 connectionId: request.connectionId,
695 requestId: request.requestId,
696 ok: true,
697 result: "legacy",
698 }),
699 ).pipe(Effect.forkScoped);
700 yield* Effect.yieldNow;
701
702 expect(
703 yield* broker.invoke<string>({ scope, operation: "resize", input: { mode: "fill" } }),
704 ).toBe("capable");
705 }),
706);
707
708it.effect("does not move a live legacy assignment to another runtime for resize", () =>
709 Effect.gen(function* () {
710 const broker = yield* makeBroker;
711 const legacyRequests = requestsFrom(
712 yield* broker.connect(makeHost({ clientId: "client-legacy" })),
713 );
714 yield* Stream.runForEach(legacyRequests, (request) =>
715 broker.respond({
716 clientId: "client-legacy",
717 connectionId: request.connectionId,
718 requestId: request.requestId,
719 ok: true,
720 result: "legacy",
721 }),
722 ).pipe(Effect.forkScoped);
723 yield* Effect.yieldNow;
724
725 expect(yield* broker.invoke<string>({ scope, operation: "status", input: {} })).toBe("legacy");
726
727 const capableRequests = requestsFrom(
728 yield* broker.connect(
729 makeHost({ clientId: "client-capable", supportedOperations: ["resize"] }),
730 ),
731 );
732 yield* Stream.runForEach(capableRequests, (request) =>
733 broker.respond({
734 clientId: "client-capable",
735 connectionId: request.connectionId,
736 requestId: request.requestId,
737 ok: true,
738 result: "capable",
739 }),
740 ).pipe(Effect.forkScoped);
741 yield* Effect.yieldNow;
742
743 const error = yield* broker
744 .invoke<void>({ scope, operation: "resize", input: { mode: "fill" } })
745 .pipe(Effect.flip);
746 expect(error).toBeInstanceOf(PreviewAutomationNoAvailableHostError);
747 expect(yield* broker.invoke<string>({ scope, operation: "status", input: {} })).toBe("legacy");
748 }),
749);
750
751it.effect("ignores stale focus updates for a different environment", () =>
752 Effect.gen(function* () {
753 const broker = yield* makeBroker;
754 let firstConnectionId = "";
755 const firstRequests = requestsFrom(
756 yield* broker.connect(makeHost({ clientId: "client-first" })),
757 (connectionId) => {
758 firstConnectionId = connectionId;
759 },
760 );
761 const secondRequests = requestsFrom(
762 yield* broker.connect(makeHost({ clientId: "client-second" })),
763 );
764 yield* Stream.runForEach(firstRequests, (request) =>
765 broker.respond({
766 clientId: "client-first",
767 connectionId: request.connectionId,
768 requestId: request.requestId,
769 ok: true,
770 result: "first",
771 }),
772 ).pipe(Effect.forkScoped);
773 yield* Stream.runForEach(secondRequests, (request) =>
774 broker.respond({
775 clientId: "client-second",
776 connectionId: request.connectionId,
777 requestId: request.requestId,
778 ok: true,
779 result: "second",
780 }),
781 ).pipe(Effect.forkScoped);
782 yield* Effect.yieldNow;
783
784 yield* broker.focusHost({
785 clientId: "client-first",
786 environmentId: EnvironmentId.make("environment-stale"),
787 connectionId: firstConnectionId,
788 focused: true,
789 });
790
791 expect(yield* broker.invoke<string>({ scope, operation: "status", input: {} })).toBe("second");
792 }),
793);
794
795it.effect("fails over a pinned provider session only after its host disconnects", () =>
796 Effect.gen(function* () {
797 const broker = yield* makeBroker;
798 const firstTabId = PreviewTabId.make("tab-on-first-host");
799 let firstConnectionId = "";
800 let secondRoutedTabId: PreviewTabId | undefined;
801 const firstRequests = requestsFrom(
802 yield* broker.connect(makeHost({ clientId: "client-first" })),
803 (connectionId) => {
804 firstConnectionId = connectionId;
805 },
806 );
807 const secondRequests = requestsFrom(
808 yield* broker.connect(makeHost({ clientId: "client-second" })),
809 );
810 const firstConsumer = yield* Stream.runForEach(firstRequests, (request) =>
811 broker.respond({
812 clientId: "client-first",
813 connectionId: request.connectionId,
814 requestId: request.requestId,
815 ok: true,
816 result: request.operation === "open" ? { host: "first", tabId: firstTabId } : "first",
817 }),
818 ).pipe(Effect.forkScoped);
819 yield* Stream.runForEach(secondRequests, (request) => {
820 secondRoutedTabId = request.tabId;
821 return broker.respond({
822 clientId: "client-second",
823 connectionId: request.connectionId,
824 requestId: request.requestId,
825 ok: true,
826 result: "second",
827 });
828 }).pipe(Effect.forkScoped);
829 yield* Effect.yieldNow;
830
831 yield* broker.focusHost({
832 clientId: "client-first",
833 environmentId: scope.environmentId,
834 connectionId: firstConnectionId,
835 focused: true,
836 liveTabs: [{ threadId: scope.thread.threadId, tabId: firstTabId }],
837 });
838 expect(yield* broker.invoke({ scope, operation: "open", input: {} })).toEqual({
839 host: "first",
840 tabId: firstTabId,
841 });
842
843 yield* Fiber.interrupt(firstConsumer);
844 yield* Effect.yieldNow;
845
846 expect(yield* broker.invoke<string>({ scope, operation: "status", input: {} })).toBe("second");
847 expect(secondRoutedTabId).toBeUndefined();
848 }),
849);
850
851it.effect("keeps a replacement stream authoritative when the old stream finalizes", () =>
852 Effect.gen(function* () {
853 const broker = yield* makeBroker;
854 let firstConnectionId = "";
855 let replacementConnectionId = "";
856 const firstRequests = requestsFrom(yield* broker.connect(makeHost()), (connectionId) => {
857 firstConnectionId = connectionId;
858 });
859 yield* Stream.runDrain(firstRequests).pipe(Effect.forkScoped);
860 yield* Effect.yieldNow;
861
862 const replacementRequests = requestsFrom(yield* broker.connect(makeHost()), (connectionId) => {
863 replacementConnectionId = connectionId;
864 });
865 yield* Stream.runForEach(replacementRequests, (request) =>
866 broker.respond({
867 clientId: "client-1",
868 connectionId: request.connectionId,
869 requestId: request.requestId,
870 ok: true,
871 result: "replacement",
872 }),
873 ).pipe(Effect.forkScoped);
874 yield* Effect.yieldNow;
875
876 expect(replacementConnectionId).not.toBe(firstConnectionId);
877 const result = yield* broker.invoke<string>({ scope, operation: "status", input: {} });
878 expect(result).toBe("replacement");
879 }),
880);
881
882it.effect("does not carry a tab id across a replacement automation stream", () =>
883 Effect.gen(function* () {
884 const broker = yield* makeBroker;
885 const openedTabId = PreviewTabId.make("tab-first-webcontents");
886 const firstRequests = requestsFrom(yield* broker.connect(makeHost()));
887 yield* Stream.runForEach(firstRequests, (request) =>
888 broker.respond({
889 clientId: "client-1",
890 connectionId: request.connectionId,
891 requestId: request.requestId,
892 ok: true,
893 result:
894 request.operation === "open" ? { host: "first", tabId: openedTabId } : { host: "first" },
895 }),
896 ).pipe(Effect.forkScoped);
897 yield* Effect.yieldNow;
898
899 expect(yield* broker.invoke({ scope, operation: "open", input: {} })).toEqual({
900 host: "first",
901 tabId: openedTabId,
902 });
903
904 const routedRequests: RoutedRequest[] = [];
905 const replacementRequests = requestsFrom(yield* broker.connect(makeHost()));
906 yield* Stream.runForEach(replacementRequests, (request) => {
907 routedRequests.push(request);
908 return broker.respond({
909 clientId: "client-1",
910 connectionId: request.connectionId,
911 requestId: request.requestId,
912 ok: true,
913 result: "replacement",
914 });
915 }).pipe(Effect.forkScoped);
916 yield* Effect.yieldNow;
917
918 expect(yield* broker.invoke<string>({ scope, operation: "status", input: {} })).toBe(
919 "replacement",
920 );
921 expect(routedRequests.at(-1)?.tabId).toBeUndefined();
922 }),
923);
924
925it.effect("fails requests assigned to the stream that is replaced", () =>
926 Effect.gen(function* () {
927 const broker = yield* makeBroker;
928 const requests = requestsFrom(yield* broker.connect(makeHost()));
929 yield* Stream.runDrain(requests).pipe(Effect.forkScoped);
930 const pending = yield* broker
931 .invoke<void>({ scope, operation: "status", input: {} })
932 .pipe(Effect.flip, Effect.forkScoped);
933 yield* Effect.yieldNow;
934
935 const replacementRequests = requestsFrom(yield* broker.connect(makeHost()));
936 yield* Stream.runDrain(replacementRequests).pipe(Effect.forkScoped);
937
938 const error = yield* Fiber.join(pending);
939 expect(error).toBeInstanceOf(PreviewAutomationClientDisconnectedError);
940 expect(error).toMatchObject({
941 operation: "status",
942 environmentId: scope.environmentId,
943 threadId: scope.thread.threadId,
944 providerSessionId: scope.thread.providerSessionId,
945 providerInstanceId: scope.thread.providerInstanceId,
946 clientId: "client-1",
947 requestId: "preview-0",
948 timeoutMs: 15_000,
949 });
950 }),
951);
952
953it.effect("accepts responses only from the host that received the request", () =>
954 Effect.gen(function* () {
955 const broker = yield* makeBroker;
956 const requests = requestsFrom(yield* broker.connect(makeHost()));
957 yield* Stream.runForEach(requests, (request) =>
958 Effect.gen(function* () {
959 yield* broker.respond({
960 clientId: "client-foreign",
961 connectionId: request.connectionId,
962 requestId: request.requestId,
963 ok: true,
964 result: "foreign",
965 });
966 yield* broker.respond({
967 clientId: "client-1",
968 connectionId: "connection-stale",
969 requestId: request.requestId,
970 ok: true,
971 result: "stale",
972 });
973 yield* broker.respond({
974 clientId: "client-1",
975 connectionId: request.connectionId,
976 requestId: request.requestId,
977 ok: true,
978 result: "owner",
979 });
980 }),
981 ).pipe(Effect.forkScoped);
982 yield* Effect.yieldNow;
983
984 const result = yield* broker.invoke<string>({ scope, operation: "status", input: {} });
985 expect(result).toBe("owner");
986 }),
987);
988
989it.effect("keeps each thread's current tab when threads share a provider session", () =>
990 Effect.gen(function* () {
991 const broker = yield* makeBroker;
992 const routed: Array<RoutedRequest> = [];
993 const requests = requestsFrom(yield* broker.connect(makeHost(), { preferred: true }));
994 yield* Stream.runForEach(requests, (request) => {
995 routed.push(request);
996 return broker.respond({
997 clientId: "client-1",
998 connectionId: request.connectionId,
999 requestId: request.requestId,
1000 ok: true,
1001 result: { tabId: request.tabId ?? `tab-${request.threadId}` },
1002 });
1003 }).pipe(Effect.forkScoped);
1004 yield* Effect.yieldNow;
1005 // Codex runs every thread of an instance in one provider session.
1006 const otherThread = {
1007 ...scope,
1008 thread: { ...scope.thread, threadId: ThreadId.make("thread-2") },
1009 };
1010
1011 yield* broker.invoke({ scope, operation: "open", input: {} });
1012 yield* broker.invoke({ scope: otherThread, operation: "snapshot", input: {} });
1013 yield* broker.invoke({ scope, operation: "snapshot", input: {} });
1014 yield* broker.invoke({ scope: otherThread, operation: "snapshot", input: {} });
1015
1016 expect(routed.map(({ threadId, tabId }) => [threadId, tabId])).toEqual([
1017 ["thread-1", undefined],
1018 ["thread-2", undefined],
1019 ["thread-1", "tab-thread-1"],
1020 ["thread-2", "tab-thread-2"],
1021 ]);
1022 expect(routed[0]!.agentSessionId).not.toBe(routed[1]!.agentSessionId);
1023 }),
1024);