GodopuandClaude Opus 5.5
2826e709ae
- tests/_local_mqtt.py: in-process MQTT 3.1.1 broker on 127.0.0.1 (what the repo uses: protocol level 4, clean session, QoS 0/1, retained). - conftest: one session broker named by MQTT_BROKER/MQTT_PORT/MQTT_TLS; other redirecting/auth MQTT_* dropped and restored afterwards. - Harnesses of the 22 broker-dependent tests (#18 acceptance1/2/3/5, #6 t1_1/2/7/8/14/15/16/20, #19 MQTT loop x2, #33 b3/r3/b7, #30 s9/s10/r2) use it; the "broker unreachable" skips are gone (a local broker that does not deliver is a failure). acceptance1/5 and #19 also assert on the subscriber, runner and audit logs the real producers write. - tests/test_local_mqtt.py: conformance with the real clients and the real publish_event -> job_subscriber path (C0), an AST guard against public broker literals and MQTT_* deleted without a replacement (C2), and the session's broker config (C3). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
363 lines
16 KiB
Python
363 lines
16 KiB
Python
"""Issue #33: the job budget travels, the wait caps follow it, and the two
|
|
clocks that close a job carry different labels.
|
|
|
|
- B1: the real wrapper takes the budget from --timeout > $MAM_JOB_TIMEOUT_S > 3600.
|
|
- B2: the loop runner derives its wait caps from the budget (+300 s) and hands
|
|
the wrapper that budget explicitly; both hops run for real.
|
|
- B3: the real wrapper + subscriber label a spent budget `wall_clock_timeout`
|
|
(subscriber rc 4) and silence `deadline_timeout` (rc 2); caller rc is 2 for
|
|
both. These need the live broker from the environment (like test_issue6 t1_1).
|
|
- B4: the remote runner's two deadline wordings lead with the new label.
|
|
- B5: the sub-job table files `wall_clock_timeout` closes as unconfirmed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from _local_mqtt import local_broker_env
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
REGISTRY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
|
|
PUBLISH = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
|
|
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
|
|
UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
|
|
|
|
|
|
def _env(home: Path, extra=None, broker: bool = False) -> dict:
|
|
env = dict(os.environ)
|
|
env["DELEGATE_JOB_PYTHON"] = sys.executable
|
|
env["AGENT_PYTHON_BIN"] = sys.executable
|
|
env["HOME"] = str(home)
|
|
env["WORKSPACE_ROOT"] = str(home)
|
|
env["PYTHONPATH"] = str(SKILLS)
|
|
env["DELEGATE_JOB_LOGS_DIR"] = str(home / "logs")
|
|
for k in ("MAM_JOBS_DIR", "MAM_REAL_ROOT", "MAM_JOB_ID", "MAM_JOB_TIMEOUT_S", "MAM_CHILD_WAIT_S",
|
|
"MAM_WAIT_GRACE_S"):
|
|
env.pop(k, None)
|
|
for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"):
|
|
env.pop(k, None)
|
|
if broker:
|
|
env.update(local_broker_env()) # Issue #26: the session's in-process broker
|
|
else:
|
|
env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
|
|
env.update(UNREACHABLE)
|
|
if extra:
|
|
env.update({k: str(v) for k, v in extra.items()})
|
|
return env
|
|
|
|
|
|
def _dry_submit(env: dict, reg: Path, *extra: str) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--dry-run",
|
|
"--registry-dir", str(reg), "--prompt", "p", *extra],
|
|
capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=120)
|
|
|
|
|
|
def _jid(res: subprocess.CompletedProcess) -> str:
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
m = re.search(r"registered job:\s*(\S+)", res.stdout)
|
|
assert m, res.stdout
|
|
return m.group(1)
|
|
|
|
|
|
def _record(reg: Path, jid: str) -> dict:
|
|
return json.loads((reg / f"{jid}.json").read_text(encoding="utf-8"))
|
|
|
|
|
|
def _events(reg: Path, jid: str) -> list:
|
|
p = reg / f"{jid}.events.log"
|
|
return [json.loads(l) for l in p.read_text().splitlines() if l.strip()] if p.exists() else []
|
|
|
|
|
|
def _publish(env: dict, reg: Path, jid: str, event: str, *extra: str) -> subprocess.CompletedProcess:
|
|
tok = _record(reg, jid)["dispatch_token"]
|
|
res = subprocess.run([sys.executable, str(PUBLISH), "--registry-dir", str(reg), "--job", jid,
|
|
"--event", event, "--expect-dispatch-token", tok, *extra],
|
|
capture_output=True, text=True, env=env, timeout=120)
|
|
assert res.returncode in (0, 2), res.stderr
|
|
return res
|
|
|
|
|
|
def _kill_timer(reg: Path, jid: str) -> None:
|
|
try:
|
|
os.kill(int((reg / f"{jid}.heartbeat.pid").read_text().strip()), signal.SIGKILL)
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B1 - the wrapper's budget: --timeout > MAM_JOB_TIMEOUT_S > 3600
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("env_val,flag,expected,warns", [
|
|
(None, None, 3600, False),
|
|
("7", None, 7, False),
|
|
("7", "9", 9, False),
|
|
("abc", None, 3600, True),
|
|
])
|
|
def test_b1_the_wrapper_takes_the_budget_from_flag_env_or_default(tmp_path, env_val, flag, expected, warns):
|
|
env = _env(tmp_path, {"MAM_JOB_TIMEOUT_S": env_val} if env_val else None)
|
|
res = _dry_submit(env, tmp_path / "reg", *(["--timeout", flag] if flag else []))
|
|
assert _record(tmp_path / "reg", _jid(res))["timeout_sec"] == expected
|
|
assert ("MAM_JOB_TIMEOUT_S=abc ignored" in res.stderr) is warns, res.stderr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B2 - the runner: budget -> wait caps, and the budget handed to the wrapper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _runner(tmp_path, monkeypatch, *argv, **env):
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
|
monkeypatch.setenv("MAM_JOBS_DIR", str(tmp_path / "reg"))
|
|
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
|
|
for k in ("MAM_REAL_ROOT", "MAM_JOB_TIMEOUT_S", "MAM_CHILD_WAIT_S", "MAM_WAIT_GRACE_S", "MAM_JOB_TREE_MAX"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
for k, v in env.items():
|
|
monkeypatch.setenv(k, str(v))
|
|
from lib_py.loop import runner as lr
|
|
r = lr.LoopRunner(["--creator", "c", "--task", "t", *argv])
|
|
r.parse_args()
|
|
return r
|
|
|
|
|
|
def test_b2_the_wait_caps_are_the_budget_plus_300(tmp_path, monkeypatch):
|
|
assert _runner(tmp_path, monkeypatch, "--job-timeout", "10").job_wait_s == 310
|
|
assert _runner(tmp_path, monkeypatch, MAM_JOB_TIMEOUT_S=20).job_wait_s == 320
|
|
assert _runner(tmp_path, monkeypatch, "--job-timeout", "10", MAM_JOB_TIMEOUT_S=20).job_wait_s == 310
|
|
r = _runner(tmp_path, monkeypatch)
|
|
assert (r.job_budget, r.job_wait_s) == (3600, 3900) # no option: today's numbers
|
|
assert "--job-timeout" in "\n".join(_usage_lines(r))
|
|
|
|
|
|
def _usage_lines(r):
|
|
import io, contextlib
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
|
|
try:
|
|
r.print_usage()
|
|
except SystemExit:
|
|
pass
|
|
return buf.getvalue().splitlines()
|
|
|
|
|
|
def test_b2_an_invalid_job_timeout_is_refused(tmp_path, monkeypatch):
|
|
with pytest.raises(SystemExit):
|
|
_runner(tmp_path, monkeypatch, "--job-timeout", "0")
|
|
|
|
|
|
def test_b2_the_child_wait_defaults_to_the_budget_and_an_explicit_value_wins(tmp_path, monkeypatch, capsys):
|
|
env = _env(tmp_path)
|
|
reg = tmp_path / "reg"
|
|
p = _jid(_dry_submit(env, reg))
|
|
_jid(_dry_submit(env, reg, "--parent-job", p))
|
|
|
|
start = time.time()
|
|
r = _runner(tmp_path, monkeypatch, "--job-timeout", "1", MAM_WAIT_GRACE_S=1)
|
|
assert r.wait_for_descendants(p, interval=0.5) is False
|
|
assert 2 <= time.time() - start < 8
|
|
assert "ended after 2s (MAM_CHILD_WAIT_S default: job budget 1 + 1)" in capsys.readouterr().out
|
|
|
|
start = time.time()
|
|
r = _runner(tmp_path, monkeypatch, "--job-timeout", "600", MAM_CHILD_WAIT_S=2)
|
|
assert r.wait_for_descendants(p, interval=0.5) is False
|
|
assert 2 <= time.time() - start < 8
|
|
assert "ended after 2s (MAM_CHILD_WAIT_S)" in capsys.readouterr().out
|
|
|
|
|
|
def test_b2_the_runner_hands_the_wrapper_its_budget_and_the_wrapper_registers_it(tmp_path, monkeypatch):
|
|
# hop 1: the runner execs the wrapper with --timeout (a recording wrapper at the control root)
|
|
rec = tmp_path / "rec.args"
|
|
w = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
w.parent.mkdir(parents=True)
|
|
w.write_text(f'#!/usr/bin/env bash\nprintf "%s\\n" "$@" > "{rec}"\necho "registered job: fake33"\n')
|
|
w.chmod(0o755)
|
|
r = _runner(tmp_path, monkeypatch, "--job-timeout", "10")
|
|
assert r.delegate_job("c", "claude", "Worker", "task") == "fake33"
|
|
args = rec.read_text().splitlines()
|
|
assert args[args.index("--timeout") + 1] == "10"
|
|
# env only: still passed explicitly
|
|
r = _runner(tmp_path, monkeypatch, MAM_JOB_TIMEOUT_S=20)
|
|
assert r.contract_args()[:2] == ["--timeout", "20"]
|
|
|
|
# hop 2: the real wrapper registers what the runner hands it
|
|
reg = tmp_path / "reg2"
|
|
res = _dry_submit(_env(tmp_path), reg, *_runner(tmp_path, monkeypatch, "--job-timeout", "10").contract_args())
|
|
assert _record(reg, _jid(res))["timeout_sec"] == 10
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B3 - the real wrapper + subscriber: which clock closed the job
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _run_wrapper_bg(env: dict, reg: Path, *extra: str) -> subprocess.Popen:
|
|
return subprocess.Popen(["bash", str(WRAPPER), "submit", "--agent", "human", "--registry-dir", str(reg),
|
|
"--workdir", env["HOME"], "--prompt", "b3", *extra],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env, cwd=env["HOME"])
|
|
|
|
|
|
def _wait_subscribed(reg: Path, timeout: float = 30) -> str:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
for f in reg.glob("*.subscriber.out"):
|
|
if "SUBSCRIBED" in f.read_text(errors="replace"):
|
|
return f.name.split(".")[0]
|
|
time.sleep(0.1)
|
|
raise AssertionError("the subscriber never subscribed (is the broker up?)")
|
|
|
|
|
|
def _last_error(reg: Path, jid: str) -> str:
|
|
errs = [e for e in _events(reg, jid) if e.get("event") == "error"]
|
|
assert errs, _events(reg, jid)
|
|
return errs[-1].get("detail", "")
|
|
|
|
|
|
def test_b3_wall_clock_before_idle_is_labelled_wall_clock_timeout(tmp_path):
|
|
env = _env(tmp_path, broker=True)
|
|
reg = tmp_path / "jobs"
|
|
proc = _run_wrapper_bg(env, reg, "--timeout", "3", "--idle-timeout", "30")
|
|
out, err = proc.communicate(timeout=90)
|
|
assert proc.returncode == 2, out + err
|
|
jid = next(reg.glob("*.json")).stem
|
|
assert _record(reg, jid)["status"] == "error"
|
|
assert _last_error(reg, jid) == "wall_clock_timeout"
|
|
assert "wall-clock timeout" in (reg / f"{jid}.subscriber.out").read_text()
|
|
assert "closed by wall_clock clock" in err
|
|
|
|
|
|
def test_b3_silence_is_still_labelled_deadline_timeout(tmp_path):
|
|
env = _env(tmp_path, broker=True)
|
|
reg = tmp_path / "jobs"
|
|
proc = _run_wrapper_bg(env, reg, "--timeout", "30", "--idle-timeout", "2")
|
|
try:
|
|
jid = _wait_subscribed(reg)
|
|
_publish(env, reg, jid, "started") # then nothing: silence for K x N
|
|
out, err = proc.communicate(timeout=90)
|
|
finally:
|
|
proc.kill()
|
|
assert proc.returncode == 2, out + err
|
|
assert _record(reg, jid)["status"] == "error"
|
|
assert _last_error(reg, jid) == "deadline_timeout"
|
|
assert "idle timeout" in (reg / f"{jid}.subscriber.out").read_text()
|
|
assert "closed by no_event clock" in err
|
|
|
|
|
|
def test_b3_a_heartbeating_job_that_spends_its_budget_is_wall_clock_timeout(tmp_path):
|
|
"""The #33 motivating case: a live agent (heartbeats keep coming) whose
|
|
work outlasts the budget was reported as `deadline_timeout`."""
|
|
env = _env(tmp_path, broker=True)
|
|
reg = tmp_path / "jobs"
|
|
proc = _run_wrapper_bg(env, reg, "--timeout", "6", "--idle-timeout", "3")
|
|
jid = None
|
|
try:
|
|
jid = _wait_subscribed(reg)
|
|
_publish(env, reg, jid, "started", "--heartbeat", "1")
|
|
out, err = proc.communicate(timeout=90)
|
|
finally:
|
|
proc.kill()
|
|
if jid:
|
|
_kill_timer(reg, jid)
|
|
assert proc.returncode == 2, out + err
|
|
assert _last_error(reg, jid) == "wall_clock_timeout"
|
|
assert "wall-clock timeout" in (reg / f"{jid}.subscriber.out").read_text()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B4 - the remote runner's deadlines are wall clocks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _dispatch(timeout_sec=1) -> dict:
|
|
return {"job_id": uuid.uuid4().hex[:8], "brief": "b", "dispatch_token": "tok-" + uuid.uuid4().hex,
|
|
"attempt_id": 1, "auth_token": None, "timeout_sec": timeout_sec, "heartbeat_sec": 30,
|
|
"heartbeat_misses": 4, "ack_timeout_sec": 5}
|
|
|
|
|
|
@pytest.mark.parametrize("cmd,prefix", [("sleep 5", "wall_clock_timeout: agent command exceeded timeout_sec 1"),
|
|
(None, "wall_clock_timeout (remote runner)")])
|
|
def test_b4_the_remote_runner_labels_its_deadline_wall_clock_timeout(tmp_path, monkeypatch, cmd, prefix):
|
|
from lib_py.delegation import remote
|
|
for k, v in _env(tmp_path).items():
|
|
monkeypatch.setenv(k, v)
|
|
monkeypatch.chdir(tmp_path)
|
|
reg = tmp_path / "jobs"
|
|
d = _dispatch()
|
|
try:
|
|
remote.handle_dispatch(d, str(reg), "s", cmd)
|
|
assert _record(reg, d["job_id"])["status"] == "error"
|
|
assert _last_error(reg, d["job_id"]).startswith(prefix)
|
|
finally:
|
|
_kill_timer(reg, d["job_id"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B5 - the sub-job table: a wall-clock close is unconfirmed, like a deadline close
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("request_kind,detail", [
|
|
("wall_clock_timeout", "wall_clock_timeout"),
|
|
("wall_clock_timeout", "wall_clock_timeout: agent command exceeded timeout_sec 1"),
|
|
("deadline_timeout", "deadline_timeout"),
|
|
])
|
|
def test_b5_a_wall_clock_close_is_closed_unconfirmed(tmp_path, monkeypatch, request_kind, detail):
|
|
env = _env(tmp_path)
|
|
reg = tmp_path / "reg"
|
|
p = _jid(_dry_submit(env, reg))
|
|
c = _jid(_dry_submit(env, reg, "--parent-job", p))
|
|
tok = _record(reg, c)["dispatch_token"]
|
|
res = subprocess.run([sys.executable, str(REGISTRY), "--registry-dir", str(reg), "close", "--job", c,
|
|
"--request", request_kind, "--expect-dispatch-token", tok],
|
|
capture_output=True, text=True, env=env, timeout=120)
|
|
assert res.returncode == 0, res.stderr
|
|
assert _record(reg, c)["status"] == "error"
|
|
_publish(env, reg, c, "error", "--detail", detail)
|
|
block = _runner(tmp_path, monkeypatch)._subjobs_block([p])
|
|
assert f"| CLOSED-UNCONFIRMED | {detail} |" in block, block
|
|
|
|
|
|
# Review 5ec8ef33 R3: the broker-unavailable wait that runs into the deadline
|
|
# is the wall clock too (caller rc stays 3); broker_up is the passing twin.
|
|
@pytest.mark.parametrize("broker_up", [True, False])
|
|
def test_r3_a_budget_spent_waiting_on_an_unreachable_broker_is_wall_clock_timeout(tmp_path, broker_up):
|
|
env = _env(tmp_path, broker=broker_up)
|
|
reg = tmp_path / "jobs"
|
|
p = _run_wrapper_bg(env, reg, "--timeout", "4", "--idle-timeout", "30")
|
|
try:
|
|
out, err = p.communicate(timeout=90)
|
|
jid = next(reg.glob("*.json")).stem
|
|
assert p.returncode == (2 if broker_up else 3), out + err
|
|
assert _record(reg, jid)["status"] == "error"
|
|
assert _last_error(reg, jid) == "wall_clock_timeout", err
|
|
finally:
|
|
if p.poll() is None:
|
|
p.kill()
|
|
p.communicate(timeout=10)
|
|
|
|
|
|
|
|
# Issue #33 follow-up (job 60da22f6): `delegate-job wait` passes the subscriber's
|
|
# code through; a spent budget stays rc 2 for its callers, as before the split.
|
|
@pytest.mark.parametrize("state,rc", [("running", 2), ("completed", 0)])
|
|
def test_b7_wait_keeps_a_spent_budget_at_rc_2(tmp_path, state, rc):
|
|
env = _env(tmp_path, broker=True)
|
|
reg = tmp_path / "reg"
|
|
jid = _jid(_dry_submit(env, reg))
|
|
if state == "completed":
|
|
_publish(env, reg, jid, "completed")
|
|
res = subprocess.run(["bash", str(WRAPPER), "wait", "--job", jid, "--registry-dir", str(reg), "--timeout", "2"],
|
|
capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=60)
|
|
assert res.returncode == rc, res.stdout + res.stderr
|