diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 84c650f75fe32..69eaa0657fffe 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4001,15 +4001,11 @@ export const layer: Layer.Layer = 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' diff --git a/apps/server/src/persistence/Migrations.test.ts b/apps/server/src/persistence/Migrations.test.ts index 54e4a711aa8a1..36c7f8fa3830b 100644 --- a/apps/server/src/persistence/Migrations.test.ts +++ b/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 = []; + 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* () { diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 1f3fb4bc6b8bc..545ea8231ba31 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/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); diff --git a/apps/server/src/persistence/Migrations/076_StartupRecoveryIndexes.ts b/apps/server/src/persistence/Migrations/076_StartupRecoveryIndexes.ts new file mode 100644 index 0000000000000..f608a043990c9 --- /dev/null +++ b/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') + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 4fefa61ec010f..297ef9f1c582c 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/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:" }))), );