/** * ProviderServiceLive - Cross-provider orchestration layer. * * Routes validated transport/API calls to provider adapters through * `ProviderAdapterRegistry` and `ProviderSessionDirectory`, and exposes a * unified provider event stream for subscribers. * * It does not implement provider protocol details (adapter concern). * * @module ProviderServiceLive */import { NonNegativeInt, ThreadId, ProviderInterruptTurnInput, ProviderRespondToRequestInput, ProviderRespondToUserInputInput, ProviderSendTurnInput, ProviderSessionStartInput, ProviderStopSessionInput, type ProviderRuntimeEvent, type ProviderSession,} from "@t3tools/contracts";import { Effect, Layer, Option, PubSub, Queue, Schema, SchemaIssue, Stream } from "effect"; import { ProviderValidationError } from "../Errors.ts";import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";import { ProviderSessionDirectory, type ProviderRuntimeBinding,} from "../Services/ProviderSessionDirectory.ts";import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; export interface ProviderServiceLiveOptions { readonly canonicalEventLogPath?: string; readonly canonicalEventLogger?: EventNdjsonLogger;} const ProviderRollbackConversationInput = Schema.Struct({ threadId: ThreadId, numTurns: NonNegativeInt,}); function toValidationError( operation: string, issue: string, cause?: unknown,): ProviderValidationError { return new ProviderValidationError({ operation, issue, ...(cause !== undefined ? { cause } : {}), });} const decodeInputOrValidationError = <S extends Schema.Top>(input: { readonly operation: string; readonly schema: S; readonly payload: unknown;}) => Schema.decodeUnknownEffect(input.schema)(input.payload).pipe( Effect.mapError( (schemaError) => new ProviderValidationError({ operation: input.operation, issue: SchemaIssue.makeFormatterDefault()(schemaError.issue), cause: schemaError, }), ), ); function toRuntimeStatus(session: ProviderSession): "starting" | "running" | "stopped" | "error" { switch (session.status) { case "connecting": return "starting"; case "error": return "error"; case "closed": return "stopped"; case "ready": case "running": default: return "running"; }} function toRuntimePayloadFromSession(session: ProviderSession): Record<string, unknown> { return { cwd: session.cwd ?? null, model: session.model ?? null, activeTurnId: session.activeTurnId ?? null, lastError: session.lastError ?? null, };} function readPersistedCwd( runtimePayload: ProviderRuntimeBinding["runtimePayload"],): string | undefined { if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) { return undefined; } const rawCwd = "cwd" in runtimePayload ? runtimePayload.cwd : undefined; if (typeof rawCwd !== "string") return undefined; const trimmed = rawCwd.trim(); return trimmed.length > 0 ? trimmed : undefined;} const makeProviderService = (options?: ProviderServiceLiveOptions) => Effect.gen(function* () { const canonicalEventLogger = options?.canonicalEventLogger ?? (options?.canonicalEventLogPath !== undefined ? yield* makeEventNdjsonLogger(options.canonicalEventLogPath, { stream: "canonical", }) : undefined); const registry = yield* ProviderAdapterRegistry; const directory = yield* ProviderSessionDirectory; const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>(); const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>(); const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> => Effect.succeed(event).pipe( Effect.tap((canonicalEvent) => canonicalEventLogger ? canonicalEventLogger.write(canonicalEvent, null) : Effect.void, ), Effect.flatMap((canonicalEvent) => PubSub.publish(runtimeEventPubSub, canonicalEvent)), Effect.asVoid, ); const upsertSessionBinding = ( session: ProviderSession, threadId: ThreadId, ) => directory.upsert({ threadId, provider: session.provider, runtimeMode: session.runtimeMode, status: toRuntimeStatus(session), ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}), runtimePayload: toRuntimePayloadFromSession(session), }); const providers = yield* registry.listProviders(); const adapters = yield* Effect.forEach(providers, (provider) => registry.getByProvider(provider), ); const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> => publishRuntimeEvent(event); const worker = Effect.forever( Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)), ); yield* Effect.forkScoped(worker); yield* Effect.forEach(adapters, (adapter) => Stream.runForEach(adapter.streamEvents, (event) => Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid), ).pipe(Effect.forkScoped), ).pipe(Effect.asVoid); const recoverSessionForThread = (input: { readonly binding: ProviderRuntimeBinding; readonly operation: string; }) => Effect.gen(function* () { const adapter = yield* registry.getByProvider(input.binding.provider); const hasResumeCursor = input.binding.resumeCursor !== null && input.binding.resumeCursor !== undefined; const hasActiveSession = yield* adapter.hasSession(input.binding.threadId); if (hasActiveSession) { const activeSessions = yield* adapter.listSessions(); const existing = activeSessions.find((session) => session.threadId === input.binding.threadId); if (existing) { yield* upsertSessionBinding(existing, input.binding.threadId); return { adapter, session: existing } as const; } } if (!hasResumeCursor) { return yield* toValidationError( input.operation, `Cannot recover thread '${input.binding.threadId}' because no provider resume state is persisted.`, ); } const persistedCwd = readPersistedCwd(input.binding.runtimePayload); const resumed = yield* adapter.startSession({ threadId: input.binding.threadId, provider: input.binding.provider, ...(persistedCwd ? { cwd: persistedCwd } : {}), ...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}), runtimeMode: input.binding.runtimeMode ?? "full-access", }); if (resumed.provider !== adapter.provider) { return yield* toValidationError( input.operation, `Adapter/provider mismatch while recovering thread '${input.binding.threadId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`, ); } yield* upsertSessionBinding(resumed, input.binding.threadId); return { adapter, session: resumed } as const; }); const resolveRoutableSession = (input: { readonly threadId: ThreadId; readonly operation: string; readonly allowRecovery: boolean; }) => Effect.gen(function* () { const bindingOption = yield* directory.getBinding(input.threadId); const binding = Option.getOrUndefined(bindingOption); if (!binding) { return yield* toValidationError( input.operation, `Cannot route thread '${input.threadId}' because no persisted provider binding exists.`, ); } const adapter = yield* registry.getByProvider(binding.provider); const hasRequestedSession = yield* adapter.hasSession(input.threadId); if (hasRequestedSession) { return { adapter, threadId: input.threadId, isActive: true } as const; } if (!input.allowRecovery) { return { adapter, threadId: input.threadId, isActive: false } as const; } const recovered = yield* recoverSessionForThread({ binding, operation: input.operation }); return { adapter: recovered.adapter, threadId: input.threadId, isActive: true } as const; }); const startSession: ProviderServiceShape["startSession"] = (threadId, rawInput) => Effect.gen(function* () { const parsed = yield* decodeInputOrValidationError({ operation: "ProviderService.startSession", schema: ProviderSessionStartInput, payload: rawInput, }); const input = { ...parsed, threadId, provider: parsed.provider ?? "codex", }; const adapter = yield* registry.getByProvider(input.provider); const session = yield* adapter.startSession(input); if (session.provider !== adapter.provider) { return yield* toValidationError( "ProviderService.startSession", `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`, ); } yield* upsertSessionBinding(session, threadId); return session; }); const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) => Effect.gen(function* () { const parsed = yield* decodeInputOrValidationError({ operation: "ProviderService.sendTurn", schema: ProviderSendTurnInput, payload: rawInput, }); const input = { ...parsed, attachments: parsed.attachments ?? [], }; if (!input.input && input.attachments.length === 0) { return yield* toValidationError( "ProviderService.sendTurn", "Either input text or at least one attachment is required", ); } const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.sendTurn", allowRecovery: true, }); const turn = yield* routed.adapter.sendTurn(input); yield* directory.upsert({ threadId: input.threadId, provider: routed.adapter.provider, status: "running", ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}), runtimePayload: { activeTurnId: turn.turnId, lastRuntimeEvent: "provider.sendTurn", lastRuntimeEventAt: new Date().toISOString(), }, }); return turn; }); const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) => Effect.gen(function* () { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.interruptTurn", schema: ProviderInterruptTurnInput, payload: rawInput, }); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.interruptTurn", allowRecovery: true, }); yield* routed.adapter.interruptTurn(routed.threadId, input.turnId); }); const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) => Effect.gen(function* () { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.respondToRequest", schema: ProviderRespondToRequestInput, payload: rawInput, }); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.respondToRequest", allowRecovery: true, }); yield* routed.adapter.respondToRequest(routed.threadId, input.requestId, input.decision); }); const respondToUserInput: ProviderServiceShape["respondToUserInput"] = (rawInput) => Effect.gen(function* () { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.respondToUserInput", schema: ProviderRespondToUserInputInput, payload: rawInput, }); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.respondToUserInput", allowRecovery: true, }); yield* routed.adapter.respondToUserInput(routed.threadId, input.requestId, input.answers); }); const stopSession: ProviderServiceShape["stopSession"] = (rawInput) => Effect.gen(function* () { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.stopSession", schema: ProviderStopSessionInput, payload: rawInput, }); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.stopSession", allowRecovery: false, }); if (routed.isActive) { yield* routed.adapter.stopSession(routed.threadId); } yield* directory.remove(input.threadId); }); const listSessions: ProviderServiceShape["listSessions"] = () => Effect.gen(function* () { const sessionsByProvider = yield* Effect.forEach(adapters, (adapter) => adapter.listSessions()); const activeSessions = sessionsByProvider.flatMap((sessions) => sessions); const persistedBindings = yield* directory .listThreadIds() .pipe( Effect.flatMap((threadIds) => Effect.forEach( threadIds, (threadId) => directory.getBinding(threadId).pipe( Effect.orElseSucceed(() => Option.none<ProviderRuntimeBinding>()), ), { concurrency: "unbounded" }, ), ), Effect.orElseSucceed(() => [] as Array<Option.Option<ProviderRuntimeBinding>>), ); const bindingsByThreadId = new Map<ThreadId, ProviderRuntimeBinding>(); for (const bindingOption of persistedBindings) { const binding = Option.getOrUndefined(bindingOption); if (binding) { bindingsByThreadId.set(binding.threadId, binding); } } return activeSessions.map((session) => { const binding = bindingsByThreadId.get(session.threadId); if (!binding) { return session; } return { ...session, ...(session.resumeCursor === undefined && binding.resumeCursor !== undefined ? { resumeCursor: binding.resumeCursor } : {}), ...(binding.runtimeMode !== undefined ? { runtimeMode: binding.runtimeMode } : {}), }; }); }); const getCapabilities: ProviderServiceShape["getCapabilities"] = (provider) => registry.getByProvider(provider).pipe(Effect.map((adapter) => adapter.capabilities)); const rollbackConversation: ProviderServiceShape["rollbackConversation"] = (rawInput) => Effect.gen(function* () { const input = yield* decodeInputOrValidationError({ operation: "ProviderService.rollbackConversation", schema: ProviderRollbackConversationInput, payload: rawInput, }); if (input.numTurns === 0) { return; } const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.rollbackConversation", allowRecovery: true, }); yield* routed.adapter.rollbackThread(routed.threadId, input.numTurns); }); const stopAll: ProviderServiceShape["stopAll"] = () => Effect.gen(function* () { const threadIds = yield* directory.listThreadIds(); yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid); yield* Effect.forEach(threadIds, (threadId) => directory.getProvider(threadId).pipe( Effect.flatMap((provider) => directory.upsert({ threadId, provider, status: "stopped", runtimePayload: { activeTurnId: null, lastRuntimeEvent: "provider.stopAll", lastRuntimeEventAt: new Date().toISOString(), }, }), ), ), ).pipe(Effect.asVoid); }); return { startSession, sendTurn, interruptTurn, respondToRequest, respondToUserInput, stopSession, listSessions, getCapabilities, rollbackConversation, stopAll, streamEvents: Stream.fromPubSub(runtimeEventPubSub), } satisfies ProviderServiceShape; }); export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService()); export function makeProviderServiceLive(options?: ProviderServiceLiveOptions) { return Layer.effect(ProviderService, makeProviderService(options));}
/**
* ProviderServiceLive - Cross-provider orchestration layer.
*
* Routes validated transport/API calls to provider adapters through
* `ProviderAdapterRegistry` and `ProviderSessionDirectory`, and exposes a
* unified provider event stream for subscribers.
*
* It does not implement provider protocol details (adapter concern).
*
* @module ProviderServiceLive
*/
import {
NonNegativeInt,
ThreadId,
ProviderInterruptTurnInput,
ProviderRespondToRequestInput,
ProviderRespondToUserInputInput,
ProviderSendTurnInput,
ProviderSessionStartInput,
ProviderStopSessionInput,
type ProviderRuntimeEvent,
type ProviderSession,
} from "@t3tools/contracts";
import { Effect, Layer, Option, PubSub, Queue, Schema, SchemaIssue, Stream } from "effect";
import { ProviderValidationError } from "../Errors.ts";
import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";
import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";
import {
ProviderSessionDirectory,
type ProviderRuntimeBinding,
} from "../Services/ProviderSessionDirectory.ts";
import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
export interface ProviderServiceLiveOptions {
readonly canonicalEventLogPath?: string;
readonly canonicalEventLogger?: EventNdjsonLogger;
}
const ProviderRollbackConversationInput = Schema.Struct({
threadId: ThreadId,
numTurns: NonNegativeInt,
});
function toValidationError(
operation: string,
issue: string,
cause?: unknown,
): ProviderValidationError {
return new ProviderValidationError({
operation,
issue,
...(cause !== undefined ? { cause } : {}),
});
}
const decodeInputOrValidationError = <S extends Schema.Top>(input: {
readonly operation: string;
readonly schema: S;
readonly payload: unknown;
}) =>
Schema.decodeUnknownEffect(input.schema)(input.payload).pipe(
Effect.mapError(
(schemaError) =>
new ProviderValidationError({
operation: input.operation,
issue: SchemaIssue.makeFormatterDefault()(schemaError.issue),
cause: schemaError,
}),
),
);
function toRuntimeStatus(session: ProviderSession): "starting" | "running" | "stopped" | "error" {
switch (session.status) {
case "connecting":
return "starting";
case "error":
return "error";
case "closed":
return "stopped";
case "ready":
case "running":
default:
return "running";
}
}
function toRuntimePayloadFromSession(session: ProviderSession): Record<string, unknown> {
return {
cwd: session.cwd ?? null,
model: session.model ?? null,
activeTurnId: session.activeTurnId ?? null,
lastError: session.lastError ?? null,
};
}
function readPersistedCwd(
runtimePayload: ProviderRuntimeBinding["runtimePayload"],
): string | undefined {
if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) {
return undefined;
}
const rawCwd = "cwd" in runtimePayload ? runtimePayload.cwd : undefined;
if (typeof rawCwd !== "string") return undefined;
const trimmed = rawCwd.trim();
return trimmed.length > 0 ? trimmed : undefined;
}
const makeProviderService = (options?: ProviderServiceLiveOptions) =>
Effect.gen(function* () {
const canonicalEventLogger =
options?.canonicalEventLogger ??
(options?.canonicalEventLogPath !== undefined
? yield* makeEventNdjsonLogger(options.canonicalEventLogPath, {
stream: "canonical",
})
: undefined);
const registry = yield* ProviderAdapterRegistry;
const directory = yield* ProviderSessionDirectory;
const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
Effect.succeed(event).pipe(
Effect.tap((canonicalEvent) =>
canonicalEventLogger
? canonicalEventLogger.write(canonicalEvent, null)
: Effect.void,
),
Effect.flatMap((canonicalEvent) => PubSub.publish(runtimeEventPubSub, canonicalEvent)),
Effect.asVoid,
);
const upsertSessionBinding = (
session: ProviderSession,
threadId: ThreadId,
) =>
directory.upsert({
threadId,
provider: session.provider,
runtimeMode: session.runtimeMode,
status: toRuntimeStatus(session),
...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}),
runtimePayload: toRuntimePayloadFromSession(session),
});
const providers = yield* registry.listProviders();
const adapters = yield* Effect.forEach(providers, (provider) =>
registry.getByProvider(provider),
);
const processRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
publishRuntimeEvent(event);
const worker = Effect.forever(
Queue.take(runtimeEventQueue).pipe(Effect.flatMap(processRuntimeEvent)),
);
yield* Effect.forkScoped(worker);
yield* Effect.forEach(adapters, (adapter) =>
Stream.runForEach(adapter.streamEvents, (event) =>
Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid),
).pipe(Effect.forkScoped),
).pipe(Effect.asVoid);
const recoverSessionForThread = (input: {
readonly binding: ProviderRuntimeBinding;
readonly operation: string;
}) =>
Effect.gen(function* () {
const adapter = yield* registry.getByProvider(input.binding.provider);
const hasResumeCursor =
input.binding.resumeCursor !== null && input.binding.resumeCursor !== undefined;
const hasActiveSession = yield* adapter.hasSession(input.binding.threadId);
if (hasActiveSession) {
const activeSessions = yield* adapter.listSessions();
const existing = activeSessions.find((session) => session.threadId === input.binding.threadId);
if (existing) {
yield* upsertSessionBinding(existing, input.binding.threadId);
return { adapter, session: existing } as const;
}
}
if (!hasResumeCursor) {
return yield* toValidationError(
input.operation,
`Cannot recover thread '${input.binding.threadId}' because no provider resume state is persisted.`,
);
}
const persistedCwd = readPersistedCwd(input.binding.runtimePayload);
const resumed = yield* adapter.startSession({
threadId: input.binding.threadId,
provider: input.binding.provider,
...(persistedCwd ? { cwd: persistedCwd } : {}),
...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}),
runtimeMode: input.binding.runtimeMode ?? "full-access",
});
if (resumed.provider !== adapter.provider) {
return yield* toValidationError(
input.operation,
`Adapter/provider mismatch while recovering thread '${input.binding.threadId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`,
);
}
yield* upsertSessionBinding(resumed, input.binding.threadId);
return { adapter, session: resumed } as const;
});
const resolveRoutableSession = (input: {
readonly threadId: ThreadId;
readonly operation: string;
readonly allowRecovery: boolean;
}) =>
Effect.gen(function* () {
const bindingOption = yield* directory.getBinding(input.threadId);
const binding = Option.getOrUndefined(bindingOption);
if (!binding) {
return yield* toValidationError(
input.operation,
`Cannot route thread '${input.threadId}' because no persisted provider binding exists.`,
);
}
const adapter = yield* registry.getByProvider(binding.provider);
const hasRequestedSession = yield* adapter.hasSession(input.threadId);
if (hasRequestedSession) {
return { adapter, threadId: input.threadId, isActive: true } as const;
}
if (!input.allowRecovery) {
return { adapter, threadId: input.threadId, isActive: false } as const;
}
const recovered = yield* recoverSessionForThread({ binding, operation: input.operation });
return { adapter: recovered.adapter, threadId: input.threadId, isActive: true } as const;
});
const startSession: ProviderServiceShape["startSession"] = (threadId, rawInput) =>
Effect.gen(function* () {
const parsed = yield* decodeInputOrValidationError({
operation: "ProviderService.startSession",
schema: ProviderSessionStartInput,
payload: rawInput,
});
const input = {
...parsed,
threadId,
provider: parsed.provider ?? "codex",
};
const adapter = yield* registry.getByProvider(input.provider);
const session = yield* adapter.startSession(input);
if (session.provider !== adapter.provider) {
return yield* toValidationError(
"ProviderService.startSession",
`Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`,
);
}
yield* upsertSessionBinding(session, threadId);
return session;
});
const sendTurn: ProviderServiceShape["sendTurn"] = (rawInput) =>
Effect.gen(function* () {
const parsed = yield* decodeInputOrValidationError({
operation: "ProviderService.sendTurn",
schema: ProviderSendTurnInput,
payload: rawInput,
});
const input = {
...parsed,
attachments: parsed.attachments ?? [],
};
if (!input.input && input.attachments.length === 0) {
return yield* toValidationError(
"ProviderService.sendTurn",
"Either input text or at least one attachment is required",
);
}
const routed = yield* resolveRoutableSession({
threadId: input.threadId,
operation: "ProviderService.sendTurn",
allowRecovery: true,
});
const turn = yield* routed.adapter.sendTurn(input);
yield* directory.upsert({
threadId: input.threadId,
provider: routed.adapter.provider,
status: "running",
...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}),
runtimePayload: {
activeTurnId: turn.turnId,
lastRuntimeEvent: "provider.sendTurn",
lastRuntimeEventAt: new Date().toISOString(),
},
});
return turn;
});
const interruptTurn: ProviderServiceShape["interruptTurn"] = (rawInput) =>
Effect.gen(function* () {
const input = yield* decodeInputOrValidationError({
operation: "ProviderService.interruptTurn",
schema: ProviderInterruptTurnInput,
payload: rawInput,
});
const routed = yield* resolveRoutableSession({
threadId: input.threadId,
operation: "ProviderService.interruptTurn",
allowRecovery: true,
});
yield* routed.adapter.interruptTurn(routed.threadId, input.turnId);
});
const respondToRequest: ProviderServiceShape["respondToRequest"] = (rawInput) =>
Effect.gen(function* () {
const input = yield* decodeInputOrValidationError({
operation: "ProviderService.respondToRequest",
schema: ProviderRespondToRequestInput,
payload: rawInput,
});
const routed = yield* resolveRoutableSession({
threadId: input.threadId,
operation: "ProviderService.respondToRequest",
allowRecovery: true,
});
yield* routed.adapter.respondToRequest(routed.threadId, input.requestId, input.decision);
});
const respondToUserInput: ProviderServiceShape["respondToUserInput"] = (rawInput) =>
Effect.gen(function* () {
const input = yield* decodeInputOrValidationError({
operation: "ProviderService.respondToUserInput",
schema: ProviderRespondToUserInputInput,
payload: rawInput,
});
const routed = yield* resolveRoutableSession({
threadId: input.threadId,
operation: "ProviderService.respondToUserInput",
allowRecovery: true,
});
yield* routed.adapter.respondToUserInput(routed.threadId, input.requestId, input.answers);
});
const stopSession: ProviderServiceShape["stopSession"] = (rawInput) =>
Effect.gen(function* () {
const input = yield* decodeInputOrValidationError({
operation: "ProviderService.stopSession",
schema: ProviderStopSessionInput,
payload: rawInput,
});
const routed = yield* resolveRoutableSession({
threadId: input.threadId,
operation: "ProviderService.stopSession",
allowRecovery: false,
});
if (routed.isActive) {
yield* routed.adapter.stopSession(routed.threadId);
}
yield* directory.remove(input.threadId);
});
const listSessions: ProviderServiceShape["listSessions"] = () =>
Effect.gen(function* () {
const sessionsByProvider = yield* Effect.forEach(adapters, (adapter) => adapter.listSessions());
const activeSessions = sessionsByProvider.flatMap((sessions) => sessions);
const persistedBindings = yield* directory
.listThreadIds()
.pipe(
Effect.flatMap((threadIds) =>
Effect.forEach(
threadIds,
(threadId) =>
directory.getBinding(threadId).pipe(
Effect.orElseSucceed(() => Option.none<ProviderRuntimeBinding>()),
),
{ concurrency: "unbounded" },
),
),
Effect.orElseSucceed(() => [] as Array<Option.Option<ProviderRuntimeBinding>>),
);
const bindingsByThreadId = new Map<ThreadId, ProviderRuntimeBinding>();
for (const bindingOption of persistedBindings) {
const binding = Option.getOrUndefined(bindingOption);
if (binding) {
bindingsByThreadId.set(binding.threadId, binding);
}
}
return activeSessions.map((session) => {
const binding = bindingsByThreadId.get(session.threadId);
if (!binding) {
return session;
}
return {
...session,
...(session.resumeCursor === undefined && binding.resumeCursor !== undefined
? { resumeCursor: binding.resumeCursor }
: {}),
...(binding.runtimeMode !== undefined ? { runtimeMode: binding.runtimeMode } : {}),
};
});
});
const getCapabilities: ProviderServiceShape["getCapabilities"] = (provider) =>
registry.getByProvider(provider).pipe(Effect.map((adapter) => adapter.capabilities));
const rollbackConversation: ProviderServiceShape["rollbackConversation"] = (rawInput) =>
Effect.gen(function* () {
const input = yield* decodeInputOrValidationError({
operation: "ProviderService.rollbackConversation",
schema: ProviderRollbackConversationInput,
payload: rawInput,
});
if (input.numTurns === 0) {
return;
}
const routed = yield* resolveRoutableSession({
threadId: input.threadId,
operation: "ProviderService.rollbackConversation",
allowRecovery: true,
});
yield* routed.adapter.rollbackThread(routed.threadId, input.numTurns);
});
const stopAll: ProviderServiceShape["stopAll"] = () =>
Effect.gen(function* () {
const threadIds = yield* directory.listThreadIds();
yield* Effect.forEach(adapters, (adapter) => adapter.stopAll()).pipe(Effect.asVoid);
yield* Effect.forEach(threadIds, (threadId) =>
directory.getProvider(threadId).pipe(
Effect.flatMap((provider) =>
directory.upsert({
threadId,
provider,
status: "stopped",
runtimePayload: {
activeTurnId: null,
lastRuntimeEvent: "provider.stopAll",
lastRuntimeEventAt: new Date().toISOString(),
},
}),
),
),
).pipe(Effect.asVoid);
});
return {
startSession,
sendTurn,
interruptTurn,
respondToRequest,
respondToUserInput,
stopSession,
listSessions,
getCapabilities,
rollbackConversation,
stopAll,
streamEvents: Stream.fromPubSub(runtimeEventPubSub),
} satisfies ProviderServiceShape;
});
export const ProviderServiceLive = Layer.effect(ProviderService, makeProviderService());
export function makeProviderServiceLive(options?: ProviderServiceLiveOptions) {
return Layer.effect(ProviderService, makeProviderService(options));
}