Files
deepseek-harness/packages/agent/src/index.ts
T
Tianyi Cui 5a5b7d19c3 fix(agent): contain a throwing agent/disposed listener in the register disposer (Codex review)
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).
2026-06-20 07:47:24 +08:00

201 lines
8.0 KiB
TypeScript

/**
* Agent registry service. Tracks live agents so plugins can find them without
* depending on the concrete loop package. Agent creation belongs to the loop.
*
* @module @deepseek-ai/dsh-agent
*/
import { Context, Service } from 'cordis'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type { Agent, AgentOptions } from './types.ts'
export * from './types.ts'
declare module 'cordis' {
interface Context {
agents: AgentRegistry
}
}
/**
* Options for programmatically creating an agent through the registry factory
* ({@link AgentRegistry.create}). The caller supplies the live `sessionId`
* (e.g. an ACP-generated id) and optional session metadata (the validated
* `cwd`, fork lineage); the factory creates the session, the agent, and wires
* them together.
*/
export interface CreateAgentOptions {
/** The agent's id (the registry handle). */
agentId: string
/** The live session's id (NOT derived from agentId). */
sessionId: string
/**
* Session creation metadata: validated absolute `cwd` and `parentSession`
* fork lineage. Mirrors the `cwd`/`parentSession` fields of
* {@link CreateSessionOptions.meta} in dsh-session (the internal-only
* `createdAt`, used when reconstructing a persisted session, is deliberately
* excluded — a factory caller never sets it).
*/
meta?: { cwd?: string; parentSession?: SessionId }
/** Per-agent options (model, system prompt). */
agentOptions?: AgentOptions
}
/**
* Options for resuming an agent on a persisted session
* ({@link AgentRegistry.resume}).
*/
export interface ResumeAgentOptions {
/** The agent's id (the registry handle). */
agentId: string
/** The persisted session id to load and resume on. */
resumeSessionId: string
/** Per-agent options (model, system prompt). */
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<void>
}
/**
* The agent-creation factory the loop implementation provides to the registry
* via {@link AgentRegistry.setFactory}. Kept on the `dsh-agent` interface so
* consumers (e.g. the ACP bridge) program against `ctx.agents` without
* depending on the concrete `dsh-agent-loop` package.
*/
export interface AgentFactory {
/**
* 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`). Returns an {@link AgentHandle}.
*/
resume(options: ResumeAgentOptions): Promise<AgentHandle>
}
/** Thrown when create/resume is called before an agent factory is registered. */
const NO_FACTORY_MESSAGE = 'no agent factory registered (load an agent-loop plugin)'
/**
* Agent registry (`ctx.agents`): tracks live agents so UI, hook, and
* orchestrator plugins can find them without depending on the concrete loop
* package. Agent *creation* is provided by whichever plugin implements the
* {@link AgentFactory} (phase 1: `@deepseek-ai/dsh-agent-loop`), registered via
* {@link setFactory}.
*/
export class AgentRegistry extends Service {
private store = new Map<string, Agent>()
private factory: AgentFactory | undefined
constructor(ctx: Context) {
super(ctx, 'agents')
}
/**
* Register the agent-creation factory (the loop calls this on construction,
* effect-scoped). Throws if a factory is already registered. Returns the
* disposer; on dispose the factory slot is cleared.
*/
setFactory(factory: AgentFactory): () => void {
const dispose = this.ctx.effect(() => {
if (this.factory !== undefined) throw new Error('an agent factory is already registered')
this.factory = factory
return () => { this.factory = undefined }
}, 'agents.setFactory()')
return () => void dispose()
}
/**
* 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. Returns an {@link AgentHandle} — the owner disposes it to tear
* down exactly this agent.
*/
create(options: CreateAgentOptions): AgentHandle {
if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE)
return this.factory.createAgent(options)
}
/**
* 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. Returns an {@link AgentHandle}.
*/
async resume(options: ResumeAgentOptions): Promise<AgentHandle> {
if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE)
return this.factory.resume(options)
}
/**
* Register a live agent. Throws if an agent with the same id is already
* registered. Emits `agent/created` on registration and `agent/disposed`
* when the calling fiber is disposed. Returns the disposer.
*/
register(agent: Agent): () => void {
const dispose = this.ctx.effect(function* (this: AgentRegistry) {
if (this.store.has(agent.id)) {
throw new Error(`agent "${agent.id}" is already registered`)
}
this.store.set(agent.id, agent)
// Yield the rollback BEFORE emitting `agent/created`: a generator effect
// collects each yielded disposer before the next step runs, so a
// throwing `agent/created` listener rolls the entry back instead of
// leaking it (a leak would wedge the duplicate-id check until restart).
// The duplicate throw above fires before any mutation — it leaks nothing.
yield () => {
this.store.delete(agent.id)
// 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()')
// ctx.effect's disposer returns Promise<void>; our disposer API is
// synchronous fire-and-forget — discard the (always-resolved) promise.
return () => void dispose()
}
get(id: string): Agent | undefined {
return this.store.get(id)
}
list(): Agent[] {
return [...this.store.values()]
}
}
export default AgentRegistry