diff --git a/packages/core/src/catalog.ts b/packages/core/src/catalog.ts index ab34db8ea4..59db758448 100644 --- a/packages/core/src/catalog.ts +++ b/packages/core/src/catalog.ts @@ -1,7 +1,8 @@ export * as Catalog from "./catalog" import { makeLocationNode } from "./effect/app-node" -import { Array, Context, Effect, Layer, Option, Order, pipe } from "effect" +import { Array, Context, Effect, Exit, Layer, Option, Order, pipe, Schema, Stream, SubscriptionRef } from "effect" +import type { Scope } from "effect" import { Catalog } from "@opencode-ai/schema/catalog" import { ModelV2 } from "./model" import { ProviderV2 } from "./provider" @@ -18,11 +19,31 @@ export type DefaultModel = { providerID: ProviderV2.ID; modelID: ModelV2.ID } export const Event = Catalog.Event +export class IncompleteError extends Schema.TaggedErrorClass()("Catalog.IncompleteError", {}) { + override get message() { + return "Initial catalog discovery did not complete successfully" + } +} + type Data = { providers: Map defaultModel?: DefaultModel } +type InitialState = { + readonly discovery: "idle" | "pending" | "succeeded" | "failed" + readonly pending: number + readonly succeeded: number + readonly failed: number + readonly version: number +} + +type Selection = + | { readonly _tag: "pending" } + | { readonly _tag: "selected"; readonly model: ModelV2.Info } + | { readonly _tag: "absent" } + | { readonly _tag: "incomplete" } + export type Draft = { provider: { list: () => readonly ProviderRecord[] @@ -42,6 +63,10 @@ export type Draft = { } export interface Interface extends State.Transformable { + readonly initial: { + readonly discover: (work: Effect.Effect) => Effect.Effect + readonly source: (work: Effect.Effect) => Effect.Effect + } readonly provider: { readonly get: (providerID: ProviderV2.ID) => Effect.Effect readonly all: () => Effect.Effect @@ -51,6 +76,10 @@ export interface Interface extends State.Transformable { readonly get: (providerID: ProviderV2.ID, modelID: ModelV2.ID) => Effect.Effect readonly all: () => Effect.Effect readonly available: () => Effect.Effect + readonly select: ( + ref: ModelV2.Ref | undefined, + supported: (model: ModelV2.Info) => boolean, + ) => Effect.Effect readonly default: () => Effect.Effect readonly small: (providerID: ProviderV2.ID) => Effect.Effect } @@ -63,6 +92,67 @@ const layer = Layer.effect( Effect.gen(function* () { const events = yield* EventV2.Service const integrations = yield* Integration.Service + const initial = yield* SubscriptionRef.make({ + discovery: "idle", + pending: 0, + succeeded: 0, + failed: 0, + version: 0, + }) + + const updateInitial = (update: (state: InitialState) => Omit) => + SubscriptionRef.update(initial, (state) => ({ ...update(state), version: state.version + 1 })) + + const discover: Interface["initial"]["discover"] = Effect.fn("CatalogV2.initial.discover")(function* (work) { + yield* Effect.acquireUseRelease( + SubscriptionRef.modify(initial, (state) => [ + state.discovery === "idle", + state.discovery === "idle" + ? { ...state, discovery: "pending" as const, version: state.version + 1 } + : state, + ]), + () => work, + (tracked, exit) => + tracked + ? updateInitial((state) => ({ + discovery: Exit.isSuccess(exit) ? "succeeded" : "failed", + pending: state.pending, + succeeded: state.succeeded, + failed: state.failed, + })) + : Effect.void, + ).pipe( + Effect.tapCause((cause) => Effect.logError("initial catalog discovery failed", { cause })), + Effect.ignoreCause, + ) + }) + + const source: Interface["initial"]["source"] = Effect.fn("CatalogV2.initial.source")(function* (work) { + yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const tracked = yield* SubscriptionRef.modify(initial, (state) => [ + state.discovery === "pending", + state.discovery === "pending" + ? { ...state, pending: state.pending + 1, version: state.version + 1 } + : state, + ]) + yield* restore(work).pipe( + tracked + ? Effect.onExit((exit) => + updateInitial((state) => ({ + discovery: state.discovery, + pending: state.pending - 1, + succeeded: state.succeeded + (Exit.isSuccess(exit) ? 1 : 0), + failed: state.failed + (Exit.isFailure(exit) ? 1 : 0), + })), + ) + : (effect) => effect, + Effect.ignoreCause, + Effect.forkScoped({ startImmediately: true }), + ) + }), + ) + }) const available = (provider: ProviderV2.Info, integration: Integration.Info | undefined) => { if (provider.disabled) return false @@ -157,11 +247,17 @@ const layer = Layer.effect( finalize: Effect.fn("CatalogV2.finalize")(function* (catalog) { yield* events.publish(Event.Updated, {}) }), + committed: () => updateInitial((state) => state), }) const result: Interface = { transform: state.transform, reload: state.reload, + initial: { + discover, + source, + }, + provider: { get: Effect.fn("CatalogV2.provider.get")(function* (providerID) { return state.get().providers.get(providerID)?.provider @@ -202,6 +298,45 @@ const layer = Layer.effect( return (yield* result.model.all()).filter((model) => providers.has(model.providerID) && model.enabled) }), + select: Effect.fn("CatalogV2.model.select")(function* (ref, supported) { + const inspect: () => Effect.Effect = Effect.fnUntraced(function* () { + const status = yield* SubscriptionRef.get(initial) + + if (ref) { + const selected = (yield* result.model.available()).find( + (model) => model.providerID === ref.providerID && model.id === ref.id, + ) + if (selected && status.discovery !== "idle" && status.discovery !== "pending") + return { _tag: "selected", model: selected } + } + + if (status.discovery === "idle" || status.discovery === "pending") return { _tag: "pending" } + if (status.pending > 0) return { _tag: "pending" } + if (status.discovery === "failed" || status.failed > 0) return { _tag: "incomplete" } + if (ref) return { _tag: "absent" } + + const defaultModel = yield* result.model.default() + if (defaultModel && supported(defaultModel)) return { _tag: "selected", model: defaultModel } + const selected = (yield* result.model.available()).find(supported) + return selected ? { _tag: "selected", model: selected } : { _tag: "absent" } + }) + const resolve = (selection: Selection): Effect.Effect => { + if (selection._tag === "selected") return Effect.succeed(selection.model) + if (selection._tag === "incomplete") return Effect.fail(new IncompleteError()) + return Effect.succeed(undefined) + } + + const observed = yield* SubscriptionRef.changes(initial).pipe( + Stream.mapEffect(() => inspect()), + Stream.filter((selection) => selection._tag !== "pending"), + Stream.runHead, + ) + return yield* Option.match(observed, { + onNone: () => Effect.die(new Error("Catalog readiness stream ended before initial settlement")), + onSome: resolve, + }) + }), + default: Effect.fn("CatalogV2.model.default")(function* () { const defaultModel = state.get().defaultModel if (defaultModel) { diff --git a/packages/core/src/plugin/internal.ts b/packages/core/src/plugin/internal.ts index 939bb4e731..b40b1220b8 100644 --- a/packages/core/src/plugin/internal.ts +++ b/packages/core/src/plugin/internal.ts @@ -104,8 +104,9 @@ const layer = Layer.effectDiscard( Effect.gen(function* () { const plugin = yield* PluginV2.Service const sdkPlugins = yield* SdkPlugins.Service + const catalog = yield* Catalog.Service const services = Context.mergeAll( - Context.make(Catalog.Service, yield* Catalog.Service), + Context.make(Catalog.Service, catalog), Context.make(CommandV2.Service, yield* CommandV2.Service), Context.make(Integration.Service, yield* Integration.Service), Context.make(AgentV2.Service, yield* AgentV2.Service), @@ -139,37 +140,42 @@ const layer = Layer.effectDiscard( input.effect(context).pipe(Effect.provide(services)), ) - yield* State.batch( - Effect.gen(function* () { - yield* add(ConfigReferencePlugin.Plugin) - yield* add(AgentPlugin.Plugin) - yield* add(CommandPlugin.Plugin) - yield* add(SkillPlugin.Plugin) - yield* add(ModelsDevPlugin) - yield* add(ConfigExternalPlugin.Plugin) - yield* add(ApplyPatchTool.Plugin) - yield* add(EditTool.Plugin) - yield* add(GlobTool.Plugin) - yield* add(GrepTool.Plugin) - yield* add(QuestionTool.Plugin) - yield* add(ReadTool.Plugin) - yield* add(ShellTool.Plugin) - yield* add(SkillTool.Plugin) - yield* add(SubagentTool.Plugin) - yield* add(TodoWriteTool.Plugin) - yield* add(WebFetchTool.Plugin) - yield* add(WebSearchTool.Plugin) - yield* add(WriteTool.Plugin) - yield* add(ConfigAgentPlugin.Plugin) - yield* add(ConfigCommandPlugin.Plugin) - yield* add(ConfigSkillPlugin.Plugin) - for (const item of ProviderPlugins) yield* add(item) - yield* add(ConfigProviderPlugin.Plugin) - yield* add(VariantPlugin.Plugin) - // Embedder-contributed plugins are added last so they layer over config. - for (const plugin of sdkPlugins.all()) yield* add(plugin) - }), - ).pipe(Effect.withSpan("PluginInternal.boot"), Effect.forkScoped({ startImmediately: true })) + yield* catalog.initial.discover( + State.batch( + Effect.gen(function* () { + yield* add(ConfigReferencePlugin.Plugin) + yield* add(AgentPlugin.Plugin) + yield* add(CommandPlugin.Plugin) + yield* add(SkillPlugin.Plugin) + yield* add(ModelsDevPlugin) + yield* add(ConfigExternalPlugin.Plugin) + yield* add(ApplyPatchTool.Plugin) + yield* add(EditTool.Plugin) + yield* add(GlobTool.Plugin) + yield* add(GrepTool.Plugin) + yield* add(QuestionTool.Plugin) + yield* add(ReadTool.Plugin) + yield* add(ShellTool.Plugin) + yield* add(SkillTool.Plugin) + yield* add(SubagentTool.Plugin) + yield* add(TodoWriteTool.Plugin) + yield* add(WebFetchTool.Plugin) + yield* add(WebSearchTool.Plugin) + yield* add(WriteTool.Plugin) + yield* add(ConfigAgentPlugin.Plugin) + yield* add(ConfigCommandPlugin.Plugin) + yield* add(ConfigSkillPlugin.Plugin) + for (const item of ProviderPlugins) yield* add(item) + yield* add(ConfigProviderPlugin.Plugin) + yield* add(VariantPlugin.Plugin) + // Embedder-contributed plugins are added last so they layer over config. + for (const plugin of sdkPlugins.all()) yield* add(plugin) + }), + ), + ).pipe( + Effect.withSpan("PluginInternal.boot"), + Effect.forkScoped({ startImmediately: true }), + ) }), ) diff --git a/packages/core/src/plugin/models-dev.ts b/packages/core/src/plugin/models-dev.ts index 27c176ed67..a05739738c 100644 --- a/packages/core/src/plugin/models-dev.ts +++ b/packages/core/src/plugin/models-dev.ts @@ -1,6 +1,7 @@ import { define } from "./internal" import type { ModelV2Info } from "@opencode-ai/sdk/v2/types" import { Effect, Stream } from "effect" +import { Catalog } from "../catalog" import { EventV2 } from "../event" import { ModelV2 } from "../model" import { ModelsDev } from "../models-dev" @@ -199,9 +200,10 @@ function applyModel( export const ModelsDevPlugin = define({ id: "models-dev", effect: Effect.fn(function* (ctx) { + const catalog = yield* Catalog.Service const modelsDev = yield* ModelsDev.Service const events = yield* EventV2.Service - const loaded = { data: yield* modelsDev.get() } + const loaded: { data: Record } = { data: {} } yield* ctx.integration.transform((integrations) => { for (const item of Object.values(loaded.data)) { if (item.env.length === 0) continue @@ -252,15 +254,16 @@ export const ModelsDevPlugin = define({ } } }) + const refresh = modelsDev.get().pipe( + Effect.tap((data) => Effect.sync(() => (loaded.data = data))), + Effect.andThen(ctx.integration.reload()), + Effect.andThen(ctx.catalog.reload()), + Effect.tapCause((cause) => Effect.logError("failed to load models.dev catalog", { cause })), + ) yield* events.subscribe(ModelsDev.Event.Refreshed).pipe( - Stream.runForEach(() => - modelsDev.get().pipe( - Effect.tap((data) => Effect.sync(() => (loaded.data = data))), - Effect.andThen(ctx.integration.reload()), - Effect.andThen(ctx.catalog.reload()), - ), - ), + Stream.runForEach(() => refresh.pipe(Effect.ignoreCause)), Effect.forkScoped({ startImmediately: true }), ) + yield* catalog.initial.source(refresh) }), }) diff --git a/packages/core/src/plugin/provider/openai.ts b/packages/core/src/plugin/provider/openai.ts index 990a82554e..2820c0b88a 100644 --- a/packages/core/src/plugin/provider/openai.ts +++ b/packages/core/src/plugin/provider/openai.ts @@ -4,6 +4,7 @@ import { define } from "@opencode-ai/plugin/v2/effect/plugin" import { Deferred, Effect, Option, Schema, Semaphore, Stream } from "effect" import type { Scope } from "effect" import { Credential } from "../../credential" +import { Catalog } from "../../catalog" import { EventV2 } from "../../event" import { InstallationVersion } from "../../installation/version" import { Integration } from "../../integration" @@ -162,14 +163,13 @@ export const OpenAIPlugin = define({ id: "openai", effect: Effect.fn(function* (ctx) { const events = yield* EventV2.Service + const catalog = yield* Catalog.Service const loading = Semaphore.makeUnsafe(1) let chatgpt = false 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.catch(() => Effect.succeed(undefined))) - : undefined + const credential = connection ? yield* ctx.integration.connection.resolve(connection) : undefined chatgpt = OpenAICodex.isChatGPT(credential) }) @@ -204,13 +204,16 @@ export const OpenAIPlugin = define({ } }) - const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.catalog.reload()))) + const refresh = () => + loading + .withPermit(load().pipe(Effect.andThen(ctx.catalog.reload()))) + .pipe(Effect.tapCause((cause) => Effect.logWarning("failed to refresh OpenAI catalog", { cause }))) yield* events.subscribe(Integration.Event.ConnectionUpdated).pipe( Stream.filter((event) => event.data.integrationID === Integration.ID.make("openai")), - Stream.runForEach(refresh), + Stream.runForEach(() => refresh().pipe(Effect.ignoreCause)), Effect.forkScoped({ startImmediately: true }), ) - yield* refresh().pipe(Effect.forkScoped) + yield* catalog.initial.source(refresh()) yield* ctx.aisdk.sdk( Effect.fn(function* (evt) { if (evt.package !== "@ai-sdk/openai") return diff --git a/packages/core/src/plugin/provider/opencode.ts b/packages/core/src/plugin/provider/opencode.ts index bfa84ecd78..0402465d23 100644 --- a/packages/core/src/plugin/provider/opencode.ts +++ b/packages/core/src/plugin/provider/opencode.ts @@ -5,6 +5,7 @@ import { define } from "@opencode-ai/plugin/v2/effect/plugin" import type { CredentialValue } from "@opencode-ai/sdk/v2/types" import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http" import { EventV2 } from "../../event" +import { Catalog } from "../../catalog" import { Credential } from "../../credential" import { Integration } from "../../integration" import { ModelV2 } from "../../model" @@ -74,10 +75,11 @@ function oauth(http: HttpClient.HttpClient) { } satisfies IntegrationOAuthMethodRegistration } -export const OpencodePlugin = define({ +export const OpencodePlugin = define({ id: "opencode", effect: Effect.fn(function* (ctx) { const events = yield* EventV2.Service + const catalog = yield* Catalog.Service const http = yield* HttpClient.HttpClient const loading = Semaphore.makeUnsafe(1) let connected = false @@ -89,13 +91,7 @@ export const OpencodePlugin = define Effect.succeed(undefined))) : undefined connected = connection !== undefined - providers = credential - ? yield* fetchProviders(http, credential).pipe( - Effect.catch((cause) => - Effect.logWarning("failed to load OpenCode provider config", { cause }).pipe(Effect.as(undefined)), - ), - ) - : undefined + providers = credential ? yield* fetchProviders(http, credential) : undefined }) yield* ctx.integration.transform((draft) => { @@ -182,13 +178,16 @@ export const OpencodePlugin = define loading.withPermit(load().pipe(Effect.andThen(ctx.catalog.reload()))) + const refresh = () => + loading + .withPermit(load().pipe(Effect.andThen(ctx.catalog.reload()))) + .pipe(Effect.tapCause((cause) => Effect.logWarning("failed to load OpenCode provider config", { cause }))) yield* events.subscribe(Integration.Event.ConnectionUpdated).pipe( Stream.filter((event) => event.data.integrationID === Integration.ID.make("opencode")), - Stream.runForEach(refresh), + Stream.runForEach(() => refresh().pipe(Effect.ignoreCause)), Effect.forkScoped({ startImmediately: true }), ) - yield* refresh().pipe(Effect.forkScoped) + yield* catalog.initial.source(refresh()) }), }) diff --git a/packages/core/src/session/runner/model.ts b/packages/core/src/session/runner/model.ts index 2b81e609ea..b707463971 100644 --- a/packages/core/src/session/runner/model.ts +++ b/packages/core/src/session/runner/model.ts @@ -69,6 +69,7 @@ export class UnsupportedApiError extends Schema.TaggedErrorClass model.providerID === session.model?.providerID && model.id === session.model.id, - ) - : defaultModel && supported(defaultModel) - ? defaultModel - : (yield* catalog.model.available()).find(supported) + const selected = yield* catalog.model.select(session.model, supported) if (!selected && session.model) return yield* new ModelUnavailableError({ providerID: session.model.providerID, diff --git a/packages/core/src/state.ts b/packages/core/src/state.ts index ab3457fc18..35350cf7f5 100644 --- a/packages/core/src/state.ts +++ b/packages/core/src/state.ts @@ -48,6 +48,8 @@ export interface Options { readonly draft: MakeDraft /** Runs after all active transforms and before the rebuilt state becomes visible. */ readonly finalize?: (draft: DraftApi) => Effect.Effect + /** Runs after the rebuilt state becomes visible. */ + readonly committed?: () => Effect.Effect } export interface Interface extends Transformable { @@ -67,6 +69,7 @@ export function create(options: Options): Inte const api = options.draft(next) if (options.finalize) yield* options.finalize(api) state = next + if (options.committed) yield* options.committed() }) const apply = (transform: TransformCallback, draft: DraftApi) => diff --git a/packages/core/test/catalog.test.ts b/packages/core/test/catalog.test.ts index 04bc006888..4ae7b1efc2 100644 --- a/packages/core/test/catalog.test.ts +++ b/packages/core/test/catalog.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { Effect, Fiber, Layer, Stream } from "effect" +import { DateTime, Deferred, Effect, Exit, Fiber, Layer, Scope, Stream } from "effect" import { Catalog } from "@opencode-ai/core/catalog" import { Integration } from "@opencode-ai/core/integration" import { Credential } from "@opencode-ai/core/credential" @@ -9,7 +9,10 @@ import { EventV2 } from "@opencode-ai/core/event" import { Location } from "@opencode-ai/core/location" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" +import { ProjectV2 } from "@opencode-ai/core/project" import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" import { location } from "./fixture/location" import { testEffect } from "./lib/effect" @@ -27,8 +30,341 @@ const catalogLayer = AppNodeBuilder.build( [[Location.node, locationLayer]], ) const it = testEffect(catalogLayer) +const sessionModelLayer = AppNodeBuilder.build( + LayerNode.group([Catalog.node, EventV2.node, Credential.node, Integration.node, SessionRunnerModel.node]), + [[Location.node, locationLayer]], +) +const sessionModelIt = testEffect(sessionModelLayer) + +const session = (providerID?: ProviderV2.ID, modelID?: ModelV2.ID) => + SessionV2.Info.make({ + id: SessionV2.ID.make("ses_catalog_readiness"), + projectID: ProjectV2.ID.global, + title: "test", + ...(providerID === undefined || modelID === undefined ? {} : { model: { providerID, id: modelID } }), + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + time: { created: DateTime.makeUnsafe(0), updated: DateTime.makeUnsafe(0) }, + location: { directory: AbsolutePath.make("test") }, + }) + +const addSupportedModel = (catalog: Catalog.Interface, providerID: ProviderV2.ID, modelID: ModelV2.ID) => + catalog.transform((draft) => { + draft.provider.update(providerID, (provider) => { + provider.api = { type: "aisdk", package: "@ai-sdk/openai", settings: {} } + }) + draft.model.update(providerID, modelID, () => {}) + }) describe("CatalogV2", () => { + it.effect("keeps available snapshots nonblocking during initial catalog work", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const release = yield* Deferred.make() + yield* catalog.initial.discover(catalog.initial.source(Deferred.await(release))) + + expect(yield* catalog.model.available()).toEqual([]) + + yield* Deferred.succeed(release, undefined) + }), + ) + + it.effect("waits until an explicit model appears", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const providerID = ProviderV2.ID.make("delayed-provider") + const modelID = ModelV2.ID.make("delayed-model") + const release = yield* Deferred.make() + const completed = yield* Deferred.make() + yield* catalog.initial.discover( + catalog.initial.source( + Deferred.await(release).pipe(Effect.andThen(addSupportedModel(catalog, providerID, modelID))), + ), + ) + const selected = yield* catalog.model + .select(ModelV2.Ref.make({ providerID, id: modelID }), () => true) + .pipe( + Effect.tap((model) => Deferred.succeed(completed, model)), + Effect.forkScoped({ startImmediately: true }), + ) + + expect(yield* Deferred.isDone(completed)).toBe(false) + yield* Deferred.succeed(release, undefined) + + expect((yield* Fiber.join(selected))?.id).toBe(modelID) + }), + ) + + it.effect("waits for discovery contributors before returning an existing explicit model", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const providerID = ProviderV2.ID.make("overlaid-provider") + const modelID = ModelV2.ID.make("overlaid-model") + const modelVisible = yield* Deferred.make() + const releaseOverlay = yield* Deferred.make() + const completed = yield* Deferred.make() + yield* catalog.initial + .discover( + Effect.gen(function* () { + yield* addSupportedModel(catalog, providerID, modelID) + yield* Deferred.succeed(modelVisible, undefined) + yield* Deferred.await(releaseOverlay) + yield* catalog.transform((draft) => + draft.model.update(providerID, modelID, (model) => { + model.name = "Configured model" + }), + ) + }), + ) + .pipe(Effect.forkScoped({ startImmediately: true })) + yield* Deferred.await(modelVisible) + const selected = yield* catalog.model + .select(ModelV2.Ref.make({ providerID, id: modelID }), () => true) + .pipe( + Effect.tap((model) => Deferred.succeed(completed, model)), + Effect.forkScoped({ startImmediately: true }), + ) + + expect(yield* Deferred.isDone(completed)).toBe(false) + yield* Deferred.succeed(releaseOverlay, undefined) + + expect((yield* Fiber.join(selected))?.name).toBe("Configured model") + }), + ) + + it.effect("selects an explicit model as soon as it arrives without waiting for unrelated initial work", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const providerID = ProviderV2.ID.make("ready-provider") + const modelID = ModelV2.ID.make("ready-model") + const releaseModel = yield* Deferred.make() + const releaseUnrelated = yield* Deferred.make() + yield* catalog.initial.discover( + Effect.gen(function* () { + yield* catalog.initial.source( + Deferred.await(releaseModel).pipe(Effect.andThen(addSupportedModel(catalog, providerID, modelID))), + ) + yield* catalog.initial.source(Deferred.await(releaseUnrelated)) + }), + ) + const selected = yield* catalog.model + .select(ModelV2.Ref.make({ providerID, id: modelID }), () => true) + .pipe(Effect.forkScoped({ startImmediately: true })) + + yield* Deferred.succeed(releaseModel, undefined) + + expect((yield* Fiber.join(selected))?.id).toBe(modelID) + expect(yield* Deferred.isDone(releaseUnrelated)).toBe(false) + yield* Deferred.succeed(releaseUnrelated, undefined) + }), + ) + + it.effect("waits for a pending source after another initial source fails", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const providerID = ProviderV2.ID.make("recovering-provider") + const modelID = ModelV2.ID.make("recovering-model") + const releaseModel = yield* Deferred.make() + const completed = yield* Deferred.make() + yield* catalog.initial.discover( + Effect.gen(function* () { + yield* catalog.initial.source(Effect.fail(new Error("unrelated source failed"))) + yield* catalog.initial.source( + Deferred.await(releaseModel).pipe(Effect.andThen(addSupportedModel(catalog, providerID, modelID))), + ) + }), + ) + const selected = yield* catalog.model + .select(ModelV2.Ref.make({ providerID, id: modelID }), () => true) + .pipe( + Effect.tap((model) => Deferred.succeed(completed, model)), + Effect.forkScoped({ startImmediately: true }), + ) + + expect(yield* Deferred.isDone(completed)).toBe(false) + yield* Deferred.succeed(releaseModel, undefined) + + expect((yield* Fiber.join(selected))?.id).toBe(modelID) + }), + ) + + it.effect("waits for a pending source after discovery fails", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const providerID = ProviderV2.ID.make("discovery-failure-provider") + const modelID = ModelV2.ID.make("discovery-failure-model") + const releaseModel = yield* Deferred.make() + const completed = yield* Deferred.make() + yield* catalog.initial.discover( + Effect.gen(function* () { + yield* catalog.initial.source( + Deferred.await(releaseModel).pipe(Effect.andThen(addSupportedModel(catalog, providerID, modelID))), + ) + return yield* Effect.fail(new Error("discovery failed after source registration")) + }), + ) + const selected = yield* catalog.model + .select(ModelV2.Ref.make({ providerID, id: modelID }), () => true) + .pipe( + Effect.tap((model) => Deferred.succeed(completed, model)), + Effect.forkScoped({ startImmediately: true }), + ) + + expect(yield* Deferred.isDone(completed)).toBe(false) + yield* Deferred.succeed(releaseModel, undefined) + + expect((yield* Fiber.join(selected))?.id).toBe(modelID) + }), + ) + + it.effect("observes discovery settlement completed before lookup subscribes", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + yield* catalog.initial.discover(Effect.void) + + const selected = yield* catalog.model.select( + ModelV2.Ref.make({ providerID: ProviderV2.ID.make("missing"), id: ModelV2.ID.make("missing") }), + () => true, + ) + + expect(selected).toBeUndefined() + }), + ) + + it.effect("waits for initial settlement before choosing an omitted model", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const earlyProviderID = ProviderV2.ID.make("early-provider") + const earlyModelID = ModelV2.ID.make("early-model") + const finalProviderID = ProviderV2.ID.make("final-provider") + const finalModelID = ModelV2.ID.make("final-model") + const release = yield* Deferred.make() + const completed = yield* Deferred.make() + yield* addSupportedModel(catalog, earlyProviderID, earlyModelID) + yield* catalog.initial.discover( + catalog.initial.source( + Deferred.await(release).pipe( + Effect.andThen( + catalog.transform((draft) => { + draft.provider.update(finalProviderID, (provider) => { + provider.api = { type: "aisdk", package: "@ai-sdk/openai", settings: {} } + }) + draft.model.update(finalProviderID, finalModelID, () => {}) + draft.model.default.set(finalProviderID, finalModelID) + }), + ), + ), + ), + ) + const selected = yield* catalog.model + .select(undefined, () => true) + .pipe( + Effect.tap((model) => Deferred.succeed(completed, model)), + Effect.forkScoped({ startImmediately: true }), + ) + + expect(yield* Deferred.isDone(completed)).toBe(false) + yield* Deferred.succeed(release, undefined) + + expect((yield* Fiber.join(selected))?.id).toBe(finalModelID) + }), + ) + + sessionModelIt.effect("reports an unavailable explicit model after successful initial settlement", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const models = yield* SessionRunnerModel.Service + const providerID = ProviderV2.ID.make("missing") + const modelID = ModelV2.ID.make("missing") + yield* catalog.initial.discover(Effect.void) + + const failure = yield* models.resolve(session(providerID, modelID)).pipe(Effect.flip) + + expect(failure).toMatchObject({ + _tag: "SessionRunnerModel.ModelUnavailableError", + providerID, + modelID, + }) + }), + ) + + sessionModelIt.effect("reports catalog incompleteness when a source fails after discovery settles", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const models = yield* SessionRunnerModel.Service + const release = yield* Deferred.make() + const completed = yield* Deferred.make() + yield* catalog.initial.discover( + catalog.initial.source( + Deferred.await(release).pipe(Effect.andThen(Effect.fail(new Error("initial source failed")))), + ), + ) + const result = yield* models + .resolve(session(ProviderV2.ID.make("missing"), ModelV2.ID.make("missing"))) + .pipe( + Effect.flip, + Effect.tap((failure) => Deferred.succeed(completed, failure)), + Effect.forkScoped({ startImmediately: true }), + ) + + expect(yield* Deferred.isDone(completed)).toBe(false) + yield* Deferred.succeed(release, undefined) + + const failure = yield* Fiber.join(result) + + expect(failure).toMatchObject({ _tag: "Catalog.IncompleteError" }) + }), + ) + + it.effect("settles interrupted initial sources as incomplete", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const sourceScope = yield* Scope.make() + const started = yield* Deferred.make() + yield* catalog.initial.discover( + catalog.initial + .source(Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never))) + .pipe(Scope.provide(sourceScope)), + ) + yield* Deferred.await(started) + + yield* Scope.close(sourceScope, Exit.void) + + const failure = yield* catalog.model + .select( + ModelV2.Ref.make({ providerID: ProviderV2.ID.make("missing"), id: ModelV2.ID.make("missing") }), + () => true, + ) + .pipe(Effect.flip) + expect(failure).toMatchObject({ _tag: "Catalog.IncompleteError" }) + }), + ) + + sessionModelIt.effect("preserves the supported-model fallback after initial settlement", () => + Effect.gen(function* () { + const catalog = yield* Catalog.Service + const models = yield* SessionRunnerModel.Service + const unsupportedProviderID = ProviderV2.ID.make("unsupported-provider") + const unsupportedModelID = ModelV2.ID.make("unsupported-model") + const supportedProviderID = ProviderV2.ID.make("supported-provider") + const supportedModelID = ModelV2.ID.make("supported-model") + yield* catalog.transform((draft) => { + draft.provider.update(unsupportedProviderID, () => {}) + draft.model.update(unsupportedProviderID, unsupportedModelID, () => {}) + draft.model.default.set(unsupportedProviderID, unsupportedModelID) + draft.provider.update(supportedProviderID, (provider) => { + provider.api = { type: "aisdk", package: "@ai-sdk/openai", settings: {} } + }) + draft.model.update(supportedProviderID, supportedModelID, () => {}) + }) + yield* catalog.initial.discover(Effect.void) + + const selected = yield* models.resolve(session()) + + expect(selected.ref).toMatchObject({ providerID: supportedProviderID, id: supportedModelID }) + }), + ) + it.effect("publishes an updated event after catalog changes", () => Effect.gen(function* () { const catalog = yield* Catalog.Service diff --git a/packages/core/test/location-layer.test.ts b/packages/core/test/location-layer.test.ts index 0f17b0dd2b..cd66efba38 100644 --- a/packages/core/test/location-layer.test.ts +++ b/packages/core/test/location-layer.test.ts @@ -1,13 +1,13 @@ import fs from "fs/promises" import path from "path" import { describe, expect } from "bun:test" -import { DateTime, Effect, Equal, Hash, Schema } from "effect" +import { DateTime, Deferred, Effect, Equal, Hash, Schema } from "effect" import { define } from "@opencode-ai/plugin/v2/effect" import { AgentV2 } from "@opencode-ai/core/agent" import { Catalog } from "@opencode-ai/core/catalog" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" -import { LocationServiceMap } from "@opencode-ai/core/location-services" +import { buildLocationServiceMap, LocationServiceMap } from "@opencode-ai/core/location-services" import { Location } from "@opencode-ai/core/location" import { PluginV2 } from "@opencode-ai/core/plugin" import { ModelV2 } from "@opencode-ai/core/model" @@ -16,6 +16,7 @@ import { ProviderV2 } from "@opencode-ai/core/provider" import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionV2 } from "@opencode-ai/core/session" import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" +import { SdkPlugins } from "@opencode-ai/core/plugin/sdk" import { tmpdir } from "./fixture/tmpdir" import { testEffect } from "./lib/effect" import { toolDefinitions, waitForTool } from "./lib/tool" @@ -27,6 +28,53 @@ import { ToolRegistry } from "../src/tool/registry" const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, LocationServiceMap.node]))) describe("LocationServiceMap", () => { + it.live("acquires location services while initial catalog discovery is blocked", () => + Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), + ).pipe( + Effect.flatMap((dir) => + Effect.gen(function* () { + const started = yield* Deferred.make() + const release = yield* Deferred.make() + const store = SdkPlugins.makeStore() + store.plugins.set( + "blocked-catalog-source", + define({ + id: "blocked-catalog-source", + effect: (ctx) => + ctx.catalog + .transform((draft) => draft.provider.update("blocked-catalog-source", () => {})) + .pipe( + Effect.andThen(Deferred.succeed(started, undefined)), + Effect.andThen(Deferred.await(release)), + Effect.asVoid, + ), + }), + ) + + return yield* Effect.gen(function* () { + const locations = yield* LocationServiceMap.Service + const context = yield* locations.contextEffect( + Location.Ref.make({ directory: AbsolutePath.make(dir.path) }), + ) + yield* Deferred.await(started) + + const snapshot = yield* Catalog.Service.use((catalog) => catalog.model.available()).pipe( + Effect.provide(context), + ) + expect(Array.isArray(snapshot)).toBe(true) + yield* Deferred.succeed(release, undefined) + }).pipe( + Effect.provide(buildLocationServiceMap([[SdkPlugins.node, SdkPlugins.layerWithStore(store)]]), { + local: true, + }), + ) + }), + ), + ), + ) + it.live("reuses cached services for constructed and decoded location refs", () => Effect.acquireRelease( Effect.promise(() => tmpdir()), diff --git a/packages/core/test/state.test.ts b/packages/core/test/state.test.ts index 505cd724e7..e0f9144e37 100644 --- a/packages/core/test/state.test.ts +++ b/packages/core/test/state.test.ts @@ -56,6 +56,26 @@ describe("State", () => { }), ) + it.effect("runs committed after the rebuilt state is visible", () => + Effect.gen(function* () { + let visible: string[] = [] + const state = State.create({ + initial: () => ({ values: [] as string[] }), + draft: (draft) => ({ add: (item: string) => draft.values.push(item) }), + committed: () => + Effect.sync(() => { + visible = [...state.get().values] + }), + }) + + yield* state.transform((editor) => { + editor.add("committed") + }) + + expect(visible).toEqual(["committed"]) + }), + ) + it.effect("disposes a transform once and rebuilds remaining state", () => Effect.gen(function* () { const state = State.create({