From 70c8b754960557d21ac3770d6855c744a3bdcc7a Mon Sep 17 00:00:00 2001 From: Will White Date: Mon, 19 Oct 2026 12:28:42 +1000 Subject: [PATCH 1/2] fix(server): reconcile stale tasks created by legacy steers --- .../DelegatedCompletionDelivery.test.ts | 267 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 87 ++++-- .../ProjectionRecovery.test.ts | 13 +- .../src/orchestration-v2/ProjectionStore.ts | 45 ++- 4 files changed, 376 insertions(+), 36 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 2fb707a58f2d..2c0d3d1ef111 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -22,6 +22,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; import * as Tracer from "effect/Tracer"; +import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as ServerConfig from "../config.ts"; @@ -921,6 +922,272 @@ const seedRestartCancelledChild = (input: { return { taskId, childThreadId, childRunId }; }); +it.effect.each([ + "completed", + "failed", + "cancelled", + "interrupted", + "running", + "waiting", + "retry-running", + "retry-completed", +] as const)("recovers legacy steered tasks with a %s run", (scenario) => + Effect.gen(function* () { + const retry = scenario === "retry-running" || scenario === "retry-completed"; + const status = + scenario === "retry-running" + ? "running" + : scenario === "retry-completed" + ? "completed" + : scenario; + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:legacy-steer-parent"); + const projectId = ProjectId.make("project:legacy-steer"); + const runId = RunId.make("run:legacy-steer-parent"); + const rootNodeId = NodeId.make("node:legacy-steer-parent"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:legacy-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "legacy-steer-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + const original = (yield* orchestrator.getThreadProjection(threadId)).subagents.find( + (task) => task.id === child.taskId, + )!; + const taskIds = [0, 1, 2, 3].map((key) => + NodeId.make( + `node:delegated-task:${encodeURIComponent(`command:mcp:legacy:thread-send:${key}`)}`, + ), + ); + yield* eventSink.write({ + commandId: reconcileCommandId("legacy-steers"), + events: taskIds.flatMap((taskId, index) => [ + { + id: EventId.make(`event:legacy-task:${index}`), + type: "subagent.updated" as const, + threadId, + occurredAt: now, + payload: { + ...original, + id: taskId, + status: "running" as const, + result: null, + completedAt: null, + completionDelivery: { + state: index === 0 ? ("pending" as const) : ("acknowledged" as const), + observedByRunId: index === 0 ? null : runId, + }, + }, + }, + { + id: EventId.make(`event:legacy-item:${index}`), + type: "turn-item.updated" as const, + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:legacy:${index}`), + threadId, + runId, + nodeId: taskId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: index + 1, + status: "running" as const, + title: "Historical review", + startedAt: now, + completedAt: null, + updatedAt: now, + type: "subagent" as const, + subagentId: taskId, + origin: "app_owned" as const, + driver, + providerInstanceId: modelSelection.instanceId, + childThreadId: child.childThreadId, + prompt: "Follow-up guidance", + result: null, + }, + }, + { + id: EventId.make(`event:legacy-node:${index}`), + type: "node.updated" as const, + threadId, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId, + parentNodeId: rootNodeId, + rootNodeId, + kind: "subagent" as const, + status: "running" as const, + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }, + { + id: EventId.make(`event:legacy-message:${index}`), + type: "message.updated" as const, + threadId: child.childThreadId, + occurredAt: now, + payload: { + id: MessageId.make(`message:mcp:legacy:thread-send:${index}`), + threadId: child.childThreadId, + runId: child.childRunId, + nodeId: null, + role: "user" as const, + text: "Follow-up guidance", + attachments: [], + streaming: false, + createdBy: "agent" as const, + creationSource: "mcp" as const, + senderThreadId: threadId, + createdAt: now, + updatedAt: now, + }, + }, + ]), + }); + const parentRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.id === runId, + )!; + yield* eventSink.write({ + commandId: reconcileCommandId("legacy-parent-ended"), + events: [ + { + id: EventId.make("event:legacy-parent-ended"), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }, + ], + }); + const before = yield* orchestrator.getThreadProjection(threadId); + assert.lengthOf( + derivePendingBackgroundWork({ + latestRun: before.runs[0], + runs: before.runs, + turnItems: before.turnItems, + providerThreads: before.providerThreads, + }), + 4, + ); + assert.lengthOf( + (yield* orchestrator.getThreadShell(threadId))!.pendingBackgroundTasks ?? [], + 4, + ); + const childRun = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (run) => run.id === child.childRunId, + )!; + yield* eventSink.write({ + commandId: reconcileCommandId("legacy-run-outcome"), + events: [ + { + id: EventId.make("event:legacy-run-outcome"), + type: "run.updated", + threadId: child.childThreadId, + runId: child.childRunId, + occurredAt: now, + payload: { + ...childRun, + status: retry ? "cancelled" : status, + completedAt: !retry && (status === "running" || status === "waiting") ? null : now, + }, + }, + (() => { + const event = runEvent({ + threadId: child.childThreadId, + runId: RunId.make("run:legacy-new-work"), + ordinal: 2, + status: retry && status === "completed" ? "completed" : "running", + now, + }); + return { + ...event, + payload: { + ...event.payload, + ...(retry ? { restartContinuationOfRunId: child.childRunId } : {}), + }, + }; + })(), + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const recovered = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + recovered.subagents.filter((task) => taskIds.includes(task.id)).map((task) => task.status), + taskIds.map(() => (["running", "waiting"].includes(status) ? "running" : status)), + ); + assert.isTrue( + recovered.subagents + .filter((task) => taskIds.includes(task.id)) + .every((task) => + ["running", "waiting"].includes(status) + ? task.completedAt === null && task.result === null + : DateTime.toEpochMillis(task.completedAt!) === DateTime.toEpochMillis(now) && + task.result !== null, + ), + ); + assert.equal( + recovered.subagents.find((task) => task.id === taskIds[1])?.completionDelivery?.state, + "acknowledged", + ); + const expectedCount = ["running", "waiting"].includes(status) ? 4 : 0; + assert.lengthOf( + derivePendingBackgroundWork({ + latestRun: recovered.runs[0], + runs: recovered.runs, + turnItems: recovered.turnItems, + providerThreads: recovered.providerThreads, + }), + expectedCount, + ); + assert.lengthOf( + (yield* orchestrator.getThreadShell(threadId))!.pendingBackgroundTasks ?? [], + expectedCount, + ); + assert.deepEqual( + recovered.nodes.filter((node) => taskIds.includes(node.id)).map((node) => node.status), + taskIds.map(() => (["running", "waiting"].includes(status) ? "running" : status)), + ); + const sequence = yield* eventSink.latestSequence(); + yield* orchestrator.recoverDelegatedTasks; + assert.equal(yield* eventSink.latestSequence(), sequence); + assert.equal( + (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (run) => run.id === RunId.make("run:legacy-new-work"), + )?.status, + retry && status === "completed" ? "completed" : "running", + ); + }).pipe(Effect.provide(layerTest)), +); + it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( "subagent follow-ups", (it) => { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3f5ed287bcef..6117d394b93f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10253,11 +10253,46 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return; } const taskRuns = childControls.runs.filter((run) => run.delegatedTaskId === options.taskId); + // Before steers stopped registering tasks, the running provider could overwrite + // delegatedTaskId from its startup snapshot. MCP message and task ids retain + // the same command identity even when several steers joined that run. + const legacySteerMessage = + taskRuns.length === 0 + ? childControls.messages.find( + (message) => + message.role === "user" && + message.creationSource === "mcp" && + message.senderThreadId === childControls.thread.lineage.parentThreadId && + message.id.startsWith("message:mcp:") && + idAllocator.derive.delegatedTaskNode({ + commandId: CommandId.make(message.id.replace(/^message:/, "command:")), + }) === options.taskId, + ) + : undefined; + let legacySteerRun = + legacySteerMessage === undefined + ? undefined + : childControls.runs.find( + (run) => + run.id === legacySteerMessage.runId && run.userMessageId !== legacySteerMessage.id, + ); + if (legacySteerRun !== undefined) { + const continuationIds = new Set([legacySteerRun.id]); + for (const run of childControls.runs.toSorted((a, b) => a.ordinal - b.ordinal)) { + if ( + run.restartContinuationOfRunId !== undefined && + continuationIds.has(run.restartContinuationOfRunId) + ) { + continuationIds.add(run.id); + legacySteerRun = run; + } + } + } const childRun = taskRuns.length === 0 ? options.taskId === forkedFrom.nodeId ? options.resultRun - : undefined + : legacySteerRun : taskRuns.toSorted((a, b) => b.ordinal - a.ordinal)[0]; if (childRun === undefined) return; const outcome = runOutcome(childRun); @@ -10311,7 +10346,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio candidate.origin === "app_owned" && candidate.childThreadId === childThreadId, ); - if (task === undefined) { + if ( + task === undefined || + (legacySteerRun !== undefined && !isOrchestrationV2WorkActive(task.status)) + ) { return; } const existingResultTransfer = parentProjection.contextTransfers.find( @@ -10322,7 +10360,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (transfer.sourcePoint.runId === childRun.id || (options.taskId === forkedFrom.nodeId && transfer.sourcePoint.runId === undefined)), ); - if (existingResultTransfer !== undefined) { + if ( + existingResultTransfer !== undefined && + (legacySteerRun === undefined || task.result !== null) + ) { return; } @@ -10336,12 +10377,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const parentTurnItem = parentProjection.turnItems.find( (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id, ); + const completedAt = legacySteerRun?.completedAt ?? now; const updatedTask: OrchestrationV2Subagent = { ...task, providerThreadId: childRun.providerThreadId, status: terminalStatus, result: result.text, - completedAt: now, + completedAt, updatedAt: now, }; const completionPlan = yield* planDelegatedCompletionDelivery({ @@ -10351,11 +10393,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedTask, now, }); - const resultTransferId = yield* idAllocator.allocate.contextTransfer({ - sourceThreadId: childThreadId, - targetThreadId: parentThreadId, - type: "subagent_result", - }); + const resultTransferId = + existingResultTransfer?.id ?? + (yield* idAllocator.allocate.contextTransfer({ + sourceThreadId: childThreadId, + targetThreadId: parentThreadId, + type: "subagent_result", + })); const childProviderThread = childRun.providerThreadId === null ? undefined @@ -10369,6 +10413,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (candidate) => candidate.id === parentRun.providerThreadId, ); const resultHandoff: OrchestrationV2ContextHandoff | null = + existingResultTransfer !== undefined || parentRun === undefined || childProviderThread === undefined || parentProviderThread === undefined @@ -10483,7 +10528,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...parentNode, status: terminalStatus, providerThreadId: childRun.providerThreadId, - completedAt: now, + completedAt, }, }, ]), @@ -10501,7 +10546,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...parentTurnItem, status: terminalStatus, result: result.text, - completedAt: now, + completedAt, updatedAt: now, }, }, @@ -10518,14 +10563,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: resultHandoff, }, ]), - { - type: "context-transfer.created", - threadId: parentThreadId, - ...(parentRun === undefined ? {} : { runId: parentRun.id }), - providerInstanceId: childRun.providerInstanceId, - occurredAt: now, - payload: resultTransfer, - }, + ...(existingResultTransfer === undefined + ? [ + { + type: "context-transfer.created" as const, + threadId: parentThreadId, + ...(parentRun === undefined ? {} : { runId: parentRun.id }), + providerInstanceId: childRun.providerInstanceId, + occurredAt: now, + payload: resultTransfer, + }, + ] + : []), ]); if (completionPlan.offer && completionPlan.parentRun !== undefined) { diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6e0dd990ce38..d70a1874b199 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -444,9 +444,11 @@ it.effect("recovers terminal subagent results until their cross-thread transfer ); it.effect.each([ - { name: "SQLite", layer: layerTest }, - { name: "memory", layer: ProjectionStore.layerMemory }, -])("recovers only recorded follow-up results with $name projections", ({ layer }) => + { name: "SQLite", layer: layerTest, legacy: false }, + { name: "memory", layer: ProjectionStore.layerMemory, legacy: false }, + { name: "legacy SQLite", layer: layerTest, legacy: true }, + { name: "legacy memory", layer: ProjectionStore.layerMemory, legacy: true }, +])("recovers follow-up results with $name projections", ({ layer, legacy }) => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; @@ -483,7 +485,10 @@ it.effect.each([ yield* createRun(child, "completed", { ordinal: 2 }); assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []); const taskId = NodeId.make("node:followup"); - yield* createRun(child, "completed", { ordinal: 3, delegatedTaskId: taskId }); + yield* createRun(child, "completed", { + ordinal: 3, + ...(legacy ? {} : { delegatedTaskId: taskId }), + }); yield* createRun(child, "running", { ordinal: 4 }); const task = { id: taskId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 98eefbe2ed1c..191d4e53fc95 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -579,19 +579,22 @@ function needsRecovery( projection.thread.lineage.relationshipToParent === "subagent" && parentThreadId !== null && projection.thread.forkedFrom?.type === "node" && - ((parent?.subagents ?? []).some( - (task) => - task.origin === "app_owned" && - task.result === null && - task.childThreadId === projection.thread.id && - projection.runs.some( - (run) => - run.delegatedTaskId === task.id && - ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( - run.status, - ), - ), - ) || + ((parent?.subagents ?? []).some((task) => { + if ( + task.origin !== "app_owned" || + task.result !== null || + task.childThreadId !== projection.thread.id + ) + return false; + const boundRuns = projection.runs.filter((run) => run.delegatedTaskId === task.id); + const candidates = + boundRuns.length === 0 && ["pending", "running", "waiting"].includes(task.status) + ? projection.runs + : boundRuns; + return candidates.some((run) => + ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(run.status), + ); + }) || (["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( latestUnheldRun(projection.runs)?.status ?? "idle", ) && @@ -3703,6 +3706,22 @@ export const layer: Layer.Layer = AND task.child_thread_id = child.thread_id AND json_extract(task.payload_json, '$.origin') = 'app_owned' AND json_extract(task.payload_json, '$.result') IS NULL + ) OR EXISTS ( + SELECT 1 FROM orchestration_v2_projection_subagents AS task + WHERE task.child_thread_id = child.thread_id + AND task.origin = 'app_owned' + AND task.status IN ('pending', 'running', 'waiting') + AND json_extract(task.payload_json, '$.result') IS NULL + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS bound + WHERE bound.thread_id = child.thread_id + AND json_extract(bound.payload_json, '$.delegatedTaskId') = task.subagent_id + ) + AND EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS ended + WHERE ended.thread_id = child.thread_id + AND ended.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + ) ) OR ( -- The original task still waits for the latest unheld work. (SELECT status FROM orchestration_v2_projection_runs From d925cac5090ba9a7e0a37383567b9bb68b24c485 Mon Sep 17 00:00:00 2001 From: Will White Date: Mon, 19 Oct 2026 12:35:40 +1000 Subject: [PATCH 2/2] fix(server): preserve legacy task identity across partial run bindings --- .../DelegatedCompletionDelivery.test.ts | 47 ++++++++++++++++--- .../src/orchestration-v2/Orchestrator.ts | 44 ++++++++--------- 2 files changed, 64 insertions(+), 27 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 2c0d3d1ef111..f8353f9c66a1 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -931,13 +931,24 @@ it.effect.each([ "waiting", "retry-running", "retry-completed", + "bound-duplicate", + "bound-retry-running", + "bound-retry-completed", + "unrelated-retry", ] as const)("recovers legacy steered tasks with a %s run", (scenario) => Effect.gen(function* () { - const retry = scenario === "retry-running" || scenario === "retry-completed"; + const retry = + scenario === "retry-running" || + scenario === "retry-completed" || + scenario === "bound-retry-running" || + scenario === "bound-retry-completed"; const status = - scenario === "retry-running" + scenario === "retry-running" || scenario === "bound-retry-running" ? "running" - : scenario === "retry-completed" + : scenario === "retry-completed" || + scenario === "bound-retry-completed" || + scenario === "bound-duplicate" || + scenario === "unrelated-retry" ? "completed" : scenario; const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -1105,7 +1116,7 @@ it.effect.each([ const childRun = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( (run) => run.id === child.childRunId, )!; - yield* eventSink.write({ + yield* eventSink.writeWithEffects({ commandId: reconcileCommandId("legacy-run-outcome"), events: [ { @@ -1117,6 +1128,7 @@ it.effect.each([ payload: { ...childRun, status: retry ? "cancelled" : status, + ...(scenario.startsWith("bound-") ? { delegatedTaskId: taskIds[3]! } : {}), completedAt: !retry && (status === "running" || status === "waiting") ? null : now, }, }, @@ -1125,7 +1137,12 @@ it.effect.each([ threadId: child.childThreadId, runId: RunId.make("run:legacy-new-work"), ordinal: 2, - status: retry && status === "completed" ? "completed" : "running", + status: + scenario === "unrelated-retry" + ? "cancelled" + : retry && status === "completed" + ? "completed" + : "running", now, }); return { @@ -1137,6 +1154,20 @@ it.effect.each([ }; })(), ], + effects: + scenario === "unrelated-retry" + ? [ + { + id: "effect:restart-continuation:run:legacy-new-work", + commandId: reconcileCommandId("legacy-run-outcome"), + threadId: child.childThreadId, + request: { + type: "provider-runtime.continue", + sourceRunId: RunId.make("run:legacy-new-work"), + }, + }, + ] + : [], }); yield* orchestrator.recoverDelegatedTasks; const recovered = yield* orchestrator.getThreadProjection(threadId); @@ -1183,7 +1214,11 @@ it.effect.each([ (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( (run) => run.id === RunId.make("run:legacy-new-work"), )?.status, - retry && status === "completed" ? "completed" : "running", + scenario === "unrelated-retry" + ? "cancelled" + : retry && status === "completed" + ? "completed" + : "running", ); }).pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6117d394b93f..728f5bd53901 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10256,19 +10256,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Before steers stopped registering tasks, the running provider could overwrite // delegatedTaskId from its startup snapshot. MCP message and task ids retain // the same command identity even when several steers joined that run. - const legacySteerMessage = - taskRuns.length === 0 - ? childControls.messages.find( - (message) => - message.role === "user" && - message.creationSource === "mcp" && - message.senderThreadId === childControls.thread.lineage.parentThreadId && - message.id.startsWith("message:mcp:") && - idAllocator.derive.delegatedTaskNode({ - commandId: CommandId.make(message.id.replace(/^message:/, "command:")), - }) === options.taskId, - ) - : undefined; + const legacySteerMessage = childControls.messages.find( + (message) => + message.role === "user" && + message.creationSource === "mcp" && + message.senderThreadId === childControls.thread.lineage.parentThreadId && + message.id.startsWith("message:mcp:") && + idAllocator.derive.delegatedTaskNode({ + commandId: CommandId.make(message.id.replace(/^message:/, "command:")), + }) === options.taskId, + ); let legacySteerRun = legacySteerMessage === undefined ? undefined @@ -10276,8 +10273,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (run) => run.id === legacySteerMessage.runId && run.userMessageId !== legacySteerMessage.id, ); + const continuationIds = new Set(); if (legacySteerRun !== undefined) { - const continuationIds = new Set([legacySteerRun.id]); + continuationIds.add(legacySteerRun.id); for (const run of childControls.runs.toSorted((a, b) => a.ordinal - b.ordinal)) { if ( run.restartContinuationOfRunId !== undefined && @@ -10289,11 +10287,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } } const childRun = - taskRuns.length === 0 - ? options.taskId === forkedFrom.nodeId - ? options.resultRun - : legacySteerRun - : taskRuns.toSorted((a, b) => b.ordinal - a.ordinal)[0]; + [...taskRuns, ...(legacySteerRun === undefined ? [] : [legacySteerRun])].toSorted( + (a, b) => b.ordinal - a.ordinal, + )[0] ?? (options.taskId === forkedFrom.nodeId ? options.resultRun : undefined); if (childRun === undefined) return; const outcome = runOutcome(childRun); const terminalStatus = delegatedTaskTerminalStatus(outcome); @@ -10302,7 +10298,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } if ( yield* childAwaitsRestartContinuation( - taskRuns.length === 0 ? childControls.runs : taskRuns, + legacySteerRun !== undefined + ? childControls.runs.filter( + (run) => continuationIds.has(run.id) || run.delegatedTaskId === options.taskId, + ) + : taskRuns.length === 0 + ? childControls.runs + : taskRuns, childRun, options?.settledContinuationOf, ) @@ -10377,7 +10379,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const parentTurnItem = parentProjection.turnItems.find( (candidate) => candidate.type === "subagent" && candidate.subagentId === task.id, ); - const completedAt = legacySteerRun?.completedAt ?? now; + const completedAt = legacySteerRun === undefined ? now : (childRun.completedAt ?? now); const updatedTask: OrchestrationV2Subagent = { ...task, providerThreadId: childRun.providerThreadId,