diff --git a/packages/core/src/plugin/provider/azure.ts b/packages/core/src/plugin/provider/azure.ts index 5144b03a92b..cab1745a2cb 100644 --- a/packages/core/src/plugin/provider/azure.ts +++ b/packages/core/src/plugin/provider/azure.ts @@ -5,7 +5,6 @@ import { define } from "@opencode/plugin/effect/plugin" import { Form } from "@opencode/schema/form" import { AppProcess } from "@opencode/util/process" 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" @@ -15,6 +14,7 @@ import { SharedCache } from "../../shared-cache.js" import { which } from "../../util/which.js" import type { PluginInternal } from "../internal.js" import { configuredSettings } from "./configured.js" +import { activeCredential } from "./credential-settings.js" import { ProviderRefresh } from "./refresh.js" const cognitiveScope = "https://cognitiveservices.azure.com/.default" @@ -66,7 +66,6 @@ export function make( effect: Effect.fn(function* (ctx) { const configured = yield* configuredSettings(Provider.ID.azure) const processes = yield* AppProcess.Service - const bus = yield* Bus.Service const credentials = yield* Credential.Service const providers = yield* Provider.Service const cache = yield* SharedCache.Service @@ -304,21 +303,16 @@ export function make( const credential = yield* ctx.integration.connection .resolve(connection) .pipe(Effect.orElseSucceed(() => undefined)) - if ( - !credential || - credential.type === "oauth" || - (credential.type === "external" && credential.methodID !== methodID) - ) - return + if (!usable(credential)) 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), + usable(current) + ? deployments(url, name, current) + : Effect.fail(new Error("Azure connection has no usable credential")), ), // 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))), @@ -369,11 +363,7 @@ export function make( yield* ctx.provider.transform((evt) => { for (const item of evt.list()) { - if ( - item.provider.id !== Provider.ID.azure && - !item.provider.package.startsWith("@opencode/ai/providers/azure/") - ) - continue + if (!azureProvider(item.provider)) continue const resourceName = resourceFor(item.provider) const websocket = responsesWebSocketCapable(item.provider) if (!resourceName && !websocket) continue @@ -400,11 +390,7 @@ export function make( }) yield* ctx.model.transform((models) => { for (const item of models.provider.list()) { - if ( - item.provider.id !== Provider.ID.azure && - !item.provider.package.startsWith("@opencode/ai/providers/azure/") - ) - continue + if (!azureProvider(item.provider)) continue const resourceName = resourceFor(item.provider) if (!resourceName) continue for (const model of item.models.values()) { @@ -415,11 +401,7 @@ export function make( ) if (baseURL === model.settings.baseURL) continue models.update(item.provider.id, model.id, (draft) => { - if (typeof draft.settings?.baseURL === "string") - draft.settings.baseURL = expandResourceName( - draft.settings.baseURL, - resolveResourceName(draft.settings, resourceName) ?? resourceName, - ) + if (typeof draft.settings?.baseURL === "string") draft.settings.baseURL = baseURL }) } } @@ -437,12 +419,9 @@ 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(() => - rebind().pipe(Effect.andThen(FiberHandle.run(discovery, ProviderRefresh.switched(discover, apply)))), - ), - Effect.forkScoped({ startImmediately: true }), + yield* ProviderRefresh.onSwitch( + "azure", + rebind().pipe(Effect.andThen(FiberHandle.run(discovery, ProviderRefresh.switched(discover, apply)))), ) // Deployments load in the background so startup never waits on Azure; the catalog serves until they arrive. yield* rebind().pipe(Effect.andThen(FiberHandle.run(discovery, refresh)), Effect.forkScoped) @@ -455,10 +434,7 @@ export function make( // 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. const bearer = Effect.fn("AzurePlugin.bearer")(function* (url: string) { - const connection = yield* ctx.integration.connection.active(Provider.ID.azure) - const credential = connection - ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) - : undefined + const { credential } = yield* activeCredential(ctx, Provider.ID.azure) if (credential?.type !== "external" || credential.methodID !== methodID) return const target = new URL(url) const scope = @@ -472,7 +448,6 @@ export function make( "http.request", (evt) => Effect.gen(function* () { - if (evt.model.providerID !== Provider.ID.azure) return const authorization = yield* bearer(evt.request.url) if (!authorization) return evt.request.headers.delete("api-key") @@ -486,7 +461,6 @@ export function make( "experimental.ws.handshake", (evt) => Effect.gen(function* () { - if (evt.model.providerID !== Provider.ID.azure) return const authorization = yield* bearer(evt.url) if (!authorization) return delete evt.headers["api-key"] @@ -523,6 +497,15 @@ function expandResourceName(baseURL: string, resourceName: string) { .replaceAll("${AZURE_COGNITIVE_SERVICES_RESOURCE_NAME}", resourceName) } +function azureProvider(provider: { readonly id: string; readonly package: string }) { + return provider.id === Provider.ID.azure || provider.package.startsWith("@opencode/ai/providers/azure/") +} + +// API keys and Azure CLI sessions can list deployments; other sign-ins cannot. +function usable(credential: Credential.Value | undefined): credential is Credential.Key | Credential.External { + return credential?.type === "key" || (credential?.type === "external" && credential.methodID === methodID) +} + // Both the API key and Azure CLI methods store their resource form answer as credential metadata. function credentialResource(credential: Credential.Value | undefined) { const resource = diff --git a/packages/core/src/plugin/provider/chatgpt.ts b/packages/core/src/plugin/provider/chatgpt.ts index 6821a0bb1e3..2e4a88fe4a9 100644 --- a/packages/core/src/plugin/provider/chatgpt.ts +++ b/packages/core/src/plugin/provider/chatgpt.ts @@ -2,12 +2,11 @@ import type { IntegrationOAuthMethodRegistration } from "@opencode/plugin/effect import type { Context } from "@opencode/plugin/effect/plugin" import { define } from "@opencode/plugin/effect/plugin" // import { Deferred, Duration, Effect, Option, Schema, Semaphore, Stream } from "effect" -import { Deferred, Effect, Equal, Option, Schema, Semaphore, Stream } from "effect" +import { Deferred, Effect, Equal, Option, Schema, Semaphore } from "effect" import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" import type { Server, ServerResponse } from "node:http" import { App } from "../../app.js" import { Credential } from "../../credential.js" -import { Bus } from "../../bus.js" import { Integration } from "../../integration.js" // import { IntegrationConnection } from "../../integration/connection.js" import { Model } from "../../model.js" @@ -16,6 +15,8 @@ import { Provider } from "../../provider.js" import { SessionAffinity } from "../../session/affinity.js" import type { PluginInternal } from "../internal.js" import { configuredSettings } from "./configured.js" +import { activeCredential } from "./credential-settings.js" +import { ProviderRefresh } from "./refresh.js" // First-time sign-in registers a user-owned client; OpenAI returns its issued client ID on the callback. const registrationClientID = "dynamic_agent_client" @@ -210,7 +211,6 @@ function listen(server: Server) { export const ChatGPTPlugin = define({ id: "opencode.provider.chatgpt", effect: Effect.fn(function* (ctx) { - const bus = yield* Bus.Service const credentials = yield* Credential.Service const loading = Semaphore.makeUnsafe(1) let chatgpt: Credential.OAuth | undefined @@ -219,10 +219,7 @@ export const ChatGPTPlugin = define({ const load = Effect.fn("ChatGPTPlugin.load")(function* () { // const previous = IntegrationConnection.key(source) - const connection = yield* ctx.integration.connection.active(integrationID) - const credential = connection - ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) - : undefined + const { connection, credential } = yield* activeCredential(ctx, integrationID) chatgpt = credential?.type === "oauth" && credential.methodID === methodID ? credential : undefined const previous = source source = chatgpt ? connection : undefined @@ -359,12 +356,10 @@ export const ChatGPTPlugin = define({ // }), // ) // }) - const reload = () => loading.withPermit(load().pipe(Effect.flatMap((changed) => (changed ? ctx.provider.reload() : Effect.void)))) // .pipe(Effect.andThen(refreshModels().pipe(Effect.forkScoped, Effect.asVoid))) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === integrationID), - Stream.runForEach(reload), - Effect.forkScoped({ startImmediately: true }), + yield* ProviderRefresh.onSwitch( + integrationID, + loading.withPermit(ProviderRefresh.reloadIfChanged(load(), ctx.provider.reload())), ) // yield* refreshModels().pipe(Effect.forkScoped) // yield* Effect.sleep(Duration.minutes(15)).pipe(Effect.andThen(refreshModels), Effect.forever, Effect.forkScoped) diff --git a/packages/core/src/plugin/provider/claude-agent/integration.ts b/packages/core/src/plugin/provider/claude-agent/integration.ts index 3e6f09490af..900f36a0fd0 100644 --- a/packages/core/src/plugin/provider/claude-agent/integration.ts +++ b/packages/core/src/plugin/provider/claude-agent/integration.ts @@ -2,10 +2,8 @@ export * as ClaudeIntegration from "./integration.js" import type { Context } from "@opencode/plugin/effect/plugin" import { Effect, Stream, SubscriptionRef } from "effect" -import { Bus } from "../../../bus.js" -import { Credential } from "../../../credential.js" -import { Integration } from "../../../integration.js" import type { IntegrationConnection } from "../../../integration/connection.js" +import { ProviderRefresh } from "../refresh.js" import { ClaudeProbe } from "./claude/probe.js" import type { ClaudeStatus } from "./claude/status.js" import { ClaudeProvider } from "./provider.js" @@ -46,12 +44,7 @@ export const register = Effect.fn("ClaudeIntegration.register")(function* ( // Check again when the user connects. opencode keeps a problem per connection, so a new connection starts with // none, and the user may have just run `claude auth login`. - const bus = yield* Bus.Service - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === Integration.ID.make(ClaudeProvider.ID)), - Stream.runForEach(() => status.refresh()), - Effect.forkScoped({ startImmediately: true }), - ) + yield* ProviderRefresh.onSwitch(ClaudeProvider.ID, status.refresh()) }) function problem(state: ClaudeStatus.State): IntegrationConnection.Status | undefined { diff --git a/packages/core/src/plugin/provider/credential-settings.ts b/packages/core/src/plugin/provider/credential-settings.ts index 686acb125f3..d44671f93fd 100644 --- a/packages/core/src/plugin/provider/credential-settings.ts +++ b/packages/core/src/plugin/provider/credential-settings.ts @@ -1,8 +1,23 @@ -import { Effect, Stream } from "effect" +import { Effect } from "effect" import type { Context } from "@opencode/plugin/effect/plugin" -import { Bus } from "../../bus.js" import { Credential } from "../../credential.js" import type { Provider } from "../../provider.js" +import { ProviderRefresh } from "./refresh.js" + +/** Resolves the integration's active connection to its credential, or undefined when it cannot be resolved. */ +export const activeCredential = Effect.fnUntraced(function* ( + ctx: Context, + integrationID: string, +): Effect.fn.Return<{ + readonly connection: Effect.Success> + readonly credential: Credential.Value | undefined +}> { + const connection = yield* ctx.integration.connection.active(integrationID) + const credential = connection + ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) + : undefined + return { connection, credential } +}) /** * Tracks provider settings a plugin derives from the active connection's stored credential, such as login form @@ -16,7 +31,6 @@ export const credentialSettings = Effect.fn("ProviderPlugin.credentialSettings") select: (value: Credential.Value) => Provider.Settings, ) { const credentials = yield* Credential.Service - const bus = yield* Bus.Service const load = Effect.fn("ProviderPlugin.credentialSettings.load")(function* () { const connection = yield* ctx.integration.connection.active(integrationID) // Reads the stored value rather than resolving the connection, so startup never refreshes anything. @@ -25,17 +39,14 @@ export const credentialSettings = Effect.fn("ProviderPlugin.credentialSettings") return stored ? select(stored.value) : {} }) const current = { settings: yield* load() } - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === integrationID), - Stream.runForEach(() => - Effect.gen(function* () { - const next = yield* load() - if (JSON.stringify(next) === JSON.stringify(current.settings)) return - current.settings = next - yield* ctx.provider.reload() - }), - ), - Effect.forkScoped({ startImmediately: true }), + yield* ProviderRefresh.onSwitch( + integrationID, + Effect.gen(function* () { + const next = yield* load() + if (JSON.stringify(next) === JSON.stringify(current.settings)) return + current.settings = next + yield* ctx.provider.reload() + }), ) return current }) diff --git a/packages/core/src/plugin/provider/digitalocean.ts b/packages/core/src/plugin/provider/digitalocean.ts index 8a58d119d18..097dcb8b8a3 100644 --- a/packages/core/src/plugin/provider/digitalocean.ts +++ b/packages/core/src/plugin/provider/digitalocean.ts @@ -12,6 +12,7 @@ import { OauthCallbackPage } from "../../oauth/page.js" import { Provider } from "../../provider.js" import { SharedCache } from "../../shared-cache.js" import type { PluginInternal } from "../internal.js" +import { activeCredential } from "./credential-settings.js" import { ProviderRefresh } from "./refresh.js" const providerID = Provider.ID.make("digitalocean") @@ -125,10 +126,7 @@ export const DigitalOceanPlugin = define({ 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 + const { connection, credential } = yield* activeCredential(ctx, integrationID) if (!connection || credential?.type !== "oauth" || credential.methodID !== methodID) return { connection, value: { routers: [] } } const key = `digitalocean:${IntegrationConnection.key(connection)}` diff --git a/packages/core/src/plugin/provider/github-copilot.ts b/packages/core/src/plugin/provider/github-copilot.ts index 61f136bf96a..439a7399cec 100644 --- a/packages/core/src/plugin/provider/github-copilot.ts +++ b/packages/core/src/plugin/provider/github-copilot.ts @@ -14,6 +14,7 @@ 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 { activeCredential } from "./credential-settings.js" import { ProviderRefresh } from "./refresh.js" import type { PluginInternal } from "../internal.js" @@ -176,10 +177,7 @@ export const GithubCopilotPlugin = define({ 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 + const { connection, credential } = yield* activeCredential(ctx, "github-copilot") if (!connection || credential?.type !== "oauth") return { connection, value: {} } const url = copilotBaseURL(credential.metadata) ?? baseURL() const headers = (yield* providers.get(Provider.ID.githubCopilot))?.headers ?? {} @@ -254,7 +252,6 @@ export const GithubCopilotPlugin = define({ yield* ctx.aisdk.hook( "sdk", Effect.fn(function* (evt) { - if (evt.model.providerID !== Provider.ID.githubCopilot) return if (evt.package !== "@ai-sdk/github-copilot") return evt.options.fetch = copilotFetch( typeof evt.options.apiKey === "string" ? evt.options.apiKey : undefined, @@ -264,12 +261,12 @@ export const GithubCopilotPlugin = define({ const mod = yield* Effect.promise(() => import("../../github-copilot/copilot-provider.js")) evt.sdk = mod.createOpenaiCompatible(evt.options) }), + { providerID: Provider.ID.githubCopilot }, ) yield* ctx.session.hook( "model.request", (evt) => Effect.gen(function* () { - if (evt.model.providerID !== Provider.ID.githubCopilot) return const session = yield* ctx.session .get({ sessionID: evt.sessionID }) .pipe(Effect.orElseSucceed(() => undefined)) @@ -287,7 +284,6 @@ export const GithubCopilotPlugin = define({ "title", (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 @@ -310,7 +306,6 @@ export const GithubCopilotPlugin = define({ "http.request", (evt) => Effect.gen(function* () { - if (evt.model.providerID !== Provider.ID.githubCopilot) return const token = evt.request.headers.get("x-api-key") if (!token) return const text = yield* Effect.promise(() => evt.request.clone().text()) @@ -322,7 +317,6 @@ export const GithubCopilotPlugin = define({ yield* ctx.aisdk.hook( "language", Effect.fn(function* (evt) { - if (evt.model.providerID !== Provider.ID.githubCopilot) return if (evt.sdk.responses === undefined && evt.sdk.chat === undefined) { evt.language = evt.sdk.languageModel(evt.model.modelID ?? evt.model.id) return @@ -346,6 +340,7 @@ export const GithubCopilotPlugin = define({ id.startsWith("mai-code-") evt.language = responses ? evt.sdk.responses(id) : evt.sdk.chat(id) }), + { providerID: Provider.ID.githubCopilot }, ) }), } satisfies PluginInternal.InternalPlugin) diff --git a/packages/core/src/plugin/provider/gitlab.ts b/packages/core/src/plugin/provider/gitlab.ts index e009f67d12a..e266316a989 100644 --- a/packages/core/src/plugin/provider/gitlab.ts +++ b/packages/core/src/plugin/provider/gitlab.ts @@ -1,12 +1,11 @@ import os from "os" import { App } from "../../app.js" -import { Clock, Deferred, Effect, Equal, Exit, Option, Schema, Semaphore, Stream } from "effect" +import { Clock, Deferred, Effect, Equal, Exit, Option, Schema, Semaphore } from "effect" import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" import type { Server, ServerResponse } from "node:http" import { define } from "@opencode/plugin/effect/plugin" import { Form } from "@opencode/schema/form" import type { DiscoveredWorkflowModel } from "gitlab-ai-provider" -import { Bus } from "../../bus.js" import { Credential } from "../../credential.js" import { Integration } from "../../integration.js" import { IntegrationConnection } from "../../integration/connection.js" @@ -15,6 +14,7 @@ import { OauthCallbackPage } from "../../oauth/page.js" import { Provider } from "../../provider.js" import type { PluginInternal } from "../internal.js" import { credentialSettings, metadataSettings } from "./credential-settings.js" +import { ProviderRefresh } from "./refresh.js" const providerID = Provider.ID.gitlab const integrationID = Integration.ID.make("gitlab") @@ -55,7 +55,6 @@ export const GitLabPlugin = define({ id: "opencode.provider.gitlab", effect: Effect.fn(function* (ctx) { const providers = yield* Provider.Service - const bus = yield* Bus.Service const http = yield* HttpClient.HttpClient const loading = Semaphore.makeUnsafe(1) const loaded: { @@ -294,13 +293,9 @@ export const GitLabPlugin = define({ ], }) }) - const refresh = () => loading.withPermit(load().pipe(Effect.flatMap((changed) => (changed ? ctx.provider.reload() : Effect.void)))) - 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) + const refresh = loading.withPermit(ProviderRefresh.reloadIfChanged(load(), ctx.provider.reload())) + yield* ProviderRefresh.onSwitch(integrationID, refresh) + yield* refresh.pipe(Effect.forkScoped) yield* ctx.aisdk.hook( "sdk", @@ -330,7 +325,6 @@ export const GitLabPlugin = define({ yield* ctx.aisdk.hook( "language", Effect.fn(function* (evt) { - if (evt.model.providerID !== Provider.ID.gitlab) return const featureFlags = typeof evt.options.featureFlags === "object" && evt.options.featureFlags ? evt.options.featureFlags : {} const id = evt.model.modelID ?? evt.model.id @@ -355,6 +349,7 @@ export const GitLabPlugin = define({ featureFlags, }) }), + { providerID: Provider.ID.gitlab }, ) }), } satisfies PluginInternal.InternalPlugin) diff --git a/packages/core/src/plugin/provider/modal.ts b/packages/core/src/plugin/provider/modal.ts index a4e9905f765..5062f548102 100644 --- a/packages/core/src/plugin/provider/modal.ts +++ b/packages/core/src/plugin/provider/modal.ts @@ -6,6 +6,7 @@ import { ModalModels } from "../../modal/models.js" import { Provider } from "../../provider.js" import { SharedCache } from "../../shared-cache.js" import type { PluginInternal } from "../internal.js" +import { activeCredential } from "./credential-settings.js" import { ProviderRefresh } from "./refresh.js" const providerID = Provider.ID.make("modal") @@ -26,10 +27,7 @@ export const ModalPlugin = define({ 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 { connection, credential } = yield* activeCredential(ctx, "modal") 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 diff --git a/packages/core/src/plugin/provider/openai-context-caps.ts b/packages/core/src/plugin/provider/openai-context-caps.ts index 45c13e5d839..c13f7307957 100644 --- a/packages/core/src/plugin/provider/openai-context-caps.ts +++ b/packages/core/src/plugin/provider/openai-context-caps.ts @@ -1,10 +1,10 @@ import { define } from "@opencode/plugin/effect/plugin" -import { Effect, Stream } from "effect" -import { Bus } from "../../bus.js" -import { Credential } from "../../credential.js" +import { Effect } from "effect" import { Integration } from "../../integration.js" import { Provider } from "../../provider.js" import type { PluginInternal } from "../internal.js" +import { activeCredential } from "./credential-settings.js" +import { ProviderRefresh } from "./refresh.js" const integrationID = Integration.ID.make("openai") const methodIDs = new Set(["chatgpt-browser", "chatgpt-headless", "chatgpt-token-sharing"]) @@ -17,14 +17,10 @@ const currentInput = 500_000 export const OpenAIContextCapsPlugin = define({ id: "opencode.provider.openai-context-caps", effect: Effect.fn(function* (ctx) { - const bus = yield* Bus.Service let signedIn = false const load = Effect.fn("OpenAIContextCapsPlugin.load")(function* () { - const connection = yield* ctx.integration.connection.active(integrationID) - const credential = connection - ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) - : undefined + const { credential } = yield* activeCredential(ctx, integrationID) const previous = signedIn signedIn = credential?.type === "oauth" && methodIDs.has(credential.methodID) return signedIn !== previous @@ -41,11 +37,7 @@ export const OpenAIContextCapsPlugin = define({ }) } }) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === integrationID), - Stream.runForEach(() => load().pipe(Effect.flatMap((changed) => (changed ? ctx.provider.reload() : Effect.void)))), - Effect.forkScoped({ startImmediately: true }), - ) + yield* ProviderRefresh.onSwitch(integrationID, ProviderRefresh.reloadIfChanged(load(), ctx.provider.reload())) }), } satisfies PluginInternal.InternalPlugin) diff --git a/packages/core/src/plugin/provider/openai.ts b/packages/core/src/plugin/provider/openai.ts index 413329b3fff..a03e3e35f26 100644 --- a/packages/core/src/plugin/provider/openai.ts +++ b/packages/core/src/plugin/provider/openai.ts @@ -1,17 +1,18 @@ import type { IntegrationOAuthMethodRegistration } from "@opencode/plugin/effect/integration" import { define } from "@opencode/plugin/effect/plugin" import type { SessionRequest } from "@opencode/plugin/effect/session" -import { Deferred, Effect, Option, Schema, Semaphore, Stream } from "effect" +import { Deferred, Effect, Option, Schema, Semaphore } from "effect" import type { Server } from "node:http" import { App } from "../../app.js" import { Credential } from "../../credential.js" -import { Bus } from "../../bus.js" import { Integration } from "../../integration.js" import { OauthCallbackPage } from "../../oauth/page.js" import { Provider } from "../../provider.js" import { SessionAffinity } from "../../session/affinity.js" import type { PluginInternal } from "../internal.js" import { configuredSettings } from "./configured.js" +import { activeCredential } from "./credential-settings.js" +import { ProviderRefresh } from "./refresh.js" const clientID = "app_EMoamEEZ73f0CkXaXp7hrann" const issuer = "https://auth.openai.com" @@ -234,15 +235,11 @@ const headless = (app: App.Info) => export const OpenAIPlugin = define({ id: "opencode.provider.openai", effect: Effect.fn(function* (ctx) { - const bus = yield* Bus.Service const loading = Semaphore.makeUnsafe(1) let chatgpt: Credential.OAuth | undefined const load = Effect.fn("OpenAIPlugin.load")(function* () { - const connection = yield* ctx.integration.connection.active("openai") - const credential = connection - ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.orElseSucceed(() => undefined)) - : undefined + const { credential } = yield* activeCredential(ctx, "openai") const previous = chatgpt chatgpt = credential?.type === "oauth" && @@ -331,11 +328,9 @@ export const OpenAIPlugin = define({ }) for (const name of ["context", "compaction"] as const) yield* ctx.session.hook(name, omitOutputLimit, { providerID: Provider.ID.openai }) - const refresh = () => loading.withPermit(load().pipe(Effect.flatMap((changed) => (changed ? ctx.provider.reload() : Effect.void)))) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === Integration.ID.make("openai")), - Stream.runForEach(refresh), - Effect.forkScoped({ startImmediately: true }), + yield* ProviderRefresh.onSwitch( + "openai", + loading.withPermit(ProviderRefresh.reloadIfChanged(load(), ctx.provider.reload())), ) }), } satisfies PluginInternal.InternalPlugin) diff --git a/packages/core/src/plugin/provider/opencode.ts b/packages/core/src/plugin/provider/opencode.ts index 2420f2f6e54..18098bbb333 100644 --- a/packages/core/src/plugin/provider/opencode.ts +++ b/packages/core/src/plugin/provider/opencode.ts @@ -1,4 +1,4 @@ -import { Clock, Duration, Effect, Equal, FiberHandle, Option, Schema, Scope, Semaphore, Stream } from "effect" +import { Clock, Duration, Effect, Equal, FiberHandle, Option, Schema, Scope, Semaphore } from "effect" import type { IntegrationOAuthMethodRegistration } from "@opencode/plugin/effect/integration" import { define } from "@opencode/plugin/effect/plugin" import type { SessionHttpResponse } from "@opencode/plugin/effect/session" @@ -19,6 +19,7 @@ import { AbsolutePath } from "../../schema.js" import { Skill } from "../../skill.js" import { SkillDiscovery } from "../../skill/discovery.js" import { WebSearch } from "../../websearch.js" +import { ProviderRefresh } from "./refresh.js" const clientID = "opencode-cli" const methodID = Integration.MethodID.make("device") @@ -174,7 +175,6 @@ export const OpencodePlugin = define< id: "opencode.provider.opencode", effect: Effect.fn(function* (ctx) { const disableFree = ["1", "true"].includes(process.env.OPENCODE_DISABLE_FREE_MODELS?.toLowerCase() ?? "") - const bus = yield* Bus.Service const client = yield* HttpClient.HttpClient // Every request here goes to the Console, which reads the User-Agent to tell which OpenCode a member runs // and whether it evaluates the policies it is being sent. @@ -335,14 +335,8 @@ export const OpencodePlugin = define< execute: (input: WebSearch.ProviderInput) => Effect.gen(function* () { const active = yield* ctx.integration.connection.active("opencode") - if ( - !active || - (connection.type === "credential" - ? active.type !== "credential" || active.id !== connection.id - : active.type !== "env" || active.name !== connection.name) - ) { + if (!active || IntegrationConnection.key(active) !== IntegrationConnection.key(connection)) return yield* Effect.fail(new Error("OpenCode Console connection changed")) - } const credential = yield* ctx.integration.connection.resolve(active) if (!credential || credential.type === "external") return yield* Effect.fail(new Error("OpenCode Console is not connected")) @@ -495,11 +489,7 @@ export const OpencodePlugin = define< onlyIfMissing: true, }) }) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === Integration.ID.make("opencode")), - Stream.runForEach(() => refresh().pipe(Effect.andThen(poll()))), - Effect.forkScoped({ startImmediately: true }), - ) + yield* ProviderRefresh.onSwitch("opencode", refresh().pipe(Effect.andThen(poll()))) yield* poll() }), }) diff --git a/packages/core/src/plugin/provider/refresh.ts b/packages/core/src/plugin/provider/refresh.ts index cd9c78004b1..ef1aa6252ff 100644 --- a/packages/core/src/plugin/provider/refresh.ts +++ b/packages/core/src/plugin/provider/refresh.ts @@ -11,6 +11,21 @@ type Connection = Parameters[0] /** How a load asks `SharedCache.get` for its list. */ export type Options = { readonly interval?: Duration.Input; readonly force?: boolean } +/** Runs `effect` each time the active credential of `integrationID` switches, until the scope closes. */ +export const onSwitch = (integrationID: string, effect: Effect.Effect) => + Effect.gen(function* () { + const bus = yield* Bus.Service + yield* bus.subscribe(Credential.Event.Switched).pipe( + Stream.filter((event) => event.data.integrationID === integrationID), + Stream.runForEach(() => effect), + Effect.forkScoped({ startImmediately: true }), + ) + }) + +/** Runs `reload` when `load` reports that the derived state changed. */ +export const reloadIfChanged = (load: Effect.Effect, reload: Effect.Effect) => + load.pipe(Effect.flatMap((changed) => (changed ? reload : Effect.void))) + /** * 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 @@ -37,7 +52,6 @@ export const start = Effect.fnUntraced(function* (input: { 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() @@ -55,11 +69,7 @@ export const start = Effect.fnUntraced(function* (input: { .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* onSwitch(input.integrationID, FiberHandle.run(loading, switched(load, apply))) yield* FiberHandle.run(loading, refresh, { onlyIfMissing: true }).pipe(Effect.forkScoped) yield* Effect.sleep(interval).pipe( Effect.andThen(FiberHandle.run(loading, refresh, { onlyIfMissing: true })), diff --git a/packages/core/src/plugin/provider/sap-ai-core.ts b/packages/core/src/plugin/provider/sap-ai-core.ts index 4085378e84f..a2de5bc06b5 100644 --- a/packages/core/src/plugin/provider/sap-ai-core.ts +++ b/packages/core/src/plugin/provider/sap-ai-core.ts @@ -11,7 +11,6 @@ export const SapAICorePlugin = define({ yield* ctx.aisdk.hook( "sdk", Effect.fn(function* (evt) { - if (evt.model.providerID !== Provider.ID.make("sap-ai-core")) return const serviceKey = process.env.AICORE_SERVICE_KEY ?? (typeof evt.options.serviceKey === "string" ? evt.options.serviceKey : undefined) @@ -27,13 +26,14 @@ export const SapAICorePlugin = define({ : {}, ) }), + { providerID: Provider.ID.make("sap-ai-core") }, ) yield* ctx.aisdk.hook( "language", Effect.fn(function* (evt) { - if (evt.model.providerID !== Provider.ID.make("sap-ai-core")) return evt.language = evt.sdk(evt.model.modelID ?? evt.model.id) }), + { providerID: Provider.ID.make("sap-ai-core") }, ) }), }) diff --git a/packages/core/src/plugin/provider/snowflake-cortex.ts b/packages/core/src/plugin/provider/snowflake-cortex.ts index 564dfba3fa0..1745b87df12 100644 --- a/packages/core/src/plugin/provider/snowflake-cortex.ts +++ b/packages/core/src/plugin/provider/snowflake-cortex.ts @@ -1,14 +1,14 @@ import { define } from "@opencode/plugin/effect/plugin" import { Form } from "@opencode/schema/form" -import { Clock, Deferred, Effect, Option, Schema, Stream } from "effect" +import { Clock, Deferred, Effect, Option, Schema } from "effect" import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http" import { App } from "../../app.js" -import { Bus } from "../../bus.js" import { Credential } from "../../credential.js" import { Integration } from "../../integration.js" import { OauthCallbackPage } from "../../oauth/page.js" import { Provider } from "../../provider.js" import { configuredSettings } from "./configured.js" +import { ProviderRefresh } from "./refresh.js" const providerID = Provider.ID.make("snowflake-cortex") const integrationID = Integration.ID.make(providerID) @@ -40,7 +40,6 @@ export const SnowflakeCortexPlugin = define({ const http = yield* HttpClient.HttpClient const credentials = yield* Credential.Service const integrations = yield* Integration.Service - const bus = yield* Bus.Service const configured = yield* configuredSettings(providerID) const account = { value: "" } const saved = Effect.gen(function* () { @@ -201,11 +200,7 @@ export const SnowflakeCortexPlugin = define({ }) } }) - yield* bus.subscribe(Credential.Event.Switched).pipe( - Stream.filter((event) => event.data.integrationID === integrationID), - Stream.runForEach(() => load.pipe(Effect.flatMap((changed) => (changed ? ctx.provider.reload() : Effect.void)))), - Effect.forkScoped({ startImmediately: true }), - ) + yield* ProviderRefresh.onSwitch(integrationID, ProviderRefresh.reloadIfChanged(load, ctx.provider.reload())) yield* ctx.session.hook( "http.request", (event) =>