diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 3b34434017e9d..6886912fc4cca 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2807,11 +2807,25 @@ it.effect("SQL shells read a subagent family and a worktree's threads", () => assert.deepEqual([...(yield* family(id("missing")))], []); // Includes another root's idle child on the same path, which a shell may leave out. - const onWorktree = yield* store.getShellSnapshot({ location: "active", worktreePath }); + const { statements, tracer } = traceSqlStatements(); + const onWorktree = yield* store + .getShellSnapshot({ location: "active", worktreePath }) + .pipe(Effect.withTracer(tracer)); assert.sameMembers( onWorktree.threads.map((thread) => thread.id), [id("root"), id("child"), id("other-child")], ); + // A path names a few threads; scanning every thread parses every payload. + const sql = yield* SqlClient.SqlClient; + const threadQuery = statements.find((statement) => + statement.includes("AS latest_run_stop_requested"), + ); + assert.isDefined(threadQuery); + const threadPlan = (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${threadQuery}`, + [], + )).map((row) => row.detail); + assert.notInclude(threadPlan, "SCAN t", threadPlan.join("\n")); // A fork keeps its deleted source's root, so the deleted root still names the live tree. yield* store.apply({ id: EventId.make("event:family:root-deleted"), @@ -6629,6 +6643,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { assert.isNotEmpty(countQueries); for (const countQuery of countQueries) { for (const detail of yield* planOf(countQuery)) { + // A thread's counts read only each run's status, never its row and payload. + if (/^SEARCH (r|run) /.test(detail)) { + assert.include(detail, "COVERING INDEX", detail); + } if (detail.includes("orchestration_v2_projection_turn_items")) { // Only held input is read from the sparse index; transcript history stays unread. assert.include( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 841cd0a0b5b7d..721006481e83e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -7666,12 +7666,11 @@ export const layer: Layer.Layer = ( SELECT item.payload_json FROM orchestration_v2_projection_turn_items item - INNER JOIN orchestration_v2_projection_runs r ON r.run_id = item.run_id - WHERE r.run_id = presented.run_id - AND r.status = 'failed' + WHERE presented.status = 'failed' AND item.thread_id = t.thread_id + AND item.run_id = presented.run_id AND item.type = 'error' AND item.status = 'failed' - AND item.node_id IS json_extract(r.payload_json, '$.rootNodeId') + AND item.node_id IS json_extract(presented.payload_json, '$.rootNodeId') ORDER BY item.updated_at DESC, item.ordinal DESC, item.turn_item_id DESC LIMIT 1 ) AS terminal_failure_payload_json, @@ -7789,9 +7788,9 @@ export const layer: Layer.Layer = projectId === undefined ? sql`` : sql` AND t.project_id = ${projectId}` }${ location === "active" - ? sql` AND json_extract(t.payload_json, '$.archivedAt') IS NULL` + ? sql` AND t.archived_at IS NULL` : location === "archive" - ? sql` AND json_extract(t.payload_json, '$.archivedAt') IS NOT NULL` + ? sql` AND t.archived_at IS NOT NULL` : sql`` }${ unsettledOnly @@ -7800,7 +7799,8 @@ export const layer: Layer.Layer = }${ familyOf === undefined ? sql`` - : sql` AND json_extract(t.payload_json, '$.lineage.rootThreadId') = ( + : sql` AND CASE WHEN json_valid(t.payload_json) + THEN json_extract(t.payload_json, '$.lineage.rootThreadId') END = ( SELECT json_extract(member.payload_json, '$.lineage.rootThreadId') FROM orchestration_v2_projection_threads member WHERE member.thread_id = ${familyOf} @@ -7808,7 +7808,8 @@ export const layer: Layer.Layer = }${ worktreePath === undefined ? sql`` - : sql` AND json_extract(t.payload_json, '$.worktreePath') = ${worktreePath}` + : sql` AND CASE WHEN json_valid(t.payload_json) + THEN json_extract(t.payload_json, '$.worktreePath') END = ${worktreePath}` }${threadIds === undefined ? sql`` : sql` AND t.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)}))`} ORDER BY t.updated_at ASC, t.thread_id ASC `; @@ -7836,6 +7837,7 @@ export const layer: Layer.Layer = WHEN EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS run + INDEXED BY orchestration_v2_projection_runs_run_status_idx WHERE run.run_id = per_run.run_key AND ( run.status IN ('queued', 'cancelled') @@ -7882,7 +7884,10 @@ export const layer: Layer.Layer = SUM(CASE WHEN counts.run_key = '' THEN counts.item_count ELSE 0 END) AS runless_item_count FROM orchestration_v2_projection_turn_item_counts counts - LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_key + -- The primary key would load each run's whole row just for its status. + LEFT JOIN orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_run_status_idx + ON r.run_id = counts.run_key WHERE counts.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) GROUP BY counts.thread_id `; @@ -8742,7 +8747,7 @@ export const layer: Layer.Layer = ) AS has_remote_tasks FROM orchestration_v2_projection_threads t WHERE t.deleted_at IS NULL - AND json_extract(t.payload_json, '$.archivedAt') IS NULL + AND t.archived_at IS NULL ORDER BY t.updated_at ASC, t.thread_id ASC `; return rows.map((row): ActiveShellIndexEntry => ({ diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 6399ee7b85950..c076c0119d2f9 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -96,6 +96,7 @@ import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; import Migration0086 from "./Migrations/086_QueuedTurnInputIndex.ts"; import Migration0087 from "./Migrations/087_ProjectionProjectsCustomTitle.ts"; import Migration0088 from "./Migrations/088_TurnItemTypeStatusIndex.ts"; +import Migration0089 from "./Migrations/089_ShellThreadReadIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -187,6 +188,7 @@ const migrationEntries = [ [86, "QueuedTurnInputIndex", Migration0086], [87, "ProjectionProjectsCustomTitle", Migration0087], [88, "TurnItemTypeStatusIndex", Migration0088], + [89, "ShellThreadReadIndexes", Migration0089], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/089_ShellThreadReadIndexes.ts b/apps/server/src/persistence/Migrations/089_ShellThreadReadIndexes.ts new file mode 100644 index 0000000000000..25133aa84be65 --- /dev/null +++ b/apps/server/src/persistence/Migrations/089_ShellThreadReadIndexes.ts @@ -0,0 +1,26 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Keep these expressions identical to ProjectionStore's shell worktree and family filters. + // json_valid keeps a malformed payload writable, as the other payload indexes do. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_threads_worktree_path_idx + ON orchestration_v2_projection_threads( + CASE WHEN json_valid(payload_json) THEN json_extract(payload_json, '$.worktreePath') END + ) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_threads_root_thread_idx + ON orchestration_v2_projection_threads( + CASE WHEN json_valid(payload_json) THEN json_extract(payload_json, '$.lineage.rootThreadId') END + ) + `; + // Shell item counts read only a run's status; the primary key would load the whole row. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_run_status_idx + ON orchestration_v2_projection_runs(run_id, status) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index a2bba7d790eaf..d1158a15e2cbe 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -68,6 +68,7 @@ describe("V2 preview upgrade", () => { [86, "QueuedTurnInputIndex"], [87, "ProjectionProjectsCustomTitle"], [88, "TurnItemTypeStatusIndex"], + [89, "ShellThreadReadIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -179,6 +180,7 @@ describe("V2 preview upgrade", () => { [86, "QueuedTurnInputIndex"], [87, "ProjectionProjectsCustomTitle"], [88, "TurnItemTypeStatusIndex"], + [89, "ShellThreadReadIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );