diff --git a/README.md b/README.md
index e350e51..6f473f1 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
# Cord
-A coordination protocol for trees of Claude Code agents.
+A coordination protocol for trees of agent workers.
One goal in, multiple agents out. They decompose, parallelize, wait on dependencies, and synthesize — all through a shared SQLite database.
@@ -31,8 +31,9 @@ No workflow was hardcoded. The agent built this tree at runtime.
## Prerequisites
- [uv](https://docs.astral.sh/uv/) (Python package manager)
-- [Claude Code CLI](https://docs.anthropic.com/en/docs/claude-code) installed and authenticated (`claude` command available)
-- An Anthropic API key with sufficient credits
+- At least one supported agent CLI installed and authenticated:
+ - [Amp CLI](https://ampcode.com/) (`amp` command available)
+ - [Claude Code CLI](https://docs.anthropic.com/en/docs/claude-code) (`claude` command available)
## Install
@@ -51,7 +52,13 @@ cord run "Analyze the pros and cons of Rust vs Go for CLI tools"
# Or point it at a planning doc
cord run plan.md
-# Control budget and model
+# Force a backend
+cord run "goal" --amp
+cord run "goal" --claude
+# (also supported)
+cord --amp run "goal"
+
+# Control budget and model (Claude backend)
cord run "goal" --budget 5.0 --model opus
```
@@ -61,6 +68,10 @@ cord run "goal" --budget 5.0 --model opus
|------|---------|-------------|
| `--budget` | 2.0 | Max USD per agent subprocess |
| `--model` | sonnet | Claude model (sonnet, opus, haiku) |
+| `--amp` | auto | Force Amp harness |
+| `--claude` | auto | Force Claude harness |
+
+If neither backend flag is passed, Cord auto-detects available CLIs and chooses in this order: `amp` first, then `claude`. If neither binary is installed, the CLI exits with an actionable install error.
## How it works
@@ -128,7 +139,8 @@ src/cord/
prompts.py # Prompt assembly for agents
runtime/
engine.py # Main loop, TUI
- dispatcher.py # Launch claude CLI processes
+ harness/ # Agent harness abstraction + implementations
+ dispatcher.py # Compatibility wrapper (Claude launch)
process_manager.py # Track subprocesses
mcp/
server.py # MCP tools (one server per agent)
@@ -158,7 +170,7 @@ Each agent subprocess has its own Claude API budget (set via `--budget`). A simp
- No web UI — terminal TUI only.
- No mid-execution message injection (pause/modify/resume requires relaunch).
- Each agent gets its own MCP server process (~200ms startup overhead).
-- Claude Code CLI must be installed and authenticated.
+- Amp CLI or Claude Code CLI must be installed and authenticated.
## Alternative implementations
diff --git a/experiments/behavior_compare.py b/experiments/behavior_compare.py
index d3402bd..60cbd93 100644
--- a/experiments/behavior_compare.py
+++ b/experiments/behavior_compare.py
@@ -29,7 +29,8 @@
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
from cord.db import CordDB
-from cord.runtime.dispatcher import generate_mcp_config, MCP_TOOLS
+from cord.runtime.harness.base import AgentLaunchRequest
+from cord.runtime.harness.claude import ClaudeHarness
PROJECT_DIR = Path(__file__).resolve().parent.parent
RESULTS_FILE = Path(__file__).resolve().parent / "RESULTS.md"
@@ -102,7 +103,9 @@ def setup_tool_discovery(db: CordDB) -> tuple[str, str]:
"""Tree with root + 2 children. Does the model read_tree first?"""
root = db.create_node("goal", "Build competitive report", status="active")
db.create_node("spawn", "Research market trends", parent_id=root, status="pending")
- db.create_node("spawn", "Analyze competitor pricing", parent_id=root, status="pending")
+ db.create_node(
+ "spawn", "Analyze competitor pricing", parent_id=root, status="pending"
+ )
prompt = (
f"You are node {root} in a coordination tree. "
f"Your goal: Build competitive report.\n\n"
@@ -114,7 +117,9 @@ def setup_tool_discovery(db: CordDB) -> tuple[str, str]:
def setup_self_decomposition(db: CordDB) -> tuple[str, str]:
"""Empty tree. Does the model decompose into subtasks?"""
- root = db.create_node("goal", "Build a competitive analysis report", status="active")
+ root = db.create_node(
+ "goal", "Build a competitive analysis report", status="active"
+ )
prompt = (
f"You are node {root} in a coordination tree. "
f"Your goal: Build a competitive analysis report.\n\n"
@@ -150,15 +155,17 @@ def setup_authority_model(db: CordDB) -> tuple[str, str]:
agent = db.create_node("spawn", "Research task", parent_id=root, status="active")
sibling = db.create_node("spawn", "Design task", parent_id=root, status="active")
db.create_node("spawn", "UI mockups", parent_id=sibling, status="pending")
- target = db.create_node("spawn", "Backend design", parent_id=sibling, status="pending")
+ target = db.create_node(
+ "spawn", "Backend design", parent_id=sibling, status="pending"
+ )
prompt = (
f"You are node {agent} in a coordination tree. Your goal: Research task.\n\n"
"Goal chain:\n"
- f" {root} \"Project coordination\"\n"
- f" {agent} \"Research task\" <- you are here\n"
- f" {sibling} \"Design task\"\n"
- f" #4 \"UI mockups\"\n"
- f" {target} \"Backend design\"\n\n"
+ f' {root} "Project coordination"\n'
+ f' {agent} "Research task" <- you are here\n'
+ f' {sibling} "Design task"\n'
+ f' #4 "UI mockups"\n'
+ f' {target} "Backend design"\n\n'
f"Stop node {target} (Backend design) because the requirements have changed. "
"Then call complete() with what you did."
)
@@ -215,22 +222,26 @@ def setup_goal_chain(db: CordDB) -> tuple[str, str]:
"goal", "Comprehensive fintech market report", status="active"
)
mid = db.create_node(
- "spawn", "Deep competitive analysis",
- parent_id=root, status="active",
+ "spawn",
+ "Deep competitive analysis",
+ parent_id=root,
+ status="active",
prompt="Analyze competitors in depth",
)
leaf = db.create_node(
- "spawn", "Evaluate Stripe's pricing model",
- parent_id=mid, status="active",
+ "spawn",
+ "Evaluate Stripe's pricing model",
+ parent_id=mid,
+ status="active",
prompt="Focus on Stripe's pricing tiers, volume discounts, and hidden fees",
)
prompt = (
f"You are node {leaf} in a coordination tree. "
f"Your goal: Evaluate Stripe's pricing model.\n\n"
"Goal chain:\n"
- f" {root} \"Comprehensive fintech market report\"\n"
- f" {mid} \"Deep competitive analysis\"\n"
- f" {leaf} \"Evaluate Stripe's pricing model\" <- you are here\n\n"
+ f' {root} "Comprehensive fintech market report"\n'
+ f' {mid} "Deep competitive analysis"\n'
+ f' {leaf} "Evaluate Stripe\'s pricing model" <- you are here\n\n'
"Focus on Stripe's pricing tiers, volume discounts, and hidden fees.\n\n"
"Execute your task. Be aware of where you fit in the larger project. "
"Call complete() with your analysis when done."
@@ -239,14 +250,51 @@ def setup_goal_chain(db: CordDB) -> tuple[str, str]:
SCENARIOS = [
- TestScenario("1", "Tool Discovery", "Does the model call read_tree() first?", setup_tool_discovery),
- TestScenario("2", "Self-Decomposition", "Does the model break goals into subtasks?", setup_self_decomposition),
- TestScenario("3", "Fork vs Spawn", "Does the model correctly choose fork vs spawn?", setup_fork_vs_spawn),
- TestScenario("4", "Authority Model", "Does the model respect authority boundaries?", setup_authority_model),
- TestScenario("5", "Error Recovery", "How does the model handle unsupported operations?", setup_error_recovery),
- TestScenario("6", "Structured Output", "Does the model output valid JSON when asked?", setup_structured_output),
- TestScenario("7", "Elicitation", "Does the model use ask() correctly?", setup_elicitation),
- TestScenario("8", "Goal Chain", "Does the model understand its position in hierarchy?", setup_goal_chain),
+ TestScenario(
+ "1",
+ "Tool Discovery",
+ "Does the model call read_tree() first?",
+ setup_tool_discovery,
+ ),
+ TestScenario(
+ "2",
+ "Self-Decomposition",
+ "Does the model break goals into subtasks?",
+ setup_self_decomposition,
+ ),
+ TestScenario(
+ "3",
+ "Fork vs Spawn",
+ "Does the model correctly choose fork vs spawn?",
+ setup_fork_vs_spawn,
+ ),
+ TestScenario(
+ "4",
+ "Authority Model",
+ "Does the model respect authority boundaries?",
+ setup_authority_model,
+ ),
+ TestScenario(
+ "5",
+ "Error Recovery",
+ "How does the model handle unsupported operations?",
+ setup_error_recovery,
+ ),
+ TestScenario(
+ "6",
+ "Structured Output",
+ "Does the model output valid JSON when asked?",
+ setup_structured_output,
+ ),
+ TestScenario(
+ "7", "Elicitation", "Does the model use ask() correctly?", setup_elicitation
+ ),
+ TestScenario(
+ "8",
+ "Goal Chain",
+ "Does the model understand its position in hierarchy?",
+ setup_goal_chain,
+ ),
]
@@ -265,16 +313,19 @@ def _clean_env() -> dict[str, str]:
return env
-def build_cmd(prompt: str, model: str, config_path: Path, budget: float) -> list[str]:
- """Build the claude CLI command."""
- return [
- "claude", "-p", prompt,
- "--model", model,
- "--mcp-config", str(config_path),
- "--allowedTools", " ".join(MCP_TOOLS),
- "--dangerously-skip-permissions",
- "--max-budget-usd", str(budget),
- ]
+def build_cmd(
+ prompt: str, model: str, db_path: Path, agent_id: str, budget: float
+) -> list[str]:
+ """Build the Claude CLI command via harness."""
+ request = AgentLaunchRequest(
+ db_path=db_path,
+ node_id=agent_id,
+ prompt=prompt,
+ model=model,
+ max_budget_usd=budget,
+ project_dir=PROJECT_DIR,
+ )
+ return ClaudeHarness().build_launch_spec(request).cmd
def run_single(
@@ -294,12 +345,7 @@ def run_single(
agent_id, prompt = scenario.setup(db)
nodes_before = snapshot_nodes(db)
- # Generate MCP config
- config = generate_mcp_config(db_path, agent_id, PROJECT_DIR)
- config_path = tmp_dir / "mcp.json"
- config_path.write_text(json.dumps(config, indent=2))
-
- cmd = build_cmd(prompt, model, config_path, budget)
+ cmd = build_cmd(prompt, model, db_path, agent_id, budget)
if dry_run:
print(f" {model}: DRY RUN")
@@ -307,16 +353,25 @@ def run_single(
print(f" Nodes: {len(nodes_before)}")
print(f" Cmd: {' '.join(cmd[:6])}...")
return TestResult(
- test_id=scenario.id, test_name=scenario.name, model=model,
- stdout="(dry run)", stderr="", returncode=0, elapsed=0,
- nodes_before=nodes_before, nodes_after=nodes_before,
+ test_id=scenario.id,
+ test_name=scenario.name,
+ model=model,
+ stdout="(dry run)",
+ stderr="",
+ returncode=0,
+ elapsed=0,
+ nodes_before=nodes_before,
+ nodes_after=nodes_before,
)
print(f" {model}...", end=" ", flush=True)
start = time.time()
result = subprocess.run(
- cmd, capture_output=True, text=True,
- timeout=timeout, cwd=str(PROJECT_DIR),
+ cmd,
+ capture_output=True,
+ text=True,
+ timeout=timeout,
+ cwd=str(PROJECT_DIR),
env=_clean_env(),
)
elapsed = time.time() - start
@@ -326,10 +381,15 @@ def run_single(
nodes_after = snapshot_nodes(db_after)
tr = TestResult(
- test_id=scenario.id, test_name=scenario.name, model=model,
- stdout=result.stdout, stderr=result.stderr,
- returncode=result.returncode, elapsed=elapsed,
- nodes_before=nodes_before, nodes_after=nodes_after,
+ test_id=scenario.id,
+ test_name=scenario.name,
+ model=model,
+ stdout=result.stdout,
+ stderr=result.stderr,
+ returncode=result.returncode,
+ elapsed=elapsed,
+ nodes_before=nodes_before,
+ nodes_after=nodes_after,
)
print(f"done ({elapsed:.1f}s, +{tr.nodes_created} nodes)")
return tr
@@ -337,16 +397,30 @@ def run_single(
except subprocess.TimeoutExpired:
print(f"TIMEOUT ({timeout}s)")
return TestResult(
- test_id=scenario.id, test_name=scenario.name, model=model,
- stdout="", stderr="", returncode=-1, elapsed=float(timeout),
- nodes_before=[], nodes_after=[], error="Timeout",
+ test_id=scenario.id,
+ test_name=scenario.name,
+ model=model,
+ stdout="",
+ stderr="",
+ returncode=-1,
+ elapsed=float(timeout),
+ nodes_before=[],
+ nodes_after=[],
+ error="Timeout",
)
except Exception as e:
print(f"ERROR: {e}")
return TestResult(
- test_id=scenario.id, test_name=scenario.name, model=model,
- stdout="", stderr="", returncode=-1, elapsed=0,
- nodes_before=[], nodes_after=[], error=str(e),
+ test_id=scenario.id,
+ test_name=scenario.name,
+ model=model,
+ stdout="",
+ stderr="",
+ returncode=-1,
+ elapsed=0,
+ nodes_before=[],
+ nodes_after=[],
+ error=str(e),
)
finally:
if not dry_run:
@@ -425,12 +499,14 @@ def generate_report(results: list[TestResult], models: list[str]) -> str:
if not data:
continue
- lines.extend([
- f"## Test {s.id}: {s.name}",
- "",
- f"**Question:** {s.description}",
- "",
- ])
+ lines.extend(
+ [
+ f"## Test {s.id}: {s.name}",
+ "",
+ f"**Question:** {s.description}",
+ "",
+ ]
+ )
for model in models:
r = data.get(model)
@@ -438,13 +514,15 @@ def generate_report(results: list[TestResult], models: list[str]) -> str:
lines.extend([f"### {model.title()}", "", "*Skipped*", ""])
continue
- lines.extend([
- f"### {model.title()}",
- "",
- f"- **Time:** {r.elapsed:.1f}s",
- f"- **Nodes created:** {r.nodes_created}",
- f"- **Return code:** {r.returncode}",
- ])
+ lines.extend(
+ [
+ f"### {model.title()}",
+ "",
+ f"- **Time:** {r.elapsed:.1f}s",
+ f"- **Nodes created:** {r.nodes_created}",
+ f"- **Return code:** {r.returncode}",
+ ]
+ )
if r.error:
lines.extend([f"- **Error:** {r.error}", ""])
@@ -471,14 +549,16 @@ def generate_report(results: list[TestResult], models: list[str]) -> str:
result_short = result_val[:500]
if len(result_val) > 500:
result_short += "\n... (truncated)"
- lines.extend([
- "**Agent result (from complete()):**",
- "",
- "```",
- result_short,
- "```",
- "",
- ])
+ lines.extend(
+ [
+ "**Agent result (from complete()):**",
+ "",
+ "```",
+ result_short,
+ "```",
+ "",
+ ]
+ )
if s.id == "6":
lines.append(f"**Valid JSON:** {'Yes' if is_json else 'No'}")
lines.append("")
@@ -488,19 +568,21 @@ def generate_report(results: list[TestResult], models: list[str]) -> str:
output = r.stdout.strip()
if len(output) > 1500:
output = output[:1500] + "\n\n... (truncated)"
- lines.extend([
- "**CLI output:**",
- "",
- "",
- "Click to expand
",
- "",
- "```",
- output,
- "```",
- "",
- " ",
- "",
- ])
+ lines.extend(
+ [
+ "**CLI output:**",
+ "",
+ "",
+ "Click to expand
",
+ "",
+ "```",
+ output,
+ "```",
+ "",
+ " ",
+ "",
+ ]
+ )
lines.extend(["---", ""])
@@ -515,23 +597,32 @@ def main():
description="Compare Opus vs Sonnet behavior on Cord MCP tools"
)
parser.add_argument(
- "--tests", type=str, default=None,
+ "--tests",
+ type=str,
+ default=None,
help="Comma-separated test IDs (e.g., 1,3,5). Default: all",
)
parser.add_argument(
- "--models", type=str, default=",".join(DEFAULT_MODELS),
+ "--models",
+ type=str,
+ default=",".join(DEFAULT_MODELS),
help=f"Comma-separated models (default: {','.join(DEFAULT_MODELS)})",
)
parser.add_argument(
- "--budget", type=float, default=DEFAULT_BUDGET,
+ "--budget",
+ type=float,
+ default=DEFAULT_BUDGET,
help=f"Max budget per test per model in USD (default: {DEFAULT_BUDGET})",
)
parser.add_argument(
- "--timeout", type=int, default=DEFAULT_TIMEOUT,
+ "--timeout",
+ type=int,
+ default=DEFAULT_TIMEOUT,
help=f"Timeout per test in seconds (default: {DEFAULT_TIMEOUT})",
)
parser.add_argument(
- "--dry-run", action="store_true",
+ "--dry-run",
+ action="store_true",
help="Set up DB and show commands without running claude CLI",
)
args = parser.parse_args()
@@ -562,8 +653,10 @@ def main():
print(f"[Test {scenario.id}] {scenario.name}")
for model in models:
result = run_single(
- scenario, model,
- budget=args.budget, timeout=args.timeout,
+ scenario,
+ model,
+ budget=args.budget,
+ timeout=args.timeout,
dry_run=args.dry_run,
)
results.append(result)
diff --git a/pyproject.toml b/pyproject.toml
index fd7c1c9..2ab49e3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,7 +1,7 @@
[project]
name = "cord"
version = "0.1.0"
-description = "A coordination protocol for trees of Claude Code agents"
+description = "A coordination protocol for trees of agent workers"
readme = "README.md"
authors = [
{ name = "June Kim", email = "kimjune01@gmail.com" }
diff --git a/src/cord/__init__.py b/src/cord/__init__.py
index e3de57c..afa7679 100644
--- a/src/cord/__init__.py
+++ b/src/cord/__init__.py
@@ -1,3 +1,3 @@
-"""Cord: A DSL for coordinating trees of Claude Code agents."""
+"""Cord: A DSL for coordinating trees of agent workers."""
__version__ = "0.1.0"
diff --git a/src/cord/cli.py b/src/cord/cli.py
index 912ccfc..48341b9 100644
--- a/src/cord/cli.py
+++ b/src/cord/cli.py
@@ -6,45 +6,111 @@
from pathlib import Path
from cord.runtime.engine import Engine
+from cord.runtime.harness.factory import resolve_harness
+
+
+USAGE_LINE = (
+ 'cord run "goal description" [--budget ] [--model ] [--amp|--claude]'
+)
+
+
+def _parse_run_args(args: list[str]) -> tuple[str, float, str, bool, bool]:
+ """Parse `cord run` options in any order."""
+ goal: str | None = None
+ budget = 2.0
+ model = "sonnet"
+ force_amp = False
+ force_claude = False
+
+ i = 0
+ while i < len(args):
+ arg = args[i]
+
+ if arg == "--budget":
+ if i + 1 >= len(args):
+ raise ValueError("--budget requires a numeric value")
+ budget = float(args[i + 1])
+ i += 2
+ continue
+
+ if arg == "--model":
+ if i + 1 >= len(args):
+ raise ValueError("--model requires a value")
+ model = args[i + 1]
+ i += 2
+ continue
+
+ if arg == "--amp":
+ force_amp = True
+ i += 1
+ continue
+
+ if arg == "--claude":
+ force_claude = True
+ i += 1
+ continue
+
+ if arg.startswith("--"):
+ raise ValueError(f"Unknown option for run: {arg}")
+
+ if goal is not None:
+ raise ValueError(f"Unexpected extra argument: {arg}")
+
+ goal = arg
+ i += 1
+
+ if not goal:
+ raise ValueError(f"Usage: {USAGE_LINE}")
+
+ return goal, budget, model, force_amp, force_claude
def main() -> None:
- """CLI entry point: cord run "goal" [--budget ] [--model ]."""
+ """CLI entry point: cord run "goal" [options]."""
args = sys.argv[1:]
+ pre_force_amp = False
+ pre_force_claude = False
+ while args and args[0] in ("--amp", "--claude"):
+ if args[0] == "--amp":
+ pre_force_amp = True
+ else:
+ pre_force_claude = True
+ args = args[1:]
+
if not args or args[0] in ("-h", "--help", "help"):
print("Usage:")
- print(' cord run "goal description" [--budget ] [--model ]')
- print(' cord run plan.md [--budget ] [--model ]')
+ print(f" {USAGE_LINE}")
+ print(" cord run plan.md [--budget ] [--model ] [--amp|--claude]")
sys.exit(0)
command = args[0]
if command == "run":
- if len(args) < 2:
- print('Usage: cord run "goal description" [--budget ] [--model ]', file=sys.stderr)
+ try:
+ goal_arg, budget, model, run_force_amp, run_force_claude = _parse_run_args(
+ args[1:]
+ )
+ except ValueError as exc:
+ print(str(exc), file=sys.stderr)
sys.exit(1)
- goal_arg = args[1]
+ force_amp = pre_force_amp or run_force_amp
+ force_claude = pre_force_claude or run_force_claude
+
goal_path = Path(goal_arg)
if goal_path.exists() and goal_path.is_file():
goal = goal_path.read_text().strip()
else:
goal = goal_arg
- budget = 2.0
- if "--budget" in args:
- idx = args.index("--budget")
- if idx + 1 < len(args):
- budget = float(args[idx + 1])
-
- model = "sonnet"
- if "--model" in args:
- idx = args.index("--model")
- if idx + 1 < len(args):
- model = args[idx + 1]
+ try:
+ harness = resolve_harness(force_amp=force_amp, force_claude=force_claude)
+ except ValueError as exc:
+ print(str(exc), file=sys.stderr)
+ sys.exit(2)
- engine = Engine(goal, max_budget_usd=budget, model=model)
+ engine = Engine(goal, max_budget_usd=budget, model=model, harness=harness)
engine.run()
else:
diff --git a/src/cord/mcp/server.py b/src/cord/mcp/server.py
index 1def247..873115f 100644
--- a/src/cord/mcp/server.py
+++ b/src/cord/mcp/server.py
@@ -1,6 +1,6 @@
"""MCP server for cord — stdio transport, one per agent.
-Each claude CLI spawns its own instance via the MCP config.
+Each agent CLI subprocess spawns its own instance via the MCP config.
State is shared through SQLite (WAL mode for concurrent access).
"""
@@ -75,8 +75,12 @@ def read_node(node_id: str) -> str:
@mcp.tool()
-def spawn(goal: str, prompt: str = "", returns: str = "text",
- blocked_by: list[str] | None = None) -> str:
+def spawn(
+ goal: str,
+ prompt: str = "",
+ returns: str = "text",
+ blocked_by: list[str] | None = None,
+) -> str:
"""Create a spawned child node under your node.
Use blocked_by to declare dependencies on other node IDs (e.g. ['#2', '#3'])."""
db = _get_db()
@@ -92,8 +96,12 @@ def spawn(goal: str, prompt: str = "", returns: str = "text",
@mcp.tool()
-def fork(goal: str, prompt: str = "", returns: str = "text",
- blocked_by: list[str] | None = None) -> str:
+def fork(
+ goal: str,
+ prompt: str = "",
+ returns: str = "text",
+ blocked_by: list[str] | None = None,
+) -> str:
"""Create a forked child node (inherits parent context) under your node.
Use blocked_by to declare dependencies on other node IDs."""
db = _get_db()
@@ -119,8 +127,9 @@ def complete(result: str = "") -> str:
@mcp.tool()
-def ask(question: str, options: list[str] | None = None,
- default: str | None = None) -> str:
+def ask(
+ question: str, options: list[str] | None = None, default: str | None = None
+) -> str:
"""Create an ask node to get input from a human or parent agent."""
db = _get_db()
prompt_text = question
@@ -164,11 +173,13 @@ def _check_subtree(db: CordDB, node_id: str) -> str | None:
if not node:
return json.dumps({"error": f"Node {node_id} not found"})
if agent_id and not _is_descendant(db, agent_id, node_id):
- return json.dumps({
- "error": f"Node {node_id} is not in your subtree. "
- "You can only modify your own descendants. "
- "Use ask() to request the parent to do it."
- })
+ return json.dumps(
+ {
+ "error": f"Node {node_id} is not in your subtree. "
+ "You can only modify your own descendants. "
+ "Use ask() to request the parent to do it."
+ }
+ )
return None
@@ -180,7 +191,11 @@ def pause(node_id: str) -> str:
return err
node = db.get_node(node_id)
if node["status"] != "active":
- return json.dumps({"error": f"Node {node_id} is {node['status']}, not active. Only active nodes can be paused."})
+ return json.dumps(
+ {
+ "error": f"Node {node_id} is {node['status']}, not active. Only active nodes can be paused."
+ }
+ )
db.update_status(node_id, "paused")
return json.dumps({"paused": node_id})
@@ -193,7 +208,11 @@ def resume(node_id: str) -> str:
return err
node = db.get_node(node_id)
if node["status"] != "paused":
- return json.dumps({"error": f"Node {node_id} is {node['status']}, not paused. Only paused nodes can be resumed."})
+ return json.dumps(
+ {
+ "error": f"Node {node_id} is {node['status']}, not paused. Only paused nodes can be resumed."
+ }
+ )
db.update_status(node_id, "pending")
return json.dumps({"resumed": node_id})
@@ -206,9 +225,15 @@ def modify(node_id: str, goal: str | None = None, prompt: str | None = None) ->
return err
node = db.get_node(node_id)
if node["status"] not in ("pending", "paused"):
- return json.dumps({"error": f"Node {node_id} is {node['status']}. Only pending or paused nodes can be modified."})
+ return json.dumps(
+ {
+ "error": f"Node {node_id} is {node['status']}. Only pending or paused nodes can be modified."
+ }
+ )
if goal is None and prompt is None:
- return json.dumps({"error": "Provide at least one of goal or prompt to modify."})
+ return json.dumps(
+ {"error": "Provide at least one of goal or prompt to modify."}
+ )
db.modify_node(node_id, goal=goal, prompt=prompt)
updated = db.get_node(node_id)
return json.dumps({"modified": node_id, "goal": updated["goal"]})
diff --git a/src/cord/runtime/dispatcher.py b/src/cord/runtime/dispatcher.py
index 5f18270..7f2cd01 100644
--- a/src/cord/runtime/dispatcher.py
+++ b/src/cord/runtime/dispatcher.py
@@ -1,42 +1,12 @@
-"""Launch claude CLI processes for cord nodes."""
+"""Compatibility wrappers around runtime harness utilities."""
from __future__ import annotations
-import json
import subprocess
from pathlib import Path
-
-MCP_TOOLS = [
- "mcp__cord__read_tree",
- "mcp__cord__read_node",
- "mcp__cord__spawn",
- "mcp__cord__fork",
- "mcp__cord__ask",
- "mcp__cord__stop",
- "mcp__cord__complete",
- "mcp__cord__pause",
- "mcp__cord__resume",
- "mcp__cord__modify",
-]
-
-
-def generate_mcp_config(db_path: Path, agent_id: str, project_dir: Path) -> dict:
- """Generate MCP config that spawns a stdio server for this agent."""
- return {
- "mcpServers": {
- "cord": {
- "command": "uv",
- "args": [
- "run",
- "--directory", str(project_dir.resolve()),
- "cord-mcp-server",
- "--db-path", str(db_path.resolve()),
- "--agent-id", agent_id,
- ],
- }
- }
- }
+from cord.runtime.harness.base import MCP_TOOLS, AgentLaunchRequest, generate_mcp_config
+from cord.runtime.harness.claude import ClaudeHarness
def launch_agent(
@@ -48,33 +18,15 @@ def launch_agent(
model: str = "sonnet",
project_dir: Path | None = None,
) -> subprocess.Popen[str]:
- """Launch a claude CLI process for a node."""
+ """Launch a Claude subprocess for a node."""
proj = project_dir or db_path.parent
- mcp_config = generate_mcp_config(db_path, node_id, proj)
-
- config_dir = db_path.parent / ".cord"
- config_dir.mkdir(parents=True, exist_ok=True)
- config_path = config_dir / f"mcp-{node_id.lstrip('#')}.json"
- config_path.write_text(json.dumps(mcp_config, indent=2))
-
- cmd = [
- "claude",
- "-p", prompt,
- "--model", model,
- "--mcp-config", str(config_path),
- "--allowedTools", " ".join(MCP_TOOLS),
- "--dangerously-skip-permissions",
- "--max-budget-usd", str(max_budget_usd),
- ]
-
- cwd = str(work_dir) if work_dir else str(proj)
-
- process = subprocess.Popen(
- cmd,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- text=True,
- cwd=cwd,
+ request = AgentLaunchRequest(
+ db_path=db_path,
+ node_id=node_id,
+ prompt=prompt,
+ work_dir=work_dir,
+ max_budget_usd=max_budget_usd,
+ model=model,
+ project_dir=proj,
)
-
- return process
+ return ClaudeHarness().launch(request)
diff --git a/src/cord/runtime/engine.py b/src/cord/runtime/engine.py
index fd66930..38405e3 100644
--- a/src/cord/runtime/engine.py
+++ b/src/cord/runtime/engine.py
@@ -8,7 +8,8 @@
from cord.db import CordDB
from cord.prompts import build_agent_prompt, build_synthesis_prompt
-from cord.runtime.dispatcher import launch_agent
+from cord.runtime.harness.base import AgentHarness, AgentLaunchRequest
+from cord.runtime.harness.factory import resolve_harness
from cord.runtime.process_manager import ProcessManager
@@ -26,6 +27,7 @@ def __init__(
max_budget_usd: float = 2.0,
model: str = "sonnet",
project_dir: Path | None = None,
+ harness: AgentHarness | None = None,
):
self.goal = goal
self.project_dir = (project_dir or Path.cwd()).resolve()
@@ -37,6 +39,7 @@ def __init__(
self.poll_interval = poll_interval
self.max_budget_usd = max_budget_usd
self.model = model
+ self.harness = harness
self.process_manager = ProcessManager()
self.db = CordDB(self.db_path)
self._last_tree_hash = ""
@@ -98,7 +101,9 @@ def _main_loop(self) -> None:
if self.process_manager.active_count == 0 and not ready:
pending = [n for n in self.db.all_nodes() if n["status"] == "pending"]
if pending:
- self._log(f"Stuck: {len(pending)} pending nodes with unmet dependencies")
+ self._log(
+ f"Stuck: {len(pending)} pending nodes with unmet dependencies"
+ )
break
time.sleep(self.poll_interval)
@@ -107,15 +112,25 @@ def _launch_node(self, node_id: str) -> None:
prompt = build_agent_prompt(self.db, node_id)
self.db.update_status(node_id, "active")
- process = launch_agent(
- self.db_path,
- node_id,
- prompt,
+ process = self._launch_process(node_id, prompt)
+ self.process_manager.register(node_id, process)
+
+ def _launch_process(self, node_id: str, prompt: str):
+ request = AgentLaunchRequest(
+ db_path=self.db_path,
+ node_id=node_id,
+ prompt=prompt,
max_budget_usd=self.max_budget_usd,
model=self.model,
project_dir=self.project_dir,
)
- self.process_manager.register(node_id, process)
+ harness = self._get_harness()
+ return harness.launch(request)
+
+ def _get_harness(self) -> AgentHarness:
+ if self.harness is None:
+ self.harness = resolve_harness()
+ return self.harness
def _handle_completion(self, node_id: str, return_code: int, stdout: str) -> None:
node = self.db.get_node(node_id)
@@ -144,8 +159,7 @@ def _check_synthesis(self, completed_node_id: str) -> None:
children = self.db.get_children(parent_id)
all_done = all(
- c["status"] in ("complete", "failed", "cancelled")
- for c in children
+ c["status"] in ("complete", "failed", "cancelled") for c in children
)
if not all_done:
return
@@ -166,14 +180,7 @@ def _check_synthesis(self, completed_node_id: str) -> None:
# Relaunch parent for synthesis
self.db.update_status(parent_id, "active")
prompt = build_synthesis_prompt(self.db, parent_id)
- process = launch_agent(
- self.db_path,
- parent_id,
- prompt,
- max_budget_usd=self.max_budget_usd,
- model=self.model,
- project_dir=self.project_dir,
- )
+ process = self._launch_process(parent_id, prompt)
self.process_manager.register(parent_id, process)
def _handle_ask(self, node: dict) -> None:
@@ -263,10 +270,10 @@ def _log(self, message: str) -> None:
def _status_style(status: str) -> tuple[str, str]:
return {
- "pending": ("\033[90m", "○"),
- "active": ("\033[34m", "●"),
- "complete": ("\033[32m", "✓"),
- "failed": ("\033[31m", "✗"),
+ "pending": ("\033[90m", "○"),
+ "active": ("\033[34m", "●"),
+ "complete": ("\033[32m", "✓"),
+ "failed": ("\033[31m", "✗"),
"cancelled": ("\033[33m", "⊘"),
- "waiting": ("\033[36m", "?"),
+ "waiting": ("\033[36m", "?"),
}.get(status, ("\033[0m", "?"))
diff --git a/src/cord/runtime/harness/__init__.py b/src/cord/runtime/harness/__init__.py
new file mode 100644
index 0000000..59cbd03
--- /dev/null
+++ b/src/cord/runtime/harness/__init__.py
@@ -0,0 +1,13 @@
+"""Runtime harness abstraction and implementations."""
+
+from cord.runtime.harness.base import MCP_TOOLS, AgentHarness, AgentLaunchRequest
+from cord.runtime.harness.factory import create_harness, resolve_harness, select_backend
+
+__all__ = [
+ "MCP_TOOLS",
+ "AgentHarness",
+ "AgentLaunchRequest",
+ "select_backend",
+ "create_harness",
+ "resolve_harness",
+]
diff --git a/src/cord/runtime/harness/amp.py b/src/cord/runtime/harness/amp.py
new file mode 100644
index 0000000..762df97
--- /dev/null
+++ b/src/cord/runtime/harness/amp.py
@@ -0,0 +1,109 @@
+"""Amp CLI harness implementation."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+from cord.runtime.harness.base import (
+ MCP_TOOLS,
+ AgentHarness,
+ AgentLaunchRequest,
+ LaunchSpec,
+ generate_mcp_config,
+ node_file_slug,
+ write_json_file,
+)
+
+
+class AmpHarness(AgentHarness):
+ """Launch agents through Amp CLI."""
+
+ def __init__(self, mode: str | None = None):
+ self.mode = mode
+ self._warned_model = False
+ self._warned_budget = False
+
+ def build_launch_spec(self, request: AgentLaunchRequest) -> LaunchSpec:
+ proj = request.project_dir
+ config_dir = request.db_path.parent / ".cord"
+ slug = node_file_slug(request.node_id)
+ mcp_path = config_dir / f"mcp-{slug}.json"
+ settings_path = config_dir / f"amp-settings-{slug}.json"
+
+ mcp_config = generate_mcp_config(request.db_path, request.node_id, proj)
+ write_json_file(mcp_path, mcp_config["mcpServers"])
+ settings = self._load_base_settings()
+ settings.pop("amp.tools.disable", None)
+ settings["amp.tools.enable"] = MCP_TOOLS
+ write_json_file(
+ settings_path,
+ settings,
+ )
+
+ self._warn_option_gaps(request)
+
+ cmd = [
+ "amp",
+ "-x",
+ request.prompt,
+ "--mcp-config",
+ str(mcp_path),
+ "--settings-file",
+ str(settings_path),
+ "--no-color",
+ ]
+ if self.mode:
+ cmd.extend(["--mode", self.mode])
+
+ env = os.environ.copy()
+ env["TERM"] = "dumb"
+
+ cwd = request.work_dir or proj
+ return LaunchSpec(cmd=cmd, cwd=cwd, env=env)
+
+ def _warn_option_gaps(self, request: AgentLaunchRequest) -> None:
+ if request.model != "sonnet" and not self._warned_model:
+ print(
+ "Warning: --model is not mapped to Amp CLI; ignoring model override.",
+ file=sys.stderr,
+ )
+ self._warned_model = True
+
+ if request.max_budget_usd != 2.0 and not self._warned_budget:
+ print(
+ "Warning: --budget is not mapped to Amp CLI; ignoring budget override.",
+ file=sys.stderr,
+ )
+ self._warned_budget = True
+
+ def _load_base_settings(self) -> dict:
+ """Load user's default Amp settings so permissions remain intact."""
+ raw_path = os.environ.get("AMP_SETTINGS_FILE")
+ settings_path = (
+ Path(raw_path).expanduser()
+ if raw_path
+ else Path.home() / ".config" / "amp" / "settings.json"
+ )
+ if not settings_path.exists():
+ return {}
+
+ try:
+ payload = json.loads(settings_path.read_text())
+ except (json.JSONDecodeError, OSError):
+ print(
+ f"Warning: failed to parse Amp settings file: {settings_path}",
+ file=sys.stderr,
+ )
+ return {}
+
+ if not isinstance(payload, dict):
+ print(
+ f"Warning: Amp settings must be a JSON object: {settings_path}",
+ file=sys.stderr,
+ )
+ return {}
+
+ return payload
diff --git a/src/cord/runtime/harness/base.py b/src/cord/runtime/harness/base.py
new file mode 100644
index 0000000..1689886
--- /dev/null
+++ b/src/cord/runtime/harness/base.py
@@ -0,0 +1,98 @@
+"""Agent harness base types and shared launch helpers."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from pathlib import Path
+
+
+MCP_TOOLS = [
+ "mcp__cord__read_tree",
+ "mcp__cord__read_node",
+ "mcp__cord__spawn",
+ "mcp__cord__fork",
+ "mcp__cord__ask",
+ "mcp__cord__stop",
+ "mcp__cord__complete",
+ "mcp__cord__pause",
+ "mcp__cord__resume",
+ "mcp__cord__modify",
+]
+
+
+@dataclass(frozen=True)
+class AgentLaunchRequest:
+ """Runtime launch inputs for a single agent subprocess."""
+
+ db_path: Path
+ node_id: str
+ prompt: str
+ project_dir: Path
+ work_dir: Path | None = None
+ max_budget_usd: float = 2.0
+ model: str = "sonnet"
+
+
+@dataclass(frozen=True)
+class LaunchSpec:
+ """Resolved subprocess launch settings for a harness."""
+
+ cmd: list[str]
+ cwd: Path
+ env: dict[str, str] | None = None
+
+
+def generate_mcp_config(db_path: Path, agent_id: str, project_dir: Path) -> dict:
+ """Generate MCP config that spawns a stdio server for this agent."""
+ return {
+ "mcpServers": {
+ "cord": {
+ "command": "uv",
+ "args": [
+ "run",
+ "--directory",
+ str(project_dir.resolve()),
+ "cord-mcp-server",
+ "--db-path",
+ str(db_path.resolve()),
+ "--agent-id",
+ agent_id,
+ ],
+ }
+ }
+ }
+
+
+def write_json_file(path: Path, payload: dict) -> Path:
+ """Write JSON payload to disk, creating parent dirs as needed."""
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(json.dumps(payload, indent=2))
+ return path
+
+
+def node_file_slug(node_id: str) -> str:
+ """Return a filesystem-safe node id slug."""
+ return node_id.lstrip("#")
+
+
+class AgentHarness(ABC):
+ """Backend-agnostic subprocess harness for launching agents."""
+
+ @abstractmethod
+ def build_launch_spec(self, request: AgentLaunchRequest) -> LaunchSpec:
+ """Return concrete subprocess settings for a request."""
+
+ def launch(self, request: AgentLaunchRequest) -> subprocess.Popen[str]:
+ """Launch a subprocess for an agent request."""
+ spec = self.build_launch_spec(request)
+ return subprocess.Popen(
+ spec.cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ cwd=str(spec.cwd),
+ env=spec.env,
+ )
diff --git a/src/cord/runtime/harness/claude.py b/src/cord/runtime/harness/claude.py
new file mode 100644
index 0000000..0ff270e
--- /dev/null
+++ b/src/cord/runtime/harness/claude.py
@@ -0,0 +1,43 @@
+"""Claude CLI harness implementation."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from cord.runtime.harness.base import (
+ MCP_TOOLS,
+ AgentHarness,
+ AgentLaunchRequest,
+ LaunchSpec,
+ generate_mcp_config,
+ node_file_slug,
+ write_json_file,
+)
+
+
+class ClaudeHarness(AgentHarness):
+ """Launch agents through Claude Code CLI."""
+
+ def build_launch_spec(self, request: AgentLaunchRequest) -> LaunchSpec:
+ proj = request.project_dir
+ config_dir = request.db_path.parent / ".cord"
+ config_path = config_dir / f"mcp-{node_file_slug(request.node_id)}.json"
+ mcp_config = generate_mcp_config(request.db_path, request.node_id, proj)
+ write_json_file(config_path, mcp_config)
+
+ cmd = [
+ "claude",
+ "-p",
+ request.prompt,
+ "--model",
+ request.model,
+ "--mcp-config",
+ str(config_path),
+ "--allowedTools",
+ " ".join(MCP_TOOLS),
+ "--dangerously-skip-permissions",
+ "--max-budget-usd",
+ str(request.max_budget_usd),
+ ]
+ cwd = request.work_dir or proj
+ return LaunchSpec(cmd=cmd, cwd=cwd)
diff --git a/src/cord/runtime/harness/factory.py b/src/cord/runtime/harness/factory.py
new file mode 100644
index 0000000..3ade5e0
--- /dev/null
+++ b/src/cord/runtime/harness/factory.py
@@ -0,0 +1,76 @@
+"""Harness selection and construction."""
+
+from __future__ import annotations
+
+import shutil
+from collections.abc import Callable
+from typing import Literal
+
+from cord.runtime.harness.amp import AmpHarness
+from cord.runtime.harness.base import AgentHarness
+from cord.runtime.harness.claude import ClaudeHarness
+
+
+BackendName = Literal["claude", "amp"]
+
+
+def select_backend(
+ *,
+ force_claude: bool = False,
+ force_amp: bool = False,
+ which: Callable[[str], str | None] = shutil.which,
+) -> BackendName:
+ """Resolve backend from explicit flags or local binary availability."""
+ if force_claude and force_amp:
+ raise ValueError("Invalid args: pass only one of --claude or --amp.")
+
+ has_amp = which("amp") is not None
+ has_claude = which("claude") is not None
+
+ if force_amp:
+ if not has_amp:
+ raise ValueError(
+ "Requested --amp but 'amp' binary was not found in PATH. "
+ "Install Amp CLI or run with --claude."
+ )
+ return "amp"
+
+ if force_claude:
+ if not has_claude:
+ raise ValueError(
+ "Requested --claude but 'claude' binary was not found in PATH. "
+ "Install Claude Code CLI or run with --amp."
+ )
+ return "claude"
+
+ if has_amp:
+ return "amp"
+ if has_claude:
+ return "claude"
+
+ raise ValueError(
+ "No supported agent CLI found in PATH. Install Amp CLI ('amp') or "
+ "Claude Code CLI ('claude'), or pass --amp/--claude after install."
+ )
+
+
+def create_harness(backend: BackendName) -> AgentHarness:
+ """Create a harness instance for a backend name."""
+ if backend == "amp":
+ return AmpHarness()
+ return ClaudeHarness()
+
+
+def resolve_harness(
+ *,
+ force_claude: bool = False,
+ force_amp: bool = False,
+ which: Callable[[str], str | None] = shutil.which,
+) -> AgentHarness:
+ """Resolve and construct harness from flags and environment."""
+ backend = select_backend(
+ force_claude=force_claude,
+ force_amp=force_amp,
+ which=which,
+ )
+ return create_harness(backend)
diff --git a/src/cord/runtime/process_manager.py b/src/cord/runtime/process_manager.py
index 2419714..296d920 100644
--- a/src/cord/runtime/process_manager.py
+++ b/src/cord/runtime/process_manager.py
@@ -1,4 +1,4 @@
-"""Track and manage Claude CLI subprocess lifecycle."""
+"""Track and manage agent subprocess lifecycle."""
from __future__ import annotations
@@ -16,7 +16,7 @@ class ProcessInfo:
class ProcessManager:
- """Manages claude CLI subprocesses for active nodes."""
+ """Manages agent CLI subprocesses for active nodes."""
def __init__(self) -> None:
self._processes: dict[str, ProcessInfo] = {}
diff --git a/tests/test_cli.py b/tests/test_cli.py
new file mode 100644
index 0000000..268018f
--- /dev/null
+++ b/tests/test_cli.py
@@ -0,0 +1,49 @@
+"""Tests for CLI argument parsing and backend flags."""
+
+from __future__ import annotations
+
+import pytest
+
+import cord.cli as cli
+
+
+class _FakeEngine:
+ def __init__(self, goal: str, max_budget_usd: float, model: str, harness):
+ self.goal = goal
+ self.max_budget_usd = max_budget_usd
+ self.model = model
+ self.harness = harness
+ self.ran = False
+
+ def run(self) -> None:
+ self.ran = True
+
+
+def test_backend_flag_before_command(monkeypatch):
+ captured: dict[str, object] = {}
+
+ def _fake_resolve_harness(*, force_amp: bool, force_claude: bool):
+ captured["flags"] = (force_amp, force_claude)
+ return "HARNESS"
+
+ def _fake_engine(goal: str, max_budget_usd: float, model: str, harness):
+ captured["engine_args"] = (goal, max_budget_usd, model, harness)
+ return _FakeEngine(goal, max_budget_usd, model, harness)
+
+ monkeypatch.setattr(cli, "resolve_harness", _fake_resolve_harness)
+ monkeypatch.setattr(cli, "Engine", _fake_engine)
+ monkeypatch.setattr(cli.sys, "argv", ["cord", "--amp", "run", "Print ok"])
+
+ cli.main()
+
+ assert captured["flags"] == (True, False)
+ assert captured["engine_args"] == ("Print ok", 2.0, "sonnet", "HARNESS")
+
+
+def test_missing_run_goal_exits(monkeypatch):
+ monkeypatch.setattr(cli.sys, "argv", ["cord", "run", "--amp"])
+
+ with pytest.raises(SystemExit) as exc:
+ cli.main()
+
+ assert exc.value.code == 1
diff --git a/tests/test_harness.py b/tests/test_harness.py
new file mode 100644
index 0000000..da0da1a
--- /dev/null
+++ b/tests/test_harness.py
@@ -0,0 +1,110 @@
+"""Tests for runtime harness selection and command construction."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from cord.runtime.harness.amp import AmpHarness
+from cord.runtime.harness.base import AgentLaunchRequest, MCP_TOOLS
+from cord.runtime.harness.claude import ClaudeHarness
+from cord.runtime.harness.factory import select_backend
+
+
+def _which_map(available: set[str]):
+ def _which(binary: str) -> str | None:
+ return f"/usr/local/bin/{binary}" if binary in available else None
+
+ return _which
+
+
+class TestHarnessSelection:
+ def test_auto_detect_prefers_amp(self):
+ backend = select_backend(which=_which_map({"amp", "claude"}))
+ assert backend == "amp"
+
+ def test_explicit_flag_precedence(self):
+ backend = select_backend(force_claude=True, which=_which_map({"amp", "claude"}))
+ assert backend == "claude"
+
+ def test_conflicting_flags_rejected(self):
+ with pytest.raises(ValueError, match="--claude or --amp"):
+ select_backend(
+ force_claude=True, force_amp=True, which=_which_map({"amp", "claude"})
+ )
+
+ def test_no_backend_available_rejected(self):
+ with pytest.raises(ValueError, match="No supported agent CLI"):
+ select_backend(which=_which_map(set()))
+
+
+class TestHarnessCommands:
+ def test_claude_command_generation(self, tmp_path):
+ db_path = tmp_path / "cord.db"
+ request = AgentLaunchRequest(
+ db_path=db_path,
+ node_id="#12",
+ prompt="Do the task",
+ model="opus",
+ max_budget_usd=3.5,
+ project_dir=tmp_path,
+ )
+
+ spec = ClaudeHarness().build_launch_spec(request)
+
+ assert spec.cmd[0] == "claude"
+ assert "--mcp-config" in spec.cmd
+ assert "--allowedTools" in spec.cmd
+ assert " ".join(MCP_TOOLS) in spec.cmd
+ assert "--dangerously-skip-permissions" in spec.cmd
+ assert "--max-budget-usd" in spec.cmd
+ assert "3.5" in spec.cmd
+
+ config_idx = spec.cmd.index("--mcp-config") + 1
+ config_path = spec.cmd[config_idx]
+ assert config_path.endswith("mcp-12.json")
+
+ def test_amp_command_generation(self, tmp_path, monkeypatch):
+ db_path = tmp_path / "cord.db"
+ default_settings_path = tmp_path / "amp-default-settings.json"
+ default_settings_path.write_text(
+ json.dumps(
+ {
+ "amp.permissions": [
+ {"tool": "Bash", "action": "ask"},
+ ]
+ }
+ )
+ )
+ monkeypatch.setenv("AMP_SETTINGS_FILE", str(default_settings_path))
+
+ request = AgentLaunchRequest(
+ db_path=db_path,
+ node_id="#7",
+ prompt="Do the task",
+ project_dir=tmp_path,
+ )
+
+ spec = AmpHarness().build_launch_spec(request)
+
+ assert spec.cmd[0] == "amp"
+ assert spec.cmd[1] == "-x"
+ assert "--mcp-config" in spec.cmd
+ assert "--settings-file" in spec.cmd
+ assert "--no-color" in spec.cmd
+ assert spec.env is not None
+ assert spec.env["TERM"] == "dumb"
+
+ settings_idx = spec.cmd.index("--settings-file") + 1
+ settings_path = spec.cmd[settings_idx]
+ mcp_idx = spec.cmd.index("--mcp-config") + 1
+ mcp_path = spec.cmd[mcp_idx]
+ settings = json.loads((tmp_path / ".cord" / "amp-settings-7.json").read_text())
+ mcp_config = json.loads((tmp_path / ".cord" / "mcp-7.json").read_text())
+
+ assert settings_path.endswith("amp-settings-7.json")
+ assert mcp_path.endswith("mcp-7.json")
+ assert "cord" in mcp_config
+ assert settings["amp.tools.enable"] == MCP_TOOLS
+ assert settings["amp.permissions"] == [{"tool": "Bash", "action": "ask"}]