From 71fc57aca795bc9e3052eb09fc98b0575b62c195 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 11:58:27 +0000 Subject: [PATCH 1/6] perf(server): finalizing a subagent reads only that child's records When a delegated child thread finishes, finalizing it read the parent's whole subagent list and every context transfer to find that child's task and the transfers it sent back. The parent read now asks for only that child's rows, through two record filters that seek the existing child-thread and source-thread indexes. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 68 ++++++ .../src/orchestration-v2/Orchestrator.ts | 10 +- .../orchestration-v2/ProjectionStore.test.ts | 230 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 26 +- docs/internals/performance-regressions.md | 3 + 5 files changed, 331 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3f0f9cf0342e..74596067a739 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -26,6 +26,7 @@ import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; +import * as Statement from "effect/sql/Statement"; import * as Tracer from "effect/Tracer"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; @@ -2403,6 +2404,73 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + // Finalizing a child reads, from its parent, only that child's tasks and the transfers it sent. + it.effect("finalizing a delegated child reads only that child's records from its parent", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + 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"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:narrow-finalize-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "narrow-finalize-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + const parentReads: Array = []; + const observe: Statement.Transformer = (statement) => { + const [query, params] = statement.compile(); + if ( + params.includes(threadId) && + /FROM orchestration_v2_projection_(subagents|context_transfers)\b/.test(query) + ) { + parentReads.push(query); + } + return Effect.succeed(statement); + }; + yield* orchestrator.recoverDelegatedTasks.pipe( + Effect.provideService(Statement.CurrentTransformer, observe), + ); + const task = (yield* orchestrator.getThreadProjection(threadId)).subagents.find( + (candidate) => candidate.id === child.taskId, + ); + assert.equal(task?.status, "completed"); + // The finalize read names the child: its tasks by child thread, its transfers by source. + assert.isTrue( + parentReads.some( + (query) => + query.includes("FROM orchestration_v2_projection_subagents") && + /AND child_thread_id (=|IN)/.test(query), + ), + "the parent's subagent read is not narrowed to the child", + ); + assert.isTrue( + parentReads.some( + (query) => + query.includes("FROM orchestration_v2_projection_context_transfers") && + /AND source_thread_id (=|IN)/.test(query), + ), + "the parent's context-transfer read is not narrowed to the child", + ); + }), + ); + 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 b4bdbd705c29..603c4492a5f4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12747,10 +12747,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) { if ( task.origin !== "app_owned" || diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index c6187a97ea8d..bfc9f47c0faf 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1682,6 +1682,236 @@ 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_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"))); + } + // 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_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 b5a0f2cfb555..d49158f080db 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -352,6 +352,8 @@ export interface ProjectionRecordFilter { readonly contextTransferStatuses?: ReadonlyArray; readonly subagentIds?: ReadonlyArray; readonly subagentRunIds?: ReadonlyArray; + readonly subagentChildThreadIds?: ReadonlyArray; + readonly contextTransferSourceThreadIds?: ReadonlyArray; readonly nodeIds?: ReadonlyArray; readonly turnItemTypes?: ReadonlyArray; readonly turnItemStatuses?: ReadonlyArray; @@ -3427,7 +3429,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` @@ -3436,6 +3442,14 @@ export const layer: Layer.Layer = WHERE thread_id = ${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 !== undefined ? sql`AND (${ @@ -3670,13 +3684,21 @@ 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` SELECT payload_json FROM orchestration_v2_projection_context_transfers WHERE (source_thread_id = ${threadId} OR target_thread_id = ${threadId}) + ${ + 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)}))` + } ${filter?.contextTransferStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.contextTransferStatuses)}))`} ORDER BY rowid ASC ` diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 4458f1af9346..dcfb0752f5e4 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -105,6 +105,9 @@ queries use a covering thread/run index for item counts and a partial index for 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 From f126ef3d1f5a2b27a62cdc3e0c06f962b986a8d0 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 14:43:03 +0000 Subject: [PATCH 2/6] perf(server): the terminal finalize read is narrowed to the finishing child Finalizing a finished delegated child read the parent a second time to build its result and delivery, and that read still loaded every task, every transfer and every provider thread of the parent. It now asks for the child's tasks and the transfers it sent, reads the other tasks of a delivery it joins by id, and no longer loads provider threads and turns it does not use. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 65 +++++++++++++------ .../src/orchestration-v2/Orchestrator.ts | 36 ++++++---- 2 files changed, 67 insertions(+), 34 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 74596067a739..9baf0df10a48 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2404,7 +2404,8 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); - // Finalizing a child reads, from its parent, only that child's tasks and the transfers it sent. + // Finalizing a child reads, from its parent, only that child's tasks and the transfers it sent, + // and the 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; @@ -2413,15 +2414,31 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { 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: NodeId.make("node:narrow-finalize-seed"), - deliveryState: "acknowledged", + 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, @@ -2447,27 +2464,33 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { yield* orchestrator.recoverDelegatedTasks.pipe( Effect.provideService(Statement.CurrentTransformer, observe), ); - const task = (yield* orchestrator.getThreadProjection(threadId)).subagents.find( - (candidate) => candidate.id === child.taskId, - ); + const projection = yield* orchestrator.getThreadProjection(threadId); + const task = projection.subagents.find((candidate) => candidate.id === child.taskId); assert.equal(task?.status, "completed"); - // The finalize read names the child: its tasks by child thread, its transfers by source. - assert.isTrue( - parentReads.some( - (query) => - query.includes("FROM orchestration_v2_projection_subagents") && - /AND child_thread_id (=|IN)/.test(query), - ), - "the parent's subagent read is not narrowed to the child", + // 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"), ); - assert.isTrue( - parentReads.some( - (query) => - query.includes("FROM orchestration_v2_projection_context_transfers") && - /AND source_thread_id (=|IN)/.test(query), - ), - "the parent's context-transfer read is not narrowed to the child", + 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", + ); + // 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}`); }), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 603c4492a5f4..ea0c173e2bfd 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12252,7 +12252,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; @@ -12544,19 +12544,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItems: childResult.turnItems, }; const parentThreadId = childControls.thread.lineage.parentThreadId; + // Only this child's tasks and result transfers; a delivery's other tasks are read by id below. const parentProjection = yield* projectionStore.getThreadRecords( parentThreadId, - [ - "runs", - "messages", - "subagents", - "providerThreads", - "providerTurns", - "nodes", - "turnItems", - "contextTransfers", - ], - { turnItemTypes: ["subagent"], messageRoles: ["user"] }, + ["runs", "messages", "subagents", "nodes", "turnItems", "contextTransfers"], + { + turnItemTypes: ["subagent"], + messageRoles: ["user"], + subagentChildThreadIds: [childThreadId], + contextTransferSourceThreadIds: [childThreadId], + }, ); const task = parentProjection.subagents.find( (candidate) => @@ -12597,8 +12594,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completedAt: now, updatedAt: now, }; + // The delivery this result may join names its other tasks in its text. + const deliveryTaskIds = (parentRun?.delegatedCompletion?.delivery?.taskIds ?? []).filter( + (taskId) => !parentProjection.subagents.some((candidate) => candidate.id === taskId), + ); + const deliveryTasks = + deliveryTaskIds.length === 0 + ? [] + : (yield* projectionStore.getThreadRecords(parentThreadId, ["subagents"], { + subagentIds: deliveryTaskIds, + })).subagents; const completionPlan = yield* planDelegatedCompletionDelivery({ - parentProjection, + parentProjection: { + ...parentProjection, + subagents: [...parentProjection.subagents, ...deliveryTasks], + }, parentRun, task, updatedTask, From 17c0954aacd5e82d835c9d8e3c887252d503f1bb Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 15:45:38 +0000 Subject: [PATCH 3/6] perf(server): finalize reads only the task's node, card and delivery message The terminal-result read loaded every parent node, subagent card and user message to use one of each. It now asks for the task's node and card by id, and reads the delivery message it may join by id, with that delivery's other tasks. The status mirror for a follow-up run reads the node and card the same way. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 202 +++++++++++++++++- .../src/orchestration-v2/Orchestrator.ts | 35 +-- 2 files changed, 219 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 9baf0df10a48..c502174da7e3 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2311,6 +2311,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("queues a child result without creating a provider history handoff", () => Effect.gen(function* () { @@ -2404,11 +2506,12 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); - // Finalizing a child reads, from its parent, only that child's tasks and the transfers it sent, - // and the other tasks of a delivery it joins by id. + // 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"); @@ -2450,12 +2553,81 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { 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 observe: Statement.Transformer = (statement) => { const [query, params] = statement.compile(); if ( params.includes(threadId) && - /FROM orchestration_v2_projection_(subagents|context_transfers)\b/.test(query) + (/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); } @@ -2486,6 +2658,30 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { [], "a parent context-transfer read is not narrowed", ); + // 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]); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ea0c173e2bfd..93e87414f8bf 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12477,7 +12477,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 = { @@ -12544,15 +12544,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItems: childResult.turnItems, }; const parentThreadId = childControls.thread.lineage.parentThreadId; - // Only this child's tasks and result transfers; a delivery's other tasks are read by id below. + // Only this child's tasks, node, card and result transfers; a delivery is read by id below. const parentProjection = yield* projectionStore.getThreadRecords( parentThreadId, - ["runs", "messages", "subagents", "nodes", "turnItems", "contextTransfers"], + ["runs", "subagents", "nodes", "turnItems", "contextTransfers"], { turnItemTypes: ["subagent"], - messageRoles: ["user"], subagentChildThreadIds: [childThreadId], contextTransferSourceThreadIds: [childThreadId], + nodeIds: [options.task.id], + turnItemNodeIds: [options.task.id], }, ); const task = parentProjection.subagents.find( @@ -12594,20 +12595,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completedAt: now, updatedAt: now, }; - // The delivery this result may join names its other tasks in its text. - const deliveryTaskIds = (parentRun?.delegatedCompletion?.delivery?.taskIds ?? []).filter( - (taskId) => !parentProjection.subagents.some((candidate) => candidate.id === taskId), - ); - const deliveryTasks = - deliveryTaskIds.length === 0 - ? [] - : (yield* projectionStore.getThreadRecords(parentThreadId, ["subagents"], { - subagentIds: deliveryTaskIds, - })).subagents; + // 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), + ), + }); const completionPlan = yield* planDelegatedCompletionDelivery({ parentProjection: { ...parentProjection, - subagents: [...parentProjection.subagents, ...deliveryTasks], + messages: deliveryRecords.messages, + subagents: [...parentProjection.subagents, ...deliveryRecords.subagents], }, parentRun, task, From 7bab847afd6318fe3509a66dfac8eda4d0eb4806 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Thu, 12 Nov 2026 11:27:52 +0000 Subject: [PATCH 4/6] perf(server): a card read by node seeks the node index A turn-item read filtered by node ids still walked every turn item of the thread through the thread index, so finalizing a subagent visited every card of its parent to return one. It now seeks the node index, as the other id-filtered reads already do. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 56 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 2 +- 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index f605ef82b536..5ff3de949f42 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2212,6 +2212,62 @@ const parentRecordsNarrowToOneChild = Effect.gen(function* () { 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.include( + cardPlan[0] ?? "", + "USING INDEX orchestration_v2_projection_turn_items_node_ordinal_idx (node_id=?)", + ); // Several children read through the id list. const both = yield* store.getThreadRecords(parent, fields, { subagentChildThreadIds: [child, sibling], diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 3802f09d0833..3d49039220f1 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -3463,7 +3463,7 @@ export const layer: Layer.Layer = readsConversationItemsOnly(filter) ? sql`INDEXED BY orchestration_v2_projection_turn_items_conversation_idx WHERE thread_id = ${threadId} AND type NOT IN ${CONVERSATION_INDEX_EXCLUDED_TYPES}` - : sql`WHERE thread_id = ${threadId}` + : sql`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)}))`} From 09a547b56bafdaf1f0463bdb44968915262c1abd Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Thu, 12 Nov 2026 11:52:33 +0000 Subject: [PATCH 5/6] test(server): the finalize test checks each transfer read names the child The read observer kept only query text, so a transfer read bound to another thread still looked narrowed. It now keeps each transfer read's bound ids and requires the finishing child's thread among them. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 975e55ebcc73..32358a4c594b 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -4059,6 +4059,7 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { ]), }); const parentReads: Array = []; + const transferReadParams: Array> = []; const observe: Statement.Transformer = (statement) => { const [query, params] = statement.compile(); if ( @@ -4069,6 +4070,9 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { )) ) { parentReads.push(query); + if (query.includes("FROM orchestration_v2_projection_context_transfers")) { + transferReadParams.push(params); + } } return Effect.succeed(statement); }; @@ -4097,6 +4101,12 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { [], "a parent context-transfer read is not narrowed", ); + // 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/], From 47c7d9fc16dd46c39cf27a8297328afdf38c35ae Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Sat, 14 Nov 2026 19:17:23 +0000 Subject: [PATCH 6/6] perf(server): finalize reads only the task's run and its delivery's run The result read loaded every run of the parent to use two: the task's run and the run its delivery message started. It now reads the first by id, and the second by id once the delivery message is known. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 37 ++++++++++++++++++- .../src/orchestration-v2/Orchestrator.ts | 20 +++++++--- 2 files changed, 50 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 32358a4c594b..bff7efe7c5d5 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -29,6 +29,7 @@ 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"; @@ -4060,8 +4061,21 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }); const parentReads: Array = []; const transferReadParams: Array> = []; - const observe: Statement.Transformer = (statement) => { + // 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) || @@ -4101,6 +4115,27 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { [], "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)), diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 2542c91c35a5..cda2822f0349 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13929,10 +13929,10 @@ 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 result transfers; a delivery is read by id below. + // Only this child's tasks, node, card and transfers; runs and a delivery are read by id. const parentProjection = yield* projectionStore.getThreadRecords( parentThreadId, - ["runs", "subagents", "nodes", "turnItems", "contextTransfers"], + ["subagents", "nodes", "turnItems", "contextTransfers"], { turnItemTypes: ["subagent"], subagentChildThreadIds: [childThreadId], @@ -13967,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, @@ -14003,9 +14004,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio !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, + runs: [...(parentRun === undefined ? [] : [parentRun]), ...deliveryRuns], messages: deliveryRecords.messages, subagents: [...parentProjection.subagents, ...deliveryRecords.subagents], },