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