fix(sqlite): preserve filenames when applying URI options

Quote the decoded path before building file: URIs in session and memory services. Otherwise # or ? truncates the filename and drops mode=ro, while literal percent escapes select a different database.

Extend public-service regressions across reserved characters, relative and absolute paths, and no-query/read-only/read-write modes. The original head fails 36 cases with 54 passing controls; the fix passes all 90. Affected session, registry, factory and SQLite memory suites: 1039 passed, 1 xfailed on Linux/Python 3.12.14. Changed-file pre-commit and targeted mypy pass. Full tox and Windows were not run.

Author: feiiiiii5

src/google/adk/memory/_sqlite_memory_service.py

@@ -28,6 +28,7 @@
 from typing import Literal
 from typing import TYPE_CHECKING
 import unicodedata
+from urllib.parse import quote
 from urllib.parse import unquote
 from urllib.parse import urlparse
 
@@ -140,7 +141,9 @@ def _parse_db_path(db_path: str | Path) -> tuple[str, str, bool]:
 
   if parsed.query:
     # sqlite3 only treats the filename as a URI when it starts with `file:`.
-    return normalized_path, f"file:{normalized_path}?{parsed.query}", True
+    # Escape filename delimiters and literal percent escapes before URI parsing.
+    connect_uri = f"file:{quote(normalized_path)}?{parsed.query}"
+    return normalized_path, connect_uri, True
 
   return normalized_path, normalized_path, False
 

src/google/adk/sessions/sqlite_session_service.py

@@ -24,6 +24,7 @@
 import sqlite3
 from typing import Any
 from typing import cast
+from urllib.parse import quote
 from urllib.parse import unquote
 from urllib.parse import urlparse
 
@@ -177,7 +178,9 @@ def _parse_db_path(db_path: str) -> tuple[str, str, bool]:
 
   if parsed.query:
     # sqlite3 only treats the filename as a URI when it starts with `file:`.
-    return normalized_path, f"file:{normalized_path}?{parsed.query}", True
+    # Escape filename delimiters and literal percent escapes before URI parsing.
+    connect_uri = f"file:{quote(normalized_path)}?{parsed.query}"
+    return normalized_path, connect_uri, True
 
   return normalized_path, normalized_path, False
 

tests/unittests/cli/test_service_registry.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
 from pathlib import Path
 import sqlite3
 from types import SimpleNamespace
@@ -176,25 +177,51 @@ async def test_sqlite_session_uri_preserves_read_only_mode(
     await session_service.close()
 
 
-async def test_sqlite_session_uri_decodes_database_path(
+@pytest.mark.parametrize("relative", [False, True])
+@pytest.mark.parametrize("mode", [None, "ro", "rw"])
+@pytest.mark.parametrize(
+    "filename",
+    [
+        "sessions #1.db",
+        pytest.param(
+            "sessions ?1.db",
+            marks=pytest.mark.skipif(
+                os.name == "nt",
+                reason="Question marks are not valid Windows filenames",
+            ),
+        ),
+        "sessions %23.db",
+        "sessions +&= ü.db",
+        "sessions.db",
+    ],
+)
+async def test_sqlite_session_uri_preserves_path_and_mode(
     registry: service_registry.ServiceRegistry,
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
+    relative: bool,
+    mode: str | None,
+    filename: str,
 ) -> None:
-  """An encoded SQLite URI opens the existing database at its decoded path."""
+  """Encoded filenames select the same database and retain URI write permissions."""
+  monkeypatch.chdir(tmp_path)
   monkeypatch.setattr(
       "google.adk.sessions.sqlite_session_service.SqliteSessionService",
       SqliteSessionService,
   )
-  db_path = tmp_path / "sessions #1.db"
+  db_path = tmp_path / filename
   writable_service = SqliteSessionService(str(db_path))
-  session = await writable_service.create_session(
-      app_name="app", user_id="user", state={"count": 1}
-  )
-  await writable_service.close()
+  try:
+    session = await writable_service.create_session(
+        app_name="app", user_id="user", state={"count": 1}
+    )
+  finally:
+    await writable_service.close()
 
+  uri_path = filename if relative else db_path.as_posix()
+  query = f"?mode={mode}&cache=shared" if mode else ""
   session_service = registry.create_session_service(
-      f"sqlite:///{quote(db_path.as_posix())}"
+      f"sqlite:///{quote(uri_path)}{query}"
   )
   assert isinstance(session_service, SqliteSessionService)
   try:
@@ -203,9 +230,25 @@ async def test_sqlite_session_uri_decodes_database_path(
     )
     assert stored_session is not None
     assert stored_session.state == {"count": 1}
+    if mode == "ro":
+      with pytest.raises(sqlite3.OperationalError, match="readonly"):
+        await session_service.create_session(app_name="app", user_id="user")
+    else:
+      await session_service.create_session(
+          app_name="app", user_id="user", session_id="new-session"
+      )
   finally:
     await session_service.close()
 
+  reopened_service = SqliteSessionService(str(db_path))
+  try:
+    added_session = await reopened_service.get_session(
+        app_name="app", user_id="user", session_id="new-session"
+    )
+    assert (added_session is None) == (mode == "ro")
+  finally:
+    await reopened_service.close()
+
 
 def test_create_session_service_sqlite_ignores_unsupported_kwargs(
     registry, mock_services, caplog

tests/unittests/memory/test_sqlite_memory_service.py

@@ -17,9 +17,11 @@
 import asyncio
 from contextlib import closing
 from datetime import datetime
+import os
 from pathlib import Path
 import sqlite3
 from typing import Literal
+from urllib.parse import quote
 
 import aiosqlite
 from google.adk.events.event import Event
@@ -158,36 +160,77 @@ async def test_in_memory_uri_variants(uri):
     await service.close()
 
 
-@pytest.mark.asyncio
+@pytest.mark.parametrize("scheme", ["sqlite", "sqlite+aiosqlite"])
+@pytest.mark.parametrize("relative", [False, True])
+@pytest.mark.parametrize("mode", [None, "ro", "rw"])
 @pytest.mark.parametrize(
-    "ro_uri_fn",
+    "filename",
     [
-        lambda p: f"sqlite:///{p / 'ro_abs.db'}?mode=ro",
-        lambda p: f"sqlite+aiosqlite:///{p / 'ro_abs.db'}?mode=ro",
+        "memory #1.db",
+        pytest.param(
+            "memory ?1.db",
+            marks=pytest.mark.skipif(
+                os.name == "nt",
+                reason="Question marks are not valid Windows filenames",
+            ),
+        ),
+        "memory %23.db",
+        "memory +&= ü.db",
+        "memory.db",
     ],
 )
-async def test_readonly_uri_mode_restricts_writes(tmp_path, ro_uri_fn):
-  """Read-only URI query parameter enables searching existing rows and rejects new writes."""
-  db_path = tmp_path / "ro_abs.db"
+async def test_sqlite_uri_preserves_path_and_mode(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    scheme: str,
+    relative: bool,
+    mode: str | None,
+    filename: str,
+) -> None:
+  """Encoded filenames select the same memories and retain URI write permissions."""
+  monkeypatch.chdir(tmp_path)
+  db_path = tmp_path / filename
   rw_service = SqliteMemoryService(db_path=db_path, fts="off")
-  event = _make_event("user", "readonly test content", 1.0)
-  await rw_service.add_events_to_memory(
-      app_name="app", user_id="user", events=[event], session_id="s1"
-  )
+  event = _make_event("user", "saved content", 1.0, event_id="saved")
+  try:
+    await rw_service.add_events_to_memory(
+        app_name="app", user_id="user", events=[event], session_id="s1"
+    )
+  finally:
+    await rw_service.close()
 
-  ro_uri = ro_uri_fn(tmp_path)
-  ro_service = SqliteMemoryService(ro_uri, fts="off")
-  resp = await ro_service.search_memory(
-      app_name="app", user_id="user", query="readonly"
+  uri_path = filename if relative else db_path.as_posix()
+  query = f"?mode={mode}&cache=shared" if mode else ""
+  service = SqliteMemoryService(
+      f"{scheme}:///{quote(uri_path)}{query}", fts="off"
   )
-  assert len(resp.memories) == 1
-  assert resp.memories[0].content.parts[0].text == "readonly test content"
+  try:
+    response = await service.search_memory(
+        app_name="app", user_id="user", query="saved"
+    )
+    assert len(response.memories) == 1
+    assert response.memories[0].content.parts[0].text == "saved content"
+    new_event = _make_event("user", "new content", 2.0, event_id="new")
+    if mode == "ro":
+      with pytest.raises(sqlite3.OperationalError, match="readonly"):
+        await service.add_events_to_memory(
+            app_name="app", user_id="user", events=[new_event], session_id="s1"
+        )
+    else:
+      await service.add_events_to_memory(
+          app_name="app", user_id="user", events=[new_event], session_id="s1"
+      )
+  finally:
+    await service.close()
 
-  with pytest.raises((sqlite3.OperationalError, aiosqlite.OperationalError)):
-    new_event = _make_event("user", "cannot write", 2.0)
-    await ro_service.add_events_to_memory(
-        app_name="app", user_id="user", events=[new_event], session_id="s1"
+  reopened_service = SqliteMemoryService(db_path=db_path, fts="off")
+  try:
+    response = await reopened_service.search_memory(
+        app_name="app", user_id="user", query="new"
     )
+    assert len(response.memories) == (0 if mode == "ro" else 1)
+  finally:
+    await reopened_service.close()
 
 
 def test_sqlite_uri_does_not_create_sqlite_colon_directory(tmp_path):