tmpl/multi-agent-mux
1
0 Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

test(opencode): the 2.x schema and CLI, a fake opencode, and an opt-in real-binary contract (Issue #60)

Browse Source 
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
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-18 14:58:29 +09:00
1 parent a311c23303
commit f0d1c000f9
5 files changed
+353 -147
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_a4_adapter_contract.py
+41 -121
View File
@@ -3,6 +3,7 @@
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
@@ -132,73 +133,19 @@ def shlex_quote(s):
import shlex
return shlex.quote(s)
def _seed_opencode_session(db_path, session_id, ws_path, epoch_sec):
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)
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.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):
def test_purge_artifacts_composite(tmp_path, monkeypatch):
import sqlite3
from lib_py.agents.base import DiscoveryContext
ws = str(tmp_path / "ws")
@@ -272,60 +219,31 @@ def test_purge_artifacts_composite(tmp_path):
assert not os.path.exists(g_path)
assert not os.path.exists(os.path.dirname(g_path))
# 5. OpenCode (SQLite tables session, message, part)
# 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)
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.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 len(purged_o) == 1
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 WHERE id='uuid-o'").fetchone()[0] == 0
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()
@@ -353,10 +271,10 @@ def test_adapter_spawn_and_resume_specs():
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'
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'
@@ -821,12 +739,13 @@ def test_opencode_real_schema_directory_and_time_created_ms(tmp_path):
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)")
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
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)))
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()
@@ -844,7 +763,8 @@ def test_opencode_real_schema_directory_and_time_created_ms(tmp_path):
def test_opencode_resolve_db_cli_path(tmp_path, monkeypatch):
"""Verify _resolve_db correctly executes opencode db path and falls back to candidate paths."""
"""_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
@@ -855,18 +775,15 @@ def test_opencode_resolve_db_cli_path(tmp_path, monkeypatch):
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
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" = "db" ] && [ "$2" = "path" ]; then
echo "{expected_db}"
if [ "$1" = "debug" ] && [ "$2" = "paths" ]; then
echo "home {home}"
echo "data {tmp_path}/elsewhere/opencode"
echo "db {expected_db}"
exit 0
fi
exit 1
@@ -877,6 +794,9 @@ exit 1
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