clover's git

the-knife

1	/**
2	* ProviderRegistry - Provider snapshot service.
3	*
4	* Owns provider install/auth/version/model snapshots and exposes the latest
5	* provider state to transport layers.
6	*
7	* The live layer aggregates per-instance snapshot streams into a
8	* single materialized list.
9	*
10	* Historically this Layer composed four per-kind Live Layers
11	* (`CodexProviderLive`, `ClaudeProviderLive`, …) that each exposed a
12	* `ManagedServerProvider`. Those Lives were deleted during the driver /
13	* instance refactor — every driver now carries its `snapshot: ManagedServerProvider`
14	* bundled onto the `ProviderInstance` the registry produces.
15	*
16	* Each configured instance (including multi-instance setups like
17	* `codex_personal` + `codex_work`) contributes one `ProviderSnapshotSource`,
18	* keyed by `instanceId`. Instances whose driver is unavailable or whose
19	* config failed to decode are merged from `instanceRegistry.listUnavailable`
20	* as shadow snapshots so the UI can render their exact unavailable reason.
21	*
22	* Cache paths on disk are now keyed by `instanceId`. Because
23	* `defaultInstanceIdForDriver(kind) === kind` for built-in kinds, existing
24	* `<kind>.json` files remain the on-disk location for that driver's default
25	* instance. Identity-less legacy cache contents are ignored and replaced by
26	* the first live refresh.
27	*
28	* @module ProviderRegistry
29	*/
30	import {
31	defaultInstanceIdForDriver,
32	isProviderWorkspaceSnapshotCurrent,
33	ProviderDriverKind,
34	type ProviderInstanceId,
35	type ServerProvider,
36	type ServerProviderUpdateState,
37	} from "@t3tools/contracts";
38	import * as Cause from "effect/Cause";
39	import * as Context from "effect/Context";
40	import * as DateTime from "effect/DateTime";
41	import * as Deferred from "effect/Deferred";
42	import * as Effect from "effect/Effect";
43	import * as Equal from "effect/Equal";
44	import * as FileSystem from "effect/FileSystem";
45	import * as Layer from "effect/Layer";
46	import * as Option from "effect/Option";
47	import * as Path from "effect/Path";
48	import * as PubSub from "effect/PubSub";
49	import * as Ref from "effect/Ref";
50	import * as Stream from "effect/Stream";
51	import * as Semaphore from "effect/Semaphore";
52
53	import * as ModelManifest from "./ModelManifest.ts";
54	import { applyProviderCompatibility } from "./providerCompatibility.ts";
55	import * as ServerConfig from "../config.ts";
56	import * as ProviderInstanceRegistry from "./ProviderInstanceRegistry.ts";
57	import {
58	hydrateCachedProvider,
59	isCachedProviderCorrelated,
60	orderProviderSnapshots,
61	readProviderStatusCache,
62	resolveProviderStatusCachePath,
63	writeProviderStatusCache,
64	} from "./providerStatusCache.ts";
65	import type {
66	ProviderInstance,
67	ProviderWorkspaceSnapshot,
68	} from "@t3tools/provider-core/server/driver";
69	import {
70	makeManualOnlyProviderMaintenanceCapabilities,
71	type ProviderMaintenanceCapabilities,
72	} from "@t3tools/provider-core/server/maintenanceResolver";
73	import type { ProviderSnapshotSource } from "./builtInProviderCatalog.ts";
74
75	export type ProviderMaintenanceActionKind = "update";
76
77	export class ProviderRegistry extends Context.Service<
78	ProviderRegistry,
79	{
80	/**
81	* Read the latest provider snapshots for every configured instance.
82	* Multiple snapshots may share the same `provider` kind (multiple
83	* instances of the same driver) and disambiguate via `instanceId`.
84	*/
85	readonly getProviders: Effect.Effect<ReadonlyArray<ServerProvider>>;
86
87	/**
88	* Refresh all providers, or the default instance of the specified
89	* kind when supplied.
90	*
91	* Retained for back-compat with legacy call sites (WS refresh RPC,
92	* orchestration metrics). New code should prefer `refreshInstance`.
93	*
94	* @deprecated prefer `refreshInstance` for new call sites.
95	*/
96	readonly refresh: (
97	provider?: ProviderDriverKind,
98	) => Effect.Effect<ReadonlyArray<ServerProvider>>;
99
100	/**
101	* Refresh the specific configured instance. Returns the updated snapshot
102	* list. When the instance id is unknown the call resolves with the
103	* currently cached list (no error) — matching the legacy `refresh` shim
104	* behaviour so transport layers don't have to special-case unknowns.
105	*/
106	readonly refreshInstance: (
107	instanceId: ProviderInstanceId,
108	) => Effect.Effect<ReadonlyArray<ServerProvider>>;
109
110	/**
111	* Fill the skills and slash commands snapshot for one cwd. A cwd that
112	* already has a snapshot is left alone unless `fresh` is set. A fresh scan
113	* also refreshes the instance's machine snapshot and drops other
114	* instances' snapshots for the cwd.
115	*/
116	readonly refreshWorkspaceSnapshot: (input: {
117	readonly instanceId: ProviderInstanceId;
118	readonly cwd: string;
119	readonly fresh?: boolean;
120	}) => Effect.Effect<ReadonlyArray<ServerProvider>>;
121
122	/**
123	* Resolve the maintenance capabilities owned by one live provider instance.
124	* Falls back to manual-only capabilities when the instance is not live.
125	* `fresh` re-derives ownership from the executable instead of the cache.
126	*/
127	readonly getProviderMaintenanceCapabilitiesForInstance: (
128	instanceId: ProviderInstanceId,
129	provider: ProviderDriverKind,
130	options?: { readonly fresh?: boolean },
131	) => Effect.Effect<ProviderMaintenanceCapabilities>;
132
133	/**
134	* Apply volatile maintenance-action state to one configured instance.
135	* This state is never persisted to disk. Today only update actions are
136	* projected onto `ServerProvider.updateState`; install/auth actions can
137	* extend this action map without adding driver-scoped APIs.
138	*/
139	readonly setProviderMaintenanceActionState: (input: {
140	readonly instanceId: ProviderInstanceId;
141	readonly action: ProviderMaintenanceActionKind;
142	readonly state: ServerProviderUpdateState | null;
143	}) => Effect.Effect<ReadonlyArray<ServerProvider>>;
144
145	/**
146	* Stream of provider snapshot updates — one emission per aggregated
147	* change. The array contains the full current state.
148	*/
149	readonly streamChanges: Stream.Stream<ReadonlyArray<ServerProvider>>;
150	}
151	>()("t3/provider/ProviderRegistry") {}
152
153	const loadProviders = (
154	providerSources: ReadonlyArray<ProviderSnapshotSource>,
155	): Effect.Effect<ReadonlyArray<ServerProvider>> =>
156	Effect.forEach(
157	providerSources,
158	(providerSource) =>
159	providerSource.getSnapshot.pipe(
160	Effect.flatMap((snapshot) => correlateSnapshotWithSource(providerSource, snapshot)),
161	),
162	{
163	concurrency: "unbounded",
164	},
165	);
166
167	const makeManualProviderMaintenanceCapabilities = (provider: ProviderDriverKind) =>
168	makeManualOnlyProviderMaintenanceCapabilities({
169	provider,
170	packageName: null,
171	});
172
173	const hasModelCapabilities = (model: ServerProvider["models"][number]): boolean =>
174	(model.capabilities?.optionDescriptors?.length ?? 0) > 0;
175
176	const MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER = 16;
177
178	function dropProviderWorkspaceSnapshot(provider: ServerProvider, cwd: string): ServerProvider {
179	return provider.workspaceSnapshots?.some((snapshot) => snapshot.cwd === cwd)
180	? {
181	...provider,
182	workspaceSnapshots: provider.workspaceSnapshots.filter((snapshot) => snapshot.cwd !== cwd),
183	}
184	: provider;
185	}
186
187	export function upsertProviderWorkspaceSnapshot(
188	provider: ServerProvider,
189	cwd: string,
190	scopedSnapshot: ProviderWorkspaceSnapshot,
191	): ServerProvider {
192	const workspaceSnapshot = {
193	cwd,
194	checkedAt: scopedSnapshot.checkedAt,
195	slashCommands: scopedSnapshot.slashCommandsPending
196	? (provider.workspaceSnapshots?.find((snapshot) => snapshot.cwd === cwd)?.slashCommands ??
197	scopedSnapshot.slashCommands)
198	: scopedSnapshot.slashCommands,
199	...(scopedSnapshot.slashCommandsPending ? { slashCommandsPending: true } : {}),
200	skills: scopedSnapshot.skills,
201	} satisfies NonNullable<ServerProvider["workspaceSnapshots"]>[number];
202	return {
203	...provider,
204	workspaceSnapshots: [
205	...(provider.workspaceSnapshots ?? []).filter((snapshot) => snapshot.cwd !== cwd),
206	workspaceSnapshot,
207	].slice(-MAX_WORKSPACE_SNAPSHOTS_PER_PROVIDER),
208	};
209	}
210
211	const shouldRetainMissingProviderModels = (provider: ServerProvider): boolean => {
212	if (provider.driver === ProviderDriverKind.make("acpRegistry")) {
213	// ACP Registry discovery probes return the agent's complete inventory, so
214	// a completed probe (ready and authenticated) replaces the model list —
215	// otherwise agents that rename or collapse models leave stale entries
216	// pinned forever through the snapshot cache. Readiness-only and failed
217	// probe snapshots only know the "default" placeholder and stay partial.
218	return !(
219	provider.installed &&
220	provider.status === "ready" &&
221	provider.auth.status === "authenticated"
222	);
223	}
224
225	const isAntigravity = provider.driver === ProviderDriverKind.make("antigravity");
226	const isCodex = provider.driver === ProviderDriverKind.make("codex");
227	if (!isAntigravity && !isCodex && provider.driver !== ProviderDriverKind.make("opencode")) {
228	return true;
229	}
230
231	if (
232	(isAntigravity || isCodex) &&
233	(!provider.enabled || provider.auth.status === "unauthenticated")
234	) {
235	return false;
236	}
237
238	// Successful discovery replaces these inventories so cached retired models disappear.
239	// Antigravity's local health check does not authenticate or discover models.
240	const isPendingAntigravityAuthentication =
241	isAntigravity && provider.status === "warning" && provider.auth.status === "unknown";
242	const isPendingInitialProbe =
243	provider.enabled && !provider.installed && provider.status === "warning";
244	const didInstalledProviderProbeFail = provider.installed && provider.status === "error";
245	return (
246	isPendingAntigravityAuthentication || isPendingInitialProbe || didInstalledProviderProbeFail
247	);
248	};
249
250	const shouldRetainMissingOpenCodeMetadata = (provider: ServerProvider): boolean =>
251	provider.driver === ProviderDriverKind.make("opencode") &&
252	shouldRetainMissingProviderModels(provider);
253
254	const mergeProviderModels = (
255	provider: ServerProvider,
256	previousModels: ReadonlyArray<ServerProvider["models"][number]>,
257	nextModels: ReadonlyArray<ServerProvider["models"][number]>,
258	): ReadonlyArray<ServerProvider["models"][number]> => {
259	const shouldRetainMissingModels = shouldRetainMissingProviderModels(provider);
260	// Custom rows are derived from settings and every snapshot carries the full
261	// current list, so a custom model missing from `nextModels` was removed by
262	// the user and must not be resurrected from the previous snapshot.
263	// A model the installed CLI is too old to run was offered by the pending
264	// snapshot, before the version was known; retaining it would make it
265	// selectable again.
266	const updateRequiredSlugs = new Set(provider.updateRequiredModels?.map((model) => model.slug));
267	const retainablePreviousModels = previousModels.filter(
268	(model) => !model.isCustom && !updateRequiredSlugs.has(model.slug),
269	);
270
271	if (shouldRetainMissingModels && nextModels.length === 0 && retainablePreviousModels.length > 0) {
272	return retainablePreviousModels;
273	}
274
275	const previousBySlug = new Map(previousModels.map((model) => [model.slug, model] as const));
276	const mergedModels = nextModels.map((model) => {
277	const previousModel = previousBySlug.get(model.slug);
278	if (!previousModel || hasModelCapabilities(model) || !hasModelCapabilities(previousModel)) {
279	return model;
280	}
281	return {
282	...model,
283	capabilities: previousModel.capabilities,
284	};
285	});
286	const nextSlugs = new Set(nextModels.map((model) => model.slug));
287	return shouldRetainMissingModels
288	? [...mergedModels, ...retainablePreviousModels.filter((model) => !nextSlugs.has(model.slug))]
289	: mergedModels;
290	};
291
292	/**
293	* Antigravity's health check only initializes the agent, so after a server
294	* restart it reports the account as unchecked. The saved Google login still
295	* works, and the previous snapshot proves it. Carry that account state until
296	* a session, refresh, or sign-out reports something new. A confirmed missing
297	* installation, sign-out, disabled instance, or a changed sign-in method is
298	* never overridden.
299	*/
300	const carrySavedAntigravityAccount = (
301	previousProvider: ServerProvider,
302	nextProvider: ServerProvider,
303	): Pick<ServerProvider, "auth" | "status"> | undefined => {
304	const antigravity = ProviderDriverKind.make("antigravity");
305	if (
306	nextProvider.driver !== antigravity ||
307	previousProvider.driver !== antigravity ||
308	!nextProvider.enabled ||
309	nextProvider.auth.status !== "unknown" ||
310	previousProvider.auth.status !== "authenticated" ||
311	(nextProvider.auth.type !== undefined &&
312	nextProvider.auth.type !== previousProvider.auth.type) ||
313	(!nextProvider.installed && nextProvider.status !== "warning")
314	) {
315	return undefined;
316	}
317	// The pending boot probe (`installed: false`, warning) and a failed probe
318	// keep their own status; only a passed health check reads as ready.
319	const status =
320	nextProvider.installed && nextProvider.status === "warning" ? "ready" : nextProvider.status;
321	return { auth: previousProvider.auth, status };
322	};
323
324	export const mergeProviderSnapshot = (
325	previousProvider: ServerProvider | undefined,
326	nextProvider: ServerProvider,
327	): ServerProvider => {
328	if (!previousProvider) {
329	return nextProvider;
330	}
331	const savedAccount = carrySavedAntigravityAccount(previousProvider, nextProvider);
332	// "Google account access is not checked yet" describes the probe, not the
333	// account; it must not outlive the state it explained.
334	const { message: _uncheckedMessage, ...nextWithoutMessage } = nextProvider;
335	return {
336	...(savedAccount?.status === "ready" ? nextWithoutMessage : nextProvider),
337	...savedAccount,
338	models: mergeProviderModels(nextProvider, previousProvider.models, nextProvider.models),
339	...(nextProvider.workspaceSnapshots !== undefined
340	? { workspaceSnapshots: nextProvider.workspaceSnapshots }
341	: previousProvider.workspaceSnapshots !== undefined
342	? { workspaceSnapshots: previousProvider.workspaceSnapshots }
343	: {}),
344	...(shouldRetainMissingOpenCodeMetadata(nextProvider)
345	? {
346	slashCommands:
347	nextProvider.slashCommands.length === 0
348	? previousProvider.slashCommands
349	: nextProvider.slashCommands,
350	skills: nextProvider.skills.length === 0 ? previousProvider.skills : nextProvider.skills,
351	}
352	: {}),
353	};
354	};
355
356	export const mergeProviderSnapshots = (
357	previousProviders: ReadonlyArray<ServerProvider>,
358	nextProviders: ReadonlyArray<ServerProvider>,
359	): ReadonlyArray<ServerProvider> => {
360	const mergedProviders = new Map(
361	previousProviders.map((provider) => [snapshotInstanceKey(provider), provider] as const),
362	);
363
364	for (const provider of nextProviders) {
365	mergedProviders.set(
366	snapshotInstanceKey(provider),
367	mergeProviderSnapshot(mergedProviders.get(snapshotInstanceKey(provider)), provider),
368	);
369	}
370
371	return orderProviderSnapshots([...mergedProviders.values()]);
372	};
373
374	export const selectProvidersByKind = (
375	providers: ReadonlyArray<ServerProvider>,
376	providerKinds: ReadonlySet<ProviderDriverKind>,
377	): ReadonlyArray<ServerProvider> =>
378	providers.filter((provider) => providerKinds.has(provider.driver));
379
380	const haveProvidersChanged = (
381	previousProviders: ReadonlyArray<ServerProvider>,
382	nextProviders: ReadonlyArray<ServerProvider>,
383	): boolean => !Equal.equals(previousProviders, nextProviders);
384
385	const correlateSnapshotWithSource = (
386	source: ProviderSnapshotSource,
387	snapshot: ServerProvider,
388	): Effect.Effect<ServerProvider> => {
389	if (snapshot.instanceId !== source.instanceId) {
390	return Effect.die(
391	new Error(
392	`Provider snapshot instance mismatch: source '${source.instanceId}' emitted '${snapshot.instanceId}'.`,
393	),
394	);
395	}
396	if (snapshot.driver !== source.driverKind) {
397	return Effect.die(
398	new Error(
399	`Provider snapshot driver mismatch for instance '${source.instanceId}': source '${source.driverKind}' emitted '${snapshot.driver}'.`,
400	),
401	);
402	}
403	return Effect.succeed(snapshot);
404	};
405
406	/**
407	* Key a snapshot for aggregation and persistence. Snapshot sources
408	* must be correlated by instance id before reaching this map; missing
409	* identities are defects, not runtime routing fallbacks.
410	*/
411	const snapshotInstanceKey = (provider: ServerProvider): ProviderInstanceId => {
412	return provider.instanceId;
413	};
414
415	// Project a live `ProviderInstance` into the aggregator's consumption
416	// shape. Each call re-captures the instance's `snapshot` closures, so
417	// after `ProviderInstanceRegistry` rebuilds an instance (e.g. because
418	// its settings changed), a fresh source rides the new PubSub instead
419	// of a closed one.
420	const buildSnapshotSource = (instance: ProviderInstance): ProviderSnapshotSource => ({
421	instanceId: instance.instanceId,
422	driverKind: instance.driverKind,
423	getSnapshot: instance.snapshot.getSnapshot,
424	refresh: instance.snapshot.refresh,
425	streamChanges: instance.snapshot.streamChanges,
426	});
427
428	export const layer = Layer.effect(
429	ProviderRegistry,
430	Effect.gen(function* () {
431	const instanceRegistry = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
432	const manifestService = yield* ModelManifest.ModelManifest;
433	const serviceScope = yield* Effect.scope;
434	const config = yield* ServerConfig.ServerConfig;
435	const fileSystem = yield* FileSystem.FileSystem;
436	const path = yield* Path.Path;
437
438	// Aggregator PubSub — consumers (WS gateway, etc.) subscribe here for
439	// coalesced updates across every instance.
440	const changesPubSub = yield* Effect.acquireRelease(
441	PubSub.unbounded<ReadonlyArray<ServerProvider>>(),
442	PubSub.shutdown,
443	);
444
445	// Boot-only: hydrate `providersRef` from the on-disk per-instance
446	// cache so the UI has something to render during the first refresh.
447	// Instances added post-boot skip this path; their first entry in
448	// `providersRef` comes from the reactive `syncLiveSources` pass
449	// below.
450	const bootInstances = yield* instanceRegistry.listInstances;
451	const bootSources = bootInstances.map(buildSnapshotSource);
452	const fallbackProviders = yield* loadProviders(bootSources);
453	const fallbackByInstance = new Map<ProviderInstanceId, ServerProvider>();
454	for (let index = 0; index < fallbackProviders.length; index++) {
455	const provider = fallbackProviders[index];
456	const source = bootSources[index];
457	if (provider === undefined || source === undefined) {
458	continue;
459	}
460	fallbackByInstance.set(source.instanceId, provider);
461	}
462
463	const cachedProviders = yield* Effect.forEach(
464	bootSources,
465	(source) =>
466	Effect.gen(function* () {
467	// One cache file per configured instance. For the default
468	// instance of a built-in kind the path equals `<kind>.json` —
469	// identical to the legacy filename. We still require the cache
470	// payload to carry matching instance id + driver kind; old
471	// identity-less payloads are discarded and the awaited refresh
472	// below repopulates the cache.
473	const filePath = yield* resolveProviderStatusCachePath({
474	cacheDir: config.providerStatusCacheDir,
475	instanceId: source.instanceId,
476	}).pipe(Effect.provideService(Path.Path, path));
477	const fallbackProvider = fallbackByInstance.get(source.instanceId);
478	if (fallbackProvider === undefined) {
479	return undefined;
480	}
481	return yield* readProviderStatusCache(filePath).pipe(
482	Effect.provideService(FileSystem.FileSystem, fileSystem),
483	Effect.flatMap((cachedProvider) => {
484	if (cachedProvider === undefined) {
485	return Effect.void.pipe(Effect.as(undefined as ServerProvider | undefined));
486	}
487	const correlation = {
488	cachedProvider,
489	fallbackProvider,
490	} as const;
491	if (!isCachedProviderCorrelated(correlation)) {
492	return Effect.logWarning("provider status cache identity mismatch, ignoring", {
493	path: filePath,
494	instanceId: source.instanceId,
495	cachedInstanceId: cachedProvider.instanceId ?? null,
496	driver: source.driverKind,
497	cachedDriver: cachedProvider.driver ?? null,
498	}).pipe(Effect.as(undefined as ServerProvider | undefined));
499	}
500	return Effect.succeed(hydrateCachedProvider(correlation));
501	}),
502	);
503	}),
504	{ concurrency: "unbounded" },
505	).pipe(
506	Effect.map((providers) =>
507	orderProviderSnapshots(
508	providers.filter((provider): provider is ServerProvider => provider !== undefined),
509	),
510	),
511	);
512	const initialManifest = yield* manifestService.current;
513	const classifyCompatibility = (
514	provider: ServerProvider,
515	manifest: ModelManifest.ModelManifestData,
516	) =>
517	applyProviderCompatibility(
518	provider,
519	manifest.compatibility,
520	ModelManifest.BUNDLED_MODEL_MANIFEST.compatibility,
521	);
522	const providersRef = yield* Ref.make<ReadonlyArray<ServerProvider>>(
523	cachedProviders.map((provider) => classifyCompatibility(provider, initialManifest)),
524	);
525	const workspaceRefreshesRef = yield* Ref.make<
526	ReadonlyMap<ProviderInstance, ReadonlySet<string>>
527	>(new Map());
528	const maintenanceActionStatesRef = yield* Ref.make<
529	ReadonlyMap<ProviderInstanceId, { readonly update?: ServerProviderUpdateState | undefined }>
530	>(new Map());
531
532	// Live-source registry — the dynamic counterpart to the boot-time
533	// `bootSources`. Keyed by `instanceId`; the stored `ProviderInstance`
534	// reference is used for identity equality so "no-op" reconciles
535	// (settings unchanged) skip re-subscribing + re-probing.
536	const liveSubsRef = yield* Ref.make<ReadonlyMap<ProviderInstanceId, ProviderInstance>>(
537	new Map(),
538	);
539	// Serialize `syncLiveSources` so a rapid burst of reconciles doesn't
540	// interleave two passes clobbering each other's fiber bookkeeping.
541	const syncSemaphore = yield* Semaphore.make(1);
542
543	const getLiveSources: Effect.Effect<ReadonlyArray<ProviderSnapshotSource>> = Ref.get(
544	liveSubsRef,
545	).pipe(Effect.map((map) => Array.from(map.values(), buildSnapshotSource)));
546
547	const persistProvider = (provider: ServerProvider) =>
548	Effect.gen(function* () {
549	// Persist every instance — the file name is the instance id, so
550	// multi-instance setups (e.g. `codex_personal`, `codex_work`) each
551	// get their own cache. We resolve the path fresh so snapshots
552	// produced by newly-added instances post-boot still land on disk
553	// without the aggregator holding a stale `cachePathByInstance`
554	// entry.
555	const key = snapshotInstanceKey(provider);
556	const filePath = yield* resolveProviderStatusCachePath({
557	cacheDir: config.providerStatusCacheDir,
558	instanceId: key,
559	}).pipe(Effect.provideService(Path.Path, path));
560	const { workspaceSnapshots: _workspaceSnapshots, ...machineProvider } = provider;
561	yield* writeProviderStatusCache({ filePath, provider: machineProvider }).pipe(
562	Effect.provideService(FileSystem.FileSystem, fileSystem),
563	Effect.provideService(Path.Path, path),
564	Effect.tapError(Effect.logError),
565	Effect.ignore,
566	);
567	});
568
569	const applyProviderUpdateState = Effect.fn("applyProviderUpdateState")(function* (
570	provider: ServerProvider,
571	) {
572	const maintenanceActionStates = yield* Ref.get(maintenanceActionStatesRef);
573	const updateState = maintenanceActionStates.get(provider.instanceId)?.update;
574	if (!updateState) {
575	const { updateState: _updateState, ...providerWithoutUpdateState } = provider;
576	return providerWithoutUpdateState;
577	}
578	return {
579	...provider,
580	updateState,
581	};
582	});
583
584	const upsertProviders = Effect.fn("upsertProviders")(function* (
585	nextProviders: ReadonlyArray<ServerProvider>,
586	options?: {
587	readonly publish?: boolean;
588	readonly persist?: boolean;
589	readonly replace?: boolean;
590	},
591	) {
592	const manifest = yield* manifestService.current;
593	const nextProvidersWithUpdateState = yield* Effect.forEach(
594	nextProviders,
595	applyProviderUpdateState,
596	{
597	concurrency: "unbounded",
598	},
599	);
600	const [previousProviders, providers, providersToPersist] = yield* Ref.modify(
601	providersRef,
602	(previousProviders) => {
603	const mergedProviders = new Map(
604	previousProviders.map((provider) => [snapshotInstanceKey(provider), provider] as const),
605	);
606	const updatedKeys = new Set<ProviderInstanceId>();
607
608	for (const provider of nextProvidersWithUpdateState) {
609	const key = snapshotInstanceKey(provider);
610	updatedKeys.add(key);
611	mergedProviders.set(
612	key,
613	options?.replace === true
614	? provider
615	: mergeProviderSnapshot(mergedProviders.get(key), provider),
616	);
617	}
618
619	const providers = orderProviderSnapshots(
620	[...mergedProviders.values()].map((provider) =>
621	classifyCompatibility(provider, manifest),
622	),
623	);
624	const providersToPersist = providers.filter((provider) =>
625	updatedKeys.has(snapshotInstanceKey(provider)),
626	);
627	return [[previousProviders, providers, providersToPersist] as const, providers];
628	},
629	);
630
631	if (haveProvidersChanged(previousProviders, providers)) {
632	if (options?.persist !== false) {
633	yield* Effect.forEach(providersToPersist, persistProvider, {
634	concurrency: "unbounded",
635	discard: true,
636	});
637	}
638	if (options?.publish !== false) {
639	yield* PubSub.publish(changesPubSub, providers);
640	}
641	}
642
643	return providers;
644	});
645
646	const compatibilityRefreshRunning = yield* Ref.make(false);
647	const syncProvider = Effect.fn("syncProvider")(function* (
648	provider: ServerProvider,
649	options?: {
650	readonly publish?: boolean;
651	},
652	) {
653	const providers = yield* upsertProviders([provider], options);
654	// Reclassify the current read model after fetching. Never republish the
655	// probe captured before the fetch: a newer health result may have landed.
656	if (!(yield* Ref.getAndSet(compatibilityRefreshRunning, true))) {
657	yield* manifestService.refresh.pipe(
658	Effect.andThen(upsertProviders([], { persist: false })),
659	Effect.ensuring(Ref.set(compatibilityRefreshRunning, false)),
660	Effect.forkIn(serviceScope),
661	);
662	}
663	return providers;
664	});
665
666	const setProviderMaintenanceActionState = Effect.fn("setProviderMaintenanceActionState")(
667	function* (input: {
668	readonly instanceId: ProviderInstanceId;
669	readonly action: "update";
670	readonly state: ServerProviderUpdateState | null;
671	}) {
672	yield* Ref.update(maintenanceActionStatesRef, (previous) => {
673	const previousActions = previous.get(input.instanceId);
674	const nextActions = { ...previousActions };
675	if (input.state === null || input.state.status === "idle") {
676	delete nextActions[input.action];
677	} else {
678	nextActions[input.action] = input.state;
679	}
680
681	const next = new Map(previous);
682	if (Object.keys(nextActions).length === 0) {
683	next.delete(input.instanceId);
684	} else {
685	next.set(input.instanceId, nextActions);
686	}
687	return next;
688	});
689
690	const existingProviders = yield* Ref.get(providersRef);
691	const matchingProvider = existingProviders.find(
692	(candidate) => candidate.instanceId === input.instanceId,
693	);
694	if (!matchingProvider) {
695	return existingProviders;
696	}
697
698	const nextProvider = yield* applyProviderUpdateState(matchingProvider);
699	return yield* upsertProviders([nextProvider], {
700	persist: false,
701	});
702	},
703	);
704
705	const refreshOneSource = Effect.fn("refreshOneSource")(function* (
706	providerSource: ProviderSnapshotSource,
707	) {
708	return yield* providerSource.refresh.pipe(
709	Effect.flatMap((nextProvider) =>
710	correlateSnapshotWithSource(providerSource, nextProvider).pipe(
711	Effect.flatMap(syncProvider),
712	),
713	),
714	);
715	});
716
717	type RefreshAllCompletion = Deferred.Deferred<ReadonlyArray<ServerProvider>>;
718	const refreshAllInFlightRef = yield* Ref.make<Option.Option<RefreshAllCompletion>>(
719	Option.none(),
720	);
721
722	// Untargeted refreshes probe every live source, so any read-scoped client
723	// can request them. Concurrent callers share one in-flight pass instead of
724	// each starting their own set of provider processes.
725	const refreshAll = Effect.fn("refreshAll")(function* () {
726	const claimed = yield* Ref.modify(
727	refreshAllInFlightRef,
728	(
729	inFlight,
730	): readonly [
731	{ readonly owner: boolean; readonly completion: RefreshAllCompletion },
732	Option.Option<RefreshAllCompletion>,
733	] => {
734	if (Option.isSome(inFlight)) {
735	return [{ owner: false, completion: inFlight.value }, inFlight];
736	}
737	const completion = Deferred.makeUnsafe<ReadonlyArray<ServerProvider>>();
738	return [{ owner: true, completion }, Option.some(completion)];
739	},
740	);
741	if (claimed.owner) {
742	// Detached so an interrupted first caller cannot cancel the pass that
743	// other callers are already awaiting.
744	yield* getLiveSources.pipe(
745	Effect.flatMap((sources) =>
746	Effect.forEach(sources, (source) => refreshOneSource(source), {
747	concurrency: "unbounded",
748	discard: true,
749	}),
750	),
751	Effect.andThen(Ref.get(providersRef)),
752	Effect.onExit((exit) =>
753	Ref.set(refreshAllInFlightRef, Option.none()).pipe(
754	Effect.andThen(Deferred.done(claimed.completion, exit)),
755	),
756	),
757	Effect.forkDetach,
758	);
759	}
760	return yield* Deferred.await(claimed.completion);
761	});
762
763	const refresh = Effect.fn("refresh")(function* (provider?: ProviderDriverKind) {
764	if (provider === undefined) {
765	return yield* refreshAll();
766	}
767	// Kind-scoped refreshes target the default instance for that driver.
768	const defaultInstanceId = defaultInstanceIdForDriver(provider);
769	const sources = yield* getLiveSources;
770	const providerSource = sources.find(
771	(candidate) => candidate.instanceId === defaultInstanceId,
772	);
773	if (!providerSource) {
774	return yield* Ref.get(providersRef);
775	}
776	return yield* refreshOneSource(providerSource);
777	});
778
779	const refreshInstance = Effect.fn("refreshInstance")(function* (
780	instanceId: ProviderInstanceId,
781	) {
782	const sources = yield* getLiveSources;
783	const providerSource = sources.find((candidate) => candidate.instanceId === instanceId);
784	if (!providerSource) {
785	return yield* Ref.get(providersRef);
786	}
787	return yield* refreshOneSource(providerSource);
788	});
789
790	const getProviderMaintenanceCapabilitiesForInstance = Effect.fn(
791	"getProviderMaintenanceCapabilitiesForInstance",
792	)(function* (
793	instanceId: ProviderInstanceId,
794	provider: ProviderDriverKind,
795	options?: { readonly fresh?: boolean },
796	) {
797	// Read the instance registry, not `liveSubsRef`: the latter trails
798	// reconciliation, and an update must never run a retired instance's
799	// command against a freshly configured executable.
800	const instance = yield* instanceRegistry.getInstance(instanceId);
801	if (!instance || instance.driverKind !== provider) {
802	return makeManualProviderMaintenanceCapabilities(provider);
803	}
804	return yield* instance.snapshot.resolveMaintenance(options);
805	});
806
807	/**
808	* Diff the aggregator's live-source set against the current
809	* `ProviderInstanceRegistry` and:
810	* - subscribe to each newly-added or rebuilt instance's
811	* `streamChanges` (so periodic + enrichment refreshes land in
812	* `providersRef`);
813	* - read each newly-added/rebuilt instance's current snapshot after
814	* subscribing, closing the race with its independently-running
815	* background startup probe;
816	* - prune `providersRef` of instances that no longer exist.
817	*
818	* Provider refreshes are owned by each managed provider and never run
819	* on this layer's construction path. Consumers see cached or pending
820	* snapshots immediately, then receive live probe results through the
821	* already-attached change stream.
822	*
823	* Per-instance subscription fibers are not tracked explicitly. When
824	* a rebuilt instance's old child scope closes, its PubSub shuts
825	* down and our `Stream.runForEach` fiber exits naturally.
826	*/
827	const syncLiveSources = syncSemaphore.withPermits(1)(
828	Effect.gen(function* () {
829	const instances = yield* instanceRegistry.listInstances;
830	const unavailableProviders = yield* instanceRegistry.listUnavailable;
831	const nextByInstance = new Map<ProviderInstanceId, ProviderInstance>(
832	instances.map((instance) => [instance.instanceId, instance] as const),
833	);
834	const knownInstanceIds = new Set<ProviderInstanceId>(nextByInstance.keys());
835	for (const provider of unavailableProviders) {
836	knownInstanceIds.add(snapshotInstanceKey(provider));
837	}
838	const previousSubs = yield* Ref.get(liveSubsRef);
839
840	// Carry over subscriptions for instances whose identity is
841	// unchanged (reconcile treated them as no-op). Instances that
842	// disappeared, or were rebuilt with a different reference,
843	// fall through to the "newly-added" branch below.
844	const carriedOver = new Map<ProviderInstanceId, ProviderInstance>();
845	for (const [instanceId, previousInstance] of previousSubs) {
846	const nextInstance = nextByInstance.get(instanceId);
847	if (nextInstance !== undefined && nextInstance === previousInstance) {
848	carriedOver.set(instanceId, previousInstance);
849	}
850	}
851
852	// Collect new/rebuilt instances in `nextByInstance` insertion
853	// order (which preserves settings-author order).
854	const newlyAdded: Array<readonly [ProviderInstanceId, ProviderInstance]> = [];
855	for (const [instanceId, instance] of nextByInstance) {
856	if (carriedOver.has(instanceId)) {
857	continue;
858	}
859	newlyAdded.push([instanceId, instance] as const);
860	}
861
862	const rebuiltInstanceIds = new Set(
863	newlyAdded
864	.map(([instanceId]) => instanceId)
865	.filter((instanceId) => previousSubs.has(instanceId)),
866	);
867	if (rebuiltInstanceIds.size > 0) {
868	const [previousProviders, providers] = yield* Ref.modify(
869	providersRef,
870	(previousProviders) => {
871	const providers = previousProviders.map((provider) => {
872	if (!rebuiltInstanceIds.has(provider.instanceId)) return provider;
873	const { workspaceSnapshots: _workspaceSnapshots, ...machineSnapshot } = provider;
874	return machineSnapshot;
875	});
876	return [[previousProviders, providers] as const, providers];
877	},
878	);
879	if (haveProvidersChanged(previousProviders, providers)) {
880	yield* PubSub.publish(changesPubSub, providers);
881	}
882	}
883
884	// Fork long-lived subscriptions to each new/rebuilt instance's
885	// change stream before reading its current snapshot. If the
886	// driver's own initial probe finishes during this sync, either
887	// the current read or the active subscriber observes the result.
888	for (const [, instance] of newlyAdded) {
889	const source = buildSnapshotSource(instance);
890	yield* Stream.runForEach(source.streamChanges, (provider) =>
891	correlateSnapshotWithSource(source, provider).pipe(Effect.flatMap(syncProvider)),
892	).pipe(Effect.forkScoped);
893	}
894	yield* Effect.yieldNow;
895
896	// Snapshot current state without starting a probe. Managed providers
897	// launch their startup refresh independently, so this closes the
898	// subscription race without putting external work on the registry
899	// or HTTP server construction path.
900	yield* Effect.forEach(
901	newlyAdded,
902	([, instance]) =>
903	Effect.gen(function* () {
904	const source = buildSnapshotSource(instance);
905	const provider = yield* source.getSnapshot;
906	yield* correlateSnapshotWithSource(source, provider).pipe(
907	Effect.flatMap(syncProvider),
908	);
909	}).pipe(Effect.ignoreCause({ log: true })),
910	{ concurrency: "unbounded", discard: true },
911	);
912	yield* upsertProviders(unavailableProviders, {
913	persist: false,
914	replace: true,
915	});
916
917	const nextSubs = new Map(carriedOver);
918	for (const [instanceId, instance] of newlyAdded) {
919	nextSubs.set(instanceId, instance);
920	}
921	yield* Ref.set(liveSubsRef, nextSubs);
922
923	// Drop aggregator state for instances that have disappeared —
924	// otherwise the UI would keep rendering ghosts.
925	const [previousProviders, providers] = yield* Ref.modify(
926	providersRef,
927	(previousProviders) => {
928	const providers = orderProviderSnapshots(
929	previousProviders.filter((provider) =>
930	knownInstanceIds.has(snapshotInstanceKey(provider)),
931	),
932	);
933	return [[previousProviders, providers] as const, providers];
934	},
935	);
936	if (haveProvidersChanged(previousProviders, providers)) {
937	yield* PubSub.publish(changesPubSub, providers);
938	}
939	yield* Ref.update(maintenanceActionStatesRef, (previous) => {
940	const next = new Map(previous);
941	for (const instanceId of previous.keys()) {
942	if (!knownInstanceIds.has(instanceId)) {
943	next.delete(instanceId);
944	}
945	}
946	return next;
947	});
948	}),
949	);
950	const syncLiveSourcesAndContinue = syncLiveSources.pipe(
951	Effect.catchCause((cause) => {
952	if (Cause.hasInterruptsOnly(cause)) {
953	return Effect.interrupt;
954	}
955	return Effect.logError(
956	"provider registry instance sync failed; keeping subscription alive",
957	{
958	cause: Cause.pretty(cause),
959	},
960	);
961	}),
962	);
963
964	// Seed `providersRef` with the boot-time fallback snapshots so
965	// consumers calling `getProviders` immediately after layer build see
966	// a populated list — even before the first `syncLiveSources` refresh
967	// resolves. Cached snapshots (already in `providersRef`) merge with
968	// these via `upsertProviders` so on-disk state wins where present
969	// and pending fallbacks fill the gaps.
970	yield* upsertProviders(fallbackProviders, { publish: false });
971	// Subscribe to registry mutations BEFORE running the initial sync.
972	// `subscribeChanges` acquires the dequeue synchronously in this
973	// fibre; the subscription is active the instant this `yield*`
974	// returns. Forking the consumer loop later cannot lose a publish
975	// because no publish can reach a not-yet-subscribed dequeue.
976	//
977	// (Contrast with the pre-fix code that did
978	// `Stream.runForEach(instanceRegistry.streamChanges, …).pipe(Effect.forkScoped)`.
979	// `Stream.fromPubSub` defers `PubSub.subscribe` to stream start,
980	// and `forkScoped` only schedules the fibre — so a reconcile that
981	// published between "fibre scheduled" and "fibre starts running"
982	// was dropped, which made any settings change that replaced an
983	// instance never propagate to the aggregator's `providersRef`.)
984	// Subscribe to registry mutations BEFORE running the initial sync.
985	// `subscribeChanges` acquires the `PubSub.Subscription` synchronously
986	// in this fibre; the subscription is registered with the PubSub the
987	// instant this `yield*` returns, so any subsequent publish is
988	// buffered in the subscription regardless of when the consumer
989	// fibre below actually starts running.
990	//
991	// (Contrast with the pre-fix code that did
992	// `Stream.runForEach(instanceRegistry.streamChanges, …).pipe(Effect.forkScoped)`.
993	// `instanceRegistry.streamChanges` is `Stream.fromPubSub(changes)`,
994	// which defers `PubSub.subscribe` to stream start. `forkScoped` only
995	// schedules the consumer fibre — so a reconcile that published
996	// between "fibre scheduled" and "fibre starts running + subscribes"
997	// was dropped, which made any settings change that replaced an
998	// instance never propagate to the aggregator's `providersRef`.)
999	const instanceChanges = yield* instanceRegistry.subscribeChanges;
1000	// Initial sync attaches subscriptions and snapshots current state for
1001	// every instance present at boot. Provider probes are already running in
1002	// their managed background fibers and never block this layer.
1003	yield* syncLiveSources;
1004	// React to registry mutations — instance added / removed / rebuilt.
1005	// `Stream.fromSubscription` builds a stream over the pre-acquired
1006	// subscription rather than subscribing on stream start, which is
1007	// what closes the race.
1008	yield* Stream.runForEach(
1009	Stream.fromSubscription(instanceChanges),
1010	() => syncLiveSourcesAndContinue,
1011	).pipe(Effect.forkScoped);
1012
1013	const recoverRefreshFailure = Effect.fn("recoverRefreshFailure")(function* (
1014	cause: Cause.Cause<unknown>,
1015	) {
1016	if (Cause.hasInterruptsOnly(cause)) {
1017	return yield* Effect.interrupt;
1018	}
1019	yield* Effect.logError("provider registry refresh failed; preserving cached providers", {
1020	cause: Cause.pretty(cause),
1021	});
1022	return yield* Ref.get(providersRef);
1023	});
1024
1025	const updateProviders = (
1026	update: (providers: ReadonlyArray<ServerProvider>) => ReadonlyArray<ServerProvider>,
1027	) =>
1028	Ref.modify(providersRef, (currentProviders) => {
1029	const nextProviders = update(currentProviders);
1030	return [[currentProviders, nextProviders] as const, nextProviders];
1031	}).pipe(
1032	Effect.tap(([previousProviders, nextProviders]) =>
1033	haveProvidersChanged(previousProviders, nextProviders)
1034	? PubSub.publish(changesPubSub, nextProviders)
1035	: Effect.void,
1036	),
1037	Effect.map(([, nextProviders]) => nextProviders),
1038	);
1039
1040	const refreshWorkspaceSnapshot = Effect.fn("refreshWorkspaceSnapshot")(function* (input: {
1041	readonly instanceId: ProviderInstanceId;
1042	readonly cwd: string;
1043	readonly fresh?: boolean;
1044	}) {
1045	// Fresh scans drop other instances' snapshots for this cwd first, so a
1046	// composer on one of them scans again on next use, even when this
1047	// instance is gone or cannot be scanned.
1048	if (input.fresh) {
1049	yield* updateProviders((providers) =>
1050	providers.map((candidate) =>
1051	candidate.instanceId === input.instanceId
1052	? candidate
1053	: dropProviderWorkspaceSnapshot(candidate, input.cwd),
1054	),
1055	);
1056	}
1057	const providers = yield* Ref.get(providersRef);
1058	const provider = providers.find((candidate) => candidate.instanceId === input.instanceId);
1059	const workspaceSnapshotOf = (candidate: ServerProvider | undefined) =>
1060	candidate?.workspaceSnapshots?.find((s) => s.cwd === input.cwd);
1061	const scannedFrom = workspaceSnapshotOf(provider);
1062	const now = yield* DateTime.now;
1063	if (
1064	!provider ||
1065	!provider.enabled ||
1066	(!input.fresh &&
1067	scannedFrom &&
1068	!scannedFrom.slashCommandsPending &&
1069	isProviderWorkspaceSnapshotCurrent(scannedFrom, DateTime.toEpochMillis(now)))
1070	) {
1071	return providers;
1072	}
1073	// Drivers spread their machine snapshot, whose `checkedAt` is the last
1074	// health check. The TTL needs the time this scan started reading files.
1075	const scannedAt = DateTime.formatIso(now);
1076	const instance = yield* instanceRegistry.getInstance(input.instanceId);
1077	if (!instance?.snapshotForCwd) return providers;
1078	const claimed = yield* Ref.modify(workspaceRefreshesRef, (refreshes) => {
1079	const current = refreshes.get(instance);
1080	if (current?.has(input.cwd)) return [false, refreshes] as const;
1081	const next = new Map(refreshes);
1082	next.set(instance, new Set(current).add(input.cwd));
1083	return [true, next] as const;
1084	});
1085	// A fresh scan never joins a running one, which may predate the change.
1086	if (!claimed && !input.fresh) return yield* Ref.get(providersRef);
1087	// Fresh scans also re-read the machine snapshot after invalidating caches.
1088	const refreshMachineSnapshot = input.fresh
1089	? (instance.invalidateCaches ?? Effect.void).pipe(
1090	Effect.andThen(refreshInstance(input.instanceId)),
1091	)
1092	: Effect.void;
1093	return yield* refreshMachineSnapshot.pipe(
1094	Effect.andThen(instance.snapshotForCwd(input.cwd)),
1095	Effect.flatMap((scopedSnapshot) =>
1096	scopedSnapshot.status === "error" && scopedSnapshot.slashCommandsPending === undefined
1097	? Ref.get(providersRef)
1098	: instanceRegistry.getInstance(input.instanceId).pipe(
1099	Effect.flatMap((currentInstance) => {
1100	if (currentInstance !== instance) return Ref.get(providersRef);
1101	// Write only if the cwd's snapshot did not change during the
1102	// scan. A session event or another scan that landed first is newer.
1103	return updateProviders((currentProviders) =>
1104	currentProviders.map((candidate) =>
1105	candidate.instanceId === input.instanceId &&
1106	Equal.equals(workspaceSnapshotOf(candidate), scannedFrom)
1107	? upsertProviderWorkspaceSnapshot(candidate, input.cwd, {
1108	...scopedSnapshot,
1109	checkedAt: scannedAt,
1110	})
1111	: candidate,
1112	),
1113	);
1114	}),
1115	),
1116	),
1117	Effect.ensuring(
1118	claimed
1119	? Ref.update(workspaceRefreshesRef, (refreshes) => {
1120	const next = new Map(refreshes);
1121	const current = new Set(next.get(instance));
1122	current.delete(input.cwd);
1123	if (current.size) next.set(instance, current);
1124	else next.delete(instance);
1125	return next;
1126	})
1127	: Effect.void,
1128	),
1129	);
1130	});
1131
1132	return {
1133	getProviders: Ref.get(providersRef),
1134	refresh: (provider?: ProviderDriverKind) =>
1135	refresh(provider).pipe(Effect.catchCause(recoverRefreshFailure)),
1136	refreshInstance: (instanceId: ProviderInstanceId) =>
1137	refreshInstance(instanceId).pipe(Effect.catchCause(recoverRefreshFailure)),
1138	refreshWorkspaceSnapshot: (input) =>
1139	refreshWorkspaceSnapshot(input).pipe(Effect.catchCause(recoverRefreshFailure)),
1140	getProviderMaintenanceCapabilitiesForInstance,
1141	setProviderMaintenanceActionState,
1142	get streamChanges() {
1143	return Stream.fromPubSub(changesPubSub);
1144	},
1145	} satisfies ProviderRegistry["Service"];
1146	}),
1147	);