From 2df41ee1d33afa0ae78c51175e4970a77aa76712 Mon Sep 17 00:00:00 2001
From: Tianyi Cui <53024+tianyicui@users.noreply.github.com>
Date: Mon, 15 Jun 2026 00:25:17 +0800
Subject: [PATCH] fix: make the six registration methods atomic under a
 throwing change-listener (P1-1)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

llm.registerAdapter, agents.register, sessions.create, systemPrompt.section,
systemPrompt.tools, and tools.register each mutated state, emitted a change
event, then returned the disposer. In Cordis a synchronous throw before the
effect returns its disposer leaves nothing for the fiber to collect, so a
throwing change-listener leaked the registry entry permanently — HMR/dispose
could not clean it, and the duplicate-name/already-exists check stayed wedged
until restart.

Convert each to the generator-effect pattern already proven in
AgentLoop.create: mutate state, `yield` the disposer that undoes it (collected
before the next step runs, so it is torn down if a later step throws), THEN
emit the change event. The existing duplicate-name throws are unchanged — they
fire before any mutation, so they correctly leak nothing. No public API change:
generator effects are still synchronous SyncEffects and register() keeps
returning its fire-and-forget disposer wrapper.

Tests: a listener-throw rollback test for all six methods — register with a
change-listener that throws, assert the call throws AND the registry is clean
(entry absent; a subsequent listener-free register of the same name succeeds
and contributes exactly once). For systemPrompt (no duplicate-name check) the
two tests assert assembly is clean. Verified each fails against the pre-fix
emit-before-return-disposer form.
---
 packages/agent/src/index.ts                   | 13 +++++--
 packages/agent/tests/agent.spec.ts            | 21 ++++++++++
 packages/llm/src/index.ts                     | 14 +++++--
 packages/llm/tests/service.spec.ts            | 22 +++++++++++
 packages/session/src/index.ts                 | 14 +++++--
 packages/session/tests/session.spec.ts        | 23 +++++++++++
 packages/system-prompt/src/index.ts           | 21 ++++++----
 .../system-prompt/tests/system-prompt.spec.ts | 38 +++++++++++++++++++
 packages/tools/src/index.ts                   | 13 +++++--
 packages/tools/tests/tools.spec.ts            | 21 ++++++++++
 10 files changed, 176 insertions(+), 24 deletions(-)

diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts
index a4d20b93a4..3118e6f9e7 100644
--- a/packages/agent/src/index.ts
+++ b/packages/agent/src/index.ts
@@ -35,17 +35,22 @@ export class AgentRegistry extends Service {
    * when the calling fiber is disposed. Returns the disposer.
    */
   register(agent: Agent): () => void {
-    const dispose = this.ctx.effect(() => {
+    const dispose = this.ctx.effect(function* (this: AgentRegistry) {
       if (this.store.has(agent.id)) {
         throw new Error(`agent "${agent.id}" is already registered`)
       }
       this.store.set(agent.id, agent)
-      this.ctx.emit('agent/created', agent)
-      return () => {
+      // Yield the rollback BEFORE emitting `agent/created`: a generator effect
+      // collects each yielded disposer before the next step runs, so a
+      // throwing `agent/created` listener rolls the entry back instead of
+      // leaking it (a leak would wedge the duplicate-id check until restart).
+      // The duplicate throw above fires before any mutation — it leaks nothing.
+      yield () => {
         this.store.delete(agent.id)
         this.ctx.emit('agent/disposed', agent)
       }
-    }, 'agents.register()')
+      this.ctx.emit('agent/created', agent)
+    }.bind(this), 'agents.register()')
     // ctx.effect's disposer returns Promise<void>; our disposer API is
     // synchronous fire-and-forget — discard the (always-resolved) promise.
     return () => void dispose()
diff --git a/packages/agent/tests/agent.spec.ts b/packages/agent/tests/agent.spec.ts
index 5d66035421..d83c00a006 100644
--- a/packages/agent/tests/agent.spec.ts
+++ b/packages/agent/tests/agent.spec.ts
@@ -52,4 +52,25 @@ describe('AgentRegistry', () => {
     await fiber.dispose()
     expect(ctx.agents.list().map(a => a.id)).toEqual(['main'])
   })
+
+  it('rolls back the agent entry when an agent/created listener throws (P1-1)', async () => {
+    const ctx = new Context()
+    await ctx.plugin(AgentRegistry)
+
+    let threw = false
+    ctx.on('agent/created', () => {
+      if (!threw) { threw = true; throw new Error('boom created listener') }
+    })
+
+    // The throwing emit must roll the entry back, not leak it.
+    expect(() => ctx.agents.register(stubAgent('main'))).toThrow('boom created listener')
+    expect(ctx.agents.get('main')).toBeUndefined() // rolled back, not leaked
+
+    // A subsequent listener-free register of the SAME id succeeds and is
+    // tracked exactly once (the duplicate-id check is not wedged).
+    const dispose = ctx.agents.register(stubAgent('main'))
+    expect(ctx.agents.list().map(a => a.id)).toEqual(['main'])
+    dispose()
+    expect(ctx.agents.get('main')).toBeUndefined()
+  })
 })
diff --git a/packages/llm/src/index.ts b/packages/llm/src/index.ts
index 268289f465..460316ea50 100644
--- a/packages/llm/src/index.ts
+++ b/packages/llm/src/index.ts
@@ -80,19 +80,25 @@ export class LlmService extends Service {
    * fiber.
    */
   registerAdapter(models: string[], adapter: LlmAdapter): () => void {
-    const dispose = this.ctx.effect(() => {
+    const dispose = this.ctx.effect(function* (this: LlmService) {
       for (const model of models) {
         if (this.adapters.has(model)) {
           throw new LlmError(`an adapter for model "${model}" is already registered`, 'DUPLICATE_ADAPTER')
         }
       }
       for (const model of models) this.adapters.set(model, adapter)
-      this.ctx.emit('llm/adapter-change')
-      return () => {
+      // Yield the rollback BEFORE emitting the change event: a generator effect
+      // collects each yielded disposer before running the next step, so a
+      // throwing `llm/adapter-change` listener rolls the mutation back instead
+      // of leaking the entry (which would wedge the duplicate check until
+      // restart). The duplicate throws above fire before any mutation, so they
+      // correctly leak nothing.
+      yield () => {
         for (const model of models) this.adapters.delete(model)
         this.ctx.emit('llm/adapter-change')
       }
-    }, 'llm.registerAdapter()')
+      this.ctx.emit('llm/adapter-change')
+    }.bind(this), 'llm.registerAdapter()')
     // ctx.effect's disposer returns Promise<void>; our disposer API is
     // synchronous fire-and-forget — discard the (always-resolved) promise.
     return () => void dispose()
diff --git a/packages/llm/tests/service.spec.ts b/packages/llm/tests/service.spec.ts
index 24bb7b359c..35333c0ffd 100644
--- a/packages/llm/tests/service.spec.ts
+++ b/packages/llm/tests/service.spec.ts
@@ -146,4 +146,26 @@ describe('LlmService', () => {
       expect((error as LlmError).code).toBe('DUPLICATE_ADAPTER')
     }
   })
+
+  it('rolls back the adapter entry when an adapter-change listener throws (P1-1)', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+
+    // A change listener that throws on the FIRST emit only.
+    let threw = false
+    ctx.on('llm/adapter-change', () => {
+      if (!threw) { threw = true; throw new Error('boom change listener') }
+    })
+
+    // The throwing emit must roll the mutation back, not leak it.
+    expect(() => ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))).toThrow('boom change listener')
+    expect(ctx.llm.models()).toEqual([]) // entry rolled back, not leaked
+
+    // A subsequent listener-free register of the SAME model succeeds and
+    // contributes exactly once (the duplicate check is not wedged).
+    const dispose = ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))
+    expect(ctx.llm.models()).toEqual(['m1'])
+    dispose()
+    expect(ctx.llm.models()).toEqual([])
+  })
 })
diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts
index 75147899f0..f7e019ddcb 100644
--- a/packages/session/src/index.ts
+++ b/packages/session/src/index.ts
@@ -167,15 +167,21 @@ export class SessionStore extends Service {
     const sessionId = SessionId(id ?? `session-${++this.counter}`)
     if (this.store.has(sessionId)) throw new Error(`session "${sessionId}" already exists`)
     const session = new Session(sessionId, seed)
-    this.ctx.effect(() => {
+    this.ctx.effect(function* (this: SessionStore) {
       session.onAppend = (event) => { this.ctx.emit('session/event', session, event) }
       this.store.set(sessionId, session)
-      this.ctx.emit('session/created', session)
-      return () => {
+      // Yield the rollback BEFORE emitting `session/created`: a generator
+      // effect collects each yielded disposer before the next step runs, so a
+      // throwing `session/created` listener detaches onAppend and removes the
+      // store entry instead of leaking them (a leak would wedge the
+      // already-exists check until restart). The duplicate throw above fires
+      // before any mutation — it leaks nothing.
+      yield () => {
         session.onAppend = undefined
         this.store.delete(sessionId)
       }
-    }, 'sessions.create()')
+      this.ctx.emit('session/created', session)
+    }.bind(this), 'sessions.create()')
     return session
   }
 
diff --git a/packages/session/tests/session.spec.ts b/packages/session/tests/session.spec.ts
index 740ce78bda..42e4fa6ec9 100644
--- a/packages/session/tests/session.spec.ts
+++ b/packages/session/tests/session.spec.ts
@@ -132,4 +132,27 @@ describe('SessionStore', () => {
     session.append('user/message', { content: [{ type: 'text', text: 'late' }], source: { kind: 'user' } })
     expect(observed).toBe(0)
   })
+
+  it('rolls back the session (and onAppend) when a session/created listener throws (P1-1)', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+
+    let threw = false
+    ctx.on('session/created', () => {
+      if (!threw) { threw = true; throw new Error('boom created listener') }
+    })
+
+    // The throwing emit must roll the store entry back, not leak it.
+    expect(() => ctx.sessions.create('fixed')).toThrow('boom created listener')
+    expect(ctx.sessions.get('fixed')).toBeUndefined() // rolled back, not leaked
+
+    // A subsequent create of the SAME id succeeds (the already-exists check is
+    // not wedged) and its onAppend is correctly wired (events observable).
+    const events: SessionEvent[] = []
+    ctx.on('session/event', (_session, event) => void events.push(event))
+    const session = ctx.sessions.create('fixed')
+    expect(ctx.sessions.get('fixed')).toBe(session)
+    session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })
+    expect(events).toHaveLength(1)
+  })
 })
diff --git a/packages/system-prompt/src/index.ts b/packages/system-prompt/src/index.ts
index c64e08495d..ec89a3460c 100644
--- a/packages/system-prompt/src/index.ts
+++ b/packages/system-prompt/src/index.ts
@@ -73,16 +73,20 @@ export class SystemPrompt extends Service {
    * fiber is disposed. Emits `system-prompt/change` on register/unregister.
    */
   section(section: PromptSection): () => void {
-    const dispose = this.ctx.effect(() => {
+    const dispose = this.ctx.effect(function* (this: SystemPrompt) {
       this.sections.push(section)
-      this.ctx.emit('system-prompt/change')
-      return () => {
+      // Yield the rollback BEFORE emitting `system-prompt/change`: a generator
+      // effect collects each yielded disposer before the next step runs, so a
+      // throwing change listener removes the section instead of leaking it into
+      // every future assembly.
+      yield () => {
         const index = this.sections.indexOf(section)
         /* v8 ignore next 3 -- defensive: section was registered, so indexOf is guaranteed >= 0 */
         if (index >= 0) this.sections.splice(index, 1)
         this.ctx.emit('system-prompt/change')
       }
-    }, 'systemPrompt.section()')
+      this.ctx.emit('system-prompt/change')
+    }.bind(this), 'systemPrompt.section()')
     // ctx.effect's disposer returns Promise<void>; our disposer API is
     // synchronous fire-and-forget — discard the (always-resolved) promise.
     return () => void dispose()
@@ -94,16 +98,17 @@ export class SystemPrompt extends Service {
    * removed when the calling fiber is disposed. Emits `system-prompt/change`.
    */
   tools(provider: () => ToolSchema[]): () => void {
-    const dispose = this.ctx.effect(() => {
+    const dispose = this.ctx.effect(function* (this: SystemPrompt) {
       this.toolProviders.push(provider)
-      this.ctx.emit('system-prompt/change')
-      return () => {
+      // Yield the rollback BEFORE emitting `system-prompt/change` (see section()).
+      yield () => {
         const index = this.toolProviders.indexOf(provider)
         /* v8 ignore next 3 -- defensive: provider was registered, so indexOf is guaranteed >= 0 */
         if (index >= 0) this.toolProviders.splice(index, 1)
         this.ctx.emit('system-prompt/change')
       }
-    }, 'systemPrompt.tools()')
+      this.ctx.emit('system-prompt/change')
+    }.bind(this), 'systemPrompt.tools()')
     // ctx.effect's disposer returns Promise<void>; our disposer API is
     // synchronous fire-and-forget — discard the (always-resolved) promise.
     return () => void dispose()
diff --git a/packages/system-prompt/tests/system-prompt.spec.ts b/packages/system-prompt/tests/system-prompt.spec.ts
index d15f3d46b3..43e9240412 100644
--- a/packages/system-prompt/tests/system-prompt.spec.ts
+++ b/packages/system-prompt/tests/system-prompt.spec.ts
@@ -34,6 +34,44 @@ describe('SystemPrompt', () => {
     expect(assembly.tools).toHaveLength(0)
   })
 
+  it('rolls back a section when a system-prompt/change listener throws (P1-1)', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SystemPrompt)
+
+    // Throw on the first emit only. Note the rollback path itself emits
+    // system-prompt/change, so a multi-shot guard would also fire on rollback;
+    // a single-shot guard isolates the register's own emit.
+    let threw = false
+    const off = ctx.on('system-prompt/change', () => {
+      if (!threw) { threw = true; throw new Error('boom change listener') }
+    })
+
+    expect(() => ctx.systemPrompt.section({ name: 'p', order: 0, text: 'persona' })).toThrow('boom change listener')
+    expect((await ctx.systemPrompt.assemble()).sections).toHaveLength(0) // nothing leaked
+
+    // Subsequent listener-free register contributes exactly once.
+    off()
+    ctx.systemPrompt.section({ name: 'p', order: 0, text: 'persona' })
+    expect((await ctx.systemPrompt.assemble()).sections.map(s => s.name)).toEqual(['p'])
+  })
+
+  it('rolls back a tool provider when a system-prompt/change listener throws (P1-1)', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SystemPrompt)
+
+    let threw = false
+    const off = ctx.on('system-prompt/change', () => {
+      if (!threw) { threw = true; throw new Error('boom change listener') }
+    })
+
+    expect(() => ctx.systemPrompt.tools(() => [{ name: 't', description: '', parameters: {} }])).toThrow('boom change listener')
+    expect((await ctx.systemPrompt.assemble()).tools).toHaveLength(0) // nothing leaked
+
+    off()
+    ctx.systemPrompt.tools(() => [{ name: 't', description: '', parameters: {} }])
+    expect((await ctx.systemPrompt.assemble()).tools.map(t => t.name)).toEqual(['t'])
+  })
+
   it('composes multiple system-prompt/assemble waterfall listeners in order', async () => {
     const ctx = new Context()
     await ctx.plugin(SystemPrompt)
diff --git a/packages/tools/src/index.ts b/packages/tools/src/index.ts
index 45c40cdadc..8e8c106147 100644
--- a/packages/tools/src/index.ts
+++ b/packages/tools/src/index.ts
@@ -140,17 +140,22 @@ export class ToolRegistry extends Service {
    * with the calling fiber. Emits `tools/change` on register/unregister.
    */
   register(definition: ToolDefinition): () => void {
-    const dispose = this.ctx.effect(() => {
+    const dispose = this.ctx.effect(function* (this: ToolRegistry) {
       if (this.store.has(definition.name)) {
         throw new Error(`tool "${definition.name}" is already registered`)
       }
       this.store.set(definition.name, definition)
-      this.ctx.emit('tools/change')
-      return () => {
+      // Yield the rollback BEFORE emitting `tools/change`: a generator effect
+      // collects each yielded disposer before the next step runs, so a throwing
+      // `tools/change` listener removes the tool instead of leaking it (a leak
+      // would wedge the duplicate-name check until restart). The duplicate
+      // throw above fires before any mutation — it leaks nothing.
+      yield () => {
         this.store.delete(definition.name)
         this.ctx.emit('tools/change')
       }
-    }, 'tools.register()')
+      this.ctx.emit('tools/change')
+    }.bind(this), 'tools.register()')
     // ctx.effect's disposer returns Promise<void>; our disposer API is
     // synchronous fire-and-forget — discard the (always-resolved) promise.
     return () => void dispose()
diff --git a/packages/tools/tests/tools.spec.ts b/packages/tools/tests/tools.spec.ts
index 86e3d6c531..7338a2c9c8 100644
--- a/packages/tools/tests/tools.spec.ts
+++ b/packages/tools/tests/tools.spec.ts
@@ -147,6 +147,27 @@ describe('ToolRegistry', () => {
     dispose()
     expect(ctx.tools.schemas().map(t => t.name)).toEqual(['echo'])
   })
+
+  it('rolls back the tool entry when a tools/change listener throws (P1-1)', async () => {
+    const ctx = await setup()
+
+    let threw = false
+    ctx.on('tools/change', () => {
+      if (!threw) { threw = true; throw new Error('boom change listener') }
+    })
+
+    // The throwing emit must roll the entry back, not leak it.
+    expect(() => ctx.tools.register(echoTool)).toThrow('boom change listener')
+    expect(ctx.tools.get('echo')).toBeUndefined() // rolled back, not leaked
+    expect(ctx.tools.schemas()).toHaveLength(0)
+
+    // A subsequent listener-free register of the SAME name succeeds and is
+    // exposed exactly once (the duplicate-name check is not wedged).
+    const dispose = ctx.tools.register(echoTool)
+    expect(ctx.tools.schemas().map(t => t.name)).toEqual(['echo'])
+    dispose()
+    expect(ctx.tools.get('echo')).toBeUndefined()
+  })
 })
 
 describe('defineTool / schema DSL', () => {