fix(managed): carry X-Database-Id on result reads; keep API error bodies

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- `ManagedDatabaseClient.fetch_table` now carries the `X-Database-Id` scope header on the result poll, the query-run poll, and the Arrow fetch — not only on the query submit. Results of database-scoped queries are themselves database-scoped, so every read against an existing synced table (merge/append loads, dlt state restore) failed with `400: Bad Request` once the table had data (dlthubworker#70).
+- API error messages now include the response body (flattened, truncated to 500 chars). `400: Bad Request` alone hid the server's actual explanation.
 
 ## [0.6.0] - 2026-06-30
 

hotdata_framework/databases.py

@@ -58,4 +58,9 @@ def managed_database_from_detail(detail: Any) -> ManagedDatabase:
 
 
 def api_error_message(exc: ApiException) -> str:
-    return exc.reason or str(exc)
+    reason = exc.reason or str(exc)
+    # Keep the response body: it carries the API's actual explanation.
+    body = getattr(exc, "body", None)
+    if body:
+        return f"{reason}: {' '.join(str(body).split())[:500]}"
+    return reason

hotdata_framework/errors.py

@@ -23,6 +23,11 @@ def classify_sdk_error(error: Exception) -> HotdataError:
     if isinstance(error, ApiException):
         status_code = int(error.status or 0)
         message = f"{status_code}: {error.reason or 'unknown error'}"
+        # The response body is where the API explains itself (e.g. which
+        # header is missing) — without it "400: Bad Request" is undebuggable.
+        body = getattr(error, "body", None)
+        if body:
+            message = f"{message} — {' '.join(str(body).split())[:500]}"
         if status_code in (408, 409, 425, 429):
             return HotdataTransientError(message)
         if 500 <= status_code <= 599:

hotdata_framework/managed_client.py

@@ -110,10 +110,26 @@ def operation() -> pa.Table | None:
             result_id = self._query_database_scoped(sql, database_id=db.id)
             if result_id is None:
                 return None
-            return ArrowResultsApi(self._runtime.api).get_result_arrow(result_id)
+            return self._fetch_result_arrow(result_id, database_id=db.id)
 
         return self._request_with_retry(operation)
 
+    def _fetch_result_arrow(self, result_id: str, *, database_id: str) -> pa.Table:
+        """Fetch a ready result as Arrow, carrying the database scope.
+
+        Results of database-scoped queries are themselves database-scoped —
+        the results endpoints reject requests without an ``X-Database-Id``
+        header. The Arrow helper takes no per-call headers, so the header is
+        set as a client default for the duration of the call and removed
+        after (this client is not shared across threads).
+        """
+        api = self._runtime.api
+        api.set_default_header("X-Database-Id", database_id)
+        try:
+            return ArrowResultsApi(api).get_result_arrow(result_id)
+        finally:
+            api.default_headers.pop("X-Database-Id", None)
+
     def _poll(
         self,
         fetch: Callable[[], S],
@@ -146,26 +162,34 @@ def _query_database_scoped(self, sql: str, *, database_id: str) -> str | None:
             # under ``result_id``; that result may be ``processing`` when the
             # inline preview returns, so wait for ``ready`` before the caller
             # fetches it as Arrow.
-            return self._wait_result_ready(raw.result_id)
+            return self._wait_result_ready(raw.result_id, database_id=database_id)
         if isinstance(raw, AsyncQueryResponse):
-            return self._wait_result_ready(self._await_query_run(raw.query_run_id))
+            run_result = self._await_query_run(raw.query_run_id, database_id=database_id)
+            return self._wait_result_ready(run_result, database_id=database_id)
         return None
 
-    def _await_query_run(self, query_run_id: str) -> str | None:
+    def _await_query_run(self, query_run_id: str, *, database_id: str) -> str | None:
         runs = QueryRunsApi(self._runtime.api)
         run = self._poll(
-            lambda: runs.get_query_run(query_run_id),
+            # Runs (like results) of database-scoped queries are database-scoped.
+            lambda: runs.get_query_run(
+                query_run_id, _headers={"X-Database-Id": database_id}
+            ),
             is_ready=lambda r: r.status == "succeeded",
             describe="Query",
         )
         return run.result_id
 
-    def _wait_result_ready(self, result_id: str | None) -> str | None:
+    def _wait_result_ready(self, result_id: str | None, *, database_id: str) -> str | None:
         if result_id is None:
             return None
         results = ResultsApi(self._runtime.api)
         self._poll(
-            lambda: results.get_result(result_id),
+            # The stored result of a database-scoped query 400s without the
+            # database scope header.
+            lambda: results.get_result(
+                result_id, _headers={"X-Database-Id": database_id}
+            ),
             is_ready=lambda r: r.status == "ready",
             describe=f"Result {result_id}",
         )

tests/test_errors.py

@@ -0,0 +1,49 @@
+"""Error message construction: the API's response body must survive.
+
+"400: Bad Request" alone is undebuggable; the body carries the server's
+actual explanation (e.g. which header was missing). Regression for the
+opaque load failures in dlthubworker#70.
+"""
+
+from __future__ import annotations
+
+from hotdata.rest import ApiException
+
+from hotdata_framework.databases import api_error_message
+from hotdata_framework.errors import (
+    HotdataTerminalError,
+    HotdataTransientError,
+    classify_sdk_error,
+)
+
+BODY = '{"error":{"code":"BAD_REQUEST","message":"X-Database-Id header is required"}}'
+
+
+def test_classify_sdk_error_includes_response_body() -> None:
+    err = classify_sdk_error(ApiException(status=400, reason="Bad Request", body=BODY))
+    assert isinstance(err, HotdataTerminalError)
+    assert "400: Bad Request" in str(err)
+    assert "X-Database-Id header is required" in str(err)
+
+
+def test_classify_sdk_error_without_body_keeps_short_form() -> None:
+    err = classify_sdk_error(ApiException(status=409, reason="Conflict"))
+    assert isinstance(err, HotdataTransientError)
+    assert str(err) == "409: Conflict"
+
+
+def test_classify_sdk_error_truncates_and_flattens_body() -> None:
+    noisy = "x\n" * 1000
+    err = classify_sdk_error(ApiException(status=500, reason="ISE", body=noisy))
+    assert "\n" not in str(err)
+    assert len(str(err)) < 600
+
+
+def test_api_error_message_includes_body() -> None:
+    msg = api_error_message(ApiException(status=400, reason="Bad Request", body=BODY))
+    assert msg.startswith("Bad Request: ")
+    assert "X-Database-Id header is required" in msg
+
+
+def test_api_error_message_without_body() -> None:
+    assert api_error_message(ApiException(status=404, reason="Not Found")) == "Not Found"

tests/test_managed_client.py

@@ -50,7 +50,7 @@ class FakeResultsApi:
         def __init__(self, api: object) -> None:
             pass
 
-        def get_result(self, result_id: str) -> Any:
+        def get_result(self, result_id: str, **kwargs: Any) -> Any:
             status = next(statuses)
             calls.append(f"get_result:{status}")
             return SimpleNamespace(status=status, result_id=result_id, error_message=None)
@@ -75,13 +75,7 @@ def get_result_arrow(self, result_id: str) -> pa.Table:
         max_retries=1,
         retry_backoff_seconds=0.0,
     )
-    client._runtime = SimpleNamespace(  # type: ignore[assignment]
-        api=object(),
-        resolve_managed_database=lambda name: SimpleNamespace(id="db1", default_connection_id="c"),
-        list_managed_tables=lambda database, schema=None: [
-            SimpleNamespace(table="orders", synced=True)
-        ],
-    )
+    client._runtime = _fake_runtime()
 
     table = client.fetch_table(database="mydb", schema="public", table="orders")
 
@@ -91,3 +85,86 @@ def get_result_arrow(self, result_id: str) -> pa.Table:
     assert "get_result:processing" in calls
     assert "get_result:ready" in calls
     assert calls.index("arrow") > calls.index("get_result:ready")
+
+
+class _FakeApiClient:
+    """Just enough of the generated ApiClient's default-header surface."""
+
+    def __init__(self) -> None:
+        self.default_headers: dict[str, str] = {}
+
+    def set_default_header(self, name: str, value: str) -> None:
+        self.default_headers[name] = value
+
+
+def _fake_runtime(api: object | None = None) -> SimpleNamespace:
+    return SimpleNamespace(
+        api=api if api is not None else _FakeApiClient(),
+        resolve_managed_database=lambda name: SimpleNamespace(id="db1", default_connection_id="c"),
+        list_managed_tables=lambda database, schema=None: [
+            SimpleNamespace(table="orders", synced=True)
+        ],
+    )
+
+
+def test_fetch_table_carries_database_scope_on_result_reads(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Results (and runs) of a database-scoped query are database-scoped:
+    the results endpoints 400 with "X-Database-Id header is required" when
+    the header is missing. ``fetch_table`` must carry the database id on the
+    result poll and the Arrow fetch, not only on the query submit.
+
+    Regression: reruns/append loads against an existing synced table failed
+    with an opaque ``400: Bad Request`` (dlthubworker#70) because both reads
+    omitted the header.
+    """
+    seen_headers: list[dict[str, Any] | None] = []
+    arrow_headers: list[str | None] = []
+    fake_api = _FakeApiClient()
+
+    class FakeQueryApi:
+        def __init__(self, api: object) -> None:
+            pass
+
+        def query(self, request: object, *, x_database_id: str) -> QueryResponse:
+            assert x_database_id == "db1"
+            return _query_response("rslt1")
+
+    class FakeResultsApi:
+        def __init__(self, api: object) -> None:
+            pass
+
+        def get_result(self, result_id: str, *, _headers: dict[str, Any] | None = None) -> Any:
+            seen_headers.append(_headers)
+            return SimpleNamespace(status="ready", result_id=result_id, error_message=None)
+
+    class FakeArrowResultsApi:
+        def __init__(self, api: _FakeApiClient) -> None:
+            self._api = api
+
+        def get_result_arrow(self, result_id: str) -> pa.Table:
+            # The scoped default header must be present DURING the fetch.
+            arrow_headers.append(self._api.default_headers.get("X-Database-Id"))
+            return pa.table({"id": [1]})
+
+    monkeypatch.setattr(mc, "QueryApi", FakeQueryApi)
+    monkeypatch.setattr(mc, "ResultsApi", FakeResultsApi)
+    monkeypatch.setattr(mc, "ArrowResultsApi", FakeArrowResultsApi)
+
+    client = mc.ManagedDatabaseClient(
+        api_key="k",
+        workspace_id="w",
+        api_base_url="https://example.test",
+        max_retries=1,
+        retry_backoff_seconds=0.0,
+    )
+    client._runtime = _fake_runtime(fake_api)
+
+    table = client.fetch_table(database="mydb", schema="public", table="orders")
+
+    assert table is not None
+    assert seen_headers == [{"X-Database-Id": "db1"}]
+    assert arrow_headers == ["db1"]
+    # The scoped header is removed once the fetch completes.
+    assert "X-Database-Id" not in fake_api.default_headers