diff --git a/packages/acp/tests/dispose.spec.ts b/packages/acp/tests/dispose.spec.ts index 6ff89b400e..72e7ba42a9 100644 --- a/packages/acp/tests/dispose.spec.ts +++ b/packages/acp/tests/dispose.spec.ts @@ -184,6 +184,43 @@ describe('acp bridge — disposal & HMR safety', () => { await harness.dispose() }) + it('a turn aborted BY the dispose still flushes its closing turn/end to disk (durability, mid-turn)', async () => { + // The teardown-order contract only earns its keep when the closing events are + // produced BY the dispose itself. Here the model stream HANGS, so the turn is + // still open when teardown runs: the composite agent effect stops the loop, + // the loop unwinds and appends `turn/end {disposed}` + runs its final + // `session/flush` — all while `onAppend` is still attached (the session + // detach is the LAST disposer in the same effect's LIFO chain) — and only + // THEN is the session detached. If the order were inverted (or the session + // were a racing SIBLING effect), the abort-produced `turn/end` would never + // reach disk and a re-load would instead show crash-recovery's synthetic + // `interrupted` closer. Re-load from disk and assert the REAL `disposed` + // reason landed — proving the loop's own closing event was captured, not a + // recovered substitute. + const harness = await makeBridgeHarness({ storageDir, script: ['hang'] }) + await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) + const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }) + const agent = harness.ctx.agents.get(sessionId)! + void harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }).catch(() => {}) + await new Promise(r => setTimeout(r, 30)) + expect(agent.status).toBe('running') + // The turn is OPEN in the log (turn/start appended, no turn/end yet). + const openTurnEnds = agent.session.events.filter(e => e.type === 'turn/end').length + + // Dispose JUST the bridge: a fiber unload that must STILL honor the ordered + // teardown (the composite effect runs its disposer chain as a unit). + await harness.acpFiber.dispose() + expect(harness.ctx.agents.get(sessionId)).toBeUndefined() + + // The loop's own `turn/end {disposed}` is on disk (re-load: the world, not + // self-report) — NOT a crash-recovery `interrupted` substitute. + const reloaded = await harness.ctx.sessionPersistence.load(SessionId(sessionId)) + const persistedTurnEnds = reloaded.events.filter(e => e.type === 'turn/end') + expect(persistedTurnEnds.length).toBe(openTurnEnds + 1) + expect(persistedTurnEnds.at(-1)!.data.reason).toMatchObject({ kind: 'disposed' }) + await harness.dispose() + }) + it('per-session AgentHandle dispose leaves sibling agents untouched', async () => { // The factory returns a per-agent AgentHandle whose dispose() tears down // EXACTLY that agent + its session — RFC 011 isolation. Create two agents diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index 7cc90ba3da..3963b24594 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -120,10 +120,11 @@ export class AgentLoop extends Service implements AgentFactory { */ create(id: string, options: AgentOptions = {}): ReactLoopAgent { this.assertAgentIdFree(id) - // Config/programmatic path: the session is owned by THIS fiber (the plain - // create()), so disposing the AgentLoop/caller fiber removes it. No - // AgentHandle is needed — the register+start effect is fiber-owned too. - const session = this.ctx.sessions.create(`${id}-session-${randomUUID()}`, { meta: {} }) + // Config/programmatic path: prepare the session and let start() fold its + // lifecycle into the agent's composite effect (so a fiber unload tears the + // session + agent down as one ordered chain, capturing the loop's closing + // flush). The whole effect is owned by THIS fiber; no AgentHandle is needed. + const session = this.ctx.sessions.prepare(`${id}-session-${randomUUID()}`, { meta: {} }) const { agent } = this.start(AgentId(id), options, session) return agent } @@ -136,12 +137,12 @@ export class AgentLoop extends Service implements AgentFactory { * an {@link AgentHandle} the owner disposes to tear down exactly this agent. */ createAgent(options: CreateAgentOptions): AgentHandle { - // Check the agent id BEFORE creating the session: register() would reject a - // duplicate id only AFTER sessions.create(), leaving an orphaned live - // session (and lazy persistence state) that blocks reuse of that id. + // Check the agent id BEFORE preparing the session: register() would reject a + // duplicate id only AFTER the session enters the store, leaving an orphaned + // live session (and lazy persistence state) that blocks reuse of that id. this.assertAgentIdFree(options.agentId) - const owned = this.ctx.sessions.createOwned(options.sessionId, { meta: options.meta ?? {} }) - return this.startOwned(AgentId(options.agentId), options.agentOptions ?? {}, owned) + const session = this.ctx.sessions.prepare(options.sessionId, { meta: options.meta ?? {} }) + return this.startOwned(AgentId(options.agentId), options.agentOptions ?? {}, session) } /** @@ -193,15 +194,16 @@ export class AgentLoop extends Service implements AgentFactory { const { meta, events } = await persistence.load(SessionId(options.resumeSessionId)) // Re-check the agent id AFTER the await: the pre-load check above can go // stale while load() is pending (a concurrent resume/create may register the - // same id). Re-checking immediately before sessions.create() keeps the + // same id). Re-checking immediately before prepare()/start keeps the // "no orphaned session on a duplicate id" guarantee under concurrency. this.assertAgentIdFree(options.agentId) // Reconstruct the live session with the FULL persisted header (createdAt, // cwd, lineage) so resume preserves identity, not just the cwd. The seed // events make lastTurnNumber/deriveMessages continue; the backend already // has state (cursor) from the load above, so onCreated is a no-op and the - // seed is not re-persisted. - const owned = this.ctx.sessions.createOwned(options.resumeSessionId, { + // seed is not re-persisted. prepare() (not create()) so the session + // lifecycle folds into the agent's composite effect (ordered teardown). + const session = this.ctx.sessions.prepare(options.resumeSessionId, { seed: events, meta: { createdAt: meta.createdAt, @@ -209,14 +211,14 @@ export class AgentLoop extends Service implements AgentFactory { ...meta.parentSession !== undefined ? { parentSession: meta.parentSession } : {}, }, }) - return this.startOwned(AgentId(options.agentId), options.agentOptions ?? {}, owned) + return this.startOwned(AgentId(options.agentId), options.agentOptions ?? {}, session) } /** - * Reject a duplicate agent id BEFORE any session is created, so a failed - * factory call never leaves an orphaned live session (and lazy persistence - * state) behind. `register()` enforces the same uniqueness, but only after - * `sessions.create()` has already run. + * Reject a duplicate agent id BEFORE the session is entered into the store, so + * a failed factory call never leaves an orphaned live session (and lazy + * persistence state) behind. `register()` enforces the same uniqueness, but + * only after the session has already entered the store. */ private assertAgentIdFree(id: string): void { if (this.ctx.agents.get(id) !== undefined) { @@ -225,53 +227,55 @@ export class AgentLoop extends Service implements AgentFactory { } /** - * Shared: construct a ReactLoopAgent, register it, and start its loop. The - * register + loop-stop disposers live in ONE generator effect so they run - * LIFO on dispose (the loop-stop disposer — yielded last — runs first, then - * the registry unregister), so a throwing stop() cannot leak the registry - * entry. Returns the agent plus the effect's disposer (`disposeAgent`); the - * effect is owned by the caller fiber, so disposing that fiber also tears the - * agent down — the disposer is for an OWNER that needs to tear down ONE agent. + * Shared: construct a ReactLoopAgent over a PREPARED (not-yet-entered) + * session, then build the ONE composite effect that owns the whole agent + * lifecycle — session entry, registry registration, and the loop. Keeping all + * three in a SINGLE effect (not sibling effects) is load-bearing: a fiber + * unload disposes sibling effects CONCURRENTLY (`Promise.all`), which would + * race the session detach against the loop's closing flush and drop the + * closing `turn/end`. Inside one effect the disposers run as an ORDERED LIFO + * chain — the runtime awaits each disposer's returned promise before the next: + * + * yield session-detach (disposed LAST — detach onAppend + remove entry) + * yield register (disposed 2nd — unregister) + * yield stop-and-drain (disposed FIRST — request loop stop, await agent.done) + * + * So on teardown: the loop is stopped and AWAITED to exit (its final + * `session/flush` + `turn/end` fire through the still-attached `onAppend`), + * THEN the agent is unregistered, THEN the session is detached — capturing the + * closing events before detach, whether the trigger is the handle's `dispose()` + * OR a fiber unload. Rollback safety: each yield runs before the next mutation, + * so a throwing `session/created`/`agent/created` listener unwinds the + * already-yielded disposers instead of leaking. + * + * Returns the agent plus the composite effect's disposer (`disposeAgent`). */ private start(id: AgentId, options: AgentOptions, session: Session): { agent: ReactLoopAgent; disposeAgent: () => Promise } { const agent = new ReactLoopAgent(this.ctx, id, options, session) const dispose = this.ctx.effect(function* (this: AgentLoop) { + yield this.ctx.sessions.enter(session) + this.ctx.sessions.announce(session) yield this.ctx.agents.register(agent) - yield agent.start() + const stop = agent.start() + // Disposed FIRST (LIFO): request loop stop (sync), then AWAIT the loop's + // actual exit so its closing flush lands while onAppend (yielded above, + // disposed later) is still attached. + yield async () => { stop(); await agent.done } }.bind(this), 'agentLoop.start()') return { agent, disposeAgent: async () => { await dispose() } } } /** - * Build an {@link AgentHandle} for an OWNED session + agent. The handle's - * `dispose()` tears down exactly this agent in the order durability requires: - * - * 1. run `disposeAgent` — the register+start effect's disposer. LIFO runs - * `agent.start()`'s (synchronous) disposer first: it sets `disposed`, - * aborts the in-flight step, and unblocks the loop's idle wait. Then the - * registry unregister runs. The loop has NOT necessarily exited yet — the - * start disposer only REQUESTS exit, it does not await it. - * 2. `await agent.done` — the loop-exit promise. The loop unwinds and runs - * its final `session/flush` + `turn/end`, delivered through the still- - * attached `session.onAppend` → `session/event`, so persistence captures - * the closing events. Only now is the agent truly quiescent. - * 3. run the session disposer — detach `onAppend` and remove the store - * entry. Done LAST so step 2's final flush is not dropped. + * Build an {@link AgentHandle} for a PREPARED session + a fresh agent. The + * handle's `dispose()` just runs the composite effect's disposer (see + * {@link start}) — which stops the loop, awaits its exit (final flush + * captured), unregisters the agent, and detaches the session, in that order. + * The same composite effect is what a fiber unload disposes, so both teardown + * triggers honor the ordering identically. */ - private startOwned( - id: AgentId, - options: AgentOptions, - owned: { session: Session; dispose: () => Promise }, - ): AgentHandle { - const { agent, disposeAgent } = this.start(id, options, owned.session) - return { - agent, - dispose: async () => { - await disposeAgent() // stop the loop (sync) + unregister - await agent.done // wait for the loop to actually exit (final flush captured) - await owned.dispose() // detach onAppend + remove the session store entry - }, - } + private startOwned(id: AgentId, options: AgentOptions, session: Session): AgentHandle { + const { agent, disposeAgent } = this.start(id, options, session) + return { agent, dispose: disposeAgent } } } diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 05fdb0c83e..8d1471d5f3 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -219,36 +219,48 @@ export class SessionStore extends Service { } /** - * Create a session. `options.seed` populates the session with a copy of - * those events (replay/fork); `options.meta` attaches creation metadata - * (validated absolute `cwd`, `parentSession` lineage) as the immutable - * {@link SessionHeader} (the store fills `version`/`id`/`createdAt`). The - * session is a Cordis effect: disposing the calling fiber stops event - * notification and removes the session from the store. + * Create a session owned by the calling fiber: disposing that fiber stops + * event notification and removes the session from the store. `options.seed` + * populates the session with a copy of those events (replay/fork); + * `options.meta` attaches creation metadata (validated absolute `cwd`, + * `parentSession` lineage) as the immutable {@link SessionHeader} (the store + * fills `version`/`id`/`createdAt`). + * + * For an agent whose session must be torn down IN ORDER with its loop (so the + * loop's final flush is captured before `onAppend` detaches), do NOT use this + * — fold the session lifecycle into the agent's own effect via + * {@link prepare} + {@link enter} + {@link announce} (see `dsh-agent-loop`'s + * `startOwned`). * * @throws if a session with `id` already exists, or if `meta.cwd` is a * non-absolute path (storage backends key directories off it). */ create(id?: string, options?: CreateSessionOptions): Session { - // Discard the store-removal disposer: a plain create() is owned by the - // calling fiber (disposing the fiber removes the session). An owner that - // needs to remove ONE session independently uses createOwned(). - return this.createOwned(id, options).session + const session = this.prepare(id, options) + // Single effect owned by the calling fiber. Yield the detach BEFORE + // announcing so a throwing `session/created` listener rolls the attach back + // (the generator effect disposes already-yielded disposers on a throw) + // instead of leaking the store entry + onAppend. + this.ctx.effect(function* (this: SessionStore) { + yield this.enter(session) + this.announce(session) + }.bind(this), 'sessions.create()') + return session } /** - * Like {@link create}, but ALSO returns the disposer for the session's - * store-removal effect — so an owner can remove exactly THIS session (detach - * `onAppend`, delete the store entry) without disposing the whole fiber. + * Build a session WITHOUT entering it into the store — validate the id/cwd and + * construct the {@link Session} (with its immutable {@link SessionHeader}). + * Pairs with {@link enter} + {@link announce}: a caller that owns a composite + * `ctx.effect` (the agent factory) folds the session lifecycle into that ONE + * effect so a fiber unload tears the session + agent down as a single ORDERED + * chain rather than as racing sibling effects — which would detach `onAppend` + * before the loop's closing `session/flush`, dropping the closing events. * - * Used by the agent factory's {@link AgentHandle} teardown: an owned agent's - * `dispose()` stops the loop, awaits quiescence, unregisters the agent, and - * THEN runs this session disposer — so the loop's final `session/flush` - * (delivered via `onAppend` → `session/event`) is captured before `onAppend` - * is detached. The disposer is async (a cordis effect disposer) to compose - * with the agent teardown's promise chain. + * @throws if a session with `id` already exists, or if `meta.cwd` is a + * non-absolute path. */ - createOwned(id?: string, options?: CreateSessionOptions): { session: Session; dispose: () => Promise } { + prepare(id?: string, options?: CreateSessionOptions): Session { const sessionId = SessionId(id ?? `session-${++this.counter}`) if (this.store.has(sessionId)) throw new Error(`session "${sessionId}" already exists`) const cwd = options?.meta?.cwd @@ -262,25 +274,38 @@ export class SessionStore extends Service { ...cwd !== undefined ? { cwd } : {}, ...options?.meta?.parentSession !== undefined ? { parentSession: options.meta.parentSession } : {}, } - const session = new Session(sessionId, options?.seed, header) - const dispose = this.ctx.effect(function* (this: SessionStore) { - session.onAppend = (event) => { this.ctx.emit('session/event', session, event) } - this.store.set(sessionId, session) - // 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) - } - this.ctx.emit('session/created', session) - }.bind(this), 'sessions.create()') - // ctx.effect's disposer returns Promise; normalize to an always-async - // disposer for the owner. - return { session, dispose: async () => { await dispose() } } + return new Session(sessionId, options?.seed, header) + } + + /** + * Enter a {@link prepare}d session into the store: wire `onAppend` → + * `session/event` and add it to the store. Returns the DETACH disposer + * (`onAppend = undefined` + store removal). Does NOT emit `session/created` — + * the caller yields this disposer inside its effect and THEN calls + * {@link announce}, so a throwing `session/created` listener rolls the attach + * back instead of leaking it. + * + * The id was already validated by {@link prepare}, which runs in the SAME + * synchronous sequence as `enter` (a config/factory caller does + * `prepare()` → `ctx.effect(generator)`, and a synchronous generator effect + * iterates inline — no await between them), so no concurrent create can claim + * the id in the gap. `enter` therefore does not re-check; it is not a public + * reservation primitive. + */ + enter(session: Session): () => void { + session.onAppend = (event) => { this.ctx.emit('session/event', session, event) } + this.store.set(session.id, session) + return () => { + session.onAppend = undefined + this.store.delete(session.id) + } + } + + /** Emit `session/created` for an {@link enter}ed session. Separate from + * {@link enter} so the caller can yield the detach disposer first (rollback + * safety — see {@link enter}). */ + announce(session: Session): void { + this.ctx.emit('session/created', session) } get(id: string): Session | undefined {