From 2a4d89a4bd0a9d365e74fbb64baf7685f4bf3972 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 06:44:35 +0800 Subject: [PATCH 1/8] feat(agent): return an AgentHandle with an async per-agent disposer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The agent factory (`ctx.agents.create`/`resume`, the `AgentFactory` seam) now returns `AgentHandle = { agent; dispose(): Promise }` instead of a bare `Agent`. The disposer is a capability: only the holder can tear down exactly this agent — stop its loop, await the loop's exit (true quiescence, not just the `disposed` status flip), unregister it, and remove its session from the store. The teardown ORDER is load-bearing for durability. The loop appends its final `turn/end` + runs `session/flush` AFTER an abort, delivered through `session.onAppend` → `session/event`; if the session-store effect (which detaches `onAppend`) were torn down first, those closing events would never reach persistence. So `dispose()`: 1. runs the register+start effect disposer (sync: request loop stop), 2. `await agent.done` (loop exits, final flush captured), THEN 3. runs the session disposer (detach onAppend + delete store entry). `SessionStore.createOwned()` exposes the session-create effect's disposer (plain `create()` discards it — fiber-owned). `AgentLoop` funnels both factory entrypoints (`createAgent`, `resumeWith`) through a shared `startOwned` that composes the ordered teardown; the config path keeps a fiber-owned agent by discarding the handle. `ctx.agents.get(id)` still returns a bare `Agent` — the handle is only for the owner that created it. --- packages/agent-loop/src/index.ts | 75 +++++++++++++++---- .../tests/config-session-id.spec.ts | 2 +- packages/agent-loop/tests/resume.spec.ts | 20 ++--- packages/agent/src/index.ts | 38 ++++++++-- packages/agent/tests/agent.spec.ts | 14 +++- packages/session/src/index.ts | 25 ++++++- 6 files changed, 133 insertions(+), 41 deletions(-) diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index f118959fce..7cc90ba3da 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -11,7 +11,7 @@ import { Context, Service } from 'cordis' import { randomUUID } from 'node:crypto' import z from 'schemastery' import { AgentId } from '@deepseek-ai/dsh-agent' -import type { Agent, AgentFactory, AgentOptions, CreateAgentOptions, ResumeAgentOptions } from '@deepseek-ai/dsh-agent' +import type { AgentFactory, AgentHandle, AgentOptions, CreateAgentOptions, ResumeAgentOptions } from '@deepseek-ai/dsh-agent' import type {} from '@deepseek-ai/dsh-llm' import { SessionId } from '@deepseek-ai/dsh-session' import type { Session } from '@deepseek-ai/dsh-session' @@ -120,23 +120,28 @@ 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: {} }) - return this.start(AgentId(id), options, session) + const { agent } = this.start(AgentId(id), options, session) + return agent } /** * Programmatic factory create ({@link AgentFactory}): an agent on a * caller-supplied `sessionId` (NOT `${id}-session`), with optional session * metadata (validated `cwd`, lineage). The ACP bridge uses this so the - * client-generated session id becomes the live/persisted session id. + * client-generated session id becomes the live/persisted session id. Returns + * an {@link AgentHandle} the owner disposes to tear down exactly this agent. */ - createAgent(options: CreateAgentOptions): 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. this.assertAgentIdFree(options.agentId) - const session = this.ctx.sessions.create(options.sessionId, { meta: options.meta ?? {} }) - return this.start(AgentId(options.agentId), options.agentOptions ?? {}, session) + const owned = this.ctx.sessions.createOwned(options.sessionId, { meta: options.meta ?? {} }) + return this.startOwned(AgentId(options.agentId), options.agentOptions ?? {}, owned) } /** @@ -151,7 +156,7 @@ export class AgentLoop extends Service implements AgentFactory { * forever) — callers that need resume (ACP) inject `sessionPersistence`, so * by the time this runs the service exists. */ - async resume(options: ResumeAgentOptions): Promise { + async resume(options: ResumeAgentOptions): Promise { // Read the service through `ctx.get('sessionPersistence')` — a direct // global-store lookup keyed by the isolate symbol — NOT // `this.ctx.sessionPersistence`. AgentLoop deliberately does NOT inject @@ -183,7 +188,7 @@ export class AgentLoop extends Service implements AgentFactory { * sessions store + registry are still read through `this.ctx` (both are in * AgentLoop's static inject, so they resolve fine). */ - private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise { + private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise { this.assertAgentIdFree(options.agentId) const { meta, events } = await persistence.load(SessionId(options.resumeSessionId)) // Re-check the agent id AFTER the await: the pre-load check above can go @@ -196,7 +201,7 @@ export class AgentLoop extends Service implements AgentFactory { // 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 session = this.ctx.sessions.create(options.resumeSessionId, { + const owned = this.ctx.sessions.createOwned(options.resumeSessionId, { seed: events, meta: { createdAt: meta.createdAt, @@ -204,7 +209,7 @@ export class AgentLoop extends Service implements AgentFactory { ...meta.parentSession !== undefined ? { parentSession: meta.parentSession } : {}, }, }) - return this.start(AgentId(options.agentId), options.agentOptions ?? {}, session) + return this.startOwned(AgentId(options.agentId), options.agentOptions ?? {}, owned) } /** @@ -219,16 +224,54 @@ export class AgentLoop extends Service implements AgentFactory { } } - /** Shared: construct a ReactLoopAgent, register it, and start its loop (LIFO). */ - private start(id: AgentId, options: AgentOptions, session: Session): ReactLoopAgent { + /** + * 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. + */ + private start(id: AgentId, options: AgentOptions, session: Session): { agent: ReactLoopAgent; disposeAgent: () => Promise } { const agent = new ReactLoopAgent(this.ctx, id, options, session) - // Generator effect: stop and unregister are independent disposables - // (LIFO), so a throwing stop() cannot leak the registry entry. - this.ctx.effect(function* (this: AgentLoop) { + const dispose = this.ctx.effect(function* (this: AgentLoop) { yield this.ctx.agents.register(agent) yield agent.start() }.bind(this), 'agentLoop.start()') - return agent + 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. + */ + 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 + }, + } } } diff --git a/packages/agent-loop/tests/config-session-id.spec.ts b/packages/agent-loop/tests/config-session-id.spec.ts index 13a62bca8c..d4753432bc 100644 --- a/packages/agent-loop/tests/config-session-id.spec.ts +++ b/packages/agent-loop/tests/config-session-id.spec.ts @@ -78,7 +78,7 @@ describe('config-driven session id', () => { await ctx1.plugin(AgentLoop, { agents: [] }) await ctx1.plugin(SessionPersistenceJsonl, { root }) ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first')])) - const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sticky-1' }) as ReactLoopAgent + const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sticky-1' }).agent as ReactLoopAgent a1.send([{ type: 'text', text: 'remember me' }], { source: { kind: 'user' } }) await waitForIdle(ctx1, a1) await ctx1.fiber.dispose() diff --git a/packages/agent-loop/tests/resume.spec.ts b/packages/agent-loop/tests/resume.spec.ts index 10313fdd26..bc655a32f2 100644 --- a/packages/agent-loop/tests/resume.spec.ts +++ b/packages/agent-loop/tests/resume.spec.ts @@ -43,7 +43,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { it('createAgent uses the caller-supplied sessionId (not ${id}-session)', async () => { const adapter = new MockAdapter([textResponse('hi')]) const { ctx } = await persistentHarness(adapter) - const agent = ctx.agents.create({ agentId: 'a1', sessionId: 'custom-session', meta: { cwd: '/w' } }) + const { agent } = ctx.agents.create({ agentId: 'a1', sessionId: 'custom-session', meta: { cwd: '/w' } }) expect(agent.session.id).toBe('custom-session') expect(agent.session.header.cwd).toBe('/w') await ctx.fiber.dispose() @@ -63,7 +63,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { it('createAgent works without meta (no cwd)', async () => { const adapter = new MockAdapter([textResponse('hi')]) const { ctx } = await persistentHarness(adapter) - const agent = ctx.agents.create({ agentId: 'a-nometa', sessionId: 'nometa-session' }) + const { agent } = ctx.agents.create({ agentId: 'a-nometa', sessionId: 'nometa-session' }) expect(agent.session.id).toBe('nometa-session') expect(agent.session.header.cwd).toBeUndefined() await ctx.fiber.dispose() @@ -73,7 +73,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { // Lifecycle 1: create a no-cwd session and run a turn. const adapter1 = new MockAdapter([textResponse('a')]) const { ctx: ctx1, root } = await persistentHarness(adapter1) - const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'nocwd-sess' }) as ReactLoopAgent + const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'nocwd-sess' }).agent as ReactLoopAgent a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } }) await waitForIdle(ctx1, a1) await ctx1.fiber.dispose() @@ -89,7 +89,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { await ctx2.plugin(AgentLoop, { agents: [] }) await ctx2.plugin(SessionPersistenceJsonl, { root }) ctx2.llm.registerAdapter(['mock'], adapter2) - const a2 = await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'nocwd-sess' }) as ReactLoopAgent + const a2 = (await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'nocwd-sess' })).agent as ReactLoopAgent expect(a2.session.header.cwd).toBeUndefined() await ctx2.fiber.dispose() }) @@ -120,7 +120,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { await ctx2.plugin(AgentLoop, { agents: [] }) await ctx2.plugin(SessionPersistenceJsonl, { root }) ctx2.llm.registerAdapter(['mock'], adapter2) - const a2 = await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'forked-sess' }) as ReactLoopAgent + const a2 = (await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'forked-sess' })).agent as ReactLoopAgent expect(a2.session.header.parentSession).toBe('parent-sess') expect(a2.session.header.cwd).toBe('/w') await ctx2.fiber.dispose() @@ -133,7 +133,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { // disk, since a crash before the next turn would otherwise lose it. const adapter1 = new MockAdapter([textResponse('answer')]) const { ctx: ctx1, root } = await persistentHarness(adapter1) - const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'inject-sess', meta: { cwd: '/w' } }) as ReactLoopAgent + const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'inject-sess', meta: { cwd: '/w' } }).agent as ReactLoopAgent a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } }) await waitForIdle(ctx1, a1) a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } }) @@ -158,7 +158,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { // drop it on reload (the bug this guards). const adapter1 = new MockAdapter([textResponse('answer')]) const { ctx: ctx1, root } = await persistentHarness(adapter1) - const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'inject-sess', meta: { cwd: '/w' } }) as ReactLoopAgent + const a1 = ctx1.agents.create({ agentId: 'm', sessionId: 'inject-sess', meta: { cwd: '/w' } }).agent as ReactLoopAgent a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } }) await waitForIdle(ctx1, a1) a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } }) @@ -176,7 +176,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { await ctx2.plugin(AgentLoop, { agents: [] }) await ctx2.plugin(SessionPersistenceJsonl, { root }) ctx2.llm.registerAdapter(['mock'], adapter2) - const a2 = await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'inject-sess' }) as ReactLoopAgent + const a2 = (await ctx2.agents.resume({ agentId: 'm', resumeSessionId: 'inject-sess' })).agent as ReactLoopAgent const flat = JSON.stringify(a2.session.deriveMessages()) expect(flat).toContain('background task 42 finished') await ctx2.fiber.dispose() @@ -186,7 +186,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { // Lifecycle 1: run one full turn, persisting it. const adapter1 = new MockAdapter([textResponse('first answer')]) const { ctx: ctx1, root } = await persistentHarness(adapter1) - const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sess-resume', meta: { cwd: '/w' } }) as ReactLoopAgent + const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sess-resume', meta: { cwd: '/w' } }).agent as ReactLoopAgent a1.send([{ type: 'text', text: 'first question' }], { source: { kind: 'user' } }) await waitForIdle(ctx1, a1) const events1 = [...a1.session.events] @@ -206,7 +206,7 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => { await ctx2.plugin(SessionPersistenceJsonl, { root }) ctx2.llm.registerAdapter(['mock'], adapter2) - const a2 = await ctx2.agents.resume({ agentId: 'main', resumeSessionId: 'sess-resume' }) as ReactLoopAgent + const a2 = (await ctx2.agents.resume({ agentId: 'main', resumeSessionId: 'sess-resume' })).agent as ReactLoopAgent // The resumed session carries the prior history… expect(a2.session.id).toBe('sess-resume') expect(a2.session.events.length).toBe(events1.length) diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index c9181081a3..2063812963 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -54,6 +54,23 @@ export interface ResumeAgentOptions { agentOptions?: AgentOptions } +/** + * An owned agent plus its disposer, returned by {@link AgentRegistry.create} / + * {@link AgentRegistry.resume}. The disposer is a CAPABILITY: only the holder + * can tear this agent down. `dispose()` unregisters the agent, stops its loop, + * awaits the loop's exit (quiescence — NOT just the `disposed` status flip), and + * removes the agent's session from the store, in an order that captures the + * loop's final `session/flush` before the session is detached. + * + * `ctx.agents.get(id)` still returns a bare {@link Agent} — the handle is only + * for the OWNER that created it. Config-created agents (the loop's own startup) + * are owned by the loop fiber and never need a handle. + */ +export interface AgentHandle { + agent: Agent + dispose(): Promise +} + /** * The agent-creation factory the loop implementation provides to the registry * via {@link AgentRegistry.setFactory}. Kept on the `dsh-agent` interface so @@ -61,14 +78,18 @@ export interface ResumeAgentOptions { * depending on the concrete `dsh-agent-loop` package. */ export interface AgentFactory { - /** Create, start, and register a new agent on a caller-supplied session id. */ - createAgent(options: CreateAgentOptions): Agent + /** + * Create, start, and register a new agent on a caller-supplied session id. + * Returns an {@link AgentHandle} — the owner disposes it to tear down exactly + * this agent (unregister + stop loop + await quiescence + remove session). + */ + createAgent(options: CreateAgentOptions): AgentHandle /** * Load a persisted session and resume an agent on it. Async because it awaits * `ctx.sessionPersistence.load`; must be called after that service exists - * (consumers inject `sessionPersistence`). + * (consumers inject `sessionPersistence`). Returns an {@link AgentHandle}. */ - resume(options: ResumeAgentOptions): Promise + resume(options: ResumeAgentOptions): Promise } /** Thrown when create/resume is called before an agent factory is registered. */ @@ -107,9 +128,10 @@ export class AgentRegistry extends Service { * Create, start, and register a new agent through the registered factory. * Distinct from {@link register} (which records an already-constructed * agent): this constructs the agent and its session. Throws if no factory is - * registered. + * registered. Returns an {@link AgentHandle} — the owner disposes it to tear + * down exactly this agent. */ - create(options: CreateAgentOptions): Agent { + create(options: CreateAgentOptions): AgentHandle { if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE) return this.factory.createAgent(options) } @@ -117,9 +139,9 @@ export class AgentRegistry extends Service { /** * Load a persisted session and resume an agent on it through the registered * factory. Rejects if no factory is registered; the factory rejects if - * session persistence is not configured. + * session persistence is not configured. Returns an {@link AgentHandle}. */ - async resume(options: ResumeAgentOptions): Promise { + async resume(options: ResumeAgentOptions): Promise { if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE) return this.factory.resume(options) } diff --git a/packages/agent/tests/agent.spec.ts b/packages/agent/tests/agent.spec.ts index 28072154f6..98f072ab10 100644 --- a/packages/agent/tests/agent.spec.ts +++ b/packages/agent/tests/agent.spec.ts @@ -82,8 +82,14 @@ describe('AgentRegistry factory seam', () => { function stubFactory() { const calls: { create: unknown[]; resume: unknown[] } = { create: [], resume: [] } const factory: import('@deepseek-ai/dsh-agent').AgentFactory = { - createAgent(options) { calls.create.push(options); return stubAgent(options.agentId) }, - resume(options) { calls.resume.push(options); return Promise.resolve(stubAgent(options.agentId)) }, + createAgent(options) { + calls.create.push(options) + return { agent: stubAgent(options.agentId), dispose: () => Promise.resolve() } + }, + resume(options) { + calls.resume.push(options) + return Promise.resolve({ agent: stubAgent(options.agentId), dispose: () => Promise.resolve() }) + }, } return { factory, calls } } @@ -102,11 +108,11 @@ describe('AgentRegistry factory seam', () => { ctx.agents.setFactory(factory) const created = ctx.agents.create({ agentId: 'c1', sessionId: 'sess-1', meta: { cwd: '/w' } }) - expect(created.id).toBe('c1') + expect(created.agent.id).toBe('c1') expect(calls.create).toEqual([{ agentId: 'c1', sessionId: 'sess-1', meta: { cwd: '/w' } }]) const resumed = await ctx.agents.resume({ agentId: 'r1', resumeSessionId: 'old-sess' }) - expect(resumed.id).toBe('r1') + expect(resumed.agent.id).toBe('r1') expect(calls.resume).toEqual([{ agentId: 'r1', resumeSessionId: 'old-sess' }]) }) diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 4796c05f51..05fdb0c83e 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -230,6 +230,25 @@ export class SessionStore extends Service { * 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 + } + + /** + * 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. + * + * 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. + */ + createOwned(id?: string, options?: CreateSessionOptions): { session: Session; dispose: () => Promise } { const sessionId = SessionId(id ?? `session-${++this.counter}`) if (this.store.has(sessionId)) throw new Error(`session "${sessionId}" already exists`) const cwd = options?.meta?.cwd @@ -244,7 +263,7 @@ export class SessionStore extends Service { ...options?.meta?.parentSession !== undefined ? { parentSession: options.meta.parentSession } : {}, } const session = new Session(sessionId, options?.seed, header) - this.ctx.effect(function* (this: SessionStore) { + 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 @@ -259,7 +278,9 @@ export class SessionStore extends Service { } this.ctx.emit('session/created', session) }.bind(this), 'sessions.create()') - return session + // ctx.effect's disposer returns Promise; normalize to an always-async + // disposer for the owner. + return { session, dispose: async () => { await dispose() } } } get(id: string): Session | undefined { From ee4cad3ada7924b5bbb62e6c646f4975cb4009f0 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 06:44:58 +0800 Subject: [PATCH 2/8] feat(acp): dispose each session's agent on disconnect/teardown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bridge now holds each session's `AgentHandle` disposer in its `SessionRecord` and runs it on teardown (client disconnect or fiber dispose) instead of the old `abort()` + `whenIdle()` drain that left agents registered. A bare client disconnect now leaves NO registered agent and NO session-store entry — not an idled-but-still-registered one. The queue-aware `cancel()` inside the disposer also closes the former pre-step best-effort window (a turn about to start is dropped), so teardown reaches true quiescence. The `session/load`-races-teardown leak is fixed: if the bridge closed while `resume()` was pending, the just-resumed handle is disposed before throwing, so it leaves no orphan (it has no SessionRecord, so quiesce() never sees it). Tests: the disconnect test now asserts (through the SAME memoized teardown) that the agent is unregistered AND its session removed; a durability test re-loads the persisted log after dispose and asserts the closing turn/end is on disk (guards the teardown-order contract); a sibling-isolation test proves one handle's dispose() leaves other agents untouched. Docs: agent / agent-loop / acp READMEs, architecture.md, and the stale in-code quiesce() ownership comment updated to the per-agent disposal model; the now-resolved TODO(rfc010-agent-disposal) / TODO(rfc010-cancel-prestep) teardown notes removed. --- docs/architecture.md | 2 +- .../proposed/2026-06-14-acp-multi-session.md | 2 +- examples/coding-agent/tests/resume.e2e.ts | 6 +- packages/acp/README.md | 4 +- packages/acp/src/index.ts | 94 +++++++++++-------- packages/acp/tests/dispose.spec.ts | 87 +++++++++++++++-- packages/acp/tests/edges.spec.ts | 2 +- packages/agent-loop/README.md | 6 +- packages/agent/README.md | 6 +- 9 files changed, 150 insertions(+), 59 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index e28a384eb3..9a81b1c6d3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -48,7 +48,7 @@ Dependency rule: plugins depend on interface packages, never on `dsh-agent-loop` | `ctx.sessionPersistence` | `SessionPersistence` (abstract) | dsh-session-persistence | durable persistence seam: create/append/load/list sessions | | `ctx.systemPrompt` | `SystemPrompt` | dsh-system-prompt | ordered sections + tool schemas → `assemble()` | | `ctx.tools` | `ToolRegistry` | dsh-tools | tool definitions; `execute()` through waterfall | -| `ctx.agents` | `AgentRegistry` | dsh-agent | live `Agent` handles + the create/resume factory seam | +| `ctx.agents` | `AgentRegistry` | dsh-agent | live `Agent` handles + the create/resume factory seam (returns an `AgentHandle` = `{ agent, dispose() }` for owned per-agent teardown) | | `ctx.agentLoop` | `AgentLoop` | dsh-agent-loop | creates `ReactLoopAgent`s and drives their loops | | `ctx.bash` | `BashExecutor` (abstract) | dsh-bash | bash execution seam: foreground runs + background tasks | diff --git a/docs/rfc/proposed/2026-06-14-acp-multi-session.md b/docs/rfc/proposed/2026-06-14-acp-multi-session.md index 8cc2237eff..e4d7bbbe41 100644 --- a/docs/rfc/proposed/2026-06-14-acp-multi-session.md +++ b/docs/rfc/proposed/2026-06-14-acp-multi-session.md @@ -3,7 +3,7 @@ Status: proposed -> **Implementation status:** the multi-session bridge (steps 1, 3, 4) and the bash task-ownership isolation are implemented in `packages/acp` + `packages/tool-bash`. **Per-session *permission* ownership is deferred** — it depends on [the ACP support permission gate](2026-06-14-acp-agent-client-protocol.md) (`TODO(rfc010-permission-gate)`), which is itself deferred; the `agent→sessionId` reverse map the gate will route through is in place. Step 2's "real per-session disposer scope" is also deferred (`TODO(rfc010-agent-disposal)`): the bridge demuxes via id-keyed maps and global `ctx.on` listeners (correct and leak-free — disposal drains every session in parallel to quiescence), and a per-agent disposer seam is the follow-up. Status stays `proposed` until per-session permission ownership lands. +> **Implementation status:** the multi-session bridge (steps 1, 3, 4) and the bash task-ownership isolation are implemented in `packages/acp` + `packages/tool-bash`. **Per-session *permission* ownership is deferred** — it depends on [the ACP support permission gate](2026-06-14-acp-agent-client-protocol.md) (`TODO(rfc010-permission-gate)`), which is itself deferred; the `agent→sessionId` reverse map the gate will route through is in place. Step 2's per-session disposer scope is now implemented (see [agent lifecycle & ownership seams](2026-06-18-agent-lifecycle-and-ownership-seams.md)): the factory returns a per-agent `AgentHandle` whose `dispose()` stops the loop, awaits quiescence, unregisters the agent, and removes its session, so a bare client disconnect leaves no registered agent or session-store entry. Status stays `proposed` until per-session permission ownership lands. ## Problem diff --git a/examples/coding-agent/tests/resume.e2e.ts b/examples/coding-agent/tests/resume.e2e.ts index b720efa8c9..cf2d910138 100644 --- a/examples/coding-agent/tests/resume.e2e.ts +++ b/examples/coding-agent/tests/resume.e2e.ts @@ -41,7 +41,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('resume: continue a persisted ses agentId: 'resume-1', sessionId: SESSION_ID, agentOptions: { model: 'deepseek-v4-flash', systemPrompt: SYSTEM_PROMPT }, - }) as ReactLoopAgent + }).agent as ReactLoopAgent first.send([{ type: 'text', text: `Remember this code for later: ${SECRET}. Just acknowledge it.` }]) await waitForIdle(ctx, first) await ctx.fiber.dispose() @@ -51,11 +51,11 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('resume: continue a persisted ses // session. The loaded event log seeds the live session, so the model sees // run 1's exchange as conversation history. ctx = await codingHarness(process.cwd(), root) - const resumed = await ctx.agents.resume({ + const resumed = (await ctx.agents.resume({ agentId: 'resume-2', resumeSessionId: SESSION_ID, agentOptions: { model: 'deepseek-v4-flash', systemPrompt: SYSTEM_PROMPT }, - }) as ReactLoopAgent + })).agent as ReactLoopAgent expect(resumed.session.id).toBe(SESSION_ID) // The prior user turn is in the rehydrated log before the model is asked. expect(JSON.stringify(resumed.session.deriveMessages())).toContain(SECRET) diff --git a/packages/acp/README.md b/packages/acp/README.md index 0bacaeef07..5cdfa3a202 100644 --- a/packages/acp/README.md +++ b/packages/acp/README.md @@ -61,13 +61,11 @@ A `session/prompt` resolves (or rejects) exactly once, keyed off the canonical s ## Disposal & disconnect -Teardown reaches quiescence: for EVERY live session settle any pending prompt as `cancelled`, `agent.abort()`, then `await agent.whenIdle()` — the interface-level quiescence signal (NOT `agent/status('disposed')`, which fires before the driver exits). The agents drain in parallel. The same teardown runs on a **client disconnect** (`conn.closed` resolves when the editor quits / the transport EOFs), so a vanished client never leaves an orphaned running agent whose `session/update` writes are silently swallowed. The two paths are idempotent and memoized (the first clears the `sessions` map; a second caller awaits the same teardown promise). +Teardown reaches quiescence: for EVERY live session settle any pending prompt as `cancelled`, then run that session's [`AgentHandle`](../agent/README.md) `dispose()` — which stops the loop with the queue-aware `cancel()`, `await`s the loop's exit (the final `turn/end` + `session/flush` are captured while the session is still attached), unregisters the agent, and removes its session from the store. The per-session disposes run in parallel. The same teardown runs on a **client disconnect** (`conn.closed` resolves when the editor quits / the transport EOFs), so a vanished client never leaves an orphaned running — or idled-but-still-registered — agent whose `session/update` writes are silently swallowed. The two paths are idempotent and memoized (the first clears the `sessions` map; a second caller awaits the same teardown promise). ## Known limitations (tracked TODOs) - **`TODO(rfc010-permission-gate)`** — the `tools/execute` permission gate (`session/request_permission`) is NOT implemented; tools run with the executor's full authority. The `agent→sessionId` reverse map is in place so the gate can route a permission request (which receives only `exec.agent`) back to its originating session. [ACP support](../../docs/rfc/proposed/2026-06-14-acp-agent-client-protocol.md) and [ACP multi-session](../../docs/rfc/proposed/2026-06-14-acp-multi-session.md) stay `proposed` until the gate (and per-session permission ownership) land. -- **`TODO(rfc010-cancel-prestep)`** — `session/cancel` is now the queue-aware `agent.cancel()` (a running step is aborted, queued + steering work is cleared, and a turn about to start is dropped), so a queued-but-not-yet-started prompt no longer runs and a later prompt cannot be batched into the cancelled turn. **Teardown/disconnect still use the older `agent.abort('disposed')` + `whenIdle()`**, so the best-effort window remains there: disposal/disconnect can return while one short queued turn per session still runs. PR D's per-agent disposer switches teardown to the queue-aware path and closes this; the single-in-flight-per-session rule bounds the worst case to one extra prompt per session until then. -- **`TODO(rfc010-agent-disposal)`** — the factory (`ctx.agents.create`/`resume`) returns no per-agent disposer, so teardown aborts+drains each agent but cannot individually unregister it; on a bare client disconnect (no host dispose) the idled agents linger in `ctx.agents` until the host context disposes. A reconnect spins up a fresh context, so this strands no work; a per-agent disposal seam is the follow-up. - **`additionalDirectories`** — rejected. A session operates in its single `cwd` (see Per-session cwd); widening the tool/filesystem scope to extra roots is a separate sandbox concern, not yet implemented. ## stdout is the protocol diff --git a/packages/acp/src/index.ts b/packages/acp/src/index.ts index 040f4b10d1..7edc2dae0d 100644 --- a/packages/acp/src/index.ts +++ b/packages/acp/src/index.ts @@ -138,6 +138,13 @@ export const Config: Schema = Schema.object({ interface SessionRecord { sessionId: string agent: Agent + /** + * The owned-agent disposer (from the {@link AgentHandle} the factory returned). + * Teardown calls it to unregister this ONE agent, stop its loop, await + * quiescence, and remove its session — instead of leaving it for the bridge + * fiber to reclaim. + */ + dispose: () => Promise /** * Resolves tool-owned presentation for THIS session's tool calls and remembers * each in-flight call's `(name, args)` so the matching `tool/result` can find @@ -434,14 +441,21 @@ export function apply(ctx: Context, config: AcpConfig): void { validateWorkspaceParams(params) validateMcpServers(params) const sessionId = randomUUID() - const agent = agents.create({ + const handle = agents.create({ agentId: sessionId, sessionId, meta: { cwd: params.cwd }, agentOptions: agentOptions(config), }) - bySession.set(agent, sessionId) - sessions.set(sessionId, { sessionId, agent, presenter: makePresenter(), terminalEnabled: terminalOutputCap, inflight: undefined }) + bySession.set(handle.agent, sessionId) + sessions.set(sessionId, { + sessionId, + agent: handle.agent, + dispose: () => handle.dispose(), + presenter: makePresenter(), + terminalEnabled: terminalOutputCap, + inflight: undefined, + }) return Promise.resolve({ sessionId }) }, @@ -483,30 +497,38 @@ export function apply(ctx: Context, config: AcpConfig): void { throw invalidParams(`session ${params.sessionId} cwd mismatch: persisted ${persistedCwd}, requested ${params.cwd}`) } } - const agent = await agents.resume({ + const handle = await agents.resume({ agentId: params.sessionId, resumeSessionId: params.sessionId, agentOptions: agentOptions(config), }) // The bridge may have torn down (disposal / client disconnect) while // resume() was pending. Its listeners are gone, so installing a record - // now would resurrect a live agent the bridge can no longer drive or - // tear down. Bail: the just-resumed agent is reclaimed with the host - // context (no per-agent disposer — TODO(rfc010-agent-disposal)). - /* v8 ignore next 3 -- the in-memory test transport rejects the in-flight + // now would resurrect a live agent the bridge can no longer drive. Bail — + // and tear down the just-resumed agent (unregister + stop + remove its + // session) before throwing, so it does not leak: it has no SessionRecord, + // so quiesce() would never see it. + /* v8 ignore next 4 -- the in-memory test transport rejects the in-flight session/load request the instant it closes (before this post-await code runs), so the guard can't be hit in tests; it protects the real stdio path, where a closed pipe need not reject a mid-flight handler. */ if (closed) { + await handle.dispose() throw invalidParams('connection closed during session/load') } + const agent = handle.agent bySession.set(agent, params.sessionId) // Snapshot the terminal capability ONCE for this session (used by both // the replay below and the post-load live stream) so a later // `initialize` can't desync the call/result of a tool card. const terminalEnabled = terminalOutputCap const record: SessionRecord = { - sessionId: params.sessionId, agent, presenter: makePresenter(), terminalEnabled, inflight: undefined, + sessionId: params.sessionId, + agent, + dispose: () => handle.dispose(), + presenter: makePresenter(), + terminalEnabled, + inflight: undefined, } sessions.set(params.sessionId, record) // Replay the persisted event log to the client as session/update. Use @@ -604,35 +626,27 @@ export function apply(ctx: Context, config: AcpConfig): void { /** * Tear ALL live sessions down to quiescence (AGENTS.md "dispose must reach - * quiescence"): for each session settle any pending prompt `cancelled`, abort - * the agent, and AWAIT it draining via the interface-level `whenIdle()` signal - * (NOT `agent/status('disposed')`, which fires before the driver exits). The - * agents drain in parallel. Idempotent — clears the `sessions` map first and - * memoizes, so a second call (close racing dispose) is a no-op. + * quiescence"): for each session settle any pending prompt `cancelled`, then + * run that session's {@link AgentHandle} `dispose()` — which stops the loop + * with the queue-aware cancel, AWAITS the loop's exit (the final + * `turn/end` + `session/flush` are captured while `onAppend` is still + * attached), unregisters the agent, and removes its session from the store. + * The per-session disposes run in parallel. Idempotent — clears the `sessions` + * map first and memoizes, so a second call (close racing dispose) is a no-op. * Shared by Cordis disposal AND client disconnect (`conn.closed`). * - * Caveat (same window as TODO(rfc010-cancel-prestep)): if teardown lands in - * the pre-step window — `agent.send()` queued a turn but the loop has not yet - * flipped to `running` — `abort()` has no live `AbortController` to signal and - * `whenIdle()` returns immediately (status is still `idle`), so that queued - * turn may still start and run after teardown returns. Reaching true - * quiescence in that window needs a queue-aware loop cancel primitive (a - * loop-level change); the single-in-flight-per-session rule bounds the worst - * case to one short queued turn per session. - * - * The agents are NOT individually disposed/unregistered here. The factory - * (`ctx.agents.create`/`resume`) registers each via `AgentLoop.start`'s - * `this.ctx.effect(...)`; because the factory is reached through this bridge's - * traceable service proxy, that effect's `this.ctx` is the CALLER context (the - * bridge fiber), so every registry entry is bound to the bridge fiber and is - * reclaimed when the bridge fiber disposes (whole-context dispose, or an - * ACP-only HMR `acpFiber.dispose()` — both unregister all the bridge's - * agents). What this teardown path handles is a bare client disconnect, which - * resolves `conn.closed` WITHOUT disposing the fiber: each live agent is - * idled+aborted here but stays in `ctx.agents` until the fiber is disposed. - * Since a reconnect spins up a fresh context, the lingering idle agents strand - * no work. A per-agent disposal seam (unregister on disconnect) is a follow-up - * (TODO(rfc010-agent-disposal)). + * Per-agent disposal closes the former pre-step best-effort window: the + * queue-aware `cancel()` (RFC 011) drops a turn about to start, so a queued- + * but-not-yet-running prompt never runs after teardown. A bare client + * disconnect (resolves `conn.closed` WITHOUT disposing the fiber) thus leaves + * NO registered agent and NO session-store entry — not an idled-but-still- + * registered one. When the fiber IS disposed (whole-context or an ACP-only HMR + * `acpFiber.dispose()`), this same memoized teardown runs first; the factory's + * register+start+session effects are ALSO bound to the bridge fiber (the + * factory is reached through this bridge's traceable service proxy, so + * `AgentLoop.start`'s `this.ctx.effect(...)` binds to the CALLER context — the + * bridge fiber), so any agent this path did not reach is still reclaimed by + * fiber disposal. */ let quiescing: Promise | undefined const quiesce = (): Promise => { @@ -650,8 +664,12 @@ export function apply(ctx: Context, config: AcpConfig): void { quiescing = (async () => { await Promise.all(recs.map(async (rec) => { settlePrompt(rec, 'cancelled') - rec.agent.abort('disposed') - await rec.agent.whenIdle() + // Per-agent dispose (the AgentHandle disposer): unregister this agent, + // stop its loop with the queue-aware cancel, await quiescence (the loop + // exit + final flush), and remove its session — so a bare client + // disconnect leaves NO registered agent and NO session-store entry, not + // just an idled-but-still-registered one. + await rec.dispose() })) })() return quiescing diff --git a/packages/acp/tests/dispose.spec.ts b/packages/acp/tests/dispose.spec.ts index 45e351ced5..6ff89b400e 100644 --- a/packages/acp/tests/dispose.spec.ts +++ b/packages/acp/tests/dispose.spec.ts @@ -3,7 +3,8 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk' -import { makeBridgeHarness } from './harness.ts' +import { SessionId } from '@deepseek-ai/dsh-session' +import { makeBridgeHarness, textResponse } from './harness.ts' describe('acp bridge — disposal & HMR safety', () => { let storageDir: string @@ -82,10 +83,11 @@ describe('acp bridge — disposal & HMR safety', () => { await harness.dispose() }) - it('a client disconnect mid-prompt tears the session down to quiescence', async () => { + it('a client disconnect mid-prompt disposes the session (no registered agent left)', async () => { // The ACP transport closes (editor quits) while a turn runs. The bridge must - // settle the in-flight prompt cancelled and abort+drain the agent rather - // than leaving an orphaned running agent whose updates are swallowed. + // settle the in-flight prompt cancelled and DISPOSE the agent (PR D's + // per-agent AgentHandle teardown) rather than leaving an orphaned running — + // or even idled-but-still-registered — agent whose updates are swallowed. 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: [] }) @@ -96,13 +98,25 @@ describe('acp bridge — disposal & HMR safety', () => { await new Promise(r => setTimeout(r, 30)) expect(agent.status).toBe('running') - // Sever the transport — the bridge's conn.closed teardown runs and drives - // the agent to quiescence on its OWN (assert before any dispose() runs). + // Sever the transport — the bridge's conn.closed teardown runs and drives the + // agent's AgentHandle dispose to quiescence on its OWN (before any dispose()). await harness.closeClientTransport() await agent.whenIdle() - expect(agent.status).toBe('idle') + // The agent's loop has stopped: status `disposed`. + expect(agent.status).toBe('disposed') - await harness.dispose() // idempotent with the close teardown + // Await the bridge teardown to completion WITHOUT tearing down the root + // agents/sessions services (so we can still query them). acpFiber.dispose() + // invokes the SAME memoized quiesce() the disconnect started and awaits its + // promise — which resolves only after every rec.dispose() (loop exit + + // session removal) has finished, closing the whenIdle()/owned.dispose() + // microtask race. The AgentHandle dispose has run: the agent is unregistered + // and its session removed from the store, not merely idled (the old + // behavior). The services live on the root ctx, so they survive this. + await harness.acpFiber.dispose() + expect(harness.ctx.agents.get(sessionId)).toBeUndefined() + expect(harness.ctx.sessions.get(sessionId)).toBeUndefined() + await harness.dispose() }) it('a client disconnect racing fiber dispose both reach quiescence (shared teardown)', async () => { @@ -140,4 +154,61 @@ describe('acp bridge — disposal & HMR safety', () => { await new Promise(r => setTimeout(r, 10)) expect(harness.updates.length).toBe(before) }) + + it('the final turn closing events are persisted across an AgentHandle dispose (durability)', async () => { + // The teardown-ORDER guarantee: a per-agent dispose must stop the loop, + // AWAIT its exit (so the loop's final `turn/end` + `session/flush` fire + // through the still-attached `session.onAppend` → `session/event`), and only + // THEN detach onAppend + remove the session. If the order were inverted + // (detach first), the closing events would never reach persistence. Drive a + // CLEAN turn to completion, dispose JUST the bridge, then re-load the + // persisted log from disk and assert the closing turn/end is on disk — the + // world, not the agent's self-report. + const harness = await makeBridgeHarness({ storageDir, script: [textResponse('done')] }) + await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) + const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }) + await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }) + const liveEvents = harness.ctx.agents.get(sessionId)!.session.events.length + expect(liveEvents).toBeGreaterThan(0) + + // Tear down JUST the bridge (the AgentHandle dispose runs to quiescence). + await harness.acpFiber.dispose() + expect(harness.ctx.agents.get(sessionId)).toBeUndefined() + + // Re-load the session from disk: every live event (incl. the closing + // turn/end) was flushed before the session was detached. + const reloaded = await harness.ctx.sessionPersistence.load(SessionId(sessionId)) + expect(reloaded.events.length).toBe(liveEvents) + const last = reloaded.events.at(-1)! + expect(last.type).toBe('turn/end') + 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 + // directly through the registry factory (the same path the ACP bridge uses), + // dispose one handle, and assert the other survives, registered and + // queryable, with its session still in the store. + const harness = await makeBridgeHarness({ storageDir, script: [] }) + const handleA = harness.ctx.agents.create({ + agentId: 'sib-a', sessionId: 'sib-a', agentOptions: { model: 'mock' }, + }) + const handleB = harness.ctx.agents.create({ + agentId: 'sib-b', sessionId: 'sib-b', agentOptions: { model: 'mock' }, + }) + expect(harness.ctx.agents.get('sib-a')).toBe(handleA.agent) + expect(harness.ctx.agents.get('sib-b')).toBe(handleB.agent) + + await handleA.dispose() + // A is gone — unregistered AND its session removed from the store. + expect(harness.ctx.agents.get('sib-a')).toBeUndefined() + expect(harness.ctx.sessions.get('sib-a')).toBeUndefined() + expect(handleA.agent.status).toBe('disposed') + // B is wholly unaffected. + expect(harness.ctx.agents.get('sib-b')).toBe(handleB.agent) + expect(harness.ctx.sessions.get('sib-b')).toBeDefined() + expect(handleB.agent.status).not.toBe('disposed') + await harness.dispose() + }) }) diff --git a/packages/acp/tests/edges.spec.ts b/packages/acp/tests/edges.spec.ts index 9484368322..b9e2377908 100644 --- a/packages/acp/tests/edges.spec.ts +++ b/packages/acp/tests/edges.spec.ts @@ -25,7 +25,7 @@ describe('acp bridge — demux & config edges', () => { await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }) const before = harness.updates.length - const foreign = harness.ctx.agents.create({ agentId: 'foreign', sessionId: 'foreign-session', agentOptions: { model: 'mock' } }) + const { agent: foreign } = harness.ctx.agents.create({ agentId: 'foreign', sessionId: 'foreign-session', agentOptions: { model: 'mock' } }) foreign.send([{ type: 'text', text: 'hi' }]) await foreign.whenIdle() await new Promise(r => setTimeout(r, 10)) diff --git a/packages/agent-loop/README.md b/packages/agent-loop/README.md index e5e3a03d2d..5d79d90147 100644 --- a/packages/agent-loop/README.md +++ b/packages/agent-loop/README.md @@ -12,8 +12,10 @@ This is the only package in the harness that contains concrete loop logic. Every `AgentLoop` also implements the `AgentFactory` seam and registers itself via `ctx.agents.setFactory(this)`, so plugins create/resume agents through `ctx.agents` (the interface): -- `ctx.agents.create({ agentId, sessionId, meta?, agentOptions? })` — programmatic create on a caller-supplied `sessionId` (e.g. an ACP-generated id), NOT `${id}-session`. -- `ctx.agents.resume({ agentId, resumeSessionId, agentOptions? })` — load a persisted session via `ctx.sessionPersistence` ([session persistence](../../docs/rfc/implemented/2026-06-14-session-persistence.md)) and resume an agent on it. The live session id is the resumed id; turn numbering and derived history continue from the loaded log. Requires a session-persistence backend (NOT hard-injected — non-persistent demos still work; `resume` rejects with a clear error when persistence is absent). +- `ctx.agents.create({ agentId, sessionId, meta?, agentOptions? }): AgentHandle` — programmatic create on a caller-supplied `sessionId` (e.g. an ACP-generated id), NOT `${id}-session`. Returns an [`AgentHandle`](../agent/README.md) — the owner disposes it to tear down exactly this agent (stop loop + await quiescence + unregister + remove session). +- `ctx.agents.resume({ agentId, resumeSessionId, agentOptions? }): Promise` — load a persisted session via `ctx.sessionPersistence` ([session persistence](../../docs/rfc/implemented/2026-06-14-session-persistence.md)) and resume an agent on it. The live session id is the resumed id; turn numbering and derived history continue from the loaded log. Requires a session-persistence backend (NOT hard-injected — non-persistent demos still work; `resume` rejects with a clear error when persistence is absent). Returns an `AgentHandle`. + +The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loop fiber (it discards the handle) — only the programmatic factory callers (the ACP bridge) hold a handle and own per-agent teardown. ### Injected services diff --git a/packages/agent/README.md b/packages/agent/README.md index d815761ae4..846ab17444 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -17,8 +17,10 @@ Tracks live agents so UI, hook, and orchestrator plugins can find them without i Agent *creation* is provided by whichever plugin implements `AgentFactory` (phase 1: `dsh-agent-loop`), registered via `setFactory`. This keeps creation on the `dsh-agent` interface so consumers (UI, the ACP bridge) program against `ctx.agents` without depending on the concrete loop package. - `ctx.agents.setFactory(factory: AgentFactory): () => void` — register the creation factory (the loop calls this on construction). Throws on a second factory; the slot clears on dispose. -- `ctx.agents.create(options: CreateAgentOptions): Agent` — construct, start, AND register a new agent on a caller-supplied `sessionId` (with optional `meta.cwd`). Distinct from `register` (which only records). Throws if no factory is registered. -- `ctx.agents.resume(options: ResumeAgentOptions): Promise` — load a persisted session ([session persistence](../../docs/rfc/implemented/2026-06-14-session-persistence.md)) and resume an agent on it. Async; rejects if no factory is registered, or if the factory finds session persistence unconfigured. +- `ctx.agents.create(options: CreateAgentOptions): AgentHandle` — construct, start, AND register a new agent on a caller-supplied `sessionId` (with optional `meta.cwd`). Distinct from `register` (which only records). Throws if no factory is registered. +- `ctx.agents.resume(options: ResumeAgentOptions): Promise` — load a persisted session ([session persistence](../../docs/rfc/implemented/2026-06-14-session-persistence.md)) and resume an agent on it. Async; rejects if no factory is registered, or if the factory finds session persistence unconfigured. + +`AgentHandle = { agent: Agent; dispose(): Promise }`. The disposer is a **capability** — only the holder can tear this agent down. `dispose()` stops the loop, `await`s its exit (quiescence — NOT just the `disposed` status flip), unregisters the agent, and removes its session from the store, in an order that captures the loop's final `session/flush` before the session is detached. `ctx.agents.get(id)` still returns a bare `Agent` — the handle is only for the OWNER that created it. The ACP bridge is the production consumer (one handle per session, disposed on disconnect/teardown); config-created agents are owned by the loop fiber and never need a handle. ### Events From a53a56ff482aebb91778ed4f0f497605d5ee18d6 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 07:12:29 +0800 Subject: [PATCH 3/8] fix(agent-loop): fold session lifecycle into the agent effect for ordered teardown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stronger durability test (dispose MID-turn, then re-load from disk) caught that the original two-sibling-effect design dropped the loop's closing `turn/end` on the bare fiber-dispose path: a fiber unload disposes sibling effects CONCURRENTLY (`Promise.all`, vendor/cordis/fiber.ts), so the session-create effect detached `onAppend` racing the loop's final `session/flush` — the re-loaded log showed crash-recovery's synthetic `interrupted` closer instead of the real `disposed` reason. The disconnect path happened to work (only `quiesce()` ran), but the contract must hold uniformly. Fix: fold the session lifecycle INTO the agent's single composite effect. `SessionStore` now exposes `prepare` (validate + construct, no store entry), `enter` (attach onAppend + store, returns detach), and `announce` (emit session/created), replacing the sibling-effect `createOwned`. `AgentLoop.start` builds ONE effect that yields, in order: session-detach, register, then stop-and-`await agent.done`. LIFO disposal runs them as an ORDERED chain (the runtime awaits each disposer's promise before the next), so the loop is stopped and awaited to exit — its closing flush captured through the still- attached onAppend — BEFORE the session detaches, whether the trigger is the handle's dispose() OR a fiber unload. The config path uses prepare()+start too, so it gets the same ordered teardown. All three factory entrypoints now funnel through the one composite builder. The mid-turn durability test asserts the REAL `disposed` reason lands on disk (not a recovered `interrupted` substitute), proving the closing event was captured rather than reconstructed. --- packages/acp/tests/dispose.spec.ts | 37 ++++++++++ packages/agent-loop/src/index.ts | 110 +++++++++++++++-------------- packages/session/src/index.ts | 103 +++++++++++++++++---------- 3 files changed, 158 insertions(+), 92 deletions(-) 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 { From 7a94d36c46e4af9a894bf244bf753bb39fbbb176 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 07:12:52 +0800 Subject: [PATCH 4/8] =?UTF-8?q?docs(dsh-code-review):=20test=20sufficiency?= =?UTF-8?q?=20=E2=80=94=20real=20usage,=20not=20just=20100%=20coverage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expand the "Test quality" reviewer check: 100% coverage proves lines ran, not that the feature works the way it ships. Judge sufficiency on two axes — would the test fail on a regression, and does it exercise the REAL thing (genuine collaborator, real entry path, verify the world) rather than faking inputs just enough to cover every line. Call out the specific trap of a happy-path test that hits a line whose PURPOSE is a mid-flight/error/recovery scenario it never actually drives — the exact gap a clean-turn "durability" test would miss. --- .agents/skills/dsh-code-review/SKILL.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.agents/skills/dsh-code-review/SKILL.md b/.agents/skills/dsh-code-review/SKILL.md index 8d0792624c..ac59587e4d 100644 --- a/.agents/skills/dsh-code-review/SKILL.md +++ b/.agents/skills/dsh-code-review/SKILL.md @@ -42,7 +42,7 @@ Where your independent reasoning earns its keep. Start here, then keep going acr - **e2e verifies the world, not the agent's self-report.** For real-API tests, confirm the assertion re-runs the command/checks the file externally — a keyword probe lets a cheating agent pass (see AGENTS.md e2e bullet). For a behavior change to the agent's real flows, a no-key/mock test alone is usually insufficient: a with-key e2e (especially a smoke test that boots the real example and checks the world) is cheap here and catches "green units, broken product" — encourage it rather than treating real-API tests as expensive (see AGENTS.md § Secrets / .env). - **Plugin export shape + real-loader coverage.** A new/changed `cordis.yml`-loaded plugin: is it a function/namespace plugin (`name`/`inject`/`Config`/`apply` named exports) with NO `export default`? A stray default export makes the Loader's `unwrapExports` drop `inject` and the plugin crashes at load with `cannot get property … without inject` — invisible to hand-built `ctx.plugin({...})` tests and to line coverage. Confirm there's a test driving it through the REAL loader path (the no-key subprocess e2e for ACP is the model). And any opportunistic read of a service NOT in `static inject` should use `ctx.get(name)`, not `ctx.` (the property proxy throws through a foreign shadow). See [packages/AGENTS.md](../../../packages/AGENTS.md) and [docs/postmortem/0001](../../../docs/postmortem/0001-acp-default-export-drops-inject.md). - **Seam discipline.** New swappable capability? Check it's split per the capability-seams RFC (interface / impl / consumer), and that the consumer injects the interface key, never an implementation type. -- **Test quality.** A test that passes but asserts the wrong thing is worse than none. Check that new tests would actually fail if the behavior regressed, and that they exercise the contract (events fired, disposal reached) rather than restating the implementation. +- **Test quality — sufficiency, not just coverage.** 100% per-file coverage and a green suite are necessary, not sufficient: they prove the lines *ran*, not that the feature *works the way it ships*. Judge whether the tests are sufficient on two axes. (1) **Would they fail if the behavior regressed?** A test that passes but asserts the wrong thing — or restates the implementation instead of the contract (events fired, disposal reached, the world changed) — is worse than none. (2) **Do they exercise the REAL thing, the way it's actually used?** Prefer the genuine collaborator over a fake, drive the change through its real entry path (the cordis Loader, the ACP bridge, a booted subprocess — not a hand-built `ctx.plugin({...})` that bypasses `unwrapExports`), and verify the WORLD (re-read the file/log/registry externally), not the agent's self-report. A test that fakes the inputs just enough to cover every line will agree with whatever the author assumed; the real thing won't. When a test sets up a *clean/happy* path to reach a line, ask whether the line's PURPOSE is exercised — e.g. a durability/teardown path "tested" by a fully-completed turn never proves the mid-flight teardown it exists for; a torn-tail recovery branch covered by a well-formed log never proves recovery. Flag tests that hit the line but not the scenario. See AGENTS.md § Defensive patterns "Line coverage is not behavior coverage" and "Prefer the REAL implementation over a mock/stand-in in tests". - **Snapshot coverage for transcript/UX changes.** If the PR changes the editor-facing transcript or end-to-end agent UX — the ACP bridge's event→update translation, the agent loop's observable output, tool presentation, or anything an editor renders — it must add or update a snapshot scenario (`examples/*/tests/**/*.snapshot.ts`, goldens under `examples/acp-agent/tests/snapshots/`) or note explicitly why none applies (AGENTS.md § Conventions). Review the golden diff itself: a changed `stdout.golden.txt` / `session.golden.txt` is a behavior change in disguise — confirm it's intended, not an accidental regression someone re-recorded away. A pure internal refactor with no observable-output change is exempt, but the PR should say so. See [docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md](../../../docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md). - **Intent and contracts.** Does the change do what the PR says, and honor the documented contract on *both* sides of every seam it touches (see AGENTS.md "Honor cross-seam contracts on BOTH sides")? From 5a5b7d19c3580af94025709995e3f0d816c315ee Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 07:47:24 +0800 Subject: [PATCH 5/8] fix(agent): contain a throwing agent/disposed listener in the register disposer (Codex review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex found a real teardown-leak (A): the AgentHandle's composite effect runs its disposers as a `.then()` chain, and the register disposer emitted `agent/disposed` UNCONTAINED. A throwing listener rejected the chain, skipping the LATER session-detach disposer — stranding the session in the store with `onAppend` attached (a leak AND a durability hole, since the new composite design relies on detach running). Verified by tracing fiber.ts:299-301 (`task = task.then(dispose)`) against the yield order in AgentLoop.start. Wrap the disposer's `agent/disposed` emit in try/catch + logger.warn (the store entry is already removed before the emit — the useful state is captured — so logging and continuing is correct, mirroring the guarded `agent/status` emit in ReactLoopAgent). The sibling `agent/created` emit stays uncontained on purpose: its throw is MEANT to propagate and roll the registration back. Regression test (acp dispose.spec): register a throwing `agent/disposed` listener, drive a clean turn, dispose, assert the session was STILL removed. Confirmed it FAILS without the guard (the throw escapes dispose and detach is skipped) and passes with it. Also (B): document the new `prepare`/`enter`/`announce` ordered-teardown lifecycle primitives in the dsh-session README (they are public cross-package methods now consumed by dsh-agent-loop). --- packages/acp/tests/dispose.spec.ts | 25 +++++++++++++++++++++++++ packages/agent/src/index.ts | 17 ++++++++++++++++- packages/session/README.md | 10 ++++++++++ 3 files changed, 51 insertions(+), 1 deletion(-) 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 | From de0c4605bd59918667008b155bb6c443e5db46e6 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 11:55:56 +0800 Subject: [PATCH 6/8] =?UTF-8?q?docs(acp):=20correct=20teardown=20wording?= =?UTF-8?q?=20=E2=80=94=20dispose=20uses=20the=20disposed=20path,=20not=20?= =?UTF-8?q?cancel()=20(review)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A reviewer noted the quiesce() comment + ACP README said `AgentHandle.dispose()` stops the loop "with the queue-aware cancel", but the handle delegates to the start-disposer's `stop(); await agent.done`, where `stop()` sets `disposed` and aborts the current controller — it does NOT call `agent.cancel()`. The pre-step teardown window is still closed (the disposed promise wakes the parked loop and `isDisposed()` breaks before a turn starts), but the mechanism is the DISPOSED path and a mid-flight turn ends with reason `disposed`, not `aborted`. Corrected the comment and the README to describe the actual path. (This commit follows the merge of PR C's `cancel(reason)` fix up into this branch.) --- packages/acp/README.md | 2 +- packages/acp/src/index.ts | 26 ++++++++++++++------------ 2 files changed, 15 insertions(+), 13 deletions(-) diff --git a/packages/acp/README.md b/packages/acp/README.md index 5cdfa3a202..78c35eff86 100644 --- a/packages/acp/README.md +++ b/packages/acp/README.md @@ -61,7 +61,7 @@ A `session/prompt` resolves (or rejects) exactly once, keyed off the canonical s ## Disposal & disconnect -Teardown reaches quiescence: for EVERY live session settle any pending prompt as `cancelled`, then run that session's [`AgentHandle`](../agent/README.md) `dispose()` — which stops the loop with the queue-aware `cancel()`, `await`s the loop's exit (the final `turn/end` + `session/flush` are captured while the session is still attached), unregisters the agent, and removes its session from the store. The per-session disposes run in parallel. The same teardown runs on a **client disconnect** (`conn.closed` resolves when the editor quits / the transport EOFs), so a vanished client never leaves an orphaned running — or idled-but-still-registered — agent whose `session/update` writes are silently swallowed. The two paths are idempotent and memoized (the first clears the `sessions` map; a second caller awaits the same teardown promise). +Teardown reaches quiescence: for EVERY live session settle any pending prompt as `cancelled`, then run that session's [`AgentHandle`](../agent/README.md) `dispose()` — which stops the loop (sets `disposed` + aborts the in-flight step), `await`s the loop's exit (the final `turn/end` + `session/flush` are captured while the session is still attached), unregisters the agent, and removes its session from the store. A turn cut off mid-flight by teardown ends with reason `disposed` (not `aborted` — `dispose()` uses the disposed path, not `session/cancel`'s queue-aware `cancel()`). The per-session disposes run in parallel. The same teardown runs on a **client disconnect** (`conn.closed` resolves when the editor quits / the transport EOFs), so a vanished client never leaves an orphaned running — or idled-but-still-registered — agent whose `session/update` writes are silently swallowed. The two paths are idempotent and memoized (the first clears the `sessions` map; a second caller awaits the same teardown promise). ## Known limitations (tracked TODOs) diff --git a/packages/acp/src/index.ts b/packages/acp/src/index.ts index d6eda1cc04..d107c2c0cb 100644 --- a/packages/acp/src/index.ts +++ b/packages/acp/src/index.ts @@ -630,19 +630,21 @@ export function apply(ctx: Context, config: AcpConfig): void { * Tear ALL live sessions down to quiescence (AGENTS.md "dispose must reach * quiescence"): for each session settle any pending prompt `cancelled`, then * run that session's {@link AgentHandle} `dispose()` — which stops the loop - * with the queue-aware cancel, AWAITS the loop's exit (the final - * `turn/end` + `session/flush` are captured while `onAppend` is still + * (sets `disposed`, aborts the in-flight step), AWAITS the loop's exit (the + * final `turn/end` + `session/flush` are captured while `onAppend` is still * attached), unregisters the agent, and removes its session from the store. * The per-session disposes run in parallel. Idempotent — clears the `sessions` * map first and memoizes, so a second call (close racing dispose) is a no-op. * Shared by Cordis disposal AND client disconnect (`conn.closed`). * - * Per-agent disposal closes the former pre-step best-effort window: the - * queue-aware `cancel()` (RFC 011) drops a turn about to start, so a queued- - * but-not-yet-running prompt never runs after teardown. A bare client - * disconnect (resolves `conn.closed` WITHOUT disposing the fiber) thus leaves - * NO registered agent and NO session-store entry — not an idled-but-still- - * registered one. When the fiber IS disposed (whole-context or an ACP-only HMR + * Per-agent disposal closes the former pre-step best-effort window — but via + * the DISPOSED path, not `cancel()`: the start-disposer resolves `handle.disposed`, + * which wakes the parked loop, and `isDisposed()` breaks the loop before a + * queued-but-not-yet-running turn can start (a turn cut off mid-flight ends + * with reason `disposed`, not `aborted`). A bare client disconnect (resolves + * `conn.closed` WITHOUT disposing the fiber) thus leaves NO registered agent + * and NO session-store entry — not an idled-but-still-registered one. When the + * fiber IS disposed (whole-context or an ACP-only HMR * `acpFiber.dispose()`), this same memoized teardown runs first; the factory's * register+start+session effects are ALSO bound to the bridge fiber (the * factory is reached through this bridge's traceable service proxy, so @@ -667,10 +669,10 @@ export function apply(ctx: Context, config: AcpConfig): void { await Promise.all(recs.map(async (rec) => { settlePrompt(rec, 'cancelled') // Per-agent dispose (the AgentHandle disposer): unregister this agent, - // stop its loop with the queue-aware cancel, await quiescence (the loop - // exit + final flush), and remove its session — so a bare client - // disconnect leaves NO registered agent and NO session-store entry, not - // just an idled-but-still-registered one. + // stop its loop (sets disposed + aborts the in-flight step), await + // quiescence (the loop exit + final flush), and remove its session — so + // a bare client disconnect leaves NO registered agent and NO + // session-store entry, not just an idled-but-still-registered one. await rec.dispose() })) })() From 083a6fc9902c16f291374ebfb98b7007fd4402ed Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 13:06:28 +0800 Subject: [PATCH 7/8] fix(agent): re-check id in enter() + memoize AgentHandle.dispose() (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two blocking lifecycle findings from the deep review: - `SessionStore.enter()` is a public cross-package primitive that a caller can separate from `prepare()` by arbitrary work, so it must re-check the id: a stale prepared session could otherwise overwrite a live store entry of the same id, and the stale session's detach disposer would later delete the REAL session. Re-add the duplicate-id throw (removed earlier on a coverage rationale that only held for the back-to-back internal caller). Tests cover the stale-overwrite rejection and the prepare/enter/announce lifecycle (which also covers the throw branch). - `AgentHandle.dispose()` exposed the raw single-shot cordis effect disposer, so a concurrent/second dispose() returned immediately (effect epoch already cleared) instead of awaiting the in-flight teardown — violating the dispose(): Promise contract that every caller observes the same quiescence boundary. Memoize the disposal promise in startOwned. Regression test gates the loop's final flush, fires two dispose() calls, and asserts the second stays pending until the first's teardown completes (fails without the memo). --- packages/acp/tests/dispose.spec.ts | 43 ++++++++++++++++++++++++++ packages/agent-loop/src/index.ts | 14 +++++++-- packages/session/src/index.ts | 16 ++++++---- packages/session/tests/session.spec.ts | 34 ++++++++++++++++++++ 4 files changed, 99 insertions(+), 8 deletions(-) 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) From 44762efbd7e387201c52fef78260225b77907f7e Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 20 Jun 2026 13:51:42 +0800 Subject: [PATCH 8/8] docs(acp): drop PR-letter ref from dispose test comment (review) The disconnect-mid-prompt test comment said "PR D's per-agent AgentHandle teardown", narrating the change's origin. Per the repo doc-current-state convention, state the mechanism (the session's AgentHandle teardown) without naming the PR that introduced it. --- packages/acp/tests/dispose.spec.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/acp/tests/dispose.spec.ts b/packages/acp/tests/dispose.spec.ts index 3196994a0e..dd27cc7e34 100644 --- a/packages/acp/tests/dispose.spec.ts +++ b/packages/acp/tests/dispose.spec.ts @@ -85,7 +85,7 @@ describe('acp bridge — disposal & HMR safety', () => { it('a client disconnect mid-prompt disposes the session (no registered agent left)', async () => { // The ACP transport closes (editor quits) while a turn runs. The bridge must - // settle the in-flight prompt cancelled and DISPOSE the agent (PR D's + // settle the in-flight prompt cancelled and DISPOSE the agent (the session's // per-agent AgentHandle teardown) rather than leaving an orphaned running — // or even idled-but-still-registered — agent whose updates are swallowed. const harness = await makeBridgeHarness({ storageDir, script: ['hang'] })