refactor(core): share credential helpers in provider plugins (#61179)

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 =

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)

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 {

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<ReturnType<Context["integration"]["connection"]["active"]>>
+  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
 })

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)}`

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)

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)

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

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)
 

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)

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()
   }),
 })

packages/core/src/plugin/provider/refresh.ts

@@ -11,6 +11,21 @@ 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 }
 
+/** Runs `effect` each time the active credential of `integrationID` switches, until the scope closes. */
+export const onSwitch = <E, R>(integrationID: string, effect: Effect.Effect<unknown, E, R>) =>
+  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 = <E, R>(load: Effect.Effect<boolean, E, R>, reload: Effect.Effect<void>) =>
+  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* <A>(input: {
   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>()
@@ -55,11 +69,7 @@ export const start = Effect.fnUntraced(function* <A>(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 })),

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") },
     )
   }),
 })

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) =>