Compare commits

..
4 Commits
57 changed files with 441 additions and 2567 deletions
+1 -2
View File
@@ -417,7 +417,6 @@ jobs:
- uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3.6.0
- uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
if: github.ref_name == 'beta'
with:
name: opencode-preview-cli
path: packages/cli/dist
@@ -480,7 +479,7 @@ jobs:
OPENCODE_VERSION: ${{ needs.version.outputs.version }}
OPENCODE_CHANNEL: ${{ (github.ref_name == 'beta' && 'beta') || 'prod' }}
OPENCODE_CLI_TARGET: ${{ matrix.settings.target }}
OPENCODE_CLI_DIST: ${{ (github.ref_name == 'beta' && format('{0}/packages/cli/dist', github.workspace)) || '' }}
OPENCODE_CLI_DIST: ${{ github.workspace }}/packages/cli/dist
- name: Build
run: bun run build
+5
View File
@@ -87,6 +87,11 @@ stdenv.mkDerivation (finalAttrs: {
cd packages/desktop
export OPENCODE_CLI_DIST="$TMPDIR/desktop-cli"
cli_package=$(bun -e 'import { getCurrentCli } from "./scripts/utils.ts"; console.log(getCurrentCli().package.replace("@opencode-ai/", ""))')
mkdir -p "$OPENCODE_CLI_DIST/$cli_package/bin"
cp ${lib.getExe opencode} "$OPENCODE_CLI_DIST/$cli_package/bin/opencode2"
bun run build
npx electron-builder --dir \
--config electron-builder.config.ts \
@@ -1,10 +1,5 @@
import { describe, expect, test } from "bun:test"
import {
normalizeNewSessionWorktree,
resolveNewSessionBranch,
resolveNewSessionGit,
resolveNewSessionWorktree,
} from "./controller"
import { resolveNewSessionBranch, resolveNewSessionGit, resolveNewSessionWorktree } from "./controller"
describe("new session workspace selection", () => {
test("uses main when the workspace bar is unavailable", () => {
@@ -12,8 +7,6 @@ describe("new session workspace selection", () => {
resolveNewSessionWorktree({
enabled: false,
selected: "/project/feature",
directory: "/project/feature",
projectWorktree: "/project",
}),
).toBe("main")
})
@@ -22,31 +15,21 @@ describe("new session workspace selection", () => {
expect(
resolveNewSessionWorktree({
enabled: true,
directory: "/project/feature",
projectWorktree: "/project",
fallback: "create",
}),
).toBe("create")
expect(
resolveNewSessionWorktree({
enabled: true,
directory: "/project/feature",
projectWorktree: "/project",
fallback: "main",
}),
).toBe("/project")
})
test("normalizes main to the project root outside the main worktree", () => {
expect(normalizeNewSessionWorktree("main", "/project/feature", "/project")).toBe("/project")
expect(normalizeNewSessionWorktree("main", "/project", "/project")).toBe("main")
})
test("treats equivalent Windows roots as the main worktree", () => {
expect(
resolveNewSessionWorktree({ enabled: true, directory: "C:\\Repo\\", projectWorktree: "c:/repo" }),
).toBe("main")
expect(normalizeNewSessionWorktree("main", "C:\\Repo\\", "c:/repo")).toBe("main")
})
test("keeps local selection when the cached project path is stale", () => {
const input = { enabled: true, directory: "C:/Projects/repo", projectWorktree: "D:/Projects/repo" }
expect(resolveNewSessionWorktree(input)).toBe("main")
expect(resolveNewSessionWorktree({ ...input, selected: "/worktree" })).toBe("/worktree")
})
test("resolves the branch from the active location", () => {
@@ -11,27 +11,15 @@ import { normalizeProjectInfo } from "@/runtime/server/global-sync/utils"
import {
isWorkspaceDirectory,
isWorkspaceSelection,
sameDirectory,
workspaceDefaultSelection,
workspaceDirectories,
workspaceSelectionDestination,
} from "@/workspaces/paths"
export function resolveNewSessionWorktree(input: {
enabled: boolean
selected?: string
directory: string
projectWorktree?: string
fallback?: string
}) {
export function resolveNewSessionWorktree(input: { enabled: boolean; selected?: string; fallback?: string }) {
if (!input.enabled) return "main"
if (input.selected) return input.selected
return normalizeNewSessionWorktree(input.fallback ?? "main", input.directory, input.projectWorktree)
}
export function normalizeNewSessionWorktree(value: string, directory: string, projectWorktree?: string) {
if (value === "main" && projectWorktree && !sameDirectory(directory, projectWorktree)) return projectWorktree
return value
return input.fallback ?? "main"
}
export function resolveNewSessionBranch(input: {
@@ -92,8 +80,6 @@ export function createNewSessionWorkspaceController(input: {
resolveNewSessionWorktree({
enabled: visible(),
selected: selected(),
directory: sdk().directory,
projectWorktree: currentProject()?.worktree,
fallback: fallback(),
}),
)
@@ -145,7 +131,7 @@ export function createNewSessionWorkspaceController(input: {
remember,
set: (worktree: string) => {
input.setSelectedBranch(undefined)
input.setSelectedWorktree(normalizeNewSessionWorktree(worktree, sdk().directory, currentProject()?.worktree))
input.setSelectedWorktree(worktree)
remember(worktree)
},
create: (branch: string) => {
+30 -15
View File
@@ -1,25 +1,40 @@
import { createMemo, type Accessor } from "solid-js"
import { createMemo, createSignal, onCleanup, type Accessor } from "solid-js"
import createPresence from "solid-presence"
export function createAnimatedPresence<T>(
value: Accessor<T | undefined>,
element: Accessor<HTMLElement | null>,
identity?: Accessor<unknown>,
minimumDuration = 0,
) {
const animation = createMemo<{ identity?: unknown; show: boolean; animate: boolean; value: T | undefined }>(
(previous) => {
const currentIdentity = identity?.()
const current = value()
const show = current !== undefined
const same = !identity || previous?.identity === currentIdentity
return {
identity: currentIdentity,
show,
animate: previous !== undefined && same && (previous.animate || previous.show !== show),
value: current ?? (same ? previous?.value : undefined),
}
},
)
const [tick, setTick] = createSignal(0)
const animation = createMemo<{
identity?: unknown
show: boolean
animate: boolean
value: T | undefined
started: number
}>((previous) => {
tick()
const currentIdentity = identity?.()
const current = value()
const same = !identity || previous?.identity === currentIdentity
const started = same && previous?.show ? previous.started : performance.now()
const remaining =
current === undefined && same && previous?.show ? minimumDuration - (performance.now() - started) : 0
const show = current !== undefined || remaining > 0
if (remaining > 0) {
const timer = setTimeout(() => setTick((value) => value + 1), remaining)
onCleanup(() => clearTimeout(timer))
}
return {
identity: currentIdentity,
show,
started,
animate: previous !== undefined && same && (previous.animate || previous.show !== show),
value: current ?? (same ? previous?.value : undefined),
}
})
const presence = createPresence({ show: () => animation().show, element })
return {
...presence,
@@ -539,7 +539,12 @@ function MessageTimelineView(
.findLast((ref) => blocking.has(ref.partID))?.partID
})
const [backgroundHintRef, setBackgroundHintRef] = createSignal<HTMLDivElement>()
const backgroundHintPresence = createAnimatedPresence(backgroundHintPartID, () => backgroundHintRef() ?? null)
const backgroundHintPresence = createAnimatedPresence(
backgroundHintPartID,
() => backgroundHintRef() ?? null,
sessionID,
1000,
)
const showWorking = createMemo(() => {
const id = sessionID()
if (!id || sessionStatus().type !== "busy") return false
+1 -2
View File
@@ -400,8 +400,7 @@ function turnStart(messageID: string, slash: PreparedPrompt["slash"], skill: Ski
async function loadCatalog(client: OpenCodeClient, cwd: string): Promise<Catalog> {
const location = { directory: cwd }
await client.plugin.awaitActivation({ location })
// Some providers discover models in the background after activation has settled.
// Location plugins initialize asynchronously, so the first ACP request may observe an empty catalog.
const deadline = Date.now() + 5_000
let missing = "No models are available"
while (Date.now() < deadline) {
@@ -1,50 +1,9 @@
import { describe, expect, test } from "bun:test"
import type { McpServer, SessionConfigOption } from "@agentclientprotocol/sdk"
import { makeACPFixture, makeSession, secondModel, testModel } from "./service-fixture"
import { makeACPFixture, makeSession, secondModel } from "./service-fixture"
import { flattenSelectOptions, requireSelectOption } from "./subprocess"
describe("acp service directory behavior", () => {
test("does not cache an available model before plugin activation settles", async () => {
const requested = Promise.withResolvers<void>()
const release = Promise.withResolvers<void>()
let ready = false
await using fixture = makeACPFixture({
fetch(request) {
requested.resolve()
if (request.path === "/api/plugin/await-activation") {
return release.promise.then(() => {
ready = true
return new Response(null, { status: 204 })
})
}
if (!ready && request.path === "/api/model") {
return Response.json({ data: [{ ...testModel, providerID: "ambient" }] })
}
if (!ready && request.path === "/api/model/default") {
return Response.json({ data: { ...testModel, providerID: "ambient" } })
}
if (request.path === "/api/session" && request.method === "POST") {
return Response.json({ data: { ...makeSession("ses_ready"), model: undefined } })
}
return undefined
},
})
const pending = fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
try {
await requested.promise
expect(fixture.requests.map((request) => request.path)).toEqual(["/api/plugin/await-activation"])
expect(fixture.requests[0]?.query["location[directory]"]).toBe("/workspace")
release.resolve()
expect(currentValue(await pending, "model")).toBe("test/test-model")
expect(
fixture.requests.find((request) => request.path === "/api/session" && request.method === "POST")?.body,
).toMatchObject({ model: { providerID: "test", id: "test-model" } })
} finally {
release.resolve()
await pending.catch(() => {})
}
})
test("creates sessions from a catalog shared by concurrent callers in the same cwd", async () => {
let created = 0
await using fixture = makeACPFixture({
@@ -68,14 +27,7 @@ describe("acp service directory behavior", () => {
expect(currentValue(first[0], "model")).toBe("test/test-model")
expect(currentValue(first[0], "mode")).toBe("build")
expect(
[
"/api/plugin/await-activation",
"/api/model",
"/api/model/default",
"/api/agent",
"/api/command",
"/api/skill",
].map((path) =>
["/api/model", "/api/model/default", "/api/agent", "/api/command", "/api/skill"].map((path) =>
fixture.requests
.filter((request) => request.path === path)
.map((request) => request.query["location[directory]"]),
@@ -86,7 +38,6 @@ describe("acp service directory behavior", () => {
["/workspace", "/other"],
["/workspace", "/other"],
["/workspace", "/other"],
["/workspace", "/other"],
])
expect(
fixture.requests
-1
View File
@@ -152,7 +152,6 @@ export function makeACPFixture(options: FixtureOptions = {}) {
const directory = request.query["location[directory]"] ?? "/workspace"
const location = { directory, project: { id: "global", directory } }
if (request.path === "/api/plugin/await-activation") return new Response(null, { status: 204 })
if (request.path === "/api/event") {
let controller: ReadableStreamDefaultController<Uint8Array> | undefined
return new Response(
-1
View File
@@ -18,7 +18,6 @@ describe("acp service", () => {
body: request.method === "GET" ? undefined : await request.json().catch(() => undefined),
})
const location = { directory: "/workspace", project: { id: "global", directory: "/workspace" } }
if (url.pathname === "/api/plugin/await-activation") return new Response(null, { status: 204 })
if (url.pathname === "/api/model") return Response.json({ location, data: [model] })
if (url.pathname === "/api/model/default") return Response.json({ location, data: model })
if (url.pathname === "/api/agent") return Response.json({ location, data: [agent] })
-9
View File
@@ -88,14 +88,6 @@ export type PluginListInput = {
export type PluginListOutput = { readonly location: Location.Info; readonly data: ReadonlyArray<Plugin.Info> }
export type PluginListOperation<E = never> = (input?: PluginListInput) => Effect.Effect<PluginListOutput, E>
export type PluginAwaitActivationInput = {
readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined
}
export type PluginAwaitActivationOutput = void
export type PluginAwaitActivationOperation<E = never> = (
input?: PluginAwaitActivationInput,
) => Effect.Effect<PluginAwaitActivationOutput, E>
export type PluginCheckInput = {
readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined
readonly target?: string | undefined
@@ -112,7 +104,6 @@ export type PluginUpdateOperation<E = never> = (input: PluginUpdateInput) => Eff
export interface PluginApi<E = never> {
readonly list: PluginListOperation<E>
readonly awaitActivation: PluginAwaitActivationOperation<E>
readonly check: PluginCheckOperation<E>
readonly update: PluginUpdateOperation<E>
}
@@ -15,8 +15,6 @@ import type {
AgentGetOutput,
PluginListInput,
PluginListOutput,
PluginAwaitActivationInput,
PluginAwaitActivationOutput,
PluginCheckInput,
PluginCheckOutput,
PluginUpdateInput,
@@ -328,11 +326,6 @@ const EndpointPluginList = (raw: RawClient["server.plugin"]) => (input?: PluginL
raw["plugin.list"]({ query: { location: input?.["location"] } }).pipe(Effect.mapError(mapClientError)),
)
const EndpointPluginAwaitActivation = (raw: RawClient["server.plugin"]) => (input?: PluginAwaitActivationInput) =>
preserveEffect<PluginAwaitActivationOutput>()(
raw["plugin.awaitActivation"]({ query: { location: input?.["location"] } }).pipe(Effect.mapError(mapClientError)),
)
const EndpointPluginCheck = (raw: RawClient["server.plugin"]) => (input?: PluginCheckInput) =>
preserveEffect<PluginCheckOutput>()(
raw["plugin.check"]({ query: { location: input?.["location"] }, payload: { target: input?.["target"] } }).pipe(
@@ -349,7 +342,6 @@ const EndpointPluginUpdate = (raw: RawClient["server.plugin"]) => (input: Plugin
const adaptGroupPlugin = (raw: RawClient["server.plugin"]) => ({
list: EndpointPluginList(raw),
awaitActivation: EndpointPluginAwaitActivation(raw),
check: EndpointPluginCheck(raw),
update: EndpointPluginUpdate(raw),
})
@@ -9,8 +9,6 @@ import type {
AgentGetOutput,
PluginListInput,
PluginListOutput,
PluginAwaitActivationInput,
PluginAwaitActivationOutput,
PluginCheckInput,
PluginCheckOutput,
PluginUpdateInput,
@@ -472,18 +470,6 @@ export function make(options: ClientOptions) {
},
requestOptions,
),
awaitActivation: (input?: PluginAwaitActivationInput, requestOptions?: RequestOptions) =>
request<PluginAwaitActivationOutput>(
{
method: "POST",
path: `/api/plugin/await-activation`,
query: { location: input?.["location"] },
successStatus: 204,
declaredStatuses: [400, 401],
empty: true,
},
requestOptions,
),
check: (input?: PluginCheckInput, requestOptions?: RequestOptions) =>
request<PluginCheckOutput>(
{
@@ -2585,14 +2585,6 @@ export type PluginListOutput = {
data: Array<PluginInfo>
}
export type PluginAwaitActivationInput = {
readonly location?: {
readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined
}["location"]
}
export type PluginAwaitActivationOutput = void
export type PluginCheckInput = {
readonly location?: {
readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined
+1 -1
View File
@@ -87,7 +87,7 @@ const layer = Layer.effect(
draft.agents.delete(id)
},
}),
notify: bus.publish(Agent.Event.Updated, {}).pipe(Effect.asVoid),
finalize: () => bus.publish(Agent.Event.Updated, {}).pipe(Effect.asVoid),
})
const selectable = (agent: Info | undefined) =>
agent && agent.mode !== "subagent" && !agent.hidden ? agent : undefined
+3 -1
View File
@@ -135,7 +135,9 @@ const layer = Layer.effect(
}
return result
},
notify: bus.publish(Catalog.Event.Updated, {}).pipe(Effect.asVoid, Effect.withSpan("Catalog.notify")),
finalize: Effect.fn("Catalog.finalize")(function* () {
yield* bus.publish(Catalog.Event.Updated, {})
}),
})
const result: Interface = {
transform: state.transform,
+1 -1
View File
@@ -60,7 +60,7 @@ export const layer = Layer.effect(
draft: (draft) => ({
add: (definition) => draft.set(definition.name, definition),
}),
notify: bus.publish(Command.Event.Updated, {}).pipe(Effect.asVoid),
finalize: () => bus.publish(Command.Event.Updated, {}).pipe(Effect.asVoid),
})
const info = (definition: Definition) =>
Info.make({
+2
View File
@@ -44,6 +44,8 @@ export class GrepInput extends Schema.Class<GrepInput>("FileSystem.GrepInput")({
pattern: Schema.String,
path: Schema.optionalKey(RelativePath),
include: Schema.optionalKey(Schema.String),
literal: Schema.optionalKey(Schema.Boolean),
caseSensitive: Schema.optionalKey(Schema.Boolean),
limit: Schema.optionalKey(PositiveInt),
}) {}
@@ -25,6 +25,7 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/Lo
const layer = Layer.effect(
Service,
Effect.gen(function* () {
let current: readonly string[] = []
const listeners = new Set<(ignore: readonly string[]) => Effect.Effect<void>>()
const state = State.create<Data, Draft>({
name: "location-watcher-policy",
@@ -33,9 +34,11 @@ const layer = Layer.effect(
add: (ignore) => draft.ignore.push(...ignore),
list: () => draft.ignore,
}),
notify: Effect.forEach(listeners, (listener) => listener(current()), { discard: true }),
finalize: (draft) =>
Effect.sync(() => {
current = [...draft.list()]
}).pipe(Effect.andThen(Effect.forEach(listeners, (listener) => listener(current), { discard: true }))),
})
const current = (): readonly string[] => state.get().ignore
const observe = Effect.fn("LocationWatcherPolicy.observe")(function* (
listener: (ignore: readonly string[]) => Effect.Effect<void>,
) {
@@ -53,7 +56,7 @@ const layer = Layer.effect(
return Service.of({
transform: state.transform,
reload: state.reload,
current,
current: () => current,
observe,
})
}),
+1 -1
View File
@@ -74,7 +74,7 @@ export const layer = (options?: Options) =>
draft.available = false
},
}),
notify: bus.publish(Event.Updated, {}).pipe(Effect.asVoid),
finalize: () => bus.publish(Event.Updated, {}).pipe(Effect.asVoid),
})
const source = (value: ReadonlyArray<File> | Instructions.Unavailable | Instructions.Removed) =>
+17 -13
View File
@@ -328,7 +328,7 @@ const layer = Layer.effect(
},
},
}),
notify: bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
finalize: () => bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
})
const createCredential = Effect.fnUntraced(function* (input: Parameters<Credential.Interface["create"]>[0]) {
@@ -400,17 +400,21 @@ const layer = Layer.effect(
}
yield* Effect.gen(function* () {
const persistence = yield* Effect.suspend(() => {
const implementation = state
.get()
.integrations.get(attempt.integrationID)
?.implementations.get(attempt.methodID)
return createCredential({
integrationID: attempt.integrationID,
label: attempt.label ?? implementation?.label?.(exit.value),
value: exit.value,
})
}).pipe(Effect.asVoid, Effect.exit)
const implementation = state
.get()
.integrations.get(attempt.integrationID)
?.implementations.get(attempt.methodID)
const persistence = yield* Effect.sync(() => attempt.label ?? implementation?.label?.(exit.value)).pipe(
Effect.flatMap((label) =>
createCredential({
integrationID: attempt.integrationID,
label,
value: exit.value,
}),
),
Effect.asVoid,
Effect.exit,
)
const settledAt = yield* Clock.currentTimeMillis
const terminal: TerminalAttempt = Exit.isSuccess(persistence)
? {
@@ -428,7 +432,7 @@ const layer = Layer.effect(
}
// Persisting attempts cannot be cancelled, expired, or claimed again.
yield* SynchronizedRef.update(attempts, (current) => new Map(current).set(attemptID, terminal))
yield* persistence
if (Exit.isFailure(persistence)) yield* Effect.failCause(persistence.cause)
}).pipe(Effect.ensuring(close(attempt.scope)))
}, Effect.uninterruptible)
+5 -6
View File
@@ -6,7 +6,7 @@ import { ephemeral } from "@opencode-ai/schema/event"
import type { Session } from "@opencode-ai/schema/session"
import { createHash } from "node:crypto"
import { isDeepStrictEqual } from "node:util"
import { Cause, Context, Effect, Exit, FiberSet, Latch, Layer, Schema, Scope, Semaphore, Stream, Types } from "effect"
import { Cause, Context, Effect, Exit, FiberSet, Latch, Layer, Schema, Scope, Stream, Types } from "effect"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Credential } from "../credential.js"
import { Bus } from "../bus.js"
@@ -615,9 +615,8 @@ export const layer = (options?: Options) =>
let applied: Map<ServerName, Mcp.ServerConfig> | undefined
const overrides = new Map<ServerName, Mcp.ServerConfig | false>()
const reconcileLock = Semaphore.makeUnsafe(1)
const reconcile = Effect.fnUntraced(function* () {
const servers = state.get().servers
const reconcile = Effect.fnUntraced(function* (next: Draft) {
const servers = new Map(next.list())
if (!applied && entries.size === 0) {
for (const [name, server] of servers) {
entries.set(name, {
@@ -678,7 +677,7 @@ export const layer = (options?: Options) =>
Stream.runForEach((event) => Effect.sync(() => fork(reconnect(event.data.integrationID)))),
),
)
const state: State.Interface<Data, Draft> = State.create<Data, Draft>({
const state = State.create<Data, Draft>({
name: "mcp",
initial: () => ({
servers: new Map(
@@ -703,7 +702,7 @@ export const layer = (options?: Options) =>
},
remove: (server) => draft.servers.delete(ServerName.make(server)),
}),
notify: State.reconcile(root, fork, () => reconcileLock.withPermit(reconcile())),
finalize: reconcile,
})
// Suspend so each await sees current entries; a bare Map iterator is exhausted after one run.
+22 -16
View File
@@ -3,7 +3,8 @@ export * as ModelResolver from "./model-resolver.js"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { LanguageModel } from "@opencode-ai/ai"
import { Auth } from "@opencode-ai/ai/route"
import { Context, Effect, Layer, Schema, Struct } from "effect"
import { Context, Effect, Layer, Schema } from "effect"
import { produce } from "immer"
import { AISDK } from "./aisdk.js"
import { AISDKNative } from "./aisdk-native.js"
import { Catalog } from "./catalog.js"
@@ -94,12 +95,11 @@ export const withVariant = (
)
return Effect.succeed(
variant
? {
...model,
settings: Provider.mergeOverlay(model.settings, variant.settings),
headers: Provider.mergeHeaders(model.headers, variant.headers),
body: Provider.mergeOverlay(model.body, variant.body),
}
? produce(model, (draft) => {
draft.settings = Provider.mergeOverlay(draft.settings, variant.settings)
draft.headers = Provider.mergeHeaders(draft.headers, variant.headers)
draft.body = Provider.mergeOverlay(draft.body, variant.body)
})
: model,
)
}
@@ -147,7 +147,10 @@ const resolveCatalogModel = Effect.fn("ModelResolver.resolveCatalogModel")(funct
...configuration,
}) ?? {},
)
return yield* loadAISDK({ ...resolved, settings }).pipe(Effect.mapError(() => unsupported(resolved)))
const runtime = produce(resolved, (draft) => {
draft.settings = settings
})
return yield* loadAISDK(runtime).pipe(Effect.mapError(() => unsupported(resolved)))
}
if (!native) return yield* unsupported(resolved)
@@ -157,7 +160,7 @@ const resolveCatalogModel = Effect.fn("ModelResolver.resolveCatalogModel")(funct
Effect.mapError(() => unsupported(resolved)),
)
const settings = {
...(credential ? Struct.omit(mapped, ["accessToken", "apiKey", "authToken"]) : mapped),
...(credential ? withoutNativeAuthSettings(mapped) : mapped),
...(resolved.canonical === undefined ? {} : { provider: resolved.canonical }),
...nativeCredentialSettings(specifier, credential),
headers: Provider.mergeHeaders(mapping?.headers, resolved.headers),
@@ -179,13 +182,11 @@ const resolveCatalogModel = Effect.fn("ModelResolver.resolveCatalogModel")(funct
function prepareRuntimeModel(model: Info, credential: Credential.Value | undefined) {
if (model.settings?.apiKey !== "" && (credential?.type !== "key" || credential.metadata === undefined)) return model
return {
...model,
...(model.settings?.apiKey === "" ? { settings: Struct.omit(model.settings, ["apiKey"]) } : {}),
...(credential?.type === "key" && credential.metadata !== undefined
? { body: Provider.mergeOverlay(model.body, credential.metadata) }
: {}),
}
return produce(model, (draft) => {
if (draft.settings?.apiKey === "") delete draft.settings.apiKey
if (credential?.type === "key" && credential.metadata !== undefined)
draft.body = Provider.mergeOverlay(draft.body, credential.metadata)
})
}
function validateProviderVariables(
@@ -242,6 +243,11 @@ const nativeCredentialSettings = (specifier: string, credential: Credential.Valu
return { apiKey: credential.access }
}
const withoutNativeAuthSettings = (settings: Record<string, unknown>) => {
const { accessToken: _accessToken, apiKey: _apiKey, authToken: _authToken, ...rest } = settings
return rest
}
const unsupported = (model: Info) =>
new UnsupportedPackageError({
providerID: model.providerID,
+1 -1
View File
@@ -169,7 +169,7 @@ const layer = Layer.effect(
lock.withPermit(
Effect.gen(function* () {
active.clear()
yield* State.shutdown(Scope.close(scope, exit))
yield* State.batch(Scope.close(scope, exit), { flush: false })
}),
)
yield* Effect.addFinalizer(close)
+41 -35
View File
@@ -1,7 +1,7 @@
export * as Reference from "./reference.js"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Context, Effect, Layer, Scope } from "effect"
import { Context, Effect, Layer, Scope, Types } from "effect"
import { Reference } from "@opencode-ai/schema/reference"
import { Global } from "@opencode-ai/util/global"
import { Bus } from "./bus.js"
@@ -25,7 +25,7 @@ export const Info = Reference.Info
export type Info = Reference.Info
type Data = {
sources: Map<string, Source>
sources: Map<string, Types.DeepMutable<Source>>
}
type Draft = {
@@ -47,34 +47,10 @@ const layer = Layer.effect(
const bus = yield* Bus.Service
const cache = yield* RepositoryCache.Service
const scope = yield* Scope.Scope
const list = (): Info[] =>
Array.from(state.get().sources).flatMap(([name, source]) => {
const info = {
name,
source,
...(source.description === undefined ? {} : { description: source.description }),
...(source.hidden === undefined ? {} : { hidden: source.hidden }),
}
if (source.type === "local") return [Info.make({ ...info, path: source.path })]
const repository = Repository.parse(source.repository)
if (!repository || !Repository.isRemote(repository)) return []
if (source.branch) {
try {
Repository.validateBranch(source.branch)
} catch {
return []
}
}
return [
Info.make({
...info,
path: AbsolutePath.make(Repository.cachePath(global.repos, repository, source.branch)),
}),
]
})
const materialized = new Map<string, Info>()
const refresh = Effect.fn("Reference.refresh")(function* () {
yield* Effect.forEach(
list(),
Array.from(materialized.values()),
(reference) =>
Effect.gen(function* () {
if (reference.source.type !== "git") return
@@ -95,14 +71,44 @@ const layer = Layer.effect(
name: "reference",
initial: () => ({ sources: new Map() }),
draft: (draft) => ({
add: (name, source) => draft.sources.set(name, source),
add: (name, source) => draft.sources.set(name, source as Types.DeepMutable<Source>),
remove: (name) => draft.sources.delete(name),
list: () => Array.from(draft.sources),
}),
notify: Effect.gen(function* () {
yield* refresh().pipe(Effect.forkIn(scope))
yield* bus.publish(Reference.Event.Updated, {})
list: () => Array.from(draft.sources.entries()) as [string, Source][],
}),
finalize: (draft) =>
Effect.gen(function* () {
materialized.clear()
for (const [name, source] of draft.list()) {
const info = {
name,
source,
...(source.description === undefined ? {} : { description: source.description }),
...(source.hidden === undefined ? {} : { hidden: source.hidden }),
}
if (source.type === "local") {
materialized.set(name, Info.make({ ...info, path: source.path }))
continue
}
const repository = Repository.parse(source.repository)
if (!repository || !Repository.isRemote(repository)) continue
if (source.branch) {
try {
Repository.validateBranch(source.branch)
} catch {
continue
}
}
materialized.set(
name,
Info.make({
...info,
path: AbsolutePath.make(Repository.cachePath(global.repos, repository, source.branch)),
}),
)
}
yield* refresh().pipe(Effect.forkIn(scope))
yield* bus.publish(Reference.Event.Updated, {})
}),
})
// Check independently of session activity; the shared cache throttles Git work daily.
@@ -112,7 +118,7 @@ const layer = Layer.effect(
transform: state.transform,
reload: state.reload,
list: Effect.fn("Reference.list")(function* () {
return list()
return Array.from(materialized.values())
}),
})
}),
+4
View File
@@ -74,6 +74,8 @@ export interface GrepInput {
readonly pattern: string
readonly file?: string
readonly include?: string
readonly literal?: boolean
readonly caseSensitive?: boolean
readonly limit: number
readonly signal?: AbortSignal
}
@@ -220,6 +222,8 @@ const layer = Layer.effect(
"--json",
"--hidden",
"--no-messages",
...(input.literal ? ["--fixed-strings"] : []),
...(input.caseSensitive === false ? ["--ignore-case"] : []),
...(input.include ? [`--glob=${input.include}`] : []),
"--glob=!**/.git/**",
"--",
+1 -1
View File
@@ -109,7 +109,7 @@ const layer = Layer.effect(
draft.skills.delete(ID.make(id))
},
}),
notify: bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid),
finalize: () => bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid),
})
return Service.of({
+101 -120
View File
@@ -1,9 +1,9 @@
export * as State from "./state.js"
import { Cause, Clock, Context, Deferred, Effect, Exit, Fiber, Scope } from "effect"
import { Clock, Context, Deferred, Effect, Scope, Semaphore } from "effect"
/**
* A synchronous, replayable edit to the current domain state.
* A replayable transform applied to a draft during reload.
*
* Domain drafts expose readable and writable state while preserving concise
* plugin/config code. Transforms synchronously rebuild derived state.
@@ -16,14 +16,13 @@ export interface Registration {
}
/**
* Registers a scoped transform. Reads rebuild by applying every registered transform in order.
* Closing the owning Scope removes the transform and invalidates the current value.
* Registers and applies a scoped transform. Closing the owning Scope removes
* the transform and reloads the materialized state.
*/
export type Transform<DraftApi> = (
transform: TransformCallback<DraftApi>,
) => Effect.Effect<Registration, never, Scope.Scope>
/** Invalidates the current value after captured inputs change and coalesces notifications. */
export type Reload = () => Effect.Effect<void>
export interface Transformable<DraftApi> {
@@ -33,8 +32,8 @@ export interface Transformable<DraftApi> {
type Batch = {
active: boolean
readonly shutdown: boolean
readonly notifications: Set<Effect.Effect<void>>
readonly flush: boolean
readonly reloads: Set<Reload>
}
const CurrentBatch = Context.Reference<Batch | undefined>("@opencode/State/CurrentBatch", {
@@ -42,49 +41,16 @@ const CurrentBatch = Context.Reference<Batch | undefined>("@opencode/State/Curre
})
const reloadDebounce = 500
/** Coalesces notifications until the effect completes. Reads inside stay fresh; nothing is rolled back. */
export function batch<A, E, R>(effect: Effect.Effect<A, E, R>) {
return run(effect, false)
}
/**
* Runs the effect as shutdown: States changed inside it close permanently and never notify again,
* including debounced reloads already waiting.
*/
export function shutdown<A, E, R>(effect: Effect.Effect<A, E, R>) {
return run(effect, true)
}
function run<A, E, R>(effect: Effect.Effect<A, E, R>, shutdown: boolean) {
return Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
const current = yield* CurrentBatch
if (current?.active && !shutdown) return yield* restore(effect)
const batch: Batch = { active: true, shutdown, notifications: new Set() }
const exit = yield* restore(effect.pipe(Effect.provideService(CurrentBatch, batch))).pipe(Effect.exit)
batch.active = false
// A shutdown batch never collects notifications: changed() closes the State instead.
const notifications = yield* Effect.forEach(batch.notifications, (notify) => restore(notify).pipe(Effect.exit))
// Aggregate ordinary failures across domains, while allowing cancellation to stop observer work.
yield* Exit.asVoidAll([exit, ...notifications])
return yield* exit
}),
)
}
/**
* A `notify` that runs resource reconciliation in the owning layer's FiberSet and awaits it, so work
* queued behind the layer's locks is interrupted with the layer. That interruption is not a failure.
*/
export function reconcile(
root: Scope.Scope,
fork: (effect: Effect.Effect<void>) => Fiber.Fiber<void>,
work: () => Effect.Effect<void>,
): Effect.Effect<void> {
/** flush: false is terminal teardown: states whose transforms are removed stop rebuilding, including pending reloads. */
export function batch<A, E, R>(effect: Effect.Effect<A, E, R>, options: { readonly flush?: boolean } = {}) {
return Effect.gen(function* () {
const exit = yield* Fiber.await(fork(work()))
if (Exit.isFailure(exit) && root.state._tag === "Closed" && Cause.hasInterruptsOnly(exit.cause)) return
yield* exit
const current = yield* CurrentBatch
if (current?.active && options.flush !== false) return yield* effect
const batch: Batch = { active: true, flush: options.flush !== false, reloads: new Set() }
const exit = yield* effect.pipe(Effect.provideService(CurrentBatch, batch), Effect.exit)
batch.active = false
if (batch.flush) yield* Effect.forEach(batch.reloads, (reload) => reload(), { discard: true })
return yield* exit
})
}
@@ -95,113 +61,128 @@ export const inherit = Effect.fnUntraced(function* () {
export interface Options<State, DraftApi> {
readonly name?: string
/** Creates the empty base value for every rebuild. */
/** Creates the base value for initial state and every scoped-transform reload. */
readonly initial: () => State
/** Wraps mutable state in a domain-specific draft API. */
readonly draft: MakeDraft<State, DraftApi>
/**
* Observes current state outside the read path. Batched changes notify at
* batch completion; reloads debounce notifications. Resource reconciliation
* owns its execution scope and coordination.
* Runs after the rebuilt state becomes visible. Update events published here
* act as read barriers: subscribers refetching on the event observe the
* committed state.
*/
readonly notify?: Effect.Effect<void>
readonly finalize?: (draft: DraftApi) => Effect.Effect<void>
}
export interface Interface<State, DraftApi> extends Transformable<DraftApi> {
/**
* Rebuilds synchronously when transforms changed since the last read. Each rebuild produces a new
* value and never touches earlier ones, so callers may retain what they read.
*/
readonly get: () => State
}
export function create<State, DraftApi>(options: Options<State, DraftApi>): Interface<State, DraftApi> {
let state = options.initial()
const transforms: { run: TransformCallback<DraftApi> }[] = []
let dirty = false
let transforms: { run: TransformCallback<DraftApi> }[] = []
let generation = 0
let requestedAt = 0
let running = false
let closed = false
let pending: Deferred.Deferred<void> | undefined
let waiters: { generation: number; done: Deferred.Deferred<void> }[] = []
const semaphore = Semaphore.makeUnsafe(1)
const get = () => {
if (closed || !dirty) return state
const next = options.initial()
const draft = options.draft(next)
for (const transform of transforms) transform.run(draft)
// Only a complete fold becomes visible; a throwing callback leaves the previous value and stays dirty.
const commit = Effect.fn("State.commit")(function* (next: State) {
state = next
dirty = false
return state
}
if (options.finalize) yield* options.finalize(options.draft(next))
})
// One stable value per State, so a batch's notification Set holds it at most once.
const notify: Effect.Effect<void> = Effect.gen(function* () {
const materialize = Effect.fnUntraced(function* () {
if (closed) return
get()
if (options.notify) yield* options.notify
}).pipe(Effect.withSpan("State.notify"))
const next = options.initial()
const api = options.draft(next)
for (const transform of transforms) {
yield* Effect.sync(() => {
transform.run(api)
})
}
yield* commit(next)
})
const changed = (debounce: boolean) =>
Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
if (closed) return
dirty = true
const batch = yield* CurrentBatch
if (batch?.active) {
if (batch.shutdown) {
closed = true
return
}
batch.notifications.add(notify)
return
}
if (!debounce) {
yield* restore(notify)
return
}
const materializeReload = () => semaphore.withPermit(materialize())
const clock = yield* Clock.Clock
requestedAt = clock.currentTimeMillisUnsafe()
const done = pending ?? Deferred.makeUnsafe<void>()
if (!pending) {
pending = done
yield* Effect.gen(function* () {
do {
const remaining = requestedAt + reloadDebounce - clock.currentTimeMillisUnsafe()
if (remaining > 0) yield* Effect.sleep(remaining)
} while (clock.currentTimeMillisUnsafe() < requestedAt + reloadDebounce)
// Observers can request and await another reload without joining their own notification.
pending = undefined
yield* notify.pipe(Deferred.into(done))
}).pipe(Effect.forkDetach)
}
yield* restore(Deferred.await(done))
}),
)
const rebuild = (): Effect.Effect<void> =>
Effect.gen(function* () {
const clock = yield* Clock.Clock
const remaining = requestedAt + reloadDebounce - clock.currentTimeMillisUnsafe()
if (remaining > 0) yield* Effect.sleep(remaining)
if (clock.currentTimeMillisUnsafe() < requestedAt + reloadDebounce) return yield* rebuild()
const target = generation
const exit = yield* materializeReload().pipe(Effect.exit)
const completed = waiters.filter((waiter) => waiter.generation <= target)
waiters = waiters.filter((waiter) => waiter.generation > target)
yield* Effect.forEach(completed, (waiter) => Deferred.done(waiter.done, exit), {
concurrency: "unbounded",
discard: true,
})
if (generation > target) return yield* rebuild()
running = false
})
const reload = Effect.fnUntraced(function* () {
if (closed) return
const done = Deferred.makeUnsafe<void>()
const clock = yield* Clock.Clock
generation++
requestedAt = clock.currentTimeMillisUnsafe()
waiters.push({ generation, done })
if (!running) {
running = true
yield* rebuild().pipe(Effect.forkDetach)
}
yield* Deferred.await(done)
})
return {
get,
get: () => state,
transform: Effect.fn("State.transform")(function* (update) {
yield* Effect.annotateCurrentSpan("state", options.name ?? "anonymous")
const scope = yield* Scope.Scope
return yield* Effect.uninterruptible(
Effect.gen(function* () {
const transform = { run: update }
let active = true
const dispose = Effect.uninterruptible(
Effect.suspend(() => {
const index = transforms.indexOf(transform)
if (index < 0) return Effect.void
transforms.splice(index, 1)
return changed(false)
semaphore.withPermit(
Effect.suspend(() => {
if (!active) return Effect.void
active = false
transforms = transforms.filter((item) => item !== transform)
return Effect.gen(function* () {
const batch = yield* CurrentBatch
if (batch?.active) {
// Detached debounced reloads must also stay quiet after teardown.
if (!batch.flush) {
closed = true
return
}
batch.reloads.add(materializeReload)
return
}
yield* materialize()
})
}),
),
)
yield* semaphore.withPermit(
Effect.sync(() => {
transforms = [...transforms, transform]
}),
)
transforms.push(transform)
yield* Scope.addFinalizer(scope, dispose)
yield* changed(false)
const batch = yield* CurrentBatch
if (batch?.active) batch.reloads.add(materializeReload)
else yield* materializeReload()
return { dispose }
}),
)
}),
reload: () => changed(true),
reload,
}
}
+1 -3
View File
@@ -196,8 +196,7 @@ const layer = Layer.effect(
draft.tools.delete(id)
},
}),
// Read errors when the notification runs, not when the State is created.
notify: Effect.suspend(() =>
finalize: () =>
Effect.forEach(
state.get().errors,
({ kind, name, namespace, error }) =>
@@ -208,7 +207,6 @@ const layer = Layer.effect(
}),
{ discard: true },
),
),
})
return Service.of({
+12 -2
View File
@@ -18,7 +18,7 @@ export const Input = Schema.Struct({
pattern: FileSystem.GrepInput.fields.pattern
.check(Schema.isMinLength(1, { message: "Pattern must not be empty" }))
.annotate({
description: "Regular expression to search for in file contents (ripgrep syntax)",
description: "Regular expression or literal text to match in file contents.",
}),
path: Schema.optionalKey(RelativePath).annotate({
description: "File or directory to search. Defaults to the current working directory.",
@@ -26,6 +26,12 @@ export const Input = Schema.Struct({
include: FileSystem.GrepInput.fields.include.annotate({
description: 'Glob pattern to filter files (for example, "*.js" or "*.{ts,tsx}")',
}),
literal: FileSystem.GrepInput.fields.literal.annotate({
description: "Treat `pattern` as exact text instead of a regular expression (default: false).",
}),
caseSensitive: FileSystem.GrepInput.fields.caseSensitive.annotate({
description: "Use case-sensitive matching (default: true).",
}),
limit: FileSystem.GrepInput.fields.limit.annotate({
description: `Maximum number of matching lines to return (default: ${FileSystem.DEFAULT_SEARCH_LIMIT})`,
}),
@@ -70,7 +76,7 @@ export const Plugin = {
name,
options: { codemode: false },
description:
"Search file contents using regular expressions. Use it to locate specific code, symbols, or text patterns, and narrow searches with `path` or `include`. Returns matching file paths, line numbers, and line previews.",
"Search file contents using ripgrep's regular expression syntax or literal text matching. Use it to locate specific code, symbols, or text patterns, and narrow searches with `path` or `include`. Returns matching file paths, line numbers, and line previews.",
input: Input,
output: Output,
execute: (input, context) =>
@@ -92,6 +98,8 @@ export const Plugin = {
root: ".",
path: input.path,
include: input.include,
literal: input.literal,
caseSensitive: input.caseSensitive,
limit: input.limit,
},
sessionID: context.sessionID,
@@ -112,6 +120,8 @@ export const Plugin = {
pattern: input.pattern,
file: type === "file" ? path.basename(root) : undefined,
include: input.include,
literal: input.literal,
caseSensitive: input.caseSensitive,
limit: limit + 1,
})
.pipe(
+10 -29
View File
@@ -1,7 +1,7 @@
export * as Vcs from "./vcs.js"
import path from "path"
import { Cause, Context, Effect, FiberSet, Layer, Schema, Semaphore, Stream } from "effect"
import { Cause, Context, Effect, Layer, Schema, Stream } from "effect"
import type { VcsDefinition, VcsDraft } from "@opencode-ai/plugin/effect/vcs"
import { FileDiff } from "@opencode-ai/schema/file-diff"
import { FileSystem } from "@opencode-ai/schema/filesystem"
@@ -55,11 +55,8 @@ const layer = Layer.effect(
const fs = yield* FSUtil.Service
const location = yield* Location.Service
const bus = yield* Bus.Service
const root = yield* Effect.scope
const fork = yield* FiberSet.makeRuntime<never, void, never>()
const vcs = location.vcs
const current: { info: Info } = { info: { branch: {} } }
const refreshLock = Semaphore.makeUnsafe(1)
const scope = {
directory: location.directory,
worktree: location.project.directory,
@@ -81,7 +78,7 @@ const layer = Layer.effect(
set: (selection) => (draft.selection = selection),
},
}),
notify: State.reconcile(root, fork, () => refresh()),
finalize: () => refresh(),
})
const selected = () => {
const value = state.get()
@@ -120,23 +117,13 @@ const layer = Layer.effect(
}),
)
const refresh = Effect.fn("Vcs.refresh")(function* () {
const changed = yield* Effect.gen(function* () {
const provider = selected()
const next: Info = provider
? yield* protect(provider, "info", provider.info(scope).pipe(Effect.flatMap(decodeInfo)), { branch: {} })
: { branch: {} }
const changed = current.info.branch.current !== next.branch.current
current.info = next
return changed
}).pipe(Semaphore.withPermit(refreshLock))
if (!changed) return
// Legacy listeners can publish nested updates before streams and SSE receive
// this event. Re-announce the latest branch if publication was overtaken.
while (true) {
const branch = current.info.branch.current
yield* bus.publish(VcsEvent.BranchUpdated, { branch })
if (branch === current.info.branch.current) return
}
const provider = selected()
const next: Info = provider
? yield* protect(provider, "info", provider.info(scope).pipe(Effect.flatMap(decodeInfo)), { branch: {} })
: { branch: {} }
const changed = current.info.branch.current !== next.branch.current
current.info = next
if (changed) yield* bus.publish(VcsEvent.BranchUpdated, { branch: next.branch.current })
})
if (vcs) {
@@ -148,13 +135,7 @@ const layer = Layer.effect(
yield* bus.subscribe(FileSystem.Event.Changed).pipe(
Stream.filter((event) => isBranchMetadata(event.data.file)),
Stream.runForEach((event) =>
refresh().pipe(
Effect.catchCauseIf(
(cause) => !Cause.hasInterrupts(cause),
(cause) => Effect.logWarning("vcs refresh failed", { file: event.data.file, cause }),
),
Effect.withSpan("Vcs.refreshBranch", { attributes: { file: event.data.file } }),
),
refresh().pipe(Effect.withSpan("Vcs.refreshBranch", { attributes: { file: event.data.file } })),
),
Effect.forkScoped({ startImmediately: true }),
)
+1 -1
View File
@@ -88,7 +88,7 @@ const layer = Layer.effect(
set: (selection) => (draft.selection = selection),
},
}),
notify: bus.publish(WebSearch.Event.Updated, {}).pipe(Effect.asVoid),
finalize: () => bus.publish(WebSearch.Event.Updated, {}).pipe(Effect.asVoid),
})
const requireProvider = (providers: Map<ID, ProviderImplementation>, providerID: ID) => {
-49
View File
@@ -1,6 +1,4 @@
import { describe, expect } from "bun:test"
import { LanguageModel } from "@opencode-ai/ai"
import { OpenAIChat } from "@opencode-ai/ai/protocols"
import { Effect, Fiber, Layer, Stream } from "effect"
import { TestClock } from "effect/testing"
import { Catalog } from "@opencode-ai/core/catalog"
@@ -11,7 +9,6 @@ import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Bus } from "@opencode-ai/core/bus"
import { Location } from "@opencode-ai/core/location"
import { Model } from "@opencode-ai/core/model"
import { ModelResolver } from "@opencode-ai/core/model-resolver"
import { Provider } from "@opencode-ai/core/provider"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { location } from "./fixture/location"
@@ -33,52 +30,6 @@ const catalogLayer = AppNodeBuilder.build(
const it = testEffect(catalogLayer)
describe("Catalog", () => {
;["variant", "empty-key", "metadata", "aisdk"].forEach((path) =>
it.effect(`keeps nested catalog values editable after ${path} model resolution`, () =>
Effect.gen(function* () {
const catalog = yield* Catalog.Service
const providerID = Provider.ID.make("resolve-fixture")
const modelID = Model.ID.make("fixture-model")
yield* catalog.transform((draft) =>
draft.model.update(providerID, modelID, (model) => {
model.package = path === "aisdk" ? Provider.aisdk("@ai-sdk/fixture") : "@opencode-ai/ai/providers/openai"
model.settings = {
apiKey: path === "empty-key" ? "" : "fixture-key",
baseURL: "https://fixture.example/v1",
}
model.variants = [{ id: Model.VariantID.make("high"), body: { reasoning: { effort: "high" } } }]
}),
)
const selected = required(yield* catalog.model.get(providerID, modelID))
if (path === "variant") yield* ModelResolver.withVariant(selected, Model.VariantID.make("high"))
if (path !== "variant")
yield* ModelResolver.fromCatalogModel(
selected,
path === "metadata"
? Credential.Key.make({ type: "key", key: "fixture-key", metadata: { tenant: "fixture" } })
: undefined,
{
loadAISDK: () =>
Effect.succeed(LanguageModel.make({ id: modelID, provider: providerID, route: OpenAIChat.route })),
},
)
yield* catalog.transform((draft) =>
draft.model.update(providerID, modelID, (model) => {
model.limit.context = 100_000
model.capabilities.tools = false
model.variants.push({ id: Model.VariantID.make("other") })
}),
)
expect(required(yield* catalog.model.get(providerID, modelID))).toMatchObject({
limit: { context: 100_000 },
capabilities: { tools: false },
variants: [{ id: "high" }, { id: "other" }],
})
}),
),
)
it.effect("publishes an updated event after catalog changes", () =>
Effect.gen(function* () {
const catalog = yield* Catalog.Service
@@ -1,70 +0,0 @@
import { describe, expect } from "bun:test"
import { Effect, Scope } from "effect"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LocationWatcherPolicy } from "@opencode-ai/core/filesystem/location-watcher-policy"
import { State } from "@opencode-ai/core/state"
import { testEffect } from "../lib/effect"
const it = testEffect(AppNodeBuilder.build(LocationWatcherPolicy.node))
describe("LocationWatcherPolicy", () => {
it.effect("reads batched registrations and disposals before notifying observers", () =>
Effect.gen(function* () {
const policy = yield* LocationWatcherPolicy.Service
const observed: string[][] = []
yield* policy.observe((ignore) =>
Effect.sync(() => {
expect(policy.current()).toEqual(ignore)
observed.push([...ignore])
}),
)
yield* State.batch(
Effect.gen(function* () {
yield* policy.transform((draft) => draft.add(["node_modules"]))
expect(policy.current()).toEqual(["node_modules"])
const overlay = yield* policy.transform((draft) => draft.add([".git"]))
expect(policy.current()).toEqual(["node_modules", ".git"])
expect(observed).toEqual([])
yield* overlay.dispose
expect(policy.current()).toEqual(["node_modules"])
expect(observed).toEqual([])
}),
)
expect(observed).toEqual([["node_modules"]])
}),
)
it.effect("passes the latest policy to later observers after a reentrant registration", () =>
Effect.gen(function* () {
const policy = yield* LocationWatcherPolicy.Service
const scope = yield* Scope.Scope
const observed: string[][] = []
let reentered = false
yield* policy.observe(() =>
Effect.gen(function* () {
if (reentered) return
reentered = true
yield* policy.transform((draft) => draft.add([".git"])).pipe(Scope.provide(scope))
expect(policy.current()).toEqual(["node_modules", ".git"])
}),
)
yield* policy.observe((ignore) =>
Effect.sync(() => {
expect(policy.current()).toEqual(ignore)
observed.push([...ignore])
}),
)
yield* policy.transform((draft) => draft.add(["node_modules"]))
expect(policy.current()).toEqual(["node_modules", ".git"])
expect(observed).toEqual([
["node_modules", ".git"],
["node_modules", ".git"],
])
}),
)
})
-57
View File
@@ -7,7 +7,6 @@ import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Bus } from "@opencode-ai/core/bus"
import { Integration } from "@opencode-ai/core/integration"
import { State } from "@opencode-ai/core/state"
import { testEffect } from "./lib/effect"
const it = testEffect(AppNodeBuilder.build(LayerNode.group([Integration.node, Credential.node, Bus.node])))
@@ -430,62 +429,6 @@ describe("Integration", () => {
}),
)
it.effect("fails and closes OAuth attempts when a pending transform throws during persistence", () =>
Effect.gen(function* () {
const integrations = yield* Integration.Service
const credentials = yield* Credential.Service
const integrationID = Integration.ID.make("replay-fixture")
const methodID = Integration.MethodID.make("code")
let closed = false
yield* integrations.transform((editor) =>
editor.method.update({
integrationID,
method: { id: methodID, type: "oauth", label: "Fixture" },
authorize: () =>
Effect.addFinalizer(() => Effect.sync(() => (closed = true))).pipe(
Effect.as({
mode: "code" as const,
url: "https://example.com/authorize",
instructions: "Enter the fixture code",
callback: () =>
Effect.succeed(
Credential.OAuth.make({
type: "oauth",
methodID,
access: "fixture-access",
refresh: "fixture-refresh",
expires: 1,
}),
),
}),
),
}),
)
const attempt = yield* integrations.oauth.connect({ integrationID, methodID })
yield* State.batch(
Effect.gen(function* () {
const failure = new Error("integration transform failed")
yield* integrations.transform(() => {
throw failure
})
const exit = yield* integrations.oauth
.complete({ integrationID, attemptID: attempt.attemptID, code: "fixture-code" })
.pipe(Effect.exit)
expect(Exit.isFailure(exit) && Cause.squash(exit.cause)).toBe(failure)
expect(yield* integrations.oauth.status({ integrationID, attemptID: attempt.attemptID })).toEqual({
status: "failed",
message: failure.message,
time: attempt.time,
})
expect(closed).toBe(true)
expect(yield* credentials.list(integrationID)).toEqual([])
}).pipe(Effect.scoped),
)
}),
)
it.effect("expires abandoned OAuth attempts", () =>
Effect.gen(function* () {
const integrations = yield* Integration.Service
+4 -181
View File
@@ -34,24 +34,9 @@ import { McpStdio } from "@opencode-ai/core/mcp/stdio"
import { Permission } from "@opencode-ai/core/permission"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { Session } from "@opencode-ai/core/session"
import { State } from "@opencode-ai/core/state"
import { McpTool } from "@opencode-ai/core/tool/mcp"
import { Tool } from "@opencode-ai/core/tool"
import {
Context,
Deferred,
Effect,
Exit,
Fiber,
Layer,
PubSub,
Ref,
Schedule,
Schema,
Scope,
Sink,
Stream,
} from "effect"
import { Deferred, Effect, Exit, Fiber, Layer, PubSub, Ref, Schedule, Schema, Sink, Stream } from "effect"
import { TestClock } from "effect/testing"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { ExitCode, makeHandle, ProcessId } from "effect/unstable/process/ChildProcessSpawner"
@@ -84,7 +69,7 @@ function resourceServer(
listChanged?: boolean
emptyElicitation?: boolean
urlElicitation?: boolean
respond?: (request: Request) => Response | undefined | Promise<Response | undefined>
respond?: (request: Request) => Response | undefined
} = {},
) {
return Effect.acquireRelease(
@@ -193,7 +178,7 @@ function resourceServer(
if (typeof body === "object" && body !== null && "method" in body && body.method === "initialize") {
state.initializations += 1
}
return (await input.respond?.(request)) ?? transport.handleRequest(request)
return input.respond?.(request) ?? transport.handleRequest(request)
},
})
return {
@@ -1346,9 +1331,6 @@ testEffect(resourceMcpLayer(new ConfigMCP.Local({ type: "local", command: ["unus
)
expect(yield* service.tools()).toHaveLength(2)
yield* service.transform((draft) => draft.update("dynamic", (server) => (server.codemode = false)))
expect((yield* service.tools()).map((tool) => tool.codemode)).toEqual([false, false])
const settings = { disabled: true }
yield* service.transform((draft) => {
draft.update("dynamic", (server) => {
@@ -1433,7 +1415,7 @@ test("isolates nested configured MCP mutations and reconciles them", async () =>
expect(published.filter((type) => type === McpEvent.StatusChanged.type)).toHaveLength(1)
yield* service.transform((draft) =>
draft.update("resources", (server) => {
if (server.type === "remote" && server.headers) server.headers.Authorization = "transformed"
if (server.type === "remote") server.headers = { Authorization: "transformed" }
}),
)
@@ -1444,41 +1426,6 @@ test("isolates nested configured MCP mutations and reconciles them", async () =>
)
})
testEffect(Layer.empty).live("batches MCP transforms without connecting intermediate configurations", () =>
Effect.gen(function* () {
const server = yield* resourceServer()
yield* Effect.gen(function* () {
const service = yield* Mcp.Service
const registrations = yield* State.batch(
Effect.gen(function* () {
const added = yield* service.transform((draft) =>
draft.set("dynamic", {
type: "remote",
url: server.url,
oauth: false,
}),
)
expect((yield* service.servers()).some((server) => server.name === "dynamic")).toBe(false)
const disabled = yield* service.transform((draft) =>
draft.update("dynamic", (config) => (config.disabled = true)),
)
return [added, disabled]
}),
)
expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({
status: "disabled",
})
expect(yield* service.tools()).toEqual([])
yield* State.batch(Effect.forEach(registrations, (registration) => registration.dispose))
expect((yield* service.servers()).some((server) => server.name === "dynamic")).toBe(false)
expect(server.state.initializations).toBe(0)
}).pipe(
Effect.provide(resourceMcpLayer(new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }))),
)
}),
)
test("reconciles only changed MCP server config", async () => {
await Effect.runPromise(
Effect.scoped(
@@ -1569,130 +1516,6 @@ test("reconciles only changed MCP server config", async () => {
)
})
testEffect(Layer.empty).live("keeps MCP config snapshots stable during an in-flight replacement", () =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const accepted = yield* Deferred.make<void>()
const server = yield* resourceServer({
respond: (request) =>
request.method !== "POST"
? undefined
: Effect.runPromise(
Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release)), Effect.as(undefined)),
),
})
yield* Effect.gen(function* () {
const service = yield* Mcp.Service
const replacing = yield* service
.transform((draft) => draft.update("resources", (config) => (config.disabled = false)))
.pipe(Effect.forkScoped({ startImmediately: true }))
yield* Deferred.await(started)
const restoring = yield* State.batch(
Effect.gen(function* () {
yield* service.transform((draft) => draft.update("resources", (config) => (config.disabled = true)))
yield* Deferred.succeed(accepted, undefined)
}),
).pipe(Effect.forkScoped({ startImmediately: true }))
yield* Deferred.await(accepted)
expect((yield* service.servers())[0]?.status).toEqual({ status: "pending" })
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(replacing)
yield* Fiber.join(restoring)
expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" })
expect(yield* service.tools()).toEqual([])
expect(server.state.initializations).toBe(1)
}).pipe(
Effect.ensuring(Deferred.succeed(release, undefined)),
Effect.provide(
resourceMcpLayer(new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, disabled: true })),
),
)
}),
)
const shutdownIt = testEffect(
AppNodeBuilder.build(
LayerNode.group([Bus.node, Integration.node, Credential.node, Form.node, Environment.node, Location.node]),
[
Location.node.replace(
Layer.succeed(
Location.Service,
Location.Service.of(location({ directory: AbsolutePath.make(import.meta.dir) })),
),
),
Environment.node.replace(hostEnvironmentLayer),
],
),
)
shutdownIt.effect("discards in-flight and queued MCP notifications after its layer closes", () =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const entered = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const root = yield* Scope.make()
yield* Effect.addFinalizer(() =>
Deferred.succeed(release, undefined).pipe(
Effect.andThen(State.shutdown(Scope.close(root, Exit.void))),
Effect.andThen(TestClock.adjust("500 millis")),
),
)
const context = yield* Layer.buildWithScope(Mcp.layer(), root)
const service = Context.get(context, Mcp.Service)
const observed: string[] = []
let block = false
yield* Effect.acquireRelease(
bus.listen((event) =>
Effect.gen(function* () {
if (event.type !== McpEvent.StatusChanged.type) return
observed.push(Schema.decodeUnknownSync(McpEvent.StatusChanged.data)(event.data).server)
if (!block) return
block = false
yield* Deferred.succeed(entered, undefined)
yield* Deferred.await(release)
}),
),
(unsubscribe) => unsubscribe,
)
const source = { url: "https://example.com/initial", added: false }
yield* service
.transform((draft) => {
draft.set("fixture", { type: "remote", url: source.url, oauth: false, disabled: true })
if (source.added) draft.set("queued", { type: "local", command: ["unused"], disabled: true })
})
.pipe(Scope.provide(root))
block = true
source.url = "https://example.com/first"
source.added = true
const first = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* TestClock.adjust("500 millis")
yield* Deferred.await(entered)
source.url = "https://example.com/second"
const second = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* TestClock.adjust("500 millis")
const shutdown = yield* State.shutdown(Scope.close(root, Exit.void)).pipe(
Effect.forkChild({ startImmediately: true }),
)
yield* TestClock.adjust("1 millis")
expect(shutdown.pollUnsafe()).toBeDefined()
expect(first.pollUnsafe()).toBeDefined()
expect(second.pollUnsafe()).toBeDefined()
expect(yield* Deferred.isDone(release)).toBe(false)
yield* Fiber.join(shutdown)
observed.length = 0
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(first)
yield* Fiber.join(second)
expect(observed).toEqual([])
expect((yield* service.servers()).map((server) => server.name)).toEqual([Mcp.ServerName.make("fixture")])
}),
)
test("serializes concurrent MCP lifecycle operations", async () => {
await Effect.runPromise(
Effect.scoped(
+1 -69
View File
@@ -1,10 +1,7 @@
import { expect } from "bun:test"
import path from "path"
import { Clock, Effect } from "effect"
import { TestClock } from "effect/testing"
import { Effect } from "effect"
import { Command } from "@opencode-ai/core/command"
import { Credential } from "@opencode-ai/core/credential"
import { Integration } from "@opencode-ai/core/integration"
import { Plugin } from "@opencode-ai/core/plugin"
import { PluginModule } from "@opencode-ai/core/plugin/module"
import { Session } from "@opencode-ai/schema/session"
@@ -122,68 +119,3 @@ it.effect("reloading a plugin replaces its command implementation", () =>
expect(output).toEqual(["before", "after"])
}),
)
it.effect("refreshes expired OAuth credentials through the context during activation", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const credentials = yield* Credential.Service
const integrations = yield* Integration.Service
const clock = yield* Clock.Clock
const integrationID = Integration.ID.make("refresh-fixture")
const methodID = Integration.MethodID.make("oauth")
const expired = Credential.OAuth.make({
type: "oauth",
methodID,
access: "expired-access",
refresh: "fixture-refresh",
expires: (yield* Clock.currentTimeMillis) + 60_000,
})
const stored = yield* credentials.create({ integrationID, label: "Fixture", value: expired })
yield* TestClock.adjust("2 minutes")
const refreshed = Credential.OAuth.make({
...expired,
access: "fresh-access",
refresh: "rotated-refresh",
expires: (yield* Clock.currentTimeMillis) + 3_600_000,
})
const refreshes: Credential.OAuth[] = []
const resolved: Array<Credential.Value | undefined> = []
yield* plugins.activate([
{
id: "oauth-refresh",
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
yield* ctx.integration.transform((draft) =>
draft.method.update({
integrationID,
method: { id: methodID, type: "oauth", label: "Fixture" },
authorize: () => Effect.die("unexpected authorization"),
refresh: (value) =>
Effect.sync(() => {
refreshes.push(value)
return refreshed
}),
}),
)
// The method registered above must be readable before the activation batch ends.
const connection = yield* ctx.integration.connection.active(integrationID)
if (!connection) return yield* Effect.die("fixture connection not found")
resolved.push(yield* ctx.integration.connection.resolve(connection).pipe(Effect.orDie))
}).pipe(
// Plugin activation isolates ambient services, including the test clock.
Effect.provideService(Clock.Clock, clock),
),
},
])
expect(yield* plugins.list()).toMatchObject([{ id: "oauth-refresh", state: { status: "active" } }])
expect(resolved).toEqual([refreshed])
expect((yield* credentials.get(stored.id))?.value).toEqual(refreshed)
expect(yield* integrations.connection.resolve({ type: "credential", id: stored.id, label: stored.label })).toEqual(
refreshed,
)
expect(refreshes).toEqual([expired])
}),
)
+2 -136
View File
@@ -1,10 +1,7 @@
import { describe, expect } from "bun:test"
import { Deferred, Effect, Exit, Layer, Scope } from "effect"
import { Effect, Exit, Layer, Scope } from "effect"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { Bus } from "@opencode-ai/core/bus"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { State } from "@opencode-ai/core/state"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { Global } from "@opencode-ai/util/global"
import { Reference } from "@opencode-ai/core/reference"
import { Repository } from "@opencode-ai/core/repository"
@@ -14,140 +11,9 @@ import { it } from "./lib/effect"
const cache = Layer.mock(RepositoryCache.Service, {
ensure: () => Effect.die("unexpected Git materialization"),
})
const referenceLayer = AppNodeBuilder.build(LayerNode.group([Reference.node, Bus.node]), [
RepositoryCache.node.replace(cache),
])
const referenceLayer = AppNodeBuilder.build(Reference.node, [RepositoryCache.node.replace(cache)])
describe("Reference", () => {
it.effect("reads batched references before cache work and update events", () => {
const operations: RepositoryCache.EnsureInput[] = []
const started = Deferred.makeUnsafe<void>()
const release = Deferred.makeUnsafe<void>()
const cache = Layer.succeed(RepositoryCache.Service, {
ensure: (input) =>
Effect.gen(function* () {
operations.push(input)
yield* Deferred.succeed(started, undefined)
yield* Deferred.await(release)
return {
repository: input.reference.label,
host: input.reference.host,
remote: input.reference.remote,
localPath: Repository.cachePath(Global.Path.repos, input.reference, input.branch),
status: "cached",
} satisfies RepositoryCache.Result
}),
})
const referenceLayer = AppNodeBuilder.build(LayerNode.group([Reference.node, Bus.node]), [
RepositoryCache.node.replace(cache),
])
return Effect.gen(function* () {
const references = yield* Reference.Service
const bus = yield* Bus.Service
const observed: string[][] = []
yield* Effect.acquireRelease(
bus.listen((event) =>
event.type === Reference.Event.Updated.type
? references.list().pipe(
Effect.map((infos) => {
observed.push(infos.map((info) => info.name))
}),
)
: Effect.void,
),
(unsubscribe) => unsubscribe,
)
yield* State.batch(
Effect.gen(function* () {
yield* references.transform((draft) =>
draft.add("docs", Reference.LocalSource.make({ type: "local", path: AbsolutePath.make("/docs") })),
)
expect((yield* references.list()).map((info) => info.name)).toEqual(["docs"])
yield* references.transform((draft) => {
draft.add(
"sdk",
Reference.GitSource.make({
type: "git",
repository: "owner/repo",
branch: "feature/docs",
description: "SDK documentation",
hidden: true,
}),
)
draft.add("invalid", Reference.GitSource.make({ type: "git", repository: "invalid" }))
draft.add(
"invalid-branch",
Reference.GitSource.make({ type: "git", repository: "owner/repo", branch: "../escape" }),
)
draft.add("file", Reference.GitSource.make({ type: "git", repository: "file:///docs" }))
})
const infos = yield* references.list()
expect(infos.map((info) => info.name)).toEqual(["docs", "sdk"])
expect(infos[1]).toMatchObject({
path: Repository.cachePath(Global.Path.repos, Repository.parseRemote("owner/repo"), "feature/docs"),
description: "SDK documentation",
hidden: true,
})
yield* Effect.yieldNow
expect(operations).toEqual([])
expect(observed).toEqual([])
}),
)
expect(observed).toEqual([["docs", "sdk"]])
yield* Deferred.await(started)
expect(operations).toEqual([
{ reference: Repository.parseRemote("owner/repo"), branch: "feature/docs", refresh: "daily" },
])
expect((yield* references.list()).map((info) => info.name)).toEqual(["docs", "sdk"])
yield* Effect.yieldNow
expect(operations).toHaveLength(1)
expect(observed).toHaveLength(1)
yield* Deferred.succeed(release, undefined)
}).pipe(Effect.scoped, Effect.provide(referenceLayer))
})
it.effect("lets update listeners replace references and refetch current info", () =>
Effect.gen(function* () {
const references = yield* Reference.Service
const bus = yield* Bus.Service
const scope = yield* Scope.Scope
const observed: string[][] = []
let reentered = false
const first = yield* bus.listen((event) =>
Effect.gen(function* () {
if (event.type !== Reference.Event.Updated.type || reentered) return
reentered = true
yield* references
.transform((draft) =>
draft.add("docs", Reference.LocalSource.make({ type: "local", path: AbsolutePath.make("/new") })),
)
.pipe(Scope.provide(scope))
}),
)
const second = yield* bus.listen((event) =>
event.type === Reference.Event.Updated.type
? references.list().pipe(
Effect.map((infos) => {
observed.push(infos.map((info) => info.path))
}),
)
: Effect.void,
)
yield* Effect.addFinalizer(() => first.pipe(Effect.andThen(second)))
yield* references.transform((draft) =>
draft.add("docs", Reference.LocalSource.make({ type: "local", path: AbsolutePath.make("/old") })),
)
expect((yield* references.list()).map((info) => info.path)).toEqual([AbsolutePath.make("/new")])
expect(observed).toEqual([["/new"], ["/new"]])
}).pipe(Effect.scoped, Effect.provide(referenceLayer)),
)
it.effect("registers normalized sources for the owning scope", () =>
Effect.gen(function* () {
const references = yield* Reference.Service
-207
View File
@@ -1,207 +0,0 @@
import { describe, expect, test } from "bun:test"
import { State } from "@opencode-ai/core/state"
import { Effect } from "effect"
import { FastCheck } from "effect/testing"
type Operation = { multiply: number; add: number }
type Value = { value: number; order: number[] }
const operation = FastCheck.record({
multiply: FastCheck.constantFrom(-3, -2, 2, 3),
add: FastCheck.integer({ min: -9, max: 9 }),
})
const source = FastCheck.integer({ min: -100, max: 100 })
const target = FastCheck.integer({ min: 0, max: 1 })
const command = FastCheck.oneof(
{
weight: 4,
arbitrary: FastCheck.record({ type: FastCheck.constant("append"), target, callback: FastCheck.nat(5) }),
},
{ weight: 3, arbitrary: FastCheck.record({ type: FastCheck.constant("read"), target }) },
{
weight: 2,
arbitrary: FastCheck.record({ type: FastCheck.constant("dispose"), target, registration: FastCheck.nat(100) }),
},
{ weight: 2, arbitrary: FastCheck.record({ type: FastCheck.constant("reload"), target, source }) },
)
// Affine transforms do not generally commute; the modulus keeps long traces exact.
function apply(value: number, operation: Operation) {
return (value * operation.multiply + operation.add) % 10_007
}
const parameters = { numRuns: 300 }
describe("State replay properties", () => {
test("matches a full fold across reads, registrations, removals and batched reloads", () =>
FastCheck.assert(
FastCheck.property(
FastCheck.tuple(source, source),
FastCheck.array(operation, { minLength: 1, maxLength: 6 }),
FastCheck.array(FastCheck.array(command, { maxLength: 48 }), { minLength: 1, maxLength: 8 }),
(initial, operations, batches) =>
Effect.gen(function* () {
const sources = [...initial]
const notifications = [0, 0]
let calls = 0
const states = sources.map((_, index) =>
State.create({
initial: (): Value => ({ value: sources[index], order: [] }),
draft: (draft) => draft,
notify: Effect.sync(() => void notifications[index]++),
}),
)
const callbacks = operations.map((operation, index) => (draft: Value) => {
calls++
draft.value = apply(draft.value, operation)
draft.order.push(index)
})
const registrations: { handle: State.Registration; callback: number; active: boolean }[][] = [[], []]
const expected = (index: number): Value => {
const order = registrations[index].filter((entry) => entry.active).map((entry) => entry.callback)
return {
value: order.reduce((value, callback) => apply(value, operations[callback]), sources[index]),
order,
}
}
yield* Effect.forEach(
batches,
(commands) =>
Effect.gen(function* () {
const before = [...notifications]
const dirty = new Set<number>()
yield* State.batch(
Effect.gen(function* () {
yield* Effect.forEach(
commands,
(command) =>
Effect.gen(function* () {
const state = states[command.target]
switch (command.type) {
case "append": {
const callback = command.callback % callbacks.length
const handle = yield* state.transform(callbacks[callback])
registrations[command.target].push({ handle, callback, active: true })
dirty.add(command.target)
return
}
case "read":
expect(state.get()).toEqual(expected(command.target))
return
case "dispose": {
const entries = registrations[command.target]
if (!entries.length) return
const entry = entries[command.registration % entries.length]
yield* entry.handle.dispose
if (!entry.active) return
entry.active = false
dirty.add(command.target)
return
}
case "reload":
sources[command.target] = command.source
yield* state.reload()
dirty.add(command.target)
}
}),
{ discard: true },
)
expect(notifications).toEqual(before)
}),
)
expect(notifications).toEqual(before.map((count, index) => count + Number(dirty.has(index))))
const flushed = calls
states.forEach((state, index) => expect(state.get()).toEqual(expected(index)))
expect(calls).toBe(flushed)
}),
{ discard: true },
)
}).pipe(Effect.scoped, Effect.runSync),
),
parameters,
))
test("rebuilds all active callbacks once per change, reads for free otherwise, and never touches retained values", () =>
FastCheck.assert(
FastCheck.property(
FastCheck.array(
FastCheck.record({
append: FastCheck.array(operation, { minLength: 1, maxLength: 8 }),
reads: FastCheck.integer({ min: 1, max: 5 }),
}),
{ minLength: 1, maxLength: 8 },
),
FastCheck.nat(100),
(chunks, removal) =>
Effect.gen(function* () {
let calls = 0
const state = State.create({
initial: () => ({ value: 1, order: new Array<number>() }),
draft: (draft) => draft,
})
const registrations: State.Registration[] = []
const retained: { value: Value; snapshot: Value }[] = []
let settled = 0
const remember = () => {
const value = state.get()
retained.push({ value, snapshot: { value: value.value, order: [...value.order] } })
}
yield* State.batch(
Effect.gen(function* () {
yield* Effect.forEach(
chunks,
(chunk) =>
Effect.gen(function* () {
const before = calls
const added = yield* Effect.forEach(chunk.append, (operation) =>
state.transform((draft) => {
calls++
draft.value = apply(draft.value, operation)
draft.order.push(draft.order.length)
}),
)
registrations.push(...added)
expect(calls).toBe(before)
// The first read after a change replays every active callback; later reads do nothing.
state.get()
expect(calls).toBe(before + registrations.length)
Array.from({ length: chunk.reads }).forEach(() => {
const again = state.get()
expect(again).toBe(state.get())
})
expect(calls).toBe(before + registrations.length)
remember()
}),
{ discard: true },
)
const removed = registrations[removal % registrations.length]
yield* removed.dispose
const before = calls
state.get()
expect(calls).toBe(before + registrations.length - 1)
remember()
const replayed = calls
yield* removed.dispose
state.get()
expect(calls).toBe(replayed)
yield* state.reload()
yield* state.reload()
expect(calls).toBe(replayed)
settled = replayed
}),
)
// Batch end notifies, and the two reloads left the value dirty: one more full rebuild.
expect(calls).toBe(settled + registrations.length - 1)
state.get()
expect(calls).toBe(settled + registrations.length - 1)
retained.forEach((entry) => expect(entry.value).toEqual(entry.snapshot))
expect(new Set(retained.map((entry) => entry.value)).size).toBe(retained.length)
}).pipe(Effect.scoped, Effect.runSync),
),
parameters,
))
})
+18 -510
View File
@@ -1,8 +1,10 @@
import { describe, expect } from "bun:test"
import { State } from "@opencode-ai/core/state"
import { Cause, Deferred, Effect, Exit, Fiber, Scope } from "effect"
import { Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect"
import { TestClock } from "effect/testing"
import { it } from "./lib/effect"
import { testEffect } from "./lib/effect"
const it = testEffect(Layer.empty)
describe("State", () => {
it.effect("commits a transform atomically when its updater is interrupted", () =>
@@ -13,9 +15,8 @@ describe("State", () => {
const state = State.create({
initial: () => ({ values: [] as string[] }),
draft: (draft) => ({ add: (value: string) => draft.values.push(value) }),
notify: block
? Deferred.succeed(rebuilding, undefined).pipe(Effect.andThen(Deferred.await(release)))
: Effect.void,
finalize: () =>
block ? Deferred.succeed(rebuilding, undefined).pipe(Effect.andThen(Deferred.await(release))) : Effect.void,
})
const scope = yield* Scope.make()
const fiber = yield* state
@@ -35,20 +36,20 @@ describe("State", () => {
}),
)
it.effect("makes current state visible before notifying", () =>
it.effect("commits rebuilt state before finalize runs", () =>
Effect.gen(function* () {
const observed: string[][] = []
const state: State.Interface<{ values: string[] }, { add: (item: string) => void }> = State.create({
initial: () => ({ values: [] as string[] }),
draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
notify: Effect.sync(() => observed.push([...state.get().values])),
finalize: () => Effect.sync(() => observed.push([...state.get().values])),
})
yield* state.transform((draft) => {
draft.add("value")
})
// Update events publish from notify, so consumers reading on the event
// Update events publish from finalize, so consumers reading on the event
// must observe the rebuilt state, not the previous one.
expect(observed).toEqual([["value"]])
}),
@@ -97,18 +98,18 @@ describe("State", () => {
}),
)
it.effect("batches notifications across domains", () =>
it.effect("batches automatic rebuilds", () =>
Effect.gen(function* () {
let finalized = 0
const first = State.create({
initial: () => ({ values: [] as string[] }),
draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
notify: Effect.sync(() => finalized++),
finalize: () => Effect.sync(() => finalized++),
})
const second = State.create({
initial: () => ({ values: [] as string[] }),
draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
notify: Effect.sync(() => finalized++),
finalize: () => Effect.sync(() => finalized++),
})
yield* State.batch(
@@ -139,7 +140,7 @@ describe("State", () => {
const state = State.create({
initial: () => ({ values: [] as string[] }),
draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
notify: Effect.sync(() => finalized++),
finalize: () => Effect.sync(() => finalized++),
})
const scope = yield* Scope.make()
yield* Scope.addFinalizer(
@@ -151,7 +152,7 @@ describe("State", () => {
const pending = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* TestClock.adjust("250 millis")
yield* State.shutdown(Scope.close(scope, Exit.void))
yield* State.batch(Scope.close(scope, Exit.void), { flush: false })
expect(disposed).toBe(1)
expect(finalized).toBe(1)
@@ -169,12 +170,12 @@ describe("State", () => {
const closing = State.create({
initial: () => ({}),
draft: (draft) => draft,
notify: Effect.sync(() => finalized.push("closing")),
finalize: () => Effect.sync(() => finalized.push("closing")),
})
const live = State.create({
initial: () => ({}),
draft: (draft) => draft,
notify: Effect.sync(() => finalized.push("live")),
finalize: () => Effect.sync(() => finalized.push("live")),
})
const scope = yield* Scope.make()
yield* closing.transform(() => {}).pipe(Scope.provide(scope))
@@ -183,7 +184,7 @@ describe("State", () => {
yield* State.batch(
Effect.gen(function* () {
yield* live.transform(() => {})
yield* State.shutdown(Scope.close(scope, Exit.void))
yield* State.batch(Scope.close(scope, Exit.void), { flush: false })
}),
)
expect(finalized).toEqual(["live"])
@@ -196,7 +197,7 @@ describe("State", () => {
const state = State.create({
initial: () => ({ values: [] as string[] }),
draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
notify: Effect.sync(() => finalized++),
finalize: () => Effect.sync(() => finalized++),
})
yield* state.transform((draft) => {
draft.add("value")
@@ -216,496 +217,3 @@ describe("State", () => {
}),
)
})
describe("State rebuild", () => {
it.effect("leaves a retained value untouched when later registrations rebuild", () =>
Effect.gen(function* () {
const state = State.create({
initial: () => ({ values: new Array<string>(), tags: new Map<string, number>() }),
draft: (data) => data,
})
yield* State.batch(
Effect.gen(function* () {
yield* state.transform((draft) => {
draft.values.push("first")
draft.tags.set("first", 1)
})
const retained = state.get()
expect(state.get()).toBe(retained)
yield* state.transform((draft) => {
draft.values.push("second")
draft.tags.set("second", 2)
})
const current = state.get()
expect(current).not.toBe(retained)
expect(current.values).toEqual(["first", "second"])
expect(Array.from(current.tags.keys())).toEqual(["first", "second"])
expect(retained.values).toEqual(["first"])
expect(Array.from(retained.tags.keys())).toEqual(["first"])
}),
)
}),
)
it.effect("recreates the draft with every rebuild", () =>
Effect.gen(function* () {
let drafts = 0
const state = State.create({
initial: () => ({ values: new Array<number>() }),
draft: (data) => {
drafts++
let sequence = 0
return { add: () => data.values.push(++sequence) }
},
})
yield* State.batch(
Effect.gen(function* () {
yield* state.transform((draft) => draft.add())
expect(state.get().values).toEqual([1])
expect(drafts).toBe(1)
yield* state.transform((draft) => draft.add())
expect(state.get().values).toEqual([1, 2])
expect(drafts).toBe(2)
yield* state.reload()
expect(state.get().values).toEqual([1, 2])
expect(drafts).toBe(3)
}),
)
}),
)
it.effect("rebuilds lazily on read and notifies even without a final read", () =>
Effect.gen(function* () {
const calls: string[] = []
const notifications: string[][] = []
const state: State.Interface<{ values: string[] }, { values: string[] }> = State.create({
initial: () => ({ values: new Array<string>() }),
draft: (data) => data,
notify: Effect.sync(() => notifications.push([...state.get().values])),
})
expect(state.get().values).toEqual([])
yield* State.batch(Effect.void)
expect(notifications).toEqual([])
yield* State.batch(
Effect.gen(function* () {
yield* state.transform((draft) => {
calls.push("first")
draft.values.push("first")
})
yield* state.transform((draft) => {
calls.push("second")
draft.values.push("second")
})
expect(calls).toEqual([])
const view = state.get()
expect(view.values).toEqual(["first", "second"])
expect(state.get()).toBe(view)
expect(calls).toEqual(["first", "second"])
yield* state.transform((draft) => {
calls.push("third")
draft.values.push("third")
})
expect(calls).toEqual(["first", "second"])
expect(state.get()).not.toBe(view)
expect(state.get().values).toEqual(["first", "second", "third"])
expect(view.values).toEqual(["first", "second"])
expect(calls).toEqual(["first", "second", "first", "second", "third"])
yield* state.transform((draft) => {
calls.push("fourth")
draft.values.push("fourth")
})
expect(notifications).toEqual([])
}),
)
expect(calls.slice(5)).toEqual(["first", "second", "third", "fourth"])
expect(notifications).toEqual([["first", "second", "third", "fourth"]])
}),
)
it.effect("replays every callback after each change outside a batch", () =>
Effect.gen(function* () {
const calls: number[] = []
const state = State.create({ initial: () => ({ value: 2 }), draft: (data) => data })
yield* state.transform((draft) => {
calls.push(1)
draft.value += 3
})
yield* state.transform((draft) => {
calls.push(2)
draft.value *= 4
})
// Each registration outside a batch notifies immediately, and notification materializes.
expect(calls).toEqual([1, 1, 2])
expect(state.get().value).toBe(20)
expect(calls).toEqual([1, 1, 2])
yield* state.transform((draft) => {
calls.push(3)
draft.value -= 1
})
expect(state.get().value).toBe(19)
expect(calls).toEqual([1, 1, 2, 1, 2, 3])
}),
)
;[0, 1, 2].forEach((removed) =>
it.effect(`rebuilds noncommutative edits after removing position ${removed}`, () =>
Effect.gen(function* () {
const calls: number[] = []
const state = State.create({ initial: () => ({ value: 5 }), draft: (data) => data })
const registrations = yield* State.batch(
Effect.all([
state.transform((draft) => {
calls.push(0)
draft.value += 1
}),
state.transform((draft) => {
calls.push(1)
draft.value *= 3
}),
state.transform((draft) => {
calls.push(2)
draft.value -= 4
}),
]),
)
expect(state.get().value).toBe(14)
const registration = registrations[removed]
if (!registration) throw new Error("missing registration")
calls.length = 0
yield* registration.dispose
expect(state.get().value).toBe([11, 2, 18][removed])
expect(calls).toEqual([0, 1, 2].filter((index) => index !== removed))
calls.length = 0
yield* registration.dispose
expect(calls).toEqual([])
}),
),
)
it.effect("keeps equal callback registrations independently disposable", () =>
Effect.gen(function* () {
const state = State.create({ initial: () => ({ value: 0 }), draft: (data) => data })
const callback = (draft: { value: number }) => draft.value++
const first = yield* state.transform(callback)
const second = yield* state.transform(callback)
expect(state.get().value).toBe(2)
yield* first.dispose
expect(state.get().value).toBe(1)
yield* first.dispose
expect(state.get().value).toBe(1)
yield* second.dispose
expect(state.get().value).toBe(0)
}),
)
it.effect("does not evaluate a pending callback removed before the first read", () =>
Effect.gen(function* () {
let calls = 0
const state = State.create({ initial: () => ({ value: 0 }), draft: (data) => data })
yield* State.batch(
Effect.gen(function* () {
const registration = yield* state.transform(() => calls++)
yield* registration.dispose
expect(state.get().value).toBe(0)
}),
)
expect(calls).toBe(0)
}),
)
it.effect("invalidates on reload and reads new inputs before notification", () =>
Effect.gen(function* () {
let source = 1
let calls = 0
let notifications = 0
const state = State.create({
initial: () => ({ value: 0 }),
draft: (data) => data,
notify: Effect.sync(() => notifications++),
})
yield* state.transform((draft) => {
calls++
draft.value += source
})
notifications = 0
source = 2
const reload = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
expect(state.get().value).toBe(2)
expect(calls).toBe(2)
expect(notifications).toBe(0)
yield* TestClock.adjust("500 millis")
yield* Fiber.join(reload)
expect(calls).toBe(2)
expect(notifications).toBe(1)
yield* State.batch(
Effect.gen(function* () {
source = 3
yield* state.reload()
expect(state.get().value).toBe(3)
yield* state.transform((draft) => (draft.value *= 10))
expect(state.get().value).toBe(30)
expect(calls).toBe(4)
expect(notifications).toBe(1)
}),
)
expect(notifications).toBe(2)
}),
)
it.effect("resamples a changing initial value even when there are no transforms", () =>
Effect.gen(function* () {
let source = 1
const state = State.create({ initial: () => ({ value: source }), draft: (data) => data })
expect(state.get().value).toBe(1)
// A captured input changed but nothing invalidated the value, so reads stay cached.
source = 2
expect(state.get().value).toBe(1)
yield* State.batch(
Effect.gen(function* () {
yield* state.reload()
expect(state.get().value).toBe(2)
}),
)
}),
)
it.effect("keeps the previous value when a rebuild throws and retries on the next read", () =>
Effect.gen(function* () {
let fail = true
let initializations = 0
const state = State.create({
initial: () => {
initializations++
return { values: new Array<string>() }
},
draft: (data) => data,
})
yield* state.transform((draft) => draft.values.push("first"))
const before = state.get()
expect(initializations).toBe(2)
yield* State.batch(
Effect.gen(function* () {
yield* state.transform((draft) => {
draft.values.push("second")
if (fail) throw new Error("failed edit")
})
expect(() => state.get()).toThrow("failed edit")
expect(initializations).toBe(3)
expect(() => state.get()).toThrow("failed edit")
expect(initializations).toBe(4)
// The partially edited container is discarded; the last complete value is untouched.
expect(before.values).toEqual(["first"])
fail = false
expect(state.get().values).toEqual(["first", "second"])
expect(initializations).toBe(5)
yield* state.transform((draft) => draft.values.push("third"))
expect(state.get().values).toEqual(["first", "second", "third"])
expect(initializations).toBe(6)
}),
)
}),
)
it.effect("recovers by disposing a failing callback without keeping its partial edits", () =>
Effect.gen(function* () {
const state = State.create({ initial: () => ({ value: 2 }), draft: (data) => data })
yield* state.transform((draft) => (draft.value *= 3))
yield* State.batch(
Effect.gen(function* () {
const failing = yield* state.transform((draft) => {
draft.value += 100
throw new Error("bad callback")
})
expect(() => state.get()).toThrow("bad callback")
yield* failing.dispose
expect(state.get().value).toBe(6)
yield* state.transform((draft) => (draft.value += 1))
expect(state.get().value).toBe(7)
}),
)
}),
)
})
describe("State notification boundaries", () => {
;["body", "observer"].forEach((phase) =>
it.effect(
`cancels remaining observers when interrupted during the ${phase}, without rolling back registrations`,
() =>
Effect.gen(function* () {
const entered = yield* Deferred.make<void>()
const observed: string[] = []
let block = true
const first = State.create({
initial: () => ({ value: 0 }),
draft: (data) => data,
notify: Effect.gen(function* () {
observed.push("first")
if (phase !== "observer" || !block) return
yield* Deferred.succeed(entered, undefined)
yield* Effect.never
}),
})
const second = State.create({
initial: () => ({ value: 0 }),
draft: (data) => data,
notify: Effect.sync(() => observed.push("second")),
})
const writer = yield* State.batch(
Effect.gen(function* () {
yield* first.transform((draft) => draft.value++)
yield* second.transform((draft) => draft.value++)
if (phase !== "body") return
yield* Deferred.succeed(entered, undefined)
yield* Effect.never
}),
).pipe(Effect.forkChild({ startImmediately: true }))
yield* Deferred.await(entered)
yield* Fiber.interrupt(writer)
block = false
expect(Exit.hasInterrupts(yield* Fiber.await(writer))).toBe(true)
expect([first.get().value, second.get().value]).toEqual([1, 1])
expect(observed).toEqual(phase === "body" ? [] : ["first"])
}),
),
)
it.effect("shares nested live batches and does not retain an escaped batch", () =>
Effect.gen(function* () {
let notifications = 0
const state = State.create({
initial: () => ({ value: 0 }),
draft: (data) => data,
notify: Effect.sync(() => notifications++),
})
const inherit = yield* State.batch(
Effect.gen(function* () {
yield* state.transform((draft) => draft.value++)
yield* State.batch(state.transform((draft) => draft.value++))
expect(state.get().value).toBe(2)
expect(notifications).toBe(0)
return yield* State.inherit()
}),
)
expect(notifications).toBe(1)
yield* inherit(state.transform((draft) => draft.value++))
expect(state.get().value).toBe(3)
expect(notifications).toBe(2)
}),
)
it.effect("lets observers read other pending domains and register more edits", () =>
Effect.gen(function* () {
const scope = yield* Scope.Scope
const observed: number[] = []
let added = false
const other = State.create({ initial: () => ({ value: 0 }), draft: (data) => data })
const state: State.Interface<object, object> = State.create({
initial: () => ({}),
draft: (data) => data,
notify: Effect.gen(function* () {
observed.push(other.get().value)
if (added) return
added = true
yield* state.transform(() => {}).pipe(Scope.provide(scope))
}),
})
yield* State.batch(
Effect.gen(function* () {
yield* state.transform(() => {})
yield* other.transform((draft) => (draft.value = 42))
}),
)
expect(observed).toEqual([42, 42])
}),
)
it.effect("allows a debounced observer to await another reload", () =>
Effect.gen(function* () {
let source = 1
let reloadAgain = false
const observed: number[] = []
const state: State.Interface<{ value: number }, { value: number }> = State.create({
initial: () => ({ value: 0 }),
draft: (data) => data,
notify: Effect.gen(function* () {
observed.push(state.get().value)
if (!reloadAgain) return
reloadAgain = false
source = 3
yield* state.reload()
}),
})
yield* state.transform((draft) => (draft.value = source))
source = 2
reloadAgain = true
const reload = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* TestClock.adjust("1 second")
yield* Fiber.join(reload)
expect(observed).toEqual([1, 2, 3])
}),
)
it.effect("attempts every domain notification and preserves both batch and observer failures", () =>
Effect.gen(function* () {
const observed: string[] = []
let fail = true
const first = State.create({
initial: () => ({}),
draft: (data) => data,
notify: Effect.suspend(() => (fail ? Effect.die("observer failed") : Effect.void)),
})
const second = State.create({
initial: () => ({}),
draft: (data) => data,
notify: Effect.sync(() => observed.push("second")),
})
const exit = yield* State.batch(
Effect.gen(function* () {
yield* first.transform(() => {})
yield* second.transform(() => {})
return yield* Effect.fail("body failed")
}),
).pipe(Effect.exit)
fail = false
expect(observed).toEqual(["second"])
expect(Exit.isFailure(exit)).toBe(true)
if (Exit.isFailure(exit)) {
expect(Cause.pretty(exit.cause)).toContain("observer failed")
expect(Cause.pretty(exit.cause)).toContain("body failed")
}
}),
)
it.effect("keeps cancelled reload callers independent and shares notification results", () =>
Effect.gen(function* () {
let fail = false
let notifications = 0
const state = State.create({
initial: () => ({}),
draft: (data) => data,
notify: Effect.sync(() => {
notifications++
if (fail) throw new Error("notification failed")
}),
})
yield* state.transform(() => {})
notifications = 0
fail = true
const cancelled = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
const first = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
const second = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* Fiber.interrupt(cancelled)
yield* TestClock.adjust("500 millis")
const exits = yield* Fiber.awaitAll([first, second])
fail = false
expect(exits.every(Exit.isFailure)).toBe(true)
expect(notifications).toBe(1)
const recovered = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* TestClock.adjust("500 millis")
yield* Fiber.join(recovered)
expect(notifications).toBe(2)
}),
)
})
+3 -40
View File
@@ -242,7 +242,6 @@ describe("Tool", () => {
draft.add({ ...constant("overlay"), name: "echo", options: { codemode: false } })
})
.pipe(Scope.provide(scope))
// Each registration outside a batch notifies immediately, and every rebuild replays all transforms.
expect(runs).toEqual(["base", "base", "overlay"])
expect((yield* executeTool(service, call("echo"))).output).toEqual({ text: "overlay" })
@@ -255,7 +254,7 @@ describe("Tool", () => {
}),
)
it.effect("reads pending tools inside a batch and suppresses terminal teardown replay", () =>
it.effect("batches tool publication and suppresses terminal teardown replay", () =>
Effect.gen(function* () {
const service = yield* Tool.Service
const runs: string[] = []
@@ -271,14 +270,13 @@ describe("Tool", () => {
draft.add({ ...constant("overlay"), name: "echo", options: { codemode: false } })
})
expect(runs).toEqual([])
expect((yield* service.snapshot()).definitions.map((tool) => tool.name)).toEqual(["echo", "execute"])
expect(runs).toEqual(["base", "overlay"])
expect((yield* service.snapshot()).definitions.map((tool) => tool.name)).toEqual(["execute"])
}).pipe(Scope.provide(scope)),
)
expect(runs).toEqual(["base", "overlay"])
expect((yield* executeTool(service, call("echo"))).output).toEqual({ text: "overlay" })
yield* State.shutdown(Scope.close(scope, Exit.void))
yield* State.batch(Scope.close(scope, Exit.void), { flush: false })
expect(runs).toEqual(["base", "overlay"])
}),
)
@@ -508,41 +506,6 @@ describe("Tool", () => {
}),
)
it.effect("retains namespace descriptions in executable snapshots after appended transforms", () =>
Effect.gen(function* () {
const service = yield* Tool.Service
yield* service.transform((draft) => {
draft.namespace({ name: "acme", description: "Archival operations" })
draft.add({ ...make(), options: { namespace: "acme" } })
})
const advertised = yield* service.snapshot()
yield* service.transform((draft) => {
draft.namespace({ name: "acme", description: "Billing operations" })
})
const current = yield* service.snapshot()
expect(advertised.codeModeCatalog?.tools).toMatchObject([{ name: "acme", description: "Archival operations" }])
expect(current.codeModeCatalog?.tools).toMatchObject([{ name: "acme", description: "Billing operations" }])
const search = (snapshot: Tool.Snapshot, query: string) =>
snapshot.execute({
...call("execute"),
call: {
type: "tool-call",
id: `namespace-${query}`,
name: "execute",
input: {
code: `return search({ query: ${JSON.stringify(query)} }).items.map(item => item.path).join(",")`,
},
},
})
expect((yield* search(advertised, "archival")).output).toMatchObject({ output: "tools.acme.echo" })
expect((yield* search(advertised, "billing")).output).toMatchObject({ output: "" })
expect((yield* search(current, "archival")).output).toMatchObject({ output: "" })
expect((yield* search(current, "billing")).output).toMatchObject({ output: "tools.acme.echo" })
}),
)
it.effect("preserves a top-level tool that also has child tools", () =>
Effect.gen(function* () {
const service = yield* Tool.Service
+3 -286
View File
@@ -2,8 +2,7 @@ import { $ } from "bun"
import { describe, expect } from "bun:test"
import fs from "fs/promises"
import path from "path"
import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Scope, Stream } from "effect"
import { TestClock } from "effect/testing"
import { Cause, Effect, Exit, Fiber, Layer, Stream } from "effect"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { AppProcess } from "@opencode-ai/util/process"
import { FSUtil } from "@opencode-ai/util/fs-util"
@@ -11,7 +10,6 @@ import { Git } from "@opencode-ai/core/git"
import { Bus } from "@opencode-ai/core/bus"
import { Location } from "@opencode-ai/core/location"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { State } from "@opencode-ai/core/state"
import { Vcs } from "@opencode-ai/core/vcs"
import { VcsGitPlugin } from "@opencode-ai/core/plugin/vcs/git"
import type { VcsDefinition, VcsDiffInput } from "@opencode-ai/plugin/effect/vcs"
@@ -19,15 +17,9 @@ import { FileSystem } from "@opencode-ai/schema/filesystem"
import { VcsEvent } from "@opencode-ai/schema/vcs-event"
import { location } from "./fixture/location"
import { tmpdir } from "./fixture/tmpdir"
import { it, testEffect } from "./lib/effect"
import { it } from "./lib/effect"
import { host } from "./plugin/host"
const Done = Bus.ephemeral({ type: "test.vcs.done", schema: {} })
const here = Location.node.replace(
Layer.succeed(Location.Service, Location.Service.of(location({ directory: AbsolutePath.make(import.meta.dir) }))),
)
const synthetic = testEffect(LayerNode.compile(LayerNode.group([Vcs.node, Bus.node]), { replacements: [here] }))
const provide = (directory: string, input: { git?: boolean; worktree?: string } = {}) =>
Effect.provide(
LayerNode.compile(LayerNode.group([Vcs.node, Bus.node, Location.node, AppProcess.node, FSUtil.node, Git.node]), {
@@ -154,44 +146,6 @@ describe("Vcs", () => {
),
)
synthetic.effect("reads batched providers without refreshing intermediate selections", () =>
Effect.gen(function* () {
const vcs = yield* Vcs.Service
const reads: string[] = []
yield* State.batch(
Effect.gen(function* () {
yield* vcs.transform((draft) => {
draft.add(
provider({
info: () =>
Effect.sync(() => {
reads.push("intermediate")
return { branch: { current: "intermediate" } }
}),
}),
)
draft.default.set("custom")
})
expect((yield* vcs.status())[0]?.file).toBe("file.txt")
expect(yield* vcs.info()).toEqual({ branch: {} })
yield* vcs.transform((draft) =>
draft.add(
provider({
info: () =>
Effect.sync(() => {
reads.push("final")
return { branch: { current: "final" } }
}),
}),
),
)
}),
)
expect(reads).toEqual(["final"])
expect(yield* vcs.info()).toEqual({ branch: { current: "final" } })
}),
)
it.live("passes location scope and bounded diff options to providers", () =>
withTmp((directory) =>
Effect.gen(function* () {
@@ -256,16 +210,8 @@ describe("Vcs", () => {
withTmp((directory) =>
Effect.gen(function* () {
const vcs = yield* Vcs.Service
let interrupt = false
yield* vcs.transform((draft) => {
draft.add(
provider({
info: () => (interrupt ? Effect.interrupt : Effect.succeed({ branch: { current: "feature" } })),
status: () => Effect.never,
diff: () => Effect.never,
base: () => Effect.never,
}),
)
draft.add(provider({ status: () => Effect.never, diff: () => Effect.never, base: () => Effect.never }))
draft.default.set("custom")
})
@@ -281,239 +227,10 @@ describe("Vcs", () => {
yield* Fiber.interrupt(base)
const cancelled = yield* Fiber.await(base)
expect(Exit.isFailure(cancelled) && Cause.hasInterrupts(cancelled.cause)).toBeTrue()
interrupt = true
const refreshed = yield* vcs.reload().pipe(Effect.exit)
expect(Exit.isFailure(refreshed) && Cause.hasInterruptsOnly(refreshed.cause)).toBeTrue()
}).pipe(provide(directory)),
),
)
it.effect("stops in-flight and queued VCS reloads when its layer closes", () =>
Effect.gen(function* () {
const bus = yield* Bus.Service
const entered = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const root = yield* Scope.make()
yield* Effect.addFinalizer(() =>
Deferred.succeed(release, undefined).pipe(
Effect.andThen(State.shutdown(Scope.close(root, Exit.void))),
Effect.andThen(TestClock.adjust("500 millis")),
),
)
const context = yield* Layer.buildWithScope(
LayerNode.compile(Vcs.node, { replacements: [Bus.node.replace(Layer.succeed(Bus.Service, bus)), here] }),
root,
)
const vcs = Context.get(context, Vcs.Service)
const reads: string[] = []
const observed: (string | undefined)[] = []
yield* Effect.acquireRelease(
bus.listen((event) =>
Effect.sync(() => {
if (event.type !== VcsEvent.BranchUpdated.type) return
observed.push(Schema.decodeUnknownSync(VcsEvent.BranchUpdated.data)(event.data).branch)
}),
),
(unsubscribe) => unsubscribe,
)
let branch = "initial"
let block = false
yield* vcs
.transform((draft) => {
draft.add(
provider({
info: () =>
Effect.gen(function* () {
const value = branch
reads.push(value)
if (block) {
block = false
yield* Deferred.succeed(entered, undefined)
yield* Deferred.await(release)
}
return { branch: { current: value } }
}),
}),
)
draft.default.set("custom")
})
.pipe(Scope.provide(root))
observed.length = 0
block = true
const first = yield* vcs.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* TestClock.adjust("500 millis")
yield* Deferred.await(entered)
branch = "late"
const second = yield* vcs.reload().pipe(Effect.forkChild({ startImmediately: true }))
yield* TestClock.adjust("500 millis")
expect(reads).toEqual(["initial", "initial"])
expect(first.pollUnsafe()).toBeUndefined()
expect(second.pollUnsafe()).toBeUndefined()
const snapshot = yield* vcs.info()
const shutdown = yield* State.shutdown(Scope.close(root, Exit.void)).pipe(
Effect.forkChild({ startImmediately: true }),
)
yield* TestClock.adjust("1 millis")
expect(shutdown.pollUnsafe()).toBeDefined()
expect(first.pollUnsafe()).toBeDefined()
expect(second.pollUnsafe()).toBeDefined()
expect(yield* Deferred.isDone(release)).toBe(false)
yield* Fiber.join(shutdown)
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(first)
yield* Fiber.join(second)
expect(reads).toEqual(["initial", "initial"])
expect(observed).toEqual([])
expect(yield* vcs.info()).toBe(snapshot)
}).pipe(Effect.provide(LayerNode.compile(Bus.node))),
)
it.live("keeps watching HEAD changes after a transform replay failure", () =>
withGit((directory) =>
Effect.gen(function* () {
const vcs = yield* Vcs.Service
const bus = yield* Bus.Service
const replayed = yield* Deferred.make<void>()
const faulty = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(faulty, Exit.void))
let branch = "initial"
yield* vcs.transform((draft) =>
draft.add(provider({ id: "git", info: () => Effect.sync(() => ({ branch: { current: branch } })) })),
)
const failure = new Error("fixture replay failed")
let replays = 0
const failed = yield* vcs
.transform(() => {
if (++replays === 2) Deferred.doneUnsafe(replayed, Exit.void)
throw failure
})
.pipe(Scope.provide(faulty), Effect.exit)
expect(Exit.isFailure(failed) && Cause.squash(failed.cause)).toBe(failure)
yield* bus.publish(FileSystem.Event.Changed, { file: path.join(directory, ".git", "HEAD"), event: "change" })
yield* Deferred.await(replayed).pipe(Effect.timeout("1 second"))
yield* Effect.yieldNow
const status = yield* vcs.status().pipe(Effect.exit)
expect(Exit.isFailure(status) && Cause.squash(status.cause)).toBe(failure)
expect((yield* vcs.info()).branch.current).toBe("initial")
branch = "recovered"
yield* Scope.close(faulty, Exit.void)
expect((yield* vcs.info()).branch.current).toBe("recovered")
const updated = yield* bus
.subscribe(VcsEvent.BranchUpdated)
.pipe(Stream.runHead, Effect.timeout("1 second"), Effect.forkScoped({ startImmediately: true }))
branch = "after-recovery"
yield* bus.publish(FileSystem.Event.Changed, { file: path.join(directory, ".git", "HEAD"), event: "change" })
expect(Option.getOrUndefined(yield* Fiber.join(updated))).toMatchObject({ data: { branch: "after-recovery" } })
expect((yield* vcs.info()).branch.current).toBe("after-recovery")
}),
),
)
it.live("serializes filesystem and config refreshes while reading the latest desired provider", () =>
withGit((directory) =>
Effect.gen(function* () {
const vcs = yield* Vcs.Service
const bus = yield* Bus.Service
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const accepted = yield* Deferred.make<void>()
const reads: string[] = []
yield* vcs.transform((draft) =>
draft.add(
provider({
id: "git",
info: () =>
Effect.gen(function* () {
reads.push(reads.length === 0 ? "initial" : "filesystem")
if (reads.length === 1) return { branch: { current: "initial" } }
yield* Deferred.succeed(started, undefined)
yield* Deferred.await(release)
return { branch: { current: "filesystem" } }
}),
}),
),
)
const updates = yield* bus
.subscribe(VcsEvent.BranchUpdated)
.pipe(Stream.take(2), Stream.runLast, Effect.forkScoped({ startImmediately: true }))
yield* Effect.gen(function* () {
yield* bus.publish(FileSystem.Event.Changed, { file: path.join(directory, ".git", "HEAD"), event: "change" })
yield* Deferred.await(started)
const configured = yield* State.batch(
Effect.gen(function* () {
yield* vcs.transform((draft) =>
draft.add(
provider({
id: "git",
info: () =>
Effect.sync(() => {
reads.push("config")
return { branch: { current: "config" } }
}),
status: () => Effect.succeed([{ file: "config.txt", additions: 1, deletions: 0, status: "added" }]),
}),
),
)
expect((yield* vcs.status())[0]?.file).toBe("config.txt")
expect(yield* vcs.info()).toEqual({ branch: { current: "initial" } })
yield* Deferred.succeed(accepted, undefined)
}),
).pipe(Effect.forkScoped({ startImmediately: true }))
yield* Deferred.await(accepted)
expect(reads).toEqual(["initial", "filesystem"])
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(configured)
expect(Option.getOrUndefined(yield* Fiber.join(updates))?.data.branch).toBe("config")
expect(yield* vcs.info()).toEqual({ branch: { current: "config" } })
expect(reads).toEqual(["initial", "filesystem", "config"])
}).pipe(Effect.ensuring(Deferred.succeed(release, undefined)))
}),
),
)
synthetic.effect("keeps branch streams current when listeners change the selected provider", () =>
Effect.gen(function* () {
const vcs = yield* Vcs.Service
const bus = yield* Bus.Service
const scope = yield* Effect.scope
const updates = yield* bus.subscribe([VcsEvent.BranchUpdated, Done]).pipe(
Stream.takeUntil((event) => event.type === Done.type),
Stream.runCollect,
Effect.forkScoped({ startImmediately: true }),
)
const unsubscribe = yield* bus.listen((event) => {
if (
event.type !== VcsEvent.BranchUpdated.type ||
Schema.decodeUnknownSync(VcsEvent.BranchUpdated.data)(event.data).branch !== "feature"
)
return Effect.void
return vcs
.transform((draft) =>
draft.add(provider({ info: () => Effect.succeed({ branch: { current: "listener" } }) })),
)
.pipe(Scope.provide(scope), Effect.asVoid)
})
yield* Effect.gen(function* () {
yield* vcs.transform((draft) => {
draft.add(provider())
draft.default.set("custom")
})
yield* bus.publish(Done, {})
const events = (yield* Fiber.join(updates)).filter((event) => event.type === VcsEvent.BranchUpdated.type)
expect(yield* vcs.info()).toEqual({ branch: { current: "listener" } })
expect(events.length).toBeGreaterThanOrEqual(2)
expect(events.at(-1)?.data.branch).toBe((yield* vcs.info()).branch.current)
}).pipe(Effect.ensuring(unsubscribe))
}),
)
it.live("lists local branches by recent activity", () =>
withGit((directory) =>
Effect.gen(function* () {
+22
View File
@@ -17,3 +17,25 @@ bundle the assets as an application. The resulting app will be in `dist/`.
```bash
bun run build && bun run package
```
Production builds require a prebuilt V2 CLI distribution. The release workflow supplies the artifact from the same run:
```bash
OPENCODE_CHANNEL=prod OPENCODE_CLI_DIST=/absolute/path/to/packages/cli/dist bun run build
OPENCODE_CHANNEL=prod bun run package
```
Set `OPENCODE_CLI_TARGET` when packaging for a different architecture. The CLI is placed outside `app.asar` in the
application's resources directory, and packaging fails if it is missing.
CLI preparation uses these channel rules:
| Channel | Without `OPENCODE_CLI_DIST` | With `OPENCODE_CLI_DIST` |
| -------------------------------------- | ------------------------------ | --------------------------------------------- |
| `dev`, `local`, unset, or unrecognized | Download the dev CLI | Download the dev CLI; ignore the distribution |
| `beta` | Download the beta CLI | Copy the supplied CLI; fail if it is missing |
| `prod`, `latest` | Fail before changing resources | Copy the supplied CLI; fail if it is missing |
`bun dev` is separate from packaging: it uses local renderer/server mode, the dev app identity, and the CLI source by
default. `bun dev --download-server <version>` instead downloads that CLI version for local development. Neither path
requires `OPENCODE_CLI_DIST` or runs the production prebuild.
@@ -190,19 +190,8 @@ for (const channel of ["dev", "beta"] as const) {
{
from: "resources/",
to: "",
filter: ["opencode-cli*"],
filter: ["opencode-cli", "opencode-cli.exe"],
},
])
})
}
test("does not bundle the CLI in prod builds", async () => {
const previous = process.env.OPENCODE_CHANNEL
process.env.OPENCODE_CHANNEL = "prod"
const module = await import("./electron-builder.config.ts?no-cli-resource=prod")
const config = module.default as Configuration
if (previous === undefined) delete process.env.OPENCODE_CHANNEL
else process.env.OPENCODE_CHANNEL = previous
expect(config.extraResources).toEqual([])
})
+17 -10
View File
@@ -1,4 +1,5 @@
import { execFile } from "node:child_process"
import { stat } from "node:fs/promises"
import path from "node:path"
import { fileURLToPath } from "node:url"
import { promisify } from "node:util"
@@ -45,6 +46,7 @@ export function macSignOptions(options: CustomMacSignOptions): CustomMacSignOpti
const channel = (() => {
const raw = process.env.OPENCODE_CHANNEL
if (raw === "dev" || raw === "beta" || raw === "prod") return raw
if (raw === "latest") return "prod"
return "dev"
})()
@@ -84,16 +86,21 @@ const getBase = (appId: string): Configuration => ({
"!**/node_modules/js-yaml/dist/{js-yaml.js,js-yaml.min.js,*.map}",
"!**/node_modules/js-yaml/bin{,/**/*}",
],
extraResources:
channel !== "prod"
? [
{
from: "resources/",
to: "",
filter: ["opencode-cli*"],
},
]
: [],
extraResources: [
{
from: "resources/",
to: "",
filter: ["opencode-cli", "opencode-cli.exe"],
},
],
afterPack: async (context) => {
const cli = path.join(
context.packager.getResourcesDir(context.appOutDir),
context.electronPlatformName === "win32" ? "opencode-cli.exe" : "opencode-cli",
)
const file = await stat(cli)
if (!file.isFile() || file.size === 0) throw new Error(`Bundled CLI must be a non-empty file: ${cli}`)
},
mac: {
category: "public.app-category.developer-tools",
icon: `resources/icons/icon.icns`,
+2 -1
View File
@@ -34,7 +34,8 @@ export default defineConfig(({ command }) => ({
dedupe: ["effect"],
},
define: {
"import.meta.env.OPENCODE_CHANNEL": JSON.stringify(channel),
// Local renderer/server mode still uses the dev application identity and updater policy.
"import.meta.env.OPENCODE_CHANNEL": JSON.stringify(channel === "local" ? "dev" : channel),
},
build: {
minify: command === "build",
+7 -1
View File
@@ -4,9 +4,15 @@ import { $ } from "bun"
import { copyBuiltCliToResources, downloadCliToResources, resolveChannel } from "./utils"
const channel = resolveChannel()
if (channel === "prod" && !Bun.env.OPENCODE_CLI_DIST) {
throw new Error("OPENCODE_CLI_DIST is required for production desktop builds")
}
await $`bun ./scripts/copy-icons.ts ${channel}`
await $`bun ./scripts/copy-metainfo.ts ${channel}`
if (channel === "dev") await downloadCliToResources()
if (channel === "beta" && Bun.env.OPENCODE_CLI_DIST) await copyBuiltCliToResources(Bun.env.OPENCODE_CLI_DIST)
if ((channel === "beta" || channel === "prod") && Bun.env.OPENCODE_CLI_DIST) {
await copyBuiltCliToResources(Bun.env.OPENCODE_CLI_DIST)
}
if (channel === "beta" && !Bun.env.OPENCODE_CLI_DIST) await downloadCliToResources("beta")
+1
View File
@@ -10,6 +10,7 @@ export type Channel = "dev" | "beta" | "prod"
export function resolveChannel(): Channel {
const raw = Bun.env.OPENCODE_CHANNEL
if (raw === "dev" || raw === "beta" || raw === "prod") return raw
if (raw === "latest") return "prod"
return "dev"
}
+1 -1
View File
@@ -45,7 +45,7 @@ yield *
})
```
Registry reads rebuild synchronously when registrations changed, applying every transform in registration order to a fresh value; unchanged registries return the previous value. Values read earlier are never mutated. Notifications and resource reconciliation run separately from that materialization.
OpenCode rebuilds the domain when a transform is registered or disposed. A rebuild starts from fresh domain state and runs every active transform in registration order.
Available transform hooks are namespaced by domain:
-76
View File
@@ -486,82 +486,6 @@
"summary": "List plugins"
}
},
"/api/plugin/await-activation": {
"post": {
"tags": ["plugin"],
"operationId": "v2.plugin.awaitActivation",
"parameters": [
{
"name": "location",
"in": "query",
"schema": {
"anyOf": [
{
"type": "object",
"properties": {
"directory": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
},
"workspace": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
}
},
"additionalProperties": false
},
{
"type": "null"
}
]
},
"required": false,
"style": "deepObject",
"explode": true
}
],
"security": [],
"responses": {
"204": {
"description": "<No Content>"
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
}
},
"description": "Wait for configured plugin activation at a Location to settle, including missing-package installs. Completion does not imply every plugin succeeded or background resource discovery finished. Cancelling this wait does not cancel activation.",
"summary": "Wait for plugin activation"
}
},
"/api/plugin/check": {
"post": {
"tags": ["plugin"],
-15
View File
@@ -20,21 +20,6 @@ export const PluginGroup = HttpApiGroup.make("server.plugin")
}),
),
)
.add(
HttpApiEndpoint.post("plugin.awaitActivation", "/api/plugin/await-activation", {
query: LocationQuery,
success: HttpApiSchema.NoContent,
})
.annotateMerge(locationQueryOpenApi)
.annotateMerge(
OpenApi.annotations({
identifier: "v2.plugin.awaitActivation",
summary: "Wait for plugin activation",
description:
"Wait for configured plugin activation at a Location to settle, including missing-package installs. Completion does not imply every plugin succeeded or background resource discovery finished. Cancelling this wait does not cancel activation.",
}),
),
)
.add(
HttpApiEndpoint.post("plugin.check", "/api/plugin/check", {
query: LocationQuery,
-1
View File
@@ -13,7 +13,6 @@ export const PluginHandler = HttpApiBuilder.group(Api, "server.plugin", (handler
return yield* response(Plugin.Service.use((plugin) => plugin.list()))
}),
)
.handle("plugin.awaitActivation", () => Plugin.Service.use((plugin) => plugin.awaitActivation))
.handle("plugin.check", (ctx) =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
@@ -1,192 +0,0 @@
import { expect } from "bun:test"
import { mkdir } from "node:fs/promises"
import path from "node:path"
import { SdkPlugins } from "@opencode-ai/core/plugin/sdk"
import { Plugin } from "@opencode-ai/plugin/effect"
import { Context, Deferred, Effect, Fiber, Layer } from "effect"
import { HttpEffect, HttpRouter, HttpServer } from "effect/unstable/http"
import { tmpdirScoped } from "../../core/test/fixture/tmpdir"
import { it } from "../../core/test/lib/effect"
import { createRoutes } from "../src/routes"
const fixture = Effect.fn(function* (plugin: Plugin.Plugin) {
const tmp = yield* tmpdirScoped("opencode-plugin-activation-")
const first = path.join(tmp.path, "first")
const second = path.join(tmp.path, "second")
const config = path.join(tmp.path, "config")
yield* Effect.promise(() => Promise.all([first, second, config].map((directory) => mkdir(directory))))
const context = yield* Layer.build(
createRoutes({
password: "secret",
database: { path: ":memory:" },
models: { fetch: false },
fs: { filewatcher: false },
config: {
directory: config,
project: false,
content: JSON.stringify({
providers: {
acme: {
models: {
reasoner: { name: "Configured Reasoner", limit: { context: 96_000, output: 8_000 } },
},
},
},
}),
},
}).pipe(Layer.provide(HttpServer.layerServices)),
)
const sdk = Context.get(context, SdkPlugins.Service)
yield* sdk.register(plugin)
const handler = Context.get(context, HttpRouter.HttpRouter).asHttpEffect().pipe(HttpEffect.toWebHandlerWith(context))
return {
first,
second,
request: (method: "GET" | "POST", route: string, directory = first, signal?: AbortSignal) =>
Effect.promise((interruption) => {
const url = new URL(route, "http://opencode.local")
url.searchParams.set("location[directory]", directory)
return handler(
new Request(url, {
method,
headers: { authorization: `Basic ${btoa("opencode:secret")}` },
signal: signal ?? interruption,
}),
)
}),
}
})
it.live(
"awaits activation only for the requested location without blocking model or plugin snapshots",
() =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const server = yield* fixture(
Plugin.define({
id: "slow-plugin",
effect: (ctx) =>
Effect.gen(function* () {
if (path.basename(ctx.location.directory) !== "first") return
yield* Deferred.succeed(started, undefined)
yield* Deferred.await(release)
}),
}),
)
const pending = yield* server.request("POST", "/api/plugin/await-activation").pipe(Effect.forkScoped)
yield* Deferred.await(started)
expect(pending.pollUnsafe()).toBeUndefined()
const models = yield* server.request("GET", "/api/model")
expect(models.status).toBe(200)
expect(yield* Effect.promise(() => models.json())).toMatchObject({
location: { directory: server.first },
data: expect.not.arrayContaining([expect.objectContaining({ providerID: "acme", id: "reasoner" })]),
})
const plugins = yield* server.request("GET", "/api/plugin")
expect(plugins.status).toBe(200)
expect(yield* Effect.promise(() => plugins.json())).toMatchObject({ location: { directory: server.first } })
const second = yield* server.request("POST", "/api/plugin/await-activation", server.second)
expect(second.status).toBe(204)
expect(pending.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(release, undefined)
const response = yield* Fiber.join(pending)
expect(response.status).toBe(204)
expect(yield* Effect.promise(() => response.text())).toBe("")
const configured = yield* server.request("GET", "/api/model")
expect(configured.status).toBe(200)
expect(yield* Effect.promise(() => configured.json())).toMatchObject({
location: { directory: server.first },
data: expect.arrayContaining([
expect.objectContaining({
providerID: "acme",
id: "reasoner",
name: "Configured Reasoner",
limit: { context: 96_000, output: 8_000 },
}),
]),
})
const active = yield* server.request("GET", "/api/plugin")
expect(active.status).toBe(200)
expect(yield* Effect.promise(() => active.json())).toMatchObject({
data: expect.arrayContaining([
expect.objectContaining({ id: "slow-plugin", source: { type: "sdk" }, state: { status: "active" } }),
]),
})
}).pipe(Effect.timeout("10 seconds")),
15_000,
)
it.live(
"aborting an activation wait does not cancel plugin setup",
() =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const completed = yield* Deferred.make<void>()
const interrupted = yield* Deferred.make<void>()
const server = yield* fixture(
Plugin.define({
id: "slow-plugin",
effect: () =>
Effect.gen(function* () {
yield* Deferred.succeed(started, undefined)
yield* Deferred.await(release)
yield* Deferred.succeed(completed, undefined)
}).pipe(Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined))),
}),
)
const controller = new AbortController()
yield* Effect.addFinalizer(() => Effect.sync(() => controller.abort()))
const pending = yield* server
.request("POST", "/api/plugin/await-activation", server.first, controller.signal)
.pipe(Effect.forkScoped)
yield* Deferred.await(started)
controller.abort()
// HttpEffect resolves a cancelled Web request with 499 rather than rejecting its Promise.
expect((yield* Fiber.join(pending)).status).toBe(499)
expect(yield* Deferred.isDone(interrupted)).toBe(false)
expect(yield* Deferred.isDone(completed)).toBe(false)
yield* Deferred.succeed(release, undefined)
expect((yield* server.request("POST", "/api/plugin/await-activation")).status).toBe(204)
expect(yield* Deferred.isDone(completed)).toBe(true)
expect(yield* Deferred.isDone(interrupted)).toBe(false)
const plugins = yield* server.request("GET", "/api/plugin")
expect(plugins.status).toBe(200)
expect(yield* Effect.promise(() => plugins.json())).toMatchObject({
data: expect.arrayContaining([expect.objectContaining({ id: "slow-plugin", state: { status: "active" } })]),
})
}).pipe(Effect.timeout("10 seconds")),
15_000,
)
it.live(
"settles activation when plugin setup fails and exposes the failure in the inventory",
() =>
Effect.gen(function* () {
const server = yield* fixture(
Plugin.define({
id: "failing-plugin",
effect: () => Effect.die(new Error("fixture setup failed")),
}),
)
expect((yield* server.request("POST", "/api/plugin/await-activation")).status).toBe(204)
const plugins = yield* server.request("GET", "/api/plugin")
expect(plugins.status).toBe(200)
expect(yield* Effect.promise(() => plugins.json())).toMatchObject({
location: { directory: server.first },
data: expect.arrayContaining([
expect.objectContaining({
id: "failing-plugin",
source: { type: "sdk" },
state: { status: "failed", error: expect.stringContaining("fixture setup failed") },
}),
]),
})
}).pipe(Effect.timeout("10 seconds")),
15_000,
)
+64 -96
View File
@@ -1,106 +1,74 @@
import fs from "node:fs/promises"
import path from "node:path"
import { expect } from "bun:test"
import { SdkPlugins } from "@opencode-ai/core/plugin/sdk"
import { Plugin } from "@opencode-ai/plugin/effect"
import { Context, Deferred, Effect, Fiber, Layer } from "effect"
import { HttpEffect, HttpRouter, HttpServer } from "effect/unstable/http"
import { tmpdirScoped } from "../../core/test/fixture/tmpdir"
import { Effect, Schedule } from "effect"
import { tmpdir } from "../../core/test/fixture/tmpdir"
import { it } from "../../core/test/lib/effect"
import { createRoutes } from "../src/routes"
import { startServer } from "./fixture/server"
it.live(
"lists and gets providers without blocking on plugin initialization",
"lists providers without blocking on plugin initialization",
() =>
Effect.gen(function* () {
const tmp = yield* tmpdirScoped("opencode-provider-endpoints-")
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const context = yield* Layer.build(
createRoutes({
password: "secret",
database: { path: ":memory:" },
models: { fetch: false },
fs: { filewatcher: false },
config: {
directory: tmp.path,
project: false,
content: JSON.stringify({
providers: {
custom: {
name: "Configured Custom Provider",
package: "@opencode-ai/ai/providers/openai-compatible",
settings: { apiKey: "secret" },
models: { chat: {} },
},
},
}),
},
}).pipe(Layer.provide(HttpServer.layerServices)),
const fixture = yield* configuredProvider("opencode-provider-list-endpoint-")
const url = new URL("/api/provider", fixture.server.base)
url.searchParams.set("location[directory]", fixture.path)
yield* Effect.promise(async () => {
const response = await fetch(url, { headers: fixture.server.headers })
if (response.status !== 200) return false
const body: unknown = await response.json()
return isRecord(body) && Array.isArray(body["data"])
? body["data"].some((provider) => isRecord(provider) && provider["id"] === "custom")
: false
}).pipe(
Effect.filterOrFail((found) => found),
Effect.retry(Schedule.spaced("10 millis")),
Effect.timeout("2 seconds"),
)
const sdk = Context.get(context, SdkPlugins.Service)
yield* sdk.register(
Plugin.define({
id: "slow-plugin",
effect: () =>
Effect.gen(function* () {
yield* Deferred.succeed(started, undefined)
yield* Deferred.await(release)
}),
}),
)
const handler = Context.get(context, HttpRouter.HttpRouter)
.asHttpEffect()
.pipe(HttpEffect.toWebHandlerWith(context))
const request = (method: "GET" | "POST", route: string) =>
Effect.promise((signal) => {
const url = new URL(route, "http://opencode.local")
url.searchParams.set("location[directory]", tmp.path)
return handler(
new Request(url, {
method,
headers: { authorization: `Basic ${btoa("opencode:secret")}` },
signal,
}),
)
})
const pending = yield* request("POST", "/api/plugin/await-activation").pipe(Effect.forkScoped)
yield* Deferred.await(started)
// Config providers activate after SDK plugins; reads must return the current snapshot without waiting.
const list = yield* request("GET", "/api/provider").pipe(Effect.timeout("2 seconds"))
expect(list.status).toBe(200)
expect(yield* Effect.promise(() => list.json())).toMatchObject({
location: { directory: tmp.path },
data: expect.not.arrayContaining([expect.objectContaining({ id: "custom" })]),
})
const get = yield* request("GET", "/api/provider/custom").pipe(Effect.timeout("2 seconds"))
expect(get.status).toBe(404)
expect(yield* Effect.promise(() => get.json())).toMatchObject({
_tag: "ProviderNotFoundError",
providerID: "custom",
})
expect(pending.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(release, undefined)
expect((yield* Fiber.join(pending)).status).toBe(204)
const provider = {
id: "custom",
name: "Configured Custom Provider",
activation: "enabled",
package: "@opencode-ai/ai/providers/openai-compatible",
settings: { apiKey: "secret" },
}
const configuredList = yield* request("GET", "/api/provider").pipe(Effect.timeout("2 seconds"))
expect(configuredList.status).toBe(200)
expect(yield* Effect.promise(() => configuredList.json())).toMatchObject({
location: { directory: tmp.path },
data: expect.arrayContaining([expect.objectContaining(provider)]),
})
const configuredGet = yield* request("GET", "/api/provider/custom").pipe(Effect.timeout("2 seconds"))
expect(configuredGet.status).toBe(200)
expect(yield* Effect.promise(() => configuredGet.json())).toMatchObject({
location: { directory: tmp.path },
data: provider,
})
}),
15_000,
)
it.live(
"gets providers without blocking on plugin initialization",
() =>
Effect.gen(function* () {
const fixture = yield* configuredProvider("opencode-provider-get-endpoint-")
const url = new URL("/api/provider/custom", fixture.server.base)
url.searchParams.set("location[directory]", fixture.path)
const body: unknown = yield* Effect.tryPromise({
try: async () => {
const response = await fetch(url, { headers: fixture.server.headers })
if (response.status !== 200) throw new Error(`Provider not ready: ${response.status}`)
return response.json()
},
catch: (cause) => cause,
}).pipe(Effect.retry(Schedule.spaced("10 millis")), Effect.timeout("2 seconds"))
if (!isRecord(body) || !isRecord(body["data"])) throw new Error("Expected a provider response")
expect(body["data"]["id"]).toBe("custom")
}),
15_000,
)
const configuredProvider = Effect.fnUntraced(function* (prefix: string) {
const tmp = yield* Effect.acquireDisposable(Effect.promise(() => tmpdir(prefix)))
yield* Effect.promise(() =>
fs.writeFile(
path.join(tmp.path, "opencode.json"),
JSON.stringify({
providers: {
custom: {
package: "@opencode-ai/ai/providers/openai-compatible",
settings: { apiKey: "secret" },
models: { chat: {} },
},
},
}),
),
)
return { server: yield* startServer(tmp.path), path: tmp.path }
})
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
}
-76
View File
@@ -486,82 +486,6 @@
"summary": "List plugins"
}
},
"/api/plugin/await-activation": {
"post": {
"tags": ["plugin"],
"operationId": "v2.plugin.awaitActivation",
"parameters": [
{
"name": "location",
"in": "query",
"schema": {
"anyOf": [
{
"type": "object",
"properties": {
"directory": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
},
"workspace": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
}
},
"additionalProperties": false
},
{
"type": "null"
}
]
},
"required": false,
"style": "deepObject",
"explode": true
}
],
"security": [],
"responses": {
"204": {
"description": "<No Content>"
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
}
},
"description": "Wait for configured plugin activation at a Location to settle, including missing-package installs. Completion does not imply every plugin succeeded or background resource discovery finished. Cancelling this wait does not cancel activation.",
"summary": "Wait for plugin activation"
}
},
"/api/plugin/check": {
"post": {
"tags": ["plugin"],
-76
View File
@@ -486,82 +486,6 @@
"summary": "List plugins"
}
},
"/api/plugin/await-activation": {
"post": {
"tags": ["plugin"],
"operationId": "v2.plugin.awaitActivation",
"parameters": [
{
"name": "location",
"in": "query",
"schema": {
"anyOf": [
{
"type": "object",
"properties": {
"directory": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
},
"workspace": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
]
}
},
"additionalProperties": false
},
{
"type": "null"
}
]
},
"required": false,
"style": "deepObject",
"explode": true
}
],
"security": [],
"responses": {
"204": {
"description": "<No Content>"
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
}
},
"description": "Wait for configured plugin activation at a Location to settle, including missing-package installs. Completion does not imply every plugin succeeded or background resource discovery finished. Cancelling this wait does not cancel activation.",
"summary": "Wait for plugin activation"
}
},
"/api/plugin/check": {
"post": {
"tags": ["plugin"],
@@ -117,10 +117,8 @@ export default Plugin.define({
## Transforms
Transforms are synchronous edits to OpenCode's domain state, and each builds on earlier registrations.
Registry reads such as `ctx.catalog.model.list()` apply pending edits before returning, including during plugin setup;
startup batches update notifications, not read visibility. Resource status APIs still report the state of running
resources: a registered definition does not mean its connection or other resource work has completed.
Transforms are a central pattern in the plugin API and modify how OpenCode works. Plugins register
transforms and each builds on the changes made before it.
Say we have a plugin that adds one model to the catalog.
@@ -136,16 +134,10 @@ export default Plugin.define({
model.cost = [{ input: 2, output: 12, cache: { read: 0.2, write: 2 } }]
})
})
const models = await ctx.catalog.model.list() // Includes the model registered above.
},
})
```
Any registration, removal, or `reload()` marks the registry changed; the next read rebuilds it by replaying every
active transform in registration order onto a fresh value. Keep transforms cheap and repeatable. A value you have
already read is never modified by later rebuilds.
A later plugin can enforce a maximum output price across every model, including models added by earlier plugins.
```ts title="plugins/model-budget/index.ts"
@@ -167,8 +159,8 @@ export default Plugin.define({
})
```
Captured inputs are not watched automatically. Load external data before the synchronous callback, then call
`reload()` after those inputs change.
Now say the first plugin dynamically fetches can fetch its model list from a
dynamic source. It can call `reload` when that list changes.
```ts title="plugins/models/index.ts"
import { Plugin } from "@opencode-ai/plugin"
@@ -801,8 +793,9 @@ interface StorageScanResult {
### Tools
Register, update, and remove tools with a synchronous transform, including in Promise plugins. Load external data
before registering or reloading. A later valid registration overrides the same effective tool name.
Register, update, and remove tools with a transform. The callback is synchronous, including in Promise plugins; load external
data before registering or reloading. OpenCode replays active transforms in registration order on a fresh draft.
For the same effective tool name, a later valid registration overrides an earlier one.
```ts
const registration = await ctx.tool.transform((draft) => {
@@ -861,10 +854,9 @@ definition it overrode. Disposal is idempotent, and unloading the plugin also di
await registration.dispose()
```
Each model request captures a stable, executable tool snapshot. Later transforms, reloads, and disposal affect future
snapshots, not the definitions, Code Mode namespace descriptions, or executors already captured. Executors that close
over mutable plugin data still observe that data; capture a value inside the transform when it must remain tied to
that definition.
Each model request captures a tool snapshot. Reload and disposal affect future snapshots, not the definitions or
executors already captured by an existing request. Executors that close over mutable plugin data still observe
that data; capture a value inside the transform when it must remain tied to that definition.
#### Reference