GodopuandClaude Opus 5.5
4c86107846
codex, PR #51 round 1. wrapper._sleep computed the span and the remainder with time.time(). The broker re-observation (up to 3 s) and the disk-heartbeat grace (1 s) use it, so a wall clock stepped back during the wait stretched it by the step: on a mocked clock stepped back 10 s, an already completed job was seen after 11 s / 13 s instead of 1 s / 3 s. The old bash ran `sleep 1` / `sleep "$left"`, which do not follow the wall clock. _sleep now measures on time.monotonic(). The deadline (START_TIME + TIMEOUT), the evidence window and event timestamps stay on the wall clock (_now), as in the bash. test_u_a_wait_is_not_stretched_by_a_wall_clock_step_back (1 s, 3 s) swaps the module's time for a clock stepped back 10 s during the first sleep; on the previous code it measures 11.0 / 13.0 and fails. The comparison test and the finalize-related behaviour tests (issue33, issue18, issue14) on macOS / Python 3.9: 321 passed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
631 lines
30 KiB
Python
631 lines
30 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
|
||
|
||
|
||
# --- 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
|