multi-agent-mux/tests/test_issue60_linux_fixes.py

"""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).
"""

from __future__ import annotations

import os
import re
import shutil
import sqlite3
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

# 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)


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',
])
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