From 8b142db54562f5d8bf3ed6277d15cf53d4fbc1ce Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 15:34:52 +0000 Subject: [PATCH 1/8] test: reproduce #30911, stopping a queued follow-up task kills another task's turn Cancelling a delegated follow-up task whose only child run is queued, via task_cancel or the Agents panel stop, still stops the whole child thread and interrupts the running turn that belongs to an earlier task. Co-Authored-By: Claude Opus 5.5 --- ...OrchestratorMcpToolkit.integration.test.ts | 499 ++++++++++-------- .../src/operations/commands.test.ts | 96 +++- 2 files changed, 366 insertions(+), 229 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 10a6f3c30c602..66e0363dd607d 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -193,6 +193,239 @@ const layerUnusedScheduledTaskStub = Layer.succeed( }), ); +const makeQueuedFollowupScenario = Effect.fn("makeQueuedFollowupScenario")(function* ( + result: string, +) { + const cwd = yield* checkpointWorkspace("delegated-task-status"); + const rawTranscript = yield* readDelegatedTaskStatusTranscript(result); + const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript( + materializeReplayTranscriptWorkspace(rawTranscript, cwd), + ); + const layerOrchestrator = ProviderReplayHarness.layerProviderReplay( + { + name: "delegated-task-status/codex", + transcript, + commands: [], + runtimePolicyOverride: { cwd }, + }, + CodexOrchestratorReplayHarness, + ); + const layerOrchestration = Layer.merge( + layerOrchestrator, + ThreadManagementService.layer.pipe( + Layer.provide(layerOrchestrator), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory))), + ), + ); + const layerProviderRegistry = ProviderRegistryMock.layer([ + makeProviderSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: codexModel, + }), + ]); + const layerTest = layerOrchestratorToolkit.pipe( + Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), + Layer.provideMerge(layerOrchestration), + Layer.provide( + CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript).pipe( + Layer.provide(McpProviderSessions.layer), + ), + ), + Layer.provide(layerProviderRegistry), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerUnusedScheduledTaskStub), + Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), + Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), + Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ProjectService.ProjectService)({})), + Layer.provideMerge( + SecretRequests.layer.pipe( + Layer.provide(layerMemorySecretStore), + Layer.provide(layerOrchestration), + ), + ), + Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), + Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ); + + const start = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const server = yield* McpServer.McpServer; + const parentCreate = yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:mcp-replay-parent:create"), + threadId: parentThreadId, + projectId, + title: "MCP replay parent", + modelSelection: codexSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:mcp-replay-parent:start"), + threadId: parentThreadId, + messageId: MessageId.make("message:mcp-replay-parent:start"), + text: parentPrompt, + attachments: [], + modelSelection: codexSelection, + dispatchMode: { type: "start_immediately" }, + }); + yield* orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: parentCreate.sequence, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "provider-turn.updated" && + stored.event.payload.status === "running", + ), + Stream.runHead, + Effect.flatMap( + Option.match({ + onNone: () => Effect.die("Parent provider turn did not start."), + onSome: () => Effect.void, + }), + ), + ); + + const invocation: McpInvocationContext.McpInvocationScope = { + environmentId: EnvironmentId.make("environment:mcp-replay"), + requestNamespace: "mcp-provider-session-replay-parent", + thread: { + threadId: parentThreadId, + providerSessionId: "mcp-provider-session-replay-parent", + providerInstanceId: codexInstanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + const invoke = (name: string, args: Record, scope = invocation) => + server + .callTool({ name, arguments: args }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + ); + + const delegationStartSequence = yield* orchestrator.getThreadEventSequence(parentThreadId); + const delegatedCall = yield* invoke("delegate_task", { + task: delegatedPrompt, + target: { + providerInstanceId: codexInstanceId, + model: codexModel, + }, + mode: "async", + clientRequestId: "delegate-codex-replay-1", + }); + const delegatedStart = yield* decodeDelegateTaskResult(delegatedCall.structuredContent).pipe( + Effect.orDie, + ); + expect(delegatedStart.childRunId).not.toBeNull(); + yield* orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: delegationStartSequence, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "context-transfer.created" && + stored.event.payload.type === "subagent_result" && + stored.event.payload.sourceThreadId === delegatedStart.childThreadId && + stored.event.payload.sourcePoint.runId === delegatedStart.childRunId, + ), + Stream.runHead, + Effect.flatMap( + Option.match({ + onNone: () => Effect.die("Delegated result transfer was not created."), + onSome: () => Effect.void, + }), + ), + ); + const delegatedStatusCall = yield* invoke("task_status", { + taskId: delegatedStart.taskId, + }); + const delegated = yield* decodeDelegateTaskResult(delegatedStatusCall.structuredContent).pipe( + Effect.orDie, + ); + expect(delegated).toMatchObject({ + status: "completed", + summary: delegatedResult, + providerInstanceId: codexInstanceId, + hasPendingChildRuns: false, + latestTerminalRunId: delegated.childRunId, + latestTerminalStatus: "completed", + latestTerminalSummary: null, + latestTerminalResultContextTransferId: null, + }); + expect(delegated.resultContextTransferId).not.toBeNull(); + + const delegatedChild = yield* orchestrator.getThreadProjection( + localChild(delegated.childThreadId), + ); + expect(delegatedChild.thread.lineage.relationshipToParent).toBe("subagent"); + expect(isProviderNativeSubagentThread(delegatedChild.thread)).toBe(false); + const followupStartSequence = yield* orchestrator.getThreadEventSequence( + localChild(delegated.childThreadId), + ); + const runningFollowupCall = yield* invoke("t3_thread_send", { + threadId: localChild(delegated.childThreadId), + message: cancellationPrompt, + clientRequestId: "delegated-child-replay-running-1", + }); + const runningFollowup = yield* decodeThreadSendResult( + runningFollowupCall.structuredContent, + ).pipe(Effect.orDie); + yield* orchestrator + .streamStoredEventsFrom({ + threadId: localChild(delegated.childThreadId), + afterSequence: followupStartSequence, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "provider-turn.updated" && + stored.event.payload.status === "running", + ), + Stream.runHead, + Effect.flatMap( + Option.match({ + onNone: () => Effect.die("Follow-up provider turn did not start."), + onSome: () => Effect.void, + }), + ), + ); + + const queuedFollowupCall = yield* invoke("t3_thread_send", { + threadId: localChild(delegated.childThreadId), + message: queuedFollowupPrompt, + mode: "queue", + clientRequestId: "delegated-child-replay-queued-1", + }); + const queuedFollowup = yield* decodeThreadSendResult(queuedFollowupCall.structuredContent).pipe( + Effect.orDie, + ); + expect(queuedFollowup).toMatchObject({ + status: "queued", + delivery: "queued", + }); + return { orchestrator, invoke, invocation, delegated, runningFollowup, queuedFollowup }; + }); + return { layer: layerTest, start }; +}); + describe("orchestrator MCP toolkit", () => { it.live.each([ { label: "distinct", result: queuedFollowupResult, latestSummary: queuedFollowupResult }, @@ -201,234 +434,10 @@ describe("orchestrator MCP toolkit", () => { ])("reports $label child results", ({ result, latestSummary }) => Effect.scoped( Effect.gen(function* () { - const cwd = yield* checkpointWorkspace("delegated-task-status"); - const rawTranscript = yield* readDelegatedTaskStatusTranscript(result); - const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript( - materializeReplayTranscriptWorkspace(rawTranscript, cwd), - ); - const layerOrchestrator = ProviderReplayHarness.layerProviderReplay( - { - name: "delegated-task-status/codex", - transcript, - commands: [], - runtimePolicyOverride: { cwd }, - }, - CodexOrchestratorReplayHarness, - ); - const layerOrchestration = Layer.merge( - layerOrchestrator, - ThreadManagementService.layer.pipe( - Layer.provide(layerOrchestrator), - Layer.provide(ServerSettings.layerTest()), - Layer.provide( - CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), - ), - ), - ); - const layerProviderRegistry = ProviderRegistryMock.layer([ - makeProviderSnapshot({ - instanceId: codexInstanceId, - driver: ProviderDriverKind.make("codex"), - model: codexModel, - }), - ]); - const layerTest = layerOrchestratorToolkit.pipe( - Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), - Layer.provideMerge(layerOrchestration), - Layer.provide( - CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript).pipe( - Layer.provide(McpProviderSessions.layer), - ), - ), - Layer.provide(layerProviderRegistry), - Layer.provide(ServerSettings.layerTest()), - Layer.provide(layerUnusedScheduledTaskStub), - Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), - Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), - Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), - Layer.provide(Layer.mock(ProjectService.ProjectService)({})), - Layer.provideMerge( - SecretRequests.layer.pipe( - Layer.provide(layerMemorySecretStore), - Layer.provide(layerOrchestration), - ), - ), - Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), - Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), - ); - + const scenario = yield* makeQueuedFollowupScenario(result); yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const server = yield* McpServer.McpServer; - const parentCreate = yield* orchestrator.dispatch({ - type: "thread.create", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("command:mcp-replay-parent:create"), - threadId: parentThreadId, - projectId, - title: "MCP replay parent", - modelSelection: codexSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: cwd, - }); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("command:mcp-replay-parent:start"), - threadId: parentThreadId, - messageId: MessageId.make("message:mcp-replay-parent:start"), - text: parentPrompt, - attachments: [], - modelSelection: codexSelection, - dispatchMode: { type: "start_immediately" }, - }); - yield* orchestrator - .streamStoredEventsFrom({ - threadId: parentThreadId, - afterSequence: parentCreate.sequence, - }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "provider-turn.updated" && - stored.event.payload.status === "running", - ), - Stream.runHead, - Effect.flatMap( - Option.match({ - onNone: () => Effect.die("Parent provider turn did not start."), - onSome: () => Effect.void, - }), - ), - ); - - const invocation: McpInvocationContext.McpInvocationScope = { - environmentId: EnvironmentId.make("environment:mcp-replay"), - requestNamespace: "mcp-provider-session-replay-parent", - thread: { - threadId: parentThreadId, - providerSessionId: "mcp-provider-session-replay-parent", - providerInstanceId: codexInstanceId, - }, - client: undefined, - capabilities: new Set(["orchestration"]), - issuedAt: 1, - }; - const invoke = (name: string, args: Record, scope = invocation) => - server - .callTool({ name, arguments: args }) - .pipe( - Effect.provideService(McpInvocationContext.McpInvocationContext, scope), - Effect.provideService(McpSchema.McpServerClient, client), - ); - - const delegationStartSequence = - yield* orchestrator.getThreadEventSequence(parentThreadId); - const delegatedCall = yield* invoke("delegate_task", { - task: delegatedPrompt, - target: { - providerInstanceId: codexInstanceId, - model: codexModel, - }, - mode: "async", - clientRequestId: "delegate-codex-replay-1", - }); - const delegatedStart = yield* decodeDelegateTaskResult( - delegatedCall.structuredContent, - ).pipe(Effect.orDie); - expect(delegatedStart.childRunId).not.toBeNull(); - yield* orchestrator - .streamStoredEventsFrom({ - threadId: parentThreadId, - afterSequence: delegationStartSequence, - }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "context-transfer.created" && - stored.event.payload.type === "subagent_result" && - stored.event.payload.sourceThreadId === delegatedStart.childThreadId && - stored.event.payload.sourcePoint.runId === delegatedStart.childRunId, - ), - Stream.runHead, - Effect.flatMap( - Option.match({ - onNone: () => Effect.die("Delegated result transfer was not created."), - onSome: () => Effect.void, - }), - ), - ); - const delegatedStatusCall = yield* invoke("task_status", { - taskId: delegatedStart.taskId, - }); - const delegated = yield* decodeDelegateTaskResult( - delegatedStatusCall.structuredContent, - ).pipe(Effect.orDie); - expect(delegated).toMatchObject({ - status: "completed", - summary: delegatedResult, - providerInstanceId: codexInstanceId, - hasPendingChildRuns: false, - latestTerminalRunId: delegated.childRunId, - latestTerminalStatus: "completed", - latestTerminalSummary: null, - latestTerminalResultContextTransferId: null, - }); - expect(delegated.resultContextTransferId).not.toBeNull(); - - const delegatedChild = yield* orchestrator.getThreadProjection( - localChild(delegated.childThreadId), - ); - expect(delegatedChild.thread.lineage.relationshipToParent).toBe("subagent"); - expect(isProviderNativeSubagentThread(delegatedChild.thread)).toBe(false); - const followupStartSequence = yield* orchestrator.getThreadEventSequence( - localChild(delegated.childThreadId), - ); - const runningFollowupCall = yield* invoke("t3_thread_send", { - threadId: localChild(delegated.childThreadId), - message: cancellationPrompt, - clientRequestId: "delegated-child-replay-running-1", - }); - const runningFollowup = yield* decodeThreadSendResult( - runningFollowupCall.structuredContent, - ).pipe(Effect.orDie); - yield* orchestrator - .streamStoredEventsFrom({ - threadId: localChild(delegated.childThreadId), - afterSequence: followupStartSequence, - }) - .pipe( - Stream.filter( - (stored) => - stored.event.type === "provider-turn.updated" && - stored.event.payload.status === "running", - ), - Stream.runHead, - Effect.flatMap( - Option.match({ - onNone: () => Effect.die("Follow-up provider turn did not start."), - onSome: () => Effect.void, - }), - ), - ); - - const queuedFollowupCall = yield* invoke("t3_thread_send", { - threadId: localChild(delegated.childThreadId), - message: queuedFollowupPrompt, - mode: "queue", - clientRequestId: "delegated-child-replay-queued-1", - }); - const queuedFollowup = yield* decodeThreadSendResult( - queuedFollowupCall.structuredContent, - ).pipe(Effect.orDie); - expect(queuedFollowup).toMatchObject({ - status: "queued", - delivery: "queued", - }); + const { orchestrator, invoke, invocation, delegated, runningFollowup, queuedFollowup } = + yield* scenario.start; const pendingProjection = yield* orchestrator.getThreadProjection( localChild(delegated.childThreadId), @@ -697,7 +706,41 @@ describe("orchestrator MCP toolkit", () => { ); expect(capabilities.isError).not.toBe(true); expect(messageReads).toHaveLength(messageReadsBeforeCapabilities); - }).pipe(Effect.provide(layerTest)); + }).pipe(Effect.provide(scenario.layer)); + }), + ), + ); + + it.live("cancelling a queued follow-up task leaves another task's running turn alone", () => + Effect.scoped( + Effect.gen(function* () { + const scenario = yield* makeQueuedFollowupScenario(queuedFollowupResult); + yield* Effect.gen(function* () { + const { orchestrator, invoke, delegated, runningFollowup, queuedFollowup } = + yield* scenario.start; + const childThreadId = localChild(delegated.childThreadId); + + const cancelCall = yield* invoke("task_cancel", { + taskId: queuedFollowup.taskId, + reason: "Drop only the queued follow-up.", + clientRequestId: "cancel-queued-followup-task-1", + }); + expect(cancelCall.isError).not.toBe(true); + + const child = yield* orchestrator.getThreadProjection(childThreadId); + expect({ + running: child.runs.find((run) => run.id === runningFollowup.runId)?.status, + queued: child.runs.find((run) => run.id === queuedFollowup.runId)?.status, + interruptRequestsForRunningTurn: child.turnItems.filter( + (item) => + item.runId === runningFollowup.runId && item.type === "run_interrupt_request", + ).length, + }).toEqual({ + running: "running", + queued: "cancelled", + interruptRequestsForRunningTurn: 0, + }); + }).pipe(Effect.provide(scenario.layer)); }), ), ); diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 821a92721d357..6dcdbdd89bb27 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -60,6 +60,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct readonly projects: ProjectMutation[]; readonly launches?: OrchestrationV2ThreadLaunchInput[]; readonly projection?: OrchestrationV2ThreadProjection; + readonly childProjections?: ReadonlyMap; readonly projectionRequests?: ThreadId[]; readonly advertiseThreadStop?: boolean; }) { @@ -74,7 +75,9 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct }) => Effect.sync(() => { input.projectionRequests?.push(requestInput.threadId); - return input.projection ?? v2Projection; + return ( + input.childProjections?.get(requestInput.threadId) ?? input.projection ?? v2Projection + ); }), [ORCHESTRATION_V2_WS_METHODS.launchThread]: (launchInput: OrchestrationV2ThreadLaunchInput) => Effect.sync(() => { @@ -531,6 +534,97 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); + it.effect("stopping a queued follow-up task leaves another task's running child turn alone", () => + Effect.gen(function* () { + const childThreadId = ThreadId.make("app-owned-child-thread"); + const runningTaskId = NodeId.make("running-follow-up-task"); + const queuedTaskId = NodeId.make("queued-follow-up-task"); + const queuedRunId = RunId.make("queued-follow-up-run"); + const task = (id: NodeId, status: "running" | "pending") => ({ + id, + threadId: v2ThreadId, + runId: RunId.make("parent-run"), + parentNodeId: NodeId.make("owner-node"), + origin: "app_owned" as const, + createdBy: "agent" as const, + driver: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Work", + title: null, + model: null, + result: null, + status, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + }); + const childRun = ( + id: RunId, + ordinal: number, + status: "running" | "queued", + taskId: NodeId, + ) => ({ + id, + threadId: childThreadId, + ordinal, + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: v2Projection.thread.modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message-${id}`), + rootNodeId: null, + activeAttemptId: null, + status, + requestedAt: v2Now, + startedAt: status === "running" ? v2Now : null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + delegatedTaskId: taskId, + }); + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ + commands, + projects: [], + advertiseThreadStop: true, + projection: { + ...v2Projection, + subagents: [task(runningTaskId, "running"), task(queuedTaskId, "pending")], + }, + childProjections: new Map([ + [ + childThreadId, + { + ...v2Projection, + thread: { ...v2Projection.thread, id: childThreadId }, + runs: [ + childRun(RunId.make("running-follow-up-run"), 2, "running", runningTaskId), + childRun(queuedRunId, 3, "queued", queuedTaskId), + ], + }, + ], + ]), + }); + + yield* interruptThreadTurn({ threadId: v2ThreadId, subagentId: queuedTaskId }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + ); + + expect( + commands.filter((command) => "threadId" in command && command.threadId === childThreadId), + ).toEqual([ + { + type: "queued-run.cancel", + commandId: expect.any(String), + threadId: childThreadId, + runId: queuedRunId, + }, + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); + it.effect("stops a thread without relying on background work in the history window", () => Effect.gen(function* () { const commands: OrchestrationV2Command[] = []; From 4889f8fb4dcb24aa78eb772a1a06d17f06c7c722 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 15:42:43 +0000 Subject: [PATCH 2/8] fix(server): stopping a queued follow-up task leaves another task's turn running Cancelling a delegated follow-up task whose own child runs are all still queued, through task_cancel or the Agents panel stop, stopped the whole child thread and interrupted the live turn that belongs to another task. Both paths now cancel only that task's queued runs when none of its runs is live. Every other case still stops the child thread as before. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/mcp/OrchestratorMcpService.ts | 55 +++++++++++-------- .../client-runtime/src/operations/commands.ts | 16 ++++++ packages/contracts/src/orchestrationV2.ts | 19 +++++++ 3 files changed, 66 insertions(+), 24 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index e53ed03a7c6fa..301a6bcac1aec 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -62,6 +62,7 @@ import { type ServerProviderModel, ThreadId, isOrchestrationV2WorkActive, + queuedOnlyDelegatedTaskRuns, resolveProjectProviderInstanceEnabled, SECRET_REQUEST_UNSAVED_NOTE, } from "@t3tools/contracts"; @@ -2636,32 +2637,38 @@ const make = Effect.gen(function* () { const stopChild = Effect.gen(function* () { const commandId = stableCommandId({ scope, requestKey: key, operation: "cancel-task" }); const reason = agentStopReason(input.reason); - yield* threadManagement - .dispatch({ - type: "thread.stop", - commandId, - threadId: childThreadId, - stopDelegatedTasks: false, - reason, - }) - .pipe( - Effect.mapError((error) => - failure( - "task_not_cancellable", - `Unable to stop delegated task ${input.taskId}: ${errorMessage(error)}`, + const { runs } = yield* threadManagement + .getThreadRecords(childThreadId, ["runs"]) + .pipe(Effect.mapError(threadManagementFailure)); + // The child's live turn belongs to another task when this one's turns are all queued. + if (queuedOnlyDelegatedTaskRuns(runs, input.taskId).length === 0) { + yield* threadManagement + .dispatch({ + type: "thread.stop", + commandId, + threadId: childThreadId, + stopDelegatedTasks: false, + reason, + }) + .pipe( + Effect.mapError((error) => + failure( + "task_not_cancellable", + `Unable to stop delegated task ${input.taskId}: ${errorMessage(error)}`, + ), ), - ), - ); - yield* threadManagement - .stopDelegatedTasks({ threadId: childThreadId, commandId, reason }) - .pipe( - Effect.mapError((error) => - failure( - "task_not_cancellable", - `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`, + ); + yield* threadManagement + .stopDelegatedTasks({ threadId: childThreadId, commandId, reason }) + .pipe( + Effect.mapError((error) => + failure( + "task_not_cancellable", + `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`, + ), ), - ), - ); + ); + } const stopped = yield* threadManagement .getThreadRecords(childThreadId, ["runs"]) .pipe(Effect.mapError(threadManagementFailure)); diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 777f7e2adb4f5..2474b0e534695 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -27,6 +27,7 @@ import { type ThreadEnvMode, type UploadChatAttachment, type GuestMessageId, + queuedOnlyDelegatedTaskRuns, } from "@t3tools/contracts"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; @@ -800,6 +801,21 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea subagentId: subagent.id, }).pipe(Effect.ignore); } + const child = yield* getProjection(subagent.childThreadId); + // The child's live turn belongs to another task when this one's turns are all queued. + const queued = queuedOnlyDelegatedTaskRuns(child.runs, subagent.id); + if (queued.length > 0) { + let result = { sequence: 0 }; + for (const run of queued) { + result = yield* dispatch({ + type: "queued-run.cancel", + commandId: yield* allocateCommandId(result.sequence === 0 ? input : {}), + threadId: subagent.childThreadId, + runId: run.id, + }); + } + return result; + } threadId = subagent.childThreadId; } else { if (subagent?.runId == null) return { sequence: 0 }; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 785f41e3aa0fb..9fd1f64606a99 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -928,6 +928,25 @@ export function isOrchestrationV2WorkActive( return status === "pending" || status === "running" || status === "waiting"; } +/** + * A delegated task's queued child runs, when none of its runs is live. Stopping such a task + * cancels only these and leaves the child's live turn, which belongs to another task. + */ +export function queuedOnlyDelegatedTaskRuns( + runs: ReadonlyArray, + taskId: NodeId, +): ReadonlyArray { + const own = runs.filter((run) => run.delegatedTaskId === taskId); + const live = own.some( + (run) => + run.status === "preparing" || + run.status === "starting" || + run.status === "running" || + run.status === "waiting", + ); + return live ? [] : own.filter((run) => run.status === "queued"); +} + export const OrchestrationV2CheckpointScope = Schema.Struct({ id: CheckpointScopeId, threadId: ThreadId, From 8cfef86125fc2bf411f8bb298ca58deabbc9d00a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 16:02:37 +0000 Subject: [PATCH 3/8] fix(server): stopping a delegated task never interrupts another task's live turn Cancelling a follow-up task, or retrying the cancel after it settled, still stopped the whole child thread when its live turn belonged to another task. task_cancel and the Agents stop now leave the child running whenever its live run is bound to a different task and cancel only this task's queued runs. If this task's queued turn starts before that cancel lands, it is stopped through the normal thread stop. The web relationships row stops the task that owns the child's live work rather than its latest queued task. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 104 ++++++++++++++++++ apps/server/src/mcp/OrchestratorMcpService.ts | 50 ++++++--- ...OrchestratorMcpToolkit.integration.test.ts | 6 +- .../chat/ThreadRelationshipsControl.tsx | 9 +- .../client-runtime/src/operations/commands.ts | 45 +++++--- packages/contracts/src/orchestrationV2.ts | 22 ++-- 6 files changed, 188 insertions(+), 48 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 0faf43acb9dd5..008384bc403ab 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -818,6 +818,110 @@ describe("OrchestratorMcpService", () => { }), ); + it.effect("stops a queued follow-up task whose turn starts while it is cancelled", () => + Effect.gen(function* () { + const parentThreadId = ThreadId.make("thread:mcp-cancel-promoted-parent"); + const childThreadId = ThreadId.make("thread:mcp-cancel-promoted-child"); + const otherTaskId = NodeId.make("node:mcp-cancel-promoted-other-task"); + const taskId = NodeId.make("node:mcp-cancel-promoted-task"); + const otherRunId = RunId.make("run:mcp-cancel-promoted-other"); + const queuedRunId = RunId.make("run:mcp-cancel-promoted-queued"); + const dispatched = yield* Ref.make>([]); + const parentProjection = { + thread: { id: parentThreadId }, + runs: [], + contextTransfers: [], + subagents: [ + { + id: taskId, + threadId: parentThreadId, + origin: "app_owned", + childThreadId, + driver: "codex", + model: "gpt-5.6-terra", + result: null, + completionDelivery: { state: "pending" }, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const childProjection = (runs: ReadonlyArray) => + ({ + thread: { id: childThreadId, title: "Child" }, + runs, + contextTransfers: [], + messages: [], + subagents: [], + providerThreads: [], + runtimeRequests: [], + }) as unknown as OrchestrationV2ThreadProjection; + const child = yield* Ref.make( + childProjection([ + { id: otherRunId, ordinal: 2, status: "running", delegatedTaskId: otherTaskId }, + { id: queuedRunId, ordinal: 3, status: "queued", delegatedTaskId: taskId }, + ]), + ); + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: (threadId) => + threadId === parentThreadId ? Effect.succeed(parentProjection) : Ref.get(child), + getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), + listAppOwnedChildThreadIds: () => Effect.succeed([]), + stopDelegatedTasks: () => Effect.void, + dispatch: (command) => + Effect.gen(function* () { + yield* Ref.update(dispatched, (commands) => [...commands, command]); + if (command.type !== "queued-run.cancel") return {} as never; + // The other task's turn ended and promoted this task's turn before the cancel. + yield* Ref.set( + child, + childProjection([ + { id: otherRunId, ordinal: 2, status: "completed", delegatedTaskId: otherTaskId }, + { id: queuedRunId, ordinal: 3, status: "running", delegatedTaskId: taskId }, + ]), + ); + return yield* Effect.fail(new Error("Run is not queued.") as never); + }), + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:mcp-cancel-promoted"), + requestNamespace: "provider-session:mcp-cancel-promoted", + thread: { + threadId: parentThreadId, + providerSessionId: "provider-session:mcp-cancel-promoted", + providerInstanceId: ProviderInstanceId.make("codex"), + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const result = yield* service.cancelTask(scope, { + taskId, + clientRequestId: "cancel-promoted-task", + }); + assert.equal(result.status, "cancel_requested"); + assert.deepEqual( + (yield* Ref.get(dispatched)).map((command) => command.type), + ["queued-run.cancel", "thread.stop", "delegated_task.completion-delivery.dispose"], + ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + }), + ); + it.effect("refuses to cancel a task whose child now runs above the parent's modes", () => Effect.gen(function* () { const parentThreadId = ThreadId.make("thread:mcp-cancel-dispose-failed-parent"); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 301a6bcac1aec..1ca642b15f1af 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -62,7 +62,7 @@ import { type ServerProviderModel, ThreadId, isOrchestrationV2WorkActive, - queuedOnlyDelegatedTaskRuns, + delegatedTaskYieldsLiveRun, resolveProjectProviderInstanceEnabled, SECRET_REQUEST_UNSAVED_NOTE, } from "@t3tools/contracts"; @@ -2637,11 +2637,7 @@ const make = Effect.gen(function* () { const stopChild = Effect.gen(function* () { const commandId = stableCommandId({ scope, requestKey: key, operation: "cancel-task" }); const reason = agentStopReason(input.reason); - const { runs } = yield* threadManagement - .getThreadRecords(childThreadId, ["runs"]) - .pipe(Effect.mapError(threadManagementFailure)); - // The child's live turn belongs to another task when this one's turns are all queued. - if (queuedOnlyDelegatedTaskRuns(runs, input.taskId).length === 0) { + const stopThread = Effect.gen(function* () { yield* threadManagement .dispatch({ type: "thread.stop", @@ -2668,16 +2664,17 @@ const make = Effect.gen(function* () { ), ), ); - } - const stopped = yield* threadManagement - .getThreadRecords(childThreadId, ["runs"]) - .pipe(Effect.mapError(threadManagementFailure)); - for (const run of stopped.runs) { - if ( - run.status !== "queued" || - (run.delegatedTaskId !== input.taskId && run.id !== current.childRunId) - ) - continue; + }); + const readRuns = threadManagement.getThreadRecords(childThreadId, ["runs"]).pipe( + Effect.map(({ runs }) => runs), + Effect.mapError(threadManagementFailure), + ); + const ownRun = (run: OrchestrationV2Run) => + run.delegatedTaskId === input.taskId || run.id === current.childRunId; + const yieldsLiveRun = delegatedTaskYieldsLiveRun(yield* readRuns, input.taskId); + if (!yieldsLiveRun) yield* stopThread; + for (const run of yield* readRuns) { + if (run.status !== "queued" || !ownRun(run)) continue; // Stop preserves the queue for Resume. Cancelling a task must also // remove that task's queued turn, including a previously held one. yield* threadManagement @@ -2691,7 +2688,24 @@ const make = Effect.gen(function* () { threadId: childThreadId, runId: run.id, }) - .pipe(Effect.mapError(threadManagementFailure)); + .pipe( + Effect.mapError(threadManagementFailure), + // A turn that started meanwhile is no longer queued; it is stopped below. + Effect.catch((error) => + Effect.flatMap(readRuns, (runs) => + runs.some((latest) => latest.id === run.id && latest.status === "queued") + ? Effect.fail(error) + : Effect.void, + ), + ), + ); + } + // The other task's turn can end and promote this task's queued turn before its cancel lands. + if ( + yieldsLiveRun && + (yield* readRuns).some((run) => ownRun(run) && ThreadManagementService.isActiveRun(run)) + ) { + yield* stopThread; } }); // Checked above; a user raising one of these threads meanwhile is @@ -2742,7 +2756,7 @@ const make = Effect.gen(function* () { return subagents.some((task) => task.id === input.taskId && task.status === "cancelled"); }).pipe(Effect.mapError(threadManagementFailure)); if (isTerminalTaskStatus(current.status)) { - // The child can still run work for another turn, so it is stopped either way. + // The child can still run work for another turn; another task's live turn is left alone. yield* stopWithinLimit; const cancelled = yield* cancelOrphan; yield* disposeCompletionDelivery; diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 66e0363dd607d..3580c5b8b6ab1 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -720,12 +720,14 @@ describe("orchestrator MCP toolkit", () => { yield* scenario.start; const childThreadId = localChild(delegated.childThreadId); - const cancelCall = yield* invoke("task_cancel", { + const cancelQueued = invoke("task_cancel", { taskId: queuedFollowup.taskId, reason: "Drop only the queued follow-up.", clientRequestId: "cancel-queued-followup-task-1", }); - expect(cancelCall.isError).not.toBe(true); + expect((yield* cancelQueued).isError).not.toBe(true); + // A retry finds the task already settled and must still leave the other task's turn. + expect((yield* cancelQueued).isError).not.toBe(true); const child = yield* orchestrator.getThreadProjection(childThreadId); expect({ diff --git a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx index c91d5acebc90a..ef49182b6442e 100644 --- a/apps/web/src/components/chat/ThreadRelationshipsControl.tsx +++ b/apps/web/src/components/chat/ThreadRelationshipsControl.tsx @@ -613,9 +613,16 @@ export function ThreadRelationshipsPanel(props: { ) => { if (stoppingId !== null) return; setStoppingId(childThreadId); + // The row shows the child's live work, which an earlier task than its latest one can own. + const owner = + projected.findLast( + (subagent) => + subagent.childThreadId === childThreadId && + (subagent.status === "running" || subagent.status === "waiting"), + ) ?? agent; const result = await interruptTurn({ environmentId: props.environmentId, - input: { threadId: agent.threadId, subagentId: agent.id }, + input: { threadId: owner.threadId, subagentId: owner.id }, }); setStoppingId(null); if (result._tag === "Failure") { diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 2474b0e534695..3088f917b32ef 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -27,7 +27,7 @@ import { type ThreadEnvMode, type UploadChatAttachment, type GuestMessageId, - queuedOnlyDelegatedTaskRuns, + delegatedTaskYieldsLiveRun, } from "@t3tools/contracts"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; @@ -801,22 +801,35 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea subagentId: subagent.id, }).pipe(Effect.ignore); } - const child = yield* getProjection(subagent.childThreadId); - // The child's live turn belongs to another task when this one's turns are all queued. - const queued = queuedOnlyDelegatedTaskRuns(child.runs, subagent.id); - if (queued.length > 0) { - let result = { sequence: 0 }; - for (const run of queued) { - result = yield* dispatch({ - type: "queued-run.cancel", - commandId: yield* allocateCommandId(result.sequence === 0 ? input : {}), - threadId: subagent.childThreadId, - runId: run.id, - }); - } - return result; + const childThreadId = subagent.childThreadId; + const child = yield* getProjection(childThreadId); + // The child's live turn belongs to another task, so only this task's queued turns go. + if (delegatedTaskYieldsLiveRun(child.runs, subagent.id)) { + const cancelled = yield* Effect.gen(function* () { + let result = { sequence: 0 }; + for (const run of child.runs) { + if (run.status !== "queued" || run.delegatedTaskId !== subagent.id) continue; + result = yield* dispatch({ + type: "queued-run.cancel", + commandId: yield* allocateCommandId({}), + threadId: childThreadId, + runId: run.id, + }); + } + return result; + }).pipe( + // A queued turn that started meanwhile owns the live turn now; the stop below ends it. + Effect.catch((error) => + Effect.flatMap(getProjection(childThreadId), (latest) => + delegatedTaskYieldsLiveRun(latest.runs, subagent.id) + ? Effect.fail(error) + : Effect.succeed(null), + ), + ), + ); + if (cancelled !== null) return cancelled; } - threadId = subagent.childThreadId; + threadId = childThreadId; } else { if (subagent?.runId == null) return { sequence: 0 }; return yield* dispatch({ diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 9fd1f64606a99..247e16bd2f9c7 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -929,22 +929,22 @@ export function isOrchestrationV2WorkActive( } /** - * A delegated task's queued child runs, when none of its runs is live. Stopping such a task - * cancels only these and leaves the child's live turn, which belongs to another task. + * Whether the child's live run belongs to another delegated task. Stopping this task then + * cancels only its own queued runs instead of stopping the child thread. */ -export function queuedOnlyDelegatedTaskRuns( +export function delegatedTaskYieldsLiveRun( runs: ReadonlyArray, taskId: NodeId, -): ReadonlyArray { - const own = runs.filter((run) => run.delegatedTaskId === taskId); - const live = own.some( +): boolean { + return runs.some( (run) => - run.status === "preparing" || - run.status === "starting" || - run.status === "running" || - run.status === "waiting", + (run.status === "preparing" || + run.status === "starting" || + run.status === "running" || + run.status === "waiting") && + run.delegatedTaskId !== undefined && + run.delegatedTaskId !== taskId, ); - return live ? [] : own.filter((run) => run.status === "queued"); } export const OrchestrationV2CheckpointScope = Schema.Struct({ From 22d4c203c3b6c4594dd7899b56f001b46d4e08ef Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 16:33:32 +0000 Subject: [PATCH 4/8] fix(server): a cancelled task's started turn is interrupted by its run id Cancelling a queued delegated task fell back to a whole-thread stop when its turn started before the queued cancel landed. That stop could reach another task's turn the queue promoted in the meantime, and on clients it held that task's queued turn until Resume. The fallback now interrupts exactly the started run with run.interrupt, which the orchestrator rejects once the run has ended; a run that already ended is treated as settled. The MCP task_cancel path and the shared client stop path both do this, so no thread-wide stop runs from the queued-only branch. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 228 ++++++++++-------- apps/server/src/mcp/OrchestratorMcpService.ts | 67 +++-- .../src/operations/commands.test.ts | 225 +++++++++++------ .../client-runtime/src/operations/commands.ts | 63 +++-- packages/contracts/src/orchestrationV2.ts | 11 +- 5 files changed, 369 insertions(+), 225 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 008384bc403ab..db65179e7fb49 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -818,107 +818,139 @@ describe("OrchestratorMcpService", () => { }), ); - it.effect("stops a queued follow-up task whose turn starts while it is cancelled", () => - Effect.gen(function* () { - const parentThreadId = ThreadId.make("thread:mcp-cancel-promoted-parent"); - const childThreadId = ThreadId.make("thread:mcp-cancel-promoted-child"); - const otherTaskId = NodeId.make("node:mcp-cancel-promoted-other-task"); - const taskId = NodeId.make("node:mcp-cancel-promoted-task"); - const otherRunId = RunId.make("run:mcp-cancel-promoted-other"); - const queuedRunId = RunId.make("run:mcp-cancel-promoted-queued"); - const dispatched = yield* Ref.make>([]); - const parentProjection = { - thread: { id: parentThreadId }, - runs: [], - contextTransfers: [], - subagents: [ - { - id: taskId, - threadId: parentThreadId, - origin: "app_owned", - childThreadId, - driver: "codex", - model: "gpt-5.6-terra", - result: null, - completionDelivery: { state: "pending" }, - }, - ], - } as unknown as OrchestrationV2ThreadProjection; - const childProjection = (runs: ReadonlyArray) => - ({ - thread: { id: childThreadId, title: "Child" }, - runs, - contextTransfers: [], - messages: [], - subagents: [], - providerThreads: [], - runtimeRequests: [], - }) as unknown as OrchestrationV2ThreadProjection; - const child = yield* Ref.make( - childProjection([ - { id: otherRunId, ordinal: 2, status: "running", delegatedTaskId: otherTaskId }, - { id: queuedRunId, ordinal: 3, status: "queued", delegatedTaskId: taskId }, - ]), - ); - const layerDependencies = Layer.mergeAll( - NodeServices.layer, - NodeCrypto.layer, - Layer.mock(ThreadManagementService.ThreadManagementService)({ - getThreadRecords: (threadId) => - threadId === parentThreadId ? Effect.succeed(parentProjection) : Ref.get(child), - getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), - listAppOwnedChildThreadIds: () => Effect.succeed([]), - stopDelegatedTasks: () => Effect.void, - dispatch: (command) => - Effect.gen(function* () { - yield* Ref.update(dispatched, (commands) => [...commands, command]); - if (command.type !== "queued-run.cancel") return {} as never; - // The other task's turn ended and promoted this task's turn before the cancel. - yield* Ref.set( - child, - childProjection([ - { id: otherRunId, ordinal: 2, status: "completed", delegatedTaskId: otherTaskId }, - { id: queuedRunId, ordinal: 3, status: "running", delegatedTaskId: taskId }, - ]), - ); - return yield* Effect.fail(new Error("Run is not queued.") as never); - }), - }), - ServerSettings.layerTest(), - Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), - Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ - list: () => Effect.succeed([]), - }), - Layer.mock(ProjectService.ProjectService)({}), - Layer.mock(SecretRequests.SecretRequests)({}), - Layer.mock(RemoteDelegation.RemoteDelegation)({}), - Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), - ); - const scope: McpInvocationScope = { - environmentId: EnvironmentId.make("environment:mcp-cancel-promoted"), - requestNamespace: "provider-session:mcp-cancel-promoted", - thread: { + // Task C's queued turn is promoted when another task's turn ends, before C's queued cancel lands. + const cancelPromotedTask = Effect.fn("cancelPromotedTask")(function* (input: { + readonly name: string; + // C's turn also ends, and D's queued turn starts, before C's interrupt lands. + readonly endsBeforeInterrupt: boolean; + }) { + const parentThreadId = ThreadId.make(`thread:${input.name}-parent`); + const childThreadId = ThreadId.make(`thread:${input.name}-child`); + const otherTaskId = NodeId.make(`node:${input.name}-other-task`); + const taskId = NodeId.make(`node:${input.name}-task`); + const laterTaskId = NodeId.make(`node:${input.name}-later-task`); + const otherRunId = RunId.make(`run:${input.name}-other`); + const runId = RunId.make(`run:${input.name}-run`); + const laterRunId = RunId.make(`run:${input.name}-later`); + const dispatched = yield* Ref.make>([]); + const parentProjection = { + thread: { id: parentThreadId }, + runs: [], + contextTransfers: [], + subagents: [ + { + id: taskId, threadId: parentThreadId, - providerSessionId: "provider-session:mcp-cancel-promoted", - providerInstanceId: ProviderInstanceId.make("codex"), + origin: "app_owned", + childThreadId, + driver: "codex", + model: "gpt-5.6-terra", + result: null, + completionDelivery: { state: "pending" }, }, - client: undefined, - capabilities: new Set(["orchestration"]), - issuedAt: 1, - }; + ], + } as unknown as OrchestrationV2ThreadProjection; + const childProjection = (statuses: readonly [string, string, string]) => + ({ + thread: { id: childThreadId, title: "Child" }, + runs: [ + { id: otherRunId, ordinal: 2, status: statuses[0], delegatedTaskId: otherTaskId }, + { id: runId, ordinal: 3, status: statuses[1], delegatedTaskId: taskId }, + { id: laterRunId, ordinal: 4, status: statuses[2], delegatedTaskId: laterTaskId }, + ], + contextTransfers: [], + messages: [], + subagents: [], + providerThreads: [], + runtimeRequests: [], + }) as unknown as OrchestrationV2ThreadProjection; + const child = yield* Ref.make(childProjection(["running", "queued", "queued"])); + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: (threadId) => + threadId === parentThreadId ? Effect.succeed(parentProjection) : Ref.get(child), + getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), + listAppOwnedChildThreadIds: () => Effect.succeed([]), + stopDelegatedTasks: () => Effect.void, + dispatch: (command) => + Effect.gen(function* () { + yield* Ref.update(dispatched, (commands) => [...commands, command]); + if (command.type === "queued-run.cancel") { + yield* Ref.set(child, childProjection(["completed", "running", "queued"])); + return yield* Effect.fail(new Error("Run is not queued.") as never); + } + if (command.type === "run.interrupt" && input.endsBeforeInterrupt) { + yield* Ref.set(child, childProjection(["completed", "completed", "running"])); + return yield* Effect.fail(new Error("Run is not interruptible.") as never); + } + return {} as never; + }), + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + }), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + const scope: McpInvocationScope = { + environmentId: EnvironmentId.make(`environment:${input.name}`), + requestNamespace: `provider-session:${input.name}`, + thread: { + threadId: parentThreadId, + providerSessionId: `provider-session:${input.name}`, + providerInstanceId: ProviderInstanceId.make("codex"), + }, + client: undefined, + capabilities: new Set(["orchestration"]), + issuedAt: 1, + }; + const result = yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + return yield* service.cancelTask(scope, { taskId, clientRequestId: `${input.name}-cancel` }); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + return { + status: result.status, + runId, + commands: (yield* Ref.get(dispatched)).map((command) => ({ + type: command.type, + runId: "runId" in command ? command.runId : undefined, + })), + }; + }); - yield* Effect.gen(function* () { - const service = yield* OrchestratorMcpService.OrchestratorMcpService; - const result = yield* service.cancelTask(scope, { - taskId, - clientRequestId: "cancel-promoted-task", - }); - assert.equal(result.status, "cancel_requested"); - assert.deepEqual( - (yield* Ref.get(dispatched)).map((command) => command.type), - ["queued-run.cancel", "thread.stop", "delegated_task.completion-delivery.dispose"], - ); - }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + it.effect("stops a queued follow-up task whose turn starts while it is cancelled", () => + Effect.gen(function* () { + const cancelled = yield* cancelPromotedTask({ + name: "mcp-cancel-promoted", + endsBeforeInterrupt: false, + }); + assert.equal(cancelled.status, "cancel_requested"); + assert.deepEqual(cancelled.commands, [ + { type: "queued-run.cancel", runId: cancelled.runId }, + { type: "run.interrupt", runId: cancelled.runId }, + { type: "delegated_task.completion-delivery.dispose", runId: undefined }, + ]); + }), + ); + + it.effect("never stops the turn the queue starts after a cancelled task's turn ends", () => + Effect.gen(function* () { + const cancelled = yield* cancelPromotedTask({ + name: "mcp-cancel-promoted-ended", + endsBeforeInterrupt: true, + }); + assert.equal(cancelled.status, "cancel_requested"); + assert.deepEqual(cancelled.commands, [ + { type: "queued-run.cancel", runId: cancelled.runId }, + { type: "run.interrupt", runId: cancelled.runId }, + { type: "delegated_task.completion-delivery.dispose", runId: undefined }, + ]); }), ); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 1ca642b15f1af..d8a6e12d4eeaf 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2637,7 +2637,39 @@ const make = Effect.gen(function* () { const stopChild = Effect.gen(function* () { const commandId = stableCommandId({ scope, requestKey: key, operation: "cancel-task" }); const reason = agentStopReason(input.reason); - const stopThread = Effect.gen(function* () { + const readRuns = threadManagement.getThreadRecords(childThreadId, ["runs"]).pipe( + Effect.map(({ runs }) => runs), + Effect.mapError(threadManagementFailure), + ); + const readRun = (runId: RunId) => + Effect.map(readRuns, (runs) => runs.find((run) => run.id === runId)); + // By its id: a thread-wide stop could reach another task's turn the queue promoted + // meanwhile. A turn that ended before the interrupt landed needs nothing more. + const interruptOwnRun = (runId: RunId) => + threadManagement + .dispatch({ + type: "run.interrupt", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: `cancel-task-run:${runId}`, + }), + threadId: childThreadId, + runId, + reason, + }) + .pipe( + Effect.mapError(threadManagementFailure), + Effect.catch((error) => + Effect.flatMap(readRun(runId), (run) => + run !== undefined && ThreadManagementService.isActiveRun(run) + ? Effect.fail(error) + : Effect.void, + ), + ), + ); + // The child's live turn belongs to another task, so only this task's queued turns go. + if (!delegatedTaskYieldsLiveRun(yield* readRuns, input.taskId)) { yield* threadManagement .dispatch({ type: "thread.stop", @@ -2664,17 +2696,13 @@ const make = Effect.gen(function* () { ), ), ); - }); - const readRuns = threadManagement.getThreadRecords(childThreadId, ["runs"]).pipe( - Effect.map(({ runs }) => runs), - Effect.mapError(threadManagementFailure), - ); - const ownRun = (run: OrchestrationV2Run) => - run.delegatedTaskId === input.taskId || run.id === current.childRunId; - const yieldsLiveRun = delegatedTaskYieldsLiveRun(yield* readRuns, input.taskId); - if (!yieldsLiveRun) yield* stopThread; + } for (const run of yield* readRuns) { - if (run.status !== "queued" || !ownRun(run)) continue; + if ( + run.status !== "queued" || + (run.delegatedTaskId !== input.taskId && run.id !== current.childRunId) + ) + continue; // Stop preserves the queue for Resume. Cancelling a task must also // remove that task's queued turn, including a previously held one. yield* threadManagement @@ -2690,23 +2718,18 @@ const make = Effect.gen(function* () { }) .pipe( Effect.mapError(threadManagementFailure), - // A turn that started meanwhile is no longer queued; it is stopped below. + // A turn that started meanwhile is no longer queued. Effect.catch((error) => - Effect.flatMap(readRuns, (runs) => - runs.some((latest) => latest.id === run.id && latest.status === "queued") + Effect.flatMap(readRun(run.id), (latest) => + latest?.status === "queued" ? Effect.fail(error) - : Effect.void, + : latest !== undefined && ThreadManagementService.isActiveRun(latest) + ? interruptOwnRun(run.id) + : Effect.void, ), ), ); } - // The other task's turn can end and promote this task's queued turn before its cancel lands. - if ( - yieldsLiveRun && - (yield* readRuns).some((run) => ownRun(run) && ThreadManagementService.isActiveRun(run)) - ) { - yield* stopThread; - } }); // Checked above; a user raising one of these threads meanwhile is // caught again under that thread's lock, which leaves it running. diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 6dcdbdd89bb27..21397885f10b6 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -15,6 +15,7 @@ import { TurnItemId, WS_METHODS, type OrchestrationV2Command, + OrchestrationV2DispatchCommandError, type OrchestrationV2ThreadLaunchInput, type OrchestrationV2ThreadProjection, type ProjectMutation, @@ -63,13 +64,16 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct readonly childProjections?: ReadonlyMap; readonly projectionRequests?: ThreadId[]; readonly advertiseThreadStop?: boolean; + readonly onCommand?: ( + command: OrchestrationV2Command, + ) => Effect.Effect; }) { const client = { [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => Effect.sync(() => { input.commands.push(command); return { sequence: input.commands.length }; - }), + }).pipe(Effect.tap(() => input.onCommand?.(command) ?? Effect.void)), [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: (requestInput: { readonly threadId: ThreadId; }) => @@ -534,92 +538,155 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); + // Task B's turn runs on the child thread while tasks C and D are queued behind it. + const stopQueuedFollowupTask = Effect.fn("stopQueuedFollowupTask")(function* ( + // C's turn had started, or also ended, when its queued cancel landed. + promoted?: "running" | "completed", + ) { + const childThreadId = ThreadId.make("app-owned-child-thread"); + const runningTaskId = NodeId.make("running-follow-up-task"); + const queuedTaskId = NodeId.make("queued-follow-up-task"); + const laterTaskId = NodeId.make("later-follow-up-task"); + const queuedRunId = RunId.make("queued-follow-up-run"); + const task = (id: NodeId, status: "running" | "pending") => ({ + id, + threadId: v2ThreadId, + runId: RunId.make("parent-run"), + parentNodeId: NodeId.make("owner-node"), + origin: "app_owned" as const, + createdBy: "agent" as const, + driver: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + providerThreadId: null, + childThreadId, + nativeTaskRef: null, + prompt: "Work", + title: null, + model: null, + result: null, + status, + startedAt: v2Now, + completedAt: null, + updatedAt: v2Now, + }); + const childRun = ( + id: RunId, + ordinal: number, + status: "running" | "queued" | "completed", + taskId: NodeId, + ) => ({ + id, + threadId: childThreadId, + ordinal, + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: v2Projection.thread.modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message-${id}`), + rootNodeId: null, + activeAttemptId: null, + status, + requestedAt: v2Now, + startedAt: status === "queued" ? null : v2Now, + completedAt: status === "completed" ? v2Now : null, + checkpointId: null, + contextHandoffId: null, + delegatedTaskId: taskId, + }); + const childProjection = ( + running: "running" | "completed", + queued: "running" | "queued" | "completed", + ) => ({ + ...v2Projection, + thread: { ...v2Projection.thread, id: childThreadId }, + runs: [ + childRun(RunId.make("running-follow-up-run"), 2, running, runningTaskId), + childRun(queuedRunId, 3, queued, queuedTaskId), + childRun(RunId.make("later-follow-up-run"), 4, "queued", laterTaskId), + ], + }); + const childProjections = new Map([[childThreadId, childProjection("running", "queued")]]); + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ + commands, + projects: [], + advertiseThreadStop: true, + projection: { + ...v2Projection, + subagents: [ + task(runningTaskId, "running"), + task(queuedTaskId, "pending"), + task(laterTaskId, "pending"), + ], + }, + childProjections, + onCommand: (command) => { + if (command.type !== "queued-run.cancel" || promoted === undefined) return Effect.void; + childProjections.set(childThreadId, childProjection("completed", promoted)); + return Effect.fail( + new OrchestrationV2DispatchCommandError({ + commandId: command.commandId, + commandType: command.type, + message: "Run is not queued.", + }), + ); + }, + }); + + yield* interruptThreadTurn({ threadId: v2ThreadId, subagentId: queuedTaskId }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + ); + return { + childThreadId, + queuedRunId, + commands: commands.filter( + (command) => "threadId" in command && command.threadId === childThreadId, + ), + }; + }); + it.effect("stopping a queued follow-up task leaves another task's running child turn alone", () => Effect.gen(function* () { - const childThreadId = ThreadId.make("app-owned-child-thread"); - const runningTaskId = NodeId.make("running-follow-up-task"); - const queuedTaskId = NodeId.make("queued-follow-up-task"); - const queuedRunId = RunId.make("queued-follow-up-run"); - const task = (id: NodeId, status: "running" | "pending") => ({ - id, - threadId: v2ThreadId, - runId: RunId.make("parent-run"), - parentNodeId: NodeId.make("owner-node"), - origin: "app_owned" as const, - createdBy: "agent" as const, - driver: ProviderDriverKind.make("codex"), - providerInstanceId: ProviderInstanceId.make("codex"), - providerThreadId: null, - childThreadId, - nativeTaskRef: null, - prompt: "Work", - title: null, - model: null, - result: null, - status, - startedAt: v2Now, - completedAt: null, - updatedAt: v2Now, - }); - const childRun = ( - id: RunId, - ordinal: number, - status: "running" | "queued", - taskId: NodeId, - ) => ({ - id, - threadId: childThreadId, - ordinal, - providerInstanceId: ProviderInstanceId.make("codex"), - modelSelection: v2Projection.thread.modelSelection, - providerThreadId: null, - userMessageId: MessageId.make(`message-${id}`), - rootNodeId: null, - activeAttemptId: null, - status, - requestedAt: v2Now, - startedAt: status === "running" ? v2Now : null, - completedAt: null, - checkpointId: null, - contextHandoffId: null, - delegatedTaskId: taskId, - }); - const commands: OrchestrationV2Command[] = []; - const supervisor = yield* makeSupervisor({ - commands, - projects: [], - advertiseThreadStop: true, - projection: { - ...v2Projection, - subagents: [task(runningTaskId, "running"), task(queuedTaskId, "pending")], + const stopped = yield* stopQueuedFollowupTask(); + expect(stopped.commands).toEqual([ + { + type: "queued-run.cancel", + commandId: expect.any(String), + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, }, - childProjections: new Map([ - [ - childThreadId, - { - ...v2Projection, - thread: { ...v2Projection.thread, id: childThreadId }, - runs: [ - childRun(RunId.make("running-follow-up-run"), 2, "running", runningTaskId), - childRun(queuedRunId, 3, "queued", queuedTaskId), - ], - }, - ], - ]), - }); + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); - yield* interruptThreadTurn({ threadId: v2ThreadId, subagentId: queuedTaskId }).pipe( - Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), - ); + it.effect("stopping a follow-up task whose turn just started interrupts only that turn", () => + Effect.gen(function* () { + const stopped = yield* stopQueuedFollowupTask("running"); + expect(stopped.commands).toEqual([ + { + type: "queued-run.cancel", + commandId: expect.any(String), + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, + }, + { + type: "run.interrupt", + commandId: expect.any(String), + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, + }, + ]); + }).pipe(Effect.provide(layerTestRandom)), + ); - expect( - commands.filter((command) => "threadId" in command && command.threadId === childThreadId), - ).toEqual([ + it.effect("stopping a follow-up task whose turn already ended holds no other task's turn", () => + Effect.gen(function* () { + const stopped = yield* stopQueuedFollowupTask("completed"); + expect(stopped.commands).toEqual([ { type: "queued-run.cancel", commandId: expect.any(String), - threadId: childThreadId, - runId: queuedRunId, + threadId: stopped.childThreadId, + runId: stopped.queuedRunId, }, ]); }).pipe(Effect.provide(layerTestRandom)), diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 3088f917b32ef..324e034b62fa6 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -28,6 +28,7 @@ import { type UploadChatAttachment, type GuestMessageId, delegatedTaskYieldsLiveRun, + isOrchestrationV2RunActive, } from "@t3tools/contracts"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; @@ -805,29 +806,47 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea const child = yield* getProjection(childThreadId); // The child's live turn belongs to another task, so only this task's queued turns go. if (delegatedTaskYieldsLiveRun(child.runs, subagent.id)) { - const cancelled = yield* Effect.gen(function* () { - let result = { sequence: 0 }; - for (const run of child.runs) { - if (run.status !== "queued" || run.delegatedTaskId !== subagent.id) continue; - result = yield* dispatch({ - type: "queued-run.cancel", - commandId: yield* allocateCommandId({}), - threadId: childThreadId, - runId: run.id, - }); - } - return result; - }).pipe( - // A queued turn that started meanwhile owns the live turn now; the stop below ends it. - Effect.catch((error) => - Effect.flatMap(getProjection(childThreadId), (latest) => - delegatedTaskYieldsLiveRun(latest.runs, subagent.id) - ? Effect.fail(error) - : Effect.succeed(null), + const readRun = (runId: RunId) => + Effect.map(getProjection(childThreadId), (latest) => + latest.runs.find((run) => run.id === runId), + ); + const settled = { sequence: 0 }; + // By its id: a thread-wide stop could reach, or hold, another task's turn. A turn that + // ended before the interrupt landed needs nothing more. + const interruptOwnRun = (runId: RunId) => + Effect.flatMap(allocateCommandId({}), (commandId) => + dispatch({ type: "run.interrupt", commandId, threadId: childThreadId, runId }), + ).pipe( + Effect.catch((error) => + Effect.flatMap(readRun(runId), (run) => + run !== undefined && isOrchestrationV2RunActive(run.status) + ? Effect.fail(error) + : Effect.succeed(settled), + ), ), - ), - ); - if (cancelled !== null) return cancelled; + ); + let result = settled; + for (const run of child.runs) { + if (run.status !== "queued" || run.delegatedTaskId !== subagent.id) continue; + result = yield* dispatch({ + type: "queued-run.cancel", + commandId: yield* allocateCommandId({}), + threadId: childThreadId, + runId: run.id, + }).pipe( + // A turn that started meanwhile is no longer queued. + Effect.catch((error) => + Effect.flatMap(readRun(run.id), (latest) => + latest?.status === "queued" + ? Effect.fail(error) + : latest !== undefined && isOrchestrationV2RunActive(latest.status) + ? interruptOwnRun(run.id) + : Effect.succeed(settled), + ), + ), + ); + } + return result; } threadId = childThreadId; } else { diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 247e16bd2f9c7..3f73aa23d6817 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -928,6 +928,12 @@ export function isOrchestrationV2WorkActive( return status === "pending" || status === "running" || status === "waiting"; } +export function isOrchestrationV2RunActive(status: OrchestrationV2Run["status"]): boolean { + return ( + status === "preparing" || status === "starting" || status === "running" || status === "waiting" + ); +} + /** * Whether the child's live run belongs to another delegated task. Stopping this task then * cancels only its own queued runs instead of stopping the child thread. @@ -938,10 +944,7 @@ export function delegatedTaskYieldsLiveRun( ): boolean { return runs.some( (run) => - (run.status === "preparing" || - run.status === "starting" || - run.status === "running" || - run.status === "waiting") && + isOrchestrationV2RunActive(run.status) && run.delegatedTaskId !== undefined && run.delegatedTaskId !== taskId, ); From fcf065076aa3c001f4753e95c19f6ac45243eea1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 16:48:07 +0000 Subject: [PATCH 5/8] fix(server): cancelling a queued task reads the child's runs once A turn the queue promoted between the ownership check and the queued-run sweep was neither cancelled nor interrupted, so task_cancel returned cancel_requested while it kept running. Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/OrchestratorMcpService.test.ts | 35 +++++++++++++++++-- apps/server/src/mcp/OrchestratorMcpService.ts | 7 ++-- 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index db65179e7fb49..52c1e47f1d63a 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -823,6 +823,8 @@ describe("OrchestratorMcpService", () => { readonly name: string; // C's turn also ends, and D's queued turn starts, before C's interrupt lands. readonly endsBeforeInterrupt: boolean; + // The promotion lands right after the cancel's Nth read of the child's runs. + readonly promotesAfterRunsRead?: number; }) { const parentThreadId = ThreadId.make(`thread:${input.name}-parent`); const childThreadId = ThreadId.make(`thread:${input.name}-child`); @@ -865,12 +867,25 @@ describe("OrchestratorMcpService", () => { runtimeRequests: [], }) as unknown as OrchestrationV2ThreadProjection; const child = yield* Ref.make(childProjection(["running", "queued", "queued"])); + const runsReads = yield* Ref.make(0); const layerDependencies = Layer.mergeAll( NodeServices.layer, NodeCrypto.layer, Layer.mock(ThreadManagementService.ThreadManagementService)({ - getThreadRecords: (threadId) => - threadId === parentThreadId ? Effect.succeed(parentProjection) : Ref.get(child), + getThreadRecords: (threadId, records) => + threadId === parentThreadId + ? Effect.succeed(parentProjection) + : Effect.tap(Ref.get(child), () => + records.length === 1 && records[0] === "runs" + ? Effect.flatMap( + Ref.updateAndGet(runsReads, (reads) => reads + 1), + (reads) => + reads === input.promotesAfterRunsRead + ? Ref.set(child, childProjection(["completed", "running", "queued"])) + : Effect.void, + ) + : Effect.void, + ), getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), listAppOwnedChildThreadIds: () => Effect.succeed([]), stopDelegatedTasks: () => Effect.void, @@ -939,6 +954,22 @@ describe("OrchestratorMcpService", () => { }), ); + it.effect("stops a queued follow-up task whose turn starts right after the cancel reads it", () => + Effect.gen(function* () { + const cancelled = yield* cancelPromotedTask({ + name: "mcp-cancel-promoted-after-read", + endsBeforeInterrupt: false, + promotesAfterRunsRead: 2, + }); + assert.equal(cancelled.status, "cancel_requested"); + assert.deepEqual(cancelled.commands, [ + { type: "queued-run.cancel", runId: cancelled.runId }, + { type: "run.interrupt", runId: cancelled.runId }, + { type: "delegated_task.completion-delivery.dispose", runId: undefined }, + ]); + }), + ); + it.effect("never stops the turn the queue starts after a cancelled task's turn ends", () => Effect.gen(function* () { const cancelled = yield* cancelPromotedTask({ diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index d8a6e12d4eeaf..a50fee8d449bf 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2668,8 +2668,11 @@ const make = Effect.gen(function* () { ), ), ); + // One snapshot for the check and the sweep: a turn promoted between two reads would + // be neither cancelled nor interrupted. + const runs = yield* readRuns; // The child's live turn belongs to another task, so only this task's queued turns go. - if (!delegatedTaskYieldsLiveRun(yield* readRuns, input.taskId)) { + if (!delegatedTaskYieldsLiveRun(runs, input.taskId)) { yield* threadManagement .dispatch({ type: "thread.stop", @@ -2697,7 +2700,7 @@ const make = Effect.gen(function* () { ), ); } - for (const run of yield* readRuns) { + for (const run of runs) { if ( run.status !== "queued" || (run.delegatedTaskId !== input.taskId && run.id !== current.childRunId) From d835d32afcf374b88356c304f04ad5e0435e0674 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 23:10:16 +0000 Subject: [PATCH 6/8] test(server): drop a duplicate cancel race case and fix mocks after merge The promote-after-second-read case exercised the same path as the cancel-time promotion once cancellation reads the child's runs once. Mock projections now carry providerTurns, which main's run model lookup reads. The client's active run lookup reuses isOrchestrationV2RunActive. AI-assisted; verified with the tests and tools listed above. --- .../src/mcp/OrchestratorMcpService.test.ts | 37 ++----------------- .../client-runtime/src/operations/commands.ts | 8 +--- 2 files changed, 5 insertions(+), 40 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 34bd959b5b4f4..c99a5909bc105 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -830,8 +830,6 @@ describe("OrchestratorMcpService", () => { readonly name: string; // C's turn also ends, and D's queued turn starts, before C's interrupt lands. readonly endsBeforeInterrupt: boolean; - // The promotion lands right after the cancel's Nth read of the child's runs. - readonly promotesAfterRunsRead?: number; }) { const parentThreadId = ThreadId.make(`thread:${input.name}-parent`); const childThreadId = ThreadId.make(`thread:${input.name}-child`); @@ -845,6 +843,7 @@ describe("OrchestratorMcpService", () => { const parentProjection = { thread: { id: parentThreadId }, runs: [], + providerTurns: [], contextTransfers: [], subagents: [ { @@ -871,28 +870,16 @@ describe("OrchestratorMcpService", () => { messages: [], subagents: [], providerThreads: [], + providerTurns: [], runtimeRequests: [], }) as unknown as OrchestrationV2ThreadProjection; const child = yield* Ref.make(childProjection(["running", "queued", "queued"])); - const runsReads = yield* Ref.make(0); const layerDependencies = Layer.mergeAll( NodeServices.layer, NodeCrypto.layer, Layer.mock(ThreadManagementService.ThreadManagementService)({ - getThreadRecords: (threadId, records) => - threadId === parentThreadId - ? Effect.succeed(parentProjection) - : Effect.tap(Ref.get(child), () => - records.length === 1 && records[0] === "runs" - ? Effect.flatMap( - Ref.updateAndGet(runsReads, (reads) => reads + 1), - (reads) => - reads === input.promotesAfterRunsRead - ? Ref.set(child, childProjection(["completed", "running", "queued"])) - : Effect.void, - ) - : Effect.void, - ), + getThreadRecords: (threadId) => + threadId === parentThreadId ? Effect.succeed(parentProjection) : Ref.get(child), getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), listAppOwnedChildThreadIds: () => Effect.succeed([]), stopDelegatedTasks: () => Effect.void, @@ -961,22 +948,6 @@ describe("OrchestratorMcpService", () => { }), ); - it.effect("stops a queued follow-up task whose turn starts right after the cancel reads it", () => - Effect.gen(function* () { - const cancelled = yield* cancelPromotedTask({ - name: "mcp-cancel-promoted-after-read", - endsBeforeInterrupt: false, - promotesAfterRunsRead: 2, - }); - assert.equal(cancelled.status, "cancel_requested"); - assert.deepEqual(cancelled.commands, [ - { type: "queued-run.cancel", runId: cancelled.runId }, - { type: "run.interrupt", runId: cancelled.runId }, - { type: "delegated_task.completion-delivery.dispose", runId: undefined }, - ]); - }), - ); - it.effect("never stops the turn the queue starts after a cancelled task's turn ends", () => Effect.gen(function* () { const cancelled = yield* cancelPromotedTask({ diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 64755d97eee2a..91bb9ac2ef1a1 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -880,13 +880,7 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea } if (runId === undefined) { const projection = yield* getProjection(threadId); - runId = projection.runs.findLast( - (run) => - run.status === "preparing" || - run.status === "starting" || - run.status === "running" || - run.status === "waiting", - )?.id; + runId = projection.runs.findLast((run) => isOrchestrationV2RunActive(run.status))?.id; if (runId === undefined) { const latestRun = projection.runs.at(-1); if ( From 3c05c1c12c7b4562bb620b648e7bb628ca2f8076 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 23:39:50 +0000 Subject: [PATCH 7/8] fix(server): a cancelled task's promoted turn stops the tasks it delegated When cancelling a queued task races with its turn starting, the fallback interrupted that turn by id but left the tasks it had delegated running. stopDelegatedTasks now takes an optional runId that limits it to the tasks that run delegated, with their whole subtrees, and the fallback calls it for the promoted run, so another task's live turn and delegated work are still left alone. AI-assisted; verified with the tests and tools listed above. --- .../src/mcp/OrchestratorMcpService.test.ts | 8 ++- apps/server/src/mcp/OrchestratorMcpService.ts | 62 +++++++++++++------ .../ThreadManagementService.ts | 5 +- 3 files changed, 53 insertions(+), 22 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index c99a5909bc105..921e212a516b0 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -874,6 +874,7 @@ describe("OrchestratorMcpService", () => { runtimeRequests: [], }) as unknown as OrchestrationV2ThreadProjection; const child = yield* Ref.make(childProjection(["running", "queued", "queued"])); + const descendantStops = yield* Ref.make>([]); const layerDependencies = Layer.mergeAll( NodeServices.layer, NodeCrypto.layer, @@ -882,7 +883,7 @@ describe("OrchestratorMcpService", () => { threadId === parentThreadId ? Effect.succeed(parentProjection) : Ref.get(child), getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), listAppOwnedChildThreadIds: () => Effect.succeed([]), - stopDelegatedTasks: () => Effect.void, + stopDelegatedTasks: (stop) => Ref.update(descendantStops, (runs) => [...runs, stop.runId]), dispatch: (command) => Effect.gen(function* () { yield* Ref.update(dispatched, (commands) => [...commands, command]); @@ -926,6 +927,7 @@ describe("OrchestratorMcpService", () => { return { status: result.status, runId, + descendantStops: yield* Ref.get(descendantStops), commands: (yield* Ref.get(dispatched)).map((command) => ({ type: command.type, runId: "runId" in command ? command.runId : undefined, @@ -945,6 +947,8 @@ describe("OrchestratorMcpService", () => { { type: "run.interrupt", runId: cancelled.runId }, { type: "delegated_task.completion-delivery.dispose", runId: undefined }, ]); + // Only the tasks the promoted turn delegated, not the other task's. + assert.deepEqual(cancelled.descendantStops, [cancelled.runId]); }), ); @@ -960,6 +964,8 @@ describe("OrchestratorMcpService", () => { { type: "run.interrupt", runId: cancelled.runId }, { type: "delegated_task.completion-delivery.dispose", runId: undefined }, ]); + // Only the tasks the promoted turn delegated, not the other task's. + assert.deepEqual(cancelled.descendantStops, [cancelled.runId]); }), ); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 3f914c1e15081..c6d27fbce0764 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2650,30 +2650,52 @@ const make = Effect.gen(function* () { const readRun = (runId: RunId) => Effect.map(readRuns, (runs) => runs.find((run) => run.id === runId)); // By its id: a thread-wide stop could reach another task's turn the queue promoted - // meanwhile. A turn that ended before the interrupt landed needs nothing more. - const interruptOwnRun = (runId: RunId) => - threadManagement - .dispatch({ - type: "run.interrupt", - commandId: stableCommandId({ - scope, - requestKey: key, - operation: `cancel-task-run:${runId}`, - }), + // meanwhile. A turn that ended before the interrupt landed needs nothing more, but the + // tasks it delegated still go, and only those. + const stopOwnRun = Effect.fn("OrchestratorMcpService.stopOwnRun")(function* ( + run: OrchestrationV2Run, + ) { + const runCommandId = stableCommandId({ + scope, + requestKey: key, + operation: `cancel-task-run:${run.id}`, + }); + if (ThreadManagementService.isActiveRun(run)) { + yield* threadManagement + .dispatch({ + type: "run.interrupt", + commandId: runCommandId, + threadId: childThreadId, + runId: run.id, + reason, + }) + .pipe( + Effect.mapError(threadManagementFailure), + Effect.catch((error) => + Effect.flatMap(readRun(run.id), (latest) => + latest !== undefined && ThreadManagementService.isActiveRun(latest) + ? Effect.fail(error) + : Effect.void, + ), + ), + ); + } + yield* threadManagement + .stopDelegatedTasks({ threadId: childThreadId, - runId, + commandId: runCommandId, reason, + runId: run.id, }) .pipe( - Effect.mapError(threadManagementFailure), - Effect.catch((error) => - Effect.flatMap(readRun(runId), (run) => - run !== undefined && ThreadManagementService.isActiveRun(run) - ? Effect.fail(error) - : Effect.void, + Effect.mapError((error) => + failure( + "task_not_cancellable", + `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`, ), ), ); + }); // One snapshot for the check and the sweep: a turn promoted between two reads would // be neither cancelled nor interrupted. const runs = yield* readRuns; @@ -2732,9 +2754,9 @@ const make = Effect.gen(function* () { Effect.flatMap(readRun(run.id), (latest) => latest?.status === "queued" ? Effect.fail(error) - : latest !== undefined && ThreadManagementService.isActiveRun(latest) - ? interruptOwnRun(run.id) - : Effect.void, + : latest === undefined + ? Effect.void + : stopOwnRun(latest), ), ), ); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index de12768ac7b17..bcebdce3b246f 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -487,6 +487,8 @@ export interface ThreadManagementServiceShape { readonly threadId: ThreadId; readonly commandId: CommandId; readonly reason?: string | undefined; + /** Only the tasks this run delegated, with their whole subtrees. */ + readonly runId?: RunId | undefined; }) => Effect.Effect< void, Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError @@ -1477,6 +1479,7 @@ const make = Effect.gen(function* () { const stoppedChildThreadIds = new Set(); for (const task of subagents) { if (task.origin !== "app_owned") continue; + if (input.runId !== undefined && task.runId !== input.runId) continue; if (task.childThreadId === null) { // A task in a linked environment ends here; its follower then stops it there. // One with its result keeps it, but its thread there may run follow-up work. @@ -1523,7 +1526,7 @@ const make = Effect.gen(function* () { stopDelegatedTasks: false, ...(input.reason === undefined ? {} : { reason: input.reason }), }).pipe( - Effect.andThen(stopDelegatedTasks({ ...input, threadId })), + Effect.andThen(stopDelegatedTasks({ ...input, threadId, runId: undefined })), Effect.catch((error) => Effect.logWarning("Unable to stop a delegated task", { parentThreadId: input.threadId, From 29a6a1a2f458e1daee9d9975ce5d42b013324ae6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 23:59:40 +0000 Subject: [PATCH 8/8] fix(client-runtime): a stopped task's promoted turn stops the tasks it delegated When stopping a queued follow-up task races with its turn starting, the client interrupted that turn by id but left the tasks it had delegated running. It now stops the child thread of each live app-owned task that run delegated, which also stops their own delegated work, and leaves another task's turn and delegated work alone. AI-assisted; verified with the tests and tools listed above. --- .../src/operations/commands.test.ts | 32 +++++++++- .../client-runtime/src/operations/commands.ts | 59 ++++++++++++++----- 2 files changed, 75 insertions(+), 16 deletions(-) diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 21397885f10b6..a4a9847d6b5d3 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -548,6 +548,7 @@ describe("V2 environment commands", () => { const queuedTaskId = NodeId.make("queued-follow-up-task"); const laterTaskId = NodeId.make("later-follow-up-task"); const queuedRunId = RunId.make("queued-follow-up-run"); + const grandchildThreadId = ThreadId.make("grandchild-thread"); const task = (id: NodeId, status: "running" | "pending") => ({ id, threadId: v2ThreadId, @@ -603,6 +604,24 @@ describe("V2 environment commands", () => { childRun(queuedRunId, 3, queued, queuedTaskId), childRun(RunId.make("later-follow-up-run"), 4, "queued", laterTaskId), ], + // Once promoted, the queued task's turn delegated a task of its own, beside the other's. + subagents: + queued === "queued" + ? [] + : [ + { + ...task(NodeId.make("other-grandchild-task"), "running"), + threadId: childThreadId, + runId: RunId.make("running-follow-up-run"), + childThreadId: ThreadId.make("other-grandchild-thread"), + }, + { + ...task(NodeId.make("grandchild-task"), "running"), + threadId: childThreadId, + runId: queuedRunId, + childThreadId: grandchildThreadId, + }, + ], }); const childProjections = new Map([[childThreadId, childProjection("running", "queued")]]); const commands: OrchestrationV2Command[] = []; @@ -638,8 +657,9 @@ describe("V2 environment commands", () => { return { childThreadId, queuedRunId, + grandchildThreadId, commands: commands.filter( - (command) => "threadId" in command && command.threadId === childThreadId, + (command) => "threadId" in command && command.threadId !== v2ThreadId, ), }; }); @@ -674,6 +694,11 @@ describe("V2 environment commands", () => { threadId: stopped.childThreadId, runId: stopped.queuedRunId, }, + { + type: "thread.stop", + commandId: expect.any(String), + threadId: stopped.grandchildThreadId, + }, ]); }).pipe(Effect.provide(layerTestRandom)), ); @@ -688,6 +713,11 @@ describe("V2 environment commands", () => { threadId: stopped.childThreadId, runId: stopped.queuedRunId, }, + { + type: "thread.stop", + commandId: expect.any(String), + threadId: stopped.grandchildThreadId, + }, ]); }).pipe(Effect.provide(layerTestRandom)), ); diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 91bb9ac2ef1a1..24a2d22ba60f4 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -24,11 +24,13 @@ import { type RuntimeMode, type RuntimeRequestId, type ThreadId, + type OrchestrationV2Run, type ThreadEnvMode, type UploadChatAttachment, type GuestMessageId, delegatedTaskYieldsLiveRun, isOrchestrationV2RunActive, + isOrchestrationV2WorkActive, } from "@t3tools/contracts"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; @@ -816,19 +818,46 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea ); const settled = { sequence: 0 }; // By its id: a thread-wide stop could reach, or hold, another task's turn. A turn that - // ended before the interrupt landed needs nothing more. - const interruptOwnRun = (runId: RunId) => - Effect.flatMap(allocateCommandId({}), (commandId) => - dispatch({ type: "run.interrupt", commandId, threadId: childThreadId, runId }), - ).pipe( - Effect.catch((error) => - Effect.flatMap(readRun(runId), (run) => - run !== undefined && isOrchestrationV2RunActive(run.status) - ? Effect.fail(error) - : Effect.succeed(settled), + // ended before the interrupt landed needs nothing more, but the tasks it delegated + // still go, and only those. + const stopOwnRun = Effect.fn("EnvironmentCommands.stopOwnRun")(function* ( + run: OrchestrationV2Run, + ) { + let result = settled; + if (isOrchestrationV2RunActive(run.status)) { + result = yield* dispatch({ + type: "run.interrupt", + commandId: yield* allocateCommandId({}), + threadId: childThreadId, + runId: run.id, + }).pipe( + Effect.catch((error) => + Effect.flatMap(readRun(run.id), (latest) => + latest !== undefined && isOrchestrationV2RunActive(latest.status) + ? Effect.fail(error) + : Effect.succeed(settled), + ), ), - ), - ); + ); + } + // A child thread's stop also stops the tasks it delegated, all the way down. + const { subagents } = yield* getProjection(childThreadId); + for (const task of subagents) { + if ( + task.runId !== run.id || + task.origin !== "app_owned" || + task.childThreadId === null || + !isOrchestrationV2WorkActive(task.status) + ) + continue; + result = yield* dispatch({ + type: "thread.stop", + commandId: yield* allocateCommandId({}), + threadId: task.childThreadId, + }); + } + return result; + }); let result = settled; for (const run of child.runs) { if (run.status !== "queued" || run.delegatedTaskId !== subagent.id) continue; @@ -843,9 +872,9 @@ export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThrea Effect.flatMap(readRun(run.id), (latest) => latest?.status === "queued" ? Effect.fail(error) - : latest !== undefined && isOrchestrationV2RunActive(latest.status) - ? interruptOwnRun(run.id) - : Effect.succeed(settled), + : latest === undefined + ? Effect.succeed(settled) + : stopOwnRun(latest), ), ), );