diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 426da22121e3..bff7efe7c5d5 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -29,7 +29,9 @@ import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; +import * as Statement from "effect/sql/Statement"; import * as Tracer from "effect/Tracer"; import * as TestClock from "effect/testing/TestClock"; @@ -3706,6 +3708,108 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }, ); +// The run-update listener mirrors a follow-up's status onto its task outside any test fiber, so +// its statements are recorded from the layer down, for one parent thread at a time. +let recordReadsOf: ThreadId | undefined; +const recordedReads: Array = []; +const layerRecordingReads = Layer.succeed(Statement.CurrentTransformer, (statement) => { + const [query, params] = statement.compile(); + if (recordReadsOf !== undefined && params.includes(recordReadsOf)) recordedReads.push(query); + return Effect.succeed(statement); +}); + +it.layer( + Layer.merge(layerTest, ProviderContinuationRequests.layer).pipe( + Layer.provide(layerRecordingReads), + ), +)("subagent follow-up status", (it) => { + it.effect("mirrors a follow-up's status reading only its task's node and card", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make("thread:narrow-status-parent"); + const projectId = ProjectId.make("project:narrow-status"); + const parentRunId = RunId.make("run:narrow-status-parent"); + const rootNodeId = NodeId.make("node:narrow-status-parent"); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make("node:narrow-status-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: "narrow-status-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + for (const name of ["active", "queued"]) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make(`command:narrow-status:${name}`), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId: MessageId.make(`message:narrow-status:${name}`), + text: name, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + } + const queued = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (run) => run.ordinal === 3, + )!; + assert.equal(queued.status, "queued"); + const afterSequence = yield* eventSink.latestSequence(); + recordReadsOf = parentThreadId; + yield* orchestrator.dispatch({ + type: "thread.stop", + threadId: child.childThreadId, + commandId: CommandId.make("command:narrow-status:stop"), + }); + yield* eventSink.stream({ afterSequence, eventType: "subagent.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === queued.delegatedTaskId && + stored.event.payload.status === "idle", + ), + Stream.take(1), + Stream.runHead, + ); + recordReadsOf = undefined; + const paused = yield* orchestrator.getThreadProjection(parentThreadId); + assert.equal(paused.nodes.find((row) => row.id === queued.delegatedTaskId)?.status, "idle"); + assert.equal( + paused.turnItems.find( + (row) => row.type === "subagent" && row.subagentId === queued.delegatedTaskId, + )?.status, + "idle", + ); + const nodeReads = recordedReads.filter((query) => + /SELECT payload_json\s+FROM orchestration_v2_projection_(nodes|turn_items)\b/.test(query), + ); + assert.isNotEmpty(nodeReads); + assert.deepEqual( + nodeReads.filter((query) => !/AND (node\.)?node_id IN/.test(query)), + [], + "a parent node or card read is not narrowed", + ); + }), + ); +}); + it.layer(layerTest)("delegated tasks across a server restart", (it) => { it.effect.each([ { name: "releases", key: "", pendingBackgroundTasks: [], released: true }, @@ -3842,6 +3946,234 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + // Finalizing a child reads, from its parent, only that child's task, node, card and the transfers + // it sent, and the message and other tasks of a delivery it joins by id. + it.effect("finalizing a delegated child reads only that child's records from its parent", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:narrow-finalize-parent"); + const projectId = ProjectId.make("project:narrow-finalize"); + const runId = RunId.make("run:narrow-finalize-parent"); + const rootNodeId = NodeId.make("node:narrow-finalize-parent"); + const siblingTaskId = NodeId.make("node:narrow-finalize-seed"); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: siblingTaskId, + deliveryState: "claimed", + completionWake: "always", + deliveryTaskIds: [siblingTaskId], + now, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:narrow-finalize-delivery"), + threadId, + messageId, + text: `Delegated task ${siblingTaskId} reached a terminal state.`, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [siblingTaskId] }, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "narrow-finalize-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + const cards = [ + { taskId: siblingTaskId, status: "completed" as const, ordinal: 100 }, + { taskId: child.taskId, status: "running" as const, ordinal: 101 }, + ]; + yield* sink.write({ + commandId: CommandId.make("command:narrow-finalize-cards"), + events: cards.flatMap(({ taskId, status, ordinal }) => [ + { + id: EventId.make(`event:narrow-finalize-node:${taskId}`), + type: "node.updated" as const, + threadId, + runId, + nodeId: taskId, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId, + parentNodeId: rootNodeId, + rootNodeId, + kind: "subagent" as const, + status, + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: status === "completed" ? now : null, + }, + }, + { + id: EventId.make(`event:narrow-finalize-card:${taskId}`), + type: "turn-item.updated" as const, + threadId, + runId, + nodeId: taskId, + occurredAt: now, + payload: { + id: TurnItemId.make(`turn-item:narrow-finalize:${taskId}`), + type: "subagent" as const, + threadId, + runId, + nodeId: taskId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status, + title: null, + startedAt: now, + completedAt: status === "completed" ? now : null, + updatedAt: now, + subagentId: taskId, + origin: "app_owned" as const, + driver, + providerInstanceId: modelSelection.instanceId, + childThreadId: taskId === child.taskId ? child.childThreadId : null, + prompt: `Run ${taskId}.`, + result: status === "completed" ? "child finished" : null, + }, + }, + ]), + }); + const parentReads: Array = []; + const transferReadParams: Array> = []; + // The parent runs read while the child's result is recorded, with their bound ids. + const resultRunReads: Array]> = []; + const recordsResult = (span: Tracer.AnySpan | undefined): boolean => + span?._tag === "Span" && + (span.name === "Orchestrator.finalizeAppOwnedSubagentTask" || + recordsResult(Option.getOrUndefined(span.parent))); + const observe: Statement.Transformer = (statement, _sql, _fiber, span) => { + const [query, params] = statement.compile(); + if ( + params.includes(threadId) && + query.includes("FROM orchestration_v2_projection_runs") && + recordsResult(span) + ) { + resultRunReads.push([query, params]); + } + if ( + params.includes(threadId) && + (/FROM orchestration_v2_projection_(subagents|context_transfers)\b/.test(query) || + /SELECT payload_json\s+FROM orchestration_v2_projection_(nodes|turn_items|messages)\b/.test( + query, + )) + ) { + parentReads.push(query); + if (query.includes("FROM orchestration_v2_projection_context_transfers")) { + transferReadParams.push(params); + } + } + return Effect.succeed(statement); + }; + yield* orchestrator.recoverDelegatedTasks.pipe( + Effect.provideService(Statement.CurrentTransformer, observe), + ); + const projection = yield* orchestrator.getThreadProjection(threadId); + const task = projection.subagents.find((candidate) => candidate.id === child.taskId); + assert.equal(task?.status, "completed"); + // Every finalize read names its rows: tasks by child thread or id, transfers by source. + const subagentReads = parentReads.filter((query) => + query.includes("FROM orchestration_v2_projection_subagents"), + ); + const transferReads = parentReads.filter((query) => + query.includes("FROM orchestration_v2_projection_context_transfers"), + ); + assert.isNotEmpty(subagentReads); + assert.isNotEmpty(transferReads); + assert.deepEqual( + subagentReads.filter((query) => !/AND (child_thread_id|subagent_id) (=|IN)/.test(query)), + [], + "a parent subagent read is not narrowed", + ); + assert.deepEqual( + transferReads.filter((query) => !/AND source_thread_id (=|IN)/.test(query)), + [], + "a parent context-transfer read is not narrowed", + ); + // Only the task's run and the run its delivery message started, each read by id. + const deliveryRunId = projection.runs.find((run) => run.userMessageId === messageId)?.id; + assert.isDefined(deliveryRunId); + assert.isNotEmpty(resultRunReads); + assert.deepEqual( + resultRunReads.filter(([query]) => !/AND run_id IN/.test(query)), + [], + "a parent runs read is not narrowed", + ); + assert.deepEqual( + [ + ...new Set( + resultRunReads.flatMap(([, params]) => + params + .filter((param) => param !== threadId) + .flatMap((param) => JSON.parse(String(param)) as Array), + ), + ), + ].toSorted(), + [runId, deliveryRunId].toSorted(), + ); + // An existing result transfer is found only when the read asks for this child's. + assert.deepEqual( + transferReadParams.filter((params) => !params.includes(child.childThreadId)), + [], + "a parent context-transfer read names another thread", + ); + // Nodes and cards by the task's node, messages by id. + for (const [table, narrowed] of [ + ["nodes", /AND node\.node_id IN/], + ["turn_items", /AND node_id IN/], + ["messages", /AND message_id IN/], + ] as const) { + const reads = parentReads.filter((query) => + query.includes(`FROM orchestration_v2_projection_${table}`), + ); + assert.isNotEmpty(reads, table); + assert.deepEqual( + reads.filter((query) => !narrowed.test(query)), + [], + `a parent ${table} read is not narrowed`, + ); + } + // The child's node and card still end with its task. + assert.equal(projection.nodes.find((row) => row.id === child.taskId)?.status, "completed"); + assert.equal( + projection.turnItems.find( + (row) => row.type === "subagent" && row.subagentId === child.taskId, + )?.status, + "completed", + ); + // The delivery the child joins still names the sibling it was opened for. + const message = projection.messages.find((candidate) => candidate.id === messageId); + assert.deepEqual(message?.delegatedCompletion?.taskIds, [siblingTaskId, child.taskId]); + assert.include(message?.text ?? "", `Delegated task ${siblingTaskId} [completed]`); + assert.include(message?.text ?? "", `Delegated task ${child.taskId}`); + }), + ); + 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 a3ad41ec9eb4..cda2822f0349 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13592,7 +13592,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; @@ -13862,7 +13862,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const parent = yield* projectionStore.getThreadRecords( childControls.thread.lineage.parentThreadId, ["nodes", "turnItems"], - { turnItemTypes: ["subagent"] }, + { turnItemTypes: ["subagent"], nodeIds: [task.id], turnItemNodeIds: [task.id] }, ); const now = yield* DateTime.now; const scope = { @@ -13929,19 +13929,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItems: childResult.turnItems, }; const parentThreadId = childControls.thread.lineage.parentThreadId; + // Only this child's tasks, node, card and transfers; runs and a delivery are read by id. const parentProjection = yield* projectionStore.getThreadRecords( parentThreadId, - [ - "runs", - "messages", - "subagents", - "providerThreads", - "providerTurns", - "nodes", - "turnItems", - "contextTransfers", - ], - { turnItemTypes: ["subagent"], messageRoles: ["user"] }, + ["subagents", "nodes", "turnItems", "contextTransfers"], + { + turnItemTypes: ["subagent"], + subagentChildThreadIds: [childThreadId], + contextTransferSourceThreadIds: [childThreadId], + nodeIds: [options.task.id], + turnItemNodeIds: [options.task.id], + }, ); const task = parentProjection.subagents.find( (candidate) => @@ -13969,10 +13967,11 @@ 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 parentRunsById = (runIds: ReadonlyArray) => + projectionStore + .getThreadRecords(parentThreadId, ["runs"], { runIds }) + .pipe(Effect.map((records) => records.runs)); + const parentRun = task.runId === null ? undefined : (yield* parentRunsById([task.runId]))[0]; const parentNode = parentProjection.nodes.find((candidate) => candidate.id === task.id); const parentTurnItem = parentProjection.turnItems.find( (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id, @@ -13992,8 +13991,32 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } : {}), }; + // The delivery this result may join: its message, and the other tasks its text names. + const delivery = parentRun?.delegatedCompletion?.delivery ?? null; + const deliveryRecords = + delivery === null + ? { messages: [], subagents: [] } + : yield* projectionStore.getThreadRecords(parentThreadId, ["messages", "subagents"], { + messageRoles: ["user"], + messageIds: [delivery.messageId], + subagentIds: delivery.taskIds.filter( + (taskId) => + !parentProjection.subagents.some((candidate) => candidate.id === taskId), + ), + }); + // The delivery message's run decides whether that delivery can still take this result. + const deliveryRunId = deliveryRecords.messages[0]?.runId ?? null; + const deliveryRuns = + deliveryRunId === null || deliveryRunId === parentRun?.id + ? [] + : yield* parentRunsById([deliveryRunId]); const completionPlan = yield* planDelegatedCompletionDelivery({ - parentProjection, + parentProjection: { + ...parentProjection, + runs: [...(parentRun === undefined ? [] : [parentRun]), ...deliveryRuns], + messages: deliveryRecords.messages, + subagents: [...parentProjection.subagents, ...deliveryRecords.subagents], + }, parentRun, task, updatedTask, @@ -14124,10 +14147,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) return; const progress = delegatedTaskProgress(childControls); - const parent = yield* projectionStore.getThreadRecords(parentThreadId, [ - "subagents", - "contextTransfers", - ]); + // Only this child's tasks and result transfers matter; a parent can hold thousands. + const parent = yield* projectionStore.getThreadRecords( + parentThreadId, + ["subagents", "contextTransfers"], + { subagentChildThreadIds: [childThreadId], contextTransferSourceThreadIds: [childThreadId] }, + ); for (const task of parent.subagents) { // A cancelled task does not wait for the work its cancel is stopping. const cancelled = task.cancelledRunId !== undefined; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 3b34434017e9..66ccddea713b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2588,6 +2588,296 @@ const subagentRecordsSelectById = Effect.gen(function* () { assert.deepEqual(yield* read([]), []); }); +const recordsAt = (second: number) => DateTime.makeUnsafe(Date.UTC(2026, 0, 1, 0, 0, second)); +const recordsThreadCreated = (threadId: ThreadId) => + ({ + id: EventId.make(`event:${threadId}:created`), + type: "thread.created", + threadId, + providerInstanceId, + occurredAt: recordsAt(0), + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:records"), + title: threadId, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: recordsAt(0), + updatedAt: recordsAt(0), + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }) as never; + +const queryPlan = Effect.fn("queryPlan")(function* (statement: string | undefined) { + assert.isDefined(statement); + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql.unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${statement}`); + return rows.map((row) => row.detail); +}); + +// finalizeAppOwnedSubagent's parent read: one child's tasks and the transfers it sent, by index, +// without depending on an index existing or on a sibling's rows decoding. +const parentRecordsNarrowToOneChild = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const sql = yield* SqlClient.SqlClient; + const parent = ThreadId.make("thread:records-parent"); + const child = ThreadId.make("thread:records-child"); + const sibling = ThreadId.make("thread:records-sibling"); + const elsewhere = ThreadId.make("thread:records-elsewhere"); + const task = (name: string, childThreadId: ThreadId | null, startedAt: number) => ({ + id: EventId.make(`event:task:${name}`), + type: "subagent.updated", + threadId: parent, + nodeId: NodeId.make(`node:${name}`), + driver, + providerInstanceId, + occurredAt: recordsAt(startedAt), + payload: { + id: NodeId.make(`node:${name}`), + threadId: parent, + runId: null, + parentNodeId: NodeId.make("node:records-parent:root"), + origin: childThreadId === null ? "provider_native" : "app_owned", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: name, + title: name, + model: null, + status: "completed", + result: null, + startedAt: recordsAt(startedAt), + completedAt: recordsAt(9), + updatedAt: recordsAt(9), + }, + }); + const transfer = ( + name: string, + type: "subagent_spawn" | "subagent_result" | "fork", + sourceThreadId: ThreadId, + targetThreadId: ThreadId, + ) => ({ + id: EventId.make(`event:transfer:${name}`), + type: "context-transfer.created", + threadId: targetThreadId, + providerInstanceId, + occurredAt: recordsAt(2), + payload: { + id: `transfer:${name}`, + type, + sourceThreadId, + targetThreadId, + sourcePoint: { threadId: sourceThreadId }, + basePoint: null, + sourceProviderInstanceId: providerInstanceId, + targetProviderInstanceId: providerInstanceId, + targetRunId: null, + status: "consumed", + resolution: null, + createdBy: "agent", + error: null, + createdAt: recordsAt(2), + updatedAt: recordsAt(2), + consumedAt: recordsAt(2), + }, + }); + yield* store.apply(recordsThreadCreated(parent)); + for (const event of [ + task("child-late", child, 5), + task("sibling", sibling, 3), + task("child-early", child, 1), + task("native", null, 2), + transfer("child-spawn", "subagent_spawn", parent, child), + transfer("child-result", "subagent_result", child, parent), + transfer("child-follow-up", "fork", child, parent), + transfer("sibling-result", "subagent_result", sibling, parent), + transfer("child-elsewhere", "fork", child, elsewhere), + transfer("fork-in", "fork", elsewhere, parent), + ]) { + yield* store.apply(event as never); + } + const fields = ["subagents", "contextTransfers"] as const; + const narrowedTo = (threadId: ThreadId) => + store.getThreadRecords(parent, fields, { + subagentChildThreadIds: [threadId], + contextTransferSourceThreadIds: [threadId], + }); + + const full = yield* store.getThreadRecords(parent, fields); + const { statements, tracer } = traceSqlStatements(); + const narrowed = yield* narrowedTo(child).pipe(Effect.withTracer(tracer)); + assert.deepEqual( + narrowed.subagents, + full.subagents.filter((candidate) => candidate.childThreadId === child), + ); + assert.deepEqual( + narrowed.subagents.map((candidate) => String(candidate.id)), + ["node:child-early", "node:child-late"], + ); + assert.deepEqual( + narrowed.contextTransfers, + full.contextTransfers.filter((candidate) => candidate.sourceThreadId === child), + ); + assert.deepEqual( + narrowed.contextTransfers.map((candidate) => String(candidate.id)), + ["transfer:child-result", "transfer:child-follow-up"], + ); + // Each read seeks the child's rows through its own index, never the parent's whole history. + for (const [table, index] of [ + ["subagents", "orchestration_v2_projection_subagents_child_thread_idx (child_thread_id=?)"], + [ + "context_transfers", + "orchestration_v2_projection_context_transfers_source_thread_status_idx (source_thread_id=?)", + ], + ]) { + const plan = yield* queryPlan( + statements.find((statement) => + statement.includes(`FROM orchestration_v2_projection_${table}`), + ), + ); + assert.include(plan[0] ?? "", `USING INDEX ${index}`); + assert.notOk(plan.some((detail) => detail.startsWith("SCAN orchestration_v2_projection"))); + } + // The finishing task's card is sought by its node, not picked from every card of the parent. + for (const [ordinal, name] of ["child-early", "sibling", "child-late"].entries()) { + const nodeId = NodeId.make(`node:${name}`); + yield* store.apply({ + id: EventId.make(`event:card:${name}`), + type: "turn-item.updated", + threadId: parent, + nodeId, + driver, + occurredAt: recordsAt(ordinal), + payload: { + id: TurnItemId.make(`item:card:${name}`), + type: "subagent", + threadId: parent, + runId: null, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: name, + startedAt: recordsAt(ordinal), + completedAt: recordsAt(9), + updatedAt: recordsAt(9), + subagentId: nodeId, + origin: "app_owned", + driver, + providerInstanceId, + childThreadId: name === "sibling" ? sibling : child, + prompt: name, + result: null, + }, + } as never); + } + const cardReads = traceSqlStatements(); + const cards = yield* store + .getThreadRecords(parent, ["turnItems"], { + turnItemTypes: ["subagent"], + turnItemNodeIds: [NodeId.make("node:child-late")], + }) + .pipe(Effect.withTracer(cardReads.tracer)); + assert.deepEqual( + cards.turnItems.map((item) => String(item.id)), + ["item:card:child-late"], + ); + const cardPlan = yield* queryPlan( + cardReads.statements.find((statement) => + statement.includes("FROM orchestration_v2_projection_turn_items"), + ), + ); + assert.isTrue( + cardPlan.some((detail) => + detail.includes( + "USING INDEX orchestration_v2_projection_turn_items_node_ordinal_idx (node_id=?)", + ), + ), + cardPlan.join(" | "), + ); + // Several children read through the id list. + const both = yield* store.getThreadRecords(parent, fields, { + subagentChildThreadIds: [child, sibling], + contextTransferSourceThreadIds: [child, sibling], + }); + assert.deepEqual( + both.subagents, + full.subagents.filter( + (candidate) => candidate.childThreadId === child || candidate.childThreadId === sibling, + ), + ); + assert.deepEqual( + both.contextTransfers, + full.contextTransfers.filter( + (candidate) => candidate.sourceThreadId === child || candidate.sourceThreadId === sibling, + ), + ); + const none = yield* store.getThreadRecords(parent, fields, { + subagentChildThreadIds: [], + contextTransferSourceThreadIds: [], + }); + assert.deepEqual([none.subagents, none.contextTransfers], [[], []]); + + // A sibling's undecodable rows fail only reads that return them; the child's still fail it. + const corrupt = (table: string, column: string, id: string) => + sql.unsafe( + `UPDATE orchestration_v2_projection_${table} SET payload_json = '{' WHERE ${column} = ?`, + [id], + ); + yield* corrupt("subagents", "subagent_id", "node:sibling"); + yield* corrupt("context_transfers", "context_transfer_id", "transfer:sibling-result"); + assert.deepEqual(yield* narrowedTo(child), narrowed); + assert.strictEqual( + (yield* Effect.flip(store.getThreadRecords(parent, fields)))._tag, + "ProjectionStoreReadError", + ); + for (const [table, column, id] of [ + ["subagents", "subagent_id", "node:child-late"], + ["context_transfers", "context_transfer_id", "transfer:child-follow-up"], + ] as const) { + const [row] = yield* sql.unsafe<{ readonly payload_json: string }>( + `SELECT payload_json FROM orchestration_v2_projection_${table} WHERE ${column} = ?`, + [id], + ); + yield* corrupt(table, column, id); + assert.strictEqual((yield* Effect.flip(narrowedTo(child)))._tag, "ProjectionStoreReadError"); + yield* sql.unsafe( + `UPDATE orchestration_v2_projection_${table} SET payload_json = ? WHERE ${column} = ?`, + [row!.payload_json, id], + ); + } + assert.deepEqual(yield* narrowedTo(child), narrowed); + + // Without the child and source indexes the read is slower, never wrong. + yield* sql`DROP INDEX orchestration_v2_projection_subagents_child_thread_idx`; + yield* sql`DROP INDEX orchestration_v2_projection_context_transfers_source_thread_status_idx`; + assert.deepEqual(yield* narrowedTo(child), narrowed); +}); + +// It drops an index and corrupts rows, so it gets its own database. +it.effect("record reads narrow a parent to one child's tasks and transfers", () => + parentRecordsNarrowToOneChild.pipe(Effect.provide(layerTest)), +); + const assertProjectScopedShells = Effect.fn("assertProjectScopedShells")(function* ( suffix: string, ) { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 841cd0a0b5b7..6285fa859f51 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -365,6 +365,8 @@ export interface ProjectionRecordFilter { readonly contextTransferStatuses?: ReadonlyArray; readonly subagentIds?: ReadonlyArray; readonly subagentRunIds?: ReadonlyArray; + readonly subagentChildThreadIds?: ReadonlyArray; + readonly contextTransferSourceThreadIds?: ReadonlyArray; readonly nodeIds?: ReadonlyArray; readonly turnItemRequestIds?: ReadonlyArray; readonly turnItemTypes?: ReadonlyArray; @@ -3482,7 +3484,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)}))`} @@ -3918,7 +3920,11 @@ 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` @@ -3927,6 +3933,14 @@ export const layer: Layer.Layer = WHERE ${threadColumn(filter?.subagentIds)} = ${threadId} ${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)}))`} + ${ + filter?.subagentChildThreadIds === undefined + ? sql`` + : // One child seeks its index by equality; an IN list hides the row count from the planner. + filter.subagentChildThreadIds.length === 1 + ? sql`AND child_thread_id = ${filter.subagentChildThreadIds[0]!}` + : sql`AND child_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.subagentChildThreadIds)}))` + } ${ filter?.omitSettledSubagents === true ? sql`AND subagent.rowid IN (${unsettledSubagentRowids(threadId)})` @@ -4195,7 +4209,8 @@ export const layer: Layer.Layer = ORDER BY rowid ASC `, (fields !== undefined && !fields.includes("contextTransfers")) || - (window === undefined && matchesNothing(filter?.contextTransferStatuses)) + (window === undefined && + matchesNothing(filter?.contextTransferSourceThreadIds, filter?.contextTransferStatuses)) ? Effect.succeed([]) : window === undefined ? sql` @@ -4216,6 +4231,13 @@ export const layer: Layer.Layer = AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)})) )` } + ${ + filter?.contextTransferSourceThreadIds === undefined + ? sql`` + : filter.contextTransferSourceThreadIds.length === 1 + ? sql`AND source_thread_id = ${filter.contextTransferSourceThreadIds[0]!}` + : sql`AND source_thread_id IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferSourceThreadIds)}))` + } ORDER BY rowid ASC ` : sql` diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 8dbf42fc45f7..2c369dfe81f7 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -131,6 +131,9 @@ queries read item counts from a per-thread, per-run table that turn-item trigger use a partial index for nonterminal background items, including idle items. Latest user-message reads use an index ordered by update time. +Finalizing a delegated task reads only that child's tasks, and the transfers it sent to the +parent, from the parent, through the subagent child-thread and transfer source-thread indexes. + Command palette results keep live VCS and linked pull-request queries leased to the visible viewport, including a small overscan region. Cached badges remain available while leases are released. Local Git status reads skip divergence calculations; callers that need ahead/behind