GodopuandClaude Opus 5
7bc3e1a288
The remote runner subscribes to every job's retained dispatch topic on the public broker and serves the first dispatch signed for its session. With the fixed session name `sim-worker` and a fixed secret, a stale retained dispatch left by an earlier run verified fine and was served instead of this run's job, which then timed out: in the observed failure the runner logged `published started/completed ... job=5eb1abcf` and `DONE rc=0` while the orchestrator waited on job 5ee3fbcf with nothing received. Each run now uses a random session name and secret, so stale dispatches cannot match. Also, per the plan: the no-event window is 10 x 2 s instead of 3 x 2 s, and a failure skips only when the runner provably published `started` for the orchestrator's own job, the orchestrator subscribed to that job's topic, and received and dropped nothing for it (broker non-delivery). Every other failure still fails. The runner's own `started --heartbeat` already covers the window. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017R3Eg7MEpSWnm1oB9ZPP6G
1298 lines
63 KiB
Python
1298 lines
63 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 re
|
||
import os
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import uuid
|
||
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, 180)
|
||
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_concurrent_start_timer_calls_leave_exactly_one_tracked_owner(tmp_path, monkeypatch):
|
||
"""Review round 4 (codex R3): two `start_timer` calls for the same
|
||
job/attempt at once - both see no timer, both spawn, one pidfile - left an
|
||
untracked timer that could publish after the terminal event. The lifecycle
|
||
lock serialises observe/drain/spawn/publish-pid, so exactly one timer
|
||
survives, the pidfile names it, and the terminal stop drains it."""
|
||
from lib_py.delegation import heartbeat, 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"))
|
||
monkeypatch.setenv("MQTT_BROKER", "127.0.0.1")
|
||
monkeypatch.setenv("MQTT_PORT", "9")
|
||
reg = tmp_path / "jobs"
|
||
jid = registry.register_job("race", registry_dir=str(reg))
|
||
tok = _job(reg, jid)["dispatch_token"]
|
||
pids = []
|
||
errors = []
|
||
|
||
def start():
|
||
try:
|
||
pids.append(heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0))
|
||
except Exception as exc: # pragma: no cover - surfaced below
|
||
errors.append(exc)
|
||
|
||
threads = [threading.Thread(target=start) for _ in range(2)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join(timeout=60)
|
||
try:
|
||
assert not errors, errors
|
||
assert len(pids) == 2 and all(pids)
|
||
time.sleep(1.0)
|
||
alive = [p for p in pids if _alive(p)]
|
||
assert len(alive) == 1, f"exactly one timer may survive two concurrent starts, alive={alive} pids={pids}"
|
||
assert _timer_pid(reg, jid) == alive[0], "the pidfile names the surviving owner"
|
||
# the terminal path drains the tracked owner and nothing else is left
|
||
assert heartbeat.stop_timer(jid, str(reg), drain=5.0) is True
|
||
assert not any(_alive(p) for p in pids)
|
||
assert _timer_pid(reg, jid) is None
|
||
finally:
|
||
for p in pids:
|
||
try:
|
||
os.kill(p, signal.SIGKILL)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def test_start_timer_after_the_terminal_barrier_does_not_arm(tmp_path, monkeypatch):
|
||
"""A `started --heartbeat` that loses the race against the terminal
|
||
publish must not leave a timer behind."""
|
||
from lib_py.delegation import heartbeat, 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("late-start", registry_dir=str(reg))
|
||
tok = _job(reg, jid)["dispatch_token"]
|
||
heartbeat.raise_barrier(jid, str(reg), dispatch_token=tok, attempt_id=1)
|
||
assert heartbeat.start_timer(jid, str(reg), 1, dispatch_token=tok, attempt_id=1, watch_pid=0) == 0
|
||
assert _timer_pid(reg, jid) is None
|
||
assert not (reg / f"{jid}.heartbeat.pid").exists()
|
||
|
||
|
||
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"
|
||
|
||
|
||
def test_remote_runner_never_rolls_a_newer_attempt_back(tmp_path, monkeypatch):
|
||
"""Review round 2 (codex R1): a validly signed dispatch for attempt 1 that
|
||
arrives after attempt 2 is stale. It must neither replace the record nor
|
||
touch its events.log, and it is never run; the same attempt under another
|
||
token is refused as a conflict."""
|
||
from lib_py.delegation import remote
|
||
|
||
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 = uuid.uuid4().hex[:8]
|
||
a1 = {"job_id": jid, "brief": "attempt one", "dispatch_token": "tok-" + uuid.uuid4().hex, "attempt_id": 1,
|
||
"auth_token": None, "timeout_sec": 30, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5}
|
||
b2 = dict(a1, brief="attempt two", dispatch_token="tok-" + uuid.uuid4().hex, attempt_id=2)
|
||
assert remote.mirror_job(a1, str(reg), "s") == jid
|
||
assert remote.mirror_job(b2, str(reg), "s") == jid, "a newer attempt is admitted"
|
||
log = reg / f"{jid}.events.log"
|
||
log.write_text(json.dumps({"event": "started", "seq": 1, "data": {"dispatch_token": b2["dispatch_token"]}}) + "\n")
|
||
record_before, log_before = (reg / f"{jid}.json").read_text(), log.read_text()
|
||
|
||
assert remote.mirror_job(a1, str(reg), "s") is None, "the replayed attempt-1 dispatch is stale"
|
||
assert remote.mirror_job(dict(b2, dispatch_token="tok-" + uuid.uuid4().hex), str(reg), "s") is None, \
|
||
"the same attempt under another token is a conflict"
|
||
assert (reg / f"{jid}.json").read_text() == record_before, "no record rollback"
|
||
assert log.read_text() == log_before, "no events.log rollback"
|
||
ran = tmp_path / "ran"
|
||
assert remote.handle_dispatch(a1, str(reg), "s", f"touch {ran}") == 0
|
||
assert not ran.exists(), "a stale dispatch is never executed"
|
||
rec = _job(reg, jid)
|
||
assert (rec["attempt_id"], rec["dispatch_token"]) == (2, b2["dispatch_token"])
|
||
assert (reg / jid / "brief.md").read_text() == "attempt two"
|
||
|
||
|
||
def test_remote_command_timeout_ends_the_whole_process_group(tmp_path, monkeypatch):
|
||
"""Review round 2 (codex R2): at timeout_sec the runner stops the whole
|
||
command - not only its shell - before it reports the error, so no
|
||
descendant keeps working past the deadline."""
|
||
from lib_py.delegation import remote
|
||
|
||
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"))
|
||
published = []
|
||
monkeypatch.setattr(remote, "_publish", lambda jid, reg, event, detail, d, heartbeat=None:
|
||
published.append((event, detail)) or 0)
|
||
reg = tmp_path / "jobs"
|
||
pid_file, marker = tmp_path / "child.pid", tmp_path / "late.marker"
|
||
child = tmp_path / "child.py"
|
||
child.write_text("import os, time\n"
|
||
f"open({str(pid_file)!r}, 'w').write(str(os.getpid()))\n"
|
||
"time.sleep(2)\n"
|
||
f"open({str(marker)!r}, 'w').write('too late')\n")
|
||
d = {"job_id": uuid.uuid4().hex[:8], "brief": "b", "dispatch_token": "tok-" + uuid.uuid4().hex, "attempt_id": 1,
|
||
"auth_token": None, "timeout_sec": 1, "heartbeat_sec": 5, "heartbeat_misses": 2, "ack_timeout_sec": 5}
|
||
pid = None
|
||
try:
|
||
# `; :` keeps the shell alive as the python child's parent (no exec)
|
||
rc = remote.handle_dispatch(d, str(reg), "s", f'"{sys.executable}" "{child}"; :')
|
||
assert rc == 124
|
||
assert published[-1][0] == "error" and "timeout_sec 1" in published[-1][1], published
|
||
pid = int(pid_file.read_text())
|
||
time.sleep(2.5)
|
||
assert not marker.exists(), "a descendant kept working past the deadline"
|
||
assert not _alive(pid)
|
||
finally:
|
||
if pid and _alive(pid):
|
||
os.kill(pid, signal.SIGKILL)
|
||
|
||
|
||
_PAUSING_TIMER = r'''
|
||
import json, os, sys, time
|
||
from lib_py.delegation import heartbeat as hb, publish_event as pe
|
||
|
||
capture, paused, go = sys.argv[1], sys.argv[2], sys.argv[3]
|
||
|
||
|
||
def fake_publish(config, topic, body, retain, extra=()):
|
||
with open(capture, "a") as fh:
|
||
fh.write(json.dumps({"who": "timer", "event": json.loads(body)["event"],
|
||
"state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")
|
||
|
||
|
||
pe._publish_once = fake_publish
|
||
real_prepare = pe.prepare_event
|
||
|
||
|
||
def paused_prepare(*a, **k):
|
||
open(paused, "w").close()
|
||
deadline = time.time() + 20
|
||
while not os.path.exists(go) and time.time() < deadline:
|
||
time.sleep(0.05)
|
||
return real_prepare(*a, **k)
|
||
|
||
|
||
pe.prepare_event = paused_prepare
|
||
argv = sys.argv[5:] # argv[4] names the module so stop_timer recognises this process as the job's timer
|
||
job, reg = argv[argv.index("--job") + 1], argv[argv.index("--registry-dir") + 1]
|
||
with open(hb.pidfile_path(job, reg), "w") as fh:
|
||
fh.write(str(os.getpid()))
|
||
sys.exit(hb.main(argv))
|
||
'''
|
||
|
||
|
||
def test_no_heartbeat_follows_the_terminal_event_even_when_one_is_in_flight(tmp_path, monkeypatch):
|
||
"""Review round 2 (codex R3): a beat already running when the agent
|
||
publishes `completed` must not publish after it. The timer's first beat is
|
||
held just before it asks for a seq; the terminal publish raises the
|
||
attempt's barrier, and the beat - released only then - is refused under the
|
||
seq lock. Nothing follows `completed` on disk or on the (stubbed) wire, and
|
||
the last retained state stays `completed`."""
|
||
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"))
|
||
monkeypatch.setenv("MAM_HEARTBEAT_DRAIN_S", "8")
|
||
reg = tmp_path / "jobs"
|
||
jid = registry.register_job("barrier", registry_dir=str(reg), auth_token="tok-" + uuid.uuid4().hex)
|
||
tok = _job(reg, jid)["dispatch_token"]
|
||
capture, paused, go = tmp_path / "wire.ndjson", tmp_path / "beat.paused", tmp_path / "beat.go"
|
||
script = tmp_path / "pausing_timer.py"
|
||
script.write_text(_PAUSING_TIMER)
|
||
env = dict(os.environ, PYTHONPATH=str(SKILLS), MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"),
|
||
DELEGATE_JOB_LOGS_DIR=str(tmp_path / "logs"))
|
||
timer = subprocess.Popen([sys.executable, str(script), str(capture), str(paused), str(go),
|
||
"lib_py.delegation.heartbeat", "--job", jid, "--registry-dir", str(reg),
|
||
"--interval", "1", "--expect-dispatch-token", tok],
|
||
cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
try:
|
||
deadline = time.time() + 15
|
||
while time.time() < deadline and not paused.exists():
|
||
time.sleep(0.05)
|
||
assert paused.exists(), "the timer's first beat never reached its publish"
|
||
|
||
def fake_publish(config, topic, body, retain, extra=()):
|
||
with open(capture, "a") as fh:
|
||
fh.write(json.dumps({"who": "terminal", "event": json.loads(body)["event"],
|
||
"state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")
|
||
|
||
monkeypatch.setattr(pe, "_publish_once", fake_publish)
|
||
result = {}
|
||
terminal = threading.Thread(target=lambda: result.update(rc=pe.main(
|
||
["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS",
|
||
"--expect-dispatch-token", tok, "--attempts", "1"])))
|
||
terminal.start()
|
||
barrier = reg / f"{jid}.heartbeat.barrier"
|
||
deadline = time.time() + 10
|
||
while time.time() < deadline and terminal.is_alive() and not barrier.exists():
|
||
time.sleep(0.02)
|
||
go.touch() # release the in-flight beat
|
||
terminal.join(timeout=30)
|
||
assert result.get("rc") == 0
|
||
assert timer.wait(timeout=30) is not None
|
||
wire = [json.loads(l) for l in capture.read_text().splitlines() if l.strip()]
|
||
assert wire and wire[-1]["event"] == "completed", f"something was published after completed: {wire}"
|
||
assert wire[-1]["state"] == ["completed"], wire
|
||
local = [e["event"] for e in _events(reg, jid)]
|
||
assert local and local[-1] == "completed", local
|
||
assert _job(reg, jid)["status"] == "completed"
|
||
finally:
|
||
if timer.poll() is None:
|
||
timer.kill()
|
||
|
||
|
||
_WIRE_HELD_TIMER = r'''
|
||
import json, os, sys, time
|
||
from lib_py.delegation import heartbeat as hb, publish_event as pe
|
||
|
||
capture, paused, go = sys.argv[1], sys.argv[2], sys.argv[3]
|
||
|
||
|
||
def held_publish(config, topic, body, retain, extra=()):
|
||
# the beat was admitted and recorded on disk already: hold it at the wire
|
||
open(paused, "w").close()
|
||
deadline = time.time() + 20
|
||
while not os.path.exists(go) and time.time() < deadline:
|
||
time.sleep(0.05)
|
||
with open(capture, "a") as fh:
|
||
fh.write(json.dumps({"who": "old-timer", "event": json.loads(body)["event"],
|
||
"state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")
|
||
|
||
|
||
pe._publish_once = held_publish
|
||
argv = sys.argv[5:] # argv[4] names the module so stop_timer recognises this process as the job's timer
|
||
job, reg = argv[argv.index("--job") + 1], argv[argv.index("--registry-dir") + 1]
|
||
with open(hb.pidfile_path(job, reg), "w") as fh:
|
||
fh.write(str(os.getpid()))
|
||
sys.exit(hb.main(argv))
|
||
'''
|
||
|
||
|
||
def test_replacing_the_timer_drains_the_old_one_before_the_terminal(tmp_path, monkeypatch):
|
||
"""Review round 3 (codex R3): a timer replaced by a second
|
||
`started --heartbeat` must be gone before it loses the pidfile. Its beat -
|
||
already admitted and on disk, held at the wire boundary - may otherwise be
|
||
published after `completed`, because the terminal publish can only find
|
||
and drain the new timer. After the fix the old timer is dead by the time
|
||
start_timer returns and nothing it held follows the terminal."""
|
||
from lib_py.delegation import heartbeat, 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("replace", registry_dir=str(reg), auth_token="tok-" + uuid.uuid4().hex)
|
||
tok = _job(reg, jid)["dispatch_token"]
|
||
capture, paused, go = tmp_path / "wire.ndjson", tmp_path / "beat.paused", tmp_path / "beat.go"
|
||
|
||
def fake_publish(config, topic, body, retain, extra=()):
|
||
with open(capture, "a") as fh:
|
||
fh.write(json.dumps({"who": "parent", "event": json.loads(body)["event"],
|
||
"state": [json.loads(b).get("status") for _t, b, _r in extra]}) + "\n")
|
||
|
||
monkeypatch.setattr(pe, "_publish_once", fake_publish)
|
||
assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "started",
|
||
"--expect-dispatch-token", tok, "--attempts", "1"]) == 0
|
||
script = tmp_path / "wire_held_timer.py"
|
||
script.write_text(_WIRE_HELD_TIMER)
|
||
env = dict(os.environ, PYTHONPATH=str(SKILLS))
|
||
old = subprocess.Popen([sys.executable, str(script), str(capture), str(paused), str(go),
|
||
"lib_py.delegation.heartbeat", "--job", jid, "--registry-dir", str(reg),
|
||
"--interval", "1", "--expect-dispatch-token", tok],
|
||
cwd=str(tmp_path), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
new_pid = None
|
||
try:
|
||
deadline = time.time() + 15
|
||
while time.time() < deadline and not paused.exists():
|
||
time.sleep(0.05)
|
||
assert paused.exists(), "the old timer's beat never reached the wire boundary"
|
||
|
||
new_pid = heartbeat.start_timer(jid, str(reg), 60, dispatch_token=tok, python=sys.executable, watch_pid=0)
|
||
assert old.poll() is not None, "the replaced timer must be gone before its pidfile is handed over"
|
||
assert _timer_pid(reg, jid) == new_pid
|
||
|
||
assert pe.main(["--registry-dir", str(reg), "--job", jid, "--event", "completed", "--detail", "PASS",
|
||
"--expect-dispatch-token", tok, "--attempts", "1"]) == 0
|
||
go.touch() # would release the held beat if its timer were still alive
|
||
time.sleep(0.5)
|
||
wire = [json.loads(l) for l in capture.read_text().splitlines() if l.strip()]
|
||
assert all(w["who"] != "old-timer" for w in wire), f"the replaced timer published: {wire}"
|
||
assert wire[-1]["event"] == "completed" and wire[-1]["state"] == ["completed"], wire
|
||
assert [e["event"] for e in _events(reg, jid)][-1] == "completed"
|
||
deadline = time.time() + 5
|
||
while time.time() < deadline and _alive(new_pid):
|
||
time.sleep(0.1)
|
||
assert not _alive(new_pid), "the terminal publish stops the replacement timer"
|
||
finally:
|
||
if old.poll() is None:
|
||
old.kill()
|
||
if new_pid and _alive(new_pid):
|
||
os.kill(new_pid, signal.SIGKILL)
|
||
|
||
|
||
def test_an_idle_timer_stops_gracefully_well_before_the_drain_deadline(tmp_path, monkeypatch, caplog):
|
||
"""Review round 5 (codex F1): the stopper holds the lifecycle lock while it
|
||
waits for the timer to exit, so the exiting timer must not wait for that
|
||
lock. An idle timer (nothing in flight) ends on TERM at once - well before
|
||
a generous drain deadline and without the SIGKILL fallback."""
|
||
import logging
|
||
from lib_py.delegation import heartbeat, 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("idle-stop", registry_dir=str(reg))
|
||
tok = _job(reg, jid)["dispatch_token"]
|
||
pid = heartbeat.start_timer(jid, str(reg), 60, dispatch_token=tok, python=sys.executable, watch_pid=0)
|
||
try:
|
||
assert pid and _alive(pid)
|
||
time.sleep(2.0) # past interpreter start-up: TERM handler installed, nothing in flight
|
||
caplog.set_level(logging.WARNING, logger="delegate_job.heartbeat")
|
||
t0 = time.monotonic()
|
||
assert heartbeat.stop_timer(jid, str(reg), drain=10) is True
|
||
elapsed = time.monotonic() - t0
|
||
assert elapsed < 3.0, f"an idle timer took {elapsed:.1f}s to stop (drain deadline 10 s)"
|
||
assert "did not drain" not in caplog.text, caplog.text
|
||
assert not _alive(pid)
|
||
assert not (reg / f"{jid}.heartbeat.pid").exists()
|
||
finally:
|
||
if _alive(pid):
|
||
os.kill(pid, signal.SIGKILL)
|
||
|
||
|
||
def test_a_timer_whose_pid_cannot_be_recorded_is_killed_at_once(tmp_path, monkeypatch):
|
||
"""Review round 5 (codex F2): if publishing the new pid fails after the
|
||
timer was spawned, that exact process is killed and reaped before the
|
||
lifecycle lock is released - no untracked owner survives, no partial
|
||
pidfile is left, and the lock is free for the next owner."""
|
||
import builtins
|
||
import errno
|
||
import fcntl
|
||
from lib_py.delegation import heartbeat, 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("pidfile-fault", registry_dir=str(reg))
|
||
pidfile = str(reg / f"{jid}.heartbeat.pid")
|
||
spawned = []
|
||
real_popen, real_open = heartbeat.subprocess.Popen, builtins.open
|
||
|
||
def recording_popen(*a, **k):
|
||
proc = real_popen(*a, **k)
|
||
args = a[0] if a else k.get("args", [])
|
||
if any("lib_py.delegation.heartbeat" in str(x) for x in args):
|
||
spawned.append(proc)
|
||
return proc
|
||
|
||
def failing_open(file, mode="r", *a, **k):
|
||
if str(file) == pidfile and "w" in mode:
|
||
raise OSError(errno.ENOSPC, "No space left on device", pidfile)
|
||
return real_open(file, mode, *a, **k)
|
||
|
||
monkeypatch.setattr(heartbeat.subprocess, "Popen", recording_popen)
|
||
monkeypatch.setattr(builtins, "open", failing_open)
|
||
try:
|
||
with pytest.raises(OSError) as exc:
|
||
heartbeat.start_timer(jid, str(reg), 60, python=sys.executable, watch_pid=0)
|
||
finally:
|
||
monkeypatch.setattr(builtins, "open", real_open)
|
||
try:
|
||
assert exc.value.errno == errno.ENOSPC
|
||
assert len(spawned) == 1, spawned
|
||
proc = spawned[0]
|
||
assert proc.poll() is not None, "the spawned timer must not survive untracked"
|
||
assert proc.returncode == -signal.SIGKILL
|
||
assert not os.path.exists(pidfile)
|
||
with open(reg / f"{jid}.heartbeat.lock", "a+") as fh: # the lifecycle lock is free again
|
||
fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
|
||
finally:
|
||
for proc in spawned:
|
||
if proc.poll() is None:
|
||
proc.kill()
|
||
proc.wait(timeout=5)
|
||
|
||
|
||
def test_a_timer_that_cannot_be_spawned_leaves_nothing_behind(tmp_path, monkeypatch):
|
||
"""Review round 5 (codex F2), the other half: when the spawn itself fails
|
||
the error propagates, no pidfile is written, and the lifecycle lock is
|
||
released and serves the next owner at once."""
|
||
import fcntl
|
||
from lib_py.delegation import heartbeat, 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("spawn-fault", registry_dir=str(reg))
|
||
with pytest.raises(OSError):
|
||
heartbeat.start_timer(jid, str(reg), 60, python=str(tmp_path / "no-such-python"), watch_pid=0)
|
||
assert not (reg / f"{jid}.heartbeat.pid").exists()
|
||
with open(reg / f"{jid}.heartbeat.lock", "a+") as fh:
|
||
fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
|
||
pid = heartbeat.start_timer(jid, str(reg), 60, python=sys.executable, watch_pid=0)
|
||
try:
|
||
assert pid and _timer_pid(reg, jid) == pid and _alive(pid)
|
||
finally:
|
||
heartbeat.stop_timer(jid, str(reg), drain=5)
|
||
if _alive(pid):
|
||
os.kill(pid, signal.SIGKILL)
|
||
|
||
|
||
def _signed_state(job_id, seq, token, dispatch_token, status="running", attempt_id=1):
|
||
import hashlib, hmac
|
||
from lib_py.delegation import publish_event as pe
|
||
p = pe.build_payload(job_id, seq, "state", "progress", {"dispatch_token": dispatch_token, "attempt_id": attempt_id},
|
||
token)
|
||
p.update(status=status, last_event="progress")
|
||
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_state_renews_liveness_only_when_it_advances(tmp_path, monkeypatch):
|
||
"""Review round 2 (codex R4): the same retained snapshot re-delivered (as on
|
||
every reconnect) or an older one is still queued - a terminal status must
|
||
be recoverable - but does not renew the idle clock; only a state newer than
|
||
every event and state already accepted does."""
|
||
from lib_py.delegation import job_subscriber as js
|
||
|
||
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", 10, "progress", token)))
|
||
for _ in range(3): # duplicate delivery of an old snapshot
|
||
w.on_message(None, None, _Msg("x/state", _signed("j1", 1, "state", token, status="running", last_event="started")))
|
||
w.on_message(None, None, _Msg("x/state", _signed("j1", 10, "state", token, status="running", last_event="progress")))
|
||
w.on_message(None, None, _Msg("x/state", _signed("j1", 11, "state", token, status="running", last_event="progress")))
|
||
states = [p for _t, p in list(w.events.queue) if p["event"] == "state"]
|
||
assert [p["_renews_liveness"] for p in states] == [False, False, False, False, True]
|
||
|
||
|
||
def test_a_replayed_retained_state_does_not_keep_a_silent_job_alive(tmp_path, monkeypatch):
|
||
"""Review round 2 (codex R4), end to end: the agent went silent after seq 10
|
||
and the broker keeps re-delivering an old retained snapshot, as it does on
|
||
every reconnect. The subscriber must still reach its idle deadline rather
|
||
than run into the wall clock."""
|
||
from lib_py.delegation import job_subscriber as js, 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"
|
||
token = "tok-" + uuid.uuid4().hex
|
||
jid = registry.register_job("replay", registry_dir=str(reg), auth_token=token)
|
||
rec = _job(reg, jid)
|
||
rec["last_seq"] = 10
|
||
(reg / f"{jid}.json").write_text(json.dumps(rec))
|
||
state = _signed_state(jid, 1, token, rec["dispatch_token"])
|
||
topic = rec["topic_prefix"] + "/state"
|
||
|
||
class FakeClient:
|
||
def __init__(self):
|
||
self.on_message = self.on_connect = self.on_subscribe = self.on_disconnect = None
|
||
self.stopped = threading.Event()
|
||
|
||
def reconnect_delay_set(self, **_kw):
|
||
pass
|
||
|
||
def connect(self, *_a, **_kw):
|
||
return 0
|
||
|
||
def subscribe(self, *_a, **_kw):
|
||
return (0, 1)
|
||
|
||
def disconnect(self):
|
||
pass
|
||
|
||
def loop_stop(self):
|
||
self.stopped.set()
|
||
|
||
def loop_start(self):
|
||
def run():
|
||
self.on_connect(self, None, None, 0, None)
|
||
self.on_subscribe(self, None, 1, [1], None)
|
||
while not self.stopped.is_set():
|
||
self.on_message(self, None, _Msg(topic, state))
|
||
time.sleep(0.3)
|
||
threading.Thread(target=run, daemon=True).start()
|
||
|
||
fake = FakeClient()
|
||
monkeypatch.setattr(js, "make_client", lambda *_a, **_kw: fake)
|
||
started = time.monotonic()
|
||
rc = js.main(["--job", jid, "--registry-dir", str(reg), "--idle-timeout", "2", "--ack-timeout", "0",
|
||
"--timeout", "20"])
|
||
elapsed = time.monotonic() - started
|
||
fake.stopped.set()
|
||
assert rc == 2
|
||
assert elapsed < 8, f"the replayed state kept the silent job alive for {elapsed:.1f}s"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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 _skip_if_broker_did_not_deliver(reg: Path, runner_log: Path) -> None:
|
||
"""acceptance5 runs over a public broker. The remote runner already covers
|
||
the idle window with `started --heartbeat` (remote.py handle_dispatch). The
|
||
observed failures were of one kind: the runner published started and
|
||
completed successfully, the orchestrator had subscribed to the right topic,
|
||
and it received nothing at all for the job - the broker did not deliver.
|
||
Only that proven non-delivery skips; a drop (HMAC, identity), a wrong
|
||
topic, or a runner that failed to publish still fails."""
|
||
log = runner_log.read_text(errors="replace")
|
||
m = re.search(r"published started seq=1 job=(\w+)", log)
|
||
if not m:
|
||
return
|
||
jid = m.group(1)
|
||
rec = reg / f"{jid}.json"
|
||
if not rec.exists():
|
||
return
|
||
topic = json.loads(rec.read_text()).get("topic_prefix") or ""
|
||
out = reg / f"{jid}.subscriber.out"
|
||
sub = out.read_text(errors="replace") if out.exists() else ""
|
||
if (topic and f"SUBSCRIBED {topic}/events" in sub
|
||
and f"job={jid}" not in sub and f"drop event for job {jid}" not in sub):
|
||
pytest.skip(f"public broker did not deliver the runner's events to the orchestrator "
|
||
f"(runner published started for {jid}; subscriber received nothing for it)")
|
||
|
||
|
||
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)
|
||
# A fresh session name and secret per run: dispatches are retained on the
|
||
# (public) broker and the runner subscribes to every job's dispatch topic,
|
||
# so with fixed values a stale dispatch left by an earlier run was served
|
||
# instead of this run's job, which then timed out (observed: runner DONE
|
||
# for one job id, the orchestrator waiting on another).
|
||
tag = os.urandom(4).hex()
|
||
session = f"sim-worker-{tag}"
|
||
secret = f"issue18-shared-secret-{tag}"
|
||
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", session, "--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, f"remote:{session}", "remote",
|
||
["--heartbeat", "2", "--heartbeat-misses", "10", "--ack-timeout", "60"],
|
||
env_extra={"MAM_DISPATCH_SECRET": secret}, timeout_s=120)
|
||
stdout, stderr = proc.communicate(timeout=180)
|
||
jid = _wait_job_id(reg)
|
||
if proc.returncode != 0:
|
||
_skip_if_broker_did_not_deliver(reg, runner_log)
|
||
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"]
|