diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 066722ea6625f..9b13954075e85 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -91,6 +91,7 @@ import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterReg import { subagentResultForRun, delegatedTaskProgress, + delegatedTaskResultRun, } from "@t3tools/provider-core/server/subagentProjection"; import { DispatchModeLimit, @@ -500,16 +501,19 @@ function endsTaskWait(result: OrchestratorMcpDelegateTaskResult): boolean { const isApprovalRequestKind = Schema.is(ProviderRequestKind); function pendingTaskApproval( - child: Pick, + child: Pick, taskId: NodeId, originalRun: OrchestrationV2Run | undefined, ) { + // A follow-up promoted to Steer waits on approvals in the run it joined. + const joinedRunId = delegatedTaskResultRun(child, taskId)?.id; const taskRunIds = new Set( child.runs .filter( (run) => !ThreadManagementService.isTerminalRunStatus(run.status) && (run.delegatedTaskId === taskId || + run.id === joinedRunId || // Legacy tasks own only their original run, not later turns in the child thread. (run.delegatedTaskId === undefined && run.id === originalRun?.id)), ) @@ -544,7 +548,11 @@ function directChildTaskIds( const ids = new Set(); if (target.thread.forkedFrom?.type === "node") ids.add(target.thread.forkedFrom.nodeId); for (const run of target.runs) { - if (returnedRunIds.has(run.id) && run.delegatedTaskId !== undefined) { + // A promoted follow-up's cancelled run can be off the page while its joined run is on it. + if ( + run.delegatedTaskId !== undefined && + (returnedRunIds.has(run.id) || run.status === "cancelled") + ) { ids.add(run.delegatedTaskId); } } @@ -553,8 +561,7 @@ function directChildTaskIds( function directAppOwnedChildTasks( parent: Pick, - target: Pick, - returnedRunIds: ReadonlySet, + target: Pick, ): ReadonlyArray { if ( target.thread.lineage.parentThreadId !== parent.thread.id || @@ -562,23 +569,10 @@ function directAppOwnedChildTasks( ) { return []; } - const original = parent.subagents.find( - (task) => - task.origin === "app_owned" && - task.threadId === parent.thread.id && - task.id === - (target.thread.forkedFrom?.type === "node" ? target.thread.forkedFrom.nodeId : undefined) && - task.childThreadId === target.thread.id, - ); - const taskIds = new Set( - target.runs - .filter((run) => returnedRunIds.has(run.id)) - .map((run) => run.delegatedTaskId ?? original?.id), - ); return parent.subagents.filter( (task) => - taskIds.has(task.id) && task.origin === "app_owned" && + task.threadId === parent.thread.id && task.childThreadId === target.thread.id, ); } @@ -586,10 +580,12 @@ function directAppOwnedChildTasks( // The child's transfers include the results it sent, so the parent's transfers are never read. function taskResultTransfer( parentThreadId: ThreadId, - child: Pick, + child: Pick, task: OrchestrationV2Subagent, ) { const originalRun = delegatedTaskRun(child, task); + // A follow-up promoted to Steer shares the transfer of the run it joined. + const resultRun = delegatedTaskResultRun(child, task.id); const resultRunIds = new Set( child.runs .filter( @@ -600,6 +596,7 @@ function taskResultTransfer( .map((run) => run.id), ); if (originalRun !== undefined) resultRunIds.add(originalRun.id); + if (resultRun !== undefined) resultRunIds.add(resultRun.id); return child.contextTransfers.find( (transfer) => transfer.type === "subagent_result" && @@ -611,24 +608,17 @@ function taskResultTransfer( } function pageIncludesTerminalTaskResult(input: { - readonly parentThreadId: ThreadId; readonly page: ReadonlyArray; - readonly task: OrchestrationV2Subagent; + readonly run: OrchestrationV2Run; readonly target: Pick< OrchestrationV2ThreadProjection, - "thread" | "runs" | "contextTransfers" | "messages" | "turnItems" | "providerTurns" + "thread" | "messages" | "turnItems" | "providerTurns" >; readonly maxChars: number; }): boolean { - const transfer = taskResultTransfer(input.parentThreadId, input.target, input.task); - if (transfer === undefined) return false; - const run = - transfer.sourcePoint.runId === undefined - ? delegatedTaskRun(input.target, input.task) - : input.target.runs.find((run) => run.id === transfer.sourcePoint.runId); - if (run === undefined || !isTerminalTaskStatus(taskStatusForRun(run))) return false; - - const result = subagentResultForRun(input.target, run); + if (!isTerminalTaskStatus(taskStatusForRun(input.run))) return false; + + const result = subagentResultForRun(input.target, input.run); if (result.messageId === null && result.turnItemId === null) return false; return input.page.some((row) => { @@ -1489,9 +1479,7 @@ const make = Effect.gen(function* () { const progress = delegatedTaskProgress(childControls); const taskResultRun = childRun?.delegatedTaskId === task.id - ? childControls.runs - .filter((run) => run.delegatedTaskId === task.id) - .toSorted((a, b) => b.ordinal - a.ordinal)[0] + ? delegatedTaskResultRun(childControls, task.id) : progress.resultRun; const taskWorkState = childRun?.delegatedTaskId === task.id @@ -3052,7 +3040,6 @@ const make = Effect.gen(function* () { ), }, target, - returnedRunIds, ); if ( parent !== undefined && @@ -3060,10 +3047,25 @@ const make = Effect.gen(function* () { tasks.length > 0 && (input.textOffset ?? 0) === 0 ) { + // Promotion moves the queued user message to its joined run, possibly outside this page. + const taskMessages = yield* threadManagement + .getThreadRecords(target.thread.id, ["messages"], { + messageIds: target.runs + .filter((run) => run.status === "cancelled" && run.delegatedTaskId !== undefined) + .map((run) => run.userMessageId) + .filter((id): id is MessageId => id !== null), + }) + .pipe(Effect.mapError(threadManagementFailure)); + const taskTarget = { ...target, messages: taskMessages.messages }; for (const task of tasks) { - const transfer = taskResultTransfer(parent.thread.id, target, task); - const resultRunId = transfer?.sourcePoint.runId ?? delegatedTaskRun(target, task)?.id; - const resultRunIds = resultRunId === undefined ? [] : [resultRunId]; + const transfer = taskResultTransfer(parent.thread.id, taskTarget, task); + if (transfer === undefined) continue; + const run = + transfer.sourcePoint.runId === undefined + ? delegatedTaskRun(target, task) + : target.runs.find((candidate) => candidate.id === transfer.sourcePoint.runId); + if (run === undefined || !returnedRunIds.has(run.id)) continue; + const resultRunIds = [run.id]; const resultRecords = yield* threadManagement .getThreadRecords(target.thread.id, ["messages", "turnItems"], { messageRoles: ["assistant"], @@ -3074,9 +3076,8 @@ const make = Effect.gen(function* () { .pipe(Effect.mapError(threadManagementFailure)); if ( pageIncludesTerminalTaskResult({ - parentThreadId: parent.thread.id, page, - task, + run, target: { ...target, ...resultRecords }, maxChars, }) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index a5c123157bae1..597959fa61874 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -7,6 +7,7 @@ import { CheckpointId, CommandId, EventId, + EnvironmentId, MessageId, type ModelSelection, NodeId, @@ -14,6 +15,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, ProviderThreadId, RunId, RuntimeRequestId, @@ -54,6 +56,17 @@ import * as ProviderContinuationRequests from "@t3tools/provider-core/server/Pro import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderContinuationService from "./ProviderContinuationService.ts"; +import { makeSubagentConversationArtifacts } from "@t3tools/provider-core/server/subagentProjection"; +import * as OrchestratorMcpService from "../mcp/OrchestratorMcpService.ts"; +import type { McpInvocationScope } from "../mcp/McpInvocationContext.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ThreadLaunchService from "./ThreadLaunchService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import * as RemoteDelegation from "../peer/RemoteDelegation.ts"; + const layerPlatformTest = Layer.merge( Layer.mergeAll(NodeServices.layer, NodeCrypto.layer), Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({ @@ -4126,3 +4139,532 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); }); + +const layerSteeredTaskTest = OrchestratorMcpService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(GitVcsDriver.GitVcsDriver)({}), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ServerSettings.layerTest(), + NodeServices.layer, + NodeCrypto.layer, + ), + ), + Layer.provideMerge(layerTest), +); + +const seedSteeredTasks = Effect.fnUntraced(function* ( + name: string, + { queueHeld = false, monitor = false } = {}, +) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make(`${name}:parent`); + const parentRunId = RunId.make(`${name}:parent`); + const rootNodeId = NodeId.make(`${name}:parent`); + const projectId = ProjectId.make(`${name}:project`); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + runId: parentRunId, + rootNodeId, + projectId, + taskId: NodeId.make(`${name}:seed`), + deliveryState: "delivered", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + parentRunId, + projectId, + rootNodeId, + name, + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTask(child.childThreadId, child.childRunId); + const original = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === child.taskId, + )!; + const steeredTaskId = NodeId.make(`${name}:steered`); + const promotedTaskId = NodeId.make(`${name}:promoted`); + const promotedMessageId = MessageId.make(`${name}:promoted`); + let monitorRun: OrchestrationV2Run | undefined; + if (monitor) { + const messageId = MessageId.make(`${name}:monitor`); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make(`${name}:monitor`), + threadId: child.childThreadId, + messageId, + text: "A monitor reported new output.", + attachments: [], + notification: { + source: { kind: "monitor" }, + outcome: "updated", + summary: "Monitor updated", + }, + dispatchMode: { type: "queue_after_active" }, + }); + monitorRun = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (run) => run.userMessageId === messageId, + )!; + assert.isUndefined(monitorRun.delegatedTaskId); + assert.lengthOf( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.filter( + (task) => task.childThreadId === child.childThreadId, + ), + 1, + ); + } + const steeredRun: OrchestrationV2Run = { + ...(monitorRun ?? { + ...runEvent({ + threadId: child.childThreadId, + runId: RunId.make(`${name}:steered`), + ordinal: 2, + status: "running", + now, + }).payload, + delegatedTaskId: steeredTaskId, + }), + status: "running", + }; + const promotedRun = { + ...runEvent({ + threadId: child.childThreadId, + runId: RunId.make(`${name}:promoted`), + ordinal: 3, + status: "cancelled", + userMessageId: promotedMessageId, + now, + }).payload, + delegatedTaskId: promotedTaskId, + startedAt: null, + }; + const conversationEvents = (role: "user" | "assistant", runId: RunId, messageId: MessageId) => { + const { message, turnItem } = makeSubagentConversationArtifacts({ + threadId: child.childThreadId, + rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + messageId, + turnItemId: TurnItemId.make(`${messageId}:item`), + role, + text: role === "user" ? "Also check the queued follow-up." : "Shared result.", + ordinal: role === "user" ? 1 : 2, + now, + }); + return [ + { + id: EventId.make(`${messageId}:message`), + type: "message.updated" as const, + threadId: child.childThreadId, + runId, + occurredAt: now, + payload: { ...message, runId }, + }, + { + id: EventId.make(`${messageId}:item`), + type: "turn-item.updated" as const, + threadId: child.childThreadId, + runId, + occurredAt: now, + payload: { ...turnItem, runId }, + }, + ]; + }; + const beforePromotion = yield* sink.latestSequence(); + yield* sink.write({ + events: [ + ...(monitor ? [promotedTaskId] : [steeredTaskId, promotedTaskId]).map((id) => ({ + id: EventId.make(`${id}:task-seed`), + type: "subagent.updated" as const, + threadId: parentThreadId, + runId: parentRunId, + occurredAt: now, + payload: { + ...original, + id, + status: + id === promotedTaskId + ? queueHeld + ? ("idle" as const) + : ("pending" as const) + : ("running" as const), + result: null, + completedAt: null, + completionDelivery: undefined, + }, + })), + ...[steeredRun, promotedRun].map((run) => ({ + id: EventId.make(`${run.id}:run-seed`), + type: "run.updated" as const, + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: run, + })), + ...conversationEvents("user", steeredRun.id, promotedMessageId), + ], + }); + const finish = Effect.fnUntraced(function* (run: OrchestrationV2Run) { + const messageId = MessageId.make(`${run.id}:result`); + yield* sink.write({ + events: [ + ...conversationEvents("assistant", run.id, messageId), + { + id: EventId.make(`${run.id}:finished`), + type: "run.updated", + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: { ...run, completedAt: now }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTask(child.childThreadId, run.id); + return TurnItemId.make(`${messageId}:item`); + }); + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("steered-test"), + requestNamespace: name, + thread: { + threadId: parentThreadId, + providerSessionId: name, + providerInstanceId: modelSelection.instanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + return { + ...child, + parentThreadId, + beforePromotion, + steeredRun, + promotedRun, + promotedTaskId, + steeredTaskId, + now, + finish, + scope, + }; +}); + +it.layer(layerSteeredTaskTest)("steered delegated follow-ups", (it) => { + it.effect.each([ + { status: "completed", owner: "delegated" }, + { status: "failed", owner: "delegated" }, + { status: "completed", owner: "monitor" }, + ] as const)( + "shares the running turn outcome and delivers each task once ($status, $owner)", + ({ status, owner }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const seeded = yield* seedSteeredTasks(`steered-${status}-${owner}`, { + queueHeld: status === "failed", + monitor: owner === "monitor", + }); + yield* orchestrator.recoverDelegatedTask(seeded.childThreadId, seeded.promotedRun.id); + yield* (yield* EventSink.EventSinkV2) + .stream({ afterSequence: seeded.beforePromotion, eventType: "subagent.updated" }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === seeded.promotedTaskId && + stored.event.payload.status === "running", + ), + Stream.take(1), + Stream.runHead, + ); + const activeParent = yield* orchestrator.getThreadProjection(seeded.parentThreadId); + assert.equal( + activeParent.subagents.find((task) => task.id === seeded.promotedTaskId)?.status, + "running", + ); + const working = yield* service.taskStatus(seeded.scope, seeded.promotedTaskId); + assert.equal(working.status, "running"); + assert.equal(working.workState, "working"); + assert.isNull(working.summary); + assert.isTrue( + yield* orchestrator.delegatedTaskResultPending( + seeded.childThreadId, + seeded.promotedTaskId, + ), + ); + + const sink = yield* EventSink.EventSinkV2; + const unrelated = runEvent({ + threadId: seeded.childThreadId, + runId: RunId.make(`${status}-${owner}:unrelated`), + ordinal: 4, + status: "cancelled", + now: seeded.now, + }); + yield* sink.writeWithEffects({ + commandId: reconcileCommandId(`${status}-${owner}:unrelated`), + events: [ + owner === "monitor" + ? unrelated + : { + ...unrelated, + payload: { + ...unrelated.payload, + delegatedTaskId: NodeId.make(`${status}:unrelated`), + }, + }, + ], + effects: [ + { + id: `effect:restart-continuation:${unrelated.runId}`, + commandId: reconcileCommandId(`${status}-${owner}:unrelated`), + threadId: seeded.childThreadId, + request: { type: "provider-runtime.continue", sourceRunId: unrelated.runId }, + }, + ], + }); + yield* seeded.finish({ ...seeded.steeredRun, status }); + const settled = yield* orchestrator.getThreadProjection(seeded.parentThreadId); + const shared = settled.contextTransfers.filter( + (transfer) => transfer.sourcePoint.runId === seeded.steeredRun.id, + ); + assert.equal( + settled.subagents.find((task) => task.id === seeded.promotedTaskId)?.status, + status, + ); + assert.lengthOf(shared, 1); + yield* orchestrator.recoverDelegatedTasks; + yield* orchestrator.recoverDelegatedTasks; + const again = yield* orchestrator.getThreadProjection(seeded.parentThreadId); + assert.deepEqual(again.subagents, settled.subagents); + assert.deepEqual(again.contextTransfers, settled.contextTransfers); + assert.deepEqual(again.runs, settled.runs); + for (const id of owner === "monitor" + ? [seeded.promotedTaskId] + : [seeded.steeredTaskId, seeded.promotedTaskId]) { + const result = yield* service.taskStatus(seeded.scope, id); + assert.equal(result.status, status); + assert.equal(result.summary, "Shared result."); + assert.equal(result.resultContextTransferId, shared[0]!.id); + } + assert.isFalse( + yield* orchestrator.delegatedTaskResultPending( + seeded.childThreadId, + seeded.promotedTaskId, + ), + ); + }), + ); + + it.effect( + "holds a promoted task through two restart cuts and shares the continuation result", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const sink = yield* EventSink.EventSinkV2; + const seeded = yield* seedSteeredTasks("steered-restarts"); + let run: OrchestrationV2Run = seeded.steeredRun; + for (const ordinal of [4, 5]) { + yield* sink.writeWithEffects({ + commandId: reconcileCommandId(`steered-restarts-${ordinal}`), + events: [ + { + id: EventId.make(`${run.id}:cut`), + type: "run.updated", + threadId: seeded.childThreadId, + runId: run.id, + occurredAt: seeded.now, + payload: { ...run, status: "cancelled", completedAt: seeded.now }, + }, + ], + effects: [ + { + id: `effect:restart-continuation:${run.id}`, + commandId: reconcileCommandId(`steered-restarts-${ordinal}`), + threadId: seeded.childThreadId, + request: { type: "provider-runtime.continue", sourceRunId: run.id }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + assert.isTrue( + yield* orchestrator.delegatedTaskResultPending( + seeded.childThreadId, + seeded.promotedTaskId, + ), + ); + assert.equal( + (yield* service.taskStatus(seeded.scope, seeded.promotedTaskId)).workState, + "working", + ); + run = { + ...seeded.steeredRun, + id: RunId.make(`steered-restarts:${ordinal}`), + ordinal, + restartContinuationOfRunId: run.id, + status: "starting", + startedAt: null, + }; + yield* sink.write({ + events: [ + { + id: EventId.make(`${run.id}:continued`), + type: "run.updated", + threadId: seeded.childThreadId, + runId: run.id, + occurredAt: seeded.now, + payload: run, + }, + ], + }); + } + yield* seeded.finish({ ...run, status: "completed", startedAt: seeded.now }); + const result = yield* service.taskStatus(seeded.scope, seeded.promotedTaskId); + assert.equal(result.status, "completed"); + assert.equal(result.summary, "Shared result."); + const parent = yield* orchestrator.getThreadProjection(seeded.parentThreadId); + const transfers = parent.contextTransfers.filter( + (transfer) => transfer.sourcePoint.runId === run.id, + ); + assert.lengthOf(transfers, 1); + assert.equal(result.resultContextTransferId, transfers[0]!.id); + }), + ); + + it.effect.each(["delegated", "monitor"] as const)( + "reports an approval pending in the joined run to the promoted task (%s)", + (owner) => + Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const sink = yield* EventSink.EventSinkV2; + const seeded = yield* seedSteeredTasks(`steered-approval-${owner}`, { + monitor: owner === "monitor", + }); + const nodeId = NodeId.make(`steered-approval-${owner}:node`); + const requestId = RuntimeRequestId.make(`steered-approval-${owner}:request`); + yield* sink.write({ + events: [ + { + id: EventId.make(`${nodeId}:seed`), + type: "node.updated", + threadId: seeded.childThreadId, + runId: seeded.steeredRun.id, + nodeId, + occurredAt: seeded.now, + payload: { + id: nodeId, + threadId: seeded.childThreadId, + runId: seeded.steeredRun.id, + rootNodeId: nodeId, + parentNodeId: null, + kind: "root_turn", + status: "running", + countsForRun: true, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: seeded.now, + completedAt: null, + }, + }, + { + id: EventId.make(`${requestId}:seed`), + type: "runtime-request.updated", + threadId: seeded.childThreadId, + runId: seeded.steeredRun.id, + nodeId, + occurredAt: seeded.now, + payload: { + id: requestId, + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "command", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: ProviderSessionId.make(`steered-approval-${owner}`), + }, + createdAt: seeded.now, + resolvedAt: null, + }, + }, + ], + }); + const result = yield* service.taskStatus(seeded.scope, seeded.promotedTaskId); + assert.deepEqual(result.pendingApproval, { + requestId, + kind: "command", + runId: seeded.steeredRun.id, + }); + }), + ); + + it.effect.each(["delegated", "monitor"] as const)( + "acknowledges tasks when a result page excludes the promoted user message (%s)", + (owner) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const seeded = yield* seedSteeredTasks(`steered-page-${owner}`, { + monitor: owner === "monitor", + }); + const itemId = yield* seeded.finish({ ...seeded.steeredRun, status: "completed" }); + const settled = yield* orchestrator.getThreadProjection(seeded.parentThreadId); + assert.lengthOf( + settled.contextTransfers.filter( + (transfer) => transfer.sourcePoint.runId === seeded.steeredRun.id, + ), + 1, + ); + yield* orchestrator.recoverDelegatedTasks; + assert.deepEqual( + (yield* orchestrator.getThreadProjection(seeded.parentThreadId)).contextTransfers, + settled.contextTransfers, + ); + const page = yield* service.readThread(seeded.scope, { + threadId: seeded.childThreadId, + itemId, + limit: 1, + }); + assert.lengthOf(page.items, 1); + assert.equal(page.items[0]?.type, "assistant_message"); + const parent = yield* orchestrator.getThreadProjection(seeded.parentThreadId); + for (const id of owner === "monitor" + ? [seeded.promotedTaskId] + : [seeded.steeredTaskId, seeded.promotedTaskId]) { + assert.equal( + parent.subagents.find((task) => task.id === id)?.completionDelivery?.state, + "acknowledged", + id, + ); + } + const result = yield* service.taskStatus(seeded.scope, seeded.promotedTaskId); + assert.equal(result.summary, "Shared result."); + assert.equal( + result.resultContextTransferId, + parent.contextTransfers.find( + (transfer) => transfer.sourcePoint.runId === seeded.steeredRun.id, + )?.id, + ); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3157f55cacfa6..242d1f7fcddda 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -183,6 +183,7 @@ import { makeSubagentChildThread, subagentResultForRun, delegatedTaskProgress, + delegatedTaskResultRun, subagentThreadTitle, } from "@t3tools/provider-core/server/subagentProjection"; import { @@ -503,6 +504,29 @@ export class OrchestratorV2 extends Context.Service, + run: OrchestrationV2Run, +): ReadonlyArray { + if (run.delegatedTaskId !== undefined) { + return runs.filter((candidate) => candidate.delegatedTaskId === run.delegatedTaskId); + } + // Ownerless runs (a monitor turn a task joined) share no task id; only the + // run's own restart continuations can still change its result. + const chain = [run]; + const seen = new Set([run.id]); + for (let index = 0; index < chain.length; index += 1) { + for (const candidate of runs) { + if (candidate.restartContinuationOfRunId === chain[index]!.id && !seen.has(candidate.id)) { + seen.add(candidate.id); + chain.push(candidate); + } + } + } + return chain; +} + /** Run ordinals can skip values: a queued run moves past newer runs when it starts. */ function nextRunOrdinal(projection: Pick): number { return projection.runs.reduce((max, run) => Math.max(max, run.ordinal), 0) + 1; @@ -12930,12 +12954,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? options.task.id === forkedFrom.nodeId ? options.resultRun : undefined - : taskRuns.toSorted((a, b) => b.ordinal - a.ordinal)[0]; + : delegatedTaskResultRun(childControls, options.task.id); if (childRun === undefined) return; + const steered = taskRuns.length > 0 && childRun.delegatedTaskId !== options.task.id; const outcome = runOutcome(childRun); const terminalStatus = delegatedTaskTerminalStatus(outcome); if (terminalStatus === null) { - if (childRun.delegatedTaskId === undefined || childRun.id !== options.syncRunId) return; + if ( + (childRun.delegatedTaskId === undefined || childRun.id !== options.syncRunId) && + !(steered && taskRuns.some((run) => run.id === options.syncRunId)) + ) + return; const task = options.task; const status: OrchestrationV2Subagent["status"] = childRun.status === "queued" @@ -12992,7 +13021,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } if ( yield* childAwaitsRestartContinuation( - taskRuns.length === 0 ? childControls.runs : taskRuns, + taskRuns.length === 0 ? childControls.runs : runsOfTask(childControls.runs, childRun), childRun, options?.settledContinuationOf, ) @@ -13047,10 +13076,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (transfer.sourcePoint.runId === childRun.id || (options.task.id === forkedFrom.nodeId && transfer.sourcePoint.runId === undefined)), ); - if (existingResultTransfer !== undefined) { + // A promoted task may share a transfer before its own result is settled. + if (steered ? task.result !== null : existingResultTransfer !== undefined) { return; } + const sharesResultTransfer = + steered && (childRun.delegatedTaskId !== undefined || existingResultTransfer !== undefined); const now = yield* DateTime.now; const result = subagentResultForRun(childProjection, { ...childRun, status: outcome }); const parentRun = @@ -13157,14 +13189,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, }, ]), - { - type: "context-transfer.created", - threadId: parentThreadId, - ...(parentRun === undefined ? {} : { runId: parentRun.id }), - providerInstanceId: childRun.providerInstanceId, - occurredAt: now, - payload: resultTransfer, - }, + ...(sharesResultTransfer + ? [] + : [ + { + type: "context-transfer.created" as const, + threadId: parentThreadId, + ...(parentRun === undefined ? {} : { runId: parentRun.id }), + providerInstanceId: childRun.providerInstanceId, + occurredAt: now, + payload: resultTransfer, + }, + ]), ]; yield* writeSystemEvents(completionEvents, [], { updateDelegatedCompletionOnly: true }); @@ -14443,10 +14479,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio { messageRoles: ["user"] }, ); if (taskId !== undefined) { - const runs = child.runs.filter((run) => run.delegatedTaskId === taskId); - const run = runs.toSorted((a, b) => b.ordinal - a.ordinal)[0]; + const run = delegatedTaskResultRun(child, taskId); if (run === undefined || delegatedTaskTerminalStatus(run.status) === null) return true; - return yield* childAwaitsRestartContinuation(runs, run); + return yield* childAwaitsRestartContinuation(runsOfTask(child.runs, run), run); } const progress = delegatedTaskProgress(child); if (progress.state !== "result_available") return true; diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 62145567f6651..a06310e3cfba6 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -248,6 +248,31 @@ export function subagentResultForRun( }; } +/** Promotion cancels the queued run; its moved message identifies the turn that now owns the work. */ +export function delegatedTaskResultRun( + child: { + readonly runs: OrchestrationV2ThreadProjection["runs"]; + readonly messages: ReadonlyArray>; + }, + taskId: NodeId, +): OrchestrationV2Run | undefined { + const run = child.runs + .filter((candidate) => candidate.delegatedTaskId === taskId) + .toSorted((a, b) => b.ordinal - a.ordinal)[0]; + if (run?.status !== "cancelled") return run; + const steeredRunId = child.messages.find((message) => message.id === run.userMessageId)?.runId; + let resolved = child.runs.find( + (candidate) => candidate.id === steeredRunId && candidate.id !== run.id, + ); + if (resolved === undefined) return run; + const continuationOf = (from: OrchestrationV2Run) => + child.runs.find((candidate) => candidate.restartContinuationOfRunId === from.id); + for (let next = continuationOf(resolved); next !== undefined; next = continuationOf(next)) { + resolved = next; + } + return resolved; +} + export function delegatedTaskProgress(projection: { readonly runs: OrchestrationV2ThreadProjection["runs"]; readonly messages: ReadonlyArray<