From c42f4426e2a42bdac1de7992960d884b44190b99 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 11 Nov 2026 12:56:25 -0800 Subject: [PATCH 1/4] perf(server): delegated tasks start and finish without reading the whole parent thread Starting, finishing and cancelling a delegated task read whole record sets of the parent thread to use one run, one node, one turn item and one task. On a long parent that decoded tens of megabytes for each task. These paths now read those records by id: - Task start reads the parent run and node by id, then that run's provider thread, attempt and provider turn. - Task finish reads the task, its node and its turn item by id, the child's own transfers, and only the run, message and tasks that delivery planning uses. - Task cancel reads the one task. - The stop check reads one entry of the partial stop-request index. ProjectionStore gains three id filters (attemptIds, providerTurnIds, subagentChildThreadIds) and hasRunInterruptRequest. A turn item read by node id now uses the node index. Co-Authored-By: Claude Fable 5.2 --- .../Orchestrator.control-reads.test.ts | 222 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 156 +++++++----- .../orchestration-v2/ProjectionStore.test.ts | 53 +++-- .../src/orchestration-v2/ProjectionStore.ts | 53 ++++- .../ProviderTurnControlService.test.ts | 1 + 5 files changed, 409 insertions(+), 76 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index d79d1a2d9ca97..5d032beb18941 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -1,3 +1,5 @@ +import * as NodeSqlite from "node:sqlite"; + import { assert, it } from "@effect/vitest"; import { CommandId, @@ -22,10 +24,13 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Metric from "effect/Metric"; +import * as Option from "effect/Option"; import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; +import type * as Tracer from "effect/Tracer"; import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import { orchestrationCommandDuration, orchestrationCommandsTotal, @@ -41,6 +46,13 @@ import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +declare module "node:sqlite" { + interface StatementSync { + // Node 26 has this; @types/node 24 does not declare it yet. + stat(counter: "vmStep"): number; + } +} + const instanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId, model: "gpt-5.1-codex" }; const adapter = { @@ -1628,6 +1640,216 @@ it.effect.each([false, true])("rejects thread.create for an existing id (deleted }).pipe(Effect.provide(layerTest)), ); +it.effect("starts and finishes a delegated task without reading the parent's history", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const eventSink = yield* EventSink.EventSinkV2; + const sql = yield* SqlClient.SqlClient; + const parentThreadId = ThreadId.make("thread:long-parent"); + const now = yield* DateTime.now; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-long-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:long-parent"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-long-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:long-parent"), + text: "Delegate the work", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const parent = yield* projections.getThreadRecords(parentThreadId, [ + "runs", + "attempts", + "providerThreads", + ]); + const parentRun = parent.runs[0]!; + const providerTurnId = ProviderTurnId.make("provider-turn:long-parent"); + const nativeRef = (nativeId: string) => ({ + driver: adapter.driver, + nativeId, + strength: "strong" as const, + }); + for (const event of [ + { + type: "provider-thread.updated", + payload: { ...parent.providerThreads[0]!, nativeThreadRef: nativeRef("native:parent") }, + }, + // A steered run: its attempt points at a provider turn that another attempt started. + { + type: "provider-turn.updated", + payload: { + id: providerTurnId, + providerThreadId: parentRun.providerThreadId!, + nodeId: parentRun.rootNodeId!, + runAttemptId: null, + nativeTurnRef: nativeRef("native:parent:turn"), + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + { type: "run-attempt.updated", payload: { ...parent.attempts[0]!, providerTurnId } }, + ] as const) { + yield* projections.apply({ + ...event, + id: EventId.make(`event:long-parent:${event.type}`), + threadId: parentThreadId, + runId: parentRun.id, + occurredAt: now, + }); + } + + // Keep the statements of the task start and finish operations, not of the recovery around them. + const taskSpans = [ + "orchestrationV2.dispatch.delegatedTaskRequest", + "Orchestrator.finalizeAppOwnedSubagent", + ]; + const statements: Array]> = []; + const recordStatements = Effect.provideService( + Statement.CurrentTransformer, + (statement, _make, _fiber, span) => { + for ( + let parent = Option.some(span); + Option.isSome(parent) && parent.value._tag === "Span"; + parent = parent.value.parent + ) { + if (taskSpans.includes(parent.value.name)) { + statements.push(statement.compile()); + break; + } + } + return Effect.succeed(statement); + }, + ); + yield* orchestrator + .dispatch({ + type: "delegated_task.request", + commandId: CommandId.make("delegate-long-parent"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Do the work", + completionWake: "always", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "agent", + creationSource: "mcp", + }) + .pipe(recordStatements); + const started = yield* projections.getThreadRecords(parentThreadId, ["subagents", "turnItems"]); + const task = started.subagents[0]!; + const childThreadId = task.childThreadId!; + assert.equal( + started.turnItems.find((item) => item.type === "subagent")?.providerTurnId, + providerTurnId, + ); + const child = yield* projections.getThreadRecords(childThreadId, ["runs", "contextTransfers"]); + const spawnPoint = child.contextTransfers[0]?.sourcePoint; + assert.deepEqual( + [spawnPoint?.providerThreadRef, spawnPoint?.providerTurnRef], + [nativeRef("native:parent"), nativeRef("native:parent:turn")], + ); + + // The run listener skips reconcile commands, so the recovery below finishes the task here. + yield* eventSink.write({ + commandId: CommandId.make("command:runtime-reconcile:long-parent"), + events: [ + { + id: EventId.make("event:long-parent:child-completed"), + type: "run.updated", + threadId: childThreadId, + runId: child.runs[0]!.id, + occurredAt: now, + payload: { ...child.runs[0]!, status: "completed", completedAt: now }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks.pipe(recordStatements); + const finished = yield* projections.getThreadRecords(parentThreadId, [ + "subagents", + "nodes", + "turnItems", + "runs", + ]); + assert.deepEqual( + [ + finished.subagents[0]?.status, + finished.subagents[0]?.completionDelivery?.state, + finished.nodes.find((node) => node.id === task.id)?.status, + finished.turnItems.find((item) => item.type === "subagent")?.status, + finished.runs[0]?.delegatedCompletion?.delivery?.taskIds, + ], + ["completed", "claimed", "completed", "completed", [task.id]], + ); + + // Start and finish read the task's own records, so their VM steps, which are deterministic, + // must stay flat on a copy with a long parent history. Left out: the counts that number the + // task and its wake, which walk one index, and provider turn history, whose read has no + // index on the attempt and walks the thread's turn rows without decoding them. + const copyPath = "file:/delegated-task-reads?vfs=memdb"; + using copy = new NodeSqlite.DatabaseSync(copyPath); + yield* sql`VACUUM INTO ${copyPath}`; + const reads = statements.filter( + ([query]) => /^\s*(SELECT|WITH)\b/.test(query) && !query.includes("COUNT(*)"), + ); + const readSteps = () => + reads.reduce((steps, [query, params]) => { + const statement = copy.prepare(query); + statement.all(...(params as ReadonlyArray)); + return steps + statement.stat("vmStep"); + }, 0); + const shortSteps = readSteps(); + assert.isAbove(shortSteps, 0); + const addHistory = (table: string, key: string, renamed: ReadonlyArray = []) => { + const name = `orchestration_v2_projection_${table}`; + const columns = copy + .prepare(`SELECT name FROM pragma_table_info('${name}')`) + .all() + .map((column) => String(column.name)); + const values = columns.map((column) => + column === key || renamed.includes(column) + ? `CASE WHEN ${column} = '${parentThreadId}' THEN ${column} ELSE ${column} || ':history:' || history.n END` + : column === "ordinal" + ? "ordinal + history.n" + : column, + ); + copy.exec(`WITH RECURSIVE history(n) AS (SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 300) + INSERT INTO ${name} (${columns.join(", ")}) + SELECT ${values.join(", ")} FROM ${name}, history + WHERE '${parentThreadId}' IN (${columns.filter((column) => column.endsWith("thread_id")).join(", ")})`); + }; + addHistory("nodes", "node_id"); + addHistory("turn_items", "turn_item_id", ["node_id"]); + addHistory("subagents", "subagent_id", ["child_thread_id"]); + addHistory("context_transfers", "context_transfer_id", [ + "source_thread_id", + "target_thread_id", + ]); + addHistory("runs", "run_id"); + addHistory("run_attempts", "attempt_id", ["run_id"]); + addHistory("messages", "message_id"); + assert.isAtMost(readSteps(), shortSteps * 1.1); + }).pipe(Effect.provide(layerTest)), +); + it.effect("replays a thread.create receipt without changing the thread", () => 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 7af71e48d67ea..e02d52eaeadf2 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1689,6 +1689,59 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]), ]; + /** One run's provider thread, active attempt and provider turn, as `contextSourcePointForRun` reads them. */ + const readRunProviderRecords = + /* The delegated task operation that calls this owns the span. */ + Effect.fnUntraced(function* (threadId: ThreadId, run: OrchestrationV2Run) { + const attemptIds = run.activeAttemptId === null ? [] : [run.activeAttemptId]; + const records = yield* projectionStore.getThreadRecords( + threadId, + ["providerThreads", "providerTurns", "attempts"], + { + providerThreadIds: run.providerThreadId === null ? [] : [run.providerThreadId], + providerTurnAttemptIds: attemptIds, + attemptIds, + }, + ); + const attemptTurnId = records.attempts[0]?.providerTurnId; + if (records.providerTurns.length > 0 || attemptTurnId == null) return records; + // A steered run's attempt can point at a provider turn that another attempt started. + const { providerTurns } = yield* projectionStore.getThreadRecords( + threadId, + ["providerTurns"], + { providerTurnIds: [attemptTurnId] }, + ); + return { ...records, providerTurns }; + }); + + /** The parent records `planDelegatedCompletionDelivery` reads for one task: its run, and the queued wake's message, run and tasks. */ + const readDelegatedCompletionPlanRecords = + /* The delegated task operation that calls this owns the span. */ + Effect.fnUntraced(function* (parentThreadId: ThreadId, task: OrchestrationV2Subagent) { + const { + thread, + runs: [parentRun], + } = yield* projectionStore.getThreadRecords(parentThreadId, ["runs"], { + runIds: task.runId === null ? [] : [task.runId], + }); + const delivery = parentRun?.delegatedCompletion?.delivery; + if (delivery == null) return { thread, parentRun, runs: [], messages: [], subagents: [] }; + const { messages } = yield* projectionStore.getThreadRecords(parentThreadId, ["messages"], { + messageIds: [delivery.messageId], + messageRoles: ["user"], + }); + const deliveryRunId = messages[0]?.runId; + const { runs, subagents } = yield* projectionStore.getThreadRecords( + parentThreadId, + ["runs", "subagents"], + { + runIds: deliveryRunId == null ? [] : [deliveryRunId], + subagentIds: [...delivery.taskIds, task.id], + }, + ); + return { thread, parentRun, runs, messages, subagents }; + }); + const offerDelegatedCompletionDelivery = /* This helper runs for individual orchestration events; the enclosing operation owns the span. */ Effect.fnUntraced(function* (threadId: ThreadId, parentRunId: RunId) { @@ -8419,23 +8472,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, effects: Ref.Ref>, ) { - const parentProjection = yield* projectionStore - .getThreadRecords( - command.parentThreadId, - ["runs", "nodes", "providerThreads", "providerTurns", "attempts"], - // The parent node must belong to the active run, whose nodes this keeps. - { omitSettledLeafNodes: true }, - ) - .pipe( - Effect.mapError( - (cause) => - new OrchestratorProjectionError({ - threadId: command.parentThreadId, - cause, - }), - ), - ); - const parentRun = parentProjection.runs.find( + const parentProjectionError = (cause: unknown) => + new OrchestratorProjectionError({ threadId: command.parentThreadId, cause }); + const parentControls = yield* projectionStore + .getThreadRecords(command.parentThreadId, ["runs", "nodes"], { + runIds: [command.parentRunId], + nodeIds: [command.parentNodeId], + }) + .pipe(Effect.mapError(parentProjectionError)); + const parentRun = parentControls.runs.find( (candidate) => candidate.id === command.parentRunId, ); if (parentRun === undefined || !isActiveRun(parentRun)) { @@ -8452,7 +8497,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Parent run ${command.parentRunId} is stopping.`, }); } - const parentNode = parentProjection.nodes.find( + const parentNode = parentControls.nodes.find( (candidate) => candidate.id === command.parentNodeId, ); if ( @@ -8466,6 +8511,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Parent node ${command.parentNodeId} is not part of active run ${parentRun.id}.`, }); } + const parentProjection = yield* readRunProviderRecords( + command.parentThreadId, + parentRun, + ).pipe(Effect.mapError(parentProjectionError)); const targetAdapter = yield* providerAdapters.get(command.modelSelection.instanceId).pipe( Effect.mapError( @@ -11460,16 +11509,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command: OrchestrationV2ServerCommand, threadId: ThreadId, runId: RunId, - ) => - projectionStore - .getThreadRecords(threadId, ["turnItems"], { - turnItemTypes: ["run_interrupt_request"], - turnItemRunIds: [runId], - }) - .pipe( - Effect.map(({ turnItems }) => turnItems.length > 0), - mapDispatchError(command), - ); + ) => projectionStore.hasRunInterruptRequest(threadId, runId).pipe(mapDispatchError(command)); const markStoppedRun = Effect.fnUntraced(function* (input: { readonly command: Extract< @@ -12955,7 +12995,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio )(function* (input: { readonly parentProjection: Pick< OrchestrationV2ThreadProjection, - "thread" | "messages" | "runs" | "subagents" | "providerTurns" + "thread" | "messages" | "runs" | "subagents" >; readonly parentRun: OrchestrationV2Run | undefined; readonly task: OrchestrationV2Subagent; @@ -13180,7 +13220,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const parent = yield* projectionStore.getThreadRecords( childControls.thread.lineage.parentThreadId, ["nodes", "turnItems"], - { turnItemTypes: ["subagent"] }, + { nodeIds: [task.id], turnItemNodeIds: [task.id], turnItemTypes: ["subagent"] }, ); const now = yield* DateTime.now; const scope = { @@ -13247,21 +13287,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItems: childResult.turnItems, }; const parentThreadId = childControls.thread.lineage.parentThreadId; - const parentProjection = yield* projectionStore.getThreadRecords( + // The task's node and turn item share its id. + const parentTask = 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 = parentTask.subagents.find( (candidate) => candidate.id === options.task.id && candidate.origin === "app_owned" && @@ -13273,7 +13310,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // A cancelled task's completed run ends it interrupted. const cancelled = task.cancelledRunId === childRun.id; if (cancelled && terminalStatus === "completed") outcome = terminalStatus = "interrupted"; - const existingResultTransfer = parentProjection.contextTransfers.find( + // Every result the child sent is among its own transfers, far fewer than the parent's. + const childTransfers = yield* projectionStore.getThreadRecords(childThreadId, [ + "contextTransfers", + ]); + const existingResultTransfer = childTransfers.contextTransfers.find( (transfer) => transfer.type === "subagent_result" && transfer.sourceThreadId === childThreadId && @@ -13287,12 +13328,12 @@ 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 - ? undefined - : parentProjection.runs.find((candidate) => candidate.id === task.runId); - const parentNode = parentProjection.nodes.find((candidate) => candidate.id === task.id); - const parentTurnItem = parentProjection.turnItems.find( + const { parentRun, ...parentProjection } = yield* readDelegatedCompletionPlanRecords( + parentThreadId, + task, + ); + const parentNode = parentTask.nodes.find((candidate) => candidate.id === task.id); + const parentTurnItem = parentTask.turnItems.find( (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id, ); const updatedTask: OrchestrationV2Subagent = { @@ -13441,8 +13482,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) return; const progress = delegatedTaskProgress(childControls); - const parent = yield* projectionStore.getThreadRecords(parentThreadId, [ - "subagents", + // Each task of this child names it, so the parent's other tasks stay unread. + const parent = yield* projectionStore.getThreadRecords(parentThreadId, ["subagents"], { + subagentChildThreadIds: [childThreadId], + }); + const { contextTransfers } = yield* projectionStore.getThreadRecords(childThreadId, [ "contextTransfers", ]); for (const task of parent.subagents) { @@ -13454,9 +13498,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (task.id === childControls.thread.forkedFrom.nodeId && (cancelled ? progress.state === "working" : 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; @@ -14764,7 +14810,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio input.parentThreadId, Effect.gen(function* () { const { subagents } = yield* projectionStore - .getThreadRecords(input.parentThreadId, ["subagents"]) + .getThreadRecords(input.parentThreadId, ["subagents"], { subagentIds: [input.taskId] }) .pipe(mapDispatchError(command)); const task = subagents.find( (candidate) => diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 2ba22350330b7..2ec13b70e2b95 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -720,26 +720,51 @@ it.effect("idle compaction filters keep the latest run, live work and the newest // An id filter reads by primary key, not every row of the thread. const byId = traceSqlStatements(); const found = yield* store - .getThreadRecords(threadId, ["runs", "nodes", "subagents", "messages"], { - runIds: [RunId.make("run:idle-compaction:4")], - nodeIds: [nodeId], - subagentIds: [nodeId], - messageIds: [MessageId.make("message:idle-compaction:4")], - }) + .getThreadRecords( + threadId, + ["runs", "nodes", "subagents", "messages", "attempts", "providerTurns"], + { + runIds: [RunId.make("run:idle-compaction:4")], + nodeIds: [nodeId], + subagentIds: [nodeId], + messageIds: [MessageId.make("message:idle-compaction:4")], + attemptIds: [RunAttemptId.make("attempt:idle-compaction")], + providerTurnIds: [ProviderTurnId.make("turn:idle-old")], + }, + ) .pipe(Effect.withTracer(byId.tracer)); assert.deepEqual( found.runs.map((run) => run.id), ["run:idle-compaction:4"], ); - for (const table of ["runs", "nodes", "subagents", "messages"]) { - const query = byId.statements.find((statement) => - statement.includes(`FROM orchestration_v2_projection_${table}`), + assert.deepEqual( + found.providerTurns.map((turn) => turn.id), + ["turn:idle-old"], + ); + // A task's turn item and a child thread's tasks are found through their own indexes. + yield* store + .getThreadRecords(threadId, ["subagents", "turnItems"], { + subagentChildThreadIds: [threadId], + turnItemNodeIds: [nodeId], + }) + .pipe(Effect.withTracer(byId.tracer)); + for (const [table, index] of [ + ...["runs", "nodes", "subagents", "messages", "run_attempts", "provider_turns"].map( + (name) => [name, `sqlite_autoindex_orchestration_v2_projection_${name}_1`] as const, + ), + ["subagents", "orchestration_v2_projection_subagents_child_thread_idx"], + ["turn_items", "orchestration_v2_projection_turn_items_node_ordinal_idx"], + ] as const) { + const plans = yield* Effect.forEach( + byId.statements.filter((statement) => + statement.includes(`FROM orchestration_v2_projection_${table}`), + ), + (query) => sql.unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${query}`), + ); + assert.isTrue( + plans.some((plan) => plan.some((row) => row.detail.includes(`USING INDEX ${index}`))), + `${table} is not read through ${index}`, ); - assert.isDefined(query); - const plan = (yield* sql.unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${query}`)) - .map((row) => row.detail) - .join("\n"); - assert.include(plan, `USING INDEX sqlite_autoindex_orchestration_v2_projection_${table}_1`); } assert.deepEqual( records.providerTurns.map((turn) => turn.id), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9e27210350642..67aa9c2c357e3 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -342,7 +342,9 @@ export interface ProjectionRuntimeResponseContext { export interface ProjectionRecordFilter { readonly providerThreadIds?: ReadonlyArray; + readonly providerTurnIds?: ReadonlyArray; readonly providerTurnAttemptIds?: ReadonlyArray; + readonly attemptIds?: ReadonlyArray; readonly messageRoles?: ReadonlyArray; readonly monitorNotificationsOnly?: boolean; readonly turnItemRunId?: RunId; @@ -362,6 +364,7 @@ export interface ProjectionRecordFilter { readonly contextTransferStatuses?: ReadonlyArray; readonly subagentIds?: ReadonlyArray; readonly subagentRunIds?: ReadonlyArray; + readonly subagentChildThreadIds?: ReadonlyArray; readonly nodeIds?: ReadonlyArray; readonly turnItemRequestIds?: ReadonlyArray; readonly turnItemTypes?: ReadonlyArray; @@ -637,6 +640,11 @@ export interface ProjectionStoreV2Shape { requestId: TurnItemId, resultId: TurnItemId, ) => Effect.Effect; + /** Whether Stop reached the run. Reads one index entry, however long the thread is. */ + readonly hasRunInterruptRequest: ( + threadId: ThreadId, + runId: RunId, + ) => Effect.Effect; readonly isNodeStored: ( node: OrchestrationV2ExecutionNode, ) => Effect.Effect; @@ -3334,7 +3342,7 @@ export const layer: Layer.Layer = ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items - WHERE thread_id = ${threadId} + WHERE ${threadColumn(filter?.turnItemNodeIds)} = ${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?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`} @@ -3576,13 +3584,15 @@ export const layer: Layer.Layer = )) ORDER BY ordinal ASC `, - fields !== undefined && !fields.includes("attempts") + (fields !== undefined && !fields.includes("attempts")) || + (window === undefined && matchesNothing(filter?.attemptIds)) ? Effect.succeed([]) : window === undefined ? sql` SELECT payload_json FROM orchestration_v2_projection_run_attempts - WHERE thread_id = ${threadId} + WHERE ${threadColumn(filter?.attemptIds)} = ${threadId} + ${filter?.attemptIds === undefined ? sql`` : sql`AND attempt_id IN (SELECT value FROM json_each(${encodeIdList(filter.attemptIds)}))`} ORDER BY run_id ASC, attempt_ordinal ASC ` : sql` @@ -3632,14 +3642,19 @@ export const layer: Layer.Layer = ), (fields !== undefined && !fields.includes("subagents")) || (window === undefined && - (matchesNothing(filter?.subagentIds) || matchesNothing(filter?.subagentRunIds))) + matchesNothing( + filter?.subagentIds, + filter?.subagentRunIds, + filter?.subagentChildThreadIds, + )) ? Effect.succeed([]) : window === undefined ? sql` SELECT payload_json FROM orchestration_v2_projection_subagents AS subagent - WHERE ${threadColumn(filter?.subagentIds)} = ${threadId} + WHERE ${threadColumn(filter?.subagentIds ?? filter?.subagentChildThreadIds)} = ${threadId} ${filter?.subagentIds === undefined ? sql`` : sql`AND subagent_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentIds)}))`} + ${filter?.subagentChildThreadIds === undefined ? sql`` : sql`AND child_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentChildThreadIds)}))`} ${filter?.subagentRunIds === undefined ? sql`` : sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentRunIds)}))`} ${ filter?.omitSettledSubagents !== undefined @@ -3721,13 +3736,18 @@ export const layer: Layer.Layer = `, (fields !== undefined && !fields.includes("providerTurns")) || (window === undefined && - matchesNothing(filter?.providerTurnAttemptIds, filter?.providerTurnThreadIds)) + matchesNothing( + filter?.providerTurnIds, + filter?.providerTurnAttemptIds, + filter?.providerTurnThreadIds, + )) ? Effect.succeed([]) : window === undefined ? sql` SELECT payload_json FROM orchestration_v2_projection_provider_turns - WHERE thread_id = ${threadId} + WHERE ${threadColumn(filter?.providerTurnIds)} = ${threadId} + ${filter?.providerTurnIds === undefined ? sql`` : sql`AND provider_turn_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnIds)}))`} ${filter?.providerTurnAttemptIds === undefined ? sql`` : sql`AND run_attempt_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnAttemptIds)}))`} ${filter?.providerTurnThreadIds === undefined ? sql`` : sql`AND provider_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnThreadIds)}))`} ${ @@ -5255,6 +5275,24 @@ export const layer: Layer.Layer = effect.pipe(Effect.mapError(controlReadError(threadId))), ); + const hasRunInterruptRequest: ProjectionStoreV2Shape["hasRunInterruptRequest"] = ( + threadId, + runId, + ) => + sql + .withTransaction( + Effect.gen(function* () { + yield* requireThread(threadId); + // The literal type lets SQLite use the partial stop-request index. + const rows = yield* sql<{ readonly found: number }>` + SELECT 1 AS found FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${threadId} AND run_id = ${runId} AND type = 'run_interrupt_request' + LIMIT 1`; + return rows.length > 0; + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); + const isNodeStored: ProjectionStoreV2Shape["isNodeStored"] = Effect.fn( "ProjectionStore.isNodeStored", )( @@ -7781,6 +7819,7 @@ export const layer: Layer.Layer = getHeldQueuedRuns: readOnReader(getHeldQueuedRuns), getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), + hasRunInterruptRequest: readOnReader(hasRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), getSubagentCount: readOnReader(getSubagentCount), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 25daa45a847fa..586d58e80f679 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -233,6 +233,7 @@ it.effect( getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"), getPlan: () => Effect.die("unused"), hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"), + hasRunInterruptRequest: () => Effect.die("unused stop read"), getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"), searchThread: () => Effect.die("unused"), searchThreadStream: () => Stream.empty, From 7f72f5f06aaff3b5fb614b36dbb3a02331aa23b6 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 11 Nov 2026 13:14:00 -0800 Subject: [PATCH 2/4] fix(server): a delegated task start reads the parent run and its provider records in one transaction Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/Orchestrator.ts | 48 ++++--------------- .../src/orchestration-v2/ProjectionStore.ts | 45 +++++++++++++++++ .../ProviderTurnControlService.test.ts | 1 + 3 files changed, 55 insertions(+), 39 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e02d52eaeadf2..90f22a5c9b458 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1689,31 +1689,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]), ]; - /** One run's provider thread, active attempt and provider turn, as `contextSourcePointForRun` reads them. */ - const readRunProviderRecords = - /* The delegated task operation that calls this owns the span. */ - Effect.fnUntraced(function* (threadId: ThreadId, run: OrchestrationV2Run) { - const attemptIds = run.activeAttemptId === null ? [] : [run.activeAttemptId]; - const records = yield* projectionStore.getThreadRecords( - threadId, - ["providerThreads", "providerTurns", "attempts"], - { - providerThreadIds: run.providerThreadId === null ? [] : [run.providerThreadId], - providerTurnAttemptIds: attemptIds, - attemptIds, - }, - ); - const attemptTurnId = records.attempts[0]?.providerTurnId; - if (records.providerTurns.length > 0 || attemptTurnId == null) return records; - // A steered run's attempt can point at a provider turn that another attempt started. - const { providerTurns } = yield* projectionStore.getThreadRecords( - threadId, - ["providerTurns"], - { providerTurnIds: [attemptTurnId] }, - ); - return { ...records, providerTurns }; - }); - /** The parent records `planDelegatedCompletionDelivery` reads for one task: its run, and the queued wake's message, run and tasks. */ const readDelegatedCompletionPlanRecords = /* The delegated task operation that calls this owns the span. */ @@ -8472,15 +8447,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, effects: Ref.Ref>, ) { - const parentProjectionError = (cause: unknown) => - new OrchestratorProjectionError({ threadId: command.parentThreadId, cause }); - const parentControls = yield* projectionStore - .getThreadRecords(command.parentThreadId, ["runs", "nodes"], { - runIds: [command.parentRunId], - nodeIds: [command.parentNodeId], - }) - .pipe(Effect.mapError(parentProjectionError)); - const parentRun = parentControls.runs.find( + const parentProjection = yield* projectionStore + .getRunProviderRecords(command.parentThreadId, command.parentRunId, command.parentNodeId) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.parentThreadId, cause }), + ), + ); + const parentRun = parentProjection.runs.find( (candidate) => candidate.id === command.parentRunId, ); if (parentRun === undefined || !isActiveRun(parentRun)) { @@ -8497,7 +8471,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Parent run ${command.parentRunId} is stopping.`, }); } - const parentNode = parentControls.nodes.find( + const parentNode = parentProjection.nodes.find( (candidate) => candidate.id === command.parentNodeId, ); if ( @@ -8511,10 +8485,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Parent node ${command.parentNodeId} is not part of active run ${parentRun.id}.`, }); } - const parentProjection = yield* readRunProviderRecords( - command.parentThreadId, - parentRun, - ).pipe(Effect.mapError(parentProjectionError)); const targetAdapter = yield* providerAdapters.get(command.modelSelection.instanceId).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 67aa9c2c357e3..cb6a6fd3de9de 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -645,6 +645,15 @@ export interface ProjectionStoreV2Shape { threadId: ThreadId, runId: RunId, ) => Effect.Effect; + /** One run and node with the run's provider thread, active attempt and provider turn, read together. */ + readonly getRunProviderRecords: ( + threadId: ThreadId, + runId: RunId, + nodeId: NodeId, + ) => Effect.Effect< + ProjectionRecords<"runs" | "nodes" | "providerThreads" | "providerTurns" | "attempts">, + ProjectionStoreV2Error + >; readonly isNodeStored: ( node: OrchestrationV2ExecutionNode, ) => Effect.Effect; @@ -5913,6 +5922,41 @@ export const layer: Layer.Layer = Effect.mapError(controlReadError(threadId)), ); + const getRunProviderRecords: ProjectionStoreV2Shape["getRunProviderRecords"] = ( + threadId, + runId, + nodeId, + ) => + sql + .withTransaction( + Effect.gen(function* () { + const controls = yield* getThreadRecords(threadId, ["runs", "nodes"], { + runIds: [runId], + nodeIds: [nodeId], + }); + const run = controls.runs[0]; + const attemptIds = run?.activeAttemptId == null ? [] : [run.activeAttemptId]; + const records = yield* getThreadRecords( + threadId, + ["providerThreads", "providerTurns", "attempts"], + { + providerThreadIds: run?.providerThreadId == null ? [] : [run.providerThreadId], + providerTurnAttemptIds: attemptIds, + attemptIds, + }, + ); + const attemptTurnId = records.attempts[0]?.providerTurnId; + if (records.providerTurns.length > 0 || attemptTurnId == null) + return { ...controls, ...records }; + // A steered run's attempt can point at a provider turn that another attempt started. + const { providerTurns } = yield* getThreadRecords(threadId, ["providerTurns"], { + providerTurnIds: [attemptTurnId], + }); + return { ...controls, ...records, providerTurns }; + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); + const getThreadSnapshot: ProjectionStoreV2Shape["getThreadSnapshot"] = (threadId) => sql .withTransaction( @@ -7820,6 +7864,7 @@ export const layer: Layer.Layer = getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), hasRunInterruptRequest: readOnReader(hasRunInterruptRequest), + getRunProviderRecords: readOnReader(getRunProviderRecords), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), getSubagentCount: readOnReader(getSubagentCount), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 586d58e80f679..8e1d78734d527 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -234,6 +234,7 @@ it.effect( getPlan: () => Effect.die("unused"), hasUnpairedRunInterruptRequest: () => Effect.die("unused interrupt read"), hasRunInterruptRequest: () => Effect.die("unused stop read"), + getRunProviderRecords: () => Effect.die("unused run provider read"), getThreadAttachmentIds: () => Effect.die("Unused attachment lookup"), searchThread: () => Effect.die("unused"), searchThreadStream: () => Stream.empty, From 6dc09c0c28ae197417f6e6ac7c4f0127d7fc4c55 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 19:06:18 +0000 Subject: [PATCH 3/4] refactor(server): delegated task reads trace as named ProjectionStore operations Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ProjectionStore.ts | 91 +++++++++---------- 1 file changed, 43 insertions(+), 48 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7624aa0bfdc56..8cfb9572b4e6e 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -5686,23 +5686,21 @@ export const layer: Layer.Layer = effect.pipe(Effect.mapError(controlReadError(threadId))), ); - const hasRunInterruptRequest: ProjectionStoreV2Shape["hasRunInterruptRequest"] = ( - threadId, - runId, - ) => - sql - .withTransaction( - Effect.gen(function* () { - yield* requireThread(threadId); - // The literal type lets SQLite use the partial stop-request index. - const rows = yield* sql<{ readonly found: number }>` + const hasRunInterruptRequest: ProjectionStoreV2Shape["hasRunInterruptRequest"] = Effect.fn( + "ProjectionStore.hasRunInterruptRequest", + )( + function* (threadId, runId) { + yield* requireThread(threadId); + // The literal type lets SQLite use the partial stop-request index. + const rows = yield* sql<{ readonly found: number }>` SELECT 1 AS found FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND run_id = ${runId} AND type = 'run_interrupt_request' LIMIT 1`; - return rows.length > 0; - }), - ) - .pipe(Effect.mapError(controlReadError(threadId))); + return rows.length > 0; + }, + (effect, threadId, _runId) => + sql.withTransaction(effect).pipe(Effect.mapError(controlReadError(threadId))), + ); const isNodeStored: ProjectionStoreV2Shape["isNodeStored"] = Effect.fn( "ProjectionStore.isNodeStored", @@ -6408,40 +6406,37 @@ export const layer: Layer.Layer = Effect.mapError(controlReadError(threadId)), ); - const getRunProviderRecords: ProjectionStoreV2Shape["getRunProviderRecords"] = ( - threadId, - runId, - nodeId, - ) => - sql - .withTransaction( - Effect.gen(function* () { - const controls = yield* getThreadRecords(threadId, ["runs", "nodes"], { - runIds: [runId], - nodeIds: [nodeId], - }); - const run = controls.runs[0]; - const attemptIds = run?.activeAttemptId == null ? [] : [run.activeAttemptId]; - const records = yield* getThreadRecords( - threadId, - ["providerThreads", "providerTurns", "attempts"], - { - providerThreadIds: run?.providerThreadId == null ? [] : [run.providerThreadId], - providerTurnAttemptIds: attemptIds, - attemptIds, - }, - ); - const attemptTurnId = records.attempts[0]?.providerTurnId; - if (records.providerTurns.length > 0 || attemptTurnId == null) - return { ...controls, ...records }; - // A steered run's attempt can point at a provider turn that another attempt started. - const { providerTurns } = yield* getThreadRecords(threadId, ["providerTurns"], { - providerTurnIds: [attemptTurnId], - }); - return { ...controls, ...records, providerTurns }; - }), - ) - .pipe(Effect.mapError(controlReadError(threadId))); + const getRunProviderRecords: ProjectionStoreV2Shape["getRunProviderRecords"] = Effect.fn( + "ProjectionStore.getRunProviderRecords", + )( + function* (threadId, runId, nodeId) { + const controls = yield* getThreadRecords(threadId, ["runs", "nodes"], { + runIds: [runId], + nodeIds: [nodeId], + }); + const run = controls.runs[0]; + const attemptIds = run?.activeAttemptId == null ? [] : [run.activeAttemptId]; + const records = yield* getThreadRecords( + threadId, + ["providerThreads", "providerTurns", "attempts"], + { + providerThreadIds: run?.providerThreadId == null ? [] : [run.providerThreadId], + providerTurnAttemptIds: attemptIds, + attemptIds, + }, + ); + const attemptTurnId = records.attempts[0]?.providerTurnId; + if (records.providerTurns.length > 0 || attemptTurnId == null) + return { ...controls, ...records }; + // A steered run's attempt can point at a provider turn that another attempt started. + const { providerTurns } = yield* getThreadRecords(threadId, ["providerTurns"], { + providerTurnIds: [attemptTurnId], + }); + return { ...controls, ...records, providerTurns }; + }, + (effect, threadId, _runId, _nodeId) => + sql.withTransaction(effect).pipe(Effect.mapError(controlReadError(threadId))), + ); const getThreadSnapshot: ProjectionStoreV2Shape["getThreadSnapshot"] = (threadId) => sql From 781863b0ea735db890f19162c93f27db1b74b77c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 23:09:19 +0000 Subject: [PATCH 4/4] perf(server): a delegated task start finds the parent's provider turn by its attempt index The start's provider-turn lookup by attempt had no index and walked every provider turn of the parent thread, so its cost still grew with history. Co-Authored-By: Claude Opus 5.5 --- .../Orchestrator.control-reads.test.ts | 4 ++-- apps/server/src/orchestration-v2/ProjectionStore.ts | 2 +- apps/server/src/persistence/Migrations.ts | 2 ++ .../Migrations/093_ProviderTurnAttemptIndex.ts | 11 +++++++++++ .../persistence/reconcileV2PreviewMigration.test.ts | 2 ++ 5 files changed, 18 insertions(+), 3 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/093_ProviderTurnAttemptIndex.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index de5c4aee628ba..d2d930972de6d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -2074,8 +2074,7 @@ it.effect("starts and finishes a delegated task without reading the parent's his // Start and finish read the task's own records, so their VM steps, which are deterministic, // must stay flat on a copy with a long parent history. Left out: the counts that number the - // task and its wake, which walk one index, and provider turn history, whose read has no - // index on the attempt and walks the thread's turn rows without decoding them. + // task and its wake, which walk one index. const copyPath = "file:/delegated-task-reads?vfs=memdb"; using copy = new NodeSqlite.DatabaseSync(copyPath); yield* sql`VACUUM INTO ${copyPath}`; @@ -2118,6 +2117,7 @@ it.effect("starts and finishes a delegated task without reading the parent's his addHistory("runs", "run_id"); addHistory("run_attempts", "attempt_id", ["run_id"]); addHistory("messages", "message_id"); + addHistory("provider_turns", "provider_turn_id", ["run_attempt_id"]); assert.isAtMost(readSteps(), shortSteps * 1.1); }).pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 414f865008dfe..16b49070c1dab 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4105,7 +4105,7 @@ export const layer: Layer.Layer = ? sql` SELECT payload_json FROM orchestration_v2_projection_provider_turns - WHERE ${threadColumn(filter?.providerTurnIds)} = ${threadId} + WHERE ${threadColumn(filter?.providerTurnIds ?? filter?.providerTurnAttemptIds)} = ${threadId} ${filter?.providerTurnIds === undefined ? sql`` : sql`AND provider_turn_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnIds)}))`} ${ recentRunIds === null diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 1279fe9e60364..586c59729c5ce 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -100,6 +100,7 @@ import Migration0089 from "./Migrations/089_ShellThreadReadIndexes.ts"; import Migration0090 from "./Migrations/090_AuthConnectDevices.ts"; import Migration0091 from "./Migrations/091_AuthConnectDeniedRequests.ts"; import Migration0092 from "./Migrations/092_QueuedTurnInputIndexJsonValid.ts"; +import Migration0093 from "./Migrations/093_ProviderTurnAttemptIndex.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -195,6 +196,7 @@ const migrationEntries = [ [90, "AuthConnectDevices", Migration0090], [91, "AuthConnectDeniedRequests", Migration0091], [92, "QueuedTurnInputIndexJsonValid", Migration0092], + [93, "ProviderTurnAttemptIndex", Migration0093], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/093_ProviderTurnAttemptIndex.ts b/apps/server/src/persistence/Migrations/093_ProviderTurnAttemptIndex.ts new file mode 100644 index 0000000000000..8e687ba31597f --- /dev/null +++ b/apps/server/src/persistence/Migrations/093_ProviderTurnAttemptIndex.ts @@ -0,0 +1,11 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE INDEX orchestration_v2_projection_provider_turns_attempt_idx + ON orchestration_v2_projection_provider_turns(run_attempt_id) + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index c1ac178cf84b7..ad20d3ef6f35c 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -72,6 +72,7 @@ describe("V2 preview upgrade", () => { [90, "AuthConnectDevices"], [91, "AuthConnectDeniedRequests"], [92, "QueuedTurnInputIndexJsonValid"], + [93, "ProviderTurnAttemptIndex"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -187,6 +188,7 @@ describe("V2 preview upgrade", () => { [90, "AuthConnectDevices"], [91, "AuthConnectDeniedRequests"], [92, "QueuedTurnInputIndexJsonValid"], + [93, "ProviderTurnAttemptIndex"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );