1import type {
2 EventId,
3 MessageId,
4 ModelSelection,
5 NodeId,
6 OrchestrationV2AppThread,
7 OrchestrationV2Actor,
8 OrchestrationV2ConversationMessage,
9 OrchestrationV2CreationSource,
10 OrchestrationV2DomainEvent,
11 OrchestrationV2ProviderRef,
12 OrchestrationV2Run,
13 OrchestrationV2ThreadProjection,
14 OrchestrationV2TurnItem,
15 ProviderInstanceId,
16 ProviderThreadId,
17 ProviderTurnId,
18 ThreadId,
19 TurnItemId,
20} from "@t3tools/contracts";
21import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
22import * as DateTime from "effect/DateTime";
23import * as Effect from "effect/Effect";
24import { isOrchestrationV2WorkActive } from "@t3tools/contracts";
25
26function trimmed(value: string | null | undefined): string | undefined {
27 const result = value?.trim();
28 return result && result.length > 0 ? result : undefined;
29}
30
31export function subagentThreadTitle(input: {
32 readonly parentTitle: string;
33 readonly title?: string | null;
34 readonly prompt: string;
35 readonly ordinal: number;
36}): string {
37 const detail = trimmed(input.title) ?? trimmed(input.prompt);
38 if (detail === undefined) {
39 return `${input.parentTitle} subagent ${input.ordinal}`;
40 }
41 const clipped = detail.length > 72 ? `${detail.slice(0, 69)}...` : detail;
42 return clipped;
43}
44
45export function makeSubagentChildThread(input: {
46 readonly parentThread: OrchestrationV2AppThread;
47 readonly childThreadId: ThreadId;
48 readonly parentNodeId: NodeId;
49 readonly activeProviderThreadId: ProviderThreadId | null;
50 readonly providerInstanceId: ProviderInstanceId;
51 readonly modelSelection: ModelSelection;
52 readonly title: string;
53 readonly now: DateTime.Utc;
54 readonly createdBy: OrchestrationV2Actor;
55 readonly creationSource: OrchestrationV2CreationSource;
56}): OrchestrationV2AppThread {
57 return {
58 ...input.parentThread,
59 createdBy: input.createdBy,
60 creationSource: input.creationSource,
61 id: input.childThreadId,
62 title: input.title,
63 linkedPullRequest: null,
64 pullRequests: [],
65 historyOrigin: undefined,
66 providerInstanceId: input.providerInstanceId,
67 modelSelection: input.modelSelection,
68 activeProviderThreadId: input.activeProviderThreadId,
69 lineage: {
70 parentThreadId: input.parentThread.id,
71 relationshipToParent: "subagent",
72 rootThreadId: input.parentThread.lineage.rootThreadId,
73 },
74 forkedFrom: {
75 type: "node",
76 nodeId: input.parentNodeId,
77 },
78 createdAt: input.now,
79 updatedAt: input.now,
80 archivedAt: null,
81 settledOverride: null,
82 settledAt: null,
83 snoozedUntil: null,
84 snoozedAt: null,
85 lastVisitedAt: null,
86 deletedAt: null,
87 };
88}
89
90export function makeSubagentConversationArtifacts(input: {
91 readonly messageId: MessageId;
92 readonly senderThreadId?: ThreadId;
93 readonly turnItemId: TurnItemId;
94 readonly threadId: ThreadId;
95 readonly rootNodeId: NodeId;
96 readonly providerThreadId: ProviderThreadId | null;
97 readonly providerTurnId: ProviderTurnId | null;
98 readonly nativeItemRef: OrchestrationV2ProviderRef | null;
99 readonly role: "user" | "assistant";
100 readonly text: string;
101 readonly ordinal: number;
102 readonly now: DateTime.Utc;
103}): {
104 readonly message: OrchestrationV2ConversationMessage;
105 readonly turnItem: OrchestrationV2TurnItem;
106} {
107 const message: OrchestrationV2ConversationMessage = {
108 createdBy: "agent",
109 creationSource: "provider",
110 id: input.messageId,
111 threadId: input.threadId,
112 runId: null,
113 nodeId: input.rootNodeId,
114 role: input.role,
115 ...(input.role === "user" && input.senderThreadId !== undefined
116 ? { senderThreadId: input.senderThreadId }
117 : {}),
118 text: input.text,
119 attachments: [],
120 streaming: false,
121 createdAt: input.now,
122 updatedAt: input.now,
123 };
124 const base = {
125 id: input.turnItemId,
126 threadId: input.threadId,
127 runId: null,
128 nodeId: input.rootNodeId,
129 providerThreadId: input.providerThreadId,
130 providerTurnId: input.providerTurnId,
131 nativeItemRef: input.nativeItemRef,
132 parentItemId: null,
133 ordinal: input.ordinal,
134 status: "completed" as const,
135 title: null,
136 startedAt: input.now,
137 completedAt: input.now,
138 updatedAt: input.now,
139 messageId: input.messageId,
140 text: input.text,
141 };
142 const turnItem: OrchestrationV2TurnItem =
143 input.role === "user"
144 ? {
145 ...base,
146 createdBy: "agent",
147 creationSource: "provider",
148 type: "user_message",
149 ...(input.senderThreadId === undefined ? {} : { senderThreadId: input.senderThreadId }),
150 inputIntent: "turn_start",
151 attachments: [],
152 }
153 : {
154 ...base,
155 type: "assistant_message",
156 streaming: false,
157 };
158 return { message, turnItem };
159}
160
161export function subagentResultForRun(
162 projection: Pick<OrchestrationV2ThreadProjection, "messages" | "turnItems">,
163 run: OrchestrationV2Run,
164): {
165 readonly text: string;
166 readonly messageId: OrchestrationV2ConversationMessage["id"] | null;
167 readonly turnItemId: OrchestrationV2TurnItem["id"] | null;
168} {
169 const message =
170 projection.messages
171 .filter(
172 (candidate) =>
173 candidate.runId === run.id &&
174 candidate.role === "assistant" &&
175 candidate.text.trim().length > 0,
176 )
177 .toSorted(
178 (left, right) =>
179 DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
180 )[0] ?? null;
181 const turnItem =
182 projection.turnItems
183 .filter(
184 (
185 candidate,
186 ): candidate is Extract<OrchestrationV2TurnItem, { readonly type: "assistant_message" }> =>
187 candidate.runId === run.id &&
188 candidate.type === "assistant_message" &&
189 candidate.text.trim().length > 0,
190 )
191 .toSorted((left, right) => right.ordinal - left.ordinal)[0] ?? null;
192 const failure =
193 run.status === "failed"
194 ? projection.turnItems
195 .filter((item) => item.runId === run.id && item.type === "error")
196 .toSorted((left, right) => right.ordinal - left.ordinal)[0]
197 : undefined;
198 const text =
199 (failure?.type === "error" ? failure.failure.message : undefined) ??
200 message?.text ??
201 turnItem?.text ??
202 (run.status === "completed"
203 ? "Child task completed without an assistant result."
204 : `Child task ended with status ${run.status}.`);
205 return {
206 text,
207 messageId: failure === undefined ? (message?.id ?? turnItem?.messageId ?? null) : null,
208 turnItemId: failure?.id ?? turnItem?.id ?? null,
209 };
210}
211
212/** A finished turn can still own live children or queued completion follow-ups. */
213export function delegatedTaskProgress(projection: {
214 readonly runs: OrchestrationV2ThreadProjection["runs"];
215 readonly messages: ReadonlyArray<
216 Pick<OrchestrationV2ConversationMessage, "runId" | "notification">
217 >;
218 readonly subagents: ReadonlyArray<
219 Pick<OrchestrationV2ThreadProjection["subagents"][number], "status" | "completionDelivery">
220 >;
221 readonly providerThreads: ReadonlyArray<
222 Pick<OrchestrationV2ThreadProjection["providerThreads"][number], "pendingBackgroundTasks">
223 >;
224}) {
225 const terminal = (status: string) =>
226 ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes(status);
227 const monitorRuns = new Set(
228 projection.messages
229 .filter((message) => message.notification?.source.kind === "monitor")
230 .map((message) => message.runId),
231 );
232 const workRuns = projection.runs.filter(
233 (run) => !monitorRuns.has(run.id) && run.status !== "rolled_back",
234 );
235 // A held queue waits for the user to resume it (after Stop, a restart, or a
236 // provider failure), so its runs are not work the task still owes.
237 const active = workRuns.some(
238 (run) => !terminal(run.status) && !(run.status === "queued" && run.queueHeld === true),
239 );
240 const children =
241 projection.subagents.some(
242 (task) =>
243 isOrchestrationV2WorkActive(task.status) ||
244 // Publishing a child's result precedes scheduling its parent's wake.
245 // The parent still owes that follow-up even between those transactions.
246 task.completionDelivery?.state === "pending" ||
247 task.completionDelivery?.state === "claimed",
248 ) ||
249 projection.providerThreads.some((thread) => (thread.pendingBackgroundTasks?.length ?? 0) > 0);
250 const resultRun = workRuns
251 .filter((run) => terminal(run.status) && (run.startedAt !== null || run.ordinal === 1))
252 .toSorted((a, b) => (runRanAfter(a, b) ? -1 : runRanAfter(b, a) ? 1 : 0))[0];
253 return {
254 state:
255 active || resultRun === undefined
256 ? ("working" as const)
257 : children
258 ? ("waiting_for_children" as const)
259 : ("result_available" as const),
260 resultRun,
261 };
262}
263
264/** Ids that `emitted` already wrote with an event of `type`. */
265function writtenIds(
266 emitted: ReadonlyArray<OrchestrationV2DomainEvent>,
267 type: "subagent.updated" | "node.updated" | "turn-item.updated",
268): ReadonlySet<string> {
269 return new Set(emitted.flatMap((event) => (event.type === type ? [event.payload.id] : [])));
270}
271
272/**
273 * Ends a provider-native subagent, its node and its turn item once the provider
274 * process that ran it is gone, since no terminal event will come from it. Each
275 * record is skipped if already settled or already ended by `emitted`.
276 */
277export const endOrphanedNativeSubagent = <E>(input: {
278 readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "subagents" | "turnItems">;
279 readonly subagentId: NodeId;
280 readonly status: "completed" | "failed" | "interrupted" | "cancelled";
281 readonly completedAt?: DateTime.Utc;
282 readonly result?: string | null;
283 readonly now: DateTime.Utc;
284 readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>;
285 readonly allocateEventId: () => Effect.Effect<EventId, E>;
286}) =>
287 Effect.gen(function* () {
288 const { now, status, subagentId, completedAt = now } = input;
289 const subagent = input.projection.subagents.find((entry) => entry.id === subagentId);
290 if (subagent?.origin !== "provider_native") return [];
291 const node = input.projection.nodes.find((entry) => entry.id === subagentId);
292 const item = input.projection.turnItems.find(
293 (entry) => entry.type === "subagent" && entry.subagentId === subagentId,
294 );
295 const scope = {
296 threadId: subagent.threadId,
297 ...(subagent.runId === null ? {} : { runId: subagent.runId }),
298 nodeId: subagentId,
299 providerInstanceId: subagent.providerInstanceId,
300 occurredAt: now,
301 };
302 const events: Array<OrchestrationV2DomainEvent> = [];
303 if (
304 isOrchestrationV2WorkActive(subagent.status) &&
305 !writtenIds(input.emitted, "subagent.updated").has(subagentId)
306 ) {
307 events.push({
308 ...scope,
309 id: yield* input.allocateEventId(),
310 type: "subagent.updated",
311 driver: subagent.driver,
312 payload: {
313 ...subagent,
314 status,
315 completedAt,
316 updatedAt: now,
317 ...(input.result === undefined ? {} : { result: input.result }),
318 },
319 });
320 }
321 if (
322 node !== undefined &&
323 isOrchestrationV2WorkActive(node.status) &&
324 !writtenIds(input.emitted, "node.updated").has(subagentId)
325 ) {
326 events.push({
327 ...scope,
328 id: yield* input.allocateEventId(),
329 type: "node.updated",
330 payload: { ...node, status, completedAt },
331 });
332 }
333 if (
334 item !== undefined &&
335 isOrchestrationV2WorkActive(item.status) &&
336 !writtenIds(input.emitted, "turn-item.updated").has(item.id)
337 ) {
338 events.push({
339 ...scope,
340 id: yield* input.allocateEventId(),
341 type: "turn-item.updated",
342 payload: {
343 ...item,
344 status,
345 completedAt,
346 updatedAt: now,
347 ...(input.result === undefined ? {} : { result: input.result }),
348 },
349 });
350 }
351 return events;
352 });
353
354/**
355 * Ends a provider-native subagent thread's work, a runless root turn plus the
356 * items under it (Claude's live progress), which only the provider process that
357 * ran it could settle. Skips records `emitted` already ended.
358 */
359export const endRunlessRootTurns = <E>(input: {
360 readonly threadId: ThreadId;
361 readonly providerInstanceId: ProviderInstanceId;
362 readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "turnItems">;
363 readonly status: "completed" | "failed" | "interrupted" | "cancelled";
364 readonly completedAt?: DateTime.Utc;
365 readonly now: DateTime.Utc;
366 readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>;
367 readonly allocateEventId: () => Effect.Effect<EventId, E>;
368}) =>
369 Effect.gen(function* () {
370 const { now, status, completedAt = now } = input;
371 const endedNodeIds = writtenIds(input.emitted, "node.updated");
372 const endedItemIds = writtenIds(input.emitted, "turn-item.updated");
373 const scope = {
374 threadId: input.threadId,
375 providerInstanceId: input.providerInstanceId,
376 occurredAt: now,
377 };
378 const events: Array<OrchestrationV2DomainEvent> = [];
379 for (const node of input.projection.nodes) {
380 if (
381 node.kind !== "root_turn" ||
382 node.runId !== null ||
383 !isOrchestrationV2WorkActive(node.status)
384 ) {
385 continue;
386 }
387 // Recovery may have ended this node through one of its items; its other
388 // items still need ending.
389 if (!endedNodeIds.has(node.id)) {
390 events.push({
391 ...scope,
392 id: yield* input.allocateEventId(),
393 type: "node.updated",
394 nodeId: node.id,
395 payload: { ...node, status, completedAt },
396 });
397 }
398 for (const item of input.projection.turnItems) {
399 if (
400 item.nodeId !== node.id ||
401 item.runId !== null ||
402 !isOrchestrationV2WorkActive(item.status) ||
403 endedItemIds.has(item.id)
404 ) {
405 continue;
406 }
407 events.push({
408 ...scope,
409 id: yield* input.allocateEventId(),
410 type: "turn-item.updated",
411 nodeId: node.id,
412 payload: {
413 ...item,
414 status:
415 status === "completed" &&
416 (item.type === "command_execution" || item.type === "dynamic_tool")
417 ? "failed"
418 : status,
419 completedAt,
420 updatedAt: now,
421 ...(item.type === "reasoning" || item.type === "assistant_message"
422 ? { streaming: false }
423 : {}),
424 },
425 });
426 }
427 }
428 return events;
429 });