From 3dda05ac50609a2743b69512f1a8f97097b3ae4d Mon Sep 17 00:00:00 2001 From: ScottN-PV Date: Mon, 19 Oct 2026 14:13:07 -0400 Subject: [PATCH 1/2] fix(server): keep follow-up tasks pending until nested work finishes --- .../src/mcp/OrchestratorMcpService.test.ts | 50 +++- apps/server/src/mcp/OrchestratorMcpService.ts | 15 +- .../DelegatedCompletionDelivery.test.ts | 282 +++++++++++++++++- .../src/orchestration-v2/Orchestrator.ts | 40 ++- 4 files changed, 362 insertions(+), 25 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 015cf1f576964..69e1a81785266 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -62,8 +62,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"); @@ -82,7 +87,7 @@ describe("OrchestratorMcpService", () => { childThreadId, driver: "codex", model: "gpt-5.6-terra", - result: "terminal result", + result: failedBeforeStart || queued ? null : "terminal result", completionDelivery: { state: "pending" }, }, ], @@ -92,9 +97,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"), }, { @@ -107,14 +116,16 @@ describe("OrchestratorMcpService", () => { ], contextTransfers: [], messages: [], + turnItems: [], subagents: [], providerThreads: [], runtimeRequests: [], } as unknown as OrchestrationV2ThreadProjection; - let hasNestedWork = true; + let hasNestedWork = !failedBeforeStart && !queued; const layerDependencies = Layer.mergeAll( NodeServices.layer, Layer.mock(ThreadManagementService.ThreadManagementService)({ + delegatedTaskResultPending: () => Effect.succeed(false), getThreadRecords: (threadId) => Effect.succeed( threadId === parentThreadId @@ -132,7 +143,11 @@ describe("OrchestratorMcpService", () => { ? { ...childProjection, subagents: [ - { ...parentProjection.subagents[0]!, status: "running" as const }, + { + ...parentProjection.subagents[0]!, + runId: childRunId, + status: "running" as const, + }, ], } : childProjection, @@ -144,7 +159,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), ), @@ -174,16 +189,29 @@ 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.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.equal(yield* Ref.get(acknowledgementAttempts), 0); + hasNestedWork = false; + } const error = yield* service.taskStatus(scope, taskId).pipe(Effect.flip); assert.equal(error.code, "orchestration_error"); 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 f1ac8c5703815..56ef09f99976d 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -89,7 +89,7 @@ import { DispatchModeLimit, type DispatchModeRefusal, } from "../orchestration-v2/DispatchModeLimit.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"; @@ -1449,11 +1449,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 e58c8a1e528f8..744f3776023b6 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -14,6 +14,8 @@ import { ProviderDriverKind, ProviderInstanceId, ProviderThreadId, + ProviderSessionId, + ProviderTurnId, RunId, ThreadId, TurnItemId, @@ -37,6 +39,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"; @@ -76,7 +79,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, @@ -1407,10 +1448,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", @@ -1490,6 +1534,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) @@ -1509,6 +1557,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({ @@ -1576,24 +1643,214 @@ 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({ 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) => @@ -1608,9 +1865,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, @@ -1636,7 +1898,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 8c8620088fa94..2a64e81eb4453 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -96,6 +96,7 @@ import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts"; import { notificationTurnItem } from "@t3tools/provider-core/server/notification"; import { cancelledBeforeStart, + neverStarted, isRestartContinuationSource, runOutcome, } from "./RestartBackgroundNote.ts"; @@ -4923,7 +4924,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; @@ -5103,6 +5104,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 under the thread lock, using the live // session's capabilities. Never interrupt/restart a turn for a notification. if ( @@ -5123,6 +5135,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)) @@ -10681,6 +10694,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"], @@ -12027,6 +12054,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); From 9df1a246edb4dd4537b5e1b30d2e0b8860f111a2 Mon Sep 17 00:00:00 2001 From: ScottN-PV Date: Fri, 30 Oct 2026 13:11:43 -0400 Subject: [PATCH 2/2] test(server): keep the competing run's finish out of queue promotion --- .../src/orchestration-v2/DelegatedCompletionDelivery.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 90558e43cf1bd..0279aa11a40b0 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -2087,7 +2087,10 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( (task) => task.id === run.delegatedTaskId, )!.result, ); - yield* eventSink.write({ events: [finishRun(child.childThreadId, competingRun)] }); + yield* eventSink.write({ + commandId: reconcileCommandId("competing-followup-finish"), + events: [finishRun(child.childThreadId, competingRun)], + }); } yield* eventSink.write({ commandId: reconcileCommandId(`nested-continuation-finish-${outcome}`),