1import {
2 CheckpointId,
3 CheckpointRef,
4 CheckpointScopeId,
5 MessageId,
6 ContextHandoffId,
7 ProviderSessionId,
8 ProviderThreadId,
9 NodeId,
10 EventId,
11 ProjectId,
12 ProviderDriverKind,
13 ProviderInstanceId,
14 ThreadId,
15 TurnItemId,
16 type OrchestrationV2DomainEvent,
17 type OrchestrationV2StoredEvent,
18 type OrchestrationV2ThreadProjection,
19 type OrchestrationV2ThreadStreamItem,
20 OrchestrationV2TurnItem,
21 OrchestrationV2TurnItemJson,
22} from "@t3tools/contracts";
23import { describe, expect, it } from "@effect/vitest";
24import * as DateTime from "effect/DateTime";
25import * as Schema from "effect/Schema";
26​
27import { coalesceLiveUpdatedEvents } from "./ThreadLiveEventCoalescer.ts";
28import {
29 makeLiveBodyDeduper,
30 projectTurnItemForWire,
31 projectTurnItemForDetail,
32 projectDomainEventForWire,
33 projectStoredEventForWire,
34} from "./WireProjection.ts";
35import { threadShellFromProjection } from "./ProjectionStore.ts";
36import { MAX_TOOL_OUTPUT_IMAGES, toolOutputImages } from "@t3tools/shared/toolOutput";
37​
38const decodeTurnItem = Schema.decodeUnknownSync(OrchestrationV2TurnItem);
39const encodeTurnItemJson = Schema.encodeSync(OrchestrationV2TurnItemJson);
40const decodeTurnItemJson = Schema.decodeUnknownSync(OrchestrationV2TurnItemJson);
41​
42const base = {
43 id: TurnItemId.make("tool-1"),
44 type: "dynamic_tool" as const,
45 threadId: ThreadId.make("thread-1"),
46 runId: null,
47 nodeId: null,
48 providerThreadId: null,
49 providerTurnId: null,
50 nativeItemRef: null,
51 parentItemId: null,
52 ordinal: 1,
53 status: "completed" as const,
54 title: "MCP tool",
55 toolName: "mcp__github__fetch_pr",
56 input: { pr: 42 },
57 startedAt: DateTime.makeUnsafe("2026-08-13T00:00:00.000Z"),
58 completedAt: DateTime.makeUnsafe("2026-08-13T00:00:01.000Z"),
59 updatedAt: DateTime.makeUnsafe("2026-08-13T00:00:01.000Z"),
60};
61​
62describe("orchestration V2 wire projection", () => {
63 it("keeps copied handoff transcripts out of activity items and live events", () => {
64 const item = {
65 ...base,
66 type: "handoff" as const,
67 contextHandoffId: ContextHandoffId.make("handoff:wire"),
68 fromProviderThreadIds: [],
69 toProviderThreadId: ProviderThreadId.make("target"),
70 fromProviderInstanceIds: [],
71 toProviderInstanceId: ProviderInstanceId.make("codex"),
72 strategy: "full_thread_summary" as const,
73 summary: "PRIVATE_HANDOFF_TRANSCRIPT",
74 };
75 const projected = projectTurnItemForWire(item);
76 expect(projected).not.toHaveProperty("summary");
77 expect(decodeTurnItem(projected)).toMatchObject({ contextHandoffId: item.contextHandoffId });
78 const event = {
79 id: EventId.make("handoff:wire"),
80 type: "turn-item.updated" as const,
81 threadId: base.threadId,
82 occurredAt: base.updatedAt,
83 payload: item,
84 };
85 expect(JSON.stringify(projectDomainEventForWire(event))).not.toContain(item.summary);
86 expect(item.summary).toBe("PRIVATE_HANDOFF_TRANSCRIPT");
87 });
88​
89 it("preserves image metadata through wire and JSON contracts while redacting output", () => {
90 const item = {
91 ...base,
92 toolName: "Read",
93 input: { file_path: "/workspace/reference.png" },
94 viewedImagePath: "/workspace/reference.png",
95 output: { data: "private-image-data" },
96 } satisfies OrchestrationV2TurnItem;
97 const projected = projectTurnItemForWire(item);
98 const live = decodeTurnItem(projected);
99 const encoded = encodeTurnItemJson(live);
100 const json = decodeTurnItemJson(encoded);
101 const decoded = decodeTurnItem(json);
102 expect(decoded).toMatchObject({ viewedImagePath: "/workspace/reference.png" });
103 expect(decoded).not.toHaveProperty("output");
104 expect(item.output.data).toBe("private-image-data");
105 });
106​
107 it("keeps transcript bodies out of shell rows", () => {
108 const now = DateTime.makeUnsafe("2026-08-13T00:00:00.000Z");
109 const threadId = ThreadId.make("thread-shell-budget");
110 const projection = {
111 thread: {
112 createdBy: "user",
113 creationSource: "web",
114 id: threadId,
115 projectId: ProjectId.make("project-shell-budget"),
116 title: "Shell payload budget",
117 providerInstanceId: ProviderInstanceId.make("codex"),
118 modelSelection: null,
119 runtimeMode: "full-access",
120 interactionMode: "default",
121 branch: null,
122 worktreePath: null,
123 lineage: {
124 parentThreadId: null,
125 relationshipToParent: null,
126 rootThreadId: threadId,
127 },
128 forkedFrom: null,
129 activeProviderThreadId: null,
130 createdAt: now,
131 updatedAt: now,
132 archivedAt: null,
133 settledOverride: null,
134 settledAt: null,
135 deletedAt: null,
136 },
137 runs: [],
138 attempts: [],
139 runtimeRequests: [],
140 messages: [
141 {
142 id: MessageId.make("message-shell-budget"),
143 role: "assistant",
144 text: "x".repeat(1_000_000),
145 updatedAt: now,
146 },
147 ],
148 providerSessions: [],
149 providerThreads: [],
150 turnItems: [],
151 visibleTurnItems: [],
152 plans: [],
153 updatedAt: now,
154 } as unknown as OrchestrationV2ThreadProjection;
155​
156 const shell = threadShellFromProjection(projection);
157​
158 expect(shell.latestVisibleMessage).toBeNull();
159 expect(JSON.stringify(shell).length).toBeLessThan(2_000);
160 });
161​
162 it("omits even small dynamic tool results while retaining input", () => {
163 const item = { ...base, output: { ok: true } } satisfies OrchestrationV2TurnItem;
164 expect(projectTurnItemForWire(item)).toEqual({ ...base, outputOmitted: true });
165 expect(item.output).toEqual({ ok: true });
166 });
167​
168 it("uses encoded JSON bytes for strings near the dynamic-value limit", () => {
169 const small = '"'.repeat(8_191);
170 const large = '"'.repeat(8_192);
171 const smallItem = { ...base, input: small };
172 expect(projectTurnItemForWire(smallItem)).toEqual(smallItem);
173 const projected = projectTurnItemForWire({ ...base, input: large });
174 expect(projected.type === "dynamic_tool" ? projected.input : null).toEqual({
175 summary: '"'.repeat(159) + "…",
176 truncated: true,
177 });
178​
179 const boundary = projectTurnItemForWire({
180 ...base,
181 input: { text: "a".repeat(149) + "😀" + "b".repeat(17_000) },
182 });
183 expect(boundary.type === "dynamic_tool" ? boundary.input : null).toEqual({
184 summary: '{"text":"' + "a".repeat(149) + "…",
185 truncated: true,
186 });
187 });
188​
189 it("truncates detail at a UTF-8 boundary without changing the source", () => {
190 const marker = "\n… output truncated for transport";
191 const prefixBytes = 32_768 - Buffer.byteLength(marker);
192 const progress = "a".repeat(prefixBytes - 1) + "😀" + "x".repeat(100_000);
193 const item = {
194 ...base,
195 type: "subagent" as const,
196 subagentId: NodeId.make("child-agent"),
197 origin: "app_owned" as const,
198 driver: ProviderDriverKind.make("codex"),
199 providerInstanceId: ProviderInstanceId.make("codex"),
200 childThreadId: null,
201 prompt: "æ¼¢".repeat(12_000),
202 result: null,
203 progress,
204 };
205 const projected = projectTurnItemForWire(item);
206 expect(projectTurnItemForWire(projected)).toEqual(projected);
207 expect(
208 Buffer.byteLength(projected.type === "subagent" ? projected.prompt : ""),
209 ).toBeLessThanOrEqual(32_768);
210 expect(projected.type === "subagent" ? projected.progress : null).toBe(
211 "a".repeat(prefixBytes - 1) + marker,
212 );
213 expect(item.progress).toBe(progress);
214 });
215​
216 it("leaves a screenshot's bytes out of the fetched tool output", () => {
217 const data = "A".repeat(600_000);
218 const item = {
219 ...base,
220 type: "dynamic_tool" as const,
221 toolName: "mcp__t3-code__device_screenshot",
222 input: { deviceId: "phone" },
223 output: {
224 content: [{ type: "image", source: { type: "base64", media_type: "image/png", data } }],
225 },
226 };
227 expect(projectTurnItemForDetail(item)).toMatchObject({
228 output: { content: [{ type: "image", mimeType: "image/png" }] },
229 });
230 expect(item.output.content[0]?.source.data).toBe(data);
231 });
232​
233 it("keeps image markers when the rest of a tool output is too large to send", () => {
234 const image = {
235 type: "image",
236 source: { type: "base64", media_type: "image/png", data: "AAAA" },
237 };
238 const item = {
239 ...base,
240 type: "dynamic_tool" as const,
241 output: { content: [{ type: "text", text: "x".repeat(300_000) }, image] },
242 };
243 const projected = projectTurnItemForDetail(item);
244 const output = projected.type === "dynamic_tool" ? projected.output : null;
245 expect(Array.isArray(output) ? output.slice(1) : null).toEqual([
246 { type: "image", mimeType: "image/png" },
247 ]);
248 expect(JSON.stringify(output).length).toBeLessThan(270_000);
249 });
250​
251 it("bounds a tool output made of many images", () => {
252 const image = { type: "image", data: "AAAA", mimeType: "image/png" };
253 const item = {
254 ...base,
255 type: "dynamic_tool" as const,
256 output: Array.from({ length: 10_000 }, () => image),
257 };
258 const projected = projectTurnItemForDetail(item);
259 const output = projected.type === "dynamic_tool" ? projected.output : null;
260 expect(Array.isArray(output) ? output.length : null).toBe(1 + MAX_TOOL_OUTPUT_IMAGES);
261 expect(Array.isArray(output) ? String(output[0]?.text).length : null).toBeLessThan(263_000);
262 expect(toolOutputImages(output)).toHaveLength(MAX_TOOL_OUTPUT_IMAGES);
263 });
264​
265 it("keeps failure evidence without retaining command output", () => {
266 const item = {
267 ...base,
268 type: "command_execution" as const,
269 input: "cat missing-file",
270 output: "cat: missing-file: No such file or directory",
271 };
272 const projected = projectTurnItemForWire(item);
273 expect(projected).not.toHaveProperty("output");
274 expect(projected).toMatchObject({ outputIndicatesFailure: true });
275 expect(projectTurnItemForWire(projected)).toEqual(projected);
276 expect(projectTurnItemForWire({ ...item, output: "", exitCode: 2 })).toMatchObject({
277 outputIndicatesFailure: true,
278 exitCode: 2,
279 });
280 });
281​
282 it.each([
283 "logs/app.log:12: Error: ENOENT: no such file or directory",
284 "fixtures/ci.txt:3: Process exited with exit code 2",
285 ])("trusts a zero exit code over failure text in the output: %s", (output) => {
286 const projected = projectTurnItemForWire({
287 ...base,
288 type: "command_execution" as const,
289 input: "rg -n error logs fixtures",
290 output,
291 exitCode: 0,
292 });
293 expect(projected).not.toHaveProperty("outputIndicatesFailure");
294 expect(projected).toMatchObject({ exitCode: 0, outputOmitted: true });
295 });
296​
297 it("omits inline file bodies but preserves file identity and change counts", () => {
298 const item = {
299 ...base,
300 type: "file_change" as const,
301 fileName: "src/main.ts",
302 additions: 3,
303 deletions: 1,
304 diffStr: "+new code",
305 oldStr: "old code",
306 newStr: "new code",
307 };
308 const projected = projectTurnItemForWire(item);
309 expect(projected).not.toHaveProperty("diffStr");
310 expect(projected).not.toHaveProperty("oldStr");
311 expect(projected).not.toHaveProperty("newStr");
312 expect(projected).toMatchObject({ fileName: "src/main.ts", additions: 3, deletions: 1 });
313 expect(item.diffStr).toBe("+new code");
314 expect(projectTurnItemForDetail(item)).toMatchObject({ diffStr: "+new code" });
315 expect(projectTurnItemForDetail(item)).not.toHaveProperty("oldStr");
316 expect(projectTurnItemForDetail(item)).not.toHaveProperty("newStr");
317 const { diffStr: _diff, ...contentOnly } = item;
318 expect(projectTurnItemForDetail(contentOnly)).toMatchObject({
319 oldStr: "old code",
320 newStr: "new code",
321 });
322 expect(projectTurnItemForWire(contentOnly)).not.toHaveProperty("newStr");
323 const failed = projectTurnItemForWire({
324 ...item,
325 status: "failed",
326 diffStr: "String to replace not found",
327 });
328 expect(failed).toMatchObject({ diffStr: "String to replace not found" });
329 expect(failed).not.toHaveProperty("newStr");
330 });
331​
332 it("retains only result identities and failure metadata in live tool events", () => {
333 const output = {
334 isError: true,
335 structuredContent: { threadId: "child-thread", messageId: "message", text: "PRIVATE_BODY" },
336 content: [{ type: "text", text: "PRIVATE_BODY" }],
337 };
338 const event = {
339 id: EventId.make("result-event"),
340 type: "turn-item.updated" as const,
341 threadId: base.threadId,
342 occurredAt: base.updatedAt,
343 payload: { ...base, output },
344 };
345 const projected = projectDomainEventForWire(event);
346 expect(projected.payload).toMatchObject({
347 output: { isError: true, threadId: "child-thread", messageId: "message" },
348 });
349 expect(JSON.stringify(projected)).not.toContain("PRIVATE_BODY");
350 expect(event.payload.output).toBe(output);
351 expect(projectDomainEventForWire(projected)).toEqual(projected);
352 });
353​
354 it("shares one wire projection per stored event", () => {
355 const tool: OrchestrationV2StoredEvent = {
356 sequence: 1,
357 commandId: null,
358 event: {
359 id: EventId.make("shared-tool-event"),
360 type: "turn-item.updated",
361 threadId: base.threadId,
362 occurredAt: base.updatedAt,
363 payload: { ...base, output: { content: [{ type: "text", text: "PRIVATE_BODY" }] } },
364 },
365 };
366 const projected = projectStoredEventForWire(tool);
367 expect(projected).toEqual({ ...tool, event: projectDomainEventForWire(tool.event) });
368 expect(projectStoredEventForWire(tool)).toBe(projected);
369 const detached: OrchestrationV2StoredEvent = {
370 sequence: 2,
371 commandId: null,
372 event: {
373 id: EventId.make("detached-event"),
374 type: "provider-session.detached",
375 threadId: base.threadId,
376 occurredAt: base.updatedAt,
377 payload: {
378 providerSessionId: ProviderSessionId.make("session"),
379 detachedAt: base.updatedAt,
380 },
381 },
382 };
383 expect(projectStoredEventForWire(detached)).toBe(detached);
384 });
385});
386​
387describe("live body deduper", () => {
388 const at = base.updatedAt;
389 const messageId = MessageId.make("message-live");
390 const event = (
391 sequence: number,
392 rest: Pick<OrchestrationV2DomainEvent, "type" | "payload">,
393 ): Extract<OrchestrationV2ThreadStreamItem, { kind: "event" }> => ({
394 kind: "event",
395 sequence,
396 event: {
397 id: EventId.make(`event-${sequence}`),
398 threadId: base.threadId,
399 occurredAt: at,
400 ...rest,
401 } as OrchestrationV2DomainEvent,
402 });
403 const message = (sequence: number, text: string) =>
404 event(sequence, {
405 type: "message.updated",
406 payload: {
407 id: messageId,
408 threadId: base.threadId,
409 runId: null,
410 nodeId: null,
411 role: "assistant",
412 text,
413 attachments: [],
414 streaming: true,
415 createdBy: "agent",
416 creationSource: "provider",
417 createdAt: at,
418 updatedAt: at,
419 },
420 });
421 const reply = (sequence: number, text: string) =>
422 event(sequence, {
423 type: "turn-item.updated",
424 payload: {
425 ...base,
426 status: "running",
427 type: "assistant_message",
428 messageId,
429 text,
430 streaming: true,
431 },
432 });
433 const payloadOf = (item: OrchestrationV2ThreadStreamItem) =>
434 item.kind === "event" ? item.event.payload : undefined;
435​
436 it("blanks a reply that repeats the message delivered before it, after coalescing", () => {
437 const delivered = coalesceLiveUpdatedEvents([
438 message(1, "a"),
439 reply(2, "a"),
440 message(3, "ab"),
441 reply(4, "ab"),
442 ]).map(makeLiveBodyDeduper());
443​
444 expect(
445 delivered.map((item) => [item.kind === "event" && item.sequence, payloadOf(item)]),
446 ).toEqual([
447 [3, message(3, "ab").event.payload],
448 [4, { ...reply(4, "ab").event.payload, text: "", textOmitted: true }],
449 ]);
450 });
451​
452 it("caps a large checkpoint's files for the wire and still dedupes its item", () => {
453 const files = Array.from({ length: 250 }, (_, index) => ({
454 path: `vendor/file-${index}.ts`,
455 kind: "modified",
456 additions: 2,
457 deletions: 1,
458 }));
459 const checkpointId = CheckpointId.make("checkpoint-large");
460 const scopeId = CheckpointScopeId.make("scope-large");
461 const captured = event(1, {
462 type: "checkpoint.captured",
463 payload: {
464 id: checkpointId,
465 threadId: base.threadId,
466 scopeId,
467 runId: null,
468 nodeId: NodeId.make("node-large"),
469 parentCheckpointId: null,
470 ordinalWithinScope: 1,
471 appRunOrdinal: null,
472 ref: CheckpointRef.make("refs/t3/checkpoints/large"),
473 status: "ready",
474 files,
475 capturedAt: at,
476 },
477 });
478 const item = event(2, {
479 type: "turn-item.updated",
480 payload: { ...base, type: "checkpoint", checkpointId, scopeId, files },
481 });
482 const dedupe = makeLiveBodyDeduper();
483 const [record, card] = [captured, item].map((streamed) =>
484 dedupe({ ...streamed, event: projectDomainEventForWire(streamed.event) }),
485 );
486​
487 const filesTotal = { count: 250, additions: 500, deletions: 250 };
488 expect(payloadOf(record!)).toMatchObject({ files: files.slice(0, 200), filesTotal });
489 expect(payloadOf(card!)).toMatchObject({ files: [], filesOmitted: true, filesTotal });
490 });
491​
492 it("ships a reply whole when its message arrives after it", () => {
493 const first = reply(1, "a");
494 expect([first, message(2, "a")].map(makeLiveBodyDeduper())[0]).toBe(first);
495 });
496});