"""Issue #60, Linux check: the defects the opencode 2.x check on Linux found. 1. The 2.x home screen's placeholder (`Ask anything… ""`, 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. """ 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 · `. 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', ]) 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 three states: `clean` (closed: no sidecars), `leftover` (both sidecars left behind by a writer that is gone, the rows only in the -wal), `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() elif state == "leftover": src = tmp_path / "src" / "opencode.db" src.parent.mkdir() conn = _wal_writer(src, ws) for suffix in ("", "-wal", "-shm"): 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"]), ("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 # --------------------------------------------------------------------------- # 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