1import {
2 EnvironmentId,
3 ProjectId,
4 RunId,
5 RuntimeRequestId,
6 ThreadId,
7 type OrchestrationV2ShellSnapshot,
8} from "@t3tools/contracts";
9import * as Option from "effect/Option";
10import { Atom, AtomRegistry } from "effect/reactivity";
11import { describe, expect, it } from "vite-plus/test";
12
13import { PrimaryConnectionTarget } from "../connection/model.ts";
14import { v2Now, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
15import { applyShellStreamEvent } from "./shellReducer.ts";
16import { isThreadWorking } from "./threadInbox.ts";
17import { childThreadInputsTitle, createEnvironmentThreadShellAtoms } from "./threadShell.ts";
18
19const environmentId = EnvironmentId.make("environment-v2");
20const remoteEnvironmentId = EnvironmentId.make("remote-environment-v2");
21const otherProjectId = ProjectId.make("other-project");
22
23function makeHarness(environmentIds: ReadonlyArray<EnvironmentId> = [environmentId]) {
24 const snapshotAtom = Atom.family((_environmentId: EnvironmentId) =>
25 Atom.make<OrchestrationV2ShellSnapshot | null>(v2ShellSnapshot),
26 );
27 const catalogValueAtom = Atom.make({
28 isReady: true,
29 entries: new Map(
30 environmentIds.map((id) => [
31 id,
32 {
33 target: new PrimaryConnectionTarget({
34 environmentId: id,
35 label: "Environment",
36 httpBaseUrl: "https://example.test",
37 wsBaseUrl: "wss://example.test",
38 }),
39 profile: Option.none(),
40 enabled: true,
41 },
42 ]),
43 ),
44 });
45 return {
46 registry: AtomRegistry.make(),
47 snapshotAtom,
48 catalogValueAtom,
49 threads: createEnvironmentThreadShellAtoms({ catalogValueAtom, snapshotAtom }),
50 };
51}
52
53describe("v2 thread shell lists", () => {
54 it("preserves ordered reference arrays when a middle thread changes", () => {
55 const { registry, threads, snapshotAtom } = makeHarness();
56 const snapshot = {
57 ...v2ShellSnapshot,
58 threads: ["a", "b", "c"].map((id) => ({ ...v2ThreadShell, id: ThreadId.make(id) })),
59 };
60 registry.set(snapshotAtom(environmentId), snapshot);
61 const dispose = registry.mount(threads.threadRefsAtom);
62 const before = registry.get(threads.threadRefsAtom);
63 registry.set(
64 snapshotAtom(environmentId),
65 applyShellStreamEvent(snapshot, {
66 kind: "thread.updated",
67 location: "active",
68 sequence: 1,
69 thread: { ...snapshot.threads[1]!, title: "Updated" },
70 }),
71 );
72 expect(registry.get(threads.threadRefsAtom)).toBe(before);
73 dispose();
74 registry.dispose();
75 });
76
77 it("keeps navigation stable on hidden subagent updates and retains user forks", () => {
78 const { registry, threads, snapshotAtom } = makeHarness();
79 const root = v2ThreadShell;
80 const child = {
81 ...root,
82 id: ThreadId.make("child"),
83 lineage: {
84 ...root.lineage,
85 parentThreadId: root.id,
86 relationshipToParent: "subagent" as const,
87 },
88 };
89 const fork = {
90 ...child,
91 id: ThreadId.make("fork"),
92 lineage: { ...child.lineage, relationshipToParent: "fork" as const },
93 };
94 const snapshot = { ...v2ShellSnapshot, threads: [root, child, fork] };
95 registry.set(snapshotAtom(environmentId), snapshot);
96 const dispose = registry.mount(threads.navigationThreadShellsAtom);
97 const before = registry.get(threads.navigationThreadShellsAtom);
98 expect(before.map((thread) => thread.id)).toEqual([root.id, fork.id]);
99 registry.set(snapshotAtom(environmentId), {
100 ...snapshot,
101 threads: [root, { ...child, title: "Child streaming" }, fork],
102 });
103 expect(registry.get(threads.navigationThreadShellsAtom)).toBe(before);
104 expect(registry.get(threads.threadShellsAtom)).toHaveLength(3);
105 dispose();
106 registry.dispose();
107 });
108
109 it("counts a subagent question on its parent only while the subagent runs", () => {
110 const { registry, threads, snapshotAtom } = makeHarness();
111 const root = v2ThreadShell;
112 const child = {
113 ...root,
114 id: ThreadId.make("child"),
115 creationSource: "mcp" as const,
116 status: "running" as const,
117 activityRunStatus: "running" as const,
118 lineage: {
119 ...root.lineage,
120 parentThreadId: root.id,
121 relationshipToParent: "subagent" as const,
122 },
123 pendingRuntimeRequest: {
124 id: RuntimeRequestId.make("question"),
125 kind: "user_input" as const,
126 createdAt: v2Now,
127 },
128 };
129 const inputs = threads.childThreadInputsAtom({ environmentId, threadId: root.id });
130 const dispose = registry.mount(inputs);
131 registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [root, child] });
132 expect(registry.get(inputs).map((thread) => thread.id)).toEqual([child.id]);
133 // A follow-up queued behind the blocked run leaves the question live.
134 registry.set(snapshotAtom(environmentId), {
135 ...v2ShellSnapshot,
136 threads: [root, { ...child, status: "queued" as const }],
137 });
138 expect(registry.get(inputs).map((thread) => thread.id)).toEqual([child.id]);
139 // An async question stays pending after the run that asked it completes.
140 registry.set(snapshotAtom(environmentId), {
141 ...v2ShellSnapshot,
142 threads: [root, { ...child, status: "completed" as const, activityRunStatus: null }],
143 });
144 expect(registry.get(inputs)).toEqual([]);
145 // Older servers omit the activity run, so the latest run's status decides.
146 const { activityRunStatus: _omitted, ...legacyChild } = child;
147 registry.set(snapshotAtom(environmentId), {
148 ...v2ShellSnapshot,
149 threads: [root, { ...legacyChild, status: "completed" as const }],
150 });
151 expect(registry.get(inputs)).toEqual([]);
152 dispose();
153 registry.dispose();
154 });
155
156 it("brings a parent back from Working while a nested subagent waits for approval", () => {
157 const { registry, threads, snapshotAtom } = makeHarness();
158 const child = {
159 ...v2ThreadShell,
160 id: ThreadId.make("review"),
161 creationSource: "mcp" as const,
162 status: "running" as const,
163 activityRunStatus: "running" as const,
164 lineage: {
165 ...v2ThreadShell.lineage,
166 parentThreadId: v2ThreadShell.id,
167 relationshipToParent: "subagent" as const,
168 },
169 };
170 const grandchild = {
171 ...child,
172 id: ThreadId.make("review-helper"),
173 lineage: { ...child.lineage, parentThreadId: child.id },
174 };
175 // The parent's turn ended; the delegated review keeps it in Working.
176 const root = {
177 ...v2ThreadShell,
178 latestRunId: RunId.make("root-run"),
179 status: "completed" as const,
180 pendingBackgroundTasks: [
181 { taskId: "review", kind: "subagent" as const, childThreadId: child.id },
182 ],
183 };
184 const approval = {
185 id: RuntimeRequestId.make("cat-file"),
186 kind: "command" as const,
187 createdAt: v2Now,
188 };
189 const parent = threads.threadShellAtom({ environmentId, threadId: root.id });
190 const inputs = threads.childThreadInputsAtom({ environmentId, threadId: root.id });
191 const disposeParent = registry.mount(parent);
192 const disposeInputs = registry.mount(inputs);
193 registry.set(snapshotAtom(environmentId), {
194 ...v2ShellSnapshot,
195 threads: [root, child, grandchild],
196 });
197 expect(isThreadWorking(registry.get(parent)!)).toBe(true);
198
199 registry.set(snapshotAtom(environmentId), {
200 ...v2ShellSnapshot,
201 threads: [root, child, { ...grandchild, pendingRuntimeRequest: approval }],
202 });
203 expect(registry.get(parent)?.pendingSubagentRequest).toBe("approval");
204 expect(isThreadWorking(registry.get(parent)!)).toBe(false);
205 const asking = registry.get(inputs);
206 expect(asking.map((thread) => thread.id)).toEqual([grandchild.id]);
207 expect(childThreadInputsTitle(asking)).toBe("Subagent needs approval");
208
209 registry.set(snapshotAtom(environmentId), {
210 ...v2ShellSnapshot,
211 threads: [root, child, grandchild],
212 });
213 expect(registry.get(parent)?.pendingSubagentRequest).toBeNull();
214 expect(isThreadWorking(registry.get(parent)!)).toBe(true);
215 disposeInputs();
216 disposeParent();
217 registry.dispose();
218 });
219
220 it("leaves switched-off environments out of navigation", () => {
221 const { registry, threads, catalogValueAtom } = makeHarness([
222 environmentId,
223 remoteEnvironmentId,
224 ]);
225 const dispose = registry.mount(threads.navigationThreadShellsAtom);
226 const environmentsOf = () =>
227 new Set(registry.get(threads.navigationThreadShellsAtom).map((t) => t.environmentId));
228 expect(environmentsOf()).toEqual(new Set([environmentId, remoteEnvironmentId]));
229
230 const setRemoteEnabled = (enabled: boolean) => {
231 const catalog = registry.get(catalogValueAtom);
232 const entries = new Map(catalog.entries);
233 entries.set(remoteEnvironmentId, { ...entries.get(remoteEnvironmentId)!, enabled });
234 registry.set(catalogValueAtom, { ...catalog, entries });
235 };
236 setRemoteEnabled(false);
237 expect(environmentsOf()).toEqual(new Set([environmentId]));
238 setRemoteEnabled(true);
239 expect(environmentsOf()).toEqual(new Set([environmentId, remoteEnvironmentId]));
240 dispose();
241 registry.dispose();
242 });
243
244 it("shares point and list values without retaining an atom for every listed thread", () => {
245 const harness = makeHarness();
246 const snapshot = {
247 ...v2ShellSnapshot,
248 threads: Array.from({ length: 200 }, (_, index) => ({
249 ...v2ThreadShell,
250 id: ThreadId.make(`thread-${index}`),
251 })),
252 };
253 harness.registry.set(harness.snapshotAtom(environmentId), snapshot);
254 const listAtom = harness.threads.threadShellsAtom;
255 const projectListAtom = harness.threads.threadShellsForProjectRefsAtom([
256 { environmentId, projectId: v2ThreadShell.projectId },
257 ]);
258 const disposeList = harness.registry.mount(listAtom);
259 const disposeProjectList = harness.registry.mount(projectListAtom);
260 try {
261 const before = harness.registry.get(listAtom);
262 expect(before).toHaveLength(200);
263 expect(harness.registry.get(projectListAtom)).toEqual(before);
264 expect(harness.registry.getNodes().size).toBeLessThan(20);
265 const firstAtom = harness.threads.threadShellAtom({
266 environmentId,
267 threadId: snapshot.threads[0]!.id,
268 });
269 expect(harness.registry.get(firstAtom)).toBe(before[0]);
270
271 harness.registry.set(harness.snapshotAtom(environmentId), {
272 ...snapshot,
273 snapshotSequence: 1,
274 threads: snapshot.threads.map((thread, index) =>
275 index === 199 ? { ...thread, title: "Updated last thread" } : thread,
276 ),
277 });
278 const after = harness.registry.get(listAtom);
279 expect(after[0]).toBe(before[0]);
280 expect(after.at(-1)).not.toBe(before.at(-1));
281 expect(after.at(-1)?.title).toBe("Updated last thread");
282 expect(harness.registry.get(projectListAtom).at(-1)).toBe(after.at(-1));
283 expect(harness.registry.get(firstAtom)).toBe(after[0]);
284 expect(harness.registry.getNodes().size).toBeLessThan(20);
285 } finally {
286 disposeProjectList();
287 disposeList();
288 harness.registry.dispose();
289 }
290 });
291
292 it("preserves project memberships on updates and keeps the same thread separate per environment", () => {
293 const harness = makeHarness([environmentId, remoteEnvironmentId]);
294 const otherThread = {
295 ...v2ThreadShell,
296 id: ThreadId.make("other-thread"),
297 projectId: otherProjectId,
298 };
299 const snapshot = { ...v2ShellSnapshot, threads: [v2ThreadShell, otherThread] };
300 harness.registry.set(harness.snapshotAtom(environmentId), snapshot);
301 const membershipAtom = harness.threads.environmentThreadRefsByProjectAtom(environmentId);
302 const listAtom = harness.threads.threadShellsForProjectRefsAtom([
303 { environmentId: remoteEnvironmentId, projectId: v2ThreadShell.projectId },
304 { environmentId, projectId: v2ThreadShell.projectId },
305 ]);
306 const disposeList = harness.registry.mount(listAtom);
307 try {
308 const membership = harness.registry.get(membershipAtom);
309 const before = harness.registry.get(listAtom);
310 expect(before).toHaveLength(2);
311 expect(before[0]).not.toBe(before[1]);
312 expect(before.map((thread) => thread.environmentId)).toEqual([
313 remoteEnvironmentId,
314 environmentId,
315 ]);
316
317 harness.registry.set(harness.snapshotAtom(environmentId), {
318 ...snapshot,
319 threads: [v2ThreadShell, { ...otherThread, title: "Changed elsewhere" }],
320 });
321 expect(harness.registry.get(membershipAtom)).toBe(membership);
322 expect(harness.registry.get(listAtom)).toBe(before);
323
324 harness.registry.set(harness.snapshotAtom(environmentId), {
325 ...snapshot,
326 threads: [{ ...v2ThreadShell, projectId: otherProjectId }, otherThread],
327 });
328 expect(harness.registry.get(listAtom)).toEqual([before[0]]);
329 harness.registry.set(harness.snapshotAtom(remoteEnvironmentId), {
330 ...v2ShellSnapshot,
331 threads: [],
332 });
333 expect(harness.registry.get(listAtom)).toEqual([]);
334 } finally {
335 disposeList();
336 harness.registry.dispose();
337 }
338 });
339});