1import {
2 CheckpointId,
3 CheckpointRef,
4 CheckpointScopeId,
5 EnvironmentId,
6 EventId,
7 MessageId,
8 NodeId,
9 RunId,
10 TurnItemId,
11 type OrchestrationV2Checkpoint,
12 type OrchestrationV2ConversationMessage,
13 type OrchestrationV2Run,
14 type OrchestrationV2ThreadProjection,
15 type OrchestrationV2TurnItem,
16} from "@t3tools/contracts";
17import { describe, expect, it } from "@effect/vitest";
18import * as DateTime from "effect/DateTime";
19import * as Option from "effect/Option";
20import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
21
22import { v2Now, v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts";
23import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts";
24import type { ThreadStructureVisibleTurnItem } from "./models.ts";
25import {
26 createEnvironmentThreadDetailAtoms,
27 deriveWorkspaceMutationId,
28 sameThreadStructure,
29} from "./threadDetail.ts";
30import { mergeOlderHistoryIntoProjection } from "./threadHistoryMerge.ts";
31import { EMPTY_ENVIRONMENT_THREAD_STATE } from "./threadState.ts";
32
33const later = DateTime.makeUnsafe("2026-06-20T00:00:01.000Z");
34
35const assistant: OrchestrationV2TurnItem = {
36 id: TurnItemId.make("item-assistant"),
37 threadId: v2ThreadId,
38 runId: RunId.make("run-1"),
39 nodeId: null,
40 providerThreadId: null,
41 providerTurnId: null,
42 nativeItemRef: null,
43 parentItemId: null,
44 ordinal: 1,
45 status: "running",
46 title: null,
47 startedAt: v2Now,
48 completedAt: null,
49 updatedAt: v2Now,
50 type: "assistant_message",
51 messageId: MessageId.make("message-assistant"),
52 text: "Hel",
53 streaming: true,
54};
55
56function withItem(item: OrchestrationV2TurnItem, updatedAt = v2Now) {
57 return {
58 ...v2Projection,
59 turnItems: [item],
60 visibleTurnItems: [
61 {
62 position: 0,
63 visibility: "local" as const,
64 sourceThreadId: v2ThreadId,
65 sourceItemId: item.id,
66 item,
67 },
68 ],
69 updatedAt,
70 };
71}
72
73describe("sameThreadStructure", () => {
74 const before = withItem(assistant);
75
76 it("holds through streamed text", () => {
77 const next = withItem({ ...assistant, text: "Hello", updatedAt: later }, later);
78 expect(sameThreadStructure(before, next)).toBe(true);
79 });
80
81 it("sees the item finish streaming", () => {
82 const next = withItem({ ...assistant, text: "Hello", streaming: false, status: "completed" });
83 expect(sameThreadStructure(before, next)).toBe(false);
84 });
85
86 it("sees a new item", () => {
87 const next = { ...before, turnItems: [assistant, { ...assistant, id: TurnItemId.make("b") }] };
88 expect(sameThreadStructure(before, next)).toBe(false);
89 });
90
91 it("sees other fields change on a streaming item", () => {
92 const next = withItem({ ...assistant, text: "Hello", title: "Answer" });
93 expect(sameThreadStructure(before, next)).toBe(false);
94 });
95
96 it("holds through a visit mark", () => {
97 const thread = { ...before.thread, lastVisitedAt: later, updatedAt: later };
98 expect(sameThreadStructure(before, { ...before, thread })).toBe(true);
99 });
100
101 it("sees other thread changes", () => {
102 const thread = { ...before.thread, title: "Renamed" };
103 expect(sameThreadStructure(before, { ...before, thread })).toBe(false);
104 });
105
106 it("sees changes outside the turn items", () => {
107 expect(sameThreadStructure(before, { ...before, runs: [] })).toBe(false);
108 });
109});
110
111describe("lastVisitedAtAtom", () => {
112 it("follows a visit mark the quiet thread holds through", () => {
113 const projection = withItem(assistant);
114 const state = Atom.make(
115 AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(projection) }),
116 );
117 const details = createEnvironmentThreadDetailAtoms(() => state);
118 const ref = { environmentId: EnvironmentId.make("env"), threadId: v2ThreadId };
119 const registry = AtomRegistry.make();
120 const unmountQuiet = registry.mount(details.quietThreadAtom(ref));
121 const unmountVisit = registry.mount(details.lastVisitedAtAtom(ref));
122 const quietBefore = registry.get(details.quietThreadAtom(ref));
123
124 const visited = { ...projection, thread: { ...projection.thread, lastVisitedAt: later } };
125 registry.set(
126 state,
127 AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(visited) }),
128 );
129
130 expect(registry.get(details.quietThreadAtom(ref))).toBe(quietBefore);
131 expect(registry.get(details.lastVisitedAtAtom(ref))).toBe(later);
132 unmountQuiet();
133 unmountVisit();
134 registry.dispose();
135 });
136});
137
138describe("quietThreadAtom", () => {
139 const ref = { environmentId: EnvironmentId.make("environment-1"), threadId: v2ThreadId };
140
141 function setup(initial: OrchestrationV2ThreadProjection) {
142 const stateAtom = Atom.make(
143 AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(initial) }),
144 );
145 const quietAtom = createEnvironmentThreadDetailAtoms(() => stateAtom).quietThreadAtom(ref);
146 const registry = AtomRegistry.make();
147 registry.mount(quietAtom);
148 return {
149 read: () => registry.get(quietAtom),
150 [Symbol.dispose]: () => registry.dispose(),
151 push: (projection: OrchestrationV2ThreadProjection | null) =>
152 registry.set(
153 stateAtom,
154 AsyncResult.success({
155 ...EMPTY_ENVIRONMENT_THREAD_STATE,
156 data: Option.fromNullishOr(projection),
157 }),
158 ),
159 };
160 }
161
162 it("keeps the same value through streamed text", () => {
163 const thread = setup(withItem(assistant));
164 const first = thread.read();
165 thread.push(withItem({ ...assistant, text: "Hello", updatedAt: later }, later));
166 thread.push(withItem({ ...assistant, text: "Hello there", updatedAt: later }, later));
167 expect(thread.read()).toBe(first);
168 });
169
170 it("passes a structure change on", () => {
171 const thread = setup(withItem(assistant));
172 const first = thread.read();
173 thread.push(withItem({ ...assistant, text: "Hello", updatedAt: later }, later));
174 thread.push(withItem({ ...assistant, text: "Hello", status: "completed", streaming: false }));
175 const next = thread.read();
176 expect(next).not.toBe(first);
177 expect(next?.projection.turnItems[0]).toMatchObject({ status: "completed", streaming: false });
178 });
179
180 it("discards structure held before the thread resets", () => {
181 const initial = withItem(assistant);
182 using thread = setup(initial);
183 const first = thread.read();
184 thread.push(replaceItem(initial, { ...assistant, text: "Hello" }));
185 expect(thread.read()).toBe(first);
186 thread.push(null);
187 expect(thread.read()).toBeNull();
188 const reset = withItem({ ...assistant, title: "Restored", text: "New snapshot" });
189 thread.push(reset);
190 expect(thread.read()).not.toBe(first);
191 expect(thread.read()?.projection.turnItems[0]).toMatchObject({ title: "Restored" });
192 thread.push(replaceItem(reset, { ...assistant, title: "Restored", text: "More text" }));
193 expect(thread.read()?.projection.turnItems[0]).toMatchObject({ title: "Restored" });
194 });
195
196 it("leaves out the fields that streaming moves", () => {
197 const thread = setup(withItem(assistant));
198 const projection = thread.read()?.projection;
199 expect(projection).not.toHaveProperty("updatedAt");
200 expect(projection?.thread).not.toHaveProperty("updatedAt");
201 expect(projection?.thread).not.toHaveProperty("lastVisitedAt");
202 expect(projection?.turnItems[0]).not.toHaveProperty("text");
203 expect(projection?.turnItems[0]).not.toHaveProperty("updatedAt");
204 expect(projection?.visibleTurnItems[0]?.item).not.toHaveProperty("text");
205 expect(projection?.visibleTurnItems[0]?.item).toMatchObject({
206 messageId: assistant.messageId,
207 streaming: true,
208 });
209 });
210});
211
212describe("timelineVisibleTurnItemsAtom", () => {
213 const ref = { environmentId: EnvironmentId.make("environment-1"), threadId: v2ThreadId };
214
215 it("holds through streamed assistant text while the message reads its own text", () => {
216 const stateAtom = Atom.make(
217 AsyncResult.success({
218 ...EMPTY_ENVIRONMENT_THREAD_STATE,
219 data: Option.some<OrchestrationV2ThreadProjection>(withItem(assistant)),
220 }),
221 );
222 const details = createEnvironmentThreadDetailAtoms(() => stateAtom);
223 const timelineAtom = details.timelineVisibleTurnItemsAtom(ref);
224 const textAtom = details.assistantMessageTextAtom(ref, assistant.id);
225 const registry = AtomRegistry.make();
226 registry.mount(timelineAtom);
227 registry.mount(textAtom);
228 const push = (item: OrchestrationV2TurnItem) =>
229 registry.set(
230 stateAtom,
231 AsyncResult.success({
232 ...EMPTY_ENVIRONMENT_THREAD_STATE,
233 data: Option.some(withItem(item, later)),
234 }),
235 );
236 const first = registry.get(timelineAtom);
237
238 push({ ...assistant, text: "Hello", updatedAt: later });
239 expect(registry.get(timelineAtom)).toBe(first);
240 expect(registry.get(textAtom)).toBe("Hello");
241
242 push({ ...assistant, text: "Hello", streaming: false, status: "completed" });
243 expect(registry.get(timelineAtom)[0]?.item).toMatchObject({ text: "Hello", streaming: false });
244 registry.dispose();
245 });
246});
247
248it("only compares replaced entries in a streamed reducer delta", () => {
249 const size = 100;
250 let slotReads = 0;
251 const countReads = <T>(items: ReadonlyArray<T>): ReadonlyArray<T> =>
252 new Proxy(items, {
253 get(target, key, receiver) {
254 if (typeof key === "string" && /^\d+$/.test(key)) slotReads++;
255 return Reflect.get(target, key, receiver);
256 },
257 });
258 const items = Array.from({ length: size }, (_, index) => ({
259 ...assistant,
260 id: TurnItemId.make(`item-${index}`),
261 ordinal: index,
262 }));
263 const before = {
264 ...v2Projection,
265 turnItems: countReads(items),
266 visibleTurnItems: countReads(
267 items.map((item, position) => ({
268 item,
269 position,
270 visibility: "local" as const,
271 sourceThreadId: v2ThreadId,
272 sourceItemId: item.id,
273 })),
274 ),
275 };
276 const next = applyOrchestrationV2ProjectionEvent(before, {
277 id: EventId.make("text-delta"),
278 type: "turn-item.updated",
279 threadId: v2ThreadId,
280 occurredAt: later,
281 payload: { ...items.at(-1)!, text: "Hello", updatedAt: later },
282 });
283 slotReads = 0;
284
285 expect(sameThreadStructure(before, next!)).toBe(true);
286 expect(slotReads).toBeLessThanOrEqual(2);
287});
288
289describe("derived thread state through a streamed message", () => {
290 const ref = { environmentId: EnvironmentId.make("environment-1"), threadId: v2ThreadId };
291 const runId = RunId.make("run-1");
292 const streamed: OrchestrationV2ConversationMessage = {
293 id: MessageId.make("message-assistant"),
294 threadId: v2ThreadId,
295 runId,
296 nodeId: null,
297 role: "assistant",
298 text: "Hel",
299 attachments: [],
300 streaming: true,
301 createdBy: "agent",
302 creationSource: "provider",
303 createdAt: v2Now,
304 updatedAt: v2Now,
305 };
306 const queuedMessage: OrchestrationV2ConversationMessage = {
307 ...streamed,
308 id: MessageId.make("message-queued"),
309 runId: RunId.make("run-2"),
310 role: "user",
311 text: "Then run the tests",
312 streaming: false,
313 createdBy: "user",
314 creationSource: "mobile",
315 };
316 const queuedRun: OrchestrationV2Run = {
317 id: RunId.make("run-2"),
318 threadId: v2ThreadId,
319 ordinal: 2,
320 providerInstanceId: v2Projection.thread.providerInstanceId,
321 modelSelection: v2Projection.thread.modelSelection,
322 providerThreadId: null,
323 userMessageId: queuedMessage.id,
324 rootNodeId: null,
325 activeAttemptId: null,
326 status: "queued",
327 requestedAt: v2Now,
328 startedAt: null,
329 completedAt: null,
330 checkpointId: null,
331 contextHandoffId: null,
332 };
333 const checkpoint = (ordinal: number): OrchestrationV2Checkpoint => ({
334 id: CheckpointId.make(`checkpoint-${ordinal}`),
335 threadId: v2ThreadId,
336 scopeId: CheckpointScopeId.make("scope"),
337 runId,
338 nodeId: NodeId.make("node"),
339 parentCheckpointId: null,
340 ordinalWithinScope: ordinal,
341 appRunOrdinal: ordinal,
342 ref: CheckpointRef.make(`refs/t3/checkpoint-${ordinal}`),
343 status: "ready",
344 files: [],
345 capturedAt: v2Now,
346 });
347 const initial: OrchestrationV2ThreadProjection = {
348 ...v2Projection,
349 runs: [queuedRun],
350 messages: [streamed, queuedMessage],
351 checkpoints: [checkpoint(1)],
352 };
353
354 function setup() {
355 const stateAtom = Atom.make(
356 AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(initial) }),
357 );
358 const details = createEnvironmentThreadDetailAtoms(() => stateAtom);
359 const queue = details.queueWorkflowAtom(ref);
360 const checkpoints = details.checkpointSummariesAtom(ref);
361 const registry = AtomRegistry.make();
362 registry.mount(queue);
363 registry.mount(checkpoints);
364 let projection = initial;
365 return {
366 queue: () => registry.get(queue),
367 checkpoints: () => registry.get(checkpoints),
368 apply: (event: Parameters<typeof applyOrchestrationV2ProjectionEvent>[1]) => {
369 projection = applyOrchestrationV2ProjectionEvent(projection, event)!;
370 registry.set(
371 stateAtom,
372 AsyncResult.success({ ...EMPTY_ENVIRONMENT_THREAD_STATE, data: Option.some(projection) }),
373 );
374 },
375 [Symbol.dispose]: () => registry.dispose(),
376 };
377 }
378
379 const messageUpdated = (message: OrchestrationV2ConversationMessage) => ({
380 id: EventId.make(`event-${message.id}-${message.text.length}`),
381 type: "message.updated" as const,
382 threadId: v2ThreadId,
383 occurredAt: later,
384 payload: message,
385 });
386
387 it("keeps the queue workflow through streamed text and follows a queued message edit", () => {
388 using thread = setup();
389 const first = thread.queue();
390 thread.apply(messageUpdated({ ...streamed, text: "Hello", updatedAt: later }));
391 thread.apply(messageUpdated({ ...streamed, text: "Hello there", updatedAt: later }));
392 expect(thread.queue()).toBe(first);
393
394 thread.apply(messageUpdated({ ...queuedMessage, text: "Then run lint", updatedAt: later }));
395 expect(thread.queue()?.queuedRuns.map((queued) => queued.text)).toEqual(["Then run lint"]);
396 });
397
398 it("keeps checkpoint summaries through streamed text and follows new assistant messages and checkpoints", () => {
399 using thread = setup();
400 const first = thread.checkpoints();
401 thread.apply(messageUpdated({ ...streamed, text: "Hello", updatedAt: later }));
402 thread.apply(messageUpdated({ ...streamed, text: "Hello there", updatedAt: later }));
403 expect(thread.checkpoints()).toBe(first);
404
405 const followUp = { ...streamed, id: MessageId.make("message-follow-up"), text: "Done" };
406 thread.apply(messageUpdated(followUp));
407 expect(thread.checkpoints().map((summary) => summary.assistantMessageId)).toEqual([
408 followUp.id,
409 ]);
410
411 thread.apply({
412 id: EventId.make("checkpoint-2"),
413 type: "checkpoint.captured",
414 threadId: v2ThreadId,
415 occurredAt: later,
416 payload: checkpoint(2),
417 });
418 expect(thread.checkpoints().map((summary) => summary.checkpointTurnCount)).toEqual([1, 2]);
419 });
420});
421
422function replaceItem(projection: OrchestrationV2ThreadProjection, item: OrchestrationV2TurnItem) {
423 return applyOrchestrationV2ProjectionEvent(projection, {
424 id: EventId.make(`event-${item.id}`),
425 type: "turn-item.updated",
426 threadId: v2ThreadId,
427 occurredAt: item.updatedAt,
428 payload: item,
429 })!;
430}
431
432it("matches full structure comparisons across streamed, batched, and reordered updates", () => {
433 const message = { ...assistant, ordinal: 2 };
434 const reasoning = {
435 ...assistant,
436 id: TurnItemId.make("reasoning"),
437 type: "reasoning" as const,
438 ordinal: 1,
439 };
440 const initial = mergeOlderHistoryIntoProjection(withItem(message), [
441 withItem(reasoning).visibleTurnItems[0]!,
442 ]);
443 const updates = [
444 { ...message, text: "Hello", updatedAt: later },
445 { ...reasoning, text: "Thinking", updatedAt: later },
446 { ...reasoning, text: "Done", status: "completed" as const, streaming: false },
447 { ...message, text: "Hello there", updatedAt: later },
448 { ...message, text: "Hello there", ordinal: 0 },
449 { ...message, text: "More text", ordinal: 0 },
450 { ...message, text: "More text", ordinal: 2 },
451 { ...message, text: "More text", runId: RunId.make("another-run") },
452 ];
453 const projections = [initial];
454 const withoutHints = (projection: OrchestrationV2ThreadProjection) => ({
455 ...projection,
456 turnItems: projection.turnItems.slice(),
457 visibleTurnItems: projection.visibleTurnItems.slice(),
458 });
459 const consecutive: boolean[] = [];
460 for (const item of updates) {
461 const previous = projections.at(-1)!;
462 const next = replaceItem(previous, item);
463 consecutive.push(sameThreadStructure(previous, next));
464 // All earlier snapshots include readers that skip one or several updates in a batch.
465 for (const before of projections) {
466 expect(sameThreadStructure(before, next)).toBe(
467 sameThreadStructure(withoutHints(before), withoutHints(next)),
468 );
469 }
470 projections.push(next);
471 }
472
473 // A resumed snapshot can reuse item ids while changing a different row.
474 const snapshot = withoutHints(projections[1]!);
475 const resumed = replaceItem(snapshot, { ...reasoning, title: "Restored reasoning" });
476 const sibling = replaceItem(initial, { ...message, text: "Another stream" });
477 const reconnected = replaceItem(resumed, { ...message, text: "Resumed stream" });
478 for (const before of [initial, sibling, projections[1]!, resumed]) {
479 expect(sameThreadStructure(before, reconnected)).toBe(
480 sameThreadStructure(withoutHints(before), withoutHints(reconnected)),
481 );
482 }
483 expect(sameThreadStructure(sibling, reconnected)).toBe(false);
484 expect(sameThreadStructure(resumed, reconnected)).toBe(true);
485
486 expect(consecutive).toEqual([true, true, false, true, false, true, false, false]);
487 expect(sameThreadStructure(initial, projections[4]!)).toBe(false);
488 expect(projections[5]!.visibleTurnItems.map((row) => [row.item.id, row.position])).toEqual([
489 [assistant.id, 0],
490 [reasoning.id, 1],
491 ]);
492 expect(projections[7]!.visibleTurnItems.map((row) => [row.item.id, row.position])).toEqual([
493 [reasoning.id, 0],
494 [assistant.id, 1],
495 ]);
496});
497
498describe("deriveWorkspaceMutationId", () => {
499 const item = (id: string, type: string, status: string) =>
500 ({
501 item: { id: TurnItemId.make(id), type, status },
502 }) as unknown as ThreadStructureVisibleTurnItem;
503
504 it("keys on the latest finished command or file change and the latest checkpoint", () => {
505 expect(
506 deriveWorkspaceMutationId(
507 [
508 item("edit", "file_change", "completed"),
509 item("reply", "assistant_message", "completed"),
510 item("command", "command_execution", "running"),
511 ],
512 "2026-01-01T00:00:00.000Z",
513 ),
514 ).toBe(JSON.stringify(["edit", "2026-01-01T00:00:00.000Z"]));
515 expect(
516 deriveWorkspaceMutationId(
517 [item("edit", "file_change", "completed"), item("mcp", "dynamic_tool", "completed")],
518 null,
519 ),
520 ).toBe(JSON.stringify(["mcp", null]));
521 });
522
523 it("is null before anything could have touched the workspace", () => {
524 expect(deriveWorkspaceMutationId([item("reply", "assistant_message", "completed")], null)).toBe(
525 null,
526 );
527 });
528});