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.gen(function* () {
123 const replies =
124 yield* Queue.unbounded<
125 Effect.Effect<OrchestrationV2ThreadFamily, OrchestrationV2GetShellSnapshotError>
126 >();
127 const client = {
128 [ORCHESTRATION_V2_WS_METHODS.getThreadFamily]: () => Effect.flatten(Queue.take(replies)),
129 } as unknown as WsRpcProtocolClient;
130 const session = {
131 client,
132 initialConfig: Effect.succeed({ shellThreadFamilies: true } as never),
133 ready: Effect.void,
134 probe: Effect.void,
135 closed: Effect.never,
136 } as unknown as RpcSession.RpcSession;
137 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
138 target: TARGET,
139 state: yield* SubscriptionRef.make<SupervisorConnectionState>({
140 ...AVAILABLE_CONNECTION_STATE,
141 desired: true,
142 network: "online",
143 phase: "connected",
144 attempt: 1,
145 generation: 1,
146 }),
147 session: yield* SubscriptionRef.make(Option.some(session)),
148 prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
149 connect: Effect.void,
150 disconnect: Effect.void,
151 retryNow: Effect.void,
152 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
153 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (
154 _environmentId,
155 effect,
156 ) => Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
157 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
158 _environmentId,
159 stream,
160 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
161 const runtime = Atom.runtime(
162 Layer.succeed(
163 EnvironmentRegistry.EnvironmentRegistry,
164 EnvironmentRegistry.EnvironmentRegistry.of({
165 run,
166 followStream,
167 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
168 ),
169 );
170 const parent = row("parent");
171 const child = row("idle-child");
172 const shellSnapshot = {
173 schemaVersion: 1,
174 snapshotSequence: 3,
175 projects: [],
176 threads: [parent],
177 archivedThreads: [],
178 } as OrchestrationV2ShellSnapshot;
179 const shellAt = (generation: number): EnvironmentShellState => ({
180 ...EMPTY_SHELL_STATE,
181 status: "live",
182 snapshot: Option.some(shellSnapshot),
183 generation,
184 });
185 const shellState = Atom.make<AsyncResult.AsyncResult<EnvironmentShellState>>(
186 AsyncResult.success(shellAt(0)),
187 );
188 const families = createThreadFamilyAtoms(runtime, () => shellState);
189 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
190 Effect.sync(() => registry.dispose()),
191 );
192 const family = families.familyAtom(TARGET.environmentId, parent.id);
193 const mergedIds = () =>
194 registry.get(families.snapshotAtom(TARGET.environmentId))?.threads.map(({ id }) => id);
195 // Waits for the read to finish either way; the assertions read its outcome.
196 const settle = AtomRegistry.getResult(registry, family, { suspendOnWaiting: true }).pipe(
197 Effect.ignore,
198 Effect.tap(() => Effect.yieldNow),
199 );
200 registry.mount(families.snapshotAtom(TARGET.environmentId));
201 families.mount(registry, TARGET.environmentId, parent.id);
202
203 // A cached shell from before this session's first snapshot reads nothing.
204 expect(threadFamilyRead(registry.get(family))).toBe("pending");
205 expect(yield* Queue.size(replies)).toBe(0);
206
207 registry.set(shellState, AsyncResult.success(shellAt(1)));
208 yield* Queue.offer(replies, Effect.succeed({ snapshotSequence: 3, threads: [child] }));
209 yield* settle;
210 expect(threadFamilyRead(registry.get(family))).toBe("settled");
211 expect(mergedIds()).toEqual(["parent", "idle-child"]);
212
213 // A new authoritative snapshot never reuses the old read, even while the next one waits
214 // or fails.
215 registry.set(shellState, AsyncResult.success(shellAt(2)));
216 expect(threadFamilyRead(registry.get(family))).toBe("pending");
217 expect(mergedIds()).toEqual(["parent"]);
218 yield* Queue.offer(
219 replies,
220 Effect.fail(new OrchestrationV2GetShellSnapshotError({ message: "read failed" })),
221 );
222 yield* settle;
223 expect(threadFamilyRead(registry.get(family))).toBe("failed");
224 expect(mergedIds()).toEqual(["parent"]);
225
226 families.refresh(registry, TARGET.environmentId, parent.id);
227 expect(threadFamilyRead(registry.get(family))).toBe("pending");
228 yield* Queue.offer(replies, Effect.succeed({ snapshotSequence: 5, threads: [child] }));
229 yield* settle;
230 expect(threadFamilyRead(registry.get(family))).toBe("settled");
231 expect(mergedIds()).toEqual(["parent", "idle-child"]);
232 }),
233 );
234});