diff --git a/Sensor/adr_sensor/parsers/opencode_parser.py b/Sensor/adr_sensor/parsers/opencode_parser.py index c5e838ae..fdfadc22 100644 --- a/Sensor/adr_sensor/parsers/opencode_parser.py +++ b/Sensor/adr_sensor/parsers/opencode_parser.py @@ -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 @@ -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) diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index ffeb2fd2..9d161848 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -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)