From 18f3f50b8b19b8ae442197aac29ceef76e43d1ed Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Sat, 10 Oct 2026 21:12:49 -0700 Subject: [PATCH 1/2] fix(server): queued runs land on the timeline where they started A run's turn items were banded by its run ordinal, which is fixed when the run is queued. A message queued before a steer therefore rendered above the steer even though the steer's run started first (#16987). Each run now gets its own timeline band the first time it writes an item, which is when it starts. TurnItemPositionStore owns the new orchestration_v2_run_timeline_bands table. Migration 061 and the projection rebuild both derive a run's band from the positions its items already hold, so existing threads keep their order. The run ordinal itself is unchanged. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 17 +-- .../orchestration-v2/ProjectionMaintenance.ts | 11 ++ .../orchestration-v2/TurnItemPositionStore.ts | 63 ++++++----- .../src/orchestration-v2/runtimeLayer.test.ts | 100 ++++++++++++++++++ apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/055_OrchestrationV2.test.ts | 4 +- .../Migrations/061_RunTimelineBands.test.ts | 84 +++++++++++++++ .../Migrations/061_RunTimelineBands.ts | 28 +++++ .../reconcileV2PreviewMigration.test.ts | 2 + 9 files changed, 269 insertions(+), 42 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/061_RunTimelineBands.test.ts create mode 100644 apps/server/src/persistence/Migrations/061_RunTimelineBands.ts diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 039d1872da6f..c64bdaa1d9de 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -306,28 +306,17 @@ const layerBase: Layer.Layer< }); }); - const normalizeEvents = (events: ReadonlyArray) => { - 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) => + Effect.forEach( events, (event): Effect.Effect => 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) => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 8c10de314202..e9e3fffd9e42 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/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.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 ( diff --git a/apps/server/src/orchestration-v2/TurnItemPositionStore.ts b/apps/server/src/orchestration-v2/TurnItemPositionStore.ts index 016fe4f84017..0022b4bbe29a 100644 --- a/apps/server/src/orchestration-v2/TurnItemPositionStore.ts +++ b/apps/server/src/orchestration-v2/TurnItemPositionStore.ts @@ -16,16 +16,17 @@ export class TurnItemPositionStoreError extends Schema.TaggedError Effect.Effect; readonly normalize: ( item: OrchestrationV2TurnItem, - runOrdinal?: number, ) => Effect.Effect; } @@ -39,25 +40,36 @@ export const layer: Layer.Layer + // 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 - 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 })), + ), }); }), ); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 4d3ff429f0c6..c0fd0fa4510f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/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) => diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 35c1a3fdce4b..4f67e5a27c1d 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/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); diff --git a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts index 5e45ed05d302..81652d0a8938 100644 --- a/apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts +++ b/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 }>` diff --git a/apps/server/src/persistence/Migrations/061_RunTimelineBands.test.ts b/apps/server/src/persistence/Migrations/061_RunTimelineBands.test.ts new file mode 100644 index 000000000000..d362863b97f1 --- /dev/null +++ b/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, + ); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/061_RunTimelineBands.ts b/apps/server/src/persistence/Migrations/061_RunTimelineBands.ts new file mode 100644 index 000000000000..20effb4e0486 --- /dev/null +++ b/apps/server/src/persistence/Migrations/061_RunTimelineBands.ts @@ -0,0 +1,28 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Each run's turn items sit in one band of a thread's timeline, assigned when + // the run writes its first item. TurnItemPositionStore owns this table. + yield* sql` + CREATE TABLE IF NOT EXISTS orchestration_v2_run_timeline_bands ( + thread_id TEXT NOT NULL, + run_id TEXT NOT NULL, + band INTEGER NOT NULL, + PRIMARY KEY (thread_id, run_id) + ) + `; + // Runs that already wrote items keep the band their positions were allocated + // in, so stored items keep their order. Runs without items get a band when + // they start. + 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 + ON CONFLICT(thread_id, run_id) DO NOTHING + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 675d5f5db021..090401d50e5d 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -41,6 +41,7 @@ describe("V2 preview upgrade", () => { [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], [60, "ThreadSnapshotWindowIndexes"], + [61, "RunTimelineBands"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -124,6 +125,7 @@ describe("V2 preview upgrade", () => { [58, "WebhookRelayDeliveries"], [59, "McpAppModelContext"], [60, "ThreadSnapshotWindowIndexes"], + [61, "RunTimelineBands"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From bafb06dc1769f43a10618c7d7b962f1504a4fbc0 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Sat, 10 Oct 2026 21:24:41 -0700 Subject: [PATCH 2/2] refactor(server): keep the timeline band size private to its store Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/ProjectionMaintenance.ts | 2 +- apps/server/src/orchestration-v2/TurnItemPositionStore.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index e9e3fffd9e42..eeedc4f93c17 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -174,7 +174,7 @@ export const layer: Layer.Layer< } // 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.BAND_SIZE (integer division). + // 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) diff --git a/apps/server/src/orchestration-v2/TurnItemPositionStore.ts b/apps/server/src/orchestration-v2/TurnItemPositionStore.ts index 0022b4bbe29a..6b547c16aa85 100644 --- a/apps/server/src/orchestration-v2/TurnItemPositionStore.ts +++ b/apps/server/src/orchestration-v2/TurnItemPositionStore.ts @@ -17,7 +17,7 @@ export class TurnItemPositionStoreError extends Schema.TaggedError