vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-09-05 02:24:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

feat(core): disable plugins after transform failures (#47083)

Browse source 
Disable failed registration groups, rebuild healthy state, and report plugin failures with safe diagnostic references. Keep cleanup outside activation locks and preserve disabled revisions across unrelated reloads. Cover deferred hook and RPC cleanup with real-service regression tests.
This commit is contained in:
Kit Langton 2026-09-03 12:38:11 -04:00 • committed by GitHub
parent c370a1bdd0
commit f40ecefdef
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 918 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

167
packages/core/src/plugin.ts
View file

@ -5,7 +5,7 @@ import { Plugin } from "@opencode-ai/schema/plugin"
import { Node } from "@opencode-ai/util/effect/app-node"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import type { PersistentPty } from "./persistent-pty.js"
import { Cause, Context, Effect, Exit, Latch, Layer, Logger, References, Scope, Semaphore } from "effect"
import { Cause, Context, Effect, Exit, Latch, Layer, Logger, Queue, References, Scope, Semaphore } from "effect"
import { Bus } from "./bus.js"
import { KV } from "./kv.js"
import { PluginHost } from "./plugin/host.js"
@ -26,43 +26,64 @@ const layer = Layer.effect(
const lock = Semaphore.makeUnsafe(1)
const ready = yield* Latch.make(true)
const pending = new Set<object>()
const hold = () =>
Effect.sync(() => {
const token = {}
pending.add(token)
ready.closeUnsafe()
return Effect.sync(() => {
if (pending.delete(token) && pending.size === 0) ready.openUnsafe()
})
let closed = false
const holdUnsafe = () => {
if (closed) return Effect.void
const token = {}
pending.add(token)
ready.closeUnsafe()
return Effect.sync(() => {
if (pending.delete(token) && pending.size === 0) ready.openUnsafe()
})
}
const hold = () => Effect.sync(holdUnsafe)
const pendingFailures = yield* Queue.unbounded<PendingFailure>()
let discovered: readonly Failure[] = []
let inventory: Plugin.Info[] = []
const list = Effect.fn("Plugin.list")(function* () {
return inventory
})
const host = yield* PluginHost.make({ list })
const load = Effect.fnUntraced(function* (plugin: Generation) {
const child = yield* Scope.fork(scope)
const activation: Activation = { plugin, scope: yield* Scope.fork(scope) }
const inherit = yield* State.inherit()
const loaded = yield* Effect.suspend(() =>
const grouped = State.group((failure, refresh) => {
activation.failure = {
error: `Plugin disabled after ${failure.state}.transform failed. Check server logs for details.`,
ref: `err_${crypto.randomUUID().slice(0, 8)}`,
}
Queue.offerUnsafe(pendingFailures, {
plugin,
scope: activation.scope,
failure,
refresh,
ref: activation.failure.ref,
release: holdUnsafe(),
})
})
const exit = yield* Effect.suspend(() =>
plugin.effect({ ...host, storage: PluginHost.storage(kv, plugin.id) }),
).pipe(
grouped,
inherit,
Effect.updateContext((context: Context.Context<never>) =>
Context.make(Scope.Scope, child).pipe(
Context.make(Scope.Scope, activation.scope).pipe(
Context.add(Logger.CurrentLoggers, Context.get(context, Logger.CurrentLoggers)),
Context.add(References.MinimumLogLevel, Context.get(context, References.MinimumLogLevel)),
),
),
Effect.withSpan("Plugin.load", { attributes: { "plugin.id": plugin.id } }),
Effect.onExit((exit) => (Exit.isFailure(exit) ? Scope.close(child, exit) : Effect.void)),
Effect.onExit((exit) =>
Exit.isFailure(exit) && !activation.failure ? Scope.close(activation.scope, exit) : Effect.void,
),
Effect.exit,
)
if (Exit.isSuccess(loaded)) return { scope: child } as const
if (activation.failure || Exit.isSuccess(exit)) return { activation } as const
yield* Effect.logWarning("failed to load plugin", {
"plugin.id": plugin.id,
cause: loaded.cause,
cause: exit.cause,
})
return { error: Cause.pretty(loaded.cause) } as const
return { error: Cause.pretty(exit.cause) } as const
})
const activate = Effect.fn("Plugin.activate")(function* (
@ -81,6 +102,8 @@ const layer = Layer.effect(
() =>
lock.withPermit(
Effect.gen(function* () {
if (closed) return
discovered = failures
const current = Array.from(active.values())
const changed = definitions.findIndex((definition, index) => {
const entry = current[index]
@ -108,29 +131,36 @@ const layer = Layer.effect(
([id, slot]) =>
Effect.gen(function* () {
active.delete(id)
if (slot.loaded) yield* Scope.close(slot.loaded.scope, Exit.void)
if (slot.activation && !slot.activation.failure)
yield* Scope.close(slot.activation.scope, Exit.void)
}),
{ discard: true },
)
for (const definition of definitions.slice(prefix)) {
const loaded = yield* load(definition)
if (loaded.scope !== undefined) {
const slot = previous.get(definition.id)
// Reordering healthy registrations does not authorize retrying a failed revision.
if (slot?.activation?.failure && slot.plugin.revision === definition.revision) {
active.set(definition.id, { ...slot, plugin: definition })
continue
}
const result = yield* load(definition)
if (result.activation !== undefined) {
active.set(definition.id, {
plugin: definition,
loaded: { plugin: definition, scope: loaded.scope },
activation: result.activation,
})
continue
}
active.set(definition.id, { plugin: definition, error: loaded.error })
active.set(definition.id, { plugin: definition, error: result.error })
const fallback = previous.get(definition.id)?.loaded
if (!fallback) continue
const fallback = slot?.activation
if (!fallback || fallback.failure) continue
const restored = yield* load(fallback.plugin)
if (restored.scope !== undefined) {
if (restored.activation !== undefined) {
active.set(definition.id, {
plugin: definition,
loaded: { plugin: fallback.plugin, scope: restored.scope },
error: loaded.error,
activation: restored.activation,
error: result.error,
})
continue
}
@ -149,9 +179,68 @@ const layer = Layer.effect(
)
})
yield* Queue.take(pendingFailures).pipe(
Effect.flatMap((item) =>
Effect.gen(function* () {
yield* Effect.logWarning("disabled plugin after transform failure", {
"plugin.id": item.plugin.id,
state: item.failure.state,
ref: item.ref,
cause: Cause.die(item.failure.cause),
})
yield* lock.withPermit(
Effect.gen(function* () {
if (closed) return
// Failure is already recorded on its exact activation, so an old queued item
// cannot disable a replacement and teardown need not wait for this worker.
inventory = [...Array.from(active.values()).map(slotInfo), ...discovered]
const refreshed = yield* State.batch(item.refresh).pipe(Effect.exit)
yield* bus.publish(Plugin.Event.Updated, {})
if (Exit.isFailure(refreshed))
yield* Effect.logWarning("failed to refresh state after disabling plugin", {
"plugin.id": item.plugin.id,
ref: item.ref,
cause: refreshed.cause,
})
}),
)
}).pipe(
// Cleanup must also be scheduled if an inventory observer fails. User finalizers
// may await readiness, so never join them under the activation lock or readiness hold.
Effect.ensuring(
Scope.close(item.scope, Exit.void).pipe(
Effect.catchCause((cause) =>
Effect.logWarning("failed to clean up disabled plugin", {
"plugin.id": item.plugin.id,
ref: item.ref,
cause,
}),
),
Effect.forkScoped({ startImmediately: true }),
),
),
Effect.catchCauseIf(
(cause) => !Cause.hasInterrupts(cause),
(cause) =>
Effect.logError("failed to report disabled plugin", {
"plugin.id": item.plugin.id,
ref: item.ref,
cause,
}),
),
Effect.ensuring(item.release),
),
),
Effect.forever,
Effect.forkScoped,
)
const close = (exit: Exit.Exit<unknown, unknown>) =>
lock.withPermit(
Effect.gen(function* () {
closed = true
pending.clear()
ready.openUnsafe()
active.clear()
yield* State.shutdown(Scope.close(scope, exit))
}),
@ -168,19 +257,37 @@ const layer = Layer.effect(
}),
)
// `plugin` is the definition the slot was last asked to run; `loaded` is the generation actually
// running, which stays an older fallback while the requested revision keeps failing setup.
// `plugin` is the requested definition; `activation` is its last activation, which may have
// failed or be an older fallback while the requested revision keeps failing setup.
type Slot = {
readonly plugin: Generation
readonly loaded?: { readonly plugin: Generation; readonly scope: Scope.Closeable }
readonly activation?: Activation
readonly error?: string
}
// Share the activation across slot snapshots so teardown sees failures synchronously,
// including failures discovered after activate() has captured its previous slots.
type Activation = {
readonly plugin: Generation
readonly scope: Scope.Closeable
failure?: { readonly error: string; readonly ref: string }
}
type PendingFailure = {
readonly plugin: Generation
readonly scope: Scope.Closeable
readonly failure: State.Failure
readonly refresh: Effect.Effect<void>
readonly ref: string
readonly release: Effect.Effect<void>
}
function slotInfo(slot: Slot): Plugin.Info {
const failure = slot.activation?.failure ?? (slot.error === undefined ? undefined : { error: slot.error })
return {
id: Plugin.ID.make(slot.plugin.id),
source: slot.plugin.source ?? { type: "builtin" },
state: slot.error === undefined ? { status: "active" } : { status: "failed", error: slot.error },
state: failure === undefined ? { status: "active" } : { status: "failed", ...failure },
features: { server: true, ...slot.plugin.features },
}
}

105
packages/core/src/state.ts
View file

@ -31,6 +31,49 @@ export interface Transformable<Editor> {
readonly reload: Reload
}
export interface Failure {
readonly state: string
readonly cause: unknown
}
type GroupedRegistration = {
readonly remove: () => boolean
readonly notify: Effect.Effect<void>
}
type RegistrationGroup = {
failed: boolean
readonly registrations: Set<GroupedRegistration>
readonly report: (failure: Failure, refresh: Effect.Effect<void>) => void
}
const CurrentGroup = Context.Reference<RegistrationGroup | undefined>("@opencode/State/CurrentGroup", {
defaultValue: () => undefined,
})
/**
* Groups registrations without coupling State to plugin identity or asynchronous cleanup.
* A failed group is detached synchronously; its supervisor must run refresh and close its scope.
*/
export function group(report: RegistrationGroup["report"]) {
const group: RegistrationGroup = { failed: false, registrations: new Set(), report }
return <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.provideService(effect, CurrentGroup, group)
}
function disable(group: RegistrationGroup, failure: Failure) {
if (group.failed) return
group.failed = true
const notifications = new Set<Effect.Effect<void>>()
for (const registration of group.registrations) {
registration.remove()
notifications.add(registration.notify)
}
group.report(
failure,
Effect.forEach(notifications, (notify) => notify, { discard: true }),
)
}
type Batch = {
active: boolean
readonly shutdown: boolean
@ -112,19 +155,38 @@ export interface Interface<State, Editor> extends Transformable<Editor> {
export function create<State, Editor>(options: Options<State, Editor>): Interface<State, Editor> {
let state = options.initial()
const transforms: { run: TransformCallback<Editor> }[] = []
const transforms = new Set<{ run: TransformCallback<Editor>; group: RegistrationGroup | undefined }>()
let dirty = false
let closed = false
let version = 0
const invalidate = () => {
dirty = true
version++
}
const get = () => {
if (closed || !dirty) return state
const next = options.initial()
const editor = options.editor(next)
for (const transform of transforms) transform.run(editor)
// Only a complete fold becomes visible; a throwing callback leaves the previous value and stays dirty.
state = next
dirty = false
return state
while (true) {
const started = version
const next = options.initial()
const editor = options.editor(next)
for (const transform of transforms) {
try {
transform.run(editor)
} catch (cause) {
if (!transform.group) throw cause
disable(transform.group, { state: options.name ?? "anonymous", cause })
}
// A nested read can disable a group that already contributed to this candidate.
if (version !== started) break
}
if (version !== started) continue
// Ungrouped failures still propagate; grouped failures restart from a fresh candidate.
state = next
dirty = false
return state
}
}
// One stable value per State, so a batch's notification Set holds it at most once.
@ -137,7 +199,7 @@ export function create<State, Editor>(options: Options<State, Editor>): Interfac
const changed = Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
if (closed) return
dirty = true
invalidate()
const batch = yield* CurrentBatch
if (batch?.active) {
if (batch.shutdown) {
@ -156,18 +218,23 @@ export function create<State, Editor>(options: Options<State, Editor>): Interfac
transform: Effect.fn("State.transform")(function* (update) {
yield* Effect.annotateCurrentSpan("state", options.name ?? "anonymous")
const scope = yield* Scope.Scope
const group = yield* CurrentGroup
if (group?.failed) return { dispose: Effect.void }
return yield* Effect.uninterruptible(
Effect.gen(function* () {
const transform = { run: update }
const dispose = Effect.uninterruptible(
Effect.suspend(() => {
const index = transforms.indexOf(transform)
if (index < 0) return Effect.void
transforms.splice(index, 1)
return changed
}),
)
transforms.push(transform)
const transform = { run: update, group }
const registration: GroupedRegistration = {
remove: () => {
if (!transforms.delete(transform)) return false
group?.registrations.delete(registration)
invalidate()
return true
},
notify: changed,
}
const dispose = Effect.uninterruptible(Effect.suspend(() => (registration.remove() ? changed : Effect.void)))
transforms.add(transform)
group?.registrations.add(registration)
yield* Scope.addFinalizer(scope, dispose)
yield* changed
return { dispose }

305
packages/core/test/plugin-failure.test.ts Normal file
View file

@ -0,0 +1,305 @@
import { expect } from "bun:test"
import { Deferred, Effect, Exit, Fiber, Schema } from "effect"
import { Bus } from "@opencode-ai/core/bus"
import { Command } from "@opencode-ai/core/command"
import { Plugin } from "@opencode-ai/core/plugin"
import { PluginHooks } from "@opencode-ai/core/plugin/hooks"
import { Rpc } from "@opencode-ai/core/rpc"
import { testEffect } from "./lib/effect"
import { PluginTestLayer } from "./plugin/fixture"
const it = testEffect(PluginTestLayer)
it.live("removes a failed plugin's hooks and RPC handlers without affecting healthy plugins", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const hooks = yield* PluginHooks.Service
const rpc = yield* Rpc.Service
const cleaned = yield* Deferred.make<void>()
const invoked: string[] = []
let fail = false
yield* plugins.activate(
["broken", "healthy"].map((id) => ({
id,
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
// Finalizers run in reverse order, so this signals after registration cleanup.
if (id === "broken") yield* Effect.addFinalizer(() => Deferred.succeed(cleaned, undefined))
yield* ctx.command.transform((editor) => {
editor.add({ name: id, execute: () => Effect.void })
if (id === "broken" && fail) throw new Error("transform failed")
})
yield* ctx.shell.hook("create.before", () => Effect.sync(() => void invoked.push(id)))
yield* ctx.rpc
.register(
Rpc.define({ id, methods: { check: { input: Schema.Struct({}), output: Schema.String } }, events: {} }),
{ check: () => Effect.succeed(id) },
)
.pipe(Effect.orDie)
if (id === "broken") yield* Effect.addFinalizer(() => plugins.awaitActivation)
}),
})),
)
const trigger = hooks.trigger("shell", "create.before", {
command: "echo fixture",
cwd: ".",
timeout: 1_000,
shell: "sh",
env: {},
})
yield* trigger
expect(invoked).toEqual(["broken", "healthy"])
expect(yield* rpc.call("broken", "check", {})).toBe("broken")
expect(yield* rpc.call("healthy", "check", {})).toBe("healthy")
expect((yield* commands.list()).map((command) => command.name)).toEqual(["broken", "healthy"])
fail = true
yield* commands.reload()
yield* Deferred.await(cleaned).pipe(Effect.timeout("1 second"))
invoked.length = 0
yield* trigger
expect(invoked).toEqual(["healthy"])
expect(yield* rpc.call("broken", "check", {}).pipe(Effect.flip)).toMatchObject({ type: "rpc.unavailable" })
expect(yield* rpc.call("healthy", "check", {})).toBe("healthy")
expect((yield* commands.list()).map((command) => command.name)).toEqual(["healthy"])
expect((yield* plugins.list()).map((plugin) => plugin.state.status)).toEqual(["failed", "active"])
}),
)
Array.of("reload", "teardown").forEach((boundary) =>
it.live(`does not join queued failed-plugin cleanup during ${boundary}`, () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const entered = yield* Deferred.make<void>()
const escape = yield* Deferred.make<void>()
const cleaned = yield* Deferred.make<void>()
let fail = false
yield* plugins.activate([
{
id: "changing",
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
yield* ctx.command.transform((editor) => {
editor.add({ name: "changing", description: "old", execute: () => Effect.void })
if (fail) throw new Error("changing failed")
})
yield* Effect.addFinalizer(() =>
Deferred.succeed(entered, undefined).pipe(
Effect.andThen(plugins.awaitActivation.pipe(Effect.raceFirst(Deferred.await(escape)))),
Effect.andThen(Deferred.succeed(cleaned, undefined)),
),
)
}),
},
{
id: "trigger",
revision: "1",
effect: () =>
Effect.addFinalizer(() =>
boundary === "teardown"
? commands.reload().pipe(Effect.andThen(commands.list()), Effect.asVoid)
: Effect.void,
),
},
])
fail = true
const activation = yield* (boundary === "reload" ? commands.reload() : Effect.void).pipe(
Effect.andThen(
plugins.activate([
{
id: "changing",
revision: "2",
effect: (ctx) =>
ctx.command
.transform((editor) =>
editor.add({ name: "changing", description: "new", execute: () => Effect.void }),
)
.pipe(Effect.asVoid),
},
]),
),
Effect.forkChild({ startImmediately: true }),
)
yield* Deferred.await(entered)
const result = yield* Fiber.join(activation).pipe(Effect.timeout("250 millis"), Effect.exit)
// Allow teardown to finish even if activation incorrectly joins the old finalizer.
yield* Deferred.succeed(escape, undefined)
yield* Fiber.join(activation)
yield* plugins.awaitActivation
yield* Deferred.await(cleaned)
expect(Exit.isSuccess(result)).toBe(true)
expect((yield* plugins.list())[0]?.state).toEqual({ status: "active" })
expect(yield* commands.get("changing")).toMatchObject({ description: "new" })
}),
),
)
it.live("does not restore a disabled generation with a pending failure when its replacement fails setup", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const loads: string[] = []
let fail = false
const generation = (revision: string): Plugin.Generation => ({
id: "replacement",
revision,
effect: (ctx) =>
Effect.gen(function* () {
loads.push(revision)
if (revision === "2") yield* Effect.die("setup failed")
yield* ctx.command.transform((editor) => {
editor.add({ name: "replacement", execute: () => Effect.void })
if (fail) throw new Error("replay failed")
})
}),
})
yield* plugins.activate([generation("1")])
fail = true
yield* commands.reload()
yield* plugins.activate([generation("2")])
yield* plugins.awaitActivation
expect(loads).toEqual(["1", "2"])
expect((yield* plugins.list())[0]?.state.status).toBe("failed")
expect(yield* commands.get("replacement")).toBeUndefined()
}),
)
Array.of("pending", "reported").forEach((status) =>
it.live(`preserves ${status} failures when an earlier plugin changes`, () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const loads: string[] = []
let fail = true
const generation = (id: string, revision: string): Plugin.Generation => ({
id,
revision,
effect: (ctx) =>
Effect.gen(function* () {
loads.push(`${id}@${revision}`)
yield* ctx.command.transform((editor) => {
editor.add({ name: id, execute: () => Effect.void })
if (id === "broken" && fail) throw new Error("broken failed")
})
}),
})
const broken = generation("broken", "1")
const later = generation("later", "1")
yield* plugins.activate([generation("earlier", "1"), broken, later])
if (status === "reported") yield* plugins.awaitActivation
fail = false
yield* plugins.activate([generation("earlier", "2"), broken, later])
yield* plugins.awaitActivation
expect(loads).toEqual(["earlier@1", "broken@1", "later@1", "earlier@2", "later@1"])
const failed = (yield* plugins.list())[1]?.state
expect(failed).toMatchObject({ status: "failed", ref: expect.stringMatching(/^err_/) })
expect((yield* commands.list()).map((entry) => entry.name)).toEqual(["earlier", "later"])
// Reordering and removing other plugins must preserve the same failure too.
yield* plugins.activate([later, broken, generation("earlier", "2")])
yield* plugins.awaitActivation
expect((yield* plugins.list())[1]?.state).toEqual(failed)
expect((yield* commands.list()).map((entry) => entry.name)).toEqual(["later", "earlier"])
yield* plugins.activate([broken, later])
yield* plugins.awaitActivation
expect((yield* plugins.list())[0]?.state).toEqual(failed)
expect(loads.filter((entry) => entry === "broken@1")).toHaveLength(1)
yield* plugins.activate([generation("broken", "2"), later])
yield* plugins.awaitActivation
expect((yield* plugins.list())[0]?.state).toEqual({ status: "active" })
expect(yield* commands.get("broken")).toBeDefined()
expect(loads.filter((entry) => entry === "broken@2")).toHaveLength(1)
}),
),
)
it.live("continues failure reporting and cleanup after a plugin update observer fails", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const bus = yield* Bus.Service
const cleaned: string[] = []
let fail = false
let failPublication = true
yield* plugins.activate(
["first", "second"].map((id) => ({
id,
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
yield* Effect.addFinalizer(() => Effect.sync(() => void cleaned.push(id)))
yield* ctx.command.transform((editor) => {
editor.add({ name: id, execute: () => Effect.void })
if (fail) throw new Error(`${id} failed`)
})
}),
})),
)
yield* Effect.acquireRelease(
bus.listen((event) => {
if (event.type !== Plugin.Event.Updated.type || !failPublication) return Effect.void
failPublication = false
return Effect.die("observer failed")
}),
(unsubscribe) => unsubscribe,
)
fail = true
yield* commands.reload()
const ready = yield* plugins.awaitActivation.pipe(Effect.timeout("250 millis"), Effect.exit)
expect(Exit.isSuccess(ready)).toBe(true)
expect((yield* plugins.list()).map((entry) => entry.state.status)).toEqual(["failed", "failed"])
expect(cleaned.toSorted()).toEqual(["first", "second"])
expect(yield* commands.list()).toEqual([])
}),
)
it.live("settles readiness before shutdown joins a disabled plugin's finalizers", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const entered = yield* Deferred.make<void>()
const escape = yield* Deferred.make<void>()
let fail = false
yield* plugins.activate([
{
id: "closing",
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
yield* ctx.command.transform((editor) => {
editor.add({ name: "closing", execute: () => Effect.void })
if (fail) throw new Error("closing failed")
})
yield* Effect.addFinalizer(() =>
Deferred.succeed(entered, undefined).pipe(
Effect.andThen(plugins.awaitActivation.pipe(Effect.raceFirst(Deferred.await(escape)))),
),
)
}),
},
])
fail = true
const shutdown = yield* commands
.reload()
.pipe(Effect.andThen(plugins.close(Exit.void)), Effect.forkChild({ startImmediately: true }))
yield* Deferred.await(entered)
const result = yield* Fiber.join(shutdown).pipe(Effect.timeout("250 millis"), Effect.exit)
// Release the fixture even on the old implementation, rather than hanging test teardown.
yield* Deferred.succeed(escape, undefined)
yield* Fiber.join(shutdown)
expect(Exit.isSuccess(result)).toBe(true)
const release = yield* plugins.hold()
yield* plugins.awaitActivation
let restarted = false
yield* plugins.activate([
{ id: "after-close", revision: "1", effect: () => Effect.sync(() => void (restarted = true)) },
])
yield* release
expect(restarted).toBe(false)
}),
)

253
packages/core/test/plugin.test.ts
View file

@ -1,12 +1,14 @@
import { expect } from "bun:test"
import path from "path"
import { Clock, Effect } from "effect"
import { Clock, Deferred, Effect } from "effect"
import { TestClock } from "effect/testing"
import { Command } from "@opencode-ai/core/command"
import { Bus } from "@opencode-ai/core/bus"
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 { fromPromise } from "@opencode-ai/plugin/promise/adapter"
import { Session } from "@opencode-ai/schema/session"
import { testEffect } from "./lib/effect"
import { PluginTestLayer } from "./plugin/fixture"
@ -154,6 +156,57 @@ it.effect("reports a failed plugin without blocking a healthy plugin", () =>
}),
)
it.effect("disables a plugin whose transform fails after setup without publishing its partial edits", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const integrations = yield* Integration.Service
let cleaned = false
yield* plugins.activate([
{
id: "before",
revision: "1",
effect: (ctx) =>
ctx.command
.transform((editor) => editor.add({ name: "shared", description: "original", execute: () => Effect.void }))
.pipe(Effect.asVoid),
},
{
id: "broken",
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
yield* Effect.addFinalizer(() => Effect.sync(() => void (cleaned = true)))
yield* ctx.integration.transform((editor) => editor.update("broken", (entry) => (entry.name = "Broken")))
yield* ctx.command.transform((editor) => {
editor.add({ name: "shared", description: "partial", execute: () => Effect.void })
throw new Error("replay failed")
})
}),
},
{
id: "after",
revision: "1",
effect: (ctx) =>
ctx.command
.transform((editor) => editor.add({ name: "healthy", execute: () => Effect.void }))
.pipe(Effect.asVoid),
},
])
yield* plugins.awaitActivation
expect(cleaned).toBe(true)
expect((yield* plugins.list()).find((plugin) => plugin.id === "broken")?.state).toMatchObject({
status: "failed",
error: expect.stringContaining("command.transform failed"),
ref: expect.stringMatching(/^err_/),
})
expect(yield* commands.get("shared")).toMatchObject({ description: "original" })
expect(yield* commands.get("healthy")).toBeDefined()
expect(yield* integrations.get(Integration.ID.make("broken"))).toBeUndefined()
}),
)
it.effect("keeps the suffix after a failed plugin alive across identical activations", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
@ -196,6 +249,204 @@ it.effect("keeps the suffix after a failed plugin alive across identical activat
}),
)
it.effect("attributes replay failure to the broken plugin rather than a later plugin reading the registry", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
yield* plugins.activate([
{
id: "broken-plugin",
revision: "1",
effect: (ctx) =>
ctx.command
.transform(() => {
throw new Error("plugin failed")
})
.pipe(Effect.asVoid),
},
{
id: "reader",
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
yield* ctx.command.list().pipe(Effect.orDie)
yield* ctx.command.transform((editor) => editor.add({ name: "reader", execute: () => Effect.void }))
}),
},
])
yield* plugins.awaitActivation
expect((yield* plugins.list()).map((entry) => `${entry.id}:${entry.state.status}`)).toEqual([
"broken-plugin:failed",
"reader:active",
])
expect(yield* commands.get("reader")).toBeDefined()
}),
)
it.effect("disables plugins after runtime reload failures without retrying an unchanged generation", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const bus = yield* Bus.Service
const reported: string[] = []
yield* Effect.acquireRelease(
bus.listen((event) =>
event.type === Plugin.Event.Updated.type
? plugins.list().pipe(
Effect.tap((items) => Effect.sync(() => void reported.push(items[0]?.state.status ?? "empty"))),
Effect.asVoid,
)
: Effect.void,
),
(unsubscribe) => unsubscribe,
)
let fail = false
let loads = 0
let reload = () => Effect.void
const generation = (revision: string): Plugin.Generation => ({
id: "runtime",
revision,
effect: (ctx) =>
Effect.gen(function* () {
loads++
reload = ctx.command.reload
yield* ctx.command.transform((editor) => {
editor.add({ name: "runtime", execute: () => Effect.void })
if (fail) throw new Error("private failure detail")
})
}),
})
const discovery = {
source: { type: "local" as const, path: "/missing" },
state: { status: "failed" as const, error: "Import failed" },
features: { server: true },
} satisfies Plugin.Info
yield* plugins.activate([generation("1")], [discovery])
expect(yield* commands.get("runtime")).toBeDefined()
fail = true
yield* reload()
yield* plugins.awaitActivation
const inventory = yield* plugins.list()
expect(inventory[0]?.state).toMatchObject({ status: "failed", ref: expect.stringMatching(/^err_/) })
expect(JSON.stringify(inventory[0]?.state)).not.toContain("private failure detail")
expect(reported.at(-1)).toBe("failed")
expect(inventory[1]).toEqual(discovery)
expect(yield* commands.get("runtime")).toBeUndefined()
fail = false
yield* plugins.activate([generation("1")], [discovery])
expect(loads).toBe(1)
expect(yield* commands.get("runtime")).toBeUndefined()
yield* plugins.activate([generation("2")], [discovery])
expect(loads).toBe(2)
expect((yield* plugins.list())[0]?.state).toEqual({ status: "active" })
expect(yield* commands.get("runtime")).toBeDefined()
}),
)
it.effect("disables plugins after replay failures discovered during setup without restoring the old generation", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const cleaned = yield* Deferred.make<void>()
const loads: string[] = []
const generation = (revision: string): Plugin.Generation => ({
id: "replacement",
revision,
effect: (ctx) =>
Effect.gen(function* () {
loads.push(revision)
if (revision === "2")
yield* Effect.addFinalizer(() =>
plugins.awaitActivation.pipe(Effect.andThen(Deferred.succeed(cleaned, undefined))),
)
yield* ctx.command.transform((editor) => {
editor.add({ name: "replacement", execute: () => Effect.void })
if (revision === "2") throw new Error("replay failure")
})
if (revision === "2") {
yield* ctx.command.list().pipe(Effect.orDie)
yield* Effect.die("subsequent setup failure")
}
}),
})
yield* plugins.activate([generation("1")])
yield* plugins.activate([generation("2")])
yield* plugins.awaitActivation
yield* Deferred.await(cleaned)
expect(loads).toEqual(["1", "2"])
expect((yield* plugins.list())[0]?.state).toMatchObject({
status: "failed",
error: expect.stringContaining("command.transform"),
})
expect(yield* commands.get("replacement")).toBeUndefined()
}),
)
it.effect("does not let asynchronous plugin cleanup block recovered registry readiness", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
const cleaned = yield* Deferred.make<void>()
yield* plugins.activate([
{
id: "async-cleanup",
revision: "1",
effect: (ctx) =>
Effect.gen(function* () {
yield* ctx.command.transform(() => {
throw new Error("failed")
})
yield* Effect.addFinalizer(() =>
plugins.awaitActivation.pipe(Effect.andThen(Deferred.succeed(cleaned, undefined))),
)
}),
},
{
id: "healthy",
revision: "1",
effect: (ctx) =>
ctx.command
.transform((editor) => editor.add({ name: "healthy", execute: () => Effect.void }))
.pipe(Effect.asVoid),
},
])
yield* plugins.awaitActivation
yield* Deferred.await(cleaned)
expect(yield* commands.get("healthy")).toBeDefined()
expect((yield* plugins.list())[0]?.state.status).toBe("failed")
}),
)
it.live("retains Promise plugin groups for later registrations and ignores a disabled group's attempts", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service
const commands = yield* Command.Service
let register = async () => {}
const definition = fromPromise({
id: "promise-plugin",
setup(ctx) {
register = async () => {
await ctx.command.transform((editor) => {
editor.add({ name: "late", execute: async () => {} })
throw new Error("late Promise failure")
})
}
},
})
yield* plugins.activate([{ ...definition, revision: "1" }])
yield* Effect.promise(register)
yield* plugins.awaitActivation
expect((yield* plugins.list())[0]?.state).toMatchObject({
status: "failed",
error: expect.stringContaining("command.transform"),
})
expect(yield* commands.get("late")).toBeUndefined()
yield* Effect.promise(register)
expect(yield* commands.get("late")).toBeUndefined()
}),
)
it.effect("reloading a plugin replaces its command implementation", () =>
Effect.gen(function* () {
const plugins = yield* Plugin.Service

138
packages/core/test/state-group.test.ts Normal file
View file

@ -0,0 +1,138 @@
import { expect } from "bun:test"
import { Effect } from "effect"
import { State } from "@opencode-ai/core/state"
import { it } from "./lib/effect"
it.effect("detaches every registration of a failed group and refreshes every affected domain", () =>
Effect.gen(function* () {
const notices: string[] = []
const failures: State.Failure[] = []
let refresh = Effect.void
let fail = false
let calls = 0
const grouped = State.group((failure, changed) => {
failures.push(failure)
refresh = changed
})
const first = State.create({
name: "first",
initial: () => ({ values: [] as string[] }),
editor: (value) => value,
notify: () => Effect.sync(() => void notices.push("first")),
})
const second = State.create({
name: "second",
initial: () => ({ values: [] as string[] }),
editor: (value) => value,
notify: () => Effect.sync(() => void notices.push("second")),
})
yield* first.transform((editor) => editor.values.push("healthy"))
const registration = yield* first.transform((editor) => editor.values.push("grouped")).pipe(grouped)
yield* first.transform((editor) => editor.values.push("also grouped")).pipe(grouped)
yield* second
.transform((editor) => {
calls++
editor.values.push("partial")
if (fail) throw new Error("broken")
})
.pipe(grouped)
const before = first.get()
notices.length = 0
fail = true
yield* second.reload()
expect(first.get().values).toEqual(["healthy"])
expect(second.get().values).toEqual([])
expect(before.values).toEqual(["healthy", "grouped", "also grouped"])
expect(failures).toHaveLength(1)
expect(failures[0]?.state).toBe("second")
expect(calls).toBe(2)
notices.length = 0
// The group deduplicates its domain notifications without relying on an outer batch.
yield* refresh
expect(notices.toSorted()).toEqual(["first", "second"])
yield* registration.dispose
expect(notices).toHaveLength(2)
yield* first.transform((editor) => editor.values.push("resurrected")).pipe(grouped)
expect(first.get().values).toEqual(["healthy"])
expect(failures).toHaveLength(1)
}),
)
it.effect("restarts an outer candidate when a nested read disables one of its contributors", () =>
Effect.gen(function* () {
let fail = false
const grouped = State.group(() => {})
const inner = State.create({ initial: () => ({ value: 0 }), editor: (value) => value })
const outer = State.create({ initial: () => ({ value: 0 }), editor: (value) => value })
yield* outer.transform((editor) => (editor.value += 10)).pipe(grouped)
yield* inner
.transform((editor) => {
editor.value = 5
if (fail) throw new Error("inner failed")
})
.pipe(grouped)
yield* outer.transform((editor) => (editor.value += inner.get().value + 1))
expect(outer.get().value).toBe(16)
fail = true
yield* State.batch(
Effect.gen(function* () {
yield* inner.reload()
yield* outer.reload()
expect(outer.get().value).toBe(1)
expect(inner.get().value).toBe(0)
}),
)
}),
)
it.effect("disables multiple failing groups once each before publishing a complete fold", () =>
Effect.gen(function* () {
const reported: string[] = []
const first = State.group(() => reported.push("first"))
const second = State.group(() => reported.push("second"))
const state = State.create({ initial: () => ({ values: [] as string[] }), editor: (value) => value })
yield* State.batch(
Effect.gen(function* () {
yield* state
.transform((editor) => {
editor.values.push("first")
throw "first failed"
})
.pipe(first)
yield* state
.transform((editor) => {
editor.values.push("second")
throw { message: "second failed" }
})
.pipe(second)
yield* state.transform((editor) => editor.values.push("healthy"))
}),
)
expect(state.get().values).toEqual(["healthy"])
expect(reported).toEqual(["first", "second"])
yield* state.reload()
expect(reported).toEqual(["first", "second"])
}),
)
it.effect("does not disable a group for a notification failure after successful replay", () =>
Effect.gen(function* () {
let reported = 0
let fail = true
const grouped = State.group(() => reported++)
const state = State.create({
initial: () => ({ value: 0 }),
editor: (value) => value,
notify: () => (fail ? Effect.die("observer failed") : Effect.void),
})
yield* state.transform((editor) => editor.value++).pipe(grouped, Effect.exit)
expect(reported).toBe(0)
expect(state.get().value).toBe(1)
fail = false
yield* state.reload()
expect(state.get().value).toBe(1)
}),
)