diff --git a/packages/acp/tests/dispose.spec.ts b/packages/acp/tests/dispose.spec.ts index 72e7ba42a9..d3af9ccd6c 100644 --- a/packages/acp/tests/dispose.spec.ts +++ b/packages/acp/tests/dispose.spec.ts @@ -248,4 +248,29 @@ describe('acp bridge — disposal & HMR safety', () => { expect(handleB.agent.status).not.toBe('disposed') await harness.dispose() }) + + it('a throwing agent/disposed listener does not prevent session removal (composite-effect containment)', async () => { + // The AgentHandle teardown folds session-detach, register, and loop-stop + // into ONE composite effect whose disposers run as a `.then()` chain. The + // register disposer emits `agent/disposed`; if a listener throws and the + // emit is UNCONTAINED, the rejected chain skips the LATER session-detach + // disposer — stranding the session in the store with `onAppend` attached (a + // leak AND a durability hole, since the new design relies on detach + // running). The emit must be contained. Register a throwing listener, drive + // a clean turn, dispose, and assert the session was STILL removed. + const harness = await makeBridgeHarness({ storageDir, script: [textResponse('ok')] }) + harness.ctx.on('agent/disposed', () => { throw new Error('boom disposed listener') }) + const handle = harness.ctx.agents.create({ + agentId: 'guard-a', sessionId: 'guard-a', agentOptions: { model: 'mock' }, + }) + handle.agent.send([{ type: 'text', text: 'go' }]) + await handle.agent.whenIdle() + expect(harness.ctx.sessions.get('guard-a')).toBeDefined() + + // Dispose: the throwing listener must NOT break the chain before detach. + await handle.dispose() + expect(harness.ctx.agents.get('guard-a')).toBeUndefined() + expect(harness.ctx.sessions.get('guard-a')).toBeUndefined() // detach still ran + await harness.dispose() + }) }) diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 2063812963..cd66156052 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -164,7 +164,22 @@ export class AgentRegistry extends Service { // The duplicate throw above fires before any mutation — it leaks nothing. yield () => { this.store.delete(agent.id) - this.ctx.emit('agent/disposed', agent) + // CONTAIN a throwing `agent/disposed` listener: this disposer runs as + // one link in the owning fiber/effect's disposal chain, and Cordis + // chains later disposers with `task.then(next)` — so an UNCAUGHT throw + // here rejects the chain and SKIPS every later disposer. When this + // registration shares a composite effect with a session (the agent + // factory's `AgentLoop.start`, where the session-detach disposer runs + // AFTER this one), a swallowed-less throw would strand the session in + // the store with `onAppend` attached — a leak AND a durability hole. + // The store entry is already removed above (the useful state), so + // logging the listener bug and continuing is correct (mirrors the + // guarded `agent/status` emit in dsh-agent-loop's ReactLoopAgent). + try { + this.ctx.emit('agent/disposed', agent) + } catch (error: unknown) { + this.ctx.logger.warn(`agent "${agent.id}": agent/disposed listener threw: ${String(error)}`) + } } this.ctx.emit('agent/created', agent) }.bind(this), 'agents.register()') diff --git a/packages/session/README.md b/packages/session/README.md index bdd825e2e7..dabe316a28 100644 --- a/packages/session/README.md +++ b/packages/session/README.md @@ -12,6 +12,16 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall - `ctx.sessions.get(id: string): Session | undefined` - `ctx.sessions.list(): Session[]` +#### Advanced: ordered-teardown lifecycle primitives + +`create()` covers the common case (the session is owned by the calling fiber). When a session must be torn down **in order with another resource** — so a final flush is captured before `onAppend` detaches — `create()`'s self-contained effect is wrong, because a fiber unload disposes sibling effects *concurrently*. For that, split the lifecycle and fold it into the owner's single effect: + +- `ctx.sessions.prepare(id?, options?): Session` — validate the id/cwd and construct the `Session`, WITHOUT entering it into the store. Same options as `create`. +- `ctx.sessions.enter(session): () => void` — wire `onAppend` → `session/event` and add the session to the store; returns the DETACH disposer. Does NOT emit `session/created` (the caller yields the disposer first, then calls `announce`, so a throwing listener rolls the attach back). The id was already validated by `prepare`, which runs in the same synchronous sequence, so `enter` does not re-check. +- `ctx.sessions.announce(session): void` — emit `session/created` for an entered session. + +`dsh-agent-loop`'s `AgentLoop.start` is the canonical consumer: it yields `enter`'s detach disposer, the registry unregister, and the loop-stop disposer into ONE composite effect, so teardown stops + awaits the loop (final flush captured) BEFORE detaching the session — whether the trigger is the `AgentHandle`'s `dispose()` or a fiber unload. + ### Events | Event | Mode | Purpose |