multi-agent-mux/tests/test_issue71_herdr_names.py

"""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
UNI = "\u0130stanbul-creator-claude"                           # str.lower() differs from tr here
NL = "newline-creator-claude\n"                                 # $(...) would drop the line break
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, UNI, NL])
@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", "-n", "-e", "a\\nb",
         NL, "a\nb", "\n", "x\r\ny", "Up\nper"]
UNI_NAMES = [UNI, "caf\u00e9-creator-claude", "\u00fc" * 40, "kelvin-\u212a", "\u65e5\u672c-creator-claude",
             "\u0130" * 20 + "-creator-claude", "\u00fc\n"]


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


@pytest.mark.parametrize("hashers", [["python3"], ["shasum", "python3"]])
def test_a_name_that_is_not_ascii_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 UNI_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_a_name_that_is_not_ascii_needs_python3(tmp_path):
    defn = _definition(tmp_path)
    d = _tools_dir(tmp_path, ["tr", "sed", "awk", "cut"])
    for name in UNI_NAMES:
        res = _sanitize_with_path(defn, d, name)
        assert res.returncode != 0 and res.stdout == "" and "not ASCII" in res.stderr, (name, res.stderr)


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"])


@pytest.mark.parametrize("retry", ["entry-point", "any-write"])
def test_a_retry_repairs_the_yaml_a_failed_migration_left_behind(own_workspace_root, tmp_path, monkeypatch, retry):
    """The DB is committed before the YAML is replaced; if that fails, the next run rewrites the YAML."""
    from lib_py import registry_format
    ws = tmp_path / "ws"
    yaml_path = _write_old_registry(ws, with_db=True)
    real_replace, failed = os.replace, []

    def replace_once(src, dst, *a, **kw):
        if os.path.abspath(str(dst)) == str(yaml_path) and not failed:
            failed.append(dst)
            raise OSError("injected: YAML not replaced")
        return real_replace(src, dst, *a, **kw)

    monkeypatch.setenv("YAML_PATH", "")
    monkeypatch.setenv("AGENT_SESSIONS_MUTATION", "")
    monkeypatch.setattr(os, "replace", replace_once)
    assert registry_format.main(["--yaml", str(yaml_path)]) == 1 and failed
    monkeypatch.setattr(os, "replace", real_replace)
    assert _state(ws)["registry_format"] == 2
    assert registry_format.stored_version(yaml.safe_load(yaml_path.read_text())) == 1

    if retry == "entry-point":
        res = _entry(yaml_path)
    else:
        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), "YAML and DB agree"
    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, server="default", env=None):
    script = sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
    flag = ["--herdr-session", server] if server else []
    return subprocess.run(["bash", str(script), "--session", name] + flag + list(extra),
                          capture_output=True, text=True, cwd=str(sandbox), timeout=120,
                          env=dict(os.environ, **(env or {})))


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_uses_the_environments_server(mam_sandbox, mock_herdr, registry_with_a_row):
    """No --herdr-session: the server is HERDR_SESSION_NAME, as create chose it, not one derived from the workspace."""
    _set_mock(mock_herdr, agents={"orphan73": _agent(mam_sandbox, "w1:p74")}, calls=[])
    res = _stop(mam_sandbox, "orphan73", server=None, env={"HERDR_SESSION_NAME": "wanted-env-server"})
    assert res.returncode == 0, res.stdout + res.stderr
    assert "wanted-env-server" in res.stdout
    calls = json.loads(Path(mock_herdr).read_text())["calls"]
    # every call on an agent or pane names that server (`session list` and `config check` name none)
    targeted = [c for c in calls if "agent" in c[:3] or "pane" in c[:3]]
    assert targeted and all(c[:2] == ["--session", "wanted-env-server"] for c in targeted), calls
    assert ["--session", "wanted-env-server", "agent", "get", "orphan73"] in calls
    assert _registry_bytes(mam_sandbox) == registry_with_a_row


def test_stop_does_not_take_an_unreadable_row_for_no_row(mam_sandbox, mock_herdr):
    """Only "no such row" leads to the row-less path; a lookup that fails otherwise stops."""
    db = mam_sandbox / ".mam" / "agent-sessions.db"
    if db.exists():
        db.unlink()
    (mam_sandbox / ".mam" / "agent-sessions.yaml").write_text("herdr_sessions:\n- 42\n")
    _set_mock(mock_herdr, agents={"orphan74": _agent(mam_sandbox, "w1:p75")})
    res = _stop(mam_sandbox, "orphan74")
    assert res.returncode == 1, res.stdout + res.stderr
    assert "cannot read the row" in res.stderr
    assert "orphan74" in json.loads(Path(mock_herdr).read_text())["agents"]


@pytest.mark.parametrize("broken", ["yaml-without-db", "db"])
def test_stop_does_not_take_an_unreadable_registry_for_no_row(mam_sandbox, mock_herdr, broken):
    """load_state_json's {} on a read error is not "no row": a registered session is not closed row-less."""
    name = "orphan75-creator-claude"
    _seed(mam_sandbox, [{"name": name, "status": "running", "role": "creator", "agent": "claude",
                         "pane": {"cwd": str(mam_sandbox), "cmd": "claude"}}])
    mam = mam_sandbox / ".mam"
    if broken == "yaml-without-db":
        for f in mam.glob("agent-sessions.db*"):
            f.unlink()
        (mam / "agent-sessions.yaml").write_text("herdr_sessions: [\n  - {name: " + name + "\n")
    else:
        for f in mam.glob("agent-sessions.db-*"):
            f.unlink()
        (mam / "agent-sessions.db").write_bytes(b"this is not a database" * 64)
    _set_mock(mock_herdr, agents={name: _agent(mam_sandbox, "w1:p76")})
    res = _stop(mam_sandbox, name)
    assert res.returncode == 1, res.stdout + res.stderr
    assert "cannot read" in res.stderr and "no registry row" not in res.stdout
    assert name in json.loads(Path(mock_herdr).read_text())["agents"]


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"