fix(sessions): honor Windows sqlite session URIs

src/google/adk/cli/service_registry.py

@@ -268,15 +268,17 @@ def database_session_factory(uri: str, **kwargs: Any) -> BaseSessionService:
     return DatabaseSessionService(db_url=uri, **kwargs_copy)
 
   def sqlite_session_factory(uri: str, **kwargs: Any) -> BaseSessionService:
+    from ..sessions.sqlite_session_service import _parse_db_path
     from ..sessions.sqlite_session_service import SqliteSessionService
 
     parsed = urlparse(uri)
-    db_path = parsed.path
-    if not db_path:
+    if not parsed.path:
       # Treat sqlite:// without a path as an in-memory session service.
       return memory_session_factory("memory://", **kwargs)
-    elif db_path.startswith("/"):
-      db_path = db_path[1:]
+
+    # Same unquote / Windows drive rules as SqliteSessionService so a
+    # percent-encoded path is not stored as a literal "%20" filename.
+    db_path, _, _ = _parse_db_path(uri)
 
     # SqliteSessionService only accepts db_path, warn if extra kwargs provided
     ignored_kwargs = {k: v for k, v in kwargs.items() if k != "agents_dir"}

src/google/adk/sessions/sqlite_session_service.py

@@ -26,6 +26,7 @@
 from typing import Optional
 from urllib.parse import unquote
 from urllib.parse import urlparse
+from urllib.request import url2pathname
 
 import aiosqlite
 from google.adk.platform import time as platform_time
@@ -118,6 +119,33 @@
 ])
 
 
+def _windows_drive_path_from_url_path(raw_path: str) -> str | None:
+  """Returns a Windows filesystem path for a drive-letter sqlite URL path.
+
+  SQLAlchemy documents ``sqlite:///C:/path/to.db`` (three slashes plus the
+  drive). ``urlparse`` yields ``/C:/path/to.db``. Four slashes plus a drive
+  (``sqlite:////C:/path/to.db``) yields ``//C:/path/to.db``. ``file://``
+  artifact URIs already run those shapes through ``url2pathname``; sqlite
+  session URIs must do the same so a space-containing Windows path is not
+  left as a leading-slash URL path.
+  """
+  if os.name != "nt":
+    return None
+
+  path = raw_path.replace("\\", "/")
+  # Four slashes before a drive letter leave an extra leading slash.
+  if (
+      path.startswith("//")
+      and len(path) >= 4
+      and path[2].isalpha()
+      and path[3] == ":"
+  ):
+    path = path[1:]
+  if len(path) >= 3 and path[0] == "/" and path[1].isalpha() and path[2] == ":":
+    return url2pathname(path)
+  return None
+
+
 def _parse_db_path(db_path: str) -> tuple[str, str, bool]:
   """Normalizes a SQLite db path from a URL or filesystem path.
 
@@ -132,6 +160,7 @@ def _parse_db_path(db_path: str) -> tuple[str, str, bool]:
     conventions:
       - `sqlite:///relative.db` is a path relative to the current working dir.
       - `sqlite:////absolute.db` is an absolute filesystem path.
+      - `sqlite:///C:/path/to.db` is a Windows absolute path.
   """
   if not db_path.startswith(("sqlite:", "sqlite+aiosqlite:")):
     return db_path, db_path, False
@@ -141,11 +170,15 @@ def _parse_db_path(db_path: str) -> tuple[str, str, bool]:
   if not raw_path:
     return db_path, db_path, False
 
-  normalized_path = raw_path
-  if normalized_path.startswith("//"):
-    normalized_path = normalized_path[1:]
-  elif normalized_path.startswith("/"):
-    normalized_path = normalized_path[1:]
+  windows_path = _windows_drive_path_from_url_path(raw_path)
+  if windows_path is not None:
+    normalized_path = windows_path
+  elif raw_path.startswith("//"):
+    normalized_path = raw_path[1:]
+  elif raw_path.startswith("/"):
+    normalized_path = raw_path[1:]
+  else:
+    normalized_path = raw_path
 
   if parsed.query:
     # sqlite3 only treats the filename as a URI when it starts with `file:`.

tests/unittests/cli/test_service_registry.py

@@ -128,6 +128,27 @@ def test_create_artifact_service_gcs(registry, mock_services):
   )
 
 
+def test_sqlite_session_factory_normalizes_windows_sqlite_uri(
+    registry, mock_services, monkeypatch
+):
+  monkeypatch.setattr(
+      "google.adk.sessions.sqlite_session_service.os",
+      SimpleNamespace(name="nt"),
+  )
+  mocked_url2pathname = mock.Mock(return_value=r"C:\tmp\adk sessions.db")
+  monkeypatch.setattr(
+      "google.adk.sessions.sqlite_session_service.url2pathname",
+      mocked_url2pathname,
+  )
+
+  registry.create_session_service("sqlite:///C:/tmp/adk%20sessions.db")
+
+  mocked_url2pathname.assert_called_once_with("/C:/tmp/adk sessions.db")
+  mock_services["sqlite_session"].assert_called_once_with(
+      db_path=r"C:\tmp\adk sessions.db"
+  )
+
+
 def test_file_artifact_factory_normalizes_windows_file_uri(monkeypatch):
   monkeypatch.setattr(service_registry, "os", SimpleNamespace(name="nt"))
   mocked_url2pathname = mock.Mock(return_value=r"C:\tmp\adk artifacts")

tests/unittests/sessions/test_session_service.py

@@ -39,6 +39,7 @@
 from google.adk.sessions.schemas.v0 import DynamicPickleType
 from google.adk.sessions.schemas.v1 import StorageSession
 from google.adk.sessions.session import Session
+from google.adk.sessions.sqlite_session_service import _parse_db_path
 from google.adk.sessions.sqlite_session_service import SqliteSessionService
 from google.adk.sessions.vertex_ai_session_service import VertexAiSessionService
 from google.adk.tools.tool_confirmation import ToolConfirmation
@@ -294,7 +295,9 @@ async def test_sqlite_session_service_preserves_uri_query_parameters(
     conn.execute('CREATE TABLE IF NOT EXISTS t (id INTEGER)')
     conn.commit()
 
-  service = SqliteSessionService(f'sqlite+aiosqlite:///{db_path}?mode=ro')
+  service = SqliteSessionService(
+      f'sqlite+aiosqlite:///{db_path.as_posix()}?mode=ro'
+  )
   # `mode=ro` opens the DB read-only; schema creation should fail.
   with pytest.raises(sqlite3.OperationalError, match=r'readonly'):
     await service.create_session(app_name='app', user_id='user')
@@ -303,12 +306,73 @@ async def test_sqlite_session_service_preserves_uri_query_parameters(
 @pytest.mark.asyncio
 async def test_sqlite_session_service_accepts_absolute_sqlite_urls(tmp_path):
   abs_db_path = tmp_path / 'absolute.db'
-  abs_url = 'sqlite+aiosqlite:////' + str(abs_db_path).lstrip('/')
+  # Three slashes plus a POSIX path: Unix stays `////tmp/...`, Windows
+  # becomes the documented `sqlite:///C:/path/to.db` form.
+  abs_url = f'sqlite+aiosqlite:///{abs_db_path.as_posix()}'
   service = SqliteSessionService(abs_url)
   await service.create_session(app_name='app', user_id='user')
   assert abs_db_path.exists()
 
 
+@pytest.mark.asyncio
+async def test_sqlite_session_service_decodes_windows_percent_encoded_path(
+    tmp_path,
+):
+  if os.name != 'nt':
+    pytest.skip('Windows drive-letter sqlite URIs')
+
+  abs_db_path = tmp_path / 'adk sessions.db'
+  encoded_path = abs_db_path.as_posix().replace(' ', '%20')
+  service = SqliteSessionService(f'sqlite+aiosqlite:///{encoded_path}')
+  await service.create_session(app_name='app', user_id='user')
+  assert abs_db_path.exists()
+  assert not (tmp_path / 'adk%20sessions.db').exists()
+
+
+def test_parse_db_path_keeps_relative_sqlite_url_on_windows(monkeypatch):
+  monkeypatch.setattr(
+      'google.adk.sessions.sqlite_session_service.os.name',
+      'nt',
+  )
+  mocked_url2pathname = mock.Mock(
+      side_effect=AssertionError(
+          'relative sqlite URIs must not go through url2pathname'
+      )
+  )
+  monkeypatch.setattr(
+      'google.adk.sessions.sqlite_session_service.url2pathname',
+      mocked_url2pathname,
+  )
+  fs_path, connect_path, is_uri = _parse_db_path('sqlite:///test.db')
+  assert fs_path == 'test.db'
+  assert connect_path == 'test.db'
+  assert is_uri is False
+  mocked_url2pathname.assert_not_called()
+
+
+def test_parse_db_path_windows_drive_and_percent_encoding(monkeypatch):
+  monkeypatch.setattr(
+      'google.adk.sessions.sqlite_session_service.os.name',
+      'nt',
+  )
+  monkeypatch.setattr(
+      'google.adk.sessions.sqlite_session_service.url2pathname',
+      lambda path: r'C:\tmp\adk sessions.db',
+  )
+
+  fs_path, connect_path, is_uri = _parse_db_path(
+      'sqlite:///C:/tmp/adk%20sessions.db'
+  )
+  assert fs_path == r'C:\tmp\adk sessions.db'
+  assert connect_path == r'C:\tmp\adk sessions.db'
+  assert is_uri is False
+
+  fs_path, _, _ = _parse_db_path(
+      'sqlite+aiosqlite:////C:/tmp/adk%20sessions.db'
+  )
+  assert fs_path == r'C:\tmp\adk sessions.db'
+
+
 @pytest.mark.asyncio
 async def test_get_empty_session(session_service):
   assert not await session_service.get_session(