feat(create/monitor): implement automatic UUID materialization, auto-pinning, and drift-C2 resolution (13/13 PASS)

tests/conftest.py

@@ -248,7 +248,16 @@ elif cmd1 == "agent":
         }
         
         # Generate session/conversation UUID and files to simulate agent startup
-        session_uuid = str(uuid.uuid4())
+        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",
@@ -261,7 +270,7 @@ elif cmd1 == "agent":
 
         # 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_abs = os.path.realpath(cwd or "TMP_PATH_PLACEHOLDER")
         ws_key = ws_abs.replace('/', '-').replace('_', '-')
 
         if agent_type == "claude":

tests/test_tier3_integration.py

@@ -107,7 +107,7 @@ def test_integration_stop_purge_combination(mam_sandbox, mock_herdr, mock_agents
     claude_uuid = session["claude_session_id_own"]
     
     # Locate dynamically generated conversation file in claude projects
-    key = str(tmp_path).replace('/', '-').replace('_', '-')
+    key = os.path.realpath(str(tmp_path)).replace('/', '-').replace('_', '-')
     proj_dir = tmp_path / ".claude" / "projects" / key
     assert proj_dir.exists(), f"Expected projects directory {proj_dir} to exist"
     

tests/test_uuid_target.py

@@ -0,0 +1,429 @@
+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_t11_legacy_isolation_row(mam_sandbox, mock_herdr, mock_agents):
+    """T-11: legacy isolation row resolved from isolation.root"""
+    iso_root = str(mam_sandbox / "iso_root")
+    ws = str(mam_sandbox / "iso_ws")
+    os.makedirs(ws, exist_ok=True)
+
+    ws_key = os.path.realpath(ws).replace("/", "-").replace("_", "-")
+    proj_dir = os.path.join(iso_root, "projects", ws_key)
+    os.makedirs(proj_dir, exist_ok=True)
+    u_val = str(uuid.uuid4())
+    with open(os.path.join(proj_dir, f"{u_val}.jsonl"), "w") as f:
+        f.write(json.dumps({"type": "queue-operation", "sessionId": u_val}) + "\n")
+        f.write(json.dumps({"type": "user", "sessionId": u_val, "cwd": ws}) + "\n")
+
+    mutation = f"""
+entry = {{
+    'name': 'iso-session',
+    'status': 'stopped',
+    'role': 'creator',
+    'claude_session_id_own': '{u_val}',
+    'isolation': {{'root': '{iso_root}', 'uuid': 'iso-uuid-1'}},
+    'pane': {{'cwd': '{ws}'}}
+}}
+d.setdefault('herdr_sessions', []).append(entry)
+"""
+    res_m = run_mutation(mam_sandbox, mutation)
+    assert res_m.returncode == 0, f"mutation failed: {res_m.stderr}"
+
+    resolve_script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-resume" / "scripts" / "resolve_session_id.sh"
+    cmd = [
+        "bash", str(resolve_script),
+        "--workspace", ws,
+        "--agent", "claude",
+        "--session", "iso-session"
+    ]
+    res = subprocess.run(cmd, capture_output=True, text=True, cwd=str(mam_sandbox))
+    assert res.returncode == 0
+    assert res.stdout.strip() == u_val
+
+
+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}"