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