/** * 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 { Cause, Effect, Equal, FileSystem, Layer, Path, PubSub, Ref, Stream } from "effect";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`); * - force-refresh each newly-added/rebuilt instance and feed the * result directly into `providersRef`, bypassing the PubSub * attachment race that otherwise drops the initial probe; * - prune `providersRef` of instances that no longer exist. * * Initial refreshes are awaited in parallel rather than forked, so * callers (layer build; `streamChanges` watcher) see fully-probed * state on return. This matters for layer build in particular: * consumers reading `getProviders` immediately after layer build * expect the probe to have already landed. * * Per-instance subscription fibers are not tracked explicitly. When * a rebuilt instance's old child scope closes, its PubSub shuts * down and our `Stream.runForEach` fiber exits naturally.Lines 1–500 of 687. Load more
/** * 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 { Cause, Effect, Equal, FileSystem, Layer, Path, PubSub, Ref, Stream } from "effect"; 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`); * - force-refresh each newly-added/rebuilt instance and feed the * result directly into `providersRef`, bypassing the PubSub * attachment race that otherwise drops the initial probe; * - prune `providersRef` of instances that no longer exist. * * Initial refreshes are awaited in parallel rather than forked, so * callers (layer build; `streamChanges` watcher) see fully-probed * state on return. This matters for layer build in particular: * consumers reading `getProviders` immediately after layer build * expect the probe to have already landed. * * Per-instance subscription fibers are not tracked explicitly. When * a rebuilt instance's old child scope closes, its PubSub shuts * down and our `Stream.runForEach` fiber exits naturally.