diff --git a/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts b/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts index 0d7bfdc3e8c7a..a92b4023b72be 100644 --- a/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts +++ b/apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts @@ -28,6 +28,7 @@ const state = vi.hoisted(() => ({ modelUnavailable: vi.fn(() => false), familyMount: vi.fn(() => () => {}), threadShells: null as unknown as Atom.Writable>, + pendingThreadIds: null as unknown as Atom.Writable>, manager: null as unknown as ReturnType< typeof import("./thread-outbox-manager").createThreadOutboxManager >, @@ -87,8 +88,11 @@ vi.mock("./server", async () => { vi.mock("./threads", async () => { const { Atom, AsyncResult } = await import("effect/reactivity"); state.threadShells = Atom.make>([state.thread]).pipe(Atom.keepAlive); + state.pendingThreadIds = Atom.make>([]).pipe(Atom.keepAlive); + const pendingThreadIdsAtom = Atom.map(state.pendingThreadIds, (ids) => new Set(ids)); return { threadEnvironment: { + pendingThreadIdsAtom: () => pendingThreadIdsAtom, startTurn: state.start, updateMetadata: state.metadata, setRuntimeMode: state.runtime, @@ -140,29 +144,40 @@ vi.mock("./use-thread-outbox", async () => { threadOutboxShellStatusesAtom: Atom.make(new Map([["secondary", "live"]])), }; }); -vi.mock("./use-composer-drafts", async () => { - const { Atom } = await import("effect/reactivity"); - return { - composerDraftsAtom: Atom.make({}), - removeDeliveredCloudQueuedMessage: async () => {}, - appendComposerDraftAttachments: vi.fn(), - flushComposerDrafts: vi.fn(), - getComposerDraftSnapshot: vi.fn(() => ({ text: state.draftText, attachments: [] })), - mergeComposerDraftContent: vi.fn(async (_key: string, input: { text: string }) => { - state.draftText += `\n${input.text}`; - }), - replaceComposerDraftAttachments: vi.fn(), - setFreshSessionPending: vi.fn(), - undoComposerDraftMerge: vi.fn(), - updateComposerDraftSettings: vi.fn(), - waitForComposerDraftsLoaded: vi.fn(), - }; -}); +vi.mock("expo-file-system", () => ({ + Directory: class { + create() {} + }, + File: class { + exists = false; + create() {} + write() {} + moveSync() {} + }, + Paths: { document: { uri: "file:///documents" } }, +})); +// The real draft store keeps mode picks; draft text stays a fake the restore tests read. +vi.mock("./use-composer-drafts", async (importOriginal) => ({ + ...(await importOriginal()), + removeDeliveredCloudQueuedMessage: async () => {}, + appendComposerDraftAttachments: vi.fn(), + flushComposerDrafts: vi.fn(), + getComposerDraftSnapshot: vi.fn(() => ({ text: state.draftText, attachments: [] })), + mergeComposerDraftContent: vi.fn(async (_key: string, input: { text: string }) => { + state.draftText += `\n${input.text}`; + }), + replaceComposerDraftAttachments: vi.fn(), + setFreshSessionPending: vi.fn(), + undoComposerDraftMerge: vi.fn(), + updateComposerDraftSettings: vi.fn(), + waitForComposerDraftsLoaded: vi.fn(async () => {}), +})); import { appAtomRegistry } from "./atom-registry"; import { environmentSession } from "./session"; import { dispatchingQueuedMessageIdAtom, editingQueuedMessageIdsAtom } from "./use-thread-outbox"; import { createThreadOutboxDelivery } from "./threadOutboxDelivery"; +import { composerDraftsAtom, waitForComposerDraftsLoaded } from "./use-composer-drafts"; const message = (overrides: Partial = {}): QueuedThreadMessage => ({ environmentId: EnvironmentId.make("secondary"), @@ -213,7 +228,9 @@ beforeEach(() => { state.config.shellThreadFamilies = false; state.modelUnavailable.mockReset().mockReturnValue(false); state.familyMount.mockReset().mockReturnValue(() => {}); + appAtomRegistry.set(composerDraftsAtom, {}); appAtomRegistry.set(state.threadShells, [state.thread]); + appAtomRegistry.set(state.pendingThreadIds, []); appAtomRegistry.set(state.manager.queuedMessagesByThreadKeyAtom, {}); appAtomRegistry.set(dispatchingQueuedMessageIdAtom, null); appAtomRegistry.set(editingQueuedMessageIdsAtom, {}); @@ -406,6 +423,94 @@ describe("queued task operation access", () => { expect(remaining()).toEqual([queued]); }); + it("waits for a pending mode change to settle before trusting the thread's mode", async () => { + setOperateGrant("secondary", true); + // Resume's mode change is in flight: the shell still holds Full access, as the server does. + appAtomRegistry.set(state.pendingThreadIds, ["thread"]); + const queued = message({ runtimeMode: "approval-required" }); + await state.manager.enqueue(queued); + startDelivery(); + expect(appAtomRegistry.get(dispatchingQueuedMessageIdAtom)).toBeNull(); + expect(state.runtime).not.toHaveBeenCalled(); + + // The server refused it, so delivery applies the mode itself and stops when that is refused too. + state.runtime.mockResolvedValueOnce( + AsyncResult.failure(Cause.fail(new Error("Background work would end."))), + ); + const settled = Promise.withResolvers(); + const release = appAtomRegistry.subscribe(dispatchingQueuedMessageIdAtom, (id) => { + if (id === null) settled.resolve(); + }); + appAtomRegistry.set(state.pendingThreadIds, []); + await settled.promise.finally(release); + expect(state.runtime).toHaveBeenCalledOnce(); + expect(state.start).not.toHaveBeenCalled(); + expect(remaining()).toEqual([queued]); + }); + + it.each([ + { queued: ["approval-required"], pick: "approval-required", failed: null, kept: false }, + { queued: ["full-access"], pick: "full-access", failed: null, kept: false }, + { queued: ["approval-required"], pick: "approval-required", failed: "mode", kept: true }, + // Saved drafts could still load afterwards and bring back the draft the release emptied. + { queued: ["approval-required"], pick: "approval-required", failed: "drafts", kept: true }, + // Picked again after the Full access message was queued; that message still moves the thread. + { + queued: ["approval-required", "full-access"], + pick: "approval-required", + failed: null, + kept: true, + }, + // A message without a mode, as an MCP app sends, follows the thread and moves nothing. + { + queued: ["approval-required", undefined], + pick: "approval-required", + failed: null, + kept: false, + }, + ] as const)( + "keeps the composer's mode pick until a delivered send applied it last ($queued, failed: $failed)", + async ({ queued, pick, failed, kept }) => { + setOperateGrant("secondary", true); + const draft = { text: "unsent follow-up", attachments: [] }; + appAtomRegistry.set(composerDraftsAtom, { + "secondary:thread": { ...draft, runtimeMode: pick }, + }); + for (const [index, runtimeMode] of queued.entries()) { + await state.manager.enqueue( + message({ + messageId: MessageId.make(`message-${index}`), + commandId: CommandId.make(`command-${index}`), + createdAt: `2026-09-05T00:00:0${index}.000Z`, + ...(runtimeMode === undefined ? {} : { runtimeMode }), + }), + ); + } + if (failed === "drafts") { + vi.mocked(waitForComposerDraftsLoaded).mockRejectedValueOnce(new Error("Read failed.")); + } + if (failed === "mode") { + state.runtime.mockResolvedValueOnce( + AsyncResult.failure(Cause.fail(new Error("Mode change refused."))), + ); + await runDrain(); + } else { + const delivered = Promise.withResolvers(); + const release = appAtomRegistry.subscribe( + state.manager.queuedMessagesByThreadKeyAtom, + () => { + if (remaining().length === 0) delivered.resolve(); + }, + ); + startDelivery(); + await delivered.promise.finally(release); + } + expect(appAtomRegistry.get(composerDraftsAtom)["secondary:thread"]).toEqual( + kept ? { ...draft, runtimeMode: pick } : draft, + ); + }, + ); + it.each([ { isCreation: false, accepted: false }, { isCreation: true, accepted: true }, diff --git a/apps/mobile/src/state/threadOutboxDelivery.ts b/apps/mobile/src/state/threadOutboxDelivery.ts index b3feeae3bb801..46b95dccc2f56 100644 --- a/apps/mobile/src/state/threadOutboxDelivery.ts +++ b/apps/mobile/src/state/threadOutboxDelivery.ts @@ -20,6 +20,7 @@ import { managedRuntimeModeDenial, PROVIDER_SEND_TURN_MAX_ATTACHMENTS, type EnvironmentId, + type ThreadId, } from "@t3tools/contracts"; import { buildTemporaryWorktreeBranchName } from "@t3tools/shared/git"; import * as Cause from "effect/Cause"; @@ -78,6 +79,7 @@ import { type ComposerDraft, getComposerDraftSnapshot, mergeComposerDraftContent, + releaseComposerDraftRuntimeMode, replaceComposerDraftAttachments, removeDeliveredCloudQueuedMessage, undoComposerDraftMerge, @@ -592,7 +594,9 @@ async function preserveUploadedAttachmentsForEditor( const deliveryInputsAtom = Atom.make((get) => { const connectedEnvironments = get(workspaceConnections.environmentsAtom); const operableEnvironments = new Set(); + const pendingThreadIds = new Map>(); for (const { environmentId } of connectedEnvironments) { + pendingThreadIds.set(environmentId, get(threadEnvironment.pendingThreadIdsAtom(environmentId))); if ( sessionHasScope( get(environmentSession.sessionStateAtom(environmentId)), @@ -612,6 +616,7 @@ const deliveryInputsAtom = Atom.make((get) => { serverConfigs: get(environmentServerConfigsAtom), connectedEnvironments, operableEnvironments, + pendingThreadIds, }; }).pipe(Atom.withLabel("mobile:thread-outbox:delivery-inputs")); @@ -831,6 +836,24 @@ export function createThreadOutboxDelivery(): { return false; } } + // Loading restores the saved draft for a key this release empties, so it must finish first. + const draftsLoaded = await waitForComposerDraftsLoaded().then( + () => true, + () => false, + ); + const threadKey = scopedThreadKey(queuedMessage.environmentId, queuedMessage.threadId); + // A newer pick can share this mode while another queued message still moves the thread off it. + const modeChangeQueued = ( + appAtomRegistry.get(threadOutboxManager.queuedMessagesByThreadKeyAtom)[threadKey] ?? [] + ).some( + (message) => + message.messageId !== queuedMessage.messageId && + message.runtimeMode !== undefined && + message.runtimeMode !== settings.runtimeMode, + ); + if (draftsLoaded && !modeChangeQueued) { + releaseComposerDraftRuntimeMode(threadKey, settings.runtimeMode); + } if (settings.interactionMode !== thread.interactionMode) { if (!hasAccess()) return true; @@ -1191,6 +1214,7 @@ export function createThreadOutboxDelivery(): { serverConfigs, connectedEnvironments, operableEnvironments, + pendingThreadIds, } = appAtomRegistry.get(deliveryInputsAtom); if (dispatchingQueuedMessageId !== null) { return; @@ -1329,6 +1353,15 @@ export function createThreadOutboxDelivery(): { if (deliveryAction === "send" && !operableEnvironments.has(nextQueuedMessage.environmentId)) { continue; } + // An unconfirmed optimistic change, such as Resume's mode, can still be refused; the send + // must compare against what the server holds, or it skips a mode change it still needs. + if ( + deliveryAction === "send" && + creation === undefined && + pendingThreadIds.get(nextQueuedMessage.environmentId)?.has(nextQueuedMessage.threadId) + ) { + continue; + } const dispatchStep = resolveThreadOutboxDispatchStep({ deliveryAction, fileAttachments: nextQueuedMessage.attachments.filter( diff --git a/apps/mobile/src/state/use-composer-drafts.ts b/apps/mobile/src/state/use-composer-drafts.ts index 643d348093208..e8e387dd3b2a1 100644 --- a/apps/mobile/src/state/use-composer-drafts.ts +++ b/apps/mobile/src/state/use-composer-drafts.ts @@ -1613,6 +1613,17 @@ export function releaseComposerDraftModelSelection( }); } +/** Lets the thread's mode show again once a send applied this pick, so later changes elsewhere hold. */ +export function releaseComposerDraftRuntimeMode(draftKey: string, runtimeMode: RuntimeMode): void { + updateComposerDrafts((current) => { + const existing = current[draftKey]; + // A different pick made after this message was queued still waits for its own send. + if (existing?.runtimeMode !== runtimeMode) return current; + const { runtimeMode: _runtimeMode, ...draft } = existing; + return withComposerDraft(current, draftKey, draft); + }); +} + export function setFreshSessionPending(draftKey: string, pending: boolean): void { if ((getComposerDraftSnapshot(draftKey).freshContext === true) === pending) return; updateComposerDraftSettings(draftKey, { freshContext: pending ? true : undefined }); diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 031ef68a14502..a37642e78d881 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -382,6 +382,7 @@ describe("remote thread lifecycle commands", () => { ["setAutoSettle", { enabled: false }, { autoSettleDisabledAt: expect.any(Object) }], ["reorderPin", { orderKey: "b" }, { pinOrderKey: "b" }], ["reorderActive", { orderKey: "b" }, { activeOrderKey: "b" }], + ["setRuntimeMode", {}, { runtimeMode: "approval-required" }], ] as const; it.effect.each(actions)( @@ -417,6 +418,7 @@ describe("remote thread lifecycle commands", () => { enabled: false, orderKey: "a", snoozedUntil: "2099-01-01T00:00:00.000Z", + runtimeMode: "approval-required", ...input, }, }); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 0e9de03f5c066..6f86a5d0773f1 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -556,5 +556,9 @@ export function createThreadEnvironmentAtoms( ...thread, modelSelection: input.modelSelection, })), + setRuntimeMode: optimistic.wrap(commands.setRuntimeMode, (thread, input) => ({ + ...thread, + runtimeMode: input.runtimeMode, + })), }; }