1import {
2 EnvironmentId,
3 type ServerConfig,
4 type ServerConfigStreamEvent,
5 type ServerLifecycleWelcomePayload,
6 WS_METHODS,
7} from "@t3tools/contracts";
8import { describe, expect, it } from "@effect/vitest";
9import * as Cause from "effect/Cause";
10import * as Deferred from "effect/Deferred";
11import * as Duration from "effect/Duration";
12import * as Effect from "effect/Effect";
13import * as Exit from "effect/Exit";
14import * as Fiber from "effect/Fiber";
15import * as Option from "effect/Option";
16import * as Queue from "effect/Queue";
17import * as Ref from "effect/Ref";
18import * as Scope from "effect/Scope";
19import * as Stream from "effect/Stream";
20import * as SubscriptionRef from "effect/SubscriptionRef";
21import * as TestClock from "effect/testing/TestClock";
22import { RpcClientError } from "effect/rpc";
23import * as Socket from "effect/socket/Socket";
24
25import {
26 AVAILABLE_CONNECTION_STATE,
27 PrimaryConnectionTarget,
28 type PreparedConnection,
29} from "../connection/model.ts";
30import * as EnvironmentSupervisor from "../connection/supervisor.ts";
31import * as Persistence from "../platform/persistence.ts";
32import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
33import type { RpcSession } from "../rpc/session.ts";
34import {
35 applyServerWelcomeEvent,
36 makeEnvironmentServerWelcomeState,
37 makeEnvironmentServerConfigState,
38 isLegacyUpdateHandoffLoss,
39 matchesServerUpdateReadyEvent,
40 matchesServerUpdateResumeEvent,
41 nudgeReconnectDuringUpdateRestart,
42 resolveServerConfigValue,
43 resolveServerWelcomeState,
44 resolveServerUpdateProgressResult,
45 serverUpdateStateForProgressEvent,
46 serverUpdateStateForServerVersion,
47 validateServerUpdateReadyEvent,
48 waitForNextEnvironmentReconnect,
49 waitForDesktopUpdateTarget,
50 runDesktopCommitWithReconnectObserver,
51} from "./server.ts";
52import { applyServerConfigProjection } from "./serverConfigProjection.ts";
53
54const CONFIG = {
55 availableEditors: [],
56 issues: [],
57 keybindings: {},
58 keybindingsConfigPath: null,
59 observability: null,
60 providers: [],
61 settings: {},
62 // Capabilities drive version-skew behaviour in the projection, so the
63 // fixture carries them rather than leaving the field absent.
64 environment: { capabilities: { environmentThemes: true } },
65} as unknown as ServerConfig;
66
67const snapshotEvent = (config: ServerConfig): ServerConfigStreamEvent => ({
68 version: 1,
69 type: "snapshot",
70 config,
71});
72
73const TARGET = new PrimaryConnectionTarget({
74 environmentId: EnvironmentId.make("environment-1"),
75 label: "Test environment",
76 httpBaseUrl: "https://environment.example.test",
77 wsBaseUrl: "wss://environment.example.test",
78});
79
80function session(client: WsRpcProtocolClient): RpcSession {
81 return {
82 client,
83 initialConfig: Effect.succeed(CONFIG),
84 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
85 ready: Effect.void,
86 probe: Effect.void,
87 closed: Effect.never,
88 };
89}
90
91describe("update restart reconnect nudges", () => {
92 it.effect("retries a desktop commit that was lost before delivery", () =>
93 Effect.gen(function* () {
94 const readyEvents =
95 yield* Queue.unbounded<Parameters<typeof matchesServerUpdateReadyEvent>[1]>();
96 const ready = (serverVersion: string) =>
97 ({
98 version: 1 as const,
99 sequence: 1,
100 type: "ready" as const,
101 payload: {
102 at: "2026-09-01T00:00:00.000Z",
103 environment: { serverVersion },
104 },
105 }) as Parameters<typeof matchesServerUpdateReadyEvent>[1];
106 yield* Queue.offerAll(readyEvents, [ready("0.0.30"), ready("0.0.31")]);
107 const retries = yield* Ref.make(0);
108 const disconnect = new RpcClientError.RpcClientError({
109 reason: new Socket.SocketCloseError({ code: 1006 }),
110 });
111
112 const result = yield* waitForDesktopUpdateTarget(
113 "0.0.31",
114 Queue.take(readyEvents),
115 Ref.update(retries, (count) => count + 1).pipe(Effect.andThen(Effect.fail(disconnect))),
116 );
117
118 expect(result.payload.environment.serverVersion).toBe("0.0.31");
119 expect(yield* Ref.get(retries)).toBe(1);
120 }),
121 );
122 it.effect("observes a fast reconnect even when the caller awaits it later", () =>
123 Effect.gen(function* () {
124 const states = yield* Queue.unbounded<{ readonly phase: string }>();
125 const reconnected = yield* waitForNextEnvironmentReconnect(Stream.fromQueue(states)).pipe(
126 Effect.forkChild,
127 );
128 yield* Queue.offerAll(states, [
129 { phase: "connected" },
130 { phase: "backoff" },
131 { phase: "connected" },
132 ]);
133
134 yield* Fiber.join(reconnected);
135 }),
136 );
137 it.effect("arms the retry observer before a commit can disconnect", () =>
138 Effect.gen(function* () {
139 const allowSubscription = yield* Deferred.make<void>();
140 const subscriptionStarted = yield* Deferred.make<void>();
141 const states = yield* Queue.unbounded<{ readonly phase: string }>();
142 const commits = yield* Ref.make(0);
143 const disconnect = new RpcClientError.RpcClientError({
144 reason: new Socket.SocketCloseError({ code: 1006 }),
145 });
146 const stateChanges = Stream.unwrap(
147 Deferred.succeed(subscriptionStarted, undefined).pipe(
148 Effect.andThen(Deferred.await(allowSubscription)),
149 Effect.as(Stream.fromQueue(states)),
150 ),
151 );
152 const retry = yield* runDesktopCommitWithReconnectObserver(
153 stateChanges,
154 Ref.update(commits, (count) => count + 1).pipe(
155 Effect.andThen(Queue.offerAll(states, [{ phase: "backoff" }, { phase: "connected" }])),
156 Effect.andThen(Effect.fail(disconnect)),
157 ),
158 ).pipe(Effect.flip, Effect.forkChild);
159
160 yield* Deferred.await(subscriptionStarted);
161 expect(yield* Ref.get(commits)).toBe(0);
162 yield* Deferred.succeed(allowSubscription, undefined);
163 yield* Queue.offer(states, { phase: "connected" });
164
165 expect(yield* Fiber.join(retry)).toBe(disconnect);
166 expect(yield* Ref.get(commits)).toBe(1);
167 }),
168 );
169 it.effect("retries once per backoff entry instead of only the first", () =>
170 Effect.gen(function* () {
171 const retries = yield* Ref.make(0);
172 const states = [
173 { phase: "backoff" },
174 { phase: "connecting" },
175 { phase: "backoff" },
176 { phase: "backoff" },
177 ];
178
179 yield* nudgeReconnectDuringUpdateRestart({
180 stateChanges: Stream.fromIterable(states),
181 retryNow: Ref.update(retries, (count) => count + 1),
182 interval: Duration.zero,
183 });
184
185 // Three backoff entries, three nudges. The old one-shot behavior fired
186 // once and then let the supervisor's ladder stretch to 16-second gaps.
187 expect(yield* Ref.get(retries)).toBe(3);
188 }),
189 );
190
191 it.effect("paces nudges so a fast-failing connection cannot spin", () =>
192 Effect.gen(function* () {
193 const retries = yield* Ref.make(0);
194
195 const fiber = yield* Effect.forkChild(
196 nudgeReconnectDuringUpdateRestart({
197 stateChanges: Stream.fromIterable([{ phase: "backoff" }, { phase: "backoff" }]),
198 retryNow: Ref.update(retries, (count) => count + 1),
199 }),
200 { startImmediately: true },
201 );
202
203 // Each nudge waits out the interval first, so nothing fires immediately.
204 yield* TestClock.adjust(Duration.millis(999));
205 expect(yield* Ref.get(retries)).toBe(0);
206
207 yield* TestClock.adjust(Duration.millis(1));
208 expect(yield* Ref.get(retries)).toBe(1);
209
210 yield* TestClock.adjust(Duration.seconds(1));
211 expect(yield* Ref.get(retries)).toBe(2);
212
213 yield* Fiber.join(fiber);
214 }).pipe(Effect.provide(TestClock.layer())),
215 );
216
217 it.effect("retries rejected credentials only while the update restart is in progress", () =>
218 Effect.gen(function* () {
219 const retries = yield* Ref.make(0);
220
221 yield* nudgeReconnectDuringUpdateRestart({
222 stateChanges: Stream.fromIterable([
223 { phase: "blocked", lastFailure: { reason: "permission" } },
224 {
225 phase: "blocked",
226 lastFailure: {
227 reason: "authentication",
228 detail: "The environment credential is invalid.",
229 },
230 },
231 { phase: "blocked", lastFailure: { reason: "configuration" } },
232 ]),
233 retryNow: Ref.update(retries, (count) => count + 1),
234 interval: Duration.zero,
235 });
236
237 expect(yield* Ref.get(retries)).toBe(1);
238 }),
239 );
240});
241
242describe("server state projection", () => {
243 it("only treats a legacy transport interruption as an unacknowledged handoff", () => {
244 expect(isLegacyUpdateHandoffLoss(Cause.interrupt(1))).toBe(true);
245 expect(
246 isLegacyUpdateHandoffLoss(
247 Cause.fail(
248 new RpcClientError.RpcClientError({
249 reason: new Socket.SocketCloseError({ code: 1006 }),
250 }),
251 ),
252 ),
253 ).toBe(true);
254 expect(
255 isLegacyUpdateHandoffLoss(
256 Cause.fail(
257 new RpcClientError.RpcClientError({
258 reason: new Socket.SocketOpenError({
259 kind: "Unknown",
260 cause: new Error("connection refused"),
261 }),
262 }),
263 ),
264 ),
265 ).toBe(false);
266 expect(
267 isLegacyUpdateHandoffLoss(
268 Cause.fail(
269 new RpcClientError.RpcClientError({
270 reason: new RpcClientError.RpcClientDefect({
271 message: "incompatible protocol",
272 cause: new Error("invalid response"),
273 }),
274 }),
275 ),
276 ),
277 ).toBe(false);
278 expect(isLegacyUpdateHandoffLoss(Cause.fail(new Error("Install failed.")))).toBe(false);
279 });
280
281 it.effect("resumes after the progress stream disconnects following completion", () => {
282 const result = {
283 targetVersion: "0.0.31",
284 method: "respawn" as const,
285 };
286 const disconnect = new RpcClientError.RpcClientError({
287 reason: new Socket.SocketCloseError({ code: 1006 }),
288 });
289
290 return Effect.gen(function* () {
291 const resumed = yield* resolveServerUpdateProgressResult(
292 result.targetVersion,
293 Option.some(result),
294 Exit.fail(disconnect),
295 );
296 expect(resumed).toEqual(result);
297 });
298 });
299
300 it("projects streamed update milestones into the shared operation state", () => {
301 expect(
302 serverUpdateStateForProgressEvent("0.0.30", "0.0.31", {
303 type: "progress",
304 stage: "installing",
305 }),
306 ).toEqual({
307 status: "running",
308 stage: "installing",
309 fromVersion: "0.0.30",
310 targetVersion: "0.0.31",
311 });
312 expect(
313 serverUpdateStateForProgressEvent("0.0.30", "0.0.31", {
314 type: "complete",
315 result: { targetVersion: "0.0.31", method: "respawn" },
316 }),
317 ).toEqual({
318 status: "running",
319 stage: "resuming",
320 fromVersion: "0.0.30",
321 targetVersion: "0.0.31",
322 });
323 });
324
325 it("keeps active update state and hides stale failures after a version change", () => {
326 const running = {
327 status: "running" as const,
328 stage: "resuming" as const,
329 fromVersion: "0.0.30",
330 targetVersion: "0.0.31",
331 };
332 const failed = {
333 status: "failed" as const,
334 stage: "installing" as const,
335 fromVersion: "0.0.30",
336 targetVersion: "0.0.31",
337 message: "Install failed.",
338 };
339
340 expect(serverUpdateStateForServerVersion(running, "0.0.31")).toBe(running);
341 expect(serverUpdateStateForServerVersion(failed, "0.0.30")).toBe(failed);
342 expect(serverUpdateStateForServerVersion(failed, null)).toBe(failed);
343 expect(serverUpdateStateForServerVersion(failed, "0.0.31")).toEqual({ status: "idle" });
344 });
345
346 it.effect("correlates launcher outcomes and fails immediately after rollback", () =>
347 Effect.gen(function* () {
348 const result = {
349 targetVersion: "0.0.31",
350 method: "boot-service" as const,
351 updateId: "update-1",
352 };
353 const ready = (status: "committed" | "rolled-back") =>
354 ({
355 version: 1 as const,
356 sequence: 1,
357 type: "ready" as const,
358 payload: {
359 at: "2026-08-01T00:00:00.000Z",
360 environment: { serverVersion: status === "committed" ? "0.0.31" : "0.0.30" },
361 updateOutcome: {
362 id: "update-1",
363 fromVersion: "0.0.30",
364 targetVersion: "0.0.31",
365 status,
366 ...(status === "rolled-back" ? { reason: "prepared-timeout" } : {}),
367 },
368 },
369 }) as Parameters<typeof matchesServerUpdateReadyEvent>[1];
370
371 expect(matchesServerUpdateReadyEvent(result, ready("committed"))).toBe(true);
372 yield* validateServerUpdateReadyEvent(result, ready("committed"));
373 const rollback = yield* Effect.flip(
374 validateServerUpdateReadyEvent(result, ready("rolled-back")),
375 );
376 expect(rollback.message).toBe("prepared-timeout");
377 }),
378 );
379
380 it("requires tokenless desktop updates to reach the target version", () => {
381 const ready = (serverVersion: string) =>
382 ({
383 version: 1 as const,
384 sequence: 1,
385 type: "ready" as const,
386 payload: {
387 at: "2026-09-01T00:00:00.000Z",
388 environment: { serverVersion },
389 },
390 }) as Parameters<typeof matchesServerUpdateResumeEvent>[1];
391
392 expect(
393 matchesServerUpdateResumeEvent(
394 { targetVersion: "0.0.31", method: "desktop-app" },
395 ready("0.0.30"),
396 ),
397 ).toBe(false);
398 expect(
399 matchesServerUpdateResumeEvent(
400 {
401 targetVersion: "0.0.31",
402 method: "desktop-app",
403 desktopUpdateToken: "update-1",
404 },
405 ready("0.0.30"),
406 ),
407 ).toBe(true);
408 });
409
410 it("applies every config category to the projected snapshot", () => {
411 const snapshot = applyServerConfigProjection(Option.none(), {
412 version: 1,
413 type: "snapshot",
414 config: CONFIG,
415 });
416 const settings = { ...CONFIG.settings };
417 const projected = applyServerConfigProjection(snapshot, {
418 version: 1,
419 type: "settingsUpdated",
420 payload: { settings },
421 });
422
423 const result = Option.getOrThrow(projected);
424 expect(result.config.settings).toBe(settings);
425 expect(result.latestEvent.type).toBe("settingsUpdated");
426 });
427
428 it("carries published environment themes in and out of the projected snapshot", () => {
429 const snapshot = applyServerConfigProjection(Option.none(), {
430 version: 1,
431 type: "snapshot",
432 config: CONFIG,
433 });
434 const themes = [
435 {
436 id: "nightfall",
437 name: "Nightfall",
438 appearance: "dark",
439 canvas: "#1a1b26",
440 accent: "#7aa2f7",
441 },
442 ] as const;
443
444 const published = applyServerConfigProjection(snapshot, {
445 version: 1,
446 type: "environmentThemesUpdated",
447 payload: { themes },
448 });
449 expect(Option.getOrThrow(published).config.environmentThemes).toEqual(themes);
450
451 // A machine that stops publishing has to clear the palettes, not freeze
452 // clients on the last set it sent.
453 const unpublished = applyServerConfigProjection(published, {
454 version: 1,
455 type: "environmentThemesUpdated",
456 payload: { themes: [] },
457 });
458 expect(Option.getOrThrow(unpublished).config.environmentThemes).toBeUndefined();
459 });
460
461 // A snapshot never carries published themes, so taking it wholesale would
462 // clear them on every reconnect and repaint anyone wearing one.
463 it("keeps published themes across a reconnect snapshot", () => {
464 const themes = [
465 {
466 id: "nightfall",
467 name: "Nightfall",
468 appearance: "dark",
469 canvas: "#1a1b26",
470 accent: "#7aa2f7",
471 },
472 ] as const;
473
474 const withThemes = applyServerConfigProjection(
475 applyServerConfigProjection(Option.none(), { version: 1, type: "snapshot", config: CONFIG }),
476 { version: 1, type: "environmentThemesUpdated", payload: { themes } },
477 );
478 expect(Option.getOrThrow(withThemes).config.environmentThemes).toEqual(themes);
479
480 const afterReconnect = applyServerConfigProjection(withThemes, {
481 version: 1,
482 type: "snapshot",
483 config: CONFIG,
484 });
485 expect(Option.getOrThrow(afterReconnect).config.environmentThemes).toEqual(themes);
486
487 // A server that predates the feature never sends another theme event, so
488 // carrying the set forward would leave a palette nothing can update.
489 const downgraded = applyServerConfigProjection(withThemes, {
490 version: 1,
491 type: "snapshot",
492 config: {
493 ...CONFIG,
494 environment: { capabilities: {} },
495 } as unknown as ServerConfig,
496 });
497 expect(Option.getOrThrow(downgraded).config.environmentThemes).toBeUndefined();
498 });
499
500 it("keeps a current welcome on ready and rejects a buffered welcome from the old session", () => {
501 const firstSession = session({} as WsRpcProtocolClient);
502 const secondSession = session({} as WsRpcProtocolClient);
503 const welcome = {
504 environment: {} as ServerLifecycleWelcomePayload["environment"],
505 cwd: "/repo",
506 projectName: "repo",
507 } as ServerLifecycleWelcomePayload;
508 const initial = {
509 currentSession: firstSession,
510 welcomeSession: firstSession,
511 welcome: null,
512 };
513 const afterWelcome = applyServerWelcomeEvent(initial, firstSession, {
514 type: "welcome",
515 payload: welcome,
516 });
517 const afterReady = applyServerWelcomeEvent(afterWelcome, firstSession, {
518 type: "ready",
519 payload: {},
520 });
521 const afterSwitch = { ...afterReady, currentSession: secondSession };
522 const afterBufferedOldWelcome = applyServerWelcomeEvent(afterSwitch, firstSession, {
523 type: "welcome",
524 payload: { ...welcome, cwd: "/stale" },
525 });
526
527 expect(afterReady).toBe(afterWelcome);
528 expect(resolveServerWelcomeState(afterReady)).toBe(welcome);
529 expect(afterBufferedOldWelcome).toBe(afterSwitch);
530 expect(resolveServerWelcomeState(afterBufferedOldWelcome)).toBeNull();
531 });
532
533 it.effect("checks the authoritative session before accepting a buffered welcome", () =>
534 Effect.gen(function* () {
535 const firstEvents = yield* Queue.unbounded<{
536 readonly type: "welcome" | "ready";
537 readonly payload: unknown;
538 }>();
539 const firstSubscribed = yield* Deferred.make<void>();
540 const firstClient = {
541 [WS_METHODS.subscribeServerLifecycle]: () =>
542 Stream.fromEffect(Deferred.succeed(firstSubscribed, undefined)).pipe(
543 Stream.drain,
544 Stream.concat(Stream.fromQueue(firstEvents)),
545 ),
546 } as unknown as WsRpcProtocolClient;
547 const firstSession = session(firstClient);
548 const secondSession = session({} as WsRpcProtocolClient);
549 const supervisorSession = yield* SubscriptionRef.make(Option.some(firstSession));
550 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
551 target: TARGET,
552 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
553 session: supervisorSession,
554 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
555 connect: Effect.void,
556 disconnect: Effect.void,
557 retryNow: Effect.void,
558 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
559 const staleWelcome = {
560 environment: {} as ServerLifecycleWelcomePayload["environment"],
561 cwd: "/stale",
562 projectName: "stale",
563 } as ServerLifecycleWelcomePayload;
564
565 yield* Effect.scoped(
566 Effect.gen(function* () {
567 const state = yield* makeEnvironmentServerWelcomeState().pipe(
568 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
569 );
570 yield* Deferred.await(firstSubscribed);
571
572 // Model the point after the ref changed but before either subscriber
573 // processed its publication.
574 supervisorSession.value = Option.some(secondSession);
575 const handled = yield* SubscriptionRef.changes(state).pipe(
576 Stream.filter(
577 (value) => value.currentSession === secondSession || value.welcome === staleWelcome,
578 ),
579 Stream.runHead,
580 Effect.map(Option.getOrThrow),
581 Effect.forkChild,
582 );
583 yield* Queue.offer(firstEvents, { type: "welcome", payload: staleWelcome });
584
585 const next = yield* Fiber.join(handled);
586 expect(next.currentSession).toBe(secondSession);
587 expect(resolveServerWelcomeState(next)).toBeNull();
588 }),
589 );
590 }),
591 );
592
593 it.effect("reads the authoritative session after waiting for the welcome state lock", () =>
594 Effect.gen(function* () {
595 const firstSubscribed = yield* Deferred.make<void>();
596 const firstClient = {
597 [WS_METHODS.subscribeServerLifecycle]: () =>
598 Stream.fromEffect(Deferred.succeed(firstSubscribed, undefined)).pipe(Stream.drain),
599 } as unknown as WsRpcProtocolClient;
600 const secondClient = {
601 [WS_METHODS.subscribeServerLifecycle]: () => Stream.never,
602 } as unknown as WsRpcProtocolClient;
603 const firstSession = session(firstClient);
604 const secondSession = session(secondClient);
605 const thirdSession = session({} as WsRpcProtocolClient);
606 const supervisorSession = yield* SubscriptionRef.make(Option.some(firstSession));
607 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
608 target: TARGET,
609 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
610 session: supervisorSession,
611 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
612 connect: Effect.void,
613 disconnect: Effect.void,
614 retryNow: Effect.void,
615 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
616
617 yield* Effect.scoped(
618 Effect.gen(function* () {
619 const state = yield* makeEnvironmentServerWelcomeState().pipe(
620 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
621 );
622 yield* Deferred.await(firstSubscribed);
623 const changed = yield* SubscriptionRef.changes(state).pipe(
624 Stream.filter((value) => value.currentSession !== firstSession),
625 Stream.runHead,
626 Effect.map(Option.getOrThrow),
627 Effect.forkChild,
628 );
629
630 yield* state.semaphore.withPermit(
631 Effect.gen(function* () {
632 yield* SubscriptionRef.set(supervisorSession, Option.some(secondSession));
633 yield* Effect.yieldNow;
634 yield* Effect.yieldNow;
635 yield* Effect.yieldNow;
636 supervisorSession.value = Option.some(thirdSession);
637 }),
638 );
639
640 expect((yield* Fiber.join(changed)).currentSession).toBe(thirdSession);
641 }),
642 );
643 }),
644 );
645
646 it.effect("clears a welcome until the reconnected session sends its own", () =>
647 Effect.gen(function* () {
648 const firstEvents = yield* Queue.unbounded<{
649 readonly type: "welcome" | "ready";
650 readonly payload: unknown;
651 }>();
652 const secondEvents = yield* Queue.unbounded<{
653 readonly type: "welcome" | "ready";
654 readonly payload: unknown;
655 }>();
656 const firstClient = {
657 [WS_METHODS.subscribeServerLifecycle]: () => Stream.fromQueue(firstEvents),
658 } as unknown as WsRpcProtocolClient;
659 const secondClient = {
660 [WS_METHODS.subscribeServerLifecycle]: () => Stream.fromQueue(secondEvents),
661 } as unknown as WsRpcProtocolClient;
662 const firstSession = session(firstClient);
663 const secondSession = session(secondClient);
664 const supervisorSession = yield* SubscriptionRef.make(Option.some(firstSession));
665 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
666 target: TARGET,
667 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
668 session: supervisorSession,
669 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
670 connect: Effect.void,
671 disconnect: Effect.void,
672 retryNow: Effect.void,
673 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
674 const firstWelcome = {
675 environment: {} as ServerLifecycleWelcomePayload["environment"],
676 cwd: "/first",
677 projectName: "first",
678 } as ServerLifecycleWelcomePayload;
679 const secondWelcome = {
680 environment: {} as ServerLifecycleWelcomePayload["environment"],
681 cwd: "/second",
682 projectName: "second",
683 } as ServerLifecycleWelcomePayload;
684
685 yield* Effect.scoped(
686 Effect.gen(function* () {
687 const state = yield* makeEnvironmentServerWelcomeState().pipe(
688 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
689 );
690 const nextResolved = (
691 predicate: (value: ServerLifecycleWelcomePayload | null) => boolean,
692 ) =>
693 SubscriptionRef.changes(state).pipe(
694 Stream.map(resolveServerWelcomeState),
695 Stream.filter(predicate),
696 Stream.runHead,
697 Effect.map(Option.getOrThrow),
698 );
699
700 const first = yield* nextResolved((value) => value === firstWelcome).pipe(
701 Effect.forkChild,
702 );
703 yield* Queue.offer(firstEvents, { type: "welcome", payload: firstWelcome });
704 expect(yield* Fiber.join(first)).toBe(firstWelcome);
705
706 const cleared = yield* nextResolved((value) => value === null).pipe(Effect.forkChild);
707 yield* SubscriptionRef.set(supervisorSession, Option.some(secondSession));
708 expect(yield* Fiber.join(cleared)).toBeNull();
709 expect(resolveServerWelcomeState(yield* SubscriptionRef.get(state))).toBeNull();
710
711 const second = yield* nextResolved((value) => value === secondWelcome).pipe(
712 Effect.forkChild,
713 );
714 yield* Queue.offer(secondEvents, { type: "welcome", payload: secondWelcome });
715 expect(yield* Fiber.join(second)).toBe(secondWelcome);
716 }),
717 );
718 }),
719 );
720
721 it("prefers an active session config over cache until a live event arrives", () => {
722 const config = (source: string, serverVersion: string) =>
723 ({
724 ...CONFIG,
725 environment: { serverVersion },
726 settings: { source },
727 }) as unknown as ServerConfig;
728 const cached = config("cache", "0.0.29");
729 const staleLive = config("stale-live", "0.0.29");
730 const initial = config("session", "0.0.30");
731 const live = config("live", "0.0.30");
732
733 expect(
734 resolveServerConfigValue(
735 {
736 config: cached,
737 latestEvent: snapshotEvent(cached),
738 source: "cache",
739 },
740 initial,
741 ),
742 ).toBe(initial);
743 expect(
744 resolveServerConfigValue(
745 {
746 config: staleLive,
747 latestEvent: snapshotEvent(staleLive),
748 source: "live",
749 },
750 initial,
751 ),
752 ).toBe(initial);
753 expect(
754 resolveServerConfigValue(
755 {
756 config: live,
757 latestEvent: snapshotEvent(live),
758 source: "live",
759 },
760 initial,
761 ),
762 ).toBe(live);
763 });
764
765 it.effect("throttles server configuration writes and flushes the latest on teardown", () =>
766 Effect.gen(function* () {
767 const events = yield* Queue.unbounded<ServerConfigStreamEvent>();
768 const client = {
769 [WS_METHODS.subscribeServerConfig]: () => Stream.fromQueue(events),
770 } as unknown as WsRpcProtocolClient;
771 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
772 target: TARGET,
773 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
774 session: yield* SubscriptionRef.make(Option.some(session(client))),
775 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
776 connect: Effect.void,
777 disconnect: Effect.void,
778 retryNow: Effect.void,
779 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
780 const savedConfigs = yield* Queue.unbounded<ServerConfig>();
781 const cache = Persistence.EnvironmentCacheStore.of({
782 loadShell: () => Effect.succeed(Option.none()),
783 saveShell: () => Effect.void,
784 loadThread: () => Effect.succeed(Option.none()),
785 saveThread: () => Effect.void,
786 removeThread: () => Effect.void,
787 loadServerConfig: () => Effect.succeed(Option.some(CONFIG)),
788 saveServerConfig: (_environmentId, config) => Queue.offer(savedConfigs, config),
789 loadVcsRefs: () => Effect.succeed(Option.none()),
790 saveVcsRefs: () => Effect.void,
791 removeVcsRefs: () => Effect.void,
792 clearVcsRefs: () => Effect.void,
793 clear: () => Effect.void,
794 });
795
796 const scope = yield* Scope.make();
797 const state = yield* makeEnvironmentServerConfigState({}).pipe(
798 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
799 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
800 Effect.provideService(Scope.Scope, scope),
801 );
802 const sendConfig = Effect.fn(function* (cwd: string) {
803 yield* Queue.offer(events, { version: 1, type: "snapshot", config: { ...CONFIG, cwd } });
804 yield* SubscriptionRef.changes(state).pipe(
805 Stream.filter((value) => Option.isSome(value) && value.value.config.cwd === cwd),
806 Stream.runHead,
807 );
808 });
809 yield* sendConfig("/first");
810 yield* TestClock.adjust("500 millis");
811 expect((yield* Queue.take(savedConfigs)).cwd).toBe("/first");
812 for (let index = 1; index <= 9; index++) {
813 yield* sendConfig(`/repo-${index}`);
814 yield* TestClock.adjust("1 second");
815 }
816 expect(yield* Queue.poll(savedConfigs)).toEqual(Option.none());
817 yield* TestClock.adjust("1 second");
818 expect((yield* Queue.take(savedConfigs)).cwd).toBe("/repo-9");
819 yield* sendConfig("/final");
820 yield* Scope.close(scope, Exit.void);
821 expect((yield* Queue.take(savedConfigs)).cwd).toBe("/final");
822 expect(yield* Queue.poll(savedConfigs)).toEqual(Option.none());
823 }),
824 );
825
826 it.effect("starts from cached configuration and persists the live projection", () =>
827 Effect.gen(function* () {
828 const events = yield* Queue.unbounded<ServerConfigStreamEvent>();
829 const client = {
830 [WS_METHODS.subscribeServerConfig]: () => Stream.fromQueue(events),
831 } as unknown as WsRpcProtocolClient;
832 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
833 target: TARGET,
834 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
835 session: yield* SubscriptionRef.make(Option.some(session(client))),
836 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
837 connect: Effect.void,
838 disconnect: Effect.void,
839 retryNow: Effect.void,
840 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
841 const savedConfigs = yield* Queue.unbounded<ServerConfig>();
842 const cache = Persistence.EnvironmentCacheStore.of({
843 loadShell: () => Effect.succeedNone,
844 saveShell: () => Effect.void,
845 loadThread: () => Effect.succeedNone,
846 saveThread: () => Effect.void,
847 removeThread: () => Effect.void,
848 loadServerConfig: () => Effect.succeedSome(CONFIG),
849 saveServerConfig: (_environmentId, config) => Queue.offer(savedConfigs, config),
850 loadVcsRefs: () => Effect.succeedNone,
851 saveVcsRefs: () => Effect.void,
852 removeVcsRefs: () => Effect.void,
853 clearVcsRefs: () => Effect.void,
854 clear: () => Effect.void,
855 });
856
857 yield* Effect.scoped(
858 Effect.gen(function* () {
859 const state = yield* makeEnvironmentServerConfigState({}).pipe(
860 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
861 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
862 );
863 expect(Option.getOrThrow(yield* SubscriptionRef.get(state)).config).toBe(CONFIG);
864
865 const providers: ServerConfig["providers"] = [];
866 yield* Queue.offer(events, {
867 version: 1,
868 type: "providerStatuses",
869 payload: { providers },
870 });
871 const projected = yield* SubscriptionRef.changes(state).pipe(
872 Stream.filter((value) =>
873 Option.match(value, {
874 onNone: () => false,
875 onSome: (projection) => projection.latestEvent.type === "providerStatuses",
876 }),
877 ),
878 Stream.runHead,
879 );
880 expect(Option.getOrThrow(Option.getOrThrow(projected)).config.providers).toBe(providers);
881 }),
882 );
883
884 expect((yield* Queue.take(savedConfigs)).providers).toEqual([]);
885 }),
886 );
887
888 it.effect("does not rewrite cached configuration when no live update arrives", () =>
889 Effect.gen(function* () {
890 const client = {
891 [WS_METHODS.subscribeServerConfig]: () => Stream.empty,
892 } as unknown as WsRpcProtocolClient;
893 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
894 target: TARGET,
895 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
896 session: yield* SubscriptionRef.make(Option.some(session(client))),
897 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
898 connect: Effect.void,
899 disconnect: Effect.void,
900 retryNow: Effect.void,
901 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
902 const savedConfigs = yield* Queue.unbounded<ServerConfig>();
903 const cache = Persistence.EnvironmentCacheStore.of({
904 loadShell: () => Effect.succeedNone,
905 saveShell: () => Effect.void,
906 loadThread: () => Effect.succeedNone,
907 saveThread: () => Effect.void,
908 removeThread: () => Effect.void,
909 loadServerConfig: () => Effect.succeedSome(CONFIG),
910 saveServerConfig: (_environmentId, config) => Queue.offer(savedConfigs, config),
911 loadVcsRefs: () => Effect.succeedNone,
912 saveVcsRefs: () => Effect.void,
913 removeVcsRefs: () => Effect.void,
914 clearVcsRefs: () => Effect.void,
915 clear: () => Effect.void,
916 });
917
918 yield* Effect.scoped(
919 makeEnvironmentServerConfigState({}).pipe(
920 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
921 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
922 ),
923 );
924
925 expect(yield* Queue.poll(savedConfigs)).toEqual(Option.none());
926 }),
927 );
928});