1import {
2 acknowledgedThreadMessagesAtom,
3 pruneCaughtUpAcknowledgedThreadMessages,
4} from "./acknowledged-thread-messages";
5import {
6 CommandId,
7 EnvironmentId,
8 MessageId,
9 ProjectId,
10 ProviderInstanceId,
11 ThreadId,
12} from "@t3tools/contracts";
13import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test";
14
15import type { PreparedTurnAttachments } from "../lib/attachmentUpload";
16
17const harness = vi.hoisted(() => ({
18 canOperate: true,
19 threadState: null as unknown as import("effect/reactivity").Atom.Writable<{
20 readonly data: import("effect/Option").Option<{
21 readonly messages: ReadonlyArray<{ readonly id: string }>;
22 }>;
23 readonly sequence: number;
24 }>,
25 manager: null as unknown as ReturnType<
26 typeof import("./thread-outbox-manager").createThreadOutboxManager
27 >,
28 removePersistedFile: vi.fn(async () => undefined),
29 removeOutboxMessage: vi.fn(async (_message: QueuedThreadMessage) => undefined),
30 prepareTurnAttachments: vi.fn<typeof import("../lib/attachmentUpload").prepareTurnAttachments>(),
31 draftFile: (() => {
32 let document = "";
33 let writeError: Error | null = null;
34 return {
35 setDocument(value: unknown) {
36 document = JSON.stringify(value);
37 },
38 setWriteError(error: Error | null) {
39 writeError = error;
40 },
41 Directory: class {
42 create() {}
43 },
44 File: class {
45 exists = true;
46 parentDirectory = null;
47
48 create() {}
49
50 moveSync() {}
51
52 async text() {
53 return document;
54 }
55
56 write(value: string) {
57 if (writeError) {
58 throw writeError;
59 }
60 document = value;
61 }
62 },
63 };
64 })(),
65}));
66
67vi.mock("react-native", () => ({ Alert: { alert: vi.fn() } }));
68
69vi.mock("expo-file-system", () => ({
70 Directory: harness.draftFile.Directory,
71 File: harness.draftFile.File,
72 Paths: { document: "/documents" },
73}));
74
75vi.mock("../lib/composerImages", () => ({
76 removePersistedComposerAttachmentFile: harness.removePersistedFile,
77}));
78
79vi.mock("../lib/uuid", () => ({
80 uuidv4: () => "00000000-0000-4000-8000-000000000000",
81 randomHex: () => "abcd",
82}));
83
84vi.mock("../lib/attachmentUpload", () => ({
85 prepareTurnAttachments: harness.prepareTurnAttachments,
86}));
87
88vi.mock("./projects", () => ({ environmentProjects: {} }));
89
90vi.mock("./workspace", () => ({ workspaceConnections: {} }));
91
92vi.mock("./server", async () => {
93 const { Atom } = await import("effect/reactivity");
94 return { serverEnvironment: { configValueAtom: Atom.family(() => Atom.make(null)) } };
95});
96
97vi.mock("./git", () => ({ gitEnvironment: {} }));
98
99vi.mock("./threads", async () => {
100 const { Atom } = await import("effect/reactivity");
101 const Option = await import("effect/Option");
102 harness.threadState = Atom.make({
103 data: Option.none<{ readonly messages: ReadonlyArray<{ readonly id: string }> }>(),
104 sequence: 0,
105 }).pipe(Atom.keepAlive);
106 return {
107 threadEnvironment: {},
108 environmentThreadDetails: { stateAtom: () => harness.threadState },
109 };
110});
111
112vi.mock("./session", () => ({
113 environmentSession: {},
114 readEnvironmentScope: () => harness.canOperate,
115 sessionHasScope: () => harness.canOperate,
116}));
117
118vi.mock("./use-thread-outbox", async () => {
119 const { Atom } = await import("effect/reactivity");
120 return {
121 editingQueuedMessageIdsAtom: Atom.make<Record<string, boolean>>({}).pipe(Atom.keepAlive),
122 dispatchingQueuedMessageIdAtom: Atom.make(null).pipe(Atom.keepAlive),
123 retryingQueuedMessageIdsAtom: Atom.make({}).pipe(Atom.keepAlive),
124 threadOutboxShellStatusesAtom: Atom.make(new Map()),
125 };
126});
127
128vi.mock("./thread-outbox", async () => {
129 const { createThreadOutboxManager } = await import("./thread-outbox-manager");
130 const { appAtomRegistry } = await import("./atom-registry");
131 harness.manager = createThreadOutboxManager({
132 registry: appAtomRegistry,
133 storage: {
134 load: async () => ({ messages: [], errors: [] }),
135 write: async () => undefined,
136 remove: (message) => harness.removeOutboxMessage(message),
137 },
138 });
139 const manager = harness.manager;
140 return {
141 threadOutboxManager: manager,
142 flushThreadOutbox: async () => undefined,
143 confirmThreadOutboxMessageQueued: (message: never) => manager.confirmQueued(message),
144 updateThreadOutboxMessage: (message: never, expectedRevision?: number) =>
145 manager.update(message, expectedRevision),
146 threadOutboxRevision: (messageId: never) => manager.revisionOf(messageId),
147 };
148});
149
150import * as Option from "effect/Option";
151import { scopedThreadKey } from "../lib/scopedEntities";
152import { appAtomRegistry } from "./atom-registry";
153import { pendingThreadCreationOutcomesAtom } from "./pending-thread-creation";
154import { clearThreadComposerError, threadComposerErrorsAtom } from "./thread-composer-error";
155import type { QueuedThreadMessage } from "./thread-outbox-model";
156import * as composerDrafts from "./use-composer-drafts";
157import { recoverFailedThreadDraft } from "./recover-failed-thread-draft";
158import { editingQueuedMessageIdsAtom } from "./use-thread-outbox";
159import {
160 completeQueuedMessageDelivery,
161 prepareQueuedMessageAttachments,
162 recoverEditedCreationAfterDelivery,
163 removeAcknowledgedExistingThreadMessage,
164 restoreRejectedQueuedMessage,
165} from "./threadOutboxDelivery";
166
167function queuedMessage(input: {
168 readonly messageId: string;
169 readonly text: string;
170 readonly fileUri?: string;
171}): QueuedThreadMessage {
172 return {
173 environmentId: EnvironmentId.make("environment-1"),
174 threadId: ThreadId.make("thread-1"),
175 messageId: MessageId.make(input.messageId),
176 commandId: CommandId.make(`command-${input.messageId}`),
177 text: input.text,
178 attachments: input.fileUri
179 ? [
180 {
181 id: `file-${input.messageId}`,
182 type: "file",
183 name: "report.pdf",
184 mimeType: "application/pdf",
185 sizeBytes: 42,
186 fileUri: input.fileUri,
187 },
188 ]
189 : [],
190 createdAt: "2026-08-24T12:00:00.000Z",
191 };
192}
193
194function withReusedFileUpload(
195 message: QueuedThreadMessage,
196 attachmentId: string,
197): QueuedThreadMessage {
198 return {
199 ...message,
200 attachments: message.attachments.map((attachment) =>
201 attachment.type === "file"
202 ? {
203 ...attachment,
204 uploadedAttachmentId: attachmentId,
205 uploadEnvironmentId: message.environmentId,
206 }
207 : attachment,
208 ),
209 };
210}
211
212function remainingMessages(): ReadonlyArray<QueuedThreadMessage> {
213 return Object.values(appAtomRegistry.get(harness.manager.queuedMessagesByThreadKeyAtom)).flat();
214}
215
216beforeEach(() => {
217 appAtomRegistry.set(acknowledgedThreadMessagesAtom, []);
218 appAtomRegistry.set(harness.threadState, { data: Option.none(), sequence: 0 });
219 harness.canOperate = true;
220 harness.draftFile.setDocument({ schemaVersion: 1, drafts: {} });
221});
222
223afterEach(() => {
224 appAtomRegistry.set(harness.manager.queuedMessagesByThreadKeyAtom, {});
225 appAtomRegistry.set(composerDrafts.composerDraftsAtom, {});
226 appAtomRegistry.set(composerDrafts.composerCloudDraftsAtom, { accountId: null, signedOut: {} });
227 appAtomRegistry.set(editingQueuedMessageIdsAtom, {});
228 appAtomRegistry.set(pendingThreadCreationOutcomesAtom, {});
229 appAtomRegistry.set(threadComposerErrorsAtom, {});
230 harness.draftFile.setWriteError(null);
231 harness.removePersistedFile.mockClear();
232 harness.removeOutboxMessage.mockClear();
233 harness.prepareTurnAttachments.mockReset();
234});
235
236describe("thread outbox attachment preparation", () => {
237 it("keeps a denied message queued without starting its attachment upload", async () => {
238 const message = queuedMessage({
239 messageId: "denied-queued-message",
240 text: "Keep this queued",
241 fileUri: "file:///documents/keep.pdf",
242 });
243 await harness.manager.enqueue(message);
244 harness.prepareTurnAttachments.mockResolvedValueOnce({
245 status: "ready",
246 attachments: [],
247 draftAttachments: message.attachments,
248 pendingAttachmentIds: [],
249 });
250 harness.canOperate = false;
251 await expect(prepareQueuedMessageAttachments(message)).resolves.toEqual({
252 status: "abandoned",
253 });
254 expect(harness.prepareTurnAttachments).not.toHaveBeenCalled();
255 expect(remainingMessages()).toEqual([message]);
256 });
257
258 it("abandons reused uploads when an editor saves changed text during verification", async () => {
259 const message = withReusedFileUpload(
260 queuedMessage({
261 messageId: "message-reused-upload-race",
262 text: "original text",
263 fileUri: "file:///documents/t3-composer-attachments/reused.pdf",
264 }),
265 "pending-reused-upload",
266 );
267 const preparationStarted = Promise.withResolvers<void>();
268 const preparationBarrier = Promise.withResolvers<PreparedTurnAttachments>();
269 harness.prepareTurnAttachments.mockImplementationOnce(async () => {
270 preparationStarted.resolve();
271 return preparationBarrier.promise;
272 });
273 await harness.manager.enqueue(message);
274 appAtomRegistry.set(editingQueuedMessageIdsAtom, { [message.messageId]: true });
275
276 const preparation = prepareQueuedMessageAttachments(message);
277 await preparationStarted.promise;
278 const edited = { ...message, text: "saved editor text" };
279 await harness.manager.update(edited);
280 appAtomRegistry.set(editingQueuedMessageIdsAtom, {});
281 preparationBarrier.resolve({
282 status: "ready",
283 attachments: [],
284 draftAttachments: message.attachments,
285 pendingAttachmentIds: ["pending-reused-upload"],
286 });
287
288 await expect(preparation).resolves.toEqual({ status: "abandoned" });
289 expect(remainingMessages()).toEqual([edited]);
290 });
291
292 it("uses the known next revision after persisting uploaded references", async () => {
293 const message = queuedMessage({
294 messageId: "message-new-upload-revision",
295 text: "upload this file",
296 fileUri: "file:///documents/t3-composer-attachments/new.pdf",
297 });
298 const uploadedAttachments = message.attachments.map((attachment) =>
299 attachment.type === "file"
300 ? {
301 ...attachment,
302 uploadedAttachmentId: "pending-new-upload",
303 uploadEnvironmentId: message.environmentId,
304 }
305 : attachment,
306 );
307 harness.prepareTurnAttachments.mockImplementationOnce(async (input) => {
308 expect(await input.persistUploadedReferences?.(uploadedAttachments)).toBe("persisted");
309 return {
310 status: "ready",
311 attachments: [],
312 draftAttachments: uploadedAttachments,
313 pendingAttachmentIds: ["pending-new-upload"],
314 };
315 });
316 await harness.manager.enqueue(message);
317 const revision = harness.manager.revisionOf(message.messageId);
318
319 const result = await prepareQueuedMessageAttachments(message);
320
321 expect(result).toMatchObject({
322 status: "ready",
323 persistedMessage: { attachments: uploadedAttachments },
324 deliveryRevision: revision + 1,
325 });
326 expect(harness.manager.revisionOf(message.messageId)).toBe(revision + 1);
327 });
328
329 it("does not prepare a payload that was already replaced", async () => {
330 const message = queuedMessage({ messageId: "message-stale-before-upload", text: "old" });
331 await harness.manager.enqueue(message);
332 const edited = { ...message, text: "new" };
333 await harness.manager.update(edited);
334
335 await expect(prepareQueuedMessageAttachments(message)).resolves.toEqual({
336 status: "abandoned",
337 });
338 expect(harness.prepareTurnAttachments).not.toHaveBeenCalled();
339 expect(remainingMessages()).toEqual([edited]);
340 });
341});
342
343describe("thread outbox drain delivery cleanup", () => {
344 it("drops caught-up acknowledged rows at the next delivery without subscribing to the thread", async () => {
345 const first = queuedMessage({ messageId: "awaiting-echo", text: "Delivered" });
346 const second = queuedMessage({ messageId: "next-delivery", text: "Also delivered" });
347 await harness.manager.enqueue(first);
348 await harness.manager.enqueue(second);
349
350 await expect(
351 completeQueuedMessageDelivery(first, harness.manager.revisionOf(first.messageId), 7),
352 ).resolves.toBe("removed");
353 expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([
354 { ...first, sequence: 7 },
355 ]);
356
357 // The thread applied the message, but its bounded window does not hold it.
358 appAtomRegistry.set(harness.threadState, {
359 data: Option.some({ messages: [{ id: "unrelated" }] }),
360 sequence: 7,
361 });
362 expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toHaveLength(1);
363
364 await expect(
365 completeQueuedMessageDelivery(second, harness.manager.revisionOf(second.messageId), 9),
366 ).resolves.toBe("removed");
367 expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([
368 { ...second, sequence: 9 },
369 ]);
370 });
371
372 it("drops an acknowledged row once the open thread catches up, so a reload cannot resurrect it", async () => {
373 const message = queuedMessage({ messageId: "paged-out-later", text: "Delivered" });
374 await harness.manager.enqueue(message);
375 await completeQueuedMessageDelivery(message, harness.manager.revisionOf(message.messageId), 7);
376
377 pruneCaughtUpAcknowledgedThreadMessages(
378 scopedThreadKey(message.environmentId, message.threadId),
379 {
380 sequence: 6,
381 messages: [],
382 },
383 );
384 expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([
385 { ...message, sequence: 7 },
386 ]);
387
388 pruneCaughtUpAcknowledgedThreadMessages(
389 scopedThreadKey(message.environmentId, message.threadId),
390 {
391 sequence: 7,
392 messages: [],
393 },
394 );
395 expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([]);
396 });
397
398 it("removes an acknowledged outbox item even when the sign-out archive write fails", async () => {
399 const message = queuedMessage({ messageId: "archive-write-failure", text: "Delivered" });
400 await harness.manager.enqueue(message);
401 await composerDrafts.archiveCloudComposerDrafts("account-a", new Set([message.environmentId]), [
402 message.environmentId,
403 ]);
404 harness.draftFile.setWriteError(new Error("Draft storage unavailable"));
405
406 await expect(
407 completeQueuedMessageDelivery(
408 message,
409 harness.manager.revisionOf(message.messageId),
410 undefined,
411 ),
412 ).resolves.toBe("removed");
413 expect(remainingMessages()).toEqual([]);
414
415 harness.draftFile.setWriteError(null);
416 await composerDrafts.flushComposerDrafts();
417 appAtomRegistry.set(composerDrafts.composerCloudDraftsAtom, { accountId: null, signedOut: {} });
418 composerDrafts.resetComposerDraftsLoadState();
419 await composerDrafts.restoreCloudComposerDrafts("account-a");
420 expect(remainingMessages()).toEqual([]);
421 });
422
423 it.each([false, true])(
424 "does not restore a message delivered after the sign-out snapshot (outbox already cleared: %s)",
425 async (cleared) => {
426 const message = queuedMessage({
427 messageId: "delivered-during-sign-out",
428 text: "Already delivered",
429 });
430 await harness.manager.enqueue(message);
431 const deliveryRevision = harness.manager.revisionOf(message.messageId);
432 await composerDrafts.archiveCloudComposerDrafts(
433 "account-a",
434 new Set([message.environmentId]),
435 [message.environmentId],
436 );
437 expect(
438 appAtomRegistry.get(composerDrafts.composerCloudDraftsAtom).signedOut["account-a"]
439 ?.queuedMessages,
440 ).toEqual([message]);
441
442 if (cleared) await harness.manager.clearEnvironment(message.environmentId);
443 await expect(
444 completeQueuedMessageDelivery(message, deliveryRevision, undefined),
445 ).resolves.toBe(cleared ? "edited" : "removed");
446
447 appAtomRegistry.set(composerDrafts.composerCloudDraftsAtom, {
448 accountId: null,
449 signedOut: {},
450 });
451 composerDrafts.resetComposerDraftsLoadState();
452 await composerDrafts.restoreCloudComposerDrafts("account-a");
453 expect(remainingMessages()).toEqual([]);
454 },
455 );
456
457 it("preserves an archived edit when an older payload finishes delivery", async () => {
458 const message = queuedMessage({ messageId: "edited-during-sign-out", text: "Original" });
459 await harness.manager.enqueue(message);
460 const deliveryRevision = harness.manager.revisionOf(message.messageId);
461 const edited = { ...message, text: "Keep this edit" };
462 await harness.manager.update(edited);
463 await composerDrafts.archiveCloudComposerDrafts("account-a", new Set([message.environmentId]), [
464 message.environmentId,
465 ]);
466 await harness.manager.clearEnvironment(message.environmentId);
467 await expect(completeQueuedMessageDelivery(message, deliveryRevision, undefined)).resolves.toBe(
468 "edited",
469 );
470 await composerDrafts.restoreCloudComposerDrafts("account-a");
471 expect(remainingMessages()).toEqual([edited]);
472 });
473
474 it("retries only cleanup after an acknowledged send removal fails", async () => {
475 const message = queuedMessage({ messageId: "message-acknowledged", text: "delivered" });
476 const acknowledged = new Set([message.messageId]);
477 harness.removeOutboxMessage.mockRejectedValueOnce(new Error("storage unavailable"));
478 await harness.manager.enqueue(message);
479
480 await expect(removeAcknowledgedExistingThreadMessage(message, acknowledged)).resolves.toBe(
481 false,
482 );
483 expect(remainingMessages()).toEqual([message]);
484 expect(acknowledged).toEqual(new Set([message.messageId]));
485
486 await expect(removeAcknowledgedExistingThreadMessage(message, acknowledged)).resolves.toBe(
487 true,
488 );
489 expect(remainingMessages()).toEqual([]);
490 expect(acknowledged).toEqual(new Set());
491 });
492
493 it("keeps an edited message and its files when delivery cleanup loses the revision race", async () => {
494 const message = queuedMessage({
495 messageId: "message-edited",
496 text: "original",
497 fileUri: "file:///documents/t3-composer-attachments/report.pdf",
498 });
499 await harness.manager.enqueue(message);
500 const deliveryRevision = harness.manager.revisionOf(message.messageId);
501 const edited = { ...message, text: "edited while the turn delivered" };
502 await harness.manager.update(edited);
503
504 await expect(completeQueuedMessageDelivery(message, deliveryRevision, undefined)).resolves.toBe(
505 "edited",
506 );
507
508 expect(remainingMessages()).toEqual([edited]);
509 expect(harness.removePersistedFile).not.toHaveBeenCalled();
510 });
511
512 it("removes the delivered message when no edit was accepted", async () => {
513 const message = queuedMessage({ messageId: "message-clean", text: "hello" });
514 await harness.manager.enqueue(message);
515 const deliveryRevision = harness.manager.revisionOf(message.messageId);
516
517 await expect(completeQueuedMessageDelivery(message, deliveryRevision, undefined)).resolves.toBe(
518 "removed",
519 );
520
521 expect(remainingMessages()).toEqual([]);
522 expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([message]);
523 });
524
525 it("keeps a delivered message when its editor opens during storage removal", async () => {
526 const message = queuedMessage({
527 messageId: "message-editor-removal-race",
528 text: "keep editor changes",
529 fileUri: "file:///documents/t3-composer-attachments/editor-race.pdf",
530 });
531 const removeStarted = Promise.withResolvers<void>();
532 const removeBarrier = Promise.withResolvers<void>();
533 harness.removeOutboxMessage.mockImplementationOnce(async () => {
534 removeStarted.resolve();
535 await removeBarrier.promise;
536 });
537 await harness.manager.enqueue(message);
538 const deliveryRevision = harness.manager.revisionOf(message.messageId);
539
540 const cleanup = completeQueuedMessageDelivery(message, deliveryRevision, undefined);
541 await removeStarted.promise;
542 appAtomRegistry.set(editingQueuedMessageIdsAtom, { [message.messageId]: true });
543 removeBarrier.resolve();
544
545 await expect(cleanup).resolves.toBe("edited");
546 expect(remainingMessages()).toEqual([message]);
547 expect(harness.removePersistedFile).not.toHaveBeenCalled();
548 });
549});
550
551describe("thread outbox delivered creation recovery", () => {
552 it("keeps an edit accepted while the older payload is persisted to the draft", async () => {
553 const message = queuedMessage({
554 messageId: "message-recovery-race",
555 text: "original queued text",
556 fileUri: "file:///documents/t3-composer-attachments/report.pdf",
557 });
558 const originalMergeComposerDraftContent = composerDrafts.mergeComposerDraftContent;
559 const mergeCompleted = Promise.withResolvers<void>();
560 const releaseRecovery = Promise.withResolvers<void>();
561 const mergeSpy = vi
562 .spyOn(composerDrafts, "mergeComposerDraftContent")
563 .mockImplementation(async (draftKey, content) => {
564 const result = await originalMergeComposerDraftContent(draftKey, content);
565 mergeCompleted.resolve();
566 await releaseRecovery.promise;
567 return result;
568 });
569
570 try {
571 await harness.manager.enqueue(message);
572 const recovery = recoverEditedCreationAfterDelivery(message);
573 await mergeCompleted.promise;
574
575 const newer = { ...message, text: "edited while recovery persisted the draft" };
576 await harness.manager.update(newer);
577
578 releaseRecovery.resolve();
579 await expect(recovery).resolves.toBe(false);
580
581 expect(remainingMessages()).toEqual([newer]);
582 expect(
583 composerDrafts.getComposerDraftSnapshot(`${message.environmentId}:${message.threadId}`),
584 ).toMatchObject({ text: message.text, attachments: [] });
585 expect(harness.removePersistedFile).not.toHaveBeenCalled();
586 } finally {
587 releaseRecovery.resolve();
588 mergeSpy.mockRestore();
589 }
590 });
591
592 it("leaves recovery to an editor that opens while the draft persists", async () => {
593 const message = queuedMessage({
594 messageId: "message-recovery-editor",
595 text: "recover this text",
596 fileUri: "file:///documents/t3-composer-attachments/editor.pdf",
597 });
598 const originalMergeComposerDraftContent = composerDrafts.mergeComposerDraftContent;
599 const mergeCompleted = Promise.withResolvers<void>();
600 const releaseRecovery = Promise.withResolvers<void>();
601 const mergeSpy = vi
602 .spyOn(composerDrafts, "mergeComposerDraftContent")
603 .mockImplementation(async (draftKey, content) => {
604 const result = await originalMergeComposerDraftContent(draftKey, content);
605 mergeCompleted.resolve();
606 await releaseRecovery.promise;
607 return result;
608 });
609
610 try {
611 await harness.manager.enqueue(message);
612 const recovery = recoverEditedCreationAfterDelivery(message);
613 await mergeCompleted.promise;
614 appAtomRegistry.set(editingQueuedMessageIdsAtom, { [message.messageId]: true });
615
616 releaseRecovery.resolve();
617 await expect(recovery).resolves.toBe(true);
618
619 expect(remainingMessages()).toEqual([message]);
620 expect(
621 composerDrafts.getComposerDraftSnapshot(`${message.environmentId}:${message.threadId}`),
622 ).toMatchObject({ text: message.text, attachments: [] });
623 expect(harness.removePersistedFile).not.toHaveBeenCalled();
624 } finally {
625 releaseRecovery.resolve();
626 mergeSpy.mockRestore();
627 }
628 });
629
630 it("retries a failed removal without duplicating recovered draft content", async () => {
631 const message = queuedMessage({
632 messageId: "message-recovery-removal",
633 text: "recover once",
634 fileUri: "file:///documents/t3-composer-attachments/retry.pdf",
635 });
636 const draftKey = `${message.environmentId}:${message.threadId}`;
637 const removeSpy = vi
638 .spyOn(harness.manager, "remove")
639 .mockRejectedValueOnce(new Error("storage unavailable"));
640
641 try {
642 await harness.manager.enqueue(message);
643
644 await expect(recoverEditedCreationAfterDelivery(message)).resolves.toBe(false);
645 expect(remainingMessages()).toEqual([message]);
646
647 await expect(recoverEditedCreationAfterDelivery(message)).resolves.toBe(true);
648
649 const draft = composerDrafts.getComposerDraftSnapshot(draftKey);
650 expect(draft.text).toBe(message.text);
651 expect(draft.attachments).toEqual(message.attachments);
652 expect(remainingMessages()).toEqual([]);
653 expect(harness.removePersistedFile).not.toHaveBeenCalled();
654 } finally {
655 removeSpy.mockRestore();
656 }
657 });
658
659 it("keeps the queue entry when the recovered draft cannot persist", async () => {
660 const message = queuedMessage({
661 messageId: "message-recovery-persistence",
662 text: "recover after persistence returns",
663 });
664 await harness.manager.enqueue(message);
665 harness.draftFile.setWriteError(new Error("disk full"));
666
667 await expect(recoverEditedCreationAfterDelivery(message)).resolves.toBe(false);
668
669 expect(remainingMessages()).toEqual([message]);
670 });
671});
672
673describe("thread outbox recovery rollback", () => {
674 it("reopens a rejected task with setup edits and every attachment, even above the send cap", async () => {
675 const message = queuedMessage({ messageId: "failed-setup", text: "Original prompt" });
676 const sourceKey = `${message.environmentId}:${message.threadId}`;
677 const targetKey = "new-task:restored-failed-setup";
678 const files = Array.from(
679 { length: 10 },
680 (_, index) =>
681 queuedMessage({
682 messageId: `attachment-${index}`,
683 text: "",
684 fileUri: `file:///file-${index}`,
685 }).attachments[0]!,
686 );
687 appAtomRegistry.set(composerDrafts.composerDraftsAtom, {
688 [targetKey]: { text: message.text, attachments: files.slice(0, 8) },
689 [sourceKey]: { text: "Please include tests", attachments: files.slice(8) },
690 });
691 await recoverFailedThreadDraft(message);
692 expect(composerDrafts.getComposerDraftSnapshot(targetKey)).toMatchObject({
693 text: "Original prompt\n\nPlease include tests",
694 attachments: files,
695 });
696 expect(composerDrafts.getComposerDraftSnapshot(sourceKey)).toMatchObject({
697 text: "",
698 attachments: [],
699 });
700 await recoverFailedThreadDraft(message);
701 expect(composerDrafts.getComposerDraftSnapshot(targetKey).text).toBe(
702 "Original prompt\n\nPlease include tests",
703 );
704 });
705
706 it("keeps setup edits recoverable when saving their recovery draft fails", async () => {
707 const message = queuedMessage({ messageId: "failed-save", text: "Original prompt" });
708 const sourceKey = `${message.environmentId}:${message.threadId}`;
709 appAtomRegistry.set(composerDrafts.composerDraftsAtom, {
710 "new-task:restored-failed-save": { text: message.text, attachments: [] },
711 [sourceKey]: { text: "Follow-up", attachments: [] },
712 });
713 harness.draftFile.setWriteError(new Error("disk full"));
714 await expect(recoverFailedThreadDraft(message)).rejects.toThrow("Composer draft persistence");
715 expect(composerDrafts.getComposerDraftSnapshot(sourceKey).text).toBe("Follow-up");
716 harness.draftFile.setWriteError(null);
717 await recoverFailedThreadDraft(message);
718 expect(composerDrafts.getComposerDraftSnapshot("new-task:restored-failed-save").text).toBe(
719 "Original prompt\n\nFollow-up",
720 );
721 });
722
723 it("restores a rejected new task as its own draft for the project", async () => {
724 const message: QueuedThreadMessage = {
725 ...queuedMessage({ messageId: "message-creation-restore", text: "new task text" }),
726 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6-sol" },
727 creation: {
728 projectId: ProjectId.make("project-1"),
729 workspaceMode: "local",
730 branch: null,
731 worktreePath: null,
732 },
733 };
734 await harness.manager.enqueue(message);
735
736 await expect(restoreRejectedQueuedMessage(message, "rejected by server")).resolves.toBe(
737 "restored",
738 );
739
740 expect(
741 composerDrafts.getComposerDraftSnapshot(`new-task:restored-${message.messageId}`),
742 ).toMatchObject({
743 text: message.text,
744 attachments: message.attachments,
745 modelSelection: message.modelSelection,
746 project: {
747 environmentId: message.environmentId,
748 projectId: message.creation!.projectId,
749 createdAt: message.createdAt,
750 },
751 });
752 expect(remainingMessages()).toEqual([]);
753 expect(appAtomRegistry.get(threadComposerErrorsAtom)).toEqual({});
754 expect(
755 appAtomRegistry.get(pendingThreadCreationOutcomesAtom)[
756 `${message.environmentId}:${message.threadId}`
757 ],
758 ).toEqual({ kind: "failed", message, reason: "rejected by server" });
759 });
760
761 it("does not record a creation outcome for a rejected follow-up message", async () => {
762 const message = queuedMessage({ messageId: "message-followup-restore", text: "follow up" });
763 await harness.manager.enqueue(message);
764
765 await expect(restoreRejectedQueuedMessage(message, "rejected")).resolves.toBe("restored");
766
767 expect(appAtomRegistry.get(pendingThreadCreationOutcomesAtom)).toEqual({});
768 expect(
769 appAtomRegistry.get(threadComposerErrorsAtom)[`${message.environmentId}:${message.threadId}`],
770 ).toEqual({ message: "rejected", messageId: message.messageId });
771 });
772
773 it("leaves no error when the restored text is resent before recovery finishes", async () => {
774 const message = queuedMessage({ messageId: "message-resent", text: "resend me" });
775 const threadKey = `${message.environmentId}:${message.threadId}`;
776 await harness.manager.enqueue(message);
777 const unsubscribe = appAtomRegistry.subscribe(composerDrafts.composerDraftsAtom, (drafts) => {
778 if (drafts[threadKey]?.text === "resend me") {
779 unsubscribe();
780 clearThreadComposerError(threadKey);
781 void composerDrafts.clearComposerDraftContent(threadKey);
782 }
783 });
784
785 await expect(restoreRejectedQueuedMessage(message, "rejected")).resolves.toBe("restored");
786
787 expect(appAtomRegistry.get(threadComposerErrorsAtom)).toEqual({});
788 });
789
790 it("withdraws the error when an edit makes the recovery back out", async () => {
791 const message = queuedMessage({ messageId: "message-edited-mid-recovery", text: "edit me" });
792 const threadKey = `${message.environmentId}:${message.threadId}`;
793 await harness.manager.enqueue(message);
794 const unsubscribe = appAtomRegistry.subscribe(composerDrafts.composerDraftsAtom, (drafts) => {
795 if (drafts[threadKey]?.text === "edit me") {
796 unsubscribe();
797 appAtomRegistry.set(editingQueuedMessageIdsAtom, { [message.messageId]: true });
798 }
799 });
800
801 await expect(restoreRejectedQueuedMessage(message, "rejected")).resolves.toBe("deferred");
802
803 expect(appAtomRegistry.get(threadComposerErrorsAtom)).toEqual({});
804 expect(remainingMessages()).toEqual([message]);
805 });
806
807 it("rolls a failed recovery merge back so the retry cannot duplicate the text", async () => {
808 const message = queuedMessage({ messageId: "message-restore", text: "queued text" });
809 const draftKey = `${message.environmentId}:${message.threadId}`;
810 appAtomRegistry.set(composerDrafts.composerDraftsAtom, {
811 [draftKey]: { text: "typed offline", attachments: [] },
812 });
813 await harness.manager.enqueue(message);
814
815 harness.draftFile.setWriteError(new Error("disk full"));
816 await expect(restoreRejectedQueuedMessage(message, "too large")).resolves.toBe("retry");
817
818 expect(composerDrafts.getComposerDraftSnapshot(draftKey).text).toBe("typed offline");
819 expect(remainingMessages()).toEqual([message]);
820
821 harness.draftFile.setWriteError(null);
822 await expect(restoreRejectedQueuedMessage(message, "too large")).resolves.toBe("restored");
823
824 expect(composerDrafts.getComposerDraftSnapshot(draftKey).text).toBe(
825 "typed offline\n\nqueued text",
826 );
827 expect(remainingMessages()).toEqual([]);
828 expect(appAtomRegistry.get(threadComposerErrorsAtom)[draftKey]?.message).toBe("too large");
829 });
830});