fix(managed): enumerate in-flight statuses, not terminal ones

Follow-up from review on the query-run wait.

Both polls listed the statuses that mean finished and treated everything
else as "keep waiting". That arrangement is what let `interrupted` be
polled for the full five-minute timeout: a status absent from the list
was indistinguishable from one still running. Listing the in-flight side
instead -- `running` for a query run, `pending`/`processing` for a result
-- makes an unrecognised status raise on the first pass with its own name
in the message, so the next status the API adds costs one round trip
rather than a timeout.

That also fixes both status bugs in `HotdataClient`, which still had
`cancelled` in its terminal sets and lacked `interrupted` entirely, so
`execute_sql` kept the behaviour this branch documents as a bug. Doing
both keeps the changelog entry true for the whole package rather than for
one of its two clients.

The async-submit test asserted the model attribute, which would still
pass if the `async` alias were missing or misspelled -- the server would
then ignore the field, answer synchronously with a full preview, and
silently restore what this branch removes. It now asserts the serialised
wire key.

Author: anoop-narang

CHANGELOG.md

@@ -35,16 +35,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - fix(managed): recognise `interrupted`, and drop a run status the API never sends.
 
-  The query-run poll treated `failed` and `cancelled` as the terminal failures.
-  `cancelled` is not a status this API returns. `interrupted` is — a run whose
-  server was replaced before it finished — and it matched neither branch, so the
-  poll ran to its five-minute timeout and raised `TimeoutError` instead of
-  failing fast.
-
-  An interrupted run is safe to re-run, so it is now raised as transient and the
-  surrounding retry re-submits the query. `classify_sdk_error` passes an
-  already-classified error through unchanged, rather than demoting a
-  caller-raised transient error to terminal.
+  Both `ManagedDatabaseClient` and `HotdataClient` treated `failed` and
+  `cancelled` as the terminal run failures. `cancelled` is not a status this API
+  returns. `interrupted` is — a run whose server was replaced before it finished
+  — and it matched neither, so an interrupted run was polled for the full
+  five-minute timeout and then raised `TimeoutError`: a retryable condition
+  hidden behind a long wait and an error naming the wrong problem.
+
+  On `ManagedDatabaseClient` an interrupted run is now raised as transient, so
+  the surrounding retry re-submits the query. That needed `classify_sdk_error` to
+  pass an already-classified error through unchanged rather than demoting a
+  caller-raised transient error to terminal. `HotdataClient.execute_sql` now
+  fails fast on it with the run's own message.
+
+  Both polls now enumerate the statuses that mean *still in flight* rather than
+  the ones that mean *finished*, for query runs and for results alike. Listing
+  the terminal side treats an unrecognised status as "keep waiting", which is
+  precisely how `interrupted` came to be waited out; listing the in-flight side
+  makes the next status the API adds fail on the first pass, naming itself.
 
 ## [0.13.0] - 2026-08-27
 

hotdata_framework/client.py

@@ -77,8 +77,15 @@
 _INDEX_TYPES = frozenset(get_args(IndexType))
 _VECTOR_METRICS = frozenset(get_args(VectorMetric))
 
-_TERMINAL = frozenset({"succeeded", "failed", "cancelled"})
-_RESULT_FAILURE = frozenset({"failed", "cancelled"})
+# Enumerate the in-flight statuses, not the terminal ones. A poll that lists
+# what is terminal treats anything it does not recognise as "keep waiting", so a
+# status the API adds later -- or one that was simply missed -- costs the full
+# timeout before surfacing. Listing what is still in flight makes an unknown
+# status raise on the first pass, naming itself. `interrupted` was missed by
+# exactly the other arrangement, and `cancelled` was listed here without being a
+# status this API sends.
+_RUN_IN_FLIGHT = frozenset({"running"})
+_RESULT_IN_FLIGHT = frozenset({"pending", "processing"})
 # Jobs have no "cancelled" state; "partially_succeeded" carries an error_message.
 _JOB_TERMINAL = frozenset({"succeeded", "partially_succeeded", "failed"})
 
@@ -918,7 +925,7 @@ def _poll_query_run(
         last = None
         while time.monotonic() < deadline:
             last = runs.get_query_run(query_run_id)
-            if last.status in _TERMINAL:
+            if last.status not in _RUN_IN_FLIGHT:
                 return last
             time.sleep(interval_s)
         raise TimeoutError(
@@ -1027,7 +1034,7 @@ def _wait_result_ready(
             last = results.get_result(result_id)
             if last.status == "ready":
                 return last
-            if last.status in _RESULT_FAILURE:
+            if last.status not in _RESULT_IN_FLIGHT:
                 raise RuntimeError(last.error_message or f"Result {last.status}")
             time.sleep(interval_s)
         raise TimeoutError(

hotdata_framework/managed_client.py

@@ -40,6 +40,11 @@ class ManagedDatabaseClient:
     database lifecycle.
     """
 
+    # The only status a query run reports while still in flight. Listed this way
+    # round so an unknown status raises immediately rather than being polled to
+    # the timeout; see `_await_query_run`.
+    _RUN_IN_FLIGHT = frozenset({"running"})
+
     _QUERY_TIMEOUT_SECONDS = 300.0
     _POLL_INTERVAL_SECONDS = 0.4
     _MAX_BACKOFF_SECONDS = 30.0
@@ -178,8 +183,12 @@ def _await_query_run(self, query_run_id: str, *, database_id: str) -> str | None
                 raise HotdataTransientError(
                     run.error_message or f"Query run {query_run_id} was interrupted"
                 )
-            if run.status == "failed":
-                raise RuntimeError(run.error_message or f"Query run {query_run_id} failed")
+            # Anything not still in flight is terminal, whether or not this
+            # client has heard of it. Enumerating the terminal statuses instead
+            # would poll an unrecognised one to the timeout -- which is exactly
+            # how `interrupted` came to be waited out for five minutes.
+            if run.status not in self._RUN_IN_FLIGHT:
+                raise RuntimeError(run.error_message or f"Query run {query_run_id} {run.status}")
             time.sleep(self._POLL_INTERVAL_SECONDS)
         raise TimeoutError(
             f"Query run {query_run_id} did not finish within {self._QUERY_TIMEOUT_SECONDS}s"

tests/test_client.py

@@ -245,20 +245,52 @@ def test_list_qualified_table_names_passes_connection_id():
     assert it.call_args.kwargs["connection_id"] == "conn_a"
 
 
-def test_wait_result_ready_raises_on_cancelled():
+def test_wait_result_ready_raises_on_any_status_that_is_not_in_flight():
+    """A result neither `ready` nor still being saved is terminal.
+
+    The poll enumerates the in-flight statuses rather than the terminal ones, so
+    a status this client has never heard of raises on the first pass instead of
+    being waited out. `cancelled` -- which this poll listed as terminal, and
+    which the API does not send -- is as good a stand-in as any.
+    """
     client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
 
     class FakeResultsApi:
         def get_result(self, result_id: str):
-            return SimpleNamespace(status="cancelled", error_message=None)
+            return SimpleNamespace(status="something_new", error_message=None)
 
     with (
         patch.object(client, "_results_api", return_value=FakeResultsApi()),
-        pytest.raises(RuntimeError, match="cancelled"),
+        pytest.raises(RuntimeError, match="something_new"),
     ):
         client._wait_result_ready("res_1", timeout_s=0.1, interval_s=0)
 
 
+def test_poll_query_run_returns_promptly_on_interrupted():
+    """`interrupted` is terminal, so the poll must stop on it.
+
+    It was absent from the terminal set, so an interrupted run was polled for the
+    full timeout and then raised `TimeoutError` -- hiding a retryable condition
+    behind a five-minute wait, behind an error naming the wrong problem.
+    """
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    calls: list[str] = []
+
+    class FakeQueryRunsApi:
+        def get_query_run(self, query_run_id: str):
+            calls.append(query_run_id)
+            return SimpleNamespace(
+                status="interrupted", error_message="instance lost", result_id=None
+            )
+
+    with patch.object(client, "_query_runs_api", return_value=FakeQueryRunsApi()):
+        run = client._poll_query_run("qrun_1", timeout_s=30.0, interval_s=0)
+
+    assert run.status == "interrupted"
+    # One request, not a timeout's worth.
+    assert len(calls) == 1
+
+
 def test_connection_id_by_name_raises_on_duplicate_names():
     client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
     listing = SimpleNamespace(

tests/test_managed_client.py

@@ -673,7 +673,12 @@ def get_result_arrow(self, result_id: str, **kwargs: Any) -> pa.Table:
 
     assert table is not None
     assert len(requests) == 1
-    assert requests[0].var_async is True
+    # The attribute alone is not the contract: the field only reaches the server
+    # as `async`. Were the alias missing or misspelled, the attribute assertion
+    # would still pass, the server would ignore the field and answer
+    # synchronously with a full preview -- restoring the exact behaviour this
+    # change removes, silently.
+    assert requests[0].to_dict()["async"] is True
 
 
 def test_async_reply_is_followed_through_to_arrow(
@@ -742,3 +747,49 @@ def get_result_arrow(self, result_id: str, **kwargs: Any) -> pa.Table:
         "get_query_run:succeeded",
         "arrow:rslt-from-run",
     ]
+
+
+def test_unknown_run_status_fails_fast_instead_of_polling_to_the_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Only `running` means "still in flight"; anything else is terminal.
+
+    Enumerating the terminal statuses is what let `interrupted` be polled for
+    five minutes, so the test runs the other way round: a status this client has
+    never seen raises on the first pass and names itself.
+    """
+    calls: list[str] = []
+
+    class FakeQueryApi:
+        def __init__(self, api: object) -> None:
+            pass
+
+        def query(self, request: object, *, x_database_id: str) -> AsyncQueryResponse:
+            return _async_query_response()
+
+    class FakeQueryRunsApi:
+        def __init__(self, api: object) -> None:
+            pass
+
+        def get_query_run(self, query_run_id: str, **kwargs: Any) -> Any:
+            calls.append("get_query_run")
+            return SimpleNamespace(status="evicted", result_id=None, error_message=None)
+
+    monkeypatch.setattr(mc, "QueryApi", FakeQueryApi)
+    monkeypatch.setattr(mc, "QueryRunsApi", FakeQueryRunsApi)
+    monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None)
+
+    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()
+
+    with pytest.raises(HotdataTerminalError, match="evicted"):
+        client.fetch_table(database="mydb", schema="public", table="orders")
+
+    # One poll, not a timeout's worth.
+    assert len(calls) == 1