Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions Sensor/adr_sensor/parsers/opencode_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -155,12 +155,17 @@ def _find_db_file(base: Path) -> Optional[Path]:
candidates = OpencodeParser._find_db_files(base)
return candidates[0] if candidates else None

@staticmethod
def _read_only_uri(db_path: Path) -> str:
"""Build a read-only SQLite URI, percent-encoding characters such as '#', '?' and '%'."""
return f"{Path(db_path).resolve().as_uri()}?mode=ro"

@staticmethod
def _has_session_schema(db_path: Path) -> bool:
"""Return True if the database has opencode's session tables."""
conn = None
try:
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
conn = sqlite3.connect(OpencodeParser._read_only_uri(db_path), uri=True)
rows = conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
except (sqlite3.Error, OSError):
return False
Expand Down Expand Up @@ -224,7 +229,7 @@ def _parse_sqlite(self, db_path: Path) -> List[AgentEvent]:
conn = None
try:
# Open read-only so we never disturb a live opencode process.
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
conn = sqlite3.connect(OpencodeParser._read_only_uri(db_path), uri=True)
conn.row_factory = sqlite3.Row

sessions = self._get_sqlite_sessions(conn)
Expand Down
26 changes: 26 additions & 0 deletions Sensor/tests/test_parsers.py
Original file line number Diff line number Diff line change
Expand Up @@ -2309,6 +2309,32 @@ def test_session_database_error_is_logged_as_a_warning(self, tmp_path, capsys):
assert "[OPENCODE] Error processing session ses_locked: database is locked" in captured.err
assert "Error processing session" not in captured.out

@pytest.mark.parametrize("dirname", ["data#1", "100%data", "what?"])
def test_reads_database_under_path_with_uri_special_characters(self, tmp_path, dirname):
base = tmp_path / dirname
base.mkdir()
_build_opencode_db(
base / "opencode.db",
sessions=[{"id": "ses_uri"}],
messages=[{"id": "msg_uri", "session_id": "ses_uri", "data": {"role": "user"}}],
parts=[
{
"id": "prt_uri",
"message_id": "msg_uri",
"session_id": "ses_uri",
"data": {"type": "text", "text": "hello from a special path"},
}
],
)

parser = _make_opencode_parser(base)
entries = parser.parse_all()

assert parser.backend == "sqlite"
assert [entry.session_id for entry in entries] == ["opencode_uri"]
# A '#' must not truncate the path and create a stray database file.
assert sorted(path.name for path in tmp_path.iterdir()) == [dirname]

def test_unreadable_database_still_records_database_error(self, tmp_path):
db_path = tmp_path / "opencode.db"
db_path.write_bytes(b"not a sqlite database" * 10)
Expand Down