multi-agent-mux/tests/test_a4_adapter_contract.py

# test_a4_adapter_contract.py — Contract tests for BaseAgentAdapter and resolve_home

import os, pytest
from lib_py.paths import resolve_home
from lib_py.agents.registry import get_adapter, own_key, agent_of_row

def test_resolve_home_contract():
    # 1. When explicit home_dir is passed
    assert resolve_home('/custom/home') == '/custom/home'

    # 2. When HOME_DIR env var is set
    orig = os.environ.get('HOME_DIR')
    try:
        os.environ['HOME_DIR'] = '/env/home'
        assert resolve_home() == '/env/home'
    finally:
        if orig is None:
            os.environ.pop('HOME_DIR', None)
        else:
            os.environ['HOME_DIR'] = orig

    # 3. When HOME_DIR env var is NOT set, falls back to HOME / expanduser (~), NOT '' or '/'
    env_copy = os.environ.copy()
    env_copy.pop('HOME_DIR', None)
    home_val = env_copy.get('HOME') or os.path.expanduser('~')
    assert resolve_home() == home_val

def test_agent_adapter_registry():
    for agent in ('claude', 'agy', 'hermes', 'cline'):
        adapter = get_adapter(agent)
        assert adapter is not None
        assert adapter.name == agent
        assert adapter.own_key == f"{agent}_{'session' if agent == 'claude' else 'conversation'}_id_own"
        assert own_key(agent) == adapter.own_key

def test_agent_of_row_priority():
    # Priority 1: Explicit 'agent' field
    row1 = {'agent': 'agy', 'name': 'some-name-creator-claude'}
    assert agent_of_row(row1) == 'agy'

    # Priority 2: Session name suffix
    row2 = {'name': 'my-workspace-planner-hermes'}
    assert agent_of_row(row2) == 'hermes'

    # Priority 3: pane.cmd exact match
    row3 = {'pane': {'cmd': 'cline'}}
    assert agent_of_row(row3) == 'cline'

def test_adapter_required_properties():
    from lib_py.agents.base import BaseAgentAdapter
    base = BaseAgentAdapter()
    for prop in ('name', 'own_key', 'ready_tokens', 'exit_key', 'delegate_agent_key', 'identity_cache_fields'):
        with pytest.raises(NotImplementedError):
            getattr(base, prop)

    expected = {
        'claude': ('Anthropic|Assistant|Chat|Welcome', '/exit', 'claude-code', ('session_id', 'session_jsonl', 'session_size_bytes', 'session_lines')),
        'agy': ('Antigravity', 'Exit', 'antigravity-cli', ('conversation_id', 'conversation_db', 'conversation_brain_dir')),
        'hermes': ('Hermes', '/exit', 'hermes-agent', ('session_id',)),
        'cline': ('Cline|history|Chat|What can I do|slash commands', '/exit', 'cline-agent', ('session_id',)),
    }
    for agent, (toks, exitk, delk, cache_f) in expected.items():
        adapter = get_adapter(agent)
        assert adapter is not None
        assert adapter.ready_tokens == toks
        assert adapter.exit_key == exitk
        assert adapter.delegate_agent_key == delk
        assert adapter.identity_cache_fields == cache_f

def test_facts_bridge_eval_contract():
    import subprocess, sys
    from pathlib import Path
    env = os.environ.copy()
    skills_dir = str(Path(__file__).resolve().parent.parent / ".agents" / "skills")
    env["PYTHONPATH"] = f"{skills_dir}:{env.get('PYTHONPATH', '')}"
    for agent in ('claude', 'agy', 'hermes', 'cline'):
        res = subprocess.run([sys.executable, "-m", "lib_py.agents", "facts", agent], capture_output=True, text=True, env=env)
        assert res.returncode == 0
        facts_output = res.stdout

        # Verify eval in bash with set -euo pipefail
        bash_cmd = f"""
set -euo pipefail
eval {shlex_quote(facts_output)}
echo "AGENT=$MAM_AGENT_NAME|OWN=$MAM_OWN_KEY|TOK=$MAM_READY_TOKENS|EXIT=$MAM_EXIT_KEY|DEL=$MAM_DELEGATE_AGENT_KEY|PH=$MAM_INPUT_PLACEHOLDER"
"""
        res_bash = subprocess.run(["bash", "-c", bash_cmd], capture_output=True, text=True)
        assert res_bash.returncode == 0, f"Bash eval failed for {agent}:\nStdout: {res_bash.stdout}\nStderr: {res_bash.stderr}"
        if agent == 'cline':
            assert "PH=Ask anything..." in res_bash.stdout

def shlex_quote(s):
    import shlex
    return shlex.quote(s)

def test_purge_artifacts_composite(tmp_path):
    import sqlite3
    from lib_py.agents.base import DiscoveryContext
    ws = str(tmp_path / "ws")
    home = str(tmp_path / "home")
    os.makedirs(ws, exist_ok=True)
    os.makedirs(home, exist_ok=True)

    # 1. Claude
    claude_adapter = get_adapter('claude')
    claude_ctx = DiscoveryContext(workspace=ws, agent_name='claude', home_dir=home)
    c_path = claude_adapter.artifact_path('uuid-c', claude_ctx)
    os.makedirs(os.path.dirname(c_path), exist_ok=True)
    with open(c_path, 'w') as f:
        f.write('{"sessionId": "uuid-c"}')
    assert os.path.exists(c_path)
    purged_c = claude_adapter.purge_artifacts('uuid-c', claude_ctx)
    assert len(purged_c) == 1
    assert not os.path.exists(c_path)

    # 2. Agy (both DB file and brain dir)
    agy_adapter = get_adapter('agy')
    agy_ctx = DiscoveryContext(workspace=ws, agent_name='agy', home_dir=home)
    agy_db = agy_adapter.artifact_path('uuid-a', agy_ctx)
    os.makedirs(os.path.dirname(agy_db), exist_ok=True)
    with open(agy_db, 'w') as f:
        f.write('mock db')
    agy_brain = f"{home}/.gemini/antigravity-cli/brain/uuid-a"
    os.makedirs(agy_brain, exist_ok=True)
    with open(f"{agy_brain}/note.txt", 'w') as f:
        f.write('brain note')
    purged_a = agy_adapter.purge_artifacts('uuid-a', agy_ctx)
    assert len(purged_a) == 2
    assert not os.path.exists(agy_db)
    assert not os.path.exists(agy_brain)

    # 3. Hermes (JSON file and SQLite rows)
    hermes_adapter = get_adapter('hermes')
    hermes_ctx = DiscoveryContext(workspace=ws, agent_name='hermes', home_dir=home)
    h_json = hermes_adapter.artifact_path('uuid-h', hermes_ctx)
    os.makedirs(os.path.dirname(h_json), exist_ok=True)
    with open(h_json, 'w') as f:
        f.write('{}')
    h_db = f"{home}/.hermes/state.db"
    os.makedirs(os.path.dirname(h_db), exist_ok=True)
    conn = sqlite3.connect(h_db)
    conn.execute("CREATE TABLE IF NOT EXISTS sessions (id TEXT, cwd TEXT)")
    conn.execute("CREATE TABLE IF NOT EXISTS messages (session_id TEXT, msg TEXT)")
    conn.execute("INSERT INTO sessions VALUES (?, ?)", ('uuid-h', ws))
    conn.execute("INSERT INTO messages VALUES (?, ?)", ('uuid-h', 'hello'))
    conn.commit()
    conn.close()

    purged_h = hermes_adapter.purge_artifacts('uuid-h', hermes_ctx)
    assert len(purged_h) == 2
    assert not os.path.exists(h_json)
    conn = sqlite3.connect(h_db)
    assert conn.execute("SELECT count(*) FROM sessions WHERE id='uuid-h'").fetchone()[0] == 0
    assert conn.execute("SELECT count(*) FROM messages WHERE session_id='uuid-h'").fetchone()[0] == 0
    conn.close()

    # 4. Cline (sessions dir)
    cline_adapter = get_adapter('cline')
    cline_ctx = DiscoveryContext(workspace=ws, agent_name='cline', home_dir=home)
    cline_dir = f"{home}/.cline/data/sessions/uuid-cl"
    os.makedirs(cline_dir, exist_ok=True)
    with open(f"{cline_dir}/uuid-cl.json", 'w') as f:
        f.write('{"session_id": "uuid-cl"}')
    purged_cl = cline_adapter.purge_artifacts('uuid-cl', cline_ctx)
    assert len(purged_cl) == 1
    assert not os.path.exists(cline_dir)

def test_adapter_spawn_and_resume_specs():
    claude = get_adapter('claude')
    assert claude.spawn_spec('claude', 'u1') == 'claude --dangerously-skip-permissions --session-id u1'
    assert claude.spawn_spec('claude', '', use_wrapper=True) == 'claude --dangerously-skip-permissions'
    assert claude.resume_spec('claude', 'u1', materialized=True) == 'claude --dangerously-skip-permissions -r u1'
    assert claude.resume_spec('claude', 'u1', materialized=False) == 'claude --dangerously-skip-permissions --session-id u1'

    agy = get_adapter('agy')
    assert agy.spawn_spec('agy', 'u1') == 'agy --dangerously-skip-permissions'
    assert agy.resume_spec('agy', 'u1', materialized=True) == 'agy --dangerously-skip-permissions --conversation u1'

    hermes = get_adapter('hermes')
    assert hermes.spawn_spec('hermes', 'u1') == 'hermes'
    assert hermes.resume_spec('hermes', 'u1', materialized=True) == 'hermes --resume u1'

    cline = get_adapter('cline')
    assert cline.spawn_spec('cline', 'u1') == 'cline -i'
    assert cline.resume_spec('cline', 'u1', materialized=True) == 'cline -i --id u1'
    assert cline.resume_spec('cline', 'u1', materialized=False) == 'cline -i'

def test_adapter_auth_ok(tmp_path, monkeypatch):
    monkeypatch.setenv("HOME_DIR", str(tmp_path))
    # Claude auth runner
    claude = get_adapter('claude')
    assert claude.auth_ok(run_cmd=lambda cmd: (0, '{"loggedIn": true}', '')) is True
    assert claude.auth_ok(run_cmd=lambda cmd: (1, '{"loggedIn": false}', 'error')) is False

    # Agy auth file check
    agy = get_adapter('agy')
    assert agy.auth_ok() is False
    oauth_file = tmp_path / ".gemini" / "oauth_creds.json"
    oauth_file.parent.mkdir(parents=True, exist_ok=True)
    oauth_file.write_text("{}")
    assert agy.auth_ok() is True

    # Hermes & Cline always True
    assert get_adapter('hermes').auth_ok() is True
    assert get_adapter('cline').auth_ok() is True

def test_adapter_discover(tmp_path):
    import sqlite3
    from lib_py.agents.base import DiscoveryContext
    ws = str(tmp_path / "ws")
    home = str(tmp_path / "home")
    os.makedirs(ws, exist_ok=True)
    os.makedirs(home, exist_ok=True)

    # 1. Claude
    claude = get_adapter('claude')
    ctx_c = DiscoveryContext(workspace=ws, agent_name='claude', home_dir=home)
    c_proj = f"{ctx_c.claude_dir}/{ctx_c.ws_key}"
    os.makedirs(c_proj, exist_ok=True)
    with open(f"{c_proj}/u-c1.jsonl", 'w') as f:
        f.write('{"sessionId": "u-c1"}\n')
    assert claude.discover(ctx_c) == ['u-c1']

    # 2. Agy
    agy = get_adapter('agy')
    ctx_a = DiscoveryContext(workspace=ws, agent_name='agy', home_dir=home)
    a_db = f"{home}/.gemini/antigravity-cli/conversations/u-a1.db"
    os.makedirs(os.path.dirname(a_db), exist_ok=True)
    conn = sqlite3.connect(a_db)
    conn.execute("CREATE TABLE steps (id INT)")
    conn.execute("INSERT INTO steps VALUES (1)")
    conn.commit()
    conn.close()
    lc = f"{home}/.gemini/antigravity-cli/cache/last_conversations.json"
    os.makedirs(os.path.dirname(lc), exist_ok=True)
    with open(lc, 'w') as f:
        import json
        json.dump({ws: 'u-a1'}, f)
    assert agy.discover(ctx_a) == ['u-a1']

    # 3. Hermes
    hermes = get_adapter('hermes')
    ctx_h = DiscoveryContext(workspace=ws, agent_name='hermes', home_dir=home)
    h_db = f"{home}/.hermes/state.db"
    os.makedirs(os.path.dirname(h_db), exist_ok=True)
    conn = sqlite3.connect(h_db)
    conn.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)")
    conn.execute("INSERT INTO sessions (id, cwd) VALUES ('u-h1', ?)", (ws,))
    conn.commit()
    conn.close()
    assert hermes.discover(ctx_h) == ['u-h1']

    # 4. Cline
    cline = get_adapter('cline')
    ctx_cl = DiscoveryContext(workspace=ws, agent_name='cline', home_dir=home)
    cl_sess = f"{home}/.cline/data/sessions/u-cl1"
    os.makedirs(cl_sess, exist_ok=True)
    with open(f"{cl_sess}/u-cl1.json", 'w') as f:
        f.write('{"session_id": "u-cl1", "cwd": "' + ws + '"}')
    assert cline.discover(ctx_cl) == ['u-cl1']