1import { describe, expect, it } from "@effect/vitest";
2import type {
3 ApplicationStoredEvent,
4 OrchestrationV2ShellSnapshot,
5 OrchestrationV2ShellStreamItem,
6 OrchestrationV2ThreadShell,
7} from "@t3tools/contracts";
8import { ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
9import type * as Cause from "effect/Cause";
10import * as DateTime from "effect/DateTime";
11import * as Deferred from "effect/Deferred";
12import * as Duration from "effect/Duration";
13import * as Effect from "effect/Effect";
14import * as PubSub from "effect/PubSub";
15import * as Queue from "effect/Queue";
16import * as Stream from "effect/Stream";
17import * as TestClock from "effect/testing/TestClock";
18
19import {
20 activeShellRowsForSubscriber,
21 bufferShellLiveStream,
22 composeShellStreamWithEnrichment,
23 dedupeShellEnrichment,
24 omitIdleSubagentShells,
25 omitOlderSettledShells,
26 shellStreamItemFromEnrichmentRefresh,
27 shellStreamItemsFromInitialSnapshot,
28 shouldPublishShellApplicationEvent,
29 skipUnchangedThreadShells,
30} from "./ShellStream.ts";
31
32const emptyShellSnapshot = {
33 schemaVersion: 1,
34 snapshotSequence: 0,
35 projects: [],
36 threads: [],
37 archivedThreads: [],
38} as OrchestrationV2ShellSnapshot;
39
40describe("subagent shell events", () => {
41 const stored = (threadId: string, type: string, itemType = "command_execution") =>
42 ({ event: { threadId, type, payload: { type: itemType } } }) as ApplicationStoredEvent;
43
44 it("keeps regular threads and subagent state changes while omitting subagent transcript updates", () => {
45 const subagents = new Set(["child"]);
46 expect(
47 shouldPublishShellApplicationEvent(stored("parent", "turn-item.updated"), subagents),
48 ).toBe(true);
49 expect(
50 shouldPublishShellApplicationEvent(stored("child", "turn-item.updated"), subagents),
51 ).toBe(false);
52 for (const itemType of ["secret_request", "link_request", "user_message"]) {
53 expect(
54 shouldPublishShellApplicationEvent(
55 stored("child", "turn-item.updated", itemType),
56 subagents,
57 ),
58 ).toBe(true);
59 }
60 expect(shouldPublishShellApplicationEvent(stored("child", "node.updated"), subagents)).toBe(
61 false,
62 );
63 expect(
64 shouldPublishShellApplicationEvent(stored("child", "runtime-request.updated"), subagents),
65 ).toBe(true);
66 expect(shouldPublishShellApplicationEvent(stored("child", "run.updated"), subagents)).toBe(
67 true,
68 );
69 });
70
71 it("tracks subagents created after the subscription starts", () => {
72 const subagents = new Set<string>();
73 const created = {
74 event: {
75 threadId: "new-child",
76 type: "thread.created",
77 payload: { lineage: { relationshipToParent: "subagent" } },
78 },
79 } as ApplicationStoredEvent;
80 expect(shouldPublishShellApplicationEvent(created, subagents)).toBe(true);
81 expect(
82 shouldPublishShellApplicationEvent(stored("new-child", "message.updated"), subagents),
83 ).toBe(false);
84 expect(
85 shouldPublishShellApplicationEvent(stored("new-child", "thread.deleted"), subagents),
86 ).toBe(true);
87 expect(subagents.has("new-child")).toBe(true);
88 });
89});
90
91it.effect("coalesces a slow shell subscriber by identity and location in sequence order", () =>
92 Effect.scoped(
93 Effect.gen(function* () {
94 const input = yield* Queue.unbounded<
95 Extract<OrchestrationV2ShellStreamItem, { readonly sequence: number }>,
96 Cause.Done
97 >();
98 const drained = yield* Deferred.make<void>();
99 const shell = shellFixture({ id: ThreadId.make("same"), title: "Before" });
100 const project = {
101 id: ProjectId.make("same"),
102 title: "Before",
103 workspaceRoot: "/workspace/same",
104 repositoryIdentity: null,
105 defaultModelSelection: null,
106 scripts: [],
107 createdAt: "2026-09-01T00:00:00Z",
108 updatedAt: "2026-09-01T00:00:00Z",
109 };
110 const initial = {
111 kind: "thread.updated" as const,
112 sequence: 1,
113 location: "active" as const,
114 thread: shell,
115 };
116 const updates = [
117 { kind: "project.updated", sequence: 2, project },
118 { ...initial, sequence: 3 },
119 { ...initial, sequence: 4, location: "archive" },
120 { ...initial, sequence: 5, thread: { ...shell, id: ThreadId.make("other") } },
121 {
122 kind: "project.updated",
123 sequence: 6,
124 project: { ...project, id: ProjectId.make("other") },
125 },
126 { kind: "project.removed", sequence: 7, projectId: project.id },
127 { kind: "thread.removed", sequence: 8, location: "active", threadId: shell.id },
128 { kind: "project.updated", sequence: 9, project: { ...project, title: "Recreated" } },
129 { ...initial, sequence: 10, thread: { ...shell, title: "Recreated" } },
130 { kind: "thread.removed", sequence: 11, location: "archive", threadId: shell.id },
131 ] satisfies Array<Extract<OrchestrationV2ShellStreamItem, { readonly sequence: number }>>;
132 const pull = yield* Stream.toPull(
133 bufferShellLiveStream(
134 Stream.fromQueue(input).pipe(Stream.ensuring(Deferred.succeed(drained, undefined))),
135 { maxItems: 6, maxSerializedBytes: 1_500 },
136 ),
137 );
138 yield* Queue.offer(input, initial);
139 expect(yield* pull).toEqual([initial]);
140 yield* Queue.offerAll(input, updates);
141 yield* Queue.end(input);
142 // Finish producing the burst while the first batch is still awaiting its ACK.
143 yield* Deferred.await(drained);
144 expect(yield* pull).toEqual([updates[3], updates[4], updates[7], updates[8], updates[9]]);
145 }),
146 ),
147);
148
149function shellFixture(overrides: Partial<OrchestrationV2ThreadShell>): OrchestrationV2ThreadShell {
150 return { id: "thread-a", archivedAt: null, ...overrides } as OrchestrationV2ThreadShell;
151}
152
153describe("omitIdleSubagentShells", () => {
154 const row = (
155 id: string,
156 parent: string | null,
157 overrides: Partial<OrchestrationV2ThreadShell> = {},
158 ) =>
159 shellFixture({
160 id: ThreadId.make(id),
161 lineage: {
162 parentThreadId: parent === null ? null : ThreadId.make(parent),
163 relationshipToParent: parent === null ? null : "subagent",
164 rootThreadId: ThreadId.make("root"),
165 },
166 status: "completed",
167 activityRunStatus: null,
168 pendingRuntimeRequest: null,
169 pendingBackgroundTasks: [],
170 ...overrides,
171 });
172
173 it("drops idle leaf children but keeps live ones and the middle of nested chains", () => {
174 const threads = [
175 row("root", null),
176 row("idle", "root"),
177 row("running", "root", { activityRunStatus: "running" }),
178 row("queued", "root", { status: "queued" }),
179 row("asking", "root", {
180 pendingRuntimeRequest: {
181 id: "request" as never,
182 kind: "user_input",
183 createdAt: DateTime.makeUnsafe(0),
184 },
185 }),
186 row("background", "root", {
187 pendingBackgroundTasks: [{ kind: "subagent" } as never],
188 }),
189 // Idle, but a woken grandchild walks through it to reach the root's badge.
190 row("middle", "root"),
191 row("grandchild", "middle"),
192 ];
193 expect(omitIdleSubagentShells(threads).map((thread) => thread.id)).toEqual([
194 "root",
195 "running",
196 "queued",
197 "asking",
198 "background",
199 "middle",
200 ]);
201 });
202
203 it("still filters the transcript events of omitted children for a subscriber that opts in", () => {
204 const threads = [row("root", null), row("idle", "root")];
205 const omitted = activeShellRowsForSubscriber(threads, true);
206 expect(omitted.threads.map((thread) => thread.id)).toEqual(["root"]);
207 expect(
208 shouldPublishShellApplicationEvent(
209 {
210 event: {
211 threadId: "idle",
212 type: "turn-item.updated",
213 payload: { type: "command_execution" },
214 },
215 } as ApplicationStoredEvent,
216 omitted.subagentThreadIds,
217 ),
218 ).toBe(false);
219 });
220});
221
222describe("omitOlderSettledShells", () => {
223 const DAY = 24 * 60 * 60 * 1000;
224 const now = DateTime.makeUnsafe(100 * DAY);
225 const row = (
226 id: string,
227 settledDaysAgo: number | null,
228 overrides: Partial<OrchestrationV2ThreadShell> = {},
229 ) =>
230 shellFixture({
231 id: ThreadId.make(id),
232 projectId: ProjectId.make(id.startsWith("b-") ? "project-b" : "project-a"),
233 lineage: {
234 parentThreadId: null,
235 relationshipToParent: null,
236 rootThreadId: ThreadId.make(id),
237 },
238 status: "completed",
239 activityRunStatus: null,
240 pendingRuntimeRequest: null,
241 pendingBackgroundTasks: [],
242 settledOverride: settledDaysAgo === null ? null : "settled",
243 settledAt:
244 settledDaysAgo === null ? null : DateTime.makeUnsafe(100 * DAY - settledDaysAgo * DAY),
245 ...overrides,
246 });
247 const newest = Array.from({ length: 25 }, (_, index) => row(`new-${index}`, index / 10));
248 const omit = (threads: ReadonlyArray<OrchestrationV2ThreadShell>) =>
249 omitOlderSettledShells({ threads, snapshotSequence: 7, now });
250
251 it("keeps the 25 newest settled threads and counts the rest per project", () => {
252 const sideChat = row("side-chat", 0, {
253 lineage: {
254 parentThreadId: ThreadId.make("new-0"),
255 relationshipToParent: "fork",
256 rootThreadId: ThreadId.make("new-0"),
257 sideChat: true,
258 },
259 });
260 const result = omit([
261 ...newest,
262 sideChat,
263 row("old", 30),
264 row("b-old", 40),
265 row("unsettled", null),
266 ]);
267 expect(result.threads.map((thread) => thread.id)).toEqual([
268 ...newest.map((thread) => thread.id),
269 "side-chat",
270 "unsettled",
271 ]);
272 expect(result.olderSettled).toEqual({
273 before: DateTime.formatIso(DateTime.makeUnsafe(100 * DAY - 2.4 * DAY)),
274 asOfSequence: 7,
275 counts: [
276 { projectId: "project-a", count: 1 },
277 { projectId: "project-b", count: 1 },
278 ],
279 });
280 });
281
282 it("keeps threads tied with the 25th newest", () => {
283 const result = omit([...newest, row("tie", 2.4), row("older", 2.5)]);
284 expect(result.threads.map((thread) => thread.id).slice(25)).toEqual(["tie"]);
285 });
286
287 it("omits nothing with fewer than 25 settled threads, however old", () => {
288 const old = Array.from({ length: 5 }, (_, index) => row(`old-${index}`, 300 + index));
289 const result = omit(old);
290 expect(result.threads).toEqual(old);
291 expect(result.olderSettled.counts).toEqual([]);
292 });
293
294 it("keeps older rows that are snoozed, live, or above a kept subagent or side chat, and still counts them", () => {
295 const threads = [
296 ...newest,
297 row("snoozed", 30, { snoozedUntil: DateTime.makeUnsafe(101 * DAY) }),
298 row("woke", 30, { snoozedUntil: DateTime.makeUnsafe(99 * DAY) }),
299 row("running", 30, { activityRunStatus: "running" }),
300 row("parent", 30),
301 shellFixture({
302 id: ThreadId.make("child"),
303 projectId: ProjectId.make("project-a"),
304 lineage: {
305 parentThreadId: ThreadId.make("parent"),
306 relationshipToParent: "subagent",
307 rootThreadId: ThreadId.make("parent"),
308 },
309 settledOverride: null,
310 settledAt: null,
311 }),
312 row("side-chat-parent", 30),
313 row("side-chat", null, {
314 lineage: {
315 parentThreadId: ThreadId.make("side-chat-parent"),
316 relationshipToParent: "fork",
317 rootThreadId: ThreadId.make("side-chat-parent"),
318 sideChat: true,
319 },
320 }),
321 ];
322 const result = omit(threads);
323 expect(result.threads.map((thread) => thread.id).slice(25)).toEqual([
324 "snoozed",
325 "running",
326 "parent",
327 "child",
328 "side-chat-parent",
329 "side-chat",
330 ]);
331 expect(result.olderSettled.counts).toEqual([{ projectId: "project-a", count: 5 }]);
332 });
333});
334
335describe("shellStreamItemsFromInitialSnapshot", () => {
336 it("keeps thread rows out of the metadata-only enrichment frame", () => {
337 const snapshot = {
338 ...emptyShellSnapshot,
339 projects: [
340 { id: "project-a", workspaceRoot: "/workspace/a" },
341 { id: "project-b", workspaceRoot: "/workspace/b" },
342 ],
343 threads: [shellFixture({})],
344 archivedThreads: [shellFixture({ id: ThreadId.make("thread-archived"), archivedAt: null })],
345 } as unknown as OrchestrationV2ShellSnapshot;
346
347 const items = shellStreamItemsFromInitialSnapshot({
348 snapshot,
349 resolvedRepositoryIdentityRoots: ["/workspace/a"],
350 });
351
352 expect(items[0]).toEqual({ kind: "snapshot", snapshot });
353 expect(items[1]).toMatchObject({
354 kind: "snapshot",
355 snapshot: {
356 projects: [{ id: "project-a", workspaceRoot: "/workspace/a" }],
357 threads: [],
358 archivedThreads: [],
359 },
360 resolvedRepositoryIdentityRoots: ["/workspace/a"],
361 });
362 expect(JSON.stringify(items[1]).length).toBeLessThan(JSON.stringify(items[0]).length);
363 });
364});
365
366describe("composeShellStreamWithEnrichment", () => {
367 it.effect("preserves pending repository resolutions while an enrichment ACK is withheld", () =>
368 Effect.scoped(
369 Effect.gen(function* () {
370 // Match ProjectEnrichmentService's sliding notifications and the WS batching path.
371 const changes = yield* Effect.acquireRelease(
372 PubSub.sliding<{ readonly workspaceRoot: string }>(256),
373 PubSub.shutdown,
374 );
375 const subscription = yield* PubSub.subscribe(changes);
376 const pull = yield* Stream.toPull(
377 composeShellStreamWithEnrichment({
378 initial: Stream.empty,
379 tail: Stream.empty,
380 enrichment: Stream.fromSubscription(subscription).pipe(
381 Stream.groupedWithin(64, Duration.millis(25)),
382 Stream.map((changes) =>
383 shellStreamItemFromEnrichmentRefresh({
384 snapshot: emptyShellSnapshot,
385 changes,
386 }),
387 ),
388 ),
389 }),
390 );
391
392 const firstRoots = Array.from({ length: 64 }, (_, index) => `/workspace/first-${index}`);
393 yield* PubSub.publishAll(
394 changes,
395 firstRoots.map((workspaceRoot) => ({ workspaceRoot })),
396 );
397 expect(yield* pull).toMatchObject([{ resolvedRepositoryIdentityRoots: firstRoots }]);
398
399 // The next completion can be batched before the client acknowledges the first.
400 yield* PubSub.publish(changes, { workspaceRoot: "/workspace/pending" });
401 yield* Effect.yieldNow;
402 yield* TestClock.adjust(Duration.millis(25));
403 const laterRoots = Array.from({ length: 256 }, (_, index) => `/workspace/later-${index}`);
404 yield* PubSub.publishAll(
405 changes,
406 laterRoots.map((workspaceRoot) => ({ workspaceRoot })),
407 );
408
409 const resumed = yield* Stream.fromPull(Effect.succeed(pull)).pipe(
410 Stream.takeUntil(
411 (item) =>
412 item.resolvedRepositoryIdentityRoots?.includes("/workspace/later-255") ?? false,
413 ),
414 Stream.runCollect,
415 );
416 expect(resumed.flatMap((item) => item.resolvedRepositoryIdentityRoots ?? [])).toEqual([
417 "/workspace/pending",
418 ...laterRoots,
419 ]);
420 }),
421 ),
422 );
423
424 it.effect("keeps a merged shell batch charged while the client withholds its ACK", () =>
425 Effect.scoped(
426 Effect.gen(function* () {
427 const update = yield* Deferred.make<void>();
428 const closed = yield* Deferred.make<void>();
429 const initial = {
430 kind: "thread.removed" as const,
431 sequence: 1,
432 location: "active" as const,
433 threadId: ThreadId.make("thread-a"),
434 };
435 const pull = yield* Stream.toPull(
436 composeShellStreamWithEnrichment({
437 initial: Stream.empty,
438 tail: bufferShellLiveStream(
439 Stream.concat(
440 Stream.succeed(initial),
441 Stream.fromEffect(Deferred.await(update)).pipe(
442 Stream.map(() => ({ ...initial, sequence: 2 })),
443 ),
444 ).pipe(Stream.ensuring(Deferred.succeed(closed, undefined))),
445 { maxItems: 1 },
446 ),
447 enrichment: Stream.empty,
448 }),
449 );
450 expect(yield* pull).toEqual([initial]);
451 // Let the merge producer run while the client still withholds its ACK.
452 yield* Effect.yieldNow;
453 yield* Deferred.succeed(update, undefined);
454 yield* Deferred.await(closed);
455 const result = yield* Stream.fromPull(Effect.succeed(pull)).pipe(
456 Stream.runDrain,
457 Effect.result,
458 );
459 expect(result._tag).toBe("Failure");
460 if (result._tag === "Failure") expect(result.failure._tag).toBe("LiveStreamBufferError");
461 }),
462 ),
463 );
464
465 it.effect(
466 "emits every initial item before enrichment even when enrichment is already ready",
467 () =>
468 Effect.gen(function* () {
469 const initialSnapshot = {
470 ...emptyShellSnapshot,
471 snapshotSequence: 5,
472 } as OrchestrationV2ShellSnapshot;
473 const enrichmentSnapshot = {
474 ...emptyShellSnapshot,
475 snapshotSequence: 10,
476 } as OrchestrationV2ShellSnapshot;
477
478 const initialItems = shellStreamItemsFromInitialSnapshot({
479 snapshot: initialSnapshot,
480 resolvedRepositoryIdentityRoots: ["/workspace/a"],
481 });
482 const enrichment = Stream.make(
483 shellStreamItemFromEnrichmentRefresh({
484 snapshot: enrichmentSnapshot,
485 changes: [{ workspaceRoot: "/workspace/b" }],
486 }),
487 );
488 const tail = Stream.make(
489 { kind: "synchronized" as const },
490 {
491 kind: "project.removed" as const,
492 sequence: 6,
493 projectId: "project-a",
494 },
495 );
496
497 const items = Array.from(
498 yield* composeShellStreamWithEnrichment({
499 initial: Stream.fromIterable(initialItems),
500 tail,
501 enrichment,
502 }).pipe(Stream.runCollect),
503 );
504
505 expect(items.slice(0, initialItems.length)).toEqual(initialItems);
506
507 const enrichmentIndex = items.findIndex(
508 (item) =>
509 item.kind === "snapshot" &&
510 "resolvedRepositoryIdentityRoots" in item &&
511 item.resolvedRepositoryIdentityRoots?.includes("/workspace/b"),
512 );
513 expect(enrichmentIndex).toBeGreaterThanOrEqual(initialItems.length);
514
515 for (let index = 0; index < initialItems.length; index++) {
516 expect(items[index]).toEqual(initialItems[index]);
517 }
518 }),
519 );
520});
521
522describe("dedupeShellEnrichment", () => {
523 const project = {
524 id: ProjectId.make("project-a"),
525 title: "A project",
526 workspaceRoot: "/workspace/a",
527 repositoryIdentity: null,
528 defaultModelSelection: null,
529 scripts: [],
530 createdAt: "2026-09-01T00:00:00Z",
531 updatedAt: "2026-09-01T00:00:00Z",
532 };
533 const initial = {
534 kind: "snapshot" as const,
535 snapshot: { ...emptyShellSnapshot, projects: [project] },
536 };
537 const marked = { ...initial, resolvedRepositoryIdentityRoots: [project.workspaceRoot] };
538
539 it.effect(
540 "keeps initial/resume resolution and real changes but drops sequence-only repeats",
541 () =>
542 Effect.gen(function* () {
543 const repeated = {
544 ...marked,
545 snapshot: { ...marked.snapshot, snapshotSequence: 10, projects: [{ ...project }] },
546 };
547 const progressed = {
548 ...repeated,
549 resolvedRepositoryIdentityRoots: [project.workspaceRoot, "/workspace/b"],
550 };
551 const changed = {
552 ...progressed,
553 snapshot: { ...progressed.snapshot, projects: [{ ...project, title: "Renamed" }] },
554 };
555 const items = yield* Stream.make(
556 initial,
557 marked,
558 repeated,
559 progressed,
560 changed,
561 changed,
562 ).pipe(dedupeShellEnrichment, Stream.runCollect);
563 expect(items).toEqual([initial, marked, progressed, changed]);
564 const resumed = Stream.make(marked, repeated).pipe(dedupeShellEnrichment);
565 expect(yield* Stream.runCollect(resumed)).toEqual([marked]);
566 expect(yield* Stream.runCollect(resumed)).toEqual([marked]);
567 }),
568 );
569
570 it.effect("keeps authoritative snapshots and invalidates metadata after project deltas", () =>
571 Effect.gen(function* () {
572 const removed = { kind: "project.removed" as const, projectId: project.id, sequence: 2 };
573 const updated = { kind: "project.updated" as const, project, sequence: 3 };
574 const withThreads = {
575 ...marked,
576 snapshot: { ...marked.snapshot, threads: [shellFixture({})] },
577 };
578 const values = [
579 initial,
580 initial,
581 marked,
582 removed,
583 marked,
584 updated,
585 marked,
586 withThreads,
587 withThreads,
588 ];
589 expect(
590 yield* Stream.fromIterable(values).pipe(dedupeShellEnrichment, Stream.runCollect),
591 ).toEqual(values);
592 }),
593 );
594
595 it.effect(
596 "deduplicates across thread deltas without dropping those deltas or identity clears",
597 () =>
598 Effect.gen(function* () {
599 const delta = {
600 kind: "thread.removed" as const,
601 threadId: ThreadId.make("thread-a"),
602 location: "active" as const,
603 sequence: 2,
604 };
605 const resolved = {
606 ...marked,
607 snapshot: {
608 ...marked.snapshot,
609 projects: [
610 {
611 ...project,
612 repositoryIdentity: {
613 canonicalKey: "github.com/test/repo",
614 locator: {
615 source: "git-remote" as const,
616 remoteName: "origin",
617 remoteUrl: "https://github.com/test/repo.git",
618 },
619 },
620 },
621 ],
622 },
623 };
624 expect(
625 yield* Stream.make(marked, delta, marked, resolved, marked).pipe(
626 dedupeShellEnrichment,
627 Stream.runCollect,
628 ),
629 ).toEqual([marked, delta, resolved, marked]);
630 }),
631 );
632});
633
634describe("skipUnchangedThreadShells", () => {
635 const at = (ms: number) => DateTime.makeUnsafe(Date.parse("2026-10-01T00:00:00.000Z") + ms);
636 const shell = (
637 id: string,
638 overrides: Partial<OrchestrationV2ThreadShell> = {},
639 ): OrchestrationV2ThreadShell => ({
640 id: ThreadId.make(id),
641 projectId: ProjectId.make("project-a"),
642 title: "Thread",
643 providerInstanceId: ProviderInstanceId.make("codex"),
644 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
645 runtimeMode: "full-access",
646 interactionMode: "default",
647 worktreePath: null,
648 activeProviderThreadId: null,
649 lineage: { rootThreadId: ThreadId.make(id), parentThreadId: null, relationshipToParent: null },
650 forkedFrom: null,
651 createdBy: "user",
652 creationSource: "web",
653 activeRunId: null,
654 latestVisibleMessage: null,
655 hasActionableProposedPlan: false,
656 itemCount: 1,
657 visibleItemCount: 1,
658 branch: null,
659 status: "running",
660 pendingRuntimeRequest: null,
661 pendingBackgroundTasks: [],
662 latestRunId: null,
663 latestUserMessageAt: null,
664 createdAt: at(0),
665 updatedAt: at(0),
666 archivedAt: null,
667 settledOverride: null,
668 settledAt: null,
669 deletedAt: null,
670 ...overrides,
671 });
672 type Step =
673 | { readonly advanceMs: number }
674 | Exclude<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" | "synchronized" }>;
675 const updated = (sequence: number, thread: OrchestrationV2ThreadShell): Step => ({
676 kind: "thread.updated",
677 sequence,
678 location: "active",
679 thread,
680 });
681 const run = (steps: ReadonlyArray<Step>) =>
682 Stream.fromIterable(steps).pipe(
683 Stream.mapEffect((step) =>
684 "advanceMs" in step
685 ? TestClock.adjust(Duration.millis(step.advanceMs)).pipe(Effect.as(null))
686 : Effect.succeed(step),
687 ),
688 Stream.filter((item) => item !== null),
689 skipUnchangedThreadShells,
690 Stream.runCollect,
691 Effect.map((items) => Array.from(items, (item) => ("sequence" in item ? item.sequence : -1))),
692 );
693
694 it.effect("sends the next update after a removal", () =>
695 Effect.gen(function* () {
696 const sent = yield* run([
697 updated(1, shell("a")),
698 { kind: "thread.removed", sequence: 2, location: "active", threadId: ThreadId.make("a") },
699 updated(3, shell("a")),
700 ]);
701 expect(sent).toEqual([1, 2, 3]);
702 }),
703 );
704});