multi-agent-mux/tests/test_issue36_prompt_transport.py

"""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, os, 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.get("--registry-dir", ".mam/jobs"))  # the shell loop passes none
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"])
review = "Self-Review" if "Self-Review" in prompt else kind
with open(".mam/calls.log", "a", encoding="utf-8") as log:  # what reached the agent, in bytes
    log.write("%s\t%d\n" % (review, len(prompt.encode("utf-8"))))
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(".mam/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\n" + "p" * int(os.environ.get("FAKE_PLAN_BYTES", "5")) + "\n"
else:
    grow = int(os.environ.get("FAKE_GROW_BYTES", "0"))
    if grow and kind == "Worker":  # the implementation: one big new file in the tree
        Path("big.txt").write_text(("g" * 63 + "\n") * (grow // 64), encoding="utf-8")
    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"})


# ---------------------------------------------------------------------------
# S6 - lib.sh's submit helper (create_session's onboarding and --submit-job)
# and the delegation CLI behind it take a prompt over the cap by file
# ---------------------------------------------------------------------------

LIB_SH = SKILLS / "lib.sh"


def _helper_env(root: Path, extra=None) -> dict:
    return _env(root, extra={"MAM_JOBS_DIR": root / "reg", **(extra or {})})


def test_s6_delegation_cli_submit_takes_a_prompt_file(tmp_path):
    text = _big_text()
    pf = tmp_path / "p.md"
    pf.write_text(text, encoding="utf-8")
    res = subprocess.run([sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt-file", str(pf),
                          "--agent", "claude-code", "--agent-session", "herdr:x"],
                         capture_output=True, text=True, env=_helper_env(tmp_path), cwd=str(tmp_path), timeout=60)
    assert res.returncode == 0, res.stderr
    jid = res.stdout.strip().splitlines()[-1]
    assert json.loads((tmp_path / "reg" / f"{jid}.json").read_text(encoding="utf-8"))["prompt"] == text

    both = subprocess.run([sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "x",
                           "--prompt-file", str(pf)],
                          capture_output=True, text=True, env=_helper_env(tmp_path), cwd=str(tmp_path), timeout=60)
    assert both.returncode == 2 and "not allowed with argument" in both.stderr


@pytest.mark.parametrize("delegation_py", ["1", "0"])  # the Python helper and its legacy twin
def test_s6b_delegate_submit_job_passes_a_prompt_over_the_cap(tmp_path, delegation_py):
    text = _big_text()
    pf = tmp_path / "p.md"
    pf.write_text(text, encoding="utf-8")
    # The prompt reaches the shell function as a bash argument (no exec, no cap),
    # exactly as create_session.sh hands it over.
    res = subprocess.run(["bash", "-c", 'source "$1" && delegate_submit_job "$(cat "$2")" claude-code herdr:x',
                          "_", str(LIB_SH), str(pf)],
                         capture_output=True, text=True, cwd=str(tmp_path), timeout=60,
                         env=_helper_env(tmp_path, {"MAM_DELEGATION_PY": delegation_py}))
    assert res.returncode == 0, res.stderr
    jid = res.stdout.strip().splitlines()[-1]
    assert json.loads((tmp_path / "reg" / f"{jid}.json").read_text(encoding="utf-8"))["prompt"] == text
    assert sorted(p.name for p in (tmp_path / ".mam").glob("prompt-*")) == []


# ---------------------------------------------------------------------------
# S7 - the legacy shell loop (MAM_LOOP_PY=0): the real run_loop.sh hands every
# prompt over by file, and a registration failure never ends it with rc 0.
# run_loop.sh always runs the wrapper next to itself, so the skills tree is
# copied under tmp_path (as the mam_sandbox fixture does) and only that copy's
# wrapper is replaced by the agent simulator.
# ---------------------------------------------------------------------------

def _shell_ctl(tmp_path: Path, refuse: str) -> Path:
    import shutil
    ctl = tmp_path / "ctl"
    shutil.copytree(SKILLS, ctl / ".agents" / "skills", ignore=shutil.ignore_patterns("__pycache__"))
    (ctl / ".gitignore").write_text(".mam/\n.agents/reports/\n")
    agent = ctl / "fake-agent.py"
    agent.write_text(FAKE_AGENT, encoding="utf-8")
    _fake_wrapper(ctl, f'exec "{sys.executable}" "{agent}" {refuse} "$@"\n')
    _repo(ctl)  # the base holds the copied tree, so the diff is only what the "Creator" writes
    _git(ctl, "add", "-A")
    _git(ctl, "commit", "-q", "-m", "skills")
    (ctl / ".mam").mkdir()
    (ctl / ".mam" / "agent-sessions.yaml").write_text(
        "herdr_sessions:\n"
        "- {name: c, agent: claude, role: creator, status: running}\n"
        "- {name: r, agent: claude, role: reviewer, status: running}\n"
        "- {name: p, agent: claude, role: planner, status: running}\n", encoding="utf-8")
    return ctl


def _run_shell_loop(ctl: Path, argv, extra=None) -> subprocess.CompletedProcess:
    env = _env(ctl, extra={
        "MAM_LOOP_PY": "0", "MAM_LOOP_NO_FREEZE": "1",  # no /tmp snapshot of the skills
        "MAM_LOOP_MARKER": ctl / ".mam" / "loop-marker",
        "AGENT_SESSIONS_YAML": ctl / ".mam" / "agent-sessions.yaml",
        "WORKSPACE_ROOT": ctl, **(extra or {})},
        drop=("MAM_LOOP_WORKTREE", "MAM_LOOP_BASE", "MAM_VERSION", "MAM_STATE_JSON"))
    run_loop = ctl / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
    return subprocess.run(["bash", str(run_loop), "--creator", "c", "--task", "t36", "--reviewer", "r", *argv],
                          capture_output=True, text=True, env=env, cwd=str(ctl), timeout=300)


def _calls(ctl: Path) -> list:
    return [tuple(line.split("\t")) for line in (ctl / ".mam" / "calls.log").read_text().splitlines()]


def test_s7_shell_loop_hands_prompts_over_the_cap_to_the_wrapper(tmp_path):
    ctl = _shell_ctl(tmp_path, "none")
    # a 140,000-byte plan makes the execution prompt, and a 150,000-byte new
    # file makes the collected diff (so the review prompt), exceed the cap
    res = _run_shell_loop(ctl, ["--plan", "--plan-talk", "0", "--max-loop", "2"],
                          {"FAKE_PLAN_BYTES": BIG, "FAKE_GROW_BYTES": 150_000})
    assert res.returncode == 0, res.stdout + res.stderr
    assert "Argument list too long" not in res.stderr
    sizes = {}
    for kind, n in _calls(ctl):
        sizes.setdefault(kind, []).append(int(n))
    assert max(sizes["Worker"]) > MAX_ARG_STRLEN      # execution prompt with the plan
    assert max(sizes["Reviewer"]) > MAX_ARG_STRLEN    # cross review with the diff
    assert sorted(p.name for p in (ctl / ".mam").glob("prompt-*")) == []


@pytest.mark.parametrize("refuse, argv", [
    ("Worker", ["--max-loop", "1"]),   # the implementation job
    ("comply", ["--max-loop", "2"]),   # the compliance job after an overruled rebuttal; round 2 would pass
])
def test_s7b_shell_loop_ends_nonzero_when_a_job_cannot_be_registered(tmp_path, refuse, argv):
    """The shell loop runs under set -e: a failed submit inside $(...) ends it
    at that line with the wrapper's code, so the cause must reach stderr."""
    ctl = _shell_ctl(tmp_path, refuse)
    res = _run_shell_loop(ctl, argv)
    out = res.stdout + res.stderr
    assert res.returncode == 2, out
    assert f"{refuse} registration refused" in res.stderr
    assert "delegate_job_safe failed (exit 2)" in res.stderr
    assert "100% PASS" not in out