1import {
2 EnvironmentId,
3 NodeId,
4 ProviderDriverKind,
5 RuntimeRequestId,
6 TurnItemId,
7 type OrchestrationV2ThreadProjection,
8 ThreadId,
9 RunId,
10 MessageId,
11 type ScopedThreadRef,
12} from "@t3tools/contracts";
13import * as Option from "effect/Option";
14import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
15import { describe, expect, it } from "vite-plus/test";
16
17import { v2Projection, v2Now } from "./orchestrationV2TestFixtures.ts";
18import { EMPTY_THREAD_HISTORY_META } from "./threadHistoryMerge.ts";
19import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts";
20import type { EnvironmentThreadState } from "./threads.ts";
21
22const ref: ScopedThreadRef = {
23 environmentId: EnvironmentId.make("environment-detail"),
24 threadId: ThreadId.make(v2Projection.thread.id),
25};
26
27function threadState(
28 partial: Pick<EnvironmentThreadState, "data" | "status" | "error"> &
29 Partial<Pick<EnvironmentThreadState, "history">>,
30): EnvironmentThreadState {
31 return { ...partial, history: partial.history ?? EMPTY_THREAD_HISTORY_META };
32}
33
34describe("createEnvironmentThreadDetailAtoms", () => {
35 it("adds environment scope while preserving the pristine projection", () => {
36 const initial: AsyncResult.AsyncResult<EnvironmentThreadState, never> = AsyncResult.success(
37 threadState({
38 data: Option.some(v2Projection),
39 status: "cached",
40 error: Option.none(),
41 }),
42 );
43 const sourceAtom = Atom.make(initial);
44 const details = createEnvironmentThreadDetailAtoms(() => sourceAtom);
45 const registry = AtomRegistry.make();
46
47 const thread = registry.get(details.threadAtom(ref));
48 expect(thread).toEqual({ environmentId: ref.environmentId, projection: v2Projection });
49 expect(thread?.projection).toBe(v2Projection);
50 expect(registry.get(details.visibleTurnItemsAtom(ref))).toBe(v2Projection.visibleTurnItems);
51 expect(registry.get(details.statusAtom(ref))).toBe("cached");
52 expect(registry.get(details.historyAtom(ref))).toBe(EMPTY_THREAD_HISTORY_META);
53
54 registry.set(
55 sourceAtom,
56 AsyncResult.success(
57 threadState({
58 data: Option.some(v2Projection),
59 status: "synchronizing",
60 error: Option.none(),
61 }),
62 ),
63 );
64
65 expect(registry.get(details.threadAtom(ref))).toBe(thread);
66 expect(registry.get(details.statusAtom(ref))).toBe("synchronizing");
67
68 const progressiveHistory = {
69 ...EMPTY_THREAD_HISTORY_META,
70 historyCursor: "cursor-1",
71 hasMoreHistory: true,
72 };
73 registry.set(
74 sourceAtom,
75 AsyncResult.success(
76 threadState({
77 data: Option.some(v2Projection),
78 status: "live",
79 error: Option.some("Stream interrupted."),
80 history: progressiveHistory,
81 }),
82 ),
83 );
84
85 expect(registry.get(details.threadAtom(ref))).toBe(thread);
86 expect(registry.get(details.visibleTurnItemsAtom(ref))).toBe(v2Projection.visibleTurnItems);
87 expect(registry.get(details.statusAtom(ref))).toBe("live");
88 expect(registry.get(details.errorAtom(ref))).toBe("Stream interrupted.");
89 expect(registry.get(details.historyAtom(ref))).toBe(progressiveHistory);
90
91 registry.set(
92 sourceAtom,
93 AsyncResult.success(
94 threadState({
95 data: Option.none(),
96 status: "deleted",
97 error: Option.none(),
98 }),
99 ),
100 );
101
102 expect(registry.get(details.threadAtom(ref))).toBeNull();
103 expect(registry.get(details.visibleTurnItemsAtom(ref))).toEqual([]);
104 expect(registry.get(details.statusAtom(ref))).toBe("deleted");
105 expect(registry.get(details.errorAtom(ref))).toBeNull();
106 expect(registry.get(details.historyAtom(ref))).toBe(EMPTY_THREAD_HISTORY_META);
107 registry.dispose();
108 });
109});
110
111it("does not notify queue consumers for streaming text or execution nodes", () => {
112 const projection = {
113 ...v2Projection,
114 runs: [
115 {
116 id: RunId.make("queued-run"),
117 threadId: v2Projection.thread.id,
118 ordinal: 1,
119 providerInstanceId: v2Projection.thread.providerInstanceId,
120 modelSelection: v2Projection.thread.modelSelection,
121 providerThreadId: null,
122 userMessageId: MessageId.make("queued-message"),
123 rootNodeId: null,
124 activeAttemptId: null,
125 status: "queued" as const,
126 requestedAt: v2Now,
127 startedAt: null,
128 completedAt: null,
129 checkpointId: null,
130 contextHandoffId: null,
131 },
132 ],
133 };
134 const source = Atom.make(
135 AsyncResult.success(
136 threadState({ data: Option.some(projection), status: "live", error: Option.none() }),
137 ),
138 );
139 const details = createEnvironmentThreadDetailAtoms(() => source);
140 const registry = AtomRegistry.make();
141 const dispose = registry.mount(details.queueWorkflowAtom(ref));
142 const before = registry.get(details.queueWorkflowAtom(ref));
143 const disposeCount = registry.mount(details.queuedCountAtom(ref));
144 expect(registry.get(details.queuedCountAtom(ref))).toBe(1);
145 registry.set(
146 source,
147 AsyncResult.success(
148 threadState({
149 data: Option.some({
150 ...projection,
151 nodes: [...v2Projection.nodes],
152 turnItems: [...v2Projection.turnItems],
153 }),
154 status: "live",
155 error: Option.none(),
156 }),
157 ),
158 );
159 expect(registry.get(details.queueWorkflowAtom(ref))).toBe(before);
160 expect(registry.get(details.queuedCountAtom(ref))).toBe(1);
161 registry.set(
162 source,
163 AsyncResult.success(
164 threadState({ data: Option.none(), status: "deleted", error: Option.none() }),
165 ),
166 );
167 expect(registry.get(details.queueWorkflowAtom(ref))).toBeNull();
168 expect(registry.get(details.queuedCountAtom(ref))).toBe(0);
169 disposeCount();
170 dispose();
171 registry.dispose();
172});
173
174it("does not notify the agents pill for unrelated projection updates", () => {
175 const projection: OrchestrationV2ThreadProjection = {
176 ...v2Projection,
177 runs: [
178 {
179 id: RunId.make("active-run"),
180 threadId: v2Projection.thread.id,
181 ordinal: 1,
182 providerInstanceId: v2Projection.thread.providerInstanceId,
183 modelSelection: v2Projection.thread.modelSelection,
184 providerThreadId: null,
185 userMessageId: MessageId.make("message"),
186 rootNodeId: null,
187 activeAttemptId: null,
188 status: "running" as const,
189 requestedAt: v2Now,
190 startedAt: v2Now,
191 completedAt: null,
192 checkpointId: null,
193 contextHandoffId: null,
194 },
195 ],
196 subagents: [
197 {
198 id: NodeId.make("subagent-1"),
199 threadId: v2Projection.thread.id,
200 runId: RunId.make("active-run"),
201 parentNodeId: NodeId.make("root"),
202 origin: "app_owned",
203 createdBy: "agent",
204 driver: ProviderDriverKind.make("codex"),
205 providerInstanceId: v2Projection.thread.providerInstanceId,
206 providerThreadId: null,
207 childThreadId: ThreadId.make("thread-child"),
208 nativeTaskRef: null,
209 prompt: "Do the thing",
210 title: "Worker",
211 model: null,
212 status: "running",
213 result: null,
214 startedAt: v2Now,
215 completedAt: null,
216 updatedAt: v2Now,
217 },
218 ],
219 };
220 const source = Atom.make(
221 AsyncResult.success(
222 threadState({ data: Option.some(projection), status: "live", error: Option.none() }),
223 ),
224 );
225 const details = createEnvironmentThreadDetailAtoms(() => source);
226 const registry = AtomRegistry.make();
227 const dispose = registry.mount(details.turnSubagentsAtom(ref));
228 const before = registry.get(details.turnSubagentsAtom(ref));
229 expect(before?.liveCount).toBe(1);
230 registry.set(
231 source,
232 AsyncResult.success(
233 threadState({
234 data: Option.some({ ...projection, turnItems: [...v2Projection.turnItems] }),
235 status: "live",
236 error: Option.none(),
237 }),
238 ),
239 );
240 expect(registry.get(details.turnSubagentsAtom(ref))).toBe(before);
241 dispose();
242 registry.dispose();
243});
244
245it("keeps worktree and pending requests stable during output updates, then reflects resolution", () => {
246 const projection: OrchestrationV2ThreadProjection = {
247 ...v2Projection,
248 thread: { ...v2Projection.thread, worktreePath: "/workspace/task" },
249 runtimeRequests: [
250 {
251 id: RuntimeRequestId.make("approval"),
252 nodeId: NodeId.make("approval-node"),
253 providerTurnId: null,
254 nativeRequestRef: null,
255 kind: "command",
256 status: "pending",
257 responseCapability: { type: "not_resumable", reason: "Provider exited" },
258 createdAt: v2Now,
259 resolvedAt: null,
260 },
261 ],
262 };
263 const source = Atom.make(
264 AsyncResult.success(
265 threadState({ data: Option.some(projection), status: "live", error: Option.none() }),
266 ),
267 );
268 const details = createEnvironmentThreadDetailAtoms(() => source);
269 const registry = AtomRegistry.make();
270 const dispose = registry.mount(details.pendingRequestsAtom(ref));
271 const before = registry.get(details.pendingRequestsAtom(ref));
272 expect(before?.approvals).toHaveLength(1);
273 const streaming: OrchestrationV2ThreadProjection = {
274 ...projection,
275 turnItems: [
276 {
277 id: TurnItemId.make("output"),
278 threadId: projection.thread.id,
279 runId: null,
280 nodeId: null,
281 providerThreadId: null,
282 providerTurnId: null,
283 nativeItemRef: null,
284 parentItemId: null,
285 ordinal: 1,
286 status: "running",
287 title: null,
288 startedAt: v2Now,
289 completedAt: null,
290 updatedAt: v2Now,
291 type: "command_execution",
292 input: "pwd",
293 output: "new output",
294 exitCode: undefined,
295 },
296 ],
297 };
298 registry.set(
299 source,
300 AsyncResult.success(
301 threadState({ data: Option.some(streaming), status: "live", error: Option.none() }),
302 ),
303 );
304 expect(registry.get(details.pendingRequestsAtom(ref))).toBe(before);
305 expect(registry.get(details.worktreePathAtom(ref))).toBe("/workspace/task");
306 registry.set(
307 source,
308 AsyncResult.success(
309 threadState({
310 data: Option.some({
311 ...streaming,
312 thread: { ...projection.thread, worktreePath: "/workspace/moved" },
313 runtimeRequests: projection.runtimeRequests.map((request) => ({
314 ...request,
315 status: "resolved",
316 resolvedAt: v2Now,
317 })),
318 }),
319 status: "live",
320 error: Option.none(),
321 }),
322 ),
323 );
324 expect(registry.get(details.pendingRequestsAtom(ref))?.approvals).toEqual([]);
325 expect(registry.get(details.worktreePathAtom(ref))).toBe("/workspace/moved");
326 dispose();
327 registry.dispose();
328});