From fbdb81d72644c2ca8f51cf5e3ddbcf3ff8a7dabf Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Fri, 13 Nov 2026 01:02:23 -0600 Subject: [PATCH] fix(core): give local model discovery 15 seconds instead of 1 (#63700) --- packages/core/src/plugin/provider/local.ts | 54 ++++++--- .../core/test/plugin/provider-vllm.test.ts | 104 +++++++++++++++++- 2 files changed, 144 insertions(+), 14 deletions(-) diff --git a/packages/core/src/plugin/provider/local.ts b/packages/core/src/plugin/provider/local.ts index 30e056bd47a..dfebe744066 100644 --- a/packages/core/src/plugin/provider/local.ts +++ b/packages/core/src/plugin/provider/local.ts @@ -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 export function createLocalProviderPlugin(input: { @@ -36,7 +40,8 @@ export function createLocalProviderPlugin(input: { readonly model: (item: NoInfer) => LocalModel }) { const discovery = new Map() - const discoveryLock = Semaphore.makeUnsafe(1) + // One lock per endpoint, so an unresponsive host does not delay discovery of another. + const discoveryLocks = new Map() const suffix = input.stripPathSuffix ?? /\/v1$/ return function make(origin?: string, interval: Duration.Input = "30 seconds") { @@ -55,7 +60,7 @@ export function createLocalProviderPlugin(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(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(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(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(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: (path: string, schema: Schema.Codec) => 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: ( path: string, bodySchema: Schema.Codec, @@ -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 } } diff --git a/packages/core/test/plugin/provider-vllm.test.ts b/packages/core/test/plugin/provider-vllm.test.ts index bb2ffceb99e..5399ec14224 100644 --- a/packages/core/test/plugin/provider-vllm.test.ts +++ b/packages/core/test/plugin/provider-vllm.test.ts @@ -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() + const release = Promise.withResolvers() + 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() + const aborted = Promise.withResolvers() + return { + arrived, + aborted, + slow: Bun.serve({ + port: 0, + fetch: (request) => { + request.signal.addEventListener("abort", () => aborted.resolve()) + arrived.resolve() + return new Promise(() => {}) + }, + }), + 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(() => {