fix(server): a held queue no longer strands later queued messages

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

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -4687,11 +4687,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           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<ProjectionStoreV2> = 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",
             )
           );
         }),

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;