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 d2d930972de6..6ff513755624 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -2671,7 +2671,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. diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 2ea2e730e61d..156223fee207 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1311,17 +1311,8 @@ it.effect("record reads filter notification sources without dropping matching me .pipe(Effect.withTracer(tracer)); assert.lengthOf(userMessages.messages, 3); const ordinalLookup = statements.find((statement) => statement.includes("MIN(ordinal)")); - assert.isDefined(ordinalLookup); - const sql = yield* SqlClient.SqlClient; - const plan = (yield* sql.unsafe<{ readonly detail: string }>( - `EXPLAIN QUERY PLAN ${ordinalLookup}`, - [threadId, "[]"], - )).map((row) => row.detail); - // User-only reads must stay on a type-bounded index instead of scanning every turn item. - assert.isTrue( - plan.some((detail) => /turn_items_(user_message|thread_type)_idx/.test(detail)), - plan.join("\n"), - ); + // User-only reads must seek the message-ordinal index instead of scanning every turn item. + yield* assertReadsThroughMessageOrdinalIndex(ordinalLookup, ["thread_id=?", "=?"]); // With every user item loaded, the items give the order and the lookup does not run. const withItems = traceSqlStatements(); const ordered = yield* store @@ -2613,6 +2604,192 @@ 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); +}); + +// 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, keys: ReadonlyArray) { + const plan = yield* queryPlan(statement); + 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"), + ); + }, +); + +// 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 94's index. +const messageOrdinalsReadThroughTheirIndex = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + 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)); + yield* assertReadsThroughMessageOrdinalIndex( + statements.find((statement) => statement.includes("MIN(ordinal) AS ordinal")), + ["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) seeks only its own messages in the same index. + 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"), + ); + yield* assertReadsThroughMessageOrdinalIndex(bounded, ["thread_id=?", "=?"]); + } +}); + +it.effect("reads message ordinals through their index", () => + 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 0b0528d1ad26..a003ade1ae8f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4341,7 +4341,7 @@ export const layer: Layer.Layer = decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows), ]); const messageOrdinals = new Map(); - // The literal `type = 'user_message'` lets SQLite pick the partial user-message index. + // Only user items can name a user message, so a user-only read needs no assistant items. const userOnly = messages.every((message) => message.role === "user"); // The loaded items already order the messages when they hold every item the lookup reads. const loadsEveryItemOf = (type: OrchestrationV2TurnItem["type"]) => @@ -4367,11 +4367,13 @@ export const layer: Layer.Layer = (filter.messageRoles.includes("user") && filter.messageRoles.includes("assistant"))) ? undefined : messages.map((message) => message.id); + // Migration 94 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 ${userOnly ? sql`type = 'user_message'` : sql`type IN ('user_message', 'assistant_message')`} + 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 586c59729c5c..0e6b50228511 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -101,6 +101,7 @@ import Migration0090 from "./Migrations/090_AuthConnectDevices.ts"; import Migration0091 from "./Migrations/091_AuthConnectDeniedRequests.ts"; import Migration0092 from "./Migrations/092_QueuedTurnInputIndexJsonValid.ts"; import Migration0093 from "./Migrations/093_ProviderTurnAttemptIndex.ts"; +import Migration0094 from "./Migrations/094_TurnItemMessageOrdinalIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -197,6 +198,7 @@ const migrationEntries = [ [91, "AuthConnectDeniedRequests", Migration0091], [92, "QueuedTurnInputIndexJsonValid", Migration0092], [93, "ProviderTurnAttemptIndex", Migration0093], + [94, "TurnItemMessageOrdinalIndex", Migration0094], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.test.ts b/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.test.ts new file mode 100644 index 000000000000..1a8ea4ecfed4 --- /dev/null +++ b/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.test.ts @@ -0,0 +1,58 @@ +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"; + +const ordinalsOf = (threadId: string) => ` + 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) + 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} + ) + `; + }); + +// 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: 93 }); + 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), + [94], + ); + 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' + `; + + assert.deepEqual(yield* sql.unsafe(ordinalsOf("thread-a")), [{ message_id: "m1", ordinal: 2 }]); + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), +); diff --git a/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.ts b/apps/server/src/persistence/Migrations/094_TurnItemMessageOrdinalIndex.ts new file mode 100644 index 000000000000..99688323cddd --- /dev/null +++ b/apps/server/src/persistence/Migrations/094_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 ad20d3ef6f35..953a2229ce26 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -73,6 +73,7 @@ describe("V2 preview upgrade", () => { [91, "AuthConnectDeniedRequests"], [92, "QueuedTurnInputIndexJsonValid"], [93, "ProviderTurnAttemptIndex"], + [94, "TurnItemMessageOrdinalIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -189,6 +190,7 @@ describe("V2 preview upgrade", () => { [91, "AuthConnectDeniedRequests"], [92, "QueuedTurnInputIndexJsonValid"], [93, "ProviderTurnAttemptIndex"], + [94, "TurnItemMessageOrdinalIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 8dbf42fc45f7..25ecabd80c5a 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -131,6 +131,13 @@ queries read item counts from a per-thread, per-run table that turn-item trigger use a partial index for nonterminal 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 94), so it does not +parse every message item's payload; the index's expression and `WHERE` must stay identical to the +query's. Migration 94 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