"""Issue #17 item 4: the delegate-job wrapper's decisions moved from bash to
Python (lib_py/delegation/wrapper.py, lib_py/ints.py) with no behaviour change.

- U: unit tables for the new Python code (every input -> label / exit code,
  invalid input, boundaries).
- C: comparison - the same input through the old bash functions (base
  7209750, tests/data/issue17_item4_old_decisions.sh) and through the new
  wrapper gives the same output, the same FINAL_JOB_STATUS and the same
  registry.py / publish_event.py calls.
- S: static guard - the wrapper no longer carries the moved decisions.
Runs on macOS and Linux: no /proc, no broker (registry.py / publish_event.py
are stand-ins that log their argv).
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import time
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"
LIB_INT = SKILLS / "lib_int.sh"
INTS_PY = SKILLS / "lib_py" / "ints.py"
OLD = REPO_ROOT / "tests" / "data" / "issue17_item4_old_decisions.sh"

if str(SKILLS) not in sys.path:
    sys.path.insert(0, str(SKILLS))

from lib_py import ints  # noqa: E402
from lib_py.delegation import wrapper as decide  # noqa: E402


# ---------------------------------------------------------------------------
# U - unit tables
# ---------------------------------------------------------------------------

def test_u_subscriber_codes_mirror_job_subscriber():
    from lib_py.delegation import job_subscriber
    assert (decide.SUB_RC_COMPLETED, decide.SUB_RC_ERROR, decide.SUB_RC_NO_EVENT, decide.SUB_RC_BROKER_UNAVAILABLE,
            decide.SUB_RC_WALL_CLOCK) == (job_subscriber.RC_COMPLETED, job_subscriber.RC_ERROR, job_subscriber.RC_NO_EVENT,
                                         job_subscriber.RC_BROKER_UNAVAILABLE, job_subscriber.RC_WALL_CLOCK)


@pytest.mark.parametrize("disk", [None, "", "pending", "running", "completed", "error", "cancelled", "garbage"])
@pytest.mark.parametrize("rc,want", [(0, "completed"), (1, "error"), (2, "timeout"), (3, None), (4, "timeout"),
                                     (5, "timeout"), (124, "timeout"), (-1, "timeout")])
def test_u_sub_status_table(tmp_path, rc, want, disk):
    if disk == "garbage":
        (tmp_path / "j.json").write_text("{not json")
    elif disk is not None:
        (tmp_path / "j.json").write_text(json.dumps({"status": disk}))
    if want is None:  # broker unavailable: a terminal record on disk wins
        want = disk if disk in ("completed", "error", "cancelled") else "broker_unavailable"
    assert decide.sub_status(rc, str(tmp_path), "j") == want


@pytest.mark.parametrize("rc,up", [(0, True), (3, True), (1, False), (2, False), (4, False), (143, False)])
def test_u_subscriber_up(rc, up):
    assert decide.subscriber_up(rc) is up


@pytest.mark.parametrize("fc", [None, "", "1", "0", "yes"])
@pytest.mark.parametrize("status,code", [("completed", 0), ("error", 1), ("timeout", 2), ("broker_unavailable", 3),
                                         ("cancelled", 5), ("running", 1), ("", 1), ("pending", 1)])
def test_u_exit_code_table(monkeypatch, status, code, fc):
    if fc is None:
        monkeypatch.delenv("MAM_DELEGATE_FAIL_CLOSED", raising=False)
    else:
        monkeypatch.setenv("MAM_DELEGATE_FAIL_CLOSED", fc)
    assert decide.exit_code(status) == (0 if fc == "0" else code)


@pytest.mark.parametrize("rc,want", [(0, 0), (1, 1), (2, 2), (3, 3), (4, 2), (5, 5)])
def test_u_wait_rc(rc, want):
    assert decide.wait_rc(rc) == want


@pytest.mark.parametrize("env,session,agent,watch,msg", [
    (None, "herdr:c", "claude", True, ""),
    ("", "herdr:c", "claude", True, ""),
    ("1", "c", "codex", True, ""),
    ("0", "herdr:c", "claude", False, ""),
    ("0", "remote:r", "claude", False, ""),
    (None, "remote:r", "claude", False, "remote job"),
    (None, "remote:human", "human", False, "remote job"),
    (None, "herdr:c", "human", False, "human job"),
    (None, "herdr:human", "claude", False, "human job"),
    (None, "human", "claude", False, "human job"),
    (None, "herdr:herdr:human", "claude", True, ""),  # ${session#herdr:} strips one prefix
])
def test_u_stall_watch_table(monkeypatch, capsys, env, session, agent, watch, msg):
    if env is None:
        monkeypatch.delenv("MAM_STALL_WATCH", raising=False)
    else:
        monkeypatch.setenv("MAM_STALL_WATCH", env)
    assert decide.stall_watch(session, agent) is watch
    err = capsys.readouterr().err
    assert (msg in err) if msg else err == ""


def _events(reg: Path, jid: str, *stamps: str) -> None:
    with open(reg / f"{jid}.events.log", "w") as fh:
        for s in stamps:
            fh.write((json.dumps({"event": "progress", "timestamp": s}) if s else "") + "\n")


def test_u_recent_disk_heartbeat(tmp_path):
    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    old = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - 100))
    assert decide.recent_disk_heartbeat(str(tmp_path), "j", 10) is False  # no file
    _events(tmp_path, "j", old, now, "")
    assert decide.recent_disk_heartbeat(str(tmp_path), "j", 10) is True   # the last non-blank line
    _events(tmp_path, "j", now, old)
    assert decide.recent_disk_heartbeat(str(tmp_path), "j", 10) is False
    assert decide.recent_disk_heartbeat(str(tmp_path), "j", 200) is True
    for bad in ("{not json", json.dumps({"event": "x"}), json.dumps({"timestamp": "yesterday"})):
        (tmp_path / "j.events.log").write_text(bad + "\n")
        assert decide.recent_disk_heartbeat(str(tmp_path), "j", 10**9) is False


def _ns(**kw):
    import argparse
    base = dict(job="j", registry_dir="", status="timeout", start_time=None, timeout=60, idle_timeout=20,
                agent="claude", session="herdr:c")
    base.update(kw)
    return argparse.Namespace(**base)


@pytest.mark.parametrize("status,session,agent,step", [
    ("timeout", "herdr:c", "claude", "probe"), ("timeout", "c", "codex", "probe"),
    ("timeout", "remote:c", "claude", "go"), ("timeout", "herdr:human", "claude", "go"),
    ("timeout", "herdr:c", "human", "go"), ("timeout", "", "claude", "go"), ("timeout", "herdr:", "claude", "go"),
    ("error", "herdr:c", "claude", "go"), ("broker_unavailable", "herdr:c", "claude", "go"),
    ("cancelled", "herdr:c", "claude", "go"),
])
def test_u_finalize_begin_steps(tmp_path, monkeypatch, status, session, agent, step):
    monkeypatch.delenv("MAM_DELEGATE_FAIL_CLOSED", raising=False)
    out = decide.finalize_begin(_ns(registry_dir=str(tmp_path), status=status, session=session, agent=agent,
                                    start_time=1000, timeout=60, idle_timeout=20)).split()
    assert out[0] == step and out[1] == status and out[3] == "1060"
    assert abs(int(out[2]) - (int(time.time()) + 20)) <= 1


def test_u_finalize_begin_keep_done_and_disk(tmp_path, monkeypatch, capsys):
    monkeypatch.delenv("MAM_DELEGATE_FAIL_CLOSED", raising=False)
    reg = str(tmp_path)
    assert decide.finalize_begin(_ns(registry_dir=reg, status="completed")) == "keep"
    monkeypatch.setenv("MAM_DELEGATE_FAIL_CLOSED", "0")
    assert decide.finalize_begin(_ns(registry_dir=reg, status="error")) == "keep"
    monkeypatch.setenv("MAM_DELEGATE_FAIL_CLOSED", "1")
    (tmp_path / "j.json").write_text(json.dumps({"status": "completed"}))
    assert decide.finalize_begin(_ns(registry_dir=reg, status="timeout")) == "done completed"
    for disk in ("error", "cancelled"):
        (tmp_path / "j.json").write_text(json.dumps({"status": disk}))
        assert decide.finalize_begin(_ns(registry_dir=reg, status="timeout")).split()[:2] == ["probe", disk]
    (tmp_path / "j.json").write_text(json.dumps({"status": "running"}))
    (tmp_path / "j.stall.out").write_text("a\nb\nc\nd\n")
    capsys.readouterr()
    assert decide.finalize_begin(_ns(registry_dir=reg, status="error")).split()[:2] == ["go", "error"]
    assert capsys.readouterr().err == "stall watch log for j (last 3 lines):\nb\nc\nd\n"


def test_u_cli_refuses_bad_input(tmp_path):
    env = {**os.environ, "PYTHONPATH": str(SKILLS)}
    for argv in (["sub-status", "x", "--job", "j", "--registry-dir", "."], ["sub-status", "1"], ["exit-code"],
                 ["wait-rc", "1.5"], ["nope"], [], ["finalize-end", "--job", "j"]):
        res = subprocess.run([sys.executable, "-m", "lib_py.delegation.wrapper", *argv], capture_output=True,
                             text=True, env=env, timeout=60)
        assert res.returncode == 2 and res.stdout == "", (argv, res.stdout, res.stderr)


@pytest.mark.parametrize("value,minimum,want", [
    ("999999999", 1, 999999999), ("1000000000", 1, None), ("0999999999", 1, 999999999), ("1", 1, 1),
    ("0", 1, None), ("0", 0, 0), ("00", 0, 0), (" 7\n", 7, 7), ("6", 7, None), ("+5", 0, None), ("5 5", 0, None),
    ("٥", 0, None), ("", 0, None), (None, 0, None)])
def test_u_ints_boundaries(value, minimum, want):
    assert ints.positive_int(value, minimum) == want


def test_u_ints_cli(tmp_path):
    def run(*a):
        return subprocess.run([sys.executable, "-I", str(INTS_PY), *a], capture_output=True, text=True, timeout=60)
    assert run("int", "0700", "1").stdout == "700\n"
    assert run("int", "abc", "1").returncode == 1
    res = run("knob", "K", "abc", "30", "1")
    assert res.stdout == "30\n" and "WARNING: K=abc ignored (not a decimal integer >= 1); using 30" in res.stderr
    assert run("knob", "K", "", "$X", "0").stdout == "$X\n"  # the default is printed as given
    for bad in ([], ["int", "1"], ["knob", "K", "1", "2"], ["other", "1", "2"]):
        assert run(*bad).returncode == 2, bad


# ---------------------------------------------------------------------------
# C - old bash vs new wrapper, same input, same result
# ---------------------------------------------------------------------------

def _shim(tmp_path: Path) -> Path:
    """The wrapper's $PY: registry.py / publish_event.py / job_subscriber.py and
    the stall watcher are stand-ins that log their argv; anything else is the
    real interpreter."""
    shim = tmp_path / "py-shim"
    shim.write_text(
        "#!/bin/sh\n"
        'case "$1" in\n'
        '  */scripts/registry.py) shift; echo "registry.py $*" >>"$CALL_LOG"; echo close-out; echo close-err >&2; '
        'exit "${CLOSE_RC:-0}";;\n'
        '  */scripts/publish_event.py) shift; echo "publish_event.py $*" >>"$CALL_LOG"; echo pub-out; '
        'echo pub-err >&2; exit 0;;\n'
        '  */scripts/job_subscriber.py) shift; echo "job_subscriber.py $*" >>"$CALL_LOG"; exit "${SUB_RC:-0}";;\n'
        "esac\n"
        'case "$*" in *lib_py.stall_watch*) echo "stall_watch $*" >>"$CALL_LOG"; exit 0;; esac\n'
        f'exec "{sys.executable}" "$@"\n')
    shim.chmod(0o755)
    return shim


def _env(tmp_path: Path, **extra) -> dict:
    # WORKSPACE_ROOT is this test's tmp_path, never inherited: sourcing the
    # wrapper sources lib.sh, which rewrites $WORKSPACE_ROOT/.mam/shim/herdr -
    # and unset, lib.sh would derive the checkout itself (Issue #17 follow-up).
    env = {k: v for k, v in os.environ.items()
           if not k.startswith(("HERDR_", "MAM_", "SKS_")) and k not in ("PYTHONPATH", "WORKSPACE_ROOT")}
    env.update(MAM_DELEGATE_JOB_NO_MAIN="1", MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"), HOME=str(tmp_path),
               WORKSPACE_ROOT=str(tmp_path), DELEGATE_JOB_LOGS_DIR=str(tmp_path / "logs"))
    env.update({k: str(v) for k, v in extra.items() if v is not None})
    return env


def _run_both(tmp_path: Path, body: str, env_extra: dict, setup=None):
    """Run `body` after sourcing the new wrapper, once as is and once with the
    old functions sourced over it. Returns [(rc, stdout, stderr, calls)] old, new."""
    out = []
    for which in ("old", "new"):
        work = tmp_path / which
        reg = work / "reg"
        reg.mkdir(parents=True)
        if setup:
            setup(reg)
        log = work / "calls.log"
        log.write_text("")
        shim = _shim(work)
        script = (f'source "{WRAPPER}"\n' + (f'source "{OLD}"\n' if which == "old" else "")
                  + f'REGISTRY_DIR="{reg}"; PY="{shim}"\n' + body)
        res = subprocess.run(["bash", "-c", script], capture_output=True, text=True, cwd=str(work), timeout=120,
                             env=_env(work, CALL_LOG=log, DELEGATE_JOB_PYTHON=shim, **env_extra))
        norm = lambda t: t.replace(str(work), "<W>")  # noqa: E731
        calls = re.sub(r"--watch-pid \d+", "--watch-pid <pid>", norm(log.read_text()))
        out.append((res.returncode, norm(res.stdout), norm(res.stderr), calls))
    return out


def _finalize_body(status, sub_rc, phase, *, typ="direct", agent="claude", session="herdr:c", timeout=5, idle=0,
                   start_offset=-100, tok="tok1", reason="", fifth=None):
    return ("_agent_still_working() { return 1; }\n"
            "_sanitize_herdr_agent_name() { printf '%s\\n' \"$1\"; }\n"
            f'TYPE="{typ}"; AGENT="{agent}"; AGENT_SESSION="{session}"; TIMEOUT={timeout}; IDLE_TIMEOUT={idle}\n'
            f'START_TIME=$(( $(date +%s) + {start_offset} )); _disp_tok="{tok}"; PREFLIGHT_REASON="{reason}"\n'
            'FINAL_JOB_STATUS="unset"\n'
            f'finalize_unfinished j1 "{status}" "{sub_rc}" "{phase}"' + (f' "{fifth}"' if fifth else "") + "\n"
            'echo "FINAL=$FINAL_JOB_STATUS"\n')


def _disk(status):
    def setup(reg):
        if status is not None:
            (reg / "j1.json").write_text(json.dumps({"job_id": "j1", "status": status}))
    return setup


PAIRS = [("error", 1), ("timeout", 2), ("timeout", 4), ("broker_unavailable", 3), ("completed", 0), ("error", 2)]
PHASES = ["direct", "inject", "preflight", "idle_wait", "subscriber", "worker"]


@pytest.mark.parametrize("phase", PHASES)
@pytest.mark.parametrize("status,sub_rc", PAIRS)
def test_c_finalize_labels_closes_and_events(tmp_path, status, sub_rc, phase):
    """Past the wall clock: every status x subscriber rc x phase."""
    old, new = _run_both(tmp_path, _finalize_body(status, sub_rc, phase, reason="target_not_found"), {},
                         setup=_disk("running"))
    assert old == new, (old, new)
    assert "FINAL=" in new[1]


VARIANTS = [
    # (status, sub_rc, phase, body kwargs, disk, env)
    ("timeout", 2, "direct", dict(start_offset=0, timeout=60, idle=0), "running", {}),       # idle clock
    ("timeout", 2, "direct", dict(start_offset=0, timeout=60, idle=0, agent="human"), "running", {}),
    ("timeout", 2, "direct", dict(session="remote:r"), "running", {}),
    ("timeout", 2, "worker", dict(fifth="herdr:other"), "running", {}),
    ("timeout", 2, "direct", dict(typ="loop"), "running", {}),
    ("error", 1, "inject", dict(typ="discuss"), "running", {}),
    ("timeout", 2, "direct", {}, "completed", {}),
    ("timeout", 2, "direct", {}, "error", {}),
    ("broker_unavailable", 3, "direct", {}, "cancelled", {}),
    ("broker_unavailable", 3, "direct", {}, None, {}),
    ("error", 1, "direct", {}, "garbage", {}),
    ("timeout", 4, "direct", dict(tok=""), "running", {}),
    ("error", 1, "preflight", dict(reason=""), "running", {}),
    ("error", 1, "preflight", dict(reason="target_not_found", tok=""), "running", {}),
    ("error", 2, "idle_wait", dict(reason="target_not_found"), "running", {}),
    ("timeout", 2, "direct", {}, "running", {"CLOSE_RC": "4"}),
    ("error", 1, "inject", {}, "running", {"CLOSE_RC": "1"}),
    ("error", 1, "direct", {}, "running", {"MAM_DELEGATE_FAIL_CLOSED": "0"}),
    ("completed", 0, "direct", {}, "error", {}),
    ("broker_unavailable", 3, "direct", dict(start_offset=0, timeout=1), "running", {}),      # one 1 s re-observation
]


@pytest.mark.parametrize("status,sub_rc,phase,kw,disk,env", VARIANTS)
def test_c_finalize_variants(tmp_path, status, sub_rc, phase, kw, disk, env):
    def setup(reg):
        if disk == "garbage":
            (reg / "j1.json").write_text("{not json")
        else:
            _disk(disk)(reg)
        (reg / "j1.stall.out").write_text("one\ntwo\nthree\nfour\n")
    old, new = _run_both(tmp_path, _finalize_body(status, sub_rc, phase, **kw), env, setup=setup)
    assert old == new, (old, new)


def test_c_finalize_heartbeat_grace_runs_to_the_wall_clock(tmp_path):
    """Fresh disk heartbeats keep the grace open until the deadline (2 s)."""
    def setup(reg):
        _disk("running")(reg)
        _events(reg, "j1", time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + 30)))
    t0 = time.time()
    old, new = _run_both(tmp_path, _finalize_body("timeout", 2, "direct", start_offset=0, timeout=2, idle=1), {},
                         setup=setup)
    assert old == new, (old, new)
    assert "wall_clock_timeout" in new[3] and time.time() - t0 >= 3


def test_c_finalize_grace_sees_a_late_completion(tmp_path):
    """A record that turns completed during the grace ends it as completed."""
    body = ("( sleep 1; printf '%s' '{\"status\": \"completed\"}' > \"$REGISTRY_DIR/j1.json\" ) &\n"
            + _finalize_body("timeout", 2, "direct", start_offset=0, timeout=30, idle=10))
    old, new = _run_both(tmp_path, body, {}, setup=_disk("running"))
    assert old == new, (old, new)
    assert "FINAL=completed" in new[1] and new[3] == ""


def _gone(pgid: int) -> bool:
    try:
        os.killpg(pgid, 0)
    except ProcessLookupError:
        return True
    except PermissionError:
        return False
    return False


@pytest.mark.parametrize("status,sub_rc,idle", [("broker_unavailable", 3, 0), ("timeout", 2, 10)])
def test_c_finalize_ends_with_its_wrapper(tmp_path, status, sub_rc, idle):
    """Review R1: TERM to the wrapper during either grace ends the grace too -
    the wrapper exits at once, nothing it started outlives it for long, and no
    close or event follows after the deadline (6 s)."""
    results = []
    for which in ("old", "new"):
        work = tmp_path / which
        reg = work / "reg"
        reg.mkdir(parents=True)
        record = json.dumps({"job_id": "j1", "status": "running"})
        (reg / "j1.json").write_text(record)
        log = work / "calls.log"
        log.write_text("")
        shim = _shim(work)
        out = work / "out.txt"
        script = (f'source "{WRAPPER}"\n' + (f'source "{OLD}"\n' if which == "old" else "")
                  + f'REGISTRY_DIR="{reg}"; PY="{shim}"\n'
                  + 'echo READY\n' + _finalize_body(status, sub_rc, "direct", start_offset=0, timeout=6, idle=idle))
        with open(out, "w") as fh:
            proc = subprocess.Popen(["bash", "-c", script], stdout=fh, stderr=subprocess.STDOUT, cwd=str(work),
                                    env=_env(work, CALL_LOG=log, DELEGATE_JOB_PYTHON=shim), start_new_session=True)
        try:
            assert _until_true(lambda: "READY" in out.read_text(), 30), out.read_text()
            time.sleep(2)  # inside the grace
            assert proc.poll() is None, out.read_text()
            t0 = time.monotonic()
            proc.terminate()  # the wrapper only
            proc.wait(timeout=10)
            exit_s = time.monotonic() - t0
            group_gone = _until_true(lambda: _gone(proc.pid), 6)
            time.sleep(6)  # past the deadline
            results.append((exit_s < 2, group_gone, log.read_text(), (reg / "j1.json").read_text() == record))
        finally:
            if not _gone(proc.pid):
                os.killpg(proc.pid, 9)
    assert results[0] == results[1] == (True, True, "", True), results


def _until_true(pred, limit):
    end = time.monotonic() + limit
    while time.monotonic() < end:
        if pred():
            return True
        time.sleep(0.1)
    return pred()


@pytest.mark.parametrize("rc", [0, 1, 2, 3, 4, 5])
def test_c_sub_status_exit_code_and_subscriber_up(tmp_path, rc):
    body = (f'st=$(_map_sub_rc {rc} j1 2>/dev/null || _decide sub-status {rc} --job j1 --registry-dir "$REGISTRY_DIR")\n'
            'echo "status=$st"\n')
    # the old file defines _map_sub_rc; the new wrapper does not
    for disk in (None, "running", "completed", "error", "cancelled"):
        old, new = _run_both(tmp_path / str(disk), body, {}, setup=_disk(disk))
        assert old == new, (disk, old, new)
    for fc in ("1", "0"):
        code = (f'if declare -F _rc_for >/dev/null; then if [ "${{MAM_DELEGATE_FAIL_CLOSED:-1}}" = "0" ]; then echo 0; '
                f'else _rc_for "$(_map_sub_rc {rc} j1)"; fi; else _decide exit-code "$(_decide sub-status {rc} --job j1 '
                f'--registry-dir "$REGISTRY_DIR")"; fi\n')
        old, new = _run_both(tmp_path / f"fc{fc}", code, {"MAM_DELEGATE_FAIL_CLOSED": fc})
        assert old == new, (fc, old, new)
    up = (f'if declare -F _rc_for >/dev/null; then sub_exit={rc}; '
          f'if [ $sub_exit -eq $SUB_RC_COMPLETED ] || [ $sub_exit -eq $SUB_RC_BROKER_UNAVAILABLE ]; then echo up; '
          f'else echo down; fi; else _decide subscriber-up {rc} && echo up || echo down; fi\n')
    old, new = _run_both(tmp_path / "up", up, {})
    assert old == new, (old, new)


@pytest.mark.parametrize("rc", [0, 1, 2, 3, 4, 5])
@pytest.mark.parametrize("job", ["", "j1"])
def test_c_cmd_wait(tmp_path, rc, job):
    body = (f'set +e; cmd_wait {"--job " + job if job else ""} --timeout 5; echo "rc=$?"\n')
    old, new = _run_both(tmp_path, body, {"SUB_RC": rc})
    assert old == new, (old, new)
    assert f"rc={2 if rc == 4 else rc}" in new[1]


@pytest.mark.parametrize("env,session,agent", [
    (None, "herdr:c", "claude"), ("0", "herdr:c", "claude"), (None, "remote:r", "claude"),
    (None, "herdr:c", "human"), (None, "herdr:human", "codex"), ("", "c", "codex")])
@pytest.mark.parametrize("tok,att", [("", ""), ("tok1", "2")])
def test_c_start_stall_watch(tmp_path, env, session, agent, tok, att):
    body = (f'JOB_ID=j1; start_stall_watch "{session}" "{agent}" "{tok}" "{att}" >out.txt\n'
            '[ -z "$STALL_PID" ] || wait "$STALL_PID"\n'
            'sed "s/pid: [0-9]*/pid: N/" out.txt; echo "spawned=${STALL_PID:+yes}"\n')
    old, new = _run_both(tmp_path, body, {"MAM_STALL_WATCH": env, "HERDR_SESSION_NAME": "hs"})
    assert old == new, (old, new)


INT_INPUTS = ["30", "0700", "0900", " 30 ", "0", "-5", "abc", "", "1e3", "9999999999", "000000000030",
              "999999999", "1000000000", " 30 ", "٣٠", "\t7\n", "9" * 5000]


@pytest.mark.parametrize("locale", ["C", "en_US.UTF-8"])
@pytest.mark.parametrize("minimum", [0, 1])
def test_c_mam_int_and_knob(tmp_path, locale, minimum):
    old_fns = subprocess.run(["sed", "-n", "/^mam_int()/,/^}/p;/^mam_int_knob()/,/^}/p", str(OLD)],
                             capture_output=True, text=True, check=True).stdout
    probe = ('for v in "${VALUES[@]}"; do printf "[%s]" "$(mam_int "$v" "$MIN" || echo invalid)"; '
             'K="$v" mam_int_knob K 42 "$MIN"; done; echo; mam_int_knob UNSET_KNOB 9 "$MIN"\n')
    arr = "VALUES=(" + " ".join("$'" + v.encode("unicode_escape").decode().replace("'", "\\'") + "'"
                                for v in INT_INPUTS) + ")\n"
    outs = []
    for src in (old_fns, f'source "{LIB_INT}"\n'):
        res = subprocess.run(["bash", "-c", src + arr + probe], capture_output=True, text=True, timeout=120,
                             env={**os.environ, "MIN": str(minimum), "LC_ALL": locale, "LANG": locale})
        outs.append((res.returncode, res.stdout, res.stderr))
    assert outs[0] == outs[1], outs


def _mtime(path: Path):
    try:
        return path.stat().st_mtime_ns
    except OSError:
        return None


def test_c_helpers_write_nothing_to_an_inherited_workspace(tmp_path, monkeypatch):
    """v4.8.7 review (claude): _env kept an inherited WORKSPACE_ROOT, so every
    helper that sources the wrapper (lib.sh) rewrote <that workspace>/.mam/
    shim/herdr - another workspace's, during the PR #51 review. With a sentinel
    workspace in the environment, nothing appears under it, the checkout's own
    shim is not rewritten either, and the shell sees the test's tmp dir."""
    sentinel = tmp_path / "sentinel-ws"
    sentinel.mkdir()
    monkeypatch.setenv("WORKSPACE_ROOT", str(sentinel))
    checkout_shim = REPO_ROOT / ".mam" / "shim" / "herdr"
    before = _mtime(checkout_shim)
    both = _run_both(tmp_path / "both", 'echo "ws=$WORKSPACE_ROOT"\n', {})
    work = tmp_path / "direct"
    work.mkdir()
    res = subprocess.run(["bash", "-c", f'source "{WRAPPER}"; echo "ws=$WORKSPACE_ROOT"'], capture_output=True,
                         text=True, cwd=str(work), timeout=60, env=_env(work))
    assert sorted(str(p.relative_to(sentinel)) for p in sentinel.rglob("*")) == [], "written into the sentinel"
    assert _mtime(checkout_shim) == before, "the checkout's shim was rewritten"
    for rc, out, err, _ in both:
        assert rc == 0 and "ws=<W>" in out, (rc, out, err)
    assert res.returncode == 0 and f"ws={work}" in res.stdout, (res.stdout, res.stderr)
    assert (work / ".mam" / "shim" / "herdr").is_file(), "lib.sh wrote its shim into the test's own root"


# ---------------------------------------------------------------------------
# S - static guard
# ---------------------------------------------------------------------------

def test_s_the_wrapper_carries_no_moved_decision():
    text = "\n".join(l for l in WRAPPER.read_text().splitlines() if not l.lstrip().startswith("#"))
    for name in ("_map_sub_rc", "_rc_for", "_disk_status", "_recent_disk_heartbeat", "FINALIZE_CLOCK"):
        assert name not in text, name
    for knob in ("MAM_DELEGATE_FAIL_CLOSED", "MAM_STALL_WATCH"):
        assert "${" + knob not in text, knob
    # subscriber codes are compared nowhere in bash any more
    assert not re.search(r"-(eq|ne) +\"?\$SUB_RC_", text)
    assert not re.search(r"= +\"?\$SUB_RC_", text)
    for label in ("wall_clock_timeout", "deadline_timeout", "agent_blocked", "session_busy"):
        assert label not in text, label
    assert "BASH_REMATCH" not in LIB_INT.read_text()


def test_u_finalize_end_leaves_without_side_effects_once_its_wrapper_is_gone(tmp_path):
    """Review R1, in process: a dead --watch-pid stops the wait and every side effect."""
    import argparse
    dead = subprocess.Popen(["true"])
    dead.wait()
    (tmp_path / "j.json").write_text(json.dumps({"status": "running"}))
    log = tmp_path / "calls.log"
    a = argparse.Namespace(job="j", registry_dir=str(tmp_path), status="broker_unavailable", start_time=int(time.time()),
                           timeout=30, idle_timeout=0, sub_rc=3, phase="direct", type="direct", final="broker_unavailable",
                           evidence_until=0, dispatch_token="t", preflight_reason="", python=str(_shim(tmp_path)),
                           scripts_dir=str(tmp_path / "scripts"), watch_pid=dead.pid)
    os.environ["CALL_LOG"] = str(log)
    try:
        t0 = time.monotonic()
        with pytest.raises(decide.OwnerGone):
            decide.finalize_end(a)
        assert time.monotonic() - t0 < 2
        a.status, a.start_time = "error", 0  # no wait at all: the close is still withheld
        with pytest.raises(decide.OwnerGone):
            decide.finalize_end(a)
    finally:
        del os.environ["CALL_LOG"]
    assert not log.exists() or log.read_text() == ""
    a.watch_pid = os.getpid()  # alive: the close goes ahead
    os.environ["CALL_LOG"] = str(log)
    try:
        assert decide.finalize_end(a) == "broker_unavailable"
    finally:
        del os.environ["CALL_LOG"]
    assert "registry.py" in log.read_text()


def test_u_an_exited_unreaped_owner_is_not_alive():
    """Review R1 round 2: a zombie answers kill(0); it is not a live wrapper."""
    z = subprocess.Popen(["true"])
    try:
        assert _until_true(lambda: subprocess.run(["ps", "-o", "stat=", "-p", str(z.pid)], capture_output=True,
                                                  text=True).stdout.strip().startswith("Z"), 10)
        os.kill(z.pid, 0)  # still answers ...
        assert decide._alive(z.pid) is False  # ... but is gone
    finally:
        z.wait()
    assert decide._alive(z.pid) is False
    assert decide._alive(os.getpid()) is True


def _terminate_unreaped(script: str, cwd: Path, env: dict, ready: Path):
    """TERM to the wrapper, then communicate() with no wait() first - the
    captured-output caller's pattern, where the wrapper stays a zombie while a
    survivor holds its pipes (review R1 round 2). Returns (rc, seconds)."""
    proc = subprocess.Popen(["bash", "-c", script], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                            cwd=str(cwd), env=env, start_new_session=True)
    try:
        assert _until_true(lambda: ready.exists(), 30), "the wrapper never reached the grace"
        time.sleep(2)  # inside the grace
        t0 = time.monotonic()
        proc.terminate()
        proc.communicate(timeout=60)
        return proc.returncode, time.monotonic() - t0
    finally:
        if not _gone(proc.pid):
            os.killpg(proc.pid, 9)


@pytest.mark.parametrize("status,sub_rc,idle", [("broker_unavailable", 3, 0), ("timeout", 2, 10)])
def test_c_finalize_ends_with_an_unreaped_wrapper(tmp_path, status, sub_rc, idle):
    results = []
    for which in ("old", "new"):
        work = tmp_path / which
        reg = work / "reg"
        reg.mkdir(parents=True)
        record = json.dumps({"job_id": "j1", "status": "running"})
        (reg / "j1.json").write_text(record)
        log = work / "calls.log"
        log.write_text("")
        shim = _shim(work)
        script = (f'source "{WRAPPER}"\n' + (f'source "{OLD}"\n' if which == "old" else "")
                  + f'REGISTRY_DIR="{reg}"; PY="{shim}"\ntouch "{work}/ready"\n'
                  + _finalize_body(status, sub_rc, "direct", start_offset=0, timeout=6, idle=idle))
        rc, took = _terminate_unreaped(script, work, _env(work, CALL_LOG=log, DELEGATE_JOB_PYTHON=shim), work / "ready")
        time.sleep(6)  # past the deadline
        results.append((rc, took < 4.5, log.read_text(), (reg / "j1.json").read_text() == record))
    assert results[0] == results[1] == (-15, True, "", True), results


def test_c_an_unreaped_wrapper_leaves_the_real_record_alone(tmp_path):
    """The same with the real registry CLI and a real dispatch token (loop
    type: no publication): the record is still pending after the deadline."""
    registry = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
    states = []
    for which in ("old", "new"):
        work = tmp_path / which
        reg = work / "reg"
        reg.mkdir(parents=True)
        env = _env(work, DELEGATE_JOB_PYTHON=sys.executable)
        res = subprocess.run([sys.executable, str(registry), "--registry-dir", str(reg), "register", "--prompt", "p",
                              "--agent", "claude", "--agent-session", "herdr:x", "--job-type", "loop"],
                             capture_output=True, text=True, env=env, timeout=60)
        assert res.returncode == 0, res.stderr
        jid = res.stdout.strip().splitlines()[-1]
        tok = json.loads((reg / f"{jid}.json").read_text())["dispatch_token"]
        script = (f'source "{WRAPPER}"\n' + (f'source "{OLD}"\n' if which == "old" else "")
                  + f'REGISTRY_DIR="{reg}"; PY="{sys.executable}"; TYPE=loop; AGENT=human; AGENT_SESSION=herdr:x\n'
                  + f'TIMEOUT=6; IDLE_TIMEOUT=10; START_TIME=$(date +%s); _disp_tok="{tok}"\n'
                  + f'touch "{work}/ready"; finalize_unfinished {jid} timeout 2 worker\n')
        rc, took = _terminate_unreaped(script, work, env, work / "ready")
        time.sleep(6)
        states.append((rc, took < 4.5, json.loads((reg / f"{jid}.json").read_text())["status"]))
    assert states[0] == states[1] == (-15, True, "pending"), states


# --- review R1 (codex): a wall clock stepped back does not stretch a wait ----

class _SteppedClock:
    """A `time` stand-in: wall and monotonic both advance on sleep(), and the
    wall clock is stepped back once, during the first sleep."""

    def __init__(self, step_back: float):
        self.wall, self.mono, self.step_back, self.slept = 1000.0, 50.0, step_back, 0.0

    def time(self):
        return self.wall

    def monotonic(self):
        return self.mono

    def sleep(self, dt):
        self.wall += dt
        self.mono += dt
        self.slept += dt
        if self.step_back:
            self.wall -= self.step_back
            self.step_back = 0


@pytest.mark.parametrize("seconds", [1, 3])
def test_u_a_wait_is_not_stretched_by_a_wall_clock_step_back(monkeypatch, seconds):
    """The broker re-observation (up to 3 s) and the disk-heartbeat grace (1 s)
    re-read the record after `seconds`, as bash's `sleep N` did, even when the
    wall clock steps back 10 s meanwhile (on the old wall-clock arithmetic the
    wait became seconds + 10)."""
    import argparse
    clock = _SteppedClock(step_back=10)
    monkeypatch.setattr(decide, "time", clock)
    decide._sleep(argparse.Namespace(watch_pid=None), seconds)
    assert seconds <= clock.slept <= seconds + 0.5, clock.slept


# ---------------------------------------------------------------------------
# F - a decision wrapper.py cannot make ends the wrapper with 1 (N2)
# ---------------------------------------------------------------------------

def _failing_py(tmp_path: Path, decision: str, mode: str) -> Path:
    """The wrapper's $PY: the real interpreter, except that `decision` fails -
    with a traceback and rc 1 (a broken module), with rc 127 (an interpreter
    that cannot run), or with rc 0 and no answer.
    job_subscriber.py is a stand-in exiting $SUB_RC when SUB_STANDIN=1."""
    shim = tmp_path / "py-fail"
    fail = {"error": 'echo "ModuleNotFoundError: No module named lib_py.delegation.wrapper" >&2; exit 1',
            "gone": 'echo "py: not found" >&2; exit 127', "empty": "exit 0"}[mode]
    shim.write_text(
        "#!/bin/sh\n"
        'if [ "${SUB_STANDIN:-0}" = 1 ]; then case "$1" in */scripts/job_subscriber.py) exit "${SUB_RC:-0}";; esac; fi\n'
        f'case "$*" in *"lib_py.delegation.wrapper {decision} "*|*"lib_py.delegation.wrapper {decision}") {fail};; esac\n'
        f'exec "{sys.executable}" "$@"\n')
    shim.chmod(0o755)
    return shim


def _human_submit(tmp_path: Path, py: str, typ: str = "direct", **extra) -> subprocess.CompletedProcess:
    """The real wrapper, one human job, an unreachable broker: the subscriber
    ends with rc 3, the job is closed at its 3 s wall clock and the wrapper
    exits through sub-status, finalize-begin/-end and exit-code (rc 3)."""
    env = {k: v for k, v in os.environ.items()
           if not k.startswith(("HERDR_", "MAM_", "SKS_", "MQTT_")) and k not in ("PYTHONPATH", "WORKSPACE_ROOT")}
    env.update(HOME=str(tmp_path), WORKSPACE_ROOT=str(tmp_path), DELEGATE_JOB_PYTHON=py,
               AGENT_PYTHON_BIN=sys.executable, PYTHONPATH=str(SKILLS), DELEGATE_JOB_LOGS_DIR=str(tmp_path / "logs"),
               MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"), MQTT_BROKER="127.0.0.1", MQTT_PORT="9", MQTT_TLS="0")
    env.update({k: str(v) for k, v in extra.items()})
    loop = ["--reviewer", "human", "--reviewer-session", "herdr:human"] if typ != "direct" else []
    return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--type", typ, *loop,
                           "--registry-dir", str(tmp_path / "reg"),
                           "--workdir", str(tmp_path), "--prompt", "n2", "--timeout", "3", "--idle-timeout", "2"],
                          capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=120)


def test_f_the_normal_path_is_unchanged(tmp_path):
    res = _human_submit(tmp_path, sys.executable)
    assert res.returncode == 3, res.stdout[-800:] + res.stderr[-800:]
    assert res.stdout.strip().splitlines()[-1] == str(tmp_path / "logs" / next((tmp_path / "reg").glob("*.json")).stem)
    assert "could not decide" not in res.stderr


@pytest.mark.parametrize("decision,mode", [
    ("exit-code", "error"), ("exit-code", "empty"),
    ("sub-status", "error"), ("sub-status", "gone"), ("sub-status", "empty"),
    ("finalize-begin", "error"), ("finalize-begin", "empty"),
    ("finalize-end", "error"), ("finalize-end", "gone"),  # it may answer nothing (its wrapper gone): rc only
])
def test_f_a_failed_decision_ends_the_submit_with_1(tmp_path, decision, mode):
    res = _human_submit(tmp_path, str(_failing_py(tmp_path, decision, mode)))
    assert res.returncode == 1, (res.returncode, res.stderr[-800:])
    assert f"ERROR: the wrapper could not decide '{decision}'" in res.stderr, res.stderr[-800:]


@pytest.mark.parametrize("mode", ["error", "gone", "empty"])
@pytest.mark.parametrize("decision,typ", [("subscriber-up", "direct"), ("subscriber-up", "loop"),
                                          ("stall-watch", "direct")])
def test_f_a_failed_yes_no_decision_ends_the_submit_with_1(tmp_path, decision, typ, mode):
    """Review R1 (codex): a yes/no decision that fails must not pass for "no".
    The subscriber stand-in ends with 0 before SUBSCRIBED: read as "not up",
    the error path mapped rc 0 to completed and the submit ended with 0."""
    res = _human_submit(tmp_path, str(_failing_py(tmp_path, decision, mode)), typ, SUB_STANDIN=1, SUB_RC=0)
    assert res.returncode == 1, (res.returncode, res.stderr[-800:])
    assert f"ERROR: the wrapper could not decide '{decision}'" in res.stderr, res.stderr[-800:]


@pytest.mark.parametrize("typ", ["direct", "loop"])
@pytest.mark.parametrize("sub_rc,up", [(0, True), (1, False)])
def test_f_yes_and_no_keep_their_branches(tmp_path, typ, sub_rc, up):
    """The same runs without a failure: "yes" goes on, "no" takes the error path."""
    res = _human_submit(tmp_path, str(_failing_py(tmp_path, "none", "error")), typ, SUB_STANDIN=1, SUB_RC=sub_rc)
    assert "could not decide" not in res.stderr, res.stderr[-800:]
    assert ("subscriber died early" in res.stderr) is (not up), res.stderr[-800:]
    if typ == "direct":
        assert res.returncode == (0 if up else 1), (res.returncode, res.stderr[-800:])


@pytest.mark.parametrize("rc,up", [(0, True), (3, True), (1, False), (2, False), (4, False)])
def test_u_yes_no_print(capsys, rc, up):
    assert decide.main(["subscriber-up", str(rc), "--print"]) == 0
    assert capsys.readouterr().out == ("yes\n" if up else "no\n")
    assert decide.main(["subscriber-up", str(rc)]) == (0 if up else 1)  # the rc form is unchanged
    assert decide.main(["stall-watch", "herdr:c", "claude", "--print"]) == 0
    assert capsys.readouterr().out == "yes\n"
    assert decide.main(["stall-watch", "remote:r", "claude", "--print"]) == 0
    assert capsys.readouterr().out == "no\n"


@pytest.mark.parametrize("mode", ["error", "empty"])
def test_f_a_failed_wait_rc_ends_wait_with_1(tmp_path, mode):
    py = _failing_py(tmp_path, "wait-rc", mode)
    script = f'source "{WRAPPER}"; PY="{py}"; set +e; (cmd_wait --job j1 --timeout 5); echo "rc=$?"'
    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True, cwd=str(tmp_path), timeout=60,
                         env=_env(tmp_path, SUB_STANDIN=1, SUB_RC=4, DELEGATE_JOB_PYTHON=py, WORKSPACE_ROOT=tmp_path))
    assert "rc=1" in res.stdout, (res.stdout, res.stderr[-800:])
    assert "ERROR: the wrapper could not decide 'wait-rc'" in res.stderr, res.stderr[-800:]


def test_s_no_printed_decision_is_read_raw():
    """Every printed answer goes through _decided/_exit_for/_decided_yes (the
    yes/no ones with --print), and finalize-end is checked by its rc."""
    text = "\n".join(l for l in WRAPPER.read_text().splitlines() if not l.lstrip().startswith("#"))
    raw = [l.strip() for l in text.splitlines() if "$(_decide" in l]
    assert raw == ['__out="$(_decide "$@")" || __rc=$?',
                   '{ out="$(_decide finalize-end --job "$job_id" --registry-dir "$REGISTRY_DIR" --status "$job_status" \\'], raw
    assert 'exit "$(' not in text
    assert not re.search(r"\b_decide (subscriber-up|stall-watch)\b", text), "a yes/no decision read by its rc"
    assert text.count("_decided_yes subscriber-up") == 2 and text.count("_decided_yes stall-watch") == 1