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.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) => 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}`, );