/** * 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 {  ModelSelection,  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";import { ServerSettingsService } from "../../serverSettings.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 modelSelection?: 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?.modelSelection !== undefined ? { modelSelection: extra.modelSelection } : {}),    ...(extra?.lastRuntimeEvent !== undefined ? { lastRuntimeEvent: extra.lastRuntimeEvent } : {}),    ...(extra?.lastRuntimeEventAt !== undefined      ? { lastRuntimeEventAt: extra.lastRuntimeEventAt }      : {}),  };} function readPersistedModelSelection(  runtimePayload: ProviderRuntimeBinding["runtimePayload"],): ModelSelection | undefined {  if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) {    return undefined;  }  const raw = "modelSelection" in runtimePayload ? runtimePayload.modelSelection : undefined;  return Schema.is(ModelSelection)(raw) ? raw : undefined;} 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 = Effect.fn("makeProviderService")(function* (  options?: ProviderServiceLiveOptions,) {  const analytics = yield* Effect.service(AnalyticsService);  const serverSettings = yield* ServerSettingsService;  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 modelSelection?: 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 = Effect.fn("recoverSessionForThread")(function* (input: {    readonly binding: ProviderRuntimeBinding;    readonly operation: string;  }) {    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 persistedModelSelection = readPersistedModelSelection(input.binding.runtimePayload);     const resumed = yield* adapter.startSession({      threadId: input.binding.threadId,      provider: input.binding.provider,      ...(persistedCwd ? { cwd: persistedCwd } : {}),      ...(persistedModelSelection ? { modelSelection: persistedModelSelection } : {}),      ...(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 = Effect.fn("resolveRoutableSession")(function* (input: {    readonly threadId: ThreadId;    readonly operation: string;    readonly allowRecovery: boolean;  }) {    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"] = Effect.fn("startSession")(    function* (threadId, rawInput) {      const parsed = yield* decodeInputOrValidationError({        operation: "ProviderService.startSession",        schema: ProviderSessionStartInput,        payload: rawInput,      });       const input = {        ...parsed,        threadId,        provider: parsed.provider ?? "codex",      };      const settings = yield* serverSettings.getSettings.pipe(        Effect.mapError((error) =>          toValidationError(            "ProviderService.startSession",            `Failed to load provider settings: ${error.message}`,            error,          ),        ),      );      if (!settings.providers[input.provider].enabled) {        return yield* toValidationError(          "ProviderService.startSession",          `Provider '${input.provider}' is disabled in T3 Code settings.`,        );      }      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, {        modelSelection: input.modelSelection,      });      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.modelSelection?.model === "string" &&          input.modelSelection.model.trim().length > 0,      });       return session;    },  );   const sendTurn: ProviderServiceShape["sendTurn"] = Effect.fn("sendTurn")(function* (rawInput) {    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: {        ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}),        activeTurnId: turn.turnId,        lastRuntimeEvent: "provider.sendTurn",        lastRuntimeEventAt: new Date().toISOString(),      },    });    yield* analytics.record("provider.turn.sent", {      provider: routed.adapter.provider,      model: input.modelSelection?.model,      interactionMode: input.interactionMode,      attachmentCount: input.attachments.length,      hasInput: typeof input.input === "string" && input.input.trim().length > 0,    });    return turn;  });   const interruptTurn: ProviderServiceShape["interruptTurn"] = Effect.fn("interruptTurn")(    function* (rawInput) {      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"] = Effect.fn("respondToRequest")(    function* (rawInput) {      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"] = Effect.fn(    "respondToUserInput",  )(function* (rawInput) {    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"] = Effect.fn("stopSession")(    function* (rawInput) {      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.fn("listSessions")(    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"];

Lines 1–500 of 602. Load more