1import { describe, expect, it } from "@effect/vitest";
2import type {
3 ApplicationStoredEvent,
4 OrchestrationV2ShellSnapshot,
5 OrchestrationV2ShellStreamItem,
6 OrchestrationV2StoredEvent,
7 OrchestrationV2ThreadShell,
8} from "@t3tools/contracts";
9import { ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
10import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
11import * as DateTime from "effect/DateTime";
12import * as Duration from "effect/Duration";
13import * as Effect from "effect/Effect";
14import * as Option from "effect/Option";
15import * as SqlClient from "effect/sql/SqlClient";
16import * as Stream from "effect/Stream";
17import * as TestClock from "effect/testing/TestClock";
18
19import {
20 archivedShellStreamItemFromThreadShell,
21 buildActiveShellSnapshot,
22 coalesceShellApplicationEvents,
23 coalesceStoredThreadEvents,
24 composeShellStreamWithEnrichment,
25 dedupeShellEnrichment,
26 loadShellSnapshotParts,
27 shellStreamItemFromEnrichmentRefresh,
28 shellStreamItemFromThreadShell,
29 shellStreamItemsFromInitialSnapshot,
30 shellStreamItemsFromResumeSnapshot,
31 skipUnchangedThreadShells,
32} from "./ShellStream.ts";
33
34function project(sequence: number, id: string): ApplicationStoredEvent {
35 return {
36 sequence,
37 aggregateKind: "project",
38 aggregateId: id,
39 } as ApplicationStoredEvent;
40}
41
42function thread(sequence: number, id: string): ApplicationStoredEvent {
43 return {
44 sequence,
45 event: { threadId: id },
46 } as ApplicationStoredEvent;
47}
48
49const emptyShellSnapshot = {
50 schemaVersion: 1,
51 snapshotSequence: 0,
52 projects: [],
53 threads: [],
54 archivedThreads: [],
55} as OrchestrationV2ShellSnapshot;
56
57describe("buildActiveShellSnapshot", () => {
58 it("never duplicates archived rows into the regular shell", () => {
59 const active = shellFixture({ archivedAt: null });
60 const archived = shellFixture({
61 id: ThreadId.make("thread-archived"),
62 archivedAt: "2026-07-30T00:00:00.000Z" as never,
63 });
64
65 expect(
66 buildActiveShellSnapshot({
67 projects: [],
68 threads: {
69 schemaVersion: 1,
70 snapshotSequence: 4,
71 threads: [active],
72 archivedThreads: [archived],
73 },
74 snapshotSequence: 7,
75 }),
76 ).toEqual({
77 schemaVersion: 1,
78 snapshotSequence: 7,
79 projects: [],
80 threads: [active],
81 archivedThreads: [],
82 });
83 });
84});
85
86describe("coalesceShellApplicationEvents", () => {
87 it("keeps the newest event per aggregate and preserves sequence order", () => {
88 expect(
89 coalesceShellApplicationEvents([
90 thread(2, "thread-a"),
91 project(3, "project-a"),
92 thread(4, "thread-b"),
93 thread(5, "thread-a"),
94 project(6, "project-a"),
95 ]).map((event) => event.sequence),
96 ).toEqual([4, 5, 6]);
97 });
98});
99
100function storedThreadEvent(
101 sequence: number,
102 threadId: string,
103 event: Record<string, unknown> = {},
104): OrchestrationV2StoredEvent {
105 return { sequence, event: { threadId, ...event } } as OrchestrationV2StoredEvent;
106}
107
108describe("loadShellSnapshotParts", () => {
109 it.effect("decodes the threads after the read transaction releases the connection", () =>
110 Effect.gen(function* () {
111 const sql = yield* SqlClient.SqlClient;
112 const steps: Array<string> = [];
113 const step = <A>(name: string, value: A) =>
114 Effect.serviceOption(sql.transactionService).pipe(
115 Effect.map((transaction) => {
116 steps.push(`${name}:${Option.isSome(transaction) ? "in" : "out of"} transaction`);
117 return value;
118 }),
119 );
120 const thread = shellFixture({});
121
122 const snapshot = yield* loadShellSnapshotParts({
123 sql,
124 readThreads: step(
125 "read threads",
126 step("decode threads", {
127 schemaVersion: 1,
128 snapshotSequence: 3,
129 threads: [thread],
130 archivedThreads: [],
131 }),
132 ),
133 listProjects: step("list projects", []),
134 latestSequence: step("latest sequence", 7),
135 });
136
137 expect(steps).toEqual([
138 "read threads:in transaction",
139 "list projects:in transaction",
140 "latest sequence:in transaction",
141 "decode threads:out of transaction",
142 ]);
143 expect(snapshot.snapshotSequence).toBe(7);
144 expect(snapshot.threads.threads).toEqual([thread]);
145 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
146 );
147});
148
149function shellFixture(overrides: Partial<OrchestrationV2ThreadShell>): OrchestrationV2ThreadShell {
150 return { id: "thread-a", archivedAt: null, ...overrides } as OrchestrationV2ThreadShell;
151}
152
153describe("coalesceStoredThreadEvents", () => {
154 it("keeps the newest stored event per thread and preserves sequence order", () => {
155 expect(
156 coalesceStoredThreadEvents([
157 storedThreadEvent(2, "thread-a"),
158 storedThreadEvent(3, "thread-b"),
159 storedThreadEvent(5, "thread-a"),
160 ]).map((stored) => stored.sequence),
161 ).toEqual([3, 5]);
162 });
163});
164
165describe("shellStreamItemFromThreadShell", () => {
166 it("emits an active thread update when the shell is not archived", () => {
167 const shell = shellFixture({ archivedAt: null });
168 expect(
169 shellStreamItemFromThreadShell({ stored: storedThreadEvent(4, "thread-a"), shell }),
170 ).toEqual({
171 kind: "thread.updated",
172 sequence: 4,
173 location: "active",
174 thread: shell,
175 });
176 });
177
178 it("removes archived threads from the active-only shell", () => {
179 const shell = shellFixture({ archivedAt: "2026-07-30T00:00:00.000Z" as never });
180 expect(
181 shellStreamItemFromThreadShell({ stored: storedThreadEvent(4, "thread-a"), shell }),
182 ).toEqual({
183 kind: "thread.removed",
184 sequence: 4,
185 location: "active",
186 threadId: "thread-a",
187 });
188 });
189
190 it("emits an active-shell removal when an archived thread is deleted", () => {
191 expect(
192 shellStreamItemFromThreadShell({
193 stored: storedThreadEvent(6, "thread-a", {
194 type: "thread.deleted",
195 payload: { archivedAt: "2026-07-30T00:00:00.000Z" },
196 }),
197 shell: null,
198 }),
199 ).toEqual({
200 kind: "thread.removed",
201 sequence: 6,
202 location: "active",
203 threadId: "thread-a",
204 });
205 });
206
207 it("emits a removal from the active list for other missing shells", () => {
208 expect(
209 shellStreamItemFromThreadShell({
210 stored: storedThreadEvent(6, "thread-a", {
211 type: "thread.deleted",
212 payload: { archivedAt: null },
213 }),
214 shell: null,
215 }),
216 ).toEqual({
217 kind: "thread.removed",
218 sequence: 6,
219 location: "active",
220 threadId: "thread-a",
221 });
222 });
223});
224
225describe("archivedShellStreamItemFromThreadShell", () => {
226 it("emits an update for an archived shell", () => {
227 const shell = shellFixture({ archivedAt: "2026-07-30T00:00:00.000Z" as never });
228 expect(
229 archivedShellStreamItemFromThreadShell({ stored: storedThreadEvent(4, "thread-a"), shell }),
230 ).toEqual({ kind: "thread.updated", sequence: 4, thread: shell });
231 });
232
233 it("ignores active threads that never touched the archive", () => {
234 expect(
235 archivedShellStreamItemFromThreadShell({
236 stored: storedThreadEvent(4, "thread-a", { type: "thread.settled" }),
237 shell: shellFixture({ archivedAt: null }),
238 }),
239 ).toBeNull();
240 });
241
242 it("emits a removal when a thread leaves the archive", () => {
243 expect(
244 archivedShellStreamItemFromThreadShell({
245 stored: storedThreadEvent(4, "thread-a", { type: "thread.unarchived" }),
246 shell: shellFixture({ archivedAt: null }),
247 }),
248 ).toEqual({ kind: "thread.removed", sequence: 4, threadId: "thread-a" });
249 });
250
251 it("emits a removal when an archived thread is deleted", () => {
252 expect(
253 archivedShellStreamItemFromThreadShell({
254 stored: storedThreadEvent(4, "thread-a", {
255 type: "thread.deleted",
256 payload: { archivedAt: "2026-07-30T00:00:00.000Z" },
257 }),
258 shell: null,
259 }),
260 ).toEqual({ kind: "thread.removed", sequence: 4, threadId: "thread-a" });
261 });
262});
263
264describe("shellStreamItemFromEnrichmentRefresh", () => {
265 it("batches nearby completion roots onto one snapshot item", () => {
266 expect(
267 shellStreamItemFromEnrichmentRefresh({
268 snapshot: emptyShellSnapshot,
269 changes: [
270 { workspaceRoot: "/workspace/a" },
271 { workspaceRoot: "/workspace/b" },
272 { workspaceRoot: "/workspace/a" },
273 ],
274 }),
275 ).toEqual({
276 kind: "snapshot",
277 snapshot: emptyShellSnapshot,
278 resolvedRepositoryIdentityRoots: ["/workspace/a", "/workspace/b"],
279 });
280 });
281});
282
283describe("shellStreamItemsFromInitialSnapshot", () => {
284 it("keeps thread rows out of the metadata-only enrichment frame", () => {
285 const snapshot = {
286 ...emptyShellSnapshot,
287 projects: [
288 { id: "project-a", workspaceRoot: "/workspace/a" },
289 { id: "project-b", workspaceRoot: "/workspace/b" },
290 ],
291 threads: [shellFixture({})],
292 archivedThreads: [shellFixture({ id: ThreadId.make("thread-archived"), archivedAt: null })],
293 } as unknown as OrchestrationV2ShellSnapshot;
294
295 const items = shellStreamItemsFromInitialSnapshot({
296 snapshot,
297 resolvedRepositoryIdentityRoots: ["/workspace/a"],
298 });
299
300 expect(items[0]).toEqual({ kind: "snapshot", snapshot });
301 expect(items[1]).toMatchObject({
302 kind: "snapshot",
303 snapshot: {
304 projects: [{ id: "project-a", workspaceRoot: "/workspace/a" }],
305 threads: [],
306 archivedThreads: [],
307 },
308 resolvedRepositoryIdentityRoots: ["/workspace/a"],
309 });
310 expect(JSON.stringify(items[1]).length).toBeLessThan(JSON.stringify(items[0]).length);
311 });
312
313 it("emits unmarked authoritative then same-sequence marked enrichment when roots resolved", () => {
314 const snapshot = {
315 ...emptyShellSnapshot,
316 snapshotSequence: 7,
317 } as OrchestrationV2ShellSnapshot;
318
319 expect(
320 shellStreamItemsFromInitialSnapshot({
321 snapshot,
322 resolvedRepositoryIdentityRoots: ["/workspace/a", "/workspace/a"],
323 }),
324 ).toEqual([
325 { kind: "snapshot", snapshot },
326 {
327 kind: "snapshot",
328 snapshot,
329 resolvedRepositoryIdentityRoots: ["/workspace/a"],
330 },
331 ]);
332 });
333
334 it("emits only the unmarked authoritative snapshot when no roots resolved", () => {
335 expect(
336 shellStreamItemsFromInitialSnapshot({
337 snapshot: emptyShellSnapshot,
338 resolvedRepositoryIdentityRoots: [],
339 }),
340 ).toEqual([{ kind: "snapshot", snapshot: emptyShellSnapshot }]);
341 });
342});
343
344describe("shellStreamItemsFromResumeSnapshot", () => {
345 it("never repeats the authoritative shell snapshot", () => {
346 expect(
347 shellStreamItemsFromResumeSnapshot({
348 snapshot: {
349 ...emptyShellSnapshot,
350 threads: [shellFixture({})],
351 },
352 resolvedRepositoryIdentityRoots: [],
353 }),
354 ).toEqual([]);
355 });
356});
357
358describe("composeShellStreamWithEnrichment", () => {
359 it.effect(
360 "emits every initial item before enrichment even when enrichment is already ready",
361 () =>
362 Effect.gen(function* () {
363 const initialSnapshot = {
364 ...emptyShellSnapshot,
365 snapshotSequence: 5,
366 } as OrchestrationV2ShellSnapshot;
367 const enrichmentSnapshot = {
368 ...emptyShellSnapshot,
369 snapshotSequence: 10,
370 } as OrchestrationV2ShellSnapshot;
371
372 const initialItems = shellStreamItemsFromInitialSnapshot({
373 snapshot: initialSnapshot,
374 resolvedRepositoryIdentityRoots: ["/workspace/a"],
375 });
376 // Enrichment stream is fully ready before the composed stream is pulled.
377 const enrichment = Stream.make(
378 shellStreamItemFromEnrichmentRefresh({
379 snapshot: enrichmentSnapshot,
380 changes: [{ workspaceRoot: "/workspace/b" }],
381 }),
382 );
383 const tail = Stream.make(
384 { kind: "synchronized" as const },
385 {
386 kind: "project.removed" as const,
387 sequence: 6,
388 projectId: "project-a",
389 },
390 );
391
392 const items = Array.from(
393 yield* composeShellStreamWithEnrichment({
394 initial: Stream.fromIterable(initialItems),
395 tail,
396 enrichment,
397 }).pipe(Stream.runCollect),
398 );
399
400 expect(items.slice(0, initialItems.length)).toEqual(initialItems);
401
402 const enrichmentIndex = items.findIndex(
403 (item) =>
404 item.kind === "snapshot" &&
405 "resolvedRepositoryIdentityRoots" in item &&
406 item.resolvedRepositoryIdentityRoots?.includes("/workspace/b"),
407 );
408 expect(enrichmentIndex).toBeGreaterThanOrEqual(initialItems.length);
409
410 for (let index = 0; index < initialItems.length; index++) {
411 expect(items[index]).toEqual(initialItems[index]);
412 }
413 }),
414 );
415
416 it.effect("still interleaves enrichment with the post-prefix tail after initials drain", () =>
417 Effect.gen(function* () {
418 const items = Array.from(
419 yield* composeShellStreamWithEnrichment({
420 initial: Stream.make("initial-unmarked", "initial-marked"),
421 tail: Stream.make("tail-a", "tail-b"),
422 enrichment: Stream.make("enrichment"),
423 }).pipe(Stream.runCollect),
424 );
425
426 expect(items.slice(0, 2)).toEqual(["initial-unmarked", "initial-marked"]);
427 expect(items).toContain("enrichment");
428 expect(items).toContain("tail-a");
429 expect(items).toContain("tail-b");
430 expect(items.indexOf("enrichment")).toBeGreaterThanOrEqual(2);
431 }),
432 );
433});
434
435describe("dedupeShellEnrichment", () => {
436 const project = {
437 id: ProjectId.make("project-a"),
438 title: "A project",
439 workspaceRoot: "/workspace/a",
440 repositoryIdentity: null,
441 defaultModelSelection: null,
442 scripts: [],
443 createdAt: "2026-09-01T00:00:00Z",
444 updatedAt: "2026-09-01T00:00:00Z",
445 };
446 const initial = {
447 kind: "snapshot" as const,
448 snapshot: { ...emptyShellSnapshot, projects: [project] },
449 };
450 const marked = { ...initial, resolvedRepositoryIdentityRoots: [project.workspaceRoot] };
451
452 it.effect(
453 "keeps initial/resume resolution and real changes but drops sequence-only repeats",
454 () =>
455 Effect.gen(function* () {
456 const repeated = {
457 ...marked,
458 snapshot: { ...marked.snapshot, snapshotSequence: 10, projects: [{ ...project }] },
459 };
460 const progressed = {
461 ...repeated,
462 resolvedRepositoryIdentityRoots: [project.workspaceRoot, "/workspace/b"],
463 };
464 const changed = {
465 ...progressed,
466 snapshot: { ...progressed.snapshot, projects: [{ ...project, title: "Renamed" }] },
467 };
468 const items = yield* Stream.make(
469 initial,
470 marked,
471 repeated,
472 progressed,
473 changed,
474 changed,
475 ).pipe(dedupeShellEnrichment, Stream.runCollect);
476 expect(items).toEqual([initial, marked, progressed, changed]);
477 const resumed = Stream.make(marked, repeated).pipe(dedupeShellEnrichment);
478 expect(yield* Stream.runCollect(resumed)).toEqual([marked]);
479 expect(yield* Stream.runCollect(resumed)).toEqual([marked]);
480 }),
481 );
482
483 it.effect("keeps authoritative snapshots and invalidates metadata after project deltas", () =>
484 Effect.gen(function* () {
485 const removed = { kind: "project.removed" as const, projectId: project.id, sequence: 2 };
486 const updated = { kind: "project.updated" as const, project, sequence: 3 };
487 const withThreads = {
488 ...marked,
489 snapshot: { ...marked.snapshot, threads: [shellFixture({})] },
490 };
491 const values = [
492 initial,
493 initial,
494 marked,
495 removed,
496 marked,
497 updated,
498 marked,
499 withThreads,
500 withThreads,
501 ];
502 expect(
503 yield* Stream.fromIterable(values).pipe(dedupeShellEnrichment, Stream.runCollect),
504 ).toEqual(values);
505 }),
506 );
507
508 it.effect(
509 "deduplicates across thread deltas without dropping those deltas or identity clears",
510 () =>
511 Effect.gen(function* () {
512 const delta = {
513 kind: "thread.removed" as const,
514 threadId: ThreadId.make("thread-a"),
515 location: "active" as const,
516 sequence: 2,
517 };
518 const resolved = {
519 ...marked,
520 snapshot: {
521 ...marked.snapshot,
522 projects: [
523 {
524 ...project,
525 repositoryIdentity: {
526 canonicalKey: "github.com/test/repo",
527 locator: {
528 source: "git-remote" as const,
529 remoteName: "origin",
530 remoteUrl: "https://github.com/test/repo.git",
531 },
532 },
533 },
534 ],
535 },
536 };
537 expect(
538 yield* Stream.make(marked, delta, marked, resolved, marked).pipe(
539 dedupeShellEnrichment,
540 Stream.runCollect,
541 ),
542 ).toEqual([marked, delta, resolved, marked]);
543 }),
544 );
545});
546
547describe("skipUnchangedThreadShells", () => {
548 const at = (ms: number) => DateTime.makeUnsafe(Date.parse("2026-10-01T00:00:00.000Z") + ms);
549 const shell = (
550 id: string,
551 overrides: Partial<OrchestrationV2ThreadShell> = {},
552 ): OrchestrationV2ThreadShell => ({
553 id: ThreadId.make(id),
554 projectId: ProjectId.make("project-a"),
555 title: "Thread",
556 providerInstanceId: ProviderInstanceId.make("codex"),
557 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
558 runtimeMode: "full-access",
559 interactionMode: "default",
560 worktreePath: null,
561 activeProviderThreadId: null,
562 lineage: { rootThreadId: ThreadId.make(id), parentThreadId: null, relationshipToParent: null },
563 forkedFrom: null,
564 createdBy: "user",
565 creationSource: "web",
566 activeRunId: null,
567 latestVisibleMessage: null,
568 hasActionableProposedPlan: false,
569 itemCount: 1,
570 visibleItemCount: 1,
571 branch: null,
572 status: "running",
573 pendingRuntimeRequest: null,
574 pendingBackgroundTasks: [],
575 latestRunId: null,
576 latestUserMessageAt: null,
577 createdAt: at(0),
578 updatedAt: at(0),
579 archivedAt: null,
580 settledOverride: null,
581 settledAt: null,
582 deletedAt: null,
583 ...overrides,
584 });
585 type Step =
586 | { readonly advanceMs: number }
587 | Exclude<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" | "synchronized" }>;
588 const updated = (sequence: number, thread: OrchestrationV2ThreadShell): Step => ({
589 kind: "thread.updated",
590 sequence,
591 location: "active",
592 thread,
593 });
594 const run = (steps: ReadonlyArray<Step>) =>
595 Stream.fromIterable(steps).pipe(
596 Stream.mapEffect((step) =>
597 "advanceMs" in step
598 ? TestClock.adjust(Duration.millis(step.advanceMs)).pipe(Effect.as(null))
599 : Effect.succeed(step),
600 ),
601 Stream.filter((item) => item !== null),
602 skipUnchangedThreadShells,
603 Stream.runCollect,
604 Effect.map((items) => Array.from(items, (item) => ("sequence" in item ? item.sequence : -1))),
605 );
606
607 it.effect("drops shells that only moved updatedAt, and resends them after a while", () =>
608 Effect.gen(function* () {
609 const sent = yield* run([
610 updated(1, shell("a")),
611 updated(2, shell("a", { updatedAt: at(100) })),
612 updated(3, shell("a", { updatedAt: at(200), itemCount: 2, visibleItemCount: 2 })),
613 updated(4, shell("a", { updatedAt: at(300), itemCount: 2, visibleItemCount: 2 })),
614 updated(5, shell("b")),
615 // The resend window is 5 s.
616 { advanceMs: 5_000 },
617 updated(6, shell("a", { updatedAt: at(5_300), itemCount: 2, visibleItemCount: 2 })),
618 ]);
619 expect(sent).toEqual([1, 3, 5, 6]);
620 }),
621 );
622
623 it.effect("sends the next update after a removal", () =>
624 Effect.gen(function* () {
625 const sent = yield* run([
626 updated(1, shell("a")),
627 { kind: "thread.removed", sequence: 2, location: "active", threadId: ThreadId.make("a") },
628 updated(3, shell("a")),
629 ]);
630 expect(sent).toEqual([1, 2, 3]);
631 }),
632 );
633});