diff --git a/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts new file mode 100644 index 0000000000000000000000000000000000000000..9d239d34b65f03377b9d144d756e74dfada57b20 --- /dev/null +++ b/apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts @@ -0,0 +1,185 @@ +import { + CommandId, + MessageId, + type OrchestrationV2DomainEvent, + type OrchestrationV2RuntimeRequest, + type OrchestrationV2Subagent, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; + +import { reportLabel } from "@t3tools/provider-core/server/notification"; + +import { forkParked } from "../serverActivation.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; + +/** + * 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") {} + +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 (!["completed", "failed", "waiting"].includes(event.payload.status)) return; + const parent = yield* projections.getThreadRecords(event.threadId, ["subagents"]); + if (parent.thread.archivedAt !== null || parent.thread.deletedAt !== null) return; + for (const task of parent.subagents) { + if ( + task.origin !== "app_owned" || + task.runId !== event.payload.id || + task.completionWake !== "settled_only" || + task.completionDelivery?.state === "disposed" || + task.childThreadId === null || + ["completed", "failed", "cancelled", "interrupted"].includes(task.status) + ) { + continue; + } + const child = yield* projections.getThreadRecords(task.childThreadId, [ + "runtimeRequests", + ]); + for (const request of child.runtimeRequests) { + if ( + request.kind !== "user_input" || + request.status !== "pending" || + request.responseCapability.type !== "live" + ) { + continue; + } + yield* wakeParent(task, request).pipe( + Effect.catchCause((cause) => + Effect.logWarning("catching up a delegated task question failed", { + taskId: task.id, + requestId: request.id, + cause, + }), + ), + ); + } + } + return; + } + if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") { + return; + } + const child = yield* projections.getThread(event.threadId); + const parentThreadId = child.lineage.parentThreadId; + if (parentThreadId === null) return; + const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id); + // 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}`); + if (event.payload.status !== "pending" || request?.status !== "pending") { + const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]); + const queued = parent.runs.find( + (run) => run.status === "queued" && run.userMessageId === messageId, + ); + if (queued !== undefined) { + yield* engine.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`), + threadId: parentThreadId, + runId: queued.id, + }); + } + return; + } + const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]); + const { runs } = yield* projections.getThreadRecords(child.id, ["runs"]); + const run = + runs.find((run) => run.id === event.runId) ?? + runs.toSorted((a, b) => b.ordinal - a.ordinal)[0]; + const taskId = + run?.delegatedTaskId ?? + (child.forkedFrom?.type === "node" ? child.forkedFrom.nodeId : undefined); + const task = parent.subagents.find( + (candidate) => candidate.origin === "app_owned" && candidate.id === taskId, + ); + if ( + task === undefined || + parent.thread.archivedAt !== null || + parent.thread.deletedAt !== null + ) { + return; + } + // A blocking delegate_task wait delivers the question itself. + if ( + task.completionWake === "settled_only" && + event.payload.responseCapability.type === "live" && + parent.runs.some( + (run) => + run.id === task.runId && + (run.status === "preparing" || run.status === "starting" || run.status === "running"), + ) + ) { + return; + } + yield* wakeParent(task, request); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("waking the parent of a delegated task question failed", { + threadId: event.threadId, + cause, + }), + ), + ); + + const start: DelegatedTaskQuestionReactor["Service"]["start"] = () => + forkParked( + Stream.runForEach(engine.streamDomainEvents, handle).pipe( + Effect.catchCause((cause) => + Effect.logWarning("delegated task question event stream failed", { cause }), + ), + ), + ); + + return { start, handle } satisfies DelegatedTaskQuestionReactor["Service"]; +}); + +export const layer = Layer.effect(DelegatedTaskQuestionReactor, make); 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 8e5fca79cf5534da2b7d6b5845477d195e4fd366..de616556b4eb06fbbb63bbe9bffc443235ed7081 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,8 @@ 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 EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; @@ -624,3 +627,336 @@ 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", + completionWake: "settled_only", + modelSelection, + runtimeMode: "full-access", + 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; + 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: run.rootNodeId!, + providerTurnId: null, + nativeRequestRef: null, + kind, + status: "pending", + 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", "live"); + const asyncQuestion = requestEvent("event:async-question", "request:async-question"); + 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, approval], + }); + + yield* reactor.handle(question); + assert.isFalse( + (yield* projections.getThreadProjection(parentThreadId)).messages.some( + (message) => message.notification !== undefined, + ), + ); + 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* 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( + (message) => message.notification !== undefined, + ); + assert.include(wakes[0]?.text, "asked a question and is still working"); + 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', + }, + ], + ); + + const wakeRun = (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find( + (run) => run.userMessageId === wakes[0]!.id, + )!; + assert.equal(wakeRun.status, "queued"); + const resolved = { + ...asyncQuestion, + id: EventId.make("event:question-resolved"), + payload: { ...asyncQuestion.payload, status: "resolved" as const, resolvedAt: now }, + }; + yield* eventSink.write({ events: [resolved] }); + yield* reactor.handle(resolved); + assert.equal( + (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find( + (run) => run.id === wakeRun.id, + )?.status, + "cancelled", + ); + 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; + 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] }); + 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, + ); + 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", + "live", + ); + yield* eventSink.write({ events: [terminalQuestion] }); + yield* reactor.handle(terminalQuestion); + assert.equal( + (yield* projections.getThreadProjection(parentThreadId)).messages.filter( + (message) => message.notification !== undefined, + ).length, + 3, + ); + 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, + 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}.`), + ), + ); + 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", + "live", + "user_input", + followup, + ); + yield* eventSink.write({ events: [followupQuestion] }); + yield* reactor.handle(followupQuestion); + const followupWake = (yield* projections.getThreadRecords(parentThreadId, [ + "messages", + ])).messages.find((message) => + message.text.includes(`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/apps/server/src/server.ts b/apps/server/src/server.ts index 8148479f8318447e1153174c06862f0dbf793ac0..23981a932e0524e226470ef2afdc991b4b574f3d 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 8b701832440355d0aa74325c54485cf48d6d3256..9b9412af654209583445f2da251003ba6442a9b0 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -458,6 +458,19 @@ persisted events and then follows live events, so finalization also runs after a server restart. An existing `subagent_result` transfer makes finalization idempotent. +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. If the question is answered or dismissed before +the notification starts a turn, that queued turn is cancelled. Approvals are not +routed, because those tools cannot answer them. +A parent still blocked in a `delegate_task` wait gets the blocking question +through that wait instead. +If the provider cuts that wait short, a still-pending live question wakes the +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. diff --git a/packages/provider-core/src/server/notification.ts b/packages/provider-core/src/server/notification.ts index b90fb02aaf7a56dab244665c8a37205955cb1164..87acef16f2ea26752bf7d53a8e552fb04a4a729e 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