1import { describe, expect, it } from "@effect/vitest";
2import { EnvironmentNotRegisteredError } from "@t3tools/client-runtime/connection";
3import { isTransportConnectionErrorMessage } from "@t3tools/client-runtime/errors";
4import { EnvironmentRpcUnavailableError } from "@t3tools/client-runtime/rpc";
5import {
6 CommandId,
7 ComposerContextId,
8 EnvironmentAuthorizationError,
9 EnvironmentId,
10 MessageId,
11 OrchestrationDispatchCommandError,
12 ProjectId,
13 ProviderInstanceId,
14 ThreadId,
15} from "@t3tools/contracts";
16import { AtomRegistry } from "effect/reactivity";
17import * as RpcClientError from "effect/rpc/RpcClientError";
18import * as Socket from "effect/socket/Socket";
19import { onTestFinished, vi } from "vite-plus/test";
20
21const outboxFiles = vi.hoisted(() => new Map<string, string | Error>());
22
23vi.mock("expo-file-system", () => {
24 class Directory {
25 create() {}
26
27 list() {
28 return Array.from(outboxFiles.keys(), (name) => new File(name));
29 }
30 }
31
32 class File {
33 readonly name: string;
34 readonly parentDirectory = new Directory();
35
36 constructor(...parts: [string] | [Directory, string]) {
37 this.name = parts.length === 1 ? parts[0] : parts[1];
38 }
39
40 get exists() {
41 return outboxFiles.has(this.name);
42 }
43
44 create() {
45 outboxFiles.set(this.name, "");
46 }
47
48 write(contents: string) {
49 outboxFiles.set(this.name, contents);
50 }
51
52 moveSync(file: File) {
53 const contents = outboxFiles.get(this.name);
54 if (contents === undefined) throw new Error("Missing file");
55 outboxFiles.set(file.name, contents);
56 outboxFiles.delete(this.name);
57 }
58
59 delete() {
60 outboxFiles.delete(this.name);
61 }
62
63 async text(): Promise<string> {
64 const contents = outboxFiles.get(this.name);
65 if (contents instanceof Error) throw contents;
66 if (contents === undefined) throw new Error("Missing file");
67 return contents;
68 }
69 }
70
71 return {
72 File,
73 Directory,
74 Paths: { document: "/documents" },
75 };
76});
77
78import {
79 decodeQueuedThreadMessage,
80 encodeQueuedThreadMessage,
81 groupQueuedThreadMessages,
82 isQueuedThreadCreationSendable,
83 modelSelectionsEqual,
84 resolveThreadOutboxDeliveryAction,
85 resolveThreadOutboxDispatchStep,
86 resolveThreadOutboxFailureAction,
87 resolveQueuedThreadSettings,
88 shouldRetryThreadOutboxDelivery,
89 threadOutboxRetryDelayMs,
90 type QueuedThreadMessage,
91} from "./thread-outbox-model";
92import { createThreadOutboxManager, ThreadOutboxManagerError } from "./thread-outbox-manager";
93import {
94 expoThreadOutboxStorage,
95 ThreadOutboxStorageError,
96 type ThreadOutboxLoadResult,
97 type ThreadOutboxStorage,
98} from "./thread-outbox-storage";
99
100function queuedMessage(input: {
101 readonly environmentId?: string;
102 readonly threadId?: string;
103 readonly messageId: string;
104 readonly createdAt: string;
105}): QueuedThreadMessage {
106 return {
107 environmentId: EnvironmentId.make(input.environmentId ?? "environment-1"),
108 threadId: ThreadId.make(input.threadId ?? "thread-1"),
109 messageId: MessageId.make(input.messageId),
110 commandId: CommandId.make(`command-${input.messageId}`),
111 text: input.messageId,
112 attachments: [],
113 createdAt: input.createdAt,
114 };
115}
116
117describe("thread outbox", () => {
118 it("retains structured context through a persisted offline queue round trip", () => {
119 const message: QueuedThreadMessage = {
120 ...queuedMessage({ messageId: "context-message", createdAt: "2026-09-06T12:00:00.000Z" }),
121 text: "[Build](t3-context://v1/terminal/build-output)",
122 context: {
123 version: 1,
124 records: [
125 {
126 version: 1,
127 kind: "terminal",
128 contextId: ComposerContextId.make("build-output"),
129 label: "Build",
130 terminalId: "main",
131 terminalLabel: "Terminal",
132 lineStart: 2,
133 lineEnd: 2,
134 text: "Build failed",
135 },
136 ],
137 },
138 };
139 expect(
140 decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))),
141 ).toEqual(message);
142 });
143 it("retains queue mode when a queued provider switch reloads from storage", () => {
144 const message: QueuedThreadMessage = {
145 ...queuedMessage({ messageId: "queued-switch", createdAt: "2026-09-17T09:00:00.000Z" }),
146 dispatchMode: "queue",
147 };
148 expect(
149 decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))),
150 ).toEqual(message);
151 });
152 it.each(["read", "json", "schema"] as const)(
153 "recovers usable messages without permitting cleanup after a record %s failure",
154 async (failure) => {
155 onTestFinished(() => outboxFiles.clear());
156 const first = queuedMessage({
157 messageId: "message-1",
158 createdAt: "2026-06-08T10:00:01.000Z",
159 });
160 const second = queuedMessage({
161 environmentId: "environment-2",
162 messageId: "message-2",
163 createdAt: "2026-06-08T10:00:02.000Z",
164 });
165 // Put the unreadable record first to check that later records still load.
166 outboxFiles.set(
167 "message-2.json",
168 failure === "read"
169 ? new Error("storage unavailable")
170 : failure === "json"
171 ? "{"
172 : JSON.stringify({ ...second, schemaVersion: 999 }),
173 );
174 outboxFiles.set("message-1.json", JSON.stringify(encodeQueuedThreadMessage(first)));
175 const unreadable = outboxFiles.get("message-2.json");
176
177 await expect(expoThreadOutboxStorage.load()).resolves.toMatchObject({
178 messages: [first],
179 errors: [{ operation: "read-message", fileName: "message-2.json" }],
180 });
181
182 const registry = AtomRegistry.make();
183 onTestFinished(() => registry.dispose());
184 const manager = createThreadOutboxManager({
185 registry,
186 storage: expoThreadOutboxStorage,
187 warn: () => {},
188 });
189 await expect(manager.load()).resolves.toBe(false);
190 const recovered = registry.get(manager.queuedMessagesByThreadKeyAtom)[
191 "environment-1:thread-1"
192 ]![0]!;
193 expect(recovered).toEqual(first);
194 await expect(manager.confirmQueued(recovered)).resolves.toBe(true);
195 const edited = { ...recovered, text: "Edited after recovery" };
196 await expect(manager.update(edited)).resolves.toBe(true);
197 const beforeRetry = registry.get(manager.queuedMessagesByThreadKeyAtom);
198 await expect(manager.load()).resolves.toBe(false);
199 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toBe(beforeRetry);
200 await expect(manager.clearEnvironment(first.environmentId)).rejects.toMatchObject({
201 operation: "clear-environment-load",
202 });
203 expect(outboxFiles.get("message-2.json")).toBe(unreadable);
204 expect(outboxFiles.has("message-1.json")).toBe(true);
205
206 // A delivered readable message can leave the queue while the failed
207 // record stays intact. Its attachment cleanup has a separate guard.
208 await expect(manager.remove(edited)).resolves.toBe(edited);
209 expect(outboxFiles.has("message-1.json")).toBe(false);
210
211 outboxFiles.set("message-2.json", JSON.stringify(encodeQueuedThreadMessage(second)));
212 await expect(manager.load()).resolves.toBe(true);
213 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
214 "environment-2:thread-1": [second],
215 });
216 await expect(expoThreadOutboxStorage.load()).resolves.toEqual({
217 messages: [second],
218 errors: [],
219 });
220 },
221 );
222
223 it("preserves queued messages when environment cleanup cannot read the outbox", async () => {
224 const registry = AtomRegistry.make();
225 onTestFinished(() => registry.dispose());
226 const message = queuedMessage({
227 messageId: "message-1",
228 createdAt: "2026-06-08T10:00:01.000Z",
229 });
230 const stored = new Map<MessageId, QueuedThreadMessage>();
231 const manager = createThreadOutboxManager({
232 registry,
233 warn: () => {},
234 storage: {
235 load: async () => {
236 throw new Error("storage unavailable");
237 },
238 write: async (entry) => {
239 stored.set(entry.messageId, entry);
240 },
241 remove: async (entry) => {
242 stored.delete(entry.messageId);
243 },
244 },
245 });
246 await manager.enqueue(message);
247
248 await expect(manager.clearEnvironment(message.environmentId)).rejects.toMatchObject({
249 operation: "clear-environment-load",
250 });
251 expect([...stored.values()]).toEqual([message]);
252 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
253 "environment-1:thread-1": [message],
254 });
255 });
256
257 it("keeps in-session edits and removals when an incomplete load is retried", async () => {
258 const registry = AtomRegistry.make();
259 onTestFinished(() => registry.dispose());
260 const message = queuedMessage({
261 messageId: "message-retried",
262 createdAt: "2026-06-08T10:00:01.000Z",
263 });
264 const started = Promise.withResolvers<void>();
265 const response = Promise.withResolvers<ThreadOutboxLoadResult>();
266 const load = vi.fn<ThreadOutboxStorage["load"]>(async () => ({
267 messages: [message],
268 errors: [],
269 }));
270 load.mockImplementationOnce(() => {
271 started.resolve();
272 return response.promise;
273 });
274 const manager = createThreadOutboxManager({
275 registry,
276 storage: { load, write: async () => {}, remove: async () => {} },
277 warn: () => {},
278 });
279 const loading = manager.load();
280 await started.promise;
281 const edited = { ...message, text: "Accepted while storage was being read" };
282 const writing = manager.enqueue(edited);
283 response.resolve({
284 messages: [message],
285 errors: [
286 new ThreadOutboxStorageError({
287 operation: "read-message",
288 environmentId: null,
289 threadId: null,
290 messageId: null,
291 fileName: "unreadable.json",
292 cause: new Error("unreadable record"),
293 }),
294 ],
295 });
296 await expect(loading).resolves.toBe(false);
297 await writing;
298 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
299 "environment-1:thread-1": [edited],
300 });
301 await expect(manager.confirmQueued(edited)).resolves.toBe(true);
302
303 await manager.remove(edited);
304 // A later repaired record can contain a stale copy of a removed message.
305 await expect(manager.load()).resolves.toBe(true);
306 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});
307 expect(load).toHaveBeenCalledTimes(2);
308 });
309
310 it("groups messages by scoped thread and preserves creation order", () => {
311 const later = queuedMessage({
312 messageId: "message-2",
313 createdAt: "2026-06-08T10:00:02.000Z",
314 });
315 const earlier = queuedMessage({
316 messageId: "message-1",
317 createdAt: "2026-06-08T10:00:01.000Z",
318 });
319
320 expect(groupQueuedThreadMessages([later, earlier])).toEqual({
321 "environment-1:thread-1": [earlier, later],
322 });
323 });
324
325 it("decodes the persisted schema and rejects incomplete messages", () => {
326 const message = queuedMessage({
327 messageId: "message-1",
328 createdAt: "2026-06-08T10:00:01.000Z",
329 });
330
331 expect(
332 decodeQueuedThreadMessage({
333 schemaVersion: 1,
334 ...message,
335 }),
336 ).toEqual(message);
337 expect(() =>
338 decodeQueuedThreadMessage({
339 schemaVersion: 1,
340 environmentId: "environment-1",
341 }),
342 ).toThrow();
343 });
344
345 it("persists generic attachment paths without embedding their contents", () => {
346 const message = {
347 ...queuedMessage({
348 messageId: "message-file",
349 createdAt: "2026-06-08T10:00:01.000Z",
350 }),
351 attachments: [
352 {
353 id: "file-1",
354 type: "file" as const,
355 name: "report.pdf",
356 mimeType: "application/pdf",
357 sizeBytes: 42,
358 fileUri: "file:///documents/report.pdf",
359 uploadedAttachmentId: "pending-report-pdf",
360 uploadEnvironmentId: EnvironmentId.make("environment-1"),
361 },
362 ],
363 } satisfies QueuedThreadMessage;
364
365 expect(decodeQueuedThreadMessage(encodeQueuedThreadMessage(message))).toEqual(message);
366 });
367
368 it("reads file-backed images from v4 queued messages", () => {
369 const message = {
370 ...queuedMessage({
371 messageId: "message-image",
372 createdAt: "2026-06-08T10:00:01.000Z",
373 }),
374 attachments: [
375 {
376 id: "image-1",
377 type: "image" as const,
378 name: "photo.png",
379 mimeType: "image/png",
380 sizeBytes: 3,
381 fileUri: "file:///documents/t3-composer-attachments/photo.png",
382 previewUri: "file:///documents/t3-composer-attachments/photo.png",
383 uploadedAttachmentId: "pending-photo-png",
384 uploadEnvironmentId: EnvironmentId.make("environment-1"),
385 },
386 ],
387 } satisfies QueuedThreadMessage;
388
389 expect(decodeQueuedThreadMessage({ ...message, schemaVersion: 4 })).toEqual(message);
390 });
391
392 it("persists the exact selector snapshot while remaining compatible with v1 messages", () => {
393 const legacyMessage = queuedMessage({
394 messageId: "message-1",
395 createdAt: "2026-06-08T10:00:01.000Z",
396 });
397 const selectedMessage = {
398 ...legacyMessage,
399 modelSelection: {
400 instanceId: ProviderInstanceId.make("codex"),
401 model: "gpt-5.4",
402 options: [{ id: "reasoningEffort", value: "xhigh" }],
403 },
404 runtimeMode: "approval-required",
405 interactionMode: "plan",
406 } satisfies QueuedThreadMessage;
407
408 expect(decodeQueuedThreadMessage(encodeQueuedThreadMessage(selectedMessage))).toEqual(
409 selectedMessage,
410 );
411 expect(
412 resolveQueuedThreadSettings(legacyMessage, {
413 modelSelection: selectedMessage.modelSelection,
414 runtimeMode: selectedMessage.runtimeMode,
415 interactionMode: selectedMessage.interactionMode,
416 }),
417 ).toEqual({
418 modelSelection: selectedMessage.modelSelection,
419 runtimeMode: selectedMessage.runtimeMode,
420 interactionMode: selectedMessage.interactionMode,
421 });
422 });
423
424 it("compares model options as part of the queued settings change", () => {
425 const base = {
426 instanceId: ProviderInstanceId.make("codex"),
427 model: "gpt-5.4",
428 options: [{ id: "reasoningEffort", value: "medium" }],
429 } as const;
430
431 expect(modelSelectionsEqual(base, base)).toBe(true);
432 expect(
433 modelSelectionsEqual(base, {
434 ...base,
435 options: [{ id: "reasoningEffort", value: "xhigh" }],
436 }),
437 ).toBe(false);
438 });
439
440 it("normalizes queued plan mode against the queued provider, not the current thread", () => {
441 const codex = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6-sol" };
442 const antigravity = {
443 instanceId: ProviderInstanceId.make("google-personal"),
444 model: "gemini-test-thinking",
445 options: [{ id: "native-option", value: "keep-this-choice" }],
446 };
447 const providers = [
448 { instanceId: codex.instanceId, showInteractionModeToggle: true },
449 { instanceId: antigravity.instanceId, showInteractionModeToggle: false },
450 ];
451 const message = {
452 ...queuedMessage({ messageId: "queued-plan", createdAt: "2026-09-02T10:00:00.000Z" }),
453 text: "/plan inspect the project",
454 modelSelection: antigravity,
455 interactionMode: "plan",
456 } satisfies QueuedThreadMessage;
457
458 expect(
459 resolveQueuedThreadSettings(
460 message,
461 { modelSelection: codex, runtimeMode: "approval-required", interactionMode: "plan" },
462 providers,
463 ),
464 ).toEqual({
465 modelSelection: antigravity,
466 runtimeMode: "approval-required",
467 interactionMode: "default",
468 });
469 expect(
470 resolveQueuedThreadSettings(
471 { ...message, modelSelection: codex },
472 {
473 modelSelection: antigravity,
474 runtimeMode: "approval-required",
475 interactionMode: "default",
476 },
477 providers,
478 ).interactionMode,
479 ).toBe("plan");
480 });
481
482 it("normalizes a legacy queued message that inherits unsupported plan mode", () => {
483 const modelSelection = {
484 instanceId: ProviderInstanceId.make("google-personal"),
485 model: "gemini-test-thinking",
486 };
487 expect(
488 resolveQueuedThreadSettings(
489 queuedMessage({ messageId: "legacy-plan", createdAt: "2026-09-02T10:00:00.000Z" }),
490 { modelSelection, runtimeMode: "approval-required", interactionMode: "plan" },
491 [{ instanceId: modelSelection.instanceId, showInteractionModeToggle: false }],
492 ).interactionMode,
493 ).toBe("default");
494 });
495
496 it("backs off queued delivery retries and caps them at sixteen seconds", () => {
497 expect([1, 2, 3, 4, 5, 6].map(threadOutboxRetryDelayMs)).toEqual([
498 1_000, 2_000, 4_000, 8_000, 16_000, 16_000,
499 ]);
500 });
501
502 it("serializes mutations even when an earlier mutation is slower", async () => {
503 const registry = AtomRegistry.make();
504 const manager = createThreadOutboxManager({
505 registry,
506 storage: {
507 load: async () => ({ messages: [], errors: [] }),
508 write: async () => undefined,
509 remove: async () => undefined,
510 },
511 });
512 const order: string[] = [];
513 let releaseFirst!: () => void;
514 const firstBlocked = new Promise<void>((resolve) => {
515 releaseFirst = resolve;
516 });
517
518 const first = manager.serialize(async () => {
519 order.push("first:start");
520 await firstBlocked;
521 order.push("first:end");
522 });
523 const second = manager.serialize(async () => {
524 order.push("second");
525 });
526
527 await Promise.resolve();
528 expect(order).toEqual(["first:start"]);
529 releaseFirst();
530 await Promise.all([first, second]);
531 expect(order).toEqual(["first:start", "first:end", "second"]);
532 registry.dispose();
533 });
534
535 it("holds the mutation queue while persisted messages are loading", async () => {
536 const registry = AtomRegistry.make();
537 const message = queuedMessage({
538 messageId: "message-1",
539 createdAt: "2026-06-08T10:00:01.000Z",
540 });
541 const stored = new Map([[message.messageId, message]]);
542 let loadCalls = 0;
543 let removeCalls = 0;
544 let releaseInitialLoad!: () => void;
545 const initialLoadBlocked = new Promise<void>((resolve) => {
546 releaseInitialLoad = resolve;
547 });
548 const storage: ThreadOutboxStorage = {
549 load: async () => {
550 loadCalls += 1;
551 if (loadCalls === 1) {
552 await initialLoadBlocked;
553 }
554 return { messages: [...stored.values()], errors: [] };
555 },
556 write: async () => undefined,
557 remove: async (candidate) => {
558 removeCalls += 1;
559 stored.delete(candidate.messageId);
560 },
561 };
562 const manager = createThreadOutboxManager({ registry, storage });
563
564 const loading = manager.load();
565 await Promise.resolve();
566 const clearing = manager.clearEnvironment(message.environmentId);
567 await Promise.resolve();
568 await Promise.resolve();
569
570 expect(loadCalls).toBe(1);
571 expect(removeCalls).toBe(0);
572
573 releaseInitialLoad();
574 await Promise.all([loading, clearing]);
575 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});
576 registry.dispose();
577 });
578
579 it("reports structured load failures and permits a retry", async () => {
580 const registry = AtomRegistry.make();
581 const loadCause = new Error("storage unavailable");
582 const warnings: Array<{ message: string; error: unknown }> = [];
583 let loadCalls = 0;
584 const manager = createThreadOutboxManager({
585 registry,
586 storage: {
587 load: async () => {
588 loadCalls += 1;
589 if (loadCalls === 1) throw loadCause;
590 return { messages: [], errors: [] };
591 },
592 write: async () => undefined,
593 remove: async () => undefined,
594 },
595 warn: (message, error) => warnings.push({ message, error }),
596 });
597
598 await manager.load();
599 expect(warnings).toEqual([
600 {
601 message: "[thread-outbox] failed to load persisted messages",
602 error: new ThreadOutboxManagerError({
603 operation: "load",
604 environmentId: null,
605 threadId: null,
606 messageId: null,
607 cause: loadCause,
608 }),
609 },
610 ]);
611
612 await manager.load();
613 expect(loadCalls).toBe(2);
614 registry.dispose();
615 });
616
617 it("keeps atom state aligned with durable writes and removals", async () => {
618 const registry = AtomRegistry.make();
619 const stored = new Map<MessageId, QueuedThreadMessage>();
620 const removalCause = new Error("remove failed");
621 let failRemoval = true;
622 const storage: ThreadOutboxStorage = {
623 load: async () => ({ messages: [...stored.values()], errors: [] }),
624 write: async (message) => {
625 stored.set(message.messageId, message);
626 },
627 remove: async (message) => {
628 if (failRemoval) {
629 throw removalCause;
630 }
631 stored.delete(message.messageId);
632 },
633 };
634 const manager = createThreadOutboxManager({ registry, storage });
635 const message = queuedMessage({
636 messageId: "message-1",
637 createdAt: "2026-06-08T10:00:01.000Z",
638 });
639
640 await manager.enqueue(message);
641 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
642 "environment-1:thread-1": [message],
643 });
644
645 await expect(manager.remove(message)).rejects.toEqual(
646 new ThreadOutboxManagerError({
647 operation: "remove",
648 environmentId: message.environmentId,
649 threadId: message.threadId,
650 messageId: message.messageId,
651 cause: removalCause,
652 }),
653 );
654 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
655 "environment-1:thread-1": [message],
656 });
657
658 failRemoval = false;
659 await manager.remove(message);
660 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});
661 registry.dispose();
662 });
663
664 it("publishes an enqueued message before the durable write resolves", async () => {
665 const registry = AtomRegistry.make();
666 let releaseWrite!: () => void;
667 const writeBlocked = new Promise<void>((resolve) => {
668 releaseWrite = resolve;
669 });
670 const manager = createThreadOutboxManager({
671 registry,
672 storage: {
673 load: async () => ({ messages: [], errors: [] }),
674 write: async () => writeBlocked,
675 remove: async () => undefined,
676 },
677 });
678 const message = queuedMessage({
679 messageId: "message-1",
680 createdAt: "2026-06-08T10:00:01.000Z",
681 });
682
683 const enqueueing = manager.enqueue(message);
684 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
685 "environment-1:thread-1": [message],
686 });
687
688 releaseWrite();
689 await enqueueing;
690 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
691 "environment-1:thread-1": [message],
692 });
693 registry.dispose();
694 });
695
696 it("rolls an enqueued message back out when the durable write fails", async () => {
697 const registry = AtomRegistry.make();
698 const writeCause = new Error("disk full");
699 const manager = createThreadOutboxManager({
700 registry,
701 storage: {
702 load: async () => ({ messages: [], errors: [] }),
703 write: async () => {
704 throw writeCause;
705 },
706 remove: async () => undefined,
707 },
708 });
709 const message = queuedMessage({
710 messageId: "message-1",
711 createdAt: "2026-06-08T10:00:01.000Z",
712 });
713
714 await expect(manager.enqueue(message)).rejects.toEqual(
715 new ThreadOutboxManagerError({
716 operation: "enqueue",
717 environmentId: message.environmentId,
718 threadId: message.threadId,
719 messageId: message.messageId,
720 cause: writeCause,
721 }),
722 );
723 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});
724 registry.dispose();
725 });
726
727 it("drops the disk entry when a failed enqueue leaves no queued message behind", async () => {
728 const registry = AtomRegistry.make();
729 const removed: string[] = [];
730 const manager = createThreadOutboxManager({
731 registry,
732 storage: {
733 load: async () => ({ messages: [], errors: [] }),
734 write: async () => {
735 throw new Error("disk full");
736 },
737 remove: async (message) => {
738 removed.push(message.messageId);
739 },
740 },
741 });
742 const message = queuedMessage({
743 messageId: "message-1",
744 createdAt: "2026-06-08T10:00:01.000Z",
745 });
746
747 // A concurrent update losing its race can compensate-write this payload
748 // to disk before this write fails; rollback must clear that copy or a
749 // restart resurrects the message.
750 await expect(manager.enqueue(message)).rejects.toBeInstanceOf(ThreadOutboxManagerError);
751 expect(removed).toEqual(["message-1"]);
752 registry.dispose();
753 });
754
755 it("keeps a same-id retry queued when the first attempt's write fails", async () => {
756 const registry = AtomRegistry.make();
757 let failNextWrite = true;
758 let releaseFirstWrite!: () => void;
759 const firstWriteBlocked = new Promise<void>((resolve) => {
760 releaseFirstWrite = resolve;
761 });
762 const manager = createThreadOutboxManager({
763 registry,
764 storage: {
765 load: async () => ({ messages: [], errors: [] }),
766 write: async () => {
767 if (failNextWrite) {
768 failNextWrite = false;
769 await firstWriteBlocked;
770 throw new Error("disk full");
771 }
772 },
773 remove: async () => undefined,
774 },
775 });
776 const message = queuedMessage({
777 messageId: "message-1",
778 createdAt: "2026-06-08T10:00:01.000Z",
779 });
780 const retried = { ...message, text: "retried" };
781
782 const first = manager.enqueue(message);
783 const second = manager.enqueue(retried);
784 releaseFirstWrite();
785 await expect(first).rejects.toBeInstanceOf(ThreadOutboxManagerError);
786 await second;
787
788 // The failed first attempt must not roll back the retry that replaced it.
789 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
790 "environment-1:thread-1": [retried],
791 });
792 await expect(manager.confirmQueued(retried)).resolves.toBe(true);
793 await expect(manager.confirmQueued(message)).resolves.toBe(false);
794 registry.dispose();
795 });
796
797 it("replaces an existing message when an enqueue retry uses the same id", async () => {
798 const registry = AtomRegistry.make();
799 const manager = createThreadOutboxManager({
800 registry,
801 storage: {
802 load: async () => ({ messages: [], errors: [] }),
803 write: async () => undefined,
804 remove: async () => undefined,
805 },
806 });
807 const message = queuedMessage({
808 messageId: "message-1",
809 createdAt: "2026-06-08T10:00:01.000Z",
810 });
811 const retried = { ...message, text: "retried" };
812
813 await manager.enqueue(message);
814 await manager.enqueue(retried);
815
816 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
817 "environment-1:thread-1": [retried],
818 });
819 registry.dispose();
820 });
821
822 it("updates a queued message in place but never resurrects a removed one", async () => {
823 const registry = AtomRegistry.make();
824 const stored = new Map<MessageId, QueuedThreadMessage>();
825 const storage: ThreadOutboxStorage = {
826 load: async () => ({ messages: [...stored.values()], errors: [] }),
827 write: async (message) => {
828 stored.set(message.messageId, message);
829 },
830 remove: async (message) => {
831 stored.delete(message.messageId);
832 },
833 };
834 const manager = createThreadOutboxManager({ registry, storage });
835 const message = queuedMessage({
836 messageId: "message-1",
837 createdAt: "2026-06-08T10:00:01.000Z",
838 });
839
840 await manager.enqueue(message);
841 const edited = { ...message, text: "edited" };
842 await expect(manager.update(edited)).resolves.toBe(true);
843 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
844 "environment-1:thread-1": [edited],
845 });
846 expect(stored.get(message.messageId)).toEqual(edited);
847
848 await manager.remove(edited);
849 await expect(manager.update({ ...message, text: "stale flush" })).resolves.toBe(false);
850 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});
851 expect(stored.size).toBe(0);
852 registry.dispose();
853 });
854
855 it("rejects a stale revision before its payload reaches durable storage", async () => {
856 const registry = AtomRegistry.make();
857 const writes: string[] = [];
858 const manager = createThreadOutboxManager({
859 registry,
860 storage: {
861 load: async () => ({ messages: [], errors: [] }),
862 write: async (message) => {
863 writes.push(message.text);
864 },
865 remove: async () => undefined,
866 },
867 });
868 const original = queuedMessage({
869 messageId: "message-edit-race",
870 createdAt: "2026-06-08T10:00:01.000Z",
871 });
872 const edited = { ...original, text: "keep my changes" };
873
874 await manager.enqueue(original);
875 // Revision captured before slow work (an attachment upload) starts.
876 const revision = manager.revisionOf(original.messageId);
877 await manager.update(edited);
878
879 await expect(manager.update({ ...original, text: "stale upload" }, revision)).resolves.toBe(
880 false,
881 );
882 // The losing writer was rejected before persisting: no stale payload can
883 // sit on disk waiting to resurrect on the next load.
884 expect(writes).toEqual([original.text, "keep my changes"]);
885 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
886 "environment-1:thread-1": [edited],
887 });
888 registry.dispose();
889 });
890
891 it("does not publish a stale attachment update after a replacement appears during its write", async () => {
892 const registry = AtomRegistry.make();
893 const writes: string[] = [];
894 let resumeWrite: () => void = () => {};
895 let signalWriteStarted: () => void = () => {};
896 const writeStarted = new Promise<void>((resolve) => {
897 signalWriteStarted = resolve;
898 });
899 const writeBarrier = new Promise<void>((resolve) => {
900 resumeWrite = resolve;
901 });
902 const manager = createThreadOutboxManager({
903 registry,
904 storage: {
905 load: async () => ({ messages: [], errors: [] }),
906 write: async (message) => {
907 writes.push(message.text);
908 if (message.text === "stale upload") {
909 signalWriteStarted();
910 await writeBarrier;
911 }
912 },
913 remove: async () => undefined,
914 },
915 });
916 const original = queuedMessage({
917 messageId: "message-write-race",
918 createdAt: "2026-06-08T10:00:01.000Z",
919 });
920 const replacement = { ...original, text: "newer edit" };
921
922 await manager.enqueue(original);
923 const update = manager.update(
924 { ...original, text: "stale upload" },
925 manager.revisionOf(original.messageId),
926 );
927 await writeStarted;
928 const enqueue = manager.enqueue(replacement);
929 resumeWrite();
930
931 await expect(update).resolves.toBe(false);
932 // The losing update re-writes the winning payload inside its own
933 // mutation, before the replacement's serialized write lands, so a crash
934 // between the two cannot leave the stale payload on disk.
935 expect(writes).toEqual([original.text, "stale upload", "newer edit", "newer edit"]);
936 await enqueue;
937 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
938 "environment-1:thread-1": [replacement],
939 });
940 registry.dispose();
941 });
942
943 it("refuses to remove a message that was rewritten after the removal decision", async () => {
944 const registry = AtomRegistry.make();
945 const stored = new Map<MessageId, QueuedThreadMessage>();
946 const manager = createThreadOutboxManager({
947 registry,
948 storage: {
949 load: async () => ({ messages: [...stored.values()], errors: [] }),
950 write: async (message) => {
951 stored.set(message.messageId, message);
952 },
953 remove: async (message) => {
954 stored.delete(message.messageId);
955 },
956 },
957 });
958 const original = queuedMessage({
959 messageId: "message-remove-race",
960 createdAt: "2026-06-08T10:00:01.000Z",
961 });
962 const edited = { ...original, text: "edited while restoring" };
963
964 await manager.enqueue(original);
965 // Revision captured when restore-to-composer read the payload it intends
966 // to remove; the edit accepted afterwards must survive the removal.
967 const revision = manager.revisionOf(original.messageId);
968 await manager.update(edited);
969
970 await expect(manager.remove(original, revision)).resolves.toBe(null);
971 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
972 "environment-1:thread-1": [edited],
973 });
974 expect(stored.get(original.messageId)).toEqual(edited);
975
976 await expect(manager.remove(edited, manager.revisionOf(edited.messageId))).resolves.toEqual(
977 edited,
978 );
979 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});
980 registry.dispose();
981 });
982
983 it("keeps a retry enqueued when its publish races a revision-checked removal", async () => {
984 const registry = AtomRegistry.make();
985 const stored = new Map<MessageId, QueuedThreadMessage>();
986 const removeStarted = Promise.withResolvers<void>();
987 const removeBarrier = Promise.withResolvers<void>();
988 const replacementWriteStarted = Promise.withResolvers<void>();
989 const replacementWriteBarrier = Promise.withResolvers<void>();
990 const original = queuedMessage({
991 messageId: "message-remove-enqueue-race",
992 createdAt: "2026-06-08T10:00:01.000Z",
993 });
994 const retried = { ...original, text: "retried" };
995 const manager = createThreadOutboxManager({
996 registry,
997 storage: {
998 load: async () => ({ messages: [...stored.values()], errors: [] }),
999 write: async (message) => {
1000 if (message === retried) {
1001 replacementWriteStarted.resolve();
1002 await replacementWriteBarrier.promise;
1003 }
1004 stored.set(message.messageId, message);
1005 },
1006 remove: async (message) => {
1007 removeStarted.resolve();
1008 await removeBarrier.promise;
1009 stored.delete(message.messageId);
1010 },
1011 },
1012 });
1013
1014 await manager.enqueue(original);
1015 const removal = manager.remove(original, manager.revisionOf(original.messageId));
1016 let removalSettled = false;
1017 void removal.then(() => {
1018 removalSettled = true;
1019 });
1020 await removeStarted.promise;
1021 // Published synchronously while the durable remove is still in flight.
1022 const enqueue = manager.enqueue(retried);
1023 removeBarrier.resolve();
1024 await replacementWriteStarted.promise;
1025
1026 // The canceled removal itself restores the durable winner. The queued
1027 // enqueue write has not had a chance to run yet.
1028 expect(removalSettled).toBe(false);
1029 replacementWriteBarrier.resolve();
1030 await expect(removal).resolves.toBe(null);
1031 expect(stored.get(original.messageId)).toEqual(retried);
1032 await enqueue;
1033 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
1034 "environment-1:thread-1": [retried],
1035 });
1036 expect(stored.get(original.messageId)).toEqual(retried);
1037 registry.dispose();
1038 });
1039
1040 it("restores a message when its live removal predicate changes during storage removal", async () => {
1041 const registry = AtomRegistry.make();
1042 const stored = new Map<MessageId, QueuedThreadMessage>();
1043 const removeStarted = Promise.withResolvers<void>();
1044 const removeBarrier = Promise.withResolvers<void>();
1045 const manager = createThreadOutboxManager({
1046 registry,
1047 storage: {
1048 load: async () => ({ messages: [...stored.values()], errors: [] }),
1049 write: async (message) => {
1050 stored.set(message.messageId, message);
1051 },
1052 remove: async (message) => {
1053 removeStarted.resolve();
1054 await removeBarrier.promise;
1055 stored.delete(message.messageId);
1056 },
1057 },
1058 });
1059 const message = queuedMessage({
1060 messageId: "message-remove-predicate-race",
1061 createdAt: "2026-06-08T10:00:01.000Z",
1062 });
1063 let canRemove = true;
1064
1065 await manager.enqueue(message);
1066 const removal = manager.remove(message, manager.revisionOf(message.messageId), () => canRemove);
1067 await removeStarted.promise;
1068 canRemove = false;
1069 removeBarrier.resolve();
1070
1071 await expect(removal).resolves.toBe(null);
1072 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
1073 "environment-1:thread-1": [message],
1074 });
1075 expect(stored.get(message.messageId)).toEqual(message);
1076 registry.dispose();
1077 });
1078
1079 it("preserves concurrent enqueues while clearing an environment", async () => {
1080 const registry = AtomRegistry.make();
1081 const stored = new Map<MessageId, QueuedThreadMessage>();
1082 const removeStarted = Promise.withResolvers<void>();
1083 const removeBarrier = Promise.withResolvers<void>();
1084 const manager = createThreadOutboxManager({
1085 registry,
1086 storage: {
1087 load: async () => ({ messages: [...stored.values()], errors: [] }),
1088 write: async (message) => {
1089 stored.set(message.messageId, message);
1090 },
1091 remove: async (message) => {
1092 if (message.environmentId === EnvironmentId.make("environment-clear")) {
1093 removeStarted.resolve();
1094 await removeBarrier.promise;
1095 }
1096 stored.delete(message.messageId);
1097 },
1098 },
1099 });
1100 const replaced = queuedMessage({
1101 environmentId: "environment-clear",
1102 messageId: "message-replaced-during-clear",
1103 createdAt: "2026-06-08T10:00:01.000Z",
1104 });
1105 const removed = queuedMessage({
1106 environmentId: "environment-clear",
1107 messageId: "message-removed-by-clear",
1108 createdAt: "2026-06-08T10:00:02.000Z",
1109 });
1110 const kept = queuedMessage({
1111 environmentId: "environment-keep",
1112 messageId: "message-other-environment",
1113 createdAt: "2026-06-08T10:00:03.000Z",
1114 });
1115 const replacement = { ...replaced, text: "replacement" };
1116 const added = queuedMessage({
1117 environmentId: "environment-clear",
1118 messageId: "message-added-during-clear",
1119 createdAt: "2026-06-08T10:00:04.000Z",
1120 });
1121
1122 await Promise.all([manager.enqueue(replaced), manager.enqueue(removed), manager.enqueue(kept)]);
1123 const clearing = manager.clearEnvironment(replaced.environmentId);
1124 await removeStarted.promise;
1125 const replacing = manager.enqueue(replacement);
1126 const adding = manager.enqueue(added);
1127 removeBarrier.resolve();
1128
1129 await expect(clearing).resolves.toEqual([removed]);
1130 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
1131 "environment-clear:thread-1": [replacement, added],
1132 "environment-keep:thread-1": [kept],
1133 });
1134 expect(stored.get(replacement.messageId)).toEqual(replacement);
1135 expect(stored.has(removed.messageId)).toBe(false);
1136
1137 await Promise.all([replacing, adding]);
1138 expect([...stored.values()]).toEqual(expect.arrayContaining([replacement, added, kept]));
1139 registry.dispose();
1140 });
1141
1142 it("does not restore a message removed before a queued environment clear starts", async () => {
1143 const registry = AtomRegistry.make();
1144 const stored = new Map<MessageId, QueuedThreadMessage>();
1145 const removeStarted = Promise.withResolvers<void>();
1146 const removeBarrier = Promise.withResolvers<void>();
1147 let removeCalls = 0;
1148 const manager = createThreadOutboxManager({
1149 registry,
1150 storage: {
1151 load: async () => ({ messages: [...stored.values()], errors: [] }),
1152 write: async (message) => {
1153 stored.set(message.messageId, message);
1154 },
1155 remove: async (message) => {
1156 removeCalls += 1;
1157 if (removeCalls === 1) {
1158 removeStarted.resolve();
1159 await removeBarrier.promise;
1160 }
1161 stored.delete(message.messageId);
1162 },
1163 },
1164 });
1165 const message = queuedMessage({
1166 environmentId: "environment-clear",
1167 messageId: "message-removed-before-clear",
1168 createdAt: "2026-06-08T10:00:01.000Z",
1169 });
1170
1171 await manager.enqueue(message);
1172 const removal = manager.remove(message);
1173 await removeStarted.promise;
1174 const clearing = manager.clearEnvironment(message.environmentId);
1175 removeBarrier.resolve();
1176
1177 await expect(removal).resolves.toEqual(message);
1178 await expect(clearing).resolves.toEqual([]);
1179 expect(removeCalls).toBe(1);
1180 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({});
1181 expect(stored.has(message.messageId)).toBe(false);
1182 registry.dispose();
1183 });
1184
1185 it("keeps an enqueue published while an environment clear waits to start", async () => {
1186 const registry = AtomRegistry.make();
1187 const stored = new Map<MessageId, QueuedThreadMessage>();
1188 const mutationStarted = Promise.withResolvers<void>();
1189 const mutationBarrier = Promise.withResolvers<void>();
1190 let removeCalls = 0;
1191 const manager = createThreadOutboxManager({
1192 registry,
1193 storage: {
1194 load: async () => ({ messages: [...stored.values()], errors: [] }),
1195 write: async (message) => {
1196 stored.set(message.messageId, message);
1197 },
1198 remove: async () => {
1199 removeCalls += 1;
1200 },
1201 },
1202 });
1203 const blocker = manager.serialize(async () => {
1204 mutationStarted.resolve();
1205 await mutationBarrier.promise;
1206 });
1207 await mutationStarted.promise;
1208 const clearing = manager.clearEnvironment(EnvironmentId.make("environment-clear"));
1209 const added = queuedMessage({
1210 environmentId: "environment-clear",
1211 messageId: "message-enqueued-before-clear-start",
1212 createdAt: "2026-06-08T10:00:01.000Z",
1213 });
1214 const enqueue = manager.enqueue(added);
1215 mutationBarrier.resolve();
1216
1217 await blocker;
1218 await expect(clearing).resolves.toEqual([]);
1219 expect(removeCalls).toBe(0);
1220 expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({
1221 "environment-clear:thread-1": [added],
1222 });
1223 await enqueue;
1224 expect(stored.get(added.messageId)).toEqual(added);
1225 registry.dispose();
1226 });
1227
1228 it("removes an already-created pending task before the file-capability gate runs", () => {
1229 // The creation's startTurn already made the thread, so the resolver wants
1230 // the queued message removed. A missing server config (or missing file
1231 // support) must not turn that into a restore, which would duplicate the
1232 // task as a draft.
1233 const fileAttachments = [{ name: "report.pdf", sizeBytes: 42 }];
1234 expect(
1235 resolveThreadOutboxDispatchStep({
1236 deliveryAction: "remove",
1237 fileAttachments,
1238 serverConfig: null,
1239 }),
1240 ).toEqual({ step: "remove" });
1241 expect(
1242 resolveThreadOutboxDispatchStep({
1243 deliveryAction: "remove",
1244 fileAttachments,
1245 serverConfig: { maxFileUploadBytes: undefined },
1246 }),
1247 ).toEqual({ step: "remove" });
1248 });
1249
1250 it("retries instead of parking a file message while the server config loads", () => {
1251 expect(
1252 resolveThreadOutboxDispatchStep({
1253 deliveryAction: "send",
1254 fileAttachments: [{ name: "report.pdf", sizeBytes: 42 }],
1255 serverConfig: null,
1256 }),
1257 ).toEqual({ step: "retry" });
1258 });
1259
1260 it("gates a sending file message on the server's file support and limit", () => {
1261 expect(
1262 resolveThreadOutboxDispatchStep({
1263 deliveryAction: "send",
1264 fileAttachments: [{ name: "report.pdf", sizeBytes: 42 }],
1265 serverConfig: { maxFileUploadBytes: undefined },
1266 }),
1267 ).toEqual({ step: "restore", reason: "This server does not support file attachments." });
1268 expect(
1269 resolveThreadOutboxDispatchStep({
1270 deliveryAction: "send",
1271 fileAttachments: [{ name: "big.zip", sizeBytes: 2 * 1024 * 1024 }],
1272 serverConfig: { maxFileUploadBytes: 1024 * 1024 },
1273 }),
1274 ).toEqual({ step: "restore", reason: "'big.zip' exceeds the 1 MB attachment limit." });
1275 expect(
1276 resolveThreadOutboxDispatchStep({
1277 deliveryAction: "send",
1278 fileAttachments: [{ name: "report.pdf", sizeBytes: 42 }],
1279 serverConfig: { maxFileUploadBytes: 1024 * 1024 },
1280 }),
1281 ).toEqual({ step: "send" });
1282 });
1283
1284 it("waits for provider capabilities before sending text-only queued messages", () => {
1285 expect(
1286 resolveThreadOutboxDispatchStep({
1287 deliveryAction: "send",
1288 fileAttachments: [],
1289 serverConfig: null,
1290 }),
1291 ).toEqual({ step: "retry" });
1292 expect(
1293 resolveThreadOutboxDispatchStep({
1294 deliveryAction: "send",
1295 fileAttachments: [],
1296 serverConfig: { maxFileUploadBytes: undefined },
1297 }),
1298 ).toEqual({ step: "send" });
1299 });
1300
1301 it("only removes a missing-thread message after shell synchronization is live", () => {
1302 expect(
1303 resolveThreadOutboxDeliveryAction({
1304 isCreation: false,
1305 threadExists: false,
1306 shellStatus: "synchronizing",
1307 environmentConnected: true,
1308 threadBusy: false,
1309 }),
1310 ).toBe("wait");
1311 expect(
1312 resolveThreadOutboxDeliveryAction({
1313 isCreation: false,
1314 threadExists: false,
1315 shellStatus: "live",
1316 environmentConnected: true,
1317 threadBusy: false,
1318 }),
1319 ).toBe("remove");
1320 expect(
1321 resolveThreadOutboxDeliveryAction({
1322 isCreation: false,
1323 threadExists: true,
1324 shellStatus: "live",
1325 environmentConnected: true,
1326 threadBusy: false,
1327 }),
1328 ).toBe("send");
1329 });
1330
1331 it("sends existing-thread messages whenever connected so queued messages can steer", () => {
1332 expect(
1333 resolveThreadOutboxDeliveryAction({
1334 isCreation: false,
1335 threadExists: true,
1336 shellStatus: "live",
1337 environmentConnected: true,
1338 threadBusy: true,
1339 }),
1340 ).toBe("send");
1341 expect(
1342 resolveThreadOutboxDeliveryAction({
1343 isCreation: false,
1344 threadExists: true,
1345 shellStatus: "live",
1346 environmentConnected: false,
1347 threadBusy: true,
1348 }),
1349 ).toBe("wait");
1350 });
1351
1352 it("sends queued creations once connected and live, removing already-created ones", () => {
1353 expect(
1354 resolveThreadOutboxDeliveryAction({
1355 isCreation: true,
1356 threadExists: false,
1357 shellStatus: "cached",
1358 environmentConnected: false,
1359 threadBusy: false,
1360 }),
1361 ).toBe("wait");
1362 // Connected but not yet synchronized: a previously delivered creation may
1363 // simply not be visible yet — sending now could duplicate the thread.
1364 expect(
1365 resolveThreadOutboxDeliveryAction({
1366 isCreation: true,
1367 threadExists: false,
1368 shellStatus: "synchronizing",
1369 environmentConnected: true,
1370 threadBusy: false,
1371 }),
1372 ).toBe("wait");
1373 expect(
1374 resolveThreadOutboxDeliveryAction({
1375 isCreation: true,
1376 threadExists: false,
1377 shellStatus: "live",
1378 environmentConnected: true,
1379 threadBusy: false,
1380 }),
1381 ).toBe("send");
1382 expect(
1383 resolveThreadOutboxDeliveryAction({
1384 isCreation: true,
1385 threadExists: true,
1386 shellStatus: "live",
1387 environmentConnected: true,
1388 threadBusy: true,
1389 }),
1390 ).toBe("remove");
1391 });
1392
1393 it("round-trips queued PR checkout sources and gates incomplete creations from sending", () => {
1394 const base = queuedMessage({
1395 messageId: "message-1",
1396 createdAt: "2026-06-08T10:00:01.000Z",
1397 });
1398 const creationMessage = {
1399 ...base,
1400 modelSelection: {
1401 instanceId: ProviderInstanceId.make("codex"),
1402 model: "gpt-5.4",
1403 },
1404 creation: {
1405 projectId: ProjectId.make("project-1"),
1406 workspaceMode: "worktree",
1407 branch: "fork/feature",
1408 worktreePath: null,
1409 startFromOrigin: false,
1410 pullRequestReference: "https://github.com/acme/repo/pull/123",
1411 },
1412 } satisfies QueuedThreadMessage;
1413
1414 expect(decodeQueuedThreadMessage(encodeQueuedThreadMessage(creationMessage))).toEqual(
1415 creationMessage,
1416 );
1417 expect(isQueuedThreadCreationSendable(creationMessage)).toBe(true);
1418 expect(
1419 isQueuedThreadCreationSendable({
1420 ...creationMessage,
1421 creation: { ...creationMessage.creation, branch: null },
1422 }),
1423 ).toBe(false);
1424 expect(
1425 isQueuedThreadCreationSendable({
1426 ...creationMessage,
1427 creation: { ...creationMessage.creation, branch: "" },
1428 }),
1429 ).toBe(false);
1430 expect(isQueuedThreadCreationSendable({ ...creationMessage, modelSelection: undefined })).toBe(
1431 false,
1432 );
1433 expect(isQueuedThreadCreationSendable(base)).toBe(false);
1434 expect(isQueuedThreadCreationSendable({ ...creationMessage, text: " " })).toBe(false);
1435 expect(
1436 isQueuedThreadCreationSendable({
1437 ...creationMessage,
1438 text: "",
1439 attachments: [
1440 {
1441 id: "image-1",
1442 type: "image",
1443 name: "photo.png",
1444 mimeType: "image/png",
1445 sizeBytes: 3,
1446 fileUri: "file:///documents/t3-composer-attachments/photo.png",
1447 previewUri: "file:///documents/t3-composer-attachments/photo.png",
1448 uploadedAttachmentId: "pending-photo-png",
1449 uploadEnvironmentId: EnvironmentId.make("environment-1"),
1450 },
1451 ],
1452 }),
1453 ).toBe(true);
1454 });
1455
1456 it("retries transport failures but drops deterministic command failures", () => {
1457 expect(shouldRetryThreadOutboxDelivery(new Error("Socket is not connected"))).toBe(true);
1458 expect(
1459 shouldRetryThreadOutboxDelivery({
1460 _tag: "ConnectionTransientError",
1461 message: "temporarily unavailable",
1462 }),
1463 ).toBe(true);
1464 expect(shouldRetryThreadOutboxDelivery(new Error("Thread no longer exists"))).toBe(false);
1465 expect(
1466 shouldRetryThreadOutboxDelivery(
1467 new OrchestrationDispatchCommandError({ message: "Thread no longer exists" }),
1468 ),
1469 ).toBe(false);
1470 expect(
1471 shouldRetryThreadOutboxDelivery(
1472 new EnvironmentAuthorizationError({
1473 message: "Missing scope",
1474 requiredScope: "orchestration:operate",
1475 }),
1476 ),
1477 ).toBe(false);
1478 });
1479
1480 // A pending task created offline drains the moment the phone reconnects,
1481 // which is exactly when the socket is most likely to drop again. Every way a
1482 // request can fail in flight must retry; a restore turns the pending task
1483 // into a draft and it disappears from the list.
1484 it("retries every in-flight transport failure by tag, not by message text", () => {
1485 const socketReasons = [
1486 new Socket.SocketReadError({ cause: new Error("The network connection was lost.") }),
1487 new Socket.SocketWriteError({ cause: new Error("Broken pipe") }),
1488 new Socket.SocketCloseError({ code: 1006 }),
1489 new Socket.SocketOpenError({ kind: "Timeout", cause: new Error("timeout") }),
1490 ];
1491 for (const reason of socketReasons) {
1492 const error = new RpcClientError.RpcClientError({ reason });
1493 expect(isTransportConnectionErrorMessage(error.message)).toBe(
1494 reason._tag === "SocketCloseError" || reason._tag === "SocketOpenError",
1495 );
1496 expect(shouldRetryThreadOutboxDelivery(error)).toBe(true);
1497 }
1498 expect(
1499 shouldRetryThreadOutboxDelivery(
1500 new RpcClientError.RpcClientError({
1501 reason: new RpcClientError.RpcClientDefect({
1502 message: "Error decoding message",
1503 cause: new Error("Unexpected end of JSON input"),
1504 }),
1505 }),
1506 ),
1507 ).toBe(true);
1508 expect(
1509 shouldRetryThreadOutboxDelivery(
1510 new EnvironmentRpcUnavailableError({
1511 environmentId: "environment-1",
1512 message: "Home is not connected.",
1513 }),
1514 ),
1515 ).toBe(true);
1516 expect(
1517 shouldRetryThreadOutboxDelivery(
1518 new EnvironmentNotRegisteredError({ environmentId: EnvironmentId.make("environment-1") }),
1519 ),
1520 ).toBe(true);
1521 });
1522
1523 it("retains queued messages when settings synchronization fails before startTurn", () => {
1524 const deterministicFailure = new Error("Thread no longer exists");
1525
1526 expect(
1527 resolveThreadOutboxFailureAction({
1528 stage: "settings-sync",
1529 error: deterministicFailure,
1530 interrupted: false,
1531 }),
1532 ).toBe("retry");
1533 expect(
1534 resolveThreadOutboxFailureAction({
1535 stage: "start-turn",
1536 error: deterministicFailure,
1537 interrupted: false,
1538 }),
1539 ).toBe("restore");
1540 });
1541});