feat(acp): multiplex N concurrent ACP sessions + bash task ownership (RFC 011)
Lifts the RFC 010 single-session-per-connection cap: the bridge now runs N concurrent sessions over one connection, each mapped to its own LoopAgent. - packages/acp: live sessions held in a Map<sessionId, SessionRecord> with an agent→sessionId reverse WeakMap so agent/* events (which carry only the Agent) demux in O(1). Every session/event and agent/status is routed strictly to its owning record — concurrent sessions never cross-settle or interleave their session/update notifications. Per-session state: one in-flight prompt each, session/cancel aborts+settles only its own agent/prompt, session/load reserves a per-id load slot (distinct ids load concurrently; re-loading a live id is rejected), and disposal drains every live session in parallel to quiescence. - packages/tool-bash: record each background task's owning agent at spawn and keep it for the executor's lifetime (NOT cleared on completion). bash_output/bash_kill reject a task owned by a different agent (a task with no owner is open; a no-agent caller can't access an owned task). Task ids are global and predictable, so this is the fence that stops one session's agent from reading/killing another session's background task. - Per-session permission ownership and a per-agent disposer seam stay deferred (depend on the deferred permission gate); the reverse map the gate will route through is in place. RFC 011 stays `proposed`. - Tests: two sessions stream concurrently without interleave; cross-session cancel isolation; per-session in-flight enforcement; dispose-all-to-quiescence; bash cross-session read/kill rejected (+ no-agent and unowned-task cases). - Docs: RFC 011 implementation-status note; acp + tool-bash READMEs; example MVP-limitations updated. 100% per-file coverage maintained.
This commit is contained in:
Tianyi Cui
@@ -52,12 +52,17 @@ describe('acp bridge', () => {
|
||||
expect(text).toBe('hello there')
|
||||
})
|
||||
|
||||
it('rejects a second session/new (single-session MVP)', async () => {
|
||||
it('allows multiple concurrent sessions, each with a distinct id', async () => {
|
||||
harness = await makeBridgeHarness({ storageDir, script: [] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
await expect(harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }))
|
||||
.rejects.toThrow(/single session/)
|
||||
const a = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
const b = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
expect(a.sessionId).toBeTruthy()
|
||||
expect(b.sessionId).toBeTruthy()
|
||||
expect(a.sessionId).not.toBe(b.sessionId)
|
||||
// Both agents are live and independently registered.
|
||||
expect(harness.ctx.agents.get(a.sessionId)).toBeDefined()
|
||||
expect(harness.ctx.agents.get(b.sessionId)).toBeDefined()
|
||||
})
|
||||
|
||||
it('rejects a non-absolute cwd and a cwd that differs from the launch dir', async () => {
|
||||
|
||||
@@ -107,8 +107,10 @@ export interface BridgeHarness {
|
||||
ctx: Context
|
||||
client: ClientSideConnection
|
||||
adapter: MockAdapter
|
||||
/** Every `session/update` the bridge pushed, in order. */
|
||||
/** Every `session/update` the bridge pushed, in order (payload only). */
|
||||
updates: CapturedUpdate[]
|
||||
/** Same, but tagged with each update's `sessionId` (for multi-session demux assertions). */
|
||||
sessionUpdates: { sessionId: string; update: CapturedUpdate }[]
|
||||
/** Permission requests the bridge issued (none until the gate lands). */
|
||||
permissionRequests: RequestPermissionRequest[]
|
||||
/** Decide each permission request's outcome (default: cancelled). */
|
||||
@@ -181,11 +183,13 @@ export async function makeBridgeHarness(options: {
|
||||
const clientStream: Stream = ndJsonStream(clientOutput, a2c.readable)
|
||||
|
||||
const updates: CapturedUpdate[] = []
|
||||
const sessionUpdates: { sessionId: string; update: CapturedUpdate }[] = []
|
||||
const permissionRequests: RequestPermissionRequest[] = []
|
||||
const harness: BridgeHarness = {
|
||||
ctx,
|
||||
adapter,
|
||||
updates,
|
||||
sessionUpdates,
|
||||
permissionRequests,
|
||||
onPermission: () => ({ outcome: { outcome: 'cancelled' } }),
|
||||
onSessionUpdateError: undefined,
|
||||
@@ -204,6 +208,7 @@ export async function makeBridgeHarness(options: {
|
||||
const makeClient = (_agent: AcpAgent): Client => ({
|
||||
sessionUpdate(params: SessionNotification): Promise<void> {
|
||||
updates.push(params.update)
|
||||
sessionUpdates.push({ sessionId: params.sessionId, update: params.update })
|
||||
// Let a test force the bridge's notify() error path.
|
||||
if (harness.onSessionUpdateError) return Promise.reject(new Error('client update rejected'))
|
||||
return Promise.resolve()
|
||||
|
||||
@@ -120,11 +120,18 @@ describe('acp bridge — session/load replay', () => {
|
||||
.rejects.toThrow(/launch directory/)
|
||||
})
|
||||
|
||||
it('rejects load when a session already exists (single-session MVP)', async () => {
|
||||
live = await makeBridgeHarness({ storageDir, script: [] })
|
||||
it('allows loading alongside an existing session but rejects re-loading the SAME id', async () => {
|
||||
// Multi-session: a load can coexist with a live session, but loading an id
|
||||
// that is already live is rejected (it is already loaded).
|
||||
live = await makeBridgeHarness({ storageDir, script: [textResponse('one')] })
|
||||
await live.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
await live.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
await expect(live.client.loadSession({ sessionId: 'other', cwd: process.cwd(), mcpServers: [] }))
|
||||
.rejects.toThrow(/single session/)
|
||||
const { sessionId } = await live.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
await live.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'hi' }] })
|
||||
// A different new session coexists.
|
||||
const other = await live.client.newSession({ cwd: process.cwd(), mcpServers: [] })
|
||||
expect(other.sessionId).not.toBe(sessionId)
|
||||
// Re-loading the already-live id is rejected.
|
||||
await expect(live.client.loadSession({ sessionId, cwd: process.cwd(), mcpServers: [] }))
|
||||
.rejects.toThrow(/already loaded/)
|
||||
})
|
||||
})
|
||||
|
||||
128
packages/acp/tests/multi-session.spec.ts
Normal file
128
packages/acp/tests/multi-session.spec.ts
Normal file
@@ -0,0 +1,128 @@
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
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, textResponse, type BridgeHarness, type CapturedUpdate } from './harness.ts'
|
||||
|
||||
/** Text of the agent_message_chunk updates scoped to one session id. */
|
||||
function messageTextFor(updates: { sessionId?: string; update: CapturedUpdate }[], sessionId: string): string {
|
||||
return updates
|
||||
.filter(u => u.sessionId === sessionId && u.update.sessionUpdate === 'agent_message_chunk')
|
||||
.map(u => (u.update.sessionUpdate === 'agent_message_chunk' && u.update.content.type === 'text' ? u.update.content.text : ''))
|
||||
.join('')
|
||||
}
|
||||
|
||||
describe('acp bridge — RFC 011 multi-session isolation', () => {
|
||||
let storageDir: string
|
||||
let harness: BridgeHarness | undefined
|
||||
|
||||
beforeEach(async () => { storageDir = await mkdtemp(join(tmpdir(), 'acp-multi-')) })
|
||||
afterEach(async () => {
|
||||
if (harness) await harness.dispose()
|
||||
harness = undefined
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
it('two sessions stream concurrently without interleaving their updates', async () => {
|
||||
// Each session's prompt answer must arrive only on its own sessionId. The
|
||||
// scripted adapter answers in send order; both prompts run, and the bridge
|
||||
// demuxes every chunk by session id.
|
||||
harness = await makeBridgeHarness({ storageDir, script: [textResponse('answer-A'), textResponse('answer-B')] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const a = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
const b = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
|
||||
const [ra, rb] = await Promise.all([
|
||||
harness.client.prompt({ sessionId: a, prompt: [{ type: 'text', text: 'go A' }] }),
|
||||
harness.client.prompt({ sessionId: b, prompt: [{ type: 'text', text: 'go B' }] }),
|
||||
])
|
||||
expect(ra.stopReason).toBe('end_turn')
|
||||
expect(rb.stopReason).toBe('end_turn')
|
||||
|
||||
// A's text landed only on A; B's only on B (strict id demux, no interleave).
|
||||
expect(messageTextFor(harness.sessionUpdates, a)).toContain('answer-A')
|
||||
expect(messageTextFor(harness.sessionUpdates, a)).not.toContain('answer-B')
|
||||
expect(messageTextFor(harness.sessionUpdates, b)).toContain('answer-B')
|
||||
expect(messageTextFor(harness.sessionUpdates, b)).not.toContain('answer-A')
|
||||
})
|
||||
|
||||
it('cancel in one session leaves the other session untouched', async () => {
|
||||
// Session A hangs; session B completes normally. Cancelling A settles ONLY
|
||||
// A as cancelled and never disturbs B's stream or result.
|
||||
harness = await makeBridgeHarness({ storageDir, script: ['hang', textResponse('B done')] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const a = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
const b = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
|
||||
const aPromise = harness.client.prompt({ sessionId: a, prompt: [{ type: 'text', text: 'hang A' }] })
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
await harness.client.cancel({ sessionId: a })
|
||||
expect((await aPromise).stopReason).toBe('cancelled')
|
||||
|
||||
// B runs to completion, unaffected by A's cancel.
|
||||
const rb = await harness.client.prompt({ sessionId: b, prompt: [{ type: 'text', text: 'go B' }] })
|
||||
expect(rb.stopReason).toBe('end_turn')
|
||||
expect(messageTextFor(harness.sessionUpdates, b)).toContain('B done')
|
||||
})
|
||||
|
||||
it('enforces one in-flight prompt PER session independently', async () => {
|
||||
harness = await makeBridgeHarness({ storageDir, script: ['hang', 'hang'] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const a = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
const b = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
|
||||
// One in-flight prompt in EACH session is allowed (independent limits).
|
||||
const aPromise = harness.client.prompt({ sessionId: a, prompt: [{ type: 'text', text: 'one A' }] })
|
||||
const bPromise = harness.client.prompt({ sessionId: b, prompt: [{ type: 'text', text: 'one B' }] })
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
// A second prompt in A is rejected, but B's in-flight prompt is unaffected.
|
||||
await expect(harness.client.prompt({ sessionId: a, prompt: [{ type: 'text', text: 'two A' }] }))
|
||||
.rejects.toThrow(/already in flight/)
|
||||
|
||||
await harness.client.cancel({ sessionId: a })
|
||||
await harness.client.cancel({ sessionId: b })
|
||||
expect((await aPromise).stopReason).toBe('cancelled')
|
||||
expect((await bPromise).stopReason).toBe('cancelled')
|
||||
})
|
||||
|
||||
it('a cancel for a non-existent session id is a silent no-op (does not touch others)', async () => {
|
||||
harness = await makeBridgeHarness({ storageDir, script: [textResponse('A done')] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const a = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
await expect(harness.client.cancel({ sessionId: 'ghost' })).resolves.toBeUndefined()
|
||||
// A still works after a cancel for an unknown id.
|
||||
const ra = await harness.client.prompt({ sessionId: a, prompt: [{ type: 'text', text: 'go A' }] })
|
||||
expect(ra.stopReason).toBe('end_turn')
|
||||
})
|
||||
|
||||
it('disposing the whole bridge drains all live sessions to quiescence', async () => {
|
||||
harness = await makeBridgeHarness({ storageDir, script: ['hang', 'hang'] })
|
||||
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
|
||||
const a = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
const b = (await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })).sessionId
|
||||
const agentA = harness.ctx.agents.get(a)!
|
||||
const agentB = harness.ctx.agents.get(b)!
|
||||
|
||||
// Wait deterministically for BOTH agents to enter `running` (not a fixed
|
||||
// sleep — agent startup latency is unbounded on a loaded worker).
|
||||
const running = (agent: typeof agentA) => agent.status === 'running'
|
||||
? Promise.resolve()
|
||||
: new Promise<void>((resolve) => {
|
||||
const dispose = harness!.ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'running') { dispose(); resolve() }
|
||||
})
|
||||
})
|
||||
void harness.client.prompt({ sessionId: a, prompt: [{ type: 'text', text: 'go A' }] }).catch(() => {})
|
||||
void harness.client.prompt({ sessionId: b, prompt: [{ type: 'text', text: 'go B' }] }).catch(() => {})
|
||||
await Promise.all([running(agentA), running(agentB)])
|
||||
expect(agentA.status).toBe('running')
|
||||
expect(agentB.status).toBe('running')
|
||||
|
||||
await harness.ctx.fiber.dispose()
|
||||
// BOTH agents drained (not still running) — teardown reached quiescence
|
||||
// across all sessions, not just one.
|
||||
expect(agentA.status).not.toBe('running')
|
||||
expect(agentB.status).not.toBe('running')
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user