1import {
2 DEFAULT_SERVER_SETTINGS,
3 EnvironmentId,
4 ServerConfig,
5 type ServerConfig as ServerConfigType,
6 ServerConfigStreamEvent,
7 type ServerConfigStreamEvent as ServerConfigStreamEventType,
8 ServerProvider,
9 WS_METHODS,
10 UsageLimitSourceId,
11} from "@t3tools/contracts";
12import { splitSocketFrame } from "@t3tools/shared/socketFrameParts";
13import { describe, expect, it } from "@effect/vitest";
14import * as Cause from "effect/Cause";
15import * as Deferred from "effect/Deferred";
16import * as Effect from "effect/Effect";
17import * as Exit from "effect/Exit";
18import * as Fiber from "effect/Fiber";
19import * as Layer from "effect/Layer";
20import * as Option from "effect/Option";
21import * as Queue from "effect/Queue";
22import * as Schema from "effect/Schema";
23import * as Stream from "effect/Stream";
24import * as SubscriptionRef from "effect/SubscriptionRef";
25import * as Socket from "effect/socket/Socket";
26import * as TestClock from "effect/testing/TestClock";
27
28import {
29 AVAILABLE_CONNECTION_STATE,
30 ConnectionBlockedError,
31 ConnectionTransientError,
32 PrimaryConnectionTarget,
33 type PreparedConnection,
34} from "../connection/model.ts";
35import * as EnvironmentSupervisor from "../connection/supervisor.ts";
36import * as Persistence from "../platform/persistence.ts";
37import * as RpcSession from "./session.ts";
38import { makeEnvironmentServerConfigState } from "../state/server.ts";
39import { applyServerConfigProjection } from "../state/serverConfigProjection.ts";
40
41type SocketEventType = "open" | "message" | "close" | "error";
42type SocketEvent = {
43 readonly code?: number;
44 readonly data?: unknown;
45 readonly reason?: string;
46 readonly type: SocketEventType;
47};
48type SocketListener = (event: SocketEvent) => void;
49
50class TestWebSocket {
51 static readonly CONNECTING = 0;
52 static readonly OPEN = 1;
53 static readonly CLOSING = 2;
54 static readonly CLOSED = 3;
55
56 readyState = TestWebSocket.CONNECTING;
57 readonly sent: string[] = [];
58 readonly url: string;
59 readonly sentChanges: Queue.Queue<void>;
60 private readonly listeners = new Map<SocketEventType, Set<SocketListener>>();
61
62 constructor(url: string, sentChanges: Queue.Queue<void>) {
63 this.url = url;
64 this.sentChanges = sentChanges;
65 }
66
67 addEventListener(type: SocketEventType, listener: SocketListener) {
68 const listeners = this.listeners.get(type) ?? new Set<SocketListener>();
69 listeners.add(listener);
70 this.listeners.set(type, listeners);
71 }
72
73 removeEventListener(type: SocketEventType, listener: SocketListener) {
74 this.listeners.get(type)?.delete(listener);
75 }
76
77 send(data: string) {
78 this.sent.push(data);
79 Queue.offerUnsafe(this.sentChanges, undefined);
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 encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
166const encodeServerConfig = Schema.encodeSync(ServerConfig);
167const encodeServerConfigStreamEvent = Schema.encodeSync(ServerConfigStreamEvent);
168const encodeDefect = Schema.encodeSync(Schema.Defect());
169const ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG);
170const THEME_SERVER_CONFIG: ServerConfigType = {
171 ...SERVER_CONFIG,
172 environment: {
173 ...SERVER_CONFIG.environment,
174 capabilities: {
175 ...SERVER_CONFIG.environment.capabilities,
176 environmentThemes: true,
177 },
178 },
179};
180const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG);
181const SOURCE_SERVER_CONFIG: ServerConfigType = {
182 ...THEME_SERVER_CONFIG,
183 environment: {
184 ...THEME_SERVER_CONFIG.environment,
185 capabilities: { ...THEME_SERVER_CONFIG.environment.capabilities, usageLimitSources: true },
186 },
187};
188const PROVIDER = Schema.decodeSync(ServerProvider)({
189 instanceId: "codex",
190 driver: "codex",
191 enabled: true,
192 installed: true,
193 version: "1.0.0",
194 status: "ready",
195 auth: { status: "authenticated" },
196 checkedAt: "2026-04-10T00:00:00.000Z",
197 models: [{ slug: "gpt", name: "GPT", isCustom: false, capabilities: null }],
198 slashCommands: [{ name: "compact" }],
199});
200const SOURCE_EVENT: ServerConfigStreamEventType = {
201 version: 1,
202 type: "usageLimitSourcesUpdated",
203 payload: {
204 sources: [
205 {
206 id: UsageLimitSourceId.make("proxy"),
207 kind: "cliproxy",
208 label: "Proxy",
209 checkedAt: "2026-09-04T00:00:00Z",
210 accounts: [],
211 },
212 ],
213 },
214};
215
216const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* (
217 options: RpcSession.RpcSessionOptions = {},
218) {
219 const sentChanges = yield* Queue.unbounded<void>();
220 const sockets = Object.assign([] as TestWebSocket[], { changes: yield* Queue.unbounded<void>() });
221 const layerConstructor = Layer.succeed(Socket.WebSocketConstructor, (url) => {
222 const socket = new TestWebSocket(url, sentChanges);
223 sockets.push(socket);
224 Queue.offerUnsafe(sockets.changes, undefined);
225 return socket as unknown as globalThis.WebSocket;
226 });
227 const layer = RpcSession.layer(options).pipe(Layer.provide(layerConstructor));
228 const factory = yield* RpcSession.RpcSessionFactory.pipe(Effect.provide(layer));
229 return { factory, sockets };
230});
231
232const awaitSocket = Effect.fn("TestRpcSessionFactory.awaitSocket")(function* (
233 sockets: ReadonlyArray<TestWebSocket> & { readonly changes: Queue.Queue<void> },
234 index = 0,
235) {
236 for (;;) {
237 const socket = sockets[index];
238 if (socket) {
239 return socket;
240 }
241 yield* Queue.take(sockets.changes);
242 }
243});
244
245const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* (
246 socket: TestWebSocket,
247 index = 0,
248) {
249 for (;;) {
250 const request = socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)[index];
251 if (request) {
252 return request;
253 }
254 yield* Queue.take(socket.sentChanges);
255 }
256});
257
258const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialConfig")(function* (
259 socket: TestWebSocket,
260 config: unknown = ENCODED_SERVER_CONFIG,
261 payload: unknown = {},
262) {
263 const request = yield* awaitRequest(socket);
264 expect(request).toMatchObject({
265 _tag: "Request",
266 tag: WS_METHODS.subscribeServerConfig,
267 payload,
268 });
269 socket.serverMessage(
270 encodeJson({
271 _tag: "Chunk",
272 requestId: request.id,
273 values: [{ version: 1, type: "snapshot", config }],
274 }),
275 );
276});
277
278const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents")(function* (
279 socket: TestWebSocket,
280 events: ReadonlyArray<ServerConfigStreamEventType>,
281) {
282 const request = yield* awaitRequest(socket);
283 socket.serverMessage(
284 encodeJson({
285 _tag: "Chunk",
286 requestId: request.id,
287 values: events.map((event) => encodeServerConfigStreamEvent(event)),
288 }),
289 );
290});
291
292describe("RpcSessionFactory", () => {
293 it.effect("connects while a large config snapshot downloads at 4000 bytes per second", () =>
294 Effect.gen(function* () {
295 const { factory, sockets } = yield* makeFactory();
296 const session = yield* factory.connect(PREPARED);
297 const readyFiber = yield* Effect.forkChild(session.ready);
298 const socket = yield* awaitSocket(sockets);
299 socket.open();
300 const request = yield* awaitRequest(socket);
301 // Multi-byte text, so parts must be sized in UTF-8 bytes and never split a surrogate pair.
302 const config = { ...SERVER_CONFIG, cwd: "/" + "é€😀x".repeat(32 * 1024) };
303 const frame = encodeJson({
304 _tag: "Chunk",
305 requestId: request.id,
306 values: [{ version: 1, type: "snapshot", config: encodeServerConfig(config) }],
307 });
308 const parts = new URL(socket.url).searchParams.has("frameParts")
309 ? splitSocketFrame(frame)
310 : [frame];
311
312 for (const part of parts) {
313 const bytes = new TextEncoder().encode(part);
314 yield* TestClock.adjust((bytes.length / 4000) * 1000);
315 socket.serverMessage(new TextDecoder().decode(bytes));
316 }
317 yield* Fiber.join(readyFiber);
318
319 expect(yield* session.initialConfig).toEqual(config);
320 }),
321 );
322
323 it.effect("owns one scoped websocket attempt and exposes readiness and closure", () =>
324 Effect.gen(function* () {
325 const { factory, sockets } = yield* makeFactory();
326 const session = yield* factory.connect(PREPARED);
327 const readyFiber = yield* Effect.forkChild(session.ready);
328 const socket = yield* awaitSocket(sockets);
329
330 expect(socket.url).toBe(`${PREPARED.socketUrl}&frameParts=1`);
331 socket.open();
332 yield* completeInitialConfig(socket);
333 yield* Fiber.join(readyFiber);
334
335 const config = yield* session.initialConfig;
336 expect(config).toEqual(SERVER_CONFIG);
337 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
338 1,
339 );
340
341 const probeFiber = yield* Effect.forkChild(session.probe);
342 const probeRequest = yield* awaitRequest(socket, 1);
343 expect(probeRequest).toMatchObject({
344 _tag: "Request",
345 tag: WS_METHODS.serverProbe,
346 payload: {},
347 });
348 socket.serverMessage(
349 encodeJson({
350 _tag: "Exit",
351 requestId: probeRequest.id,
352 exit: {
353 _tag: "Success",
354 value: {},
355 },
356 }),
357 );
358 yield* Fiber.join(probeFiber);
359
360 expect(
361 socket.sent
362 .map((message) => decodeJson(message))
363 .filter(isRpcRequest)
364 .map((request) => request.tag),
365 ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverProbe]);
366
367 socket.close(1012, "service restart");
368 const error = yield* Effect.flip(session.closed);
369 const configStreamError = yield* session
370 .subscribeServerConfig({})
371 .pipe(Stream.runDrain, Effect.flip);
372
373 expect(error).toBeInstanceOf(ConnectionTransientError);
374 expect(error).toMatchObject({
375 reason: "transport",
376 message: "Test environment disconnected.",
377 });
378 expect(configStreamError).toMatchObject({ _tag: "RpcClientError" });
379 yield* Effect.yieldNow;
380 expect(sockets).toHaveLength(1);
381 }),
382 );
383
384 it.effect("counts every inbound frame, including ones that answer no request", () =>
385 Effect.gen(function* () {
386 const { factory, sockets } = yield* makeFactory();
387 const session = yield* factory.connect(PREPARED);
388 const readyFiber = yield* Effect.forkChild(session.ready);
389 const socket = yield* awaitSocket(sockets);
390 socket.open();
391 yield* completeInitialConfig(socket);
392 yield* Fiber.join(readyFiber);
393 const before = yield* session.inboundFrames;
394
395 socket.serverMessage(encodeJson({ _tag: "Pong" }));
396
397 expect(yield* session.inboundFrames).toBe(before + 1);
398 }),
399 );
400
401 it.effect("closes the websocket when the session scope is released", () =>
402 Effect.gen(function* () {
403 const { factory, sockets } = yield* makeFactory();
404
405 yield* Effect.scoped(
406 Effect.gen(function* () {
407 const session = yield* factory.connect(PREPARED);
408 const readyFiber = yield* Effect.forkChild(session.ready);
409 const socket = yield* awaitSocket(sockets);
410 socket.open();
411 yield* completeInitialConfig(socket);
412 yield* Fiber.join(readyFiber);
413 }),
414 );
415
416 expect(sockets[0]?.readyState).toBe(TestWebSocket.CLOSED);
417 }),
418 );
419
420 it.effect("ends a stream at once when its socket closes", () =>
421 Effect.gen(function* () {
422 const { factory, sockets } = yield* makeFactory();
423 const session = yield* factory.connect(PREPARED);
424 const readyFiber = yield* Effect.forkChild(session.ready);
425 const socket = yield* awaitSocket(sockets);
426 socket.open();
427 yield* completeInitialConfig(socket);
428 yield* Fiber.join(readyFiber);
429 // Like the supervisor: the closed session interrupts its streams, and a
430 // stream interrupt must not wait for a socket that is gone.
431 const ended = yield* Effect.forkChild(
432 Effect.raceFirst(
433 Effect.flip(session.closed),
434 session.client[WS_METHODS.subscribeServerLifecycle]({}).pipe(
435 Stream.runDrain,
436 Effect.andThen(Effect.never),
437 ),
438 ),
439 );
440 yield* awaitRequest(socket, 1);
441 socket.close(1006);
442 yield* TestClock.adjust("999 millis");
443
444 expect(ended.pollUnsafe()).toBeDefined();
445 // A request made after the close still fails instead of waiting forever.
446 const lateProbe = yield* Effect.forkChild(session.probe);
447 yield* TestClock.adjust("999 millis");
448 expect(lateProbe.pollUnsafe()?._tag).toBe("Failure");
449 }).pipe(Effect.scoped, Effect.provide(TestClock.layer())),
450 );
451
452 it.effect("replays current config and broadcasts updates to every subscriber", () =>
453 Effect.scoped(
454 Effect.gen(function* () {
455 const { factory, sockets } = yield* makeFactory();
456 const session = yield* factory.connect(PREPARED);
457 const readyFiber = yield* Effect.forkChild(session.ready);
458 const socket = yield* awaitSocket(sockets);
459 socket.open();
460 yield* completeInitialConfig(socket);
461 yield* Fiber.join(readyFiber);
462
463 const collectTwo = session
464 .subscribeServerConfig({})
465 .pipe(Stream.take(2), Stream.runCollect);
466 const firstSubscriber = yield* Effect.forkChild(collectTwo);
467 const secondSubscriber = yield* Effect.forkChild(collectTwo);
468 yield* Effect.yieldNow;
469
470 const shortcut = {
471 key: "k",
472 metaKey: false,
473 ctrlKey: false,
474 shiftKey: false,
475 altKey: false,
476 modKey: true,
477 };
478 const request = yield* awaitRequest(socket);
479 socket.serverMessage(
480 encodeJson({
481 _tag: "Chunk",
482 requestId: request.id,
483 values: [
484 {
485 version: 1,
486 type: "keybindingsUpdated",
487 payload: {
488 keybindings: [{ command: "terminal.toggle", shortcut }],
489 issues: [],
490 },
491 },
492 ],
493 }),
494 );
495
496 const firstEvents = Array.from(yield* Fiber.join(firstSubscriber));
497 const secondEvents = Array.from(yield* Fiber.join(secondSubscriber));
498 expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]);
499 expect(secondEvents).toEqual(firstEvents);
500
501 const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead);
502 expect(replay).toMatchObject({
503 _tag: "Some",
504 value: {
505 type: "snapshot",
506 config: { keybindings: [{ command: "terminal.toggle", shortcut }] },
507 },
508 });
509 }),
510 ),
511 );
512
513 it.effect.each([
514 { usageLimitSources: true },
515 { environmentThemes: true, usageLimitSources: true },
516 ])("replays usage sources and removal with %j", (options) =>
517 Effect.scoped(
518 Effect.gen(function* () {
519 const { factory, sockets } = yield* makeFactory(options);
520 const session = yield* factory.connect(PREPARED);
521 const ready = yield* Effect.forkChild(session.ready);
522 const socket = yield* awaitSocket(sockets);
523 socket.open();
524 yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), options);
525 yield* Fiber.join(ready);
526 const observed = yield* Queue.unbounded<ServerConfigStreamEventType>();
527 yield* session.subscribeServerConfig().pipe(
528 Stream.runForEach((event) => Queue.offer(observed, event)),
529 Effect.forkChild,
530 );
531 expect((yield* Queue.take(observed)).type).toBe("snapshot");
532 const themes: ServerConfigStreamEventType[] = options.environmentThemes
533 ? [{ version: 1, type: "environmentThemesUpdated", payload: { themes: [] } }]
534 : [];
535 for (const event of themes) {
536 yield* publishConfigEvents(socket, [event]);
537 expect(yield* Queue.take(observed)).toEqual(event);
538 }
539 const events: ServerConfigStreamEventType[] = [
540 SOURCE_EVENT,
541 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
542 SOURCE_EVENT,
543 ];
544 for (const event of events) {
545 yield* publishConfigEvents(socket, [event]);
546 expect(yield* Queue.take(observed)).toEqual(event);
547 const started = yield* Deferred.make<void>();
548 const replay = yield* session.subscribeServerConfig(options).pipe(
549 Stream.tap(() => Deferred.succeed(started, undefined)),
550 Stream.takeUntil((item) => item.type === "keybindingsUpdated"),
551 Stream.runCollect,
552 Effect.forkChild,
553 );
554 yield* Deferred.await(started);
555 const marker: ServerConfigStreamEventType = {
556 version: 1,
557 type: "keybindingsUpdated",
558 payload: { keybindings: [], issues: [] },
559 };
560 yield* publishConfigEvents(socket, [marker]);
561 expect(yield* Queue.take(observed)).toEqual(marker);
562 const replayed = Array.from(yield* Fiber.join(replay));
563 expect(replayed.slice(1)).toEqual([...themes, event, marker]);
564 let projection = applyServerConfigProjection(Option.none(), {
565 version: 1,
566 type: "snapshot",
567 config: SOURCE_SERVER_CONFIG,
568 });
569 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
570 for (const item of replayed) projection = applyServerConfigProjection(projection, item);
571 expect(Option.getOrThrow(projection).config.usageLimitSources).toEqual(
572 event.type === "usageLimitSourcesUpdated" && event.payload.sources.length > 0
573 ? event.payload.sources
574 : undefined,
575 );
576 }
577 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
578 1,
579 );
580 }),
581 ),
582 );
583
584 it.effect("replays theme updates and deletion as authoritative events", () =>
585 Effect.scoped(
586 Effect.gen(function* () {
587 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
588 const session = yield* factory.connect(PREPARED);
589 const readyFiber = yield* Effect.forkChild(session.ready);
590 const socket = yield* awaitSocket(sockets);
591 socket.open();
592 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {
593 environmentThemes: true,
594 });
595 yield* Fiber.join(readyFiber);
596
597 const firstThemes = [
598 {
599 id: "nightfall",
600 name: "Nightfall",
601 appearance: "dark" as const,
602 canvas: "#1a1b26",
603 accent: "#7aa2f7",
604 },
605 ];
606 const replacementThemes = [
607 {
608 id: "midnight",
609 name: "Midnight",
610 appearance: "dark" as const,
611 canvas: "#000000",
612 accent: "#ffffff",
613 },
614 ];
615 const subscriberStarted = yield* Deferred.make<void>();
616 const subscriber = yield* session.subscribeServerConfig({ environmentThemes: true }).pipe(
617 Stream.mapEffect((event) =>
618 Deferred.succeed(subscriberStarted, undefined).pipe(Effect.as(event)),
619 ),
620 Stream.take(4),
621 Stream.runCollect,
622 Effect.forkChild,
623 );
624 yield* Deferred.await(subscriberStarted);
625 yield* publishConfigEvents(socket, [
626 {
627 version: 1,
628 type: "environmentThemesUpdated",
629 payload: { themes: firstThemes },
630 },
631 {
632 version: 1,
633 type: "environmentThemesUpdated",
634 payload: { themes: replacementThemes },
635 },
636 {
637 version: 1,
638 type: "environmentThemesUpdated",
639 payload: { themes: [] },
640 },
641 ]);
642
643 const liveEvents = Array.from(yield* Fiber.join(subscriber));
644 expect(liveEvents.map((event) => event.type)).toEqual([
645 "snapshot",
646 "environmentThemesUpdated",
647 "environmentThemesUpdated",
648 "environmentThemesUpdated",
649 ]);
650 expect(liveEvents[2]).toMatchObject({ payload: { themes: replacementThemes } });
651
652 const replay = Array.from(
653 yield* session
654 .subscribeServerConfig({ environmentThemes: true })
655 .pipe(Stream.take(2), Stream.runCollect),
656 );
657 expect(replay.map((event) => event.type)).toEqual(["snapshot", "environmentThemesUpdated"]);
658 expect(replay[1]).toMatchObject({ payload: { themes: [] } });
659
660 let projection = applyServerConfigProjection(Option.none(), {
661 version: 1,
662 type: "snapshot",
663 config: THEME_SERVER_CONFIG,
664 });
665 projection = applyServerConfigProjection(projection, {
666 version: 1,
667 type: "environmentThemesUpdated",
668 payload: { themes: firstThemes },
669 });
670 for (const event of replay) {
671 projection = applyServerConfigProjection(projection, event);
672 }
673 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
674 }),
675 ),
676 );
677
678 it.effect("recovers a slow subscriber after it misses theme and usage-source deletion", () =>
679 Effect.scoped(
680 Effect.gen(function* () {
681 const { factory, sockets } = yield* makeFactory({
682 environmentThemes: true,
683 usageLimitSources: true,
684 });
685 const session = yield* factory.connect(PREPARED);
686 const readyFiber = yield* Effect.forkChild(session.ready);
687 const socket = yield* awaitSocket(sockets);
688 socket.open();
689 yield* completeInitialConfig(
690 socket,
691 encodeServerConfig({ ...SOURCE_SERVER_CONFIG, providers: [PROVIDER] }),
692 { environmentThemes: true, usageLimitSources: true },
693 );
694 yield* Fiber.join(readyFiber);
695
696 const slowSubscriberStarted = yield* Deferred.make<void>();
697 const releaseSlowSubscriber = yield* Deferred.make<void>();
698 let firstEvent = true;
699 const slowSubscriber = yield* session
700 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
701 .pipe(
702 Stream.mapEffect((event) => {
703 if (!firstEvent) return Effect.succeed(event);
704 firstEvent = false;
705 return Deferred.succeed(slowSubscriberStarted, undefined).pipe(
706 Effect.andThen(Deferred.await(releaseSlowSubscriber)),
707 Effect.as(event),
708 );
709 }),
710 Stream.take(4),
711 Stream.runCollect,
712 Effect.forkChild,
713 );
714 yield* Deferred.await(slowSubscriberStarted);
715
716 const firstThemes = [
717 {
718 id: "nightfall",
719 name: "Nightfall",
720 appearance: "dark" as const,
721 canvas: "#1a1b26",
722 accent: "#7aa2f7",
723 },
724 ];
725 const themeEvents: ServerConfigStreamEventType[] = [
726 {
727 version: 1,
728 type: "environmentThemesUpdated",
729 payload: { themes: firstThemes },
730 },
731 {
732 version: 1,
733 type: "environmentThemesUpdated",
734 payload: {
735 themes: [{ ...firstThemes[0]!, name: "Nightfall 2" }],
736 },
737 },
738 {
739 version: 1,
740 type: "environmentThemesUpdated",
741 payload: { themes: [] },
742 },
743 ];
744 const settingsEvents = Array.from({ length: 65 }, (): ServerConfigStreamEventType => ({
745 version: 1,
746 type: "settingsUpdated",
747 payload: { settings: DEFAULT_SERVER_SETTINGS },
748 }));
749 const sourceEvents: ServerConfigStreamEventType[] = [
750 SOURCE_EVENT,
751 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
752 ];
753 // A delta carries no lists; a subscriber that misses it must still get its result.
754 const { models: _models, slashCommands: _commands, skills: _skills, ...status } = PROVIDER;
755 const providerEvent: ServerConfigStreamEventType = {
756 version: 1,
757 type: "providerStatusDeltas",
758 payload: {
759 providers: [{ ...status, checkedAt: "2026-04-10T00:05:00.000Z", message: "Checked" }],
760 },
761 };
762 const allEvents = [providerEvent, ...themeEvents, ...sourceEvents, ...settingsEvents];
763 const observedByFastSubscriber = yield* Queue.unbounded<ServerConfigStreamEventType>();
764 yield* session
765 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
766 .pipe(
767 Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)),
768 Effect.forkChild,
769 );
770 expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot");
771 for (const event of allEvents) {
772 yield* publishConfigEvents(socket, [event]);
773 expect(yield* Queue.take(observedByFastSubscriber)).toEqual(event);
774 }
775 yield* Deferred.succeed(releaseSlowSubscriber, undefined);
776
777 const recovered = Array.from(yield* Fiber.join(slowSubscriber));
778 expect(recovered.map((event) => event.type)).toEqual([
779 "snapshot",
780 "snapshot",
781 "environmentThemesUpdated",
782 "usageLimitSourcesUpdated",
783 ]);
784 expect(recovered[1]).toMatchObject({
785 config: {
786 providers: [{ ...PROVIDER, checkedAt: "2026-04-10T00:05:00.000Z", message: "Checked" }],
787 },
788 });
789 expect(recovered[2]).toMatchObject({ payload: { themes: [] } });
790 expect(recovered[3]).toMatchObject({ payload: { sources: [] } });
791
792 let projection = applyServerConfigProjection(Option.none(), {
793 version: 1,
794 type: "snapshot",
795 config: SOURCE_SERVER_CONFIG,
796 });
797 projection = applyServerConfigProjection(projection, themeEvents[0]!);
798 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
799 for (const event of recovered.slice(1)) {
800 projection = applyServerConfigProjection(projection, event);
801 }
802 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
803 expect(Option.getOrThrow(projection).config.usageLimitSources).toBeUndefined();
804 }),
805 ),
806 );
807
808 it.effect("closes the session when the config source dies", () =>
809 Effect.scoped(
810 Effect.gen(function* () {
811 const { factory, sockets } = yield* makeFactory();
812 const session = yield* factory.connect(PREPARED);
813 const readyFiber = yield* Effect.forkChild(session.ready);
814 const socket = yield* awaitSocket(sockets);
815 socket.open();
816 yield* completeInitialConfig(socket);
817 yield* Fiber.join(readyFiber);
818
819 const closedFiber = yield* session.closed.pipe(Effect.exit, Effect.forkChild);
820 socket.serverMessage(
821 encodeJson({
822 _tag: "Defect",
823 defect: encodeDefect(new Error("config stream died")),
824 }),
825 );
826
827 const closed = yield* Fiber.join(closedFiber);
828 expect(Exit.isFailure(closed)).toBe(true);
829 if (Exit.isFailure(closed)) {
830 expect(Cause.hasDies(closed.cause)).toBe(true);
831 }
832 }),
833 ),
834 );
835
836 it.effect.each([{ failure: "defect" as const }, { failure: "typed" as const }])(
837 "keeps durable config state alive after an owned $failure failure",
838 ({ failure }) =>
839 Effect.scoped(
840 Effect.gen(function* () {
841 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
842 const firstSession = yield* factory.connect(PREPARED);
843 const firstReady = yield* Effect.forkChild(firstSession.ready);
844 const firstSocket = yield* awaitSocket(sockets);
845 firstSocket.open();
846 yield* completeInitialConfig(firstSocket, ENCODED_THEME_SERVER_CONFIG, {
847 environmentThemes: true,
848 });
849 yield* Fiber.join(firstReady);
850
851 const activeSession = yield* SubscriptionRef.make(Option.some(firstSession));
852 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
853 target: TARGET,
854 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
855 session: activeSession,
856 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
857 connect: Effect.void,
858 disconnect: Effect.void,
859 retryNow: Effect.void,
860 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
861 const cache = Persistence.EnvironmentCacheStore.of({
862 loadShell: () => Effect.succeedNone,
863 saveShell: () => Effect.void,
864 loadThread: () => Effect.succeedNone,
865 saveThread: () => Effect.void,
866 removeThread: () => Effect.void,
867 loadServerConfig: () => Effect.succeedNone,
868 saveServerConfig: () => Effect.void,
869 loadVcsRefs: () => Effect.succeedNone,
870 saveVcsRefs: () => Effect.void,
871 removeVcsRefs: () => Effect.void,
872 clearVcsRefs: () => Effect.void,
873 clear: () => Effect.void,
874 });
875 const configState = yield* makeEnvironmentServerConfigState({
876 environmentThemes: true,
877 }).pipe(
878 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
879 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
880 );
881 const awaitConfig = (predicate: (config: ServerConfigType) => boolean) =>
882 SubscriptionRef.changes(configState).pipe(
883 Stream.filter(Option.isSome),
884 Stream.map((projection) => projection.value.config),
885 Stream.filter(predicate),
886 Stream.runHead,
887 Effect.map(Option.getOrThrow),
888 );
889
890 const firstThemes = [
891 {
892 id: "first-theme",
893 name: "First theme",
894 appearance: "dark" as const,
895 canvas: "#111111",
896 accent: "#ffffff",
897 },
898 ];
899 const firstThemeState = yield* awaitConfig(
900 (config) => config.environmentThemes?.[0]?.id === "first-theme",
901 ).pipe(Effect.forkChild);
902 yield* publishConfigEvents(firstSocket, [
903 {
904 version: 1,
905 type: "environmentThemesUpdated",
906 payload: { themes: firstThemes },
907 },
908 ]);
909 expect((yield* Fiber.join(firstThemeState)).environmentThemes).toEqual(firstThemes);
910
911 const firstClosed = yield* firstSession.closed.pipe(Effect.exit, Effect.forkChild);
912 const firstRequest = yield* awaitRequest(firstSocket);
913 firstSocket.serverMessage(
914 failure === "defect"
915 ? encodeJson({
916 _tag: "Defect",
917 defect: encodeDefect(new Error("config stream died")),
918 })
919 : encodeJson({
920 _tag: "Exit",
921 requestId: firstRequest.id,
922 exit: {
923 _tag: "Failure",
924 cause: [
925 {
926 _tag: "Fail",
927 error: {
928 _tag: "EnvironmentAuthorizationError",
929 message: "config subscription rejected",
930 requiredScope: "orchestration:read",
931 },
932 },
933 ],
934 },
935 }),
936 );
937 const firstClosedExit = yield* Fiber.join(firstClosed);
938 expect(Exit.isFailure(firstClosedExit)).toBe(true);
939 if (failure === "typed" && Exit.isFailure(firstClosedExit)) {
940 expect(Cause.squash(firstClosedExit.cause)).toBeInstanceOf(ConnectionBlockedError);
941 expect(Cause.squash(firstClosedExit.cause)).toMatchObject({ reason: "permission" });
942 }
943 yield* SubscriptionRef.set(activeSession, Option.none());
944
945 const recoveredConfig = {
946 ...THEME_SERVER_CONFIG,
947 environment: {
948 ...THEME_SERVER_CONFIG.environment,
949 label: "Recovered environment",
950 },
951 } satisfies ServerConfigType;
952 const secondSession = yield* factory.connect(PREPARED);
953 const secondReady = yield* Effect.forkChild(secondSession.ready);
954 const secondSocket = yield* awaitSocket(sockets, 1);
955 secondSocket.open();
956 yield* completeInitialConfig(secondSocket, encodeServerConfig(recoveredConfig), {
957 environmentThemes: true,
958 });
959 yield* Fiber.join(secondReady);
960
961 const recoveredState = yield* awaitConfig(
962 (config) => config.environment.label === "Recovered environment",
963 ).pipe(Effect.forkChild);
964 yield* SubscriptionRef.set(activeSession, Option.some(secondSession));
965 expect((yield* Fiber.join(recoveredState)).environmentThemes).toEqual(firstThemes);
966
967 const recoveredThemes = [
968 {
969 id: "recovered-theme",
970 name: "Recovered theme",
971 appearance: "dark" as const,
972 canvas: "#000000",
973 accent: "#eeeeee",
974 },
975 ];
976 const liveRecoveredState = yield* awaitConfig(
977 (config) => config.environmentThemes?.[0]?.id === "recovered-theme",
978 ).pipe(Effect.forkChild);
979 yield* publishConfigEvents(secondSocket, [
980 {
981 version: 1,
982 type: "environmentThemesUpdated",
983 payload: { themes: recoveredThemes },
984 },
985 ]);
986 expect((yield* Fiber.join(liveRecoveredState)).environmentThemes).toEqual(
987 recoveredThemes,
988 );
989 }),
990 ),
991 );
992
993 it.effect("rejects a server config for a different environment", () =>
994 Effect.gen(function* () {
995 const { factory, sockets } = yield* makeFactory();
996 const session = yield* factory.connect(PREPARED);
997 const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready));
998 const configFiber = yield* session
999 .subscribeServerConfig({})
1000 .pipe(Stream.runHead, Effect.flip, Effect.forkChild);
1001 const customConfigFiber = yield* session
1002 .subscribeServerConfig({ environmentThemes: true })
1003 .pipe(Stream.runHead, Effect.flip, Effect.forkChild);
1004 const socket = yield* awaitSocket(sockets);
1005 socket.open();
1006 yield* completeInitialConfig(socket, {
1007 ...ENCODED_SERVER_CONFIG,
1008 environment: {
1009 ...ENCODED_SERVER_CONFIG.environment,
1010 environmentId: "environment-2",
1011 },
1012 });
1013
1014 const error = yield* Fiber.join(readyFiber);
1015 expect(error).toMatchObject({
1016 reason: "configuration",
1017 message: "Connected environment environment-2 does not match environment-1.",
1018 });
1019 expect((yield* Fiber.join(configFiber))._tag).toBe("RpcClientError");
1020 expect((yield* Fiber.join(customConfigFiber))._tag).toBe("RpcClientError");
1021 }),
1022 );
1023
1024 it.effect("keeps reading replies after closing a stream with a full buffer", () =>
1025 Effect.gen(function* () {
1026 const { factory, sockets } = yield* makeFactory();
1027 const session = yield* factory.connect(PREPARED);
1028 const readyFiber = yield* Effect.forkChild(session.ready);
1029 const socket = yield* awaitSocket(sockets);
1030 socket.open();
1031 yield* completeInitialConfig(socket);
1032 yield* Fiber.join(readyFiber);
1033
1034 const consuming = yield* Deferred.make<void>();
1035 const streamFiber = yield* session.client[WS_METHODS.subscribeServerConfig]({}).pipe(
1036 Stream.runForEach(() =>
1037 Deferred.succeed(consuming, undefined).pipe(Effect.andThen(Effect.never)),
1038 ),
1039 Effect.forkChild,
1040 );
1041 const streamRequest = yield* awaitRequest(socket, 1);
1042 const snapshot = { version: 1, type: "snapshot", config: ENCODED_SERVER_CONFIG };
1043 socket.serverMessage(
1044 encodeJson({
1045 _tag: "Chunk",
1046 requestId: streamRequest.id,
1047 values: Array.from({ length: 64 }, () => snapshot),
1048 }),
1049 );
1050 yield* Deferred.await(consuming);
1051 yield* Fiber.interrupt(streamFiber);
1052
1053 const probeFiber = yield* Effect.forkChild(session.probe);
1054 const probeRequest = yield* awaitRequest(socket, 2);
1055 expect(probeRequest).toMatchObject({ tag: WS_METHODS.serverProbe });
1056 socket.serverMessage(
1057 encodeJson({
1058 _tag: "Exit",
1059 requestId: probeRequest.id,
1060 exit: { _tag: "Success", value: {} },
1061 }),
1062 );
1063 yield* Fiber.join(probeFiber);
1064 }).pipe(Effect.scoped),
1065 );
1066
1067 it.effect("reaches ready when a newer server sends unknown config members", () =>
1068 Effect.gen(function* () {
1069 const { factory, sockets } = yield* makeFactory();
1070 const session = yield* factory.connect(PREPARED);
1071 const readyFiber = yield* Effect.forkChild(session.ready);
1072 const socket = yield* awaitSocket(sockets);
1073 socket.open();
1074
1075 const shortcut = {
1076 key: "p",
1077 metaKey: false,
1078 ctrlKey: false,
1079 shiftKey: false,
1080 altKey: false,
1081 modKey: true,
1082 };
1083 yield* completeInitialConfig(socket, {
1084 ...ENCODED_SERVER_CONFIG,
1085 keybindings: [
1086 { command: "someFuture.toggle", shortcut },
1087 { command: "terminal.toggle", shortcut },
1088 ],
1089 issues: [{ kind: "keybindings.future-issue", message: "From a newer server" }],
1090 availableEditors: ["some-future-editor", "zed"],
1091 });
1092 yield* Fiber.join(readyFiber);
1093
1094 const config = yield* session.initialConfig;
1095 expect(config.keybindings).toEqual([{ command: "terminal.toggle", shortcut }]);
1096 expect(config.issues).toEqual([]);
1097 expect(config.availableEditors).toEqual(["zed"]);
1098 }),
1099 );
1100});