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 28cf03313b2d6..d2d930972de6d 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, @@ -24,8 +26,10 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; 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"; @@ -44,6 +48,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 = { @@ -1901,6 +1912,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. + 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"); + addHistory("provider_turns", "provider_turn_id", ["run_attempt_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 226b7d703d0b1..57337a06f6c74 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1978,6 +1978,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ]), ]; + /** 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) { @@ -9309,19 +9337,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio 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 }, - ) + .getRunProviderRecords(command.parentThreadId, command.parentRunId, command.parentNodeId) .pipe( Effect.mapError( - (cause) => - new OrchestratorProjectionError({ - threadId: command.parentThreadId, - cause, - }), + (cause) => new OrchestratorProjectionError({ threadId: command.parentThreadId, cause }), ), ); const parentRun = parentProjection.runs.find( @@ -12501,16 +12520,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< @@ -14028,7 +14038,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; @@ -14298,7 +14308,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 = { @@ -14376,21 +14386,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" && @@ -14402,7 +14409,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 && @@ -14416,12 +14427,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 = { @@ -14571,8 +14582,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) { @@ -14584,9 +14598,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; @@ -16007,7 +16023,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 50f2a0a12d1ae..2ea2e730e61d9 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -723,26 +723,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`); } // The filters every command reads with seek the rows they keep, not every row of the thread. const command = traceSqlStatements(); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2d31d3e9bde54..0b0528d1ad269 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -348,7 +348,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; @@ -368,6 +370,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; @@ -697,6 +700,20 @@ 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; + /** 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; @@ -3533,7 +3550,7 @@ export const layer: Layer.Layer = }); const turnItemMatch = (threadId: ThreadId, filter: ProjectionRecordFilter | undefined) => sql` - thread_id = ${threadId} + ${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)}))`} @@ -3886,13 +3903,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)}))`} ${inRecentRuns("run_id")} ORDER BY run_id ASC, attempt_ordinal ASC ` @@ -3969,14 +3988,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 === true @@ -4073,13 +4097,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 ?? filter?.providerTurnAttemptIds)} = ${threadId} + ${filter?.providerTurnIds === undefined ? sql`` : sql`AND provider_turn_id IN (SELECT value FROM json_each(${encodeIdList(filter.providerTurnIds)}))`} ${ recentRunIds === null ? sql`` @@ -5659,6 +5688,22 @@ export const layer: Layer.Layer = effect.pipe(Effect.mapError(controlReadError(threadId))), ); + 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; + }, + (effect, threadId, _runId) => + sql.withTransaction(effect).pipe(Effect.mapError(controlReadError(threadId))), + ); + const isNodeStored: ProjectionStoreV2Shape["isNodeStored"] = Effect.fn( "ProjectionStore.isNodeStored", )( @@ -6363,6 +6408,38 @@ export const layer: Layer.Layer = 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 .withTransaction( @@ -9129,6 +9206,8 @@ export const layer: Layer.Layer = getHeldQueuedRuns: readOnReader(getHeldQueuedRuns), getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), + hasRunInterruptRequest: readOnReader(hasRunInterruptRequest), + getRunProviderRecords: readOnReader(getRunProviderRecords), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), getTranscriptByteFloor: readOnReader(getTranscriptByteFloor), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index c3d1096cb3c93..532d019ddd6fb 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -235,6 +235,8 @@ it.effect( getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"), 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, 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:" }))), );