1import {
2 EnvironmentId,
3 ORCHESTRATION_V2_WS_METHODS,
4 OrchestrationV2GetShellSnapshotError,
5 ProjectId,
6 ThreadId,
7 type OrchestrationV2OlderSettledThreadsPage,
8 type OrchestrationV2SearchThreadsResult,
9 type OrchestrationV2ShellSnapshot,
10 type OrchestrationV2ThreadShell,
11} from "@t3tools/contracts";
12import { describe, expect, it } from "@effect/vitest";
13import * as Clock from "effect/Clock";
14import * as DateTime from "effect/DateTime";
15import * as Deferred from "effect/Deferred";
16import * as Effect from "effect/Effect";
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 { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
23import * as TestClock from "effect/testing/TestClock";
24
25import {
26 AVAILABLE_CONNECTION_STATE,
27 PrimaryConnectionTarget,
28 type PreparedConnection,
29 type SupervisorConnectionState,
30} from "../connection/model.ts";
31import * as EnvironmentRegistry from "../connection/registry.ts";
32import * as EnvironmentSupervisor from "../connection/supervisor.ts";
33import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
34import type * as RpcSession from "../rpc/session.ts";
35import { EMPTY_SHELL_STATE, type EnvironmentShellState } from "./shell.ts";
36import {
37 createOlderSettledThreadAtoms,
38 isListedSettledThread,
39 olderSettledCountKey,
40 settledShelfFrontier,
41 unloadedOlderSettledCount,
42 wantedOlderSettledPages,
43 type OlderSettled,
44 type OlderSettledPages,
45} from "./olderSettledThreads.ts";
46
47const DAY = 24 * 60 * 60 * 1000;
48const environmentId = EnvironmentId.make("env-a");
49const project = ProjectId.make("project-a");
50const before = DateTime.formatIso(DateTime.makeUnsafe(50 * DAY));
51const older: OlderSettled = {
52 before,
53 beforeMs: 50 * DAY,
54 counts: new Map([[project, 3]]),
55};
56
57const row = (
58 id: string,
59 input: {
60 readonly settledDay?: number;
61 readonly createdDay?: number;
62 readonly subagent?: boolean;
63 },
64) =>
65 ({
66 id: ThreadId.make(id),
67 projectId: project,
68 lineage: {
69 parentThreadId: input.subagent === true ? ThreadId.make("parent") : null,
70 relationshipToParent: input.subagent === true ? "subagent" : null,
71 rootThreadId: ThreadId.make(id),
72 },
73 settledOverride: input.settledDay === undefined ? null : "settled",
74 settledAt: input.settledDay === undefined ? null : DateTime.makeUnsafe(input.settledDay * DAY),
75 createdAt: DateTime.makeUnsafe((input.createdDay ?? 1) * DAY),
76 }) as unknown as OrchestrationV2ThreadShell;
77
78describe("olderSettledCountKey", () => {
79 const key = (
80 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
81 removedAt: ReadonlyMap<ThreadId, number> = new Map(),
82 ) => olderSettledCountKey(threads, older.beforeMs, removedAt);
83 const base = [row("kept-old", { settledDay: 10 }), row("open", { createdDay: 60 })];
84
85 it("changes on every event that can change the older count", () => {
86 const start = key(base);
87 // A removal, even of an id removed before (the map keeps its size).
88 expect(key(base, new Map([[ThreadId.make("x"), 4]]))).not.toBe(start);
89 expect(key(base, new Map([[ThreadId.make("x"), 9]]))).not.toBe(
90 key(base, new Map([[ThreadId.make("x"), 4]])),
91 );
92 // An omitted thread that arrives unsettled.
93 expect(key([...base, row("woken", { createdDay: 5 })])).not.toBe(start);
94 // A local row that turns older (a backdated auto-settle).
95 expect(key([base[0]!, row("open", { createdDay: 60, settledDay: 20 })])).not.toBe(start);
96 });
97
98 it("stays put for new threads and subagents", () => {
99 const start = key(base);
100 expect(key([...base, row("new", { createdDay: 70 })])).toBe(start);
101 expect(key([...base, row("new-settled", { createdDay: 70, settledDay: 71 })])).toBe(start);
102 expect(key([...base, row("child", { createdDay: 5, subagent: true })])).toBe(start);
103 });
104});
105
106describe("settled shelf listing", () => {
107 const pages = (frontier: OlderSettledPages["frontier"]): OlderSettledPages => ({
108 reads: [],
109 frontier,
110 status: "ready",
111 });
112 const at = (day: number) => DateTime.formatIso(DateTime.makeUnsafe(day * DAY));
113
114 it("lists rows at or above the highest frontier still loading", () => {
115 // No page yet: the cut is the frontier, so only rows newer than it list.
116 const one = settledShelfFrontier([{ older, pages: pages({ settledAt: before }) }]);
117 expect(isListedSettledThread(row("recent", { settledDay: 60 }), one)).toBe(true);
118 expect(isListedSettledThread(row("old", { settledDay: 30 }), one)).toBe(false);
119 // A row that became older inside the loaded range lists at its place.
120 const loaded = settledShelfFrontier([{ older, pages: pages({ settledAt: at(20) }) }]);
121 expect(isListedSettledThread(row("old", { settledDay: 30 }), loaded)).toBe(true);
122 expect(isListedSettledThread(row("older", { settledDay: 10 }), loaded)).toBe(false);
123 // All loaded, or a shell that keeps every settled thread: everything lists.
124 expect(settledShelfFrontier([{ older, pages: pages(null) }])).toBeNull();
125 expect(settledShelfFrontier([{ older: null, pages: pages(null) }])).toBeNull();
126 });
127
128 it("holds back another environment's rows below an environment still loading", () => {
129 const frontier = settledShelfFrontier([
130 { older, pages: pages({ settledAt: at(40) }) },
131 { older, pages: pages({ settledAt: at(20) }) },
132 ]);
133 expect(frontier).toEqual({ settledAt: at(40) });
134 expect(isListedSettledThread(row("b-loaded", { settledDay: 30 }), frontier)).toBe(false);
135 // A failed environment holds no one back.
136 expect(
137 settledShelfFrontier([
138 { older, pages: { ...pages({ settledAt: at(40) }), status: "failed" } },
139 { older, pages: pages({ settledAt: at(20) }) },
140 ]),
141 ).toEqual({ settledAt: at(20) });
142 });
143
144 it("on a settle-time tie, stops at the cursor that admits fewest rows", () => {
145 const strict = pages({ settledAt: at(30) });
146 const afterM = pages({ settledAt: at(30), threadId: ThreadId.make("m") });
147 const afterK = pages({ settledAt: at(30), threadId: ThreadId.make("k") });
148 const frontier = (...all: ReadonlyArray<OlderSettledPages>) =>
149 settledShelfFrontier(all.map((each) => ({ older, pages: each })));
150 expect(frontier(strict, afterM)).toEqual(afterM.frontier);
151 expect(frontier(afterM, strict)).toEqual(afterM.frontier);
152 expect(frontier(afterM, afterK)).toEqual(afterK.frontier);
153 });
154
155 it("follows the server's id order on a settle-time tie", () => {
156 const frontier = { settledAt: at(30), threadId: ThreadId.make("m") };
157 expect(isListedSettledThread(row("a", { settledDay: 30 }), frontier)).toBe(true);
158 expect(isListedSettledThread(row("z", { settledDay: 30 }), frontier)).toBe(false);
159 });
160});
161
162it("counts older threads the client does not hold, once", () => {
163 const local = [
164 // Kept by the snapshot (snoozed) and loaded by a page: the server counts both.
165 { source: row("snoozed-old", { settledDay: 10 }), environmentId },
166 { source: row("paged-old", { settledDay: 20 }), environmentId },
167 { source: row("recent", { settledDay: 60 }), environmentId },
168 ];
169 const olderSettled = () => older;
170 expect(
171 unloadedOlderSettledCount({
172 olderSettled,
173 scopes: [{ environmentId, projectIds: null }],
174 threads: local,
175 }),
176 ).toBe(1);
177 expect(
178 unloadedOlderSettledCount({
179 olderSettled,
180 scopes: [{ environmentId, projectIds: [ProjectId.make("other")] }],
181 threads: local,
182 }),
183 ).toBe(0);
184 // Two projects in one environment: each subtracts only its own loaded rows.
185 const other = ProjectId.make("project-b");
186 const twoProjects = (): OlderSettled => ({
187 ...older,
188 counts: new Map([
189 [project, 10],
190 [other, 10],
191 ]),
192 });
193 expect(
194 unloadedOlderSettledCount({
195 olderSettled: twoProjects,
196 scopes: [{ environmentId, projectIds: [project, other] }],
197 threads: [
198 { source: row("a-old", { settledDay: 10 }), environmentId },
199 { source: { ...row("b-old", { settledDay: 10 }), projectId: other }, environmentId },
200 ],
201 }),
202 ).toBe(18);
203});
204
205describe("wantedOlderSettledPages", () => {
206 const otherEnvironment = EnvironmentId.make("env-b");
207 const full = (settledAt: string): OlderSettledPages => ({
208 reads: [],
209 frontier: { settledAt, threadId: ThreadId.make("t") },
210 status: "ready",
211 });
212 const wanted = (input: {
213 readonly requested?: ReadonlyMap<EnvironmentId, number>;
214 readonly environments: ReadonlyMap<EnvironmentId, { readonly pages: OlderSettledPages }>;
215 readonly listed: number;
216 readonly visible: number;
217 }) =>
218 wantedOlderSettledPages({
219 requested: input.requested ?? new Map(),
220 environments: new Map(
221 [...input.environments].map(([id, { pages }]) => [id, { older, pages }] as const),
222 ),
223 listedCount: input.listed,
224 visibleCount: input.visible,
225 });
226
227 it("reads the next page of the environment holding the shelf back", () => {
228 const environments = new Map([
229 [environmentId, { pages: full("2026-01-10T00:00:00.000Z") }],
230 [otherEnvironment, { pages: full("2026-01-20T00:00:00.000Z") }],
231 ]);
232 expect(wanted({ environments, listed: 9, visible: 10 })).toEqual(
233 new Map([[otherEnvironment, 1]]),
234 );
235 // A deleted row drops the listed count again.
236 expect(
237 wanted({ requested: new Map([[environmentId, 1]]), environments, listed: 34, visible: 35 }),
238 ).toEqual(
239 new Map([
240 [environmentId, 1],
241 [otherEnvironment, 1],
242 ]),
243 );
244 });
245
246 it("waits for pending pages and stops at a short page or a closed shelf", () => {
247 const one = (pages: OlderSettledPages) => new Map([[environmentId, { pages }]]);
248 const pending = { ...full(before), status: "pending" as const };
249 const requested = new Map([[environmentId, 1]]);
250 expect(wanted({ requested, environments: one(pending), listed: 0, visible: 10 })).toBe(
251 requested,
252 );
253 const done = { ...full(before), frontier: null };
254 expect(wanted({ requested, environments: one(done), listed: 3, visible: 10 })).toBe(requested);
255 expect(wanted({ requested, environments: one(full(before)), listed: 10, visible: 10 })).toBe(
256 requested,
257 );
258 expect(wanted({ requested, environments: one(full(before)), listed: 0, visible: 0 })).toBe(
259 requested,
260 );
261 });
262});
263
264const TARGET = new PrimaryConnectionTarget({
265 environmentId,
266 label: "Older settled environment",
267 httpBaseUrl: "https://older.example.test",
268 wsBaseUrl: "wss://older.example.test",
269});
270
271// The atoms over a fake connected environment whose `getOlderSettledThreads` and `searchThreads`
272// take their replies, in order, from `replies`.
273const makeAtoms = Effect.fn(function* () {
274 const replies =
275 yield* Queue.unbounded<
276 Effect.Effect<
277 OrchestrationV2OlderSettledThreadsPage | OrchestrationV2SearchThreadsResult,
278 OrchestrationV2GetShellSnapshotError
279 >
280 >();
281 const requests: unknown[] = [];
282 // Each request, with a signal for when its read is stopped.
283 const requestLog = yield* Queue.unbounded<{
284 readonly input: unknown;
285 readonly stopped: Deferred.Deferred<void>;
286 }>();
287 const reply = (input: unknown) =>
288 Effect.gen(function* () {
289 requests.push(input);
290 const stopped = yield* Deferred.make<void>();
291 yield* Queue.offer(requestLog, { input, stopped });
292 return yield* Effect.flatten(Queue.take(replies)).pipe(
293 Effect.onInterrupt(() => Deferred.succeed(stopped, undefined)),
294 );
295 });
296 const client = {
297 [ORCHESTRATION_V2_WS_METHODS.getOlderSettledThreads]: reply,
298 [ORCHESTRATION_V2_WS_METHODS.searchThreads]: reply,
299 } as unknown as WsRpcProtocolClient;
300 const session = {
301 client,
302 initialConfig: Effect.succeed({} as never),
303 ready: Effect.void,
304 probe: Effect.void,
305 closed: Effect.never,
306 } as unknown as RpcSession.RpcSession;
307 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
308 target: TARGET,
309 state: yield* SubscriptionRef.make<SupervisorConnectionState>({
310 ...AVAILABLE_CONNECTION_STATE,
311 desired: true,
312 network: "online",
313 phase: "connected",
314 attempt: 1,
315 generation: 1,
316 }),
317 session: yield* SubscriptionRef.make(Option.some(session)),
318 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
319 connect: Effect.void,
320 disconnect: Effect.void,
321 retryNow: Effect.void,
322 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
323 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (_environmentId, effect) =>
324 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
325 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
326 _environmentId,
327 stream,
328 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
329 const runtime = Atom.runtime(
330 Layer.mergeAll(
331 Layer.succeed(
332 EnvironmentRegistry.EnvironmentRegistry,
333 EnvironmentRegistry.EnvironmentRegistry.of({
334 run,
335 followStream,
336 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
337 ),
338 // The test's clock, so the count re-read's wait moves with `TestClock.adjust`.
339 Layer.succeed(Clock.Clock, yield* Clock.Clock),
340 ),
341 );
342 const shellState = Atom.make<AsyncResult.AsyncResult<EnvironmentShellState>>(
343 AsyncResult.success(EMPTY_SHELL_STATE),
344 );
345 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
346 Effect.sync(() => registry.dispose()),
347 );
348 return {
349 replies,
350 requests,
351 requestLog,
352 atoms: createOlderSettledThreadAtoms(runtime, () => shellState),
353 registry,
354 setShell: (state: Partial<EnvironmentShellState>) =>
355 registry.set(shellState, AsyncResult.success({ ...EMPTY_SHELL_STATE, ...state })),
356 // The first value that passes, from a subscription.
357 until: <A>(atom: Atom.Atom<A>, done: (value: A) => boolean) =>
358 AtomRegistry.toStream(registry, atom).pipe(
359 Stream.filter(done),
360 Stream.runHead,
361 Effect.map(Option.getOrThrow),
362 ),
363 };
364});
365
366const shellSnapshot = (input: {
367 readonly sequence: number;
368 readonly asOfSequence: number;
369 readonly count: number;
370}) =>
371 Option.some({
372 schemaVersion: 1,
373 snapshotSequence: input.sequence,
374 projects: [],
375 threads: [row("recent", { settledDay: 60 })],
376 archivedThreads: [],
377 olderSettled: {
378 before,
379 asOfSequence: input.asOfSequence,
380 counts: [{ projectId: project, count: input.count }],
381 },
382 } as unknown as OrchestrationV2ShellSnapshot);
383
384describe("createOlderSettledThreadAtoms", () => {
385 it.effect("keeps counts current: the snapshot's own, then one read after changes settle", () =>
386 Effect.gen(function* () {
387 const { replies, requests, atoms, registry, setShell, until } = yield* makeAtoms();
388 const olderAtom = atoms.olderSettledAtom(environmentId);
389 const count = (value: OlderSettled | null) => value?.counts.get(project);
390 registry.mount(olderAtom);
391
392 // A cached shell: its counts stand, read nothing.
393 setShell({ snapshot: shellSnapshot({ sequence: 5, asOfSequence: 3, count: 5 }) });
394 expect(count(registry.get(olderAtom))).toBe(5);
395 // This session's snapshot: its counts are current until something changes.
396 setShell({
397 snapshot: shellSnapshot({ sequence: 3, asOfSequence: 3, count: 5 }),
398 generation: 1,
399 });
400 expect(count(registry.get(olderAtom))).toBe(5);
401
402 // A removal can change the count: one read once changes stop for a moment.
403 yield* Queue.offer(
404 replies,
405 Effect.succeed({
406 snapshotSequence: 4,
407 threads: [],
408 counts: [{ projectId: project, count: 4 }],
409 }),
410 );
411 setShell({
412 snapshot: shellSnapshot({ sequence: 4, asOfSequence: 3, count: 5 }),
413 generation: 1,
414 removedAt: new Map([[ThreadId.make("gone"), 4]]),
415 });
416 expect(count(registry.get(olderAtom))).toBe(5);
417 yield* TestClock.adjust("500 millis");
418 expect(count(yield* until(olderAtom, (value) => count(value) === 4))).toBe(4);
419 expect(requests).toEqual([{ before: { settledAt: before }, limit: 0 }]);
420
421 // A new snapshot brings its own counts and drops the read.
422 setShell({
423 snapshot: shellSnapshot({ sequence: 9, asOfSequence: 9, count: 7 }),
424 generation: 2,
425 });
426 expect(count(registry.get(olderAtom))).toBe(7);
427 expect(requests).toHaveLength(1);
428 }),
429 );
430
431 it.effect("reads pages in order until a short one, and retries a failed page", () =>
432 Effect.gen(function* () {
433 const { replies, requests, atoms, registry, setShell, until } = yield* makeAtoms();
434 const scope = { environmentId, projectIds: null };
435 const snapshot = shellSnapshot({ sequence: 3, asOfSequence: 3, count: 30 });
436
437 // A cached shell holds no other environment back: nothing can load yet.
438 setShell({ snapshot });
439 expect(registry.get(atoms.pagesAtom(scope, 1))).toEqual({
440 reads: [],
441 frontier: null,
442 status: "ready",
443 });
444
445 setShell({ snapshot, generation: 1 });
446 expect(registry.get(atoms.pagesAtom(scope, 0)).frontier).toEqual({ settledAt: before });
447 const full = Array.from({ length: 25 }, (_, index) =>
448 row(`full-${String(index).padStart(2, "0")}`, { settledDay: 40 - index }),
449 );
450 yield* Queue.offer(
451 replies,
452 Effect.succeed({ snapshotSequence: 3, threads: full, counts: [] }),
453 );
454 yield* Queue.offer(
455 replies,
456 Effect.succeed({
457 snapshotSequence: 3,
458 threads: [row("short", { settledDay: 10 })],
459 counts: [],
460 }),
461 );
462 const twoPages = atoms.pagesAtom(scope, 2);
463 registry.mount(twoPages);
464 const done = yield* until(twoPages, (pages) => pages.reads.length === 2);
465 expect(done.frontier).toBeNull();
466 expect(done.status).toBe("ready");
467 // The second page continues from the first page's last row.
468 expect(requests).toEqual([
469 { before: { settledAt: before }, limit: 25 },
470 {
471 before: { settledAt: DateTime.formatIso(full[24]!.settledAt!), threadId: full[24]!.id },
472 limit: 25,
473 },
474 ]);
475
476 // A new snapshot reads again; a failed page shows until a retry reads it.
477 const onePage = atoms.pagesAtom(scope, 1);
478 registry.mount(onePage);
479 setShell({ snapshot, generation: 2 });
480 expect(registry.get(onePage).status).toBe("pending");
481 yield* Queue.offer(
482 replies,
483 Effect.fail(new OrchestrationV2GetShellSnapshotError({ message: "read failed" })),
484 );
485 expect((yield* until(onePage, (pages) => pages.status === "failed")).frontier).toEqual({
486 settledAt: before,
487 });
488 atoms.retryPages(registry, scope, 1);
489 expect(registry.get(onePage).status).toBe("pending");
490 yield* Queue.offer(
491 replies,
492 Effect.succeed({ snapshotSequence: 3, threads: full.slice(0, 3), counts: [] }),
493 );
494 const retried = yield* until(onePage, (pages) => pages.status === "ready");
495 expect(retried.reads).toHaveLength(1);
496 expect(retried.frontier).toBeNull();
497 }),
498 );
499
500 it.effect("searches read once per generation, and nothing before the first snapshot", () =>
501 Effect.gen(function* () {
502 const { replies, requests, atoms, registry, setShell, until } = yield* makeAtoms();
503 const snapshot = shellSnapshot({ sequence: 3, asOfSequence: 3, count: 30 });
504 const input = { query: "old", olderSettledBefore: before, titlesOnly: true };
505 const search = atoms.searchAtom(environmentId, input);
506
507 // That snapshot would drop the rows: loading while it is on its way, idle while offline.
508 setShell({ snapshot, status: "synchronizing" });
509 registry.mount(search);
510 expect(registry.get(search).waiting).toBe(true);
511 setShell({ snapshot, status: "cached" });
512 expect(registry.get(search).waiting).toBe(false);
513
514 const found = { matches: [], snapshotSequence: 3, threads: [row("old", { settledDay: 9 })] };
515 setShell({ snapshot, status: "live", generation: 1 });
516 yield* Queue.offer(replies, Effect.succeed(found));
517 yield* until(search, AsyncResult.isSuccess);
518 // A new snapshot reads the rows again.
519 setShell({ snapshot, status: "live", generation: 2 });
520 yield* Queue.offer(replies, Effect.succeed({ ...found, snapshotSequence: 8 }));
521 yield* until(
522 search,
523 (result) => AsyncResult.getOrElse(result, () => null)?.snapshotSequence === 8,
524 );
525 expect(requests).toEqual([input, input]);
526 }),
527 );
528
529 it.effect("growing reads only the next page, and collapsing stops the shelf's reads", () =>
530 Effect.gen(function* () {
531 const { replies, requestLog, atoms, registry, setShell, until } = yield* makeAtoms();
532 setShell({
533 snapshot: shellSnapshot({ sequence: 3, asOfSequence: 3, count: 30 }),
534 generation: 1,
535 });
536 const scope = { environmentId, projectIds: null };
537 const full = Array.from({ length: 25 }, (_, index) =>
538 row(`full-${String(index).padStart(2, "0")}`, { settledDay: 40 - index }),
539 );
540 const releaseOne = registry.mount(atoms.pagesAtom(scope, 1));
541 expect((yield* Queue.take(requestLog)).input).toEqual({
542 before: { settledAt: before },
543 limit: 25,
544 });
545 yield* Queue.offer(
546 replies,
547 Effect.succeed({ snapshotSequence: 3, threads: full, counts: [] }),
548 );
549 yield* until(atoms.pagesAtom(scope, 1), (pages) => pages.reads.length === 1);
550
551 // React builds the next shelf before the last one lets go, so page 1 is not read again.
552 registry.get(atoms.pagesAtom(scope, 2));
553 releaseOne();
554 const releaseTwo = registry.mount(atoms.pagesAtom(scope, 2));
555 const second = yield* Queue.take(requestLog);
556 expect(second.input).toEqual({
557 before: { settledAt: DateTime.formatIso(full[24]!.settledAt!), threadId: full[24]!.id },
558 limit: 25,
559 });
560
561 // Collapsing stops the pending read: nothing is left to run again on a reconnect.
562 releaseTwo();
563 yield* Deferred.await(second.stopped);
564 }),
565 );
566});