"""Issue #71: herdr agent names in the registry and the lib.sh shim. - attach_command is `herdr --session agent attach ` from one builder (create, resume, reconcile); kill_command/start_command are gone. - the shim's `agent attach` resolves its target like get/read/prompt. - one bash sanitize rule, equal to the Python one, with no silent wrong name. - an old registry is migrated by the writer (YAML and DB), with an entry point. - stop closes a session that has no registry row, only inside this workspace. - stall_watch works out the herdr name once. """ import hashlib import json import os import shutil import sqlite3 import subprocess import sys from pathlib import Path import pytest import yaml REPO = Path(__file__).resolve().parent.parent SKILLS = REPO / ".agents" / "skills" sys.path.insert(0, str(SKILLS)) from lib_py.agents.sanitize import sanitize_herdr_agent_name # noqa: E402 LONG = "lab-issue71-a-very-long-branch-name-creator-claude" # > 32 characters UPPER = "Lab-Upper-creator-claude" PLAIN = "lab-plain-creator-claude" # needs no conversion STOP = "bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh" def test_the_names_need_converting_and_the_plain_one_does_not(): assert len(LONG) > 32 and sanitize_herdr_agent_name(LONG) != LONG assert sanitize_herdr_agent_name(UPPER) == UPPER.lower() assert sanitize_herdr_agent_name(PLAIN) == PLAIN @pytest.mark.parametrize("name", [LONG, UPPER, PLAIN]) @pytest.mark.parametrize("server", ["default", "mam-x", None]) def test_attach_command_names_the_server_and_the_herdr_name(name, server): from lib_py.agents.sanitize import attach_command want = "herdr --session %s agent attach %s" % (server or "default", sanitize_herdr_agent_name(name)) assert attach_command(server, name) == want def test_attach_command_quotes_a_server_that_needs_it(): from lib_py.agents.sanitize import attach_command assert attach_command("my server", PLAIN) == "herdr --session 'my server' agent attach " + PLAIN # --- writers ----------------------------------------------------------------- def _rows(sandbox): return yaml.safe_load((sandbox / ".mam" / "agent-sessions.yaml").read_text())["herdr_sessions"] def _db_rows(sandbox): conn = sqlite3.connect(str(sandbox / ".mam" / "agent-sessions.db")) try: return {r[0]: json.loads(r[1]) for r in conn.execute("SELECT name, data FROM sessions")} finally: conn.close() def _set_mock(mock_herdr, **kw): state = json.loads(Path(mock_herdr).read_text()) state.update(kw) Path(mock_herdr).write_text(json.dumps(state)) def _assert_new_form(row, server): assert row["attach_command"] == "herdr --session %s agent attach %s" % ( server, sanitize_herdr_agent_name(row["name"])) assert "kill_command" not in row and "start_command" not in row @pytest.mark.parametrize("name", [LONG, UPPER, PLAIN]) @pytest.mark.parametrize("server", ["default", "mam-iso71"]) def test_create_writes_the_attach_command(mam_sandbox, mock_herdr, mock_agents, name, server): _set_mock(mock_herdr, strict_agent_names=True) script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" res = subprocess.run(["bash", str(script), "--workspace", str(mam_sandbox), "--agent", "claude", "--role", "creator", "--session", name, "--herdr-session", server, "--no-onboard"], capture_output=True, text=True, cwd=str(mam_sandbox), timeout=300) assert res.returncode == 0, res.stdout + res.stderr (row,) = [r for r in _rows(mam_sandbox) if r["name"] == name] _assert_new_form(row, server) agents = json.loads(Path(mock_herdr).read_text())["agents"] assert sanitize_herdr_agent_name(name) in agents, "the attach name is the one herdr registered" def _seed(sandbox, rows): lib = sandbox / ".agents" / "skills" / "lib.sh" mutation = "d['herdr_sessions'] = " + repr(rows) + "\n" res = subprocess.run(["bash", "-c", 'source "$1" && atomic_dump_yaml "$AGENT_SESSIONS_YAML"', "_", str(lib)], input=mutation, capture_output=True, text=True) assert res.returncode == 0, res.stderr def _resume_update(sandbox, name, server): script = sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "update_yaml_resumed.sh" return subprocess.run(["bash", str(script), "--session", name, "--uuid", "11111111-2222-3333-4444-555555555555", "--agent", "claude", "--herdr-session", server, "--workspace", str(sandbox)], capture_output=True, text=True, cwd=str(sandbox), timeout=120) @pytest.mark.parametrize("name", [LONG, UPPER, PLAIN]) @pytest.mark.parametrize("server", ["default", "mam-iso71"]) @pytest.mark.parametrize("has_row", [True, False]) def test_resume_writes_the_attach_command(mam_sandbox, mock_herdr, mock_agents, name, server, has_row): if has_row: _seed(mam_sandbox, [{"name": name, "status": "stopped", "role": "creator", "herdr_session": "old-srv", "herdr_server": "old-srv", "pane": {"cwd": str(mam_sandbox), "cmd": "claude"}}]) res = _resume_update(mam_sandbox, name, server) assert res.returncode == 0, res.stdout + res.stderr (row,) = [r for r in _rows(mam_sandbox) if r["name"] == name] assert row["status"] == "running" _assert_new_form(row, server) @pytest.mark.parametrize("name", [LONG, UPPER, PLAIN]) def test_reconcile_registers_with_the_attach_command(mam_sandbox, mock_herdr, mock_agents, name): # The pane must exist, or reconcile never reaches the registration. _set_mock(mock_herdr, agents={name: { "agent": "claude", "status": "running", "cwd": str(mam_sandbox), "workspace_id": "w1", "pid": 9999, "pane_id": "w1:p71", "command": "claude", "buffer": "Ready"}}) assert not any(r["name"] == name for r in _rows(mam_sandbox)) script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" res = subprocess.run(["bash", str(script), "--once"], capture_output=True, text=True, cwd=str(mam_sandbox), timeout=300) assert res.returncode == 0, res.stdout + res.stderr rows = [r for r in _rows(mam_sandbox) if r["name"] == name] assert len(rows) == 1, "reconcile wrote no row: " + res.stdout + res.stderr assert rows[0]["last_visible_note"] == "auto-registered by monitor" _assert_new_form(rows[0], rows[0]["herdr_session"]) # --- shim: agent attach -------------------------------------------------------- def _run_lib(sandbox, body, env=None): lib = sandbox / ".agents" / "skills" / "lib.sh" return subprocess.run(["bash", "-c", 'source "$1"\n' + body, "_", str(lib)], capture_output=True, text=True, cwd=str(sandbox), env=dict(os.environ, **(env or {})), timeout=120) @pytest.mark.parametrize("name", [LONG, UPPER, PLAIN]) def test_wrapper_attach_calls_herdr_with_the_herdr_name(mam_sandbox, mock_herdr, name): herdr_name = sanitize_herdr_agent_name(name) _set_mock(mock_herdr, strict_agent_names=True, agents={herdr_name: { "agent": "claude", "status": "running", "cwd": str(mam_sandbox), "workspace_id": "w1", "pid": 9999, "pane_id": "w1:p71", "command": "claude", "buffer": "Ready"}}) res = _run_lib(mam_sandbox, 'herdr agent attach "$0" --takeover'.replace("$0", name)) calls = [c for c in json.loads(Path(mock_herdr).read_text())["calls"] if c[:2] == ["agent", "attach"]] assert calls == [["agent", "attach", herdr_name, "--takeover"]], (calls, res.stderr) # --- one sanitize rule ---------------------------------------------------------- NAMES = ["", "plain", "Upper-Case", "9lead", "-dash", "has space/and.dot", "x" * 32, "y" * 33, LONG, LONG.upper(), "creator-claude-fix-issue-99-long-branch"] def test_lib_sh_has_one_definition(): text = (SKILLS / "lib.sh").read_text() assert text.count("_sanitize_herdr_agent_name() {") == 1 assert "00000000" not in text assert "lambda s: s.lower()[:32]" not in text def test_the_generated_shim_carries_the_same_definition(mam_sandbox): res = _run_lib(mam_sandbox, "declare -f _sanitize_herdr_agent_name") assert res.returncode == 0, res.stderr shim = (mam_sandbox / ".mam" / "shim" / "herdr").read_text() assert shim.startswith("#!/usr/bin/env bash\n") assert shim.count("_sanitize_herdr_agent_name ()") == 1 assert res.stdout.strip() in shim def _definition(tmp_path): """The function lib.sh defines, as bash prints it (sourced with a tmp workspace).""" res = subprocess.run(["bash", "-c", 'source "$1" >/dev/null 2>&1; declare -f _sanitize_herdr_agent_name', "_", str(SKILLS / "lib.sh")], capture_output=True, text=True, env=dict(os.environ, WORKSPACE_ROOT=str(tmp_path / "defn-ws"))) assert res.returncode == 0 and res.stdout, res.stderr return res.stdout def _tools_dir(tmp_path, tools): d = tmp_path / ("bin-" + "-".join(tools)) d.mkdir() for t in tools: src = shutil.which(t) assert src, t (d / t).symlink_to(src) return d def _sanitize_with_path(defn, path_dir, name): bash = shutil.which("bash") return subprocess.run([bash, "--noprofile", "--norc", "-c", defn + '\n_sanitize_herdr_agent_name "$1"', "_", name], capture_output=True, text=True, env={"PATH": str(path_dir), "LC_ALL": "C"}) @pytest.mark.parametrize("hashers", [["shasum"], ["sha1sum"], ["openssl"], ["python3"]]) def test_each_hasher_alone_matches_python(tmp_path, hashers): if not shutil.which(hashers[0]): pytest.skip(hashers[0] + " not installed") defn = _definition(tmp_path) d = _tools_dir(tmp_path, ["tr", "sed", "awk", "cut"] + hashers) for name in NAMES: res = _sanitize_with_path(defn, d, name) assert res.returncode == 0, (name, res.stderr) assert res.stdout == sanitize_herdr_agent_name(name) + "\n", name def test_no_hasher_at_all_is_an_error_not_a_name(tmp_path): defn = _definition(tmp_path) d = _tools_dir(tmp_path, ["tr", "sed", "awk", "cut"]) res = _sanitize_with_path(defn, d, LONG) assert res.returncode != 0 and res.stdout == "" and "ERROR" in res.stderr # a short name needs no hash and still works res = _sanitize_with_path(defn, d, UPPER) assert res.returncode == 0 and res.stdout == UPPER.lower() + "\n" def _no_hash_bin(tmp_path): """Stubs: shasum/sha1sum/openssl fail; python3 runs, except the name hash.""" d = tmp_path / "nohash" d.mkdir() for t in ("shasum", "sha1sum", "openssl"): (d / t).write_text("#!/bin/sh\nexit 1\n") (d / t).chmod(0o755) (d / "python3").write_text('#!/bin/sh\ncase "$*" in *hashlib.sha1*) exit 1 ;; esac\nexec %s "$@"\n' % shutil.which("python3")) (d / "python3").chmod(0o755) return d def test_spawn_does_not_go_on_without_the_name(mam_sandbox, mock_herdr, tmp_path): d = _no_hash_bin(tmp_path) body = 'export PATH="%s:$PATH"\nherdr new-session -d -s "%s" -c "%s" claude' % (d, LONG, mam_sandbox) res = _run_lib(mam_sandbox, body) assert res.returncode != 0, res.stdout + res.stderr assert "cannot make the herdr name" in res.stderr calls = json.loads(Path(mock_herdr).read_text())["calls"] assert not [c for c in calls if c[:2] in (["agent", "start"], ["pane", "split"], ["workspace", "create"])], calls def test_lookups_report_failure_without_the_name(mam_sandbox, mock_herdr, tmp_path): d = _no_hash_bin(tmp_path) for verb in ('has-session -t "%s"', 'agent get "%s"', 'kill-session -t "%s"'): res = _run_lib(mam_sandbox, 'export PATH="%s:$PATH"\nherdr %s' % (d, verb % LONG)) assert res.returncode != 0 and "cannot make the herdr name" in res.stderr, (verb, res.stderr) # --- registry migration ---------------------------------------------------------- def _old_rows(ws): def row(name, status, srv): return {"name": name, "status": status, "role": "creator", "herdr_session": srv, "herdr_server": srv, "pane": {"index": 0, "pid": 1, "cmd": "claude", "cwd": str(ws)}, "start_command": "HERDR_SESSION_NAME=%s herdr new-session -d -s %s" % (srv, name), "attach_command": "HERDR_SESSION_NAME=%s herdr agent attach %s" % (srv, name), "kill_command": "HERDR_SESSION_NAME=%s herdr kill-session -t %s" % (srv, name)} return [row(LONG, "running", "mam-x"), row(UPPER, "stopped", "default"), row(PLAIN, "terminated", "mam-y")] def _write_old_registry(ws, with_db): mam = ws / ".mam" mam.mkdir(parents=True, exist_ok=True) top = {"orchestrator_uuids": ["u-1"], "snapshot": {"taken_at": "2026-10-01T00:00:00Z"}} d = dict(top, herdr_sessions=_old_rows(ws)) (mam / "agent-sessions.yaml").write_text(yaml.safe_dump(d, sort_keys=False)) if with_db: conn = sqlite3.connect(str(mam / "agent-sessions.db")) 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(top),)) for s in d["herdr_sessions"]: conn.execute("INSERT INTO sessions VALUES (?, ?, ?, ?)", (s["name"], s["status"], s["pane"]["cwd"], json.dumps(s))) conn.commit() conn.close() return mam / "agent-sessions.yaml" def _entry(yaml_path): return subprocess.run([sys.executable, "-m", "lib_py.registry_format", "--yaml", str(yaml_path)], capture_output=True, text=True, env=dict(os.environ, PYTHONPATH=str(SKILLS)), timeout=60) def _state(ws): conn = sqlite3.connect(str(ws / ".mam" / "agent-sessions.db")) try: return json.loads(conn.execute("SELECT data FROM state WHERE id=1").fetchone()[0]) finally: conn.close() def test_migrate_is_pure_and_idempotent(tmp_path): from lib_py import registry_format old = {"herdr_sessions": _old_rows(tmp_path), "snapshot": {}} before = json.dumps(old, sort_keys=True) new, n = registry_format.migrate(old) assert json.dumps(old, sort_keys=True) == before, "the input is not modified" assert n == 3 and new[registry_format.FORMAT_KEY] == registry_format.CURRENT == 2 for s in new["herdr_sessions"]: _assert_new_form(s, s["herdr_session"]) again, n2 = registry_format.migrate(new) assert n2 == 0 and again == new assert registry_format.stored_version({}) == 1 @pytest.mark.parametrize("with_db", [True, False]) def test_entry_point_migrates_yaml_and_db_once(tmp_path, with_db): ws = tmp_path / "ws" yaml_path = _write_old_registry(ws, with_db) res = _entry(yaml_path) assert res.returncode == 0, res.stderr assert res.stdout == "registry-format: 1 -> 2, rows changed: 3\n" y = yaml.safe_load(yaml_path.read_text()) db = _db_rows(ws) assert y["registry_format"] == 2 and _state(ws)["registry_format"] == 2 assert y["orchestrator_uuids"] == ["u-1"] and y["snapshot"]["taken_at"] == "2026-10-01T00:00:00Z" assert {s["name"]: s for s in y["herdr_sessions"]} == db, "YAML and DB agree" for s in y["herdr_sessions"]: _assert_new_form(s, s["herdr_session"]) assert {s["name"]: s["status"] for s in y["herdr_sessions"]} == {LONG: "running", UPPER: "stopped", PLAIN: "terminated"} text = yaml_path.read_text() res = _entry(yaml_path) assert res.returncode == 0, res.stderr assert res.stdout == "registry-format: 2 -> 2, rows changed: 0\n" assert yaml_path.read_text() == text and _db_rows(ws) == db def test_any_write_migrates_and_rewrites_the_yaml(own_workspace_root, tmp_path): """A mutation that changes nothing still leaves YAML and DB in the new form (point A).""" ws = tmp_path / "ws" yaml_path = _write_old_registry(ws, with_db=True) res = subprocess.run(["bash", "-c", 'source "$1" && atomic_dump_yaml "$2"', "_", str(SKILLS / "lib.sh"), str(yaml_path)], input="pass\n", capture_output=True, text=True, timeout=60) assert res.returncode == 0, res.stderr y = yaml.safe_load(yaml_path.read_text()) assert y["registry_format"] == 2 assert {s["name"]: s for s in y["herdr_sessions"]} == _db_rows(ws) for s in y["herdr_sessions"]: _assert_new_form(s, s["herdr_session"]) def test_a_new_registry_is_written_current(own_workspace_root, tmp_path): yaml_path = tmp_path / "fresh" / ".mam" / "agent-sessions.yaml" yaml_path.parent.mkdir(parents=True) res = subprocess.run(["bash", "-c", 'source "$1" && atomic_dump_yaml "$2"', "_", str(SKILLS / "lib.sh"), str(yaml_path)], input="d['herdr_sessions'] = []\n", capture_output=True, text=True, timeout=60) assert res.returncode == 0, res.stderr assert yaml.safe_load(yaml_path.read_text())["registry_format"] == 2 def test_entry_point_fails_without_a_registry(tmp_path): res = _entry(tmp_path / "none" / "agent-sessions.yaml") assert res.returncode != 0 and "error" in res.stderr and res.stdout == "" # --- stop: a session with no registry row ---------------------------------------- def _registry_bytes(sandbox): out = {} for f in ("agent-sessions.yaml", "agent-sessions.db"): p = sandbox / ".mam" / f out[f] = hashlib.sha256(p.read_bytes()).hexdigest() if p.exists() else None return out def _stop(sandbox, name, *extra): script = sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh" return subprocess.run(["bash", str(script), "--session", name, "--herdr-session", "default"] + list(extra), capture_output=True, text=True, cwd=str(sandbox), timeout=120) def _agent(cwd, pane): return {"agent": "claude", "status": "running", "cwd": str(cwd), "workspace_id": "w1", "pid": 9999, "pane_id": pane, "command": "claude", "buffer": "Ready"} @pytest.fixture def registry_with_a_row(mam_sandbox): _seed(mam_sandbox, [{"name": "other-creator-claude", "status": "stopped", "role": "creator", "pane": {"cwd": str(mam_sandbox), "cmd": "claude"}}]) return _registry_bytes(mam_sandbox) def test_stop_closes_a_rowless_session_of_this_workspace(mam_sandbox, mock_herdr, registry_with_a_row): # no agent suffix and no row: there is no agent kind, and none is needed _set_mock(mock_herdr, agents={"orphan71": _agent(mam_sandbox / "sub", "w1:p71")}) res = _stop(mam_sandbox, "orphan71") assert res.returncode == 0, res.stdout + res.stderr assert "orphan71" not in json.loads(Path(mock_herdr).read_text())["agents"] assert _registry_bytes(mam_sandbox) == registry_with_a_row def test_stop_refuses_a_rowless_session_of_another_workspace(mam_sandbox, mock_herdr, registry_with_a_row, tmp_path_factory): elsewhere = tmp_path_factory.mktemp("elsewhere") _set_mock(mock_herdr, agents={"foreign71-creator-claude": _agent(elsewhere, "w1:p72")}) res = _stop(mam_sandbox, "foreign71-creator-claude") assert res.returncode == 5, res.stdout + res.stderr assert "outside this workspace" in res.stderr assert "foreign71-creator-claude" in json.loads(Path(mock_herdr).read_text())["agents"] calls = json.loads(Path(mock_herdr).read_text())["calls"] assert not [c for c in calls if c[:2] == ["pane", "close"]], calls assert _registry_bytes(mam_sandbox) == registry_with_a_row def test_stop_reports_a_rowless_session_that_is_not_there(mam_sandbox, mock_herdr, registry_with_a_row): _set_mock(mock_herdr, strict_agent_names=True) res = _stop(mam_sandbox, "absent71") assert res.returncode == 1, res.stdout + res.stderr assert "has no agent of that name" in res.stderr assert _registry_bytes(mam_sandbox) == registry_with_a_row def test_stop_without_a_row_purges_nothing(mam_sandbox, mock_herdr, registry_with_a_row): _set_mock(mock_herdr, agents={"orphan72": _agent(mam_sandbox, "w1:p73")}) res = _stop(mam_sandbox, "orphan72", "--purge-conversation", "--yes") assert res.returncode == 1, res.stdout + res.stderr assert "orphan72" in json.loads(Path(mock_herdr).read_text())["agents"] assert _registry_bytes(mam_sandbox) == registry_with_a_row # --- stall_watch ----------------------------------------------------------------- def test_stall_watch_works_out_the_herdr_name_once(monkeypatch): from lib_py import stall_watch seen = [] real = stall_watch.sanitize_herdr_agent_name monkeypatch.setattr(stall_watch, "sanitize_herdr_agent_name", lambda n: seen.append(n) or real(n)) class Args(object): target, agent, herdr_session, registry_dir, job = LONG, None, "mam-x", "/nonexistent", "j" sw = stall_watch.StallWatch(Args()) assert seen == [LONG] assert sw.herdr_name == sw.src.name == sanitize_herdr_agent_name(LONG) assert sw.src.target == LONG and sw.src.agent == "claude"