feat(core): quarantine failed plugin transforms

packages/core/src/plugin.ts

@@ -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,15 +26,19 @@ 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 quarantines = yield* Queue.unbounded<Quarantine>()
+    let discovered: readonly Failure[] = []
     let inventory: Plugin.Info[] = []
     const list = Effect.fn("Plugin.list")(function* () {
       return inventory
@@ -43,9 +47,23 @@ const layer = Layer.effect(
     const load = Effect.fnUntraced(function* (plugin: Generation) {
       const child = yield* Scope.fork(scope)
       const inherit = yield* State.inherit()
+      let failed: Quarantine | undefined
+      const owned = State.owner((failure, refresh) => {
+        failed = {
+          plugin,
+          scope: child,
+          failure,
+          refresh,
+          error: `Plugin disabled after ${failure.state}.transform failed. Check server logs for details.`,
+          ref: `err_${crypto.randomUUID().slice(0, 8)}`,
+          release: holdUnsafe(),
+        }
+        Queue.offerUnsafe(quarantines, failed)
+      })
       const loaded = yield* Effect.suspend(() =>
         plugin.effect({ ...host, storage: PluginHost.storage(kv, plugin.id) }),
       ).pipe(
+        owned,
         inherit,
         Effect.updateContext((context: Context.Context<never>) =>
           Context.make(Scope.Scope, child).pipe(
@@ -54,15 +72,16 @@ const layer = Layer.effect(
           ),
         ),
         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) && !failed ? Scope.close(child, exit) : Effect.void)),
         Effect.exit,
       )
+      if (failed) return { error: failed.error, ref: failed.ref, quarantined: true } as const
       if (Exit.isSuccess(loaded)) return { scope: child } as const
       yield* Effect.logWarning("failed to load plugin", {
         "plugin.id": plugin.id,
         cause: loaded.cause,
       })
-      return { error: Cause.pretty(loaded.cause) } as const
+      return { error: Cause.pretty(loaded.cause), ref: undefined, quarantined: false } as const
     })
 
     const activate = Effect.fn("Plugin.activate")(function* (
@@ -81,6 +100,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]
@@ -121,7 +142,8 @@ const layer = Layer.effect(
                       })
                       continue
                     }
-                    active.set(definition.id, { plugin: definition, error: loaded.error })
+                    active.set(definition.id, { plugin: definition, error: loaded.error, ref: loaded.ref })
+                    if (loaded.quarantined) continue
 
                     const fallback = previous.get(definition.id)?.loaded
                     if (!fallback) continue
@@ -149,9 +171,71 @@ const layer = Layer.effect(
       )
     })
 
+    yield* Queue.take(quarantines).pipe(
+      Effect.flatMap((item) =>
+        Effect.gen(function* () {
+          yield* Effect.logWarning("quarantined 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
+              const id = Plugin.ID.make(item.plugin.id)
+              const slot = active.get(id)
+              // A queued failure from an old generation must never disable its replacement.
+              if (slot?.loaded?.scope === item.scope)
+                active.set(id, { plugin: slot.plugin, error: item.error, ref: item.ref })
+              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 plugin quarantine", {
+                  "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 quarantined 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 quarantined 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))
         }),
@@ -174,13 +258,24 @@ type Slot = {
   readonly plugin: Generation
   readonly loaded?: { readonly plugin: Generation; readonly scope: Scope.Closeable }
   readonly error?: string
+  readonly ref?: string
+}
+
+type Quarantine = {
+  readonly plugin: Generation
+  readonly scope: Scope.Closeable
+  readonly failure: State.Failure
+  readonly refresh: Effect.Effect<void>
+  readonly error: string
+  readonly ref: string
+  readonly release: Effect.Effect<void>
 }
 
 function slotInfo(slot: Slot): Plugin.Info {
   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: slot.error === undefined ? { status: "active" } : { status: "failed", error: slot.error, ref: slot.ref },
     features: { server: true, ...slot.plugin.features },
   }
 }

packages/core/src/state.ts

@@ -31,6 +31,49 @@ export interface Transformable<Editor> {
   readonly reload: Reload
 }
 
+export interface Failure {
+  readonly state: string
+  readonly cause: unknown
+}
+
+type OwnedRegistration = {
+  readonly remove: () => boolean
+  readonly notify: Effect.Effect<void>
+}
+
+type Owner = {
+  failed: boolean
+  readonly registrations: Set<OwnedRegistration>
+  readonly report: (failure: Failure, refresh: Effect.Effect<void>) => void
+}
+
+const CurrentOwner = Context.Reference<Owner | undefined>("@opencode/State/CurrentOwner", {
+  defaultValue: () => undefined,
+})
+
+/**
+ * Groups registrations without coupling State to plugin identity or asynchronous cleanup.
+ * A failed owner is detached synchronously; its supervisor must run refresh and close its scope.
+ */
+export function owner(report: Owner["report"]) {
+  const owner: Owner = { failed: false, registrations: new Set(), report }
+  return <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.provideService(effect, CurrentOwner, owner)
+}
+
+function quarantine(owner: Owner, failure: Failure) {
+  if (owner.failed) return
+  owner.failed = true
+  const notifications = new Set<Effect.Effect<void>>()
+  for (const registration of owner.registrations) {
+    registration.remove()
+    notifications.add(registration.notify)
+  }
+  owner.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: { run: TransformCallback<Editor>; owner: Owner | 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.owner) throw cause
+          quarantine(transform.owner, { state: options.name ?? "anonymous", cause })
+        }
+        // A nested read can quarantine an owner that already contributed to this candidate.
+        if (version !== started) break
+      }
+      if (version !== started) continue
+      // Unowned failures still propagate; owned 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,25 @@ 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 owner = yield* CurrentOwner
+      if (owner?.failed) return { dispose: Effect.void }
       return yield* Effect.uninterruptible(
         Effect.gen(function* () {
-          const transform = { run: update }
-          const dispose = Effect.uninterruptible(
-            Effect.suspend(() => {
+          const transform = { run: update, owner }
+          const registration: OwnedRegistration = {
+            remove: () => {
               const index = transforms.indexOf(transform)
-              if (index < 0) return Effect.void
+              if (index < 0) return false
               transforms.splice(index, 1)
-              return changed
-            }),
-          )
+              owner?.registrations.delete(registration)
+              invalidate()
+              return true
+            },
+            notify: changed,
+          }
+          const dispose = Effect.uninterruptible(Effect.suspend(() => (registration.remove() ? changed : Effect.void)))
           transforms.push(transform)
+          owner?.registrations.add(registration)
           yield* Scope.addFinalizer(scope, dispose)
           yield* changed
           return { dispose }

packages/core/test/plugin-quarantine.test.ts

@@ -0,0 +1,100 @@
+import { expect } from "bun:test"
+import { Deferred, Effect, Exit } from "effect"
+import { Bus } from "@opencode-ai/core/bus"
+import { Command } from "@opencode-ai/core/command"
+import { Plugin } from "@opencode-ai/core/plugin"
+import { testEffect } from "./lib/effect"
+import { PluginTestLayer } from "./plugin/fixture"
+
+const it = testEffect(PluginTestLayer)
+
+it.live("continues quarantine processing 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 quarantined 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>()
+    const closed = 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
+    yield* commands
+      .reload()
+      .pipe(
+        Effect.andThen(plugins.close(Exit.void)),
+        Effect.andThen(Deferred.succeed(closed, undefined)),
+        Effect.forkChild({ startImmediately: true }),
+      )
+    yield* Deferred.await(entered)
+    const result = yield* Deferred.await(closed).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* Deferred.await(closed)
+    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)
+  }),
+)

packages/core/test/plugin.test.ts

@@ -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("quarantines 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 its owner 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-owner",
+        revision: "1",
+        effect: (ctx) =>
+          ctx.command
+            .transform(() => {
+              throw new Error("owner 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-owner:failed",
+      "reader:active",
+    ])
+    expect(yield* commands.get("reader")).toBeDefined()
+  }),
+)
+
+it.effect("quarantines 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("quarantines 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 ownership for later registrations and ignores a quarantined owner's attempts", () =>
+  Effect.gen(function* () {
+    const plugins = yield* Plugin.Service
+    const commands = yield* Command.Service
+    let register = async () => {}
+    const definition = fromPromise({
+      id: "promise-owner",
+      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

packages/core/test/state-owner.test.ts

@@ -0,0 +1,136 @@
+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 owner 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 owned = State.owner((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("owned")).pipe(owned)
+    yield* second
+      .transform((editor) => {
+        calls++
+        editor.values.push("partial")
+        if (fail) throw new Error("broken")
+      })
+      .pipe(owned)
+    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", "owned"])
+    expect(failures).toHaveLength(1)
+    expect(failures[0]?.state).toBe("second")
+    expect(calls).toBe(2)
+
+    notices.length = 0
+    yield* State.batch(refresh)
+    expect(notices.toSorted()).toEqual(["first", "second"])
+    yield* registration.dispose
+    expect(notices).toHaveLength(2)
+    yield* first.transform((editor) => editor.values.push("resurrected")).pipe(owned)
+    expect(first.get().values).toEqual(["healthy"])
+    expect(failures).toHaveLength(1)
+  }),
+)
+
+it.effect("restarts an outer candidate when a nested read quarantines one of its contributors", () =>
+  Effect.gen(function* () {
+    let fail = false
+    const owned = State.owner(() => {})
+    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(owned)
+    yield* inner
+      .transform((editor) => {
+        editor.value = 5
+        if (fail) throw new Error("inner failed")
+      })
+      .pipe(owned)
+    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("quarantines multiple failing owners once each before publishing a complete fold", () =>
+  Effect.gen(function* () {
+    const reported: string[] = []
+    const first = State.owner(() => reported.push("first"))
+    const second = State.owner(() => 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 quarantine an owner for a notification failure after successful replay", () =>
+  Effect.gen(function* () {
+    let reported = 0
+    let fail = true
+    const owned = State.owner(() => 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(owned, Effect.exit)
+    expect(reported).toBe(0)
+    expect(state.get().value).toBe(1)
+    fail = false
+    yield* state.reload()
+    expect(state.get().value).toBe(1)
+  }),
+)