fix(server): delegated results no longer trigger a completion alert before the parent resumes

apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts

@@ -1,4 +1,5 @@
 import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
+import { backgroundWorkHoldsCompletion } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
 import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
 import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
 import * as NodeServices from "@effect/platform-node/NodeServices";
@@ -2242,6 +2243,118 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))(
         assert.notEqual(deliveryRun?.status, "queued");
       }),
     );
+    it.effect(
+      "keeps the parent shell unsettled while its delegated completion wake is pending",
+      () =>
+        Effect.gen(function* () {
+          const orchestrator = yield* Orchestrator.OrchestratorV2;
+          const sink = yield* EventSink.EventSinkV2;
+          const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
+          const now = yield* DateTime.now;
+          const threadId = ThreadId.make("thread:pending-wake-shell");
+          yield* seedParentWithTerminalTask({
+            threadId,
+            projectId: ProjectId.make("project:pending-wake-shell"),
+            runId: RunId.make("run:pending-wake-shell-seed"),
+            rootNodeId: NodeId.make("node:pending-wake-shell-seed"),
+            taskId: NodeId.make("node:pending-wake-shell-seed-task"),
+            deliveryState: "acknowledged",
+            parentRunStatus: "completed",
+            now,
+          });
+          const messageId = MessageId.make("message:pending-wake-shell");
+          yield* orchestrator.dispatch({
+            type: "message.dispatch",
+            createdBy: "user",
+            creationSource: "web",
+            commandId: CommandId.make("command:pending-wake-shell-send"),
+            threadId,
+            messageId,
+            text: "Delegate this and end your turn",
+            attachments: [],
+            modelSelection,
+            dispatchMode: { type: "start_immediately" },
+          });
+          const parentRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find(
+            (run) => run.userMessageId === messageId,
+          )!;
+          const runId = parentRun.id;
+          yield* orchestrator.dispatch({
+            type: "delegated_task.request",
+            createdBy: "agent",
+            creationSource: "mcp",
+            commandId: CommandId.make("command:pending-wake-shell-delegate"),
+            parentThreadId: threadId,
+            parentRunId: runId,
+            parentNodeId: parentRun.rootNodeId!,
+            task: "Run in the background",
+            modelSelection,
+            runtimeMode: "full-access",
+            interactionMode: "default",
+            completionWake: "always",
+          });
+          const started = yield* orchestrator.getThreadProjection(threadId);
+          const task = started.subagents.find((row) => row.runId === runId && row.childThreadId)!;
+          yield* sink.write({
+            events: [
+              {
+                ...runEvent({
+                  threadId,
+                  runId,
+                  ordinal: parentRun.ordinal,
+                  status: "completed",
+                  now,
+                }),
+                payload: {
+                  ...started.runs.find((run) => run.id === runId)!,
+                  status: "completed",
+                  completedAt: now,
+                },
+              },
+            ],
+          });
+          assert.isTrue(
+            backgroundWorkHoldsCompletion(
+              (yield* orchestrator.getThreadShell(threadId))?.pendingBackgroundTasks ?? [],
+            ),
+          );
+
+          const childThreadId = task.childThreadId!;
+          const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!;
+          yield* sink.write({
+            events: [
+              {
+                ...runEvent({
+                  threadId: childThreadId,
+                  runId: childRun.id,
+                  ordinal: childRun.ordinal,
+                  status: "completed",
+                  now,
+                }),
+                payload: { ...childRun, status: "completed", completedAt: now },
+              },
+            ],
+          });
+          let wake = yield* requests.take;
+          while (wake.threadId !== threadId) wake = yield* requests.take;
+          const projection = yield* orchestrator.getThreadProjection(threadId);
+          assert.equal(
+            projection.subagents.find((row) => row.id === task.id)?.completionDelivery?.state,
+            "claimed",
+          );
+          assert.isFalse(
+            projection.runs.some(
+              (run) => run.userMessageId === wake.delegatedCompletion?.messageId,
+            ),
+          );
+
+          const shell = yield* orchestrator.getThreadShell(threadId);
+          assert.isTrue(
+            backgroundWorkHoldsCompletion(shell?.pendingBackgroundTasks ?? []),
+            `parent shell reads settled (status ${shell?.status}, no held background work) while its delegated completion wake is still pending`,
+          );
+        }),
+    );
     it.effect("pauses queued follow-up cards and resumes them", () =>
       Effect.gen(function* () {
         const orchestrator = yield* Orchestrator.OrchestratorV2;

apps/server/src/orchestration-v2/NotificationMailbox.ts

@@ -2,7 +2,10 @@ import {
   latestProviderTurnForAttempt,
   type MessageId,
   type OrchestrationV2ConversationMessage,
+  type OrchestrationV2Run,
+  type OrchestrationV2Subagent,
   type OrchestrationV2ThreadProjection,
+  type OrchestrationV2TurnItem,
 } from "@t3tools/contracts";
 
 export function isBufferedProviderNotification(
@@ -35,3 +38,79 @@ export function isUndeliveredMailboxSteer(
         "completed")
   );
 }
+
+type DelegatedCompletionRecords = Pick<
+  OrchestrationV2ThreadProjection,
+  "messages" | "runs" | "providerTurns"
+>;
+
+/** The run's claimed delegated completion still owes a wake: none dispatched, or its steer went unread. */
+export function pendingDelegatedCompletionWake(
+  projection: DelegatedCompletionRecords,
+  run: Pick<OrchestrationV2Run, "delegatedCompletion">,
+) {
+  const delivery = run.delegatedCompletion?.delivery;
+  if (run.delegatedCompletion?.disposition !== "open" || delivery == null) return undefined;
+  const dispatched = projection.messages.some((message) => message.id === delivery.messageId);
+  return dispatched && !isUndeliveredMailboxSteer(projection, delivery.messageId)
+    ? undefined
+    : delivery;
+}
+
+const TERMINAL_TASK_STATUSES: ReadonlyArray<OrchestrationV2Subagent["status"]> = [
+  "completed",
+  "failed",
+  "cancelled",
+  "interrupted",
+];
+
+/**
+ * Finished delegated task items whose completion wake has not started its run: claimed into a
+ * pending wake, or left pending behind a running wake, whose end reserves them a successor. A run
+ * without its provider thread is never offered a wake, so its tasks hold nothing.
+ */
+export function delegatedTaskItemsAwaitingWake<Item extends OrchestrationV2TurnItem>(
+  projection: DelegatedCompletionRecords & {
+    readonly providerThreads: ReadonlyArray<{ readonly id: string }>;
+    readonly subagents: ReadonlyArray<
+      Pick<
+        OrchestrationV2Subagent,
+        "id" | "runId" | "origin" | "status" | "completionWake" | "completionDelivery"
+      >
+    >;
+    readonly turnItems: ReadonlyArray<Item>;
+  },
+): ReadonlyArray<Item> {
+  const taskIds = new Set<string>();
+  for (const run of projection.runs) {
+    if (
+      run.providerThreadId === null ||
+      !projection.providerThreads.some((thread) => thread.id === run.providerThreadId)
+    ) {
+      continue;
+    }
+    for (const taskId of pendingDelegatedCompletionWake(projection, run)?.taskIds ?? []) {
+      taskIds.add(taskId);
+    }
+    if (run.delegatedCompletion?.disposition !== "open" || run.delegatedCompletion.delivery == null)
+      continue;
+    // Mirrors the successor reservation: a live run's blocking wait takes settled_only results.
+    const runIsLive = ["preparing", "starting", "running"].includes(run.status);
+    for (const task of projection.subagents) {
+      if (
+        task.runId === run.id &&
+        task.origin === "app_owned" &&
+        task.completionDelivery?.state === "pending" &&
+        TERMINAL_TASK_STATUSES.includes(task.status) &&
+        (!runIsLive || task.completionWake === "always")
+      ) {
+        taskIds.add(task.id);
+      }
+    }
+  }
+  if (taskIds.size === 0) return [];
+  return projection.turnItems.filter(
+    (item) =>
+      item.type === "subagent" && item.origin === "app_owned" && taskIds.has(item.subagentId),
+  );
+}

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -130,6 +130,7 @@ import {
 import {
   isBufferedProviderNotification,
   isUndeliveredMailboxSteer,
+  pendingDelegatedCompletionWake,
 } from "./NotificationMailbox.ts";
 import { EventSinkV2 } from "./EventSink.ts";
 import {
@@ -1557,18 +1558,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         messageIds: [delivery.messageId],
       });
       const message = messages.messages[0];
-      if (message !== undefined) {
-        if (message.delegatedCompletion === undefined || message.runId === null) return;
-        const runs = yield* projectionStore.getThreadRecords(threadId, ["runs"], {
-          runIds: [message.runId],
-        });
-        const run = runs.runs[0];
-        const turns = yield* projectionStore.getThreadRecords(threadId, ["providerTurns"], {
-          providerTurnAttemptIds: run?.activeAttemptId == null ? [] : [run.activeAttemptId],
-        });
-        if (!isUndeliveredMailboxSteer({ ...messages, ...runs, ...turns }, delivery.messageId))
-          return;
-      }
+      const runs =
+        message?.delegatedCompletion === undefined || message.runId === null
+          ? []
+          : (yield* projectionStore.getThreadRecords(threadId, ["runs"], {
+              runIds: [message.runId],
+            })).runs;
+      const attemptId = runs[0]?.activeAttemptId;
+      const providerTurns =
+        attemptId == null
+          ? []
+          : (yield* projectionStore.getThreadRecords(threadId, ["providerTurns"], {
+              providerTurnAttemptIds: [attemptId],
+            })).providerTurns;
+      if (
+        pendingDelegatedCompletionWake(
+          { messages: messages.messages, runs, providerTurns },
+          parentRun,
+        ) === undefined
+      )
+        return;
       const projection = yield* projectionStore.getThreadRecords(threadId, ["providerThreads"], {
         providerThreadIds: parentRun.providerThreadId === null ? [] : [parentRun.providerThreadId],
       });