perf(server): avoid historical scans during startup recovery

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

@@ -4001,15 +4001,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                       ELSE 0 END
                 )
                 SELECT thread_id FROM orchestration_v2_projection_runs
-                WHERE status IN ('preparing', 'starting', 'running', 'waiting')
-                UNION
-                -- A held queue already went through recovery; rereading it on
-                -- every boot costs a projection read per held thread.
-                SELECT thread_id FROM orchestration_v2_projection_runs
-                WHERE status = 'queued'
-                  AND CASE WHEN json_valid(payload_json)
+                -- SQLite requires the exact partial-index predicate to avoid historical runs.
+                WHERE status IN ('queued', 'preparing', 'starting', 'running', 'waiting')
+                  AND (status <> 'queued' OR CASE WHEN json_valid(payload_json)
                     THEN json_extract(payload_json, '$.queueHeld') IS NOT 1
-                    ELSE 1 END
+                    ELSE 1 END)
                 UNION
                 SELECT thread_id FROM orchestration_v2_projection_runtime_requests
                 WHERE status = 'pending'

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

@@ -2,6 +2,9 @@ import { assert, describe, it } from "@effect/vitest";
 import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Tracer from "effect/Tracer";
+import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
 import * as Exit from "effect/Exit";
 import * as FileSystem from "effect/FileSystem";
 import * as Path from "effect/Path";
@@ -127,6 +130,167 @@ describe("runMigrations", () => {
   );
 });
 
+it.effect("upgrades startup recovery lookups without changing candidate selection", () =>
+  Effect.gen(function* () {
+    const sql = yield* SqlClient.SqlClient;
+    const store = yield* ProjectionStore.ProjectionStoreV2;
+    yield* runMigrations({ toMigrationInclusive: 75 });
+
+    for (const threadId of [
+      "history",
+      "active",
+      "queued",
+      "held",
+      "malformed",
+      "native",
+      "child",
+      "wrong-kind",
+      "attached-root",
+      "deleted",
+      "archived",
+      "duplicate",
+    ]) {
+      yield* sql`INSERT INTO orchestration_v2_projection_threads ${sql.insert({
+        thread_id: threadId,
+        project_id: "project",
+        title: threadId,
+        default_provider: "codex",
+        runtime_mode: "full-access",
+        interaction_mode: "default",
+        created_at: "2026-10-20",
+        updated_at: "2026-10-20",
+        deleted_at: threadId === "deleted" ? "2026-10-20" : null,
+        archived_at: threadId === "archived" ? "2026-10-20" : null,
+        payload_json: "{}",
+      })}`;
+    }
+    for (const [threadId, status, payload] of [
+      ["active", "running", '{"queueHeld":true}'],
+      ["queued", "queued", "{}"],
+      ["held", "queued", '{"queueHeld":true}'],
+      ["malformed", "queued", "{"],
+      ["deleted", "running", "{}"],
+      ["archived", "waiting", "{}"],
+      ["duplicate", "running", "{}"],
+      ["duplicate", "queued", "{}"],
+    ] as const) {
+      yield* sql`INSERT INTO orchestration_v2_projection_runs ${sql.insert({
+        run_id: `${threadId}-${status}`,
+        thread_id: threadId,
+        ordinal: status === "queued" ? 2 : 1,
+        provider: "codex",
+        status,
+        requested_at: "2026-10-20",
+        payload_json: payload,
+      })}`;
+    }
+    for (const [id, threadId, origin, status, childThreadId] of [
+      ["native-live", "native", "provider_native", "waiting", null],
+      ["app-live", "wrong-kind", "app_owned", "running", null],
+      ["child-link", "history", "app_owned", "completed", "child"],
+      ["wrong-kind-link", "history", "app_owned", "completed", "wrong-kind"],
+      ["attached-link", "history", "app_owned", "completed", "attached-root"],
+    ] as const) {
+      yield* sql`INSERT INTO orchestration_v2_projection_subagents ${sql.insert({
+        subagent_id: id,
+        thread_id: threadId,
+        parent_node_id: "parent",
+        provider: "codex",
+        origin,
+        status,
+        child_thread_id: childThreadId,
+        updated_at: "2026-10-20",
+        payload_json: "{}",
+      })}`;
+    }
+    for (const [threadId, kind, runId] of [
+      ["child", "root_turn", null],
+      ["wrong-kind", "tool_call", null],
+      ["attached-root", "root_turn", "finished-run"],
+    ] as const) {
+      yield* sql`INSERT INTO orchestration_v2_projection_nodes ${sql.insert({
+        node_id: `${threadId}-node`,
+        thread_id: threadId,
+        run_id: runId,
+        root_node_id: `${threadId}-node`,
+        kind,
+        status: "running",
+        payload_json: "{}",
+      })}`;
+    }
+
+    // Completed records dominate a real database. Their large payloads should
+    // not need to be read just to find the few remaining recovery candidates.
+    yield* sql`
+      WITH RECURSIVE history(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2048)
+      INSERT INTO orchestration_v2_projection_runs
+        (run_id, thread_id, ordinal, provider, status, requested_at, payload_json)
+      SELECT 'historical-run-' || n, 'history', n, 'codex', 'completed', '2026-10-20',
+        json_object('padding', printf('%.*c', 2048, 'x')) FROM history
+    `;
+    yield* sql`
+      WITH RECURSIVE history(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2048)
+      INSERT INTO orchestration_v2_projection_nodes
+        (node_id, thread_id, root_node_id, kind, status, payload_json)
+      SELECT 'historical-node-' || n, 'child', 'child-node', 'tool_call', 'completed',
+        json_object('padding', printf('%.*c', 2048, 'x')) FROM history
+    `;
+    yield* sql`
+      WITH RECURSIVE history(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 2048)
+      INSERT INTO orchestration_v2_projection_subagents
+        (subagent_id, thread_id, parent_node_id, provider, origin, status, updated_at, payload_json)
+      SELECT 'historical-subagent-' || n, 'history', 'parent', 'codex',
+        'provider_native', 'completed', '2026-10-20',
+        json_object('padding', printf('%.*c', 2048, 'x')) FROM history
+    `;
+
+    const expected = ["active", "archived", "child", "duplicate", "malformed", "native", "queued"];
+    assert.deepStrictEqual((yield* store.getRecoveryThreadIds("runtime")).toSorted(), expected);
+    assert.deepStrictEqual(yield* runMigrations(), [[76, "StartupRecoveryIndexes"]]);
+    assert.deepStrictEqual(yield* runMigrations(), []);
+
+    const statements: Array<string> = [];
+    const tracer = Tracer.make({
+      span(options) {
+        const span = new Tracer.NativeSpan(options);
+        const end = span.end.bind(span);
+        span.end = (time, exit) => {
+          end(time, exit);
+          const query = span.attributes.get("db.query.text");
+          if (typeof query === "string") statements.push(query);
+        };
+        return span;
+      },
+    });
+    const candidates = yield* store.getRecoveryThreadIds("runtime").pipe(Effect.withTracer(tracer));
+    assert.deepStrictEqual(candidates.toSorted(), expected);
+
+    const statement = statements.find((query) => query.includes("pending_provider_threads"));
+    assert.isDefined(statement);
+    const plan = (yield* sql.unsafe<{ readonly detail: string }>(
+      `EXPLAIN QUERY PLAN ${statement}`,
+    )).map(({ detail }) => detail);
+    const details = plan.join("\n");
+    assert.include(details, "orchestration_v2_projection_runs_recovery_idx");
+    assert.include(details, "orchestration_v2_projection_nodes_unattached_recovery_idx");
+    assert.include(details, "orchestration_v2_projection_subagents_live_native_idx");
+    assert.isFalse(
+      plan.some((detail) => detail === "SCAN orchestration_v2_projection_runs"),
+      details,
+    );
+    assert.isFalse(
+      plan.some((detail) => detail === "SCAN orchestration_v2_projection_subagents"),
+      details,
+    );
+  }).pipe(
+    Effect.provide(
+      ProjectionStore.layer.pipe(
+        Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })),
+      ),
+    ),
+  ),
+);
+
 // The npm launcher reads the highest migration id from these file names.
 it.effect("registers every migration file", () =>
   Effect.gen(function* () {

apps/server/src/persistence/Migrations.ts

@@ -78,6 +78,7 @@ import Migration0072 from "./Migrations/072_WaitingLinkRequestIndex.ts";
 import Migration0073 from "./Migrations/073_ActiveTurnItemIndexes.ts";
 import Migration0074 from "./Migrations/074_ExcludeAttachmentPlaceholdersFromSearch.ts";
 import Migration0075 from "./Migrations/075_SubagentDeliveryState.ts";
+import Migration0076 from "./Migrations/076_StartupRecoveryIndexes.ts";
 
 const migrationEntries = [
   [1, "OrchestrationEvents", Migration0001],
@@ -156,6 +157,7 @@ const migrationEntries = [
   [73, "ActiveTurnItemIndexes", Migration0073],
   [74, "ExcludeAttachmentPlaceholdersFromSearch", Migration0074],
   [75, "SubagentDeliveryState", Migration0075],
+  [76, "StartupRecoveryIndexes", Migration0076],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/076_StartupRecoveryIndexes.ts

@@ -0,0 +1,20 @@
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/sql/SqlClient";
+
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  // Startup recovery only needs thread IDs. Keep historical payloads out of
+  // these lookups, including root nodes without an attached run.
+  yield* sql`
+    CREATE INDEX orchestration_v2_projection_nodes_unattached_recovery_idx
+    ON orchestration_v2_projection_nodes(thread_id, kind, status)
+    WHERE run_id IS NULL
+  `;
+  yield* sql`
+    CREATE INDEX orchestration_v2_projection_subagents_live_native_idx
+    ON orchestration_v2_projection_subagents(thread_id)
+    WHERE origin = 'provider_native'
+      AND status IN ('pending', 'running', 'waiting')
+  `;
+});

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

@@ -55,6 +55,7 @@ describe("V2 preview upgrade", () => {
         [73, "ActiveTurnItemIndexes"],
         [74, "ExcludeAttachmentPlaceholdersFromSearch"],
         [75, "SubagentDeliveryState"],
+        [76, "StartupRecoveryIndexes"],
       ]);
       assert.deepStrictEqual(yield* runMigrations(), []);
       assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
@@ -153,6 +154,7 @@ describe("V2 preview upgrade", () => {
         [73, "ActiveTurnItemIndexes"],
         [74, "ExcludeAttachmentPlaceholdersFromSearch"],
         [75, "SubagentDeliveryState"],
+        [76, "StartupRecoveryIndexes"],
       ]);
     }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
   );