import os import json import uuid import pytest import subprocess import yaml from pathlib import Path 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, cwd=str(mam_sandbox)) return res def test_t1_create_assigned_session(mam_sandbox, mock_herdr, mock_agents): """T-1: create immediately has claude_session_id_own as uuid, source=assigned, cmd_full with --session-id""" ws = str(mam_sandbox / "project") os.makedirs(ws, exist_ok=True) script_path = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" cmd = [ "bash", str(script_path), "--workspace", ws, "--agent", "claude", "--role", "creator", "--session", "test-t1-claude", "--no-onboard" ] res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0, f"create failed: {res.stderr}" yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" with open(yaml_path) as f: data = yaml.safe_load(f) session = [s for s in data.get("herdr_sessions", []) if s.get("name") == "test-t1-claude"][0] cid = session.get("claude_session_id_own") assert cid is not None and len(cid) > 0 assert session.get("session_id_source") == "assigned" assert session.get("session_id_verified") is False assert "--session-id" in session.get("pane", {}).get("cmd_full", "") def test_t2_assigned_id_promotion(mam_sandbox, mock_herdr, mock_agents): """T-2: assigned ID is unverified before transcript appears, verified after transcript materializes and reconcile runs""" ws = str(mam_sandbox / "project2") os.makedirs(ws, exist_ok=True) create_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" cmd = [ "bash", str(create_script), "--workspace", ws, "--agent", "claude", "--role", "creator", "--session", "test-t2-claude", "--no-onboard" ] res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0, f"create failed: {res.stderr}" yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" with open(yaml_path) as f: data = yaml.safe_load(f) session = [s for s in data.get("herdr_sessions", []) if s.get("name") == "test-t2-claude"][0] cid = session.get("claude_session_id_own") assert session.get("session_id_verified") is False # Simulate transcript materialization on disk ws_key = os.path.realpath(ws).replace("/", "-").replace("_", "-") c_dir = os.path.expanduser("~/.claude/projects") t_dir = os.path.join(c_dir, ws_key) os.makedirs(t_dir, exist_ok=True) with open(os.path.join(t_dir, f"{cid}.jsonl"), "w") as f: f.write(json.dumps({"type": "queue-operation", "sessionId": cid}) + "\n") f.write(json.dumps({"type": "user", "sessionId": cid, "cwd": ws}) + "\n") # Run reconcile once res2 = subprocess.run(["bash", str(reconcile_script), "--once"], capture_output=True, text=True, cwd=str(mam_sandbox)) assert res2.returncode == 0, f"reconcile failed: {res2.stderr}" with open(yaml_path) as f: data = yaml.safe_load(f) session = [s for s in data.get("herdr_sessions", []) if s.get("name") == "test-t2-claude"][0] assert session.get("session_id_verified") is True assert session.get("last_visible_status") == "pinned" def test_t3_different_cwd_transcript_not_pinned(mam_sandbox): """T-3: transcript belonging to a different cwd is not pinned""" ws1 = str(mam_sandbox / "ws1") ws2 = str(mam_sandbox / "ws2") os.makedirs(ws1, exist_ok=True) os.makedirs(ws2, exist_ok=True) test_uuid = str(uuid.uuid4()) ws1_key = os.path.realpath(ws1).replace("/", "-").replace("_", "-") c_dir = os.path.expanduser("~/.claude/projects") t_dir = os.path.join(c_dir, ws1_key) os.makedirs(t_dir, exist_ok=True) t_file = os.path.join(t_dir, f"{test_uuid}.jsonl") with open(t_file, "w") as f: f.write(json.dumps({"type": "queue-operation", "sessionId": test_uuid}) + "\n") f.write(json.dumps({"type": "user", "sessionId": test_uuid, "cwd": ws2}) + "\n") lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" cmd = [ "bash", "-c", f'source "{lib_path}" && verify_session_uuid "{ws1}" "claude" "{test_uuid}" \'{{"pane":{{"cwd":"{ws1}"}}}}\' discover' ] res = subprocess.run(cmd, cwd=str(mam_sandbox)) assert res.returncode != 0 def test_t4_ambiguous_candidates_reported(mam_sandbox, mock_herdr, mock_agents): """T-4: 2 candidate transcripts -> not pinning + C-ambiguous reported""" ws = str(mam_sandbox / "ambig_ws") os.makedirs(ws, exist_ok=True) session_name = "ambig-creator-claude" create_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" cmd = [ "bash", str(create_script), "--workspace", ws, "--agent", "claude", "--role", "creator", "--session", session_name, "--no-onboard" ] subprocess.run(cmd, check=True, cwd=str(mam_sandbox)) # Reset own ID to None to simulate legacy session without assigned ID mutation = f""" for s in d.get('herdr_sessions', []): if s.get('name') == '{session_name}': s['claude_session_id_own'] = None s['session_id_source'] = None """ res = run_mutation(mam_sandbox, mutation) assert res.returncode == 0 ws_key = os.path.realpath(ws).replace("/", "-").replace("_", "-") c_dir = os.path.expanduser("~/.claude/projects") t_dir = os.path.join(c_dir, ws_key) os.makedirs(t_dir, exist_ok=True) u1, u2 = str(uuid.uuid4()), str(uuid.uuid4()) for u in (u1, u2): with open(os.path.join(t_dir, f"{u}.jsonl"), "w") as f: f.write(json.dumps({"type": "queue-operation", "sessionId": u}) + "\n") f.write(json.dumps({"type": "user", "sessionId": u, "cwd": ws}) + "\n") reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" res2 = subprocess.run(["bash", str(reconcile_script), "--once"], capture_output=True, text=True, cwd=str(mam_sandbox)) assert res2.returncode == 0 yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" with open(yaml_path) as f: d = yaml.safe_load(f) s = [x for x in d["herdr_sessions"] if x["name"] == session_name][0] assert s.get("claude_session_id_own") is None assert "ambiguous" in s.get("last_visible_status", "") def test_t5_custom_session_name_pinned(mam_sandbox, mock_herdr, mock_agents): """T-5: custom --session name also gets pinned""" ws = str(mam_sandbox / "custom_name_ws") os.makedirs(ws, exist_ok=True) create_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" reconcile_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" cmd = [ "bash", str(create_script), "--workspace", ws, "--agent", "claude", "--role", "creator", "--session", "my-custom-session-name", "--no-onboard" ] res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0 yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" with open(yaml_path) as f: data = yaml.safe_load(f) session = [s for s in data.get("herdr_sessions", []) if s.get("name") == "my-custom-session-name"][0] cid = session.get("claude_session_id_own") # Simulate transcript materialization on disk ws_key = os.path.realpath(ws).replace("/", "-").replace("_", "-") c_dir = os.path.expanduser("~/.claude/projects") t_dir = os.path.join(c_dir, ws_key) os.makedirs(t_dir, exist_ok=True) with open(os.path.join(t_dir, f"{cid}.jsonl"), "w") as f: f.write(json.dumps({"type": "queue-operation", "sessionId": cid}) + "\n") f.write(json.dumps({"type": "user", "sessionId": cid, "cwd": ws}) + "\n") res2 = subprocess.run(["bash", str(reconcile_script), "--once"], capture_output=True, text=True, cwd=str(mam_sandbox)) assert res2.returncode == 0 with open(yaml_path) as f: data = yaml.safe_load(f) session = [s for s in data.get("herdr_sessions", []) if s.get("name") == "my-custom-session-name"][0] assert session.get("session_id_verified") is True def test_t6_viewport_no_path_degraded(mam_sandbox): """T-6: viewport without path returns 2 (degraded)""" lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" cmd = [ "bash", "-c", f'source "{lib_path}" && _pane_capture() {{ echo "some random text without slash paths"; }} && _herdr() {{ return 0; }} && verify_tui_viewport "sess" "claude" "/tmp/myproj"' ] res = subprocess.run(cmd, cwd=str(mam_sandbox)) assert res.returncode == 2 def test_t7_viewport_different_project_mismatch(mam_sandbox): """T-7: viewport showing DIFFERENT project returns 1 (mismatch)""" lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" cmd = [ "bash", "-c", f'source "{lib_path}" && _pane_capture() {{ echo "Accessing workspace: /Users/godopu16/other_project"; }} && _herdr() {{ return 0; }} && verify_tui_viewport "sess" "claude" "/Users/godopu16/myproj"' ] res = subprocess.run(cmd, cwd=str(mam_sandbox)) assert res.returncode == 1 def test_t8_resume_unmaterialized_assigned_id(mam_sandbox, mock_herdr, mock_agents): """T-8: unmaterialized assigned ID resume uses --session-id instead of -r""" ws = str(mam_sandbox / "resume_unmat") os.makedirs(ws, exist_ok=True) create_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" stop_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh" resume_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh" # Create session cmd = [ "bash", str(create_script), "--workspace", ws, "--agent", "claude", "--role", "creator", "--session", "test-t8-claude", "--no-onboard" ] subprocess.run(cmd, check=True, cwd=str(mam_sandbox)) # Stop session without sending any prompt subprocess.run(["bash", str(stop_script), "--session", "test-t8-claude", "--agent", "claude"], check=True, cwd=str(mam_sandbox)) # Remove transcript if created by mock yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" with open(yaml_path) as f: d = yaml.safe_load(f) session = [s for s in d["herdr_sessions"] if s["name"] == "test-t8-claude"][0] uuid_val = session["claude_session_id_own"] ws_key = os.path.realpath(ws).replace("/", "-").replace("_", "-") jsonl_path = os.path.expanduser(f"~/.claude/projects/{ws_key}/{uuid_val}.jsonl") if os.path.exists(jsonl_path): os.remove(jsonl_path) # Run dry-run resume res = subprocess.run([ "bash", str(resume_script), "--workspace", ws, "--agent", "claude", "--session", "test-t8-claude", "--dry-run" ], capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0 assert "--session-id" in res.stdout def test_t9_pane_capture_json_unwrap(mam_sandbox): """T-9: _pane_capture returns real text, unwrapping JSON envelope""" raw_json = json.dumps({ "id": "cli:agent:read", "result": { "read": { "format": "text", "text": "Accessing workspace:\n/Users/godopu16/myproj" } } }) lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" cmd = [ "bash", "-c", f'source "{lib_path}" && _sks_herdr() {{ echo \'{raw_json}\'; }} && _pane_capture "sess"' ] res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0 assert "Accessing workspace:" in res.stdout assert "{" not in res.stdout def test_t10_resume_workspace_paths(mam_sandbox, mock_herdr, mock_agents): """T-10: resume with various workspace path forms (absolute, trailing slash, dotdot, relative, symlink)""" real_ws = mam_sandbox / "wsprobe" / "real" os.makedirs(real_ws, exist_ok=True) link_ws = mam_sandbox / "wsprobe" / "link" os.symlink(real_ws, link_ws) create_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" stop_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh" resume_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resume_session.sh" ws_str = str(real_ws) cmd = [ "bash", str(create_script), "--workspace", ws_str, "--agent", "claude", "--role", "creator", "--session", "test-t10-claude", "--no-onboard" ] subprocess.run(cmd, check=True, cwd=str(mam_sandbox)) subprocess.run(["bash", str(stop_script), "--session", "test-t10-claude", "--agent", "claude"], check=True, cwd=str(mam_sandbox)) # Test path variations variations = [ ws_str, ws_str + "/", str(real_ws / ".." / "real"), str(link_ws) ] for v in variations: res = subprocess.run([ "bash", str(resume_script), "--workspace", v, "--agent", "claude", "--session", "test-t10-claude", "--dry-run" ], capture_output=True, text=True, cwd=str(mam_sandbox)) assert res.returncode == 0, f"failed for variation: {v}" assert f"would spawn:" in res.stdout def test_t12_other_workspace_assigned_row_revalidate_fails(mam_sandbox): """T-12: assigned row belonging to another workspace does not pass revalidate""" ws1 = str(mam_sandbox / "target_ws") ws2 = str(mam_sandbox / "other_ws") os.makedirs(ws1, exist_ok=True) os.makedirs(ws2, exist_ok=True) u_val = str(uuid.uuid4()) row_json = json.dumps({ "session_id_source": "assigned", "session_id_verified": False, "pane": {"cwd": ws2} }) lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" cmd = [ "bash", "-c", f'source "{lib_path}" && verify_session_uuid "{ws1}" "claude" "{u_val}" \'{row_json}\' revalidate' ] res = subprocess.run(cmd, cwd=str(mam_sandbox)) assert res.returncode != 0 def test_t13_mam_workspace_key_equivalence(mam_sandbox): """T-13: mam_workspace_key (shell) matches python workspace_key across absolute, trailing slash, dotdot, and symlink""" real_ws = mam_sandbox / "ws_real" os.makedirs(real_ws, exist_ok=True) link_ws = mam_sandbox / "ws_link" os.symlink(real_ws, link_ws) lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" variations = [ str(real_ws), str(real_ws) + "/", str(real_ws / ".." / "ws_real"), str(link_ws) ] for v in variations: cmd = ["bash", "-c", f'source "{lib_path}" && mam_workspace_key "{v}"'] res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(mam_sandbox)) sh_key = res.stdout.strip() # Python realpath key abs_path = os.path.realpath(v) py_key = abs_path.replace("/", "-").replace("_", "-") assert sh_key == py_key, f"Mismatch for {v}: shell={sh_key}, py={py_key}"