517 lines
17 KiB
Python
517 lines
17 KiB
Python
import os
|
|
import shutil
|
|
import json
|
|
import uuid
|
|
import sqlite3
|
|
import pytest
|
|
|
|
@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 = "/home/godopu16/PuKi/laa/canary_projects/multi-agent-mux/.agents/skills"
|
|
shutil.copytree(src_skills, tmp_path / "skills", ignore=shutil.ignore_patterns('multi-agent-mux-ui'))
|
|
shutil.copytree(src_skills, tmp_path / ".agents" / "skills", ignore=shutil.ignore_patterns('multi-agent-mux-ui'))
|
|
|
|
# 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))
|
|
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
|
|
|
|
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", "w")
|
|
fcntl.flock(lock_f, fcntl.LOCK_EX)
|
|
|
|
with open(state_file, 'r') as f:
|
|
state = json.load(f)
|
|
|
|
# Record the command call
|
|
state["calls"].append(sys.argv[1:])
|
|
|
|
def save_state():
|
|
with open(state_file, 'w') as f:
|
|
json.dump(state, f, indent=2)
|
|
|
|
# Save calls immediately so they persist even if we exit early or error out
|
|
save_state()
|
|
|
|
args = sys.argv[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":
|
|
res = {"workspaces": state.get("workspaces", [])}
|
|
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
|
|
else:
|
|
i += 1
|
|
workspaces = state.get("workspaces", [])
|
|
if not any(w["label"] == label for w in workspaces):
|
|
ws_id = f"w{len(workspaces) + 1}"
|
|
workspaces.append({"workspace_id": ws_id, "label": label, "cwd": cwd})
|
|
state["workspaces"] = workspaces
|
|
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 = ""
|
|
# 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:]
|
|
|
|
i = 0
|
|
while i < len(opts):
|
|
if opts[i] == "--workspace":
|
|
ws = opts[i+1]
|
|
i += 2
|
|
elif opts[i] == "--cwd":
|
|
cwd = opts[i+1]
|
|
i += 2
|
|
elif opts[i] == "--env":
|
|
env_val = opts[i+1]
|
|
if "=" in env_val:
|
|
k, v = env_val.split("=", 1)
|
|
os.environ[k] = v
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
|
|
# Determine the agent type (claude, agy, hermes, cline)
|
|
agent_type = "claude"
|
|
if agent_cmd:
|
|
if "claude" in agent_cmd[0]:
|
|
agent_type = "claude"
|
|
elif "agy" in agent_cmd[0]:
|
|
agent_type = "agy"
|
|
elif "hermes" in agent_cmd[0]:
|
|
agent_type = "hermes"
|
|
elif "cline" in agent_cmd[0]:
|
|
agent_type = "cline"
|
|
else:
|
|
# guess from name
|
|
if "claude" in name:
|
|
agent_type = "claude"
|
|
elif "agy" in name:
|
|
agent_type = "agy"
|
|
elif "hermes" in name:
|
|
agent_type = "hermes"
|
|
elif "cline" in name:
|
|
agent_type = "cline"
|
|
|
|
# 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")
|
|
|
|
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": "w1:p1",
|
|
"command": " ".join(agent_cmd),
|
|
"buffer": buffer_content
|
|
}
|
|
|
|
# Generate session/conversation UUID and files to simulate agent startup
|
|
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.abspath(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()
|
|
sys.exit(0)
|
|
elif cmd2 == "get":
|
|
if len(args) < 3:
|
|
sys.exit(1)
|
|
name = args[2]
|
|
agents = state.get("agents", {})
|
|
if name in agents:
|
|
agent_data = agents[name]
|
|
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}))
|
|
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", {})
|
|
if name in agents:
|
|
buffer_content = agents[name].get("buffer", "Ready")
|
|
print(buffer_content)
|
|
sys.exit(0)
|
|
else:
|
|
sys.stderr.write("Agent " + name + " not found\\\\n")
|
|
sys.exit(1)
|
|
elif cmd2 == "send":
|
|
if len(args) < 4:
|
|
sys.exit(1)
|
|
name = args[2]
|
|
text = args[3]
|
|
agents = state.get("agents", {})
|
|
if name in agents:
|
|
agents[name]["sent_text"] = agents[name].get("sent_text", "") + text
|
|
if text == "C-m":
|
|
agents[name]["buffer"] = agents[name].get("buffer", "") + "\\nesc to interrupt"
|
|
else:
|
|
agents[name]["buffer"] = agents[name].get("buffer", "") + "\\n" + text
|
|
if "/exit" in text or "exit" in text or "Exit" in text:
|
|
agents[name]["status"] = "stopped"
|
|
state["agents"] = agents
|
|
save_state()
|
|
sys.exit(0)
|
|
else:
|
|
sys.stderr.write("Agent " + name + " not found\\\\n")
|
|
sys.exit(1)
|
|
|
|
elif cmd1 == "session":
|
|
if len(args) < 3:
|
|
sys.exit(1)
|
|
cmd2 = args[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 == "pane":
|
|
if len(args) < 4:
|
|
sys.exit(1)
|
|
cmd2 = args[1]
|
|
if cmd2 == "send-keys":
|
|
name = args[2]
|
|
key = args[3]
|
|
agents = state.get("agents", {})
|
|
if name in agents:
|
|
agents[name]["sent_keys"] = agents[name].get("sent_keys", []) + [key]
|
|
if key in ("Enter", "C-m"):
|
|
agents[name]["buffer"] = agents[name].get("buffer", "") + "\\nesc to interrupt"
|
|
state["agents"] = agents
|
|
save_state()
|
|
sys.exit(0)
|
|
else:
|
|
sys.exit(1)
|
|
|
|
elif cmd1 == "ls":
|
|
if "-F" in args:
|
|
for name, data in state.get("agents", {}).items():
|
|
print(f"{name}|999999")
|
|
sys.exit(0)
|
|
with open("/tmp/debug_mock_herdr.log", "a") as f_debug:
|
|
f_debug.write(f"ARGS: {sys.argv[1:]} | AGENTS: {list(state.get('agents', {}).keys())} | PATH: {os.path.exists(state_file)}\\n")
|
|
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
|