fix: stopping a queued delegated task no longer interrupts another task's live turn

apps/server/src/mcp/OrchestratorMcpService.test.ts

@@ -825,6 +825,150 @@ describe("OrchestratorMcpService", () => {
     }),
   );
 
+  // 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<ReadonlyArray<OrchestrationV2ServerCommand>>([]);
+    const parentProjection = {
+      thread: { id: parentThreadId },
+      runs: [],
+      providerTurns: [],
+      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 = (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: [],
+        providerTurns: [],
+        runtimeRequests: [],
+      }) as unknown as OrchestrationV2ThreadProjection;
+    const child = yield* Ref.make(childProjection(["running", "queued", "queued"]));
+    const descendantStops = yield* Ref.make<ReadonlyArray<RunId | undefined>>([]);
+    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: (stop) => Ref.update(descendantStops, (runs) => [...runs, stop.runId]),
+        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,
+      descendantStops: yield* Ref.get(descendantStops),
+      commands: (yield* Ref.get(dispatched)).map((command) => ({
+        type: command.type,
+        runId: "runId" in command ? command.runId : undefined,
+      })),
+    };
+  });
+
+  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 },
+      ]);
+      // Only the tasks the promoted turn delegated, not the other task's.
+      assert.deepEqual(cancelled.descendantStops, [cancelled.runId]);
+    }),
+  );
+
+  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 },
+      ]);
+      // Only the tasks the promoted turn delegated, not the other task's.
+      assert.deepEqual(cancelled.descendantStops, [cancelled.runId]);
+    }),
+  );
+
   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");

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -63,6 +63,7 @@ import {
   type ServerProviderModel,
   ThreadId,
   isOrchestrationV2WorkActive,
+  delegatedTaskYieldsLiveRun,
   latestProviderTurnForAttempt,
   resolveProjectProviderInstanceEnabled,
   SECRET_REQUEST_UNSAVED_NOTE,
@@ -2642,36 +2643,92 @@ 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 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, 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,
+              commandId: runCommandId,
+              reason,
+              runId: run.id,
+            })
+            .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)}`,
+            );
+        });
+        // 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(runs, input.taskId)) {
+          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 stopped = yield* threadManagement
-          .getThreadRecords(childThreadId, ["runs"])
-          .pipe(Effect.mapError(threadManagementFailure));
-        for (const run of stopped.runs) {
+            );
+          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)}`,
+                ),
+              ),
+            );
+        }
+        for (const run of runs) {
           if (
             run.status !== "queued" ||
             (run.delegatedTaskId !== input.taskId && run.id !== current.childRunId)
@@ -2690,7 +2747,19 @@ 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.
+              Effect.catch((error) =>
+                Effect.flatMap(readRun(run.id), (latest) =>
+                  latest?.status === "queued"
+                    ? Effect.fail(error)
+                    : latest === undefined
+                      ? Effect.void
+                      : stopOwnRun(latest),
+                ),
+              ),
+            );
         }
       });
       // Checked above; a user raising one of these threads meanwhile is
@@ -2741,7 +2810,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;