From 30c04bc0b2683724ab4dfb100484e29df891eed0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 17:16:33 +0000 Subject: [PATCH 1/2] test(server): an async delegated task's approval wakes its idle parent Fails on main: DelegatedTaskQuestionReactor routes only user_input requests, so a child's command approval never notifies the parent agent (#22925). AI-assisted; verified with the tests and tools listed above. --- .../Orchestrator.control-reads.test.ts | 100 ++++++++++++++++++ 1 file changed, 100 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index f7773665aa9b0..96d1815b6d7ca 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/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) => From d9a1ed3a06b0888b30246e6c2b0fd5e77d17f43e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 17:36:02 +0000 Subject: [PATCH 2/2] fix(server): a delegated task waiting for an approval notifies its idle parent DelegatedTaskQuestionReactor routed only user_input requests, so an async delegated child stuck on a command or other approval never told its parent agent, and the Orchestrator also treated any such queued notice as stale. Live approval requests now queue an advisory notice that names the child thread and request and says the agent cannot approve it; the notice is cancelled once the approval resolves or the task settles, as for questions. Fixes pingdotgg/t3code#22925 AI-assisted; verified with the tests and tools listed above. --- .../DelegatedTaskQuestionReactor.ts | 37 ++++++++++++++----- .../src/orchestration-v2/Orchestrator.ts | 1 - 2 files changed, 28 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 9fd17580e333e..64538b67fae25 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/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" && diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c186c40f9e73a..6a7fe0f35eae7 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/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}`, );