1import {
2 EnvironmentId,
3 ORCHESTRATION_V2_WS_METHODS,
4 type OrchestrationV2ShellSnapshot,
5 type OrchestrationV2ShellStreamItem,
6 ThreadId,
7 type ThreadPullRequestLink,
8} from "@t3tools/contracts";
9import { describe, expect, it } from "@effect/vitest";
10import * as Deferred from "effect/Deferred";
11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";
13import * as Fiber from "effect/Fiber";
14import * as Option from "effect/Option";
15import * as Queue from "effect/Queue";
16import * as Ref from "effect/Ref";
17import * as Scope from "effect/Scope";
18import * as Stream from "effect/Stream";
19import * as SubscriptionRef from "effect/SubscriptionRef";
20import * as TestClock from "effect/testing/TestClock";
21
22import {
23 AVAILABLE_CONNECTION_STATE,
24 PrimaryConnectionTarget,
25 type PreparedConnection,
26} from "../connection/model.ts";
27import * as EnvironmentSupervisor from "../connection/supervisor.ts";
28import * as ConnectionWakeups from "../connection/wakeups.ts";
29import * as Persistence from "../platform/persistence.ts";
30import * as RpcSession from "../rpc/session.ts";
31import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
32import { makeEnvironmentShellState } from "./shell.ts";
33import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
34import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
35
36const TARGET = new PrimaryConnectionTarget({
37 environmentId: EnvironmentId.make("environment-1"),
38 label: "Test environment",
39 httpBaseUrl: "https://environment.example.test",
40 wsBaseUrl: "wss://environment.example.test",
41});
42
43const PREPARED: PreparedConnection = {
44 environmentId: TARGET.environmentId,
45 label: TARGET.label,
46 httpBaseUrl: TARGET.httpBaseUrl,
47 socketUrl: TARGET.wsBaseUrl,
48 httpAuthorization: null,
49 target: TARGET,
50};
51
52const LIVE_SHELL_SNAPSHOT: OrchestrationV2ShellSnapshot = {
53 ...v2ShellSnapshot,
54 snapshotSequence: 1,
55};
56
57function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
58 return {
59 client,
60 initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never),
61 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
62 ready: Effect.void,
63 probe: Effect.void,
64 closed: Effect.never,
65 };
66}
67
68describe("environment shell synchronization", () => {
69 it.effect("publishes live state before persistence and preserves it when ready", () =>
70 Effect.gen(function* () {
71 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
72 const client = {
73 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
74 } as unknown as WsRpcProtocolClient;
75 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
76 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
77 Option.some(session(client)),
78 );
79 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
80 target: TARGET,
81 state: supervisorState,
82 session: activeSession,
83 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
84 connect: Effect.void,
85 disconnect: Effect.void,
86 retryNow: Effect.void,
87 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
88 const cache = Persistence.EnvironmentCacheStore.of({
89 loadShell: () => Effect.succeed(Option.none()),
90 saveShell: () => Effect.void,
91 loadThread: () => Effect.succeed(Option.none()),
92 saveThread: () => Effect.void,
93 removeThread: () => Effect.void,
94 loadServerConfig: () => Effect.succeedNone,
95 saveServerConfig: () => Effect.void,
96 loadVcsRefs: () => Effect.succeedNone,
97 saveVcsRefs: () => Effect.void,
98 removeVcsRefs: () => Effect.void,
99 clearVcsRefs: () => Effect.void,
100 clear: () => Effect.void,
101 });
102 // Cold cache with no HTTP snapshot available → falls back to the
103 // socket-embedded snapshot.
104 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
105 load: () => Effect.succeedNone,
106 });
107 const shellState = yield* makeEnvironmentShellState().pipe(
108 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
109 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
110 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
111 );
112
113 yield* SubscriptionRef.set(supervisorState, {
114 desired: true,
115 network: "online",
116 phase: "connecting",
117 stage: "synchronizing",
118 attempt: 1,
119 generation: 0,
120 lastFailure: null,
121 retryAt: null,
122 });
123 yield* Queue.offer(events, {
124 kind: "snapshot",
125 snapshot: LIVE_SHELL_SNAPSHOT,
126 });
127 const synchronizing = yield* SubscriptionRef.changes(shellState).pipe(
128 Stream.filter((state) => state.status === "synchronizing" && Option.isSome(state.snapshot)),
129 Stream.runHead,
130 );
131 expect(Option.getOrThrow(Option.getOrThrow(synchronizing).snapshot)).toEqual(
132 LIVE_SHELL_SNAPSHOT,
133 );
134
135 yield* Queue.offer(events, { kind: "synchronized" });
136 yield* SubscriptionRef.changes(shellState).pipe(
137 Stream.filter((state) => state.status === "live"),
138 Stream.runHead,
139 );
140
141 yield* SubscriptionRef.set(supervisorState, {
142 desired: true,
143 network: "online",
144 phase: "connected",
145 stage: null,
146 attempt: 1,
147 generation: 1,
148 lastFailure: null,
149 retryAt: null,
150 });
151 for (let index = 0; index < 10; index += 1) {
152 yield* Effect.yieldNow;
153 }
154
155 const state = yield* SubscriptionRef.get(shellState);
156 expect(state.status).toBe("live");
157 expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT);
158 }),
159 );
160
161 it.live.each([
162 { bufferSize: Infinity, expectedSequences: [51] },
163 // RpcClient defaults to a 16-event buffer, which splits larger server chunks.
164 { bufferSize: 16, expectedSequences: [17, 33, 49, 51] },
165 ])("batches live events with a $bufferSize event buffer", ({ bufferSize, expectedSequences }) =>
166 Effect.gen(function* () {
167 const events = yield* Queue.bounded<OrchestrationV2ShellStreamItem>(bufferSize);
168 const batchSnapshot = { ...LIVE_SHELL_SNAPSHOT, threads: [] };
169 const client = {
170 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
171 } as unknown as WsRpcProtocolClient;
172 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
173 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
174 Option.some(session(client)),
175 );
176 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
177 target: TARGET,
178 state: supervisorState,
179 session: activeSession,
180 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
181 connect: Effect.void,
182 disconnect: Effect.void,
183 retryNow: Effect.void,
184 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
185 const cache = Persistence.EnvironmentCacheStore.of({
186 loadShell: () => Effect.succeed(Option.none()),
187 saveShell: () => Effect.void,
188 loadThread: () => Effect.succeed(Option.none()),
189 saveThread: () => Effect.void,
190 removeThread: () => Effect.void,
191 loadServerConfig: () => Effect.succeed(Option.none()),
192 saveServerConfig: () => Effect.void,
193 loadVcsRefs: () => Effect.succeed(Option.none()),
194 saveVcsRefs: () => Effect.void,
195 removeVcsRefs: () => Effect.void,
196 clearVcsRefs: () => Effect.void,
197 clear: () => Effect.void,
198 });
199 const shellState = yield* makeEnvironmentShellState().pipe(
200 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
201 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
202 Effect.provideService(
203 ShellSnapshotLoader.ShellSnapshotLoader,
204 ShellSnapshotLoader.ShellSnapshotLoader.of({ load: () => Effect.succeed(Option.none()) }),
205 ),
206 );
207 yield* SubscriptionRef.set(supervisorState, {
208 desired: true,
209 network: "online",
210 phase: "connected",
211 stage: null,
212 attempt: 1,
213 generation: 1,
214 lastFailure: null,
215 retryAt: null,
216 });
217 yield* Queue.offer(events, { kind: "snapshot", snapshot: batchSnapshot });
218 yield* Queue.offer(events, { kind: "synchronized" });
219 yield* SubscriptionRef.changes(shellState).pipe(
220 Stream.filter((state) => state.status === "live"),
221 Stream.runHead,
222 );
223
224 // Observe before publishing so no batch can arrive before the subscription.
225 const observed = yield* SubscriptionRef.changes(shellState).pipe(
226 Stream.drop(1),
227 Stream.takeUntil(
228 (state) => Option.isSome(state.snapshot) && state.snapshot.value.threads.length === 50,
229 ),
230 Stream.runCollect,
231 Effect.forkScoped({ startImmediately: true }),
232 );
233 yield* Queue.offerAll(
234 events,
235 Array.from({ length: 50 }, (_, index) => ({
236 kind: "thread.updated" as const,
237 sequence: 2 + index,
238 location: "active" as const,
239 thread: { ...v2ShellSnapshot.threads[0]!, id: `thread-${index}` } as never,
240 })),
241 );
242 const states = yield* Fiber.join(observed);
243 const snapshots = states.map((state) => Option.getOrThrow(state.snapshot));
244 expect(snapshots.map((snapshot) => snapshot.snapshotSequence)).toEqual(expectedSequences);
245 expect(snapshots.at(-1)!.threads.map((thread) => thread.id)).toEqual(
246 Array.from({ length: 50 }, (_, index) => `thread-${index}`),
247 );
248 }).pipe(Effect.scoped),
249 );
250
251 it.effect("refreshes a warm shell cache from HTTP before resuming", () =>
252 Effect.gen(function* () {
253 const cachedSnapshot: OrchestrationV2ShellSnapshot = {
254 ...v2ShellSnapshot,
255 snapshotSequence: 5,
256 };
257 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
258 const capturedAfterSequence = yield* SubscriptionRef.make<number | undefined>(undefined);
259 const capturedCompletionMarker = yield* Ref.make(false);
260 const loaderCalls = yield* SubscriptionRef.make(0);
261 const httpSnapshot: OrchestrationV2ShellSnapshot = {
262 ...v2ShellSnapshot,
263 snapshotSequence: 9,
264 };
265 const client = {
266 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: {
267 readonly afterSequence?: number;
268 readonly requestCompletionMarker?: true;
269 }) =>
270 Stream.unwrap(
271 Effect.all([
272 Ref.set(capturedCompletionMarker, input.requestCompletionMarker === true),
273 SubscriptionRef.set(capturedAfterSequence, input.afterSequence),
274 ]).pipe(Effect.as(Stream.fromQueue(events))),
275 ),
276 } as unknown as WsRpcProtocolClient;
277 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
278 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
279 Option.some(session(client)),
280 );
281 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
282 target: TARGET,
283 state: supervisorState,
284 session: activeSession,
285 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
286 connect: Effect.void,
287 disconnect: Effect.void,
288 retryNow: Effect.void,
289 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
290 const cache = Persistence.EnvironmentCacheStore.of({
291 loadShell: () => Effect.succeedSome(cachedSnapshot),
292 saveShell: () => Effect.void,
293 loadThread: () => Effect.succeedNone,
294 saveThread: () => Effect.void,
295 removeThread: () => Effect.void,
296 loadServerConfig: () => Effect.succeedNone,
297 saveServerConfig: () => Effect.void,
298 loadVcsRefs: () => Effect.succeedNone,
299 saveVcsRefs: () => Effect.void,
300 removeVcsRefs: () => Effect.void,
301 clearVcsRefs: () => Effect.void,
302 clear: () => Effect.void,
303 });
304 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
305 load: () =>
306 SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(
307 Effect.as(Option.some(httpSnapshot)),
308 ),
309 });
310 const shellState = yield* makeEnvironmentShellState().pipe(
311 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
312 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
313 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
314 );
315
316 // Wait until the subscription is established from the warm cache.
317 yield* SubscriptionRef.changes(capturedAfterSequence).pipe(
318 Stream.filter((value) => value !== undefined),
319 Stream.runHead,
320 );
321
322 expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(9);
323 expect(yield* Ref.get(capturedCompletionMarker)).toBe(true);
324 expect(yield* SubscriptionRef.get(loaderCalls)).toBe(1);
325 const synchronizing = yield* SubscriptionRef.get(shellState);
326 expect(synchronizing.status).toBe("synchronizing");
327 expect(Option.getOrThrow(synchronizing.snapshot)).toEqual(httpSnapshot);
328
329 yield* Queue.offer(events, { kind: "synchronized" });
330 yield* SubscriptionRef.changes(shellState).pipe(
331 Stream.filter((value) => value.status === "live"),
332 Stream.runHead,
333 );
334 }),
335 );
336
337 it.effect("resubscribes from the in-memory shell cursor when the app becomes active", () =>
338 Effect.gen(function* () {
339 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
340 const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();
341 const loaderCalls = yield* Ref.make(0);
342 const capturedAfterSequences = yield* Ref.make<ReadonlyArray<number | undefined>>([]);
343 const client = {
344 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: {
345 readonly afterSequence?: number;
346 }) =>
347 Stream.unwrap(
348 Ref.update(capturedAfterSequences, (captured) => [
349 ...captured,
350 input.afterSequence,
351 ]).pipe(Effect.as(Stream.fromQueue(events))),
352 ),
353 } as unknown as WsRpcProtocolClient;
354 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
355 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
356 Option.some(session(client)),
357 );
358 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
359 target: TARGET,
360 state: supervisorState,
361 session: activeSession,
362 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
363 connect: Effect.void,
364 disconnect: Effect.void,
365 retryNow: Effect.void,
366 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
367 const cache = Persistence.EnvironmentCacheStore.of({
368 loadShell: () => Effect.succeedSome(LIVE_SHELL_SNAPSHOT),
369 saveShell: () => Effect.void,
370 loadThread: () => Effect.succeedNone,
371 saveThread: () => Effect.void,
372 removeThread: () => Effect.void,
373 loadServerConfig: () => Effect.succeedNone,
374 saveServerConfig: () => Effect.void,
375 loadVcsRefs: () => Effect.succeedNone,
376 saveVcsRefs: () => Effect.void,
377 removeVcsRefs: () => Effect.void,
378 clearVcsRefs: () => Effect.void,
379 clear: () => Effect.void,
380 });
381 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
382 load: () =>
383 Ref.updateAndGet(loaderCalls, (count) => count + 1).pipe(
384 Effect.map((count) =>
385 Option.some({ ...LIVE_SHELL_SNAPSHOT, snapshotSequence: count * 10 }),
386 ),
387 ),
388 });
389 const shellState = yield* makeEnvironmentShellState().pipe(
390 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
391 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
392 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
393 Effect.provideService(
394 ConnectionWakeups.ConnectionWakeups,
395 ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }),
396 ),
397 );
398
399 // A new session starts from an authoritative HTTP snapshot.
400 for (let attempt = 0; attempt < 100; attempt += 1) {
401 if ((yield* Ref.get(capturedAfterSequences)).length >= 1) break;
402 yield* Effect.yieldNow;
403 }
404 expect(yield* Ref.get(capturedAfterSequences)).toEqual([10]);
405 yield* Queue.offer(events, { kind: "synchronized" });
406 yield* SubscriptionRef.changes(shellState).pipe(
407 Stream.filter((value) => value.status === "live"),
408 Stream.runHead,
409 );
410
411 // A newer snapshot arrives on the stream and advances the cursor.
412 yield* Queue.offer(events, {
413 kind: "snapshot",
414 snapshot: { ...LIVE_SHELL_SNAPSHOT, snapshotSequence: 40 },
415 });
416 yield* SubscriptionRef.changes(shellState).pipe(
417 Stream.filter(
418 (value) => Option.isSome(value.snapshot) && value.snapshot.value.snapshotSequence === 40,
419 ),
420 Stream.runHead,
421 );
422
423 yield* Queue.offer(wakeups, "application-active");
424 for (let attempt = 0; attempt < 100; attempt += 1) {
425 if ((yield* Ref.get(capturedAfterSequences)).length >= 2) break;
426 yield* Effect.yieldNow;
427 }
428 expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40]);
429 yield* Queue.offer(events, { kind: "synchronized" });
430
431 yield* Queue.offer(wakeups, "application-active-probe");
432 for (let attempt = 0; attempt < 100; attempt += 1) {
433 if ((yield* Ref.get(capturedAfterSequences)).length >= 3) break;
434 yield* Effect.yieldNow;
435 }
436 expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40, 40]);
437
438 yield* Queue.offer(wakeups, "application-active-reconnect");
439 for (let attempt = 0; attempt < 10; attempt += 1) {
440 yield* Effect.yieldNow;
441 }
442 expect((yield* Ref.get(capturedAfterSequences)).length).toBe(3);
443 expect(yield* Ref.get(loaderCalls)).toBe(1);
444
445 // Replacing the session performs another authoritative refresh.
446 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
447 for (let attempt = 0; attempt < 100; attempt += 1) {
448 if ((yield* Ref.get(capturedAfterSequences)).length >= 4) break;
449 yield* Effect.yieldNow;
450 }
451 expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40, 40, 20]);
452 expect(yield* Ref.get(loaderCalls)).toBe(2);
453 }),
454 );
455
456 it.effect("throttles shell writes and flushes the latest snapshot on disconnect", () =>
457 Effect.gen(function* () {
458 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
459 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
460 const client = {
461 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
462 } as unknown as WsRpcProtocolClient;
463 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
464 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
465 Option.some(session(client)),
466 );
467 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
468 target: TARGET,
469 state: supervisorState,
470 session: activeSession,
471 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
472 connect: Effect.void,
473 disconnect: Effect.void,
474 retryNow: Effect.void,
475 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
476 const cache = Persistence.EnvironmentCacheStore.of({
477 loadShell: () => Effect.succeed(Option.none()),
478 saveShell: (_environmentId, snapshot) =>
479 Ref.update(saved, (values) => [...values, snapshot]),
480 loadThread: () => Effect.succeed(Option.none()),
481 saveThread: () => Effect.void,
482 removeThread: () => Effect.void,
483 loadServerConfig: () => Effect.succeed(Option.none()),
484 saveServerConfig: () => Effect.void,
485 loadVcsRefs: () => Effect.succeed(Option.none()),
486 saveVcsRefs: () => Effect.void,
487 removeVcsRefs: () => Effect.void,
488 clearVcsRefs: () => Effect.void,
489 clear: () => Effect.void,
490 });
491 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
492 load: () => Effect.succeed(Option.none()),
493 });
494
495 const shellState = yield* makeEnvironmentShellState().pipe(
496 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
497 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
498 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
499 );
500
501 const sendSnapshot = Effect.fn(function* (sequence: number) {
502 yield* Queue.offer(events, {
503 kind: "snapshot",
504 snapshot: { ...LIVE_SHELL_SNAPSHOT, snapshotSequence: sequence },
505 });
506 yield* SubscriptionRef.changes(shellState).pipe(
507 Stream.filter(
508 (state) =>
509 Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === sequence,
510 ),
511 Stream.runHead,
512 );
513 });
514 yield* sendSnapshot(1);
515 yield* TestClock.adjust("500 millis");
516 for (let sequence = 2; sequence <= 10; sequence++) {
517 yield* sendSnapshot(sequence);
518 yield* TestClock.adjust("1 second");
519 }
520 expect((yield* Ref.get(saved)).map((snapshot) => snapshot.snapshotSequence)).toEqual([1]);
521 yield* TestClock.adjust("1 second");
522 expect((yield* Ref.get(saved)).map((snapshot) => snapshot.snapshotSequence)).toEqual([1, 10]);
523 yield* sendSnapshot(11);
524 yield* SubscriptionRef.set(supervisorState, {
525 ...AVAILABLE_CONNECTION_STATE,
526 desired: false,
527 });
528 yield* SubscriptionRef.changes(shellState).pipe(
529 Stream.filter((state) => state.status === "cached"),
530 Stream.runHead,
531 );
532 // The disconnect flush runs independently from the connection-state update.
533 yield* TestClock.adjust("0 millis");
534 expect((yield* Ref.get(saved)).map((snapshot) => snapshot.snapshotSequence)).toEqual([
535 1, 10, 11,
536 ]);
537 }),
538 );
539
540 it.effect("flushes the latest live snapshot without blocking connection state updates", () =>
541 Effect.gen(function* () {
542 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
543 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
544 const saveStarted = yield* Deferred.make<void>();
545 const releaseSave = yield* Deferred.make<void>();
546 const client = {
547 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
548 } as unknown as WsRpcProtocolClient;
549 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
550 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
551 Option.some(session(client)),
552 );
553 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
554 target: TARGET,
555 state: supervisorState,
556 session: activeSession,
557 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
558 connect: Effect.void,
559 disconnect: Effect.void,
560 retryNow: Effect.void,
561 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
562 const cache = Persistence.EnvironmentCacheStore.of({
563 loadShell: () => Effect.succeed(Option.none()),
564 saveShell: (_environmentId, snapshot) =>
565 Deferred.succeed(saveStarted, undefined).pipe(
566 Effect.andThen(Deferred.await(releaseSave)),
567 Effect.andThen(Ref.update(saved, (values) => [...values, snapshot])),
568 ),
569 loadThread: () => Effect.succeed(Option.none()),
570 saveThread: () => Effect.void,
571 removeThread: () => Effect.void,
572 loadServerConfig: () => Effect.succeed(Option.none()),
573 saveServerConfig: () => Effect.void,
574 loadVcsRefs: () => Effect.succeed(Option.none()),
575 saveVcsRefs: () => Effect.void,
576 removeVcsRefs: () => Effect.void,
577 clearVcsRefs: () => Effect.void,
578 clear: () => Effect.void,
579 });
580 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
581 load: () => Effect.succeed(Option.none()),
582 });
583
584 const shellState = yield* makeEnvironmentShellState().pipe(
585 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
586 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
587 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
588 );
589
590 yield* SubscriptionRef.set(supervisorState, {
591 desired: true,
592 network: "online",
593 phase: "connecting",
594 stage: "synchronizing",
595 attempt: 1,
596 generation: 0,
597 lastFailure: null,
598 retryAt: null,
599 });
600 yield* Queue.offer(events, {
601 kind: "snapshot",
602 snapshot: LIVE_SHELL_SNAPSHOT,
603 });
604 yield* Queue.offer(events, { kind: "synchronized" });
605 yield* SubscriptionRef.changes(shellState).pipe(
606 Stream.filter((state) => state.status === "live"),
607 Stream.runHead,
608 );
609
610 yield* SubscriptionRef.set(supervisorState, {
611 desired: true,
612 network: "online",
613 phase: "connecting",
614 stage: "synchronizing",
615 attempt: 2,
616 generation: 1,
617 lastFailure: null,
618 retryAt: null,
619 });
620 yield* SubscriptionRef.set(supervisorState, {
621 desired: false,
622 network: "online",
623 phase: "available",
624 stage: null,
625 attempt: 2,
626 generation: 2,
627 lastFailure: null,
628 retryAt: null,
629 });
630 yield* Deferred.await(saveStarted);
631
632 yield* SubscriptionRef.set(supervisorState, {
633 desired: true,
634 network: "online",
635 phase: "connecting",
636 stage: "synchronizing",
637 attempt: 3,
638 generation: 3,
639 lastFailure: null,
640 retryAt: null,
641 });
642 for (let index = 0; index < 100; index += 1) {
643 if ((yield* SubscriptionRef.get(shellState)).status === "synchronizing") break;
644 yield* Effect.yieldNow;
645 }
646
647 expect((yield* SubscriptionRef.get(shellState)).status).toBe("synchronizing");
648 expect(yield* Ref.get(saved)).toEqual([]);
649
650 yield* Deferred.succeed(releaseSave, undefined);
651 for (let index = 0; index < 100; index += 1) {
652 if ((yield* Ref.get(saved)).length > 0) break;
653 yield* Effect.yieldNow;
654 }
655
656 expect(yield* Ref.get(saved)).toEqual([LIVE_SHELL_SNAPSHOT]);
657 }),
658 );
659
660 it.effect("re-persists same-sequence enrichment when an older in-flight save completes", () =>
661 Effect.gen(function* () {
662 const repositoryIdentity = {
663 canonicalKey: "github.com/example/repo",
664 locator: {
665 source: "git-remote" as const,
666 remoteName: "origin",
667 remoteUrl: "https://github.com/example/repo.git",
668 },
669 };
670 const unenrichedSnapshot: OrchestrationV2ShellSnapshot = {
671 ...LIVE_SHELL_SNAPSHOT,
672 snapshotSequence: 1,
673 projects: [{ ...v2Project, repositoryIdentity: null }],
674 };
675 const enrichedSnapshot: OrchestrationV2ShellSnapshot = {
676 ...LIVE_SHELL_SNAPSHOT,
677 snapshotSequence: 1,
678 projects: [{ ...v2Project, repositoryIdentity }],
679 };
680
681 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
682 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
683 const unenrichedSaveStarted = yield* Deferred.make<void>();
684 const releaseUnenrichedSave = yield* Deferred.make<void>();
685 const client = {
686 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
687 } as unknown as WsRpcProtocolClient;
688 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
689 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
690 Option.some(session(client)),
691 );
692 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
693 target: TARGET,
694 state: supervisorState,
695 session: activeSession,
696 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
697 connect: Effect.void,
698 disconnect: Effect.void,
699 retryNow: Effect.void,
700 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
701 const cache = Persistence.EnvironmentCacheStore.of({
702 loadShell: () => Effect.succeed(Option.none()),
703 saveShell: (_environmentId, snapshot) =>
704 Effect.gen(function* () {
705 if (snapshot.projects[0]?.repositoryIdentity == null) {
706 yield* Deferred.succeed(unenrichedSaveStarted, undefined);
707 yield* Deferred.await(releaseUnenrichedSave);
708 }
709 yield* Ref.update(saved, (values) => [...values, snapshot]);
710 }),
711 loadThread: () => Effect.succeed(Option.none()),
712 saveThread: () => Effect.void,
713 removeThread: () => Effect.void,
714 loadServerConfig: () => Effect.succeed(Option.none()),
715 saveServerConfig: () => Effect.void,
716 loadVcsRefs: () => Effect.succeed(Option.none()),
717 saveVcsRefs: () => Effect.void,
718 removeVcsRefs: () => Effect.void,
719 clearVcsRefs: () => Effect.void,
720 clear: () => Effect.void,
721 });
722 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
723 load: () => Effect.succeed(Option.none()),
724 });
725
726 const shellState = yield* makeEnvironmentShellState().pipe(
727 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
728 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
729 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
730 );
731
732 yield* SubscriptionRef.set(supervisorState, {
733 desired: true,
734 network: "online",
735 phase: "connecting",
736 stage: "synchronizing",
737 attempt: 1,
738 generation: 0,
739 lastFailure: null,
740 retryAt: null,
741 });
742 yield* Queue.offer(events, {
743 kind: "snapshot",
744 snapshot: unenrichedSnapshot,
745 });
746 yield* Queue.offer(events, { kind: "synchronized" });
747 yield* SubscriptionRef.changes(shellState).pipe(
748 Stream.filter((state) => state.status === "live"),
749 Stream.runHead,
750 );
751
752 // Disconnect flush starts an immediate persist of the unenriched snapshot
753 // (debounced path needs TestClock advancement; flush does not).
754 yield* SubscriptionRef.set(supervisorState, {
755 desired: false,
756 network: "online",
757 phase: "available",
758 stage: null,
759 attempt: 1,
760 generation: 1,
761 lastFailure: null,
762 retryAt: null,
763 });
764 yield* Deferred.await(unenrichedSaveStarted);
765
766 // Same sequence, different content. Session stays open so the stream can
767 // still apply enrichment while the older save is in flight.
768 yield* Queue.offer(events, {
769 kind: "snapshot",
770 snapshot: enrichedSnapshot,
771 });
772 yield* SubscriptionRef.changes(shellState).pipe(
773 Stream.filter(
774 (state) =>
775 Option.isSome(state.snapshot) &&
776 state.snapshot.value.projects[0]?.repositoryIdentity != null,
777 ),
778 Stream.runHead,
779 );
780
781 // Completing the older same-sequence save must re-persist the latest object.
782 yield* Deferred.succeed(releaseUnenrichedSave, undefined);
783 for (let index = 0; index < 100; index += 1) {
784 const values = yield* Ref.get(saved);
785 if (values.length > 0 && values.at(-1)?.projects[0]?.repositoryIdentity != null) {
786 break;
787 }
788 yield* Effect.yieldNow;
789 }
790
791 const finalSaved = (yield* Ref.get(saved)).at(-1);
792 expect(finalSaved?.snapshotSequence).toBe(1);
793 expect(finalSaved?.projects[0]?.repositoryIdentity).toEqual(repositoryIdentity);
794 }),
795 );
796
797 it.effect("scope teardown flush stays stable while a save is blocked", () =>
798 Effect.gen(function* () {
799 const lateSnapshot: OrchestrationV2ShellSnapshot = {
800 ...LIVE_SHELL_SNAPSHOT,
801 snapshotSequence: 2,
802 projects: [{ ...v2Project, title: "Late stream event" }],
803 };
804 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
805 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
806 const saveStarted = yield* Deferred.make<void>();
807 const releaseSave = yield* Deferred.make<void>();
808 const client = {
809 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
810 } as unknown as WsRpcProtocolClient;
811 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
812 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
813 Option.some(session(client)),
814 );
815 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
816 target: TARGET,
817 state: supervisorState,
818 session: activeSession,
819 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
820 connect: Effect.void,
821 disconnect: Effect.void,
822 retryNow: Effect.void,
823 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
824 const cache = Persistence.EnvironmentCacheStore.of({
825 loadShell: () => Effect.succeed(Option.none()),
826 saveShell: (_environmentId, snapshot) =>
827 Deferred.succeed(saveStarted, undefined).pipe(
828 Effect.andThen(Deferred.await(releaseSave)),
829 Effect.andThen(Ref.update(saved, (values) => [...values, snapshot])),
830 ),
831 loadThread: () => Effect.succeed(Option.none()),
832 saveThread: () => Effect.void,
833 removeThread: () => Effect.void,
834 loadServerConfig: () => Effect.succeed(Option.none()),
835 saveServerConfig: () => Effect.void,
836 loadVcsRefs: () => Effect.succeed(Option.none()),
837 saveVcsRefs: () => Effect.void,
838 removeVcsRefs: () => Effect.void,
839 clearVcsRefs: () => Effect.void,
840 clear: () => Effect.void,
841 });
842 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
843 load: () => Effect.succeed(Option.none()),
844 });
845
846 const scope = yield* Scope.make();
847 const shellState = yield* makeEnvironmentShellState().pipe(
848 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
849 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
850 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
851 Scope.provide(scope),
852 );
853
854 yield* SubscriptionRef.set(supervisorState, {
855 desired: true,
856 network: "online",
857 phase: "connecting",
858 stage: "synchronizing",
859 attempt: 1,
860 generation: 0,
861 lastFailure: null,
862 retryAt: null,
863 });
864 yield* Queue.offer(events, {
865 kind: "snapshot",
866 snapshot: LIVE_SHELL_SNAPSHOT,
867 });
868 yield* Queue.offer(events, { kind: "synchronized" });
869 yield* SubscriptionRef.changes(shellState).pipe(
870 Stream.filter((state) => state.status === "live"),
871 Stream.runHead,
872 );
873
874 // Close scope so the finalizer flush blocks on save. Workers must be
875 // interrupted before that flush; late stream events must not extend it.
876 const closeFiber = yield* Effect.forkChild(Scope.close(scope, Exit.void));
877 yield* Deferred.await(saveStarted);
878 yield* Queue.offer(events, {
879 kind: "snapshot",
880 snapshot: lateSnapshot,
881 });
882 for (let index = 0; index < 50; index += 1) {
883 yield* Effect.yieldNow;
884 }
885 yield* Deferred.succeed(releaseSave, undefined);
886 yield* Fiber.join(closeFiber);
887
888 const values = yield* Ref.get(saved);
889 expect(values.length).toBeGreaterThan(0);
890 expect(values.every((snapshot) => snapshot.snapshotSequence === 1)).toBe(true);
891 expect(values.every((snapshot) => snapshot.projects[0]?.title !== "Late stream event")).toBe(
892 true,
893 );
894 }),
895 );
896
897 it.effect("applies authoritative lower-sequence HTTP resets over client-ahead cache", () =>
898 Effect.gen(function* () {
899 const cachedSnapshot: OrchestrationV2ShellSnapshot = {
900 ...v2ShellSnapshot,
901 snapshotSequence: 20,
902 projects: [{ ...v2Project, title: "Client ahead" }],
903 };
904 const httpSnapshot: OrchestrationV2ShellSnapshot = {
905 ...v2ShellSnapshot,
906 snapshotSequence: 4,
907 projects: [{ ...v2Project, title: "Server reset" }],
908 };
909 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
910 const client = {
911 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
912 } as unknown as WsRpcProtocolClient;
913 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
914 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
915 target: TARGET,
916 state: supervisorState,
917 session: yield* SubscriptionRef.make(Option.some(session(client))),
918 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
919 connect: Effect.void,
920 disconnect: Effect.void,
921 retryNow: Effect.void,
922 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
923 const cache = Persistence.EnvironmentCacheStore.of({
924 loadShell: () => Effect.succeed(Option.some(cachedSnapshot)),
925 saveShell: () => Effect.void,
926 loadThread: () => Effect.succeed(Option.none()),
927 saveThread: () => Effect.void,
928 removeThread: () => Effect.void,
929 loadServerConfig: () => Effect.succeed(Option.none()),
930 saveServerConfig: () => Effect.void,
931 loadVcsRefs: () => Effect.succeed(Option.none()),
932 saveVcsRefs: () => Effect.void,
933 removeVcsRefs: () => Effect.void,
934 clearVcsRefs: () => Effect.void,
935 clear: () => Effect.void,
936 });
937 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
938 load: () => Effect.succeed(Option.some(httpSnapshot)),
939 });
940 const shellState = yield* makeEnvironmentShellState().pipe(
941 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
942 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
943 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
944 );
945
946 yield* SubscriptionRef.changes(shellState).pipe(
947 Stream.filter(
948 (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 4,
949 ),
950 Stream.runHead,
951 );
952
953 const state = yield* SubscriptionRef.get(shellState);
954 expect(Option.getOrThrow(state.snapshot).snapshotSequence).toBe(4);
955 expect(Option.getOrThrow(state.snapshot).projects[0]?.title).toBe("Server reset");
956 }),
957 );
958
959 it.effect("fills deferred cached pull request links without overwriting newer rows", () =>
960 Effect.gen(function* () {
961 const link = (number: number): ThreadPullRequestLink => ({
962 host: "github.com",
963 repository: "pingdotgg/t3code",
964 number,
965 url: `https://github.com/pingdotgg/t3code/pull/${number}`,
966 source: "agent",
967 linkedAt: "2026-06-20T00:00:00.000Z",
968 snapshot: null,
969 stack: null,
970 });
971 const otherThreadId = ThreadId.make("thread-other");
972 const cachedSnapshot: OrchestrationV2ShellSnapshot = {
973 ...v2ShellSnapshot,
974 snapshotSequence: 5,
975 threads: [v2ThreadShell, { ...v2ThreadShell, id: otherThreadId }],
976 };
977 const linksReady = yield* Deferred.make<void>();
978 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
979 const client = {
980 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
981 } as unknown as WsRpcProtocolClient;
982 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
983 target: TARGET,
984 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
985 session: yield* SubscriptionRef.make(Option.some(session(client))),
986 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
987 connect: Effect.void,
988 disconnect: Effect.void,
989 retryNow: Effect.void,
990 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
991 const cache = Persistence.EnvironmentCacheStore.of({
992 loadShell: () =>
993 Effect.succeedSome({
994 ...cachedSnapshot,
995 loadPullRequests: Deferred.await(linksReady).pipe(
996 Effect.as(
997 new Map([
998 [v2ThreadShell.id, [link(1)]],
999 [otherThreadId, [link(2)]],
1000 ]),
1001 ),
1002 ),
1003 }),
1004 saveShell: () => Effect.void,
1005 loadThread: () => Effect.succeedNone,
1006 saveThread: () => Effect.void,
1007 removeThread: () => Effect.void,
1008 loadServerConfig: () => Effect.succeedNone,
1009 saveServerConfig: () => Effect.void,
1010 loadVcsRefs: () => Effect.succeedNone,
1011 saveVcsRefs: () => Effect.void,
1012 removeVcsRefs: () => Effect.void,
1013 clearVcsRefs: () => Effect.void,
1014 clear: () => Effect.void,
1015 });
1016 // No HTTP snapshot, so the cached rows stay until the socket sends deltas.
1017 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1018 load: () => Effect.succeedNone,
1019 });
1020 const shellState = yield* makeEnvironmentShellState().pipe(
1021 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1022 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1023 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1024 );
1025
1026 const threadsOf = (state: {
1027 readonly snapshot: Option.Option<OrchestrationV2ShellSnapshot>;
1028 }) => Option.getOrThrow(state.snapshot).threads;
1029 expect(threadsOf(yield* SubscriptionRef.get(shellState))).toEqual(cachedSnapshot.threads);
1030
1031 // The server updates one row before the cached links finish decoding.
1032 const serverRow = { ...v2ThreadShell, title: "From server", pullRequests: [link(9)] };
1033 yield* Queue.offer(events, {
1034 kind: "thread.updated",
1035 sequence: 6,
1036 location: "active",
1037 thread: serverRow,
1038 });
1039 yield* SubscriptionRef.changes(shellState).pipe(
1040 Stream.filter((state) => threadsOf(state)[0]?.title === "From server"),
1041 Stream.runHead,
1042 );
1043
1044 yield* Deferred.succeed(linksReady, undefined);
1045 const filled = yield* SubscriptionRef.changes(shellState).pipe(
1046 Stream.map(threadsOf),
1047 Stream.filter((threads) => threads[1]?.pullRequests !== undefined),
1048 Stream.runHead,
1049 Effect.map(Option.getOrThrow),
1050 );
1051 expect(filled[0]).toBe(serverRow);
1052 expect(filled[1]?.pullRequests).toEqual([link(2)]);
1053 }),
1054 );
1055
1056 it.effect("shows HTTP rows before their pull request links and saves them once filled", () =>
1057 Effect.gen(function* () {
1058 const link: ThreadPullRequestLink = {
1059 host: "github.com",
1060 repository: "pingdotgg/t3code",
1061 number: 7,
1062 url: "https://github.com/pingdotgg/t3code/pull/7",
1063 source: "agent",
1064 linkedAt: "2026-06-20T00:00:00.000Z",
1065 snapshot: null,
1066 stack: null,
1067 };
1068 const httpSnapshot: OrchestrationV2ShellSnapshot = {
1069 ...v2ShellSnapshot,
1070 snapshotSequence: 3,
1071 };
1072 const linksReady = yield* Deferred.make<void>();
1073 const saved = yield* Queue.unbounded<OrchestrationV2ShellSnapshot>();
1074 const client = {
1075 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.never,
1076 } as unknown as WsRpcProtocolClient;
1077 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
1078 target: TARGET,
1079 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
1080 session: yield* SubscriptionRef.make(Option.some(session(client))),
1081 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
1082 connect: Effect.void,
1083 disconnect: Effect.void,
1084 retryNow: Effect.void,
1085 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
1086 const cache = Persistence.EnvironmentCacheStore.of({
1087 loadShell: () => Effect.succeedNone,
1088 saveShell: (_environmentId, snapshot) => Queue.offer(saved, snapshot),
1089 loadThread: () => Effect.succeedNone,
1090 saveThread: () => Effect.void,
1091 removeThread: () => Effect.void,
1092 loadServerConfig: () => Effect.succeedNone,
1093 saveServerConfig: () => Effect.void,
1094 loadVcsRefs: () => Effect.succeedNone,
1095 saveVcsRefs: () => Effect.void,
1096 removeVcsRefs: () => Effect.void,
1097 clearVcsRefs: () => Effect.void,
1098 clear: () => Effect.void,
1099 });
1100 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1101 load: () =>
1102 Effect.succeedSome({
1103 ...httpSnapshot,
1104 loadPullRequests: Deferred.await(linksReady).pipe(
1105 Effect.as(new Map([[v2ThreadShell.id, [link]]])),
1106 ),
1107 }),
1108 });
1109 const shellState = yield* makeEnvironmentShellState().pipe(
1110 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1111 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1112 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1113 );
1114
1115 const rows = yield* SubscriptionRef.changes(shellState).pipe(
1116 Stream.filter((state) => Option.isSome(state.snapshot)),
1117 Stream.runHead,
1118 Effect.map((state) => Option.getOrThrow(Option.getOrThrow(state).snapshot)),
1119 );
1120 expect(rows.threads[0]?.pullRequests).toBeUndefined();
1121 expect(rows).not.toHaveProperty("loadPullRequests");
1122
1123 yield* Deferred.succeed(linksReady, undefined);
1124 const filled = yield* SubscriptionRef.changes(shellState).pipe(
1125 Stream.map((state) => Option.getOrThrow(state.snapshot).threads[0]),
1126 Stream.filter((thread) => thread?.pullRequests !== undefined),
1127 Stream.runHead,
1128 Effect.map(Option.getOrThrow),
1129 );
1130 expect(filled?.pullRequests).toEqual([link]);
1131
1132 // The cache must end up with the links: the last save carries them.
1133 yield* TestClock.adjust("30 seconds");
1134 const saves = yield* Queue.takeAll(saved);
1135 expect(saves.at(-1)?.threads[0]?.pullRequests).toEqual([link]);
1136 }),
1137 );
1138
1139 it.effect("does not save rows whose deferred links are still pending", () =>
1140 Effect.gen(function* () {
1141 const link: ThreadPullRequestLink = {
1142 host: "github.com",
1143 repository: "pingdotgg/t3code",
1144 number: 8,
1145 url: "https://github.com/pingdotgg/t3code/pull/8",
1146 source: "agent",
1147 linkedAt: "2026-06-20T00:00:00.000Z",
1148 snapshot: null,
1149 stack: null,
1150 };
1151 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
1152 const client = {
1153 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.never,
1154 } as unknown as WsRpcProtocolClient;
1155 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
1156 target: TARGET,
1157 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
1158 session: yield* SubscriptionRef.make(Option.some(session(client))),
1159 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
1160 connect: Effect.void,
1161 disconnect: Effect.void,
1162 retryNow: Effect.void,
1163 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
1164 const cache = Persistence.EnvironmentCacheStore.of({
1165 loadShell: () => Effect.succeedNone,
1166 saveShell: (_environmentId, snapshot) => Ref.update(saved, (all) => [...all, snapshot]),
1167 loadThread: () => Effect.succeedNone,
1168 saveThread: () => Effect.void,
1169 removeThread: () => Effect.void,
1170 loadServerConfig: () => Effect.succeedNone,
1171 saveServerConfig: () => Effect.void,
1172 loadVcsRefs: () => Effect.succeedNone,
1173 saveVcsRefs: () => Effect.void,
1174 removeVcsRefs: () => Effect.void,
1175 clearVcsRefs: () => Effect.void,
1176 clear: () => Effect.void,
1177 });
1178 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1179 load: () =>
1180 Effect.succeedSome({
1181 ...v2ShellSnapshot,
1182 snapshotSequence: 3,
1183 // Links never finish decoding before the shell closes.
1184 loadPullRequests: Effect.never.pipe(Effect.as(new Map([[v2ThreadShell.id, [link]]]))),
1185 }),
1186 });
1187 const scope = yield* Scope.make();
1188 const shellState = yield* makeEnvironmentShellState().pipe(
1189 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1190 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1191 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1192 Scope.provide(scope),
1193 );
1194 yield* SubscriptionRef.changes(shellState).pipe(
1195 Stream.filter((state) => Option.isSome(state.snapshot)),
1196 Stream.runHead,
1197 );
1198
1199 // Neither the throttled worker nor the closing flush may save the rows without links.
1200 yield* TestClock.adjust("30 seconds");
1201 yield* Scope.close(scope, Exit.void);
1202 expect(yield* Ref.get(saved)).toEqual([]);
1203 }),
1204 );
1205
1206 it.effect("drops an older snapshot's links once a newer snapshot reuses its rows", () =>
1207 Effect.gen(function* () {
1208 const link = (number: number): ThreadPullRequestLink => ({
1209 host: "github.com",
1210 repository: "pingdotgg/t3code",
1211 number,
1212 url: `https://github.com/pingdotgg/t3code/pull/${number}`,
1213 source: "agent",
1214 linkedAt: "2026-06-20T00:00:00.000Z",
1215 snapshot: null,
1216 stack: null,
1217 });
1218 const firstLinks = yield* Deferred.make<void>();
1219 const secondLinks = yield* Deferred.make<void>();
1220 // Set once each snapshot's links start decoding, i.e. once its rows are applied.
1221 const firstFillStarted = yield* Deferred.make<void>();
1222 const loads = yield* Ref.make(0);
1223 const subscriptions = yield* Ref.make(0);
1224 const client = {
1225 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () =>
1226 Stream.unwrap(Ref.update(subscriptions, (n) => n + 1).pipe(Effect.as(Stream.never))),
1227 } as unknown as WsRpcProtocolClient;
1228 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
1229 Option.some(session(client)),
1230 );
1231 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
1232 target: TARGET,
1233 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
1234 session: activeSession,
1235 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
1236 connect: Effect.void,
1237 disconnect: Effect.void,
1238 retryNow: Effect.void,
1239 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
1240 const cache = Persistence.EnvironmentCacheStore.of({
1241 loadShell: () => Effect.succeedNone,
1242 saveShell: () => Effect.void,
1243 loadThread: () => Effect.succeedNone,
1244 saveThread: () => Effect.void,
1245 removeThread: () => Effect.void,
1246 loadServerConfig: () => Effect.succeedNone,
1247 saveServerConfig: () => Effect.void,
1248 loadVcsRefs: () => Effect.succeedNone,
1249 saveVcsRefs: () => Effect.void,
1250 removeVcsRefs: () => Effect.void,
1251 clearVcsRefs: () => Effect.void,
1252 clear: () => Effect.void,
1253 });
1254 // Both snapshots carry the same row; only its pull request links differ.
1255 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1256 load: () =>
1257 Ref.updateAndGet(loads, (n) => n + 1).pipe(
1258 Effect.map((n) =>
1259 Option.some({
1260 ...v2ShellSnapshot,
1261 snapshotSequence: n,
1262 loadPullRequests: (n === 1
1263 ? Deferred.succeed(firstFillStarted, undefined).pipe(
1264 Effect.andThen(Deferred.await(firstLinks)),
1265 )
1266 : Deferred.await(secondLinks)
1267 ).pipe(Effect.as(new Map([[v2ThreadShell.id, [link(n)]]]))),
1268 }),
1269 ),
1270 ),
1271 });
1272 const shellState = yield* makeEnvironmentShellState().pipe(
1273 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1274 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1275 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1276 );
1277 yield* Deferred.await(firstFillStarted);
1278
1279 // A new session loads a second snapshot that reuses the bare row.
1280 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
1281 yield* SubscriptionRef.changes(shellState).pipe(
1282 Stream.filter(
1283 (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 2,
1284 ),
1285 Stream.runHead,
1286 );
1287
1288 // The superseded first fill finishes first, then the second.
1289 yield* Deferred.succeed(firstLinks, undefined);
1290 yield* Deferred.succeed(secondLinks, undefined);
1291 const filled = yield* SubscriptionRef.changes(shellState).pipe(
1292 Stream.map((state) => Option.getOrThrow(state.snapshot).threads[0]?.pullRequests),
1293 Stream.filter((links) => links !== undefined),
1294 Stream.runHead,
1295 Effect.map(Option.getOrThrow),
1296 );
1297 expect(filled).toEqual([link(2)]);
1298 }),
1299 );
1300});