1import { desktopTerminalSocket } from "./desktopTerminal.ts";
2import {
3 DEFAULT_SERVER_SETTINGS,
4 EnvironmentId,
5 ProviderDriverKind,
6 ProviderInstanceId,
7 ServerConfig,
8 type ServerConfig as ServerConfigType,
9 ServerConfigStreamEvent,
10 type ServerConfigStreamEvent as ServerConfigStreamEventType,
11 WS_METHODS,
12 UsageLimitSourceId,
13} from "@t3tools/contracts";
14import { describe, expect, it } from "@effect/vitest";
15import * as Cause from "effect/Cause";
16import * as Deferred from "effect/Deferred";
17import * as Effect from "effect/Effect";
18import * as Exit from "effect/Exit";
19import * as Fiber from "effect/Fiber";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Queue from "effect/Queue";
23import * as Schema from "effect/Schema";
24import * as Stream from "effect/Stream";
25import * as SubscriptionRef from "effect/SubscriptionRef";
26import * as TestClock from "effect/testing/TestClock";
27import * as Socket from "effect/socket/Socket";
28
29import {
30 AVAILABLE_CONNECTION_STATE,
31 ConnectionBlockedError,
32 ConnectionTransientError,
33 PrimaryConnectionTarget,
34 RelayConnectionTarget,
35 type PreparedConnection,
36} from "../connection/model.ts";
37import * as EnvironmentSupervisor from "../connection/supervisor.ts";
38import * as Persistence from "../platform/persistence.ts";
39import * as RpcSession from "./session.ts";
40import { makeEnvironmentServerConfigState } from "../state/server.ts";
41import { applyServerConfigProjection } from "../state/serverConfigProjection.ts";
42import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
43
44type SocketEventType = "open" | "message" | "close" | "error";
45type SocketEvent = {
46 readonly code?: number;
47 readonly data?: unknown;
48 readonly reason?: string;
49 readonly type: SocketEventType;
50};
51type SocketListener = (event: SocketEvent) => void;
52
53class TestWebSocket {
54 static readonly CONNECTING = 0;
55 static readonly OPEN = 1;
56 static readonly CLOSING = 2;
57 static readonly CLOSED = 3;
58
59 readyState = TestWebSocket.CONNECTING;
60 readonly sent: string[] = [];
61 readonly url: string;
62 private readonly listeners = new Map<SocketEventType, Set<SocketListener>>();
63
64 constructor(url: string) {
65 this.url = url;
66 }
67
68 addEventListener(type: SocketEventType, listener: SocketListener) {
69 const listeners = this.listeners.get(type) ?? new Set<SocketListener>();
70 listeners.add(listener);
71 this.listeners.set(type, listeners);
72 }
73
74 removeEventListener(type: SocketEventType, listener: SocketListener) {
75 this.listeners.get(type)?.delete(listener);
76 }
77
78 send(data: string) {
79 this.sent.push(data);
80 }
81
82 close(code = 1000, reason = "") {
83 if (this.readyState === TestWebSocket.CLOSED) {
84 return;
85 }
86 this.readyState = TestWebSocket.CLOSED;
87 this.emit("close", { code, reason, type: "close" });
88 }
89
90 open() {
91 this.readyState = TestWebSocket.OPEN;
92 this.emit("open", { type: "open" });
93 }
94
95 serverMessage(data: string) {
96 this.emit("message", { data, type: "message" });
97 }
98
99 private emit(type: SocketEventType, event: SocketEvent) {
100 for (const listener of this.listeners.get(type) ?? []) {
101 listener(event);
102 }
103 }
104}
105
106const TARGET = new PrimaryConnectionTarget({
107 environmentId: EnvironmentId.make("environment-1"),
108 label: "Test environment",
109 httpBaseUrl: "https://environment.example.test",
110 wsBaseUrl: "wss://environment.example.test",
111});
112
113const PREPARED: PreparedConnection = {
114 environmentId: TARGET.environmentId,
115 label: TARGET.label,
116 httpBaseUrl: TARGET.httpBaseUrl,
117 socketUrl: "wss://environment.example.test/ws?wsTicket=test",
118 httpAuthorization: null,
119 target: TARGET,
120};
121
122const SERVER_CONFIG: ServerConfigType = {
123 environment: {
124 environmentId: TARGET.environmentId,
125 label: TARGET.label,
126 platform: {
127 os: "darwin",
128 arch: "arm64",
129 },
130 serverVersion: "0.0.0-test",
131 capabilities: {
132 repositoryIdentity: true,
133 connectionProbe: true,
134 },
135 },
136 auth: {
137 policy: "loopback-browser",
138 bootstrapMethods: ["one-time-token"],
139 sessionMethods: ["browser-session-cookie", "bearer-access-token"],
140 sessionCookieName: "t3_session",
141 },
142 cwd: "/tmp/workspace",
143 keybindingsConfigPath: "/tmp/workspace/keybindings.json",
144 keybindings: [],
145 issues: [],
146 providers: [],
147 availableEditors: [],
148 observability: {
149 logsDirectoryPath: "/tmp/logs",
150 localTracingEnabled: false,
151 otlpTracesEnabled: false,
152 otlpMetricsEnabled: false,
153 otlpLogsEnabled: false,
154 },
155 settings: DEFAULT_SERVER_SETTINGS,
156};
157
158const RpcRequest = Schema.TaggedStruct("Request", {
159 id: Schema.Union([Schema.String, Schema.Number]),
160 payload: Schema.Unknown,
161 tag: Schema.String,
162});
163const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
164const isRpcRequest = Schema.is(RpcRequest);
165const isPing = Schema.is(Schema.TaggedStruct("Ping", {}));
166const encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
167const encodeServerConfig = Schema.encodeSync(ServerConfig);
168const encodeServerConfigStreamEvent = Schema.encodeSync(ServerConfigStreamEvent);
169const encodeDefect = Schema.encodeSync(Schema.Defect());
170const ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG);
171const THEME_SERVER_CONFIG: ServerConfigType = {
172 ...SERVER_CONFIG,
173 environment: {
174 ...SERVER_CONFIG.environment,
175 capabilities: {
176 ...SERVER_CONFIG.environment.capabilities,
177 environmentThemes: true,
178 },
179 },
180};
181const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG);
182const SOURCE_SERVER_CONFIG: ServerConfigType = {
183 ...THEME_SERVER_CONFIG,
184 environment: {
185 ...THEME_SERVER_CONFIG.environment,
186 capabilities: { ...THEME_SERVER_CONFIG.environment.capabilities, usageLimitSources: true },
187 },
188};
189const SOURCE_EVENT: ServerConfigStreamEventType = {
190 version: 1,
191 type: "usageLimitSourcesUpdated",
192 payload: {
193 sources: [
194 {
195 id: UsageLimitSourceId.make("proxy"),
196 kind: "cliproxy",
197 label: "Proxy",
198 checkedAt: "2026-09-04T00:00:00Z",
199 accounts: [],
200 },
201 ],
202 },
203};
204
205const LEGACY_SERVER_CONFIG = {
206 ...ENCODED_SERVER_CONFIG,
207 environment: {
208 ...ENCODED_SERVER_CONFIG.environment,
209 capabilities: {
210 repositoryIdentity: true,
211 },
212 },
213};
214
215const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* (
216 options: RpcSession.RpcSessionOptions = {},
217) {
218 const sockets: TestWebSocket[] = [];
219 const layerConstructor = Layer.succeed(Socket.WebSocketConstructor, (url) => {
220 const socket = new TestWebSocket(url);
221 sockets.push(socket);
222 return socket as unknown as globalThis.WebSocket;
223 });
224 const layer = RpcSession.layer(options).pipe(Layer.provide(layerConstructor));
225 const factory = yield* RpcSession.RpcSessionFactory.pipe(Effect.provide(layer));
226 return { factory, sockets };
227});
228
229const awaitSocket = Effect.fn("TestRpcSessionFactory.awaitSocket")(function* (
230 sockets: ReadonlyArray<TestWebSocket>,
231 index = 0,
232) {
233 for (let attempt = 0; attempt < 100; attempt += 1) {
234 const socket = sockets[index];
235 if (socket) {
236 return socket;
237 }
238 yield* Effect.yieldNow;
239 }
240 return yield* Effect.die(new Error("Expected the RPC protocol to create a websocket."));
241});
242
243const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* (
244 socket: TestWebSocket,
245 index = 0,
246) {
247 for (let attempt = 0; attempt < 100; attempt += 1) {
248 const request = socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)[index];
249 if (request) {
250 return request;
251 }
252 yield* Effect.yieldNow;
253 }
254 return yield* Effect.die(new Error("Expected the RPC protocol to send a request."));
255});
256
257const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialConfig")(function* (
258 socket: TestWebSocket,
259 config: unknown = ENCODED_SERVER_CONFIG,
260 payload: unknown = {},
261) {
262 const request = yield* awaitRequest(socket);
263 expect(request).toMatchObject({
264 _tag: "Request",
265 tag: WS_METHODS.subscribeServerConfig,
266 payload,
267 });
268 socket.serverMessage(
269 encodeJson({
270 _tag: "Chunk",
271 requestId: request.id,
272 values: [{ version: 1, type: "snapshot", config }],
273 }),
274 );
275});
276
277const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents")(function* (
278 socket: TestWebSocket,
279 events: ReadonlyArray<ServerConfigStreamEventType>,
280) {
281 const request = yield* awaitRequest(socket);
282 socket.serverMessage(
283 encodeJson({
284 _tag: "Chunk",
285 requestId: request.id,
286 values: events.map((event) => encodeServerConfigStreamEvent(event)),
287 }),
288 );
289});
290
291describe("RpcSessionFactory", () => {
292 it.effect("uses the local pipe for terminal RPC without opening a terminal WebSocket", () =>
293 Effect.gen(function* () {
294 const incoming = yield* Queue.unbounded<string>();
295 let respond: (data: string) => void = () => {};
296 let onClose: () => void = () => {};
297 let opened = 0;
298 let closed = 0;
299 const { factory, sockets } = yield* makeFactory({
300 terminalSocket: (prepared) =>
301 desktopTerminalSocket(
302 {
303 supportsLocalTerminal: ({ environmentId }) =>
304 environmentId === PREPARED.environmentId,
305 openLocalTerminal: (input, receive, disconnected) => {
306 expect(input.url).toContain("wsTicket=pipe-ticket");
307 opened += 1;
308 respond = receive;
309 onClose = disconnected;
310 return {
311 send: (data) => {
312 Queue.offerUnsafe(incoming, data);
313 },
314 close: () => {
315 closed += 1;
316 },
317 };
318 },
319 },
320 prepared,
321 ),
322 });
323 const session = yield* factory.connect(
324 PREPARED,
325 Effect.succeed({ ...PREPARED, socketUrl: `${PREPARED.socketUrl}&wsTicket=pipe-ticket` }),
326 );
327 const main = yield* awaitSocket(sockets);
328 main.open();
329 yield* completeInitialConfig(main);
330 yield* session.ready;
331 expect(opened).toBe(0);
332 const write = yield* session.client[WS_METHODS.terminalWrite]({
333 threadId: "thread",
334 terminalId: "term-1",
335 data: "j",
336 }).pipe(Effect.forkChild);
337 const message = JSON.parse(yield* Queue.take(incoming));
338 expect(message.tag).toBe(WS_METHODS.terminalWrite);
339 expect(message.payload.data).toBe("j");
340 respond(
341 encodeJson({ _tag: "Exit", requestId: message.id, exit: { _tag: "Success", value: null } }),
342 );
343 yield* Fiber.join(write);
344 expect(sockets).toHaveLength(1);
345 expect(opened).toBe(1);
346 onClose();
347 expect((yield* Effect.exit(session.closed))._tag).toBe("Failure");
348 expect(closed).toBeLessThanOrEqual(1);
349 }).pipe(Effect.scoped),
350 );
351
352 it.effect(
353 "opens one terminal socket lazily with fresh authorization and isolates blocked metadata",
354 () =>
355 Effect.gen(function* () {
356 const { factory, sockets } = yield* makeFactory();
357 let authorizations = 0;
358 const session = yield* factory.connect(
359 PREPARED,
360 Effect.sync(() => {
361 authorizations += 1;
362 return {
363 ...PREPARED,
364 socketUrl: `${PREPARED.socketUrl}&wsTicket=fresh-terminal-ticket`,
365 };
366 }),
367 );
368 const primary = yield* awaitSocket(sockets);
369 primary.open();
370 yield* completeInitialConfig(primary);
371 yield* session.ready;
372 const metadata = yield* session.client[WS_METHODS.subscribeTerminalMetadata](
373 {},
374 { asQueue: true },
375 );
376 const metadataRequest = yield* awaitRequest(primary, 1);
377 expect(authorizations).toBe(0);
378 expect(sockets).toHaveLength(1);
379 primary.serverMessage(
380 encodeJson({
381 _tag: "Chunk",
382 requestId: metadataRequest.id,
383 values: Array.from({ length: 17 }, () => ({
384 type: "remove",
385 threadId: "thread",
386 terminalId: "term-1",
387 })),
388 }),
389 );
390 yield* Effect.yieldNow;
391 expect(yield* Queue.size(metadata)).toBe(16);
392 const probed = yield* Deferred.make<void>();
393 const probe = yield* Effect.forkChild(
394 session.probe.pipe(Effect.tap(() => Deferred.succeed(probed, undefined))),
395 );
396 const probeRequest = yield* awaitRequest(primary, 2);
397 primary.serverMessage(
398 encodeJson({
399 _tag: "Exit",
400 requestId: probeRequest.id,
401 exit: { _tag: "Success", value: {} },
402 }),
403 );
404 const observed = yield* session.client[WS_METHODS.terminalObserve](
405 { threadId: "thread", terminalId: "term-1" },
406 { asQueue: true },
407 );
408 const terminal = yield* awaitSocket(sockets, 1);
409 terminal.open();
410 const observeRequest = yield* awaitRequest(terminal);
411 expect(observeRequest.tag).toBe(WS_METHODS.terminalObserve);
412 expect(terminal.url).toContain("wsTicket=fresh-terminal-ticket");
413 const write = yield* session.client[WS_METHODS.terminalWrite]({
414 threadId: "thread",
415 terminalId: "term-1",
416 data: "j",
417 }).pipe(Effect.forkChild);
418 const writeRequest = yield* awaitRequest(terminal, 1);
419 terminal.serverMessage(
420 encodeJson({
421 _tag: "Exit",
422 requestId: writeRequest.id,
423 exit: { _tag: "Success", value: null },
424 }),
425 );
426 terminal.serverMessage(
427 encodeJson({
428 _tag: "Chunk",
429 requestId: observeRequest.id,
430 values: [{ type: "output", threadId: "thread", terminalId: "term-1", data: "j" }],
431 }),
432 );
433 yield* Fiber.join(write);
434 expect((yield* Queue.take(observed)).type).toBe("output");
435 expect(yield* Deferred.isDone(probed)).toBe(false);
436 expect(authorizations).toBe(1);
437 expect(sockets).toHaveLength(2);
438 yield* Queue.take(metadata);
439 yield* Fiber.join(probe);
440 }),
441 );
442
443 it.effect(
444 "closes both transports with the session and obtains fresh authorization on its replacement",
445 () =>
446 Effect.gen(function* () {
447 const { factory, sockets } = yield* makeFactory();
448 let authorizations = 0;
449 const prepareTerminal = Effect.sync(() => ({
450 ...PREPARED,
451 socketUrl: `${PREPARED.socketUrl}&wsTicket=terminal-${++authorizations}`,
452 }));
453 for (let attempt = 0; attempt < 2; attempt += 1) {
454 yield* Effect.scoped(
455 Effect.gen(function* () {
456 const session = yield* factory.connect(PREPARED, prepareTerminal);
457 const primary = yield* awaitSocket(sockets, attempt * 2);
458 primary.open();
459 yield* completeInitialConfig(primary);
460 yield* session.ready;
461 const resize = yield* session.client[WS_METHODS.terminalResize]({
462 threadId: "thread",
463 terminalId: "term-1",
464 cols: 80,
465 rows: 24,
466 }).pipe(Effect.forkChild);
467 const terminal = yield* awaitSocket(sockets, attempt * 2 + 1);
468 terminal.open();
469 const request = yield* awaitRequest(terminal);
470 terminal.serverMessage(
471 encodeJson({
472 _tag: "Exit",
473 requestId: request.id,
474 exit: { _tag: "Success", value: null },
475 }),
476 );
477 yield* Fiber.join(resize);
478 expect(terminal.url).toContain(`wsTicket=terminal-${attempt + 1}`);
479 (attempt === 0 ? primary : terminal).close(1012, "restart");
480 expect(yield* Effect.flip(session.closed)).toMatchObject({ reason: "transport" });
481 }),
482 );
483 expect(sockets[attempt * 2]?.readyState).toBe(TestWebSocket.CLOSED);
484 expect(sockets[attempt * 2 + 1]?.readyState).toBe(TestWebSocket.CLOSED);
485 }
486 expect(authorizations).toBe(2);
487 }),
488 );
489
490 it.effect("keeps terminal initialization alive when its first command is interrupted", () =>
491 Effect.gen(function* () {
492 const { factory, sockets } = yield* makeFactory();
493 const authorizing = yield* Deferred.make<void>();
494 const authorized = yield* Deferred.make<PreparedConnection>();
495 let authorizations = 0;
496 const session = yield* factory.connect(
497 PREPARED,
498 Effect.gen(function* () {
499 authorizations += 1;
500 yield* Deferred.succeed(authorizing, undefined);
501 return yield* Deferred.await(authorized);
502 }),
503 );
504 const primary = yield* awaitSocket(sockets);
505 primary.open();
506 yield* completeInitialConfig(primary);
507 yield* session.ready;
508 const first = yield* session.client[WS_METHODS.terminalWrite]({
509 threadId: "thread",
510 terminalId: "term-1",
511 data: "first",
512 }).pipe(Effect.forkChild);
513 yield* Deferred.await(authorizing);
514 yield* Fiber.interrupt(first);
515 yield* Deferred.succeed(authorized, PREPARED);
516 const terminal = yield* awaitSocket(sockets, 1);
517 terminal.open();
518 const second = yield* session.client[WS_METHODS.terminalWrite]({
519 threadId: "thread",
520 terminalId: "term-1",
521 data: "second",
522 }).pipe(Effect.forkChild);
523 const request = yield* awaitRequest(terminal);
524 expect(request.payload).toMatchObject({ data: "second" });
525 terminal.serverMessage(
526 encodeJson({ _tag: "Exit", requestId: request.id, exit: { _tag: "Success", value: null } }),
527 );
528 yield* Fiber.join(second);
529 expect(authorizations).toBe(1);
530 expect(primary.readyState).toBe(TestWebSocket.OPEN);
531 expect(terminal.readyState).toBe(TestWebSocket.OPEN);
532 }),
533 );
534
535 it.effect("keeps the terminal transport after its observing caller finishes", () =>
536 Effect.gen(function* () {
537 const { factory, sockets } = yield* makeFactory();
538 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
539 const primary = yield* awaitSocket(sockets);
540 primary.open();
541 yield* completeInitialConfig(primary);
542 yield* session.ready;
543 const observing = yield* session.client[WS_METHODS.terminalObserve]({
544 threadId: "thread",
545 terminalId: "term-1",
546 }).pipe(Stream.take(1), Stream.runCollect, Effect.forkChild);
547 const terminal = yield* awaitSocket(sockets, 1);
548 terminal.open();
549 const observeRequest = yield* awaitRequest(terminal);
550 terminal.serverMessage(
551 encodeJson({
552 _tag: "Chunk",
553 requestId: observeRequest.id,
554 values: [{ type: "output", threadId: "thread", terminalId: "term-1", data: "a" }],
555 }),
556 );
557 yield* Fiber.join(observing);
558 const writing = yield* session.client[WS_METHODS.terminalWrite]({
559 threadId: "thread",
560 terminalId: "term-1",
561 data: "j",
562 }).pipe(Effect.forkChild);
563 const writeRequest = yield* awaitRequest(terminal, 1);
564 terminal.serverMessage(
565 encodeJson({
566 _tag: "Exit",
567 requestId: writeRequest.id,
568 exit: { _tag: "Success", value: null },
569 }),
570 );
571 yield* Fiber.join(writing);
572 expect(sockets).toHaveLength(2);
573 expect(terminal.readyState).toBe(TestWebSocket.OPEN);
574 }),
575 );
576
577 it.effect.each(["write", "observe", "queue"] as const)(
578 "fails %s and the session if terminal authorization is revoked before first use",
579 (consumer) =>
580 Effect.gen(function* () {
581 const { factory, sockets } = yield* makeFactory();
582 const revoked = new ConnectionBlockedError({
583 reason: "permission",
584 detail: "Terminal access was revoked.",
585 });
586 const session = yield* factory.connect(PREPARED, Effect.fail(revoked));
587 const primary = yield* awaitSocket(sockets);
588 primary.open();
589 yield* completeInitialConfig(primary);
590 yield* session.ready;
591 const input = { threadId: "thread", terminalId: "term-1" };
592 const call =
593 consumer === "write"
594 ? session.client[WS_METHODS.terminalWrite]({ ...input, data: "j" })
595 : consumer === "observe"
596 ? session.client[WS_METHODS.terminalObserve](input).pipe(Stream.runDrain)
597 : Effect.flatMap(
598 session.client[WS_METHODS.terminalObserve](input, { asQueue: true }),
599 (queue) => Stream.fromQueue(queue).pipe(Stream.runDrain),
600 );
601 const result = yield* Effect.flip(call);
602 expect(result._tag).toBe("RpcClientError");
603 expect(yield* Effect.flip(session.closed)).toBe(revoked);
604 expect(sockets).toHaveLength(1);
605 }),
606 );
607
608 it.effect("owns one scoped websocket attempt and exposes readiness and closure", () =>
609 Effect.gen(function* () {
610 const { factory, sockets } = yield* makeFactory();
611 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
612 const readyFiber = yield* Effect.forkChild(session.ready);
613 const socket = yield* awaitSocket(sockets);
614
615 expect(socket.url).toBe(PREPARED.socketUrl);
616 socket.open();
617 yield* completeInitialConfig(socket);
618 yield* Fiber.join(readyFiber);
619
620 const config = yield* session.initialConfig;
621 expect(config).toEqual(SERVER_CONFIG);
622 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
623 1,
624 );
625
626 const probeFiber = yield* Effect.forkChild(session.probe);
627 const probeRequest = yield* awaitRequest(socket, 1);
628 expect(probeRequest).toMatchObject({
629 _tag: "Request",
630 tag: WS_METHODS.serverProbe,
631 payload: {},
632 });
633 socket.serverMessage(
634 encodeJson({
635 _tag: "Exit",
636 requestId: probeRequest.id,
637 exit: {
638 _tag: "Success",
639 value: {},
640 },
641 }),
642 );
643 yield* Fiber.join(probeFiber);
644
645 expect(
646 socket.sent
647 .map((message) => decodeJson(message))
648 .filter(isRpcRequest)
649 .map((request) => request.tag),
650 ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverProbe]);
651
652 socket.close(1012, "service restart");
653 const error = yield* Effect.flip(session.closed);
654 const configStreamError = yield* session
655 .subscribeServerConfig({})
656 .pipe(Stream.runDrain, Effect.flip);
657
658 expect(error).toBeInstanceOf(ConnectionTransientError);
659 expect(error).toMatchObject({
660 reason: "transport",
661 message: "Test environment disconnected.",
662 });
663 expect(configStreamError).toMatchObject({ _tag: "RpcClientError" });
664 yield* Effect.yieldNow;
665 expect(sockets).toHaveLength(1);
666 }),
667 );
668
669 it.effect("closes the websocket when the session scope is released", () =>
670 Effect.gen(function* () {
671 const { factory, sockets } = yield* makeFactory();
672
673 yield* Effect.scoped(
674 Effect.gen(function* () {
675 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
676 const readyFiber = yield* Effect.forkChild(session.ready);
677 const socket = yield* awaitSocket(sockets);
678 socket.open();
679 yield* completeInitialConfig(socket);
680 yield* Fiber.join(readyFiber);
681 }),
682 );
683
684 expect(sockets[0]?.readyState).toBe(TestWebSocket.CLOSED);
685 }),
686 );
687
688 it.effect("replays current config and broadcasts updates to every subscriber", () =>
689 Effect.scoped(
690 Effect.gen(function* () {
691 const { factory, sockets } = yield* makeFactory();
692 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
693 const readyFiber = yield* Effect.forkChild(session.ready);
694 const socket = yield* awaitSocket(sockets);
695 socket.open();
696 yield* completeInitialConfig(socket);
697 yield* Fiber.join(readyFiber);
698
699 const collectTwo = session
700 .subscribeServerConfig({})
701 .pipe(Stream.take(2), Stream.runCollect);
702 const firstSubscriber = yield* Effect.forkChild(collectTwo);
703 const secondSubscriber = yield* Effect.forkChild(collectTwo);
704 yield* Effect.yieldNow;
705
706 const shortcut = {
707 key: "k",
708 metaKey: false,
709 ctrlKey: false,
710 shiftKey: false,
711 altKey: false,
712 modKey: true,
713 };
714 const request = yield* awaitRequest(socket);
715 socket.serverMessage(
716 encodeJson({
717 _tag: "Chunk",
718 requestId: request.id,
719 values: [
720 {
721 version: 1,
722 type: "keybindingsUpdated",
723 payload: {
724 keybindings: [{ command: "terminal.toggle", shortcut }],
725 issues: [],
726 },
727 },
728 ],
729 }),
730 );
731
732 const firstEvents = Array.from(yield* Fiber.join(firstSubscriber));
733 const secondEvents = Array.from(yield* Fiber.join(secondSubscriber));
734 expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]);
735 expect(secondEvents).toEqual(firstEvents);
736
737 const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead);
738 expect(replay).toMatchObject({
739 _tag: "Some",
740 value: {
741 type: "snapshot",
742 config: { keybindings: [{ command: "terminal.toggle", shortcut }] },
743 },
744 });
745 }),
746 ),
747 );
748
749 it.effect.each([
750 { environmentThemes: true },
751 { usageLimitSources: true },
752 { environmentThemes: true, usageLimitSources: true },
753 ])("shares only a config subscription with the same opt-ins: %j", (options) =>
754 Effect.scoped(
755 Effect.gen(function* () {
756 const { factory, sockets } = yield* makeFactory(options);
757 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
758 const readyFiber = yield* Effect.forkChild(session.ready);
759 const socket = yield* awaitSocket(sockets);
760 socket.open();
761 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, options);
762 yield* Fiber.join(readyFiber);
763
764 const shared = yield* session.subscribeServerConfig(options).pipe(Stream.runHead);
765 expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } });
766 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
767 1,
768 );
769
770 const fallbackFiber = yield* session
771 .subscribeServerConfig({})
772 .pipe(Stream.runHead, Effect.forkChild);
773 const fallbackRequest = yield* awaitRequest(socket, 1);
774 expect(fallbackRequest).toMatchObject({
775 tag: WS_METHODS.subscribeServerConfig,
776 payload: {},
777 });
778 socket.serverMessage(
779 encodeJson({
780 _tag: "Chunk",
781 requestId: fallbackRequest.id,
782 values: [
783 {
784 version: 1,
785 type: "snapshot",
786 config: ENCODED_THEME_SERVER_CONFIG,
787 },
788 ],
789 }),
790 );
791 expect(yield* Fiber.join(fallbackFiber)).toMatchObject({
792 _tag: "Some",
793 value: { type: "snapshot" },
794 });
795 }),
796 ),
797 );
798
799 it.effect.each([
800 { usageLimitSources: true },
801 { environmentThemes: true, usageLimitSources: true },
802 ])("replays usage sources, removal, and capability downgrade with %j", (options) =>
803 Effect.scoped(
804 Effect.gen(function* () {
805 const { factory, sockets } = yield* makeFactory(options);
806 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
807 const ready = yield* Effect.forkChild(session.ready);
808 const socket = yield* awaitSocket(sockets);
809 socket.open();
810 yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), options);
811 yield* Fiber.join(ready);
812 const observed = yield* Queue.unbounded<ServerConfigStreamEventType>();
813 yield* session.subscribeServerConfig(options).pipe(
814 Stream.runForEach((event) => Queue.offer(observed, event)),
815 Effect.forkChild,
816 );
817 expect((yield* Queue.take(observed)).type).toBe("snapshot");
818 const themes: ServerConfigStreamEventType[] = options.environmentThemes
819 ? [{ version: 1, type: "environmentThemesUpdated", payload: { themes: [] } }]
820 : [];
821 for (const event of themes) {
822 yield* publishConfigEvents(socket, [event]);
823 expect(yield* Queue.take(observed)).toEqual(event);
824 }
825 const events: ServerConfigStreamEventType[] = [
826 SOURCE_EVENT,
827 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
828 SOURCE_EVENT,
829 { version: 1, type: "snapshot", config: THEME_SERVER_CONFIG },
830 ];
831 for (const event of events) {
832 yield* publishConfigEvents(socket, [event]);
833 expect(yield* Queue.take(observed)).toEqual(event);
834 const started = yield* Deferred.make<void>();
835 const replay = yield* session.subscribeServerConfig(options).pipe(
836 Stream.tap(() => Deferred.succeed(started, undefined)),
837 Stream.takeUntil((item) => item.type === "keybindingsUpdated"),
838 Stream.runCollect,
839 Effect.forkChild,
840 );
841 yield* Deferred.await(started);
842 // A live end marker makes a missing or stale replay event fail without a timeout.
843 const marker: ServerConfigStreamEventType = {
844 version: 1,
845 type: "keybindingsUpdated",
846 payload: { keybindings: [], issues: [] },
847 };
848 yield* publishConfigEvents(socket, [marker]);
849 expect(yield* Queue.take(observed)).toEqual(marker);
850 const replayed = Array.from(yield* Fiber.join(replay));
851 expect(replayed.slice(1)).toEqual([
852 ...themes,
853 ...(event.type === "snapshot" ? [] : [event]),
854 marker,
855 ]);
856 let projection = applyServerConfigProjection(Option.none(), {
857 version: 1,
858 type: "snapshot",
859 config: SOURCE_SERVER_CONFIG,
860 });
861 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
862 for (const item of replayed) projection = applyServerConfigProjection(projection, item);
863 expect(Option.getOrThrow(projection).config.usageLimitSources).toEqual(
864 event.type === "usageLimitSourcesUpdated" && event.payload.sources.length > 0
865 ? event.payload.sources
866 : undefined,
867 );
868 }
869 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
870 1,
871 );
872 }),
873 ),
874 );
875
876 it.effect("replays theme updates and deletion as authoritative events", () =>
877 Effect.scoped(
878 Effect.gen(function* () {
879 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
880 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
881 const readyFiber = yield* Effect.forkChild(session.ready);
882 const socket = yield* awaitSocket(sockets);
883 socket.open();
884 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {
885 environmentThemes: true,
886 });
887 yield* Fiber.join(readyFiber);
888
889 const firstThemes = [
890 {
891 id: "nightfall",
892 name: "Nightfall",
893 appearance: "dark" as const,
894 canvas: "#1a1b26",
895 accent: "#7aa2f7",
896 },
897 ];
898 const replacementThemes = [
899 {
900 id: "midnight",
901 name: "Midnight",
902 appearance: "dark" as const,
903 canvas: "#000000",
904 accent: "#ffffff",
905 },
906 ];
907 const subscriberStarted = yield* Deferred.make<void>();
908 const subscriber = yield* session.subscribeServerConfig({ environmentThemes: true }).pipe(
909 Stream.mapEffect((event) =>
910 Deferred.succeed(subscriberStarted, undefined).pipe(Effect.as(event)),
911 ),
912 Stream.take(4),
913 Stream.runCollect,
914 Effect.forkChild,
915 );
916 yield* Deferred.await(subscriberStarted);
917 yield* publishConfigEvents(socket, [
918 {
919 version: 1,
920 type: "environmentThemesUpdated",
921 payload: { themes: firstThemes },
922 },
923 {
924 version: 1,
925 type: "environmentThemesUpdated",
926 payload: { themes: replacementThemes },
927 },
928 {
929 version: 1,
930 type: "environmentThemesUpdated",
931 payload: { themes: [] },
932 },
933 ]);
934
935 const liveEvents = Array.from(yield* Fiber.join(subscriber));
936 expect(liveEvents.map((event) => event.type)).toEqual([
937 "snapshot",
938 "environmentThemesUpdated",
939 "environmentThemesUpdated",
940 "environmentThemesUpdated",
941 ]);
942 expect(liveEvents[2]).toMatchObject({ payload: { themes: replacementThemes } });
943
944 const replay = Array.from(
945 yield* session
946 .subscribeServerConfig({ environmentThemes: true })
947 .pipe(Stream.take(2), Stream.runCollect),
948 );
949 expect(replay.map((event) => event.type)).toEqual(["snapshot", "environmentThemesUpdated"]);
950 expect(replay[1]).toMatchObject({ payload: { themes: [] } });
951
952 let projection = applyServerConfigProjection(Option.none(), {
953 version: 1,
954 type: "snapshot",
955 config: THEME_SERVER_CONFIG,
956 });
957 projection = applyServerConfigProjection(projection, {
958 version: 1,
959 type: "environmentThemesUpdated",
960 payload: { themes: firstThemes },
961 });
962 for (const event of replay) {
963 projection = applyServerConfigProjection(projection, event);
964 }
965 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
966 }),
967 ),
968 );
969
970 it.effect("recovers a slow subscriber after it misses theme and usage-source deletion", () =>
971 Effect.scoped(
972 Effect.gen(function* () {
973 const { factory, sockets } = yield* makeFactory({
974 environmentThemes: true,
975 usageLimitSources: true,
976 });
977 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
978 const readyFiber = yield* Effect.forkChild(session.ready);
979 const socket = yield* awaitSocket(sockets);
980 socket.open();
981 yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), {
982 environmentThemes: true,
983 usageLimitSources: true,
984 });
985 yield* Fiber.join(readyFiber);
986
987 const slowSubscriberStarted = yield* Deferred.make<void>();
988 const releaseSlowSubscriber = yield* Deferred.make<void>();
989 let firstEvent = true;
990 const slowSubscriber = yield* session
991 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
992 .pipe(
993 Stream.mapEffect((event) => {
994 if (!firstEvent) return Effect.succeed(event);
995 firstEvent = false;
996 return Deferred.succeed(slowSubscriberStarted, undefined).pipe(
997 Effect.andThen(Deferred.await(releaseSlowSubscriber)),
998 Effect.as(event),
999 );
1000 }),
1001 Stream.take(4),
1002 Stream.runCollect,
1003 Effect.forkChild,
1004 );
1005 yield* Deferred.await(slowSubscriberStarted);
1006
1007 const firstThemes = [
1008 {
1009 id: "nightfall",
1010 name: "Nightfall",
1011 appearance: "dark" as const,
1012 canvas: "#1a1b26",
1013 accent: "#7aa2f7",
1014 },
1015 ];
1016 const themeEvents: ServerConfigStreamEventType[] = [
1017 {
1018 version: 1,
1019 type: "environmentThemesUpdated",
1020 payload: { themes: firstThemes },
1021 },
1022 {
1023 version: 1,
1024 type: "environmentThemesUpdated",
1025 payload: {
1026 themes: [{ ...firstThemes[0]!, name: "Nightfall 2" }],
1027 },
1028 },
1029 {
1030 version: 1,
1031 type: "environmentThemesUpdated",
1032 payload: { themes: [] },
1033 },
1034 ];
1035 const settingsEvents = Array.from({ length: 65 }, (): ServerConfigStreamEventType => ({
1036 version: 1,
1037 type: "settingsUpdated",
1038 payload: { settings: DEFAULT_SERVER_SETTINGS },
1039 }));
1040 const sourceEvents: ServerConfigStreamEventType[] = [
1041 SOURCE_EVENT,
1042 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
1043 ];
1044 const allEvents = [...themeEvents, ...sourceEvents, ...settingsEvents];
1045 const observedByFastSubscriber = yield* Queue.unbounded<ServerConfigStreamEventType>();
1046 yield* session
1047 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
1048 .pipe(
1049 Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)),
1050 Effect.forkChild,
1051 );
1052 expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot");
1053 for (const event of allEvents) {
1054 yield* publishConfigEvents(socket, [event]);
1055 expect(yield* Queue.take(observedByFastSubscriber)).toEqual(event);
1056 }
1057 yield* Deferred.succeed(releaseSlowSubscriber, undefined);
1058
1059 const recovered = Array.from(yield* Fiber.join(slowSubscriber));
1060 expect(recovered.map((event) => event.type)).toEqual([
1061 "snapshot",
1062 "snapshot",
1063 "environmentThemesUpdated",
1064 "usageLimitSourcesUpdated",
1065 ]);
1066 expect(recovered[2]).toMatchObject({ payload: { themes: [] } });
1067 expect(recovered[3]).toMatchObject({ payload: { sources: [] } });
1068
1069 let projection = applyServerConfigProjection(Option.none(), {
1070 version: 1,
1071 type: "snapshot",
1072 config: SOURCE_SERVER_CONFIG,
1073 });
1074 projection = applyServerConfigProjection(projection, themeEvents[0]!);
1075 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
1076 for (const event of recovered.slice(1)) {
1077 projection = applyServerConfigProjection(projection, event);
1078 }
1079 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
1080 expect(Option.getOrThrow(projection).config.usageLimitSources).toBeUndefined();
1081 }),
1082 ),
1083 );
1084
1085 it.effect("closes the session when the config source dies", () =>
1086 Effect.scoped(
1087 Effect.gen(function* () {
1088 const { factory, sockets } = yield* makeFactory();
1089 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1090 const readyFiber = yield* Effect.forkChild(session.ready);
1091 const socket = yield* awaitSocket(sockets);
1092 socket.open();
1093 yield* completeInitialConfig(socket);
1094 yield* Fiber.join(readyFiber);
1095
1096 const closedFiber = yield* session.closed.pipe(Effect.exit, Effect.forkChild);
1097 socket.serverMessage(
1098 encodeJson({
1099 _tag: "Defect",
1100 defect: encodeDefect(new Error("config stream died")),
1101 }),
1102 );
1103
1104 const closed = yield* Fiber.join(closedFiber);
1105 expect(Exit.isFailure(closed)).toBe(true);
1106 if (Exit.isFailure(closed)) {
1107 expect(Cause.hasDies(closed.cause)).toBe(true);
1108 }
1109 }),
1110 ),
1111 );
1112
1113 it.effect.each([{ failure: "defect" as const }, { failure: "typed" as const }])(
1114 "keeps durable config state alive after an owned $failure failure",
1115 ({ failure }) =>
1116 Effect.scoped(
1117 Effect.gen(function* () {
1118 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
1119 const firstSession = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1120 const firstReady = yield* Effect.forkChild(firstSession.ready);
1121 const firstSocket = yield* awaitSocket(sockets);
1122 firstSocket.open();
1123 yield* completeInitialConfig(firstSocket, ENCODED_THEME_SERVER_CONFIG, {
1124 environmentThemes: true,
1125 });
1126 yield* Fiber.join(firstReady);
1127
1128 const activeSession = yield* SubscriptionRef.make(Option.some(firstSession));
1129 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
1130 target: TARGET,
1131 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
1132 session: activeSession,
1133 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
1134 connect: Effect.void,
1135 disconnect: Effect.void,
1136 retryNow: Effect.void,
1137 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
1138 const cache = Persistence.EnvironmentCacheStore.of({
1139 loadShell: () => Effect.succeedNone,
1140 saveShell: () => Effect.void,
1141 loadThread: () => Effect.succeedNone,
1142 saveThread: () => Effect.void,
1143 removeThread: () => Effect.void,
1144 loadServerConfig: () => Effect.succeedNone,
1145 saveServerConfig: () => Effect.void,
1146 loadVcsRefs: () => Effect.succeedNone,
1147 saveVcsRefs: () => Effect.void,
1148 removeVcsRefs: () => Effect.void,
1149 clearVcsRefs: () => Effect.void,
1150 clear: () => Effect.void,
1151 });
1152 const configState = yield* makeEnvironmentServerConfigState({
1153 environmentThemes: true,
1154 }).pipe(
1155 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1156 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1157 );
1158 const awaitConfig = (predicate: (config: ServerConfigType) => boolean) =>
1159 SubscriptionRef.changes(configState).pipe(
1160 Stream.filter(Option.isSome),
1161 Stream.map((projection) => projection.value.config),
1162 Stream.filter(predicate),
1163 Stream.runHead,
1164 Effect.map(Option.getOrThrow),
1165 );
1166
1167 const firstThemes = [
1168 {
1169 id: "first-theme",
1170 name: "First theme",
1171 appearance: "dark" as const,
1172 canvas: "#111111",
1173 accent: "#ffffff",
1174 },
1175 ];
1176 const firstThemeState = yield* awaitConfig(
1177 (config) => config.environmentThemes?.[0]?.id === "first-theme",
1178 ).pipe(Effect.forkChild);
1179 yield* publishConfigEvents(firstSocket, [
1180 {
1181 version: 1,
1182 type: "environmentThemesUpdated",
1183 payload: { themes: firstThemes },
1184 },
1185 ]);
1186 expect((yield* Fiber.join(firstThemeState)).environmentThemes).toEqual(firstThemes);
1187
1188 const firstClosed = yield* firstSession.closed.pipe(Effect.exit, Effect.forkChild);
1189 const firstRequest = yield* awaitRequest(firstSocket);
1190 firstSocket.serverMessage(
1191 failure === "defect"
1192 ? encodeJson({
1193 _tag: "Defect",
1194 defect: encodeDefect(new Error("config stream died")),
1195 })
1196 : encodeJson({
1197 _tag: "Exit",
1198 requestId: firstRequest.id,
1199 exit: {
1200 _tag: "Failure",
1201 cause: [
1202 {
1203 _tag: "Fail",
1204 error: {
1205 _tag: "EnvironmentAuthorizationError",
1206 message: "config subscription rejected",
1207 requiredScope: "orchestration:read",
1208 },
1209 },
1210 ],
1211 },
1212 }),
1213 );
1214 const firstClosedExit = yield* Fiber.join(firstClosed);
1215 expect(Exit.isFailure(firstClosedExit)).toBe(true);
1216 if (failure === "typed" && Exit.isFailure(firstClosedExit)) {
1217 expect(Cause.squash(firstClosedExit.cause)).toBeInstanceOf(ConnectionBlockedError);
1218 expect(Cause.squash(firstClosedExit.cause)).toMatchObject({ reason: "permission" });
1219 }
1220 yield* SubscriptionRef.set(activeSession, Option.none());
1221
1222 const recoveredConfig = {
1223 ...THEME_SERVER_CONFIG,
1224 environment: {
1225 ...THEME_SERVER_CONFIG.environment,
1226 label: "Recovered environment",
1227 },
1228 } satisfies ServerConfigType;
1229 const secondSession = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1230 const secondReady = yield* Effect.forkChild(secondSession.ready);
1231 const secondSocket = yield* awaitSocket(sockets, 1);
1232 secondSocket.open();
1233 yield* completeInitialConfig(secondSocket, encodeServerConfig(recoveredConfig), {
1234 environmentThemes: true,
1235 });
1236 yield* Fiber.join(secondReady);
1237
1238 const recoveredState = yield* awaitConfig(
1239 (config) => config.environment.label === "Recovered environment",
1240 ).pipe(Effect.forkChild);
1241 yield* SubscriptionRef.set(activeSession, Option.some(secondSession));
1242 expect((yield* Fiber.join(recoveredState)).environmentThemes).toEqual(firstThemes);
1243
1244 const recoveredThemes = [
1245 {
1246 id: "recovered-theme",
1247 name: "Recovered theme",
1248 appearance: "dark" as const,
1249 canvas: "#000000",
1250 accent: "#eeeeee",
1251 },
1252 ];
1253 const liveRecoveredState = yield* awaitConfig(
1254 (config) => config.environmentThemes?.[0]?.id === "recovered-theme",
1255 ).pipe(Effect.forkChild);
1256 yield* publishConfigEvents(secondSocket, [
1257 {
1258 version: 1,
1259 type: "environmentThemesUpdated",
1260 payload: { themes: recoveredThemes },
1261 },
1262 ]);
1263 expect((yield* Fiber.join(liveRecoveredState)).environmentThemes).toEqual(
1264 recoveredThemes,
1265 );
1266 }),
1267 ),
1268 );
1269
1270 it.effect.each<{
1271 readonly event: ServerConfigStreamEventType;
1272 readonly expectedConfig: Partial<ServerConfigType>;
1273 }>([
1274 {
1275 event: {
1276 version: 1,
1277 type: "providerStatuses",
1278 payload: {
1279 providers: [
1280 {
1281 instanceId: ProviderInstanceId.make("codex"),
1282 driver: ProviderDriverKind.make("codex"),
1283 enabled: true,
1284 installed: true,
1285 version: "1.0.0",
1286 status: "ready",
1287 auth: { status: "authenticated" },
1288 checkedAt: "2026-08-27T00:00:00.000Z",
1289 models: [],
1290 slashCommands: [],
1291 skills: [],
1292 },
1293 ],
1294 },
1295 },
1296 expectedConfig: {
1297 providers: [
1298 {
1299 instanceId: ProviderInstanceId.make("codex"),
1300 driver: ProviderDriverKind.make("codex"),
1301 enabled: true,
1302 installed: true,
1303 version: "1.0.0",
1304 status: "ready",
1305 auth: { status: "authenticated" },
1306 checkedAt: "2026-08-27T00:00:00.000Z",
1307 models: [],
1308 slashCommands: [],
1309 skills: [],
1310 },
1311 ],
1312 },
1313 },
1314 {
1315 event: {
1316 version: 1,
1317 type: "settingsUpdated",
1318 payload: {
1319 settings: {
1320 ...DEFAULT_SERVER_SETTINGS,
1321 newWorktreesStartFromOrigin: !DEFAULT_SERVER_SETTINGS.newWorktreesStartFromOrigin,
1322 },
1323 },
1324 },
1325 expectedConfig: {
1326 settings: {
1327 ...DEFAULT_SERVER_SETTINGS,
1328 newWorktreesStartFromOrigin: !DEFAULT_SERVER_SETTINGS.newWorktreesStartFromOrigin,
1329 },
1330 },
1331 },
1332 ])(
1333 "preserves $event.type events and includes them in replay snapshots",
1334 ({ event, expectedConfig }) =>
1335 Effect.scoped(
1336 Effect.gen(function* () {
1337 const { factory, sockets } = yield* makeFactory();
1338 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1339 const readyFiber = yield* Effect.forkChild(session.ready);
1340 const socket = yield* awaitSocket(sockets);
1341 socket.open();
1342 yield* completeInitialConfig(socket);
1343 yield* Fiber.join(readyFiber);
1344
1345 const subscriber = yield* session
1346 .subscribeServerConfig({})
1347 .pipe(Stream.take(2), Stream.runCollect, Effect.forkChild);
1348 yield* Effect.yieldNow;
1349
1350 const request = yield* awaitRequest(socket);
1351 socket.serverMessage(
1352 encodeJson({
1353 _tag: "Chunk",
1354 requestId: request.id,
1355 values: [encodeServerConfigStreamEvent(event)],
1356 }),
1357 );
1358
1359 const events = Array.from(yield* Fiber.join(subscriber));
1360 expect(events[1]).toEqual(event);
1361
1362 const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead);
1363 expect(replay).toMatchObject({
1364 _tag: "Some",
1365 value: {
1366 type: "snapshot",
1367 config: expectedConfig,
1368 },
1369 });
1370 }),
1371 ),
1372 );
1373
1374 it.effect("rejects a server config for a different environment", () =>
1375 Effect.gen(function* () {
1376 const { factory, sockets } = yield* makeFactory();
1377 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1378 const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready));
1379 const configFiber = yield* session
1380 .subscribeServerConfig({})
1381 .pipe(Stream.runHead, Effect.flip, Effect.forkChild);
1382 const customConfigFiber = yield* session
1383 .subscribeServerConfig({ environmentThemes: true })
1384 .pipe(Stream.runHead, Effect.flip, Effect.forkChild);
1385 const socket = yield* awaitSocket(sockets);
1386 socket.open();
1387 yield* completeInitialConfig(socket, {
1388 ...ENCODED_SERVER_CONFIG,
1389 environment: {
1390 ...ENCODED_SERVER_CONFIG.environment,
1391 environmentId: "environment-2",
1392 },
1393 });
1394
1395 const error = yield* Fiber.join(readyFiber);
1396 expect(error).toMatchObject({
1397 reason: "configuration",
1398 message: "Connected environment environment-2 does not match environment-1.",
1399 });
1400 expect((yield* Fiber.join(configFiber))._tag).toBe("RpcClientError");
1401 expect((yield* Fiber.join(customConfigFiber))._tag).toBe("RpcClientError");
1402 }),
1403 );
1404
1405 it.effect("closes the session after 15 seconds without a frame from the server", () =>
1406 Effect.gen(function* () {
1407 const { factory, sockets } = yield* makeFactory();
1408 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1409 const readyFiber = yield* Effect.forkChild(session.ready);
1410 const closedFiber = yield* Effect.forkChild(Effect.flip(session.closed));
1411 const socket = yield* awaitSocket(sockets);
1412
1413 socket.open();
1414 yield* completeInitialConfig(socket);
1415 yield* Fiber.join(readyFiber);
1416
1417 yield* TestClock.adjust("15 seconds");
1418 expect(closedFiber.pollUnsafe()).toBeUndefined();
1419 expect(socket.sent.map((message) => decodeJson(message)).filter(isPing)).toEqual([
1420 { _tag: "Ping" },
1421 { _tag: "Ping" },
1422 { _tag: "Ping" },
1423 ]);
1424
1425 yield* TestClock.adjust("5 seconds");
1426 const error = yield* Fiber.join(closedFiber);
1427 expect(error).toBeInstanceOf(ConnectionTransientError);
1428 expect(error).toMatchObject({
1429 reason: "transport",
1430 detail: "Test environment stopped responding.",
1431 });
1432 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
1433 );
1434
1435 it.effect("keeps reading replies after closing a stream with a full buffer", () =>
1436 Effect.gen(function* () {
1437 const { factory, sockets } = yield* makeFactory();
1438 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1439 const readyFiber = yield* Effect.forkChild(session.ready);
1440 const socket = yield* awaitSocket(sockets);
1441 socket.open();
1442 yield* completeInitialConfig(socket);
1443 yield* Fiber.join(readyFiber);
1444
1445 // The consumer takes one event and stops pulling, so the stream buffer fills.
1446 const consuming = yield* Deferred.make<void>();
1447 const streamFiber = yield* session.client[WS_METHODS.subscribeServerConfig]({}).pipe(
1448 Stream.runForEach(() =>
1449 Deferred.succeed(consuming, undefined).pipe(Effect.andThen(Effect.never)),
1450 ),
1451 Effect.forkChild,
1452 );
1453 const streamRequest = yield* awaitRequest(socket, 1);
1454 const snapshot = { version: 1, type: "snapshot", config: ENCODED_SERVER_CONFIG };
1455 socket.serverMessage(
1456 encodeJson({
1457 _tag: "Chunk",
1458 requestId: streamRequest.id,
1459 values: Array.from({ length: 64 }, () => snapshot),
1460 }),
1461 );
1462 yield* Deferred.await(consuming);
1463 yield* Fiber.interrupt(streamFiber);
1464
1465 const probeFiber = yield* Effect.forkChild(session.probe);
1466 const probeRequest = yield* awaitRequest(socket, 2);
1467 expect(probeRequest).toMatchObject({ tag: WS_METHODS.serverProbe });
1468 socket.serverMessage(
1469 encodeJson({
1470 _tag: "Exit",
1471 requestId: probeRequest.id,
1472 exit: { _tag: "Success", value: {} },
1473 }),
1474 );
1475 yield* Fiber.join(probeFiber);
1476 }).pipe(Effect.scoped),
1477 );
1478
1479 it.effect("reaches ready when a newer server sends unknown config members", () =>
1480 Effect.gen(function* () {
1481 const { factory, sockets } = yield* makeFactory();
1482 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1483 const readyFiber = yield* Effect.forkChild(session.ready);
1484 const socket = yield* awaitSocket(sockets);
1485 socket.open();
1486
1487 const shortcut = {
1488 key: "p",
1489 metaKey: false,
1490 ctrlKey: false,
1491 shiftKey: false,
1492 altKey: false,
1493 modKey: true,
1494 };
1495 yield* completeInitialConfig(socket, {
1496 ...ENCODED_SERVER_CONFIG,
1497 keybindings: [
1498 { command: "someFuture.toggle", shortcut },
1499 { command: "terminal.toggle", shortcut },
1500 ],
1501 issues: [{ kind: "keybindings.future-issue", message: "From a newer server" }],
1502 availableEditors: ["some-future-editor", "zed"],
1503 });
1504 yield* Fiber.join(readyFiber);
1505
1506 const config = yield* session.initialConfig;
1507 expect(config.keybindings).toEqual([{ command: "terminal.toggle", shortcut }]);
1508 expect(config.issues).toEqual([]);
1509 expect(config.availableEditors).toEqual(["zed"]);
1510 }),
1511 );
1512
1513 it.effect("uses the legacy config RPC for probes when the server lacks the capability", () =>
1514 Effect.scoped(
1515 Effect.gen(function* () {
1516 const { factory, sockets } = yield* makeFactory();
1517 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
1518 const readyFiber = yield* Effect.forkChild(session.ready);
1519 const socket = yield* awaitSocket(sockets);
1520
1521 socket.open();
1522 yield* completeInitialConfig(socket, LEGACY_SERVER_CONFIG);
1523 yield* Fiber.join(readyFiber);
1524
1525 const probeFiber = yield* Effect.forkChild(session.probe);
1526 const probeRequest = yield* awaitRequest(socket, 1);
1527 expect(probeRequest).toMatchObject({
1528 _tag: "Request",
1529 tag: WS_METHODS.serverGetConfig,
1530 payload: {},
1531 });
1532 socket.serverMessage(
1533 encodeJson({
1534 _tag: "Exit",
1535 requestId: probeRequest.id,
1536 exit: {
1537 _tag: "Success",
1538 value: LEGACY_SERVER_CONFIG,
1539 },
1540 }),
1541 );
1542 yield* Fiber.join(probeFiber);
1543
1544 expect(
1545 socket.sent
1546 .map((message) => decodeJson(message))
1547 .filter(isRpcRequest)
1548 .map((request) => request.tag),
1549 ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverGetConfig]);
1550 }),
1551 ),
1552 );
1553
1554 it.effect.each([
1555 { relay: false, label: "direct" },
1556 { relay: true, label: "relay" },
1557 ])("fails readiness when the $label websocket never opens", ({ relay }) =>
1558 Effect.gen(function* () {
1559 const { factory, sockets } = yield* makeFactory();
1560
1561 const error = yield* Effect.scoped(
1562 Effect.gen(function* () {
1563 const session = yield* factory.connect(
1564 {
1565 ...PREPARED,
1566 target: relay
1567 ? new RelayConnectionTarget({
1568 environmentId: TARGET.environmentId,
1569 label: TARGET.label,
1570 })
1571 : TARGET,
1572 },
1573 Effect.succeed(PREPARED),
1574 );
1575 const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready));
1576 yield* awaitSocket(sockets);
1577
1578 yield* TestClock.adjust("15 seconds");
1579 return yield* Fiber.join(readyFiber);
1580 }),
1581 );
1582
1583 expect(error).toBeInstanceOf(ConnectionTransientError);
1584 expect(error).toMatchObject({
1585 reason: "transport",
1586 message: `Test environment could not establish a WebSocket connection.${relay ? ` ${NETWORK_BLOCKING_HINT}` : ""}`,
1587 });
1588 expect(sockets[0]?.readyState).toBe(TestWebSocket.CLOSED);
1589 }).pipe(Effect.provide(TestClock.layer())),
1590 );
1591});