mirrors/opencode
1
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-11-14 20:21:39 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): give local model discovery 15 seconds instead of 1 (#63700)

Browse Source
This commit is contained in:
Aiden Cline
2026-11-13 01:02:23 -06:00
committed by GitHub
Unverified
parent 908d140231
commit fbdb81d726
2 changed files with 144 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/plugin/provider/local.ts
+41 -13
View File
@@ -1,6 +1,6 @@
import { define } from "@opencode/plugin/effect/plugin"
import type { Entry } from "@opencode/schema/config"
import { Duration, Effect, Option, Schedule, type Schema, Semaphore, Stream } from "effect"
import { Duration, Effect, FiberSet, Option, Schedule, type Schema, Semaphore, Stream } from "effect"
import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"
import photonWasm from "#photon-wasm"
import { Config } from "../../config.js"
@@ -24,6 +24,10 @@ export type LocalModel = {
readonly type?: "chat" | "decision"
}
// Remote and tunnelled hosts can take several seconds to connect. Discovery runs in the background, so waiting
// only delays when the models appear.
const requestTimeout = "15 seconds"
export type LocalDiscoveryClient = ReturnType<typeof createClient>
export function createLocalProviderPlugin<Discovered>(input: {
@@ -36,7 +40,8 @@ export function createLocalProviderPlugin<Discovered>(input: {
readonly model: (item: NoInfer<Discovered>) => LocalModel
}) {
const discovery = new Map<string, { checked: number; apiKey?: string; models?: readonly Discovered[] }>()
const discoveryLock = Semaphore.makeUnsafe(1)
// One lock per endpoint, so an unresponsive host does not delay discovery of another.
const discoveryLocks = new Map<string, Semaphore.Semaphore>()
const suffix = input.stripPathSuffix ?? /\/v1$/
return function make(origin?: string, interval: Duration.Input = "30 seconds") {
@@ -55,7 +60,7 @@ export function createLocalProviderPlugin<Discovered>(input: {
? undefined
: input.origin)
const source = { current: configured(yield* config.entries(), input.providerID, defaultOrigin, suffix) }
const loaded = { models: [] as readonly Discovered[], connected: false, hash: "[]" }
const loaded = { models: [] as readonly Discovered[], connected: false, hash: "[]", failing: false }
yield* ctx.integration.transform((integrations) => {
// A key connection lives on the integration (env keys only exist while it does), so keep it then.
@@ -102,7 +107,9 @@ export function createLocalProviderPlugin<Discovered>(input: {
const discover = Effect.fn("LocalProviderPlugin.discover")(function* () {
const current = source.current
if (!current.endpoint) return undefined
return yield* discoveryLock.withPermit(
const lock = discoveryLocks.get(current.root) ?? Semaphore.makeUnsafe(1)
discoveryLocks.set(current.root, lock)
return yield* lock.withPermit(
Effect.gen(function* () {
// A key saved with /connect or set in the integration's env var wins over config, as it does for requests.
const active = yield* activeCredential(ctx, input.providerID)
@@ -125,6 +132,7 @@ export function createLocalProviderPlugin<Discovered>(input: {
const refresh = Effect.fn("LocalProviderPlugin.refresh")(function* () {
const result = yield* discover()
loaded.failing = false
if (!result?.models || result.source !== source.current) return
const hash = JSON.stringify([result.connected, result.models])
if (hash === loaded.hash) return
@@ -136,8 +144,25 @@ export function createLocalProviderPlugin<Discovered>(input: {
})
// Keep the last successful inventory through transient outages instead of flickering model availability.
yield* refresh().pipe(Effect.ignore, Effect.repeat(Schedule.spaced(interval)), Effect.forkScoped)
yield* ProviderRefresh.onSwitch(input.providerID, refresh().pipe(Effect.ignore))
// Only a configured endpoint is expected to answer, and it is reported once per outage.
// A reload doesn't wait for the attempt in flight, so a failure for a replaced source is dropped.
const attempt = Effect.suspend(() => {
const current = source.current
return refresh().pipe(
Effect.catch((cause) => {
if (current !== source.current || !current.explicit || loaded.failing) return Effect.void
loaded.failing = true
return Effect.logWarning(`failed to discover ${input.name} models`, {
baseURL: current.baseURL,
cause,
})
}),
)
})
yield* attempt.pipe(Effect.repeat(Schedule.spaced(interval)), Effect.forkScoped)
// Event handlers start discovery without waiting on it, so a slow endpoint doesn't hold up the next event.
const background = yield* FiberSet.make()
yield* ProviderRefresh.onSwitch(input.providerID, FiberSet.run(background, attempt))
const reload = Effect.fn("LocalProviderPlugin.reload")(function* () {
const next = configured(yield* config.entries(), input.providerID, defaultOrigin, suffix)
if (
@@ -150,9 +175,10 @@ export function createLocalProviderPlugin<Discovered>(input: {
loaded.models = []
loaded.connected = false
loaded.hash = "[]"
loaded.failing = false
yield* ctx.integration.reload()
yield* ctx.provider.reload()
yield* refresh().pipe(Effect.ignore)
yield* FiberSet.run(background, attempt)
})
yield* ctx.event
.subscribe({ types: ["config.updated"] })
@@ -175,11 +201,12 @@ function createClient(
return {
root,
apiKey,
get: (path: string) => http.execute(auth(HttpClientRequest.get(endpoint(path)))).pipe(Effect.timeout("1 second")),
get: (path: string) =>
http.execute(auth(HttpClientRequest.get(endpoint(path)))).pipe(Effect.timeout(requestTimeout)),
getJson: <A, I>(path: string, schema: Schema.Codec<A, I>) =>
http
.execute(auth(HttpClientRequest.get(endpoint(path))))
.pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(schema)), Effect.timeout("1 second")),
.pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(schema)), Effect.timeout(requestTimeout)),
postJson: <B, BI, A, AI>(
path: string,
bodySchema: Schema.Codec<B, BI>,
@@ -190,7 +217,7 @@ function createClient(
HttpClientRequest.schemaBodyJson(bodySchema)(body),
Effect.flatMap(http.execute),
Effect.flatMap(HttpClientResponse.schemaBodyJson(responseSchema)),
Effect.timeout("1 second"),
Effect.timeout(requestTimeout),
),
}
}
@@ -205,9 +232,10 @@ function configured(entries: readonly Entry[], providerID: string, origin: strin
: ""
).replace(/\/+$/, "")
const apiKey = typeof settings?.apiKey === "string" ? settings.apiKey : undefined
if (!URL.canParse(baseURL)) return { baseURL, apiKey }
const explicit = typeof settings?.baseURL === "string"
if (!URL.canParse(baseURL)) return { baseURL, apiKey, explicit }
const url = new URL(baseURL)
if (url.protocol !== "http:" && url.protocol !== "https:") return { baseURL, apiKey }
if (url.protocol !== "http:" && url.protocol !== "https:") return { baseURL, apiKey, explicit }
const prefix = url.pathname.replace(/\/+$/, "").replace(suffix, "")
const endpoint = (path: string) => {
const next = new URL(url)
@@ -216,5 +244,5 @@ function configured(entries: readonly Entry[], providerID: string, origin: strin
next.hash = ""
return next.toString()
}
return { baseURL, apiKey, root: endpoint(""), endpoint }
return { baseURL, apiKey, explicit, root: endpoint(""), endpoint }
}
packages/core/test/plugin/provider-vllm.test.ts
+103 -1
View File
@@ -9,7 +9,8 @@ import { make, VLLMPlugin } from "@opencode/core/plugin/provider/vllm"
import { Provider } from "@opencode/core/provider"
import { Document, Event, Info } from "@opencode/schema/config"
import { describe, expect } from "bun:test"
import { Duration, Effect, Layer, Schema } from "effect"
import { Duration, Effect, Layer, Logger, Schema } from "effect"
import { TestClock } from "effect/testing"
import { eventually } from "../lib/eventually"
import { testEffect } from "../lib/effect"
import { PluginTestLayer } from "./fixture"
@@ -99,6 +100,107 @@ describe("VLLMPlugin", () => {
),
)
it.effect("waits for a slow endpoint instead of dropping its models", () =>
Effect.acquireUseRelease(
Effect.sync(() => {
const arrived = Promise.withResolvers<void>()
const release = Promise.withResolvers<void>()
return {
arrived,
release,
server: Bun.serve({
port: 0,
fetch: async (request) => {
if (new URL(request.url).pathname === "/health") return new Response()
arrived.resolve()
await release.promise
return Response.json({ object: "list", data: [remoteModel("slow-model")] })
},
}),
}
}),
({ arrived, release, server }) =>
Effect.gen(function* () {
const modelState = yield* Model.Service
yield* addPlugin(server.url.origin)
yield* Effect.promise(() => arrived.promise)
// The request is still open after ten seconds, as it is for a remote host that is slow to connect.
yield* TestClock.adjust("10 seconds")
release.resolve()
expect(
yield* eventually(
modelState.get(Provider.ID.make("vllm"), Model.ID.make("slow-model")),
(model) => model !== undefined,
),
).toBeDefined()
}),
({ release, server }) =>
Effect.promise(() => {
release.resolve()
return server.stop(true)
}),
),
)
it.effect("does not report a replaced endpoint's failure against the new one", () =>
Effect.acquireUseRelease(
Effect.sync(() => {
const arrived = Promise.withResolvers<void>()
const aborted = Promise.withResolvers<void>()
return {
arrived,
aborted,
slow: Bun.serve({
port: 0,
fetch: (request) => {
request.signal.addEventListener("abort", () => aborted.resolve())
arrived.resolve()
return new Promise<Response>(() => {})
},
}),
healthy: Bun.serve({
port: 0,
fetch: (request) =>
new URL(request.url).pathname === "/health"
? new Response()
: Response.json({ object: "list", data: [remoteModel("healthy-model")] }),
}),
}
}),
({ arrived, aborted, slow, healthy }) =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const modelState = yield* Model.Service
const config = yield* Config.Test
const warnings: unknown[] = []
yield* config.setEntries([configuration({ baseURL: `${slow.url.origin}/v1` })])
yield* addPlugin(slow.url.origin).pipe(
Effect.provide(
Logger.layer([
Logger.make((entry) => {
if (entry.logLevel === "Warn") warnings.push(entry.message)
}),
]),
),
)
yield* Effect.promise(() => arrived.promise)
yield* config.setEntries([configuration({ baseURL: `${healthy.url.origin}/v1` })])
yield* bus.publish(Event.Updated, {})
yield* eventually(
modelState.get(Provider.ID.make("vllm"), Model.ID.make("healthy-model")),
(model) => model !== undefined,
)
// The request to the replaced endpoint is still open, and now times out.
yield* TestClock.adjust("15 seconds")
yield* Effect.promise(() => aborted.promise)
expect(warnings).toEqual([])
}),
({ slow, healthy }) => Effect.promise(() => Promise.all([slow.stop(true), healthy.stop(true)])),
),
)
it.live("refreshes inventory while retaining the last success through transient failures", () =>
Effect.acquireUseRelease(
Effect.sync(() => {