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.gen(function* () {
454 const { factory, sockets } = yield* makeFactory();
455 const session = yield* factory.connect(PREPARED);
456 const readyFiber = yield* Effect.forkChild(session.ready);
457 const socket = yield* awaitSocket(sockets);
458 socket.open();
459 yield* completeInitialConfig(socket);
460 yield* Fiber.join(readyFiber);
461
462 const collectTwo = session.subscribeServerConfig({}).pipe(Stream.take(2), Stream.runCollect);
463 const firstSubscriber = yield* Effect.forkChild(collectTwo);
464 const secondSubscriber = yield* Effect.forkChild(collectTwo);
465 yield* Effect.yieldNow;
466
467 const shortcut = {
468 key: "k",
469 metaKey: false,
470 ctrlKey: false,
471 shiftKey: false,
472 altKey: false,
473 modKey: true,
474 };
475 const request = yield* awaitRequest(socket);
476 socket.serverMessage(
477 encodeJson({
478 _tag: "Chunk",
479 requestId: request.id,
480 values: [
481 {
482 version: 1,
483 type: "keybindingsUpdated",
484 payload: {
485 keybindings: [{ command: "terminal.toggle", shortcut }],
486 issues: [],
487 },
488 },
489 ],
490 }),
491 );
492
493 const firstEvents = Array.from(yield* Fiber.join(firstSubscriber));
494 const secondEvents = Array.from(yield* Fiber.join(secondSubscriber));
495 expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]);
496 expect(secondEvents).toEqual(firstEvents);
497
498 const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead);
499 expect(replay).toMatchObject({
500 _tag: "Some",
501 value: {
502 type: "snapshot",
503 config: { keybindings: [{ command: "terminal.toggle", shortcut }] },
504 },
505 });
506 }),
507 );
508
509 it.effect.each([
510 { usageLimitSources: true },
511 { environmentThemes: true, usageLimitSources: true },
512 ])("replays usage sources and removal with %j", (options) =>
513 Effect.gen(function* () {
514 const { factory, sockets } = yield* makeFactory(options);
515 const session = yield* factory.connect(PREPARED);
516 const ready = yield* Effect.forkChild(session.ready);
517 const socket = yield* awaitSocket(sockets);
518 socket.open();
519 yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), options);
520 yield* Fiber.join(ready);
521 const observed = yield* Queue.unbounded<ServerConfigStreamEventType>();
522 yield* session.subscribeServerConfig().pipe(
523 Stream.runForEach((event) => Queue.offer(observed, event)),
524 Effect.forkChild,
525 );
526 expect((yield* Queue.take(observed)).type).toBe("snapshot");
527 const themes: ServerConfigStreamEventType[] = options.environmentThemes
528 ? [{ version: 1, type: "environmentThemesUpdated", payload: { themes: [] } }]
529 : [];
530 for (const event of themes) {
531 yield* publishConfigEvents(socket, [event]);
532 expect(yield* Queue.take(observed)).toEqual(event);
533 }
534 const events: ServerConfigStreamEventType[] = [
535 SOURCE_EVENT,
536 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
537 SOURCE_EVENT,
538 ];
539 for (const event of events) {
540 yield* publishConfigEvents(socket, [event]);
541 expect(yield* Queue.take(observed)).toEqual(event);
542 const started = yield* Deferred.make<void>();
543 const replay = yield* session.subscribeServerConfig(options).pipe(
544 Stream.tap(() => Deferred.succeed(started, undefined)),
545 Stream.takeUntil((item) => item.type === "keybindingsUpdated"),
546 Stream.runCollect,
547 Effect.forkChild,
548 );
549 yield* Deferred.await(started);
550 const marker: ServerConfigStreamEventType = {
551 version: 1,
552 type: "keybindingsUpdated",
553 payload: { keybindings: [], issues: [] },
554 };
555 yield* publishConfigEvents(socket, [marker]);
556 expect(yield* Queue.take(observed)).toEqual(marker);
557 const replayed = Array.from(yield* Fiber.join(replay));
558 expect(replayed.slice(1)).toEqual([...themes, event, marker]);
559 let projection = applyServerConfigProjection(Option.none(), {
560 version: 1,
561 type: "snapshot",
562 config: SOURCE_SERVER_CONFIG,
563 });
564 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
565 for (const item of replayed) projection = applyServerConfigProjection(projection, item);
566 expect(Option.getOrThrow(projection).config.usageLimitSources).toEqual(
567 event.type === "usageLimitSourcesUpdated" && event.payload.sources.length > 0
568 ? event.payload.sources
569 : undefined,
570 );
571 }
572 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
573 1,
574 );
575 }),
576 );
577
578 it.effect("replays theme updates and deletion as authoritative events", () =>
579 Effect.gen(function* () {
580 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
581 const session = yield* factory.connect(PREPARED);
582 const readyFiber = yield* Effect.forkChild(session.ready);
583 const socket = yield* awaitSocket(sockets);
584 socket.open();
585 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {
586 environmentThemes: true,
587 });
588 yield* Fiber.join(readyFiber);
589
590 const firstThemes = [
591 {
592 id: "nightfall",
593 name: "Nightfall",
594 appearance: "dark" as const,
595 canvas: "#1a1b26",
596 accent: "#7aa2f7",
597 },
598 ];
599 const replacementThemes = [
600 {
601 id: "midnight",
602 name: "Midnight",
603 appearance: "dark" as const,
604 canvas: "#000000",
605 accent: "#ffffff",
606 },
607 ];
608 const subscriberStarted = yield* Deferred.make<void>();
609 const subscriber = yield* session.subscribeServerConfig({ environmentThemes: true }).pipe(
610 Stream.mapEffect((event) =>
611 Deferred.succeed(subscriberStarted, undefined).pipe(Effect.as(event)),
612 ),
613 Stream.take(4),
614 Stream.runCollect,
615 Effect.forkChild,
616 );
617 yield* Deferred.await(subscriberStarted);
618 yield* publishConfigEvents(socket, [
619 {
620 version: 1,
621 type: "environmentThemesUpdated",
622 payload: { themes: firstThemes },
623 },
624 {
625 version: 1,
626 type: "environmentThemesUpdated",
627 payload: { themes: replacementThemes },
628 },
629 {
630 version: 1,
631 type: "environmentThemesUpdated",
632 payload: { themes: [] },
633 },
634 ]);
635
636 const liveEvents = Array.from(yield* Fiber.join(subscriber));
637 expect(liveEvents.map((event) => event.type)).toEqual([
638 "snapshot",
639 "environmentThemesUpdated",
640 "environmentThemesUpdated",
641 "environmentThemesUpdated",
642 ]);
643 expect(liveEvents[2]).toMatchObject({ payload: { themes: replacementThemes } });
644
645 const replay = Array.from(
646 yield* session
647 .subscribeServerConfig({ environmentThemes: true })
648 .pipe(Stream.take(2), Stream.runCollect),
649 );
650 expect(replay.map((event) => event.type)).toEqual(["snapshot", "environmentThemesUpdated"]);
651 expect(replay[1]).toMatchObject({ payload: { themes: [] } });
652
653 let projection = applyServerConfigProjection(Option.none(), {
654 version: 1,
655 type: "snapshot",
656 config: THEME_SERVER_CONFIG,
657 });
658 projection = applyServerConfigProjection(projection, {
659 version: 1,
660 type: "environmentThemesUpdated",
661 payload: { themes: firstThemes },
662 });
663 for (const event of replay) {
664 projection = applyServerConfigProjection(projection, event);
665 }
666 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
667 }),
668 );
669
670 it.effect("recovers a slow subscriber after it misses theme and usage-source deletion", () =>
671 Effect.gen(function* () {
672 const { factory, sockets } = yield* makeFactory({
673 environmentThemes: true,
674 usageLimitSources: true,
675 });
676 const session = yield* factory.connect(PREPARED);
677 const readyFiber = yield* Effect.forkChild(session.ready);
678 const socket = yield* awaitSocket(sockets);
679 socket.open();
680 yield* completeInitialConfig(
681 socket,
682 encodeServerConfig({ ...SOURCE_SERVER_CONFIG, providers: [PROVIDER] }),
683 { environmentThemes: true, usageLimitSources: true },
684 );
685 yield* Fiber.join(readyFiber);
686
687 const slowSubscriberStarted = yield* Deferred.make<void>();
688 const releaseSlowSubscriber = yield* Deferred.make<void>();
689 let firstEvent = true;
690 const slowSubscriber = yield* session
691 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
692 .pipe(
693 Stream.mapEffect((event) => {
694 if (!firstEvent) return Effect.succeed(event);
695 firstEvent = false;
696 return Deferred.succeed(slowSubscriberStarted, undefined).pipe(
697 Effect.andThen(Deferred.await(releaseSlowSubscriber)),
698 Effect.as(event),
699 );
700 }),
701 Stream.take(4),
702 Stream.runCollect,
703 Effect.forkChild,
704 );
705 yield* Deferred.await(slowSubscriberStarted);
706
707 const firstThemes = [
708 {
709 id: "nightfall",
710 name: "Nightfall",
711 appearance: "dark" as const,
712 canvas: "#1a1b26",
713 accent: "#7aa2f7",
714 },
715 ];
716 const themeEvents: ServerConfigStreamEventType[] = [
717 {
718 version: 1,
719 type: "environmentThemesUpdated",
720 payload: { themes: firstThemes },
721 },
722 {
723 version: 1,
724 type: "environmentThemesUpdated",
725 payload: {
726 themes: [{ ...firstThemes[0]!, name: "Nightfall 2" }],
727 },
728 },
729 {
730 version: 1,
731 type: "environmentThemesUpdated",
732 payload: { themes: [] },
733 },
734 ];
735 const settingsEvents = Array.from({ length: 65 }, (): ServerConfigStreamEventType => ({
736 version: 1,
737 type: "settingsUpdated",
738 payload: { settings: DEFAULT_SERVER_SETTINGS },
739 }));
740 const sourceEvents: ServerConfigStreamEventType[] = [
741 SOURCE_EVENT,
742 { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [] } },
743 ];
744 // A delta carries no lists; a subscriber that misses it must still get its result.
745 const { models: _models, slashCommands: _commands, skills: _skills, ...status } = PROVIDER;
746 const providerEvent: ServerConfigStreamEventType = {
747 version: 1,
748 type: "providerStatusDeltas",
749 payload: {
750 providers: [{ ...status, checkedAt: "2026-04-10T00:05:00.000Z", message: "Checked" }],
751 },
752 };
753 const allEvents = [providerEvent, ...themeEvents, ...sourceEvents, ...settingsEvents];
754 const observedByFastSubscriber = yield* Queue.unbounded<ServerConfigStreamEventType>();
755 yield* session
756 .subscribeServerConfig({ environmentThemes: true, usageLimitSources: true })
757 .pipe(
758 Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)),
759 Effect.forkChild,
760 );
761 expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot");
762 for (const event of allEvents) {
763 yield* publishConfigEvents(socket, [event]);
764 expect(yield* Queue.take(observedByFastSubscriber)).toEqual(event);
765 }
766 yield* Deferred.succeed(releaseSlowSubscriber, undefined);
767
768 const recovered = Array.from(yield* Fiber.join(slowSubscriber));
769 expect(recovered.map((event) => event.type)).toEqual([
770 "snapshot",
771 "snapshot",
772 "environmentThemesUpdated",
773 "usageLimitSourcesUpdated",
774 ]);
775 expect(recovered[1]).toMatchObject({
776 config: {
777 providers: [{ ...PROVIDER, checkedAt: "2026-04-10T00:05:00.000Z", message: "Checked" }],
778 },
779 });
780 expect(recovered[2]).toMatchObject({ payload: { themes: [] } });
781 expect(recovered[3]).toMatchObject({ payload: { sources: [] } });
782
783 let projection = applyServerConfigProjection(Option.none(), {
784 version: 1,
785 type: "snapshot",
786 config: SOURCE_SERVER_CONFIG,
787 });
788 projection = applyServerConfigProjection(projection, themeEvents[0]!);
789 projection = applyServerConfigProjection(projection, SOURCE_EVENT);
790 for (const event of recovered.slice(1)) {
791 projection = applyServerConfigProjection(projection, event);
792 }
793 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
794 expect(Option.getOrThrow(projection).config.usageLimitSources).toBeUndefined();
795 }),
796 );
797
798 it.effect("closes the session when the config source dies", () =>
799 Effect.gen(function* () {
800 const { factory, sockets } = yield* makeFactory();
801 const session = yield* factory.connect(PREPARED);
802 const readyFiber = yield* Effect.forkChild(session.ready);
803 const socket = yield* awaitSocket(sockets);
804 socket.open();
805 yield* completeInitialConfig(socket);
806 yield* Fiber.join(readyFiber);
807
808 const closedFiber = yield* session.closed.pipe(Effect.exit, Effect.forkChild);
809 socket.serverMessage(
810 encodeJson({
811 _tag: "Defect",
812 defect: encodeDefect(new Error("config stream died")),
813 }),
814 );
815
816 const closed = yield* Fiber.join(closedFiber);
817 expect(Exit.isFailure(closed)).toBe(true);
818 if (Exit.isFailure(closed)) {
819 expect(Cause.hasDies(closed.cause)).toBe(true);
820 }
821 }),
822 );
823
824 it.effect.each([{ failure: "defect" as const }, { failure: "typed" as const }])(
825 "keeps durable config state alive after an owned $failure failure",
826 ({ failure }) =>
827 Effect.gen(function* () {
828 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
829 const firstSession = yield* factory.connect(PREPARED);
830 const firstReady = yield* Effect.forkChild(firstSession.ready);
831 const firstSocket = yield* awaitSocket(sockets);
832 firstSocket.open();
833 yield* completeInitialConfig(firstSocket, ENCODED_THEME_SERVER_CONFIG, {
834 environmentThemes: true,
835 });
836 yield* Fiber.join(firstReady);
837
838 const activeSession = yield* SubscriptionRef.make(Option.some(firstSession));
839 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
840 target: TARGET,
841 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
842 session: activeSession,
843 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
844 connect: Effect.void,
845 disconnect: Effect.void,
846 retryNow: Effect.void,
847 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
848 const cache = Persistence.EnvironmentCacheStore.of({
849 loadShell: () => Effect.succeedNone,
850 saveShell: () => Effect.void,
851 loadThread: () => Effect.succeedNone,
852 saveThread: () => Effect.void,
853 removeThread: () => Effect.void,
854 loadServerConfig: () => Effect.succeedNone,
855 saveServerConfig: () => Effect.void,
856 loadVcsRefs: () => Effect.succeedNone,
857 saveVcsRefs: () => Effect.void,
858 removeVcsRefs: () => Effect.void,
859 clearVcsRefs: () => Effect.void,
860 clear: () => Effect.void,
861 });
862 const configState = yield* makeEnvironmentServerConfigState({
863 environmentThemes: true,
864 }).pipe(
865 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
866 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
867 );
868 const awaitConfig = (predicate: (config: ServerConfigType) => boolean) =>
869 SubscriptionRef.changes(configState).pipe(
870 Stream.filter(Option.isSome),
871 Stream.map((projection) => projection.value.config),
872 Stream.filter(predicate),
873 Stream.runHead,
874 Effect.map(Option.getOrThrow),
875 );
876
877 const firstThemes = [
878 {
879 id: "first-theme",
880 name: "First theme",
881 appearance: "dark" as const,
882 canvas: "#111111",
883 accent: "#ffffff",
884 },
885 ];
886 const firstThemeState = yield* awaitConfig(
887 (config) => config.environmentThemes?.[0]?.id === "first-theme",
888 ).pipe(Effect.forkChild);
889 yield* publishConfigEvents(firstSocket, [
890 {
891 version: 1,
892 type: "environmentThemesUpdated",
893 payload: { themes: firstThemes },
894 },
895 ]);
896 expect((yield* Fiber.join(firstThemeState)).environmentThemes).toEqual(firstThemes);
897
898 const firstClosed = yield* firstSession.closed.pipe(Effect.exit, Effect.forkChild);
899 const firstRequest = yield* awaitRequest(firstSocket);
900 firstSocket.serverMessage(
901 failure === "defect"
902 ? encodeJson({
903 _tag: "Defect",
904 defect: encodeDefect(new Error("config stream died")),
905 })
906 : encodeJson({
907 _tag: "Exit",
908 requestId: firstRequest.id,
909 exit: {
910 _tag: "Failure",
911 cause: [
912 {
913 _tag: "Fail",
914 error: {
915 _tag: "EnvironmentAuthorizationError",
916 message: "config subscription rejected",
917 requiredScope: "orchestration:read",
918 },
919 },
920 ],
921 },
922 }),
923 );
924 const firstClosedExit = yield* Fiber.join(firstClosed);
925 expect(Exit.isFailure(firstClosedExit)).toBe(true);
926 if (failure === "typed" && Exit.isFailure(firstClosedExit)) {
927 expect(Cause.squash(firstClosedExit.cause)).toBeInstanceOf(ConnectionBlockedError);
928 expect(Cause.squash(firstClosedExit.cause)).toMatchObject({ reason: "permission" });
929 }
930 yield* SubscriptionRef.set(activeSession, Option.none());
931
932 const recoveredConfig = {
933 ...THEME_SERVER_CONFIG,
934 environment: {
935 ...THEME_SERVER_CONFIG.environment,
936 label: "Recovered environment",
937 },
938 } satisfies ServerConfigType;
939 const secondSession = yield* factory.connect(PREPARED);
940 const secondReady = yield* Effect.forkChild(secondSession.ready);
941 const secondSocket = yield* awaitSocket(sockets, 1);
942 secondSocket.open();
943 yield* completeInitialConfig(secondSocket, encodeServerConfig(recoveredConfig), {
944 environmentThemes: true,
945 });
946 yield* Fiber.join(secondReady);
947
948 const recoveredState = yield* awaitConfig(
949 (config) => config.environment.label === "Recovered environment",
950 ).pipe(Effect.forkChild);
951 yield* SubscriptionRef.set(activeSession, Option.some(secondSession));
952 expect((yield* Fiber.join(recoveredState)).environmentThemes).toEqual(firstThemes);
953
954 const recoveredThemes = [
955 {
956 id: "recovered-theme",
957 name: "Recovered theme",
958 appearance: "dark" as const,
959 canvas: "#000000",
960 accent: "#eeeeee",
961 },
962 ];
963 const liveRecoveredState = yield* awaitConfig(
964 (config) => config.environmentThemes?.[0]?.id === "recovered-theme",
965 ).pipe(Effect.forkChild);
966 yield* publishConfigEvents(secondSocket, [
967 {
968 version: 1,
969 type: "environmentThemesUpdated",
970 payload: { themes: recoveredThemes },
971 },
972 ]);
973 expect((yield* Fiber.join(liveRecoveredState)).environmentThemes).toEqual(recoveredThemes);
974 }),
975 );
976
977 it.effect("rejects a server config for a different environment", () =>
978 Effect.gen(function* () {
979 const { factory, sockets } = yield* makeFactory();
980 const session = yield* factory.connect(PREPARED);
981 const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready));
982 const configFiber = yield* session
983 .subscribeServerConfig({})
984 .pipe(Stream.runHead, Effect.flip, Effect.forkChild);
985 const customConfigFiber = yield* session
986 .subscribeServerConfig({ environmentThemes: true })
987 .pipe(Stream.runHead, Effect.flip, Effect.forkChild);
988 const socket = yield* awaitSocket(sockets);
989 socket.open();
990 yield* completeInitialConfig(socket, {
991 ...ENCODED_SERVER_CONFIG,
992 environment: {
993 ...ENCODED_SERVER_CONFIG.environment,
994 environmentId: "environment-2",
995 },
996 });
997
998 const error = yield* Fiber.join(readyFiber);
999 expect(error).toMatchObject({
1000 reason: "configuration",
1001 message: "Connected environment environment-2 does not match environment-1.",
1002 });
1003 expect((yield* Fiber.join(configFiber))._tag).toBe("RpcClientError");
1004 expect((yield* Fiber.join(customConfigFiber))._tag).toBe("RpcClientError");
1005 }),
1006 );
1007
1008 it.effect("keeps reading replies after closing a stream with a full buffer", () =>
1009 Effect.gen(function* () {
1010 const { factory, sockets } = yield* makeFactory();
1011 const session = yield* factory.connect(PREPARED);
1012 const readyFiber = yield* Effect.forkChild(session.ready);
1013 const socket = yield* awaitSocket(sockets);
1014 socket.open();
1015 yield* completeInitialConfig(socket);
1016 yield* Fiber.join(readyFiber);
1017
1018 const consuming = yield* Deferred.make<void>();
1019 const streamFiber = yield* session.client[WS_METHODS.subscribeServerConfig]({}).pipe(
1020 Stream.runForEach(() =>
1021 Deferred.succeed(consuming, undefined).pipe(Effect.andThen(Effect.never)),
1022 ),
1023 Effect.forkChild,
1024 );
1025 const streamRequest = yield* awaitRequest(socket, 1);
1026 const snapshot = { version: 1, type: "snapshot", config: ENCODED_SERVER_CONFIG };
1027 socket.serverMessage(
1028 encodeJson({
1029 _tag: "Chunk",
1030 requestId: streamRequest.id,
1031 values: Array.from({ length: 64 }, () => snapshot),
1032 }),
1033 );
1034 yield* Deferred.await(consuming);
1035 yield* Fiber.interrupt(streamFiber);
1036
1037 const probeFiber = yield* Effect.forkChild(session.probe);
1038 const probeRequest = yield* awaitRequest(socket, 2);
1039 expect(probeRequest).toMatchObject({ tag: WS_METHODS.serverProbe });
1040 socket.serverMessage(
1041 encodeJson({
1042 _tag: "Exit",
1043 requestId: probeRequest.id,
1044 exit: { _tag: "Success", value: {} },
1045 }),
1046 );
1047 yield* Fiber.join(probeFiber);
1048 }),
1049 );
1050
1051 it.effect("reaches ready when a newer server sends unknown config members", () =>
1052 Effect.gen(function* () {
1053 const { factory, sockets } = yield* makeFactory();
1054 const session = yield* factory.connect(PREPARED);
1055 const readyFiber = yield* Effect.forkChild(session.ready);
1056 const socket = yield* awaitSocket(sockets);
1057 socket.open();
1058
1059 const shortcut = {
1060 key: "p",
1061 metaKey: false,
1062 ctrlKey: false,
1063 shiftKey: false,
1064 altKey: false,
1065 modKey: true,
1066 };
1067 yield* completeInitialConfig(socket, {
1068 ...ENCODED_SERVER_CONFIG,
1069 keybindings: [
1070 { command: "someFuture.toggle", shortcut },
1071 { command: "terminal.toggle", shortcut },
1072 ],
1073 issues: [{ kind: "keybindings.future-issue", message: "From a newer server" }],
1074 availableEditors: ["some-future-editor", "zed"],
1075 });
1076 yield* Fiber.join(readyFiber);
1077
1078 const config = yield* session.initialConfig;
1079 expect(config.keybindings).toEqual([{ command: "terminal.toggle", shortcut }]);
1080 expect(config.issues).toEqual([]);
1081 expect(config.availableEditors).toEqual(["zed"]);
1082 }),
1083 );
1084});