diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index 52f8d5901f2..fba9e7ec75e 100644 --- a/packages/core/src/agent.ts +++ b/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 diff --git a/packages/core/src/catalog.ts b/packages/core/src/catalog.ts index c2ce177f1fb..07a7a634ea6 100644 --- a/packages/core/src/catalog.ts +++ b/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, diff --git a/packages/core/src/command.ts b/packages/core/src/command.ts index d72da0826cf..b3a216464ad 100644 --- a/packages/core/src/command.ts +++ b/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({ diff --git a/packages/core/src/filesystem/location-watcher-policy.ts b/packages/core/src/filesystem/location-watcher-policy.ts index fcf4f89e7fd..40f20a76fd1 100644 --- a/packages/core/src/filesystem/location-watcher-policy.ts +++ b/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 diff --git a/packages/core/src/instruction-discovery.ts b/packages/core/src/instruction-discovery.ts index 4e687c6350d..d6bc4d2f9be 100644 --- a/packages/core/src/instruction-discovery.ts +++ b/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 | Instructions.Unavailable | Instructions.Removed) => diff --git a/packages/core/src/integration.ts b/packages/core/src/integration.ts index b7ad2139767..21a9339c16f 100644 --- a/packages/core/src/integration.ts +++ b/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[0]) { diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index d689fcc7417..ef2eb6b1d19 100644 --- a/packages/core/src/mcp/index.ts +++ b/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. diff --git a/packages/core/src/reference.ts b/packages/core/src/reference.ts index f3177157ccd..5e2eb22a75d 100644 --- a/packages/core/src/reference.ts +++ b/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. diff --git a/packages/core/src/skill.ts b/packages/core/src/skill.ts index bc5570a923e..1bcae148fe1 100644 --- a/packages/core/src/skill.ts +++ b/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({ diff --git a/packages/core/src/state.ts b/packages/core/src/state.ts index 1dd56f28b37..27ef20b40f9 100644 --- a/packages/core/src/state.ts +++ b/packages/core/src/state.ts @@ -73,7 +73,7 @@ function run(effect: Effect.Effect, 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 { /** Wraps mutable state in a domain-specific draft API. */ readonly draft: MakeDraft /** - * 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 + readonly notify?: (state: State) => Effect.Effect } export interface Interface extends Transformable { @@ -137,8 +137,8 @@ export function create(options: Options): Inte // One stable value per State, so a batch's notification Set holds it at most once. const notify: Effect.Effect = 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) => diff --git a/packages/core/src/tool.ts b/packages/core/src/tool.ts index 2f6680fded1..9fd0ceae695 100644 --- a/packages/core/src/tool.ts +++ b/packages/core/src/tool.ts @@ -149,7 +149,7 @@ const layer = Layer.effect( } }) - const state: State.Interface = State.create({ + const state = State.create({ 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({ diff --git a/packages/core/src/vcs.ts b/packages/core/src/vcs.ts index d609fccd201..78afb17e27f 100644 --- a/packages/core/src/vcs.ts +++ b/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() diff --git a/packages/core/src/websearch.ts b/packages/core/src/websearch.ts index 226912c33b1..3c5b7ee6032 100644 --- a/packages/core/src/websearch.ts +++ b/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, providerID: ID) => { diff --git a/packages/core/test/state-replay.test.ts b/packages/core/test/state-replay.test.ts index db310f93584..51b630075cd 100644 --- a/packages/core/test/state-replay.test.ts +++ b/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) => { diff --git a/packages/core/test/state.test.ts b/packages/core/test/state.test.ts index a5fcbe8802e..ab893d77bd4 100644 --- a/packages/core/test/state.test.ts +++ b/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() }), 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 = 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