/** * ProviderRegistryLive — aggregates per-instance snapshot streams into a * single materialized list. * * Historically this Layer composed four per-kind Live Layers * (`CodexProviderLive`, `ClaudeProviderLive`, …) that each exposed a * `ServerProviderShape`. Those Lives were deleted during the driver / * instance refactor — every driver now carries its `snapshot: ServerProviderShape` * bundled onto the `ProviderInstance` the registry produces. * * Each configured instance (including multi-instance setups like * `codex_personal` + `codex_work`) contributes one `ProviderSnapshotSource`, * keyed by `instanceId`. Instances whose driver is unavailable or whose * config failed to decode are merged from `instanceRegistry.listUnavailable` * as shadow snapshots so the UI can render their exact unavailable reason. * * Cache paths on disk are now keyed by `instanceId`. Because * `defaultInstanceIdForDriver(kind) === kind` for built-in kinds, existing * `<kind>.json` files remain the on-disk location for that driver's default * instance. Identity-less legacy cache contents are ignored and replaced by * the first live refresh. * * @module ProviderRegistryLive */import {  defaultInstanceIdForDriver,  ProviderDriverKind,  type ProviderInstanceId,  type ServerProvider,  type ServerProviderUpdateState,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Effect from "effect/Effect";import * as Equal from "effect/Equal";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Path from "effect/Path";import * as PubSub from "effect/PubSub";import * as Ref from "effect/Ref";import * as Stream from "effect/Stream";import * as Semaphore from "effect/Semaphore"; import { ServerConfig } from "../../config.ts";import { ProviderInstanceRegistry } from "../Services/ProviderInstanceRegistry.ts";import { ProviderRegistry, type ProviderRegistryShape } from "../Services/ProviderRegistry.ts";import {  hydrateCachedProvider,  isCachedProviderCorrelated,  orderProviderSnapshots,  readProviderStatusCache,  resolveProviderStatusCachePath,  writeProviderStatusCache,} from "../providerStatusCache.ts";import type { ProviderInstance } from "../ProviderDriver.ts";import { makeManualOnlyProviderMaintenanceCapabilities } from "../providerMaintenance.ts";import type { ProviderSnapshotSource } from "../builtInProviderCatalog.ts"; const loadProviders = (  providerSources: ReadonlyArray<ProviderSnapshotSource>,): Effect.Effect<ReadonlyArray<ServerProvider>> =>  Effect.forEach(    providerSources,    (providerSource) =>      providerSource.getSnapshot.pipe(        Effect.flatMap((snapshot) => correlateSnapshotWithSource(providerSource, snapshot)),      ),    {      concurrency: "unbounded",    },  ); const makeManualProviderMaintenanceCapabilities = (provider: ProviderDriverKind) =>  makeManualOnlyProviderMaintenanceCapabilities({    provider,    packageName: null,  }); const hasModelCapabilities = (model: ServerProvider["models"][number]): boolean =>  (model.capabilities?.optionDescriptors?.length ?? 0) > 0; const mergeProviderModels = (  previousModels: ReadonlyArray<ServerProvider["models"][number]>,  nextModels: ReadonlyArray<ServerProvider["models"][number]>,): ReadonlyArray<ServerProvider["models"][number]> => {  if (nextModels.length === 0 && previousModels.length > 0) {    return previousModels;  }   const previousBySlug = new Map(previousModels.map((model) => [model.slug, model] as const));  const mergedModels = nextModels.map((model) => {    const previousModel = previousBySlug.get(model.slug);    if (!previousModel || hasModelCapabilities(model) || !hasModelCapabilities(previousModel)) {      return model;    }    return {      ...model,      capabilities: previousModel.capabilities,    };  });  const nextSlugs = new Set(nextModels.map((model) => model.slug));  return [...mergedModels, ...previousModels.filter((model) => !nextSlugs.has(model.slug))];}; export const mergeProviderSnapshot = (  previousProvider: ServerProvider | undefined,  nextProvider: ServerProvider,): ServerProvider =>  !previousProvider    ? nextProvider    : {        ...nextProvider,        models: mergeProviderModels(previousProvider.models, nextProvider.models),      }; export const mergeProviderSnapshots = (  previousProviders: ReadonlyArray<ServerProvider>,  nextProviders: ReadonlyArray<ServerProvider>,): ReadonlyArray<ServerProvider> => {  const mergedProviders = new Map(    previousProviders.map((provider) => [snapshotInstanceKey(provider), provider] as const),  );   for (const provider of nextProviders) {    mergedProviders.set(      snapshotInstanceKey(provider),      mergeProviderSnapshot(mergedProviders.get(snapshotInstanceKey(provider)), provider),    );  }   return orderProviderSnapshots([...mergedProviders.values()]);}; export const selectProvidersByKind = (  providers: ReadonlyArray<ServerProvider>,  providerKinds: ReadonlySet<ProviderDriverKind>,): ReadonlyArray<ServerProvider> =>  providers.filter((provider) => providerKinds.has(provider.driver)); export const haveProvidersChanged = (  previousProviders: ReadonlyArray<ServerProvider>,  nextProviders: ReadonlyArray<ServerProvider>,): boolean => !Equal.equals(previousProviders, nextProviders); const correlateSnapshotWithSource = (  source: ProviderSnapshotSource,  snapshot: ServerProvider,): Effect.Effect<ServerProvider> => {  if (snapshot.instanceId !== source.instanceId) {    return Effect.die(      new Error(        `Provider snapshot instance mismatch: source '${source.instanceId}' emitted '${snapshot.instanceId}'.`,      ),    );  }  if (snapshot.driver !== source.driverKind) {    return Effect.die(      new Error(        `Provider snapshot driver mismatch for instance '${source.instanceId}': source '${source.driverKind}' emitted '${snapshot.driver}'.`,      ),    );  }  return Effect.succeed(snapshot);}; /** * Key a snapshot for aggregation and persistence. Snapshot sources * must be correlated by instance id before reaching this map; missing * identities are defects, not runtime routing fallbacks. */const snapshotInstanceKey = (provider: ServerProvider): ProviderInstanceId => {  return provider.instanceId;}; // Project a live `ProviderInstance` into the aggregator's consumption// shape. Each call re-captures the instance's `snapshot` closures, so// after `ProviderInstanceRegistry` rebuilds an instance (e.g. because// its settings changed), a fresh source rides the new PubSub instead// of a closed one.const buildSnapshotSource = (instance: ProviderInstance): ProviderSnapshotSource => ({  instanceId: instance.instanceId,  driverKind: instance.driverKind,  getSnapshot: instance.snapshot.getSnapshot,  refresh: instance.snapshot.refresh,  streamChanges: instance.snapshot.streamChanges,}); export const ProviderRegistryLive = Layer.effect(  ProviderRegistry,  Effect.gen(function* () {    const instanceRegistry = yield* ProviderInstanceRegistry;    const config = yield* ServerConfig;    const fileSystem = yield* FileSystem.FileSystem;    const path = yield* Path.Path;     // Aggregator PubSub — consumers (WS gateway, etc.) subscribe here for    // coalesced updates across every instance.    const changesPubSub = yield* Effect.acquireRelease(      PubSub.unbounded<ReadonlyArray<ServerProvider>>(),      PubSub.shutdown,    );     // Boot-only: hydrate `providersRef` from the on-disk per-instance    // cache so the UI has something to render during the first refresh.    // Instances added post-boot skip this path; their first entry in    // `providersRef` comes from the reactive `syncLiveSources` pass    // below.    const bootInstances = yield* instanceRegistry.listInstances;    const bootSources = bootInstances.map(buildSnapshotSource);    const fallbackProviders = yield* loadProviders(bootSources);    const fallbackByInstance = new Map<ProviderInstanceId, ServerProvider>();    for (let index = 0; index < fallbackProviders.length; index++) {      const provider = fallbackProviders[index];      const source = bootSources[index];      if (provider === undefined || source === undefined) {        continue;      }      fallbackByInstance.set(source.instanceId, provider);    }     const cachedProviders = yield* Effect.forEach(      bootSources,      (source) =>        Effect.gen(function* () {          // One cache file per configured instance. For the default          // instance of a built-in kind the path equals `<kind>.json` —          // identical to the legacy filename. We still require the cache          // payload to carry matching instance id + driver kind; old          // identity-less payloads are discarded and the awaited refresh          // below repopulates the cache.          const filePath = yield* resolveProviderStatusCachePath({            cacheDir: config.providerStatusCacheDir,            instanceId: source.instanceId,          }).pipe(Effect.provideService(Path.Path, path));          const fallbackProvider = fallbackByInstance.get(source.instanceId);          if (fallbackProvider === undefined) {            return undefined;          }          return yield* readProviderStatusCache(filePath).pipe(            Effect.provideService(FileSystem.FileSystem, fileSystem),            Effect.flatMap((cachedProvider) => {              if (cachedProvider === undefined) {                return Effect.void.pipe(Effect.as(undefined as ServerProvider | undefined));              }              const correlation = {                cachedProvider,                fallbackProvider,              } as const;              if (!isCachedProviderCorrelated(correlation)) {                return Effect.logWarning("provider status cache identity mismatch, ignoring", {                  path: filePath,                  instanceId: source.instanceId,                  cachedInstanceId: cachedProvider.instanceId ?? null,                  driver: source.driverKind,                  cachedDriver: cachedProvider.driver ?? null,                }).pipe(Effect.as(undefined as ServerProvider | undefined));              }              return Effect.succeed(hydrateCachedProvider(correlation));            }),          );        }),      { concurrency: "unbounded" },    ).pipe(      Effect.map((providers) =>        orderProviderSnapshots(          providers.filter((provider): provider is ServerProvider => provider !== undefined),        ),      ),    );    const providersRef = yield* Ref.make<ReadonlyArray<ServerProvider>>(cachedProviders);    const maintenanceActionStatesRef = yield* Ref.make<      ReadonlyMap<ProviderInstanceId, { readonly update?: ServerProviderUpdateState | undefined }>    >(new Map());     // Live-source registry — the dynamic counterpart to the boot-time    // `bootSources`. Keyed by `instanceId`; the stored `ProviderInstance`    // reference is used for identity equality so "no-op" reconciles    // (settings unchanged) skip re-subscribing + re-probing.    const liveSubsRef = yield* Ref.make<ReadonlyMap<ProviderInstanceId, ProviderInstance>>(      new Map(),    );    // Serialize `syncLiveSources` so a rapid burst of reconciles doesn't    // interleave two passes clobbering each other's fiber bookkeeping.    const syncSemaphore = yield* Semaphore.make(1);     const getLiveSources: Effect.Effect<ReadonlyArray<ProviderSnapshotSource>> = Ref.get(      liveSubsRef,    ).pipe(Effect.map((map) => Array.from(map.values(), buildSnapshotSource)));     const persistProvider = (provider: ServerProvider) =>      Effect.gen(function* () {        // Persist every instance — the file name is the instance id, so        // multi-instance setups (e.g. `codex_personal`, `codex_work`) each        // get their own cache. We resolve the path fresh so snapshots        // produced by newly-added instances post-boot still land on disk        // without the aggregator holding a stale `cachePathByInstance`        // entry.        const key = snapshotInstanceKey(provider);        const filePath = yield* resolveProviderStatusCachePath({          cacheDir: config.providerStatusCacheDir,          instanceId: key,        }).pipe(Effect.provideService(Path.Path, path));        yield* writeProviderStatusCache({ filePath, provider }).pipe(          Effect.provideService(FileSystem.FileSystem, fileSystem),          Effect.provideService(Path.Path, path),          Effect.tapError(Effect.logError),          Effect.ignore,        );      });     const applyProviderUpdateState = Effect.fn("applyProviderUpdateState")(function* (      provider: ServerProvider,    ) {      const maintenanceActionStates = yield* Ref.get(maintenanceActionStatesRef);      const updateState = maintenanceActionStates.get(provider.instanceId)?.update;      if (!updateState) {        const { updateState: _updateState, ...providerWithoutUpdateState } = provider;        return providerWithoutUpdateState;      }      return {        ...provider,        updateState,      };    });     const upsertProviders = Effect.fn("upsertProviders")(function* (      nextProviders: ReadonlyArray<ServerProvider>,      options?: {        readonly publish?: boolean;        readonly persist?: boolean;        readonly replace?: boolean;      },    ) {      const nextProvidersWithUpdateState = yield* Effect.forEach(        nextProviders,        applyProviderUpdateState,        {          concurrency: "unbounded",        },      );      const [previousProviders, providers, providersToPersist] = yield* Ref.modify(        providersRef,        (previousProviders) => {          const mergedProviders = new Map(            previousProviders.map((provider) => [snapshotInstanceKey(provider), provider] as const),          );          const updatedKeys = new Set<ProviderInstanceId>();           for (const provider of nextProvidersWithUpdateState) {            const key = snapshotInstanceKey(provider);            updatedKeys.add(key);            mergedProviders.set(              key,              options?.replace === true                ? provider                : mergeProviderSnapshot(mergedProviders.get(key), provider),            );          }           const providers = orderProviderSnapshots([...mergedProviders.values()]);          const providersToPersist = providers.filter((provider) =>            updatedKeys.has(snapshotInstanceKey(provider)),          );          return [[previousProviders, providers, providersToPersist] as const, providers];        },      );       if (haveProvidersChanged(previousProviders, providers)) {        if (options?.persist !== false) {          yield* Effect.forEach(providersToPersist, persistProvider, {            concurrency: "unbounded",            discard: true,          });        }        if (options?.publish !== false) {          yield* PubSub.publish(changesPubSub, providers);        }      }       return providers;    });     const syncProvider = Effect.fn("syncProvider")(function* (      provider: ServerProvider,      options?: {        readonly publish?: boolean;      },    ) {      return yield* upsertProviders([provider], options);    });     const setProviderMaintenanceActionState = Effect.fn("setProviderMaintenanceActionState")(      function* (input: {        readonly instanceId: ProviderInstanceId;        readonly action: "update";        readonly state: ServerProviderUpdateState | null;      }) {        yield* Ref.update(maintenanceActionStatesRef, (previous) => {          const previousActions = previous.get(input.instanceId);          const nextActions = { ...previousActions };          if (input.state === null || input.state.status === "idle") {            delete nextActions[input.action];          } else {            nextActions[input.action] = input.state;          }           const next = new Map(previous);          if (Object.keys(nextActions).length === 0) {            next.delete(input.instanceId);          } else {            next.set(input.instanceId, nextActions);          }          return next;        });         const existingProviders = yield* Ref.get(providersRef);        const matchingProvider = existingProviders.find(          (candidate) => candidate.instanceId === input.instanceId,        );        if (!matchingProvider) {          return existingProviders;        }         const nextProvider = yield* applyProviderUpdateState(matchingProvider);        return yield* upsertProviders([nextProvider], {          persist: false,        });      },    );     const refreshOneSource = Effect.fn("refreshOneSource")(function* (      providerSource: ProviderSnapshotSource,    ) {      return yield* providerSource.refresh.pipe(        Effect.flatMap((nextProvider) =>          correlateSnapshotWithSource(providerSource, nextProvider).pipe(            Effect.flatMap(syncProvider),          ),        ),      );    });     const refreshAll = Effect.fn("refreshAll")(function* () {      const sources = yield* getLiveSources;      return yield* Effect.forEach(sources, (source) => refreshOneSource(source), {        concurrency: "unbounded",        discard: true,      }).pipe(Effect.andThen(Ref.get(providersRef)));    });     const refresh = Effect.fn("refresh")(function* (provider?: ProviderDriverKind) {      if (provider === undefined) {        return yield* refreshAll();      }      // Kind-scoped refreshes target the default instance for that driver.      const defaultInstanceId = defaultInstanceIdForDriver(provider);      const sources = yield* getLiveSources;      const providerSource = sources.find(        (candidate) => candidate.instanceId === defaultInstanceId,      );      if (!providerSource) {        return yield* Ref.get(providersRef);      }      return yield* refreshOneSource(providerSource);    });     const refreshInstance = Effect.fn("refreshInstance")(function* (      instanceId: ProviderInstanceId,    ) {      const sources = yield* getLiveSources;      const providerSource = sources.find((candidate) => candidate.instanceId === instanceId);      if (!providerSource) {        return yield* Ref.get(providersRef);      }      return yield* refreshOneSource(providerSource);    });     const getProviderMaintenanceCapabilitiesForInstance = Effect.fn(      "getProviderMaintenanceCapabilitiesForInstance",    )(function* (instanceId: ProviderInstanceId, provider: ProviderDriverKind) {      const instance = Array.from((yield* Ref.get(liveSubsRef)).values()).find(        (candidate) => candidate.instanceId === instanceId,      );      return (        instance?.snapshot.maintenanceCapabilities ??        makeManualProviderMaintenanceCapabilities(provider)      );    });     /**     * Diff the aggregator's live-source set against the current     * `ProviderInstanceRegistry` and:     *   - subscribe to each newly-added or rebuilt instance's     *     `streamChanges` (so periodic + enrichment refreshes land in     *     `providersRef`);     *   - read each newly-added/rebuilt instance's current snapshot after     *     subscribing, closing the race with its independently-running     *     background startup probe;     *   - prune `providersRef` of instances that no longer exist.     *     * Provider refreshes are owned by each managed provider and never run

Lines 1–500 of 698. Load more