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 grantedEnvironments: new Set<string>(),
17 draftText: "Keep this draft separate",
18 cleanups: [] as Array<() => void>,
19 start: vi.fn(),
20 metadata: vi.fn(),
21 runtime: vi.fn(),
22 interaction: vi.fn(),
23 prepare: vi.fn(),
24 checkout: vi.fn(),
25 manager: null as unknown as ReturnType<
26 typeof import("./thread-outbox-manager").createThreadOutboxManager
27 >,
28 thread: {
29 environmentId: "secondary",
30 id: "thread",
31 modelSelection: { instanceId: "codex", model: "gpt-5.4" },
32 runtimeMode: "full-access",
33 interactionMode: "default",
34 session: null,
35 },
36 config: {
37 providers: [{ instanceId: "codex", driver: "codex" }],
38 environment: { capabilities: {} },
39 },
40}));
41vi.mock("react", () => ({
42 useCallback: <A>(callback: A) => callback,
43 useEffect: (effect: () => void | (() => void)) => {
44 const cleanup = effect();
45 if (cleanup) state.cleanups.push(cleanup);
46 },
47 useRef: <A>(value: A) => ({ current: value }),
48 useState: <A>(initial: A) => [initial, vi.fn()],
49}));
50vi.mock("react-native", () => ({ Alert: { alert: vi.fn() } }));
51vi.mock("@effect/atom-react", async () => {
52 const { appAtomRegistry } = await import("./atom-registry");
53 return { useAtomValue: <A>(atom: Atom.Atom<A>) => appAtomRegistry.get(atom) };
54});
55vi.mock("./session", () => ({
56 readEnvironmentScope: (environmentId: string, scope: string) =>
57 scope === AuthOrchestrationOperateScope && state.grantedEnvironments.has(environmentId),
58 useEnvironmentsWithScope: () => state.grantedEnvironments,
59}));
60vi.mock("./entities", () => ({
61 useProjects: () => [],
62 useThreadShells: () => [state.thread],
63 useServerConfigs: () => new Map([["secondary", state.config]]),
64}));
65vi.mock("./server", async () => {
66 const { Atom } = await import("effect/reactivity");
67 return {
68 serverEnvironment: { configValueAtom: Atom.family((_id: string) => Atom.make(state.config)) },
69 };
70});
71vi.mock("./threads", async () => {
72 const { Atom } = await import("effect/reactivity");
73 return {
74 threadEnvironment: {
75 startTurn: state.start,
76 updateMetadata: state.metadata,
77 setRuntimeMode: state.runtime,
78 setInteractionMode: state.interaction,
79 },
80 environmentThreadShells: { threadShellsAtom: Atom.make([state.thread]) },
81 };
82});
83vi.mock("./use-atom-command", () => ({ useAtomCommand: <A>(command: A) => command }));
84vi.mock("./git", () => ({ gitEnvironment: { preparePullRequestThread: state.checkout } }));
85vi.mock("../lib/modelOptions", () => ({ isModelSelectionUnavailable: () => false }));
86vi.mock("../lib/uuid", () => ({ uuidv4: () => "uuid", randomHex: () => "abcd" }));
87vi.mock("../lib/attachmentUpload", () => ({ prepareTurnAttachments: state.prepare }));
88vi.mock("./use-remote-environment-registry", () => ({
89 setPendingConnectionError: vi.fn(),
90 useRemoteConnectionStatus: () => ({
91 connectedEnvironments: [{ environmentId: "secondary", connectionState: "connected" }],
92 }),
93}));
94vi.mock("./thread-outbox", async () => {
95 const { createThreadOutboxManager } = await import("./thread-outbox-manager");
96 const { appAtomRegistry } = await import("./atom-registry");
97 state.manager = createThreadOutboxManager({
98 registry: appAtomRegistry,
99 storage: {
100 load: async () => ({ messages: [], errors: [] }),
101 write: async () => {},
102 remove: async () => {},
103 },
104 });
105 return {
106 threadOutboxManager: state.manager,
107 threadOutboxRevision: (id: MessageId) => state.manager.revisionOf(id),
108 confirmThreadOutboxMessageQueued: (message: QueuedThreadMessage) =>
109 state.manager.confirmQueued(message),
110 updateThreadOutboxMessage: (message: QueuedThreadMessage, revision?: number) =>
111 state.manager.update(message, revision),
112 };
113});
114vi.mock("./thread-outbox-removal", () => ({
115 removeThreadOutboxMessage: (message: QueuedThreadMessage, revision?: number) =>
116 state.manager.remove(message, revision),
117}));
118vi.mock("./use-thread-outbox", async () => {
119 const { Atom } = await import("effect/reactivity");
120 const { appAtomRegistry } = await import("./atom-registry");
121 await import("./thread-outbox");
122 return {
123 editingQueuedMessageIdsAtom: Atom.make({}).pipe(Atom.keepAlive),
124 dispatchingQueuedMessageIdAtom: Atom.make<MessageId | null>(null).pipe(Atom.keepAlive),
125 useThreadOutboxMessages: () => appAtomRegistry.get(state.manager.queuedMessagesByThreadKeyAtom),
126 useThreadOutboxShellStatuses: () => new Map([["secondary", "live"]]),
127 };
128});
129vi.mock("./use-composer-drafts", async () => {
130 const { Atom } = await import("effect/reactivity");
131 return {
132 composerDraftsAtom: Atom.make({}),
133 removeDeliveredCloudQueuedMessage: async () => {},
134 appendComposerDraftAttachments: vi.fn(),
135 flushComposerDrafts: vi.fn(),
136 getComposerDraftSnapshot: vi.fn(() => ({ text: state.draftText, attachments: [] })),
137 mergeComposerDraftContent: vi.fn(async (_key: string, input: { text: string }) => {
138 state.draftText += `\n${input.text}`;
139 }),
140 replaceComposerDraftAttachments: vi.fn(),
141 undoComposerDraftMerge: vi.fn(),
142 updateComposerDraftSettings: vi.fn(),
143 waitForComposerDraftsLoaded: vi.fn(),
144 };
145});
146
147import { appAtomRegistry } from "./atom-registry";
148import { dispatchingQueuedMessageIdAtom } from "./use-thread-outbox";
149import { useThreadOutboxDrain } from "./use-thread-outbox-drain";
150
151const message = (overrides: Partial<QueuedThreadMessage> = {}): QueuedThreadMessage => ({
152 environmentId: EnvironmentId.make("secondary"),
153 threadId: ThreadId.make("thread"),
154 commandId: CommandId.make("command"),
155 messageId: MessageId.make("message"),
156 text: "Keep this queued",
157 attachments: [],
158 createdAt: "2026-09-05T00:00:00.000Z",
159 ...overrides,
160});
161const remaining = () =>
162 Object.values(appAtomRegistry.get(state.manager.queuedMessagesByThreadKeyAtom)).flat();
163
164function runDrain() {
165 const settled = Promise.withResolvers<void>();
166 const release = appAtomRegistry.subscribe(dispatchingQueuedMessageIdAtom, (id) => {
167 if (id === null) settled.resolve();
168 });
169 useThreadOutboxDrain();
170 return settled.promise.finally(release);
171}
172
173beforeEach(() => {
174 state.grantedEnvironments = new Set(["primary"]);
175 state.draftText = "Keep this draft separate";
176 appAtomRegistry.set(state.manager.queuedMessagesByThreadKeyAtom, {});
177 appAtomRegistry.set(dispatchingQueuedMessageIdAtom, null);
178 for (const command of [
179 state.start,
180 state.metadata,
181 state.runtime,
182 state.interaction,
183 state.checkout,
184 ]) {
185 command.mockReset().mockResolvedValue(AsyncResult.success(undefined));
186 }
187 state.prepare.mockReset().mockImplementation(async ({ attachments }) => ({
188 status: "ready",
189 attachments: [],
190 draftAttachments: attachments,
191 pendingAttachmentIds: [],
192 }));
193});
194
195describe("queued PR checkout", () => {
196 const queuedPullRequest = () =>
197 message({
198 threadId: ThreadId.make("new-pr-thread"),
199 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
200 creation: {
201 projectId: ProjectId.make("project"),
202 projectCwd: "/repo",
203 workspaceMode: "worktree",
204 branch: "fork/feature",
205 worktreePath: null,
206 pullRequestReference: "https://github.com/acme/repo/pull/123",
207 },
208 });
209 const preparedCheckout = {
210 branch: "pr-123",
211 worktreePath: "/repo/.t3/worktrees/pr-123",
212 isOnPullRequestHead: true,
213 };
214
215 it("waits for the provider's PR checkout and launches in that worktree", async () => {
216 state.grantedEnvironments.add("secondary");
217 await state.manager.enqueue(queuedPullRequest());
218 const started = Promise.withResolvers<void>();
219 const checkout =
220 Promise.withResolvers<AsyncResult.AsyncResult<typeof preparedCheckout, Error>>();
221 state.checkout.mockImplementationOnce(() => {
222 started.resolve();
223 return checkout.promise;
224 });
225 const drained = runDrain();
226 await started.promise;
227 expect(state.start).not.toHaveBeenCalled();
228 expect(state.checkout).toHaveBeenCalledWith({
229 environmentId: "secondary",
230 input: { cwd: "/repo", reference: "https://github.com/acme/repo/pull/123", mode: "worktree" },
231 });
232 checkout.resolve(AsyncResult.success(preparedCheckout));
233 await drained;
234 expect(state.start).toHaveBeenCalledOnce();
235 const bootstrap = state.start.mock.calls[0]?.[0].input.bootstrap;
236 expect(bootstrap.createThread).toMatchObject({
237 branch: "pr-123",
238 worktreePath: preparedCheckout.worktreePath,
239 });
240 expect(bootstrap.prepareWorktree).toBeUndefined();
241 expect(remaining()).toEqual([]);
242 });
243
244 it("keeps the PR source queued on a transport failure without starting the agent", async () => {
245 state.grantedEnvironments.add("secondary");
246 const queued = queuedPullRequest();
247 await state.manager.enqueue(queued);
248 state.checkout.mockResolvedValueOnce(
249 AsyncResult.failure(Cause.fail(new Error("Socket is not connected"))),
250 );
251 await runDrain();
252 expect(state.start).not.toHaveBeenCalled();
253 expect(remaining()).toEqual([queued]);
254 });
255
256 it("restores the PR draft when the reused worktree cannot reach the PR head", async () => {
257 state.grantedEnvironments.add("secondary");
258 await state.manager.enqueue(queuedPullRequest());
259 state.checkout.mockResolvedValueOnce(
260 AsyncResult.success({ ...preparedCheckout, isOnPullRequestHead: false }),
261 );
262 await runDrain();
263 expect(state.start).not.toHaveBeenCalled();
264 expect(remaining()).toEqual([]);
265 const { updateComposerDraftSettings } = await import("./use-composer-drafts");
266 expect(updateComposerDraftSettings).toHaveBeenCalledWith(
267 expect.any(String),
268 expect.objectContaining({
269 workspaceSelection: expect.objectContaining({
270 pullRequestReference: queuedPullRequest().creation?.pullRequestReference,
271 }),
272 }),
273 );
274 });
275
276 it("stops launch if operation access is revoked while checking out the PR", async () => {
277 state.grantedEnvironments.add("secondary");
278 const queued = queuedPullRequest();
279 await state.manager.enqueue(queued);
280 state.checkout.mockImplementationOnce(async () => {
281 state.grantedEnvironments.delete("secondary");
282 return AsyncResult.success(preparedCheckout);
283 });
284 await runDrain();
285 expect(state.start).not.toHaveBeenCalled();
286 expect(remaining()).toEqual([queued]);
287 });
288
289 it("does not launch a stale task edited while the PR checkout was running", async () => {
290 state.grantedEnvironments.add("secondary");
291 const queued = queuedPullRequest();
292 await state.manager.enqueue(queued);
293 const edited = { ...queued, text: "Updated instructions" };
294 state.checkout.mockImplementationOnce(async () => {
295 await state.manager.update(edited);
296 return AsyncResult.success(preparedCheckout);
297 });
298 await runDrain();
299 expect(state.start).not.toHaveBeenCalled();
300 expect(remaining()).toEqual([edited]);
301 });
302});
303afterEach(() => {
304 state.cleanups.splice(0).forEach((cleanup) => cleanup());
305});
306
307describe("queued task operation access", () => {
308 it("parks a connected read-only target, then delivers its unchanged queue after a grant", async () => {
309 const queued = message();
310 await state.manager.enqueue(queued);
311 useThreadOutboxDrain();
312 expect(state.start).not.toHaveBeenCalled();
313 expect(state.prepare).not.toHaveBeenCalled();
314 expect(remaining()).toEqual([queued]);
315 state.grantedEnvironments.add("secondary");
316 await runDrain();
317 expect(state.start).toHaveBeenCalledOnce();
318 expect(remaining()).toEqual([]);
319 });
320
321 it("stops the remaining settings and turn commands when an earlier update loses access", async () => {
322 state.grantedEnvironments.add("secondary");
323 const queued = message({
324 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "different-model" },
325 runtimeMode: "approval-required",
326 interactionMode: "plan",
327 });
328 await state.manager.enqueue(queued);
329 state.runtime.mockImplementationOnce(async () => {
330 state.grantedEnvironments.delete("secondary");
331 return AsyncResult.success(undefined);
332 });
333 await runDrain();
334 expect(state.runtime).toHaveBeenCalledOnce();
335 expect(state.interaction).not.toHaveBeenCalled();
336 expect(state.prepare).not.toHaveBeenCalled();
337 expect(state.start).not.toHaveBeenCalled();
338 expect(remaining()).toEqual([queued]);
339 });
340
341 it.each([
342 { isCreation: false, accepted: false },
343 { isCreation: true, accepted: false },
344 { isCreation: false, accepted: true },
345 { isCreation: true, accepted: true },
346 ])(
347 "preserves a rejected queue but removes an accepted turn after access loss (new task: $isCreation, accepted: $accepted)",
348 async ({ isCreation, accepted }) => {
349 state.grantedEnvironments.add("secondary");
350 const queued = message(
351 isCreation
352 ? {
353 threadId: ThreadId.make("new-thread"),
354 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
355 creation: {
356 projectId: ProjectId.make("project"),
357 projectCwd: "/repo",
358 workspaceMode: "local",
359 branch: null,
360 worktreePath: null,
361 },
362 }
363 : {},
364 );
365 await state.manager.enqueue(queued);
366 const started = Promise.withResolvers<void>();
367 const delivery = Promise.withResolvers<AsyncResult.AsyncResult<void, Error>>();
368 state.start.mockImplementationOnce(() => {
369 started.resolve();
370 return delivery.promise;
371 });
372 const drained = runDrain();
373 await started.promise;
374 state.grantedEnvironments.delete("secondary");
375 delivery.resolve(
376 accepted
377 ? AsyncResult.success(undefined)
378 : AsyncResult.failure(Cause.fail(new Error("This connection cannot control this task."))),
379 );
380 await drained;
381
382 expect(remaining()).toEqual(accepted ? [] : [queued]);
383 expect(state.draftText).toBe("Keep this draft separate");
384 if (!accepted) {
385 state.grantedEnvironments.add("secondary");
386 await runDrain();
387 expect(state.start).toHaveBeenCalledTimes(2);
388 expect(remaining()).toEqual([]);
389 expect(state.draftText).toBe("Keep this draft separate");
390 }
391 },
392 );
393
394 it.each([
395 { isCreation: false, uploadFails: false },
396 { isCreation: true, uploadFails: false },
397 { isCreation: false, uploadFails: true },
398 { isCreation: true, uploadFails: true },
399 ])(
400 "keeps the message queued after upload revocation (new task: $isCreation, upload fails: $uploadFails)",
401 async ({ isCreation, uploadFails }) => {
402 state.grantedEnvironments.add("secondary");
403 const queued = message(
404 isCreation
405 ? {
406 threadId: ThreadId.make("new-thread"),
407 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
408 creation: {
409 projectId: ProjectId.make("project"),
410 projectCwd: "/repo",
411 workspaceMode: "local",
412 branch: null,
413 worktreePath: null,
414 },
415 }
416 : {},
417 );
418 await state.manager.enqueue(queued);
419 state.prepare.mockImplementationOnce(async ({ attachments }) => {
420 state.grantedEnvironments.delete("secondary");
421 if (uploadFails) throw new Error("This connection cannot upload attachments.");
422 return {
423 status: "ready",
424 attachments: [],
425 draftAttachments: attachments,
426 pendingAttachmentIds: [],
427 };
428 });
429 await runDrain();
430 expect(state.prepare).toHaveBeenCalledOnce();
431 expect(state.start).not.toHaveBeenCalled();
432 expect(remaining()).toEqual([queued]);
433 },
434 );
435});