fix(server): allow queued messages ahead of completion deliveries

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,19 +9699,26 @@ 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(
       reordered,
       (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;

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<OrchestrationV2ThreadProjection, "runs" | "messages">,
 ): ReadonlyArray<OrchestrationV2Run> {
@@ -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) ||

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));
       }),