1import {
2 AuthOrchestrationOperateScope,
3 CommandId,
4 EnvironmentId,
5 MessageId,
6 ProjectId,
7 ProviderInstanceId,
8 ThreadId,
9} from "@t3tools/contracts";
10import * as Cause from "effect/Cause";
11import { AsyncResult, type Atom } from "effect/reactivity";
12import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test";
13import type { QueuedThreadMessage } from "./thread-outbox-model";
14
15const state = vi.hoisted(() => ({
16 draftText: "Keep this draft separate",
17 cleanups: [] as Array<() => void>,
18 delivery: null as {
19 readonly drain: () => void;
20 readonly dispose: () => void;
21 } | null,
22 start: vi.fn(),
23 metadata: vi.fn(),
24 runtime: vi.fn(),
25 interaction: vi.fn(),
26 prepare: vi.fn(),
27 checkout: vi.fn(),
28 modelUnavailable: vi.fn(() => false),
29 familyMount: vi.fn(() => () => {}),
30 threadShells: null as unknown as Atom.Writable<ReadonlyArray<unknown>>,
31 manager: null as unknown as ReturnType<
32 typeof import("./thread-outbox-manager").createThreadOutboxManager
33 >,
34 thread: {
35 environmentId: "secondary",
36 id: "thread",
37 modelSelection: { instanceId: "codex", model: "gpt-5.4" },
38 runtimeMode: "full-access",
39 interactionMode: "default",
40 session: null,
41 },
42 config: {
43 shellThreadFamilies: false,
44 providers: [{ instanceId: "codex", driver: "codex", models: [] }],
45 environment: { capabilities: {} },
46 },
47}));
48vi.mock("react-native", () => ({ Alert: { alert: vi.fn() } }));
49vi.mock("@t3tools/client-runtime/state/runtime", async (importOriginal) => ({
50 ...(await importOriginal<typeof import("@t3tools/client-runtime/state/runtime")>()),
51 runAtomCommand: (_registry: unknown, command: (input: unknown) => unknown, input: unknown) =>
52 command(input),
53}));
54vi.mock("./session", async () => {
55 const { Atom } = await import("effect/reactivity");
56 const { appAtomRegistry } = await import("./atom-registry");
57 // A session's state is reduced to whether it grants the operate scope.
58 const sessionStateAtom = Atom.family((_id: string) => Atom.make(false).pipe(Atom.keepAlive));
59 const sessionHasScope = (granted: boolean, scope: string) =>
60 granted && scope === AuthOrchestrationOperateScope;
61 return {
62 environmentSession: { sessionStateAtom },
63 sessionHasScope,
64 readEnvironmentScope: (environmentId: string, scope: string) =>
65 sessionHasScope(appAtomRegistry.get(sessionStateAtom(environmentId)), scope),
66 };
67});
68vi.mock("./projects", async () => {
69 const { Atom } = await import("effect/reactivity");
70 return { environmentProjects: { projectsAtom: Atom.make([]) } };
71});
72vi.mock("./workspace", async () => {
73 const { Atom } = await import("effect/reactivity");
74 return {
75 workspaceConnections: {
76 environmentsAtom: Atom.make([{ environmentId: "secondary", connectionState: "connected" }]),
77 },
78 };
79});
80vi.mock("./server", async () => {
81 const { Atom } = await import("effect/reactivity");
82 return {
83 serverEnvironment: { configValueAtom: Atom.family((_id: string) => Atom.make(state.config)) },
84 environmentServerConfigsAtom: Atom.make(new Map([["secondary", state.config]])),
85 };
86});
87vi.mock("./threads", async () => {
88 const { Atom, AsyncResult } = await import("effect/reactivity");
89 state.threadShells = Atom.make<ReadonlyArray<unknown>>([state.thread]).pipe(Atom.keepAlive);
90 return {
91 threadEnvironment: {
92 startTurn: state.start,
93 updateMetadata: state.metadata,
94 setRuntimeMode: state.runtime,
95 setInteractionMode: state.interaction,
96 },
97 environmentThreadShells: { threadShellsAtom: state.threadShells },
98 threadFamilies: {
99 mount: state.familyMount,
100 familyAtom: () => Atom.make(AsyncResult.success({})),
101 refresh: vi.fn(),
102 },
103 environmentThreadDetails: { stateAtom: () => Atom.make(null) },
104 };
105});
106vi.mock("./git", () => ({ gitEnvironment: { preparePullRequestThread: state.checkout } }));
107vi.mock("../lib/modelOptions", () => ({ isModelSelectionUnavailable: state.modelUnavailable }));
108vi.mock("../lib/uuid", () => ({ uuidv4: () => "uuid", randomHex: () => "abcd" }));
109vi.mock("../lib/attachmentUpload", () => ({ prepareTurnAttachments: state.prepare }));
110vi.mock("./thread-outbox", async () => {
111 const { createThreadOutboxManager } = await import("./thread-outbox-manager");
112 const { appAtomRegistry } = await import("./atom-registry");
113 state.manager = createThreadOutboxManager({
114 registry: appAtomRegistry,
115 storage: {
116 load: async () => ({ messages: [], errors: [] }),
117 write: async () => {},
118 remove: async () => {},
119 },
120 });
121 return {
122 threadOutboxManager: state.manager,
123 threadOutboxRevision: (id: MessageId) => state.manager.revisionOf(id),
124 confirmThreadOutboxMessageQueued: (message: QueuedThreadMessage) =>
125 state.manager.confirmQueued(message),
126 updateThreadOutboxMessage: (message: QueuedThreadMessage, revision?: number) =>
127 state.manager.update(message, revision),
128 };
129});
130vi.mock("./thread-outbox-removal", () => ({
131 removeThreadOutboxMessage: (message: QueuedThreadMessage, revision?: number) =>
132 state.manager.remove(message, revision),
133}));
134vi.mock("./use-thread-outbox", async () => {
135 const { Atom } = await import("effect/reactivity");
136 return {
137 editingQueuedMessageIdsAtom: Atom.make({}).pipe(Atom.keepAlive),
138 dispatchingQueuedMessageIdAtom: Atom.make<MessageId | null>(null).pipe(Atom.keepAlive),
139 retryingQueuedMessageIdsAtom: Atom.make({}).pipe(Atom.keepAlive),
140 threadOutboxShellStatusesAtom: Atom.make(new Map([["secondary", "live"]])),
141 };
142});
143vi.mock("./use-composer-drafts", async () => {
144 const { Atom } = await import("effect/reactivity");
145 return {
146 composerDraftsAtom: Atom.make({}),
147 removeDeliveredCloudQueuedMessage: async () => {},
148 appendComposerDraftAttachments: vi.fn(),
149 flushComposerDrafts: vi.fn(),
150 getComposerDraftSnapshot: vi.fn(() => ({ text: state.draftText, attachments: [] })),
151 mergeComposerDraftContent: vi.fn(async (_key: string, input: { text: string }) => {
152 state.draftText += `\n${input.text}`;
153 }),
154 replaceComposerDraftAttachments: vi.fn(),
155 setFreshSessionPending: vi.fn(),
156 undoComposerDraftMerge: vi.fn(),
157 updateComposerDraftSettings: vi.fn(),
158 waitForComposerDraftsLoaded: vi.fn(),
159 };
160});
161
162import { appAtomRegistry } from "./atom-registry";
163import { environmentSession } from "./session";
164import { dispatchingQueuedMessageIdAtom, editingQueuedMessageIdsAtom } from "./use-thread-outbox";
165import { createThreadOutboxDelivery } from "./threadOutboxDelivery";
166
167const message = (overrides: Partial<QueuedThreadMessage> = {}): QueuedThreadMessage => ({
168 environmentId: EnvironmentId.make("secondary"),
169 threadId: ThreadId.make("thread"),
170 commandId: CommandId.make("command"),
171 messageId: MessageId.make("message"),
172 text: "Keep this queued",
173 attachments: [],
174 createdAt: "2026-09-05T00:00:00.000Z",
175 ...overrides,
176});
177const remaining = () =>
178 Object.values(appAtomRegistry.get(state.manager.queuedMessagesByThreadKeyAtom)).flat();
179
180function setOperateGrant(environmentId: string, granted: boolean): void {
181 // The mocked session atom holds the grant itself (see the ./session mock).
182 const sessionAtom = environmentSession.sessionStateAtom(EnvironmentId.make(environmentId));
183 appAtomRegistry.set(sessionAtom as unknown as Atom.Writable<boolean>, granted);
184}
185
186function runDrain() {
187 const settled = Promise.withResolvers<void>();
188 const release = appAtomRegistry.subscribe(dispatchingQueuedMessageIdAtom, (id) => {
189 if (id === null) settled.resolve();
190 });
191 startDelivery();
192 return settled.promise.finally(release);
193}
194
195/** Starts the service once per test; later calls run another drain pass. */
196function startDelivery(): void {
197 if (state.delivery !== null) {
198 state.delivery.drain();
199 return;
200 }
201 const delivery = createThreadOutboxDelivery();
202 state.delivery = delivery;
203 state.cleanups.push(() => {
204 delivery.dispose();
205 state.delivery = null;
206 });
207}
208
209beforeEach(() => {
210 setOperateGrant("primary", true);
211 setOperateGrant("secondary", false);
212 state.draftText = "Keep this draft separate";
213 state.config.shellThreadFamilies = false;
214 state.modelUnavailable.mockReset().mockReturnValue(false);
215 state.familyMount.mockReset().mockReturnValue(() => {});
216 appAtomRegistry.set(state.threadShells, [state.thread]);
217 appAtomRegistry.set(state.manager.queuedMessagesByThreadKeyAtom, {});
218 appAtomRegistry.set(dispatchingQueuedMessageIdAtom, null);
219 appAtomRegistry.set(editingQueuedMessageIdsAtom, {});
220 for (const command of [
221 state.start,
222 state.metadata,
223 state.runtime,
224 state.interaction,
225 state.checkout,
226 ]) {
227 command.mockReset().mockResolvedValue(AsyncResult.success(undefined));
228 }
229 state.start.mockResolvedValue(AsyncResult.success({}));
230 state.prepare.mockReset().mockImplementation(async ({ attachments }) => ({
231 status: "ready",
232 attachments: [],
233 draftAttachments: attachments,
234 pendingAttachmentIds: [],
235 }));
236});
237
238describe("queued PR checkout", () => {
239 const queuedPullRequest = () =>
240 message({
241 threadId: ThreadId.make("new-pr-thread"),
242 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
243 creation: {
244 projectId: ProjectId.make("project"),
245 projectCwd: "/repo",
246 workspaceMode: "worktree",
247 branch: "fork/feature",
248 worktreePath: null,
249 pullRequestReference: "https://github.com/acme/repo/pull/123",
250 },
251 });
252 const preparedCheckout = {
253 branch: "pr-123",
254 worktreePath: "/repo/.t3/worktrees/pr-123",
255 isOnPullRequestHead: true,
256 };
257
258 it("waits for the provider's PR checkout and launches in that worktree", async () => {
259 setOperateGrant("secondary", true);
260 await state.manager.enqueue(queuedPullRequest());
261 const started = Promise.withResolvers<void>();
262 const checkout =
263 Promise.withResolvers<AsyncResult.AsyncResult<typeof preparedCheckout, Error>>();
264 state.checkout.mockImplementationOnce(() => {
265 started.resolve();
266 return checkout.promise;
267 });
268 const drained = runDrain();
269 await started.promise;
270 expect(state.start).not.toHaveBeenCalled();
271 expect(state.checkout).toHaveBeenCalledWith({
272 environmentId: "secondary",
273 input: { cwd: "/repo", reference: "https://github.com/acme/repo/pull/123", mode: "worktree" },
274 });
275 checkout.resolve(AsyncResult.success(preparedCheckout));
276 await drained;
277 expect(state.start).toHaveBeenCalledOnce();
278 const bootstrap = state.start.mock.calls[0]?.[0].input.bootstrap;
279 expect(bootstrap.createThread).toMatchObject({
280 branch: "pr-123",
281 worktreePath: preparedCheckout.worktreePath,
282 });
283 expect(bootstrap.prepareWorktree).toBeUndefined();
284 expect(remaining()).toEqual([]);
285 });
286
287 it("keeps the PR source queued on a transport failure without starting the agent", async () => {
288 setOperateGrant("secondary", true);
289 const queued = queuedPullRequest();
290 await state.manager.enqueue(queued);
291 state.checkout.mockResolvedValueOnce(
292 AsyncResult.failure(Cause.fail(new Error("Socket is not connected"))),
293 );
294 await runDrain();
295 expect(state.start).not.toHaveBeenCalled();
296 expect(remaining()).toEqual([queued]);
297 });
298
299 it("stops launch if operation access is revoked while checking out the PR", async () => {
300 setOperateGrant("secondary", true);
301 const queued = queuedPullRequest();
302 await state.manager.enqueue(queued);
303 state.checkout.mockImplementationOnce(async () => {
304 setOperateGrant("secondary", false);
305 return AsyncResult.success(preparedCheckout);
306 });
307 await runDrain();
308 expect(state.start).not.toHaveBeenCalled();
309 expect(remaining()).toEqual([queued]);
310 });
311
312 it("does not launch a stale task edited while the PR checkout was running", async () => {
313 setOperateGrant("secondary", true);
314 const queued = queuedPullRequest();
315 await state.manager.enqueue(queued);
316 const edited = { ...queued, text: "Updated instructions" };
317 state.checkout.mockImplementationOnce(async () => {
318 await state.manager.update(edited);
319 return AsyncResult.success(preparedCheckout);
320 });
321 await runDrain();
322 expect(state.start).not.toHaveBeenCalled();
323 expect(remaining()).toEqual([edited]);
324 });
325});
326afterEach(() => {
327 state.cleanups.splice(0).forEach((cleanup) => cleanup());
328});
329
330describe("queued task operation access", () => {
331 it("parks a connected read-only target, then delivers its unchanged queue after a grant", async () => {
332 const queued = message();
333 await state.manager.enqueue(queued);
334 startDelivery();
335 expect(state.start).not.toHaveBeenCalled();
336 expect(state.prepare).not.toHaveBeenCalled();
337 expect(remaining()).toEqual([queued]);
338 setOperateGrant("secondary", true);
339 await runDrain();
340 expect(state.start).toHaveBeenCalledOnce();
341 expect(remaining()).toEqual([]);
342 });
343
344 it("backs off instead of redelivering at once when a delivery hits a defect", async () => {
345 setOperateGrant("secondary", true);
346 const queued = message();
347 await state.manager.enqueue(queued);
348 // Awaited steps catch their own failures, so a defect in the delivery's own code is what
349 // rejects a delivery.
350 state.modelUnavailable.mockImplementationOnce(() => {
351 throw new TypeError("Cannot read properties of undefined");
352 });
353 await runDrain();
354 expect(state.modelUnavailable).toHaveBeenCalledOnce();
355 expect(state.prepare).not.toHaveBeenCalled();
356 expect(state.start).not.toHaveBeenCalled();
357 expect(appAtomRegistry.get(dispatchingQueuedMessageIdAtom)).toBeNull();
358 expect(remaining()).toEqual([queued]);
359 });
360
361 it("starts one delivery when mounting a thread family changes the rows mid-pass", async () => {
362 setOperateGrant("secondary", true);
363 state.config.shellThreadFamilies = true;
364 const subAgentThread = { ...state.thread, id: "sub-agent" };
365 // A cached family merges its rows into the shell list as soon as it is mounted.
366 state.familyMount.mockImplementationOnce(() => {
367 appAtomRegistry.set(state.threadShells, [state.thread, subAgentThread]);
368 return () => {};
369 });
370 const queued = message({ threadId: ThreadId.make("sub-agent") });
371 startDelivery();
372 appAtomRegistry.set(editingQueuedMessageIdsAtom, { [queued.messageId]: true });
373 await state.manager.enqueue(queued);
374 using confirm = vi.spyOn(state.manager, "confirmQueued");
375 const delivered = Promise.withResolvers<void>();
376 const release = appAtomRegistry.subscribe(dispatchingQueuedMessageIdAtom, (id) => {
377 if (id === null) delivered.resolve();
378 });
379 // Ending the edit wakes the drain through a registry notification, the path on which the
380 // mount's synchronous row change notifies the drain again mid-pass.
381 appAtomRegistry.set(editingQueuedMessageIdsAtom, {});
382 expect(state.familyMount).toHaveBeenCalledOnce();
383 expect(confirm).toHaveBeenCalledOnce();
384 await delivered.promise.finally(release);
385 expect(state.start).toHaveBeenCalledOnce();
386 expect(remaining()).toEqual([]);
387 });
388
389 it("stops the remaining settings and turn commands when an earlier update loses access", async () => {
390 setOperateGrant("secondary", true);
391 const queued = message({
392 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "different-model" },
393 runtimeMode: "approval-required",
394 interactionMode: "plan",
395 });
396 await state.manager.enqueue(queued);
397 state.runtime.mockImplementationOnce(async () => {
398 setOperateGrant("secondary", false);
399 return AsyncResult.success(undefined);
400 });
401 await runDrain();
402 expect(state.runtime).toHaveBeenCalledOnce();
403 expect(state.interaction).not.toHaveBeenCalled();
404 expect(state.prepare).not.toHaveBeenCalled();
405 expect(state.start).not.toHaveBeenCalled();
406 expect(remaining()).toEqual([queued]);
407 });
408
409 it.each([
410 { isCreation: false, accepted: false },
411 { isCreation: true, accepted: true },
412 ])(
413 "preserves a rejected queue but removes an accepted turn after access loss (new task: $isCreation, accepted: $accepted)",
414 async ({ isCreation, accepted }) => {
415 setOperateGrant("secondary", true);
416 const queued = message(
417 isCreation
418 ? {
419 threadId: ThreadId.make("new-thread"),
420 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
421 creation: {
422 projectId: ProjectId.make("project"),
423 projectCwd: "/repo",
424 workspaceMode: "local",
425 branch: null,
426 worktreePath: null,
427 },
428 }
429 : {},
430 );
431 await state.manager.enqueue(queued);
432 const started = Promise.withResolvers<void>();
433 const delivery =
434 Promise.withResolvers<AsyncResult.AsyncResult<{ readonly sequence?: number }, Error>>();
435 state.start.mockImplementationOnce(() => {
436 started.resolve();
437 return delivery.promise;
438 });
439 const drained = runDrain();
440 await started.promise;
441 setOperateGrant("secondary", false);
442 delivery.resolve(
443 accepted
444 ? AsyncResult.success({})
445 : AsyncResult.failure(Cause.fail(new Error("This connection cannot control this task."))),
446 );
447 await drained;
448
449 expect(remaining()).toEqual(accepted ? [] : [queued]);
450 expect(state.draftText).toBe("Keep this draft separate");
451 if (!accepted) {
452 setOperateGrant("secondary", true);
453 await runDrain();
454 expect(state.start).toHaveBeenCalledTimes(2);
455 expect(remaining()).toEqual([]);
456 expect(state.draftText).toBe("Keep this draft separate");
457 }
458 },
459 );
460
461 it.each([
462 { isCreation: false, uploadFails: false },
463 { isCreation: true, uploadFails: false },
464 { isCreation: false, uploadFails: true },
465 { isCreation: true, uploadFails: true },
466 ])(
467 "keeps the message queued after upload revocation (new task: $isCreation, upload fails: $uploadFails)",
468 async ({ isCreation, uploadFails }) => {
469 setOperateGrant("secondary", true);
470 const queued = message(
471 isCreation
472 ? {
473 threadId: ThreadId.make("new-thread"),
474 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
475 creation: {
476 projectId: ProjectId.make("project"),
477 projectCwd: "/repo",
478 workspaceMode: "local",
479 branch: null,
480 worktreePath: null,
481 },
482 }
483 : {},
484 );
485 await state.manager.enqueue(queued);
486 state.prepare.mockImplementationOnce(async ({ attachments }) => {
487 setOperateGrant("secondary", false);
488 if (uploadFails) throw new Error("This connection cannot upload attachments.");
489 return {
490 status: "ready",
491 attachments: [],
492 draftAttachments: attachments,
493 pendingAttachmentIds: [],
494 };
495 });
496 await runDrain();
497 expect(state.prepare).toHaveBeenCalledOnce();
498 expect(state.start).not.toHaveBeenCalled();
499 expect(remaining()).toEqual([queued]);
500 },
501 );
502});
503
504describe("compaction before a queued send", () => {
505 const compactTurn = {
506 commandId: "command:compact",
507 message: {
508 messageId: "message:compact",
509 role: "user",
510 text: "/compact",
511 attachments: [],
512 idleCompaction: "onReturn",
513 },
514 dispatchMode: "queue",
515 };
516
517 it("queues /compact ahead of the message and resends the same /compact on a retry", async () => {
518 setOperateGrant("secondary", true);
519 const queued = message({ compactBeforeSend: true, dispatchMode: "queue" });
520 await state.manager.enqueue(queued);
521 state.start
522 .mockResolvedValueOnce(AsyncResult.success({}))
523 .mockResolvedValueOnce(AsyncResult.failure(Cause.fail(new Error("Socket is not connected"))));
524 await runDrain();
525 expect(remaining()).toEqual([queued]);
526
527 // A restarted app delivers the persisted message again.
528 state.cleanups.splice(0).forEach((cleanup) => cleanup());
529 await runDrain();
530 expect(state.start.mock.calls.map(([call]) => call.input)).toEqual([
531 expect.objectContaining(compactTurn),
532 expect.objectContaining({ commandId: "command", dispatchMode: "queue" }),
533 expect.objectContaining(compactTurn),
534 expect.objectContaining({ commandId: "command", dispatchMode: "queue" }),
535 ]);
536 expect(remaining()).toEqual([]);
537 });
538
539 it("holds the message when its /compact cannot be sent", async () => {
540 setOperateGrant("secondary", true);
541 const queued = message({ compactBeforeSend: true, dispatchMode: "queue" });
542 await state.manager.enqueue(queued);
543 state.start.mockResolvedValueOnce(
544 AsyncResult.failure(Cause.fail(new Error("Socket is not connected"))),
545 );
546 await runDrain();
547 expect(state.start).toHaveBeenCalledOnce();
548 expect(state.start.mock.calls[0]?.[0].input).toMatchObject(compactTurn);
549 expect(remaining()).toEqual([queued]);
550 });
551});