@@ -352,16 +352,17 @@ def __init__(
           event.listen(db_engine.sync_engine, "connect", _set_sqlite_pragma)
 
       except Exception as e:
+        redacted_url = _schema_check_utils._redact_db_url(db_url)
         if isinstance(e, ArgumentError):
           raise ValueError(
-              f"Invalid database URL format or argument '{db_url}'."
+              f"Invalid database URL format or argument '{redacted_url}'."
           ) from e
         if isinstance(e, ImportError):
           raise ValueError(
-              f"Database related module not found for URL '{db_url}'."
+              f"Database related module not found for URL '{redacted_url}'."
           ) from e
         raise ValueError(
-            f"Failed to create database engine for URL '{db_url}'"
+            f"Failed to create database engine for URL '{redacted_url}'"
         ) from e
     else:
       self._owns_db_engine = False
 
@@ -22,6 +22,7 @@
   from sqlalchemy import create_engine as create_sync_engine
   from sqlalchemy import inspect
   from sqlalchemy import text
+  from sqlalchemy.engine import make_url
 except ImportError:
   pass
 
@@ -31,6 +32,9 @@
 
 logger = logging.getLogger("google_adk." + __name__)
 
+_UNPARSEABLE_DB_URL = "<unparseable database URL>"
+_REDACTED_QUERY_VALUE = "REDACTED"
+
 SCHEMA_VERSION_KEY = "schema_version"
 SCHEMA_VERSION_0_PICKLE = "0"
 SCHEMA_VERSION_1_JSON = "1"
@@ -125,6 +129,24 @@ def to_sync_url(db_url: str) -> str:
   return db_url
 
 
+def _redact_db_url(db_url: str) -> str:
+  """Returns the URL with its credentials masked, for logs and error messages.
+
+  A database URL carries the password in the userinfo component, and drivers
+  also accept secrets as query parameters, so every query value is masked
+  rather than only the ones with a recognizable name. Redaction happens while
+  an error is being reported, so it never raises: an unparseable URL yields a
+  fixed placeholder rather than the original string.
+  """
+  try:
+    url = make_url(db_url)
+    if url.query:
+      url = url.set(query={key: _REDACTED_QUERY_VALUE for key in url.query})
+    return str(url.render_as_string(hide_password=True))
+  except Exception:  # pylint: disable=broad-except
+    return _UNPARSEABLE_DB_URL
+
+
 def get_db_schema_version(db_url: str) -> str:
   """Reads schema version from DB.
 
@@ -146,7 +168,7 @@ def get_db_schema_version(db_url: str) -> str:
   except Exception:
     logger.warning(
         "Failed to get schema version from database %s.",
-        db_url,
+        _redact_db_url(db_url),
     )
     raise
   finally:
 
@@ -284,7 +284,10 @@ def migrate(
   source_sync_url = _schema_check_utils.to_sync_url(source_db_url)
   dest_sync_url = _schema_check_utils.to_sync_url(dest_db_url)
 
-  logger.info(f"Connecting to source database: {source_db_url}")
+  logger.info(
+      "Connecting to source database: %s",
+      _schema_check_utils._redact_db_url(source_db_url),
+  )
   if allow_unsafe_unpickling:
     logger.warning(
         "Unsafe pickle migration mode is enabled. Only use this with a trusted"
@@ -297,7 +300,10 @@ def migrate(
     logger.error(f"Failed to connect to source database: {e}")
     raise RuntimeError(f"Failed to connect to source database: {e}") from e
 
-  logger.info(f"Connecting to destination database: {dest_db_url}")
+  logger.info(
+      "Connecting to destination database: %s",
+      _schema_check_utils._redact_db_url(dest_db_url),
+  )
   try:
     dest_engine = create_engine(dest_sync_url)
     v1.Base.metadata.create_all(dest_engine)
 
@@ -38,7 +38,10 @@ def migrate(source_db_url: str, dest_db_path: str) -> None:
   # them automatically converted to 'sqlite://...' for migration.
   source_sync_url = _schema_check_utils.to_sync_url(source_db_url)
 
-  logger.info(f"Connecting to source database: {source_db_url}")
+  logger.info(
+      "Connecting to source database: %s",
+      _schema_check_utils._redact_db_url(source_db_url),
+  )
   try:
     engine = create_engine(source_sync_url)
     v0_schema.Base.metadata.create_all(
 
@@ -82,8 +82,9 @@ def upgrade(
   current_version = _schema_check_utils.get_db_schema_version(source_db_url)
   if current_version == LATEST_VERSION:
     logger.info(
-        f"Database {source_db_url} is already at latest version"
-        f" {LATEST_VERSION}. No migration needed."
+        "Database %s is already at latest version %s. No migration needed.",
+        _schema_check_utils._redact_db_url(source_db_url),
+        LATEST_VERSION,
     )
     return
 
@@ -118,7 +119,10 @@ def upgrade(
         logger.debug("Created temp db %s for step %d", out_url, i + 1)
 
       logger.info(
-          f"Migrating from {in_url} to {out_url} (schema v{end_version})..."
+          "Migrating from %s to %s (schema v%s)...",
+          _schema_check_utils._redact_db_url(in_url),
+          _schema_check_utils._redact_db_url(out_url),
+          end_version,
       )
       if migrate_func is migrate_from_sqlalchemy_pickle.migrate:
         migrate_func(
 
@@ -18,9 +18,11 @@
 import contextlib
 from datetime import datetime
 from datetime import timezone
+import logging
 import os
 import pickle
 import time
+from unittest import mock
 
 from fastapi.openapi.models import HTTPBearer
 from google.adk.auth.auth_tool import AuthConfig
@@ -29,6 +31,8 @@
 from google.adk.events.ui_widget import UiWidget
 from google.adk.sessions.migration import _schema_check_utils
 from google.adk.sessions.migration import migrate_from_sqlalchemy_pickle as mfsp
+from google.adk.sessions.migration import migrate_from_sqlalchemy_sqlite as mfss
+from google.adk.sessions.migration import migration_runner
 from google.adk.sessions.schemas import v0
 from google.adk.sessions.schemas import v1
 from google.adk.tools.tool_confirmation import ToolConfirmation
@@ -116,6 +120,118 @@ def test_to_sync_url_empty_string(self):
     assert _schema_check_utils.to_sync_url("") == ""
 
 
+class TestRedactDbUrl:
+  """Tests for the _redact_db_url function."""
+
+  def test_password_is_masked(self):
+    redacted = _schema_check_utils._redact_db_url(
+        "postgresql+asyncpg://user:sup3r-s3cret@host:5432/db"
+    )
+    assert redacted == "postgresql+asyncpg://user:***@host:5432/db"
+
+  def test_unparseable_url_falls_back_to_placeholder(self):
+    """Redaction runs while reporting an error, so it must never raise."""
+    assert (
+        _schema_check_utils._redact_db_url("definitely not a url sup3r-s3cret")
+        == "<unparseable database URL>"
+    )
+
+  def test_query_parameter_values_are_masked(self):
+    """Drivers accept secrets as query parameters, so every value is masked."""
+    redacted = _schema_check_utils._redact_db_url(
+        "postgresql://user@host:5432/db?password=sup3r-s3cret&sslmode=require"
+    )
+    assert redacted == (
+        "postgresql://user@host:5432/db?password=REDACTED&sslmode=REDACTED"
+    )
+
+  def test_schema_version_failure_warning_hides_password(self, caplog):
+    db_url = "postgresql+asyncpg://user:sup3r-s3cret@host:5432/db"
+
+    with mock.patch.object(
+        _schema_check_utils,
+        "create_sync_engine",
+        side_effect=RuntimeError("boom"),
+    ):
+      with caplog.at_level(logging.WARNING):
+        with pytest.raises(RuntimeError):
+          _schema_check_utils.get_db_schema_version(db_url)
+
+    assert "sup3r-s3cret" not in caplog.text
+    assert "postgresql+asyncpg://user:***@host:5432/db" in caplog.text
+
+
+_SOURCE_URL = "postgresql+asyncpg://user:sup3r-s3cret@host:5432/src"
+_DEST_URL = "postgresql+asyncpg://user:0ther-s3cret@host:5432/dst"
+
+
+class TestMigrationLogsHidePassword:
+  """These entry points log their URLs on every run, not only on failure."""
+
+  def test_pickle_migration_connect_logs_are_redacted(self, caplog):
+    with mock.patch.object(
+        mfsp,
+        "create_engine",
+        side_effect=[mock.MagicMock(), RuntimeError("boom")],
+    ):
+      with caplog.at_level(logging.INFO):
+        with pytest.raises(RuntimeError):
+          mfsp.migrate(_SOURCE_URL, _DEST_URL)
+
+    assert "sup3r-s3cret" not in caplog.text
+    assert "0ther-s3cret" not in caplog.text
+    assert "postgresql+asyncpg://user:***@host:5432/src" in caplog.text
+    assert "postgresql+asyncpg://user:***@host:5432/dst" in caplog.text
+
+  def test_sqlite_migration_connect_log_is_redacted(self, caplog, tmp_path):
+    with mock.patch.object(
+        mfss, "create_engine", side_effect=RuntimeError("boom")
+    ):
+      with caplog.at_level(logging.INFO):
+        with pytest.raises(SystemExit):
+          mfss.migrate(_SOURCE_URL, str(tmp_path / "dest.db"))
+
+    assert "sup3r-s3cret" not in caplog.text
+    assert "postgresql+asyncpg://user:***@host:5432/src" in caplog.text
+
+  def test_runner_up_to_date_log_is_redacted(self, caplog):
+    with mock.patch.object(
+        _schema_check_utils,
+        "get_db_schema_version",
+        return_value=migration_runner.LATEST_VERSION,
+    ):
+      with caplog.at_level(logging.INFO):
+        migration_runner.upgrade(_SOURCE_URL, _DEST_URL)
+
+    assert "sup3r-s3cret" not in caplog.text
+    assert "postgresql+asyncpg://user:***@host:5432/src" in caplog.text
+
+  def test_runner_migration_step_log_is_redacted(self, caplog):
+    mock_migrate = mock.Mock()
+    with mock.patch.object(
+        _schema_check_utils,
+        "get_db_schema_version",
+        return_value=_schema_check_utils.SCHEMA_VERSION_0_PICKLE,
+    ):
+      with mock.patch.dict(
+          migration_runner.MIGRATIONS,
+          {
+              _schema_check_utils.SCHEMA_VERSION_0_PICKLE: (
+                  _schema_check_utils.SCHEMA_VERSION_1_JSON,
+                  mock_migrate,
+              )
+          },
+      ):
+        with caplog.at_level(logging.INFO):
+          migration_runner.upgrade(_SOURCE_URL, _DEST_URL)
+
+    mock_migrate.assert_called_once_with(_SOURCE_URL, _DEST_URL)
+    assert "sup3r-s3cret" not in caplog.text
+    assert "0ther-s3cret" not in caplog.text
+    assert "postgresql+asyncpg://user:***@host:5432/src" in caplog.text
+    assert "postgresql+asyncpg://user:***@host:5432/dst" in caplog.text
+
+
 def test_migrate_from_sqlalchemy_pickle(tmp_path):
   """Tests for migrate_from_sqlalchemy_pickle."""
   source_db_path = tmp_path / "source_pickle.db"
 
@@ -45,6 +45,7 @@
 from sqlalchemy import select
 from sqlalchemy import text
 from sqlalchemy import update
+from sqlalchemy.exc import ArgumentError
 from sqlalchemy.ext.asyncio import create_async_engine
 from sqlalchemy.pool import StaticPool
 
@@ -2236,6 +2237,48 @@ async def test_database_session_service_requires_one_argument():
     )
 
 
+@pytest.mark.parametrize(
+    'raised_error',
+    [
+        RuntimeError('boom'),
+        ArgumentError('bad argument'),
+        ImportError('no driver'),
+    ],
+)
+def test_database_session_service_engine_error_hides_password(raised_error):
+  """Engine creation errors must not put the DB password in the message."""
+  password = 'sup3r-s3cret'
+  db_url = f'postgresql+asyncpg://user:{password}@localhost:5432/db'
+
+  with mock.patch.object(
+      database_session_service,
+      'create_async_engine',
+      side_effect=raised_error,
+  ):
+    with pytest.raises(ValueError) as exc_info:
+      DatabaseSessionService(db_url)
+
+  message = str(exc_info.value)
+  assert password not in message
+  # The redacted URL is still there, so the error stays diagnosable.
+  assert 'postgresql+asyncpg://user:***@localhost:5432/db' in message
+
+
+def test_database_session_service_malformed_url_reports_usable_error():
+  """A URL too malformed to parse still yields a usable, leak-free error."""
+  # make_url() itself rejects this, so redaction cannot parse it either and
+  # must fall back to a placeholder rather than echoing the raw string.
+  db_url = 'definitely not a url sup3r-s3cret'
+
+  with pytest.raises(ValueError) as exc_info:
+    DatabaseSessionService(db_url)
+
+  message = str(exc_info.value)
+  assert 'sup3r-s3cret' not in message
+  assert 'Invalid database URL format or argument' in message
+  assert isinstance(exc_info.value.__cause__, ArgumentError)
+
+
 @pytest.mark.asyncio
 async def test_database_session_service_sqlite_file_timestamp_read_after_reopen(
     tmp_path,