refactor(core): pass the rebuilt value to State notify (#46837)

packages/core/src/agent.ts

@@ -87,7 +87,7 @@ const layer = Layer.effect(
           draft.agents.delete(id)
         },
       }),
-      notify: 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

@@ -135,7 +135,7 @@ const layer = Layer.effect(
         }
         return result
       },
-      notify: bus.publish(Catalog.Event.Updated, {}).pipe(Effect.asVoid, Effect.withSpan("Catalog.notify")),
+      notify: () => bus.publish(Catalog.Event.Updated, {}).pipe(Effect.asVoid, Effect.withSpan("Catalog.notify")),
     })
     const result: Interface = {
       transform: state.transform,

packages/core/src/command.ts

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

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

@@ -33,7 +33,8 @@ const layer = Layer.effect(
         add: (ignore) => draft.ignore.push(...ignore),
         list: () => draft.ignore,
       }),
-      notify: Effect.forEach(listeners, (listener) => listener(current()), { discard: true }),
+      // Read per listener: a reentrant transform inside an earlier listener must reach later ones.
+      notify: () => Effect.forEach(listeners, (listener) => listener(current()), { discard: true }),
     })
     // Annotated to break the inference cycle through notify: notify reads current, current reads state.
     const current = (): readonly string[] => state.get().ignore

packages/core/src/instruction-discovery.ts

@@ -74,7 +74,7 @@ export const layer = (options?: Options) =>
             draft.available = false
           },
         }),
-        notify: 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(
           },
         },
       }),
-      notify: bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
+      notify: () => bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
     })
 
     const createCredential = Effect.fnUntraced(function* (input: Parameters<Credential.Interface["create"]>[0]) {

packages/core/src/mcp/index.ts

@@ -703,7 +703,7 @@ export const layer = (options?: Options) =>
           },
           remove: (server) => draft.servers.delete(ServerName.make(server)),
         }),
-        notify: State.reconcile(root, fork, () => reconcileLock.withPermit(reconcile())),
+        notify: () => State.reconcile(root, fork, () => reconcileLock.withPermit(reconcile())),
       })
 
       // Suspend so each await sees current entries; a bare Map iterator is exhausted after one run.

packages/core/src/reference.ts

@@ -99,10 +99,11 @@ const layer = Layer.effect(
         remove: (name) => draft.sources.delete(name),
         list: () => Array.from(draft.sources),
       }),
-      notify: Effect.gen(function* () {
-        yield* refresh().pipe(Effect.forkIn(scope))
-        yield* bus.publish(Reference.Event.Updated, {})
-      }),
+      notify: () =>
+        Effect.gen(function* () {
+          yield* refresh().pipe(Effect.forkIn(scope))
+          yield* bus.publish(Reference.Event.Updated, {})
+        }),
     })
 
     // Check independently of session activity; the shared cache throttles Git work daily.

packages/core/src/skill.ts

@@ -109,7 +109,7 @@ const layer = Layer.effect(
           draft.skills.delete(ID.make(id))
         },
       }),
-      notify: 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

@@ -73,7 +73,7 @@ function run<A, E, R>(effect: Effect.Effect<A, E, R>, shutdown: boolean) {
 }
 
 /**
- * A `notify` that runs resource reconciliation in the owning layer's FiberSet and awaits it, so work
+ * A `notify` body that runs resource reconciliation in the owning layer's FiberSet and awaits it, so work
  * queued behind the layer's locks is interrupted with the layer. That interruption is not a failure.
  */
 export function reconcile(
@@ -100,11 +100,11 @@ export interface Options<State, DraftApi> {
   /** Wraps mutable state in a domain-specific draft API. */
   readonly draft: MakeDraft<State, DraftApi>
   /**
-   * Observes current state outside the read path. Batched changes notify at
-   * batch completion; reloads debounce notifications. Resource reconciliation
-   * owns its execution scope and coordination.
+   * Observes the freshly rebuilt value outside the read path. Batched changes
+   * notify at batch completion; reloads debounce notifications. Resource
+   * reconciliation owns its execution scope and coordination.
    */
-  readonly notify?: Effect.Effect<void>
+  readonly notify?: (state: State) => Effect.Effect<void>
 }
 
 export interface Interface<State, DraftApi> extends Transformable<DraftApi> {
@@ -137,8 +137,8 @@ export function create<State, DraftApi>(options: Options<State, DraftApi>): Inte
   // One stable value per State, so a batch's notification Set holds it at most once.
   const notify: Effect.Effect<void> = Effect.gen(function* () {
     if (closed) return
-    get()
-    if (options.notify) yield* options.notify
+    const value = get()
+    if (options.notify) yield* options.notify(value)
   }).pipe(Effect.withSpan("State.notify"))
 
   const changed = (debounce: boolean) =>

packages/core/src/tool.ts

@@ -149,7 +149,7 @@ const layer = Layer.effect(
       }
     })
 
-    const state: State.Interface<Data, Draft> = State.create<Data, Draft>({
+    const state = State.create<Data, Draft>({
       name: "tool",
       initial: () => ({
         tools: new Map(),
@@ -196,10 +196,9 @@ const layer = Layer.effect(
           draft.tools.delete(id)
         },
       }),
-      // Read errors when the notification runs, not when the State is created.
-      notify: Effect.suspend(() =>
+      notify: (value) =>
         Effect.forEach(
-          state.get().errors,
+          value.errors,
           ({ kind, name, namespace, error }) =>
             Effect.logError(`Skipping invalid ${kind} registration`, {
               name,
@@ -208,7 +207,6 @@ const layer = Layer.effect(
             }),
           { discard: true },
         ),
-      ),
     })
 
     return Service.of({

packages/core/src/vcs.ts

@@ -81,7 +81,7 @@ const layer = Layer.effect(
           set: (selection) => (draft.selection = selection),
         },
       }),
-      notify: State.reconcile(root, fork, () => refresh()),
+      notify: () => State.reconcile(root, fork, () => refresh()),
     })
     const selected = () => {
       const value = state.get()

packages/core/src/websearch.ts

@@ -88,7 +88,7 @@ const layer = Layer.effect(
           set: (selection) => (draft.selection = selection),
         },
       }),
-      notify: 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/state-replay.test.ts

@@ -48,7 +48,7 @@ describe("State replay properties", () => {
               State.create({
                 initial: (): Value => ({ value: sources[index], order: [] }),
                 draft: (draft) => draft,
-                notify: Effect.sync(() => void notifications[index]++),
+                notify: () => Effect.sync(() => void notifications[index]++),
               }),
             )
             const callbacks = operations.map((operation, index) => (draft: Value) => {

packages/core/test/state.test.ts

@@ -13,9 +13,8 @@ describe("State", () => {
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (value: string) => draft.values.push(value) }),
-        notify: block
-          ? Deferred.succeed(rebuilding, undefined).pipe(Effect.andThen(Deferred.await(release)))
-          : Effect.void,
+        notify: () =>
+          block ? Deferred.succeed(rebuilding, undefined).pipe(Effect.andThen(Deferred.await(release))) : Effect.void,
       })
       const scope = yield* Scope.make()
       const fiber = yield* state
@@ -41,7 +40,7 @@ describe("State", () => {
       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])),
+        notify: () => Effect.sync(() => observed.push([...state.get().values])),
       })
 
       yield* state.transform((draft) => {
@@ -103,12 +102,12 @@ describe("State", () => {
       const first = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        notify: Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => finalized++),
       })
       const second = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        notify: Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => finalized++),
       })
 
       yield* State.batch(
@@ -139,7 +138,7 @@ describe("State", () => {
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        notify: Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => finalized++),
       })
       const scope = yield* Scope.make()
       yield* Scope.addFinalizer(
@@ -169,12 +168,12 @@ describe("State", () => {
       const closing = State.create({
         initial: () => ({}),
         draft: (draft) => draft,
-        notify: Effect.sync(() => finalized.push("closing")),
+        notify: () => Effect.sync(() => finalized.push("closing")),
       })
       const live = State.create({
         initial: () => ({}),
         draft: (draft) => draft,
-        notify: Effect.sync(() => finalized.push("live")),
+        notify: () => Effect.sync(() => finalized.push("live")),
       })
       const scope = yield* Scope.make()
       yield* closing.transform(() => {}).pipe(Scope.provide(scope))
@@ -196,7 +195,7 @@ describe("State", () => {
       const state = State.create({
         initial: () => ({ values: [] as string[] }),
         draft: (draft) => ({ add: (item: string) => draft.values.push(item) }),
-        notify: Effect.sync(() => finalized++),
+        notify: () => Effect.sync(() => finalized++),
       })
       yield* state.transform((draft) => {
         draft.add("value")
@@ -278,10 +277,10 @@ describe("State rebuild", () => {
     Effect.gen(function* () {
       const calls: string[] = []
       const notifications: string[][] = []
-      const state: State.Interface<{ values: string[] }, { values: string[] }> = State.create({
+      const state = State.create({
         initial: () => ({ values: new Array<string>() }),
         draft: (data) => data,
-        notify: Effect.sync(() => notifications.push([...state.get().values])),
+        notify: (value) => Effect.sync(() => notifications.push([...value.values])),
       })
       expect(state.get().values).toEqual([])
       yield* State.batch(Effect.void)
@@ -421,7 +420,7 @@ describe("State rebuild", () => {
       const state = State.create({
         initial: () => ({ value: 0 }),
         draft: (data) => data,
-        notify: Effect.sync(() => notifications++),
+        notify: () => Effect.sync(() => notifications++),
       })
       yield* state.transform((draft) => {
         calls++
@@ -540,17 +539,18 @@ describe("State notification boundaries", () => {
           const first = State.create({
             initial: () => ({ value: 0 }),
             draft: (data) => data,
-            notify: Effect.gen(function* () {
-              observed.push("first")
-              if (phase !== "observer" || !block) return
-              yield* Deferred.succeed(entered, undefined)
-              yield* Effect.never
-            }),
+            notify: () =>
+              Effect.gen(function* () {
+                observed.push("first")
+                if (phase !== "observer" || !block) return
+                yield* Deferred.succeed(entered, undefined)
+                yield* Effect.never
+              }),
           })
           const second = State.create({
             initial: () => ({ value: 0 }),
             draft: (data) => data,
-            notify: Effect.sync(() => observed.push("second")),
+            notify: () => Effect.sync(() => observed.push("second")),
           })
           const writer = yield* State.batch(
             Effect.gen(function* () {
@@ -577,7 +577,7 @@ describe("State notification boundaries", () => {
       const state = State.create({
         initial: () => ({ value: 0 }),
         draft: (data) => data,
-        notify: Effect.sync(() => notifications++),
+        notify: () => Effect.sync(() => notifications++),
       })
       const inherit = yield* State.batch(
         Effect.gen(function* () {
@@ -604,12 +604,13 @@ describe("State notification boundaries", () => {
       const state: State.Interface<object, object> = State.create({
         initial: () => ({}),
         draft: (data) => data,
-        notify: Effect.gen(function* () {
-          observed.push(other.get().value)
-          if (added) return
-          added = true
-          yield* state.transform(() => {}).pipe(Scope.provide(scope))
-        }),
+        notify: () =>
+          Effect.gen(function* () {
+            observed.push(other.get().value)
+            if (added) return
+            added = true
+            yield* state.transform(() => {}).pipe(Scope.provide(scope))
+          }),
       })
       yield* State.batch(
         Effect.gen(function* () {
@@ -629,13 +630,14 @@ describe("State notification boundaries", () => {
       const state: State.Interface<{ value: number }, { value: number }> = State.create({
         initial: () => ({ value: 0 }),
         draft: (data) => data,
-        notify: Effect.gen(function* () {
-          observed.push(state.get().value)
-          if (!reloadAgain) return
-          reloadAgain = false
-          source = 3
-          yield* state.reload()
-        }),
+        notify: () =>
+          Effect.gen(function* () {
+            observed.push(state.get().value)
+            if (!reloadAgain) return
+            reloadAgain = false
+            source = 3
+            yield* state.reload()
+          }),
       })
       yield* state.transform((draft) => (draft.value = source))
       source = 2
@@ -654,12 +656,12 @@ describe("State notification boundaries", () => {
       const first = State.create({
         initial: () => ({}),
         draft: (data) => data,
-        notify: Effect.suspend(() => (fail ? Effect.die("observer failed") : Effect.void)),
+        notify: () => (fail ? Effect.die("observer failed") : Effect.void),
       })
       const second = State.create({
         initial: () => ({}),
         draft: (data) => data,
-        notify: Effect.sync(() => observed.push("second")),
+        notify: () => Effect.sync(() => observed.push("second")),
       })
       const exit = yield* State.batch(
         Effect.gen(function* () {
@@ -685,10 +687,11 @@ describe("State notification boundaries", () => {
       const state = State.create({
         initial: () => ({}),
         draft: (data) => data,
-        notify: Effect.sync(() => {
-          notifications++
-          if (fail) throw new Error("notification failed")
-        }),
+        notify: () =>
+          Effect.sync(() => {
+            notifications++
+            if (fail) throw new Error("notification failed")
+          }),
       })
       yield* state.transform(() => {})
       notifications = 0