From b3ea13749c211d052c45efa5ab43168d63c3c834 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 19:23:21 +0800 Subject: [PATCH] feat(acp): multiplex N concurrent ACP sessions + bash task ownership (RFC 011) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 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. --- docs/rfc/011-acp-multi-session.md | 2 + examples/acp-agent/README.md | 2 +- packages/acp/README.md | 24 ++-- packages/acp/src/index.ts | 159 ++++++++++++----------- packages/acp/tests/bridge.spec.ts | 13 +- packages/acp/tests/harness.ts | 7 +- packages/acp/tests/load.spec.ts | 17 ++- packages/acp/tests/multi-session.spec.ts | 128 ++++++++++++++++++ packages/tool-bash/README.md | 4 + packages/tool-bash/src/index.ts | 57 ++++++-- packages/tool-bash/tests/tools.spec.ts | 100 ++++++++++++++ 11 files changed, 407 insertions(+), 106 deletions(-) create mode 100644 packages/acp/tests/multi-session.spec.ts diff --git a/docs/rfc/011-acp-multi-session.md b/docs/rfc/011-acp-multi-session.md index dca24569f8..77bbc7b664 100644 --- a/docs/rfc/011-acp-multi-session.md +++ b/docs/rfc/011-acp-multi-session.md @@ -2,6 +2,8 @@ Status: proposed +> **Implementation status:** the multi-session bridge (steps 1, 3, 4) and the bash task-ownership isolation are implemented in `packages/acp` + `packages/tool-bash`. **Per-session *permission* ownership is deferred** — it depends on the RFC 010 permission gate (`TODO(rfc010-permission-gate)`), which is itself deferred; the `agent→sessionId` reverse map the gate will route through is in place. Step 2's "real per-session disposer scope" is also deferred (`TODO(rfc010-agent-disposal)`): the bridge demuxes via id-keyed maps and global `ctx.on` listeners (correct and leak-free — disposal drains every session in parallel to quiescence), and a per-agent disposer seam is the follow-up. Status stays `proposed` until per-session permission ownership lands. + ## Problem RFC 010 ships ACP support with a single active session per connection: a second `session/new` is rejected. Editors expect to run several conversations over one agent subprocess — a user opens multiple threads, or a client pre-warms sessions. The single-session guard is a deliberate MVP scope cut, not an architectural limit; this RFC lifts it. diff --git a/examples/acp-agent/README.md b/examples/acp-agent/README.md index e010dab96b..99650c2779 100644 --- a/examples/acp-agent/README.md +++ b/examples/acp-agent/README.md @@ -32,4 +32,4 @@ Run from the repo root (the MVP requires the server's launch directory to be the ## MVP limitations -The bridge is the RFC 010 MVP: single session per connection (RFC 011 lifts this), text-only prompts, `cwd` must equal the launch directory, and the tool-permission gate is deferred (`TODO(rfc010-permission-gate)` — tools run with the executor's full authority). See `packages/acp/README.md` for the full contract. +The bridge supports N concurrent sessions per connection (RFC 011). Remaining limits: text-only prompts, `cwd` must equal the launch directory, and the tool-permission gate is deferred (`TODO(rfc010-permission-gate)` — tools run with the executor's full authority). See `packages/acp/README.md` for the full contract. diff --git a/packages/acp/README.md b/packages/acp/README.md index 78523ef774..853d03d8b7 100644 --- a/packages/acp/README.md +++ b/packages/acp/README.md @@ -1,6 +1,6 @@ # @deepseek-ai/dsh-acp -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. +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`. @@ -24,25 +24,31 @@ It is a **client-driver / UI plugin**, the structured analogue of the readline ` | ACP method | Harness seam | Notes | |---|---|---| | `initialize` | static | negotiate `protocolVersion`; advertise text-only `promptCapabilities` and `loadSession: true` | -| `session/new` | `ctx.agents.create({ sessionId, meta:{cwd} })` | single-session MVP (a 2nd is rejected — RFC 011 lifts this); `cwd` must be absolute AND equal the server launch dir; `additionalDirectories` rejected; `mcpServers` ignored | -| `session/load` | `ctx.agents.resume(...)` | replays the persisted event log to the client as `session/update` — the USER side (`user/message` → `user_message_chunk`), assistant text/reasoning (`assistant/chunk`), and tool calls/results (`tool/call` + `tool/result`). The single-session slot is reserved (`loading`) BEFORE the async resume so a pipelined `load`/`new` can't leak a second agent; the PERSISTED header `cwd` is validated via a metadata-only `list()` BEFORE resume (not just the requested `cwd`), so a mismatch rejects without ever constructing an agent. After the async resume a `closed` re-check refuses to install a record if the bridge tore down mid-load | -| `session/prompt` | `agent.send()` | text-only; rejects image/audio and empty prompts; one in-flight prompt; settles on the OWNING turn's end (a turn that ends in `error` rejects the RPC) | -| `session/cancel` | `agent.abort()` | aborts a running step + settles the prompt `cancelled` (see limitation below) | +| `session/new` | `ctx.agents.create({ sessionId, meta:{cwd} })` | creates a new session/agent; N concurrent sessions are allowed (RFC 011), keyed by id; `cwd` must be absolute AND equal the server launch dir; `additionalDirectories` rejected; `mcpServers` ignored | +| `session/load` | `ctx.agents.resume(...)` | replays the persisted event log to the client as `session/update` — the USER side (`user/message` → `user_message_chunk`), assistant text/reasoning (`assistant/chunk`), and tool calls/results (`tool/call` + `tool/result`). Re-loading an already-live id is rejected; the id's load slot is reserved (`loadingIds`) BEFORE the async resume so a pipelined load of the SAME id can't leak a second agent (distinct ids load concurrently); the PERSISTED header `cwd` is validated via a metadata-only `list()` BEFORE resume (not just the requested `cwd`), so a mismatch rejects without ever constructing an agent. After the async resume a `closed` re-check refuses to install a record if the bridge tore down mid-load | +| `session/prompt` | `agent.send()` | text-only; rejects image/audio and empty prompts; one in-flight prompt PER session (independent); settles on the OWNING turn's end (a turn that ends in `error` rejects the RPC) | +| `session/cancel` | `agent.abort()` | aborts a running step + settles the prompt `cancelled` for ONLY that session — a cancel never touches another session's stream or prompt (see limitation below) | | `session/update` | `session/event` | `agent_message_chunk` (text-delta), `agent_thought_chunk` (reasoning-delta), `user_message_chunk` (load replay), `tool_call`/`tool_call_update` | +## Multi-session (RFC 011) + +The bridge multiplexes N sessions over one connection. Live sessions are held in a `Map` (forward) with a `WeakMap` reverse map 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, so concurrent sessions never cross-settle or interleave their `session/update` notifications. State is per session: one in-flight prompt each, `session/cancel` aborts and settles only its own agent/prompt, and disposal drains every live session in parallel to quiescence. (Per-session *permission* ownership is reserved for the deferred permission gate — `TODO(rfc010-permission-gate)`.) + +Background-task isolation rides on `dsh-tool-bash`: bash task ids are global and predictable, so the tool layer records each background task's owning agent and `bash_output`/`bash_kill` reject a task owned by a different agent — one session's agent can't read or kill another's task. + ## Settle-exactly-once A `session/prompt` resolves (or rejects) exactly once, keyed off the canonical session log (the `session/event` stream), NOT the `agent/turn-start`/`agent/turn-end` events. One listener captures the prompt's owning turn from the log's `turn/start` and settles on the matching `turn/end` — the one signal that always fires (`closeTurn` appends it unconditionally, even when a boundary emit throws and the `agent/turn-end` EVENT is skipped). A prompt settles only on ITS OWN turn (`inflight.turn === turn/end.turn`), so a stale `turn/end` for a previously-cancelled turn whose end arrives late can never settle the wrong prompt. A turn that ends `error` REJECTS the RPC with an internal error carrying the failure message (ACP has no error stop reason); every other reason resolves via the codec. As a fallback, when the agent settles to `idle`/`disposed` with a prompt still pending — e.g. a `session/event` listener registered before the bridge threw and starved the bridge's listener — an `agent/status` handler reconciles the prompt from the log (the owning turn's `turn/end`, or `cancelled` if the turn was torn down without one). An empty/whitespace prompt is rejected up front — it would queue no work, so no turn would start and the RPC would hang. ## Disposal & disconnect -Teardown reaches quiescence: 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 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 (each clears the record first). +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). ## Known limitations (tracked TODOs) -- **`TODO(rfc010-permission-gate)`** — the `tools/execute` permission gate (`session/request_permission`) is NOT implemented in this PR; tools run with the executor's full authority. The ownership `WeakMap` seam is laid down so the gate (and RFC 011 per-session permission ownership) can build on it. RFC 010 stays `proposed` until the gate lands. -- **`TODO(rfc010-cancel-prestep)`** — `session/cancel` (and teardown/disconnect) is honest RPC/UI cancellation plus best-effort abort: a *running* step is aborted, but a turn that is queued-but-not-yet-started (the gap before `agent.abort()` has an `AbortController` to signal) may still run to completion. This same window means disposal/disconnect can return while one short queued turn still runs, and a prompt accepted right after a pre-step cancel can be batched into the cancelled turn (the loop merges queued messages into one turn). A loop-level queue-aware cancel will close this; the single-in-flight-prompt rule bounds the worst case to one extra prompt. -- **`TODO(rfc010-agent-disposal)`** — the factory (`ctx.agents.create`/`resume`) returns no per-agent disposer, so teardown aborts+drains the agent but cannot individually unregister it; on a bare client disconnect (no host dispose) the idled agent lingers in `ctx.agents` until the host context disposes. Single-session-per-connection makes this benign today (a reconnect spins up a fresh context); RFC 011 adds the per-session disposal seam. +- **`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. RFC 010/011 stay `proposed` until the gate (and per-session permission ownership) land. +- **`TODO(rfc010-cancel-prestep)`** — `session/cancel` (and teardown/disconnect) is honest RPC/UI cancellation plus best-effort abort: a *running* step is aborted, but a turn that is queued-but-not-yet-started (the gap before `agent.abort()` has an `AbortController` to signal) may still run to completion. This same window means disposal/disconnect can return while one short queued turn per session still runs, and a prompt accepted right after a pre-step cancel can be batched into the cancelled turn (the loop merges queued messages into one turn). A loop-level queue-aware cancel will close this; the single-in-flight-per-session rule bounds the worst case to one extra prompt per session. +- **`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. - **`cwd`** — only the server's launch directory is honored; a `session/new.cwd` (or a persisted `session/load` header cwd) that differs is rejected (RFC 010 § Deferred — no path from session cwd to the bash workdir yet). ## stdout is the protocol diff --git a/packages/acp/src/index.ts b/packages/acp/src/index.ts index 4ec7054749..71240d97c5 100644 --- a/packages/acp/src/index.ts +++ b/packages/acp/src/index.ts @@ -15,9 +15,13 @@ * that ends in `error` rejects the RPC) * - `session/cancel` → `agent.abort()` + settle the in-flight prompt * - * Single-session for the MVP (a 2nd `session/new` is rejected); RFC 011 lifts - * that. The `tools/execute` permission gate is deferred — see the - * TODO(rfc010-permission-gate) note below. + * Multi-session (RFC 011): N concurrent sessions per connection, each mapped to + * its own `LoopAgent`. Sessions are keyed by id in `sessions` (forward) with an + * `agent→sessionId` reverse map for O(1) demux of `agent/*` events; every + * `session/event` and `agent/*` event is routed strictly to its owning session + * record, so two sessions streaming at once never interleave their + * `session/update` notifications. The `tools/execute` permission gate is + * deferred — see the TODO(rfc010-permission-gate) note below. * * stdout is the protocol: this plugin must run in an example that loads NO * stdout logger (the console logger writes to stdout and would corrupt the @@ -121,9 +125,8 @@ export const Config: Schema = Schema.object({ }) /** - * Per-session bridge state. Single-entry in this MVP (RFC 011 makes the maps - * multi-entry); kept as a record from the start so RFC 011 generalizes the - * container, not the shape. + * Per-session bridge state. One per live ACP session; held in the `sessions` + * map keyed by id (RFC 011 multi-session). */ interface SessionRecord { sessionId: string @@ -169,23 +172,21 @@ export function apply(ctx: Context, config: AcpConfig): void { const agentName = config.agentName ?? 'deepseek-harness-acp' const agentVersion = config.agentVersion ?? '0.0.1' - // Single live session for the MVP. RFC 011 turns this into maps keyed by - // sessionId plus an agent→sessionId reverse map for the permission gate. - let record: SessionRecord | undefined - // True while a `session/load` is between reserving the single-session slot and - // installing its `record` (resume() is async). The session guards check BOTH - // `record` and `loading` so a pipelined load/new cannot slip past while the - // first load's resume() is pending and leak a second live agent. - let loading = false + // 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 + // `bySession` together, and removed together. + const sessions = new Map() + const bySession = new WeakMap() + // Session ids whose `session/load` is mid-`resume()` (the slot is reserved + // before the async resume so a pipelined load/new for the SAME id can't create + // two agents). Distinct ids load concurrently; a given id loads once at a time. + const loadingIds = new Set() // Set once the bridge has torn down (disposal or client disconnect). An async - // `session/load` that was mid-`resume()` when teardown ran must observe this - // after its await and NOT install a `record` (which would resurrect a live - // agent/listeners after the bridge closed). Checked after every load await. + // `session/load` mid-`resume()` when teardown ran must observe this after its + // await and NOT install a record (which would resurrect a live agent/listeners + // after the bridge closed). Checked after every load await. let closed = false - // Ownership marker: agents this bridge created. The deferred permission gate - // (TODO(rfc010-permission-gate)) and RFC 011 build on this; laid down now so - // the seam exists. A WeakMap so a disposed agent's entry is collectable. - const owned = new WeakMap() // Assigned at the bottom, before any agent event can fire (a session only // exists after `newSession`, which the client calls after construction), so @@ -207,10 +208,11 @@ export function apply(ctx: Context, config: AcpConfig): void { /** Resolve the live record for a sessionId, or throw an ACP error. */ const requireSession = (sessionId: string): SessionRecord => { - if (record === undefined || record.sessionId !== sessionId) { + const rec = sessions.get(sessionId) + if (rec === undefined) { throw invalidParams(`unknown session: ${sessionId}`) } - return record + return rec } /** Push a `session/update` notification, swallowing post-close rejections. */ @@ -252,10 +254,11 @@ export function apply(ctx: Context, config: AcpConfig): void { // already-cancelled turn whose end arrives late is ignored (see // SessionRecord.inflight). A turn that ends `error` REJECTS the prompt (ACP // has no error stop reason); other reasons resolve via the codec. Demux - // strictly by session id. + // strictly by session id: a `session/event` is routed to its own record, so + // two sessions streaming at once never cross-settle or interleave updates. ctx.on('session/event', (session, event: SessionEvent) => { - const rec = record - if (rec === undefined || session.header.id !== rec.sessionId) return + const rec = sessions.get(session.header.id) + if (rec === undefined) return streamSessionEventUpdate(rec.sessionId, event, notify) const inflight = rec.inflight if (inflight === undefined) return @@ -333,9 +336,12 @@ export function apply(ctx: Context, config: AcpConfig): void { // On a settle to idle/disposed, reconcile any still-pending prompt from the // log (covers a starved `session/event` listener — see settleFromLog). A mid- // step disposal that never appended a clean turn/end resolves `cancelled`. + // Demux via the agent→sessionId reverse map. ctx.on('agent/status', (agent, status: AgentStatus) => { - const rec = record - if (rec === undefined || owned.get(agent) !== rec.sessionId) return + const sessionId = bySession.get(agent) + if (sessionId === undefined) return + const rec = sessions.get(sessionId) + if (rec === undefined) return if (status === 'idle' || status === 'disposed') settleFromLog(rec) }) @@ -369,9 +375,6 @@ export function apply(ctx: Context, config: AcpConfig): void { newSession(params: NewSessionRequest): Promise { assertOpen() - if (record !== undefined || loading) { - throw invalidParams('this agent supports a single session; a session already exists (RFC 011 will lift this)') - } validateWorkspaceParams(params) const sessionId = randomUUID() const agent = ctx.agents.create({ @@ -380,24 +383,23 @@ export function apply(ctx: Context, config: AcpConfig): void { meta: { cwd: params.cwd }, agentOptions: agentOptions(config), }) - owned.set(agent, sessionId) - record = { sessionId, agent, inflight: undefined } + bySession.set(agent, sessionId) + sessions.set(sessionId, { sessionId, agent, inflight: undefined }) return Promise.resolve({ sessionId }) }, async loadSession(params: LoadSessionRequest): Promise { assertOpen() - if (record !== undefined || loading) { - throw invalidParams('this agent supports a single session; a session already exists (RFC 011 will lift this)') + if (sessions.has(params.sessionId) || loadingIds.has(params.sessionId)) { + throw invalidParams(`session ${params.sessionId} is already loaded`) } validateWorkspaceParams(params) - // Reserve the single-session slot BEFORE the await. Without this, two - // pipelined load/new requests could both pass the guard above while the - // first load's resume() is pending, then both install a record and leak - // a second live agent. `loading` claims the slot; it is cleared in - // `finally` so a rejected load (bad id, cwd mismatch) never wedges all - // future sessions on this connection. - loading = true + // Reserve THIS id's load slot BEFORE the await. Without it, two pipelined + // loads for the same id could both pass the guard above while the first + // resume() is pending, then both install a record and leak a second + // agent. (Distinct ids load concurrently — the set is keyed by id.) The + // slot is released in `finally` so a rejected load never wedges the id. + loadingIds.add(params.sessionId) try { // Validate the PERSISTED cwd BEFORE resuming — `list()` is a // metadata-only read (no full-log parse) — so a mismatch rejects @@ -419,7 +421,7 @@ export function apply(ctx: Context, config: AcpConfig): void { agentOptions: agentOptions(config), }) // The bridge may have torn down (disposal / client disconnect) while - // resume() was pending. Its listeners are gone, so installing `record` + // 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)). @@ -430,8 +432,8 @@ export function apply(ctx: Context, config: AcpConfig): void { if (closed) { throw invalidParams('connection closed during session/load') } - owned.set(agent, params.sessionId) - record = { sessionId: params.sessionId, agent, inflight: undefined } + bySession.set(agent, params.sessionId) + sessions.set(params.sessionId, { sessionId: params.sessionId, agent, inflight: undefined }) // Replay the persisted event log to the client as session/update. Use // the raw event log (NOT deriveMessages, which drops assistant/chunk // and trace events): RFC 010's load contract reconstructs the streamed @@ -442,7 +444,7 @@ export function apply(ctx: Context, config: AcpConfig): void { } return {} } finally { - loading = false + loadingIds.delete(params.sessionId) } }, @@ -477,13 +479,15 @@ export function apply(ctx: Context, config: AcpConfig): void { }, cancel(params: CancelNotification): Promise { - const rec = record - if (rec === undefined || rec.sessionId !== params.sessionId) return Promise.resolve() + const rec = sessions.get(params.sessionId) + if (rec === undefined) return Promise.resolve() // RFC 010: session/cancel maps to agent.abort(reason). This aborts a // RUNNING step (the turn ends 'aborted' → 'cancelled' via turn-end). - // It also settles the in-flight prompt as cancelled directly, in case - // the abort lands in the pre-step window (queued-but-not-started) where - // abort() has no AbortController to signal — see the README + // It aborts and settles ONLY this session's agent/prompt — a cancel in + // one session never touches another's stream or pending prompt (RFC 011 + // isolation). It also settles the in-flight prompt as cancelled directly, + // in case the abort lands in the pre-step window (queued-but-not-started) + // where abort() has no AbortController to signal — see the README // TODO(rfc010-cancel-prestep): a not-yet-started queued turn may still // run to completion until a loop-level cancel lands. Best-effort abort // plus honest RPC/UI cancellation. A secondary consequence of that same @@ -515,11 +519,12 @@ export function apply(ctx: Context, config: AcpConfig): void { conn = new AgentSideConnection(makeAgent, stream) /** - * Tear the live session down to quiescence (AGENTS.md "dispose must reach - * quiescence"): 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). Idempotent - * — clears `record` first, so a second call (close racing dispose) is a no-op. + * 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. * Shared by Cordis disposal AND client disconnect (`conn.closed`). * * Caveat (same window as TODO(rfc010-cancel-prestep)): if teardown lands in @@ -528,38 +533,36 @@ export function apply(ctx: Context, config: AcpConfig): void { * `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, out of the RFC 010 MVP scope); for `newSession` agents - * the worst case is one short queued turn, since the bridge enforces a single - * in-flight prompt. + * loop-level change); the single-in-flight-per-session rule bounds the worst + * case to one short queued turn per session. * - * The agent itself is NOT individually disposed/unregistered here — the - * factory (`ctx.agents.create`/`resume`) registers it on the AgentLoop fiber - * and returns no per-agent disposer, so the registry entry is reclaimed when + * 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 agent therefore lingers in `ctx.agents` until shutdown; - * since the MVP is single-session-per-connection and a reconnect spins up a - * fresh context, this does not strand work. A per-agent disposal seam is - * RFC 011 follow-up (TODO(rfc010-agent-disposal)). + * 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)). */ let quiescing: Promise | undefined const quiesce = (): Promise => { // Memoize: disposal and client-disconnect can both fire. The first call owns // the teardown; later callers await the SAME promise so `fiber.dispose()` - // never returns before an in-flight close teardown has finished (using - // `record === undefined` as the only guard would let the second caller - // return early while the first is still awaiting whenIdle()). + // never returns before an in-flight close teardown has finished. if (quiescing !== undefined) return quiescing - // Mark closed BEFORE the record check: a `session/load` mid-`resume()` (no - // record installed yet) must observe this after its await and refuse to - // install a post-teardown record. Set even when there is nothing else to do. + // Mark closed BEFORE draining: a `session/load` mid-`resume()` (no record + // installed yet) must observe this after its await and refuse to install a + // post-teardown record. Set even when there are no live sessions. closed = true - const rec = record - record = undefined - if (rec === undefined) return Promise.resolve() + const recs = [...sessions.values()] + sessions.clear() + if (recs.length === 0) return Promise.resolve() quiescing = (async () => { - settlePrompt(rec, 'cancelled') - rec.agent.abort('disposed') - await rec.agent.whenIdle() + await Promise.all(recs.map(async (rec) => { + settlePrompt(rec, 'cancelled') + rec.agent.abort('disposed') + await rec.agent.whenIdle() + })) })() return quiescing } diff --git a/packages/acp/tests/bridge.spec.ts b/packages/acp/tests/bridge.spec.ts index a1396434e9..9ff75c0ec5 100644 --- a/packages/acp/tests/bridge.spec.ts +++ b/packages/acp/tests/bridge.spec.ts @@ -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 () => { diff --git a/packages/acp/tests/harness.ts b/packages/acp/tests/harness.ts index d546706bca..9c36d1847a 100644 --- a/packages/acp/tests/harness.ts +++ b/packages/acp/tests/harness.ts @@ -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 { 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() diff --git a/packages/acp/tests/load.spec.ts b/packages/acp/tests/load.spec.ts index cde8d46d95..e78b64c710 100644 --- a/packages/acp/tests/load.spec.ts +++ b/packages/acp/tests/load.spec.ts @@ -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/) }) }) diff --git a/packages/acp/tests/multi-session.spec.ts b/packages/acp/tests/multi-session.spec.ts new file mode 100644 index 0000000000..1c20d2ba39 --- /dev/null +++ b/packages/acp/tests/multi-session.spec.ts @@ -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((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') + }) +}) diff --git a/packages/tool-bash/README.md b/packages/tool-bash/README.md index 74a4ae5011..2736884765 100644 --- a/packages/tool-bash/README.md +++ b/packages/tool-bash/README.md @@ -28,6 +28,10 @@ Result text: stdout, then a `[stderr]` section, then status markers — `[timed `task_id` → SIGTERM→SIGKILL on the task's process group. Killing an already-finished task is a reported no-op; unknown ids are errors. +### Task ownership (cross-session isolation) + +The owning agent is recorded per task id at spawn and kept for the lifetime of the loaded plugin instance (it is **not** cleared on completion). `bash_output`/`bash_kill` reject a task owned by a *different* agent with `task belongs to another session` (a task started with no agent — a non-loop caller — has no owner and is open to anyone; a call with no `exec.agent` cannot access an owned task). Task ids are global and predictable, so under multi-session ACP this ownership check is the fence that stops one session's agent from reading or killing another session's background task. (`TODO(tool-bash-owner-hmr)`: an independent HMR reload of this plugin starts a fresh map, so a task spawned before the reload becomes un-owned — acceptable as HMR is dev-only and the session boundary is one user's cooperative editor; a durable fix attaches ownership to the executor/task lifetime.) + ## Background completion notices When a background task finishes, a short notice is injected into the owning agent's session (`agent.inject()`, source `{kind: 'plugin', plugin: 'tool-bash'}`). Injection is **durable context for the next model request, not a wake-up** — an idle agent stays idle until something sends a message. That's why the tool descriptions tell the model to poll with `bash_output`. diff --git a/packages/tool-bash/src/index.ts b/packages/tool-bash/src/index.ts index 78a83b9a5f..b812b32f83 100644 --- a/packages/tool-bash/src/index.ts +++ b/packages/tool-bash/src/index.ts @@ -11,6 +11,23 @@ * message, which is why the tool descriptions tell the model to poll with * `bash_output`. * + * Task ownership: the owning agent is recorded per task id at spawn and kept + * for the lifetime of THIS plugin instance (it is NOT cleared on task + * completion — a finished task must stay un-readable / un-killable by a + * different agent). `bash_output`/`bash_kill` reject a task owned by a DIFFERENT + * agent (a task with no recorded owner is open to anyone). Task ids are global + * and predictable (`bash-1`, …); under multi-session ACP (RFC 011) this + * ownership check is the fence that stops one session's agent from reading or + * killing another session's background task. + * + * TODO(tool-bash-owner-hmr): the ownership map is per-plugin-instance, so an + * independent HMR reload of `tool-bash` (without reloading `dsh-bash`) starts a + * fresh map and a task spawned before the reload becomes un-owned (open to any + * caller). This is acceptable today — HMR is dev-only, the ACP session boundary + * is one user's cooperative editor (not an adversarial trust boundary), and the + * executor's own disposal kills its tasks — but a durable fix would attach + * ownership to the executor/task lifetime via a `dsh-bash` seam. + * * TODO(permissions): commands run with the executor's full authority. The * permission/sandbox seam is the `tools/execute` waterfall (veto/ask) plus * sandboxing `BashExecutor` implementations — see docs/architecture.md @@ -115,12 +132,33 @@ function statusLine(task: BashTask): string { } export function apply(ctx: Context): void { + // Owning agent per background task id, recorded at spawn. Kept for the + // lifetime of THIS plugin instance (NOT cleared on completion): a completed + // task must stay un-readable / un-killable by a DIFFERENT agent, so the + // ownership record outlives the task. Under multi-session ACP (RFC 011) this + // is the isolation fence — one session's agent must never read or kill + // another session's background task. A task with no recorded owner (started by + // a non-loop caller, `exec.agent` absent) is unowned and accessible to anyone. + // An independent `tool-bash` HMR reload resets this map — see the + // TODO(tool-bash-owner-hmr) note in the module doc. + const taskOwner = new Map() + + /** + * Authorize a `bash_output`/`bash_kill` call against a task's owner. Rejects + * when the task has a recorded owner and the caller is not that exact agent — + * including the conservative no-agent case (`exec.agent` absent cannot prove + * ownership of an owned task). An unowned task (no record) is allowed. + */ + const assertTaskAccess = (taskId: string, exec: { agent?: Agent }): void => { + const owner = taskOwner.get(taskId) + if (owner !== undefined && owner !== exec.agent) { + throw new Error(`task ${taskId} belongs to another session`) + } + } + // Background completion → inject a notice into the owning agent's session. - // Tracks the agent per task id; entries drop once notified. - const owners = new Map() ctx.bash.onTaskDone((task) => { - const agent = owners.get(task.id) - owners.delete(task.id) + const agent = taskOwner.get(task.id) if (!agent) return try { agent.inject( @@ -171,7 +209,7 @@ export function apply(ctx: Context): void { } if (args.run_in_background === true) { const task = ctx.bash.start(ctx.bash.resolve(request)) - if (exec.agent) owners.set(task.id, exec.agent) + if (exec.agent) taskOwner.set(task.id, exec.agent) return [{ type: 'text', text: `started background task ${task.id}` }] } const result = await ctx.bash.run(ctx.bash.resolve(request)) @@ -190,8 +228,10 @@ export function apply(ctx: Context): void { }, // execute is synchronous (registry reads + string shaping) but the // ToolDefinition contract wants a Promise — hence resolve(), not async. - execute(args) { - const read = ctx.bash.readOutput(validateTaskId(args.task_id)) + execute(args, exec) { + const id = validateTaskId(args.task_id) + assertTaskAccess(id, exec) + const read = ctx.bash.readOutput(id) let text = read.delta.length > 0 ? read.delta : '(no new output)' if (read.lossy) { const paths = [read.stdoutSpillPath, read.stderrSpillPath].filter((p): p is string => p !== undefined) @@ -208,8 +248,9 @@ export function apply(ctx: Context): void { parameters: { task_id: { type: 'string', required: true, description: 'Task id returned by the bash tool.' }, }, - execute(args) { + execute(args, exec) { const id = validateTaskId(args.task_id) + assertTaskAccess(id, exec) const killed = ctx.bash.kill(id) return Promise.resolve([{ type: 'text', diff --git a/packages/tool-bash/tests/tools.spec.ts b/packages/tool-bash/tests/tools.spec.ts index 23bcd35c01..c5c058a6f2 100644 --- a/packages/tool-bash/tests/tools.spec.ts +++ b/packages/tool-bash/tests/tools.spec.ts @@ -338,6 +338,106 @@ describe('background tools', () => { }) }) +describe('background task ownership (cross-session isolation)', () => { + /** Run a tool on behalf of a specific agent (sets exec.agent). */ + function callAs(ctx: Context, agent: import('@deepseek-ai/dsh-agent').Agent | undefined, name: string, args: unknown) { + return ctx.tools.execute({ callId: CallId(`own-${++callCounter}`), name, arguments: args, ...agent ? { agent } : {} }) + } + // Distinct identities — ownership is by agent object identity, not id. + const fakeAgent = () => ({ inject: () => undefined }) as unknown as import('@deepseek-ai/dsh-agent').Agent + + it('rejects bash_output/bash_kill for a task owned by a DIFFERENT agent', async () => { + const ctx = await setup() + const a = fakeAgent() + const b = fakeAgent() + // Agent A starts a long-running background task. + const started = await callAs(ctx, a, 'bash', { command: 'sleep 60', description: 'bg', run_in_background: true }) + const id = /task (bash-\d+)/.exec(text(started))![1]! + + // Agent B cannot read or kill A's task. + const readByB = await callAs(ctx, b, 'bash_output', { task_id: id }) + expect(readByB.isError).toBe(true) + expect(text(readByB)).toMatch(/belongs to another session/) + const killByB = await callAs(ctx, b, 'bash_kill', { task_id: id }) + expect(killByB.isError).toBe(true) + expect(text(killByB)).toMatch(/belongs to another session/) + + // The task is still running (B's kill did nothing) — A can still kill it. + const killByA = await callAs(ctx, a, 'bash_kill', { task_id: id }) + expect(killByA.isError).toBe(false) + expect(text(killByA)).toBe(`killed background task ${id}`) + }) + + it('the no-agent (non-loop) caller cannot access an owned task', async () => { + const ctx = await setup() + const a = fakeAgent() + const started = await callAs(ctx, a, 'bash', { command: 'sleep 60', description: 'bg', run_in_background: true }) + const id = /task (bash-\d+)/.exec(text(started))![1]! + // A call with no exec.agent cannot prove ownership of an owned task. + const read = await callAs(ctx, undefined, 'bash_output', { task_id: id }) + expect(read.isError).toBe(true) + expect(text(read)).toMatch(/belongs to another session/) + await callAs(ctx, a, 'bash_kill', { task_id: id }) // cleanup + }) + + it('an UNOWNED task (started with no agent) is accessible to anyone', async () => { + const ctx = await setup() + // Started by a non-loop caller (no exec.agent) → no recorded owner. + const started = await callAs(ctx, undefined, 'bash', { command: 'sleep 60', description: 'bg', run_in_background: true }) + const id = /task (bash-\d+)/.exec(text(started))![1]! + // Any agent (and the no-agent caller) may read/kill it. + const read = await callAs(ctx, fakeAgent(), 'bash_output', { task_id: id }) + expect(read.isError).toBe(false) + const killed = await callAs(ctx, undefined, 'bash_kill', { task_id: id }) + expect(killed.isError).toBe(false) + }) + + it('the owner can still access its task AFTER it completes (owner record persists)', async () => { + const ctx = await setup() + const a = fakeAgent() + const b = fakeAgent() + const started = await callAs(ctx, a, 'bash', { command: 'echo done', description: 'bg', run_in_background: true }) + const id = /task (bash-\d+)/.exec(text(started))![1]! + await ctx.bash.get(id)!.done + // Completion does NOT clear ownership: B is still rejected, A still allowed. + const readByB = await callAs(ctx, b, 'bash_output', { task_id: id }) + expect(readByB.isError).toBe(true) + expect(text(readByB)).toMatch(/belongs to another session/) + const readByA = await callAs(ctx, a, 'bash_output', { task_id: id }) + expect(readByA.isError).toBe(false) + }) + + it('documents the HMR caveat: an independent tool-bash reload resets ownership', async () => { + // The ownership map is per-plugin-instance (TODO(tool-bash-owner-hmr)). When + // ONLY tool-bash is reloaded (bash/executor + task survive), the new instance + // has an empty map, so the previously-owned task becomes unowned (open). This + // test pins that documented behavior — a regression here (e.g. an accidental + // global map) would change it. + const ctx = new Context() + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000 }) + ;(ctx.bash as LocalBashExecutor).internals = { spillDir, graceMs: 200 } + const fiber = await ctx.plugin(ToolBash) + + const a = fakeAgent() + const b = fakeAgent() + const started = await callAs(ctx, a, 'bash', { command: 'sleep 60', description: 'bg', run_in_background: true }) + const id = /task (bash-\d+)/.exec(text(started))![1]! + // Before reload: B is rejected (A owns it). + expect((await callAs(ctx, b, 'bash_output', { task_id: id })).isError).toBe(true) + + // Reload ONLY tool-bash; the executor and its running task survive. + await fiber.dispose() + await ctx.plugin(ToolBash) + expect(ctx.bash.get(id)?.status).toBe('running') + + // After reload the fresh map has no owner → B can now access it (the caveat). + expect((await callAs(ctx, b, 'bash_output', { task_id: id })).isError).toBe(false) + await callAs(ctx, b, 'bash_kill', { task_id: id }) // cleanup + }) +}) + describe('renderResult', () => { const base = { exitCode: 0 as number | null,