1import { CommandId, type OrchestrationV2ThreadProjection } from "@t3tools/contracts";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Ref from "effect/Ref";
5
6import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
7import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
8import * as ThreadManagementService from "./ThreadManagementService.ts";
9import { isUndeliveredMailboxSteer } from "./NotificationMailbox.ts";
10
11const CONTINUATION_MESSAGE_TEXT = "Background task completed.";
12
13function delegatedCompletionText(taskIds: ReadonlyArray<string>): string {
14 const taskList = taskIds.join(", ");
15 return taskIds.length === 1
16 ? `Delegated task ${taskList} reached a terminal state. Use task_status with taskId ${taskList} to read the result.`
17 : `Delegated tasks ${taskList} reached terminal states. Use task_status with each taskId to read the results.`;
18}
19
20function currentDelegatedCompletionDelivery(
21 projection: Pick<OrchestrationV2ThreadProjection, "messages" | "runs" | "providerTurns">,
22 completion: NonNullable<
23 ProviderContinuationRequests.ProviderContinuationRequest["delegatedCompletion"]
24 >,
25) {
26 const sourceRun = projection.runs.find((candidate) => candidate.id === completion.parentRunId);
27 const delivery = sourceRun?.delegatedCompletion?.delivery;
28 const alreadyDispatched = projection.messages.some(
29 (message) => message.id === completion.messageId,
30 );
31 if (
32 sourceRun?.delegatedCompletion?.disposition !== "open" ||
33 delivery === null ||
34 delivery === undefined ||
35 delivery.generation !== completion.generation ||
36 delivery.messageId !== completion.messageId ||
37 (alreadyDispatched && !isUndeliveredMailboxSteer(projection, completion.messageId))
38 ) {
39 return undefined;
40 }
41 return delivery;
42}
43
44function delegatedCompletionRetryKey(
45 request: ProviderContinuationRequests.ProviderContinuationRequest,
46 completion: NonNullable<
47 ProviderContinuationRequests.ProviderContinuationRequest["delegatedCompletion"]
48 >,
49): string {
50 return `${request.threadId}:${completion.parentRunId}:${completion.generation}:${completion.messageId}`;
51}
52
53/**
54 * Drains ProviderContinuationRequests and dispatches an internal
55 * message.dispatch per request so the wake turn buffered by the adapter is
56 * ingested as a normal run. Delegated completions are durable mailbox offers:
57 * the orchestrator selects native steering or queued delivery under its thread
58 * lock. Adapter-buffered continuations still queue behind active work.
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 const dispatchContinuation = Effect.fn("ProviderContinuationService.dispatchContinuation")(
84 function* (request: ProviderContinuationRequests.ProviderContinuationRequest) {
85 const projection = yield* threads.getThreadRecords(
86 request.threadId,
87 ["messages", "runs", "providerTurns"],
88 {
89 messageIds:
90 request.delegatedCompletion === undefined
91 ? []
92 : [request.delegatedCompletion.messageId],
93 },
94 );
95 if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) {
96 yield* Effect.logInfo("orchestration-v2.provider-continuation.thread-archived", {
97 threadId: request.threadId,
98 providerThreadId: request.providerThreadId,
99 });
100 if (request.delegatedCompletion !== undefined) {
101 yield* clearRetryAttempt(
102 delegatedCompletionRetryKey(request, request.delegatedCompletion),
103 );
104 }
105 // No continuation turn will start to clear the adapter's sticky offer.
106 if (request.clearIfCurrent !== undefined) {
107 yield* request.clearIfCurrent();
108 } else if (request.dispatchIfCurrent !== undefined) {
109 // Backward compatibility for request producers without an explicit
110 // drop callback.
111 yield* request.dispatchIfCurrent(Effect.void);
112 }
113 return;
114 }
115 if (request.delegatedCompletion !== undefined) {
116 const retryKey = delegatedCompletionRetryKey(request, request.delegatedCompletion);
117 const delivery = currentDelegatedCompletionDelivery(
118 projection,
119 request.delegatedCompletion,
120 );
121 if (delivery === undefined) {
122 yield* clearRetryAttempt(retryKey);
123 return;
124 }
125 const commandId = yield* ids.allocate.command({
126 fixtureName: "delegated-completion",
127 commandName: "dispatch",
128 });
129 yield* threads
130 .dispatch({
131 type: "message.dispatch",
132 commandId,
133 threadId: request.threadId,
134 messageId: delivery.messageId,
135 text: delegatedCompletionText(delivery.taskIds),
136 attachments: [],
137 dispatchMode: { type: "queue_after_active" },
138 createdBy: "agent",
139 creationSource: "server",
140 delegatedCompletion: {
141 parentRunId: request.delegatedCompletion.parentRunId,
142 generation: delivery.generation,
143 taskIds: delivery.taskIds,
144 },
145 })
146 .pipe(
147 // The project turned the target instance off between the cohort
148 // read above and this dispatch. The finalize that claimed this
149 // delivery already wrote the children's results to the parent's
150 // history unconditionally, so nothing is lost: acknowledge each
151 // task administratively (observedByRunId: null, the same shape
152 // task_status uses when no run observed it) to consume this
153 // delivery instead of leaving it open to retry. The parent's
154 // next turn against an enabled instance reads those results from
155 // history the same way any turn does; nothing here re-offers
156 // this generation.
157 Effect.catchTags({
158 ThreadManagementProviderInstanceDisabledError: (error) =>
159 Effect.logInfo(
160 "orchestration-v2.provider-continuation.delegated-completion-disabled",
161 {
162 threadId: request.threadId,
163 projectId: error.projectId,
164 instanceId: error.instanceId,
165 },
166 ).pipe(
167 Effect.andThen(
168 Effect.forEach(
169 delivery.taskIds,
170 (taskId) =>
171 ids.allocate
172 .command({
173 fixtureName: "delegated-completion",
174 commandName: "acknowledge-disabled",
175 })
176 .pipe(
177 Effect.flatMap((ackCommandId) =>
178 threads.dispatch({
179 type: "delegated_task.completion-delivery.acknowledge",
180 commandId: ackCommandId,
181 parentThreadId: request.threadId,
182 taskId,
183 observedByRunId: null,
184 }),
185 ),
186 ),
187 { discard: true },
188 ),
189 ),
190 Effect.andThen(
191 request.clearIfCurrent !== undefined
192 ? request.clearIfCurrent()
193 : request.dispatchIfCurrent !== undefined
194 ? request.dispatchIfCurrent(Effect.void)
195 : Effect.void,
196 ),
197 ),
198 }),
199 );
200 yield* clearRetryAttempt(retryKey);
201 return;
202 }
203 // The ordinal is display metadata only: allocate.message appends a
204 // random UUID, so a stale projection read here cannot collide ids.
205 const messageId =
206 request.messageId ??
207 (yield* ids.allocate.message({
208 threadId: request.threadId,
209 ordinal: projection.messages.length + 1,
210 }));
211 const commandId = CommandId.make(`provider-continuation:${messageId}`);
212 const dispatch = threads.dispatch({
213 type: "message.dispatch",
214 commandId,
215 threadId: request.threadId,
216 messageId,
217 text: request.detail ?? CONTINUATION_MESSAGE_TEXT,
218 notification: request.notification ?? {
219 source: { kind: "background_task" },
220 outcome: "updated",
221 summary: "Background activity updated",
222 },
223 attachments: [],
224 dispatchMode: { type: "queue_after_active" },
225 createdBy: "agent",
226 // "provider" marks an adapter-buffered wake, which ClaudeAdapterV2
227 // detects to attach the buffered CLI output and drop this text. A
228 // message_text wake has no buffered output, so it must not carry that
229 // marker or the turn settles immediately having prompted nothing.
230 creationSource: request.delivery === "message_text" ? "server" : "provider",
231 });
232 // The adapter's buffered wake has nowhere else to go once the
233 // instance is off: drop it rather than warn, since nothing here
234 // retries it either way. This catch sits outside dispatchIfCurrent so
235 // an adapter that clears its own sticky offer flag on dispatch
236 // failure (inside dispatchIfCurrent) observes the raw failure first.
237 // Not every adapter self-clears that way, so the explicit callback is
238 // called here too; clearIfCurrent implementations are idempotent, so
239 // a source that already cleared is unaffected.
240 const logInstanceDisabled = (
241 error: ThreadManagementService.ThreadManagementProviderInstanceDisabledError,
242 ) =>
243 Effect.logInfo("orchestration-v2.provider-continuation.instance-disabled", {
244 threadId: request.threadId,
245 providerThreadId: request.providerThreadId,
246 projectId: error.projectId,
247 instanceId: error.instanceId,
248 }).pipe(
249 Effect.andThen(
250 request.clearIfCurrent !== undefined ? request.clearIfCurrent() : Effect.void,
251 ),
252 );
253 if (request.dispatchIfCurrent === undefined) {
254 yield* dispatch.pipe(
255 Effect.catchTags({
256 ThreadManagementProviderInstanceDisabledError: logInstanceDisabled,
257 }),
258 );
259 return;
260 }
261 yield* request.dispatchIfCurrent(dispatch).pipe(
262 Effect.catchTags({
263 ThreadManagementProviderInstanceDisabledError: logInstanceDisabled,
264 }),
265 );
266 },
267 );
268
269 yield* requests.take.pipe(
270 Effect.flatMap((request) =>
271 dispatchContinuation(request).pipe(
272 Effect.catchCause((cause) =>
273 Effect.gen(function* () {
274 yield* Effect.logWarning("orchestration-v2.provider-continuation.dispatch-failed", {
275 threadId: request.threadId,
276 providerThreadId: request.providerThreadId,
277 cause,
278 });
279 if (request.delegatedCompletion !== undefined) {
280 const completion = request.delegatedCompletion;
281 const retryKey = delegatedCompletionRetryKey(request, completion);
282 const retryDelay = yield* nextRetryDelay(retryKey);
283 yield* Effect.gen(function* () {
284 yield* Effect.sleep(`${retryDelay} millis`);
285 const projection = yield* threads.getThreadRecords(
286 request.threadId,
287 ["messages", "runs", "providerTurns"],
288 {
289 messageIds:
290 request.delegatedCompletion === undefined
291 ? []
292 : [request.delegatedCompletion.messageId],
293 },
294 );
295 if (currentDelegatedCompletionDelivery(projection, completion) !== undefined) {
296 yield* requests.offer(request);
297 } else {
298 yield* clearRetryAttempt(retryKey);
299 }
300 }).pipe(
301 Effect.catchCause((retryCause) =>
302 Effect.logWarning("orchestration-v2.provider-continuation.retry-check-failed", {
303 threadId: request.threadId,
304 providerThreadId: request.providerThreadId,
305 cause: retryCause,
306 }).pipe(Effect.andThen(requests.offer(request))),
307 ),
308 Effect.forkScoped,
309 );
310 }
311 }),
312 ),
313 ),
314 ),
315 Effect.forever,
316 Effect.forkScoped,
317 );
318 }),
319);