1import {
2 ModelSelection,
3 OrchestrationV2ThreadProjection,
4 ProviderSessionId,
5 ThreadId,
6} from "@t3tools/contracts";
7import { modelSelectionsEqual } from "@t3tools/shared/model";
8import * as Context from "effect/Context";
9import * as DateTime from "effect/DateTime";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Option from "effect/Option";
13import * as Schema from "effect/Schema";
14
15import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
16import {
17 decideProviderSessionTransition,
18 type ProviderSessionTransition,
19} from "./ProviderSessionTransitionPolicy.ts";
20
21export interface ProviderSwitchPlanV2 {
22 readonly instanceChanged: boolean;
23 readonly releaseProviderSessionIds: ReadonlyArray<ProviderSessionId>;
24 readonly transition: ProviderSessionTransition;
25}
26
27export class ProviderSwitchPlanError extends Schema.TaggedError<ProviderSwitchPlanError>()(
28 "ProviderSwitchPlanError",
29 {
30 threadId: ThreadId,
31 targetProviderInstanceId: Schema.String,
32 cause: Schema.optional(Schema.Defect()),
33 },
34) {}
35
36export interface ProviderSwitchServiceV2Shape {
37 readonly plan: (input: {
38 readonly projection: Pick<
39 OrchestrationV2ThreadProjection,
40 "thread" | "providerSessions" | "providerThreads"
41 >;
42 readonly targetModelSelection: ModelSelection;
43 }) => Effect.Effect<ProviderSwitchPlanV2, ProviderSwitchPlanError>;
44}
45
46export class ProviderSwitchServiceV2 extends Context.Service<
47 ProviderSwitchServiceV2,
48 ProviderSwitchServiceV2Shape
49>()("t3/orchestration-v2/ProviderSwitchService/ProviderSwitchServiceV2") {}
50
51const isLiveProviderSession = (
52 session: OrchestrationV2ThreadProjection["providerSessions"][number],
53) => session.status !== "stopped" && session.status !== "error";
54
55export const layer: Layer.Layer<
56 ProviderSwitchServiceV2,
57 never,
58 ProviderAdapterRegistry.ProviderAdapterRegistryV2
59> = Layer.effect(
60 ProviderSwitchServiceV2,
61 Effect.gen(function* () {
62 const adapters = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
63 return ProviderSwitchServiceV2.of({
64 plan: Effect.fn("ProviderSwitchService.plan")(
65 function* ({ projection, targetModelSelection }) {
66 const current = projection.thread.modelSelection;
67 const instanceChanged = current.instanceId !== targetModelSelection.instanceId;
68 const getMetadata = (instanceId: typeof current.instanceId) =>
69 adapters.getMetadata !== undefined
70 ? adapters.getMetadata(instanceId)
71 : adapters.get(instanceId).pipe(
72 Effect.flatMap((adapter) =>
73 adapter.getCapabilities().pipe(
74 Effect.map((capabilities) => ({
75 driver: adapter.driver,
76 continuationKey: `${adapter.driver}:instance:${instanceId}`,
77 enabled: true,
78 capabilities,
79 })),
80 ),
81 ),
82 );
83 const currentInstance = yield* Effect.option(getMetadata(current.instanceId));
84 const targetInstance = yield* Effect.option(getMetadata(targetModelSelection.instanceId));
85 const targetAdapter = yield* Effect.option(adapters.get(targetModelSelection.instanceId));
86 const currentSessions = projection.providerSessions
87 .filter((session) => session.providerInstanceId === current.instanceId)
88 .toSorted(
89 (left, right) =>
90 DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
91 );
92 const currentSession = currentSessions.find(isLiveProviderSession);
93 const negotiatedCapabilities =
94 currentSession?.capabilities ?? currentSessions[0]?.capabilities;
95 const currentProviderThread = projection.providerThreads.find(
96 (thread) =>
97 thread.id === projection.thread.activeProviderThreadId &&
98 thread.providerInstanceId === current.instanceId &&
99 thread.nativeThreadRef !== null,
100 );
101 const selectionTransition =
102 current.instanceId === targetModelSelection.instanceId &&
103 !modelSelectionsEqual(current, targetModelSelection) &&
104 Option.isSome(targetAdapter) &&
105 Option.isSome(currentInstance) &&
106 (currentSession !== undefined || currentProviderThread !== undefined)
107 ? negotiatedCapabilities === undefined
108 ? // No session record left to switch in place; the next session starts with it.
109 ({ type: "apply_on_next_turn" } as const)
110 : yield* targetAdapter.value.planSelectionTransition({
111 current,
112 target: targetModelSelection,
113 sessionCapabilities: negotiatedCapabilities,
114 })
115 : undefined;
116 const transition =
117 Option.isNone(targetInstance) || Option.isNone(targetAdapter)
118 ? ({
119 type: "reject",
120 reason: "The target provider instance is unavailable.",
121 } as const)
122 : decideProviderSessionTransition({
123 current:
124 Option.isNone(currentInstance) ||
125 (currentSession === undefined && currentProviderThread === undefined)
126 ? null
127 : {
128 driver: currentInstance.value.driver,
129 continuationIdentity: {
130 driverKind: currentInstance.value.driver,
131 continuationKey: currentInstance.value.continuationKey,
132 },
133 modelSelection: current,
134 runtimeMode: projection.thread.runtimeMode,
135 interactionMode: projection.thread.interactionMode,
136 workspace:
137 currentSession?.cwd ??
138 projection.thread.worktreePath ??
139 "<unresolved-workspace>",
140 },
141 target: {
142 driver: targetInstance.value.driver,
143 continuationIdentity: {
144 driverKind: targetInstance.value.driver,
145 continuationKey: targetInstance.value.continuationKey,
146 },
147 modelSelection: targetModelSelection,
148 runtimeMode: projection.thread.runtimeMode,
149 interactionMode: projection.thread.interactionMode,
150 workspace:
151 projection.thread.worktreePath ??
152 currentSession?.cwd ??
153 "<unresolved-workspace>",
154 available: targetInstance.value.enabled,
155 },
156 ...(selectionTransition === undefined ? {} : { selectionTransition }),
157 });
158 if (transition.type === "reject") {
159 return yield* new ProviderSwitchPlanError({
160 threadId: projection.thread.id,
161 targetProviderInstanceId: targetModelSelection.instanceId,
162 cause: transition.reason,
163 });
164 }
165 const releaseProviderSessionIds = projection.providerSessions
166 .filter((session) => {
167 if (!isLiveProviderSession(session)) return false;
168 if (transition.type === "restart_and_resume") {
169 // Release only the replaced session; peers can serve pooled or delegated bindings.
170 return session.id === currentSession?.id;
171 }
172 if (transition.type === "create_with_handoff") {
173 return session.providerInstanceId !== targetModelSelection.instanceId;
174 }
175 return false;
176 })
177 .map((session) => session.id);
178 return {
179 instanceChanged,
180 releaseProviderSessionIds,
181 transition,
182 };
183 },
184 (effect, { projection, targetModelSelection }) =>
185 effect.pipe(
186 Effect.mapError(
187 (cause) =>
188 new ProviderSwitchPlanError({
189 threadId: projection.thread.id,
190 targetProviderInstanceId: targetModelSelection.instanceId,
191 cause,
192 }),
193 ),
194 ),
195 ),
196 });
197 }),
198);