GodopuandClaude Opus 5.5
518bedac5d
kill(pid, 0) succeeds for a zombie. A caller that captures the wrapper's
output and calls communicate() after TERM leaves the wrapper unreaped while
the finalizer holds its pipes, so the owner check passed until the deadline
and the job was still closed (codex, round 2; reproduced with the real
registry). _alive now asks `ps -o stat=` (POSIX, no /proc) once kill(0)
succeeds and treats state Z as gone, as tests/conftest.py::_pid_alive does;
no ps answer keeps kill(0)'s verdict. The owner is looked at every 0.5 s.
Tests: TERM + communicate() with no wait() for both graces (stand-ins) and
for the timeout grace with the real registry, old vs new; a zombie is not
alive. All four fail on c5cebe6.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
594 lines
29 KiB
Python
594 lines
29 KiB
Python
"""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:
|
||
env = {k: v for k, v in os.environ.items()
|
||
if not k.startswith(("HERDR_", "MAM_", "SKS_")) and k not in ("PYTHONPATH",)}
|
||
env.update(MAM_DELEGATE_JOB_NO_MAIN="1", MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"), HOME=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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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
|