fix(server): a delegated task waiting for an approval notifies its idle parent

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

@@ -4,12 +4,14 @@ import {
   type OrchestrationV2DomainEvent,
   type OrchestrationV2RuntimeRequest,
   type OrchestrationV2Subagent,
+  ProviderRequestKind,
   type RuntimeRequestId,
   type ThreadId,
 } 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";
 
@@ -19,6 +21,10 @@ import { forkParked } from "../serverActivation.ts";
 import * as Orchestrator from "./Orchestrator.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 
+const isApprovalKind = Schema.is(ProviderRequestKind);
+const notifiesParent = (kind: OrchestrationV2RuntimeRequest["kind"]) =>
+  kind === "user_input" || isApprovalKind(kind);
+
 export class DelegatedTaskQuestionReactor extends Context.Service<
   DelegatedTaskQuestionReactor,
   {
@@ -38,6 +44,9 @@ const make = Effect.gen(function* () {
     if (task.childThreadId === null) return;
     const key = `${task.threadId}:${request.id}`;
     const label = reportLabel(task.title?.trim() || task.prompt);
+    const quotedLabel = label === undefined ? "" : ` "${label}"`;
+    // An approval notice is advisory: the parent agent cannot approve on the user's behalf.
+    const approval = isApprovalKind(request.kind);
     const situation =
       request.responseCapability.type === "message"
         ? `Delegated task ${task.id} asked a question and is still working.`
@@ -47,11 +56,13 @@ const make = Effect.gen(function* () {
       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 action read, threadId ${task.childThreadId}, and requestId ${request.id}. Answer it with action 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.`,
+      text: approval
+        ? `Delegated task ${task.id} is blocked on a ${request.kind} approval (requestId ${request.id}, threadId ${task.childThreadId}) and cannot continue until the user approves or declines it in the app. You cannot approve it. Tell the user it is waiting, or stop the task with task_cancel and relaunch it if it should not need this approval. If it is no longer pending, it was resolved.`
+        : `${situation} Read it with t3_pending_request action read, threadId ${task.childThreadId}, and requestId ${request.id}. Answer it with action 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`,
+        summary: `Delegated task${quotedLabel} ${approval ? "needs approval" : "has a question"}`,
       },
       attachments: [],
       dispatchMode: { type: "queue_after_active" },
@@ -114,7 +125,7 @@ const make = Effect.gen(function* () {
           ]);
           for (const request of child.runtimeRequests) {
             if (
-              request.kind !== "user_input" ||
+              !notifiesParent(request.kind) ||
               request.status !== "pending" ||
               request.responseCapability.type !== "live"
             ) {
@@ -142,12 +153,12 @@ const make = Effect.gen(function* () {
         ) {
           return;
         }
-        // A settled task can no longer use an answer to its still-pending questions.
+        // A settled task can no longer use an answer to its still-pending requests.
         const { runtimeRequests } = yield* projections.getThreadRecords(task.childThreadId, [
           "runtimeRequests",
         ]);
-        const questions = runtimeRequests.filter((request) => request.kind === "user_input");
-        if (questions.length === 0) return;
+        const requests = runtimeRequests.filter((request) => notifiesParent(request.kind));
+        if (requests.length === 0) return;
         const child = yield* projections.getThreadRecords(task.childThreadId, ["runs", "nodes"]);
         const taskRunIds = new Set(
           child.runs
@@ -167,13 +178,21 @@ const make = Effect.gen(function* () {
         );
         yield* cancelQueuedWakes(
           task.threadId,
-          questions
+          requests
             .filter((request) => taskNodeIds.has(request.nodeId))
             .map((request) => request.id),
         );
         return;
       }
-      if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") {
+      if (event.type !== "runtime-request.updated" || !notifiesParent(event.payload.kind)) {
+        return;
+      }
+      // Only a live approval can still be approved, so only it is worth a notice.
+      if (
+        isApprovalKind(event.payload.kind) &&
+        event.payload.status === "pending" &&
+        event.payload.responseCapability.type !== "live"
+      ) {
         return;
       }
       const child = yield* projections.getThread(event.threadId);
@@ -202,7 +221,7 @@ const make = Effect.gen(function* () {
       ) {
         return;
       }
-      // A blocking delegate_task wait delivers the question itself.
+      // A blocking delegate_task wait delivers the request itself.
       if (
         task.completionWake === "settled_only" &&
         event.payload.responseCapability.type === "live" &&

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

@@ -1617,6 +1617,106 @@ it.effect("wakes the parent once when its delegated task asks a question", () =>
   }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))),
 );
 
+it.effect("wakes an idle parent when its async delegated task waits for an approval", () =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor;
+    const eventSink = yield* EventSink.EventSinkV2;
+    const parentThreadId = ThreadId.make("thread:approval-parent");
+    yield* orchestrator.dispatch({
+      type: "thread.create",
+      commandId: CommandId.make("create-approval-parent"),
+      threadId: parentThreadId,
+      projectId: ProjectId.make("project:approval"),
+      title: "Parent",
+      modelSelection,
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      branch: null,
+      worktreePath: null,
+      createdBy: "user",
+      creationSource: "web",
+    });
+    yield* orchestrator.dispatch({
+      type: "message.dispatch",
+      commandId: CommandId.make("start-approval-parent"),
+      threadId: parentThreadId,
+      messageId: MessageId.make("message:approval-parent"),
+      text: "Delegate the edit",
+      attachments: [],
+      dispatchMode: { type: "start_immediately" },
+      createdBy: "user",
+      creationSource: "web",
+    });
+    const parentRun = (yield* projections.getThreadProjection(parentThreadId)).runs[0]!;
+    yield* orchestrator.dispatch({
+      type: "delegated_task.request",
+      commandId: CommandId.make("delegate-approval-task"),
+      parentThreadId,
+      parentRunId: parentRun.id,
+      parentNodeId: parentRun.rootNodeId!,
+      task: "Edit the file",
+      completionWake: "always",
+      modelSelection,
+      runtimeMode: "approval-required",
+      interactionMode: "default",
+      createdBy: "agent",
+      creationSource: "mcp",
+    });
+    const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!;
+    const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!;
+    const now = yield* DateTime.now;
+    yield* eventSink.write({
+      events: [
+        {
+          id: EventId.make("event:approval-parent-completed"),
+          type: "run.updated",
+          threadId: parentThreadId,
+          runId: parentRun.id,
+          occurredAt: now,
+          payload: { ...parentRun, status: "completed", completedAt: now },
+        },
+      ],
+    });
+    const approval = {
+      id: EventId.make("event:child-approval"),
+      type: "runtime-request.updated",
+      threadId: task.childThreadId!,
+      runId: childRun.id,
+      occurredAt: now,
+      payload: {
+        id: RuntimeRequestId.make("request:child-approval"),
+        nodeId: childRun.rootNodeId!,
+        providerTurnId: null,
+        nativeRequestRef: null,
+        kind: "command",
+        status: "pending",
+        responseCapability: {
+          type: "live",
+          providerSessionId: ProviderSessionId.make("session:approval"),
+        },
+        createdAt: now,
+        resolvedAt: null,
+      },
+    } satisfies OrchestrationV2DomainEvent;
+    yield* eventSink.write({ events: [approval] });
+    yield* reactor.handle(approval);
+
+    const wakes = (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
+      (message) =>
+        message.notification?.source.kind === "delegated_task" &&
+        message.text.includes(approval.payload.id),
+    );
+    assert.equal(wakes.length, 1);
+    assert.deepEqual(wakes[0]?.notification?.source, {
+      kind: "delegated_task",
+      taskIds: [task.id],
+      childThreadId: task.childThreadId!,
+    });
+  }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))),
+);
+
 it.effect.each([false, true])(
   "drops a stale question wake (missed terminal event: %s)",
   (missedEvent) =>

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

@@ -2099,7 +2099,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       );
     return !child.runtimeRequests.some(
       (request) =>
-        request.kind === "user_input" &&
         request.status === "pending" &&
         messageId === `message:delegated-task-question:${threadId}:${request.id}`,
     );