diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 5e1bc9e962b5f..2c16cc546bff5 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -1476,6 +1476,143 @@ it.layer( ); }); + it.effect("invalidates model caches only for an explicit model refresh", () => + Effect.gen(function* () { + const forceRefreshCalls = yield* Ref.make(0); + const makeInstance = (name: string) => + Effect.gen(function* () { + const driver = ProviderDriverKind.make("codex"); + const instanceId = ProviderInstanceId.make(name); + const provider = { + instanceId, + driver, + status: "ready", + enabled: true, + installed: true, + auth: { status: "authenticated" }, + checkedAt: "2026-06-10T00:00:00.000Z", + version: "1.0.0", + models: [], + slashCommands: [], + skills: [], + } as const satisfies ServerProvider; + const invalidations = yield* Ref.make(0); + const maintenanceReads = yield* Ref.make>([]); + const instance = { + instanceId, + driverKind: driver, + continuationIdentity: { + driverKind: driver, + continuationKey: `codex:instance:${name}`, + }, + displayName: undefined, + enabled: true, + snapshot: { + resolveMaintenance: (options) => + Ref.update(maintenanceReads, (reads) => [...reads, options?.fresh === true]).pipe( + Effect.as( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driver, + packageName: `${name}-package`, + }), + ), + ), + getSnapshot: Effect.succeed(provider), + refresh: Effect.succeed(provider), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + invalidateCaches: Ref.update(invalidations, (count) => count + 1), + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + } satisfies ProviderInstance; + const takeMaintenanceReads = Ref.getAndSet(maintenanceReads, []); + return { instance, invalidations, takeMaintenanceReads }; + }); + const first = yield* makeInstance("codex_first"); + const second = yield* makeInstance("codex_second"); + const instances = [first.instance, second.instance]; + const layerInstanceRegistry = Layer.succeed( + ProviderInstanceRegistry.ProviderInstanceRegistry, + { + getInstance: (instanceId) => + Effect.succeed(instances.find((instance) => instance.instanceId === instanceId)), + listInstances: Effect.succeed(instances), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: Effect.flatMap(PubSub.unbounded(), PubSub.subscribe), + }, + ); + const layerModelManifest = Layer.succeed(ModelManifest.ModelManifest, { + current: Effect.succeed(ModelManifest.BUNDLED_MODEL_MANIFEST), + refresh: Effect.succeed(ModelManifest.BUNDLED_MODEL_MANIFEST), + forceRefresh: Ref.update(forceRefreshCalls, (count) => count + 1).pipe( + Effect.as(ModelManifest.BUNDLED_MODEL_MANIFEST), + ), + refreshInBackground: Effect.void, + }); + const layerLatestVersions = Layer.effect( + ProviderLatestVersions.ProviderLatestVersions, + ProviderLatestVersions.make([ + ["codex_first-package", "1.0.0"], + ["codex_second-package", "1.0.0"], + ]), + ); + const readCachedVersion = (key: string) => + ProviderLatestVersions.ProviderLatestVersions.use((latestVersions) => + latestVersions.cached(key, Effect.succeed("2.0.0")), + ); + + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + + yield* registry.refreshInstance(first.instance.instanceId); + yield* registry.refresh(); + assert.strictEqual(yield* Ref.get(forceRefreshCalls), 0); + assert.strictEqual(yield* Ref.get(first.invalidations), 0); + assert.strictEqual(yield* readCachedVersion("codex_first-package"), "1.0.0"); + yield* first.takeMaintenanceReads; + + yield* registry.invalidateModelCaches({ instanceId: first.instance.instanceId }); + // The instance refresh that follows resolves installer metadata fresh itself. + assert.deepStrictEqual(yield* first.takeMaintenanceReads, [false]); + assert.strictEqual(yield* Ref.get(forceRefreshCalls), 1); + assert.strictEqual(yield* Ref.get(first.invalidations), 1); + assert.strictEqual(yield* Ref.get(second.invalidations), 0); + assert.strictEqual(yield* readCachedVersion("codex_first-package"), "2.0.0"); + assert.strictEqual(yield* readCachedVersion("codex_second-package"), "1.0.0"); + + yield* registry.invalidateModelCaches({}); + assert.strictEqual(yield* Ref.get(first.invalidations), 2); + assert.strictEqual(yield* Ref.get(second.invalidations), 1); + assert.strictEqual(yield* readCachedVersion("codex_second-package"), "2.0.0"); + yield* first.takeMaintenanceReads; + + // Only a cached workspace refresh leaves the fresh resolution to this call. + const workspace = { instanceId: first.instance.instanceId, cwd: process.cwd() }; + yield* registry.invalidateModelCaches(workspace); + assert.deepStrictEqual(yield* first.takeMaintenanceReads, [true]); + yield* registry.invalidateModelCaches({ ...workspace, fresh: true }); + assert.deepStrictEqual(yield* first.takeMaintenanceReads, [false]); + }).pipe( + Effect.provide( + ProviderRegistry.layer.pipe( + Layer.provideMerge(layerInstanceRegistry), + Layer.provideMerge(layerModelManifest), + Layer.provideMerge(layerLatestVersions), + Layer.provide(layerUnusedMutator), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { + prefix: "t3-provider-registry-model-cache-", + }), + ), + Layer.provideMerge(NodeServices.layer), + ), + ), + ); + }), + ); + it.effect( "a second request for a cwd being scanned gets its result, even if the first leaves", () => diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index 3234bfd61960a..c1d6bb566930b 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -81,6 +81,17 @@ export class ProviderRegistry extends Context.Service< readonly fresh?: boolean; }) => Effect.Effect>; + /** + * Drop T3-owned model caches before an explicit model refresh. Pass the + * target of the refresh that follows. Background and ordinary refreshes + * keep their cache timers and must not call this. + */ + readonly invalidateModelCaches: (input: { + readonly instanceId?: ProviderInstanceId | undefined; + readonly cwd?: string | undefined; + readonly fresh?: boolean | undefined; + }) => Effect.Effect; + readonly getProviderMaintenanceCapabilitiesForInstance: ( instanceId: ProviderInstanceId, provider: ProviderDriverKind, @@ -1025,6 +1036,34 @@ export const layer = Layer.effect( return yield* Fiber.join(claim.scan); }); + const invalidateModelCaches = Effect.fn("ProviderRegistry.invalidateModelCaches")(function* ( + input: Parameters[0], + ) { + yield* manifestService.forceRefresh; + const instances = yield* instanceRegistry.listInstances; + yield* Effect.forEach( + instances.filter( + (instance) => input.instanceId === undefined || input.instanceId === instance.instanceId, + ), + (instance) => + Effect.gen(function* () { + yield* instance.invalidateCaches ?? Effect.void; + // Instance and full refreshes resolve installer metadata fresh + // themselves; a cached workspace refresh needs it resolved here. + const maintenance = yield* instance.snapshot.resolveMaintenance({ + fresh: + input.instanceId !== undefined && input.cwd !== undefined && input.fresh !== true, + }); + yield* Effect.forEach( + latestProviderVersionCacheKeys(maintenance), + latestVersions.invalidate, + { discard: true }, + ); + }), + { concurrency: "unbounded", discard: true }, + ); + }); + return { getProviders: Ref.get(providersRef), refresh: (provider?: ProviderDriverKind) => @@ -1036,6 +1075,7 @@ export const layer = Layer.effect( refreshWorkspaceSnapshot: (input) => refreshWorkspaceSnapshot(input).pipe(Effect.catchCause(recoverRefreshFailure)), refreshAfterUpdate, + invalidateModelCaches, getProviderMaintenanceCapabilitiesForInstance, setProviderMaintenanceActionState, get streamChanges() { diff --git a/apps/server/src/provider/providerMaintenanceRunner.test.ts b/apps/server/src/provider/providerMaintenanceRunner.test.ts index 6f3e49e4d4c46..77a44c7fa29a5 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.test.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.test.ts @@ -211,6 +211,7 @@ function makeRegistry( ); }, refreshWorkspaceSnapshot: () => Ref.get(providersRef), + invalidateModelCaches: () => Effect.void, getProviderMaintenanceCapabilitiesForInstance: (_instanceId, provider) => Effect.succeed(lifecycleFor(provider)), setProviderMaintenanceActionState, diff --git a/apps/server/src/provider/testUtils/providerRegistryMock.ts b/apps/server/src/provider/testUtils/providerRegistryMock.ts index a4008acc18efb..f75ecd141e71a 100644 --- a/apps/server/src/provider/testUtils/providerRegistryMock.ts +++ b/apps/server/src/provider/testUtils/providerRegistryMock.ts @@ -14,6 +14,7 @@ const makeProviderRegistryMock = ( reloadSecretsAndRefresh: () => Effect.succeed(providers), refreshAfterUpdate: () => Effect.succeed(providers), refreshWorkspaceSnapshot: () => Effect.succeed(providers), + invalidateModelCaches: () => Effect.void, getProviderMaintenanceCapabilitiesForInstance: (_instanceId, provider) => Effect.succeed(makeManualOnlyProviderMaintenanceCapabilities({ provider, packageName: null })), setProviderMaintenanceActionState: () => Effect.succeed(providers), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index cc06273187a9a..1cae6b63ee4fd 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -172,9 +172,6 @@ import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts"; import * as AcpRegistrySupport from "@t3tools/provider-acp-registry/server/AcpRegistrySupport"; import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; -import * as ModelManifest from "./provider/ModelManifest.ts"; -import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; -import { latestProviderVersionCacheKeys } from "@t3tools/provider-core/server/maintenanceResolver"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import * as ProviderAuthService from "./provider/ProviderAuthService.ts"; import { makeProviderInstallation } from "./provider/providerInstallation.ts"; @@ -1363,8 +1360,6 @@ export const layerWsRpc = ( const previewManager = yield* PreviewManager.PreviewManager; const portDiscovery = yield* PortScanner.PortDiscovery; const providerRegistry = yield* ProviderRegistry.ProviderRegistry; - const modelManifest = yield* ModelManifest.ModelManifest; - const providerLatestVersions = yield* ProviderLatestVersions.ProviderLatestVersions; const providerInstances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry; const acpRegistryCatalog = yield* AcpRegistrySupport.AcpRegistryCatalog; const acpRegistryRuntimeCoordinator = @@ -2330,30 +2325,11 @@ export const layerWsRpc = ( /* Request instrumentation already owns the span; keep this helper in that context. */ Effect.fnUntraced(function* (input) { if (input.refreshModels) { - yield* modelManifest.forceRefresh; - const instances = yield* providerInstances.listInstances; - yield* Effect.forEach( - instances.filter( - (instance) => - input.instanceId === undefined || input.instanceId === instance.instanceId, - ), - (instance) => - Effect.gen(function* () { - yield* instance.invalidateCaches ?? Effect.void; - const maintenance = yield* instance.snapshot.resolveMaintenance({ - fresh: - input.instanceId !== undefined && - input.cwd !== undefined && - input.fresh !== true, - }); - yield* Effect.forEach( - latestProviderVersionCacheKeys(maintenance), - providerLatestVersions.invalidate, - { discard: true }, - ); - }), - { concurrency: "unbounded", discard: true }, - ); + yield* providerRegistry.invalidateModelCaches({ + instanceId: input.instanceId, + cwd: input.cwd, + fresh: input.fresh, + }); } // Await refresh; RPC scope closure would interrupt a fork before the hub answered. if (input.instanceId === undefined) {