From 3ae93a058a93ead1e0d5f28934eb75816a4c201f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 11:33:18 +0000 Subject: [PATCH 1/2] test(server): shell worktree reads and item counts avoid reading every thread and run row Failing tests for #36139: a worktreePath shell read scans every thread payload, and the shell item counts read each run's table row for its status. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) 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( From f91a7d32cf368ceebace276ed6da381e10006e49 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 11:44:08 +0000 Subject: [PATCH 2/2] perf(server): shell reads of a worktree or family no longer parse every thread payload Shell thread reads filtered archivedAt, worktreePath and lineage.rootThreadId with json_extract on payload_json, so even a 2-thread worktree read scanned and parsed every thread. Shell item counts joined runs by primary key just to read status, loading each run's full row. Archive filters now use the existing archived_at column. A migration adds expression indexes for the worktree and family filters (guarded by json_valid like the other payload indexes, so malformed rows stay writable) and a (run_id, status) index the item counts read through. The terminal-failure lookup reads the already-joined presented run instead of re-joining runs, which keeps its error lookup on the failed-error index. Fixes pingdotgg/t3code#36139 Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ProjectionStore.ts | 25 +++++++++++------- apps/server/src/persistence/Migrations.ts | 2 ++ .../Migrations/089_ShellThreadReadIndexes.ts | 26 +++++++++++++++++++ .../reconcileV2PreviewMigration.test.ts | 2 ++ 4 files changed, 45 insertions(+), 10 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/089_ShellThreadReadIndexes.ts 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:" }))), );