From 45520ad4a897efb72eef1c16365cf9a86ec4dd30 Mon Sep 17 00:00:00 2001 From: Joe Petrucelli Date: Mon, 3 Aug 2026 18:40:13 -0400 Subject: [PATCH] feat(call-agent): optional cwd and permission-mode for sandboxed runs --- CHANGELOG.md | 25 ++++++++++++ README.md | 30 +++++++++++++++ claude_batch_runner/__init__.py | 4 +- claude_batch_runner/deliver.py | 33 +++++++++++----- claude_batch_runner/driver.py | 41 +++++++++++++++++++- pyproject.toml | 2 +- tests/test_deliver.py | 20 ++++++++++ tests/test_driver.py | 67 +++++++++++++++++++++++++++++++++ 8 files changed, 209 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b8cd6d..147de63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,31 @@ All notable changes to this project are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## 0.2.0 + +### Added + +- **`cwd` and `permission_mode` on `call_agent`** — two optional keyword-only + arguments. `cwd` runs the `claude` subprocess in that directory, validated to + exist first so an unusable path raises `AgentError` before a process starts + (inside `fan_out` that surfaces as the unit's `_error` row instead of taking + the batch down). `permission_mode` is passed to `claude --permission-mode`; + the accepted modes stay the CLI's vocabulary, not the runner's. This is what + a replay harness needs to run recorded, action-shaped prompts against a + throwaway fixture directory under a restrictive mode. +- **Both threaded through `fan_out` and `deliver.run`** with the same optional + contract — a campaign confines its worker, advisor, *and* grader calls alike, + so an escalation cannot step outside the sandbox its worker ran in. +- **`agent_options`** — builds the optional `call_agent` kwargs and omits + whatever is unset, which is what keeps the no-argument path byte-identical. + +### Compatibility + +Both arguments default to `None`, and when omitted nothing changes: no +`--permission-mode` in the argv, no `cwd=` keyword on the `subprocess.run` call +at all, and injected `call=` functions written against the four-positional +signature are invoked exactly as before. + ## 0.1.0 First public release. diff --git a/README.md b/README.md index 1792c3b..c048008 100644 --- a/README.md +++ b/README.md @@ -235,6 +235,36 @@ in the results rather than in a count you forgot to check. estimate crosses it. Set it as a runaway-loop stop, generously — tuning it down just moves the abort onto whichever unit reads the most. +### Confining a run to a directory + +`call_agent` takes two optional keyword arguments for running agents somewhere +other than the caller's own directory, on tighter permissions: + +```python +from claude_batch_runner import call_agent + +envelope = call_agent("doc-reviewer", prompt, schema, 5.0, + cwd="/tmp/eval-fixture", # the subprocess's working directory + permission_mode="plan") # claude --permission-mode +``` + +`cwd` must be an existing directory. A missing one raises `AgentError` *before* +the subprocess starts, so a bad path inside a fan-out comes back as that unit's +`{"_error": ...}` row rather than taking the batch down. `permission_mode` is +handed straight to the CLI — the accepted modes are its vocabulary, not the +runner's, so a mode the CLI adds later works without a release here. + +Both are accepted by `fan_out(tasks, cwd=..., permission_mode=...)`, which +applies them to every task, and by `deliver.run(campaign, name, cwd=..., +permission_mode=...)`, which applies them to every call a campaign makes — +worker, advisor, and grader alike, so an escalation cannot step outside the +sandbox its worker ran in. + +Both default to `None`, and omitting them changes nothing: no +`--permission-mode` in the argv, no `cwd` on the subprocess call, and an +injected `call=` written against the four-argument signature is invoked exactly +as before. + ## Examples - [`examples/demo-campaign.json`](examples/) — the campaign above, runnable diff --git a/claude_batch_runner/__init__.py b/claude_batch_runner/__init__.py index a917734..7141431 100644 --- a/claude_batch_runner/__init__.py +++ b/claude_batch_runner/__init__.py @@ -15,6 +15,7 @@ AgentError, AgentTask, agent_failure_detail, + agent_options, call_agent, fan_out, ) @@ -43,6 +44,7 @@ "VerifyError", "Worker", "agent_failure_detail", + "agent_options", "call_agent", "fan_out", "load_campaign", @@ -50,4 +52,4 @@ "verify_unit", ] -__version__ = "0.1.0" +__version__ = "0.2.0" diff --git a/claude_batch_runner/deliver.py b/claude_batch_runner/deliver.py index 63d545a..28e9c0c 100644 --- a/claude_batch_runner/deliver.py +++ b/claude_batch_runner/deliver.py @@ -43,14 +43,27 @@ def report_path(campaign: Campaign, name: str) -> Path: return Path(campaign.report.format(date=datetime.date.today().isoformat(), campaign=name)) -def run(campaign: Campaign, name: str, *, call: Any = None) -> CampaignResult: - """Execute `campaign` per its deliver adapter; `name` keys the report path.""" +def run( + campaign: Campaign, + name: str, + *, + call: Any = None, + cwd: str | Path | None = None, + permission_mode: str | None = None, +) -> CampaignResult: + """Execute `campaign` per its deliver adapter; `name` keys the report path. + + `cwd` and `permission_mode` (see `driver.call_agent`) confine every call the + campaign makes — worker, advisor, and grader alike, so an escalation cannot + step outside the sandbox its worker ran in. Omitted, the run is unchanged. + """ if campaign.deliver == "pr-dispatch": raise NotImplementedError(PR_DISPATCH_CONTRACT) - return _report_only(campaign, name, call or driver.call_agent) + opts = driver.agent_options(cwd, permission_mode) + return _report_only(campaign, name, call or driver.call_agent, opts) -def _report_only(campaign: Campaign, name: str, call: Any) -> CampaignResult: +def _report_only(campaign: Campaign, name: str, call: Any, opts: dict[str, Any]) -> CampaignResult: start, cond = time.monotonic(), None if campaign.escalate: # upstream gate: a bad condition fails before any tokens burn raw = campaign.escalate.condition @@ -71,16 +84,18 @@ def _report_only(campaign: Campaign, name: str, call: Any) -> CampaignResult: ) for i, u in enumerate(campaign.units, 1) } - envelopes = driver.fan_out(tasks, call=call, max_workers=campaign.parallelism) + envelopes = driver.fan_out(tasks, call=call, max_workers=campaign.parallelism, **opts) units = tuple( - _finish_unit(key, u, tasks[key], envelopes[key], campaign, cond, rubric, meter, tiers, call) + _finish_unit( + key, u, tasks[key], envelopes[key], campaign, cond, rubric, meter, tiers, call, opts + ) for key, u in zip(tasks, campaign.units) ) duration = time.monotonic() - start return CampaignResult(units, _write_report(campaign, name, units, meter, duration), duration) -def _finish_unit(key, unit, task, envelope, campaign, cond, rubric, meter, tiers, call): +def _finish_unit(key, unit, task, envelope, campaign, cond, rubric, meter, tiers, call, opts): label = (unit if isinstance(unit, str) else json.dumps(unit)).replace("|", "\\|")[:48] acc = {"tokens": 0, "metered": False} @@ -99,7 +114,7 @@ def done(outcome, verdict, revisions, note): # tokens None = no usage seen for return UnitReport(key, label, outcome, verdict, revisions, toks, note) def metered(agent, prompt, schema, budget): # verify-pass calls: grader + worker revisions - env = call(agent, prompt, schema, budget) + env = call(agent, prompt, schema, budget, **opts) tally(tiers.get(agent, "?"), env) return env @@ -116,7 +131,7 @@ def metered(agent, prompt, schema, budget): # verify-pass calls: grader + worke f"{condition!r}. Produce the authoritative output.\n" + frame("WORKER DRAFT", draft) ) try: - adv_env = call(advisor.agent, prompt, campaign.schema, UNIT_BUDGET_USD) + adv_env = call(advisor.agent, prompt, campaign.schema, UNIT_BUDGET_USD, **opts) except driver.AgentError as e: return done(FLAGGED, "-", 0, f"advisor error: {e}") tally(advisor.tier, adv_env) diff --git a/claude_batch_runner/driver.py b/claude_batch_runner/driver.py index d9205c6..5f5c0d4 100644 --- a/claude_batch_runner/driver.py +++ b/claude_batch_runner/driver.py @@ -14,6 +14,7 @@ AgentTask one unit of fan-out work call_agent single `claude -p --agent` subprocess call fan_out parallel fan-out over {key: AgentTask} + agent_options the optional call_agent kwargs, minus whatever is unset AgentError raised on any non-success or malformed result agent_failure_detail envelope-aware failure description """ @@ -25,6 +26,7 @@ import json import os import subprocess +from pathlib import Path from typing import Any DEFAULT_AGENT_TIMEOUT_S = int(os.environ.get("AGENT_TIMEOUT_S", "600")) @@ -71,17 +73,41 @@ def agent_failure_detail(proc: subprocess.CompletedProcess[str]) -> str | None: return f"exit {proc.returncode}, " + (", ".join(bits) or "is_error=True")[:600] +def agent_options( + cwd: str | Path | None = None, permission_mode: str | None = None +) -> dict[str, Any]: + """The optional `call_agent` keyword arguments, minus whatever is unset. + + Empty when both are None, so a caller's own injected `call=` written against + the four-positional-argument signature keeps working untouched. + """ + given = (("cwd", cwd), ("permission_mode", permission_mode)) + return {k: v for k, v in given if v is not None} + + def call_agent( agent_name: str, prompt: str, schema: dict[str, Any] | None, budget_usd: float, + *, + cwd: str | Path | None = None, + permission_mode: str | None = None, ) -> dict[str, Any]: """Subprocess `claude -p --agent ` and return the result envelope. When `schema` is provided, output is coerced into envelope["structured_output"]. Raises AgentError on any non-success result or parse failure. + + `cwd` runs the subprocess in that directory — checked to exist first, so an + unusable path raises before a process starts (and inside `fan_out` becomes + that unit's `_error` row rather than taking the batch down). `permission_mode` + is handed straight to `claude --permission-mode`; the accepted modes are the + CLI's vocabulary, not the runner's. Both default to None, and when omitted + the subprocess call is exactly what it was without them. """ + if cwd is not None and not Path(cwd).is_dir(): + raise AgentError(f"agent {agent_name}: cwd is not an existing directory: {cwd}") cmd = [ "claude", "-p", @@ -96,8 +122,13 @@ def call_agent( ] if schema is not None: cmd += ["--json-schema", json.dumps(schema)] + if permission_mode is not None: + cmd += ["--permission-mode", permission_mode] + where = {"cwd": cwd} if cwd is not None else {} # unset: not even cwd=None try: - proc = subprocess.run(cmd, capture_output=True, text=True, timeout=DEFAULT_AGENT_TIMEOUT_S) + proc = subprocess.run( + cmd, capture_output=True, text=True, timeout=DEFAULT_AGENT_TIMEOUT_S, **where + ) except subprocess.TimeoutExpired as e: raise AgentError( f"agent {agent_name} timed out after {DEFAULT_AGENT_TIMEOUT_S}s: {e}" @@ -117,6 +148,8 @@ def fan_out( *, call: Any = call_agent, max_workers: int | None = None, + cwd: str | Path | None = None, + permission_mode: str | None = None, ) -> dict[str, dict[str, Any]]: """Parallel subprocess fan-out over independent agent tasks. @@ -124,13 +157,17 @@ def fan_out( envelope carrying {"_error": ""} so the orchestrator can surface coverage gaps without aborting the run. `max_workers` defaults to len(tasks) — full-width fan-out; pass it to cap concurrency. + + `cwd` and `permission_mode` apply to every task (see `call_agent`); omit + them and `call` is invoked with the same four positional arguments as before. """ if not tasks: return {} + opts = agent_options(cwd, permission_mode) results: dict[str, dict[str, Any]] = {} with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers or len(tasks)) as pool: futures = { - pool.submit(call, task.agent, task.prompt, task.schema, task.budget_usd): key + pool.submit(call, task.agent, task.prompt, task.schema, task.budget_usd, **opts): key for key, task in tasks.items() } for fut in concurrent.futures.as_completed(futures): diff --git a/pyproject.toml b/pyproject.toml index 9436c34..38b21ac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "claude-batch-runner" -version = "0.1.0" +version = "0.2.0" description = "Code-orchestrated campaign runner for LLM agent fan-out: declarative specs, parallel claude -p calls, rubric verification, and cost metering." readme = "README.md" license = { text = "MIT" } diff --git a/tests/test_deliver.py b/tests/test_deliver.py index fc49fb0..0b68e33 100644 --- a/tests/test_deliver.py +++ b/tests/test_deliver.py @@ -71,3 +71,23 @@ def test_pr_dispatch_stub_and_condition_gate(tmp_path): bad = {"escalate": {"condition": "vibes are off", "advisor": {"agent": "adv"}}} with pytest.raises(bspec.SpecError, match="escalate.condition"): deliver.run(_campaign(tmp_path, ["u"], bad), "d", call=lambda *a: pytest.fail("no call")) + + +def test_run_confines_worker_advisor_and_grader_alike(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + seen: list = [] + + def confined(agent, prompt, schema, budget, **opts): + seen.append((agent, opts)) + return fake_call(agent, prompt, schema, budget) + + c = _campaign(tmp_path, ["beta"], FULL) # beta escalates: worker -> advisor -> grader + deliver.run(c, "demo", call=confined, cwd=tmp_path, permission_mode="plan") + assert {agent for agent, _ in seen} == {"w", "adv", "v"} + assert all(opts == {"cwd": tmp_path, "permission_mode": "plan"} for _, opts in seen) + + +def test_run_without_options_keeps_the_four_argument_call_contract(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + c = _campaign(tmp_path, ["beta"], FULL) # fake_call accepts exactly four positionals + assert [u.outcome for u in deliver.run(c, "demo", call=fake_call).units] == ["escalated"] diff --git a/tests/test_driver.py b/tests/test_driver.py index d7b0638..3f29a45 100644 --- a/tests/test_driver.py +++ b/tests/test_driver.py @@ -31,6 +31,18 @@ def _task(agent: str = "health-gaps", budget: float = 5.0) -> driver.AgentTask: return driver.AgentTask(agent=agent, prompt="p", schema=SCHEMA, budget_usd=budget) +def _capture_run(monkeypatch) -> dict: + """Stub subprocess.run with a success envelope; return where cmd/kwargs land.""" + captured: dict = {} + + def fake_run(cmd, **kwargs): + captured["cmd"], captured["kwargs"] = cmd, kwargs + return _proc(stdout=json.dumps(_envelope({}))) + + monkeypatch.setattr(driver.subprocess, "run", fake_run) + return captured + + # --- call_agent --------------------------------------------------------------- @@ -96,6 +108,49 @@ def test_call_agent_non_json_stdout_raises(monkeypatch): driver.call_agent("health-gaps", "p", SCHEMA, 5.0) +# --- call_agent: cwd + permission_mode ---------------------------------------- + + +def test_call_agent_omitting_options_leaves_the_call_unchanged(monkeypatch): + captured = _capture_run(monkeypatch) + driver.call_agent("health-gaps", "analyze", SCHEMA, 5.0) + assert "--permission-mode" not in captured["cmd"] + assert "cwd" not in captured["kwargs"] # not even cwd=None — today's call, byte for byte + + +def test_call_agent_cwd_reaches_the_subprocess(monkeypatch, tmp_path): + captured = _capture_run(monkeypatch) + driver.call_agent("health-gaps", "analyze", SCHEMA, 5.0, cwd=tmp_path) + assert captured["kwargs"]["cwd"] == tmp_path + + +def test_call_agent_bad_cwd_raises_before_any_dispatch(monkeypatch, tmp_path): + def never(*a, **k): + raise AssertionError("no subprocess may start for an unusable cwd") + + monkeypatch.setattr(driver.subprocess, "run", never) + (not_a_dir := tmp_path / "f.txt").write_text("x") + for bad in (tmp_path / "nope", not_a_dir): + with pytest.raises(driver.AgentError, match=r"cwd is not an existing directory"): + driver.call_agent("health-gaps", "p", SCHEMA, 5.0, cwd=bad) + + +def test_call_agent_permission_mode_appears_in_argv(monkeypatch): + captured = _capture_run(monkeypatch) + driver.call_agent("health-gaps", "p", SCHEMA, 5.0, permission_mode="acceptEdits") + cmd = captured["cmd"] + assert cmd[cmd.index("--permission-mode") + 1] == "acceptEdits" + assert "cwd" not in captured["kwargs"] + + +def test_call_agent_both_options_together(monkeypatch, tmp_path): + captured = _capture_run(monkeypatch) + driver.call_agent("health-gaps", "p", None, 5.0, cwd=tmp_path, permission_mode="plan") + cmd = captured["cmd"] + assert cmd[cmd.index("--permission-mode") + 1] == "plan" + assert captured["kwargs"]["cwd"] == tmp_path + + # --- agent_failure_detail ----------------------------------------------------- @@ -157,6 +212,18 @@ def broken(agent, prompt, schema, budget): driver.fan_out({"x": _task()}, call=broken) +def test_fan_out_forwards_options_and_passes_none_when_unset(tmp_path): + seen: list = [] + + def fake_call(agent, prompt, schema, budget, **opts): + seen.append(opts) + return _envelope({}) + + driver.fan_out({"a": _task()}, call=fake_call) + driver.fan_out({"a": _task()}, call=fake_call, cwd=tmp_path, permission_mode="plan") + assert seen == [{}, {"cwd": tmp_path, "permission_mode": "plan"}] + + def test_fan_out_max_workers_caps_concurrency(): lock, state = threading.Lock(), {"active": 0, "peak": 0}