/** * 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";import { AnalyticsService } from "../../telemetry/Services/AnalyticsService.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,  extra?: {    readonly modelOptions?: unknown;    readonly providerOptions?: unknown;    readonly lastRuntimeEvent?: string;    readonly lastRuntimeEventAt?: string;  },): Record<string, unknown> {  return {    cwd: session.cwd ?? null,    model: session.model ?? null,    activeTurnId: session.activeTurnId ?? null,    lastError: session.lastError ?? null,    ...(extra?.modelOptions !== undefined ? { modelOptions: extra.modelOptions } : {}),    ...(extra?.providerOptions !== undefined ? { providerOptions: extra.providerOptions } : {}),    ...(extra?.lastRuntimeEvent !== undefined ? { lastRuntimeEvent: extra.lastRuntimeEvent } : {}),    ...(extra?.lastRuntimeEventAt !== undefined      ? { lastRuntimeEventAt: extra.lastRuntimeEventAt }      : {}),  };} function readPersistedModelOptions(  runtimePayload: ProviderRuntimeBinding["runtimePayload"],): Record<string, unknown> | undefined {  if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) {    return undefined;  }  const raw = "modelOptions" in runtimePayload ? runtimePayload.modelOptions : undefined;  if (!raw || typeof raw !== "object" || Array.isArray(raw)) return undefined;  return raw as Record<string, unknown>;} function readPersistedProviderOptions(  runtimePayload: ProviderRuntimeBinding["runtimePayload"],): Record<string, unknown> | undefined {  if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) {    return undefined;  }  const raw = "providerOptions" in runtimePayload ? runtimePayload.providerOptions : undefined;  if (!raw || typeof raw !== "object" || Array.isArray(raw)) return undefined;  return raw as Record<string, unknown>;} 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 analytics = yield* Effect.service(AnalyticsService);    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,      extra?: {        readonly modelOptions?: unknown;        readonly providerOptions?: unknown;        readonly lastRuntimeEvent?: string;        readonly lastRuntimeEventAt?: string;      },    ) =>      directory.upsert({        threadId,        provider: session.provider,        runtimeMode: session.runtimeMode,        status: toRuntimeStatus(session),        ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}),        runtimePayload: toRuntimePayloadFromSession(session, extra),      });     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);            yield* analytics.record("provider.session.recovered", {              provider: existing.provider,              strategy: "adopt-existing",              hasResumeCursor: existing.resumeCursor !== undefined,            });            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 persistedModelOptions = readPersistedModelOptions(input.binding.runtimePayload);        const persistedProviderOptions = readPersistedProviderOptions(input.binding.runtimePayload);         const resumed = yield* adapter.startSession({          threadId: input.binding.threadId,          provider: input.binding.provider,          ...(persistedCwd ? { cwd: persistedCwd } : {}),          ...(persistedModelOptions ? { modelOptions: persistedModelOptions } : {}),          ...(persistedProviderOptions ? { providerOptions: persistedProviderOptions } : {}),          ...(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);        yield* analytics.record("provider.session.recovered", {          provider: resumed.provider,          strategy: "resume-thread",          hasResumeCursor: resumed.resumeCursor !== undefined,        });        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 persistedBinding = Option.getOrUndefined(yield* directory.getBinding(threadId));        const effectiveResumeCursor =          input.resumeCursor ??          (persistedBinding?.provider === input.provider            ? persistedBinding.resumeCursor            : undefined);        const adapter = yield* registry.getByProvider(input.provider);        const session = yield* adapter.startSession({          ...input,          ...(effectiveResumeCursor !== undefined ? { resumeCursor: effectiveResumeCursor } : {}),        });         if (session.provider !== adapter.provider) {          return yield* toValidationError(            "ProviderService.startSession",            `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`,          );        }         yield* upsertSessionBinding(session, threadId, {          modelOptions: input.modelOptions,          providerOptions: input.providerOptions,        });        yield* analytics.record("provider.session.started", {          provider: session.provider,          runtimeMode: input.runtimeMode,          hasResumeCursor: session.resumeCursor !== undefined,          hasCwd: typeof input.cwd === "string" && input.cwd.trim().length > 0,          hasModel: typeof input.model === "string" && input.model.trim().length > 0,        });         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(),          },        });        yield* analytics.record("provider.turn.sent", {          provider: routed.adapter.provider,          model: input.model,          interactionMode: input.interactionMode,          attachmentCount: input.attachments.length,          hasInput: typeof input.input === "string" && input.input.trim().length > 0,        });        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);        yield* analytics.record("provider.turn.interrupted", {          provider: routed.adapter.provider,        });      });     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);        yield* analytics.record("provider.request.responded", {          provider: routed.adapter.provider,          decision: 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);        yield* analytics.record("provider.session.stopped", {          provider: routed.adapter.provider,        });      });     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;          }           const overrides: {            resumeCursor?: ProviderSession["resumeCursor"];            runtimeMode?: ProviderSession["runtimeMode"];          } = {};          if (session.resumeCursor === undefined && binding.resumeCursor !== undefined) {

Lines 1–500 of 601. Load more