1import {
2 CommandId,
3 MessageId,
4 type OrchestrationV2DomainEvent,
5 type OrchestrationV2RuntimeRequest,
6 type OrchestrationV2Subagent,
7 type RuntimeRequestId,
8 type ThreadId,
9} from "@t3tools/contracts";
10import * as Context from "effect/Context";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import type * as Scope from "effect/Scope";
14import * as Stream from "effect/Stream";
15
16import { reportLabel } from "@t3tools/provider-core/server/notification";
17
18import { forkParked } from "../serverActivation.ts";
19import * as Orchestrator from "./Orchestrator.ts";
20import * as ProjectionStore from "./ProjectionStore.ts";
21
22export class DelegatedTaskQuestionReactor extends Context.Service<
23 DelegatedTaskQuestionReactor,
24 {
25 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
26 readonly handle: (event: OrchestrationV2DomainEvent) => Effect.Effect<void>;
27 }
28>()("t3/orchestration-v2/DelegatedTaskQuestionReactor") {}
29
30const make = Effect.gen(function* () {
31 const engine = yield* Orchestrator.OrchestratorV2;
32 const projections = yield* ProjectionStore.ProjectionStoreV2;
33
34 const wakeParent = Effect.fnUntraced(function* (
35 task: OrchestrationV2Subagent,
36 request: OrchestrationV2RuntimeRequest,
37 ) {
38 if (task.childThreadId === null) return;
39 const key = `${task.threadId}:${request.id}`;
40 const label = reportLabel(task.title?.trim() || task.prompt);
41 const situation =
42 request.responseCapability.type === "message"
43 ? `Delegated task ${task.id} asked a question and is still working.`
44 : `Delegated task ${task.id} is blocked on a question and cannot continue until it is answered.`;
45 yield* engine.dispatch({
46 type: "message.dispatch",
47 commandId: CommandId.make(`server:delegated-task-question:${key}`),
48 threadId: task.threadId,
49 messageId: MessageId.make(`message:delegated-task-question:${key}`),
50 text: `${situation} Read it with t3_pending_request action read, threadId ${task.childThreadId}, and requestId ${request.id}. Answer it with action respond when the answer follows from the task or the user's instructions; otherwise ask the user, then answer it. If it is no longer pending, it was answered or dismissed.`,
51 notification: {
52 source: { kind: "delegated_task", taskIds: [task.id], childThreadId: task.childThreadId },
53 outcome: "updated",
54 summary: `Delegated task${label === undefined ? "" : ` "${label}"`} has a question`,
55 },
56 attachments: [],
57 dispatchMode: { type: "queue_after_active" },
58 createdBy: "agent",
59 creationSource: "server",
60 });
61 });
62
63 const cancelQueuedWakes = Effect.fnUntraced(function* (
64 parentThreadId: ThreadId,
65 requestIds: ReadonlyArray<RuntimeRequestId>,
66 ) {
67 const parent = yield* projections.getThreadRecords(parentThreadId, ["runs"]);
68 for (const requestId of requestIds) {
69 const key = `${parentThreadId}:${requestId}`;
70 const messageId = MessageId.make(`message:delegated-task-question:${key}`);
71 const queued = parent.runs.find(
72 (run) => run.status === "queued" && run.userMessageId === messageId,
73 );
74 if (queued === undefined) continue;
75 yield* engine
76 .dispatch({
77 type: "queued-run.cancel",
78 commandId: CommandId.make(`server:delegated-task-question-cancel:${key}`),
79 threadId: parentThreadId,
80 runId: queued.id,
81 })
82 .pipe(
83 Effect.catchCause((cause) =>
84 Effect.logWarning("cancelling a delegated task question wake failed", {
85 threadId: parentThreadId,
86 requestId,
87 cause,
88 }),
89 ),
90 );
91 }
92 });
93
94 // Runs for every domain event; most events return before any work is needed.
95 const handle: DelegatedTaskQuestionReactor["Service"]["handle"] = Effect.fnUntraced(
96 function* (event) {
97 if (event.type === "run.updated") {
98 if (!["completed", "failed", "waiting"].includes(event.payload.status)) return;
99 const parent = yield* projections.getThreadRecords(event.threadId, ["subagents"]);
100 if (parent.thread.archivedAt !== null || parent.thread.deletedAt !== null) return;
101 for (const task of parent.subagents) {
102 if (
103 task.origin !== "app_owned" ||
104 task.runId !== event.payload.id ||
105 task.completionWake !== "settled_only" ||
106 task.completionDelivery?.state === "disposed" ||
107 task.childThreadId === null ||
108 ["completed", "failed", "cancelled", "interrupted"].includes(task.status)
109 ) {
110 continue;
111 }
112 const child = yield* projections.getThreadRecords(task.childThreadId, [
113 "runtimeRequests",
114 ]);
115 for (const request of child.runtimeRequests) {
116 if (
117 request.kind !== "user_input" ||
118 request.status !== "pending" ||
119 request.responseCapability.type !== "live"
120 ) {
121 continue;
122 }
123 yield* wakeParent(task, request).pipe(
124 Effect.catchCause((cause) =>
125 Effect.logWarning("catching up a delegated task question failed", {
126 taskId: task.id,
127 requestId: request.id,
128 cause,
129 }),
130 ),
131 );
132 }
133 }
134 return;
135 }
136 if (event.type === "subagent.updated") {
137 const task = event.payload;
138 if (
139 task.origin !== "app_owned" ||
140 task.childThreadId === null ||
141 !["completed", "failed", "cancelled", "interrupted"].includes(task.status)
142 ) {
143 return;
144 }
145 // A settled task can no longer use an answer to its still-pending questions.
146 const { runtimeRequests } = yield* projections.getThreadRecords(task.childThreadId, [
147 "runtimeRequests",
148 ]);
149 const questions = runtimeRequests.filter((request) => request.kind === "user_input");
150 if (questions.length === 0) return;
151 const child = yield* projections.getThreadRecords(task.childThreadId, ["runs", "nodes"]);
152 const taskRunIds = new Set(
153 child.runs
154 .filter(
155 (run) =>
156 run.delegatedTaskId === task.id ||
157 (run.delegatedTaskId === undefined &&
158 child.thread.forkedFrom?.type === "node" &&
159 child.thread.forkedFrom.nodeId === task.id),
160 )
161 .map((run) => run.id),
162 );
163 const taskNodeIds = new Set(
164 child.nodes
165 .filter((node) => node.runId !== null && taskRunIds.has(node.runId))
166 .map((node) => node.id),
167 );
168 yield* cancelQueuedWakes(
169 task.threadId,
170 questions
171 .filter((request) => taskNodeIds.has(request.nodeId))
172 .map((request) => request.id),
173 );
174 return;
175 }
176 if (event.type !== "runtime-request.updated" || event.payload.kind !== "user_input") {
177 return;
178 }
179 const child = yield* projections.getThread(event.threadId);
180 const parentThreadId = child.lineage.parentThreadId;
181 if (parentThreadId === null) return;
182 const request = yield* projections.getRuntimeRequest(event.threadId, event.payload.id);
183 if (event.payload.status !== "pending" || request?.status !== "pending") {
184 yield* cancelQueuedWakes(parentThreadId, [event.payload.id]);
185 return;
186 }
187 const parent = yield* projections.getThreadRecords(parentThreadId, ["subagents", "runs"]);
188 const { runs } = yield* projections.getThreadRecords(child.id, ["runs"]);
189 const run =
190 runs.find((run) => run.id === event.runId) ??
191 runs.toSorted((a, b) => b.ordinal - a.ordinal)[0];
192 const taskId =
193 run?.delegatedTaskId ??
194 (child.forkedFrom?.type === "node" ? child.forkedFrom.nodeId : undefined);
195 const task = parent.subagents.find(
196 (candidate) => candidate.origin === "app_owned" && candidate.id === taskId,
197 );
198 if (
199 task === undefined ||
200 parent.thread.archivedAt !== null ||
201 parent.thread.deletedAt !== null
202 ) {
203 return;
204 }
205 // A blocking delegate_task wait delivers the question itself.
206 if (
207 task.completionWake === "settled_only" &&
208 event.payload.responseCapability.type === "live" &&
209 parent.runs.some(
210 (run) =>
211 run.id === task.runId &&
212 (run.status === "preparing" || run.status === "starting" || run.status === "running"),
213 )
214 ) {
215 return;
216 }
217 yield* wakeParent(task, request);
218 },
219 (effect, event) =>
220 effect.pipe(
221 Effect.catchCause((cause) =>
222 Effect.logWarning("waking the parent of a delegated task question failed", {
223 threadId: event.threadId,
224 cause,
225 }),
226 ),
227 ),
228 );
229
230 const start: DelegatedTaskQuestionReactor["Service"]["start"] = () =>
231 forkParked(
232 Stream.runForEach(engine.streamDomainEvents, handle).pipe(
233 Effect.catchCause((cause) =>
234 Effect.logWarning("delegated task question event stream failed", { cause }),
235 ),
236 ),
237 );
238
239 return { start, handle } satisfies DelegatedTaskQuestionReactor["Service"];
240});
241
242export const layer = Layer.effect(DelegatedTaskQuestionReactor, make);