diff --git a/apps/server/src/orchestration-v2/QueuedRunOrder.test.ts b/apps/server/src/orchestration-v2/QueuedRunOrder.test.ts index 838ec141a4b50..9e12153771fab 100644 --- a/apps/server/src/orchestration-v2/QueuedRunOrder.test.ts +++ b/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", ]); }); }); diff --git a/apps/server/src/orchestration-v2/QueuedRunOrder.ts b/apps/server/src/orchestration-v2/QueuedRunOrder.ts index 72c5e319d69ed..1635f4ab51674 100644 --- a/apps/server/src/orchestration-v2/QueuedRunOrder.ts +++ b/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, @@ -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 = []; + 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; + } + } }