/** * 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 {  EventId,  MessageId,  ModelSelection,  NonNegativeInt,  ProviderInterruptTurnInput,  ProviderRespondToRequestInput,  ProviderRespondToUserInputInput,  RuntimeRequestId,  ProviderSendTurnInput,  ProviderSessionStartInput,  ProviderStopSessionInput,  ProviderUploadFeedbackInput,  ThreadId,  TurnId,  type ProviderInstanceId,  type ProviderDriverKind,  type ProviderRuntimeEvent,  type ProviderSession,} from "@t3tools/contracts";import { expandAssistantCitationsForProvider } from "@t3tools/shared/assistantCitations";import { causeErrorTag } from "@t3tools/shared/observability";import { getModelSelectionStringOptionValue } from "@t3tools/shared/model";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as FileSystem from "effect/FileSystem";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 { resolveAttachmentPath } from "../../attachmentStore.ts";import * as ServerConfig from "../../config.ts";import {  increment,  providerMetricAttributes,  providerRuntimeEventsTotal,  providerSessionsTotal,  providerTurnDuration,  providerTurnsTotal,  providerTurnMetricAttributes,  withMetrics,} from "../../observability/Metrics.ts";import {  ProviderAdapterRequestError,  type ProviderAdapterError,  ProviderValidationError,  ProviderWorkspaceMissingError,} 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";import * as ServerSettings from "../../serverSettings.ts";const isModelSelection = Schema.is(ModelSelection); interface PendingCompaction {  readonly completion: Deferred.Deferred<string>;  readonly native: boolean;  readonly providerInstanceId: ProviderInstanceId;  readonly requestId: MessageId | undefined;  readonly earlyEvents: ProviderRuntimeEvent[];  compactedEventObserved: boolean;  expectedTurnId: TurnId | undefined;} /** * 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;  /**   * Overrides MCP credential issuance. The real issuer reads a module-global   * registry that only a running MCP server installs, which makes the   * agent-browser-access gate unobservable from a unit test; this seam lets a   * test see whether a credential was requested at all.   */  readonly issueMcpCredential?: typeof McpSessionRegistry.issueActiveMcpCredential;  /** Same seam as `issueMcpCredential`, for observing the deny path's revoke. */  readonly revokeMcpCredential?: typeof McpSessionRegistry.revokeActiveMcpThread;} interface TurnAnalyticsMetadata {  readonly requestId: number;  readonly provider: ProviderDriverKind;  readonly startedAtMs: number;  readonly mixedModels: boolean;  readonly model?: string;  readonly effort?: string;  readonly interactionMode?: string;  readonly runtimeMode?: string;} interface ActiveTurnAnalytics {  readonly metadata: TurnAnalyticsMetadata;  readonly requestAssociated: boolean;} interface DeferredTurnAnalyticsCompletion {  readonly completionKey: string;  readonly completedAtMs: number;  readonly terminalProperties: Readonly<Record<string, unknown>>;} interface TurnAnalyticsSessionState {  readonly pendingByRequestId: Map<number, TurnAnalyticsMetadata>;  readonly activeByTurnId: Map<string, ActiveTurnAnalytics>;  readonly deferredCompletionsByTurnId: Map<string, DeferredTurnAnalyticsCompletion>;} interface TurnAnalyticsState {  readonly sessions: Map<string, TurnAnalyticsSessionState>;  readonly completedKeys: Set<string>;  readonly completedOrder: Array<string>;} const MAX_COMPLETED_TURN_ANALYTICS_KEYS = 512;const MAX_ACTIVE_TURN_ANALYTICS_PER_SESSION = 8; function setActiveTurnAnalytics(  session: TurnAnalyticsSessionState,  turnId: string,  active: ActiveTurnAnalytics,): void {  session.activeByTurnId.set(turnId, active);  while (session.activeByTurnId.size > MAX_ACTIVE_TURN_ANALYTICS_PER_SESSION) {    const oldestTurnId = session.activeByTurnId.keys().next().value;    if (oldestTurnId === undefined) return;    session.activeByTurnId.delete(oldestTurnId);  }} function turnAnalyticsSessionKey(instanceId: ProviderInstanceId, threadId: ThreadId): string {  return `${String(instanceId)}\u0000${String(threadId)}`;} function turnAnalyticsCompletionKey(  instanceId: ProviderInstanceId,  threadId: ThreadId,  turnId: string,): string {  return `${turnAnalyticsSessionKey(instanceId, threadId)}\u0000${turnId}`;} function turnEffort(modelSelection: ProviderSendTurnInput["modelSelection"]): string | undefined {  return (    getModelSelectionStringOptionValue(modelSelection, "reasoningEffort") ??    getModelSelectionStringOptionValue(modelSelection, "effort")  );} 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 continueAfterServerUpdate?: TurnId;    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?.continueAfterServerUpdate !== undefined      ? { continueAfterServerUpdate: extra.continueAfterServerUpdate }      : {}),    ...(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 serverConfig = yield* ServerConfig.ServerConfig;  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 serverSettings = yield* ServerSettings.ServerSettingsService;  const issueMcpCredential =    options?.issueMcpCredential ?? McpSessionRegistry.issueActiveMcpCredential;  const revokeMcpCredential =    options?.revokeMcpCredential ?? McpSessionRegistry.revokeActiveMcpThread;  const fileSystem = yield* FileSystem.FileSystem;  const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();  const pendingCompactions = new Map<ThreadId, PendingCompaction>();  const timedOutNativeCompactions = new Set<ThreadId>();  const settleCompaction = (threadId: ThreadId, pending: PendingCompaction, terminal: string) =>    Effect.gen(function* () {      if (pendingCompactions.get(threadId) !== pending) return false;      pendingCompactions.delete(threadId);      yield* Deferred.succeed(pending.completion, terminal);      return true;    });  const turnAnalytics = yield* Ref.make<TurnAnalyticsState>({    sessions: new Map(),    completedKeys: new Set(),    completedOrder: [],  });  let turnAnalyticsRequestId = 0;  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);   const finishTurnAnalytics = (    state: TurnAnalyticsState,    input: {      readonly sessionKey: string;      readonly turnId: string;      readonly completion: DeferredTurnAnalyticsCompletion;    },  ): Readonly<Record<string, unknown>> | undefined => {    if (state.completedKeys.has(input.completion.completionKey)) return undefined;    state.completedKeys.add(input.completion.completionKey);    state.completedOrder.push(input.completion.completionKey);    while (state.completedOrder.length > MAX_COMPLETED_TURN_ANALYTICS_KEYS) {      const expired = state.completedOrder.shift();      if (expired) state.completedKeys.delete(expired);    }     const session = state.sessions.get(input.sessionKey);    const metadata = session?.activeByTurnId.get(input.turnId)?.metadata;    session?.activeByTurnId.delete(input.turnId);    session?.deferredCompletionsByTurnId.delete(input.turnId);    if (      session &&      session.activeByTurnId.size === 0 &&      session.pendingByRequestId.size === 0 &&      session.deferredCompletionsByTurnId.size === 0    ) {      state.sessions.delete(input.sessionKey);    }     return {      ...input.completion.terminalProperties,      ...(metadata?.model ? { model: metadata.model } : {}),      ...(metadata?.effort ? { effort: metadata.effort } : {}),      ...(metadata?.interactionMode ? { interactionMode: metadata.interactionMode } : {}),      ...(metadata?.runtimeMode ? { runtimeMode: metadata.runtimeMode } : {}),      ...(metadata ? { mixedModels: metadata.mixedModels } : {}),      ...(metadata        ? { durationMs: Math.max(0, input.completion.completedAtMs - metadata.startedAtMs) }        : {}),    };  };   const recordCompletedTurnProperties = (    properties: ReadonlyArray<Readonly<Record<string, unknown>>>,  ) =>    Effect.forEach(properties, (entry) => analytics.record("provider.turn.completed", entry), {      discard: true,    });   const clearTurnAnalyticsSession = (providerInstanceId: ProviderInstanceId, threadId: ThreadId) =>    Effect.gen(function* () {      const properties = yield* Ref.modify(turnAnalytics, (state) => {        const sessionKey = turnAnalyticsSessionKey(providerInstanceId, threadId);        const session = state.sessions.get(sessionKey);        const completed: Array<Readonly<Record<string, unknown>>> = [];        if (session) {          for (const [turnId, completion] of session.deferredCompletionsByTurnId) {            const entry = finishTurnAnalytics(state, { sessionKey, turnId, completion });            if (entry) completed.push(entry);          }        }        state.sessions.delete(sessionKey);        return [completed, state] as const;      });      yield* recordCompletedTurnProperties(properties);    });   const beginTurnAnalytics = Effect.fn("beginTurnAnalytics")(function* (input: {    readonly providerInstanceId: ProviderInstanceId;    readonly provider: ProviderDriverKind;    readonly threadId: ThreadId;    readonly modelSelection: ProviderSendTurnInput["modelSelection"];    readonly interactionMode: ProviderSendTurnInput["interactionMode"];    readonly runtimeMode: string | undefined;  }) {    const startedAtMs = DateTime.toEpochMillis(yield* DateTime.now);    turnAnalyticsRequestId += 1;    const requestId = turnAnalyticsRequestId;    const effort = turnEffort(input.modelSelection);    return yield* Ref.modify(turnAnalytics, (state) => {      const key = turnAnalyticsSessionKey(input.providerInstanceId, input.threadId);      const session = state.sessions.get(key) ?? {        pendingByRequestId: new Map(),        activeByTurnId: new Map(),        deferredCompletionsByTurnId: new Map(),      };      const metadata: TurnAnalyticsMetadata = {        provider: input.provider,        startedAtMs,        mixedModels: false,        requestId,        ...(input.modelSelection?.model ? { model: input.modelSelection.model } : {}),        ...(effort ? { effort } : {}),        ...(input.interactionMode ? { interactionMode: input.interactionMode } : {}),        ...(input.runtimeMode ? { runtimeMode: input.runtimeMode } : {}),      };      session.pendingByRequestId.set(requestId, metadata);      state.sessions.set(key, session);      return [metadata, state] as const;    });  });   const clearPendingTurnAnalytics = (input: {    readonly providerInstanceId: ProviderInstanceId;    readonly threadId: ThreadId;    readonly requestId: number;  }) =>    Effect.gen(function* () {      const properties = yield* Ref.modify(turnAnalytics, (state) => {        const sessionKey = turnAnalyticsSessionKey(input.providerInstanceId, input.threadId);        const session = state.sessions.get(sessionKey);        if (!session)          return [[] as ReadonlyArray<Readonly<Record<string, unknown>>>, state] as const;        session.pendingByRequestId.delete(input.requestId);        const completed: Array<Readonly<Record<string, unknown>>> = [];        if (session.pendingByRequestId.size === 0) {          for (const [turnId, completion] of session.deferredCompletionsByTurnId) {            const entry = finishTurnAnalytics(state, { sessionKey, turnId, completion });            if (entry) completed.push(entry);          }        }        if (          session.activeByTurnId.size === 0 &&          session.pendingByRequestId.size === 0 &&          session.deferredCompletionsByTurnId.size === 0        ) {          state.sessions.delete(sessionKey);        }        return [completed, state] as const;      });      yield* recordCompletedTurnProperties(properties);    });   const associateTurnAnalytics = (input: {    readonly providerInstanceId: ProviderInstanceId;    readonly threadId: ThreadId;    readonly turnId: string;    readonly metadata: TurnAnalyticsMetadata;  }) =>    Effect.gen(function* () {      const properties = yield* Ref.modify(turnAnalytics, (state) => {        const completionKey = turnAnalyticsCompletionKey(          input.providerInstanceId,          input.threadId,          input.turnId,        );        const sessionKey = turnAnalyticsSessionKey(input.providerInstanceId, input.threadId);        const session = state.sessions.get(sessionKey);        if (!session || state.completedKeys.has(completionKey)) {          if (session) {            session.pendingByRequestId.delete(input.metadata.requestId);            if (

Lines 1–500 of 2083. Load more