# Files changed · PR \#20928 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/20928/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[fix\(server\): notify parents when delegated children need approval](https://git.cafe/pingdotgg/t3code/pull/20928.md)

Version: 12

Head commit: 8aadcc59a93e1a59936b7ae3a936d3fcaf41f4c6

Base commit: 4040867bb46a90a8f41c687d7b454b5ece97a34f

Comparison base: 4040867bb46a90a8f41c687d7b454b5ece97a34f

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

## apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts

Status: deleted; +0; −182

```diff
@@ -1,182 +1,0 @@
-import {
-  CommandId,
-  MessageId,
-  type OrchestrationV2DomainEvent,
-  type OrchestrationV2RuntimeRequest,
-  type OrchestrationV2Subagent,
-} from "@t3tools/contracts";
-import * as Context from "effect/Context";
-import * as Effect from "effect/Effect";
-import * as Layer from "effect/Layer";
-import type * as Scope from "effect/Scope";
-import * as Stream from "effect/Stream";
-
-import { reportLabel } from "@t3tools/provider-core/server/notification";
-
-import { forkParked } from "../serverActivation.ts";
-import * as Orchestrator from "./Orchestrator.ts";
-import * as ProjectionStore from "./ProjectionStore.ts";
-
-export class DelegatedTaskQuestionReactor extends Context.Service<
-  DelegatedTaskQuestionReactor,
-  {
-    readonly start: () => Effect.Effect<void, never, Scope.Scope>;
-    readonly handle: (event: OrchestrationV2DomainEvent) => Effect.Effect<void>;
-  }
->()("t3/orchestration-v2/DelegatedTaskQuestionReactor") {}
-
-const make = Effect.gen(function* () {
-  const engine = yield* Orchestrator.OrchestratorV2;
-  const projections = yield* ProjectionStore.ProjectionStoreV2;
-
-  const wakeParent = Effect.fnUntraced(function* (
-    task: OrchestrationV2Subagent,
-    request: OrchestrationV2RuntimeRequest,
-  ) {
-    if (task.childThreadId === null) return;
-    const key = `${task.threadId}:${request.id}`;
-    const label = reportLabel(task.title?.trim() || task.prompt);
-    const situation =
-      request.responseCapability.type === "message"
-        ? `Delegated task ${task.id} asked a question and is still working.`
-        : `Delegated task ${task.id} is blocked on a question and cannot continue until it is answered.`;
-    yield* engine.dispatch({
-      type: "message.dispatch",
-      commandId: CommandId.make(`server:delegated-task-question:${key}`),
-      threadId: task.threadId,
-      messageId: MessageId.make(`message:delegated-task-question:${key}`),
-      text: `${situation} Read it with t3_pending_request_read using threadId ${task.childThreadId} and requestId ${request.id}. Answer it with t3_pending_request_respond when the answer follows from the task or the user's instructions; otherwise ask the user, then answer it. If it is no longer pending, it was answered or dismissed.`,
-      notification: {
-        source: { kind: "delegated_task", taskIds: [task.id], childThreadId: task.childThreadId },
-        outcome: "updated",
-        summary: `Delegated task${label === undefined ? "" : ` "${label}"`} has a question`,
-      },
-      attachments: [],
-      dispatchMode: { type: "queue_after_active" },
-      createdBy: "agent",
-      creationSource: "server",
-    });
-  });
-
-  // Runs for every domain event; most events return before any work is needed.
-  const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = Effect.fnUntraced(
-    function* (event) {
-      if (event.type === "run.updated") {
-        if (!["completed", "failed", "waiting"].includes(event.payload.status)) return;
-        const parent = yield* projections.getThreadRecords(event.threadId, ["subagents"]);
-        if (parent.thread.archivedAt !== null || parent.thread.deletedAt !== null) return;
-        for (const task of parent.subagents) {
-          if (
-            task.origin !== "app_owned" ||
-            task.runId !== event.payload.id ||
-            task.completionWake !== "settled_only" ||
-            task.completionDelivery?.state === "disposed" ||
-            task.childThreadId === null ||
-            ["completed", "failed", "cancelled", "interrupted"].includes(task.status)
-          ) {
-            continue;
-          }
-          const child = yield* projections.getThreadRecords(task.childThreadId, [
-            "runtimeRequests",
-          ]);
-          for (const request of child.runtimeRequests) {
-            if (
-              request.kind !== "user_input" ||
-              request.status !== "pending" ||
-              request.responseCapability.type !== "live"
-            ) {
-              continue;
-            }
-            yield* wakeParent(task, request).pipe(
-              Effect.catchCause((cause) =>
-                Effect.logWarning("catching up a delegated task question failed", {
-                  taskId: task.id,
-                  requestId: request.id,
-                  cause,
-                }),
-              ),
-            );
-          }
-        }
-        return;
-      }
-      if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") {
-        return;
-      }
-      const child = yield* projections.getThread(event.threadId);
-      const parentThreadId = child.lineage.parentThreadId;
-      if (parentThreadId === null) return;
-      const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id);
-      const key = `${parentThreadId}:${event.payload.id}`;
-      const messageId = MessageId.make(`message:delegated-task-question:${key}`);
-      if (event.payload.status !== "pending" || request?.status !== "pending") {
-        const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
-        const queued = parent.runs.find(
-          (run) => run.status === "queued" && run.userMessageId === messageId,
-        );
-        if (queued !== undefined) {
-          yield* engine.dispatch({
-            type: "queued-run.cancel",
-            commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`),
-            threadId: parentThreadId,
-            runId: queued.id,
-          });
-        }
-        return;
-      }
-      const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]);
-      const { runs } = yield* projections.getThreadRecords(child.id, ["runs"]);
-      const run =
-        runs.find((run) => run.id === event.runId) ??
-        runs.toSorted((a, b) => b.ordinal - a.ordinal)[0];
-      const taskId =
-        run?.delegatedTaskId ??
-        (child.forkedFrom?.type === "node" ? child.forkedFrom.nodeId : undefined);
-      const task = parent.subagents.find(
-        (candidate) => candidate.origin === "app_owned" && candidate.id === taskId,
-      );
-      if (
-        task === undefined ||
-        parent.thread.archivedAt !== null ||
-        parent.thread.deletedAt !== null
-      ) {
-        return;
-      }
-      // A blocking delegate_task wait delivers the question itself.
-      if (
-        task.completionWake === "settled_only" &&
-        event.payload.responseCapability.type === "live" &&
-        parent.runs.some(
-          (run) =>
-            run.id === task.runId &&
-            (run.status === "preparing" || run.status === "starting" || run.status === "running"),
-        )
-      ) {
-        return;
-      }
-      yield* wakeParent(task, request);
-    },
-    (effect, event) =>
-      effect.pipe(
-        Effect.catchCause((cause) =>
-          Effect.logWarning("waking the parent of a delegated task question failed", {
-            threadId: event.threadId,
-            cause,
-          }),
-        ),
-      ),
-  );
-
-  const start: DelegatedTaskQuestionReactor["Service"]["start"] = () =>
-    forkParked(
-      Stream.runForEach(engine.streamDomainEvents, handle).pipe(
-        Effect.catchCause((cause) =>
-          Effect.logWarning("delegated task question event stream failed", { cause }),
-        ),
-      ),
-    );
-
-  return { start, handle } satisfies DelegatedTaskQuestionReactor["Service"];
-});
-
-export const layer = Layer.effect(DelegatedTaskQuestionReactor, make);
```

## apps/server/src/orchestration-v2/DelegatedTaskRequestReactor.ts

Status: added; +209; −0

```diff
@@ -1,0 +1,209 @@
+import {
+  CommandId,
+  MessageId,
+  ProviderRequestKind,
+  type OrchestrationV2DomainEvent,
+  type OrchestrationV2RuntimeRequest,
+  type OrchestrationV2Subagent,
+} from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+import type * as Scope from "effect/Scope";
+import * as Stream from "effect/Stream";
+
+import { reportLabel } from "@t3tools/provider-core/server/notification";
+
+import { forkParked } from "../serverActivation.ts";
+import * as Orchestrator from "./Orchestrator.ts";
+import * as ProjectionStore from "./ProjectionStore.ts";
+
+export class DelegatedTaskRequestReactor extends Context.Service<
+  DelegatedTaskRequestReactor,
+  {
+    readonly start: () => Effect.Effect<void, never, Scope.Scope>;
+    readonly handle: (event: OrchestrationV2DomainEvent) => Effect.Effect<void>;
+  }
+>()("t3/orchestration-v2/DelegatedTaskRequestReactor") {}
+
+const isApprovalRequestKind = Schema.is(ProviderRequestKind);
+
+const isRoutedRequestKind = (kind: OrchestrationV2RuntimeRequest["kind"]) =>
+  kind === "user_input" || isApprovalRequestKind(kind);
+
+// Questions can be answered by the parent; approvals need a live callback and only a human can decide them.
+const routesToParent = (
+  request: Pick<OrchestrationV2RuntimeRequest, "kind" | "responseCapability">,
+) =>
+  request.kind === "user_input" ||
+  (isApprovalRequestKind(request.kind) && request.responseCapability.type === "live");
+
+const isTerminalTask = (task: OrchestrationV2Subagent) =>
+  ["completed", "failed", "cancelled", "interrupted"].includes(task.status);
+
+const make = Effect.gen(function* () {
+  const engine = yield* Orchestrator.OrchestratorV2;
+  const projections = yield* ProjectionStore.ProjectionStoreV2;
+
+  const wakeParent = Effect.fnUntraced(function* (
+    task: OrchestrationV2Subagent,
+    request: OrchestrationV2RuntimeRequest,
+  ) {
+    if (task.childThreadId === null) return;
+    const key = `${task.threadId}:${request.id}`;
+    const label = reportLabel(task.title?.trim() || task.prompt);
+    const question = request.kind === "user_input";
+    const situation = question
+      ? request.responseCapability.type === "message"
+        ? `Delegated task ${task.id} asked a question and is still working.`
+        : `Delegated task ${task.id} is blocked on a question and cannot continue until it is answered.`
+      : `Delegated task ${task.id} needs the user's approval.`;
+    yield* engine.dispatch({
+      type: "message.dispatch",
+      commandId: CommandId.make(`server:delegated-task-question:${key}`),
+      threadId: task.threadId,
+      messageId: MessageId.make(`message:delegated-task-question:${key}`),
+      text: !question
+        ? `${situation} Ask the user to open [the child thread](t3-thread://v1/${task.childThreadId}) and handle request ${request.id} there. Only the user can approve; do not answer this request with t3_pending_request_respond. If it is no longer pending, no action is needed.`
+        : `${situation} Read it with t3_pending_request_read using threadId ${task.childThreadId} and requestId ${request.id}. Answer it with t3_pending_request_respond when the answer follows from the task or the user's instructions; otherwise ask the user, then answer it. If it is no longer pending, it was answered or dismissed.`,
+      notification: {
+        source: { kind: "delegated_task", taskIds: [task.id], childThreadId: task.childThreadId },
+        outcome: "updated",
+        summary: `Delegated task${label === undefined ? "" : ` "${label}"`} ${question ? "has a question" : "needs approval"}`,
+      },
+      attachments: [],
+      dispatchMode: { type: "queue_after_active" },
+      createdBy: "agent",
+      creationSource: "server",
+    });
+  });
+
+  // Runs for every domain event; most events return before any work is needed.
+  const handle: DelegatedTaskRequestReactor["Service"]["handle"] = Effect.fnUntraced(
+    function* (event) {
+      if (event.type === "run.updated") {
+        if (!["completed", "failed", "waiting"].includes(event.payload.status)) return;
+        const parent = yield* projections.getThreadRecords(event.threadId, ["subagents"]);
+        if (parent.thread.archivedAt !== null || parent.thread.deletedAt !== null) return;
+        for (const task of parent.subagents) {
+          if (
+            task.origin !== "app_owned" ||
+            task.runId !== event.payload.id ||
+            task.completionDelivery?.state === "disposed" ||
+            task.childThreadId === null ||
+            isTerminalTask(task)
+          ) {
+            continue;
+          }
+          const child = yield* projections.getThreadRecords(task.childThreadId, [
+            "runtimeRequests",
+          ]);
+          for (const request of child.runtimeRequests) {
+            // An incomplete wait upgrades the task to "always" after returning the request itself.
+            if (
+              !routesToParent(request) ||
+              request.status !== "pending" ||
+              task.completionWake !== "settled_only" ||
+              request.responseCapability.type !== "live"
+            ) {
+              continue;
+            }
+            yield* wakeParent(task, request).pipe(
+              Effect.catchCause((cause) =>
+                Effect.logWarning("catching up a delegated task request failed", {
+                  taskId: task.id,
+                  requestId: request.id,
+                  cause,
+                }),
+              ),
+            );
+          }
+        }
+        return;
+      }
+      // Withdrawal must not depend on capability: interruption ends a live approval as not_resumable.
+      if (event.type !== "runtime-request.updated" || !isRoutedRequestKind(event.payload.kind)) {
+        return;
+      }
+      const child = yield* projections.getThread(event.threadId);
+      const parentThreadId = child.lineage.parentThreadId;
+      if (parentThreadId === null) return;
+      const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id);
+      const key = `${parentThreadId}:${event.payload.id}`;
+      const messageId = MessageId.make(`message:delegated-task-question:${key}`);
+      if (event.payload.status !== "pending" || request?.status !== "pending") {
+        const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
+        const queued = parent.runs.find(
+          (run) => run.status === "queued" && run.userMessageId === messageId,
+        );
+        if (queued !== undefined) {
+          yield* engine.dispatch({
+            type: "queued-run.cancel",
+            commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`),
+            threadId: parentThreadId,
+            runId: queued.id,
+          });
+        }
+        return;
+      }
+      if (!routesToParent(request)) return;
+      const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]);
+      const { runs } = yield* projections.getThreadRecords(child.id, ["runs"]);
+      const run =
+        runs.find((run) => run.id === event.runId) ??
+        runs.toSorted((a, b) => b.ordinal - a.ordinal)[0];
+      const taskId =
+        run?.delegatedTaskId ??
+        (child.forkedFrom?.type === "node" ? child.forkedFrom.nodeId : undefined);
+      const task = parent.subagents.find(
+        (candidate) => candidate.origin === "app_owned" && candidate.id === taskId,
+      );
+      if (
+        task === undefined ||
+        task.childThreadId !== child.id ||
+        task.completionDelivery?.state === "disposed" ||
+        isTerminalTask(task) ||
+        parent.thread.archivedAt !== null ||
+        parent.thread.deletedAt !== null
+      ) {
+        return;
+      }
+      // A blocking delegate_task wait delivers live questions and approvals itself.
+      if (
+        task.completionWake === "settled_only" &&
+        (request.kind !== "user_input" || request.responseCapability.type === "live") &&
+        parent.runs.some(
+          (run) =>
+            run.id === task.runId &&
+            (run.status === "preparing" || run.status === "starting" || run.status === "running"),
+        )
+      ) {
+        return;
+      }
+      yield* wakeParent(task, request);
+    },
+    (effect, event) =>
+      effect.pipe(
+        Effect.catchCause((cause) =>
+          Effect.logWarning("waking the parent of a delegated task request failed", {
+            threadId: event.threadId,
+            cause,
+          }),
+        ),
+      ),
+  );
+
+  const start: DelegatedTaskRequestReactor["Service"]["start"] = () =>
+    forkParked(
+      Stream.runForEach(engine.streamDomainEvents, handle).pipe(
+        Effect.catchCause((cause) =>
+          Effect.logWarning("delegated task request event stream failed", { cause }),
+        ),
+      ),
+    );
+
+  return { start, handle } satisfies DelegatedTaskRequestReactor["Service"];
+});
+
+export const layer = Layer.effect(DelegatedTaskRequestReactor, make);
```

## apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts

Status: modified; +287; −3

```diff
@@ -27,7 +27,7 @@
 import * as SqlClient from "effect/sql/SqlClient";
 import * as SqlitePersistence from "../persistence/Sqlite.ts";
 import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
-import * as DelegatedTaskQuestionReactor from "./DelegatedTaskQuestionReactor.ts";
+import * as DelegatedTaskRequestReactor from "./DelegatedTaskRequestReactor.ts";
 import * as EffectOutbox from "./EffectOutbox.ts";
 import * as EventSink from "./EventSink.ts";
 import * as Orchestrator from "./Orchestrator.ts";
@@ -743,7 +743,7 @@
   Effect.gen(function* () {
     const orchestrator = yield* Orchestrator.OrchestratorV2;
     const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const reactor = yield* DelegatedTaskRequestReactor.DelegatedTaskRequestReactor;
-    const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor;
     const parentThreadId = ThreadId.make("thread:question-parent");
     yield* orchestrator.dispatch({
       type: "thread.create",
@@ -1069,9 +1069,293 @@
       taskIds: [followup.delegatedTaskId],
       childThreadId: task.childThreadId!,
     });
+  }).pipe(Effect.provide(DelegatedTaskRequestReactor.layer.pipe(Layer.provideMerge(layerTest)))),
+);
+
+const setupDelegatedApproval = Effect.fnUntraced(function* (
+  completionWake: "always" | "settled_only" = "always",
+) {
+  const engine = yield* Orchestrator.OrchestratorV2;
+  const projections = yield* ProjectionStore.ProjectionStoreV2;
+  const reactor = yield* DelegatedTaskRequestReactor.DelegatedTaskRequestReactor;
+  const sink = yield* EventSink.EventSinkV2;
+  const parentThreadId = ThreadId.make("thread:human-parent:approval");
+  yield* engine.dispatch({
+    type: "thread.create",
+    commandId: CommandId.make("human:create:approval"),
+    threadId: parentThreadId,
+    projectId: ProjectId.make("project:human-requests"),
+    title: "Parent",
+    modelSelection,
+    runtimeMode: "full-access",
+    interactionMode: "default",
+    branch: null,
+    worktreePath: null,
+    createdBy: "user",
+    creationSource: "web",
+  });
+  yield* engine.dispatch({
+    type: "message.dispatch",
+    commandId: CommandId.make("human:start:approval"),
+    threadId: parentThreadId,
+    messageId: MessageId.make("human:message:approval"),
+    text: "Delegate a task",
+    attachments: [],
+    dispatchMode: { type: "start_immediately" },
+    createdBy: "user",
+    creationSource: "web",
+  });
+  const parentRun = (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs[0]!;
+  yield* engine.dispatch({
+    type: "delegated_task.request",
+    commandId: CommandId.make("human:delegate:approval"),
+    parentThreadId,
+    parentRunId: parentRun.id,
+    parentNodeId: parentRun.rootNodeId!,
+    task: "Run the checks",
+    completionWake,
+    modelSelection,
+    runtimeMode: "approval-required",
+    interactionMode: "default",
+    createdBy: "agent",
+    creationSource: "mcp",
+  });
+  const task = (yield* projections.getThreadRecords(parentThreadId, ["subagents"])).subagents[0]!;
+  const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!;
+  const now = yield* DateTime.now;
+  const parentEnded = {
+    id: EventId.make("human:parent-ended:approval"),
+    type: "run.updated",
+    threadId: parentThreadId,
+    runId: parentRun.id,
+    occurredAt: now,
+    payload: { ...parentRun, status: "completed", completedAt: now },
+  } satisfies OrchestrationV2DomainEvent;
+  const request = (
+    kind: "command" | "permission" | "mcp-elicitation" | "auth_refresh" | "dynamic_tool_call",
+  ) => ({
+    id: EventId.make("approval:request:" + kind),
+    type: "runtime-request.updated" as const,
+    threadId: task.childThreadId!,
+    runId: childRun.id,
+    occurredAt: now,
+    payload: {
+      id: RuntimeRequestId.make("approval:request:" + kind),
+      nodeId: childRun.rootNodeId!,
+      providerTurnId: null,
-  }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))),
+      nativeRequestRef: null,
+      kind,
+      status: "pending" as const,
+      responseCapability: {
+        type: "live" as const,
+        providerSessionId: ProviderSessionId.make("approval:session"),
+      },
+      createdAt: now,
+      resolvedAt: null,
+    },
+  });
+  return { engine, projections, reactor, sink, parentThreadId, task, parentEnded, request };
+});
+
+const layerDelegatedApproval = DelegatedTaskRequestReactor.layer.pipe(
+  Layer.provideMerge(layerTest),
 );
 
+it.effect.each(["command", "mcp-elicitation"] as const)(
+  "wakes an idle parent once for %s approval and directs the human to the child",
+  (kind) =>
+    Effect.gen(function* () {
+      const { projections, reactor, sink, parentThreadId, task, parentEnded, request } =
+        yield* setupDelegatedApproval();
+      const approval = request(kind);
+      const requests = [approval, request("auth_refresh"), request("dynamic_tool_call")];
+      yield* sink.write({ events: [parentEnded, ...requests] });
+      for (const event of requests) yield* reactor.handle(event);
+      yield* reactor.handle(approval);
+      const parent = yield* projections.getThreadRecords(parentThreadId, [
+        "messages",
+        "runs",
+        "subagents",
+      ]);
+      const notices = parent.messages.filter((message) => message.notification !== undefined);
+      assert.equal(notices.length, 1);
+      const notice = notices[0]!;
+      assert.include(notice.text, "t3-thread://v1/" + task.childThreadId);
+      assert.include(notice.text, approval.payload.id);
+      assert.include(notice.text, "Only the user can approve");
+      assert.include(notice.text, "do not answer this request with t3_pending_request_respond");
+      assert.deepEqual(notice.notification, {
+        source: { kind: "delegated_task", taskIds: [task.id], childThreadId: task.childThreadId! },
+        outcome: "updated",
+        summary: 'Delegated task "Run the checks" needs approval',
+      });
+      assert.equal(parent.runs.find((run) => run.userMessageId === notice.id)?.status, "starting");
+      assert.equal(parent.subagents[0]?.status, "running");
+      assert.isNull(parent.subagents[0]?.result);
+    }).pipe(Effect.provide(layerDelegatedApproval)),
+);
+
+it.effect.each(["stop", "dispose"] as const)(
+  "withdraws resolved approval notices and fences queued and late notices after %s",
+  (action) =>
+    Effect.gen(function* () {
+      const { engine, projections, reactor, sink, parentThreadId, task, request } =
+        yield* setupDelegatedApproval();
+      const approval = request("command");
+      const pending = request("permission");
+      yield* sink.write({ events: [approval, pending] });
+      yield* reactor.handle(approval);
+      yield* reactor.handle(pending);
+      const parent = yield* projections.getThreadRecords(parentThreadId, ["messages", "runs"]);
+      const notices = parent.messages.filter((message) => message.notification !== undefined);
+      assert.equal(notices.length, 2);
+      const noticeRuns = parent.runs.filter((run) =>
+        notices.some((notice) => notice.id === run.userMessageId),
+      );
+      assert.deepEqual(
+        noticeRuns.map((run) => run.status),
+        ["queued", "queued"],
+      );
+      const resolved = {
+        ...approval,
+        id: EventId.make("approval:resolved"),
+        payload: {
+          ...approval.payload,
+          status: "resolved" as const,
+          resolvedAt: approval.occurredAt,
+        },
+      };
+      yield* sink.write({ events: [resolved] });
+      yield* reactor.handle(resolved);
+      const afterResolution = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
+      assert.deepEqual(
+        noticeRuns.map(
+          (noticeRun) => afterResolution.runs.find((run) => run.id === noticeRun.id)?.status,
+        ),
+        ["cancelled", "queued"],
+      );
+      yield* engine.dispatch(
+        action === "stop"
+          ? {
+              type: "thread.stop",
+              commandId: CommandId.make("approval:stop"),
+              threadId: parentThreadId,
+            }
+          : {
+              type: "delegated_task.completion-delivery.dispose",
+              commandId: CommandId.make("approval:dispose"),
+              parentThreadId,
+              taskId: task.id,
+            },
+      );
+      const afterCancellation = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
+      assert.deepEqual(
+        noticeRuns.map(
+          (noticeRun) => afterCancellation.runs.find((run) => run.id === noticeRun.id)?.status,
+        ),
+        ["cancelled", "cancelled"],
+      );
+      // A reactor that read the task before cancellation can still reach dispatch afterwards.
+      const refused = yield* Effect.exit(
+        engine.dispatch({
+          type: "message.dispatch",
+          commandId: CommandId.make("approval:late"),
+          threadId: parentThreadId,
+          messageId: MessageId.make("approval:late"),
+          text: notices[1]!.text,
+          notification: notices[1]!.notification!,
+          attachments: [],
+          dispatchMode: { type: "queue_after_active" },
+          createdBy: "agent",
+          creationSource: "server",
+        }),
+      );
+      assert.equal(refused._tag, "Failure");
+      assert.equal(
+        (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.length,
+        parent.messages.length,
+      );
+    }).pipe(Effect.provide(layerDelegatedApproval)),
+);
+
+it.effect.each(["cancelled", "expired"] as const)(
+  "withdraws a queued approval notice when the request is %s without a live callback",
+  (status) =>
+    Effect.gen(function* () {
+      const { projections, reactor, sink, parentThreadId, request } =
+        yield* setupDelegatedApproval();
+      const approval = request("command");
+      yield* sink.write({ events: [approval] });
+      yield* reactor.handle(approval);
+      const ended = {
+        ...approval,
+        id: EventId.make("approval:ended"),
+        payload: {
+          ...approval.payload,
+          status,
+          resolvedAt: approval.occurredAt,
+          responseCapability: {
+            type: "not_resumable" as const,
+            reason: "The run was interrupted.",
+          },
+        },
+      };
+      yield* sink.write({ events: [ended] });
+      yield* reactor.handle(ended);
+      const parent = yield* projections.getThreadRecords(parentThreadId, ["messages", "runs"]);
+      const notice = parent.messages.find((message) => message.notification !== undefined);
+      assert.equal(
+        parent.runs.find((run) => run.userMessageId === notice?.id)?.status,
+        "cancelled",
+      );
+    }).pipe(Effect.provide(layerDelegatedApproval)),
+);
+
+it.effect.each([false, true])(
+  "catches up a blocking delegation approval at run end unless the wait returned it (returned: %s)",
+  (waitReturned) =>
+    Effect.gen(function* () {
+      const { engine, projections, reactor, sink, parentThreadId, task, parentEnded, request } =
+        yield* setupDelegatedApproval("settled_only");
+      const approval = request("permission");
+      yield* sink.write({ events: [approval] });
+      yield* reactor.handle(approval);
+      assert.isFalse(
+        (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.some(
+          (message) => message.notification !== undefined,
+        ),
+      );
+      if (waitReturned) {
+        yield* engine.dispatch({
+          type: "delegated_task.wake-policy",
+          commandId: CommandId.make("approval:wait-ended"),
+          parentThreadId,
+          taskId: task.id,
+          completionWake: "always",
+        });
+      }
+      yield* sink.write({ events: [parentEnded] });
+      yield* reactor.handle(parentEnded);
+      yield* reactor.handle(parentEnded);
+      const parent = yield* projections.getThreadRecords(parentThreadId, [
+        "messages",
+        "runs",
+        "subagents",
+      ]);
+      const notices = parent.messages.filter((message) => message.notification !== undefined);
+      assert.equal(notices.length, waitReturned ? 0 : 1);
+      if (!waitReturned) {
+        assert.include(notices[0]!.text, approval.payload.id);
+        assert.equal(
+          parent.runs.find((run) => run.userMessageId === notices[0]!.id)?.status,
+          "starting",
+        );
+      }
+      assert.equal(parent.subagents[0]?.status, "running");
+      assert.isNull(parent.subagents[0]?.result);
+    }).pipe(Effect.provide(layerDelegatedApproval)),
+);
+
 it.effect.each([false, true])("rejects thread.create for an existing id (deleted: %s)", (deleted) =>
   Effect.gen(function* () {
     const orchestrator = yield* Orchestrator.OrchestratorV2;
```

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

Status: modified; +70; −0

```diff
@@ -2139,6 +2139,38 @@
     ),
   );
 
+  // The enclosing dispatch span owns this cancellation work; a helper span would duplicate it.
+  const cancelQueuedDelegatedRequests = Effect.fnUntraced(function* (input: {
+    readonly command: OrchestrationV2ServerCommand;
+    readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>;
+    readonly projection: Pick<
+      OrchestrationV2ThreadProjection,
+      "runs" | "messages" | "nodes" | "attempts"
+    >;
+    readonly taskIds: ReadonlyArray<NodeId>;
+    readonly now: DateTime.Utc;
+  }) {
+    const messageIds = new Set(
+      input.projection.messages
+        .filter((message) => {
+          const notice = message.notification;
+          return (
+            message.delegatedCompletion === undefined &&
+            notice?.source.kind === "delegated_task" &&
+            notice.source.childThreadId !== undefined &&
+            notice.outcome === "updated" &&
+            notice.source.taskIds.some((taskId) => input.taskIds.includes(taskId))
+          );
+        })
+        .map((message) => message.id),
+    );
+    for (const run of input.projection.runs) {
+      if (run.status === "queued" && messageIds.has(run.userMessageId)) {
+        yield* emitQueuedRunCancellation({ ...input, run });
+      }
+    }
+  });
+
   const dispatchDelegatedTaskCompletionDeliveryResolution = Effect.fn(
     "Orchestrator.dispatchDelegatedTaskCompletionDeliveryResolution",
   )(function* (
@@ -2204,6 +2236,15 @@
       });
       return;
     }
+    if (state === "disposed") {
+      yield* cancelQueuedDelegatedRequests({
+        command,
+        events,
+        projection,
+        taskIds: [task.id],
+        now,
+      });
+    }
     const updatedTask: OrchestrationV2Subagent = {
       ...task,
       completionDelivery: {
@@ -2331,6 +2372,7 @@
     if (cohort === undefined && tasks.length === 0) {
       return;
     }
+    yield* cancelQueuedDelegatedRequests({ ...input, taskIds: tasks.map((task) => task.id) });
     const emitEvent = emit(input.events, input.command);
     const nextDisposition = cohort?.disposition === "disposed" ? "disposed" : input.disposition;
     const nextCohort = {
@@ -4806,6 +4848,34 @@
     effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
   ) {
     let projection = yield* getProjectionWithPendingEvents(command.threadId, events);
+    const notice = command.notification;
+    if (
+      command.delegatedCompletion === undefined &&
+      notice?.source.kind === "delegated_task" &&
+      notice.source.childThreadId !== undefined &&
+      notice.outcome === "updated"
+    ) {
+      const source = notice.source;
+      // Stop and disposal can win the parent lock after the reactor reads the task.
+      if (
+        !source.taskIds.every((taskId) =>
+          projection.subagents.some(
+            (task) =>
+              task.id === taskId &&
+              task.origin === "app_owned" &&
+              task.childThreadId === source.childThreadId &&
+              task.completionDelivery?.state !== "disposed" &&
+              !isTerminalDelegatedTaskStatus(task.status),
+          ),
+        )
+      ) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: "The delegated task no longer accepts request notices.",
+        });
+      }
+    }
     if (command.manualContinuationOfRunId !== undefined) {
       const source = projection.runs.find((run) => run.id === command.manualContinuationOfRunId);
       const limited = latestRootProviderFailure(source ?? null, projection.turnItems);
```

## apps/server/src/server.ts

Status: modified; +3; −3

```diff
@@ -5,7 +5,7 @@
 import * as StorageCleanup from "./storageCleanup.ts";
 import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts";
 import * as PullRequestWatchReactor from "./orchestration-v2/PullRequestWatchReactor.ts";
-import * as DelegatedTaskQuestionReactor from "./orchestration-v2/DelegatedTaskQuestionReactor.ts";
+import * as DelegatedTaskRequestReactor from "./orchestration-v2/DelegatedTaskRequestReactor.ts";
 // Effect wraps this raw Node HTTP server after its startup and write-error guards attach.
 // @effect-diagnostics nodeBuiltinImport:off
 import * as NodeHttp from "node:http";
@@ -594,10 +594,10 @@
   ),
   Layer.effectDiscard(
     Effect.gen(function* () {
-      const service = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor;
+      const service = yield* DelegatedTaskRequestReactor.DelegatedTaskRequestReactor;
       yield* service.start();
     }),
+  ).pipe(Layer.provide(DelegatedTaskRequestReactor.layer), Layer.provide(ProjectionStoreV2.layer)),
-  ).pipe(Layer.provide(DelegatedTaskQuestionReactor.layer), Layer.provide(ProjectionStoreV2.layer)),
   ProviderUsageLimitsIngestion.layer,
   layerProviderInstallationRefresh,
   ReplayMarkers.layer,
```

## docs/orchestration-v2/orchestrator-mcp-server.md

Status: modified; +13; −7

```diff
@@ -481,13 +481,19 @@
 a notification message telling it to answer with the pending request tools or
 ask the user. The message says whether the child is blocked or, for a Codex
 async question, still working. If the question is answered or dismissed before
-the notification starts a turn, that queued turn is cancelled. Approvals are not
-routed, because those tools cannot answer them.
-A parent still blocked in a `delegate_task` wait gets the blocking question
-through that wait instead.
-If the provider cuts that wait short, a still-pending live question wakes the
+the notification starts a turn, that queued turn is cancelled. Approval requests
+also notify the parent, but tell it to ask the user to open the child thread and
+handle the request there. The pending request tools cannot approve for the user.
-parent when its launching run completes, fails, or waits for post-turn work.
-Stopping or interrupting the parent prevents this catch-up.
+A parent still blocked in a `delegate_task` wait gets the blocking question or
+live approval through that wait instead, as `hasPendingQuestion` or
+`pendingApproval`.
+If the provider cuts that wait short, a still-pending live question or approval
+wakes the parent when its launching run completes, fails, or waits for post-turn
+work.
+Stopping or interrupting the parent, or cancelling a queued completion delivery,
+disposes the affected delegated cohort. Disposing an individual task suppresses
+that task's notices. These paths cancel queued question and approval notices and
+suppress future ones, even if the child keeps running.
 
 A failed run exposes its provider error before any progress text. Successful
 results use the latest assistant content from the final work turn.
```