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