Merge branch 'feat/acp-2-bridge' into feat/acp-3-multi-session
# Conflicts: # docs/rfc/proposed/2026-06-14-acp-multi-session.md # packages/acp/src/index.ts
This commit is contained in:
Tianyi Cui
@@ -2,7 +2,7 @@
|
||||
|
||||
The **Agent Client Protocol (ACP)** bridge: exposes the DeepSeek Harness coding agent as an ACP server over JSON-RPC stdio, so editors (Zed and other ACP clients) can drive it — streaming render, tool-call display, and resumable sessions. **N concurrent sessions per connection** (RFC 011): each maps to its own `LoopAgent`, and every event is demuxed strictly by session id so two sessions streaming at once never interleave.
|
||||
|
||||
It is a **client-driver / UI plugin**, the structured analogue of the readline `stdio-chat` plugin — NOT a loop change and NOT an [ADR 0009](../../docs/adr/0009-capability-seams.md) capability seam. It consumes the existing `agent/*` event taxonomy, the `dsh-agent` create/resume factory, and `dsh-session-persistence`.
|
||||
It is a **client-driver / UI plugin**, the structured analogue of the readline `stdio-chat` plugin — NOT a loop change and NOT a [capability seam](../../docs/rfc/implemented/2026-06-13-capability-seams.md). It consumes the existing `agent/*` event taxonomy, the `dsh-agent` create/resume factory, and `dsh-session-persistence`.
|
||||
|
||||
## Service / plugin
|
||||
|
||||
|
||||
+29
-14
@@ -172,6 +172,17 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
const agentName = config.agentName ?? 'deepseek-harness-acp'
|
||||
const agentVersion = config.agentVersion ?? '0.0.1'
|
||||
|
||||
// Capture the injected services NOW, during apply(), while we are inside this
|
||||
// plugin's fiber (where `inject` grants access). The ACP method handlers run
|
||||
// LATER, from the AgentSideConnection's JSON-RPC read loop — a context that is
|
||||
// NOT this fiber's injection scope — so reading `ctx.agents` / `ctx.logger` /
|
||||
// `ctx.sessionPersistence` lazily inside a handler throws "cannot get property
|
||||
// … without inject". Resolving the references here and closing over them keeps
|
||||
// the handlers working regardless of which fiber later invokes them.
|
||||
const agents = ctx.agents
|
||||
const sessionPersistence = ctx.sessionPersistence
|
||||
const logger = ctx.logger
|
||||
|
||||
// Live sessions keyed by id (RFC 011 multi-session), plus an agent→sessionId
|
||||
// reverse map so `agent/*` events (which carry only the Agent) demux in O(1).
|
||||
// The two stay in lockstep: a record is added to `sessions` and the agent to
|
||||
@@ -225,7 +236,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
failure (closed pipe), which the in-memory test transport never induces;
|
||||
the swallow is a defensive best-effort guard like the loop's emit traps */
|
||||
void Promise.resolve(conn.sessionUpdate(notification)).catch((error: unknown) => {
|
||||
ctx.logger.warn(`acp: session/update failed: ${String(error)}`)
|
||||
logger.warn(`acp: session/update failed: ${String(error)}`)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -377,7 +388,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
assertOpen()
|
||||
validateWorkspaceParams(params)
|
||||
const sessionId = randomUUID()
|
||||
const agent = ctx.agents.create({
|
||||
const agent = agents.create({
|
||||
agentId: sessionId,
|
||||
sessionId,
|
||||
meta: { cwd: params.cwd },
|
||||
@@ -409,13 +420,13 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
// launched in workspace B: it would replay A's history while tools run
|
||||
// in B. (If the id is unknown to `list()`, fall through to resume,
|
||||
// which rejects with the backend's not-found error.)
|
||||
const meta = (await ctx.sessionPersistence.list()).find(m => m.id === params.sessionId)
|
||||
const meta = (await sessionPersistence.list()).find(m => m.id === params.sessionId)
|
||||
if (meta?.cwd !== undefined && meta.cwd !== process.cwd()) {
|
||||
throw invalidParams(
|
||||
`session was created in ${meta.cwd}, but the server's launch directory is ${process.cwd()}; honoring a different cwd is not yet supported — launch the server in the session's workspace`,
|
||||
)
|
||||
}
|
||||
const agent = await ctx.agents.resume({
|
||||
const agent = await agents.resume({
|
||||
agentId: params.sessionId,
|
||||
resumeSessionId: params.sessionId,
|
||||
agentOptions: agentOptions(config),
|
||||
@@ -536,13 +547,19 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
* loop-level change); the single-in-flight-per-session rule bounds the worst
|
||||
* case to one short queued turn per session.
|
||||
*
|
||||
* The agents themselves are NOT individually disposed/unregistered here — the
|
||||
* factory (`ctx.agents.create`/`resume`) registers each on the AgentLoop fiber
|
||||
* and returns no per-agent disposer, so registry entries are reclaimed when
|
||||
* the host context disposes. On a bare client disconnect (without a host
|
||||
* dispose) the idled agents linger in `ctx.agents` until shutdown; a reconnect
|
||||
* spins up a fresh context, so this does not strand work. A per-agent disposal
|
||||
* seam is a follow-up (TODO(rfc010-agent-disposal)).
|
||||
* 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)).
|
||||
*/
|
||||
let quiescing: Promise<void> | undefined
|
||||
const quiesce = (): Promise<void> => {
|
||||
@@ -580,7 +597,7 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
||||
mid-run), and there is nothing else to act on once the connection is gone —
|
||||
the swallow mirrors notify(). */
|
||||
void conn.closed.then(quiesce).catch((error: unknown) => {
|
||||
ctx.logger.warn(`acp: connection-close teardown failed: ${String(error)}`)
|
||||
logger.warn(`acp: connection-close teardown failed: ${String(error)}`)
|
||||
})
|
||||
/* v8 ignore stop */
|
||||
|
||||
@@ -736,5 +753,3 @@ function toolResultContent(blocks: ContentBlock[]): { type: 'content'; content:
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
export default apply
|
||||
|
||||
@@ -38,7 +38,7 @@ describe('acp bridge — disposal & HMR safety', () => {
|
||||
// stay up and the transport is still live. A late session/new must hit the
|
||||
// `closed` guard and reject — NOT create an agent the disposed bridge can no
|
||||
// longer stream or settle. Verify the world: no agent appeared.
|
||||
const harness = await makeBridgeHarness({ storageDir, script: [], childFiber: true })
|
||||
const harness = await makeBridgeHarness({ storageDir, script: [] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const before = harness.ctx.agents.list().length
|
||||
await harness.acpFiber.dispose() // tear down ONLY the bridge
|
||||
@@ -48,6 +48,25 @@ describe('acp bridge — disposal & HMR safety', () => {
|
||||
await harness.dispose()
|
||||
})
|
||||
|
||||
it('an agent created through the bridge is unregistered when ONLY the bridge fiber is disposed', async () => {
|
||||
// The factory (`ctx.agents.create`) is reached through the bridge's
|
||||
// traceable service proxy, so `AgentLoop.start`'s `this.ctx.effect(...)`
|
||||
// registration binds to the CALLER context — the bridge fiber — not the
|
||||
// AgentLoop fiber. Disposing JUST the bridge fiber (an ACP-only HMR reload)
|
||||
// must therefore reclaim the agent's registry entry, even though agents/
|
||||
// agent-loop stay up. This pins the fiber-ownership the bridge's teardown
|
||||
// doc comment relies on; if a refactor rebinds the registration to the
|
||||
// AgentLoop fiber, the agent would survive bridge dispose and this fails.
|
||||
const harness = await makeBridgeHarness({ storageDir, script: [] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
expect(harness.ctx.agents.get(sessionId)).toBeDefined()
|
||||
|
||||
await harness.acpFiber.dispose() // tear down ONLY the bridge
|
||||
expect(harness.ctx.agents.get(sessionId)).toBeUndefined()
|
||||
await harness.dispose()
|
||||
})
|
||||
|
||||
it('no agent is created by a session/new after the bridge has closed (closed guard)', async () => {
|
||||
// After teardown (here a client disconnect sets `closed`), a late
|
||||
// `session/new` must NOT create an orphan agent the bridge can no longer
|
||||
|
||||
@@ -148,8 +148,6 @@ export async function makeBridgeHarness(options: {
|
||||
script?: (StreamChunk[] | 'hang')[]
|
||||
config?: Partial<AcpConfig>
|
||||
storageDir: string
|
||||
/** Mount the bridge in a disposable child fiber (for the ACP-only-HMR test). */
|
||||
childFiber?: boolean
|
||||
} = { storageDir: '' }): Promise<BridgeHarness> {
|
||||
const adapter = new MockAdapter(options.script ?? [])
|
||||
|
||||
@@ -225,21 +223,20 @@ export async function makeBridgeHarness(options: {
|
||||
// override means "no model at all".
|
||||
const cfg: AcpConfig = { stream: agentStream, ...options.config }
|
||||
if (!(options.config && 'model' in options.config)) cfg.model = 'mock'
|
||||
// By default apply the bridge directly on the root ctx (services ungated). For
|
||||
// the ACP-only-HMR test, `childFiber: true` mounts it in a CHILD fiber instead
|
||||
// so the test can dispose JUST the bridge while the rest of the harness stays
|
||||
// up — its disposer (`harness.acpFiber.dispose()`) tears down only the
|
||||
// bridge's listeners/effect. (Child-fiber service tracing gates the async
|
||||
// persistence path, so the load-replay tests use the default direct mount.)
|
||||
if (options.childFiber) {
|
||||
harness.acpFiber = await ctx.plugin({
|
||||
name: 'acp-test',
|
||||
inject: ['agents', 'sessions', 'sessionPersistence'],
|
||||
apply: (inner: Context) => { AcpPlugin.apply(inner, cfg) },
|
||||
})
|
||||
} else {
|
||||
AcpPlugin.apply(ctx, cfg)
|
||||
}
|
||||
// Mount the bridge the way production does: as a cordis PLUGIN (via
|
||||
// `ctx.plugin` with the real `inject`), NOT `AcpPlugin.apply(ctx, cfg)`
|
||||
// directly on the root ctx. The plugin fiber is the faithful reproduction —
|
||||
// the bridge's `apply` runs inside the fiber's injection scope, and its ACP
|
||||
// handlers later run from the JSON-RPC read loop OUTSIDE that scope, exactly
|
||||
// as under the example's cordis.yml. (Mounting directly on root made every
|
||||
// service an ungated property and hid the "cannot get property … without
|
||||
// inject" failure that bit a real Zed session.) `harness.acpFiber.dispose()`
|
||||
// tears down JUST the bridge (its listeners + effect) for the HMR test.
|
||||
harness.acpFiber = await ctx.plugin({
|
||||
name: 'acp-test',
|
||||
inject: ['agents', 'sessions', 'sessionPersistence'],
|
||||
apply: (inner: Context) => { AcpPlugin.apply(inner, cfg) },
|
||||
})
|
||||
harness.client = new ClientSideConnection(makeClient, clientStream)
|
||||
|
||||
return harness
|
||||
|
||||
Reference in New Issue
Block a user