perf(server): message ordinals are read through an index, not a scan

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.

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=?", "<expr>=?"]);
     // 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<string>) {
+    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=?", "<expr>=?"]);
+  }
+});
+
+it.effect("reads message ordinals through their index", () =>
+  messageOrdinalsReadThroughTheirIndex.pipe(Effect.provide(layerTest)),
+);
+
 const assertProjectScopedShells = Effect.fn("assertProjectScopedShells")(function* (
   suffix: string,
 ) {

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

@@ -4341,7 +4341,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows),
         ]);
         const messageOrdinals = new Map<string, number>();
-        // 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<ProjectionStoreV2, never, SqlClient.SqlClient> =
               (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')
           `;

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);

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:" }))),
+);

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)
+  `;
+});

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:" }))),
   );

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