GodopuandClaude Opus 5.5
6a37c0019b
- tests: macOS SQLite keeps -wal/-shm after the last close (persistent WAL); the `clean` state removes them after the checkpointing close - the 2.x placeholder pattern requires the quoted example, so a typed `Ask anything…` is read as a draft - auth_ok takes ~/.opencode/bin/opencode only as an executable file - status.sh and reconcile.sh (drift C) open opencode.db with _ro_connect Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
383 lines
17 KiB
Python
383 lines
17 KiB
Python
"""Issue #60, Linux check: the defects the opencode 2.x check on Linux found.
|
|
|
|
1. The 2.x home screen's placeholder (`Ask anything… "<example>"`, cut
|
|
without its closing quote in a narrow box) is not typed input.
|
|
5. The adapter's queries neither create nor remove anything beside
|
|
opencode's database (a read-write open checkpoints the WAL and deletes
|
|
the sidecars on close; a plain mode=ro open creates missing ones).
|
|
6. `debug paths` (which writes a line to opencode's log) runs once per
|
|
(binary, HOME, XDG_DATA_HOME) in a process.
|
|
2. A graceful stop whose `/exit` worked says so: the agent exits instead of
|
|
turning `working`, so the delivery is unconfirmed, and no kill step ran.
|
|
3. opencode's auth check is the binary alone (2.x needs no login), and create
|
|
takes the adapter's verdict instead of running `opencode --version`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import json
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
PANES = REPO_ROOT / "tests" / "fixtures" / "panes"
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
|
|
from lib_py import stall_watch # noqa: E402
|
|
from lib_py.agents.base import DiscoveryContext # noqa: E402
|
|
from lib_py.agents.input_region import extract_input_region # noqa: E402
|
|
from lib_py.agents.registry import get_adapter # noqa: E402
|
|
from lib_py.agents.adapters import opencode as opencode_mod # noqa: E402
|
|
import _fake_opencode # noqa: E402
|
|
|
|
# 140 columns (herdr capture) and 60 columns, where the example loses its
|
|
# closing quote.
|
|
HOME_SCREENS = ("opencode_idle_2x_home.txt", "opencode_idle_2x_home_narrow.txt")
|
|
PLACEHOLDER_LINE = re.compile(r'^(\s*┃\s*)Ask anything… ".*$', re.M)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _no_cached_db_paths(monkeypatch):
|
|
monkeypatch.setattr(opencode_mod, "_DB_PATHS", {}, raising=False)
|
|
|
|
|
|
def _pane(name: str) -> str:
|
|
return "".join(l for l in (PANES / name).read_text(encoding="utf-8").splitlines(True) if not l.startswith("# "))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. the 2.x home screen's placeholder
|
|
# ---------------------------------------------------------------------------
|
|
@pytest.mark.parametrize("name", HOME_SCREENS)
|
|
def test_the_2x_home_screen_reads_as_an_empty_box(name):
|
|
assert extract_input_region(_pane(name), "opencode") == ""
|
|
|
|
|
|
def test_the_narrow_capture_has_a_cut_example():
|
|
# The case the pattern must keep: no closing quote on screen.
|
|
(line,) = [l for l in _pane(HOME_SCREENS[1]).splitlines() if "Ask anything" in l]
|
|
assert '"' in line and not line.rstrip().endswith('"')
|
|
|
|
|
|
def test_the_narrow_screen_stops_at_its_status_line_and_reads_ready():
|
|
text = _pane(HOME_SCREENS[1])
|
|
# Under ~60 columns the status line drops `auto`: `Build · <model>`.
|
|
assert re.search(r"^\s*┃\s*Build · ", text, re.M) and "Build auto" not in text
|
|
assert re.search(get_adapter("opencode").ready_tokens, text)
|
|
|
|
|
|
@pytest.mark.parametrize("name", HOME_SCREENS)
|
|
@pytest.mark.parametrize("typed", [
|
|
"please continue",
|
|
'Ask anything… "x" and fix it',
|
|
"Ask anything about the build",
|
|
'Ask anything… "a"b',
|
|
"Ask anything…",
|
|
"Ask anything...",
|
|
])
|
|
def test_typed_text_is_read_as_typed(name, typed):
|
|
text, n = PLACEHOLDER_LINE.subn(lambda m: m.group(1) + typed, _pane(name), count=1)
|
|
assert n == 1
|
|
assert extract_input_region(text, "opencode") == typed
|
|
|
|
|
|
def test_the_1x_capture_still_reads_empty():
|
|
assert extract_input_region(_pane("opencode_idle.txt"), "opencode") == ""
|
|
|
|
|
|
@pytest.mark.parametrize("name", HOME_SCREENS)
|
|
def test_stall_watch_sees_the_2x_home_box_empty(name, monkeypatch):
|
|
src = stall_watch.HerdrSource("t", "opencode")
|
|
monkeypatch.setattr(src, "_herdr", lambda *args: _pane(name))
|
|
assert src.input_empty() is True
|
|
|
|
|
|
def test_other_adapters_have_no_placeholder_pattern():
|
|
for agent in ("claude", "agy", "grok", "codex", "hermes"):
|
|
assert get_adapter(agent).input_placeholder_pattern is None, agent
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. the queries leave the files beside the database alone
|
|
# ---------------------------------------------------------------------------
|
|
V2_DDL = ("CREATE TABLE session_v2 (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT NOT NULL, time_created INTEGER NOT NULL)",
|
|
"CREATE TABLE session_message (id TEXT PRIMARY KEY, session_id TEXT)")
|
|
|
|
QUERIES = {
|
|
"verify_artifact": (lambda a, ctx, db: a.verify_artifact("ses_main", ctx), True),
|
|
"discover": (lambda a, ctx, db: a.discover(ctx), ["ses_main"]),
|
|
"_has_session": (lambda a, ctx, db: a._has_session(db, "ses_main"), True),
|
|
"_descendants": (lambda a, ctx, db: a._descendants(db, "ses_main"), ["ses_child"]),
|
|
"residual_artifacts": (lambda a, ctx, db: any("session_v2.id: 1 rows" in r
|
|
for r in a.residual_artifacts("ses_main", ctx)), True),
|
|
}
|
|
|
|
|
|
def _wal_writer(db: Path, ws: Path) -> sqlite3.Connection:
|
|
"""An open WAL connection holding the schema and two sessions: until it
|
|
closes, the rows live in the -wal only."""
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute("PRAGMA journal_mode=wal")
|
|
for stmt in V2_DDL:
|
|
conn.execute(stmt)
|
|
conn.execute("INSERT INTO session_v2 VALUES ('ses_main', NULL, ?, 2000000)", (str(ws),))
|
|
conn.execute("INSERT INTO session_v2 VALUES ('ses_child', 'ses_main', ?, 3000000)", (str(ws),))
|
|
conn.execute("INSERT INTO session_message VALUES ('msg-1', 'ses_main')")
|
|
conn.commit()
|
|
return conn
|
|
|
|
|
|
@pytest.fixture
|
|
def wal_home(tmp_path, monkeypatch):
|
|
"""(adapter, ctx, db, ws, make(state)) - make builds the database in one
|
|
of four states: `clean` (closed: no sidecars), `leftover` (both sidecars
|
|
left behind by a writer that is gone, the rows only in the -wal),
|
|
`walonly` (the same without its -shm: the committed rows still count),
|
|
`writer` (a live writer whose rows are only in the -wal)."""
|
|
home, ws = tmp_path / "home", tmp_path / "ws"
|
|
ws.mkdir()
|
|
db = home / ".local" / "share" / "opencode" / "opencode.db"
|
|
db.parent.mkdir(parents=True)
|
|
adapter = get_adapter("opencode")
|
|
monkeypatch.setattr(adapter, "_bin", lambda ctx: None) # the default path, no subprocess
|
|
ctx = DiscoveryContext(workspace=str(ws), agent_name="opencode", home_dir=str(home))
|
|
opened = []
|
|
|
|
def make(state):
|
|
if state == "clean":
|
|
_wal_writer(db, ws).close()
|
|
# macOS SQLite keeps both sidecars after the last close (persistent
|
|
# WAL); the close checkpointed them, so the Linux state is the same rows.
|
|
for suffix in ("-wal", "-shm"):
|
|
if os.path.exists(str(db) + suffix):
|
|
os.remove(str(db) + suffix)
|
|
elif state in ("leftover", "walonly"):
|
|
src = tmp_path / "src" / "opencode.db"
|
|
src.parent.mkdir()
|
|
conn = _wal_writer(src, ws)
|
|
for suffix in ("", "-wal", "-shm")[:3 if state == "leftover" else 2]:
|
|
shutil.copyfile(str(src) + suffix, str(db) + suffix)
|
|
conn.close()
|
|
else:
|
|
opened.append(_wal_writer(db, ws))
|
|
return sorted(os.listdir(db.parent))
|
|
|
|
yield adapter, ctx, str(db), ws, make
|
|
for conn in opened:
|
|
conn.close()
|
|
|
|
|
|
@pytest.mark.parametrize("query", sorted(QUERIES))
|
|
@pytest.mark.parametrize("state,listing", [
|
|
("clean", ["opencode.db"]),
|
|
("leftover", ["opencode.db", "opencode.db-shm", "opencode.db-wal"]),
|
|
("walonly", ["opencode.db", "opencode.db-wal"]),
|
|
("writer", ["opencode.db", "opencode.db-shm", "opencode.db-wal"]),
|
|
])
|
|
def test_a_query_reads_right_and_leaves_the_sidecars_alone(wal_home, state, listing, query):
|
|
adapter, ctx, db, ws, make = wal_home
|
|
assert make(state) == listing
|
|
run, expected = QUERIES[query]
|
|
assert run(adapter, ctx, db) == expected
|
|
assert sorted(os.listdir(os.path.dirname(db))) == listing
|
|
|
|
|
|
@pytest.mark.parametrize("script,count", [
|
|
("multi-agent-mux-status/scripts/status.sh", 2),
|
|
("multi-agent-mux-monitor/scripts/reconcile.sh", 1),
|
|
])
|
|
def test_status_and_reconcile_open_the_database_read_only(script, count):
|
|
text = (SKILLS / script).read_text()
|
|
assert text.count("_ro_connect(odb)") == count
|
|
assert "sqlite3.connect(odb)" not in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. `debug paths` once per (binary, HOME, XDG_DATA_HOME)
|
|
# ---------------------------------------------------------------------------
|
|
def test_debug_paths_runs_once_per_home(tmp_path, monkeypatch):
|
|
log = tmp_path / "calls.log"
|
|
fake = _fake_opencode.write(tmp_path / "fakebin" / "opencode-real")
|
|
wrapper = tmp_path / "fakebin" / "opencode"
|
|
wrapper.write_text(f'#!/bin/sh\necho "$*" >> "{log}"\nexec "{fake}" "$@"\n')
|
|
wrapper.chmod(0o755)
|
|
adapter = get_adapter("opencode")
|
|
monkeypatch.setattr(adapter, "_bin", lambda ctx: str(wrapper))
|
|
homes = [tmp_path / "home1", tmp_path / "home2"]
|
|
ctxs = [DiscoveryContext(workspace=str(tmp_path), agent_name="opencode", home_dir=str(h)) for h in homes]
|
|
|
|
def calls():
|
|
return log.read_text().splitlines() if log.exists() else []
|
|
|
|
want = [str(h / ".local" / "share" / "opencode" / "opencode.db") for h in homes]
|
|
assert [adapter._resolve_db(ctxs[0]) for _ in range(2)] == [want[0]] * 2
|
|
assert calls() == ["debug paths"]
|
|
assert adapter._resolve_db(ctxs[1]) == want[1]
|
|
assert adapter._resolve_db(ctxs[0]) == want[0]
|
|
assert calls() == ["debug paths"] * 2
|
|
|
|
|
|
def test_a_failed_debug_paths_is_not_remembered(tmp_path, monkeypatch):
|
|
log = tmp_path / "calls.log"
|
|
failing = tmp_path / "fakebin" / "opencode"
|
|
failing.parent.mkdir()
|
|
failing.write_text(f'#!/bin/sh\necho "$*" >> "{log}"\nexit 3\n')
|
|
failing.chmod(0o755)
|
|
adapter = get_adapter("opencode")
|
|
monkeypatch.setattr(adapter, "_bin", lambda ctx: str(failing))
|
|
ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="opencode", home_dir=str(tmp_path / "home"))
|
|
default = str(tmp_path / "home" / ".local" / "share" / "opencode" / "opencode.db")
|
|
assert [adapter._resolve_db(ctx) for _ in range(2)] == [default] * 2
|
|
assert log.read_text().splitlines() == ["debug paths"] * 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. the stop's words match what happened
|
|
# ---------------------------------------------------------------------------
|
|
SESSION = "i60-stop-opencode"
|
|
|
|
# A herdr in front of the conftest mock that, like a real opencode 2.x pane
|
|
# (Linux check 3-3), ends the agent on `/exit`: the prompt's wait for
|
|
# `working` then fails because the agent is gone.
|
|
EXITING_HERDR = """#!/usr/bin/env python3
|
|
import json, os, sys
|
|
args = sys.argv[1:]
|
|
if "prompt" in args and "/exit" in args:
|
|
with open(@STATE@) as f:
|
|
state = json.load(f)
|
|
state["agents"].pop(@NAME@, None)
|
|
state.setdefault("calls", []).append("herdr " + " ".join(args))
|
|
with open(@STATE@, "w") as f:
|
|
json.dump(state, f)
|
|
sys.stderr.write(json.dumps({"error": {"code": "agent_exited", "message": "agent exited"}}) + "\\n")
|
|
sys.exit(1)
|
|
os.execv(@MOCK@, [@MOCK@] + args)
|
|
"""
|
|
|
|
|
|
@pytest.mark.parametrize("exits", [True, False])
|
|
def test_the_graceful_stop_reports_what_happened(mam_sandbox, mock_herdr, mock_agents, monkeypatch, exits):
|
|
state = json.loads(Path(mock_herdr).read_text())
|
|
state["agents"][SESSION] = {"status": "running", "agent": "opencode", "agent_status": "idle",
|
|
"cwd": str(mam_sandbox), "pid": 12345, "pane_id": "w1:p1",
|
|
"command": "opencode", "buffer": "Ask anything"}
|
|
Path(mock_herdr).write_text(json.dumps(state, indent=2))
|
|
lib = mam_sandbox / ".agents" / "skills" / "lib.sh"
|
|
row = ("d['herdr_sessions'] = [{'name': '%s', 'agent': 'opencode', 'status': 'running', "
|
|
"'pane': {'cwd': '%s', 'pid': 12345, 'cmd': 'opencode'}}]" % (SESSION, mam_sandbox))
|
|
res = subprocess.run(["bash", "-c", f"source {lib} && atomic_dump_yaml {mam_sandbox / '.mam' / 'agent-sessions.yaml'}"],
|
|
input=row, capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
if exits:
|
|
wrap = mam_sandbox / "exitbin" / "herdr"
|
|
wrap.parent.mkdir()
|
|
wrap.write_text(EXITING_HERDR.replace("@STATE@", repr(str(mock_herdr))).replace("@NAME@", repr(SESSION))
|
|
.replace("@MOCK@", repr(shutil.which("herdr"))))
|
|
wrap.chmod(0o755)
|
|
monkeypatch.setenv("PATH", f"{wrap.parent}:{os.environ['PATH']}")
|
|
stop = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
|
|
res = subprocess.run(["bash", str(stop), "--session", SESSION, "--agent", "opencode"],
|
|
capture_output=True, text=True, timeout=180)
|
|
out = res.stdout + res.stderr
|
|
assert res.returncode == 0, out
|
|
calls = json.loads(Path(mock_herdr).read_text()).get("calls", [])
|
|
assert any("prompt" in c and "/exit" in c for c in calls), calls
|
|
assert "falling back to kill chain" not in out
|
|
if exits:
|
|
assert "unconfirmed" in out and "graceful: exited cleanly" in out, out
|
|
assert "kill-session" not in out and not any("kill-session" in c for c in calls), out
|
|
else:
|
|
# The kill step runs, so it is named.
|
|
assert "still alive → kill-session" in out, out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. opencode's auth check is the binary
|
|
# ---------------------------------------------------------------------------
|
|
def _executable(path: Path) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("#!/bin/sh\n")
|
|
path.chmod(0o755)
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def bare_home(tmp_path, monkeypatch):
|
|
"""An empty HOME_DIR and a PATH with no opencode."""
|
|
monkeypatch.setenv("HOME_DIR", str(tmp_path / "home"))
|
|
monkeypatch.setenv("PATH", str(tmp_path / "empty-bin"))
|
|
for var in ("ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY", "GROQ_API_KEY"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
return tmp_path
|
|
|
|
|
|
def test_keys_and_auth_json_without_the_binary_are_not_enough(bare_home, monkeypatch):
|
|
opencode = get_adapter("opencode")
|
|
assert opencode.auth_ok() is False
|
|
monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key")
|
|
assert opencode.auth_ok() is False
|
|
auth = bare_home / "home" / ".local" / "share" / "opencode" / "auth.json"
|
|
auth.parent.mkdir(parents=True)
|
|
auth.write_text("{}")
|
|
assert opencode.auth_ok() is False
|
|
|
|
|
|
def test_a_home_install_that_cannot_run_is_not_enough(bare_home):
|
|
local = bare_home / "home" / ".opencode" / "bin" / "opencode"
|
|
_executable(local).chmod(0o644)
|
|
assert get_adapter("opencode").auth_ok() is False
|
|
local.unlink()
|
|
local.mkdir()
|
|
local.chmod(0o755)
|
|
assert get_adapter("opencode").auth_ok() is False
|
|
|
|
|
|
@pytest.mark.parametrize("where", ["path", "home"])
|
|
def test_the_binary_alone_is_enough(bare_home, monkeypatch, where):
|
|
if where == "path":
|
|
_executable(bare_home / "bin" / "opencode")
|
|
monkeypatch.setenv("PATH", str(bare_home / "bin"))
|
|
else:
|
|
_executable(bare_home / "home" / ".opencode" / "bin" / "opencode")
|
|
assert get_adapter("opencode").auth_ok() is True
|
|
|
|
|
|
def test_the_auth_ok_command_exits_with_the_verdict(bare_home):
|
|
def rc(*args):
|
|
env = dict(os.environ, PYTHONPATH=str(SKILLS))
|
|
return subprocess.run([sys.executable, "-m", "lib_py.agents", "auth-ok", *args],
|
|
capture_output=True, text=True, env=env).returncode
|
|
|
|
assert rc("opencode") == 1
|
|
_executable(bare_home / "home" / ".opencode" / "bin" / "opencode")
|
|
assert rc("opencode") == 0
|
|
assert rc("no-such-agent") == 1
|
|
|
|
|
|
def test_create_decides_opencode_auth_without_running_it(mam_sandbox, mock_herdr, mock_agents, monkeypatch):
|
|
# Every opencode call create makes is logged; --version was the old check.
|
|
log = mam_sandbox / "opencode-calls.log"
|
|
wrap = mam_sandbox / "logbin" / "opencode"
|
|
wrap.parent.mkdir()
|
|
wrap.write_text(f'#!/bin/sh\necho "$*" >> "{log}"\nexec "{shutil.which("opencode")}" "$@"\n')
|
|
wrap.chmod(0o755)
|
|
monkeypatch.setenv("PATH", f"{wrap.parent}:{os.environ['PATH']}")
|
|
for var in ("ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY", "GROQ_API_KEY"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
create = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
res = subprocess.run(["bash", str(create), "--workspace", str(mam_sandbox), "--agent", "opencode",
|
|
"--role", "creator", "--no-onboard"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox), timeout=300)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
calls = log.read_text().splitlines() if log.exists() else []
|
|
assert "--version" not in calls, calls
|