fix(server): queued messages no longer wait behind every later delegate completion

apps/server/src/orchestration-v2/QueuedRunOrder.test.ts

@@ -1,51 +1,65 @@
+import * as DateTime from "effect/DateTime";
 import { describe, expect, it } from "vite-plus/test";
 
 import { queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts";
 
+const at = (time: string) => DateTime.makeUnsafe(`2026-01-01T${time}Z`);
+const delegatedCompletion = {
+  generation: 1,
+  parentRunId: "run:parent",
+  taskIds: ["task:child"],
+};
+
 describe("queued run delivery order", () => {
-  it("keeps automatic completion delivery ahead of visible queued messages", () => {
+  it("delivers completions and queued messages in request order, keeping manual queue order", () => {
     const projection = {
       messages: [
-        { id: "message:visible-first" },
-        {
-          id: "message:automatic",
-          delegatedCompletion: {
-            generation: 1,
-            parentRunId: "run:parent",
-            taskIds: ["task:child"],
-          },
-        },
-        { id: "message:visible-second" },
+        { id: "message:old" },
+        { id: "message:moved" },
+        { id: "message:earlier-completion", delegatedCompletion },
+        { id: "message:later-completion", delegatedCompletion },
       ],
       runs: [
         {
-          id: "run:visible-first",
-          ordinal: 2,
-          queuePosition: 1,
+          id: "run:old",
+          ordinal: 3,
+          queuePosition: 2,
+          requestedAt: at("10:00:00"),
           status: "queued",
-          userMessageId: "message:visible-first",
+          userMessageId: "message:old",
         },
         {
-          id: "run:automatic",
+          id: "run:moved",
           ordinal: 4,
+          queuePosition: 1,
+          requestedAt: at("10:02:00"),
+          status: "queued",
+          userMessageId: "message:moved",
+        },
+        {
+          id: "run:earlier-completion",
+          ordinal: 2,
           queuePosition: 3,
+          requestedAt: at("09:59:00"),
           status: "queued",
-          userMessageId: "message:automatic",
+          userMessageId: "message:earlier-completion",
         },
         {
-          id: "run:visible-second",
-          ordinal: 3,
-          queuePosition: 2,
+          id: "run:later-completion",
+          ordinal: 5,
+          queuePosition: 4,
+          requestedAt: at("10:03:00"),
           status: "queued",
-          userMessageId: "message:visible-second",
+          userMessageId: "message:later-completion",
         },
       ],
     } as never;
 
     expect(queuedRunsInDeliveryOrder(projection).map((run) => run.id)).toEqual([
-      "run:automatic",
-      "run:visible-first",
-      "run:visible-second",
+      "run:earlier-completion",
+      "run:moved",
+      "run:old",
+      "run:later-completion",
     ]);
   });
 });

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

@@ -1,4 +1,5 @@
 import type { OrchestrationV2Run, OrchestrationV2ThreadProjection } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
 
 export function isAutomaticCompletionRun(
   projection: Pick<OrchestrationV2ThreadProjection, "runs" | "messages">,
@@ -17,16 +18,34 @@ export function queuedRunsInDeliveryOrder(
       .filter((message) => message.delegatedCompletion !== undefined)
       .map((message) => message.id),
   );
-  return projection.runs
+  const queued = projection.runs
     .filter((run) => run.status === "queued")
-    .toSorted((left, right) => {
-      const deliveryPriority =
-        Number(automaticCompletionMessageIds.has(right.userMessageId)) -
-        Number(automaticCompletionMessageIds.has(left.userMessageId));
-      return (
-        deliveryPriority ||
+    .toSorted(
+      (left, right) =>
         (left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal) ||
-        left.ordinal - right.ordinal
-      );
-    });
+        left.ordinal - right.ordinal,
+    );
+  const completions = queued.filter((run) => automaticCompletionMessageIds.has(run.userMessageId));
+  const messages = queued.filter((run) => !automaticCompletionMessageIds.has(run.userMessageId));
+  // Completions and messages each keep their queue order and interleave by request time,
+  // so neither kind can hold the other back while more of it keeps arriving.
+  const ordered: Array<OrchestrationV2Run> = [];
+  let completionIndex = 0;
+  let messageIndex = 0;
+  for (;;) {
+    const completion = completions[completionIndex];
+    const message = messages[messageIndex];
+    if (completion === undefined || message === undefined) {
+      return [...ordered, ...completions.slice(completionIndex), ...messages.slice(messageIndex)];
+    }
+    if (
+      DateTime.toEpochMillis(completion.requestedAt) <= DateTime.toEpochMillis(message.requestedAt)
+    ) {
+      ordered.push(completion);
+      completionIndex += 1;
+    } else {
+      ordered.push(message);
+      messageIndex += 1;
+    }
+  }
 }