diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1059c88230788..24dd2aa8ec1c0 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6193,6 +6193,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio queuedRunId: runId, targetRunId: goalStopRunId, reason: "Stopped to apply /goal", + goal: true, now, }); } @@ -9537,7 +9538,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio nodeId: queuedRootNode.id, providerInstanceId: queuedRun.providerInstanceId, occurredAt: now, - payload: { ...queuedRun, queuePosition: 1 }, + payload: { + ...queuedRun, + queuePosition: 1, + ...(goalCommand === null ? {} : { queuedAheadOfCompletions: false }), + }, }); const stops = isLiveRun(target) && !(yield* stopReachedRun(command, command.threadId, target.id)); @@ -9549,6 +9554,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio queuedRunId: queuedRun.id, targetRunId: stops ? target.id : undefined, reason: "Skipped compaction to send a queued message", + goal: goalCommand !== null, now, }); } @@ -9564,7 +9570,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio nodeId: queuedRootNode.id, providerInstanceId: queuedRun.providerInstanceId, occurredAt: now, - payload: { ...queuedRun, queuePosition: 1 }, + payload: { ...queuedRun, queuePosition: 1, queuedAheadOfCompletions: false }, }); const stops = target !== undefined && @@ -9580,6 +9586,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio queuedRunId: queuedRun.id, targetRunId: stops ? target.id : undefined, reason: "Stopped to apply /goal", + goal: true, now, }); } @@ -9679,19 +9686,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: "Automatic completion deliveries cannot be reordered.", }); } - const automaticRuns = queuedRuns.filter((run) => isAutomaticCompletionRun(projection, run)); - if ( - command.beforeRunId !== null && - automaticRuns.some((run) => run.id === command.beforeRunId) - ) { - return yield* new OrchestratorDispatchError({ - commandId: command.commandId, - commandType: command.type, - cause: "Queued messages cannot be reordered ahead of automatic completion delivery.", - }); - } - const reorderableRuns = queuedRuns.filter((run) => !isAutomaticCompletionRun(projection, run)); - const withoutMoving = reorderableRuns.filter((run) => run.id !== command.runId); + const withoutMoving = queuedRuns.filter((run) => run.id !== command.runId); const beforeIndex = command.beforeRunId === null ? withoutMoving.length @@ -9704,11 +9699,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } const reordered = [ - ...automaticRuns, ...withoutMoving.slice(0, beforeIndex), moving, ...withoutMoving.slice(beforeIndex), ]; + const lastCompletionIndex = reordered.findLastIndex((run) => + isAutomaticCompletionRun(projection, run), + ); const now = yield* DateTime.now; const emitEvent = emit(events, command); yield* Effect.forEach( @@ -9716,7 +9713,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (run, index) => Effect.gen(function* () { const queuePosition = index + 1; - if (run.queuePosition === queuePosition) { + const aheadOfCompletions = + index < lastCompletionIndex && !isAutomaticCompletionRun(projection, run); + if ( + run.queuePosition === queuePosition && + (run.queuedAheadOfCompletions === true) === aheadOfCompletions + ) { return; } yield* emitEvent({ @@ -9729,6 +9731,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: { ...run, queuePosition, + ...(aheadOfCompletions || run.queuedAheadOfCompletions === true + ? { queuedAheadOfCompletions: aheadOfCompletions } + : {}), }, }); }), @@ -11799,15 +11804,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly queuedRunId: RunId; readonly targetRunId: RunId | undefined; readonly reason: string; + /** A goal goes first among queued messages, so it drops their priority over completions. */ + readonly goal: boolean; readonly now: DateTime.Utc; }) { const { command, events, effects, projection, now, targetRunId } = input; + // Completions are renumbered too: a message moved ahead of them shares their position range. const others = queuedRunsInDeliveryOrder(projection).filter( - (run) => run.id !== input.queuedRunId && !isAutomaticCompletionRun(projection, run), + (run) => run.id !== input.queuedRunId, ); for (const [index, run] of others.entries()) { const queuePosition = index + 2; - if (run.queuePosition === queuePosition) continue; + if ( + run.queuePosition === queuePosition && + (!input.goal || run.queuedAheadOfCompletions !== true) + ) + continue; yield* emit( events, command, @@ -11818,7 +11830,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), providerInstanceId: run.providerInstanceId, occurredAt: now, - payload: { ...run, queuePosition }, + payload: { + ...run, + queuePosition, + ...(input.goal ? { queuedAheadOfCompletions: false } : {}), + }, }); } if (targetRunId === undefined) return undefined; diff --git a/apps/server/src/orchestration-v2/QueuedRunOrder.ts b/apps/server/src/orchestration-v2/QueuedRunOrder.ts index 72c5e319d69ed..3d60a9fab4705 100644 --- a/apps/server/src/orchestration-v2/QueuedRunOrder.ts +++ b/apps/server/src/orchestration-v2/QueuedRunOrder.ts @@ -9,6 +9,7 @@ export function isAutomaticCompletionRun( ); } +/** Automatic completions, and messages reordered ahead of them, run before other queued messages. */ export function queuedRunsInDeliveryOrder( projection: Pick, ): ReadonlyArray { @@ -20,9 +21,10 @@ export function queuedRunsInDeliveryOrder( return projection.runs .filter((run) => run.status === "queued") .toSorted((left, right) => { - const deliveryPriority = - Number(automaticCompletionMessageIds.has(right.userMessageId)) - - Number(automaticCompletionMessageIds.has(left.userMessageId)); + const hasPriority = (run: OrchestrationV2Run) => + automaticCompletionMessageIds.has(run.userMessageId) || + run.queuedAheadOfCompletions === true; + const deliveryPriority = Number(hasPriority(right)) - Number(hasPriority(left)); return ( deliveryPriority || (left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal) || diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index bd3f4e5b2e7d8..b0308bb7167c5 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -1628,11 +1628,11 @@ it.effect("promotes a queued goal by stopping the running turn", () => ); it.effect.each(["next", "/goal ship it"])( - "promoting %s into a compaction stops it and sends it next", + "promoting %s into a compaction while a message is moved ahead of a completion", (promoted) => Effect.scoped( Effect.gen(function* () { - const { events, layer, send, read, started, steered, startFirstTurn } = yield* goalHarness( + const { events, layer, send, started, steered, startFirstTurn } = yield* goalHarness( `compaction-promote-${promoted.length}`, ); yield* Effect.gen(function* () { @@ -1642,63 +1642,128 @@ it.effect.each(["next", "/goal ship it"])( yield* send("compact", "/compact", "queue"); yield* send("queued", "queued", "queue"); yield* send("next", promoted, "queue"); - const settled = yield* orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.type === "run.updated" && - event.payload.id === started[0]!.runId && - event.payload.status === "waiting", - ), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - const compacting = yield* orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.type === "provider-turn.updated" && - event.payload.status === "running" && - event.payload.runAttemptId !== started[0]!.attemptId, - ), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; - yield* Queue.offer(events, { - type: "provider_turn.updated", - driver, - providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, + const nextStart = Effect.suspend(() => { + const known = started.map((turn) => turn.attemptId); + return orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.status === "running" && + event.payload.runAttemptId !== null && + !known.includes(event.payload.runAttemptId), + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); }); - yield* Queue.offer(events, { - type: "turn.terminal", - driver, - providerThreadId: turn.providerThreadId, - providerTurnId: turn.id, - runOrdinal: started[0]!.runOrdinal, - status: "completed", - failure: null, - threadDisposition: "reusable", + const finish = Effect.fn("finish")(function* (index: number) { + const { runId, attemptId, runOrdinal } = started[index]!; + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === runId && + (event.payload.status === "waiting" || event.payload.status === "completed"), + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const startedNext = yield* nextStart; + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns.find( + (candidate) => candidate.id === ProviderTurnId.make(`provider-turn:${attemptId}`), + )!; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + yield* Fiber.join(startedNext); }); - yield* Fiber.join(settled); - yield* worker.drain(); - yield* orchestrator.resumeQueuedRuns; - yield* worker.drain(); - yield* Fiber.join(compacting); + yield* finish(0); const compactRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( (run) => run.userMessageId === MessageId.make("message:compact"), )!; - const nextRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (run) => run.userMessageId === MessageId.make("message:next"), + const firstRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.id === started[0]!.runId, )!; + const taskIds = [NodeId.make("task:compaction-promote")]; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("event:compaction-promote-completion"), + type: "run.updated", + threadId, + runId: firstRun.id, + occurredAt: yield* DateTime.now, + payload: { + ...firstRun, + delegatedCompletion: { + disposition: "open", + nextGeneration: 2, + delivery: { + messageId: MessageId.make("message:completion"), + generation: 1, + taskIds, + }, + }, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("completion"), + threadId, + messageId: MessageId.make("message:completion"), + text: "Completion", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { parentRunId: firstRun.id, generation: 1, taskIds }, + createdBy: "agent", + creationSource: "server", + }); + const runFor = (key: string) => + Effect.map(orchestrator.getThreadProjection(threadId), ({ runs }) => + runs.find((run) => run.userMessageId === MessageId.make(`message:${key}`))!, + ); + yield* send("correction", "correction", "queue"); + // A goal is moved ahead of the completion too, so promoting it must drop that priority. + for (const key of promoted === "next" ? ["correction"] : ["next", "correction"]) { + yield* orchestrator.dispatch({ + type: "queued-run.reorder", + commandId: CommandId.make(`reorder-${key}`), + threadId, + runId: (yield* runFor(key)).id, + beforeRunId: (yield* runFor("completion")).id, + }); + } + const promotedStarted = yield* nextStart; yield* orchestrator.dispatch({ type: "queued-message.promote-to-steer", commandId: CommandId.make("promote"), threadId, - queuedRunId: nextRun.id, + queuedRunId: (yield* runFor("next")).id, targetRunId: compactRun.id, }); yield* worker.drain(); + yield* Fiber.join(promotedStarted); const projection = yield* orchestrator.getThreadProjection(threadId); assert.deepEqual(steered, []); @@ -1707,10 +1772,18 @@ it.effect.each(["next", "/goal ship it"])( (item) => item.type === "run_interrupt_request" && item.runId === compactRun.id, ), ); - assert.deepEqual((yield* read(threadId)).queue, ["queued"]); + yield* finish(2); + // A plain message leaves the reordered correction ahead of the completion; a goal, like + // any queued goal, goes after automatic completions and ahead of other messages. assert.deepEqual( - started.map((turn) => turn.message.text), - ["first", "/compact", promoted], + started.map((turn) => turn.runId), + [ + started[0]!.runId, + compactRun.id, + ...(promoted === "next" + ? [(yield* runFor("correction")).id, (yield* runFor("completion")).id] + : [(yield* runFor("completion")).id, (yield* runFor("next")).id]), + ], ); }).pipe(Effect.provide(layer)); }), diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b38fb737c91db..7009c84b6a0ff 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -22,6 +22,7 @@ import { type OrchestrationV2Run, type OrchestrationV2ShellSnapshot, type OrchestrationV2RuntimeRequest, + type OrchestrationV2ThreadProjection, ProjectId, type PullRequestDetail, type PullRequestComment, @@ -63,6 +64,7 @@ import * as VcsProcess from "../vcs/VcsProcess.ts"; import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import { queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts"; import { ROLLBACK_FAILED_MESSAGE } from "./CheckpointRollbackService.ts"; import * as EffectWorker from "./EffectWorker.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; @@ -6608,6 +6610,154 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { assert.equal(cancelAgainError._tag, "OrchestratorDispatchError"); }), ); + + it.effect("moves a queued message ahead of queued delegated completions and back", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("runtime-layer-reorder-before-completion-thread"); + const messageId = (key: string) => + MessageId.make(`runtime-layer-reorder-before-completion-${key}`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-reorder-before-completion-create"), + threadId, + projectId: ProjectId.make("runtime-layer-reorder-before-completion-project"), + title: "Reorder before completions", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/runtime-layer-reorder-before-completion", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-reorder-before-completion-active"), + threadId, + messageId: messageId("active"), + text: "Keep the provider occupied.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const queueCompletion = (generation: number) => + Effect.gen(function* () { + const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const taskIds = [ + NodeId.make(`runtime-layer-reorder-before-completion-task-${generation}`), + ]; + yield* eventSink.write({ + events: [ + { + id: EventId.make(`runtime-layer-reorder-before-completion-open-${generation}`), + type: "run.updated", + threadId, + runId: activeRun.id, + occurredAt: yield* DateTime.now, + payload: { + ...activeRun, + delegatedCompletion: { + disposition: "open", + nextGeneration: generation + 1, + delivery: { + messageId: messageId(`completion-${generation}`), + generation, + taskIds, + }, + }, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make(`runtime-layer-reorder-before-completion-${generation}`), + threadId, + messageId: messageId(`completion-${generation}`), + text: `Completion ${generation}`, + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { parentRunId: activeRun.id, generation, taskIds }, + }); + }); + yield* queueCompletion(1); + yield* queueCompletion(2); + for (const key of ["correction", "follow-up"]) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`runtime-layer-reorder-before-completion-${key}`), + threadId, + messageId: messageId(key), + text: key, + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + } + const runIdFor = (projection: OrchestrationV2ThreadProjection, key: string) => + projection.runs.find((run) => run.userMessageId === messageId(key))!.id; + const before = yield* orchestrator.getThreadProjection(threadId); + + yield* orchestrator.dispatch({ + type: "queued-run.reorder", + commandId: CommandId.make("runtime-layer-reorder-before-completion-reorder"), + threadId, + runId: runIdFor(before, "correction"), + beforeRunId: runIdFor(before, "completion-1"), + }); + + yield* queueCompletion(3); + + const deliveryOrder = Effect.gen(function* () { + const projection = yield* orchestrator.getThreadProjection(threadId); + return queuedRunsInDeliveryOrder(projection).map((run) => + run.userMessageId.replace(messageId(""), ""), + ); + }); + assert.deepEqual(yield* deliveryOrder, [ + "correction", + "completion-1", + "completion-2", + "completion-3", + "follow-up", + ]); + + const completionReorder = yield* orchestrator + .dispatch({ + type: "queued-run.reorder", + commandId: CommandId.make("runtime-layer-reorder-before-completion-completion"), + threadId, + runId: runIdFor(before, "completion-2"), + beforeRunId: runIdFor(before, "completion-1"), + }) + .pipe(Effect.flip); + assert.equal(completionReorder._tag, "OrchestratorDispatchError"); + + yield* orchestrator.dispatch({ + type: "queued-run.reorder", + commandId: CommandId.make("runtime-layer-reorder-before-completion-back"), + threadId, + runId: runIdFor(before, "correction"), + beforeRunId: null, + }); + assert.deepEqual(yield* deliveryOrder, [ + "completion-1", + "completion-2", + "completion-3", + "follow-up", + "correction", + ]); + }), + ); }); it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", (it) => { diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 785f41e3aa0fb..b335b9bd5803a 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -686,6 +686,8 @@ export const OrchestrationV2Run = Schema.Struct({ status: OrchestrationV2RunStatus, queuePosition: Schema.optional(Schema.NullOr(PositiveInt)), queueHeld: Schema.optional(Schema.Boolean), + /** Set by a reorder that placed this message before queued automatic completions. */ + queuedAheadOfCompletions: Schema.optional(Schema.Boolean), requestedAt: Schema.DateTimeUtc, startedAt: Schema.NullOr(Schema.DateTimeUtc), completedAt: Schema.NullOr(Schema.DateTimeUtc),