fix(server): queued runs land on the timeline where they started

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

@@ -306,28 +306,17 @@ const layerBase: Layer.Layer<
         });
       });
 
-    const normalizeEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) => {
-      const runOrdinals = new Map(
-        events.flatMap((event) =>
-          event.type === "run.created" || event.type === "run.updated"
-            ? [[event.payload.id, event.payload.ordinal] as const]
-            : [],
-        ),
-      );
-      return Effect.forEach(
+    const normalizeEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) =>
+      Effect.forEach(
         events,
         (event): Effect.Effect<OrchestrationV2DomainEvent, unknown> =>
           event.type === "turn-item.updated"
             ? turnItemPositions
-                .normalize(
-                  event.payload,
-                  event.payload.runId === null ? undefined : runOrdinals.get(event.payload.runId),
-                )
+                .normalize(event.payload)
                 .pipe(Effect.map((payload) => ({ ...event, payload })))
             : Effect.succeed(event),
         { concurrency: 1 },
       );
-    };
 
     const applyStoredEvents = (storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>) =>
       Effect.gen(function* () {

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

@@ -141,6 +141,7 @@ export const layer: Layer.Layer<
           yield* sql`DELETE FROM orchestration_v2_projection_runs`;
           yield* sql`DELETE FROM orchestration_v2_projection_threads`;
           yield* sql`DELETE FROM orchestration_v2_turn_item_positions`;
+          yield* sql`DELETE FROM orchestration_v2_run_timeline_bands`;
 
           const pageSize = 500;
           let lastSequence = 0;
@@ -171,6 +172,16 @@ export const layer: Layer.Layer<
             if (page.length < pageSize) break;
             yield* Effect.yieldNow;
           }
+          // Stored ordinals are already positions, so each run keeps the band
+          // its items were written in. Same backfill as migration 061; the
+          // literal is TurnItemPositionStore's BAND_SIZE (integer division).
+          yield* sql`
+            INSERT INTO orchestration_v2_run_timeline_bands (thread_id, run_id, band)
+            SELECT thread_id, run_id, MAX(ordinal / 1000000)
+            FROM orchestration_v2_projection_turn_items
+            WHERE run_id IS NOT NULL
+            GROUP BY thread_id, run_id
+          `;
           const now = DateTime.formatIso(yield* DateTime.now);
           yield* sql`
             INSERT INTO orchestration_v2_projection_metadata (

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

@@ -16,16 +16,17 @@ export class TurnItemPositionStoreError extends Schema.TaggedError<TurnItemPosit
 
 const isTurnItemPositionStoreError = Schema.is(TurnItemPositionStoreError);
 
+/** Positions per band; a turn item's band is `Math.floor(position / BAND_SIZE)`. */
+const BAND_SIZE = 1_000_000;
+
 export interface TurnItemPositionStoreV2Shape {
   readonly allocate: (input: {
     readonly threadId: ThreadId;
     readonly turnItemId: TurnItemId;
     readonly runId: RunId | null;
-    readonly runOrdinal?: number;
   }) => Effect.Effect<number, TurnItemPositionStoreError>;
   readonly normalize: (
     item: OrchestrationV2TurnItem,
-    runOrdinal?: number,
   ) => Effect.Effect<OrchestrationV2TurnItem, TurnItemPositionStoreError>;
 }
 
@@ -39,25 +40,36 @@ export const layer: Layer.Layer<TurnItemPositionStoreV2, never, SqlClient.SqlCli
   Effect.gen(function* () {
     const sql = yield* SqlClient.SqlClient;
 
-    const allocate: TurnItemPositionStoreV2Shape["allocate"] = ({
-      threadId,
-      turnItemId,
-      runId,
-      runOrdinal: suppliedRunOrdinal,
-    }) =>
+    // A thread's timeline is split into bands of BAND_SIZE positions. Runless
+    // items share band 0. A run gets the next free band when it writes its
+    // first item, which happens when it starts, so runs sort by when they
+    // started rather than by their ordinal, which is fixed when they are queued.
+    const allocate: TurnItemPositionStoreV2Shape["allocate"] = ({ threadId, turnItemId, runId }) =>
       Effect.gen(function* () {
-        const runRows =
-          runId === null
-            ? []
-            : yield* sql<{ readonly ordinal: number }>`
-                SELECT ordinal
-                FROM orchestration_v2_projection_runs
-                WHERE thread_id = ${threadId} AND run_id = ${runId}
-                LIMIT 1
-              `;
-        const runOrdinal = suppliedRunOrdinal ?? runRows[0]?.ordinal ?? null;
-        const lowerBound = runOrdinal === null ? 0 : runOrdinal * 1_000_000;
-        const upperBound = runOrdinal === null ? 999_999 : lowerBound + 999_999;
+        const selectBand = (id: RunId) => sql<{ readonly band: number }>`
+          SELECT band
+          FROM orchestration_v2_run_timeline_bands
+          WHERE thread_id = ${threadId} AND run_id = ${id}
+          LIMIT 1
+        `;
+        let band = 0;
+        if (runId !== null) {
+          // Most calls update an item of a run that already has its band.
+          let bandRows = yield* selectBand(runId);
+          if (bandRows[0] === undefined) {
+            yield* sql`
+              INSERT INTO orchestration_v2_run_timeline_bands (thread_id, run_id, band)
+              SELECT ${threadId}, ${runId}, COALESCE(MAX(band), 0) + 1
+              FROM orchestration_v2_run_timeline_bands
+              WHERE thread_id = ${threadId}
+              ON CONFLICT(thread_id, run_id) DO NOTHING
+            `;
+            bandRows = yield* selectBand(runId);
+          }
+          band = bandRows[0]?.band ?? 0;
+        }
+        const lowerBound = band * BAND_SIZE;
+        const upperBound = lowerBound + BAND_SIZE - 1;
         yield* sql`
           INSERT INTO orchestration_v2_turn_item_positions (thread_id, turn_item_id, ordinal)
           SELECT
@@ -95,13 +107,10 @@ export const layer: Layer.Layer<TurnItemPositionStoreV2, never, SqlClient.SqlCli
 
     return TurnItemPositionStoreV2.of({
       allocate,
-      normalize: (item, runOrdinal) =>
-        allocate({
-          threadId: item.threadId,
-          turnItemId: item.id,
-          runId: item.runId,
-          ...(runOrdinal === undefined ? {} : { runOrdinal }),
-        }).pipe(Effect.map((ordinal) => (item.ordinal === ordinal ? item : { ...item, ordinal }))),
+      normalize: (item) =>
+        allocate({ threadId: item.threadId, turnItemId: item.id, runId: item.runId }).pipe(
+          Effect.map((ordinal) => (item.ordinal === ordinal ? item : { ...item, ordinal })),
+        ),
     });
   }),
 );

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

@@ -4186,6 +4186,106 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
     }),
   );
 
+  it.effect("places a queued run that starts late after the runs that started before it", () =>
+    Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const projections = yield* ProjectionStore.ProjectionStoreV2;
+      const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
+      const threadId = ThreadId.make("runtime-layer-late-queued-run");
+      yield* orchestrator.dispatch({
+        type: "thread.create",
+        createdBy: "user",
+        creationSource: "web",
+        commandId: CommandId.make(`${threadId}:create`),
+        threadId,
+        projectId: ProjectId.make(`${threadId}:project`),
+        title: "Late queued run",
+        modelSelection,
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        branch: null,
+        worktreePath: process.cwd(),
+      });
+      const send = (key: string, dispatchMode: "start_immediately" | "queue_after_active") =>
+        orchestrator.dispatch({
+          type: "message.dispatch",
+          createdBy: "user",
+          creationSource: "web",
+          commandId: CommandId.make(`${threadId}:message:${key}`),
+          threadId,
+          messageId: MessageId.make(`${threadId}:message:${key}`),
+          text: key,
+          attachments: [],
+          modelSelection,
+          dispatchMode: { type: dispatchMode },
+        });
+      // A runs; B is queued behind it and held when A is stopped.
+      yield* send("A", "start_immediately");
+      yield* send("B", "queue_after_active");
+      const queued = yield* orchestrator.getThreadProjection(threadId);
+      const [runA, runB] = queued.runs;
+      yield* orchestrator.dispatch({
+        type: "run.interrupt",
+        commandId: CommandId.make(`${threadId}:interrupt`),
+        threadId,
+        runId: runA!.id,
+        holdQueue: true,
+      });
+      // C is sent later, so it has a higher run ordinal, but it starts first.
+      yield* send("C", "start_immediately");
+      const started = yield* orchestrator.getThreadProjection(threadId);
+      const runC = started.runs.find((run) => run.userMessageId === `${threadId}:message:C`)!;
+      assert.isAbove(runC.ordinal, runB!.ordinal);
+      assert.equal(started.runs.find((run) => run.id === runB!.id)?.status, "queued");
+      yield* orchestrator.dispatch({
+        type: "run.interrupt",
+        commandId: CommandId.make(`${threadId}:interrupt-c`),
+        threadId,
+        runId: runC.id,
+      });
+      yield* orchestrator.dispatch({
+        type: "queue.resume",
+        commandId: CommandId.make(`${threadId}:resume`),
+        threadId,
+      });
+
+      const userMessages = (
+        items: ReadonlyArray<{ readonly type: string; readonly text?: string }>,
+      ) => items.flatMap((item) => (item.type === "user_message" ? [item.text] : []));
+      const projection = yield* orchestrator.getThreadProjection(threadId);
+      assert.notEqual(projection.runs.find((run) => run.id === runB!.id)?.status, "queued");
+      assert.deepEqual(userMessages(projection.turnItems), ["A", "C", "B"]);
+      // Every item of B sorts after every item of C.
+      const lastOfC = Math.max(
+        ...projection.turnItems.filter((item) => item.runId === runC.id).map((i) => i.ordinal),
+      );
+      assert.isTrue(
+        projection.turnItems
+          .filter((item) => item.runId === runB!.id)
+          .every((item) => item.ordinal > lastOfC),
+      );
+      const window = yield* projections.getThreadSnapshotWindow(threadId, {
+        rowLimit: 100,
+        userTurnLimit: 2,
+      });
+      assert.deepEqual(userMessages(window.projection.turnItems), ["A", "C", "B"]);
+
+      // A rebuild from the stored events keeps the order, and B's band stays above C's.
+      yield* maintenance.rebuild;
+      const rebuilt = yield* orchestrator.getThreadProjection(threadId);
+      assert.deepEqual(userMessages(rebuilt.turnItems), ["A", "C", "B"]);
+      yield* orchestrator.dispatch({
+        type: "run.interrupt",
+        commandId: CommandId.make(`${threadId}:interrupt-b`),
+        threadId,
+        runId: runB!.id,
+      });
+      yield* send("D", "start_immediately");
+      const appended = yield* orchestrator.getThreadProjection(threadId);
+      assert.deepEqual(userMessages(appended.turnItems), ["A", "C", "B", "D"]);
+    }),
+  );
+
   it.effect.each(["startup", "shutdown"] as const)(
     "preserves and holds queued messages across %s until explicitly resumed",
     (trigger) =>

apps/server/src/persistence/Migrations.ts

@@ -74,6 +74,7 @@ import Migration0057 from "./Migrations/057_ScheduledTaskWebhooks.ts";
 import Migration0058 from "./Migrations/058_WebhookRelayDeliveries.ts";
 import Migration0059 from "./Migrations/059_McpAppModelContext.ts";
 import Migration0060 from "./Migrations/060_ThreadSnapshotWindowIndexes.ts";
+import Migration0061 from "./Migrations/061_RunTimelineBands.ts";
 
 /**
  * Migration loader with all migrations defined inline.
@@ -148,6 +149,7 @@ export const migrationEntries = [
   [58, "WebhookRelayDeliveries", Migration0058],
   [59, "McpAppModelContext", Migration0059],
   [60, "ThreadSnapshotWindowIndexes", Migration0060],
+  [61, "RunTimelineBands", Migration0061],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts

@@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => {
     Effect.sync(() => {
       assert.deepStrictEqual(
         migrationEntries.map(([id]) => id),
-        Array.from({ length: 60 }, (_, index) => index + 1),
+        Array.from({ length: 61 }, (_, index) => index + 1),
       );
     }),
   );
@@ -32,6 +32,7 @@ layer("055_OrchestrationV2", (it) => {
         [58, "WebhookRelayDeliveries"],
         [59, "McpAppModelContext"],
         [60, "ThreadSnapshotWindowIndexes"],
+        [61, "RunTimelineBands"],
       ]);
       assert.deepStrictEqual(yield* runMigrations(), []);
 
@@ -58,6 +59,7 @@ layer("055_OrchestrationV2", (it) => {
         { migration_id: 58, name: "WebhookRelayDeliveries" },
         { migration_id: 59, name: "McpAppModelContext" },
         { migration_id: 60, name: "ThreadSnapshotWindowIndexes" },
+        { migration_id: 61, name: "RunTimelineBands" },
       ]);
 
       const tables = yield* sql<{ readonly name: string }>`

apps/server/src/persistence/Migrations/061_RunTimelineBands.test.ts

@@ -0,0 +1,84 @@
+import { assert, it } from "@effect/vitest";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as SqlClient from "effect/sql/SqlClient";
+import { RunId, ThreadId, TurnItemId } from "@t3tools/contracts";
+
+import * as TurnItemPositionStore from "../../orchestration-v2/TurnItemPositionStore.ts";
+import { runMigrations } from "../Migrations.ts";
+
+const layer = it.layer(
+  Layer.mergeAll(
+    TurnItemPositionStore.layer.pipe(
+      Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })),
+    ),
+  ),
+);
+
+layer("061_RunTimelineBands", (it) => {
+  it.effect("keeps stored item order and starts new runs above it", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      const positions = yield* TurnItemPositionStore.TurnItemPositionStoreV2;
+      yield* runMigrations({ toMigrationInclusive: 60 });
+      const threadId = ThreadId.make("thread:bands");
+      // Run 2 was queued first but its items were stored in band 2; run 3 has
+      // items in band 3. Run 4 is still queued and has no items.
+      const stored = [
+        ["item:1", "run:1", 1_000_001],
+        ["item:2", "run:2", 2_000_001],
+        ["item:3", "run:3", 3_000_001],
+        ["item:3b", "run:3", 3_000_002],
+      ] as const;
+      for (const [itemId, runId, ordinal] of stored) {
+        yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert({
+          turn_item_id: itemId,
+          thread_id: threadId,
+          run_id: runId,
+          ordinal,
+          type: "user_message",
+          status: "completed",
+          updated_at: "2026-10-01T00:00:00.000Z",
+          payload_json: "{}",
+        })}`;
+        yield* sql`INSERT INTO orchestration_v2_turn_item_positions ${sql.insert({
+          thread_id: threadId,
+          turn_item_id: itemId,
+          ordinal,
+        })}`;
+      }
+
+      yield* runMigrations();
+      assert.deepStrictEqual(
+        yield* sql`
+          SELECT run_id, band FROM orchestration_v2_run_timeline_bands ORDER BY band
+        `,
+        [
+          { run_id: "run:1", band: 1 },
+          { run_id: "run:2", band: 2 },
+          { run_id: "run:3", band: 3 },
+        ],
+      );
+
+      // An existing run keeps appending in its band.
+      assert.strictEqual(
+        yield* positions.allocate({
+          threadId,
+          turnItemId: TurnItemId.make("item:3c"),
+          runId: RunId.make("run:3"),
+        }),
+        3_000_003,
+      );
+      // A queued run that starts now goes above every existing band.
+      assert.strictEqual(
+        yield* positions.allocate({
+          threadId,
+          turnItemId: TurnItemId.make("item:4"),
+          runId: RunId.make("run:4"),
+        }),
+        4_000_001,
+      );
+    }),
+  );
+});