Align runtime contract with adapter usage and ambiguity safeguards.

Document the adapter-facing HotdataClient methods in the runtime contract, add duplicate-connection-name protection, and support explicit connection_id column resolution with regression coverage.

Author: eddietejeda

CONTRACT.md

@@ -39,6 +39,12 @@ Adapters should import from `hotdata_runtime` and treat this surface as the stab
 - `from_env()` resolves runtime context from env vars and selected workspace.
 - `execute_sql(sql)` returns `QueryResult` or raises `RuntimeError`/`TimeoutError`.
 - `get_result(result_id)` returns a ready `QueryResult` and waits for readiness when needed.
+- `connections()` returns the connections API wrapper for adapter UI/status features.
+- `query_runs()` returns the query-runs API wrapper for adapter history views.
+- `results()` returns the results API wrapper for adapter result pickers.
+- `list_qualified_table_names(...)` returns sorted fully qualified table names.
+- `columns_for_qualified(qualified, connection_id=...)` resolves table columns, and
+  adapters should pass `connection_id` when known.
 
 ### `QueryResult`
 

hotdata_runtime/client.py

@@ -144,9 +144,26 @@ def list_qualified_table_names(
 
     def connection_id_by_name(self) -> dict[str, str]:
         listing = self.connections().list_connections()
-        return {c.name: c.id for c in listing.connections}
+        id_map: dict[str, str] = {}
+        duplicate_names: set[str] = set()
+        for c in listing.connections:
+            if c.name in id_map and id_map[c.name] != c.id:
+                duplicate_names.add(c.name)
+            id_map[c.name] = c.id
+        if duplicate_names:
+            names = ", ".join(sorted(duplicate_names))
+            raise RuntimeError(
+                f"Duplicate connection names found: {names}. "
+                "Use an explicit connection_id."
+            )
+        return id_map
 
-    def columns_for_qualified(self, qualified: str) -> list[TableInfo]:
+    def columns_for_qualified(
+        self,
+        qualified: str,
+        *,
+        connection_id: str | None = None,
+    ) -> list[TableInfo]:
         parts = qualified.split(".")
         if len(parts) < 3:
             raise ValueError(
@@ -157,10 +174,12 @@ def columns_for_qualified(self, qualified: str) -> list[TableInfo]:
             parts[1],
             ".".join(parts[2:]),
         )
-        id_map = self.connection_id_by_name()
-        conn_id = id_map.get(conn_name)
-        if not conn_id:
-            raise KeyError(f"Unknown connection {conn_name!r}")
+        conn_id = connection_id
+        if conn_id is None:
+            id_map = self.connection_id_by_name()
+            conn_id = id_map.get(conn_name)
+            if not conn_id:
+                raise KeyError(f"Unknown connection {conn_name!r}")
         resp = self._information_schema().information_schema(
             connection_id=conn_id,
             var_schema=schema_name,

tests/test_client.py

@@ -81,6 +81,24 @@ def test_pick_workspace_falls_back_to_first(monkeypatch: pytest.MonkeyPatch):
         assert pick_workspace("k", "https://api.hotdata.dev", None) == "ws_1"
 
 
+def test_resolve_workspace_selection_source_first(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.delenv("HOTDATA_WORKSPACE", raising=False)
+    monkeypatch.delenv("HOTDATA_WORKSPACE_ID", raising=False)
+    items = [
+        SimpleNamespace(public_id="ws_1", active=False),
+        SimpleNamespace(public_id="ws_2", active=False),
+    ]
+    listing = SimpleNamespace(workspaces=items)
+    with patch("hotdata_runtime.env.WorkspacesApi") as Api:
+        Api.return_value.list_workspaces.return_value = listing
+        resolved = resolve_workspace_selection(
+            "k", "https://api.hotdata.dev", None
+        )
+    assert resolved.workspace_id == "ws_1"
+    assert resolved.source == "first"
+    assert resolved.workspaces == items
+
+
 def test_resolve_workspace_selection_returns_workspaces_and_source(
     monkeypatch: pytest.MonkeyPatch,
 ):
@@ -121,3 +139,48 @@ def get_result(self, result_id: str):
     with patch.object(client, "_results_api", return_value=FakeResultsApi()):
         with pytest.raises(RuntimeError, match="cancelled"):
             client._wait_result_ready("res_1", timeout_s=0.1, interval_s=0)
+
+
+def test_connection_id_by_name_raises_on_duplicate_names():
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    listing = SimpleNamespace(
+        connections=[
+            SimpleNamespace(name="warehouse", id="conn_1"),
+            SimpleNamespace(name="warehouse", id="conn_2"),
+        ]
+    )
+
+    class FakeConnectionsApi:
+        def list_connections(self):
+            return listing
+
+    with patch.object(client, "connections", return_value=FakeConnectionsApi()):
+        with pytest.raises(RuntimeError, match="Duplicate connection names"):
+            client.connection_id_by_name()
+
+
+def test_columns_for_qualified_prefers_explicit_connection_id():
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    col = SimpleNamespace(name="a", data_type="INTEGER", nullable=True)
+    table = SimpleNamespace(columns=[col])
+    response = SimpleNamespace(tables=[table])
+
+    class FakeInformationSchemaApi:
+        def __init__(self):
+            self.kwargs = None
+
+        def information_schema(self, **kwargs):
+            self.kwargs = kwargs
+            return response
+
+    fake_api = FakeInformationSchemaApi()
+    with patch.object(client, "_information_schema", return_value=fake_api), patch.object(
+        client, "connection_id_by_name"
+    ) as id_map:
+        cols = client.columns_for_qualified(
+            "warehouse.public.orders",
+            connection_id="conn_explicit",
+        )
+    id_map.assert_not_called()
+    assert cols == [col]
+    assert fake_api.kwargs["connection_id"] == "conn_explicit"