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