GodopuandClaude Fable 5.1
0550320590
Review round 1 (claude NOT PASS) corrections: - B2: the timer now watches a process - `--watch-pid`, or by default the nearest agent-CLI ancestor of the publishing command - and stops within one period once it is gone, so a crashed agent stops heartbeating and is closed after K x N; a remote runner watches itself. Documented as process liveness (an alive-but-stuck agent keeps heartbeating). - R1: `_wait_started` accepts an events.log `started` only when it carries the current dispatch token (loop jobs share one events.log). - R2: the remote runner ignores a re-sent dispatch of an attempt that already reached its terminal. - R3: registry.md EOF blank line (git diff --check clean). - R4: the ack error states the effective budget when the wall clock cut it. - P3: stop_timer verifies the pidfile pid is this job's timer before signalling; a retained state message keeps its own seq watermark so it can never make the monotonic guard drop its own event. - MULTI_AGENT_RULES.md timeout section updated for the contract. - tests: agent-process death stops the timer, recycled-pid safety, attempt-scoped started, re-sent dispatch; acceptance 3 now kills the agent process instead of the timer. Help captures refreshed (--watch-pid). The wrapper can be sourced with MAM_DELEGATE_JOB_NO_MAIN=1 for helper-level tests. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
735 lines
36 KiB
Python
735 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
"""Issue #18 — liveness and delivery are decided by the event protocol, not by
|
||
screen scraping.
|
||
|
||
Contract (registry fields, wrapper options):
|
||
* heartbeat obligation - the agent publishes `progress` at least every N s;
|
||
`publish_event --event started --heartbeat N` starts an automatic timer
|
||
that does it for the agent and stops itself at the terminal event.
|
||
* unresponsive - no event (disk or wire) for K x N s closes the job
|
||
as deadline_timeout; herdr status and pane text are diagnostics only.
|
||
* delivery - `started` must arrive within T s of the injection,
|
||
else the inject fails closed (rc 1, registry error, single injection).
|
||
* wire durability - every event is followed by a retained
|
||
`<prefix>/state` message on the same connection; a subscriber that
|
||
(re)connects recovers the status at once and records seq gaps.
|
||
* remote transport - `--agent-session remote:<name>` delivers the brief
|
||
over a signed retained `<prefix>/dispatch`; a runner on another host with
|
||
its own registry and broker credentials runs it, and the orchestrator's
|
||
subscriber mirrors its events into the orchestrator's registry.
|
||
|
||
Tests marked "broker" talk to the configured broker (the public default in the
|
||
sandbox); each stays under a minute.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
SKILLS = REPO_ROOT / ".agents" / "skills"
|
||
LIB_SH = SKILLS / "lib.sh"
|
||
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
||
SCRIPTS = SKILLS / "multi-agent-mux-delegate-job" / "scripts"
|
||
PUBLISH_PY = SCRIPTS / "publish_event.py"
|
||
REGISTRY_PY = SCRIPTS / "registry.py"
|
||
REMOTE_PY = SCRIPTS / "remote.py"
|
||
|
||
UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
|
||
|
||
|
||
def _env(cwd: Path, extra=None) -> dict:
|
||
env = dict(os.environ)
|
||
env["DELEGATE_JOB_PYTHON"] = sys.executable
|
||
env["AGENT_PYTHON_BIN"] = sys.executable
|
||
env["WORKSPACE_ROOT"] = str(cwd)
|
||
env["HOME"] = str(cwd)
|
||
env["MAM_ENV_FILE"] = str(cwd / ".mam.env.absent")
|
||
env["DELEGATE_JOB_LOGS_DIR"] = str(cwd / "logs")
|
||
env["PYTHONPATH"] = str(SKILLS)
|
||
for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"):
|
||
env.pop(k, None)
|
||
if extra:
|
||
env.update({k: str(v) for k, v in extra.items()})
|
||
return env
|
||
|
||
|
||
def _events(reg_dir: Path, jid: str) -> list:
|
||
p = reg_dir / f"{jid}.events.log"
|
||
if not p.exists():
|
||
return []
|
||
return [json.loads(l) for l in p.read_text().splitlines() if l.strip()]
|
||
|
||
|
||
def _job(reg_dir: Path, jid: str) -> dict:
|
||
return json.loads((reg_dir / f"{jid}.json").read_text())
|
||
|
||
|
||
def _wait_job_id(reg_dir: Path, timeout=20) -> str:
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
files = list(reg_dir.glob("*.json"))
|
||
if files:
|
||
return files[0].stem
|
||
time.sleep(0.1)
|
||
raise AssertionError("job was not registered")
|
||
|
||
|
||
def _publish(reg_dir: Path, jid: str, event: str, env: dict, *extra) -> subprocess.CompletedProcess:
|
||
"""Publish as the agent would: with the dispatch identity the wrapper's
|
||
injected command carries (`--expect-dispatch-token`)."""
|
||
tok = _job(reg_dir, jid).get("dispatch_token")
|
||
ident = ["--expect-dispatch-token", tok] if tok else []
|
||
return subprocess.run([sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir), "--job", jid,
|
||
"--event", event, "--attempts", "2", *ident, *extra],
|
||
capture_output=True, text=True, env=env, cwd=str(reg_dir.parent), timeout=120)
|
||
|
||
|
||
def _timer_pid(reg_dir: Path, jid: str):
|
||
p = reg_dir / f"{jid}.heartbeat.pid"
|
||
try:
|
||
return int(p.read_text().strip())
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
|
||
def _alive(pid) -> bool:
|
||
if not pid:
|
||
return False
|
||
try:
|
||
os.kill(pid, 0)
|
||
except ProcessLookupError:
|
||
return False
|
||
return subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True).stdout.strip() not in ("", "Z", "Z+")
|
||
|
||
|
||
def _kill_timer(reg_dir: Path, jid: str):
|
||
pid = _timer_pid(reg_dir, jid)
|
||
if pid:
|
||
try:
|
||
os.kill(pid, signal.SIGKILL)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. registry: the contract is part of the record
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_register_derives_the_idle_window_from_the_contract(tmp_path, monkeypatch):
|
||
from lib_py.delegation import registry
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
|
||
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
|
||
reg = tmp_path / "jobs"
|
||
jid = registry.register_job("contract", registry_dir=str(reg))
|
||
rec = _job(reg, jid)
|
||
assert (rec["heartbeat_sec"], rec["heartbeat_misses"], rec["ack_timeout_sec"]) == (30, 4, 60)
|
||
assert rec["idle_timeout_sec"] == 120 == registry.liveness_window(30, 4)
|
||
assert rec["transport"] == "herdr"
|
||
|
||
jid2 = registry.register_job("contract", registry_dir=str(reg), heartbeat_sec=5, heartbeat_misses=3)
|
||
assert _job(reg, jid2)["idle_timeout_sec"] == 15
|
||
jid3 = registry.register_job("contract", registry_dir=str(reg), heartbeat_sec=5, heartbeat_misses=3,
|
||
idle_timeout_sec=40)
|
||
assert _job(reg, jid3)["idle_timeout_sec"] == 40, "an explicit idle timeout is kept"
|
||
|
||
|
||
def test_wrapper_registers_the_contract_and_derives_idle_timeout(tmp_path):
|
||
"""--dry-run registers the job without any herdr; K x N shows up as idle."""
|
||
reg = tmp_path / "jobs"
|
||
reg.mkdir()
|
||
res = subprocess.run(
|
||
["bash", str(WRAPPER), "submit", "--agent", "human", "--prompt", "contract", "--dry-run",
|
||
"--registry-dir", str(reg), "--workdir", str(tmp_path),
|
||
"--heartbeat", "7", "--heartbeat-misses", "3", "--ack-timeout", "11"],
|
||
capture_output=True, text=True, env=_env(tmp_path, UNREACHABLE), cwd=str(tmp_path), timeout=60)
|
||
assert res.returncode == 0, res.stderr
|
||
jid = _wait_job_id(reg)
|
||
rec = _job(reg, jid)
|
||
assert (rec["heartbeat_sec"], rec["heartbeat_misses"], rec["ack_timeout_sec"], rec["idle_timeout_sec"]) == (7, 3, 11, 21)
|
||
assert "--event started --heartbeat 7" in res.stdout, "the injected start command must arm the timer"
|
||
assert "within 11 s" in res.stdout
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. the automatic heartbeat (no broker needed: events land on disk first)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_started_with_heartbeat_beats_on_disk_until_the_terminal_event(tmp_path):
|
||
reg = tmp_path / "jobs"
|
||
env = _env(tmp_path, UNREACHABLE)
|
||
jid = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "register",
|
||
"--prompt", "hb", "--heartbeat-sec", "1", "--heartbeat-misses", "3"],
|
||
capture_output=True, text=True, env=env, cwd=str(tmp_path)).stdout.strip()
|
||
assert jid
|
||
try:
|
||
res = _publish(reg, jid, "started", env, "--heartbeat", "1", "--attempts", "1")
|
||
assert res.returncode == 2, "the broker is unreachable, the event is still recorded locally"
|
||
pid = _timer_pid(reg, jid)
|
||
assert pid and _alive(pid), "a detached timer must be running"
|
||
time.sleep(4.5)
|
||
beats = [e for e in _events(reg, jid) if e["event"] == "progress"]
|
||
assert len(beats) >= 3, [e["event"] for e in _events(reg, jid)]
|
||
assert all(e["detail"] == "heartbeat" for e in beats)
|
||
assert all(e["data"].get("dispatch_token") == _job(reg, jid)["dispatch_token"] for e in beats), \
|
||
"heartbeats carry the dispatch identity"
|
||
seqs = [e["seq"] for e in _events(reg, jid)]
|
||
assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs)
|
||
|
||
res = _publish(reg, jid, "completed", env, "--attempts", "1")
|
||
assert res.returncode == 2
|
||
deadline = time.time() + 5
|
||
while time.time() < deadline and _alive(pid):
|
||
time.sleep(0.1)
|
||
assert not _alive(pid), "the terminal event must stop the timer"
|
||
assert _timer_pid(reg, jid) is None
|
||
assert _events(reg, jid)[-1]["event"] == "completed", "no heartbeat may trail the terminal event"
|
||
assert _job(reg, jid)["status"] == "completed"
|
||
finally:
|
||
_kill_timer(reg, jid)
|
||
|
||
|
||
def test_heartbeat_timer_stops_on_its_own_when_the_job_is_redispatched(tmp_path):
|
||
reg = tmp_path / "jobs"
|
||
env = _env(tmp_path, UNREACHABLE)
|
||
jid = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "register",
|
||
"--prompt", "hb"], capture_output=True, text=True, env=env, cwd=str(tmp_path)).stdout.strip()
|
||
try:
|
||
_publish(reg, jid, "started", env, "--heartbeat", "1", "--attempts", "1")
|
||
pid = _timer_pid(reg, jid)
|
||
assert _alive(pid)
|
||
# An operator moves the job to the next attempt: the old attempt's timer must not keep beating.
|
||
_publish(reg, jid, "completed", env, "--attempts", "1")
|
||
subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "update", "--job", jid,
|
||
"--status", "pending", "--attempt-id", "2"], capture_output=True, text=True, env=env, cwd=str(tmp_path))
|
||
deadline = time.time() + 5
|
||
while time.time() < deadline and _alive(pid):
|
||
time.sleep(0.1)
|
||
assert not _alive(pid)
|
||
finally:
|
||
_kill_timer(reg, jid)
|
||
|
||
|
||
def test_heartbeat_stops_when_the_watched_agent_process_dies(tmp_path):
|
||
"""Review B2: the timer follows the agent's life. The 'agent' here is a
|
||
shell that runs the start command with --watch-pid $$ and then idles;
|
||
killing that shell must stop the heartbeats within about one period."""
|
||
reg = tmp_path / "jobs"
|
||
env = _env(tmp_path, UNREACHABLE)
|
||
jid = subprocess.run([sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg), "register",
|
||
"--prompt", "hb"], capture_output=True, text=True, env=env, cwd=str(tmp_path)).stdout.strip()
|
||
agent = subprocess.Popen(
|
||
["bash", "-c", f'"{sys.executable}" "{PUBLISH_PY}" --registry-dir "{reg}" --job {jid} --event started '
|
||
f'--heartbeat 1 --attempts 1 --watch-pid $$ >/dev/null 2>&1; sleep 120'],
|
||
cwd=str(tmp_path), env=env)
|
||
try:
|
||
deadline = time.time() + 15
|
||
while time.time() < deadline and not _timer_pid(reg, jid):
|
||
time.sleep(0.1)
|
||
pid = _timer_pid(reg, jid)
|
||
assert pid and _alive(pid)
|
||
time.sleep(2.5)
|
||
before = len([e for e in _events(reg, jid) if e["event"] == "progress"])
|
||
assert before >= 2
|
||
agent.kill() # the agent crashes
|
||
agent.wait(timeout=10)
|
||
deadline = time.time() + 6
|
||
while time.time() < deadline and _alive(pid):
|
||
time.sleep(0.1)
|
||
assert not _alive(pid), "timer must stop once the watched agent is gone"
|
||
time.sleep(1.5)
|
||
after = len([e for e in _events(reg, jid) if e["event"] == "progress"])
|
||
assert after <= before + 2, f"heartbeats kept coming after the agent died: {before} -> {after}"
|
||
assert _job(reg, jid)["status"] == "running", "the timer stops; the orchestrator's K x N closes the job"
|
||
finally:
|
||
if agent.poll() is None:
|
||
agent.kill()
|
||
_kill_timer(reg, jid)
|
||
|
||
|
||
def test_stop_timer_never_signals_a_recycled_pid(tmp_path, monkeypatch):
|
||
"""A stale pidfile (timer crashed) may name a pid that now belongs to
|
||
something else; stop_timer must only signal a heartbeat timer of this job."""
|
||
from lib_py.delegation import heartbeat
|
||
|
||
reg = tmp_path / "jobs"
|
||
reg.mkdir()
|
||
bystander = subprocess.Popen(["sleep", "30"])
|
||
try:
|
||
(reg / "job1.heartbeat.pid").write_text(str(bystander.pid))
|
||
assert heartbeat.running_timer("job1", str(reg)) == bystander.pid
|
||
assert heartbeat.stop_timer("job1", str(reg)) is False
|
||
assert bystander.poll() is None, "the bystander must not be killed"
|
||
assert not (reg / "job1.heartbeat.pid").exists(), "the stale pidfile is removed"
|
||
finally:
|
||
bystander.kill()
|
||
|
||
|
||
def test_heartbeat_is_refused_on_a_terminal_event(tmp_path, monkeypatch):
|
||
from lib_py.delegation import publish_event as pe, registry
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
|
||
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
|
||
reg = tmp_path / "jobs"
|
||
jid = registry.register_job("hb", registry_dir=str(reg))
|
||
assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--heartbeat", "5"]) == 1
|
||
assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "progress", "--heartbeat", "0"]) == 1
|
||
assert _events(reg, jid) == [], "a refused call consumes no seq and records nothing"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. wire durability: retained state next to every event, seq gaps detected
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_every_event_carries_a_signed_retained_state_message(tmp_path, monkeypatch):
|
||
from lib_py.delegation import mqtt_common, publish_event as pe, registry
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
|
||
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
|
||
reg = tmp_path / "jobs"
|
||
jid = registry.register_job("state", registry_dir=str(reg), auth_token="tok-" + "a" * 20)
|
||
sent = []
|
||
|
||
def fake_publish(config, topic, body, retain, extra=()):
|
||
sent.append((topic, json.loads(body), retain, [(t, json.loads(b), r) for t, b, r in extra]))
|
||
|
||
monkeypatch.setattr(pe, "_publish_once", fake_publish)
|
||
tok = _job(reg, jid)["dispatch_token"]
|
||
assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "started", "--expect-dispatch-token", tok]) == 0
|
||
assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS",
|
||
"--expect-dispatch-token", tok]) == 0
|
||
prefix = _job(reg, jid)["topic_prefix"]
|
||
assert [s[0] for s in sent] == [f"{prefix}/events"] * 2
|
||
for (topic, ev, retain, extra), status in zip(sent, ("running", "completed")):
|
||
assert len(extra) == 1
|
||
st_topic, st, st_retain = extra[0]
|
||
assert st_topic == f"{prefix}/state" and st_retain is True
|
||
assert st["event"] == "state" and st["status"] == status and st["last_event"] == ev["event"]
|
||
assert st["seq"] == ev["seq"] and st["job_id"] == jid
|
||
assert mqtt_common.verify_hmac(st, "tok-" + "a" * 20), "state must be signed with the job token"
|
||
assert st["data"]["dispatch_token"] == ev["data"]["dispatch_token"]
|
||
assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "progress", "--no-state",
|
||
"--expect-dispatch-token", tok]) == 0
|
||
assert len(sent) == 3 and sent[-1][3] == [], "--no-state publishes the event alone"
|
||
|
||
|
||
class _Msg:
|
||
def __init__(self, topic, payload):
|
||
self.topic = topic
|
||
self.payload = json.dumps(payload).encode()
|
||
|
||
|
||
def _signed(job_id, seq, event, token, **fields):
|
||
from lib_py.delegation import publish_event as pe
|
||
p = pe.build_payload(job_id, seq, event, "", {"dispatch_token": "d1", "attempt_id": 1}, token)
|
||
if fields:
|
||
# re-sign after adding state fields, as build_state_payload does
|
||
import hashlib, hmac
|
||
p.update(fields)
|
||
p["data"].pop("hmac_sig", None)
|
||
sp = {k: v for k, v in p.items() if k != "data"}
|
||
sp["data"] = dict(p["data"])
|
||
p["data"]["hmac_sig"] = hmac.new(token.encode(), json.dumps(sp, sort_keys=True, separators=(",", ":")).encode(),
|
||
hashlib.sha256).hexdigest()
|
||
return p
|
||
|
||
|
||
def test_subscriber_records_seq_gaps_and_recovers_terminal_from_state(tmp_path, monkeypatch):
|
||
from lib_py.delegation import job_subscriber as js, mqtt_common
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
|
||
token = "t" * 24
|
||
w = js._Watcher({"j1"}, {"j1": token}, {"j1": 0}, {"j1": "d1"}, {"j1": 1})
|
||
w.on_message(None, None, _Msg("x/events", _signed("j1", 1, "started", token)))
|
||
w.on_message(None, None, _Msg("x/events", _signed("j1", 2, "progress", token)))
|
||
w.on_message(None, None, _Msg("x/events", _signed("j1", 5, "progress", token))) # 3 and 4 were lost
|
||
audit = [e for e in mqtt_common.iter_logged_events("j1", mqtt_common.get_logs_dir()) if e["event"] == "seq_gap"]
|
||
assert len(audit) == 1 and (audit[0]["expected_seq"], audit[0]["seq"], audit[0]["via"]) == (3, 5, "events")
|
||
assert w.events.qsize() == 3
|
||
|
||
# a retained state after a reconnect: newer seq than anything seen, terminal status
|
||
st = _signed("j1", 8, "state", token, status="completed", last_event="completed")
|
||
w.on_message(None, None, _Msg("x/state", st))
|
||
assert w.events.qsize() == 4
|
||
topic, payload = list(w.events.queue)[-1]
|
||
assert topic == "x/state" and payload["event"] == "state" and payload["status"] == "completed"
|
||
gaps = [e for e in mqtt_common.iter_logged_events("j1", mqtt_common.get_logs_dir()) if e["event"] == "seq_gap"]
|
||
assert len(gaps) == 2 and gaps[-1]["via"] == "state"
|
||
# a forged state (wrong token) is dropped like a forged event
|
||
bad = _signed("j1", 9, "state", "wrong", status="error", last_event="error")
|
||
w.on_message(None, None, _Msg("x/state", bad))
|
||
assert w.events.qsize() == 4
|
||
# a state for a stale attempt (other dispatch token) is dropped
|
||
stale = _signed("j1", 10, "state", token, status="error", last_event="error")
|
||
stale["data"]["dispatch_token"] = "d0"
|
||
w.on_message(None, None, _Msg("x/state", stale))
|
||
assert w.events.qsize() == 4
|
||
|
||
|
||
def test_subscriber_subscribes_to_the_state_topic(tmp_path):
|
||
src = (SKILLS / "lib_py" / "delegation" / "job_subscriber.py").read_text()
|
||
assert 'state_topics.append(f"{prefix}/state")' in src
|
||
assert "for topic in subscribed_topics + state_topics:" in src
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. delivery: `started` decides, screen signals do not
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _codex_like_mock(pane_before: str, pane_after: str, code: str = "timeout") -> str:
|
||
return f'''
|
||
_pane_quiescent() {{ return 0; }}
|
||
_pane_dialog_open() {{ return 1; }}
|
||
FLAG="$WORKSPACE_ROOT/.prompt-submitted"
|
||
rm -f "$FLAG"
|
||
_sks_herdr() {{
|
||
case "${{1:-}}" in
|
||
agent)
|
||
case "${{2:-}}" in
|
||
prompt) touch "$FLAG"; echo '{{"error":{{"code":"{code}","message":"x"}}}}' >&2; return 1 ;;
|
||
get) echo '{{"result":{{"agent":{{"agent_status":"idle"}}}}}}'; return 0 ;;
|
||
esac ;;
|
||
capture-pane)
|
||
if [ -e "$FLAG" ]; then printf '%s\\n' "{pane_after}"; else printf '%s\\n' "{pane_before}"; fi
|
||
return 0 ;;
|
||
paste-buffer|set-buffer|delete-buffer|send-keys)
|
||
echo "PASTE_PATH_USED: $*" >&2; return 0 ;;
|
||
esac
|
||
return 0
|
||
}}
|
||
'''
|
||
|
||
|
||
def _run_lib_sh(snippet: str, cwd: Path, env=None, timeout=60):
|
||
return subprocess.run(["bash", "-c", f'source "{LIB_SH}" >/dev/null 2>&1; {snippet}'],
|
||
capture_output=True, text=True, cwd=str(cwd), env=_env(cwd, env), timeout=timeout)
|
||
|
||
|
||
def test_send_keys_safe_is_confirmed_by_the_job_started_event(tmp_path, monkeypatch):
|
||
"""herdr says `timeout`, agent_status stays idle, the pane never changes:
|
||
the only thing that confirms delivery is the job's own `started`."""
|
||
from lib_py.delegation import registry
|
||
|
||
monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
|
||
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
|
||
reg = tmp_path / "jobs"
|
||
jid = registry.register_job("ack", registry_dir=str(reg))
|
||
env = _env(tmp_path, {**UNREACHABLE, "SKS_DELIVERY_GRACE_S": "14", "SKS_REGISTRY_DIR": str(reg)})
|
||
|
||
def agent_starts_later():
|
||
time.sleep(2)
|
||
subprocess.run([sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg), "--job", jid,
|
||
"--event", "started", "--attempts", "1"], capture_output=True, env=env, cwd=str(tmp_path))
|
||
|
||
t = threading.Thread(target=agent_starts_later, daemon=True)
|
||
t.start()
|
||
started = time.time()
|
||
res = subprocess.run(["bash", "-c", f'source "{LIB_SH}" >/dev/null 2>&1; '
|
||
+ _codex_like_mock("idle", "idle") + f'send_keys_safe "reviewer-codex-01" "prompt" "{jid}"'],
|
||
capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60)
|
||
t.join()
|
||
assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"
|
||
assert time.time() - started < 11, "confirmed once started landed, not at the end of the 14 s grace"
|
||
assert "PASTE_PATH_USED" not in res.stderr
|
||
|
||
|
||
def test_send_keys_safe_ignores_pane_change_and_busy_token(tmp_path):
|
||
"""The screen moving on, or showing a busy token, is not delivery any more:
|
||
with no `started` the accepted submission still fails closed (rc 4)."""
|
||
for after in ("Working (0s • esc to interrupt)\\n› Ask Codex", "• OK\\n› Ask Codex"):
|
||
res = _run_lib_sh(_codex_like_mock("idle", after) + 'send_keys_safe "reviewer-codex-01" "prompt" "adhoc"',
|
||
tmp_path, env={**UNREACHABLE, "SKS_DELIVERY_GRACE_S": "2"})
|
||
assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
|
||
assert "delivery unknown/failed" in res.stderr
|
||
assert "PASTE_PATH_USED" not in res.stderr
|
||
|
||
|
||
def test_wait_started_only_accepts_this_attempts_started(tmp_path):
|
||
"""Review R1: a loop job keeps one events.log across iterations. A
|
||
`started` left by the previous attempt (other dispatch token) must not
|
||
acknowledge the new injection; one with the current token must."""
|
||
reg = tmp_path / "jobs"
|
||
reg.mkdir()
|
||
jid = "abcd1234"
|
||
(reg / f"{jid}.json").write_text(json.dumps({"job_id": jid, "status": "pending", "dispatch_token": "new-token"}))
|
||
log = reg / f"{jid}.events.log"
|
||
log.write_text(json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": "old-token"}}) + "\n")
|
||
env = _env(tmp_path, {"MAM_DELEGATE_JOB_NO_MAIN": "1"})
|
||
|
||
def wait_started(extra: str) -> subprocess.CompletedProcess:
|
||
# the wrapper sets -e: capture the helper's rc instead of dying on it
|
||
script = (f'source "{WRAPPER}"; set +e; REGISTRY_DIR="{reg}"; PY="{sys.executable}"; TIMEOUT=60; START_TIME=$(date +%s); '
|
||
f'_disp_tok="new-token"; {extra} _wait_started {jid} 3; echo "rc=$?"')
|
||
return subprocess.run(["bash", "-c", script], capture_output=True, text=True, env=env, cwd=str(tmp_path), timeout=60)
|
||
|
||
res = wait_started("")
|
||
assert "rc=1" in res.stdout, f"old attempt's started must not count\n{res.stdout}\n{res.stderr}"
|
||
with open(log, "a") as fh:
|
||
fh.write(json.dumps({"event": "started", "seq": 2, "data": {"dispatch_token": "new-token"}}) + "\n")
|
||
res = wait_started("")
|
||
assert "rc=0" in res.stdout and "recorded 'started' in events.log" in res.stderr, res.stderr
|
||
|
||
|
||
def test_remote_runner_ignores_a_resent_dispatch_of_a_finished_attempt(tmp_path, monkeypatch):
|
||
"""Review R2: a retained dispatch delivered again (runner restart, broker
|
||
replay) must not re-run an attempt that already reached its terminal."""
|
||
from lib_py.delegation import remote, registry
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
monkeypatch.setenv("MAM_ENV_FILE", str(tmp_path / ".mam.env.absent"))
|
||
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
|
||
reg = tmp_path / "jobs"
|
||
d = {"job_id": "feed1234", "brief": "b", "dispatch_token": "tok-1", "attempt_id": 1, "auth_token": None,
|
||
"timeout_sec": 30, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5}
|
||
assert remote.mirror_job(d, str(reg), "s") == "feed1234"
|
||
assert remote.mirror_job(d, str(reg), "s") == "feed1234", "same attempt, still open: idempotent"
|
||
from lib_py.delegation import mqtt_common
|
||
mqtt_common.update_job_status("feed1234", str(reg), _source="registry_cli", status="completed")
|
||
assert remote.mirror_job(d, str(reg), "s") is None, "finished attempt: the re-sent dispatch is ignored"
|
||
assert remote.handle_dispatch(d, str(reg), "s", "echo should-not-run") == 0
|
||
d2 = dict(d, dispatch_token="tok-2", attempt_id=2)
|
||
assert remote.mirror_job(d2, str(reg), "s") == "feed1234", "a new attempt is mirrored afresh"
|
||
assert _job(reg, "feed1234")["dispatch_token"] == "tok-2" and _job(reg, "feed1234")["status"] == "pending"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. wrapper acceptance criteria (broker)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _headless_agent(state_file: Path, name: str, agent: str = "codex") -> None:
|
||
state = json.loads(state_file.read_text())
|
||
# agent_status pinned idle, prompt `timeout`, empty pane: nothing to scrape
|
||
state["agents"][name] = {"agent": agent, "agent_status": "idle", "prompt_wait_result": "timeout", "buffer": ""}
|
||
state_file.write_text(json.dumps(state, indent=2))
|
||
|
||
|
||
def _submit(tmp_path: Path, reg: Path, session: str, agent: str, extra_args, env_extra=None, timeout_s=60):
|
||
env = _env(tmp_path, {"SKS_IDLE_TIMEOUT_MS": "500", "SKS_EMPTY_GIVEUP": "1", **(env_extra or {})})
|
||
proc = subprocess.Popen(
|
||
["bash", str(WRAPPER), "submit", "--agent", agent, "--agent-session", session, "--prompt", "issue 18",
|
||
"--registry-dir", str(reg), "--workdir", str(tmp_path), "--timeout", str(timeout_s), *extra_args],
|
||
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||
return proc, env
|
||
|
||
|
||
def _prompt_calls(state_file: Path) -> int:
|
||
"""`agent prompt` invocations; the isolated wrapper prefixes `--session <name>`."""
|
||
n = 0
|
||
for c in json.loads(state_file.read_text()).get("calls", []):
|
||
c = list(c)
|
||
if len(c) >= 2 and c[0] == "--session":
|
||
c = c[2:]
|
||
if len(c) >= 2 and c[0] == "agent" and c[1] == "prompt":
|
||
n += 1
|
||
return n
|
||
|
||
|
||
def test_acceptance1_headless_agent_survives_on_heartbeats_alone(tmp_path, mock_herdr):
|
||
"""herdr idle throughout, empty pane: `started --heartbeat 2` keeps the job
|
||
alive past several K x N windows (2 x 2 = 4 s) and `completed` yields rc 0."""
|
||
reg = tmp_path / "jobs"
|
||
reg.mkdir()
|
||
_headless_agent(Path(mock_herdr), "headless-codex")
|
||
proc, env = _submit(tmp_path, reg, "herdr:headless-codex", "codex",
|
||
["--heartbeat", "2", "--heartbeat-misses", "2", "--ack-timeout", "40"], timeout_s=90)
|
||
jid = _wait_job_id(reg)
|
||
try:
|
||
time.sleep(1)
|
||
assert _publish(reg, jid, "started", env, "--heartbeat", "2").returncode == 0
|
||
time.sleep(16) # four idle windows without any agent-published event
|
||
assert _job(reg, jid)["status"] == "running", "heartbeats alone must keep the job open"
|
||
assert _publish(reg, jid, "completed", env, "--detail", "PASS").returncode == 0
|
||
stdout, stderr = proc.communicate(timeout=120)
|
||
finally:
|
||
_kill_timer(reg, jid)
|
||
if proc.poll() is None:
|
||
proc.kill()
|
||
assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}"
|
||
assert _job(reg, jid)["status"] == "completed"
|
||
assert _prompt_calls(Path(mock_herdr)) == 1
|
||
beats = [e for e in _events(reg, jid) if e["event"] == "progress"]
|
||
assert len(beats) >= 5, len(beats)
|
||
brief = (reg / jid / "brief.md").read_text()
|
||
assert "heartbeat every 2 s" in brief and "after 2 missed heartbeats" in brief and "within 40 s" in brief
|
||
|
||
|
||
def test_acceptance2_no_started_within_T_fails_closed_without_reinjection(tmp_path, mock_herdr):
|
||
reg = tmp_path / "jobs"
|
||
reg.mkdir()
|
||
_headless_agent(Path(mock_herdr), "mute-codex")
|
||
started = time.time()
|
||
proc, env = _submit(tmp_path, reg, "herdr:mute-codex", "codex",
|
||
["--ack-timeout", "4", "--heartbeat", "2", "--heartbeat-misses", "2"],
|
||
env_extra={"SKS_DELIVERY_GRACE_S": "1"}, timeout_s=60)
|
||
stdout, stderr = proc.communicate(timeout=120)
|
||
jid = _wait_job_id(reg)
|
||
assert proc.returncode == 1, f"rc={proc.returncode}\n{stdout}\n{stderr}"
|
||
assert "no 'started' within 4s" in stderr
|
||
assert _job(reg, jid)["status"] == "error"
|
||
assert _prompt_calls(Path(mock_herdr)) == 1, "single injection: never re-injected"
|
||
assert time.time() - started < 45, "the ack wait is bounded by T, not the wall clock"
|
||
ev = [e for e in _events(reg, jid) if e["event"] == "error"]
|
||
assert ev and ev[-1]["detail"] == "agent_blocked"
|
||
|
||
|
||
def test_acceptance3_lost_heartbeats_close_the_job_and_a_late_completed_is_refused(tmp_path, mock_herdr):
|
||
reg = tmp_path / "jobs"
|
||
reg.mkdir()
|
||
_headless_agent(Path(mock_herdr), "dying-codex")
|
||
proc, env = _submit(tmp_path, reg, "herdr:dying-codex", "codex",
|
||
["--heartbeat", "1", "--heartbeat-misses", "3", "--ack-timeout", "40"], timeout_s=60)
|
||
jid = _wait_job_id(reg)
|
||
t0 = None
|
||
agent = None
|
||
try:
|
||
time.sleep(1)
|
||
tok = _job(reg, jid)["dispatch_token"]
|
||
# the agent: a shell that runs the injected start command (timer bound
|
||
# to it via --watch-pid $$) and then keeps "working" silently
|
||
agent = subprocess.Popen(
|
||
["bash", "-c", f'"{sys.executable}" "{PUBLISH_PY}" --registry-dir "{reg}" --job {jid} --event started '
|
||
f'--expect-dispatch-token {tok} --heartbeat 1 --watch-pid $$; sleep 300'],
|
||
cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
deadline = time.time() + 30
|
||
while time.time() < deadline and not _timer_pid(reg, jid):
|
||
time.sleep(0.2)
|
||
assert _timer_pid(reg, jid), "the start command must have armed the timer"
|
||
time.sleep(4)
|
||
agent.kill() # the agent dies: its timer stops, no more heartbeats
|
||
agent.wait(timeout=10)
|
||
t0 = time.time()
|
||
stdout, stderr = proc.communicate(timeout=120)
|
||
finally:
|
||
if agent is not None and agent.poll() is None:
|
||
agent.kill()
|
||
_kill_timer(reg, jid)
|
||
if proc.poll() is None:
|
||
proc.kill()
|
||
assert proc.returncode == 2, f"rc={proc.returncode}\n{stdout}\n{stderr}"
|
||
assert _job(reg, jid)["status"] == "error"
|
||
assert "herdr/pane says busy=" in stderr, "screen state is logged as a diagnostic only"
|
||
# K x N = 3 s idle + one grace window: closed long before the 60 s wall clock
|
||
assert time.time() - t0 < 30, f"took {time.time() - t0:.0f}s"
|
||
late = _publish(reg, jid, "completed", env, "--detail", "too late")
|
||
assert late.returncode in (0, 2, 4)
|
||
assert _job(reg, jid)["status"] == "error", "terminal immutability (CAS) refuses the late completed"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 6. remote transport (broker): no shared disk, no pane, only the wire
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path):
|
||
orch = tmp_path / "orchestrator"
|
||
remote = tmp_path / "remote-host"
|
||
for d in (orch, remote):
|
||
(d / "jobs").mkdir(parents=True)
|
||
secret = "issue18-shared-secret"
|
||
runner_env = _env(remote, {"MAM_DISPATCH_SECRET": secret, "MAM_JOBS_DIR": str(remote / "jobs")})
|
||
runner_log = remote / "runner.log"
|
||
with open(runner_log, "wb") as log:
|
||
runner = subprocess.Popen(
|
||
[sys.executable, str(REMOTE_PY), "run", "--session", "sim-worker", "--registry-dir", str(remote / "jobs"),
|
||
"--exec", 'echo "remote work done for $JOB_ID"', "--once", "--timeout", "150"],
|
||
cwd=str(remote), env=runner_env, stdout=log, stderr=subprocess.STDOUT)
|
||
try:
|
||
deadline = time.time() + 45
|
||
while time.time() < deadline and "SUBSCRIBED" not in runner_log.read_text(errors="replace"):
|
||
time.sleep(0.2)
|
||
if "SUBSCRIBED" not in runner_log.read_text(errors="replace"):
|
||
pytest.skip("broker unreachable: " + runner_log.read_text(errors="replace")[-300:])
|
||
|
||
reg = orch / "jobs"
|
||
proc, env = _submit(orch, reg, "remote:sim-worker", "remote",
|
||
["--heartbeat", "2", "--heartbeat-misses", "3", "--ack-timeout", "60"],
|
||
env_extra={"MAM_DISPATCH_SECRET": secret}, timeout_s=120)
|
||
stdout, stderr = proc.communicate(timeout=180)
|
||
jid = _wait_job_id(reg)
|
||
assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
|
||
assert "subscriber received started" in stderr or "advanced to 'running'" in stderr
|
||
assert _job(reg, jid)["status"] == "completed", \
|
||
f"the subscriber mirrors the remote terminal into the registry\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
|
||
assert _job(reg, jid)["transport"] == "remote"
|
||
# The runner's own status sync follows its wire publish, so the
|
||
# orchestrator can observe `completed` a moment before the remote
|
||
# record says so: wait for the runner before reading its registry.
|
||
runner.wait(timeout=60)
|
||
# the remote side kept its own record with the orchestrator's identity
|
||
mirror = _job(remote / "jobs", jid)
|
||
assert mirror["status"] == "completed" and mirror["transport"] == "remote"
|
||
assert mirror["dispatch_token"] == _job(reg, jid)["dispatch_token"]
|
||
assert mirror["auth_token"] == _job(reg, jid)["auth_token"]
|
||
assert (remote / "jobs" / jid / "brief.md").read_text().startswith("issue 18")
|
||
# the agent command's last stdout line became the completed detail
|
||
assert "remote work done for " + jid in stdout, stdout
|
||
assert runner.returncode == 0
|
||
assert "살아있는 에이전트 세션" not in stdout, "no pane injection path for a remote session"
|
||
finally:
|
||
if runner.poll() is None:
|
||
runner.kill()
|
||
|
||
|
||
def test_remote_runner_refuses_unsigned_dispatches(tmp_path, monkeypatch):
|
||
from lib_py.delegation import remote
|
||
|
||
d = {"kind": "dispatch", "job_id": "j", "session": "s", "schema_version": 1}
|
||
assert not remote.verify_dispatch(d, "secret")
|
||
signed = remote.sign_dispatch(d, "secret")
|
||
assert remote.verify_dispatch(signed, "secret")
|
||
assert not remote.verify_dispatch(signed, "other")
|
||
assert not remote.verify_dispatch(dict(signed, session="hijacked"), "secret")
|
||
assert not remote.verify_dispatch(signed, None)
|
||
monkeypatch.delenv("MAM_DISPATCH_SECRET", raising=False)
|
||
assert remote.run_runner("s", str(tmp_path), allow_unsigned=False) == 1, "a runner without a secret refuses to run"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 7. loop runner: contract knobs pass through; remote rows get remote:<name>
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_loop_runner_passes_the_contract_and_remote_label(tmp_path, monkeypatch):
|
||
from lib_py.loop import runner as loop_runner
|
||
|
||
monkeypatch.chdir(tmp_path)
|
||
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
||
monkeypatch.setenv("MAM_STATE_JSON", json.dumps({"herdr_sessions": [
|
||
{"name": "creator-claude-01", "agent": "claude", "role": "creator", "status": "running"},
|
||
{"name": "reviewer-far-01", "agent": "claude", "role": "reviewer", "status": "running", "transport": "remote"},
|
||
]}))
|
||
monkeypatch.setenv("MAM_HEARTBEAT_SEC", "20")
|
||
monkeypatch.setenv("MAM_HEARTBEAT_MISSES", "3")
|
||
monkeypatch.setenv("MAM_ACK_TIMEOUT_S", "45")
|
||
r = loop_runner.LoopRunner(["--creator", "creator-claude-01", "--task", "t"])
|
||
r.parse_args()
|
||
assert r.session_label("creator-claude-01") == "herdr:creator-claude-01"
|
||
assert r.session_label("reviewer-far-01") == "remote:reviewer-far-01"
|
||
assert r.contract_args() == ["--heartbeat", "20", "--heartbeat-misses", "3", "--ack-timeout", "45"]
|
||
seen = []
|
||
|
||
def fake_run(cmd, **_kw):
|
||
seen.append(cmd)
|
||
class P: # noqa: D401 - tiny stand-in
|
||
returncode = 0
|
||
stdout = "registered job: deadbeef\n"
|
||
stderr = ""
|
||
return P()
|
||
|
||
monkeypatch.setattr(loop_runner.subprocess, "run", fake_run)
|
||
assert r.delegate_job("reviewer-far-01", "claude", "Reviewer", "review") == "deadbeef"
|
||
cmd = seen[0]
|
||
assert cmd[cmd.index("--agent-session") + 1] == "remote:reviewer-far-01"
|
||
assert cmd[-6:] == ["--heartbeat", "20", "--heartbeat-misses", "3", "--ack-timeout", "45"]
|