From 9e0d6ecca22a8dc4241a48bf6495640e46a52d4e Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 5 Nov 2026 22:27:13 -0300 Subject: [PATCH 1/7] fix(server): thread shell reads no longer scan every finished run The shell query looked for live runs by walking a thread's whole ordinal index and reading each run row, and sorted every run to find the one that ended last. Use the status index for the live-run lookups and add an index matching the latest-executed-run ordering. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 107 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 4 + apps/server/src/persistence/Migrations.ts | 2 + .../081_ProjectionRunsExecutedIndex.ts | 14 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 5 files changed, 129 insertions(+) create mode 100644 apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index c584dd27a8d23..a433de4e6eb74 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6522,6 +6522,113 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect( + "reads the shell of a thread with a long finished run history without scanning it", + () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const suffix = "shell-long-history"; + const threadId = ThreadId.make(`thread:${suffix}`); + const now = yield* DateTime.now; + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Long finished history", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + for (let ordinal = 1; ordinal <= 3; ordinal++) { + const runId = RunId.make(`run:${suffix}:${ordinal}`); + const rootNodeId = NodeId.make(`node:${suffix}:${ordinal}`); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:run-${ordinal}`), + type: "run.created", + threadId, + runId, + nodeId: rootNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${suffix}:${ordinal}`), + rootNodeId, + activeAttemptId: null, + status: "completed" as const, + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + } + + const { statements, tracer } = traceSqlStatements(); + const shell = yield* projectionStore + .getThreadShell(threadId) + .pipe(Effect.withTracer(tracer)); + assert.isNotNull(shell); + const statement = statements.find((query) => query.includes("AS activity_run_status")); + assert.isDefined(statement); + const plan = yield* sql.unsafe<{ + readonly id: number; + readonly parent: number; + readonly detail: string; + }>(`EXPLAIN QUERY PLAN ${statement}`); + // With no live run, a lookup that walks the ordinal index reads every run the thread has had. + const liveRunLookups = plan.filter((row) => + row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_"), + ); + assert.isAbove(liveRunLookups.length, 0); + for (const lookup of liveRunLookups) { + assert.include( + lookup.detail, + "orchestration_v2_projection_runs_thread_status_idx (thread_id=? AND status=?)", + ); + } + // The newest run needs no sort, but the run that ended last must come from an ordered index. + const candidateLookups = plan.filter((row) => + row.detail.startsWith("SEARCH candidate USING INDEX orchestration_v2_projection_runs_"), + ); + assert.isAbove(candidateLookups.length, 0); + const sortedCandidates = plan.filter( + (row) => + row.detail === "USE TEMP B-TREE FOR ORDER BY" && + candidateLookups.some((lookup) => lookup.parent === row.parent), + ); + assert.deepEqual(sortedCandidates, []); + }), + ); + it.effect("selects the latest waiting secret only from active runs", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 23874c6014fe0..46c5879c646c9 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6171,6 +6171,7 @@ export const layer: Layer.Layer = ( SELECT r.status FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6185,6 +6186,7 @@ export const layer: Layer.Layer = r.requested_at ) FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6246,6 +6248,7 @@ export const layer: Layer.Layer = SELECT secret.payload_json -- Keep runs outermost so completed history is never scanned for a pending secret. FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx CROSS JOIN orchestration_v2_projection_turn_items secret INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx WHERE r.thread_id = t.thread_id @@ -6294,6 +6297,7 @@ export const layer: Layer.Layer = LEFT JOIN orchestration_v2_projection_runs active ON active.run_id = ( SELECT r.run_id FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running') AND NOT ${isSilentShellRun("r")} diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 6a9c57e18cc43..475ec454f5bf9 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -83,6 +83,7 @@ import Migration0077 from "./Migrations/077_RepairBooleanSettledOverride.ts"; import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationIndex.ts"; import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; +import Migration0081 from "./Migrations/081_ProjectionRunsExecutedIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -166,6 +167,7 @@ const migrationEntries = [ [78, "DropOrchestrationEventCorrelationIndex", Migration0078], [79, "DelegatedRecoveryIndexes", Migration0079], [80, "ScheduledTaskThreadMode", Migration0080], + [81, "ProjectionRunsExecutedIndex", Migration0081], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts b/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts new file mode 100644 index 0000000000000..74838b3f96909 --- /dev/null +++ b/apps/server/src/persistence/Migrations/081_ProjectionRunsExecutedIndex.ts @@ -0,0 +1,14 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Matches the shell's latest-executed-run ordering, so it reads the newest rows instead of sorting every run. + yield* sql` + CREATE INDEX orchestration_v2_projection_runs_executed_idx + ON orchestration_v2_projection_runs( + thread_id, (completed_at IS NULL), completed_at, ordinal, run_id + ) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 84fa4ddb44c77..5d5f524d0ec96 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -60,6 +60,7 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProjectionRunsExecutedIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -163,6 +164,7 @@ describe("V2 preview upgrade", () => { [78, "DropOrchestrationEventCorrelationIndex"], [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], + [81, "ProjectionRunsExecutedIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 676afc44ec3e826a630a2d61ad2a8743d6e67adf Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 5 Nov 2026 22:36:10 -0300 Subject: [PATCH 2/7] test(server): require every live-run shell lookup to use the status index The plan assertion passed if any one lookup used the index, and a lookup that fell back to a scan was filtered out instead of failing. Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/orchestration-v2/ProjectionStore.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a433de4e6eb74..771bc7ec5aad7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6608,7 +6608,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const liveRunLookups = plan.filter((row) => row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_"), ); - assert.isAbove(liveRunLookups.length, 0); + assert.lengthOf(liveRunLookups, 4); + assert.isFalse(plan.some((row) => /^SCAN (?:r|candidate)\b/.test(row.detail))); for (const lookup of liveRunLookups) { assert.include( lookup.detail, From e7b5c1d9d54f70953c84e408404d4a43a1bcfe0e Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 14:46:05 -0300 Subject: [PATCH 3/7] perf(server): reads of a long thread no longer touch every row On a thread with tens of thousands of items, the shell item count looked up each run in the main runs table, the command read walked every item to find user messages and errors, and the provider-thread read looked each node up through a non-covering index. Add compact covering indexes for those lookups and hint them where the planner prefers the primary key or the ordinal index. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 143 +++++++++++++++++- .../src/orchestration-v2/ProjectionStore.ts | 23 ++- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/082_CommandReadIndexes.ts | 22 +++ .../reconcileV2PreviewMigration.test.ts | 2 + 5 files changed, 184 insertions(+), 8 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 771bc7ec5aad7..f0119e317b0fc 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6605,8 +6605,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { readonly detail: string; }>(`EXPLAIN QUERY PLAN ${statement}`); // With no live run, a lookup that walks the ordinal index reads every run the thread has had. - const liveRunLookups = plan.filter((row) => - row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_"), + const liveRunLookups = plan.filter( + (row) => + row.detail.startsWith("SEARCH r USING INDEX orchestration_v2_projection_runs_") && + row.detail.includes("(thread_id=?"), ); assert.lengthOf(liveRunLookups, 4); assert.isFalse(plan.some((row) => /^SCAN (?:r|candidate)\b/.test(row.detail))); @@ -6630,6 +6632,143 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("reads a long thread's records for a command without touching every row", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const suffix = "command-long-history"; + const threadId = ThreadId.make(`thread:${suffix}`); + const now = yield* DateTime.now; + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Long command history", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + for (const ordinal of [1, 2]) { + const messageId = MessageId.make(`message:${suffix}:${ordinal}`); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:message-${ordinal}`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId: null, + nodeId: null, + role: "user", + text: `prompt ${ordinal}`, + attachments: [], + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + }, + }); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:item-${ordinal}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`${suffix}:item:${ordinal}`), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: "prompt", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId, + inputIntent: "turn_start", + text: `prompt ${ordinal}`, + attachments: [], + }, + }); + } + + const { statements, tracer } = traceSqlStatements(); + yield* projectionStore + .getThreadRecords(threadId, ["runs", "nodes", "providerThreads", "messages", "turnItems"], { + turnItemTypes: ["user_message", "error"], + messageRoles: ["user"], + omitSettledLeafNodes: true, + omitSettledSubagents: true, + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); + const planOf = Effect.fnUntraced(function* (marker: string) { + const statement = statements.find((query) => query.includes(marker)); + assert.isDefined(statement, marker); + return (yield* sql.unsafe<{ readonly detail: string }>( + `EXPLAIN QUERY PLAN ${statement}`, + )).map((row) => row.detail); + }); + // Each read must reach only the rows it returns; a thread can hold tens of thousands. + const expectations = [ + { + marker: "(type || '') IN (SELECT value FROM json_each", + uses: "turn_items_thread_type_idx (=? AND =?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "MIN(ordinal) AS ordinal", + uses: "turn_items_thread_type_idx (=? AND =?)", + avoids: "turn_items_thread_run_idx", + }, + { + marker: "owner_node_id IN (", + uses: "COVERING INDEX orchestration_v2_projection_nodes_thread_node_idx (=?)", + avoids: "nodes_thread_run_idx", + }, + { + marker: "AS runless_item_count", + uses: "r USING COVERING INDEX orchestration_v2_projection_runs_id_status_idx (=?)", + avoids: "sqlite_autoindex_orchestration_v2_projection_runs_1", + }, + ] as const; + for (const { marker, uses, avoids } of expectations) { + const plan = (yield* planOf(marker)).join("\n"); + assert.include(plan, uses, marker); + assert.notInclude(plan, avoids, marker); + } + }), + ); + it.effect("selects the latest waiting secret only from active runs", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 46c5879c646c9..c80840a4b10e7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3139,8 +3139,14 @@ export const layer: Layer.Layer = ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} - ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} + ${ + // The ordinal index would read every item of the thread to find these types. + filter?.turnItemTypes === undefined + ? sql`` + : sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + } + WHERE ${filter?.turnItemTypes === undefined ? sql`thread_id = ${threadId}` : sql`(thread_id || '') = ${threadId}`} + ${filter?.turnItemTypes === undefined ? sql`` : sql`AND (type || '') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} @@ -3493,9 +3499,10 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_provider_threads WHERE (thread_id = ${threadId} OR owner_node_id IN ( - SELECT node_id + SELECT node_id || '' FROM orchestration_v2_projection_nodes - WHERE thread_id = ${threadId} + INDEXED BY orchestration_v2_projection_nodes_thread_node_idx + WHERE (thread_id || '') = ${threadId} ) OR provider_thread_id IN ( SELECT provider_thread_id @@ -3732,7 +3739,8 @@ export const layer: Layer.Layer = const rows = yield* sql<{ message_id: string; ordinal: number }>` SELECT json_extract(payload_json, '$.messageId') AS message_id, MIN(ordinal) AS ordinal FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} AND type IN ('user_message', 'assistant_message') + INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx + WHERE (thread_id || '') = ${threadId} AND (type || '') IN ('user_message', 'assistant_message') ${loadedMessageIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.messageId') IN (SELECT value FROM json_each(${encodeIdList(loadedMessageIds)}))`} GROUP BY json_extract(payload_json, '$.messageId') `; @@ -6398,7 +6406,10 @@ export const layer: Layer.Layer = WHERE thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) GROUP BY thread_id, run_id ) counts - LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_id + -- The covering index answers each run from a compact page set instead of the runs table. + LEFT JOIN orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_id_status_idx + ON (r.run_id || '') = (counts.run_id || '') GROUP BY counts.thread_id `; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 475ec454f5bf9..0c80c57fae550 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -84,6 +84,7 @@ import Migration0078 from "./Migrations/078_DropOrchestrationEventCorrelationInd import Migration0079 from "./Migrations/079_DelegatedRecoveryIndexes.ts"; import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionRunsExecutedIndex.ts"; +import Migration0082 from "./Migrations/082_CommandReadIndexes.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -168,6 +169,7 @@ const migrationEntries = [ [79, "DelegatedRecoveryIndexes", Migration0079], [80, "ScheduledTaskThreadMode", Migration0080], [81, "ProjectionRunsExecutedIndex", Migration0081], + [82, "CommandReadIndexes", Migration0082], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts b/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts new file mode 100644 index 0000000000000..a7ab4e95074b6 --- /dev/null +++ b/apps/server/src/persistence/Migrations/082_CommandReadIndexes.ts @@ -0,0 +1,22 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Each lets a read of one long thread skip the rows it does not return. Every column a query + // could match is wrapped in `|| ''`, so only a lookup that writes the same expression can use + // them. Plain indexes were chosen for unrelated statements over their partial indexes. + yield* sql` + CREATE INDEX orchestration_v2_projection_turn_items_thread_type_idx + ON orchestration_v2_projection_turn_items((thread_id || ''), (type || ''), ordinal, turn_item_id) + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_nodes_thread_node_idx + ON orchestration_v2_projection_nodes((thread_id || ''), (node_id || '')) + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_runs_id_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 5d5f524d0ec96..39cb418baf47a 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -61,6 +61,7 @@ describe("V2 preview upgrade", () => { [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], [81, "ProjectionRunsExecutedIndex"], + [82, "CommandReadIndexes"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -165,6 +166,7 @@ describe("V2 preview upgrade", () => { [79, "DelegatedRecoveryIndexes"], [80, "ScheduledTaskThreadMode"], [81, "ProjectionRunsExecutedIndex"], + [82, "CommandReadIndexes"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); From 72bd82bc36e0ac446bac9539625519bc94e586ec Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 6 Nov 2026 20:21:18 +0000 Subject: [PATCH 4/7] perf(server): run- and node-scoped turn-item reads keep their own index --- .../src/orchestration-v2/ProjectionStore.test.ts | 11 +++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 16 +++++++++++----- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 4721352fe1411..b15918547a5c0 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6795,6 +6795,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { omitSettledSubagents: true, }) .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["command_execution"], + turnItemRunId: RunId.make(`run:${suffix}`), + }) + .pipe(Effect.withTracer(tracer)); yield* projectionStore.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); const planOf = Effect.fnUntraced(function* (marker: string) { const statement = statements.find((query) => query.includes(marker)); @@ -6810,6 +6816,11 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { uses: "turn_items_thread_type_idx (=? AND =?)", avoids: "turn_items_thread_ordinal_idx", }, + { + marker: "AND type IN (SELECT value FROM json_each", + uses: "SEARCH orchestration_v2_projection_turn_items USING INDEX", + avoids: "turn_items_thread_type_idx", + }, { marker: "MIN(ordinal) AS ordinal", uses: "turn_items_thread_type_idx (=? AND =?)", diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7106c95ea37f2..cedf45bedf321 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3141,6 +3141,11 @@ export const layer: Layer.Layer = readonly work_last_ordinal?: number | null; readonly work_item_count?: number | null; }; + const byTypeOnly = + filter?.turnItemTypes !== undefined && + filter.turnItemRunId === undefined && + filter.turnItemRunIds === undefined && + filter.turnItemNodeIds === undefined; const boundedTurnItemRows: ReadonlyArray = (fields !== undefined && !fields.includes("turnItems")) || (window === undefined && @@ -3158,13 +3163,14 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_turn_items ${ // The ordinal index would read every item of the thread to find these types. - filter?.turnItemTypes === undefined - ? sql`` - : sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + // Run and node filters keep their narrower indexes. + byTypeOnly + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + : sql`` } - WHERE ${filter?.turnItemTypes === undefined ? sql`thread_id = ${threadId}` : sql`(thread_id || '') = ${threadId}`} + WHERE ${byTypeOnly ? sql`(thread_id || '') = ${threadId}` : sql`thread_id = ${threadId}`} ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} - ${filter?.turnItemTypes === undefined ? sql`` : sql`AND (type || '') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} + ${filter?.turnItemTypes === undefined ? sql`` : byTypeOnly ? sql`AND (type || '') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`} ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} From 8f7ed10af92a8fa8c91636f052b171ea6a534902 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 16:32:09 -0300 Subject: [PATCH 5/7] perf(server): turn-start guards read one run instead of the recovery projection isCurrentAttemptInStatus and shouldFinalizeRun loaded and decoded the whole runtime recovery projection to compare a run's status and active attempt, and a turn start calls them about seven times. Read the run by id instead. Co-Authored-By: Claude Sonnet 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 13 ++++++++ .../src/orchestration-v2/ProjectionStore.ts | 20 ++++++++++++ .../ProviderTurnControlService.test.ts | 1 + .../ProviderTurnStartService.test.ts | 5 +++ .../ProviderTurnStartService.ts | 31 ++++++++++--------- 5 files changed, 55 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index b15918547a5c0..98e19055bc76b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6694,6 +6694,19 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { candidateLookups.some((lookup) => lookup.parent === row.parent), ); assert.deepEqual(sortedCandidates, []); + + // A turn-start guard needs one run, not the thread's whole recovery projection. + const runRead = traceSqlStatements(); + const run = yield* projectionStore + .getRun(threadId, RunId.make(`run:${suffix}:2`)) + .pipe(Effect.withTracer(runRead.tracer)); + assert.equal(run?.ordinal, 2); + assert.isFalse( + runRead.statements.some((query) => /turn_items|projection_nodes|messages/.test(query)), + ); + assert.isUndefined( + yield* projectionStore.getRun(threadId, RunId.make(`run:${suffix}:absent`)), + ); }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index cedf45bedf321..5996313397008 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -612,6 +612,10 @@ export interface ProjectionStoreV2Shape { readonly getRuntimeRecoveryProjection: ( threadId: ThreadId, ) => Effect.Effect; + readonly getRun: ( + threadId: ThreadId, + runId: RunId, + ) => Effect.Effect; readonly getPendingNativeUserInputs: ( threadId: ThreadId, providerTurnId: ProviderTurnId, @@ -5507,6 +5511,21 @@ export const layer: Layer.Layer = ) .pipe(Effect.mapError(controlReadError(threadId))); + const getRun: ProjectionStoreV2Shape["getRun"] = (threadId, runId) => + sql + .withTransaction( + Effect.gen(function* () { + yield* requireThread(threadId); + const rows = yield* sql` + SELECT payload_json FROM orchestration_v2_projection_runs + WHERE run_id = ${runId} AND thread_id = ${threadId}`; + return rows[0] === undefined + ? undefined + : yield* decodeRunPayload(rows[0].payload_json); + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); + const canStartQueuedRun: ProjectionStoreV2Shape["canStartQueuedRun"] = (threadId) => sql .withTransaction( @@ -7518,6 +7537,7 @@ export const layer: Layer.Layer = getCheckpointCaptureContext: readOnReader(getCheckpointCaptureContext), getCheckpointRollbackContext: readOnReader(getCheckpointRollbackContext), getRunMessage: readOnReader(getRunMessage), + getRun: readOnReader(getRun), canStartQueuedRun: readOnReader(canStartQueuedRun), getHeldQueuedRuns: readOnReader(getHeldQueuedRuns), getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 55e2a9e50e3ef..6a52db7147cec 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -230,6 +230,7 @@ it.effect( getTurnStartContext: () => Effect.die("unused"), getTurnStartHistory: () => Effect.die("unused"), getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"), + getRun: () => Effect.die("unused getRun"), getPlan: () => Effect.die("unused"), hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"), getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index d1b5168258874..0859fa15e56cc 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -970,6 +970,11 @@ function makeLocalCommandHarness(input: { (m.text.trim().toLowerCase() !== "/compact" || m.attachments.length > 0), ), }), + getRun: (_threadId, id) => + Effect.as( + failReadIfRunning, + projection.runs.find((candidate) => candidate.id === id), + ), getTurnStartHistory: () => input.sideChat !== undefined ? Effect.succeed([]) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 54b16d7aa2aef..2f5cdd03df2bb 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -238,12 +238,13 @@ export const layer: Layer.Layer< >; }) => { const isCurrentAttemptInStatus = (expectedStatus: OrchestrationV2Run["status"]) => - projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( - Effect.map((current) => { - const run = current.runs.find((candidate) => candidate.id === input.runId); - return run?.activeAttemptId === input.attemptId && run.status === expectedStatus; - }), - ); + projectionStore + .getRun(input.threadId, input.runId) + .pipe( + Effect.map( + (run) => run?.activeAttemptId === input.attemptId && run.status === expectedStatus, + ), + ); return { isCurrentAttemptInStatus, loadInheritedBackgroundTurnItems: () => @@ -261,15 +262,15 @@ export const layer: Layer.Layer< ), shouldStartProviderTurn: () => isCurrentAttemptInStatus("running"), shouldFinalizeRun: () => - projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( - Effect.map((current) => { - const run = current.runs.find((candidate) => candidate.id === input.runId); - return ( - run?.activeAttemptId === input.attemptId && - (run.status === "starting" || run.status === "running") - ); - }), - ), + projectionStore + .getRun(input.threadId, input.runId) + .pipe( + Effect.map( + (run) => + run?.activeAttemptId === input.attemptId && + (run.status === "starting" || run.status === "running"), + ), + ), hasUnpairedRunInterruptRequest: () => projectionStore .hasUnpairedRunInterruptRequest( From 6d1e4e07986a8f7c15b6c86d312d9cc77f35f7dd Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 18:44:38 -0300 Subject: [PATCH 6/7] perf(server): settling a subagent and MCP calls read less of a long parent thread Settling an app-owned subagent loaded every node, user message, subagent, provider thread/turn and context transfer of the parent, to use one task, its node, its item and its delivery cohort. On a dispatcher thread with 35k nodes and 7k prompts that cost over a second of reads per child. It now reads the records it names, and takes the result-transfer check from the child side. MCP calls that identify their caller loaded the caller thread's provider threads and turns, which no tool reads. Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/mcp/OrchestratorMcpService.ts | 21 +----- .../DelegatedCompletionDelivery.test.ts | 53 +++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 65 +++++++++++++------ 3 files changed, 101 insertions(+), 38 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 9d15408483a3e..e6d6ac4593729 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1081,15 +1081,7 @@ const make = Effect.gen(function* () { const loadProjection = (threadId: ThreadId, includeUserMessages = true) => loadThreadRecords( threadId, - [ - "runs", - "messages", - "providerThreads", - "providerTurns", - "runtimeRequests", - "contextTransfers", - "turnItems", - ], + ["runs", "messages", "runtimeRequests", "contextTransfers", "turnItems"], { turnItemTypes: [], messageRoles: includeUserMessages ? ["user"] : [] }, ); @@ -1123,16 +1115,7 @@ const make = Effect.gen(function* () { threadManagement .getProjectThreadRecords( { projectId, threadId }, - [ - "runs", - "messages", - "subagents", - "providerThreads", - "providerTurns", - "runtimeRequests", - "contextTransfers", - "turnItems", - ], + ["runs", "messages", "subagents", "runtimeRequests", "contextTransfers", "turnItems"], { turnItemTypes: [], messageRoles: ["user"] }, ) .pipe(Effect.mapError(threadManagementFailure)); diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index db3a2759e3ac5..6fb060d164db5 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2763,6 +2763,59 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + it.effect("settles a child without reading all of the parent's nodes", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:bounded-finalize-parent"); + const projectId = ProjectId.make("project:bounded-finalize"); + const runId = RunId.make("run:bounded-finalize-parent"); + const rootNodeId = NodeId.make("node:bounded-finalize-parent"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:bounded-finalize-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "bounded-finalize-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + const statements: 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; + }, + }); + yield* orchestrator.recoverDelegatedTasks.pipe(Effect.withTracer(tracer)); + const settled = yield* orchestrator.getThreadProjection(threadId); + assert.equal(settled.subagents.find((task) => task.id === child.taskId)?.status, "completed"); + // A dispatcher parent holds thousands of nodes; settling one child names its own. + const nodeReads = statements.filter((query) => + query.includes("FROM orchestration_v2_projection_nodes"), + ); + assert.isNotEmpty(nodeReads); + assert.isTrue(nodeReads.every((query) => query.includes("node_id IN"))); + }), + ); + it.effect("holds a restart-cancelled child for its continuation's result", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcec0a13cf20a..e9076edbefbb5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12803,21 +12803,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItems: childResult.turnItems, }; const parentThreadId = childControls.thread.lineage.parentThreadId; - const parentProjection = yield* projectionStore.getThreadRecords( + // A dispatcher parent holds thousands of tasks, runs and prompts, so read only this task's + // records and the delivery cohort the completion plan joins. + const taskRecords = yield* projectionStore.getThreadRecords( parentThreadId, - [ - "runs", - "messages", - "subagents", - "providerThreads", - "providerTurns", - "nodes", - "turnItems", - "contextTransfers", - ], - { turnItemTypes: ["subagent"], messageRoles: ["user"] }, + ["subagents", "nodes", "turnItems"], + { + subagentIds: [options.task.id], + nodeIds: [options.task.id], + turnItemNodeIds: [options.task.id], + turnItemTypes: ["subagent"], + }, ); - const task = parentProjection.subagents.find( + const task = taskRecords.subagents.find( (candidate) => candidate.id === options.task.id && candidate.origin === "app_owned" && @@ -12826,7 +12824,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (task === undefined) { return; } - const existingResultTransfer = parentProjection.contextTransfers.find( + // The result transfer runs from the child, so the child's side lists it too. + const { contextTransfers } = yield* projectionStore.getThreadRecords(childThreadId, [ + "contextTransfers", + ]); + const existingResultTransfer = contextTransfers.find( (transfer) => transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId && @@ -12840,12 +12842,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const now = yield* DateTime.now; const result = subagentResultForRun(childProjection, { ...childRun, status: outcome }); - const parentRun = - task.runId === null + const taskRunRecords = yield* projectionStore.getThreadRecords(parentThreadId, ["runs"], { + runIds: task.runId === null ? [] : [task.runId], + }); + const parentRun = taskRunRecords.runs.find((candidate) => candidate.id === task.runId); + const delivery = parentRun?.delegatedCompletion?.delivery; + const deliveryRecords = + delivery == null + ? undefined + : yield* projectionStore.getThreadRecords(parentThreadId, ["messages", "subagents"], { + messageIds: [delivery.messageId], + subagentIds: [...new Set([...delivery.taskIds, task.id])], + }); + const deliveryRunId = deliveryRecords?.messages[0]?.runId ?? null; + const deliveryRun = + deliveryRunId === null || deliveryRunId === parentRun?.id ? undefined - : parentProjection.runs.find((candidate) => candidate.id === task.runId); - const parentNode = parentProjection.nodes.find((candidate) => candidate.id === task.id); - const parentTurnItem = parentProjection.turnItems.find( + : (yield* projectionStore.getThreadRecords(parentThreadId, ["runs"], { + runIds: [deliveryRunId], + })).runs[0]; + const parentProjection = { + thread: taskRecords.thread, + runs: [ + ...(parentRun === undefined ? [] : [parentRun]), + ...(deliveryRun ? [deliveryRun] : []), + ], + messages: deliveryRecords?.messages ?? [], + subagents: deliveryRecords?.subagents ?? [task], + providerTurns: [], + }; + const parentNode = taskRecords.nodes.find((candidate) => candidate.id === task.id); + const parentTurnItem = taskRecords.turnItems.find( (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id, ); const updatedTask: OrchestrationV2Subagent = { From 860ff723ae9901a1f75a40fea1c4d36f1b5a9141 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 6 Nov 2026 19:32:01 -0300 Subject: [PATCH 7/7] perf(server): run-, node- and child-scoped reads name their index Without statistics SQLite answers a turn-item read filtered by run or node by walking the thread's whole ordinal index (2.2 s on a thread with tens of thousands of items), whether the filter is a run id, a run id list or node ids. Name the matching index for each, and drop the always-false OR arm from the run list so the run id can be a search key. A restart continuation does this read once per interrupted run. Finalizing a child at startup loaded every task and context transfer of its parent to find its own; read the child's tasks through the child-thread index and its result transfers from the child side. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 13 +++++-- .../orchestration-v2/ProjectionStore.test.ts | 38 +++++++++++++++++-- .../src/orchestration-v2/ProjectionStore.ts | 24 +++++++++--- 3 files changed, 62 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e9076edbefbb5..c040da43a5aef 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13033,8 +13033,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) return; const progress = delegatedTaskProgress(childControls); - const parent = yield* projectionStore.getThreadRecords(parentThreadId, [ - "subagents", + // A dispatcher parent holds thousands of tasks and transfers; this child owns a few of each. + const parent = yield* projectionStore.getThreadRecords(parentThreadId, ["subagents"], { + subagentChildThreadId: childThreadId, + }); + const { contextTransfers } = yield* projectionStore.getThreadRecords(childThreadId, [ "contextTransfers", ]); for (const task of parent.subagents) { @@ -13044,9 +13047,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (task.id === childControls.thread.forkedFrom.nodeId && progress.state !== "result_available") || (task.result !== null && - parent.contextTransfers.some( + contextTransfers.some( (transfer) => - transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId, + transfer.type === "subagent_result" && + transfer.sourceThreadId === childThreadId && + transfer.targetThreadId === parentThreadId, )) ) continue; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 98e19055bc76b..107ed4ff4bb7f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6814,6 +6814,23 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { turnItemRunId: RunId.make(`run:${suffix}`), }) .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["run_interrupt_request"], + turnItemRunIds: [RunId.make(`run:${suffix}`)], + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["turnItems"], { + turnItemTypes: ["subagent"], + turnItemNodeIds: [NodeId.make(`node:${suffix}`)], + }) + .pipe(Effect.withTracer(tracer)); + yield* projectionStore + .getThreadRecords(threadId, ["subagents"], { + subagentChildThreadId: ThreadId.make(`thread:${suffix}:child`), + }) + .pipe(Effect.withTracer(tracer)); yield* projectionStore.getThreadShell(threadId).pipe(Effect.withTracer(tracer)); const planOf = Effect.fnUntraced(function* (marker: string) { const statement = statements.find((query) => query.includes(marker)); @@ -6830,9 +6847,24 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { avoids: "turn_items_thread_ordinal_idx", }, { - marker: "AND type IN (SELECT value FROM json_each", - uses: "SEARCH orchestration_v2_projection_turn_items USING INDEX", - avoids: "turn_items_thread_type_idx", + marker: "AND run_id = ?", + uses: "turn_items_thread_run_idx (thread_id=? AND run_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND run_id IN (SELECT value FROM json_each", + uses: "turn_items_thread_run_idx (thread_id=? AND run_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND node_id IN (SELECT value FROM json_each", + uses: "turn_items_node_ordinal_idx (node_id=?)", + avoids: "turn_items_thread_ordinal_idx", + }, + { + marker: "AND child_thread_id = ?", + uses: "subagents_child_thread_idx (child_thread_id=?)", + avoids: "subagents_thread_idx", }, { marker: "MIN(ordinal) AS ordinal", diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 5996313397008..0c889ad7dbe41 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -353,6 +353,8 @@ export interface ProjectionRecordFilter { readonly contextTransferStatuses?: ReadonlyArray; readonly subagentIds?: ReadonlyArray; readonly subagentRunIds?: ReadonlyArray; + /** Only the tasks whose child thread this is. */ + readonly subagentChildThreadId?: ThreadId; readonly nodeIds?: ReadonlyArray; readonly turnItemRequestIds?: ReadonlyArray; readonly turnItemTypes?: ReadonlyArray; @@ -3145,6 +3147,10 @@ export const layer: Layer.Layer = readonly work_last_ordinal?: number | null; readonly work_item_count?: number | null; }; + const byRun = + filter?.turnItemRunId !== undefined || + (filter?.turnItemRunIds !== undefined && !filter.turnItemRunIds.includes(null)); + const byNode = !byRun && filter?.turnItemNodeIds !== undefined; const byTypeOnly = filter?.turnItemTypes !== undefined && filter.turnItemRunId === undefined && @@ -3166,11 +3172,15 @@ export const layer: Layer.Layer = SELECT payload_json FROM orchestration_v2_projection_turn_items ${ - // The ordinal index would read every item of the thread to find these types. - // Run and node filters keep their narrower indexes. - byTypeOnly - ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` - : sql`` + // Without statistics SQLite walks the thread's ordinal index and reads every + // item of a long thread; each filter names the index that reaches only its rows. + byRun + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx` + : byNode + ? sql`INDEXED BY orchestration_v2_projection_turn_items_node_ordinal_idx` + : byTypeOnly + ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_type_idx` + : sql`` } WHERE ${byTypeOnly ? sql`(thread_id || '') = ${threadId}` : sql`thread_id = ${threadId}`} ${filter?.turnItemRequestIds === undefined ? sql`` : sql`AND json_extract(payload_json, '$.requestId') IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRequestIds)}))`} @@ -3178,7 +3188,7 @@ export const layer: Layer.Layer = ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`} ${filter?.turnItemNodeIds === undefined ? sql`` : sql`AND node_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemNodeIds)}))`} - ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} + ${filter?.turnItemRunIds === undefined ? sql`` : filter.turnItemRunIds.includes(null) ? sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR run_id IS NULL)` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))}))`} ORDER BY ordinal ASC, turn_item_id ASC ` : yield* sql` @@ -3464,7 +3474,9 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_subagents AS subagent + ${filter?.subagentChildThreadId === undefined ? sql`` : sql`INDEXED BY orchestration_v2_projection_subagents_child_thread_idx`} WHERE thread_id = ${threadId} + ${filter?.subagentChildThreadId === undefined ? sql`` : sql`AND child_thread_id = ${filter.subagentChildThreadId}`} ${filter?.subagentIds === undefined ? sql`` : sql`AND subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentIds)}))`} ${filter?.subagentRunIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentRunIds)}))`} ${