Open History to see the last commit
1import {
2 EnvironmentId,
3 EventId,
4 ORCHESTRATION_V2_WS_METHODS,
5 TurnItemId,
6 type OrchestrationV2ThreadHistoryPage,
7 type OrchestrationV2ThreadProjection,
8 type OrchestrationV2ThreadDetailSnapshot,
9 type OrchestrationV2ThreadStreamItem,
10} from "@t3tools/contracts";
11import { afterEach, describe, expect, it, vi } from "@effect/vitest";
12import * as Deferred from "effect/Deferred";
13import * as Effect from "effect/Effect";
14import * as Fiber from "effect/Fiber";
15import { HttpClient, HttpClientResponse } from "effect/http";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as Queue from "effect/Queue";
19import * as Stream from "effect/Stream";
20import * as SubscriptionRef from "effect/SubscriptionRef";
21import { Atom, AtomRegistry } from "effect/reactivity";
22
23import type { ConnectionCatalogEntry } from "../connection/catalog.ts";
24import * as EnvironmentRegistry from "../connection/registry.ts";
25import {
26 AVAILABLE_CONNECTION_STATE,
27 PrimaryConnectionTarget,
28 type NetworkStatus,
29 type PreparedConnection,
30} from "../connection/model.ts";
31import * as EnvironmentSupervisor from "../connection/supervisor.ts";
32import * as Persistence from "../platform/persistence.ts";
33import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
34import type { RpcSession } from "../rpc/session.ts";
35import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts";
36import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts";
37import * as ThreadHistoryController from "./threadHistoryController.ts";
38import { createEnvironmentThreadStateAtoms, type EnvironmentThreadState } from "./threads.ts";
39import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
40
41const TARGET = new PrimaryConnectionTarget({
42 environmentId: EnvironmentId.make("environment-1"),
43 label: "Test environment",
44 httpBaseUrl: "https://environment.example.test",
45 wsBaseUrl: "wss://environment.example.test",
46});
47const THREAD_ID = v2ThreadId;
48const THREAD: OrchestrationV2ThreadProjection = {
49 ...v2Projection,
50 thread: { ...v2Projection.thread, title: "Cached thread" },
51};
52const SNAPSHOT: OrchestrationV2ThreadDetailSnapshot = { snapshotSequence: 7, projection: THREAD };
53
54const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: {
55 readonly snapshot?: OrchestrationV2ThreadDetailSnapshot;
56 readonly socketSnapshot?: false;
57}) {
58 const subscriptions = yield* Queue.unbounded<{
59 readonly afterSequence: number | undefined;
60 readonly events: Queue.Queue<OrchestrationV2ThreadStreamItem>;
61 readonly closed: Deferred.Deferred<void>;
62 }>();
63 const olderLoads = yield* Queue.unbounded<{
64 readonly cursor: string | null;
65 readonly response: Deferred.Deferred<OrchestrationV2ThreadHistoryPage>;
66 readonly closed: Deferred.Deferred<void>;
67 }>();
68 const snapshot = options?.snapshot ?? SNAPSHOT;
69 let httpLoads = 0;
70 let diskLoads = 0;
71 let opened = 0;
72 let active = 0;
73 const client = {
74 [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) =>
75 Stream.unwrap(
76 Effect.gen(function* () {
77 const events = yield* Queue.unbounded<OrchestrationV2ThreadStreamItem>();
78 const closed = yield* Deferred.make<void>();
79 yield* Effect.acquireRelease(
80 Effect.sync(() => {
81 opened += 1;
82 active += 1;
83 }),
84 () =>
85 Effect.sync(() => {
86 active -= 1;
87 }).pipe(Effect.andThen(Deferred.succeed(closed, undefined))),
88 );
89 yield* Queue.offer(subscriptions, { afterSequence: input.afterSequence, events, closed });
90 // Like a paging server, a fresh subscribe opens with the bounded snapshot.
91 if (input.afterSequence === undefined && options?.socketSnapshot !== false) {
92 yield* Queue.offer(events, { kind: "snapshot", ...snapshot });
93 }
94 return Stream.fromQueue(events);
95 }),
96 ),
97 } as unknown as WsRpcProtocolClient;
98 const session: RpcSession = {
99 client,
100 initialConfig: Effect.succeed({
101 threadResumeCompletionMarker: true,
102 threadSnapshotPagination: true,
103 } as never),
104 subscribeServerConfig: (input = {}) => client.subscribeServerConfig(input),
105 ready: Effect.void,
106 probe: Effect.void,
107 inboundFrames: Effect.succeed(0),
108 closed: Effect.never,
109 };
110 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
111 target: TARGET,
112 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
113 session: yield* SubscriptionRef.make(Option.some(session)),
114 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
115 Option.some({
116 environmentId: TARGET.environmentId,
117 label: TARGET.label,
118 httpBaseUrl: TARGET.httpBaseUrl,
119 socketUrl: TARGET.wsBaseUrl,
120 httpAuthorization: null,
121 target: TARGET,
122 }),
123 ),
124 connect: Effect.void,
125 disconnect: Effect.void,
126 retryNow: Effect.void,
127 });
128 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
129 entries: yield* SubscriptionRef.make<ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>>(
130 new Map(),
131 ),
132 networkStatus: yield* SubscriptionRef.make<NetworkStatus>("online"),
133 start: Effect.void,
134 register: () => Effect.die("Unexpected environment registration"),
135 assertKnownAddress: () => Effect.die("Unexpected address check"),
136 reconcilePlatform: () => Effect.die("Unexpected environment reconciliation"),
137 remove: () => Effect.die("Unexpected environment removal"),
138 removeRoute: () => Effect.die("Unexpected route removal"),
139 reorderRoutes: () => Effect.die("Unexpected route reorder"),
140 removeRelayEnvironments: () => Effect.die("Unexpected environment removal"),
141 retryNow: () => Effect.void,
142 setEnabled: () => Effect.die("Unexpected environment toggle"),
143 setCompatibility: () => Effect.die("Unexpected compatibility update"),
144 state: () => SubscriptionRef.get(supervisor.state),
145 stateChanges: () => SubscriptionRef.changes(supervisor.state),
146 run: (_environmentId, effect) =>
147 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
148 runStream: (_environmentId, stream) =>
149 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
150 followStream: (_environmentId, stream) =>
151 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
152 });
153 const historyController = yield* Effect.service(
154 ThreadHistoryController.ThreadHistoryController,
155 ).pipe(Effect.provide(ThreadHistoryController.layer));
156 const historyHttpClient = HttpClient.make((request, url) =>
157 Effect.gen(function* () {
158 const response = yield* Deferred.make<OrchestrationV2ThreadHistoryPage>();
159 const closed = yield* Deferred.make<void>();
160 yield* Effect.addFinalizer(() => Deferred.succeed(closed, undefined));
161 yield* Queue.offer(olderLoads, { cursor: url.searchParams.get("cursor"), response, closed });
162 const page = yield* Deferred.await(response);
163 return HttpClientResponse.fromWeb(request, Response.json(page));
164 }).pipe(Effect.scoped),
165 );
166 const runtime = Atom.runtime(
167 Layer.mergeAll(
168 Layer.succeed(ThreadHistoryController.ThreadHistoryController, historyController),
169 Layer.succeed(HttpClient.HttpClient, historyHttpClient),
170 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
171 Layer.succeed(
172 Persistence.EnvironmentCacheStore,
173 Persistence.EnvironmentCacheStore.of({
174 loadShell: () => Effect.succeedNone,
175 saveShell: () => Effect.void,
176 loadThread: () =>
177 Effect.sync(() => {
178 diskLoads += 1;
179 return Option.none();
180 }),
181 saveThread: () => Effect.void,
182 removeThread: () => Effect.void,
183 loadServerConfig: () => Effect.succeedNone,
184 saveServerConfig: () => Effect.void,
185 loadVcsRefs: () => Effect.succeedNone,
186 saveVcsRefs: () => Effect.void,
187 removeVcsRefs: () => Effect.void,
188 clearVcsRefs: () => Effect.void,
189 clear: () => Effect.void,
190 }),
191 ),
192 Layer.succeed(
193 ThreadSnapshotLoader.ThreadSnapshotLoader,
194 ThreadSnapshotLoader.ThreadSnapshotLoader.of({
195 load: () =>
196 Effect.sync(() => {
197 httpLoads += 1;
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("keeps warm data and resumes a completed cursor without loading another snapshot", () =>
266 Effect.gen(function* () {
267 const h = yield* makeHarness();
268 const unmount = h.registry.mount(h.stateAtom);
269 const first = yield* Queue.take(h.subscriptions);
270 yield* Queue.offer(first.events, {
271 kind: "event",
272 sequence: 8,
273 event: {
274 id: EventId.make("metadata-1"),
275 type: "thread.metadata-updated",
276 threadId: THREAD_ID,
277 occurredAt: THREAD.thread.updatedAt,
278 payload: { ...THREAD.thread, title: "Retained title" },
279 },
280 });
281 yield* Queue.offer(first.events, { kind: "synchronized" });
282 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
283 const before = currentThread(h.registry, h.stateAtom);
284 unmount();
285 yield* Deferred.await(first.closed);
286 const remount = h.registry.mount(h.stateAtom);
287 expect(currentThread(h.registry, h.stateAtom)).toBe(before);
288 const next = yield* Queue.take(h.subscriptions);
289 expect(next.afterSequence).toBe(8);
290 expect(h.counts()).toEqual({ httpLoads: 0, diskLoads: 1, opened: 2, active: 1 });
291 remount();
292 yield* Deferred.await(next.closed);
293 }),
294 );
295
296 it.effect("cancels older-page work on unmount and permits it again on a warm return", () =>
297 Effect.gen(function* () {
298 const h = yield* makeHarness({
299 snapshot: {
300 ...SNAPSHOT,
301 historyCursor: "older-1",
302 hasMoreHistory: true,
303 },
304 });
305 const unmount = h.registry.mount(h.stateAtom);
306 const first = yield* Queue.take(h.subscriptions);
307 const loading = yield* h.loadEarlier().pipe(Effect.forkScoped);
308 const older = yield* Queue.take(h.olderLoads);
309 expect(h.registry.get(h.stateAtom).history.loading).toBe(true);
310 unmount();
311 yield* Deferred.await(first.closed);
312 yield* Deferred.await(older.closed);
313 yield* Fiber.await(loading);
314 const remount = h.registry.mount(h.stateAtom);
315 const next = yield* Queue.take(h.subscriptions);
316 expect(h.registry.get(h.stateAtom).history.loading).toBe(false);
317 const retrying = yield* h.loadEarlier().pipe(Effect.forkScoped);
318 const retried = yield* Queue.take(h.olderLoads);
319 expect(retried.cursor).toBe("older-1");
320 expect(h.counts().httpLoads).toBe(0);
321 remount();
322 yield* Deferred.await(next.closed);
323 yield* Deferred.await(retried.closed);
324 yield* Fiber.await(retrying);
325 }),
326 );
327
328 it.effect("opens from the bounded socket snapshot without HTTP and merges older history", () =>
329 Effect.gen(function* () {
330 const h = yield* makeHarness({ socketSnapshot: false });
331 const unmount = h.registry.mount(h.stateAtom);
332 const subscription = yield* Queue.take(h.subscriptions);
333 expect(subscription.afterSequence).toBeUndefined();
334
335 yield* Queue.offer(subscription.events, {
336 kind: "snapshot",
337 snapshotSequence: 12,
338 projection: THREAD,
339 historyCursor: "socket-older-1",
340 hasMoreHistory: true,
341 latestLocalTurnOrdinal: 4,
342 });
343 yield* Queue.offer(subscription.events, { kind: "synchronized" });
344 yield* observeState(
345 h.registry,
346 h.stateAtom,
347 (state) => state.status === "live" && state.history.historyCursor === "socket-older-1",
348 );
349
350 const loading = yield* h.loadEarlier().pipe(Effect.forkScoped);
351 const request = yield* Queue.take(h.olderLoads);
352 expect(request.cursor).toBe("socket-older-1");
353 const olderItem = {
354 id: TurnItemId.make("socket-fallback-older-item"),
355 type: "command_execution" as const,
356 threadId: THREAD_ID,
357 runId: null,
358 nodeId: null,
359 providerThreadId: null,
360 providerTurnId: null,
361 nativeItemRef: null,
362 parentItemId: null,
363 ordinal: 1,
364 status: "completed" as const,
365 title: "Earlier command",
366 input: "pwd",
367 output: "/workspace",
368 exitCode: 0,
369 startedAt: THREAD.thread.createdAt,
370 completedAt: THREAD.thread.createdAt,
371 updatedAt: THREAD.thread.createdAt,
372 };
373 yield* Deferred.succeed(request.response, {
374 snapshotSequence: 12,
375 items: [
376 {
377 position: 0,
378 visibility: "local",
379 sourceThreadId: THREAD_ID,
380 sourceItemId: olderItem.id,
381 item: olderItem,
382 },
383 ],
384 nextCursor: null,
385 hasMoreHistory: false,
386 });
387 expect(yield* Fiber.join(loading)).toEqual({ _tag: "loaded" });
388
389 const state = h.registry.get(h.stateAtom);
390 expect(Option.getOrThrow(state.data).visibleTurnItems.map((row) => row.sourceItemId)).toEqual(
391 [olderItem.id],
392 );
393 expect(state.history).toMatchObject({
394 historyCursor: null,
395 hasMoreHistory: false,
396 loading: false,
397 error: null,
398 expanded: true,
399 latestLocalTurnOrdinal: 4,
400 });
401 expect(h.counts().httpLoads).toBe(0);
402
403 unmount();
404 yield* Deferred.await(subscription.closed);
405 }),
406 );
407});
408
409it.effect("keeps snapshot boundaries and completion markers ordered inside a replay batch", () =>
410 Effect.gen(function* () {
411 const h = yield* makeHarness();
412 const unmount = h.registry.mount(h.stateAtom);
413 const first = yield* Queue.take(h.subscriptions);
414 const title = (sequence: number, value: string): OrchestrationV2ThreadStreamItem => ({
415 kind: "event",
416 sequence,
417 event: {
418 id: EventId.make(`batch-title-${sequence}`),
419 type: "thread.metadata-updated",
420 threadId: THREAD_ID,
421 occurredAt: THREAD.thread.updatedAt,
422 payload: { ...THREAD.thread, title: value },
423 },
424 });
425 yield* Queue.offerAll(first.events, [
426 title(8, "Before snapshot"),
427 {
428 kind: "snapshot",
429 snapshotSequence: 20,
430 projection: THREAD,
431 historyCursor: "older-batch",
432 hasMoreHistory: true,
433 latestLocalTurnOrdinal: 10,
434 },
435 title(21, "After snapshot"),
436 title(19, "Stale replay"),
437 { kind: "synchronized" },
438 ]);
439 const state = yield* observeState(h.registry, h.stateAtom, (value) => value.status === "live");
440 expect(Option.getOrThrow(state.data).thread.title).toBe("After snapshot");
441 expect(state.history.historyCursor).toBe("older-batch");
442 expect(state.history.latestLocalTurnOrdinal).toBe(10);
443 unmount();
444 yield* Deferred.await(first.closed);
445 const remount = h.registry.mount(h.stateAtom);
446 const next = yield* Queue.take(h.subscriptions);
447 expect(next.afterSequence).toBe(21);
448 remount();
449 yield* Deferred.await(next.closed);
450 }),
451);