Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
30 changes: 30 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion claude_batch_runner/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
AgentError,
AgentTask,
agent_failure_detail,
agent_options,
call_agent,
fan_out,
)
Expand Down Expand Up @@ -43,11 +44,12 @@
"VerifyError",
"Worker",
"agent_failure_detail",
"agent_options",
"call_agent",
"fan_out",
"load_campaign",
"load_rubric",
"verify_unit",
]

__version__ = "0.1.0"
__version__ = "0.2.0"
33 changes: 24 additions & 9 deletions claude_batch_runner/deliver.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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}

Expand All @@ -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

Expand All @@ -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)
Expand Down
41 changes: 39 additions & 2 deletions claude_batch_runner/driver.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
"""
Expand All @@ -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"))
Expand Down Expand Up @@ -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 <name>` 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",
Expand All @@ -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}"
Expand All @@ -117,20 +148,26 @@ 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.

Returns: {key: envelope}. Tasks that fail are reported with a synthetic
envelope carrying {"_error": "<msg>"} 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):
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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" }
Expand Down
20 changes: 20 additions & 0 deletions tests/test_deliver.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
67 changes: 67 additions & 0 deletions tests/test_driver.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ---------------------------------------------------------------


Expand Down Expand Up @@ -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 -----------------------------------------------------


Expand Down Expand Up @@ -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}

Expand Down
Loading