diff --git a/packages/acp/tests/dispose.spec.ts b/packages/acp/tests/dispose.spec.ts index d3af9ccd6c..3196994a0e 100644 --- a/packages/acp/tests/dispose.spec.ts +++ b/packages/acp/tests/dispose.spec.ts @@ -273,4 +273,47 @@ describe('acp bridge — disposal & HMR safety', () => { expect(harness.ctx.sessions.get('guard-a')).toBeUndefined() // detach still ran await harness.dispose() }) + + it('concurrent AgentHandle dispose() calls all await the SAME teardown (memoized)', async () => { + // The handle's dispose() must memoize: the underlying cordis effect disposer + // is single-shot, so a second dispose() while the first is mid-teardown would + // otherwise resolve IMMEDIATELY (effect epoch already cleared) — before the + // first call's await agent.done + final flush finished. Every caller must + // observe the same quiescence boundary. + const harness = await makeBridgeHarness({ storageDir, script: ['hang'] }) + const handle = harness.ctx.agents.create({ + agentId: 'conc-a', sessionId: 'conc-a', agentOptions: { model: 'mock' }, + }) + // Drive a turn that hangs in the model stream, so the loop is mid-turn when + // disposed — its exit runs a final session/flush we can gate to hold the + // teardown observably in-flight. + handle.agent.send([{ type: 'text', text: 'go' }]) + await new Promise(r => setTimeout(r, 30)) + expect(handle.agent.status).toBe('running') + let releaseFlush!: () => void + const flushGate = new Promise((resolve) => { releaseFlush = resolve }) + harness.ctx.on('session/flush', () => flushGate) + + // First dispose enters teardown (aborts the hanging step) and blocks in the + // gated final flush. + const first = handle.dispose() + let firstSettled = false + void first.then(() => { firstSettled = true }) + await new Promise(r => setTimeout(r, 20)) + expect(firstSettled).toBe(false) + + // Second dispose MUST await the same in-flight teardown, not resolve early. + const second = handle.dispose() + let secondSettled = false + void second.then(() => { secondSettled = true }) + await new Promise(r => setTimeout(r, 20)) + expect(secondSettled).toBe(false) // memoized: still pending with the first + + // Release the flush; both resolve together and the session is gone. + releaseFlush() + await Promise.all([first, second]) + expect(harness.ctx.agents.get('conc-a')).toBeUndefined() + expect(harness.ctx.sessions.get('conc-a')).toBeUndefined() + await harness.dispose() + }) }) diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index 3963b24594..5c25eb197d 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -267,15 +267,25 @@ export class AgentLoop extends Service implements AgentFactory { /** * Build an {@link AgentHandle} for a PREPARED session + a fresh agent. The - * handle's `dispose()` just runs the composite effect's disposer (see + * handle's `dispose()` 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. + * + * `dispose()` is MEMOIZED: the underlying cordis effect disposer is + * single-shot (a second call returns immediately because the effect's epoch is + * already cleared, NOT awaiting the in-flight teardown), so concurrent/repeated + * `dispose()` calls would otherwise resolve before the first call's + * `await agent.done` + final flush completed. Memoizing the promise makes every + * caller observe the SAME quiescence boundary, honoring the + * `AgentHandle.dispose(): Promise` contract (mirrors the ACP `quiesce()` + * helper). */ private startOwned(id: AgentId, options: AgentOptions, session: Session): AgentHandle { const { agent, disposeAgent } = this.start(id, options, session) - return { agent, dispose: disposeAgent } + let disposing: Promise | undefined + return { agent, dispose: () => (disposing ??= disposeAgent()) } } } diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 8d1471d5f3..57210c3431 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -285,14 +285,18 @@ export class SessionStore extends Service { * {@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. + * Re-checks the id for a duplicate: `prepare` and `enter` are public + * cross-package primitives and a caller may interleave arbitrary work (or + * another create) between them, so a stale prepared session must NOT overwrite + * a live store entry of the same id — its detach disposer would later delete + * the REAL session. The {@link create} convenience and the agent factory call + * the two back-to-back so they never trip this, but the public seam cannot + * assume that. + * + * @throws if a session with this id is already in the store. */ enter(session: Session): () => void { + if (this.store.has(session.id)) throw new Error(`session "${session.id}" already exists`) session.onAppend = (event) => { this.ctx.emit('session/event', session, event) } this.store.set(session.id, session) return () => { diff --git a/packages/session/tests/session.spec.ts b/packages/session/tests/session.spec.ts index ac40a8ea3f..593eed36c0 100644 --- a/packages/session/tests/session.spec.ts +++ b/packages/session/tests/session.spec.ts @@ -221,6 +221,40 @@ describe('SessionStore', () => { expect(forked.deriveMessages()).toEqual(a.deriveMessages()) }) + it('enter() rejects a stale prepared session whose id is already live (no overwrite)', async () => { + // prepare()/enter() are public cross-package primitives that a caller may + // separate with arbitrary work. A stale prepared session must NOT overwrite + // a live store entry of the same id — its detach disposer would later delete + // the REAL session, breaking the store-uniqueness invariant. + const ctx = new Context() + await ctx.plugin(SessionStore) + const stale = ctx.sessions.prepare('racy') + const live = ctx.sessions.create('racy') + expect(() => ctx.sessions.enter(stale)).toThrow(/already exists/) + // The live session is intact and still the store entry. + expect(ctx.sessions.get('racy')).toBe(live) + }) + + it('prepare() + enter() + announce() register a session and emit session/created', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const created: Session[] = [] + ctx.on('session/created', session => void created.push(session)) + + const session = ctx.sessions.prepare('lifecycle') + // prepare alone does NOT enter the store. + expect(ctx.sessions.get('lifecycle')).toBeUndefined() + const detach = ctx.sessions.enter(session) + expect(ctx.sessions.get('lifecycle')).toBe(session) + // enter does NOT announce. + expect(created).toEqual([]) + ctx.sessions.announce(session) + expect(created).toEqual([session]) + // The detach disposer removes the entry + stops notification. + detach() + expect(ctx.sessions.get('lifecycle')).toBeUndefined() + }) + it('synthesizes a minimal v1 header for a bare-created session', async () => { const ctx = new Context() await ctx.plugin(SessionStore)