diff --git a/docs/core-data-structures/subagent.md b/docs/core-data-structures/subagent.md index d9fdbfa215..5ae83b40fa 100644 --- a/docs/core-data-structures/subagent.md +++ b/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. diff --git a/packages/subagent/subagent/src/index.ts b/packages/subagent/subagent/src/index.ts index cf42746fef..c7d954f09c 100644 --- a/packages/subagent/subagent/src/index.ts +++ b/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)}`) + } } } diff --git a/packages/subagent/subagent/tests/service.spec.ts b/packages/subagent/subagent/tests/service.spec.ts index be743a7a3f..6876c6cd80 100644 --- a/packages/subagent/subagent/tests/service.spec.ts +++ b/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', () => {