1import {
2 CommandId,
3 ModelSelection,
4 type OrchestrationV2Command,
5 OrchestrationV2ProviderCapabilities,
6 type OrchestrationV2Run,
7 OrchestrationV2ThreadProjection,
8 ProviderInstanceId,
9 ProviderTurnId,
10 RunId,
11 ThreadId,
12} from "@t3tools/contracts";
13import * as Context from "effect/Context";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as Schema from "effect/Schema";
17
18export const MessageDispatchDecisionV2 = Schema.Union([
19 Schema.Struct({
20 type: Schema.Literal("start_run"),
21 modelSelection: ModelSelection,
22 }),
23 Schema.Struct({
24 type: Schema.Literal("steer_active"),
25 targetRunId: RunId,
26 providerTurnId: ProviderTurnId,
27 }),
28 Schema.Struct({
29 type: Schema.Literal("restart_active"),
30 targetRunId: RunId,
31 interruptProviderTurnId: ProviderTurnId,
32 }),
33 Schema.Struct({
34 type: Schema.Literal("queue_after_active"),
35 activeRunId: RunId,
36 }),
37 Schema.Struct({
38 type: Schema.Literal("switch_provider"),
39 fromProviderInstanceId: ProviderInstanceId,
40 toModelSelection: ModelSelection,
41 }),
42]);
43export type MessageDispatchDecisionV2 = typeof MessageDispatchDecisionV2.Type;
44
45export const SteeringExecutionPolicyV2 = Schema.Literals(["active_steering", "interrupt_restart"]);
46export type SteeringExecutionPolicyV2 = typeof SteeringExecutionPolicyV2.Type;
47
48export const ForkExecutionPolicyV2 = Schema.Literals(["native_fork", "portable_context"]);
49export type ForkExecutionPolicyV2 = typeof ForkExecutionPolicyV2.Type;
50
51export const CommandPolicyCapability = Schema.Literals([
52 "queued_messages",
53 "active_steering",
54 "interrupt",
55 "interrupt_restart_steering",
56 "native_fork",
57 "fork_from_turn",
58 "rollback",
59 "rollback_snapshot",
60 "context_handoff",
61 "strong_terminal_status",
62]);
63export type CommandPolicyCapability = typeof CommandPolicyCapability.Type;
64
65export class CommandPolicyMessageDispatchError extends Schema.TaggedError<CommandPolicyMessageDispatchError>()(
66 "CommandPolicyMessageDispatchError",
67 {
68 commandId: CommandId,
69 threadId: ThreadId,
70 cause: Schema.optional(Schema.Defect()),
71 },
72) {
73 override get message(): string {
74 return `Failed to choose message dispatch policy for command ${this.commandId}.`;
75 }
76}
77
78export class CommandPolicyUnsupportedError extends Schema.TaggedError<CommandPolicyUnsupportedError>()(
79 "CommandPolicyUnsupportedError",
80 {
81 commandId: CommandId,
82 threadId: ThreadId,
83 requestedMode: Schema.String,
84 providerInstanceId: ProviderInstanceId,
85 },
86) {
87 override get message(): string {
88 return `${this.providerInstanceId} cannot satisfy message dispatch mode ${this.requestedMode} for command ${this.commandId}.`;
89 }
90}
91
92export class CommandPolicyCapabilityUnsupportedError extends Schema.TaggedError<CommandPolicyCapabilityUnsupportedError>()(
93 "CommandPolicyCapabilityUnsupportedError",
94 {
95 commandId: CommandId,
96 threadId: ThreadId,
97 providerInstanceId: ProviderInstanceId,
98 capability: CommandPolicyCapability,
99 detail: Schema.String,
100 },
101) {
102 override get message(): string {
103 return `${this.providerInstanceId} cannot satisfy ${this.capability} for command ${this.commandId}: ${this.detail}`;
104 }
105}
106
107export const CommandPolicyV2Error = Schema.Union([
108 CommandPolicyMessageDispatchError,
109 CommandPolicyUnsupportedError,
110 CommandPolicyCapabilityUnsupportedError,
111]);
112export type CommandPolicyV2Error = typeof CommandPolicyV2Error.Type;
113
114type MessageDispatchMode = Extract<
115 OrchestrationV2Command,
116 { readonly type: "message.dispatch" }
117>["dispatchMode"];
118
119/** Resolve client intent from the state serialized by the thread dispatch lock. */
120export function resolveMessageDispatchIntent(
121 projection: OrchestrationV2ThreadProjection,
122 requestedMode: MessageDispatchMode,
123 deliveryIntent?: "auto" | "steer" | "restart",
124): MessageDispatchMode {
125 if (deliveryIntent === undefined) return requestedMode;
126
127 const activeRun = projection.runs.findLast(
128 (run) =>
129 run.status === "preparing" ||
130 run.status === "starting" ||
131 run.status === "running" ||
132 run.status === "waiting",
133 );
134 if (activeRun === undefined) return { type: "start_immediately" };
135 if (deliveryIntent === "steer") {
136 return { type: "steer_active", targetRunId: activeRun.id };
137 }
138 if (deliveryIntent === "restart") {
139 return { type: "restart_active", targetRunId: activeRun.id };
140 }
141 if (activeRun.status === "preparing" || activeRun.status === "starting") {
142 return { type: "queue_after_active" };
143 }
144
145 const providerThread = projection.providerThreads.find(
146 (candidate) => candidate.id === activeRun.providerThreadId,
147 );
148 const providerSession =
149 providerThread?.providerSessionId == null
150 ? undefined
151 : projection.providerSessions.find(
152 (candidate) => candidate.id === providerThread.providerSessionId,
153 );
154 const capabilities = providerSession?.capabilities.turns;
155 if (capabilities?.supportsActiveSteering === true) {
156 return { type: "steer_active", targetRunId: activeRun.id };
157 }
158 if (capabilities?.supportsQueuedMessages === true) {
159 return { type: "queue_after_active" };
160 }
161 if (capabilities?.supportsSteeringByInterruptRestart === true) {
162 return { type: "restart_active", targetRunId: activeRun.id };
163 }
164 return { type: "queue_after_active" };
165}
166
167interface CapabilityCheckInput {
168 readonly commandId: CommandId;
169 readonly threadId: ThreadId;
170 readonly providerInstanceId: ProviderInstanceId;
171 readonly capabilities: OrchestrationV2ProviderCapabilities;
172}
173
174export interface CommandPolicyV2Shape {
175 readonly decideMessageDispatch: (input: {
176 readonly commandId: CommandId;
177 readonly projection: OrchestrationV2ThreadProjection;
178 readonly requestedModelSelection?: ModelSelection;
179 readonly requestedMode:
180 | { readonly type: "steer_active"; readonly targetRunId: RunId }
181 | { readonly type: "restart_active"; readonly targetRunId: RunId }
182 | { readonly type: "queue_after_active" }
183 | { readonly type: "start_immediately" };
184 readonly capabilities: OrchestrationV2ProviderCapabilities;
185 }) => Effect.Effect<MessageDispatchDecisionV2, CommandPolicyV2Error>;
186 readonly ensureQueuedMessages: (
187 input: CapabilityCheckInput,
188 ) => Effect.Effect<void, CommandPolicyV2Error>;
189 readonly decideSteeringExecution: (
190 input: CapabilityCheckInput & {
191 readonly forceRestart?: boolean;
192 },
193 ) => Effect.Effect<SteeringExecutionPolicyV2, CommandPolicyV2Error>;
194 readonly ensureInterrupt: (
195 input: CapabilityCheckInput,
196 ) => Effect.Effect<void, CommandPolicyV2Error>;
197 readonly ensureNativeFork: (
198 input: CapabilityCheckInput & {
199 readonly fromSpecificTurn: boolean;
200 },
201 ) => Effect.Effect<void, CommandPolicyV2Error>;
202 readonly decideForkExecution: (
203 input: CapabilityCheckInput & {
204 readonly sameProvider: boolean;
205 readonly hasStrongNativeSource: boolean;
206 readonly sourceRunStatus: OrchestrationV2Run["status"];
207 readonly fromSpecificTurn: boolean;
208 },
209 ) => Effect.Effect<ForkExecutionPolicyV2, CommandPolicyV2Error>;
210 readonly ensureRollback: (
211 input: CapabilityCheckInput,
212 ) => Effect.Effect<void, CommandPolicyV2Error>;
213 readonly ensureContextHandoff: (
214 input: CapabilityCheckInput & {
215 readonly strategy: "fork_delta_context" | "delta_context" | "full_thread_summary";
216 },
217 ) => Effect.Effect<void, CommandPolicyV2Error>;
218}
219
220export class CommandPolicyV2 extends Context.Service<CommandPolicyV2, CommandPolicyV2Shape>()(
221 "t3/orchestration-v2/CommandPolicy/CommandPolicyV2",
222) {}
223
224function unsupported(
225 input: CapabilityCheckInput,
226 capability: CommandPolicyCapability,
227 detail: string,
228) {
229 return new CommandPolicyCapabilityUnsupportedError({
230 commandId: input.commandId,
231 threadId: input.threadId,
232 providerInstanceId: input.providerInstanceId,
233 capability,
234 detail,
235 });
236}
237
238const ensureQueuedMessages: CommandPolicyV2Shape["ensureQueuedMessages"] = (input) =>
239 input.capabilities.turns.supportsQueuedMessages
240 ? Effect.void
241 : Effect.fail(
242 unsupported(
243 input,
244 "queued_messages",
245 "providerInstanceId does not support app-owned queued turns",
246 ),
247 );
248
249const decideSteeringExecution: CommandPolicyV2Shape["decideSteeringExecution"] = (input) => {
250 if (!input.forceRestart && input.capabilities.turns.supportsActiveSteering) {
251 return Effect.succeed("active_steering");
252 }
253 if (
254 input.capabilities.turns.supportsInterrupt &&
255 input.capabilities.turns.supportsSteeringByInterruptRestart
256 ) {
257 return Effect.succeed("interrupt_restart");
258 }
259 return Effect.fail(
260 unsupported(
261 input,
262 input.forceRestart || input.capabilities.turns.supportsInterrupt
263 ? "interrupt_restart_steering"
264 : "active_steering",
265 input.forceRestart
266 ? "providerInstanceId cannot satisfy a required interrupt-and-restart"
267 : "providerInstanceId cannot steer active turns directly or by interrupt-and-restart",
268 ),
269 );
270};
271
272const ensureInterrupt: CommandPolicyV2Shape["ensureInterrupt"] = (input) =>
273 input.capabilities.turns.supportsInterrupt
274 ? Effect.void
275 : Effect.fail(
276 unsupported(input, "interrupt", "providerInstanceId does not support turn interrupts"),
277 );
278
279const ensureNativeFork: CommandPolicyV2Shape["ensureNativeFork"] = (input) => {
280 if (!input.capabilities.threads.canForkThread) {
281 return Effect.fail(
282 unsupported(input, "native_fork", "providerInstanceId does not support native thread forks"),
283 );
284 }
285 if (input.fromSpecificTurn && !input.capabilities.threads.canForkFromTurn) {
286 return Effect.fail(
287 unsupported(
288 input,
289 "fork_from_turn",
290 "providerInstanceId cannot fork from a specific completed turn",
291 ),
292 );
293 }
294 if (input.capabilities.identity.nativeThreadIds !== "strong") {
295 return Effect.fail(
296 unsupported(
297 input,
298 "native_fork",
299 "providerInstanceId does not expose strong native thread ids",
300 ),
301 );
302 }
303 return Effect.void;
304};
305
306const ensureRollback: CommandPolicyV2Shape["ensureRollback"] = (input) => {
307 if (
308 !input.capabilities.threads.canRollbackThread ||
309 !input.capabilities.checkpointing.providerCanRollbackConversation
310 ) {
311 return Effect.fail(
312 unsupported(input, "rollback", "providerInstanceId conversation rollback is unavailable"),
313 );
314 }
315 if (!input.capabilities.checkpointing.providerRollbackReturnsSnapshot) {
316 return Effect.fail(
317 unsupported(
318 input,
319 "rollback_snapshot",
320 "rollback must return a providerInstanceId thread snapshot",
321 ),
322 );
323 }
324 return Effect.void;
325};
326
327const ensureContextHandoff: CommandPolicyV2Shape["ensureContextHandoff"] = (input) => {
328 if (!input.capabilities.context.canConsumeHandoffSummaries) {
329 return Effect.fail(
330 unsupported(input, "context_handoff", "providerInstanceId cannot consume handoff summaries"),
331 );
332 }
333 if (!input.capabilities.context.acceptsSyntheticUserContext) {
334 return Effect.fail(
335 unsupported(
336 input,
337 "context_handoff",
338 "providerInstanceId cannot receive synthetic user context",
339 ),
340 );
341 }
342 if (
343 (input.strategy === "fork_delta_context" || input.strategy === "delta_context") &&
344 !input.capabilities.context.supportsDeltaHandoff
345 ) {
346 return Effect.fail(
347 unsupported(input, "context_handoff", "providerInstanceId does not support delta handoff"),
348 );
349 }
350 if (
351 input.strategy === "full_thread_summary" &&
352 !input.capabilities.context.supportsFullThreadHandoff
353 ) {
354 return Effect.fail(
355 unsupported(
356 input,
357 "context_handoff",
358 "providerInstanceId does not support full-thread handoff",
359 ),
360 );
361 }
362 return Effect.void;
363};
364
365const decideForkExecution: CommandPolicyV2Shape["decideForkExecution"] = (input) => {
366 // Unsuccessful runs may have no native turn or assistant cursor. Forking
367 // those at native head can include later turns, so use the bounded transcript.
368 const canForkNatively =
369 (input.sourceRunStatus === "completed" || input.sourceRunStatus === "waiting") &&
370 input.sameProvider &&
371 input.hasStrongNativeSource &&
372 input.capabilities.threads.canForkThread &&
373 (!input.fromSpecificTurn || input.capabilities.threads.canForkFromTurn) &&
374 input.capabilities.identity.nativeThreadIds === "strong";
375
376 if (canForkNatively) {
377 return Effect.succeed("native_fork");
378 }
379
380 return ensureContextHandoff({
381 commandId: input.commandId,
382 threadId: input.threadId,
383 providerInstanceId: input.providerInstanceId,
384 capabilities: input.capabilities,
385 strategy: "full_thread_summary",
386 }).pipe(Effect.as("portable_context"));
387};
388
389const decideMessageDispatch: CommandPolicyV2Shape["decideMessageDispatch"] = (input) => {
390 const activeRun = input.projection.runs.find(
391 (run) =>
392 run.status === "preparing" ||
393 run.status === "starting" ||
394 run.status === "running" ||
395 run.status === "waiting",
396 );
397 const modelSelection = input.requestedModelSelection ?? input.projection.thread.modelSelection;
398
399 switch (input.requestedMode.type) {
400 case "steer_active": {
401 if (activeRun?.id !== input.requestedMode.targetRunId) {
402 return Effect.fail(
403 new CommandPolicyUnsupportedError({
404 commandId: input.commandId,
405 threadId: input.projection.thread.id,
406 requestedMode: input.requestedMode.type,
407 providerInstanceId: modelSelection.instanceId,
408 }),
409 );
410 }
411 const providerTurnId =
412 activeRun.activeAttemptId === null
413 ? undefined
414 : input.projection.providerTurns.find(
415 (turn) =>
416 turn.runAttemptId === activeRun.activeAttemptId && turn.status === "running",
417 )?.id;
418 return providerTurnId === undefined
419 ? Effect.fail(
420 new CommandPolicyMessageDispatchError({
421 commandId: input.commandId,
422 threadId: input.projection.thread.id,
423 cause: `No running providerInstanceId turn found for active run ${activeRun.id}.`,
424 }),
425 )
426 : Effect.succeed({
427 type: "steer_active",
428 targetRunId: activeRun.id,
429 providerTurnId,
430 });
431 }
432 case "restart_active": {
433 if (activeRun?.id !== input.requestedMode.targetRunId || activeRun.activeAttemptId === null) {
434 return Effect.fail(
435 new CommandPolicyUnsupportedError({
436 commandId: input.commandId,
437 threadId: input.projection.thread.id,
438 requestedMode: input.requestedMode.type,
439 providerInstanceId: modelSelection.instanceId,
440 }),
441 );
442 }
443 const providerTurnId = input.projection.providerTurns.find(
444 (turn) => turn.runAttemptId === activeRun.activeAttemptId && turn.status === "running",
445 )?.id;
446 return providerTurnId === undefined
447 ? Effect.fail(
448 new CommandPolicyMessageDispatchError({
449 commandId: input.commandId,
450 threadId: input.projection.thread.id,
451 cause: `No running providerInstanceId turn found for active run ${activeRun.id}.`,
452 }),
453 )
454 : Effect.succeed({
455 type: "restart_active",
456 targetRunId: activeRun.id,
457 interruptProviderTurnId: providerTurnId,
458 });
459 }
460 case "queue_after_active":
461 return activeRun === undefined
462 ? Effect.succeed({ type: "start_run", modelSelection })
463 : ensureQueuedMessages({
464 commandId: input.commandId,
465 threadId: input.projection.thread.id,
466 providerInstanceId: modelSelection.instanceId,
467 capabilities: input.capabilities,
468 }).pipe(Effect.as({ type: "queue_after_active", activeRunId: activeRun.id }));
469 case "start_immediately":
470 if (activeRun === undefined) {
471 return Effect.succeed({ type: "start_run", modelSelection });
472 }
473 return ensureQueuedMessages({
474 commandId: input.commandId,
475 threadId: input.projection.thread.id,
476 providerInstanceId: modelSelection.instanceId,
477 capabilities: input.capabilities,
478 }).pipe(Effect.as({ type: "queue_after_active", activeRunId: activeRun.id }));
479 }
480};
481
482export const layer: Layer.Layer<CommandPolicyV2> = Layer.succeed(CommandPolicyV2, {
483 decideMessageDispatch,
484 ensureQueuedMessages,
485 decideSteeringExecution,
486 ensureInterrupt,
487 ensureNativeFork,
488 decideForkExecution,
489 ensureRollback,
490 ensureContextHandoff,
491});