perf(server): index slow projection lookups

apps/server/src/persistence/Migrations.ts

@@ -69,6 +69,7 @@ import Migration0063 from "./Migrations/063_DropUnusedV1Projections.ts";
 import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.ts";
 import Migration0065 from "./Migrations/065_CodexUsageNativeTurnIndex.ts";
 import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts";
+import Migration0067 from "./Migrations/067_OrchestrationSlowQueryIndexes.ts";
 
 export const migrationEntries = [
   [1, "OrchestrationEvents", Migration0001],
@@ -138,6 +139,7 @@ export const migrationEntries = [
   [64, "CanonicalizeThreadPullRequestLinks", Migration0064],
   [65, "CodexUsageNativeTurnIndex", Migration0065],
   [66, "RepairClientSettlementTimestamps", Migration0066],
+  [67, "OrchestrationSlowQueryIndexes", Migration0067],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

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

@@ -15,7 +15,7 @@ layer("055_OrchestrationV2", (it) => {
     Effect.sync(() => {
       assert.deepStrictEqual(
         migrationEntries.map(([id]) => id),
-        Array.from({ length: 66 }, (_, index) => index + 1),
+        Array.from({ length: 67 }, (_, index) => index + 1),
       );
     }),
   );
@@ -40,6 +40,7 @@ layer("055_OrchestrationV2", (it) => {
         [64, "CanonicalizeThreadPullRequestLinks"],
         [65, "CodexUsageNativeTurnIndex"],
         [66, "RepairClientSettlementTimestamps"],
+        [67, "OrchestrationSlowQueryIndexes"],
       ]);
       assert.deepStrictEqual(yield* runMigrations(), []);
 
@@ -72,6 +73,7 @@ layer("055_OrchestrationV2", (it) => {
         { migration_id: 64, name: "CanonicalizeThreadPullRequestLinks" },
         { migration_id: 65, name: "CodexUsageNativeTurnIndex" },
         { migration_id: 66, name: "RepairClientSettlementTimestamps" },
+        { migration_id: 67, name: "OrchestrationSlowQueryIndexes" },
       ]);
 
       const tables = yield* sql<{ readonly name: string }>`

apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.test.ts

@@ -0,0 +1,109 @@
+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 layer = it.layer(NodeSqliteClient.layer({ filename: ":memory:" }));
+
+layer("067_OrchestrationSlowQueryIndexes", (it) => {
+  it.effect("creates the indexes and the planner uses them for the slow queries", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      yield* runMigrations({ toMigrationInclusive: 66 });
+      yield* sql`
+        INSERT INTO orchestration_v2_projection_turn_items (
+          turn_item_id, thread_id, run_id, ordinal, type, status, payload_json, updated_at
+        ) VALUES (
+          'item-1', 'thread-1', NULL, 1, 'user_message', 'completed',
+          '{"messageId":"message-1"}', '2026-01-01'
+        )
+      `;
+      const beforeMessages = yield* sql`
+        SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal
+        FROM orchestration_v2_projection_turn_items
+        WHERE thread_id = 'thread-1' AND type IN ('user_message', 'assistant_message')
+        GROUP BY json_extract(payload_json, '$.messageId')
+      `;
+      const beforeRunning = yield* sql<{ readonly detail: string }>`
+        EXPLAIN QUERY PLAN
+        SELECT thread_id FROM orchestration_v2_projection_runs WHERE status = 'running'
+      `;
+      assert.isFalse(
+        beforeRunning.some((row) => row.detail.includes("runs_recovery_idx")),
+        beforeRunning.map((row) => row.detail).join("\n"),
+      );
+      yield* runMigrations({ toMigrationInclusive: 67 });
+      assert.deepEqual(yield* runMigrations(), []);
+
+      const afterRunning = yield* sql<{ readonly detail: string }>`
+        EXPLAIN QUERY PLAN
+        SELECT thread_id FROM orchestration_v2_projection_runs WHERE status = 'running'
+      `;
+      assert.isTrue(
+        afterRunning.some((row) => row.detail.includes("runs_running_idx")),
+        afterRunning.map((row) => row.detail).join("\n"),
+      );
+      const recoveryPlan = yield* sql<{ readonly detail: string }>`
+        EXPLAIN QUERY PLAN
+        SELECT thread_id FROM orchestration_v2_projection_runs
+        WHERE thread_id = 'thread-1'
+          AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting')
+      `;
+      assert.isTrue(
+        recoveryPlan.some((row) => row.detail.includes("runs_recovery_idx")),
+        recoveryPlan.map((row) => row.detail).join("\n"),
+      );
+
+      const rows = yield* sql<{ readonly name: string }>`
+        SELECT name FROM sqlite_master WHERE type = 'index'
+      `;
+      const indexes = new Set(rows.map((row) => row.name));
+      for (const name of [
+        "orchestration_v2_projection_turn_items_thread_type_idx",
+        "orchestration_v2_projection_turn_items_thread_message_idx",
+        "orchestration_v2_projection_turn_items_run_status_idx",
+        "orchestration_v2_projection_runs_running_idx",
+        "orchestration_v2_projection_runs_user_message_idx",
+        "orchestration_v2_projection_provider_turns_thread_completed_idx",
+        "idx_orchestration_events_aggregate_version_sequence",
+      ]) {
+        assert.isTrue(indexes.has(name), name);
+      }
+
+      const afterMessages = yield* sql`
+        SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal
+        FROM orchestration_v2_projection_turn_items
+        WHERE thread_id = 'thread-1' AND type IN ('user_message', 'assistant_message')
+        GROUP BY json_extract(payload_json, '$.messageId')
+      `;
+      assert.deepEqual(afterMessages, beforeMessages);
+      const messagePlan = yield* sql<{ readonly detail: string }>`
+        EXPLAIN QUERY PLAN
+        SELECT json_extract(payload_json, '$.messageId'), MIN(ordinal)
+        FROM orchestration_v2_projection_turn_items
+        WHERE thread_id = 'thread-1' AND type IN ('user_message', 'assistant_message')
+        GROUP BY json_extract(payload_json, '$.messageId')
+      `;
+      assert.isTrue(
+        messagePlan.some((row) =>
+          row.detail.includes("orchestration_v2_projection_turn_items_thread_message_idx"),
+        ),
+        messagePlan.map((row) => row.detail).join("\n"),
+      );
+
+      const runPlan = yield* sql<{ readonly detail: string }>`
+        EXPLAIN QUERY PLAN
+        SELECT thread_id FROM orchestration_v2_projection_runs
+        WHERE json_extract(payload_json, '$.userMessageId') = 'message-1'
+      `;
+      assert.isTrue(
+        runPlan.some((row) =>
+          row.detail.includes("orchestration_v2_projection_runs_user_message_idx"),
+        ),
+        runPlan.map((row) => row.detail).join("\n"),
+      );
+    }),
+  );
+});

apps/server/src/persistence/Migrations/067_OrchestrationSlowQueryIndexes.ts

@@ -0,0 +1,39 @@
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/sql/SqlClient";
+
+// Each expression index repeats the json_extract text of the query it serves.
+// SQLite only uses an expression index when the query expression matches it.
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_thread_type_idx
+    ON orchestration_v2_projection_turn_items(thread_id, type)
+  `;
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_thread_message_idx
+    ON orchestration_v2_projection_turn_items(thread_id, json_extract(payload_json, '$.messageId'))
+    WHERE type IN ('user_message', 'assistant_message')
+  `;
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS orchestration_v2_projection_turn_items_run_status_idx
+    ON orchestration_v2_projection_turn_items(run_id, status)
+  `;
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_running_idx
+    ON orchestration_v2_projection_runs(thread_id)
+    WHERE status = 'running'
+  `;
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_user_message_idx
+    ON orchestration_v2_projection_runs(json_extract(payload_json, '$.userMessageId'))
+  `;
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS orchestration_v2_projection_provider_turns_thread_completed_idx
+    ON orchestration_v2_projection_provider_turns(thread_id, completed_at DESC)
+  `;
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS idx_orchestration_events_aggregate_version_sequence
+    ON orchestration_events(aggregate_kind, application_event_version, sequence)
+  `;
+});

apps/server/src/persistence/reconcileV2PreviewMigration.test.ts

@@ -46,6 +46,7 @@ describe("V2 preview upgrade", () => {
         [64, "CanonicalizeThreadPullRequestLinks"],
         [65, "CodexUsageNativeTurnIndex"],
         [66, "RepairClientSettlementTimestamps"],
+        [67, "OrchestrationSlowQueryIndexes"],
       ]);
       assert.deepStrictEqual(yield* runMigrations(), []);
       assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
@@ -135,6 +136,7 @@ describe("V2 preview upgrade", () => {
         [64, "CanonicalizeThreadPullRequestLinks"],
         [65, "CodexUsageNativeTurnIndex"],
         [66, "RepairClientSettlementTimestamps"],
+        [67, "OrchestrationSlowQueryIndexes"],
       ]);
     }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
   );