From c1d4a5873821dd0a37c8c4150e189c0d4f3d0dc6 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 11:58:42 +0000 Subject: [PATCH 1/7] perf(server): message ordinals are read through an index, not a scan Ordering a thread's messages by their first user or assistant turn item grouped every such item by its message id through a temporary B-tree, so the read grew with the thread. Migration 80 adds a partial expression index on the thread, message id and ordinal of user and assistant items with a valid JSON payload, and the ordinal read gains the matching json_valid term. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 192 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 3 + apps/server/src/persistence/Migrations.ts | 2 + .../080_TurnItemMessageOrdinalIndex.test.ts | 138 +++++++++++++ .../080_TurnItemMessageOrdinalIndex.ts | 20 ++ .../reconcileV2PreviewMigration.test.ts | 2 + docs/internals/performance-regressions.md | 7 + 7 files changed, 364 insertions(+) create mode 100644 apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts create mode 100644 apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index c6187a97ea8d..f2f12cce6041 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1682,6 +1682,198 @@ const subagentRecordsSelectById = Effect.gen(function* () { assert.deepEqual(yield* read([]), []); }); +const recordsAt = (second: number) => DateTime.makeUnsafe(Date.UTC(2026, 0, 1, 0, 0, second)); +const recordsThreadCreated = (threadId: ThreadId) => + ({ + id: EventId.make(`event:${threadId}:created`), + type: "thread.created", + threadId, + providerInstanceId, + occurredAt: recordsAt(0), + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:records"), + title: threadId, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: recordsAt(0), + updatedAt: recordsAt(0), + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }) as never; + +const queryPlan = Effect.fn("queryPlan")(function* (statement: string | undefined) { + assert.isDefined(statement); + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql.unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${statement}`); + return rows.map((row) => row.detail); +}); + +// Each message's first item orders it, not the stored message order (the assistant reply is stored +// first); the message-ordinal statement reads those ordinals through migration 80's index. +const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread:message-ordinals"); + yield* store.apply(recordsThreadCreated(threadId)); + const conversation = (name: string, firstOrdinal: number) => { + const runId = RunId.make(`run:${name}`); + const nodeId = NodeId.make(`node:${name}:root`); + const user = MessageId.make(`message:${name}:user`); + const assistant = MessageId.make(`message:${name}:assistant`); + const item = (index: number) => ({ + id: TurnItemId.make(`item:${name}:${index}`), + threadId, + runId, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: firstOrdinal + index, + status: "completed", + title: null, + startedAt: recordsAt(2), + completedAt: recordsAt(2), + updatedAt: recordsAt(2), + }); + const message = (id: MessageId, role: "user" | "assistant", second: number) => ({ + id: EventId.make(`event:${id}`), + type: "message.updated", + threadId, + occurredAt: recordsAt(second), + payload: { + createdBy: role === "user" ? "user" : "agent", + creationSource: role === "user" ? "web" : "provider", + id, + threadId, + runId, + nodeId, + role, + text: `${name} ${role}`, + attachments: [], + streaming: false, + createdAt: recordsAt(second), + updatedAt: recordsAt(second), + }, + }); + const turnItem = (index: number, payload: object) => ({ + id: EventId.make(`event:item:${name}:${index}`), + type: "turn-item.updated", + threadId, + occurredAt: recordsAt(2), + payload: { ...item(index), ...payload }, + }); + const events = [ + message(assistant, "assistant", 1), + message(user, "user", 2), + turnItem(0, { + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId: user, + inputIntent: "turn_start", + text: `${name} user`, + attachments: [], + }), + // A message streamed over two items keeps its first ordinal. + ...[1, 2].map((index) => + turnItem(index, { + type: "assistant_message", + messageId: assistant, + text: `${name} assistant ${index}`, + streaming: false, + }), + ), + turnItem(3, { + type: "dynamic_tool", + toolName: "tool", + input: {}, + output: { messageId: user }, + }), + ]; + return { events, user, assistant }; + }; + const first = conversation("first", 1); + const second = conversation("second", 10); + for (const event of [...first.events, ...second.events]) yield* store.apply(event as never); + + const { statements, tracer } = traceSqlStatements(); + const indexed = yield* store + .getThreadRecords(threadId, ["messages"]) + .pipe(Effect.withTracer(tracer)); + assert.deepEqual( + yield* queryPlan(statements.find((statement) => statement.includes("MIN(ordinal) AS ordinal"))), + [ + "SEARCH orchestration_v2_projection_turn_items USING INDEX orchestration_v2_projection_turn_items_message_ordinal_idx (thread_id=?)", + ], + ); + assert.deepEqual( + indexed.messages.map((message) => message.id), + [first.user, first.assistant, second.user, second.assistant], + ); + // A narrow read (ids, runs or one role) recovers only its own messages' ordinals, through the + // same index; the statement it sends, rerun with its bindings, returns just those rows. + const narrowReads = [ + { + filter: { messageIds: [second.assistant, first.user] }, + loaded: [first.user, second.assistant], + }, + { filter: { messageRunIds: [RunId.make("run:first")] }, loaded: [first.user, first.assistant] }, + { filter: { messageRoles: ["user" as const] }, loaded: [first.user, second.user] }, + ]; + for (const { filter, loaded } of narrowReads) { + const narrow = traceSqlStatements(); + const narrowed = yield* store + .getThreadRecords(threadId, ["messages"], filter) + .pipe(Effect.withTracer(narrow.tracer)); + assert.deepEqual( + narrowed.messages.map((message) => message.id), + loaded, + ); + const bounded = narrow.statements.find((statement) => + statement.includes("MIN(ordinal) AS ordinal"), + ); + assert.include(bounded, "IN (SELECT value FROM json_each("); + assert.deepEqual(yield* queryPlan(bounded), [ + "SEARCH orchestration_v2_projection_turn_items USING INDEX orchestration_v2_projection_turn_items_message_ordinal_idx (thread_id=? AND =?)", + "LIST SUBQUERY 1", + "SCAN json_each VIRTUAL TABLE INDEX 1:", + ]); + const rows = yield* sql.unsafe<{ readonly message_id: string }>(bounded!, [ + threadId, + JSON.stringify(loaded), + ]); + assert.sameMembers( + rows.map((row) => row.message_id), + loaded, + ); + } + yield* sql`DROP INDEX orchestration_v2_projection_turn_items_message_ordinal_idx`; + assert.deepEqual( + (yield* store.getThreadRecords(threadId, ["messages"])).messages, + indexed.messages, + ); +}); + +// It drops the index, so it gets its own database. +it.effect("reads message ordinals through their index and orders messages as without it", () => + messageOrdinalsReadThroughTheirIndex.pipe(Effect.provide(layerTest)), +); + const assertProjectScopedShells = Effect.fn("assertProjectScopedShells")(function* ( suffix: string, ) { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b5a0f2cfb555..ee74a6150cae 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3732,10 +3732,13 @@ export const layer: Layer.Layer = (filter.messageRoles.includes("user") && filter.messageRoles.includes("assistant"))) ? undefined : messages.map((message) => message.id); + // Migration 80 indexes exactly this expression, type list and json_valid term: changing + // any, even the IN order, silently drops the index. ProjectionStore.test.ts asserts the plan. const rows = yield* sql<{ message_id: string; ordinal: number }>` SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND type IN ('user_message', 'assistant_message') + AND json_valid(payload_json) ${loadedMessageIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.messageId') IN (SELECT value FROM json_each(${encodeIdList(loadedMessageIds)}))`} GROUP BY json_extract(payload_json, '$.messageId') `; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 606770d1b650..67483c973805 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -83,6 +83,7 @@ import Migration0077 from "./Migrations/077_RepairBooleanSettledOverride.ts"; import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationIndex.ts"; import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; +import Migration0080 from "./Migrations/080_TurnItemMessageOrdinalIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -165,6 +166,7 @@ const migrationEntries = [ [77, "RepairBooleanSettledOverride", Migration0077], [78, "DropOrchestrationEventCorrelationIndex", Migration0078], [79, "DelegatedRecoveryIndexes", Migration0079], + [80, "TurnItemMessageOrdinalIndex", Migration0080], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts b/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts new file mode 100644 index 000000000000..2af2c28cf82c --- /dev/null +++ b/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts @@ -0,0 +1,138 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import TurnItemMessageOrdinalIndex from "./080_TurnItemMessageOrdinalIndex.ts"; + +// The statement without the index's json_valid term, which the indexed one (below) must answer +// the same over JSON payloads. +const unindexedOrdinalsOf = (threadId: string) => ` + SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal + FROM orchestration_v2_projection_turn_items NOT INDEXED + WHERE thread_id = '${threadId}' AND type IN ('user_message', 'assistant_message') + GROUP BY json_extract(payload_json, '$.messageId') + ORDER BY message_id +`; + +const ordinalsOf = (threadId: string, from: string) => ` + SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal + FROM ${from} + WHERE thread_id = '${threadId}' AND type IN ('user_message', 'assistant_message') + AND json_valid(payload_json) + GROUP BY json_extract(payload_json, '$.messageId') + ORDER BY message_id +`; + +const insertTurnItem = ( + id: string, + threadId: string, + ordinal: number, + type: string, + payloadJson: string, +) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql` + INSERT INTO orchestration_v2_projection_turn_items ( + turn_item_id, thread_id, run_id, ordinal, type, status, updated_at, payload_json + ) + VALUES ( + ${id}, ${threadId}, NULL, ${ordinal}, ${type}, 'completed', + '2026-11-01T00:00:00.000Z', ${payloadJson} + ) + `; + }); + +const usesTheIndex = (query: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const plan = yield* sql.unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${query}`); + assert.include( + plan[0]?.detail, + "USING INDEX orchestration_v2_projection_turn_items_message_ordinal_idx", + ); + }); + +it.effect( + "adds the message-ordinal index once, re-runs cleanly and keeps the grouped ordinals", + () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 79 }); + for (const [id, threadId, ordinal, type, messageId] of [ + ["a1", "thread-a", 1, "user_message", "m1"], + ["a2", "thread-a", 3, "assistant_message", "m2"], + ["a3", "thread-a", 2, "assistant_message", "m2"], + ["a4", "thread-a", 4, "dynamic_tool", "m3"], + ["a5", "thread-a", 5, "user_message", "m3"], + ["b1", "thread-b", 1, "assistant_message", "m1"], + ] as const) { + yield* sql` + INSERT INTO orchestration_v2_projection_turn_items ( + turn_item_id, thread_id, run_id, ordinal, type, status, updated_at, payload_json + ) + VALUES ( + ${id}, ${threadId}, NULL, ${ordinal}, ${type}, 'completed', + '2026-11-01T00:00:00.000Z', ${JSON.stringify({ messageId })} + ) + `; + } + const expected = [ + { message_id: "m1", ordinal: 1 }, + { message_id: "m2", ordinal: 2 }, + { message_id: "m3", ordinal: 5 }, + ]; + assert.deepEqual(yield* sql.unsafe(unindexedOrdinalsOf("thread-a")), expected); + + assert.deepEqual( + (yield* runMigrations()).map(([id]) => id), + [80], + ); + const index = () => sql<{ readonly sql: string }>` + SELECT sql FROM sqlite_master + WHERE type = 'index' AND name = 'orchestration_v2_projection_turn_items_message_ordinal_idx' + `; + const created = yield* index(); + assert.lengthOf(created, 1); + yield* TurnItemMessageOrdinalIndex; + assert.deepEqual(yield* index(), created); + assert.deepEqual(yield* runMigrations(), []); + + const indexed = "orchestration_v2_projection_turn_items"; + yield* usesTheIndex(ordinalsOf("thread-a", indexed)); + assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a", indexed)), expected); + assert.deepEqual( + yield* sql.unsafe(ordinalsOf("thread-b", indexed)), + yield* sql.unsafe(unindexedOrdinalsOf("thread-b")), + ); + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), +); + +// Without the index a payload that is not JSON fails no write; with it, it must fail none either. +it.effect("migrates over a message whose payload is not JSON, and takes one after it", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 79 }); + yield* insertTurnItem("bad-1", "thread-a", 1, "assistant_message", "{not json"); + yield* insertTurnItem("ok-1", "thread-a", 2, "user_message", '{"messageId":"m1"}'); + + assert.deepEqual( + (yield* runMigrations()).map(([id]) => id), + [80], + ); + yield* insertTurnItem("bad-2", "thread-a", 3, "user_message", "{not json"); + yield* insertTurnItem("ok-2", "thread-a", 4, "assistant_message", '{"messageId":"m1"}'); + yield* sql` + UPDATE orchestration_v2_projection_turn_items SET payload_json = '{still not json' + WHERE turn_item_id = 'ok-2' + `; + + const indexed = "orchestration_v2_projection_turn_items"; + yield* usesTheIndex(ordinalsOf("thread-a", indexed)); + assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a", indexed)), [ + { message_id: "m1", ordinal: 2 }, + ]); + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), +); diff --git a/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.ts b/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.ts new file mode 100644 index 000000000000..99688323cddd --- /dev/null +++ b/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.ts @@ -0,0 +1,20 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +/** + * A projection read groups a thread's message items by `$.messageId` for each message's first + * ordinal; without this index it parses every message item's payload. The expression and the + * WHERE must stay identical to that query's, or SQLite will not use the index. `json_valid` keeps + * a payload that is not JSON out of it: json_extract would fail its write. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_message_ordinal_idx + ON orchestration_v2_projection_turn_items( + thread_id, json_extract(payload_json, '$.messageId'), ordinal + ) + WHERE type IN ('user_message', 'assistant_message') AND json_valid(payload_json) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 5f51c237fd76..5967ae14b1aa 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -59,6 +59,7 @@ describe("V2 preview upgrade", () => { [77, "RepairBooleanSettledOverride"], [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], + [80, "TurnItemMessageOrdinalIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -161,6 +162,7 @@ describe("V2 preview upgrade", () => { [77, "RepairBooleanSettledOverride"], [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], + [80, "TurnItemMessageOrdinalIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 4458f1af9346..993916c435de 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -105,6 +105,13 @@ queries use a covering thread/run index for item counts and a partial index for background items, including idle items. Latest user-message reads use an index ordered by update time. +A projection read that orders messages by their first user or assistant turn item groups those +items by `$.messageId` through a partial index on that expression (migration 80), so it does not +parse every message item's payload; the index's expression and `WHERE` must stay identical to the +query's. Migration 80 scans the turn-item table once on first start; a message item whose payload +is not valid JSON is left out of the index. Each message-item write re-parses the old and new +payload for the index, so its cost grows with the payload. + Command palette results keep live VCS and linked pull-request queries leased to the visible viewport, including a small overscan region. Cached badges remain available while leases are released. Local Git status reads skip divergence calculations; callers that need ahead/behind From a0ea9f2c6ec6607e7d0e4463d196c5d91644ed91 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 14:17:53 +0000 Subject: [PATCH 2/7] test(server): message-ordinal plan checks assert the index, not SQLite's wording The store test now checks that the message-ordinal statement's plan names the message-ordinal index and has no temporary B-tree, instead of pinning SQLite's exact plan text, which changes between SQLite versions. The migration test drops its own copy of that plan check (the store test covers the real statement) and keeps its result comparisons. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 31 +++++++++++++------ .../080_TurnItemMessageOrdinalIndex.test.ts | 26 +++------------- 2 files changed, 26 insertions(+), 31 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index f2f12cce6041..005aa355b6ad 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1722,6 +1722,24 @@ const queryPlan = Effect.fn("queryPlan")(function* (statement: string | undefine return rows.map((row) => row.detail); }); +// SQLite's plan wording changes between versions, so only the index name and the absence of a +// temporary B-tree are pinned. +const assertReadsThroughMessageOrdinalIndex = Effect.fn("assertReadsThroughMessageOrdinalIndex")( + function* (statement: string | undefined) { + const plan = yield* queryPlan(statement); + assert.isTrue( + plan.some((detail) => + detail.includes("orchestration_v2_projection_turn_items_message_ordinal_idx"), + ), + plan.join("\n"), + ); + assert.isFalse( + plan.some((detail) => detail.includes("TEMP B-TREE")), + plan.join("\n"), + ); + }, +); + // Each message's first item orders it, not the stored message order (the assistant reply is stored // first); the message-ordinal statement reads those ordinals through migration 80's index. const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { @@ -1815,11 +1833,8 @@ const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { const indexed = yield* store .getThreadRecords(threadId, ["messages"]) .pipe(Effect.withTracer(tracer)); - assert.deepEqual( - yield* queryPlan(statements.find((statement) => statement.includes("MIN(ordinal) AS ordinal"))), - [ - "SEARCH orchestration_v2_projection_turn_items USING INDEX orchestration_v2_projection_turn_items_message_ordinal_idx (thread_id=?)", - ], + yield* assertReadsThroughMessageOrdinalIndex( + statements.find((statement) => statement.includes("MIN(ordinal) AS ordinal")), ); assert.deepEqual( indexed.messages.map((message) => message.id), @@ -1848,11 +1863,7 @@ const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { statement.includes("MIN(ordinal) AS ordinal"), ); assert.include(bounded, "IN (SELECT value FROM json_each("); - assert.deepEqual(yield* queryPlan(bounded), [ - "SEARCH orchestration_v2_projection_turn_items USING INDEX orchestration_v2_projection_turn_items_message_ordinal_idx (thread_id=? AND =?)", - "LIST SUBQUERY 1", - "SCAN json_each VIRTUAL TABLE INDEX 1:", - ]); + yield* assertReadsThroughMessageOrdinalIndex(bounded); const rows = yield* sql.unsafe<{ readonly message_id: string }>(bounded!, [ threadId, JSON.stringify(loaded), diff --git a/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts b/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts index 2af2c28cf82c..df55ed584d60 100644 --- a/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts +++ b/apps/server/src/persistence/Migrations/080_TurnItemMessageOrdinalIndex.test.ts @@ -16,9 +16,9 @@ const unindexedOrdinalsOf = (threadId: string) => ` ORDER BY message_id `; -const ordinalsOf = (threadId: string, from: string) => ` +const ordinalsOf = (threadId: string) => ` SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal - FROM ${from} + FROM orchestration_v2_projection_turn_items WHERE thread_id = '${threadId}' AND type IN ('user_message', 'assistant_message') AND json_valid(payload_json) GROUP BY json_extract(payload_json, '$.messageId') @@ -45,16 +45,6 @@ const insertTurnItem = ( `; }); -const usesTheIndex = (query: string) => - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - const plan = yield* sql.unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${query}`); - assert.include( - plan[0]?.detail, - "USING INDEX orchestration_v2_projection_turn_items_message_ordinal_idx", - ); - }); - it.effect( "adds the message-ordinal index once, re-runs cleanly and keeps the grouped ordinals", () => @@ -100,11 +90,9 @@ it.effect( assert.deepEqual(yield* index(), created); assert.deepEqual(yield* runMigrations(), []); - const indexed = "orchestration_v2_projection_turn_items"; - yield* usesTheIndex(ordinalsOf("thread-a", indexed)); - assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a", indexed)), expected); + assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a")), expected); assert.deepEqual( - yield* sql.unsafe(ordinalsOf("thread-b", indexed)), + yield* sql.unsafe(ordinalsOf("thread-b")), yield* sql.unsafe(unindexedOrdinalsOf("thread-b")), ); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), @@ -129,10 +117,6 @@ it.effect("migrates over a message whose payload is not JSON, and takes one afte WHERE turn_item_id = 'ok-2' `; - const indexed = "orchestration_v2_projection_turn_items"; - yield* usesTheIndex(ordinalsOf("thread-a", indexed)); - assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a", indexed)), [ - { message_id: "m1", ordinal: 2 }, - ]); + assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a")), [{ message_id: "m1", ordinal: 2 }]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 7d535a8e91586350b699d600ed4cbb774dcca9ff Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 15:00:20 +0000 Subject: [PATCH 3/7] test(server): a narrowed message-ordinal read must seek the message key The structural plan check for the bounded narrow reads also requires the message-ordinal index search to seek the message id expression, not just the thread, so a read that filters the whole thread after the index fails. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 005aa355b6ad..6675aec95279 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1722,17 +1722,16 @@ const queryPlan = Effect.fn("queryPlan")(function* (statement: string | undefine return rows.map((row) => row.detail); }); -// SQLite's plan wording changes between versions, so only the index name and the absence of a -// temporary B-tree are pinned. +// SQLite's plan wording changes between versions, so only the index search, the keys it seeks +// and the absence of a temporary B-tree are pinned. const assertReadsThroughMessageOrdinalIndex = Effect.fn("assertReadsThroughMessageOrdinalIndex")( - function* (statement: string | undefined) { + function* (statement: string | undefined, keys: ReadonlyArray) { const plan = yield* queryPlan(statement); - assert.isTrue( - plan.some((detail) => - detail.includes("orchestration_v2_projection_turn_items_message_ordinal_idx"), - ), - plan.join("\n"), + const search = plan.find((detail) => + detail.includes("orchestration_v2_projection_turn_items_message_ordinal_idx"), ); + assert.isDefined(search, plan.join("\n")); + for (const key of keys) assert.include(search, key); assert.isFalse( plan.some((detail) => detail.includes("TEMP B-TREE")), plan.join("\n"), @@ -1835,6 +1834,7 @@ const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { .pipe(Effect.withTracer(tracer)); yield* assertReadsThroughMessageOrdinalIndex( statements.find((statement) => statement.includes("MIN(ordinal) AS ordinal")), + ["thread_id=?"], ); assert.deepEqual( indexed.messages.map((message) => message.id), @@ -1863,7 +1863,8 @@ const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { statement.includes("MIN(ordinal) AS ordinal"), ); assert.include(bounded, "IN (SELECT value FROM json_each("); - yield* assertReadsThroughMessageOrdinalIndex(bounded); + // The selected messages are sought in the index, not filtered from the whole thread. + yield* assertReadsThroughMessageOrdinalIndex(bounded, ["thread_id=?", "=?"]); const rows = yield* sql.unsafe<{ readonly message_id: string }>(bounded!, [ threadId, JSON.stringify(loaded), From aa191911a675636fca9c22d542e1b027b3759994 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Thu, 12 Nov 2026 23:50:08 +0000 Subject: [PATCH 4/7] test(server): the non-JSON payload written after migration is an assistant message Main's queued-input index (migration 86) evaluates json_extract on user messages, so a user message whose payload is not JSON is refused before the message-ordinal index sees it. The test now writes the non-JSON item the schema still accepts and checks the index refuses none of them. Co-Authored-By: Claude Opus 5.5 --- .../Migrations/087_TurnItemMessageOrdinalIndex.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/server/src/persistence/Migrations/087_TurnItemMessageOrdinalIndex.test.ts b/apps/server/src/persistence/Migrations/087_TurnItemMessageOrdinalIndex.test.ts index d03370148b6b..204dc4035196 100644 --- a/apps/server/src/persistence/Migrations/087_TurnItemMessageOrdinalIndex.test.ts +++ b/apps/server/src/persistence/Migrations/087_TurnItemMessageOrdinalIndex.test.ts @@ -98,7 +98,8 @@ it.effect( }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); -// Without the index a payload that is not JSON fails no write; with it, it must fail none either. +// The index must fail no write the schema accepts: a message item whose payload is not JSON. +// (A user message's non-JSON payload is already refused by migration 86's queued-input index.) it.effect("migrates over a message whose payload is not JSON, and takes one after it", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -110,7 +111,7 @@ it.effect("migrates over a message whose payload is not JSON, and takes one afte (yield* runMigrations()).map(([id]) => id), [87], ); - yield* insertTurnItem("bad-2", "thread-a", 3, "user_message", "{not json"); + yield* insertTurnItem("bad-2", "thread-a", 3, "assistant_message", "{not json"); yield* insertTurnItem("ok-2", "thread-a", 4, "assistant_message", '{"messageId":"m1"}'); yield* sql` UPDATE orchestration_v2_projection_turn_items SET payload_json = '{still not json' From ff7cae6745a1e771909df39a96b07ceeba2acfa0 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Fri, 13 Nov 2026 04:49:28 +0000 Subject: [PATCH 5/7] test(server): a full message-ordering pass may search the message-ordinal index by thread The control-read check counts a turn-item search bounded only by the thread as a history walk. The full ordering pass that Stop and steer already accept now reads the message-ordinal index, which holds only message items, so its thread search reads the same rows as the typed search it replaces, without the temporary B-tree. A narrowed pass must still seek the message id. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.control-reads.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 21afe0b0574c..dc545c2e92f9 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -2428,7 +2428,9 @@ it.effect("queued-message commands read a long thread's history without walking if ( (detail.includes("TEMP B-TREE FOR GROUP BY") && query.includes("json_each")) || (/^SEARCH .*(node|turn_item).* \(thread_id=\?\)$/.test(detail) && - !detail.includes("USING COVERING INDEX")) || + !detail.includes("USING COVERING INDEX") && + // That full ordering pass, through an index that holds only message items. + !(detail.includes("message_ordinal_idx") && !query.includes("json_each"))) || // Every active run in the environment, not only this thread's. /^SEARCH .*runs.* \(status=\?\)$/.test(detail) || // A status-filtered read that fetches every typed row to check its status. From 5239938f6cb458aee0c89992e78290a359ff9ded Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Sat, 14 Nov 2026 19:58:08 +0000 Subject: [PATCH 6/7] test(server): the non-JSON payload written after migration is a user message again Main's queued-input index now guards json_extract with json_valid, so a user message whose payload is not JSON is writable again. The migration test goes back to writing one after the message-ordinal index exists. Co-Authored-By: Claude Opus 5.5 --- .../Migrations/093_TurnItemMessageOrdinalIndex.test.ts | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts b/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts index e743cc57c999..7b2799c6ee68 100644 --- a/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts +++ b/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts @@ -98,8 +98,7 @@ it.effect( }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); -// The index must fail no write the schema accepts: a message item whose payload is not JSON. -// (A user message's non-JSON payload is already refused by migration 86's queued-input index.) +// Without the index a payload that is not JSON fails no write; with it, it must fail none either. it.effect("migrates over a message whose payload is not JSON, and takes one after it", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -111,7 +110,7 @@ it.effect("migrates over a message whose payload is not JSON, and takes one afte (yield* runMigrations()).map(([id]) => id), [93], ); - yield* insertTurnItem("bad-2", "thread-a", 3, "assistant_message", "{not json"); + yield* insertTurnItem("bad-2", "thread-a", 3, "user_message", "{not json"); yield* insertTurnItem("ok-2", "thread-a", 4, "assistant_message", '{"messageId":"m1"}'); yield* sql` UPDATE orchestration_v2_projection_turn_items SET payload_json = '{still not json' From 0f9c1cc11c045d17c8b39e15e547faadb6685cb0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 15 Nov 2026 00:46:34 +0000 Subject: [PATCH 7/7] test(server): keep only the message-ordinal checks that catch a regression The plan test still fails when the query drifts from migration 93's index, and the migration test still covers payloads that are not JSON. Dropped the re-run of the narrowed statement, the drop-index comparison, and the migration idempotency case. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 22 +------ .../093_TurnItemMessageOrdinalIndex.test.ts | 64 ------------------- 2 files changed, 2 insertions(+), 84 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index b80d40c89327..e642ea0c0a53 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2640,7 +2640,6 @@ const assertReadsThroughMessageOrdinalIndex = Effect.fn("assertReadsThroughMessa // first); the message-ordinal statement reads those ordinals through migration 93's index. const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; - const sql = yield* SqlClient.SqlClient; const threadId = ThreadId.make("thread:message-ordinals"); yield* store.apply(recordsThreadCreated(threadId)); const conversation = (name: string, firstOrdinal: number) => { @@ -2737,8 +2736,7 @@ const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { indexed.messages.map((message) => message.id), [first.user, first.assistant, second.user, second.assistant], ); - // A narrow read (ids, runs or one role) recovers only its own messages' ordinals, through the - // same index; the statement it sends, rerun with its bindings, returns just those rows. + // A narrow read (ids, runs or one role) seeks only its own messages in the same index. const narrowReads = [ { filter: { messageIds: [second.assistant, first.user] }, @@ -2759,27 +2757,11 @@ const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { const bounded = narrow.statements.find((statement) => statement.includes("MIN(ordinal) AS ordinal"), ); - assert.include(bounded, "IN (SELECT value FROM json_each("); - // The selected messages are sought in the index, not filtered from the whole thread. yield* assertReadsThroughMessageOrdinalIndex(bounded, ["thread_id=?", "=?"]); - const rows = yield* sql.unsafe<{ readonly message_id: string }>(bounded!, [ - threadId, - JSON.stringify(loaded), - ]); - assert.sameMembers( - rows.map((row) => row.message_id), - loaded, - ); } - yield* sql`DROP INDEX orchestration_v2_projection_turn_items_message_ordinal_idx`; - assert.deepEqual( - (yield* store.getThreadRecords(threadId, ["messages"])).messages, - indexed.messages, - ); }); -// It drops the index, so it gets its own database. -it.effect("reads message ordinals through their index and orders messages as without it", () => +it.effect("reads message ordinals through their index", () => messageOrdinalsReadThroughTheirIndex.pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts b/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts index 7b2799c6ee68..e44b3c06cf72 100644 --- a/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts +++ b/apps/server/src/persistence/Migrations/093_TurnItemMessageOrdinalIndex.test.ts @@ -4,17 +4,6 @@ import * as Effect from "effect/Effect"; import * as SqlClient from "effect/sql/SqlClient"; import { runMigrations } from "../Migrations.ts"; -import TurnItemMessageOrdinalIndex from "./093_TurnItemMessageOrdinalIndex.ts"; - -// The statement without the index's json_valid term, which the indexed one (below) must answer -// the same over JSON payloads. -const unindexedOrdinalsOf = (threadId: string) => ` - SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal - FROM orchestration_v2_projection_turn_items NOT INDEXED - WHERE thread_id = '${threadId}' AND type IN ('user_message', 'assistant_message') - GROUP BY json_extract(payload_json, '$.messageId') - ORDER BY message_id -`; const ordinalsOf = (threadId: string) => ` SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal @@ -45,59 +34,6 @@ const insertTurnItem = ( `; }); -it.effect( - "adds the message-ordinal index once, re-runs cleanly and keeps the grouped ordinals", - () => - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - yield* runMigrations({ toMigrationInclusive: 92 }); - for (const [id, threadId, ordinal, type, messageId] of [ - ["a1", "thread-a", 1, "user_message", "m1"], - ["a2", "thread-a", 3, "assistant_message", "m2"], - ["a3", "thread-a", 2, "assistant_message", "m2"], - ["a4", "thread-a", 4, "dynamic_tool", "m3"], - ["a5", "thread-a", 5, "user_message", "m3"], - ["b1", "thread-b", 1, "assistant_message", "m1"], - ] as const) { - yield* sql` - INSERT INTO orchestration_v2_projection_turn_items ( - turn_item_id, thread_id, run_id, ordinal, type, status, updated_at, payload_json - ) - VALUES ( - ${id}, ${threadId}, NULL, ${ordinal}, ${type}, 'completed', - '2026-11-01T00:00:00.000Z', ${JSON.stringify({ messageId })} - ) - `; - } - const expected = [ - { message_id: "m1", ordinal: 1 }, - { message_id: "m2", ordinal: 2 }, - { message_id: "m3", ordinal: 5 }, - ]; - assert.deepEqual(yield* sql.unsafe(unindexedOrdinalsOf("thread-a")), expected); - - assert.deepEqual( - (yield* runMigrations()).map(([id]) => id), - [93], - ); - const index = () => sql<{ readonly sql: string }>` - SELECT sql FROM sqlite_master - WHERE type = 'index' AND name = 'orchestration_v2_projection_turn_items_message_ordinal_idx' - `; - const created = yield* index(); - assert.lengthOf(created, 1); - yield* TurnItemMessageOrdinalIndex; - assert.deepEqual(yield* index(), created); - assert.deepEqual(yield* runMigrations(), []); - - assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a")), expected); - assert.deepEqual( - yield* sql.unsafe(ordinalsOf("thread-b")), - yield* sql.unsafe(unindexedOrdinalsOf("thread-b")), - ); - }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), -); - // Without the index a payload that is not JSON fails no write; with it, it must fail none either. it.effect("migrates over a message whose payload is not JSON, and takes one after it", () => Effect.gen(function* () {