GodopuandClaude Opus 5.5
f0d1c000f9
tests/_fake_opencode.py implements the two commands MAM calls (debug paths, session delete) as measured on 2.0.32; unit tests never run a real opencode (it would migrate the test database). test_real_opencode_2x_contract runs a real 2.x opencode, when installed, in a home and tmp dir of its own with --standalone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
1089 lines
47 KiB
Python
1089 lines
47 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
|
||
import _fake_opencode # tests/ is on sys.path (conftest)
|
||
|
||
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',
|
||
'codex': 'codex_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|Ask anything|tab agents|ctrl\\+p|Build auto|commands', '/exit', 'opencode-cli', ('session_id',)),
|
||
'codex': ('Codex|Ask Codex to do anything', 'Ctrl+C', 'codex-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,
|
||
'codex': 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', 'codex'):
|
||
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, parent_id=None):
|
||
"""A session row of the opencode 2.x schema (session_v2, Issue #60)."""
|
||
import sqlite3
|
||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||
conn = sqlite3.connect(db_path)
|
||
conn.execute("CREATE TABLE IF NOT EXISTS session_v2 (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT NOT NULL, time_created INTEGER NOT NULL)")
|
||
conn.execute("INSERT OR REPLACE INTO session_v2 (id, parent_id, directory, time_created) VALUES (?, ?, ?, ?)",
|
||
(session_id, parent_id, ws_path, int(float(epoch_sec) * 1000)))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def test_purge_artifacts_composite(tmp_path, monkeypatch):
|
||
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('0c1a5f3e-6b2d-4c8e-9a7f-1d2e3f4a5b6c', claude_ctx)
|
||
os.makedirs(os.path.dirname(c_path), exist_ok=True)
|
||
with open(c_path, 'w') as f:
|
||
f.write('{"sessionId": "0c1a5f3e-6b2d-4c8e-9a7f-1d2e3f4a5b6c"}')
|
||
assert os.path.exists(c_path)
|
||
purged_c = claude_adapter.purge_artifacts('0c1a5f3e-6b2d-4c8e-9a7f-1d2e3f4a5b6c', 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 2.x: `opencode session delete` removes the session (the
|
||
# fake opencode, as the real one does), and the rows a 1.x opencode left
|
||
# behind (message, part) are deleted directly (Issue #60).
|
||
opencode_adapter = get_adapter('opencode')
|
||
fake = _fake_opencode.write(tmp_path / "fakebin" / "opencode")
|
||
monkeypatch.setattr(opencode_adapter, "_bin", lambda ctx: fake)
|
||
opencode_ctx = DiscoveryContext(workspace=ws, agent_name='opencode', home_dir=home)
|
||
o_db = opencode_adapter.artifact_path('uuid-o', opencode_ctx)
|
||
assert o_db == f"{home}/.local/share/opencode/opencode.db"
|
||
_seed_opencode_session(o_db, 'uuid-o', ws, 2000.0)
|
||
conn = sqlite3.connect(o_db)
|
||
conn.execute("CREATE TABLE session_message (id TEXT PRIMARY KEY, session_id TEXT)")
|
||
conn.execute("CREATE TABLE message (id TEXT PRIMARY KEY, session_id TEXT)")
|
||
conn.execute("CREATE TABLE part (id TEXT PRIMARY KEY, session_id TEXT)")
|
||
conn.execute("INSERT INTO session_message VALUES ('sm1', 'uuid-o')")
|
||
conn.execute("INSERT INTO message VALUES ('m1', 'uuid-o')")
|
||
conn.execute("INSERT INTO part VALUES ('p1', 'uuid-o')")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
purged_o = opencode_adapter.purge_artifacts('uuid-o', opencode_ctx)
|
||
assert purged_o == ["opencode session (with its child sessions): uuid-o", "sqlite rows for session: uuid-o"]
|
||
conn = sqlite3.connect(o_db)
|
||
assert conn.execute("SELECT count(*) FROM session_v2 WHERE id='uuid-o'").fetchone()[0] == 0
|
||
assert conn.execute("SELECT count(*) FROM session_message WHERE session_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 --standalone'
|
||
assert opencode.spawn_spec('opencode', '') == 'opencode --auto --standalone'
|
||
assert opencode.resume_spec('opencode', 'u1', materialized=True) == 'opencode --session u1 --auto --standalone'
|
||
assert opencode.resume_spec('opencode', 'u1', materialized=False) == 'opencode --auto --standalone'
|
||
|
||
codex = get_adapter('codex')
|
||
assert codex.spawn_spec('codex', 'u1') == 'codex --dangerously-bypass-approvals-and-sandbox'
|
||
assert codex.spawn_spec('codex', '') == 'codex --dangerously-bypass-approvals-and-sandbox'
|
||
assert codex.resume_spec('codex', 'u1', materialized=True) == 'codex resume --dangerously-bypass-approvals-and-sandbox u1'
|
||
assert codex.resume_spec('codex', 'u1', materialized=False) == 'codex --dangerously-bypass-approvals-and-sandbox'
|
||
|
||
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
|
||
|
||
# Codex auth check (env var, auth.json file, or CLI)
|
||
codex = get_adapter('codex')
|
||
assert codex.auth_ok(run_cmd=lambda cmd: (1, '', '')) is False
|
||
monkeypatch.setenv("OPENAI_API_KEY", "test-key")
|
||
assert codex.auth_ok() is True
|
||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||
codex_auth = tmp_path / ".codex" / "auth.json"
|
||
codex_auth.parent.mkdir(parents=True, exist_ok=True)
|
||
codex_auth.write_text("{}")
|
||
assert codex.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']
|
||
|
||
# 6. Codex
|
||
codex = get_adapter('codex')
|
||
ctx_x = DiscoveryContext(workspace=ws, agent_name='codex', home_dir=home)
|
||
codex_home = f"{home}/.codex"
|
||
os.makedirs(codex_home, exist_ok=True)
|
||
c_db = f"{codex_home}/state_5.sqlite"
|
||
conn = sqlite3.connect(c_db)
|
||
conn.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT)")
|
||
c_rollout = f"{codex_home}/sessions/2026/09/05/rollout-001-u-x1.jsonl"
|
||
os.makedirs(os.path.dirname(c_rollout), exist_ok=True)
|
||
with open(c_rollout, 'w') as f:
|
||
f.write('{"payload": {"cwd": "' + ws + '", "id": "u-x1"}}\n')
|
||
conn.execute("INSERT INTO threads VALUES ('u-x1', ?, ?)", (ws, c_rollout))
|
||
conn.commit()
|
||
conn.close()
|
||
assert codex.discover(ctx_x) == ['u-x1']
|
||
|
||
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'), ('codex', 'Ctrl+C')]:
|
||
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',
|
||
'codex': 'codex-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" ;;
|
||
codex) delegate_agent="codex-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,}'
|
||
# Issue #30: the input box itself is a ┃ block (herdr 0.9.1 capture).
|
||
assert adapter.input_block_prefix == '┃'
|
||
|
||
|
||
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_v2 (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT NOT NULL, time_created INTEGER NOT NULL)")
|
||
|
||
epoch_floor_sec = 1724920000
|
||
ins = "INSERT INTO session_v2 (id, directory, time_created) VALUES (?, ?, ?)"
|
||
conn.execute(ins, ('u-target-fresh', ws1, int((epoch_floor_sec + 100) * 1000)))
|
||
conn.execute(ins, ('u-target-old', ws1, int((epoch_floor_sec - 100) * 1000)))
|
||
conn.execute(ins, ('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):
|
||
"""_resolve_db reads the `db` line of `opencode debug paths` (2.x; the 1.x
|
||
`db path` only prints the help) and falls back to the default path."""
|
||
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"{tmp_path}/elsewhere/opencode/opencode.db"
|
||
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" = "debug" ] && [ "$2" = "paths" ]; then
|
||
echo "home {home}"
|
||
echo "data {tmp_path}/elsewhere/opencode"
|
||
echo "db {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
|
||
|
||
monkeypatch.setattr(adapter, "_bin", lambda ctx: None)
|
||
assert adapter._resolve_db(ctx) == f"{home}/.local/share/opencode/opencode.db"
|
||
|
||
|
||
def _seed_codex_session(home_dir, session_id, ws_path, epoch_sec):
|
||
import sqlite3, json, time
|
||
codex_home = f"{home_dir}/.codex"
|
||
os.makedirs(codex_home, exist_ok=True)
|
||
db_path = f"{codex_home}/state_5.sqlite"
|
||
conn = sqlite3.connect(db_path)
|
||
conn.execute("CREATE TABLE IF NOT EXISTS threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT)")
|
||
sessions_dir = f"{codex_home}/sessions/2026/09/05"
|
||
os.makedirs(sessions_dir, exist_ok=True)
|
||
rollout_file = f"{sessions_dir}/rollout-001-{session_id}.jsonl"
|
||
with open(rollout_file, "w") as rf:
|
||
rf.write(json.dumps({"payload": {"cwd": ws_path, "id": session_id}}) + "\n")
|
||
if epoch_sec:
|
||
os.utime(rollout_file, (epoch_sec, epoch_sec))
|
||
conn.execute("INSERT OR REPLACE INTO threads (id, cwd, rollout_path) VALUES (?, ?, ?)", (session_id, ws_path, rollout_file))
|
||
conn.commit()
|
||
conn.close()
|
||
return rollout_file
|
||
|
||
|
||
def test_codex_purge_artifacts(tmp_path):
|
||
from lib_py.agents.registry import get_adapter
|
||
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)
|
||
|
||
adapter = get_adapter('codex')
|
||
ctx = DiscoveryContext(workspace=ws, agent_name='codex', home_dir=home)
|
||
rollout = _seed_codex_session(home, 'u-del', ws, 2000.0)
|
||
assert os.path.exists(rollout)
|
||
|
||
purged = adapter.purge_artifacts('u-del', ctx)
|
||
assert len(purged) == 2
|
||
assert not os.path.exists(rollout)
|
||
import sqlite3
|
||
conn = sqlite3.connect(f"{home}/.codex/state_5.sqlite")
|
||
assert conn.execute("SELECT count(*) FROM threads WHERE id='u-del'").fetchone()[0] == 0
|
||
conn.close()
|
||
|
||
|
||
def test_codex_verify_artifact_epoch_filtering(tmp_path):
|
||
from lib_py.agents.registry import get_adapter
|
||
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)
|
||
|
||
adapter = get_adapter('codex')
|
||
ctx_old = DiscoveryContext(workspace=ws, agent_name='codex', home_dir=home, epoch=1500)
|
||
_seed_codex_session(home, 'u-old', ws, 1000.0)
|
||
_seed_codex_session(home, '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_codex_reconcile_dual_direction_epoch_guards(tmp_path):
|
||
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('codex')
|
||
_seed_codex_session(home, 'u-hist', ws, 1000.0)
|
||
_seed_codex_session(home, '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='codex', home_dir=home, epoch=epoch, row=s_row)) if verify_session_uuid(ws, 'codex', u, s_row, home_dir=home, mode="discover")]
|
||
assert valid == ['u-fresh1']
|
||
assert len(valid) == 1
|
||
|
||
_seed_codex_session(home, 'u-fresh2', ws, 2100.0)
|
||
|
||
valid2 = [u for u in adapter.discover(DiscoveryContext(workspace=ws, agent_name='codex', home_dir=home, epoch=epoch, row=s_row)) if verify_session_uuid(ws, 'codex', u, s_row, home_dir=home, mode="discover")]
|
||
assert len(valid2) == 2
|
||
|
||
|
||
def test_codex_reconcile_full_block_integration(tmp_path):
|
||
"""Verify codex drift-C block behavior with sibling_claimed exclusion."""
|
||
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('codex')
|
||
_seed_codex_session(home, 'u-claimed', ws, 2000.0)
|
||
_seed_codex_session(home, 'u-target', ws, 2001.0)
|
||
|
||
epoch = 1900
|
||
s_eval = {
|
||
'name': 'codex-target-01',
|
||
'pane': {'cwd': ws},
|
||
'herdr_session_epoch': epoch,
|
||
'_sibling_claimed_uuids': ['u-claimed']
|
||
}
|
||
|
||
candidates = adapter.discover(DiscoveryContext(workspace=ws, agent_name='codex', 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, 'codex', uuid, s_eval, home_dir=home, mode="discover"):
|
||
valid_candidates.append(uuid)
|
||
|
||
assert valid_candidates == ['u-target']
|
||
|
||
|
||
def test_codex_verify_artifact_metadata_fail_closed(tmp_path, monkeypatch):
|
||
"""R4 / T1: Verify verify_artifact fails closed on missing/invalid metadata, id mismatch, or db inconsistency."""
|
||
import json, sqlite3
|
||
from lib_py.agents.registry import get_adapter
|
||
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)
|
||
adapter = get_adapter('codex')
|
||
ctx = DiscoveryContext(workspace=ws, agent_name='codex', home_dir=home)
|
||
|
||
uuid = "12345678-1234-1234-1234-123456789abc"
|
||
rollout_file = tmp_path / "rollout.jsonl"
|
||
rollout_file.write_text("{}\n")
|
||
|
||
# Install finder override using pytest monkeypatch before all cases
|
||
monkeypatch.setattr(adapter, "_find_rollout", lambda u, c: str(rollout_file))
|
||
|
||
# Assert finder resolves the fixture before testing malformed contents
|
||
resolved_path = adapter._find_rollout(uuid, ctx)
|
||
assert resolved_path == str(rollout_file)
|
||
assert os.path.exists(resolved_path)
|
||
|
||
# Parameterized invalid metadata cases
|
||
invalid_payloads = [
|
||
"{}", # Empty dict (R4 finding)
|
||
json.dumps({"payload": {"cwd": ws}}), # Missing payload.id
|
||
json.dumps({"payload": {"id": uuid}}), # Missing payload.cwd
|
||
json.dumps({"payload": {"id": "other-id", "cwd": ws}}), # Mismatched session id
|
||
json.dumps({"payload": {"id": uuid, "cwd": "/wrong/path"}}), # Mismatched workspace
|
||
"not-valid-json", # Non-JSON content
|
||
json.dumps([]), # Non-dict JSON
|
||
json.dumps({"payload": "not-a-dict"}), # Non-dict payload
|
||
]
|
||
|
||
for inv in invalid_payloads:
|
||
rollout_file.write_text(inv + "\n")
|
||
assert adapter.verify_artifact(uuid, ctx) is False, f"Expected False for payload: {inv}"
|
||
|
||
# Database / transcript metadata inconsistency check
|
||
db_path = tmp_path / "state_5.sqlite"
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT)")
|
||
conn.execute("INSERT INTO threads VALUES (?, ?, ?)", (uuid, "/conflicting/db/cwd", str(rollout_file)))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
monkeypatch.setattr(adapter, "_state_db", lambda c: str(db_path))
|
||
# Even if transcript has valid ws, DB cwd conflict must fail closed
|
||
rollout_file.write_text(json.dumps({"payload": {"id": uuid, "cwd": ws}}) + "\n")
|
||
assert adapter.verify_artifact(uuid, ctx) is False
|
||
|
||
# Valid control case (matching DB cwd and transcript cwd)
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.execute("UPDATE threads SET cwd=? WHERE id=?", (ws, uuid))
|
||
conn.commit()
|
||
conn.close()
|
||
assert adapter.verify_artifact(uuid, ctx) is True
|
||
|
||
|
||
def test_codex_discover_empty_stale_db_fallback(tmp_path):
|
||
"""R2: Verify discover falls back to JSONL transcripts when DB is empty or has stale rows."""
|
||
import json, sqlite3, time
|
||
from lib_py.agents.registry import get_adapter
|
||
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)
|
||
codex_home = tmp_path / "home" / ".codex"
|
||
os.makedirs(codex_home, exist_ok=True)
|
||
|
||
adapter = get_adapter('codex')
|
||
ctx = DiscoveryContext(workspace=ws, agent_name='codex', home_dir=home)
|
||
|
||
# Create a valid date-nested rollout on disk
|
||
uuid_valid = "11111111-1111-1111-1111-111111111111"
|
||
sessions_dir = codex_home / "sessions" / "2026" / "09" / "05"
|
||
sessions_dir.mkdir(parents=True, exist_ok=True)
|
||
valid_file = sessions_dir / f"rollout-2026-09-05T00-00-00-{uuid_valid}.jsonl"
|
||
valid_file.write_text(json.dumps({"payload": {"id": uuid_valid, "cwd": ws}}) + "\n")
|
||
|
||
# Empty DB: threads table exists with 0 rows
|
||
db_path = codex_home / "state_5.sqlite"
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT, updated_at INTEGER)")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
# Should fall through to JSONL file and discover uuid_valid
|
||
found = adapter.discover(ctx)
|
||
assert uuid_valid in found
|
||
|
||
# Stale DB: insert row pointing to nonexistent file
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.execute("INSERT INTO threads (id, cwd, rollout_path, updated_at) VALUES (?, ?, ?, ?)",
|
||
("stale-uuid", ws, str(codex_home / "nonexistent.jsonl"), int(time.time())))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
found_stale = adapter.discover(ctx)
|
||
assert uuid_valid in found_stale
|
||
|
||
|
||
def test_codex_discover_workspace_scoping_over_50_foreign_rows(tmp_path):
|
||
"""R3: Verify global LIMIT does not hide target-workspace threads behind >50 foreign threads."""
|
||
import json, sqlite3, time
|
||
from pathlib import Path
|
||
from lib_py.agents.registry import get_adapter
|
||
from lib_py.agents.base import DiscoveryContext
|
||
ws_target = str(tmp_path / "ws_target")
|
||
ws_foreign = str(tmp_path / "ws_foreign")
|
||
home = str(tmp_path / "home")
|
||
os.makedirs(ws_target, exist_ok=True)
|
||
os.makedirs(ws_foreign, exist_ok=True)
|
||
os.makedirs(home, exist_ok=True)
|
||
|
||
adapter = get_adapter('codex')
|
||
ctx_target = DiscoveryContext(workspace=ws_target, agent_name='codex', home_dir=home)
|
||
|
||
codex_home = Path(home) / ".codex"
|
||
sessions_dir = codex_home / "sessions" / "2026" / "09" / "05"
|
||
sessions_dir.mkdir(parents=True, exist_ok=True)
|
||
db_path = codex_home / "state_5.sqlite"
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.execute("CREATE TABLE IF NOT EXISTS threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT, updated_at INTEGER)")
|
||
|
||
now = int(time.time())
|
||
# Seed 1 valid target thread
|
||
target_uuid = "target-1234-1234-1234-123456789abc"
|
||
target_file = sessions_dir / f"rollout-001-{target_uuid}.jsonl"
|
||
target_file.write_text(json.dumps({"payload": {"id": target_uuid, "cwd": ws_target}}) + "\n")
|
||
conn.execute("INSERT INTO threads (id, cwd, rollout_path, updated_at) VALUES (?, ?, ?, ?)",
|
||
(target_uuid, ws_target, str(target_file), now - 100))
|
||
|
||
# Seed 60 newer foreign threads
|
||
for i in range(60):
|
||
f_uuid = f"foreign-{i:04d}-1234-1234-123456789abc"
|
||
f_file = sessions_dir / f"rollout-001-{f_uuid}.jsonl"
|
||
f_file.write_text(json.dumps({"payload": {"id": f_uuid, "cwd": ws_foreign}}) + "\n")
|
||
conn.execute("INSERT INTO threads (id, cwd, rollout_path, updated_at) VALUES (?, ?, ?, ?)",
|
||
(f_uuid, ws_foreign, str(f_file), now + i))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
candidates = adapter.discover(ctx_target)
|
||
assert target_uuid in candidates
|
||
|
||
|
||
def test_codex_discover_symlink_canonical_alias(tmp_path):
|
||
"""R3: Verify discover resolves symlink workspace aliases correctly."""
|
||
from lib_py.agents.registry import get_adapter
|
||
from lib_py.agents.base import DiscoveryContext
|
||
real_ws = tmp_path / "real_ws"
|
||
real_ws.mkdir()
|
||
symlink_ws = tmp_path / "symlink_ws"
|
||
symlink_ws.symlink_to(real_ws, target_is_directory=True)
|
||
home = str(tmp_path / "home")
|
||
os.makedirs(home, exist_ok=True)
|
||
|
||
adapter = get_adapter('codex')
|
||
ctx = DiscoveryContext(workspace=str(symlink_ws), agent_name='codex', home_dir=home)
|
||
|
||
uuid = "sym-1234-1234-1234-123456789abc"
|
||
_seed_codex_session(home, uuid, str(real_ws), 2000.0)
|
||
|
||
candidates = adapter.discover(ctx)
|
||
assert uuid in candidates
|