/** * 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, Schema, SchemaIssue, Stream } from "effect"; import {  increment,  providerMetricAttributes,  providerRuntimeEventsTotal,  providerSessionsTotal,  providerTurnDuration,  providerTurnsTotal,  providerTurnMetricAttributes,  withMetrics,} from "../../observability/Metrics.ts";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 runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();   const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>    Effect.succeed(event).pipe(      Effect.tap((canonicalEvent) =>        canonicalEventLogger          ? canonicalEventLogger.write(canonicalEvent, canonicalEvent.threadId)          : 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> =>    increment(providerRuntimeEventsTotal, {      provider: event.provider,      eventType: event.type,    }).pipe(Effect.andThen(publishRuntimeEvent(event)));   yield* Effect.forEach(adapters, (adapter) =>    Stream.runForEach(adapter.streamEvents, processRuntimeEvent).pipe(Effect.forkScoped),  ).pipe(Effect.asVoid);   const recoverSessionForThread = Effect.fn("recoverSessionForThread")(function* (input: {    readonly binding: ProviderRuntimeBinding;    readonly operation: string;  }) {    yield* Effect.annotateCurrentSpan({      "provider.operation": "recover-session",      "provider.kind": input.binding.provider,      "provider.thread_id": input.binding.threadId,    });    return yield* 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 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;    }).pipe(      withMetrics({        counter: providerSessionsTotal,        attributes: providerMetricAttributes(input.binding.provider, {          operation: "recover",        }),      }),    );  });   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 stopStaleSessionsForThread = Effect.fn("stopStaleSessionsForThread")(function* (input: {    readonly threadId: ThreadId;    readonly currentProvider: ProviderSession["provider"];  }) {    yield* Effect.forEach(      adapters,      (adapter) =>        adapter.provider === input.currentProvider          ? Effect.void          : Effect.gen(function* () {              const hasSession = yield* adapter.hasSession(input.threadId);              if (!hasSession) {                return;              }               yield* adapter.stopSession(input.threadId).pipe(                Effect.tap(() =>                  analytics.record("provider.session.stopped", {                    provider: adapter.provider,                  }),                ),                Effect.catchCause((cause) =>                  Effect.logWarning("provider.session.stop-stale-failed", {                    threadId: input.threadId,                    provider: adapter.provider,                    cause,                  }),                ),              );            }),      { discard: true },    );  });   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",      };      yield* Effect.annotateCurrentSpan({        "provider.operation": "start-session",        "provider.kind": input.provider,        "provider.thread_id": threadId,        "provider.runtime_mode": input.runtimeMode,      });      return yield* Effect.gen(function* () {        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 effectiveCwd =          input.cwd ??          (persistedBinding?.provider === input.provider            ? readPersistedCwd(persistedBinding.runtimePayload)            : undefined);        yield* Effect.annotateCurrentSpan({          "provider.resume_cursor.source":            input.resumeCursor !== undefined              ? "request"              : effectiveResumeCursor !== undefined && persistedBinding?.provider === input.provider                ? "persisted"                : "none",          "provider.resume_cursor.present": effectiveResumeCursor !== undefined,          "provider.cwd.source":            input.cwd !== undefined              ? "request"              : effectiveCwd !== undefined && persistedBinding?.provider === input.provider                ? "persisted"                : "none",          "provider.cwd.effective": effectiveCwd ?? "",        });        const adapter = yield* registry.getByProvider(input.provider);        const session = yield* adapter.startSession({          ...input,          ...(effectiveCwd !== undefined ? { cwd: effectiveCwd } : {}),          ...(effectiveResumeCursor !== undefined ? { resumeCursor: effectiveResumeCursor } : {}),        });         if (session.provider !== adapter.provider) {          return yield* toValidationError(            "ProviderService.startSession",            `Adapter/provider mismatch: requested '${adapter.provider}', received '${session.provider}'.`,          );        }         yield* stopStaleSessionsForThread({          threadId,          currentProvider: adapter.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 effectiveCwd === "string" && effectiveCwd.trim().length > 0,          hasModel:            typeof input.modelSelection?.model === "string" &&            input.modelSelection.model.trim().length > 0,        });         return session;      }).pipe(        withMetrics({          counter: providerSessionsTotal,          attributes: providerMetricAttributes(input.provider, {            operation: "start",          }),        }),      );    },  );   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",      );    }    yield* Effect.annotateCurrentSpan({      "provider.operation": "send-turn",      "provider.thread_id": input.threadId,      "provider.interaction_mode": input.interactionMode,      "provider.attachment_count": input.attachments.length,    });    let metricProvider = "unknown";    let metricModel = input.modelSelection?.model;    return yield* Effect.gen(function* () {      const routed = yield* resolveRoutableSession({        threadId: input.threadId,        operation: "ProviderService.sendTurn",        allowRecovery: true,      });      metricProvider = routed.adapter.provider;      metricModel = input.modelSelection?.model;      yield* Effect.annotateCurrentSpan({        "provider.kind": routed.adapter.provider,        ...(input.modelSelection?.model ? { "provider.model": input.modelSelection.model } : {}),      });      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;    }).pipe(      withMetrics({

Lines 1–500 of 825. Load more