mirror of
https://github.com/open-multi-agent/open-multi-agent.git
synced 2026-10-19 01:31:07 -05:00
* feat(agent): orchestrate external coding agents over ACP (#204) Add an AgentBackend seam so an Agent can run on either the LLM AgentRunner or an external agent. First backend: an ACP client that drives a coding CLI (e.g. Claude Code via its ACP adapter) as a first-class team member — same task DAG, shared memory, cascade, and token budget as LLM agents. - Extract AgentBackend ({ run, stream }) that AgentRunner already implements; Agent.getBackend() branches on AgentConfig.backend. - AcpBackend spawns the agent over stdio JSON-RPC and maps session updates to a RunResult (text, tool calls, usage); handles permissions and abort. - ACP SDK as a lazy optional peer; new /acp subpath export. - docs/external-agents.md, a hybrid-team example, and tests (in-process fake agent + subprocess integration). Experimental: ACP is young; token accounting maps ACP's single context figure and client-side fs proxying is not implemented yet. * fix(agent): harden ACP backend after review (#204) Follow-up fixes on the ACP external-agent backend (PR #360): - Token accounting: usage_update.used is a cumulative context-size gauge, not a per-turn delta. Record each turn's increment so summing across a reused session telescopes to the latest figure instead of double-counting. - systemPrompt: ACP has no system-prompt field, so prepend the agent's systemPrompt to the session's first turn (once). It was previously dropped and only fed coordinator routing. - Abort: when the signal is already aborted before a turn starts, return empty output without prompting (no in-flight turn to cancel). - Startup: clear the cached start() promise on failure so a later run retries instead of re-throwing the same rejection forever. - Permissions: auto-approve now prefers least-privilege allow_once over a session-wide allow_always. Docs (external-agents.md, CLAUDE.md) updated to match. ACP tests 10 to 15; lint + full suite (1135) green.
This commit is contained in:
Jack Chen authored and GitHub committed
1 parent
f0375eebea
commit
ffb2905017
22 files changed
+1465
-34
No files matched your search
@@ -46,7 +46,7 @@ The framework's key feature. The coordinator receives goal + roster → emits a
|
||||
|-------|-------|----------------|
|
||||
| Orchestrator | `orchestrator/orchestrator.ts`, `scheduler.ts` | Top-level API, task decomposition, retry/backoff, coordinator |
|
||||
| Team | `team/team.ts`, `messaging.ts` | Agent roster, MessageBus (point-to-point + broadcast), SharedMemory binding |
|
||||
| Agent | `agent/agent.ts`, `runner.ts`, `pool.ts`, `structured-output.ts`, `loop-detector.ts` | Lifecycle (idle→running→completed/error), conversation loop, concurrency pool + per-agent mutex, structured-output validation, loop detection |
|
||||
| Agent | `agent/agent.ts`, `runner.ts`, `pool.ts`, `structured-output.ts`, `loop-detector.ts`, `acp-backend.ts` | Lifecycle (idle→running→completed/error), conversation loop, concurrency pool + per-agent mutex, structured-output validation, loop detection; `AgentBackend` seam so an `Agent` runs on either an LLM `AgentRunner` or an external ACP subprocess |
|
||||
| Task | `task/queue.ts`, `task.ts` | Dependency-aware queue, auto-unblock on completion, cascade failure to dependents |
|
||||
| Tool | `tool/framework.ts`, `executor.ts`, `mcp.ts`, `text-tool-extractor.ts`, `built-in/` | `defineTool()` + Zod, ToolRegistry, parallel batch exec, MCP bridge, local-model text tool-call fallback, filesystem sandbox |
|
||||
| LLM | `llm/adapter.ts` + 12 per-provider files + `openai-common.ts` + `reasoning-fallback.ts` | `LLMAdapter` (`chat` + `stream`); lazy `createAdapter()` factory; `baseURL` for OpenAI-compatible servers; cross-provider reasoning round-tripping |
|
||||
@@ -55,7 +55,7 @@ The framework's key feature. The coordinator receives goal + roster → emits a
|
||||
| CLI | `cli/oma.ts` | Shell/CI entry; built to `dist/cli/oma.js`, exposed as the `oma` npm bin |
|
||||
| Utils | `utils/*.ts` | Semaphore, token accounting, keyword helpers, trace plumbing, secret/PII redaction |
|
||||
| Types / Errors | `types.ts`, `errors.ts` | All interfaces in one file (avoids circular deps); shared error types |
|
||||
| Exports | `index.ts`, `mcp.ts`, `ai-sdk.ts` | Root + `/mcp` + `/ai-sdk` subpaths so optional peers don't break the main import |
|
||||
| Exports | `index.ts`, `mcp.ts`, `ai-sdk.ts`, `acp.ts` | Root + `/mcp` + `/ai-sdk` + `/acp` subpaths so optional peers don't break the main import |
|
||||
|
||||
### Non-obvious invariants
|
||||
|
||||
@@ -67,6 +67,7 @@ Behavior that isn't visible from any single file and will cause bugs if missed:
|
||||
- **Filesystem tools are sandboxed, `bash` is not** — `file_read/file_write/file_edit/grep/glob` resolve every path (symlinks included) within `AgentConfig.cwd` / `OrchestratorConfig.defaultCwd`, defaulting to `<cwd>/.agent-workspace`. `null` disables the sandbox; `process.cwd()` widens it. → [docs/tool-configuration.md](docs/tool-configuration.md)
|
||||
- **Reasoning is dropped unless opted in** — provider-native `ReasoningBlock`s the target adapter can't echo are silently dropped unless `AgentConfig.preserveReasoningAsText` is on (then converted to inline `<thinking>` text). `<thinking>` text is never parsed back into a signed block. → [docs/context-management.md](docs/context-management.md)
|
||||
- **Local-model tool-call fallback** — `text-tool-extractor.ts` only runs when the server emits no native `tool_calls` (Ollama/vLLM/LM Studio); native calls always win.
|
||||
- **External (ACP) agents run their own loop** — `AgentConfig.backend = { kind: 'acp', … }` swaps the LLM `AgentRunner` for an `AcpBackend` (spawns a coding CLI over ACP). The runner's tool loop / sandbox / context strategy don't apply; OMA is the ACP *client*, the subprocess does its own file IO in `cwd` (no `fs/*` proxying yet), permission prompts default to auto-approve, `systemPrompt` is prepended to the session's first turn (ACP has no system-prompt field), and ACP's cumulative context-token figure is recorded as a per-turn delta into `tokenUsage.input_tokens` (no `usage_update` ⇒ not budget-gated). Everything downstream (`pool`/`scheduler`/`queue`/memory/budget) is backend-agnostic. → [docs/external-agents.md](docs/external-agents.md)
|
||||
- **Secrets are auto-redacted** from traces, bash output, and dashboard payloads (`utils/redaction.ts`).
|
||||
|
||||
### Subsystem docs
|
||||
@@ -82,6 +83,7 @@ Detailed behavior is documented in `docs/` — the single source of truth, so up
|
||||
| Checkpoint/resume over `MemoryStore` | `memory/checkpoint.ts`, `orchestrator/orchestrator.ts` | [checkpoint.md](docs/checkpoint.md) |
|
||||
| Tracing, progress events, dashboard | `utils/trace.ts`, `dashboard/` | [observability.md](docs/observability.md) |
|
||||
| CLI usage + JSON schemas | `cli/oma.ts` | [cli.md](docs/cli.md) |
|
||||
| External coding agents over ACP | `agent/acp-backend.ts` | [external-agents.md](docs/external-agents.md) |
|
||||
|
||||
### Adding an LLM Adapter
|
||||
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
# External Agents (ACP)
|
||||
|
||||
OMA can orchestrate **external agents that run as local processes** alongside its
|
||||
LLM-backed agents. The first supported kind is a coding agent driven over the
|
||||
[Agent Client Protocol (ACP)](https://agentclientprotocol.com) — a JSON-RPC-over-stdio
|
||||
standard implemented by Gemini CLI, Claude Code, Codex, and others.
|
||||
|
||||
An ACP agent is a first-class team member: it sits in the same task DAG, writes to
|
||||
the same shared memory, cascades failure to its dependents, and counts against the
|
||||
same token budget as any LLM agent. The motivating shape is a **hybrid team** — an
|
||||
LLM planner decomposes the goal, an external coding agent writes the code, an LLM
|
||||
reviewer audits the diff — all in one `runTeam` / `runTasks` call.
|
||||
|
||||
## Quick start
|
||||
|
||||
Declare an agent with `backend` instead of a model. Everything else about the team
|
||||
is unchanged:
|
||||
|
||||
```typescript
|
||||
import { OpenMultiAgent } from '@open-multi-agent/core'
|
||||
|
||||
const oma = new OpenMultiAgent({ defaultModel: 'claude-sonnet-4-6', defaultProvider: 'anthropic' })
|
||||
|
||||
const team = oma.createTeam('hybrid-dev', {
|
||||
name: 'hybrid-dev',
|
||||
agents: [
|
||||
{ name: 'planner', systemPrompt: 'Break the task into a short plan. Do not write code.' },
|
||||
{
|
||||
name: 'coder',
|
||||
systemPrompt: 'Writes and edits code by running an external coding CLI.',
|
||||
backend: {
|
||||
kind: 'acp',
|
||||
command: 'npx',
|
||||
args: ['-y', '@agentclientprotocol/claude-agent-acp'], // Claude Code over ACP
|
||||
cwd: process.cwd(),
|
||||
},
|
||||
},
|
||||
{ name: 'reviewer', systemPrompt: 'Review the change and summarize risks. Do not edit files.' },
|
||||
],
|
||||
sharedMemory: true,
|
||||
})
|
||||
|
||||
const result = await oma.runTeam(team, 'Add a slugify() utility with tests, then review it.')
|
||||
```
|
||||
|
||||
The coordinator routes the coding work to `coder` based on its roster description; the
|
||||
ACP subprocess does the file edits; `reviewer` then reads the result from shared memory.
|
||||
|
||||
A runnable version is at
|
||||
[`examples/integrations/external-agent-acp.ts`](../packages/core/examples/integrations/external-agent-acp.ts).
|
||||
|
||||
## Installation
|
||||
|
||||
ACP support requires the optional peer dependency, loaded lazily so it never affects
|
||||
consumers that don't use it:
|
||||
|
||||
```bash
|
||||
npm install @agentclientprotocol/sdk
|
||||
```
|
||||
|
||||
You also need an ACP-speaking agent. Set the backend's `command` / `args` to launch
|
||||
it — any ACP agent works. Common choices:
|
||||
|
||||
| Agent | `command` / `args` | Notes |
|
||||
|-------|--------------------|-------|
|
||||
| **Claude Code** | `npx -y @agentclientprotocol/claude-agent-acp` | Official [Claude Agent SDK adapter](https://github.com/agentclientprotocol/claude-agent-acp) (Claude Code has no native ACP). Auth via `ANTHROPIC_API_KEY`. |
|
||||
| Gemini CLI | `gemini --acp` | Native ACP. Note: Google is reportedly retiring the free-tier Gemini CLI (and its `--experimental-acp` flag) — verify availability before relying on it. |
|
||||
| Codex | `codex-acp` (or `codex --experimental-acp`) | Experimental ACP support. |
|
||||
|
||||
The example below uses Claude Code, which fits OMA's Anthropic-centric default and
|
||||
needs only one key (`ANTHROPIC_API_KEY`) for the whole team.
|
||||
|
||||
## Configuration
|
||||
|
||||
`AgentConfig.backend` takes an `AgentBackendConfig` (a discriminated union; only
|
||||
`kind: 'acp'` exists today):
|
||||
|
||||
| Field | Type | Default | Meaning |
|
||||
|-------|------|---------|---------|
|
||||
| `kind` | `'acp'` | — | Backend discriminant. |
|
||||
| `command` | `string` | — | Executable to spawn (`'npx'`, `'gemini'`, …). |
|
||||
| `args` | `string[]` | `[]` | Arguments passed to `command`. |
|
||||
| `env` | `Record<string,string>` | — | Extra env vars, merged over `process.env`. |
|
||||
| `cwd` | `string` | `process.cwd()` | Directory the agent reads and edits. |
|
||||
| `permission` | `'auto-approve' \| 'reject' \| fn` | `'auto-approve'` | How to answer permission prompts (below). |
|
||||
|
||||
When `backend` is set, the LLM-specific fields (`model`, `provider`, `adapter`,
|
||||
sampling, `tools`, context strategy) do not apply — the external agent runs its own
|
||||
loop, and `model` becomes optional. The agent's `systemPrompt` is the exception: it
|
||||
still shapes the external agent because OMA — lacking any ACP system-prompt field —
|
||||
prepends it to the agent's first prompt (once per session), on top of seeding the
|
||||
coordinator's routing as it does for every agent.
|
||||
|
||||
### Permissions
|
||||
|
||||
ACP agents ask the client to approve sensitive tool calls (editing a file, running a
|
||||
command). Because OMA runs agents autonomously inside a DAG, the default is
|
||||
`'auto-approve'` (it picks the least-privilege `allow_once` when offered, otherwise
|
||||
`allow_always`). Tighten it as needed:
|
||||
|
||||
```typescript
|
||||
backend: {
|
||||
kind: 'acp',
|
||||
command: 'npx',
|
||||
args: ['-y', '@agentclientprotocol/claude-agent-acp'],
|
||||
// Reject everything…
|
||||
permission: 'reject',
|
||||
// …or decide per request.
|
||||
permission: (req) => req.kind !== 'delete' && !req.title.includes('rm -rf'),
|
||||
}
|
||||
```
|
||||
|
||||
The callback receives a minimal, SDK-agnostic `{ title, kind, optionKinds }` and returns
|
||||
`true` to approve / `false` to reject.
|
||||
|
||||
> **Security.** Unlike OMA's filesystem-tool sandbox, an ACP agent accesses `cwd`
|
||||
> directly — it is a local subprocess with your permissions. Scope `cwd` to a project
|
||||
> you trust the agent with, and use `permission` to gate destructive actions.
|
||||
|
||||
## How it works
|
||||
|
||||
OMA takes the ACP **client** role. On the first run for an agent it spawns the
|
||||
subprocess, frames its stdio as newline-delimited JSON-RPC, `initialize`s, and opens a
|
||||
`session/new` in `cwd`. Each `agent.run(prompt)` then sends one `session/prompt` turn
|
||||
and drains `session/update` notifications into a normal agent result:
|
||||
|
||||
| ACP update / stop | Maps to |
|
||||
|-------------------|---------|
|
||||
| `agent_message_chunk` (text) | streamed `text` deltas + the result's `output` |
|
||||
| `tool_call` / `tool_call_update` | entries in `result.toolCalls` |
|
||||
| `usage_update` (`used`) | `result.tokenUsage` (see caveat) |
|
||||
| stop `end_turn` | success |
|
||||
| stop `max_tokens` / `max_turn_requests` | success with `budgetExceeded` (stopped early) |
|
||||
| stop `refusal` | task failure (cascades to dependents) |
|
||||
| stop `cancelled` | returns partial output (from an abort) |
|
||||
|
||||
The seam is the `AgentBackend` interface (`run` + `stream`) that `AgentRunner` already
|
||||
implements — so the pool, scheduler, task queue, shared memory, and budget accounting
|
||||
treat an ACP agent exactly like an LLM agent, with no special cases.
|
||||
|
||||
### Token accounting caveat
|
||||
|
||||
ACP reports a single **context-token** figure (`usage_update.used` — "tokens
|
||||
currently in context"), not an input/output split, and it is *cumulative* across a
|
||||
session, not a per-turn delta. Because OMA reuses one session across an agent's
|
||||
turns, it records each turn's usage as the **increment** since the previous reading
|
||||
and stores it as `tokenUsage.input_tokens` (with `output_tokens: 0`) — so summing
|
||||
across turns telescopes to the latest figure instead of double-counting. That total
|
||||
aggregates into the run and honors `maxTokenBudget`. An agent that emits no
|
||||
`usage_update` reports `{0, 0}` and is therefore **not** budget-gated — size the
|
||||
budget on LLM agents, or bound the ACP agent with its own `--max-*` flags.
|
||||
|
||||
## Programmatic API
|
||||
|
||||
Most users only touch `backend`. To construct a backend directly, import from the
|
||||
`@open-multi-agent/core/acp` subpath (this is where the optional peer is loaded):
|
||||
|
||||
```typescript
|
||||
import { createAcpBackend } from '@open-multi-agent/core/acp'
|
||||
|
||||
const backend = createAcpBackend({ command: 'npx', args: ['-y', '@agentclientprotocol/claude-agent-acp'] })
|
||||
const result = await backend.run([{ role: 'user', content: [{ type: 'text', text: 'refactor foo.ts' }] }])
|
||||
await backend.dispose() // close the connection and kill the subprocess
|
||||
```
|
||||
|
||||
## v1 scope
|
||||
|
||||
What this release does **not** do yet (open an issue with a real use case to pull any
|
||||
of these forward):
|
||||
|
||||
- **Client role only.** OMA drives external agents; it does not expose OMA agents *as*
|
||||
an ACP agent to editors.
|
||||
- **No `fs/*` proxying.** The agent does its own filesystem access within `cwd`; OMA
|
||||
does not yet proxy ACP file operations through its sandbox. Agents that require the
|
||||
client to serve files are not supported.
|
||||
- **ACP only.** Bespoke per-CLI adapters and non-coding business-system CLIs are out of
|
||||
scope. ACP already covers the major coding agents.
|
||||
- **No cost-based budgets.** Budgeting is token-based; `usage_update.cost` is ignored.
|
||||
- **Subprocess lifetime.** An orchestrated ACP agent's subprocess lives until the
|
||||
process exits (there is no per-agent disposal hook in `runTeam` / `runTasks`).
|
||||
Use the programmatic API + `dispose()` when you need explicit teardown.
|
||||
Generated
+15
@@ -15,6 +15,16 @@
|
||||
"node": ">=18.0.0"
|
||||
}
|
||||
},
|
||||
"node_modules/@agentclientprotocol/sdk": {
|
||||
"version": "1.2.1",
|
||||
"resolved": "https://registry.npmjs.org/@agentclientprotocol/sdk/-/sdk-1.2.1.tgz",
|
||||
"integrity": "sha512-jwYUdOQR7tc+Zfch53VL4JJyUNK/46q03uUTYb+PjECsmnNl94XFXOfYLJ8RBpMNidXd1rpOAVgb0vqD98xImA==",
|
||||
"dev": true,
|
||||
"license": "Apache-2.0",
|
||||
"peerDependencies": {
|
||||
"zod": "^3.25.0 || ^4.0.0"
|
||||
}
|
||||
},
|
||||
"node_modules/@ai-sdk/gateway": {
|
||||
"version": "4.0.6",
|
||||
"resolved": "https://registry.npmjs.org/@ai-sdk/gateway/-/gateway-4.0.6.tgz",
|
||||
@@ -6467,6 +6477,7 @@
|
||||
"oma": "dist/cli/oma.js"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@agentclientprotocol/sdk": "^1.2.1",
|
||||
"@ai-sdk/openai": "^4.0.0",
|
||||
"@ai-sdk/provider": "^4.0.0",
|
||||
"@aws-sdk/client-bedrock-runtime": "^3.1040.0",
|
||||
@@ -6483,12 +6494,16 @@
|
||||
"node": ">=18.0.0"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"@agentclientprotocol/sdk": "^1.2.0",
|
||||
"@aws-sdk/client-bedrock-runtime": "^3.700.0",
|
||||
"@google/genai": "^1.48.0",
|
||||
"@modelcontextprotocol/sdk": "^1.18.0",
|
||||
"ai": "^5.0.0 || ^6.0.0 || ^7.0.0"
|
||||
},
|
||||
"peerDependenciesMeta": {
|
||||
"@agentclientprotocol/sdk": {
|
||||
"optional": true
|
||||
},
|
||||
"@aws-sdk/client-bedrock-runtime": {
|
||||
"optional": true
|
||||
},
|
||||
|
||||
@@ -14,6 +14,8 @@ contributed them. Current:
|
||||
jobs — N Open Design landing-page variants generated in parallel via
|
||||
`runTasks()`, each variant driven `create_project → start_run → poll get_run`
|
||||
to a terminal status by deterministic code.
|
||||
- `external-agent-acp.ts`: external coding agents (Gemini CLI / Claude Code) as
|
||||
team members via the `acp` backend. See [docs/external-agents.md](../../../../docs/external-agents.md).
|
||||
- `trace-observability.ts`: the `onTrace` API.
|
||||
- `with-vercel-ai-sdk/`: Next.js + AI SDK + `runTeam()`.
|
||||
- `express-customer-support/`: Express REST API + `runTasks()`. Contributed
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
/**
|
||||
* External coding agents over ACP (Agent Client Protocol)
|
||||
*
|
||||
* A hybrid team: LLM-backed agents plan and review, while an *external* coding
|
||||
* agent (a local CLI driven over ACP) writes the code — all in one OMA task DAG
|
||||
* with shared memory, cascade-on-failure, and unified token accounting. The
|
||||
* coordinator decomposes the goal and routes the coding work to the ACP agent
|
||||
* because of its roster description; the LLM reviewer then audits what it wrote.
|
||||
*
|
||||
* The ACP agent is declared with `backend: { kind: 'acp', ... }` instead of a
|
||||
* model. OMA spawns it, exchanges JSON-RPC over stdio, and turns each prompt
|
||||
* turn into a normal agent result — so it behaves like any other team member.
|
||||
*
|
||||
* Run:
|
||||
* npx tsx packages/core/examples/integrations/external-agent-acp.ts
|
||||
*
|
||||
* Prerequisites:
|
||||
* - ANTHROPIC_API_KEY one key for the whole team: the LLM
|
||||
* planner + reviewer AND the Claude Code coder
|
||||
* - @agentclientprotocol/sdk installed (the optional ACP peer)
|
||||
* - Node.js with `npx` on PATH. The coder runs Claude Code over ACP via the
|
||||
* official adapter `npx @agentclientprotocol/claude-agent-acp` (auto-downloaded
|
||||
* on first run). Swap `command`/`args` below for any other ACP agent — e.g.
|
||||
* Gemini CLI (`gemini --acp`) or Codex (`codex-acp`). See docs/external-agents.md.
|
||||
*/
|
||||
|
||||
import { OpenMultiAgent } from '../../src/index.js'
|
||||
|
||||
if (!process.env.ANTHROPIC_API_KEY?.trim()) {
|
||||
console.error('Missing ANTHROPIC_API_KEY: needed for the planner/reviewer and the Claude Code coder.')
|
||||
process.exit(1)
|
||||
}
|
||||
|
||||
// The directory the coding agent is allowed to read and edit. Point this at a
|
||||
// scratch project you trust the external agent with.
|
||||
const projectDir = process.env.OMA_ACP_PROJECT_DIR ?? process.cwd()
|
||||
|
||||
const oma = new OpenMultiAgent({
|
||||
defaultModel: 'claude-sonnet-4-6',
|
||||
defaultProvider: 'anthropic',
|
||||
maxConcurrency: 3,
|
||||
})
|
||||
|
||||
const team = oma.createTeam('hybrid-dev', {
|
||||
name: 'hybrid-dev',
|
||||
agents: [
|
||||
{
|
||||
name: 'planner',
|
||||
systemPrompt:
|
||||
'You break a software task into a short, concrete implementation plan. ' +
|
||||
'You do not write code yourself.',
|
||||
},
|
||||
{
|
||||
// External coding agent — no model; runs Claude Code as a local subprocess
|
||||
// over ACP, via the official Claude Agent SDK adapter.
|
||||
name: 'coder',
|
||||
systemPrompt:
|
||||
'Writes and edits code in the project by running Claude Code. ' +
|
||||
'Assign all file-editing and code-writing work here.',
|
||||
backend: {
|
||||
kind: 'acp',
|
||||
command: 'npx',
|
||||
args: ['-y', '@agentclientprotocol/claude-agent-acp'],
|
||||
cwd: projectDir,
|
||||
// OMA runs autonomously, so tool prompts are auto-approved within `cwd`.
|
||||
// Use 'reject' or a function to gate specific actions.
|
||||
permission: 'auto-approve',
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'reviewer',
|
||||
systemPrompt:
|
||||
'You review a code change for correctness and clarity and summarize the diff, ' +
|
||||
'calling out risks. You do not edit files.',
|
||||
},
|
||||
],
|
||||
sharedMemory: true,
|
||||
})
|
||||
|
||||
const result = await oma.runTeam(
|
||||
team,
|
||||
'Add a `slugify(text)` utility (lowercase, spaces and punctuation to single hyphens) ' +
|
||||
'with a couple of unit tests, then review the change.',
|
||||
{
|
||||
onProgress: (event) => {
|
||||
if (event.type === 'task_complete') {
|
||||
console.log(`✓ ${event.agent} finished task ${event.task}`)
|
||||
}
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
console.log('\n=== Final output ===\n')
|
||||
console.log(result.agentResults.get('coordinator')?.output ?? '(no synthesis)')
|
||||
console.log('\nTotal tokens:', result.totalTokenUsage)
|
||||
@@ -21,6 +21,10 @@
|
||||
"types": "./dist/index.d.ts",
|
||||
"import": "./dist/index.js"
|
||||
},
|
||||
"./acp": {
|
||||
"types": "./dist/acp.d.ts",
|
||||
"import": "./dist/acp.js"
|
||||
},
|
||||
"./mcp": {
|
||||
"types": "./dist/mcp.d.ts",
|
||||
"import": "./dist/mcp.js"
|
||||
@@ -76,12 +80,16 @@
|
||||
"zod": "^3.23.0"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"@agentclientprotocol/sdk": "^1.2.0",
|
||||
"@aws-sdk/client-bedrock-runtime": "^3.700.0",
|
||||
"@google/genai": "^1.48.0",
|
||||
"@modelcontextprotocol/sdk": "^1.18.0",
|
||||
"ai": "^5.0.0 || ^6.0.0 || ^7.0.0"
|
||||
},
|
||||
"peerDependenciesMeta": {
|
||||
"@agentclientprotocol/sdk": {
|
||||
"optional": true
|
||||
},
|
||||
"@aws-sdk/client-bedrock-runtime": {
|
||||
"optional": true
|
||||
},
|
||||
@@ -96,6 +104,7 @@
|
||||
}
|
||||
},
|
||||
"devDependencies": {
|
||||
"@agentclientprotocol/sdk": "^1.2.1",
|
||||
"@ai-sdk/openai": "^4.0.0",
|
||||
"@ai-sdk/provider": "^4.0.0",
|
||||
"@aws-sdk/client-bedrock-runtime": "^3.1040.0",
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
/**
|
||||
* @fileoverview `@open-multi-agent/core/acp` — orchestrate external coding agents
|
||||
* (Gemini CLI, Claude Code, Codex, …) over the Agent Client Protocol (ACP).
|
||||
*
|
||||
* Most users never import this module: set {@link AgentBackendConfig} on an
|
||||
* agent's `backend` field and OMA loads the backend for you. Import from here only
|
||||
* to construct a backend directly (advanced / programmatic use):
|
||||
*
|
||||
* ```ts
|
||||
* import { createAcpBackend } from '@open-multi-agent/core/acp'
|
||||
* const backend = createAcpBackend({ command: 'npx', args: ['-y', '@agentclientprotocol/claude-agent-acp'] })
|
||||
* ```
|
||||
*
|
||||
* Requires the optional peer dependency `@agentclientprotocol/sdk`.
|
||||
*/
|
||||
|
||||
export { AcpBackend, createAcpBackend } from './agent/acp-backend.js'
|
||||
export type { AcpBackendOptions, AcpConnection } from './agent/acp-backend.js'
|
||||
export type { AgentBackend } from './agent/runner.js'
|
||||
export type {
|
||||
AgentBackendConfig,
|
||||
AcpPermissionPolicy,
|
||||
AcpPermissionRequest,
|
||||
} from './types.js'
|
||||
@@ -0,0 +1,501 @@
|
||||
/**
|
||||
* @fileoverview {@link AgentBackend} that drives an external coding agent over the
|
||||
* Agent Client Protocol (ACP).
|
||||
*
|
||||
* OMA takes the ACP **client** role: it spawns the agent (a local CLI such as
|
||||
* Gemini CLI or Claude Code) as a subprocess, exchanges JSON-RPC over stdio, and
|
||||
* turns one `session/prompt` turn into a {@link RunResult}. This lets an ACP agent
|
||||
* sit in the same task DAG as LLM agents — with shared memory, cascade-on-failure,
|
||||
* and token budget — because it satisfies the same {@link AgentBackend} contract.
|
||||
*
|
||||
* Requires the optional peer dependency `@agentclientprotocol/sdk`, loaded lazily
|
||||
* so the three-runtime-dependency promise holds for consumers that never use ACP.
|
||||
* Reached only through `AgentConfig.backend` (see {@link Agent}) or the
|
||||
* `@open-multi-agent/core/acp` subpath — never imported by the package root.
|
||||
*
|
||||
* v1 scope: OMA is the client only; the agent performs its own filesystem access
|
||||
* within `cwd` (client-side `fs/*` proxying is intentionally not implemented yet).
|
||||
*/
|
||||
|
||||
import { spawn } from 'node:child_process'
|
||||
import { resolve as resolvePath } from 'node:path'
|
||||
import { Readable, Writable } from 'node:stream'
|
||||
|
||||
import type {
|
||||
ActiveSession,
|
||||
CancelNotification,
|
||||
ClientApp,
|
||||
ClientConnection,
|
||||
ClientContext,
|
||||
PermissionOption,
|
||||
RequestPermissionRequest,
|
||||
RequestPermissionResponse,
|
||||
StopReason,
|
||||
} from '@agentclientprotocol/sdk'
|
||||
|
||||
import type {
|
||||
AcpPermissionPolicy,
|
||||
AcpPermissionRequest,
|
||||
ContentBlock,
|
||||
LLMMessage,
|
||||
StreamEvent,
|
||||
TokenUsage,
|
||||
ToolCallRecord,
|
||||
} from '../types.js'
|
||||
import type { AgentBackend, RunOptions, RunResult } from './runner.js'
|
||||
|
||||
/** The `@agentclientprotocol/sdk` module shape, resolved lazily. */
|
||||
type AcpSdkModule = typeof import('@agentclientprotocol/sdk')
|
||||
|
||||
const ZERO_USAGE: TokenUsage = { input_tokens: 0, output_tokens: 0 }
|
||||
|
||||
/** A live ACP client connection paired with its teardown. */
|
||||
export interface AcpConnection {
|
||||
readonly connection: ClientConnection
|
||||
dispose(): Promise<void> | void
|
||||
}
|
||||
|
||||
/** Options for {@link createAcpBackend} / {@link AcpBackend}. */
|
||||
export interface AcpBackendOptions {
|
||||
/** Executable to spawn (e.g. `'npx'`, `'gemini'`, `'codex-acp'`). */
|
||||
readonly command: string
|
||||
/** Arguments passed to `command` (e.g. `['-y', '@agentclientprotocol/claude-agent-acp']`). */
|
||||
readonly args?: readonly string[]
|
||||
/** Extra environment variables for the subprocess, merged over `process.env`. */
|
||||
readonly env?: Readonly<Record<string, string>>
|
||||
/** Working directory the agent reads and edits. Defaults to `process.cwd()`. */
|
||||
readonly cwd?: string
|
||||
/** How to answer the agent's permission prompts. Defaults to `'auto-approve'`. */
|
||||
readonly permission?: AcpPermissionPolicy
|
||||
/**
|
||||
* The agent's role / system prompt. ACP has no session-level system-prompt
|
||||
* field, so this is prepended to the *first* `session/prompt` turn (once per
|
||||
* session) to convey the agent's role to the external CLI. Empty/whitespace is
|
||||
* ignored.
|
||||
*/
|
||||
readonly systemPrompt?: string
|
||||
/** Agent name, used as the ACP client identity and in error messages. */
|
||||
readonly agentName?: string
|
||||
/** Model/label, forwarded only for diagnostics. */
|
||||
readonly model?: string
|
||||
/** Advisory per-call timeout hint (unused today; reserved). */
|
||||
readonly callTimeoutMs?: number
|
||||
/**
|
||||
* Test seam. Given the configured {@link ClientApp}, return an open connection
|
||||
* instead of spawning `command` — e.g. `app.connect(fakeAgentApp)` for an
|
||||
* in-process fake agent. Defaults to spawning a subprocess and framing its
|
||||
* stdio as newline-delimited JSON-RPC.
|
||||
*/
|
||||
readonly connect?: (app: ClientApp, sdk: AcpSdkModule) => AcpConnection | Promise<AcpConnection>
|
||||
}
|
||||
|
||||
/** Construct an ACP-backed {@link AgentBackend}. The subprocess is spawned lazily on first run. */
|
||||
export function createAcpBackend(options: AcpBackendOptions): AgentBackend {
|
||||
return new AcpBackend(options)
|
||||
}
|
||||
|
||||
/**
|
||||
* {@link AgentBackend} implementation backed by an ACP subprocess.
|
||||
*
|
||||
* The subprocess + session are created on the first `run`/`stream` and reused for
|
||||
* subsequent calls (preserving conversation context). Call {@link dispose} to
|
||||
* close the connection and kill the subprocess.
|
||||
*/
|
||||
export class AcpBackend implements AgentBackend {
|
||||
private readonly options: AcpBackendOptions
|
||||
private sdk: AcpSdkModule | null = null
|
||||
private conn: AcpConnection | null = null
|
||||
private ctx: ClientContext | null = null
|
||||
private session: ActiveSession | null = null
|
||||
private starting: Promise<void> | null = null
|
||||
/**
|
||||
* Cumulative context-token reading from the last `usage_update`. ACP reports a
|
||||
* running total ("tokens currently in context"), not a per-turn delta, so each
|
||||
* turn's usage is reported as the increment over this value (see {@link stream}).
|
||||
*/
|
||||
private lastUsedTokens = 0
|
||||
/** Role prompt to prepend to the first turn of the session; cleared once sent. */
|
||||
private systemPromptPending: string | undefined
|
||||
|
||||
constructor(options: AcpBackendOptions) {
|
||||
this.options = options
|
||||
this.systemPromptPending = options.systemPrompt?.trim() || undefined
|
||||
}
|
||||
|
||||
/** Run one prompt turn to completion and return the aggregated {@link RunResult}. */
|
||||
async run(messages: LLMMessage[], options: RunOptions = {}): Promise<RunResult> {
|
||||
let result: RunResult = {
|
||||
messages: [],
|
||||
output: '',
|
||||
toolCalls: [],
|
||||
tokenUsage: ZERO_USAGE,
|
||||
turns: 0,
|
||||
}
|
||||
for await (const event of this.stream(messages, options)) {
|
||||
if (event.type === 'done') {
|
||||
result = event.data as RunResult
|
||||
} else if (event.type === 'error') {
|
||||
throw event.data
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
/** Run one prompt turn, yielding `text` deltas then a terminal `done`/`error` event. */
|
||||
async *stream(messages: LLMMessage[], options: RunOptions = {}): AsyncGenerator<StreamEvent> {
|
||||
const label = this.options.agentName ?? 'agent'
|
||||
const userText = lastUserText(messages)
|
||||
if (!userText) {
|
||||
yield { type: 'error', data: new Error(`ACP backend "${label}" received an empty prompt.`) }
|
||||
return
|
||||
}
|
||||
|
||||
let session: ActiveSession
|
||||
let ctx: ClientContext
|
||||
try {
|
||||
const started = await this.ensureSession()
|
||||
session = started.session
|
||||
ctx = started.ctx
|
||||
} catch (err) {
|
||||
yield { type: 'error', data: toError(err) }
|
||||
return
|
||||
}
|
||||
|
||||
const abort = options.abortSignal
|
||||
const toolCalls: ToolCallRecord[] = []
|
||||
// Tool metadata seen on the initial `tool_call` event, finalized on the
|
||||
// completing `tool_call_update` (which may omit the title/input).
|
||||
const toolMeta = new Map<string, { start: number; title?: string; input?: Record<string, unknown> }>()
|
||||
const assistantText: string[] = []
|
||||
// The latest cumulative context reading seen this turn; undefined ⇒ the
|
||||
// agent sent no usage_update, so this turn reports zero usage.
|
||||
let usedThisTurn: number | undefined
|
||||
|
||||
// Wire abort → session/cancel; a well-behaved agent replies with a
|
||||
// `cancelled` stop, which ends the drain loop below.
|
||||
const onAbort = (): void => {
|
||||
void ctx
|
||||
.notify('session/cancel', { sessionId: session.sessionId } satisfies CancelNotification)
|
||||
.catch(() => {})
|
||||
}
|
||||
// If the caller already aborted before we start a turn, don't prompt at all —
|
||||
// there is no in-flight turn to cancel. Return empty output, mirroring a turn
|
||||
// that was cancelled immediately.
|
||||
if (abort?.aborted) {
|
||||
yield {
|
||||
type: 'done',
|
||||
data: { messages: [], output: '', toolCalls: [], tokenUsage: ZERO_USAGE, turns: 0 },
|
||||
}
|
||||
return
|
||||
}
|
||||
abort?.addEventListener('abort', onAbort, { once: true })
|
||||
|
||||
try {
|
||||
// ACP has no session-level system prompt, so convey the agent's role by
|
||||
// prepending it to the first prompt of the session (once). Later turns
|
||||
// reuse the same session, which already carries it.
|
||||
const promptText = this.systemPromptPending
|
||||
? `${this.systemPromptPending}\n\n${userText}`
|
||||
: userText
|
||||
this.systemPromptPending = undefined
|
||||
|
||||
// Fire the prompt; its completion is also queued as a `stop` message for
|
||||
// nextUpdate(). Attach a no-op catch now, then propagate after draining.
|
||||
const promptPromise = session.prompt(promptText)
|
||||
promptPromise.catch(() => {})
|
||||
|
||||
let stopReason: StopReason = 'end_turn'
|
||||
while (true) {
|
||||
const msg = await session.nextUpdate()
|
||||
if (msg.kind === 'stop') {
|
||||
stopReason = msg.stopReason
|
||||
break
|
||||
}
|
||||
const update = msg.update
|
||||
switch (update.sessionUpdate) {
|
||||
case 'agent_message_chunk': {
|
||||
if (update.content.type === 'text') {
|
||||
assistantText.push(update.content.text)
|
||||
yield { type: 'text', data: update.content.text }
|
||||
}
|
||||
break
|
||||
}
|
||||
case 'tool_call': {
|
||||
toolMeta.set(update.toolCallId, {
|
||||
start: Date.now(),
|
||||
title: update.title ?? undefined,
|
||||
input: asRecord(update.rawInput),
|
||||
})
|
||||
break
|
||||
}
|
||||
case 'tool_call_update': {
|
||||
if (update.status === 'completed' || update.status === 'failed') {
|
||||
const meta = toolMeta.get(update.toolCallId)
|
||||
toolCalls.push({
|
||||
toolName: update.title ?? meta?.title ?? update.toolCallId,
|
||||
input: update.rawInput !== undefined ? asRecord(update.rawInput) : meta?.input ?? {},
|
||||
output: toolOutputText(update.rawOutput, update.content),
|
||||
duration: Math.max(0, Date.now() - (meta?.start ?? Date.now())),
|
||||
})
|
||||
}
|
||||
break
|
||||
}
|
||||
case 'usage_update': {
|
||||
// ACP reports a single *cumulative* context-token figure ("tokens
|
||||
// currently in context"), not an input/output split or a per-turn
|
||||
// delta. Keep the latest reading; it is converted to an incremental
|
||||
// figure below (see docs/external-agents.md).
|
||||
usedThisTurn = update.used
|
||||
break
|
||||
}
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Surface a late prompt rejection (e.g. transport failure at turn end).
|
||||
await promptPromise
|
||||
|
||||
const output = assistantText.join('')
|
||||
|
||||
// A refusal is a task failure — dependents should not proceed on it.
|
||||
if (stopReason === 'refusal') {
|
||||
yield {
|
||||
type: 'error',
|
||||
data: new Error(`ACP agent "${label}" refused the task${output ? `: ${output}` : '.'}`),
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
const assistantMsg = output
|
||||
? ({ role: 'assistant', content: [{ type: 'text', text: output }] } satisfies LLMMessage)
|
||||
: undefined
|
||||
if (assistantMsg) options.onMessage?.(assistantMsg)
|
||||
|
||||
// Convert ACP's cumulative context reading into this turn's incremental
|
||||
// usage, so summing token usage across a reused session telescopes to the
|
||||
// final figure instead of double-counting. A turn with no usage_update
|
||||
// reports 0 and leaves the running total untouched.
|
||||
let inputTokens = 0
|
||||
if (usedThisTurn !== undefined) {
|
||||
inputTokens = Math.max(0, usedThisTurn - this.lastUsedTokens)
|
||||
this.lastUsedTokens = usedThisTurn
|
||||
}
|
||||
|
||||
// `max_tokens` / `max_turn_requests` mean the agent stopped before finishing;
|
||||
// reuse the "hit a limit, may be incomplete" signal so the task soft-fails.
|
||||
const budgetExceeded = stopReason === 'max_tokens' || stopReason === 'max_turn_requests'
|
||||
const result: RunResult = {
|
||||
messages: assistantMsg ? [assistantMsg] : [],
|
||||
output,
|
||||
toolCalls,
|
||||
tokenUsage: { input_tokens: inputTokens, output_tokens: 0 },
|
||||
turns: 1,
|
||||
...(budgetExceeded ? { budgetExceeded: true } : {}),
|
||||
}
|
||||
yield { type: 'done', data: result }
|
||||
} catch (err) {
|
||||
yield { type: 'error', data: toError(err) }
|
||||
} finally {
|
||||
if (abort) abort.removeEventListener('abort', onAbort)
|
||||
}
|
||||
}
|
||||
|
||||
/** Close the ACP connection and kill the subprocess. Safe to call more than once. */
|
||||
async dispose(): Promise<void> {
|
||||
try {
|
||||
this.session?.dispose()
|
||||
} catch {
|
||||
// best-effort
|
||||
}
|
||||
if (this.conn) {
|
||||
await this.conn.dispose()
|
||||
}
|
||||
this.session = null
|
||||
this.ctx = null
|
||||
this.conn = null
|
||||
this.starting = null
|
||||
// Reset per-session accounting: a fresh session restarts its cumulative
|
||||
// token reading from zero and re-prepends the role prompt on its first turn.
|
||||
this.lastUsedTokens = 0
|
||||
this.systemPromptPending = this.options.systemPrompt?.trim() || undefined
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Internals
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private async ensureSession(): Promise<{ session: ActiveSession; ctx: ClientContext }> {
|
||||
if (this.session && this.ctx) {
|
||||
return { session: this.session, ctx: this.ctx }
|
||||
}
|
||||
if (!this.starting) {
|
||||
this.starting = this.start()
|
||||
}
|
||||
try {
|
||||
await this.starting
|
||||
} catch (err) {
|
||||
// Clear the cached promise so a later run can retry, instead of re-throwing
|
||||
// this same rejection forever (e.g. after a transient spawn failure).
|
||||
this.starting = null
|
||||
throw err
|
||||
}
|
||||
return { session: this.session!, ctx: this.ctx! }
|
||||
}
|
||||
|
||||
private async start(): Promise<void> {
|
||||
const sdk = this.sdk ?? (this.sdk = await loadAcpSdk())
|
||||
const app = sdk
|
||||
.client({ name: this.options.agentName ?? 'open-multi-agent' })
|
||||
.onRequest('session/request_permission', ({ params }) => this.resolvePermission(params))
|
||||
|
||||
const connector = this.options.connect ?? defaultSpawnConnector(this.options)
|
||||
this.conn = await connector(app, sdk)
|
||||
this.ctx = this.conn.connection.agent
|
||||
|
||||
await this.ctx.request('initialize', {
|
||||
protocolVersion: sdk.PROTOCOL_VERSION,
|
||||
clientCapabilities: {},
|
||||
})
|
||||
this.session = await this.ctx.buildSession(resolveCwd(this.options.cwd)).start()
|
||||
}
|
||||
|
||||
private async resolvePermission(
|
||||
params: RequestPermissionRequest,
|
||||
): Promise<RequestPermissionResponse> {
|
||||
const options = params.options
|
||||
// Prefer the least-privilege option: a one-time `allow_once` / `reject_once`
|
||||
// over a session-wide `allow_always` / `reject_always`, so a single auto
|
||||
// decision doesn't silently broaden scope for the rest of the session.
|
||||
const allow = pick(options, 'allow_once') ?? pick(options, 'allow_always')
|
||||
const reject = pick(options, 'reject_once') ?? pick(options, 'reject_always')
|
||||
const policy = this.options.permission ?? 'auto-approve'
|
||||
|
||||
let approve: boolean
|
||||
if (policy === 'auto-approve') {
|
||||
approve = true
|
||||
} else if (policy === 'reject') {
|
||||
approve = false
|
||||
} else {
|
||||
const request: AcpPermissionRequest = {
|
||||
title: params.toolCall.title ?? '',
|
||||
kind: params.toolCall.kind ?? undefined,
|
||||
optionKinds: options.map((o) => o.kind),
|
||||
}
|
||||
approve = await policy(request)
|
||||
}
|
||||
|
||||
if (approve && allow) {
|
||||
return { outcome: { outcome: 'selected', optionId: allow.optionId } }
|
||||
}
|
||||
if (!approve && reject) {
|
||||
return { outcome: { outcome: 'selected', optionId: reject.optionId } }
|
||||
}
|
||||
return { outcome: { outcome: 'cancelled' } }
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Module helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async function loadAcpSdk(): Promise<AcpSdkModule> {
|
||||
try {
|
||||
return await import('@agentclientprotocol/sdk')
|
||||
} catch (err) {
|
||||
throw new Error(
|
||||
`The 'acp' agent backend requires the optional peer dependency ` +
|
||||
`'@agentclientprotocol/sdk'. Install it with: npm install @agentclientprotocol/sdk\n` +
|
||||
`Original error: ${err instanceof Error ? err.message : String(err)}`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/** Default connector: spawn the CLI and frame its stdio as newline-delimited JSON-RPC. */
|
||||
function defaultSpawnConnector(
|
||||
options: AcpBackendOptions,
|
||||
): (app: ClientApp, sdk: AcpSdkModule) => AcpConnection {
|
||||
return (app, sdk) => {
|
||||
const child = spawn(options.command, [...(options.args ?? [])], {
|
||||
cwd: resolveCwd(options.cwd),
|
||||
env: { ...process.env, ...options.env },
|
||||
stdio: ['pipe', 'pipe', 'inherit'],
|
||||
})
|
||||
if (!child.stdin || !child.stdout) {
|
||||
child.kill()
|
||||
throw new Error(`Failed to open stdio pipes for ACP agent command "${options.command}".`)
|
||||
}
|
||||
const stream = sdk.ndJsonStream(
|
||||
Writable.toWeb(child.stdin) as unknown as WritableStream<Uint8Array>,
|
||||
Readable.toWeb(child.stdout) as unknown as ReadableStream<Uint8Array>,
|
||||
)
|
||||
const connection = app.connect(stream)
|
||||
// If the process can't start (e.g. command not found), close the connection
|
||||
// so pending requests reject instead of hanging.
|
||||
child.once('error', () => {
|
||||
try {
|
||||
connection.close()
|
||||
} catch {
|
||||
// best-effort
|
||||
}
|
||||
})
|
||||
return {
|
||||
connection,
|
||||
dispose: () => {
|
||||
try {
|
||||
connection.close()
|
||||
} catch {
|
||||
// best-effort
|
||||
}
|
||||
if (!child.killed) child.kill()
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function pick(options: readonly PermissionOption[], kind: string): PermissionOption | undefined {
|
||||
return options.find((o) => o.kind === kind)
|
||||
}
|
||||
|
||||
/** Extract the text of the most recent user message (the new turn for the ACP session). */
|
||||
function lastUserText(messages: LLMMessage[]): string {
|
||||
for (let i = messages.length - 1; i >= 0; i--) {
|
||||
const message = messages[i]
|
||||
if (message.role !== 'user') continue
|
||||
return message.content
|
||||
.filter((b): b is Extract<ContentBlock, { type: 'text' }> => b.type === 'text')
|
||||
.map((b) => b.text)
|
||||
.join('\n')
|
||||
.trim()
|
||||
}
|
||||
return ''
|
||||
}
|
||||
|
||||
function asRecord(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === 'object' && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {}
|
||||
}
|
||||
|
||||
function toolOutputText(rawOutput: unknown, content: unknown): string {
|
||||
if (typeof rawOutput === 'string') return rawOutput
|
||||
if (rawOutput != null) return safeStringify(rawOutput)
|
||||
if (Array.isArray(content) && content.length > 0) return safeStringify(content)
|
||||
return ''
|
||||
}
|
||||
|
||||
function safeStringify(value: unknown): string {
|
||||
try {
|
||||
return JSON.stringify(value)
|
||||
} catch {
|
||||
return String(value)
|
||||
}
|
||||
}
|
||||
|
||||
function resolveCwd(cwd?: string): string {
|
||||
return resolvePath(cwd ?? process.cwd())
|
||||
}
|
||||
|
||||
function toError(err: unknown): Error {
|
||||
return err instanceof Error ? err : new Error(String(err))
|
||||
}
|
||||
@@ -37,6 +37,7 @@
|
||||
*/
|
||||
|
||||
import type {
|
||||
AgentBackendConfig,
|
||||
AgentConfig,
|
||||
AgentState,
|
||||
AgentRunResult,
|
||||
@@ -52,7 +53,7 @@ import type { ToolDefinition as FrameworkToolDefinition, ToolRegistry } from '..
|
||||
import type { ToolExecutor } from '../tool/executor.js'
|
||||
import { defaultWorkspaceDir } from '../tool/built-in/path-safety.js'
|
||||
import { createAdapter } from '../llm/adapter.js'
|
||||
import { AgentRunner, type RunnerOptions, type RunOptions, type RunResult } from './runner.js'
|
||||
import { AgentRunner, type AgentBackend, type RunnerOptions, type RunOptions, type RunResult } from './runner.js'
|
||||
import {
|
||||
buildStructuredOutputInstruction,
|
||||
extractJSON,
|
||||
@@ -84,7 +85,7 @@ export class Agent {
|
||||
readonly name: string
|
||||
readonly config: AgentConfig
|
||||
|
||||
private runner: AgentRunner | null = null
|
||||
private backend: AgentBackend | null = null
|
||||
private state: AgentState
|
||||
private readonly _toolRegistry: ToolRegistry
|
||||
private readonly _toolExecutor: ToolExecutor
|
||||
@@ -106,11 +107,12 @@ export class Agent {
|
||||
this.name = config.name
|
||||
// `model` is optional on AgentConfig so orchestrated agents can inherit
|
||||
// `OrchestratorConfig.defaultModel`. A standalone Agent has no orchestrator
|
||||
// to inherit from, so it must declare a model explicitly.
|
||||
if (config.model === undefined) {
|
||||
// to inherit from, so it must declare a model explicitly — unless it runs on
|
||||
// an external `backend` (e.g. an ACP coding CLI), which needs no LLM model.
|
||||
if (config.model === undefined && config.backend === undefined) {
|
||||
throw new Error(
|
||||
`Agent "${config.name}" has no model. Set 'model' in its config, or run it ` +
|
||||
`through OpenMultiAgent to inherit 'defaultModel'.`,
|
||||
`Agent "${config.name}" has no model. Set 'model' in its config, or a 'backend', ` +
|
||||
`or run it through OpenMultiAgent to inherit 'defaultModel'.`,
|
||||
)
|
||||
}
|
||||
this.config = config
|
||||
@@ -129,14 +131,24 @@ export class Agent {
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Lazily create the {@link AgentRunner}.
|
||||
* Lazily create the {@link AgentBackend} that executes this agent's runs.
|
||||
*
|
||||
* The adapter is created asynchronously (it may lazy-import provider SDKs),
|
||||
* so we defer construction until the first `run` / `prompt` / `stream` call.
|
||||
* Defaults to an {@link AgentRunner} (the LLM conversation loop). When
|
||||
* {@link AgentConfig.backend} is set, delegates to an external backend (e.g. an
|
||||
* ACP coding CLI) instead. Construction is async because it may lazy-import a
|
||||
* provider SDK or the optional ACP peer, so we defer it to the first
|
||||
* `run` / `prompt` / `stream` call.
|
||||
*/
|
||||
private async getRunner(): Promise<AgentRunner> {
|
||||
if (this.runner !== null) {
|
||||
return this.runner
|
||||
private async getBackend(): Promise<AgentBackend> {
|
||||
if (this.backend !== null) {
|
||||
return this.backend
|
||||
}
|
||||
|
||||
// External backend: it runs its own agentic loop, so none of the
|
||||
// LLM-specific configuration below applies.
|
||||
if (this.config.backend) {
|
||||
this.backend = await this.createExternalBackend(this.config.backend)
|
||||
return this.backend
|
||||
}
|
||||
|
||||
const provider = this.config.provider ?? 'anthropic'
|
||||
@@ -185,14 +197,45 @@ export class Agent {
|
||||
compressReasoningText: this.config.compressReasoningText,
|
||||
}
|
||||
|
||||
this.runner = new AgentRunner(
|
||||
this.backend = new AgentRunner(
|
||||
adapter,
|
||||
this._toolRegistry,
|
||||
this._toolExecutor,
|
||||
runnerOptions,
|
||||
)
|
||||
|
||||
return this.runner
|
||||
return this.backend
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a non-LLM {@link AgentBackend} from {@link AgentConfig.backend}.
|
||||
*
|
||||
* The backend module is loaded via dynamic `import()` so its optional peer SDK
|
||||
* (e.g. `@agentclientprotocol/sdk`) is only resolved when a backend is used.
|
||||
*/
|
||||
private async createExternalBackend(backend: AgentBackendConfig): Promise<AgentBackend> {
|
||||
switch (backend.kind) {
|
||||
case 'acp': {
|
||||
const { createAcpBackend } = await import('./acp-backend.js')
|
||||
return createAcpBackend({
|
||||
command: backend.command,
|
||||
args: backend.args,
|
||||
env: backend.env,
|
||||
cwd: backend.cwd,
|
||||
permission: backend.permission,
|
||||
// ACP has no system-prompt field; the backend prepends this to the
|
||||
// agent's first turn so its configured role reaches the external CLI.
|
||||
systemPrompt: this.config.systemPrompt,
|
||||
agentName: this.name,
|
||||
model: this.config.model,
|
||||
callTimeoutMs: this.config.callTimeoutMs,
|
||||
})
|
||||
}
|
||||
default:
|
||||
throw new Error(
|
||||
`Agent "${this.name}": unknown backend kind "${(backend as { kind: string }).kind}".`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
@@ -333,7 +376,7 @@ export class Agent {
|
||||
this.applyHookContext(messages, modified, hookCtx.prompt)
|
||||
}
|
||||
|
||||
const runner = await this.getRunner()
|
||||
const backend = await this.getBackend()
|
||||
const internalOnMessage = (msg: LLMMessage) => {
|
||||
this.state.messages.push(msg)
|
||||
callerOptions?.onMessage?.(msg)
|
||||
@@ -365,7 +408,7 @@ export class Agent {
|
||||
}
|
||||
effectiveTraceOptions = runOptions
|
||||
|
||||
const result = await runner.run(messages, runOptions)
|
||||
const result = await backend.run(messages, runOptions)
|
||||
this.state.tokenUsage = addUsage(this.state.tokenUsage, result.tokenUsage)
|
||||
|
||||
if (result.budgetExceeded) {
|
||||
@@ -383,7 +426,7 @@ export class Agent {
|
||||
let validated = await this.validateStructuredOutput(
|
||||
messages,
|
||||
result,
|
||||
runner,
|
||||
backend,
|
||||
runOptions,
|
||||
)
|
||||
// --- afterRun hook ---
|
||||
@@ -456,7 +499,7 @@ export class Agent {
|
||||
private async validateStructuredOutput(
|
||||
originalMessages: LLMMessage[],
|
||||
result: RunResult,
|
||||
runner: AgentRunner,
|
||||
backend: AgentBackend,
|
||||
runOptions: RunOptions,
|
||||
): Promise<AgentRunResult> {
|
||||
const schema = this.config.outputSchema!
|
||||
@@ -497,7 +540,7 @@ export class Agent {
|
||||
errorFeedbackMessage,
|
||||
]
|
||||
|
||||
const retryResult = await runner.run(retryMessages, runOptions)
|
||||
const retryResult = await backend.run(retryMessages, runOptions)
|
||||
this.state.tokenUsage = addUsage(this.state.tokenUsage, retryResult.tokenUsage)
|
||||
|
||||
const mergedTokenUsage = addUsage(result.tokenUsage, retryResult.tokenUsage)
|
||||
@@ -552,7 +595,7 @@ export class Agent {
|
||||
this.applyHookContext(messages, modified, hookCtx.prompt)
|
||||
}
|
||||
|
||||
const runner = await this.getRunner()
|
||||
const backend = await this.getBackend()
|
||||
// Fresh timeout per stream call, same as executeRun.
|
||||
const timeoutSignal = this.config.timeoutMs !== undefined && this.config.timeoutMs > 0
|
||||
? AbortSignal.timeout(this.config.timeoutMs)
|
||||
@@ -575,7 +618,7 @@ export class Agent {
|
||||
}
|
||||
effectiveTraceOptions = runOptions
|
||||
|
||||
for await (const event of runner.stream(messages, runOptions)) {
|
||||
for await (const event of backend.stream(messages, runOptions)) {
|
||||
if (event.type === 'done') {
|
||||
const result = event.data as RunResult
|
||||
this.state.tokenUsage = addUsage(this.state.tokenUsage, result.tokenUsage)
|
||||
|
||||
@@ -218,6 +218,25 @@ export interface RunResult {
|
||||
readonly budgetExceeded?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* The execution seam behind every {@link Agent}: given a conversation, produce a
|
||||
* {@link RunResult} (and a matching {@link StreamEvent} stream). {@link AgentRunner}
|
||||
* is the LLM implementation; alternative backends (e.g. an external coding agent
|
||||
* over ACP — see `@open-multi-agent/core/acp`) implement the same contract so an
|
||||
* `Agent` can drive either without the orchestrator, pool, or team knowing which.
|
||||
*
|
||||
* The contract mirrors {@link AgentRunner.run} / {@link AgentRunner.stream}:
|
||||
* `stream()` yields incremental events and MUST end with a single
|
||||
* `{ type: 'done', data: RunResult }` (or `{ type: 'error', data }` on failure);
|
||||
* `run()` returns that same aggregated {@link RunResult}.
|
||||
*/
|
||||
export interface AgentBackend {
|
||||
/** Run the conversation to completion and return the aggregated result. */
|
||||
run(messages: LLMMessage[], options?: RunOptions): Promise<RunResult>
|
||||
/** Run the conversation and yield {@link StreamEvent}s, ending with `done`. */
|
||||
stream(messages: LLMMessage[], options?: RunOptions): AsyncIterable<StreamEvent>
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Internal helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -377,7 +396,7 @@ function loopWarningText(kind: 'tool_repetition' | 'text_repetition'): string {
|
||||
* console.log(result.output)
|
||||
* ```
|
||||
*/
|
||||
export class AgentRunner {
|
||||
export class AgentRunner implements AgentBackend {
|
||||
private readonly maxTurns: number
|
||||
private summarizeCache: {
|
||||
oldSignature: string
|
||||
|
||||
@@ -65,6 +65,9 @@ export { renderTeamRunDashboard } from './dashboard/render-team-run-dashboard.js
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export { Agent } from './agent/agent.js'
|
||||
// The execution seam behind every Agent. `AgentRunner` (the LLM loop) implements
|
||||
// it; the ACP backend (`@open-multi-agent/core/acp`) is an alternative impl.
|
||||
export type { AgentBackend, RunOptions, RunResult } from './agent/runner.js'
|
||||
export { LoopDetector } from './agent/loop-detector.js'
|
||||
export { buildStructuredOutputInstruction, extractJSON, validateOutput } from './agent/structured-output.js'
|
||||
export { AgentPool, Semaphore } from './agent/pool.js'
|
||||
@@ -167,6 +170,9 @@ export type {
|
||||
AgentConfig,
|
||||
AgentState,
|
||||
AgentRunResult,
|
||||
AgentBackendConfig,
|
||||
AcpPermissionPolicy,
|
||||
AcpPermissionRequest,
|
||||
BeforeRunHookContext,
|
||||
ToolCallRecord,
|
||||
LoopDetectionConfig,
|
||||
|
||||
@@ -400,6 +400,62 @@ export interface BeforeRunHookContext {
|
||||
readonly agent: AgentConfig
|
||||
}
|
||||
|
||||
/**
|
||||
* A minimal, SDK-agnostic view of an ACP `session/request_permission` prompt,
|
||||
* passed to a {@link AgentBackendConfig.permission} callback so callers can decide
|
||||
* without importing `@agentclientprotocol/sdk`.
|
||||
*/
|
||||
export interface AcpPermissionRequest {
|
||||
/** Human-readable title of the pending tool call (e.g. `"Edit src/app.ts"`). */
|
||||
readonly title: string
|
||||
/** ACP tool kind when the agent provides one (e.g. `'edit'`, `'execute'`, `'read'`). */
|
||||
readonly kind?: string
|
||||
/** The option kinds the agent offered (e.g. `'allow_once'`, `'reject_always'`). */
|
||||
readonly optionKinds: readonly string[]
|
||||
}
|
||||
|
||||
/**
|
||||
* How an ACP-backed agent answers a permission prompt. OMA runs agents
|
||||
* autonomously inside a task DAG, so the default is `'auto-approve'`.
|
||||
* - `'auto-approve'` — select an `allow_*` option, preferring the least-privilege
|
||||
* `allow_once` over a session-wide `allow_always` (falls back to cancel).
|
||||
* - `'reject'` — select a `reject_*` option, preferring `reject_once` (falls back to cancel).
|
||||
* - function — decide per request; `true` approves, `false` rejects.
|
||||
*/
|
||||
export type AcpPermissionPolicy =
|
||||
| 'auto-approve'
|
||||
| 'reject'
|
||||
| ((request: AcpPermissionRequest) => boolean | Promise<boolean>)
|
||||
|
||||
/**
|
||||
* Configuration for running an external agent over the Agent Client Protocol
|
||||
* (ACP) as an OMA team member — see {@link AgentConfig.backend}. The agent is a
|
||||
* local subprocess (a coding CLI such as Gemini CLI or Claude Code) that runs its
|
||||
* own agentic loop, while OMA drives it, collects its output and token usage, and
|
||||
* schedules it in the task DAG alongside LLM agents.
|
||||
*
|
||||
* A discriminated union keyed by {@link kind}; only `'acp'` exists today.
|
||||
* Requires the optional peer `@agentclientprotocol/sdk`.
|
||||
*/
|
||||
export interface AgentBackendConfig {
|
||||
/** Backend discriminant. Only `'acp'` is supported today. */
|
||||
readonly kind: 'acp'
|
||||
/** Executable to spawn (e.g. `'npx'`, `'gemini'`, `'codex-acp'`). */
|
||||
readonly command: string
|
||||
/** Arguments passed to `command` (e.g. `['-y', '@agentclientprotocol/claude-agent-acp']`). */
|
||||
readonly args?: readonly string[]
|
||||
/** Extra environment variables for the subprocess, merged over `process.env`. */
|
||||
readonly env?: Readonly<Record<string, string>>
|
||||
/**
|
||||
* Working directory the agent reads and edits. Defaults to `process.cwd()`.
|
||||
* Unlike OMA's filesystem-tool sandbox, an ACP agent accesses this directory
|
||||
* directly — scope it to a project you trust the external agent with.
|
||||
*/
|
||||
readonly cwd?: string
|
||||
/** How to answer the agent's permission prompts. Defaults to `'auto-approve'`. */
|
||||
readonly permission?: AcpPermissionPolicy
|
||||
}
|
||||
|
||||
/** Static configuration for a single agent. */
|
||||
export interface AgentConfig {
|
||||
readonly name: string
|
||||
@@ -419,6 +475,16 @@ export interface AgentConfig {
|
||||
* and `region` are ignored for LLM calls.
|
||||
*/
|
||||
readonly adapter?: LLMAdapter
|
||||
/**
|
||||
* Run this agent on an external {@link AgentBackendConfig} (e.g. a coding CLI
|
||||
* over the Agent Client Protocol) instead of an LLM adapter. When set, the
|
||||
* LLM-specific fields (`model`, `provider`, `adapter`, sampling, tools, context
|
||||
* strategy) do not apply — the external agent runs its own loop — but the agent
|
||||
* still participates in the task DAG, shared memory, cascade-on-failure, and
|
||||
* token budget like any other team member. Requires the optional peer
|
||||
* `@agentclientprotocol/sdk`; import the backend from `@open-multi-agent/core/acp`.
|
||||
*/
|
||||
readonly backend?: AgentBackendConfig
|
||||
readonly provider?: SupportedProvider
|
||||
/**
|
||||
* Custom base URL for OpenAI-compatible APIs (Ollama, vLLM, LM Studio, etc.).
|
||||
|
||||
@@ -0,0 +1,429 @@
|
||||
/**
|
||||
* Tests for the ACP agent backend (`src/agent/acp-backend.ts`).
|
||||
*
|
||||
* The unit tests wire the backend to an in-process fake ACP *agent* built with
|
||||
* the SDK's `agent()` helper (via the backend's `connect` test seam), so no
|
||||
* subprocess or real coding CLI is needed. One integration test spawns a real
|
||||
* subprocess (`fixtures/fake-acp-agent.mjs`) through the public `Agent` API to
|
||||
* exercise the full spawn → stdio JSON-RPC → RunResult path.
|
||||
*/
|
||||
import { describe, it, expect } from 'vitest'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { agent, PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
|
||||
import type { AgentApp, AgentRequestHandlersByMethod } from '@agentclientprotocol/sdk'
|
||||
|
||||
import { AcpBackend, type AcpBackendOptions } from '../src/agent/acp-backend.js'
|
||||
import { Agent } from '../src/agent/agent.js'
|
||||
import { ToolRegistry } from '../src/tool/framework.js'
|
||||
import { ToolExecutor } from '../src/tool/executor.js'
|
||||
import type { AcpPermissionPolicy, LLMMessage } from '../src/types.js'
|
||||
|
||||
type PromptHandler = AgentRequestHandlersByMethod['session/prompt']
|
||||
|
||||
function userMsg(text: string): LLMMessage {
|
||||
return { role: 'user', content: [{ type: 'text', text }] }
|
||||
}
|
||||
|
||||
/** Extract the text an ACP agent received in a `session/prompt` turn. */
|
||||
function promptText(prompt: unknown): string {
|
||||
const blocks = (Array.isArray(prompt) ? prompt : [prompt]) as Array<{ type?: string; text?: string }>
|
||||
return blocks.find((b) => b?.type === 'text')?.text ?? ''
|
||||
}
|
||||
|
||||
/** Build a fake ACP agent whose prompt turn is driven by `promptHandler`. */
|
||||
function fakeAgent(promptHandler: PromptHandler): AgentApp {
|
||||
return agent({ name: 'fake' })
|
||||
.onRequest('initialize', () => ({ protocolVersion: PROTOCOL_VERSION, agentCapabilities: {} }))
|
||||
.onRequest('session/new', () => ({ sessionId: 'test-session' }))
|
||||
.onRequest('session/prompt', promptHandler)
|
||||
}
|
||||
|
||||
/** An {@link AcpBackend} connected in-process to `fake` (no subprocess). */
|
||||
function backendFor(
|
||||
fake: AgentApp,
|
||||
permission?: AcpPermissionPolicy,
|
||||
extra?: Partial<AcpBackendOptions>,
|
||||
): AcpBackend {
|
||||
return new AcpBackend({
|
||||
command: 'unused',
|
||||
agentName: 'fake',
|
||||
...(permission ? { permission } : {}),
|
||||
...extra,
|
||||
connect: (clientApp) => {
|
||||
const connection = clientApp.connect(fake)
|
||||
return {
|
||||
connection,
|
||||
dispose: () => {
|
||||
try {
|
||||
connection.close()
|
||||
} catch {
|
||||
// best-effort
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
describe('AcpBackend (in-process fake agent)', () => {
|
||||
it('maps agent_message_chunk text and usage_update into a RunResult', async () => {
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Hello ' } },
|
||||
})
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'world' } },
|
||||
})
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'usage_update', used: 1234, size: 200000 },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}),
|
||||
)
|
||||
try {
|
||||
const result = await backend.run([userMsg('hi')])
|
||||
expect(result.output).toBe('Hello world')
|
||||
expect(result.tokenUsage).toEqual({ input_tokens: 1234, output_tokens: 0 })
|
||||
expect(result.turns).toBe(1)
|
||||
expect(result.budgetExceeded).toBeUndefined()
|
||||
expect(result.messages).toHaveLength(1)
|
||||
expect(result.messages[0]?.role).toBe('assistant')
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('streams text deltas and ends with a done event carrying the RunResult', async () => {
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'a' } },
|
||||
})
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'b' } },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}),
|
||||
)
|
||||
try {
|
||||
const deltas: string[] = []
|
||||
let done = false
|
||||
for await (const event of backend.stream([userMsg('hi')])) {
|
||||
if (event.type === 'text') deltas.push(event.data as string)
|
||||
else if (event.type === 'done') done = true
|
||||
}
|
||||
expect(deltas).toEqual(['a', 'b'])
|
||||
expect(done).toBe(true)
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('records tool calls, carrying the title from tool_call to its completion', async () => {
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: {
|
||||
sessionUpdate: 'tool_call',
|
||||
toolCallId: 't1',
|
||||
title: 'Edit file',
|
||||
kind: 'edit',
|
||||
status: 'pending',
|
||||
rawInput: { path: 'a.ts' },
|
||||
},
|
||||
})
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: {
|
||||
sessionUpdate: 'tool_call_update',
|
||||
toolCallId: 't1',
|
||||
status: 'completed',
|
||||
rawOutput: 'patched',
|
||||
},
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}),
|
||||
)
|
||||
try {
|
||||
const result = await backend.run([userMsg('edit a.ts')])
|
||||
expect(result.toolCalls).toHaveLength(1)
|
||||
expect(result.toolCalls[0]).toMatchObject({
|
||||
toolName: 'Edit file',
|
||||
input: { path: 'a.ts' },
|
||||
output: 'patched',
|
||||
})
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('auto-approves permission requests by selecting an allow option', async () => {
|
||||
const backend = backendFor(fakeAgent(reportPermissionChoice))
|
||||
try {
|
||||
const result = await backend.run([userMsg('go')])
|
||||
expect(result.output).toBe('allow')
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('auto-approve prefers the least-privilege allow_once over allow_always', async () => {
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
const response = await client.request('session/request_permission', {
|
||||
sessionId: params.sessionId,
|
||||
toolCall: { toolCallId: 't1', title: 'Edit', kind: 'edit', status: 'pending' },
|
||||
options: [
|
||||
// `allow_always` is listed first on purpose.
|
||||
{ optionId: 'always', name: 'Always', kind: 'allow_always' },
|
||||
{ optionId: 'once', name: 'Once', kind: 'allow_once' },
|
||||
],
|
||||
})
|
||||
const chosen =
|
||||
response.outcome.outcome === 'selected' ? response.outcome.optionId : 'cancelled'
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: chosen } },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}),
|
||||
)
|
||||
try {
|
||||
const result = await backend.run([userMsg('go')])
|
||||
// Auto-approve must pick `allow_once`, not the first (`allow_always`) option.
|
||||
expect(result.output).toBe('once')
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it("rejects permission requests under the 'reject' policy", async () => {
|
||||
const backend = backendFor(fakeAgent(reportPermissionChoice), 'reject')
|
||||
try {
|
||||
const result = await backend.run([userMsg('go')])
|
||||
expect(result.output).toBe('reject')
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('honors a custom permission function', async () => {
|
||||
const seen: string[] = []
|
||||
const policy: AcpPermissionPolicy = (req) => {
|
||||
seen.push(req.title)
|
||||
return false
|
||||
}
|
||||
const backend = backendFor(fakeAgent(reportPermissionChoice), policy)
|
||||
try {
|
||||
const result = await backend.run([userMsg('go')])
|
||||
expect(result.output).toBe('reject')
|
||||
expect(seen).toEqual(['Run cmd'])
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('throws when the agent refuses the turn', async () => {
|
||||
const backend = backendFor(fakeAgent(async () => ({ stopReason: 'refusal' })))
|
||||
try {
|
||||
await expect(backend.run([userMsg('x')])).rejects.toThrow(/refused/i)
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('flags budgetExceeded when the agent stops on max_tokens', async () => {
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'partial' } },
|
||||
})
|
||||
return { stopReason: 'max_tokens' }
|
||||
}),
|
||||
)
|
||||
try {
|
||||
const result = await backend.run([userMsg('x')])
|
||||
expect(result.budgetExceeded).toBe(true)
|
||||
expect(result.output).toBe('partial')
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('errors on an empty prompt', async () => {
|
||||
const backend = backendFor(fakeAgent(async () => ({ stopReason: 'end_turn' })))
|
||||
try {
|
||||
await expect(backend.run([])).rejects.toThrow(/empty prompt/i)
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('reports token usage as a per-turn delta across a reused session', async () => {
|
||||
// `usage_update.used` is cumulative context occupancy, so a reused session
|
||||
// reports a growing total. The backend must record each turn's increment,
|
||||
// not the running total, or summing across turns double-counts.
|
||||
const readings = [1000, 2500, 3000]
|
||||
let turn = 0
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'usage_update', used: readings[turn++]!, size: 200000 },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}),
|
||||
)
|
||||
try {
|
||||
const first = await backend.run([userMsg('turn 1')])
|
||||
const second = await backend.run([userMsg('turn 2')])
|
||||
const third = await backend.run([userMsg('turn 3')])
|
||||
expect(first.tokenUsage.input_tokens).toBe(1000)
|
||||
expect(second.tokenUsage.input_tokens).toBe(1500) // 2500 - 1000
|
||||
expect(third.tokenUsage.input_tokens).toBe(500) // 3000 - 2500
|
||||
// The sum telescopes to the latest cumulative reading (3000), not 6500.
|
||||
const total =
|
||||
first.tokenUsage.input_tokens +
|
||||
second.tokenUsage.input_tokens +
|
||||
third.tokenUsage.input_tokens
|
||||
expect(total).toBe(3000)
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('prepends systemPrompt to the first turn only, then reuses the session', async () => {
|
||||
const seen: string[] = []
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
seen.push(promptText(params.prompt))
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'ok' } },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}),
|
||||
undefined,
|
||||
{ systemPrompt: 'You are a coder.' },
|
||||
)
|
||||
try {
|
||||
await backend.run([userMsg('do X')])
|
||||
await backend.run([userMsg('do Y')])
|
||||
expect(seen[0]).toBe('You are a coder.\n\ndo X')
|
||||
expect(seen[1]).toBe('do Y')
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('does not start a turn when the abort signal is already aborted', async () => {
|
||||
let prompted = false
|
||||
const backend = backendFor(
|
||||
fakeAgent(async ({ params, client }) => {
|
||||
prompted = true
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'x' } },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}),
|
||||
)
|
||||
try {
|
||||
const result = await backend.run([userMsg('go')], { abortSignal: AbortSignal.abort() })
|
||||
expect(result.output).toBe('')
|
||||
expect(prompted).toBe(false)
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('retries session startup after a failed start instead of caching the rejection', async () => {
|
||||
let attempts = 0
|
||||
const fake = fakeAgent(async ({ params, client }) => {
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'ok' } },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
})
|
||||
const backend = new AcpBackend({
|
||||
command: 'unused',
|
||||
agentName: 'fake',
|
||||
connect: (clientApp) => {
|
||||
attempts++
|
||||
if (attempts === 1) throw new Error('spawn failed')
|
||||
const connection = clientApp.connect(fake)
|
||||
return {
|
||||
connection,
|
||||
dispose: () => {
|
||||
try {
|
||||
connection.close()
|
||||
} catch {
|
||||
// best-effort
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
})
|
||||
try {
|
||||
await expect(backend.run([userMsg('go')])).rejects.toThrow(/spawn failed/)
|
||||
const result = await backend.run([userMsg('go')])
|
||||
expect(result.output).toBe('ok')
|
||||
expect(attempts).toBe(2)
|
||||
} finally {
|
||||
await backend.dispose()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('AcpBackend via Agent (subprocess integration)', () => {
|
||||
it('runs a real ACP agent subprocess through Agent.run', async () => {
|
||||
const fixture = fileURLToPath(new URL('./fixtures/fake-acp-agent.mjs', import.meta.url))
|
||||
const registry = new ToolRegistry()
|
||||
const executor = new ToolExecutor(registry)
|
||||
const agentInstance = new Agent(
|
||||
{ name: 'coder', backend: { kind: 'acp', command: process.execPath, args: [fixture] } },
|
||||
registry,
|
||||
executor,
|
||||
)
|
||||
try {
|
||||
const result = await agentInstance.run('build the thing')
|
||||
expect(result.success).toBe(true)
|
||||
expect(result.output).toBe('done: build the thing')
|
||||
expect(result.tokenUsage.input_tokens).toBe(42)
|
||||
} finally {
|
||||
// Reach into the Agent's cached backend to tear the subprocess down.
|
||||
await (agentInstance as unknown as { backend?: AcpBackend }).backend?.dispose()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
/**
|
||||
* Prompt handler that asks the client to approve a tool call, then reports the
|
||||
* selected option id (or `cancelled`) back as its assistant message.
|
||||
*/
|
||||
const reportPermissionChoice: PromptHandler = async ({ params, client }) => {
|
||||
const response = await client.request('session/request_permission', {
|
||||
sessionId: params.sessionId,
|
||||
toolCall: { toolCallId: 't1', title: 'Run cmd', kind: 'execute', status: 'pending' },
|
||||
options: [
|
||||
{ optionId: 'allow', name: 'Allow', kind: 'allow_once' },
|
||||
{ optionId: 'reject', name: 'Reject', kind: 'reject_once' },
|
||||
],
|
||||
})
|
||||
const chosen =
|
||||
response.outcome.outcome === 'selected' ? response.outcome.optionId : 'cancelled'
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: chosen } },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
}
|
||||
@@ -50,7 +50,7 @@ function buildMockAgent(config: AgentConfig, responseText: string) {
|
||||
temperature: config.temperature,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
|
||||
return { agent, calls }
|
||||
}
|
||||
@@ -330,7 +330,7 @@ describe('Agent hooks — beforeRun / afterRun', () => {
|
||||
model: config.model,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
|
||||
// Directly call run which creates a single text-only user message.
|
||||
// To test mixed content, we need to go through the private executeRun.
|
||||
|
||||
@@ -48,7 +48,7 @@ function buildMockAgent(config: AgentConfig, responseText: string): Agent {
|
||||
temperature: config.temperature,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
return agent
|
||||
}
|
||||
|
||||
|
||||
+39
@@ -0,0 +1,39 @@
|
||||
// Minimal ACP *agent* over stdio, used by the acp-backend integration test.
|
||||
// Echoes the incoming prompt text back as an assistant message, reports a fixed
|
||||
// token usage, then ends the turn. Hermetic: no network, no real model.
|
||||
import { agent, ndJsonStream, PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
|
||||
import { Readable, Writable } from 'node:stream'
|
||||
|
||||
function firstText(prompt) {
|
||||
const blocks = Array.isArray(prompt) ? prompt : [prompt]
|
||||
const block = blocks.find((b) => b && b.type === 'text')
|
||||
return block ? block.text : ''
|
||||
}
|
||||
|
||||
const app = agent({ name: 'fake-acp-agent' })
|
||||
.onRequest('initialize', () => ({
|
||||
protocolVersion: PROTOCOL_VERSION,
|
||||
agentCapabilities: {},
|
||||
}))
|
||||
.onRequest('session/new', () => ({ sessionId: 'test-session' }))
|
||||
.onRequest('session/prompt', async ({ params, client }) => {
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: {
|
||||
sessionUpdate: 'agent_message_chunk',
|
||||
content: { type: 'text', text: `done: ${firstText(params.prompt)}` },
|
||||
},
|
||||
})
|
||||
await client.notify('session/update', {
|
||||
sessionId: params.sessionId,
|
||||
update: { sessionUpdate: 'usage_update', used: 42, size: 1000 },
|
||||
})
|
||||
return { stopReason: 'end_turn' }
|
||||
})
|
||||
|
||||
app.connect(
|
||||
ndJsonStream(
|
||||
Writable.toWeb(process.stdout),
|
||||
Readable.toWeb(process.stdin),
|
||||
),
|
||||
)
|
||||
@@ -49,7 +49,7 @@ function buildMockAgent(config: AgentConfig, responses: string[]): Agent {
|
||||
temperature: config.temperature,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
|
||||
return agent
|
||||
}
|
||||
|
||||
@@ -57,7 +57,7 @@ function buildMockAgent(config: AgentConfig, responseText: string) {
|
||||
maxTurns: config.maxTurns,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
|
||||
return { agent, calls }
|
||||
}
|
||||
|
||||
@@ -55,7 +55,7 @@ function buildMockAgent(config: AgentConfig, responses: string[]): Agent {
|
||||
temperature: config.temperature,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
return agent
|
||||
}
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ function buildAgentWithAdapter(config: AgentConfig, adapter: LLMAdapter) {
|
||||
maxTurns: config.maxTurns,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
|
||||
return agent
|
||||
}
|
||||
|
||||
@@ -196,7 +196,7 @@ function buildMockAgentWithAdapter(config: AgentConfig, adapter: LLMAdapter): Ag
|
||||
const executor = new ToolExecutor(registry)
|
||||
const agent = new Agent(config, registry, executor)
|
||||
|
||||
// Inject a pre-built runner so `getRunner()` returns it without calling createAdapter.
|
||||
// Inject a pre-built runner so `getBackend()` returns it without calling createAdapter.
|
||||
const runner = new AgentRunner(adapter, registry, executor, {
|
||||
model: config.model,
|
||||
systemPrompt: config.systemPrompt,
|
||||
@@ -205,7 +205,7 @@ function buildMockAgentWithAdapter(config: AgentConfig, adapter: LLMAdapter): Ag
|
||||
temperature: config.temperature,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
|
||||
return agent
|
||||
}
|
||||
|
||||
@@ -78,7 +78,7 @@ function buildMockAgent(
|
||||
temperature: config.temperature,
|
||||
agentName: config.name,
|
||||
})
|
||||
;(agent as any).runner = runner
|
||||
;(agent as any).backend = runner
|
||||
|
||||
return agent
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user