diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9199640..07d76ad 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,14 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-### Fixed
-
-- The API key now resolves from `HOTDATA_API_KEY` following the README quickstart: setting the env var (with no `credentials=` argument) populates the destination, instead of leaving `credentials.api_key` unset and failing with an opaque `NoneType` error deep in `hotdata-framework`. Missing `api_key`/`workspace_id` now raises a clear `ConfigurationValueError` at setup naming the missing field.
-
 ### Changed
 
 - **Breaking:** `workspace_id` moved out of `HotdataCredentials` (authentication) to a top-level `hotdata(workspace_id=...)` param / `HotdataClientConfiguration` field (configuration), matching the SDK's `Configuration(api_key=, workspace_id=)` shape. It is a **param with no environment-variable fallback** — the `HOTDATA_WORKSPACE` env var is no longer read on this path (the API key remains env-backed, as a secret). Passing `workspace_id` inside a `credentials={...}` dict still works but is deprecated (hoisted with a `DeprecationWarning`); constructing `HotdataCredentials(workspace_id=...)` now raises `TypeError` — pass `workspace_id=` to `hotdata(...)` instead.
 
+### Fixed
+
+- The API key now resolves from `HOTDATA_API_KEY` following the README quickstart: setting the env var (with no `credentials=` argument) populates the destination, instead of leaving `credentials.api_key` unset and failing with an opaque `NoneType` error deep in `hotdata-framework`. Missing `api_key`/`workspace_id` now raises a clear `ConfigurationValueError` at setup naming the missing field.
+- Managed-database name resolution is now collision-safe and resolved once per run. Hotdata database names are not unique, and the destination previously took the first `list_databases` match on every operation — so a name collision could silently read from, write to, or **drop** the wrong database. It now raises a clear error when a name matches more than one database, resolves the name to its record a single time per run (cached on the shared config), and addresses every subsequent operation (load/add/list/query/drop) by id.
+
 
 ## [0.10.0] - 2026-07-20
 
diff --git a/src/hotdata_dlt_destination/hotdata_client.py b/src/hotdata_dlt_destination/hotdata_client.py
index 13b1675..38e0ffe 100644
--- a/src/hotdata_dlt_destination/hotdata_client.py
+++ b/src/hotdata_dlt_destination/hotdata_client.py
@@ -5,20 +5,82 @@
 from hotdata_framework.databases import ManagedDatabase
 from hotdata_framework.managed_client import ManagedDatabaseClient
 
+from hotdata_dlt_destination.errors import HotdataTerminalError
+
 
 class HotdataClient(ManagedDatabaseClient):
     """Managed-database client used by the dlt destination.
 
-    Adds cross-run schema evolution on top of the shared ``hotdata_framework``
-    client. The base client only creates a managed database with its initial
-    tables; this override additionally reconciles tables on an already-existing
-    database. When a later run requires a table that the database is missing,
-    the table is declared in place via ``add_managed_table`` (the table is added
-    empty and populated by the subsequent load) — no data is moved and existing
-    tables, including dlt's ``_dlt_version`` / ``_dlt_loads`` /
-    ``_dlt_pipeline_state`` bookkeeping, are left untouched.
+    Adds two things on top of the shared ``hotdata_framework`` client:
+
+    * **Cross-run schema evolution** — when a later run requires a table the
+      database is missing, the table is declared in place via
+      ``add_managed_table`` (added empty, populated by the subsequent load); no
+      data is moved and existing tables, including dlt's ``_dlt_version`` /
+      ``_dlt_loads`` / ``_dlt_pipeline_state`` bookkeeping, are left untouched.
+    * **Collision-safe, resolve-once addressing** — a database name is resolved
+      to its record once per run (cached via :meth:`bind_run_cache`) and every
+      subsequent operation addresses the database by id. Resolution raises on an
+      ambiguous name instead of silently taking the first match.
     """
 
+    # Run-scoped store bound via bind_run_cache(); resolution is cached on it so
+    # the whole run reuses one resolved/created record.
+    _run_cache: object | None = None
+
+    def bind_run_cache(self, cache: object) -> None:
+        """Bind a run-scoped store so a database resolves to its record once.
+
+        ``cache`` is any object that tolerates a ``_hotdata_resolved_db``
+        attribute — in practice the shared ``HotdataClientConfiguration``
+        instance, which every client built for a run points at.
+        """
+        self._run_cache = cache
+
+    # --- resolution -------------------------------------------------------
+
+    def _collision_safe_resolve(self, name_or_id: str) -> ManagedDatabase:
+        """Resolve a name/id to its record, raising on an ambiguous name.
+
+        Hotdata database names are not unique. Taking the first match can read,
+        write, or drop the wrong database, so a name that matches more than one
+        database raises instead. An id (matched exactly) is unambiguous.
+        """
+        databases = self._request_with_retry(self._runtime.list_managed_databases)
+        by_name = [db for db in databases if db.description == name_or_id]
+        if len(by_name) > 1:
+            raise HotdataTerminalError(
+                f"Managed database name {name_or_id!r} is ambiguous: "
+                f"{len(by_name)} databases share it (ids: {sorted(db.id for db in by_name)}). "
+                "Address it by id to disambiguate."
+            )
+        if by_name:
+            return by_name[0]
+        by_id = [db for db in databases if db.id == name_or_id]
+        if by_id:
+            return by_id[0]
+        raise KeyError(name_or_id)
+
+    def _resolve(self, name_or_id: str) -> ManagedDatabase:
+        """Resolve once per run, then serve the cached (id-addressable) record."""
+        cache = self._run_cache
+        if cache is not None:
+            cached = getattr(cache, "_hotdata_resolved_db", None)
+            if cached is not None and name_or_id in (
+                cached.id,
+                getattr(cached, "description", None),
+            ):
+                return cached
+        db = self._collision_safe_resolve(name_or_id)
+        self._cache_db(db)
+        return db
+
+    def _cache_db(self, db: ManagedDatabase | None) -> None:
+        if self._run_cache is not None:
+            self._run_cache._hotdata_resolved_db = db
+
+    # --- lifecycle --------------------------------------------------------
+
     def ensure_managed_database(
         self,
         name: str,
@@ -30,72 +92,72 @@ def ensure_managed_database(
     ) -> ManagedDatabase:
         # keys: table name -> key columns (enables delete/update/upsert on it)
         keys = keys or {}
-        runtime = self._runtime
 
-        # Resolve is called directly (not via _request_with_retry) so its KeyError
-        # "not found" signal is preserved rather than mapped to a terminal error.
         try:
-            db = runtime.resolve_managed_database(name)
+            db = self._resolve(name)
         except KeyError:
             if not create_if_missing:
                 raise
-            return self._request_with_retry(
-                lambda: runtime.create_managed_database(
+            db = self._request_with_retry(
+                lambda: self._runtime.create_managed_database(
                     description=name, schema=schema, tables=sorted(set(tables)), keys=keys
                 )
             )
+            self._cache_db(db)
+            return db
 
         existing = {
             managed_table.table
             for managed_table in self._request_with_retry(
-                lambda: runtime.list_managed_tables(name, schema=schema)
+                lambda: self._runtime.list_managed_tables(db.id, schema=schema)
             )
         }
         # Declare any newly-required tables additively, in place, carrying their
         # key. dlt calls ``initialize_storage`` with the full table set before any
         # load job runs, so by load time this is normally a no-op.
         for table in sorted(set(tables) - existing):
-            self._add_managed_table(name, table, schema=schema, key=keys.get(table))
+            self._add_managed_table(db.id, table, schema=schema, key=keys.get(table))
         return db
 
     def _add_managed_table(
-        self, name: str, table: str, *, schema: str, key: list[str] | None = None
+        self, database: str, table: str, *, schema: str, key: list[str] | None = None
     ) -> None:
-        runtime = self._runtime
         self._request_with_retry(
-            lambda: runtime.add_managed_table(name, table, schema=schema, key=key)
+            lambda: self._runtime.add_managed_table(database, table, schema=schema, key=key)
         )
 
     def drop_managed_database(self, name: str) -> None:
         """Delete the managed database if it exists (used for dlt dev_mode / refresh)."""
-        runtime = self._runtime
         try:
-            db = runtime.resolve_managed_database(name)
+            db = self._resolve(name)
         except KeyError:
             return
-        self._request_with_retry(lambda: runtime.delete_managed_database(db.id))
+        self._request_with_retry(lambda: self._runtime.delete_managed_database(db.id))
+        self._cache_db(None)
 
     def resolve_managed_database(self, name: str) -> ManagedDatabase:
-        """Resolve a managed database by display name to its record (carrying ``.id``).
+        """Resolve a managed database by display name (or id) to its record.
 
-        Delegates to the runtime client, preserving its ``KeyError`` "not found" signal.
+        Raises ``KeyError`` when nothing matches and ``HotdataTerminalError`` when
+        the name is shared by more than one database.
         """
-        return self._runtime.resolve_managed_database(name)
+        return self._resolve(name)
+
+    def load_managed_table(self, database: str, table: str, **kwargs):
+        """Load parquet into a managed table, addressing the database by id."""
+        db = self._resolve(database)
+        return super().load_managed_table(db.id, table, **kwargs)
 
     def execute_sql(self, sql: str, *, database: str) -> pa.Table:
         """Run a SQL query scoped to ``database`` and return the result as Arrow.
 
-        The read/dataset interface goes through here. The base client has no
-        general query entrypoint of its own — only the private database-scoped
-        submit + Arrow fetch that :meth:`fetch_table` uses — so this mirrors that
-        dance for arbitrary SQL: resolve the managed database name to its id,
-        submit the query, poll until the result is ready, and fetch it as a
-        ``pyarrow.Table``. An empty table is returned when the query produces no
-        out-of-band result (e.g. a statement with no result set).
+        Resolves the managed database to its id (once per run), submits the query,
+        polls until the result is ready, and fetches it as a ``pyarrow.Table``. An
+        empty table is returned when the query produces no out-of-band result.
         """
 
         def operation() -> pa.Table:
-            db = self._runtime.resolve_managed_database(database)
+            db = self._resolve(database)
             result_id = self._query_database_scoped(sql, database_id=db.id)
             if result_id is None:
                 return pa.table({})
@@ -109,10 +171,38 @@ def operation() -> pa.Table:
 
     def list_managed_tables(self, database: str, *, schema: str) -> list:
         """List the managed tables in ``database``/``schema`` (used by ``has_dataset``)."""
-        runtime = self._runtime
+        db = self._resolve(database)
         return self._request_with_retry(
-            lambda: runtime.list_managed_tables(database, schema=schema)
+            lambda: self._runtime.list_managed_tables(db.id, schema=schema)
         )
 
+    def table_is_synced(self, database: str, table: str, *, schema: str) -> bool:
+        db = self._resolve(database)
+        for managed_table in self._request_with_retry(
+            lambda: self._runtime.list_managed_tables(db.id, schema=schema)
+        ):
+            if managed_table.table == table:
+                return managed_table.synced
+        return False
+
+    def fetch_table(self, *, database: str, schema: str, table: str) -> pa.Table | None:
+        def operation() -> pa.Table | None:
+            db = self._resolve(database)
+            if not self._table_is_synced_for(db, table, schema=schema):
+                return None
+            sql = f'SELECT * FROM "default"."{schema}"."{table}"'
+            result_id = self._query_database_scoped(sql, database_id=db.id)
+            if result_id is None:
+                return None
+            return self._fetch_result_arrow(result_id, database_id=db.id)
+
+        return self._request_with_retry(operation)
+
+    def _table_is_synced_for(self, db: ManagedDatabase, table: str, *, schema: str) -> bool:
+        for managed_table in self._runtime.list_managed_tables(db.id, schema=schema):
+            if managed_table.table == table:
+                return managed_table.synced
+        return False
+
 
 __all__ = ["HotdataClient"]
diff --git a/src/hotdata_dlt_destination/job_client.py b/src/hotdata_dlt_destination/job_client.py
index f102a94..2a1347c 100644
--- a/src/hotdata_dlt_destination/job_client.py
+++ b/src/hotdata_dlt_destination/job_client.py
@@ -107,6 +107,8 @@ def _hotdata_api(config: HotdataClientConfiguration) -> Iterator[HotdataClient]:
         max_retries=config.max_retries,
         retry_backoff_seconds=config.retry_backoff_seconds,
     )
+    # Share the run's resolved-database cache across every short-lived client.
+    api.bind_run_cache(config)
     try:
         yield api
     finally:
diff --git a/src/hotdata_dlt_destination/sql_client.py b/src/hotdata_dlt_destination/sql_client.py
index 9f7b10b..39edde0 100644
--- a/src/hotdata_dlt_destination/sql_client.py
+++ b/src/hotdata_dlt_destination/sql_client.py
@@ -177,6 +177,7 @@ def open_connection(self) -> HotdataClient:
             max_retries=self._config.max_retries,
             retry_backoff_seconds=self._config.retry_backoff_seconds,
         )
+        self._client.bind_run_cache(self._config)
         return self._client
 
     def close_connection(self) -> None:
diff --git a/tests/test_client.py b/tests/test_client.py
index 20d6f9e..b9ba6a7 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -2,189 +2,174 @@
 
 import pytest
 
+from hotdata_dlt_destination.errors import HotdataTerminalError
 from hotdata_dlt_destination.hotdata_client import HotdataClient
 
 
-def test_upload_and_load_managed_table() -> None:
-    class FakeRuntime:
-        def __init__(self) -> None:
-            self.upload_calls = 0
-            self.load_calls = 0
+def _db(db_id: str, name: str, conn: str = "conn") -> SimpleNamespace:
+    return SimpleNamespace(id=db_id, description=name, default_connection_id=conn)
 
-        def upload_parquet(self, path: str) -> str:
-            self.upload_calls += 1
-            assert path.endswith(".parquet")
-            return "upload_1"
-
-        def load_managed_table(
-            self,
-            database: str,
-            table: str,
-            *,
-            schema: str,
-            upload_id: str,
-            mode: str = "replace",
-            key: list[str] | None = None,
-        ) -> SimpleNamespace:
-            self.load_calls += 1
-            assert database == "dlt"
-            assert table == "orders"
-            assert schema == "public"
-            assert upload_id == "upload_1"
-            return SimpleNamespace(
-                connection_id="conn_1",
-                schema_name=schema,
-                table_name=table,
-                row_count=1,
-                full_name=f"{database}.{schema}.{table}",
-            )
-
-        def close(self) -> None:
-            return None
 
+def _client(runtime, *, cache=None) -> HotdataClient:
     client = HotdataClient(
         api_key="k",
-        workspace_id="ws_1",
+        workspace_id="ws",
         api_base_url="https://api.hotdata.dev",
-        max_retries=2,
+        max_retries=1,
         retry_backoff_seconds=0.0,
     )
-    fake_runtime = FakeRuntime()
-    client._runtime = fake_runtime
+    client._runtime = runtime
+    if cache is not None:
+        client.bind_run_cache(cache)
+    return client
 
-    upload_id = client.upload_parquet("/tmp/batch.parquet")
-    loaded = client.load_managed_table(
-        "dlt",
-        "orders",
-        schema="public",
-        upload_id=upload_id,
-    )
 
-    assert upload_id == "upload_1"
-    assert loaded.full_name == "dlt.public.orders"
-    assert fake_runtime.upload_calls == 1
-    assert fake_runtime.load_calls == 1
+# --- resolution: collision-safe, id-addressed, resolve-once ---------------
+
+
+def test_resolve_returns_single_match_by_name() -> None:
+    class FakeRuntime:
+        def list_managed_databases(self):
+            return [_db("db_1", "dlt")]
+
+        def close(self):
+            return None
+
+    client = _client(FakeRuntime())
+    assert client.resolve_managed_database("dlt").id == "db_1"
     client.close()
 
 
-def test_fetch_table_rows_skips_unsynced_tables() -> None:
+def test_resolve_by_id() -> None:
     class FakeRuntime:
-        def list_managed_tables(self, database: str, *, schema: str):
-            assert database == "dlt"
-            assert schema == "public"
-            return [SimpleNamespace(table="orders", synced=False)]
+        def list_managed_databases(self):
+            return [_db("db_1", "dlt")]
 
-        def close(self) -> None:
+        def close(self):
             return None
 
-    client = HotdataClient(
-        api_key="k",
-        workspace_id="ws_1",
-        api_base_url="https://api.hotdata.dev",
-        max_retries=1,
-        retry_backoff_seconds=0.0,
-    )
-    client._runtime = FakeRuntime()
+    client = _client(FakeRuntime())
+    assert client.resolve_managed_database("db_1").id == "db_1"
+    client.close()
 
-    rows = client.fetch_table_rows(database="dlt", schema="public", table="orders")
-    assert rows == []
+
+def test_resolve_missing_raises_keyerror() -> None:
+    class FakeRuntime:
+        def list_managed_databases(self):
+            return []
+
+        def close(self):
+            return None
+
+    client = _client(FakeRuntime())
+    with pytest.raises(KeyError):
+        client.resolve_managed_database("dlt")
     client.close()
 
 
-def test_fetch_table_rows_reads_synced_table() -> None:
-    # Patch module-level QueryApi and ArrowResultsApi so no real HTTP happens.
-    # The client implementation now lives in hotdata_framework.managed_client
-    # (re-exported here as HotdataClient), so patch the symbols there.
-    import hotdata_framework.managed_client as _mod
-    import pyarrow as pa
-    from hotdata.models.query_response import QueryResponse as _QR
+def test_resolve_raises_on_ambiguous_name() -> None:
+    # Hotdata names are not unique: >1 match must raise, never silently pick one.
+    class FakeRuntime:
+        def list_managed_databases(self):
+            return [_db("db_1", "dlt"), _db("db_2", "dlt")]
 
-    class FakeQueryApi:
-        def __init__(self, api):
-            pass
+        def close(self):
+            return None
 
-        def query(self, request, *, x_database_id):
-            assert x_database_id == "db_1"
-            assert 'SELECT * FROM "default"."public"."orders"' in request.sql
-            return _QR(
-                columns=["id", "name"],
-                rows=[[1, "alpha"]],
-                row_count=1,
-                preview_row_count=1,
-                truncated=False,
-                nullable=[False, False],
-                result_id="result_1",
-                query_run_id="qrun_1",
-                execution_time_ms=1,
-            )
+    client = _client(FakeRuntime())
+    with pytest.raises(HotdataTerminalError, match="ambiguous"):
+        client.resolve_managed_database("dlt")
+    client.close()
 
-    class FakeResultsApi:
-        def __init__(self, api):
-            pass
 
-        def get_result(self, result_id, *, x_database_id=None):
-            assert result_id == "result_1"
-            return SimpleNamespace(status="ready", result_id=result_id, error_message=None)
+def test_resolves_once_and_reuses_run_cache() -> None:
+    # The name is resolved once per run; a second client sharing the cache reuses it.
+    class FakeRuntime:
+        def __init__(self):
+            self.list_calls = 0
 
-    class FakeArrowResultsApi:
-        def __init__(self, api):
-            pass
+        def list_managed_databases(self):
+            self.list_calls += 1
+            return [_db("db_1", "dlt")]
 
-        # Required in the hotdata 0.6.0 SDK; framework >=0.6.1 passes it on
-        # every result read.
-        def get_result_arrow(self, result_id, *, x_database_id):
-            assert result_id == "result_1"
-            assert x_database_id == "db_1"
-            return pa.table({"id": [1], "name": ["alpha"]})
+        def list_managed_tables(self, database, *, schema):
+            assert database == "db_1"  # addressed by id after the first resolve
+            return []
+
+        def close(self):
+            return None
+
+    rt = FakeRuntime()
+    cache = SimpleNamespace()
+    client = _client(rt, cache=cache)
+    client.resolve_managed_database("dlt")
+    client.list_managed_tables("dlt", schema="public")
+
+    client2 = _client(rt, cache=cache)
+    assert client2.resolve_managed_database("dlt").id == "db_1"
+
+    assert rt.list_calls == 1
+    assert cache._hotdata_resolved_db.id == "db_1"
+    client.close()
+    client2.close()
 
+
+# --- upload / load: addressed by id ---------------------------------------
+
+
+def test_upload_and_load_managed_table_addresses_by_id() -> None:
     class FakeRuntime:
-        api = None
+        def __init__(self) -> None:
+            self.upload_calls = 0
+            self.load_database = None
 
-        def list_managed_tables(self, database: str, *, schema: str):
-            return [SimpleNamespace(table="orders", synced=True)]
+        def list_managed_databases(self):
+            return [_db("db_dlt", "dlt")]
 
-        def resolve_managed_database(self, name):
-            return SimpleNamespace(id="db_1")
+        def upload_parquet(self, path: str) -> str:
+            self.upload_calls += 1
+            assert path.endswith(".parquet")
+            return "upload_1"
+
+        def load_managed_table(
+            self, database, table, *, schema, upload_id, mode="replace", key=None
+        ):
+            self.load_database = database
+            return SimpleNamespace(
+                connection_id="conn_1",
+                schema_name=schema,
+                table_name=table,
+                row_count=1,
+                full_name=f"{database}.{schema}.{table}",
+            )
 
         def close(self) -> None:
             return None
 
-    orig_query_api = _mod.QueryApi
-    orig_results_api = _mod.ResultsApi
-    orig_arrow_api = _mod.ArrowResultsApi
-    _mod.QueryApi = FakeQueryApi
-    _mod.ResultsApi = FakeResultsApi
-    _mod.ArrowResultsApi = FakeArrowResultsApi
+    rt = FakeRuntime()
+    client = _client(rt)
 
-    client = HotdataClient(
-        api_key="k",
-        workspace_id="ws_1",
-        api_base_url="https://api.hotdata.dev",
-        max_retries=1,
-        retry_backoff_seconds=0.0,
-    )
-    client._runtime = FakeRuntime()
-
-    try:
-        rows = client.fetch_table_rows(database="dlt", schema="public", table="orders")
-        assert rows == [{"id": 1, "name": "alpha"}]
-    finally:
-        _mod.QueryApi = orig_query_api
-        _mod.ResultsApi = orig_results_api
-        _mod.ArrowResultsApi = orig_arrow_api
+    upload_id = client.upload_parquet("/tmp/batch.parquet")
+    loaded = client.load_managed_table("dlt", "orders", schema="public", upload_id=upload_id)
+
+    assert upload_id == "upload_1"
+    # the load is addressed by the resolved id, not the display name
+    assert rt.load_database == "db_dlt"
+    assert loaded.full_name == "db_dlt.public.orders"
+    assert rt.upload_calls == 1
     client.close()
 
 
+# --- fetch_table / execute_sql: carry the database scope ------------------
+
+
 def _patch_query_apis(monkeypatch, arrow_table, *, arrow_in_hotdata_client: bool) -> dict:
     """Patch the query/result APIs so no real HTTP happens.
 
     ``fetch_table`` resolves ArrowResultsApi from hotdata_framework.managed_client;
     ``execute_sql`` resolves it from hotdata_dlt_destination.hotdata_client.
-    Returns a recorder of the ``x_database_id`` scopes each read carried —
-    the hotdata 0.6.0 SDK requires the scope on ``get_result_arrow``, and
-    framework >=0.6.1 / ``execute_sql`` must pass it on every result read.
+    Returns a recorder of the ``x_database_id`` scopes each read carried.
     """
-    import pyarrow as pa  # noqa: F401  (arrow_table already built by caller)
     from hotdata.models.query_response import QueryResponse as _QR
 
     scopes: dict[str, list] = {"result": [], "arrow": []}
@@ -232,22 +217,10 @@ def get_result_arrow(self, result_id, *, x_database_id):
     return scopes
 
 
-def _client_with_runtime(runtime) -> HotdataClient:
-    client = HotdataClient(
-        api_key="k",
-        workspace_id="ws_1",
-        api_base_url="https://api.hotdata.dev",
-        max_retries=1,
-        retry_backoff_seconds=0.0,
-    )
-    client._runtime = runtime
-    return client
-
-
 def test_fetch_table_carries_database_scope(monkeypatch) -> None:
     # Hosted result endpoints reject requests without the database scope;
     # fetch_table (merge / state-sync read-back) must carry it on the result
-    # poll and the Arrow fetch (native x_database_id since framework 0.6.1).
+    # poll and the Arrow fetch, addressing by the resolved id.
     import pyarrow as pa
 
     scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1]}), arrow_in_hotdata_client=False)
@@ -255,16 +228,17 @@ def test_fetch_table_carries_database_scope(monkeypatch) -> None:
     class FakeRuntime:
         api = None
 
+        def list_managed_databases(self):
+            return [_db("db_42", "dlt")]
+
         def list_managed_tables(self, database, *, schema):
+            assert database == "db_42"
             return [SimpleNamespace(table="orders", synced=True)]
 
-        def resolve_managed_database(self, name):
-            return SimpleNamespace(id="db_42")
-
         def close(self):
             return None
 
-    client = _client_with_runtime(FakeRuntime())
+    client = _client(FakeRuntime())
     table = client.fetch_table(database="dlt", schema="public", table="orders")
     assert table is not None and table.num_rows == 1
     assert scopes["result"] == ["db_42"]
@@ -273,7 +247,6 @@ def close(self):
 
 
 def test_execute_sql_carries_database_scope(monkeypatch) -> None:
-    # The read/dataset path carries the scope too.
     import pyarrow as pa
 
     scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1, 2]}), arrow_in_hotdata_client=True)
@@ -281,13 +254,13 @@ def test_execute_sql_carries_database_scope(monkeypatch) -> None:
     class FakeRuntime:
         api = None
 
-        def resolve_managed_database(self, name):
-            return SimpleNamespace(id="db_99")
+        def list_managed_databases(self):
+            return [_db("db_99", "dlt")]
 
         def close(self):
             return None
 
-    client = _client_with_runtime(FakeRuntime())
+    client = _client(FakeRuntime())
     table = client.execute_sql('SELECT * FROM "default"."public"."spans"', database="dlt")
     assert table.num_rows == 2
     assert scopes["result"] == ["db_99"]
@@ -295,23 +268,36 @@ def close(self):
     client.close()
 
 
+def test_fetch_table_rows_skips_unsynced_tables() -> None:
+    class FakeRuntime:
+        def list_managed_databases(self):
+            return [_db("db_1", "dlt")]
+
+        def list_managed_tables(self, database, *, schema):
+            assert database == "db_1"
+            return [SimpleNamespace(table="orders", synced=False)]
+
+        def close(self) -> None:
+            return None
+
+    client = _client(FakeRuntime())
+    rows = client.fetch_table_rows(database="dlt", schema="public", table="orders")
+    assert rows == []
+    client.close()
+
+
 def test_fetch_table_rows_returns_empty_when_table_missing() -> None:
     class FakeRuntime:
-        def list_managed_tables(self, database: str, *, schema: str):
+        def list_managed_databases(self):
+            return [_db("db_1", "dlt")]
+
+        def list_managed_tables(self, database, *, schema):
             return []
 
         def close(self) -> None:
             return None
 
-    client = HotdataClient(
-        api_key="k",
-        workspace_id="ws_1",
-        api_base_url="https://api.hotdata.dev",
-        max_retries=1,
-        retry_backoff_seconds=0.0,
-    )
-    client._runtime = FakeRuntime()
-
+    client = _client(FakeRuntime())
     rows = client.fetch_table_rows(database="dlt", schema="public", table="orders")
     assert rows == []
     client.close()
@@ -324,7 +310,7 @@ class _EvoRuntime:
     """Fake runtime tracking the managed-database lifecycle calls."""
 
     def __init__(self, existing_db, existing_tables) -> None:
-        self._existing_db = existing_db  # SimpleNamespace(id=...) or None
+        self._existing_db = existing_db  # _db(...) or None
         self._existing_tables = list(existing_tables)
         self.created: list[tuple] = []
         self.deleted: list[str] = []
@@ -332,12 +318,10 @@ def __init__(self, existing_db, existing_tables) -> None:
         self.uploaded: list[str] = []
         self.loaded: list[tuple] = []
 
-    def resolve_managed_database(self, name):
-        if self._existing_db is None:
-            raise KeyError(name)
-        return self._existing_db
+    def list_managed_databases(self):
+        return [self._existing_db] if self._existing_db is not None else []
 
-    def list_managed_tables(self, name, *, schema):
+    def list_managed_tables(self, database, *, schema):
         return [SimpleNamespace(table=t) for t in self._existing_tables]
 
     def add_managed_table(self, database, table, *, schema, key=None):
@@ -347,7 +331,7 @@ def add_managed_table(self, database, table, *, schema, key=None):
 
     def create_managed_database(self, *, description, schema, tables, keys=None):
         self.created.append((description, schema, list(tables)))
-        return SimpleNamespace(id="new_db")
+        return _db("new_db", description)
 
     def delete_managed_database(self, db_id):
         self.deleted.append(db_id)
@@ -364,21 +348,9 @@ def close(self):
         return None
 
 
-def _client_with(runtime) -> HotdataClient:
-    client = HotdataClient(
-        api_key="k",
-        workspace_id="ws",
-        api_base_url="https://api.hotdata.dev",
-        max_retries=1,
-        retry_backoff_seconds=0.0,
-    )
-    client._runtime = runtime
-    return client
-
-
 def test_ensure_creates_when_missing() -> None:
     rt = _EvoRuntime(existing_db=None, existing_tables=[])
-    client = _client_with(rt)
+    client = _client(rt)
     client.ensure_managed_database("db", schema="public", tables=["orders"], create_if_missing=True)
     assert rt.created == [("db", "public", ["orders"])]
     assert rt.deleted == []
@@ -387,7 +359,7 @@ def test_ensure_creates_when_missing() -> None:
 
 def test_ensure_raises_when_missing_and_no_create() -> None:
     rt = _EvoRuntime(existing_db=None, existing_tables=[])
-    client = _client_with(rt)
+    client = _client(rt)
     with pytest.raises(KeyError):
         client.ensure_managed_database(
             "db", schema="public", tables=["orders"], create_if_missing=False
@@ -396,27 +368,26 @@ def test_ensure_raises_when_missing_and_no_create() -> None:
 
 
 def test_ensure_noop_when_all_tables_present() -> None:
-    rt = _EvoRuntime(
-        existing_db=SimpleNamespace(id="db_1"), existing_tables=["orders", "customers"]
-    )
-    client = _client_with(rt)
+    rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=["orders", "customers"])
+    client = _client(rt)
     client.ensure_managed_database("db", schema="public", tables=["orders"], create_if_missing=True)
     assert rt.deleted == []
     assert rt.created == []
+    assert rt.added == []
     client.close()
 
 
 def test_ensure_adds_missing_table_without_recreate() -> None:
-    rt = _EvoRuntime(existing_db=SimpleNamespace(id="db_1"), existing_tables=["orders"])
-    client = _client_with(rt)
+    rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=["orders"])
+    client = _client(rt)
 
     client.ensure_managed_database(
         "db", schema="public", tables=["orders", "customers"], create_if_missing=True
     )
 
-    # The missing table is declared in place; the database is never deleted or
-    # recreated and no data is moved.
-    assert rt.added == [("db", "customers", "public")]
+    # The missing table is declared in place, addressed by the resolved id; the
+    # database is never deleted or recreated and no data is moved.
+    assert rt.added == [("db_1", "customers", "public")]
     assert rt.deleted == []
     assert rt.created == []
     assert rt.uploaded == []
@@ -424,9 +395,25 @@ def test_ensure_adds_missing_table_without_recreate() -> None:
     client.close()
 
 
-def test_drop_managed_database_deletes_when_present() -> None:
-    rt = _EvoRuntime(existing_db=SimpleNamespace(id="db_1"), existing_tables=[])
-    client = _client_with(rt)
+def test_ensure_raises_on_ambiguous_name() -> None:
+    class _Ambiguous(_EvoRuntime):
+        def list_managed_databases(self):
+            return [_db("db_1", "db"), _db("db_2", "db")]
+
+    rt = _Ambiguous(existing_db=_db("db_1", "db"), existing_tables=[])
+    client = _client(rt)
+    with pytest.raises(HotdataTerminalError, match="ambiguous"):
+        client.ensure_managed_database(
+            "db", schema="public", tables=["orders"], create_if_missing=True
+        )
+    # never created a duplicate on the ambiguity
+    assert rt.created == []
+    client.close()
+
+
+def test_drop_managed_database_deletes_by_id_when_present() -> None:
+    rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=[])
+    client = _client(rt)
     client.drop_managed_database("db")
     assert rt.deleted == ["db_1"]
     client.close()
@@ -434,7 +421,18 @@ def test_drop_managed_database_deletes_when_present() -> None:
 
 def test_drop_managed_database_noop_when_absent() -> None:
     rt = _EvoRuntime(existing_db=None, existing_tables=[])
-    client = _client_with(rt)
+    client = _client(rt)
     client.drop_managed_database("db")
     assert rt.deleted == []
     client.close()
+
+
+def test_drop_clears_run_cache() -> None:
+    rt = _EvoRuntime(existing_db=_db("db_1", "db"), existing_tables=[])
+    cache = SimpleNamespace()
+    client = _client(rt, cache=cache)
+    client.resolve_managed_database("db")
+    assert cache._hotdata_resolved_db.id == "db_1"
+    client.drop_managed_database("db")
+    assert cache._hotdata_resolved_db is None
+    client.close()
diff --git a/tests/test_e2e_inmemory.py b/tests/test_e2e_inmemory.py
index 4825ba3..6f41949 100644
--- a/tests/test_e2e_inmemory.py
+++ b/tests/test_e2e_inmemory.py
@@ -55,6 +55,12 @@ def resolve_managed_database(self, name_or_id):
             raise KeyError(name)
         return SimpleNamespace(id=self.name_to_id[name], default_connection_id="conn")
 
+    def list_managed_databases(self):
+        return [
+            SimpleNamespace(id=db_id, description=name, default_connection_id="conn")
+            for name, db_id in self.name_to_id.items()
+        ]
+
     def list_managed_tables(self, database, *, schema=None):
         name = self.id_to_name.get(database, database)
         return [
@@ -70,7 +76,7 @@ def create_managed_database(self, *, description, schema, tables, keys=None, exp
         self.declared[description] = set(tables)
         for t in tables:
             self.keys[(description, schema, t)] = list((keys or {}).get(t, []))
-        return SimpleNamespace(id=db_id, default_connection_id="conn")
+        return SimpleNamespace(id=db_id, description=description, default_connection_id="conn")
 
     def delete_managed_database(self, name_or_id):
         name = self.id_to_name.get(name_or_id, name_or_id)
diff --git a/tests/test_job_client.py b/tests/test_job_client.py
index 4044be4..92b3860 100644
--- a/tests/test_job_client.py
+++ b/tests/test_job_client.py
@@ -31,6 +31,9 @@ class FakeApi:
         def __init__(self, **_kwargs: object) -> None:
             self._pending: pa.Table | None = None
 
+        def bind_run_cache(self, cache: object) -> None:
+            return None
+
         def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing):
             return SimpleNamespace(id="db_1")
 
@@ -196,6 +199,9 @@ class RecordingApi:
         def __init__(self, **_kwargs: object) -> None:
             self._pending = None
 
+        def bind_run_cache(self, cache: object) -> None:
+            return None
+
         def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing):
             calls["keys"] = keys
             return SimpleNamespace(id="db_1")
@@ -487,6 +493,9 @@ class MissingApi:
         def __init__(self, **_kwargs: object) -> None:
             pass
 
+        def bind_run_cache(self, cache: object) -> None:
+            return None
+
         def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing):
             raise KeyError(name)