diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 3bd9439d308b4..b28b576853095 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -78,8 +78,13 @@ const layerOrchestratorService = OrchestratorMcpService.layer.pipe( ); describe("OrchestratorMcpService", () => { - it.effect("retries terminal acknowledgement with a fresh command id", () => + const acknowledgementTest = + "retries terminal acknowledgement with a fresh command id (follow-up: %s)"; + const cases = [false, true, "never-started", "queued"] as const; + it.effect.each(cases)(acknowledgementTest, (followup) => Effect.gen(function* () { + const failedBeforeStart = followup === "never-started"; + const queued = followup === "queued"; const parentThreadId = ThreadId.make("thread:mcp-ack-parent"); const childThreadId = ThreadId.make("thread:mcp-ack-child"); const childRunId = RunId.make("run:mcp-ack-child"); @@ -107,7 +112,7 @@ describe("OrchestratorMcpService", () => { childThreadId, driver: "codex", model: "gpt-5.6-terra", - result: "terminal result", + result: failedBeforeStart || queued ? null : "terminal result", completionDelivery: { state: "pending" }, }, ], @@ -117,9 +122,13 @@ describe("OrchestratorMcpService", () => { runs: [ { id: childRunId, + ...(followup ? { delegatedTaskId: taskId } : {}), ordinal: 1, status: "completed", startedAt: DateTime.makeUnsafe("2026-10-03T10:00:00Z"), + ...(failedBeforeStart || queued + ? { ordinal: 3, status: failedBeforeStart ? "failed" : "queued", startedAt: null } + : {}), completedAt: DateTime.makeUnsafe("2026-10-03T10:10:00Z"), }, { @@ -146,7 +155,7 @@ describe("OrchestratorMcpService", () => { providerThreads: [], runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; - let hasNestedWork = true; + let hasNestedWork = !failedBeforeStart && !queued; let settleAfterTaskRead = false; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -159,7 +168,11 @@ describe("OrchestratorMcpService", () => { ? { ...childProjection, subagents: [ - { ...parentProjection.subagents[0]!, status: "running" as const }, + { + ...parentProjection.subagents[0]!, + runId: childRunId, + status: "running" as const, + }, ], } : childProjection; @@ -189,7 +202,7 @@ describe("OrchestratorMcpService", () => { ]).pipe( Effect.andThen(Ref.updateAndGet(acknowledgementAttempts, (count) => count + 1)), Effect.flatMap((attempt) => - attempt === 1 + attempt === 1 && !hasNestedWork ? Effect.fail(new Error("simulated acknowledgement failure") as never) : Effect.succeed({} as never), ), @@ -220,20 +233,35 @@ describe("OrchestratorMcpService", () => { yield* Effect.gen(function* () { const service = yield* OrchestratorMcpService.OrchestratorMcpService; - const pending = yield* service.taskStatus(scope, taskId); - assert.equal(pending.status, "running"); - assert.equal(pending.workState, "waiting_for_children"); - assert.isNull(pending.summary); - assert.isNull(pending.resultContextTransferId); - assert.equal(yield* Ref.get(acknowledgementAttempts), 0); - hasNestedWork = false; + if (queued) { + const pending = yield* service.taskStatus(scope, taskId); + assert.equal(pending.status, "queued"); + assert.equal(pending.workState, "working"); + return; + } + if (!failedBeforeStart) { + const pending = yield* service.taskStatus(scope, taskId); + assert.equal(pending.status, "running"); + assert.equal(pending.workState, "waiting_for_children"); + assert.isNull(pending.summary); + assert.isNull(pending.resultContextTransferId); + assert.equal(yield* Ref.get(acknowledgementAttempts), 0); + hasNestedWork = false; + } const error = yield* service.taskStatus(scope, taskId).pipe(Effect.flip); assert.equal(error.code, "orchestration_error"); - // Completion lands after task selection, before the remaining parent/child reads. - hasNestedWork = true; - settleAfterTaskRead = true; + if (!failedBeforeStart) { + // Completion lands after task selection, before the remaining parent/child reads. + hasNestedWork = true; + settleAfterTaskRead = true; + } const result = yield* service.taskStatus(scope, taskId); + if (failedBeforeStart) { + assert.equal(result.status, "failed"); + assert.equal(result.workState, "result_available"); + return; + } assert.equal(result.status, "completed"); assert.equal(result.summary, "terminal result"); assert.equal(result.latestTerminalRunId, childRunId); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index b937453f39b22..2e87e7c716913 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -96,7 +96,7 @@ import type { ProjectionRecordField, ProjectionRecordFilter, } from "../orchestration-v2/ProjectionStore.ts"; -import { runOutcome } from "../orchestration-v2/RestartBackgroundNote.ts"; +import { neverStarted, runOutcome } from "../orchestration-v2/RestartBackgroundNote.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; @@ -1518,11 +1518,20 @@ const make = Effect.gen(function* () { .filter((run) => run.delegatedTaskId === task.id) .toSorted((a, b) => b.ordinal - a.ordinal)[0] : progress.resultRun; + const taskRuns = childControls.runs.filter((run) => run.delegatedTaskId === task.id); const taskWorkState = childRun?.delegatedTaskId === task.id - ? isTerminalTaskStatus(taskStatusForRun(taskResultRun)) + ? taskResultRun !== undefined && + isTerminalTaskStatus(taskStatusForRun(taskResultRun)) && + taskRuns.every(neverStarted) ? "result_available" - : "working" + : delegatedTaskProgress({ + ...childControls, + runs: taskRuns, + subagents: childControls.subagents.filter((subagent) => + taskRuns.some((run) => run.id === subagent.runId), + ), + }).state : progress.state; const resultRunIds = [ ...new Set( diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index cabc322d31331..84accc9faa0b2 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -15,6 +15,8 @@ import { ProviderDriverKind, ProviderInstanceId, ProviderThreadId, + ProviderSessionId, + ProviderTurnId, RunId, RuntimeRequestId, ThreadId, @@ -42,6 +44,7 @@ import * as VcsProcess from "../vcs/VcsProcess.ts"; import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; +import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { continueRestartedRun } from "./RestartContinuation.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; @@ -82,7 +85,45 @@ const orchestrationAdapter = { driver, getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), - openSession: () => Effect.die("sessions are not used by delegated completion tests"), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const unused = () => Effect.die("unused session operation"); + return { + instanceId: modelSelection.instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: modelSelection.instanceId, + status: "ready", + cwd: process.cwd(), + model: modelSelection.model, + capabilities: { + ...CodexProviderCapabilitiesV2, + turns: { + ...CodexProviderCapabilitiesV2.turns, + supportsActiveSteering: true, + activeSteeringInterruptsTools: false, + }, + }, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.never, + ensureThread: unused, + resumeThread: unused, + startTurn: unused, + steerTurn: unused, + interruptTurn: unused, + respondToRuntimeRequest: unused, + readThreadSnapshot: unused, + rollbackThread: unused, + forkThread: unused, + }; + }), } as ProviderAdapter.ProviderAdapterV2["Service"]; const providerInstance = { instanceId: modelSelection.instanceId, @@ -1817,10 +1858,13 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( const outcomes = [ "completed", "failed", + "never-started", "interrupted", "restarted", "stopped", "overlapping", + "nested", + "nested-active", ] as const; it.effect.each(outcomes)( "delivers a %s follow-up independently of the original task", @@ -1900,6 +1944,10 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId); let run = childProjection.runs.find((row) => row.userMessageId === messageId)!; assert.isDefined(run.delegatedTaskId); + const resultPending = orchestrator.delegatedTaskResultPending( + child.childThreadId, + run.delegatedTaskId, + ); const workingParent = yield* orchestrator.getShellSnapshot(); const backgroundTask = workingParent.threads .find((thread) => thread.id === parentThreadId) @@ -1919,6 +1967,25 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( dispatchMode: { type: "queue_after_active" }, }); } + if (outcome === "nested" || outcome === "nested-active") { + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make(`command:nested-delegate-${outcome}`), + parentThreadId: child.childThreadId, + parentRunId: run.id, + parentNodeId: run.rootNodeId!, + task: "Check the follow-up.", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + }); + } + const nested = (yield* orchestrator.getThreadProjection( + child.childThreadId, + )).subagents.find((task) => task.runId === run.id); if (outcome === "restarted") { const cut = { ...run, status: "cancelled" as const, startedAt: now, completedAt: now }; yield* eventSink.writeWithEffects({ @@ -1986,24 +2053,217 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( payload: { ...run, status: - outcome === "stopped" - ? "interrupted" - : outcome === "restarted" || outcome === "overlapping" - ? "completed" - : outcome, - startedAt: now, + outcome === "never-started" + ? "queued" + : outcome === "stopped" + ? "interrupted" + : outcome === "restarted" || + outcome === "overlapping" || + outcome === "nested" || + outcome === "nested-active" + ? "completed" + : outcome, + startedAt: outcome === "never-started" ? null : now, + ...(outcome === "never-started" ? { activeAttemptId: null } : {}), completedAt: now, }, }, ], }); + if (outcome === "never-started") { + yield* orchestrator.dispatch({ + type: "queue.resume", + threadId: child.childThreadId, + commandId: CommandId.make("command:followup-start-fails"), + }); + const failed = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (row) => row.id === run.id, + )!; + assert.equal(failed.status, "failed"); + assert.isNull(failed.startedAt); + assert.isAbove(failed.ordinal, 1); + } yield* orchestrator.recoverDelegatedTasks; + if (nested !== undefined) { + assert.isTrue(yield* resultPending); + const finishRun = (threadId: ThreadId, finishedRun: OrchestrationV2Run) => ({ + id: EventId.make(`event:finish:${finishedRun.id}`), + type: "run.updated" as const, + threadId, + runId: finishedRun.id, + occurredAt: now, + payload: { + ...finishedRun, + status: "completed" as const, + startedAt: now, + completedAt: now, + }, + }); + let competingRun: OrchestrationV2Run | undefined; + if (outcome === "nested-active") { + const competingMessageId = MessageId.make("message:competing-followup"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:competing-followup"), + threadId: child.childThreadId, + senderThreadId: parentThreadId, + messageId: competingMessageId, + text: "Work on another follow-up.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + const projection = yield* orchestrator.getThreadProjection(child.childThreadId); + competingRun = projection.runs.find( + (row) => row.userMessageId === competingMessageId, + )!; + const providerSessionId = ProviderSessionId.make("session:competing-followup"); + const sessions = yield* ProviderSessionManagerV2; + yield* sessions.open({ + threadId: child.childThreadId, + providerSessionId, + modelSelection, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }, + }); + const providerThread = projection.providerThreads[0]!; + competingRun = { + ...competingRun, + status: "running", + startedAt: now, + }; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:competing-provider-thread"), + type: "provider-thread.updated", + threadId: child.childThreadId, + occurredAt: now, + payload: { ...providerThread, providerSessionId }, + }, + { + ...runEvent({ + threadId: child.childThreadId, + runId: competingRun.id, + ordinal: competingRun.ordinal, + status: "running", + now, + }), + payload: competingRun, + }, + { + id: EventId.make("event:competing-provider-turn"), + type: "provider-turn.updated", + threadId: child.childThreadId, + runId: competingRun.id, + occurredAt: now, + payload: { + id: ProviderTurnId.make("turn:competing"), + providerThreadId: providerThread.id, + nodeId: competingRun.rootNodeId!, + runAttemptId: competingRun.activeAttemptId!, + nativeTurnRef: { driver, nativeId: "competing", strength: "strong" }, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + ], + }); + } + const nestedRun = (yield* orchestrator.getThreadProjection(nested.childThreadId!)) + .runs[0]!; + yield* eventSink.write({ + commandId: reconcileCommandId(`nested-finish-${outcome}`), + events: [finishRun(nested.childThreadId!, nestedRun)], + }); + yield* orchestrator.recoverDelegatedTasks; + let nestedWake = yield* requests.take; + while (nestedWake.threadId !== child.childThreadId) nestedWake = yield* requests.take; + const delivery = nestedWake.delegatedCompletion!; + const wakeMessageId = delivery.messageId; + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make(`command:nested-wake-${outcome}`), + threadId: child.childThreadId, + messageId: wakeMessageId, + text: "Nested checker finished.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { + parentRunId: run.id, + generation: delivery.generation, + taskIds: [nested.id], + }, + }); + const afterWake = yield* orchestrator.getThreadProjection(child.childThreadId); + const continuation = afterWake.runs.find((row) => row.userMessageId === wakeMessageId)!; + assert.isDefined(continuation); + assert.equal(continuation.delegatedTaskId, run.delegatedTaskId); + if (competingRun !== undefined) { + assert.deepEqual( + afterWake.runs.find((row) => row.id === competingRun.id), + competingRun, + ); + yield* orchestrator.recoverDelegatedTasks; + assert.isNull( + (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === run.delegatedTaskId, + )!.result, + ); + yield* eventSink.write({ + commandId: reconcileCommandId("competing-followup-finish"), + events: [finishRun(child.childThreadId, competingRun)], + }); + } + yield* eventSink.write({ + commandId: reconcileCommandId(`nested-continuation-finish-${outcome}`), + events: [ + finishRun(child.childThreadId, continuation), + { + id: EventId.make(`event:nested-final-message-${outcome}`), + type: "message.updated", + threadId: child.childThreadId, + runId: continuation.id, + occurredAt: now, + payload: { + id: MessageId.make(`message:nested-final-${outcome}`), + threadId: child.childThreadId, + runId: continuation.id, + nodeId: continuation.rootNodeId, + role: "assistant", + text: "FINAL", + createdBy: "agent", + creationSource: "provider", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + run = continuation; + yield* orchestrator.recoverDelegatedTasks; + yield* orchestrator.recoverDelegatedTasks; + } const finished = yield* orchestrator.getThreadProjection(parentThreadId); assert.deepEqual( finished.subagents.find((task) => task.id === child.taskId), original, ); const followup = finished.subagents.find((task) => task.id === run.delegatedTaskId)!; + if (nested !== undefined) assert.equal(followup.result, "FINAL"); + if (outcome === "overlapping" || outcome === "never-started") { + assert.isFalse(yield* resultPending); + } if (outcome === "overlapping") { const later = finished.subagents.find( (task) => @@ -2018,9 +2278,14 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( followup.status, outcome === "stopped" ? "interrupted" - : outcome === "restarted" || outcome === "overlapping" + : outcome === "restarted" || + outcome === "overlapping" || + outcome === "nested" || + outcome === "nested-active" ? "completed" - : outcome, + : outcome === "never-started" + ? "failed" + : outcome, ); assert.equal( followup.completionDelivery?.state, @@ -2046,7 +2311,7 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( transfer.type === "subagent_result" && transfer.sourceThreadId === child.childThreadId, ).length, - 2, + outcome === "nested-active" ? 3 : 2, ); yield* orchestrator.recoverDelegatedTasks; const recovered = yield* orchestrator.getThreadProjection(parentThreadId); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 600764704e57f..5df533c538bf5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -122,6 +122,7 @@ import { cancelledBeforeStart, continuationFailedBeforePrompt, continuationStartRetryDelayMs, + neverStarted, isRestartContinuationSource, runOutcome, } from "./RestartBackgroundNote.ts"; @@ -5513,7 +5514,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio projection = applyToProjection(projection, event); } - const continuedTaskId = projection.runs.find( + let continuedTaskId = projection.runs.find( (run) => run.id === command.restartContinuationOfRunId, )?.delegatedTaskId; @@ -5732,6 +5733,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio taskIds: delivery.taskIds, }; } + if (delegatedCompletion !== undefined && projection.thread.lineage.parentThreadId !== null) { + const source = projection.runs.find((run) => run.id === delegatedCompletion.parentRunId); + const parent = yield* projectionStore + .getThreadRecords(projection.thread.lineage.parentThreadId, ["subagents"]) + .pipe(mapDispatchError(command)); + const task = parent.subagents.find((task) => task.id === source?.delegatedTaskId); + // Nested results continue the pending follow-up rather than creating another obligation. + if (task?.result === null && task.completionDelivery?.state !== "disposed") { + continuedTaskId = task.id; + } + } // Route durable mailbox deliveries and delegated task questions under the // thread lock, using the live session's capabilities. A question queued // behind a long parent turn is never seen. Never interrupt/restart a turn @@ -5756,6 +5768,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); if ( active !== undefined && + (continuedTaskId === undefined || active.delegatedTaskId === continuedTaskId) && activeTurn !== undefined && providerThread?.providerSessionId != null && (activeMessage === undefined || !isNativeMaintenanceCommand(activeMessage)) @@ -12472,6 +12485,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return; } + // Each follow-up waits for its own nested work, independently of later queued follow-ups. + if ( + taskRuns.length > 0 && + !taskRuns.every(neverStarted) && + delegatedTaskProgress({ + ...childControls, + runs: taskRuns, + subagents: childControls.subagents.filter((task) => + taskRuns.some((run) => run.id === task.runId), + ), + }).state !== "result_available" + ) + return; + const childResult = yield* projectionStore.getThreadRecords( childThreadId, ["messages", "turnItems"], @@ -13972,6 +13999,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const runs = child.runs.filter((run) => run.delegatedTaskId === taskId); const run = runs.toSorted((a, b) => b.ordinal - a.ordinal)[0]; if (run === undefined || delegatedTaskTerminalStatus(run.status) === null) return true; + if (runs.every(neverStarted)) return false; + if ( + delegatedTaskProgress({ + ...child, + runs, + subagents: child.subagents.filter((task) => + runs.some((taskRun) => taskRun.id === task.runId), + ), + }).state !== "result_available" + ) + return true; return yield* childAwaitsRestartContinuation(runs, run); } const progress = delegatedTaskProgress(child);