"""Issue #72: the attach hints create, stop and the delegate-job wrapper print are the session row's attach_command (Issue #71); with no row, the one lib_py.agents.sanitize.attach_command builds for the server the script used. """ import json import os import re import subprocess import sys import time 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 attach_command, sanitize_herdr_agent_name # noqa: E402 LONG = "lab-issue72-a-very-long-branch-name-creator-claude" # > 32 characters UPPER = "Lab-Upper72-creator-claude" 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 _rows(sandbox): return yaml.safe_load((sandbox / ".mam" / "agent-sessions.yaml").read_text())["herdr_sessions"] 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 _row(name, server, cwd, pid=1): return {"name": name, "status": "running", "role": "creator", "agent": "claude", "herdr_session": server, "herdr_server": server, "attach_command": attach_command(server, name), "pane": {"index": 0, "pid": pid, "cmd": "claude", "cwd": str(cwd)}} def _seeded_attach(sandbox, name, server): (row,) = [r for r in _rows(sandbox) if r["name"] == name] assert row["attach_command"] == attach_command(server, name) return row["attach_command"] def _agent(cwd, pane, pid=9999): return {"agent": "claude", "status": "running", "agent_status": "idle", "cwd": str(cwd), "workspace_id": "w1", "pid": pid, "pane_id": pane, "command": "claude", "buffer": "Ready"} # --- create --------------------------------------------------------------------- @pytest.mark.parametrize("name", [LONG, UPPER]) def test_create_prints_the_rows_attach_command(mam_sandbox, mock_herdr, mock_agents, name): assert sanitize_herdr_agent_name(name) != name, "the herdr name differs from the session name" _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", "mam-iso72", "--no-onboard"], capture_output=True, text=True, cwd=str(mam_sandbox), timeout=300, env=dict(os.environ, MAM_NO_SUBSCRIBER_DAEMON="1")) assert res.returncode == 0, res.stdout + res.stderr (row,) = [r for r in _rows(mam_sandbox) if r["name"] == name] hints = [l for l in res.stdout.splitlines() if l.startswith("Attach:")] assert hints == ["Attach: " + row["attach_command"]], res.stdout assert row["attach_command"] == attach_command("mam-iso72", name) # --- stop: the stuck-TUI message ------------------------------------------------ def _undying_herdr(mam_sandbox): """The mock herdr behind a wrapper that ignores every way to close a pane, so the session survives stop's kill chain.""" bin_dir = mam_sandbox / "bin" (bin_dir / "herdr").rename(bin_dir / "herdr-mock72") (bin_dir / "herdr").write_text( '#!/usr/bin/env bash\nfor a in "$@"; do case "$a" in kill-session|close) exit 0 ;; esac; done\n' 'exec "%s" "$@"\n' % (bin_dir / "herdr-mock72")) (bin_dir / "herdr").chmod(0o755) @pytest.mark.parametrize("name", [LONG, UPPER]) def test_stop_stuck_tui_message_shows_the_rows_attach_command(mam_sandbox, mock_herdr, mock_agents, name): # The SIGKILL fallback kills the pane pid: make it a child of this test. pane = subprocess.Popen(["sleep", "300"]) try: # The row's server is not the one passed on the command line: the row wins. _seed(mam_sandbox, [_row(name, "mam-row72", mam_sandbox, pid=pane.pid)]) want = _seeded_attach(mam_sandbox, name, "mam-row72") _set_mock(mock_herdr, strict_agent_names=True, agents={sanitize_herdr_agent_name(name): _agent(mam_sandbox, "w1:p72", pid=pane.pid)}) _undying_herdr(mam_sandbox) script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh" res = subprocess.run(["bash", str(script), "--session", name, "--agent", "claude", "--herdr-session", "mam-arg72", "--purge-conversation", "--yes"], capture_output=True, text=True, cwd=str(mam_sandbox), timeout=300, env=dict(os.environ, SKS_IDLE_TIMEOUT_MS="500")) finally: pane.kill() pane.wait() assert res.returncode == 4, res.stdout + res.stderr hints = [l for l in res.stderr.splitlines() if "Diagnose the stuck TUI" in l] assert hints == [" Diagnose the stuck TUI (%s), then re-run" % want], res.stderr # --- delegate-job wrapper --------------------------------------------------------- def _delegate(mam_sandbox, mock_herdr, name): """Submit a job to herdr:, acknowledge it as the agent would, and return (stdout, the herdr server the wrapper prompted).""" reg_dir = mam_sandbox / ".mam" / "jobs" reg_dir.mkdir(parents=True, exist_ok=True) wrapper = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" publish = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py" env = dict(os.environ, DELEGATE_JOB_PYTHON=sys.executable, MQTT_HOST="127.0.0.1", MQTT_PORT="65534", SKS_IDLE_TIMEOUT_MS="500", MAM_INJECT_ACK_GRACE_S="20") proc = subprocess.Popen(["bash", str(wrapper), "submit", "--agent", "claude-code", "--agent-session", "herdr:" + name, "--prompt", "issue 72 attach hint", "--registry-dir", str(reg_dir), "--workdir", str(mam_sandbox), "--timeout", "6", "--idle-timeout", "2"], cwd=str(mam_sandbox), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: tok, jid = "", "" deadline = time.time() + 60 while time.time() < deadline and not tok: for f in reg_dir.glob("*.json"): try: tok, jid = json.loads(f.read_text()).get("dispatch_token") or "", f.stem except ValueError: pass time.sleep(0.2) assert tok, "the job was not dispatched" ack = subprocess.run([sys.executable, str(publish), "--registry-dir", str(reg_dir), "--job", jid, "--event", "started", "--expect-dispatch-token", tok, "--attempt-id", "1"], capture_output=True, text=True, cwd=str(mam_sandbox), env=env, timeout=60) # publish_event records `started` locally before its (here unreachable) broker publish assert '"started"' in (reg_dir / (jid + ".events.log")).read_text(), ack.stderr stdout, stderr = proc.communicate(timeout=180) finally: if proc.poll() is None: proc.kill() proc.communicate() calls = json.loads(Path(mock_herdr).read_text())["calls"] prompts = [c for c in calls if "agent" in c[:3] and "prompt" in c[:4]] assert prompts, (calls, stdout, stderr) servers = {c[1] if c[0] == "--session" else "default" for c in prompts} assert len(servers) == 1, prompts return stdout + stderr, servers.pop() def _delegate_hint(out): hints = [m.group(1) for m in (re.search(r"\(연결하려면: (.*)\)$", l) for l in out.splitlines()) if m] assert len(hints) == 1, out return hints[0] @pytest.mark.parametrize("name", [LONG, UPPER]) def test_delegate_prints_the_rows_attach_command(mam_sandbox, mock_herdr, mock_agents, name): _seed(mam_sandbox, [_row(name, "mam-row72", mam_sandbox)]) want = _seeded_attach(mam_sandbox, name, "mam-row72") _set_mock(mock_herdr, strict_agent_names=True, agents={sanitize_herdr_agent_name(name): _agent(mam_sandbox, "w1:p72")}) out, server = _delegate(mam_sandbox, mock_herdr, name) assert server == "mam-row72" assert _delegate_hint(out) == want @pytest.mark.parametrize("name", [LONG, UPPER]) def test_delegate_without_a_row_builds_it_for_the_server_it_used(mam_sandbox, mock_herdr, mock_agents, name): assert not any(r["name"] == name for r in _rows(mam_sandbox)) _set_mock(mock_herdr, strict_agent_names=True, agents={sanitize_herdr_agent_name(name): _agent(mam_sandbox, "w1:p72")}) out, server = _delegate(mam_sandbox, mock_herdr, name) assert _delegate_hint(out) == attach_command(server, name)