1import {
2 CommandId,
3 type MessageId,
4 type OrchestrationV2ThreadProjection,
5} from "@t3tools/contracts";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";
9import * as Ref from "effect/Ref";
10import * as Stream from "effect/Stream";
11
12import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
13import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
14import type * as Orchestrator from "./Orchestrator.ts";
15import * as ThreadManagementService from "./ThreadManagementService.ts";
16import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts";
17
18const CONTINUATION_MESSAGE_TEXT = "Background task completed.";
19
20function delegatedCompletionText(taskIds: ReadonlyArray<string>): string {
21 const taskList = taskIds.join(", ");
22 return taskIds.length === 1
23 ? `Delegated task ${taskList} reached a terminal state. Use task_status with taskId ${taskList} to read the result.`
24 : `Delegated tasks ${taskList} reached terminal states. Use task_status with each taskId to read the results.`;
25}
26
27function currentDelegatedCompletionDelivery(
28 projection: Pick<OrchestrationV2ThreadProjection, "messages" | "runs" | "providerTurns">,
29 completion: NonNullable<
30 ProviderContinuationRequests.ProviderContinuationRequest["delegatedCompletion"]
31 >,
32) {
33 const sourceRun = projection.runs.find((candidate) => candidate.id === completion.parentRunId);
34 const delivery = sourceRun?.delegatedCompletion?.delivery;
35 const alreadyDispatched = projection.messages.some(
36 (message) => message.id === completion.messageId,
37 );
38 if (
39 sourceRun?.delegatedCompletion?.disposition !== "open" ||
40 delivery === null ||
41 delivery === undefined ||
42 delivery.generation !== completion.generation ||
43 delivery.messageId !== completion.messageId ||
44 (alreadyDispatched && !isUndeliveredMailboxSteer(projection, completion.messageId))
45 ) {
46 return undefined;
47 }
48 return delivery;
49}
50
51function delegatedCompletionRetryKey(
52 request: ProviderContinuationRequests.ProviderContinuationRequest,
53 completion: NonNullable<
54 ProviderContinuationRequests.ProviderContinuationRequest["delegatedCompletion"]
55 >,
56): string {
57 return `${request.threadId}:${completion.parentRunId}:${completion.generation}:${completion.messageId}`;
58}
59
60export const layer = Layer.effectDiscard(
61 Effect.gen(function* () {
62 const ids = yield* IdAllocator.IdAllocatorV2;
63 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
64 const threads = yield* ThreadManagementService.ThreadManagementService;
65 const retryAttempts = yield* Ref.make(new Map<string, number>());
66
67 const clearRetryAttempt = (key: string) =>
68 Ref.update(retryAttempts, (current) => {
69 if (!current.has(key)) return current;
70 const updated = new Map(current);
71 updated.delete(key);
72 return updated;
73 });
74
75 const nextRetryDelay = (key: string) =>
76 Ref.modify(retryAttempts, (current) => {
77 const attempt = current.get(key) ?? 0;
78 const updated = new Map(current);
79 updated.set(key, attempt + 1);
80 return [Math.min(100 * 2 ** Math.min(attempt, 6), 5_000), updated] as const;
81 });
82
83 // Cancelling or stopping a queued continuation ends it unstarted; only the request can release the adapter's wake latch.
84 const releaseIfNeverStarted = Effect.fn("ProviderContinuationService.releaseIfNeverStarted")(
85 function* (
86 request: ProviderContinuationRequests.ProviderContinuationRequest,
87 messageId: MessageId,
88 dispatched: Orchestrator.OrchestratorV2DispatchResult,
89 ) {
90 const onRunEndedWithoutStarting = request.onRunEndedWithoutStarting;
91 if (onRunEndedWithoutStarting === undefined) return;
92 const runId = dispatched.storedEvents.flatMap(({ event }) =>
93 event.type === "run.created" && event.payload.userMessageId === messageId
94 ? [event.payload.id]
95 : [],
96 )[0];
97 if (runId === undefined) return;
98 yield* threads
99 .streamStoredEventsFrom({
100 threadId: request.threadId,
101 afterSequence: dispatched.sequence,
102 eventType: "run.updated",
103 })
104 .pipe(
105 Stream.filter(
106 ({ event }) =>
107 event.type === "run.updated" &&
108 event.payload.id === runId &&
109 ThreadManagementService.isTerminalRunStatus(event.payload.status),
110 ),
111 Stream.runHead,
112 Effect.flatMap((ended) =>
113 Option.isSome(ended) &&
114 ended.value.event.type === "run.updated" &&
115 ended.value.event.payload.startedAt === null
116 ? onRunEndedWithoutStarting()
117 : Effect.void,
118 ),
119 Effect.catchCause((cause) =>
120 Effect.logWarning("orchestration-v2.provider-continuation.run-watch-failed", {
121 threadId: request.threadId,
122 providerThreadId: request.providerThreadId,
123 cause,
124 }),
125 ),
126 Effect.forkScoped,
127 );
128 },
129 );
130
131 const dispatchContinuation = Effect.fn("ProviderContinuationService.dispatchContinuation")(
132 function* (request: ProviderContinuationRequests.ProviderContinuationRequest) {
133 const projection = yield* threads.getThreadRecords(
134 request.threadId,
135 ["messages", "runs", "providerTurns"],
136 {
137 messageIds:
138 request.delegatedCompletion === undefined
139 ? []
140 : [request.delegatedCompletion.messageId],
141 },
142 );
143 if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) {
144 yield* Effect.logInfo("orchestration-v2.provider-continuation.thread-archived", {
145 threadId: request.threadId,
146 providerThreadId: request.providerThreadId,
147 });
148 if (request.delegatedCompletion !== undefined) {
149 yield* clearRetryAttempt(
150 delegatedCompletionRetryKey(request, request.delegatedCompletion),
151 );
152 }
153 if (request.clearIfCurrent !== undefined) {
154 yield* request.clearIfCurrent();
155 } else if (request.dispatchIfCurrent !== undefined) {
156 yield* request.dispatchIfCurrent(Effect.void);
157 }
158 return;
159 }
160 if (request.delegatedCompletion !== undefined) {
161 const retryKey = delegatedCompletionRetryKey(request, request.delegatedCompletion);
162 const delivery = currentDelegatedCompletionDelivery(
163 projection,
164 request.delegatedCompletion,
165 );
166 if (delivery === undefined) {
167 yield* clearRetryAttempt(retryKey);
168 return;
169 }
170 const commandId = yield* ids.allocate.command({
171 fixtureName: "delegated-completion",
172 commandName: "dispatch",
173 });
174 yield* threads
175 .dispatch({
176 type: "message.dispatch",
177 commandId,
178 threadId: request.threadId,
179 messageId: delivery.messageId,
180 text: delegatedCompletionText(delivery.taskIds),
181 attachments: [],
182 dispatchMode: { type: "queue_after_active" },
183 createdBy: "agent",
184 creationSource: "server",
185 delegatedCompletion: {
186 parentRunId: request.delegatedCompletion.parentRunId,
187 generation: delivery.generation,
188 taskIds: delivery.taskIds,
189 },
190 })
191 .pipe(
192 Effect.catchTags({
193 ThreadManagementProviderInstanceDisabledError: (error) =>
194 Effect.logInfo(
195 "orchestration-v2.provider-continuation.delegated-completion-disabled",
196 {
197 threadId: request.threadId,
198 projectId: error.projectId,
199 instanceId: error.instanceId,
200 },
201 ).pipe(
202 Effect.andThen(
203 Effect.forEach(
204 delivery.taskIds,
205 (taskId) =>
206 ids.allocate
207 .command({
208 fixtureName: "delegated-completion",
209 commandName: "dispose-disabled",
210 })
211 .pipe(
212 Effect.flatMap((disposeCommandId) =>
213 threads.dispatch({
214 type: "delegated_task.completion-delivery.dispose",
215 commandId: disposeCommandId,
216 parentThreadId: request.threadId,
217 taskId,
218 }),
219 ),
220 ),
221 { discard: true },
222 ),
223 ),
224 Effect.andThen(
225 request.clearIfCurrent !== undefined
226 ? request.clearIfCurrent()
227 : request.dispatchIfCurrent !== undefined
228 ? request.dispatchIfCurrent(Effect.void)
229 : Effect.void,
230 ),
231 ),
232 }),
233 );
234 yield* clearRetryAttempt(retryKey);
235 return;
236 }
237 const messageId =
238 request.messageId ??
239 (yield* ids.allocate.message({
240 threadId: request.threadId,
241 ordinal: projection.messages.length + 1,
242 }));
243 const commandId = CommandId.make(`provider-continuation:${messageId}`);
244 const dispatch = threads
245 .dispatch({
246 type: "message.dispatch",
247 commandId,
248 threadId: request.threadId,
249 messageId,
250 text: request.detail ?? CONTINUATION_MESSAGE_TEXT,
251 notification: request.notification ?? {
252 source: { kind: "background_task" },
253 outcome: "updated",
254 summary: "Background activity updated",
255 },
256 attachments: [],
257 dispatchMode: { type: "queue_after_active" },
258 createdBy: "agent",
259 // Only buffered wakes use creationSource "provider"; Claude would drain an empty buffer instead of prompting message_text.
260 creationSource: request.delivery === "message_text" ? "server" : "provider",
261 })
262 .pipe(Effect.tap((dispatched) => releaseIfNeverStarted(request, messageId, dispatched)));
263 // Catch after dispatchIfCurrent so adapters see raw failures first; clearIfCurrent is idempotent and covers others.
264 const logInstanceDisabled = (
265 error: ThreadManagementService.ThreadManagementProviderInstanceDisabledError,
266 ) =>
267 Effect.logInfo("orchestration-v2.provider-continuation.instance-disabled", {
268 threadId: request.threadId,
269 providerThreadId: request.providerThreadId,
270 projectId: error.projectId,
271 instanceId: error.instanceId,
272 }).pipe(
273 Effect.andThen(
274 request.clearIfCurrent !== undefined ? request.clearIfCurrent() : Effect.void,
275 ),
276 );
277 if (request.dispatchIfCurrent === undefined) {
278 yield* dispatch.pipe(
279 Effect.catchTags({
280 ThreadManagementProviderInstanceDisabledError: logInstanceDisabled,
281 }),
282 );
283 return;
284 }
285 yield* request.dispatchIfCurrent(dispatch).pipe(
286 Effect.catchTags({
287 ThreadManagementProviderInstanceDisabledError: logInstanceDisabled,
288 }),
289 );
290 },
291 );
292
293 yield* requests.take.pipe(
294 Effect.flatMap((request) =>
295 dispatchContinuation(request).pipe(
296 Effect.catchCause((cause) =>
297 Effect.gen(function* () {
298 yield* Effect.logWarning("orchestration-v2.provider-continuation.dispatch-failed", {
299 threadId: request.threadId,
300 providerThreadId: request.providerThreadId,
301 cause,
302 });
303 if (request.delegatedCompletion !== undefined) {
304 const completion = request.delegatedCompletion;
305 const retryKey = delegatedCompletionRetryKey(request, completion);
306 const retryDelay = yield* nextRetryDelay(retryKey);
307 yield* Effect.gen(function* () {
308 yield* Effect.sleep(`${retryDelay} millis`);
309 const projection = yield* threads.getThreadRecords(
310 request.threadId,
311 ["messages", "runs", "providerTurns"],
312 {
313 messageIds:
314 request.delegatedCompletion === undefined
315 ? []
316 : [request.delegatedCompletion.messageId],
317 },
318 );
319 if (currentDelegatedCompletionDelivery(projection, completion) !== undefined) {
320 yield* requests.offer(request);
321 } else {
322 yield* clearRetryAttempt(retryKey);
323 }
324 }).pipe(
325 Effect.catchCause((retryCause) =>
326 Effect.logWarning("orchestration-v2.provider-continuation.retry-check-failed", {
327 threadId: request.threadId,
328 providerThreadId: request.providerThreadId,
329 cause: retryCause,
330 }).pipe(Effect.andThen(requests.offer(request))),
331 ),
332 Effect.forkScoped,
333 );
334 } else if (request.clearIfCurrent !== undefined) {
335 yield* request.clearIfCurrent();
336 }
337 }),
338 ),
339 ),
340 ),
341 Effect.forever,
342 Effect.forkScoped,
343 );
344 }),
345);