Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions apps/mobile/src/state/thread-outbox-model.ts
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@ export const QueuedThreadMessageSchema = Schema.Struct({
context: Schema.optional(OrchestrationMessageContext),
attachments: Schema.Array(DraftComposerAttachmentSchema),
modelSelection: Schema.optional(ModelSelection),
// Written with every stored model since follow-ups carry only the user's pick. Entries without it
// come from builds that stored this client's cached thread model on every follow-up.
modelPicked: Schema.optional(Schema.Literal(true)),
dispatchMode: Schema.optional(Schema.Literals(["auto", "queue", "steer", "restart"])),
compactBeforeSend: Schema.optional(Schema.Boolean),
runtimeMode: Schema.optional(RuntimeMode),
Expand Down Expand Up @@ -133,12 +136,16 @@ export function encodeQueuedThreadMessage(message: QueuedThreadMessage): unknown
return encodeStoredQueuedThreadMessage({
schemaVersion: THREAD_OUTBOX_SCHEMA_VERSION,
...message,
...(message.modelSelection === undefined ? {} : { modelPicked: true as const }),
});
}

export function decodeQueuedThreadMessage(value: unknown): QueuedThreadMessage {
const { schemaVersion: _, ...message } = decodeStoredQueuedThreadMessage(value);
return message;
const { schemaVersion: _, modelPicked, ...message } = decodeStoredQueuedThreadMessage(value);
if (message.creation !== undefined || modelPicked === true) return message;
// Possibly a stale cached model rather than a pick: let the server use the thread's own.
const { modelSelection: _unconfirmed, ...followUp } = message;
return followUp;
}

export function groupQueuedThreadMessages(
Expand Down
31 changes: 31 additions & 0 deletions apps/mobile/src/state/thread-outbox.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,37 @@ describe("thread outbox", () => {
decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))),
).toEqual(message);
});
it("keeps a picked follow-up model through a persisted round trip", () => {
const message: QueuedThreadMessage = {
...queuedMessage({ messageId: "picked-model", createdAt: "2026-10-23T23:00:00.000Z" }),
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6-sol" },
};

expect(
decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))),
).toEqual(message);
});

it("drops the model an older build stored on a follow-up, so the server keeps the thread's own", () => {
const stale = { instanceId: "claudeAgent", model: "claude-opus-5-5" };
const base = queuedMessage({ messageId: "older-entry", createdAt: "2026-10-23T23:00:00.000Z" });
const stored = { schemaVersion: 3, ...base, modelSelection: stale };

expect(decodeQueuedThreadMessage(stored)).toEqual(base);
// A new thread needs its model, so an older creation entry keeps it.
const creation = {
projectId: ProjectId.make("project-1"),
workspaceMode: "local" as const,
branch: null,
worktreePath: null,
};
expect(decodeQueuedThreadMessage({ ...stored, creation })).toEqual({
...base,
modelSelection: stale,
creation,
});
});

it.each(["read", "json", "schema"] as const)(
"recovers usable messages without permitting cleanup after a record %s failure",
async (failure) => {
Expand Down
48 changes: 48 additions & 0 deletions apps/mobile/src/state/thread-task-permissions.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,54 @@ describe("mobile task permissions", () => {
});
});

describe("queued follow-up model", () => {
it("queues an unpicked follow-up despite an unavailable cached thread model", async () => {
state.grantedEnvironments.add("secondary");
state.serverConfig = {
...state.serverConfig,
providers: [{ instanceId: "codex", driver: "codex", enabled: false }],
} as unknown as ServerConfig;

expect(await useThreadComposerState().onSendMessage()).toBe("message");

expect(state.enqueue).toHaveBeenCalledExactlyOnceWith(
expect.objectContaining({ text: "Keep my draft", threadId: "thread" }),
);
const [queued] = state.enqueue.mock.calls[0] as unknown as [object];
expect(queued).not.toHaveProperty("modelSelection");
expect(state.clearDraft).toHaveBeenCalledOnce();
});

it("preserves an explicit unavailable pick instead of queueing it", async () => {
state.grantedEnvironments.add("secondary");
state.serverConfig = {
...state.serverConfig,
providers: [{ instanceId: "codex", driver: "codex", enabled: false }],
} as unknown as ServerConfig;
state.draft.modelSelection = {
instanceId: ProviderInstanceId.make("codex"),
model: "gpt-5.4",
};

expect(await useThreadComposerState().onSendMessage()).toBeNull();

expect(state.enqueue).not.toHaveBeenCalled();
expect(state.clearDraft).not.toHaveBeenCalled();
});

it("queues a follow-up with no model when the user picked none", async () => {
state.grantedEnvironments.add("secondary");
const composer = useThreadComposerState();

expect(await composer.onSendMessage()).toBe("message");

// The outbox can deliver after the thread moved elsewhere; only the server knows its model then.
expect(state.enqueue).toHaveBeenCalledOnce();
const [queued] = state.enqueue.mock.calls[0] as unknown as [object];
expect(queued).not.toHaveProperty("modelSelection");
});
});

describe("mobile handoff preview and submission", () => {
it("follows project enablement and provider availability for the selected destination", async () => {
state.grantedEnvironments.add("secondary");
Expand Down
39 changes: 38 additions & 1 deletion apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,7 @@ vi.mock("./use-composer-drafts", async () => {
}),
replaceComposerDraftAttachments: vi.fn(),
undoComposerDraftMerge: vi.fn(),
releaseDeliveredComposerModelPick: vi.fn(async () => {}),
updateComposerDraftSettings: vi.fn(),
waitForComposerDraftsLoaded: vi.fn(),
};
Expand Down Expand Up @@ -325,6 +326,40 @@ afterEach(() => {
state.cleanups.splice(0).forEach((cleanup) => cleanup());
});

describe("queued follow-up model", () => {
it("delivers a follow-up queued without a pick with no model, so the server keeps the thread's own", async () => {
setOperateGrant("secondary", true);
await state.manager.enqueue(message());

await runDrain();

expect(state.start).toHaveBeenCalledOnce();
expect(state.start.mock.calls[0]?.[0].input).not.toHaveProperty("modelSelection");
});

it("delivers a follow-up without a pick even when this client's copy of the thread model looks unavailable", async () => {
setOperateGrant("secondary", true);
state.modelUnavailable.mockReturnValue(true);
await state.manager.enqueue(message());

await runDrain();

expect(state.start).toHaveBeenCalledOnce();
expect(remaining()).toEqual([]);
});

it("delivers the model the user picked with the follow-up", async () => {
setOperateGrant("secondary", true);
const picked = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6-sol" };
await state.manager.enqueue(message({ modelSelection: picked }));

await runDrain();

expect(state.start).toHaveBeenCalledOnce();
expect(state.start.mock.calls[0]?.[0].input.modelSelection).toEqual(picked);
});
});

describe("queued task operation access", () => {
it("parks a connected read-only target, then delivers its unchanged queue after a grant", async () => {
const queued = message();
Expand All @@ -341,7 +376,9 @@ describe("queued task operation access", () => {

it("backs off instead of redelivering at once when a delivery hits a defect", async () => {
setOperateGrant("secondary", true);
const queued = message();
const queued = message({
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6-sol" },
});
await state.manager.enqueue(queued);
// Awaited steps catch their own failures, so a defect in the delivery's own code is what
// rejects a delivery.
Expand Down
22 changes: 19 additions & 3 deletions apps/mobile/src/state/threadOutboxDelivery.ts
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ import {
undoComposerDraftMerge,
updateComposerDraftSettings,
waitForComposerDraftsLoaded,
releaseDeliveredComposerModelPick,
} from "./use-composer-drafts";
import {
dispatchingQueuedMessageIdAtom,
Expand Down Expand Up @@ -775,7 +776,13 @@ export function createThreadOutboxDelivery(): {
);
if (!serverConfig) return false;
const settings = resolveQueuedThreadSettings(queuedMessage, thread, serverConfig.providers);
if (isModelSelectionUnavailable(serverConfig, settings.modelSelection, thread.projectId)) {
// Without a pick the server delivers on the thread's own model, which this client's copy
// of the thread can misreport right after reconnecting; the server checks that one itself.
const picked = queuedMessage.modelSelection !== undefined;
if (
picked &&
isModelSelectionUnavailable(serverConfig, settings.modelSelection, thread.projectId)
) {
const { title, detail } = describeUnavailableModelSelection(
serverConfig,
settings.modelSelection,
Expand Down Expand Up @@ -860,7 +867,10 @@ export function createThreadOutboxDelivery(): {
serverEnvironment.configValueAtom(queuedMessage.environmentId),
);
if (!currentConfig) return false;
if (isModelSelectionUnavailable(currentConfig, settings.modelSelection, thread.projectId)) {
if (
picked &&
isModelSelectionUnavailable(currentConfig, settings.modelSelection, thread.projectId)
) {
const { title, detail } = describeUnavailableModelSelection(
currentConfig,
settings.modelSelection,
Expand Down Expand Up @@ -924,7 +934,7 @@ export function createThreadOutboxDelivery(): {
),
attachments: prepared.attachments,
},
modelSelection: sendSettings.modelSelection,
...(picked ? { modelSelection: sendSettings.modelSelection } : {}),
titleSeed: deriveThreadTitleSeed({
text: queuedMessage.text,
attachments: queuedMessage.attachments,
Expand All @@ -944,6 +954,12 @@ export function createThreadOutboxDelivery(): {
return restoreQueuedMessage(persistedMessage, failure.message);
}
acknowledgedExistingThreadMessageIds.add(persistedMessage.messageId);
if (queuedMessage.modelSelection) {
await releaseDeliveredComposerModelPick(
scopedThreadKey(queuedMessage.environmentId, queuedMessage.threadId),
queuedMessage.modelSelection,
);
}
const delivered =
(await completeQueuedMessageDelivery(
persistedMessage,
Expand Down
18 changes: 18 additions & 0 deletions apps/mobile/src/state/use-composer-drafts.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,7 @@ import {
archiveCloudComposerDrafts,
clearComposerDraftContent,
clearComposerDraftContentState,
releaseDeliveredComposerModelPick,
subscribeComposerDraftModelReconciliation,
clearComposerDraftsEnvironment,
ComposerDraftPersistenceError,
Expand Down Expand Up @@ -549,6 +550,23 @@ describe("mobile composer drafts", () => {
expect(getComposerDraftSnapshot(key)).toEqual({ ...before, modelSelection: undefined });
});

it("releases a pick once a message carrying it is delivered", async () => {
const key = "environment-1:thread-1";
const pick = { instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-opus-5-5" };
setComposerDraftText(key, "next follow-up");
updateComposerDraftSettings(key, { modelSelection: pick, interactionMode: "plan" });
const before = getComposerDraftSnapshot(key);

await releaseDeliveredComposerModelPick(key, { ...pick, model: "claude-sonnet-5-5" });
expect(getComposerDraftSnapshot(key)).toEqual(before);

await releaseDeliveredComposerModelPick(key, {
...pick,
options: [{ id: "fastMode", value: false }],
});
expect(getComposerDraftSnapshot(key)).toEqual({ ...before, modelSelection: undefined });
});

it("retains an unavailable provider pick through differing durable model updates", () => {
const environmentId = EnvironmentId.make("environment-1");
const threadId = ThreadId.make("thread-1");
Expand Down
15 changes: 15 additions & 0 deletions apps/mobile/src/state/use-composer-drafts.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1521,6 +1521,21 @@ export function updateComposerDraftSettings(
});
}

export async function releaseDeliveredComposerModelPick(
draftKey: string,
delivered: ModelSelection,
): Promise<void> {
await waitForComposerDraftsLoaded();
updateComposerDrafts((current) => {
const draft = current[draftKey];
if (!draft?.modelSelection || !draftModelPickMatchesThread(draft.modelSelection, delivered)) {
return current;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}
const { modelSelection: _modelSelection, ...rest } = draft;
return withComposerDraft(current, draftKey, rest);
});
}

export function subscribeComposerDraftModelReconciliation(input: {
readonly registry: AtomRegistry.AtomRegistry;
readonly threadRefsAtom: Atom.Atom<ReadonlyArray<ScopedThreadRef>>;
Expand Down
9 changes: 6 additions & 3 deletions apps/mobile/src/state/use-thread-composer-state.ts
Original file line number Diff line number Diff line change
Expand Up @@ -562,11 +562,12 @@ export function useThreadComposerState() {
const serverConfig = selectedEnvironmentRuntime?.serverConfig;
if (
selectedEnvironmentRuntime?.connectionState === "connected" &&
isModelSelectionUnavailable(serverConfig, modelSelection, thread.projectId)
draft.modelSelection !== undefined &&
isModelSelectionUnavailable(serverConfig, draft.modelSelection, thread.projectId)
) {
const { title, detail } = describeUnavailableModelSelection(
serverConfig,
modelSelection,
draft.modelSelection,
thread.projectId,
);
Dialog.alert(title, detail);
Expand Down Expand Up @@ -692,7 +693,9 @@ export function useThreadComposerState() {
text,
attachments,
context: draft.context,
modelSelection,
// Only a pick travels with the message. The thread's model is read by the server when the
// outbox delivers, because this client's copy of it can be stale by then.
...(draft.modelSelection === undefined ? {} : { modelSelection: draft.modelSelection }),
runtimeMode: draft.runtimeMode ?? thread.runtimeMode,
interactionMode: resolveProviderInteractionMode(
provider,
Expand Down
17 changes: 16 additions & 1 deletion apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5710,7 +5710,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
]);
projection = applyToProjection(projection, event);
}
const modelSelection = command.modelSelection ?? projection.thread.modelSelection;
let modelSelection = command.modelSelection ?? projection.thread.modelSelection;
let dispatchMode = resolveMessageDispatchIntent(
projection,
command.dispatchMode,
Expand Down Expand Up @@ -5995,6 +5995,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
command.createdBy === "user" &&
(command.creationSource === "web" || command.creationSource === "mobile") &&
command.scheduledTaskId === undefined;
// A queued pick reaches the thread only when its run starts; unpicked messages behind it follow it.
// Monitor updates follow whatever runs before them; goal commands stay on the provider validated above.
const newestQueuedPick =
shouldQueue &&
sentByUserFromClient &&
command.modelSelection === undefined &&
parsedGoalCommand === null
? queuedRunsInDeliveryOrder(projection).findLast(
(run) =>
!followsCurrentConversation(
projection.messages.find((message) => message.id === run.userMessageId),
),
)
: undefined;
if (newestQueuedPick) modelSelection = newestQueuedPick.modelSelection;
Comment thread
vitalyiegorov marked this conversation as resolved.
Resolved
const heldRunsToRelease =
sentByUserFromClient && startsOrQueuesRun && !worktreeSetupUnresolved
? projection.runs.filter((run) => run.status === "queued" && run.queueHeld === true)
Expand Down
Loading
Loading