# 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():
    EXPECTED_OWN_KEYS = {
        'claude': 'claude_session_id_own',
        'agy': 'agy_conversation_id_own',
        'hermes': 'hermes_conversation_id_own',
        'grok': 'grok_session_id_own',
        'opencode': 'opencode_session_id_own',
    }
    for agent, expected in EXPECTED_OWN_KEYS.items():
        adapter = get_adapter(agent)
        assert adapter is not None
        assert adapter.name == agent
        assert adapter.own_key == expected
        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': 'grok'}}
    assert agent_of_row(row3) == 'grok'

def test_agent_of_row_pane_cmd_binary_path_and_failure():
    # pane.cmd 가 절대 경로 형태여도 해석된다
    assert agent_of_row({'pane': {'cmd': '/usr/local/bin/agy'}}) == 'agy'
    # 세 경로 모두 실패하면 None — 호출자가 오류를 소유한다
    assert agent_of_row({}, session_name='bad-session-name') is None
    # 입양 조회용 match_cmd=False 에서는 pane.cmd 를 보지 않는다
    assert agent_of_row({'name': 'agy-creator-01', 'pane': {'cmd': 'agy'}},
                        match_cmd=False) is None

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|Claude Code|Opus|Sonnet|Haiku', '/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|Welcome to Hermes Agent|NOUS HERMES', '/exit', 'hermes-agent', ('session_id',)),
        'grok': ('Grok|xAI|Assistant|❯|>>>', '/exit', 'grok-build', ('session_id', 'session_jsonl')),
        'opencode': ('OpenCode|Chat', '/exit', 'opencode-cli', ('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
        # 2-Tier readiness default contract
        assert adapter.strong_ready_tokens == adapter.ready_tokens
        assert adapter.weak_ready_tokens == ''


def test_adapter_modal_tokens_contract():
    """F-8: Verify adapter-level modal_tokens contract (T-2d)."""
    expected_modal = {
        'claude': 'Try the new fullscreen renderer\\?',
        'agy': None,
        'hermes': None,
        'grok': None,
        'opencode': None,
    }
    for agent, expected in expected_modal.items():
        adapter = get_adapter(agent)
        assert adapter is not None
        assert adapter.modal_tokens == expected, f"modal_tokens mismatch for {agent}"


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', 'grok', 'opencode'):
        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|STRONG=$MAM_STRONG_READY_TOKENS|WEAK=$MAM_WEAK_READY_TOKENS"
"""
        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}"
        assert f"STRONG={get_adapter(agent).strong_ready_tokens}" in res_bash.stdout
        assert "WEAK=" in res_bash.stdout

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

def _seed_opencode_session(db_path, session_id, ws_path, epoch_sec):
    import sqlite3
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("PRAGMA foreign_keys = OFF")
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = {row[0] for row in cursor.fetchall()}
    if "session" not in tables:
        conn.execute("CREATE TABLE session (id TEXT PRIMARY KEY, directory TEXT, time_created INTEGER, cwd TEXT, created_at REAL)")
    if "project" in tables:
        cursor.execute("PRAGMA table_info(project)")
        p_cols = [col[1] for col in cursor.fetchall()]
        p_ins = ["id"]
        p_vals = ["default"]
        if "worktree" in p_cols:
            p_ins.append("worktree")
            p_vals.append(ws_path)
        if "time_created" in p_cols:
            p_ins.append("time_created")
            p_vals.append(0)
        if "time_updated" in p_cols:
            p_ins.append("time_updated")
            p_vals.append(0)
        conn.execute(f"INSERT OR IGNORE INTO project ({', '.join(p_ins)}) VALUES ({', '.join(['?']*len(p_vals))})", p_vals)

    cursor.execute("PRAGMA table_info(session)")
    cols = [col[1] for col in cursor.fetchall()]

    insert_cols = ["id"]
    vals = [session_id]
    if "project_id" in cols:
        insert_cols.append("project_id")
        vals.append("default")
    if "slug" in cols:
        insert_cols.append("slug")
        vals.append("default")
    if "title" in cols:
        insert_cols.append("title")
        vals.append("default")
    if "version" in cols:
        insert_cols.append("version")
        vals.append("1.0.0")
    if "directory" in cols:
        insert_cols.append("directory")
        vals.append(ws_path)
    if "cwd" in cols:
        insert_cols.append("cwd")
        vals.append(ws_path)
    if "time_created" in cols:
        insert_cols.append("time_created")
        vals.append(int(float(epoch_sec) * 1000))
    if "time_updated" in cols:
        insert_cols.append("time_updated")
        vals.append(int(float(epoch_sec) * 1000))
    if "created_at" in cols:
        insert_cols.append("created_at")
        vals.append(float(epoch_sec))

    placeholders = ", ".join(["?"] * len(vals))
    col_str = ", ".join(insert_cols)
    conn.execute(f"INSERT OR REPLACE INTO session ({col_str}) VALUES ({placeholders})", vals)
    conn.commit()
    conn.close()


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. Grok (session directory containing chat_history.jsonl)
    grok_adapter = get_adapter('grok')
    grok_ctx = DiscoveryContext(workspace=ws, agent_name='grok', home_dir=home)
    g_path = grok_adapter.artifact_path('uuid-g', grok_ctx)
    os.makedirs(os.path.dirname(g_path), exist_ok=True)
    with open(g_path, 'w') as f:
        f.write('{"session_id": "uuid-g"}')
    assert os.path.exists(g_path)
    purged_g = grok_adapter.purge_artifacts('uuid-g', grok_ctx)
    assert len(purged_g) == 1
    assert not os.path.exists(g_path)
    assert not os.path.exists(os.path.dirname(g_path))

    # 5. OpenCode (SQLite tables session, message, part)
    opencode_adapter = get_adapter('opencode')
    opencode_ctx = DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home)
    o_db = opencode_adapter.artifact_path('uuid-o', opencode_ctx)
    _seed_opencode_session(o_db, 'uuid-o', ws, 2000.0)
    conn = sqlite3.connect(o_db)
    cursor = conn.cursor()
    cursor.execute("PRAGMA foreign_keys = OFF")
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = {row[0] for row in cursor.fetchall()}
    if "message" not in tables:
        conn.execute("CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT)")
    if "part" not in tables:
        conn.execute("CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT)")

    cursor.execute("PRAGMA table_info(message)")
    m_cols = [c[1] for c in cursor.fetchall()]
    m_ins = ["id", "session_id"] if "id" in m_cols else ["session_id"]
    m_vals = ["m1", "uuid-o"] if "id" in m_cols else ["uuid-o"]
    if "time_created" in m_cols:
        m_ins.append("time_created")
        m_vals.append(0)
    if "time_updated" in m_cols:
        m_ins.append("time_updated")
        m_vals.append(0)
    if "data" in m_cols:
        m_ins.append("data")
        m_vals.append("{}")
    conn.execute(f"INSERT OR REPLACE INTO message ({', '.join(m_ins)}) VALUES ({', '.join(['?']*len(m_vals))})", m_vals)

    cursor.execute("PRAGMA table_info(part)")
    p_cols = [c[1] for c in cursor.fetchall()]
    p_ins = ["id", "session_id"] if "id" in p_cols else ["session_id"]
    p_vals = ["p1", "uuid-o"] if "id" in p_cols else ["uuid-o"]
    if "message_id" in p_cols:
        p_ins.append("message_id")
        p_vals.append("m1")
    if "time_created" in p_cols:
        p_ins.append("time_created")
        p_vals.append(0)
    if "time_updated" in p_cols:
        p_ins.append("time_updated")
        p_vals.append(0)
    if "data" in p_cols:
        p_ins.append("data")
        p_vals.append("{}")
    conn.execute(f"INSERT OR REPLACE INTO part ({', '.join(p_ins)}) VALUES ({', '.join(['?']*len(p_vals))})", p_vals)
    conn.commit()
    conn.close()

    purged_o = opencode_adapter.purge_artifacts('uuid-o', opencode_ctx)
    assert len(purged_o) == 1
    conn = sqlite3.connect(o_db)
    assert conn.execute("SELECT count(*) FROM session WHERE id='uuid-o'").fetchone()[0] == 0
    assert conn.execute("SELECT count(*) FROM message WHERE session_id='uuid-o'").fetchone()[0] == 0
    assert conn.execute("SELECT count(*) FROM part WHERE session_id='uuid-o'").fetchone()[0] == 0
    conn.close()

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 --yolo --accept-hooks'
    assert hermes.resume_spec('hermes', 'u1', materialized=True) == 'hermes --resume u1 --no-restore-cwd --yolo --accept-hooks'
    assert hermes.resume_spec('hermes', 'u1', materialized=False) == 'hermes --yolo --accept-hooks'

    grok = get_adapter('grok')
    assert grok.spawn_spec('grok', 'u1') == 'grok --session-id u1 --permission-mode bypassPermissions'
    assert grok.spawn_spec('grok', '') == 'grok --permission-mode bypassPermissions'
    assert grok.resume_spec('grok', 'u1', materialized=True) == 'grok --resume u1 --permission-mode bypassPermissions'
    assert grok.resume_spec('grok', 'u1', materialized=False) == 'grok --session-id u1 --permission-mode bypassPermissions'

    opencode = get_adapter('opencode')
    assert opencode.spawn_spec('opencode', 'u1') == 'opencode --auto --agent build'
    assert opencode.spawn_spec('opencode', '') == 'opencode --auto --agent build'
    assert opencode.resume_spec('opencode', 'u1', materialized=True) == 'opencode --session u1 --auto --agent build'
    assert opencode.resume_spec('opencode', 'u1', materialized=False) == 'opencode --auto --agent build'

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

    # Grok auth check (env var or auth.json file)
    grok = get_adapter('grok')
    assert grok.auth_ok() is False
    monkeypatch.setenv("XAI_API_KEY", "test-key")
    assert grok.auth_ok() is True
    monkeypatch.delenv("XAI_API_KEY", raising=False)
    grok_auth = tmp_path / ".grok" / "auth.json"
    grok_auth.parent.mkdir(parents=True, exist_ok=True)
    grok_auth.write_text("{}")
    assert grok.auth_ok() is True

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

    # OpenCode auth check (provider env var or auth.json file)
    opencode = get_adapter('opencode')
    assert opencode.auth_ok() is False
    monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key")
    assert opencode.auth_ok() is True
    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
    opencode_auth = tmp_path / ".local" / "share" / "opencode" / "auth.json"
    opencode_auth.parent.mkdir(parents=True, exist_ok=True)
    opencode_auth.write_text("{}")
    assert opencode.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. Grok
    grok = get_adapter('grok')
    ctx_g = DiscoveryContext(workspace=ws, agent_name='grok', home_dir=home)
    g_sess = f"{home}/.grok/sessions/{grok._ws_dir(ctx_g)}/u-g1"
    os.makedirs(g_sess, exist_ok=True)
    with open(f"{g_sess}/chat_history.jsonl", 'w') as f:
        f.write('{"session_id": "u-g1", "cwd": "' + ws + '"}\n')
    assert grok.discover(ctx_g) == ['u-g1']

    # 5. OpenCode
    opencode = get_adapter('opencode')
    ctx_o = DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home)
    o_db = opencode.artifact_path('u-o1', ctx_o)
    _seed_opencode_session(o_db, 'u-o1', ws, 2000.0)
    assert opencode.discover(ctx_o) == ['u-o1']

def test_cli_bridge_subcommands_and_quote_safety():
    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', '')}"

    # 1. spawn-spec
    res = subprocess.run([sys.executable, "-m", "lib_py.agents", "spawn-spec", "claude", "/path with spaces/claude", "uuid-test", "0"], capture_output=True, text=True, env=env)
    assert res.returncode == 0
    assert res.stdout.strip() == "/path with spaces/claude --dangerously-skip-permissions --session-id uuid-test"

    # 2. resume-spec with single quotes in workspace path
    res = subprocess.run([sys.executable, "-m", "lib_py.agents", "resume-spec", "claude", "/bin/claude", "uuid-test", "/tmp/bob's ws"], capture_output=True, text=True, env=env)
    assert res.returncode == 0
    assert res.stdout.strip() == "/bin/claude --dangerously-skip-permissions --session-id uuid-test"

    # 3. exit-key
    for agent, expected_key in [('claude', '/exit'), ('agy', 'Exit'), ('hermes', '/exit'), ('grok', '/exit'), ('opencode', '/exit')]:
        res = subprocess.run([sys.executable, "-m", "lib_py.agents", "exit-key", agent], capture_output=True, text=True, env=env)
        assert res.returncode == 0
        assert res.stdout.strip() == expected_key

def test_delegate_agent_resolution_and_fallback():
    import subprocess
    expected_map = {
        'claude': 'claude-code',
        'agy': 'antigravity-cli',
        'hermes': 'hermes-agent',
        'grok': 'grok-build',
        'opencode': 'opencode-cli',
    }
    # 1. Adapter property
    for agent, expected_key in expected_map.items():
        adapter = get_adapter(agent)
        assert adapter.delegate_agent_key == expected_key

    # 2. Shell fallback resolution when MAM_DELEGATE_AGENT_KEY is unset (R1 fallback)
    for agent, expected_key in expected_map.items():
        sh_snippet = f'''
        AGENT="{agent}"
        MAM_DELEGATE_AGENT_KEY=""
        delegate_agent="${{MAM_DELEGATE_AGENT_KEY:-}}"
        if [ -z "$delegate_agent" ]; then
          case "$AGENT" in
            claude) delegate_agent="claude-code" ;;
            hermes) delegate_agent="hermes-agent" ;;
            agy)    delegate_agent="antigravity-cli" ;;
            grok)   delegate_agent="grok-build" ;;
            opencode) delegate_agent="opencode-cli" ;;
            *) echo "ERROR: cannot resolve delegate agent key for '$AGENT'" >&2; exit 2 ;;
          esac
        fi
        echo "$delegate_agent"
        '''
        res = subprocess.run(["bash", "-c", sh_snippet], capture_output=True, text=True)
        assert res.returncode == 0
        assert res.stdout.strip() == expected_key

def test_wait_for_tui_ready_missing_tokens_diagnostic(mam_sandbox):
    import subprocess
    lib_sh = mam_sandbox / "skills" / "lib.sh"
    cmd = f'source "{lib_sh}" && wait_for_tui_ready "dummy-sess" "bogus-agent"'
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True)
    assert res.returncode != 0
    assert "no ready tokens for agent 'bogus-agent'" in res.stderr


def test_hermes_input_region_properties():
    adapter = get_adapter('hermes')
    assert adapter.input_prompt == '❯'
    assert adapter.input_placeholder == ''
    assert adapter.input_rule_pattern == '─{10,}'


def test_hermes_verify_artifact_started_at_epoch_filtering(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)

    hdb = f"{home}/.hermes/state.db"
    os.makedirs(os.path.dirname(hdb), exist_ok=True)
    conn = sqlite3.connect(hdb)
    conn.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at REAL)")
    conn.execute("INSERT INTO sessions VALUES ('u-old', ?, 1000.0)", (ws,))
    conn.execute("INSERT INTO sessions VALUES ('u-new', ?, 2000.0)", (ws,))
    conn.commit()
    conn.close()

    adapter = get_adapter('hermes')
    ctx_old = DiscoveryContext(workspace=ws, agent_name='hermes', home_dir=home, epoch=1500)
    assert adapter.verify_artifact('u-old', ctx_old) is False
    assert adapter.verify_artifact('u-new', ctx_old) is True


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

    hdb = f"{home}/.hermes/state.db"
    os.makedirs(os.path.dirname(hdb), exist_ok=True)
    conn = sqlite3.connect(hdb)
    conn.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at REAL)")
    # Direction 1: Old historical session (started_at=1000) + 1 new session (started_at=2000)
    conn.execute("INSERT INTO sessions VALUES ('u-hist', ?, 1000.0)", (ws,))
    conn.execute("INSERT INTO sessions VALUES ('u-fresh1', ?, 2000.0)", (ws,))
    conn.commit()

    epoch = 1500
    s_row = {'name': 'sess1', 'pane': {'cwd': ws}, 'herdr_session_epoch': epoch}

    # Query with epoch filter (Rev.2 pattern)
    rows = conn.execute(
        "SELECT id FROM sessions WHERE cwd=? AND started_at >= ? ORDER BY started_at DESC LIMIT 20",
        (ws, epoch)
    ).fetchall()
    valid = [u for (u,) in rows if verify_session_uuid(ws, 'hermes', u, s_row, home_dir=home, mode="discover")]
    assert valid == ['u-fresh1']
    assert len(valid) == 1  # Exactly 1, no false C-ambiguous!

    # Direction 2: Add a second concurrent fresh session (started_at=2100) -> must detect C-ambiguous
    conn.execute("INSERT INTO sessions VALUES ('u-fresh2', ?, 2100.0)", (ws,))
    conn.commit()
    conn.close()

    rows2 = sqlite3.connect(hdb).execute(
        "SELECT id FROM sessions WHERE cwd=? AND started_at >= ? ORDER BY started_at DESC LIMIT 20",
        (ws, epoch)
    ).fetchall()
    valid2 = [u for (u,) in rows2 if verify_session_uuid(ws, 'hermes', u, s_row, home_dir=home, mode="discover")]
    assert len(valid2) == 2  # Ambiguous detected correctly!


def test_hermes_verify_artifact_spawn_epoch_timing_f1(tmp_path):
    """F1 Regression: Ensure spawn-time epoch capture allows freshly started session to be verified."""
    import sqlite3, time
    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. Pre-spawn epoch capture (T0)
    spawn_epoch = int(time.time())

    # 2. Process spawns and writes session row (T0 + 0.1s)
    session_started_at = spawn_epoch + 0.1
    hdb = f"{home}/.hermes/state.db"
    os.makedirs(os.path.dirname(hdb), exist_ok=True)
    conn = sqlite3.connect(hdb)
    conn.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at REAL)")
    conn.execute("INSERT INTO sessions VALUES ('u-fresh', ?, ?)", (ws, session_started_at))
    conn.execute("INSERT INTO sessions VALUES ('u-old', ?, ?)", (ws, spawn_epoch - 3600.0))
    conn.commit()
    conn.close()

    adapter = get_adapter('hermes')
    # Context with pre-spawn epoch
    ctx_spawn = DiscoveryContext(workspace=ws, agent_name='hermes', home_dir=home, epoch=spawn_epoch)
    assert adapter.verify_artifact('u-fresh', ctx_spawn) is True
    assert adapter.verify_artifact('u-old', ctx_spawn) is False

    # Discover with pre-spawn epoch returns the fresh candidate only
    assert adapter.discover(ctx_spawn) == ['u-fresh']


def test_hermes_reconcile_full_block_integration(tmp_path):
    """F3: Verify hermes drift-C block behavior with sibling_claimed exclusion."""
    import sqlite3
    from lib_py.agents.base import DiscoveryContext
    from lib_py.verify_session import verify_session_uuid
    ws = str(tmp_path / "ws")
    home = str(tmp_path / "home")
    os.makedirs(ws, exist_ok=True)
    os.makedirs(home, exist_ok=True)

    hdb = f"{home}/.hermes/state.db"
    os.makedirs(os.path.dirname(hdb), exist_ok=True)
    conn = sqlite3.connect(hdb)
    conn.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at REAL)")
    conn.execute("INSERT INTO sessions VALUES ('u-claimed', ?, 2000.0)", (ws,))
    conn.execute("INSERT INTO sessions VALUES ('u-target', ?, 2001.0)", (ws,))
    conn.commit()
    conn.close()

    epoch = 1900
    s_eval = {
        'name': 'hermes-target-01',
        'pane': {'cwd': ws},
        'herdr_session_epoch': epoch,
        '_sibling_claimed_uuids': ['u-claimed']
    }

    # Simulate reconcile.sh candidate gathering
    conn = sqlite3.connect(hdb)
    rows = conn.execute(
        "SELECT id FROM sessions WHERE cwd=? AND started_at >= ? ORDER BY started_at DESC LIMIT 20",
        (ws, epoch)
    ).fetchall()
    conn.close()

    valid_candidates = []
    for (uuid,) in rows:
        if uuid in s_eval['_sibling_claimed_uuids']:
            continue
        if verify_session_uuid(ws, 'hermes', uuid, s_eval, home_dir=home, mode="discover"):
            valid_candidates.append(uuid)

    assert valid_candidates == ['u-target']


def test_opencode_input_region_properties():
    adapter = get_adapter('opencode')
    assert adapter.input_prompt == '❯'
    assert adapter.input_placeholder == ''
    assert adapter.input_rule_pattern == '─{10,}'


def test_opencode_verify_artifact_created_at_epoch_filtering(tmp_path):
    import sqlite3
    from lib_py.agents.base import DiscoveryContext
    ws = str(tmp_path / "ws")
    home = str(tmp_path / "home")
    adapter = get_adapter('opencode')
    ctx_old = DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home, epoch=1500)
    odb = adapter.artifact_path('u-old', ctx_old)
    _seed_opencode_session(odb, 'u-old', ws, 1000.0)
    _seed_opencode_session(odb, 'u-new', ws, 2000.0)

    assert adapter.verify_artifact('u-old', ctx_old) is False
    assert adapter.verify_artifact('u-new', ctx_old) is True


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

    adapter = get_adapter('opencode')
    ctx = DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home)
    odb = adapter.artifact_path('u-hist', ctx)
    _seed_opencode_session(odb, 'u-hist', ws, 1000.0)
    _seed_opencode_session(odb, 'u-fresh1', ws, 2000.0)

    epoch = 1500
    s_row = {'name': 'sess1', 'pane': {'cwd': ws}, 'herdr_session_epoch': epoch}

    valid = [u for u in adapter.discover(DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home, epoch=epoch, row=s_row)) if verify_session_uuid(ws, 'opencode', u, s_row, home_dir=home, mode="discover")]
    assert valid == ['u-fresh1']
    assert len(valid) == 1

    _seed_opencode_session(odb, 'u-fresh2', ws, 2100.0)

    valid2 = [u for u in adapter.discover(DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home, epoch=epoch, row=s_row)) if verify_session_uuid(ws, 'opencode', u, s_row, home_dir=home, mode="discover")]
    assert len(valid2) == 2


def test_opencode_reconcile_full_block_integration(tmp_path):
    """Verify opencode drift-C block behavior with sibling_claimed exclusion."""
    import sqlite3
    from lib_py.agents.registry import get_adapter
    from lib_py.agents.base import DiscoveryContext
    from lib_py.verify_session import verify_session_uuid
    ws = str(tmp_path / "ws")
    home = str(tmp_path / "home")
    os.makedirs(ws, exist_ok=True)
    os.makedirs(home, exist_ok=True)

    adapter = get_adapter('opencode')
    ctx = DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home)
    odb = adapter.artifact_path('u-target', ctx)
    _seed_opencode_session(odb, 'u-claimed', ws, 2000.0)
    _seed_opencode_session(odb, 'u-target', ws, 2001.0)

    epoch = 1900
    s_eval = {
        'name': 'opencode-target-01',
        'pane': {'cwd': ws},
        'herdr_session_epoch': epoch,
        '_sibling_claimed_uuids': ['u-claimed']
    }

    candidates = adapter.discover(DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home, epoch=epoch, row=s_eval))

    valid_candidates = []
    for uuid in candidates:
        if uuid in s_eval['_sibling_claimed_uuids']:
            continue
        if verify_session_uuid(ws, 'opencode', uuid, s_eval, home_dir=home, mode="discover"):
            valid_candidates.append(uuid)

    assert valid_candidates == ['u-target']


def test_opencode_real_schema_directory_and_time_created_ms(tmp_path):
    """Verify opencode adapter behavior with the real live schema (directory, time_created in ms)."""
    import sqlite3
    from lib_py.agents.registry import get_adapter
    from lib_py.agents.base import DiscoveryContext

    ws1 = str(tmp_path / "ws1")
    ws2 = str(tmp_path / "ws2")
    home = str(tmp_path / "home")
    os.makedirs(ws1, exist_ok=True)
    os.makedirs(ws2, exist_ok=True)
    os.makedirs(home, exist_ok=True)

    odb = f"{home}/.local/share/opencode/opencode.db"
    os.makedirs(os.path.dirname(odb), exist_ok=True)
    conn = sqlite3.connect(odb)
    conn.execute("CREATE TABLE session (id TEXT PRIMARY KEY, directory TEXT NOT NULL, time_created INTEGER NOT NULL)")

    epoch_floor_sec = 1724920000
    conn.execute("INSERT INTO session VALUES ('u-target-fresh', ?, ?)", (ws1, int((epoch_floor_sec + 100) * 1000)))
    conn.execute("INSERT INTO session VALUES ('u-target-old', ?, ?)", (ws1, int((epoch_floor_sec - 100) * 1000)))
    conn.execute("INSERT INTO session VALUES ('u-foreign', ?, ?)", (ws2, int((epoch_floor_sec + 200) * 1000)))
    conn.commit()
    conn.close()

    adapter = get_adapter('opencode')
    ctx_ws1 = DiscoveryContext(workspace=ws1, agent_name='opencode', home_dir=home, epoch=epoch_floor_sec)

    # verify_artifact assertions
    assert adapter.verify_artifact('u-target-fresh', ctx_ws1) is True
    assert adapter.verify_artifact('u-target-old', ctx_ws1) is False  # Stale epoch rejected
    assert adapter.verify_artifact('u-foreign', ctx_ws1) is False     # Foreign workspace rejected

    # discover assertions
    discovered = adapter.discover(ctx_ws1)
    assert discovered == ['u-target-fresh']


def test_opencode_resolve_db_cli_path(tmp_path, monkeypatch):
    """Verify _resolve_db correctly executes opencode db path and falls back to candidate paths."""
    import stat
    from lib_py.agents.registry import get_adapter
    from lib_py.agents.base import DiscoveryContext

    adapter = get_adapter('opencode')
    home = str(tmp_path / "home")
    ws = str(tmp_path / "ws")
    os.makedirs(home, exist_ok=True)
    os.makedirs(ws, exist_ok=True)

    expected_db = f"{home}/.local/share/opencode/custom_opencode.db"
    os.makedirs(os.path.dirname(expected_db), exist_ok=True)
    with open(expected_db, "w") as f:
        f.write("mock db")

    # Create mock opencode binary that outputs custom DB path
    bin_dir = tmp_path / "bin"
    os.makedirs(bin_dir, exist_ok=True)
    mock_bin = bin_dir / "opencode"
    mock_bin.write_text(f"""#!/bin/sh
if [ "$1" = "db" ] && [ "$2" = "path" ]; then
    echo "{expected_db}"
    exit 0
fi
exit 1
""")
    mock_bin.chmod(mock_bin.stat().st_mode | stat.S_IEXEC)

    monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ.get('PATH', '')}")
    ctx = DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home)
    assert adapter._resolve_db(ctx) == expected_db