GodopuandClaude Opus 5.5
8b046390be
MAM_WAIT_GRACE_S silently fell back to 300 on anything but bare digits
(" 30 " too), MAM_CHILD_WAIT_S went through float() (abc raised ValueError,
0.5 and 1e3 were taken) and MAM_JOB_TREE_MAX through int() (abc and an
empty value raised). All three now go through runner._int_knob, which is
_positive_int's rule (PR #43 review grok): decimal, blanks around allowed,
an invalid value warns once and the default is used. Minimums:
MAM_WAIT_GRACE_S >= 0 (wait exactly the budget), MAM_CHILD_WAIT_S >= 0
(look once; test_issue34 uses 0), MAM_JOB_TREE_MAX >= 1 (a cap of 0 calls
every tree "over"; not waiting is MAM_CHILD_WAIT_S=0).
The loop SKILL.md now names MAM_WAIT_GRACE_S in the budget row and the
sub-job paragraph (PR #41 M6), with the knobs' integer rules.
Tests: RUNNER_INT_KNOBS in test_issue26_ints.py holds the minimums; the
real wait_for_descendants is driven per knob, and a static check ties the
call-site minimums to the table (15 of the new cases fail at 0051fd6; the
14 that pass are valid values whose result did not change).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe
459 lines
25 KiB
Python
459 lines
25 KiB
Python
"""Issue #26 (B): integer knobs that reach bash arithmetic are read as decimal.
|
|
|
|
bash reads 0700 as octal 448 and fails on 0900; the runner reads 700. Both now
|
|
go through one rule (lib.sh mam_int / runner._positive_int) with a per-knob
|
|
minimum, and INT_KNOBS below is the one place that minimum is written.
|
|
|
|
- B1: one input table through the real wrapper and the real runner.
|
|
- B2: flags are normalised or refused.
|
|
- B3: the derived idle timeout uses decimal values.
|
|
- B4: the ack waits read 0900 as 900 and keep 0 as "check once".
|
|
- B5: static guard - no env default reaches arithmetic directly, and every
|
|
normalised knob uses the minimum in INT_KNOBS.
|
|
"""
|
|
|
|
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_SH = SKILLS / "lib.sh"
|
|
|
|
# knob -> (minimum, default). The minimum is 0 where 0 already means "do not
|
|
# wait / check once" (SKS_BUSY_TIMEOUT=0 is test_issue6_turn_handshake T2-12's
|
|
# contract); 1 where 0 would close a job at once.
|
|
INT_KNOBS = {
|
|
"MAM_JOB_TIMEOUT_S": (1, "3600"),
|
|
"MAM_HEARTBEAT_SEC": (1, "30"),
|
|
"MAM_HEARTBEAT_MISSES": (1, "4"),
|
|
"MAM_ACK_TIMEOUT_S": (1, "180"),
|
|
"MAM_INJECT_ACK_GRACE_S": (0, "$ACK_TIMEOUT"),
|
|
"SKS_DIALOG_TIMEOUT": (0, "30"),
|
|
"SKS_BUSY_TIMEOUT": (0, "120"),
|
|
"SKS_DELIVERY_GRACE_S": (0, "15"),
|
|
# _pane_quiescent's `for ((i = 0; i < tries; i++))` (review R2); 0 keeps "headless" (rc 2)
|
|
"SKS_QUIESCENT_TRIES": (0, "20"),
|
|
# delegate_wait_started's ack_s (MAM_ONBOARD_ACK_S / DELEGATE_JOB_ACK_S via create_session.sh)
|
|
"delegate_wait_started:ack_s": (0, "180"),
|
|
# wrapper flags
|
|
"--timeout": (1, None),
|
|
"--idle-timeout": (1, None),
|
|
"--heartbeat": (1, None),
|
|
"--heartbeat-misses": (1, None),
|
|
"--ack-timeout": (1, None),
|
|
}
|
|
|
|
# The loop runner's own integer knobs (follow-up F4a): read by runner._int_knob
|
|
# (_positive_int's rule) with these minimums; an invalid value warns once and
|
|
# the default is used - never an exception. knob -> (minimum, default)
|
|
RUNNER_INT_KNOBS = {
|
|
"MAM_WAIT_GRACE_S": (0, 300), # 0: wait exactly the job budget
|
|
"MAM_CHILD_WAIT_S": (0, None), # 0: look at the sub-jobs once; default: budget + grace
|
|
"MAM_JOB_TREE_MAX": (1, 50), # 0 would refuse every tree as "over"; not waiting is MAM_CHILD_WAIT_S=0
|
|
}
|
|
|
|
# The last two are longer than Python's int() digit limit (4300): the rule must
|
|
# bound the significant digits before converting (review R3).
|
|
INPUTS = ["30", "0700", "0900", " 30 ", "0", "-5", "abc", "", "1e3", "9999999999", "000000000030",
|
|
"9" * 5000, "0" * 5000 + "30"]
|
|
|
|
# Integer-looking env defaults in the shell scripts that do NOT reach bash
|
|
# arithmetic, each with where the value goes. B5 fails on any `${MAM_/SKS_/
|
|
# DELEGATE_JOB_/RECONCILE_…:-<digit>…}` that is neither here nor normalised
|
|
# through INT_KNOBS, so a new knob has to be classified to land.
|
|
NOT_ARITHMETIC = {
|
|
"MAM_DELEGATE_FAIL_CLOSED": "string compare", "MAM_STALL_WATCH": "string compare",
|
|
"MAM_ALLOW_UNSIGNED_DISPATCH": "string compare", "MAM_DELEGATE_JOB_NO_MAIN": "string compare",
|
|
"MAM_NO_AUTOSTART": "string compare", "MAM_SHIM_DEBUG": "string compare",
|
|
"MAM_NO_SUBSCRIBER_DAEMON": "string compare", "MAM_DELEGATION_PY": "string compare",
|
|
"MAM_AUTO_TRUST": "string compare", "MAM_KILL_ON_READY_TIMEOUT": "string compare",
|
|
"MAM_LOOP_PY": "string compare", "MAM_LOOP_FREEZE": "string compare", "MAM_LOOP_NO_FREEZE": "string compare",
|
|
"MAM_LOOP_FREEZE_OWNED": "string compare", "MAM_AUTODETECT_FORCE_FAIL": "string compare",
|
|
"MAM_WS_LABEL_EXPLICIT": "passed to python", "SKS_IDLE_HANDSHAKE": "string compare",
|
|
"SKS_DIALOG_ESCAPE": "string compare", "SKS_PROMPT_WAIT": "string compare",
|
|
"MAM_MIN_PANE_COLS": "argv of python -m lib_py.layout", "MAM_MIN_PANE_ROWS": "argv of python -m lib_py.layout",
|
|
"MAM_MAX_PANE_COLS": "argv of python -m lib_py.layout", "MAM_MAX_PANE_ROWS": "argv of python -m lib_py.layout",
|
|
"MAM_BUFFER_GC_MINUTES": "find -mmin", "MAM_DIALOG_TAIL_LINES": "tail -n",
|
|
"SKS_IDLE_TIMEOUT_MS": "argv of the python idle watcher", "MAM_DIALOG_RESEND_S": "python int()",
|
|
"MAM_READY_EMPTY_GIVEUP": "[ -ge ] (decimal)", "SKS_EMPTY_GIVEUP": "[ -ge ] (decimal)",
|
|
"SKS_IDLE_RETRIES": "[ -le ] (decimal)", "SKS_IDLE_RETRY_BACKOFF_S": "sleep",
|
|
"SKS_QUIESCENT_INTERVAL": "sleep", "MAM_LOCK_STEAL_RETRIES": "[ -le ] (decimal)",
|
|
"MAM_DIFF_MAX_BYTES": "[ -gt ] (decimal)", "MAM_DIFF_MAX_LINES": "[ -gt ] (decimal)",
|
|
"RECONCILE_POLL_INTERVAL": "sleep",
|
|
"DELEGATE_JOB_ACK_S": "create_session.sh -> delegate_wait_started, which runs mam_int on it",
|
|
}
|
|
|
|
|
|
def _expected(value: str, minimum: int):
|
|
s = value.strip(" \t\n\r\v\f") # ASCII blanks only, in every locale (review R2)
|
|
if not s.isascii() or not s.isdigit():
|
|
return None
|
|
s = s.lstrip("0") or "0"
|
|
if len(s) > 9:
|
|
return None
|
|
n = int(s)
|
|
return n if n >= minimum else None
|
|
|
|
|
|
def _id(value: str) -> str:
|
|
return repr(value) if len(value) <= 12 else f"{value[:3]}...x{len(value)}"
|
|
|
|
|
|
def _base_env(home: Path, extra=None) -> dict:
|
|
env = {k: v for k, v in os.environ.items() if k not in INT_KNOBS and not k.startswith("MAM_JOB")}
|
|
env.update({"HOME": str(home), "WORKSPACE_ROOT": str(home), "MAM_ENV_FILE": str(home / ".mam.env.absent"),
|
|
"DELEGATE_JOB_PYTHON": sys.executable, "AGENT_PYTHON_BIN": sys.executable,
|
|
"PYTHONPATH": str(SKILLS), "DELEGATE_JOB_LOGS_DIR": str(home / "logs")})
|
|
if extra:
|
|
env.update(extra)
|
|
return env
|
|
|
|
|
|
def _dry_submit(home: Path, extra_env=None, *flags) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--dry-run",
|
|
"--registry-dir", str(home / "reg"), "--prompt", "p", *flags],
|
|
capture_output=True, text=True, env=_base_env(home, extra_env), cwd=str(home), timeout=120)
|
|
|
|
|
|
def _record(home: Path, res: subprocess.CompletedProcess) -> dict:
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
|
|
return json.loads((home / "reg" / f"{jid}.json").read_text())
|
|
|
|
|
|
def _runner(home: Path, extra_env=None, *argv) -> subprocess.CompletedProcess:
|
|
"""The real runner in its own process: parse_args, then the budget and the
|
|
contract flags it would hand the wrapper."""
|
|
code = ("import json, sys; from lib_py.loop import runner as lr; "
|
|
"r = lr.LoopRunner(['--creator', 'c', '--task', 't'] + sys.argv[1:]); r.parse_args(); "
|
|
"print(json.dumps({'budget': r.job_budget, 'contract': r.contract_args()}))")
|
|
return subprocess.run([sys.executable, "-c", code, *argv], capture_output=True, text=True,
|
|
env=_base_env(home, extra_env), cwd=str(home), timeout=60)
|
|
|
|
|
|
def _runner_wait(home: Path, extra_env=None, kids: int = 0) -> tuple:
|
|
"""The real runner waiting on a sub-job tree of `kids` open jobs (budget 1 s):
|
|
({grace, ok}, everything it printed)."""
|
|
code = ("import json, sys; from lib_py.loop import runner as lr; "
|
|
"r = lr.LoopRunner(['--creator', 'c', '--task', 't', '--job-timeout', '1']); r.parse_args(); "
|
|
"lr.job_registry.descendants_of = lambda root, d: [{'job_id': 'k%d' % i, 'status': 'running'} "
|
|
"for i in range(int(sys.argv[1]))]; "
|
|
"ok = r.wait_for_descendants('root', interval=0.1); "
|
|
"print(json.dumps({'grace': r.job_wait_s - r.job_budget, 'ok': ok}))")
|
|
res = subprocess.run([sys.executable, "-c", code, str(kids)], capture_output=True, text=True,
|
|
env=_base_env(home, {k: v for k, v in (extra_env or {}).items()}), cwd=str(home), timeout=60)
|
|
assert res.returncode == 0 and "Traceback" not in res.stderr, res.stderr[-800:]
|
|
return json.loads(res.stdout.strip().splitlines()[-1]), res.stdout + res.stderr
|
|
|
|
|
|
def _bash(snippet: str, home: Path, extra_env=None) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["bash", "-c", f'source "{LIB_SH}" >/dev/null 2>&1; {snippet}'], capture_output=True,
|
|
text=True, env=_base_env(home, extra_env), cwd=str(home), timeout=60)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B1 - one table, the real wrapper and the real runner agree
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("value", INPUTS, ids=_id)
|
|
def test_b1_wrapper_and_runner_give_the_same_budget(tmp_path, value):
|
|
minimum, default = INT_KNOBS["MAM_JOB_TIMEOUT_S"]
|
|
want = _expected(value, minimum)
|
|
budget = want if want is not None else int(default)
|
|
warns = want is None and value != ""
|
|
|
|
wres = _dry_submit(tmp_path, {"MAM_JOB_TIMEOUT_S": value})
|
|
assert _record(tmp_path, wres)["timeout_sec"] == budget
|
|
assert ("MAM_JOB_TIMEOUT_S=" in wres.stderr and "ignored" in wres.stderr) is warns, wres.stderr
|
|
# The record goes through registry.py's int(); the wrapper's own deadlines
|
|
# (START_TIME + TIMEOUT) are bash arithmetic on TIMEOUT - check that value.
|
|
ares = subprocess.run(["bash", "-c", f'source "{WRAPPER}"; parse_opts; echo "arith=$(( 0 + TIMEOUT ))"'],
|
|
capture_output=True, text=True, cwd=str(tmp_path), timeout=60,
|
|
env=_base_env(tmp_path, {"MAM_JOB_TIMEOUT_S": value, "MAM_DELEGATE_JOB_NO_MAIN": "1"}))
|
|
assert f"arith={budget}" in ares.stdout, (ares.stdout, ares.stderr)
|
|
|
|
rres = _runner(tmp_path, {"MAM_JOB_TIMEOUT_S": value})
|
|
assert rres.returncode == 0 and "Traceback" not in rres.stderr, rres.stderr[-500:]
|
|
got = json.loads(rres.stdout.strip().splitlines()[-1])
|
|
assert got["budget"] == budget
|
|
assert got["contract"][:2] == ["--timeout", str(budget)]
|
|
assert ("MAM_JOB_TIMEOUT_S=" in rres.stderr and "ignored" in rres.stderr) is warns, rres.stderr
|
|
|
|
|
|
@pytest.mark.parametrize("minimum", [0, 1])
|
|
def test_b1_mam_int_and_positive_int_are_one_rule(tmp_path, minimum):
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
from lib_py.loop.runner import _positive_int
|
|
|
|
script = "; ".join(f'printf "%s\\n" "$(mam_int {json.dumps(v)} {minimum} || echo invalid)"' for v in INPUTS)
|
|
res = _bash(script, tmp_path)
|
|
bash = [None if l == "invalid" else int(l) for l in res.stdout.splitlines()]
|
|
want = [_expected(v, minimum) for v in INPUTS]
|
|
assert bash == want, (res.stdout, res.stderr)
|
|
assert [_positive_int(v, minimum) for v in INPUTS] == want
|
|
assert want[INPUTS.index("0")] == (0 if minimum == 0 else None)
|
|
assert want[INPUTS.index("0700")] == 700
|
|
|
|
|
|
# Review R2: bash's [[:space:]] and [0-9] follow the locale; the rule is ASCII.
|
|
UNICODE_INPUTS = ["\u200330\u2003", "\u00a030", "30\u3000", "\u0663\u0660", "\uff13\uff10"]
|
|
|
|
|
|
@pytest.mark.parametrize("locale", ["C", "en_US.UTF-8"])
|
|
@pytest.mark.parametrize("value", UNICODE_INPUTS, ids=lambda v: v.encode("unicode_escape").decode())
|
|
def test_b1_unicode_blanks_and_digits_are_invalid_in_every_locale(tmp_path, locale, value):
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
from lib_py.loop.runner import _positive_int
|
|
|
|
loc = {"LC_ALL": locale, "LANG": locale}
|
|
res = _bash('mam_int "$PROBE" 1 || echo invalid', tmp_path, {**loc, "PROBE": value})
|
|
assert res.stdout.strip() == "invalid", (res.stdout, res.stderr)
|
|
assert _positive_int(value) is None and _expected(value, 1) is None
|
|
rec = _record(tmp_path, _dry_submit(tmp_path, {**loc, "MAM_JOB_TIMEOUT_S": value}))
|
|
rres = _runner(tmp_path, {**loc, "MAM_JOB_TIMEOUT_S": value})
|
|
assert rres.returncode == 0, rres.stderr[-500:]
|
|
got = json.loads(rres.stdout.strip().splitlines()[-1])
|
|
assert rec["timeout_sec"] == got["budget"] == 3600, (rec["timeout_sec"], got)
|
|
|
|
|
|
@pytest.mark.parametrize("knob", ["SKS_DIALOG_TIMEOUT", "SKS_BUSY_TIMEOUT", "SKS_DELIVERY_GRACE_S"])
|
|
def test_b1_sks_knobs_keep_zero_and_read_decimal(tmp_path, knob):
|
|
minimum, default = INT_KNOBS[knob]
|
|
for value, want, warns in (("0", "0", False), ("0900", "900", False), ("", default, False),
|
|
("abc", default, True), ("-1", default, True)):
|
|
res = _bash(f"mam_int_knob {knob} {default} {minimum}", tmp_path, {knob: value})
|
|
assert res.stdout.strip() == want, (value, res.stdout, res.stderr)
|
|
assert ("ignored" in res.stderr) is warns, (value, res.stderr)
|
|
|
|
|
|
@pytest.mark.parametrize("value", INPUTS, ids=_id)
|
|
def test_b1_runner_wait_grace_is_decimal_or_the_default(tmp_path, value):
|
|
minimum, default = RUNNER_INT_KNOBS["MAM_WAIT_GRACE_S"]
|
|
want = _expected(value, minimum)
|
|
got, out = _runner_wait(tmp_path, {"MAM_WAIT_GRACE_S": value})
|
|
assert got["grace"] == (default if want is None else want), out[-500:]
|
|
assert ("WARNING: MAM_WAIT_GRACE_S=" in out and "ignored" in out) is (want is None and value != ""), out[-500:]
|
|
|
|
|
|
@pytest.mark.parametrize("value,waited,warns", [("0", "0s (MAM_CHILD_WAIT_S)", False),
|
|
(" 0 ", "0s (MAM_CHILD_WAIT_S)", False),
|
|
("01", "1s (MAM_CHILD_WAIT_S)", False),
|
|
("abc", "1s (MAM_CHILD_WAIT_S default", True),
|
|
("-5", "1s (MAM_CHILD_WAIT_S default", True),
|
|
("0.5", "1s (MAM_CHILD_WAIT_S default", True),
|
|
("9999999999", "1s (MAM_CHILD_WAIT_S default", True),
|
|
("", "1s (MAM_CHILD_WAIT_S default", False)])
|
|
def test_b1_runner_child_wait_is_decimal_or_the_default(tmp_path, value, waited, warns):
|
|
"""MAM_CHILD_WAIT_S=abc raised ValueError; the default is budget + grace (1 + 0 here)."""
|
|
got, out = _runner_wait(tmp_path, {"MAM_CHILD_WAIT_S": value, "MAM_WAIT_GRACE_S": "0"}, kids=1)
|
|
assert got["ok"] is False and f"ended after {waited}" in out, out[-500:]
|
|
assert ("WARNING: MAM_CHILD_WAIT_S=" in out and "ignored" in out) is warns, out[-500:]
|
|
|
|
|
|
@pytest.mark.parametrize("value,over,warns", [("2", True, False), ("03", False, False), (" 2 ", True, False),
|
|
("0", False, True), ("abc", False, True), ("-1", False, True),
|
|
("", False, False)])
|
|
def test_b1_runner_job_tree_max_is_decimal_or_the_default(tmp_path, value, over, warns):
|
|
"""Three open sub-jobs against the cap; an invalid cap is 50 with a warning."""
|
|
got, out = _runner_wait(tmp_path, {"MAM_JOB_TREE_MAX": value, "MAM_CHILD_WAIT_S": "0"}, kids=3)
|
|
assert got["ok"] is False
|
|
assert ("over MAM_JOB_TREE_MAX=" in out) is over, out[-500:]
|
|
assert ("ended after 0s" in out) is (not over), out[-500:]
|
|
assert ("WARNING: MAM_JOB_TREE_MAX=" in out and "ignored" in out) is warns, out[-500:]
|
|
|
|
|
|
def test_b5_runner_knobs_use_the_table_minimum():
|
|
src = (SKILLS / "lib_py" / "loop" / "runner.py").read_text()
|
|
calls = dict(re.findall(r'_int_knob\("([A-Z_]+)", (\d+)\)', src))
|
|
assert {k: int(v) for k, v in calls.items()} == {k: m for k, (m, _) in RUNNER_INT_KNOBS.items()}
|
|
for name in RUNNER_INT_KNOBS:
|
|
assert f'os.environ.get("{name}"' not in src, f"{name} is read raw"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B2 - flags
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_b2_wrapper_flags_are_decimal_or_refused(tmp_path):
|
|
rec = _record(tmp_path, _dry_submit(tmp_path, None, "--timeout", "0700", "--heartbeat", "010",
|
|
"--heartbeat-misses", "04", "--ack-timeout", "0090", "--idle-timeout", "0900"))
|
|
assert (rec["timeout_sec"], rec["heartbeat_sec"], rec["heartbeat_misses"], rec["ack_timeout_sec"],
|
|
rec["idle_timeout_sec"]) == (700, 10, 4, 90, 900)
|
|
ok = _dry_submit(tmp_path, None, "--timeout", "30")
|
|
assert _record(tmp_path, ok)["timeout_sec"] == 30
|
|
for flag in ("--timeout", "--idle-timeout", "--heartbeat", "--heartbeat-misses", "--ack-timeout"):
|
|
for bad in ("abc", "0", "-5", "9999999999", "9" * 5000, ""): # "" given explicitly (review R3)
|
|
res = _dry_submit(tmp_path, None, flag, bad)
|
|
assert res.returncode == 1, (flag, bad[:20], res.stdout, res.stderr[-500:])
|
|
assert f"ERROR: {flag} {bad}: not a positive decimal integer" in res.stderr, res.stderr[-500:]
|
|
assert "registered job" not in res.stdout
|
|
|
|
|
|
def test_b2_runner_job_timeout_is_decimal_or_refused(tmp_path):
|
|
res = _runner(tmp_path, None, "--job-timeout", "0700")
|
|
assert res.returncode == 0, res.stderr
|
|
assert json.loads(res.stdout.strip().splitlines()[-1])["budget"] == 700
|
|
for bad in ("0", "abc", "-5", "9999999999", "9" * 5000):
|
|
res = _runner(tmp_path, None, "--job-timeout", bad)
|
|
assert res.returncode == 1, (bad[:20], res.stderr[-500:])
|
|
assert "--job-timeout requires a positive decimal integer" in res.stderr
|
|
assert "Traceback" not in res.stderr, res.stderr[-500:]
|
|
res = _runner(tmp_path, None, "--job-timeout", "0" * 5000 + "30")
|
|
assert res.returncode == 0 and "Traceback" not in res.stderr, res.stderr[-500:]
|
|
assert json.loads(res.stdout.strip().splitlines()[-1])["budget"] == 30
|
|
|
|
|
|
def test_b2_runner_passes_normalised_contract_flags(tmp_path):
|
|
res = _runner(tmp_path, {"MAM_HEARTBEAT_SEC": "0030", "MAM_HEARTBEAT_MISSES": " 04 ", "MAM_ACK_TIMEOUT_S": "0"})
|
|
assert res.returncode == 0, res.stderr
|
|
contract = json.loads(res.stdout.strip().splitlines()[-1])["contract"]
|
|
assert contract[2:] == ["--heartbeat", "30", "--heartbeat-misses", "4"], contract
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B3 - the derived idle timeout
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_b3_idle_timeout_derives_from_decimal_values(tmp_path):
|
|
rec = _record(tmp_path, _dry_submit(tmp_path, {"MAM_HEARTBEAT_SEC": "010", "MAM_HEARTBEAT_MISSES": "04"}))
|
|
assert (rec["heartbeat_sec"], rec["heartbeat_misses"], rec["idle_timeout_sec"]) == (10, 4, 40)
|
|
|
|
|
|
@pytest.mark.parametrize("knob,default", [("MAM_HEARTBEAT_SEC", 30), ("MAM_HEARTBEAT_MISSES", 4),
|
|
("MAM_ACK_TIMEOUT_S", 180)])
|
|
def test_b3_invalid_contract_env_falls_back_with_a_warning(tmp_path, knob, default):
|
|
field = {"MAM_HEARTBEAT_SEC": "heartbeat_sec", "MAM_HEARTBEAT_MISSES": "heartbeat_misses",
|
|
"MAM_ACK_TIMEOUT_S": "ack_timeout_sec"}[knob]
|
|
res = _dry_submit(tmp_path, {knob: "0"})
|
|
assert _record(tmp_path, res)[field] == default
|
|
assert f"WARNING: {knob}=0 ignored" in res.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B4 - the ack waits
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _job(reg: Path, jid: str, tok: str, started: bool) -> None:
|
|
reg.mkdir(parents=True, exist_ok=True)
|
|
(reg / f"{jid}.json").write_text(json.dumps({"job_id": jid, "status": "pending", "dispatch_token": tok}))
|
|
if started:
|
|
(reg / f"{jid}.events.log").write_text(
|
|
json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": tok}}) + "\n")
|
|
|
|
|
|
@pytest.mark.parametrize("grace,started,rc", [("0900", True, 0), ("0", True, 0), ("0", False, 1)])
|
|
def test_b4_wrapper_inject_ack_grace(tmp_path, grace, started, rc):
|
|
reg = tmp_path / "jobs"
|
|
_job(reg, "abcd1234", "tok", started)
|
|
script = (f'source "{WRAPPER}"; set +e; REGISTRY_DIR="{reg}"; PY="{sys.executable}"; TIMEOUT=60; '
|
|
f'START_TIME=$(date +%s); _disp_tok="tok"; _inject_acknowledged abcd1234; echo "rc=$?"')
|
|
t0 = time.time()
|
|
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True, cwd=str(tmp_path), timeout=60,
|
|
env=_base_env(tmp_path, {"MAM_DELEGATE_JOB_NO_MAIN": "1", "MAM_INJECT_ACK_GRACE_S": grace}))
|
|
assert f"rc={rc}" in res.stdout, (res.stdout, res.stderr)
|
|
assert "value too great for base" not in res.stderr and "ignored" not in res.stderr, res.stderr
|
|
assert time.time() - t0 < 10, "0 must mean check once, not wait"
|
|
|
|
|
|
@pytest.mark.parametrize("ack,started,rc", [("0900", True, 0), ("0", True, 0), ("0", False, 1)])
|
|
def test_b4_delegate_wait_started(tmp_path, ack, started, rc):
|
|
reg = tmp_path / "jobs"
|
|
_job(reg, "abcd1234", "tok", started)
|
|
t0 = time.time()
|
|
res = _bash(f'delegate_wait_started abcd1234 {ack} tok; echo "rc=$?"', tmp_path, {"MAM_JOBS_DIR": str(reg)})
|
|
assert f"rc={rc}" in res.stdout, (res.stdout, res.stderr)
|
|
assert "value too great for base" not in res.stderr and "ignored" not in res.stderr, res.stderr
|
|
assert time.time() - t0 < 10
|
|
|
|
|
|
def test_b4_delegate_wait_started_rejects_garbage_with_a_warning(tmp_path):
|
|
reg = tmp_path / "jobs"
|
|
_job(reg, "abcd1234", "tok", True)
|
|
res = _bash('delegate_wait_started abcd1234 abc tok; echo "rc=$?"', tmp_path, {"MAM_JOBS_DIR": str(reg)})
|
|
assert "rc=0" in res.stdout and "onboarding ack timeout abc ignored" in res.stderr, res.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B5 - static guard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ARITH = re.compile(r"\$\(\((.*?)\)\)|\(\((.*?)\)\)|\[\[(.*?)\]\]")
|
|
_ENV_DEFAULT = re.compile(r"\$\{(MAM|SKS|DELEGATE_JOB)_[A-Z0-9_]*:-")
|
|
|
|
|
|
@pytest.mark.parametrize("path", [WRAPPER, LIB_SH], ids=lambda p: p.name)
|
|
def test_b5_no_env_default_inside_arithmetic(path):
|
|
bad = []
|
|
for n, line in enumerate(path.read_text().splitlines(), 1):
|
|
for m in _ARITH.finditer(line):
|
|
arith, cmd, test = m.groups()
|
|
body = arith if arith is not None else cmd if cmd is not None else test
|
|
is_arith = test is None or re.search(r"\s-(gt|ge|lt|le|eq|ne)\s", body)
|
|
if is_arith and _ENV_DEFAULT.search(body):
|
|
bad.append(f"{path.name}:{n}: {line.strip()}")
|
|
assert not bad, "normalise with mam_int_knob first:\n" + "\n".join(bad)
|
|
|
|
|
|
def test_b5_every_normalised_knob_uses_the_table_minimum():
|
|
src = WRAPPER.read_text() + "\n" + LIB_SH.read_text()
|
|
calls = re.findall(r"mam_int_knob ([A-Z_]+) (\S+) (\d+)\)", src)
|
|
seen = {}
|
|
for name, default, minimum in calls:
|
|
assert name in INT_KNOBS, f"{name} is normalised but missing from INT_KNOBS"
|
|
want_min, want_default = INT_KNOBS[name]
|
|
assert int(minimum) == want_min, (name, minimum)
|
|
assert default.strip('"') == want_default, (name, default)
|
|
seen[name] = True
|
|
env_knobs = {k for k in INT_KNOBS if re.fullmatch(r"[A-Z_]+", k)}
|
|
assert set(seen) == env_knobs, f"in INT_KNOBS but never normalised: {env_knobs - set(seen)}"
|
|
|
|
# delegate_wait_started's argument and the wrapper's flags
|
|
assert re.search(r'ack_s=\$\(mam_int "\$ack_s" 0\)', src)
|
|
loop = re.search(r"for _flag in ([^;]+); do\n(.*?)done", WRAPPER.read_text(), re.S)
|
|
flags = {f.split(":")[0] for f in loop.group(1).split()}
|
|
assert {f"--{f}" for f in flags} == {k for k in INT_KNOBS if k.startswith("--")}
|
|
assert 'mam_int "${!_var}" 1' in loop.group(2)
|
|
|
|
|
|
def test_b5_every_integer_env_default_is_classified():
|
|
"""Review R2: a knob can reach arithmetic through a function argument, which
|
|
the direct-pattern guard above cannot see. Every integer-looking env default
|
|
in the shell code is therefore either normalised (INT_KNOBS) or listed in
|
|
NOT_ARITHMETIC with where it goes."""
|
|
files = [WRAPPER, *sorted(SKILLS.glob("**/*.sh"))]
|
|
found = {}
|
|
for f in files:
|
|
for m in re.finditer(r"\$\{((?:MAM|SKS|DELEGATE_JOB|RECONCILE)_[A-Z0-9_]*):-[0-9]", f.read_text()):
|
|
found.setdefault(m.group(1), f.relative_to(REPO_ROOT))
|
|
raw_normalised = {k: v for k, v in found.items() if k in INT_KNOBS}
|
|
assert not raw_normalised, f"normalised knobs read raw (use mam_int_knob): {raw_normalised}"
|
|
unclassified = {k: str(v) for k, v in found.items() if k not in NOT_ARITHMETIC}
|
|
assert not unclassified, f"classify in INT_KNOBS or NOT_ARITHMETIC: {unclassified}"
|
|
stale = set(NOT_ARITHMETIC) - set(found)
|
|
assert not stale, f"no longer in the code: {stale}"
|
|
|
|
|
|
@pytest.mark.parametrize("value,rc", [("900", 0), ("0900", 0), ("0", 2)])
|
|
def test_b5_pane_quiescent_reads_the_tries_knob_as_decimal(tmp_path, value, rc):
|
|
"""The consuming function with the call site's normalisation (review R2):
|
|
0900 used to raise 'value too great for base' and return 2 (headless)."""
|
|
line = next(l for l in LIB_SH.read_text().splitlines() if l.strip().startswith('_pane_quiescent "$sess"'))
|
|
assert '"$(mam_int_knob SKS_QUIESCENT_TRIES 20 0)"' in line
|
|
res = _bash('_pane_capture() { echo stable; }; '
|
|
'_pane_quiescent s "$(mam_int_knob SKS_QUIESCENT_TRIES 20 0)" 0; echo "rc=$?"',
|
|
tmp_path, {"SKS_QUIESCENT_TRIES": value})
|
|
assert f"rc={rc}" in res.stdout, (res.stdout, res.stderr)
|
|
assert "value too great" not in res.stderr and "ignored" not in res.stderr
|