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 {