fix(core): preserve observer lifecycle and branch freshness

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

@@ -25,19 +25,15 @@ 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>({
+    const state: State.Interface<Data, Draft> = State.create<Data, Draft>({
       name: "location-watcher-policy",
       initial: () => ({ ignore: [] }),
       draft: (draft) => ({
         add: (ignore) => draft.ignore.push(...ignore),
         list: () => draft.ignore,
       }),
-      finalize: (draft) =>
-        Effect.sync(() => {
-          current = [...draft.list()]
-        }).pipe(Effect.andThen(Effect.forEach(listeners, (listener) => listener(current), { discard: true }))),
+      notify: () => Effect.forEach(listeners, (listener) => listener(state.get().ignore), { discard: true }),
     })
     const observe = Effect.fn("LocationWatcherPolicy.observe")(function* (
       listener: (ignore: readonly string[]) => Effect.Effect<void>,
@@ -56,7 +52,7 @@ const layer = Layer.effect(
     return Service.of({
       transform: state.transform,
       reload: state.reload,
-      current: () => current,
+      current: () => state.get().ignore,
       observe,
     })
   }),

packages/core/src/instruction-discovery.ts

@@ -74,7 +74,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]) {
@@ -402,11 +402,13 @@ const layer = Layer.effect(
           }
 
           yield* Effect.gen(function* () {
-            const implementation = state
-              .get()
-              .integrations.get(attempt.integrationID)
-              ?.implementations.get(attempt.methodID)
-            const persistence = yield* Effect.sync(() => attempt.label ?? implementation?.label?.(exit.value)).pipe(
+            const persistence = yield* Effect.sync(() => {
+              const implementation = state
+                .get()
+                .integrations.get(attempt.integrationID)
+                ?.implementations.get(attempt.methodID)
+              return attempt.label ?? implementation?.label?.(exit.value)
+            }).pipe(
               Effect.flatMap((label) =>
                 createCredential({
                   integrationID: attempt.integrationID,

packages/core/src/mcp/index.ts

@@ -5,7 +5,7 @@ import { McpEvent } from "@opencode-ai/schema/mcp-event"
 import { ephemeral } from "@opencode-ai/schema/event"
 import { createHash } from "node:crypto"
 import { isDeepStrictEqual } from "node:util"
-import { Cause, Context, Effect, Exit, FiberSet, Latch, Layer, Schema, Scope, Stream, Types } from "effect"
+import { Cause, Context, Effect, Exit, FiberSet, Latch, Layer, Schema, Scope, Semaphore, Stream, Types } from "effect"
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
 import { Credential } from "../credential.js"
 import { Bus } from "../bus.js"
@@ -613,8 +613,10 @@ export const layer = (options?: Options) =>
 
       let applied: Map<ServerName, Mcp.ServerConfig> | undefined
       const overrides = new Map<ServerName, Mcp.ServerConfig | false>()
-      const reconcile = Effect.fnUntraced(function* (next: Draft) {
-        const servers = new Map(next.list())
+      const reconcileLock = Semaphore.makeUnsafe(1)
+      const reconcile = Effect.fnUntraced(function* () {
+        if (root.state._tag === "Closed") return
+        const servers = new Map(state.get().servers)
         if (!applied && entries.size === 0) {
           for (const [name, server] of servers) {
             entries.set(name, {
@@ -676,7 +678,7 @@ export const layer = (options?: Options) =>
           Effect.ignore,
         ),
       )
-      const state = State.create<Data, Draft>({
+      const state: State.Interface<Data, Draft> = State.create<Data, Draft>({
         name: "mcp",
         initial: () => ({
           servers: new Map(
@@ -701,7 +703,7 @@ export const layer = (options?: Options) =>
           },
           remove: (server) => draft.servers.delete(ServerName.make(server)),
         }),
-        finalize: reconcile,
+        notify: () => reconcileLock.withPermit(reconcile()),
       })
 
       // Suspend so each await sees current entries; a bare Map iterator is exhausted after one run.

packages/core/src/reference.ts

@@ -26,6 +26,7 @@ export type Info = Reference.Info
 
 type Data = {
   sources: Map<string, Types.DeepMutable<Source>>
+  materialized: Map<string, Info>
 }
 
 type Draft = {
@@ -47,61 +48,71 @@ const layer = Layer.effect(
     const bus = yield* Bus.Service
     const cache = yield* RepositoryCache.Service
     const scope = yield* Scope.Scope
-    const materialized = new Map<string, Info>()
-    const state = State.create<Data, Draft>({
+    const state: State.Interface<Data, Draft> = State.create<Data, Draft>({
       name: "reference",
-      initial: () => ({ sources: new Map() }),
+      initial: () => ({ sources: new Map(), materialized: new Map() }),
       draft: (draft) => ({
         add: (name, source) => draft.sources.set(name, source as Types.DeepMutable<Source>),
         remove: (name) => draft.sources.delete(name),
         list: () => Array.from(draft.sources.entries()) as [string, Source][],
       }),
-      finalize: (draft) =>
-        Effect.gen(function* () {
-          materialized.clear()
-          for (const [name, source] of draft.list()) {
-            if (source.type === "local") {
-              materialized.set(
-                name,
-                Info.make({
-                  name,
-                  path: source.path,
-                  ...(source.description === undefined ? {} : { description: source.description }),
-                  ...(source.hidden === undefined ? {} : { hidden: source.hidden }),
-                  source,
-                }),
-              )
-              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(
+      prepare: (data) => {
+        for (const [name, source] of data.sources) {
+          if (source.type === "local") {
+            data.materialized.set(
               name,
               Info.make({
                 name,
-                path: AbsolutePath.make(Repository.cachePath(global.repos, repository, source.branch)),
+                path: source.path,
                 ...(source.description === undefined ? {} : { description: source.description }),
                 ...(source.hidden === undefined ? {} : { hidden: source.hidden }),
                 source,
               }),
             )
-            yield* cache.ensure({ reference: repository, branch: source.branch, refresh: true }).pipe(
-              Effect.catchCause((cause) =>
-                Effect.logWarning("failed to materialize reference", {
-                  name,
-                  repository: source.repository,
-                  cause,
-                }),
-              ),
-              Effect.forkIn(scope),
-            )
+            continue
+          }
+          const repository = Repository.parse(source.repository)
+          if (!repository || !Repository.isRemote(repository)) continue
+          if (source.branch) {
+            try {
+              Repository.validateBranch(source.branch)
+            } catch {
+              continue
+            }
+          }
+          data.materialized.set(
+            name,
+            Info.make({
+              name,
+              path: AbsolutePath.make(Repository.cachePath(global.repos, repository, source.branch)),
+              ...(source.description === undefined ? {} : { description: source.description }),
+              ...(source.hidden === undefined ? {} : { hidden: source.hidden }),
+              source,
+            }),
+          )
+        }
+      },
+      notify: () =>
+        Effect.gen(function* () {
+          for (const info of state.get().materialized.values()) {
+            const source = info.source
+            if (source.type !== "git") continue
+            yield* cache
+              .ensure({
+                reference: Repository.parseRemote(source.repository),
+                branch: source.branch,
+                refresh: true,
+              })
+              .pipe(
+                Effect.catchCause((cause) =>
+                  Effect.logWarning("failed to materialize reference", {
+                    name: info.name,
+                    repository: source.repository,
+                    cause,
+                  }),
+                ),
+                Effect.forkIn(scope),
+              )
           }
           yield* bus.publish(Reference.Event.Updated, {})
         }),
@@ -111,7 +122,7 @@ const layer = Layer.effect(
       transform: state.transform,
       reload: state.reload,
       list: Effect.fn("Reference.list")(function* () {
-        return Array.from(materialized.values())
+        return Array.from(state.get().materialized.values())
       }),
     })
   }),

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 } from "effect"
 
 /**
  * A replayable transform applied to a draft during reload.
@@ -16,13 +16,14 @@ export interface Registration {
 }
 
 /**
- * Registers and applies a scoped transform. Closing the owning Scope removes
- * the transform and reloads the materialized state.
+ * Registers a scoped transform and invalidates the derived state. Closing the
+ * owning Scope removes the transform. Reads synchronously replay pending changes.
  */
 export type Transform<DraftApi> = (
   transform: TransformCallback<DraftApi>,
 ) => Effect.Effect<Registration, never, Scope.Scope>
 
+/** Invalidates captured inputs immediately and coalesces change notifications. */
 export type Reload = () => Effect.Effect<void>
 
 export interface Transformable<DraftApi> {
@@ -33,7 +34,7 @@ export interface Transformable<DraftApi> {
 type Batch = {
   active: boolean
   readonly flush: boolean
-  readonly reloads: Set<Reload>
+  readonly notifications: Set<Reload>
 }
 
 const CurrentBatch = Context.Reference<Batch | undefined>("@opencode/State/CurrentBatch", {
@@ -41,17 +42,24 @@ const CurrentBatch = Context.Reference<Batch | undefined>("@opencode/State/Curre
 })
 const reloadDebounce = 500
 
-/** flush: false is terminal teardown: states whose transforms are removed stop rebuilding, including pending reloads. */
+/** Batches notifications, not read visibility. flush: false is terminal teardown. */
 export function batch<A, E, R>(effect: Effect.Effect<A, E, R>, options: { readonly flush?: boolean } = {}) {
-  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 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
-  })
+  return Effect.uninterruptibleMask((restore) =>
+    Effect.gen(function* () {
+      const current = yield* CurrentBatch
+      if (current?.active && options.flush !== false) return yield* restore(effect)
+      const batch: Batch = { active: true, flush: options.flush !== false, notifications: new Set() }
+      const exit = yield* restore(effect.pipe(Effect.provideService(CurrentBatch, batch))).pipe(Effect.exit)
+      batch.active = false
+      const notifications = batch.flush
+        ? yield* Effect.forEach(batch.notifications, (notify) => restore(notify()).pipe(Effect.exit))
+        : []
+      // Accepted writes are not rolled back: one failed observer must not hide
+      // the other states' changes, or replace the batch body's failure.
+      yield* Exit.asVoidAll([exit, ...notifications])
+      return yield* exit
+    }),
+  )
 }
 
 export const inherit = Effect.fnUntraced(function* () {
@@ -65,124 +73,117 @@ export interface Options<State, DraftApi> {
   readonly initial: () => State
   /** Wraps mutable state in a domain-specific draft API. */
   readonly draft: MakeDraft<State, DraftApi>
+  /** Synchronously completes derived data after ordered transform replay. */
+  readonly prepare?: (state: State) => void
   /**
-   * Runs after the rebuilt state becomes visible. Update events published here
-   * act as read barriers: subscribers refetching on the event observe the
-   * committed state.
+   * Observes accepted changes outside the read path. Batched writes notify at
+   * batch completion; reloads debounce notifications. Reads never run this hook.
+   * Resource reconciliation owns any coordination it requires.
    */
-  readonly finalize?: (draft: DraftApi) => Effect.Effect<void>
+  readonly notify?: () => Effect.Effect<void>
 }
 
 export interface Interface<State, DraftApi> extends Transformable<DraftApi> {
+  /** Returns the latest accepted state, replaying stale inputs synchronously. */
   readonly get: () => State
 }
 
 export function create<State, DraftApi>(options: Options<State, DraftApi>): Interface<State, DraftApi> {
   let state = options.initial()
-  let transforms: { run: TransformCallback<DraftApi> }[] = []
-  let generation = 0
+  const transforms = new Set<{ run: TransformCallback<DraftApi> }>()
+  let dirty = false
   let requestedAt = 0
   let running = false
   let closed = false
-  let waiters: { generation: number; done: Deferred.Deferred<void> }[] = []
-  const semaphore = Semaphore.makeUnsafe(1)
+  let waiters: Deferred.Deferred<void>[] = []
 
-  const commit = Effect.fn("State.commit")(function* (next: State) {
-    state = next
-    if (options.finalize) yield* options.finalize(options.draft(next))
-  })
-
-  const materialize = Effect.fnUntraced(function* () {
-    if (closed) return
+  const get = () => {
+    if (!dirty || closed) return state
     const next = options.initial()
     const api = options.draft(next)
-    for (const transform of transforms) {
-      yield* Effect.sync(() => {
-        transform.run(api)
-      })
-    }
-    yield* commit(next)
+    transforms.forEach((transform) => transform.run(api))
+    options.prepare?.(next)
+    state = next
+    dirty = false
+    return state
+  }
+
+  const notify = Effect.fn("State.notify")(function* () {
+    if (closed) return
+    get()
+    if (options.notify) yield* options.notify()
   })
 
-  const materializeReload = () => semaphore.withPermit(materialize())
-
-  const rebuild = (): Effect.Effect<void> =>
+  const publish = (): 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()
+      if (clock.currentTimeMillisUnsafe() < requestedAt + reloadDebounce) return yield* publish()
 
-      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), {
+      // Release scheduling ownership before observers run: an observer may
+      // request and await another reload without joining this notification.
+      const completed = waiters
+      waiters = []
+      running = false
+      const exit = yield* notify().pipe(Effect.exit)
+      yield* Effect.forEach(completed, (done) => Deferred.done(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)
-  })
+  const changed = (debounce: boolean) =>
+    Effect.uninterruptibleMask((restore) =>
+      Effect.gen(function* () {
+        if (closed) return
+        if (debounce) dirty = true
+        const batch = yield* CurrentBatch
+        if (batch?.active) {
+          if (!batch.flush) {
+            closed = true
+            return
+          }
+          batch.notifications.add(notify)
+          return
+        }
+        if (!debounce) return yield* restore(notify())
+
+        const done = Deferred.makeUnsafe<void>()
+        const clock = yield* Clock.Clock
+        requestedAt = clock.currentTimeMillisUnsafe()
+        waiters.push(done)
+        if (!running) {
+          running = true
+          yield* publish().pipe(Effect.forkDetach)
+        }
+        yield* restore(Deferred.await(done))
+      }),
+    )
 
   return {
-    get: () => state,
+    get,
     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(
-            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]
+            Effect.suspend(() => {
+              if (!transforms.delete(transform)) return Effect.void
+              dirty = true
+              return changed(false)
             }),
           )
+          transforms.add(transform)
+          dirty = true
           yield* Scope.addFinalizer(scope, dispose)
-          const batch = yield* CurrentBatch
-          if (batch?.active) batch.reloads.add(materializeReload)
-          else yield* materializeReload()
+          yield* changed(false)
           return { dispose }
         }),
       )
     }),
-    reload,
+    reload: () => changed(true),
   }
 }

packages/core/src/tool.ts

@@ -185,7 +185,7 @@ const layer = Layer.effect(
           draft.tools.delete(id)
         },
       }),
-      finalize: () =>
+      notify: () =>
         Effect.forEach(
           state.get().errors,
           ({ tool, error }) =>

packages/core/src/vcs.ts

@@ -1,7 +1,7 @@
 export * as Vcs from "./vcs.js"
 
 import path from "path"
-import { Cause, Context, Effect, Layer, Schema, Stream } from "effect"
+import { Cause, Context, Effect, Layer, Schema, Semaphore, 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"
@@ -49,6 +49,7 @@ const layer = Layer.effect(
     const bus = yield* Bus.Service
     const vcs = location.vcs
     const current: { info: Info } = { info: { branch: {} } }
+    const refreshLock = Semaphore.makeUnsafe(1)
     const scope = {
       directory: location.directory,
       worktree: location.project.directory,
@@ -69,7 +70,7 @@ const layer = Layer.effect(
           set: (selection) => (draft.selection = selection),
         },
       }),
-      finalize: () => refresh(),
+      notify: () => refresh(),
     })
     const selected = () => {
       const value = state.get()
@@ -87,13 +88,23 @@ const layer = Layer.effect(
         ),
       )
     const refresh = Effect.fn("Vcs.refresh")(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
-      if (changed) yield* bus.publish(VcsEvent.BranchUpdated, { branch: next.branch.current })
+      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(refreshLock.withPermit)
+      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
+      }
     })
 
     if (vcs) {

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/catalog.test.ts

@@ -11,6 +11,7 @@ import { Location } from "@opencode-ai/core/location"
 import { Model } from "@opencode-ai/core/model"
 import { Provider } from "@opencode-ai/core/provider"
 import { AbsolutePath } from "@opencode-ai/core/schema"
+import { State } from "@opencode-ai/core/state"
 import { location } from "./fixture/location"
 import { testEffect } from "./lib/effect"
 
@@ -30,6 +31,57 @@ const catalogLayer = AppNodeBuilder.build(
 const it = testEffect(catalogLayer)
 
 describe("Catalog", () => {
+  it.effect("reads available and default models inside a batch before publishing", () =>
+    Effect.gen(function* () {
+      const catalog = yield* Catalog.Service
+      const bus = yield* Bus.Service
+      const observed: string[] = []
+      const unsubscribe = yield* bus.listen((event) =>
+        event.type === Catalog.Event.Updated.type
+          ? catalog.model.default().pipe(
+              Effect.map((model) => {
+                observed.push(model?.id ?? "none")
+              }),
+            )
+          : Effect.void,
+      )
+      yield* Effect.addFinalizer(() => unsubscribe)
+      const providerID = Provider.ID.make("test")
+      const old = Model.ID.make("old")
+      const newest = Model.ID.make("new")
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* catalog.transform((draft) => {
+            draft.provider.update(providerID, () => {})
+            draft.model.update(providerID, old, (model) => {
+              model.time.released = 1000
+            })
+            draft.model.update(providerID, newest, (model) => {
+              model.time.released = 2000
+            })
+            draft.model.default.set(providerID, old)
+          })
+          expect((yield* catalog.model.available()).map((model) => model.id)).toEqual([newest, old])
+          expect((yield* catalog.model.default())?.id).toBe(old)
+
+          const overlay = yield* catalog.transform((draft) =>
+            draft.model.update(providerID, old, (model) => {
+              model.enabled = false
+            }),
+          )
+          expect((yield* catalog.model.available()).map((model) => model.id)).toEqual([newest])
+          expect((yield* catalog.model.default())?.id).toBe(newest)
+          yield* overlay.dispose
+          expect((yield* catalog.model.default())?.id).toBe(old)
+          expect(observed).toEqual([])
+        }),
+      )
+
+      expect(observed).toEqual([old])
+    }),
+  )
+
   it.effect("publishes an updated event after catalog changes", () =>
     Effect.gen(function* () {
       const catalog = yield* Catalog.Service
@@ -291,6 +343,7 @@ describe("Catalog", () => {
 
       configured = false
       const reload = yield* catalog.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      expect((yield* catalog.model.default())?.id).toBe(newest)
       yield* TestClock.adjust("500 millis")
       yield* Fiber.join(reload)
       expect((yield* catalog.model.default())?.id).toBe(newest)

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

@@ -0,0 +1,127 @@
+import { describe, expect } from "bun:test"
+import { Effect, Fiber, Scope } from "effect"
+import { TestClock } from "effect/testing"
+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 without notifying observers", () =>
+    Effect.gen(function* () {
+      const policy = yield* LocationWatcherPolicy.Service
+      const observed: string[][] = []
+      yield* policy.observe((ignore) =>
+        Effect.sync(() => {
+          observed.push([...ignore])
+        }),
+      )
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* policy.transform((draft) => draft.add(["base"]))
+          const overlay = yield* policy.transform((draft) => draft.add(["overlay"]))
+          const snapshot = policy.current()
+          expect(snapshot).toEqual(["base", "overlay"])
+          expect(observed).toEqual([])
+
+          yield* overlay.dispose
+          expect(policy.current()).toEqual(["base"])
+          expect(snapshot).toEqual(["base", "overlay"])
+          expect(observed).toEqual([])
+        }),
+      )
+
+      expect(observed).toEqual([["base"]])
+    }),
+  )
+
+  it.effect("reads reloaded patterns before debounced observer reconciliation", () =>
+    Effect.gen(function* () {
+      const policy = yield* LocationWatcherPolicy.Service
+      const observed: string[][] = []
+      let ignore = ["first"]
+      yield* policy.observe((ignore) =>
+        Effect.sync(() => {
+          observed.push([...ignore])
+        }),
+      )
+      yield* policy.transform((draft) => draft.add(ignore))
+      const snapshot = policy.current()
+      observed.length = 0
+
+      ignore = ["second"]
+      const reload = yield* policy.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      expect(policy.current()).toEqual(["second"])
+      expect(snapshot).toEqual(["first"])
+      expect(observed).toEqual([])
+
+      yield* TestClock.adjust("500 millis")
+      yield* Fiber.join(reload)
+      expect(observed).toEqual([["second"]])
+    }),
+  )
+
+  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(["inner"])).pipe(Scope.provide(scope))
+        }),
+      )
+      yield* policy.observe((ignore) =>
+        Effect.sync(() => {
+          observed.push([...ignore])
+        }),
+      )
+
+      yield* policy.transform((draft) => draft.add(["outer"]))
+
+      expect(policy.current()).toEqual(["outer", "inner"])
+      expect(observed).toEqual([
+        ["outer", "inner"],
+        ["outer", "inner"],
+      ])
+    }),
+  )
+
+  it.effect("allows an observer to await a reload and keeps later observers current", () =>
+    Effect.gen(function* () {
+      const policy = yield* LocationWatcherPolicy.Service
+      const observed: string[][] = []
+      let ignore = ["first"]
+      let reentered = false
+      yield* policy.observe(() =>
+        Effect.gen(function* () {
+          if (reentered) return
+          reentered = true
+          ignore = ["second"]
+          yield* policy.reload()
+        }),
+      )
+      yield* policy.observe((ignore) =>
+        Effect.sync(() => {
+          observed.push([...ignore])
+        }),
+      )
+
+      const writer = yield* policy
+        .transform((draft) => draft.add(ignore))
+        .pipe(Effect.forkChild({ startImmediately: true }))
+      expect(policy.current()).toEqual(["second"])
+      expect(observed).toEqual([])
+
+      yield* TestClock.adjust("500 millis")
+      yield* Fiber.join(writer)
+      expect(observed).toEqual([["second"], ["second"]])
+    }),
+  )
+})

packages/core/test/integration-replay.test.ts

@@ -0,0 +1,73 @@
+import { describe, expect } from "bun:test"
+import { Cause, Effect, Exit, Fiber } from "effect"
+import { TestClock } from "effect/testing"
+import { Credential } from "@opencode-ai/core/credential"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { Integration } from "@opencode-ai/core/integration"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { testEffect } from "./lib/effect"
+
+const it = testEffect(AppNodeBuilder.build(LayerNode.group([Integration.node, Credential.node])))
+
+describe("Integration replay", () => {
+  it.effect("fails and closes an OAuth attempt when fresh implementation replay throws", () =>
+    Effect.gen(function* () {
+      const integrations = yield* Integration.Service
+      const credentials = yield* Credential.Service
+      const integrationID = Integration.ID.make("replay-test")
+      const methodID = Integration.MethodID.make("code")
+      const source = { fail: false, closed: false }
+      const failure = new Error("integration transform replay failed")
+      yield* integrations.transform((editor) => {
+        if (source.fail) throw failure
+        editor.method.update({
+          integrationID,
+          method: { id: methodID, type: "oauth", label: "Fixture" },
+          authorize: () =>
+            Effect.addFinalizer(() => Effect.sync(() => (source.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: "dummy-access",
+                      refresh: "dummy-refresh",
+                      expires: Number.MAX_SAFE_INTEGER,
+                    }),
+                  ),
+              }),
+            ),
+        })
+      })
+
+      const attempt = yield* integrations.oauth.connect({ integrationID, methodID, label: "Fixture" })
+      source.fail = true
+      const reload = yield* integrations.reload().pipe(Effect.exit, Effect.forkChild({ startImmediately: true }))
+      yield* Effect.addFinalizer(() =>
+        Effect.gen(function* () {
+          source.fail = false
+          yield* TestClock.adjust("500 millis")
+          yield* Fiber.join(reload)
+        }),
+      )
+
+      const exit = yield* integrations.oauth
+        .complete({ integrationID, attemptID: attempt.attemptID, code: "dummy-code" })
+        .pipe(Effect.exit)
+
+      expect(Exit.isFailure(exit) && Cause.hasDies(exit.cause)).toBe(true)
+      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(source.closed).toBe(true)
+      expect(yield* credentials.list(integrationID)).toEqual([])
+    }),
+  )
+})

packages/core/test/integration.test.ts

@@ -7,6 +7,7 @@ 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])))
@@ -262,6 +263,102 @@ describe("Integration", () => {
     }),
   )
 
+  it.effect("resolves stored OAuth with refresh registrations made inside a batch", () =>
+    Effect.gen(function* () {
+      const integrations = yield* Integration.Service
+      const credentials = yield* Credential.Service
+      const integrationID = Integration.ID.make("acme")
+      const method = Integration.OAuthMethod.make({
+        id: Integration.MethodID.make("browser"),
+        type: "oauth",
+        label: "Browser",
+      })
+      const expired = Credential.OAuth.make({
+        type: "oauth",
+        methodID: method.id,
+        access: "expired",
+        refresh: "refresh",
+        expires: 0,
+      })
+      const fresh = Credential.OAuth.make({
+        ...expired,
+        access: "fresh",
+        refresh: "fresh-refresh",
+        expires: (yield* Clock.currentTimeMillis) + Duration.toMillis(Duration.hours(1)),
+      })
+      const stored = yield* credentials.create({ integrationID, label: "Personal", value: expired })
+      const connection = { type: "credential" as const, id: stored.id, label: stored.label }
+      const calls: string[] = []
+      const implementation = {
+        integrationID,
+        method,
+        authorize: () => Effect.die("unexpected authorization"),
+        refresh: (value: Credential.OAuth) =>
+          Effect.sync(() => {
+            expect(value).toEqual(expired)
+            calls.push("original")
+            return fresh
+          }),
+      }
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* integrations.transform((editor) => editor.method.update(implementation))
+          expect(yield* integrations.connection.resolve(connection)).toEqual(fresh)
+          expect((yield* credentials.get(stored.id))?.value).toEqual(fresh)
+          expect(calls).toEqual(["original"])
+
+          expect(yield* integrations.connection.resolve(connection)).toEqual(fresh)
+          expect(calls).toEqual(["original"])
+
+          yield* credentials.update(stored.id, { value: expired })
+          const overridden = Credential.OAuth.make({ ...fresh, access: "override" })
+          const override = yield* integrations.transform((editor) =>
+            editor.method.update({
+              ...implementation,
+              refresh: (value) =>
+                Effect.sync(() => {
+                  expect(value).toEqual(expired)
+                  calls.push("override")
+                  return overridden
+                }),
+            }),
+          )
+          expect(yield* integrations.connection.resolve(connection)).toEqual(overridden)
+          expect((yield* credentials.get(stored.id))?.value).toEqual(overridden)
+
+          yield* override.dispose
+          yield* credentials.update(stored.id, { value: expired })
+          expect(yield* integrations.connection.resolve(connection)).toEqual(fresh)
+          expect(calls).toEqual(["original", "override", "original"])
+
+          yield* credentials.update(stored.id, { value: expired })
+          const removal = yield* integrations.transform((editor) => editor.method.remove(integrationID, method))
+          expect(yield* integrations.connection.resolve(connection)).toEqual(expired)
+          expect((yield* credentials.get(stored.id))?.value).toEqual(expired)
+          expect(calls).toEqual(["original", "override", "original"])
+
+          yield* removal.dispose
+          expect(yield* integrations.connection.resolve(connection)).toEqual(fresh)
+          yield* credentials.update(stored.id, { value: expired })
+          yield* integrations.transform((editor) => editor.method.update({ ...implementation, refresh: undefined }))
+          expect(yield* integrations.connection.resolve(connection)).toEqual(expired)
+          expect((yield* credentials.get(stored.id))?.value).toEqual(expired)
+          expect(calls).toEqual(["original", "override", "original", "original"])
+
+          const failure = new Error("refresh failed")
+          yield* integrations.transform((editor) =>
+            editor.method.update({ ...implementation, refresh: () => Effect.fail(failure) }),
+          )
+          expect(yield* integrations.connection.resolve(connection).pipe(Effect.flip)).toEqual(
+            new Integration.AuthorizationError({ cause: failure }),
+          )
+          expect((yield* credentials.get(stored.id))?.value).toEqual(expired)
+        }),
+      )
+    }),
+  )
+
   it.effect("completes code OAuth once and stores the credential", () =>
     Effect.gen(function* () {
       const integrations = yield* Integration.Service

packages/core/test/mcp.test.ts

@@ -33,9 +33,25 @@ 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 { DateTime, Deferred, Effect, Exit, Fiber, Layer, PubSub, Ref, Schedule, Schema, Sink, Stream } from "effect"
+import {
+  Context,
+  DateTime,
+  Deferred,
+  Effect,
+  Exit,
+  Fiber,
+  Layer,
+  PubSub,
+  Ref,
+  Schedule,
+  Schema,
+  Scope,
+  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"
@@ -66,7 +82,7 @@ function resourceServer(
     listChanged?: boolean
     emptyElicitation?: boolean
     urlElicitation?: boolean
-    respond?: (request: Request) => Response | undefined
+    respond?: (request: Request) => Response | undefined | Promise<Response | undefined>
   } = {},
 ) {
   return Effect.acquireRelease(
@@ -158,7 +174,7 @@ function resourceServer(
           if (typeof body === "object" && body !== null && "method" in body && body.method === "initialize") {
             state.initializations += 1
           }
-          return input.respond?.(request) ?? transport.handleRequest(request)
+          return (await input.respond?.(request)) ?? transport.handleRequest(request)
         },
       })
       return {
@@ -1325,6 +1341,121 @@ test("reconciles only changed MCP server config", async () => {
   )
 })
 
+testEffect(Layer.empty).live("serializes MCP config restoration behind 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
+      expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" })
+      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 })),
+      ),
+    )
+  }),
+)
+
+testEffect(
+  AppNodeBuilder.build(
+    LayerNode.group([Bus.node, Integration.node, Credential.node, Form.node, Environment.node, Location.node]),
+    [
+      [
+        Location.node,
+        Layer.succeed(
+          Location.Service,
+          Location.Service.of(location({ directory: AbsolutePath.make(import.meta.dir) })),
+        ),
+      ],
+      [Environment.node, hostEnvironmentLayer],
+    ],
+  ),
+).effect("discards 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.batch(Scope.close(root, Exit.void), { flush: false })),
+        Effect.andThen(TestClock.adjust("500 millis")),
+      ),
+    )
+    const context = yield* Layer.build(Mcp.layer()).pipe(Scope.provide(root))
+    const service = Context.get(context, Mcp.Service)
+    const observed: string[] = []
+    let block = false
+    const unsubscribe = yield* 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)
+      }),
+    )
+    yield* Effect.addFinalizer(() => 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"
+    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"
+    source.added = true
+    const second = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))
+    yield* TestClock.adjust("500 millis")
+
+    yield* State.batch(Scope.close(root, Exit.void), { flush: false })
+    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(

packages/core/test/plugin.test.ts

@@ -1,10 +1,12 @@
 import { describe, expect } from "bun:test"
 import { ToolFailure } from "@opencode-ai/ai"
-import { Context, Effect, Exit, Fiber, Schema, Stream } from "effect"
+import { Clock, Context, Duration, Effect, Exit, Fiber, Schema, Stream } from "effect"
 import { Plugin as EffectPlugin } from "@opencode-ai/plugin/effect"
 import { Config as ConfigSchema } from "@opencode-ai/schema/config"
 import { Agent } from "@opencode-ai/core/agent"
 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 { PluginHost } from "@opencode-ai/core/plugin/host"
 import { PluginRuntime } from "@opencode-ai/core/plugin/runtime"
@@ -53,6 +55,64 @@ describe("Plugin", () => {
     }),
   )
 
+  it.effect("refreshes its own stored OAuth connection during plugin activation", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      const credentials = yield* Credential.Service
+      const integrationID = Integration.ID.make("acme")
+      const methodID = Integration.MethodID.make("browser")
+      const expired = Credential.OAuth.make({
+        type: "oauth",
+        methodID,
+        access: "expired",
+        refresh: "refresh",
+        expires: 0,
+      })
+      const fresh = Credential.OAuth.make({
+        ...expired,
+        access: "fresh",
+        refresh: "fresh-refresh",
+        expires: (yield* Clock.currentTimeMillis) + Duration.toMillis(Duration.hours(1)),
+      })
+      const stored = yield* credentials.create({ integrationID, label: "Personal", value: expired })
+      const resolved: (Credential.Value | undefined)[] = []
+      const refreshed: Credential.OAuth[] = []
+
+      yield* plugins.activate([
+        versioned(
+          EffectPlugin.define({
+            id: "oauth-refresh",
+            effect: (ctx) =>
+              Effect.gen(function* () {
+                yield* ctx.integration.transform((editor) =>
+                  editor.method.update({
+                    integrationID,
+                    method: { id: methodID, type: "oauth", label: "Browser" },
+                    authorize: () => Effect.die("unexpected authorization"),
+                    refresh: (value) =>
+                      Effect.sync(() => {
+                        refreshed.push(value)
+                        return fresh
+                      }),
+                  }),
+                )
+                const connection = yield* ctx.integration.connection.active(integrationID)
+                if (!connection) return yield* Effect.die("stored connection missing")
+                resolved.push(yield* ctx.integration.connection.resolve(connection).pipe(Effect.orDie))
+              }),
+          }),
+        ),
+      ])
+
+      expect(resolved).toEqual([fresh])
+      expect(refreshed).toEqual([expired])
+      expect((yield* credentials.get(stored.id))?.value).toEqual(fresh)
+      expect(yield* plugins.list()).toEqual([
+        { id: Plugin.ID.make("oauth-refresh"), source: { type: "builtin" }, status: "active", tui: false },
+      ])
+    }),
+  )
+
   it.live("exposes public events through the plugin context", () =>
     Effect.gen(function* () {
       const plugins = yield* Plugin.Service

packages/core/test/reference.test.ts

@@ -2,7 +2,9 @@ import { describe, expect } from "bun:test"
 import { Effect, Exit, Layer, Scope } from "effect"
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Bus } from "@opencode-ai/core/bus"
 import { AbsolutePath } from "@opencode-ai/core/schema"
+import { State } from "@opencode-ai/core/state"
 import { Global } from "@opencode-ai/util/global"
 import { Reference } from "@opencode-ai/core/reference"
 import { Repository } from "@opencode-ai/core/repository"
@@ -12,9 +14,128 @@ import { it } from "./lib/effect"
 const cache = Layer.mock(RepositoryCache.Service, {
   ensure: () => Effect.die("unexpected Git materialization"),
 })
-const referenceLayer = AppNodeBuilder.build(Reference.node, [[RepositoryCache.node, cache]])
+const referenceLayer = AppNodeBuilder.build(LayerNode.group([Reference.node, Bus.node]), [
+  [RepositoryCache.node, cache],
+])
 
 describe("Reference", () => {
+  it.effect("prepares batched references before cache work or update events", () => {
+    const operations: RepositoryCache.EnsureInput[] = []
+    const cache = Layer.mock(RepositoryCache.Service, {
+      ensure: (input) =>
+        Effect.sync(() => {
+          operations.push(input)
+          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, cache],
+    ])
+
+    return Effect.gen(function* () {
+      const references = yield* Reference.Service
+      const bus = yield* Bus.Service
+      const observed: string[][] = []
+      const unsubscribe = yield* bus.listen((event) =>
+        event.type === Reference.Event.Updated.type
+          ? references.list().pipe(
+              Effect.map((infos) => {
+                observed.push(infos.map((info) => info.name))
+              }),
+            )
+          : Effect.void,
+      )
+      yield* Effect.addFinalizer(() => unsubscribe)
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* references.transform((draft) => {
+            draft.add("docs", Reference.LocalSource.make({ type: "local", path: AbsolutePath.make("/docs") }))
+            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,
+          })
+          expect(operations).toEqual([])
+          expect(observed).toEqual([])
+        }),
+      )
+
+      expect(observed).toEqual([["docs", "sdk"]])
+      yield* Effect.yieldNow
+      expect(
+        operations.map((input) => ({
+          repository: input.reference.label,
+          branch: input.branch,
+          refresh: input.refresh,
+        })),
+      ).toEqual([{ repository: "owner/repo", branch: "feature/docs", refresh: true }])
+    }).pipe(Effect.provide(referenceLayer))
+  })
+
+  it.effect("lets update listeners replace references and refetch the latest projection", () =>
+    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.provide(referenceLayer)),
+  )
+
   it.effect("registers normalized sources for the owning scope", () =>
     Effect.gen(function* () {
       const references = yield* Reference.Service

packages/core/test/state.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect } from "bun:test"
 import { State } from "@opencode-ai/core/state"
-import { Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect"
+import { Cause, Deferred, Effect, Exit, Fiber, Layer, Scheduler, Scope } from "effect"
 import { TestClock } from "effect/testing"
 import { testEffect } from "./lib/effect"
 
@@ -15,7 +15,7 @@ describe("State", () => {
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (value: string) => draft.values.push(value) }),
-        finalize: () =>
+        notify: () =>
           block ? Deferred.succeed(rebuilding, undefined).pipe(Effect.andThen(Deferred.await(release))) : Effect.void,
       })
       const scope = yield* Scope.make()
@@ -36,20 +36,20 @@ describe("State", () => {
     }),
   )
 
-  it.effect("commits rebuilt state before finalize runs", () =>
+  it.effect("makes rebuilt state visible before notifying", () =>
     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) }),
-        finalize: () => Effect.sync(() => observed.push([...state.get().values])),
+        notify: () => Effect.sync(() => observed.push([...state.get().values])),
       })
 
       yield* state.transform((draft) => {
         draft.add("value")
       })
 
-      // Update events publish from finalize, so consumers reading on the event
+      // Update events publish from notify, so consumers reading on the event
       // must observe the rebuilt state, not the previous one.
       expect(observed).toEqual([["value"]])
     }),
@@ -76,6 +76,303 @@ describe("State", () => {
     }),
   )
 
+  it.effect("reads registrations and disposals inside a batch without publishing", () =>
+    Effect.gen(function* () {
+      const observed: string[][] = []
+      let replays = 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.sync(() => observed.push([...state.get().values])),
+      })
+      const scope = yield* Scope.make()
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* state
+            .transform((draft) => {
+              replays++
+              draft.add("value")
+            })
+            .pipe(Scope.provide(scope))
+
+          const snapshot = state.get()
+          expect(snapshot.values).toEqual(["value"])
+          expect(state.get()).toBe(snapshot)
+          expect(replays).toBe(1)
+          expect(observed).toEqual([])
+
+          yield* Scope.close(scope, Exit.void)
+          expect(state.get().values).toEqual([])
+          expect(snapshot.values).toEqual(["value"])
+          expect(observed).toEqual([])
+        }),
+      )
+
+      expect(observed).toEqual([[]])
+    }),
+  )
+
+  it.effect("reads a requested reload without waiting for its notification debounce", () =>
+    Effect.gen(function* () {
+      let value = "first"
+      let replays = 0
+      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])),
+      })
+      yield* state.transform((draft) => {
+        replays++
+        draft.add(value)
+      })
+      const snapshot = state.get()
+      observed.length = 0
+
+      value = "second"
+      const reload = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("50 millis")
+
+      expect(state.get().values).toEqual(["second"])
+      expect(snapshot.values).toEqual(["first"])
+      expect(replays).toBe(2)
+      expect(observed).toEqual([])
+
+      yield* TestClock.adjust("450 millis")
+      yield* Fiber.join(reload)
+      expect(observed).toEqual([["second"]])
+      expect(replays).toBe(2)
+    }),
+  )
+
+  it.effect("can await reload inside a batch while deferring its notification", () =>
+    Effect.gen(function* () {
+      let value = "first"
+      let notifications = 0
+      const state = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+        notify: () => Effect.sync(() => notifications++),
+      })
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* state.transform((draft) => draft.add(value))
+          expect(state.get().values).toEqual(["first"])
+          value = "second"
+          yield* state.reload()
+          expect(state.get().values).toEqual(["second"])
+          expect(notifications).toBe(0)
+        }),
+      )
+
+      expect(notifications).toBe(1)
+    }),
+  )
+
+  it.effect("prepares derived data during reads without running observers", () =>
+    Effect.gen(function* () {
+      let notifications = 0
+      const state = State.create({
+        initial: () => ({ values: [] as string[], joined: "" }),
+        draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+        prepare: (data) => {
+          data.joined = data.values.join(",")
+        },
+        notify: () => Effect.sync(() => notifications++),
+      })
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* state.transform((draft) => draft.add("first"))
+          yield* state.transform((draft) => draft.add("second"))
+          expect(state.get().joined).toBe("first,second")
+          expect(notifications).toBe(0)
+        }),
+      )
+
+      expect(notifications).toBe(1)
+    }),
+  )
+
+  it.effect("keeps replay failures observable without replacing the previous snapshot", () =>
+    Effect.gen(function* () {
+      let fail = false
+      const state = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+        prepare: () => {
+          if (fail) throw new Error("preparation failed")
+        },
+      })
+      yield* state.transform((draft) => draft.add("first"))
+      const snapshot = state.get()
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* state.transform((draft) => draft.add("second"))
+          fail = true
+          expect(() => state.get()).toThrow("preparation failed")
+          expect(() => state.get()).toThrow("preparation failed")
+          expect(snapshot.values).toEqual(["first"])
+          fail = false
+          expect(state.get().values).toEqual(["first", "second"])
+        }),
+      )
+    }),
+  )
+
+  it.effect("allows an observer to await a registration on the same state", () =>
+    Effect.gen(function* () {
+      const scope = yield* Scope.Scope
+      let added = false
+      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.gen(function* () {
+            observed.push([...state.get().values])
+            if (added) return
+            added = true
+            yield* state.transform((draft) => draft.add("second")).pipe(Scope.provide(scope))
+          }),
+      })
+
+      yield* state.transform((draft) => draft.add("first"))
+      expect(observed).toEqual([["first"], ["first", "second"]])
+      expect(state.get().values).toEqual(["first", "second"])
+    }),
+  )
+
+  it.effect("allows a debounced observer to await another reload", () =>
+    Effect.gen(function* () {
+      let value = "first"
+      let reloadAgain = false
+      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.gen(function* () {
+            observed.push([...state.get().values])
+            if (!reloadAgain) return
+            reloadAgain = false
+            value = "third"
+            yield* state.reload()
+          }),
+      })
+      yield* state.transform((draft) => draft.add(value))
+      observed.length = 0
+
+      value = "second"
+      reloadAgain = true
+      const reload = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("1 second")
+      yield* Fiber.join(reload)
+
+      expect(observed).toEqual([["second"], ["third"]])
+      expect(state.get().values).toEqual(["third"])
+    }),
+  )
+
+  it.effect("keeps reload waiters associated with their own notification results", () =>
+    Effect.gen(function* () {
+      const entered = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      let value = "first"
+      let block = false
+      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.gen(function* () {
+            observed.push([...state.get().values])
+            if (!block) return
+            block = false
+            yield* Deferred.succeed(entered, undefined)
+            yield* Deferred.await(release)
+            return yield* Effect.die(new Error("first notification failed"))
+          }),
+      })
+      yield* state.transform((draft) => draft.add(value))
+      observed.length = 0
+
+      value = "second"
+      block = true
+      const first = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("500 millis")
+      yield* Deferred.await(entered)
+
+      value = "third"
+      const second = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      expect(state.get().values).toEqual(["third"])
+      yield* TestClock.adjust("500 millis")
+      yield* Fiber.join(second)
+      expect(first.pollUnsafe()).toBeUndefined()
+      expect(observed).toEqual([["second"], ["third"]])
+
+      yield* Deferred.succeed(release, undefined)
+      const exit = yield* Fiber.await(first)
+      expect(Exit.isFailure(exit)).toBe(true)
+      if (Exit.isFailure(exit)) expect(Cause.pretty(exit.cause)).toContain("first notification failed")
+    }),
+  )
+
+  it.effect("continues publishing when a reload caller is cancelled while scheduling its worker", () =>
+    Effect.gen(function* () {
+      let value = "first"
+      let notifications = 0
+      let interrupted = false
+      const state = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+        notify: () => Effect.sync(() => notifications++),
+      })
+      yield* state.transform((draft) => draft.add(value))
+      notifications = 0
+
+      value = "second"
+      const cancelled = yield* Effect.withFiber((fiber) => {
+        const base = new Scheduler.MixedScheduler("sync")
+        const scheduler: Scheduler.Scheduler = {
+          executionMode: base.executionMode,
+          // Keep the first scheduled task at the detached worker handoff.
+          shouldYield: () => false,
+          makeDispatcher: () => {
+            const dispatcher = base.makeDispatcher()
+            return {
+              scheduleTask: (task, priority) => {
+                if (!interrupted) {
+                  interrupted = true
+                  fiber.interruptUnsafe()
+                }
+                dispatcher.scheduleTask(task, priority)
+              },
+              flush: () => dispatcher.flush(),
+            }
+          },
+        }
+        return state.reload().pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
+      }).pipe(Effect.forkChild({ startImmediately: true }))
+      const exit = yield* Fiber.await(cancelled)
+      expect(interrupted).toBe(true)
+      expect(Exit.hasInterrupts(exit)).toBe(true)
+      expect(state.get().values).toEqual(["second"])
+      yield* TestClock.adjust("500 millis")
+      expect(notifications).toBe(1)
+
+      value = "third"
+      const reload = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("500 millis")
+      yield* Fiber.join(reload)
+      expect(state.get().values).toEqual(["third"])
+      expect(notifications).toBe(2)
+    }),
+  )
+
   it.effect("disposes a transform once and rebuilds remaining state", () =>
     Effect.gen(function* () {
       const state = State.create({
@@ -98,18 +395,18 @@ describe("State", () => {
     }),
   )
 
-  it.effect("batches automatic rebuilds", () =>
+  it.effect("batches notifications", () =>
     Effect.gen(function* () {
       let finalized = 0
       const first = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        finalize: () => Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => finalized++),
       })
       const second = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        finalize: () => Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => finalized++),
       })
 
       yield* State.batch(
@@ -133,14 +430,123 @@ describe("State", () => {
     }),
   )
 
+  it.effect("closes a batched observer's owning scope without losing the body's failure", () =>
+    Effect.gen(function* () {
+      const entered = yield* Deferred.make<void>()
+      const release = yield* Deferred.make<void>()
+      const registrations = yield* Scope.make()
+      const owner = yield* Scope.make()
+      yield* Effect.addFinalizer(() =>
+        Deferred.succeed(release, undefined).pipe(
+          Effect.andThen(Scope.close(owner, Exit.void)),
+          Effect.andThen(State.batch(Scope.close(registrations, Exit.void), { flush: false })),
+        ),
+      )
+      const state = State.create({
+        initial: () => ({}),
+        draft: (draft) => draft,
+        notify: () => Deferred.succeed(entered, undefined).pipe(Effect.andThen(Deferred.await(release))),
+      })
+      const writer = yield* State.batch(
+        state.transform(() => {}).pipe(Scope.provide(registrations), Effect.andThen(Effect.fail("batch body failed"))),
+      ).pipe(Effect.forkIn(owner, { startImmediately: true }))
+      yield* Deferred.await(entered)
+
+      const shutdown = yield* Scope.close(owner, Exit.void).pipe(Effect.forkChild({ startImmediately: true }))
+      yield* TestClock.adjust("1 millis")
+      expect(shutdown.pollUnsafe()).toBeDefined()
+      expect(yield* Deferred.isDone(release)).toBe(false)
+      const exit = yield* Fiber.await(writer)
+      expect(Exit.hasInterrupts(exit)).toBe(true)
+      if (Exit.isFailure(exit)) expect(Cause.pretty(exit.cause)).toContain("batch body failed")
+    }),
+  )
+
+  it.effect("lets batch observers read the other states' accepted changes", () =>
+    Effect.gen(function* () {
+      const observed: string[][] = []
+      const first = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+        notify: () => Effect.sync(() => observed.push([...second.get().values])),
+      })
+      const second = State.create({
+        initial: () => ({ values: [] as string[] }),
+        draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
+      })
+
+      yield* State.batch(
+        Effect.gen(function* () {
+          yield* first.transform((draft) => draft.add("first"))
+          yield* second.transform((draft) => draft.add("second"))
+        }),
+      )
+
+      expect(observed).toEqual([["second"]])
+    }),
+  )
+  ;["replay", "notification"].forEach((failure) =>
+    it.effect(`notifies the other states when a batch ${failure} fails`, () =>
+      Effect.gen(function* () {
+        let fail = true
+        const observed: string[] = []
+        const first = State.create({
+          initial: () => ({}),
+          draft: (draft) => draft,
+          notify: () => Effect.sync(() => observed.push("first")),
+        })
+        const failing = State.create({
+          initial: () => ({}),
+          draft: (draft) => draft,
+          prepare: () => {
+            if (fail && failure === "replay") throw new Error("replay failed")
+          },
+          notify: () =>
+            fail ? Effect.die(new Error("notification failed")) : Effect.sync(() => observed.push("failing")),
+        })
+        const last = State.create({
+          initial: () => ({}),
+          draft: (draft) => draft,
+          notify: () => Effect.sync(() => observed.push("last")),
+        })
+
+        const exit = yield* State.batch(
+          Effect.gen(function* () {
+            yield* first.transform(() => {})
+            yield* failing.transform(() => {})
+            yield* last.transform(() => {})
+            return yield* Effect.die(new Error("batch failed"))
+          }),
+        ).pipe(Effect.exit)
+        fail = false
+
+        expect(observed).toEqual(["first", "last"])
+        expect(Exit.isFailure(exit)).toBe(true)
+        if (Exit.isFailure(exit)) {
+          expect(Cause.pretty(exit.cause)).toContain("batch failed")
+          expect(Cause.pretty(exit.cause)).toContain(`${failure} failed`)
+        }
+
+        const reload = yield* failing.reload().pipe(Effect.forkChild({ startImmediately: true }))
+        yield* TestClock.adjust("500 millis")
+        yield* Fiber.join(reload)
+        expect(observed).toEqual(["first", "last", "failing"])
+      }),
+    ),
+  )
+
   it.effect("discards teardown rebuilds and pending reloads while still running cleanup", () =>
     Effect.gen(function* () {
       let finalized = 0
+      let prepared = 0
       let disposed = 0
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        finalize: () => Effect.sync(() => finalized++),
+        prepare: () => {
+          prepared++
+        },
+        notify: () => Effect.sync(() => finalized++),
       })
       const scope = yield* Scope.make()
       yield* Scope.addFinalizer(
@@ -148,19 +554,25 @@ describe("State", () => {
         Effect.sync(() => disposed++),
       )
       const registration = yield* state.transform((draft) => draft.add("value")).pipe(Scope.provide(scope))
+      const snapshot = state.get()
       expect(finalized).toBe(1)
+      expect(prepared).toBe(1)
 
       const pending = yield* state.reload().pipe(Effect.forkChild({ startImmediately: true }))
       yield* TestClock.adjust("250 millis")
       yield* State.batch(Scope.close(scope, Exit.void), { flush: false })
       expect(disposed).toBe(1)
       expect(finalized).toBe(1)
+      expect(state.get()).toBe(snapshot)
+      expect(prepared).toBe(1)
 
       yield* TestClock.adjust("500 millis")
       yield* Fiber.join(pending)
       yield* registration.dispose
       yield* state.reload()
       expect(finalized).toBe(1)
+      expect(state.get()).toBe(snapshot)
+      expect(prepared).toBe(1)
     }),
   )
 
@@ -170,12 +582,12 @@ describe("State", () => {
       const closing = State.create({
         initial: () => ({}),
         draft: (draft) => draft,
-        finalize: () => Effect.sync(() => finalized.push("closing")),
+        notify: () => Effect.sync(() => finalized.push("closing")),
       })
       const live = State.create({
         initial: () => ({}),
         draft: (draft) => draft,
-        finalize: () => Effect.sync(() => finalized.push("live")),
+        notify: () => Effect.sync(() => finalized.push("live")),
       })
       const scope = yield* Scope.make()
       yield* closing.transform(() => {}).pipe(Scope.provide(scope))
@@ -197,7 +609,7 @@ describe("State", () => {
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        finalize: () => Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => finalized++),
       })
       yield* state.transform((draft) => {
         draft.add("value")

packages/core/test/tool-registry.test.ts

@@ -311,7 +311,7 @@ describe("Tool", () => {
     }),
   )
 
-  it.effect("replays empty sources on reload and keeps advertised snapshots", () =>
+  it.effect("reads refreshed sources before notifications and keeps advertised snapshots", () =>
     Effect.gen(function* () {
       const service = yield* Tool.Service
       let source: Info[] = []
@@ -321,24 +321,24 @@ describe("Tool", () => {
       const tool = { ...constant("first"), name: "echo", options: { codemode: false } }
       source = [tool]
       const first = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))
-      yield* TestClock.adjust("500 millis")
-      yield* Fiber.join(first)
       const advertised = yield* service.snapshot()
       expect((yield* advertised.execute(call("echo"))).output).toEqual({ text: "first" })
+      yield* TestClock.adjust("500 millis")
+      yield* Fiber.join(first)
 
       tool.execute = constant("second").execute
       expect((yield* advertised.execute(call("echo"))).output).toEqual({ text: "first" })
       const second = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      expect((yield* executeTool(service, call("echo"))).output).toEqual({ text: "second" })
       yield* TestClock.adjust("500 millis")
       yield* Fiber.join(second)
-      expect((yield* executeTool(service, call("echo"))).output).toEqual({ text: "second" })
       expect((yield* advertised.execute(call("echo"))).output).toEqual({ text: "first" })
 
       source = []
       const removed = yield* service.reload().pipe(Effect.forkChild({ startImmediately: true }))
+      expect((yield* service.snapshot()).definitions.map((tool) => tool.name)).toEqual(["execute"])
       yield* TestClock.adjust("500 millis")
       yield* Fiber.join(removed)
-      expect((yield* service.snapshot()).definitions.map((tool) => tool.name)).toEqual(["execute"])
       expect((yield* advertised.execute(call("echo"))).output).toEqual({ text: "first" })
     }),
   )
@@ -370,7 +370,7 @@ describe("Tool", () => {
     }),
   )
 
-  it.effect("batches tool publication and suppresses terminal teardown replay", () =>
+  it.effect("batches tool notifications with fresh snapshots and suppresses terminal teardown replay", () =>
     Effect.gen(function* () {
       const service = yield* Tool.Service
       const runs: string[] = []
@@ -386,7 +386,8 @@ describe("Tool", () => {
             draft.add({ ...constant("overlay"), name: "echo", options: { codemode: false } })
           })
           expect(runs).toEqual([])
-          expect((yield* service.snapshot()).definitions.map((tool) => tool.name)).toEqual(["execute"])
+          expect((yield* service.snapshot()).definitions.map((tool) => tool.name)).toEqual(["echo", "execute"])
+          expect(runs).toEqual(["base", "overlay"])
         }).pipe(Scope.provide(scope)),
       )
 
@@ -545,23 +546,32 @@ describe("Tool", () => {
     }),
   )
 
-  it.effect("logs invalid tool definitions without dropping healthy tools", () => {
+  it.effect("compiles healthy tools before notifying invalid definition diagnostics", () => {
     const output: unknown[] = []
     const logger = Logger.map(Logger.formatStructured, (entry) => {
       output.push(entry.message)
     })
     return Effect.gen(function* () {
       const service = yield* Tool.Service
-      yield* service.transform((draft) => {
-        draft.add({ ...make(), name: "healthy", options: { codemode: false } })
-        draft.add({
-          name: "phone_type",
-          input: Schema.Struct({}),
-          execute: () => Effect.succeed({ content: "ok" }),
-          options: { codemode: false },
-        } as unknown as Info)
-        draft.add({ ...make(), name: "codemode" })
-      })
+      const snapshot = yield* State.batch(
+        Effect.gen(function* () {
+          yield* service.transform((draft) => {
+            draft.add({ ...make(), name: "healthy", options: { codemode: false } })
+            draft.add({
+              name: "phone_type",
+              input: Schema.Struct({}),
+              execute: () => Effect.succeed({ content: "ok" }),
+              options: { codemode: false },
+            } as unknown as Info)
+            draft.add({ ...make(), name: "codemode" })
+          })
+          const snapshot = yield* service.snapshot()
+          expect(snapshot.definitions.map((tool) => tool.name)).toEqual(["healthy", "execute"])
+          expect(snapshot.codeModeCatalog?.map((tool) => tool.path)).toEqual(["codemode"])
+          expect(output).toEqual([])
+          return snapshot
+        }),
+      )
 
       expect(output).toEqual([
         [
@@ -573,9 +583,6 @@ describe("Tool", () => {
           },
         ],
       ])
-      const snapshot = yield* service.snapshot()
-      expect(snapshot.definitions.map((tool) => tool.name)).toEqual(["healthy", "execute"])
-      expect(snapshot.codeModeCatalog?.map((tool) => tool.path)).toEqual(["codemode"])
       expect((yield* snapshot.execute(call("phone_type")).pipe(Effect.flip)).message).toBe("Unknown tool: phone_type")
     }).pipe(Effect.provide(Logger.layer([logger])))
   })

packages/core/test/vcs.test.ts

@@ -2,12 +2,13 @@ import { $ } from "bun"
 import { describe, expect } from "bun:test"
 import fs from "fs/promises"
 import path from "path"
-import { Cause, Effect, Exit, Fiber, Layer, Stream } from "effect"
+import { Cause, Deferred, Effect, Exit, Fiber, Layer, Schema, Scope, Stream } from "effect"
 import { LayerNode } from "@opencode-ai/util/effect/layer-node"
 import { AppProcess } from "@opencode-ai/util/process"
 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"
@@ -18,6 +19,8 @@ import { tmpdir } from "./fixture/tmpdir"
 import { it } from "./lib/effect"
 import { host } from "./plugin/host"
 
+const Done = Bus.ephemeral({ type: "test.vcs.done", schema: {} })
+
 const provide = (directory: string, input: { git?: boolean } = {}) =>
   Effect.provide(
     LayerNode.compile(LayerNode.group([Vcs.node, Bus.node, Location.node, AppProcess.node]), [
@@ -209,6 +212,152 @@ describe("Vcs", () => {
     ),
   )
 
+  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.runCollect, 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((yield* Fiber.join(updates)).at(-1)?.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)))
+      }),
+    ),
+  )
+
+  it.live("keeps branch streams current when listeners change the selected provider", () =>
+    withTmp((directory) =>
+      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))
+      }).pipe(provide(directory)),
+    ),
+  )
+
+  it.live("does not roll back branch streams when an older listener finishes late", () =>
+    withTmp((directory) =>
+      Effect.gen(function* () {
+        const vcs = yield* Vcs.Service
+        const bus = yield* Bus.Service
+        const entered = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        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) =>
+          event.type === VcsEvent.BranchUpdated.type &&
+          Schema.decodeUnknownSync(VcsEvent.BranchUpdated.data)(event.data).branch === "older"
+            ? Deferred.succeed(entered, undefined).pipe(Effect.andThen(Deferred.await(release)))
+            : Effect.void,
+        )
+
+        yield* Effect.gen(function* () {
+          const older = yield* vcs
+            .transform((draft) => {
+              draft.add(provider({ info: () => Effect.succeed({ branch: { current: "older" } }) }))
+              draft.default.set("custom")
+            })
+            .pipe(Effect.forkScoped({ startImmediately: true }))
+          yield* Deferred.await(entered)
+          yield* vcs.transform((draft) =>
+            draft.add(provider({ info: () => Effect.succeed({ branch: { current: "newer" } }) })),
+          )
+          expect(older.pollUnsafe()).toBeUndefined()
+          expect((yield* vcs.info()).branch.current).toBe("newer")
+
+          yield* Deferred.succeed(release, undefined)
+          yield* Fiber.join(older)
+          yield* bus.publish(Done, {})
+          const events = (yield* Fiber.join(updates)).filter((event) => event.type === VcsEvent.BranchUpdated.type)
+          expect(events.length).toBeGreaterThanOrEqual(2)
+          expect(events.at(-1)?.data.branch).toBe((yield* vcs.info()).branch.current)
+        }).pipe(Effect.ensuring(Deferred.succeed(release, undefined).pipe(Effect.andThen(unsubscribe))))
+      }).pipe(provide(directory)),
+    ),
+  )
+
   it.live("lists local branches by recent activity", () =>
     withGit((directory) =>
       Effect.gen(function* () {

packages/server/test/event-feed.test.ts

@@ -1,10 +1,17 @@
 import { describe, expect, test } from "bun:test"
 import { Agent } from "@opencode-ai/core/agent"
 import { Bus } from "@opencode-ai/core/bus"
+import { Database } from "@opencode-ai/core/database/database"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { Location } from "@opencode-ai/core/location"
+import { Vcs } from "@opencode-ai/core/vcs"
 import { Credential } from "@opencode-ai/schema/credential"
 import { Event } from "@opencode-ai/schema/event"
 import { IntegrationID } from "@opencode-ai/schema/integration-id"
-import { Deferred, Effect, Exit, Fiber, Option, Schema, Stream } from "effect"
+import { VcsEvent } from "@opencode-ai/schema/vcs-event"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Deferred, Effect, Exit, Fiber, Option, Schema, Scope, Stream } from "effect"
+import { tempLocationLayer } from "../../core/test/fixture/location"
 import { it } from "../../core/test/lib/effect"
 import { EventFeed } from "../src/event-feed"
 
@@ -44,6 +51,59 @@ describe("EventFeed", () => {
     expect(EventFeed.frame(payload)).toBe(`data: ${JSON.stringify(payload)}\n\n`)
   })
 
+  it.effect("delivers the latest VCS branch after an earlier legacy listener reenters", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const vcs = yield* Vcs.Service
+      const scope = yield* Scope.Scope
+      const provider = {
+        id: "fixture",
+        name: "Fixture",
+        info: () => Effect.succeed({ branch: { current: "outer" } }),
+        branches: () => Effect.succeed([]),
+        status: () => Effect.succeed([]),
+        diff: () => Effect.succeed([]),
+      }
+      const unsubscribe = yield* bus.listen((event) =>
+        event.type === VcsEvent.BranchUpdated.type &&
+        Schema.decodeUnknownSync(VcsEvent.BranchUpdated.data)(event.data).branch === "outer"
+          ? vcs
+              .transform((draft) =>
+                draft.add({ ...provider, info: () => Effect.succeed({ branch: { current: "inner" } }) }),
+              )
+              .pipe(Scope.provide(scope), Effect.asVoid)
+          : Effect.void,
+      )
+      const feed = yield* EventFeed.make(bus.listen, {
+        encode: (event) => (event.type === VcsEvent.BranchUpdated.type ? (event.data.branch ?? "none") : event.type),
+      })
+      const stream = yield* feed.subscribe
+      const received = yield* stream.pipe(
+        Stream.takeUntil((frame) => frame === Agent.Event.Updated.type, { excludeLast: true }),
+        Stream.runLast,
+        Effect.forkScoped({ startImmediately: true }),
+      )
+
+      yield* vcs.transform((draft) => {
+        draft.add(provider)
+        draft.default.set(provider.id)
+      })
+      yield* unsubscribe
+      yield* bus.publish(Agent.Event.Updated, {})
+
+      const info = yield* vcs.info()
+      expect(info.branch.current).toBe("inner")
+      expect(Option.getOrUndefined(yield* Fiber.join(received))).toBe(info.branch.current)
+    }).pipe(
+      Effect.provide(
+        AppNodeBuilder.build(LayerNode.group([Vcs.node, Bus.node]), [
+          [Location.node, tempLocationLayer],
+          [Database.node, Database.configured({ path: ":memory:" })],
+        ]),
+      ),
+    ),
+  )
+
   it.effect("encodes once and delivers the same frame to every subscriber", () =>
     Effect.gen(function* () {
       let encodes = 0

packages/www/src/docs/content/build/plugins/index.mdx

@@ -95,10 +95,10 @@ Pass plugin options with the object form in `opencode.json(c)`.
     {
       "package": "./plugins/company.ts",
       "options": {
-        "strict": true
-      }
-    }
-  ]
+        "strict": true,
+      },
+    },
+  ],
 }
 ```
 
@@ -120,6 +120,10 @@ export default Plugin.define({
 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.
 
+Transform callbacks are synchronous and should only edit their draft. Load external data before registering or
+reloading a transform. Registry reads replay pending changes when needed, so definitions registered earlier in
+setup are readable without waiting for all plugins to finish setup. Update notifications are batched separately.
+
 Say we have a plugin that adds one model to the catalog.
 
 ```ts title="plugins/models.ts"
@@ -159,8 +163,7 @@ export default Plugin.define({
 })
 ```
 
-Now say the first plugin dynamically fetches can fetch its model list from a
-dynamic source. It can call `reload` when that list changes.
+Now say the first plugin fetches its model list from a dynamic source. It can call `reload` when that list changes.
 
 ```ts title="plugins/models.ts"
 import { Plugin } from "@opencode-ai/plugin"
@@ -190,7 +193,12 @@ export default Plugin.define({
 })
 ```
 
-`reload` replays every catalog transform in order, so the output-price policy still filters the refreshed models.
+`reload` invalidates the catalog without changing transform order, so the output-price policy still filters the
+refreshed models. Reads replay those changes immediately rather than waiting for the update notification's
+500 ms debounce. If nothing reads the catalog, it is rebuilt before the notification instead.
+
+Reading definitions does not start or await background resource work. MCP connections, Git reference checkouts,
+and cached VCS information retain their own lifecycle and readiness behavior.
 
 ## API
 
@@ -468,14 +476,23 @@ interface IntegrationContext {
     key(input: IntegrationConnectKeyInput, requestOptions?: RequestOptions): Promise<void>
   }
   oauth: {
-    connect(input: IntegrationOauthConnectInput, requestOptions?: RequestOptions): Promise<IntegrationOauthConnectOutput>
+    connect(
+      input: IntegrationOauthConnectInput,
+      requestOptions?: RequestOptions,
+    ): Promise<IntegrationOauthConnectOutput>
     status(input: IntegrationOauthStatusInput, requestOptions?: RequestOptions): Promise<IntegrationOauthStatusOutput>
     complete(input: IntegrationOauthCompleteInput, requestOptions?: RequestOptions): Promise<void>
     cancel(input: IntegrationOauthCancelInput, requestOptions?: RequestOptions): Promise<void>
   }
   command: {
-    connect(input: IntegrationCommandConnectInput, requestOptions?: RequestOptions): Promise<IntegrationCommandConnectOutput>
-    status(input: IntegrationCommandStatusInput, requestOptions?: RequestOptions): Promise<IntegrationCommandStatusOutput>
+    connect(
+      input: IntegrationCommandConnectInput,
+      requestOptions?: RequestOptions,
+    ): Promise<IntegrationCommandConnectOutput>
+    status(
+      input: IntegrationCommandStatusInput,
+      requestOptions?: RequestOptions,
+    ): Promise<IntegrationCommandStatusOutput>
     cancel(input: IntegrationCommandCancelInput, requestOptions?: RequestOptions): Promise<void>
   }
   transform(callback: (draft: IntegrationDraft) => void): Promise<Registration>
@@ -1201,10 +1218,7 @@ await ctx.shell.hook("create.before", (event) => {
 
 ```ts
 interface ShellHookContext {
-  hook(
-    name: "create.before",
-    callback: (event: ShellCreateBefore) => Promise<void> | void,
-  ): Promise<Registration>
+  hook(name: "create.before", callback: (event: ShellCreateBefore) => Promise<void> | void): Promise<Registration>
 }
 
 interface ShellCreateBefore {