Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 71 Packages Projects Releases Wiki Activity

refactor(core): move the models.dev catalog cache from disk to KV (#41649)

Browse Source
This commit is contained in:
Kit Langton
2026-08-10 22:40:41 -04:00
committed by GitHub
parent 5d8c48718b
commit 518af92c5b
2 changed files with 128 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/models-dev.ts
+63 -57
View File
@@ -1,11 +1,8 @@
import path from "path"
import { Context, Duration, Effect, Layer, Option, Schedule, Schema } from "effect"
import { Context, Duration, Effect, Layer, Option, Schedule, Schema, Semaphore } from "effect"
import { HttpClient, HttpClientRequest } from "effect/unstable/http"
import { ModelsDev } from "@opencode-ai/schema/models-dev"
import { Money } from "@opencode-ai/schema/money"
import { App } from "./app"
import { Global } from "@opencode-ai/util/global"
import { Flock } from "@opencode-ai/util/flock"
import { Hash } from "@opencode-ai/util/hash"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Bus } from "./bus"
@@ -13,6 +10,7 @@ import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { httpClient } from "@opencode-ai/util/effect/app-node-platform"
import { Model } from "./model"
import { Provider } from "./provider"
import { KV } from "./kv"
export const CatalogModelStatus = Schema.Literals(["alpha", "beta", "deprecated"])
export type CatalogModelStatus = typeof CatalogModelStatus.Type
@@ -537,6 +535,18 @@ export type Options = typeof Options.Type
export class Service extends Context.Service<Service, Interface>()("@opencode/ModelsDev") {}
const CatalogJson = Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown))
const Cache = Schema.Struct({
updatedAt: Schema.Number,
body: CatalogJson,
})
const defaultSource = "https://models.opencode.ai"
function cacheKey(source: string) {
if (source === defaultSource) return "models-dev:catalog"
return `models-dev:catalog:${Hash.fast(source)}`
}
export const layer = (options?: Options) =>
Layer.effect(
Service,
@@ -544,7 +554,7 @@ export const layer = (options?: Options) =>
const fs = yield* FSUtil.Service
const bus = yield* Bus.Service
const app = yield* App.Metadata
const global = yield* Global.Service
const kv = yield* KV.Service
const http = HttpClient.filterStatusOk(
(yield* HttpClient.HttpClient).pipe(
HttpClient.retryTransient({
@@ -555,21 +565,28 @@ export const layer = (options?: Options) =>
),
)
const source = options?.url || "https://models.opencode.ai"
const source = options?.url || defaultSource
const fetch = options?.fetch ?? true
const userAgent = App.useragent(app)
const filepath = path.join(
global.cache,
source === "https://models.opencode.ai" ? "models.json" : `models-${Hash.fast(source)}.json`,
)
const key = cacheKey(source)
const ttl = Duration.minutes(5)
const lockKey = `models-dev:${filepath}`
const lock = Semaphore.makeUnsafe(1)
const loadFromCache = Effect.fnUntraced(function* () {
const value = yield* kv.get(key)
const cached = Schema.decodeUnknownOption(Cache)(value)
if (Option.isSome(cached))
return {
catalog: cached.value.body as Record<string, SourceProvider>,
updatedAt: cached.value.updatedAt,
}
if (value !== undefined) yield* kv.remove(key)
})
const fresh = Effect.fnUntraced(function* () {
const stat = yield* fs.stat(filepath).pipe(Effect.catch(() => Effect.succeed(undefined)))
if (!stat) return false
const mtime = Option.getOrElse(stat.mtime, () => new Date(0)).getTime()
return Date.now() - mtime < Duration.toMillis(ttl)
const cached = yield* loadFromCache()
if (!cached) return false
return Date.now() - cached.updatedAt < Duration.toMillis(ttl)
})
const fetchApi = Effect.fn("ModelsDev.fetchApi")(function* () {
@@ -581,15 +598,12 @@ export const layer = (options?: Options) =>
)
})
const loadFromDisk = fs.readJson(options?.file ?? filepath).pipe(
Effect.map((input) => input as Record<string, SourceProvider>),
Effect.catch((error) => {
if (options?.file === undefined && error._tag === "FileSystemError" && error.method === "readJson") {
return fs.remove(filepath, { force: true }).pipe(Effect.ignore, Effect.as(undefined))
}
return Effect.succeed(undefined)
}),
)
const loadFromFile = options?.file
? fs.readJson(options.file).pipe(
Effect.map((input) => input as Record<string, SourceProvider>),
Effect.catch(() => Effect.succeed(undefined)),
)
: Effect.succeed(undefined)
const loadSnapshot = Effect.sync(() =>
typeof OPENCODE_MODELS_DEV === "undefined" ? undefined : OPENCODE_MODELS_DEV,
@@ -597,33 +611,27 @@ export const layer = (options?: Options) =>
const fetchAndWrite = Effect.fn("ModelsDev.fetchAndWrite")(function* () {
const text = yield* fetchApi()
const tempfile = `${filepath}.${process.pid}.${Date.now()}.tmp`
yield* fs.writeWithDirs(tempfile, text).pipe(
Effect.andThen(fs.rename(tempfile, filepath)),
Effect.catch((error) =>
Effect.gen(function* () {
yield* fs.remove(tempfile, { force: true }).pipe(Effect.ignore)
return yield* Effect.fail(error)
}),
),
)
return text
const catalog = (yield* Schema.decodeUnknownEffect(CatalogJson)(text)) as Record<string, SourceProvider>
yield* kv.set(key, { updatedAt: Date.now(), body: text })
return catalog
})
const populate = Effect.gen(function* () {
const fromDisk = yield* loadFromDisk
if (fromDisk) return normalize(fromDisk)
const fromFile = yield* loadFromFile
if (fromFile) return normalize(fromFile)
const cached = options?.file ? undefined : yield* loadFromCache()
if (cached) return normalize(cached.catalog)
const bundled = yield* loadSnapshot
if (bundled) return normalize(bundled)
if (!fetch) return []
// Flock is cross-process: concurrent opencode CLIs can race on this cache file.
const text = yield* Effect.scoped(
const catalog = yield* lock.withPermit(
Effect.gen(function* () {
yield* Flock.effect(lockKey)
const stored = options?.file ? undefined : yield* loadFromCache()
if (stored) return stored.catalog
return yield* fetchAndWrite()
}),
)
return normalize(JSON.parse(text) as Record<string, SourceProvider>)
return normalize(catalog)
}).pipe(Effect.withSpan("ModelsDev.populate"), Effect.orDie)
const [cachedGet, invalidate] = yield* Effect.cachedInvalidateWithTTL(populate, Duration.infinity)
@@ -631,21 +639,19 @@ export const layer = (options?: Options) =>
const get = (): Effect.Effect<readonly Snapshot[]> => cachedGet
const refresh = Effect.fn("ModelsDev.refresh")(function* (force = false) {
if (!force && (yield* fresh())) return
yield* Effect.scoped(
Effect.gen(function* () {
yield* Flock.effect(lockKey)
// Re-check under the lock: another process may have refreshed between
// our outer check and lock acquisition.
if (!force && (yield* fresh())) return
yield* fetchAndWrite()
yield* invalidate
yield* bus.publish(ModelsDev.Event.Refreshed, {})
}),
).pipe(
Effect.tapCause((cause) => Effect.logError("Failed to fetch models.dev", { cause: cause })),
Effect.ignore,
)
yield* lock
.withPermit(
Effect.gen(function* () {
if (!force && (yield* fresh())) return
yield* fetchAndWrite()
yield* invalidate
yield* bus.publish(ModelsDev.Event.Refreshed, {})
}),
)
.pipe(
Effect.tapCause((cause) => Effect.logError("Failed to fetch models.dev", { cause: cause })),
Effect.ignore,
)
})
if (fetch && !process.argv.includes("--get-yargs-completions")) {
@@ -661,7 +667,7 @@ export function configured(options?: Options) {
return makeGlobalNode({
service: Service,
layer: layer(options),
deps: [FSUtil.node, Bus.node, App.node, Global.node, httpClient],
deps: [FSUtil.node, Bus.node, App.node, KV.node, httpClient],
})
}
packages/core/test/models.test.ts
+65 -55
View File
@@ -1,19 +1,17 @@
import { describe, expect, beforeEach, afterAll, test } from "bun:test"
import { describe, expect, test } from "bun:test"
import { Money } from "@opencode-ai/schema/money"
import { Effect, Layer, Ref } from "effect"
import { HttpClient, HttpClientResponse } from "effect/unstable/http"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LayerNodePlatform } from "@opencode-ai/util/effect/app-node-platform"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Global } from "@opencode-ai/util/global"
import { KV } from "@opencode-ai/core/kv"
import { Model } from "@opencode-ai/core/model"
import { ModelsDev } from "@opencode-ai/core/models-dev"
import { Provider } from "@opencode-ai/core/provider"
import { it } from "./lib/effect"
import { readFile, rm, writeFile, utimes, mkdir } from "fs/promises"
import path from "path"
const cacheFile = path.join(Global.Path.cache, "models.json")
const cacheKey = "models-dev:catalog"
test("normalizes permissive interleaved values to compatibility", () => {
expect(Model.compatibility("reasoning_text")).toEqual({ reasoningField: "reasoning_text" })
@@ -164,7 +162,18 @@ const makeMockClient = (state: Ref.Ref<MockState>) =>
}),
)
const buildLayer = (state: Ref.Ref<MockState>, options: ModelsDev.Options = { fetch: false }) =>
interface MockCache {
readonly values: Map<string, KV.Value>
}
const makeMockKV = (cache: MockCache) =>
Layer.mock(KV.Service, {
get: (key) => Effect.sync(() => cache.values.get(key)),
set: (key, value) => Effect.sync(() => cache.values.set(key, value)).pipe(Effect.asVoid),
remove: (key) => Effect.sync(() => cache.values.delete(key)).pipe(Effect.asVoid),
})
const buildLayer = (state: Ref.Ref<MockState>, cache: MockCache, options: ModelsDev.Options = { fetch: false }) =>
// Layer.fresh is required because the ModelsDev implementation is a module-level Layer constant,
// and Effect.provide uses a process-global MemoMap by default — without fresh,
// every test would reuse the cachedInvalidateWithTTL state from the first run.
@@ -172,31 +181,20 @@ const buildLayer = (state: Ref.Ref<MockState>, options: ModelsDev.Options = { fe
AppNodeBuilder.build(ModelsDev.node, [
[ModelsDev.node, ModelsDev.configured(options)],
[LayerNodePlatform.httpClient, Layer.succeed(HttpClient.HttpClient, makeMockClient(state))],
[KV.node, makeMockKV(cache)],
]),
)
const writeCacheText = (text: string, mtimeMs?: number) =>
Effect.promise(async () => {
await mkdir(Global.Path.cache, { recursive: true })
await writeFile(cacheFile, text)
if (mtimeMs !== undefined) {
const t = mtimeMs / 1000
await utimes(cacheFile, t, t)
}
})
const makeCache = (): MockCache => ({ values: new Map() })
const writeCache = (data: object, mtimeMs?: number) => writeCacheText(JSON.stringify(data), mtimeMs)
const writeCacheText = (cache: MockCache, text: string, updatedAt = Date.now()) =>
cache.values.set(cacheKey, { updatedAt, body: text })
const provided = <A, E>(state: Ref.Ref<MockState>, eff: Effect.Effect<A, E, ModelsDev.Service>) =>
eff.pipe(Effect.provide(buildLayer(state)))
const writeCache = (cache: MockCache, data: object, updatedAt?: number) =>
writeCacheText(cache, JSON.stringify(data), updatedAt)
beforeEach(async () => {
await rm(cacheFile, { force: true })
})
afterAll(async () => {
await rm(cacheFile, { force: true })
})
const provided = <A, E>(state: Ref.Ref<MockState>, cache: MockCache, eff: Effect.Effect<A, E, ModelsDev.Service>) =>
eff.pipe(Effect.provide(buildLayer(state, cache)))
const initialState: MockState = {
body: JSON.stringify(fixture),
@@ -205,12 +203,14 @@ const initialState: MockState = {
}
describe("ModelsDev Service", () => {
it.live("get() returns normalized snapshots from disk when cache file exists", () =>
it.live("get() returns normalized snapshots from KV when a cache entry exists", () =>
Effect.gen(function* () {
yield* writeCache(fixture)
const cache = makeCache()
writeCache(cache, fixture)
const state = yield* Ref.make(initialState)
const result = yield* provided(
state,
cache,
ModelsDev.Service.use((s) => s.get()),
)
expect(result).toEqual(fixtureSnapshot)
@@ -219,11 +219,13 @@ describe("ModelsDev Service", () => {
}),
)
it.live("get() returns empty catalog when disk empty, fetch disabled, and no bundled snapshot is injected", () =>
it.live("get() returns empty catalog when KV is empty, fetch disabled, and no bundled snapshot is injected", () =>
Effect.gen(function* () {
const cache = makeCache()
const state = yield* Ref.make(initialState)
const result = yield* provided(
state,
cache,
ModelsDev.Service.use((s) => s.get()),
)
expect(result).toEqual([])
@@ -232,14 +234,15 @@ describe("ModelsDev Service", () => {
}),
)
it.live("get() recovers from a corrupted cache file by fetching a fresh catalog", () =>
it.live("get() recovers from a corrupted KV entry by fetching a fresh catalog", () =>
Effect.gen(function* () {
yield* writeCacheText("{")
const cache = makeCache()
writeCacheText(cache, "{")
const state = yield* Ref.make({ ...initialState, body: JSON.stringify(fixture2) })
const context = yield* Layer.build(buildLayer(state, { fetch: true }))
const context = yield* Layer.build(buildLayer(state, cache, { fetch: true }))
const result = yield* ModelsDev.Service.use((s) => s.get()).pipe(Effect.provide(context))
expect(result).toEqual(fixture2Snapshot)
expect(yield* Effect.promise(() => readFile(cacheFile, "utf8"))).toBe(JSON.stringify(fixture2))
expect(cache.values.get(cacheKey)).toMatchObject({ body: JSON.stringify(fixture2) })
const final = yield* Ref.get(state)
expect(final.calls.length).toBe(1)
}),
@@ -247,9 +250,10 @@ describe("ModelsDev Service", () => {
it.live("uses the default models URL when the configured URL is empty", () =>
Effect.gen(function* () {
const cache = makeCache()
const state = yield* Ref.make(initialState)
yield* ModelsDev.Service.use((service) => service.get()).pipe(
Effect.provide(buildLayer(state, { url: "", fetch: true })),
Effect.provide(buildLayer(state, cache, { url: "", fetch: true })),
)
expect((yield* Ref.get(state)).calls[0]?.url).toBe("https://models.opencode.ai/api.json")
}),
@@ -257,32 +261,31 @@ describe("ModelsDev Service", () => {
it.live("get() is single-flight under concurrent calls", () =>
Effect.gen(function* () {
yield* writeCache(fixture)
const cache = makeCache()
const state = yield* Ref.make(initialState)
const results = yield* provided(
state,
Effect.gen(function* () {
const svc = yield* ModelsDev.Service
return yield* Effect.all([svc.get(), svc.get(), svc.get(), svc.get(), svc.get()], {
concurrency: "unbounded",
})
}),
)
const results = yield* Effect.gen(function* () {
const svc = yield* ModelsDev.Service
return yield* Effect.all([svc.get(), svc.get(), svc.get(), svc.get(), svc.get()], {
concurrency: "unbounded",
})
}).pipe(Effect.provide(buildLayer(state, cache, { fetch: true })))
for (const result of results) expect(result).toEqual(fixtureSnapshot)
expect((yield* Ref.get(state)).calls.length).toBe(1)
}),
)
it.live("get() caches across calls (later disk writes are ignored until invalidate)", () =>
it.live("get() caches across calls (later KV writes are ignored until invalidate)", () =>
Effect.gen(function* () {
yield* writeCache(fixture)
const cache = makeCache()
writeCache(cache, fixture)
const state = yield* Ref.make(initialState)
const first = yield* provided(
state,
cache,
Effect.gen(function* () {
const svc = yield* ModelsDev.Service
const a = yield* svc.get()
// mutate disk between calls — cache should mask the change
yield* writeCache(fixture2)
writeCache(cache, fixture2)
const b = yield* svc.get()
return { a, b }
}),
@@ -294,10 +297,12 @@ describe("ModelsDev Service", () => {
it.live("refresh(true) fetches via HttpClient and updates the cache", () =>
Effect.gen(function* () {
yield* writeCache(fixture)
const cache = makeCache()
writeCache(cache, fixture)
const state = yield* Ref.make({ ...initialState, body: JSON.stringify(fixture2) })
const result = yield* provided(
state,
cache,
Effect.gen(function* () {
const svc = yield* ModelsDev.Service
const before = yield* svc.get()
@@ -308,6 +313,7 @@ describe("ModelsDev Service", () => {
)
expect(result.before).toEqual(fixtureSnapshot)
expect(result.after).toEqual(fixture2Snapshot)
expect(cache.values.get(cacheKey)).toMatchObject({ body: JSON.stringify(fixture2) })
const final = yield* Ref.get(state)
expect(final.calls.length).toBe(1)
expect(final.calls[0].url).toContain("/api.json")
@@ -315,13 +321,14 @@ describe("ModelsDev Service", () => {
}),
)
it.live("refresh(false) skips fetch when on-disk file is fresh", () =>
it.live("refresh(false) skips fetch when the KV entry is fresh", () =>
Effect.gen(function* () {
// Fresh: mtime within the 5-minute TTL.
yield* writeCache(fixture, Date.now() - 1000)
const cache = makeCache()
writeCache(cache, fixture, Date.now() - 1000)
const state = yield* Ref.make({ ...initialState, body: JSON.stringify(fixture2) })
yield* provided(
state,
cache,
ModelsDev.Service.use((s) => s.refresh(false)),
)
const final = yield* Ref.get(state)
@@ -329,13 +336,14 @@ describe("ModelsDev Service", () => {
}),
)
it.live("refresh(false) fetches when on-disk file is stale", () =>
it.live("refresh(false) fetches when the KV entry is stale", () =>
Effect.gen(function* () {
// Stale: mtime 10 minutes ago, beyond the 5-minute TTL.
yield* writeCache(fixture, Date.now() - 10 * 60 * 1000)
const cache = makeCache()
writeCache(cache, fixture, Date.now() - 10 * 60 * 1000)
const state = yield* Ref.make({ ...initialState, body: JSON.stringify(fixture2) })
const after = yield* provided(
state,
cache,
Effect.gen(function* () {
const svc = yield* ModelsDev.Service
yield* svc.refresh(false)
@@ -350,10 +358,12 @@ describe("ModelsDev Service", () => {
it.live("refresh swallows HTTP errors and leaves cache intact", () =>
Effect.gen(function* () {
yield* writeCache(fixture)
const cache = makeCache()
writeCache(cache, fixture)
const state = yield* Ref.make({ ...initialState, status: 500, body: "boom" })
const result = yield* provided(
state,
cache,
Effect.gen(function* () {
const svc = yield* ModelsDev.Service
yield* svc.refresh(true)