GodopuandClaude Fable 5.1
6c99fedf40
open_jobs reports a live pending record nobody signals on as `injected`: its producer registered and injected it without a waiter (create --onboard, a job sent by hand), so it holds the session like a running or admitted job, and a waiter of the same second with a lower job id, or one registered a moment earlier and still in its preflight, no longer passes it in the FIFO. The waiter signals on its own record before its first scan, so its record is never mistaken for one. lib_py.pane_lock resolves the shim's ambient target (no -L, HERDR_SESSION_NAME empty or `default`) to the server of the pane the caller runs in - HERDR_SESSION, else HERDR_SOCKET_PATH's layout, as lib.sh _herdr_caller_server reads them - so an ambient and a named call into one server claim under one lock; outside a pane it stays the default server's. Tests: u2h (the tie, the later-registered onboarding, the signalled control), u1/u4 labels, c10 forces the same-second registration, p1b, s2 (real herdr: held lock stops both call forms; mixed concurrent calls take distinct panes). Two FIFO fixtures now carry a waiter's signal. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
504 lines
28 KiB
Python
504 lines
28 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"),
|
|
# Follow-up F4b: `[ -ge ]`/`[ -le ]`/`[ -gt ]` operands and integer sleeps, where
|
|
# "abc" was a `[` error or a sleep error. lib.sh, reconcile.sh (which sources
|
|
# it), and loop_lock.sh / diff_collect.sh, which run without lib.sh and
|
|
# source lib_int.sh themselves.
|
|
"MAM_READY_EMPTY_GIVEUP": (1, "5"), # consecutive empty captures; 0 would act as 1
|
|
"SKS_EMPTY_GIVEUP": (1, "3"), # the same, in _pane_quiescent
|
|
"SKS_IDLE_RETRIES": (0, "2"), # 0: one wait, no retry
|
|
"SKS_IDLE_RETRY_BACKOFF_S": (0, "10"), # sleep between retries; test_issue6 uses 0
|
|
"MAM_LOCK_STEAL_RETRIES": (0, "3"), # 0: one attempt, no steal retry
|
|
"MAM_DIFF_MAX_BYTES": (1, "200000"), # 0 would truncate every diff
|
|
"MAM_DIFF_MAX_LINES": (1, "4000"),
|
|
"RECONCILE_POLL_INTERVAL": (1, "15"), # sleep; 0 would poll in a busy loop
|
|
# Issue #50: create's wait for the onboarding job to end (0 would exit 45 at
|
|
# once) and the wrapper's registry wait (0: decide at once - clear or busy)
|
|
"MAM_ONBOARD_WAIT_S": (1, "600"),
|
|
"MAM_DISPATCH_WAIT_S": (0, "600"),
|
|
# 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.
|
|
# Issue #17 item 4: MAM_DELEGATE_FAIL_CLOSED and MAM_STALL_WATCH are read by
|
|
# lib_py/delegation/wrapper.py now, not by shell code; MAM_PANE_CLAIM_WAIT_S
|
|
# (minimum 1, default 120) by lib_py/pane_lock.py (test_issue50_pane_lock P5).
|
|
NOT_ARITHMETIC = {
|
|
"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()",
|
|
# sleep's fractional seconds are its meaning here (default 0.5): mam_int would refuse them
|
|
"SKS_QUIESCENT_INTERVAL": "sleep, fractional by default (0.5) - left raw",
|
|
"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 = "\n".join(f.read_text() for f in [WRAPPER, *sorted(SKILLS.glob("**/*.sh"))])
|
|
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}"
|
|
|
|
|
|
LOOP_LOCK = SKILLS / "multi-agent-mux-loop" / "scripts" / "loop_lock.sh"
|
|
DIFF_COLLECT = SKILLS / "multi-agent-mux-loop" / "scripts" / "diff_collect.sh"
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["abc", "0900", " 7 ", "-1", "1.5", "0", ""], ids=_id)
|
|
@pytest.mark.parametrize("script,knobs", [(LOOP_LOCK, ["MAM_LOCK_STEAL_RETRIES"]),
|
|
(DIFF_COLLECT, ["MAM_DIFF_MAX_BYTES", "MAM_DIFF_MAX_LINES"])],
|
|
ids=lambda v: v.name if isinstance(v, Path) else "")
|
|
def test_b5_scripts_sourced_without_lib_sh_read_their_knobs_as_decimal(tmp_path, script, knobs, value):
|
|
"""F4b: these files are sourced alone (the runner, the tests): they bring
|
|
the integer rule with them, and "abc" no longer reaches `[ -le ]`/`[ -gt ]`."""
|
|
probe = "; ".join(f'printf "%s=%s\\n" {k} "${k}"' for k in knobs)
|
|
res = subprocess.run(["bash", "-c", f'source "{script}"; {probe}'], capture_output=True, text=True,
|
|
env=_base_env(tmp_path, {k: value for k in knobs}), cwd=str(tmp_path), timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
for k in knobs:
|
|
minimum, default = INT_KNOBS[k]
|
|
want = _expected(value, minimum)
|
|
assert f"{k}={default if want is None else want}" in res.stdout.splitlines(), (res.stdout, res.stderr)
|
|
assert (f"WARNING: {k}=" in res.stderr) is (want is None and value != ""), res.stderr
|
|
|
|
|
|
def test_b5_pane_quiescent_gives_up_on_empty_captures_with_a_bad_knob(tmp_path):
|
|
"""F4b: SKS_EMPTY_GIVEUP=abc was a `[` error on every empty capture."""
|
|
res = _bash('_pane_capture() { :; }; _pane_quiescent s 20 0; echo "rc=$?"', tmp_path, {"SKS_EMPTY_GIVEUP": "abc"})
|
|
assert "rc=2" in res.stdout, (res.stdout, res.stderr)
|
|
assert "integer expression expected" not in res.stderr, res.stderr
|
|
assert "WARNING: SKS_EMPTY_GIVEUP=abc ignored" in res.stderr, res.stderr
|
|
|
|
|
|
@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
|