Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(acp): dispose each session's agent on disconnect/teardown

Browse Source 
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.
This commit is contained in:
Tianyi Cui
2026-06-20 06:44:58 +08:00
parent 2a4d89a4bd
commit ee4cad3ada
9 changed files with 150 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/acp/README.md
View File

@@ -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

94
packages/acp/src/index.ts
View File

@@ -138,6 +138,13 @@ export const Config: Schema<AcpConfig> = 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<void>
/**
* 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<void> | undefined
const quiesce = (): Promise<void> => {
@@ -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

87
packages/acp/tests/dispose.spec.ts
View File

@@ -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()
})
})

2
packages/acp/tests/edges.spec.ts
View File

@@ -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))

6
packages/agent-loop/README.md
View File

@@ -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<AgentHandle>` — 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

6
packages/agent/README.md
View File

@@ -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<Agent>` — 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<AgentHandle>` — 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<void> }`. 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