From 234775647294939d35f6f47948f56a647a80fc4b Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 01:29:31 +0530 Subject: [PATCH 1/4] fix(server): isolate provider model refresh failures --- apps/server/src/provider/ProviderRegistry.ts | 57 ++++++ .../providerMaintenanceRunner.test.ts | 1 + .../testUtils/providerRegistryMock.ts | 1 + apps/server/src/ws.test.ts | 185 ++++++++++++++++++ apps/server/src/ws.ts | 26 +-- 5 files changed, 245 insertions(+), 25 deletions(-) diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index fb6aeaac26b63..c335ec83e9c57 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,57 @@ 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, + }), + ), + 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 +1136,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..e122c54d3b517 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -34,6 +34,7 @@ import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/http"; @@ -81,6 +82,10 @@ 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 * as ProviderInstanceRegistryMutator from "./provider/ProviderInstanceRegistryMutator.ts"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; +import { ProviderDriverError } from "@t3tools/provider-core/server/errors"; +import { makeManualOnlyProviderMaintenanceCapabilities } from "@t3tools/provider-core/server/maintenanceResolver"; import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; @@ -1346,6 +1351,186 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( ), ); +it.effect.each(["all", "single"] as const)( + "%s model refresh preserves cached models when an instance fails", + (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 thirdStarted = yield* Deferred.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, + ) { + const snapshot = yield* Ref.make({ + instanceId, + driver, + enabled: true, + 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: true, + snapshot: { + getSnapshot: Ref.get(snapshot), + refresh: Ref.get(snapshot), + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: null, + }), + ), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + refreshModels: () => + Effect.gen(function* () { + if (instanceId === firstId) { + return yield* new ProviderDriverError({ + driver, + instanceId, + detail: "Model refresh failed.", + }); + } + 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), + ]; + const changes = yield* PubSub.unbounded(); + 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: PubSub.subscribe(changes), + }); + const registryLayer = ProviderRegistry.layer.pipe( + Layer.provide( + Layer.mergeAll( + instanceLayer, + ModelManifest.layerTest, + Layer.mock(ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator)({}), + Layer.mock(ProviderLatestVersions.ProviderLatestVersions)({}), + ), + ), + ); + const responses = + yield* Queue.unbounded< + Exit.Exit<{ readonly providers: ReadonlyArray }, unknown> + >(); + const services = Layer.mergeAll( + layerUnusedConfigRpcServices, + instanceLayer, + 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)({}), + ); + const server = yield* RpcServer.makeNoSerialization( + RpcGroup.make(WsServerRefreshProvidersRpc), + { + ...WS_RPC_SERVER_OPTIONS, + onFromServer: (response) => + response._tag === "Exit" ? Queue.offer(responses, response.exit) : Effect.void, + }, + ).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)), + ), + ); + yield* server.write(0, { + _tag: "Request", + id: RpcMessage.RequestId("refresh-models"), + tag: WS_METHODS.serverRefreshProviders, + payload: { refreshModels: true, ...(target === "single" ? { instanceId: firstId } : {}) }, + headers: Headers.empty, + }); + const response = yield* Queue.take(responses); + if (target === "single") { + assert.equal(response._tag, "Failure"); + if (response._tag === "Failure") { + assert.match(Cause.pretty(response.cause), /Model refresh failed/); + } + assert.equal(yield* Ref.get(maxRefreshes), 0); + return; + } + assert.equal(response._tag, "Success"); + if (response._tag === "Success") { + assert.deepEqual( + Object.fromEntries( + response.value.providers.map((provider) => [ + provider.instanceId, + provider.models.map((entry) => entry.slug), + ]), + ), + { + [firstId]: ["cached-model"], + [secondId]: ["refreshed-model"], + [thirdId]: ["refreshed-model"], + }, + ); + assert.equal(yield* Ref.get(maxRefreshes), 2); + } + }).pipe( + Effect.provide( + Layer.mergeAll( + 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..971c60161279b 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2497,31 +2497,7 @@ 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( - Effect.mapError( - (error) => - new ProviderSetupError({ - instanceId: instance.instanceId, - operation: "refresh-models", - detail: error.detail, - }), - ), - ); - providers = yield* providerRegistry.refreshInstance(instance.instanceId); - } + providers = yield* providerRegistry.refreshModels(input.instanceId); } if (input.workspaceSnapshotsOnDemand !== true) return { providers }; const commandProviders = From 8322524ebbb3a984e58e7d7a3af5da6224b49ac9 Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 02:27:21 +0530 Subject: [PATCH 2/4] fix(server): retain model refresh error causes --- apps/server/src/provider/ProviderRegistry.ts | 1 + apps/server/src/ws.test.ts | 1 + 2 files changed, 2 insertions(+) diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index c335ec83e9c57..de3783710fef2 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -718,6 +718,7 @@ export const layer = Layer.effect( instanceId: instance.instanceId, operation: "refresh-models", detail: error.detail, + cause: error, }), ), Effect.andThen( diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index e122c54d3b517..2c0f2ff785635 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1498,6 +1498,7 @@ it.effect.each(["all", "single"] as const)( assert.equal(response._tag, "Failure"); if (response._tag === "Failure") { assert.match(Cause.pretty(response.cause), /Model refresh failed/); + assert.match(Cause.pretty(response.cause), /ProviderDriverError/); } assert.equal(yield* Ref.get(maxRefreshes), 0); return; From c9d556f03049a3cf595607969f44d4ce45a13e06 Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 05:12:24 +0530 Subject: [PATCH 3/4] fix(server): keep model refresh causes out of RPC responses --- apps/server/src/ws.test.ts | 78 +++++++++++++++++++++++++++----------- apps/server/src/ws.ts | 10 ++++- 2 files changed, 64 insertions(+), 24 deletions(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 2c0f2ff785635..564161f58ea8e 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -35,10 +35,11 @@ import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +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"; @@ -149,6 +150,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; @@ -1359,6 +1364,13 @@ it.effect.each(["all", "single"] as const)( const firstId = ProviderInstanceId.make("antigravity_first"); const secondId = ProviderInstanceId.make("antigravity_second"); const thirdId = ProviderInstanceId.make("antigravity_third"); + 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 thirdStarted = yield* Deferred.make(); const activeRefreshes = yield* Ref.make(0); const maxRefreshes = yield* Ref.make(0); @@ -1401,11 +1413,7 @@ it.effect.each(["all", "single"] as const)( refreshModels: () => Effect.gen(function* () { if (instanceId === firstId) { - return yield* new ProviderDriverError({ - driver, - instanceId, - detail: "Model refresh failed.", - }); + return yield* driverError; } const active = yield* Ref.updateAndGet(activeRefreshes, (count) => count + 1); yield* Ref.update(maxRefreshes, (count) => Math.max(count, active)); @@ -1446,10 +1454,35 @@ it.effect.each(["all", "single"] as const)( ), ), ); - const responses = - yield* Queue.unbounded< - Exit.Exit<{ readonly providers: ReadonlyArray }, unknown> - >(); + if (target === "single") { + const failure = yield* ProviderRegistry.ProviderRegistry.pipe( + Effect.flatMap((registry) => registry.refreshModels(firstId)), + Effect.flip, + Effect.provide(registryLayer), + ); + assert.strictEqual(failure.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, instanceLayer, @@ -1463,14 +1496,7 @@ it.effect.each(["all", "single"] as const)( Layer.mock(ServerEnvironment.ServerEnvironment)({}), Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), ); - const server = yield* RpcServer.makeNoSerialization( - RpcGroup.make(WsServerRefreshProvidersRpc), - { - ...WS_RPC_SERVER_OPTIONS, - onFromServer: (response) => - response._tag === "Exit" ? Queue.offer(responses, response.exit) : Effect.void, - }, - ).pipe( + yield* RpcServer.make(RpcGroup.make(WsServerRefreshProvidersRpc), WS_RPC_SERVER_OPTIONS).pipe( Effect.provide( layerWsRpc( { @@ -1485,20 +1511,25 @@ it.effect.each(["all", "single"] as const)( { secretReferences: true }, ).pipe(Layer.provide(services)), ), + Effect.provideService(RpcServer.Protocol, protocol), + Effect.forkScoped, ); - yield* server.write(0, { + const write = yield* Deferred.await(receive); + yield* write(0, { _tag: "Request", - id: RpcMessage.RequestId("refresh-models"), + id: "refresh-models", tag: WS_METHODS.serverRefreshProviders, payload: { refreshModels: true, ...(target === "single" ? { instanceId: firstId } : {}) }, - headers: Headers.empty, + headers: [], }); - const response = yield* Queue.take(responses); + const encoded = yield* Queue.take(responses); + assert.notInclude(JSON.stringify(encoded), privateCode); + const response = yield* decodeModelRefreshExit(encoded.exit); if (target === "single") { assert.equal(response._tag, "Failure"); if (response._tag === "Failure") { assert.match(Cause.pretty(response.cause), /Model refresh failed/); - assert.match(Cause.pretty(response.cause), /ProviderDriverError/); + assert.notMatch(Cause.pretty(response.cause), /ProviderDriverError/); } assert.equal(yield* Ref.get(maxRefreshes), 0); return; @@ -1523,6 +1554,7 @@ it.effect.each(["all", "single"] as const)( }).pipe( Effect.provide( Layer.mergeAll( + RpcSerialization.layerJson, Layer.fresh(layerConfigSources), Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), Layer.mock(ServerBrowser.ServerBrowser)({}), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 971c60161279b..be000ef80ef1e 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2497,7 +2497,15 @@ export const layerWsRpc = ( ? providerRegistry.refreshInstance(input.instanceId) : providerRegistry.refresh(); if (input.refreshModels) { - providers = yield* providerRegistry.refreshModels(input.instanceId); + // Native causes can contain OAuth URLs; keep them inside the provider service. + providers = yield* providerRegistry + .refreshModels(input.instanceId) + .pipe( + Effect.mapError( + ({ instanceId, operation, detail }) => + new ProviderSetupError({ instanceId, operation, detail }), + ), + ); } if (input.workspaceSnapshotsOnDemand !== true) return { providers }; const commandProviders = From 21042d4dbef1904753ace7c6a4e5e8f3a28b6a12 Mon Sep 17 00:00:00 2001 From: sidreyas Date: Sat, 7 Nov 2026 05:59:42 +0530 Subject: [PATCH 4/4] test(server): cover model refresh at the registry boundary --- .../src/provider/ProviderRegistry.test.ts | 162 ++++++++++++++++++ apps/server/src/ws.test.ts | 143 +++++----------- 2 files changed, 201 insertions(+), 104 deletions(-) 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/ws.test.ts b/apps/server/src/ws.test.ts index 564161f58ea8e..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,9 +33,9 @@ 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 Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -83,10 +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 * as ProviderInstanceRegistryMutator from "./provider/ProviderInstanceRegistryMutator.ts"; -import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; import { ProviderDriverError } from "@t3tools/provider-core/server/errors"; -import { makeManualOnlyProviderMaintenanceCapabilities } from "@t3tools/provider-core/server/maintenanceResolver"; import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; @@ -1357,13 +1355,11 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( ); it.effect.each(["all", "single"] as const)( - "%s model refresh preserves cached models when an instance fails", + "%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 secondId = ProviderInstanceId.make("antigravity_second"); - const thirdId = ProviderInstanceId.make("antigravity_third"); const privateCode = "private-model-refresh-oauth-code"; const driverError = new ProviderDriverError({ driver, @@ -1371,15 +1367,9 @@ it.effect.each(["all", "single"] as const)( detail: "Model refresh failed.", cause: new Error(`https://oauth.invalid/callback?code=${privateCode}`), }); - const thirdStarted = yield* Deferred.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, - ) { - const snapshot = yield* Ref.make({ - instanceId, + const providers: ReadonlyArray = [ + { + instanceId: firstId, driver, enabled: true, installed: true, @@ -1387,81 +1377,34 @@ it.effect.each(["all", "single"] as const)( status: "ready", auth: { status: "authenticated" }, checkedAt: "2026-10-25T00:00:00.000Z", - models: [model("cached-model")], + models: [{ slug: "model", name: "Model", isCustom: false, capabilities: null }], slashCommands: [], skills: [], - }); - return { - instanceId, - driverKind: driver, - continuationIdentity: { driverKind: driver, continuationKey: instanceId }, - displayName: undefined, - 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, - }, - refreshModels: () => - Effect.gen(function* () { - 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), + }, ]; - const changes = yield* PubSub.unbounded(); - const instanceLayer = Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { - listInstances: Effect.succeed(instances), - getInstance: (id) => - Effect.succeed(instances.find((instance) => instance.instanceId === id)), - listUnavailable: Effect.succeed([]), + 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, - subscribeChanges: PubSub.subscribe(changes), - }); - const registryLayer = ProviderRegistry.layer.pipe( - Layer.provide( - Layer.mergeAll( - instanceLayer, - ModelManifest.layerTest, - Layer.mock(ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator)({}), - Layer.mock(ProviderLatestVersions.ProviderLatestVersions)({}), + 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, + }), + ), + ), ), - ), - ); - if (target === "single") { - const failure = yield* ProviderRegistry.ProviderRegistry.pipe( - Effect.flatMap((registry) => registry.refreshModels(firstId)), - Effect.flip, - Effect.provide(registryLayer), - ); - assert.strictEqual(failure.cause, driverError); - } + }); const responses = yield* Queue.unbounded(); const receive = yield* Deferred.make[0]>(); const protocol = yield* RpcServer.Protocol.make((write) => @@ -1485,7 +1428,6 @@ it.effect.each(["all", "single"] as const)( ); const services = Layer.mergeAll( layerUnusedConfigRpcServices, - instanceLayer, registryLayer, Layer.mock(UsageLimitSources.UsageLimitSources)({ refresh: Effect.void }), Layer.mock(ExternalLauncher.ExternalLauncher)({}), @@ -1523,33 +1465,26 @@ it.effect.each(["all", "single"] as const)( 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") { - assert.match(Cause.pretty(response.cause), /Model refresh failed/); - assert.notMatch(Cause.pretty(response.cause), /ProviderDriverError/); + 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); + } } - assert.equal(yield* Ref.get(maxRefreshes), 0); return; } assert.equal(response._tag, "Success"); if (response._tag === "Success") { - assert.deepEqual( - Object.fromEntries( - response.value.providers.map((provider) => [ - provider.instanceId, - provider.models.map((entry) => entry.slug), - ]), - ), - { - [firstId]: ["cached-model"], - [secondId]: ["refreshed-model"], - [thirdId]: ["refreshed-model"], - }, - ); - assert.equal(yield* Ref.get(maxRefreshes), 2); + assert.deepEqual(response.value, { providers }); } }).pipe( Effect.provide(