Skip to content

fix(orchestrator): defer queued provider handoff until delivery #12226

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
juliusmarminge merged 12 commits into from
Sep 17, 2026
Merged

fix(orchestrator): defer queued provider handoff until delivery #12226

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
4b033a3
fix(orchestrator): defer queued provider handoff until delivery
juliusmarminge Sep 17, 2026
d1c2161
fix(orchestrator): preserve imported native provider mapping
juliusmarminge Sep 17, 2026
8196141
fix(queue): retain queued delivery mode and native continuation
juliusmarminge Sep 17, 2026
3c74ce1
fix(queue): carry steered provider context into queued turns
juliusmarminge Sep 17, 2026
305ed6a
fix(orchestrator): record selected provider during steering restart
juliusmarminge Sep 17, 2026
5b53c10
fix(orchestrator): record model selection when dispatching
juliusmarminge Sep 17, 2026
2645797
fix(queue): validate the selected provider capability
juliusmarminge Sep 17, 2026
ff8e024
fix(server): restore imported history for queued provider takeover
juliusmarminge Sep 17, 2026
d8ed6ca
fix(queue): advance after queued provider handoff rejection
juliusmarminge Sep 17, 2026
0b45a61
fix(queue): fail queued turns when startup cannot proceed
juliusmarminge Sep 17, 2026
c9f0de2
fix(queue): show queued handoff in conversation timeline
juliusmarminge Sep 17, 2026
cbbd1ce
test(queue): cover queued provider session detachment
juliusmarminge Sep 17, 2026
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
1 change: 1 addition & 0 deletions apps/mobile/src/state/thread-outbox-model.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ export const QueuedThreadMessageSchema = Schema.Struct({
context: Schema.optional(OrchestrationMessageContext),
attachments: Schema.Array(DraftComposerAttachmentSchema),
modelSelection: Schema.optional(ModelSelection),
dispatchMode: Schema.optional(Schema.Literals(["auto", "queue", "steer", "restart"])),
runtimeMode: Schema.optional(RuntimeMode),
interactionMode: Schema.optional(ProviderInteractionMode),
// Present when the queued item creates a brand-new thread (pending task)
Expand Down
9 changes: 9 additions & 0 deletions apps/mobile/src/state/thread-outbox.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,15 @@ describe("thread outbox", () => {
decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))),
).toEqual(message);
});
it("retains queue mode when a queued provider switch reloads from storage", () => {
const message: QueuedThreadMessage = {
...queuedMessage({ messageId: "queued-switch", createdAt: "2026-09-17T09:00:00.000Z" }),
dispatchMode: "queue",
};
expect(
decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))),
).toEqual(message);
});
it.each(["read", "json", "schema"] as const)(
"recovers usable messages without permitting cleanup after a record %s failure",
async (failure) => {
Expand Down
20 changes: 0 additions & 20 deletions apps/mobile/src/state/use-thread-outbox-drain.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,6 @@ import {
import { removeThreadOutboxMessage } from "./thread-outbox-removal";
import {
isQueuedThreadCreationSendable,
modelSelectionsEqual,
resolveThreadOutboxDeliveryAction,
resolveThreadOutboxDispatchStep,
resolveThreadOutboxFailureAction,
Expand Down Expand Up @@ -546,9 +545,6 @@ async function preserveUploadedAttachmentsForEditor(

export function useThreadOutboxDrain(): void {
const startTurn = useAtomCommand(threadEnvironment.startTurn, { reportFailure: false });
const updateThreadMetadata = useAtomCommand(threadEnvironment.updateMetadata, {
reportFailure: false,
});
const setThreadRuntimeMode = useAtomCommand(threadEnvironment.setRuntimeMode, {
reportFailure: false,
});
Expand Down Expand Up @@ -705,21 +701,6 @@ export function useThreadOutboxDrain(): void {
}
const { reportFailure } = makeDeliveryHelpers(queuedMessage);

if (!modelSelectionsEqual(settings.modelSelection, thread.modelSelection)) {
const updateResult = await updateThreadMetadata({
environmentId: queuedMessage.environmentId,
input: {
commandId: settingsCommandId(queuedMessage, "model-selection"),
threadId: queuedMessage.threadId,
modelSelection: settings.modelSelection,
},
});
if (AsyncResult.isFailure(updateResult)) {
reportFailure(updateResult, "settings-sync");
return false;
}
}

if (settings.runtimeMode !== thread.runtimeMode) {
const runtimeResult = await setThreadRuntimeMode({
environmentId: queuedMessage.environmentId,
Expand Down Expand Up @@ -855,7 +836,6 @@ export function useThreadOutboxDrain(): void {
setThreadInteractionMode,
setThreadRuntimeMode,
startTurn,
updateThreadMetadata,
restoreQueuedMessage,
],
);
Expand Down
Loading
Loading