Open History to see the last commit
1import {
2 EnvironmentId,
3 ORCHESTRATION_V2_WS_METHODS,
4 OrchestrationV2GetShellSnapshotError,
5 ThreadId,
6 type OrchestrationV2ShellSnapshot,
7 type OrchestrationV2ThreadFamily,
8 type OrchestrationV2ThreadShell,
9} from "@t3tools/contracts";
10import { describe, expect, it } from "@effect/vitest";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Queue from "effect/Queue";
15import * as Stream from "effect/Stream";
16import * as SubscriptionRef from "effect/SubscriptionRef";
17import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
18
19import {
20 AVAILABLE_CONNECTION_STATE,
21 PrimaryConnectionTarget,
22 type PreparedConnection,
23 type SupervisorConnectionState,
24} from "../connection/model.ts";
25import * as EnvironmentRegistry from "../connection/registry.ts";
26import * as EnvironmentSupervisor from "../connection/supervisor.ts";
27import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
28import type * as RpcSession from "../rpc/session.ts";
29import { EMPTY_SHELL_STATE, type EnvironmentShellState } from "./shell.ts";
30import {
31 createThreadFamilyAtoms,
32 threadFamilyRead,
33 threadFamilyRowsMissingFromShell,
34} from "./threadFamilies.ts";
35
36const row = (id: string, title = id) =>
37 ({ id: ThreadId.make(id), title }) as unknown as OrchestrationV2ThreadShell;
38
39describe("threadFamilyRowsMissingFromShell", () => {
40 it("adds only the rows the shell left out, once, with shell rows winning", () => {
41 const missing = threadFamilyRowsMissingFromShell({
42 shellThreads: [row("parent"), row("live", "live from shell")],
43 families: [
44 { snapshotSequence: 4, threads: [row("live", "stale family copy"), row("idle")] },
45 { snapshotSequence: 4, threads: [row("idle"), row("other")] },
46 ],
47 removedAt: new Map(),
48 });
49 expect(missing.map((thread) => thread.id)).toEqual(["idle", "other"]);
50 });
51
52 it("drops a family copy read before the shell removed that thread", () => {
53 const missing = threadFamilyRowsMissingFromShell({
54 shellThreads: [],
55 families: [
56 {
57 snapshotSequence: 5,
58 threads: [row("archived-after-read"), row("removed-before-read")],
59 },
60 ],
61 removedAt: new Map([
62 [ThreadId.make("archived-after-read"), 7],
63 [ThreadId.make("removed-before-read"), 3],
64 ]),
65 });
66 expect(missing.map((thread) => thread.id)).toEqual(["removed-before-read"]);
67 });
68});
69
70const TARGET = new PrimaryConnectionTarget({
71 environmentId: EnvironmentId.make("environment-family"),
72 label: "Family environment",
73 httpBaseUrl: "https://family.example.test",
74 wsBaseUrl: "wss://family.example.test",
75});
76
77describe("createThreadFamilyAtoms", () => {
78 it("merges mounted rows behind shell rows until released", () => {
79 const runtime = Atom.runtime(
80 Layer.succeed(
81 EnvironmentRegistry.EnvironmentRegistry,
82 {} as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"],
83 ),
84 );
85 const shellState = Atom.make<AsyncResult.AsyncResult<EnvironmentShellState>>(
86 AsyncResult.success<EnvironmentShellState>({
87 ...EMPTY_SHELL_STATE,
88 status: "live",
89 generation: 1,
90 snapshot: Option.some({
91 schemaVersion: 1,
92 snapshotSequence: 9,
93 projects: [],
94 threads: [row("kept")],
95 archivedThreads: [],
96 olderSettled: { before: "2026-10-01T00:00:00.000Z", asOfSequence: 4, counts: [] },
97 } as OrchestrationV2ShellSnapshot),
98 removedAt: new Map([[ThreadId.make("removed"), 9]]),
99 }),
100 );
101 const families = createThreadFamilyAtoms(runtime, () => shellState);
102 const registry = AtomRegistry.make();
103 const merged = () => registry.get(families.snapshotAtom(TARGET.environmentId))?.threads;
104 registry.mount(families.snapshotAtom(TARGET.environmentId));
105 // A settled shelf page read at sequence 5: the shell's copy wins, a later removal drops a row.
106 const page = Atom.make<ReadonlyArray<OrchestrationV2ThreadFamily>>([
107 { snapshotSequence: 5, threads: [row("kept", "page copy"), row("old"), row("removed")] },
108 // Read before the latest authoritative snapshot (sequence 4), whose removals it cleared.
109 { snapshotSequence: 3, threads: [row("stale-search-row")] },
110 ]);
111 const release = families.mountRows(registry, TARGET.environmentId, "older-settled:page", page);
112 expect(merged()?.map(({ id, title }) => [id, title])).toEqual([
113 ["kept", "kept"],
114 ["old", "old"],
115 ]);
116 release();
117 expect(merged()?.map(({ id }) => id)).toEqual(["kept"]);
118 registry.dispose();
119 });
120
121 it.effect("reads a family per shell generation and retries a failed read", () =>
122 Effect.scoped(
123 Effect.gen(function* () {
124 const replies =
125 yield* Queue.unbounded<
126 Effect.Effect<OrchestrationV2ThreadFamily, OrchestrationV2GetShellSnapshotError>
127 >();
128 const client = {
129 [ORCHESTRATION_V2_WS_METHODS.getThreadFamily]: () => Effect.flatten(Queue.take(replies)),
130 } as unknown as WsRpcProtocolClient;
131 const session = {
132 client,
133 initialConfig: Effect.succeed({ shellThreadFamilies: true } as never),
134 ready: Effect.void,
135 probe: Effect.void,
136 closed: Effect.never,
137 } as unknown as RpcSession.RpcSession;
138 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
139 target: TARGET,
140 state: yield* SubscriptionRef.make<SupervisorConnectionState>({
141 ...AVAILABLE_CONNECTION_STATE,
142 desired: true,
143 network: "online",
144 phase: "connected",
145 attempt: 1,
146 generation: 1,
147 }),
148 session: yield* SubscriptionRef.make(Option.some(session)),
149 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
150 connect: Effect.void,
151 disconnect: Effect.void,
152 retryNow: Effect.void,
153 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
154 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
155 _environmentId,
156 effect,
157 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
158 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
159 _environmentId,
160 stream,
161 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
162 const runtime = Atom.runtime(
163 Layer.succeed(
164 EnvironmentRegistry.EnvironmentRegistry,
165 EnvironmentRegistry.EnvironmentRegistry.of({
166 run,
167 followStream,
168 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
169 ),
170 );
171 const parent = row("parent");
172 const child = row("idle-child");
173 const shellSnapshot = {
174 schemaVersion: 1,
175 snapshotSequence: 3,
176 projects: [],
177 threads: [parent],
178 archivedThreads: [],
179 } as OrchestrationV2ShellSnapshot;
180 const shellAt = (generation: number): EnvironmentShellState => ({
181 ...EMPTY_SHELL_STATE,
182 status: "live",
183 snapshot: Option.some(shellSnapshot),
184 generation,
185 });
186 const shellState = Atom.make<AsyncResult.AsyncResult<EnvironmentShellState>>(
187 AsyncResult.success(shellAt(0)),
188 );
189 const families = createThreadFamilyAtoms(runtime, () => shellState);
190 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
191 Effect.sync(() => registry.dispose()),
192 );
193 const family = families.familyAtom(TARGET.environmentId, parent.id);
194 const mergedIds = () =>
195 registry.get(families.snapshotAtom(TARGET.environmentId))?.threads.map(({ id }) => id);
196 // Waits for the read to finish either way; the assertions read its outcome.
197 const settle = AtomRegistry.getResult(registry, family, { suspendOnWaiting: true }).pipe(
198 Effect.ignore,
199 Effect.tap(() => Effect.yieldNow),
200 );
201 registry.mount(families.snapshotAtom(TARGET.environmentId));
202 families.mount(registry, TARGET.environmentId, parent.id);
203
204 // A cached shell from before this session's first snapshot reads nothing.
205 expect(threadFamilyRead(registry.get(family))).toBe("pending");
206 expect(yield* Queue.size(replies)).toBe(0);
207
208 registry.set(shellState, AsyncResult.success(shellAt(1)));
209 yield* Queue.offer(replies, Effect.succeed({ snapshotSequence: 3, threads: [child] }));
210 yield* settle;
211 expect(threadFamilyRead(registry.get(family))).toBe("settled");
212 expect(mergedIds()).toEqual(["parent", "idle-child"]);
213
214 // A new authoritative snapshot never reuses the old read, even while the next one waits
215 // or fails.
216 registry.set(shellState, AsyncResult.success(shellAt(2)));
217 expect(threadFamilyRead(registry.get(family))).toBe("pending");
218 expect(mergedIds()).toEqual(["parent"]);
219 yield* Queue.offer(
220 replies,
221 Effect.fail(new OrchestrationV2GetShellSnapshotError({ message: "read failed" })),
222 );
223 yield* settle;
224 expect(threadFamilyRead(registry.get(family))).toBe("failed");
225 expect(mergedIds()).toEqual(["parent"]);
226
227 families.refresh(registry, TARGET.environmentId, parent.id);
228 expect(threadFamilyRead(registry.get(family))).toBe("pending");
229 yield* Queue.offer(replies, Effect.succeed({ snapshotSequence: 5, threads: [child] }));
230 yield* settle;
231 expect(threadFamilyRead(registry.get(family))).toBe("settled");
232 expect(mergedIds()).toEqual(["parent", "idle-child"]);
233 }),
234 ),
235 );
236});