From eb57e62eb782b01554efaf9334d38156ec6b7e77 Mon Sep 17 00:00:00 2001 From: flamboh Date: Wed, 14 Oct 2026 17:40:16 +0000 Subject: [PATCH 01/14] fix(server): a delegated task's question now wakes its parent agent --- .../DelegatedTaskQuestionReactor.test.ts | 191 ++++++++++++++++++ .../DelegatedTaskQuestionReactor.ts | 115 +++++++++++ apps/server/src/server.ts | 7 + .../orchestrator-mcp-server.md | 6 + 4 files changed, 319 insertions(+) create mode 100644 apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts create mode 100644 apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts new file mode 100644 index 000000000000..cc4b0593ef16 --- /dev/null +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts @@ -0,0 +1,191 @@ +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + NodeId, + type OrchestrationV2DomainEvent, + type OrchestrationV2RuntimeRequest, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + RuntimeRequestId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import * as DelegatedTaskQuestionReactor from "./DelegatedTaskQuestionReactor.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; + +const instanceId = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId, model: "gpt-5.1-codex" }; +const adapter = { + instanceId, + driver: ProviderDriverKind.make("codex"), + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), + openSession: () => Effect.die("No provider process needed for delegated task questions"), +} as ProviderAdapter.ProviderAdapterV2["Service"]; +const layerDatabase = SqlitePersistence.layerMemory; +const layerProjections = ProjectionStore.layer.pipe(Layer.provide(layerDatabase)); +const layerHarness = ProviderReplayHarness.layerWithRegistry( + { name: "delegated-task-question" }, + ProviderAdapterRegistry.layerFromAdapters([adapter]), + { databaseLayer: layerDatabase, runEffectWorker: false }, +); +const layerTest = DelegatedTaskQuestionReactor.layer.pipe( + Layer.provideMerge(Layer.mergeAll(layerDatabase, layerProjections, layerHarness)), +); + +const requestEvent = (input: { + readonly threadId: ThreadId; + readonly requestId: string; + readonly kind: OrchestrationV2RuntimeRequest["kind"]; + readonly status?: OrchestrationV2RuntimeRequest["status"]; + readonly now: DateTime.Utc; +}): OrchestrationV2DomainEvent => ({ + id: EventId.make(`event:${input.requestId}`), + type: "runtime-request.updated", + threadId: input.threadId, + occurredAt: input.now, + payload: { + id: RuntimeRequestId.make(input.requestId), + nodeId: NodeId.make(`node:${input.requestId}`), + providerTurnId: null, + nativeRequestRef: null, + kind: input.kind, + status: input.status ?? "pending", + responseCapability: { type: "message" }, + createdAt: input.now, + resolvedAt: null, + }, +}); + +const seedDelegatedTask = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const parentThreadId = ThreadId.make("thread:question-parent"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-question-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:question"), + 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-question-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:question-parent"), + text: "Delegate the research", + 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-question-task"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Research the official questions", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + createdBy: "agent", + creationSource: "mcp", + }); + const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!; + return { parentThreadId, childThreadId: task.childThreadId!, taskId: task.id }; +}); + +const notificationMessages = (threadId: ThreadId) => + ProjectionStore.ProjectionStoreV2.pipe( + Effect.flatMap((projections) => projections.getThreadProjection(threadId)), + Effect.map((projection) => + projection.messages.filter((message) => message.notification !== undefined), + ), + ); + +it.effect("wakes the parent once when its delegated task asks a question", () => + Effect.gen(function* () { + const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; + const { parentThreadId, childThreadId, taskId } = yield* seedDelegatedTask; + const now = yield* DateTime.now; + const question = requestEvent({ + threadId: childThreadId, + requestId: "request:question", + kind: "user_input", + now, + }); + + yield* reactor.handle(question); + // The same request updated while pending must not queue a second wake. + yield* reactor.handle(question); + + const messages = yield* notificationMessages(parentThreadId); + assert.equal(messages.length, 1); + const [message] = messages; + assert.deepEqual(message?.notification, { + source: { kind: "delegated_task", taskIds: [taskId], childThreadId }, + outcome: "updated", + summary: 'Delegated task "Research the official questions" has a question', + }); + assert.include(message?.text, `threadId ${childThreadId}`); + assert.include(message?.text, "t3_pending_request_respond"); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("ignores approvals, answered questions and questions outside delegated tasks", () => + Effect.gen(function* () { + const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; + const { parentThreadId, childThreadId } = yield* seedDelegatedTask; + const now = yield* DateTime.now; + + yield* reactor.handle( + requestEvent({ + threadId: childThreadId, + requestId: "request:approval", + kind: "command", + now, + }), + ); + yield* reactor.handle( + requestEvent({ + threadId: childThreadId, + requestId: "request:answered", + kind: "user_input", + status: "resolved", + now, + }), + ); + yield* reactor.handle( + requestEvent({ + threadId: parentThreadId, + requestId: "request:parent-question", + kind: "user_input", + now, + }), + ); + + assert.deepEqual(yield* notificationMessages(parentThreadId), []); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts new file mode 100644 index 000000000000..2124d56414e8 --- /dev/null +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -0,0 +1,115 @@ +import { CommandId, MessageId, type OrchestrationV2DomainEvent } 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 { forkParked } from "../serverActivation.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; + +const LABEL_MAX_LENGTH = 80; + +function taskLabel(title: string | null | undefined, prompt: string): string { + const firstLine = (title?.trim() || prompt).trim().split("\n")[0]?.trim() ?? ""; + return firstLine.length > LABEL_MAX_LENGTH + ? `${firstLine.slice(0, LABEL_MAX_LENGTH - 1)}…` + : firstLine; +} + +/** + * Wakes the parent of an app-owned delegated task when the task asks a question. The question + * blocks the task until someone answers it, and the user rarely watches child threads, so the + * parent agent is told to answer it or to ask the user. Only questions are routed: approvals + * cannot be answered through the pending request tools. + */ +export class DelegatedTaskQuestionReactor extends Context.Service< + DelegatedTaskQuestionReactor, + { + readonly start: () => Effect.Effect; + readonly handle: (event: OrchestrationV2DomainEvent) => Effect.Effect; + } +>()("t3/orchestration-v2/DelegatedTaskQuestionReactor") {} + +/** @public Service construction is part of the canonical Effect module API. */ +export const make = Effect.gen(function* () { + const engine = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + + const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = (event) => + Effect.gen(function* () { + if ( + event.type !== "runtime-request.updated" || + event.payload.kind !== "user_input" || + event.payload.status !== "pending" + ) { + return; + } + const child = yield* projections.getThread(event.threadId); + const parentThreadId = child.lineage.parentThreadId; + if ( + child.lineage.relationshipToParent !== "subagent" || + parentThreadId === null || + child.forkedFrom?.type !== "node" + ) { + return; + } + const taskId = child.forkedFrom.nodeId; + const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents"]); + const task = parent.subagents.find( + (candidate) => + candidate.id === taskId && + candidate.origin === "app_owned" && + candidate.childThreadId === child.id, + ); + if ( + task === undefined || + parent.thread.archivedAt !== null || + parent.thread.deletedAt !== null || + parent.thread.settledOverride === "settled" || + parent.thread.settledAt !== null + ) { + return; + } + const label = taskLabel(task.title, task.prompt); + // Keyed by the request, so a request updated while still pending wakes the parent once. + const key = `${parentThreadId}:${event.payload.id}`; + yield* engine.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`server:delegated-task-question:${key}`), + threadId: parentThreadId, + messageId: MessageId.make(`message:delegated-task-question:${key}`), + text: `Delegated task ${taskId} is blocked on a question and cannot continue until it is answered. Read it with t3_pending_request_list and t3_pending_request_read using threadId ${child.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.`, + notification: { + source: { kind: "delegated_task", taskIds: [taskId], childThreadId: child.id }, + outcome: "updated", + summary: `Delegated task${label.length === 0 ? "" : ` "${label}"`} has a question`, + }, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + }); + }).pipe( + Effect.catch((error) => + Effect.logWarning("waking the parent of a delegated task question failed", { + threadId: event.threadId, + error, + }), + ), + ); + + 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); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 8148479f8318..23981a932e05 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -6,6 +6,7 @@ import * as Semaphore from "effect/Semaphore"; 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"; // @effect-diagnostics nodeBuiltinImport:off import * as NodeHttp from "node:http"; @@ -571,6 +572,12 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( Layer.provide(layerPullRequestService), Layer.provide(ProjectionStoreV2.layer), ), + Layer.effectDiscard( + Effect.gen(function* () { + const service = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; + yield* service.start(); + }), + ).pipe(Layer.provide(DelegatedTaskQuestionReactor.layer), Layer.provide(ProjectionStoreV2.layer)), // Subscribes to `account.rate-limits.updated` so usage bars track live // telemetry instead of waiting for the next status probe. ProviderUsageLimitsIngestion.layer, diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 20f6456b1a34..31a8637fba9c 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -452,6 +452,12 @@ persisted events and then follows live events, so finalization also runs after a server restart. An existing `subagent_result` transfer makes finalization idempotent. +A child question blocks the child until someone answers it, and users rarely +watch child threads. When a child opens a pending `user_input` request, the +parent gets a notification message telling it to answer with the pending +request tools or ask the user. Approvals are not routed, because those tools +cannot answer them. + A failed run exposes its provider error before any progress text. Successful results use the latest assistant content from the final work turn. From 828e20c47de9e7b9ee736ec3e1b8bf12d247d727 Mon Sep 17 00:00:00 2001 From: flamboh Date: Wed, 14 Oct 2026 17:44:59 +0000 Subject: [PATCH 02/14] refactor(server): reuse notification label for delegated task questions --- .../DelegatedTaskQuestionReactor.ts | 17 +++++------------ .../provider-core/src/server/notification.ts | 2 +- 2 files changed, 6 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 2124d56414e8..acc725c46193 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -5,19 +5,12 @@ 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"; -const LABEL_MAX_LENGTH = 80; - -function taskLabel(title: string | null | undefined, prompt: string): string { - const firstLine = (title?.trim() || prompt).trim().split("\n")[0]?.trim() ?? ""; - return firstLine.length > LABEL_MAX_LENGTH - ? `${firstLine.slice(0, LABEL_MAX_LENGTH - 1)}…` - : firstLine; -} - /** * Wakes the parent of an app-owned delegated task when the task asks a question. The question * blocks the task until someone answers it, and the user rarely watches child threads, so the @@ -72,7 +65,7 @@ export const make = Effect.gen(function* () { ) { return; } - const label = taskLabel(task.title, task.prompt); + const label = reportLabel(task.title?.trim() || task.prompt); // Keyed by the request, so a request updated while still pending wakes the parent once. const key = `${parentThreadId}:${event.payload.id}`; yield* engine.dispatch({ @@ -80,11 +73,11 @@ export const make = Effect.gen(function* () { commandId: CommandId.make(`server:delegated-task-question:${key}`), threadId: parentThreadId, messageId: MessageId.make(`message:delegated-task-question:${key}`), - text: `Delegated task ${taskId} is blocked on a question and cannot continue until it is answered. Read it with t3_pending_request_list and t3_pending_request_read using threadId ${child.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.`, + text: `Delegated task ${taskId} is blocked on a question and cannot continue until it is answered. Read it with t3_pending_request_list and t3_pending_request_read using threadId ${child.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, someone already answered it.`, notification: { source: { kind: "delegated_task", taskIds: [taskId], childThreadId: child.id }, outcome: "updated", - summary: `Delegated task${label.length === 0 ? "" : ` "${label}"`} has a question`, + summary: `Delegated task${label === undefined ? "" : ` "${label}"`} has a question`, }, attachments: [], dispatchMode: { type: "queue_after_active" }, diff --git a/packages/provider-core/src/server/notification.ts b/packages/provider-core/src/server/notification.ts index b90fb02aaf7a..87acef16f2ea 100644 --- a/packages/provider-core/src/server/notification.ts +++ b/packages/provider-core/src/server/notification.ts @@ -25,7 +25,7 @@ export type BackgroundWorkReport = BackgroundWork & { readonly outcome: Notifica const LABEL_MAX_LENGTH = 80; -function reportLabel(label: string | undefined): string | undefined { +export function reportLabel(label: string | undefined): string | undefined { const firstLine = label?.trim().split("\n")[0]?.trim(); if (firstLine === undefined || firstLine.length === 0) return undefined; return firstLine.length > LABEL_MAX_LENGTH From 88473cbd880fad7d975a1aa53ffead7d55605516 Mon Sep 17 00:00:00 2001 From: flamboh Date: Wed, 14 Oct 2026 18:21:37 +0000 Subject: [PATCH 03/14] test(server): fold the delegated task question test into control reads --- .../DelegatedTaskQuestionReactor.test.ts | 191 ------------------ .../Orchestrator.control-reads.test.ts | 98 +++++++++ 2 files changed, 98 insertions(+), 191 deletions(-) delete mode 100644 apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts deleted file mode 100644 index cc4b0593ef16..000000000000 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.test.ts +++ /dev/null @@ -1,191 +0,0 @@ -import { assert, it } from "@effect/vitest"; -import { - CommandId, - EventId, - MessageId, - NodeId, - type OrchestrationV2DomainEvent, - type OrchestrationV2RuntimeRequest, - ProjectId, - ProviderDriverKind, - ProviderInstanceId, - RuntimeRequestId, - ThreadId, -} from "@t3tools/contracts"; -import * as DateTime from "effect/DateTime"; -import * as Effect from "effect/Effect"; -import * as Layer from "effect/Layer"; - -import * as SqlitePersistence from "../persistence/Sqlite.ts"; -import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; -import * as DelegatedTaskQuestionReactor from "./DelegatedTaskQuestionReactor.ts"; -import * as Orchestrator from "./Orchestrator.ts"; -import * as ProjectionStore from "./ProjectionStore.ts"; -import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; -import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; -import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; - -const instanceId = ProviderInstanceId.make("codex"); -const modelSelection = { instanceId, model: "gpt-5.1-codex" }; -const adapter = { - instanceId, - driver: ProviderDriverKind.make("codex"), - getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), - planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), - openSession: () => Effect.die("No provider process needed for delegated task questions"), -} as ProviderAdapter.ProviderAdapterV2["Service"]; -const layerDatabase = SqlitePersistence.layerMemory; -const layerProjections = ProjectionStore.layer.pipe(Layer.provide(layerDatabase)); -const layerHarness = ProviderReplayHarness.layerWithRegistry( - { name: "delegated-task-question" }, - ProviderAdapterRegistry.layerFromAdapters([adapter]), - { databaseLayer: layerDatabase, runEffectWorker: false }, -); -const layerTest = DelegatedTaskQuestionReactor.layer.pipe( - Layer.provideMerge(Layer.mergeAll(layerDatabase, layerProjections, layerHarness)), -); - -const requestEvent = (input: { - readonly threadId: ThreadId; - readonly requestId: string; - readonly kind: OrchestrationV2RuntimeRequest["kind"]; - readonly status?: OrchestrationV2RuntimeRequest["status"]; - readonly now: DateTime.Utc; -}): OrchestrationV2DomainEvent => ({ - id: EventId.make(`event:${input.requestId}`), - type: "runtime-request.updated", - threadId: input.threadId, - occurredAt: input.now, - payload: { - id: RuntimeRequestId.make(input.requestId), - nodeId: NodeId.make(`node:${input.requestId}`), - providerTurnId: null, - nativeRequestRef: null, - kind: input.kind, - status: input.status ?? "pending", - responseCapability: { type: "message" }, - createdAt: input.now, - resolvedAt: null, - }, -}); - -const seedDelegatedTask = Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const parentThreadId = ThreadId.make("thread:question-parent"); - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make("create-question-parent"), - threadId: parentThreadId, - projectId: ProjectId.make("project:question"), - 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-question-parent"), - threadId: parentThreadId, - messageId: MessageId.make("message:question-parent"), - text: "Delegate the research", - 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-question-task"), - parentThreadId, - parentRunId: parentRun.id, - parentNodeId: parentRun.rootNodeId!, - task: "Research the official questions", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - completionWake: "always", - createdBy: "agent", - creationSource: "mcp", - }); - const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!; - return { parentThreadId, childThreadId: task.childThreadId!, taskId: task.id }; -}); - -const notificationMessages = (threadId: ThreadId) => - ProjectionStore.ProjectionStoreV2.pipe( - Effect.flatMap((projections) => projections.getThreadProjection(threadId)), - Effect.map((projection) => - projection.messages.filter((message) => message.notification !== undefined), - ), - ); - -it.effect("wakes the parent once when its delegated task asks a question", () => - Effect.gen(function* () { - const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; - const { parentThreadId, childThreadId, taskId } = yield* seedDelegatedTask; - const now = yield* DateTime.now; - const question = requestEvent({ - threadId: childThreadId, - requestId: "request:question", - kind: "user_input", - now, - }); - - yield* reactor.handle(question); - // The same request updated while pending must not queue a second wake. - yield* reactor.handle(question); - - const messages = yield* notificationMessages(parentThreadId); - assert.equal(messages.length, 1); - const [message] = messages; - assert.deepEqual(message?.notification, { - source: { kind: "delegated_task", taskIds: [taskId], childThreadId }, - outcome: "updated", - summary: 'Delegated task "Research the official questions" has a question', - }); - assert.include(message?.text, `threadId ${childThreadId}`); - assert.include(message?.text, "t3_pending_request_respond"); - }).pipe(Effect.provide(layerTest)), -); - -it.effect("ignores approvals, answered questions and questions outside delegated tasks", () => - Effect.gen(function* () { - const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; - const { parentThreadId, childThreadId } = yield* seedDelegatedTask; - const now = yield* DateTime.now; - - yield* reactor.handle( - requestEvent({ - threadId: childThreadId, - requestId: "request:approval", - kind: "command", - now, - }), - ); - yield* reactor.handle( - requestEvent({ - threadId: childThreadId, - requestId: "request:answered", - kind: "user_input", - status: "resolved", - now, - }), - ); - yield* reactor.handle( - requestEvent({ - threadId: parentThreadId, - requestId: "request:parent-question", - kind: "user_input", - now, - }), - ); - - assert.deepEqual(yield* notificationMessages(parentThreadId), []); - }).pipe(Effect.provide(layerTest)), -); 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 8e5fca79cf55..1254a97a519c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -4,6 +4,7 @@ import { MessageId, EventId, NodeId, + type OrchestrationV2DomainEvent, PlanId, ProjectId, ProviderDriverKind, @@ -23,6 +24,7 @@ import * as Layer from "effect/Layer"; 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 Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; @@ -624,3 +626,99 @@ it.effect("keeps delegated child pull-request links independent of the parent", assert.deepEqual(parentAfterChildLink.thread.pullRequests, parent.thread.pullRequests); }).pipe(Effect.provide(layerTest)), ); + +it.effect("wakes the parent once when its delegated task asks a question", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor; + const parentThreadId = ThreadId.make("thread:question-parent"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-question-parent"), + threadId: parentThreadId, + projectId: ProjectId.make("project:question"), + 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-question-parent"), + threadId: parentThreadId, + messageId: MessageId.make("message:question-parent"), + text: "Delegate the research", + 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-question-task"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRun.rootNodeId!, + task: "Research the sources", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "agent", + creationSource: "mcp", + }); + const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!; + const now = yield* DateTime.now; + const question: OrchestrationV2DomainEvent = { + id: EventId.make("event:question"), + type: "runtime-request.updated", + threadId: task.childThreadId!, + occurredAt: now, + payload: { + id: RuntimeRequestId.make("request:question"), + nodeId: NodeId.make("node:question"), + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: now, + resolvedAt: null, + }, + }; + + yield* reactor.handle(question); + yield* reactor.handle(question); + yield* reactor.handle({ + ...question, + payload: { + ...question.payload, + id: RuntimeRequestId.make("request:approval"), + kind: "command", + }, + }); + + const wakes = (yield* projections.getThreadProjection(parentThreadId)).messages.filter( + (message) => message.notification !== undefined, + ); + assert.deepEqual( + wakes.map((message) => message.notification), + [ + { + source: { + kind: "delegated_task", + taskIds: [task.id], + childThreadId: task.childThreadId!, + }, + outcome: "updated", + summary: 'Delegated task "Research the sources" has a question', + }, + ], + ); + }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), +); From 9059d139d8dc6b1295a6cb11212717eab508def9 Mon Sep 17 00:00:00 2001 From: flamboh Date: Wed, 14 Oct 2026 18:29:03 +0000 Subject: [PATCH 04/14] fix(server): an answered question no longer wakes the parent --- .../DelegatedTaskQuestionReactor.ts | 5 ++++- .../Orchestrator.control-reads.test.ts | 20 +++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index acc725c46193..27b7a2156ab2 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -39,9 +39,12 @@ export const make = Effect.gen(function* () { ) { return; } - const child = yield* projections.getThread(event.threadId); + const childRecords = yield* projections.getThreadRecords(event.threadId, ["runtimeRequests"]); + const child = childRecords.thread; const parentThreadId = child.lineage.parentThreadId; if ( + childRecords.runtimeRequests.find((request) => request.id === event.payload.id)?.status !== + "pending" || child.lineage.relationshipToParent !== "subagent" || parentThreadId === null || child.forkedFrom?.type !== "node" 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 1254a97a519c..4b5580fd85ef 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -25,6 +25,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 EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; @@ -692,8 +693,27 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }, }; + const answered = { + ...question, + id: EventId.make("event:answered"), + payload: { ...question.payload, id: RuntimeRequestId.make("request:answered") }, + }; + const eventSink = yield* EventSink.EventSinkV2; + yield* eventSink.write({ + events: [ + question, + answered, + { + ...answered, + id: EventId.make("event:answered-resolved"), + payload: { ...answered.payload, status: "resolved", resolvedAt: now }, + }, + ], + }); + yield* reactor.handle(question); yield* reactor.handle(question); + yield* reactor.handle(answered); yield* reactor.handle({ ...question, payload: { From 01c3015b8a797bd7fecf916be71078c3ab239e24 Mon Sep 17 00:00:00 2001 From: flamboh Date: Wed, 14 Oct 2026 18:37:40 +0000 Subject: [PATCH 05/14] fix(server): a background question no longer reads as blocking the task --- .../src/orchestration-v2/DelegatedTaskQuestionReactor.ts | 6 +++++- .../orchestration-v2/Orchestrator.control-reads.test.ts | 1 + docs/orchestration-v2/orchestrator-mcp-server.md | 9 +++++---- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 27b7a2156ab2..2adde971ffb9 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -71,12 +71,16 @@ export const make = Effect.gen(function* () { const label = reportLabel(task.title?.trim() || task.prompt); // Keyed by the request, so a request updated while still pending wakes the parent once. const key = `${parentThreadId}:${event.payload.id}`; + const situation = + event.payload.responseCapability.type === "message" + ? `Delegated task ${taskId} asked a question and is still working.` + : `Delegated task ${taskId} 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: parentThreadId, messageId: MessageId.make(`message:delegated-task-question:${key}`), - text: `Delegated task ${taskId} is blocked on a question and cannot continue until it is answered. Read it with t3_pending_request_list and t3_pending_request_read using threadId ${child.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, someone already answered it.`, + text: `${situation} Read it with t3_pending_request_list and t3_pending_request_read using threadId ${child.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: [taskId], childThreadId: child.id }, outcome: "updated", 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 4b5580fd85ef..b1387daf7e96 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -726,6 +726,7 @@ it.effect("wakes the parent once when its delegated task asks a question", () => const wakes = (yield* projections.getThreadProjection(parentThreadId)).messages.filter( (message) => message.notification !== undefined, ); + assert.include(wakes[0]?.text, "asked a question and is still working"); assert.deepEqual( wakes.map((message) => message.notification), [ diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 31a8637fba9c..63318f46cad3 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -452,10 +452,11 @@ persisted events and then follows live events, so finalization also runs after a server restart. An existing `subagent_result` transfer makes finalization idempotent. -A child question blocks the child until someone answers it, and users rarely -watch child threads. When a child opens a pending `user_input` request, the -parent gets a notification message telling it to answer with the pending -request tools or ask the user. Approvals are not routed, because those tools +Users rarely watch child threads, so a child question would otherwise go +unanswered. When a child opens a pending `user_input` request, the parent gets +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. Approvals are not routed, because those tools cannot answer them. A failed run exposes its provider error before any progress text. Successful From 92c5c37f327fa38fb608d70a8b4cf23735dd6297 Mon Sep 17 00:00:00 2001 From: flamboh Date: Wed, 14 Oct 2026 18:46:40 +0000 Subject: [PATCH 06/14] test(server): store the approval before checking it is not routed --- .../Orchestrator.control-reads.test.ts | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) 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 b1387daf7e96..e1b35e1765d5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -698,6 +698,15 @@ it.effect("wakes the parent once when its delegated task asks a question", () => id: EventId.make("event:answered"), payload: { ...question.payload, id: RuntimeRequestId.make("request:answered") }, }; + const approval = { + ...question, + id: EventId.make("event:approval"), + payload: { + ...question.payload, + id: RuntimeRequestId.make("request:approval"), + kind: "command", + }, + } satisfies OrchestrationV2DomainEvent; const eventSink = yield* EventSink.EventSinkV2; yield* eventSink.write({ events: [ @@ -708,20 +717,14 @@ it.effect("wakes the parent once when its delegated task asks a question", () => id: EventId.make("event:answered-resolved"), payload: { ...answered.payload, status: "resolved", resolvedAt: now }, }, + approval, ], }); yield* reactor.handle(question); yield* reactor.handle(question); yield* reactor.handle(answered); - yield* reactor.handle({ - ...question, - payload: { - ...question.payload, - id: RuntimeRequestId.make("request:approval"), - kind: "command", - }, - }); + yield* reactor.handle(approval); const wakes = (yield* projections.getThreadProjection(parentThreadId)).messages.filter( (message) => message.notification !== undefined, From 6d82648a6b21c96a559f2f1e50511ec15c3e074e Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 14 Oct 2026 20:52:57 +0000 Subject: [PATCH 07/14] fix(server): simplify delegated question wake routing --- .../DelegatedTaskQuestionReactor.ts | 25 +++----- .../Orchestrator.control-reads.test.ts | 57 ++++++++----------- 2 files changed, 32 insertions(+), 50 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 2adde971ffb9..e658bcb923d5 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -25,8 +25,7 @@ export class DelegatedTaskQuestionReactor extends Context.Service< } >()("t3/orchestration-v2/DelegatedTaskQuestionReactor") {} -/** @public Service construction is part of the canonical Effect module API. */ -export const make = Effect.gen(function* () { +const make = Effect.gen(function* () { const engine = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; @@ -39,25 +38,14 @@ export const make = Effect.gen(function* () { ) { return; } - const childRecords = yield* projections.getThreadRecords(event.threadId, ["runtimeRequests"]); - const child = childRecords.thread; + const child = yield* projections.getThread(event.threadId); const parentThreadId = child.lineage.parentThreadId; - if ( - childRecords.runtimeRequests.find((request) => request.id === event.payload.id)?.status !== - "pending" || - child.lineage.relationshipToParent !== "subagent" || - parentThreadId === null || - child.forkedFrom?.type !== "node" - ) { - return; - } - const taskId = child.forkedFrom.nodeId; + if (parentThreadId === null) return; + const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id); + if (request?.status !== "pending") return; const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents"]); const task = parent.subagents.find( - (candidate) => - candidate.id === taskId && - candidate.origin === "app_owned" && - candidate.childThreadId === child.id, + (candidate) => candidate.origin === "app_owned" && candidate.childThreadId === child.id, ); if ( task === undefined || @@ -68,6 +56,7 @@ export const make = Effect.gen(function* () { ) { return; } + const taskId = task.id; const label = reportLabel(task.title?.trim() || task.prompt); // Keyed by the request, so a request updated while still pending wakes the parent once. const key = `${parentThreadId}:${event.payload.id}`; 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 e1b35e1765d5..d3da5806a6b9 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -675,38 +675,31 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }); const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!; const now = yield* DateTime.now; - const question: OrchestrationV2DomainEvent = { - id: EventId.make("event:question"), - type: "runtime-request.updated", - threadId: task.childThreadId!, - occurredAt: now, - payload: { - id: RuntimeRequestId.make("request:question"), - nodeId: NodeId.make("node:question"), - providerTurnId: null, - nativeRequestRef: null, - kind: "user_input", - status: "pending", - responseCapability: { type: "message" }, - createdAt: now, - resolvedAt: null, - }, - }; - - const answered = { - ...question, - id: EventId.make("event:answered"), - payload: { ...question.payload, id: RuntimeRequestId.make("request:answered") }, - }; - const approval = { - ...question, - id: EventId.make("event:approval"), - payload: { - ...question.payload, - id: RuntimeRequestId.make("request:approval"), - kind: "command", - }, - } satisfies OrchestrationV2DomainEvent; + const requestEvent = ( + eventId: string, + requestId: string, + kind: "user_input" | "command" = "user_input", + ) => + ({ + id: EventId.make(eventId), + type: "runtime-request.updated", + threadId: task.childThreadId!, + occurredAt: now, + payload: { + id: RuntimeRequestId.make(requestId), + nodeId: NodeId.make(`node:${requestId}`), + providerTurnId: null, + nativeRequestRef: null, + kind, + status: "pending", + responseCapability: { type: "message" }, + createdAt: now, + resolvedAt: null, + }, + }) satisfies OrchestrationV2DomainEvent; + const question = requestEvent("event:question", "request:question"); + const answered = requestEvent("event:answered", "request:answered"); + const approval = requestEvent("event:approval", "request:approval", "command"); const eventSink = yield* EventSink.EventSinkV2; yield* eventSink.write({ events: [ From 4b7122ff9825a791c0f4170d1aa455f9f351fa2b Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 14 Oct 2026 21:21:01 +0000 Subject: [PATCH 08/14] fix(server): route follow-up questions and cancel stale wakes --- .../DelegatedTaskQuestionReactor.ts | 63 ++++++++++----- .../Orchestrator.control-reads.test.ts | 81 +++++++++++++++++++ .../orchestrator-mcp-server.md | 5 +- 3 files changed, 128 insertions(+), 21 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index e658bcb923d5..f92a125c29c7 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -31,35 +31,60 @@ const make = Effect.gen(function* () { const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = (event) => Effect.gen(function* () { - if ( - event.type !== "runtime-request.updated" || - event.payload.kind !== "user_input" || - event.payload.status !== "pending" - ) { + 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); - if (request?.status !== "pending") return; - const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents"]); + // Keyed by the request, so a request updated while still pending wakes the parent once. + const key = `${parentThreadId}:${event.payload.id}`; + const messageId = MessageId.make(`message:delegated-task-question:${key}`); + const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]); + if (event.payload.status !== "pending" || request?.status !== "pending") { + 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 controls = yield* projections.getThreadRecords(child.id, [ + "runs", + "nodes", + "providerTurns", + ]); + const turnNodeId = controls.providerTurns.find( + (turn) => turn.id === request.providerTurnId, + )?.nodeId; + const runId = + event.runId ?? + controls.nodes.find((node) => node.id === request.nodeId)?.runId ?? + controls.nodes.find((node) => node.id === turnNodeId)?.runId; + const run = + controls.runs.find((run) => run.id === runId) ?? + controls.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.childThreadId === child.id, + (candidate) => candidate.origin === "app_owned" && candidate.id === taskId, ); if ( task === undefined || parent.thread.archivedAt !== null || - parent.thread.deletedAt !== null || - parent.thread.settledOverride === "settled" || - parent.thread.settledAt !== null + parent.thread.deletedAt !== null ) { return; } - const taskId = task.id; const label = reportLabel(task.title?.trim() || task.prompt); - // Keyed by the request, so a request updated while still pending wakes the parent once. - const key = `${parentThreadId}:${event.payload.id}`; const situation = event.payload.responseCapability.type === "message" ? `Delegated task ${taskId} asked a question and is still working.` @@ -68,10 +93,10 @@ const make = Effect.gen(function* () { type: "message.dispatch", commandId: CommandId.make(`server:delegated-task-question:${key}`), threadId: parentThreadId, - messageId: MessageId.make(`message:delegated-task-question:${key}`), - text: `${situation} Read it with t3_pending_request_list and t3_pending_request_read using threadId ${child.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.`, + messageId, + text: `${situation} Read it with t3_pending_request_read using threadId ${child.id} and requestId ${event.payload.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: [taskId], childThreadId: child.id }, + source: { kind: "delegated_task", taskIds: [task.id], childThreadId: child.id }, outcome: "updated", summary: `Delegated task${label === undefined ? "" : ` "${label}"`} has a question`, }, @@ -81,10 +106,10 @@ const make = Effect.gen(function* () { creationSource: "server", }); }).pipe( - Effect.catch((error) => + Effect.catchCause((cause) => Effect.logWarning("waking the parent of a delegated task question failed", { threadId: event.threadId, - error, + cause, }), ), ); 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 d3da5806a6b9..c794243a50c2 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -737,5 +737,86 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }, ], ); + + const wakeRun = (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find( + (run) => run.userMessageId === wakes[0]!.id, + )!; + assert.equal(wakeRun.status, "queued"); + const resolved = { + ...question, + id: EventId.make("event:question-resolved"), + payload: { ...question.payload, status: "resolved" as const, resolvedAt: now }, + }; + yield* eventSink.write({ events: [resolved] }); + yield* reactor.handle(resolved); + yield* reactor.handle(resolved); + assert.equal( + (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find( + (run) => run.id === wakeRun.id, + )?.status, + "cancelled", + ); + + const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!; + yield* eventSink.write({ + commandId: CommandId.make("command:runtime-reconcile:question-followup"), + events: [ + { + id: EventId.make("event:question-child-completed"), + type: "run.updated", + threadId: task.childThreadId!, + occurredAt: now, + payload: { ...childRun, status: "completed", completedAt: now }, + }, + { + id: EventId.make("event:question-task-completed"), + type: "subagent.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { + ...task, + status: "completed", + result: "Sources researched.", + completedAt: now, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("question-followup"), + threadId: task.childThreadId!, + messageId: MessageId.make("message:question-followup"), + text: "Check the sources again", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const followup = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs.find( + (run) => run.userMessageId === "message:question-followup", + )!; + assert.isDefined(followup.delegatedTaskId); + assert.notEqual(followup.delegatedTaskId, task.id); + const followupQuestion = { + ...requestEvent("event:followup-question", "request:followup-question"), + runId: followup.id, + }; + yield* eventSink.write({ events: [followupQuestion] }); + yield* reactor.handle(followupQuestion); + const followupWake = (yield* projections.getThreadRecords(parentThreadId, [ + "messages", + ])).messages.find( + (message) => + message.id === + `message:delegated-task-question:${parentThreadId}:${followupQuestion.payload.id}`, + )!; + assert.include(followupWake.text, `Delegated task ${followup.delegatedTaskId}`); + assert.include(followupWake.text, `requestId ${followupQuestion.payload.id}`); + assert.deepEqual(followupWake.notification?.source, { + kind: "delegated_task", + taskIds: [followup.delegatedTaskId], + childThreadId: task.childThreadId!, + }); }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), ); diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 63318f46cad3..52a0919d6bf2 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -456,8 +456,9 @@ Users rarely watch child threads, so a child question would otherwise go unanswered. When a child opens a pending `user_input` request, the parent gets 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. Approvals are not routed, because those tools -cannot answer them. +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 failed run exposes its provider error before any progress text. Successful results use the latest assistant content from the final work turn. From 6b2cc98c1209d2fc8844ca168550690a18a93336 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 14 Oct 2026 21:43:00 +0000 Subject: [PATCH 09/14] refactor(server): simplify delegated question task lookup --- .../DelegatedTaskQuestionReactor.ts | 20 +++++-------------- .../Orchestrator.control-reads.test.ts | 12 +++-------- 2 files changed, 8 insertions(+), 24 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index f92a125c29c7..9b5dafefea9e 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -41,8 +41,8 @@ const make = Effect.gen(function* () { // Keyed by the request, so a request updated while still pending wakes the parent once. const key = `${parentThreadId}:${event.payload.id}`; const messageId = MessageId.make(`message:delegated-task-question:${key}`); - const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]); 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, ); @@ -56,21 +56,11 @@ const make = Effect.gen(function* () { } return; } - const controls = yield* projections.getThreadRecords(child.id, [ - "runs", - "nodes", - "providerTurns", - ]); - const turnNodeId = controls.providerTurns.find( - (turn) => turn.id === request.providerTurnId, - )?.nodeId; - const runId = - event.runId ?? - controls.nodes.find((node) => node.id === request.nodeId)?.runId ?? - controls.nodes.find((node) => node.id === turnNodeId)?.runId; + const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents"]); + const { runs } = yield* projections.getThreadRecords(child.id, ["runs"]); const run = - controls.runs.find((run) => run.id === runId) ?? - controls.runs.toSorted((a, b) => b.ordinal - a.ordinal)[0]; + 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); 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 c794243a50c2..ca91e2e25c21 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -704,7 +704,6 @@ it.effect("wakes the parent once when its delegated task asks a question", () => yield* eventSink.write({ events: [ question, - answered, { ...answered, id: EventId.make("event:answered-resolved"), @@ -749,7 +748,6 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }; yield* eventSink.write({ events: [resolved] }); yield* reactor.handle(resolved); - yield* reactor.handle(resolved); assert.equal( (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find( (run) => run.id === wakeRun.id, @@ -796,7 +794,6 @@ it.effect("wakes the parent once when its delegated task asks a question", () => const followup = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs.find( (run) => run.userMessageId === "message:question-followup", )!; - assert.isDefined(followup.delegatedTaskId); assert.notEqual(followup.delegatedTaskId, task.id); const followupQuestion = { ...requestEvent("event:followup-question", "request:followup-question"), @@ -806,12 +803,9 @@ it.effect("wakes the parent once when its delegated task asks a question", () => yield* reactor.handle(followupQuestion); const followupWake = (yield* projections.getThreadRecords(parentThreadId, [ "messages", - ])).messages.find( - (message) => - message.id === - `message:delegated-task-question:${parentThreadId}:${followupQuestion.payload.id}`, - )!; - assert.include(followupWake.text, `Delegated task ${followup.delegatedTaskId}`); + ])).messages + .filter((message) => message.notification !== undefined) + .at(-1)!; assert.include(followupWake.text, `requestId ${followupQuestion.payload.id}`); assert.deepEqual(followupWake.notification?.source, { kind: "delegated_task", From 8400dccffc9249e6ec3b730b08413d5b25358d03 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 14 Oct 2026 21:51:55 +0000 Subject: [PATCH 10/14] test(server): find the follow-up question wake by its request id --- .../orchestration-v2/Orchestrator.control-reads.test.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) 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 ca91e2e25c21..d8be2e3958ff 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -794,6 +794,7 @@ it.effect("wakes the parent once when its delegated task asks a question", () => const followup = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs.find( (run) => run.userMessageId === "message:question-followup", )!; + assert.isDefined(followup.delegatedTaskId); assert.notEqual(followup.delegatedTaskId, task.id); const followupQuestion = { ...requestEvent("event:followup-question", "request:followup-question"), @@ -803,10 +804,9 @@ it.effect("wakes the parent once when its delegated task asks a question", () => yield* reactor.handle(followupQuestion); const followupWake = (yield* projections.getThreadRecords(parentThreadId, [ "messages", - ])).messages - .filter((message) => message.notification !== undefined) - .at(-1)!; - assert.include(followupWake.text, `requestId ${followupQuestion.payload.id}`); + ])).messages.find((message) => + message.text.includes(`requestId ${followupQuestion.payload.id}`), + )!; assert.deepEqual(followupWake.notification?.source, { kind: "delegated_task", taskIds: [followup.delegatedTaskId], From a572e44ec286f9b57982457efbce7968da42eefe Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Thu, 15 Oct 2026 00:35:51 +0000 Subject: [PATCH 11/14] fix(server): a parent blocked in a wait isn't woken again for the same question --- .../DelegatedTaskQuestionReactor.ts | 10 +++++++++- .../Orchestrator.control-reads.test.ts | 15 +++++++++++++++ docs/orchestration-v2/orchestrator-mcp-server.md | 1 + 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 9b5dafefea9e..03a19171bab4 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -10,6 +10,7 @@ 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"; +import { isTerminalRunStatus } from "./ThreadManagementService.ts"; /** * Wakes the parent of an app-owned delegated task when the task asks a question. The question @@ -56,7 +57,7 @@ const make = Effect.gen(function* () { } return; } - const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents"]); + 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) ?? @@ -74,6 +75,13 @@ const make = Effect.gen(function* () { ) { return; } + // A blocking delegate_task wait delivers the question itself. + if ( + task.completionWake === "settled_only" && + parent.runs.some((run) => run.id === task.runId && !isTerminalRunStatus(run.status)) + ) { + return; + } const label = reportLabel(task.title?.trim() || task.prompt); const situation = event.payload.responseCapability.type === "message" 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 d8be2e3958ff..32a4f24e0a5c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -667,6 +667,7 @@ it.effect("wakes the parent once when its delegated task asks a question", () => parentRunId: parentRun.id, parentNodeId: parentRun.rootNodeId!, task: "Research the sources", + completionWake: "settled_only", modelSelection, runtimeMode: "full-access", interactionMode: "default", @@ -713,6 +714,20 @@ it.effect("wakes the parent once when its delegated task asks a question", () => ], }); + yield* reactor.handle(question); + assert.isFalse( + (yield* projections.getThreadProjection(parentThreadId)).messages.some( + (message) => message.notification !== undefined, + ), + ); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("question-wait-ended"), + parentThreadId, + taskId: task.id, + completionWake: "always", + }); + yield* reactor.handle(question); yield* reactor.handle(question); yield* reactor.handle(answered); diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 52a0919d6bf2..82c558a5cda4 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -459,6 +459,7 @@ 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 `delegate_task` wait gets the question through that wait instead. A failed run exposes its provider error before any progress text. Successful results use the latest assistant content from the final work turn. From 1428e0da2143e7f1090d91bdfee3cbf809870ab3 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Thu, 15 Oct 2026 00:59:22 +0000 Subject: [PATCH 12/14] fix(server): route async delegated questions during waits --- .../DelegatedTaskQuestionReactor.ts | 8 +- .../Orchestrator.control-reads.test.ts | 92 +++++++++++++++---- .../orchestrator-mcp-server.md | 3 +- 3 files changed, 84 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index 03a19171bab4..f26c24c5a815 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -10,7 +10,6 @@ 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"; -import { isTerminalRunStatus } from "./ThreadManagementService.ts"; /** * Wakes the parent of an app-owned delegated task when the task asks a question. The question @@ -78,7 +77,12 @@ const make = Effect.gen(function* () { // A blocking delegate_task wait delivers the question itself. if ( task.completionWake === "settled_only" && - parent.runs.some((run) => run.id === task.runId && !isTerminalRunStatus(run.status)) + event.payload.responseCapability.type === "live" && + parent.runs.some( + (run) => + run.id === task.runId && + (run.status === "preparing" || run.status === "starting" || run.status === "running"), + ) ) { return; } 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 32a4f24e0a5c..efa73ea40f87 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -679,6 +679,7 @@ it.effect("wakes the parent once when its delegated task asks a question", () => const requestEvent = ( eventId: string, requestId: string, + responseCapability: "live" | "message" = "message", kind: "user_input" | "command" = "user_input", ) => ({ @@ -693,18 +694,23 @@ it.effect("wakes the parent once when its delegated task asks a question", () => nativeRequestRef: null, kind, status: "pending", - responseCapability: { type: "message" }, + responseCapability: + responseCapability === "live" + ? { type: "live", providerSessionId: ProviderSessionId.make("session:question") } + : { type: "message" }, createdAt: now, resolvedAt: null, }, }) satisfies OrchestrationV2DomainEvent; - const question = requestEvent("event:question", "request:question"); + const question = requestEvent("event:question", "request:question", "live"); + const asyncQuestion = requestEvent("event:async-question", "request:async-question"); const answered = requestEvent("event:answered", "request:answered"); - const approval = requestEvent("event:approval", "request:approval", "command"); + const approval = requestEvent("event:approval", "request:approval", "message", "command"); const eventSink = yield* EventSink.EventSinkV2; yield* eventSink.write({ events: [ question, + asyncQuestion, { ...answered, id: EventId.make("event:answered-resolved"), @@ -720,16 +726,14 @@ it.effect("wakes the parent once when its delegated task asks a question", () => (message) => message.notification !== undefined, ), ); - yield* orchestrator.dispatch({ - type: "delegated_task.wake-policy", - commandId: CommandId.make("question-wait-ended"), - parentThreadId, - taskId: task.id, - completionWake: "always", - }); - - yield* reactor.handle(question); - yield* reactor.handle(question); + yield* reactor.handle(asyncQuestion); + assert.equal( + (yield* projections.getThreadProjection(parentThreadId)).messages.filter( + (message) => message.notification !== undefined, + ).length, + 1, + ); + yield* reactor.handle(asyncQuestion); yield* reactor.handle(answered); yield* reactor.handle(approval); @@ -757,9 +761,9 @@ it.effect("wakes the parent once when its delegated task asks a question", () => )!; assert.equal(wakeRun.status, "queued"); const resolved = { - ...question, + ...asyncQuestion, id: EventId.make("event:question-resolved"), - payload: { ...question.payload, status: "resolved" as const, resolvedAt: now }, + payload: { ...asyncQuestion.payload, status: "resolved" as const, resolvedAt: now }, }; yield* eventSink.write({ events: [resolved] }); yield* reactor.handle(resolved); @@ -769,7 +773,63 @@ it.effect("wakes the parent once when its delegated task asks a question", () => )?.status, "cancelled", ); - + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:question-parent-completed"), + type: "run.updated", + threadId: parentThreadId, + runId: parentRun.id, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }, + ], + }); + const terminalQuestion = requestEvent( + "event:terminal-question", + "request:terminal-question", + "live", + ); + yield* eventSink.write({ events: [terminalQuestion] }); + yield* reactor.handle(terminalQuestion); + assert.equal( + (yield* projections.getThreadProjection(parentThreadId)).messages.filter( + (message) => message.notification !== undefined, + ).length, + 2, + ); + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:question-parent-running-again"), + type: "run.updated", + threadId: parentThreadId, + runId: parentRun.id, + occurredAt: now, + payload: { ...parentRun, status: "running", completedAt: null }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("terminal-question-wait-ended"), + parentThreadId, + taskId: task.id, + completionWake: "always", + }); + const upgradedQuestion = requestEvent( + "event:upgraded-question", + "request:upgraded-question", + "live", + ); + yield* eventSink.write({ events: [upgradedQuestion] }); + yield* reactor.handle(upgradedQuestion); + assert.equal( + (yield* projections.getThreadProjection(parentThreadId)).messages.filter( + (message) => message.notification !== undefined, + ).length, + 3, + ); const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!; yield* eventSink.write({ commandId: CommandId.make("command:runtime-reconcile:question-followup"), diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 82c558a5cda4..c935b404b18c 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -459,7 +459,8 @@ 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 `delegate_task` wait gets the question through that wait instead. +A parent still blocked in a `delegate_task` wait gets the blocking question +through that wait instead. A failed run exposes its provider error before any progress text. Successful results use the latest assistant content from the final work turn. From 6ca2e53a7e8c6564e0e2cfd465d8063226789f25 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Thu, 15 Oct 2026 02:25:04 +0000 Subject: [PATCH 13/14] fix(server): a question skipped during a cut-off wait still reaches the parent --- .../DelegatedTaskQuestionReactor.ts | 110 ++++++++++++--- .../Orchestrator.control-reads.test.ts | 129 +++++++++++++----- .../orchestrator-mcp-server.md | 2 + 3 files changed, 187 insertions(+), 54 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index f26c24c5a815..c50d7b61b5f1 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -1,4 +1,10 @@ -import { CommandId, MessageId, type OrchestrationV2DomainEvent } from "@t3tools/contracts"; +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"; @@ -29,8 +35,88 @@ 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", + }); + }); + const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = (event) => Effect.gen(function* () { + if (event.type === "run.updated") { + if ( + event.payload.status === "preparing" || + event.payload.status === "starting" || + event.payload.status === "running" + ) { + 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.childThreadId === null || + task.status === "completed" || + task.status === "failed" || + task.status === "cancelled" || + task.status === "interrupted" + ) { + continue; + } + const child = yield* projections.getThreadRecords(task.childThreadId, [ + "runtimeRequests", + "nodes", + "runs", + ]); + for (const request of child.runtimeRequests) { + if ( + request.kind !== "user_input" || + request.status !== "pending" || + request.responseCapability.type !== "live" + ) { + continue; + } + // Follow-ups share the child thread, so scope each request to its run's task. + const node = child.nodes.find((node) => node.id === request.nodeId); + const run = child.runs.find((run) => run.id === node?.runId); + if (run === undefined) continue; + const taskId = + run.delegatedTaskId ?? + (child.thread.forkedFrom?.type === "node" + ? child.thread.forkedFrom.nodeId + : undefined); + if (taskId !== task.id) continue; + yield* wakeParent(task, request); + } + } + return; + } if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") { return; } @@ -86,27 +172,7 @@ const make = Effect.gen(function* () { ) { return; } - const label = reportLabel(task.title?.trim() || task.prompt); - const situation = - event.payload.responseCapability.type === "message" - ? `Delegated task ${taskId} asked a question and is still working.` - : `Delegated task ${taskId} 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: parentThreadId, - messageId, - text: `${situation} Read it with t3_pending_request_read using threadId ${child.id} and requestId ${event.payload.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: child.id }, - outcome: "updated", - summary: `Delegated task${label === undefined ? "" : ` "${label}"`} has a question`, - }, - attachments: [], - dispatchMode: { type: "queue_after_active" }, - createdBy: "agent", - creationSource: "server", - }); + yield* wakeParent(task, request); }).pipe( Effect.catchCause((cause) => Effect.logWarning("waking the parent of a delegated task question failed", { 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 efa73ea40f87..7250e05c649a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -675,21 +675,24 @@ it.effect("wakes the parent once when its delegated task asks a question", () => 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; const requestEvent = ( eventId: string, requestId: string, responseCapability: "live" | "message" = "message", kind: "user_input" | "command" = "user_input", + run = childRun, ) => ({ id: EventId.make(eventId), type: "runtime-request.updated", threadId: task.childThreadId!, + runId: run.id, occurredAt: now, payload: { id: RuntimeRequestId.make(requestId), - nodeId: NodeId.make(`node:${requestId}`), + nodeId: run.rootNodeId!, providerTurnId: null, nativeRequestRef: null, kind, @@ -704,20 +707,11 @@ it.effect("wakes the parent once when its delegated task asks a question", () => }) satisfies OrchestrationV2DomainEvent; const question = requestEvent("event:question", "request:question", "live"); const asyncQuestion = requestEvent("event:async-question", "request:async-question"); - const answered = requestEvent("event:answered", "request:answered"); + const answered = requestEvent("event:answered", "request:answered", "live"); const approval = requestEvent("event:approval", "request:approval", "message", "command"); const eventSink = yield* EventSink.EventSinkV2; yield* eventSink.write({ - events: [ - question, - asyncQuestion, - { - ...answered, - id: EventId.make("event:answered-resolved"), - payload: { ...answered.payload, status: "resolved", resolvedAt: now }, - }, - approval, - ], + events: [question, asyncQuestion, answered, approval], }); yield* reactor.handle(question); @@ -735,6 +729,16 @@ it.effect("wakes the parent once when its delegated task asks a question", () => ); yield* reactor.handle(asyncQuestion); yield* reactor.handle(answered); + yield* eventSink.write({ + events: [ + { + ...answered, + id: EventId.make("event:answered-resolved"), + payload: { ...answered.payload, status: "resolved", resolvedAt: now }, + }, + ], + }); + yield* reactor.handle(answered); yield* reactor.handle(approval); const wakes = (yield* projections.getThreadProjection(parentThreadId)).messages.filter( @@ -773,18 +777,37 @@ it.effect("wakes the parent once when its delegated task asks a question", () => )?.status, "cancelled", ); - yield* eventSink.write({ - events: [ - { - id: EventId.make("event:question-parent-completed"), - type: "run.updated", - threadId: parentThreadId, - runId: parentRun.id, - occurredAt: now, - payload: { ...parentRun, status: "completed", completedAt: now }, - }, - ], - }); + const parentCompleted = { + id: EventId.make("event:question-parent-completed"), + type: "run.updated", + threadId: parentThreadId, + runId: parentRun.id, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + } satisfies OrchestrationV2DomainEvent; + yield* eventSink.write({ events: [parentCompleted] }); + yield* reactor.handle(parentCompleted); + yield* reactor.handle(parentCompleted); + const caughtUp = (yield* projections.getThreadProjection(parentThreadId)).messages.filter( + (message) => message.notification !== undefined, + ); + assert.equal(caughtUp.length, 2); + assert.equal( + caughtUp.filter((message) => message.text.includes(`requestId ${question.payload.id}.`)) + .length, + 1, + ); + assert.include(caughtUp[1]?.text, "is blocked on a question"); + assert.isFalse( + caughtUp.some((message) => message.text.includes(`requestId ${answered.payload.id}.`)), + ); + yield* reactor.handle(question); + assert.equal( + (yield* projections.getThreadProjection(parentThreadId)).messages.filter( + (message) => message.notification !== undefined, + ).length, + 2, + ); const terminalQuestion = requestEvent( "event:terminal-question", "request:terminal-question", @@ -796,7 +819,7 @@ it.effect("wakes the parent once when its delegated task asks a question", () => (yield* projections.getThreadProjection(parentThreadId)).messages.filter( (message) => message.notification !== undefined, ).length, - 2, + 3, ); yield* eventSink.write({ events: [ @@ -828,9 +851,20 @@ it.effect("wakes the parent once when its delegated task asks a question", () => (yield* projections.getThreadProjection(parentThreadId)).messages.filter( (message) => message.notification !== undefined, ).length, - 3, + 4, + ); + const waitDelivered = requestEvent("event:wait-delivered", "request:wait-delivered", "live"); + const waitDeliveredParentCompleted = { + ...parentCompleted, + id: EventId.make("event:wait-delivered-parent-completed"), + }; + yield* eventSink.write({ events: [waitDelivered, waitDeliveredParentCompleted] }); + yield* reactor.handle(waitDeliveredParentCompleted); + assert.isFalse( + (yield* projections.getThreadProjection(parentThreadId)).messages.some((message) => + message.text.includes(`requestId ${waitDelivered.payload.id}.`), + ), ); - const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!; yield* eventSink.write({ commandId: CommandId.make("command:runtime-reconcile:question-followup"), events: [ @@ -871,12 +905,42 @@ it.effect("wakes the parent once when its delegated task asks a question", () => )!; assert.isDefined(followup.delegatedTaskId); assert.notEqual(followup.delegatedTaskId, task.id); - const followupQuestion = { - ...requestEvent("event:followup-question", "request:followup-question"), - runId: followup.id, - }; + const followupQuestion = requestEvent( + "event:followup-question", + "request:followup-question", + "live", + "user_input", + followup, + ); yield* eventSink.write({ events: [followupQuestion] }); - yield* reactor.handle(followupQuestion); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("followup-question-wait"), + parentThreadId, + taskId: followup.delegatedTaskId!, + completionWake: "settled_only", + }); + const followupParent = yield* projections.getThreadRecords(parentThreadId, [ + "subagents", + "runs", + ]); + const followupTask = followupParent.subagents.find( + (row) => row.id === followup.delegatedTaskId, + )!; + const followupParentRun = followupParent.runs.find((run) => run.id === followupTask.runId)!; + const followupParentCompleted = { + ...parentCompleted, + id: EventId.make("event:followup-parent-completed"), + runId: followupParentRun.id, + payload: { ...followupParentRun, status: "completed" as const, completedAt: now }, + }; + yield* eventSink.write({ events: [followupParentCompleted] }); + yield* reactor.handle(followupParentCompleted); + assert.isFalse( + (yield* projections.getThreadProjection(parentThreadId)).messages.some((message) => + message.text.includes(`requestId ${waitDelivered.payload.id}.`), + ), + ); const followupWake = (yield* projections.getThreadRecords(parentThreadId, [ "messages", ])).messages.find((message) => @@ -887,5 +951,6 @@ it.effect("wakes the parent once when its delegated task asks a question", () => taskIds: [followup.delegatedTaskId], childThreadId: task.childThreadId!, }); + yield* reactor.handle(followupQuestion); }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), ); diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index c935b404b18c..4f72b59e653a 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -461,6 +461,8 @@ 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 +parent when its launching run ends. A failed run exposes its provider error before any progress text. Successful results use the latest assistant content from the final work turn. From 35ba9631bece433070f626efdc432bd3725119f4 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Thu, 15 Oct 2026 02:49:04 +0000 Subject: [PATCH 14/14] fix(server): keep disposed delegated questions from waking parents --- .../DelegatedTaskQuestionReactor.ts | 36 ++++------ .../Orchestrator.control-reads.test.ts | 66 ++++++++++--------- .../orchestrator-mcp-server.md | 3 +- 3 files changed, 50 insertions(+), 55 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts index c50d7b61b5f1..9d239d34b65f 100644 --- a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -67,13 +67,7 @@ const make = Effect.gen(function* () { const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = (event) => Effect.gen(function* () { if (event.type === "run.updated") { - if ( - event.payload.status === "preparing" || - event.payload.status === "starting" || - event.payload.status === "running" - ) { - return; - } + 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) { @@ -81,18 +75,14 @@ const make = Effect.gen(function* () { task.origin !== "app_owned" || task.runId !== event.payload.id || task.completionWake !== "settled_only" || + task.completionDelivery?.state === "disposed" || task.childThreadId === null || - task.status === "completed" || - task.status === "failed" || - task.status === "cancelled" || - task.status === "interrupted" + ["completed", "failed", "cancelled", "interrupted"].includes(task.status) ) { continue; } const child = yield* projections.getThreadRecords(task.childThreadId, [ "runtimeRequests", - "nodes", - "runs", ]); for (const request of child.runtimeRequests) { if ( @@ -102,17 +92,15 @@ const make = Effect.gen(function* () { ) { continue; } - // Follow-ups share the child thread, so scope each request to its run's task. - const node = child.nodes.find((node) => node.id === request.nodeId); - const run = child.runs.find((run) => run.id === node?.runId); - if (run === undefined) continue; - const taskId = - run.delegatedTaskId ?? - (child.thread.forkedFrom?.type === "node" - ? child.thread.forkedFrom.nodeId - : undefined); - if (taskId !== task.id) continue; - yield* wakeParent(task, request); + 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; 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 7250e05c649a..de616556b4eb 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -785,8 +785,42 @@ it.effect("wakes the parent once when its delegated task asks a question", () => occurredAt: now, payload: { ...parentRun, status: "completed", completedAt: now }, } satisfies OrchestrationV2DomainEvent; + for (const status of [ + "preparing", + "starting", + "running", + "queued", + "cancelled", + "rolled_back", + "interrupted", + ] as const) { + yield* reactor.handle({ ...parentCompleted, payload: { ...parentRun, status } }); + } + assert.equal( + (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.length, + 2, + ); yield* eventSink.write({ events: [parentCompleted] }); - yield* reactor.handle(parentCompleted); + for (const state of ["disposed", "pending"] as const) { + yield* eventSink.write({ + events: [ + { + id: EventId.make(`event:question-task-${state}`), + type: "subagent.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { ...task, completionDelivery: { state, observedByRunId: null } }, + }, + ], + }); + yield* reactor.handle(parentCompleted); + if (state === "disposed") { + assert.equal( + (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.length, + 2, + ); + } + } yield* reactor.handle(parentCompleted); const caughtUp = (yield* projections.getThreadProjection(parentThreadId)).messages.filter( (message) => message.notification !== undefined, @@ -913,34 +947,7 @@ it.effect("wakes the parent once when its delegated task asks a question", () => followup, ); yield* eventSink.write({ events: [followupQuestion] }); - yield* orchestrator.dispatch({ - type: "delegated_task.wake-policy", - commandId: CommandId.make("followup-question-wait"), - parentThreadId, - taskId: followup.delegatedTaskId!, - completionWake: "settled_only", - }); - const followupParent = yield* projections.getThreadRecords(parentThreadId, [ - "subagents", - "runs", - ]); - const followupTask = followupParent.subagents.find( - (row) => row.id === followup.delegatedTaskId, - )!; - const followupParentRun = followupParent.runs.find((run) => run.id === followupTask.runId)!; - const followupParentCompleted = { - ...parentCompleted, - id: EventId.make("event:followup-parent-completed"), - runId: followupParentRun.id, - payload: { ...followupParentRun, status: "completed" as const, completedAt: now }, - }; - yield* eventSink.write({ events: [followupParentCompleted] }); - yield* reactor.handle(followupParentCompleted); - assert.isFalse( - (yield* projections.getThreadProjection(parentThreadId)).messages.some((message) => - message.text.includes(`requestId ${waitDelivered.payload.id}.`), - ), - ); + yield* reactor.handle(followupQuestion); const followupWake = (yield* projections.getThreadRecords(parentThreadId, [ "messages", ])).messages.find((message) => @@ -951,6 +958,5 @@ it.effect("wakes the parent once when its delegated task asks a question", () => taskIds: [followup.delegatedTaskId], childThreadId: task.childThreadId!, }); - yield* reactor.handle(followupQuestion); }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))), ); diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 4f72b59e653a..2eb962e58819 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -462,7 +462,8 @@ 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 -parent when its launching run ends. +parent when its launching run completes, fails, or waits for post-turn work. +Stopping or interrupting the parent prevents this catch-up. A failed run exposes its provider error before any progress text. Successful results use the latest assistant content from the final work turn.