1import { describe, expect, it } from "vite-plus/test";
2import {
3 type ThreadPullRequestLink,
4 type OrchestrationV2DomainEvent,
5 type OrchestrationV2Run,
6 type OrchestrationV2ThreadProjection,
7 type OrchestrationV2TurnItem,
8 MessageId,
9 ProjectId,
10 ProviderInstanceId,
11 NodeId,
12 ProviderThreadId,
13 ProviderTurnId,
14 RunId,
15 ThreadId,
16 TurnItemId,
17} from "@t3tools/contracts";
18import * as DateTime from "effect/DateTime";
19
20import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts";
21
22const now = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z");
23const threadId = ThreadId.make("thread-reducer");
24const runId = RunId.make("run-reducer");
25const run = {
26 id: runId,
27 threadId,
28 ordinal: 1,
29 providerInstanceId: ProviderInstanceId.make("codex"),
30 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
31 providerThreadId: null,
32 userMessageId: MessageId.make("message-reducer"),
33 rootNodeId: null,
34 activeAttemptId: null,
35 status: "completed",
36 requestedAt: now,
37 startedAt: now,
38 completedAt: now,
39 checkpointId: null,
40 contextHandoffId: null,
41} satisfies OrchestrationV2Run;
42
43function commandItem(id: string, output = "done", ordinal = 1): OrchestrationV2TurnItem {
44 return {
45 id: TurnItemId.make(id),
46 threadId,
47 runId,
48 nodeId: null,
49 providerThreadId: null,
50 providerTurnId: null,
51 nativeItemRef: null,
52 parentItemId: null,
53 ordinal,
54 status: "completed",
55 title: null,
56 startedAt: now,
57 completedAt: now,
58 updatedAt: now,
59 type: "command_execution",
60 input: "pwd",
61 output,
62 exitCode: 0,
63 };
64}
65const emptyProjection = {
66 thread: {
67 id: threadId,
68 projectId: ProjectId.make("project-reducer"),
69 title: "Reducer",
70 providerInstanceId: ProviderInstanceId.make("codex"),
71 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
72 runtimeMode: "full-access",
73 interactionMode: "default",
74 branch: null,
75 worktreePath: null,
76 activeProviderThreadId: null,
77 lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
78 forkedFrom: null,
79 createdBy: "user",
80 creationSource: "web",
81 createdAt: now,
82 updatedAt: now,
83 archivedAt: null,
84 settledOverride: null,
85 settledAt: null,
86 lastVisitedAt: null,
87 deletedAt: null,
88 },
89 runs: [],
90 attempts: [],
91 nodes: [],
92 subagents: [],
93 providerSessions: [],
94 providerThreads: [],
95 providerTurns: [],
96 runtimeRequests: [],
97 messages: [],
98 plans: [],
99 turnItems: [],
100 checkpointScopes: [],
101 checkpoints: [],
102 contextHandoffs: [],
103 contextTransfers: [],
104 visibleTurnItems: [],
105 updatedAt: now,
106} as OrchestrationV2ThreadProjection;
107
108describe("applyOrchestrationV2ProjectionEvent", () => {
109 it("applies a compact PR sync without replacing other links or thread metadata", () => {
110 const link = (number: number): ThreadPullRequestLink => ({
111 host: "github.com",
112 repository: "pingdotgg/t3code",
113 number,
114 url: `https://github.com/pingdotgg/t3code/pull/${number}`,
115 source: "manual",
116 linkedAt: DateTime.formatIso(now),
117 snapshot: null,
118 stack: null,
119 });
120 const links = [link(1), link(2)];
121 const projection = {
122 ...emptyProjection,
123 thread: { ...emptyProjection.thread, pullRequests: links },
124 };
125 const updated = { ...links[1]!, source: "agent" as const };
126 const next = applyOrchestrationV2ProjectionEvent(projection, {
127 id: "compact-sync" as OrchestrationV2DomainEvent["id"],
128 type: "thread.pull-request-link-updated",
129 threadId,
130 occurredAt: now,
131 payload: updated,
132 });
133 expect(next?.thread).toEqual({ ...projection.thread, pullRequests: [links[0], updated] });
134 expect(next?.thread.pullRequests?.[0]).toBe(links[0]);
135 });
136
137 it("keeps live token usage when the terminal provider turn omits it", () => {
138 const providerTurnId = ProviderTurnId.make("provider-turn-reducer");
139 const running = {
140 id: providerTurnId,
141 providerThreadId: ProviderThreadId.make("provider-thread-reducer"),
142 nodeId: NodeId.make("provider-node-reducer"),
143 runAttemptId: null,
144 nativeTurnRef: null,
145 ordinal: 1,
146 status: "running" as const,
147 startedAt: now,
148 completedAt: null,
149 tokenUsage: {
150 usedTokens: 50_000,
151 maxTokens: 200_000,
152 updatedAt: "2026-08-29T00:00:00.000Z",
153 },
154 };
155 const projection = { ...emptyProjection, providerTurns: [running] };
156 const event = {
157 id: "event-provider-turn-terminal",
158 type: "provider-turn.updated",
159 threadId,
160 driver: "codex",
161 occurredAt: now,
162 payload: {
163 ...running,
164 status: "completed",
165 completedAt: now,
166 tokenUsage: undefined,
167 },
168 } as OrchestrationV2DomainEvent;
169
170 const next = applyOrchestrationV2ProjectionEvent(projection, event);
171
172 expect(next?.providerTurns[0]?.status).toBe("completed");
173 expect(next?.providerTurns[0]?.tokenUsage).toEqual(running.tokenUsage);
174 });
175
176 it("applies thread lifecycle payloads instead of leaving stale metadata", () => {
177 const archivedAt = DateTime.makeUnsafe("2026-06-20T01:00:00.000Z");
178 const event = {
179 id: "event-archive",
180 type: "thread.archived",
181 threadId,
182 occurredAt: archivedAt,
183 payload: { ...emptyProjection.thread, archivedAt, updatedAt: archivedAt },
184 } as OrchestrationV2DomainEvent;
185
186 const next = applyOrchestrationV2ProjectionEvent(emptyProjection, event);
187 expect(next?.thread.archivedAt).toEqual(archivedAt);
188 expect(next?.updatedAt).toEqual(archivedAt);
189 });
190
191 it("ignores events for another thread", () => {
192 const event = {
193 id: "event-other",
194 type: "thread.deleted",
195 threadId: ThreadId.make("thread-other"),
196 occurredAt: now,
197 payload: { ...emptyProjection.thread, id: ThreadId.make("thread-other"), deletedAt: now },
198 } as OrchestrationV2DomainEvent;
199
200 expect(applyOrchestrationV2ProjectionEvent(emptyProjection, event)).toBe(emptyProjection);
201 });
202
203 it("preserves visible row identity when run updates do not change membership", () => {
204 const item = commandItem("item-stable");
205 const visibleTurnItems = [
206 {
207 position: 0,
208 visibility: "local" as const,
209 sourceThreadId: threadId,
210 sourceItemId: item.id,
211 item,
212 },
213 ];
214 const projection = {
215 ...emptyProjection,
216 runs: [run],
217 turnItems: [item],
218 visibleTurnItems,
219 };
220 const event = {
221 id: "event-run-update",
222 type: "run.updated",
223 threadId,
224 runId,
225 occurredAt: now,
226 payload: { ...run, status: "completed" },
227 } as OrchestrationV2DomainEvent;
228
229 const next = applyOrchestrationV2ProjectionEvent(projection, event);
230 expect(next?.visibleTurnItems).toBe(visibleTurnItems);
231 expect(next?.visibleTurnItems[0]).toBe(visibleTurnItems[0]);
232 });
233
234 it("replaces only the updated visible item when membership is unchanged", () => {
235 const first = commandItem("item-first", "first");
236 const second = commandItem("item-second", "second");
237 const firstRow = {
238 position: 0,
239 visibility: "local" as const,
240 sourceThreadId: threadId,
241 sourceItemId: first.id,
242 item: first,
243 };
244 const secondRow = {
245 position: 1,
246 visibility: "local" as const,
247 sourceThreadId: threadId,
248 sourceItemId: second.id,
249 item: second,
250 };
251 const updated = commandItem("item-first", "streamed output");
252 const projection = {
253 ...emptyProjection,
254 runs: [run],
255 turnItems: [first, second],
256 visibleTurnItems: [firstRow, secondRow],
257 };
258 const event = {
259 id: "event-item-update",
260 type: "turn-item.updated",
261 threadId,
262 runId,
263 occurredAt: now,
264 payload: updated,
265 } as OrchestrationV2DomainEvent;
266
267 const next = applyOrchestrationV2ProjectionEvent(projection, event);
268 expect(next?.visibleTurnItems).not.toBe(projection.visibleTurnItems);
269 expect(next?.visibleTurnItems[0]).not.toBe(firstRow);
270 expect(next?.visibleTurnItems[0]?.item).toBe(updated);
271 expect(next?.visibleTurnItems[1]).toBe(secondRow);
272 });
273
274 it("inserts live turn items by authoritative ordinal", () => {
275 const queuedFuture = commandItem("item-queued-future", "queued", 300);
276 const activeAssistant = commandItem("item-active-assistant", "done", 201);
277 const queuedRow = {
278 position: 0,
279 visibility: "local" as const,
280 sourceThreadId: threadId,
281 sourceItemId: queuedFuture.id,
282 item: queuedFuture,
283 };
284 const projection = {
285 ...emptyProjection,
286 runs: [run],
287 turnItems: [queuedFuture],
288 visibleTurnItems: [queuedRow],
289 };
290 const event = {
291 id: "event-active-assistant",
292 type: "turn-item.updated",
293 threadId,
294 runId,
295 occurredAt: now,
296 payload: activeAssistant,
297 } as OrchestrationV2DomainEvent;
298
299 const next = applyOrchestrationV2ProjectionEvent(projection, event);
300 expect(next?.visibleTurnItems.map((row) => row.item.id)).toEqual([
301 activeAssistant.id,
302 queuedFuture.id,
303 ]);
304 expect(next?.visibleTurnItems.map((row) => row.position)).toEqual([0, 1]);
305 });
306
307 describe("partial timeline", () => {
308 const queuedRunId = RunId.make("run-queued");
309 const steerRunId = RunId.make("run-steer");
310 const queuedRun = { ...run, id: queuedRunId, ordinal: 2, status: "queued" } as const;
311 const steerRun = { ...run, id: steerRunId, ordinal: 3, status: "completed" } as const;
312 const runItem = (id: string, itemRunId: RunId, ordinal: number) => ({
313 ...commandItem(id, id, ordinal),
314 runId: itemRunId,
315 });
316 const windowOf = (
317 runs: ReadonlyArray<OrchestrationV2Run>,
318 items: ReadonlyArray<OrchestrationV2TurnItem>,
319 ): OrchestrationV2ThreadProjection => ({
320 ...emptyProjection,
321 runs,
322 turnItems: items,
323 visibleTurnItems: items.map((item, position) => ({
324 position,
325 visibility: "local" as const,
326 sourceThreadId: threadId,
327 sourceItemId: item.id,
328 item,
329 })),
330 });
331 const itemEvent = (payload: OrchestrationV2TurnItem) =>
332 ({
333 id: `event-${payload.id}`,
334 type: "turn-item.updated",
335 threadId,
336 occurredAt: now,
337 payload,
338 }) as OrchestrationV2DomainEvent;
339 const runEvent = (payload: OrchestrationV2Run) =>
340 ({
341 id: `event-${payload.id}-${payload.status}`,
342 type: "run.updated",
343 threadId,
344 occurredAt: now,
345 payload,
346 }) as OrchestrationV2DomainEvent;
347
348 it("shows a queued run that starts after a newer steer raised the watermark", () => {
349 // Run 2 was queued first, run 3 was steered in and ran first. Its items
350 // sit in run 3's band and already raised the watermark past run 2's band.
351 const steerItems = [
352 runItem("item-steer-user", steerRunId, 3_000_001),
353 runItem("item-steer-reply", steerRunId, 3_000_002),
354 ];
355 let projection: OrchestrationV2ThreadProjection | null = windowOf(
356 [run, queuedRun, steerRun],
357 steerItems,
358 );
359 let latestLocalTurnOrdinal = 3_000_002;
360 // Same order the orchestrator commits a queued start: the user message
361 // lands while the run is still queued, then the run starts and replies.
362 const queuedUser = runItem("item-queued-user", queuedRunId, 2_000_001);
363 const queuedReply = runItem("item-queued-reply", queuedRunId, 2_000_002);
364 for (const event of [
365 itemEvent(queuedUser),
366 runEvent({ ...queuedRun, status: "running" }),
367 itemEvent(queuedReply),
368 runEvent({ ...queuedRun, status: "completed" }),
369 ]) {
370 projection = applyOrchestrationV2ProjectionEvent(projection, event, {
371 partialTimeline: true,
372 latestLocalTurnOrdinal,
373 });
374 if (event.type === "turn-item.updated") {
375 latestLocalTurnOrdinal = Math.max(latestLocalTurnOrdinal, event.payload.ordinal);
376 }
377 }
378
379 expect(projection?.visibleTurnItems.map((row) => row.item.id)).toEqual([
380 queuedUser.id,
381 queuedReply.id,
382 ...steerItems.map((item) => item.id),
383 ]);
384 expect(projection?.turnItems.map((item) => item.id)).toContain(queuedReply.id);
385 });
386
387 it("keeps a running run's item when the window starts after its earlier items", () => {
388 // The bounded window begins inside run 3; run 2 started and kept going,
389 // but none of its rows made it into the window.
390 const runningRun = { ...queuedRun, status: "running" } as const;
391 const steerItems = [
392 runItem("item-steer-user", steerRunId, 3_000_001),
393 runItem("item-steer-reply", steerRunId, 3_000_002),
394 ];
395 const projection = windowOf([run, runningRun, steerRun], steerItems);
396 const reply = runItem("item-running-reply", queuedRunId, 2_000_003);
397
398 const next = applyOrchestrationV2ProjectionEvent(projection, itemEvent(reply), {
399 partialTimeline: true,
400 latestLocalTurnOrdinal: 3_000_002,
401 });
402
403 expect(next?.visibleTurnItems.map((row) => row.item.id)).toEqual([
404 reply.id,
405 ...steerItems.map((item) => item.id),
406 ]);
407 });
408
409 it.each([
410 { name: "keeps an item of a run already in the window", itemRunId: steerRunId, kept: true },
411 { name: "drops an item of a finished run outside the window", itemRunId: runId, kept: false },
412 ])("$name that arrives below the watermark", ({ itemRunId, kept }) => {
413 const first = runItem("item-first", steerRunId, 3_000_001);
414 const later = runItem("item-later", steerRunId, 3_000_003);
415 const late = runItem("item-late", itemRunId, 3_000_002);
416 const projection = windowOf([run, steerRun], [first, later]);
417
418 const next = applyOrchestrationV2ProjectionEvent(projection, itemEvent(late), {
419 partialTimeline: true,
420 latestLocalTurnOrdinal: later.ordinal,
421 });
422 if (kept) {
423 expect(next?.visibleTurnItems.map((row) => row.item.id)).toEqual([
424 first.id,
425 late.id,
426 later.id,
427 ]);
428 } else {
429 expect(next).toBe(projection);
430 }
431 });
432 });
433
434 it("removes only hidden local items while preserving inherited rows", () => {
435 const inherited = commandItem("item-inherited");
436 const local = commandItem("item-local");
437 const inheritedRow = {
438 position: 0,
439 visibility: "inherited" as const,
440 sourceThreadId: ThreadId.make("thread-source"),
441 sourceItemId: inherited.id,
442 item: inherited,
443 };
444 const localRow = {
445 position: 1,
446 visibility: "local" as const,
447 sourceThreadId: threadId,
448 sourceItemId: local.id,
449 item: local,
450 };
451 const projection = {
452 ...emptyProjection,
453 runs: [run],
454 turnItems: [local],
455 visibleTurnItems: [inheritedRow, localRow],
456 };
457 const event = {
458 id: "event-run-rollback",
459 type: "run.updated",
460 threadId,
461 runId,
462 occurredAt: now,
463 payload: { ...run, status: "rolled_back" },
464 } as OrchestrationV2DomainEvent;
465
466 const next = applyOrchestrationV2ProjectionEvent(projection, event);
467 expect(next?.visibleTurnItems).toEqual([inheritedRow]);
468 expect(next?.visibleTurnItems[0]).toBe(inheritedRow);
469 });
470});
471
472it("does not scan every row against every run for a streaming item update", () => {
473 let runReads = 0;
474 const runs = Array.from({ length: 100 }, (_, index) => ({
475 ...run,
476 get id() {
477 runReads++;
478 return RunId.make(`run-${index}`);
479 },
480 }));
481 const items = Array.from({ length: 1000 }, (_, index) =>
482 commandItem(`item-${index}`, "before", index),
483 );
484 const projection = {
485 ...emptyProjection,
486 runs,
487 turnItems: items,
488 visibleTurnItems: items.map((item, position) => ({
489 item,
490 position,
491 visibility: "local" as const,
492 sourceThreadId: threadId,
493 sourceItemId: item.id,
494 })),
495 };
496 const payload = commandItem("item-999", "after", 999);
497 const next = applyOrchestrationV2ProjectionEvent(projection, {
498 id: "stream-update",
499 type: "turn-item.updated",
500 threadId,
501 occurredAt: now,
502 payload,
503 } as OrchestrationV2DomainEvent);
504 expect(next?.visibleTurnItems.at(-1)?.item).toBe(payload);
505 expect(next?.visibleTurnItems[0]).toBe(projection.visibleTurnItems[0]);
506 expect(runReads).toBeLessThanOrEqual(100);
507});