Open History to see the last commit
1import { assert, describe, it } from "@effect/vitest";
2import {
3 AuthOrchestrationReadScope,
4 AuthSessionId,
5 CheckpointId,
6 CheckpointRef,
7 CheckpointScopeId,
8 EnvironmentAuthenticatedAuth,
9 EnvironmentAuthenticatedPrincipal,
10 EnvironmentHttpApi,
11 EnvironmentId,
12 EventId,
13 MessageId,
14 NodeId,
15 ORCHESTRATION_PROTOCOL_HEADER,
16 ORCHESTRATION_PROTOCOL_VERSION,
17 OrchestrationV2ThreadBoundedSnapshot,
18 OrchestrationV2ThreadHistoryPage,
19 OrchestrationV2ThreadStreamItem,
20 ProjectId,
21 ProviderDriverKind,
22 ProviderInstanceId,
23 RunId,
24 ThreadId,
25 TurnItemId,
26 type OrchestrationV2DomainEvent,
27 type OrchestrationV2StoredEvent,
28 type OrchestrationV2Subagent,
29 type OrchestrationV2ThreadProjection,
30 type OrchestrationV2TurnItem,
31} from "@t3tools/contracts";
32import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
33import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
34import * as NodeServices from "@effect/platform-node/NodeServices";
35import * as DateTime from "effect/DateTime";
36import * as Effect from "effect/Effect";
37import * as Layer from "effect/Layer";
38import * as Schema from "effect/Schema";
39import * as Stream from "effect/Stream";
40import * as HttpApi from "effect/http-api/HttpApi";
41import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
42import { Etag, HttpRouter } from "effect/http";
43
44import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
45import * as SqlitePersistence from "../persistence/Sqlite.ts";
46import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
47import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
48import { subscribeOrchestrationV2Thread } from "../ws.ts";
49import * as OrchestrationHttp from "./http.ts";
50import * as ProjectionStore from "./ProjectionStore.ts";
51import * as ProjectStore from "./ProjectStore.ts";
52import * as ThreadManagementService from "./ThreadManagementService.ts";
53
54const decodeBounded = Schema.decodeUnknownSync(
55 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)),
56);
57const decodeHistoryPage = Schema.decodeUnknownSync(
58 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadHistoryPage)),
59);
60const decodeStreamItem = Schema.decodeUnknownSync(
61 Schema.toCodecJson(OrchestrationV2ThreadStreamItem),
62);
63const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem));
64
65const provider = ProviderInstanceId.make("codex");
66const modelSelection = { instanceId: provider, model: "gpt-5.4" };
67const projectId = ProjectId.make("project:compact-transport");
68const PARENT = ThreadId.make("thread:compact-transport:parent");
69const FORK = ThreadId.make("thread:compact-transport:fork");
70const LONG = ThreadId.make("thread:compact-transport:long");
71const BODIES = ThreadId.make("thread:compact-transport:bodies");
72const parentRun = RunId.make("run:compact-transport:parent");
73const longRun = RunId.make("run:compact-transport:long");
74const bodiesRun = RunId.make("run:compact-transport:bodies");
75
76const store = Layer.mergeAll(
77 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
78 SqlitePersistence.layerMemory,
79);
80const management = Layer.unwrap(
81 Effect.gen(function* () {
82 const projections = yield* ProjectionStore.ProjectionStoreV2;
83 return Layer.mock(ThreadManagementService.ThreadManagementService)({
84 ensureLegacyTranscript: () => Effect.void,
85 getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie),
86 getThreadSnapshotWindow: (id, options) =>
87 projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
88 getThreadHistoryPage: (id, cursor, throughEntryId, conversationOnly) =>
89 projections
90 .getThreadHistoryPage(id, cursor, throughEntryId, conversationOnly)
91 .pipe(Effect.orDie),
92 streamStoredEventsFrom: (input) =>
93 input?.threadId === BODIES ? Stream.fromIterable(bodiesLiveTail()) : Stream.empty,
94 });
95 }),
96);
97
98class OrchestrationApi extends HttpApi.make("environment").add(
99 EnvironmentHttpApi.groups.orchestration,
100) {}
101const auth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) =>
102 effect.pipe(
103 Effect.provideService(EnvironmentAuthenticatedPrincipal, {
104 sessionId: AuthSessionId.make("compact-transport"),
105 subject: "compact-transport",
106 method: "browser-session-cookie",
107 scopes: new Set([AuthOrchestrationReadScope]),
108 }),
109 ),
110);
111
112function threadCreated(id: ThreadId, at: DateTime.Utc, forkOf?: RunId): OrchestrationV2DomainEvent {
113 return {
114 id: EventId.make(`event:${id}:created`),
115 type: "thread.created",
116 threadId: id,
117 occurredAt: at,
118 payload: {
119 createdBy: "user",
120 creationSource: "web",
121 id,
122 projectId,
123 title: String(id),
124 providerInstanceId: provider,
125 modelSelection,
126 runtimeMode: "full-access",
127 interactionMode: "default",
128 branch: null,
129 worktreePath: null,
130 activeProviderThreadId: null,
131 lineage:
132 forkOf === undefined
133 ? { parentThreadId: null, relationshipToParent: null, rootThreadId: id }
134 : { parentThreadId: PARENT, relationshipToParent: "fork", rootThreadId: PARENT },
135 forkedFrom: forkOf === undefined ? null : { type: "run", threadId: PARENT, runId: forkOf },
136 createdAt: at,
137 updatedAt: at,
138 archivedAt: null,
139 settledOverride: null,
140 settledAt: null,
141 lastVisitedAt: null,
142 deletedAt: null,
143 },
144 };
145}
146
147function runCreated(
148 threadId: ThreadId,
149 runId: RunId,
150 at: DateTime.Utc,
151): OrchestrationV2DomainEvent {
152 const rootNodeId = NodeId.make(`node:${runId}`);
153 return {
154 id: EventId.make(`event:${runId}:created`),
155 type: "run.created",
156 threadId,
157 runId,
158 nodeId: rootNodeId,
159 driver: ProviderDriverKind.make("codex"),
160 occurredAt: at,
161 payload: {
162 id: runId,
163 threadId,
164 ordinal: 1,
165 providerInstanceId: provider,
166 modelSelection,
167 providerThreadId: null,
168 userMessageId: MessageId.make(`message:${runId}`),
169 rootNodeId,
170 activeAttemptId: null,
171 status: "completed",
172 requestedAt: at,
173 startedAt: at,
174 completedAt: at,
175 checkpointId: null,
176 contextHandoffId: null,
177 },
178 };
179}
180
181function itemUpdated(
182 threadId: ThreadId,
183 runId: RunId | null,
184 key: string,
185 ordinal: number,
186 at: DateTime.Utc,
187 type:
188 | "command_execution"
189 | "run_interrupt_request"
190 | "run_interrupt_result" = "command_execution",
191): OrchestrationV2DomainEvent {
192 const base = {
193 id: TurnItemId.make(`turn-item:${threadId}:${key}`),
194 threadId,
195 runId,
196 nodeId: null,
197 providerThreadId: null,
198 providerTurnId: null,
199 nativeItemRef: null,
200 parentItemId: null,
201 ordinal,
202 status: "completed" as const,
203 title: key,
204 startedAt: at,
205 completedAt: at,
206 updatedAt: at,
207 };
208 const payload: OrchestrationV2TurnItem =
209 type === "command_execution"
210 ? { ...base, type, input: `echo ${key}`, output: `output ${key}`, exitCode: 0 }
211 : { ...base, type, message: key };
212 return {
213 id: EventId.make(`event:${threadId}:${key}`),
214 type: "turn-item.updated",
215 threadId,
216 ...(runId === null ? {} : { runId }),
217 occurredAt: at,
218 payload,
219 };
220}
221
222// A conversation turn whose records (message, checkpoint, subagent) repeat in its turn items.
223function bodiesEvents(at: DateTime.Utc): OrchestrationV2DomainEvent[] {
224 const event = (key: string, rest: Pick<OrchestrationV2DomainEvent, "type" | "payload">) =>
225 ({
226 id: EventId.make(`event:${BODIES}:${key}`),
227 threadId: BODIES,
228 runId: bodiesRun,
229 occurredAt: at,
230 ...rest,
231 }) as OrchestrationV2DomainEvent;
232 const item = (key: string, ordinal: number, fields: object) =>
233 event(`${key}-item`, {
234 type: "turn-item.updated",
235 payload: {
236 id: TurnItemId.make(`turn-item:${BODIES}:${key}`),
237 threadId: BODIES,
238 runId: bodiesRun,
239 nodeId: null,
240 providerThreadId: null,
241 providerTurnId: null,
242 nativeItemRef: null,
243 parentItemId: null,
244 ordinal,
245 status: "completed",
246 title: null,
247 startedAt: at,
248 completedAt: at,
249 updatedAt: at,
250 ...fields,
251 } as OrchestrationV2TurnItem,
252 });
253 const message = (key: string, role: "user" | "assistant", text: string) =>
254 event(`${key}-message`, {
255 type: "message.updated",
256 payload: {
257 id: MessageId.make(`message:${key === "prompt" ? bodiesRun : `${BODIES}:${key}`}`),
258 threadId: BODIES,
259 runId: bodiesRun,
260 nodeId: null,
261 role,
262 text,
263 attachments: [],
264 streaming: false,
265 createdBy: role === "user" ? "user" : "agent",
266 creationSource: role === "user" ? "web" : "provider",
267 createdAt: at,
268 updatedAt: at,
269 },
270 });
271 const files = [{ path: "src/index.ts", kind: "modified", additions: 4, deletions: 2 }];
272 return [
273 threadCreated(BODIES, at),
274 runCreated(BODIES, bodiesRun, at),
275 message("prompt", "user", "Please fix the build."),
276 item("prompt", 1, {
277 type: "user_message",
278 messageId: MessageId.make(`message:${bodiesRun}`),
279 inputIntent: "turn_start",
280 text: "Please fix the build.",
281 attachments: [],
282 createdBy: "user",
283 creationSource: "web",
284 }),
285 event("task-record", { type: "subagent.updated", payload: bodiesSubagent(at) }),
286 item("task", 2, {
287 type: "subagent",
288 subagentId: bodiesSubagent(at).id,
289 origin: "app_owned",
290 driver: "codex",
291 providerInstanceId: provider,
292 childThreadId: null,
293 prompt: bodiesSubagent(at).prompt,
294 result: bodiesSubagent(at).result,
295 }),
296 message("reply", "assistant", "Fixed the import."),
297 item("reply", 3, {
298 type: "assistant_message",
299 messageId: MessageId.make(`message:${BODIES}:reply`),
300 text: "Fixed the import.",
301 streaming: false,
302 }),
303 event("checkpoint-record", {
304 type: "checkpoint.captured",
305 payload: {
306 id: CheckpointId.make(`checkpoint:${BODIES}`),
307 threadId: BODIES,
308 scopeId: CheckpointScopeId.make(`scope:${BODIES}`),
309 runId: bodiesRun,
310 nodeId: NodeId.make(`node:${bodiesRun}`),
311 parentCheckpointId: null,
312 ordinalWithinScope: 1,
313 appRunOrdinal: 1,
314 ref: CheckpointRef.make(`refs/t3/${BODIES}`),
315 status: "ready",
316 files,
317 capturedAt: at,
318 },
319 }),
320 item("checkpoint", 4, {
321 type: "checkpoint",
322 checkpointId: CheckpointId.make(`checkpoint:${BODIES}`),
323 scopeId: CheckpointScopeId.make(`scope:${BODIES}`),
324 files,
325 }),
326 ];
327}
328
329function bodiesSubagent(at: DateTime.Utc): OrchestrationV2Subagent {
330 return {
331 id: NodeId.make(`node:${BODIES}:task`),
332 threadId: BODIES,
333 runId: bodiesRun,
334 parentNodeId: NodeId.make(`node:${bodiesRun}`),
335 origin: "app_owned",
336 createdBy: "agent",
337 driver: ProviderDriverKind.make("codex"),
338 providerInstanceId: provider,
339 providerThreadId: null,
340 childThreadId: null,
341 nativeTaskRef: null,
342 prompt: "Find the failing import.",
343 title: null,
344 model: null,
345 status: "completed",
346 result: "It was src/index.ts.",
347 startedAt: at,
348 completedAt: at,
349 updatedAt: at,
350 };
351}
352
353// Live after the snapshot: a reply pair, then a subagent update that repeats its prompt.
354function bodiesLiveTail(): OrchestrationV2StoredEvent[] {
355 const at = DateTime.makeUnsafe("2026-10-01T00:00:00.000Z");
356 const [reply, replyItem] = bodiesEvents(at).filter((event) =>
357 String(event.id).includes(":reply-"),
358 );
359 return [
360 reply!,
361 replyItem!,
362 {
363 id: EventId.make(`event:${BODIES}:task-progress`),
364 type: "subagent.updated",
365 threadId: BODIES,
366 runId: bodiesRun,
367 occurredAt: at,
368 payload: { ...bodiesSubagent(at), progress: "Checked the build." },
369 } as OrchestrationV2DomainEvent,
370 ].map((event, index) => ({ sequence: 1_000 + index, commandId: null, event }));
371}
372
373const seed = Effect.gen(function* () {
374 const projections = yield* ProjectionStore.ProjectionStoreV2;
375 const at = yield* DateTime.now;
376 const events: OrchestrationV2DomainEvent[] = [
377 threadCreated(PARENT, at),
378 runCreated(PARENT, parentRun, at),
379 ...[1, 2, 3].map((n) => itemUpdated(PARENT, parentRun, `parent-${n}`, n, at)),
380 threadCreated(FORK, at, parentRun),
381 ...[1, 2, 3, 4].map((n) => itemUpdated(FORK, null, `fork-${n}`, n, at)),
382 threadCreated(LONG, at),
383 runCreated(LONG, longRun, at),
384 itemUpdated(LONG, longRun, "request", 1, at, "run_interrupt_request"),
385 ...Array.from({ length: 90 }, (_, index) =>
386 itemUpdated(LONG, longRun, `tool-${index}`, index + 2, at),
387 ),
388 itemUpdated(LONG, longRun, "result", 92, at, "run_interrupt_result"),
389 ...bodiesEvents(at),
390 ];
391 for (const event of events) yield* projections.apply(event);
392});
393
394const TestLayer = Layer.mergeAll(
395 management,
396 Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, {
397 getEnvironmentId: Effect.succeed(EnvironmentId.make("compact-transport")),
398 }),
399 Layer.effectDiscard(seed),
400 Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}),
401 Layer.mock(ProjectStore.ProjectStoreV2)({}),
402 Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({}),
403).pipe(Layer.provideMerge(store));
404
405const withHttp = <A>(use: (get: (path: string) => Promise<Response>) => Promise<A>) =>
406 Effect.gen(function* () {
407 const context = yield* Effect.context<Layer.Success<typeof TestLayer>>();
408 const appLayer = HttpApiBuilder.layer(OrchestrationApi).pipe(
409 Layer.provide(OrchestrationHttp.layer),
410 Layer.provide(auth),
411 Layer.provide(Layer.succeedContext(context)),
412 Layer.provide(NodeHttpPlatform.layer),
413 Layer.provide(Etag.layerWeak),
414 Layer.provide(NodeServices.layer),
415 );
416 return yield* Effect.acquireUseRelease(
417 Effect.sync(() => HttpRouter.toWebHandler(appLayer, { disableLogger: true })),
418 ({ handler }) =>
419 Effect.promise(() =>
420 use((path) =>
421 handler(
422 new Request(`http://environment.test${path}`, {
423 headers: {
424 [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION),
425 },
426 }),
427 ),
428 ),
429 ),
430 ({ dispose }) => Effect.promise(() => dispose()),
431 );
432 });
433
434const firstSnapshot = (input: Parameters<typeof subscribeOrchestrationV2Thread>[0]) =>
435 subscribeOrchestrationV2Thread(input).pipe(
436 Effect.flatMap((stream) => Stream.runHead(stream)),
437 Effect.map((head) => {
438 if (head._tag !== "Some" || head.value.kind !== "snapshot") throw new Error("No snapshot");
439 const decoded = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(head.value))));
440 if (decoded.kind !== "snapshot") throw new Error("No snapshot");
441 return { raw: head.value, decoded };
442 }),
443 );
444
445function itemIds(projection: OrchestrationV2ThreadProjection): string[] {
446 return projection.turnItems.map((item) => String(item.id));
447}
448
449it.layer(TestLayer)("compact bounded snapshot transport", (it) => {
450 describe.each([
451 ["fork", FORK],
452 ["long run with a retained interrupt request", LONG],
453 ])("%s", (_, threadId) => {
454 it.effect("HTTP: legacy, ignored and opted-in requests restore to the same snapshot", () =>
455 withHttp(async (get) => {
456 const base = `/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`;
457 const legacyBody = await (await get(base)).text();
458 const ignoredBody = await (await get(`${base}?compactTurnItems=yes&unknown=1`)).text();
459 const compactResponse = await get(`${base}?compactTurnItems=1`);
460 assert.strictEqual(compactResponse.status, 200);
461 const compactBody = await compactResponse.text();
462
463 assert.notInclude(legacyBody, "turnItemsOmitLocalVisible");
464 assert.strictEqual(ignoredBody, legacyBody);
465 assert.isBelow(compactBody.length, legacyBody.length);
466
467 const legacy = decodeBounded(legacyBody);
468 const compact = decodeBounded(compactBody);
469 assert.isTrue(compact.turnItemsOmitLocalVisible);
470 const { turnItemsOmitLocalVisible: _marker, projection: _p, ...compactRest } = compact;
471 const { projection: _legacyProjection, ...legacyRest } = legacy;
472 assert.deepStrictEqual(compactRest, legacyRest);
473 assert.deepStrictEqual(boundedSnapshotProjection(compact), legacy.projection);
474 return { legacy, compact };
475 }).pipe(
476 Effect.tap(({ legacy }) =>
477 Effect.sync(() => {
478 if (threadId === FORK) {
479 const visibility = legacy.projection.visibleTurnItems.map((row) => row.visibility);
480 assert.deepStrictEqual(visibility, [
481 "inherited",
482 "inherited",
483 "inherited",
484 "synthetic",
485 "local",
486 "local",
487 "local",
488 "local",
489 ]);
490 } else {
491 const ids = itemIds(legacy.projection);
492 assert.isTrue(legacy.hasMoreHistory);
493 assert.strictEqual(ids.at(-1), `turn-item:${LONG}:request`);
494 assert.isFalse(
495 legacy.projection.visibleTurnItems.some((row) => row.sourceItemId === ids.at(-1)),
496 );
497 }
498 }),
499 ),
500 ),
501 );
502
503 it.effect("HTTP: the ID table is opt-in and decodes to the same snapshot", () =>
504 withHttp(async (get) => {
505 const base = `/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded?compactTurnItems=1`;
506 const plainBody = await (await get(base)).text();
507 const tableBody = await (await get(`${base}&compactIds=1`)).text();
508 const wire = JSON.parse(tableBody);
509
510 assert.notProperty(JSON.parse(plainBody), "ids");
511 assert.isArray(wire.ids);
512 assert.isNumber(wire.projection.visibleTurnItems[0].item.id);
513 assert.isNumber(wire.projection.thread.id);
514 assert.isBelow(tableBody.length, plainBody.length);
515 const decoded = decodeBounded(tableBody);
516 assert.deepStrictEqual(decoded, decodeBounded(plainBody));
517
518 // Only the long thread has older history to page.
519 if (threadId !== LONG) return;
520 assert.isNotNull(decoded.historyCursor);
521 const history = `/api/orchestration/threads/${encodeURIComponent(threadId)}/history?cursor=${encodeURIComponent(decoded.historyCursor ?? "")}`;
522 const plainPage = await (await get(history)).text();
523 const tablePage = await (await get(`${history}&compactIds=1`)).text();
524 assert.isArray(JSON.parse(tablePage).ids);
525 const page = decodeHistoryPage(tablePage);
526 assert.isAbove(page.items.length, 0);
527 assert.deepStrictEqual(page, decodeHistoryPage(plainPage));
528 }),
529 );
530
531 it.effect("WS: the ID table is opt-in and decodes to the same snapshot", () =>
532 Effect.gen(function* () {
533 const input = { threadId, acceptBoundedSnapshot: true, acceptCompactTurnItems: true };
534 const plain = yield* firstSnapshot(input);
535 const table = yield* firstSnapshot({ ...input, acceptCompactIds: true });
536
537 assert.notProperty(encodeStreamItem(plain.raw), "ids");
538 assert.property(encodeStreamItem(table.raw), "ids");
539 assert.deepStrictEqual(table.decoded, plain.decoded);
540 }),
541 );
542
543 it.effect("WS: bounded snapshots stay unchanged unless the client opts in", () =>
544 Effect.gen(function* () {
545 const legacy = yield* firstSnapshot({ threadId, acceptBoundedSnapshot: true });
546 const compact = yield* firstSnapshot({
547 threadId,
548 acceptBoundedSnapshot: true,
549 acceptCompactTurnItems: true,
550 });
551 const full = yield* firstSnapshot({ threadId, acceptCompactTurnItems: true });
552
553 assert.isUndefined(legacy.decoded.turnItemsOmitLocalVisible);
554 assert.isUndefined(full.decoded.turnItemsOmitLocalVisible);
555 assert.isUndefined(full.decoded.historyCursor);
556 assert.isTrue(compact.decoded.turnItemsOmitLocalVisible);
557 assert.isBelow(
558 compact.decoded.projection.turnItems.length,
559 legacy.decoded.projection.turnItems.length,
560 );
561 assert.deepStrictEqual(
562 boundedSnapshotProjection(compact.decoded),
563 legacy.decoded.projection,
564 );
565 assert.deepStrictEqual(
566 {
567 ...compact.decoded,
568 projection: legacy.decoded.projection,
569 turnItemsOmitLocalVisible: undefined,
570 },
571 { ...legacy.decoded, turnItemsOmitLocalVisible: undefined },
572 );
573 }),
574 );
575 });
576
577 describe("deduped bodies", () => {
578 const markers = ["textOmitted", "filesOmitted", "promptOmitted", "resultOmitted"];
579
580 it.effect(
581 "HTTP: opted-in snapshots blank repeated bodies and restore to the same snapshot",
582 () =>
583 withHttp(async (get) => {
584 const base = `/api/orchestration/threads/${encodeURIComponent(BODIES)}/bounded`;
585 const compactBody = await (await get(`${base}?compactTurnItems=1`)).text();
586 const dedupedBody = await (await get(`${base}?compactTurnItems=1&dedupeBodies=1`)).text();
587
588 for (const marker of markers) {
589 assert.notInclude(compactBody, marker);
590 assert.include(dedupedBody, marker);
591 }
592 assert.isBelow(dedupedBody.length, compactBody.length);
593 assert.deepStrictEqual(
594 boundedSnapshotProjection(decodeBounded(dedupedBody)),
595 boundedSnapshotProjection(decodeBounded(compactBody)),
596 );
597 // The client sends both opt-ins: bodies restore after the ID table resolves.
598 const tabledBody = await (
599 await get(`${base}?compactTurnItems=1&dedupeBodies=1&compactIds=1`)
600 ).text();
601 assert.property(JSON.parse(tabledBody), "ids");
602 assert.deepStrictEqual(
603 boundedSnapshotProjection(decodeBounded(tabledBody)),
604 boundedSnapshotProjection(decodeBounded(compactBody)),
605 );
606 }),
607 );
608
609 it.effect(
610 "WS: bounded and full snapshots restore, and the live tail sends each body once",
611 () =>
612 Effect.gen(function* () {
613 const subscribe = (
614 input: Partial<Parameters<typeof subscribeOrchestrationV2Thread>[0]>,
615 ) =>
616 subscribeOrchestrationV2Thread({ threadId: BODIES, ...input }).pipe(
617 Effect.flatMap(Stream.runCollect),
618 Effect.map((items) =>
619 Array.from(items, (item) =>
620 decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(item)))),
621 ),
622 ),
623 );
624 const bounded = { acceptBoundedSnapshot: true, acceptCompactTurnItems: true };
625 for (const mode of [bounded, {}]) {
626 const [plainSnapshot, ...plainTail] = yield* subscribe(mode);
627 const [dedupedSnapshot, ...dedupedTail] = yield* subscribe({
628 ...mode,
629 acceptDedupedBodies: true,
630 });
631 if (plainSnapshot?.kind !== "snapshot" || dedupedSnapshot?.kind !== "snapshot") {
632 throw new Error("No snapshot");
633 }
634 assert.deepStrictEqual(
635 boundedSnapshotProjection(dedupedSnapshot),
636 boundedSnapshotProjection(plainSnapshot),
637 );
638 const payloads = (tail: typeof plainTail) =>
639 tail.map((item) => (item.kind === "event" ? item.event.payload : null));
640 const [reply, replyItem, progress] = payloads(plainTail);
641 assert.deepStrictEqual(payloads(dedupedTail), [
642 reply,
643 { ...replyItem, text: "", textOmitted: true },
644 // The snapshot delivered this subagent's record, so its unchanged bodies repeat.
645 { ...progress, prompt: "", promptOmitted: true, result: null, resultOmitted: true },
646 ]);
647 }
648 }),
649 );
650 });
651});