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