GodopuandClaude Opus 5.5
ec7f5c4e57
With MAM_DELEGATION_PY=0 the monitor was left in the caller's process group (decision 2), on every restart the update makes. The legacy body now spawns it under job control (`set -m`, macOS has no setsid(1)) and restores the caller's option. The update tests assert the restarted monitor's PGID in both modes, and their teardown kills helper groups and this test's exact monitors instead of `pkill -f <ws>`, then reports anything left that names the workspace. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
967 lines
43 KiB
Python
967 lines
43 KiB
Python
"""Issue #72 - the multi-agent-mux-update skill.
|
|
|
|
Every test updates a copy of one workspace installed from this checkout
|
|
(MAM_REPO_URL, MAM_SKIP_VENV=1). herdr is a fake that answers `session list
|
|
--json` and `--session S agent list` from a JSON spec, so the pre-check's view
|
|
of the live agents is set per test; the monitor daemon is a stub reconcile.sh
|
|
that holds .mam/monitor.lock like the real one. Nothing here touches a real
|
|
herdr server or the control root's monitor.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
UPDATE_REL = ".agents/skills/multi-agent-mux-update/scripts/update.sh"
|
|
MONITOR_REL = ".agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh"
|
|
|
|
FAKE_HERDR = r'''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
spec = json.load(open(os.environ["FAKE_HERDR_SPEC"]))
|
|
a = sys.argv[1:]
|
|
server = "default"
|
|
if a[:1] == ["--session"]:
|
|
server, a = a[1], a[2:]
|
|
with open(os.environ["FAKE_HERDR_SPEC"] + ".calls", "a") as f:
|
|
f.write(json.dumps([server] + a) + "\n")
|
|
if a[:2] == ["session", "list"]:
|
|
if spec.get("fail") == "session-list":
|
|
sys.exit(1)
|
|
print(json.dumps({"sessions": [{"name": n, "running": r} for n, r in spec.get("sessions", {"default": True}).items()]}))
|
|
elif a[:2] == ["agent", "list"]:
|
|
if spec.get("fail") == "agent-list:" + server:
|
|
sys.exit(1)
|
|
print(json.dumps({"result": {"agents": spec.get("agents", {}).get(server, [])}}))
|
|
elif a[:2] == ["agent", "get"]:
|
|
hit = [x for x in spec.get("agents", {}).get(server, []) if a[2] in (x.get("name"), x.get("pane_id"))]
|
|
if not hit:
|
|
sys.exit(1)
|
|
print(json.dumps({"result": {"agent": hit[0]}}))
|
|
elif a[:2] == ["pane", "close"]:
|
|
spec.setdefault("agents", {})[server] = [x for x in spec.get("agents", {}).get(server, [])
|
|
if x.get("pane_id") != a[2]]
|
|
json.dump(spec, open(os.environ["FAKE_HERDR_SPEC"], "w"))
|
|
'''
|
|
|
|
STUB_MONITOR = r'''#!/usr/bin/env bash
|
|
# Issue #72 test stub of reconcile.sh: a wrapper whose python child holds monitor.lock.
|
|
d="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd -P)"
|
|
echo "start $$ $*" >> "$d/.mam/stub-monitor.log"
|
|
python3 -c '
|
|
import fcntl, os, signal, sys, time
|
|
if os.environ.get("STUB_MONITOR_IGNORE_TERM") == "1":
|
|
signal.signal(signal.SIGTERM, signal.SIG_IGN)
|
|
f = open(sys.argv[1], "w")
|
|
fcntl.flock(f, fcntl.LOCK_EX)
|
|
time.sleep(600)
|
|
' "$d/.mam/monitor.lock" &
|
|
child=$!
|
|
if [ "${STUB_MONITOR_IGNORE_TERM:-0}" = "1" ]; then
|
|
trap '' TERM
|
|
else
|
|
trap 'kill $child 2>/dev/null; exit 143' TERM
|
|
fi
|
|
while kill -0 "$child" 2>/dev/null; do wait "$child"; done
|
|
'''
|
|
|
|
|
|
def _clean_env():
|
|
env = {k: v for k, v in os.environ.items()
|
|
if not k.startswith(("HERDR_", "XDG_", "MAM_", "STUB_"))
|
|
and k not in ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "DELEGATE_JOB_LOGS_DIR", "PYTHONPATH")}
|
|
env.update(MAM_REPO_URL=str(REPO_ROOT), MAM_SKIP_VENV="1",
|
|
MAM_INSTALLER_URL="file://" + str(REPO_ROOT / "deploy" / "install.sh"),
|
|
MAM_ENV_FILE="/nonexistent/.mam.env", PYTHONDONTWRITEBYTECODE="1")
|
|
return env
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def template(tmp_path_factory):
|
|
ws = tmp_path_factory.mktemp("tpl72") / "ws"
|
|
ws.mkdir()
|
|
res = subprocess.run(["bash", str(REPO_ROOT / "deploy" / "install.sh"), str(ws)], env=_clean_env(),
|
|
capture_output=True, text=True, timeout=600)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
return ws
|
|
|
|
|
|
class Box:
|
|
def __init__(self, root: Path, ws: Path):
|
|
self.root, self.ws = root, ws
|
|
self.mam = ws / ".mam"
|
|
self.bin = root / "bin"
|
|
self.spec = root / "herdr.json"
|
|
self.env = _clean_env()
|
|
self.env["PATH"] = "%s:%s" % (self.bin, os.environ["PATH"])
|
|
self.env["FAKE_HERDR_SPEC"] = str(self.spec)
|
|
self.helpers = []
|
|
self.herdr()
|
|
|
|
def herdr(self, sessions=None, agents=None, fail=None):
|
|
self.spec.write_text(json.dumps({"sessions": sessions or {"default": True},
|
|
"agents": agents or {}, "fail": fail}))
|
|
|
|
def update(self, *args, env=None, timeout=300):
|
|
e = dict(self.env, **(env or {}))
|
|
return subprocess.run(["bash", str(self.ws / UPDATE_REL), "--yes", *args], cwd=str(self.ws), env=e,
|
|
stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=timeout)
|
|
|
|
def lib(self, script, env=None, timeout=120):
|
|
e = dict(self.env, WORKSPACE_ROOT=str(self.ws), **(env or {}))
|
|
return subprocess.run(["bash", "-c", 'source "$1/.agents/skills/lib.sh"\n' + script, "_", str(self.ws)],
|
|
env=e, capture_output=True, text=True, timeout=timeout)
|
|
|
|
def installer(self, body):
|
|
p = self.root / "installer.sh"
|
|
p.write_text("#!/usr/bin/env bash\nws=\"$1\"\n" + body)
|
|
self.env["MAM_INSTALLER_URL"] = "file://" + str(p)
|
|
|
|
def helper(self, *argv, **kw):
|
|
p = subprocess.Popen(list(argv), start_new_session=True, **kw)
|
|
self.helpers.append(p)
|
|
return p
|
|
|
|
def stub_monitor(self):
|
|
p = self.ws / MONITOR_REL
|
|
p.write_text(STUB_MONITOR)
|
|
|
|
def start_monitor(self, env=None):
|
|
"""The way create and resume start it: lib.sh start_watchdog (detached, so
|
|
not a child of this test that would linger as a zombie)."""
|
|
res = self.lib('start_watchdog test "$WORKSPACE_ROOT"', env=dict(env or {}))
|
|
assert res.returncode == 0 and res.stdout.strip(), res.stdout + res.stderr
|
|
_until(lambda: _lock_held(self.mam / "monitor.lock"))
|
|
|
|
def monitor_pids(self):
|
|
return _exact_monitor_pids(self.ws / MONITOR_REL)
|
|
|
|
def decoy(self):
|
|
"""A process that only names this workspace's monitor in an argument."""
|
|
return self.helper("/bin/bash", "-c", "sleep 120 & wait $!",
|
|
"bash %s --subscribe" % (self.ws / MONITOR_REL))
|
|
|
|
def monitor_starts(self):
|
|
p = self.mam / "stub-monitor.log"
|
|
return p.read_text().count("start ") if p.exists() else 0
|
|
|
|
|
|
def _until(cond, timeout=15.0):
|
|
end = time.monotonic() + timeout
|
|
while time.monotonic() < end:
|
|
if cond():
|
|
return True
|
|
time.sleep(0.1)
|
|
raise AssertionError("condition not met within %ss" % timeout)
|
|
|
|
|
|
def _exact_monitor_pids(script) -> list:
|
|
"""`bash <script> --subscribe ...` processes, judged here independently of
|
|
the code under test (a `pgrep -f` would also count a decoy)."""
|
|
out = subprocess.run(["ps", "-A", "-ww", "-o", "pid=", "-o", "args="], capture_output=True, text=True).stdout
|
|
want = "%s --subscribe" % script
|
|
pids = []
|
|
for line in out.splitlines():
|
|
pid, _, args = line.strip().partition(" ")
|
|
argv0, _, rest = args.partition(" ")
|
|
if os.path.basename(argv0) == "bash" and (rest == want or rest.startswith(want + " ")):
|
|
pids.append(pid)
|
|
return pids
|
|
|
|
|
|
def _descendants(pid) -> list:
|
|
kids = subprocess.run(["pgrep", "-P", str(pid)], capture_output=True, text=True).stdout.split()
|
|
return [d for k in kids for d in [k] + _descendants(k)]
|
|
|
|
|
|
def _processes_naming(path) -> list:
|
|
"""Live processes whose command line names PATH (read only, to report leftovers)."""
|
|
out = subprocess.run(["ps", "-A", "-ww", "-o", "pid=", "-o", "args="], capture_output=True, text=True).stdout
|
|
return [line.strip() for line in out.splitlines()
|
|
if str(path) in line and line.split(None, 1)[0] != str(os.getpid())]
|
|
|
|
|
|
def _lock_held(path: Path) -> bool:
|
|
if not path.exists():
|
|
return False
|
|
with open(path, "a") as fh:
|
|
try:
|
|
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
fcntl.flock(fh, fcntl.LOCK_UN)
|
|
return False
|
|
except BlockingIOError:
|
|
return True
|
|
|
|
|
|
@pytest.fixture
|
|
def box(template, tmp_path):
|
|
root = Path(os.path.realpath(tmp_path))
|
|
ws = root / "ws"
|
|
shutil.copytree(template, ws, symlinks=True)
|
|
(root / "bin").mkdir()
|
|
herdr = root / "bin" / "herdr"
|
|
herdr.write_text(FAKE_HERDR)
|
|
herdr.chmod(0o755)
|
|
b = Box(root, ws)
|
|
try:
|
|
yield b
|
|
finally:
|
|
# Only what this test started: its helpers' process groups, and the
|
|
# monitors whose exact command line is this test's own workspace
|
|
# (detached, so in a group of their own), with their children. No
|
|
# pattern kill: the host runs other workspaces' monitors.
|
|
for p in b.helpers:
|
|
try:
|
|
os.killpg(p.pid, signal.SIGKILL)
|
|
except OSError:
|
|
pass
|
|
p.wait()
|
|
for pid in _exact_monitor_pids(ws / MONITOR_REL):
|
|
for q in [pid] + _descendants(pid):
|
|
try:
|
|
os.kill(int(q), signal.SIGKILL)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
_until(lambda: not _processes_naming(ws), timeout=10)
|
|
except AssertionError:
|
|
raise AssertionError("processes of this test outlived it: %s" % _processes_naming(ws))
|
|
|
|
|
|
# --- helpers ---------------------------------------------------------------------
|
|
|
|
SKIP_SNAPSHOT = ("shim/", "update-admission.lock", "agent-sessions.db.lock", "monitor.lock", "stub-monitor.log",
|
|
"agent-sessions.db-wal", "agent-sessions.db-shm")
|
|
|
|
|
|
def _db_content(p: Path) -> str:
|
|
"""The registry DB by its rows: on macOS SQLite keeps -wal/-shm after close and
|
|
a read rewrites -shm, and a checkpoint moves the same rows between -wal and
|
|
the file, so neither file's bytes say whether the data changed. A file that
|
|
is not a database is compared by its bytes."""
|
|
conn = sqlite3.connect(str(p))
|
|
try:
|
|
return "DB:" + hashlib.sha256("\n".join(conn.iterdump()).encode()).hexdigest()
|
|
except sqlite3.DatabaseError:
|
|
return hashlib.sha256(p.read_bytes()).hexdigest()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def snapshot(ws: Path) -> dict:
|
|
out = {}
|
|
for dp, dns, fns in os.walk(ws):
|
|
dns[:] = [d for d in dns if d != "__pycache__"]
|
|
for f in fns:
|
|
p = Path(dp) / f
|
|
rel = str(p.relative_to(ws))
|
|
if rel.startswith(".mam/") and any(s in rel for s in SKIP_SNAPSHOT):
|
|
continue
|
|
if rel == ".mam/agent-sessions.db":
|
|
out[rel] = _db_content(p)
|
|
continue
|
|
out[rel] = "L:" + os.readlink(p) if p.is_symlink() else hashlib.sha256(p.read_bytes()).hexdigest()
|
|
return out
|
|
|
|
|
|
def write_rows(ws: Path, rows, **top):
|
|
d = dict(top, herdr_sessions=rows)
|
|
(ws / ".mam" / "agent-sessions.yaml").write_text(yaml.safe_dump(d, sort_keys=False))
|
|
|
|
|
|
def row(name, status="running", server="default", cwd=None, **extra):
|
|
r = {"name": name, "status": status, "role": "creator", "herdr_session": server, "herdr_server": server,
|
|
"pane": {"cwd": cwd or "/elsewhere", "cmd": "claude"}}
|
|
r.update(extra)
|
|
return r
|
|
|
|
|
|
def write_job(d: Path, jid, status="running", session="herdr:x", **extra):
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
rec = dict({"job_id": jid, "status": status, "agent_session": session}, **extra)
|
|
(d / (jid + ".json")).write_text(json.dumps(rec))
|
|
|
|
|
|
def live_marker(path: Path, box: Box):
|
|
"""A marker owned by a live process (pid + lstart as loop_lock.sh writes them)."""
|
|
p = box.helper("sleep", "120")
|
|
ls = subprocess.run(["ps", "-p", str(p.pid), "-o", "lstart="], capture_output=True, text=True).stdout
|
|
ls = " ".join(ls.split())
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("pid=%d\nlstart=%s\nstarted=x\n" % (p.pid, ls))
|
|
return p
|
|
|
|
|
|
def dead_marker(path: Path):
|
|
p = subprocess.Popen(["true"])
|
|
p.wait()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("pid=%d\nlstart=Thu Jan 1 00:00:00 1970\nstarted=x\n" % p.pid)
|
|
|
|
|
|
def assert_refused_unchanged(box: Box, before: dict, res, *needles):
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert "update refused; nothing was changed" in res.stderr, res.stderr
|
|
for n in needles:
|
|
assert n in res.stderr, (n, res.stderr)
|
|
assert snapshot(box.ws) == before
|
|
assert not (box.mam / "update-in-progress").exists(), "the refusing update releases its marker"
|
|
assert not (box.mam / "update-backup").exists()
|
|
|
|
|
|
# --- success and the #8 state rules ------------------------------------------------
|
|
|
|
def test_update_in_place_keeps_mam_and_merges_install_state(box):
|
|
mam = box.mam
|
|
(mam / "version.txt").write_text("source=old\ncommit=stale0000\nfetched_at=then\nmethod=git\n")
|
|
(mam / "herdr_workspace_id").write_text("w9\n")
|
|
for d in ("briefs", "diagnostics", "resumelogs", "stoplogs", "jobs", "delegate_job_logs/j1"):
|
|
(mam / d).mkdir(parents=True, exist_ok=True)
|
|
(mam / d / "keep.txt").write_text(d)
|
|
write_job(mam / "jobs", "done1", status="completed")
|
|
(mam / "multi-agent-mux-monitor.log").write_text("log\n")
|
|
(box.ws / ".mam.env").write_text("MQTT_BROKER=example\n")
|
|
with open(mam / "install_state", "a") as fh:
|
|
fh.write("custom_marker=kept\n")
|
|
write_rows(box.ws, [row("a", status="stopped")], orchestrator_uuids=["u1"])
|
|
keep = {k: v for k, v in snapshot(box.ws).items()
|
|
if k.startswith((".mam/herdr_workspace_id", ".mam/briefs", ".mam/diagnostics", ".mam/resumelogs",
|
|
".mam/stoplogs", ".mam/jobs", ".mam/delegate_job_logs", ".mam/multi-agent-mux-monitor.log"))
|
|
or k == ".mam.env"}
|
|
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
after = snapshot(box.ws)
|
|
assert {k: after.get(k) for k in keep} == keep, "everything that is not the installer's is kept"
|
|
assert "stale0000" not in (mam / "version.txt").read_text()
|
|
state = (mam / "install_state").read_text().splitlines()
|
|
assert "custom_marker=kept" in state and state.count("gitignore_created=1") == 1, state
|
|
assert not (box.ws / ".mam_deploy" / "update.sh").exists()
|
|
assert not (mam / "update-backup").exists() and not (mam / "update-in-progress").exists()
|
|
assert "Update complete." in res.stdout
|
|
|
|
|
|
def test_old_format_registry_is_converted_by_the_new_code(box):
|
|
old = row("old-one", status="stopped", start_command="HERDR_SESSION_NAME=x herdr new-session",
|
|
kill_command="HERDR_SESSION_NAME=x herdr kill-session", attach_command="HERDR_SESSION_NAME=x herdr agent attach old-one")
|
|
write_rows(box.ws, [old])
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "registry-format: 1 -> 2, rows changed: 1" in res.stdout
|
|
y = yaml.safe_load((box.mam / "agent-sessions.yaml").read_text())
|
|
assert y["registry_format"] == 2
|
|
conn = sqlite3.connect(str(box.mam / "agent-sessions.db"))
|
|
try:
|
|
db = {r["name"]: r for r in (json.loads(x[0]) for x in conn.execute("SELECT data FROM sessions"))}
|
|
st = json.loads(conn.execute("SELECT data FROM state WHERE id=1").fetchone()[0])
|
|
finally:
|
|
conn.close()
|
|
assert {r["name"]: r for r in y["herdr_sessions"]} == db and st["registry_format"] == 2
|
|
assert "kill_command" not in db["old-one"] and "start_command" not in db["old-one"]
|
|
|
|
|
|
def test_no_registry_is_nothing_to_convert(box):
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "registry: nothing to convert (no registry)" in res.stdout
|
|
assert not (box.mam / "agent-sessions.yaml").exists()
|
|
|
|
|
|
def test_a_failed_conversion_keeps_the_install_and_exits_nonzero(box, tmp_path):
|
|
src = tmp_path / "src"
|
|
shutil.copytree(REPO_ROOT, src, symlinks=True,
|
|
ignore=shutil.ignore_patterns(".git", ".venv", ".mam", "__pycache__", "tests", "nats-docker"))
|
|
fmt = src / ".agents" / "skills" / "lib_py" / "registry_format.py"
|
|
fmt.write_text("import sys\nif __name__ == '__main__':\n sys.exit(3)\n" + fmt.read_text())
|
|
box.env["MAM_REPO_URL"] = str(src)
|
|
box.stub_monitor()
|
|
box.start_monitor()
|
|
write_rows(box.ws, [row("a", status="stopped")])
|
|
res = box.update()
|
|
assert res.returncode == 3, res.stdout + res.stderr
|
|
assert "the registry was not converted" in res.stderr
|
|
assert (box.ws / ".agents/skills/lib_py/registry_format.py").read_text() == fmt.read_text(), "install kept"
|
|
assert box.monitor_pids() and box.monitor_starts() == 2, "the monitor is started again anyway"
|
|
assert not (box.mam / "update-backup").exists() and not (box.mam / "update-in-progress").exists()
|
|
|
|
|
|
def test_refuses_outside_an_install_and_without_yes(box):
|
|
(box.ws / ".mam_deploy" / "remove.sh").unlink()
|
|
res = box.update()
|
|
assert res.returncode == 1 and "not an installed MAM workspace" in res.stderr
|
|
shutil.copy(REPO_ROOT / "deploy" / "remove.sh", box.ws / ".mam_deploy" / "remove.sh")
|
|
res = subprocess.run(["bash", str(box.ws / UPDATE_REL)], env=box.env, stdin=subprocess.DEVNULL,
|
|
capture_output=True, text=True, timeout=60)
|
|
assert res.returncode == 1 and "-y/--yes" in res.stderr
|
|
res = box.update("--force")
|
|
assert res.returncode == 2, "--force is not an alias of --yes"
|
|
|
|
|
|
# --- refusals (decision 1) ---------------------------------------------------------
|
|
|
|
def test_refuses_a_running_row(box):
|
|
write_rows(box.ws, [row("busy-creator", cwd="/somewhere/else")])
|
|
before = snapshot(box.ws)
|
|
res = box.update()
|
|
assert_refused_unchanged(box, before, res, "session busy-creator is running", "stop_session.sh --session busy-creator")
|
|
|
|
|
|
def test_refuses_an_unreadable_registry(box):
|
|
(box.mam / "agent-sessions.db").write_text("not a database")
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(), "cannot be read")
|
|
|
|
|
|
def test_refuses_a_rowless_agent_on_any_running_server(box):
|
|
box.herdr(sessions={"default": True, "picked-by-hand": True, "down": False},
|
|
agents={"picked-by-hand": [{"name": "stray", "pane_id": "w3:p1", "cwd": str(box.ws / "sub")}],
|
|
"down": [{"name": "ghost", "pane_id": "w4:p1", "cwd": str(box.ws)}]})
|
|
before = snapshot(box.ws)
|
|
res = box.update()
|
|
assert_refused_unchanged(box, before, res, "herdr agent stray (pane w3:p1, server picked-by-hand)")
|
|
assert "ghost" not in res.stderr, "a server that is not running is not asked"
|
|
calls = (box.root / "herdr.json.calls").read_text()
|
|
assert '"down"' not in calls
|
|
|
|
|
|
def test_the_rowless_agent_hint_stops_it_on_the_server_it_was_found_on(box):
|
|
"""R3: the printed stop command names the server; without it stop looks on
|
|
the workspace's server, where a same-named agent runs outside the workspace."""
|
|
write_rows(box.ws, [])
|
|
box.herdr(sessions={"default": True, "unrecorded-server": True},
|
|
agents={"unrecorded-server": [{"name": "stray", "pane_id": "w3:p1", "cwd": str(box.ws / "sub")}],
|
|
"default": [{"name": "stray", "pane_id": "w1:p1", "cwd": str(box.root / "elsewhere")}]})
|
|
res = box.update()
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
line = next(ln for ln in res.stderr.splitlines() if "herdr agent stray" in ln)
|
|
cmd = line.split("stop it: ", 1)[1]
|
|
assert "--herdr-session unrecorded-server" in cmd, cmd
|
|
stop = subprocess.run(["bash", "-c", cmd], cwd=str(box.ws), env=box.env, stdin=subprocess.DEVNULL,
|
|
capture_output=True, text=True, timeout=120)
|
|
assert stop.returncode == 0, stop.stdout + stop.stderr
|
|
agents = json.loads(box.spec.read_text())["agents"]
|
|
assert agents["unrecorded-server"] == []
|
|
assert [x["pane_id"] for x in agents["default"]] == ["w1:p1"], "the same name elsewhere is left alone"
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
|
|
|
|
def test_an_agent_outside_the_workspace_does_not_refuse(box):
|
|
box.herdr(agents={"default": [{"name": "other", "pane_id": "w1:p2", "cwd": str(box.root / "elsewhere")}]})
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
|
|
|
|
@pytest.mark.parametrize("fail", ["session-list", "agent-list:default"])
|
|
def test_refuses_when_herdr_cannot_answer(box, fail):
|
|
box.herdr(fail=fail)
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(), "cannot list the live agents")
|
|
|
|
|
|
def test_open_jobs_refuse_and_terminal_or_dry_run_do_not(box, tmp_path):
|
|
jobs = box.mam / "jobs"
|
|
write_job(jobs, "done1", status="completed")
|
|
write_job(jobs, "err1", status="error")
|
|
write_job(jobs, "can1", status="cancelled")
|
|
write_job(jobs, "dry1", status="pending", dry_run=True)
|
|
assert box.update().returncode == 0
|
|
|
|
write_job(jobs, "open1", status="pending", session="remote:x")
|
|
before = snapshot(box.ws)
|
|
res = box.update()
|
|
assert_refused_unchanged(box, before, res, "job open1", "close --job open1 --request cancelled")
|
|
(jobs / "open1.json").unlink()
|
|
|
|
(jobs / "broken.json").write_text("{not json")
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(), "job record %s cannot be read" % (jobs / "broken.json"))
|
|
(jobs / "broken.json").unlink()
|
|
|
|
extra = tmp_path / "mam-jobs-dir"
|
|
write_job(extra, "env1", status="running")
|
|
box.env["MAM_JOBS_DIR"] = str(extra)
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(), "job env1")
|
|
del box.env["MAM_JOBS_DIR"]
|
|
|
|
|
|
@pytest.mark.skipif(os.geteuid() == 0, reason="root reads a mode-000 directory")
|
|
def test_a_jobs_directory_that_cannot_be_listed_refuses(box, tmp_path):
|
|
"""R2: an unreadable .mam/jobs (or MAM_JOBS_DIR under an unreadable parent)
|
|
is not an empty registry."""
|
|
jobs = box.mam / "jobs"
|
|
write_job(jobs, "hidden1", status="running")
|
|
before = snapshot(box.ws)
|
|
jobs.chmod(0)
|
|
try:
|
|
res = box.update()
|
|
finally:
|
|
jobs.chmod(0o755)
|
|
assert_refused_unchanged(box, before, res, "jobs directory %s cannot be listed" % jobs)
|
|
(jobs / "hidden1.json").unlink()
|
|
|
|
parent = tmp_path / "locked"
|
|
write_job(parent / "jobs", "env1", status="running")
|
|
box.env["MAM_JOBS_DIR"] = str(parent / "jobs")
|
|
before = snapshot(box.ws)
|
|
parent.chmod(0)
|
|
try:
|
|
res = box.update()
|
|
finally:
|
|
parent.chmod(0o755)
|
|
assert_refused_unchanged(box, before, res, "jobs directory %s cannot be listed" % (parent / "jobs"))
|
|
del box.env["MAM_JOBS_DIR"]
|
|
|
|
|
|
def test_other_control_planes_count_only_jobs_addressed_to_this_registry(box, tmp_path):
|
|
other = tmp_path / "other-plane" / "jobs"
|
|
write_job(other, "theirs", status="running", session="herdr:not-ours")
|
|
write_rows(box.ws, [row("wt-creator", status="stopped", control_plane={"mam_jobs_dir": str(other)})])
|
|
assert box.update().returncode == 0, "a job of another registry's session is not ours"
|
|
write_job(other, "ours", status="running", session="herdr:wt-creator")
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(), "job ours")
|
|
|
|
|
|
def test_loop_marker_live_refuses_stale_does_not(box):
|
|
live_marker(box.mam / "loop-guard-active", box)
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(), "a review loop is running")
|
|
dead_marker(box.mam / "loop-guard-active")
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "stale loop marker" in res.stdout
|
|
|
|
|
|
# --- the calling agent (decision 7, N2) ----------------------------------------------
|
|
|
|
def _me_env(server="mam-x", pane="w5:p1"):
|
|
return {"HERDR_SESSION": server, "HERDR_PANE_ID": pane}
|
|
|
|
|
|
def test_the_calling_agent_its_row_and_its_running_job_are_excluded(box):
|
|
me = {"name": "me-creator", "pane_id": "w5:p1", "cwd": str(box.ws)}
|
|
box.herdr(sessions={"default": True, "mam-x": True}, agents={"mam-x": [me]})
|
|
write_rows(box.ws, [row("me-creator", server="mam-x", cwd=str(box.ws))])
|
|
write_job(box.mam / "jobs", "myjob", status="running", session="herdr:me-creator")
|
|
res = box.update(env=_me_env())
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
|
|
write_job(box.mam / "jobs", "myjob2", status="pending", session="herdr:me-creator")
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(env=_me_env()), "job myjob2")
|
|
|
|
|
|
def test_same_pane_id_on_another_server_is_not_excluded(box):
|
|
box.herdr(sessions={"default": True, "mam-x": True, "mam-y": True},
|
|
agents={"mam-x": [{"name": "me", "pane_id": "w5:p1", "cwd": str(box.ws)}],
|
|
"mam-y": [{"name": "twin", "pane_id": "w5:p1", "cwd": str(box.ws)}]})
|
|
before = snapshot(box.ws)
|
|
res = box.update(env=_me_env())
|
|
assert_refused_unchanged(box, before, res, "herdr agent twin (pane w5:p1, server mam-y)")
|
|
assert "agent me " not in res.stderr
|
|
|
|
|
|
def test_a_nameless_caller_pane_is_excluded_by_pane_only(box):
|
|
box.herdr(sessions={"default": True, "mam-x": True},
|
|
agents={"mam-x": [{"name": None, "pane_id": "w5:p1", "cwd": str(box.ws)}]})
|
|
assert box.update(env=_me_env()).returncode == 0
|
|
write_job(box.mam / "jobs", "nameless", status="running", session="herdr:")
|
|
write_job(box.mam / "jobs", "nonename", status="running", session="herdr:None")
|
|
before = snapshot(box.ws)
|
|
assert_refused_unchanged(box, before, box.update(env=_me_env()), "job nameless", "job nonename")
|
|
|
|
|
|
@pytest.mark.parametrize("env", [{}, {"HERDR_PANE_ID": "w5:p1"}, {"HERDR_SESSION": "mam-x"}])
|
|
def test_nothing_is_excluded_when_the_caller_is_not_identified(box, env):
|
|
box.herdr(sessions={"default": True, "mam-x": True},
|
|
agents={"mam-x": [{"name": "me", "pane_id": "w5:p1", "cwd": str(box.ws)}]})
|
|
before = snapshot(box.ws)
|
|
res = box.update(env=env)
|
|
assert_refused_unchanged(box, before, res, "herdr agent me", "nothing was excluded as the calling agent")
|
|
|
|
|
|
# --- the monitor (decision 2, U3) ----------------------------------------------------
|
|
|
|
def test_a_running_monitor_is_stopped_and_started_again(box):
|
|
box.stub_monitor()
|
|
box.start_monitor()
|
|
first = set(box.monitor_pids())
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "monitor: stopping the daemon" in res.stdout and "monitor: restarted" in res.stdout
|
|
now = set(box.monitor_pids())
|
|
assert now and not (now & first) and box.monitor_starts() == 2
|
|
|
|
|
|
def test_no_monitor_before_means_none_after(box):
|
|
box.stub_monitor()
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert not box.monitor_pids() and box.monitor_starts() == 0
|
|
|
|
|
|
def test_no_subscriber_daemon_flag_wins(box):
|
|
box.stub_monitor()
|
|
box.start_monitor()
|
|
res = box.update(env={"MAM_NO_SUBSCRIBER_DAEMON": "1"})
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "monitor: not restarted (MAM_NO_SUBSCRIBER_DAEMON=1)" in res.stdout
|
|
assert not box.monitor_pids()
|
|
|
|
|
|
def test_a_monitor_that_ignores_term_is_killed_restarted_and_nothing_installed(box):
|
|
box.stub_monitor()
|
|
box.start_monitor(env={"STUB_MONITOR_IGNORE_TERM": "1"})
|
|
box.installer("touch \"$ws/INSTALLER-RAN\"\n")
|
|
before = snapshot(box.ws)
|
|
res = box.update(env={"MAM_UPDATE_MONITOR_STOP_S": "1"})
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert "it was killed and is started again" in res.stderr
|
|
assert not (box.ws / "INSTALLER-RAN").exists()
|
|
assert snapshot(box.ws) == before
|
|
assert box.monitor_pids() and box.monitor_starts() == 2
|
|
assert not (box.mam / "update-in-progress").exists()
|
|
|
|
|
|
def test_a_monitor_lock_held_by_an_unknown_process_refuses(box):
|
|
lock = box.mam / "monitor.lock"
|
|
with open(lock, "w") as fh:
|
|
fcntl.flock(fh, fcntl.LOCK_EX)
|
|
before = snapshot(box.ws)
|
|
res = box.update()
|
|
assert_refused_unchanged(box, before, res, "monitor.lock is held by a monitor the skills did not start")
|
|
|
|
|
|
@pytest.mark.parametrize("finder", ["python", "shell"])
|
|
def test_only_this_workspaces_exact_monitor_command_line_is_its_monitor(tmp_path, finder):
|
|
"""The update signals what `lib_py.update monitor` returns, and the legacy
|
|
start_watchdog (MAM_DELEGATION_PY=0) reuses what lib.sh _monitor_pids finds:
|
|
`bash <this workspace's reconcile.sh> --subscribe ...` only - not another
|
|
workspace's daemon the path matched as a regex (ws.a, wsXa), nor a process
|
|
that names the path in an argument. A path with a space is still found."""
|
|
root = Path(os.path.realpath(tmp_path))
|
|
procs = []
|
|
|
|
def daemon_in(ws):
|
|
script = ws / MONITOR_REL
|
|
script.parent.mkdir(parents=True)
|
|
script.write_text("sleep 120 & wait $!\n")
|
|
procs.append(subprocess.Popen(["bash", str(script), "--subscribe", "--idle-timeout", "0"],
|
|
start_new_session=True))
|
|
return str(procs[-1].pid)
|
|
|
|
def monitor(ws):
|
|
if finder == "shell":
|
|
res = subprocess.run(["bash", "-c", 'source "$1" && _monitor_pids "$2"', "_",
|
|
str(REPO_ROOT / ".agents" / "skills" / "lib.sh"), str(ws / MONITOR_REL)],
|
|
env=dict(_clean_env(), WORKSPACE_ROOT=str(root)), capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
return res.stdout.split()
|
|
res = subprocess.run([sys.executable, "-m", "lib_py.update", "monitor", "--workspace", str(ws)],
|
|
cwd=str(REPO_ROOT / ".agents" / "skills"), capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
return json.loads(res.stdout)["pids"]
|
|
|
|
try:
|
|
a, other, spaced = root / "ws.a", root / "wsXa", root / "w s"
|
|
a.mkdir()
|
|
daemon_in(other)
|
|
procs.append(subprocess.Popen(["/bin/bash", "-c", "sleep 120 & wait $!",
|
|
"bash %s --subscribe" % (a / MONITOR_REL)], start_new_session=True))
|
|
pid = daemon_in(spaced)
|
|
_until(lambda: monitor(spaced) == [pid])
|
|
assert monitor(a) == []
|
|
finally:
|
|
for p in procs:
|
|
try:
|
|
os.killpg(p.pid, signal.SIGKILL)
|
|
except OSError:
|
|
pass
|
|
p.wait()
|
|
|
|
|
|
@pytest.mark.parametrize("case", ["installed", "killed", "rolled-back"])
|
|
@pytest.mark.parametrize("mode", ["0", "1"])
|
|
def test_every_restart_starts_a_real_monitor_in_both_delegation_modes(box, mode, case):
|
|
"""MAM_DELEGATION_PY=0 keeps the legacy start_watchdog, the switch the
|
|
delegate skill documents. Its restart, after an install, after a stop
|
|
timeout or after a rollback, must start this workspace's monitor again, out
|
|
of the caller's process group (decision 2), and not report a process that
|
|
only names the monitor path in an argument."""
|
|
env = {"MAM_DELEGATION_PY": mode}
|
|
box.stub_monitor()
|
|
box.start_monitor(env=dict(env, STUB_MONITOR_IGNORE_TERM="1") if case == "killed" else env)
|
|
first = set(box.monitor_pids())
|
|
decoy = box.decoy()
|
|
_until(lambda: str(decoy.pid) in subprocess.run(["ps", "-A", "-ww", "-o", "pid=", "-o", "args="],
|
|
capture_output=True, text=True).stdout)
|
|
if case == "killed":
|
|
env["MAM_UPDATE_MONITOR_STOP_S"] = "1"
|
|
if case == "rolled-back":
|
|
box.installer("exit 7\n")
|
|
res = box.update(env=env)
|
|
assert res.returncode == (0 if case == "installed" else 1), res.stdout + res.stderr
|
|
assert {"installed": "monitor: restarted", "killed": "it was killed and is started again",
|
|
"rolled-back": "rolled back"}[case] in res.stdout + res.stderr
|
|
assert "PID %s)" % decoy.pid not in res.stdout, "the decoy is reported as the monitor"
|
|
_until(lambda: bool(box.monitor_pids()))
|
|
assert not (set(box.monitor_pids()) & first) and box.monitor_starts() == 2
|
|
# box.update runs update.sh in this test's process group: the caller's.
|
|
assert all(os.getpgid(int(p)) != os.getpgrp() for p in box.monitor_pids()), "left in the caller's group"
|
|
assert decoy.poll() is None, "the decoy was not signalled"
|
|
|
|
|
|
# --- rollback (decisions 3, 10; U4) --------------------------------------------------
|
|
|
|
FAILING_INSTALL = """
|
|
echo broken >> "$ws/.agents/skills/lib.sh"
|
|
rm -f "$ws/.mam_deploy/remove.sh"
|
|
echo new > "$ws/MESSAGING.md"
|
|
mkdir -p "$ws/.venv/lib" && echo junk > "$ws/.venv/lib/junk"
|
|
echo junk > "$ws/.mam/version.txt"
|
|
echo junk > "$ws/.mam/asset_hashes.txt"
|
|
mkdir -p "$ws/.agents/skills/half-installed" && echo x > "$ws/.agents/skills/half-installed/f"
|
|
python3 -c 'import yaml,sys; p=sys.argv[1]; d=yaml.safe_load(open(p)); d["herdr_sessions"]=[]; yaml.safe_dump(d, open(p, "w"))' "$ws/.mam/agent-sessions.yaml"
|
|
exit 7
|
|
"""
|
|
|
|
|
|
def test_a_failed_install_rolls_everything_back(box):
|
|
write_rows(box.ws, [row("a", status="stopped")])
|
|
db = box.mam / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA wal_autocheckpoint=0")
|
|
conn.execute("CREATE TABLE state (id INTEGER PRIMARY KEY, data TEXT)")
|
|
conn.execute("CREATE TABLE sessions (name TEXT PRIMARY KEY, status TEXT, pane_cwd TEXT, data JSON)")
|
|
conn.execute("INSERT INTO state VALUES (1, ?)", (json.dumps({"orchestrator_uuids": ["in-wal"]}),))
|
|
conn.execute("INSERT INTO sessions VALUES ('a', 'stopped', '/x', ?)", (json.dumps(row("a", status="stopped")),))
|
|
conn.commit()
|
|
assert (box.mam / "agent-sessions.db-wal").stat().st_size > 0, "the commit is still in the WAL"
|
|
box.stub_monitor()
|
|
box.start_monitor()
|
|
box.installer(FAILING_INSTALL)
|
|
before = snapshot(box.ws)
|
|
try:
|
|
res = box.update()
|
|
finally:
|
|
conn.close()
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert "the install failed (rc 7); rolling back" in res.stderr and "rolled back" in res.stdout
|
|
after = snapshot(box.ws)
|
|
after.pop(".mam/agent-sessions.db", None)
|
|
before.pop(".mam/agent-sessions.db", None)
|
|
before.pop(".mam/agent-sessions.db-wal", None)
|
|
before.pop(".mam/agent-sessions.db-shm", None)
|
|
assert after == before, sorted(set(after.items()) ^ set(before.items()))[:10]
|
|
assert not (box.ws / "MESSAGING.md").exists() and not (box.ws / ".venv").exists()
|
|
check = sqlite3.connect(str(db))
|
|
try:
|
|
st = json.loads(check.execute("SELECT data FROM state WHERE id=1").fetchone()[0])
|
|
finally:
|
|
check.close()
|
|
assert st == {"orchestrator_uuids": ["in-wal"]}, "the backup carried the WAL's commit"
|
|
assert not (box.mam / "update-backup").exists() and not (box.mam / "update-in-progress").exists()
|
|
assert box.monitor_pids() and box.monitor_starts() == 2, "the restored tree starts the monitor again"
|
|
|
|
box.env["MAM_INSTALLER_URL"] = "file://" + str(REPO_ROOT / "deploy" / "install.sh")
|
|
res = box.update()
|
|
assert res.returncode == 0, "the skill runs again after a rollback: " + res.stdout + res.stderr
|
|
|
|
|
|
# --- leftovers (U4) ------------------------------------------------------------------
|
|
|
|
def _leftover(box: Box, state):
|
|
bk = box.mam / "update-backup"
|
|
(bk / "files" / ".agents" / "skills").mkdir(parents=True)
|
|
shutil.copy2(box.ws / ".agents/skills/lib.sh", bk / "files/.agents/skills/lib.sh")
|
|
(bk / "items").write_text(".agents/skills/lib.sh\nMESSAGING.md\n")
|
|
(bk / "absent").write_text("MESSAGING.md\n")
|
|
(bk / "registry").mkdir()
|
|
(bk / "registry" / "agent-sessions.yaml").write_text("herdr_sessions: []\n")
|
|
if state is not None:
|
|
(bk / "state").write_text(state + "\n")
|
|
return bk
|
|
|
|
|
|
@pytest.mark.parametrize("state", [None, "backing-up", "installed"])
|
|
def test_leftover_that_changed_nothing_or_finished_is_removed(box, state):
|
|
_leftover(box, state)
|
|
(box.ws / "MESSAGING.md").write_text("must survive: nothing is restored\n")
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "removing the backup a previous update left" in res.stdout
|
|
assert (box.ws / "MESSAGING.md").exists()
|
|
assert not (box.mam / "update-backup").exists()
|
|
|
|
|
|
def test_leftover_installing_restores_files_but_not_the_registry(box):
|
|
_leftover(box, "installing")
|
|
pristine = (box.ws / ".agents/skills/lib.sh").read_bytes()
|
|
with open(box.ws / ".agents/skills/lib.sh", "a") as fh:
|
|
fh.write("# half installed\n")
|
|
(box.ws / "MESSAGING.md").write_text("made by the dead install\n")
|
|
write_rows(box.ws, [row("written-after-crash", status="stopped")])
|
|
box.installer("grep -q 'half installed' \"$ws/.agents/skills/lib.sh\" && exit 9\n"
|
|
"test -e \"$ws/MESSAGING.md\" && exit 8\n"
|
|
"cp \"$ws/.mam/agent-sessions.yaml\" \"$ws/../seen-registry.yaml\"\nexit 0\n")
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "the previous update was rolled back" in res.stdout
|
|
assert (box.ws / ".agents/skills/lib.sh").read_bytes() == pristine
|
|
assert "written-after-crash" in (box.root / "seen-registry.yaml").read_text(), "registry left as it was"
|
|
assert not (box.mam / "update-backup").exists()
|
|
|
|
|
|
def test_leftover_whose_restore_fails_is_kept_and_nothing_more_changes(box):
|
|
bk = _leftover(box, "installing")
|
|
(bk / "items").write_text(".agents/skills/lib.sh\n.agents/skills/missing-from-backup\n")
|
|
box.stub_monitor()
|
|
box.start_monitor()
|
|
box.installer("touch \"$ws/INSTALLER-RAN\"\n")
|
|
res = box.update()
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert "restoring the previous update's backup failed" in res.stderr
|
|
assert bk.exists() and (bk / "state").read_text().strip() == "installing"
|
|
assert not (box.ws / "INSTALLER-RAN").exists()
|
|
assert not box.monitor_pids(), "the monitor stays stopped"
|
|
assert not (box.mam / "update-in-progress").exists()
|
|
|
|
|
|
def test_a_state_write_failure_before_the_install_changes_nothing_and_restarts_the_monitor(box):
|
|
real_mv = shutil.which("mv")
|
|
fake = box.bin / "mv"
|
|
fake.write_text("#!/usr/bin/env bash\n"
|
|
"src=\"${@: -2:1}\"\n"
|
|
"case \"$src\" in */update-backup/state.tmp) grep -qx installing \"$src\" && exit 1 ;; esac\n"
|
|
"exec %s \"$@\"\n" % real_mv)
|
|
fake.chmod(0o755)
|
|
box.stub_monitor()
|
|
box.start_monitor()
|
|
box.installer("touch \"$ws/INSTALLER-RAN\"\n")
|
|
res = box.update()
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert not (box.ws / "INSTALLER-RAN").exists()
|
|
assert box.monitor_starts() == 2, "the monitor it stopped is started again"
|
|
_until(lambda: bool(box.monitor_pids()))
|
|
assert not (box.mam / "update-backup").exists() and not (box.mam / "update-in-progress").exists()
|
|
assert "nothing was changed" in res.stderr
|
|
|
|
|
|
def test_the_backup_lives_in_the_workspace_and_is_gone_after_success(box):
|
|
box.installer("test -f \"$ws/.mam/update-backup/state\" && grep -qx installing \"$ws/.mam/update-backup/state\" "
|
|
"&& test -f \"$ws/.mam/update-backup/files/.agents/skills/lib.sh\" && touch \"$ws/../saw-backup\"\n")
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert (box.root / "saw-backup").exists()
|
|
assert not (box.mam / "update-backup").exists()
|
|
|
|
|
|
# --- the marker and the admission lock (decision 8) -----------------------------------
|
|
|
|
def test_a_live_update_marker_refuses_a_second_update_and_is_left_alone(box):
|
|
live_marker(box.mam / "update-in-progress", box)
|
|
text = (box.mam / "update-in-progress").read_text()
|
|
before = snapshot(box.ws)
|
|
res = box.update()
|
|
assert res.returncode == 1 and "another update is in progress" in res.stderr, res.stderr
|
|
assert (box.mam / "update-in-progress").read_text() == text
|
|
assert snapshot(box.ws) == before
|
|
|
|
|
|
def test_a_stale_update_marker_is_taken_over(box):
|
|
dead_marker(box.mam / "update-in-progress")
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "stale update marker" in res.stdout
|
|
assert not (box.mam / "update-in-progress").exists()
|
|
|
|
|
|
def test_two_updates_at_once_one_proceeds(box):
|
|
box.installer("touch \"$ws/../first-installing\"; sleep 4; exit 0\n")
|
|
first = box.helper("bash", str(box.ws / UPDATE_REL), "--yes", cwd=str(box.ws), env=box.env,
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
_until(lambda: (box.root / "first-installing").exists())
|
|
marker = (box.mam / "update-in-progress").read_text()
|
|
second = box.update()
|
|
assert second.returncode == 1 and "another update is in progress" in second.stderr, second.stderr
|
|
assert (box.mam / "update-in-progress").read_text() == marker, "the running update's marker is untouched"
|
|
out, err = first.communicate(timeout=120)
|
|
assert first.returncode == 0, out + err
|
|
|
|
|
|
def test_entries_are_refused_while_the_copy_installs(box):
|
|
box.installer('if bash -c \'source "$1/.agents/skills/lib.sh" && mam_update_admit\' _ "$ws" 2> "$ws/../admit.err"; '
|
|
'then echo admitted > "$ws/../admit"; else echo refused > "$ws/../admit"; fi\n')
|
|
res = box.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert (box.root / "admit").read_text().strip() == "refused"
|
|
assert "an update of %s is in progress" % box.ws in (box.root / "admit.err").read_text()
|
|
|
|
|
|
@pytest.mark.parametrize("wait,needle", [("30", "session late-row is running"), ("1", "admission lock past")])
|
|
def test_an_entry_holding_the_admission_lock_is_waited_for_or_refuses(box, wait, needle):
|
|
script = ('mam_update_admit || exit 1\necho admitted\nsleep 3\n'
|
|
'printf "herdr_sessions:\\n- name: late-row\\n status: running\\n" > "$WORKSPACE_ROOT/.mam/agent-sessions.yaml"\n'
|
|
'mam_update_admit_release\nsleep 30\n')
|
|
entry = box.helper("bash", "-c", 'source "$1/.agents/skills/lib.sh"\n' + script, "_", str(box.ws),
|
|
env=dict(box.env, WORKSPACE_ROOT=str(box.ws)), stdout=subprocess.PIPE, text=True)
|
|
assert entry.stdout.readline().strip() == "admitted"
|
|
box.installer("touch \"$ws/INSTALLER-RAN\"\n")
|
|
res = box.update(env={"MAM_UPDATE_ADMISSION_WAIT_S": wait})
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
assert needle in res.stderr, res.stderr
|
|
assert not (box.ws / "INSTALLER-RAN").exists()
|
|
assert not (box.mam / "update-in-progress").exists()
|
|
|
|
|
|
# --- the first transition (decision 4, criterion 5) ------------------------------------
|
|
|
|
def test_the_old_updater_installs_the_skill_which_then_converts_the_registry(tmp_path):
|
|
"""A workspace installed by v4.11.0 still has .mam_deploy/update.sh. Run once,
|
|
it installs the new version - the skill, and no updater left behind - and
|
|
copies the registry unconverted; the skill run right after converts it."""
|
|
old_src = tmp_path / "v4.11.0"
|
|
old_src.mkdir()
|
|
arch = subprocess.run(["git", "-C", str(REPO_ROOT), "archive", "0e1e2e8"], capture_output=True)
|
|
if arch.returncode != 0:
|
|
pytest.skip("no git history to rebuild v4.11.0 from")
|
|
subprocess.run(["tar", "-x", "-C", str(old_src)], input=arch.stdout, check=True)
|
|
root = Path(os.path.realpath(tmp_path))
|
|
ws = root / "ws"
|
|
ws.mkdir()
|
|
(root / "bin").mkdir()
|
|
(root / "bin" / "herdr").write_text(FAKE_HERDR)
|
|
(root / "bin" / "herdr").chmod(0o755)
|
|
b = Box(root, ws)
|
|
res = subprocess.run(["bash", str(old_src / "deploy" / "install.sh"), str(ws)],
|
|
env=dict(b.env, MAM_REPO_URL=str(old_src)), capture_output=True, text=True, timeout=600)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert (ws / ".mam_deploy" / "update.sh").exists()
|
|
write_rows(ws, [row("old-one", status="stopped", kill_command="x", start_command="y")])
|
|
|
|
res = subprocess.run(["bash", str(ws / ".mam_deploy" / "update.sh"), "--force"], cwd=str(ws), env=b.env,
|
|
capture_output=True, text=True, timeout=600, stdin=subprocess.DEVNULL)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert (ws / UPDATE_REL).exists() and not (ws / ".mam_deploy" / "update.sh").exists()
|
|
assert "registry_format" not in yaml.safe_load((ws / ".mam" / "agent-sessions.yaml").read_text())
|
|
|
|
res = b.update()
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
assert "registry-format: 1 -> 2, rows changed: 1" in res.stdout
|
|
assert yaml.safe_load((ws / ".mam" / "agent-sessions.yaml").read_text())["registry_format"] == 2
|