843 lines
35 KiB
Python
843 lines
35 KiB
Python
# 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
|
||
|
||
|
||
|
||
|
||
|