From 9fffed89033a096bc32df66fdefd80de4a11a630 Mon Sep 17 00:00:00 2001 From: Ben Brougher Date: Wed, 21 Oct 2026 21:54:38 -0700 Subject: [PATCH] fix: discover models without blocking plugin activation OpenCode can load this plugin in a separate process for every location, so waiting on CLIProxyAPI or models.dev can exceed the activation budget. Register the provider first and fill models from a shared disk cache and background refresh. A failed fetch no longer refreshes the cache timestamp, and a process that loses the metadata lock waits for the winner's cache. --- CHANGELOG.md | 12 ++ README.md | 33 ++-- src/cache.test.ts | 128 +++++++++++++++ src/cache.ts | 223 +++++++++++++++++++++++++ src/index.test.ts | 408 ++++++++++++++++++++++++++++++++++++++++++---- src/index.ts | 292 ++++++++++++++++++++++++++++----- 6 files changed, 1020 insertions(+), 76 deletions(-) create mode 100644 src/cache.test.ts create mode 100644 src/cache.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 95941db..5ce06c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project follows [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Fixed + +- Register the provider before any network request and load both the + CLIProxyAPI catalog and optional models.dev metadata in the background. + OpenCode can activate this plugin in a separate process for every location, + so waiting for either request could exceed the activation budget and leave + the plugin failed. A disk cache, shared by server URL, supplies the last + catalogs immediately. A file lock ensures only one process downloads the + metadata catalog; the others wait for that download. A failed request does + not extend the cache lifetime, and disposing the plugin stops further + provider reloads. + ## [0.3.0] - 2026-10-04 ### Added diff --git a/README.md b/README.md index 255f9f4..79d31ca 100644 --- a/README.md +++ b/README.md @@ -12,13 +12,13 @@ This is the OpenCode V2 port of targets OpenCode V1. The V1 plugin API does not run in V2, so this package is a separate release built on `@opencode/plugin`. -The plugin discovers CLIProxyAPI's live `/v1/models` catalog whenever it loads -(and refreshes it in the background), then registers them as a provider with -`ctx.provider.transform`. Available models appear in the normal `/models` -picker under **CLIProxyAPI**. Model names, capabilities, limits, costs, and -reasoning variants are enriched from live metadata on -[models.dev](https://models.dev/). Deployment limits that the CLIProxyAPI server -itself reports take precedence (see +The plugin registers a provider with `ctx.provider.transform` immediately, then +discovers CLIProxyAPI's live `/v1/models` catalog in the background. Available +models appear in the normal `/models` picker under **CLIProxyAPI**. Model names, +capabilities, limits, costs, and reasoning variants are enriched from live +metadata on [models.dev](https://models.dev/), also in the background. Neither +request can delay plugin activation. Deployment limits that the CLIProxyAPI +server itself reports take precedence (see [Live model metadata](#live-model-metadata)). No model IDs are hard-coded. Claude models owned by Anthropic use OpenCode's Anthropic Messages runtime @@ -89,12 +89,23 @@ pick up changes immediately. | `providerName` | `CLIProxyAPI` | Name displayed in the model picker | | `protocol` | `chat` | Default protocol for non-Anthropic models: `chat` uses `/chat/completions`; `responses` uses `/responses`. Claude models still use `/v1/messages`. | | `modelMetadataURL` | `https://models.dev/api.json` | Dynamic model metadata. Set to `false` to disable enrichment and use only inferred defaults. | -| `discoveryTimeoutMs` | `10000` | Startup model-discovery timeout | +| `discoveryTimeoutMs` | `10000` | CLIProxyAPI model-discovery timeout. Also bounds the optional, non-blocking metadata request. | | `refreshIntervalMs` | `300000` | Background catalog refresh interval. Set to `0` to refresh only at startup. | -If model metadata cannot be reached, the plugin logs a warning and keeps the -CLIProxyAPI-discovered models available with inferred capabilities and the -configured default protocol. +The plugin registers the provider before contacting either service, so a slow +CLIProxyAPI server or models.dev cannot exceed OpenCode's plugin activation +budget. The model list is filled in as soon as discovery finishes and is +reloaded when optional metadata arrives. If metadata is slow or cannot be +reached, the plugin logs a warning and keeps the CLIProxyAPI-discovered models +with inferred capabilities and the configured default protocol. + +OpenCode may load the plugin in a separate process for every location. The last +successful catalogs are cached under `~/.cache/opencode2-cliproxyapi` (or +`CLIPROXYAPI_CACHE_DIR`) so later processes can register models immediately. +The catalog cache is shared by server URL, not by API key. A file lock lets +only one process download the metadata catalog; the others wait for that +download and then read the cache. A failed request does not extend the cache +lifetime, and disposing the plugin stops it from reloading the provider. ### Optional environment variables diff --git a/src/cache.test.ts b/src/cache.test.ts new file mode 100644 index 0000000..72dc1db --- /dev/null +++ b/src/cache.test.ts @@ -0,0 +1,128 @@ +import { mkdtemp, mkdir, readFile, stat, utimes, writeFile } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { describe, expect, test } from "bun:test" +import { + acquireMetadataLock, + cacheInternals, + catalogCacheIsFresh, + LOCK_STALE_MS, + metadataCacheIsFresh, + readCatalogCache, + readMetadataCache, + writeCatalogCache, + writeMetadataCache, +} from "./cache.js" + +async function root() { + return mkdtemp(join(tmpdir(), "cliproxyapi-cache-")) +} + +describe("disk cache", () => { + test("round-trips the model catalog and metadata", async () => { + const directory = await root() + await writeCatalogCache("http://cliproxy.test/v1", [{ id: "chat-model" }], directory) + await writeMetadataCache( + "https://models.dev/api.json", + { chat: { models: { "chat-model": { name: "Chat Model" } } } }, + directory, + ) + + expect(await readCatalogCache("http://cliproxy.test/v1", directory)).toMatchObject({ + models: [{ id: "chat-model" }], + }) + const metadata = await readMetadataCache("https://models.dev/api.json", directory) + expect(metadata?.catalog.chat?.models["chat-model"]?.name).toBe("Chat Model") + expect(metadataCacheIsFresh(metadata?.fetchedAt ?? 0)).toBe(true) + expect(catalogCacheIsFresh(Date.now())).toBe(true) + expect(catalogCacheIsFresh(Date.now() - 31_000)).toBe(false) + expect(metadataCacheIsFresh(Date.now() - 7 * 60 * 60_000)).toBe(false) + }) + + test("lets only one process hold the metadata lock", async () => { + const directory = await root() + const first = await acquireMetadataLock("https://models.dev/api.json", directory) + const second = await acquireMetadataLock("https://models.dev/api.json", directory) + + expect(first).toBeFunction() + expect(second).toBeUndefined() + + await first?.() + const third = await acquireMetadataLock("https://models.dev/api.json", directory) + expect(third).toBeFunction() + await third?.() + }) + + test("ignores a malformed cache file", async () => { + const directory = await root() + await Bun.write(join(directory, "not-used.json"), "{") + expect(await readCatalogCache("missing", directory)).toBeUndefined() + expect(await readMetadataCache("missing", directory)).toBeUndefined() + }) + + test("rejects a catalog entry that is missing a model id", async () => { + const directory = await root() + const path = cacheInternals.catalogPath("http://cliproxy.test/v1", directory) + await mkdir(directory, { recursive: true }) + await writeFile( + path, + JSON.stringify({ + version: 1, + fetchedAt: Date.now(), + models: [{ ownedBy: "acme" }, { id: "" }], + }), + ) + + expect(await readCatalogCache("http://cliproxy.test/v1", directory)).toBeUndefined() + }) + + test("removes a temporary file when replacing the cache fails", async () => { + const directory = await root() + const path = join(directory, "nested", "catalog.json") + await mkdir(path, { recursive: true }) + + await expect(cacheInternals.writeJson(path, { ok: false })).rejects.toThrow() + const entries = await Array.fromAsync( + new Bun.Glob("*.tmp").scan({ cwd: join(directory, "nested") }), + ) + expect(entries).toEqual([]) + }) + + test("replaces a stale metadata lock and leaves a live lock alone", async () => { + const directory = await root() + const url = "https://models.dev/api.json" + const path = cacheInternals.lockPath(`metadata-${cacheInternals.cacheKey(url)}`, directory) + await mkdir(directory, { recursive: true }) + await writeFile(path, "stale") + const stale = new Date(Date.now() - LOCK_STALE_MS - 1_000) + await utimes(path, stale, stale) + + const recovered = await acquireMetadataLock(url, directory) + expect(recovered).toBeFunction() + expect(await readFile(path, "utf8")).not.toBe("stale") + + const blocked = await acquireMetadataLock(url, directory) + expect(blocked).toBeUndefined() + const age = Date.now() - (await stat(path)).mtimeMs + expect(age).toBeLessThan(LOCK_STALE_MS) + + await recovered?.() + expect(await readFile(path, "utf8").catch(() => undefined)).toBeUndefined() + }) + + test("does not delete a lock that another process has taken over", async () => { + const directory = await root() + const url = "https://models.dev/api.json" + const path = cacheInternals.lockPath(`metadata-${cacheInternals.cacheKey(url)}`, directory) + await mkdir(directory, { recursive: true }) + await writeFile(path, "previous") + const stale = new Date(Date.now() - LOCK_STALE_MS - 1_000) + await utimes(path, stale, stale) + + const recovered = await acquireMetadataLock(url, directory) + await writeFile(path, "taken-over") + await recovered?.() + + expect(await readFile(path, "utf8")).toBe("taken-over") + }) +}) diff --git a/src/cache.ts b/src/cache.ts new file mode 100644 index 0000000..db26ae8 --- /dev/null +++ b/src/cache.ts @@ -0,0 +1,223 @@ +import { createHash } from "node:crypto" +import { mkdir, open, readFile, rename, stat, unlink, writeFile } from "node:fs/promises" +import { homedir } from "node:os" +import { dirname, join } from "node:path" +import type { CatalogModel, ModelMetadataCatalog } from "./catalog.js" + +const CACHE_VERSION = 1 +const METADATA_TTL_MS = 6 * 60 * 60_000 +const CATALOG_TTL_MS = 30_000 +// Longer than one metadata request. discoveryTimeoutMs defaults to 10s, and a +// cold models.dev download can use most of that. A shorter stale window lets a +// second process steal the lock from a download that is still running. +export const LOCK_STALE_MS = 3 * 60_000 + +type CatalogCache = { + version: number + fetchedAt: number + models: CatalogModel[] +} + +type MetadataCache = { + version: number + fetchedAt: number + catalog: ModelMetadataCatalog +} + +export function cacheRoot() { + return ( + process.env.CLIPROXYAPI_CACHE_DIR ?? + join(process.env.XDG_CACHE_HOME ?? join(homedir(), ".cache"), "opencode2-cliproxyapi") + ) +} + +function cacheKey(value: string) { + return createHash("sha256").update(value).digest("hex").slice(0, 16) +} + +function catalogPath(key: string, root: string) { + return join(root, `catalog-${cacheKey(key)}.json`) +} + +function metadataPath(url: string, root: string) { + return join(root, `metadata-${cacheKey(url)}.json`) +} + +function lockPath(name: string, root: string) { + return join(root, `${name}.lock`) +} + +async function readJson(path: string): Promise { + try { + return JSON.parse(await readFile(path, "utf8")) + } catch { + return undefined + } +} + +async function writeJson(path: string, value: unknown) { + const temporary = `${path}.${process.pid}.${Date.now()}.${Math.random().toString(16).slice(2)}.tmp` + await mkdir(dirname(path), { recursive: true }) + try { + await writeFile(temporary, JSON.stringify(value)) + await rename(temporary, path) + } catch (error) { + await unlink(temporary).catch(() => {}) + throw error + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +} + +function isCatalogModel(value: unknown): value is CatalogModel { + if (!isRecord(value) || typeof value.id !== "string" || value.id.trim() === "") return false + if (value.ownedBy !== undefined && typeof value.ownedBy !== "string") return false + if (value.live === undefined) return true + if (!isRecord(value.live)) return false + + const live = value.live + if (live.name !== undefined && typeof live.name !== "string") return false + if (live.limit === undefined) return true + if (!isRecord(live.limit)) return false + const limit = live.limit + return ( + (limit.context === undefined || typeof limit.context === "number") && + (limit.output === undefined || typeof limit.output === "number") + ) +} + +export async function readCatalogCache( + key: string, + root = cacheRoot(), +): Promise<{ models: CatalogModel[]; fetchedAt: number } | undefined> { + const value = await readJson(catalogPath(key, root)) + if (!isRecord(value) || value.version !== CACHE_VERSION) return + if (!Array.isArray(value.models) || !value.models.every(isCatalogModel)) return + if (typeof value.fetchedAt !== "number") return + + return { models: value.models, fetchedAt: value.fetchedAt } +} + +export async function writeCatalogCache(key: string, models: CatalogModel[], root = cacheRoot()) { + const value: CatalogCache = { + version: CACHE_VERSION, + fetchedAt: Date.now(), + models, + } + await writeJson(catalogPath(key, root), value) +} + +export async function readMetadataCache( + url: string, + root = cacheRoot(), +): Promise<{ catalog: ModelMetadataCatalog; fetchedAt: number } | undefined> { + const value = await readJson(metadataPath(url, root)) + if (!isRecord(value) || value.version !== CACHE_VERSION) return + if (!isRecord(value.catalog) || typeof value.fetchedAt !== "number") return + + return { + catalog: value.catalog as ModelMetadataCatalog, + fetchedAt: value.fetchedAt, + } +} + +export function metadataCacheIsFresh(fetchedAt: number, now = Date.now()) { + return now - fetchedAt < METADATA_TTL_MS +} + +export function catalogCacheIsFresh(fetchedAt: number, now = Date.now()) { + return now - fetchedAt < CATALOG_TTL_MS +} + +export async function writeMetadataCache( + url: string, + catalog: ModelMetadataCatalog, + root = cacheRoot(), +) { + const value: MetadataCache = { + version: CACHE_VERSION, + fetchedAt: Date.now(), + catalog, + } + await writeJson(metadataPath(url, root), value) +} + +async function tryAcquire(path: string): Promise<(() => Promise) | undefined> { + const token = `${process.pid}.${Date.now()}.${Math.random()}` + let handle: Awaited> | undefined + try { + handle = await open(path, "wx") + } catch (error) { + if (isErrno(error, "EEXIST")) return + throw error + } + + try { + await handle.writeFile(token) + } catch (error) { + // This exclusive create belongs to this attempt. Remove it so a failed + // write cannot leave a lock that no process will release. + await unlink(path).catch(() => {}) + throw error + } finally { + await handle.close() + } + + return async () => { + // A stale lock may already have been taken over. Only the owner may remove it. + const current = await readFile(path, "utf8").catch(() => undefined) + if (current === token) await unlink(path).catch(() => {}) + } +} + +function isErrno(error: unknown, code: string) { + return typeof error === "object" && error !== null && "code" in error && error.code === code +} + +/** + * Allow only one plugin process to download the large metadata catalog at a + * time. OpenCode may load this plugin in a separate process for every location, + * so an in-memory promise is not enough to stop a download stampede. + * + * Returns undefined when another live process holds the lock. Throws when the + * lock directory itself cannot be created, so the caller can fall back to an + * uncoordinated download instead of skipping metadata entirely. + */ +export async function acquireMetadataLock( + url: string, + root = cacheRoot(), +): Promise<(() => Promise) | undefined> { + const path = lockPath(`metadata-${cacheKey(url)}`, root) + await mkdir(root, { recursive: true }) + + const first = await tryAcquire(path) + if (first) return first + + // Recover a lock left behind by a crashed plugin process. A lock younger + // than LOCK_STALE_MS is still owned by a live download. + let stale = false + try { + const info = await stat(path) + stale = Date.now() - info.mtimeMs > LOCK_STALE_MS + } catch (error) { + if (!isErrno(error, "ENOENT")) throw error + stale = true + } + + if (!stale) return + + await unlink(path).catch((error) => { + if (!isErrno(error, "ENOENT")) throw error + }) + return tryAcquire(path) +} + +export const cacheInternals = { + cacheKey, + catalogPath, + metadataPath, + lockPath, + writeJson, +} diff --git a/src/index.test.ts b/src/index.test.ts index a61f4f2..0bfb1f7 100644 --- a/src/index.test.ts +++ b/src/index.test.ts @@ -1,12 +1,55 @@ +import { mkdtempSync } from "node:fs" +import { mkdir } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" + +let metadataURLCounter = 0 + +function uniqueMetadataURL() { + metadataURLCounter += 1 + return `https://models.test/${metadataURLCounter}/api.json` +} + +function uniqueBaseURL() { + metadataURLCounter += 1 + return `http://cliproxy.test:${8300 + metadataURLCounter}` +} + +async function useCacheDir() { + const directory = mkdtempSync(join(tmpdir(), "cliproxyapi-test-")) + process.env.CLIPROXYAPI_CACHE_DIR = directory + return directory +} + +async function restoreFetch(original: typeof globalThis.fetch) { + await pluginTesting.settled + globalThis.fetch = original +} + import { describe, expect, test } from "bun:test" import type { Model, Plugin as PluginTypes, Provider } from "@opencode/plugin" -import plugin, { buildModel, resolveLimit } from "./index.js" +import { + acquireMetadataLock, + cacheInternals, + readCatalogCache, + writeCatalogCache, + writeMetadataCache, +} from "./cache.js" +import plugin, { buildModel, pluginTesting, resolveLimit } from "./index.js" type Registered = { info: Provider.Info models: readonly Model.Info[] } +async function waitUntil(predicate: () => boolean, timeoutMs = 2_000) { + const deadline = Date.now() + timeoutMs + while (!predicate()) { + if (Date.now() >= deadline) throw new Error("timed out waiting for condition") + await Bun.sleep(5) + } +} + function context( options: Record, existing?: { settings?: Record }, @@ -44,32 +87,48 @@ function context( describe("plugin", () => { test("registers a provider from the discovered CLIProxyAPI catalog", async () => { + await useCacheDir() const requests: Request[] = [] const originalFetch = globalThis.fetch globalThis.fetch = async (input, init) => { - requests.push(new Request(input as RequestInfo, init)) + const request = new Request(input as RequestInfo, init) + requests.push(request) + if (request.url.endsWith("/api.json") && request.url !== "http://cliproxy.test:8317/v1/models") { + return Response.json({ + acme: { + models: { + "gpt-5.6-terra": { + name: "GPT 5.6 Terra", + limit: { context: 400_000, output: 128_000 }, + }, + }, + }, + }) + } return Response.json({ data: [{ id: "gpt-5.6-terra" }, { id: "gemini-3.1-flash-image" }], }) } try { + const metadataURL = uniqueMetadataURL() const { ctx, registered } = context({ baseURL: "http://cliproxy.test:8317", apiKey: "secret", + modelMetadataURL: metadataURL, }) await plugin.setup(ctx) + await waitUntil(() => registered.at(-1)?.models[0]?.limit?.context === 400_000) expect(requests).toHaveLength(2) const modelRequest = requests.find( (request) => request.url === "http://cliproxy.test:8317/v1/models", ) expect(modelRequest?.headers.get("authorization")).toBe("Bearer secret") - expect(requests.some((request) => request.url === "https://models.dev/api.json")).toBe(true) + expect(requests.some((request) => request.url === metadataURL)).toBe(true) - expect(registered).toHaveLength(1) - expect(registered[0]?.info).toMatchObject({ + expect(registered.at(-1)?.info).toMatchObject({ id: "cliproxyapi", name: "CLIProxyAPI", activation: "enabled", @@ -79,16 +138,21 @@ describe("plugin", () => { apiKey: "secret", }, }) - expect(registered[0]?.models.map((model) => model.id)).toEqual([ + expect(registered.at(-1)?.models.map((model) => model.id)).toEqual([ "gpt-5.6-terra", "gemini-3.1-flash-image", ]) + expect(registered.at(-1)?.models[0]?.limit).toEqual({ + context: 400_000, + output: 128_000, + }) } finally { - globalThis.fetch = originalFetch + await restoreFetch(originalFetch) } }) test("reads connection settings from an existing provider config", async () => { + await useCacheDir() const requests: Request[] = [] const originalFetch = globalThis.fetch globalThis.fetch = async (input, init) => { @@ -118,11 +182,12 @@ describe("plugin", () => { apiKey: "from-config", }) } finally { - globalThis.fetch = originalFetch + await restoreFetch(originalFetch) } }) test("selects the responses package when configured", async () => { + await useCacheDir() const originalFetch = globalThis.fetch globalThis.fetch = async () => Response.json({ data: [{ id: "chat-model" }] }) @@ -139,28 +204,304 @@ describe("plugin", () => { "@opencode/ai/providers/openai/responses", ) } finally { - globalThis.fetch = originalFetch + await restoreFetch(originalFetch) } }) test("keeps discovered models available when model metadata is unavailable", async () => { + await useCacheDir() const originalFetch = globalThis.fetch globalThis.fetch = async (input) => { - if (String(input) === "https://models.dev/api.json") { + if (String(input).startsWith("https://models.test/")) { return Response.json({ error: "unavailable" }, { status: 503 }) } return Response.json({ data: [{ id: "chat-model", owned_by: "acme" }] }) } try { - const { ctx, registered } = context({ baseURL: "http://cliproxy.test:8317" }) + const { ctx, registered } = context({ + baseURL: "http://cliproxy.test:8317", + modelMetadataURL: uniqueMetadataURL(), + }) + + await plugin.setup(ctx) + await waitUntil(() => registered.at(-1)?.models[0]?.id === "chat-model") + + expect(registered.at(-1)?.models.map((model) => model.id)).toEqual(["chat-model"]) + expect(registered.at(-1)?.models[0]?.package).toBeUndefined() + } finally { + await restoreFetch(originalFetch) + } + }) + + test("registers models before optional metadata resolves", async () => { + await useCacheDir() + const originalFetch = globalThis.fetch + let resolveMetadata: (response: Response) => void = () => {} + const metadataResponse = new Promise((resolve) => { + resolveMetadata = resolve + }) + let metadataRequests = 0 + + const metadataURL = uniqueMetadataURL() + const baseURL = uniqueBaseURL() + globalThis.fetch = async (input) => { + if (String(input) === metadataURL) { + metadataRequests++ + return metadataResponse + } + return Response.json({ data: [{ id: "chat-model" }] }) + } + + try { + const { ctx, registered } = context({ + baseURL, + modelMetadataURL: metadataURL, + }) + + await plugin.setup(ctx) + + expect(registered[0]?.models).toEqual([]) + await waitUntil(() => registered.at(-1)?.models[0]?.name === "Chat Model") + await waitUntil(() => metadataRequests === 1) + + resolveMetadata( + Response.json({ chat: { models: { "chat-model": { name: "Enriched Model" } } } }), + ) + await waitUntil(() => registered.at(-1)?.models[0]?.name === "Enriched Model") + expect(registered.at(-1)?.models[0]?.name).toBe("Enriched Model") + } finally { + await restoreFetch(originalFetch) + } + }) + + test("registers models from the disk cache before the catalog request resolves", async () => { + await useCacheDir() + const originalFetch = globalThis.fetch + let resolveCatalog: (response: Response) => void = () => {} + const catalogResponse = new Promise((resolve) => { + resolveCatalog = resolve + }) + globalThis.fetch = async () => catalogResponse + + try { + const { ctx, registered } = context({ + baseURL: "http://cached.test:8317", + modelMetadataURL: false, + }) + await writeCatalogCache("http://cached.test:8317/v1", [ + { id: "cached-model", live: { name: "Cached Model" } }, + ]) + + await plugin.setup(ctx) + + expect(registered[0]?.models.map((model) => model.id)).toEqual(["cached-model"]) + expect(registered[0]?.models[0]?.name).toBe("Cached Model") + resolveCatalog(Response.json({ data: [{ id: "live-model", display_name: "Live Model" }] })) + await waitUntil(() => registered.at(-1)?.models[0]?.name === "Live Model") + } finally { + resolveCatalog(Response.json({ data: [] })) + await restoreFetch(originalFetch) + } + }) + + test("registers the provider before the catalog request resolves", async () => { + await useCacheDir() + const originalFetch = globalThis.fetch + let resolveCatalog: (response: Response) => void = () => {} + const catalogResponse = new Promise((resolve) => { + resolveCatalog = resolve + }) + + const baseURL = uniqueBaseURL() + globalThis.fetch = async () => catalogResponse + + try { + const { ctx, registered } = context({ + baseURL, + modelMetadataURL: false, + }) await plugin.setup(ctx) - expect(registered[0]?.models.map((model) => model.id)).toEqual(["chat-model"]) - expect(registered[0]?.models[0]?.package).toBeUndefined() + expect(registered[0]?.models).toEqual([]) + + resolveCatalog(Response.json({ data: [{ id: "chat-model" }] })) + await waitUntil(() => registered.at(-1)?.models[0]?.name === "Chat Model") + expect(registered.at(-1)?.models[0]?.name).toBe("Chat Model") + } finally { + resolveCatalog(Response.json({ data: [{ id: "chat-model" }] })) + await restoreFetch(originalFetch) + } + }) + + test("shares one metadata request across concurrent plugin instances", async () => { + await useCacheDir() + const originalFetch = globalThis.fetch + let releaseMetadata: () => void = () => {} + const metadataGate = new Promise((resolve) => { + releaseMetadata = resolve + }) + let metadataRequests = 0 + + const metadataURL = uniqueMetadataURL() + const baseURL = uniqueBaseURL() + globalThis.fetch = async (input) => { + if (String(input) === metadataURL) { + metadataRequests++ + await metadataGate + return Response.json({ + chat: { models: { "chat-model": { name: "Shared Metadata" } } }, + }) + } + return Response.json({ data: [{ id: "chat-model" }] }) + } + + try { + const first = context({ baseURL, modelMetadataURL: metadataURL }) + const second = context({ baseURL, modelMetadataURL: metadataURL }) + + await Promise.all([plugin.setup(first.ctx), plugin.setup(second.ctx)]) + + expect(first.registered[0]?.models).toEqual([]) + expect(second.registered[0]?.models).toEqual([]) + await waitUntil(() => first.registered.at(-1)?.models[0]?.name === "Chat Model") + await waitUntil(() => second.registered.at(-1)?.models[0]?.name === "Chat Model") + await waitUntil(() => metadataRequests === 1) + expect(metadataRequests).toBe(1) + + releaseMetadata() + await waitUntil(() => first.registered.at(-1)?.models[0]?.name === "Shared Metadata") + await waitUntil(() => second.registered.at(-1)?.models[0]?.name === "Shared Metadata") + expect(first.registered.at(-1)?.models[0]?.name).toBe("Shared Metadata") + expect(second.registered.at(-1)?.models[0]?.name).toBe("Shared Metadata") } finally { - globalThis.fetch = originalFetch + releaseMetadata() + await restoreFetch(originalFetch) + } + }) + + test("shares one catalog cache across API keys and does not refresh it after a failure", async () => { + const directory = await useCacheDir() + const baseURL = uniqueBaseURL() + const normalized = `${baseURL}/v1` + const originalFetch = globalThis.fetch + let catalogRequests = 0 + globalThis.fetch = async () => { + catalogRequests++ + if (catalogRequests === 1) return Response.json({ data: [{ id: "shared-model" }] }) + return Response.json({ error: "down" }, { status: 503 }) + } + + try { + const first = context({ baseURL, apiKey: "one", modelMetadataURL: false }) + await plugin.setup(first.ctx) + await waitUntil(() => first.registered.at(-1)?.models[0]?.id === "shared-model") + await pluginTesting.settled + + const cached = await readCatalogCache(normalized, directory) + expect(cached?.models).toEqual([{ id: "shared-model" }]) + const fetchedAt = cached?.fetchedAt + + const second = context({ baseURL, apiKey: "two", modelMetadataURL: false }) + await plugin.setup(second.ctx) + expect(second.registered[0]?.models.map((model) => model.id)).toEqual(["shared-model"]) + + await pluginTesting.refresh?.() + const afterFailure = await readCatalogCache(normalized, directory) + expect(afterFailure?.fetchedAt).toBe(fetchedAt) + expect(afterFailure?.models).toEqual([{ id: "shared-model" }]) + } finally { + await restoreFetch(originalFetch) + } + }) + + test("waits for another process to finish the metadata download", async () => { + await useCacheDir() + const metadataURL = uniqueMetadataURL() + const baseURL = uniqueBaseURL() + const originalFetch = globalThis.fetch + let metadataRequests = 0 + + globalThis.fetch = async (input) => { + if (String(input) === metadataURL) { + metadataRequests++ + return Response.json({ chat: { models: { "chat-model": { name: "Should Not Fetch" } } } }) + } + return Response.json({ data: [{ id: "chat-model" }] }) + } + + const release = await acquireMetadataLock(metadataURL) + expect(release).toBeFunction() + + try { + const { ctx, registered } = context({ + baseURL, + modelMetadataURL: metadataURL, + discoveryTimeoutMs: 2_000, + }) + const setup = plugin.setup(ctx) + await waitUntil(() => registered.at(-1)?.models[0]?.name === "Chat Model") + await writeMetadataCache(metadataURL, { + chat: { models: { "chat-model": { name: "Locked Metadata" } } }, + }) + await waitUntil(() => registered.at(-1)?.models[0]?.name === "Locked Metadata") + expect(metadataRequests).toBe(0) + await setup + } finally { + await release?.() + await restoreFetch(originalFetch) + } + }) + + test("does not reload the provider after the plugin is disposed", async () => { + await useCacheDir() + const originalFetch = globalThis.fetch + let resolveCatalog: (response: Response) => void = () => {} + const catalogResponse = new Promise((resolve) => { + resolveCatalog = resolve + }) + globalThis.fetch = async () => catalogResponse + + try { + const { ctx, registered } = context({ + baseURL: uniqueBaseURL(), + modelMetadataURL: false, + }) + const cleanup = await plugin.setup(ctx) + expect(registered).toHaveLength(1) + const pending = pluginTesting.refresh?.() + + await cleanup?.() + resolveCatalog(Response.json({ data: [{ id: "late-model" }] })) + await pending + + expect(registered).toHaveLength(1) + expect(registered[0]?.models).toEqual([]) + } finally { + resolveCatalog(Response.json({ data: [] })) + await restoreFetch(originalFetch) + } + }) + + test("ignores a catalog cache entry that is not a model list", async () => { + const directory = await useCacheDir() + const baseURL = uniqueBaseURL() + await mkdir(directory, { recursive: true }) + await Bun.write( + cacheInternals.catalogPath(`${baseURL}/v1`, directory), + JSON.stringify({ version: 1, fetchedAt: Date.now(), models: [{ ownedBy: "acme" }] }), + ) + const originalFetch = globalThis.fetch + globalThis.fetch = async () => Response.json({ data: [{ id: "live-model" }] }) + + try { + const { ctx, registered } = context({ baseURL, modelMetadataURL: false }) + await plugin.setup(ctx) + expect(registered[0]?.models).toEqual([]) + await waitUntil(() => registered.at(-1)?.models[0]?.id === "live-model") + } finally { + await restoreFetch(originalFetch) } }) }) @@ -420,6 +761,7 @@ describe("plugin live metadata", () => { }) test("uses live limits when enrichment is disabled or fails", async () => { + await useCacheDir() const originalFetch = globalThis.fetch try { for (const modelMetadataURL of [false, "https://catalog.test/api.json"] as const) { @@ -431,38 +773,48 @@ describe("plugin live metadata", () => { } const { ctx, registered } = context({ baseURL: "http://cliproxy.test:8317", modelMetadataURL }) await plugin.setup(ctx) + await waitUntil(() => registered.at(-1)?.models[0]?.id === "local/qwen3:8b") const model = registered.at(-1)?.models[0] expect(model?.limit).toEqual({ context: 16_384, output: 4_096 }) expect(model?.name).toBe("qwen3:8b (Ollama)") } } finally { - globalThis.fetch = originalFetch + await restoreFetch(originalFetch) } }) test("refresh applies changed and removed live metadata for an unchanged model ID", async () => { + await useCacheDir() const originalFetch = globalThis.fetch const responses = [enriched(8_192), enriched(32_768), enriched(undefined)] - let calls = 0 - globalThis.fetch = async () => Response.json(responses[Math.min(calls++, responses.length - 1)]) + let response = responses[0]! + globalThis.fetch = async () => Response.json(response) let cleanup: void | (() => void | Promise) try { const { ctx, registered } = context({ - baseURL: "http://cliproxy.test:8317", + baseURL: uniqueBaseURL(), modelMetadataURL: false, - refreshIntervalMs: 10, + refreshIntervalMs: 0, }) cleanup = await plugin.setup(ctx) - expect(registered.at(-1)?.models[0]?.limit).toEqual({ context: 8_192, output: 2_048 }) - - const waitFor = async (predicate: () => boolean) => { - for (let i = 0; i < 200 && !predicate(); i++) await Bun.sleep(5) - } - await waitFor(() => registered.at(-1)?.models[0]?.limit.context === 32_768) - expect(registered.at(-1)?.models[0]?.limit).toEqual({ context: 32_768, output: 8_192 }) - await waitFor(() => registered.at(-1)?.models[0]?.limit.context !== 32_768) + await waitUntil(() => + registered.some((entry) => entry.models[0]?.limit?.context === 8_192), + ) + await pluginTesting.settled + expect( + registered.find((entry) => entry.models[0]?.limit?.context === 8_192)?.models[0]?.limit, + ).toEqual({ context: 8_192, output: 2_048 }) + + response = responses[1]! + await pluginTesting.refresh?.() + expect( + registered.find((entry) => entry.models[0]?.limit?.context === 32_768)?.models[0]?.limit, + ).toEqual({ context: 32_768, output: 8_192 }) + + response = responses[2]! + await pluginTesting.refresh?.() const defaults = buildModel({ providerID: "cliproxyapi" as Provider.ID, model: { id: "local/qwen3:8b" }, @@ -472,7 +824,7 @@ describe("plugin live metadata", () => { expect(registered.at(-1)?.models.map((model) => model.id)).toEqual(["local/qwen3:8b"]) } finally { if (typeof cleanup === "function") await cleanup() - globalThis.fetch = originalFetch + await restoreFetch(originalFetch) } }) }) diff --git a/src/index.ts b/src/index.ts index 803422a..b0e0ddf 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,13 @@ import { Model, Plugin, Provider } from "@opencode/plugin" +import { + acquireMetadataLock, + catalogCacheIsFresh, + metadataCacheIsFresh, + readCatalogCache, + readMetadataCache, + writeCatalogCache, + writeMetadataCache, +} from "./cache.js" import { discoverModelMetadata, discoverModels, @@ -15,6 +24,14 @@ const DEFAULT_PROVIDER_NAME = "CLIProxyAPI" const DEFAULT_DISCOVERY_TIMEOUT_MS = 10_000 const DEFAULT_MODEL_METADATA_URL = "https://models.dev/api.json" const DEFAULT_REFRESH_MS = 5 * 60_000 +const METADATA_LOCK_POLL_MS = 50 + +// Lets tests drive a refresh without waiting on the interval timer, and wait +// until the background refresh has released the network. +export const pluginTesting = { + refresh: undefined as undefined | (() => Promise), + settled: Promise.resolve() as Promise, +} const CHAT_PACKAGE = "@opencode/ai/providers/openai-compatible" // OpenCode resolves these package names from inside its own compiled bundle, not @@ -72,6 +89,10 @@ export default Plugin.define({ options.apiKey ?? process.env.CLIPROXY_API_KEY ?? stringOption(existingSettings.apiKey) const timeoutMs = options.discoveryTimeoutMs ?? DEFAULT_DISCOVERY_TIMEOUT_MS const refreshMs = options.refreshIntervalMs ?? DEFAULT_REFRESH_MS + const metadataURL = options.modelMetadataURL ?? DEFAULT_MODEL_METADATA_URL + // The catalog is not secret and does not vary by credential. Keying only on + // the server lets every location share one cache entry. + const catalogCacheKey = baseURL const source = { info: { @@ -87,34 +108,82 @@ export default Plugin.define({ models: [] as Model.Info[], } - const refresh = async (initial: boolean) => { - const [catalog, metadataDiscovery] = await Promise.all([ - discoverModels({ baseURL, apiKey, timeoutMs }), - discoverMetadata({ - url: options.modelMetadataURL ?? DEFAULT_MODEL_METADATA_URL, - timeoutMs, - }), - ]) - - if (metadataDiscovery.error) console.warn(`[cliproxyapi] ${metadataDiscovery.error}`) - - source.models = catalog.map((model) => - buildModel({ - providerID, - model, - metadata: metadataDiscovery.catalog, - }), - ) + let catalog: CatalogModel[] = [] + let metadata: ModelMetadataCatalog = {} + let registered = false + let closed = false + let refreshPromise: Promise | undefined - if (!initial) await ctx.provider.reload() - console.log(`[cliproxyapi] discovered ${source.models.length} models from ${baseURL}`) + const rebuild = async (reload: boolean) => { + if (closed) return + source.models = catalog.map((model) => buildModel({ providerID, model, metadata })) + if (reload && registered && !closed) await ctx.provider.reload() } - await refresh(true) + const refreshCatalog = async (reload: boolean, fresh = false) => { + let discovered: CatalogModel[] + try { + discovered = await loadCatalog({ baseURL, apiKey, timeoutMs, fresh }) + } catch (error) { + if (closed) return + // Keep the previous entry unchanged, including its fetchedAt. Rewriting + // it would make a failed server look freshly discovered. + const recovered = await readCatalogCache(catalogCacheKey).catch(() => undefined) + if (!recovered?.models.length) throw error + catalog = recovered.models + await rebuild(reload) + const message = error instanceof Error ? error.message : String(error) + console.warn(`[cliproxyapi] catalog refresh failed, using cached models: ${message}`) + return + } + + catalog = discovered + await writeCatalogCache(catalogCacheKey, catalog).catch((error) => { + const message = error instanceof Error ? error.message : String(error) + console.warn(`[cliproxyapi] could not update the catalog cache: ${message}`) + }) + if (closed) return + await rebuild(reload) + console.log(`[cliproxyapi] discovered ${catalog.length} models from ${baseURL}`) + } + const refreshMetadata = async () => { + if (metadataURL === false || closed || catalog.length === 0) return + const result = await loadModelMetadata({ + url: metadataURL, + timeoutMs, + isClosed: () => closed, + }) + if (closed || catalog.length === 0) return + // A failed download must not wipe metadata already in use. An empty + // successful catalog is authoritative and clears stale enrichment. + if (result.stale) { + if (Object.keys(metadata).length > 0 || Object.keys(result.catalog).length === 0) return + } + metadata = result.catalog + await rebuild(true) + } + + // A previous process may already have the catalogs. Use them immediately so + // models are available even when this process loses the download race, then + // refresh in the background. + const [cachedCatalog, cachedMetadata] = await Promise.all([ + readCatalogCache(catalogCacheKey).catch(() => undefined), + metadataURL === false + ? Promise.resolve(undefined) + : readMetadataCache(metadataURL).catch(() => undefined), + ]) + if (cachedCatalog?.models.length) catalog = cachedCatalog.models + if (cachedMetadata) metadata = cachedMetadata.catalog + if (catalog.length > 0) await rebuild(false) + + // Register the provider immediately so activation never depends on a + // network request. The model list is filled in by the background refresh + // below, and the provider is reloaded once it is available. await ctx.provider.transform((editor) => { editor.add({ info: source.info, models: source.models }) }) + registered = true await ctx.session.hook( "model.request", @@ -125,17 +194,48 @@ export default Plugin.define({ { providerID }, ) - if (refreshMs <= 0) return - - const timer = setInterval(() => { - void refresh(false).catch((error) => { + const refresh = () => { + if (closed) return Promise.resolve() + if (refreshPromise) return refreshPromise + + refreshPromise = (async () => { + try { + // Fetch the CLIProxyAPI catalog first. Metadata is large and must not + // compete with model discovery when several locations load at once. + await refreshCatalog(true, true) + if (closed) return + await refreshMetadata() + } finally { + refreshPromise = undefined + } + })() + pluginTesting.settled = refreshPromise.catch((error) => { console.warn( `[cliproxyapi] catalog refresh failed: ${error instanceof Error ? error.message : String(error)}`, ) }) + + return refreshPromise + } + pluginTesting.refresh = refresh + + void refresh() + + const cleanup = () => { + closed = true + if (pluginTesting.refresh === refresh) pluginTesting.refresh = undefined + } + + if (refreshMs <= 0) return cleanup + + const timer = setInterval(() => { + void refresh() }, refreshMs) timer.unref?.() - return () => clearInterval(timer) + return () => { + cleanup() + clearInterval(timer) + } }, }) @@ -272,21 +372,139 @@ function resolveMetadata( return { npm: owner?.npm } } -function discoverMetadata(input: { +// OpenCode may activate the same plugin once per location, sometimes in a +// separate process. Share in-flight requests within this process, and use the +// disk cache and metadata lock to coordinate across processes. +const catalogRequests = new Map>() + +function loadCatalog(input: { + baseURL: string + apiKey?: string + timeoutMs: number + fresh?: boolean +}) { + // Share a download within one process only when it uses the same credential. + // The disk cache is keyed by server alone; see catalogCacheKey. + const requestKey = `${input.baseURL}\0${input.apiKey ?? ""}` + const existing = catalogRequests.get(requestKey) + if (existing) return existing + + const pending = (async () => { + if (!input.fresh) { + const cached = await readCatalogCache(input.baseURL).catch(() => undefined) + if (cached && catalogCacheIsFresh(cached.fetchedAt) && cached.models.length > 0) { + return cached.models + } + } + return discoverModels(input) + })().finally(() => { + catalogRequests.delete(requestKey) + }) + + catalogRequests.set(requestKey, pending) + return pending +} + +type ModelMetadataDiscovery = { + catalog: ModelMetadataCatalog + /** True when this is a fallback, not a completed fetch. */ + stale?: boolean + error?: string +} + +const metadataRequests = new Map>() + +function loadModelMetadata(input: { url: string | false timeoutMs: number -}): Promise<{ catalog: ModelMetadataCatalog; error?: string }> { + isClosed?: () => boolean +}): Promise { if (input.url === false) return Promise.resolve({ catalog: {} }) - return discoverModelMetadata({ - url: input.url, - timeoutMs: input.timeoutMs, - }) - .then((catalog) => ({ catalog })) - .catch((error) => ({ - catalog: {}, - error: error instanceof Error ? error.message : String(error), - })) + const url = input.url + const existing = metadataRequests.get(url) + if (existing) return existing + + const pending = discoverSharedMetadata(url, input.timeoutMs, input.isClosed ?? (() => false)).finally( + () => { + metadataRequests.delete(url) + }, + ) + + metadataRequests.set(url, pending) + return pending +} + +async function discoverSharedMetadata( + url: string, + timeoutMs: number, + isClosed: () => boolean, +): Promise { + const cached = await readMetadataCache(url).catch(() => undefined) + if (cached && metadataCacheIsFresh(cached.fetchedAt)) return { catalog: cached.catalog } + + let release: (() => Promise) | undefined + try { + release = await acquireMetadataLock(url) + } catch (error) { + // A cache directory that cannot be created should not disable enrichment. + const message = error instanceof Error ? error.message : String(error) + console.warn(`[cliproxyapi] metadata lock unavailable, downloading anyway: ${message}`) + return fetchMetadata(url, timeoutMs, cached?.catalog) + } + + if (!release) { + // Another process is already downloading. Wait for its cache instead of + // starting a second download or skipping enrichment until the next refresh. + const waited = await waitForSharedMetadata(url, timeoutMs, isClosed) + if (waited && metadataCacheIsFresh(waited.fetchedAt)) return { catalog: waited.catalog } + return { catalog: waited?.catalog ?? cached?.catalog ?? {}, stale: true } + } + + try { + return await fetchMetadata(url, timeoutMs, cached?.catalog) + } finally { + await release() + } +} + +async function fetchMetadata( + url: string, + timeoutMs: number, + fallback: ModelMetadataCatalog | undefined, +): Promise { + try { + const catalog = await discoverModelMetadata({ url, timeoutMs }) + // A completed response is authoritative, including an empty catalog. + await writeMetadataCache(url, catalog).catch((error) => { + const message = error instanceof Error ? error.message : String(error) + console.warn(`[cliproxyapi] could not update the metadata cache: ${message}`) + }) + return { catalog } + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + console.warn(`[cliproxyapi] ${message}`) + return { catalog: fallback ?? {}, stale: true, error: message } + } +} + +async function waitForSharedMetadata( + url: string, + timeoutMs: number, + isClosed: () => boolean, +) { + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + if (isClosed()) return + const cached = await readMetadataCache(url).catch(() => undefined) + if (cached && metadataCacheIsFresh(cached.fetchedAt)) return cached + await sleep(METADATA_LOCK_POLL_MS) + } + return readMetadataCache(url).catch(() => undefined) +} + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)) } function readOptions(input: Record): ConnectorOptions {