/** * ProviderService - Cross-provider session and turn service. * * 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). * Uses Effect `Context.Service` for dependency injection and returns typed * domain errors for validation, session, codex, and checkpoint workflows. * * @module ProviderService */import {  ModelSelection,  NonNegativeInt,  ThreadId,  ProviderInterruptTurnInput,  ProviderRespondToRequestInput,  ProviderRespondToUserInputInput,  ProviderSendTurnInput,  ProviderSessionStartInput,  ProviderStopSessionInput,  ProviderDriverKind,  ProviderInstanceId,  type ProviderRuntimeEvent,  type ProviderSession,  type ProviderTurnStartResult,} from "@t3tools/contracts";import * as Context from "effect/Context";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,  type ProviderServiceError,  ProviderValidationError,} from "./Errors.ts";import type {  ProviderAdapterCapabilities,  ProviderAdapterShape,} from "./Services/ProviderAdapter.ts";import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";import * as ProviderSessionDirectory from "./ProviderSessionDirectory.ts";import { type EventNdjsonLogger } from "./Layers/EventNdjsonLogger.ts";import * as ProviderEventLoggers from "./ProviderEventLoggers.ts";import * as AnalyticsService from "../telemetry/AnalyticsService.ts";import {  clearAllMcpProviderSessions,  clearMcpProviderSession,  setMcpProviderSession,} from "../mcp/McpProviderSession.ts";import {  issueActiveMcpCredential,  revokeActiveMcpThread,  revokeAllActiveMcpCredentials,} from "../mcp/McpSessionRegistry.ts"; export class ProviderBindingInstanceIdMissingError extends Schema.TaggedErrorClass<ProviderBindingInstanceIdMissingError>()(  "ProviderBindingInstanceIdMissingError",  {    operation: Schema.String,    provider: Schema.optional(ProviderDriverKind),  },) {  override get message(): string {    const provider = this.provider === undefined ? "" : ` for provider '${this.provider}'`;    return `Provider binding is missing an instance id${provider} in ${this.operation}.`;  }} export class ProviderRuntimeEventCorrelationError extends Schema.TaggedErrorClass<ProviderRuntimeEventCorrelationError>()(  "ProviderRuntimeEventCorrelationError",  {    sourceInstanceId: ProviderInstanceId,    sourceProvider: ProviderDriverKind,    eventInstanceId: Schema.optional(ProviderInstanceId),    eventProvider: ProviderDriverKind,  },) {  override get message(): string {    const eventInstanceId = this.eventInstanceId ?? "unattributed";    return `Provider runtime event correlation failed for source instance '${this.sourceInstanceId}' (${this.sourceProvider}) and emitted instance '${eventInstanceId}' (${this.eventProvider}).`;  }} export class ProviderSessionBindingCorrelationError extends Schema.TaggedErrorClass<ProviderSessionBindingCorrelationError>()(  "ProviderSessionBindingCorrelationError",  {    threadId: ThreadId,    sessionProvider: ProviderDriverKind,    bindingProvider: ProviderDriverKind,    sessionInstanceId: ProviderInstanceId,    bindingInstanceId: ProviderInstanceId,  },) {  override get message(): string {    return `Active provider session and persisted binding disagree for thread '${this.threadId}'.`;  }} export class ProviderService extends Context.Service<  ProviderService,  {    /** Start a provider session. */    readonly startSession: (      threadId: ThreadId,      input: ProviderSessionStartInput,    ) => Effect.Effect<ProviderSession, ProviderServiceError>;     /** Send a turn through the adapter bound to the session. */    readonly sendTurn: (      input: ProviderSendTurnInput,    ) => Effect.Effect<ProviderTurnStartResult, ProviderServiceError>;     /** Interrupt a running provider turn. */    readonly interruptTurn: (      input: ProviderInterruptTurnInput,    ) => Effect.Effect<void, ProviderServiceError>;     /** Respond to a provider approval request. */    readonly respondToRequest: (      input: ProviderRespondToRequestInput,    ) => Effect.Effect<void, ProviderServiceError>;     /** Respond to a structured provider user-input request. */    readonly respondToUserInput: (      input: ProviderRespondToUserInputInput,    ) => Effect.Effect<void, ProviderServiceError>;     /** Stop a provider session. */    readonly stopSession: (      input: ProviderStopSessionInput,    ) => Effect.Effect<void, ProviderServiceError>;     /**     * List active provider sessions.     *     * Aggregates runtime session lists from all registered adapters.     */    readonly listSessions: () => Effect.Effect<ReadonlyArray<ProviderSession>>;     /** Read capabilities for the adapter bound to a configured provider instance. */    readonly getCapabilities: (      instanceId: ProviderInstanceId,    ) => Effect.Effect<ProviderAdapterCapabilities, ProviderServiceError>;     /** Read routing metadata for a configured provider instance. */    readonly getInstanceInfo: (      instanceId: ProviderInstanceId,    ) => Effect.Effect<ProviderAdapterRegistry.ProviderInstanceRoutingInfo, ProviderServiceError>;     /** Roll back provider conversation state by a number of turns. */    readonly rollbackConversation: (input: {      readonly threadId: ThreadId;      readonly numTurns: number;    }) => Effect.Effect<void, ProviderServiceError>;     /**     * Canonical provider runtime event stream.     *     * Fan-out is owned by ProviderService (not by a standalone event-bus service).     */    readonly streamEvents: Stream.Stream<ProviderRuntimeEvent>;  }>()("t3/provider/ProviderService") {} 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 ProviderServiceOptions {  readonly canonicalEventLogger?: EventNdjsonLogger;} const ProviderRollbackConversationInput = Schema.Struct({  threadId: ThreadId,  numTurns: NonNegativeInt,}); 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 ProviderBindingInstanceIdMissingError({    operation,    ...(payload.provider === undefined ? {} : { provider: payload.provider }),  });}; const correlateRuntimeEventWithInstance = (  source: {    readonly instanceId: ProviderInstanceId;    readonly provider: ProviderDriverKind;  },  event: ProviderRuntimeEvent,): ProviderRuntimeEvent => {  if (    event.provider !== source.provider ||    (event.providerInstanceId !== undefined && event.providerInstanceId !== source.instanceId)  ) {    throw new ProviderRuntimeEventCorrelationError({      sourceInstanceId: source.instanceId,      sourceProvider: source.provider,      ...(event.providerInstanceId === undefined        ? {}        : { eventInstanceId: event.providerInstanceId }),      eventProvider: event.provider,    });  }  return { ...event, providerInstanceId: source.instanceId };}; export const make = Effect.fn("ProviderService.make")(function* (options?: ProviderServiceOptions) {  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) =>    issueActiveMcpCredential({ threadId, providerInstanceId }).pipe(      Effect.tap((credential) =>        credential ? Effect.sync(() => setMcpProviderSession(credential.config)) : Effect.void,      ),    );  const clearMcpSession = (threadId: ThreadId) =>    revokeActiveMcpThread(threadId).pipe(      Effect.tap(() => Effect.sync(() => 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(          new ProviderValidationError({            operation,            issue: 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) {

Lines 1–500 of 1199. Load more