Skip to content
Closed
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
2 changes: 1 addition & 1 deletion apps/mobile/src/features/threads/ThreadQueueControl.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,7 @@ export function ThreadQueueSheet({ route }: StaticScreenProps<QueueTarget>) {
>
{workflow?.isHeld && queuedRuns.length > 0 ? (
<View className="gap-2 py-3">
<Text className="text-sm text-foreground-muted">Queue held after restart</Text>
<Text className="text-sm text-foreground-muted">Queue paused</Text>
<MaterialButton
label="Resume queue"
disabled={resuming || busyRunId !== null}
Expand Down
32 changes: 32 additions & 0 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5216,6 +5216,38 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
(transfer) => transfer.id !== pendingMergeBackTransfer?.id,
);
const now = yield* DateTime.now;
// A new message someone types into a client after Stop or a failure is the
// go-ahead the held queue waits for, so the held messages follow it instead of
// stranding. Scheduled prompts keep the client they were created from, and a
// late steer is re-sent under its original identity, so neither counts.
// Messages held because their worktree was never created stay held: once this
// run ends, nothing would stop them from running in the project checkout.
const worktreeMissing =
(latestRun?.status === "failed" || latestRun?.status === "interrupted") &&
latestRun.workspacePreparation?.type === "worktree" &&
projection.thread.worktreePath === null;
Comment on lines +5225 to +5228

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High orchestration-v2/Orchestrator.ts:5225

After a failed worktree preparation and an intervening turn, this code releases all held runs even though thread.worktreePath is still null; those runs then execute in the project checkout instead of the intended worktree. worktreeMissing only checks latestRun, so the completed intervening run hides the failed preparation; preserve that hold while any failed or interrupted worktree preparation remains unresolved.

-      const worktreeMissing =
-        (latestRun?.status === "failed" || latestRun?.status === "interrupted") &&
-        latestRun.workspacePreparation?.type === "worktree" &&
+      const worktreeMissing =
+        projection.runs.some(
+          (run) =>
+            (run.status === "failed" || run.status === "interrupted") &&
+            run.workspacePreparation?.type === "worktree",
+        ) &&
         projection.thread.worktreePath === null;
🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/orchestration-v2/Orchestrator.ts around lines 5225-5228:

After a failed worktree preparation and an intervening turn, this code releases all held runs even though `thread.worktreePath` is still `null`; those runs then execute in the project checkout instead of the intended worktree. `worktreeMissing` only checks `latestRun`, so the completed intervening run hides the failed preparation; preserve that hold while any failed or interrupted worktree preparation remains unresolved.

if (
command.createdBy === "user" &&
(command.creationSource === "web" || command.creationSource === "mobile") &&
command.scheduledTaskId === undefined &&
!projection.messages.some((message) => message.id === command.messageId) &&
!worktreeMissing
) {
for (const heldRun of projection.runs) {
if (heldRun.status !== "queued" || heldRun.queueHeld !== true) continue;
yield* emit(
events,
command,
)({
type: "run.updated",
threadId: command.threadId,
runId: heldRun.id,
providerInstanceId: heldRun.providerInstanceId,
occurredAt: now,
payload: { ...heldRun, queueHeld: false },
});
}
}
if (
!modelSelectionsEqual(projection.thread.modelSelection, modelSelection) ||
projection.thread.providerInstanceId !== modelSelection.instanceId
Expand Down
101 changes: 101 additions & 0 deletions apps/server/src/orchestration-v2/ThreadStop.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
ProviderInstanceId,
ProviderSessionId,
ProviderTurnId,
ScheduledTaskId,
ThreadId,
TurnItemId,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -705,3 +706,103 @@ it.effect(
assert.isTrue(Exit.isFailure(unrelated));
}).pipe(Effect.provide(layerTest)),
);

/** Stops a thread's only run with a message queued behind it, leaving that message held. */
const stopWithHeldQueue = (threadId: ThreadId, number: number) =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
yield* createWatchingThread(threadId, number);
yield* send(threadId, "compact", "start_immediately");
yield* send(threadId, "waiting", "queue_after_active");
const stopped = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
yield* orchestrator.dispatch({
type: "run.interrupt",
commandId: CommandId.make(`stop:${threadId}`),
threadId,
runId: stopped.id,
holdQueue: true,
});
assert.deepEqual((yield* threadState(threadId)).runs, ["interrupted", "queued:held"]);
});

it.effect("a message sent from a client after Stop releases the held queue behind it", () =>
Effect.gen(function* () {
const threadId = ThreadId.make("thread:stop-then-send");
yield* stopWithHeldQueue(threadId, 5);
yield* send(threadId, "next", "start_immediately");
assert.deepEqual((yield* threadState(threadId)).runs, ["interrupted", "queued", "starting"]);
}).pipe(Effect.provide(layerTest)),
);

it.effect("a scheduled prompt after Stop leaves the held queue held", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const threadId = ThreadId.make("thread:stop-then-automatic");
yield* stopWithHeldQueue(threadId, 6);
// A scheduled prompt carries the client its task was created from.
yield* orchestrator.dispatch({
type: "message.dispatch",
commandId: CommandId.make("send:stop-then-automatic:scheduled"),
threadId,
messageId: MessageId.make("message:stop-then-automatic:scheduled"),
scheduledTaskId: ScheduledTaskId.make("scheduled-task:nightly"),
text: "nightly",
attachments: [],
dispatchMode: { type: "start_immediately" },
createdBy: "user",
creationSource: "web",
});
assert.deepEqual((yield* threadState(threadId)).runs, [
"interrupted",
"queued:held",
"starting",
]);
}).pipe(Effect.provide(layerTest)),
);

it.effect("a late steer re-sent after Stop leaves the held queue held", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const threadId = ThreadId.make("thread:stop-then-late-steer");
yield* stopWithHeldQueue(threadId, 8);
// A steer that lands after its turn ended is re-sent under the message it already has.
const stopped = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
yield* orchestrator.dispatch({
type: "message.dispatch",
commandId: CommandId.make("send:stop-then-late-steer"),
threadId,
messageId: stopped.userMessageId,
text: "compact",
attachments: [],
dispatchMode: { type: "start_immediately" },
createdBy: "user",
creationSource: "web",
});
assert.include((yield* threadState(threadId)).runs, "queued:held");
}).pipe(Effect.provide(layerTest)),
);

it.effect("a client message leaves messages held behind a worktree that was never created", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const projections = yield* ProjectionStore.ProjectionStoreV2;
const threadId = ThreadId.make("thread:stop-without-worktree");
yield* stopWithHeldQueue(threadId, 7);
// Stop landed during worktree setup, so the thread never got its worktree.
const stopped = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
yield* projections.apply({
id: EventId.make("event:stop-without-worktree:preparation"),
type: "run.updated",
threadId,
runId: stopped.id,
occurredAt: yield* DateTime.now,
payload: { ...stopped, workspacePreparation: { type: "worktree", baseRef: "main" } },
});
yield* send(threadId, "next", "start_immediately");
assert.deepEqual((yield* threadState(threadId)).runs, [
"interrupted",
"queued:held",
"starting",
]);
}).pipe(Effect.provide(layerTest)),
);
4 changes: 3 additions & 1 deletion apps/web/src/components/chat/QueuedRunsControl.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -263,7 +263,9 @@ export function QueuedRunsControl({
<ComposerBanner.Icon>
<ListOrderedIcon />
</ComposerBanner.Icon>
<ComposerBanner.Content className="text-muted-foreground">Queued</ComposerBanner.Content>
<ComposerBanner.Content className="text-muted-foreground">
{workflow?.isHeld ? "Paused" : "Queued"}
</ComposerBanner.Content>
<ComposerBanner.Actions>
<ComposerBanner.Count>{items.length}</ComposerBanner.Count>
<ComposerBanner.ToggleIcon expanded={expanded} />
Expand Down
4 changes: 1 addition & 3 deletions docs/user/composer.md
Original file line number Diff line number Diff line change
Expand Up @@ -186,9 +186,7 @@ Queued messages appear above the composer. Rows show a thumbnail of any attached
the text. Drag a row by its handle to reorder it, use the handle's arrow keys, promote the message
to a steer, or remove it.

If the server restarts, saved queued messages keep their order and are held. Press
**Resume** in an empty composer on web or desktop, or **Resume queue** in the queue
sheet on mobile, to continue. You can edit, reorder, or remove held messages without starting them.
Queued messages pause after you press **Stop**, when the provider fails or worktree setup fails, and when the server restarts. They keep their order, and the queue shows that it is paused. Sending a new message continues the queue after it, except after a failed worktree setup, where retrying the setup continues it. To continue without sending anything, press **Resume** in an empty composer on web or desktop, or **Resume queue** in the queue sheet on mobile. You can edit, reorder, or remove paused messages without starting them.

The pencil on a queued row opens that message in the composer for editing. The original message
stays in the queue until you save, and its row is highlighted while you edit. The message's
Expand Down
Loading