1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodePath from "node:path";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { expect, it } from "@effect/vitest";
6import {
7 ChatAttachmentId,
8 CommandId,
9 EventId,
10 MessageId,
11 ProjectId,
12 ProviderInstanceId,
13 RuntimeRequestId,
14 ThreadId,
15 TurnItemId,
16 type OrchestrationV2Command,
17 type OrchestrationV2ServerCommand,
18 type OrchestrationV2StoredEvent,
19 type UserInputAttachments,
20} from "@t3tools/contracts";
21import * as DateTime from "effect/DateTime";
22import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";
24import { createPendingAttachmentId, resolveAttachmentPath } from "../attachmentStore.ts";
25import * as ServerConfig from "../config.ts";
26import * as ServerSettings from "../serverSettings.ts";
27import {
28 OrchestratorCommandPreviouslyRejectedError,
29 OrchestratorDispatchError,
30} from "./Orchestrator.ts";
31import * as ThreadManagementService from "./ThreadManagementService.ts";
32import { dispatchCommand, sendToThread } from "./ThreadMessageIntake.ts";
33
34const layerIntakeTest = ServerConfig.layerTest(process.cwd(), {
35 prefix: "t3-question-intake-",
36}).pipe(Layer.provideMerge(NodeServices.layer), Layer.provideMerge(ServerSettings.layerTest()));
37
38const layerFailingDispatch = (captured: OrchestrationV2ServerCommand[]) =>
39 Layer.mock(ThreadManagementService.ThreadManagementService)({
40 dispatch: (command) => {
41 captured.push(command);
42 return Effect.fail(
43 new OrchestratorDispatchError({
44 commandId: command.commandId,
45 commandType: command.type,
46 cause: "Stop after intake",
47 }),
48 );
49 },
50 });
51
52const NOW = DateTime.makeUnsafe("2026-09-13T00:00:00.000Z");
53
54const answeredRespondEvents = (input: {
55 readonly threadId: ThreadId;
56 readonly requestId: string;
57 readonly attachmentsByQuestionId: UserInputAttachments;
58}): OrchestrationV2StoredEvent[] => [
59 {
60 sequence: 1,
61 commandId: null,
62 event: {
63 id: EventId.make("evt-recorded-answer"),
64 type: "turn-item.updated",
65 threadId: input.threadId,
66 occurredAt: NOW,
67 payload: {
68 id: TurnItemId.make("item-recorded-answer"),
69 type: "user_input_request",
70 threadId: input.threadId,
71 runId: null,
72 nodeId: null,
73 providerThreadId: null,
74 providerTurnId: null,
75 nativeItemRef: null,
76 parentItemId: null,
77 ordinal: 1,
78 status: "completed",
79 title: null,
80 startedAt: null,
81 completedAt: NOW,
82 updatedAt: NOW,
83 requestId: RuntimeRequestId.make(input.requestId),
84 questions: [],
85 questionAnswer: {
86 requestId: input.requestId,
87 answers: {},
88 attachmentsByQuestionId: input.attachmentsByQuestionId,
89 },
90 },
91 },
92 },
93];
94
95// The first attempt answered with text only: clients omit attachmentsByQuestionId
96// entirely, so the recorded turn item resolves the request without a questionAnswer.
97const textOnlyRespondEvents = (input: {
98 readonly threadId: ThreadId;
99 readonly requestId: string;
100}): OrchestrationV2StoredEvent[] => [
101 {
102 sequence: 1,
103 commandId: null,
104 event: {
105 id: EventId.make("evt-recorded-answer"),
106 type: "turn-item.updated",
107 threadId: input.threadId,
108 occurredAt: NOW,
109 payload: {
110 id: TurnItemId.make("item-recorded-answer"),
111 type: "user_input_request",
112 threadId: input.threadId,
113 runId: null,
114 nodeId: null,
115 providerThreadId: null,
116 providerTurnId: null,
117 nativeItemRef: null,
118 parentItemId: null,
119 ordinal: 1,
120 status: "completed",
121 title: null,
122 startedAt: null,
123 completedAt: NOW,
124 updatedAt: NOW,
125 requestId: RuntimeRequestId.make(input.requestId),
126 questions: [],
127 },
128 },
129 },
130];
131
132it.effect("claims question uploads and passes readable paths through the V2 request command", () =>
133 Effect.gen(function* () {
134 const config = yield* ServerConfig.ServerConfig;
135 const id = ChatAttachmentId.make(createPendingAttachmentId()!);
136 NodeFS.writeFileSync(
137 NodePath.join(config.attachmentsDir, `${id}.png`),
138 new Uint8Array([1, 2, 3]),
139 );
140 const captured: OrchestrationV2ServerCommand[] = [];
141 yield* dispatchCommand({
142 type: "runtime-request.respond",
143 commandId: CommandId.make("answer-1"),
144 threadId: ThreadId.make("thread-answer"),
145 requestId: RuntimeRequestId.make("request-1"),
146 answers: { q: ["Selected option"] },
147 attachmentsByQuestionId: {
148 q: [{ type: "image", id, name: "screen.png", mimeType: "image/png", sizeBytes: 3 }],
149 },
150 }).pipe(Effect.provide(layerFailingDispatch(captured)), Effect.result);
151 expect(captured).toHaveLength(1);
152 const command = captured[0]!;
153 expect(command.type).toBe("runtime-request.respond");
154 if (command.type !== "runtime-request.respond") return;
155 const attachment = command.attachmentsByQuestionId?.q?.[0];
156 expect(attachment).toBeDefined();
157 const path = resolveAttachmentPath({
158 attachmentsDir: config.attachmentsDir,
159 attachment: attachment!,
160 });
161 expect(path).not.toBeNull();
162 expect(NodeFS.readFileSync(path!)).toEqual(Buffer.from([1, 2, 3]));
163 expect(command.answers?.q).toEqual([
164 "Selected option",
165 `Attached image "screen.png": "${path}"`,
166 ]);
167 }).pipe(Effect.provide(layerIntakeTest)),
168);
169
170it.effect("rolls back earlier question claims when a later pending upload is missing", () =>
171 Effect.gen(function* () {
172 const config = yield* ServerConfig.ServerConfig;
173 const threadId = ThreadId.make("thread-q-fail");
174 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
175 const existingId = ChatAttachmentId.make("thread-q-fail-00000000-0000-4000-8000-000000000001");
176 NodeFS.writeFileSync(
177 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
178 new Uint8Array([1, 2, 3]),
179 );
180 NodeFS.writeFileSync(
181 NodePath.join(config.attachmentsDir, `${existingId}.png`),
182 new Uint8Array([9, 9, 9, 9]),
183 );
184 const captured: OrchestrationV2ServerCommand[] = [];
185 const result = yield* dispatchCommand({
186 type: "runtime-request.respond",
187 commandId: CommandId.make("answer-missing-pending"),
188 threadId,
189 requestId: RuntimeRequestId.make("request-missing-pending"),
190 answers: { q1: ["one"], q2: ["two"] },
191 attachmentsByQuestionId: {
192 q1: [
193 {
194 type: "image",
195 id: existingId,
196 name: "keep.png",
197 mimeType: "image/png",
198 sizeBytes: 4,
199 },
200 {
201 type: "image",
202 id: pendingId,
203 name: "screen.png",
204 mimeType: "image/png",
205 sizeBytes: 3,
206 },
207 ],
208 q2: [
209 {
210 type: "image",
211 id: ChatAttachmentId.make(createPendingAttachmentId()!),
212 name: "gone.png",
213 mimeType: "image/png",
214 sizeBytes: 3,
215 },
216 ],
217 },
218 }).pipe(Effect.provide(layerFailingDispatch(captured)), Effect.result);
219 expect(result._tag).toBe("Failure");
220 if (result._tag === "Failure") {
221 expect(result.failure._tag).toBe("AttachmentClaimError");
222 }
223 expect(captured).toHaveLength(0);
224 // The pending sources stay behind so the user can retry the response.
225 expect(NodeFS.readFileSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toEqual(
226 Buffer.from([1, 2, 3]),
227 );
228 // The pre-existing thread attachment is retained; only the new copy is rolled back.
229 expect(
230 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
231 entry.startsWith("thread-q-fail-"),
232 ),
233 ).toEqual([`${existingId}.png`]);
234 }).pipe(Effect.provide(layerIntakeTest)),
235);
236
237it.effect("rolls back earlier question claims when attachment path preparation fails", () =>
238 Effect.gen(function* () {
239 const config = yield* ServerConfig.ServerConfig;
240 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
241 NodeFS.writeFileSync(
242 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
243 new Uint8Array([1, 2, 3]),
244 );
245 const captured: OrchestrationV2ServerCommand[] = [];
246 const result = yield* dispatchCommand({
247 type: "runtime-request.respond",
248 commandId: CommandId.make("answer-missing-claimed"),
249 threadId: ThreadId.make("thread-path-fail"),
250 requestId: RuntimeRequestId.make("request-missing-claimed"),
251 answers: { q1: ["one"], q2: ["two"] },
252 attachmentsByQuestionId: {
253 q1: [
254 {
255 type: "image",
256 id: pendingId,
257 name: "screen.png",
258 mimeType: "image/png",
259 sizeBytes: 3,
260 },
261 ],
262 q2: [
263 {
264 type: "image",
265 id: ChatAttachmentId.make("thread-path-fail-00000000-0000-4000-8000-000000000099"),
266 name: "missing.png",
267 mimeType: "image/png",
268 sizeBytes: 3,
269 },
270 ],
271 },
272 }).pipe(Effect.provide(layerFailingDispatch(captured)), Effect.result);
273 expect(result._tag).toBe("Failure");
274 if (result._tag === "Failure") {
275 expect(result.failure._tag).toBe("AttachmentClaimError");
276 }
277 expect(captured).toHaveLength(0);
278 expect(NodeFS.readFileSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toEqual(
279 Buffer.from([1, 2, 3]),
280 );
281 expect(
282 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
283 entry.startsWith("thread-path-fail-"),
284 ),
285 ).toEqual([]);
286 }).pipe(Effect.provide(layerIntakeTest)),
287);
288
289it.effect("retains claimed copies when dispatch failure may have been accepted", () =>
290 Effect.gen(function* () {
291 const config = yield* ServerConfig.ServerConfig;
292 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
293 NodeFS.writeFileSync(
294 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
295 new Uint8Array([1, 2, 3]),
296 );
297 const captured: OrchestrationV2ServerCommand[] = [];
298 const result = yield* dispatchCommand({
299 type: "runtime-request.respond",
300 commandId: CommandId.make("answer-ambiguous"),
301 threadId: ThreadId.make("thread-ambiguous"),
302 requestId: RuntimeRequestId.make("request-ambiguous"),
303 answers: { q: ["one"] },
304 attachmentsByQuestionId: {
305 q: [
306 {
307 type: "image",
308 id: pendingId,
309 name: "screen.png",
310 mimeType: "image/png",
311 sizeBytes: 3,
312 },
313 ],
314 },
315 }).pipe(Effect.provide(layerFailingDispatch(captured)), Effect.result);
316 expect(result._tag).toBe("Failure");
317 expect(captured).toHaveLength(1);
318 // The response may have been accepted before the error, so the copy that
319 // backs the answer's path stays for the provider to read.
320 const threadFiles = NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
321 entry.startsWith("thread-ambiguous-"),
322 );
323 expect(threadFiles).toHaveLength(1);
324 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
325 }).pipe(Effect.provide(layerIntakeTest)),
326);
327
328it.effect("releases claimed copies when the command was already rejected", () =>
329 Effect.gen(function* () {
330 const config = yield* ServerConfig.ServerConfig;
331 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
332 NodeFS.writeFileSync(
333 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
334 new Uint8Array([1, 2, 3]),
335 );
336 // A retry of a previously rejected command claims fresh copies, then the
337 // orchestrator replays the rejection without touching them.
338 const result = yield* dispatchCommand({
339 type: "runtime-request.respond",
340 commandId: CommandId.make("answer-rejected"),
341 threadId: ThreadId.make("thread-rejected"),
342 requestId: RuntimeRequestId.make("request-rejected"),
343 answers: { q: ["one"] },
344 attachmentsByQuestionId: {
345 q: [
346 {
347 type: "image",
348 id: pendingId,
349 name: "screen.png",
350 mimeType: "image/png",
351 sizeBytes: 3,
352 },
353 ],
354 },
355 }).pipe(
356 Effect.provide(
357 Layer.mock(ThreadManagementService.ThreadManagementService)({
358 dispatch: (command) =>
359 Effect.fail(
360 new OrchestratorCommandPreviouslyRejectedError({
361 commandId: command.commandId,
362 commandType: command.type,
363 detail: "rejected earlier",
364 }),
365 ),
366 }),
367 ),
368 Effect.result,
369 );
370 expect(result._tag).toBe("Failure");
371 expect(
372 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
373 entry.startsWith("thread-rejected-"),
374 ),
375 ).toEqual([]);
376 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
377 }).pipe(Effect.provide(layerIntakeTest)),
378);
379
380it.effect("releases claimed copies when dispatch replays an earlier accepted response", () =>
381 Effect.gen(function* () {
382 const config = yield* ServerConfig.ServerConfig;
383 const threadId = ThreadId.make("thread-replay");
384 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
385 NodeFS.writeFileSync(
386 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
387 new Uint8Array([1, 2, 3]),
388 );
389 // The first attempt was accepted but its response was lost. The retry's
390 // dispatch replays the stored events, which reference the first claim's
391 // attachment id — not this attempt's fresh copy.
392 const earlierAttachment = {
393 type: "image" as const,
394 id: ChatAttachmentId.make("thread-replay-00000000-0000-4000-8000-0000000000aa"),
395 name: "screen.png",
396 mimeType: "image/png",
397 sizeBytes: 3,
398 };
399 NodeFS.writeFileSync(
400 NodePath.join(config.attachmentsDir, `${earlierAttachment.id}.png`),
401 new Uint8Array([4, 4, 4]),
402 );
403 const storedEvents = answeredRespondEvents({
404 threadId,
405 requestId: "request-replay",
406 attachmentsByQuestionId: { q: [earlierAttachment] },
407 });
408 const result = yield* dispatchCommand({
409 type: "runtime-request.respond",
410 commandId: CommandId.make("answer-replay"),
411 threadId,
412 requestId: RuntimeRequestId.make("request-replay"),
413 answers: { q: ["one"] },
414 attachmentsByQuestionId: {
415 q: [
416 {
417 type: "image",
418 id: pendingId,
419 name: "screen.png",
420 mimeType: "image/png",
421 sizeBytes: 3,
422 },
423 ],
424 },
425 }).pipe(
426 Effect.provide(
427 Layer.mock(ThreadManagementService.ThreadManagementService)({
428 dispatch: () => Effect.succeed({ sequence: 1, storedEvents }),
429 }),
430 ),
431 Effect.result,
432 );
433 expect(result._tag).toBe("Success");
434 // The accepted copy from the first attempt is untouched; only the
435 // retry's unreferenced copy is removed.
436 expect(
437 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
438 entry.startsWith("thread-replay-"),
439 ),
440 ).toEqual([`${earlierAttachment.id}.png`]);
441 expect(
442 NodeFS.readFileSync(NodePath.join(config.attachmentsDir, `${earlierAttachment.id}.png`)),
443 ).toEqual(Buffer.from([4, 4, 4]));
444 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
445 }).pipe(Effect.provide(layerIntakeTest)),
446);
447
448it.effect("releases claimed copies when the recorded answer has no attachments", () =>
449 Effect.gen(function* () {
450 const config = yield* ServerConfig.ServerConfig;
451 const threadId = ThreadId.make("thread-empty-answer");
452 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
453 NodeFS.writeFileSync(
454 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
455 new Uint8Array([1, 2, 3]),
456 );
457 // The same command id was first answered without attachments; this
458 // retry's copies are unreferenced by the recorded answer.
459 const storedEvents = answeredRespondEvents({
460 threadId,
461 requestId: "request-empty-answer",
462 attachmentsByQuestionId: {},
463 });
464 const result = yield* dispatchCommand({
465 type: "runtime-request.respond",
466 commandId: CommandId.make("answer-empty"),
467 threadId,
468 requestId: RuntimeRequestId.make("request-empty-answer"),
469 answers: { q: ["one"] },
470 attachmentsByQuestionId: {
471 q: [
472 {
473 type: "image",
474 id: pendingId,
475 name: "screen.png",
476 mimeType: "image/png",
477 sizeBytes: 3,
478 },
479 ],
480 },
481 }).pipe(
482 Effect.provide(
483 Layer.mock(ThreadManagementService.ThreadManagementService)({
484 dispatch: () => Effect.succeed({ sequence: 1, storedEvents }),
485 }),
486 ),
487 Effect.result,
488 );
489 expect(result._tag).toBe("Success");
490 expect(
491 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
492 entry.startsWith("thread-empty-answer-"),
493 ),
494 ).toEqual([]);
495 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
496 }).pipe(Effect.provide(layerIntakeTest)),
497);
498
499it.effect("releases claimed copies when the recorded answer was text-only", () =>
500 Effect.gen(function* () {
501 const config = yield* ServerConfig.ServerConfig;
502 const threadId = ThreadId.make("thread-text-only-answer");
503 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
504 NodeFS.writeFileSync(
505 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
506 new Uint8Array([1, 2, 3]),
507 );
508 // The first attempt accepted a text-only response, so its recorded turn
509 // item has no questionAnswer at all. The retried command id uploads an
510 // attachment whose fresh copy is unreferenced by that answer.
511 const storedEvents = textOnlyRespondEvents({
512 threadId,
513 requestId: "request-text-only",
514 });
515 const result = yield* dispatchCommand({
516 type: "runtime-request.respond",
517 commandId: CommandId.make("answer-text-only"),
518 threadId,
519 requestId: RuntimeRequestId.make("request-text-only"),
520 answers: { q: ["one"] },
521 attachmentsByQuestionId: {
522 q: [
523 {
524 type: "image",
525 id: pendingId,
526 name: "screen.png",
527 mimeType: "image/png",
528 sizeBytes: 3,
529 },
530 ],
531 },
532 }).pipe(
533 Effect.provide(
534 Layer.mock(ThreadManagementService.ThreadManagementService)({
535 dispatch: () => Effect.succeed({ sequence: 1, storedEvents }),
536 }),
537 ),
538 Effect.result,
539 );
540 expect(result._tag).toBe("Success");
541 expect(
542 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
543 entry.startsWith("thread-text-only-answer-"),
544 ),
545 ).toEqual([]);
546 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
547 }).pipe(Effect.provide(layerIntakeTest)),
548);
549
550it.effect("retains claimed copies referenced by a fresh recorded answer", () =>
551 Effect.gen(function* () {
552 const config = yield* ServerConfig.ServerConfig;
553 const threadId = ThreadId.make("thread-fresh-answer");
554 const firstId = ChatAttachmentId.make(createPendingAttachmentId()!);
555 const secondId = ChatAttachmentId.make(createPendingAttachmentId()!);
556 NodeFS.writeFileSync(
557 NodePath.join(config.attachmentsDir, `${firstId}.png`),
558 new Uint8Array([1, 2, 3]),
559 );
560 NodeFS.writeFileSync(
561 NodePath.join(config.attachmentsDir, `${secondId}.png`),
562 new Uint8Array([7, 7]),
563 );
564 const result = yield* dispatchCommand({
565 type: "runtime-request.respond",
566 commandId: CommandId.make("answer-fresh"),
567 threadId,
568 requestId: RuntimeRequestId.make("request-fresh"),
569 answers: { q1: ["one"], q2: ["two"] },
570 attachmentsByQuestionId: {
571 q1: [
572 {
573 type: "image",
574 id: firstId,
575 name: "one.png",
576 mimeType: "image/png",
577 sizeBytes: 3,
578 },
579 ],
580 q2: [
581 {
582 type: "image",
583 id: secondId,
584 name: "two.png",
585 mimeType: "image/png",
586 sizeBytes: 2,
587 },
588 ],
589 },
590 }).pipe(
591 Effect.provide(
592 Layer.mock(ThreadManagementService.ThreadManagementService)({
593 // A fresh commit records exactly the attachments this attempt claimed.
594 dispatch: (command) =>
595 Effect.succeed({
596 sequence: 1,
597 storedEvents: answeredRespondEvents({
598 threadId,
599 requestId: "request-fresh",
600 attachmentsByQuestionId:
601 command.type === "runtime-request.respond"
602 ? (command.attachmentsByQuestionId ?? {})
603 : {},
604 }),
605 }),
606 }),
607 ),
608 Effect.result,
609 );
610 expect(result._tag).toBe("Success");
611 expect(
612 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
613 entry.startsWith("thread-fresh-answer-"),
614 ),
615 ).toHaveLength(2);
616 }).pipe(Effect.provide(layerIntakeTest)),
617);
618
619it.effect("releases claimed copies when preparation dies with a defect", () =>
620 Effect.gen(function* () {
621 const config = yield* ServerConfig.ServerConfig;
622 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
623 NodeFS.writeFileSync(
624 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
625 new Uint8Array([1, 2, 3]),
626 );
627 const poisoned = {
628 get id(): string {
629 throw new Error("poisoned");
630 },
631 };
632 const captured: OrchestrationV2ServerCommand[] = [];
633 const result = yield* dispatchCommand({
634 type: "runtime-request.respond",
635 commandId: CommandId.make("answer-defect"),
636 threadId: ThreadId.make("thread-defect"),
637 requestId: RuntimeRequestId.make("request-defect"),
638 answers: { q1: ["one"], q2: ["two"] },
639 attachmentsByQuestionId: {
640 q1: [
641 {
642 type: "image",
643 id: pendingId,
644 name: "screen.png",
645 mimeType: "image/png",
646 sizeBytes: 3,
647 },
648 ],
649 q2: [poisoned] as never,
650 },
651 }).pipe(Effect.provide(layerFailingDispatch(captured)), Effect.exit);
652 expect(result._tag).toBe("Failure");
653 expect(captured).toHaveLength(0);
654 expect(
655 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
656 entry.startsWith("thread-defect-"),
657 ),
658 ).toEqual([]);
659 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
660 }).pipe(Effect.provide(layerIntakeTest)),
661);
662
663it.effect("a retried response re-claims the preserved pending uploads", () =>
664 Effect.gen(function* () {
665 const config = yield* ServerConfig.ServerConfig;
666 const threadId = ThreadId.make("thread-retry");
667 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
668 const lateId = ChatAttachmentId.make(createPendingAttachmentId()!);
669 NodeFS.writeFileSync(
670 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
671 new Uint8Array([1, 2, 3]),
672 );
673 const command: OrchestrationV2Command = {
674 type: "runtime-request.respond",
675 commandId: CommandId.make("answer-retry"),
676 threadId,
677 requestId: RuntimeRequestId.make("request-retry"),
678 answers: { q1: ["one"], q2: ["two"] },
679 attachmentsByQuestionId: {
680 q1: [
681 {
682 type: "image",
683 id: pendingId,
684 name: "screen.png",
685 mimeType: "image/png",
686 sizeBytes: 3,
687 },
688 ],
689 q2: [
690 {
691 type: "image",
692 id: lateId,
693 name: "late.png",
694 mimeType: "image/png",
695 sizeBytes: 2,
696 },
697 ],
698 },
699 };
700 const captured: OrchestrationV2ServerCommand[] = [];
701 const first = yield* dispatchCommand(command).pipe(
702 Effect.provide(layerFailingDispatch(captured)),
703 Effect.result,
704 );
705 expect(first._tag).toBe("Failure");
706 expect(captured).toHaveLength(0);
707
708 NodeFS.writeFileSync(
709 NodePath.join(config.attachmentsDir, `${lateId}.png`),
710 new Uint8Array([7, 7]),
711 );
712 const retry = yield* dispatchCommand(command).pipe(
713 Effect.provide(
714 Layer.mock(ThreadManagementService.ThreadManagementService)({
715 dispatch: (dispatched) => {
716 captured.push(dispatched);
717 return Effect.succeed({ sequence: 1, storedEvents: [] });
718 },
719 }),
720 ),
721 Effect.result,
722 );
723 expect(retry._tag).toBe("Success");
724 const dispatched = captured[0]!;
725 expect(dispatched.type).toBe("runtime-request.respond");
726 if (dispatched.type !== "runtime-request.respond") return;
727 for (const attachment of [
728 ...(dispatched.attachmentsByQuestionId?.q1 ?? []),
729 ...(dispatched.attachmentsByQuestionId?.q2 ?? []),
730 ]) {
731 const path = resolveAttachmentPath({
732 attachmentsDir: config.attachmentsDir,
733 attachment,
734 });
735 expect(path).not.toBeNull();
736 expect(NodeFS.existsSync(path!)).toBe(true);
737 }
738 expect(
739 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
740 entry.startsWith("thread-retry-"),
741 ),
742 ).toHaveLength(2);
743 }).pipe(Effect.provide(layerIntakeTest)),
744);
745
746it.effect("applies the image budget across all questions before dispatch", () =>
747 Effect.gen(function* () {
748 const attachments = Array.from({ length: 5 }, (_, index) => ({
749 type: "image" as const,
750 id: ChatAttachmentId.make(`existing-budget-${index}`),
751 name: `${index}.png`,
752 mimeType: "image/png",
753 sizeBytes: 10 * 1024 * 1024,
754 }));
755 const captured: OrchestrationV2ServerCommand[] = [];
756 const result = yield* Effect.exit(
757 dispatchCommand({
758 type: "runtime-request.respond",
759 commandId: CommandId.make("command-image-budget"),
760 threadId: ThreadId.make("thread-image-budget"),
761 requestId: RuntimeRequestId.make("question-image-budget"),
762 answers: {},
763 attachmentsByQuestionId: { q1: attachments, q2: attachments },
764 }).pipe(Effect.provide(layerFailingDispatch(captured))),
765 );
766 expect(result._tag).toBe("Failure");
767 if (result._tag === "Failure") expect(String(result.cause)).toContain("80 MiB");
768 expect(captured).toEqual([]);
769 }).pipe(Effect.provide(layerIntakeTest)),
770);
771
772it.effect("releases claimed copies when a project disables the message's provider instance", () =>
773 Effect.gen(function* () {
774 const config = yield* ServerConfig.ServerConfig;
775 const pendingId = ChatAttachmentId.make(createPendingAttachmentId()!);
776 NodeFS.writeFileSync(
777 NodePath.join(config.attachmentsDir, `${pendingId}.png`),
778 new Uint8Array([1, 2, 3]),
779 );
780 const result = yield* sendToThread({
781 projectId: ProjectId.make("project-disabled-send"),
782 commandId: CommandId.make("command-disabled-send"),
783 threadId: ThreadId.make("thread-disabled-send"),
784 messageId: MessageId.make("message-disabled-send"),
785 text: "Hello",
786 attachments: [
787 {
788 type: "image",
789 id: pendingId,
790 name: "screen.png",
791 mimeType: "image/png",
792 sizeBytes: 3,
793 },
794 ],
795 mode: "auto",
796 createdBy: "user",
797 creationSource: "web",
798 }).pipe(
799 Effect.provide(
800 Layer.mock(ThreadManagementService.ThreadManagementService)({
801 sendToThread: (input) =>
802 Effect.fail(
803 new ThreadManagementService.ThreadManagementProviderInstanceDisabledError({
804 projectId: input.projectId,
805 instanceId: ProviderInstanceId.make("codex"),
806 }),
807 ),
808 }),
809 ),
810 Effect.result,
811 );
812 expect(result._tag).toBe("Failure");
813 expect(
814 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
815 entry.startsWith("thread-disabled-send-"),
816 ),
817 ).toEqual([]);
818 expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`))).toBe(true);
819 }).pipe(Effect.provide(layerIntakeTest)),
820);