authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-01 09:19:38-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-01 16:19:38+00:00
log556a195c9de5554a6de978422a0160d01585a906
tree4680419e6969cfe8fc0c1c2cc3c450375b73a94d
parent80b282ece7be0b8c32c8d0573bc6569d3b7ecbe1
signature Signed by PGP key B5690EEEBB952194

[codex] migrate Effect.fn in apps/server/src/provider/Layers/ProviderService.ts (#1639)

Co-authored-by: codex <codex@users.noreply.github.com>

1 files changed, 428 insertions(+), 430 deletions(-)

apps/server/src/provider/Layers/ProviderService.ts+428-430
......@@ -131,472 +131,470 @@ function readPersistedCwd(
131131 return trimmed.length > 0 ? trimmed : undefined;
132132}
133133
134const makeProviderService = (options?: ProviderServiceLiveOptions) =>
135 Effect.gen(function* () {
136 const analytics = yield* Effect.service(AnalyticsService);
137 const serverSettings = yield* ServerSettingsService;
138 const canonicalEventLogger =
139 options?.canonicalEventLogger ??
140 (options?.canonicalEventLogPath !== undefined
141 ? yield* makeEventNdjsonLogger(options.canonicalEventLogPath, {
142 stream: "canonical",
143 })
144 : undefined);
145
146 const registry = yield* ProviderAdapterRegistry;
147 const directory = yield* ProviderSessionDirectory;
148 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
149 const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
150
151 const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
152 Effect.succeed(event).pipe(
153 Effect.tap((canonicalEvent) =>
154 canonicalEventLogger ? canonicalEventLogger.write(canonicalEvent, null) : Effect.void,
155 ),
156 Effect.flatMap((canonicalEvent) => PubSub.publish(runtimeEventPubSub, canonicalEvent)),
157 Effect.asVoid,
158 );
159
160 const upsertSessionBinding = (
161 session: ProviderSession,
162 threadId: ThreadId,
163 extra?: {
164 readonly modelSelection?: unknown;
165 readonly lastRuntimeEvent?: string;
166 readonly lastRuntimeEventAt?: string;
167 },
168 ) =>
169 directory.upsert({
170 threadId,
171 provider: session.provider,
172 runtimeMode: session.runtimeMode,
173 status: toRuntimeStatus(session),
174 ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}),
175 runtimePayload: toRuntimePayloadFromSession(session, extra),
176 });
177
178 const providers = yield* registry.listProviders();
179 const adapters = yield* Effect.forEach(providers, (provider) =>
180 registry.getByProvider(provider),
181 );
182 const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
183 publishRuntimeEvent(event);
184
185 const worker = Effect.forever(
186 Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)),
134const makeProviderService = Effect.fn("makeProviderService")(function* (
135 options?: ProviderServiceLiveOptions,
136) {
137 const analytics = yield* Effect.service(AnalyticsService);
138 const serverSettings = yield* ServerSettingsService;
139 const canonicalEventLogger =
140 options?.canonicalEventLogger ??
141 (options?.canonicalEventLogPath !== undefined
142 ? yield* makeEventNdjsonLogger(options.canonicalEventLogPath, {
143 stream: "canonical",
144 })
145 : undefined);
146
147 const registry = yield* ProviderAdapterRegistry;
148 const directory = yield* ProviderSessionDirectory;
149 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
150 const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
151
152 const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
153 Effect.succeed(event).pipe(
154 Effect.tap((canonicalEvent) =>
155 canonicalEventLogger ? canonicalEventLogger.write(canonicalEvent, null) : Effect.void,
156 ),
157 Effect.flatMap((canonicalEvent) => PubSub.publish(runtimeEventPubSub, canonicalEvent)),
158 Effect.asVoid,
187159 );
188 yield* Effect.forkScoped(worker);
189
190 yield* Effect.forEach(adapters, (adapter) =>
191 Stream.runForEach(adapter.streamEvents, (event) =>
192 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid),
193 ).pipe(Effect.forkScoped),
194 ).pipe(Effect.asVoid);
195
196 const recoverSessionForThread = (input: {
197 readonly binding: ProviderRuntimeBinding;
198 readonly operation: string;
199 }) =>
200 Effect.gen(function* () {
201 const adapter = yield* registry.getByProvider(input.binding.provider);
202 const hasResumeCursor =
203 input.binding.resumeCursor !== null && input.binding.resumeCursor !== undefined;
204 const hasActiveSession = yield* adapter.hasSession(input.binding.threadId);
205 if (hasActiveSession) {
206 const activeSessions = yield* adapter.listSessions();
207 const existing = activeSessions.find(
208 (session) => session.threadId === input.binding.threadId,
209 );
210 if (existing) {
211 yield* upsertSessionBinding(existing, input.binding.threadId);
212 yield* analytics.record("provider.session.recovered", {
213 provider: existing.provider,
214 strategy: "adopt-existing",
215 hasResumeCursor: existing.resumeCursor !== undefined,
216 });
217 return { adapter, session: existing } as const;
218 }
219 }
220160
221 if (!hasResumeCursor) {
222 return yield* toValidationError(
223 input.operation,
224 `Cannot recover thread '${input.binding.threadId}' because no provider resume state is persisted.`,
225 );
226 }
227
228 const persistedCwd = readPersistedCwd(input.binding.runtimePayload);
229 const persistedModelSelection = readPersistedModelSelection(input.binding.runtimePayload);
230
231 const resumed = yield* adapter.startSession({
232 threadId: input.binding.threadId,
233 provider: input.binding.provider,
234 ...(persistedCwd ? { cwd: persistedCwd } : {}),
235 ...(persistedModelSelection ? { modelSelection: persistedModelSelection } : {}),
236 ...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}),
237 runtimeMode: input.binding.runtimeMode ?? "full-access",
238 });
239 if (resumed.provider !== adapter.provider) {
240 return yield* toValidationError(
241 input.operation,
242 `Adapter/provider mismatch while recovering thread '${input.binding.threadId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`,
243 );
244 }
245
246 yield* upsertSessionBinding(resumed, input.binding.threadId);
161 const upsertSessionBinding = (
162 session: ProviderSession,
163 threadId: ThreadId,
164 extra?: {
165 readonly modelSelection?: unknown;
166 readonly lastRuntimeEvent?: string;
167 readonly lastRuntimeEventAt?: string;
168 },
169 ) =>
170 directory.upsert({
171 threadId,
172 provider: session.provider,
173 runtimeMode: session.runtimeMode,
174 status: toRuntimeStatus(session),
175 ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}),
176 runtimePayload: toRuntimePayloadFromSession(session, extra),
177 });
178
179 const providers = yield* registry.listProviders();
180 const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider));
181 const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
182 publishRuntimeEvent(event);
183
184 const worker = Effect.forever(
185 Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)),
186 );
187 yield* Effect.forkScoped(worker);
188
189 yield* Effect.forEach(adapters, (adapter) =>
190 Stream.runForEach(adapter.streamEvents, (event) =>
191 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid),
192 ).pipe(Effect.forkScoped),
193 ).pipe(Effect.asVoid);
194
195 const recoverSessionForThread = Effect.fn("recoverSessionForThread")(function* (input: {
196 readonly binding: ProviderRuntimeBinding;
197 readonly operation: string;
198 }) {
199 const adapter = yield* registry.getByProvider(input.binding.provider);
200 const hasResumeCursor =
201 input.binding.resumeCursor !== null && input.binding.resumeCursor !== undefined;
202 const hasActiveSession = yield* adapter.hasSession(input.binding.threadId);
203 if (hasActiveSession) {
204 const activeSessions = yield* adapter.listSessions();
205 const existing = activeSessions.find(
206 (session) => session.threadId === input.binding.threadId,
207 );
208 if (existing) {
209 yield* upsertSessionBinding(existing, input.binding.threadId);
247210 yield* analytics.record("provider.session.recovered", {
248 provider: resumed.provider,
249 strategy: "resume-thread",
250 hasResumeCursor: resumed.resumeCursor !== undefined,
211 provider: existing.provider,
212 strategy: "adopt-existing",
213 hasResumeCursor: existing.resumeCursor !== undefined,
251214 });
252 return { adapter, session: resumed } as const;
253 });
215 return { adapter, session: existing } as const;
216 }
217 }
218
219 if (!hasResumeCursor) {
220 return yield* toValidationError(
221 input.operation,
222 `Cannot recover thread '${input.binding.threadId}' because no provider resume state is persisted.`,
223 );
224 }
225
226 const persistedCwd = readPersistedCwd(input.binding.runtimePayload);
227 const persistedModelSelection = readPersistedModelSelection(input.binding.runtimePayload);
228
229 const resumed = yield* adapter.startSession({
230 threadId: input.binding.threadId,
231 provider: input.binding.provider,
232 ...(persistedCwd ? { cwd: persistedCwd } : {}),
233 ...(persistedModelSelection ? { modelSelection: persistedModelSelection } : {}),
234 ...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}),
235 runtimeMode: input.binding.runtimeMode ?? "full-access",
236 });
237 if (resumed.provider !== adapter.provider) {
238 return yield* toValidationError(
239 input.operation,
240 `Adapter/provider mismatch while recovering thread '${input.binding.threadId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`,
241 );
242 }
243
244 yield* upsertSessionBinding(resumed, input.binding.threadId);
245 yield* analytics.record("provider.session.recovered", {
246 provider: resumed.provider,
247 strategy: "resume-thread",
248 hasResumeCursor: resumed.resumeCursor !== undefined,
249 });
250 return { adapter, session: resumed } as const;
251 });
254252
255 const resolveRoutableSession = (input: {
256 readonly threadId: ThreadId;
257 readonly operation: string;
258 readonly allowRecovery: boolean;
259 }) =>
260 Effect.gen(function* () {
261 const bindingOption = yield* directory.getBinding(input.threadId);
262 const binding = Option.getOrUndefined(bindingOption);
263 if (!binding) {
264 return yield* toValidationError(
265 input.operation,
266 `Cannot route thread '${input.threadId}' because no persisted provider binding exists.`,
267 );
268 }
269 const adapter = yield* registry.getByProvider(binding.provider);
253 const resolveRoutableSession = Effect.fn("resolveRoutableSession")(function* (input: {
254 readonly threadId: ThreadId;
255 readonly operation: string;
256 readonly allowRecovery: boolean;
257 }) {
258 const bindingOption = yield* directory.getBinding(input.threadId);
259 const binding = Option.getOrUndefined(bindingOption);
260 if (!binding) {
261 return yield* toValidationError(
262 input.operation,
263 `Cannot route thread '${input.threadId}' because no persisted provider binding exists.`,
264 );
265 }
266 const adapter = yield* registry.getByProvider(binding.provider);
270267
271 const hasRequestedSession = yield* adapter.hasSession(input.threadId);
272 if (hasRequestedSession) {
273 return { adapter, threadId: input.threadId, isActive: true } as const;
274 }
268 const hasRequestedSession = yield* adapter.hasSession(input.threadId);
269 if (hasRequestedSession) {
270 return { adapter, threadId: input.threadId, isActive: true } as const;
271 }
275272
276 if (!input.allowRecovery) {
277 return { adapter, threadId: input.threadId, isActive: false } as const;
278 }
273 if (!input.allowRecovery) {
274 return { adapter, threadId: input.threadId, isActive: false } as const;
275 }
279276
280 const recovered = yield* recoverSessionForThread({ binding, operation: input.operation });
281 return { adapter: recovered.adapter, threadId: input.threadId, isActive: true } as const;
282 });
277 const recovered = yield* recoverSessionForThread({ binding, operation: input.operation });
278 return { adapter: recovered.adapter, threadId: input.threadId, isActive: true } as const;
279 });
283280
284 const startSession: ProviderServiceShape["startSession"] = (threadId, rawInput) =>
285 Effect.gen(function* () {
286 const parsed = yield* decodeInputOrValidationError({
287 operation: "ProviderService.startSession",
288 schema: ProviderSessionStartInput,
289 payload: rawInput,
290 });
281 const startSession: ProviderServiceShape["startSession"] = Effect.fn("startSession")(
282 function* (threadId, rawInput) {
283 const parsed = yield* decodeInputOrValidationError({
284 operation: "ProviderService.startSession",
285 schema: ProviderSessionStartInput,
286 payload: rawInput,
287 });
291288
292 const input = {
293 ...parsed,
294 threadId,
295 provider: parsed.provider ?? "codex",
296 };
297 const settings = yield* serverSettings.getSettings.pipe(
298 Effect.mapError((error) =>
299 toValidationError(
300 "ProviderService.startSession",
301 `Failed to load provider settings: ${error.message}`,
302 error,
303 ),
289 const input = {
290 ...parsed,
291 threadId,
292 provider: parsed.provider ?? "codex",
293 };
294 const settings = yield* serverSettings.getSettings.pipe(
295 Effect.mapError((error) =>
296 toValidationError(
297 "ProviderService.startSession",
298 `Failed to load provider settings: ${error.message}`,
299 error,
304300 ),
301 ),
302 );
303 if (!settings.providers[input.provider].enabled) {
304 return yield* toValidationError(
305 "ProviderService.startSession",
306 `Provider '${input.provider}' is disabled in T3 Code settings.`,
305307 );
306 if (!settings.providers[input.provider].enabled) {
307 return yield* toValidationError(
308 "ProviderService.startSession",
309 `Provider '${input.provider}' is disabled in T3 Code settings.`,
310 );
311 }
312 const persistedBinding = Option.getOrUndefined(yield* directory.getBinding(threadId));
313 const effectiveResumeCursor =
314 input.resumeCursor ??
315 (persistedBinding?.provider === input.provider
316 ? persistedBinding.resumeCursor
317 : undefined);
318 const adapter = yield* registry.getByProvider(input.provider);
319 const session = yield* adapter.startSession({
320 ...input,
321 ...(effectiveResumeCursor !== undefined ? { resumeCursor: effectiveResumeCursor } : {}),
322 });
323
324 if (session.provider !== adapter.provider) {
325 return yield* toValidationError(
326 "ProviderService.startSession",
327 `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`,
328 );
329 }
308 }
309 const persistedBinding = Option.getOrUndefined(yield* directory.getBinding(threadId));
310 const effectiveResumeCursor =
311 input.resumeCursor ??
312 (persistedBinding?.provider === input.provider ? persistedBinding.resumeCursor : undefined);
313 const adapter = yield* registry.getByProvider(input.provider);
314 const session = yield* adapter.startSession({
315 ...input,
316 ...(effectiveResumeCursor !== undefined ? { resumeCursor: effectiveResumeCursor } : {}),
317 });
330318
331 yield* upsertSessionBinding(session, threadId, {
332 modelSelection: input.modelSelection,
333 });
334 yield* analytics.record("provider.session.started", {
335 provider: session.provider,
336 runtimeMode: input.runtimeMode,
337 hasResumeCursor: session.resumeCursor !== undefined,
338 hasCwd: typeof input.cwd === "string" && input.cwd.trim().length > 0,
339 hasModel:
340 typeof input.modelSelection?.model === "string" &&
341 input.modelSelection.model.trim().length > 0,
342 });
319 if (session.provider !== adapter.provider) {
320 return yield* toValidationError(
321 "ProviderService.startSession",
322 `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`,
323 );
324 }
343325
344 return session;
326 yield* upsertSessionBinding(session, threadId, {
327 modelSelection: input.modelSelection,
328 });
329 yield* analytics.record("provider.session.started", {
330 provider: session.provider,
331 runtimeMode: input.runtimeMode,
332 hasResumeCursor: session.resumeCursor !== undefined,
333 hasCwd: typeof input.cwd === "string" && input.cwd.trim().length > 0,
334 hasModel:
335 typeof input.modelSelection?.model === "string" &&
336 input.modelSelection.model.trim().length > 0,
345337 });
346338
347 const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) =>
348 Effect.gen(function* () {
349 const parsed = yield* decodeInputOrValidationError({
350 operation: "ProviderService.sendTurn",
351 schema: ProviderSendTurnInput,
352 payload: rawInput,
353 });
339 return session;
340 },
341 );
354342
355 const input = {
356 ...parsed,
357 attachments: parsed.attachments ?? [],
358 };
359 if (!input.input && input.attachments.length === 0) {
360 return yield* toValidationError(
361 "ProviderService.sendTurn",
362 "Either input text or at least one attachment is required",
363 );
364 }
365 const routed = yield* resolveRoutableSession({
366 threadId: input.threadId,
367 operation: "ProviderService.sendTurn",
368 allowRecovery: true,
369 });
370 const turn = yield* routed.adapter.sendTurn(input);
371 yield* directory.upsert({
372 threadId: input.threadId,
373 provider: routed.adapter.provider,
374 status: "running",
375 ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}),
376 runtimePayload: {
377 ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}),
378 activeTurnId: turn.turnId,
379 lastRuntimeEvent: "provider.sendTurn",
380 lastRuntimeEventAt: new Date().toISOString(),
381 },
382 });
383 yield* analytics.record("provider.turn.sent", {
384 provider: routed.adapter.provider,
385 model: input.modelSelection?.model,
386 interactionMode: input.interactionMode,
387 attachmentCount: input.attachments.length,
388 hasInput: typeof input.input === "string" && input.input.trim().length > 0,
389 });
390 return turn;
391 });
343 const sendTurn: ProviderServiceShape["sendTurn"] = Effect.fn("sendTurn")(function* (rawInput) {
344 const parsed = yield* decodeInputOrValidationError({
345 operation: "ProviderService.sendTurn",
346 schema: ProviderSendTurnInput,
347 payload: rawInput,
348 });
349
350 const input = {
351 ...parsed,
352 attachments: parsed.attachments ?? [],
353 };
354 if (!input.input && input.attachments.length === 0) {
355 return yield* toValidationError(
356 "ProviderService.sendTurn",
357 "Either input text or at least one attachment is required",
358 );
359 }
360 const routed = yield* resolveRoutableSession({
361 threadId: input.threadId,
362 operation: "ProviderService.sendTurn",
363 allowRecovery: true,
364 });
365 const turn = yield* routed.adapter.sendTurn(input);
366 yield* directory.upsert({
367 threadId: input.threadId,
368 provider: routed.adapter.provider,
369 status: "running",
370 ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}),
371 runtimePayload: {
372 ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}),
373 activeTurnId: turn.turnId,
374 lastRuntimeEvent: "provider.sendTurn",
375 lastRuntimeEventAt: new Date().toISOString(),
376 },
377 });
378 yield* analytics.record("provider.turn.sent", {
379 provider: routed.adapter.provider,
380 model: input.modelSelection?.model,
381 interactionMode: input.interactionMode,
382 attachmentCount: input.attachments.length,
383 hasInput: typeof input.input === "string" && input.input.trim().length > 0,
384 });
385 return turn;
386 });
392387
393 const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) =>
394 Effect.gen(function* () {
395 const input = yield* decodeInputOrValidationError({
396 operation: "ProviderService.interruptTurn",
397 schema: ProviderInterruptTurnInput,
398 payload: rawInput,
399 });
400 const routed = yield* resolveRoutableSession({
401 threadId: input.threadId,
402 operation: "ProviderService.interruptTurn",
403 allowRecovery: true,
404 });
405 yield* routed.adapter.interruptTurn(routed.threadId, input.turnId);
406 yield* analytics.record("provider.turn.interrupted", {
407 provider: routed.adapter.provider,
408 });
388 const interruptTurn: ProviderServiceShape["interruptTurn"] = Effect.fn("interruptTurn")(
389 function* (rawInput) {
390 const input = yield* decodeInputOrValidationError({
391 operation: "ProviderService.interruptTurn",
392 schema: ProviderInterruptTurnInput,
393 payload: rawInput,
409394 });
410
411 const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) =>
412 Effect.gen(function* () {
413 const input = yield* decodeInputOrValidationError({
414 operation: "ProviderService.respondToRequest",
415 schema: ProviderRespondToRequestInput,
416 payload: rawInput,
417 });
418 const routed = yield* resolveRoutableSession({
419 threadId: input.threadId,
420 operation: "ProviderService.respondToRequest",
421 allowRecovery: true,
422 });
423 yield* routed.adapter.respondToRequest(routed.threadId, input.requestId, input.decision);
424 yield* analytics.record("provider.request.responded", {
425 provider: routed.adapter.provider,
426 decision: input.decision,
427 });
395 const routed = yield* resolveRoutableSession({
396 threadId: input.threadId,
397 operation: "ProviderService.interruptTurn",
398 allowRecovery: true,
399 });
400 yield* routed.adapter.interruptTurn(routed.threadId, input.turnId);
401 yield* analytics.record("provider.turn.interrupted", {
402 provider: routed.adapter.provider,
428403 });
404 },
405 );
429406
430 const respondToUserInput: ProviderServiceShape["respondToUserInput"] = (rawInput) =>
431 Effect.gen(function* () {
432 const input = yield* decodeInputOrValidationError({
433 operation: "ProviderService.respondToUserInput",
434 schema: ProviderRespondToUserInputInput,
435 payload: rawInput,
436 });
437 const routed = yield* resolveRoutableSession({
438 threadId: input.threadId,
439 operation: "ProviderService.respondToUserInput",
440 allowRecovery: true,
441 });
442 yield* routed.adapter.respondToUserInput(routed.threadId, input.requestId, input.answers);
407 const respondToRequest: ProviderServiceShape["respondToRequest"] = Effect.fn("respondToRequest")(
408 function* (rawInput) {
409 const input = yield* decodeInputOrValidationError({
410 operation: "ProviderService.respondToRequest",
411 schema: ProviderRespondToRequestInput,
412 payload: rawInput,
413 });
414 const routed = yield* resolveRoutableSession({
415 threadId: input.threadId,
416 operation: "ProviderService.respondToRequest",
417 allowRecovery: true,
418 });
419 yield* routed.adapter.respondToRequest(routed.threadId, input.requestId, input.decision);
420 yield* analytics.record("provider.request.responded", {
421 provider: routed.adapter.provider,
422 decision: input.decision,
443423 });
424 },
425 );
444426
445 const stopSession: ProviderServiceShape["stopSession"] = (rawInput) =>
446 Effect.gen(function* () {
447 const input = yield* decodeInputOrValidationError({
448 operation: "ProviderService.stopSession",
449 schema: ProviderStopSessionInput,
450 payload: rawInput,
451 });
452 const routed = yield* resolveRoutableSession({
453 threadId: input.threadId,
454 operation: "ProviderService.stopSession",
455 allowRecovery: false,
456 });
457 if (routed.isActive) {
458 yield* routed.adapter.stopSession(routed.threadId);
459 }
460 yield* directory.remove(input.threadId);
461 yield* analytics.record("provider.session.stopped", {
462 provider: routed.adapter.provider,
463 });
427 const respondToUserInput: ProviderServiceShape["respondToUserInput"] = Effect.fn(
428 "respondToUserInput",
429 )(function* (rawInput) {
430 const input = yield* decodeInputOrValidationError({
431 operation: "ProviderService.respondToUserInput",
432 schema: ProviderRespondToUserInputInput,
433 payload: rawInput,
434 });
435 const routed = yield* resolveRoutableSession({
436 threadId: input.threadId,
437 operation: "ProviderService.respondToUserInput",
438 allowRecovery: true,
439 });
440 yield* routed.adapter.respondToUserInput(routed.threadId, input.requestId, input.answers);
441 });
442
443 const stopSession: ProviderServiceShape["stopSession"] = Effect.fn("stopSession")(
444 function* (rawInput) {
445 const input = yield* decodeInputOrValidationError({
446 operation: "ProviderService.stopSession",
447 schema: ProviderStopSessionInput,
448 payload: rawInput,
449 });
450 const routed = yield* resolveRoutableSession({
451 threadId: input.threadId,
452 operation: "ProviderService.stopSession",
453 allowRecovery: false,
464454 });
455 if (routed.isActive) {
456 yield* routed.adapter.stopSession(routed.threadId);
457 }
458 yield* directory.remove(input.threadId);
459 yield* analytics.record("provider.session.stopped", {
460 provider: routed.adapter.provider,
461 });
462 },
463 );
465464
466 const listSessions: ProviderServiceShape["listSessions"] = () =>
467 Effect.gen(function* () {
468 const sessionsByProvider = yield* Effect.forEach(adapters, (adapter) =>
469 adapter.listSessions(),
470 );
471 const activeSessions = sessionsByProvider.flatMap((sessions) => sessions);
472 const persistedBindings = yield* directory.listThreadIds().pipe(
473 Effect.flatMap((threadIds) =>
474 Effect.forEach(
475 threadIds,
476 (threadId) =>
477 directory
478 .getBinding(threadId)
479 .pipe(Effect.orElseSucceed(() => Option.none<ProviderRuntimeBinding>())),
480 { concurrency: "unbounded" },
481 ),
465 const listSessions: ProviderServiceShape["listSessions"] = Effect.fn("listSessions")(
466 function* () {
467 const sessionsByProvider = yield* Effect.forEach(adapters, (adapter) =>
468 adapter.listSessions(),
469 );
470 const activeSessions = sessionsByProvider.flatMap((sessions) => sessions);
471 const persistedBindings = yield* directory.listThreadIds().pipe(
472 Effect.flatMap((threadIds) =>
473 Effect.forEach(
474 threadIds,
475 (threadId) =>
476 directory
477 .getBinding(threadId)
478 .pipe(Effect.orElseSucceed(() => Option.none<ProviderRuntimeBinding>())),
479 { concurrency: "unbounded" },
482480 ),
483 Effect.orElseSucceed(() => [] as Array<Option.Option<ProviderRuntimeBinding>>),
484 );
485 const bindingsByThreadId = new Map<ThreadId, ProviderRuntimeBinding>();
486 for (const bindingOption of persistedBindings) {
487 const binding = Option.getOrUndefined(bindingOption);
488 if (binding) {
489 bindingsByThreadId.set(binding.threadId, binding);
490 }
481 ),
482 Effect.orElseSucceed(() => [] as Array<Option.Option<ProviderRuntimeBinding>>),
483 );
484 const bindingsByThreadId = new Map<ThreadId, ProviderRuntimeBinding>();
485 for (const bindingOption of persistedBindings) {
486 const binding = Option.getOrUndefined(bindingOption);
487 if (binding) {
488 bindingsByThreadId.set(binding.threadId, binding);
491489 }
490 }
492491
493 return activeSessions.map((session) => {
494 const binding = bindingsByThreadId.get(session.threadId);
495 if (!binding) {
496 return session;
497 }
498
499 const overrides: {
500 resumeCursor?: ProviderSession["resumeCursor"];
501 runtimeMode?: ProviderSession["runtimeMode"];
502 } = {};
503 if (session.resumeCursor === undefined && binding.resumeCursor !== undefined) {
504 overrides.resumeCursor = binding.resumeCursor;
505 }
506 if (binding.runtimeMode !== undefined) {
507 overrides.runtimeMode = binding.runtimeMode;
508 }
509 return Object.assign({}, session, overrides);
510 });
511 });
512
513 const getCapabilities: ProviderServiceShape["getCapabilities"] = (provider) =>
514 registry.getByProvider(provider).pipe(Effect.map((adapter) => adapter.capabilities));
492 return activeSessions.map((session) => {
493 const binding = bindingsByThreadId.get(session.threadId);
494 if (!binding) {
495 return session;
496 }
515497
516 const rollbackConversation: ProviderServiceShape["rollbackConversation"] = (rawInput) =>
517 Effect.gen(function* () {
518 const input = yield* decodeInputOrValidationError({
519 operation: "ProviderService.rollbackConversation",
520 schema: ProviderRollbackConversationInput,
521 payload: rawInput,
522 });
523 if (input.numTurns === 0) {
524 return;
498 const overrides: {
499 resumeCursor?: ProviderSession["resumeCursor"];
500 runtimeMode?: ProviderSession["runtimeMode"];
501 } = {};
502 if (session.resumeCursor === undefined && binding.resumeCursor !== undefined) {
503 overrides.resumeCursor = binding.resumeCursor;
525504 }
526 const routed = yield* resolveRoutableSession({
527 threadId: input.threadId,
528 operation: "ProviderService.rollbackConversation",
529 allowRecovery: true,
530 });
531 yield* routed.adapter.rollbackThread(routed.threadId, input.numTurns);
532 yield* analytics.record("provider.conversation.rolled_back", {
533 provider: routed.adapter.provider,
534 turns: input.numTurns,
535 });
505 if (binding.runtimeMode !== undefined) {
506 overrides.runtimeMode = binding.runtimeMode;
507 }
508 return Object.assign({}, session, overrides);
536509 });
510 },
511 );
537512
538 const runStopAll = () =>
539 Effect.gen(function* () {
540 const threadIds = yield* directory.listThreadIds();
541 const activeSessions = yield* Effect.forEach(adapters, (adapter) =>
542 adapter.listSessions(),
543 ).pipe(
544 Effect.map((sessionsByAdapter) => sessionsByAdapter.flatMap((sessions) => sessions)),
545 );
546 yield* Effect.forEach(activeSessions, (session) =>
547 upsertSessionBinding(session, session.threadId, {
548 lastRuntimeEvent: "provider.stopAll",
549 lastRuntimeEventAt: new Date().toISOString(),
550 }),
551 ).pipe(Effect.asVoid);
552 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
553 yield* Effect.forEach(threadIds, (threadId) =>
554 directory.getProvider(threadId).pipe(
555 Effect.flatMap((provider) =>
556 directory.upsert({
557 threadId,
558 provider,
559 status: "stopped",
560 runtimePayload: {
561 activeTurnId: null,
562 lastRuntimeEvent: "provider.stopAll",
563 lastRuntimeEventAt: new Date().toISOString(),
564 },
565 }),
566 ),
567 ),
568 ).pipe(Effect.asVoid);
569 yield* analytics.record("provider.sessions.stopped_all", {
570 sessionCount: threadIds.length,
571 });
572 yield* analytics.flush;
573 });
513 const getCapabilities: ProviderServiceShape["getCapabilities"] = (provider) =>
514 registry.getByProvider(provider).pipe(Effect.map((adapter) => adapter.capabilities));
515
516 const rollbackConversation: ProviderServiceShape["rollbackConversation"] = Effect.fn(
517 "rollbackConversation",
518 )(function* (rawInput) {
519 const input = yield* decodeInputOrValidationError({
520 operation: "ProviderService.rollbackConversation",
521 schema: ProviderRollbackConversationInput,
522 payload: rawInput,
523 });
524 if (input.numTurns === 0) {
525 return;
526 }
527 const routed = yield* resolveRoutableSession({
528 threadId: input.threadId,
529 operation: "ProviderService.rollbackConversation",
530 allowRecovery: true,
531 });
532 yield* routed.adapter.rollbackThread(routed.threadId, input.numTurns);
533 yield* analytics.record("provider.conversation.rolled_back", {
534 provider: routed.adapter.provider,
535 turns: input.numTurns,
536 });
537 });
574538
575 yield* Effect.addFinalizer(() =>
576 Effect.catch(runStopAll(), (cause) =>
577 Effect.logWarning("failed to stop provider service", { cause }),
539 const runStopAll = Effect.fn("runStopAll")(function* () {
540 const threadIds = yield* directory.listThreadIds();
541 const activeSessions = yield* Effect.forEach(adapters, (adapter) =>
542 adapter.listSessions(),
543 ).pipe(Effect.map((sessionsByAdapter) => sessionsByAdapter.flatMap((sessions) => sessions)));
544 yield* Effect.forEach(activeSessions, (session) =>
545 upsertSessionBinding(session, session.threadId, {
546 lastRuntimeEvent: "provider.stopAll",
547 lastRuntimeEventAt: new Date().toISOString(),
548 }),
549 ).pipe(Effect.asVoid);
550 yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
551 yield* Effect.forEach(threadIds, (threadId) =>
552 directory.getProvider(threadId).pipe(
553 Effect.flatMap((provider) =>
554 directory.upsert({
555 threadId,
556 provider,
557 status: "stopped",
558 runtimePayload: {
559 activeTurnId: null,
560 lastRuntimeEvent: "provider.stopAll",
561 lastRuntimeEventAt: new Date().toISOString(),
562 },
563 }),
564 ),
578565 ),
579 );
580
581 return {
582 startSession,
583 sendTurn,
584 interruptTurn,
585 respondToRequest,
586 respondToUserInput,
587 stopSession,
588 listSessions,
589 getCapabilities,
590 rollbackConversation,
591 // Each access creates a fresh PubSub subscription so that multiple
592 // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each
593 // independently receive all runtime events.
594 get streamEvents(): ProviderServiceShape["streamEvents"] {
595 return Stream.fromPubSub(runtimeEventPubSub);
596 },
597 } satisfies ProviderServiceShape;
566 ).pipe(Effect.asVoid);
567 yield* analytics.record("provider.sessions.stopped_all", {
568 sessionCount: threadIds.length,
569 });
570 yield* analytics.flush;
598571 });
599572
573 yield* Effect.addFinalizer(() =>
574 Effect.catch(runStopAll(), (cause) =>
575 Effect.logWarning("failed to stop provider service", { cause }),
576 ),
577 );
578
579 return {
580 startSession,
581 sendTurn,
582 interruptTurn,
583 respondToRequest,
584 respondToUserInput,
585 stopSession,
586 listSessions,
587 getCapabilities,
588 rollbackConversation,
589 // Each access creates a fresh PubSub subscription so that multiple
590 // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each
591 // independently receive all runtime events.
592 get streamEvents(): ProviderServiceShape["streamEvents"] {
593 return Stream.fromPubSub(runtimeEventPubSub);
594 },
595 } satisfies ProviderServiceShape;
596});
597
600598export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService());
601599
602600export function makeProviderServiceLive(options?: ProviderServiceLiveOptions) {