1import { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
2import { beforeEach, describe, expect, it, vi } from "vite-plus/test";
3
4import type { QueuedThreadMessage } from "./thread-outbox-model";
5
6const harness = vi.hoisted(() => ({
7 manager: null as unknown as ReturnType<
8 typeof import("./thread-outbox-manager").createThreadOutboxManager
9 >,
10 writeGates: [] as Array<{
11 readonly promise: Promise<void>;
12 readonly started: (message: QueuedThreadMessage) => void;
13 }>,
14}));
15
16vi.mock("./thread-outbox", async () => {
17 const { createThreadOutboxManager } = await import("./thread-outbox-manager");
18 const { appAtomRegistry } = await import("./atom-registry");
19 harness.manager = createThreadOutboxManager({
20 registry: appAtomRegistry,
21 storage: {
22 load: async () => ({ messages: [], errors: [] }),
23 write: async (message) => {
24 const pending = harness.writeGates.shift();
25 if (pending) {
26 pending.started(message);
27 await pending.promise;
28 }
29 },
30 remove: async () => undefined,
31 },
32 });
33 const manager = harness.manager;
34 return {
35 threadOutboxManager: manager,
36 flushThreadOutbox: async () => undefined,
37 threadOutboxRevision: (messageId: QueuedThreadMessage["messageId"]) =>
38 manager.revisionOf(messageId),
39 updateThreadOutboxMessage: (message: QueuedThreadMessage, expectedRevision?: number) =>
40 manager.update(message, expectedRevision),
41 };
42});
43
44import { appAtomRegistry } from "./atom-registry";
45import {
46 capturePendingTaskEditorWriteBaseline,
47 flushPendingTaskEditorWrite,
48} from "./pending-task-editor-writes";
49import {
50 composerDraftsAtom,
51 getComposerDraftSnapshot,
52 type ComposerDraft,
53} from "./use-composer-drafts";
54
55function queuedMessage(messageId: string, text: string): QueuedThreadMessage {
56 return {
57 environmentId: EnvironmentId.make("environment-1"),
58 threadId: ThreadId.make("thread-1"),
59 messageId: MessageId.make(messageId),
60 commandId: CommandId.make(`command-${messageId}`),
61 text,
62 attachments: [],
63 createdAt: "2026-08-28T12:00:00.000Z",
64 };
65}
66
67function draft(text: string): ComposerDraft {
68 return {
69 text,
70 attachments: [],
71 runtimeMode: "full-access",
72 };
73}
74
75function setDraft(draftKey: string, value: ComposerDraft): void {
76 appAtomRegistry.set(composerDraftsAtom, { [draftKey]: value });
77}
78
79function queuedMessageText(messageId: QueuedThreadMessage["messageId"]): string | null {
80 const messages = Object.values(
81 appAtomRegistry.get(harness.manager.queuedMessagesByThreadKeyAtom),
82 ).flat();
83 return messages.find((message) => message.messageId === messageId)?.text ?? null;
84}
85
86function blockNextWrite() {
87 let resolveWrite!: () => void;
88 let rejectWrite!: (error: Error) => void;
89 let markStarted!: (message: QueuedThreadMessage) => void;
90 const promise = new Promise<void>((resolve, reject) => {
91 resolveWrite = resolve;
92 rejectWrite = reject;
93 });
94 const started = new Promise<QueuedThreadMessage>((resolve) => {
95 markStarted = resolve;
96 });
97 harness.writeGates.push({ promise, started: markStarted });
98 return {
99 started,
100 resolve: resolveWrite,
101 reject: rejectWrite,
102 };
103}
104
105beforeEach(() => {
106 harness.writeGates.length = 0;
107 appAtomRegistry.set(harness.manager.queuedMessagesByThreadKeyAtom, {});
108 appAtomRegistry.set(composerDraftsAtom, {});
109});
110
111describe("pending task editor writes", () => {
112 it("chains a reopened editor that closes before the previous save finishes", async () => {
113 const original = queuedMessage("message-close-before-save", "original");
114 const firstEdit = queuedMessage("message-close-before-save", "first edit");
115 const secondEdit = queuedMessage("message-close-before-save", "second edit");
116 const draftKey = "pending-task:message-close-before-save";
117 await harness.manager.enqueue(original);
118
119 setDraft(draftKey, draft(firstEdit.text));
120 const firstBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
121 const firstWriteGate = blockNextWrite();
122 const firstSave = flushPendingTaskEditorWrite({
123 message: firstEdit,
124 baseline: firstBaseline,
125 draftKey,
126 });
127 await firstWriteGate.started;
128
129 const secondBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
130 setDraft(draftKey, draft(secondEdit.text));
131 const secondWriteGate = blockNextWrite();
132 const secondSave = flushPendingTaskEditorWrite({
133 message: secondEdit,
134 baseline: secondBaseline,
135 draftKey,
136 });
137
138 firstWriteGate.resolve();
139 await expect(firstSave).resolves.toBe(false);
140 await expect(secondWriteGate.started).resolves.toMatchObject({ text: secondEdit.text });
141 secondWriteGate.resolve();
142
143 await expect(secondSave).resolves.toBe(true);
144 expect(queuedMessageText(original.messageId)).toBe(secondEdit.text);
145 });
146
147 it("keeps a captured predecessor after that predecessor finishes", async () => {
148 const original = queuedMessage("message-finished-predecessor", "original");
149 const firstEdit = queuedMessage("message-finished-predecessor", "first edit");
150 const secondEdit = queuedMessage("message-finished-predecessor", "second edit");
151 const draftKey = "pending-task:message-finished-predecessor";
152 await harness.manager.enqueue(original);
153
154 setDraft(draftKey, draft(firstEdit.text));
155 const firstWriteGate = blockNextWrite();
156 const firstSave = flushPendingTaskEditorWrite({
157 message: firstEdit,
158 baseline: capturePendingTaskEditorWriteBaseline(original.messageId),
159 draftKey,
160 });
161 await firstWriteGate.started;
162 const secondBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
163
164 firstWriteGate.resolve();
165 await expect(firstSave).resolves.toBe(true);
166
167 setDraft(draftKey, draft(secondEdit.text));
168 const secondWriteGate = blockNextWrite();
169 const secondSave = flushPendingTaskEditorWrite({
170 message: secondEdit,
171 baseline: secondBaseline,
172 draftKey,
173 });
174 await secondWriteGate.started;
175 secondWriteGate.resolve();
176
177 await expect(secondSave).resolves.toBe(true);
178 expect(queuedMessageText(original.messageId)).toBe(secondEdit.text);
179 });
180
181 it("chains three rapid editor saves in order", async () => {
182 const original = queuedMessage("message-three-saves", "original");
183 const firstEdit = queuedMessage("message-three-saves", "first edit");
184 const secondEdit = queuedMessage("message-three-saves", "second edit");
185 const thirdEdit = queuedMessage("message-three-saves", "third edit");
186 const draftKey = "pending-task:message-three-saves";
187 await harness.manager.enqueue(original);
188
189 setDraft(draftKey, draft(firstEdit.text));
190 const firstWriteGate = blockNextWrite();
191 const firstSave = flushPendingTaskEditorWrite({
192 message: firstEdit,
193 baseline: capturePendingTaskEditorWriteBaseline(original.messageId),
194 draftKey,
195 });
196 await firstWriteGate.started;
197
198 const secondBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
199 setDraft(draftKey, draft(secondEdit.text));
200 const secondWriteGate = blockNextWrite();
201 const secondSave = flushPendingTaskEditorWrite({
202 message: secondEdit,
203 baseline: secondBaseline,
204 draftKey,
205 });
206
207 const thirdBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
208 setDraft(draftKey, draft(thirdEdit.text));
209 const thirdWriteGate = blockNextWrite();
210 const thirdSave = flushPendingTaskEditorWrite({
211 message: thirdEdit,
212 baseline: thirdBaseline,
213 draftKey,
214 });
215
216 firstWriteGate.resolve();
217 await expect(firstSave).resolves.toBe(false);
218 await secondWriteGate.started;
219 secondWriteGate.resolve();
220 await expect(secondSave).resolves.toBe(false);
221 await thirdWriteGate.started;
222 thirdWriteGate.resolve();
223
224 await expect(thirdSave).resolves.toBe(true);
225 expect(queuedMessageText(original.messageId)).toBe(thirdEdit.text);
226 });
227
228 it("keeps the handed-off revision when the middle editor write fails", async () => {
229 const original = queuedMessage("message-middle-failure", "original");
230 const firstEdit = queuedMessage("message-middle-failure", "first edit");
231 const failedEdit = queuedMessage("message-middle-failure", "failed edit");
232 const finalEdit = queuedMessage("message-middle-failure", "final edit");
233 const draftKey = "pending-task:message-middle-failure";
234 await harness.manager.enqueue(original);
235
236 setDraft(draftKey, draft(firstEdit.text));
237 const firstWriteGate = blockNextWrite();
238 const firstSave = flushPendingTaskEditorWrite({
239 message: firstEdit,
240 baseline: capturePendingTaskEditorWriteBaseline(original.messageId),
241 draftKey,
242 });
243 await firstWriteGate.started;
244
245 const failedBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
246 setDraft(draftKey, draft(failedEdit.text));
247 const failedWriteGate = blockNextWrite();
248 const failedSave = flushPendingTaskEditorWrite({
249 message: failedEdit,
250 baseline: failedBaseline,
251 draftKey,
252 });
253
254 const finalBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
255 setDraft(draftKey, draft(finalEdit.text));
256 const finalWriteGate = blockNextWrite();
257 const finalSave = flushPendingTaskEditorWrite({
258 message: finalEdit,
259 baseline: finalBaseline,
260 draftKey,
261 });
262
263 firstWriteGate.resolve();
264 await expect(firstSave).resolves.toBe(false);
265 await failedWriteGate.started;
266 failedWriteGate.reject(new Error("disk full"));
267 await expect(failedSave).rejects.toMatchObject({ _tag: "ThreadOutboxManagerError" });
268 await finalWriteGate.started;
269 finalWriteGate.resolve();
270
271 await expect(finalSave).resolves.toBe(true);
272 expect(queuedMessageText(original.messageId)).toBe(finalEdit.text);
273 });
274
275 it("does not overwrite an unrelated update accepted after capture", async () => {
276 const original = queuedMessage("message-unrelated-update", "original");
277 const editorEdit = queuedMessage("message-unrelated-update", "editor edit");
278 const unrelatedEdit = queuedMessage("message-unrelated-update", "unrelated edit");
279 const draftKey = "pending-task:message-unrelated-update";
280 await harness.manager.enqueue(original);
281
282 const editorBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
283 const revision = harness.manager.revisionOf(original.messageId);
284 await expect(harness.manager.update(unrelatedEdit, revision)).resolves.toBe(true);
285 setDraft(draftKey, draft(editorEdit.text));
286
287 await expect(
288 flushPendingTaskEditorWrite({
289 message: editorEdit,
290 baseline: editorBaseline,
291 draftKey,
292 }),
293 ).resolves.toBe(false);
294 expect(queuedMessageText(original.messageId)).toBe(unrelatedEdit.text);
295 });
296
297 it("lets a later editor retry after its predecessor write fails", async () => {
298 const original = queuedMessage("message-write-retry", "original");
299 const failedEdit = queuedMessage("message-write-retry", "failed edit");
300 const retryEdit = queuedMessage("message-write-retry", "retry edit");
301 const draftKey = "pending-task:message-write-retry";
302 await harness.manager.enqueue(original);
303
304 setDraft(draftKey, draft(failedEdit.text));
305 const failedWriteGate = blockNextWrite();
306 const failedSave = flushPendingTaskEditorWrite({
307 message: failedEdit,
308 baseline: capturePendingTaskEditorWriteBaseline(original.messageId),
309 draftKey,
310 });
311 await failedWriteGate.started;
312
313 const retryBaseline = capturePendingTaskEditorWriteBaseline(original.messageId);
314 setDraft(draftKey, draft(retryEdit.text));
315 const retryWriteGate = blockNextWrite();
316 const retrySave = flushPendingTaskEditorWrite({
317 message: retryEdit,
318 baseline: retryBaseline,
319 draftKey,
320 });
321
322 failedWriteGate.reject(new Error("disk full"));
323 await expect(failedSave).rejects.toMatchObject({ _tag: "ThreadOutboxManagerError" });
324 await retryWriteGate.started;
325 retryWriteGate.resolve();
326
327 await expect(retrySave).resolves.toBe(true);
328 expect(queuedMessageText(original.messageId)).toBe(retryEdit.text);
329 });
330
331 it("does not permit cleanup after a newer editor makes the draft unsendable", async () => {
332 const original = queuedMessage("message-unsendable-draft", "original");
333 const editorEdit = queuedMessage("message-unsendable-draft", "saved edit");
334 const draftKey = "pending-task:message-unsendable-draft";
335 await harness.manager.enqueue(original);
336
337 setDraft(draftKey, draft(editorEdit.text));
338 const writeGate = blockNextWrite();
339 const save = flushPendingTaskEditorWrite({
340 message: editorEdit,
341 baseline: capturePendingTaskEditorWriteBaseline(original.messageId),
342 draftKey,
343 });
344 await writeGate.started;
345
346 setDraft(draftKey, draft(""));
347 writeGate.resolve();
348
349 await expect(save).resolves.toBe(false);
350 expect(getComposerDraftSnapshot(draftKey).text).toBe("");
351 expect(queuedMessageText(original.messageId)).toBe(editorEdit.text);
352 });
353});