1228 lines
49 KiB
Python
1228 lines
49 KiB
Python
import os
|
|
import re
|
|
import shutil
|
|
import json
|
|
import sqlite3
|
|
import subprocess
|
|
import pytest
|
|
import time
|
|
import sys
|
|
import shlex
|
|
from pathlib import Path
|
|
|
|
# Helper to run mutation on agent-sessions.yaml using atomic_dump_yaml in bash
|
|
def run_mutation(mam_sandbox, mutation_str, env=None):
|
|
lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh"
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
cmd_str = f"source {lib_path} && atomic_dump_yaml {yaml_path}"
|
|
run_env = dict(os.environ)
|
|
if env:
|
|
run_env.update(env)
|
|
res = subprocess.run(["bash", "-c", cmd_str], input=mutation_str, capture_output=True, text=True, env=run_env)
|
|
return res
|
|
|
|
def get_mqtt_common(mam_sandbox):
|
|
script_path = str(mam_sandbox / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts")
|
|
if script_path not in sys.path:
|
|
sys.path.insert(0, script_path)
|
|
import mqtt_common
|
|
return mqtt_common
|
|
|
|
|
|
# ==============================================================================
|
|
# FEATURE 1: Create Session (9 Test Cases)
|
|
# ==============================================================================
|
|
|
|
def test_comp_create_schema_validation(mam_sandbox):
|
|
"""Verify that atomic_dump_yaml validates the schema and rejects malformed formats."""
|
|
# Try setting herdr_sessions to a dictionary instead of list
|
|
mutation = "d['herdr_sessions'] = {}"
|
|
res = run_mutation(mam_sandbox, mutation)
|
|
assert "VALIDATE: herdr_sessions is not a list" in res.stderr
|
|
|
|
# Try setting status to an invalid state
|
|
mutation_invalid_status = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'session1',
|
|
'status': 'invalidstatus',
|
|
'pane': {}
|
|
}]
|
|
"""
|
|
res2 = run_mutation(mam_sandbox, mutation_invalid_status)
|
|
assert "VALIDATE:" in res2.stderr
|
|
|
|
def test_comp_create_role_immutability(mam_sandbox):
|
|
"""Verify that once a role is defined in the sessions, it cannot be mutated."""
|
|
# 1. Setup session with a role
|
|
mutation_setup = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'session-imm-role',
|
|
'status': 'running',
|
|
'role': 'Creator',
|
|
'pane': {}
|
|
}]
|
|
"""
|
|
res = run_mutation(mam_sandbox, mutation_setup)
|
|
assert res.returncode == 0
|
|
|
|
# 2. Try mutating the role
|
|
mutation_bad = """
|
|
for s in d.get('herdr_sessions', []):
|
|
if s.get('name') == 'session-imm-role':
|
|
s['role'] = 'Reviewer'
|
|
"""
|
|
res_bad = run_mutation(mam_sandbox, mutation_bad)
|
|
assert res_bad.returncode != 0
|
|
assert "role of session 'session-imm-role' cannot be modified" in res_bad.stderr
|
|
|
|
def test_comp_create_duplicate_running_ids(mam_sandbox):
|
|
"""Verify that duplicate running conversation IDs across sessions are rejected."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [
|
|
{
|
|
'name': 'sess1',
|
|
'status': 'running',
|
|
'claude_session_id_own': 'uuid-1234',
|
|
'pane': {}
|
|
},
|
|
{
|
|
'name': 'sess2',
|
|
'status': 'running',
|
|
'claude_session_id_own': 'uuid-1234',
|
|
'pane': {}
|
|
}
|
|
]
|
|
"""
|
|
res = run_mutation(mam_sandbox, mutation)
|
|
assert res.returncode != 0
|
|
assert "Duplicate running conversation ID" in res.stderr
|
|
|
|
def test_comp_create_sqlite_tables_created(mam_sandbox, mock_herdr, mock_agents):
|
|
"""Verify that tables exist and contain records after a full create_session.sh run."""
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
|
|
cmd = [
|
|
"bash", str(script_path),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator"
|
|
]
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
assert db_path.exists()
|
|
|
|
# Connect directly to the SQLite DB
|
|
conn = sqlite3.connect(str(db_path))
|
|
cursor = conn.cursor()
|
|
|
|
# Check tables
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
|
|
tables = [row[0] for row in cursor.fetchall()]
|
|
assert "state" in tables
|
|
assert "sessions" in tables
|
|
|
|
# Check entries in sessions table
|
|
cursor.execute("SELECT name, status, pane_cwd FROM sessions;")
|
|
rows = cursor.fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0][0].endswith("-creator-claude")
|
|
assert rows[0][1] == "running"
|
|
|
|
conn.close()
|
|
|
|
|
|
def test_comp_create_usage_matches_parser(mam_sandbox, mock_herdr, mock_agents):
|
|
"""Verify that create_session.sh usage documents --herdr-session and parser accepts it."""
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
res = subprocess.run(["bash", str(script), "--help"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
assert "--herdr-session" in res.stdout
|
|
assert "--herdr-server" in res.stdout
|
|
for agent in ("claude", "agy", "hermes", "cline"):
|
|
assert agent in res.stdout
|
|
|
|
# Test parser acceptance of valid flags vs unknown arg rejection
|
|
r1 = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator",
|
|
"--herdr-session", "test-sess",
|
|
"--dry-run"
|
|
], capture_output=True, text=True)
|
|
assert r1.returncode == 0
|
|
assert "unknown arg" not in r1.stderr
|
|
|
|
r2 = subprocess.run([
|
|
"bash", str(script),
|
|
"--invalid-flag-xyz"
|
|
], capture_output=True, text=True)
|
|
assert r2.returncode == 2
|
|
assert "unknown arg" in r2.stderr
|
|
|
|
|
|
def test_comp_create_herdr_session_cli_parsing_dry_run(mam_sandbox, mock_herdr, mock_agents):
|
|
"""Verify that --herdr-session and --herdr-server are parsed cleanly in --dry-run mode."""
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
res1 = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator",
|
|
"--herdr-session", "test-isolated-sess",
|
|
"--dry-run"
|
|
], capture_output=True, text=True)
|
|
assert res1.returncode == 0
|
|
assert "[dry-run] would spawn:" in res1.stdout
|
|
assert "herdr_session=test-isolated-sess" in res1.stdout
|
|
|
|
res2 = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator",
|
|
"--herdr-server", "test-isolated-srv",
|
|
"--dry-run"
|
|
], capture_output=True, text=True)
|
|
assert res2.returncode == 0
|
|
assert "[dry-run] would spawn:" in res2.stdout
|
|
assert "herdr_session=test-isolated-srv" in res2.stdout
|
|
|
|
|
|
def test_comp_create_herdr_session_default_preserved(mam_sandbox, mock_herdr, mock_agents):
|
|
"""Verify --herdr-session default is honored and not overwritten by workspace slug."""
|
|
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", "custom-proj-default-claude",
|
|
"--herdr-session", "default"
|
|
], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s["name"] == "custom-proj-default-claude"
|
|
assert s["herdr_session"] == "default"
|
|
assert s["herdr_server"] == "default"
|
|
assert "HERDR_SESSION_NAME=default" in s["start_command"]
|
|
assert "HERDR_SESSION_NAME=default" in s["attach_command"]
|
|
assert "HERDR_SESSION_NAME=default" in s["kill_command"]
|
|
|
|
|
|
def test_comp_create_herdr_session_yaml_propagation(mam_sandbox, mock_herdr, mock_agents):
|
|
"""Verify HERDR_SESSION_NAME propagation into start_command / herdr_session YAML field."""
|
|
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", "custom-proj-creator-claude",
|
|
"--herdr-session", "isolated-suite-01"
|
|
], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s["name"] == "custom-proj-creator-claude"
|
|
assert s["herdr_session"] == "isolated-suite-01"
|
|
assert s["herdr_server"] == "isolated-suite-01"
|
|
assert "HERDR_SESSION_NAME=isolated-suite-01" in s["start_command"]
|
|
assert "HERDR_SESSION_NAME=isolated-suite-01" in s["attach_command"]
|
|
assert "HERDR_SESSION_NAME=isolated-suite-01" in s["kill_command"]
|
|
|
|
|
|
def test_comp_create_herdr_workspace_parsing_and_env_fallback(mam_sandbox, mock_herdr, mock_agents):
|
|
"""T4: Verify --herdr-workspace CLI flag, HERDR_WORKSPACE env fallback, and default bare slug."""
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
|
|
# Flag passed
|
|
res1 = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator",
|
|
"--herdr-workspace", "my-explicit-label",
|
|
"--dry-run"
|
|
], capture_output=True, text=True)
|
|
assert res1.returncode == 0
|
|
assert "herdr_workspace=my-explicit-label" in res1.stdout
|
|
|
|
# Env set, flag omitted -> env wins (C-3)
|
|
res2 = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator",
|
|
"--dry-run"
|
|
], capture_output=True, text=True, env={**os.environ, "HERDR_WORKSPACE": "from-env-label"})
|
|
assert res2.returncode == 0
|
|
assert "herdr_workspace=from-env-label" in res2.stdout
|
|
|
|
# Both flag and env -> flag wins
|
|
res3 = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator",
|
|
"--herdr-workspace", "my-explicit-label",
|
|
"--dry-run"
|
|
], capture_output=True, text=True, env={**os.environ, "HERDR_WORKSPACE": "from-env-label"})
|
|
assert res3.returncode == 0
|
|
assert "herdr_workspace=my-explicit-label" in res3.stdout
|
|
|
|
# Neither -> default bare slug (D3), distinct from herdr_session
|
|
run_env = dict(os.environ)
|
|
run_env.pop("HERDR_WORKSPACE", None)
|
|
run_env.pop("HERDR_SESSION_NAME", None)
|
|
res4 = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--role", "Creator",
|
|
"--dry-run"
|
|
], capture_output=True, text=True, env=run_env)
|
|
assert res4.returncode == 0
|
|
parent = os.path.basename(os.path.dirname(str(mam_sandbox))).lower().replace('_', '-')
|
|
work = os.path.basename(str(mam_sandbox)).lower().replace('_', '-')
|
|
bare = f"{parent}-{work}".replace('_', '-')
|
|
import re
|
|
bare = re.sub(r'[^a-zA-Z0-9-]', '', bare).lstrip('-')
|
|
assert f"herdr_workspace={bare}" in res4.stdout
|
|
assert f"herdr_session=mam-{bare}" in res4.stdout
|
|
|
|
|
|
def test_comp_create_herdr_workspace_yaml_propagation(mam_sandbox, mock_herdr, mock_agents):
|
|
"""T5: Verify herdr_workspace distinct persistence in YAML and no leakage into commands."""
|
|
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", "custom-ws-creator-claude",
|
|
"--herdr-session", "isolated-sock-01",
|
|
"--herdr-workspace", "distinct-ws-label"
|
|
], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s["name"] == "custom-ws-creator-claude"
|
|
assert s["herdr_session"] == "isolated-sock-01"
|
|
assert s["herdr_server"] == "isolated-sock-01"
|
|
assert s["herdr_workspace"] == "distinct-ws-label"
|
|
assert "distinct-ws-label" not in s["start_command"]
|
|
assert "distinct-ws-label" not in s["attach_command"]
|
|
assert "distinct-ws-label" not in s["kill_command"]
|
|
|
|
|
|
def test_create_does_not_inherit_a_stale_workspace_label(mam_sandbox, mock_herdr, mock_agents):
|
|
"""T9 / D5: Recreating over a terminated row derives label afresh from --workspace."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'reuse-creator-claude',
|
|
'status': 'terminated',
|
|
'herdr_session': 'old-sock',
|
|
'herdr_server': 'old-sock',
|
|
'herdr_workspace': 'old-stale-label',
|
|
'pane': {'cwd': '/old/place', 'cmd': 'claude'}
|
|
}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
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", "reuse-creator-claude"
|
|
], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s["status"] == "running"
|
|
assert s["herdr_workspace"] != "old-stale-label"
|
|
|
|
|
|
# ==============================================================================
|
|
# FEATURE 2: Resume Session (8 Test Cases)
|
|
# ==============================================================================
|
|
|
|
def test_comp_resume_config_restore(mam_sandbox):
|
|
"""Verify that isolation details are preserved and can be read from the DB registry."""
|
|
# Write a test session with isolation details
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-session-iso',
|
|
'status': 'stopped',
|
|
'pane': {'cwd': '/tmp'},
|
|
'isolation': {
|
|
'uuid': 'iso-uuid-789',
|
|
'root': '/tmp/iso_root',
|
|
'lever': 'claude_config_dir',
|
|
'seeded': []
|
|
}
|
|
}]
|
|
"""
|
|
res = run_mutation(mam_sandbox, mutation)
|
|
assert res.returncode == 0
|
|
|
|
# Read the configuration via resume_session.sh isolation resolution check
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh"
|
|
# We run in dry-run/mock mode or just execute python command checking block
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
|
|
# Call the python parsing block from resume_session.sh directly
|
|
python_block = """
|
|
import os, json, sqlite3
|
|
name = "test-session-iso"
|
|
db_path = "DB_PATH_PLACEHOLDER"
|
|
conn = sqlite3.connect(db_path)
|
|
row = conn.execute('SELECT data FROM sessions WHERE name=?', (name,)).fetchone()
|
|
if row:
|
|
import json
|
|
s = json.loads(row[0])
|
|
print(json.dumps(s.get('isolation') or {}))
|
|
""".replace("DB_PATH_PLACEHOLDER", str(db_path))
|
|
|
|
res_py = subprocess.run(["python3", "-c", python_block], capture_output=True, text=True)
|
|
assert res_py.returncode == 0
|
|
data = json.loads(res_py.stdout)
|
|
assert data["uuid"] == "iso-uuid-789"
|
|
assert data["root"] == "/tmp/iso_root"
|
|
|
|
def test_comp_resume_metadata_read_integrity(mam_sandbox):
|
|
"""Verify metadata read integrity from the SQLite DB by changing state outside YAML."""
|
|
# Write initial YAML state
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'sess-integrity',
|
|
'status': 'stopped',
|
|
'pane': {'cwd': '/tmp'}
|
|
}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
# Update SQLite directly to terminated
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.execute("UPDATE sessions SET status='terminated' WHERE name='sess-integrity'")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# Read via atomic_dump_yaml check (which pulls from DB sessions table)
|
|
# If the DB reading has integrity, d will contain sess-integrity as status 'terminated'
|
|
mutation_check = """
|
|
for s in d.get('herdr_sessions', []):
|
|
if s.get('name') == 'sess-integrity':
|
|
print(f"STATUS={s['status']}")
|
|
"""
|
|
res = run_mutation(mam_sandbox, mutation_check)
|
|
assert "STATUS=terminated" in res.stdout
|
|
|
|
def test_comp_resume_update_yaml(mam_sandbox):
|
|
"""Verify that update_yaml_resumed.sh cleans up stop fields and marks status running."""
|
|
# Seed a stopped session with stop metadata
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-resumed-session-creator-claude',
|
|
'status': 'stopped',
|
|
'stopped_at': '2026-07-19T00:00:00Z',
|
|
'stop_reason': 'manual_stop',
|
|
'pane': {'cwd': '/tmp'}
|
|
}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "update_yaml_resumed.sh"
|
|
res = subprocess.run(["bash", str(script_path), "--session", "test-resumed-session-creator-claude", "--uuid", "new-uuid-999"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
|
|
# Verify stopped fields are popped
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
row = conn.execute("SELECT data FROM sessions WHERE name='test-resumed-session-creator-claude'").fetchone()
|
|
s = json.loads(row[0])
|
|
assert s["status"] == "running"
|
|
assert "stopped_at" not in s
|
|
assert "stop_reason" not in s
|
|
assert s["claude_session_id_own"] == "new-uuid-999"
|
|
conn.close()
|
|
|
|
def test_comp_resume_find_workspace_uuid_tier1_own_id(mam_sandbox):
|
|
"""Verify find_workspace_uuid resolves the per-row own ID properly."""
|
|
# Write a running session with a claude_session_id_own
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'ws-own-creator-claude',
|
|
'status': 'running',
|
|
'claude_session_id_own': 'own-uuid-111',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh"
|
|
# Target session mode scopes resolution to the row
|
|
cmd_str = f"source {lib_path} && find_workspace_uuid {mam_sandbox} claude ws-own-creator-claude"
|
|
# Since we are mock-checking, own_exists function mock will query disk format.
|
|
# Claude disk format requires a project file: projects/<key>/<uuid>.jsonl
|
|
# Let's create it.
|
|
key = str(mam_sandbox).replace('/', '-').replace('_', '-')
|
|
proj_dir = mam_sandbox / ".claude" / "projects" / key
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
(proj_dir / "own-uuid-111.jsonl").write_text('{"sessionId": "own-uuid-111"}')
|
|
|
|
res = subprocess.run(["bash", "-c", cmd_str], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
# Should resolve to own-uuid-111 because it exists in projects folder
|
|
assert res.stdout.strip() == "own-uuid-111"
|
|
|
|
def test_comp_resume_find_workspace_uuid_tier2_disk_scan_claude(mam_sandbox):
|
|
"""Verify find_workspace_uuid scans workspace on disk if own ID isn't set in the row."""
|
|
# Seed session without own ID
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'ws-scan-creator-claude',
|
|
'status': 'stopped',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
# Create JSONL file
|
|
key = str(mam_sandbox).replace('/', '-').replace('_', '-')
|
|
proj_dir = mam_sandbox / ".claude" / "projects" / key
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
(proj_dir / "scanned-uuid.jsonl").write_text('{"sessionId": "scanned-uuid"}')
|
|
|
|
lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh"
|
|
cmd_str = f"source {lib_path} && find_workspace_uuid {mam_sandbox} claude"
|
|
res = subprocess.run(["bash", "-c", cmd_str], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
assert res.stdout.strip() == "scanned-uuid"
|
|
|
|
|
|
def test_comp_resume_herdr_session_propagation(mam_sandbox, mock_herdr, mock_agents):
|
|
"""Verify that resume_session.sh with --herdr-session updates existing row's herdr_session."""
|
|
conv_id = "11111111-2222-3333-4444-555555555555"
|
|
key = str(mam_sandbox).replace('/', '-').replace('_', '-')
|
|
proj_dir = mam_sandbox / ".claude" / "projects" / key
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
(proj_dir / f"{conv_id}.jsonl").write_text(f'{{"sessionId": "{conv_id}"}}')
|
|
|
|
# Seed a stopped session with OLD herdr_session
|
|
mutation = f"""
|
|
d['herdr_sessions'] = [{{
|
|
'name': 'test-proj-creator-claude',
|
|
'status': 'stopped',
|
|
'herdr_session': 'OLD-HERDR-SESSION',
|
|
'herdr_server': 'OLD-HERDR-SESSION',
|
|
'claude_session_id_own': '{conv_id}',
|
|
'pane': {{'cwd': '{str(mam_sandbox)}', 'cmd': 'claude'}}
|
|
}}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh"
|
|
res = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--session", "test-proj-creator-claude",
|
|
"--herdr-session", "NEW-HERDR-SESSION"
|
|
], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s["status"] == "running"
|
|
assert s["herdr_session"] == "NEW-HERDR-SESSION"
|
|
assert s["herdr_server"] == "NEW-HERDR-SESSION"
|
|
assert "HERDR_SESSION_NAME=NEW-HERDR-SESSION" in s["attach_command"]
|
|
assert "HERDR_SESSION_NAME=NEW-HERDR-SESSION" in s["kill_command"]
|
|
|
|
|
|
def test_comp_resume_herdr_workspace_propagation(mam_sandbox, mock_herdr, mock_agents):
|
|
"""T6: Verify resume_session.sh with --herdr-workspace updates herdr_workspace while preserving herdr_session."""
|
|
conv_id = "22222222-3333-4444-5555-666666666666"
|
|
key = str(mam_sandbox).replace('/', '-').replace('_', '-')
|
|
proj_dir = mam_sandbox / ".claude" / "projects" / key
|
|
proj_dir.mkdir(parents=True, exist_ok=True)
|
|
(proj_dir / f"{conv_id}.jsonl").write_text(f'{{"sessionId": "{conv_id}"}}')
|
|
|
|
mutation = f"""
|
|
d['herdr_sessions'] = [{{
|
|
'name': 'test-proj-ws-creator-claude',
|
|
'status': 'stopped',
|
|
'herdr_session': 'PRESERVED-SESSION',
|
|
'herdr_server': 'PRESERVED-SESSION',
|
|
'herdr_workspace': 'OLD-WS-LABEL',
|
|
'claude_session_id_own': '{conv_id}',
|
|
'pane': {{'cwd': '{str(mam_sandbox)}', 'cmd': 'claude'}}
|
|
}}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh"
|
|
res = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--session", "test-proj-ws-creator-claude",
|
|
"--herdr-workspace", "NEW-WS-LABEL"
|
|
], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s["status"] == "running"
|
|
assert s["herdr_workspace"] == "NEW-WS-LABEL"
|
|
assert s["herdr_session"] == "PRESERVED-SESSION"
|
|
|
|
|
|
def test_comp_resume_herdr_workspace_new_row_branch(mam_sandbox, mock_herdr, mock_agents):
|
|
"""T7: Verify update_yaml_resumed.sh creates a new row with herdr_workspace when target is None."""
|
|
run_mutation(mam_sandbox, "d['herdr_sessions'] = []")
|
|
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "update_yaml_resumed.sh"
|
|
res = subprocess.run([
|
|
"bash", str(script),
|
|
"--workspace", str(mam_sandbox),
|
|
"--agent", "claude",
|
|
"--session", "brand-new-resumed-session",
|
|
"--uuid", "33333333-4444-5555-6666-777777777777",
|
|
"--herdr-session", "explicit-sock",
|
|
"--herdr-workspace", "explicit-ws"
|
|
], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s["name"] == "brand-new-resumed-session"
|
|
assert s["herdr_session"] == "explicit-sock"
|
|
assert s["herdr_server"] == "explicit-sock"
|
|
assert s["herdr_workspace"] == "explicit-ws"
|
|
|
|
|
|
# ==============================================================================
|
|
# FEATURE 3: Stop Session (7 Test Cases)
|
|
# ==============================================================================
|
|
|
|
def test_comp_stop_sqlite_state_update(mam_sandbox):
|
|
"""Verify that stop_session.sh updates state in SQLite to stopped."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-stop-sqlite-creator-claude',
|
|
'status': 'running',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
res = subprocess.run(["bash", str(script_path), "--session", "test-stop-sqlite-creator-claude"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
|
|
# Query database directly
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
row = conn.execute("SELECT status, data FROM sessions WHERE name='test-stop-sqlite-creator-claude'").fetchone()
|
|
assert row[0] == "stopped"
|
|
s = json.loads(row[1])
|
|
assert s["status"] == "stopped"
|
|
assert "stopped_at" in s
|
|
assert s["stop_reason"] == "manual_stop"
|
|
conn.close()
|
|
|
|
def test_comp_stop_purge_record_removal(mam_sandbox):
|
|
"""Verify that purge completely removes the session record from DB."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-purge-record-creator-claude',
|
|
'status': 'running',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
res = subprocess.run(["bash", str(script_path), "--session", "test-purge-record-creator-claude", "--purge-conversation", "--yes"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
|
|
# Check SQLite
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
row = conn.execute("SELECT COUNT(*) FROM sessions WHERE name='test-purge-record-creator-claude'").fetchone()
|
|
assert row[0] == 0
|
|
conn.close()
|
|
|
|
def test_comp_stop_lock_release(mam_sandbox):
|
|
"""Verify SQLite lock is released after running atomic_dump_yaml."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-lock-release',
|
|
'status': 'stopped',
|
|
'pane': {'cwd': '/tmp'}
|
|
}]
|
|
"""
|
|
res = run_mutation(mam_sandbox, mutation)
|
|
assert res.returncode == 0
|
|
|
|
# Since res has returned, the lock should be free.
|
|
# We test this by immediately acquiring a direct SQLite write connection
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path), timeout=0.1)
|
|
try:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
conn.execute("UPDATE sessions SET status='archived' WHERE name='test-lock-release'")
|
|
conn.commit()
|
|
except sqlite3.OperationalError as e:
|
|
pytest.fail(f"Could not acquire SQLite lock: {e}")
|
|
finally:
|
|
conn.close()
|
|
|
|
def test_comp_stop_graceful_kill_chain(mam_sandbox, mock_herdr, mock_agents):
|
|
"""Verify that the graceful stop command fallback chain runs when herdr has-session is alive."""
|
|
# Write a running session in YAML
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-graceful-chain-creator-claude',
|
|
'status': 'running',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER', 'pid': 12345}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
# Pre-populate herdr mock with the running session so it passes has-session check
|
|
with open(mock_herdr, 'r') as f:
|
|
state = json.load(f)
|
|
state["agents"]["test-graceful-chain-creator-claude"] = {
|
|
"status": "running",
|
|
"agent": "claude",
|
|
"cwd": str(mam_sandbox),
|
|
"pid": 12345,
|
|
"pane_id": "w1:p1",
|
|
"command": "claude",
|
|
"buffer": "Anthropic Claude Ready"
|
|
}
|
|
with open(mock_herdr, 'w') as f:
|
|
json.dump(state, f, indent=2)
|
|
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
# Run stop_session.sh. This should call send-keys first
|
|
res = subprocess.run(["bash", str(script_path), "--session", "test-graceful-chain-creator-claude"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
|
|
with open(mock_herdr, 'r') as f:
|
|
state = json.load(f)
|
|
|
|
# Verify mock herdr calls recorded the keys "/exit" sent
|
|
calls = state.get("calls", [])
|
|
|
|
# Should see send-keys or agent prompt call
|
|
assert any(("send" in call or "prompt" in call) and "/exit" in call for call in calls)
|
|
|
|
|
|
def test_comp_stop_agent_fallback_reads_pane_cmd(mam_sandbox):
|
|
"""B-21: --agent 생략 시 세션명에 에이전트 접미사가 없어도 레지스트리 행의
|
|
pane.cmd 로 해석된다 (라이브 `agy-creator-01` 형태)."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'agy-creator-01',
|
|
'status': 'running',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER', 'cmd': 'agy'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
res = subprocess.run(["bash", str(script), "--session", "agy-creator-01"],
|
|
capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
assert re.search(r"^\s*agent:\s+agy\s*$", res.stdout, re.M), res.stdout
|
|
|
|
|
|
def test_comp_stop_agent_fallback_prefers_explicit_agent_field(mam_sandbox):
|
|
"""우선순위 계약: 명시 `agent` 필드가 세션명 접미사와 pane.cmd 를 모두 이긴다."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'x-creator-claude',
|
|
'status': 'running',
|
|
'agent': 'hermes',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER', 'cmd': 'claude'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
res = subprocess.run(["bash", str(script), "--session", "x-creator-claude"],
|
|
capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
assert re.search(r"^\s*agent:\s+hermes\s*$", res.stdout, re.M), res.stdout
|
|
|
|
|
|
# 코드 펜스 안의 stop_session.sh 호출을 '명령 단위'로 잘라낸다.
|
|
# - 펜스 스코프: 산문 속 `stop_session.sh` 언급을 명령으로 오인하지 않는다
|
|
# (Pitfalls / When-NOT-to-use 절은 성격상 스크립트를 산문으로 언급한다).
|
|
# - 명령 단위: 한 펜스에 여러 호출이 들어 있어도 각각을 따로 검증한다
|
|
# (블록 단위로 보면 그중 하나만 --agent 를 가져도 통과해 버린다).
|
|
_FENCE_RE = re.compile(r"```(?:bash|sh)\n(.*?)```", re.S)
|
|
_STOP_CALL_RE = re.compile(r"(?:bash\s+)?\S*stop_session\.sh[^\n\\]*(?:\\\n[^\n\\]*)*")
|
|
|
|
|
|
def test_comp_docs_stop_examples_pass_agent():
|
|
"""B-21 문서 계약: 문서의 모든 stop_session.sh 예제는 --agent 를 넘긴다.
|
|
문서 변경은 뮤테이션 감도가 없으므로 이 가드가 표준의 유일한 집행 장치다."""
|
|
repo = Path(__file__).resolve().parent.parent
|
|
expected = { # 문서별 최소 예제 수 — 예제를 지워 가드를 무력화하는 것을 막는다
|
|
repo / ".agents/skills/multi-agent-mux-stop/SKILL.md": 3,
|
|
repo / "deploy/INSTALL.md": 2,
|
|
}
|
|
for doc, floor in expected.items():
|
|
seen = 0
|
|
for block in _FENCE_RE.findall(doc.read_text()):
|
|
for m in _STOP_CALL_RE.finditer(block):
|
|
snippet = m.group(0)
|
|
seen += 1
|
|
assert "--agent" in snippet, \
|
|
f"{doc.name}: stop_session.sh example without --agent:\n{snippet}"
|
|
assert seen >= floor, f"{doc.name}: expected >= {floor} examples, saw {seen}"
|
|
|
|
|
|
# ==============================================================================
|
|
# FEATURE 4: Status Query (5 Test Cases)
|
|
# ==============================================================================
|
|
|
|
def test_comp_status_read_lock(mam_sandbox):
|
|
"""Verify status.sh accesses SQLite database with a read lock, not writing."""
|
|
# Seed DB first so it exists
|
|
run_mutation(mam_sandbox, "d['herdr_sessions'] = []")
|
|
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"
|
|
|
|
# Run status.sh in a background sub-process
|
|
res = subprocess.run(["bash", str(script_path), "--json"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
|
|
# Verify no writing occurred by comparing DB file modified time
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
t1 = os.path.getmtime(db_path)
|
|
|
|
# Run status.sh again
|
|
subprocess.run(["bash", str(script_path), "--json"], capture_output=True, text=True)
|
|
t2 = os.path.getmtime(db_path)
|
|
|
|
assert t1 == t2
|
|
|
|
def test_comp_status_drift_class_parsing(mam_sandbox):
|
|
"""Verify status.sh correctly parses drifts returned by reconcile.sh."""
|
|
# Write reconcile mock logic that outputs a drift JSON
|
|
reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
|
|
drift_output = {
|
|
"timestamp": "2026-07-19T00:00:00Z",
|
|
"yaml_path": str(mam_sandbox / ".mam" / "agent-sessions.yaml"),
|
|
"herdr_sessions_alive": ["test-session|default"],
|
|
"herdr_confirmed": True,
|
|
"drifts": [{"class": "A", "name": "test-session", "msg": "drift detected"}],
|
|
"actions": []
|
|
}
|
|
|
|
# Overwrite reconcile.sh to print this JSON directly
|
|
reconcile_script.write_text(f"""#!/usr/bin/env bash
|
|
echo '{json.dumps(drift_output)}'
|
|
""")
|
|
|
|
# Seed YAML
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-session',
|
|
'status': 'running',
|
|
'pane': {'cwd': '/tmp'}
|
|
}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
status_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"
|
|
res = subprocess.run(["bash", str(status_script), "--json"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
data = json.loads(res.stdout)
|
|
assert data["sessions_detail"][0]["drift_classes"] == ["A"]
|
|
|
|
def test_comp_status_job_candidates_parsing(mam_sandbox):
|
|
"""Verify status.sh reads job statuses from audit logs or registry JSONs."""
|
|
# Write YAML session
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-job-session-creator-claude',
|
|
'status': 'running',
|
|
'delegate_job_id': 'job12345',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
# Create the status.json in both candidate folders to cover path resolutions
|
|
job_logs_dir1 = mam_sandbox / ".mam" / "delegate_job_logs" / "job12345"
|
|
job_logs_dir1.mkdir(parents=True, exist_ok=True)
|
|
(job_logs_dir1 / "status.json").write_text('{"status": "completed"}')
|
|
|
|
job_logs_dir2 = mam_sandbox.parent / ".mam" / "delegate_job_logs" / "job12345"
|
|
job_logs_dir2.mkdir(parents=True, exist_ok=True)
|
|
(job_logs_dir2 / "status.json").write_text('{"status": "completed"}')
|
|
|
|
status_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"
|
|
res = subprocess.run(["bash", str(status_script), "--json"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
data = json.loads(res.stdout)
|
|
assert data["sessions_detail"][0]["job_status"] == "completed"
|
|
|
|
def test_comp_status_no_side_effects(mam_sandbox):
|
|
"""Verify that running status.sh leaves YAML and DB unchanged."""
|
|
# Seed DB first so it exists
|
|
run_mutation(mam_sandbox, "d['herdr_sessions'] = []")
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
|
|
y_content_1 = yaml_path.read_text()
|
|
|
|
conn = sqlite3.connect(str(db_path))
|
|
d_content_1 = conn.execute("SELECT * FROM sessions").fetchall()
|
|
conn.close()
|
|
|
|
status_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"
|
|
subprocess.run(["bash", str(status_script)], capture_output=True, text=True)
|
|
|
|
y_content_2 = yaml_path.read_text()
|
|
conn = sqlite3.connect(str(db_path))
|
|
d_content_2 = conn.execute("SELECT * FROM sessions").fetchall()
|
|
conn.close()
|
|
|
|
assert y_content_1 == y_content_2
|
|
assert d_content_1 == d_content_2
|
|
|
|
def test_comp_status_yaml_to_json_structure(mam_sandbox):
|
|
"""Verify YAML-to-JSON structure translation preserves all nested maps."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-struct-creator-claude',
|
|
'status': 'running',
|
|
'pane': {
|
|
'index': 0,
|
|
'pid': 1234,
|
|
'cmd': 'claude',
|
|
'cwd': '/tmp'
|
|
}
|
|
}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
status_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"
|
|
res = subprocess.run(["bash", str(status_script), "--json"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
data = json.loads(res.stdout)
|
|
session = data["sessions_detail"][0]
|
|
assert session["pane_pid"] == 1234
|
|
assert session["pane_cwd"] == "/tmp"
|
|
|
|
|
|
def test_comp_status_displays_socket_and_workspace_columns(mam_sandbox):
|
|
"""T12: Verify status.sh displays distinct SOCKET and WORKSPACE columns."""
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-cols-creator-claude',
|
|
'status': 'running',
|
|
'herdr_session': 'socket-AAA',
|
|
'herdr_server': 'socket-AAA',
|
|
'herdr_workspace': 'label-BBB',
|
|
'pane': {'cwd': '/tmp', 'cmd': 'claude'}
|
|
}]
|
|
"""
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"
|
|
res = subprocess.run(["bash", str(script_path)], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
assert "SOCKET" in res.stdout
|
|
assert "WORKSPACE" in res.stdout
|
|
assert "socket-AAA" in res.stdout
|
|
assert "label-BBB" in res.stdout
|
|
|
|
# Also verify --json has herdr_workspace
|
|
res_json = subprocess.run(["bash", str(script_path), "--json"], capture_output=True, text=True)
|
|
assert res_json.returncode == 0
|
|
data = json.loads(res_json.stdout)
|
|
assert data["sessions_detail"][0]["herdr_workspace"] == "label-BBB"
|
|
assert data["sessions_detail"][0]["server"] == "socket-AAA"
|
|
|
|
|
|
# ==============================================================================
|
|
# FEATURE 5: Monitor/Reconcile (7 Test Cases)
|
|
# ==============================================================================
|
|
|
|
def test_comp_monitor_concurrency_lock(mam_sandbox):
|
|
"""Verify that multiple subscriber reconciles cannot run concurrently."""
|
|
# Reconcile subscribe script tries to acquire .mam/monitor.lock
|
|
# Let's write a mock subscriber that acquires the lock
|
|
workspace_root = mam_sandbox
|
|
lock_file_path = workspace_root / ".mam" / "monitor.lock"
|
|
|
|
# Hold lock in Python
|
|
import fcntl
|
|
lock_file = open(lock_file_path, 'w')
|
|
fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
|
|
# Now execute reconcile subscribe loop, it should exit immediately
|
|
reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
res = subprocess.run(["bash", str(reconcile_script), "--subscribe", "--idle-timeout", "1"], capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
assert "MQTT Monitor: another subscriber is already running" in res.stdout
|
|
|
|
lock_file.close()
|
|
|
|
def test_comp_monitor_sqlite_journal_wal(mam_sandbox):
|
|
"""Verify that SQLite database defaults to WAL mode on standard filesystems."""
|
|
mutation = """
|
|
d['herdr_sessions'] = []
|
|
"""
|
|
# Runs atomic_dump_yaml which sets journal mode
|
|
res = run_mutation(mam_sandbox, mutation)
|
|
assert res.returncode == 0
|
|
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
conn.close()
|
|
assert mode.lower() == "wal"
|
|
|
|
def test_comp_monitor_sqlite_journal_delete_nfs(mam_sandbox):
|
|
"""Verify that SQLite database falls back to DELETE mode on NFS filesystems."""
|
|
mutation = """
|
|
d['herdr_sessions'] = []
|
|
"""
|
|
# Set MAM_IS_NFS env var
|
|
res = run_mutation(mam_sandbox, mutation, env={"MAM_IS_NFS": "true"})
|
|
assert res.returncode == 0
|
|
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
conn.close()
|
|
assert mode.lower() == "delete"
|
|
|
|
def test_comp_monitor_db_schema_auto_update(mam_sandbox):
|
|
"""Verify that table structure and indexes are automatically created if DB is deleted."""
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
if db_path.exists():
|
|
db_path.unlink()
|
|
|
|
mutation = """
|
|
d['herdr_sessions'] = []
|
|
"""
|
|
res = run_mutation(mam_sandbox, mutation)
|
|
assert res.returncode == 0
|
|
|
|
assert db_path.exists()
|
|
conn = sqlite3.connect(str(db_path))
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
|
|
tables = [row[0] for row in cursor.fetchall()]
|
|
assert "state" in tables
|
|
assert "sessions" in tables
|
|
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='index';")
|
|
indexes = [row[0] for row in cursor.fetchall()]
|
|
assert "idx_sessions_pane_cwd" in indexes
|
|
conn.close()
|
|
|
|
def test_comp_monitor_monotonic_seq_updates(mam_sandbox):
|
|
"""Verify sequence monotonic updates in mqtt_common."""
|
|
mqtt_common = get_mqtt_common(mam_sandbox)
|
|
|
|
# Setup job JSON in jobs registry dir
|
|
registry_dir = mam_sandbox / ".mam" / "jobs"
|
|
registry_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
job_id = "job-seq-123"
|
|
job_data = {
|
|
"job_id": job_id,
|
|
"status": "running",
|
|
"last_seq": 0
|
|
}
|
|
with open(registry_dir / f"{job_id}.json", "w") as f:
|
|
json.dump(job_data, f)
|
|
|
|
# Get sequence multiple times
|
|
s1 = mqtt_common.next_seq(job_id, str(registry_dir))
|
|
s2 = mqtt_common.next_seq(job_id, str(registry_dir))
|
|
s3 = mqtt_common.next_seq(job_id, str(registry_dir))
|
|
|
|
assert s1 == 1
|
|
assert s2 == 2
|
|
assert s3 == 3
|
|
|
|
# Reload and check
|
|
with open(registry_dir / f"{job_id}.json", "r") as f:
|
|
loaded = json.load(f)
|
|
assert loaded["last_seq"] == 3
|
|
|
|
def test_comp_monitor_auto_state_recovery(mam_sandbox, mock_herdr):
|
|
"""Verify that reconcile.sh auto-terminates a session if herdr is confirmed dead."""
|
|
# Write a running session in YAML
|
|
mutation = """
|
|
d['herdr_sessions'] = [{
|
|
'name': 'test-autorecovery-creator-claude',
|
|
'status': 'running',
|
|
'pane': {'cwd': 'WS_PLACEHOLDER'}
|
|
}]
|
|
""".replace("WS_PLACEHOLDER", str(mam_sandbox))
|
|
run_mutation(mam_sandbox, mutation)
|
|
|
|
# Run reconcile.sh --once --emit-diff (which runs the dry-run, which does NOT write but reports)
|
|
reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
res = subprocess.run(["bash", str(reconcile_script), "--once", "--emit-diff", "--dry-run"], capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res.returncode == 0
|
|
|
|
data = json.loads(res.stdout)
|
|
drifts = data["drifts"]
|
|
# Should detect drift class A: herdr gone
|
|
assert any(dr["class"] == "A" and dr["name"] == "test-autorecovery-creator-claude" for dr in drifts)
|
|
|
|
# Now run reconcile.sh --once --emit-diff (without --dry-run) to trigger write
|
|
res_write = subprocess.run(["bash", str(reconcile_script), "--once", "--emit-diff"], capture_output=True, text=True, cwd=str(mam_sandbox))
|
|
assert res_write.returncode == 0
|
|
|
|
# Verify session is marked terminated in the SQLite DB
|
|
db_path = mam_sandbox / ".mam" / "agent-sessions.db"
|
|
conn = sqlite3.connect(str(db_path))
|
|
row = conn.execute("SELECT status FROM sessions WHERE name='test-autorecovery-creator-claude'").fetchone()
|
|
assert row[0] == "terminated"
|
|
conn.close()
|
|
|
|
def test_comp_stop_usage_matches_parser(mam_sandbox):
|
|
"""C-6: help text and parser must not drift apart."""
|
|
script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
|
|
# 에이전트 접미사 추론(:91-100)이 성립하는 이름 — rc=2 의 다섯 원인 중
|
|
# 'cannot infer agent'(:98)를 배제하기 위함 (Challenge 8b6b574f)
|
|
VALID = "test-project-creator-claude"
|
|
|
|
# 1) --help 는 성공하고, 폐지된 플래그를 광고하지 않는다
|
|
res = subprocess.run(["bash", str(script), "--help"], capture_output=True, text=True)
|
|
assert res.returncode == 0
|
|
for dead in ("--mode", "--capture-id", "--graceful"):
|
|
assert dead not in res.stdout, f"usage() still advertises {dead}"
|
|
|
|
# 1b) 검증기가 받는 에이전트는 전부 도움말에 나온다 (Rev.2 M3)
|
|
for agent in ("claude", "agy", "hermes", "cline"):
|
|
assert agent in res.stdout, f"usage() omits supported agent {agent}"
|
|
|
|
# 2) 도움말이 광고하는 플래그는 전부 파서가 받는다
|
|
# rc=2 는 5가지 원인을 공유하므로 stderr 메시지로 직접 지목한다
|
|
for flag, args in (("--reason", ["--reason", "x"]),
|
|
("--purge-conversation", ["--purge-conversation"]),
|
|
("--yes", ["--yes"]),
|
|
("--agent", ["--agent", "hermes"]),
|
|
("--herdr-session", ["--herdr-session", "isolated-sess"]),
|
|
("--herdr-workspace", ["--herdr-workspace", "isolated-ws"])):
|
|
r = subprocess.run(["bash", str(script), "--session", VALID] + args,
|
|
capture_output=True, text=True)
|
|
assert "unknown arg" not in r.stderr, f"usage() advertises {flag} but parser rejects it: {r.stderr}"
|
|
assert "deprecated" not in r.stderr, f"usage() advertises deprecated {flag}: {r.stderr}"
|
|
assert r.returncode != 2, f"{flag} -> rc=2: {r.stderr}"
|
|
|
|
# 3) 폐지된 플래그는 전용 메시지와 함께 rc=2 로 거부된다 (특별 취급 유지)
|
|
for dead in ("--mode", "--capture-id", "--graceful"):
|
|
r = subprocess.run(["bash", str(script), "--session", VALID, dead, "hard"],
|
|
capture_output=True, text=True)
|
|
assert r.returncode == 2
|
|
assert "deprecated" in r.stderr
|
|
|
|
# 4) 헤더 주석도 폐지 플래그를 사용법으로 광고하지 않는다
|
|
head = "".join(script.read_text().splitlines(keepends=True)[:35])
|
|
assert "--mode soft|hard" not in head
|
|
|
|
|
|
def test_comp_reconcile_drift_b_populates_workspace_and_server(mam_sandbox, mock_herdr, mock_agents):
|
|
"""T11 / S10: Verify drift B auto-registration populates herdr_workspace and herdr_server."""
|
|
session_name = "canary-test-creator-claude"
|
|
state = {
|
|
"workspaces": [{"workspace_id": "w1", "label": "default", "cwd": str(mam_sandbox)}],
|
|
"agents": {
|
|
session_name: {
|
|
"name": session_name,
|
|
"status": "running",
|
|
"cwd": str(mam_sandbox),
|
|
"command": "claude",
|
|
"pid": 98765
|
|
}
|
|
},
|
|
"calls": []
|
|
}
|
|
with open(mock_herdr, "w") as f:
|
|
json.dump(state, f)
|
|
|
|
run_mutation(mam_sandbox, "d['herdr_sessions'] = []")
|
|
|
|
reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
res = subprocess.run(["bash", str(reconcile_script)], capture_output=True, text=True, cwd=str(mam_sandbox), env={**os.environ, "WORKSPACE_ROOT": str(mam_sandbox)})
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
|
|
import yaml
|
|
with open(yaml_path) as f:
|
|
data = yaml.safe_load(f)
|
|
sessions = data.get("herdr_sessions", [])
|
|
matching = [s for s in sessions if s.get("name") == session_name]
|
|
assert len(matching) == 1
|
|
s = matching[0]
|
|
assert s["herdr_session"] == "default"
|
|
assert s["herdr_server"] == "default"
|
|
assert s["herdr_workspace"] != ""
|
|
assert s["herdr_workspace"] != "-"
|
|
|