From 5eafb9bc97261ea63123e134420358a9e2a7690a Mon Sep 17 00:00:00 2001
From: Rohan Dsouza <rohan@hotdata.dev>
Date: Thu, 23 Jul 2026 20:42:56 +0530
Subject: [PATCH 1/6] feat: let create-scoped API keys bootstrap a managed
 database (#55)

When an API key may create/upload/query but is forbidden from reading
/databases (403), ensure_managed_database now creates the database (the
operation the key is permitted to make) instead of failing, caches the
returned record for the run, and hands that record to subsequent
load/add/list operations so no further read is attempted.

Builds on the resolve-once cache from #39. Requires hotdata-framework 0.9.0
(managed-table ops accept a resolved ManagedDatabase and skip the read
probe). The pin bump is intentionally deferred until 0.9.0 is published;
until then the object-passing is exercised via fakes modelling that
contract.
---
 CHANGELOG.md                                  |   4 +
 src/hotdata_dlt_destination/hotdata_client.py |  56 +++++++---
 tests/test_client.py                          | 104 ++++++++++++++++--
 tests/test_e2e_inmemory.py                    |   4 +
 4 files changed, 145 insertions(+), 23 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 07d76ad..e74ba6e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Create/upload/query-scoped API keys can now bootstrap a managed database. When the key is forbidden from reading `/databases` (403), `ensure_managed_database` creates the database (the operation the key *is* permitted to make) instead of failing, caches the returned record for the run, and hands that record to subsequent load/add/list operations so no further read is attempted. **Requires `hotdata-framework>=0.9.0`** (managed-table ops accept a resolved `ManagedDatabase` and skip the read probe — hotdata-dev/sdk-python-framework#52).
+
 ### 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.
diff --git a/src/hotdata_dlt_destination/hotdata_client.py b/src/hotdata_dlt_destination/hotdata_client.py
index 38e0ffe..0561f3d 100644
--- a/src/hotdata_dlt_destination/hotdata_client.py
+++ b/src/hotdata_dlt_destination/hotdata_client.py
@@ -8,6 +8,11 @@
 from hotdata_dlt_destination.errors import HotdataTerminalError
 
 
+def _is_forbidden(exc: Exception) -> bool:
+    """True when ``exc`` wraps a 403 (create-scoped keys can't read /databases)."""
+    return getattr(getattr(exc, "__cause__", None), "status", None) == 403
+
+
 class HotdataClient(ManagedDatabaseClient):
     """Managed-database client used by the dlt destination.
 
@@ -98,29 +103,45 @@ def ensure_managed_database(
         except KeyError:
             if not create_if_missing:
                 raise
-            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
+            return self._create_and_cache(name, schema=schema, tables=tables, keys=keys)
+        except HotdataTerminalError as exc:
+            # A create/upload/query-scoped key is forbidden from reading /databases,
+            # so it can't check existence; attempt the create it is permitted to make.
+            if not (create_if_missing and _is_forbidden(exc)):
+                raise
+            return self._create_and_cache(name, schema=schema, tables=tables, keys=keys)
 
         existing = {
             managed_table.table
             for managed_table in self._request_with_retry(
-                lambda: self._runtime.list_managed_tables(db.id, schema=schema)
+                lambda: self._runtime.list_managed_tables(db, 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(db.id, table, schema=schema, key=keys.get(table))
+            self._add_managed_table(db, table, schema=schema, key=keys.get(table))
+        return db
+
+    def _create_and_cache(
+        self, name: str, *, schema: str, tables: list[str], keys: dict[str, list[str]]
+    ) -> ManagedDatabase:
+        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
 
     def _add_managed_table(
-        self, database: str, table: str, *, schema: str, key: list[str] | None = None
+        self,
+        database: str | ManagedDatabase,
+        table: str,
+        *,
+        schema: str,
+        key: list[str] | None = None,
     ) -> None:
         self._request_with_retry(
             lambda: self._runtime.add_managed_table(database, table, schema=schema, key=key)
@@ -132,7 +153,7 @@ def drop_managed_database(self, name: str) -> None:
             db = self._resolve(name)
         except KeyError:
             return
-        self._request_with_retry(lambda: self._runtime.delete_managed_database(db.id))
+        self._request_with_retry(lambda: self._runtime.delete_managed_database(db))
         self._cache_db(None)
 
     def resolve_managed_database(self, name: str) -> ManagedDatabase:
@@ -144,9 +165,12 @@ def resolve_managed_database(self, name: str) -> ManagedDatabase:
         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."""
+        """Load parquet into a managed table via the resolved database record.
+
+        Passing the resolved ``ManagedDatabase`` (not its id) lets a create-scoped
+        key load without a further read probe (framework passthrough)."""
         db = self._resolve(database)
-        return super().load_managed_table(db.id, table, **kwargs)
+        return super().load_managed_table(db, 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.
@@ -173,13 +197,13 @@ def list_managed_tables(self, database: str, *, schema: str) -> list:
         """List the managed tables in ``database``/``schema`` (used by ``has_dataset``)."""
         db = self._resolve(database)
         return self._request_with_retry(
-            lambda: self._runtime.list_managed_tables(db.id, schema=schema)
+            lambda: self._runtime.list_managed_tables(db, 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)
+            lambda: self._runtime.list_managed_tables(db, schema=schema)
         ):
             if managed_table.table == table:
                 return managed_table.synced
@@ -199,7 +223,7 @@ def operation() -> pa.Table | None:
         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):
+        for managed_table in self._runtime.list_managed_tables(db, schema=schema):
             if managed_table.table == table:
                 return managed_table.synced
         return False
diff --git a/tests/test_client.py b/tests/test_client.py
index b9ba6a7..96f70e7 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -93,7 +93,8 @@ def list_managed_databases(self):
             return [_db("db_1", "dlt")]
 
         def list_managed_tables(self, database, *, schema):
-            assert database == "db_1"  # addressed by id after the first resolve
+            # addressed by the resolved record after the first resolve
+            assert getattr(database, "id", database) == "db_1"
             return []
 
         def close(self):
@@ -134,13 +135,14 @@ def upload_parquet(self, path: str) -> str:
         def load_managed_table(
             self, database, table, *, schema, upload_id, mode="replace", key=None
         ):
-            self.load_database = database
+            db_id = getattr(database, "id", database)
+            self.load_database = db_id
             return SimpleNamespace(
                 connection_id="conn_1",
                 schema_name=schema,
                 table_name=table,
                 row_count=1,
-                full_name=f"{database}.{schema}.{table}",
+                full_name=f"{db_id}.{schema}.{table}",
             )
 
         def close(self) -> None:
@@ -232,7 +234,7 @@ def list_managed_databases(self):
             return [_db("db_42", "dlt")]
 
         def list_managed_tables(self, database, *, schema):
-            assert database == "db_42"
+            assert getattr(database, "id", database) == "db_42"
             return [SimpleNamespace(table="orders", synced=True)]
 
         def close(self):
@@ -274,7 +276,7 @@ def list_managed_databases(self):
             return [_db("db_1", "dlt")]
 
         def list_managed_tables(self, database, *, schema):
-            assert database == "db_1"
+            assert getattr(database, "id", database) == "db_1"
             return [SimpleNamespace(table="orders", synced=False)]
 
         def close(self) -> None:
@@ -325,7 +327,7 @@ 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):
-        self.added.append((database, table, schema))
+        self.added.append((getattr(database, "id", database), table, schema))
         self._existing_tables.append(table)
         return SimpleNamespace(table=table, schema=schema)
 
@@ -334,7 +336,7 @@ def create_managed_database(self, *, description, schema, tables, keys=None):
         return _db("new_db", description)
 
     def delete_managed_database(self, db_id):
-        self.deleted.append(db_id)
+        self.deleted.append(getattr(db_id, "id", db_id))
 
     def upload_parquet(self, path):
         self.uploaded.append(path)
@@ -436,3 +438,91 @@ def test_drop_clears_run_cache() -> None:
     client.drop_managed_database("db")
     assert cache._hotdata_resolved_db is None
     client.close()
+
+
+# --- #55: create/upload/query-scoped keys (forbidden reads) can bootstrap ---
+# NOTE: passing the resolved ManagedDatabase into the ops (skipping the read
+# probe) requires hotdata-framework >= 0.9.0; the pin bump is the release-gated
+# final step. These tests model that passthrough via the fake runtime.
+
+from hotdata.exceptions import ApiException, ForbiddenException  # noqa: E402
+
+
+class _CreateScopedRuntime:
+    """A create/upload/query-scoped key: reads are forbidden, create/load succeed."""
+
+    def __init__(self) -> None:
+        self.created: list[str] = []
+        self.loaded: list[str] = []
+        self.list_calls = 0
+
+    def list_managed_databases(self):
+        self.list_calls += 1
+        raise ForbiddenException(status=403, reason="ACCESS_DENIED")
+
+    def create_managed_database(self, *, description, schema, tables, keys=None):
+        self.created.append(description)
+        return _db("db_new", description)
+
+    def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None):
+        self.loaded.append(getattr(database, "id", database))
+        return SimpleNamespace(full_name=f"{getattr(database, 'id', database)}.{schema}.{table}")
+
+    def close(self):
+        return None
+
+
+def test_create_scoped_key_bootstraps_on_forbidden_read() -> None:
+    rt = _CreateScopedRuntime()
+    cache = SimpleNamespace()
+    client = _client(rt, cache=cache)
+    db = client.ensure_managed_database(
+        "dlt", schema="public", tables=["orders"], create_if_missing=True
+    )
+    # the forbidden read did not abort — the create the key *is* allowed to make ran
+    assert rt.created == ["dlt"]
+    assert db.id == "db_new"
+    assert cache._hotdata_resolved_db.id == "db_new"  # cached for the run
+    client.close()
+
+
+def test_create_scoped_load_reuses_cache_without_further_read() -> None:
+    rt = _CreateScopedRuntime()
+    cache = SimpleNamespace()
+    client = _client(rt, cache=cache)
+    client.ensure_managed_database(
+        "dlt", schema="public", tables=["orders"], create_if_missing=True
+    )
+    reads_after_create = rt.list_calls
+    # a load in the same run resolves from cache and hands the record straight through
+    client.load_managed_table("dlt", "orders", schema="public", upload_id="u1")
+    assert rt.list_calls == reads_after_create  # no further forbidden read
+    assert rt.loaded == ["db_new"]  # addressed by the created record
+    client.close()
+
+
+def test_forbidden_read_still_raises_when_create_disabled() -> None:
+    rt = _CreateScopedRuntime()
+    client = _client(rt)
+    with pytest.raises(HotdataTerminalError):
+        client.ensure_managed_database(
+            "dlt", schema="public", tables=["orders"], create_if_missing=False
+        )
+    assert rt.created == []
+    client.close()
+
+
+def test_non_forbidden_terminal_error_does_not_trigger_create() -> None:
+    class _BadRuntime(_CreateScopedRuntime):
+        def list_managed_databases(self):
+            self.list_calls += 1
+            raise ApiException(status=400, reason="bad request")
+
+    rt = _BadRuntime()
+    client = _client(rt)
+    with pytest.raises(HotdataTerminalError):
+        client.ensure_managed_database(
+            "dlt", schema="public", tables=["orders"], create_if_missing=True
+        )
+    assert rt.created == []  # a non-403 error is a real failure, not "create it"
+    client.close()
diff --git a/tests/test_e2e_inmemory.py b/tests/test_e2e_inmemory.py
index 6f41949..d7b222e 100644
--- a/tests/test_e2e_inmemory.py
+++ b/tests/test_e2e_inmemory.py
@@ -62,6 +62,7 @@ def list_managed_databases(self):
         ]
 
     def list_managed_tables(self, database, *, schema=None):
+        database = getattr(database, "id", database)  # accept a resolved ManagedDatabase or id/name
         name = self.id_to_name.get(database, database)
         return [
             SimpleNamespace(table=t, var_schema="public", synced=(name, "public", t) in self.tables)
@@ -79,6 +80,7 @@ def create_managed_database(self, *, description, schema, tables, keys=None, exp
         return SimpleNamespace(id=db_id, description=description, default_connection_id="conn")
 
     def delete_managed_database(self, name_or_id):
+        name_or_id = getattr(name_or_id, "id", name_or_id)
         name = self.id_to_name.get(name_or_id, name_or_id)
         db_id = self.name_to_id.pop(name, None)
         self.id_to_name.pop(db_id, None)
@@ -87,6 +89,7 @@ def delete_managed_database(self, name_or_id):
             del self.tables[key]
 
     def add_managed_table(self, database, table, *, schema, key=None):
+        database = getattr(database, "id", database)
         name = self.id_to_name.get(database, database)
         self.declared.setdefault(name, set()).add(table)
         self.keys[(name, schema, table)] = list(key or [])
@@ -101,6 +104,7 @@ def upload_parquet(self, path):
     def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None):
         import pyarrow as _pa
 
+        database = getattr(database, "id", database)
         name = self.id_to_name.get(database, database)
         k = (name, schema, table)
         self.load_counts[k] = self.load_counts.get(k, 0) + 1

From 88655592c33bc172429e49f8782bd636ac00dfe9 Mon Sep 17 00:00:00 2001
From: Rohan Dsouza <rohan@hotdata.dev>
Date: Fri, 24 Jul 2026 01:10:58 +0530
Subject: [PATCH 2/6] chore: require hotdata-framework>=0.9.0 for the
 ManagedDatabase passthrough (#55)

0.9.0 is released on PyPI; bump the pin so create-scoped keys can pass the
resolved ManagedDatabase into load/add/list ops and skip the read probe.
Verified against real 0.9.0: _as_managed_database returns the object as-is and
the full suite is green.
---
 pyproject.toml | 2 +-
 uv.lock        | 8 ++++----
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index 602b50d..3e3fce3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -19,7 +19,7 @@ dependencies = [
     # decls (hotdata 0.8.0) and the per-load `key` on load_managed_table plus
     # create_managed_database(keys=) / add_managed_table(key=) (framework 0.8.0).
     "hotdata>=0.8.0,<0.9",
-    "hotdata-framework>=0.8.0,<0.9",
+    "hotdata-framework>=0.9.0,<0.10",
     "pandas>=2.0",
     "pyarrow>=14",
 ]
diff --git a/uv.lock b/uv.lock
index 9829894..77bba47 100644
--- a/uv.lock
+++ b/uv.lock
@@ -350,7 +350,7 @@ dev = [
 requires-dist = [
     { name = "dlt", specifier = ">=1.28.1,<1.29" },
     { name = "hotdata", specifier = ">=0.8.0,<0.9" },
-    { name = "hotdata-framework", specifier = ">=0.8.0,<0.9" },
+    { name = "hotdata-framework", specifier = ">=0.9.0,<0.10" },
     { name = "hotdata-ibis", marker = "extra == 'ibis'", specifier = ">=0.3.1,<0.4" },
     { name = "ibis-framework", marker = "extra == 'ibis'", specifier = ">=12,<13" },
     { name = "pandas", specifier = ">=2.0" },
@@ -369,16 +369,16 @@ dev = [
 
 [[package]]
 name = "hotdata-framework"
-version = "0.8.0"
+version = "0.9.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "hotdata" },
     { name = "pandas" },
     { name = "pyarrow" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/48/24/879cbc596520023d1757eef8fd73b482404fb98e98d48a19571e8932bec2/hotdata_framework-0.8.0.tar.gz", hash = "sha256:057ebe8818298f1e8b4ef4250a4f4266514c7b51f3a4eb79425a1548885348ce", size = 100984, upload-time = "2026-07-20T07:04:50.175Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/05/95/2e8970ba68bf0001d92d0044766c0e13cb09e7378aa003c637ad571786b0/hotdata_framework-0.9.0.tar.gz", hash = "sha256:a6497ef9fc41f104c89d5636e958a1feaf1c697d98d01436329afeb75f773583", size = 102244, upload-time = "2026-07-23T19:32:56.546Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/98/cd/c5078d826a2240e6d7c6ef5f558280772ee189991af15d0332fd11c5a67c/hotdata_framework-0.8.0-py3-none-any.whl", hash = "sha256:6ba899e9015b71945bf56df873c0486382d4f11763a57afc423338bd68abb30a", size = 16330, upload-time = "2026-07-20T07:04:49.089Z" },
+    { url = "https://files.pythonhosted.org/packages/aa/52/d9bf9001d48262f84a712a3f07c07492964a4481d902117b8986b0dada9e/hotdata_framework-0.9.0-py3-none-any.whl", hash = "sha256:32e7a541823365b5d0e741c9209683406e3851de0434012631a7a7b16b029c1b", size = 16550, upload-time = "2026-07-23T19:32:55.064Z" },
 ]
 
 [[package]]

From 6c50b389877dd79e5714fc3e1eaeee4cd7d88e62 Mon Sep 17 00:00:00 2001
From: Rohan Dsouza <rohan@hotdata.dev>
Date: Fri, 24 Jul 2026 18:01:32 +0530
Subject: [PATCH 3/6] feat!: address managed databases by id only, never by
 name

Hotdata database names are not unique -- the `name` field is a display label
(description), not an identifier. The destination previously listed databases
and matched on the name for every operation, so a collision could silently
read from, write to, or drop the wrong database. Resolve strictly by id
instead:

- Bind an existing database by id via `GET /databases/{id}` (no listing, no
  name scan); every op addresses it by the resolved `ManagedDatabase` record.
- With no `database_id` and `create_database_if_missing`, create the database
  (labelled `database_name`) and log its new id so it can be pinned for
  subsequent runs -- without a pinned id, each run creates a new database.
- Create/upload/query-scoped keys bootstrap for free: the create path issues
  no read, so a key forbidden from reading `/databases` only makes the create
  it is permitted to make.

Adds `database_id` (param / `HOTDATA_DATABASE_ID` / `[destination.hotdata]`),
keeping `database_name` as a create-only label. Removes the by-name resolution
and collision guard entirely. Supersedes the interim collision-safe path (#39)
and reworks the #55 create-scoped bootstrap. Requires hotdata-framework>=0.9.0.

BREAKING CHANGE: managed databases are addressed by id, not name. To load into
the same database across runs, pin its `database_id` (printed on first-run
create). Bumps to 0.11.0.
---
 .env.example                                  |   4 +
 CHANGELOG.md                                  |   5 +-
 README.md                                     |  12 +-
 docs/runbook.md                               |   5 +-
 docs/sql-client-spec.md                       |  22 +-
 pyproject.toml                                |   2 +-
 scripts/load_test.py                          |  45 +-
 scripts/roundtrip_demo.py                     |   9 +-
 src/hotdata_dlt_destination/cli.py            |   1 +
 src/hotdata_dlt_destination/config.py         |   2 +
 src/hotdata_dlt_destination/configuration.py  |   7 +-
 src/hotdata_dlt_destination/factory.py        |   2 +
 src/hotdata_dlt_destination/hotdata_client.py | 287 +++++----
 src/hotdata_dlt_destination/ibis_backend.py   |   2 +-
 src/hotdata_dlt_destination/job_client.py     |  12 +-
 src/hotdata_dlt_destination/pipelines/demo.py |   7 +
 .../pipelines/ibis_demo.py                    |   6 +
 .../pipelines/merge_demo.py                   |   6 +
 src/hotdata_dlt_destination/sql_client.py     |   4 +-
 tests/test_client.py                          | 599 +++++++-----------
 tests/test_e2e_inmemory.py                    |  75 ++-
 tests/test_factory.py                         |  14 +
 tests/test_job_client.py                      |  20 +-
 tests/test_sql_client.py                      |  12 +-
 uv.lock                                       |   2 +-
 25 files changed, 596 insertions(+), 566 deletions(-)

diff --git a/.env.example b/.env.example
index a789212..1544ee0 100644
--- a/.env.example
+++ b/.env.example
@@ -1,4 +1,8 @@
 HOTDATA_API_KEY=
+# Existing managed database to load into, addressed by id. Printed on first-run
+# create; set it here to reuse the same database on subsequent runs.
+HOTDATA_DATABASE_ID=
+# Display label used only when creating a new managed database (not a lookup key).
 HOTDATA_DATABASE=dlt
 HOTDATA_SCHEMA=public
 HOTDATA_WRITE_DISPOSITION=append
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e74ba6e..3f9aea3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,16 +9,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Create/upload/query-scoped API keys can now bootstrap a managed database. When the key is forbidden from reading `/databases` (403), `ensure_managed_database` creates the database (the operation the key *is* permitted to make) instead of failing, caches the returned record for the run, and hands that record to subsequent load/add/list operations so no further read is attempted. **Requires `hotdata-framework>=0.9.0`** (managed-table ops accept a resolved `ManagedDatabase` and skip the read probe — hotdata-dev/sdk-python-framework#52).
+- `database_id` param (`hotdata(database_id=...)`) / `HOTDATA_DATABASE_ID` env / `[destination.hotdata] database_id` config — target an existing managed database by id. On a first run with no id, the database is created by its `database_name` label and the **new id is logged** so it can be pinned (`created managed database <id> … set database_id=<id> …`) to reuse the same database on subsequent runs.
 
 ### Changed
 
+- **Breaking:** managed databases are now addressed strictly **by id**, never by name. Hotdata database names are not unique (the `name` field is a display label / description, not an identifier), so the destination no longer lists databases and matches on the name — a practice that could silently read from, write to, or **drop** the wrong database on a collision. An existing database is bound by id via `GET /databases/{id}`; with no id and `create_database_if_missing`, one is created (labelled `database_name`) and addressed by its returned id for the run. Consequence: to load into the same database across runs you must pin its `database_id` (printed on first-run create); without it, each run creates a new database. Supersedes the interim by-name collision guard. **Requires `hotdata-framework>=0.9.0`** (managed-table ops accept a resolved `ManagedDatabase` and skip the read probe — hotdata-dev/sdk-python-framework#52).
+- Create/upload/query-scoped API keys (forbidden from reading `/databases`) can bootstrap a managed database as a direct consequence: with no `database_id` the create path issues no read at all, so the key only makes the create it *is* permitted to make, and the returned record is reused (by id) for every subsequent load/add/query in the run.
 - **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/README.md b/README.md
index 5bc43c8..359685a 100644
--- a/README.md
+++ b/README.md
@@ -85,7 +85,14 @@ export HOTDATA_API_KEY=your_api_key
 
 The API key is a secret, so it's read from the environment (or a dlt secrets provider). The workspace ID is routing, not a secret — pass it as the `workspace_id=` parameter (switching workspaces is a one-line change).
 
-That's it. On first run, the `sales` managed database is created automatically and the `orders` table is loaded.
+That's it. On first run, a managed database labelled `sales` is created automatically, the `orders` table is loaded, and the new database **id** is printed:
+
+```
+hotdata: created managed database db_abc123 (name='sales'). Pin it for future runs by
+setting database_id=db_abc123 (HOTDATA_DATABASE_ID / [destination.hotdata] database_id).
+```
+
+Managed databases are addressed by id — Hotdata database names are not unique, so a name can't identify one. To keep loading into the **same** database on later runs, pass that id (via `hotdata(database_id="db_abc123")`, `HOTDATA_DATABASE_ID`, or `[destination.hotdata] database_id` in `.dlt/config.toml`). Without a pinned id, each run creates a new database.
 
 `hotdata` supports nested/child tables, preserves dlt's internal columns (`_dlt_id`, `_dlt_load_id`), and persists schema-version, load, and pipeline-state tables in the managed database so **incremental sources resume correctly across runs**. If an existing managed database is missing a declared table on a later run, the table is added to it in place; existing tables and their data are left untouched.
 
@@ -189,7 +196,8 @@ Where `hotdata` stands against the [dlt destination capability spec](https://dlt
 |-----------|-------------|---------|-------------|
 | `api_key` | `HOTDATA_API_KEY` | required | Your Hotdata API key (a secret; passed via `credentials=` or the env var) |
 | `workspace_id` | — | required | Your Hotdata workspace ID — pass as the `hotdata(workspace_id=...)` param (no env var) |
-| `database_name` | `HOTDATA_DATABASE` | `dlt` | Managed database to load into |
+| `database_id` | `HOTDATA_DATABASE_ID` | — | Id of an existing managed database to load into. This is how a database is targeted — names are not unique, so the **id** is the identifier. Printed on first-run create; pin it to reuse the database on later runs |
+| `database_name` | `HOTDATA_DATABASE` | `dlt` | Display label used **only when creating** a new managed database (never to look one up) |
 | `schema` | `HOTDATA_SCHEMA` | `public` | Schema within the managed database |
 | `write_disposition` | `HOTDATA_WRITE_DISPOSITION` | `append` | Default write mode (see [Write modes](#write-modes)) |
 | `declared_tables` | `HOTDATA_DECLARED_TABLES` | — | All table names the pipeline will write (required for multi-table pipelines — see [Multiple tables](#multiple-tables)) |
diff --git a/docs/runbook.md b/docs/runbook.md
index b257b06..a3ad1bb 100644
--- a/docs/runbook.md
+++ b/docs/runbook.md
@@ -31,7 +31,10 @@
   uv run hotdata-dlt-demo --workspace-id <workspace_id>
   ```
 
-  Requires `HOTDATA_API_KEY` to be set; pass your workspace id as the argument.
+  Requires `HOTDATA_API_KEY` to be set; pass your workspace id as the argument. On
+  the first run the demo creates a managed database and prints its id; pass that id
+  on later runs (`--database-id <id>`) to load into the same database instead of
+  creating a new one (managed databases are addressed by id, not by name).
 
 ## Add a new pipeline
 
diff --git a/docs/sql-client-spec.md b/docs/sql-client-spec.md
index ed3b6a7..beb51c4 100644
--- a/docs/sql-client-spec.md
+++ b/docs/sql-client-spec.md
@@ -132,7 +132,7 @@ scoping (see §8):
 ```python
 def __init__(self, managed_database, schema, capabilities, config):
     super().__init__(
-        database_name=managed_database,     # used only as the execute_sql(database=) scope
+        database_name=managed_database,     # display label only; scoping resolves by id from config
         dataset_name=schema,                # "public" -> drives default.public.<table>
         staging_dataset_name=schema,        # no staging; mirror dataset_name
         capabilities=capabilities,
@@ -148,7 +148,7 @@ def __init__(self, managed_database, schema, capabilities, config):
 | `open_connection() -> HotdataClient` | Construct a `HotdataClient` from config, store on `self._client`, return it. No socket. | dlt opens the client when a dataset is materialized. |
 | `close_connection() -> None` | `self._client.close()`; clear it. | Dataset context exit. |
 | `native_connection -> HotdataClient` (property) | Return `self._client`. | Base `__getattr__` delegation; ibis backend. |
-| `execute_query(query, *args, **kwargs) -> ContextManager[DBApiCursor]` | `@contextmanager`; run `self._client.execute_sql(sql, database=self.database_name)`, wrap the returned `pyarrow.Table` in `HotdataCursor`, `yield` it. Map SDK errors via `@raise_database_error`. | **Every read** — `Relation.to_sql()` → here. |
+| `execute_query(query, *args, **kwargs) -> ContextManager[DBApiCursor]` | `@contextmanager`; run `self._client.execute_sql(sql)` (the database is resolved by id from the bound config), wrap the returned `pyarrow.Table` in `HotdataCursor`, `yield` it. Map SDK errors via `@raise_database_error`. | **Every read** — `Relation.to_sql()` → here. |
 | `execute_sql(query, *args, **kwargs) -> Sequence[Sequence] \| None` | `with self.execute_query(...) as c: return None if c.description is None else c.fetchall()`. | Base helpers, direct SQL. |
 | `begin_transaction() -> ContextManager[DBTransaction]` | No-op: `yield self`. Hotdata/DataFusion has no transactions (`supports_ddl_transactions=False`). | dlt may wrap ops in a txn. |
 | `_make_database_exception(ex) -> Exception` (static) | Map undefined-relation → `DatabaseUndefinedRelation`; transient → transient; else terminal. **Scan the whole `__cause__` chain**, not just `str(ex)`: the SDK's `classify_sdk_error` collapses the `ApiException` to `"400: Bad Request"`, so the engine's descriptive `"table … not found"` only appears deeper in the chain (in the underlying `hotdata.exceptions.BadRequestException`). | `@raise_database_error`. |
@@ -180,7 +180,8 @@ class HotdataJobClient(JobClientBase, WithStateSync, WithSqlClient):   # + WithS
     def sql_client(self) -> HotdataSqlClient:
         if self._sql_client is None:
             self._sql_client = HotdataSqlClient(
-                self.config.database_name, self.config.schema, self.capabilities, self.config
+                self.config.database_id or self.config.database_name,
+                self.config.schema, self.capabilities, self.config
             )
         return self._sql_client
 ```
@@ -238,14 +239,13 @@ A single dlt "location" splits into **two independent mechanisms**:
 
 1. **Table path — goes into the SQL.** `make_qualified_table_name("spans")` →
    `"default"."public"."spans"` because `catalog_name()="default"` and `dataset_name="public"`.
-2. **Database scoping — goes into the request, not the SQL.** The managed database is passed as
-   `execute_sql(sql, database=self.database_name)`. Query scoping is by database **id**
-   (`_query_database_scoped(database_id=...)` → `X-Database-Id` header), so our `HotdataClient.execute_sql`
-   resolves **name → id** first via `resolve_managed_database(name).id` — exactly as `fetch_table`
-   already does. (This is what the CLI's `-d` flag supplies manually as an id; the SqlClient passes the
-   *name* and the id lookup lives in our client.)
-
-**Decision — mirror the write path:** address by the destination's managed `database_name` + the fixed
+2. **Database scoping — goes into the request, not the SQL.** Query scoping is by database **id**
+   (`_query_database_scoped(database_id=...)` → `X-Database-Id` header). `HotdataClient.execute_sql`
+   resolves the run's database **by id** from the bound config (`database_id`, or the record created
+   this run) — never by name — exactly as `fetch_table` does. Managed databases are addressed by id
+   only; names are not unique and are not used to look one up.
+
+**Decision — mirror the write path:** address by the run's managed database **id** + the fixed
 `public` schema, exactly as writes do. Guarantees reads return what writes wrote, zero write-side
 change. (Consequence: dlt's pipeline `dataset_name` is not the addressing key — same as today, which is
 why load output shows `dataset None`.) Whether to make `dataset_name` idiomatic later is a joint
diff --git a/pyproject.toml b/pyproject.toml
index 3e3fce3..1ff2e06 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "hotdata-dlt-destination"
-version = "0.10.0"
+version = "0.11.0"
 description = "dlt destination for loading data into Hotdata managed databases."
 readme = "README.md"
 license = "MIT"
diff --git a/scripts/load_test.py b/scripts/load_test.py
index 6730612..bac3bb8 100644
--- a/scripts/load_test.py
+++ b/scripts/load_test.py
@@ -31,6 +31,7 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
+from types import SimpleNamespace
 
 import pyarrow as pa
 import pyarrow.parquet as pq
@@ -140,7 +141,7 @@ def run_database_load(
     workspace_id: str,
     api_base_url: str,
     do_query: bool,
-) -> list[PhaseResult]:
+) -> tuple[list[PhaseResult], str | None]:
     results: list[PhaseResult] = []
 
     client = HotdataClient(
@@ -150,23 +151,28 @@ def run_database_load(
         max_retries=3,
         retry_backoff_seconds=1.0,
     )
+    # id-first: bind a run config so the created database is cached and every
+    # subsequent op addresses it by id (database_name is only the create label).
+    run_cfg = SimpleNamespace(database_id=None, database_name=db_name)
+    client.bind_run_cache(run_cfg)
+    created_id: str | None = None
 
     try:
         # --- create database ---
         t0 = time.perf_counter()
         try:
-            client.ensure_managed_database(
-                db_name,
+            db = client.ensure_managed_database(
                 schema="public",
                 tables=["events"],
                 create_if_missing=True,
             )
+            created_id = db.id
             results.append(PhaseResult(db_name, "create_db", time.perf_counter() - t0))
         except Exception as exc:
             results.append(
                 PhaseResult(db_name, "create_db", time.perf_counter() - t0, error=str(exc))
             )
-            return results
+            return results, created_id
 
         # --- generate + write parquet ---
         table = generate_table(rows)
@@ -182,7 +188,7 @@ def run_database_load(
             results.append(
                 PhaseResult(db_name, "write_parquet", time.perf_counter() - t0, error=str(exc))
             )
-            return results
+            return results, created_id
 
         # --- upload parquet ---
         t0 = time.perf_counter()
@@ -191,7 +197,7 @@ def run_database_load(
             results.append(PhaseResult(db_name, "upload", time.perf_counter() - t0, rows=rows))
         except Exception as exc:
             results.append(PhaseResult(db_name, "upload", time.perf_counter() - t0, error=str(exc)))
-            return results
+            return results, created_id
         finally:
             if parquet_path:
                 with contextlib.suppress(OSError):
@@ -200,17 +206,17 @@ def run_database_load(
         # --- load managed table ---
         t0 = time.perf_counter()
         try:
-            client.load_managed_table(db_name, "events", schema="public", upload_id=upload_id)
+            client.load_managed_table("events", schema="public", upload_id=upload_id)
             results.append(PhaseResult(db_name, "load", time.perf_counter() - t0, rows=rows))
         except Exception as exc:
             results.append(PhaseResult(db_name, "load", time.perf_counter() - t0, error=str(exc)))
-            return results
+            return results, created_id
 
         # --- query via Arrow IPC ---
         if do_query:
             t0 = time.perf_counter()
             try:
-                result_table = client.fetch_table(database=db_name, schema="public", table="events")
+                result_table = client.fetch_table(schema="public", table="events")
                 n = len(result_table) if result_table is not None else 0
                 results.append(PhaseResult(db_name, "query", time.perf_counter() - t0, rows=n))
             except Exception as exc:
@@ -221,7 +227,7 @@ def run_database_load(
     finally:
         client.close()
 
-    return results
+    return results, created_id
 
 
 # ---------------------------------------------------------------------------
@@ -230,7 +236,7 @@ def run_database_load(
 
 
 def delete_databases(
-    db_names: list[str],
+    database_ids: list[str],
     *,
     api_key: str,
     workspace_id: str,
@@ -240,13 +246,13 @@ def delete_databases(
 
     client = RuntimeClient(api_key, workspace_id, host=api_base_url.rstrip("/"))
     try:
-        for name in db_names:
+        for db_id in database_ids:
             try:
-                db = client.resolve_managed_database(name)
-                client.delete_managed_database(db.id)
-                print(f"  deleted {name}")
+                # addressed by id (GET /databases/{id}) — no by-name lookup
+                client.delete_managed_database(db_id)
+                print(f"  deleted {db_id}")
             except Exception as exc:
-                print(f"  could not delete {name}: {exc}")
+                print(f"  could not delete {db_id}: {exc}")
     finally:
         client.close()
 
@@ -307,11 +313,14 @@ def main() -> None:
         }
 
         completed = 0
+        created_ids: list[str] = []
         for future in as_completed(futures):
             name = futures[future]
             completed += 1
             try:
-                phase_results = future.result()
+                phase_results, created_id = future.result()
+                if created_id:
+                    created_ids.append(created_id)
             except Exception as exc:
                 print(f"  [{completed:>3}/{args.databases}] {name}  FATAL: {exc}")
                 continue
@@ -339,7 +348,7 @@ def main() -> None:
     if not args.no_cleanup:
         print("Cleaning up databases...")
         delete_databases(
-            db_names,
+            created_ids,
             api_key=api_key,
             workspace_id=workspace_id,
             api_base_url=api_base_url,
diff --git a/scripts/roundtrip_demo.py b/scripts/roundtrip_demo.py
index 8fdb8a4..ab9f757 100644
--- a/scripts/roundtrip_demo.py
+++ b/scripts/roundtrip_demo.py
@@ -103,12 +103,19 @@ def read_without_dlt(workspace_id: str) -> None:
 def main() -> None:
     parser = argparse.ArgumentParser(description="dlt <-> Hotdata round trip demo.")
     parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id")
-    workspace_id = parser.parse_args().workspace_id
+    parser.add_argument(
+        "--database-id",
+        default=None,
+        help="Existing managed database id (omit to create a new one by name)",
+    )
+    args = parser.parse_args()
+    workspace_id = args.workspace_id
     pipeline = dlt.pipeline(
         pipeline_name="roundtrip_demo",
         destination=hotdata(
             credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]),
             workspace_id=workspace_id,
+            database_id=args.database_id,
             database_name=DATABASE,
             declared_tables=["spans"],
             create_database_if_missing=True,
diff --git a/src/hotdata_dlt_destination/cli.py b/src/hotdata_dlt_destination/cli.py
index 90fab95..1d7e316 100644
--- a/src/hotdata_dlt_destination/cli.py
+++ b/src/hotdata_dlt_destination/cli.py
@@ -7,6 +7,7 @@ def main() -> None:
     config = HotdataDestinationConfig.from_env()
     print("hotdata-dlt-destination is configured")
     print(f"api_base_url={config.api_base_url}")
+    print(f"database_id={config.database_id or '<unset — will create by name>'}")
     print(f"database_name={config.database_name}")
     print(f"schema={config.schema}")
     print(f"write_disposition={config.write_disposition}")
diff --git a/src/hotdata_dlt_destination/config.py b/src/hotdata_dlt_destination/config.py
index e5a9e5b..badfd9b 100644
--- a/src/hotdata_dlt_destination/config.py
+++ b/src/hotdata_dlt_destination/config.py
@@ -28,6 +28,7 @@ def _parse_backoff(value: str) -> float:
 class HotdataDestinationConfig:
     api_key: str
     database_name: str
+    database_id: str | None = None
     api_base_url: str = "https://api.hotdata.dev"
     schema: str = "public"
     write_disposition: str = "append"
@@ -43,6 +44,7 @@ def from_env(cls) -> HotdataDestinationConfig:
         return cls(
             api_key=os.environ["HOTDATA_API_KEY"],
             database_name=os.environ.get("HOTDATA_DATABASE", "dlt"),
+            database_id=os.environ.get("HOTDATA_DATABASE_ID") or None,
             api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"),
             schema=os.environ.get("HOTDATA_SCHEMA", "public"),
             write_disposition=os.environ.get("HOTDATA_WRITE_DISPOSITION", "append"),
diff --git a/src/hotdata_dlt_destination/configuration.py b/src/hotdata_dlt_destination/configuration.py
index 0ec2327..a66ca4f 100644
--- a/src/hotdata_dlt_destination/configuration.py
+++ b/src/hotdata_dlt_destination/configuration.py
@@ -32,8 +32,13 @@ class HotdataClientConfiguration(DestinationClientConfiguration):
     workspace_id: str | None = None
     """Hotdata workspace ID. Pass as a ``hotdata(workspace_id=...)`` param."""
     api_base_url: str = "https://api.hotdata.dev"
+    database_id: str | None = None
+    """Id of the managed database to load into. This is how an existing database
+    is targeted — Hotdata database names are not unique, so the id (not the name)
+    is the identifier. Printed on first-run create; pin it to reuse the database."""
     database_name: str = "dlt"
-    """Name of the managed database to load into."""
+    """Display label for the managed database, used only when creating a new one
+    (never to look one up). Pin ``database_id`` to reuse an existing database."""
     schema: str = "public"
     """Schema within the managed database."""
     write_disposition: str = "append"
diff --git a/src/hotdata_dlt_destination/factory.py b/src/hotdata_dlt_destination/factory.py
index 3bbcde2..3f3de82 100644
--- a/src/hotdata_dlt_destination/factory.py
+++ b/src/hotdata_dlt_destination/factory.py
@@ -88,6 +88,7 @@ def __init__(
         self,
         credentials: HotdataCredentials | dict[str, t.Any] | str | None = None,
         workspace_id: str | None = None,
+        database_id: str | None = None,
         database_name: str = None,
         schema: str = None,
         write_disposition: str = None,
@@ -123,6 +124,7 @@ def __init__(
         super().__init__(
             credentials=credentials,
             workspace_id=workspace_id,
+            database_id=database_id,
             database_name=database_name,
             schema=schema,
             write_disposition=write_disposition,
diff --git a/src/hotdata_dlt_destination/hotdata_client.py b/src/hotdata_dlt_destination/hotdata_client.py
index 0561f3d..11e0512 100644
--- a/src/hotdata_dlt_destination/hotdata_client.py
+++ b/src/hotdata_dlt_destination/hotdata_client.py
@@ -1,94 +1,121 @@
 from __future__ import annotations
 
 import pyarrow as pa
+from dlt.common import logger
+from hotdata.api.databases_api import DatabasesApi
 from hotdata.arrow import ResultsApi as ArrowResultsApi
-from hotdata_framework.databases import ManagedDatabase
+from hotdata_framework.databases import ManagedDatabase, managed_database_from_detail
 from hotdata_framework.managed_client import ManagedDatabaseClient
 
 from hotdata_dlt_destination.errors import HotdataTerminalError
 
 
-def _is_forbidden(exc: Exception) -> bool:
-    """True when ``exc`` wraps a 403 (create-scoped keys can't read /databases)."""
-    return getattr(getattr(exc, "__cause__", None), "status", None) == 403
+def _is_not_found(exc: Exception) -> bool:
+    """True when ``exc`` wraps a 404 (the bound database id does not exist)."""
+    return getattr(getattr(exc, "__cause__", None), "status", None) == 404
 
 
 class HotdataClient(ManagedDatabaseClient):
     """Managed-database client used by the dlt destination.
 
-    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.
+    Addressing is **id-first**: a Hotdata database is identified by its id, never
+    by name. The name (``description``) is only a display label supplied when a
+    database is created — it is never used to look one up, because Hotdata names
+    are not unique. Concretely:
+
+    * **Bind an existing database by id.** When a ``database_id`` is configured,
+      the record is fetched once via ``GET /databases/{id}`` and every operation
+      addresses the database by that record. No listing, no name scan.
+    * **Create on first run.** With no ``database_id`` and ``create_if_missing``,
+      the database is created (labelled with ``database_name``) and its new id is
+      logged so it can be pinned via ``database_id`` for subsequent runs.
+    * **Cross-run schema evolution.** When binding an existing database, tables it
+      is missing are 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.
+
+    The resolved record is cached once per run (via :meth:`bind_run_cache`) so the
+    whole run reuses a single bind/create and create-scoped keys never issue a read
+    they are not permitted to make.
     """
 
-    # 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
+    # Run-scoped configuration bound via bind_run_cache(); the resolved database
+    # and its provenance are cached on it so the whole run reuses one record.
+    _config: object | None = None
 
     def bind_run_cache(self, cache: object) -> None:
-        """Bind a run-scoped store so a database resolves to its record once.
+        """Bind the run's shared configuration.
 
-        ``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.
+        ``cache`` is the shared ``HotdataClientConfiguration`` instance every
+        client built for a run points at. Its ``database_id`` / ``database_name``
+        drive id-first resolution, and the resolved record is cached back on it
+        (``_hotdata_db``) so a single run resolves the database exactly once.
         """
-        self._run_cache = cache
+        self._config = cache
 
-    # --- resolution -------------------------------------------------------
+    # --- run cache --------------------------------------------------------
 
-    def _collision_safe_resolve(self, name_or_id: str) -> ManagedDatabase:
-        """Resolve a name/id to its record, raising on an ambiguous name.
+    def _cached_db(self) -> ManagedDatabase | None:
+        return getattr(self._config, "_hotdata_db", None) if self._config is not None else None
 
-        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.
+    def _cache_db(self, db: ManagedDatabase | None, *, created: bool) -> None:
+        if self._config is not None:
+            self._config._hotdata_db = db
+            self._config._hotdata_db_created = created
+
+    def _was_created(self) -> bool:
+        return bool(getattr(self._config, "_hotdata_db_created", False))
+
+    # --- resolution (id-first, never by name) -----------------------------
+
+    def _bind_by_id(self, database_id: str) -> ManagedDatabase:
+        """Fetch a database record by id (``GET /databases/{id}``).
+
+        Raises ``KeyError`` when the id does not exist (404); other API errors
+        propagate as ``HotdataTerminalError``/``HotdataTransientError``.
         """
-        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."
+        try:
+            detail = self._request_with_retry(
+                lambda: DatabasesApi(self._runtime.api).get_database(database_id)
             )
-        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
+        except HotdataTerminalError as exc:
+            if _is_not_found(exc):
+                raise KeyError(database_id) from exc
+            raise
+        return managed_database_from_detail(detail)
+
+    def _configured_database_id(self) -> str | None:
+        return getattr(self._config, "database_id", None)
+
+    def _require_db(self) -> ManagedDatabase:
+        """Resolve the run's database for an operation (id-first).
 
-    def _cache_db(self, db: ManagedDatabase | None) -> None:
-        if self._run_cache is not None:
-            self._run_cache._hotdata_resolved_db = db
+        Reuses the run cache, else binds the configured ``database_id``. Raises
+        ``KeyError`` when no database has been resolved and none is configured —
+        there is deliberately no by-name fallback.
+        """
+        db = self._cached_db()
+        if db is not None:
+            return db
+        database_id = self._configured_database_id()
+        if database_id:
+            db = self._bind_by_id(database_id)
+            self._cache_db(db, created=False)
+            return db
+        raise KeyError("no managed database resolved for this run (set database_id)")
+
+    def resolved_database_id(self) -> str:
+        """Return the id of the run's managed database (bound by id or created).
+
+        Raises ``KeyError`` when none is configured and none was created this run.
+        """
+        return self._require_db().id
 
     # --- lifecycle --------------------------------------------------------
 
     def ensure_managed_database(
         self,
-        name: str,
         *,
         schema: str,
         tables: list[str],
@@ -98,19 +125,55 @@ def ensure_managed_database(
         # keys: table name -> key columns (enables delete/update/upsert on it)
         keys = keys or {}
 
-        try:
-            db = self._resolve(name)
-        except KeyError:
-            if not create_if_missing:
-                raise
-            return self._create_and_cache(name, schema=schema, tables=tables, keys=keys)
-        except HotdataTerminalError as exc:
-            # A create/upload/query-scoped key is forbidden from reading /databases,
-            # so it can't check existence; attempt the create it is permitted to make.
-            if not (create_if_missing and _is_forbidden(exc)):
-                raise
-            return self._create_and_cache(name, schema=schema, tables=tables, keys=keys)
+        db = self._cached_db()
+        created = self._was_created()
+        if db is None:
+            database_id = self._configured_database_id()
+            if database_id:
+                db = self._bind_by_id(database_id)
+                created = False
+                self._cache_db(db, created=False)
+            elif create_if_missing:
+                db = self._create(schema=schema, tables=tables, keys=keys)
+                created = True
+                self._cache_db(db, created=True)
+            else:
+                raise KeyError("no managed database resolved for this run (set database_id)")
+
+        # A freshly created database already declared every table; only an
+        # existing (bound) database needs additive, in-place schema evolution.
+        if not created:
+            self._reconcile_tables(db, schema=schema, tables=tables, keys=keys)
+        return db
+
+    def _create(
+        self, *, schema: str, tables: list[str], keys: dict[str, list[str]]
+    ) -> ManagedDatabase:
+        description = getattr(self._config, "database_name", None)
+        db = self._request_with_retry(
+            lambda: self._runtime.create_managed_database(
+                description=description, schema=schema, tables=sorted(set(tables)), keys=keys
+            )
+        )
+        # Logged at WARNING (dlt's default level) so the new id is always visible:
+        # without pinning it via database_id, the next run creates another database.
+        logger.warning(
+            "hotdata: created managed database %s (name=%r). Pin it for future runs by "
+            "setting database_id=%s (HOTDATA_DATABASE_ID / [destination.hotdata] database_id).",
+            db.id,
+            description,
+            db.id,
+        )
+        return db
 
+    def _reconcile_tables(
+        self,
+        db: ManagedDatabase,
+        *,
+        schema: str,
+        tables: list[str],
+        keys: dict[str, list[str]],
+    ) -> None:
         existing = {
             managed_table.table
             for managed_table in self._request_with_retry(
@@ -122,22 +185,10 @@ def ensure_managed_database(
         # load job runs, so by load time this is normally a no-op.
         for table in sorted(set(tables) - existing):
             self._add_managed_table(db, table, schema=schema, key=keys.get(table))
-        return db
-
-    def _create_and_cache(
-        self, name: str, *, schema: str, tables: list[str], keys: dict[str, list[str]]
-    ) -> ManagedDatabase:
-        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
 
     def _add_managed_table(
         self,
-        database: str | ManagedDatabase,
+        database: ManagedDatabase,
         table: str,
         *,
         schema: str,
@@ -147,41 +198,40 @@ def _add_managed_table(
             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)."""
-        try:
-            db = self._resolve(name)
-        except KeyError:
-            return
+    def drop_managed_database(self) -> None:
+        """Delete the run's managed database if it exists (used for dlt dev_mode / refresh)."""
+        db = self._cached_db()
+        if db is None:
+            database_id = self._configured_database_id()
+            if not database_id:
+                return
+            try:
+                db = self._bind_by_id(database_id)
+            except KeyError:
+                return
         self._request_with_retry(lambda: self._runtime.delete_managed_database(db))
-        self._cache_db(None)
+        self._cache_db(None, created=False)
 
-    def resolve_managed_database(self, name: str) -> ManagedDatabase:
-        """Resolve a managed database by display name (or id) to its record.
+    # --- operations (addressed by the resolved record) --------------------
 
-        Raises ``KeyError`` when nothing matches and ``HotdataTerminalError`` when
-        the name is shared by more than one database.
-        """
-        return self._resolve(name)
-
-    def load_managed_table(self, database: str, table: str, **kwargs):
+    def load_managed_table(self, table: str, **kwargs):
         """Load parquet into a managed table via the resolved database record.
 
         Passing the resolved ``ManagedDatabase`` (not its id) lets a create-scoped
         key load without a further read probe (framework passthrough)."""
-        db = self._resolve(database)
+        db = self._require_db()
         return super().load_managed_table(db, 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.
+    def execute_sql(self, sql: str) -> pa.Table:
+        """Run a SQL query scoped to the run's database and return Arrow.
 
-        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.
+        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._resolve(database)
+            db = self._require_db()
             result_id = self._query_database_scoped(sql, database_id=db.id)
             if result_id is None:
                 return pa.table({})
@@ -193,25 +243,16 @@ def operation() -> pa.Table:
 
         return self._request_with_retry(operation)
 
-    def list_managed_tables(self, database: str, *, schema: str) -> list:
-        """List the managed tables in ``database``/``schema`` (used by ``has_dataset``)."""
-        db = self._resolve(database)
+    def list_managed_tables(self, *, schema: str) -> list:
+        """List the managed tables in the run's database/``schema`` (used by ``has_dataset``)."""
+        db = self._require_db()
         return self._request_with_retry(
             lambda: self._runtime.list_managed_tables(db, 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, 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 fetch_table(self, *, schema: str, table: str) -> pa.Table | None:
         def operation() -> pa.Table | None:
-            db = self._resolve(database)
+            db = self._require_db()
             if not self._table_is_synced_for(db, table, schema=schema):
                 return None
             sql = f'SELECT * FROM "default"."{schema}"."{table}"'
@@ -222,6 +263,10 @@ def operation() -> pa.Table | None:
 
         return self._request_with_retry(operation)
 
+    def fetch_table_rows(self, *, schema: str, table: str) -> list[dict]:
+        result = self.fetch_table(schema=schema, table=table)
+        return result.to_pylist() if result is not None else []
+
     def _table_is_synced_for(self, db: ManagedDatabase, table: str, *, schema: str) -> bool:
         for managed_table in self._runtime.list_managed_tables(db, schema=schema):
             if managed_table.table == table:
diff --git a/src/hotdata_dlt_destination/ibis_backend.py b/src/hotdata_dlt_destination/ibis_backend.py
index 9b73f8e..4565bb6 100644
--- a/src/hotdata_dlt_destination/ibis_backend.py
+++ b/src/hotdata_dlt_destination/ibis_backend.py
@@ -35,7 +35,7 @@ def ibis_connect(client: JobClientBase) -> Any:
     config = client.config
 
     with _hotdata_api(config) as api:
-        database_id = api.resolve_managed_database(config.database_name).id
+        database_id = api.resolved_database_id()
 
     return ibis.hotdata.connect(
         api_url=config.api_base_url,
diff --git a/src/hotdata_dlt_destination/job_client.py b/src/hotdata_dlt_destination/job_client.py
index 2a1347c..640631d 100644
--- a/src/hotdata_dlt_destination/job_client.py
+++ b/src/hotdata_dlt_destination/job_client.py
@@ -145,7 +145,6 @@ def _truncate_table(
         write_table_parquet(arrow_schema.empty_table(), parquet_path)
         upload_id = api.upload_parquet(parquet_path)
         api.load_managed_table(
-            config.database_name,
             table_name,
             schema=config.schema,
             upload_id=upload_id,
@@ -167,7 +166,6 @@ def _upload_table(
         write_table_parquet(arrow_table, parquet_path)
         upload_id = api.upload_parquet(parquet_path)
         api.load_managed_table(
-            config.database_name,
             table_name,
             schema=config.schema,
             upload_id=upload_id,
@@ -237,7 +235,6 @@ def _apply(api: HotdataClient, upload_table, mode: str) -> None:
                 write_table_parquet(upload_table, parquet_path)
                 upload_id = api.upload_parquet(parquet_path)
                 api.load_managed_table(
-                    contract.database_name,
                     contract.table_name,
                     schema=contract.schema,
                     upload_id=upload_id,
@@ -254,7 +251,6 @@ def _combine_and_replace(
             api: HotdataClient, combine_disposition: str, *, apply_hard_delete: bool = False
         ) -> None:
             existing = api.fetch_table(
-                database=contract.database_name,
                 schema=contract.schema,
                 table=contract.table_name,
             )
@@ -271,7 +267,6 @@ def _combine_and_replace(
         try:
             with _hotdata_api(self._config) as api:
                 api.ensure_managed_database(
-                    contract.database_name,
                     schema=contract.schema,
                     tables=_declared_tables(
                         contract=contract,
@@ -364,7 +359,7 @@ def sql_client(self) -> HotdataSqlClient:
 
         if self._sql_client is None:
             self._sql_client = HotdataSqlClient(
-                self.config.database_name,
+                self.config.database_id or self.config.database_name,
                 self.config.schema,
                 self.capabilities,
                 self.config,
@@ -398,7 +393,6 @@ def initialize_storage(self, truncate_tables: Iterable[str] = None) -> None:
         try:
             with _hotdata_api(self.config) as api:
                 api.ensure_managed_database(
-                    self.config.database_name,
                     schema=self.config.schema,
                     tables=all_tables,
                     keys=keys,
@@ -437,7 +431,6 @@ def is_storage_initialized(self) -> bool:
         with _hotdata_api(self.config) as api:
             try:
                 api.ensure_managed_database(
-                    self.config.database_name,
                     schema=self.config.schema,
                     tables=[],
                     create_if_missing=False,
@@ -452,7 +445,7 @@ def drop_storage(self) -> None:
         # recreates it on the next run.
         try:
             with _hotdata_api(self.config) as api:
-                api.drop_managed_database(self.config.database_name)
+                api.drop_managed_database()
         except HotdataTerminalError as exc:
             raise DestinationTerminalException(str(exc)) from exc
 
@@ -549,7 +542,6 @@ def _fetch_internal_rows(self, table_name: str) -> list[dict[str, Any]]:
         with _hotdata_api(self.config) as api:
             try:
                 table = api.fetch_table(
-                    database=self.config.database_name,
                     schema=self.config.schema,
                     table=table_name,
                 )
diff --git a/src/hotdata_dlt_destination/pipelines/demo.py b/src/hotdata_dlt_destination/pipelines/demo.py
index b4fbf60..9b1c767 100644
--- a/src/hotdata_dlt_destination/pipelines/demo.py
+++ b/src/hotdata_dlt_destination/pipelines/demo.py
@@ -90,6 +90,12 @@ def macro_wide_resource():
 def main() -> None:
     parser = argparse.ArgumentParser(description="Load FRED macro indicators into Hotdata.")
     parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id")
+    parser.add_argument(
+        "--database-id",
+        default=None,
+        help="Existing managed database id to load into (printed on first-run create; "
+        "omit to create a new database by name)",
+    )
     args = parser.parse_args()
     all_tables = ["macro_indicators_raw", "macro_wide"]
 
@@ -98,6 +104,7 @@ def main() -> None:
         destination=hotdata(
             credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]),
             workspace_id=args.workspace_id,
+            database_id=args.database_id,
             api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"),
             write_disposition="replace",
             declared_tables=all_tables,
diff --git a/src/hotdata_dlt_destination/pipelines/ibis_demo.py b/src/hotdata_dlt_destination/pipelines/ibis_demo.py
index 17fc0d8..eea8d49 100644
--- a/src/hotdata_dlt_destination/pipelines/ibis_demo.py
+++ b/src/hotdata_dlt_destination/pipelines/ibis_demo.py
@@ -84,12 +84,18 @@ def _read_with_ibis(pipeline: dlt.Pipeline) -> None:
 def main() -> None:
     parser = argparse.ArgumentParser(description="Load NYC taxi trips and read via ibis.")
     parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id")
+    parser.add_argument(
+        "--database-id",
+        default=None,
+        help="Existing managed database id (omit to create a new one by name)",
+    )
     args = parser.parse_args()
     pipeline = dlt.pipeline(
         pipeline_name="ibis_demo",
         destination=hotdata(
             credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]),
             workspace_id=args.workspace_id,
+            database_id=args.database_id,
             api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"),
             write_disposition="replace",
             declared_tables=["trips"],
diff --git a/src/hotdata_dlt_destination/pipelines/merge_demo.py b/src/hotdata_dlt_destination/pipelines/merge_demo.py
index 04373cd..0ce0f42 100644
--- a/src/hotdata_dlt_destination/pipelines/merge_demo.py
+++ b/src/hotdata_dlt_destination/pipelines/merge_demo.py
@@ -84,12 +84,18 @@ def _print_table(pipeline: dlt.Pipeline, label: str) -> None:
 def main() -> None:
     parser = argparse.ArgumentParser(description="Composite-key merge + hard_delete demo.")
     parser.add_argument("--workspace-id", required=True, help="Hotdata workspace id")
+    parser.add_argument(
+        "--database-id",
+        default=None,
+        help="Existing managed database id (omit to create a new one by name)",
+    )
     args = parser.parse_args()
     pipeline = dlt.pipeline(
         pipeline_name="orders_merge",
         destination=hotdata(
             credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]),
             workspace_id=args.workspace_id,
+            database_id=args.database_id,
             api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"),
             declared_tables=["orders"],
             database_name=DATABASE,
diff --git a/src/hotdata_dlt_destination/sql_client.py b/src/hotdata_dlt_destination/sql_client.py
index 39edde0..1bec649 100644
--- a/src/hotdata_dlt_destination/sql_client.py
+++ b/src/hotdata_dlt_destination/sql_client.py
@@ -200,7 +200,7 @@ def execute_query(self, query: AnyStr, *args: Any, **kwargs: Any) -> Iterator[DB
             raise DatabaseTerminalException(
                 NotImplementedError("HotdataSqlClient does not support parameterized queries")
             )
-        table = self._client.execute_sql(query, database=self.database_name)
+        table = self._client.execute_sql(query)
         yield HotdataCursor(table)
 
     def execute_sql(
@@ -246,7 +246,7 @@ def has_dataset(self) -> bool:
         # Check via the managed-DB API rather than the base's INFORMATION_SCHEMA.SCHEMATA
         # query (which binds %s params our query API can't take).
         try:
-            self._client.list_managed_tables(self.database_name, schema=self.dataset_name)
+            self._client.list_managed_tables(schema=self.dataset_name)
             return True
         except (KeyError, HotdataTerminalError, HotdataTransientError):
             return False
diff --git a/tests/test_client.py b/tests/test_client.py
index 96f70e7..86fc544 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,16 +1,21 @@
 from types import SimpleNamespace
 
 import pytest
+from hotdata.exceptions import ApiException, ForbiddenException
 
-from hotdata_dlt_destination.errors import HotdataTerminalError
 from hotdata_dlt_destination.hotdata_client import HotdataClient
 
 
-def _db(db_id: str, name: str, conn: str = "conn") -> SimpleNamespace:
+def _db(db_id: str, name: str = "dlt", conn: str = "conn") -> SimpleNamespace:
     return SimpleNamespace(id=db_id, description=name, default_connection_id=conn)
 
 
-def _client(runtime, *, cache=None) -> HotdataClient:
+def _cfg(*, database_id=None, database_name="dlt") -> SimpleNamespace:
+    # Stand-in for the shared HotdataClientConfiguration the run binds.
+    return SimpleNamespace(database_id=database_id, database_name=database_name)
+
+
+def _client(runtime, *, config=None) -> HotdataClient:
     client = HotdataClient(
         api_key="k",
         workspace_id="ws",
@@ -19,159 +24,211 @@ def _client(runtime, *, cache=None) -> HotdataClient:
         retry_backoff_seconds=0.0,
     )
     client._runtime = runtime
-    if cache is not None:
-        client.bind_run_cache(cache)
+    if config is not None:
+        client.bind_run_cache(config)
     return client
 
 
-# --- resolution: collision-safe, id-addressed, resolve-once ---------------
+def _install_get_database(monkeypatch, registry: dict) -> None:
+    """Patch the id lookup (GET /databases/{id}); unknown ids 404."""
 
+    class FakeDatabasesApi:
+        def __init__(self, api):
+            pass
 
-def test_resolve_returns_single_match_by_name() -> None:
-    class FakeRuntime:
-        def list_managed_databases(self):
-            return [_db("db_1", "dlt")]
+        def get_database(self, database_id):
+            if database_id not in registry:
+                raise ApiException(status=404, reason="not found")
+            return registry[database_id]
 
-        def close(self):
-            return None
+    monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi)
+    monkeypatch.setattr(
+        "hotdata_dlt_destination.hotdata_client.managed_database_from_detail", lambda d: d
+    )
 
-    client = _client(FakeRuntime())
-    assert client.resolve_managed_database("dlt").id == "db_1"
-    client.close()
 
+class _Runtime:
+    """Fake runtime tracking managed-database lifecycle calls (no name lookups)."""
 
-def test_resolve_by_id() -> None:
-    class FakeRuntime:
-        def list_managed_databases(self):
-            return [_db("db_1", "dlt")]
+    api = None
 
-        def close(self):
-            return None
+    def __init__(self, existing_tables=()) -> None:
+        self._existing_tables = list(existing_tables)
+        self.created: list[tuple] = []
+        self.deleted: list[str] = []
+        self.added: list[tuple] = []
+        self.uploaded: list[str] = []
+        self.loaded: list[tuple] = []
+        self.table_lists = 0
 
-    client = _client(FakeRuntime())
-    assert client.resolve_managed_database("db_1").id == "db_1"
-    client.close()
+    def list_managed_tables(self, database, *, schema):
+        self.table_lists += 1
+        return [SimpleNamespace(table=t, synced=True) for t in self._existing_tables]
 
+    def add_managed_table(self, database, table, *, schema, key=None):
+        self.added.append((getattr(database, "id", database), table, schema))
+        self._existing_tables.append(table)
+        return SimpleNamespace(table=table, schema=schema)
 
-def test_resolve_missing_raises_keyerror() -> None:
-    class FakeRuntime:
-        def list_managed_databases(self):
-            return []
+    def create_managed_database(self, *, description, schema, tables, keys=None):
+        self.created.append((description, schema, list(tables)))
+        return _db("new_db", description)
 
-        def close(self):
-            return None
+    def delete_managed_database(self, db):
+        self.deleted.append(getattr(db, "id", db))
 
-    client = _client(FakeRuntime())
-    with pytest.raises(KeyError):
-        client.resolve_managed_database("dlt")
+    def upload_parquet(self, path):
+        self.uploaded.append(path)
+        return "up_1"
+
+    def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None):
+        db_id = getattr(database, "id", database)
+        self.loaded.append((db_id, table, schema, upload_id))
+        return SimpleNamespace(
+            connection_id="c",
+            schema_name=schema,
+            table_name=table,
+            row_count=1,
+            full_name=f"{db_id}.{schema}.{table}",
+        )
+
+    def close(self):
+        return None
+
+
+# --- create on first run (no database_id) ---------------------------------
+
+
+def test_create_when_no_id_configured() -> None:
+    rt = _Runtime()
+    client = _client(rt, config=_cfg(database_id=None, database_name="sales"))
+    db = client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True)
+    # created, labelled by database_name; no read/list happened (nothing to reconcile)
+    assert rt.created == [("sales", "public", ["orders"])]
+    assert db.id == "new_db"
+    assert rt.table_lists == 0
     client.close()
 
 
-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")]
+def test_create_logs_new_id(monkeypatch) -> None:
+    messages: list[str] = []
+    monkeypatch.setattr(
+        "hotdata_dlt_destination.hotdata_client.logger.warning",
+        lambda msg, *args: messages.append(msg % args if args else msg),
+    )
+    rt = _Runtime()
+    client = _client(rt, config=_cfg(database_name="sales"))
+    client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True)
+    # the id is surfaced for the user to pin via database_id
+    assert any("new_db" in m and "database_id" in m for m in messages)
+    client.close()
 
-        def close(self):
-            return None
 
-    client = _client(FakeRuntime())
-    with pytest.raises(HotdataTerminalError, match="ambiguous"):
-        client.resolve_managed_database("dlt")
+def test_no_id_and_no_create_raises_keyerror() -> None:
+    # is_storage_initialized() path: probe with create disabled and no id -> "not there".
+    rt = _Runtime()
+    client = _client(rt, config=_cfg(database_id=None))
+    with pytest.raises(KeyError):
+        client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=False)
+    assert rt.created == []
     client.close()
 
 
-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
+# --- bind an existing database by id --------------------------------------
 
-        def list_managed_databases(self):
-            self.list_calls += 1
-            return [_db("db_1", "dlt")]
 
-        def list_managed_tables(self, database, *, schema):
-            # addressed by the resolved record after the first resolve
-            assert getattr(database, "id", database) == "db_1"
-            return []
+def test_bind_existing_by_id(monkeypatch) -> None:
+    _install_get_database(monkeypatch, {"db_1": _db("db_1", "sales")})
+    rt = _Runtime(existing_tables=["orders", "customers"])
+    client = _client(rt, config=_cfg(database_id="db_1"))
+    db = client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True)
+    # bound by id; not recreated
+    assert db.id == "db_1"
+    assert rt.created == []
+    client.close()
 
-        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")
+def test_bind_missing_id_raises_keyerror(monkeypatch) -> None:
+    _install_get_database(monkeypatch, {})  # id does not exist -> 404
+    rt = _Runtime()
+    client = _client(rt, config=_cfg(database_id="db_missing"))
+    with pytest.raises(KeyError):
+        client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=False)
+    assert rt.created == []
+    client.close()
 
-    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"
+def test_bind_by_id_evolves_schema_in_place(monkeypatch) -> None:
+    _install_get_database(monkeypatch, {"db_1": _db("db_1", "sales")})
+    rt = _Runtime(existing_tables=["orders"])
+    client = _client(rt, config=_cfg(database_id="db_1"))
+    client.ensure_managed_database(
+        schema="public", tables=["orders", "customers"], create_if_missing=True
+    )
+    # the missing table is declared in place, addressed by the resolved id
+    assert rt.added == [("db_1", "customers", "public")]
+    assert rt.created == []
+    assert rt.deleted == []
     client.close()
-    client2.close()
 
 
-# --- upload / load: addressed by id ---------------------------------------
+# --- resolve-once / run cache (populated by id or create, never by name) --
 
 
-def test_upload_and_load_managed_table_addresses_by_id() -> None:
-    class FakeRuntime:
-        def __init__(self) -> None:
-            self.upload_calls = 0
-            self.load_database = None
+def test_resolves_once_and_reuses_run_cache(monkeypatch) -> None:
+    reads = {"n": 0}
 
-        def list_managed_databases(self):
-            return [_db("db_dlt", "dlt")]
+    class FakeDatabasesApi:
+        def __init__(self, api):
+            pass
 
-        def upload_parquet(self, path: str) -> str:
-            self.upload_calls += 1
-            assert path.endswith(".parquet")
-            return "upload_1"
+        def get_database(self, database_id):
+            reads["n"] += 1
+            return _db("db_1", "sales")
 
-        def load_managed_table(
-            self, database, table, *, schema, upload_id, mode="replace", key=None
-        ):
-            db_id = getattr(database, "id", database)
-            self.load_database = db_id
-            return SimpleNamespace(
-                connection_id="conn_1",
-                schema_name=schema,
-                table_name=table,
-                row_count=1,
-                full_name=f"{db_id}.{schema}.{table}",
-            )
+    monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi)
+    monkeypatch.setattr(
+        "hotdata_dlt_destination.hotdata_client.managed_database_from_detail", lambda d: d
+    )
 
-        def close(self) -> None:
-            return None
+    cfg = _cfg(database_id="db_1")
+    rt = _Runtime(existing_tables=["orders"])
+    client = _client(rt, config=cfg)
+    client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True)
+    # a second client sharing the run config reuses the cached record
+    client2 = _client(_Runtime(existing_tables=["orders"]), config=cfg)
+    assert client2.resolved_database_id() == "db_1"
+    assert reads["n"] == 1  # bound exactly once for the run
+    assert cfg._hotdata_db.id == "db_1"
+    client.close()
+    client2.close()
 
-    rt = FakeRuntime()
-    client = _client(rt)
 
-    upload_id = client.upload_parquet("/tmp/batch.parquet")
-    loaded = client.load_managed_table("dlt", "orders", schema="public", upload_id=upload_id)
+# --- operations address the resolved record -------------------------------
+
 
-    assert upload_id == "upload_1"
-    # the load is addressed by the resolved id, not the display name
-    assert rt.load_database == "db_dlt"
+def test_load_addresses_by_resolved_record(monkeypatch) -> None:
+    _install_get_database(monkeypatch, {"db_dlt": _db("db_dlt", "dlt")})
+    rt = _Runtime()
+    client = _client(rt, config=_cfg(database_id="db_dlt"))
+    upload_id = client.upload_parquet("/tmp/batch.parquet")
+    loaded = client.load_managed_table("orders", schema="public", upload_id=upload_id)
+    assert upload_id == "up_1"
+    assert rt.loaded == [("db_dlt", "orders", "public", "up_1")]
     assert loaded.full_name == "db_dlt.public.orders"
-    assert rt.upload_calls == 1
     client.close()
 
 
-# --- fetch_table / execute_sql: carry the database scope ------------------
+def test_list_managed_tables_by_id(monkeypatch) -> None:
+    _install_get_database(monkeypatch, {"db_1": _db("db_1")})
+    rt = _Runtime(existing_tables=["orders"])
+    client = _client(rt, config=_cfg(database_id="db_1"))
+    tables = client.list_managed_tables(schema="public")
+    assert [t.table for t in tables] == ["orders"]
+    client.close()
 
 
 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.
-    """
     from hotdata.models.query_response import QueryResponse as _QR
 
     scopes: dict[str, list] = {"result": [], "arrow": []}
@@ -219,310 +276,120 @@ def get_result_arrow(self, result_id, *, x_database_id):
     return scopes
 
 
-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, addressing by the resolved id.
-    import pyarrow as pa
-
-    scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1]}), arrow_in_hotdata_client=False)
-
-    class FakeRuntime:
-        api = None
-
-        def list_managed_databases(self):
-            return [_db("db_42", "dlt")]
-
-        def list_managed_tables(self, database, *, schema):
-            assert getattr(database, "id", database) == "db_42"
-            return [SimpleNamespace(table="orders", synced=True)]
-
-        def close(self):
-            return None
-
-    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"]
-    assert scopes["arrow"] == ["db_42"]
-    client.close()
-
-
 def test_execute_sql_carries_database_scope(monkeypatch) -> None:
     import pyarrow as pa
 
+    _install_get_database(monkeypatch, {"db_99": _db("db_99")})
     scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1, 2]}), arrow_in_hotdata_client=True)
 
-    class FakeRuntime:
-        api = None
-
-        def list_managed_databases(self):
-            return [_db("db_99", "dlt")]
-
-        def close(self):
-            return None
-
-    client = _client(FakeRuntime())
-    table = client.execute_sql('SELECT * FROM "default"."public"."spans"', database="dlt")
+    client = _client(_Runtime(), config=_cfg(database_id="db_99"))
+    table = client.execute_sql('SELECT * FROM "default"."public"."spans"')
     assert table.num_rows == 2
     assert scopes["result"] == ["db_99"]
     assert scopes["arrow"] == ["db_99"]
     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 getattr(database, "id", database) == "db_1"
-            return [SimpleNamespace(table="orders", synced=False)]
-
-        def close(self) -> None:
-            return None
+def test_fetch_table_carries_database_scope(monkeypatch) -> None:
+    import pyarrow as pa
 
-    client = _client(FakeRuntime())
-    rows = client.fetch_table_rows(database="dlt", schema="public", table="orders")
-    assert rows == []
+    _install_get_database(monkeypatch, {"db_42": _db("db_42")})
+    scopes = _patch_query_apis(monkeypatch, pa.table({"id": [1]}), arrow_in_hotdata_client=False)
+    rt = _Runtime(existing_tables=["orders"])
+    client = _client(rt, config=_cfg(database_id="db_42"))
+    table = client.fetch_table(schema="public", table="orders")
+    assert table is not None and table.num_rows == 1
+    assert scopes["result"] == ["db_42"]
+    assert scopes["arrow"] == ["db_42"]
     client.close()
 
 
-def test_fetch_table_rows_returns_empty_when_table_missing() -> None:
-    class FakeRuntime:
-        def list_managed_databases(self):
-            return [_db("db_1", "dlt")]
+def test_fetch_table_rows_skips_unsynced_tables(monkeypatch) -> None:
+    _install_get_database(monkeypatch, {"db_1": _db("db_1")})
 
+    class _Unsynced(_Runtime):
         def list_managed_tables(self, database, *, schema):
-            return []
-
-        def close(self) -> None:
-            return None
-
-    client = _client(FakeRuntime())
-    rows = client.fetch_table_rows(database="dlt", schema="public", table="orders")
-    assert rows == []
-    client.close()
-
-
-# --- ensure_managed_database / drop_managed_database (schema evolution) ---
-
-
-class _EvoRuntime:
-    """Fake runtime tracking the managed-database lifecycle calls."""
-
-    def __init__(self, existing_db, existing_tables) -> None:
-        self._existing_db = existing_db  # _db(...) or None
-        self._existing_tables = list(existing_tables)
-        self.created: list[tuple] = []
-        self.deleted: list[str] = []
-        self.added: list[tuple] = []
-        self.uploaded: list[str] = []
-        self.loaded: list[tuple] = []
-
-    def list_managed_databases(self):
-        return [self._existing_db] if self._existing_db is not None else []
-
-    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):
-        self.added.append((getattr(database, "id", database), table, schema))
-        self._existing_tables.append(table)
-        return SimpleNamespace(table=table, schema=schema)
-
-    def create_managed_database(self, *, description, schema, tables, keys=None):
-        self.created.append((description, schema, list(tables)))
-        return _db("new_db", description)
-
-    def delete_managed_database(self, db_id):
-        self.deleted.append(getattr(db_id, "id", db_id))
-
-    def upload_parquet(self, path):
-        self.uploaded.append(path)
-        return "up_1"
-
-    def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None):
-        self.loaded.append((database, table, schema, upload_id))
-        return SimpleNamespace()
-
-    def close(self):
-        return None
-
-
-def test_ensure_creates_when_missing() -> None:
-    rt = _EvoRuntime(existing_db=None, existing_tables=[])
-    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 == []
-    client.close()
-
-
-def test_ensure_raises_when_missing_and_no_create() -> None:
-    rt = _EvoRuntime(existing_db=None, existing_tables=[])
-    client = _client(rt)
-    with pytest.raises(KeyError):
-        client.ensure_managed_database(
-            "db", schema="public", tables=["orders"], create_if_missing=False
-        )
-    client.close()
-
+            return [SimpleNamespace(table="orders", synced=False)]
 
-def test_ensure_noop_when_all_tables_present() -> None:
-    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 = _client(_Unsynced(), config=_cfg(database_id="db_1"))
+    assert client.fetch_table_rows(schema="public", table="orders") == []
     client.close()
 
 
-def test_ensure_adds_missing_table_without_recreate() -> None:
-    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, 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 == []
-    assert rt.loaded == []
-    client.close()
-
+# --- drop -----------------------------------------------------------------
 
-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")
+def test_drop_deletes_by_id(monkeypatch) -> None:
+    _install_get_database(monkeypatch, {"db_1": _db("db_1")})
+    rt = _Runtime()
+    cfg = _cfg(database_id="db_1")
+    client = _client(rt, config=cfg)
+    client.drop_managed_database()
     assert rt.deleted == ["db_1"]
+    assert cfg._hotdata_db is None
     client.close()
 
 
-def test_drop_managed_database_noop_when_absent() -> None:
-    rt = _EvoRuntime(existing_db=None, existing_tables=[])
-    client = _client(rt)
-    client.drop_managed_database("db")
+def test_drop_noop_without_id() -> None:
+    rt = _Runtime()
+    client = _client(rt, config=_cfg(database_id=None))
+    client.drop_managed_database()
     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()
+# --- #55: create/upload/query-scoped keys (forbidden reads) bootstrap ------
+# Under id-first there is no id to look up on a first run, so the create path
+# never issues a read the key isn't allowed to make -- create just succeeds.
 
 
-# --- #55: create/upload/query-scoped keys (forbidden reads) can bootstrap ---
-# NOTE: passing the resolved ManagedDatabase into the ops (skipping the read
-# probe) requires hotdata-framework >= 0.9.0; the pin bump is the release-gated
-# final step. These tests model that passthrough via the fake runtime.
+class _CreateScopedRuntime(_Runtime):
+    """A create-scoped key: any read (get_database / list) is forbidden."""
 
-from hotdata.exceptions import ApiException, ForbiddenException  # noqa: E402
-
-
-class _CreateScopedRuntime:
-    """A create/upload/query-scoped key: reads are forbidden, create/load succeed."""
-
-    def __init__(self) -> None:
-        self.created: list[str] = []
-        self.loaded: list[str] = []
-        self.list_calls = 0
-
-    def list_managed_databases(self):
-        self.list_calls += 1
+    def list_managed_tables(self, database, *, schema):
         raise ForbiddenException(status=403, reason="ACCESS_DENIED")
 
-    def create_managed_database(self, *, description, schema, tables, keys=None):
-        self.created.append(description)
-        return _db("db_new", description)
-
-    def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None):
-        self.loaded.append(getattr(database, "id", database))
-        return SimpleNamespace(full_name=f"{getattr(database, 'id', database)}.{schema}.{table}")
 
-    def close(self):
-        return None
+def test_create_scoped_key_bootstraps_without_read(monkeypatch) -> None:
+    # get_database must never be called on the create path.
+    called = {"get": 0}
 
+    class FakeDatabasesApi:
+        def __init__(self, api):
+            pass
 
-def test_create_scoped_key_bootstraps_on_forbidden_read() -> None:
-    rt = _CreateScopedRuntime()
-    cache = SimpleNamespace()
-    client = _client(rt, cache=cache)
-    db = client.ensure_managed_database(
-        "dlt", schema="public", tables=["orders"], create_if_missing=True
-    )
-    # the forbidden read did not abort — the create the key *is* allowed to make ran
-    assert rt.created == ["dlt"]
-    assert db.id == "db_new"
-    assert cache._hotdata_resolved_db.id == "db_new"  # cached for the run
-    client.close()
+        def get_database(self, database_id):
+            called["get"] += 1
+            raise ForbiddenException(status=403, reason="ACCESS_DENIED")
 
+    monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi)
 
-def test_create_scoped_load_reuses_cache_without_further_read() -> None:
     rt = _CreateScopedRuntime()
-    cache = SimpleNamespace()
-    client = _client(rt, cache=cache)
-    client.ensure_managed_database(
-        "dlt", schema="public", tables=["orders"], create_if_missing=True
-    )
-    reads_after_create = rt.list_calls
-    # a load in the same run resolves from cache and hands the record straight through
-    client.load_managed_table("dlt", "orders", schema="public", upload_id="u1")
-    assert rt.list_calls == reads_after_create  # no further forbidden read
-    assert rt.loaded == ["db_new"]  # addressed by the created record
+    cfg = _cfg(database_id=None, database_name="dlt")
+    client = _client(rt, config=cfg)
+    db = client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True)
+    assert rt.created == [("dlt", "public", ["orders"])]
+    assert db.id == "new_db"
+    assert called["get"] == 0  # never attempted a forbidden read
+    assert cfg._hotdata_db.id == "new_db"
     client.close()
 
 
-def test_forbidden_read_still_raises_when_create_disabled() -> None:
-    rt = _CreateScopedRuntime()
-    client = _client(rt)
-    with pytest.raises(HotdataTerminalError):
-        client.ensure_managed_database(
-            "dlt", schema="public", tables=["orders"], create_if_missing=False
-        )
-    assert rt.created == []
-    client.close()
+def test_create_scoped_load_reuses_cache_without_read(monkeypatch) -> None:
+    class FakeDatabasesApi:
+        def __init__(self, api):
+            pass
 
+        def get_database(self, database_id):
+            raise ForbiddenException(status=403, reason="ACCESS_DENIED")
 
-def test_non_forbidden_terminal_error_does_not_trigger_create() -> None:
-    class _BadRuntime(_CreateScopedRuntime):
-        def list_managed_databases(self):
-            self.list_calls += 1
-            raise ApiException(status=400, reason="bad request")
+    monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", FakeDatabasesApi)
 
-    rt = _BadRuntime()
-    client = _client(rt)
-    with pytest.raises(HotdataTerminalError):
-        client.ensure_managed_database(
-            "dlt", schema="public", tables=["orders"], create_if_missing=True
-        )
-    assert rt.created == []  # a non-403 error is a real failure, not "create it"
+    rt = _CreateScopedRuntime()
+    cfg = _cfg(database_id=None, database_name="dlt")
+    client = _client(rt, config=cfg)
+    client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True)
+    # a load in the same run resolves from cache and hands the record straight through
+    client.load_managed_table("orders", schema="public", upload_id="u1")
+    assert rt.loaded == [("new_db", "orders", "public", "u1")]
     client.close()
diff --git a/tests/test_e2e_inmemory.py b/tests/test_e2e_inmemory.py
index d7b222e..798b22e 100644
--- a/tests/test_e2e_inmemory.py
+++ b/tests/test_e2e_inmemory.py
@@ -19,6 +19,7 @@
 import pyarrow.parquet as pq
 import pytest
 from dlt.common.schema import Schema
+from hotdata.exceptions import ApiException
 from hotdata.models.query_response import QueryResponse
 
 import hotdata_dlt_destination.job_client as jc
@@ -49,17 +50,15 @@ def __init__(self) -> None:
     def close(self) -> None:
         pass
 
-    def resolve_managed_database(self, name_or_id):
-        name = self.id_to_name.get(name_or_id, name_or_id)
-        if name not in self.name_to_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 get_database(self, database_id):
+        # id-first bind: GET /databases/{id}; unknown id 404s.
+        if database_id not in self.id_to_name:
+            raise ApiException(status=404, reason="not found")
+        return SimpleNamespace(
+            id=database_id,
+            description=self.id_to_name[database_id],
+            default_connection_id="conn",
+        )
 
     def list_managed_tables(self, database, *, schema=None):
         database = getattr(database, "id", database)  # accept a resolved ManagedDatabase or id/name
@@ -234,6 +233,16 @@ def __init__(self, **kwargs):
         self._runtime = _ACTIVE["backend"]
 
 
+class _FakeDatabasesApi:
+    """Routes the id-first bind (GET /databases/{id}) to the active backend."""
+
+    def __init__(self, api):
+        pass
+
+    def get_database(self, database_id):
+        return _ACTIVE["backend"].get_database(database_id)
+
+
 @pytest.fixture
 def backend(monkeypatch):
     be = InMemoryBackend()
@@ -241,10 +250,15 @@ def backend(monkeypatch):
     monkeypatch.setattr(mc, "QueryApi", _FakeQueryApi)
     monkeypatch.setattr(mc, "ResultsApi", _FakeResultsApi)
     monkeypatch.setattr(mc, "ArrowResultsApi", _FakeArrowResultsApi)
-    # hotdata_client.py also references ArrowResultsApi (for execute_sql).
+    # hotdata_client.py also references ArrowResultsApi (for execute_sql) and
+    # DatabasesApi (for the id-first bind).
     monkeypatch.setattr(
         "hotdata_dlt_destination.hotdata_client.ArrowResultsApi", _FakeArrowResultsApi
     )
+    monkeypatch.setattr("hotdata_dlt_destination.hotdata_client.DatabasesApi", _FakeDatabasesApi)
+    monkeypatch.setattr(
+        "hotdata_dlt_destination.hotdata_client.managed_database_from_detail", lambda d: d
+    )
     monkeypatch.setattr(jc, "HotdataClient", _E2EClient)
     monkeypatch.setattr(sc, "HotdataClient", _E2EClient)
     yield be
@@ -252,9 +266,19 @@ def backend(monkeypatch):
 
 
 def _dest(database_name, declared_tables, write_disposition="append"):
+    # id-first: a managed database is addressed by id. Model the real workflow —
+    # the database exists (created once) and the pipeline pins its id — by
+    # creating it up front and binding by id. `database_name` stays the label.
+    be = _ACTIVE["backend"]
+    database_id = be.name_to_id.get(database_name)
+    if database_id is None:
+        database_id = be.create_managed_database(
+            description=database_name, schema="public", tables=[]
+        ).id
     return hotdata(
         credentials=HotdataCredentials(api_key="test"),
         workspace_id="ws_test",
+        database_id=database_id,
         database_name=database_name,
         declared_tables=declared_tables,
         write_disposition=write_disposition,
@@ -280,6 +304,32 @@ def orders():
     assert backend.rows("e2e_basic", "_dlt_version") is not None
 
 
+def test_auto_create_when_no_database_id(backend, tmp_path):
+    # First run with no database_id: the pipeline creates the managed database by
+    # its name (label) and addresses it by the returned id for the rest of the run.
+    @dlt.resource(name="orders", write_disposition="replace")
+    def orders():
+        yield [{"id": 1, "amount": 10}]
+
+    dlt.pipeline(
+        pipeline_name="p_autocreate",
+        destination=hotdata(
+            credentials=HotdataCredentials(api_key="test"),
+            workspace_id="ws_test",
+            database_name="e2e_autocreate",
+            declared_tables=["orders"],
+            write_disposition="replace",
+        ),
+        dataset_name="public",
+        pipelines_dir=str(tmp_path),
+    ).run(orders())
+
+    # Exactly one database was created, under the given label.
+    assert set(backend.name_to_id) == {"e2e_autocreate"}
+    rows = backend.rows("e2e_autocreate", "orders")
+    assert rows is not None and len(rows) == 1
+
+
 def test_replace_multi_file_pipeline_keeps_all_rows(backend, tmp_path, monkeypatch):
     # Regression for multi-file replace data loss: with a small
     # DATA_WRITER__FILE_MAX_BYTES (the ingest worker sets 200MB in production),
@@ -375,6 +425,7 @@ def events(updated=dlt.sources.incremental("updated")):  # noqa: B008  (dlt idio
     cfg = HotdataClientConfiguration(
         credentials=HotdataCredentials(api_key="test"),
         workspace_id="ws_test",
+        database_id=backend.name_to_id["e2e_state"],
         database_name="e2e_state",
         schema="public",
     )
diff --git a/tests/test_factory.py b/tests/test_factory.py
index 61859e8..fe04288 100644
--- a/tests/test_factory.py
+++ b/tests/test_factory.py
@@ -71,6 +71,20 @@ def test_workspace_id_is_none_when_not_passed(clean_env) -> None:
     assert cfg.workspace_id is None
 
 
+def test_database_id_param_flows_to_config(clean_env) -> None:
+    # An existing database is targeted by id (names are not unique identifiers).
+    clean_env.setenv("HOTDATA_API_KEY", "sk_env")
+    cfg = _resolve(hotdata(workspace_id="ws", database_id="db_abc", declared_tables=["t"]))
+    assert cfg.database_id == "db_abc"
+
+
+def test_database_id_is_none_when_not_passed(clean_env) -> None:
+    # No id means "create by name on first run"; the id is None until then.
+    clean_env.setenv("HOTDATA_API_KEY", "sk_env")
+    cfg = _resolve(hotdata(workspace_id="ws", database_name="d", declared_tables=["t"]))
+    assert cfg.database_id is None
+
+
 def test_legacy_workspace_in_credentials_dict_is_hoisted_without_mutating(clean_env) -> None:
     creds = {"api_key": "k", "workspace_id": "ws_dict"}
     with pytest.warns(DeprecationWarning):
diff --git a/tests/test_job_client.py b/tests/test_job_client.py
index 92b3860..a834fca 100644
--- a/tests/test_job_client.py
+++ b/tests/test_job_client.py
@@ -34,17 +34,17 @@ def __init__(self, **_kwargs: object) -> None:
         def bind_run_cache(self, cache: object) -> None:
             return None
 
-        def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing):
+        def ensure_managed_database(self, *, schema, tables, keys=None, create_if_missing):
             return SimpleNamespace(id="db_1")
 
-        def fetch_table(self, *, database, schema, table):
+        def fetch_table(self, *, schema, table):
             return store.get(table)
 
         def upload_parquet(self, path: str) -> str:
             self._pending = pq.read_table(path)
             return "upload_1"
 
-        def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None):
+        def load_managed_table(self, table, *, schema, upload_id, mode="replace", key=None):
             assert self._pending is not None
             # Mode-faithful, like the server: append accumulates, replace overwrites.
             existing = store.get(table)
@@ -54,7 +54,7 @@ def load_managed_table(self, database, table, *, schema, upload_id, mode="replac
                 )
             else:
                 store[table] = self._pending
-            return SimpleNamespace(full_name=f"{database}.{schema}.{table}")
+            return SimpleNamespace(full_name=f"db_1.{schema}.{table}")
 
         def close(self) -> None:
             return None
@@ -202,18 +202,18 @@ def __init__(self, **_kwargs: object) -> None:
         def bind_run_cache(self, cache: object) -> None:
             return None
 
-        def ensure_managed_database(self, name, *, schema, tables, keys=None, create_if_missing):
+        def ensure_managed_database(self, *, schema, tables, keys=None, create_if_missing):
             calls["keys"] = keys
             return SimpleNamespace(id="db_1")
 
-        def fetch_table(self, *, database, schema, table):
+        def fetch_table(self, *, schema, table):
             calls["fetches"] = calls.get("fetches", 0) + 1
 
         def upload_parquet(self, path: str) -> str:
             self._pending = pq.read_table(path)
             return "upload_1"
 
-        def load_managed_table(self, database, table, *, schema, upload_id, mode="replace", key=None):
+        def load_managed_table(self, table, *, schema, upload_id, mode="replace", key=None):
             calls.setdefault("modes", []).append(mode)
             calls["mode"] = mode
             calls["load_key"] = key
@@ -225,7 +225,7 @@ def load_managed_table(self, database, table, *, schema, upload_id, mode="replac
             )
             if reject_mode is not None and mode == reject_mode:
                 raise HotdataTerminalError(f"{table}: no declared key; required for mode={mode}")
-            return SimpleNamespace(full_name=f"{database}.{schema}.{table}")
+            return SimpleNamespace(full_name=f"db_1.{schema}.{table}")
 
         def close(self) -> None:
             return None
@@ -496,8 +496,8 @@ def __init__(self, **_kwargs: object) -> None:
         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)
+        def ensure_managed_database(self, *, schema, tables, keys=None, create_if_missing):
+            raise KeyError("db_1")
 
         def close(self) -> None:
             return None
diff --git a/tests/test_sql_client.py b/tests/test_sql_client.py
index 6b7dc4b..e80d1dc 100644
--- a/tests/test_sql_client.py
+++ b/tests/test_sql_client.py
@@ -38,14 +38,14 @@ def __init__(self, table: pa.Table | None = None, tables: object = _MISSING) ->
         self._table = table if table is not None else CANNED
         # tables=None explicitly models a missing database (list raises).
         self._tables = ["spans"] if tables is _MISSING else tables
-        self.calls: list[tuple[str, str]] = []
+        self.calls: list[str] = []
         self.closed = False
 
-    def execute_sql(self, sql: str, *, database: str) -> pa.Table:
-        self.calls.append((sql, database))
+    def execute_sql(self, sql: str) -> pa.Table:
+        self.calls.append(sql)
         return self._table
 
-    def list_managed_tables(self, database: str, *, schema: str):
+    def list_managed_tables(self, *, schema: str):
         if self._tables is None:
             raise HotdataTerminalError("database not found")
         return list(self._tables)
@@ -120,8 +120,8 @@ def test_execute_query_yields_cursor() -> None:
     sc = _client(fake)
     with sc.execute_query('SELECT * FROM "default"."public"."spans"') as cur:
         assert cur.df().shape == (3, 3)
-    # SQL passed through verbatim, scoped by managed database_name.
-    assert fake.calls == [('SELECT * FROM "default"."public"."spans"', "db")]
+    # SQL passed through verbatim; the database scope is resolved by id internally.
+    assert fake.calls == ['SELECT * FROM "default"."public"."spans"']
 
 
 def test_execute_sql_returns_rows() -> None:
diff --git a/uv.lock b/uv.lock
index 77bba47..5cb0085 100644
--- a/uv.lock
+++ b/uv.lock
@@ -321,7 +321,7 @@ wheels = [
 
 [[package]]
 name = "hotdata-dlt-destination"
-version = "0.10.0"
+version = "0.11.0"
 source = { editable = "." }
 dependencies = [
     { name = "dlt" },

From c00fb86140af4a580fbde92e047f7f54e157df0f Mon Sep 17 00:00:00 2001
From: Rohan Dsouza <rohan@hotdata.dev>
Date: Fri, 24 Jul 2026 18:10:53 +0530
Subject: [PATCH 4/6] fix: clear error when a pinned database_id was dropped;
 date 0.11.0 changelog

- ensure_managed_database: on the create path, a configured database_id that
  404s (e.g. dropped by dev_mode/refresh) now raises a clear HotdataTerminalError
  instead of an opaque KeyError -- a server-assigned id can't be recreated, so
  the message tells the user to unset database_id or pin an existing one. The
  probe path (create_if_missing=False) still gets KeyError -> "not initialized".
  Addresses claude[bot]'s review nit on PR #59.
- CHANGELOG: promote the Unreleased notes to a dated [0.11.0] section so the
  release-metadata check matches the version bump.
---
 CHANGELOG.md                                  |  2 ++
 src/hotdata_dlt_destination/hotdata_client.py | 16 +++++++++++++++-
 tests/test_client.py                          | 14 ++++++++++++++
 3 files changed, 31 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3f9aea3..68378f2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.11.0] - 2026-07-24
+
 ### Added
 
 - `database_id` param (`hotdata(database_id=...)`) / `HOTDATA_DATABASE_ID` env / `[destination.hotdata] database_id` config — target an existing managed database by id. On a first run with no id, the database is created by its `database_name` label and the **new id is logged** so it can be pinned (`created managed database <id> … set database_id=<id> …`) to reuse the same database on subsequent runs.
diff --git a/src/hotdata_dlt_destination/hotdata_client.py b/src/hotdata_dlt_destination/hotdata_client.py
index 11e0512..d974d81 100644
--- a/src/hotdata_dlt_destination/hotdata_client.py
+++ b/src/hotdata_dlt_destination/hotdata_client.py
@@ -130,7 +130,21 @@ def ensure_managed_database(
         if db is None:
             database_id = self._configured_database_id()
             if database_id:
-                db = self._bind_by_id(database_id)
+                try:
+                    db = self._bind_by_id(database_id)
+                except KeyError:
+                    # A pinned id can't be recreated (ids are server-assigned), so
+                    # on the create path a missing id is a clear terminal error, not
+                    # a silent recreate. The probe path (create_if_missing=False, e.g.
+                    # is_storage_initialized) still gets the KeyError -> "not there".
+                    if create_if_missing:
+                        raise HotdataTerminalError(
+                            f"configured database_id {database_id!r} was not found "
+                            "(it may have been dropped). A managed database cannot be "
+                            "recreated with the same id -- unset database_id to create a "
+                            "new one, or pin an existing id."
+                        ) from None
+                    raise
                 created = False
                 self._cache_db(db, created=False)
             elif create_if_missing:
diff --git a/tests/test_client.py b/tests/test_client.py
index 86fc544..010c7c4 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -3,6 +3,7 @@
 import pytest
 from hotdata.exceptions import ApiException, ForbiddenException
 
+from hotdata_dlt_destination.errors import HotdataTerminalError
 from hotdata_dlt_destination.hotdata_client import HotdataClient
 
 
@@ -158,6 +159,19 @@ def test_bind_missing_id_raises_keyerror(monkeypatch) -> None:
     client.close()
 
 
+def test_bind_missing_id_with_create_raises_clear_error(monkeypatch) -> None:
+    # dev_mode/refresh drops the pinned db, then a create-path ensure finds the id
+    # gone. It can't be recreated (ids are server-assigned) -> clear terminal error,
+    # not a raw KeyError.
+    _install_get_database(monkeypatch, {})  # pinned id no longer exists -> 404
+    rt = _Runtime()
+    client = _client(rt, config=_cfg(database_id="db_dropped"))
+    with pytest.raises(HotdataTerminalError, match="db_dropped"):
+        client.ensure_managed_database(schema="public", tables=["orders"], create_if_missing=True)
+    assert rt.created == []  # never silently recreated
+    client.close()
+
+
 def test_bind_by_id_evolves_schema_in_place(monkeypatch) -> None:
     _install_get_database(monkeypatch, {"db_1": _db("db_1", "sales")})
     rt = _Runtime(existing_tables=["orders"])

From 55d0c183c081dee4556afe370f07590c0aebac63 Mon Sep 17 00:00:00 2001
From: Rohan Dsouza <rohan@hotdata.dev>
Date: Fri, 24 Jul 2026 22:30:21 +0530
Subject: [PATCH 5/6] fix(demos): provision + pin database_id so read-back
 works under id-first

The read-showcasing demos did run() then read via dataset() in one process.
Under id-first the read has no id to resolve (its config differs from the
write's), so they crashed with "no managed database resolved". Provision the
managed database up front and pin its id for both write and read (use
--database-id to reuse an existing one). This also models the correct id-first
workflow. demo.py (write-only) is unchanged. Demo-only; no src behavior change.
---
 scripts/roundtrip_demo.py                     | 17 +++++++++++++-
 .../pipelines/ibis_demo.py                    | 21 ++++++++++++++++--
 .../pipelines/merge_demo.py                   | 22 +++++++++++++++++--
 3 files changed, 55 insertions(+), 5 deletions(-)

diff --git a/scripts/roundtrip_demo.py b/scripts/roundtrip_demo.py
index ab9f757..07f3fa6 100644
--- a/scripts/roundtrip_demo.py
+++ b/scripts/roundtrip_demo.py
@@ -110,12 +110,27 @@ def main() -> None:
     )
     args = parser.parse_args()
     workspace_id = args.workspace_id
+
+    # id-first: a managed database is addressed by id (names are not identifiers),
+    # so a read-back must know the id. Provision the database once here and pin its
+    # id for both the write and the read; pass --database-id to reuse an existing one.
+    database_id = args.database_id
+    if database_id is None:
+        from hotdata_framework.client import HotdataClient as RuntimeClient
+
+        rc = RuntimeClient(os.environ["HOTDATA_API_KEY"], workspace_id, host=API_BASE_URL.rstrip("/"))
+        database_id = rc.create_managed_database(
+            description=DATABASE, schema="public", tables=["spans"]
+        ).id
+        rc.close()
+        print(f"created managed database {database_id} (pass --database-id {database_id} to reuse)")
+
     pipeline = dlt.pipeline(
         pipeline_name="roundtrip_demo",
         destination=hotdata(
             credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]),
             workspace_id=workspace_id,
-            database_id=args.database_id,
+            database_id=database_id,
             database_name=DATABASE,
             declared_tables=["spans"],
             create_database_if_missing=True,
diff --git a/src/hotdata_dlt_destination/pipelines/ibis_demo.py b/src/hotdata_dlt_destination/pipelines/ibis_demo.py
index eea8d49..9b9ac63 100644
--- a/src/hotdata_dlt_destination/pipelines/ibis_demo.py
+++ b/src/hotdata_dlt_destination/pipelines/ibis_demo.py
@@ -90,13 +90,30 @@ def main() -> None:
         help="Existing managed database id (omit to create a new one by name)",
     )
     args = parser.parse_args()
+    api_base_url = os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev")
+
+    # id-first: the live ibis read binds the managed database by id (names are not
+    # identifiers). Provision it once and pin the id; pass --database-id to reuse.
+    database_id = args.database_id
+    if database_id is None:
+        from hotdata_framework.client import HotdataClient as RuntimeClient
+
+        rc = RuntimeClient(
+            os.environ["HOTDATA_API_KEY"], args.workspace_id, host=api_base_url.rstrip("/")
+        )
+        database_id = rc.create_managed_database(
+            description=DATABASE_NAME, schema=SCHEMA, tables=["trips"]
+        ).id
+        rc.close()
+        print(f"created managed database {database_id} (pass --database-id {database_id} to reuse)")
+
     pipeline = dlt.pipeline(
         pipeline_name="ibis_demo",
         destination=hotdata(
             credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]),
             workspace_id=args.workspace_id,
-            database_id=args.database_id,
-            api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"),
+            database_id=database_id,
+            api_base_url=api_base_url,
             write_disposition="replace",
             declared_tables=["trips"],
             database_name=DATABASE_NAME,
diff --git a/src/hotdata_dlt_destination/pipelines/merge_demo.py b/src/hotdata_dlt_destination/pipelines/merge_demo.py
index 0ce0f42..43ebde4 100644
--- a/src/hotdata_dlt_destination/pipelines/merge_demo.py
+++ b/src/hotdata_dlt_destination/pipelines/merge_demo.py
@@ -90,13 +90,31 @@ def main() -> None:
         help="Existing managed database id (omit to create a new one by name)",
     )
     args = parser.parse_args()
+    api_base_url = os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev")
+
+    # id-first: address the managed database by id (names are not identifiers), so
+    # the before/after reads below can find it. Provision it once and pin the id;
+    # pass --database-id to reuse an existing database.
+    database_id = args.database_id
+    if database_id is None:
+        from hotdata_framework.client import HotdataClient as RuntimeClient
+
+        rc = RuntimeClient(
+            os.environ["HOTDATA_API_KEY"], args.workspace_id, host=api_base_url.rstrip("/")
+        )
+        database_id = rc.create_managed_database(
+            description=DATABASE, schema=SCHEMA, tables=["orders"]
+        ).id
+        rc.close()
+        print(f"created managed database {database_id} (pass --database-id {database_id} to reuse)")
+
     pipeline = dlt.pipeline(
         pipeline_name="orders_merge",
         destination=hotdata(
             credentials=HotdataCredentials(api_key=os.environ["HOTDATA_API_KEY"]),
             workspace_id=args.workspace_id,
-            database_id=args.database_id,
-            api_base_url=os.environ.get("HOTDATA_API_BASE_URL", "https://api.hotdata.dev"),
+            database_id=database_id,
+            api_base_url=api_base_url,
             declared_tables=["orders"],
             database_name=DATABASE,
             schema=SCHEMA,

From 2b9b5cf464638d55bec0cf45edc5ca77fbed9736 Mon Sep 17 00:00:00 2001
From: Rohan Dsouza <rohan@hotdata.dev>
Date: Fri, 24 Jul 2026 22:37:17 +0530
Subject: [PATCH 6/6] fix: walk the __cause__ chain in _is_not_found for a
 dropped-id 404

_is_not_found only inspected the immediate __cause__. That matches where the
framework surfaces the 404 today, but if it ever wraps get_database errors in
an extra layer the dropped-id 404 would slip past and re-surface as an opaque
HotdataTerminalError, losing the clear "database_id was dropped" message. Walk
the whole chain (mirroring sql_client._chain_messages) so it stays robust.
Addresses claude[bot]'s review nit on PR #59.
---
 src/hotdata_dlt_destination/hotdata_client.py | 17 +++++++++++++++--
 tests/test_client.py                          | 17 +++++++++++++++++
 2 files changed, 32 insertions(+), 2 deletions(-)

diff --git a/src/hotdata_dlt_destination/hotdata_client.py b/src/hotdata_dlt_destination/hotdata_client.py
index d974d81..1c15d54 100644
--- a/src/hotdata_dlt_destination/hotdata_client.py
+++ b/src/hotdata_dlt_destination/hotdata_client.py
@@ -11,8 +11,21 @@
 
 
 def _is_not_found(exc: Exception) -> bool:
-    """True when ``exc`` wraps a 404 (the bound database id does not exist)."""
-    return getattr(getattr(exc, "__cause__", None), "status", None) == 404
+    """True when ``exc`` or anything in its ``__cause__`` chain wraps a 404.
+
+    The framework raises the mapped error ``from`` the underlying ``ApiException``,
+    so today the 404 sits one level down -- but it walks the whole chain (mirroring
+    ``sql_client._chain_messages``) so a dropped-id 404 is still recognised if the
+    framework ever wraps ``get_database`` errors in an extra layer.
+    """
+    current: BaseException | None = exc
+    for _ in range(6):
+        if current is None:
+            break
+        if getattr(current, "status", None) == 404:
+            return True
+        current = current.__cause__
+    return False
 
 
 class HotdataClient(ManagedDatabaseClient):
diff --git a/tests/test_client.py b/tests/test_client.py
index 010c7c4..ecdde77 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -159,6 +159,23 @@ def test_bind_missing_id_raises_keyerror(monkeypatch) -> None:
     client.close()
 
 
+def test_is_not_found_walks_cause_chain() -> None:
+    # A 404 buried under extra wrapping layers is still recognised, so the clear
+    # "database_id was dropped" message survives even if the framework nests errors.
+    from hotdata_dlt_destination.hotdata_client import _is_not_found
+
+    inner = ApiException(status=404, reason="not found")
+    mid = HotdataTerminalError("wrapped")
+    mid.__cause__ = inner
+    outer = HotdataTerminalError("outer")
+    outer.__cause__ = mid
+    assert _is_not_found(outer) is True
+
+    non_404 = HotdataTerminalError("x")
+    non_404.__cause__ = ApiException(status=500, reason="boom")
+    assert _is_not_found(non_404) is False
+
+
 def test_bind_missing_id_with_create_raises_clear_error(monkeypatch) -> None:
     # dev_mode/refresh drops the pinned db, then a create-path ensure finds the id
     # gone. It can't be recreated (ids are server-assigned) -> clear terminal error,