fix(orchestration): reconcile held delegated tasks after child completion

apps/server/src/mcp/OrchestratorMcpService.queued-tasks.test.ts

@@ -0,0 +1,234 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { assert, it } from "@effect/vitest";
+import {
+  EnvironmentId,
+  MessageId,
+  NodeId,
+  type OrchestrationV2ThreadProjection,
+  type OrchestrationV2ServerCommand,
+  ProjectId,
+  ProviderInstanceId,
+  RunId,
+  ThreadId,
+} from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+
+import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
+import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
+import * as ProviderRegistry from "../provider/Services/ProviderRegistry.ts";
+import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
+import type { McpInvocationScope } from "./McpInvocationContext.ts";
+import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
+
+const parentThreadId = ThreadId.make("thread:queued-task-parent");
+const childThreadId = ThreadId.make("thread:queued-task-child");
+const taskId = NodeId.make("node:queued-task");
+const now = DateTime.makeUnsafe("2026-10-03T11:00:00Z");
+const providerInstanceId = ProviderInstanceId.make("claude_two");
+const scope: McpInvocationScope = {
+  environmentId: EnvironmentId.make("environment:queued-task"),
+  threadId: parentThreadId,
+  providerSessionId: "provider-session:queued-task",
+  providerInstanceId,
+  capabilities: new Set(["orchestration"]),
+  issuedAt: 1,
+};
+
+function fixture(
+  options: {
+    published?: boolean;
+    executing?: boolean;
+    failCancellation?: boolean;
+    secondQueued?: boolean;
+    deliveryPending?: boolean;
+  } = {},
+) {
+  const projectId = ProjectId.make("project:queued-task");
+  const thread = {
+    id: childThreadId,
+    projectId,
+    title: "Child",
+    modelSelection: { instanceId: providerInstanceId, model: "claude-opus-5-5" },
+    lineage: { parentThreadId, relationshipToParent: "subagent", rootThreadId: parentThreadId },
+    createdAt: now,
+    updatedAt: now,
+    deletedAt: null,
+    archivedAt: null,
+    settledOverride: null,
+    settledAt: null,
+    runtimeMode: "full-access",
+    interactionMode: "default",
+  };
+  const run = (ordinal: number, status: string) => ({
+    id: RunId.make(`run:queued-task:${ordinal}`),
+    threadId: childThreadId,
+    ordinal,
+    status,
+    userMessageId: MessageId.make(`message:queued-task:${ordinal}`),
+    requestedAt: now,
+    startedAt: status === "queued" ? null : now,
+    completedAt: status === "failed" || status === "completed" ? now : null,
+    queueHeld: status === "queued",
+    modelSelection: thread.modelSelection,
+    providerInstanceId,
+  });
+  const child = {
+    thread,
+    runs: [
+      run(1, "failed"),
+      run(2, "queued"),
+      run(3, "completed"),
+      run(4, "completed"),
+      ...(options.executing ? [run(5, "running")] : []),
+      ...(options.secondQueued ? [run(6, "queued")] : []),
+    ],
+    messages: [
+      {
+        id: MessageId.make("result:queued-task:4"),
+        runId: RunId.make("run:queued-task:4"),
+        role: "assistant",
+        text: "Latest completed follow-up result",
+        updatedAt: now,
+      },
+    ],
+    turnItems: [],
+    providerThreads: [{ status: "idle", pendingBackgroundTasks: [] }],
+    runtimeRequests: [],
+    contextTransfers: [],
+    subagents: [],
+  } as unknown as OrchestrationV2ThreadProjection;
+  const parent = {
+    thread: {
+      ...thread,
+      id: parentThreadId,
+      lineage: { parentThreadId: null, relationshipToParent: null },
+    },
+    runs: [],
+    contextTransfers: [],
+    subagents: [
+      {
+        id: taskId,
+        threadId: parentThreadId,
+        childThreadId,
+        origin: "app_owned",
+        status: options.published ? "completed" : "running",
+        providerInstanceId,
+        model: "claude-opus-5-5",
+        result: options.published ? "Published original result" : null,
+        completionDelivery: { state: options.deliveryPending ? "pending" : "disposed" },
+      },
+    ],
+  } as unknown as OrchestrationV2ThreadProjection;
+  const commands: OrchestrationV2ServerCommand[] = [];
+  const layer = OrchestratorMcpService.layer.pipe(
+    Layer.provide(
+      Layer.mergeAll(
+        NodeServices.layer,
+        Layer.mock(ThreadManagementService.ThreadManagementService)({
+          getThreadRecords: (id) => Effect.succeed(id === parentThreadId ? parent : child),
+          getProjectThreadRecords: () => Effect.succeed(child),
+          getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }),
+          dispatch: (command) =>
+            Effect.suspend(() => {
+              commands.push(command);
+              return options.failCancellation && command.type === "queued-run.cancel"
+                ? Effect.fail(new Error("Cancellation failed") as never)
+                : Effect.succeed({} as never);
+            }),
+        }),
+        Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }),
+        Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
+          list: () => Effect.succeed([]),
+        }),
+        Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
+      ),
+    ),
+  );
+  return { child, parent, commands, layer };
+}
+
+it.effect("reports an old held continuation as queued and keeps the latest result readable", () => {
+  const { commands, layer } = fixture();
+  return Effect.gen(function* () {
+    const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+    const task = yield* service.taskStatus(scope, taskId);
+    assert.equal(task.status, "queued");
+    assert.equal(task.workState, "working");
+    assert.equal(task.hasPendingChildRuns, true);
+    assert.equal(task.latestTerminalRunId, RunId.make("run:queued-task:4"));
+    assert.equal(task.latestTerminalSummary, "Latest completed follow-up result");
+    assert.isNull(task.summary);
+    const read = yield* service.readThread(scope, { threadId: childThreadId, runLimit: 1 });
+    assert.equal(read.thread.status, "completed");
+    assert.isNull(read.thread.activeRunId);
+    assert.equal(read.thread.pendingRequestCount, 0);
+    assert.equal(read.thread.queuedRunCount, 1);
+    assert.equal(read.thread.heldQueuedRunCount, 1);
+    assert.equal(read.recentRuns.length, 1);
+    assert.deepEqual(commands, []);
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect("cancels queued-only unpublished work without a provider interruption", () => {
+  const { commands, child, layer } = fixture({ secondQueued: true });
+  return Effect.gen(function* () {
+    const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+    const result = yield* service.cancelTask(scope, { taskId, clientRequestId: "cancel-held" });
+    assert.equal(result.status, "cancel_requested");
+    assert.deepEqual(
+      commands.map((command) => command.type),
+      ["queued-run.cancel"],
+    );
+    const command = commands[0]!;
+    assert.equal(command.type, "queued-run.cancel");
+    if (command.type !== "queued-run.cancel")
+      return yield* Effect.die("Expected queue cancellation");
+    assert.equal(command.threadId, childThreadId);
+    assert.equal(command.runId, child.runs[1]!.id);
+    assert.deepEqual(command.additionalRunIds, [RunId.make("run:queued-task:6")]);
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect("keeps truly executing child work running ahead of held queued work", () => {
+  const { layer } = fixture({ executing: true });
+  return Effect.gen(function* () {
+    const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+    const result = yield* service.taskStatus(scope, taskId);
+    assert.equal(result.status, "running");
+    assert.equal(result.hasPendingChildRuns, true);
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect("does not resurrect a published task or cancel its later backing-thread work", () => {
+  const { commands, layer } = fixture({ published: true, executing: true });
+  return Effect.gen(function* () {
+    const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+    const status = yield* service.taskStatus(scope, taskId);
+    assert.equal(status.status, "completed");
+    assert.equal(status.summary, "Published original result");
+    assert.equal(status.hasPendingChildRuns, true);
+    const cancel = yield* service.cancelTask(scope, {
+      taskId,
+      clientRequestId: "cancel-published",
+    });
+    assert.equal(cancel.status, "completed");
+    assert.deepEqual(commands, []);
+  }).pipe(Effect.provide(layer));
+});
+
+it.effect("does not dispose completion delivery when queue cancellation fails", () => {
+  const { commands, layer } = fixture({ failCancellation: true, deliveryPending: true });
+  return Effect.gen(function* () {
+    const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+    const error = yield* service
+      .cancelTask(scope, { taskId, clientRequestId: "cancel-fails" })
+      .pipe(Effect.flip);
+    assert.equal(error.code, "task_not_cancellable");
+    assert.deepEqual(
+      commands.map((command) => command.type),
+      ["queued-run.cancel"],
+    );
+  }).pipe(Effect.provide(layer));
+});

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -624,6 +624,10 @@ function threadDetail(
     pendingRequestCount: projection.runtimeRequests.filter(
       (request) => request.status === "pending",
     ).length,
+    queuedRunCount: projection.runs.filter((run) => run.status === "queued").length,
+    heldQueuedRunCount: projection.runs.filter(
+      (run) => run.status === "queued" && run.queueHeld === true,
+    ).length,
     archived: projection.thread.archivedAt !== null,
     ...threadSettlement(projection.thread),
     createdAt: DateTime.formatIso(projection.thread.createdAt),
@@ -1068,9 +1072,11 @@ const make = Effect.gen(function* () {
             )
           : workState === "result_available"
             ? taskStatusForRun(progress.resultRun ?? childRun)
-            : taskStatusForRun(childRun) === "queued"
-              ? "queued"
-              : "running";
+            : progress.pendingRun !== undefined
+              ? taskStatusForRun(progress.pendingRun)
+              : taskStatusForRun(childRun) === "queued"
+                ? "queued"
+                : "running";
       const derivedResult =
         task.result !== null
           ? task.result
@@ -1521,32 +1527,59 @@ const make = Effect.gen(function* () {
         }
         const child = yield* loadProjection(current.childThreadId);
         const activeRun = ThreadManagementService.latestActiveRun(child);
-        if (activeRun === undefined) {
+        const queuedRuns = child.runs.filter((run) => run.status === "queued");
+        if (activeRun === undefined && queuedRuns.length === 0) {
           return yield* failure(
             "task_not_cancellable",
-            `Delegated task ${input.taskId} has no interruptible child run.`,
+            `Delegated task ${input.taskId} has no interruptible or queued child run.`,
           );
         }
-        yield* threadManagement
-          .dispatch({
-            type: "run.interrupt",
-            commandId: stableCommandId({
-              scope,
-              requestKey: key,
-              operation: "cancel-task",
-            }),
-            threadId: current.childThreadId,
-            runId: activeRun.id,
-            ...(input.reason === undefined ? {} : { reason: input.reason }),
-          })
-          .pipe(
-            Effect.mapError((error) =>
-              failure(
-                "task_not_cancellable",
-                `Unable to interrupt delegated task ${input.taskId}: ${errorMessage(error)}`,
+        // Cancel unstarted intent atomically so terminal reactions cannot
+        // promote a continuation that this cancellation also owns.
+        const firstQueuedRun = queuedRuns[0];
+        if (firstQueuedRun !== undefined) {
+          yield* threadManagement
+            .dispatch({
+              type: "queued-run.cancel",
+              commandId: stableCommandId({
+                scope,
+                requestKey: key,
+                operation: "cancel-task-queued",
+              }),
+              threadId: current.childThreadId,
+              runId: firstQueuedRun.id,
+              additionalRunIds: queuedRuns.slice(1).map((run) => run.id),
+            })
+            .pipe(
+              Effect.mapError((error) =>
+                failure(
+                  "task_not_cancellable",
+                  `Unable to cancel queued runs for delegated task ${input.taskId}: ${errorMessage(error)}`,
+                ),
               ),
-            ),
-          );
+            );
+        }
+        if (activeRun !== undefined)
+          yield* threadManagement
+            .dispatch({
+              type: "run.interrupt",
+              commandId: stableCommandId({
+                scope,
+                requestKey: key,
+                operation: "cancel-task",
+              }),
+              threadId: current.childThreadId,
+              runId: activeRun.id,
+              ...(input.reason === undefined ? {} : { reason: input.reason }),
+            })
+            .pipe(
+              Effect.mapError((error) =>
+                failure(
+                  "task_not_cancellable",
+                  `Unable to interrupt delegated task ${input.taskId}: ${errorMessage(error)}`,
+                ),
+              ),
+            );
         yield* disposeCompletionDelivery.pipe(
           Effect.catchCause((cause) =>
             Effect.logWarning("orchestrator-mcp.cancel-task.delivery-dispose-failed", {

apps/server/src/mcp/toolkits/orchestrator/tools.ts

@@ -84,7 +84,7 @@ const TaskStatusTool = Tool.make("task_status", {
 
 const TaskCancelTool = Tool.make("task_cancel", {
   description:
-    "Request interruption of an active T3-owned delegated task and dispose its automatic parent delivery. For a terminal task, return its existing status and dispose delivery without interrupting later child-thread runs, even when task_status reports hasPendingChildRuns=true. Published task results remain available. Use t3_thread_interrupt for a later active run.",
+    "Cancel queued continuations and request interruption of an active, unpublished T3-owned delegated task, then dispose its automatic parent delivery. A queued-only task can be cancelled without a live provider run. For a terminal task, return its existing status and dispose delivery without interrupting later child-thread runs, even when task_status reports hasPendingChildRuns=true. Published task results remain available. Use t3_thread_interrupt for a later active run.",
   parameters: OrchestratorMcpTaskCancelInput,
   success: OrchestratorMcpTaskCancelResult,
   failure: OrchestratorMcpFailure,
@@ -173,7 +173,7 @@ const ThreadListTool = Tool.make("t3_thread_list", {
 
 const ThreadReadTool = Tool.make("t3_thread_read", {
   description:
-    "Read durable state and a paginated timeline from a T3 thread in the calling project, or from a thread the user attached to this conversation as context. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units.",
+    "Read durable state and a paginated timeline from a T3 thread in the calling project, or from a thread the user attached to this conversation as context. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. pendingRequestCount counts provider approval/input requests; queuedRunCount and heldQueuedRunCount report unstarted message runs independently of the latest run status. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition. Recover long item text with itemId and textOffset=nextTextOffset until nextTextOffset is null; offsets count UTF-16 code units.",
   parameters: OrchestratorMcpThreadReadInput,
   success: OrchestratorMcpThreadReadResult,
   failure: OrchestratorMcpFailure,