/** * 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 ProviderInstanceId,  type ProviderDriverKind,  type ProviderRuntimeEvent,  type ProviderSession,} from "@t3tools/contracts";import { causeErrorTag } from "@t3tools/shared/observability";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as PubSub from "effect/PubSub";import * as Ref from "effect/Ref";import * as Schema from "effect/Schema";import * as SchemaIssue from "effect/SchemaIssue";import * as Stream from "effect/Stream"; import {  increment,  providerMetricAttributes,  providerRuntimeEventsTotal,  providerSessionsTotal,  providerTurnDuration,  providerTurnsTotal,  providerTurnMetricAttributes,  withMetrics,} from "../../observability/Metrics.ts";import { type ProviderAdapterError, ProviderValidationError } from "../Errors.ts";import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";import * as ProviderAdapterRegistry from "../Services/ProviderAdapterRegistry.ts";import * as ProviderService from "../Services/ProviderService.ts";import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts";import { type EventNdjsonLogger } from "./EventNdjsonLogger.ts";import * as ProviderEventLoggers from "./ProviderEventLoggers.ts";import * as AnalyticsService from "../../telemetry/AnalyticsService.ts";import * as McpProviderSession from "../../mcp/McpProviderSession.ts";import * as McpSessionRegistry from "../../mcp/McpSessionRegistry.ts";const isModelSelection = Schema.is(ModelSelection); /** * Hook for tests that want to override the canonical event logger pulled * from `ProviderEventLoggers`. Production wiring leaves this undefined and * reads the logger off the tag. */export interface ProviderServiceLiveOptions {  readonly canonicalEventLogger?: EventNdjsonLogger;} type ProviderServiceMethod<Name extends keyof ProviderService.ProviderService["Service"]> =  ProviderService.ProviderService["Service"][Name]; 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;}) => {  const decodeProviderRequestInput = Schema.decodeUnknownEffect(input.schema);  return decodeProviderRequestInput(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: ProviderSessionDirectory.ProviderRuntimeBinding["runtimePayload"],): ModelSelection | undefined {  if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) {    return undefined;  }  const raw = "modelSelection" in runtimePayload ? runtimePayload.modelSelection : undefined;  return isModelSelection(raw) ? raw : undefined;} function readPersistedCwd(  runtimePayload: ProviderSessionDirectory.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 dieOnMissingBindingInstanceId = (  operation: string,  payload: {    readonly providerInstanceId?: ProviderInstanceId | undefined;    readonly provider?: ProviderDriverKind | undefined;  },): ProviderInstanceId => {  if (payload.providerInstanceId !== undefined) {    return payload.providerInstanceId;  }  throw new Error(    payload.provider      ? `${operation}: provider instance id is required for provider '${payload.provider}'.`      : `${operation}: provider instance id is required.`,  );}; const correlateRuntimeEventWithInstance = (  source: {    readonly instanceId: ProviderInstanceId;    readonly provider: ProviderDriverKind;  },  event: ProviderRuntimeEvent,): ProviderRuntimeEvent => {  if (event.provider !== source.provider) {    throw new Error(      `ProviderService.streamEvents: provider instance '${source.instanceId}' is backed by driver '${source.provider}' but emitted driver '${event.provider}'.`,    );  }  if (event.providerInstanceId !== undefined && event.providerInstanceId !== source.instanceId) {    throw new Error(      `ProviderService.streamEvents: provider instance '${source.instanceId}' emitted event for instance '${event.providerInstanceId}'.`,    );  }  return { ...event, providerInstanceId: source.instanceId };}; const makeProviderService = Effect.fn("makeProviderService")(function* (  options?: ProviderServiceLiveOptions,) {  const analytics = yield* Effect.service(AnalyticsService.AnalyticsService);  const eventLoggers = yield* ProviderEventLoggers.ProviderEventLoggers;  // Options-provided logger wins (test overrides); otherwise we take whatever  // the `ProviderEventLoggers` tag exposes — `undefined` means "no canonical  // log writer is attached", which downstream code already handles as a  // no-op.  const canonicalEventLogger = options?.canonicalEventLogger ?? eventLoggers.canonical;   const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistry;  const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;  const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);  const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) =>    McpSessionRegistry.issueActiveMcpCredential({ threadId, providerInstanceId }).pipe(      Effect.tap((credential) =>        credential          ? Effect.sync(() => McpProviderSession.setMcpProviderSession(credential.config))          : Effect.void,      ),    );  const clearMcpSession = (threadId: ThreadId) =>    McpSessionRegistry.revokeActiveMcpThread(threadId).pipe(      Effect.tap(() => Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId))),    );   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 requireBindingInstanceId = (    operation: string,    payload: {      readonly providerInstanceId?: ProviderInstanceId | undefined;      readonly provider?: ProviderDriverKind | undefined;    },  ): Effect.Effect<ProviderInstanceId, ProviderValidationError> =>    payload.providerInstanceId !== undefined      ? Effect.succeed(payload.providerInstanceId)      : Effect.fail(          toValidationError(            operation,            payload.provider              ? `Provider instance id is required for provider '${payload.provider}'.`              : "Provider instance id is required.",          ),        );   const upsertSessionBinding = (    session: ProviderSession,    threadId: ThreadId,    extra?: {      readonly modelSelection?: unknown;      readonly lastRuntimeEvent?: string;      readonly lastRuntimeEventAt?: string;    },  ) =>    Effect.gen(function* () {      const providerInstanceId = yield* requireBindingInstanceId(        "ProviderService.upsertSessionBinding",        session,      );      yield* directory.upsert({        threadId,        provider: session.provider,        providerInstanceId,        runtimeMode: session.runtimeMode,        status: toRuntimeStatus(session),        ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}),        runtimePayload: toRuntimePayloadFromSession(session, extra),      });    });   const processRuntimeEvent = (    source: {      readonly instanceId: ProviderInstanceId;      readonly provider: ProviderDriverKind;    },    event: ProviderRuntimeEvent,  ): Effect.Effect<void> =>    Effect.sync(() => correlateRuntimeEventWithInstance(source, event)).pipe(      Effect.flatMap((canonicalEvent) =>        increment(providerRuntimeEventsTotal, {          provider: canonicalEvent.provider,          eventType: canonicalEvent.type,        }).pipe(Effect.andThen(publishRuntimeEvent(canonicalEvent))),      ),    );   // `subscribedAdapters` is our source-of-truth for "which instance adapters  // are currently wired into the runtime event bus". It both tracks the set  // of live subscriptions (so `reconcileInstanceSubscriptions` can diff and  // fork only the *new* or *rebuilt* ones) and serves as the dynamic adapter  // list consumed by `stopStaleSessionsForThread`, `listSessions`, and  // `runStopAll` — replacing the pre-Slice-D startup snapshot so hot-added  // instances become visible to those call sites as soon as settings edits  // land.  const subscribedAdapters = yield* Ref.make(    new Map<ProviderInstanceId, ProviderAdapterShape<ProviderAdapterError>>(),  );   const getAdapterEntries = Ref.get(subscribedAdapters).pipe(    Effect.map((map) => Array.from(map.entries())),  );   // Rebuild the map of id → adapter from the registry and fork a new event  // subscription for every instance that is either brand new or whose adapter  // identity changed (indicating the underlying `ProviderInstance` was torn  // down and rebuilt by `ProviderInstanceRegistry.reconcile`). Orphaned  // fibers for removed/replaced instances exit on their own because their  // adapter's `streamEvents` source terminates when the old scope closes.  const reconcileInstanceSubscriptions = Effect.gen(function* () {    const previous = yield* Ref.get(subscribedAdapters);    const currentIds = yield* registry.listInstances();    const next = new Map<ProviderInstanceId, ProviderAdapterShape<ProviderAdapterError>>();    for (const id of currentIds) {      const adapterOption = yield* registry        .getByInstance(id)        .pipe(Effect.tapError(Effect.logWarning), Effect.option);      if (Option.isNone(adapterOption)) continue;      const adapter = adapterOption.value;      next.set(id, adapter);      if (previous.get(id) !== adapter) {        yield* Stream.runForEach(adapter.streamEvents, (event) =>          processRuntimeEvent(            {              instanceId: id,              provider: adapter.provider,            },            event,          ),        ).pipe(Effect.forkScoped);      }    }    yield* Ref.set(subscribedAdapters, next);  });   const instanceChanges = yield* registry.subscribeChanges;  yield* reconcileInstanceSubscriptions;  yield* Stream.runForEach(    Stream.fromSubscription(instanceChanges),    () => reconcileInstanceSubscriptions,  ).pipe(Effect.forkScoped);   const recoverSessionForThread = Effect.fn("recoverSessionForThread")(function* (input: {    readonly binding: ProviderSessionDirectory.ProviderRuntimeBinding;    readonly operation: string;  }) {    const bindingInstanceId = yield* requireBindingInstanceId(input.operation, input.binding);    yield* Effect.annotateCurrentSpan({      "provider.operation": "recover-session",      "provider.kind": input.binding.provider,      "provider.instance_id": bindingInstanceId,      "provider.thread_id": input.binding.threadId,    });    return yield* Effect.gen(function* () {      const adapter = yield* registry.getByInstance(bindingInstanceId);      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, providerInstanceId: bindingInstanceId },            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);       yield* prepareMcpSession(input.binding.threadId, bindingInstanceId);      const resumed = yield* adapter        .startSession({          threadId: input.binding.threadId,          provider: input.binding.provider,          providerInstanceId: bindingInstanceId,          ...(persistedCwd ? { cwd: persistedCwd } : {}),          ...(persistedModelSelection ? { modelSelection: persistedModelSelection } : {}),          ...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}),          runtimeMode: input.binding.runtimeMode ?? "full-access",        })        .pipe(Effect.onError(() => clearMcpSession(input.binding.threadId)));      if (resumed.provider !== adapter.provider) {        yield* clearMcpSession(input.binding.threadId);        return yield* toValidationError(          input.operation,          `Adapter/provider mismatch while recovering thread '${input.binding.threadId}'. Expected '${adapter.provider}', received '${resumed.provider}'.`,        );      }       yield* upsertSessionBinding(        { ...resumed, providerInstanceId: bindingInstanceId },        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 instanceId = yield* requireBindingInstanceId(input.operation, binding);    const adapter = yield* registry.getByInstance(instanceId);     const hasRequestedSession = yield* adapter.hasSession(input.threadId);    if (hasRequestedSession) {      return {        adapter,        instanceId,        threadId: input.threadId,        isActive: true,      } as const;    }     if (!input.allowRecovery) {      return {        adapter,        instanceId,        threadId: input.threadId,        isActive: false,      } as const;    }     const recovered = yield* recoverSessionForThread({      binding,      operation: input.operation,    });    return {      adapter: recovered.adapter,      instanceId,      threadId: input.threadId,      isActive: true,    } as const;  });   const stopStaleSessionsForThread = Effect.fn("stopStaleSessionsForThread")(function* (input: {    readonly threadId: ThreadId;    readonly currentInstanceId: ProviderInstanceId;  }) {    const currentAdapters = yield* getAdapterEntries;    yield* Effect.forEach(      currentAdapters,      ([instanceId, adapter]) =>        instanceId === input.currentInstanceId          ? Effect.void          : Effect.gen(function* () {              const hasSession = yield* adapter.hasSession(input.threadId);              if (!hasSession) {                return;

Lines 1–500 of 1098. Load more