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