Contain subagent lifecycle listeners per-listener, not per-emit

A single try/catch around ctx.emit prevented a thrown subagent/start or
subagent/end listener from propagating, but cordis emit dispatches listeners in
a `.map(cb => cb())` that HALTS on the first throw — so a bad subscriber still
starved the listeners registered after it, violating the AGENTS.md
callback-boundary rule ("one bad subscriber must not starve the listeners after
it"). Resolve the listener callbacks via ctx.events.dispatch and contain each
call individually, the same per-listener guarantee BashExecutor.notifyTaskDone
gives its own listener set.

The two containment tests now register TWO listeners where the first throws and
assert the second still observes the event (start) and the settle (end) — a
regression that fails on the per-emit code (verified: reverted, watched both go
red, restored).

docs/core-data-structures/subagent.md

@@ -85,4 +85,4 @@ interface SubagentProvider {
 }
 ```
 
-The service (`ctx.subagents`) emits `subagent/start` when a run begins and `subagent/end` when it settles (see the [events catalog](../cordis-catalog/events-and-services.md)). Both emits contain a thrown listener (logged, never propagated) so one bad subscriber can neither strand a live run nor surface as an unhandled rejection on the detached settle hook.
+The service (`ctx.subagents`) emits `subagent/start` when a run begins and `subagent/end` when it settles (see the [events catalog](../cordis-catalog/events-and-services.md)). Both emits contain a thrown listener **per listener** (logged, never propagated): one bad subscriber can neither strand a live run, surface as an unhandled rejection on the detached settle hook, nor starve the listeners registered after it.

packages/subagent/subagent/src/index.ts

@@ -153,48 +153,49 @@ export class SubagentService extends Service {
     this.assertCapabilities(provider, request)
 
     const run = provider.start(request)
-    // CONTAIN lifecycle-listener throws: the run is already live, so a throwing
-    // `subagent/start` listener must NOT escape `start()` (the caller would
-    // never receive the run to dispose it — a leaked child). Emit defensively
-    // and log a thrown listener, mirroring the agent registry's `agent/created`
-    // /`agent/disposed` containment.
-    this.emitContainedStart({ provider: name, id: run.id })
+    // Emit `subagent/start` with PER-LISTENER containment (see {@link emitLifecycle}):
+    // the run is already live, so neither a throwing subscriber escaping
+    // `start()` (the caller would never receive the run to dispose it — a leaked
+    // child) NOR one bad subscriber starving the listeners after it is
+    // acceptable. `ctx.emit` halts the dispatch on the first throw, so a single
+    // surrounding try/catch is not enough — each listener is invoked and
+    // contained individually.
+    this.emitLifecycle('subagent/start', { provider: name, id: run.id })
     // Emit `subagent/end` when the run settles. The result promise does not
     // reject on a child-level failure (it resolves with stopReason 'error'),
     // so a rejection here is an infrastructure fault — surface its stop reason
     // as 'error' for the telemetry event without swallowing the rejection
-    // (the consumer still observes it via `run.result`). Containment also keeps
-    // a thrown `subagent/end` listener from becoming an unhandled rejection on
-    // this detached `.then`.
+    // (the consumer still observes it via `run.result`). Per-listener
+    // containment also keeps a thrown `subagent/end` listener from becoming an
+    // unhandled rejection on this detached `.then`.
     void run.result.then(
-      (result) => { this.emitContainedEnd({ provider: name, id: run.id, stopReason: result.stopReason }) },
-      () => { this.emitContainedEnd({ provider: name, id: run.id, stopReason: 'error' }) },
+      (result) => { this.emitLifecycle('subagent/end', { provider: name, id: run.id, stopReason: result.stopReason }) },
+      () => { this.emitLifecycle('subagent/end', { provider: name, id: run.id, stopReason: 'error' }) },
     )
     return run
   }
 
   /**
-   * Emit `subagent/start`, containing a thrown listener (log, never propagate)
-   * so one bad subscriber cannot strand the already-live run before the caller
-   * receives it to dispose.
+   * Emit a `subagent/*` lifecycle event with PER-LISTENER containment: dispatch
+   * each subscriber individually and log (never propagate) a thrown one, so one
+   * bad subscriber can neither strand the already-live run, surface as an
+   * unhandled rejection on the detached settle hook, NOR starve the listeners
+   * registered after it. A single try/catch around `ctx.emit` would not do the
+   * last part — cordis `emit` runs listeners in a `.map(cb => cb())` that halts
+   * on the first throw — so this resolves the listener callbacks via
+   * `ctx.events.dispatch` and contains each call, the same guarantee
+   * `BashExecutor.notifyTaskDone` gives its own listener set.
    */
-  private emitContainedStart(info: SubagentRunInfo): void {
-    try {
-      this.ctx.emit('subagent/start', info)
-    } catch (error: unknown) {
-      this.ctx.logger.warn(`subagent: subagent/start listener threw: ${String(error)}`)
-    }
-  }
-
-  /**
-   * Emit `subagent/end`, containing a thrown listener so it cannot surface as an
-   * unhandled rejection on the detached result-settle hook.
-   */
-  private emitContainedEnd(info: SubagentRunEndInfo): void {
-    try {
-      this.ctx.emit('subagent/end', info)
-    } catch (error: unknown) {
-      this.ctx.logger.warn(`subagent: subagent/end listener threw: ${String(error)}`)
+  private emitLifecycle(
+    name: 'subagent/start' | 'subagent/end',
+    info: SubagentRunInfo | SubagentRunEndInfo,
+  ): void {
+    for (const callback of this.ctx.events.dispatch('emit', [name, info])) {
+      try {
+        callback(info)
+      } catch (error: unknown) {
+        this.ctx.logger.warn(`subagent: ${name} listener threw: ${String(error)}`)
+      }
     }
   }
 

packages/subagent/subagent/tests/service.spec.ts

@@ -200,31 +200,37 @@ describe('SubagentService', () => {
     expect(ended).toHaveBeenCalledWith(expect.objectContaining({ provider: 'rejecter', id: run.id, stopReason: 'error' }))
   })
 
-  it('contains a throwing subagent/start listener so start() still returns the run', async () => {
+  it('contains a throwing subagent/start listener per-listener: a later listener still observes the event and start() returns the run', async () => {
     const ctx = new Context()
     await ctx.plugin(SubagentService)
     ctx.subagents.registerProvider(new StubProvider('contain'))
-    // A bad subscriber must not strand the live run: start() returns it anyway.
+    // Two listeners; the FIRST throws. Per-listener containment means the second
+    // must STILL run (a single try/catch around ctx.emit would let the first
+    // throw halt the dispatch and starve the second — the round-2 regression).
+    const second = vi.fn()
     ctx.on('subagent/start', () => { throw new Error('bad start listener') })
+    ctx.on('subagent/start', second)
 
     const run = ctx.subagents.start('contain', baseRequest())
     expect(run.id).toBeDefined()
+    expect(second).toHaveBeenCalledWith(expect.objectContaining({ provider: 'contain', id: run.id }))
     await expect(run.result).resolves.toMatchObject({ stopReason: 'completed' })
   })
 
-  it('contains a throwing subagent/end listener (no unhandled rejection on the settle hook)', async () => {
+  it('contains a throwing subagent/end listener per-listener: a later listener still observes the settle, no unhandled rejection', async () => {
     const ctx = new Context()
     await ctx.plugin(SubagentService)
     ctx.subagents.registerProvider(new StubProvider('contain-end'))
+    const second = vi.fn()
     ctx.on('subagent/end', () => { throw new Error('bad end listener') })
+    ctx.on('subagent/end', second)
 
     const run = ctx.subagents.start('contain-end', baseRequest())
     await run.result
-    // Let the detached `.then` + the contained emit run; a thrown listener here
-    // must be swallowed (logged), not escape as an unhandled rejection.
+    // Let the detached `.then` + the contained emit run.
     await Promise.resolve()
     await Promise.resolve()
-    expect(run.id).toBeDefined()
+    expect(second).toHaveBeenCalledWith(expect.objectContaining({ provider: 'contain-end', id: run.id, stopReason: 'completed' }))
   })
 
   it('SubagentError extends the shared HarnessError base', () => {