diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d85504bb264e7..92af809e0a07b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -606,10 +606,11 @@ function delegatedTaskTerminalStatus( } } +// Held runs wait for the user; runs queued behind a later run do not. function nextQueuedRun( projection: Pick, ): OrchestrationV2Run | undefined { - return queuedRunsInDeliveryOrder(projection)[0]; + return queuedRunsInDeliveryOrder(projection).find((run) => run.queueHeld !== true); } function latestStableRun( @@ -1284,8 +1285,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || - projection.runs.some(isBlockingRun) || - projection.runs.some((run) => run.status === "queued" && run.queueHeld === true) + projection.runs.some(isBlockingRun) ) { return; } @@ -5032,9 +5032,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio rootNodeId, activeAttemptId: attemptId, status: "queued", + // A message joins a hold placed on messages waiting behind the same + // run. A held queue that this run started ahead of keeps its hold. ...(worktreeMissingRun !== undefined || projection.runs.some( - (candidate) => candidate.status === "queued" && candidate.queueHeld === true, + (candidate) => + candidate.status === "queued" && + candidate.queueHeld === true && + candidate.ordinal > queueBehindRun.ordinal, ) ? { queueHeld: true } : {}), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 4f10a0512fd53..4cc5f7f2fbbf6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4687,11 +4687,11 @@ export const layer: Layer.Layer = SELECT 1 AS ready FROM orchestration_v2_projection_threads AS thread WHERE thread.thread_id = ${threadId} AND thread.archived_at IS NULL AND thread.deleted_at IS NULL AND EXISTS (SELECT 1 FROM orchestration_v2_projection_runs - WHERE thread_id = ${threadId} AND status = 'queued') + WHERE thread_id = ${threadId} AND status = 'queued' + AND json_extract(payload_json, '$.queueHeld') IS NOT 1) AND NOT EXISTS (SELECT 1 FROM orchestration_v2_projection_runs WHERE thread_id = ${threadId} - AND (status IN ('preparing', 'starting', 'running', 'waiting') - OR (status = 'queued' AND json_extract(payload_json, '$.queueHeld') = 1)))`; + AND status IN ('preparing', 'starting', 'running', 'waiting'))`; return rows.length > 0; }), ) @@ -6458,14 +6458,13 @@ export const layerMemory: Layer.Layer = Layer.effect( return ( projection.thread.archivedAt === null && projection.thread.deletedAt === null && - projection.runs.some((run) => run.status === "queued") && + projection.runs.some((run) => run.status === "queued" && run.queueHeld !== true) && !projection.runs.some( (run) => run.status === "preparing" || run.status === "starting" || run.status === "running" || - run.status === "waiting" || - (run.status === "queued" && run.queueHeld === true), + run.status === "waiting", ) ); }), diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index d122289ecc08e..48cf5db0d3237 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4287,6 +4287,152 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("starts a message queued behind a run that started ahead of a held queue", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const recovery = yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("runtime-layer-queue-behind-later-run"); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Queue behind a later run", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + const send = (key: string, type: "start_immediately" | "queue_after_active") => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:${key}`), + threadId, + messageId: MessageId.make(`${threadId}:${key}`), + text: key, + attachments: [], + modelSelection, + dispatchMode: { type }, + }); + const runFor = (key: string) => + orchestrator + .getThreadProjection(threadId) + .pipe( + Effect.map((projection) => + projection.runs.find((run) => run.userMessageId === `${threadId}:${key}`)!, + ), + ); + yield* send("active", "start_immediately"); + yield* send("held", "queue_after_active"); + yield* recovery.reconcile("startup"); + assert.isTrue((yield* runFor("held")).queueHeld); + + // The idle thread starts a new message ahead of the held queue, and a + // message queued behind that run waits for it, not for the held queue. + yield* send("later", "start_immediately"); + const later = yield* runFor("later"); + assert.equal(later.status, "starting"); + yield* send("follow-up", "queue_after_active"); + assert.notEqual((yield* runFor("follow-up")).queueHeld, true); + + const now = yield* DateTime.now; + yield* eventSink.commitCommand({ + commandId: CommandId.make(`${threadId}:complete-later`), + threadId, + commandType: "provider-runtime.reconcile", + acceptedAt: now, + events: [ + { + id: EventId.make(`${threadId}:complete-later`), + type: "run.updated", + threadId, + runId: later.id, + occurredAt: now, + payload: { ...later, status: "completed", completedAt: now }, + }, + ], + effects: [], + }); + yield* orchestrator.resumeQueuedRuns; + assert.equal((yield* runFor("follow-up")).status, "starting"); + const held = yield* runFor("held"); + assert.equal(held.status, "queued"); + assert.isTrue(held.queueHeld); + }), + ); + + it.effect("holds a message queued behind the run its held queue waits for", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("runtime-layer-queue-behind-held-run"); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Queue behind a held run", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + for (const [index, key] of ["active", "held", "follow-up"].entries()) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:${key}`), + threadId, + messageId: MessageId.make(`${threadId}:${key}`), + text: key, + attachments: [], + modelSelection, + dispatchMode: { type: index === 0 ? "start_immediately" : "queue_after_active" }, + }); + if (key !== "held") continue; + // The queue is held while the active run is still ending, as on Stop. + const held = (yield* orchestrator.getThreadProjection(threadId)).runs[1]!; + const now = yield* DateTime.now; + yield* eventSink.commitCommand({ + commandId: CommandId.make(`${threadId}:hold`), + threadId, + commandType: "provider-runtime.reconcile", + acceptedAt: now, + events: [ + { + id: EventId.make(`${threadId}:hold`), + type: "run.updated", + threadId, + runId: held.id, + occurredAt: now, + payload: { ...held, queueHeld: true }, + }, + ], + effects: [], + }); + } + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + projection.runs.map((run) => [run.status, run.queueHeld === true]), + [ + ["starting", false], + ["queued", true], + ["queued", true], + ], + ); + }), + ); + it.effect("edits and removes queued runs", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2;