authorgravatar for 97427849+flamboh@users.noreply.github.comoliver <97427849+flamboh@users.noreply.github.com> 2026-10-14 20:17:48-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-15 03:17:48+00:00
logfdacb3a3804bb535e8f5c7b733c4445e8567ea62
tree3b1ae255e784f612ac0dedaa00c51bc7c27d73f5
parent924fb0c4d4941f9d186e46030ab358ebbdb064f8
signature Signed by PGP key B5690EEEBB952194

fix(server): a delegated task's question now wakes its parent agent (#19521)

Co-authored-by: maria-rcks <maria@kuuro.net>

5 files changed, 542 insertions(+), 1 deletions(-)

apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts created+185
......@@ -0,0 +1,185 @@
1import {
2 CommandId,
3 MessageId,
4 type OrchestrationV2DomainEvent,
5 type OrchestrationV2RuntimeRequest,
6 type OrchestrationV2Subagent,
7} from "@t3tools/contracts";
8import * as Context from "effect/Context";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import type * as Scope from "effect/Scope";
12import * as Stream from "effect/Stream";
13
14import { reportLabel } from "@t3tools/provider-core/server/notification";
15
16import { forkParked } from "../serverActivation.ts";
17import * as Orchestrator from "./Orchestrator.ts";
18import * as ProjectionStore from "./ProjectionStore.ts";
19
20/**
21 * Wakes the parent of an app-owned delegated task when the task asks a question. The question
22 * blocks the task until someone answers it, and the user rarely watches child threads, so the
23 * parent agent is told to answer it or to ask the user. Only questions are routed: approvals
24 * cannot be answered through the pending request tools.
25 */
26export class DelegatedTaskQuestionReactor extends Context.Service<
27 DelegatedTaskQuestionReactor,
28 {
29 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
30 readonly handle: (event: OrchestrationV2DomainEvent) => Effect.Effect<void>;
31 }
32>()("t3/orchestration-v2/DelegatedTaskQuestionReactor") {}
33
34const make = Effect.gen(function* () {
35 const engine = yield* Orchestrator.OrchestratorV2;
36 const projections = yield* ProjectionStore.ProjectionStoreV2;
37
38 const wakeParent = Effect.fnUntraced(function* (
39 task: OrchestrationV2Subagent,
40 request: OrchestrationV2RuntimeRequest,
41 ) {
42 if (task.childThreadId === null) return;
43 const key = `${task.threadId}:${request.id}`;
44 const label = reportLabel(task.title?.trim() || task.prompt);
45 const situation =
46 request.responseCapability.type === "message"
47 ? `Delegated task ${task.id} asked a question and is still working.`
48 : `Delegated task ${task.id} is blocked on a question and cannot continue until it is answered.`;
49 yield* engine.dispatch({
50 type: "message.dispatch",
51 commandId: CommandId.make(`server:delegated-task-question:${key}`),
52 threadId: task.threadId,
53 messageId: MessageId.make(`message:delegated-task-question:${key}`),
54 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.`,
55 notification: {
56 source: { kind: "delegated_task", taskIds: [task.id], childThreadId: task.childThreadId },
57 outcome: "updated",
58 summary: `Delegated task${label === undefined ? "" : ` "${label}"`} has a question`,
59 },
60 attachments: [],
61 dispatchMode: { type: "queue_after_active" },
62 createdBy: "agent",
63 creationSource: "server",
64 });
65 });
66
67 const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = (event) =>
68 Effect.gen(function* () {
69 if (event.type === "run.updated") {
70 if (!["completed", "failed", "waiting"].includes(event.payload.status)) return;
71 const parent = yield* projections.getThreadRecords(event.threadId, ["subagents"]);
72 if (parent.thread.archivedAt !== null || parent.thread.deletedAt !== null) return;
73 for (const task of parent.subagents) {
74 if (
75 task.origin !== "app_owned" ||
76 task.runId !== event.payload.id ||
77 task.completionWake !== "settled_only" ||
78 task.completionDelivery?.state === "disposed" ||
79 task.childThreadId === null ||
80 ["completed", "failed", "cancelled", "interrupted"].includes(task.status)
81 ) {
82 continue;
83 }
84 const child = yield* projections.getThreadRecords(task.childThreadId, [
85 "runtimeRequests",
86 ]);
87 for (const request of child.runtimeRequests) {
88 if (
89 request.kind !== "user_input" ||
90 request.status !== "pending" ||
91 request.responseCapability.type !== "live"
92 ) {
93 continue;
94 }
95 yield* wakeParent(task, request).pipe(
96 Effect.catchCause((cause) =>
97 Effect.logWarning("catching up a delegated task question failed", {
98 taskId: task.id,
99 requestId: request.id,
100 cause,
101 }),
102 ),
103 );
104 }
105 }
106 return;
107 }
108 if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") {
109 return;
110 }
111 const child = yield* projections.getThread(event.threadId);
112 const parentThreadId = child.lineage.parentThreadId;
113 if (parentThreadId === null) return;
114 const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id);
115 // Keyed by the request, so a request updated while still pending wakes the parent once.
116 const key = `${parentThreadId}:${event.payload.id}`;
117 const messageId = MessageId.make(`message:delegated-task-question:${key}`);
118 if (event.payload.status !== "pending" || request?.status !== "pending") {
119 const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
120 const queued = parent.runs.find(
121 (run) => run.status === "queued" && run.userMessageId === messageId,
122 );
123 if (queued !== undefined) {
124 yield* engine.dispatch({
125 type: "queued-run.cancel",
126 commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`),
127 threadId: parentThreadId,
128 runId: queued.id,
129 });
130 }
131 return;
132 }
133 const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]);
134 const { runs } = yield* projections.getThreadRecords(child.id, ["runs"]);
135 const run =
136 runs.find((run) => run.id === event.runId) ??
137 runs.toSorted((a, b) => b.ordinal - a.ordinal)[0];
138 const taskId =
139 run?.delegatedTaskId ??
140 (child.forkedFrom?.type === "node" ? child.forkedFrom.nodeId : undefined);
141 const task = parent.subagents.find(
142 (candidate) => candidate.origin === "app_owned" && candidate.id === taskId,
143 );
144 if (
145 task === undefined ||
146 parent.thread.archivedAt !== null ||
147 parent.thread.deletedAt !== null
148 ) {
149 return;
150 }
151 // A blocking delegate_task wait delivers the question itself.
152 if (
153 task.completionWake === "settled_only" &&
154 event.payload.responseCapability.type === "live" &&
155 parent.runs.some(
156 (run) =>
157 run.id === task.runId &&
158 (run.status === "preparing" || run.status === "starting" || run.status === "running"),
159 )
160 ) {
161 return;
162 }
163 yield* wakeParent(task, request);
164 }).pipe(
165 Effect.catchCause((cause) =>
166 Effect.logWarning("waking the parent of a delegated task question failed", {
167 threadId: event.threadId,
168 cause,
169 }),
170 ),
171 );
172
173 const start: DelegatedTaskQuestionReactor["Service"]["start"] = () =>
174 forkParked(
175 Stream.runForEach(engine.streamDomainEvents, handle).pipe(
176 Effect.catchCause((cause) =>
177 Effect.logWarning("delegated task question event stream failed", { cause }),
178 ),
179 ),
180 );
181
182 return { start, handle } satisfies DelegatedTaskQuestionReactor["Service"];
183});
184
185export const layer = Layer.effect(DelegatedTaskQuestionReactor, make);
apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts+336
......@@ -4,6 +4,7 @@ import {
44 MessageId,
55 EventId,
66 NodeId,
7 type OrchestrationV2DomainEvent,
78 PlanId,
89 ProjectId,
910 ProviderDriverKind,
......@@ -23,6 +24,8 @@ import * as Layer from "effect/Layer";
2324import * as SqlClient from "effect/sql/SqlClient";
2425import * as SqlitePersistence from "../persistence/Sqlite.ts";
2526import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
27import * as DelegatedTaskQuestionReactor from "./DelegatedTaskQuestionReactor.ts";
28import * as EventSink from "./EventSink.ts";
2629import * as Orchestrator from "./Orchestrator.ts";
2730import * as ProjectionStore from "./ProjectionStore.ts";
2831import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
......@@ -624,3 +627,336 @@ it.effect("keeps delegated child pull-request links independent of the parent",
624627 assert.deepEqual(parentAfterChildLink.thread.pullRequests, parent.thread.pullRequests);
625628 }).pipe(Effect.provide(layerTest)),
626629);
630
631it.effect("wakes the parent once when its delegated task asks a question", () =>
632 Effect.gen(function* () {
633 const orchestrator = yield* Orchestrator.OrchestratorV2;
634 const projections = yield* ProjectionStore.ProjectionStoreV2;
635 const reactor = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor;
636 const parentThreadId = ThreadId.make("thread:question-parent");
637 yield* orchestrator.dispatch({
638 type: "thread.create",
639 commandId: CommandId.make("create-question-parent"),
640 threadId: parentThreadId,
641 projectId: ProjectId.make("project:question"),
642 title: "Parent",
643 modelSelection,
644 runtimeMode: "full-access",
645 interactionMode: "default",
646 branch: null,
647 worktreePath: null,
648 createdBy: "user",
649 creationSource: "web",
650 });
651 yield* orchestrator.dispatch({
652 type: "message.dispatch",
653 commandId: CommandId.make("start-question-parent"),
654 threadId: parentThreadId,
655 messageId: MessageId.make("message:question-parent"),
656 text: "Delegate the research",
657 attachments: [],
658 dispatchMode: { type: "start_immediately" },
659 createdBy: "user",
660 creationSource: "web",
661 });
662 const parentRun = (yield* projections.getThreadProjection(parentThreadId)).runs[0]!;
663 yield* orchestrator.dispatch({
664 type: "delegated_task.request",
665 commandId: CommandId.make("delegate-question-task"),
666 parentThreadId,
667 parentRunId: parentRun.id,
668 parentNodeId: parentRun.rootNodeId!,
669 task: "Research the sources",
670 completionWake: "settled_only",
671 modelSelection,
672 runtimeMode: "full-access",
673 interactionMode: "default",
674 createdBy: "agent",
675 creationSource: "mcp",
676 });
677 const task = (yield* projections.getThreadProjection(parentThreadId)).subagents[0]!;
678 const childRun = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs[0]!;
679 const now = yield* DateTime.now;
680 const requestEvent = (
681 eventId: string,
682 requestId: string,
683 responseCapability: "live" | "message" = "message",
684 kind: "user_input" | "command" = "user_input",
685 run = childRun,
686 ) =>
687 ({
688 id: EventId.make(eventId),
689 type: "runtime-request.updated",
690 threadId: task.childThreadId!,
691 runId: run.id,
692 occurredAt: now,
693 payload: {
694 id: RuntimeRequestId.make(requestId),
695 nodeId: run.rootNodeId!,
696 providerTurnId: null,
697 nativeRequestRef: null,
698 kind,
699 status: "pending",
700 responseCapability:
701 responseCapability === "live"
702 ? { type: "live", providerSessionId: ProviderSessionId.make("session:question") }
703 : { type: "message" },
704 createdAt: now,
705 resolvedAt: null,
706 },
707 }) satisfies OrchestrationV2DomainEvent;
708 const question = requestEvent("event:question", "request:question", "live");
709 const asyncQuestion = requestEvent("event:async-question", "request:async-question");
710 const answered = requestEvent("event:answered", "request:answered", "live");
711 const approval = requestEvent("event:approval", "request:approval", "message", "command");
712 const eventSink = yield* EventSink.EventSinkV2;
713 yield* eventSink.write({
714 events: [question, asyncQuestion, answered, approval],
715 });
716
717 yield* reactor.handle(question);
718 assert.isFalse(
719 (yield* projections.getThreadProjection(parentThreadId)).messages.some(
720 (message) => message.notification !== undefined,
721 ),
722 );
723 yield* reactor.handle(asyncQuestion);
724 assert.equal(
725 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
726 (message) => message.notification !== undefined,
727 ).length,
728 1,
729 );
730 yield* reactor.handle(asyncQuestion);
731 yield* reactor.handle(answered);
732 yield* eventSink.write({
733 events: [
734 {
735 ...answered,
736 id: EventId.make("event:answered-resolved"),
737 payload: { ...answered.payload, status: "resolved", resolvedAt: now },
738 },
739 ],
740 });
741 yield* reactor.handle(answered);
742 yield* reactor.handle(approval);
743
744 const wakes = (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
745 (message) => message.notification !== undefined,
746 );
747 assert.include(wakes[0]?.text, "asked a question and is still working");
748 assert.deepEqual(
749 wakes.map((message) => message.notification),
750 [
751 {
752 source: {
753 kind: "delegated_task",
754 taskIds: [task.id],
755 childThreadId: task.childThreadId!,
756 },
757 outcome: "updated",
758 summary: 'Delegated task "Research the sources" has a question',
759 },
760 ],
761 );
762
763 const wakeRun = (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find(
764 (run) => run.userMessageId === wakes[0]!.id,
765 )!;
766 assert.equal(wakeRun.status, "queued");
767 const resolved = {
768 ...asyncQuestion,
769 id: EventId.make("event:question-resolved"),
770 payload: { ...asyncQuestion.payload, status: "resolved" as const, resolvedAt: now },
771 };
772 yield* eventSink.write({ events: [resolved] });
773 yield* reactor.handle(resolved);
774 assert.equal(
775 (yield* projections.getThreadRecords(parentThreadId, ["runs"])).runs.find(
776 (run) => run.id === wakeRun.id,
777 )?.status,
778 "cancelled",
779 );
780 const parentCompleted = {
781 id: EventId.make("event:question-parent-completed"),
782 type: "run.updated",
783 threadId: parentThreadId,
784 runId: parentRun.id,
785 occurredAt: now,
786 payload: { ...parentRun, status: "completed", completedAt: now },
787 } satisfies OrchestrationV2DomainEvent;
788 for (const status of [
789 "preparing",
790 "starting",
791 "running",
792 "queued",
793 "cancelled",
794 "rolled_back",
795 "interrupted",
796 ] as const) {
797 yield* reactor.handle({ ...parentCompleted, payload: { ...parentRun, status } });
798 }
799 assert.equal(
800 (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.length,
801 2,
802 );
803 yield* eventSink.write({ events: [parentCompleted] });
804 for (const state of ["disposed", "pending"] as const) {
805 yield* eventSink.write({
806 events: [
807 {
808 id: EventId.make(`event:question-task-${state}`),
809 type: "subagent.updated",
810 threadId: parentThreadId,
811 occurredAt: now,
812 payload: { ...task, completionDelivery: { state, observedByRunId: null } },
813 },
814 ],
815 });
816 yield* reactor.handle(parentCompleted);
817 if (state === "disposed") {
818 assert.equal(
819 (yield* projections.getThreadRecords(parentThreadId, ["messages"])).messages.length,
820 2,
821 );
822 }
823 }
824 yield* reactor.handle(parentCompleted);
825 const caughtUp = (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
826 (message) => message.notification !== undefined,
827 );
828 assert.equal(caughtUp.length, 2);
829 assert.equal(
830 caughtUp.filter((message) => message.text.includes(`requestId ${question.payload.id}.`))
831 .length,
832 1,
833 );
834 assert.include(caughtUp[1]?.text, "is blocked on a question");
835 assert.isFalse(
836 caughtUp.some((message) => message.text.includes(`requestId ${answered.payload.id}.`)),
837 );
838 yield* reactor.handle(question);
839 assert.equal(
840 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
841 (message) => message.notification !== undefined,
842 ).length,
843 2,
844 );
845 const terminalQuestion = requestEvent(
846 "event:terminal-question",
847 "request:terminal-question",
848 "live",
849 );
850 yield* eventSink.write({ events: [terminalQuestion] });
851 yield* reactor.handle(terminalQuestion);
852 assert.equal(
853 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
854 (message) => message.notification !== undefined,
855 ).length,
856 3,
857 );
858 yield* eventSink.write({
859 events: [
860 {
861 id: EventId.make("event:question-parent-running-again"),
862 type: "run.updated",
863 threadId: parentThreadId,
864 runId: parentRun.id,
865 occurredAt: now,
866 payload: { ...parentRun, status: "running", completedAt: null },
867 },
868 ],
869 });
870 yield* orchestrator.dispatch({
871 type: "delegated_task.wake-policy",
872 commandId: CommandId.make("terminal-question-wait-ended"),
873 parentThreadId,
874 taskId: task.id,
875 completionWake: "always",
876 });
877 const upgradedQuestion = requestEvent(
878 "event:upgraded-question",
879 "request:upgraded-question",
880 "live",
881 );
882 yield* eventSink.write({ events: [upgradedQuestion] });
883 yield* reactor.handle(upgradedQuestion);
884 assert.equal(
885 (yield* projections.getThreadProjection(parentThreadId)).messages.filter(
886 (message) => message.notification !== undefined,
887 ).length,
888 4,
889 );
890 const waitDelivered = requestEvent("event:wait-delivered", "request:wait-delivered", "live");
891 const waitDeliveredParentCompleted = {
892 ...parentCompleted,
893 id: EventId.make("event:wait-delivered-parent-completed"),
894 };
895 yield* eventSink.write({ events: [waitDelivered, waitDeliveredParentCompleted] });
896 yield* reactor.handle(waitDeliveredParentCompleted);
897 assert.isFalse(
898 (yield* projections.getThreadProjection(parentThreadId)).messages.some((message) =>
899 message.text.includes(`requestId ${waitDelivered.payload.id}.`),
900 ),
901 );
902 yield* eventSink.write({
903 commandId: CommandId.make("command:runtime-reconcile:question-followup"),
904 events: [
905 {
906 id: EventId.make("event:question-child-completed"),
907 type: "run.updated",
908 threadId: task.childThreadId!,
909 occurredAt: now,
910 payload: { ...childRun, status: "completed", completedAt: now },
911 },
912 {
913 id: EventId.make("event:question-task-completed"),
914 type: "subagent.updated",
915 threadId: parentThreadId,
916 occurredAt: now,
917 payload: {
918 ...task,
919 status: "completed",
920 result: "Sources researched.",
921 completedAt: now,
922 },
923 },
924 ],
925 });
926 yield* orchestrator.dispatch({
927 type: "message.dispatch",
928 commandId: CommandId.make("question-followup"),
929 threadId: task.childThreadId!,
930 messageId: MessageId.make("message:question-followup"),
931 text: "Check the sources again",
932 attachments: [],
933 dispatchMode: { type: "start_immediately" },
934 createdBy: "user",
935 creationSource: "web",
936 });
937 const followup = (yield* projections.getThreadRecords(task.childThreadId!, ["runs"])).runs.find(
938 (run) => run.userMessageId === "message:question-followup",
939 )!;
940 assert.isDefined(followup.delegatedTaskId);
941 assert.notEqual(followup.delegatedTaskId, task.id);
942 const followupQuestion = requestEvent(
943 "event:followup-question",
944 "request:followup-question",
945 "live",
946 "user_input",
947 followup,
948 );
949 yield* eventSink.write({ events: [followupQuestion] });
950 yield* reactor.handle(followupQuestion);
951 const followupWake = (yield* projections.getThreadRecords(parentThreadId, [
952 "messages",
953 ])).messages.find((message) =>
954 message.text.includes(`requestId ${followupQuestion.payload.id}`),
955 )!;
956 assert.deepEqual(followupWake.notification?.source, {
957 kind: "delegated_task",
958 taskIds: [followup.delegatedTaskId],
959 childThreadId: task.childThreadId!,
960 });
961 }).pipe(Effect.provide(DelegatedTaskQuestionReactor.layer.pipe(Layer.provideMerge(layerTest)))),
962);
apps/server/src/server.ts+7
......@@ -6,6 +6,7 @@ import * as Semaphore from "effect/Semaphore";
66import * as StorageCleanup from "./storageCleanup.ts";
77import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts";
88import * as PullRequestWatchReactor from "./orchestration-v2/PullRequestWatchReactor.ts";
9import * as DelegatedTaskQuestionReactor from "./orchestration-v2/DelegatedTaskQuestionReactor.ts";
910// @effect-diagnostics nodeBuiltinImport:off
1011import * as NodeHttp from "node:http";
1112
......@@ -571,6 +572,12 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll(
571572 Layer.provide(layerPullRequestService),
572573 Layer.provide(ProjectionStoreV2.layer),
573574 ),
575 Layer.effectDiscard(
576 Effect.gen(function* () {
577 const service = yield* DelegatedTaskQuestionReactor.DelegatedTaskQuestionReactor;
578 yield* service.start();
579 }),
580 ).pipe(Layer.provide(DelegatedTaskQuestionReactor.layer), Layer.provide(ProjectionStoreV2.layer)),
574581 // Subscribes to `account.rate-limits.updated` so usage bars track live
575582 // telemetry instead of waiting for the next status probe.
576583 ProviderUsageLimitsIngestion.layer,
docs/orchestration-v2/orchestrator-mcp-server.md+13
......@@ -458,6 +458,19 @@ persisted events and then follows live events, so finalization also runs after
458458a server restart. An existing `subagent_result` transfer makes finalization
459459idempotent.
460460
461Users rarely watch child threads, so a child question would otherwise go
462unanswered. When a child opens a pending `user_input` request, the parent gets
463a notification message telling it to answer with the pending request tools or
464ask the user. The message says whether the child is blocked or, for a Codex
465async question, still working. If the question is answered or dismissed before
466the notification starts a turn, that queued turn is cancelled. Approvals are not
467routed, because those tools cannot answer them.
468A parent still blocked in a `delegate_task` wait gets the blocking question
469through that wait instead.
470If the provider cuts that wait short, a still-pending live question wakes the
471parent when its launching run completes, fails, or waits for post-turn work.
472Stopping or interrupting the parent prevents this catch-up.
473
461474A failed run exposes its provider error before any progress text. Successful
462475results use the latest assistant content from the final work turn.
463476
packages/provider-core/src/server/notification.ts+1-1
......@@ -25,7 +25,7 @@ export type BackgroundWorkReport = BackgroundWork & { readonly outcome: Notifica
2525
2626const LABEL_MAX_LENGTH = 80;
2727
28function reportLabel(label: string | undefined): string | undefined {
28export function reportLabel(label: string | undefined): string | undefined {
2929 const firstLine = label?.trim().split("\n")[0]?.trim();
3030 if (firstLine === undefined || firstLine.length === 0) return undefined;
3131 return firstLine.length > LABEL_MAX_LENGTH