fix(settings): close third-review watcher-lifecycle and write-boundary gaps

Review round three found four seam defects:

- A watch() disposer only removed the observer from the set; an invocation
  already chained onto the watcher tail still ran after disposal. Watchers
  now carry an active flag checked when a queued invocation would start,
  and the service dispose drain awaits started invocations (pendingTails)
  beside the write queues, so disposal is quiescent.
- The settings/updated manual fan-out caught only synchronous throws; an
  async listener rejection escaped as an unhandled rejection. Thenable
  returns are now contained through the shared listener diagnostic, and
  the event contract documents that the INVARIANT rethrow serves
  synchronous listeners only.
- structuredClone admitted Dates, Maps, BigInts, and cycles that YAML/JSON
  storage silently distorts on reload (a Date lands as a timestamp string,
  a Map as a plain map, a BigInt as a number). The write snapshot is now a
  single-pass cloneJsonShaped walk that rejects non-JSON values with their
  path before anything persists.
- mergeLayers' per-entry undefined guard became dead code once the clone
  strips undefined entries at the boundary; removed, with the sparse-patch
  contract restated at its enforcement point.

packages/settings/settings/src/index.ts

@@ -60,20 +60,24 @@ export interface SettingsScope<T> {
   /**
    * Observe committed changes to this namespace's resolved value. Invocations
    * of one callback run asynchronously, one at a time, in commit order; a
-   * rejection is contained and logged like a sync throw.
+   * rejection is contained and logged like a sync throw. After the disposer
+   * returns, no further invocation starts — one already queued is skipped;
+   * one already started still settles, and service disposal waits for it.
    * @param callback - invoked after each commit with the next and previous values.
    * @returns the disposer removing this observer.
    */
   watch(callback: (next: T, prev: T) => void | Promise<void>): () => void
   /**
    * Merge a partial patch into this namespace's user layer and persist it.
-   * @param patch - plain-object patch over the user section.
+   * @param patch - plain-object patch over the user section; JSON-shaped data
+   * only (non-JSON values reject with their path before anything persists).
    */
   update(patch: object): Promise<void>
   /**
    * Replace this namespace's user section wholesale; absent keys re-inherit
    * the composition `base` and schema defaults (`replace({})` resets all).
-   * @param section - the complete next user section.
+   * @param section - the complete next user section; JSON-shaped data only,
+   * as for {@link update}.
    */
   replace(section: object): Promise<void>
 }
@@ -88,6 +92,11 @@ declare module 'cordis' {
      * Committed change to one registered namespace's resolved value. Emitted
      * after the provider persisted (for `update`) or published (`provider`)
      * the change; never emitted when the resolved value is deep-equal.
+     * Listener failures are contained and logged — a sync throw and an async
+     * rejection alike — except `INVARIANT`-coded failures, which rethrow
+     * after every listener ran; that rethrow reaches the emitter only from
+     * synchronous listeners, so invariant checks on this event must not be
+     * async functions.
      * @param ns - the namespace whose resolved value changed.
      * @param next - the new resolved value.
      * @param prev - the previous resolved value.
@@ -127,17 +136,76 @@ function isPlainObject(value: unknown): value is Record<string, unknown> {
   return proto === Object.prototype || proto === null
 }
 
+/** Human label for a value rejected by the JSON-shape boundary (numbers reject inline). */
+function describeRejected(value: unknown): string {
+  if (value === undefined) return 'undefined'
+  if (typeof value === 'object' && value !== null) {
+    const proto = Object.getPrototypeOf(value) as { constructor?: { name?: string } } | null
+    const name = proto?.constructor?.name
+    return name === undefined || name === 'Object' ? 'a non-plain object' : `a ${name}`
+  }
+  return `a ${typeof value}`
+}
+
+/**
+ * Detach one write input in a single walk that doubles as the durable-boundary
+ * shape check: only JSON data (plain objects, arrays, strings, finite numbers,
+ * booleans, `null`) may reach a provider document. `structuredClone` alone
+ * would admit Dates, Maps, BigInts, and cycles that YAML/JSON storage then
+ * silently distorts on the reload round-trip. `undefined` entries in objects
+ * are skipped — the same sparse-patch semantics as {@link mergeLayers} — while
+ * an `undefined` array entry is rejected rather than coerced.
+ * @param root - plain-object write input (caller-checked).
+ * @param reject - builds the boundary error from a value label and its `$`-rooted path.
+ * @returns the detached JSON-shaped clone.
+ */
+function cloneJsonShaped(
+  root: Record<string, unknown>,
+  reject: (label: string, path: string) => TypeError,
+): Record<string, unknown> {
+  const visiting = new WeakSet<object>()
+  const clone = (value: unknown, path: string): unknown => {
+    if (value === null || typeof value === 'string' || typeof value === 'boolean') return value
+    if (typeof value === 'number') {
+      if (!Number.isFinite(value)) throw reject('a non-finite number', path)
+      return value
+    }
+    if (Array.isArray(value)) {
+      if (visiting.has(value)) throw reject('a circular reference', path)
+      visiting.add(value)
+      const entries = value.map((entry, index) => clone(entry, `${path}[${index}]`))
+      // Un-mark on exit so one object referenced twice without a cycle passes.
+      visiting.delete(value)
+      return entries
+    }
+    if (isPlainObject(value)) {
+      if (visiting.has(value)) throw reject('a circular reference', path)
+      visiting.add(value)
+      const out: Record<string, unknown> = {}
+      for (const [key, entry] of Object.entries(value)) {
+        if (entry === undefined) continue
+        out[key] = clone(entry, `${path}.${key}`)
+      }
+      visiting.delete(value)
+      return out
+    }
+    throw reject(describeRejected(value), path)
+  }
+  return clone(root, '$') as Record<string, unknown>
+}
+
 /**
  * Layer `over` onto `under`: plain objects merge recursively, every other
- * value (arrays included) replaces the lower layer wholesale, and `undefined`
- * entries in `over` are ignored so a sparse patch cannot erase lower keys.
+ * value (arrays included) replaces the lower layer wholesale. `over` never
+ * carries `undefined` entries — sections come from parsed documents and write
+ * snapshots pass {@link cloneJsonShaped}, which strips them so a sparse patch
+ * cannot erase lower keys.
  */
 function mergeLayers(under: unknown, over: unknown): unknown {
   if (over === undefined) return under
   if (!isPlainObject(under) || !isPlainObject(over)) return over
   const merged: Record<string, unknown> = { ...under }
   for (const [key, value] of Object.entries(over)) {
-    if (value === undefined) continue
     merged[key] = key in merged ? mergeLayers(merged[key], value) : value
   }
   return merged
@@ -155,6 +223,8 @@ interface SettingsWatcher {
   callback: (next: never, prev: never) => void | Promise<void>
   /** Settled tail: invocations of this callback run one at a time, in commit order. */
   tail: Promise<void>
+  /** Cleared by the disposer: a queued invocation checks this before starting. */
+  active: boolean
 }
 
 /** One live namespace registration owned by a registrant fiber. */
@@ -179,6 +249,8 @@ export abstract class Settings extends Service {
   private document: Record<string, unknown> = {}
   /** Per-namespace write chains; settled tails, so a failure never poisons the queue. */
   private readonly writeQueues = new Map<SettingsNamespace, Promise<unknown>>()
+  /** In-flight watcher invocation segments, drained by the dispose teardown. */
+  private readonly pendingTails = new Set<Promise<void>>()
   /** Set at service dispose: refuse new writes while queued ones drain. */
   private stopped = false
 
@@ -199,10 +271,12 @@ export abstract class Settings extends Service {
    */
   async* [Service.init](): AsyncGenerator<() => Promise<void> | void, void, void> {
     yield async () => {
-      // Teardown: refuse new writes, then wait until every queued write chain
-      // settles so disposal completes only once storage is quiescent.
+      // Teardown: refuse new writes and new watcher starts, then wait until
+      // every queued write chain and every started watcher invocation settles
+      // so disposal completes only once storage and observers are quiescent.
+      // Invocations queued but not yet started skip via the stopped check.
       this.stopped = true
-      await Promise.allSettled([...this.writeQueues.values()])
+      await Promise.allSettled([...this.writeQueues.values(), ...this.pendingTails])
     }
     this.publish(await this.load())
   }
@@ -252,9 +326,12 @@ export abstract class Settings extends Service {
     return {
       get: () => registration.resolved as T,
       watch: (callback) => {
-        const watcher: SettingsWatcher = { callback: callback, tail: Promise.resolve() }
+        const watcher: SettingsWatcher = { callback: callback, tail: Promise.resolve(), active: true }
         registration.watchers.add(watcher)
-        return () => registration.watchers.delete(watcher)
+        return () => {
+          watcher.active = false
+          registration.watchers.delete(watcher)
+        }
       },
       update: patch => this.update(ns, patch),
       replace: section => this.replace(ns, section),
@@ -325,13 +402,10 @@ export abstract class Settings extends Service {
       throw new TypeError(`settings ${verb} for "${ns}" must be a plain object`)
     }
     // Snapshot at call time: the queue must never read a caller-owned object
-    // the caller may keep mutating while the write waits its turn.
-    let snapshot: Record<string, unknown>
-    try {
-      snapshot = structuredClone(input)
-    } catch {
-      throw new TypeError(`settings ${verb} for "${ns}" must be JSON-shaped (structured-cloneable) data`)
-    }
+    // the caller may keep mutating while the write waits its turn. The same
+    // walk is the JSON-shape boundary check (see cloneJsonShaped).
+    const snapshot = cloneJsonShaped(input, (label, path) =>
+      new TypeError(`settings ${verb} for "${ns}" must be JSON-shaped data (found ${label} at ${path})`))
     const previous = this.writeQueues.get(ns) ?? Promise.resolve()
     // Chain past a failed predecessor: one rejected write must not poison the
     // namespace queue for every later caller.
@@ -407,11 +481,20 @@ export abstract class Settings extends Service {
       // Serialize per watcher: invocations of one callback run one at a time
       // in commit order, so a slow stale invocation can never apply after a
       // newer one. Sync throws and async rejections land in the same handler.
-      watcher.tail = watcher.tail
-        .then(() => watcher.callback(next as never, prev as never))
+      // The activity check runs when the queued invocation would start, so a
+      // disposer (or service stop) that ran while it waited prevents the
+      // start entirely; started invocations drain at service dispose.
+      const segment = watcher.tail
+        .then(() => {
+          if (!watcher.active || this.isStopped()) return
+          return watcher.callback(next as never, prev as never)
+        })
         .then(() => undefined, (error: unknown) => {
           this.warnWatcherFailure(registration.ns, error)
         })
+      watcher.tail = segment
+      this.pendingTails.add(segment)
+      void segment.then(() => this.pendingTails.delete(segment))
     }
     // Fan the event out one listener at a time (the plain emit stops at the
     // first throwing listener, starving the rest). Invariant violations are
@@ -422,14 +505,21 @@ export abstract class Settings extends Service {
     const args = ['settings/updated', registration.ns, next, prev, source]
     for (const listener of this.ctx.events.dispatch('emit', args) as Array<(...listenerArgs: unknown[]) => unknown>) {
       try {
-        listener(registration.ns, next, prev, source)
+        const returned = listener(registration.ns, next, prev, source)
+        if (returned != null && typeof (returned as PromiseLike<unknown>).then === 'function') {
+          // An emit listener may still be an async function; its rejection
+          // cannot reach the synchronous INVARIANT rethrow below, so it is
+          // contained here instead of becoming an unhandled rejection.
+          void Promise.resolve(returned as PromiseLike<unknown>).then(undefined, (error: unknown) => {
+            this.warnListenerFailure(registration.ns, error)
+          })
+        }
       } catch (error) {
         if ((error as { code?: unknown } | null)?.code === 'INVARIANT') {
           invariantFailure ??= error
           continue
         }
-        this.ctx.logger.warn('settings: a settings/updated listener for "%s" failed', registration.ns)
-        this.ctx.logger.warn(error)
+        this.warnListenerFailure(registration.ns, error)
       }
     }
     if (invariantFailure !== undefined) throw invariantFailure as Error
@@ -440,6 +530,12 @@ export abstract class Settings extends Service {
     this.ctx.logger.warn('settings: watcher for "%s" failed', ns)
     this.ctx.logger.warn(error)
   }
+
+  /** Contained-listener diagnostic shared by the sync and async failure paths. */
+  private warnListenerFailure(ns: SettingsNamespace, error: unknown): void {
+    this.ctx.logger.warn('settings: a settings/updated listener for "%s" failed', ns)
+    this.ctx.logger.warn(error)
+  }
 }
 
 export default Settings

packages/settings/settings/tests/settings.spec.ts

@@ -228,6 +228,14 @@ describe('update', () => {
     expect(scope.get()).toEqual({ theme: 'light', fontSize: 18 })
   })
 
+  it('ignores an explicit undefined entry in the composition base layer', async () => {
+    const { ctx } = await boot({ doc: { 'ui-theme': { theme: 'light' } } })
+    const scope = ctx.settings.register(settingsNamespace('ui-theme'), ThemeSchema, {
+      base: { theme: undefined, fontSize: 16 },
+    })
+    expect(scope.get()).toEqual({ theme: 'light', fontSize: 16 })
+  })
+
   it('rejects a non-object patch', async () => {
     const { ctx } = await boot()
     const scope = ctx.settings.register(settingsNamespace('ui-theme'), ThemeSchema)
@@ -433,11 +441,11 @@ describe('second review regressions', () => {
     expect(applied).toEqual([1, 2])
   })
 
-  it('rejects a plain object that is not structured-cloneable', async () => {
+  it('rejects a function value as not JSON-shaped', async () => {
     const { ctx } = await boot()
     const scope = ctx.settings.register(settingsNamespace('ui-theme'), ThemeSchema)
     await expect(scope.update({ theme: () => 'dark' }))
-      .rejects.toThrow(/JSON-shaped/)
+      .rejects.toThrow(/JSON-shaped.*function at \$\.theme/)
   })
 
   it('rejects a write still queued when the service disposes', async () => {
@@ -532,6 +540,99 @@ describe('publish', () => {
   })
 })
 
+describe('third review regressions', () => {
+  it('skips a queued watch invocation whose disposer ran before it started', async () => {
+    const { ctx, provider } = await boot()
+    const scope = ctx.settings.register(settingsNamespace('ui-theme'), ThemeSchema)
+    const watcher = vi.fn()
+    const dispose = scope.watch(watcher)
+    // The commit chains the invocation as a microtask; the disposer runs in
+    // the same synchronous frame, before that invocation could start.
+    provider.pushExternal({ 'ui-theme': { theme: 'light' } })
+    dispose()
+    await new Promise(resolve => setTimeout(resolve, 10))
+    expect(watcher).not.toHaveBeenCalled()
+  })
+
+  it('waits for an in-flight watch invocation at service dispose', async () => {
+    const { ctx, provider, fiber } = await boot()
+    const scope = ctx.settings.register(settingsNamespace('ui-theme'), ThemeSchema)
+    let release: (() => void) | undefined
+    let finished = false
+    scope.watch(async () => {
+      await new Promise<void>((resolve) => { release = resolve })
+      finished = true
+    })
+    provider.pushExternal({ 'ui-theme': { theme: 'light' } })
+    await vi.waitFor(() => { expect(release).toBeDefined() })
+    let disposed = false
+    const disposal = fiber.dispose().then(() => { disposed = true })
+    await new Promise(resolve => setTimeout(resolve, 15))
+    expect(disposed).toBe(false)
+    release!()
+    await disposal
+    expect(finished).toBe(true)
+  })
+
+  it('rejects a Date at its path before anything persists', async () => {
+    const { ctx, provider } = await boot()
+    const scope = ctx.settings.register(settingsNamespace('ui-theme'), z.object({ value: z.any() }))
+    await expect(scope.update({ value: { at: new Date(0) } }))
+      .rejects.toThrow(/JSON-shaped.*Date at \$\.value\.at/)
+    expect(provider.persisted).toEqual([])
+  })
+
+  it.each([
+    ['a Map', { value: new Map() }, /Map at \$\.value/],
+    ['a bigint', { value: [10n] }, /bigint at \$\.value\[0\]/],
+    ['a symbol', { value: Symbol('x') }, /symbol at \$\.value/],
+    ['a non-finite number', { value: Number.NaN }, /non-finite number at \$\.value/],
+    ['an undefined array entry', { value: [undefined] }, /undefined at \$\.value\[0\]/],
+    ['a class instance', { value: Object.create({ marker: true }) as object }, /non-plain object at \$\.value/],
+  ])('rejects %s that structuredClone would admit', async (_label, patch, message) => {
+    const { ctx } = await boot()
+    const scope = ctx.settings.register(settingsNamespace('ui-theme'), z.object({ value: z.any() }))
+    await expect(scope.update(patch)).rejects.toThrow(message)
+  })
+
+  it('rejects a circular patch instead of storing an alias-looped document', async () => {
+    const { ctx } = await boot()
+    const scope = ctx.settings.register(settingsNamespace('ui-theme'), z.object({ value: z.any() }))
+    const cyclic: Record<string, unknown> = {}
+    cyclic['self'] = cyclic
+    await expect(scope.update({ value: cyclic })).rejects.toThrow(/circular reference at \$\.value\.self/)
+    const loop: unknown[] = []
+    loop.push(loop)
+    await expect(scope.update({ value: loop })).rejects.toThrow(/circular reference at \$\.value\[0\]/)
+  })
+
+  it('accepts one object referenced twice without a cycle', async () => {
+    const { ctx } = await boot()
+    const scope = ctx.settings.register(settingsNamespace('ui-theme'), z.object({ value: z.any() }))
+    const shared = { leaf: 1 }
+    await scope.update({ value: { left: shared, right: shared } })
+    expect(scope.get()).toEqual({ value: { left: { leaf: 1 }, right: { leaf: 1 } } })
+  })
+
+  it('contains an async settings/updated listener rejection and keeps other listeners running', async () => {
+    const { ctx, provider } = await boot()
+    // An async listener violates the event's synchronous signature (typed
+    // consumers get a lint error for it), but an unlinted JS plugin can still
+    // register one; the cast simulates exactly that caller.
+    ctx.on('settings/updated', async () => {
+      throw new Error('async listener boom')
+    })
+    const second = vi.fn()
+    ctx.on('settings/updated', second)
+    ctx.settings.register(settingsNamespace('ui-theme'), ThemeSchema)
+    provider.pushExternal({ 'ui-theme': { theme: 'light' } })
+    expect(second).toHaveBeenCalledTimes(1)
+    // Containment gives the rejection a handler; vitest observes no unhandled
+    // rejection out of this test.
+    await new Promise(resolve => setTimeout(resolve, 10))
+  })
+})
+
 describe('watch', () => {
   it('stops after its disposer runs', async () => {
     const { ctx, provider } = await boot()