diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 9b1187493967c..700a45702327d 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -76,6 +76,7 @@ import * as ProviderInstanceRegistryMutator from "./ProviderInstanceRegistryMuta import * as ProviderRegistry from "./ProviderRegistry.ts"; import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment"; import { makeManualOnlyProviderMaintenanceCapabilities } from "@t3tools/provider-core/server/maintenanceResolver"; +import { ProviderDriverError } from "@t3tools/provider-core/server/errors"; vi.mock("@anthropic-ai/claude-agent-sdk", { spy: true }); const decodeServerProvider = Schema.decodeSync(ServerProvider); @@ -1489,6 +1490,167 @@ it.layer( ); }); + it.effect.each(["all", "single", "disabled", "uninstalled", "non-refreshable"] as const)( + "%s model refresh preserves cached models and skips ineligible instances", + (target) => + Effect.gen(function* () { + const driver = ProviderDriverKind.make("antigravity"); + const firstId = ProviderInstanceId.make("antigravity_first"); + const secondId = ProviderInstanceId.make("antigravity_second"); + const thirdId = ProviderInstanceId.make("antigravity_third"); + const skippedIds = { + disabled: ProviderInstanceId.make("antigravity_disabled"), + uninstalled: ProviderInstanceId.make("antigravity_uninstalled"), + "non-refreshable": ProviderInstanceId.make("antigravity_static"), + }; + const driverError = new ProviderDriverError({ + driver, + instanceId: firstId, + detail: "Model refresh failed.", + cause: new Error("native refresh failure"), + }); + const thirdStarted = yield* Deferred.make(); + const calls = yield* Ref.make>([]); + const activeRefreshes = yield* Ref.make(0); + const maxRefreshes = yield* Ref.make(0); + const model = (slug: string) => ({ + slug, + name: slug, + isCustom: false, + capabilities: null, + }); + const makeInstance = Effect.fn("Test.makeModelRefreshInstance")(function* ( + instanceId: ProviderInstanceId, + options: { enabled?: boolean; installed?: boolean; refreshable?: boolean } = {}, + ) { + const snapshot = yield* Ref.make({ + instanceId, + driver, + enabled: options.enabled ?? true, + installed: options.installed ?? true, + version: "1.1.1", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-10-25T00:00:00.000Z", + models: [model("cached-model")], + slashCommands: [], + skills: [], + }); + return { + instanceId, + driverKind: driver, + continuationIdentity: { driverKind: driver, continuationKey: instanceId }, + displayName: undefined, + enabled: options.enabled ?? true, + snapshot: { + getSnapshot: Ref.get(snapshot), + refresh: Ref.get(snapshot), + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + ...(options.refreshable === false + ? {} + : { + refreshModels: () => + Effect.gen(function* () { + yield* Ref.update(calls, (ids) => [...ids, instanceId]); + if (instanceId === firstId) { + return yield* driverError; + } + const active = yield* Ref.updateAndGet( + activeRefreshes, + (count) => count + 1, + ); + yield* Ref.update(maxRefreshes, (count) => Math.max(count, active)); + yield* instanceId === secondId + ? Deferred.await(thirdStarted) + : Deferred.succeed(thirdStarted, undefined); + yield* Ref.update(snapshot, (provider) => ({ + ...provider, + models: [model("refreshed-model")], + })); + yield* Ref.update(activeRefreshes, (count) => count - 1); + }), + }), + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + } satisfies ProviderInstance; + }); + const instances = [ + yield* makeInstance(firstId), + yield* makeInstance(secondId), + yield* makeInstance(thirdId), + yield* makeInstance(skippedIds.disabled, { enabled: false }), + yield* makeInstance(skippedIds.uninstalled, { installed: false }), + yield* makeInstance(skippedIds["non-refreshable"], { refreshable: false }), + ]; + const instanceLayer = Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { + listInstances: Effect.succeed(instances), + getInstance: (id) => + Effect.succeed(instances.find((instance) => instance.instanceId === id)), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: Effect.flatMap(PubSub.unbounded(), PubSub.subscribe), + }); + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + if (target === "single") { + const failure = yield* registry.refreshModels(firstId).pipe(Effect.flip); + assert.equal(failure.instanceId, firstId); + assert.equal(failure.operation, "refresh-models"); + assert.equal(failure.detail, driverError.detail); + assert.strictEqual(failure.cause, driverError); + } else { + yield* registry.refreshModels(target === "all" ? undefined : skippedIds[target]); + } + assert.sameMembers( + [...(yield* Ref.get(calls))], + target === "all" + ? [firstId, secondId, thirdId] + : target === "single" + ? [firstId] + : [], + ); + assert.equal(yield* Ref.get(maxRefreshes), target === "all" ? 2 : 0); + assert.deepEqual( + Object.fromEntries( + (yield* registry.getProviders).map((provider) => [ + provider.instanceId, + provider.models.map((entry) => entry.slug), + ]), + ), + { + [firstId]: ["cached-model"], + [secondId]: [target === "all" ? "refreshed-model" : "cached-model"], + [thirdId]: [target === "all" ? "refreshed-model" : "cached-model"], + [skippedIds.disabled]: ["cached-model"], + [skippedIds.uninstalled]: ["cached-model"], + [skippedIds["non-refreshable"]]: ["cached-model"], + }, + ); + }).pipe( + Effect.provide( + ProviderRegistry.layer.pipe( + Layer.provide(instanceLayer), + Layer.provide(layerUnusedMutator), + Layer.provide( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-model-refresh-", + }), + ), + ), + ), + ); + }), + ); + it.effect("invalidates model caches only for an explicit model refresh", () => Effect.gen(function* () { const forceRefreshCalls = yield* Ref.make(0); diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index fb6aeaac26b63..de3783710fef2 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -2,6 +2,7 @@ import { defaultInstanceIdForDriver, isProviderWorkspaceSnapshotCurrent, ProviderDriverKind, + ProviderSetupError, type ProviderInstanceId, type ServerProvider, type ServerProviderUpdateState, @@ -64,6 +65,10 @@ export class ProviderRegistry extends Context.Service< instanceId: ProviderInstanceId, ) => Effect.Effect>; + readonly refreshModels: ( + instanceId?: ProviderInstanceId, + ) => Effect.Effect, ProviderSetupError>; + /** Re-reads secret-backed environments first, so a fixed secret applies without a restart. */ readonly reloadSecretsAndRefresh: ( instanceId?: ProviderInstanceId, @@ -685,6 +690,58 @@ export const layer = Layer.effect( return yield* refreshOneSource(instance); }); + const refreshModels = Effect.fn("ProviderRegistry.refreshModels")(function* ( + instanceId?: ProviderInstanceId, + ) { + yield* syncLiveSourcesAndContinue; + const providers = yield* Ref.get(providersRef); + const instances = yield* instanceRegistry.listInstances; + yield* Effect.forEach( + instances, + (instance) => { + if ( + !instance.refreshModels || + (instanceId !== undefined && instanceId !== instance.instanceId) || + !providers.some( + (provider) => + provider.instanceId === instance.instanceId && + provider.enabled && + provider.installed, + ) + ) { + return Effect.void; + } + const refresh = instance.refreshModels().pipe( + Effect.mapError( + (error) => + new ProviderSetupError({ + instanceId: instance.instanceId, + operation: "refresh-models", + detail: error.detail, + cause: error, + }), + ), + Effect.andThen( + refreshInstance(instance.instanceId).pipe(Effect.catchCause(recoverRefreshFailure)), + ), + Effect.asVoid, + ); + return instanceId === undefined + ? refresh.pipe( + Effect.catch((error) => + Effect.logWarning("provider model refresh failed", { + instanceId: instance.instanceId, + detail: error.detail, + }), + ), + ) + : refresh; + }, + { concurrency: 2, discard: true }, + ); + return yield* Ref.get(providersRef); + }); + const reloadSecretsAndRefresh = Effect.fn("reloadSecretsAndRefresh")(function* ( instanceId?: ProviderInstanceId, ) { @@ -1080,6 +1137,7 @@ export const layer = Layer.effect( refreshWorkspaceSnapshot: (input) => refreshWorkspaceSnapshot(input).pipe(Effect.catchCause(recoverRefreshFailure)), refreshAfterUpdate, + refreshModels, invalidateModelCaches, getProviderMaintenanceCapabilitiesForInstance, setProviderMaintenanceActionState, diff --git a/apps/server/src/provider/providerMaintenanceRunner.test.ts b/apps/server/src/provider/providerMaintenanceRunner.test.ts index 77a44c7fa29a5..6f32a012053ea 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.test.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.test.ts @@ -197,6 +197,7 @@ function makeRegistry( getProviders: Ref.get(providersRef), refresh: () => Ref.get(providersRef), refreshInstance: () => Ref.get(providersRef), + refreshModels: () => Ref.get(providersRef), reloadSecretsAndRefresh: () => Ref.get(providersRef), refreshAfterUpdate: () => { const version = options?.installedVersionAfterUpdate; diff --git a/apps/server/src/provider/testUtils/providerRegistryMock.ts b/apps/server/src/provider/testUtils/providerRegistryMock.ts index f75ecd141e71a..259a89d1a4178 100644 --- a/apps/server/src/provider/testUtils/providerRegistryMock.ts +++ b/apps/server/src/provider/testUtils/providerRegistryMock.ts @@ -11,6 +11,7 @@ const makeProviderRegistryMock = ( getProviders: Effect.succeed(providers), refresh: () => Effect.succeed(providers), refreshInstance: () => Effect.succeed(providers), + refreshModels: () => Effect.succeed(providers), reloadSecretsAndRefresh: () => Effect.succeed(providers), refreshAfterUpdate: () => Effect.succeed(providers), refreshWorkspaceSnapshot: () => Effect.succeed(providers), diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 2be9e61c7c916..571d9b5cb6ea4 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -13,6 +13,7 @@ import { TurnItemId, ProviderDriverKind, ProviderInstanceId, + ProviderSetupError, SourceControlRepositoryError, WS_METHODS, WsRpcGroup, @@ -32,12 +33,14 @@ import * as Exit from "effect/Exit"; import * as Cause from "effect/Cause"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/http"; -import { RpcGroup, RpcMessage, RpcServer } from "effect/rpc"; +import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/rpc"; import * as SqlClient from "effect/sql/SqlClient"; import * as Statement from "effect/sql/Statement"; import { vi } from "vite-plus/test"; @@ -81,6 +84,7 @@ import * as ProjectCloneTracker from "./project/ProjectCloneTracker.ts"; import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; import * as ProviderAuthService from "./provider/ProviderAuthService.ts"; import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts"; +import { ProviderDriverError } from "@t3tools/provider-core/server/errors"; import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; @@ -144,6 +148,10 @@ import { withLateConfigDiscovery, } from "./ws.ts"; +const decodeModelRefreshExit = Schema.decodeUnknownEffect( + Schema.toCodecJson(Rpc.exitSchema(WsServerRefreshProvidersRpc)), +); + const snapshotConfig = (fields: Partial) => ({ availableEditors: [], settings: {}, ...fields }) as unknown as ServerConfig; @@ -1346,6 +1354,151 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( ), ); +it.effect.each(["all", "single"] as const)( + "%s model refresh RPC encodes responses without internal causes", + (target) => + Effect.gen(function* () { + const driver = ProviderDriverKind.make("antigravity"); + const firstId = ProviderInstanceId.make("antigravity_first"); + const privateCode = "private-model-refresh-oauth-code"; + const driverError = new ProviderDriverError({ + driver, + instanceId: firstId, + detail: "Model refresh failed.", + cause: new Error(`https://oauth.invalid/callback?code=${privateCode}`), + }); + const providers: ReadonlyArray = [ + { + instanceId: firstId, + driver, + enabled: true, + installed: true, + version: "1.1.1", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-10-25T00:00:00.000Z", + models: [{ slug: "model", name: "Model", isCustom: false, capabilities: null }], + slashCommands: [], + skills: [], + }, + ]; + const requests = yield* Queue.unbounded(); + const registryLayer = Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed(providers), + refresh: () => Effect.succeed(providers), + refreshInstance: () => Effect.succeed(providers), + invalidateModelCaches: () => Effect.void, + streamChanges: Stream.empty, + refreshModels: (instanceId) => + Queue.offer(requests, instanceId).pipe( + Effect.andThen( + instanceId === undefined + ? Effect.succeed(providers) + : Effect.fail( + new ProviderSetupError({ + instanceId, + operation: "refresh-models", + detail: driverError.detail, + cause: driverError, + }), + ), + ), + ), + }); + const responses = yield* Queue.unbounded(); + const receive = yield* Deferred.make[0]>(); + const protocol = yield* RpcServer.Protocol.make((write) => + Effect.gen(function* () { + yield* Deferred.succeed(receive, write); + const serialization = yield* RpcSerialization.RpcSerialization; + return { + disconnects: yield* Queue.unbounded(), + send: (_clientId, response) => + response._tag === "Exit" ? Queue.offer(responses, response) : Effect.void, + end: () => Effect.void, + clientIds: Effect.succeed(new Set([0])), + initialMessage: Effect.succeedNone, + supportsAck: true, + supportsTransferables: false, + supportsSpanPropagation: false, + supportsNotifications: true, + codecFor: serialization.codecFor, + }; + }), + ); + const services = Layer.mergeAll( + layerUnusedConfigRpcServices, + registryLayer, + Layer.mock(UsageLimitSources.UsageLimitSources)({ refresh: Effect.void }), + Layer.mock(ExternalLauncher.ExternalLauncher)({}), + Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), + Layer.mock(DirectEndpoints.DirectEndpoints)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), + Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), + Layer.mock(ServerEnvironment.ServerEnvironment)({}), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), + ); + yield* RpcServer.make(RpcGroup.make(WsServerRefreshProvidersRpc), WS_RPC_SERVER_OPTIONS).pipe( + Effect.provide( + layerWsRpc( + { + sessionId: AuthSessionId.make("model-refresh-test"), + subject: "model-refresh-test", + method: "browser-session-cookie", + scopes: [], + }, + {}, + yield* ServerBrowser.ServerBrowser, + () => Effect.void, + { secretReferences: true }, + ).pipe(Layer.provide(services)), + ), + Effect.provideService(RpcServer.Protocol, protocol), + Effect.forkScoped, + ); + const write = yield* Deferred.await(receive); + yield* write(0, { + _tag: "Request", + id: "refresh-models", + tag: WS_METHODS.serverRefreshProviders, + payload: { refreshModels: true, ...(target === "single" ? { instanceId: firstId } : {}) }, + headers: [], + }); + const encoded = yield* Queue.take(responses); + assert.equal(yield* Queue.take(requests), target === "single" ? firstId : undefined); + assert.notInclude(JSON.stringify(encoded), privateCode); + const response = yield* decodeModelRefreshExit(encoded.exit); + if (target === "single") { + assert.equal(response._tag, "Failure"); + if (response._tag === "Failure") { + const error = Cause.findErrorOption(response.cause).pipe(Option.getOrThrow); + assert.equal(error._tag, "ProviderSetupError"); + if (error._tag === "ProviderSetupError") { + assert.equal(error.instanceId, firstId); + assert.equal(error.operation, "refresh-models"); + assert.equal(error.detail, "Model refresh failed."); + assert.isUndefined(error.cause); + } + } + return; + } + assert.equal(response._tag, "Success"); + if (response._tag === "Success") { + assert.deepEqual(response.value, { providers }); + } + }).pipe( + Effect.provide( + Layer.mergeAll( + RpcSerialization.layerJson, + Layer.fresh(layerConfigSources), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ), +); + it.effect("a publish whose push fails still refreshes the new remote", () => Effect.gen(function* () { const refreshes = yield* Queue.unbounded(); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 035f8b1688d44..be000ef80ef1e 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2497,31 +2497,15 @@ export const layerWsRpc = ( ? providerRegistry.refreshInstance(input.instanceId) : providerRegistry.refresh(); if (input.refreshModels) { - const instances = yield* providerInstances.listInstances; - for (const instance of instances) { - if ( - !instance.refreshModels || - (input.instanceId !== undefined && input.instanceId !== instance.instanceId) || - !providers.some( - (provider) => - provider.instanceId === instance.instanceId && - provider.enabled && - provider.installed, - ) - ) - continue; - yield* instance.refreshModels().pipe( + // Native causes can contain OAuth URLs; keep them inside the provider service. + providers = yield* providerRegistry + .refreshModels(input.instanceId) + .pipe( Effect.mapError( - (error) => - new ProviderSetupError({ - instanceId: instance.instanceId, - operation: "refresh-models", - detail: error.detail, - }), + ({ instanceId, operation, detail }) => + new ProviderSetupError({ instanceId, operation, detail }), ), ); - providers = yield* providerRegistry.refreshInstance(instance.instanceId); - } } if (input.workspaceSnapshotsOnDemand !== true) return { providers }; const commandProviders =