feat(acp): ACP bridge — drive the coding agent from an editor over JSON-RPC stdio

Implements the RFC 010 MVP: a new `@deepseek-ai/dsh-acp` package bridges the
harness agent to the Agent Client Protocol (JSON-RPC 2.0 over newline-delimited
stdio), so Zed and other ACP editors can drive the coding agent — streaming
render, tool-call display, and resumable sessions via `session/load`.

- packages/acp: AgentSideConnection wiring; initialize/newSession/loadSession/
  prompt/cancel; a total TurnEndReason→StopReason codec; settle-once with a
  fallback chain (agent/turn-end → logged turn/end → idle); single-session
  guard; cwd-must-equal-launch-dir validation; load replays from the persisted
  event log (assistant/chunk→agent_message_chunk, tool/call/result→tool_call*).
- agent: add Agent.whenIdle() quiescence signal to the interface; LoopAgent
  implements it (resolves on the first running→idle/disposed transition). The
  bridge awaits it on disposal so teardown reaches quiescence, not just abort.
- examples: extract the shared provider/tool core into examples/base.yml;
  coding-agent nest-includes it; new examples/acp-agent serves the agent over
  ACP with JSONL persistence and no stdout logger (stdout is the protocol).
- Permission gate deferred (TODO(rfc010-permission-gate)): tools run with the
  executor's full authority; only the Agent→sessionId ownership seam is laid
  down. Cancel is best-effort for a not-yet-started queued turn
  (TODO(rfc010-cancel-prestep)). RFC 010 stays `proposed`.
- Docs: package README + Zed snippet; client-driver cookbook section; root and
  packages layout/commands; RFC 010 implementation-status note.

48 bridge tests + whenIdle coverage; 100% per-file coverage; e2e boots the
example as a subprocess and verifies a written file on disk (key-gated, with a
no-key stdout-purity check).
This commit is contained in:
Tianyi Cui
2026-06-16 18:44:31 +08:00
parent add59a3336
commit fb9636db44
39 changed files with 2819 additions and 51 deletions
+8
View File
@@ -33,10 +33,15 @@ packages/ Harness packages, all named @deepseek-ai/dsh-<name>:
bash/ abstract bash executor seam (ctx.bash) — interface only
bash-local/ local-subprocess BashExecutor implementation
tool-bash/ model-facing bash/bash_output/bash_kill tool schemas
acp/ Agent Client Protocol bridge: drive the agent from an ACP
editor (Zed) over JSON-RPC stdio
examples/ Runnable demos (not workspaces). echo-agent = mock model + echo
tool + stdio UI + JSONL persistence, wired via cordis.yml.
coding-agent = the real thing: DeepSeek V4 + bash tools
(yarn demo:coding, needs DEEPSEEK_API_KEY).
acp-agent = the coding agent exposed as an ACP server over
JSON-RPC stdio (yarn demo:acp, needs DEEPSEEK_API_KEY).
base.yml = shared provider/tool core both real demos include.
docs/ architecture.md — the design doc. adr/ — decision records (the
why behind vendoring, event-sourcing, the schema DSL, …).
rfc/ — proposals for substantial future work.
@@ -72,6 +77,9 @@ yarn demo:echo # run examples/echo-agent (no API key; type "echo hi" to
# see a tool call) — the mock skeleton
yarn demo:coding # run examples/coding-agent — the real agent (needs
# DEEPSEEK_API_KEY; give it a coding task)
yarn demo:acp # run examples/acp-agent — the coding agent as an ACP
# server over JSON-RPC stdio (needs DEEPSEEK_API_KEY;
# drive it from Zed or another ACP client)
```
## Secrets / .env
+1
View File
@@ -119,6 +119,7 @@ Tool schemas are deliberately **part of the assembly**: "what the model is told
- `steer(content)` — mid-turn injection, drained **between steps**; behaves like `send` when idle
- `inject(content)` — in-session context (`context/message` event); the next request sees it (Claude Code attachment / system-reminder analog). An inject made while the agent is *running* joins the open turn; an inject while *idle* is wrapped in a one-shot turn (`turn/start{trigger:injection}``context/message``turn/end`) so every event stays turn-enclosed (see ADR 0017).
- `abort(reason)` — aborts the in-flight step via `AbortSignal`
- `whenIdle()` — resolves once the agent reaches quiescence after settling out of `running` (resolves immediately when already idle; awaits the loop exit when disposed). The teardown signal: `abort()` then `await whenIdle()` guarantees the in-flight turn has fully stopped. Observes the transition without disposing the agent.
- `session`, `status`, `options`
**TODO(sub-agents)**: `spawn`/`fork` land on `AgentLoop.create()` — fork seeds the child Session with the parent's event log, spawn starts fresh; children are ordinary `Agent` handles so `steer()` and event subscription work uniformly. Inter-agent channels beyond these primitives are deliberately deferred.
+28 -1
View File
@@ -53,6 +53,33 @@ export function apply(ctx: Context) {
}
```
## A client-driver plugin (external protocol bridge)
A *client driver* is a UI plugin whose "user" is another program speaking a wire protocol rather than a human at a terminal. It owns the process's stdio (so it must run with **no stdout logger** — every non-protocol byte corrupts the stream), creates/resumes agents on demand through the `dsh-agent` factory seam, translates harness events (`session/event`, `agent/*`) into outbound protocol messages, and translates inbound requests back into `agent.send()` / `agent.abort()`. Two harness-specific contracts make it correct: resolve each request exactly once off a settle signal (the turn can end without its `agent/turn-end` event firing — fall back through the logged `turn/end` record), and on disposal reach quiescence (`await agent.whenIdle()` after `abort()`), not just request it.
`packages/acp` is the worked example: it bridges the agent to the Agent Client Protocol (JSON-RPC over stdio) so Zed and other ACP editors can drive it. See its README for the full method surface and the deferred-permission-gate note.
```ts
import type { Context } from 'cordis'
export const name = 'my-protocol-bridge'
export const inject = ['agents', 'sessions', 'sessionPersistence']
export function apply(ctx: Context) {
// Stream every logged assistant text/reasoning delta out to the client.
ctx.on('session/event', (_session, event) => {
if (event.type === 'assistant/chunk') {
const chunk = event.data.chunk
if (chunk.type === 'text-delta') {
// sendToClient({ kind: 'message_chunk', text: chunk.text })
}
}
})
// Inbound "prompt": create/resume an agent and feed it; settle on turn end.
// Disposal awaits quiescence: agent.abort() then await agent.whenIdle().
}
```
## Runnable wirings
Two complete examples load their plugin trees from `cordis.yml` with HMR: [`examples/echo-agent`](../../examples/echo-agent) (mock model + echo tool — the all-mock skeleton check, `yarn demo:echo`) and [`examples/coding-agent`](../../examples/coding-agent) (DeepSeek V4 + the bash tool suite — the real thing, `yarn demo:coding`).
Three complete examples load their plugin trees from `cordis.yml` with HMR: [`examples/echo-agent`](../../examples/echo-agent) (mock model + echo tool — the all-mock skeleton check, `yarn demo:echo`), [`examples/coding-agent`](../../examples/coding-agent) (DeepSeek V4 + the bash tool suite — the real thing, `yarn demo:coding`), and [`examples/acp-agent`](../../examples/acp-agent) (the same coding agent exposed as an ACP server over JSON-RPC stdio — the client-driver shape, `yarn demo:acp`). The two real demos share their provider/tool core via [`examples/base.yml`](../../examples/base.yml).
@@ -2,6 +2,8 @@
Status: proposed
> **Implementation status (MVP landed):** steps 1, 2, 3, 4, 6, 7, 8 are implemented in `packages/acp` + `examples/acp-agent`. **Step 5 (the `session/request_permission` permission gate) is deferred** — the bridge ships a pass-through (tools run with the executor's full authority) marked `TODO(rfc010-permission-gate)`, and lays down only the `WeakMap<Agent, sessionId>` ownership seam the gate will build on. Status stays `proposed` until the gate lands. One further best-effort limitation is tracked as `TODO(rfc010-cancel-prestep)`: `session/cancel` aborts a running step and settles the RPC as `cancelled`, but a turn still queued (not yet started) when the cancel arrives may execute before the abort takes effect, pending a loop-level pre-step cancel.
## Problem
The coding agent is reachable only through the readline `stdio-chat` plugin: it reads lines from stdin, calls `agent.send()`, and prints `agent/stream-chunk` to stdout. There is no structured protocol, so the agent cannot be embedded in an editor — no streaming render, no tool-call display, no permission UI, no resumable sessions.
+35
View File
@@ -0,0 +1,35 @@
# acp-agent example
The DeepSeek Harness coding agent exposed as an **Agent Client Protocol (ACP)** server over JSON-RPC stdio — drive it from Zed or any other ACP client.
```sh
yarn demo:acp # needs DEEPSEEK_API_KEY (repo-root .env or env)
```
This boots `@deepseek-ai/dsh-acp` over the shared provider/tool core (`../base.yml`), with `agent-loop` configured with **no pre-created agents** (ACP `session/new` creates them on demand) and JSONL session persistence (so `session/load` works).
## stdout is the protocol
This example loads **no stdout logger**`stdout` carries the JSON-RPC frames, and any other write corrupts them. Do not add `@cordisjs/plugin-logger-console` or a stdio UI here. Use a stderr exporter if you need logs.
## Zed configuration
Add to your Zed `settings.json` under `agent_servers`:
```json
{
"agent_servers": {
"DeepSeek Harness": {
"command": "yarn",
"args": ["demo:acp"],
"env": { "DEEPSEEK_API_KEY": "sk-…" }
}
}
}
```
Run from the repo root (the MVP requires the server's launch directory to be the workspace — see the `cwd` note in `packages/acp`).
## 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.
+49
View File
@@ -0,0 +1,49 @@
# The acp-agent plugin tree, loaded via @cordisjs/plugin-include.
#
# CRITICAL: this example loads NO stdout logger (no @cordisjs/plugin-logger-
# console, no stdio-chat). stdout is reserved for the ACP JSON-RPC protocol —
# anything else written there corrupts the frames (see packages/acp, RFC 010 §
# Risks). Use a stderr exporter if you need logging. The timer plugin is loaded
# (no stdout writes); hmr is omitted (an editor manages the subprocess).
#
# Requires DEEPSEEK_API_KEY (and optionally DEEPSEEK_BASE_URL) in the
# environment — start.ts loads the gitignored repo-root .env first.
- id: timer
name: '@cordisjs/plugin-timer'
# Shared provider/tool core (llm, sessions, system-prompt, tools, agents,
# invariants, llm-deepseek, bash-local, tool-bash). Nested include resolved
# relative to THIS file's directory.
- id: base
name: '@cordisjs/plugin-include'
config:
path: '../base.yml'
# agent-loop with NO pre-created agents: ACP `session/new` creates them on
# demand (unlike coding-agent, which pre-creates `main`).
- id: agent-loop
name: '@deepseek-ai/dsh-agent-loop'
config:
agents: []
# Durable session persistence — required by the ACP bridge for `session/load`.
- id: session-persistence
name: '@deepseek-ai/dsh-session-persistence-jsonl'
config:
root: './.sessions'
# The ACP bridge: wires AgentSideConnection to stdin/stdout.
- id: acp
name: '@deepseek-ai/dsh-acp'
config:
model: deepseek-v4-flash
systemPrompt: |
You are a coding assistant driven over the Agent Client Protocol.
Your only tools are bash (plus bash_output/bash_kill for background
tasks). Do ALL file operations through bash: read with cat/sed/head,
search with grep, write with heredocs (cat <<'EOF' > file), edit with
sed or a rewrite. Each bash call runs in a fresh shell — pass workdir
instead of cd. Check the [exit code: N] marker; verify your work. Keep
answers brief and factual.
+7
View File
@@ -0,0 +1,7 @@
{
"name": "acp-agent-example",
"description": "Runnable demo: the coding agent as an ACP server over JSON-RPC stdio (Zed & other ACP editors)",
"private": true,
"version": "0.0.1",
"type": "module"
}
+30
View File
@@ -0,0 +1,30 @@
import { pathToFileURL } from 'node:url'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
// Load DEEPSEEK_API_KEY / DEEPSEEK_BASE_URL from a gitignored repo-root .env
// (Node native). Absent file is fine — the environment may already carry them.
//
// IMPORTANT: this server speaks ACP JSON-RPC on stdout. Do NOT add any
// stdout logging here or in cordis.yml — it would corrupt the protocol frames.
// A present-but-unreadable/malformed .env is a real misconfiguration: surface
// it on STDERR (never stdout) rather than silently running with the wrong env.
try {
process.loadEnvFile(new URL('../../.env', import.meta.url).pathname)
} catch (error) {
if ((error as NodeJS.ErrnoException | null)?.code !== 'ENOENT') {
process.stderr.write(`acp-agent: failed to load .env: ${String(error)}\n`)
}
// ENOENT (no .env) is fine — rely on the ambient environment.
}
const ctx = new Context()
ctx.baseUrl = pathToFileURL(import.meta.dirname).href + '/'
await ctx.plugin(Loader)
await ctx.loader.create({
name: '@cordisjs/plugin-include',
config: {
path: './cordis.yml',
},
})
+142
View File
@@ -0,0 +1,142 @@
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { Readable, Writable } from 'node:stream'
import { mkdtemp, rm, readFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { fileURLToPath } from 'node:url'
import { afterEach, describe, expect, it } from 'vitest'
import {
ClientSideConnection,
ndJsonStream,
PROTOCOL_VERSION,
type Agent as AcpAgent,
type Client,
type RequestPermissionRequest,
type RequestPermissionResponse,
type SessionNotification,
} from '@agentclientprotocol/sdk'
/**
* End-to-end: boot examples/acp-agent as a real subprocess speaking ACP over
* its stdio, drive it with a real ClientSideConnection, send a real prompt, and
* verify the WORLD (a file the agent wrote), not the agent's self-report. Owns
* and disposes the subprocess in afterEach. Key-gated.
*
* Also asserts stdout purity (only framed JSON-RPC on stdout) — that one runs
* WITHOUT a key, since it only needs the server to boot and answer initialize.
*/
const startScript = fileURLToPath(new URL('../start.ts', import.meta.url))
// Resolve tsx's loader to an ABSOLUTE path: the subprocess runs with cwd set to
// a temp workdir (the MVP requires session cwd === process.cwd()), where a bare
// `--import tsx` would not resolve from node_modules. import.meta.resolve gives
// the worktree's tsx regardless of the child's cwd.
const tsxLoader = fileURLToPath(import.meta.resolve('tsx'))
interface Spawned {
child: ChildProcessWithoutNullStreams
client: ClientSideConnection
updates: SessionNotification['update'][]
stderr: string[]
}
function spawnAcpAgent(cwd: string): Spawned {
const child = spawn(
process.execPath,
['--import', tsxLoader, startScript],
{ cwd, env: { ...process.env }, stdio: ['pipe', 'pipe', 'pipe'] },
)
const stderr: string[] = []
child.stderr.setEncoding('utf8')
child.stderr.on('data', (chunk: string) => stderr.push(chunk))
const updates: SessionNotification['update'][] = []
const stream = ndJsonStream(
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
Readable.toWeb(child.stdout) as ReadableStream<Uint8Array>,
)
const makeClient = (_agent: AcpAgent): Client => ({
sessionUpdate(params: SessionNotification): Promise<void> {
updates.push(params.update)
return Promise.resolve()
},
requestPermission(_params: RequestPermissionRequest): Promise<RequestPermissionResponse> {
// Permission gate is deferred (TODO(rfc010-permission-gate)); the bridge
// never requests permission yet, so just allow if it ever does.
return Promise.resolve({ outcome: { outcome: 'cancelled' } })
},
})
const client = new ClientSideConnection(makeClient, stream)
return { child, client, updates, stderr }
}
let spawned: Spawned | undefined
let workdir: string | undefined
afterEach(async () => {
if (spawned) {
spawned.child.kill('SIGKILL')
spawned = undefined
}
if (workdir !== undefined) await rm(workdir, { recursive: true, force: true })
workdir = undefined
})
describe('acp-agent stdout purity (no key required)', () => {
it('emits only framed JSON-RPC on stdout', async () => {
workdir = await mkdtemp(join(tmpdir(), 'acp-e2e-'))
// Collect raw stdout bytes directly (bypass the SDK framing) to inspect.
// A dummy key lets the deepseek adapter APPLY (it only checks the key is
// present at boot, not valid — the key is used only on a real model call,
// which this purity test never triggers). So this runs WITHOUT real creds.
const child = spawn(process.execPath, ['--import', tsxLoader, startScript], {
cwd: workdir,
env: { ...process.env, DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY ?? 'sk-dummy-for-boot' },
stdio: ['pipe', 'pipe', 'pipe'],
})
const out: string[] = []
child.stdout.setEncoding('utf8')
child.stdout.on('data', (c: string) => out.push(c))
// Send a single initialize request as a newline-delimited JSON-RPC frame.
const req = JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params: { protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} } })
child.stdin.write(req + '\n')
// Give it a moment to boot + reply, then inspect stdout.
await new Promise(r => setTimeout(r, 4000))
child.kill('SIGKILL')
const lines = out.join('').split('\n').filter(l => l.trim().length > 0)
expect(lines.length).toBeGreaterThan(0)
for (const line of lines) {
// Every stdout line MUST parse as JSON (a JSON-RPC frame). A non-JSON
// line means a logger/print leaked onto the protocol channel.
expect(() => JSON.parse(line) as unknown).not.toThrow()
}
}, 30_000)
})
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('acp-agent e2e: real prompt over ACP', () => {
it('runs a real turn and the agent writes the requested file (verified on disk)', async () => {
workdir = await mkdtemp(join(tmpdir(), 'acp-e2e-'))
spawned = spawnAcpAgent(workdir)
const { client, updates } = spawned
await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
// The MVP requires cwd === the server's launch dir (its cwd is `workdir`).
const { sessionId } = await client.newSession({ cwd: workdir, mcpServers: [] })
const res = await client.prompt({
sessionId,
prompt: [{ type: 'text', text: 'Use the bash tool to write the exact text ACP_OK into a file named proof.txt in the current directory. Then stop.' }],
})
expect(['end_turn', 'max_tokens']).toContain(res.stopReason)
// Verify the WORLD, not the agent's self-report: read the file from disk.
const proof = await readFile(join(workdir, 'proof.txt'), 'utf8')
expect(proof).toContain('ACP_OK')
// And the client saw tool-call activity stream through.
expect(updates.some(u => u.sessionUpdate === 'tool_call')).toBe(true)
}, 180_000)
})
+56
View File
@@ -0,0 +1,56 @@
# Shared provider/tool core for the example agents, loaded via a nested
# @cordisjs/plugin-include from each example's cordis.yml. Contains everything
# the model and tools need; each example adds its own infra (logger/timer/hmr),
# its agent-loop config (the examples disagree — see below), and its UI plugin.
#
# Deliberately EXCLUDES:
# - the console logger: it writes to stdout, which the acp-agent reserves for
# the JSON-RPC protocol (see packages/acp). Each example loads logging itself.
# - agent-loop: AgentLoop pre-creates its configured `agents` in its
# constructor, and the examples disagree — coding-agent needs a pre-created
# `main` (its stdio-chat calls ctx.agents.get('main')), while acp-agent must
# pre-create NONE (ACP session/new creates agents on demand). So each example
# declares agent-loop with its own `agents` list.
#
# Plugin entries here use package names (resolved from node_modules), so they
# are insensitive to the baseUrl reset that plugin-include performs per file.
# Requires DEEPSEEK_API_KEY (and optionally DEEPSEEK_BASE_URL) in the env.
- id: llm
name: '@deepseek-ai/dsh-llm'
- id: sessions
name: '@deepseek-ai/dsh-session'
- id: system-prompt
name: '@deepseek-ai/dsh-system-prompt'
- id: tools
name: '@deepseek-ai/dsh-tools'
- id: agents
name: '@deepseek-ai/dsh-agent'
# Dev-mode event-contract assertions + session-log freeze (off in prod).
- id: invariants
name: '@deepseek-ai/dsh-invariants'
# The DeepSeek adapter. Swap to '@deepseek-ai/dsh-llm-pi-ai' for the pi-ai-backed
# twin (same config shape; `reasoning: high` replaces thinking/reasoningEffort).
- id: llm-deepseek
name: '@deepseek-ai/dsh-llm-deepseek'
config:
apiKey: !!js process.env.DEEPSEEK_API_KEY
baseURL: !!js process.env.DEEPSEEK_BASE_URL
models:
- deepseek-v4-flash
- deepseek-v4-pro
# Bash execution: the local executor implementation + the tool schemas.
- id: bash
name: '@deepseek-ai/dsh-bash-local'
config:
timeoutMs: 60000
- id: tool-bash
name: '@deepseek-ai/dsh-tool-bash'
+10 -39
View File
@@ -1,5 +1,6 @@
# The coding-agent plugin tree, loaded via @cordisjs/plugin-include.
# Core services first, then the real adapters/tools, then the agent itself.
# Infra (logger/timer/hmr) first, then the shared provider/tool core (nested
# include of ../base.yml), then this example's agent-loop config + UI.
#
# Requires DEEPSEEK_API_KEY (and optionally DEEPSEEK_BASE_URL) in the
# environment — start.ts loads the gitignored repo-root .env first.
@@ -15,46 +16,16 @@
config:
root: ['.']
- id: llm
name: '@deepseek-ai/dsh-llm'
- id: sessions
name: '@deepseek-ai/dsh-session'
- id: system-prompt
name: '@deepseek-ai/dsh-system-prompt'
- id: tools
name: '@deepseek-ai/dsh-tools'
- id: agents
name: '@deepseek-ai/dsh-agent'
# Dev-mode event-contract assertions + session-log freeze (off in prod).
- id: invariants
name: '@deepseek-ai/dsh-invariants'
# The DeepSeek adapter. Swap to '@deepseek-ai/dsh-llm-pi-ai' for the
# pi-ai-backed twin (same config shape; `reasoning: high` replaces
# thinking/reasoningEffort).
- id: llm-deepseek
name: '@deepseek-ai/dsh-llm-deepseek'
# Shared provider/tool core (llm, sessions, system-prompt, tools, agents,
# invariants, llm-deepseek, bash-local, tool-bash). Nested include: the path is
# resolved relative to THIS file's directory.
- id: base
name: '@cordisjs/plugin-include'
config:
apiKey: !!js process.env.DEEPSEEK_API_KEY
baseURL: !!js process.env.DEEPSEEK_BASE_URL
models:
- deepseek-v4-flash
- deepseek-v4-pro
# Bash execution: the local executor implementation + the tool schemas.
- id: bash
name: '@deepseek-ai/dsh-bash-local'
config:
timeoutMs: 60000
- id: tool-bash
name: '@deepseek-ai/dsh-tool-bash'
path: '../base.yml'
# agent-loop is per-example (NOT in base.yml): coding-agent pre-creates a `main`
# agent its stdio-chat drives via ctx.agents.get('main').
- id: agent-loop
name: '@deepseek-ai/dsh-agent-loop'
config:
+8 -3
View File
@@ -4,11 +4,16 @@ import Loader from '@cordisjs/plugin-loader'
// Load DEEPSEEK_API_KEY / DEEPSEEK_BASE_URL from a gitignored repo-root .env
// (Node >= 21.7 native). Absent file is fine — the environment may already
// carry the variables; cordis.yml reads them via the `!!js` tag.
// carry the variables; cordis.yml reads them via the `!!js` tag. A
// present-but-unreadable/malformed .env is a real misconfiguration: surface it
// rather than silently running with the wrong environment.
try {
process.loadEnvFile(new URL('../../.env', import.meta.url).pathname)
} catch {
// no .env — rely on the ambient environment
} catch (error) {
if ((error as NodeJS.ErrnoException | null)?.code !== 'ENOENT') {
process.stderr.write(`coding-agent: failed to load .env: ${String(error)}\n`)
}
// ENOENT (no .env) is fine — rely on the ambient environment.
}
// Boot a Cordis app from this example's cordis.yml — the same shape as the
+5 -1
View File
@@ -4,7 +4,11 @@
"ignoreWorkspaces": ["vendor/*"],
"workspaces": {
".": {
"entry": ["examples/echo-agent/src/*.ts", "examples/coding-agent/src/*.ts"],
"entry": [
"examples/echo-agent/src/*.ts",
"examples/coding-agent/src/*.ts",
"examples/acp-agent/tests/**/*.e2e.ts"
],
"project": ["scripts/**/*.ts", "examples/**/*.ts"]
},
"packages/*": {
+2
View File
@@ -27,9 +27,11 @@
"hygiene": "yarn knip && yarn publint && yarn constraints",
"demo:echo": "node --expose-internals --import tsx examples/echo-agent/start.ts",
"demo:coding": "node --expose-internals --import tsx examples/coding-agent/start.ts",
"demo:acp": "node --expose-internals --import tsx examples/acp-agent/start.ts",
"postinstall": "lefthook install"
},
"devDependencies": {
"@agentclientprotocol/sdk": "0.25.1",
"@stylistic/eslint-plugin": "^5.10.0",
"@types/node": "^25.3.5",
"@vitest/coverage-v8": "^4.1.8",
+2
View File
@@ -17,6 +17,7 @@ dsh-llm-deepseek ← dsh-llm (DeepSeek adapter)
dsh-llm-pi-ai ← dsh-llm (pi-ai-backed adapter)
dsh-agent-loop ← dsh-llm, dsh-session, dsh-system-prompt, dsh-tools, dsh-agent
dsh-invariants ← dsh-llm, dsh-session, dsh-agent (dev-mode contract checks)
dsh-acp ← dsh-agent, dsh-llm, dsh-session, dsh-session-persistence (ACP JSON-RPC bridge)
```
The rule: plugins depend on interfaces, never on the concrete loop. `dsh-agent-loop` is swappable — UI/hook/tool plugins keep working against the `dsh-agent` vocabulary if the loop is replaced. A swappable capability splits into interface / implementation / consumer packages (the bash trio is the template — see [ADR 0009](../docs/adr/0009-capability-seams.md)).
@@ -37,6 +38,7 @@ The rule: plugins depend on interfaces, never on the concrete loop. `dsh-agent-l
| `llm-deepseek/` | DeepSeek API adapter (hand-rolled fetch/SSE) | (registers on `ctx.llm`) |
| `llm-pi-ai/` | DeepSeek adapter via `@earendil-works/pi-ai` (design twin) | (registers on `ctx.llm`) |
| `invariants/` | Dev-mode event-contract invariants + session-log freeze | (listens on `session/*`, `agent/*`) |
| `acp/` | Agent Client Protocol bridge: serves the agent to an ACP editor over JSON-RPC stdio | (drives `ctx.agents`/`ctx.sessions`) |
Each package has its own `README.md` with purpose, service API, events, extension points, and deliberate non-goals (TODOs).
+65
View File
@@ -0,0 +1,65 @@
# @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.
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`.
## Service / plugin
`apply(ctx, config)` — wires an `AgentSideConnection` (from `@agentclientprotocol/sdk`) to `process.stdin`/`process.stdout` and implements the ACP `Agent` method surface.
`inject: ['agents', 'sessions', 'sessionPersistence']` — programs against the interface packages only (never `dsh-agent-loop`). `sessionPersistence` is required because `initialize` advertises `loadSession: true`.
### Config
| Key | Default | Meaning |
|---|---|---|
| `model` | — | Model name for created agents (must have a registered adapter). |
| `systemPrompt` | — | Per-agent system prompt. |
| `agentName` | `deepseek-harness-acp` | Server name reported in `initialize`. |
| `agentVersion` | `0.0.1` | Server version reported in `initialize`. |
## ACP method mapping
| 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/update` | `session/event` | `agent_message_chunk` (text-delta), `agent_thought_chunk` (reasoning-delta), `user_message_chunk` (load replay), `tool_call`/`tool_call_update` |
## 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).
## 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<Agent, sessionId>` 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.
- **`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
The JSON-RPC frames go on stdout, so this plugin MUST run in an example that loads **no stdout logger** (the console logger writes to stdout and would corrupt the frames). The guarantee is config-only — see `examples/acp-agent` (no console logger) and RFC 010 § Risks. A stderr exporter is fine for logging.
## Running
`yarn demo:acp` boots `examples/acp-agent` (needs `DEEPSEEK_API_KEY`). Point an ACP client at it; for Zed, add to `agent_servers`:
```json
{
"agent_servers": {
"DeepSeek Harness": {
"command": "yarn",
"args": ["demo:acp"]
}
}
}
```
+45
View File
@@ -0,0 +1,45 @@
{
"name": "@deepseek-ai/dsh-acp",
"description": "Agent Client Protocol (ACP) bridge: drive the DeepSeek Harness coding agent from an ACP editor over JSON-RPC stdio",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/index.d.ts",
"exports": {
".": {
"types": "./lib/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib",
"src"
],
"license": "BSD-3-Clause",
"dependencies": {
"@agentclientprotocol/sdk": "0.25.1",
"schemastery": "^3.17.0",
"zod": "^4.0.0"
},
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-agent-loop": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
"@deepseek-ai/dsh-session-persistence-jsonl": "^0.0.1",
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
"@deepseek-ai/dsh-tools": "^0.0.1",
"cordis": "^4.0.0-rc.6"
}
}
+103
View File
@@ -0,0 +1,103 @@
/**
* Pure translation between harness vocabulary and ACP wire types. No I/O, no
* Cordis context — every function here is total and unit-testable in isolation.
* Keeping the mapping pure is deliberate: the SDK rejects an unknown
* `stopReason`, so the {@link turnEndToStopReason} total function (with its
* exhaustive test over every `TurnEndReason` kind) is the guard that a turn
* always settles to a legal wire value.
*
* @module @deepseek-ai/dsh-acp/codec
*/
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { TurnEndReason } from '@deepseek-ai/dsh-session'
import type { ContentBlock as AcpContentBlock, StopReason } from '@agentclientprotocol/sdk'
/**
* Map a harness {@link TurnEndReason} to the ACP `StopReason` wire enum.
*
* The mapping is total over the kinds the loop actually produces today
* (`completed`/`aborted`/`error`/`disposed`/`max-tokens`). `TurnEndReason` is
* merge-extensible, so an unknown future kind falls through to `end_turn` —
* the safest default (the turn DID end; we just lack a more specific wire
* reason) — rather than throwing into the SDK, which would reject an unknown
* `stopReason` and break the prompt RPC. When a new kind gains a dedicated ACP
* reason (e.g. a future `refusal` → `refusal`), add an explicit case here.
*
* - `completed` → `end_turn` (the model chose to stop)
* - `max-tokens` → `max_tokens` (cut off at the output-token ceiling)
* - `aborted` → `cancelled` (an `agent.abort()`, e.g. from `session/cancel`)
* - `error` → `end_turn` (defensive fallback only: the bridge REJECTS the
* `session/prompt` RPC on an error turn BEFORE calling this, so
* a client sees a JSON-RPC error, not a stop reason — see
* `rejectPrompt` in index.ts. This case keeps the function total
* for any non-bridge caller / property test.)
* - `disposed` → `cancelled` (the agent was torn down mid-turn — closest to a
* cancellation from the client's perspective)
*/
export function turnEndToStopReason(reason: TurnEndReason): StopReason {
switch (reason.kind) {
case 'completed':
return 'end_turn'
case 'max-tokens':
return 'max_tokens'
case 'aborted':
return 'cancelled'
case 'disposed':
return 'cancelled'
case 'error':
return 'end_turn'
// Merge-extensible: an unknown future TurnEndReason kind still has to
// produce a legal wire value (the SDK rejects unknown stopReason), so
// default to end_turn rather than assertNever. Add an explicit case when a
// new kind gains a dedicated ACP reason.
default:
return 'end_turn'
}
}
/**
* Translate a harness {@link ContentBlock} from a prompt into ACP content for
* replay, or `undefined` for block kinds the bridge does not surface to the
* client as message content. Today only `text` maps (text-only
* `promptCapabilities`); `reasoning` is surfaced via `agent_thought_chunk`
* streaming rather than as a message block, and `tool-call`/`tool-result`/
* `image` are handled by the tool-call update path or not advertised.
*/
export function harnessBlockToAcpContent(block: ContentBlock): AcpContentBlock | undefined {
switch (block.type) {
case 'text':
return { type: 'text', text: block.text }
// reasoning → streamed as agent_thought_chunk, not a message block
// tool-call / tool-result → the tool_call / tool_call_update path
// image → not advertised (text-only promptCapabilities)
default:
return undefined
}
}
/**
* Extract plain text from an ACP prompt's content blocks, concatenating every
* `text` block. Non-text blocks are ignored here; the caller rejects a prompt
* carrying image/audio per the advertised text-only capabilities BEFORE
* calling this, so dropping them here only affects `resource`/`resource_link`
* (which carry no inline text to forward in the MVP).
*/
export function acpPromptToText(prompt: readonly AcpContentBlock[]): string {
return prompt
.filter((block): block is AcpContentBlock & { type: 'text'; text: string } => block.type === 'text')
.map(block => block.text)
.join('')
}
/**
* Whether an ACP prompt contains any content the text-only bridge cannot
* accept — i.e. ANY non-`text` block (image, audio, `resource`, `resource_link`,
* …). The caller rejects such a prompt up front rather than silently dropping
* the unsupported parts: a prompt like `[text, resource_link]` carries context
* the model would otherwise never see, so running it text-only would be silent
* data loss. When richer block kinds are supported, narrow this.
*/
export function promptHasUnsupportedContent(prompt: readonly AcpContentBlock[]): boolean {
return prompt.some(block => block.type !== 'text')
}
+737
View File
@@ -0,0 +1,737 @@
/**
* The Agent Client Protocol (ACP) bridge: a client-driver / UI plugin that
* exposes the harness agent as an ACP server over JSON-RPC stdio, so editors
* (Zed and other ACP clients) can drive it. The structured analogue of the
* readline `stdio-chat` plugin.
*
* This is NOT a loop change and NOT an ADR-0009 capability seam: it consumes
* the existing `agent/*` event taxonomy, the `dsh-agent` create/resume factory,
* and `dsh-session-persistence` (for `session/load`). It maps:
*
* - `initialize` → protocol-version negotiation, text-only capabilities
* - `session/new` → `ctx.agents.create({ sessionId, meta:{cwd} })`
* - `session/load` → `ctx.agents.resume(...)` then replay the event log
* - `session/prompt` → `agent.send()`, settle on the owning turn's end (a turn
* 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.
*
* 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
* JSON-RPC frames). The guarantee is config-only — see the package README and
* RFC 010 § Risks.
*
* @module @deepseek-ai/dsh-acp
*/
import type { Context } from 'cordis'
import { Readable, Writable } from 'node:stream'
import { randomUUID } from 'node:crypto'
import { isAbsolute } from 'node:path'
import Schema from 'schemastery'
import {
AgentSideConnection,
ndJsonStream,
PROTOCOL_VERSION,
RequestError,
type Agent as AcpAgent,
type AuthenticateRequest,
type CancelNotification,
type ContentBlock as AcpContentBlock,
type InitializeRequest,
type InitializeResponse,
type LoadSessionRequest,
type LoadSessionResponse,
type NewSessionRequest,
type NewSessionResponse,
type PromptRequest,
type PromptResponse,
type SessionNotification,
type Stream,
type StopReason,
} from '@agentclientprotocol/sdk'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
// Side-effect type import: declaration-merges `ctx.sessionPersistence` onto
// Context (the bridge injects it and reads `list()` for load cwd validation).
import type {} from '@deepseek-ai/dsh-session-persistence'
import {
acpPromptToText,
harnessBlockToAcpContent,
promptHasUnsupportedContent,
turnEndToStopReason,
} from './codec.ts'
export const name = 'acp'
// The bridge programs against the interface packages only (architecture rule:
// plugins never depend on dsh-agent-loop). `sessionPersistence` is required
// because `initialize` advertises `loadSession: true`.
export const inject = ['agents', 'sessions', 'sessionPersistence']
/**
* Build an ACP "invalid params" error whose human detail rides in the message.
* `RequestError.invalidParams(data, additionalMessage)` keeps the standard
* "Invalid params" message and appends `additionalMessage`, so we pass the
* detail as `additionalMessage` (and no structured `data`).
*/
function invalidParams(detail: string): RequestError {
return RequestError.invalidParams(undefined, detail)
}
/**
* Build an ACP "internal error" whose human detail rides in the message. Used
* to reject a `session/prompt` whose turn ended in failure: a plain `Error`
* thrown from a method handler is flattened to a generic "Internal error" on
* the wire, so we wrap the detail in the SDK's `RequestError.internalError`
* (which appends `additionalMessage`) to surface *why* the turn failed.
*/
function internalError(detail: string): RequestError {
return RequestError.internalError(undefined, detail)
}
/** Plugin config: the agent template ACP sessions are created from. */
export interface AcpConfig {
/** Model name for created agents (must have a registered adapter). */
model?: string
/** Per-agent system prompt. */
systemPrompt?: string
/** Agent/server name reported to the client in `initialize`. */
agentName?: string
/** Agent/server version reported to the client in `initialize`. */
agentVersion?: string
/**
* Transport stream override. Production omits this (the plugin wires
* `process.stdin`/`process.stdout` via `ndJsonStream`). Tests inject an
* in-memory `Stream` (e.g. an `ndJsonStream` over a `Duplex` pair) to drive
* the bridge without a subprocess. Not part of the schemastery `Config` —
* it is a runtime-only seam, never set from a `cordis.yml`.
*/
stream?: Stream
}
export const Config: Schema<AcpConfig> = Schema.object({
model: Schema.string(),
systemPrompt: Schema.string(),
agentName: Schema.string().default('deepseek-harness-acp'),
agentVersion: Schema.string().default('0.0.1'),
})
/**
* 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.
*/
interface SessionRecord {
sessionId: string
agent: Agent
/**
* The in-flight `session/prompt`, or `undefined` when none is pending. A
* prompt resolves with a {@link StopReason} or rejects with an Error (a
* turn that ended in failure). Settled exactly once via {@link settlePrompt}.
*
* `turn` is the loop turn number this prompt owns, captured from the log's
* `turn/start` after `send()`. Until then it is `undefined` (the turn has not
* begun). Only a `turn/end` whose turn number equals `turn` settles the prompt
* — so a *previous* prompt's late `turn/end` (e.g. an aborted turn whose end
* arrives after the next prompt is already installed) can never settle the
* wrong prompt. A direct cancel/dispose settle clears the whole in-flight slot,
* so a later stale `turn/end` finds no pending prompt.
*
* `logWatermark` is the session log length at the moment the prompt was
* installed (before `send()`). The settle-from-log fallback uses it to infer
* the owning `turn/start` from the canonical log even when the live
* `session/event` capture was starved (a peer listener that throws on
* `turn/start` — see `settleFromLog`): the prompt owns the FIRST `turn/start`
* appended at or after this watermark.
*/
inflight: {
resolve: (reason: StopReason) => void
reject: (error: Error) => void
turn: number | undefined
logWatermark: number
} | undefined
}
/**
* Drive the in-flight prompt's settle from the harness event stream. A turn
* can end three ways the bridge must all handle (AGENTS.md "honor cross-seam
* contracts on BOTH sides"): the normal `agent/turn-end` event; a `turn/end`
* session event WITHOUT the agent event (a boundary emit threw inside the loop,
* which still appends `turn/end`); or the agent erroring/settling to idle. The
* first of these to fire settles the prompt; `settle` is then cleared so the
* others are no-ops (settle-exactly-once).
*/
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
// 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.
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<Agent, string>()
// Assigned at the bottom, before any agent event can fire (a session only
// exists after `newSession`, which the client calls after construction), so
// `notify` never observes it unset — no undefined guard needed.
let conn: AgentSideConnection
/**
* Reject any RPC after the bridge has torn down. The `AgentSideConnection`
* receive loop can outlive the plugin fiber — under an ACP-only HMR reload the
* `agents`/`agent-loop` services stay up while the bridge's `ctx.on` listeners
* and disposer are gone — so a late `session/new`/`load`/`prompt` could create
* or drive an agent the bridge can no longer stream or settle. Every
* state-affecting handler calls this first. (`initialize`/`authenticate` are
* pure/stateless and may answer harmlessly.)
*/
const assertOpen = (): void => {
if (closed) throw internalError('the ACP bridge has been disposed')
}
/** Resolve the live record for a sessionId, or throw an ACP error. */
const requireSession = (sessionId: string): SessionRecord => {
if (record === undefined || record.sessionId !== sessionId) {
throw invalidParams(`unknown session: ${sessionId}`)
}
return record
}
/** Push a `session/update` notification, swallowing post-close rejections. */
const notify = (notification: SessionNotification): void => {
// sessionUpdate returns a promise; a closed connection rejects it. The
// update is best-effort UI feed, never load-bearing for correctness, so a
// throwing/rejecting send must not break the turn (the chunk is emitted
// inside the model step — see AGENTS.md "contain callback exceptions").
/* v8 ignore next 3 -- the rejection only fires on a stdout/connection write
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)}`)
})
}
/** Settle the in-flight prompt with a stop reason, exactly once (no-op if none pending). */
const settlePrompt = (rec: SessionRecord, reason: StopReason): void => {
const inflight = rec.inflight
if (inflight === undefined) return
rec.inflight = undefined
inflight.resolve(reason)
}
// --- Stream the harness event taxonomy to ACP session/update --------------
// All content streaming AND the prompt settle flow through `session/event`,
// the canonical log: every assistant/chunk and tool/call/result is logged, so
// translating from the log makes live streaming and `session/load` replay
// share the identical path (streamSessionEventUpdate). Both the owning-turn
// capture and the settle key off the log's own `turn/start`/`turn/end` — NOT
// the `agent/turn-start`/`agent/turn-end` EVENTS, which a throwing PEER
// listener (cordis `emit` stops at the first throw) or a boundary-emit failure
// can skip. `closeTurn` appends `turn/end` to the log unconditionally, and
// `turn/start` is appended before any step runs, so within this one listener
// we always see the prompt's turn-start (tag `inflight.turn`) then its
// turn-end (settle). A `turn/end` settles the prompt ONLY when it is the
// prompt's OWN turn (`inflight.turn === event.data.turn`) — a previous,
// 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.
ctx.on('session/event', (session, event: SessionEvent) => {
const rec = record
if (rec === undefined || session.header.id !== rec.sessionId) return
streamSessionEventUpdate(rec.sessionId, event, notify)
const inflight = rec.inflight
if (inflight === undefined) return
if (event.type === 'turn/start') {
// Tag the in-flight prompt with its owning turn — but ONLY a
// `message`-triggered turn (the kind a `send()` prompt produces). A turn
// a plugin opens between prompt-install and the prompt's own turn (an idle
// `agent.inject()` writes a one-shot `injection`-triggered turn) must NOT
// be mistaken for the prompt's turn, or its turn/end would settle the RPC
// early. The first message turn at/after install owns the prompt
// (`turn === undefined` guard); the loop batches queued messages into one
// turn, so there is exactly one.
if (inflight.turn === undefined && event.data.trigger.kind === 'message') {
inflight.turn = event.data.turn
}
return
}
// Settle only on the OWNING turn's end.
if (event.type !== 'turn/end' || inflight.turn !== event.data.turn) return
rec.inflight = undefined
const reason = event.data.reason
if (reason.kind === 'error') {
inflight.reject(internalError(`turn failed: ${reason.message}`))
} else {
inflight.resolve(turnEndToStopReason(reason))
}
})
// Settle fallback: a `session/event` listener registered BEFORE ACP that
// throws (on `turn/start` OR `turn/end`) would, via cordis `emit`'s
// stop-on-throw, starve ACP's listener above — the prompt would hang or, if
// only the turn number was missed, settle as the wrong outcome. So when the
// agent settles to `idle` (or is disposed), reconcile against the canonical
// log: determine the prompt's owning turn (the captured `turn`, or — if the
// live capture was starved — the FIRST `turn/start` appended at/after the
// install-time `logWatermark`), then settle from that turn's `turn/end`
// (reject on error, resolve via codec), or `cancelled` if no owning turn ever
// started. Never double-settles — clears `inflight` first.
const settleFromLog = (rec: SessionRecord): void => {
const inflight = rec.inflight
if (inflight === undefined) return
const events = rec.agent.session.events
// The owning turn number: the captured one, or — if the live capture was
// starved — inferred from the log as the first MESSAGE-triggered turn opened
// at/after the watermark. The message-trigger filter matches the live
// capture: a one-shot `injection` turn a plugin may open between
// prompt-install and the prompt's turn is NOT the prompt's turn. Undefined
// only if no message turn ever started for this prompt.
const owningTurn = inflight.turn ?? events.slice(inflight.logWatermark).find(
(e): e is Extract<SessionEvent, { type: 'turn/start' }> =>
e.type === 'turn/start' && e.data.trigger.kind === 'message',
)?.data.turn
// The owning turn's end in the log. If `owningTurn` is undefined (no turn
// ever started for this prompt — a torn-down-before-turn case that quiesce's
// direct settle normally pre-empts), no `turn/end` matches (turn numbers are
// >= 1) and `findLast` returns undefined, falling through to cancelled.
const end = events.findLast(
(e): e is Extract<SessionEvent, { type: 'turn/end' }> =>
e.type === 'turn/end' && e.data.turn === owningTurn,
)
rec.inflight = undefined
if (end === undefined) {
// No owning turn / no clean turn/end (torn down mid-turn) → cancelled.
inflight.resolve('cancelled')
return
}
const reason = end.data.reason
if (reason.kind === 'error') {
inflight.reject(internalError(`turn failed: ${reason.message}`))
} else {
inflight.resolve(turnEndToStopReason(reason))
}
}
// 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`.
ctx.on('agent/status', (agent, status: AgentStatus) => {
const rec = record
if (rec === undefined || owned.get(agent) !== rec.sessionId) return
if (status === 'idle' || status === 'disposed') settleFromLog(rec)
})
// --- The ACP Agent method surface -----------------------------------------
const makeAgent = (connection: AgentSideConnection): AcpAgent => {
conn = connection
return {
initialize(params: InitializeRequest): Promise<InitializeResponse> {
// Echo the client's version if we support it, else our own. We support
// exactly PROTOCOL_VERSION; any other requested version negotiates
// down to ours (the client disconnects if it can't speak it).
const protocolVersion = params.protocolVersion === PROTOCOL_VERSION ? params.protocolVersion : PROTOCOL_VERSION
return Promise.resolve({
protocolVersion,
agentInfo: { name: agentName, version: agentVersion },
agentCapabilities: {
loadSession: true,
// text-only: no image/audio/embeddedContext, no mcpCapabilities
promptCapabilities: { image: false, audio: false, embeddedContext: false },
},
authMethods: [],
})
},
authenticate(_params: AuthenticateRequest): Promise<void> {
// No auth methods advertised; nothing to do. Present because the SDK
// Agent interface requires it.
return Promise.resolve()
},
newSession(params: NewSessionRequest): Promise<NewSessionResponse> {
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({
agentId: sessionId,
sessionId,
meta: { cwd: params.cwd },
agentOptions: agentOptions(config),
})
owned.set(agent, sessionId)
record = { sessionId, agent, inflight: undefined }
return Promise.resolve({ sessionId })
},
async loadSession(params: LoadSessionRequest): Promise<LoadSessionResponse> {
assertOpen()
if (record !== undefined || loading) {
throw invalidParams('this agent supports a single session; a session already exists (RFC 011 will lift this)')
}
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
try {
// Validate the PERSISTED cwd BEFORE resuming — `list()` is a
// metadata-only read (no full-log parse) — so a mismatch rejects
// without ever constructing/registering a live agent (which would
// then leak in `ctx.agents`/`ctx.sessions` with no disposer here).
// A session persisted in workspace A must not be loaded by a server
// 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)
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({
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 `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
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) {
throw invalidParams('connection closed during session/load')
}
owned.set(agent, params.sessionId)
record = { 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
// turns — user prompts (user/message → user_message_chunk), assistant
// text and reasoning (assistant/chunk), and tool calls/results.
for (const event of agent.session.events) {
streamSessionEventUpdate(params.sessionId, event, notify)
}
return {}
} finally {
loading = false
}
},
async prompt(params: PromptRequest): Promise<PromptResponse> {
assertOpen()
const rec = requireSession(params.sessionId)
if (rec.inflight !== undefined) {
throw invalidParams('a prompt is already in flight for this session')
}
if (promptHasUnsupportedContent(params.prompt)) {
throw invalidParams('only text prompt content is supported (text-only promptCapabilities); image/audio/resource blocks are rejected rather than silently dropped')
}
const text = acpPromptToText(params.prompt)
if (text.trim().length === 0) {
// Reject up front rather than calling send(): an empty prompt would
// queue no work, no turn would start, and the RPC would hang forever
// waiting for a settle that never comes.
throw invalidParams('empty prompt')
}
// Install the in-flight slot BEFORE send() (send does not synchronously
// flip status to running; the session/event listener records the turn
// number and settle/rejects it). Capture the log length now as the
// watermark: the settle-from-log fallback infers the owning turn/start
// as the first one appended at/after it, surviving a starved live
// capture. A turn that ends in error rejects this promise (the codec
// never produces an error stop reason).
const stopReason = await new Promise<StopReason>((resolve, reject) => {
rec.inflight = { resolve, reject, turn: undefined, logWatermark: rec.agent.session.events.length }
rec.agent.send([{ type: 'text', text }])
})
return { stopReason }
},
cancel(params: CancelNotification): Promise<void> {
const rec = record
if (rec === undefined || rec.sessionId !== params.sessionId) 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
// 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
// gap: because the loop batches all queued messages into one turn, a
// prompt accepted right after a pre-step cancel can be merged into the
// same turn as the cancelled one — that turn then carries both prompts'
// text and the new prompt settles for it. Both are closed by the same
// queue-aware loop cancel; the single-in-flight rule bounds the blast
// radius to one extra prompt.
rec.agent.abort('session/cancel')
settlePrompt(rec, 'cancelled')
return Promise.resolve()
},
}
}
// --- Connection lifecycle --------------------------------------------------
// The transport stream. Production wires stdio (stdout carries the protocol);
// tests inject an in-memory pipe pair via config.stream to drive the bridge
// without a subprocess. ndJsonStream is the SDK's stdio framing helper. The
// AgentSideConnection constructor synchronously invokes makeAgent (assigning
// the outer `conn`), so `conn` is set before any agent method runs.
/* v8 ignore next 4 -- production stdio wiring; tests always inject config.stream */
const stream: Stream = config.stream ?? ndJsonStream(
Writable.toWeb(process.stdout) as WritableStream<Uint8Array>,
Readable.toWeb(process.stdin) as ReadableStream<Uint8Array>,
)
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.
* 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, 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.
*
* 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 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)).
*/
let quiescing: Promise<void> | undefined
const quiesce = (): Promise<void> => {
// 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()).
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.
closed = true
const rec = record
record = undefined
if (rec === undefined) return Promise.resolve()
quiescing = (async () => {
settlePrompt(rec, 'cancelled')
rec.agent.abort('disposed')
await rec.agent.whenIdle()
})()
return quiescing
}
// Client disconnect: when the ACP transport closes (editor quits, pipe EOF),
// the in-flight turn would otherwise keep running and its `session/update`
// writes would be silently swallowed by `notify()`. Tear the session down so
// a vanished client does not leave an orphaned running agent. `conn.closed`
// rejects/resolves once; contain any teardown throw (nothing else can act on
// it — the connection is already gone). The Cordis disposer below still runs
// on normal shutdown and is idempotent with this.
/* v8 ignore start -- the .catch arrow is a defensive guard: conn.closed
settling rejected or quiesce() throwing on an already-closed connection is
not reproducible through the in-memory test transport (it never severs
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)}`)
})
/* v8 ignore stop */
ctx.effect(() => quiesce, 'acp.connection')
}
/**
* Build per-agent options from the plugin config, omitting absent fields
* (exactOptionalPropertyTypes: never assign `undefined` to an optional key).
* Exported for unit coverage of both the present and absent branches.
*/
export function agentOptions(config: AcpConfig): { model?: string; systemPrompt?: string } {
return {
...config.model !== undefined ? { model: config.model } : {},
...config.systemPrompt !== undefined ? { systemPrompt: config.systemPrompt } : {},
}
}
/**
* Validate `session/new` params per the MVP contract: `cwd` absolute AND equal
* to the server's launch directory (there is no path from session cwd to the
* bash workdir yet — RFC 010 § Deferred — so the server must be launched in the
* workspace root, and we error loudly rather than silently run tools in the
* wrong directory); `additionalDirectories` empty (we cannot widen filesystem
* scope yet, and silently ignoring them would desync the client's scope UI).
*/
/**
* Validate the MVP `cwd`/`additionalDirectories` contract shared by
* `session/new` and `session/load`: `cwd` must be absolute AND equal the
* server's launch directory (there is no path from session cwd to the bash
* workdir yet — RFC 010 § Deferred — so the server must be launched in the
* workspace root, and we error loudly rather than silently run tools in the
* wrong directory); `additionalDirectories` must be empty (we cannot widen
* filesystem scope yet, and silently ignoring it would desync the client's
* scope UI). Both request shapes carry `cwd: string` and
* `additionalDirectories?: string[]`, so one validator covers both.
*/
function validateWorkspaceParams(params: { cwd: string; additionalDirectories?: string[] }): void {
if (!isAbsolute(params.cwd)) {
throw invalidParams(`cwd must be an absolute path: ${params.cwd}`)
}
if (params.cwd !== process.cwd()) {
throw invalidParams(
`cwd must equal the server's launch directory (${process.cwd()}); honoring an arbitrary cwd is not yet supported — launch the server in the workspace root`,
)
}
if (params.additionalDirectories !== undefined && params.additionalDirectories.length > 0) {
throw invalidParams('additionalDirectories is not supported in this MVP')
}
}
/**
* Translate a single harness {@link SessionEvent} into the `session/update`
* notification(s) it produces, pushing each via `notify`. Shared by live
* streaming (`session/event`) and `session/load` replay so both paths emit an
* identical update stream from the same event log.
*
* - `assistant/chunk` text-delta/reasoning-delta → message/thought chunks
* - `user/message` → `user_message_chunk` (text blocks) — so a `session/load`
* replay reconstructs the USER side of each turn, not just the agent's
* - `tool/call` → `tool_call` (pending)
* - `tool/result` → `tool_call_update` (completed/failed)
*
* Other event types (turn/step boundaries, context/message, usage, …) produce
* no client update.
*/
export function streamSessionEventUpdate(
sessionId: string,
event: SessionEvent,
notify: (notification: SessionNotification) => void,
): void {
switch (event.type) {
case 'assistant/chunk': {
const chunk = event.data.chunk
if (chunk.type === 'text-delta') {
notify({ sessionId, update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: chunk.text } } })
} else if (chunk.type === 'reasoning-delta') {
notify({ sessionId, update: { sessionUpdate: 'agent_thought_chunk', content: { type: 'text', text: chunk.text } } })
}
return
}
case 'user/message': {
// Replay the user's prompt so a loaded session shows both sides of each
// turn. Only text blocks carry inline content the bridge surfaces (the
// prompt path is text-only); other block kinds produce no chunk.
for (const block of event.data.content) {
const content = harnessBlockToAcpContent(block)
if (content !== undefined) {
notify({ sessionId, update: { sessionUpdate: 'user_message_chunk', content } })
}
}
return
}
case 'tool/call': {
notify({
sessionId,
update: {
sessionUpdate: 'tool_call',
toolCallId: event.data.callId,
title: event.data.name,
kind: toolKindFor(event.data.name),
status: 'in_progress',
rawInput: parseToolArguments(event.data.arguments),
},
})
return
}
case 'tool/result': {
notify({
sessionId,
update: {
sessionUpdate: 'tool_call_update',
toolCallId: event.data.callId,
status: event.data.isError ? 'failed' : 'completed',
content: toolResultContent(event.data.content),
},
})
return
}
// turn/step boundaries, context/message, steering, usage, error,
// assistant/message — no direct ACP client update.
default:
return
}
}
/** Map a harness tool name to an ACP ToolKind (best-effort; default `other`). */
function toolKindFor(name: string): 'read' | 'edit' | 'delete' | 'move' | 'search' | 'execute' | 'fetch' | 'other' {
if (name === 'bash' || name === 'bash_output' || name === 'bash_kill') return 'execute'
if (name === 'read' || name.startsWith('read')) return 'read'
if (name === 'write' || name === 'edit' || name.startsWith('edit')) return 'edit'
return 'other'
}
/** Parse a tool-call arguments JSON string for `rawInput`; raw string on failure. */
function parseToolArguments(args: string): unknown {
try {
return args ? JSON.parse(args) : {}
} catch {
// The model produced non-JSON arguments; surface the raw string rather
// than dropping it. (The harness tool layer handles validation; here we
// only feed the client's tool-call UI.)
return args
}
}
/** Map harness tool-result content blocks to ACP tool-call content (text only). */
function toolResultContent(blocks: ContentBlock[]): { type: 'content'; content: AcpContentBlock }[] {
const out: { type: 'content'; content: AcpContentBlock }[] = []
for (const block of blocks) {
const content = harnessBlockToAcpContent(block)
if (content !== undefined) out.push({ type: 'content', content })
}
return out
}
export default apply
+152
View File
@@ -0,0 +1,152 @@
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 } from './harness.ts'
/**
* End-to-end bridge specs over an in-memory transport: a real
* ClientSideConnection drives the bridge's AgentSideConnection, so every
* assertion exercises actual JSON-RPC framing and the harness event taxonomy.
*/
describe('acp bridge', () => {
let storageDir: string
let harness: BridgeHarness | undefined
beforeEach(async () => {
storageDir = await mkdtemp(join(tmpdir(), 'acp-test-'))
})
afterEach(async () => {
// e2e/integration tests own their resources (AGENTS.md): dispose even on
// failure so a flaky run never leaks a context or persistence dir.
if (harness) await harness.dispose()
harness = undefined
await rm(storageDir, { recursive: true, force: true })
})
it('initialize negotiates the protocol version and advertises capabilities', async () => {
harness = await makeBridgeHarness({ storageDir })
const res = await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
expect(res.protocolVersion).toBe(PROTOCOL_VERSION)
expect(res.agentCapabilities?.loadSession).toBe(true)
expect(res.agentCapabilities?.promptCapabilities).toMatchObject({ image: false, audio: false })
expect(res.agentInfo?.name).toBe('deepseek-harness-acp')
})
it('session/new creates a session and a full prompt turn streams text then settles end_turn', async () => {
harness = await makeBridgeHarness({ storageDir, script: [textResponse('hello there')] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
expect(sessionId).toBeTruthy()
const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'hi' }] })
expect(res.stopReason).toBe('end_turn')
// The streamed text arrived as agent_message_chunk updates.
const text = harness.updates
.filter(u => u.sessionUpdate === 'agent_message_chunk')
.map(u => (u.content.type === 'text' ? u.content.text : ''))
.join('')
expect(text).toBe('hello there')
})
it('rejects a second session/new (single-session MVP)', 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/)
})
it('rejects a non-absolute cwd and a cwd that differs from the launch dir', async () => {
harness = await makeBridgeHarness({ storageDir })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await expect(harness.client.newSession({ cwd: 'relative/path', mcpServers: [] }))
.rejects.toThrow(/absolute/)
await expect(harness.client.newSession({ cwd: '/some/other/dir', mcpServers: [] }))
.rejects.toThrow(/launch directory/)
})
it('rejects non-empty additionalDirectories', async () => {
harness = await makeBridgeHarness({ storageDir })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await expect(harness.client.newSession({ cwd: process.cwd(), mcpServers: [], additionalDirectories: ['/x'] }))
.rejects.toThrow(/additionalDirectories/)
})
it('rejects an empty prompt without hanging', async () => {
harness = await makeBridgeHarness({ storageDir, script: [] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: ' ' }] }))
.rejects.toThrow(/empty prompt/)
})
it('rejects image content in a prompt (text-only capabilities)', async () => {
harness = await makeBridgeHarness({ storageDir, script: [] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
await expect(harness.client.prompt({
sessionId,
prompt: [{ type: 'image', mimeType: 'image/png', data: 'AA==' }],
})).rejects.toThrow(/text/)
})
it('rejects a prompt carrying a non-text block alongside text (no silent context loss)', async () => {
// A text + resource_link prompt must be rejected, not run text-only with the
// resource silently dropped — that would feed the model an incomplete prompt.
harness = await makeBridgeHarness({ storageDir, script: [] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
await expect(harness.client.prompt({
sessionId,
prompt: [
{ type: 'text', text: 'fix the bug in' },
{ type: 'resource_link', uri: 'file:///x.ts', name: 'x.ts' },
],
})).rejects.toThrow(/text/)
})
it('rejects a prompt for an unknown session', async () => {
harness = await makeBridgeHarness({ storageDir })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await expect(harness.client.prompt({ sessionId: 'nope', prompt: [{ type: 'text', text: 'hi' }] }))
.rejects.toThrow(/unknown session/)
})
it('negotiates an unsupported protocol version down to the supported one', async () => {
harness = await makeBridgeHarness({ storageDir })
const res = await harness.client.initialize({ protocolVersion: 999, clientCapabilities: {} })
expect(res.protocolVersion).toBe(PROTOCOL_VERSION)
})
it('a cancel for an unknown/absent session is a silent no-op', async () => {
harness = await makeBridgeHarness({ storageDir })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
// No session created yet — cancel must not throw.
await expect(harness.client.cancel({ sessionId: 'nope' })).resolves.toBeUndefined()
})
it('authenticate is a no-op (no auth methods advertised)', async () => {
harness = await makeBridgeHarness({ storageDir })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await expect(harness.client.authenticate({ methodId: 'whatever' })).resolves.toBeDefined()
})
it('honors agentName/agentVersion/systemPrompt config', async () => {
harness = await makeBridgeHarness({
storageDir,
script: [textResponse('ok')],
config: { agentName: 'custom-agent', agentVersion: '9.9.9', systemPrompt: 'be terse' },
})
const res = await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
expect(res.agentInfo).toMatchObject({ name: 'custom-agent', version: '9.9.9' })
// Create + prompt so the systemPrompt config flows through agentOptions and
// reaches the model request.
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'hi' }] })
expect(harness.adapter.requests[0]?.system).toContain('be terse')
})
})
+65
View File
@@ -0,0 +1,65 @@
import { describe, expect, it } from 'vitest'
import type { TurnEndReason } from '@deepseek-ai/dsh-session'
import type { ContentBlock as AcpContentBlock } from '@agentclientprotocol/sdk'
import {
acpPromptToText,
harnessBlockToAcpContent,
promptHasUnsupportedContent,
turnEndToStopReason,
} from '../src/codec.ts'
describe('turnEndToStopReason', () => {
// The SDK rejects an unknown stopReason, so this must be total over every
// TurnEndReason kind and always produce a legal wire value.
it('maps every known TurnEndReason kind to a legal StopReason', () => {
expect(turnEndToStopReason({ kind: 'completed' })).toBe('end_turn')
expect(turnEndToStopReason({ kind: 'max-tokens' })).toBe('max_tokens')
expect(turnEndToStopReason({ kind: 'aborted', reason: 'x' })).toBe('cancelled')
expect(turnEndToStopReason({ kind: 'disposed' })).toBe('cancelled')
expect(turnEndToStopReason({ kind: 'error', message: 'boom' })).toBe('end_turn')
})
it('falls back to end_turn for an unknown (merge-extensible) future kind', () => {
// A plugin-added TurnEndReason variant the bridge does not yet know about
// must still produce a legal wire value, not throw into the SDK.
const future = { kind: 'refusal' } as unknown as TurnEndReason
expect(turnEndToStopReason(future)).toBe('end_turn')
})
})
describe('harnessBlockToAcpContent', () => {
it('maps a text block to ACP text content', () => {
expect(harnessBlockToAcpContent({ type: 'text', text: 'hi' })).toEqual({ type: 'text', text: 'hi' })
})
it('returns undefined for non-text blocks (reasoning/tool/image)', () => {
expect(harnessBlockToAcpContent({ type: 'reasoning', text: 'think' })).toBeUndefined()
expect(harnessBlockToAcpContent({ type: 'image', url: 'https://x/y.png', mimeType: 'image/png' })).toBeUndefined()
})
})
describe('acpPromptToText', () => {
it('concatenates text blocks and ignores non-text', () => {
const prompt: AcpContentBlock[] = [
{ type: 'text', text: 'hello ' },
{ type: 'resource_link', uri: 'file:///x', name: 'x' },
{ type: 'text', text: 'world' },
]
expect(acpPromptToText(prompt)).toBe('hello world')
})
it('returns empty string for a prompt with no text blocks', () => {
expect(acpPromptToText([{ type: 'resource_link', uri: 'file:///x', name: 'x' }])).toBe('')
})
})
describe('promptHasUnsupportedContent', () => {
it('detects image and audio blocks', () => {
expect(promptHasUnsupportedContent([{ type: 'image', mimeType: 'image/png', data: 'AA==' }])).toBe(true)
expect(promptHasUnsupportedContent([{ type: 'audio', mimeType: 'audio/wav', data: 'AA==' }])).toBe(true)
})
it('passes a text-only prompt', () => {
expect(promptHasUnsupportedContent([{ type: 'text', text: 'hi' }])).toBe(false)
})
})
+124
View File
@@ -0,0 +1,124 @@
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 } from './harness.ts'
describe('acp bridge — disposal & HMR safety', () => {
let storageDir: string
beforeEach(async () => { storageDir = await mkdtemp(join(tmpdir(), 'acp-dispose-')) })
afterEach(async () => { await rm(storageDir, { recursive: true, force: true }) })
it('disposal reaches quiescence: a running turn is aborted and awaited before dispose returns', async () => {
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: [] })
const agent = harness.ctx.agents.get(sessionId)!
// Start a prompt that hangs in the model stream.
const promptDone = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
await new Promise(r => setTimeout(r, 30))
expect(agent.status).toBe('running')
// Dispose the whole context. The bridge's teardown must abort the agent and
// AWAIT whenIdle() — so right after dispose resolves, the agent is settled
// (not still running). Proves disposal waited, not just requested.
await harness.ctx.fiber.dispose()
expect(agent.status).not.toBe('running')
// The in-flight prompt settled (cancelled) rather than hanging forever.
const res = await promptDone
expect(res.stopReason).toBe('cancelled')
})
it('after an ACP-only HMR dispose, a late session/new creates no orphan agent (closed guard)', async () => {
// Dispose JUST the bridge's fiber (an HMR reload) while agents/agent-loop
// 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 })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const before = harness.ctx.agents.list().length
await harness.acpFiber.dispose() // tear down ONLY the bridge
await expect(harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }))
.rejects.toThrow(/disposed/)
expect(harness.ctx.agents.list().length).toBe(before)
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
// drive/settle. The transport is gone so the RPC rejects; assert the world:
// no new agent appeared in the registry.
const harness = await makeBridgeHarness({ storageDir, script: [] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const before = harness.ctx.agents.list().length
await harness.closeClientTransport() // teardown → closed = true
await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }).catch(() => {})
await new Promise(r => setTimeout(r, 10))
expect(harness.ctx.agents.list().length).toBe(before)
await harness.dispose()
})
it('a client disconnect mid-prompt tears the session down to quiescence', 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.
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: [] })
const agent = harness.ctx.agents.get(sessionId)!
// Start a prompt that hangs in the model stream. The prompt RPC will never
// return (its transport is severed), so do not await it.
void harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }).catch(() => {})
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).
await harness.closeClientTransport()
await agent.whenIdle()
expect(agent.status).toBe('idle')
await harness.dispose() // idempotent with the close teardown
})
it('a client disconnect racing fiber dispose both reach quiescence (shared teardown)', async () => {
// conn.closed teardown and ctx.fiber.dispose() can fire near-simultaneously.
// They must share one teardown promise: dispose() must NOT return before the
// disconnect teardown's whenIdle() has settled (a `record === undefined`-only
// guard would let the second caller return early mid-teardown).
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: [] })
const agent = harness.ctx.agents.get(sessionId)!
void harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }).catch(() => {})
await new Promise(r => setTimeout(r, 30))
expect(agent.status).toBe('running')
// Fire both teardown paths without awaiting the first, then await both.
const close = harness.closeClientTransport()
const dispose = harness.ctx.fiber.dispose()
await Promise.all([close, dispose])
// After BOTH settle, the agent has fully drained (not still running).
expect(agent.status).not.toBe('running')
})
it('after dispose, session/update listeners are gone (no further updates emitted)', async () => {
const harness = await makeBridgeHarness({ storageDir, script: [] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
const session = harness.ctx.agents.get(sessionId)!.session
await harness.ctx.fiber.dispose()
const before = harness.updates.length
// Append an event directly to the (now-detached) session: the bridge's
// session/event listener should have been disposed, so no update fires.
session.append('turn/start', { turn: 99, trigger: { kind: 'message', source: { kind: 'user' } } })
await new Promise(r => setTimeout(r, 10))
expect(harness.updates.length).toBe(before)
})
})
+55
View File
@@ -0,0 +1,55 @@
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 } from './harness.ts'
describe('acp bridge — demux & config edges', () => {
let storageDir: string
let harness: BridgeHarness | undefined
beforeEach(async () => { storageDir = await mkdtemp(join(tmpdir(), 'acp-edge-')) })
afterEach(async () => {
if (harness) await harness.dispose()
harness = undefined
await rm(storageDir, { recursive: true, force: true })
})
it('ignores events from an agent the bridge does not own (strict id demux)', async () => {
// A second agent created directly on the registry (NOT via the bridge) runs
// a turn. Its session/event + agent/status must NOT produce ACP updates and
// must not settle anything — the bridge demuxes strictly by its own id.
harness = await makeBridgeHarness({ storageDir, script: [textResponse('foreign')] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
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' } })
foreign.send([{ type: 'text', text: 'hi' }])
await foreign.whenIdle()
await new Promise(r => setTimeout(r, 10))
// No update was emitted for the foreign agent's stream.
expect(harness.updates.length).toBe(before)
})
it('survives a session/update that the client rejects (best-effort notify)', async () => {
harness = await makeBridgeHarness({ storageDir, script: [textResponse('ok')] })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
// Make the client reject every update — the bridge's notify() must swallow
// the rejection and the prompt must still settle normally.
harness.onSessionUpdateError = () => { throw new Error('client update rejected') }
const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
expect(res.stopReason).toBe('end_turn')
})
it('accepts session/new with additionalDirectories empty', async () => {
// Exercises the defined-but-empty additionalDirectories branch (length 0 → allowed).
harness = await makeBridgeHarness({ storageDir })
await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const a = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [], additionalDirectories: [] })
expect(a.sessionId).toBeTruthy()
})
})
+241
View File
@@ -0,0 +1,241 @@
/**
* Shared test fixtures for the ACP bridge specs. A plain module (NOT a
* *.spec.ts) so importing it does not re-register a describe block.
*
* `makeBridgeHarness` builds a full in-memory cordis context (llm + session +
* system-prompt + tools + agents + agent-loop + persistence) with the ACP
* bridge wired to an in-memory transport, plus a `ClientSideConnection` on the
* other end — so a test drives the bridge exactly as an editor would, with no
* subprocess and no real stdio.
*/
import { Context } from 'cordis'
import LlmService, { CallId, type GenerateOptions, type StreamChunk } from '@deepseek-ai/dsh-llm'
import { LlmAdapter } from '@deepseek-ai/dsh-llm'
import SessionStore from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
import {
ClientSideConnection,
ndJsonStream,
type Agent as AcpAgent,
type Client,
type RequestPermissionRequest,
type RequestPermissionResponse,
type SessionNotification,
type Stream,
} from '@agentclientprotocol/sdk'
import * as AcpPlugin from '../src/index.ts'
import { type AcpConfig } from '../src/index.ts'
/** A scripted mock adapter (mirrors the agent-loop test adapter). */
class MockAdapter extends LlmAdapter {
requests: GenerateOptions[] = []
constructor(private script: (StreamChunk[] | 'hang')[]) {
super()
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
this.requests.push(options)
const entry = this.script.shift()
if (!entry) throw new Error('MockAdapter: script exhausted')
if (entry === 'hang') {
yield { type: 'block-start', index: 0, blockType: 'text' }
yield { type: 'text-delta', index: 0, text: 'partial' }
await new Promise<void>((_resolve, reject) => {
if (options.signal?.aborted) { reject(new Error('aborted')); return }
options.signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
})
return
}
for (const chunk of entry) {
if (options.signal?.aborted) throw new Error('aborted')
yield chunk
}
}
}
/** Scripted text response ending in a clean `stop` finish. */
export function textResponse(text: string): StreamChunk[] {
return [
{ type: 'block-start', index: 0, blockType: 'text' },
...Array.from(text, (char): StreamChunk => ({ type: 'text-delta', index: 0, text: char })),
{ type: 'block-end', index: 0, block: { type: 'text', text } },
{ type: 'usage', usage: { inputTokens: 5, outputTokens: text.length } },
{ type: 'finish', reason: { kind: 'stop' } },
]
}
/** Scripted response ending at the output-token ceiling (max-tokens finish). */
export function maxTokensResponse(text: string): StreamChunk[] {
return [
{ type: 'block-start', index: 0, blockType: 'text' },
...Array.from(text, (char): StreamChunk => ({ type: 'text-delta', index: 0, text: char })),
{ type: 'block-end', index: 0, block: { type: 'text', text } },
{ type: 'finish', reason: { kind: 'max-tokens' } },
]
}
/** Scripted response that fails mid-turn with a finish-error chunk. */
export function errorResponse(message: string): StreamChunk[] {
return [
{ type: 'block-start', index: 0, blockType: 'text' },
{ type: 'text-delta', index: 0, text: 'partial' },
{ type: 'finish', reason: { kind: 'error', message, code: 'PROVIDER_ERROR' } },
]
}
/** Scripted single tool call (no follow-up step scripted by default). */
export function toolCallResponse(rawCallId: string, name: string, args: object): StreamChunk[] {
const argumentsJson = JSON.stringify(args)
const id = CallId(rawCallId)
return [
{ type: 'block-start', index: 0, blockType: 'tool-call' },
{ type: 'tool-call-delta', index: 0, id, name, argumentsDelta: argumentsJson },
{ type: 'block-end', index: 0, block: { type: 'tool-call', id, name, arguments: argumentsJson } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
]
}
/** A captured `session/update` notification (the update payload only). */
export type CapturedUpdate = SessionNotification['update']
export interface BridgeHarness {
ctx: Context
client: ClientSideConnection
adapter: MockAdapter
/** Every `session/update` the bridge pushed, in order. */
updates: CapturedUpdate[]
/** Permission requests the bridge issued (none until the gate lands). */
permissionRequests: RequestPermissionRequest[]
/** Decide each permission request's outcome (default: cancelled). */
onPermission: (req: RequestPermissionRequest) => RequestPermissionResponse
/** If set, the client's sessionUpdate throws this (tests notify error path). */
onSessionUpdateError: (() => void) | undefined
/**
* Sever the client→agent transport (close the writable the agent reads),
* which ends the agent-side stream and resolves the bridge's `conn.closed` —
* simulating an editor disconnecting. Returns once the close is requested.
*/
closeClientTransport: () => Promise<void>
/**
* The child fiber the ACP bridge is mounted in. Disposing it tears down JUST
* the bridge (its `ctx.on` listeners + effect) while the rest of the harness
* stays up — an ACP-only HMR reload.
*/
acpFiber: Awaited<ReturnType<Context['plugin']>>
dispose: () => Promise<void>
storageDir: string
}
/**
* Build the bridge + a connected client over an in-memory transport pair.
*
* Two identity `TransformStream`s cross-wired (agent writes → client reads,
* client writes → agent reads) give a faithful bidirectional JSON-RPC channel.
* The bridge's `apply` receives the agent-side `Stream` via `config.stream`;
* the test holds the `ClientSideConnection`.
*
* Pass `config: { model: undefined }` to override the default `model: 'mock'`
* (the model key is dropped entirely when explicitly undefined).
*/
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 ?? [])
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SessionPersistenceJsonl, { root: options.storageDir })
ctx.llm.registerAdapter(['mock'], adapter)
// Two identity byte pipes cross-wired into the two ndJsonStreams: bytes the
// agent writes flow to the client's reader and vice versa. (ndJsonStream
// takes (output, input): the agent writes to a2c and reads from c2a; the
// client writes to c2a and reads from a2c.) The client→agent path (c2a) runs
// through a hand-held writer so a test can close it (`closeClientTransport`)
// to simulate the editor disconnecting — closing it EOFs the agent's reader
// and resolves the bridge's `conn.closed`.
const a2c = new TransformStream<Uint8Array, Uint8Array>()
const c2a = new TransformStream<Uint8Array, Uint8Array>()
const c2aWriter = c2a.writable.getWriter()
// A WritableStream the client writes into; each chunk is forwarded to the
// held c2a writer. `closeClientTransport` closes that writer directly.
const clientOutput = new WritableStream<Uint8Array>({
write: chunk => c2aWriter.write(chunk),
})
const agentStream: Stream = ndJsonStream(a2c.writable, c2a.readable)
const clientStream: Stream = ndJsonStream(clientOutput, a2c.readable)
const updates: CapturedUpdate[] = []
const permissionRequests: RequestPermissionRequest[] = []
const harness: BridgeHarness = {
ctx,
adapter,
updates,
permissionRequests,
onPermission: () => ({ outcome: { outcome: 'cancelled' } }),
onSessionUpdateError: undefined,
client: undefined as unknown as ClientSideConnection,
acpFiber: undefined as unknown as BridgeHarness['acpFiber'],
// Close the writable the CLIENT writes to (c2a) — its readable, which the
// agent's ndJsonStream consumes, then EOFs cleanly, so the bridge's
// `conn.closed` resolves and it sees the client disconnect. If the client
// connection holds a writer lock on it, abort the connection's signal path
// instead by closing through the underlying stream.
closeClientTransport: async () => { await c2aWriter.close() },
dispose: async () => { await ctx.fiber.dispose() },
storageDir: options.storageDir,
}
const makeClient = (_agent: AcpAgent): Client => ({
sessionUpdate(params: SessionNotification): Promise<void> {
updates.push(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()
},
requestPermission(params: RequestPermissionRequest): Promise<RequestPermissionResponse> {
permissionRequests.push(params)
return Promise.resolve(harness.onPermission(params))
},
})
// Wire the bridge (agent side) and the client (test side). The test config
// can override `model` (including to undefined): default to 'mock' unless the
// caller explicitly set the key (even to undefined), so a `{ model: undefined }`
// 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)
}
harness.client = new ClientSideConnection(makeClient, clientStream)
return harness
}
+130
View File
@@ -0,0 +1,130 @@
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 { SessionId } from '@deepseek-ai/dsh-session'
import { makeBridgeHarness, textResponse, type BridgeHarness, type CapturedUpdate } from './harness.ts'
/** Concatenate the text of all agent_message_chunk updates. */
function messageText(updates: CapturedUpdate[]): string {
return updates
.filter(u => u.sessionUpdate === 'agent_message_chunk')
.map(u => (u.content.type === 'text' ? u.content.text : ''))
.join('')
}
describe('acp bridge — session/load replay', () => {
let storageDir: string
let live: BridgeHarness | undefined
let loader: BridgeHarness | undefined
beforeEach(async () => { storageDir = await mkdtemp(join(tmpdir(), 'acp-load-')) })
afterEach(async () => {
if (live) await live.dispose()
if (loader) await loader.dispose()
live = loader = undefined
await rm(storageDir, { recursive: true, force: true })
})
it('replays a persisted turn from the event log as session/update on load', async () => {
// 1. Create a session and run one turn — persistence writes the event log.
live = await makeBridgeHarness({ storageDir, script: [textResponse('remembered answer')] })
await live.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await live.client.newSession({ cwd: process.cwd(), mcpServers: [] })
await live.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'remember this' }] })
// Dispose to flush + release; the on-disk log persists.
await live.dispose()
live = undefined
// 2. A fresh bridge loads the same session id and must replay the turn.
loader = await makeBridgeHarness({ storageDir, script: [] })
await loader.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const res = await loader.client.loadSession({ sessionId, cwd: process.cwd(), mcpServers: [] })
expect(res).toBeDefined()
// The replayed updates reconstruct the assistant text from the event log
// (assistant/chunk → agent_message_chunk), NOT from deriveMessages.
expect(messageText(loader.updates)).toBe('remembered answer')
// And the USER side of the turn replays too (user/message →
// user_message_chunk), so the editor transcript shows both sides.
const userText = loader.updates
.filter(u => u.sessionUpdate === 'user_message_chunk')
.map(u => (u.content.type === 'text' ? u.content.text : ''))
.join('')
expect(userText).toBe('remember this')
})
it('a load whose resume finishes after a client disconnect leaks no live session', async () => {
// A session/load is mid-resume() when the client transport closes. The load
// must NOT end up with a live registered agent for the connection that is
// already gone. (The bridge's post-await `closed` guard backs this on real
// stdio; here the SDK rejects the in-flight request on close — either way no
// agent survives.) Stall persistence so resume() is pending across the close.
live = await makeBridgeHarness({ storageDir, script: [textResponse('x')] })
await live.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await live.client.newSession({ cwd: process.cwd(), mcpServers: [] })
await live.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'hi' }] })
await live.dispose()
live = undefined
loader = await makeBridgeHarness({ storageDir, script: [] })
await loader.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const realLoad = loader.ctx.sessionPersistence.load.bind(loader.ctx.sessionPersistence)
let release!: () => void
const gate = new Promise<void>((r) => { release = r })
loader.ctx.sessionPersistence.load = async (id) => { await gate; return realLoad(id) }
const loadResult = loader.client.loadSession({ sessionId, cwd: process.cwd(), mcpServers: [] })
.then(() => 'resolved' as const, () => 'rejected' as const)
await loader.closeClientTransport() // teardown sets `closed` while load is gated
release() // resume() finishes AFTER teardown
expect(await loadResult).toBe('rejected')
// No live agent was installed for the closed connection.
expect(loader.ctx.agents.get(sessionId)).toBeUndefined()
})
it('rejects load when the persisted session cwd differs from the launch dir', async () => {
// Seed a session on disk whose header.cwd is a DIFFERENT absolute path than
// the server's launch dir, then load it requesting the launch cwd (so the
// request-cwd check passes). The bridge must still reject on the persisted
// header cwd — else it would replay that session while tools run here.
loader = await makeBridgeHarness({ storageDir, script: [] })
const otherCwd = '/some/other/workspace'
await loader.ctx.sessionPersistence.create({
version: 1, id: SessionId('elsewhere'), createdAt: 1, cwd: otherCwd, updatedAt: 1,
})
await loader.ctx.sessionPersistence.append(SessionId('elsewhere'), [
{ type: 'turn/start', seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 1, time: 0, data: { turn: 1, reason: { kind: 'completed' } } },
])
await loader.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await expect(loader.client.loadSession({ sessionId: 'elsewhere', cwd: process.cwd(), mcpServers: [] }))
.rejects.toThrow(/created in \/some\/other\/workspace/)
// The rejected load must NOT have constructed/registered a live agent (the
// cwd is validated from persisted metadata BEFORE resume) — no leak.
expect(loader.ctx.agents.get('elsewhere')).toBeUndefined()
// And a fresh newSession still works (the connection is not wedged).
const ok = await loader.client.newSession({ cwd: process.cwd(), mcpServers: [] })
expect(ok.sessionId).toBeTruthy()
})
it('rejects load for a non-absolute or mismatched cwd', async () => {
loader = await makeBridgeHarness({ storageDir, script: [] })
await loader.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await expect(loader.client.loadSession({ sessionId: 's', cwd: 'rel', mcpServers: [] }))
.rejects.toThrow(/absolute/)
await expect(loader.client.loadSession({ sessionId: 's', cwd: '/other', mcpServers: [] }))
.rejects.toThrow(/launch directory/)
})
it('rejects load when a session already exists (single-session MVP)', async () => {
live = await makeBridgeHarness({ storageDir, script: [] })
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/)
})
})
+120
View File
@@ -0,0 +1,120 @@
/**
* Property-based protocol-shape tests for the ACP update stream (RFC 001 →
* ADR 0013 precedent). Fuzz arbitrary harness `SessionEvent` sequences through
* the pure `streamSessionEventUpdate` translator and assert the invariants an
* ACP client relies on:
*
* - every emitted update is a legal `SessionUpdate` variant;
* - a `tool_call_update` for a given id is never emitted before a `tool_call`
* for that id (the client must see the pending call before its completion);
* - the translator is a pure function of the event (same event → same updates),
* so live streaming and `session/load` replay produce identical streams.
*
* Pure-function fuzzing (no live loop) keeps these deterministic — a failure is
* a real finding, not timing noise.
*/
import { describe, expect, it } from 'vitest'
import fc from 'fast-check'
import { CallId } from '@deepseek-ai/dsh-llm'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import type { SessionNotification } from '@agentclientprotocol/sdk'
import { streamSessionEventUpdate } from '../src/index.ts'
const LEGAL_UPDATE_KINDS = new Set([
'agent_message_chunk',
'agent_thought_chunk',
'tool_call',
'tool_call_update',
])
/**
* Build a WELL-FORMED harness event sequence: a list of "actions" where a tool
* result can only reference a call already opened earlier. This mirrors what
* the loop actually appends (tool/call always precedes its tool/result), so the
* ordering invariant is asserted over realistic logs, not arbitrary noise.
*/
type Action =
| { kind: 'text'; text: string }
| { kind: 'reasoning'; text: string }
| { kind: 'call'; id: string; name: string }
| { kind: 'result'; idx: number; isError: boolean }
| { kind: 'ignored' }
function actionsArb(): fc.Arbitrary<Action[]> {
const action: fc.Arbitrary<Action> = fc.oneof(
fc.string().map((text): Action => ({ kind: 'text', text })),
fc.string().map((text): Action => ({ kind: 'reasoning', text })),
fc.record({ id: fc.string({ minLength: 1 }), name: fc.string() }).map(({ id, name }): Action => ({ kind: 'call', id, name })),
fc.record({ idx: fc.nat(), isError: fc.boolean() }).map(({ idx, isError }): Action => ({ kind: 'result', idx, isError })),
fc.constant<Action>({ kind: 'ignored' }),
)
return fc.array(action, { maxLength: 30 })
}
/** Lower well-formed actions into a harness event sequence. */
function actionsToEvents(actions: Action[]): SessionEvent[] {
const events: SessionEvent[] = []
const openCalls: string[] = []
for (const a of actions) {
switch (a.kind) {
case 'text':
events.push({ type: 'assistant/chunk', seq: 0, time: 0, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: a.text } } })
break
case 'reasoning':
events.push({ type: 'assistant/chunk', seq: 0, time: 0, data: { turn: 1, step: 1, chunk: { type: 'reasoning-delta', index: 0, text: a.text } } })
break
case 'call':
openCalls.push(a.id)
events.push({ type: 'tool/call', seq: 0, time: 0, data: { turn: 1, step: 1, callId: CallId(a.id), name: a.name, arguments: '{}' } })
break
case 'result': {
// Only emit a result for an already-opened call (well-formedness).
if (openCalls.length === 0) break
const id = openCalls[a.idx % openCalls.length]!
events.push({ type: 'tool/result', seq: 0, time: 0, data: { turn: 1, step: 1, callId: CallId(id), content: [], isError: a.isError } })
break
}
case 'ignored':
events.push({ type: 'turn/end', seq: 0, time: 0, data: { turn: 1, reason: { kind: 'completed' } } })
break
}
}
return events
}
function runStream(events: SessionEvent[]): SessionNotification['update'][] {
const out: SessionNotification['update'][] = []
for (const event of events) streamSessionEventUpdate('s1', event, n => out.push(n.update))
return out
}
describe('ACP update-stream invariants (property-based)', () => {
it('every emitted update is a legal SessionUpdate variant', () => {
fc.assert(fc.property(actionsArb(), (actions) => {
for (const update of runStream(actionsToEvents(actions))) {
expect(LEGAL_UPDATE_KINDS.has(update.sessionUpdate)).toBe(true)
}
}))
})
it('never emits a tool_call_update for an id before that id\'s tool_call', () => {
fc.assert(fc.property(actionsArb(), (actions) => {
const seenCall = new Set<string>()
for (const update of runStream(actionsToEvents(actions))) {
if (update.sessionUpdate === 'tool_call') {
seenCall.add(update.toolCallId)
} else if (update.sessionUpdate === 'tool_call_update') {
expect(seenCall.has(update.toolCallId)).toBe(true)
}
}
}))
})
it('is a pure function of the event (replay equals live)', () => {
fc.assert(fc.property(actionsArb(), (actions) => {
const events = actionsToEvents(actions)
expect(runStream(events)).toEqual(runStream(events))
}))
})
})
+109
View File
@@ -0,0 +1,109 @@
import { describe, expect, it } from 'vitest'
import { CallId } from '@deepseek-ai/dsh-llm'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import type { SessionNotification } from '@agentclientprotocol/sdk'
import { streamSessionEventUpdate, agentOptions } from '../src/index.ts'
/** Collect the updates a single event produces. */
function updatesFor(event: SessionEvent): SessionNotification['update'][] {
const out: SessionNotification['update'][] = []
streamSessionEventUpdate('s1', event, n => out.push(n.update))
return out
}
function evt<T extends SessionEvent['type']>(type: T, data: Extract<SessionEvent, { type: T }>['data']): SessionEvent {
return { type, seq: 0, time: 0, data } as SessionEvent
}
describe('streamSessionEventUpdate', () => {
it('maps assistant/chunk text-delta to agent_message_chunk', () => {
expect(updatesFor(evt('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'hi' } })))
.toEqual([{ sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'hi' } }])
})
it('maps assistant/chunk reasoning-delta to agent_thought_chunk', () => {
expect(updatesFor(evt('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'reasoning-delta', index: 0, text: 'mm' } })))
.toEqual([{ sessionUpdate: 'agent_thought_chunk', content: { type: 'text', text: 'mm' } }])
})
it('produces no update for a non-text/reasoning chunk (e.g. block-start)', () => {
expect(updatesFor(evt('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'block-start', index: 0, blockType: 'text' } })))
.toEqual([])
})
it('maps tool/call to an in_progress tool_call with inferred kind and parsed rawInput', () => {
const updates = updatesFor(evt('tool/call', { turn: 1, step: 1, callId: CallId('c1'), name: 'bash', arguments: '{"command":"ls"}' }))
expect(updates).toEqual([{
sessionUpdate: 'tool_call',
toolCallId: 'c1',
title: 'bash',
kind: 'execute',
status: 'in_progress',
rawInput: { command: 'ls' },
}])
})
it('infers tool kinds: read*/write*/edit*/other', () => {
const kind = (name: string): unknown =>
updatesFor(evt('tool/call', { turn: 1, step: 1, callId: CallId('c'), name, arguments: '' }))[0]
expect((kind('read_file') as { kind: string }).kind).toBe('read')
expect((kind('write') as { kind: string }).kind).toBe('edit')
expect((kind('edit_file') as { kind: string }).kind).toBe('edit')
expect((kind('frobnicate') as { kind: string }).kind).toBe('other')
})
it('falls back to the raw argument string when tool arguments are not JSON', () => {
const update = updatesFor(evt('tool/call', { turn: 1, step: 1, callId: CallId('c1'), name: 'bash', arguments: 'not json' }))[0]
expect((update as { rawInput: unknown }).rawInput).toBe('not json')
})
it('maps tool/result to completed/failed tool_call_update with text content', () => {
const ok = updatesFor(evt('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'out' }], isError: false }))
expect(ok).toEqual([{
sessionUpdate: 'tool_call_update',
toolCallId: 'c1',
status: 'completed',
content: [{ type: 'content', content: { type: 'text', text: 'out' } }],
}])
const failed = updatesFor(evt('tool/result', { turn: 1, step: 1, callId: CallId('c2'), content: [], isError: true }))
expect((failed[0] as { status: string }).status).toBe('failed')
})
it('drops non-text tool-result content (text-only)', () => {
const update = updatesFor(evt('tool/result', {
turn: 1, step: 1, callId: CallId('c1'),
content: [{ type: 'image', url: 'https://x/y.png' }],
isError: false,
}))[0]
expect((update as { content: unknown[] }).content).toEqual([])
})
it('maps user/message text blocks to user_message_chunk (load replays the user side)', () => {
// A text block surfaces; a non-text block (here a tool-call) is skipped, so
// only the text chunk is emitted.
expect(updatesFor(evt('user/message', {
content: [
{ type: 'text', text: 'hi' },
{ type: 'tool-call', id: CallId('c'), name: 'bash', arguments: '{}' },
],
source: { kind: 'user' },
}))).toEqual([{ sessionUpdate: 'user_message_chunk', content: { type: 'text', text: 'hi' } }])
// A user/message with no text-bearing blocks produces no chunk.
expect(updatesFor(evt('user/message', { content: [], source: { kind: 'user' } }))).toEqual([])
})
it('produces no update for boundary/other event types', () => {
expect(updatesFor(evt('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }))).toEqual([])
expect(updatesFor(evt('turn/end', { turn: 1, reason: { kind: 'completed' } }))).toEqual([])
expect(updatesFor(evt('usage', { turn: 1, step: 1, usage: { inputTokens: 1, outputTokens: 1 } }))).toEqual([])
})
})
describe('agentOptions', () => {
it('includes only the fields present in config', () => {
expect(agentOptions({})).toEqual({})
expect(agentOptions({ model: 'm' })).toEqual({ model: 'm' })
expect(agentOptions({ systemPrompt: 'sp' })).toEqual({ systemPrompt: 'sp' })
expect(agentOptions({ model: 'm', systemPrompt: 'sp' })).toEqual({ model: 'm', systemPrompt: 'sp' })
})
})
+231
View File
@@ -0,0 +1,231 @@
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 { defineTool } from '@deepseek-ai/dsh-tools'
import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
import {
errorResponse,
makeBridgeHarness,
maxTokensResponse,
textResponse,
toolCallResponse,
type BridgeHarness,
} from './harness.ts'
/** Boilerplate: initialize + create one session, returning its id. */
async function newSession(h: BridgeHarness): Promise<string> {
await h.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await h.client.newSession({ cwd: process.cwd(), mcpServers: [] })
return sessionId
}
describe('acp bridge — turn outcomes', () => {
let storageDir: string
let harness: BridgeHarness | undefined
beforeEach(async () => { storageDir = await mkdtemp(join(tmpdir(), 'acp-test-')) })
afterEach(async () => {
if (harness) await harness.dispose()
harness = undefined
await rm(storageDir, { recursive: true, force: true })
})
it('maps a max-tokens turn to stopReason max_tokens', async () => {
harness = await makeBridgeHarness({ storageDir, script: [maxTokensResponse('cut off')] })
const sessionId = await newSession(harness)
const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
expect(res.stopReason).toBe('max_tokens')
})
it('rejects the prompt RPC when a turn fails (no misleading end_turn)', async () => {
// ACP has no "error" stop reason; a failed turn must surface as a rejected
// session/prompt, not a normal end_turn that hides the failure from the
// client. The bridge rejects via the turn/end{error} log record.
harness = await makeBridgeHarness({ storageDir, script: [errorResponse('provider boom')] })
const sessionId = await newSession(harness)
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
.rejects.toThrow(/turn failed: provider boom/)
})
it('streams a tool call as tool_call then tool_call_update', async () => {
harness = await makeBridgeHarness({
storageDir,
script: [toolCallResponse('c1', 'bash', { command: 'echo hi' }), textResponse('done')],
})
harness.ctx.tools.register(defineTool({
name: 'bash',
description: 'run a command',
parameters: { command: { type: 'string' } },
async execute() { return [{ type: 'text', text: 'hi\n' }] },
}))
const sessionId = await newSession(harness)
await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'run it' }] })
const toolCalls = harness.updates.filter(u => u.sessionUpdate === 'tool_call')
const toolUpdates = harness.updates.filter(u => u.sessionUpdate === 'tool_call_update')
expect(toolCalls).toHaveLength(1)
expect(toolCalls[0]).toMatchObject({ toolCallId: 'c1', title: 'bash', kind: 'execute', status: 'in_progress' })
expect(toolUpdates).toHaveLength(1)
expect(toolUpdates[0]).toMatchObject({ toolCallId: 'c1', status: 'completed' })
// Ordering invariant: the tool_call precedes its tool_call_update.
const callIdx = harness.updates.findIndex(u => u.sessionUpdate === 'tool_call')
const updIdx = harness.updates.findIndex(u => u.sessionUpdate === 'tool_call_update')
expect(callIdx).toBeLessThan(updIdx)
})
it('a failing tool yields a failed tool_call_update', async () => {
harness = await makeBridgeHarness({
storageDir,
script: [toolCallResponse('c1', 'bash', { command: 'boom' }), textResponse('ok')],
})
harness.ctx.tools.register(defineTool({
name: 'bash',
description: 'run a command',
parameters: { command: { type: 'string' } },
async execute() { throw new Error('command failed') },
}))
const sessionId = await newSession(harness)
await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'run it' }] })
const failed = harness.updates.filter(u => u.sessionUpdate === 'tool_call_update' && u.status === 'failed')
expect(failed).toHaveLength(1)
})
it('settles via the log fallback when a prior session/event listener throws (starvation)', async () => {
// A peer session/event listener that runs BEFORE the bridge's listener
// throws on turn/end (prepend: true puts it first). cordis emit stops at the
// throw, so the bridge's session/event listener never sees turn/end and
// cannot settle there. The agent/status idle-fallback must reconcile the
// prompt from the log so the RPC settles instead of hanging.
harness = await makeBridgeHarness({ storageDir, script: [textResponse('answer')] })
harness.ctx.on('session/event', (_s, event) => {
if (event.type === 'turn/end') throw new Error('peer listener boom')
}, { prepend: true })
const sessionId = await newSession(harness)
const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
expect(res.stopReason).toBe('end_turn')
})
it('log fallback REJECTS when the starved turn ended in error', async () => {
// Same starvation as above, but the turn fails: the idle-fallback must
// reject the RPC from the logged turn/end{error}, not resolve.
harness = await makeBridgeHarness({ storageDir, script: [errorResponse('starved boom')] })
harness.ctx.on('session/event', (_s, event) => {
if (event.type === 'turn/end') throw new Error('peer listener boom')
}, { prepend: true })
const sessionId = await newSession(harness)
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
.rejects.toThrow(/turn failed: starved boom/)
})
it('log fallback infers the owning turn when turn/START capture is starved', async () => {
// A peer listener throws on turn/START (not turn/end): the bridge never
// captures inflight.turn via the live stream. A throwing turn/start listener
// also FAILS the turn (the throw is recorded as the turn's error). Without
// the watermark inference the fallback would resolve `cancelled` (the bug);
// with it, it infers the owning turn from the log and REJECTS from that
// turn's error turn/end. (The model's own error is never reached — the turn
// failed at start — so the rejection carries the listener's failure.)
harness = await makeBridgeHarness({ storageDir, script: [textResponse('never runs')] })
harness.ctx.on('session/event', (_s, event) => {
if (event.type === 'turn/start') throw new Error('peer listener boom on start')
}, { prepend: true })
const sessionId = await newSession(harness)
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
.rejects.toThrow(/turn failed:/)
})
it('a between-turn injection does not settle the prompt early (message-trigger correlation)', async () => {
// A plugin injects context (a one-shot injection-triggered turn) right after
// the prompt is queued but before the prompt's own message turn runs. The
// bridge must NOT mistake the injection turn's turn/end for the prompt's —
// it correlates only to message-triggered turns. The prompt settles on its
// OWN turn with the real model answer.
harness = await makeBridgeHarness({ storageDir, script: [textResponse('real answer')] })
const sessionId = await newSession(harness)
const agent = harness.ctx.agents.get(sessionId)!
// On the queued prompt, synchronously inject a one-shot context turn (idle
// inject writes turn/start{injection} → context/message → turn/end). Fire
// once so it lands between install and the prompt turn.
let injected = false
harness.ctx.on('agent/queued', (subject) => {
if (subject === agent && !injected) {
injected = true
agent.inject([{ type: 'text', text: 'ctx note' }], { source: { kind: 'plugin', plugin: 'test' } })
}
})
const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
expect(res.stopReason).toBe('end_turn')
const text = harness.updates
.filter(u => u.sessionUpdate === 'agent_message_chunk')
.map(u => (u.content.type === 'text' ? u.content.text : ''))
.join('')
expect(text).toContain('real answer')
})
it('rejects a second prompt while one is in flight', async () => {
harness = await makeBridgeHarness({ storageDir, script: ['hang'] })
const sessionId = await newSession(harness)
// Start the first prompt but do NOT await — it hangs in the model stream.
const first = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'one' }] })
// Give the loop a tick to install the settle + start running.
await new Promise(r => setTimeout(r, 30))
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'two' }] }))
.rejects.toThrow(/already in flight/)
// Cancel to settle the first so the harness disposes cleanly.
await harness.client.cancel({ sessionId })
await first
})
it('session/cancel aborts a running turn and settles the prompt as cancelled', async () => {
harness = await makeBridgeHarness({ storageDir, script: ['hang'] })
const sessionId = await newSession(harness)
const promptDone = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
await new Promise(r => setTimeout(r, 30))
await harness.client.cancel({ sessionId })
const res = await promptDone
expect(res.stopReason).toBe('cancelled')
})
it('cancel in the pre-step window still settles the prompt cancelled exactly once', async () => {
// No script entry is consumed before cancel: cancel immediately after the
// prompt is sent, before the model step starts. The prompt must still
// settle cancelled (best-effort abort + settle), not hang.
harness = await makeBridgeHarness({ storageDir, script: [textResponse('late')] })
const sessionId = await newSession(harness)
const promptDone = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
await harness.client.cancel({ sessionId })
const res = await promptDone
expect(res.stopReason).toBe('cancelled')
// The queued turn may still start after the cancel cleared the in-flight
// slot (the documented TODO(rfc010-cancel-prestep) best-effort window): its
// turn-start then fires with no prompt to tag, and the bridge does nothing.
// Let it run to completion and assert nothing re-settles (no throw, no hang).
await harness.ctx.agents.get(sessionId)!.whenIdle()
})
it('a cancelled turn\'s late turn/end does not settle the NEXT prompt', async () => {
// Regression: prompt A runs; cancel settles A and frees the slot; A's
// aborted turn/end is still pending in the loop. Prompt B is sent before
// A's turn/end arrives. A's late turn/end (an EARLIER turn number) must NOT
// settle B — B owns a later turn. B then completes on its OWN turn/end.
harness = await makeBridgeHarness({ storageDir, script: ['hang', textResponse('B answer')] })
const sessionId = await newSession(harness)
const a = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'A' }] })
await new Promise(r => setTimeout(r, 30)) // let A start running (turn 1)
await harness.client.cancel({ sessionId })
expect((await a).stopReason).toBe('cancelled')
// Immediately send B; its turn (2) is distinct from A's (1). If A's late
// turn/end leaked onto B, B would settle 'cancelled' instead of 'end_turn'.
const b = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'B' }] })
expect(b.stopReason).toBe('end_turn')
const text = harness.updates
.filter(u => u.sessionUpdate === 'agent_message_chunk')
.map(u => (u.content.type === 'text' ? u.content.text : ''))
.join('')
expect(text).toContain('B answer')
})
})
+17
View File
@@ -0,0 +1,17 @@
{
"extends": "../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib"
},
"include": ["src"],
"references": [
{ "path": "../../vendor/cosmokit" },
{ "path": "../../vendor/cordis" },
{ "path": "../../vendor/schemastery" },
{ "path": "../llm" },
{ "path": "../session" },
{ "path": "../agent" },
{ "path": "../session-persistence" }
]
}
+61 -2
View File
@@ -30,6 +30,14 @@ export class LoopAgent implements Agent {
private resolveDisposed!: () => void
/** Resolves when the driver loop has fully exited (tests/disposal). */
done: Promise<void> = Promise.resolve()
/**
* Pending {@link whenIdle} waiters, resolved by {@link settleIdleWaiters} when
* the agent next settles out of `running`. Kept as internal agent state (NOT
* an effect-scoped `ctx.on` listener) so a concurrent fiber disposal — which
* runs the agent's own listeners' disposers — cannot drop the waiter before
* the `disposed` transition fires and leave the promise hanging.
*/
private idleWaiters: (() => void)[] = []
constructor(
private ctx: Context,
@@ -49,9 +57,26 @@ export class LoopAgent implements Agent {
private setStatus(status: AgentStatus): void {
if (this._status === status || this._status === 'disposed') return
this._status = status
// Release quiescence waiters on a transition OUT of running BEFORE emitting
// (the disposer handles the disposed transition separately). Settling first
// means a throwing `agent/status` subscriber cannot starve a `whenIdle()`
// waiter (AGENTS.md "contain callback exceptions" — a lifecycle await must
// not hang on one bad listener).
if (status !== 'running') this.settleIdleWaiters()
this.ctx.emit('agent/status', this, status)
}
/**
* Resolve and clear all pending {@link whenIdle} waiters. Called on a
* running→idle transition (from {@link setStatus}) and on disposal (from the
* {@link start} disposer, which chains `done` for true loop-exit quiescence).
*/
private settleIdleWaiters(): void {
const waiters = this.idleWaiters
this.idleWaiters = []
for (const resolve of waiters) resolve()
}
private resolveSource(options?: SendOptions): MessageSource {
return options?.source ?? { kind: 'user' }
}
@@ -147,11 +172,41 @@ export class LoopAgent implements Agent {
this.currentAbort?.abort(reason ?? 'aborted')
}
/**
* Resolve once the agent has reached quiescence after settling out of
* `running`. If it is already disposed, awaits {@link done} (the loop-exit
* promise) — `agent/status('disposed')` fires in the disposer BEFORE the
* driver loop has unwound, so it is NOT itself a quiescence signal. If it is
* idle, resolves immediately. Otherwise queues an internal waiter (see
* {@link idleWaiters}) released on the next running→idle/disposed transition,
* resolving on `idle` directly (the turn fully ended) or chaining {@link done}
* on `disposed` (wait for the loop to actually exit). Implements the
* {@link Agent.whenIdle} contract used by teardown (`abort()` then
* `await whenIdle()`).
*/
whenIdle(): Promise<void> {
if (this._status === 'disposed') return this.done
if (this._status !== 'running') return Promise.resolve()
// Register an internal waiter (resolved by settleIdleWaiters on the next
// running→idle/disposed transition), NOT an effect-scoped `ctx.on` listener:
// a concurrent fiber disposal runs this agent's listener disposers, which
// could remove a `ctx.on` waiter before the `disposed` transition fires and
// hang the promise. On disposal the disposer settles the waiter AND we chain
// `done` here for true loop-exit quiescence (status flips to disposed before
// the loop unwinds); a plain idle transition resolves directly.
return new Promise<void>((resolve) => {
this.idleWaiters.push(() => {
resolve(this._status === 'disposed' ? this.done : undefined)
})
})
}
/**
* Start the driver loop. Returns a disposer: calling it sets status to
* `disposed`, emits `agent/status('disposed')`, resolves the disposed
* promise (unblocking the idle wait), and aborts the current request if
* any. The returned `agent.done` promise resolves once the loop exits.
* promise (unblocking the idle wait), releases any `whenIdle` waiters, and
* aborts the current request if any. The returned `agent.done` promise
* resolves once the loop exits.
*/
start(): () => void {
this.done = runLoop(this.ctx, this, {
@@ -167,6 +222,10 @@ export class LoopAgent implements Agent {
if (this._status === 'disposed') return
this._status = 'disposed'
this.resolveDisposed()
// Release whenIdle waiters BEFORE the (guarded) event emit — they are
// internal state that must settle even if a listener throws below. Each
// waiter chains `done`, so it resolves only once the loop actually exits.
this.settleIdleWaiters()
this.currentAbort?.abort('disposed')
// setStatus refuses transitions out of 'disposed', so emit directly —
// 'disposed' is part of the agent/status contract. Guarded: a throwing
+112
View File
@@ -254,6 +254,118 @@ describe('LoopAgent', () => {
expect(idleTransitionCount).toBe(1) // only the final transition from running
})
it('whenIdle() resolves immediately when the agent is not running', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
// Fresh agent is idle — whenIdle() takes the not-running fast path and
// resolves without subscribing. await must not hang.
await agent.whenIdle()
expect(agent.status).not.toBe('running')
})
it('whenIdle() awaits the running→idle transition, ignoring other subjects/running events', async () => {
const adapter = new MockAdapter([textResponse('ok'), textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
const other = ctx.agentLoop.create('a2', { model: 'mock' })
// Drive `agent` into `running`, then await whenIdle() — it subscribes to
// agent/status and resolves on the first transition out of running.
const running = new Promise<void>((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
if (subject === agent && status === 'running') { dispose(); resolve() }
})
})
send(agent, 'go')
await running
expect(agent.status).toBe('running')
// While `agent`'s whenIdle is pending, churn `other` through running→idle:
// every status event it emits hits whenIdle's guard with `subject !== this`,
// so the wait must ignore them and only resolve on `agent`'s own idle.
send(other, 'go')
await agent.whenIdle()
expect(agent.status).toBe('idle')
})
it('whenIdle() subscribed while running resolves via done when the agent is then disposed', async () => {
// Covers the waiter's disposed arm: whenIdle() queues an internal waiter
// while running (not the fast path), then the disposer settles it and chains
// `done` (loop exit), not an eager resolve. A bare LoopAgent + direct
// start() disposer keeps the emit synchronous.
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
const adapter = new MockAdapter(['hang'])
ctx.llm.registerAdapter(['mock'], adapter)
const session = ctx.sessions.create('bare')
const agent = new LoopAgent(ctx, AgentId('bare'), { model: 'mock' }, session)
const dispose = agent.start()
agent.send([{ type: 'text', text: 'go' }])
await new Promise(r => setTimeout(r, 30))
expect(agent.status).toBe('running')
const idle = agent.whenIdle() // queues an internal waiter (running)
dispose() // settles the waiter synchronously; whenIdle chains done
await idle
expect(agent.status).toBe('disposed')
await agent.done
})
it('whenIdle() subscribed while running survives a FIBER dispose (no hung promise)', async () => {
// The waiter is internal agent state, NOT an effect-scoped ctx.on listener:
// disposing the OWNING fiber runs the agent's listener disposers, which would
// have dropped a ctx.on-based waiter before the 'disposed' transition and
// hung the promise. With internal waiters, the fiber disposer still settles
// it. Regression for the round-3 whenIdle finding.
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
let agent!: LoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create('scoped', { model: 'mock' })
}, { inject: ['agentLoop'] }))
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
expect(agent.status).toBe('running')
const idle = agent.whenIdle() // queued while running
await fiber.dispose() // tears the fiber down (drops agent listeners)
await idle // must resolve, not hang
expect(agent.status).toBe('disposed')
})
it('whenIdle() on a disposed agent awaits the loop exit (done), not just the status flip', async () => {
// The disposer emits agent/status('disposed') BEFORE the driver loop
// unwinds, so whenIdle() must chain `done` (true quiescence) on the
// disposed path. Dispose a running agent, then assert whenIdle() resolves
// only after `done` — i.e. the loop has actually exited.
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
let agent!: LoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create('scoped', { model: 'mock' })
}, { inject: ['agentLoop'] }))
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
let doneResolved = false
void agent.done.then(() => { doneResolved = true })
await fiber.dispose() // sets status disposed, aborts, drains the loop
expect(agent.status).toBe('disposed')
// whenIdle() must not resolve before `done` has — chaining `done` is the
// quiescence guarantee. By here dispose() awaited the loop, so done is
// settled; whenIdle resolves and done is observed resolved.
await agent.whenIdle()
expect(doneResolved).toBe(true)
})
it('abort() resolves reason to "aborted" when no reason provided', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
+1
View File
@@ -55,6 +55,7 @@ The handle every plugin programs against:
- `agent.steer(content, options?)` — steer a running turn (inject between steps); behaves like `send` when idle
- `agent.inject(content, options?)` — inject in-session context (context/message event); the next request sees it. Does not run the model. While a turn is open it joins that turn; while idle it is wrapped in a one-shot `injection` turn so every event stays turn-enclosed (ADR 0017)
- `agent.abort(reason?)` — abort the in-flight step
- `agent.whenIdle()` — resolve once the agent reaches quiescence after settling out of `running` (idle → immediately; disposed → awaits the loop exit), the signal a teardown awaits (`abort()` then `await whenIdle()`). Observes the transition without disposing the agent.
- `agent.session`, `agent.status`, `agent.options`, `agent.id`
### Extension points
+21
View File
@@ -80,6 +80,27 @@ export interface Agent {
/** Abort the in-flight step (if any); the turn ends with reason 'aborted'. */
abort(reason?: string): void
/**
* Resolve once the agent has reached quiescence after settling out of
* `running`, or immediately if it is already idle. The quiescence signal a
* teardown awaits: `agent.abort()` then `await agent.whenIdle()` guarantees
* the in-flight turn has fully stopped before the caller proceeds (a closing
* ACP connection, a disposing UI plugin), rather than returning while the
* driver is still streaming.
*
* "Quiescence", not merely "status changed": a disposed agent emits
* `agent/status('disposed')` from inside its disposer, BEFORE the driver loop
* has unwound — so `whenIdle()` resolving on `disposed` must wait for the loop
* to actually exit (the implementation chains the loop-exit promise), not just
* observe the status flip. A mid-step disposal that never reaches `idle` still
* unblocks the await this way.
*
* Distinct from disposal: `whenIdle()` observes the transition WITHOUT tearing
* the agent down. A consumer that owns the agent's lifecycle disposes it
* separately.
*/
whenIdle(): Promise<void>
// TODO(sub-agents): spawn/fork seams — semantics deliberately deferred.
// The intended shape: a creation option referencing a parent agent
// (fork = seed the child Session with the parent's event log; spawn =
+1
View File
@@ -14,6 +14,7 @@ function stubAgent(rawId: string): Agent {
steer() {},
inject() {},
abort() {},
whenIdle() { return Promise.resolve() },
}
}
+1
View File
@@ -19,6 +19,7 @@ const packages = [
'packages/bash-local',
'packages/tool-bash',
'packages/invariants',
'packages/acp',
]
const root = resolve(import.meta.dirname, '..')
+2 -1
View File
@@ -48,7 +48,8 @@
"@deepseek-ai/dsh-llm-pi-ai": ["./packages/llm-pi-ai/src"],
"@deepseek-ai/dsh-bash-local": ["./packages/bash-local/src"],
"@deepseek-ai/dsh-tool-bash": ["./packages/tool-bash/src"],
"@deepseek-ai/dsh-invariants": ["./packages/invariants/src"]
"@deepseek-ai/dsh-invariants": ["./packages/invariants/src"],
"@deepseek-ai/dsh-acp": ["./packages/acp/src"]
}
}
}
+2 -1
View File
@@ -24,6 +24,7 @@
{ "path": "./packages/llm-pi-ai" },
{ "path": "./packages/bash-local" },
{ "path": "./packages/tool-bash" },
{ "path": "./packages/invariants" }
{ "path": "./packages/invariants" },
{ "path": "./packages/acp" }
]
}
+2 -1
View File
@@ -30,7 +30,8 @@
"@deepseek-ai/dsh-llm-pi-ai": ["./packages/llm-pi-ai/src"],
"@deepseek-ai/dsh-bash-local": ["./packages/bash-local/src"],
"@deepseek-ai/dsh-tool-bash": ["./packages/tool-bash/src"],
"@deepseek-ai/dsh-invariants": ["./packages/invariants/src"]
"@deepseek-ai/dsh-invariants": ["./packages/invariants/src"],
"@deepseek-ai/dsh-acp": ["./packages/acp/src"]
}
},
"include": ["packages/*/src", "packages/*/tests", "examples", "scripts"]
+37 -2
View File
@@ -5,6 +5,15 @@ __metadata:
version: 9
cacheKey: 10c0
"@agentclientprotocol/sdk@npm:0.25.1":
version: 0.25.1
resolution: "@agentclientprotocol/sdk@npm:0.25.1"
peerDependencies:
zod: ^3.25.0 || ^4.0.0
checksum: 10c0/c2e02b52e339b31b982a6293bedb318a8006a1190dff45da2762d9d74f79368046dbaabb3c63b99b6766590ef4e126323912496287cb0feca1c4183d9875d31c
languageName: node
linkType: hard
"@anthropic-ai/sdk@npm:0.91.1":
version: 0.91.1
resolution: "@anthropic-ai/sdk@npm:0.91.1"
@@ -553,6 +562,31 @@ __metadata:
languageName: unknown
linkType: soft
"@deepseek-ai/dsh-acp@workspace:packages/acp":
version: 0.0.0-use.local
resolution: "@deepseek-ai/dsh-acp@workspace:packages/acp"
dependencies:
"@agentclientprotocol/sdk": "npm:0.25.1"
"@deepseek-ai/dsh-agent": "npm:^0.0.1"
"@deepseek-ai/dsh-agent-loop": "npm:^0.0.1"
"@deepseek-ai/dsh-llm": "npm:^0.0.1"
"@deepseek-ai/dsh-session": "npm:^0.0.1"
"@deepseek-ai/dsh-session-persistence": "npm:^0.0.1"
"@deepseek-ai/dsh-session-persistence-jsonl": "npm:^0.0.1"
"@deepseek-ai/dsh-system-prompt": "npm:^0.0.1"
"@deepseek-ai/dsh-tools": "npm:^0.0.1"
cordis: "npm:^4.0.0-rc.6"
schemastery: "npm:^3.17.0"
zod: "npm:^4.0.0"
peerDependencies:
"@deepseek-ai/dsh-agent": ^0.0.1
"@deepseek-ai/dsh-llm": ^0.0.1
"@deepseek-ai/dsh-session": ^0.0.1
"@deepseek-ai/dsh-session-persistence": ^0.0.1
cordis: ^4.0.0-rc.6
languageName: unknown
linkType: soft
"@deepseek-ai/dsh-agent-loop@npm:^0.0.1, @deepseek-ai/dsh-agent-loop@workspace:packages/agent-loop":
version: 0.0.0-use.local
resolution: "@deepseek-ai/dsh-agent-loop@workspace:packages/agent-loop"
@@ -673,6 +707,7 @@ __metadata:
version: 0.0.0-use.local
resolution: "@deepseek-ai/dsh-root@workspace:."
dependencies:
"@agentclientprotocol/sdk": "npm:0.25.1"
"@stylistic/eslint-plugin": "npm:^5.10.0"
"@types/node": "npm:^25.3.5"
"@vitest/coverage-v8": "npm:^4.1.8"
@@ -4303,7 +4338,7 @@ __metadata:
languageName: node
linkType: hard
"schemastery@npm:^3.18.0, schemastery@workspace:vendor/schemastery":
"schemastery@npm:^3.17.0, schemastery@npm:^3.18.0, schemastery@workspace:vendor/schemastery":
version: 0.0.0-use.local
resolution: "schemastery@workspace:vendor/schemastery"
dependencies:
@@ -4901,7 +4936,7 @@ __metadata:
languageName: node
linkType: hard
"zod@npm:^3.25.0 || ^4.0.0, zod@npm:^4.1.11":
"zod@npm:^3.25.0 || ^4.0.0, zod@npm:^4.0.0, zod@npm:^4.1.11":
version: 4.4.3
resolution: "zod@npm:4.4.3"
checksum: 10c0/7ea31b558e88f9faf44f31dd185e2e1cbf51fed3081787fb96cc2534749b50c0acfc6da7f0922a7353ed092dd358c7d50c28ea96c94d04af64191bd33152eca3