GodopuandClaude Opus 5.5
14d11bf062
Review of f5cadd9 (planner-reviewer-codex-01, R1): the compliance job's
registration failure was only logged, so a passing next round ended the
real loop process with rc 0. Compliance now returns 1 when it cannot be
registered or does not finish, as the correction step already does.
Re-adjudication and arbitration also return 1 when their job cannot be
registered; a registered job that yields no verdict keeps the designed
fail-closed fallback.
S4c (stubbed delegate_job, a fixture whose later rounds never pass) is
replaced by S3c: a real `python -m lib_py.loop` process whose wrapper runs
an executable that simulates the agents, refusing the compliance,
re-adjudication or arbitration registration while the next round passes.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
412 lines
17 KiB
Python
412 lines
17 KiB
Python
"""Issue #36: a delegated prompt never travels as one argv string.
|
|
|
|
Linux caps a single argv (or env) string at MAX_ARG_STRLEN = 131,072 bytes,
|
|
independent of ARG_MAX. A review prompt carries the whole diff, and the diff
|
|
collector lets up to 200,000 bytes through, so a 131,072..200,000-byte diff
|
|
killed the loop with `OSError: [Errno 7] Argument list too long: 'bash'`.
|
|
|
|
The prompt now goes by file: runner -> wrapper (--prompt-file) -> registry
|
|
(--prompt-file). These tests push prompts over the cap through the real
|
|
bash wrapper, the real registry and a real `python -m lib_py.loop` process;
|
|
subprocess is replaced only where a test is about the error handling itself
|
|
(S4, S4b), never to prove the transport.
|
|
|
|
Every test writes only under tmp_path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import errno
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
DIFF_COLLECT = SKILLS / "multi-agent-mux-loop" / "scripts" / "diff_collect.sh"
|
|
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
|
|
MAX_ARG_STRLEN = 131072
|
|
BIG = 140_000
|
|
UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
|
|
GIT_ID = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.invalid",
|
|
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.invalid"}
|
|
TRUNCATED = "!!! DIFF TRUNCATED !!!"
|
|
|
|
linux_only = pytest.mark.skipif(sys.platform != "linux", reason="MAX_ARG_STRLEN is a Linux constant")
|
|
|
|
|
|
def _env(home: Path, extra=None, drop=()) -> dict:
|
|
env = dict(os.environ)
|
|
env["DELEGATE_JOB_PYTHON"] = sys.executable
|
|
env["AGENT_PYTHON_BIN"] = sys.executable
|
|
env["HOME"] = str(home)
|
|
env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
|
|
env["PYTHONPATH"] = str(SKILLS)
|
|
for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD",
|
|
"MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT", "MAM_REAL_ROOT"):
|
|
env.pop(k, None)
|
|
env.update(UNREACHABLE)
|
|
env.update(GIT_ID)
|
|
# never this checkout's .mam (audit logs, herdr shim)
|
|
env["DELEGATE_JOB_LOGS_DIR"] = str(home / ".mam-logs")
|
|
env["WORKSPACE_ROOT"] = str(home)
|
|
if extra:
|
|
env.update({k: str(v) for k, v in extra.items()})
|
|
for k in drop:
|
|
env.pop(k, None)
|
|
return env
|
|
|
|
|
|
def _big_text(n: int = BIG) -> str:
|
|
"""ASCII with newlines, exactly n bytes."""
|
|
line = "x" * 63 + "\n"
|
|
return (line * (n // len(line) + 1))[:n]
|
|
|
|
|
|
def _dry_submit(cwd: Path, reg: Path, *prompt_args: str) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human",
|
|
"--agent-session", "herdr:x", "--dry-run",
|
|
"--registry-dir", str(reg), *prompt_args],
|
|
capture_output=True, text=True, env=_env(cwd), cwd=str(cwd), timeout=120)
|
|
|
|
|
|
def _registered_prompt(res: subprocess.CompletedProcess, reg: Path) -> str:
|
|
assert res.returncode == 0, res.stderr
|
|
m = re.search(r"registered job:\s*(\S+)", res.stdout)
|
|
assert m, res.stdout
|
|
return json.loads((reg / f"{m.group(1)}.json").read_text(encoding="utf-8"))["prompt"]
|
|
|
|
|
|
def _git(cwd: Path, *args: str) -> str:
|
|
return subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True,
|
|
check=True, env={**os.environ, **GIT_ID}).stdout.strip()
|
|
|
|
|
|
def _repo(path: Path) -> Path:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
_git(path, "init", "-q")
|
|
(path / "a.txt").write_text("a\n")
|
|
_git(path, "add", "a.txt")
|
|
_git(path, "commit", "-q", "-m", "init")
|
|
return path
|
|
|
|
|
|
def _fake_wrapper(root: Path, body: str) -> Path:
|
|
"""A wrapper at the control root, where the runner looks first."""
|
|
w = root / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
w.parent.mkdir(parents=True, exist_ok=True)
|
|
w.write_text("#!/usr/bin/env bash\n" + body, encoding="utf-8")
|
|
w.chmod(0o755)
|
|
return w
|
|
|
|
|
|
def _prompt_leftovers(root: Path) -> list:
|
|
return sorted(p.name for p in (root / ".mam").glob("prompt-*"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S1 - the real wrapper and registry take a prompt over the cap by file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_s1_wrapper_registers_a_prompt_over_the_cap_by_file(tmp_path):
|
|
text = _big_text()
|
|
pf = tmp_path / "prompt.md"
|
|
pf.write_text(text, encoding="utf-8")
|
|
reg = tmp_path / "reg"
|
|
res = _dry_submit(tmp_path, reg, "--prompt-file", str(pf))
|
|
assert _registered_prompt(res, reg) == text # byte for byte, trailing newline too
|
|
|
|
|
|
@linux_only
|
|
def test_s1b_the_same_prompt_as_an_argument_cannot_even_start(tmp_path):
|
|
"""Negative control: the failure S1 avoids is real, not a fixture artefact."""
|
|
with pytest.raises(OSError) as exc:
|
|
_dry_submit(tmp_path, tmp_path / "reg", "--prompt", _big_text())
|
|
assert exc.value.errno == errno.E2BIG
|
|
|
|
|
|
def test_s1c_prompt_still_works_and_the_two_options_exclude_each_other(tmp_path):
|
|
reg = tmp_path / "reg"
|
|
assert _registered_prompt(_dry_submit(tmp_path, reg, "--prompt", "short"), reg) == "short"
|
|
|
|
pf = tmp_path / "p.md"
|
|
pf.write_text("from file\n", encoding="utf-8")
|
|
both = _dry_submit(tmp_path, tmp_path / "reg2", "--prompt", "x", "--prompt-file", str(pf))
|
|
assert both.returncode == 1 and "not both" in both.stderr
|
|
missing = _dry_submit(tmp_path, tmp_path / "reg3", "--prompt-file", str(tmp_path / "absent.md"))
|
|
assert missing.returncode == 1 and "cannot read --prompt-file" in missing.stderr
|
|
neither = _dry_submit(tmp_path, tmp_path / "reg4")
|
|
assert neither.returncode == 1
|
|
|
|
# relative to the caller's cwd, even though the wrapper cds to --workdir
|
|
work = tmp_path / "elsewhere"
|
|
work.mkdir()
|
|
reg5 = tmp_path / "reg5"
|
|
rel = _dry_submit(tmp_path, reg5, "--prompt-file", "p.md", "--workdir", str(work))
|
|
assert _registered_prompt(rel, reg5) == "from file\n"
|
|
|
|
|
|
def test_s1d_registry_update_takes_a_prompt_file(tmp_path):
|
|
"""Hop 3 (the wrapper's loop type) updates the record by file too."""
|
|
reg = tmp_path / "reg"
|
|
assert _registered_prompt(_dry_submit(tmp_path, reg, "--prompt", "v1"), reg) == "v1"
|
|
jid = next(reg.glob("*.json")).stem
|
|
pf = tmp_path / "v2.md"
|
|
text = _big_text()
|
|
pf.write_text(text, encoding="utf-8")
|
|
res = subprocess.run([sys.executable, str(SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"),
|
|
"--registry-dir", str(reg), "update", "--job", jid, "--prompt-file", str(pf)],
|
|
capture_output=True, text=True, env=_env(tmp_path), cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
assert json.loads((reg / f"{jid}.json").read_text(encoding="utf-8"))["prompt"] == text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S2 - the runner execs the wrapper with a file, not the text
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_s2_runner_hands_the_wrapper_a_file(tmp_path, monkeypatch):
|
|
rec = tmp_path / "rec"
|
|
_fake_wrapper(tmp_path, f"""
|
|
printf '%s\\n' "$@" > "{rec}.args"
|
|
pf=""
|
|
while [ $# -gt 0 ]; do case "$1" in --prompt-file) pf="$2"; shift 2;; *) shift;; esac; done
|
|
cp "$pf" "{rec}.prompt"
|
|
echo "registered job: fake36"
|
|
""")
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
|
for k in ("MAM_REAL_ROOT", "MAM_JOBS_DIR"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
from lib_py.loop import runner as lr
|
|
r = lr.LoopRunner(["--creator", "c", "--task", "t"])
|
|
text = _big_text()
|
|
|
|
assert r.delegate_job("c", "claude", "Worker", text) == "fake36"
|
|
args = Path(f"{rec}.args").read_text().splitlines()
|
|
assert "--prompt-file" in args and "--prompt" not in args
|
|
assert Path(f"{rec}.prompt").read_text(encoding="utf-8") == text
|
|
assert _prompt_leftovers(tmp_path) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S3 - a failed delegation ends a real loop process with rc 1
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _loop_env(ctl: Path, extra=None) -> dict:
|
|
return _env(ctl, extra={
|
|
"WORKSPACE_ROOT": ctl,
|
|
"MAM_JOBS_DIR": ctl / ".mam" / "jobs",
|
|
"DELEGATE_JOB_LOGS_DIR": ctl / ".mam" / "logs",
|
|
"MAM_SESSION_BUSY_RETRY_S": "0",
|
|
"MAM_STATE_JSON": json.dumps({"herdr_sessions": [
|
|
{"name": "c", "agent": "claude", "role": "creator", "status": "running"},
|
|
{"name": "r", "agent": "claude", "role": "reviewer", "status": "running"}]}),
|
|
**(extra or {}),
|
|
}, drop=("MAM_LOOP_WORKTREE", "MAM_LOOP_BASE", "MAM_VERSION"))
|
|
|
|
|
|
def _run_loop(ctl: Path, env: dict) -> subprocess.CompletedProcess:
|
|
plan = ctl / "plan.md"
|
|
plan.write_text("# plan\n\ndo it\n", encoding="utf-8")
|
|
return subprocess.run([sys.executable, "-m", "lib_py.loop", "--creator", "c", "--task", "t36",
|
|
"--reviewer", "r", "--max-loop", "1", "--plan-file", str(plan)],
|
|
capture_output=True, text=True, env=env, cwd=str(ctl), timeout=120)
|
|
|
|
|
|
def test_s3_a_wrapper_that_fails_ends_the_loop_with_rc_1(tmp_path):
|
|
ctl = _repo(tmp_path / "ctl")
|
|
_fake_wrapper(ctl, "echo 'submit refused' >&2\nexit 2\n")
|
|
res = _run_loop(ctl, _loop_env(ctl))
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert "Failed to register Creator execution job" in res.stderr
|
|
assert "Traceback" not in res.stderr
|
|
|
|
|
|
def test_s3b_a_wrapper_that_cannot_start_ends_the_loop_with_rc_1(tmp_path):
|
|
"""exec itself fails (no bash on PATH): reported, no traceback, rc 1."""
|
|
ctl = _repo(tmp_path / "ctl")
|
|
_fake_wrapper(ctl, "echo 'registered job: never'\n")
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
git = subprocess.run(["bash", "-c", "command -v git"], capture_output=True, text=True).stdout.strip()
|
|
(bin_dir / "git").symlink_to(git)
|
|
res = _run_loop(ctl, _loop_env(ctl, {"PATH": bin_dir}))
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert "submit failed (exit 126)" in res.stderr
|
|
assert "cannot start the wrapper" in res.stderr
|
|
assert "Failed to register Creator execution job" in res.stderr
|
|
assert "Traceback" not in res.stderr
|
|
assert _prompt_leftovers(ctl) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S4 - the handler: an exec error is a failed submit, on the retry too
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def runner(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
|
monkeypatch.setenv("MAM_SESSION_BUSY_RETRY_S", "0")
|
|
for k in ("MAM_REAL_ROOT", "MAM_JOBS_DIR"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
from lib_py.loop import runner as lr
|
|
return lr, lr.LoopRunner(["--creator", "c", "--task", "t"])
|
|
|
|
|
|
def test_s4_an_exec_error_is_a_failed_submit(runner, tmp_path, monkeypatch, capsys):
|
|
lr, r = runner
|
|
|
|
def boom(*a, **k):
|
|
raise OSError(errno.E2BIG, "Argument list too long")
|
|
|
|
monkeypatch.setattr(lr.subprocess, "run", boom)
|
|
assert r.delegate_job("c", "claude", "Worker", "p" * 1000) is None
|
|
err = capsys.readouterr().err
|
|
assert "submit failed (exit 126)" in err
|
|
assert "Argument list too long" in err and "prompt 1000 bytes" in err
|
|
assert _prompt_leftovers(tmp_path) == []
|
|
|
|
|
|
def test_s4b_an_exec_error_on_the_busy_retry_is_reported_not_raised(runner, tmp_path, monkeypatch, capsys):
|
|
lr, r = runner
|
|
calls = []
|
|
|
|
def run(cmd, **k):
|
|
calls.append(cmd)
|
|
if len(calls) == 1:
|
|
return subprocess.CompletedProcess(cmd, 5, "subscriber pid: 1\n", "session busy\n")
|
|
raise OSError(errno.EAGAIN, "Resource temporarily unavailable")
|
|
|
|
monkeypatch.setattr(lr.subprocess, "run", run)
|
|
assert r.delegate_job("c", "claude", "Worker", "p") is None
|
|
assert len(calls) == 2
|
|
err = capsys.readouterr().err
|
|
assert "submit failed (exit 5)" in err
|
|
assert "retry failed (exit 126)" in err
|
|
assert "cannot start the wrapper" in err
|
|
assert _prompt_leftovers(tmp_path) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S3c - a job that cannot be registered after the rebuttal ends the loop with
|
|
# rc 1, even when the next review round would pass (real loop process; only
|
|
# the agents' outcomes are simulated, by an executable behind the wrapper)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
FAKE_AGENT = r"""
|
|
import json, sys
|
|
from pathlib import Path
|
|
args = sys.argv[3:] # argv[1] = kind to refuse, argv[2] = "submit"
|
|
opts = dict(zip(args[::2], args[1::2]))
|
|
prompt = Path(opts["--prompt-file"]).read_text(encoding="utf-8")
|
|
refuse = sys.argv[1]
|
|
jobs = Path(opts["--registry-dir"])
|
|
jobs.mkdir(parents=True, exist_ok=True)
|
|
kind = ("comply" if "재심 결과 기각" in prompt else
|
|
"adjudication" if "re-adjudication" in prompt else
|
|
"arbitration" if "재정(arbitration)" in prompt else
|
|
"correction" if "[이의제기 채널]" in prompt else
|
|
opts["--role"])
|
|
if kind == refuse:
|
|
print(kind + " registration refused", file=sys.stderr)
|
|
sys.exit(2)
|
|
if kind == "adjudication":
|
|
report = "no\n\n[ADJUDICATION: OVERRULED]\n"
|
|
elif kind == "arbitration":
|
|
report = "reviewer is right\n\n[ARBITRATION: REVIEWER]\n"
|
|
elif kind == "correction":
|
|
report = "I disagree\n\n[REBUT: r]\n"
|
|
elif kind == "Reviewer" and "Self-Review" in prompt:
|
|
report = "ok\n\n[VERDICT: PASS]\n"
|
|
elif kind == "Reviewer": # the first cross review objects, every later one passes
|
|
counter = Path.cwd() / "review-count"
|
|
n = int(counter.read_text()) + 1 if counter.exists() else 1
|
|
counter.write_text(str(n))
|
|
report = "line 3 is wrong\n\n[VERDICT: NOT PASS]\n" if n == 1 else "ok\n\n[VERDICT: PASS]\n"
|
|
elif kind == "Planner":
|
|
report = "# plan\n\ndo it\n"
|
|
else:
|
|
report = "done\n"
|
|
jid = "j%d" % len(list(jobs.glob("*.json")))
|
|
(jobs / (jid + ".json")).write_text(json.dumps({"job_id": jid, "status": "completed"}))
|
|
dest = jobs / jid / "agent-reports"
|
|
dest.mkdir(parents=True)
|
|
(dest / "report-final.md").write_text(report, encoding="utf-8")
|
|
print("registered job: " + jid)
|
|
"""
|
|
|
|
|
|
@pytest.mark.parametrize("refuse, message", [
|
|
("comply", "Failed to register Creator compliance job."),
|
|
("adjudication", "Failed to register re-adjudication job for 'r'."),
|
|
("arbitration", "Failed to register Planner arbitration job."),
|
|
])
|
|
def test_s3c_a_registration_failure_after_a_rebuttal_is_never_a_success(tmp_path, refuse, message):
|
|
ctl = _repo(tmp_path / "ctl")
|
|
agent = ctl / "fake-agent.py"
|
|
agent.write_text(FAKE_AGENT, encoding="utf-8")
|
|
_fake_wrapper(ctl, f'exec "{sys.executable}" "{agent}" {refuse} "$@"\n')
|
|
extra = {}
|
|
argv = ["--max-loop", "2"]
|
|
if refuse == "arbitration": # arbitration needs --plan and a planner
|
|
extra["MAM_STATE_JSON"] = json.dumps({"herdr_sessions": [
|
|
{"name": "c", "agent": "claude", "role": "creator", "status": "running"},
|
|
{"name": "r", "agent": "claude", "role": "reviewer", "status": "running"},
|
|
{"name": "p", "agent": "claude", "role": "planner", "status": "running"}]})
|
|
argv += ["--plan", "--plan-talk", "0"]
|
|
else:
|
|
plan = ctl / "plan.md"
|
|
plan.write_text("# plan\n\ndo it\n", encoding="utf-8")
|
|
argv += ["--plan-file", str(plan)]
|
|
res = subprocess.run([sys.executable, "-m", "lib_py.loop", "--creator", "c", "--task", "t36",
|
|
"--reviewer", "r", *argv],
|
|
capture_output=True, text=True, env=_loop_env(ctl, extra), cwd=str(ctl), timeout=120)
|
|
assert message in res.stderr, res.stdout + res.stderr
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert "100% PASS" not in res.stdout
|
|
assert "Traceback" not in res.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# S5 - the former blind spot: a 131,072..200,000-byte diff goes through whole
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _collect(repo: Path, base: str, head: str, extra=None) -> bytes:
|
|
env = {k: v for k, v in os.environ.items() if k not in ("MAM_DIFF_MAX_BYTES", "MAM_DIFF_MAX_LINES")}
|
|
env.update(extra or {})
|
|
res = subprocess.run(["bash", "-c", 'source "$1" && mam_collect_changes_diff "$2" "$3" "$4"',
|
|
"_", str(DIFF_COLLECT), str(repo), base, head],
|
|
capture_output=True, env=env, timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
return res.stdout
|
|
|
|
|
|
def test_s5_a_diff_in_the_former_blind_spot_is_delivered_whole(tmp_path):
|
|
repo = _repo(tmp_path / "src")
|
|
base = _git(repo, "rev-parse", "HEAD")
|
|
(repo / "big.txt").write_text(_big_text(150_000), encoding="utf-8")
|
|
_git(repo, "add", "big.txt")
|
|
_git(repo, "commit", "-q", "-m", "grow")
|
|
head = _git(repo, "rev-parse", "HEAD")
|
|
|
|
diff = _collect(repo, base, head)
|
|
assert TRUNCATED.encode() not in diff
|
|
assert MAX_ARG_STRLEN < len(diff) <= 200_000
|
|
|
|
pf = tmp_path / "diff.md"
|
|
pf.write_bytes(diff)
|
|
reg = tmp_path / "reg"
|
|
assert _registered_prompt(_dry_submit(tmp_path, reg, "--prompt-file", str(pf)), reg) == diff.decode("utf-8")
|
|
|
|
# the collector cap itself is unchanged: over it, the diff is still truncated and marked
|
|
assert TRUNCATED.encode() in _collect(repo, base, head, {"MAM_DIFF_MAX_BYTES": "100000"})
|