From a66964ee9fa151ab81f5f60405bc4e56c6480549 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Fri, 30 Oct 2026 20:35:16 -0500 Subject: [PATCH] refactor(core): share provider discovery through a simple cache (#60082) --- packages/core/src/github-copilot/models.ts | 32 ++-- packages/core/src/modal/models.ts | 29 ++-- packages/core/src/plugin/internal.ts | 3 + packages/core/src/plugin/provider/azure.ts | 69 ++++++-- .../core/src/plugin/provider/digitalocean.ts | 91 +++++----- .../src/plugin/provider/github-copilot.ts | 115 ++++++------- packages/core/src/plugin/provider/modal.ts | 97 ++++++----- packages/core/src/plugin/provider/refresh.ts | 70 ++++++++ packages/core/src/shared-cache.ts | 103 +++++++++++ .../core/test/github-copilot/models.test.ts | 21 ++- packages/core/test/modal/models.test.ts | 11 +- packages/core/test/plugin/fixture.ts | 2 + .../core/test/plugin/provider-azure.test.ts | 113 ++++++++---- .../test/plugin/provider-digitalocean.test.ts | 84 ++++++++- .../plugin/provider-github-copilot.test.ts | 70 +++++++- packages/core/test/shared-cache.test.ts | 161 ++++++++++++++++++ 16 files changed, 825 insertions(+), 246 deletions(-) create mode 100644 packages/core/src/plugin/provider/refresh.ts create mode 100644 packages/core/src/shared-cache.ts create mode 100644 packages/core/test/shared-cache.test.ts diff --git a/packages/core/src/github-copilot/models.ts b/packages/core/src/github-copilot/models.ts index 8af9fbf496f..f17770829a5 100644 --- a/packages/core/src/github-copilot/models.ts +++ b/packages/core/src/github-copilot/models.ts @@ -1,7 +1,8 @@ export * as CopilotModels from "./models.js" import { Money } from "@opencode/schema/money" -import { Option, Schema } from "effect" +import { Effect, Option, Schema } from "effect" +import { Headers, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" import { Model } from "../model.js" import { Provider } from "../provider.js" @@ -58,7 +59,6 @@ const RemoteModel = Schema.Struct({ }) const Response = Schema.Struct({ data: Schema.Array(Schema.Unknown) }) -const decodeResponse = Schema.decodeUnknownSync(Response) const decodeModel = Schema.decodeUnknownOption(RemoteModel) type RemoteModel = typeof RemoteModel.Type @@ -72,26 +72,32 @@ type UsableModel = RemoteModel & { } } -export async function get(baseURL: string, headers: RequestInit["headers"], existing: readonly Model.Info[]) { - return derive(baseURL, await load(baseURL, headers), existing) +export function get( + http: HttpClient.HttpClient, + baseURL: string, + headers: Headers.Input, + existing: readonly Model.Info[], +) { + return load(http, baseURL, headers).pipe(Effect.map((remote) => derive(baseURL, remote, existing))) } export type Snapshot = ReadonlyMap -export async function load(baseURL: string, headers: RequestInit["headers"]): Promise { - const response = await fetch(`${baseURL}/models`, { - headers, - signal: AbortSignal.timeout(5_000), - }) - if (!response.ok) throw new Error(`Failed to fetch Copilot models: ${response.status}`) - +export const load = Effect.fn("CopilotModels.load")(function* ( + http: HttpClient.HttpClient, + baseURL: string, + headers: Headers.Input, +) { + const response = yield* HttpClient.filterStatusOk(http) + .execute(HttpClientRequest.get(`${baseURL}/models`).pipe(HttpClientRequest.setHeaders(headers))) + .pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(Response)), Effect.timeout("5 seconds")) return new Map( - decodeResponse(await response.json()).data.flatMap((raw) => { + response.data.flatMap((raw) => { const model = Option.getOrUndefined(decodeModel(raw)) return model && usable(model) ? ([[model.id, model]] as const) : [] }), ) -} +}) /** Combine remote facts with source templates, never a previously transformed model result. */ export function derive(baseURL: string, remote: Snapshot, existing: readonly Model.Info[]) { diff --git a/packages/core/src/modal/models.ts b/packages/core/src/modal/models.ts index 2832117492a..05e8975fa16 100644 --- a/packages/core/src/modal/models.ts +++ b/packages/core/src/modal/models.ts @@ -1,7 +1,8 @@ export * as ModalModels from "./models.js" import { Money } from "@opencode/schema/money" -import { Option, Schema } from "effect" +import { Effect, Option, Schema } from "effect" +import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" import { Model } from "../model.js" import { Provider } from "../provider.js" @@ -42,33 +43,31 @@ const RemoteModel = Schema.Struct({ }) const Response = Schema.Struct({ data: Schema.Array(Schema.Unknown) }) -const decodeResponse = Schema.decodeUnknownSync(Response) const decodeModel = Schema.decodeUnknownOption(RemoteModel) type RemoteModel = typeof RemoteModel.Type -export async function get(baseURL: string, apiKey: string, existing: readonly Model.Info[]) { - return derive(baseURL, await load(baseURL, apiKey), existing) +export function get(http: HttpClient.HttpClient, baseURL: string, apiKey: string, existing: readonly Model.Info[]) { + return load(http, baseURL, apiKey).pipe(Effect.map((remote) => derive(baseURL, remote, existing))) } export type Snapshot = readonly RemoteModel[] -export async function load(baseURL: string, apiKey: string): Promise { - const response = await fetch(`${baseURL.replace(/\/+$/, "")}/models`, { - headers: { - Authorization: `Bearer ${apiKey}`, - }, - signal: AbortSignal.timeout(3_000), - }) - if (!response.ok) throw new Error(`Failed to fetch Modal models: ${response.status}`) - +export const load = Effect.fn("ModalModels.load")(function* ( + http: HttpClient.HttpClient, + baseURL: string, + apiKey: string, +) { + const response = yield* HttpClient.filterStatusOk(http) + .execute(HttpClientRequest.get(`${baseURL.replace(/\/+$/, "")}/models`).pipe(HttpClientRequest.bearerToken(apiKey))) + .pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(Response)), Effect.timeout("3 seconds")) // Decode each item tolerantly so one malformed entry cannot discard the // whole inventory. A malformed envelope still fails the fetch. - return decodeResponse(await response.json()).data.flatMap((raw) => { + return response.data.flatMap((raw) => { const model = Option.getOrUndefined(decodeModel(raw)) return model ? [model] : [] }) -} +}) /** Combine remote facts with source templates, never a previously transformed model result. */ export function derive(baseURL: string, remote: Snapshot, existing: readonly Model.Info[]) { diff --git a/packages/core/src/plugin/internal.ts b/packages/core/src/plugin/internal.ts index 7db8143c246..df14b9908b4 100644 --- a/packages/core/src/plugin/internal.ts +++ b/packages/core/src/plugin/internal.ts @@ -64,6 +64,7 @@ import { Session } from "../session.js" import { SessionCompaction } from "../session/compaction.js" import { SessionInstructions } from "../session/instructions.js" import { SubagentJob } from "../session/subagent-job.js" +import { SharedCache } from "../shared-cache.js" import { Shell } from "../shell.js" import { ShellSelect } from "../shell/select.js" import { Snapshot } from "../snapshot.js" @@ -154,6 +155,7 @@ const services = [ SessionCompaction.Service, SessionInstructions.Service, SubagentJob.Service, + SharedCache.Service, Shell.Service, ShellSelect.Service, Snapshot.Service, @@ -210,6 +212,7 @@ export const requirements = LayerNode.group([ SessionCompaction.node, SessionInstructions.node, SubagentJob.node, + SharedCache.node, Shell.node, ShellSelect.node, Snapshot.node, diff --git a/packages/core/src/plugin/provider/azure.ts b/packages/core/src/plugin/provider/azure.ts index 1df4bfc1bbc..5144b03a92b 100644 --- a/packages/core/src/plugin/provider/azure.ts +++ b/packages/core/src/plugin/provider/azure.ts @@ -1,4 +1,4 @@ -import { Clock, Effect, FiberHandle, Option, Schema, Semaphore, Stream } from "effect" +import { Clock, Effect, Equal, FiberHandle, Option, Schema, Semaphore, Stream } from "effect" import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" import { ChildProcess } from "effect/process" import { define } from "@opencode/plugin/effect/plugin" @@ -11,9 +11,11 @@ import { Integration } from "../../integration.js" import { IntegrationConnection } from "../../integration/connection.js" import { Model } from "../../model.js" import { Provider } from "../../provider.js" +import { SharedCache } from "../../shared-cache.js" import { which } from "../../util/which.js" import type { PluginInternal } from "../internal.js" import { configuredSettings } from "./configured.js" +import { ProviderRefresh } from "./refresh.js" const cognitiveScope = "https://cognitiveservices.azure.com/.default" const foundryScope = "https://ai.azure.com/.default" @@ -67,6 +69,7 @@ export function make( const bus = yield* Bus.Service const credentials = yield* Credential.Service const providers = yield* Provider.Service + const cache = yield* SharedCache.Service const http = HttpClient.filterStatusOk(yield* HttpClient.HttpClient) const tokens = new Map() // A resource keeps its Azure Resource Manager ID until it is deleted, so discovery looks it up once. @@ -259,7 +262,9 @@ export function make( // https://github.com/Azure/azure-rest-api-specs/blob/main/specification/cognitiveservices/data-plane/OpenAIAuthoring/stable/2022-12-01/azureopenai.json const deployments = (url: string, resource: string, credential: Credential.Key | Credential.External) => credential.type === "external" - ? managementDeployments(resource).pipe(Effect.catch(() => resourceDeployments(url, credential))) + ? managementDeployments(resource).pipe( + Effect.catch((error) => resourceDeployments(url, credential).pipe(Effect.mapError(() => error))), + ) : resourceDeployments(url, credential) // Local and quick, so a switch rebinds the provider before discovery for the new connection calls Azure. @@ -277,7 +282,7 @@ export function make( }), ) - const discover = Effect.fn("AzurePlugin.discover")(function* () { + const discover = Effect.fn("AzurePlugin.discover")(function* (options: ProviderRefresh.Options) { const connection = loaded.connection const settings = (yield* providers.get(Provider.ID.azure))?.settings const name = loaded.resource ?? resolveResourceName(settings) @@ -305,21 +310,38 @@ export function make( (credential.type === "external" && credential.methodID !== methodID) ) return - const found = yield* deployments(url, name, credential).pipe( - // Azure promises no order; normalize it so a reordered response does not rebuild the model list. - Effect.map((list) => list.toSorted((a, b) => a.name.localeCompare(b.name))), - Effect.catch((cause) => - Effect.logWarning("failed to sync Azure deployments", { cause }).pipe(Effect.as(undefined)), - ), - ) - if (!found) return + const found = yield* cache + .get({ + key: deploymentsKey(connection, credential, url), + ...options, + load: ctx.integration.connection.resolve(connection).pipe( + Effect.flatMap((current) => + !current || current.type === "oauth" || (current.type === "external" && current.methodID !== methodID) + ? Effect.fail(new Error("Azure connection has no usable credential")) + : deployments(url, name, current), + ), + // Azure promises no order; normalize it so a reordered response does not rebuild the model list. + Effect.map((list) => list.toSorted((a, b) => a.name.localeCompare(b.name))), + ), + }) + .pipe(Effect.orElseSucceed(() => undefined)) + return found && { connection, found } + }) + + const apply = (next: { connection: typeof loaded.connection; found: readonly Deployment[] } | undefined) => + next ? loading.withPermit(publish(next.connection, next.found)).pipe(Effect.uninterruptible) : Effect.void + + const publish = Effect.fn("AzurePlugin.publish")(function* ( + connection: typeof loaded.connection, + found: readonly Deployment[], + ) { if ( loaded.connection !== connection || IntegrationConnection.key(connection) !== IntegrationConnection.key(yield* ctx.integration.connection.active(Provider.ID.azure)) ) return - if (JSON.stringify(found) === JSON.stringify(loaded.deployments)) return + if (Equal.equals(found, loaded.deployments)) return const catalog = new Map( Array.from((yield* providers.snapshot()).records.get(Provider.ID.azure)?.models.keys() ?? [], (id) => [ id.toLowerCase(), @@ -335,7 +357,7 @@ export function make( yield* ctx.model.reload() }) - const refresh = () => rebind().pipe(Effect.andThen(FiberHandle.run(discovery, discover()))) + const refresh = discover({}).pipe(Effect.flatMap(apply)) // The connection's resource wins over configured settings for Azure itself. const resourceFor = (provider: Provider.Info) => @@ -417,12 +439,18 @@ export function make( // A switch interrupts discovery for the previous connection instead of waiting for its Azure calls. yield* bus.subscribe(Credential.Event.Switched).pipe( Stream.filter((event) => event.data.integrationID === Integration.ID.make("azure")), - Stream.runForEach(() => refresh()), + Stream.runForEach(() => + rebind().pipe(Effect.andThen(FiberHandle.run(discovery, ProviderRefresh.switched(discover, apply)))), + ), Effect.forkScoped({ startImmediately: true }), ) // Deployments load in the background so startup never waits on Azure; the catalog serves until they arrive. - // Later changes load when the connection changes, so a new deployment needs a reconnect or restart. - yield* refresh().pipe(Effect.forkScoped) + yield* rebind().pipe(Effect.andThen(FiberHandle.run(discovery, refresh)), Effect.forkScoped) + yield* Effect.sleep(SharedCache.interval).pipe( + Effect.andThen(FiberHandle.run(discovery, refresh, { onlyIfMissing: true })), + Effect.forever, + Effect.forkScoped, + ) // Entra bearer tokens are minted per request from the target URL's scope, so they are injected // at the transport hooks rather than stored as a credential. @@ -474,6 +502,15 @@ export function make( export const AzurePlugin = make() +function deploymentsKey( + connection: Parameters[0], + credential: Credential.Key | Credential.External, + url: string, +) { + const access = credential.type === "key" ? `:${SharedCache.fingerprint(credential.key)}` : "" + return `azure:${IntegrationConnection.key(connection)}${access}:${url}` +} + function resolveResourceName(settings: Readonly> | undefined, fallback?: string) { const configured = settings?.resourceName if (typeof configured === "string" && configured.trim() !== "") return configured diff --git a/packages/core/src/plugin/provider/digitalocean.ts b/packages/core/src/plugin/provider/digitalocean.ts index b0083c59aae..8a58d119d18 100644 --- a/packages/core/src/plugin/provider/digitalocean.ts +++ b/packages/core/src/plugin/provider/digitalocean.ts @@ -1,16 +1,18 @@ import type { IntegrationOAuthMethodRegistration } from "@opencode/plugin/effect/integration" import { define } from "@opencode/plugin/effect/plugin" -import { Clock, Deferred, Effect, Option, Schema, Semaphore, Stream } from "effect" +import { Clock, Deferred, Effect, Option, Schema } from "effect" import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" import type { Server } from "node:http" import { App } from "../../app.js" -import { Bus } from "../../bus.js" import { Credential } from "../../credential.js" import { Integration } from "../../integration.js" +import { IntegrationConnection } from "../../integration/connection.js" import { Model } from "../../model.js" import { OauthCallbackPage } from "../../oauth/page.js" import { Provider } from "../../provider.js" +import { SharedCache } from "../../shared-cache.js" import type { PluginInternal } from "../internal.js" +import { ProviderRefresh } from "./refresh.js" const providerID = Provider.ID.make("digitalocean") const integrationID = Integration.ID.make("digitalocean") @@ -25,6 +27,7 @@ const tokenPath = "/auth/token" const RoutersResponse = Schema.Struct({ model_routers: Schema.Array(Schema.Struct({ name: Schema.NonEmptyString })), }) +type Routers = (typeof RoutersResponse.Type)["model_routers"] const decodeCallback = Schema.decodeUnknownOption( Schema.fromJsonString( Schema.Struct({ @@ -114,42 +117,47 @@ const oauth = { export const DigitalOceanPlugin = define({ id: "opencode.provider.digitalocean", effect: Effect.fn(function* (ctx) { - const bus = yield* Bus.Service + const cache = yield* SharedCache.Service const http = HttpClient.filterStatusOk(yield* HttpClient.HttpClient) - const loading = Semaphore.makeUnsafe(1) - const loaded: { access?: string; routers: (typeof RoutersResponse.Type)["model_routers"] } = { routers: [] } - - const load = Effect.fn("DigitalOceanPlugin.load")(function* () { - const connection = yield* ctx.integration.connection.active(integrationID) - const credential = connection - ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) - : undefined - if (credential?.type !== "oauth" || credential.methodID !== methodID) { - loaded.access = undefined - loaded.routers = [] - return - } - if (loaded.access !== credential.access) { - loaded.access = credential.access - loaded.routers = [] - } - if (credential.expires !== 0 && credential.expires <= (yield* Clock.currentTimeMillis)) return - const result = yield* http - .execute( - HttpClientRequest.get(routersEndpoint).pipe( - HttpClientRequest.bearerToken(credential.access), - HttpClientRequest.acceptJson, - HttpClientRequest.setHeader("User-Agent", App.useragent(ctx.app)), - ), - ) - .pipe( - Effect.flatMap(HttpClientResponse.schemaBodyJson(RoutersResponse)), - Effect.timeout("10 seconds"), - Effect.catch((cause) => - Effect.logWarning("failed to sync DigitalOcean routers", { cause }).pipe(Effect.as(undefined)), - ), - ) - if (result) loaded.routers = result.model_routers + const discovered = yield* ProviderRefresh.start({ + integrationID, + initial: { routers: [] } as { readonly key?: string; readonly routers: Routers }, + active: ctx.integration.connection.active(integrationID), + reload: ctx.provider.reload(), + load: Effect.fn("DigitalOceanPlugin.load")(function* (current, options) { + const connection = yield* ctx.integration.connection.active(integrationID) + const credential = connection + ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) + : undefined + if (!connection || credential?.type !== "oauth" || credential.methodID !== methodID) + return { connection, value: { routers: [] } } + const key = `digitalocean:${IntegrationConnection.key(connection)}` + // An expired token cannot be refreshed, so it keeps the routers already loaded for the connection. + if (credential.expires !== 0 && credential.expires <= (yield* Clock.currentTimeMillis)) + return { connection, value: current.key === key ? current : { key, routers: [] } } + const routers = yield* cache + .get({ + key, + ...options, + load: Effect.gen(function* () { + const latest = yield* ctx.integration.connection.resolve(connection) + if (latest?.type !== "oauth" || latest.methodID !== methodID) + return yield* Effect.fail(new Error("DigitalOcean connection has no OAuth credential")) + const response = yield* http + .execute( + HttpClientRequest.get(routersEndpoint).pipe( + HttpClientRequest.bearerToken(latest.access), + HttpClientRequest.acceptJson, + HttpClientRequest.setHeader("User-Agent", App.useragent(ctx.app)), + ), + ) + .pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(RoutersResponse)), Effect.timeout("10 seconds")) + return response.model_routers.toSorted((a, b) => a.name.localeCompare(b.name)) + }), + }) + .pipe(Effect.orElseSucceed((): Routers => [])) + return { connection, value: { key, routers } } + }), }) yield* ctx.integration.transform((draft) => { @@ -157,7 +165,7 @@ export const DigitalOceanPlugin = define({ }) yield* ctx.provider.transform((evt) => { if (!evt.get(providerID)) return - for (const router of loaded.routers) { + for (const router of discovered.current.routers) { const id = `router:${router.name}` if (evt.get(providerID)?.models.has(Model.ID.make(id))) continue evt.models.update(providerID, id, (draft) => { @@ -168,13 +176,6 @@ export const DigitalOceanPlugin = define({ }) } }) - const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.provider.reload()))) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === integrationID), - Stream.runForEach(refresh), - Effect.forkScoped({ startImmediately: true }), - ) - yield* refresh().pipe(Effect.forkScoped) }), } satisfies PluginInternal.InternalPlugin) diff --git a/packages/core/src/plugin/provider/github-copilot.ts b/packages/core/src/plugin/provider/github-copilot.ts index 0cb15e010c8..61f136bf96a 100644 --- a/packages/core/src/plugin/provider/github-copilot.ts +++ b/packages/core/src/plugin/provider/github-copilot.ts @@ -1,10 +1,10 @@ import type { IntegrationOAuthMethodRegistration } from "@opencode/plugin/effect/integration" import type { SessionRequestKind, SessionTitle } from "@opencode/plugin/effect/session" import { isRecord } from "@opencode/util/record" -import { Effect, Option, Schema, Semaphore, Stream } from "effect" +import { Effect, Option, Schema } from "effect" +import { HttpClient } from "effect/http" import { IntegrationConnection } from "../../integration/connection.js" import { Credential } from "../../credential.js" -import { Bus } from "../../bus.js" import { CopilotModels } from "../../github-copilot/models.js" import { App } from "../../app.js" import { Integration } from "../../integration.js" @@ -13,6 +13,8 @@ import { Agent } from "../../agent.js" import { define } from "@opencode/plugin/effect/plugin" import { Provider } from "../../provider.js" import { SessionAffinity } from "../../session/affinity.js" +import { SharedCache } from "../../shared-cache.js" +import { ProviderRefresh } from "./refresh.js" import type { PluginInternal } from "../internal.js" const clientID = "Ov23li8tweQw6odWQebz" @@ -161,53 +163,46 @@ export const GithubCopilotPlugin = define({ id: "opencode.provider.github.copilot", effect: Effect.fn(function* (ctx) { const providers = yield* Provider.Service - const bus = yield* Bus.Service - const loading = Semaphore.makeUnsafe(1) - const loaded: { - baseURL?: string - token?: string - models?: CopilotModels.Snapshot - connection?: Effect.Success> - } = {} - - const load = Effect.fn("GithubCopilotPlugin.load")(function* () { - const connection = yield* ctx.integration.connection.active("github-copilot") - const credential = connection - ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) - : undefined - if (credential?.type !== "oauth") { - loaded.baseURL = undefined - loaded.token = undefined - loaded.models = undefined - loaded.connection = undefined - return - } - - const url = copilotBaseURL(credential.metadata) ?? baseURL() - const provider = yield* providers.get(Provider.ID.githubCopilot) - const remote = yield* Effect.tryPromise({ - try: () => - CopilotModels.load(url, { - ...provider?.headers, - Authorization: `Bearer ${credential.refresh}`, - "User-Agent": App.useragent(ctx.app), - "X-GitHub-Api-Version": apiVersion, - }), - catch: (cause) => cause, - }).pipe( - Effect.catch((cause) => - Effect.logWarning("failed to sync GitHub Copilot models", { cause }).pipe(Effect.as(undefined)), - ), - ) - if ( - IntegrationConnection.key(connection) !== - IntegrationConnection.key(yield* ctx.integration.connection.active("github-copilot")) - ) - return - loaded.baseURL = url - loaded.token = credential.refresh - loaded.models = remote - loaded.connection = connection + const cache = yield* SharedCache.Service + const http = yield* HttpClient.HttpClient + const discovered = yield* ProviderRefresh.start({ + integrationID: "github-copilot", + initial: {} as { + readonly baseURL?: string + readonly token?: string + readonly models?: CopilotModels.Snapshot + readonly connection?: Effect.Success> + }, + active: ctx.integration.connection.active("github-copilot"), + reload: ctx.provider.reload(), + load: Effect.fn("GithubCopilotPlugin.load")(function* (_, options) { + const connection = yield* ctx.integration.connection.active("github-copilot") + const credential = connection + ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) + : undefined + if (!connection || credential?.type !== "oauth") return { connection, value: {} } + const url = copilotBaseURL(credential.metadata) ?? baseURL() + const headers = (yield* providers.get(Provider.ID.githubCopilot))?.headers ?? {} + const models = yield* cache + .get({ + key: modelsKey(connection, url, headers), + ...options, + load: ctx.integration.connection.resolve(connection).pipe( + Effect.flatMap((current) => + current?.type === "oauth" + ? CopilotModels.load(http, url, { + ...headers, + Authorization: `Bearer ${current.refresh}`, + "User-Agent": App.useragent(ctx.app), + "X-GitHub-Api-Version": apiVersion, + }) + : Effect.fail(new Error("GitHub Copilot connection has no OAuth credential")), + ), + ), + }) + .pipe(Effect.orElseSucceed(() => undefined)) + return { connection, value: { baseURL: url, token: credential.refresh, models, connection } } + }), }) yield* ctx.integration.transform((editor) => { @@ -217,6 +212,7 @@ export const GithubCopilotPlugin = define({ yield* ctx.provider.transform((evt) => { const item = evt.get(Provider.ID.githubCopilot) if (!item) return + const loaded = discovered.current if (loaded.models) { evt.add({ info: item.provider, @@ -255,13 +251,6 @@ export const GithubCopilotPlugin = define({ }) } }) - const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.provider.reload()))) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === Integration.ID.make("github-copilot")), - Stream.runForEach(refresh), - Effect.forkScoped({ startImmediately: true }), - ) - yield* refresh().pipe(Effect.forkScoped) yield* ctx.aisdk.hook( "sdk", Effect.fn(function* (evt) { @@ -299,8 +288,11 @@ export const GithubCopilotPlugin = define({ (evt) => Effect.gen(function* () { if (evt.model.providerID !== Provider.ID.githubCopilot) return + const loaded = discovered.current if (!loaded.baseURL || !loaded.token || !loaded.models) return - const agent = yield* ctx.agent.get({ agentID: Agent.ID.make("title") }).pipe(Effect.orElseSucceed(() => undefined)) + const agent = yield* ctx.agent + .get({ agentID: Agent.ID.make("title") }) + .pipe(Effect.orElseSucceed(() => undefined)) if (agent?.data.model) return const model = utilityTitleModels.find((id) => loaded.models?.has(id)) if (!model) return @@ -358,6 +350,15 @@ export const GithubCopilotPlugin = define({ }), } satisfies PluginInternal.InternalPlugin) +function modelsKey( + connection: Parameters[0], + url: string, + headers: Record, +) { + const fingerprint = SharedCache.fingerprint(JSON.stringify(Object.entries(headers).toSorted())) + return `github-copilot:${IntegrationConnection.key(connection)}:${url}:${fingerprint}` +} + function normalizeDomain(input: string) { return input.replace(/^https?:\/\//, "").replace(/\/$/, "") } diff --git a/packages/core/src/plugin/provider/modal.ts b/packages/core/src/plugin/provider/modal.ts index aa47e3d77e5..a4e9905f765 100644 --- a/packages/core/src/plugin/provider/modal.ts +++ b/packages/core/src/plugin/provider/modal.ts @@ -1,12 +1,12 @@ -import { Effect, Semaphore, Stream } from "effect" +import { Effect } from "effect" +import { HttpClient } from "effect/http" import { define } from "@opencode/plugin/effect/plugin" -import { Bus } from "../../bus.js" import { IntegrationConnection } from "../../integration/connection.js" -import { Credential } from "../../credential.js" -import { Integration } from "../../integration.js" import { ModalModels } from "../../modal/models.js" import { Provider } from "../../provider.js" +import { SharedCache } from "../../shared-cache.js" import type { PluginInternal } from "../internal.js" +import { ProviderRefresh } from "./refresh.js" const providerID = Provider.ID.make("modal") @@ -14,48 +14,49 @@ export const ModalPlugin = define({ id: "opencode.provider.modal", effect: Effect.fn(function* (ctx) { const providers = yield* Provider.Service - const bus = yield* Bus.Service - const loading = Semaphore.makeUnsafe(1) - const loaded: { - baseURL?: string - models?: ModalModels.Snapshot - connection?: Effect.Success> - } = {} - - const load = Effect.fn("ModalPlugin.load")(function* () { - const connection = yield* ctx.integration.connection.active("modal") - const credential = connection - ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) - : undefined - const apiKey = credential?.type === "key" ? credential.key : process.env.MODAL_PROXY_TOKEN - const provider = yield* providers.get(providerID) - const baseURL = typeof provider?.settings?.baseURL === "string" ? provider.settings.baseURL : undefined - if (!apiKey || !baseURL) { - loaded.baseURL = undefined - loaded.models = undefined - loaded.connection = undefined - return - } - const remote = yield* Effect.tryPromise({ - try: () => ModalModels.load(baseURL, apiKey), - catch: (cause) => cause, - }).pipe( - Effect.catch((cause) => Effect.logWarning("failed to sync Modal models", { cause }).pipe(Effect.as(undefined))), - ) - if ( - IntegrationConnection.key(connection) !== - IntegrationConnection.key(yield* ctx.integration.connection.active("modal")) - ) - return - loaded.baseURL = baseURL - loaded.models = remote - loaded.connection = connection + const cache = yield* SharedCache.Service + const http = yield* HttpClient.HttpClient + const discovered = yield* ProviderRefresh.start({ + integrationID: "modal", + initial: {} as { + readonly baseURL?: string + readonly models?: ModalModels.Snapshot + readonly connection?: Effect.Success> + }, + active: ctx.integration.connection.active("modal"), + reload: ctx.provider.reload(), + load: Effect.fn("ModalPlugin.load")(function* (_, options) { + const connection = yield* ctx.integration.connection.active("modal") + const credential = connection + ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) + : undefined + const apiKey = credential?.type === "key" ? credential.key : process.env.MODAL_PROXY_TOKEN + const provider = yield* providers.get(providerID) + const baseURL = typeof provider?.settings?.baseURL === "string" ? provider.settings.baseURL : undefined + if (!apiKey || !baseURL) return { connection, value: {} } + const models = yield* cache + .get({ + key: modelsKey(connection, apiKey, baseURL), + ...options, + load: (connection ? ctx.integration.connection.resolve(connection) : Effect.undefined).pipe( + Effect.map((current) => + current?.type === "key" ? current.key : connection ? undefined : process.env.MODAL_PROXY_TOKEN, + ), + Effect.flatMap((key) => + key ? ModalModels.load(http, baseURL, key) : Effect.fail(new Error("Modal connection has no API key")), + ), + Effect.map((models) => models.toSorted((a, b) => a.id.localeCompare(b.id))), + ), + }) + .pipe(Effect.orElseSucceed(() => undefined)) + return { connection, value: { baseURL, models, connection } } + }), }) yield* ctx.provider.transform((evt) => { const item = evt.get(providerID) - if (!item) return - if (!loaded.models || !loaded.baseURL) return + const loaded = discovered.current + if (!item || !loaded.models || !loaded.baseURL) return evt.add({ info: item.provider, models: Array.from( @@ -64,12 +65,10 @@ export const ModalPlugin = define({ sourceConnection: loaded.connection, }) }) - const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.provider.reload()))) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === Integration.ID.make("modal")), - Stream.runForEach(refresh), - Effect.forkScoped({ startImmediately: true }), - ) - yield* refresh().pipe(Effect.forkScoped) }), } satisfies PluginInternal.InternalPlugin) + +function modelsKey(connection: Parameters[0], apiKey: string, baseURL: string) { + const access = IntegrationConnection.key(connection) ?? "env:MODAL_PROXY_TOKEN" + return `modal:${access}:${SharedCache.fingerprint(apiKey)}:${baseURL.replace(/\/+$/, "")}` +} diff --git a/packages/core/src/plugin/provider/refresh.ts b/packages/core/src/plugin/provider/refresh.ts new file mode 100644 index 00000000000..cd9c78004b1 --- /dev/null +++ b/packages/core/src/plugin/provider/refresh.ts @@ -0,0 +1,70 @@ +export * as ProviderRefresh from "./refresh.js" + +import { Duration, Effect, Equal, Fiber, FiberHandle, Scope, Semaphore, Stream } from "effect" +import { Bus } from "../../bus.js" +import { Credential } from "../../credential.js" +import { IntegrationConnection } from "../../integration/connection.js" +import { SharedCache } from "../../shared-cache.js" + +type Connection = Parameters[0] + +/** How a load asks `SharedCache.get` for its list. */ +export type Options = { readonly interval?: Duration.Input; readonly force?: boolean } + +/** + * After an account switch, applies the account's list from the cache at once, however old, while a forced load runs + * alongside and is applied when it finishes. With nothing cached the first step joins the forced load, so one request + * serves both; after a failed load the forced load is the retry. + */ +export const switched = (load: (options: Options) => Effect.Effect, apply: (next: A) => Effect.Effect) => + Effect.gen(function* () { + const fresh = yield* load({ force: true }).pipe(Effect.forkChild({ startImmediately: true })) + yield* load({ interval: Duration.infinity }).pipe(Effect.flatMap(apply)) + yield* Fiber.join(fresh).pipe(Effect.flatMap(apply)) + }) + +/** + * Keeps a Location's copy of a provider's shared list current: loads in the background at startup, on an account + * switch (see `switched`) and every `interval`, and reloads providers only when the value changed. `load` passes the + * options it gets to `SharedCache.get`. A switch replaces a waiting load. A value loaded for a connection that is no + * longer active is dropped. + */ +export const start = Effect.fnUntraced(function* (input: { + readonly integrationID: string + readonly initial: A + readonly interval?: Duration.Input + readonly active: Effect.Effect + readonly load: (current: A, options: Options) => Effect.Effect<{ readonly connection: Connection; readonly value: A }> + readonly reload: Effect.Effect +}): Effect.fn.Return<{ readonly current: A }, never, Bus.Service | Scope.Scope> { + const bus = yield* Bus.Service + const interval = input.interval ?? SharedCache.interval + const lock = Semaphore.makeUnsafe(1) + const loading = yield* FiberHandle.make() + const state = { current: input.initial } + const apply = (next: { readonly connection: Connection; readonly value: A }) => + lock + .withPermit( + Effect.gen(function* () { + if (IntegrationConnection.key(next.connection) !== IntegrationConnection.key(yield* input.active)) return + if (Equal.equals(next.value, state.current)) return + state.current = next.value + yield* input.reload + }), + ) + .pipe(Effect.uninterruptible) + const load = (options: Options) => Effect.suspend(() => input.load(state.current, { interval, ...options })) + const refresh = load({}).pipe(Effect.flatMap(apply)) + yield* bus.subscribe(Credential.Event.Switched).pipe( + Stream.filter((event) => event.data.integrationID === input.integrationID), + Stream.runForEach(() => FiberHandle.run(loading, switched(load, apply))), + Effect.forkScoped({ startImmediately: true }), + ) + yield* FiberHandle.run(loading, refresh, { onlyIfMissing: true }).pipe(Effect.forkScoped) + yield* Effect.sleep(interval).pipe( + Effect.andThen(FiberHandle.run(loading, refresh, { onlyIfMissing: true })), + Effect.forever, + Effect.forkScoped, + ) + return state +}) diff --git a/packages/core/src/shared-cache.ts b/packages/core/src/shared-cache.ts new file mode 100644 index 00000000000..3beca4bdf25 --- /dev/null +++ b/packages/core/src/shared-cache.ts @@ -0,0 +1,103 @@ +export * as SharedCache from "./shared-cache.js" + +import { Cause, Clock, Context, Duration, Effect, Exit, Fiber, FiberSet, Layer, Schedule } from "effect" +import { HttpClientError } from "effect/http" +import { Hash } from "@opencode/util/hash" +import { makeGlobalNode } from "@opencode/util/effect/app-node" + +/** How long a result stays fresh, and how often callers refresh, unless a caller passes its own. */ +export const interval = Duration.minutes(10) + +export interface Input { + /** Everything that changes the answer, such as the connection and base URL. Never a raw secret: see `fingerprint`. */ + readonly key: string + readonly load: Effect.Effect + /** How long a result stays fresh. Defaults to `interval`; a caller refreshing on a timer uses the same value. */ + readonly interval?: Duration.Input + /** Skips the age check, but still joins a running load. */ + readonly force?: boolean +} + +export interface Interface { + /** + * Returns the last result for `key` while it is fresh, otherwise joins the running load or starts one, so loads + * never overlap per key. A load retries `transient` failures 3 times. A failure keeps the last good value, or + * without one is returned, until the next interval. Interrupts and defects are never kept. Entries nobody asked + * for in 24 hours, or in two of their intervals if longer, are dropped. + */ + readonly get: (input: Input) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/SharedCache") {} + +/** A short, one-way stand-in for a secret in a cache key, so a secret changed in place loads again. */ +export const fingerprint = (secret: string) => Hash.sha256(secret).slice(0, 16) + +/** Network errors, timeouts and HTTP 408, 429, 500, 502, 503 or 504, as Effect's `HttpClient.retryTransient` counts them. */ +export const transient = (error: unknown) => + Cause.isTimeoutError(error) || + (HttpClientError.isHttpClientError(error) && + (error.reason._tag === "TransportError" || + (error.reason._tag === "StatusCodeError" && + [408, 429, 500, 502, 503, 504].includes(error.reason.response.status)))) + +interface Entry { + last?: { readonly exit: Exit.Exit; readonly time: number } + loading?: Fiber.Fiber + read: number + unread: number +} + +const unread = Duration.toMillis(Duration.hours(24)) + +const layer = Layer.effect( + Service, + Effect.gen(function* () { + const run = yield* FiberSet.makeRuntime() + const entries = new Map() + + const start = Effect.fnUntraced(function* (key: string, entry: Entry, load: Effect.Effect) { + const time = yield* Clock.currentTimeMillis + const exit = yield* Effect.exit( + Effect.retry(load, { schedule: Schedule.exponential("1 second"), times: 3, while: transient }), + ) + entry.read = yield* Clock.currentTimeMillis + if (Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)) return yield* exit + if (Exit.isFailure(exit)) yield* Effect.logWarning("failed to load shared value", { key, cause: exit.cause }) + if (Exit.isFailure(exit) && Cause.hasDies(exit.cause)) return yield* exit + const previous = entry.last?.exit + entry.last = { exit: Exit.isFailure(exit) && previous && Exit.isSuccess(previous) ? previous : exit, time } + return yield* entry.last.exit + }) + + const get = Effect.fnUntraced(function* (input: Input) { + const now = yield* Clock.currentTimeMillis + const age = Duration.toMillis(input.interval ?? interval) + for (const [key, item] of entries) if (!item.loading && now - item.read > item.unread) entries.delete(key) + const entry = entries.get(input.key) ?? { read: now, unread } + entries.set(input.key, entry) + entry.read = now + // A caller refreshing less often than daily must never find its entry dropped between two of its refreshes. + if (Number.isFinite(age)) entry.unread = Math.max(entry.unread, 2 * age) + const last = entry.last + // The key decides the value's type, so a stored or joined result under the same key has this caller's types. + if (!input.force && last && now - last.time < age) return yield* last.exit as Exit.Exit + // Yield first so `entry.loading` is set before a synchronous load can finish and clear it. + entry.loading ??= run( + Effect.yieldNow.pipe( + Effect.andThen(start(input.key, entry, input.load)), + Effect.ensuring( + Effect.sync(() => { + entry.loading = undefined + }), + ), + ), + ) + return yield* Fiber.join(entry.loading as Fiber.Fiber) + }) + + return Service.of({ get }) + }), +) + +export const node = makeGlobalNode({ service: Service, layer, deps: [] }) diff --git a/packages/core/test/github-copilot/models.test.ts b/packages/core/test/github-copilot/models.test.ts index 0080cf0d933..8708f083bc9 100644 --- a/packages/core/test/github-copilot/models.test.ts +++ b/packages/core/test/github-copilot/models.test.ts @@ -1,8 +1,13 @@ import { expect, test } from "bun:test" import { CopilotModels } from "@opencode/core/github-copilot/models" +import { Effect } from "effect" +import { FetchHttpClient, HttpClient } from "effect/http" import { Model } from "@opencode/core/model" import { Provider } from "@opencode/core/provider" +const run = (effect: (http: HttpClient.HttpClient) => Effect.Effect) => + Effect.runPromise(HttpClient.HttpClient.pipe(Effect.flatMap(effect), Effect.provide(FetchHttpClient.layer))) + test("defensively syncs advertised Copilot models", async () => { const server = Bun.serve({ port: 0, @@ -93,7 +98,7 @@ test("defensively syncs advertised Copilot models", async () => { ...Model.Info.default(Provider.ID.githubCopilot, Model.ID.make("stale")), modelID: Model.ID.make("stale"), }) - const models = await CopilotModels.get(server.url.origin, {}, [existing, stale]) + const models = await run((http) => CopilotModels.get(http, server.url.origin, {}, [existing, stale])) const model = models.get(Model.ID.make("gpt-5")) expect(model?.name).toBe("GPT-5 local") @@ -147,9 +152,17 @@ test("prices cache reads from either token price spelling", async () => { }) try { - const models = await CopilotModels.get(server.url.origin, {}, []) - expect(models.get(Model.ID.make("renamed"))?.cost[0]).toMatchObject({ input: 2.5, output: 15, cache: { read: 0.25 } }) - expect(models.get(Model.ID.make("legacy"))?.cost[0]).toMatchObject({ input: 2.5, output: 15, cache: { read: 0.25 } }) + const models = await run((http) => CopilotModels.get(http, server.url.origin, {}, [])) + expect(models.get(Model.ID.make("renamed"))?.cost[0]).toMatchObject({ + input: 2.5, + output: 15, + cache: { read: 0.25 }, + }) + expect(models.get(Model.ID.make("legacy"))?.cost[0]).toMatchObject({ + input: 2.5, + output: 15, + cache: { read: 0.25 }, + }) expect(models.get(Model.ID.make("unpriced"))?.cost[0]).toMatchObject({ cache: { read: 0 } }) } finally { await server.stop(true) diff --git a/packages/core/test/modal/models.test.ts b/packages/core/test/modal/models.test.ts index 07518f370c9..95d3befd9c4 100644 --- a/packages/core/test/modal/models.test.ts +++ b/packages/core/test/modal/models.test.ts @@ -1,10 +1,15 @@ import { expect, test } from "bun:test" import { ModalModels } from "@opencode/core/modal/models" +import { Effect } from "effect" +import { FetchHttpClient, HttpClient } from "effect/http" import { Model } from "@opencode/core/model" import { Provider } from "@opencode/core/provider" import { ProviderPlugins } from "@opencode/core/plugin/provider" import { Money } from "@opencode/schema/money" +const run = (effect: (http: HttpClient.HttpClient) => Effect.Effect) => + Effect.runPromise(HttpClient.HttpClient.pipe(Effect.flatMap(effect), Effect.provide(FetchHttpClient.layer))) + const providerID = Provider.ID.make("modal") test("modal plugin is registered", () => { @@ -55,7 +60,7 @@ test("maps live Modal models onto catalog templates", async () => { try { const base = template("base-model") const stale = template("stale") - const models = await ModalModels.get(`${server.url.origin}/v1`, "test-key", [base, stale]) + const models = await run((http) => ModalModels.get(http, `${server.url.origin}/v1`, "test-key", [base, stale])) expect(models.has(Model.ID.make("stale"))).toBe(false) expect(models.has(Model.ID.make("malformed"))).toBe(false) @@ -113,7 +118,7 @@ test("keeps template cost and limits when the proxy omits them", async () => { ], limit: { context: 1000, input: 500, output: 250 }, }) - const models = await ModalModels.get(server.url.origin, "test-key", [base]) + const models = await run((http) => ModalModels.get(http, server.url.origin, "test-key", [base])) const model = models.get(Model.ID.make("sparse")) expect(model?.name).toBe("hf-base catalog") expect(model?.cost[0]).toMatchObject({ input: 5, output: 10, cache: { read: 1, write: 2 } }) @@ -130,7 +135,7 @@ test("throws on proxy failure so the plugin can fail soft", async () => { }) try { - await expect(ModalModels.get(server.url.origin, "test-key", [])).rejects.toThrow() + await expect(run((http) => ModalModels.get(http, server.url.origin, "test-key", []))).rejects.toThrow() } finally { await server.stop(true) } diff --git a/packages/core/test/plugin/fixture.ts b/packages/core/test/plugin/fixture.ts index e2e1574e1fe..5bcc568a85a 100644 --- a/packages/core/test/plugin/fixture.ts +++ b/packages/core/test/plugin/fixture.ts @@ -27,6 +27,7 @@ import { Plugin } from "@opencode/core/plugin" import { PluginHooks } from "@opencode/core/plugin/hooks" import { Provider } from "@opencode/core/provider" import { Session } from "@opencode/core/session" +import { SharedCache } from "@opencode/core/shared-cache" import { InstructionEntry } from "@opencode/core/session/instruction-entry" import { PersistentPty } from "@opencode/core/persistent-pty" import { LocationServiceMap } from "@opencode/core/location-service-map" @@ -114,6 +115,7 @@ export const makePluginTestLayer = (permission = permissionLayer, generate = gen KV.node, Mcp.node, Session.node, + SharedCache.node, InstructionEntry.node, PersistentPty.node, LocationServiceMap.node, diff --git a/packages/core/test/plugin/provider-azure.test.ts b/packages/core/test/plugin/provider-azure.test.ts index 1f22e91f997..5bdf8f214cb 100644 --- a/packages/core/test/plugin/provider-azure.test.ts +++ b/packages/core/test/plugin/provider-azure.test.ts @@ -1,10 +1,11 @@ import { chmod } from "node:fs/promises" import { Agent } from "@opencode/core/agent" +import { Bus } from "@opencode/core/bus" import { describe, expect } from "bun:test" -import { Effect, Schedule, Schema } from "effect" +import { Clock, Effect, Schedule, Schema } from "effect" +import { TestClock } from "effect/testing" import { Config } from "@opencode/core/config" import { ConfigProviderPlugin } from "@opencode/core/config/plugin/provider" -import { Bus } from "@opencode/core/bus" import { Credential } from "@opencode/core/credential" import { Model } from "@opencode/core/model" import { Plugin } from "@opencode/core/plugin" @@ -169,12 +170,6 @@ const connect = Effect.fn(function* (value: Credential.Value) { return yield* credentials.create({ integrationID: azureID, value }) }) -// Republishing a switch reruns discovery for the connection that is already active. -const announceSwitch = Effect.fn(function* (credentialID: Credential.ID) { - const bus = yield* Bus.Service - yield* bus.publish(Credential.Event.Switched, { integrationID: azureID, credentialID }, { global: true }) -}) - const seedProvider = (settings?: Provider.Settings) => Effect.gen(function* () { const catalog = yield* Provider.Service @@ -222,6 +217,19 @@ const eventually = (effect: Effect.Effect, predicate: (value: Effect.retry({ times: 3000, schedule: Schedule.spaced("1 millis") }), ) +// Lets real Azure answers arrive between TestClock steps, so retries and the 10 minute refresh need no real waiting. +const eventuallyOnClock = (effect: Effect.Effect, predicate: (value: A) => boolean) => + Effect.gen(function* () { + const deadline = performance.now() + 15_000 + while (performance.now() < deadline) { + const value = yield* effect + if (predicate(value)) return value + yield* TestClock.withLive(Effect.sleep("1 millis")) + yield* TestClock.adjust("500 millis") + } + return yield* Effect.die(new Error("Timed out waiting for value")) + }) + // Lets a discovery that has received its last answer finish publishing before asserting that nothing changed. const settle = Effect.promise(() => Bun.sleep(25)) @@ -276,7 +284,6 @@ describe("AzurePlugin connecting", () => { ]) }), ) - }) describe("AzurePlugin startup", () => { @@ -424,16 +431,11 @@ describe("AzurePlugin discovery", () => { }), ) - it.live("keeps the last complete list when a later page fails", () => + it.effect("keeps the last complete list when a later page fails", () => Effect.gen(function* () { const cli = yield* fakeAzureCli() const state = { paged: false } - const fellBack = Promise.withResolvers() const azure = yield* fakeAzure({ - resource: () => { - fellBack.resolve() - return notFound() - }, management: (request) => { if (request.path.startsWith("/page-2")) return new Response("Unavailable", { status: 503 }) if (state.paged) return managementPage([managed("nano", "gpt-5-nano")], `${request.origin}/page-2`) @@ -441,18 +443,43 @@ describe("AzurePlugin discovery", () => { }, }) yield* seedCatalog - const credential = yield* connect(cliCredential()) + yield* connect(cliCredential()) yield* addPlugin(azure.endpoints).pipe(cli.provide) - yield* eventually(azureModelIDs, (ids) => ids.length === 1) + yield* eventuallyOnClock(azureModelIDs, (ids) => ids.length === 1) state.paged = true - yield* announceSwitch(credential.id) - yield* Effect.promise(() => fellBack.promise) - yield* settle + yield* TestClock.adjust("10 minutes") + const pages = Effect.sync(() => azure.requests.filter((request) => request.path.startsWith("/page-2")).length) + yield* eventuallyOnClock(pages, (count) => count === 4) + yield* Effect.repeat( + TestClock.withLive(Effect.sleep("1 millis")).pipe(Effect.andThen(TestClock.adjust("500 millis"))), + { times: 3 }, + ) expect(yield* azureModelIDs).toEqual(["mini"]) }), ) + it.effect("reports the management API error when the resource's own list fails too", () => + Effect.gen(function* () { + const cli = yield* fakeAzureCli() + const state = { failures: 1 } + const azure = yield* fakeAzure({ + management: () => + state.failures-- > 0 + ? new Response("Unavailable", { status: 503 }) + : managementPage([managed("mini", "gpt-5-mini")]), + }) + yield* seedCatalog + yield* connect(cliCredential()) + yield* addPlugin(azure.endpoints).pipe(cli.provide) + + // The resource's 404 would wait for the next refresh; the management API's 503 is retried within seconds. + expect(yield* eventuallyOnClock(azureModelIDs, (ids) => ids.length === 1)).toEqual(["mini"]) + expect(yield* Clock.currentTimeMillis).toBeLessThan(60_000) + expect(azure.requests.filter((request) => request.path.startsWith("/openai/deployments"))).toHaveLength(1) + }), + ) + // The test harness fails any request to a non-local host, so following the link would fail the test. it.live("never sends the management token to a page link outside the management endpoint", () => Effect.gen(function* () { @@ -507,28 +534,46 @@ describe("AzurePlugin discovery", () => { }) describe("AzurePlugin switching accounts", () => { - it.live("shows the new account without waiting for the previous account's discovery", () => + it.live("loads again when the API key changes in place", () => Effect.gen(function* () { - const stuck = Promise.withResolvers() - const calls = { count: 0 } const azure = yield* fakeAzure({ - resource: (request) => { - calls.count++ - if (calls.count === 2) return stuck.promise - return resourceList(listed(request.key === "secret" ? "production-a" : "production-b", "gpt-5-mini")) - }, + resource: (request) => + resourceList(listed(request.key === "secret" ? "production-a" : "production-b", "gpt-5-mini")), }) yield* seedCatalog - const first = yield* connect(keyCredential()) + const credential = yield* connect(keyCredential()) yield* addPlugin(azure.endpoints) yield* eventually(azureModelIDs, (ids) => ids.includes("production-a")) - // Start a discovery for the first account that Azure never answers. - yield* announceSwitch(first.id) - yield* eventually( - Effect.sync(() => calls.count), - (count) => count === 2, + const credentials = yield* Credential.Service + const bus = yield* Bus.Service + yield* credentials.update(credential.id, { value: keyCredential("rotated-key") }) + yield* bus.publish( + Credential.Event.Switched, + { integrationID: Integration.ID.make("azure"), credentialID: credential.id }, + { global: true }, ) + yield* eventually(azureModelIDs, (ids) => ids.includes("production-b")) + }), + ) + + it.live("shows the new account without waiting for the previous account's discovery", () => + Effect.gen(function* () { + const stuck = Promise.withResolvers() + const asked = Promise.withResolvers() + const azure = yield* fakeAzure({ + resource: (request) => { + if (request.key !== "secret") return resourceList(listed("production-b", "gpt-5-mini")) + asked.resolve() + return stuck.promise + }, + }) + yield* seedCatalog + yield* connect(keyCredential()) + yield* addPlugin(azure.endpoints) + + // Start a discovery for the first account that Azure never answers. + yield* Effect.promise(() => asked.promise) yield* connect(keyCredential("other-key", "other-resource")) const deployed = yield* eventually(azureModels, (list) => list.length === 1 && list[0]?.id === "production-b") diff --git a/packages/core/test/plugin/provider-digitalocean.test.ts b/packages/core/test/plugin/provider-digitalocean.test.ts index b69ed3bfff1..f65856318ef 100644 --- a/packages/core/test/plugin/provider-digitalocean.test.ts +++ b/packages/core/test/plugin/provider-digitalocean.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test" -import { Effect, Schedule } from "effect" +import { Deferred, Effect, Logger, Schedule } from "effect" import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" +import { TestClock } from "effect/testing" import { Credential } from "@opencode/core/credential" import { Integration } from "@opencode/core/integration" import { Model } from "@opencode/core/model" @@ -47,14 +48,20 @@ const discovery = Effect.gen(function* () { }) draft.models.update(Provider.ID.openai, Model.ID.make("router:alpha"), () => {}) }) - const remote: { status: number; body: unknown; requests: HttpClientRequest.HttpClientRequest[] } = { + const remote: { + status: number + body: unknown + requests: HttpClientRequest.HttpClientRequest[] + gate?: Deferred.Deferred + } = { status: 200, body: { model_routers: [{ name: "alpha" }, { name: "configured" }] }, requests: [], } const http = HttpClient.make((request) => - Effect.sync(() => { + Effect.gen(function* () { remote.requests.push(request) + if (remote.gate) yield* Deferred.await(remote.gate) return HttpClientResponse.fromWeb(request, Response.json(remote.body, { status: remote.status })) }), ) @@ -117,7 +124,47 @@ describe("DigitalOceanPlugin", () => { name: "Configured router", limit: { context: 256_000 }, }) + }), + ) + it.effect("loads nothing and logs nothing at startup when the access token has expired", () => + Effect.gen(function* () { + const fixture = yield* discovery + const warnings: unknown[] = [] + yield* TestClock.adjust("1 second") + yield* oauthCredential("do-token", 1) + yield* fixture.install.pipe( + Effect.provide( + Logger.layer([ + Logger.make((entry) => { + if (entry.logLevel === "Warn" || entry.logLevel === "Error") warnings.push(entry.message) + }), + ]), + ), + ) + yield* drain + expect(fixture.remote.requests).toHaveLength(0) + expect(yield* fixture.models.get(providerID, Model.ID.make("router:alpha"))).toBeUndefined() + expect(warnings).toEqual([]) + }), + ) + + it.effect("refreshes routers in the background with the latest access token", () => + Effect.gen(function* () { + const fixture = yield* discovery + const credentials = yield* Credential.Service + const credential = yield* oauthCredential() + yield* fixture.install + yield* drain + expect(fixture.remote.requests).toHaveLength(1) + + yield* credentials.update(credential.id, { + value: Credential.OAuth.make({ type: "oauth", methodID, access: "rotated-token", refresh: "", expires: 0 }), + }) + yield* TestClock.adjust("10 minutes") + yield* drain + expect(fixture.remote.requests).toHaveLength(2) + expect(fixture.remote.requests[1]?.headers.authorization).toBe("Bearer rotated-token") }), ) @@ -133,7 +180,7 @@ describe("DigitalOceanPlugin", () => { yield* drain expect(yield* fixture.models.get(providerID, Model.ID.make("router:alpha"))).toBeDefined() - fixture.remote.status = 503 + fixture.remote.status = 401 const second = yield* oauthCredential("other-account") yield* drain expect(fixture.remote.requests.at(-1)?.headers.authorization).toBe("Bearer other-account") @@ -157,6 +204,35 @@ describe("DigitalOceanPlugin", () => { }), ) + it.effect("shows a returning account's cached routers at once and refreshes them in the background", () => + Effect.gen(function* () { + const fixture = yield* discovery + const credentials = yield* Credential.Service + const first = yield* oauthCredential() + yield* fixture.install + yield* drain + fixture.remote.body = { model_routers: [{ name: "beta" }] } + yield* oauthCredential("other-account") + yield* drain + const router = (name: string) => fixture.models.get(providerID, Model.ID.make(`router:${name}`)) + expect(yield* router("beta")).toBeDefined() + + fixture.remote.body = { model_routers: [{ name: "gamma" }] } + fixture.remote.gate = yield* Deferred.make() + const count = fixture.remote.requests.length + yield* credentials.activate(first.id) + yield* drain + expect(yield* router("alpha")).toBeDefined() + expect(fixture.remote.requests).toHaveLength(count + 1) + + yield* Deferred.succeed(fixture.remote.gate, undefined) + yield* drain + expect(yield* router("alpha")).toBeUndefined() + expect(yield* router("gamma")).toBeDefined() + expect(fixture.remote.requests).toHaveLength(count + 1) + }), + ) + it.live("completes browser OAuth independently of router discovery and releases the listener", () => Effect.gen(function* () { const fixture = yield* discovery diff --git a/packages/core/test/plugin/provider-github-copilot.test.ts b/packages/core/test/plugin/provider-github-copilot.test.ts index d54f051861c..8a0a4e6ffcf 100644 --- a/packages/core/test/plugin/provider-github-copilot.test.ts +++ b/packages/core/test/plugin/provider-github-copilot.test.ts @@ -5,6 +5,8 @@ import { Session } from "@opencode/core/session" import { Location } from "@opencode/core/location" import { describe, expect, test } from "bun:test" import { Effect } from "effect" +import { HttpClient, HttpClientResponse } from "effect/http" +import { TestClock } from "effect/testing" import { Model } from "@opencode/core/model" import { ModelResolver } from "@opencode/core/model-resolver" import { Plugin } from "@opencode/core/plugin" @@ -20,6 +22,8 @@ import { import { Message, SystemPart } from "@opencode/ai/effect" import { Provider } from "@opencode/core/provider" import { Integration } from "@opencode/core/integration" +import { Credential } from "@opencode/core/credential" +import { drain } from "../lib/clock" import type { SessionRequestKind } from "@opencode/plugin/effect/session" import { fakeSelectorSdk } from "../fixture/selector" import { testEffect } from "../lib/effect" @@ -81,6 +85,62 @@ describe("GithubCopilotPlugin", () => { expect(copilotEntitlementError({})).toBeUndefined() }) + it.effect("reloads providers only when a refreshed model list changes", () => + Effect.gen(function* () { + const credentials = yield* Credential.Service + const providers = yield* Provider.Service + const requests = { count: 0 } + const http = HttpClient.make((request) => + Effect.sync(() => { + requests.count++ + return HttpClientResponse.fromWeb( + request, + Response.json({ + data: [ + { + model_picker_enabled: true, + id: "gpt-5", + name: "GPT-5", + version: "gpt-5", + capabilities: { + family: "gpt-5", + limits: { max_output_tokens: 1, max_prompt_tokens: 1 }, + supports: { tool_calls: true }, + }, + }, + ], + }), + ) + }), + ) + yield* providers.transform((draft) => { + draft.update(Provider.ID.githubCopilot, () => {}) + }) + yield* credentials.create({ + integrationID: Integration.ID.make("github-copilot"), + value: Credential.OAuth.make({ + type: "oauth", + methodID: Integration.MethodID.make("device"), + access: "token", + refresh: "token", + expires: 0, + }), + }) + yield* addPlugin().pipe(Effect.provideService(HttpClient.HttpClient, http)) + yield* drain + const folds = { count: 0 } + yield* providers.transform(() => { + folds.count++ + }) + const before = folds.count + + yield* TestClock.adjust("10 minutes") + yield* drain + expect(requests.count).toBe(2) + expect(folds.count).toBe(before) + }), + ) + it.effect("registers GitHub Copilot device OAuth", () => Effect.gen(function* () { yield* addPlugin() @@ -623,10 +683,9 @@ describe("GithubCopilotPlugin", () => { editor.models.update(Provider.ID.githubCopilot, Model.ID.make("gpt-5-chat-latest"), () => {}) }) yield* addPlugin() - expect( - required(yield* models.get(Provider.ID.githubCopilot, Model.ID.make("gpt-5-chat-latest"))) - .enabled, - ).toBe(false) + expect(required(yield* models.get(Provider.ID.githubCopilot, Model.ID.make("gpt-5-chat-latest"))).enabled).toBe( + false, + ) }), ) @@ -640,8 +699,7 @@ describe("GithubCopilotPlugin", () => { }) yield* addPlugin() expect( - required(yield* models.get(Provider.ID.make("custom-copilot"), Model.ID.make("gpt-5-chat-latest"))) - .enabled, + required(yield* models.get(Provider.ID.make("custom-copilot"), Model.ID.make("gpt-5-chat-latest"))).enabled, ).toBe(true) }), ) diff --git a/packages/core/test/shared-cache.test.ts b/packages/core/test/shared-cache.test.ts new file mode 100644 index 00000000000..346fa8fd08b --- /dev/null +++ b/packages/core/test/shared-cache.test.ts @@ -0,0 +1,161 @@ +import { describe, expect } from "bun:test" +import { Cause, Deferred, Duration, Effect, Exit, Fiber } from "effect" +import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/http" +import { TestClock } from "effect/testing" +import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" +import { SharedCache } from "@opencode/core/shared-cache" +import { testEffect } from "./lib/effect" + +const it = testEffect(AppNodeBuilder.build(SharedCache.node)) + +const request = HttpClientRequest.get("https://models.example/v1") +const status = (code: number) => + HttpClient.filterStatusOk( + HttpClient.make((request) => + Effect.succeed(HttpClientResponse.fromWeb(request, new Response(null, { status: code }))), + ), + ) + .execute(request) + .pipe(Effect.flip) + +describe("SharedCache", () => { + it.effect("counts network errors, timeouts and 408, 429 or 5xx as transient, and nothing else", () => + Effect.gen(function* () { + const reason = (value: HttpClientError.HttpClientError["reason"]) => + new HttpClientError.HttpClientError({ reason: value }) + expect(SharedCache.transient(reason(new HttpClientError.TransportError({ request })))).toBe(true) + expect(SharedCache.transient(new Cause.TimeoutError())).toBe(true) + expect(SharedCache.transient(yield* status(503))).toBe(true) + expect(SharedCache.transient(yield* status(429))).toBe(true) + expect(SharedCache.transient(yield* status(401))).toBe(false) + expect(SharedCache.transient(yield* status(404))).toBe(false) + expect(SharedCache.transient(reason(new HttpClientError.InvalidUrlError({ request })))).toBe(false) + expect(SharedCache.transient(new TypeError("fetch failed"))).toBe(false) + }), + ) + + it.effect("joins a running load, refreshes after the caller's interval, and forces without overlapping", () => + Effect.gen(function* () { + const cache = yield* SharedCache.Service + const loads = { a: 0, b: 0 } + const gate = yield* Deferred.make() + const get = (key: "a" | "b", options: { interval?: Duration.Input; force?: boolean } = {}) => + cache.get({ key, ...options, load: Deferred.await(gate).pipe(Effect.map(() => `${key}${++loads[key]}`)) }) + + const joined = yield* Effect.all([get("a"), get("a", { force: true }), get("b")], { + concurrency: "unbounded", + }).pipe(Effect.forkChild({ startImmediately: true })) + yield* Deferred.succeed(gate, undefined) + expect(yield* Fiber.join(joined)).toEqual(["a1", "a1", "b1"]) + + yield* TestClock.adjust("5 minutes") + expect(yield* get("a")).toBe("a1") + expect(yield* get("a", { interval: "1 minute" })).toBe("a2") + expect(yield* get("a", { force: true })).toBe("a3") + yield* TestClock.adjust("10 minutes") + expect(yield* get("b")).toBe("b2") + expect(loads).toEqual({ a: 3, b: 2 }) + }), + ) + + it.effect("retries transient failures, then keeps the last list or the failure until the next interval", () => + Effect.gen(function* () { + const cache = yield* SharedCache.Service + const remote = { results: [] as unknown[], loads: 0 } + const get = (force?: boolean) => + cache.get({ + key: "models", + force, + load: Effect.suspend(() => { + remote.loads++ + const result = remote.results.shift() + return typeof result === "string" ? Effect.succeed(result) : Effect.fail(result) + }), + }) + const timeout = new Cause.TimeoutError() + + remote.results = [401] + expect(yield* Effect.flip(get())).toBe(401) + yield* TestClock.adjust("9 minutes") + expect(yield* Effect.flip(get())).toBe(401) + remote.results = [timeout, timeout, "list"] + const flaky = yield* get(true).pipe(Effect.forkChild({ startImmediately: true })) + yield* Effect.yieldNow.pipe(Effect.andThen(TestClock.adjust("1 second"))) + yield* Effect.yieldNow.pipe(Effect.andThen(TestClock.adjust("2 seconds"))) + expect(yield* Fiber.join(flaky)).toBe("list") + expect(remote.loads).toBe(4) + + remote.results = [401, timeout, timeout, timeout, timeout] + expect(yield* get(true)).toBe("list") + const outage = yield* get(true).pipe(Effect.forkChild({ startImmediately: true })) + for (const wait of ["1 second", "2 seconds", "4 seconds"] as const) + yield* Effect.yieldNow.pipe(Effect.andThen(TestClock.adjust(wait))) + expect(yield* Fiber.join(outage)).toBe("list") + expect(yield* get()).toBe("list") + expect(remote.loads).toBe(9) + }), + ) + + it.effect("never keeps interrupts or defects, and an interrupted caller leaves the load to the others", () => + Effect.gen(function* () { + const cache = yield* SharedCache.Service + const gate = yield* Deferred.make() + const loads = { count: 0 } + const get = (load: Effect.Effect) => + cache.get({ + key: "models", + load: Effect.suspend(() => { + loads.count++ + return load + }), + }) + + const first = yield* get(Deferred.await(gate).pipe(Effect.as(1))).pipe( + Effect.forkChild({ startImmediately: true }), + ) + const second = yield* get(Effect.succeed(2)).pipe(Effect.forkChild({ startImmediately: true })) + yield* Effect.yieldNow + yield* Fiber.interrupt(first) + yield* Deferred.succeed(gate, undefined) + expect(yield* Fiber.join(second)).toBe(1) + + yield* TestClock.adjust("10 minutes") + expect(Exit.hasDies(yield* Effect.exit(get(Effect.die("bug"))))).toBe(true) + expect(Exit.hasInterrupts(yield* Effect.exit(get(Effect.interrupt)))).toBe(true) + expect(yield* get(Effect.succeed(3))).toBe(3) + expect(loads.count).toBe(4) + }), + ) + + it.effect("drops entries unread for 24 hours, never one still in use or loading", () => + Effect.gen(function* () { + const cache = yield* SharedCache.Service + const loads = { idle: 0, used: 0, daily: 0, slow: 0 } + const once = (key: keyof typeof loads) => + Effect.suspend(() => (++loads[key] === 1 ? Effect.succeed(key) : Effect.fail("offline"))) + const gate = yield* Deferred.make() + + expect(yield* cache.get({ key: "idle", load: once("idle") })).toBe("idle") + expect(yield* cache.get({ key: "daily", interval: "48 hours", load: once("daily") })).toBe("daily") + const getSlow = cache + .get({ key: "slow", load: Deferred.await(gate).pipe(Effect.andThen(once("slow"))) }) + .pipe(Effect.forkChild({ startImmediately: true })) + const slow = yield* getSlow + yield* Effect.yieldNow + // A Location refreshing every 10 minutes keeps its entry, so its list survives later failures. + for (let elapsed = 0; elapsed <= 25 * 60; elapsed += 10) { + expect(yield* cache.get({ key: "used", load: once("used") })).toBe("used") + yield* TestClock.adjust("10 minutes") + } + + expect(yield* Effect.flip(cache.get({ key: "idle", load: once("idle") }))).toBe("offline") + expect(loads.idle).toBe(2) + yield* TestClock.adjust("23 hours") + expect(yield* cache.get({ key: "daily", interval: "48 hours", load: once("daily") })).toBe("daily") + const joined = yield* getSlow + yield* Deferred.succeed(gate, undefined) + expect([yield* Fiber.join(slow), yield* Fiber.join(joined)]).toEqual(["slow", "slow"]) + expect(loads.slow).toBe(1) + }), + ) +})