1import {
2 defaultInstanceIdForDriver,
3 isProviderWorkspaceSnapshotCurrent,
4 isProviderWorkspaceSnapshotPending,
5 ProviderDriverKind,
6 type ProviderInstanceId,
7 type SecretReference,
8 type ServerProvider,
9 type ServerProviderUpdateState,
10} from "@t3tools/contracts";
11import * as Cause from "effect/Cause";
12import * as Context from "effect/Context";
13import * as DateTime from "effect/DateTime";
14import * as Deferred from "effect/Deferred";
15import * as Effect from "effect/Effect";
16import * as Equal from "effect/Equal";
17import * as Fiber from "effect/Fiber";
18import * as FileSystem from "effect/FileSystem";
19import * as Layer from "effect/Layer";
20import * as Option from "effect/Option";
21import * as Path from "effect/Path";
22import * as PubSub from "effect/PubSub";
23import * as Ref from "effect/Ref";
24import * as Scope from "effect/Scope";
25import * as Stream from "effect/Stream";
26import * as Semaphore from "effect/Semaphore";
27
28import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions";
29import * as ModelManifest from "./ModelManifest.ts";
30import { applyProviderCompatibility } from "./providerCompatibility.ts";
31import * as ServerConfig from "../config.ts";
32import * as ProviderInstanceRegistry from "./ProviderInstanceRegistry.ts";
33import * as ProviderInstanceRegistryMutator from "./ProviderInstanceRegistryMutator.ts";
34import {
35 hydrateCachedProvider,
36 isCachedProviderCorrelated,
37 orderProviderSnapshots,
38 readProviderStatusCache,
39 resolveProviderStatusCachePath,
40 writeProviderStatusCache,
41} from "./providerStatusCache.ts";
42import type {
43 ProviderInstance,
44 ProviderWorkspaceSnapshot,
45} from "@t3tools/provider-core/server/driver";
46import {
47 makeManualOnlyProviderMaintenanceCapabilities,
48 type ProviderMaintenanceCapabilities,
49 latestProviderVersionCacheKeys,
50} from "@t3tools/provider-core/server/maintenanceResolver";
51import type { ProviderSnapshotSource } from "./builtInProviderCatalog.ts";
52
53export type ProviderMaintenanceActionKind = "update";
54
55export class ProviderRegistry extends Context.Service<
56 ProviderRegistry,
57 {
58 readonly getProviders: Effect.Effect<ReadonlyArray<ServerProvider>>;
59
60 readonly refresh: (
61 provider?: ProviderDriverKind,
62 ) => Effect.Effect<ReadonlyArray<ServerProvider>>;
63
64 /** Unknown instance ids return the cached list without error. */
65 readonly refreshInstance: (
66 instanceId: ProviderInstanceId,
67 ) => Effect.Effect<ReadonlyArray<ServerProvider>>;
68
69 /** Re-reads secret-backed environments first, so a fixed secret applies without a restart. */
70 readonly reloadSecretsAndRefresh: (
71 instanceId?: ProviderInstanceId,
72 ) => Effect.Effect<ReadonlyArray<ServerProvider>>;
73
74 /** Rebuilds and refreshes only the instances that read a secret, with the value the caller just read. */
75 readonly applySecretAndRefresh: (
76 reference: SecretReference,
77 value: string,
78 ) => Effect.Effect<ReadonlyArray<ServerProvider>>;
79
80 /** Only refreshed instances can verify an update; the published list still includes every row. */
81 readonly refreshAfterUpdate: (
82 provider: ProviderDriverKind,
83 selectedInstanceId: ProviderInstanceId,
84 ) => Effect.Effect<ReadonlyArray<ServerProvider>>;
85
86 readonly refreshWorkspaceSnapshot: (input: {
87 readonly instanceId: ProviderInstanceId;
88 readonly cwd: string;
89 /** Directory the driver scans; the snapshot stays keyed by `cwd`. Defaults to `cwd`. */
90 readonly scanCwd?: string;
91 readonly fresh?: boolean;
92 }) => Effect.Effect<ReadonlyArray<ServerProvider>>;
93
94 /**
95 * Drop T3-owned model caches before an explicit model refresh. Pass the
96 * target of the refresh that follows. Background and ordinary refreshes
97 * keep their cache timers and must not call this.
98 */
99 readonly invalidateModelCaches: (input: {
100 readonly instanceId?: ProviderInstanceId | undefined;
101 readonly cwd?: string | undefined;
102 readonly fresh?: boolean | undefined;
103 }) => Effect.Effect<void>;
104
105 /**
106 * For an opened model picker: refetch a model manifest older than a few
107 * minutes. When it changed, recheck latest CLI versions and probe providers.
108 */
109 readonly refreshModelCatalog: Effect.Effect<void>;
110
111 readonly getProviderMaintenanceCapabilitiesForInstance: (
112 instanceId: ProviderInstanceId,
113 provider: ProviderDriverKind,
114 options?: { readonly fresh?: boolean },
115 ) => Effect.Effect<ProviderMaintenanceCapabilities>;
116
117 readonly setProviderMaintenanceActionState: (input: {
118 readonly instanceId: ProviderInstanceId;
119 readonly action: ProviderMaintenanceActionKind;
120 readonly state: ServerProviderUpdateState | null;
121 readonly targetVersion?: string | undefined;
122 }) => Effect.Effect<ReadonlyArray<ServerProvider>>;
123
124 readonly streamChanges: Stream.Stream<ReadonlyArray<ServerProvider>>;
125 }
126>()("t3/provider/ProviderRegistry") {}
127
128const loadProviders = (
129 providerSources: ReadonlyArray<ProviderSnapshotSource>,
130): Effect.Effect<ReadonlyArray<ServerProvider>> =>
131 Effect.forEach(
132 providerSources,
133 (providerSource) =>
134 providerSource.getSnapshot.pipe(
135 Effect.flatMap((snapshot) => correlateSnapshotWithSource(providerSource, snapshot)),
136 ),
137 {
138 concurrency: "unbounded",
139 },
140 );
141
142const makeManualProviderMaintenanceCapabilities = (provider: ProviderDriverKind) =>
143 makeManualOnlyProviderMaintenanceCapabilities({
144 provider,
145 packageName: null,
146 });
147
148const hasModelCapabilities = (model: ServerProvider["models"][number]): boolean =>
149 (model.capabilities?.optionDescriptors?.length ?? 0) > 0;
150
151const MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER = 16;
152
153function dropProviderWorkspaceSnapshot(provider: ServerProvider, cwd: string): ServerProvider {
154 return provider.workspaceSnapshots?.some((snapshot) => snapshot.cwd === cwd)
155 ? {
156 ...provider,
157 workspaceSnapshots: provider.workspaceSnapshots.filter((snapshot) => snapshot.cwd !== cwd),
158 }
159 : provider;
160}
161
162export function upsertProviderWorkspaceSnapshot(
163 provider: ServerProvider,
164 cwd: string,
165 scopedSnapshot: ProviderWorkspaceSnapshot,
166): ServerProvider {
167 const previousSnapshot = provider.workspaceSnapshots?.find((snapshot) => snapshot.cwd === cwd);
168 const optionDescriptors = scopedSnapshot.optionDescriptors ?? previousSnapshot?.optionDescriptors;
169 const models = scopedSnapshot.workspaceModelsPending
170 ? previousSnapshot?.models
171 : scopedSnapshot.workspaceModels;
172 const workspaceSnapshot = {
173 cwd,
174 checkedAt: scopedSnapshot.checkedAt,
175 slashCommands: scopedSnapshot.slashCommandsPending
176 ? (previousSnapshot?.slashCommands ?? scopedSnapshot.slashCommands)
177 : scopedSnapshot.slashCommands,
178 ...(scopedSnapshot.slashCommandsPending ? { slashCommandsPending: true } : {}),
179 skills: scopedSnapshot.skills,
180 ...(optionDescriptors === undefined ? {} : { optionDescriptors }),
181 ...(models === undefined ? {} : { models }),
182 ...(scopedSnapshot.workspaceModelsPending ? { modelsPending: true } : {}),
183 } satisfies NonNullable<ServerProvider["workspaceSnapshots"]>[number];
184 return {
185 ...provider,
186 workspaceSnapshots: [
187 ...(provider.workspaceSnapshots ?? []).filter((snapshot) => snapshot.cwd !== cwd),
188 workspaceSnapshot,
189 ].slice(-MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER),
190 };
191}
192
193const shouldRetainMissingProviderModels = (provider: ServerProvider): boolean => {
194 if (provider.driver === ProviderDriverKind.make("pi")) {
195 // Pi auth becomes known only after inventory discovery: empty means unauthenticated; failed/interactive discovery preserves unknown auth and cached models.
196 return !(
197 provider.installed &&
198 (provider.status === "ready" || provider.status === "warning") &&
199 provider.auth.status !== "unknown"
200 );
201 }
202
203 if (provider.driver === ProviderDriverKind.make("acpRegistry")) {
204 // Only completed ACP discovery provides nativeSessions and inventory; placeholder probes cannot establish auth for agents without sign-in.
205 return !(
206 provider.installed &&
207 provider.status === "ready" &&
208 provider.nativeSessions !== undefined
209 );
210 }
211
212 if (
213 provider.driver === ProviderDriverKind.make("grok") ||
214 provider.driver === ProviderDriverKind.make("cursor")
215 ) {
216 // Grok and Cursor report ready only after they discover a model list; the pending,
217 // failed, empty-inventory, and logged-out probes are warning/error.
218 return !(provider.installed && provider.status === "ready");
219 }
220
221 const isAntigravity = provider.driver === ProviderDriverKind.make("antigravity");
222 const isCodex = provider.driver === ProviderDriverKind.make("codex");
223 if (!isAntigravity && !isCodex && provider.driver !== ProviderDriverKind.make("opencode")) {
224 return true;
225 }
226
227 if (
228 (isAntigravity || isCodex) &&
229 (!provider.enabled || provider.auth.status === "unauthenticated")
230 ) {
231 return false;
232 }
233
234 // Antigravity's local health check does not authenticate or discover models.
235 const isPendingAntigravityAuthentication =
236 isAntigravity && provider.status === "warning" && provider.auth.status === "unknown";
237 const isPendingInitialProbe =
238 provider.enabled && !provider.installed && provider.status === "warning";
239 const isIncompleteInstalledProbe =
240 provider.installed &&
241 (provider.status === "error" || (isCodex && provider.status === "warning"));
242 return isPendingAntigravityAuthentication || isPendingInitialProbe || isIncompleteInstalledProbe;
243};
244
245const shouldRetainMissingOpenCodeMetadata = (provider: ServerProvider): boolean =>
246 provider.driver === ProviderDriverKind.make("opencode") &&
247 shouldRetainMissingProviderModels(provider);
248
249const mergeProviderModels = (
250 provider: ServerProvider,
251 previousModels: ReadonlyArray<ServerProvider["models"][number]>,
252 nextModels: ReadonlyArray<ServerProvider["models"][number]>,
253): ReadonlyArray<ServerProvider["models"][number]> => {
254 const shouldRetainMissingModels = shouldRetainMissingProviderModels(provider);
255 const updateRequiredSlugs = new Set(provider.updateRequiredModels?.map((model) => model.slug));
256 const retainablePreviousModels = previousModels.filter(
257 (model) => !model.isCustom && !updateRequiredSlugs.has(model.slug),
258 );
259
260 if (shouldRetainMissingModels && nextModels.length === 0 && retainablePreviousModels.length > 0) {
261 return retainablePreviousModels;
262 }
263
264 const previousBySlug = new Map(previousModels.map((model) => [model.slug, model] as const));
265 const mergedModels = nextModels.map((model) => {
266 const previousModel = previousBySlug.get(model.slug);
267 if (!previousModel || hasModelCapabilities(model) || !hasModelCapabilities(previousModel)) {
268 return model;
269 }
270 return {
271 ...model,
272 capabilities: previousModel.capabilities,
273 };
274 });
275 const nextSlugs = new Set(nextModels.map((model) => model.slug));
276 return shouldRetainMissingModels
277 ? [...mergedModels, ...retainablePreviousModels.filter((model) => !nextSlugs.has(model.slug))]
278 : mergedModels;
279};
280
281/** Antigravity health checks only initialize; preserve previously verified Google auth across restarts. */
282const carrySavedAntigravityAccount = (
283 previousProvider: ServerProvider,
284 nextProvider: ServerProvider,
285): Pick<ServerProvider, "auth" | "status"> | undefined => {
286 const antigravity = ProviderDriverKind.make("antigravity");
287 if (
288 nextProvider.driver !== antigravity ||
289 previousProvider.driver !== antigravity ||
290 !nextProvider.enabled ||
291 nextProvider.auth.status !== "unknown" ||
292 previousProvider.auth.status !== "authenticated" ||
293 (nextProvider.auth.type !== undefined &&
294 nextProvider.auth.type !== previousProvider.auth.type) ||
295 (!nextProvider.installed && nextProvider.status !== "warning")
296 ) {
297 return undefined;
298 }
299 const status =
300 nextProvider.installed && nextProvider.status === "warning" ? "ready" : nextProvider.status;
301 return { auth: previousProvider.auth, status };
302};
303
304export const mergeProviderSnapshot = (
305 previousProvider: ServerProvider | undefined,
306 nextProvider: ServerProvider,
307): ServerProvider => {
308 if (!previousProvider) {
309 return nextProvider;
310 }
311 const savedAccount = carrySavedAntigravityAccount(previousProvider, nextProvider);
312 const { message: _uncheckedMessage, ...nextWithoutMessage } = nextProvider;
313 const models = mergeProviderModels(nextProvider, previousProvider.models, nextProvider.models);
314 // Folder catalogs were compared with the old machine catalog, so every folder rescans.
315 const previousWorkspaceSnapshots = Equal.equals(models, previousProvider.models)
316 ? previousProvider.workspaceSnapshots
317 : previousProvider.workspaceSnapshots?.map((snapshot) => ({
318 ...snapshot,
319 modelsPending: true,
320 }));
321 return {
322 ...(savedAccount?.status === "ready" ? nextWithoutMessage : nextProvider),
323 ...savedAccount,
324 models,
325 ...(nextProvider.workspaceSnapshots !== undefined
326 ? { workspaceSnapshots: nextProvider.workspaceSnapshots }
327 : previousWorkspaceSnapshots !== undefined
328 ? { workspaceSnapshots: previousWorkspaceSnapshots }
329 : {}),
330 ...(shouldRetainMissingOpenCodeMetadata(nextProvider)
331 ? {
332 slashCommands:
333 nextProvider.slashCommands.length === 0
334 ? previousProvider.slashCommands
335 : nextProvider.slashCommands,
336 skills: nextProvider.skills.length === 0 ? previousProvider.skills : nextProvider.skills,
337 }
338 : {}),
339 };
340};
341
342const haveProvidersChanged = (
343 previousProviders: ReadonlyArray<ServerProvider>,
344 nextProviders: ReadonlyArray<ServerProvider>,
345): boolean => !Equal.equals(previousProviders, nextProviders);
346
347const correlateSnapshotWithSource = (
348 source: ProviderSnapshotSource,
349 snapshot: ServerProvider,
350): Effect.Effect<ServerProvider> => {
351 if (snapshot.instanceId !== source.instanceId) {
352 return Effect.die(
353 new Error(
354 `Provider snapshot instance mismatch: source '${source.instanceId}' emitted '${snapshot.instanceId}'.`,
355 ),
356 );
357 }
358 if (snapshot.driver !== source.driverKind) {
359 return Effect.die(
360 new Error(
361 `Provider snapshot driver mismatch for instance '${source.instanceId}': source '${source.driverKind}' emitted '${snapshot.driver}'.`,
362 ),
363 );
364 }
365 return Effect.succeed(snapshot);
366};
367
368const snapshotInstanceKey = (provider: ServerProvider): ProviderInstanceId => {
369 return provider.instanceId;
370};
371
372// Rebuilt instances have new PubSubs; capture their fresh closures instead of a closed source.
373const buildSnapshotSource = (instance: ProviderInstance): ProviderSnapshotSource => ({
374 instanceId: instance.instanceId,
375 driverKind: instance.driverKind,
376 getSnapshot: instance.snapshot.getSnapshot,
377 refresh: instance.snapshot.refresh,
378 streamChanges: instance.snapshot.streamChanges,
379});
380
381export const layer = Layer.effect(
382 ProviderRegistry,
383 Effect.gen(function* () {
384 const instanceRegistry = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
385 const instanceRegistryMutator =
386 yield* ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator;
387 const manifestService = yield* ModelManifest.ModelManifest;
388 const latestVersions = yield* ProviderLatestVersions.ProviderLatestVersions;
389 const serviceScope = yield* Effect.scope;
390 const config = yield* ServerConfig.ServerConfig;
391 const fileSystem = yield* FileSystem.FileSystem;
392 const path = yield* Path.Path;
393
394 const changesPubSub = yield* Effect.acquireRelease(
395 PubSub.unbounded<ReadonlyArray<ServerProvider>>(),
396 PubSub.shutdown,
397 );
398
399 const bootInstances = yield* instanceRegistry.listInstances;
400 const bootSources = bootInstances.map(buildSnapshotSource);
401 const fallbackProviders = yield* loadProviders(bootSources);
402 const fallbackByInstance = new Map<ProviderInstanceId, ServerProvider>();
403 for (let index = 0; index < fallbackProviders.length; index++) {
404 const provider = fallbackProviders[index];
405 const source = bootSources[index];
406 if (provider === undefined || source === undefined) {
407 continue;
408 }
409 fallbackByInstance.set(source.instanceId, provider);
410 }
411 // Read after the snapshots: a snapshot read as awaiting is the pre-check placeholder.
412 const awaitingBootInstances = new Set<ProviderInstance>();
413 for (const instance of bootInstances) {
414 if (yield* instance.snapshot.isAwaitingFirstCheck ?? Effect.succeed(false)) {
415 awaitingBootInstances.add(instance);
416 }
417 }
418
419 const bootProviders = yield* Effect.forEach(
420 bootSources,
421 (source, index) =>
422 Effect.gen(function* () {
423 const filePath = yield* resolveProviderStatusCachePath({
424 cacheDir: config.providerStatusCacheDir,
425 instanceId: source.instanceId,
426 }).pipe(Effect.provideService(Path.Path, path));
427 const fallbackProvider = fallbackByInstance.get(source.instanceId);
428 if (fallbackProvider === undefined) {
429 return undefined;
430 }
431 // A completed check is seeded by the boot upsert instead, which persists it.
432 const placeholder = awaitingBootInstances.has(bootInstances[index]!)
433 ? fallbackProvider
434 : undefined;
435 return yield* readProviderStatusCache(filePath).pipe(
436 Effect.provideService(FileSystem.FileSystem, fileSystem),
437 Effect.flatMap((cachedProvider) => {
438 if (cachedProvider === undefined) {
439 return Effect.succeed(placeholder);
440 }
441 const correlation = {
442 cachedProvider,
443 fallbackProvider,
444 } as const;
445 if (!isCachedProviderCorrelated(correlation)) {
446 return Effect.logWarning("provider status cache identity mismatch, ignoring", {
447 path: filePath,
448 instanceId: source.instanceId,
449 cachedInstanceId: cachedProvider.instanceId ?? null,
450 driver: source.driverKind,
451 cachedDriver: cachedProvider.driver ?? null,
452 }).pipe(Effect.as(placeholder));
453 }
454 return Effect.succeed(hydrateCachedProvider(correlation));
455 }),
456 );
457 }),
458 { concurrency: "unbounded" },
459 ).pipe(
460 Effect.map((providers) =>
461 orderProviderSnapshots(
462 providers.filter((provider): provider is ServerProvider => provider !== undefined),
463 ),
464 ),
465 );
466 const initialManifest = yield* manifestService.current;
467 const classifyCompatibility = (
468 provider: ServerProvider,
469 manifest: ModelManifest.ModelManifestData,
470 ) =>
471 applyProviderCompatibility(
472 provider,
473 manifest.compatibility,
474 ModelManifest.BUNDLED_MODEL_MANIFEST.compatibility,
475 );
476 const providersRef = yield* Ref.make<ReadonlyArray<ServerProvider>>(
477 bootProviders.map((provider) => classifyCompatibility(provider, initialManifest)),
478 );
479 const workspaceRefreshesRef = yield* Ref.make<
480 ReadonlyMap<ProviderInstance, ReadonlyMap<string, Deferred.Deferred<void>>>
481 >(new Map());
482 const maintenanceActionStatesRef = yield* Ref.make<
483 ReadonlyMap<
484 ProviderInstanceId,
485 { readonly update?: ServerProviderUpdateState; readonly targetVersion?: string | undefined }
486 >
487 >(new Map());
488
489 const liveSubsRef = yield* Ref.make<ReadonlyMap<ProviderInstanceId, ProviderInstance>>(
490 new Map(),
491 );
492 const syncSemaphore = yield* Semaphore.make(1);
493
494 const persistProvider = Effect.fn("ProviderRegistry.persistProvider")(function* (
495 provider: ServerProvider,
496 ) {
497 const key = snapshotInstanceKey(provider);
498 const filePath = yield* resolveProviderStatusCachePath({
499 cacheDir: config.providerStatusCacheDir,
500 instanceId: key,
501 }).pipe(Effect.provideService(Path.Path, path));
502 const { workspaceSnapshots: _workspaceSnapshots, ...machineProvider } = provider;
503 yield* writeProviderStatusCache({ filePath, provider: machineProvider }).pipe(
504 Effect.provideService(FileSystem.FileSystem, fileSystem),
505 Effect.provideService(Path.Path, path),
506 Effect.tapError(Effect.logError),
507 Effect.ignore,
508 );
509 });
510
511 const applyProviderUpdateState = Effect.fn("applyProviderUpdateState")(function* (
512 provider: ServerProvider,
513 ) {
514 const maintenanceActionStates = yield* Ref.get(maintenanceActionStatesRef);
515 const updateState = maintenanceActionStates.get(provider.instanceId)?.update;
516 if (!updateState) {
517 const { updateState: _updateState, ...providerWithoutUpdateState } = provider;
518 return providerWithoutUpdateState;
519 }
520 return {
521 ...provider,
522 updateState,
523 };
524 });
525
526 const upsertProviders = Effect.fn("upsertProviders")(function* (
527 nextProviders: ReadonlyArray<ServerProvider>,
528 options?: {
529 readonly publish?: boolean;
530 readonly persist?: boolean;
531 readonly replace?: boolean;
532 },
533 ) {
534 const manifest = yield* manifestService.current;
535 const nextProvidersWithUpdateState = yield* Effect.forEach(
536 nextProviders,
537 applyProviderUpdateState,
538 {
539 concurrency: "unbounded",
540 },
541 );
542 const [previousProviders, providers, providersToPersist] = yield* Ref.modify(
543 providersRef,
544 (previousProviders) => {
545 const mergedProviders = new Map(
546 previousProviders.map((provider) => [snapshotInstanceKey(provider), provider] as const),
547 );
548 const updatedKeys = new Set<ProviderInstanceId>();
549
550 for (const provider of nextProvidersWithUpdateState) {
551 const key = snapshotInstanceKey(provider);
552 updatedKeys.add(key);
553 mergedProviders.set(
554 key,
555 options?.replace === true
556 ? provider
557 : mergeProviderSnapshot(mergedProviders.get(key), provider),
558 );
559 }
560
561 const providers = orderProviderSnapshots(
562 [...mergedProviders.values()].map((provider) =>
563 classifyCompatibility(provider, manifest),
564 ),
565 );
566 const providersToPersist = providers.filter((provider) =>
567 updatedKeys.has(snapshotInstanceKey(provider)),
568 );
569 return [[previousProviders, providers, providersToPersist] as const, providers];
570 },
571 );
572
573 if (haveProvidersChanged(previousProviders, providers)) {
574 if (options?.persist !== false) {
575 yield* Effect.forEach(providersToPersist, persistProvider, {
576 concurrency: "unbounded",
577 discard: true,
578 });
579 }
580 if (options?.publish !== false) {
581 // A newer upsert may have committed and published during the cache write; never republish an older array.
582 yield* PubSub.publish(changesPubSub, yield* Ref.get(providersRef));
583 }
584 }
585
586 return providers;
587 });
588
589 const compatibilityRefreshRunning = yield* Ref.make(false);
590 // Reclassify the current model after fetch; republishing the old probe could overwrite newer health results.
591 const refreshCompatibility = Effect.gen(function* () {
592 if (!(yield* Ref.getAndSet(compatibilityRefreshRunning, true))) {
593 yield* manifestService.refresh.pipe(
594 Effect.andThen(upsertProviders([], { persist: false })),
595 Effect.ensuring(Ref.set(compatibilityRefreshRunning, false)),
596 Effect.forkIn(serviceScope),
597 );
598 }
599 });
600 const syncProvider = Effect.fn("syncProvider")(function* (
601 provider: ServerProvider,
602 options?: {
603 readonly publish?: boolean;
604 },
605 ) {
606 // Latest-version status cannot resolve a failed recommended-version install.
607 if (provider.versionAdvisory?.status === "current") {
608 yield* Ref.update(maintenanceActionStatesRef, (previous) => {
609 const action = previous.get(provider.instanceId);
610 if (
611 action?.targetVersion !== undefined ||
612 (action?.update?.status !== "unchanged" && action?.update?.status !== "failed")
613 ) {
614 return previous;
615 }
616 const next = new Map(previous);
617 next.delete(provider.instanceId);
618 return next;
619 });
620 }
621 const providers = yield* upsertProviders([provider], options);
622 yield* refreshCompatibility;
623 return providers;
624 });
625
626 const setProviderMaintenanceActionState = Effect.fn("setProviderMaintenanceActionState")(
627 function* (input: {
628 readonly instanceId: ProviderInstanceId;
629 readonly action: "update";
630 readonly state: ServerProviderUpdateState | null;
631 readonly targetVersion?: string | undefined;
632 }) {
633 yield* Ref.update(maintenanceActionStatesRef, (previous) => {
634 const next = new Map(previous);
635 if (input.state === null || input.state.status === "idle") {
636 next.delete(input.instanceId);
637 } else {
638 next.set(input.instanceId, { update: input.state, targetVersion: input.targetVersion });
639 }
640 return next;
641 });
642
643 const existingProviders = yield* Ref.get(providersRef);
644 const matchingProvider = existingProviders.find(
645 (candidate) => candidate.instanceId === input.instanceId,
646 );
647 if (!matchingProvider) {
648 return existingProviders;
649 }
650
651 const nextProvider = yield* applyProviderUpdateState(matchingProvider);
652 return yield* upsertProviders([nextProvider], {
653 persist: false,
654 });
655 },
656 );
657
658 const isCurrentInstance = (instance: ProviderInstance) =>
659 instanceRegistry
660 .getInstance(instance.instanceId)
661 .pipe(Effect.map((current) => current === instance));
662
663 const refreshOneSource = Effect.fn("refreshOneSource")(function* (instance: ProviderInstance) {
664 const providerSource = buildSnapshotSource(instance);
665 // Enrichment can publish a newer snapshot before refresh returns its pre-enrichment probe.
666 const probe = yield* providerSource.refresh.pipe(
667 Effect.andThen(providerSource.getSnapshot),
668 Effect.exit,
669 );
670 // Settings can replace or remove the instance mid-probe; its late result or failure is stale.
671 if (!(yield* isCurrentInstance(instance))) {
672 return yield* Ref.get(providersRef);
673 }
674 const nextProvider = yield* probe;
675 return yield* correlateSnapshotWithSource(providerSource, nextProvider).pipe(
676 Effect.flatMap(syncProvider),
677 );
678 });
679
680 type RefreshAllCompletion = Deferred.Deferred<ReadonlyArray<ServerProvider>>;
681 const refreshAllInFlightRef = yield* Ref.make<Option.Option<RefreshAllCompletion>>(
682 Option.none(),
683 );
684
685 // What the last full pass could see; a later manifest fetch does not reach providers by itself.
686 const probedManifestUpdatedAt = yield* Ref.make(initialManifest.updatedAt);
687
688 const refreshAll = Effect.fn("refreshAll")(function* () {
689 const claimed = yield* Ref.modify(
690 refreshAllInFlightRef,
691 (
692 inFlight,
693 ): readonly [
694 { readonly owner: boolean; readonly completion: RefreshAllCompletion },
695 Option.Option<RefreshAllCompletion>,
696 ] => {
697 if (Option.isSome(inFlight)) {
698 return [{ owner: false, completion: inFlight.value }, inFlight];
699 }
700 const completion = Deferred.makeUnsafe<ReadonlyArray<ServerProvider>>();
701 return [{ owner: true, completion }, Option.some(completion)];
702 },
703 );
704 if (claimed.owner) {
705 // Detach shared passes so the first caller's interruption cannot cancel other callers' work.
706 // Syncing first keeps a refresh right after a rebuild from probing the replaced instance.
707 yield* manifestService.current.pipe(
708 Effect.flatMap(({ updatedAt }) =>
709 syncLiveSourcesAndContinue.pipe(
710 Effect.andThen(Ref.get(liveSubsRef)),
711 Effect.flatMap((instances) =>
712 Effect.forEach(instances.values(), (instance) => refreshOneSource(instance), {
713 concurrency: "unbounded",
714 discard: true,
715 }),
716 ),
717 Effect.andThen(Ref.get(providersRef)),
718 Effect.tap(() => Ref.set(probedManifestUpdatedAt, updatedAt)),
719 ),
720 ),
721 Effect.onExit((exit) =>
722 Ref.set(refreshAllInFlightRef, Option.none()).pipe(
723 Effect.andThen(Deferred.done(claimed.completion, exit)),
724 ),
725 ),
726 Effect.forkDetach,
727 );
728 }
729 return yield* Deferred.await(claimed.completion);
730 });
731
732 const refresh = Effect.fn("refresh")(function* (provider?: ProviderDriverKind) {
733 if (provider === undefined) {
734 return yield* refreshAll();
735 }
736 const instance = (yield* Ref.get(liveSubsRef)).get(defaultInstanceIdForDriver(provider));
737 if (!instance) {
738 return yield* Ref.get(providersRef);
739 }
740 return yield* refreshOneSource(instance);
741 });
742
743 const refreshInstance = Effect.fn("refreshInstance")(function* (
744 instanceId: ProviderInstanceId,
745 ) {
746 yield* syncLiveSourcesAndContinue;
747 const instance = (yield* Ref.get(liveSubsRef)).get(instanceId);
748 if (!instance) {
749 return yield* Ref.get(providersRef);
750 }
751 return yield* refreshOneSource(instance);
752 });
753
754 const reloadSecretsAndRefresh = Effect.fn("reloadSecretsAndRefresh")(function* (
755 instanceId?: ProviderInstanceId,
756 ) {
757 yield* instanceRegistryMutator.reloadSecrets(instanceId);
758 return yield* instanceId === undefined ? refresh() : refreshInstance(instanceId);
759 });
760
761 const applySecretAndRefresh = Effect.fn("applySecretAndRefresh")(function* (
762 reference: SecretReference,
763 value: string,
764 ) {
765 const instanceIds = yield* instanceRegistryMutator.applySecret(reference, value);
766 yield* Effect.forEach(instanceIds, refreshInstance, { discard: true });
767 return yield* Ref.get(providersRef);
768 });
769
770 const refreshAfterUpdate = Effect.fn("refreshAfterUpdate")(function* (
771 provider: ProviderDriverKind,
772 selectedInstanceId: ProviderInstanceId,
773 ) {
774 yield* manifestService.forceRefresh;
775 const instances = (yield* instanceRegistry.listInstances).filter(
776 (instance) =>
777 instance.driverKind === provider &&
778 (instance.enabled || instance.instanceId === selectedInstanceId),
779 );
780 const failedInstances = new Set<ProviderInstanceId>();
781 const recoverInstanceFailure = Effect.fn("recoverPostUpdateRefreshFailure")(function* (
782 instanceId: ProviderInstanceId,
783 cause: Cause.Cause<unknown>,
784 ) {
785 if (Cause.hasInterruptsOnly(cause)) {
786 return yield* Effect.interrupt;
787 }
788 yield* Effect.logError("provider post-update refresh failed", {
789 instanceId,
790 cause: Cause.pretty(cause),
791 });
792 failedInstances.add(instanceId);
793 // A cached ready snapshot cannot verify a successful update.
794 return yield* updateProviders((providers) =>
795 providers.map((candidate) =>
796 candidate.instanceId === instanceId
797 ? { ...candidate, status: "error", message: "Provider refresh failed after update." }
798 : candidate,
799 ),
800 );
801 });
802 // Instances can share the replaced service, so none may invalidate before it restarts.
803 const selectedInstance = instances.find(
804 (instance) => instance.instanceId === selectedInstanceId,
805 );
806 if (selectedInstance?.afterUpdate !== undefined) {
807 // The probe below reports what is running, so a failed hook needs no status of its own.
808 yield* selectedInstance.afterUpdate.pipe(
809 Effect.catchCause((cause) =>
810 Cause.hasInterruptsOnly(cause)
811 ? Effect.interrupt
812 : Effect.logError("provider post-update hook failed", {
813 instanceId: selectedInstance.instanceId,
814 cause: Cause.pretty(cause),
815 }),
816 ),
817 );
818 }
819 // Accounts can share an installation, but discovery must use each
820 // instance's own executable, environment, and credentials.
821 yield* Effect.forEach(
822 instances,
823 (instance) =>
824 Effect.gen(function* () {
825 yield* instance.invalidateCaches ?? Effect.void;
826 const maintenance = yield* instance.snapshot.resolveMaintenance({ fresh: true });
827 yield* Effect.forEach(
828 latestProviderVersionCacheKeys(maintenance),
829 latestVersions.invalidate,
830 { discard: true },
831 );
832 }).pipe(Effect.catchCause((cause) => recoverInstanceFailure(instance.instanceId, cause))),
833 { concurrency: "unbounded", discard: true },
834 );
835 yield* Effect.forEach(
836 instances.filter((instance) => !failedInstances.has(instance.instanceId)),
837 (instance) =>
838 refreshOneSource(instance).pipe(
839 Effect.catchCause((cause) => recoverInstanceFailure(instance.instanceId, cause)),
840 ),
841 { concurrency: "unbounded", discard: true },
842 );
843 // A replaced instance's cached row did not come from this pass's probe.
844 const refreshedInstances = yield* Effect.filter(
845 instances.filter((instance) => !failedInstances.has(instance.instanceId)),
846 isCurrentInstance,
847 );
848 const refreshedInstanceIds = new Set(
849 refreshedInstances.map((instance) => instance.instanceId),
850 );
851 return (yield* Ref.get(providersRef)).filter((candidate) =>
852 refreshedInstanceIds.has(candidate.instanceId),
853 );
854 });
855
856 const getProviderMaintenanceCapabilitiesForInstance = Effect.fn(
857 "getProviderMaintenanceCapabilitiesForInstance",
858 )(function* (
859 instanceId: ProviderInstanceId,
860 provider: ProviderDriverKind,
861 options?: { readonly fresh?: boolean },
862 ) {
863 // liveSubsRef trails reconciliation and could run a retired instance's command; read the registry.
864 const instance = yield* instanceRegistry.getInstance(instanceId);
865 if (!instance || instance.driverKind !== provider) {
866 return makeManualProviderMaintenanceCapabilities(provider);
867 }
868 return yield* instance.snapshot.resolveMaintenance(options);
869 });
870
871 const syncLiveSources = syncSemaphore.withPermits(1)(
872 Effect.gen(function* () {
873 const instances = yield* instanceRegistry.listInstances;
874 const unavailableProviders = yield* instanceRegistry.listUnavailable;
875 const nextByInstance = new Map<ProviderInstanceId, ProviderInstance>(
876 instances.map((instance) => [instance.instanceId, instance] as const),
877 );
878 const knownInstanceIds = new Set<ProviderInstanceId>(nextByInstance.keys());
879 for (const provider of unavailableProviders) {
880 knownInstanceIds.add(snapshotInstanceKey(provider));
881 }
882 const previousSubs = yield* Ref.get(liveSubsRef);
883
884 const carriedOver = new Map<ProviderInstanceId, ProviderInstance>();
885 for (const [instanceId, previousInstance] of previousSubs) {
886 const nextInstance = nextByInstance.get(instanceId);
887 if (nextInstance !== undefined && nextInstance === previousInstance) {
888 carriedOver.set(instanceId, previousInstance);
889 }
890 }
891
892 const newlyAdded: Array<readonly [ProviderInstanceId, ProviderInstance]> = [];
893 for (const [instanceId, instance] of nextByInstance) {
894 if (carriedOver.has(instanceId)) {
895 continue;
896 }
897 newlyAdded.push([instanceId, instance] as const);
898 }
899
900 const rebuiltInstanceIds = new Set(
901 newlyAdded
902 .map(([instanceId]) => instanceId)
903 .filter((instanceId) => previousSubs.has(instanceId)),
904 );
905 if (rebuiltInstanceIds.size > 0) {
906 const [previousProviders, providers] = yield* Ref.modify(
907 providersRef,
908 (previousProviders) => {
909 const providers = previousProviders.map((provider) => {
910 if (!rebuiltInstanceIds.has(provider.instanceId)) return provider;
911 const { workspaceSnapshots: _workspaceSnapshots, ...machineSnapshot } = provider;
912 return machineSnapshot;
913 });
914 return [[previousProviders, providers] as const, providers];
915 },
916 );
917 if (haveProvidersChanged(previousProviders, providers)) {
918 yield* PubSub.publish(changesPubSub, providers);
919 }
920 }
921
922 // Subscribe before reading so a probe completing mid-sync is observed by at least one path.
923 for (const [, instance] of newlyAdded) {
924 const source = buildSnapshotSource(instance);
925 yield* Stream.runForEach(source.streamChanges, (provider) =>
926 correlateSnapshotWithSource(source, provider).pipe(Effect.flatMap(syncProvider)),
927 ).pipe(Effect.forkScoped);
928 }
929 yield* Effect.yieldNow;
930
931 yield* Effect.forEach(
932 newlyAdded,
933 ([, instance]) =>
934 Effect.gen(function* () {
935 const source = buildSnapshotSource(instance);
936 const provider = yield* source.getSnapshot;
937 // Boot shows the cached status until this instance's first check publishes a result.
938 if (
939 awaitingBootInstances.has(instance) &&
940 (yield* instance.snapshot.isAwaitingFirstCheck ?? Effect.succeed(false))
941 ) {
942 return yield* refreshCompatibility;
943 }
944 yield* correlateSnapshotWithSource(source, provider).pipe(
945 Effect.flatMap(syncProvider),
946 );
947 }).pipe(Effect.ignoreCause({ log: true })),
948 { concurrency: "unbounded", discard: true },
949 );
950 yield* upsertProviders(unavailableProviders, {
951 persist: false,
952 replace: true,
953 });
954
955 const nextSubs = new Map(carriedOver);
956 for (const [instanceId, instance] of newlyAdded) {
957 nextSubs.set(instanceId, instance);
958 }
959 yield* Ref.set(liveSubsRef, nextSubs);
960
961 const [previousProviders, providers] = yield* Ref.modify(
962 providersRef,
963 (previousProviders) => {
964 const providers = orderProviderSnapshots(
965 previousProviders.filter((provider) =>
966 knownInstanceIds.has(snapshotInstanceKey(provider)),
967 ),
968 );
969 return [[previousProviders, providers] as const, providers];
970 },
971 );
972 if (haveProvidersChanged(previousProviders, providers)) {
973 yield* PubSub.publish(changesPubSub, providers);
974 }
975 yield* Ref.update(maintenanceActionStatesRef, (previous) => {
976 const next = new Map(previous);
977 for (const instanceId of previous.keys()) {
978 if (!knownInstanceIds.has(instanceId)) {
979 next.delete(instanceId);
980 }
981 }
982 return next;
983 });
984 }),
985 );
986 const syncLiveSourcesAndContinue = syncLiveSources.pipe(
987 Effect.catchCause((cause) => {
988 if (Cause.hasInterruptsOnly(cause)) {
989 return Effect.interrupt;
990 }
991 return Effect.logError(
992 "provider registry instance sync failed; keeping subscription alive",
993 {
994 cause: Cause.pretty(cause),
995 },
996 );
997 }),
998 Scope.provide(serviceScope),
999 );
1000
1001 yield* upsertProviders(
1002 fallbackProviders.filter((_, index) => !awaitingBootInstances.has(bootInstances[index]!)),
1003 { publish: false },
1004 );
1005 const instanceChanges = yield* instanceRegistry.subscribeChanges;
1006 yield* syncLiveSources;
1007 yield* Stream.runForEach(
1008 Stream.fromSubscription(instanceChanges),
1009 () => syncLiveSourcesAndContinue,
1010 ).pipe(Effect.forkScoped);
1011
1012 const recoverRefreshFailure = Effect.fn("recoverRefreshFailure")(function* (
1013 cause: Cause.Cause<unknown>,
1014 ) {
1015 if (Cause.hasInterruptsOnly(cause)) {
1016 return yield* Effect.interrupt;
1017 }
1018 yield* Effect.logError("provider registry refresh failed; preserving cached providers", {
1019 cause: Cause.pretty(cause),
1020 });
1021 return yield* Ref.get(providersRef);
1022 });
1023
1024 const updateProviders = (
1025 update: (providers: ReadonlyArray<ServerProvider>) => ReadonlyArray<ServerProvider>,
1026 ) =>
1027 Ref.modify(providersRef, (currentProviders) => {
1028 const nextProviders = update(currentProviders);
1029 return [[currentProviders, nextProviders] as const, nextProviders];
1030 }).pipe(
1031 Effect.tap(([previousProviders, nextProviders]) =>
1032 haveProvidersChanged(previousProviders, nextProviders)
1033 ? PubSub.publish(changesPubSub, nextProviders)
1034 : Effect.void,
1035 ),
1036 Effect.map(([, nextProviders]) => nextProviders),
1037 );
1038
1039 const refreshWorkspaceSnapshot = Effect.fn("refreshWorkspaceSnapshot")(function* (input: {
1040 readonly instanceId: ProviderInstanceId;
1041 readonly cwd: string;
1042 readonly scanCwd?: string;
1043 readonly fresh?: boolean;
1044 }) {
1045 if (input.fresh) {
1046 yield* updateProviders((providers) =>
1047 providers.map((candidate) =>
1048 candidate.instanceId === input.instanceId
1049 ? candidate
1050 : dropProviderWorkspaceSnapshot(candidate, input.cwd),
1051 ),
1052 );
1053 }
1054 const providers = yield* Ref.get(providersRef);
1055 const provider = providers.find((candidate) => candidate.instanceId === input.instanceId);
1056 const workspaceSnapshotOf = (candidate: ServerProvider | undefined) =>
1057 candidate?.workspaceSnapshots?.find((s) => s.cwd === input.cwd);
1058 const scannedFrom = workspaceSnapshotOf(provider);
1059 const now = yield* DateTime.now;
1060 if (
1061 !provider ||
1062 !provider.enabled ||
1063 (!input.fresh &&
1064 scannedFrom &&
1065 !isProviderWorkspaceSnapshotPending(scannedFrom) &&
1066 isProviderWorkspaceSnapshotCurrent(scannedFrom, DateTime.toEpochMillis(now)))
1067 ) {
1068 return providers;
1069 }
1070 // Use scan start for TTL, not checkedAt inherited from the machine health check.
1071 const scannedAt = DateTime.formatIso(now);
1072 const instance = yield* instanceRegistry.getInstance(input.instanceId);
1073 const snapshotForCwd = instance?.snapshotForCwd;
1074 if (!instance || !snapshotForCwd) return providers;
1075 const token = yield* Deferred.make<void>();
1076 // A machine refresh can change this cwd's snapshot, so the scan publishes against the result.
1077 const refreshMachineSnapshot = input.fresh
1078 ? (instance.invalidateCaches ?? Effect.void).pipe(
1079 Effect.andThen(refreshInstance(input.instanceId)),
1080 Effect.andThen(Ref.get(providersRef)),
1081 Effect.map((current) =>
1082 workspaceSnapshotOf(
1083 current.find((candidate) => candidate.instanceId === input.instanceId),
1084 ),
1085 ),
1086 )
1087 : Effect.succeed(scannedFrom);
1088 const runScan = refreshMachineSnapshot.pipe(
1089 Effect.flatMap((scanBase) =>
1090 Effect.map(snapshotForCwd(input.scanCwd ?? input.cwd), (scopedSnapshot) => ({
1091 scanBase,
1092 scopedSnapshot,
1093 })),
1094 ),
1095 // Scans fail on error; returned inventory remains valid even when provider health reports an error.
1096 Effect.flatMap(({ scanBase, scopedSnapshot }) =>
1097 instanceRegistry.getInstance(input.instanceId).pipe(
1098 Effect.flatMap((currentInstance) => {
1099 if (currentInstance !== instance) return Ref.get(providersRef);
1100 return updateProviders((currentProviders) => {
1101 // Read ownership inside the update so a newer claim cannot race publication.
1102 if (Ref.getUnsafe(workspaceRefreshesRef).get(instance)?.get(input.cwd) !== token) {
1103 return currentProviders;
1104 }
1105 return currentProviders.map((candidate) =>
1106 candidate.instanceId === input.instanceId &&
1107 Equal.equals(workspaceSnapshotOf(candidate), scanBase)
1108 ? upsertProviderWorkspaceSnapshot(candidate, input.cwd, {
1109 ...scopedSnapshot,
1110 checkedAt: scannedAt,
1111 // Without driver command discovery, retain previous commands after failed health checks and retry.
1112 slashCommandsPending:
1113 scopedSnapshot.slashCommandsPending ?? scopedSnapshot.status === "error",
1114 })
1115 : candidate,
1116 );
1117 });
1118 }),
1119 ),
1120 ),
1121 Effect.ensuring(
1122 Ref.update(workspaceRefreshesRef, (refreshes) => {
1123 const current = refreshes.get(instance);
1124 if (current?.get(input.cwd) !== token) return refreshes;
1125 const next = new Map(refreshes);
1126 const nextCwds = new Map(current);
1127 nextCwds.delete(input.cwd);
1128 if (nextCwds.size) next.set(instance, nextCwds);
1129 else next.delete(instance);
1130 return next;
1131 }).pipe(Effect.andThen(Deferred.succeed(token, undefined))),
1132 ),
1133 );
1134 // The claim and the fork happen together, so the token is always completed. The scan is
1135 // detached so a disconnecting owner cannot cancel the result its waiters depend on.
1136 const claim = yield* Effect.uninterruptible(
1137 Effect.gen(function* () {
1138 const inFlight = yield* Ref.modify(workspaceRefreshesRef, (refreshes) => {
1139 const current = refreshes.get(instance);
1140 const running = current?.get(input.cwd);
1141 if (!input.fresh && running) return [running, refreshes] as const;
1142 const next = new Map(refreshes);
1143 next.set(instance, new Map(current).set(input.cwd, token));
1144 return [undefined, next] as const;
1145 });
1146 if (inFlight) return { inFlight } as const;
1147 return {
1148 // Starting now attaches the scan's `ensuring` before anything can close the scope.
1149 scan: yield* Effect.forkIn(runScan, serviceScope, { startImmediately: true }),
1150 } as const;
1151 }),
1152 );
1153 // Wait for the running scan: on-demand clients get its result only in this reply.
1154 if ("inFlight" in claim) {
1155 return yield* Deferred.await(claim.inFlight).pipe(Effect.andThen(Ref.get(providersRef)));
1156 }
1157 return yield* Fiber.join(claim.scan);
1158 });
1159
1160 const invalidateModelCaches = Effect.fn("ProviderRegistry.invalidateModelCaches")(function* (
1161 input: Parameters<ProviderRegistry["Service"]["invalidateModelCaches"]>[0],
1162 ) {
1163 yield* manifestService.forceRefresh;
1164 const instances = yield* instanceRegistry.listInstances;
1165 yield* Effect.forEach(
1166 instances.filter(
1167 (instance) => input.instanceId === undefined || input.instanceId === instance.instanceId,
1168 ),
1169 (instance) =>
1170 Effect.gen(function* () {
1171 yield* instance.invalidateCaches ?? Effect.void;
1172 // Instance and full refreshes resolve installer metadata fresh
1173 // themselves; a cached workspace refresh needs it resolved here.
1174 const maintenance = yield* instance.snapshot.resolveMaintenance({
1175 fresh:
1176 input.instanceId !== undefined && input.cwd !== undefined && input.fresh !== true,
1177 });
1178 yield* Effect.forEach(
1179 latestProviderVersionCacheKeys(maintenance),
1180 latestVersions.invalidate,
1181 { discard: true },
1182 );
1183 }),
1184 { concurrency: "unbounded", discard: true },
1185 );
1186 });
1187
1188 const refreshModelCatalog = Effect.fn("ProviderRegistry.refreshModelCatalog")(function* () {
1189 const { updatedAt } = yield* manifestService.refreshOnDemand;
1190 // The first pass can be one that started before the fetch, so allow one more.
1191 for (let pass = 0; pass < 2; pass++) {
1192 if ((yield* Ref.get(probedManifestUpdatedAt)) === updatedAt) return;
1193 // A new model usually ships with the CLI release that supports it.
1194 yield* latestVersions.invalidateAll;
1195 yield* refreshAll();
1196 }
1197 });
1198
1199 return {
1200 getProviders: Ref.get(providersRef),
1201 refresh: (provider?: ProviderDriverKind) =>
1202 refresh(provider).pipe(Effect.catchCause(recoverRefreshFailure)),
1203 refreshInstance: (instanceId: ProviderInstanceId) =>
1204 refreshInstance(instanceId).pipe(Effect.catchCause(recoverRefreshFailure)),
1205 reloadSecretsAndRefresh: (instanceId?: ProviderInstanceId) =>
1206 reloadSecretsAndRefresh(instanceId).pipe(Effect.catchCause(recoverRefreshFailure)),
1207 applySecretAndRefresh: (reference: SecretReference, value: string) =>
1208 applySecretAndRefresh(reference, value).pipe(Effect.catchCause(recoverRefreshFailure)),
1209 refreshWorkspaceSnapshot: (input) =>
1210 refreshWorkspaceSnapshot(input).pipe(Effect.catchCause(recoverRefreshFailure)),
1211 refreshAfterUpdate,
1212 invalidateModelCaches,
1213 refreshModelCatalog: refreshModelCatalog().pipe(
1214 Effect.catchCause(recoverRefreshFailure),
1215 Effect.asVoid,
1216 ),
1217 getProviderMaintenanceCapabilitiesForInstance,
1218 setProviderMaintenanceActionState,
1219 get streamChanges() {
1220 return Stream.fromPubSub(changesPubSub);
1221 },
1222 } satisfies ProviderRegistry["Service"];
1223 }),
1224);