fix(core): notify registries after state commits

packages/core/src/agent.ts

@@ -87,7 +87,7 @@ const layer = Layer.effect(
           draft.agents.delete(id)
         },
       }),
-      finalize: () => bus.publish(Agent.Event.Updated, {}).pipe(Effect.asVoid),
+      notify: () => bus.publish(Agent.Event.Updated, {}).pipe(Effect.asVoid),
     })
     const selectable = (agent: Info | undefined) =>
       agent && agent.mode !== "subagent" && !agent.hidden ? agent : undefined

packages/core/src/catalog.ts

@@ -134,7 +134,7 @@ const layer = Layer.effect(
         }
         return result
       },
-      finalize: Effect.fn("Catalog.finalize")(function* () {
+      notify: Effect.fn("Catalog.notify")(function* () {
         yield* bus.publish(Catalog.Event.Updated, {})
       }),
     })

packages/core/src/command.ts

@@ -60,7 +60,7 @@ export const layer = Layer.effect(
       draft: (draft) => ({
         add: (definition) => draft.set(definition.name, definition),
       }),
-      finalize: () => bus.publish(Command.Event.Updated, {}).pipe(Effect.asVoid),
+      notify: () => bus.publish(Command.Event.Updated, {}).pipe(Effect.asVoid),
     })
     const info = (definition: Definition) =>
       Info.make({

packages/core/src/filesystem/location-watcher-policy.ts

@@ -37,7 +37,11 @@ const layer = Layer.effect(
       finalize: (draft) =>
         Effect.sync(() => {
           current = [...draft.list()]
-        }).pipe(Effect.andThen(Effect.forEach(listeners, (listener) => listener(current), { discard: true }))),
+        }),
+      notify: () => {
+        const ignore = current
+        return Effect.forEach(listeners, (listener) => listener(ignore), { discard: true })
+      },
     })
     const observe = Effect.fn("LocationWatcherPolicy.observe")(function* (
       listener: (ignore: readonly string[]) => Effect.Effect<void>,

packages/core/src/instruction-discovery.ts

@@ -72,7 +72,7 @@ export const layer = (options?: Options) =>
             draft.available = false
           },
         }),
-        finalize: () => bus.publish(Event.Updated, {}).pipe(Effect.asVoid),
+        notify: () => bus.publish(Event.Updated, {}).pipe(Effect.asVoid),
       })
 
       const source = (value: ReadonlyArray<File> | Instructions.Unavailable | Instructions.Removed) =>

packages/core/src/integration.ts

@@ -328,7 +328,7 @@ const layer = Layer.effect(
           },
         },
       }),
-      finalize: () => bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
+      notify: () => bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
     })
 
     const createCredential = Effect.fnUntraced(function* (input: Parameters<Credential.Interface["create"]>[0]) {

packages/core/src/reference.ts

@@ -103,8 +103,8 @@ const layer = Layer.effect(
               Effect.forkIn(scope),
             )
           }
-          yield* bus.publish(Reference.Event.Updated, {})
         }),
+      notify: () => bus.publish(Reference.Event.Updated, {}).pipe(Effect.asVoid),
     })
 
     return Service.of({

packages/core/src/skill.ts

@@ -109,7 +109,7 @@ const layer = Layer.effect(
           draft.skills.delete(ID.make(id))
         },
       }),
-      finalize: () => bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid),
+      notify: () => bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid),
     })
 
     return Service.of({

packages/core/src/state.ts

@@ -1,6 +1,6 @@
 export * as State from "./state.js"
 
-import { Clock, Context, Deferred, Effect, Scope, Semaphore } from "effect"
+import { Clock, Context, Deferred, Effect, Exit, Scope, Semaphore } from "effect"
 
 /**
  * A replayable transform applied to a draft during reload.
@@ -30,10 +30,12 @@ export interface Transformable<DraftApi> {
   readonly reload: Reload
 }
 
+type Commit = () => Effect.Effect<Effect.Effect<void>>
+
 type Batch = {
   active: boolean
   readonly flush: boolean
-  readonly reloads: Set<Reload>
+  readonly commits: Set<Commit>
 }
 
 const CurrentBatch = Context.Reference<Batch | undefined>("@opencode/State/CurrentBatch", {
@@ -46,10 +48,13 @@ export function batch<A, E, R>(effect: Effect.Effect<A, E, R>, options: { readon
   return Effect.gen(function* () {
     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 batch: Batch = { active: true, flush: options.flush !== false, commits: 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 })
+    if (batch.flush) {
+      const notifications = yield* Effect.forEach(batch.commits, (commit) => commit())
+      yield* Effect.forEach(notifications, (notify) => notify, { discard: true })
+    }
     return yield* exit
   })
 }
@@ -65,12 +70,10 @@ export interface Options<State, DraftApi> {
   readonly initial: () => State
   /** Wraps mutable state in a domain-specific draft API. */
   readonly draft: MakeDraft<State, DraftApi>
-  /**
-   * Runs after the rebuilt state becomes visible. Update events published here
-   * act as read barriers: subscribers refetching on the event observe the
-   * committed state.
-   */
+  /** Runs after the rebuilt state becomes visible while mutation coordination is held. */
   readonly finalize?: (draft: DraftApi) => Effect.Effect<void>
+  /** Runs after the rebuilt state is finalized and mutation coordination is released. */
+  readonly notify?: (draft: DraftApi) => Effect.Effect<void>
 }
 
 export interface Interface<State, DraftApi> extends Transformable<DraftApi> {
@@ -89,11 +92,14 @@ export function create<State, DraftApi>(options: Options<State, DraftApi>): Inte
 
   const commit = Effect.fn("State.commit")(function* (next: State) {
     state = next
-    if (options.finalize) yield* options.finalize(options.draft(next))
+    const draft = options.draft(next)
+    if (options.finalize) yield* options.finalize(draft)
+    const notify = options.notify
+    return notify ? Effect.suspend(() => notify(draft)) : Effect.void
   })
 
   const materialize = Effect.fnUntraced(function* () {
-    if (closed) return
+    if (closed) return Effect.void
     const next = options.initial()
     const api = options.draft(next)
     for (const transform of transforms) {
@@ -101,10 +107,11 @@ export function create<State, DraftApi>(options: Options<State, DraftApi>): Inte
         transform.run(api)
       })
     }
-    yield* commit(next)
+    return yield* commit(next)
   })
 
-  const materializeReload = () => semaphore.withPermit(materialize())
+  const materializeCommit = () => semaphore.withPermit(materialize())
+  const materializeReload = () => materializeCommit().pipe(Effect.flatMap((notify) => notify))
 
   const rebuild = (): Effect.Effect<void> =>
     Effect.gen(function* () {
@@ -114,15 +121,19 @@ export function create<State, DraftApi>(options: Options<State, DraftApi>): Inte
       if (clock.currentTimeMillisUnsafe() < requestedAt + reloadDebounce) return yield* rebuild()
 
       const target = generation
-      const exit = yield* materializeReload().pipe(Effect.exit)
+      const committed = yield* materializeCommit().pipe(Effect.exit)
       const completed = waiters.filter((waiter) => waiter.generation <= target)
       waiters = waiters.filter((waiter) => waiter.generation > target)
+      running = false
+      if (generation > target) {
+        running = true
+        yield* rebuild().pipe(Effect.forkDetach)
+      }
+      const exit = Exit.isFailure(committed) ? committed : yield* committed.value.pipe(Effect.exit)
       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* () {
@@ -149,26 +160,28 @@ export function create<State, DraftApi>(options: Options<State, DraftApi>): Inte
           const transform = { run: update }
           let active = true
           const dispose = Effect.uninterruptible(
-            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
+            semaphore
+              .withPermit(
+                Effect.suspend(() => {
+                  if (!active) return Effect.succeed(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 Effect.void
+                      }
+                      batch.commits.add(materializeCommit)
+                      return Effect.void
                     }
-                    batch.reloads.add(materializeReload)
-                    return
-                  }
-                  yield* materialize()
-                })
-              }),
-            ),
+                    return yield* materialize()
+                  })
+                }),
+              )
+              .pipe(Effect.flatMap((notify) => notify)),
           )
           yield* semaphore.withPermit(
             Effect.sync(() => {
@@ -177,7 +190,7 @@ export function create<State, DraftApi>(options: Options<State, DraftApi>): Inte
           )
           yield* Scope.addFinalizer(scope, dispose)
           const batch = yield* CurrentBatch
-          if (batch?.active) batch.reloads.add(materializeReload)
+          if (batch?.active) batch.commits.add(materializeCommit)
           else yield* materializeReload()
           return { dispose }
         }),

packages/core/src/websearch.ts

@@ -88,7 +88,7 @@ const layer = Layer.effect(
           set: (selection) => (draft.selection = selection),
         },
       }),
-      finalize: () => bus.publish(WebSearch.Event.Updated, {}).pipe(Effect.asVoid),
+      notify: () => bus.publish(WebSearch.Event.Updated, {}).pipe(Effect.asVoid),
     })
 
     const requireProvider = (providers: Map<ID, ProviderImplementation>, providerID: ID) => {

packages/core/test/command.test.ts

@@ -1,11 +1,13 @@
 import { describe, expect } from "bun:test"
+import { Bus } from "@opencode-ai/core/bus"
 import { Command } from "@opencode-ai/core/command"
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { Session } from "@opencode-ai/schema/session"
-import { Effect } from "effect"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Effect, Scope } from "effect"
 import { testEffect } from "./lib/effect"
 
-const it = testEffect(AppNodeBuilder.build(Command.node))
+const it = testEffect(AppNodeBuilder.build(LayerNode.group([Command.node, Bus.node])))
 
 describe("Command", () => {
   it.effect("registers and executes callback commands", () =>
@@ -45,6 +47,31 @@ describe("Command", () => {
     }),
   )
 
+  it.effect("allows synchronous update listeners to mutate the registry", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const command = yield* Command.Service
+      const scope = yield* Scope.Scope
+      let reentered = false
+      const unsubscribe = yield* bus.listen((event) => {
+        if (event.type !== Command.Event.Updated.type || reentered) return Effect.void
+        reentered = true
+        return command
+          .transform((draft) => {
+            draft.add({ name: "listener", execute: () => Effect.void })
+          })
+          .pipe(Scope.provide(scope), Effect.asVoid)
+      })
+      yield* Effect.addFinalizer(() => unsubscribe)
+
+      yield* command.transform((draft) => {
+        draft.add({ name: "source", execute: () => Effect.void })
+      })
+
+      expect((yield* command.list()).map((item) => item.name)).toEqual(["source", "listener"])
+    }),
+  )
+
   it.effect("returns callback error messages without stack traces", () =>
     Effect.gen(function* () {
       const command = yield* Command.Service

packages/core/test/state.test.ts

@@ -78,9 +78,11 @@ describe("State", () => {
 
   it.effect("disposes a transform once and rebuilds remaining state", () =>
     Effect.gen(function* () {
+      let notified = 0
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+        notify: () => Effect.sync(() => notified++),
       })
       yield* state.transform((editor) => {
         editor.add("first")
@@ -92,9 +94,11 @@ describe("State", () => {
 
       yield* registration.dispose
       expect(state.get().values).toEqual(["first"])
+      expect(notified).toBe(3)
 
       yield* registration.dispose
       expect(state.get().values).toEqual(["first"])
+      expect(notified).toBe(3)
     }),
   )
 
@@ -191,29 +195,137 @@ describe("State", () => {
     }),
   )
 
+  it.effect("commits every batched state before notifying", () =>
+    Effect.gen(function* () {
+      type Registry = State.Interface<
+        { values: string[] },
+        { add: (item: string) => void; list: () => readonly string[] }
+      >
+      const observed: string[][] = []
+      const first: Registry = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({
+          add: (item: string) => draft.values.push(item),
+          list: () => draft.values,
+        }),
+        notify: (draft) => Effect.sync(() => observed.push([...draft.list(), ...second.get().values])),
+      })
+      const second: Registry = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({
+          add: (item: string) => draft.values.push(item),
+          list: () => draft.values,
+        }),
+        notify: (draft) => Effect.sync(() => observed.push([...first.get().values, ...draft.list()])),
+      })
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* first.transform((draft) => {
+            draft.add("first")
+          })
+          yield* second.transform((draft) => {
+            draft.add("second")
+          })
+        }),
+      )
+
+      expect(observed).toEqual([
+        ["first", "second"],
+        ["first", "second"],
+      ])
+    }),
+  )
+
   it.effect("debounces reload bursts", () =>
     Effect.gen(function* () {
-      let finalized = 0
+      let notified = 0
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        finalize: () => Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => notified++),
       })
       yield* state.transform((draft) => {
         draft.add("value")
       })
-      finalized = 0
+      notified = 0
 
       const first = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
       yield* TestClock.adjust("250 millis")
       const second = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
       yield* TestClock.adjust("499 millis")
-      expect(finalized).toBe(0)
+      expect(notified).toBe(0)
       yield* TestClock.adjust("1 millis")
       yield* Fiber.join(first)
       yield* Fiber.join(second)
 
-      expect(finalized).toBe(1)
+      expect(notified).toBe(1)
+    }),
+  )
+
+  it.effect("allows debounced notifications to synchronously reload", () =>
+    Effect.gen(function* () {
+      let reenter = false
+      let notified = 0
+      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.gen(function* () {
+            notified++
+            if (!reenter) return
+            reenter = false
+            yield* state.reload()
+          }),
+      })
+      yield* state.transform((draft) => {
+        draft.add("value")
+      })
+      notified = 0
+      reenter = true
+
+      const reload = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("500 millis")
+      yield* TestClock.adjust("500 millis")
+      yield* Fiber.join(reload)
+
+      expect(notified).toBe(2)
+    }),
+  )
+
+  it.effect("settles each reload after its own notification", () =>
+    Effect.gen(function* () {
+      const firstStarted = yield* Deferred.make<void>()
+      const releaseFirst = yield* Deferred.make<void>()
+      let block = false
+      let notified = 0
+      const state = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+        notify: () =>
+          Effect.gen(function* () {
+            notified++
+            if (!block || notified !== 1) return
+            yield* Deferred.succeed(firstStarted, undefined)
+            yield* Deferred.await(releaseFirst)
+          }),
+      })
+      yield* state.transform((draft) => {
+        draft.add("value")
+      })
+      notified = 0
+      block = true
+
+      const first = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("500 millis")
+      yield* Deferred.await(firstStarted)
+      const second = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("500 millis")
+      yield* Fiber.join(second)
+
+      expect(first.pollUnsafe()).toBeUndefined()
+      yield* Deferred.succeed(releaseFirst, undefined)
+      yield* Fiber.join(first)
     }),
   )
 })