tests/test_issue36_prompt_transport.py

@@ -0,0 +1,386 @@
+"""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, S4c), 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) == []
+
+
+# ---------------------------------------------------------------------------
+# S4c - a compliance job that cannot be registered is logged
+# ---------------------------------------------------------------------------
+
+def _drive_to_compliance(lr, r, monkeypatch, comply_jid):
+    """Self-review PASS, cross review NOT PASS, the correction rebuts, the
+    reviewer overrules, and the compliance delegation returns comply_jid."""
+    stub = Path(r.workspace_root) / "stub"
+    stub.mkdir(exist_ok=True)
+    files = {}
+
+    def fake_delegate(session, agent_type, role, prompt):
+        jid = f"j{len(files)}"
+        if "재심 결과 기각" in prompt:
+            return comply_jid
+        if prompt.startswith(lr.SELF_REVIEW_PROMPT) or lr.SELF_REVIEW_PROMPT in prompt:
+            body = "ok\n\n[VERDICT: PASS]\n"
+        elif "re-adjudication" in prompt:
+            body = "no\n\n[ADJUDICATION: OVERRULED]\n"
+        elif "[이의제기 채널]" in prompt:
+            body = "I disagree\n\n[REBUT: r]\n"
+        elif role == "Reviewer":
+            body = "line 3 is wrong\n\n[VERDICT: NOT PASS]\n"
+        else:
+            body = "done\n"
+        f = stub / f"{jid}.md"
+        f.write_text(body, encoding="utf-8")
+        files[jid] = f
+        return jid
+
+    monkeypatch.setattr(lr.signal, "signal", lambda *a, **k: None)
+    monkeypatch.setattr(r, "delegate_job", fake_delegate)
+    monkeypatch.setattr(r, "wait_for_job", lambda *a, **k: True)
+    monkeypatch.setattr(r, "find_report", lambda jid: files.get(jid))
+    return r.run()
+
+
+@pytest.mark.parametrize("comply_jid", [None, "jc"])
+def test_s4c_a_failed_compliance_delegation_is_logged(tmp_path, monkeypatch, capsys, comply_jid):
+    ctl = _repo(tmp_path / "ctl")
+    for k, v in _loop_env(ctl).items():
+        if k in ("WORKSPACE_ROOT", "MAM_JOBS_DIR", "MAM_STATE_JSON"):
+            monkeypatch.setenv(k, v)
+    for k in ("MAM_REAL_ROOT", "MAM_LOOP_WORKTREE", "MAM_LOOP_BASE"):
+        monkeypatch.delenv(k, raising=False)
+    from lib_py.loop import runner as lr
+    # two rounds: the correction (and so the rebuttal) runs only between rounds
+    r = lr.LoopRunner(["--creator", "c", "--task", "t36", "--reviewer", "r", "--max-loop", "2"])
+    rc = _drive_to_compliance(lr, r, monkeypatch, comply_jid)
+    out = capsys.readouterr()
+    assert "Creator must comply" in out.out + out.err
+    assert ("Failed to register Creator compliance job" in out.err) == (comply_jid is None)
+    assert rc == 1  # the review round was not passed either way; the flow is unchanged
+
+
+# ---------------------------------------------------------------------------
+# 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"})