refactor(core): share provider discovery through a simple cache (#60082)

This commit is contained in:
Aiden Cline
2026-10-30 20:35:16 -05:00
committed by GitHub
Unverified
parent 591690c375
commit a66964ee9f
16 changed files with 825 additions and 246 deletions
+19 -13
View File
@@ -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<string, UsableModel>
export async function load(baseURL: string, headers: RequestInit["headers"]): Promise<Snapshot> {
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[]) {
+14 -15
View File
@@ -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<Snapshot> {
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[]) {
+3
View File
@@ -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,
+53 -16
View File
@@ -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<string, { access: string; expires: number }>()
// 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<typeof IntegrationConnection.key>[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<Record<string, unknown>> | undefined, fallback?: string) {
const configured = settings?.resourceName
if (typeof configured === "string" && configured.trim() !== "") return configured
@@ -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)
@@ -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<ReturnType<typeof ctx.integration.connection.active>>
} = {}
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<ReturnType<typeof ctx.integration.connection.active>>
},
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<typeof IntegrationConnection.key>[0],
url: string,
headers: Record<string, string>,
) {
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(/\/$/, "")
}
+48 -49
View File
@@ -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<ReturnType<typeof ctx.integration.connection.active>>
} = {}
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<ReturnType<typeof ctx.integration.connection.active>>
},
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<typeof IntegrationConnection.key>[0], apiKey: string, baseURL: string) {
const access = IntegrationConnection.key(connection) ?? "env:MODAL_PROXY_TOKEN"
return `modal:${access}:${SharedCache.fingerprint(apiKey)}:${baseURL.replace(/\/+$/, "")}`
}
@@ -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<typeof IntegrationConnection.key>[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 = <A>(load: (options: Options) => Effect.Effect<A>, apply: (next: A) => Effect.Effect<void>) =>
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* <A>(input: {
readonly integrationID: string
readonly initial: A
readonly interval?: Duration.Input
readonly active: Effect.Effect<Connection>
readonly load: (current: A, options: Options) => Effect.Effect<{ readonly connection: Connection; readonly value: A }>
readonly reload: Effect.Effect<void>
}): 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<void, never>()
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
})
+103
View File
@@ -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<A, E> {
/** 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<A, E>
/** 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: <A, E>(input: Input<A, E>) => Effect.Effect<A, E>
}
export class Service extends Context.Service<Service, Interface>()("@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<unknown, unknown>; readonly time: number }
loading?: Fiber.Fiber<unknown, unknown>
read: number
unread: number
}
const unread = Duration.toMillis(Duration.hours(24))
const layer = Layer.effect(
Service,
Effect.gen(function* () {
const run = yield* FiberSet.makeRuntime<never, unknown, unknown>()
const entries = new Map<string, Entry>()
const start = Effect.fnUntraced(function* (key: string, entry: Entry, load: Effect.Effect<unknown, unknown>) {
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* <A, E>(input: Input<A, E>) {
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<A, E>
// 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<A, E>)
})
return Service.of({ get })
}),
)
export const node = makeGlobalNode({ service: Service, layer, deps: [] })
@@ -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 = <A, E>(effect: (http: HttpClient.HttpClient) => Effect.Effect<A, E>) =>
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)
+8 -3
View File
@@ -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 = <A, E>(effect: (http: HttpClient.HttpClient) => Effect.Effect<A, E>) =>
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)
}
+2
View File
@@ -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,
@@ -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 = <A, R>(effect: Effect.Effect<A, never, R>, 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 = <A, R>(effect: Effect.Effect<A, never, R>, 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<void>()
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<Response>()
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<Response>()
const asked = Promise.withResolvers<void>()
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")
@@ -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<void>
} = {
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<void>()
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
@@ -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)
}),
)
+161
View File
@@ -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<void>()
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<void>()
const loads = { count: 0 }
const get = (load: Effect.Effect<number>) =>
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<void>()
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)
}),
)
})