1import { projectComposerContextForProvider } from "@t3tools/shared/composerContextReferences";
2import {
3 MessageId,
4 type ModelSelection,
5 type NodeId,
6 ProviderSessionId,
7 ProviderThreadId,
8 ProviderTurnId,
9 RunAttemptId,
10 ThreadId,
11} from "@t3tools/contracts";
12import * as Clock from "effect/Clock";
13import type * as DateTime from "effect/DateTime";
14import * as Context from "effect/Context";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as Schema from "effect/Schema";
19
20import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
21import * as ProjectionStore from "./ProjectionStore.ts";
22import * as ProviderSessionManager from "./ProviderSessionManager.ts";
23
24const yieldToRuntime = Effect.yieldNow.pipe(
25 Effect.andThen(
26 Effect.promise(
27 () =>
28 new Promise<void>((resolve) => {
29 setImmediate(resolve);
30 }),
31 ),
32 ),
33);
34
35export class ProviderTurnControlError extends Schema.TaggedError<ProviderTurnControlError>()(
36 "ProviderTurnControlError",
37 {
38 threadId: ThreadId,
39 operation: Schema.Literals(["interrupt", "restart", "steer"]),
40 providerTurnId: ProviderTurnId,
41 turnCompleted: Schema.optional(Schema.Boolean),
42 steerDeclined: Schema.optional(Schema.Boolean),
43 /** The provider never confirmed the stop; Force stop is the way out. */
44 unresponsive: Schema.optional(Schema.Boolean),
45 cause: Schema.optional(Schema.Defect()),
46 },
47) {}
48
49const isProviderTurnControlError = Schema.is(ProviderTurnControlError);
50
51export interface ProviderTurnControlServiceV2Shape {
52 readonly interrupt: (input: {
53 readonly subagent?: {
54 readonly id: NodeId;
55 readonly nativeTaskId: string;
56 readonly requestedAt?: DateTime.Utc;
57 };
58 readonly threadId: ThreadId;
59 readonly providerSessionId: ProviderSessionId;
60 readonly providerThreadId: ProviderThreadId;
61 readonly providerTurnId: ProviderTurnId;
62 }) => Effect.Effect<
63 void | ProviderAdapter.ProviderAdapterV2SubagentOutcome,
64 ProviderTurnControlError
65 >;
66 readonly steer: (input: {
67 readonly threadId: ThreadId;
68 readonly providerSessionId: ProviderSessionId;
69 readonly providerThreadId: ProviderThreadId;
70 readonly providerTurnId: ProviderTurnId;
71 readonly messageId: MessageId;
72 readonly modelSelection?: ModelSelection;
73 }) => Effect.Effect<void, ProviderTurnControlError>;
74 /** Best effort: a turn that ended or moved sessions picks the selection up when it next starts. */
75 readonly applyModelSelection: (input: {
76 readonly threadId: ThreadId;
77 readonly providerSessionId: ProviderSessionId;
78 readonly providerThreadId: ProviderThreadId;
79 readonly modelSelection: ModelSelection;
80 }) => Effect.Effect<
81 void,
82 | ProjectionStore.ProjectionStoreV2Error
83 | ProviderSessionManager.ProviderSessionManagerV2Error
84 | ProviderAdapter.ProviderAdapterV2Error
85 >;
86 readonly interruptAndAwaitTerminal: (input: {
87 readonly threadId: ThreadId;
88 readonly providerSessionId: ProviderSessionId;
89 readonly replacementProviderSessionId?: ProviderSessionId;
90 readonly providerThreadId: ProviderThreadId;
91 readonly providerTurnId: ProviderTurnId;
92 readonly interruptedAttemptId: RunAttemptId;
93 }) => Effect.Effect<void, ProviderTurnControlError>;
94}
95
96export class ProviderTurnControlServiceV2 extends Context.Service<
97 ProviderTurnControlServiceV2,
98 ProviderTurnControlServiceV2Shape
99>()("t3/orchestration-v2/ProviderTurnControlService/ProviderTurnControlServiceV2") {}
100
101export const layer: Layer.Layer<
102 ProviderTurnControlServiceV2,
103 never,
104 ProjectionStore.ProjectionStoreV2 | ProviderSessionManager.ProviderSessionManagerV2
105> = Layer.effect(
106 ProviderTurnControlServiceV2,
107 Effect.gen(function* () {
108 const projections = yield* ProjectionStore.ProjectionStoreV2;
109 const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
110
111 const load = Effect.fn("ProviderTurnControlService.load")(function* (input: {
112 readonly threadId: ThreadId;
113 readonly providerSessionId: ProviderSessionId;
114 readonly replacementProviderSessionId?: ProviderSessionId;
115 readonly providerThreadId: ProviderThreadId;
116 readonly providerTurnId: ProviderTurnId;
117 readonly operation: "interrupt" | "restart" | "steer";
118 readonly messageId?: MessageId;
119 }) {
120 const context = yield* projections.getProviderControlContext(input.threadId, input);
121 const { providerThread, providerTurn } = context;
122 const targetsRecordedSession = providerThread?.providerSessionId === input.providerSessionId;
123 const targetsCommittedReplacement =
124 input.operation === "restart" &&
125 input.replacementProviderSessionId !== undefined &&
126 providerThread?.providerSessionId === input.replacementProviderSessionId;
127 if (
128 providerThread === undefined ||
129 providerTurn === undefined ||
130 (!targetsRecordedSession && !targetsCommittedReplacement) ||
131 providerTurn.providerThreadId !== providerThread.id
132 ) {
133 return yield* new ProviderTurnControlError({
134 threadId: input.threadId,
135 operation: input.operation,
136 providerTurnId: input.providerTurnId,
137 cause: "The recorded provider execution target is no longer valid.",
138 });
139 }
140 // A restart-session command commits the replacement binding before its
141 // process-bound effect runs. The old live runtime must still receive
142 // the interrupt, but only when the projection matches the exact
143 // replacement captured by that same durable effect.
144 const interruptProviderThread = targetsRecordedSession
145 ? providerThread
146 : { ...providerThread, providerSessionId: input.providerSessionId };
147 const continuingRun =
148 input.operation === "steer" &&
149 context.run?.status === "running" &&
150 context.run.activeAttemptId === providerTurn.runAttemptId;
151 if (providerTurn.status !== "running" && !continuingRun) {
152 if (input.operation === "steer") {
153 return yield* new ProviderTurnControlError({
154 threadId: input.threadId,
155 operation: "steer",
156 providerTurnId: input.providerTurnId,
157 turnCompleted: providerTurn.status === "completed",
158 cause: "The provider turn ended before the steering message was delivered.",
159 });
160 }
161 return {
162 context,
163 providerThread: interruptProviderThread,
164 providerTurn,
165 session: Option.none(),
166 };
167 }
168 const session = yield* sessions.get(input.providerSessionId);
169 if (Option.isNone(session)) {
170 // Interrupt/restart against a already-released session must not fail
171 // the durable effect (and retry 5x). The turn may still look running
172 // in projection until recovery/finalization; there is no live adapter
173 // to interrupt.
174 if (input.operation === "interrupt" || input.operation === "restart") {
175 yield* Effect.logWarning(
176 "Provider interrupt/restart found no live session; treating as already stopped",
177 {
178 threadId: input.threadId,
179 operation: input.operation,
180 providerSessionId: input.providerSessionId,
181 providerTurnId: input.providerTurnId,
182 providerTurnStatus: providerTurn.status,
183 },
184 );
185 return {
186 context,
187 providerThread: interruptProviderThread,
188 providerTurn,
189 session: Option.none(),
190 };
191 }
192 return yield* new ProviderTurnControlError({
193 threadId: input.threadId,
194 operation: input.operation,
195 providerTurnId: input.providerTurnId,
196 cause: `Provider session ${input.providerSessionId} is not active.`,
197 });
198 }
199 return { context, providerThread: interruptProviderThread, providerTurn, session };
200 });
201
202 return ProviderTurnControlServiceV2.of({
203 applyModelSelection: Effect.fn("ProviderTurnControlService.applyModelSelection")(
204 function* (input) {
205 const { providerThread, providerTurn } = yield* projections.getRunningTurnContext(
206 input.threadId,
207 );
208 if (
209 providerThread?.id !== input.providerThreadId ||
210 providerThread.providerSessionId !== input.providerSessionId ||
211 providerTurn === undefined
212 ) {
213 return;
214 }
215 const session = yield* sessions.get(input.providerSessionId);
216 if (Option.isNone(session) || session.value.applyModelSelection === undefined) return;
217 yield* session.value.applyModelSelection({
218 providerThread,
219 modelSelection: input.modelSelection,
220 });
221 },
222 ),
223 interrupt: Effect.fn("ProviderTurnControlService.interrupt")(
224 function* (input) {
225 const loaded = yield* load({ ...input, operation: "interrupt" });
226 const session = Option.isSome(loaded.session)
227 ? loaded.session
228 : yield* sessions.get(input.providerSessionId);
229 if (Option.isNone(session)) {
230 if (input.subagent !== undefined) {
231 return yield* new ProviderTurnControlError({
232 threadId: input.threadId,
233 operation: "interrupt",
234 providerTurnId: input.providerTurnId,
235 cause: "No live provider session can confirm this subagent's outcome.",
236 });
237 }
238 return;
239 }
240 if (input.subagent !== undefined) {
241 if (session.value.stopSubagent === undefined) {
242 return yield* new ProviderTurnControlError({
243 threadId: input.threadId,
244 operation: "interrupt",
245 providerTurnId: input.providerTurnId,
246 cause: "The provider does not support stopping a native subagent.",
247 });
248 }
249 return yield* session.value.stopSubagent({
250 providerThread: loaded.providerThread,
251 nativeTaskId: input.subagent.nativeTaskId,
252 ...(input.subagent.requestedAt === undefined
253 ? {}
254 : { requestedAt: input.subagent.requestedAt }),
255 });
256 }
257 // Settled turns still reach adapters, which alone know whether background work remains.
258 yield* session.value
259 .interruptTurn({
260 providerThread: loaded.providerThread,
261 providerTurnId: loaded.providerTurn.id,
262 requestRuntimeRestart: true,
263 })
264 .pipe(
265 Effect.catchTags({
266 ProviderAdapterInterruptTimeoutError: (cause) =>
267 Effect.fail(
268 new ProviderTurnControlError({
269 threadId: input.threadId,
270 operation: "interrupt",
271 providerTurnId: input.providerTurnId,
272 unresponsive: true,
273 cause,
274 }),
275 ),
276 }),
277 );
278 // Use real time; ingestion fibers do not advance TestClock and would leave Stop blocked.
279 yield* Effect.gen(function* () {
280 const deadline = (yield* Clock.currentTimeMillis) + 2_000;
281 while (
282 loaded.providerTurn.status === "running" &&
283 (yield* Clock.currentTimeMillis) < deadline
284 ) {
285 const current = yield* projections.getProviderControlContext(input.threadId, input);
286 if (
287 current.providerTurn?.status !== "running" &&
288 current.attempt?.status !== "running"
289 ) {
290 return;
291 }
292 yield* Effect.sleep("10 millis");
293 }
294 }).pipe(Effect.provideService(Clock.Clock, Clock.Clock.defaultValue()));
295 },
296 (effect, input) =>
297 effect.pipe(
298 Effect.mapError((cause) =>
299 isProviderTurnControlError(cause)
300 ? cause
301 : new ProviderTurnControlError({
302 threadId: input.threadId,
303 operation: "interrupt",
304 providerTurnId: input.providerTurnId,
305 cause,
306 }),
307 ),
308 ),
309 ),
310 interruptAndAwaitTerminal: Effect.fn("ProviderTurnControlService.interruptAndAwaitTerminal")(
311 function* (input) {
312 const loaded = yield* load({ ...input, operation: "restart" });
313 if (Option.isNone(loaded.session)) {
314 // Without a live adapter, no terminal event can arrive; proceed to detach/start without polling.
315 if (loaded.providerTurn.status === "running") {
316 yield* Effect.logWarning(
317 "Provider restart interrupt skipped; no live session for a still-projected running turn",
318 {
319 threadId: input.threadId,
320 providerSessionId: input.providerSessionId,
321 providerTurnId: input.providerTurnId,
322 },
323 );
324 }
325 return;
326 }
327
328 yield* loaded.session.value.interruptTurn({
329 providerThread: loaded.providerThread,
330 providerTurnId: loaded.providerTurn.id,
331 });
332
333 for (let remaining = 1_000; remaining > 0; remaining -= 1) {
334 const { providerTurn, attempt } = yield* projections.getProviderControlContext(
335 input.threadId,
336 {
337 providerThreadId: input.providerThreadId,
338 providerTurnId: input.providerTurnId,
339 attemptId: input.interruptedAttemptId,
340 },
341 );
342 if (
343 providerTurn !== undefined &&
344 providerTurn.status !== "running" &&
345 attempt !== undefined &&
346 attempt.status !== "running"
347 ) {
348 return;
349 }
350 // Yield through Node's event loop; detached ingestion cannot advance a deterministic Effect clock.
351 yield* yieldToRuntime;
352 }
353 return yield* new ProviderTurnControlError({
354 threadId: input.threadId,
355 operation: "restart",
356 providerTurnId: input.providerTurnId,
357 cause: `Provider turn ${input.providerTurnId} did not terminalize before restart.`,
358 });
359 },
360 (effect, input) =>
361 effect.pipe(
362 Effect.mapError((cause) =>
363 isProviderTurnControlError(cause)
364 ? cause
365 : new ProviderTurnControlError({
366 threadId: input.threadId,
367 operation: "restart",
368 providerTurnId: input.providerTurnId,
369 cause,
370 }),
371 ),
372 ),
373 ),
374 steer: Effect.fn("ProviderTurnControlService.steer")(
375 function* (input) {
376 const context = yield* projections.getProviderControlContext(input.threadId, input);
377 const ownership = context.message?.delegatedCompletion;
378 if (ownership !== undefined) {
379 const projection = yield* projections.getThreadRecords(input.threadId, ["runs"], {
380 runIds: [ownership.parentRunId],
381 });
382 const cohort = projection.runs.find(
383 (run) => run.id === ownership.parentRunId,
384 )?.delegatedCompletion;
385 if (
386 cohort?.disposition !== "open" ||
387 cohort.delivery?.messageId !== input.messageId ||
388 cohort.delivery.generation !== ownership.generation ||
389 cohort.delivery.taskIds.length === 0
390 )
391 return;
392 }
393 const loaded = yield* load({ ...input, operation: "steer" });
394 if (Option.isNone(loaded.session)) return;
395 const { message, run } = loaded.context;
396 if (message === undefined || run === undefined) {
397 return yield* new ProviderTurnControlError({
398 threadId: input.threadId,
399 operation: "steer",
400 providerTurnId: input.providerTurnId,
401 cause: "The persisted steering message or target run is missing.",
402 });
403 }
404 yield* loaded.session.value
405 .steerTurn({
406 threadId: input.threadId,
407 runId: run.id,
408 providerThread: loaded.providerThread,
409 providerTurnId: loaded.providerTurn.id,
410 ...(input.modelSelection === undefined
411 ? {}
412 : { modelSelection: input.modelSelection }),
413 message: {
414 messageId: message.id,
415 text: projectComposerContextForProvider({
416 text: message.text,
417 records: message.context?.records ?? [],
418 }),
419 attachments: message.attachments,
420 createdBy: message.createdBy,
421 creationSource: message.creationSource,
422 ...(message.scheduledTaskId === undefined
423 ? {}
424 : { scheduledTaskId: message.scheduledTaskId }),
425 ...(message.senderThreadId === undefined
426 ? {}
427 : { senderThreadId: message.senderThreadId }),
428 },
429 })
430 .pipe(
431 Effect.catchTags({
432 ProviderAdapterSteerDeclinedError: (cause) =>
433 Effect.fail(
434 new ProviderTurnControlError({
435 threadId: input.threadId,
436 operation: "steer",
437 providerTurnId: input.providerTurnId,
438 steerDeclined: true,
439 cause,
440 }),
441 ),
442 }),
443 Effect.catch((cause) =>
444 Effect.gen(function* () {
445 if (isProviderTurnControlError(cause)) return yield* cause;
446 const current = yield* projections.getProviderControlContext(
447 input.threadId,
448 input,
449 );
450 const status = current.providerTurn?.status;
451 return yield* new ProviderTurnControlError({
452 threadId: input.threadId,
453 operation: "steer",
454 providerTurnId: input.providerTurnId,
455 turnCompleted: status === "completed",
456 cause,
457 });
458 }),
459 ),
460 );
461 },
462 (effect, input) =>
463 effect.pipe(
464 Effect.mapError((cause) =>
465 isProviderTurnControlError(cause)
466 ? cause
467 : new ProviderTurnControlError({
468 threadId: input.threadId,
469 operation: "steer",
470 providerTurnId: input.providerTurnId,
471 cause,
472 }),
473 ),
474 ),
475 ),
476 });
477 }),
478);