1import {
2 EnvironmentId,
3 EventId,
4 MessageId,
5 ORCHESTRATION_V2_WS_METHODS,
6 TurnItemId,
7 type OrchestrationV2ThreadHistoryPage,
8 type OrchestrationV2ThreadProjection,
9 type OrchestrationV2ThreadDetailSnapshot,
10 type OrchestrationV2ThreadStreamItem,
11} from "@t3tools/contracts";
12import { afterEach, describe, expect, it, vi } from "@effect/vitest";
13import * as Deferred from "effect/Deferred";
14import * as Effect from "effect/Effect";
15import * as Fiber from "effect/Fiber";
16import { HttpClient, HttpClientResponse } from "effect/http";
17import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";
19import * as Queue from "effect/Queue";
20import * as Stream from "effect/Stream";
21import * as SubscriptionRef from "effect/SubscriptionRef";
22import { Atom, AtomRegistry } from "effect/reactivity";
23
24import type { ConnectionCatalogEntry } from "../connection/catalog.ts";
25import * as EnvironmentRegistry from "../connection/registry.ts";
26import {
27 AVAILABLE_CONNECTION_STATE,
28 PrimaryConnectionTarget,
29 type NetworkStatus,
30 type PreparedConnection,
31} from "../connection/model.ts";
32import * as EnvironmentSupervisor from "../connection/supervisor.ts";
33import * as Persistence from "../platform/persistence.ts";
34import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
35import type { RpcSession } from "../rpc/session.ts";
36import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts";
37import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts";
38import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts";
39import * as ThreadHistoryController from "./threadHistoryController.ts";
40import { createEnvironmentThreadStateAtoms, type EnvironmentThreadState } from "./threads.ts";
41import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
42
43const TARGET = new PrimaryConnectionTarget({
44 environmentId: EnvironmentId.make("environment-1"),
45 label: "Test environment",
46 httpBaseUrl: "https://environment.example.test",
47 wsBaseUrl: "wss://environment.example.test",
48});
49const THREAD_ID = v2ThreadId;
50const THREAD: OrchestrationV2ThreadProjection = {
51 ...v2Projection,
52 thread: { ...v2Projection.thread, title: "Cached thread" },
53};
54const SNAPSHOT: OrchestrationV2ThreadDetailSnapshot = { snapshotSequence: 7, projection: THREAD };
55
56const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: {
57 readonly snapshot?: OrchestrationV2ThreadDetailSnapshot;
58 readonly snapshotUnavailable?: boolean;
59}) {
60 const subscriptions = yield* Queue.unbounded<{
61 readonly afterSequence: number | undefined;
62 readonly events: Queue.Queue<OrchestrationV2ThreadStreamItem>;
63 readonly closed: Deferred.Deferred<void>;
64 }>();
65 const olderLoads = yield* Queue.unbounded<{
66 readonly cursor: string | null;
67 readonly response: Deferred.Deferred<OrchestrationV2ThreadHistoryPage>;
68 readonly closed: Deferred.Deferred<void>;
69 }>();
70 const snapshot = options?.snapshot ?? SNAPSHOT;
71 let httpLoads = 0;
72 let diskLoads = 0;
73 let opened = 0;
74 let active = 0;
75 const client = {
76 [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) =>
77 Stream.unwrap(
78 Effect.gen(function* () {
79 const events = yield* Queue.unbounded<OrchestrationV2ThreadStreamItem>();
80 const closed = yield* Deferred.make<void>();
81 yield* Effect.acquireRelease(
82 Effect.sync(() => {
83 opened += 1;
84 active += 1;
85 }),
86 () =>
87 Effect.sync(() => {
88 active -= 1;
89 }).pipe(Effect.andThen(Deferred.succeed(closed, undefined))),
90 );
91 yield* Queue.offer(subscriptions, { afterSequence: input.afterSequence, events, closed });
92 return Stream.fromQueue(events);
93 }),
94 ),
95 } as unknown as WsRpcProtocolClient;
96 const session: RpcSession = {
97 client,
98 initialConfig: Effect.succeed({
99 threadResumeCompletionMarker: true,
100 threadSnapshotPagination: true,
101 } as never),
102 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
103 ready: Effect.void,
104 probe: Effect.void,
105 closed: Effect.never,
106 };
107 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
108 target: TARGET,
109 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
110 session: yield* SubscriptionRef.make(Option.some(session)),
111 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
112 Option.some({
113 environmentId: TARGET.environmentId,
114 label: TARGET.label,
115 httpBaseUrl: TARGET.httpBaseUrl,
116 socketUrl: TARGET.wsBaseUrl,
117 httpAuthorization: null,
118 target: TARGET,
119 }),
120 ),
121 connect: Effect.void,
122 disconnect: Effect.void,
123 retryNow: Effect.void,
124 });
125 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
126 entries: yield* SubscriptionRef.make<ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>>(
127 new Map(),
128 ),
129 networkStatus: yield* SubscriptionRef.make<NetworkStatus>("online"),
130 start: Effect.void,
131 register: () => Effect.die("Unexpected environment registration"),
132 registerPlatform: () => Effect.die("Unexpected environment registration"),
133 reconcilePlatform: () => Effect.die("Unexpected environment reconciliation"),
134 remove: () => Effect.die("Unexpected environment removal"),
135 removeRoute: () => Effect.die("Unexpected route removal"),
136 reorderRoutes: () => Effect.die("Unexpected route reorder"),
137 removeRelayEnvironments: () => Effect.die("Unexpected environment removal"),
138 retryNow: () => Effect.void,
139 setEnabled: () => Effect.die("Unexpected environment toggle"),
140 setCompatibility: () => Effect.die("Unexpected compatibility update"),
141 state: () => SubscriptionRef.get(supervisor.state),
142 stateChanges: () => SubscriptionRef.changes(supervisor.state),
143 run: (_environmentId, effect) =>
144 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
145 runStream: (_environmentId, stream) =>
146 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
147 followStream: (_environmentId, stream) =>
148 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
149 });
150 const historyController = yield* Effect.service(
151 ThreadHistoryController.ThreadHistoryController,
152 ).pipe(Effect.provide(ThreadHistoryController.layer));
153 const historyHttpClient = HttpClient.make((request, url) =>
154 Effect.gen(function* () {
155 const response = yield* Deferred.make<OrchestrationV2ThreadHistoryPage>();
156 const closed = yield* Deferred.make<void>();
157 yield* Effect.addFinalizer(() => Deferred.succeed(closed, undefined));
158 yield* Queue.offer(olderLoads, { cursor: url.searchParams.get("cursor"), response, closed });
159 const page = yield* Deferred.await(response);
160 return HttpClientResponse.fromWeb(request, Response.json(page));
161 }).pipe(Effect.scoped),
162 );
163 const runtime = Atom.runtime(
164 Layer.mergeAll(
165 Layer.succeed(ThreadHistoryController.ThreadHistoryController, historyController),
166 Layer.succeed(HttpClient.HttpClient, historyHttpClient),
167 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
168 Layer.succeed(
169 Persistence.EnvironmentCacheStore,
170 Persistence.EnvironmentCacheStore.of({
171 loadShell: () => Effect.succeedNone,
172 saveShell: () => Effect.void,
173 loadThread: () =>
174 Effect.sync(() => {
175 diskLoads += 1;
176 return Option.none();
177 }),
178 saveThread: () => Effect.void,
179 removeThread: () => Effect.void,
180 loadServerConfig: () => Effect.succeedNone,
181 saveServerConfig: () => Effect.void,
182 loadVcsRefs: () => Effect.succeedNone,
183 saveVcsRefs: () => Effect.void,
184 removeVcsRefs: () => Effect.void,
185 clearVcsRefs: () => Effect.void,
186 clear: () => Effect.void,
187 }),
188 ),
189 Layer.succeed(
190 ThreadSnapshotLoader.ThreadSnapshotLoader,
191 ThreadSnapshotLoader.ThreadSnapshotLoader.of({
192 load: () =>
193 Effect.sync(() => {
194 httpLoads += 1;
195 if (options?.snapshotUnavailable === true) {
196 return { _tag: "unavailable" as const };
197 }
198 return {
199 _tag: "present" as const,
200 snapshot,
201 ...(snapshot.historyCursor === undefined
202 ? {}
203 : {
204 history: {
205 historyCursor: snapshot.historyCursor,
206 hasMoreHistory: snapshot.hasMoreHistory ?? false,
207 latestLocalTurnOrdinal: snapshot.latestLocalTurnOrdinal ?? null,
208 },
209 }),
210 };
211 }),
212 }),
213 ),
214 ),
215 );
216 const raw = createEnvironmentThreadStateAtoms(runtime);
217 const details = createEnvironmentThreadDetailAtoms(raw.stateAtom);
218 const ref = { environmentId: TARGET.environmentId, threadId: THREAD_ID };
219 const stateAtom = details.stateAtom(ref);
220 const makeRegistry = Effect.acquireRelease(
221 Effect.sync(() => AtomRegistry.make({ defaultIdleTTL: 60_000, timeoutResolution: 1 })),
222 (registry) => Effect.sync(() => registry.dispose()),
223 );
224 const registry = yield* makeRegistry;
225
226 return {
227 registry,
228 makeRegistry,
229 rawAtoms: raw,
230 stateAtom,
231 details,
232 ref,
233 subscriptions,
234 olderLoads,
235 loadEarlier: () => historyController.loadEarlier(TARGET.environmentId, THREAD_ID),
236 counts: () => ({ httpLoads, diskLoads, opened, active }),
237 };
238});
239
240function observeState(
241 registry: AtomRegistry.AtomRegistry,
242 atom: Atom.Atom<EnvironmentThreadState>,
243 predicate: (state: EnvironmentThreadState) => boolean,
244) {
245 return AtomRegistry.toStream(registry, atom).pipe(
246 Stream.filter(predicate),
247 Stream.runHead,
248 Effect.map(Option.getOrThrow),
249 );
250}
251
252function currentThread(
253 registry: AtomRegistry.AtomRegistry,
254 atom: Atom.Atom<EnvironmentThreadState>,
255) {
256 return Option.getOrThrow(registry.get(atom).data);
257}
258
259describe("createEnvironmentThreadStateAtoms", () => {
260 afterEach(() => {
261 vi.useRealTimers();
262 vi.restoreAllMocks();
263 });
264
265 it.effect("shares one live stream and closes it after the last detail consumer leaves", () =>
266 Effect.gen(function* () {
267 const h = yield* makeHarness();
268 const unmountMessages = h.registry.mount(h.details.visibleTurnItemsAtom(h.ref));
269 const first = yield* Queue.take(h.subscriptions);
270 const unmountStatus = h.registry.mount(h.details.statusAtom(h.ref));
271 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 1, active: 1 });
272 unmountMessages();
273 yield* Queue.offer(first.events, { kind: "synchronized" });
274 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
275 expect(h.counts().active).toBe(1);
276 unmountStatus();
277 yield* Deferred.await(first.closed);
278 expect(h.counts().active).toBe(0);
279 }),
280 );
281
282 it.effect("keeps warm data and resumes a completed cursor without loading another snapshot", () =>
283 Effect.gen(function* () {
284 const h = yield* makeHarness();
285 const unmount = h.registry.mount(h.stateAtom);
286 const first = yield* Queue.take(h.subscriptions);
287 yield* Queue.offer(first.events, {
288 kind: "event",
289 sequence: 8,
290 event: {
291 id: EventId.make("metadata-1"),
292 type: "thread.metadata-updated",
293 threadId: THREAD_ID,
294 occurredAt: THREAD.thread.updatedAt,
295 payload: { ...THREAD.thread, title: "Retained title" },
296 },
297 });
298 yield* Queue.offer(first.events, { kind: "synchronized" });
299 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
300 const before = currentThread(h.registry, h.stateAtom);
301 unmount();
302 yield* Deferred.await(first.closed);
303 const remount = h.registry.mount(h.stateAtom);
304 expect(currentThread(h.registry, h.stateAtom)).toBe(before);
305 const next = yield* Queue.take(h.subscriptions);
306 expect(next.afterSequence).toBe(8);
307 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 2, active: 1 });
308 remount();
309 yield* Deferred.await(next.closed);
310 }),
311 );
312
313 it.effect("keeps warm data when the raw atom family's weak entry is collected", () =>
314 Effect.gen(function* () {
315 const h = yield* makeHarness();
316 const oldRaw = h.rawAtoms.stateAtom(TARGET.environmentId, THREAD_ID);
317 const unmount = h.registry.mount(h.stateAtom);
318 const first = yield* Queue.take(h.subscriptions);
319 const latest = { ...THREAD, thread: { ...THREAD.thread, title: "Newer cached thread" } };
320 yield* Queue.offer(first.events, {
321 kind: "snapshot",
322 snapshotSequence: 8,
323 projection: latest,
324 });
325 yield* Queue.offer(first.events, { kind: "synchronized" });
326 yield* observeState(h.registry, h.stateAtom, (value) => value.status === "live");
327 unmount();
328 yield* Deferred.await(first.closed);
329
330 // Force the weak-family miss without depending on host GC timing.
331 const deref = WeakRef.prototype.deref;
332 vi.spyOn(WeakRef.prototype, "deref").mockImplementation(function (this: WeakRef<object>) {
333 const value = deref.call(this);
334 return value === oldRaw ? undefined : value;
335 });
336 const remount = h.registry.mount(h.stateAtom);
337 expect(currentThread(h.registry, h.stateAtom)).toBe(latest);
338 const next = yield* Queue.take(h.subscriptions);
339 expect(next.afterSequence).toBe(8);
340 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 2, active: 1 });
341 remount();
342 yield* Deferred.await(next.closed);
343 }),
344 );
345
346 it.effect("keeps cached snapshots local to each registry", () =>
347 Effect.gen(function* () {
348 const h = yield* makeHarness();
349 const unmount = h.registry.mount(h.stateAtom);
350 const first = yield* Queue.take(h.subscriptions);
351 unmount();
352 yield* Deferred.await(first.closed);
353 const otherRegistry = yield* h.makeRegistry;
354 const unmountOther = otherRegistry.mount(h.stateAtom);
355 const other = yield* Queue.take(h.subscriptions);
356 expect(h.counts()).toEqual({ httpLoads: 2, diskLoads: 2, opened: 2, active: 1 });
357 unmountOther();
358 yield* Deferred.await(other.closed);
359 }),
360 );
361
362 it.effect("expires the plain snapshot after five idle minutes", () =>
363 Effect.gen(function* () {
364 vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
365 const h = yield* makeHarness();
366 const unmount = h.registry.mount(h.stateAtom);
367 const first = yield* Queue.take(h.subscriptions);
368 unmount();
369 yield* Deferred.await(first.closed);
370 yield* Effect.yieldNow;
371 yield* Effect.promise(() => vi.advanceTimersByTimeAsync(THREAD_SNAPSHOT_IDLE_TTL_MS + 1));
372 const remount = h.registry.mount(h.stateAtom);
373 const next = yield* Queue.take(h.subscriptions);
374 expect(h.counts()).toEqual({ httpLoads: 2, diskLoads: 2, opened: 2, active: 1 });
375 remount();
376 yield* Deferred.await(next.closed);
377 }),
378 );
379
380 it.effect("cancels older-page work on unmount and permits it again on a warm return", () =>
381 Effect.gen(function* () {
382 const h = yield* makeHarness({
383 snapshot: {
384 ...SNAPSHOT,
385 historyCursor: "older-1",
386 hasMoreHistory: true,
387 },
388 });
389 const unmount = h.registry.mount(h.stateAtom);
390 const first = yield* Queue.take(h.subscriptions);
391 const loading = yield* h.loadEarlier().pipe(Effect.forkScoped);
392 const older = yield* Queue.take(h.olderLoads);
393 expect(h.registry.get(h.stateAtom).history.loading).toBe(true);
394 unmount();
395 yield* Deferred.await(first.closed);
396 yield* Deferred.await(older.closed);
397 yield* Fiber.await(loading);
398 const remount = h.registry.mount(h.stateAtom);
399 const next = yield* Queue.take(h.subscriptions);
400 expect(h.registry.get(h.stateAtom).history.loading).toBe(false);
401 const retrying = yield* h.loadEarlier().pipe(Effect.forkScoped);
402 const retried = yield* Queue.take(h.olderLoads);
403 expect(retried.cursor).toBe("older-1");
404 expect(h.counts().httpLoads).toBe(1);
405 remount();
406 yield* Deferred.await(next.closed);
407 yield* Deferred.await(retried.closed);
408 yield* Fiber.await(retrying);
409 }),
410 );
411
412 it.effect(
413 "merges older history after an HTTP failure falls back to a bounded socket snapshot",
414 () =>
415 Effect.gen(function* () {
416 const h = yield* makeHarness({ snapshotUnavailable: true });
417 const unmount = h.registry.mount(h.stateAtom);
418 const subscription = yield* Queue.take(h.subscriptions);
419 expect(subscription.afterSequence).toBeUndefined();
420
421 yield* Queue.offer(subscription.events, {
422 kind: "snapshot",
423 snapshotSequence: 12,
424 projection: THREAD,
425 historyCursor: "socket-older-1",
426 hasMoreHistory: true,
427 latestLocalTurnOrdinal: 4,
428 payloadBudgetExceeded: false,
429 });
430 yield* Queue.offer(subscription.events, { kind: "synchronized" });
431 yield* observeState(
432 h.registry,
433 h.stateAtom,
434 (state) => state.status === "live" && state.history.historyCursor === "socket-older-1",
435 );
436
437 const loading = yield* h.loadEarlier().pipe(Effect.forkScoped);
438 const request = yield* Queue.take(h.olderLoads);
439 expect(request.cursor).toBe("socket-older-1");
440 const olderItem = {
441 id: TurnItemId.make("socket-fallback-older-item"),
442 type: "command_execution" as const,
443 threadId: THREAD_ID,
444 runId: null,
445 nodeId: null,
446 providerThreadId: null,
447 providerTurnId: null,
448 nativeItemRef: null,
449 parentItemId: null,
450 ordinal: 1,
451 status: "completed" as const,
452 title: "Earlier command",
453 input: "pwd",
454 output: "/workspace",
455 exitCode: 0,
456 startedAt: THREAD.thread.createdAt,
457 completedAt: THREAD.thread.createdAt,
458 updatedAt: THREAD.thread.createdAt,
459 };
460 yield* Deferred.succeed(request.response, {
461 snapshotSequence: 12,
462 items: [
463 {
464 position: 0,
465 visibility: "local",
466 sourceThreadId: THREAD_ID,
467 sourceItemId: olderItem.id,
468 item: olderItem,
469 },
470 ],
471 nextCursor: null,
472 hasMoreHistory: false,
473 });
474 expect(yield* Fiber.join(loading)).toEqual({ _tag: "loaded" });
475
476 const state = h.registry.get(h.stateAtom);
477 expect(
478 Option.getOrThrow(state.data).visibleTurnItems.map((row) => row.sourceItemId),
479 ).toEqual([olderItem.id]);
480 expect(state.history).toMatchObject({
481 historyCursor: null,
482 hasMoreHistory: false,
483 loading: false,
484 error: null,
485 expanded: true,
486 latestLocalTurnOrdinal: 4,
487 });
488 expect(h.counts().httpLoads).toBe(1);
489
490 unmount();
491 yield* Deferred.await(subscription.closed);
492 }),
493 );
494});
495
496it.effect.each([1, 16, 500])("publishes V2 message replay once per batch of %i", (batchSize) =>
497 Effect.gen(function* () {
498 const h = yield* makeHarness();
499 const unmount = h.registry.mount(h.stateAtom);
500 const first = yield* Queue.take(h.subscriptions);
501 let updates = 0;
502 const stop = h.registry.subscribe(
503 Atom.map(h.stateAtom, (state) => Option.getOrNull(state.data)?.messages),
504 () => updates++,
505 {
506 immediate: true,
507 },
508 );
509 updates = 0;
510 const events: OrchestrationV2ThreadStreamItem[] = Array.from({ length: 500 }, (_, index) => ({
511 kind: "event",
512 sequence: 8 + index,
513 event: {
514 id: EventId.make(`replay-${index}`),
515 type: "message.updated",
516 threadId: THREAD_ID,
517 occurredAt: THREAD.thread.createdAt,
518 payload: {
519 id: MessageId.make("replayed-message"),
520 threadId: THREAD_ID,
521 runId: null,
522 nodeId: null,
523 role: "assistant",
524 text: `${index},`,
525 streaming: true,
526 attachments: [],
527 createdBy: "agent",
528 creationSource: "provider",
529 createdAt: THREAD.thread.createdAt,
530 updatedAt: THREAD.thread.createdAt,
531 },
532 },
533 }));
534 for (let offset = 0; offset < events.length; offset += batchSize) {
535 yield* Queue.offerAll(first.events, events.slice(offset, offset + batchSize));
536 const last = Math.min(offset + batchSize, events.length) - 1;
537 yield* observeState(
538 h.registry,
539 h.stateAtom,
540 (state) => Option.getOrNull(state.data)?.messages[0]?.text === `${last},`,
541 );
542 }
543 yield* Queue.offerAll(first.events, [events[499]!, events[0]!, { kind: "synchronized" }]);
544 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
545 expect(currentThread(h.registry, h.stateAtom).messages[0]?.text).toBe("499,");
546 expect(updates).toBe(Math.ceil(500 / batchSize));
547 stop();
548 unmount();
549 yield* Deferred.await(first.closed);
550 const remount = h.registry.mount(h.stateAtom);
551 const next = yield* Queue.take(h.subscriptions);
552 expect(next.afterSequence).toBe(507);
553 remount();
554 yield* Deferred.await(next.closed);
555 }),
556);
557
558it.effect("keeps snapshot boundaries and completion markers ordered inside a replay batch", () =>
559 Effect.gen(function* () {
560 const h = yield* makeHarness();
561 const unmount = h.registry.mount(h.stateAtom);
562 const first = yield* Queue.take(h.subscriptions);
563 const title = (sequence: number, value: string): OrchestrationV2ThreadStreamItem => ({
564 kind: "event",
565 sequence,
566 event: {
567 id: EventId.make(`batch-title-${sequence}`),
568 type: "thread.metadata-updated",
569 threadId: THREAD_ID,
570 occurredAt: THREAD.thread.updatedAt,
571 payload: { ...THREAD.thread, title: value },
572 },
573 });
574 yield* Queue.offerAll(first.events, [
575 title(8, "Before snapshot"),
576 {
577 kind: "snapshot",
578 snapshotSequence: 20,
579 projection: THREAD,
580 historyCursor: "older-batch",
581 hasMoreHistory: true,
582 latestLocalTurnOrdinal: 10,
583 },
584 title(21, "After snapshot"),
585 title(19, "Stale replay"),
586 { kind: "synchronized" },
587 ]);
588 const state = yield* observeState(h.registry, h.stateAtom, (value) => value.status === "live");
589 expect(Option.getOrThrow(state.data).thread.title).toBe("After snapshot");
590 expect(state.history.historyCursor).toBe("older-batch");
591 expect(state.history.latestLocalTurnOrdinal).toBe(10);
592 unmount();
593 yield* Deferred.await(first.closed);
594 const remount = h.registry.mount(h.stateAtom);
595 const next = yield* Queue.take(h.subscriptions);
596 expect(next.afterSequence).toBe(21);
597 remount();
598 yield* Deferred.await(next.closed);
599 }),
600);