multi-agent-mux/tests/conftest.py

import os
import sys
import shutil
import json
import uuid
import sqlite3
from pathlib import Path
import pytest

SKILLS_DIR = str(Path(__file__).resolve().parent.parent / ".agents" / "skills")
if SKILLS_DIR not in sys.path:
    sys.path.insert(0, SKILLS_DIR)

@pytest.fixture
def mam_sandbox(tmp_path, monkeypatch):
    """
    Manages a temporary sandboxed directory for .mam/ configuration files
    and copies the scripts from .agents/skills to prevent contaminating the real codebase.
    Also overrides environments (HOME, PATH, AGENT_SESSIONS_YAML, etc.) for isolation.
    """
    # 1. Copy the skill scripts to sandboxed tmp_path/skills and tmp_path/.agents/skills
    src_skills = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".agents", "skills"))
    shutil.copytree(src_skills, tmp_path / "skills")
    shutil.copytree(src_skills, tmp_path / ".agents" / "skills")
    
    # 2. Create sandboxed directory structure
    mam_dir = tmp_path / ".mam"
    mam_dir.mkdir(parents=True, exist_ok=True)
    
    yaml_path = mam_dir / "agent-sessions.yaml"
    # Create empty registry structure
    yaml_path.write_text("herdr_sessions: []\ndelegation_jobs: []\n")
    
    # 3. Setup mock home directory contents
    (tmp_path / ".claude" / "projects").mkdir(parents=True, exist_ok=True)
    (tmp_path / ".local" / "bin").mkdir(parents=True, exist_ok=True)
    
    # 4. Monkeypatch environment variables to point inside the sandbox
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("HOME_DIR", str(tmp_path))
    monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(tmp_path / ".claude" / "projects"))
    monkeypatch.setenv("LOCAL_BIN", str(tmp_path / ".local" / "bin"))
    monkeypatch.setenv("AGENT_SESSIONS_YAML", str(yaml_path))
    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
    monkeypatch.delenv("HERDR_SESSION_NAME", raising=False)
    monkeypatch.delenv("HERDR_SERVER_NAME", raising=False)
    monkeypatch.delenv("HERDR_WORKSPACE_ID", raising=False)
    import sys
    monkeypatch.setenv("AGENT_PYTHON_BIN", sys.executable)
    
    return tmp_path

@pytest.fixture
def mock_herdr(mam_sandbox, monkeypatch):
    """
    Generates a mock 'herdr' executable and prepends it to PATH.
    Allows tests to control and verify state via mock_herdr_state.json.
    Automatically generates mock session database files upon 'agent start'.
    """
    tmp_path = mam_sandbox
    state_file = tmp_path / "mock_herdr_state.json"
    
    # Initial mock state
    initial_state = {
        "workspaces": [
            {
                "workspace_id": "w1",
                "label": "default",
                "cwd": str(tmp_path)
            }
        ],
        "agents": {},
        "calls": []
    }
    state_file.write_text(json.dumps(initial_state, indent=2))
    
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir(parents=True, exist_ok=True)
    
    herdr_bin = bin_dir / "herdr"
    
    # Write Python implementation of mock herdr (using string replacement to avoid f-string escaping issues)
    # Double escape backslashes for newline (\\\\n) and write correctly
    code = """#!/usr/bin/env python3
import sys
import os
import json
import uuid
import sqlite3
import fcntl
import time

state_file = "STATE_FILE_PLACEHOLDER"

if not os.path.exists(state_file):
    sys.exit(0)

# Lock the state file exclusively to prevent concurrent race conditions
lock_f = open(state_file + ".lock", "a")
fcntl.flock(lock_f, fcntl.LOCK_EX)

state = {"workspaces": [], "agents": {}, "calls": []}
if os.path.exists(state_file):
    for _retry in range(50):
        try:
            with open(state_file, 'r') as f:
                content = f.read().strip()
                if content:
                    state = json.loads(content)
                    break
        except Exception:
            pass
        time.sleep(0.02)

# Record the command call
state.setdefault("calls", []).append(sys.argv[1:])
try:
    with open(state_file + ".trace", "a") as tf:
        tf.write(f"PID {os.getpid()} ARGS: {sys.argv[1:]}\\n")
        tf.write(f"   AGENTS: {list(state.get('agents', {}).keys())}\\n")
except Exception:
    pass

def save_state():
    # LOCK INVARIANT (W17):
    # lock_f (state_file + .lock) acquired at entry is deliberately NOT closed inside save_state().
    # This exclusive global process lock is the LOAD-BEARING guarantee of consistency for
    # overwriting disk_state["agents"] and disk_state["workspaces"].
    # If lock scope is ever narrowed or made fine-grained, state merging MUST switch to
    # explicit key-by-key dictionary diff merging first before removing the global lock.
    disk_state = {"workspaces": [], "agents": {}, "calls": [], "panes": []}
    if os.path.exists(state_file):
        for _retry in range(50):
            try:
                with open(state_file, 'r') as f:
                    content = f.read().strip()
                    if content:
                        disk_state = json.loads(content)
                        break
            except Exception:
                pass
            time.sleep(0.02)
    disk_state["agents"] = state.get("agents", {})
    state["agents"] = disk_state["agents"]
    
    if "workspaces" in state:
        disk_ws = disk_state.setdefault("workspaces", [])
        for w in state["workspaces"]:
            if not any(dw.get("workspace_id") == w.get("workspace_id") for dw in disk_ws):
                disk_ws.append(w)
    if "panes" in state:
        disk_panes = disk_state.setdefault("panes", [])
        for p in state["panes"]:
            matched = False
            for i, dp in enumerate(disk_panes):
                if dp.get("pane_id") == p.get("pane_id"):
                    disk_panes[i] = p
                    matched = True
                    break
            if not matched:
                disk_panes.append(p)
    disk_calls = disk_state.setdefault("calls", [])
    if sys.argv[1:] and (not disk_calls or disk_calls[-1] != sys.argv[1:]):
        disk_calls.append(sys.argv[1:])

    tmp_state = state_file + f".tmp.{os.getpid()}"
    with open(tmp_state, 'w') as f:
        json.dump(disk_state, f, indent=2)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp_state, state_file)

# Save calls immediately so they persist even if we exit early or error out
save_state()

def _sanitize_agent_name(name):
    if not name:
        return "agent"
    s = str(name).lower()
    import re, hashlib
    s = re.sub(r'[^a-z0-9_-]', '-', s)
    if not s or not s[0].isalpha():
        s = "x-" + s
    if len(s) > 32:
        h = hashlib.sha1(s.encode('utf-8')).hexdigest()[:8]
        s = f"{s[:23]}-{h}"
    return s

def _match_agent(k, target):
    if not k or not target:
        return False
    return k == target or _sanitize_agent_name(k) == target or _sanitize_agent_name(target) == k or _sanitize_agent_name(k) == _sanitize_agent_name(target)

args = sys.argv[1:]
while args and args[0] in ("-L", "--server", "-s", "--session"):
    if len(args) > 1:
        args = args[2:]
    else:
        args = args[1:]
if not args:
    sys.exit(0)

cmd1 = args[0]
if cmd1 == "workspace":
    if len(args) < 2:
        sys.exit(0)
    cmd2 = args[1]
    if cmd2 == "list":
        # W10: WorkspaceInfo has NO cwd key
        wss = []
        for w in state.get("workspaces", []):
            wss.append({"workspace_id": w["workspace_id"], "label": w.get("label", "default")})
        res = {"workspaces": wss}
        print(json.dumps({"result": res}))
        sys.exit(0)
    elif cmd2 == "create":
        label = "default"
        cwd = "."
        i = 2
        while i < len(args):
            if args[i] == "--label":
                label = args[i+1]
                i += 2
            elif args[i] == "--cwd":
                cwd = args[i+1]
                i += 2
            elif args[i] == "--env":
                env_val = args[i+1]
                if "=" in env_val:
                    k, v = env_val.split("=", 1)
                    os.environ[k] = v
                i += 2
            else:
                i += 1
        workspaces = state.get("workspaces", [])
        ws_id = f"w{len(workspaces) + 1}"
        workspaces.append({"workspace_id": ws_id, "label": label, "cwd": cwd})
        state["workspaces"] = workspaces
        
        # W10: workspace create returns full workspace_created object
        root_pane_id = f"{ws_id}:p1"
        panes = state.get("panes", [])
        panes.append({"pane_id": root_pane_id, "workspace_id": ws_id, "cwd": cwd, "tab_id": f"{ws_id}:t1"})
        state["panes"] = panes
        save_state()
        print(json.dumps({
            "result": {
                "type": "workspace_created",
                "workspace": {"workspace_id": ws_id, "label": label},
                "tab": {"tab_id": f"{ws_id}:t1"},
                "root_pane": {"pane_id": root_pane_id}
            }
        }))
        sys.exit(0)

elif cmd1 == "pane":
    if len(args) < 2:
        sys.exit(0)
    cmd2 = args[1]
    if cmd2 == "list":
        target_ws = ""
        i = 2
        while i < len(args):
            if args[i] == "--workspace":
                target_ws = args[i+1]
                i += 2
            else:
                i += 1
        panes_list = []
        seen = set()
        # Merge agent-derived panes with state["panes"] (dedupe by pane_id).
        # Label-only panes live in state["panes"] and must remain visible
        # even when other agents are registered.
        for name, data in state.get("agents", {}).items():
            ws_id = data.get("workspace_id", "w1")
            if target_ws and ws_id != target_ws:
                continue
            pid = data.get("pane_id", f"{ws_id}:p1")
            entry = {
                "pane_id": pid,
                "workspace_id": ws_id,
                "cwd": data.get("cwd", "."),
                "tab_id": data.get("tab_id", f"{ws_id}:t1"),
                "agent": data.get("agent", "claude"),
                "name": name,
            }
            if data.get("label"):
                entry["label"] = data["label"]
            panes_list.append(entry)
            seen.add(pid)
        for p in state.get("panes", []):
            if target_ws and p.get("workspace_id") != target_ws:
                continue
            pid = p.get("pane_id")
            if pid in seen:
                for existing in panes_list:
                    if existing.get("pane_id") == pid:
                        for k in ("label", "name"):
                            if p.get(k) and not existing.get(k):
                                existing[k] = p[k]
                        break
                continue
            panes_list.append(p)
            if pid:
                seen.add(pid)
        print(json.dumps({"result": {"panes": panes_list}}))
        sys.exit(0)
    elif cmd2 == "split":
        # W11: pane split handler with --env support
        i = 2
        while i < len(args):
            if args[i] == "--env":
                env_val = args[i+1]
                if "=" in env_val:
                    k, v = env_val.split("=", 1)
                    os.environ[k] = v
                i += 2
            else:
                i += 1
        print(json.dumps({"result": {"type": "pane_split", "pane": {"pane_id": "w1:p_split"}}}))
        sys.exit(0)
    elif cmd2 == "layout":
        # W11: pane layout handler returning area, panes, focused_pane_id
        focused = "w1:p1"
        panes_list = []
        agents = state.get("agents", {})
        count = max(len(agents), 1)
        w_per_pane = 184 // count
        idx = 0
        for name, data in agents.items():
            panes_list.append({
                "pane_id": data.get("pane_id", f"w1:p{idx+1}"),
                "rect": {"width": w_per_pane, "height": 78}
            })
            idx += 1
        if not panes_list:
            panes_list = [{"pane_id": "w1:p1", "rect": {"width": 184, "height": 78}}]
        print(json.dumps({
            "result": {
                "area": {"width": 184, "height": 78},
                "focused_pane_id": panes_list[0]["pane_id"],
                "panes": panes_list
            }
        }))
        sys.exit(0)
    elif cmd2 == "send-keys":
        if len(args) < 4:
            sys.exit(1)
        name = args[2]
        key = args[3]
        agents = state.get("agents", {})
        actual_name = name
        for a_name, data in agents.items():
            if data.get("pane_id") == name or a_name == name:
                actual_name = a_name
                break
        if actual_name in agents:
            agents[actual_name]["sent_keys"] = agents[actual_name].get("sent_keys", []) + [key]
            if key in ("Enter", "C-m"):
                agents[actual_name]["buffer"] = agents[actual_name].get("buffer", "") + "\\\\n\\\\nesc to interrupt"
            state["agents"] = agents
            save_state()
            sys.exit(0)
        for p in state.get("panes", []):
            if p.get("pane_id") == name:
                p["sent_keys"] = p.get("sent_keys", []) + [key]
                if key in ("Enter", "C-m"):
                    p["buffer"] = p.get("buffer", "") + "\\n\\nesc to interrupt"
                save_state()
                sys.exit(0)
        sys.exit(1)
    elif cmd2 == "send-text":
        if len(args) < 4:
            sys.exit(1)
        pane_id = args[2]
        text = args[3]
        found = False
        for a_name, data in state.get("agents", {}).items():
            if data.get("pane_id") == pane_id or a_name == pane_id:
                data["sent_text"] = data.get("sent_text", "") + text
                data["buffer"] = data.get("buffer", "") + "\\n" + text
                found = True
                break
        if not found:
            for p in state.get("panes", []):
                if p.get("pane_id") == pane_id:
                    p["sent_text"] = p.get("sent_text", "") + text
                    p["buffer"] = p.get("buffer", "") + "\\n" + text
                    found = True
                    break
        if found:
            save_state()
            sys.exit(0)
        sys.exit(1)
    elif cmd2 == "read":
        if len(args) < 3:
            sys.exit(1)
        pane_id = args[2]
        for a_name, data in state.get("agents", {}).items():
            if data.get("pane_id") == pane_id or a_name == pane_id:
                print(data.get("buffer", "Ready"))
                sys.exit(0)
        for p in state.get("panes", []):
            if p.get("pane_id") == pane_id:
                print(p.get("buffer", ""))
                sys.exit(0)
        sys.stderr.write("Pane " + pane_id + " not found\\n")
        sys.exit(1)
    elif cmd2 == "rename":
        if len(args) < 4:
            sys.exit(1)
        pane_id = args[2]
        label = args[3]
        panes = state.setdefault("panes", [])
        renamed = False
        for p in panes:
            if p.get("pane_id") == pane_id:
                p["label"] = label
                renamed = True
                break
        if not renamed:
            panes.append({"pane_id": pane_id, "label": label})
        for a_name, data in state.get("agents", {}).items():
            if data.get("pane_id") == pane_id:
                data["label"] = label
        save_state()
        sys.exit(0)
    elif cmd2 == "process-info":
        pane_id = ""
        if "--pane" in args:
            pane_id = args[args.index("--pane") + 1]
        pid = 9999
        for name, data in state.get("agents", {}).items():
            if data.get("pane_id") == pane_id or pane_id == "w1:p1":
                pid = data.get("pid", 9999)
                break
        res = {
            "process_info": {
                "foreground_processes": [{"pid": pid}],
                "shell_pid": pid
            }
        }
        print(json.dumps({"result": res}))
        sys.exit(0)
    elif cmd2 == "close":
        pane_id = args[2]
        agents = state.get("agents", {})
        to_delete = []
        for name, data in agents.items():
            if data.get("pane_id") == pane_id or pane_id == "w1:p1" or name == pane_id:
                to_delete.append(name)
        for name in to_delete:
            del agents[name]
        state["agents"] = agents
        save_state()
        sys.exit(0)

elif cmd1 == "agent":
    if len(args) < 2:
        sys.exit(0)
    cmd2 = args[1]
    if cmd2 == "list":
        agents_list = []
        for name, data in state.get("agents", {}).items():
            agents_list.append({
                "name": name,
                "agent": data.get("agent", "claude"),
                "agent_status": data.get("status", "running"),
                "cwd": data.get("cwd", ""),
                "pane_id": data.get("pane_id", "w1:p1"),
                "workspace_id": data.get("workspace_id", "w1")
            })
        print(json.dumps({"result": {"agents": agents_list}}))
        sys.exit(0)
    elif cmd2 == "start":
        if len(args) < 3:
            sys.exit(1)
        name = args[2]
        ws = ""
        cwd = ""
        pane = ""
        agent_type = ""
        # Find where -- is
        try:
            double_dash_idx = args.index("--")
            agent_cmd = args[double_dash_idx+1:]
            opts = args[3:double_dash_idx]
        except ValueError:
            agent_cmd = []
            opts = args[3:]
        
        # Whitelist herdr 0.8.0 allowed flags for agent start
        whitelist = {"--kind", "--pane", "--timeout"}
        i = 0
        unknown_flags = []
        while i < len(opts):
            opt = opts[i]
            if opt not in whitelist:
                unknown_flags.append(opt)
                i += 1
                continue
            if opt == "--kind":
                agent_type = opts[i+1]
                i += 2
            elif opt == "--pane":
                pane = opts[i+1]
                i += 2
            elif opt == "--timeout":
                i += 2
            else:
                i += 1

        if unknown_flags:
            sys.stderr.write("unknown option: " + str(unknown_flags[0]) + "\\n")
            sys.exit(1)

        if not agent_type or not pane:
            missing_arg = "--pane" if not pane else "--kind"
            sys.stderr.write("missing required " + missing_arg + "\\n")
            sys.exit(1)

        # Name validation: ^[a-z][a-z0-9_-]{0,31}$
        import re
        if not re.match(r'^[a-z][a-z0-9_-]{0,31}$', name):
            import json
            err_payload = {
                "id": "cli:agent:start",
                "error": {
                    "code": "invalid_agent_name",
                    "message": "agent name must start with a lowercase letter and contain only lowercase letters, digits, '-' or '_' (1-32 characters)"
                }
            }
            sys.stderr.write(json.dumps(err_payload) + "\\n")
            sys.exit(1)

        # TUI Welcome Tokens definition to prevent TUI readiness check timeout
        buffer_content = {
            "claude": "Anthropic Claude Ready",
            "agy": "Antigravity Ready",
            "hermes": "Hermes Ready",
            "cline": "Cline Chat Ready"
        }.get(agent_type, "Ready")

        # Look up cwd from target pane in state if not explicitly passed
        if not cwd and pane:
            for p in state.get("panes", []):
                if p.get("pane_id") == pane:
                    cwd = p.get("cwd", "")
                    ws = p.get("workspace_id", ws)
                    break
        if not ws and pane and ":" in pane:
            ws = pane.split(":")[0]

        agents = state.get("agents", {})
        agents[name] = {
            "agent": agent_type,
            "status": "running",
            "cwd": cwd or "TMP_PATH_PLACEHOLDER",
            "workspace_id": ws or "w1",
            "pid": 9999,
            "pane_id": pane or f"w1:p_{name}",
            "command": " ".join(agent_cmd),
            "buffer": buffer_content
        }
        
        # Generate session/conversation UUID and files to simulate agent startup
        session_uuid = ""
        cmd_str = " ".join(agent_cmd)
        if "--session-id" in cmd_str:
            parts = cmd_str.split()
            if "--session-id" in parts:
                _i = parts.index("--session-id")
                if _i + 1 < len(parts):
                    session_uuid = parts[_i + 1]
        if not session_uuid:
            session_uuid = str(uuid.uuid4())
        own_key_map = {
            "claude": "claude_session_id_own",
            "agy": "agy_conversation_id_own",
            "hermes": "hermes_conversation_id_own",
            "cline": "cline_conversation_id_own"
        }
        own_key = own_key_map.get(agent_type)
        if own_key:
            agents[name][own_key] = session_uuid

        # Find isolation directory (e.g. if HOME has agent_homes/<uuid>)
        home_dir = os.environ.get("HOME", "TMP_PATH_PLACEHOLDER")
        ws_abs = os.path.realpath(cwd or "TMP_PATH_PLACEHOLDER")
        ws_key = ws_abs.replace('/', '-').replace('_', '-')

        if agent_type == "claude":
            ccd = os.environ.get("CLAUDE_CONFIG_DIR")
            if ccd:
                cp_dir = os.path.join(ccd, "projects")
            else:
                cp_dir = os.environ.get("CLAUDE_PROJECT_DIR", f"{home_dir}/.claude/projects")
            proj_dir = os.path.join(cp_dir, ws_key)
            os.makedirs(proj_dir, exist_ok=True)
            jsonl_file = os.path.join(proj_dir, f"{session_uuid}.jsonl")
            with open(jsonl_file, 'w') as jf:
                jf.write(json.dumps({"sessionId": session_uuid}) + "\\n")
        elif agent_type == "agy":
            db_dir = os.path.join(home_dir, ".gemini", "antigravity-cli", "conversations")
            os.makedirs(db_dir, exist_ok=True)
            db_file = os.path.join(db_dir, f"{session_uuid}.db")
            with open(db_file, 'w') as df:
                df.write("")
            
            lc_dir = os.path.join(home_dir, ".gemini", "antigravity-cli", "cache")
            os.makedirs(lc_dir, exist_ok=True)
            lc_file = os.path.join(lc_dir, "last_conversations.json")
            lc_data = {}
            if os.path.exists(lc_file):
                try:
                    with open(lc_file, 'r') as lcf:
                        lc_data = json.load(lcf)
                except Exception:
                    pass
            lc_data[ws_abs] = session_uuid
            with open(lc_file, 'w') as lcf:
                json.dump(lc_data, lcf)
        elif agent_type == "hermes":
            hermes_dir = os.path.join(home_dir, ".hermes")
            os.makedirs(hermes_dir, exist_ok=True)
            hdb = os.path.join(hermes_dir, "state.db")
            conn = sqlite3.connect(hdb)
            conn.execute("CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, cwd TEXT, started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)")
            conn.execute("INSERT OR REPLACE INTO sessions (id, cwd) VALUES (?, ?)", (session_uuid, ws_abs))
            conn.commit()
            conn.close()
        elif agent_type == "cline":
            clin_base = os.path.join(home_dir, ".cline", "data", "sessions")
            if "agent_homes" in home_dir:
                clin_base = os.path.join(home_dir, "sessions")
            session_dir = os.path.join(clin_base, session_uuid)
            os.makedirs(session_dir, exist_ok=True)
            json_file = os.path.join(session_dir, f"{session_uuid}.json")
            with open(json_file, 'w') as jf:
                jf.write(json.dumps({"id": session_uuid}))

        state["agents"] = agents
        save_state()
        print(json.dumps({
            "result": {
                "type": "agent_started",
                "agent": {
                    "name": name,
                    "workspace_id": ws or "w1",
                    "pane_id": f"w1:p_{name}",
                    "cwd": cwd or "TMP_PATH_PLACEHOLDER"
                },
                "argv": agent_cmd
            }
        }))
        sys.exit(0)
    elif cmd2 == "get":
        if len(args) < 3:
            sys.exit(1)
        name = args[2]
        agents = state.get("agents", {})
        matched_k = None
        for k in agents:
            if _match_agent(k, name):
                matched_k = k
                break
        if matched_k:
            agent_data = agents[matched_k]
            pane_info = {
                "pid": agent_data.get("pid", 9999),
                "cwd": agent_data.get("cwd", ""),
                "command": agent_data.get("command", ""),
                "argv": agent_data.get("command", "")
            }
            print(json.dumps({
                "pane": pane_info,
                "result": {
                    "agent": {
                        "agent": agent_data.get("agent", "claude"),
                        "status": agent_data.get("status", "running"),
                        "cwd": agent_data.get("cwd", ""),
                        "pane_id": agent_data.get("pane_id", "w1:p1"),
                        "workspace_id": agent_data.get("workspace_id", "w1")
                    }
                }
            }))
            sys.exit(0)
        else:
            sys.stderr.write("Agent " + name + " not found\\n")
            sys.exit(1)
    elif cmd2 == "read":
        if len(args) < 3:
            sys.exit(1)
        name = args[2]
        agents = state.get("agents", {})
        matched_k = None
        for k in agents:
            if _match_agent(k, name):
                matched_k = k
                break
        if matched_k:
            buffer_content = agents[matched_k].get("buffer", "Ready")
            print(buffer_content)
            sys.exit(0)
        else:
            sys.stderr.write("Agent " + name + " not found\\n")
            sys.exit(1)
    elif cmd2 == "prompt":
        if len(args) < 4:
            sys.exit(1)
        name = args[2]
        text = args[3]
        agents = state.get("agents", {})
        matched_k = None
        for k in agents:
            if _match_agent(k, name):
                matched_k = k
                break
        if matched_k:
            agents[matched_k]["sent_text"] = agents[matched_k].get("sent_text", "") + text
            agents[matched_k]["buffer"] = agents[matched_k].get("buffer", "") + "\\n" + text + "\\n\\nesc to interrupt"
            state["agents"] = agents
            save_state()
            print(json.dumps({"id": "cli:agent:prompt", "result": {"type": "ok"}}))
            sys.exit(0)
        sys.stderr.write("Agent " + name + " not found\\n")
        sys.exit(1)
    else:
        # Real herdr has no `agent send`. Unknown subcommands must fail
        # (exit 1) rather than silently succeeding.
        sys.stderr.write("error: unrecognized subcommand '" + cmd2 + "'\\n")
        sys.exit(1)

elif cmd1 == "session":
    if len(args) < 2:
        sys.exit(0)
    cmd2 = args[1]
    if cmd2 == "list":
        session_names = ["custom_server", "default", "multi-agent-mux"]
        session_names.extend(list(state.get("agents", {}).keys()))
        if os.environ.get("MAM_SESS"):
            session_names.append(os.environ["MAM_SESS"])
        res = {"sessions": [{"name": s, "running": True} for s in set(session_names)]}
        print(json.dumps(res))
        sys.exit(0)
    if len(args) < 3:
        sys.exit(1)
    name = args[2]
    agents = state.get("agents", {})
    if cmd2 == "stop":
        if name in agents:
            agents[name]["status"] = "stopped"
            state["agents"] = agents
            save_state()
            sys.exit(0)
        else:
            sys.exit(1)
    elif cmd2 == "delete":
        if name in agents:
            del agents[name]
            state["agents"] = agents
            save_state()
            sys.exit(0)
        else:
            sys.exit(1)

elif cmd1 == "send-keys":
    target = ""
    key = ""
    i = 1
    while i < len(args):
        if args[i] in ("-t", "--target") and i + 1 < len(args):
            target = args[i+1]
            i += 2
        else:
            key = args[i]
            i += 1
    agents = state.get("agents", {})
    actual_name = target
    for a_name, data in agents.items():
        if data.get("pane_id") == target or a_name == target:
            actual_name = a_name
            break
    if actual_name in agents:
        agents[actual_name]["sent_keys"] = agents[actual_name].get("sent_keys", []) + [key]
        if key in ("Enter", "C-m"):
            agents[actual_name]["buffer"] = agents[actual_name].get("buffer", "") + "\\\\n\\\\nesc to interrupt"
        state["agents"] = agents
        save_state()
        sys.exit(0)
    else:
        sys.exit(0)

elif cmd1 == "capture-pane":
    target = ""
    i = 1
    while i < len(args):
        if args[i] in ("-t", "--target") and i + 1 < len(args):
            target = args[i+1]
            i += 2
        else:
            i += 1
    agents = state.get("agents", {})
    actual_name = target
    for a_name, data in agents.items():
        if data.get("pane_id") == target or a_name == target:
            actual_name = a_name
            break
    if actual_name in agents:
        print(agents[actual_name].get("buffer", "Ready"))
    else:
        print("Ready")
    sys.exit(0)

elif cmd1 == "list-panes":
    session_target = ""
    if "-t" in args:
        session_target = args[args.index("-t") + 1]
    agents = state.get("agents", {})
    matched_k = None
    for k in agents:
        if _match_agent(k, session_target):
            matched_k = k
            break
    if matched_k:
        data = agents[matched_k]
        pid = data.get("pid", 9999)
        cwd = data.get("cwd", "TMP_PATH_PLACEHOLDER")
        cmd = data.get("command", "claude")
        if "-F" in args:
            f_idx = args.index("-F")
            fmt = args[f_idx + 1] if f_idx + 1 < len(args) else ""
            if "pane_current_path" in fmt or "|" in fmt:
                print(f"{pid}|{cwd}|{cmd}")
            else:
                print(f"{pid}")
        else:
            print(f"{pid}|{cwd}|{cmd}")
        sys.exit(0)
    else:
        sys.exit(1)

elif cmd1 == "has-session":
    sess_target = ""
    if "-t" in args:
        sess_target = args[args.index("-t") + 1]
    elif len(args) > 1:
        sess_target = args[1]
    agents = state.get("agents", {})
    if any(_match_agent(k, sess_target) for k in agents):
        sys.exit(0)
    else:
        sys.exit(1)

elif cmd1 == "kill-session":
    sess_target = ""
    if "-t" in args:
        sess_target = args[args.index("-t") + 1]
    elif len(args) > 1:
        sess_target = args[1]
    agents = state.get("agents", {})
    to_del = [k for k in agents if _match_agent(k, sess_target)]
    for k in to_del:
        del agents[k]
    state["agents"] = agents
    save_state()
    sys.exit(0)

elif cmd1 == "ls":
    if "-F" in args:
        # Real POSIX seconds, not a sentinel: a below-floor value silently
        # disables the stale-transcript guard, so a mock that emits one lets
        # B-4-class regressions pass unnoticed. Tests may pin a specific
        # creation time per agent via the "created" key in the state file.
        now = int(time.time())
        for name, data in state.get("agents", {}).items():
            print(name + "|" + str(data.get("created") or now))
        sys.exit(0)
    agents_list = []
    for name, data in state.get("agents", {}).items():
        agents_list.append({
            "name": name,
            "agent": data.get("agent", "claude"),
            "agent_status": data.get("status", "running")
        })
    print(json.dumps({"result": {"agents": agents_list}}))
    sys.exit(0)

sys.exit(0)
""".replace("STATE_FILE_PLACEHOLDER", str(state_file)).replace("TMP_PATH_PLACEHOLDER", str(tmp_path))
    
    herdr_bin.write_text(code)
    herdr_bin.chmod(0o755)
    
    # Update PATH using monkeypatch.
    # Prepend bin_dir, and append standard Homebrew/system paths so lib.sh won't prepend them
    old_path = os.environ.get("PATH", "")
    extra_dirs = [
        "/home/linuxbrew/.linuxbrew/bin",
        "/home/linuxbrew/.linuxbrew/sbin",
        f"{tmp_path}/.local/bin",
        f"{tmp_path}/.npm-global/bin"
    ]
    new_path = old_path
    for d in extra_dirs:
        if d not in new_path:
            new_path = f"{new_path}:{d}"
            
    monkeypatch.setenv("PATH", f"{bin_dir}:{new_path}")
    
    return state_file

@pytest.fixture
def mock_agents(mam_sandbox, monkeypatch):
    """
    Generates mock agent binaries (claude, agy, hermes, cline)
    and places them in the sandboxed bin folder to satisfy preflight checks.
    """
    tmp_path = mam_sandbox
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir(parents=True, exist_ok=True)
    
    # 1. claude mock
    claude_bin = bin_dir / "claude"
    claude_bin.write_text("""#!/usr/bin/env python3
import sys
import json
args = sys.argv[1:]
if len(args) >= 2 and args[0] == "auth" and args[1] == "status":
    print(json.dumps({"loggedIn": True}))
    sys.exit(0)
sys.exit(0)
""")
    claude_bin.chmod(0o755)

    # 2. agy mock
    agy_bin = bin_dir / "agy"
    agy_bin.write_text("""#!/usr/bin/env python3
import sys
args = sys.argv[1:]
if len(args) >= 1 and args[0] == "models":
    print("gemini-1.5-pro\\ngemini-1.5-flash")
    sys.exit(0)
sys.exit(0)
""")
    agy_bin.chmod(0o755)

    # 3. hermes mock
    hermes_bin = bin_dir / "hermes"
    hermes_bin.write_text("""#!/usr/bin/env python3
import sys
args = sys.argv[1:]
if len(args) >= 1 and args[0] == "status":
    print("Hermes functional")
    sys.exit(0)
sys.exit(0)
""")
    hermes_bin.chmod(0o755)

    # 4. cline mock
    cline_bin = bin_dir / "cline"
    cline_bin.write_text("""#!/usr/bin/env python3
import sys
import json
args = sys.argv[1:]
if len(args) >= 1 and args[0] == "history":
    print(json.dumps([]))
    sys.exit(0)
sys.exit(0)
""")
    cline_bin.chmod(0o755)

    # 5. uuidgen mock to guarantee isolated creation UUIDs
    uuidgen_bin = bin_dir / "uuidgen"
    uuidgen_bin.write_text("""#!/usr/bin/env python3
import uuid
print(str(uuid.uuid4()))
""")
    uuidgen_bin.chmod(0o755)

    # Prepend bin directory to PATH
    old_path = os.environ.get("PATH", "")
    extra_dirs = [
        "/home/linuxbrew/.linuxbrew/bin",
        "/home/linuxbrew/.linuxbrew/sbin",
        f"{tmp_path}/.local/bin",
        f"{tmp_path}/.npm-global/bin"
    ]
    new_path = old_path
    for d in extra_dirs:
        if d not in new_path:
            new_path = f"{new_path}:{d}"
            
    monkeypatch.setenv("PATH", f"{bin_dir}:{new_path}")
    
    return bin_dir