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);