/** * 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 MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER = 16; export function upsertProviderWorkspaceSnapshot(  provider: ServerProvider,  cwd: string,  scopedSnapshot: ServerProvider,): ServerProvider {  const workspaceSnapshot = {    cwd,    checkedAt: scopedSnapshot.checkedAt,    slashCommands: scopedSnapshot.slashCommands,    skills: scopedSnapshot.skills,  } satisfies NonNullable<ServerProvider["workspaceSnapshots"]>[number];  return {    ...provider,    workspaceSnapshots: [      ...(provider.workspaceSnapshots ?? []).filter((snapshot) => snapshot.cwd !== cwd),      workspaceSnapshot,    ].slice(-MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER),  };} const shouldRetainMissingProviderModels = (provider: ServerProvider): boolean => {  const isAntigravity = provider.driver === ProviderDriverKind.make("antigravity");  const isCodex = provider.driver === ProviderDriverKind.make("codex");  if (!isAntigravity && !isCodex && provider.driver !== ProviderDriverKind.make("opencode")) {    return true;  }   if (    (isAntigravity || isCodex) &&    (!provider.enabled || provider.auth.status === "unauthenticated")  ) {    return false;  }   // Successful discovery replaces these inventories so cached retired models disappear.  // Antigravity's local health check does not authenticate or discover models.  const isPendingAntigravityAuthentication =    isAntigravity && provider.status === "warning" && provider.auth.status === "unknown";  const isPendingInitialProbe =    provider.enabled && !provider.installed && provider.status === "warning";  const didInstalledProviderProbeFail = provider.installed && provider.status === "error";  return (    isPendingAntigravityAuthentication || isPendingInitialProbe || didInstalledProviderProbeFail  );}; const shouldRetainMissingOpenCodeMetadata = (provider: ServerProvider): boolean =>  provider.driver === ProviderDriverKind.make("opencode") &&  shouldRetainMissingProviderModels(provider); const mergeProviderModels = (  provider: ServerProvider,  previousModels: ReadonlyArray<ServerProvider["models"][number]>,  nextModels: ReadonlyArray<ServerProvider["models"][number]>,): ReadonlyArray<ServerProvider["models"][number]> => {  const shouldRetainMissingModels = shouldRetainMissingProviderModels(provider);  // Custom rows are derived from settings and every snapshot carries the full  // current list, so a custom model missing from `nextModels` was removed by  // the user and must not be resurrected from the previous snapshot.  const retainablePreviousModels = previousModels.filter((model) => !model.isCustom);   if (shouldRetainMissingModels && nextModels.length === 0 && retainablePreviousModels.length > 0) {    return retainablePreviousModels;  }   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 shouldRetainMissingModels    ? [...mergedModels, ...retainablePreviousModels.filter((model) => !nextSlugs.has(model.slug))]    : mergedModels;}; /** * Antigravity's health check only initializes the agent, so after a server * restart it reports the account as unchecked. The saved Google login still * works, and the previous snapshot proves it. Carry that account state until * a session, refresh, or sign-out reports something new. A confirmed missing * installation, sign-out, disabled instance, or a changed sign-in method is * never overridden. */const carrySavedAntigravityAccount = (  previousProvider: ServerProvider,  nextProvider: ServerProvider,): Pick<ServerProvider, "auth" | "status"> | undefined => {  const antigravity = ProviderDriverKind.make("antigravity");  if (    nextProvider.driver !== antigravity ||    previousProvider.driver !== antigravity ||    !nextProvider.enabled ||    nextProvider.auth.status !== "unknown" ||    previousProvider.auth.status !== "authenticated" ||    (nextProvider.auth.type !== undefined &&      nextProvider.auth.type !== previousProvider.auth.type) ||    (!nextProvider.installed && nextProvider.status !== "warning")  ) {    return undefined;  }  // The pending boot probe (`installed: false`, warning) and a failed probe  // keep their own status; only a passed health check reads as ready.  const status =    nextProvider.installed && nextProvider.status === "warning" ? "ready" : nextProvider.status;  return { auth: previousProvider.auth, status };}; export const mergeProviderSnapshot = (  previousProvider: ServerProvider | undefined,  nextProvider: ServerProvider,): ServerProvider => {  if (!previousProvider) {    return nextProvider;  }  const savedAccount = carrySavedAntigravityAccount(previousProvider, nextProvider);  // "Google account access is not checked yet" describes the probe, not the  // account; it must not outlive the state it explained.  const { message: _uncheckedMessage, ...nextWithoutMessage } = nextProvider;  return {    ...(savedAccount?.status === "ready" ? nextWithoutMessage : nextProvider),    ...savedAccount,    models: mergeProviderModels(nextProvider, previousProvider.models, nextProvider.models),    ...(nextProvider.workspaceSnapshots !== undefined      ? { workspaceSnapshots: nextProvider.workspaceSnapshots }      : previousProvider.workspaceSnapshots !== undefined        ? { workspaceSnapshots: previousProvider.workspaceSnapshots }        : {}),    ...(shouldRetainMissingOpenCodeMetadata(nextProvider)      ? {          slashCommands:            nextProvider.slashCommands.length === 0              ? previousProvider.slashCommands              : nextProvider.slashCommands,          skills: nextProvider.skills.length === 0 ? previousProvider.skills : nextProvider.skills,        }      : {}),  };}; 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 workspaceRefreshesRef = yield* Ref.make<      ReadonlyMap<ProviderInstance, ReadonlySet<string>>    >(new Map());    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));        const { workspaceSnapshots: _workspaceSnapshots, ...machineProvider } = provider;        yield* writeProviderStatusCache({ filePath, provider: machineProvider }).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);

Lines 1–500 of 883. Load more