GodopuandClaude Opus 5.5
e4c7c8fe9b
- attach_command is `herdr --session <server> agent attach <herdr name>`
(`default` written too), built by lib_py.agents.sanitize.attach_command;
create, resume and reconcile all use it. kill_command and start_command
are no longer written, and status --json drops start_command.
- The two raw kill-session hints (create exit 43, id_order's orphan) print a
stop-skill command with --herdr-session instead.
- stop_session.sh closes a session that has no registry row, herdr only and
without --agent, when herdr runs it inside this workspace; another
workspace's session is refused (exit 5), a missing one is exit 1, one that
survives kill-session is exit 6. The registry is not changed.
- Docs: only the SKILL.md lines these changes made wrong; the skill block
pins follow. Tests that asserted the removed fields keep their intent
(issue50: create's own row; issue21: no control-root path in
attach_command; tier2: the new form and the absent fields). S11 pins the
shim to 4243da5, which changed it on purpose.
Refs #71
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
448 lines
21 KiB
Python
448 lines
21 KiB
Python
"""Issue #71: herdr agent names in the registry and the lib.sh shim.
|
|
|
|
- attach_command is `herdr --session <server> agent attach <herdr name>` 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"
|