1import {
2 OrchestrationV2ThreadDetailSnapshot,
3 EnvironmentId,
4 EventId,
5 MessageId,
6 ORCHESTRATION_V2_WS_METHODS,
7 ThreadId,
8 TurnItemId,
9 type OrchestrationV2ThreadProjection,
10 type OrchestrationV2ThreadStreamItem,
11 type OrchestrationV2TurnItem,
12} from "@t3tools/contracts";
13import { describe, expect, it } from "@effect/vitest";
14import * as DateTime from "effect/DateTime";
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 Option from "effect/Option";
20import * as Queue from "effect/Queue";
21import * as Ref from "effect/Ref";
22import * as Schema from "effect/Schema";
23import * as HubConnection from "../connection/hub.ts";
24import * as Scope from "effect/Scope";
25import * as Stream from "effect/Stream";
26import * as SubscriptionRef from "effect/SubscriptionRef";
27import * as TestClock from "effect/testing/TestClock";
28import { HttpClient, HttpClientResponse } from "effect/http";
29
30import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
31import {
32 AVAILABLE_CONNECTION_STATE,
33 PrimaryConnectionTarget,
34 type PreparedConnection,
35 type SupervisorConnectionState,
36} from "../connection/model.ts";
37import * as ConnectionWakeups from "../connection/wakeups.ts";
38import * as EnvironmentSupervisor from "../connection/supervisor.ts";
39import * as Persistence from "../platform/persistence.ts";
40import * as RpcSession from "../rpc/session.ts";
41import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts";
42import * as ThreadHistoryController from "./threadHistoryController.ts";
43import {
44 EMPTY_ENVIRONMENT_THREAD_STATE,
45 makeEnvironmentThreadState,
46 type EnvironmentThreadState,
47 type ThreadSnapshotLoadResult,
48} from "./threads.ts";
49import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
50
51const encodeArchiveSnapshot = Schema.encodeEffect(
52 Schema.toCodecJson(OrchestrationV2ThreadDetailSnapshot),
53);
54const TARGET = new PrimaryConnectionTarget({
55 environmentId: EnvironmentId.make("environment-1"),
56 label: "Test environment",
57 httpBaseUrl: "https://environment.example.test",
58 wsBaseUrl: "wss://environment.example.test",
59});
60const THREAD_ID = v2ThreadId;
61const CACHED_SNAPSHOT_SEQUENCE = 7;
62const PREPARED: PreparedConnection = {
63 environmentId: TARGET.environmentId,
64 label: TARGET.label,
65 httpBaseUrl: TARGET.httpBaseUrl,
66 socketUrl: TARGET.wsBaseUrl,
67 httpAuthorization: null,
68 target: TARGET,
69};
70const BASE_PROJECTION: OrchestrationV2ThreadProjection = {
71 ...v2Projection,
72 thread: { ...v2Projection.thread, title: "Cached thread" },
73};
74
75type TestThreadInput = OrchestrationV2ThreadStreamItem | Error;
76
77function testSession(
78 client: WsRpcProtocolClient,
79 config?: { readonly completionMarker?: boolean },
80): RpcSession.RpcSession {
81 return {
82 client,
83 initialConfig: Effect.succeed(
84 config?.completionMarker === true
85 ? ({ threadResumeCompletionMarker: true } as never)
86 : ({} as never),
87 ),
88 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
89 ready: Effect.void,
90 probe: Effect.void,
91 closed: Effect.never,
92 };
93}
94
95function awaitThreadState(
96 observed: Queue.Queue<EnvironmentThreadState>,
97 predicate: (state: EnvironmentThreadState) => boolean,
98) {
99 return Queue.take(observed).pipe(
100 Effect.repeat({
101 until: predicate,
102 }),
103 );
104}
105
106const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: {
107 readonly hub?: HubConnection.HubConnection["Service"];
108 readonly offline?: boolean;
109 readonly cached?: OrchestrationV2ThreadProjection;
110 readonly cachedHistory?: {
111 readonly historyCursor: string | null;
112 readonly hasMoreHistory: boolean;
113 readonly latestLocalTurnOrdinal?: number | null;
114 };
115 readonly httpSnapshot?: ThreadSnapshotLoadResult;
116 readonly completionMarker?: boolean;
117 readonly resumeCache?: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]>;
118 readonly loadCached?: ReturnType<Persistence.EnvironmentCacheStore["Service"]["loadThread"]>;
119 readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"];
120 readonly historyPaging?: "enabled" | "no-http" | "no-controller";
121 readonly historyHttpClient?: HttpClient.HttpClient;
122}) {
123 const inputs = yield* Queue.unbounded<TestThreadInput>();
124 const observed = yield* Queue.unbounded<EnvironmentThreadState>();
125 const latest = yield* Ref.make<EnvironmentThreadState>(EMPTY_ENVIRONMENT_THREAD_STATE);
126 const retryCount = yield* Ref.make(0);
127 const subscriptionCount = yield* Ref.make(0);
128 const loaderCalls = yield* Ref.make(0);
129 const lastSubscribeAfterSequence = yield* Ref.make<number | undefined>(undefined);
130 const lastRequestCompletionMarker = yield* Ref.make(false);
131 const lastAcceptBoundedSnapshot = yield* Ref.make<true | undefined>(undefined);
132 const lastAcceptCompactTurnItems = yield* Ref.make<true | undefined>(undefined);
133 const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();
134 const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationV2ThreadDetailSnapshot>>([]);
135 const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);
136 const supervisorState = yield* SubscriptionRef.make<SupervisorConnectionState>(
137 AVAILABLE_CONNECTION_STATE,
138 );
139 const streamFrom = (queue: Queue.Queue<TestThreadInput>) =>
140 Stream.fromQueue(queue).pipe(
141 Stream.mapEffect((input) =>
142 input instanceof Error ? Effect.fail(input) : Effect.succeed(input),
143 ),
144 );
145 const client = {
146 [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input: {
147 readonly afterSequence?: number;
148 readonly requestCompletionMarker?: true;
149 readonly acceptBoundedSnapshot?: true;
150 readonly acceptCompactTurnItems?: true;
151 }) =>
152 Stream.unwrap(
153 Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(
154 Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)),
155 Effect.andThen(
156 Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker === true),
157 ),
158 Effect.andThen(Ref.set(lastAcceptBoundedSnapshot, input.acceptBoundedSnapshot)),
159 Effect.andThen(Ref.set(lastAcceptCompactTurnItems, input.acceptCompactTurnItems)),
160 Effect.as(streamFrom(inputs)),
161 ),
162 ),
163 } as unknown as WsRpcProtocolClient;
164 const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
165 options?.offline ? Option.none() : Option.some(testSession(client, options)),
166 );
167 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
168 Option.some(PREPARED),
169 );
170 const snapshotLoader = ThreadSnapshotLoader.ThreadSnapshotLoader.of({
171 load: (_prepared, threadId) =>
172 Ref.update(loaderCalls, (count) => count + 1).pipe(
173 Effect.as(
174 threadId === THREAD_ID
175 ? (options?.httpSnapshot ??
176 ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult))
177 : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),
178 ),
179 ),
180 });
181 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
182 target: TARGET,
183 state: supervisorState,
184 session: supervisorSession,
185 prepared,
186 connect: Effect.void,
187 disconnect: Effect.void,
188 retryNow: Ref.update(retryCount, (count) => count + 1),
189 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
190 const cache = Persistence.EnvironmentCacheStore.of({
191 loadShell: () => Effect.succeedNone,
192 saveShell: () => Effect.void,
193 loadThread: (_environmentId, threadId) =>
194 options?.loadCached ??
195 Effect.succeed(
196 threadId === THREAD_ID && options?.cached !== undefined
197 ? Option.some({
198 snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,
199 projection: options.cached,
200 ...(options.cachedHistory === undefined
201 ? {}
202 : {
203 historyCursor: options.cachedHistory.historyCursor,
204 hasMoreHistory: options.cachedHistory.hasMoreHistory,
205 ...(options.cachedHistory.latestLocalTurnOrdinal === undefined
206 ? {}
207 : {
208 latestLocalTurnOrdinal: options.cachedHistory.latestLocalTurnOrdinal,
209 }),
210 }),
211 })
212 : Option.none(),
213 ),
214 saveThread: (environmentId, thread) =>
215 Ref.update(savedThreads, (current) => [...current, thread]).pipe(
216 Effect.andThen(options?.saveThread?.(environmentId, thread) ?? Effect.void),
217 ),
218 removeThread: (_environmentId, threadId) =>
219 Ref.update(removedThreads, (current) => [...current, threadId]),
220 loadServerConfig: () => Effect.succeedNone,
221 saveServerConfig: () => Effect.void,
222 loadVcsRefs: () => Effect.succeedNone,
223 saveVcsRefs: () => Effect.void,
224 removeVcsRefs: () => Effect.void,
225 clearVcsRefs: () => Effect.void,
226 clear: () => Effect.void,
227 });
228 let makeThreadState = makeEnvironmentThreadState(THREAD_ID, options?.resumeCache).pipe(
229 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
230 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
231 Effect.provideService(ThreadSnapshotLoader.ThreadSnapshotLoader, snapshotLoader),
232 Effect.provideService(
233 ConnectionWakeups.ConnectionWakeups,
234 ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }),
235 ),
236 );
237 if (options?.historyPaging !== "no-http") {
238 makeThreadState = makeThreadState.pipe(
239 Effect.provideService(
240 HttpClient.HttpClient,
241 options?.historyHttpClient ??
242 HttpClient.make(() => Effect.die("Unexpected history HTTP request")),
243 ),
244 );
245 }
246 const historyController = yield* ThreadHistoryController.ThreadHistoryController.pipe(
247 Effect.provide(ThreadHistoryController.layer),
248 );
249 if (options?.historyPaging !== "no-controller") {
250 makeThreadState = makeThreadState.pipe(
251 Effect.provideService(ThreadHistoryController.ThreadHistoryController, historyController),
252 );
253 }
254 if (options?.hub)
255 makeThreadState = makeThreadState.pipe(
256 Effect.provideService(HubConnection.HubConnection, options.hub),
257 );
258 const threadState = yield* makeThreadState;
259 yield* SubscriptionRef.changes(threadState).pipe(
260 Stream.runForEach((state) =>
261 Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))),
262 ),
263 Effect.forkScoped,
264 );
265
266 return {
267 threadState,
268 loadEarlier: (throughEntryId?: string) =>
269 historyController.loadEarlier(TARGET.environmentId, THREAD_ID, throughEntryId),
270 inputs,
271 observed,
272 latest,
273 retryCount,
274 subscriptionCount,
275 loaderCalls,
276 lastSubscribeAfterSequence,
277 lastRequestCompletionMarker,
278 lastAcceptBoundedSnapshot,
279 lastAcceptCompactTurnItems,
280 supervisorState,
281 supervisorSession,
282 savedThreads,
283 removedThreads,
284 wakeups,
285 replaceSession: SubscriptionRef.set(
286 supervisorSession,
287 Option.some(testSession(client, options)),
288 ),
289 };
290});
291
292const snapshot = (
293 projection: OrchestrationV2ThreadProjection,
294 snapshotSequence = 1,
295): Extract<OrchestrationV2ThreadStreamItem, { readonly kind: "snapshot" }> => ({
296 kind: "snapshot",
297 snapshotSequence,
298 projection,
299});
300
301const synchronized = (): OrchestrationV2ThreadStreamItem => ({ kind: "synchronized" });
302
303const titleUpdated = (title: string, sequence = 2): OrchestrationV2ThreadStreamItem => {
304 const occurredAt = DateTime.makeUnsafe("2026-06-20T01:00:00.000Z");
305 return {
306 kind: "event",
307 sequence,
308 event: {
309 id: EventId.make("event-title"),
310 type: "thread.metadata-updated",
311 threadId: THREAD_ID,
312 occurredAt,
313 payload: { ...v2Projection.thread, title, updatedAt: occurredAt },
314 },
315 };
316};
317
318const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => {
319 const occurredAt = DateTime.makeUnsafe("2026-06-20T02:00:00.000Z");
320 return {
321 kind: "event",
322 sequence,
323 event: {
324 id: EventId.make("event-deleted"),
325 type: "thread.deleted",
326 threadId: THREAD_ID,
327 occurredAt,
328 payload: { ...v2Projection.thread, updatedAt: occurredAt, deletedAt: occurredAt },
329 },
330 };
331};
332
333describe("EnvironmentThreads", () => {
334 it.effect(
335 "opens a NAS archived transcript without a source session and resumes its source sequence after reconnect",
336 () =>
337 Effect.gen(function* () {
338 const archived = {
339 ...BASE_PROJECTION,
340 thread: {
341 ...BASE_PROJECTION.thread,
342 archivedAt: DateTime.makeUnsafe("2026-06-20T02:00:00Z"),
343 },
344 };
345 const archiveSnapshot = { snapshotSequence: 81, projection: archived };
346 const hubState = yield* SubscriptionRef.make<HubConnection.HubConnectionState>({
347 ...HubConnection.EMPTY_HUB_CONNECTION_STATE,
348 connection: {
349 environmentId: EnvironmentId.make("archive-hub"),
350 label: "Archive",
351 httpBaseUrl: "https://archive.example.test",
352 },
353 });
354 const hub = HubConnection.HubConnection.of({
355 state: hubState,
356 headers: Effect.succeed({ authorization: "Bearer ephemeral-archive-access" }),
357 connect: () => Effect.void,
358 connectAuthenticated: () => Effect.void,
359 refresh: Effect.void,
360 refreshInbox: Effect.void,
361 organize: () => Effect.succeedNone,
362 disconnect: Effect.void,
363 });
364 const json = yield* encodeArchiveSnapshot(archiveSnapshot);
365 const http = HttpClient.make((request) =>
366 Effect.sync(() => {
367 expect(request.url).toBe(
368 `https://archive.example.test/api/archive/catalog/${TARGET.environmentId}/${THREAD_ID}`,
369 );
370 expect(request.headers.authorization).toBe("Bearer ephemeral-archive-access");
371 return HttpClientResponse.fromWeb(request, Response.json(json));
372 }),
373 );
374 const h = yield* makeHarness({ hub, offline: true, historyHttpClient: http });
375 const state = yield* awaitThreadState(h.observed, (value) => Option.isSome(value.data));
376 expect(state.status).toBe("cached");
377 expect(Option.getOrThrow(state.data)).toEqual(archived);
378 expect(yield* Ref.get(h.subscriptionCount)).toBe(0);
379 expect(yield* Ref.get(h.loaderCalls)).toBe(0);
380 yield* h.replaceSession;
381 yield* Queue.offer(h.inputs, titleUpdated("Source reconnected", 82));
382 const live = yield* awaitThreadState(
383 h.observed,
384 (value) =>
385 Option.isSome(value.data) && value.data.value.thread.title === "Source reconnected",
386 );
387 expect(live.status).toBe("live");
388 expect(yield* Ref.get(h.lastSubscribeAfterSequence)).toBe(81);
389 }),
390 );
391
392 it.effect("loads the server thread when its local cache read fails", () =>
393 Effect.gen(function* () {
394 const h = yield* makeHarness({
395 loadCached: Effect.fail(
396 new Persistence.ConnectionPersistenceError({
397 operation: "load-thread",
398 message: "The database connection is closing.",
399 }),
400 ),
401 httpSnapshot: {
402 _tag: "present",
403 snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },
404 },
405 });
406
407 const state = yield* awaitThreadState(h.observed, (value) => value.status === "live");
408 expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION);
409 expect(yield* Ref.get(h.loaderCalls)).toBe(1);
410 expect(yield* Ref.get(h.subscriptionCount)).toBe(1);
411 }),
412 );
413
414 it.effect.each(["disk", "HTTP"] as const)(
415 "does not rewrite an unchanged %s snapshot on navigation or warm return",
416 (source) =>
417 Effect.gen(function* () {
418 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
419 snapshot: undefined,
420 owner: undefined,
421 };
422 const firstSaved = yield* Effect.scoped(
423 Effect.gen(function* () {
424 const h = yield* makeHarness({
425 resumeCache,
426 ...(source === "disk"
427 ? { cached: BASE_PROJECTION }
428 : {
429 httpSnapshot: {
430 _tag: "present",
431 snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },
432 },
433 }),
434 });
435 yield* awaitThreadState(h.observed, (value) => value.status === "live");
436 if (source === "HTTP") yield* TestClock.adjust("500 millis");
437 return h.savedThreads;
438 }),
439 );
440 expect(yield* Ref.get(firstSaved)).toHaveLength(source === "disk" ? 0 : 1);
441 const nextSaved = yield* Effect.scoped(
442 Effect.gen(function* () {
443 const h = yield* makeHarness({ resumeCache });
444 yield* awaitThreadState(h.observed, (value) => value.status === "live");
445 return h.savedThreads;
446 }),
447 );
448 expect(yield* Ref.get(nextSaved)).toEqual([]);
449 }),
450 );
451
452 it.effect("persists a complete bounded HTTP window only once", () =>
453 Effect.gen(function* () {
454 const saved = yield* Effect.scoped(
455 Effect.gen(function* () {
456 const h = yield* makeHarness({
457 httpSnapshot: {
458 _tag: "present",
459 snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },
460 history: {
461 historyCursor: null,
462 hasMoreHistory: false,
463 latestLocalTurnOrdinal: 12,
464 },
465 },
466 });
467 yield* awaitThreadState(h.observed, (value) => value.status === "live");
468 yield* TestClock.adjust("500 millis");
469 return h.savedThreads;
470 }),
471 );
472 expect(yield* Ref.get(saved)).toEqual([
473 {
474 snapshotSequence: 7,
475 projection: BASE_PROJECTION,
476 historyCursor: null,
477 hasMoreHistory: false,
478 latestLocalTurnOrdinal: 12,
479 },
480 ]);
481 }),
482 );
483
484 it.effect("retries a failed background cache write when the scope closes", () =>
485 Effect.gen(function* () {
486 let attempts = 0;
487 yield* Effect.scoped(
488 Effect.gen(function* () {
489 const h = yield* makeHarness({
490 httpSnapshot: {
491 _tag: "present",
492 snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },
493 },
494 saveThread: () =>
495 Effect.suspend(() => {
496 attempts += 1;
497 return attempts === 1
498 ? Effect.fail(
499 new Persistence.ConnectionPersistenceError({
500 operation: "save-thread",
501 message: "Test storage failure",
502 }),
503 )
504 : Effect.void;
505 }),
506 });
507 yield* awaitThreadState(h.observed, (value) => value.status === "live");
508 yield* TestClock.adjust("500 millis");
509 expect(attempts).toBe(1);
510 }),
511 );
512 expect(attempts).toBe(2);
513 }),
514 );
515
516 it.effect("flushes newer data after an older background write completes", () =>
517 Effect.gen(function* () {
518 const writing = yield* Deferred.make<void>();
519 const written = yield* Deferred.make<void>();
520 const release = yield* Deferred.make<void>();
521 const saved = yield* Effect.scoped(
522 Effect.gen(function* () {
523 const h = yield* makeHarness({
524 cached: BASE_PROJECTION,
525 saveThread: (_environmentId, snapshot) =>
526 snapshot.snapshotSequence === 8
527 ? Deferred.succeed(writing, undefined).pipe(
528 Effect.andThen(Deferred.await(release)),
529 Effect.andThen(Deferred.succeed(written, undefined)),
530 )
531 : Effect.void,
532 });
533 yield* Queue.offer(h.inputs, titleUpdated("First update", 8));
534 yield* awaitThreadState(
535 h.observed,
536 (value) => Option.getOrNull(value.data)?.thread.title === "First update",
537 );
538 yield* TestClock.adjust("500 millis");
539 yield* Deferred.await(writing);
540 yield* Queue.offer(h.inputs, titleUpdated("Newer update", 9));
541 yield* awaitThreadState(
542 h.observed,
543 (value) => Option.getOrNull(value.data)?.thread.title === "Newer update",
544 );
545 yield* Deferred.succeed(release, undefined);
546 yield* Deferred.await(written);
547 return h.savedThreads;
548 }),
549 );
550 expect(
551 (yield* Ref.get(saved)).map((snapshot) => [
552 snapshot.snapshotSequence,
553 snapshot.projection.thread.title,
554 ]),
555 ).toEqual([
556 [8, "First update"],
557 [9, "Newer update"],
558 ]);
559 }),
560 );
561
562 it.effect("does not resume past a canceled event whose data was not applied", () =>
563 Effect.gen(function* () {
564 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
565 snapshot: undefined,
566 owner: undefined,
567 };
568 const scope = yield* Effect.acquireRelease(Scope.make(), (scope) =>
569 Scope.close(scope, Exit.void),
570 );
571 const first = yield* makeHarness({
572 httpSnapshot: {
573 _tag: "present",
574 snapshot: {
575 projection: BASE_PROJECTION,
576 snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,
577 },
578 },
579 resumeCache,
580 }).pipe(Effect.provideService(Scope.Scope, scope));
581 yield* awaitThreadState(first.observed, (value) => value.status === "live");
582 const applying = yield* Deferred.make<void>();
583 const update = titleUpdated("Not applied", 8);
584 if (update.kind !== "event") return yield* Effect.die("Expected an event");
585 let sequenceReads = 0;
586 Object.defineProperty(update, "sequence", {
587 get: () => {
588 if (++sequenceReads === 2) Deferred.doneUnsafe(applying, Exit.void);
589 return 8;
590 },
591 });
592 // Hold the projection write while the event advances its cursor. Closing
593 // the scope must resume from the last fully applied projection.
594 yield* first.threadState.semaphore.take(1);
595 yield* Queue.offer(first.inputs, update);
596 yield* Deferred.await(applying);
597 yield* Scope.close(scope, Exit.void);
598 yield* first.threadState.semaphore.release(1);
599
600 const resumed = yield* makeHarness({ resumeCache });
601 const state = yield* awaitThreadState(resumed.observed, (value) => value.status === "live");
602 expect(Option.getOrThrow(state.data).thread.title).toBe(BASE_PROJECTION.thread.title);
603 expect(yield* Ref.get(resumed.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);
604 expect(yield* Ref.get(first.savedThreads)).toEqual([
605 {
606 projection: BASE_PROJECTION,
607 snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,
608 historyCursor: null,
609 hasMoreHistory: false,
610 latestLocalTurnOrdinal: null,
611 },
612 ]);
613 }),
614 );
615
616 it.effect("prevents an old scope from replacing its successor's cached data", () =>
617 Effect.gen(function* () {
618 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
619 snapshot: undefined,
620 owner: undefined,
621 };
622 const oldScope = yield* Effect.acquireRelease(Scope.make(), (scope) =>
623 Scope.close(scope, Exit.void),
624 );
625 const old = yield* makeHarness({ cached: BASE_PROJECTION, resumeCache }).pipe(
626 Effect.provideService(Scope.Scope, oldScope),
627 );
628 yield* awaitThreadState(old.observed, (value) => value.status === "live");
629 const successor = yield* makeHarness({ resumeCache });
630 yield* Queue.offer(successor.inputs, titleUpdated("Successor title", 9));
631 yield* awaitThreadState(
632 successor.observed,
633 (value) => Option.getOrNull(value.data)?.thread.title === "Successor title",
634 );
635 yield* Queue.offer(old.inputs, titleUpdated("Old scope title", 8));
636 yield* awaitThreadState(
637 old.observed,
638 (value) => Option.getOrNull(value.data)?.thread.title === "Old scope title",
639 );
640 yield* Scope.close(oldScope, Exit.void);
641
642 expect(resumeCache.snapshot?.sequence).toBe(9);
643 expect(Option.getOrThrow(resumeCache.snapshot!.state.data).thread.title).toBe(
644 "Successor title",
645 );
646 expect(yield* Ref.get(old.savedThreads)).toEqual([]);
647 }),
648 );
649
650 it.effect("does not let a delayed cache read replace an initialized successor", () =>
651 Effect.gen(function* () {
652 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
653 snapshot: undefined,
654 owner: undefined,
655 };
656 const loading = yield* Deferred.make<void>();
657 const response = yield* Deferred.make<Option.Option<OrchestrationV2ThreadDetailSnapshot>>();
658 const oldFiber = yield* makeHarness({
659 resumeCache,
660 loadCached: Deferred.succeed(loading, undefined).pipe(
661 Effect.andThen(Deferred.await(response)),
662 ),
663 }).pipe(Effect.forkScoped);
664 yield* Deferred.await(loading);
665 const successor = yield* makeHarness({ cached: BASE_PROJECTION, resumeCache });
666 yield* awaitThreadState(successor.observed, (value) => value.status === "live");
667 yield* Deferred.succeed(
668 response,
669 Option.some({
670 snapshotSequence: 3,
671 projection: {
672 ...BASE_PROJECTION,
673 thread: { ...BASE_PROJECTION.thread, title: "Old disk data" },
674 },
675 }),
676 );
677 const old = yield* Fiber.join(oldFiber);
678 yield* awaitThreadState(old.observed, (value) => value.status === "live");
679
680 expect(resumeCache.snapshot?.sequence).toBe(CACHED_SNAPSHOT_SEQUENCE);
681 expect(Option.getOrThrow(resumeCache.snapshot!.state.data).thread.title).toBe(
682 BASE_PROJECTION.thread.title,
683 );
684 }),
685 );
686
687 it.effect("does not let an old deletion remove its successor's persisted cache", () =>
688 Effect.gen(function* () {
689 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
690 snapshot: undefined,
691 owner: undefined,
692 };
693 const old = yield* makeHarness({ cached: BASE_PROJECTION, resumeCache });
694 yield* awaitThreadState(old.observed, (value) => value.status === "live");
695 const successor = yield* makeHarness({ resumeCache });
696 yield* Queue.offer(successor.inputs, titleUpdated("Successor title", 9));
697 yield* awaitThreadState(
698 successor.observed,
699 (value) => Option.getOrNull(value.data)?.thread.title === "Successor title",
700 );
701 const update = deleted();
702 if (update.kind !== "event") return yield* Effect.die("Expected an event");
703 yield* Queue.offer(old.inputs, { ...update, sequence: 8 });
704 yield* awaitThreadState(old.observed, (value) => value.status === "deleted");
705
706 expect(resumeCache.snapshot?.sequence).toBe(9);
707 expect(yield* Ref.get(old.removedThreads)).toEqual([]);
708 }),
709 );
710
711 it.effect("retains a deletion instead of restoring the old disk snapshot", () =>
712 Effect.gen(function* () {
713 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
714 snapshot: undefined,
715 owner: undefined,
716 };
717 yield* Effect.scoped(
718 Effect.gen(function* () {
719 const first = yield* makeHarness({ cached: BASE_PROJECTION, resumeCache });
720 const update = deleted();
721 if (update.kind !== "event") return yield* Effect.die("Expected an event");
722 yield* Queue.offer(first.inputs, { ...update, sequence: 8 });
723 yield* awaitThreadState(first.observed, (value) => value.status === "deleted");
724 }),
725 );
726 const resumed = yield* makeHarness({ cached: BASE_PROJECTION, resumeCache });
727 const state = yield* awaitThreadState(
728 resumed.observed,
729 (value) => value.status === "deleted",
730 );
731 expect(Option.isNone(state.data)).toBe(true);
732 expect(yield* Ref.get(resumed.loaderCalls)).toBe(0);
733 }),
734 );
735
736 it.effect("publishes cached data immediately from a warm cache", () =>
737 Effect.gen(function* () {
738 const harness = yield* makeHarness({ cached: BASE_PROJECTION });
739 const state = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.data));
740
741 expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION);
742 expect(Option.isNone(state.error)).toBe(true);
743 }),
744 );
745
746 it.effect("resumes a warm cache via afterSequence without an HTTP fetch", () =>
747 Effect.gen(function* () {
748 const harness = yield* makeHarness({ cached: BASE_PROJECTION });
749
750 // The warm cache reaches live from the cached data, and a live event
751 // applies on top of it.
752 yield* Queue.offer(harness.inputs, titleUpdated("Live title", CACHED_SNAPSHOT_SEQUENCE + 1));
753 yield* awaitThreadState(
754 harness.observed,
755 (value) =>
756 value.status === "live" &&
757 Option.isSome(value.data) &&
758 value.data.value.thread.title === "Live title",
759 );
760
761 // The subscription resumed from the cached sequence and never fetched the
762 // full snapshot over HTTP.
763 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);
764 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
765 }),
766 );
767
768 it.effect("coalesces streaming cache writes and flushes the latest thread on teardown", () =>
769 Effect.gen(function* () {
770 const scope = yield* Scope.make();
771 const harness = yield* makeHarness().pipe(Effect.provideService(Scope.Scope, scope));
772 yield* Queue.offer(harness.inputs, snapshot(BASE_PROJECTION));
773 yield* awaitThreadState(harness.observed, (state) => state.status === "live");
774 yield* TestClock.adjust("500 millis");
775 expect(yield* Ref.get(harness.savedThreads)).toHaveLength(1);
776
777 for (let sequence = 2; sequence <= 10; sequence++) {
778 yield* Queue.offer(harness.inputs, titleUpdated(`Title ${sequence}`, sequence));
779 yield* awaitThreadState(
780 harness.observed,
781 (state) =>
782 Option.isSome(state.data) && state.data.value.thread.title === `Title ${sequence}`,
783 );
784 yield* TestClock.adjust("1 second");
785 }
786 expect(yield* Ref.get(harness.savedThreads)).toHaveLength(1);
787 yield* TestClock.adjust("1 second");
788 expect((yield* Ref.get(harness.savedThreads)).map((saved) => saved.snapshotSequence)).toEqual(
789 [1, 10],
790 );
791
792 yield* Queue.offer(harness.inputs, titleUpdated("Final title", 11));
793 yield* awaitThreadState(
794 harness.observed,
795 (state) => Option.isSome(state.data) && state.data.value.thread.title === "Final title",
796 );
797 yield* Scope.close(scope, Exit.void);
798 expect((yield* Ref.get(harness.savedThreads)).map((saved) => saved.snapshotSequence)).toEqual(
799 [1, 10, 11],
800 );
801 }),
802 );
803
804 it.effect("reduces live events and persists the latest thread", () =>
805 Effect.gen(function* () {
806 const harness = yield* makeHarness({ cached: BASE_PROJECTION });
807 yield* Queue.offer(harness.inputs, snapshot(BASE_PROJECTION));
808 yield* Queue.offer(harness.inputs, titleUpdated("Live title"));
809
810 const state = yield* awaitThreadState(
811 harness.observed,
812 (value) =>
813 value.status === "live" &&
814 Option.isSome(value.data) &&
815 value.data.value.thread.title === "Live title",
816 );
817 yield* TestClock.adjust("500 millis");
818 yield* Effect.yieldNow;
819
820 expect(Option.getOrThrow(state.data).thread.title).toBe("Live title");
821 expect((yield* Ref.get(harness.savedThreads)).at(-1)?.projection.thread.title).toBe(
822 "Live title",
823 );
824 expect((yield* Ref.get(harness.savedThreads)).at(-1)?.snapshotSequence).toBe(2);
825 }),
826 );
827
828 it.effect("keeps progressive history metadata from a bounded socket fallback", () =>
829 Effect.gen(function* () {
830 const olderItem = {
831 id: "item:older",
832 threadId: THREAD_ID,
833 runId: null,
834 nodeId: null,
835 providerThreadId: null,
836 providerTurnId: null,
837 nativeItemRef: null,
838 parentItemId: null,
839 ordinal: 1,
840 status: "completed",
841 title: null,
842 startedAt: "2026-06-20T00:00:00.000Z",
843 completedAt: "2026-06-20T00:00:00.000Z",
844 updatedAt: "2026-06-20T00:00:00.000Z",
845 type: "command_execution",
846 input: "pwd",
847 output: "older history",
848 exitCode: 0,
849 };
850 const harness = yield* makeHarness({
851 historyHttpClient: HttpClient.make((request, url) => {
852 expect(url.searchParams.get("cursor")).toBe("socket-history-cursor");
853 expect(url.searchParams.get("throughEntryId")).toBe(olderItem.id);
854 return Effect.succeed(
855 HttpClientResponse.fromWeb(
856 request,
857 Response.json({
858 snapshotSequence: 14,
859 items: [
860 {
861 position: 0,
862 visibility: "local",
863 sourceThreadId: THREAD_ID,
864 sourceItemId: olderItem.id,
865 item: olderItem,
866 },
867 ],
868 nextCursor: null,
869 hasMoreHistory: false,
870 }),
871 ),
872 );
873 }),
874 });
875 yield* Queue.offer(harness.inputs, {
876 kind: "snapshot",
877 snapshotSequence: 14,
878 projection: BASE_PROJECTION,
879 historyCursor: "socket-history-cursor",
880 hasMoreHistory: true,
881 latestLocalTurnOrdinal: 47,
882 payloadBudgetExceeded: false,
883 });
884
885 const state = yield* awaitThreadState(
886 harness.observed,
887 (value) =>
888 value.status === "live" && value.history.historyCursor === "socket-history-cursor",
889 );
890 expect(state.history).toMatchObject({
891 historyCursor: "socket-history-cursor",
892 hasMoreHistory: true,
893 loading: false,
894 error: null,
895 expanded: false,
896 latestLocalTurnOrdinal: 47,
897 });
898 // A transient cold HTTP failure still requests a bounded socket snapshot.
899 // Its cursor remains available for a later history request.
900 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBeUndefined();
901 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true);
902
903 yield* TestClock.adjust("500 millis");
904 yield* Effect.yieldNow;
905 expect((yield* Ref.get(harness.savedThreads)).at(-1)).toMatchObject({
906 snapshotSequence: 14,
907 historyCursor: "socket-history-cursor",
908 hasMoreHistory: true,
909 latestLocalTurnOrdinal: 47,
910 });
911 expect(yield* harness.loadEarlier(olderItem.id)).toEqual({ _tag: "loaded" });
912 const expanded = yield* SubscriptionRef.get(harness.threadState);
913 expect(
914 Option.getOrThrow(expanded.data).turnItems.some((item) => item.id === olderItem.id),
915 ).toBe(true);
916 expect(expanded.history.hasMoreHistory).toBe(false);
917 }),
918 );
919
920 it.effect.each(["complete", "replacement snapshot", "retry"] as const)(
921 "progressively reveals conversation history before activity, then handles %s",
922 (mode) =>
923 Effect.gen(function* () {
924 const common = {
925 threadId: THREAD_ID,
926 runId: null,
927 nodeId: null,
928 providerThreadId: null,
929 providerTurnId: null,
930 nativeItemRef: null,
931 parentItemId: null,
932 status: "completed",
933 title: null,
934 startedAt: "2026-06-20T00:00:00.000Z",
935 completedAt: "2026-06-20T00:00:00.000Z",
936 updatedAt: "2026-06-20T00:00:00.000Z",
937 };
938 const message = {
939 ...common,
940 id: "item:older-message",
941 ordinal: 1,
942 type: "assistant_message",
943 messageId: "message:older",
944 text: "COD4 result",
945 streaming: false,
946 };
947 const activity = {
948 ...common,
949 id: "item:older-command",
950 ordinal: 2,
951 type: "command_execution",
952 input: "pwd",
953 output: "",
954 exitCode: 0,
955 };
956 const projected = (item: typeof message | typeof activity, position: number) => ({
957 position,
958 visibility: "local",
959 sourceThreadId: THREAD_ID,
960 sourceItemId: item.id,
961 item,
962 });
963 const activityStarted = yield* Deferred.make<void>();
964 const releaseActivity = yield* Deferred.make<void>();
965 let activityRequests = 0;
966 const harness = yield* makeHarness({
967 historyHttpClient: HttpClient.make((request, url) =>
968 Effect.gen(function* () {
969 expect(url.searchParams.get("cursor")).toBe("progressive-cursor");
970 const conversationOnly = url.searchParams.get("view") === "conversation";
971 if (!conversationOnly && activityRequests++ === 0) {
972 yield* Deferred.succeed(activityStarted, undefined);
973 yield* Deferred.await(releaseActivity);
974 if (mode === "retry")
975 return HttpClientResponse.fromWeb(
976 request,
977 new Response("Activity unavailable", { status: 503 }),
978 );
979 }
980 return HttpClientResponse.fromWeb(
981 request,
982 Response.json({
983 snapshotSequence: 14,
984 items: conversationOnly
985 ? [projected(message, 0)]
986 : [projected(message, 0), projected(activity, 1)],
987 nextCursor: "next-cursor",
988 hasMoreHistory: true,
989 }),
990 );
991 }),
992 ),
993 });
994 yield* Queue.offer(harness.inputs, {
995 ...snapshot(BASE_PROJECTION, 14),
996 historyCursor: "progressive-cursor",
997 hasMoreHistory: true,
998 latestLocalTurnOrdinal: 47,
999 payloadBudgetExceeded: false,
1000 });
1001 yield* awaitThreadState(
1002 harness.observed,
1003 (state) => state.history.historyCursor === "progressive-cursor",
1004 );
1005 const loading = yield* harness.loadEarlier(message.messageId).pipe(Effect.forkScoped);
1006 yield* Deferred.await(activityStarted);
1007 const preview = yield* SubscriptionRef.get(harness.threadState);
1008 expect(preview.history).toMatchObject({
1009 loading: true,
1010 historyCursor: "progressive-cursor",
1011 expanded: true,
1012 });
1013 expect(
1014 Option.getOrThrow(preview.data).visibleTurnItems.map((row) => row.sourceItemId),
1015 ).toEqual([message.id]);
1016 if (mode === "replacement snapshot") {
1017 yield* Queue.offer(harness.inputs, snapshot(BASE_PROJECTION, 15));
1018 yield* awaitThreadState(
1019 harness.observed,
1020 (state) => state.history.historyCursor === null,
1021 );
1022 }
1023 yield* Deferred.succeed(releaseActivity, undefined);
1024 const result = yield* Fiber.join(loading);
1025 if (mode === "replacement snapshot") {
1026 expect(result).toEqual({ _tag: "noop" });
1027 const state = yield* SubscriptionRef.get(harness.threadState);
1028 expect(Option.getOrThrow(state.data).visibleTurnItems).toEqual(
1029 BASE_PROJECTION.visibleTurnItems,
1030 );
1031 expect(state.history.historyCursor).toBeNull();
1032 return;
1033 }
1034 if (mode === "retry") {
1035 expect(result._tag).toBe("error");
1036 // The ordinary load-earlier button must also finish a failed search load.
1037 expect(yield* harness.loadEarlier()).toEqual({ _tag: "loaded" });
1038 } else expect(result).toEqual({ _tag: "loaded" });
1039 const complete = yield* SubscriptionRef.get(harness.threadState);
1040 expect(complete.history).toMatchObject({
1041 loading: false,
1042 historyCursor: "next-cursor",
1043 error: null,
1044 latestLocalTurnOrdinal: 47,
1045 });
1046 expect(
1047 Option.getOrThrow(complete.data).visibleTurnItems.map((row) => row.sourceItemId),
1048 ).toEqual([message.id, activity.id]);
1049 expect(Option.getOrThrow(complete.data).turnItems).toHaveLength(2);
1050 }),
1051 );
1052
1053 it.effect("restores compact socket snapshot turnItems before reducing and caching", () =>
1054 Effect.gen(function* () {
1055 const at = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z");
1056 const command = (id: string, ordinal: number, threadId = THREAD_ID) =>
1057 ({
1058 id: TurnItemId.make(id),
1059 threadId,
1060 runId: null,
1061 nodeId: null,
1062 providerThreadId: null,
1063 providerTurnId: null,
1064 nativeItemRef: null,
1065 parentItemId: null,
1066 ordinal,
1067 status: "completed" as const,
1068 title: null,
1069 startedAt: at,
1070 completedAt: at,
1071 updatedAt: at,
1072 type: "command_execution" as const,
1073 input: id,
1074 output: id,
1075 exitCode: 0,
1076 }) satisfies OrchestrationV2TurnItem;
1077 const parentThread = ThreadId.make("thread:compact-parent");
1078 const inherited = command("parent-item", 1, parentThread);
1079 const local = [command("local-1", 1), command("local-2", 2)];
1080 const retainedRequest = {
1081 ...command("retained-request", 0),
1082 type: "run_interrupt_request" as const,
1083 message: "Stop",
1084 } as unknown as OrchestrationV2TurnItem;
1085 const visibleTurnItems = [
1086 {
1087 position: 0,
1088 visibility: "inherited" as const,
1089 sourceThreadId: parentThread,
1090 sourceItemId: inherited.id,
1091 item: inherited,
1092 },
1093 ...local.map((item, index) => ({
1094 position: index + 1,
1095 visibility: "local" as const,
1096 sourceThreadId: THREAD_ID,
1097 sourceItemId: item.id,
1098 item,
1099 })),
1100 ];
1101 const harness = yield* makeHarness();
1102 yield* Queue.offer(harness.inputs, {
1103 kind: "snapshot",
1104 snapshotSequence: 5,
1105 // Only the dependency that is not a visible local row is on the wire.
1106 projection: { ...BASE_PROJECTION, turnItems: [retainedRequest], visibleTurnItems },
1107 historyCursor: "compact-cursor",
1108 hasMoreHistory: true,
1109 latestLocalTurnOrdinal: 2,
1110 payloadBudgetExceeded: false,
1111 turnItemsOmitLocalVisible: true,
1112 });
1113
1114 const seeded = yield* awaitThreadState(
1115 harness.observed,
1116 (value) => value.history.historyCursor === "compact-cursor",
1117 );
1118 expect(Option.getOrThrow(seeded.data).turnItems).toEqual([...local, retainedRequest]);
1119 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true);
1120 expect(yield* Ref.get(harness.lastAcceptCompactTurnItems)).toBe(true);
1121
1122 // Live updates reduce against the restored list, not the compact one.
1123 const updated = { ...local[1]!, output: "updated", ordinal: 2 };
1124 yield* Queue.offer(harness.inputs, {
1125 kind: "event",
1126 sequence: 6,
1127 event: {
1128 id: EventId.make("event-compact-update"),
1129 type: "turn-item.updated",
1130 threadId: THREAD_ID,
1131 occurredAt: at,
1132 payload: updated,
1133 },
1134 });
1135 const live = yield* awaitThreadState(harness.observed, (value) =>
1136 Option.exists(value.data, (data) =>
1137 data.turnItems.some(
1138 (item) => item.type === "command_execution" && item.output === "updated",
1139 ),
1140 ),
1141 );
1142 expect(Option.getOrThrow(live.data).turnItems.map((item) => String(item.id))).toEqual([
1143 "local-1",
1144 "local-2",
1145 "retained-request",
1146 ]);
1147
1148 yield* TestClock.adjust("500 millis");
1149 yield* Effect.yieldNow;
1150 const saved = (yield* Ref.get(harness.savedThreads)).at(-1);
1151 expect(saved?.projection.turnItems.map((item) => String(item.id))).toEqual([
1152 "local-1",
1153 "local-2",
1154 "retained-request",
1155 ]);
1156 expect(saved && "turnItemsOmitLocalVisible" in saved).toBe(false);
1157 }),
1158 );
1159
1160 it.effect("seeds the thread from the HTTP snapshot and resumes live events", () =>
1161 Effect.gen(function* () {
1162 const httpProjection: OrchestrationV2ThreadProjection = {
1163 ...BASE_PROJECTION,
1164 thread: { ...BASE_PROJECTION.thread, title: "HTTP title" },
1165 };
1166 const harness = yield* makeHarness({
1167 httpSnapshot: {
1168 _tag: "present",
1169 snapshot: { snapshotSequence: 1, projection: httpProjection },
1170 },
1171 });
1172 // No socket snapshot is pushed; only a live event arrives over the socket.
1173 // It can only be applied if the HTTP snapshot already seeded the thread.
1174 yield* Queue.offer(harness.inputs, titleUpdated("Live title", 2));
1175
1176 const state = yield* awaitThreadState(
1177 harness.observed,
1178 (value) =>
1179 value.status === "live" &&
1180 Option.isSome(value.data) &&
1181 value.data.value.thread.title === "Live title",
1182 );
1183
1184 expect(Option.getOrThrow(state.data).thread.title).toBe("Live title");
1185 // Cold cache: the full snapshot was loaded over HTTP and the socket
1186 // resumed from that snapshot's sequence.
1187 expect(yield* Ref.get(harness.loaderCalls)).toBeGreaterThanOrEqual(1);
1188 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(1);
1189 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true);
1190 }),
1191 );
1192
1193 it.effect("installs bounded snapshot history meta and resumes via afterSequence", () =>
1194 Effect.gen(function* () {
1195 const httpProjection: OrchestrationV2ThreadProjection = {
1196 ...BASE_PROJECTION,
1197 thread: { ...BASE_PROJECTION.thread, title: "Bounded title" },
1198 };
1199 const harness = yield* makeHarness({
1200 httpSnapshot: {
1201 _tag: "present",
1202 snapshot: { snapshotSequence: 11, projection: httpProjection },
1203 history: {
1204 historyCursor: "opaque-cursor",
1205 hasMoreHistory: true,
1206 },
1207 },
1208 });
1209 yield* Queue.offer(harness.inputs, titleUpdated("Live after bounded", 12));
1210
1211 const state = yield* awaitThreadState(
1212 harness.observed,
1213 (value) =>
1214 value.status === "live" &&
1215 Option.isSome(value.data) &&
1216 value.data.value.thread.title === "Live after bounded" &&
1217 value.history.hasMoreHistory,
1218 );
1219
1220 expect(state.history).toMatchObject({
1221 historyCursor: "opaque-cursor",
1222 hasMoreHistory: true,
1223 loading: false,
1224 error: null,
1225 expanded: false,
1226 });
1227 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(11);
1228 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true);
1229 // Bounded HTTP success must not require a socket snapshot frame.
1230 expect(Option.getOrThrow(state.data).thread.title).toBe("Live after bounded");
1231 }),
1232 );
1233
1234 it.effect("persists progressive history meta with a settled bounded snapshot", () =>
1235 Effect.gen(function* () {
1236 const httpProjection: OrchestrationV2ThreadProjection = {
1237 ...BASE_PROJECTION,
1238 thread: { ...BASE_PROJECTION.thread, title: "Bounded cache title" },
1239 };
1240 const harness = yield* makeHarness({
1241 httpSnapshot: {
1242 _tag: "present",
1243 snapshot: { snapshotSequence: 4, projection: httpProjection },
1244 history: {
1245 historyCursor: "cursor-oldest",
1246 hasMoreHistory: true,
1247 },
1248 },
1249 });
1250 // Bounded install alone (settled) must enqueue progressive meta with the
1251 // projection. Never persist the partial window as a complete full snapshot.
1252 yield* awaitThreadState(
1253 harness.observed,
1254 (value) =>
1255 Option.isSome(value.data) &&
1256 value.data.value.thread.title === "Bounded cache title" &&
1257 value.history.historyCursor === "cursor-oldest",
1258 );
1259 yield* TestClock.adjust("500 millis");
1260 yield* Effect.yieldNow;
1261
1262 const savedAll = yield* Ref.get(harness.savedThreads);
1263 expect(savedAll.length).toBeGreaterThanOrEqual(1);
1264 const first = savedAll[0];
1265 expect(first?.snapshotSequence).toBe(4);
1266 expect(first?.projection.thread.title).toBe("Bounded cache title");
1267 expect(first?.historyCursor).toBe("cursor-oldest");
1268 expect(first?.hasMoreHistory).toBe(true);
1269 // No earlier complete-looking entry without progressive meta.
1270 expect(
1271 savedAll.some(
1272 (entry) =>
1273 entry.projection.thread.title === "Bounded cache title" &&
1274 entry.historyCursor === undefined &&
1275 entry.hasMoreHistory === undefined,
1276 ),
1277 ).toBe(false);
1278
1279 // A later live update still carries the progressive cursor.
1280 yield* Queue.offer(harness.inputs, titleUpdated("Settled bounded", 5));
1281 yield* awaitThreadState(
1282 harness.observed,
1283 (value) =>
1284 value.status === "live" &&
1285 Option.isSome(value.data) &&
1286 value.data.value.thread.title === "Settled bounded",
1287 );
1288 yield* TestClock.adjust("10 seconds");
1289 yield* Effect.yieldNow;
1290
1291 const saved = (yield* Ref.get(harness.savedThreads)).at(-1);
1292 expect(saved?.snapshotSequence).toBe(5);
1293 expect(saved?.projection.thread.title).toBe("Settled bounded");
1294 expect(saved?.historyCursor).toBe("cursor-oldest");
1295 expect(saved?.hasMoreHistory).toBe(true);
1296 }),
1297 );
1298
1299 it.effect("warm resume restores progressive history meta and skips HTTP", () =>
1300 Effect.gen(function* () {
1301 const harness = yield* makeHarness({
1302 cached: {
1303 ...BASE_PROJECTION,
1304 thread: { ...BASE_PROJECTION.thread, title: "Cached bounded" },
1305 },
1306 cachedHistory: {
1307 historyCursor: "warm-cursor",
1308 hasMoreHistory: true,
1309 },
1310 });
1311
1312 // Apply a live event so the subscription is known to have resumed from cache.
1313 yield* Queue.offer(
1314 harness.inputs,
1315 titleUpdated("Cached bounded live", CACHED_SNAPSHOT_SEQUENCE + 1),
1316 );
1317 const state = yield* awaitThreadState(
1318 harness.observed,
1319 (value) =>
1320 value.status === "live" &&
1321 Option.isSome(value.data) &&
1322 value.data.value.thread.title === "Cached bounded live" &&
1323 value.history.hasMoreHistory,
1324 );
1325
1326 expect(state.history).toMatchObject({
1327 historyCursor: "warm-cursor",
1328 hasMoreHistory: true,
1329 expanded: false,
1330 });
1331 // Warm progressive cache must not re-download; resume via afterSequence.
1332 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
1333 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);
1334 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true);
1335 }),
1336 );
1337
1338 it.effect("legacy warm cache without history meta stays complete (no false load-earlier)", () =>
1339 Effect.gen(function* () {
1340 const harness = yield* makeHarness({ cached: BASE_PROJECTION });
1341 const state = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.data));
1342 expect(state.history).toMatchObject({
1343 historyCursor: null,
1344 hasMoreHistory: false,
1345 expanded: false,
1346 });
1347 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
1348 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBeUndefined();
1349 }),
1350 );
1351
1352 it.effect.each(["disk", "retained"] as const)(
1353 "retains paging support through a complete bounded %s cache",
1354 (cacheKind) =>
1355 Effect.gen(function* () {
1356 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
1357 snapshot: undefined,
1358 owner: undefined,
1359 };
1360 const cold = yield* Effect.scoped(
1361 Effect.gen(function* () {
1362 const harness = yield* makeHarness({
1363 resumeCache,
1364 httpSnapshot: {
1365 _tag: "present",
1366 snapshot: { snapshotSequence: 4, projection: BASE_PROJECTION },
1367 history: {
1368 historyCursor: null,
1369 hasMoreHistory: false,
1370 latestLocalTurnOrdinal: null,
1371 },
1372 },
1373 });
1374 yield* Queue.offer(harness.inputs, titleUpdated("Complete bounded", 5));
1375 yield* awaitThreadState(
1376 harness.observed,
1377 (value) =>
1378 value.status === "live" &&
1379 Option.isSome(value.data) &&
1380 value.data.value.thread.title === "Complete bounded",
1381 );
1382 return harness;
1383 }),
1384 );
1385 const saved = (yield* Ref.get(cold.savedThreads)).at(-1);
1386 expect(saved).toMatchObject({
1387 snapshotSequence: 5,
1388 historyCursor: null,
1389 hasMoreHistory: false,
1390 latestLocalTurnOrdinal: null,
1391 });
1392
1393 const warm = yield* makeHarness(
1394 cacheKind === "retained"
1395 ? { resumeCache }
1396 : { loadCached: Effect.succeed(Option.some(saved!)) },
1397 );
1398 // The thread may have grown past the replay budget while closed. Its
1399 // warm subscription must still allow the server to bound that fallback.
1400 yield* Queue.offer(warm.inputs, titleUpdated("Grown while closed", 200));
1401 yield* awaitThreadState(
1402 warm.observed,
1403 (value) =>
1404 value.status === "live" &&
1405 Option.isSome(value.data) &&
1406 value.data.value.thread.title === "Grown while closed",
1407 );
1408 expect(yield* Ref.get(warm.loaderCalls)).toBe(0);
1409 expect(yield* Ref.get(warm.lastSubscribeAfterSequence)).toBe(5);
1410 expect(yield* Ref.get(warm.lastAcceptBoundedSnapshot)).toBe(true);
1411 }),
1412 );
1413
1414 it.effect.each(["no-http", "no-controller"] as const)(
1415 "does not negotiate bounded fallbacks with %s",
1416 (historyPaging) =>
1417 Effect.gen(function* () {
1418 for (const source of ["cache", "http"] as const) {
1419 const history = {
1420 historyCursor: "known-paging-cursor",
1421 hasMoreHistory: true,
1422 };
1423 const harness = yield* makeHarness({
1424 historyPaging,
1425 ...(source === "cache"
1426 ? { cached: BASE_PROJECTION, cachedHistory: history }
1427 : {
1428 httpSnapshot: {
1429 _tag: "present" as const,
1430 snapshot: { snapshotSequence: 4, projection: BASE_PROJECTION },
1431 history,
1432 },
1433 }),
1434 });
1435 yield* Queue.offer(harness.inputs, titleUpdated("Paging unavailable", 10));
1436 yield* awaitThreadState(
1437 harness.observed,
1438 (value) =>
1439 value.status === "live" &&
1440 Option.isSome(value.data) &&
1441 value.data.value.thread.title === "Paging unavailable",
1442 );
1443 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBeUndefined();
1444 }
1445 }),
1446 );
1447
1448 it.effect("socket snapshot clears progressive history meta left from a bounded window", () =>
1449 Effect.gen(function* () {
1450 const harness = yield* makeHarness({
1451 cached: {
1452 ...BASE_PROJECTION,
1453 thread: { ...BASE_PROJECTION.thread, title: "Warm progressive" },
1454 },
1455 cachedHistory: {
1456 historyCursor: "stale-cursor",
1457 hasMoreHistory: true,
1458 },
1459 });
1460
1461 yield* Queue.offer(
1462 harness.inputs,
1463 snapshot(
1464 {
1465 ...BASE_PROJECTION,
1466 thread: { ...BASE_PROJECTION.thread, title: "Full socket snapshot" },
1467 },
1468 CACHED_SNAPSHOT_SEQUENCE + 1,
1469 ),
1470 );
1471
1472 const state = yield* awaitThreadState(
1473 harness.observed,
1474 (value) =>
1475 value.status === "live" &&
1476 Option.isSome(value.data) &&
1477 value.data.value.thread.title === "Full socket snapshot" &&
1478 value.history.historyCursor === null &&
1479 value.history.hasMoreHistory === false,
1480 );
1481
1482 expect(state.history).toMatchObject({
1483 historyCursor: null,
1484 hasMoreHistory: false,
1485 expanded: false,
1486 loading: false,
1487 error: null,
1488 });
1489
1490 // Settled full snapshot persistence must not keep the stale cursor.
1491 yield* TestClock.adjust("500 millis");
1492 yield* Effect.yieldNow;
1493 const saved = (yield* Ref.get(harness.savedThreads)).at(-1);
1494 expect(saved?.projection.thread.title).toBe("Full socket snapshot");
1495 expect(saved?.historyCursor).toBeNull();
1496 expect(saved?.hasMoreHistory).toBe(false);
1497 }),
1498 );
1499
1500 it.effect("bounded socket fallback replaces progressive history meta during resume", () =>
1501 Effect.gen(function* () {
1502 const harness = yield* makeHarness({
1503 cached: {
1504 ...BASE_PROJECTION,
1505 thread: { ...BASE_PROJECTION.thread, title: "Warm progressive" },
1506 },
1507 cachedHistory: {
1508 historyCursor: "stale-cursor",
1509 hasMoreHistory: true,
1510 latestLocalTurnOrdinal: 3,
1511 },
1512 });
1513
1514 yield* Queue.offer(harness.inputs, {
1515 ...snapshot(
1516 {
1517 ...BASE_PROJECTION,
1518 thread: { ...BASE_PROJECTION.thread, title: "Bounded resume fallback" },
1519 },
1520 CACHED_SNAPSHOT_SEQUENCE + 1,
1521 ),
1522 historyCursor: "replacement-cursor",
1523 hasMoreHistory: true,
1524 latestLocalTurnOrdinal: 9,
1525 payloadBudgetExceeded: false,
1526 });
1527
1528 const state = yield* awaitThreadState(
1529 harness.observed,
1530 (value) =>
1531 value.status === "live" &&
1532 Option.getOrNull(value.data)?.thread.title === "Bounded resume fallback" &&
1533 value.history.historyCursor === "replacement-cursor",
1534 );
1535
1536 expect(state.history).toMatchObject({
1537 historyCursor: "replacement-cursor",
1538 hasMoreHistory: true,
1539 latestLocalTurnOrdinal: 9,
1540 expanded: false,
1541 loading: false,
1542 error: null,
1543 });
1544 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);
1545 }),
1546 );
1547
1548 it.effect("live events preserve progressive history meta under atomic setThread", () =>
1549 Effect.gen(function* () {
1550 const harness = yield* makeHarness({
1551 httpSnapshot: {
1552 _tag: "present",
1553 snapshot: {
1554 snapshotSequence: 8,
1555 projection: {
1556 ...BASE_PROJECTION,
1557 thread: { ...BASE_PROJECTION.thread, title: "Bounded seed" },
1558 },
1559 },
1560 history: {
1561 historyCursor: "keep-me",
1562 hasMoreHistory: true,
1563 },
1564 },
1565 });
1566
1567 yield* Queue.offer(harness.inputs, titleUpdated("Live preserves meta", 9));
1568 const state = yield* awaitThreadState(
1569 harness.observed,
1570 (value) =>
1571 value.status === "live" &&
1572 Option.isSome(value.data) &&
1573 value.data.value.thread.title === "Live preserves meta",
1574 );
1575
1576 expect(state.history).toMatchObject({
1577 historyCursor: "keep-me",
1578 hasMoreHistory: true,
1579 expanded: false,
1580 });
1581 }),
1582 );
1583
1584 it.effect("bounded HTTP install sets projection and progressive meta atomically", () =>
1585 Effect.gen(function* () {
1586 // One setThread with explicit history (not applyItem reset + later meta).
1587 const harness = yield* makeHarness({
1588 httpSnapshot: {
1589 _tag: "present",
1590 snapshot: {
1591 snapshotSequence: 2,
1592 projection: {
1593 ...BASE_PROJECTION,
1594 thread: { ...BASE_PROJECTION.thread, title: "Bounded atomic install" },
1595 },
1596 },
1597 history: {
1598 historyCursor: "post-install-cursor",
1599 hasMoreHistory: true,
1600 },
1601 },
1602 });
1603
1604 const state = yield* awaitThreadState(
1605 harness.observed,
1606 (value) =>
1607 Option.isSome(value.data) &&
1608 value.data.value.thread.title === "Bounded atomic install" &&
1609 value.history.historyCursor === "post-install-cursor" &&
1610 value.history.hasMoreHistory === true,
1611 );
1612
1613 expect(state.history).toMatchObject({
1614 historyCursor: "post-install-cursor",
1615 hasMoreHistory: true,
1616 loading: false,
1617 error: null,
1618 expanded: false,
1619 });
1620 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(2);
1621 }),
1622 );
1623
1624 it.effect("dropped partial-timeline turn item is a true applyItem no-op", () =>
1625 Effect.gen(function* () {
1626 const recent = {
1627 id: TurnItemId.make("item-window"),
1628 threadId: THREAD_ID,
1629 runId: null,
1630 nodeId: null,
1631 providerThreadId: null,
1632 providerTurnId: null,
1633 nativeItemRef: null,
1634 parentItemId: null,
1635 ordinal: 10,
1636 status: "completed" as const,
1637 title: null,
1638 startedAt: DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"),
1639 completedAt: DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"),
1640 updatedAt: DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"),
1641 type: "command_execution" as const,
1642 input: "pwd",
1643 output: "ok",
1644 exitCode: 0,
1645 } satisfies OrchestrationV2TurnItem;
1646 const recentRow = {
1647 position: 0,
1648 visibility: "local" as const,
1649 sourceThreadId: THREAD_ID,
1650 sourceItemId: recent.id,
1651 item: recent,
1652 };
1653 const boundedProjection: OrchestrationV2ThreadProjection = {
1654 ...BASE_PROJECTION,
1655 thread: { ...BASE_PROJECTION.thread, title: "Partial noop" },
1656 turnItems: [recent],
1657 visibleTurnItems: [recentRow],
1658 };
1659 const harness = yield* makeHarness({
1660 httpSnapshot: {
1661 _tag: "present",
1662 snapshot: {
1663 snapshotSequence: 5,
1664 projection: boundedProjection,
1665 latestLocalTurnOrdinal: 10,
1666 },
1667 history: {
1668 historyCursor: "partial-cursor",
1669 hasMoreHistory: true,
1670 latestLocalTurnOrdinal: 10,
1671 },
1672 },
1673 });
1674
1675 const seeded = yield* awaitThreadState(
1676 harness.observed,
1677 (value) =>
1678 value.status === "live" &&
1679 Option.isSome(value.data) &&
1680 value.data.value.thread.title === "Partial noop" &&
1681 value.history.historyCursor === "partial-cursor",
1682 );
1683 const seededProjection = Option.getOrThrow(seeded.data);
1684 yield* TestClock.adjust("500 millis");
1685 yield* Effect.yieldNow;
1686 const savedBefore = (yield* Ref.get(harness.savedThreads)).length;
1687
1688 const older = {
1689 ...recent,
1690 id: TurnItemId.make("item-old-outside"),
1691 ordinal: 3,
1692 output: "must-not-append",
1693 } satisfies OrchestrationV2TurnItem;
1694 yield* Queue.offer(harness.inputs, {
1695 kind: "event",
1696 sequence: 6,
1697 event: {
1698 id: EventId.make("event-old-partial"),
1699 type: "turn-item.updated",
1700 threadId: THREAD_ID,
1701 occurredAt: DateTime.makeUnsafe("2026-06-20T01:00:00.000Z"),
1702 payload: older,
1703 },
1704 });
1705
1706 // Allow the event to be processed without requiring a state transition.
1707 for (let attempt = 0; attempt < 30; attempt += 1) {
1708 yield* Effect.yieldNow;
1709 }
1710 // Drive a later unrelated title update so we know the stream continued.
1711 yield* Queue.offer(harness.inputs, titleUpdated("After dropped event", 7));
1712 const after = yield* awaitThreadState(
1713 harness.observed,
1714 (value) =>
1715 value.status === "live" &&
1716 Option.isSome(value.data) &&
1717 value.data.value.thread.title === "After dropped event",
1718 );
1719
1720 // The dropped event must not have appended the old item.
1721 expect(Option.getOrThrow(after.data).turnItems.map((item) => String(item.id))).toEqual([
1722 String(recent.id),
1723 ]);
1724 // Watermark unchanged (only advanced on successful newer turn-item applies).
1725 expect(after.history.latestLocalTurnOrdinal).toBe(10);
1726 // No persistence enqueue from the dropped event itself.
1727 const savedAfterDrop = (yield* Ref.get(harness.savedThreads)).length;
1728 expect(savedAfterDrop).toBe(savedBefore);
1729 // Seeded projection reference path: event path kept partial meta intact.
1730 expect(after.history.historyCursor).toBe("partial-cursor");
1731 expect(after.history.hasMoreHistory).toBe(true);
1732 // Title event applied; drop itself did not clear progressive meta.
1733 expect(seededProjection.turnItems.map((item) => String(item.id))).toEqual([
1734 String(recent.id),
1735 ]);
1736 }),
1737 );
1738
1739 it.effect("installs and advances latestLocalTurnOrdinal for partial progressive windows", () =>
1740 Effect.gen(function* () {
1741 const harness = yield* makeHarness({
1742 httpSnapshot: {
1743 _tag: "present",
1744 snapshot: {
1745 snapshotSequence: 3,
1746 projection: {
1747 ...BASE_PROJECTION,
1748 thread: { ...BASE_PROJECTION.thread, title: "Watermark seed" },
1749 },
1750 latestLocalTurnOrdinal: 15,
1751 },
1752 history: {
1753 historyCursor: "wm-cursor",
1754 hasMoreHistory: true,
1755 latestLocalTurnOrdinal: 15,
1756 },
1757 },
1758 });
1759
1760 const seeded = yield* awaitThreadState(
1761 harness.observed,
1762 (value) =>
1763 Option.isSome(value.data) &&
1764 value.data.value.thread.title === "Watermark seed" &&
1765 value.history.latestLocalTurnOrdinal === 15,
1766 );
1767 expect(seeded.history.latestLocalTurnOrdinal).toBe(15);
1768
1769 yield* TestClock.adjust("500 millis");
1770 yield* Effect.yieldNow;
1771 const savedSeed = (yield* Ref.get(harness.savedThreads)).at(-1);
1772 expect(savedSeed?.latestLocalTurnOrdinal).toBe(15);
1773
1774 const newer = {
1775 id: TurnItemId.make("item-newer-live"),
1776 threadId: THREAD_ID,
1777 runId: null,
1778 nodeId: null,
1779 providerThreadId: null,
1780 providerTurnId: null,
1781 nativeItemRef: null,
1782 parentItemId: null,
1783 ordinal: 22,
1784 status: "completed" as const,
1785 title: null,
1786 startedAt: DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"),
1787 completedAt: DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"),
1788 updatedAt: DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"),
1789 type: "command_execution" as const,
1790 input: "echo newer",
1791 output: "newer",
1792 exitCode: 0,
1793 } satisfies OrchestrationV2TurnItem;
1794
1795 yield* Queue.offer(harness.inputs, {
1796 kind: "event",
1797 sequence: 4,
1798 event: {
1799 id: EventId.make("event-newer-item"),
1800 type: "turn-item.updated",
1801 threadId: THREAD_ID,
1802 occurredAt: DateTime.makeUnsafe("2026-06-20T01:00:00.000Z"),
1803 payload: newer,
1804 },
1805 });
1806
1807 const advanced = yield* awaitThreadState(
1808 harness.observed,
1809 (value) =>
1810 value.status === "live" &&
1811 Option.isSome(value.data) &&
1812 value.history.latestLocalTurnOrdinal === 22 &&
1813 value.data.value.turnItems.some((item) => String(item.id) === String(newer.id)),
1814 );
1815 expect(advanced.history.latestLocalTurnOrdinal).toBe(22);
1816 expect(advanced.history.historyCursor).toBe("wm-cursor");
1817 }),
1818 );
1819
1820 it.effect("warm resume restores latestLocalTurnOrdinal from progressive cache", () =>
1821 Effect.gen(function* () {
1822 const harness = yield* makeHarness({
1823 cached: {
1824 ...BASE_PROJECTION,
1825 thread: { ...BASE_PROJECTION.thread, title: "Warm watermark" },
1826 },
1827 cachedHistory: {
1828 historyCursor: "warm-wm-cursor",
1829 hasMoreHistory: true,
1830 latestLocalTurnOrdinal: 33,
1831 },
1832 });
1833
1834 const state = yield* awaitThreadState(
1835 harness.observed,
1836 (value) =>
1837 Option.isSome(value.data) &&
1838 value.data.value.thread.title === "Warm watermark" &&
1839 value.history.latestLocalTurnOrdinal === 33,
1840 );
1841 expect(state.history).toMatchObject({
1842 historyCursor: "warm-wm-cursor",
1843 hasMoreHistory: true,
1844 latestLocalTurnOrdinal: 33,
1845 expanded: false,
1846 });
1847 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
1848 }),
1849 );
1850
1851 it.effect("marks a cold definitive HTTP miss deleted without socket subscribe or retry", () =>
1852 Effect.gen(function* () {
1853 const harness = yield* makeHarness({
1854 httpSnapshot: { _tag: "missing" },
1855 });
1856
1857 const state = yield* awaitThreadState(
1858 harness.observed,
1859 (value) => value.status === "deleted",
1860 );
1861
1862 expect(Option.isNone(state.data)).toBe(true);
1863 expect(Option.isNone(state.error)).toBe(true);
1864 expect(yield* Ref.get(harness.loaderCalls)).toBeGreaterThanOrEqual(1);
1865 expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]);
1866 expect(yield* Ref.get(harness.subscriptionCount)).toBe(0);
1867
1868 // A definitive miss must not schedule the expected-failure retry path.
1869 yield* TestClock.adjust("1 second");
1870 for (let attempt = 0; attempt < 20; attempt += 1) {
1871 yield* Effect.yieldNow;
1872 }
1873 yield* harness.replaceSession;
1874 yield* Queue.offer(harness.wakeups, "application-active");
1875 for (let attempt = 0; attempt < 20; attempt += 1) {
1876 yield* Effect.yieldNow;
1877 }
1878 expect(yield* Ref.get(harness.subscriptionCount)).toBe(0);
1879 expect(yield* Ref.get(harness.retryCount)).toBe(0);
1880 expect(yield* Ref.get(harness.loaderCalls)).toBe(1);
1881 expect((yield* Ref.get(harness.latest)).status).toBe("deleted");
1882 }),
1883 );
1884
1885 it.effect("falls back to the socket when the HTTP snapshot is only unavailable", () =>
1886 Effect.gen(function* () {
1887 const harness = yield* makeHarness({
1888 httpSnapshot: { _tag: "unavailable" },
1889 });
1890
1891 yield* Queue.offer(
1892 harness.inputs,
1893 snapshot({
1894 ...BASE_PROJECTION,
1895 thread: { ...BASE_PROJECTION.thread, title: "Socket title" },
1896 }),
1897 );
1898
1899 const state = yield* awaitThreadState(
1900 harness.observed,
1901 (value) =>
1902 value.status === "live" &&
1903 Option.isSome(value.data) &&
1904 value.data.value.thread.title === "Socket title",
1905 );
1906
1907 expect(Option.getOrThrow(state.data).thread.title).toBe("Socket title");
1908 expect(yield* Ref.get(harness.loaderCalls)).toBeGreaterThanOrEqual(1);
1909 expect(yield* Ref.get(harness.subscriptionCount)).toBe(1);
1910 expect(yield* Ref.get(harness.removedThreads)).toEqual([]);
1911 }),
1912 );
1913
1914 it.effect("ignores replayed thread events at or below the snapshot sequence", () =>
1915 Effect.gen(function* () {
1916 const harness = yield* makeHarness({ cached: BASE_PROJECTION });
1917 yield* Queue.offer(harness.inputs, snapshot(BASE_PROJECTION));
1918 yield* Queue.offer(harness.inputs, titleUpdated("Replayed title", 1));
1919 yield* Queue.offer(harness.inputs, titleUpdated("Live title", 2));
1920
1921 const state = yield* awaitThreadState(
1922 harness.observed,
1923 (value) =>
1924 value.status === "live" &&
1925 Option.isSome(value.data) &&
1926 value.data.value.thread.title === "Live title",
1927 );
1928
1929 expect(Option.getOrThrow(state.data).thread.title).toBe("Live title");
1930 }),
1931 );
1932
1933 it.effect("does not recreate a deleted cache from a queued save or finalizer", () =>
1934 Effect.gen(function* () {
1935 const scope = yield* Scope.make();
1936 const harness = yield* makeHarness().pipe(Effect.provideService(Scope.Scope, scope));
1937 yield* Queue.offer(harness.inputs, snapshot(BASE_PROJECTION));
1938 yield* awaitThreadState(harness.observed, (value) => value.status === "live");
1939 yield* TestClock.adjust("500 millis");
1940 expect(yield* Ref.get(harness.savedThreads)).toHaveLength(1);
1941
1942 yield* Queue.offer(harness.inputs, titleUpdated("Queued before deletion"));
1943 yield* awaitThreadState(
1944 harness.observed,
1945 (value) =>
1946 Option.isSome(value.data) && value.data.value.thread.title === "Queued before deletion",
1947 );
1948 yield* Queue.offer(harness.inputs, deleted());
1949 const state = yield* awaitThreadState(
1950 harness.observed,
1951 (value) => value.status === "deleted",
1952 );
1953 expect(Option.isNone(state.data)).toBe(true);
1954 expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]);
1955
1956 yield* TestClock.adjust("10 seconds");
1957 expect(yield* Ref.get(harness.savedThreads)).toHaveLength(1);
1958 yield* Scope.close(scope, Exit.void);
1959 expect(yield* Ref.get(harness.savedThreads)).toHaveLength(1);
1960 }),
1961 );
1962
1963 it.effect("preserves data after a domain failure and resumes on a replacement session", () =>
1964 Effect.gen(function* () {
1965 const harness = yield* makeHarness({ cached: BASE_PROJECTION });
1966 yield* Queue.offer(harness.inputs, snapshot(BASE_PROJECTION));
1967 yield* Queue.offer(harness.inputs, new Error("stream failed"));
1968
1969 const state = yield* awaitThreadState(harness.observed, (value) =>
1970 Option.isSome(value.error),
1971 );
1972
1973 expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION);
1974 expect(Option.getOrThrow(state.error)).toBe("stream failed");
1975 expect(yield* Ref.get(harness.retryCount)).toBe(0);
1976
1977 yield* harness.replaceSession;
1978 for (let attempt = 0; attempt < 100; attempt += 1) {
1979 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) {
1980 break;
1981 }
1982 yield* Effect.yieldNow;
1983 }
1984 yield* Queue.offer(
1985 harness.inputs,
1986 snapshot({
1987 ...BASE_PROJECTION,
1988 thread: { ...BASE_PROJECTION.thread, title: "Recovered thread" },
1989 }),
1990 );
1991 const recovered = yield* awaitThreadState(
1992 harness.observed,
1993 (value) =>
1994 value.status === "live" &&
1995 Option.isSome(value.data) &&
1996 value.data.value.thread.title === "Recovered thread",
1997 );
1998
1999 expect(Option.isNone(recovered.error)).toBe(true);
2000 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
2001 }),
2002 );
2003
2004 it.effect("recovers from a transient domain failure without replacing the session", () =>
2005 Effect.gen(function* () {
2006 const harness = yield* makeHarness();
2007 yield* Queue.offer(harness.inputs, new Error("thread not found yet"));
2008
2009 const failed = yield* awaitThreadState(harness.observed, (value) =>
2010 Option.isSome(value.error),
2011 );
2012 expect(Option.getOrThrow(failed.error)).toBe("thread not found yet");
2013 expect(yield* Ref.get(harness.subscriptionCount)).toBe(1);
2014
2015 yield* TestClock.adjust("250 millis");
2016 for (let attempt = 0; attempt < 100; attempt += 1) {
2017 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) {
2018 break;
2019 }
2020 yield* Effect.yieldNow;
2021 }
2022 yield* Queue.offer(
2023 harness.inputs,
2024 snapshot({
2025 ...BASE_PROJECTION,
2026 thread: { ...BASE_PROJECTION.thread, title: "Materialized thread" },
2027 }),
2028 );
2029
2030 const recovered = yield* awaitThreadState(
2031 harness.observed,
2032 (value) =>
2033 value.status === "live" &&
2034 Option.isSome(value.data) &&
2035 value.data.value.thread.title === "Materialized thread",
2036 );
2037
2038 expect(Option.isNone(recovered.error)).toBe(true);
2039 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
2040 expect(yield* Ref.get(harness.retryCount)).toBe(0);
2041 }),
2042 );
2043
2044 it.effect("does not overwrite a live snapshot when the supervisor becomes ready", () =>
2045 Effect.gen(function* () {
2046 const harness = yield* makeHarness({ cached: BASE_PROJECTION });
2047 yield* SubscriptionRef.set(harness.supervisorState, {
2048 desired: true,
2049 network: "online",
2050 phase: "connecting",
2051 stage: "synchronizing",
2052 attempt: 1,
2053 generation: 0,
2054 lastFailure: null,
2055 retryAt: null,
2056 });
2057 yield* Queue.offer(harness.inputs, snapshot(BASE_PROJECTION));
2058 yield* awaitThreadState(harness.observed, (value) => value.status === "live");
2059
2060 yield* SubscriptionRef.set(harness.supervisorState, {
2061 desired: true,
2062 network: "online",
2063 phase: "connected",
2064 stage: null,
2065 attempt: 1,
2066 generation: 1,
2067 lastFailure: null,
2068 retryAt: null,
2069 });
2070 for (let index = 0; index < 10; index += 1) {
2071 yield* Effect.yieldNow;
2072 }
2073
2074 expect((yield* Ref.get(harness.latest)).status).toBe("live");
2075 }),
2076 );
2077
2078 it.effect("keeps replayed updates synchronizing until the completion marker arrives", () =>
2079 Effect.gen(function* () {
2080 const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true });
2081 yield* awaitThreadState(
2082 harness.observed,
2083 (value) => value.status === "synchronizing" && Option.isSome(value.data),
2084 );
2085 expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true);
2086
2087 yield* Queue.offer(
2088 harness.inputs,
2089 titleUpdated("Caught-up title", CACHED_SNAPSHOT_SEQUENCE + 1),
2090 );
2091 const catchingUp = yield* awaitThreadState(
2092 harness.observed,
2093 (value) =>
2094 value.status === "synchronizing" &&
2095 Option.isSome(value.data) &&
2096 value.data.value.thread.title === "Caught-up title",
2097 );
2098 expect(catchingUp.status).toBe("synchronizing");
2099
2100 yield* Queue.offer(harness.inputs, synchronized());
2101 const live = yield* awaitThreadState(
2102 harness.observed,
2103 (value) => value.status === "live" && Option.isSome(value.data),
2104 );
2105 expect(Option.getOrThrow(live.data).thread.title).toBe("Caught-up title");
2106 }),
2107 );
2108
2109 it.effect("skips an unknown event type and resumes after it", () =>
2110 Effect.gen(function* () {
2111 const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true });
2112 const unknown = (sequence: number): OrchestrationV2ThreadStreamItem => ({
2113 kind: "unknown-event",
2114 sequence,
2115 eventType: "run.from-a-future-server",
2116 });
2117 const occurredAt = DateTime.makeUnsafe("2026-06-20T01:00:00.000Z");
2118 yield* Queue.offerAll(harness.inputs, [
2119 {
2120 kind: "event",
2121 sequence: CACHED_SNAPSHOT_SEQUENCE + 1,
2122 event: {
2123 id: EventId.make("event-message"),
2124 type: "message.updated",
2125 threadId: THREAD_ID,
2126 occurredAt,
2127 payload: {
2128 id: MessageId.make("message-before"),
2129 threadId: THREAD_ID,
2130 runId: null,
2131 nodeId: null,
2132 role: "assistant",
2133 text: "Before",
2134 streaming: false,
2135 attachments: [],
2136 createdBy: "agent",
2137 creationSource: "provider",
2138 createdAt: occurredAt,
2139 updatedAt: occurredAt,
2140 },
2141 },
2142 },
2143 unknown(CACHED_SNAPSHOT_SEQUENCE + 2),
2144 titleUpdated("After", CACHED_SNAPSHOT_SEQUENCE + 3),
2145 // A trailing unknown event must still advance the resume cursor.
2146 unknown(CACHED_SNAPSHOT_SEQUENCE + 4),
2147 synchronized(),
2148 ]);
2149 const live = yield* awaitThreadState(
2150 harness.observed,
2151 (value) =>
2152 value.status === "live" &&
2153 Option.isSome(value.data) &&
2154 value.data.value.thread.title === "After",
2155 );
2156 expect(Option.isNone(live.error)).toBe(true);
2157 expect(Option.getOrThrow(live.data).messages.map((message) => message.text)).toEqual([
2158 "Before",
2159 ]);
2160
2161 yield* harness.replaceSession;
2162 for (let attempt = 0; attempt < 100; attempt += 1) {
2163 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break;
2164 yield* Effect.yieldNow;
2165 }
2166 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
2167 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 4);
2168 }),
2169 );
2170
2171 it.effect("resumes replacement sessions from the latest applied sequence", () =>
2172 Effect.gen(function* () {
2173 const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true });
2174 yield* Queue.offer(
2175 harness.inputs,
2176 titleUpdated("Latest title", CACHED_SNAPSHOT_SEQUENCE + 1),
2177 );
2178 yield* Queue.offer(harness.inputs, synchronized());
2179 yield* awaitThreadState(
2180 harness.observed,
2181 (value) =>
2182 value.status === "live" &&
2183 Option.isSome(value.data) &&
2184 value.data.value.thread.title === "Latest title",
2185 );
2186
2187 yield* harness.replaceSession;
2188 for (let attempt = 0; attempt < 100; attempt += 1) {
2189 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break;
2190 yield* Effect.yieldNow;
2191 }
2192
2193 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
2194 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1);
2195 expect((yield* Ref.get(harness.latest)).status).toBe("synchronizing");
2196 }),
2197 );
2198
2199 it.effect("resubscribes on app foreground from the latest applied sequence", () =>
2200 Effect.gen(function* () {
2201 const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true });
2202 yield* Queue.offer(
2203 harness.inputs,
2204 titleUpdated("Latest title", CACHED_SNAPSHOT_SEQUENCE + 1),
2205 );
2206 yield* Queue.offer(harness.inputs, synchronized());
2207 yield* awaitThreadState(
2208 harness.observed,
2209 (value) =>
2210 value.status === "live" &&
2211 Option.isSome(value.data) &&
2212 value.data.value.thread.title === "Latest title",
2213 );
2214
2215 yield* Queue.offer(harness.wakeups, "application-active");
2216 const synchronizing = yield* awaitThreadState(
2217 harness.observed,
2218 (value) => value.status === "synchronizing" && Option.isSome(value.data),
2219 );
2220 for (let attempt = 0; attempt < 100; attempt += 1) {
2221 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break;
2222 yield* Effect.yieldNow;
2223 }
2224
2225 expect(synchronizing.status).toBe("synchronizing");
2226 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
2227 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1);
2228 expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true);
2229 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
2230
2231 yield* Queue.offer(harness.inputs, synchronized());
2232 const live = yield* awaitThreadState(
2233 harness.observed,
2234 (value) => value.status === "live" && Option.isSome(value.data),
2235 );
2236 expect(Option.getOrThrow(live.data).thread.title).toBe("Latest title");
2237
2238 yield* Queue.offer(harness.wakeups, "application-active-probe");
2239 for (let attempt = 0; attempt < 100; attempt += 1) {
2240 if ((yield* Ref.get(harness.subscriptionCount)) >= 3) break;
2241 yield* Effect.yieldNow;
2242 }
2243 expect(yield* Ref.get(harness.subscriptionCount)).toBe(3);
2244
2245 yield* Queue.offer(harness.wakeups, "application-active-reconnect");
2246 for (let attempt = 0; attempt < 10; attempt += 1) {
2247 yield* Effect.yieldNow;
2248 }
2249 expect(yield* Ref.get(harness.subscriptionCount)).toBe(3);
2250 }),
2251 );
2252});