1import {
2 type ChatAttachment,
3 CommandId,
4 type OrchestrationV2Command,
5 OrchestrationV2ProviderCapabilities,
6 type OrchestrationV2GoalCommand,
7 type OrchestrationV2ProviderGoalStatus,
8 type OrchestrationV2Run,
9 OrchestrationV2ThreadProjection,
10 ProviderInstanceId,
11 ThreadId,
12} from "@t3tools/contracts";
13import { latestExecutedRun } from "@t3tools/shared/orchestrationV2ThreadError";
14import * as Context from "effect/Context";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Schema from "effect/Schema";
18
19export const SteeringExecutionPolicyV2 = Schema.Literals(["active_steering", "interrupt_restart"]);
20export type SteeringExecutionPolicyV2 = typeof SteeringExecutionPolicyV2.Type;
21
22export const ForkExecutionPolicyV2 = Schema.Literals(["native_fork", "portable_context"]);
23export type ForkExecutionPolicyV2 = typeof ForkExecutionPolicyV2.Type;
24
25export const CommandPolicyCapability = Schema.Literals([
26 "queued_messages",
27 "active_steering",
28 "interrupt",
29 "interrupt_restart_steering",
30 "native_fork",
31 "fork_from_turn",
32 "rollback",
33 "rollback_snapshot",
34 "context_handoff",
35 "strong_terminal_status",
36]);
37export type CommandPolicyCapability = typeof CommandPolicyCapability.Type;
38
39export class CommandPolicyCapabilityUnsupportedError extends Schema.TaggedError<CommandPolicyCapabilityUnsupportedError>()(
40 "CommandPolicyCapabilityUnsupportedError",
41 {
42 commandId: CommandId,
43 threadId: ThreadId,
44 providerInstanceId: ProviderInstanceId,
45 capability: CommandPolicyCapability,
46 detail: Schema.String,
47 },
48) {
49 override get message(): string {
50 return `${this.providerInstanceId} cannot satisfy ${this.capability} for command ${this.commandId}: ${this.detail}`;
51 }
52}
53
54export const CommandPolicyV2Error = Schema.Union([CommandPolicyCapabilityUnsupportedError]);
55export type CommandPolicyV2Error = typeof CommandPolicyV2Error.Type;
56
57type MessageDispatchMode = Extract<
58 OrchestrationV2Command,
59 { readonly type: "message.dispatch" }
60>["dispatchMode"];
61
62export function isNativeMaintenanceCommand(message: {
63 readonly text: string;
64 readonly attachments: ReadonlyArray<ChatAttachment>;
65 readonly context?: import("@t3tools/contracts").OrchestrationMessageContext | undefined;
66}): boolean {
67 return (
68 message.attachments.length === 0 &&
69 ["/compact", "/logout"].includes(message.text.trim().toLowerCase())
70 );
71}
72
73/**
74 * A goal command sent into active work stops that work when it starts or
75 * replaces goal work, or ends a goal that is driving the run.
76 */
77export function goalCommandStopsActiveRun(
78 command: OrchestrationV2GoalCommand,
79 goalStatus: OrchestrationV2ProviderGoalStatus | null,
80): boolean {
81 switch (command.type) {
82 case "set":
83 case "resume":
84 return true;
85 case "clear":
86 case "pause":
87 return goalStatus === "active";
88 case "show":
89 return false;
90 }
91}
92
93export function resolveMessageDispatchIntent(
94 projection: OrchestrationV2ThreadProjection,
95 requestedMode: MessageDispatchMode,
96 deliveryIntent?: "auto" | "steer" | "restart",
97): MessageDispatchMode {
98 const mode = resolveDeliveryIntent(projection, requestedMode, deliveryIntent);
99 if (mode.type !== "steer_active") return mode;
100 // Compaction and sign-out cannot accept steers; queue follow-ups.
101 const target = projection.runs.find((run) => run.id === mode.targetRunId);
102 const targetMessage = projection.messages.find((message) => message.id === target?.userMessageId);
103 if (targetMessage !== undefined && isNativeMaintenanceCommand(targetMessage)) {
104 return { type: "queue_after_active" };
105 }
106 // A provider that can neither steer nor restart its turn takes the message after the turn.
107 const turns =
108 target === undefined || !isActiveRun(target)
109 ? undefined
110 : providerTurnCapabilities(projection, target);
111 return turns !== undefined &&
112 !turns.supportsActiveSteering &&
113 !(turns.supportsInterrupt && turns.supportsSteeringByInterruptRestart) &&
114 turns.supportsQueuedMessages
115 ? { type: "queue_after_active" }
116 : mode;
117}
118
119/** A run that still owns the thread's provider turn. */
120export function isActiveRun(run: OrchestrationV2Run): boolean {
121 return (
122 run.status === "preparing" ||
123 run.status === "starting" ||
124 run.status === "running" ||
125 run.status === "waiting"
126 );
127}
128
129/** The run a follow-up to a finished child binds to: the latest active one, else the last that ran. */
130export function followUpOwnerRun(
131 runs: ReadonlyArray<OrchestrationV2Run>,
132): OrchestrationV2Run | null {
133 return (
134 runs.filter(isActiveRun).toSorted((a, b) => b.ordinal - a.ordinal)[0] ?? latestExecutedRun(runs)
135 );
136}
137
138function providerTurnCapabilities(
139 projection: OrchestrationV2ThreadProjection,
140 run: OrchestrationV2Run,
141) {
142 const providerThread = projection.providerThreads.find(
143 (candidate) => candidate.id === run.providerThreadId,
144 );
145 const providerSession =
146 providerThread?.providerSessionId == null
147 ? undefined
148 : projection.providerSessions.find(
149 (candidate) => candidate.id === providerThread.providerSessionId,
150 );
151 return providerSession?.capabilities.turns;
152}
153
154function resolveDeliveryIntent(
155 projection: OrchestrationV2ThreadProjection,
156 requestedMode: MessageDispatchMode,
157 deliveryIntent?: "auto" | "steer" | "restart",
158): MessageDispatchMode {
159 if (deliveryIntent === undefined) return requestedMode;
160
161 const activeRun = projection.runs.findLast(isActiveRun);
162 if (activeRun === undefined) return { type: "start_immediately" };
163 if (deliveryIntent === "steer") {
164 return { type: "steer_active", targetRunId: activeRun.id };
165 }
166 if (deliveryIntent === "restart") {
167 return { type: "restart_active", targetRunId: activeRun.id };
168 }
169 // A run can be active before its provider turn starts or after it finishes.
170 // Queue automatic input in those gaps so steering cannot reject the message.
171 if (
172 activeRun.status === "preparing" ||
173 activeRun.status === "starting" ||
174 !projection.providerTurns.some(
175 (turn) => turn.runAttemptId === activeRun.activeAttemptId && turn.status === "running",
176 )
177 ) {
178 return { type: "queue_after_active" };
179 }
180
181 const capabilities = providerTurnCapabilities(projection, activeRun);
182 if (capabilities?.supportsActiveSteering === true) {
183 return { type: "steer_active", targetRunId: activeRun.id };
184 }
185 if (capabilities?.supportsQueuedMessages === true) {
186 return { type: "queue_after_active" };
187 }
188 if (capabilities?.supportsSteeringByInterruptRestart === true) {
189 return { type: "restart_active", targetRunId: activeRun.id };
190 }
191 return { type: "queue_after_active" };
192}
193
194interface CapabilityCheckInput {
195 readonly commandId: CommandId;
196 readonly threadId: ThreadId;
197 readonly providerInstanceId: ProviderInstanceId;
198 readonly capabilities: OrchestrationV2ProviderCapabilities;
199}
200
201export interface CommandPolicyV2Shape {
202 readonly ensureQueuedMessages: (
203 input: CapabilityCheckInput,
204 ) => Effect.Effect<void, CommandPolicyV2Error>;
205 readonly decideSteeringExecution: (
206 input: CapabilityCheckInput & {
207 readonly forceRestart?: boolean;
208 },
209 ) => Effect.Effect<SteeringExecutionPolicyV2, CommandPolicyV2Error>;
210 readonly ensureInterrupt: (
211 input: CapabilityCheckInput,
212 ) => Effect.Effect<void, CommandPolicyV2Error>;
213 readonly ensureNativeFork: (
214 input: CapabilityCheckInput & {
215 readonly fromSpecificTurn: boolean;
216 },
217 ) => Effect.Effect<void, CommandPolicyV2Error>;
218 readonly decideForkExecution: (
219 input: CapabilityCheckInput & {
220 readonly sameProvider: boolean;
221 readonly hasStrongNativeSource: boolean;
222 readonly sourceRunStatus: OrchestrationV2Run["status"];
223 readonly fromSpecificTurn: boolean;
224 /** Later source runs are still active; providers cannot fork a thread mid-turn. */
225 readonly sourceBusy: boolean;
226 },
227 ) => Effect.Effect<ForkExecutionPolicyV2, CommandPolicyV2Error>;
228 readonly ensureRollback: (
229 input: CapabilityCheckInput,
230 ) => Effect.Effect<void, CommandPolicyV2Error>;
231 readonly ensureContextHandoff: (
232 input: CapabilityCheckInput & {
233 readonly strategy: "fork_delta_context" | "delta_context" | "full_thread_summary";
234 },
235 ) => Effect.Effect<void, CommandPolicyV2Error>;
236}
237
238export class CommandPolicyV2 extends Context.Service<CommandPolicyV2, CommandPolicyV2Shape>()(
239 "t3/orchestration-v2/CommandPolicy/CommandPolicyV2",
240) {}
241
242function unsupported(
243 input: CapabilityCheckInput,
244 capability: CommandPolicyCapability,
245 detail: string,
246) {
247 return new CommandPolicyCapabilityUnsupportedError({
248 commandId: input.commandId,
249 threadId: input.threadId,
250 providerInstanceId: input.providerInstanceId,
251 capability,
252 detail,
253 });
254}
255
256const ensureQueuedMessages: CommandPolicyV2Shape["ensureQueuedMessages"] = (input) =>
257 input.capabilities.turns.supportsQueuedMessages
258 ? Effect.void
259 : Effect.fail(
260 unsupported(
261 input,
262 "queued_messages",
263 "providerInstanceId does not support app-owned queued turns",
264 ),
265 );
266
267const decideSteeringExecution: CommandPolicyV2Shape["decideSteeringExecution"] = (input) => {
268 if (!input.forceRestart && input.capabilities.turns.supportsActiveSteering) {
269 return Effect.succeed("active_steering");
270 }
271 if (
272 input.capabilities.turns.supportsInterrupt &&
273 input.capabilities.turns.supportsSteeringByInterruptRestart
274 ) {
275 return Effect.succeed("interrupt_restart");
276 }
277 return Effect.fail(
278 unsupported(
279 input,
280 input.forceRestart || input.capabilities.turns.supportsInterrupt
281 ? "interrupt_restart_steering"
282 : "active_steering",
283 input.forceRestart
284 ? "providerInstanceId cannot satisfy a required interrupt-and-restart"
285 : "providerInstanceId cannot steer active turns directly or by interrupt-and-restart",
286 ),
287 );
288};
289
290const ensureInterrupt: CommandPolicyV2Shape["ensureInterrupt"] = (input) =>
291 input.capabilities.turns.supportsInterrupt
292 ? Effect.void
293 : Effect.fail(
294 unsupported(input, "interrupt", "providerInstanceId does not support turn interrupts"),
295 );
296
297const ensureNativeFork: CommandPolicyV2Shape["ensureNativeFork"] = (input) => {
298 if (!input.capabilities.threads.canForkThread) {
299 return Effect.fail(
300 unsupported(input, "native_fork", "providerInstanceId does not support native thread forks"),
301 );
302 }
303 if (input.fromSpecificTurn && !input.capabilities.threads.canForkFromTurn) {
304 return Effect.fail(
305 unsupported(
306 input,
307 "fork_from_turn",
308 "providerInstanceId cannot fork from a specific completed turn",
309 ),
310 );
311 }
312 if (input.capabilities.identity.nativeThreadIds !== "strong") {
313 return Effect.fail(
314 unsupported(
315 input,
316 "native_fork",
317 "providerInstanceId does not expose strong native thread ids",
318 ),
319 );
320 }
321 return Effect.void;
322};
323
324const ensureRollback: CommandPolicyV2Shape["ensureRollback"] = (input) => {
325 if (
326 !input.capabilities.threads.canRollbackThread ||
327 !input.capabilities.checkpointing.providerCanRollbackConversation
328 ) {
329 return Effect.fail(
330 unsupported(input, "rollback", "providerInstanceId conversation rollback is unavailable"),
331 );
332 }
333 if (!input.capabilities.checkpointing.providerRollbackReturnsSnapshot) {
334 return Effect.fail(
335 unsupported(
336 input,
337 "rollback_snapshot",
338 "rollback must return a providerInstanceId thread snapshot",
339 ),
340 );
341 }
342 return Effect.void;
343};
344
345const ensureContextHandoff: CommandPolicyV2Shape["ensureContextHandoff"] = (input) => {
346 if (!input.capabilities.context.canConsumeHandoffSummaries) {
347 return Effect.fail(
348 unsupported(input, "context_handoff", "providerInstanceId cannot consume handoff summaries"),
349 );
350 }
351 if (!input.capabilities.context.acceptsSyntheticUserContext) {
352 return Effect.fail(
353 unsupported(
354 input,
355 "context_handoff",
356 "providerInstanceId cannot receive synthetic user context",
357 ),
358 );
359 }
360 if (
361 (input.strategy === "fork_delta_context" || input.strategy === "delta_context") &&
362 !input.capabilities.context.supportsDeltaHandoff
363 ) {
364 return Effect.fail(
365 unsupported(input, "context_handoff", "providerInstanceId does not support delta handoff"),
366 );
367 }
368 if (
369 input.strategy === "full_thread_summary" &&
370 !input.capabilities.context.supportsFullThreadHandoff
371 ) {
372 return Effect.fail(
373 unsupported(
374 input,
375 "context_handoff",
376 "providerInstanceId does not support full-thread handoff",
377 ),
378 );
379 }
380 return Effect.void;
381};
382
383const decideForkExecution: CommandPolicyV2Shape["decideForkExecution"] = (input) => {
384 // Unsuccessful runs may lack native cursors; fork bounded transcripts to exclude later turns.
385 const canForkNatively =
386 (input.sourceRunStatus === "completed" || input.sourceRunStatus === "waiting") &&
387 input.sameProvider &&
388 input.hasStrongNativeSource &&
389 !input.sourceBusy &&
390 input.capabilities.threads.canForkThread &&
391 (!input.fromSpecificTurn || input.capabilities.threads.canForkFromTurn) &&
392 input.capabilities.identity.nativeThreadIds === "strong";
393
394 if (canForkNatively) {
395 return Effect.succeed("native_fork");
396 }
397
398 return ensureContextHandoff({
399 commandId: input.commandId,
400 threadId: input.threadId,
401 providerInstanceId: input.providerInstanceId,
402 capabilities: input.capabilities,
403 strategy: "full_thread_summary",
404 }).pipe(Effect.as("portable_context"));
405};
406
407export const layer: Layer.Layer<CommandPolicyV2> = Layer.succeed(CommandPolicyV2, {
408 ensureQueuedMessages,
409 decideSteeringExecution,
410 ensureInterrupt,
411 ensureNativeFork,
412 decideForkExecution,
413 ensureRollback,
414 ensureContextHandoff,
415});