diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 93fec081ebb9c..1fb343eb2de5c 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -1934,6 +1934,139 @@ it.layer( }), ); + it.effect("ignores a probe from an instance that settings replaced mid-refresh", () => + Effect.gen(function* () { + const scenario = Effect.fn(function* ( + probeFails: boolean, + run: ( + registry: ProviderRegistry.ProviderRegistry["Service"], + ) => Effect.Effect>, + ) { + const driver = ProviderDriverKind.make("codex"); + const instanceId = ProviderInstanceId.make("codex"); + const provider = { + instanceId, + driver, + status: "ready", + enabled: true, + installed: true, + auth: { status: "authenticated" }, + checkedAt: "2026-06-10T00:00:00.000Z", + version: "2.0.0", + models: [], + slashCommands: [], + skills: [], + } as const satisfies ServerProvider; + const retiredProbe = { + ...provider, + status: "error", + installed: false, + version: "1.0.0", + message: "Codex CLI is not installed.", + } as const satisfies ServerProvider; + const probeStarted = yield* Deferred.make(); + const releaseProbe = yield* Deferred.make(); + const makeInstance = (snapshot: ServerProvider, refresh: Effect.Effect) => + ({ + instanceId, + driverKind: driver, + continuationIdentity: { driverKind: driver, continuationKey: "codex:instance:codex" }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + getSnapshot: Effect.succeed(snapshot), + refresh: Effect.as(refresh, snapshot), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + }) satisfies ProviderInstance; + const retiredInstance = makeInstance( + retiredProbe, + Deferred.succeed(probeStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseProbe)), + Effect.andThen(probeFails ? Effect.die(new Error("probe failed")) : Effect.void), + ), + ); + const rebuiltInstance = makeInstance(provider, Effect.void); + const instances = yield* Ref.make>([retiredInstance]); + const changes = yield* PubSub.unbounded(); + const instanceRegistryLayer = Layer.succeed( + ProviderInstanceRegistry.ProviderInstanceRegistry, + { + getInstance: (requestedId) => + Ref.get(instances).pipe( + Effect.map((current) => + current.find((instance) => instance.instanceId === requestedId), + ), + ), + listInstances: Ref.get(instances), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: PubSub.subscribe(changes), + }, + ); + const scope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)); + const runtimeServices = yield* Layer.build( + ProviderRegistry.layer.pipe( + Layer.provide(layerUnusedMutator), + Layer.provideMerge(instanceRegistryLayer), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-retired-refresh-", + }), + ), + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ), + ).pipe(Scope.provide(scope)); + + return yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + const rebuiltPublished = yield* Stream.toPull( + registry.streamChanges.pipe( + Stream.filter((providers) => providers[0]?.version === provider.version), + ), + ); + const refresh = yield* run(registry).pipe(Effect.forkChild); + yield* Deferred.await(probeStarted); + yield* Ref.set(instances, [rebuiltInstance]); + yield* PubSub.publish(changes, undefined); + yield* rebuiltPublished; + yield* Deferred.succeed(releaseProbe, undefined); + return { + provider, + result: yield* Fiber.join(refresh), + providers: yield* registry.getProviders, + }; + }).pipe(Effect.provide(runtimeServices)); + }); + const driver = ProviderDriverKind.make("codex"); + const instanceId = ProviderInstanceId.make("codex"); + + const manual = yield* scenario(false, (registry) => registry.refresh()); + expect(manual.result).toEqual([expect.objectContaining(manual.provider)]); + expect(manual.providers).toEqual([expect.objectContaining(manual.provider)]); + + // Neither the retired probe's failure nor its cached row can stand in for the replacement. + for (const probeFails of [true, false]) { + const afterUpdate = yield* scenario(probeFails, (registry) => + registry.refreshAfterUpdate(driver, instanceId), + ); + expect(afterUpdate.result).toEqual([]); + expect(afterUpdate.providers).toEqual([expect.objectContaining(afterUpdate.provider)]); + } + }), + ); + it.effect("keeps an enriched snapshot that streams in before a manual refresh returns", () => Effect.gen(function* () { const driver = ProviderDriverKind.make("acpRegistry"); diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index 76f2723cd9d4f..fb6aeaac26b63 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -444,10 +444,6 @@ export const layer = Layer.effect( ); const syncSemaphore = yield* Semaphore.make(1); - const getLiveSources: Effect.Effect> = Ref.get( - liveSubsRef, - ).pipe(Effect.map((map) => Array.from(map.values(), buildSnapshotSource))); - const persistProvider = Effect.fn("ProviderRegistry.persistProvider")(function* ( provider: ServerProvider, ) { @@ -601,17 +597,25 @@ export const layer = Layer.effect( }, ); - const refreshOneSource = Effect.fn("refreshOneSource")(function* ( - providerSource: ProviderSnapshotSource, - ) { + const isCurrentInstance = (instance: ProviderInstance) => + instanceRegistry + .getInstance(instance.instanceId) + .pipe(Effect.map((current) => current === instance)); + + const refreshOneSource = Effect.fn("refreshOneSource")(function* (instance: ProviderInstance) { + const providerSource = buildSnapshotSource(instance); // Enrichment can publish a newer snapshot before refresh returns its pre-enrichment probe. - return yield* providerSource.refresh.pipe( + const probe = yield* providerSource.refresh.pipe( Effect.andThen(providerSource.getSnapshot), - Effect.flatMap((nextProvider) => - correlateSnapshotWithSource(providerSource, nextProvider).pipe( - Effect.flatMap(syncProvider), - ), - ), + Effect.exit, + ); + // Settings can replace or remove the instance mid-probe; its late result or failure is stale. + if (!(yield* isCurrentInstance(instance))) { + return yield* Ref.get(providersRef); + } + const nextProvider = yield* probe; + return yield* correlateSnapshotWithSource(providerSource, nextProvider).pipe( + Effect.flatMap(syncProvider), ); }); @@ -640,9 +644,9 @@ export const layer = Layer.effect( // Detach shared passes so the first caller's interruption cannot cancel other callers' work. // Syncing first keeps a refresh right after a rebuild from probing the replaced instance. yield* syncLiveSourcesAndContinue.pipe( - Effect.andThen(getLiveSources), - Effect.flatMap((sources) => - Effect.forEach(sources, (source) => refreshOneSource(source), { + Effect.andThen(Ref.get(liveSubsRef)), + Effect.flatMap((instances) => + Effect.forEach(instances.values(), (instance) => refreshOneSource(instance), { concurrency: "unbounded", discard: true, }), @@ -663,27 +667,22 @@ export const layer = Layer.effect( if (provider === undefined) { return yield* refreshAll(); } - const defaultInstanceId = defaultInstanceIdForDriver(provider); - const sources = yield* getLiveSources; - const providerSource = sources.find( - (candidate) => candidate.instanceId === defaultInstanceId, - ); - if (!providerSource) { + const instance = (yield* Ref.get(liveSubsRef)).get(defaultInstanceIdForDriver(provider)); + if (!instance) { return yield* Ref.get(providersRef); } - return yield* refreshOneSource(providerSource); + return yield* refreshOneSource(instance); }); const refreshInstance = Effect.fn("refreshInstance")(function* ( instanceId: ProviderInstanceId, ) { yield* syncLiveSourcesAndContinue; - const sources = yield* getLiveSources; - const providerSource = sources.find((candidate) => candidate.instanceId === instanceId); - if (!providerSource) { + const instance = (yield* Ref.get(liveSubsRef)).get(instanceId); + if (!instance) { return yield* Ref.get(providersRef); } - return yield* refreshOneSource(providerSource); + return yield* refreshOneSource(instance); }); const reloadSecretsAndRefresh = Effect.fn("reloadSecretsAndRefresh")(function* ( @@ -744,15 +743,18 @@ export const layer = Layer.effect( yield* Effect.forEach( instances.filter((instance) => !failedInstances.has(instance.instanceId)), (instance) => - refreshOneSource(buildSnapshotSource(instance)).pipe( + refreshOneSource(instance).pipe( Effect.catchCause((cause) => recoverInstanceFailure(instance.instanceId, cause)), ), { concurrency: "unbounded", discard: true }, ); + // A replaced instance's cached row did not come from this pass's probe. + const refreshedInstances = yield* Effect.filter( + instances.filter((instance) => !failedInstances.has(instance.instanceId)), + isCurrentInstance, + ); const refreshedInstanceIds = new Set( - instances - .filter((instance) => !failedInstances.has(instance.instanceId)) - .map((instance) => instance.instanceId), + refreshedInstances.map((instance) => instance.instanceId), ); return (yield* Ref.get(providersRef)).filter((candidate) => refreshedInstanceIds.has(candidate.instanceId),