diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index 240aaa417..163c70afa 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -49,7 +49,7 @@ jobs:
       - name: "Ruff format"
         run: ruff format --diff weaviate test mock_tests integration packages/web
       - name: "Flake 8"
-        run: flake8 weaviate test mock_tests integration
+        run: flake8 weaviate test mock_tests integration packages/web
       - name: "Check release for pypi"
         run: |
           python -m build
diff --git a/ci/pyodide-e2e/e2e.py b/ci/pyodide-e2e/e2e.py
index 0ebba1391..ae92e40f5 100644
--- a/ci/pyodide-e2e/e2e.py
+++ b/ci/pyodide-e2e/e2e.py
@@ -10,14 +10,17 @@
 """
 
 import os
+import uuid
 import warnings
 
 import weaviate_client_web  # bootstraps the grpc shim + fetch transport under Emscripten
 
 import grpc
+import httpx
 import weaviate
 import weaviate.classes as wvc
-from weaviate.classes.config import DataType, Property
+from weaviate.classes.config import DataType, Property, ReferenceProperty
+from weaviate.classes.data import DataReference
 from weaviate.classes.query import Filter
 from weaviate.classes.tenants import Tenant
 from weaviate.exceptions import WeaviateBatchStreamError, WeaviateQueryError
@@ -38,6 +41,13 @@ async def main() -> None:
     assert getattr(grpc, "__weaviate_client_web_shim__", False), (
         "sys.modules['grpc'] is not the shim"
     )
+    # REST must run through the package's own fetch transport, not Pyodide's bundled
+    # httpx transport (which cannot read the null body of HEAD / 204 responses).
+    assert weaviate_client_web.is_fetch_transport_installed(), "fetch transport not installed"
+    assert getattr(
+        httpx.AsyncHTTPTransport.handle_async_request, "__weaviate_fetch_shim__", False
+    ), "httpx.AsyncHTTPTransport is not the package's fetch transport"
+    ok("self-check: package fetch transport is the active httpx transport")
 
     host = os.environ.get("WEAVIATE_HOST", "localhost")
     port = int(os.environ.get("WEAVIATE_PORT", "8090"))
@@ -66,6 +76,7 @@ async def main() -> None:
                 Property(name="title", data_type=DataType.TEXT),
                 Property(name="idx", data_type=DataType.INT),
             ],
+            references=[ReferenceProperty(name="related", target_collection=COLL)],
         )
         ok("collections.create")
 
@@ -97,6 +108,40 @@ async def main() -> None:
         assert idx.minimum == 0 and idx.maximum == 49, agg.properties
         ok("aggregate count=50 min=0 max=49")
 
+        # REST calls answered without a body (HEAD 204/404, PATCH/DELETE 204) and the
+        # batch-references path, which reads httpx's response.elapsed.
+        first, second, last = ret.uuids[0], ret.uuids[1], ret.uuids[49]
+        assert await coll.data.exists(first) is True
+        assert await coll.data.exists(uuid.uuid4()) is False
+        ok("data.exists (HEAD 204 / 404) = True / False")
+
+        await coll.data.update(uuid=first, properties={"title": "article 0 (updated)"})
+        obj = await coll.query.fetch_object_by_id(first)
+        assert obj is not None and obj.properties["title"] == "article 0 (updated)", obj
+        ok("data.update (PATCH 204) -> fetch_object_by_id sees the update")
+
+        refs = await coll.data.reference_add_many(
+            [
+                DataReference(
+                    from_property="related", from_uuid=ret.uuids[i], to_uuid=ret.uuids[i + 1]
+                )
+                for i in range(5)
+            ]
+        )
+        assert not refs.has_errors, f"reference_add_many errors: {refs.errors}"
+        assert refs.elapsed_seconds >= 0, refs
+        ok("data.reference_add_many (REST /batch/references) = 5")
+
+        await coll.data.reference_delete(from_uuid=first, from_property="related", to=second)
+        ok("data.reference_delete (DELETE 204)")
+
+        assert await coll.data.delete_by_id(last) is True
+        assert await coll.data.exists(last) is False
+        # deleting a missing object answers 204 or 404 depending on the server topology;
+        # either way it is a body-less response the transport must handle
+        assert isinstance(await coll.data.delete_by_id(last), bool)
+        ok("data.delete_by_id (DELETE 204; repeat -> 204/404) = True, then bool")
+
         await client.collections.create(
             MT_COLL,
             vector_config=wvc.config.Configure.Vectors.self_provided(),
@@ -109,6 +154,10 @@ async def main() -> None:
         assert set(tenants.keys()) == {"t1", "t2"}, f"TenantsGet: {set(tenants.keys())}"
         ok("multi-tenant create + TenantsGet = {t1, t2}")
 
+        assert await mt.tenants.exists("t1") is True
+        assert await mt.tenants.exists("t404") is False
+        ok("tenants.exists (HEAD 200 / 404) = True / False")
+
         t1 = mt.with_tenant("t1")
         ret = await t1.data.insert_many([{"title": f"tenant doc {i}"} for i in range(10)])
         assert not ret.has_errors and len(ret.uuids) == 10, f"tenant insert_many: {ret.errors}"
@@ -126,8 +175,8 @@ async def main() -> None:
             await client.collections.get("DoesNotExistXyz").query.fetch_objects(limit=1)
             raise AssertionError("expected WeaviateQueryError for nonexistent collection")
         except WeaviateQueryError as e:
-            assert "DoesNotExistXyz" in str(e) or "not" in str(e).lower(), str(e)
-        ok("error mapping: nonexistent collection -> WeaviateQueryError")
+            assert "DoesNotExistXyz" in str(e), str(e)
+        ok("error mapping: nonexistent collection -> WeaviateQueryError names the collection")
 
         try:
             async with client.batch.stream() as batch:
diff --git a/mock_tests/test_auth.py b/mock_tests/test_auth.py
index 95d64858e..cbd5e1f4a 100644
--- a/mock_tests/test_auth.py
+++ b/mock_tests/test_auth.py
@@ -1,8 +1,9 @@
 import asyncio
 import json
+import threading
 import time
 import warnings
-from typing import Union
+from typing import List, Union
 
 import grpc
 import pytest
@@ -12,7 +13,7 @@
 import weaviate
 from mock_tests.conftest import CLIENT_ID, MOCK_IP, MOCK_PORT, MOCK_PORT_GRPC
 from weaviate.connect.v4 import _ConnectionBase
-from weaviate.exceptions import MissingScopeException
+from weaviate.exceptions import MissingScopeException, UnexpectedStatusCodeError
 
 ACCESS_TOKEN = "HELLO!IamAnAccessToken"
 CLIENT_SECRET = "SomeSecret.DontTell"
@@ -123,27 +124,48 @@ def handler(request: Request) -> Response:
         assert token_requests > first  # a fresh token was fetched with the credentials
 
 
-@pytest.mark.asyncio
-async def test_token_refresh_survives_non_http_error_async(
-    weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn
-) -> None:
-    """A refresh failure that is NOT an httpx.HTTPError must warn and keep the refresher alive.
+def _reject_refreshes(weaviate_auth_mock: HTTPServer) -> List[float]:
+    """Make the IdP reject every refresh (400 invalid_grant); returns the hit timestamps."""
+    hits: List[float] = []
 
-    An IdP rejecting the refresh (400 invalid_grant) makes authlib raise OAuthError. When
-    only HTTPError was caught, the refresh task died silently — nothing awaits it, so not
-    even asyncio's 'exception was never retrieved' fired — and every later request 401ed
-    with nothing pointing at the token refresh.
-    """
-    weaviate_auth_mock.expect_request("/auth").respond_with_response(
-        Response(
+    def handler(request: Request) -> Response:
+        hits.append(time.monotonic())
+        return Response(
             json.dumps({"error": "invalid_grant", "error_description": "refresh token expired"}),
             status=400,
             content_type="application/json",
         )
-    )
+
+    weaviate_auth_mock.expect_request("/auth").respond_with_handler(handler)
     weaviate_auth_mock.expect_request(
         "/v1/schema", headers={"Authorization": "Bearer " + ACCESS_TOKEN}
     ).respond_with_json({"classes": []})
+    return hits
+
+
+def _assert_backed_off(hits: List[float], recwarn) -> None:
+    # attempts at ~1s, ~2s, ~4s after connect: a fixed 1s retry would have made 5 in 5s
+    assert 2 <= len(hits) <= 3
+    gaps = [b - a for a, b in zip(hits, hits[1:])]
+    assert all(later > earlier * 1.5 for earlier, later in zip(gaps, gaps[1:]))
+    failed = [w for w in recwarn if str(w.message).startswith("Con001")]
+    assert len(failed) == len(hits)  # one warning per failed attempt
+    assert "invalid_grant" in str(failed[0].message)
+    assert "retrying in 1s" in str(failed[0].message)
+    assert "retrying in 2s" in str(failed[1].message)
+    assert "unstable internet" not in str(failed[0].message)
+
+
+@pytest.mark.asyncio
+async def test_token_refresh_backs_off_on_persistent_failure_async(
+    weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn
+) -> None:
+    """A permanently failing refresh must neither kill the refresher nor hot-loop the IdP.
+
+    A 400 invalid_grant makes authlib raise OAuthError: warn, back off exponentially,
+    stay alive.
+    """
+    hits = _reject_refreshes(weaviate_auth_mock)
 
     async with weaviate.use_async_with_local(
         host=MOCK_IP,
@@ -157,23 +179,132 @@ async def test_token_refresh_survives_non_http_error_async(
     ) as client:
         task = getattr(client._connection, "_ConnectionBase__token_refresh_task")  # noqa: B009
         assert task is not None
-        await asyncio.sleep(2.5)  # long enough for at least two failed attempts
-        assert not task.done()  # the refresher survived the failure
+        await asyncio.sleep(5)
+        assert not task.done()  # the refresher survived the failures
         await client.collections.list_all()  # ... and the client still works
 
-    failed = [w for w in recwarn if str(w.message).startswith("Con001")]
-    assert len(failed) >= 1
-    assert "invalid_grant" in str(failed[0].message)
+    _assert_backed_off(hits, recwarn)
     # the task ended by close()'s cancellation, not by dying on the exception
     assert [w for w in recwarn if str(w.message).startswith("Con003")] == []
 
 
+def test_token_refresh_backs_off_on_persistent_failure(
+    weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn
+) -> None:
+    """Sync colour of the test above.
+
+    The daemon thread used to die silently on anything but an httpx.HTTPError; now it
+    warns and backs off like the async task.
+    """
+    hits = _reject_refreshes(weaviate_auth_mock)
+
+    threads_before = set(threading.enumerate())
+    with weaviate.connect_to_local(
+        host=MOCK_IP,
+        port=MOCK_PORT,
+        grpc_port=MOCK_PORT_GRPC,
+        auth_credentials=weaviate.auth.AuthBearerToken(
+            ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=1
+        ),
+    ) as client:
+        refreshers = [
+            t for t in set(threading.enumerate()) - threads_before if t.name == "TokenRefresh"
+        ]
+        assert len(refreshers) == 1
+        time.sleep(5)
+        assert refreshers[0].is_alive()  # survived the failures
+        client.collections.list_all()
+
+    _assert_backed_off(hits, recwarn)
+    refreshers[0].join(timeout=2)
+    assert not refreshers[0].is_alive()  # close() stops the daemon thread promptly
+
+
+class _Boom(BaseException):
+    """Escapes the refresh loop's `except Exception` like a real BaseException would."""
+
+
+@pytest.mark.asyncio
+async def test_token_refresh_death_is_surfaced_through_real_connect(
+    weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server, recwarn
+) -> None:
+    """A refresher started by connect() that dies outside the loop body must warn (Con003).
+
+    Pins the done-callback wiring in _create_background_token_refresh, which the
+    callback-only unit test below cannot see.
+    """
+    weaviate_auth_mock.expect_request(
+        "/v1/schema", headers={"Authorization": "Bearer " + ACCESS_TOKEN}
+    ).respond_with_json({"classes": []})
+
+    async def boom(*args, **kwargs):
+        raise _Boom("boom")
+
+    async with weaviate.use_async_with_local(
+        host=MOCK_IP,
+        port=MOCK_PORT,
+        grpc_port=MOCK_PORT_GRPC,
+        auth_credentials=weaviate.auth.AuthBearerToken(
+            ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=1
+        ),
+    ) as client:
+        task = getattr(client._connection, "_ConnectionBase__token_refresh_task")  # noqa: B009
+        assert task is not None
+        client._connection._client.refresh_token = boom  # type: ignore[union-attr]
+        await asyncio.sleep(2)  # the first refresh is due after ~1s
+        assert task.done() and not task.cancelled()
+
+    stopped = [w for w in recwarn if str(w.message).startswith("Con003")]
+    assert len(stopped) == 1
+    assert "_Boom" in str(stopped[0].message)
+
+
+@pytest.mark.asyncio
+async def test_failed_connect_cancels_the_refresher_and_close_is_safe_after(
+    weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server
+) -> None:
+    """A failed connect() must not leave the refresher running; close() after it is safe.
+
+    The OIDC step starts the refresher before /v1/meta is checked, and close() afterwards
+    (even twice) must neither hang nor raise.
+    """
+    weaviate_auth_mock.expect_request("/auth").respond_with_json(
+        {"access_token": ACCESS_TOKEN, "expires_in": 500, "refresh_token": REFRESH_TOKEN}
+    )
+    # oneshot handlers take precedence over the fixture's permanent /v1/meta handler
+    weaviate_auth_mock.expect_oneshot_request("/v1/meta").respond_with_response(
+        Response(status=500)
+    )
+
+    tasks_before = asyncio.all_tasks()
+    client = weaviate.use_async_with_local(
+        host=MOCK_IP,
+        port=MOCK_PORT,
+        grpc_port=MOCK_PORT_GRPC,
+        auth_credentials=weaviate.auth.AuthBearerToken(
+            ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=500
+        ),
+    )
+    with pytest.raises(UnexpectedStatusCodeError):
+        await client.connect()
+
+    refresh_tasks = [
+        t for t in asyncio.all_tasks() - tasks_before if "token_refresh" in repr(t.get_coro())
+    ]
+    assert len(refresh_tasks) == 1  # it was started ...
+    await asyncio.sleep(0)
+    assert refresh_tasks[0].cancelled()  # ... and cancelled by the failed connect
+
+    await asyncio.wait_for(client.close(), timeout=2)
+    await asyncio.wait_for(client.close(), timeout=2)
+
+
 @pytest.mark.asyncio
 async def test_token_refresh_death_outside_loop_body_is_surfaced() -> None:
     """A refresher that dies where the loop body cannot catch it must still warn.
 
-    Nothing awaits the task and _cancel_background_token_refresh drops the reference, so
-    the done-callback is the only thing that can observe such a death.
+    Nothing awaits the task while it runs (close() only gathers it, exceptions included),
+    so the done-callback is the only thing that can observe such a death.
     """
     on_done = getattr(_ConnectionBase, "_ConnectionBase__warn_if_token_refresh_died")  # noqa: B009
 
@@ -347,11 +478,52 @@ async def test_async_auth_starts_no_threads(
             t for t in asyncio.all_tasks() - tasks_before if "token_refresh" in repr(t.get_coro())
         ]
         assert len(refresh_tasks) == 1  # the refresher runs as an asyncio task instead
-    # ... and close() must cancel it, not leak it (one wait for the cancellation to land)
-    await asyncio.wait(refresh_tasks, timeout=1)
+    # ... and close() must cancel it AND await it: done as soon as close() returns
     assert refresh_tasks[0].done()
 
 
+def test_sync_reconnect_leaves_exactly_one_refresher_thread(
+    weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server
+) -> None:
+    """close() must end the daemon thread promptly, and a later connect() must not revive it.
+
+    The thread used to re-read the connection's shutdown event on every loop, so after
+    close()+connect() it picked up the NEW (unset) event and kept refreshing next to the
+    new thread — two refreshers per client.
+    """
+    weaviate_auth_mock.expect_request("/auth").respond_with_json(
+        {"access_token": ACCESS_TOKEN, "expires_in": 500, "refresh_token": REFRESH_TOKEN}
+    )
+    weaviate_auth_mock.expect_request(
+        "/v1/schema", headers={"Authorization": "Bearer " + ACCESS_TOKEN}
+    ).respond_with_json({"classes": []})
+
+    def refreshers() -> List[threading.Thread]:
+        return [t for t in set(threading.enumerate()) - threads_before if t.name == "TokenRefresh"]
+
+    threads_before = set(threading.enumerate())
+    client = weaviate.connect_to_local(
+        host=MOCK_IP,
+        port=MOCK_PORT,
+        grpc_port=MOCK_PORT_GRPC,
+        auth_credentials=weaviate.auth.AuthBearerToken(
+            ACCESS_TOKEN, refresh_token=REFRESH_TOKEN, expires_in=500
+        ),
+    )
+    (first,) = refreshers()
+    client.close()
+    first.join(timeout=2)
+    assert not first.is_alive()  # not asleep until the next (470s away) wake-up
+
+    client.connect()
+    client.collections.list_all()
+    alive = [t for t in refreshers() if t.is_alive()]
+    assert len(alive) == 1 and alive[0] is not first
+    client.close()
+    alive[0].join(timeout=2)
+    assert not alive[0].is_alive()
+
+
 def test_refresh_of_refresh(weaviate_auth_mock: HTTPServer, start_grpc_server: grpc.Server) -> None:
     """Test that refresh tokens are used to get a new refresh token token."""
     weaviate_auth_mock.expect_request(
diff --git a/packages/web/README.md b/packages/web/README.md
index 800f476ae..e1aa554f1 100644
--- a/packages/web/README.md
+++ b/packages/web/README.md
@@ -8,6 +8,10 @@ WASM workers) where there is no socket and no `grpcio` wheel.
 It is built from the same repository as `weaviate-client` and reuses its generated
 protobuf stubs — it does **not** fork code generation.
 
+Requires Weaviate ≥ 1.38.3 (the first release to serve grpc-web natively) or a grpc-web
+transcoder in front of an older server. Pyodide ≥ 0.27 recommended; verified on
+Pyodide 314.0.4 (CPython 3.14).
+
 ## How it works
 
 Under Pyodide there is no `grpcio` Emscripten wheel, and `import weaviate` hard-imports
@@ -23,31 +27,55 @@ Under Pyodide there is no `grpcio` Emscripten wheel, and `import weaviate` hard-
   (`grpc.__version__` / `grpc._utilities.first_version_is_lower`).
 
 The `GrpcWebChannel` frames unary RPCs as grpc-web (a 5-byte header + protobuf payload)
-and POSTs them via `pyodide.http.pyfetch` to a server fronted by a grpc-web transcoder
-(e.g. Envoy or [connectrpc/vanguard](https://github.com/connectrpc/vanguard-go)). Call
-metadata (API key / OIDC bearer) is folded into `fetch` headers.
-
-For REST, Pyodide ≥ 0.27 distributes a patched httpx that already routes through the
-browser's `fetch` natively — when that build is detected the package leaves it alone.
-Only when httpx resolved from PyPI (httpcore + raw sockets, which cannot work under
-WASM) does the package patch `httpx.AsyncHTTPTransport` with its own pyfetch-based
-transport.
+and POSTs them via `pyodide.http.pyfetch`. Call metadata (API key / OIDC bearer) is
+folded into `fetch` headers. Two deployments work:
+
+1. **Core-native grpc-web (Weaviate ≥ 1.38.3).** Weaviate serves grpc-web on its REST
+   port under `/v1/grpc-web`, so gRPC and REST share one host, port and TLS setup and
+   no proxy is needed. Select it with `grpc_path_prefix="/v1/grpc-web"` and point
+   `grpc_host`/`grpc_port` at the REST endpoint.
+2. **A grpc-web transcoder** (e.g. Envoy or
+   [connectrpc/vanguard](https://github.com/connectrpc/vanguard-go)) listening at the
+   root of `grpc_host:grpc_port` in front of Weaviate's native gRPC port. This is what
+   `use_async_with_local()` / `use_async_with_weaviate_cloud()` need, because those
+   helpers have no `grpc_path_prefix` parameter.
+
+For REST (`is_ready`, collection config, `/batch/references`, …) the package patches
+`httpx.AsyncHTTPTransport` with its own `pyfetch`-based transport. It does so even on
+Pyodide builds whose bundled httpx has a JS-fetch transport of its own: that transport
+cannot read the null body of HEAD requests and 204 responses (`data.exists`,
+`data.delete_by_id`, `tenants.exists`, …) and does not enforce the client's per-request
+timeouts.
 
 ## Usage
 
 With this package installed, a plain `import weaviate` is all you need — under
 Emscripten the base client imports `weaviate_client_web` itself before anything else,
-which installs the shim (and raises a clear error if the package is missing):
+which installs the shim and the fetch transport (and raises a clear error if the package
+is missing). Against Weaviate ≥ 1.38.3:
 
 ```python
 import weaviate  # bootstraps weaviate_client_web automatically under Emscripten
 
-client = weaviate.use_async_with_local(skip_init_checks=True)
-await client.connect()
+client = weaviate.use_async_with_custom(
+    http_host="localhost",
+    http_port=8080,
+    http_secure=False,
+    grpc_host="localhost",   # same host as REST
+    grpc_port=8080,          # same port as REST: grpc-web rides on the REST listener
+    grpc_secure=False,
+    grpc_path_prefix="/v1/grpc-web",
+)
+await client.connect()       # runs the gRPC health check over grpc-web
 collection = client.collections.get("Article")
 await collection.query.near_text("hello", limit=3)
 ```
 
+`use_async_with_local()` and `use_async_with_weaviate_cloud()` have no
+`grpc_path_prefix`, so under Pyodide they only work when a grpc-web transcoder answers at
+the root of `grpc_host:grpc_port`. Pass `headers={...}` / `auth_credentials=...` to
+`use_async_with_custom` as usual for API keys, OIDC or WCD.
+
 Importing the companion explicitly first also works and remains the explicit form:
 
 ```python
@@ -60,8 +88,8 @@ import weaviate
 | Feature                                                   | Kind            | Status |
 |----------------------------------------------------------|-----------------|--------|
 | Search, Aggregate, TenantsGet, BatchObjects, BatchDelete | unary gRPC      | ✅ works over grpc-web |
-| Health check (`/grpc.health.v1.Health/Check`)            | unary gRPC      | ✅ (recommend `skip_init_checks=True` + REST `/.well-known/ready`) |
-| REST (`is_ready`, config, `/batch/references`, …)         | REST            | ✅ via fetch (Pyodide's httpx build, or this package's fallback transport) |
+| Health check (`/grpc.health.v1.Health/Check`)            | unary gRPC      | ✅ runs on `connect()` over grpc-web |
+| REST (`is_ready`, config, `/batch/references`, …)         | REST            | ✅ via the package's own fetch transport |
 | API-key auth (`Auth.api_key`)                             | header          | ✅ |
 | OIDC auth (`client_credentials` / `client_password` / `bearer_token`) | REST | ✅ token fetch + asyncio-task refresh (no threads) |
 | Bulk insert: `collection.data.insert_many()`              | unary gRPC      | ✅ the supported bulk path under WASM |
@@ -70,7 +98,7 @@ import weaviate
 | Embedded Weaviate (`use_async_with_embedded`)            | subprocess      | ❌ raises "not supported under WebAssembly/Pyodide" |
 | Synchronous client                                       | —               | ❌ async-only under WASM |
 | Weaviate Agents: `AsyncQueryAgent` `run/ask/search`      | REST            | ✅ via fetch |
-| Weaviate Agents: `ask_stream` / `research_stream` (SSE)  | REST streaming  | ⚠️ degraded under the fallback transport: fully buffered, events arrive only when the run completes (and long runs can hit the request timeout) |
+| Weaviate Agents: `ask_stream` / `research_stream` (SSE)  | REST streaming  | ⚠️ degraded: the fetch transport buffers the whole response, so events arrive only when the run completes (and long runs can hit the request timeout) |
 | Weaviate Agents: sync `QueryAgent`, `TransformationAgent`, `PersonalizationAgent` | REST sync | ❌ no async flavour exists |
 
 ## Configuration not honored in the browser
@@ -82,17 +110,22 @@ under WASM:
   cannot proxy fetch requests per-client),
 - connection-pool sizing and `session_pool_max_retries`,
 - `GrpcConfig.credentials` (custom CA bundles — the browser's trust store decides TLS),
-- `GrpcConfig.channel_options`, including `grpc.max_send/receive_message_length`
-  (only `grpc-web.path_prefix` is consumed),
+- `GrpcConfig.channel_options`, including `grpc.max_send_message_length` /
+  `grpc.max_receive_message_length` (only `grpc-web.path_prefix` is consumed). The
+  practical message-size ceiling is the server's `grpcMaxMessageSize` (reported by
+  `/v1/meta`); exceeding it surfaces as `RESOURCE_EXHAUSTED`,
 - `Proxies.grpc` / `GRPC_PROXY`.
 
 ## CORS requirements (browsers)
 
-Cross-origin browser deployments must configure the grpc-web transcoder / REST endpoint
-with CORS, or failures become hard to diagnose:
+Weaviate ≥ 1.38.3 serves the CORS headers below for its `/v1/grpc-web` endpoint itself,
+with no configuration. Its request-header list is a **closed allowlist**: custom
+`headers={...}` that are not on it fail the browser's preflight. Cross-origin
+deployments that go through a grpc-web transcoder or a proxy must configure CORS there:
 
-- allow the request headers the client sends: `authorization`, `content-type`,
-  `x-grpc-web`, and any custom headers;
+- allow every request header the client sends: `content-type`, `x-grpc-web`,
+  `x-user-agent`, `grpc-timeout`, `x-weaviate-client`, `authorization` (when auth is
+  used) and `x-weaviate-cluster-url` (Weaviate Cloud);
 - expose the grpc-web status headers on responses:
   `Access-Control-Expose-Headers: grpc-status, grpc-message` — without this,
   trailers-only error responses (e.g. a bad API key) are reported as
@@ -106,3 +139,5 @@ with CORS, or failures become hard to diagnose:
 interpreter (run it in a fresh process, before importing `weaviate`). Inject a sender
 with `weaviate_client_web.set_sender(...)` (e.g. `make_httpx_sender()`) to exercise the
 transport against an Envoy/vanguard transcoder without a browser.
+`install_fetch_transport(force=True)` likewise patches httpx on CPython, given an
+importable `pyodide.http` stand-in.
diff --git a/packages/web/src/weaviate_client_web/__init__.py b/packages/web/src/weaviate_client_web/__init__.py
index f0a481d19..24820d41d 100644
--- a/packages/web/src/weaviate_client_web/__init__.py
+++ b/packages/web/src/weaviate_client_web/__init__.py
@@ -3,16 +3,26 @@
 Under Pyodide/Emscripten there is no ``grpcio`` wheel. Importing this package installs a
 pure-Python ``grpc`` shim into ``sys.modules`` (and forces the pure-Python protobuf
 runtime) so that the subsequent ``import weaviate`` succeeds and its async gRPC data path
-runs over grpc-web (``fetch``) instead of HTTP/2 sockets.
+runs over grpc-web (``fetch``) instead of HTTP/2 sockets; REST runs through the package's
+own ``fetch``-based httpx transport.
 
-Usage under Pyodide (with this package installed, a bare ``import weaviate`` suffices —
+Usage under Pyodide against Weaviate >= 1.38.3, which serves grpc-web on its REST port
+under ``/v1/grpc-web`` (with this package installed, a bare ``import weaviate`` suffices —
 the base client imports this package itself under Emscripten before anything else)::
 
     import weaviate
 
-    client = weaviate.use_async_with_local(skip_init_checks=True)
+    client = weaviate.use_async_with_custom(
+        http_host="localhost", http_port=8080, http_secure=False,
+        grpc_host="localhost", grpc_port=8080, grpc_secure=False,
+        grpc_path_prefix="/v1/grpc-web",
+    )
     await client.connect()
 
+``use_async_with_local`` / ``use_async_with_weaviate_cloud`` have no ``grpc_path_prefix``
+and therefore only work with a grpc-web transcoder (Envoy, vanguard) at the root of
+``grpc_host:grpc_port``.
+
 An explicit ``import weaviate_client_web`` before ``import weaviate`` also works and
 remains the explicit form. The shim is installed automatically only under Emscripten, so
 importing this package on a normal CPython install never clobbers a real, working
diff --git a/packages/web/src/weaviate_client_web/_channel.py b/packages/web/src/weaviate_client_web/_channel.py
index c3ba77837..d6dd4b7cc 100644
--- a/packages/web/src/weaviate_client_web/_channel.py
+++ b/packages/web/src/weaviate_client_web/_channel.py
@@ -15,10 +15,11 @@
 import asyncio
 import base64
 import math
+import sys
 import urllib.parse
 from typing import Any, Callable, Dict, List, Optional
 
-from ._framing import encode_message, split_response
+from ._framing import TruncatedFrameError, UnknownFrameFlagError, encode_message, split_response
 from ._sender import Sender, pyfetch_sender
 from ._shim import AioChannel, AioRpcError, StatusCode, status_from_int
 
@@ -36,14 +37,26 @@ def get_sender() -> Sender:
     return _default_sender
 
 
-def _encode_timeout(seconds: float) -> str:
-    """Encode a timeout as a grpc-timeout header value (``<positive int><unit>``)."""
-    # Round up so we never advertise a shorter deadline than requested (which would risk
-    # premature server-side cancellation); grpc-timeout takes a positive integer + unit.
-    millis = max(1, math.ceil(seconds * 1000))
-    if millis < 100_000_000:
-        return f"{millis}m"
-    return f"{max(1, math.ceil(seconds))}S"
+# grpc-timeout is at most 8 digits plus a unit; anything longer is rejected by the server.
+_GRPC_TIMEOUT_MAX = 100_000_000
+
+
+def _encode_timeout(seconds: Optional[float]) -> Optional[str]:
+    """Encode a timeout as a grpc-timeout header value; ``None`` means no deadline.
+
+    ``None``, non-finite values and anything beyond 99,999,999 minutes (~190 years) carry
+    no deadline. Rounds up so we never advertise a shorter deadline than requested (which
+    would risk premature server-side cancellation), moving to a coarser unit
+    (m -> S -> M) to stay within 8 digits. Hours are never emitted: grpc-web transcoders
+    (vanguard) reject any H value above 8H with HTTP 400.
+    """
+    if seconds is None or not math.isfinite(seconds):
+        return None
+    for amount, unit in ((seconds * 1000, "m"), (seconds, "S"), (seconds / 60, "M")):
+        value = max(1, math.ceil(amount))
+        if value < _GRPC_TIMEOUT_MAX:
+            return f"{value}{unit}"
+    return None
 
 
 def _fold_metadata(headers: Dict[str, str], metadata: Any) -> None:
@@ -57,9 +70,13 @@ def _fold_metadata(headers: Dict[str, str], metadata: Any) -> None:
         name = key.lower()
         if name.endswith("-bin"):
             raw = value if isinstance(value, (bytes, bytearray)) else str(value).encode()
-            headers[name] = base64.b64encode(raw).decode("ascii")
+            text = base64.b64encode(raw).decode("ascii")
         else:
-            headers[name] = value if isinstance(value, str) else str(value)
+            text = value if isinstance(value, str) else str(value)
+        # This path bypasses h11/grpcio's header validation, so keep their defence here.
+        if any(c in name or c in text for c in ("\r", "\n", "\0")):
+            raise ValueError(f"Illegal character in gRPC metadata {name!r}")
+        headers[name] = text
 
 
 def _header_lookup(headers: Dict[str, str], name: str) -> Optional[str]:
@@ -106,8 +123,8 @@ async def __call__(
 class _UnsupportedStreamMultiCallable:
     """Placeholder for ``stream_stream`` (bidirectional streaming).
 
-    Calling it raises immediately, before the ``async for`` in ``connect/v4.py:1243``
-    begins iterating.
+    Calling it raises immediately, before the ``async for`` in ``connect/v4.py`` begins
+    iterating.
     """
 
     def __init__(self, path: str) -> None:
@@ -181,16 +198,20 @@ async def _unary(
             "x-user-agent": "weaviate-client-web",
         }
         _fold_metadata(headers, metadata)
-        if timeout is not None:
-            headers["grpc-timeout"] = _encode_timeout(timeout)
+        grpc_timeout = _encode_timeout(timeout)
+        if grpc_timeout is None:
+            timeout = None  # None / non-finite: no deadline, server- or client-side
+        else:
+            headers["grpc-timeout"] = grpc_timeout
 
         url = self._base_url + self._path_prefix + path
         framed = encode_message(payload)
 
         # Send. Enforce a client-side deadline (the grpc-timeout header is server-side
         # only; pyfetch ignores its timeout arg, so without this a stalled request could
-        # hang forever). Any transport/parse failure is surfaced as AioRpcError so the
-        # client only ever sees gRPC error types (never a bare ValueError/TimeoutError).
+        # hang forever). Any transport/parse failure is surfaced as AioRpcError; the only
+        # non-gRPC error a caller can see is the ValueError from metadata validation
+        # above, raised before any I/O (as native grpcio does).
         try:
             send = self._sender(url, headers, framed, timeout)
             if timeout is not None:
@@ -208,10 +229,10 @@ async def _unary(
             # str() of transport errors can be empty (e.g. httpx.ConnectError) — always
             # include the exception type so failures stay diagnosable
             detail = f"{type(exc).__name__}: {exc}" if str(exc) else repr(exc)
-            raise AioRpcError(
-                code=StatusCode.UNAVAILABLE,
-                details=f"grpc-web transport error for {path}: {detail}",
-            ) from exc
+            details = f"grpc-web transport error for {path}: {detail}"
+            if not self._path_prefix and sys.platform == "emscripten":
+                details += " " + _no_path_prefix_hint()
+            raise AioRpcError(code=StatusCode.UNAVAILABLE, details=details) from exc
 
         try:
             return self._handle_response(status, resp_headers, body, deserialize, url)
@@ -231,11 +252,9 @@ def _handle_response(
         deserialize: Callable[[bytes], Any],
         url: str = "",
     ) -> Any:
-        # Frame-parse defensively, and never let a parse failure decide the outcome:
-        # every real error response carries a body that is NOT grpc-web framing
-        # (Weaviate's 404 JSON, an nginx error page), so reading '{' or '<' as a frame
-        # flag would report "malformed grpc-web response" and throw away the HTTP
-        # status, the URL and the server's own explanation.
+        # A frame-parse failure must never decide the outcome by itself: real error
+        # responses carry non-grpc-web bodies (Weaviate's 404 JSON, an nginx page), and
+        # the HTTP status, URL and the server's own text must survive into the error.
         messages: List[bytes] = []
         trailers: Dict[str, str] = {}
         frame_error: Optional[BaseException] = None
@@ -257,7 +276,7 @@ def _handle_response(
             # No grpc-status anywhere AND either a non-200 or a body that is not
             # grpc-web framing: a gRPC service did not answer this request at all.
             if http_status != 200 or frame_error is not None:
-                raise _non_grpc_web_error(http_status, url, body, frame_error)
+                raise _frame_error_to_rpc(http_status, url, body, frame_error)
             if messages:
                 # Every grpc-web unary response must carry a grpc-status (trailer frame
                 # or header); a proxy that drops the trailer must not read as success.
@@ -274,7 +293,12 @@ def _handle_response(
         if frame_error is not None:
             # grpc-status said OK but the body will not parse — report what actually
             # came back rather than a bare "no message frame".
-            raise _non_grpc_web_error(http_status, url, body, frame_error)
+            raise _frame_error_to_rpc(http_status, url, body, frame_error)
+        if len(messages) > 1:
+            raise AioRpcError(
+                code=StatusCode.INTERNAL,
+                details=f"unary grpc-web response carried {len(messages)} message frames",
+            )
         if not messages:
             details = "grpc-web response contained no message frame"
             if raw_status is None:
@@ -309,22 +333,53 @@ def _body_excerpt(body: bytes, limit: int = _BODY_EXCERPT_LIMIT) -> str:
     return text + ("..." if len(body) > limit else "")
 
 
+def _no_path_prefix_hint() -> str:
+    # Lazy import: this module is imported while ``weaviate/__init__`` is still
+    # bootstrapping the shim under Emscripten.
+    from weaviate.exceptions import GRPC_WEB_MIN_SERVER_VERSION, GRPC_WEB_SERVER_PATH_PREFIX
+
+    return (
+        "(no grpc_path_prefix set — for Weaviate >= "
+        f"{GRPC_WEB_MIN_SERVER_VERSION} use use_async_with_custom(..., "
+        f"grpc_path_prefix='{GRPC_WEB_SERVER_PATH_PREFIX}') on the REST port, or point "
+        "grpc_host/grpc_port at a grpc-web transcoder)"
+    )
+
+
+def _frame_error_to_rpc(
+    http_status: int, url: str, body: bytes, frame_error: Optional[BaseException]
+) -> AioRpcError:
+    """Choose the error for a body that did not parse as grpc-web frames."""
+    if http_status != 200 or isinstance(frame_error, (UnknownFrameFlagError, TruncatedFrameError)):
+        return _non_grpc_web_error(http_status, url, body, frame_error)
+    # Well-formed grpc-web up to the point of failure: a grpc-web endpoint answered but
+    # broke the protocol (compressed frame, message after trailer, …).
+    return AioRpcError(
+        code=StatusCode.INTERNAL,
+        details=f"malformed grpc-web response from {url or '<unknown url>'}: {frame_error}",
+    )
+
+
 def _non_grpc_web_error(
     http_status: int,
     url: str,
     body: bytes,
     frame_error: Optional[BaseException] = None,
 ) -> AioRpcError:
-    """Build the error for a response that is not a grpc-web response at all.
+    """Build the error for a response that is not a usable grpc-web response.
 
     Details always begin with ``HTTP <status>`` and carry the request URL plus a body
     excerpt (``weaviate/connect`` matches on that prefix). The status alone rarely
     separates "endpoint missing" from "proxy misconfigured"; the server's own body
     text usually does.
     """
+    truncated = isinstance(frame_error, TruncatedFrameError)
     what = "not a grpc-web response"
     if http_status == 200 and frame_error is not None:
-        what = f"the body is not grpc-web framing ({frame_error})"
+        if truncated:
+            what = f"the grpc-web body is truncated ({frame_error})"
+        else:
+            what = f"the body is not grpc-web framing ({frame_error})"
     parts = [f"HTTP {http_status} from {url or '<unknown url>'}: {what}."]
 
     if http_status == 404:
@@ -336,8 +391,20 @@ def _non_grpc_web_error(
             "the configured grpc-web path prefix is wrong for the proxy in front of it. "
             "Weaviate's native prefix is '/v1/grpc-web'."
         )
+    elif http_status == 405:
+        # A 405 can only come from an existing HTTP route: the prefix points at one.
+        parts.append(
+            "An HTTP route answered instead of the grpc-web endpoint (method not "
+            "allowed): the configured grpc-web path prefix is wrong. Weaviate's native "
+            "prefix is '/v1/grpc-web'."
+        )
     elif http_status in (502, 503, 504):
         parts.append("Weaviate or the proxy in front of it is unavailable.")
+    elif http_status == 200 and truncated:
+        parts.append(
+            "The response was cut short — a proxy or browser buffering limit, or the "
+            "connection dropped mid-response."
+        )
     elif http_status == 200:
         parts.append(
             "Something other than a grpc-web endpoint answered — typically a proxy "
diff --git a/packages/web/src/weaviate_client_web/_framing.py b/packages/web/src/weaviate_client_web/_framing.py
index 4449d265b..7e6a3dfc8 100644
--- a/packages/web/src/weaviate_client_web/_framing.py
+++ b/packages/web/src/weaviate_client_web/_framing.py
@@ -19,9 +19,22 @@
 
 _FLAG_TRAILER = 0x80
 _FLAG_COMPRESSED = 0x01
+_KNOWN_FLAGS = _FLAG_TRAILER | _FLAG_COMPRESSED
 _HEADER = struct.Struct(">BI")  # 1 flag byte + 4-byte big-endian length
 
 
+class FrameError(ValueError):
+    """The body is not a well-formed grpc-web response."""
+
+
+class UnknownFrameFlagError(FrameError):
+    """A flag byte outside the grpc-web set: the body is not grpc-web framing (JSON, HTML, …)."""
+
+
+class TruncatedFrameError(FrameError):
+    """The body ends before the length its frame header announces."""
+
+
 def encode_message(payload: bytes) -> bytes:
     """Frame a single (uncompressed) protobuf payload for sending."""
     return _HEADER.pack(0x00, len(payload)) + payload
@@ -30,31 +43,41 @@ def encode_message(payload: bytes) -> bytes:
 def iter_frames(buf: bytes) -> Iterator[Tuple[int, bytes]]:
     """Yield ``(flag, payload)`` for each frame in a grpc-web response body."""
     off, n = 0, len(buf)
-    while off + 5 <= n:
-        flag, length = _HEADER.unpack_from(buf, off)
+    while off < n:
+        # Validate the flag before the length so a text body ('{', '<') is reported as
+        # non-grpc-web rather than as a truncated frame with a garbage length.
+        flag = buf[off]
+        if flag & ~_KNOWN_FLAGS:
+            raise UnknownFrameFlagError(f"unknown grpc-web frame flag 0x{flag:02x} at byte {off}")
+        if off + 5 > n:
+            raise TruncatedFrameError(f"truncated grpc-web frame header at byte {off}")
+        _, length = _HEADER.unpack_from(buf, off)
         off += 5
         if off + length > n:
-            raise ValueError("truncated grpc-web frame")
+            raise TruncatedFrameError(
+                f"truncated grpc-web frame: header announces {length} bytes, {n - off} remain"
+            )
         yield flag, buf[off : off + length]
         off += length
-    if off != n:
-        raise ValueError("trailing bytes after final grpc-web frame")
 
 
 def parse_trailers(raw: bytes) -> Dict[str, str]:
     """Parse a trailer frame payload into a lower-cased header dict.
 
-    Names are ASCII by spec, but values are decoded leniently: a proxy that does not
-    percent-encode ``grpc-message``, or a server error quoting a UTF-8 collection /
-    tenant / property name, puts raw non-ASCII bytes in the trailer. Failing here would
-    throw away the ``grpc-status`` that came with it.
+    Decoded leniently on both sides of the colon: a proxy that does not percent-encode
+    ``grpc-message``, or a server error quoting a UTF-8 collection / tenant / property
+    name, puts raw non-ASCII bytes in the trailer, and one odd key must not discard the
+    ``grpc-status`` travelling with it. Lines are CRLF-terminated by spec; bare LF is
+    accepted.
     """
     out: Dict[str, str] = {}
-    for line in raw.split(b"\r\n"):
+    for line in raw.split(b"\n"):
+        line = line.rstrip(b"\r")
         if not line:
             continue
         key, _, value = line.partition(b":")
-        out[key.strip().decode("ascii").lower()] = value.strip().decode("utf-8", "replace")
+        name = key.strip().decode("utf-8", "replace").lower()
+        out[name] = value.strip().decode("utf-8", "replace")
     return out
 
 
@@ -62,13 +85,17 @@ def split_response(body: bytes) -> Tuple[List[bytes], Dict[str, str]]:
     """Split a grpc-web response body into message payloads and trailers."""
     messages: List[bytes] = []
     trailers: Dict[str, str] = {}
+    seen_trailer = False
     for flag, payload in iter_frames(body):
         if flag & _FLAG_TRAILER:
             trailers.update(parse_trailers(payload))
+            seen_trailer = True
         elif flag & _FLAG_COMPRESSED:
-            raise ValueError(
+            raise FrameError(
                 "compressed grpc-web message frames are not supported by this transport"
             )
+        elif seen_trailer:
+            raise FrameError("message frame after the trailer frame")
         else:
             messages.append(payload)
     return messages, trailers
diff --git a/packages/web/src/weaviate_client_web/_httpx_fetch.py b/packages/web/src/weaviate_client_web/_httpx_fetch.py
index f09d6bfbe..e3f0ddf4a 100644
--- a/packages/web/src/weaviate_client_web/_httpx_fetch.py
+++ b/packages/web/src/weaviate_client_web/_httpx_fetch.py
@@ -11,11 +11,11 @@
 philosophy as the grpc shim in ``_shim.py``. Responses are fully buffered, which matches
 how the base client consumes them (JSON bodies, no streaming).
 
-NOTE: Pyodide >= 0.27 distributes a patched httpx whose ``AsyncHTTPTransport`` already
-routes through JS ``fetch`` natively (``httpx/_transports/jsfetch.py``) with streaming
-support and a proper connect/read timeout split. When that build is detected, installing
-is skipped — overwriting it would replace a better implementation. This transport is the
-fallback for environments where httpx resolved from PyPI (httpcore + raw sockets).
+It installs under Emscripten even when Pyodide's bundled httpx carries its own JS-fetch
+transport (``httpx/_transports/jsfetch.py``): that transport reads ``Response.body``
+unconditionally, which is ``null`` for HEAD requests and 204 responses (``data.exists``,
+``data.delete_by_id``, ``tenants.exists``, …), and it does not enforce the per-request
+read timeout end-to-end.
 
 Known divergences from native httpx (acceptable for the weaviate client's usage):
 - the browser's fetch follows redirects internally, so httpx never sees a 3xx;
@@ -23,7 +23,7 @@
 - responses are fully buffered (no streaming).
 """
 
-import importlib.util
+import math
 import sys
 from typing import Callable, Dict, Optional
 
@@ -56,6 +56,10 @@
 
 _TIMEOUT_HINTS = ("timeout", "timed out", "abort")
 
+# JS timers take a signed 32-bit millisecond delay; anything larger overflows and fires
+# immediately, so a huge timeout would abort every request at once.
+_MAX_ABORT_SIGNAL_MS = 2**31 - 1
+
 
 async def _read_request_body(request: httpx.Request) -> bytes:
     try:
@@ -65,18 +69,25 @@ async def _read_request_body(request: httpx.Request) -> bytes:
 
 
 def _pick_timeout(request: httpx.Request) -> Optional[float]:
-    """Pick the effective deadline from httpx's timeout extension.
+    """Pick the request deadline from httpx's timeout extension: the ``read`` value only.
 
-    httpx populates ``extensions['timeout']`` with connect/read/write/pool values; the
-    read timeout is what the weaviate client configures per request. ``get(...) or``
-    chains would silently skip an explicit 0, so check for None instead.
+    The base client sets ``read`` per request and passes ``read=None`` for "no deadline"
+    while still carrying a ``pool`` value; ``connect`` is not separable under fetch and a
+    pool-acquire timeout has no meaning there, so neither may stand in for ``read``.
     """
     timeouts = request.extensions.get("timeout") or {}
-    for key in ("read", "connect", "pool"):
-        value = timeouts.get(key)
-        if value is not None:
-            return value
-    return None
+    return timeouts.get("read")
+
+
+def _abort_signal_ms(timeout: Optional[float]) -> Optional[int]:
+    """Milliseconds for ``AbortSignal.timeout``; ``None`` means no client-side deadline.
+
+    Zero, negative and non-finite timeouts all mean "no deadline". Rounded up so a
+    sub-millisecond timeout never becomes an immediate abort.
+    """
+    if timeout is None or not math.isfinite(timeout) or timeout <= 0:
+        return None
+    return min(math.ceil(timeout * 1000), _MAX_ABORT_SIGNAL_MS)
 
 
 def _map_fetch_error(
@@ -121,19 +132,21 @@ async def _fetch_handle_async_request(
         # fetch rejects GET/HEAD requests that carry a body
         kwargs["body"] = body
 
-    timeout = _pick_timeout(request)
     deadline_set = False
-    if timeout is not None and timeout > 0:
+    deadline_ms = _abort_signal_ms(_pick_timeout(request))
+    if deadline_ms is not None:
         try:
             from js import AbortSignal  # type: ignore[import-not-found]
 
-            kwargs["signal"] = AbortSignal.timeout(int(timeout * 1000))
+            kwargs["signal"] = AbortSignal.timeout(deadline_ms)
             deadline_set = True
         except Exception:  # pragma: no cover - AbortSignal.timeout availability varies
             pass
 
     try:
         response = await pyfetch(str(request.url), method=request.method, headers=headers, **kwargs)
+        # A body-less response (HEAD, 204) reads as b"": fetch resolves a null body to
+        # an empty ArrayBuffer.
         data = await response.bytes()
     except OSError as e:  # incl. pyodide.http.AbortError
         raise _map_fetch_error(e, request, deadline_set) from e
@@ -146,10 +159,12 @@ async def _fetch_handle_async_request(
         }
     except Exception:  # pragma: no cover - header shape varies across Pyodide versions
         resp_headers = {}
+    # Hand httpx an unread stream, as its own transports do: the client reads it and
+    # only then stamps ``response.elapsed``, which the batch-references path relies on.
     return httpx.Response(
         status_code=int(response.status),
         headers=resp_headers,
-        content=data,
+        stream=httpx.ByteStream(data),
         request=request,
     )
 
@@ -158,34 +173,17 @@ async def _fetch_handle_async_request(
 _fetch_handle_async_request.__weaviate_fetch_shim__ = True  # type: ignore[attr-defined]
 
 
-def _platform_httpx_has_fetch_support() -> bool:
-    """True when the running httpx is Pyodide's distributed build.
-
-    That build replaces the httpcore transport with a native JS-fetch one
-    (httpx/_transports/jsfetch.py), so the weaviate REST path already works without
-    this shim — and works better (streaming, connect/read timeout split).
-    """
-    try:
-        return importlib.util.find_spec("httpx._transports.jsfetch") is not None
-    except (ImportError, ValueError):  # pragma: no cover - exotic import states
-        return False
-
-
 def install_fetch_transport(force: bool = False) -> None:
     """Patch ``httpx.AsyncHTTPTransport`` to send requests through ``fetch``.
 
     Installs only under Emscripten unless ``force=True`` (CPython testing, where a
-    ``pyodide`` stub must be importable), and is skipped when httpx itself already has
-    fetch support (Pyodide's distributed build). Idempotent.
+    ``pyodide`` stub must be importable). Idempotent.
     """
     global _installed, _original_handle_async_request
     if _installed:
         return
-    if not force:
-        if sys.platform != "emscripten":
-            return
-        if _platform_httpx_has_fetch_support():
-            return
+    if not force and sys.platform != "emscripten":
+        return
     # Fail fast: the handler imports pyfetch per request, so a missing pyodide module
     # would otherwise surface as a confusing ModuleNotFoundError on the first request.
     from pyodide.http import pyfetch  # type: ignore[import-not-found]  # noqa: F401
diff --git a/packages/web/src/weaviate_client_web/_shim.py b/packages/web/src/weaviate_client_web/_shim.py
index dbf0d79c6..b04b8837a 100644
--- a/packages/web/src/weaviate_client_web/_shim.py
+++ b/packages/web/src/weaviate_client_web/_shim.py
@@ -8,14 +8,14 @@
 
 1. **Import surface** — every ``import grpc`` / ``from grpc(.aio) import ...`` executed
    while ``weaviate`` and its generated ``*_pb2_grpc`` stubs are imported
-   (``weaviate/config.py:4-5``, ``exceptions.py:7-8``, ``retry.py:5-6``,
-   ``connect/base.py:5-8``, ``connect/v4.py:24,29-32``, and the v6300 stub's
-   ``grpc.__version__`` / ``grpc._utilities.first_version_is_lower`` version gate).
+   (``weaviate/config.py``, ``exceptions.py``, ``retry.py``, ``connect/base.py``,
+   ``connect/v4.py``, and the v6300 stub's ``grpc.__version__`` /
+   ``grpc._utilities.first_version_is_lower`` version gate).
 2. **Runtime type contract** — :class:`AioChannel` becomes ``grpc.aio.Channel`` so the
    real grpc-web channel (which subclasses it) passes the
-   ``isinstance(..., grpc.aio.Channel)`` assertions in ``connect/v4.py`` (lines 722,
-   1241); :class:`AioRpcError` is the error the client catches and inspects via
-   ``.code()`` / ``.details()`` (``exceptions.py:62-76``, ``retry.py:30-31``).
+   ``isinstance(..., grpc.aio.Channel)`` assertions in ``connect/v4.py``;
+   :class:`AioRpcError` is the error the client catches and inspects via ``.code()`` /
+   ``.details()`` (``exceptions.py``, ``retry.py``).
 """
 
 import enum
@@ -34,8 +34,8 @@ class StatusCode(enum.Enum):
     """Mirror of ``grpc.StatusCode``.
 
     ``value`` is the canonical ``(int, str)`` tuple, matching grpcio so ``code.value[0]``
-    / ``code.value[1]`` (``exceptions.py:63,66``) and ``code.name``
-    (``connect/v4.py:1189``) behave identically.
+    / ``code.value[1]`` (``exceptions.py``) and ``code.name`` (``connect/v4.py``) behave
+    identically.
     """
 
     OK = (0, "ok")
@@ -122,11 +122,11 @@ def debug_error_string(self) -> Optional[str]:
 
 
 class StreamStreamCall:
-    """Stand-in for ``grpc.aio.StreamStreamCall`` (imported as a type at ``v4.py:31``)."""
+    """Stand-in for ``grpc.aio.StreamStreamCall`` (imported as a type by ``connect/v4.py``)."""
 
 
 class ChannelCredentials:
-    """Stand-in for ``grpc.ChannelCredentials`` (imported by ``config.py:4``)."""
+    """Stand-in for ``grpc.ChannelCredentials`` (imported by ``config.py``)."""
 
 
 def ssl_channel_credentials(*_args: Any, **_kwargs: Any) -> ChannelCredentials:
@@ -152,8 +152,8 @@ class AioChannel:
 def first_version_is_lower(_version: str, _other: str) -> bool:
     """Stand-in for ``grpc._utilities.first_version_is_lower``.
 
-    Returning ``False`` makes the v6300 stub's version gate
-    (``weaviate_pb2_grpc.py:17-29``) pass.
+    Returning ``False`` makes the v6300 stub's import-time version gate
+    (``weaviate_pb2_grpc.py``) pass.
     """
     return False
 
diff --git a/packages/web/tests/test_framing.py b/packages/web/tests/test_framing.py
index 5bcefe354..fd18a35a3 100644
--- a/packages/web/tests/test_framing.py
+++ b/packages/web/tests/test_framing.py
@@ -3,6 +3,9 @@
 import pytest
 
 from weaviate_client_web._framing import (
+    FrameError,
+    TruncatedFrameError,
+    UnknownFrameFlagError,
     encode_message,
     iter_frames,
     parse_trailers,
@@ -29,12 +32,20 @@ def test_split_response_message_and_trailer():
 
 
 def test_split_response_multiple_messages():
+    # the splitter returns every message frame; whether more than one is acceptable is
+    # the channel's decision (a unary RPC rejects it)
     body = _frame(b"a") + _frame(b"bb") + _frame(b"grpc-status:0\r\n", 0x80)
     messages, trailers = split_response(body)
     assert messages == [b"a", b"bb"]
     assert trailers["grpc-status"] == "0"
 
 
+def test_split_response_message_after_trailer_raises():
+    body = _frame(b"a") + _frame(b"grpc-status:0\r\n", 0x80) + _frame(b"late")
+    with pytest.raises(FrameError, match="after the trailer"):
+        split_response(body)
+
+
 def test_split_response_trailers_only():
     body = _frame(b"grpc-status:7\r\ngrpc-message:denied\r\n", 0x80)
     messages, trailers = split_response(body)
@@ -70,13 +81,40 @@ def test_split_response_survives_non_ascii_trailer():
     assert trailers["grpc-status"] == "7"
 
 
+def test_parse_trailers_accepts_lf_only_lines():
+    parsed = parse_trailers(b"grpc-status:0\ngrpc-message:ok\n")
+    assert parsed == {"grpc-status": "0", "grpc-message": "ok"}
+
+
+def test_parse_trailers_keeps_status_when_a_key_is_not_ascii():
+    # one odd key from a proxy must not throw away the whole block
+    parsed = parse_trailers("x-caf\u00e9:1\r\ngrpc-status:0\r\n".encode("utf-8"))
+    assert parsed["grpc-status"] == "0"
+    assert parsed["x-caf\u00e9"] == "1"
+
+
 def test_truncated_frame_raises():
     framed = encode_message(b"hello")[:-2]
-    with pytest.raises(ValueError):
+    with pytest.raises(TruncatedFrameError):
         list(iter_frames(framed))
+    with pytest.raises(TruncatedFrameError):
+        list(iter_frames(b"\x00\x00\x00"))  # shorter than one frame header
+
+
+@pytest.mark.parametrize("first_byte", [b"{", b"<", b"\x02", b"\x40", b"\xff"])
+def test_unknown_frame_flag_raises(first_byte):
+    # a JSON / HTML body, or a flag bit this transport does not know
+    body = first_byte + b"\x00\x00\x00\x01x"
+    with pytest.raises(UnknownFrameFlagError, match="unknown grpc-web frame flag"):
+        list(iter_frames(body))
+
+
+def test_frame_errors_are_value_errors():
+    assert issubclass(TruncatedFrameError, ValueError)
+    assert issubclass(UnknownFrameFlagError, ValueError)
 
 
 def test_compressed_message_frame_rejected():
     body = _frame(b"x", 0x01)
-    with pytest.raises(ValueError):
+    with pytest.raises(FrameError, match="compressed"):
         split_response(body)
diff --git a/packages/web/tests/test_httpx_fetch.py b/packages/web/tests/test_httpx_fetch.py
index 119c4bcf5..63480311e 100644
--- a/packages/web/tests/test_httpx_fetch.py
+++ b/packages/web/tests/test_httpx_fetch.py
@@ -18,19 +18,26 @@
 import httpx
 import pytest
 
-from weaviate_client_web._httpx_fetch import _fetch_handle_async_request
+from weaviate_client_web._httpx_fetch import (
+    _MAX_ABORT_SIGNAL_MS,
+    _abort_signal_ms,
+    _fetch_handle_async_request,
+)
 
 _SRC = str(pathlib.Path(__file__).resolve().parents[1] / "src")
 
 
 class FakeFetchResponse:
-    def __init__(self, status: int = 200, headers: Optional[Any] = None, body: bytes = b""):
+    def __init__(
+        self, status: int = 200, headers: Optional[Any] = None, body: Optional[bytes] = b""
+    ):
         self.status = status
         self.headers: Any = headers or {}
         self._body = body
 
     async def bytes(self) -> bytes:  # noqa: A003 - mirrors pyodide's FetchResponse API
-        return self._body
+        # a null JS body (HEAD, 204) resolves to an empty ArrayBuffer, i.e. b""
+        return b"" if self._body is None else self._body
 
 
 class FakePyfetch:
@@ -55,10 +62,31 @@ def fake_pyfetch(monkeypatch) -> FakePyfetch:
     return fetch
 
 
-def _handle(request: httpx.Request) -> httpx.Response:
+async def _handle_async(request: httpx.Request) -> httpx.Response:
     # self is unused by the handler implementation; a bare transport instance suffices
     transport = httpx.AsyncHTTPTransport.__new__(httpx.AsyncHTTPTransport)
-    return asyncio.run(_fetch_handle_async_request(transport, request))
+    response = await _fetch_handle_async_request(transport, request)
+    await response.aread()  # httpx.AsyncClient reads non-streamed responses the same way
+    return response
+
+
+def _handle(request: httpx.Request) -> httpx.Response:
+    return asyncio.run(_handle_async(request))
+
+
+class _FetchTransport(httpx.AsyncBaseTransport):
+    """Route an ``httpx.AsyncClient`` through the handler without patching httpx globally."""
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        return await _fetch_handle_async_request(self, request)  # type: ignore[arg-type]
+
+
+def _via_client(method: str, url: str, **kwargs: Any) -> httpx.Response:
+    async def main() -> httpx.Response:
+        async with httpx.AsyncClient(transport=_FetchTransport()) as client:
+            return await client.request(method, url, **kwargs)
+
+    return asyncio.run(main())
 
 
 def test_basic_get_round_trip(fake_pyfetch):
@@ -82,6 +110,44 @@ def test_response_has_request_attached_for_raise_for_status(fake_pyfetch):
         response.raise_for_status()
 
 
+@pytest.mark.parametrize("status", [204, 404])
+def test_head_response_without_body_yields_empty_content(fake_pyfetch, status):
+    # data.exists() / tenants.exists() are HEAD requests answered 204/404 with a null
+    # body; the transport must hand httpx an empty response, not fail on the missing body
+    fake_pyfetch.response = FakeFetchResponse(status=status, body=None)
+    response = _handle(httpx.Request("HEAD", "http://h:8080/v1/objects/A/uuid"))
+    assert response.status_code == status
+    assert response.content == b""
+    assert "body" not in fake_pyfetch.calls[0]
+
+
+def test_delete_204_without_body_yields_empty_content(fake_pyfetch):
+    # data.delete_by_id() / reference_delete() are answered 204 with a null body
+    fake_pyfetch.response = FakeFetchResponse(status=204, body=None)
+    response = _handle(httpx.Request("DELETE", "http://h:8080/v1/objects/A/uuid"))
+    assert response.status_code == 204
+    assert response.content == b""
+
+
+def test_body_less_response_through_async_client(fake_pyfetch):
+    # the full httpx.AsyncClient path (stream wrapping + read) on a body-less response
+    fake_pyfetch.response = FakeFetchResponse(status=204, body=None)
+    response = _via_client("HEAD", "http://h:8080/v1/objects/A/uuid")
+    assert response.status_code == 204
+    assert response.content == b""
+
+
+def test_response_through_async_client_exposes_elapsed_and_content(fake_pyfetch):
+    # the batch-references path reads ``res.elapsed``, which httpx only sets after it
+    # has read/closed a stream-backed response; a pre-loaded body never gets one
+    payload = b'[{"result": {"status": "SUCCESS"}}]'
+    fake_pyfetch.response = FakeFetchResponse(status=200, body=payload)
+    response = _via_client("POST", "http://h:8080/v1/batch/references", content=b"[]")
+    assert response.content == payload
+    assert response.json() == [{"result": {"status": "SUCCESS"}}]
+    assert response.elapsed.total_seconds() >= 0
+
+
 def test_fetch_managed_request_headers_stripped(fake_pyfetch):
     request = httpx.Request(
         "POST",
@@ -184,11 +250,22 @@ def test_read_timeout_maps_to_abort_signal_ms(fake_pyfetch, fake_abort_signal):
     assert fake_pyfetch.calls[0]["signal"] == "signal-30000"
 
 
-def test_timeout_falls_back_to_connect_then_pool(fake_pyfetch, fake_abort_signal):
+def test_read_none_means_no_deadline_even_with_pool_and_connect_set(
+    fake_pyfetch, fake_abort_signal
+):
+    # what the base client hands over for a non-finite request timeout: read=None with the
+    # session pool timeout still set; falling back to pool/connect would abort a long
+    # insert after 5 s
+    _handle(_request_with_timeout({"connect": None, "read": None, "write": None, "pool": 5}))
     _handle(_request_with_timeout({"connect": 2.0, "read": None, "write": None, "pool": 9.0}))
-    _handle(_request_with_timeout({"connect": None, "read": None, "write": None, "pool": 9.0}))
-    assert fake_abort_signal.timeouts == [2000, 9000]
-    assert [c["signal"] for c in fake_pyfetch.calls] == ["signal-2000", "signal-9000"]
+    assert fake_abort_signal.timeouts == []
+    assert all("signal" not in c for c in fake_pyfetch.calls)
+
+
+def test_read_timeout_alone_sets_the_deadline(fake_pyfetch, fake_abort_signal):
+    _handle(_request_with_timeout({"connect": None, "read": 7, "write": None, "pool": 5}))
+    assert fake_abort_signal.timeouts == [7000]
+    assert fake_pyfetch.calls[0]["signal"] == "signal-7000"
 
 
 def test_no_timeout_extension_sends_no_signal(fake_pyfetch, fake_abort_signal):
@@ -215,6 +292,40 @@ def test_zero_timeout_means_no_deadline(fake_pyfetch, fake_abort_signal):
     assert "signal" not in fake_pyfetch.calls[0]
 
 
+@pytest.mark.parametrize(
+    "timeout,expected_ms",
+    [
+        (None, None),
+        (0, None),
+        (-1, None),
+        (float("inf"), None),
+        (float("nan"), None),
+        (0.0001, 1),  # rounds up: never an immediate AbortSignal.timeout(0)
+        (30.0, 30_000),
+        (1e8, _MAX_ABORT_SIGNAL_MS),
+        (1e10, _MAX_ABORT_SIGNAL_MS),
+    ],
+)
+def test_abort_signal_ms_bounds(timeout, expected_ms):
+    assert _abort_signal_ms(timeout) == expected_ms
+
+
+def test_infinite_timeout_sends_no_signal(fake_pyfetch, fake_abort_signal):
+    # an inf read deadline reaching the transport: no signal, not an OverflowError
+    _handle(
+        _request_with_timeout({"connect": None, "read": float("inf"), "write": None, "pool": 5})
+    )
+    assert fake_abort_signal.timeouts == []
+    assert "signal" not in fake_pyfetch.calls[0]
+
+
+def test_huge_timeout_is_capped_to_int32_ms(fake_pyfetch, fake_abort_signal):
+    # setTimeout delays above 2^31-1 ms overflow and fire at once, aborting the request
+    _handle(_request_with_timeout({"connect": None, "read": 1e10, "write": None, "pool": None}))
+    assert fake_abort_signal.timeouts == [_MAX_ABORT_SIGNAL_MS]
+    assert fake_pyfetch.calls[0]["signal"] == f"signal-{_MAX_ABORT_SIGNAL_MS}"
+
+
 class RaisingPyfetch:
     def __init__(self, exc: BaseException):
         self.exc = exc
@@ -286,20 +397,6 @@ def test_crlf_in_header_value_rejected(fake_pyfetch):
     assert fake_pyfetch.calls == []
 
 
-def test_platform_jsfetch_detection(monkeypatch):
-    import importlib.machinery
-
-    from weaviate_client_web._httpx_fetch import _platform_httpx_has_fetch_support
-
-    # the dev environment runs PyPI httpx (httpcore-based): no jsfetch transport
-    assert _platform_httpx_has_fetch_support() is False
-
-    fake = types.ModuleType("httpx._transports.jsfetch")
-    fake.__spec__ = importlib.machinery.ModuleSpec("httpx._transports.jsfetch", loader=None)
-    monkeypatch.setitem(sys.modules, "httpx._transports.jsfetch", fake)
-    assert _platform_httpx_has_fetch_support() is True
-
-
 # ---------------------------------------------------------------------------
 # Install semantics: these patch httpx.AsyncHTTPTransport globally, so each
 # scenario runs in a fresh subprocess (same pattern as test_shim_install.py).
@@ -476,11 +573,12 @@ def test_force_install_without_pyodide_fails_fast():
     assert "OK" in result.stdout
 
 
-def test_emscripten_with_platform_jsfetch_skips_install():
-    # on Pyodide's distributed httpx (jsfetch transport built in), the shim must NOT
-    # overwrite the platform implementation
+def test_emscripten_installs_even_when_platform_httpx_has_jsfetch():
+    # Pyodide's bundled httpx ships a jsfetch transport that crashes on body-less
+    # responses (HEAD / 204); the shim must take over regardless of the httpx build
     result = _run(
-        """
+        prelude=_FAKE_PYODIDE_PRELUDE,
+        body="""
         import importlib.machinery, sys, types
 
         sys.platform = "emscripten"
@@ -493,11 +591,14 @@ def test_emscripten_with_platform_jsfetch_skips_install():
         import httpx
         before = httpx.AsyncHTTPTransport.handle_async_request
         from weaviate_client_web import install_fetch_transport, is_fetch_transport_installed
-        install_fetch_transport()  # no force: platform transport must win
-        assert not is_fetch_transport_installed()
-        assert httpx.AsyncHTTPTransport.handle_async_request is before
+        install_fetch_transport()  # no force: the platform alone must trigger it
+        assert is_fetch_transport_installed()
+        assert httpx.AsyncHTTPTransport.handle_async_request is not before
+        assert getattr(
+            httpx.AsyncHTTPTransport.handle_async_request, "__weaviate_fetch_shim__", False
+        ) is True
         print("OK")
-        """
+        """,
     )
     assert result.returncode == 0, result.stderr
     assert "OK" in result.stdout
diff --git a/packages/web/tests/test_single_import.py b/packages/web/tests/test_single_import.py
index aff239c0d..562f31e0a 100644
--- a/packages/web/tests/test_single_import.py
+++ b/packages/web/tests/test_single_import.py
@@ -23,6 +23,19 @@
 sysconfig.get_config_vars()
 """
 
+# The companion's bootstrap installs the fetch transport under Emscripten and fails fast
+# if pyodide.http cannot be imported, so a faked platform needs a stand-in module.
+_FAKE_PYODIDE = """
+import types
+
+_pyodide = types.ModuleType("pyodide")
+_http = types.ModuleType("pyodide.http")
+_http.pyfetch = None
+_pyodide.http = _http
+sys.modules["pyodide"] = _pyodide
+sys.modules["pyodide.http"] = _http
+"""
+
 
 def _run(
     body: str, *, prelude: str = "", path_entry: str = _SRC, no_site: bool = False
@@ -36,22 +49,16 @@ def _run(
 
 def test_bare_import_weaviate_installs_shim_under_emscripten():
     result = _run(
-        prelude=_PRIME_SYSCONFIG,
+        prelude=_PRIME_SYSCONFIG + _FAKE_PYODIDE,
         body="""
-        import importlib.machinery, types
-
         sys.platform = "emscripten"
-        # Pretend httpx is Pyodide's jsfetch build so the companion's bootstrap skips
-        # the fetch-transport install (there is no pyodide module on CPython).
-        fake = types.ModuleType("httpx._transports.jsfetch")
-        fake.__spec__ = importlib.machinery.ModuleSpec("httpx._transports.jsfetch", loader=None)
-        sys.modules["httpx._transports.jsfetch"] = fake
 
         import weaviate  # the ONLY weaviate-side import: must bootstrap the companion
 
         assert "weaviate_client_web" in sys.modules, "hook did not import the companion"
         import weaviate_client_web
         assert weaviate_client_web.is_installed()
+        assert weaviate_client_web.is_fetch_transport_installed()
         import grpc
         assert getattr(grpc, "__weaviate_client_web_shim__", False) is True
         print("OK")
@@ -101,3 +108,32 @@ def test_bare_import_with_grpc_present_falls_through_silently():
     )
     assert result.returncode == 0, result.stderr
     assert "OK" in result.stdout
+
+
+def test_bare_import_with_broken_companion_surfaces_its_own_error(tmp_path):
+    # An INSTALLED companion whose import fails (here: a missing dependency of its own)
+    # must raise that error, not the "install weaviate-client-web" hint — the hint would
+    # send the user to reinstall a package that is already there.
+    fake_pkg = tmp_path / "weaviate_client_web"
+    fake_pkg.mkdir()
+    (fake_pkg / "__init__.py").write_text(
+        "raise ModuleNotFoundError(\"No module named 'anyio'\", name='anyio')\n"
+    )
+    result = _run(
+        prelude=_PRIME_SYSCONFIG,
+        body="""
+        sys.platform = "emscripten"
+        try:
+            import weaviate
+        except ImportError as e:
+            assert e.name == "anyio", (e.name, str(e))
+            assert "anyio" in str(e), str(e)
+            assert "weaviate-client-web" not in str(e), str(e)
+            print("OK")
+        else:
+            raise AssertionError("expected the companion's own ImportError to surface")
+        """,
+        path_entry=str(tmp_path),
+    )
+    assert result.returncode == 0, result.stderr
+    assert "OK" in result.stdout
diff --git a/packages/web/tests/test_transport.py b/packages/web/tests/test_transport.py
index 3c7d30dcc..02871fa69 100644
--- a/packages/web/tests/test_transport.py
+++ b/packages/web/tests/test_transport.py
@@ -7,11 +7,17 @@
 
 import asyncio
 import struct
+import sys
 from typing import Dict, List, Optional, Tuple
 
 import pytest
 
-from weaviate_client_web._channel import GrpcWebChannel, set_sender
+from weaviate_client_web._channel import (
+    GrpcWebChannel,
+    _body_excerpt,
+    _encode_timeout,
+    set_sender,
+)
 from weaviate_client_web._shim import AioChannel, AioRpcError, StatusCode
 
 
@@ -80,7 +86,7 @@ def test_health_call_without_metadata():
     sender = FakeSender(body=_ok_response(b"pong"))
     channel = _channel(sender)
     mc = channel.unary_unary("/grpc.health.v1.Health/Check", lambda x: x, lambda b: b)
-    # mirrors connect/v4.py:316 — request + timeout, no metadata
+    # mirrors the health check in connect/v4.py — request + timeout, no metadata
     assert asyncio.run(mc(b"ping", timeout=2)) == b"pong"
 
 
@@ -118,9 +124,8 @@ def test_trailers_only_status_in_http_headers():
 # --- non-grpc-web responses -------------------------------------------------------
 #
 # Every real error response carries a body, and none of them is grpc-web framing. The
-# bodies below are verbatim shapes seen in the wild; an empty error body (the old
-# fixture) is the one shape no server or proxy produces, which is why these tests used
-# to pass against a code path that never ran.
+# bodies below are verbatim shapes seen in the wild (an empty error body is the one
+# shape no server or proxy produces).
 
 # Weaviate's own 404, verbatim from a 1.39.0 server asked for the wrong prefix.
 WEAVIATE_404_JSON = (
@@ -192,6 +197,61 @@ def test_nginx_404_html_is_reported_as_an_http_404():
     assert "malformed grpc-web response" not in err.details()
 
 
+def test_405_names_the_wrong_prefix():
+    # a 405 can only come from an existing HTTP route (measured live: a prefix pointing
+    # at /v1/objects answers "method POST is not allowed"), so the prefix is wrong
+    err = _details_of(405, b'{"code":405,"message":"method POST is not allowed, but [GET] are"}')
+    assert err.details().startswith("HTTP 405 ")
+    assert "path prefix" in err.details()
+    assert "/v1/grpc-web" in err.details()
+    assert "method POST is not allowed" in err.details()
+
+
+def test_truncated_grpc_web_body_is_reported_as_truncated_not_as_wrong_prefix():
+    # a valid frame header whose payload was cut short: the endpoint IS grpc-web, so the
+    # SPA / path-prefix hint would send the user the wrong way
+    body = _ok_response(b"reply-bytes")[:-6]
+    err = _details_of(200, body)
+    assert err.code() is StatusCode.INTERNAL
+    assert "truncated" in err.details()
+    assert "cut short" in err.details()
+    assert "single-page-app" not in err.details()
+    assert "path prefix" not in err.details()
+
+
+def test_spa_html_body_is_not_reported_as_truncated():
+    # text bodies decode to an unknown flag byte and a garbage length; they must read as
+    # "not grpc-web framing", never as a truncated grpc-web body
+    err = _details_of(200, SPA_INDEX_HTML)
+    assert "truncated" not in err.details()
+    assert "not grpc-web framing" in err.details()
+
+
+def test_message_frame_after_trailer_is_internal():
+    body = _frame(b"a") + _frame(b"grpc-status:0\r\n", 0x80) + _frame(b"late")
+    err = _details_of(200, body)
+    assert err.code() is StatusCode.INTERNAL
+    assert "malformed grpc-web response" in err.details()
+    assert "after the trailer" in err.details()
+    assert "single-page-app" not in err.details()
+
+
+def test_multiple_message_frames_in_unary_response_is_internal():
+    # a unary RPC has exactly one message; silently taking the first would hide a proxy
+    # or server that streams several
+    body = _frame(b"a") + _frame(b"bb") + _frame(b"grpc-status:0\r\n", 0x80)
+    err = _details_of(200, body)
+    assert err.code() is StatusCode.INTERNAL
+    assert "2 message frames" in err.details()
+
+
+def test_error_status_wins_over_multiple_message_frames():
+    body = _frame(b"a") + _frame(b"bb") + _frame(b"grpc-status:5\r\ngrpc-message:gone\r\n", 0x80)
+    err = _details_of(200, body)
+    assert err.code() is StatusCode.NOT_FOUND
+    assert err.details() == "gone"
+
+
 def test_spa_fallback_html_200_is_distinguishable_from_a_404():
     # An HTTP 200 serving index.html is the other half of a wrong path prefix: the app's
     # catch-all route answers instead of Weaviate. It must not read as malformed framing.
@@ -396,6 +456,110 @@ def test_grpc_timeout_header_rounds_up():
     assert sender.calls[0][1]["grpc-timeout"] == "124m"
 
 
+def test_body_excerpt_empty_and_non_printable():
+    assert _body_excerpt(b"") == "<empty>"
+    assert _body_excerpt(b"\x00\x01\x02\x7f") == "<4 non-printable bytes>"
+    assert _body_excerpt(b"ok\x00\x01") == "ok"
+
+
+@pytest.mark.parametrize(
+    "seconds,expected",
+    [
+        (None, None),
+        (float("inf"), None),
+        (float("nan"), None),
+        (0, "1m"),
+        (0.1234, "124m"),
+        (5, "5000m"),
+        (99_999, "99999000m"),
+        (100_000, "100000S"),  # 1e8 ms would be 9 digits
+        (1e8, "1666667M"),  # 1e8 s would be 9 digits
+        (1e9, "16666667M"),
+        (5_999_999_940, "99999999M"),  # the largest deadline that still fits in minutes
+        (1e10, None),  # would need hours, which transcoders reject above 8H: no deadline
+        (1e15, None),
+    ],
+)
+def test_encode_timeout_stays_within_eight_digits(seconds, expected):
+    encoded = _encode_timeout(seconds)
+    assert encoded == expected
+    if encoded is not None:
+        assert len(encoded) <= 9  # 8 digits + unit
+        assert not encoded.endswith("H")
+
+
+def test_infinite_timeout_sends_no_deadline():
+    # Timeout(query=inf): neither a grpc-timeout header nor a client-side wait
+    sender = FakeSender(body=_ok_response(b"x"))
+    channel = _channel(sender)
+    mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b)
+    assert asyncio.run(mc(b"q", timeout=float("inf"))) == b"x"
+    _, headers, _, timeout = sender.calls[0]
+    assert "grpc-timeout" not in headers
+    assert timeout is None
+
+
+def test_huge_timeout_uses_minutes_then_no_deadline():
+    # 1e8 s in milliseconds is 12 digits; the server rejects more than 8 ("timeout is
+    # too long", HTTP 400) and transcoders reject hour values above 8H, so past the
+    # minute range the request carries no deadline at all
+    sender = FakeSender(body=_ok_response(b"x"))
+    channel = _channel(sender)
+    mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b)
+    asyncio.run(mc(b"q", timeout=1e8))
+    asyncio.run(mc(b"q", timeout=1e9))
+    asyncio.run(mc(b"q", timeout=1e10))
+    assert sender.calls[0][1]["grpc-timeout"] == "1666667M"
+    assert sender.calls[1][1]["grpc-timeout"] == "16666667M"
+    assert "grpc-timeout" not in sender.calls[2][1]
+    assert sender.calls[2][3] is None  # no client-side wait either
+
+
+@pytest.mark.parametrize("bad", ["val\r\nx-injected: evil", "val\nx", "v\0"])
+def test_crlf_in_metadata_rejected(bad):
+    sender = FakeSender(body=_ok_response(b"x"))
+    channel = _channel(sender)
+    mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b)
+    with pytest.raises(ValueError, match="Illegal character"):
+        asyncio.run(mc(b"q", metadata=[("x-key", bad)]))
+    with pytest.raises(ValueError, match="Illegal character"):
+        asyncio.run(mc(b"q", metadata=[("x-key\r\n", "v")]))
+    assert sender.calls == []
+
+
+def _unavailable_details(monkeypatch, path_prefix, platform):
+    async def boom(url, headers, body, timeout):
+        raise ConnectionError("Failed to fetch")
+
+    monkeypatch.setattr(sys, "platform", platform)
+    channel = GrpcWebChannel("h:50051", secure=False, sender=boom, path_prefix=path_prefix)
+    mc = channel.unary_unary("/svc/M", lambda x: x, lambda b: b)
+    with pytest.raises(AioRpcError) as excinfo:
+        asyncio.run(mc(b"q"))
+    assert excinfo.value.code() is StatusCode.UNAVAILABLE
+    return excinfo.value.details()
+
+
+def test_unavailable_without_path_prefix_under_emscripten_hints_at_grpc_path_prefix(monkeypatch):
+    # use_async_with_local() under Pyodide builds a prefix-less channel aimed at the
+    # native gRPC port, which fetch cannot reach; the error must say what to do instead
+    details = _unavailable_details(monkeypatch, path_prefix="", platform="emscripten")
+    assert "grpc_path_prefix='/v1/grpc-web'" in details
+    assert "1.38.3" in details
+    assert "use_async_with_custom" in details
+
+
+def test_unavailable_with_path_prefix_has_no_prefix_hint(monkeypatch):
+    details = _unavailable_details(monkeypatch, path_prefix="/v1/grpc-web", platform="emscripten")
+    assert "no grpc_path_prefix" not in details
+
+
+def test_unavailable_without_path_prefix_off_emscripten_has_no_prefix_hint(monkeypatch):
+    # on CPython an empty prefix against a transcoder is the normal configuration
+    details = _unavailable_details(monkeypatch, path_prefix="", platform="linux")
+    assert "no grpc_path_prefix" not in details
+
+
 def test_close_is_awaitable_noop():
     channel = _channel(FakeSender())
     assert asyncio.run(channel.close()) is None
diff --git a/pyrightconfig.json b/pyrightconfig.json
index 396d62cfd..61eb3eaa3 100644
--- a/pyrightconfig.json
+++ b/pyrightconfig.json
@@ -1,6 +1,6 @@
 {
   "include": [
-    "weaviate", "integration"
+    "weaviate", "integration", "packages/web/src"
   ],
 
   "exclude": [
diff --git a/test/test_batch_async.py b/test/test_batch_async.py
index 5ad611b94..cc8a9a3f2 100644
--- a/test/test_batch_async.py
+++ b/test/test_batch_async.py
@@ -7,13 +7,16 @@
 """
 
 import asyncio
+import logging
+from typing import Optional
 
 import grpc
 import pytest
 
 from weaviate.collections.batch.async_ import _BatchBaseAsync, _BgTasks
 from weaviate.collections.batch.base import _BatchDataWrapper
-from weaviate.exceptions import WeaviateBatchStreamError
+from weaviate.collections.batch.batch_wrapper import _ContextManagerAsync
+from weaviate.exceptions import WeaviateBatchStreamError, _BatchStreamShutdownError
 
 
 class _NotAnException(BaseException):
@@ -155,3 +158,177 @@ class FakeConnection:
     with pytest.raises(RuntimeError, match="boom"):
         asyncio.run(batch._wait())
     assert backup.failed_objects == ["sentinel-failure"]
+
+
+async def _finished_task() -> "asyncio.Task[None]":
+    """A background task that returned normally (e.g. the server closed the stream)."""
+    task = asyncio.get_running_loop().create_task(asyncio.sleep(0))
+    await task
+    return task
+
+
+class _FakeTimeouts:
+    insert = 1
+
+
+class _FakeConnection:
+    timeout_config = _FakeTimeouts()
+
+
+def test_wait_names_unsent_data_when_the_tasks_ended_cleanly() -> None:
+    # both tasks returned normally (server closed the stream) with data still queued:
+    # that is not "the background tasks died unexpectedly", it is an early end of the
+    # stream — say what was left behind
+    async def run() -> None:
+        batch = _bare_batch(
+            bg_exception=None,
+            bg_tasks=_BgTasks(recv=await _finished_task(), loop=await _finished_task()),
+            connection=_FakeConnection(),
+            results_for_wrapper=_BatchDataWrapper(),
+            results_for_wrapper_backup=_BatchDataWrapper(),
+            batch_objects=[object(), object()],
+            batch_references=[object()],
+        )
+        await batch._wait()
+
+    with pytest.raises(
+        WeaviateBatchStreamError, match="ended with 2 objects and 1 references unsent"
+    ):
+        asyncio.run(run())
+
+
+def test_check_alive_after_a_clean_end_says_the_stream_ended() -> None:
+    async def run() -> None:
+        batch = _bare_batch(
+            bg_exception=None,
+            bg_tasks=_BgTasks(recv=await _finished_task(), loop=await _finished_task()),
+        )
+        getattr(batch, "_BatchBaseAsync__check_bg_tasks_alive")()  # noqa: B009
+
+    with pytest.raises(WeaviateBatchStreamError, match="stream has ended"):
+        asyncio.run(run())
+
+
+def test_put_gives_up_when_the_tasks_are_dead() -> None:
+    # a full queue with a dead receiver never drains: __put must return False instead of
+    # retrying (and recursing) once per second forever
+    async def run() -> bool:
+        reqs: asyncio.Queue = asyncio.Queue(maxsize=1)
+        await reqs.put(object())  # full
+        batch = _bare_batch(
+            reqs=reqs,
+            bg_exception=None,  # nothing recorded ...
+            shutdown_loop=asyncio.Event(),  # ... and no shutdown either
+            bg_tasks=_BgTasks(recv=await _dead_task("cancel"), loop=await _finished_task()),
+        )
+        put = getattr(batch, "_BatchBaseAsync__put")  # noqa: B009
+        return await asyncio.wait_for(put(object()), timeout=5)
+
+    assert asyncio.run(run()) is False
+
+
+def test_batch_stream_shutdown_error_is_in_the_taxonomy() -> None:
+    # raised on gRPC ABORTED and can surface from a clean `async with` exit
+    assert issubclass(_BatchStreamShutdownError, WeaviateBatchStreamError)
+    assert isinstance(_BatchStreamShutdownError(), Exception)
+
+
+class _FakeBatch:
+    """Stands in for _BatchBaseAsync behind the context manager."""
+
+    def __init__(self, wait_error: Optional[BaseException] = None) -> None:
+        self.wait_error = wait_error
+        self.shutdown_called = False
+        self.wait_called = False
+
+    async def _start(self) -> None:
+        pass
+
+    async def _shutdown(self) -> None:
+        self.shutdown_called = True
+
+    async def _wait(self) -> None:
+        self.wait_called = True
+        if self.wait_error is not None:
+            raise self.wait_error
+
+
+def test_aexit_raises_a_background_failure_on_a_clean_block() -> None:
+    fake = _FakeBatch(WeaviateBatchStreamError("bg died"))
+
+    async def run() -> None:
+        async with _ContextManagerAsync(fake):  # type: ignore[arg-type]
+            pass
+
+    with pytest.raises(WeaviateBatchStreamError, match="bg died"):
+        asyncio.run(run())
+
+
+def test_aexit_keeps_the_users_exception_over_a_background_failure(caplog) -> None:
+    # the block's own exception must not be REPLACED by the background failure (which
+    # used to demote it to __context__); the failure is logged instead
+    fake = _FakeBatch(WeaviateBatchStreamError("bg died"))
+
+    async def run() -> None:
+        async with _ContextManagerAsync(fake):  # type: ignore[arg-type]
+            raise ValueError("user code")
+
+    with caplog.at_level(logging.WARNING, logger="weaviate-client"):
+        with pytest.raises(ValueError, match="user code"):
+            asyncio.run(run())
+    assert fake.shutdown_called and fake.wait_called  # still drained/awaited
+    assert "bg died" in caplog.text
+
+
+def test_aexit_never_swallows_cancellation() -> None:
+    # a CancelledError leaving the block used to be replaced by the background failure,
+    # i.e. the cancellation was swallowed
+    fake = _FakeBatch(WeaviateBatchStreamError("bg died"))
+
+    async def run() -> None:
+        async with _ContextManagerAsync(fake):  # type: ignore[arg-type]
+            raise asyncio.CancelledError()
+
+    with pytest.raises(asyncio.CancelledError):
+        asyncio.run(run())
+    assert fake.wait_called
+
+
+def test_wait_with_an_infinite_insert_timeout_does_not_raise() -> None:
+    # Timeout(insert=inf) means "no deadline": the shutdown wait must not turn it into an
+    # error (the sync colour's Thread.join(inf) overflows; keep both colours consistent)
+    from weaviate.config import Timeout as TimeoutConfig
+
+    class InfiniteInsert:
+        timeout_config = TimeoutConfig(insert=float("inf"))
+
+    async def run() -> None:
+        batch = _bare_batch(
+            bg_exception=None,
+            bg_tasks=_BgTasks(recv=await _finished_task(), loop=await _finished_task()),
+            connection=InfiniteInsert(),
+            results_for_wrapper=_BatchDataWrapper(),
+            results_for_wrapper_backup=_BatchDataWrapper(),
+            batch_objects=[],
+            batch_references=[],
+        )
+        await batch._wait()
+
+    asyncio.run(run())
+
+
+def test_aexit_does_not_log_the_exception_that_is_already_propagating(caplog) -> None:
+    # flush()/add_object raised __bg_exception inside the block; _wait() re-raises the
+    # SAME object on exit — that is not a second failure to log
+    err = WeaviateBatchStreamError("bg died")
+    fake = _FakeBatch(err)
+
+    async def run() -> None:
+        async with _ContextManagerAsync(fake):  # type: ignore[arg-type]
+            raise err
+
+    with caplog.at_level(logging.WARNING, logger="weaviate-client"):
+        with pytest.raises(WeaviateBatchStreamError, match="bg died"):
+            asyncio.run(run())
+    assert fake.wait_called
+    assert "batch stream failed" not in caplog.text
diff --git a/test/test_batch_sync.py b/test/test_batch_sync.py
new file mode 100644
index 000000000..5656e7388
--- /dev/null
+++ b/test/test_batch_sync.py
@@ -0,0 +1,164 @@
+"""Unit tests for the sync batch-stream failure handling, mirroring test_batch_async.py.
+
+_wait() surfaces a background failure (or data left unsent) instead of returning as if
+the batch had succeeded, while an exception leaving the `with` block still wins.
+"""
+
+import logging
+import threading
+import time
+from typing import Optional
+
+import pytest
+
+from weaviate.collections.batch.base import _BatchDataWrapper, _BgThreads
+from weaviate.collections.batch.batch_wrapper import _ContextManagerSync
+from weaviate.collections.batch.sync import _BatchBaseSync
+from weaviate.config import Timeout as TimeoutConfig
+from weaviate.exceptions import WeaviateBatchStreamError
+
+
+def _bare_batch(**mangled) -> _BatchBaseSync:
+    batch = object.__new__(_BatchBaseSync)
+    for name, value in mangled.items():
+        setattr(batch, f"_BatchBaseSync__{name}", value)
+    return batch
+
+
+class _FakeThreads:
+    def __init__(self, alive: bool = False) -> None:
+        self.alive = alive
+
+    def join(self, timeout=None) -> None:
+        return None
+
+    def is_alive(self) -> bool:
+        return self.alive
+
+
+class _FakeTimeouts:
+    insert = 1
+
+
+class _FakeConnection:
+    timeout_config = _FakeTimeouts()
+
+
+def _batch_for_wait(**mangled) -> _BatchBaseSync:
+    defaults = {
+        "bg_exception": None,
+        "bg_threads": _FakeThreads(alive=False),
+        "connection": _FakeConnection(),
+        "results_for_wrapper": _BatchDataWrapper(),
+        "results_for_wrapper_backup": _BatchDataWrapper(),
+        "batch_objects": [],
+        "batch_references": [],
+    }
+    defaults.update(mangled)
+    return _bare_batch(**defaults)
+
+
+def test_wait_returns_quietly_when_everything_was_sent() -> None:
+    _batch_for_wait()._wait()
+
+
+def test_wait_raises_the_background_exception_and_keeps_partial_results() -> None:
+    # like the async colour: a background failure must not come back as a success
+    partial = _BatchDataWrapper()
+    partial.failed_objects = ["sentinel-failure"]  # type: ignore[list-item]
+    backup = _BatchDataWrapper()
+    batch = _batch_for_wait(
+        bg_exception=RuntimeError("boom"),
+        results_for_wrapper=partial,
+        results_for_wrapper_backup=backup,
+    )
+
+    with pytest.raises(RuntimeError, match="boom"):
+        batch._wait()
+    assert backup.failed_objects == ["sentinel-failure"]
+
+
+def test_wait_names_unsent_data_when_the_threads_are_gone() -> None:
+    batch = _batch_for_wait(batch_objects=[object(), object()], batch_references=[object()])
+    with pytest.raises(
+        WeaviateBatchStreamError, match="ended with 2 objects and 1 references unsent"
+    ):
+        batch._wait()
+
+
+def test_check_alive_raises_inside_the_taxonomy() -> None:
+    # used to be a bare Exception("Batch thread died unexpectedly")
+    batch = _bare_batch(bg_exception=None, bg_threads=_FakeThreads(alive=False))
+    with pytest.raises(WeaviateBatchStreamError, match="stream has ended"):
+        getattr(batch, "_BatchBaseSync__check_bg_threads_alive")()  # noqa: B009
+
+
+class _FakeBatch:
+    def __init__(self, wait_error: Optional[BaseException] = None) -> None:
+        self.wait_error = wait_error
+        self.shutdown_called = False
+        self.wait_called = False
+
+    def _start(self) -> None:
+        pass
+
+    def _shutdown(self) -> None:
+        self.shutdown_called = True
+
+    def _wait(self) -> None:
+        self.wait_called = True
+        if self.wait_error is not None:
+            raise self.wait_error
+
+
+def test_exit_raises_a_background_failure_on_a_clean_block() -> None:
+    fake = _FakeBatch(WeaviateBatchStreamError("bg died"))
+    with pytest.raises(WeaviateBatchStreamError, match="bg died"):
+        with _ContextManagerSync(fake):  # type: ignore[type-var]
+            pass
+
+
+def test_exit_keeps_the_users_exception_over_a_background_failure(caplog) -> None:
+    fake = _FakeBatch(WeaviateBatchStreamError("bg died"))
+    with caplog.at_level(logging.WARNING, logger="weaviate-client"):
+        with pytest.raises(ValueError, match="user code"):
+            with _ContextManagerSync(fake):  # type: ignore[type-var]
+                raise ValueError("user code")
+    assert fake.shutdown_called and fake.wait_called
+    assert "bg died" in caplog.text
+
+
+def test_exit_never_swallows_a_base_exception() -> None:
+    fake = _FakeBatch(WeaviateBatchStreamError("bg died"))
+    with pytest.raises(KeyboardInterrupt):
+        with _ContextManagerSync(fake):  # type: ignore[type-var]
+            raise KeyboardInterrupt()
+    assert fake.wait_called
+
+
+def test_wait_with_an_infinite_insert_timeout_does_not_overflow_join() -> None:
+    # Timeout(insert=inf) is accepted (it means "no deadline"), but Thread.join(inf)
+    # raises OverflowError — the shutdown wait must become "no timeout" instead
+    class InfiniteInsert:
+        timeout_config = TimeoutConfig(insert=float("inf"))
+
+    threads = _BgThreads(
+        loop=threading.Thread(target=lambda: None), recv=threading.Thread(target=lambda: None)
+    )
+    threads.start_recv()
+    threads.start_loop()
+    time.sleep(0.1)  # let both finish; is_alive()/join() are deliberately not called yet
+    _batch_for_wait(bg_threads=threads, connection=InfiniteInsert())._wait()
+
+
+def test_exit_does_not_log_the_exception_that_is_already_propagating(caplog) -> None:
+    # flush()/add_object raised __bg_exception inside the block; _wait() re-raises the
+    # SAME object on exit — that is not a second failure to log
+    err = WeaviateBatchStreamError("bg died")
+    fake = _FakeBatch(err)
+    with caplog.at_level(logging.WARNING, logger="weaviate-client"):
+        with pytest.raises(WeaviateBatchStreamError, match="bg died"):
+            with _ContextManagerSync(fake):  # type: ignore[type-var]
+                raise err
+    assert fake.wait_called
+    assert "batch stream failed" not in caplog.text
diff --git a/test/test_connection.py b/test/test_connection.py
new file mode 100644
index 000000000..28540f162
--- /dev/null
+++ b/test/test_connection.py
@@ -0,0 +1,90 @@
+"""Unit tests for weaviate.connect.v4 connection-level behaviour that needs no server."""
+
+import asyncio
+import inspect
+import threading
+import time
+
+import grpc
+import pytest
+from grpc.aio import AioRpcError, Metadata
+
+from weaviate.config import ConnectionConfig
+from weaviate.config import Timeout as TimeoutConfig
+from weaviate.connect.base import ConnectionParams
+from weaviate.connect.v4 import ConnectionAsync, ConnectionSync, _ConnectionBase
+from weaviate.exceptions import WeaviateBatchError
+from weaviate.proto.v1 import batch_pb2
+
+
+def test_connection_sync_init_mirrors_the_base_signature() -> None:
+    # explicit, typed parameters (no *args/**kwargs pass-through)
+    assert (
+        inspect.signature(ConnectionSync.__init__).parameters
+        == inspect.signature(_ConnectionBase.__init__).parameters
+    )
+
+
+def _connection_async() -> ConnectionAsync:
+    return ConnectionAsync(
+        connection_params=ConnectionParams.from_url("http://localhost:8080", 50051),
+        auth_client_secret=None,
+        timeout_config=TimeoutConfig(),
+        proxies=None,
+        trust_env=False,
+        additional_headers=None,
+        connection_config=ConnectionConfig(),
+    )
+
+
+def test_async_batch_objects_error_carries_the_grpc_details_only() -> None:
+    # like WeaviateQueryError: the message is the server's details, not the whole
+    # '<AioRpcError of RPC that terminated with: ...>' repr
+    class FailingStub:
+        async def BatchObjects(self, request, metadata=None, timeout=None):
+            raise AioRpcError(
+                grpc.StatusCode.INVALID_ARGUMENT, Metadata(), Metadata(), details="bad object"
+            )
+
+    conn = _connection_async()
+    conn._connected = True
+    conn._grpc_stub = FailingStub()  # type: ignore[assignment]
+
+    with pytest.raises(WeaviateBatchError) as excinfo:
+        asyncio.run(
+            conn.grpc_batch_objects(batch_pb2.BatchObjectsRequest(), timeout=1, max_retries=0)
+        )
+    assert excinfo.value.message == "bad object"
+    assert "AioRpcError" not in str(excinfo.value)
+
+
+def test_cancel_refresher_owned_by_another_loop_is_scheduled_not_awaited() -> None:
+    # close() from a thread/loop other than the refresher's must not try to await the
+    # task (which would hang); it schedules the cancel on the owning loop instead
+    loop = asyncio.new_event_loop()
+    started = threading.Event()
+    holder: dict = {}
+
+    async def forever() -> None:
+        started.set()
+        await asyncio.sleep(3600)
+
+    def run() -> None:
+        holder["task"] = loop.create_task(forever())
+        loop.run_forever()
+
+    thread = threading.Thread(target=run, daemon=True)
+    thread.start()
+    assert started.wait(timeout=2)
+    conn = _connection_async()
+    conn._ConnectionBase__token_refresh_task = holder["task"]  # type: ignore[attr-defined]
+    try:
+        assert conn._cancel_background_token_refresh() is None  # nothing to await here
+        deadline = time.monotonic() + 2
+        while not holder["task"].done() and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert holder["task"].cancelled()
+    finally:
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join(timeout=2)
+        loop.close()
diff --git a/test/test_connection_params.py b/test/test_connection_params.py
index 89b9a4cf4..a0dc1c7d4 100644
--- a/test/test_connection_params.py
+++ b/test/test_connection_params.py
@@ -171,3 +171,51 @@ def fake_insecure_channel(target, options=None, **kwargs):
 
     option_keys = [key for key, _ in captured["options"]]
     assert "grpc-web.path_prefix" not in option_keys
+
+
+def test_connect_to_custom_treats_root_prefix_as_native_grpc(monkeypatch) -> None:
+    # "/" normalizes to "" (native gRPC), so the async-only guard must not fire on it
+    import weaviate
+    import weaviate.connect.helpers as helpers
+
+    monkeypatch.setattr(helpers, "__connect", lambda client: client)  # module-level: not mangled
+    client = weaviate.connect_to_custom(
+        http_host="localhost",
+        http_port=8080,
+        http_secure=False,
+        grpc_host="localhost",
+        grpc_port=50051,
+        grpc_secure=False,
+        grpc_path_prefix="/",
+    )
+    assert client._connection._connection_params._grpc_web_path_prefix == ""
+
+
+def test_async_client_construction_rejects_prefix_without_shim(monkeypatch) -> None:
+    # fail at construction with actionable text, not deep inside connect() after the
+    # OIDC and /v1/meta round trips already succeeded
+    from weaviate import WeaviateAsyncClient
+
+    monkeypatch.delattr(base_mod.grpc, "__weaviate_client_web_shim__", raising=False)
+    with pytest.raises(WeaviateInvalidInputError) as excinfo:
+        WeaviateAsyncClient(_grpc_web_params())
+    msg = str(excinfo.value)
+    assert "weaviate-client-web" in msg
+    assert "install(force=True)" in msg
+    assert "set_sender(make_httpx_sender())" in msg
+    assert "import weaviate" in msg  # ... and how Pyodide differs
+
+
+def test_async_client_construction_allows_prefix_with_shim(monkeypatch) -> None:
+    from weaviate import WeaviateAsyncClient
+
+    monkeypatch.setattr(base_mod.grpc, "__weaviate_client_web_shim__", True, raising=False)
+    client = WeaviateAsyncClient(_grpc_web_params())
+    assert client._connection._connection_params._grpc_web_path_prefix == "/grpc-web"
+
+
+def test_sync_client_construction_rejects_grpc_web_prefix() -> None:
+    from weaviate import WeaviateClient
+
+    with pytest.raises(WeaviateInvalidInputError, match="async"):
+        WeaviateClient(_grpc_web_params())
diff --git a/test/test_wasm_compat.py b/test/test_wasm_compat.py
index f568d08e9..5ac1d58fc 100644
--- a/test/test_wasm_compat.py
+++ b/test/test_wasm_compat.py
@@ -6,8 +6,10 @@
 """
 
 import sys
+from typing import Optional
 
 import grpc
+import httpx
 import pytest
 from grpc.aio import AioRpcError, Metadata
 from httpx import ConnectError, ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout
@@ -16,7 +18,7 @@
 from weaviate.config import ConnectionConfig
 from weaviate.config import Timeout as TimeoutConfig
 from weaviate.connect.base import ConnectionParams
-from weaviate.connect.v4 import _ConnectionBase, _exc_detail
+from weaviate.connect.v4 import _ConnectionBase, _deadline, _exc_detail
 from weaviate.embedded import _EmbeddedBase
 from weaviate.exceptions import (
     WeaviateClosedClientError,
@@ -58,25 +60,33 @@ def test_async_client_construction_allowed_under_emscripten(monkeypatch) -> None
     assert client is not None
 
 
-def _handle_exceptions(e: Exception, error_msg: str = "") -> None:
+def _handle_exceptions(
+    e: Exception, error_msg: str = "", client: Optional[httpx.Client] = None
+) -> None:
     conn = object.__new__(_ConnectionBase)
-    # keep the bare instance's __del__ quiet (it checks these for unclosed connections)
-    conn._client = None
+    # __del__ checks these for unclosed connections; None also means "no client at all"
+    conn._client = client
     conn._grpc_channel = None
     getattr(conn, "_ConnectionBase__handle_exceptions")(e, error_msg)  # noqa: B009
 
 
-def test_httpx_closed_client_runtime_error_maps_to_closed_client() -> None:
-    # the exact message httpx raises for a closed AsyncClient/Client
+def test_runtime_error_from_a_closed_client_maps_to_closed_client() -> None:
+    # httpx raises a bare RuntimeError('Cannot send a request, as the client has been
+    # closed.'); the client's state, not the message text, is what makes it 'closed'
+    closed = httpx.Client()
+    closed.close()
     with pytest.raises(WeaviateClosedClientError):
         _handle_exceptions(RuntimeError("Cannot send a request, as the client has been closed."))
+    with pytest.raises(WeaviateClosedClientError):
+        _handle_exceptions(RuntimeError("some other wording"), client=closed)
 
 
 def test_unrelated_runtime_error_is_not_rewritten_as_closed_client() -> None:
     # Emscripten's canonical thread failure must propagate as-is, not as a misleading
     # 'client is closed - run client.connect()'
-    with pytest.raises(RuntimeError, match="can't start new thread"):
-        _handle_exceptions(RuntimeError("can't start new thread"))
+    with httpx.Client() as open_client:
+        with pytest.raises(RuntimeError, match="can't start new thread"):
+            _handle_exceptions(RuntimeError("can't start new thread"), client=open_client)
 
 
 def test_connect_error_message_includes_exception_type() -> None:
@@ -206,21 +216,52 @@ def test_rest_timeouts_are_capped_at_five_seconds_on_native_platforms() -> None:
     assert insert.read == 90
 
 
-def test_rest_connect_timeout_follows_the_request_timeout_under_emscripten(monkeypatch) -> None:
-    # under Pyodide the connect timeout bounds the entire fetch promise, so a fixed 5s
-    # silently caps every request at ~5s of wall clock
+def test_rest_deadline_under_emscripten_is_the_read_timeout(monkeypatch) -> None:
+    # the fetch transport (weaviate-client-web) reads only `read` as the whole-request
+    # deadline; connect/write/pool are never consulted there, so nothing platform-specific
+    # is computed: same httpx.Timeout as on CPython, finite -> value, non-finite -> None
     monkeypatch.setattr(sys, "platform", "emscripten")
-    conn = _connection()
 
-    insert = _get_timeout(conn, "POST")
-    assert insert.connect == 90 and insert.write == 90
-    assert insert.read == 90
-
-    query = _get_timeout(conn, "GET")
-    assert query.connect == 30
+    insert = _get_timeout(_connection(), "POST")
+    assert insert.read == 90 and insert.connect == 5.0
+    query = _get_timeout(_connection(), "GET")
     assert query.read == 30
 
-    # a configured timeout below the httpx default must not make connecting stricter
     short = _get_timeout(_connection(insert=1, query=1), "POST")
-    assert short.connect == 5.0
     assert short.read == 1
+    unbounded = _get_timeout(_connection(insert=float("inf")), "POST")
+    assert unbounded.read is None
+
+
+@pytest.mark.parametrize("value", [float("inf"), float("nan")])
+def test_deadline_maps_non_finite_to_none(value: float) -> None:
+    # Timeout(query=float('inf')) passes pydantic's ge=0; handed on as-is it overflows in
+    # the fetch layer under Pyodide (int(inf * 1000)). None is "no deadline" for both
+    # httpx and grpc.
+    assert _deadline(value) is None
+    assert _deadline(None) is None
+    assert _deadline(0) == 0
+    assert _deadline(2.5) == 2.5
+
+
+def test_infinite_rest_timeouts_become_no_read_timeout() -> None:
+    conn = _connection(insert=float("inf"), query=float("inf"))
+    insert = _get_timeout(conn, "POST")
+    assert insert.read is None
+    assert insert.connect == 5.0  # the httpx default is kept on native platforms
+    query = _get_timeout(conn, "GET")
+    assert query.read is None
+
+
+def test_deadlines_view_sanitises_every_timeout() -> None:
+    conn = _connection()
+    conn.timeout_config = TimeoutConfig(
+        query=float("inf"), insert=90, init=float("inf"), stream=float("inf")
+    )
+    deadlines = conn._deadlines
+    assert deadlines.query is None
+    assert deadlines.insert == 90
+    assert deadlines.init is None  # Timeout(init=inf) used to end in an OverflowError
+    assert deadlines.stream is None
+    # the user's own config object is left untouched
+    assert conn.timeout_config.init == float("inf")
diff --git a/weaviate/__init__.py b/weaviate/__init__.py
index 49c1dbadf..20ccff232 100644
--- a/weaviate/__init__.py
+++ b/weaviate/__init__.py
@@ -7,9 +7,13 @@
 if sys.platform == "emscripten":
     try:
         import weaviate_client_web  # noqa: F401
-    except ImportError:
+    except ImportError as exc:
         from importlib.util import find_spec
 
+        # Only an absent companion earns the install hint; a companion that is present
+        # but fails to import (a broken dependency of its own) must surface that error.
+        if not (isinstance(exc, ModuleNotFoundError) and exc.name == "weaviate_client_web"):
+            raise
         if find_spec("grpc") is None:
             raise ImportError(
                 "weaviate requires the weaviate-client-web package under "
@@ -17,7 +21,7 @@
                 "weaviate-client-web provides the grpc-web (fetch) transport in its "
                 "place. Install it (e.g. micropip.install('weaviate-client-web')) and "
                 "import weaviate again."
-            ) from None
+            ) from exc
 
 import os
 from importlib.metadata import PackageNotFoundError, version
diff --git a/weaviate/collections/batch/async_.py b/weaviate/collections/batch/async_.py
index 21310ab94..d09bbf723 100644
--- a/weaviate/collections/batch/async_.py
+++ b/weaviate/collections/batch/async_.py
@@ -39,7 +39,7 @@
 from weaviate.collections.classes.types import WeaviateProperties
 from weaviate.connect.base import _grpc_web_shim_active
 from weaviate.connect.executor import aresult
-from weaviate.connect.v4 import ConnectionAsync
+from weaviate.connect.v4 import ConnectionAsync, _deadline
 from weaviate.exceptions import (
     WeaviateBatchFailedToReestablishStreamError,
     WeaviateBatchStreamError,
@@ -194,7 +194,8 @@ async def recv_wrapper() -> None:
     async def _wait(self) -> None:
         assert self.__bg_tasks is not None
         # this is how long an insert will take to timeout for, so we wait at most this time +5s for the batch to finish after shutdown is initiated, in case the server never hangs up
-        shutdown_timeout = self.__connection.timeout_config.insert + 5
+        insert = _deadline(self.__connection.timeout_config.insert)
+        shutdown_timeout = None if insert is None else insert + 5
         try:
             await self.__bg_tasks.gather(timeout=shutdown_timeout)
         except asyncio.TimeoutError as e:
@@ -217,13 +218,14 @@ async def _wait(self) -> None:
             # surface background-task failures instead of returning partial results
             # as if the batch had succeeded
             raise self.__bg_exception
-        if (
-            len(self.__batch_objects) > 0 or len(self.__batch_references) > 0
-        ) and not self.__all_tasks_alive():
-            # the tasks are gone with data still queued and nothing recorded in
-            # __bg_exception (a BaseException escaping loop_wrapper/recv_wrapper): the
-            # batch did NOT complete, so do not return as if it had
-            raise self.__bg_task_death_cause()
+        n_objs, n_refs = len(self.__batch_objects), len(self.__batch_references)
+        if n_objs + n_refs > 0 and not self.__all_tasks_alive():
+            # the tasks are gone with data still queued: the batch did NOT complete, so
+            # do not return as if it had — whether a task died (a BaseException escaping
+            # loop_wrapper/recv_wrapper) or the server ended the stream early
+            raise self.__bg_task_death_cause() or WeaviateBatchStreamError(
+                f"batch stream ended with {n_objs} objects and {n_refs} references unsent"
+            )
 
     async def _shutdown(self) -> None:
         self.__is_stopped.set()
@@ -560,12 +562,8 @@ async def flush(self) -> None:
         """Flush the batch queue and wait for all requests to be finished."""
         # bg thread is sending objs+refs automatically, so simply wait for everything to be done
         while len(self.__batch_objects) > 0 or len(self.__batch_references) > 0:
-            # a dead task is the condition to check, not __bg_exception: loop_wrapper /
-            # recv_wrapper only catch Exception, so a BaseException (notably the
-            # asyncio.CancelledError grpc.aio raises on a cancelled streaming call) ends
-            # the task with __bg_exception unset. Nothing then drains the queues, so
-            # waiting any longer would hang forever. Mirrors the sync colour's
-            # __check_bg_threads_alive().
+            # a dead task (not just __bg_exception, which a BaseException such as grpc.aio's
+            # CancelledError leaves unset) means nothing drains the queues: raise, don't hang
             self.__check_bg_tasks_alive()
             await asyncio.sleep(0.01)
 
@@ -664,10 +662,14 @@ def __check_bg_tasks_alive(self) -> None:
         if self.__all_tasks_alive():
             return
 
-        raise self.__bg_exception or self.__bg_task_death_cause()
+        raise (
+            self.__bg_exception
+            or self.__bg_task_death_cause()
+            or WeaviateBatchStreamError("the batch stream has ended")
+        )
 
-    def __bg_task_death_cause(self) -> Exception:
-        """Explain a background task that ended without setting __bg_exception.
+    def __bg_task_death_cause(self) -> Optional[Exception]:
+        """Explain a background task that died without setting __bg_exception; None if none did.
 
         loop_wrapper/recv_wrapper only catch Exception, so a BaseException — notably the
         asyncio.CancelledError grpc.aio raises when a streaming call is cancelled — ends
@@ -689,4 +691,4 @@ def __bg_task_death_cause(self) -> Exception:
                     return exc
                 if exc is not None:
                     return WeaviateBatchStreamError(f"the background {name} task died with {exc!r}")
-        return WeaviateBatchStreamError("the background tasks died unexpectedly")
+        return None
diff --git a/weaviate/collections/batch/batch_wrapper.py b/weaviate/collections/batch/batch_wrapper.py
index a3a3598d6..eb1353dd7 100644
--- a/weaviate/collections/batch/batch_wrapper.py
+++ b/weaviate/collections/batch/batch_wrapper.py
@@ -508,7 +508,18 @@ def __init__(self, current_batch: T):
 
     def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         self.__current_batch._shutdown()
-        self.__current_batch._wait()
+        if exc_type is None:
+            self.__current_batch._wait()
+            return
+        # the exception leaving the block wins; a background failure is only logged
+        # (unless it IS the one in flight, re-raised by _wait after flush/add already did)
+        try:
+            self.__current_batch._wait()
+        except Exception as e:
+            if e is not exc_val:
+                logger.warning(
+                    f"batch stream failed while the block raised {exc_type.__name__}: {e}"
+                )
 
     def __enter__(self) -> P:
         self.__current_batch._start()
@@ -521,7 +532,21 @@ def __init__(self, current_batch: _BatchBaseAsync):
 
     async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         await self.__current_batch._shutdown()
-        await self.__current_batch._wait()
+        if exc_type is None:
+            await self.__current_batch._wait()
+            return
+        # the exception leaving the block wins (never replace a CancelledError or other
+        # BaseException); a background failure is only logged (unless it IS the one in
+        # flight, re-raised by _wait after flush/add already did)
+        try:
+            await self.__current_batch._wait()
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            if e is not exc_val:
+                logger.warning(
+                    f"batch stream failed while the block raised {exc_type.__name__}: {e}"
+                )
 
     async def __aenter__(self) -> Q:
         await self.__current_batch._start()
diff --git a/weaviate/collections/batch/client.py b/weaviate/collections/batch/client.py
index d9a2524d7..b95238e03 100644
--- a/weaviate/collections/batch/client.py
+++ b/weaviate/collections/batch/client.py
@@ -285,6 +285,11 @@ def stream(
         Args:
             concurrency: The number of concurrent streams to use when sending batches. If not provided, the default will be one.
             consistency_level: The consistency level to be used when inserting data. If not provided, the default value is `None`.
+
+        Raises:
+            WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references
+                still unsent. An exception raised inside the block is propagated unchanged; a background failure is
+                then only logged.
         """
         if self._connection._weaviate_version.is_lower_than(1, 36, 0):
             raise WeaviateUnsupportedFeatureError(
@@ -344,6 +349,11 @@ def stream(
         Args:
             concurrency: The number of concurrent streams to use when sending batches. If not provided, the default will be one.
             consistency_level: The consistency level to be used when inserting data. If not provided, the default value is `None`.
+
+        Raises:
+            WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references
+                still unsent. An exception raised inside the block is propagated unchanged; a background failure is
+                then only logged.
         """
         if self._connection._weaviate_version.is_lower_than(1, 36, 0):
             raise WeaviateUnsupportedFeatureError(
diff --git a/weaviate/collections/batch/collection.py b/weaviate/collections/batch/collection.py
index 415b2df49..565412563 100644
--- a/weaviate/collections/batch/collection.py
+++ b/weaviate/collections/batch/collection.py
@@ -310,6 +310,11 @@ def stream(
         Args:
             concurrency: The number of concurrent requests when sending batches. This controls the number of concurrent requests
                 made to Weaviate. If not provided, the default value is 1.
+
+        Raises:
+            WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references
+                still unsent. An exception raised inside the block is propagated unchanged; a background failure is
+                then only logged.
         """
         if self._connection._weaviate_version.is_lower_than(1, 36, 0):
             raise WeaviateUnsupportedFeatureError(
@@ -370,6 +375,11 @@ def stream(
         Args:
             concurrency: The number of concurrent requests when sending batches. This controls the number of concurrent requests
                 made to Weaviate. If not provided, the default value is 1.
+
+        Raises:
+            WeaviateBatchStreamError: On exit, if the background stream failed or ended with objects or references
+                still unsent. An exception raised inside the block is propagated unchanged; a background failure is
+                then only logged.
         """
         if self._connection._weaviate_version.is_lower_than(1, 36, 0):
             raise WeaviateUnsupportedFeatureError(
diff --git a/weaviate/collections/batch/sync.py b/weaviate/collections/batch/sync.py
index 6cf8c1edc..40e53b6c0 100644
--- a/weaviate/collections/batch/sync.py
+++ b/weaviate/collections/batch/sync.py
@@ -35,7 +35,7 @@
 )
 from weaviate.collections.classes.types import WeaviateProperties
 from weaviate.connect.executor import result
-from weaviate.connect.v4 import ConnectionSync
+from weaviate.connect.v4 import ConnectionSync, _deadline
 from weaviate.exceptions import (
     WeaviateBatchFailedToReestablishStreamError,
     WeaviateBatchStreamError,
@@ -134,7 +134,8 @@ def _start(self) -> None:
 
     def _wait(self) -> None:
         # this is how long an insert will take to timeout for, so we wait at most this time +5s for the batch to finish after shutdown is initiated, in case the server never hangs up
-        shutdown_timeout = self.__connection.timeout_config.insert + 5
+        insert = _deadline(self.__connection.timeout_config.insert)
+        shutdown_timeout = None if insert is None else insert + 5  # inf overflows Thread.join
         try:
             self.__bg_threads.join(shutdown_timeout)
         except TimeoutError as e:
@@ -142,7 +143,8 @@ def _wait(self) -> None:
                 "Background batch threads did not terminate after forced shutdown."
             ) from e
 
-        # copy the results to the public results
+        # copy the results to the public results — even on failure, so the user can
+        # still inspect batch.results / batch.failed_objects after catching
         self.__results_for_wrapper_backup.results = self.__results_for_wrapper.results
         self.__results_for_wrapper_backup.failed_objects = self.__results_for_wrapper.failed_objects
         self.__results_for_wrapper_backup.failed_references = (
@@ -152,6 +154,17 @@ def _wait(self) -> None:
             self.__results_for_wrapper.imported_shards
         )
 
+        if self.__bg_exception is not None:
+            # surface background-thread failures instead of returning partial results
+            # as if the batch had succeeded
+            raise self.__bg_exception
+        n_objs, n_refs = len(self.__batch_objects), len(self.__batch_references)
+        if n_objs + n_refs > 0 and not self.__all_threads_alive():
+            # the threads are gone with data still queued: the batch did NOT complete
+            raise WeaviateBatchStreamError(
+                f"batch stream ended with {n_objs} objects and {n_refs} references unsent"
+            )
+
     def _shutdown(self) -> None:
         # Shutdown the current batch and wait for all requests to be finished
         self.__is_stopped.set()
@@ -644,4 +657,4 @@ def __check_bg_threads_alive(self) -> None:
         if self.__all_threads_alive():
             return
 
-        raise self.__bg_exception or Exception("Batch thread died unexpectedly")
+        raise self.__bg_exception or WeaviateBatchStreamError("the batch stream has ended")
diff --git a/weaviate/connect/base.py b/weaviate/connect/base.py
index fd5bed8e5..4fc093023 100644
--- a/weaviate/connect/base.py
+++ b/weaviate/connect/base.py
@@ -149,6 +149,29 @@ def _grpc_web_path_prefix(self) -> str:
         cleaned = (self.grpc_path_prefix or "").strip("/")
         return f"/{cleaned}" if cleaned else ""
 
+    def _check_grpc_web_usable(self, is_async: bool) -> None:
+        """Fail fast on a grpc-web prefix this process cannot honour; a no-op for native gRPC.
+
+        A native grpcio channel would silently ignore the ``grpc-web.path_prefix`` option
+        and route over native gRPC, so the shim (which consumes it) must be in place.
+        """
+        if self._grpc_web_path_prefix == "":
+            return
+        if not is_async:
+            raise WeaviateInvalidInputError(
+                "grpc_path_prefix (grpc-web) is only supported for async clients; "
+                "use use_async_with_custom(...) / WeaviateAsyncClient"
+            )
+        if not _grpc_web_shim_active():
+            raise WeaviateInvalidInputError(
+                "grpc_path_prefix enables grpc-web, which requires the "
+                "'weaviate-client-web' package (it installs a grpc shim before "
+                "'import weaviate'); it is not active in this environment. Under Pyodide a "
+                "plain `import weaviate` activates it; on CPython call "
+                "weaviate_client_web.install(force=True) and set_sender(make_httpx_sender()) "
+                "before importing weaviate (intended for integration testing)."
+            )
+
     def _grpc_channel(
         self,
         proxies: Dict[str, str],
@@ -172,24 +195,9 @@ def _grpc_channel(
         if grpc_config is not None and grpc_config.channel_options is not None:
             options.extend(grpc_config.channel_options)
 
-        # grpc-web mode (prefix set): only valid for an async client, and only when the
-        # weaviate-client-web shim has replaced the grpc module (it consumes the
-        # grpc-web.path_prefix option). Fail fast otherwise — a native grpcio channel
-        # would silently ignore the option and route over native gRPC, a confusing
-        # misconfiguration. Nothing is added for native gRPC, so its channel options stay
-        # byte-for-byte unchanged.
+        # nothing is added for native gRPC, so its channel options stay byte-for-byte unchanged
         if (prefix := self._grpc_web_path_prefix) != "":
-            if not is_async:
-                raise WeaviateInvalidInputError(
-                    "grpc_path_prefix (grpc-web) is only supported for async clients; "
-                    "use use_async_with_custom(...) / WeaviateAsyncClient"
-                )
-            if not _grpc_web_shim_active():
-                raise WeaviateInvalidInputError(
-                    "grpc_path_prefix enables grpc-web, which requires the "
-                    "'weaviate-client-web' package (it installs a grpc shim before "
-                    "'import weaviate'); it is not active in this environment"
-                )
+            self._check_grpc_web_usable(is_async)
             options.append(("grpc-web.path_prefix", prefix))
 
         if is_async:
diff --git a/weaviate/connect/helpers.py b/weaviate/connect/helpers.py
index a2d87afdb..28153ddc1 100644
--- a/weaviate/connect/helpers.py
+++ b/weaviate/connect/helpers.py
@@ -351,22 +351,23 @@ def connect_to_custom(
         True
         >>> # The connection is automatically closed when the context is exited.
     """
-    if grpc_path_prefix:
+    connection_params = ConnectionParams.from_params(
+        http_host=http_host,
+        http_port=http_port,
+        http_secure=http_secure,
+        grpc_host=grpc_host,
+        grpc_port=grpc_port,
+        grpc_secure=grpc_secure,
+        grpc_path_prefix=grpc_path_prefix,
+    )
+    if connection_params._grpc_web_path_prefix:  # normalized: "/" is native gRPC
         raise WeaviateInvalidInputError(
             "grpc_path_prefix enables grpc-web, which is async-only; use "
             "use_async_with_custom(...) instead of connect_to_custom(...)"
         )
     return __connect(
         WeaviateClient(
-            ConnectionParams.from_params(
-                http_host=http_host,
-                http_port=http_port,
-                http_secure=http_secure,
-                grpc_host=grpc_host,
-                grpc_port=grpc_port,
-                grpc_secure=grpc_secure,
-                grpc_path_prefix=grpc_path_prefix,
-            ),
+            connection_params,
             auth_client_secret=__parse_auth_credentials(auth_credentials),
             additional_headers=headers,
             additional_config=additional_config,
diff --git a/weaviate/connect/v4.py b/weaviate/connect/v4.py
index 852da8dd9..e0c4a7fca 100644
--- a/weaviate/connect/v4.py
+++ b/weaviate/connect/v4.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import math
 import sys
 import time
 from copy import copy
@@ -111,6 +112,44 @@
 
 PERMISSION_DENIED = "PERMISSION_DENIED"
 
+# ceiling (seconds) for the exponential backoff between failed token refresh attempts
+TOKEN_REFRESH_BACKOFF_CAP = 60
+
+
+def _token_refresh_backoff(consecutive_failures: int) -> int:
+    """Seconds to wait before the next refresh attempt: 1, 2, 4, ... up to the cap."""
+    return min(2 ** min(consecutive_failures - 1, 16), TOKEN_REFRESH_BACKOFF_CAP)
+
+
+def _deadline(timeout: Union[int, float, None]) -> Union[int, float, None]:
+    """A timeout as handed to httpx/grpc: non-finite (e.g. ``float('inf')``) means no deadline.
+
+    Passed on as-is, ``inf`` overflows in the fetch layer under Pyodide; None is "no
+    timeout" for httpx and "no deadline" for grpc alike.
+    """
+    if timeout is None or not math.isfinite(timeout):
+        return None
+    return timeout
+
+
+@dataclass(frozen=True)
+class _Deadlines:
+    """The user's timeout config as handed to httpx/grpc (each value through ``_deadline``)."""
+
+    query: Union[int, float, None]
+    insert: Union[int, float, None]
+    init: Union[int, float, None]
+    stream: Union[int, float, None]
+
+    @classmethod
+    def of(cls, config: TimeoutConfig) -> "_Deadlines":
+        return cls(
+            query=_deadline(config.query),
+            insert=_deadline(config.insert),
+            init=_deadline(config.init),
+            stream=_deadline(config.stream),
+        )
+
 
 def _exc_detail(e: BaseException) -> str:
     """Format an exception for user-facing messages.
@@ -158,6 +197,8 @@ def __init__(
         self._connection_params = connection_params
         self._grpc_stub: Optional[weaviate_pb2_grpc.WeaviateStub] = None
         self._grpc_channel: Union[AsyncChannel, SyncChannel, None] = None
+        # a grpc-web prefix this process cannot honour fails here, not deep inside connect()
+        connection_params._check_grpc_web_usable(is_async=not isinstance(self, ConnectionSync))
         self.timeout_config = timeout_config
         self.__connection_config = connection_config
         self.__trust_env = trust_env
@@ -328,7 +369,7 @@ def _ping_grpc(self, colour: executor.Colour) -> Union[None, Awaitable[None]]:
                 "/grpc.health.v1.Health/Check",
                 request_serializer=health_weaviate_pb2.WeaviateHealthCheckRequest.SerializeToString,
                 response_deserializer=health_weaviate_pb2.WeaviateHealthCheckResponse.FromString,
-            )(health_weaviate_pb2.WeaviateHealthCheckRequest(), timeout=self.timeout_config.init)
+            )(health_weaviate_pb2.WeaviateHealthCheckRequest(), timeout=self._deadlines.init)
             if colour == "async":
 
                 async def execute():
@@ -388,6 +429,10 @@ def server_version(self) -> str:
         """Version of the weaviate instance."""
         return str(self._weaviate_version)
 
+    @property
+    def _deadlines(self) -> _Deadlines:
+        return _Deadlines.of(self.timeout_config)
+
     def get_proxies(self) -> Dict[str, str]:
         return self._proxies
 
@@ -432,7 +477,7 @@ def _open_connections_rest(
             async def get_oidc() -> None:
                 async with self._make_client("async") as client:
                     try:
-                        response = await client.get(oidc_url, timeout=self.timeout_config.init)
+                        response = await client.get(oidc_url, timeout=self._deadlines.init)
                     except Exception as e:
                         raise WeaviateConnectionError(
                             f"Error: {_exc_detail(e)}. \nIs Weaviate running and reachable at {self.url}?"
@@ -447,7 +492,7 @@ async def get_oidc() -> None:
 
         with self._make_client("sync") as client:
             try:
-                response = client.get(oidc_url, timeout=self.timeout_config.init)
+                response = client.get(oidc_url, timeout=self._deadlines.init)
             except Exception as e:
                 raise WeaviateConnectionError(
                     f"Error: {_exc_detail(e)}. \nIs Weaviate running and reachable at {self.url}?"
@@ -557,15 +602,17 @@ def _create_background_token_refresh(self, _auth: Optional[_Auth] = None) -> Non
         if "refresh_token" not in self._client.token and _auth is None:
             return
 
-        # a previous connect() may have left a refresher behind (e.g. a retry after a
-        # partially failed connect); stop it before replacing the shutdown event, or it
-        # would keep refreshing concurrently forever
+        # stop the refresher a previous connect() may have left behind (e.g. a retry
+        # after a partially failed connect)
         self._cancel_background_token_refresh()
 
         expires_in: int = self._client.token.get(
             "expires_in", 60
         )  # use 1minute as token lifetime if not supplied
-        self._shutdown_background_event = Event()
+        # captured by the refresher below, so that a later close()+connect() (which
+        # replaces the attribute) still stops THIS refresher and not the new one
+        shutdown = Event()
+        self._shutdown_background_event = shutdown
 
         if isinstance(self._client, AsyncOAuth2Client):
             try:
@@ -575,7 +622,9 @@ def _create_background_token_refresh(self, _auth: Optional[_Auth] = None) -> Non
             if loop is not None:
                 # The async colour refreshes on its own already-running loop: threads
                 # cannot start under WASM/Pyodide and are unnecessary here anyway.
-                task = loop.create_task(self.__periodic_token_refresh_async(expires_in, _auth))
+                task = loop.create_task(
+                    self.__periodic_token_refresh_async(expires_in, _auth, shutdown)
+                )
                 task.add_done_callback(self.__warn_if_token_refresh_died)
                 self.__token_refresh_task = task
                 return
@@ -617,12 +666,9 @@ def update_refresh_time() -> int:
             return self._client.token.get("expires_in", 60) - 30
 
         def periodic_refresh_token(refresh_time: int, _auth: Optional[_Auth]) -> None:
-            while (
-                self._shutdown_background_event is not None
-                and not self._shutdown_background_event.is_set()
-            ):
-                # use refresh token when available
-                time.sleep(max(refresh_time, 1))
+            failures = 0
+            # Event.wait instead of time.sleep so close() ends the thread promptly
+            while not shutdown.wait(timeout=max(refresh_time, 1)):
                 try:
                     if self._client is None:
                         continue
@@ -636,10 +682,12 @@ def periodic_refresh_token(refresh_time: int, _auth: Optional[_Auth]) -> None:
                         # saved credentials
                         refresh_session()
                     refresh_time = update_refresh_time()
-                except HTTPError as exc:
-                    # retry again after one second, might be an unstable connection
-                    refresh_time = 1
-                    _Warnings.token_refresh_failed(exc)
+                    failures = 0
+                except Exception as exc:
+                    # any failure (not only transport errors) keeps the refresher alive
+                    failures += 1
+                    refresh_time = _token_refresh_backoff(failures)
+                    _Warnings.token_refresh_failed(exc, refresh_time, failures)
 
         demon = Thread(
             target=periodic_refresh_token,
@@ -649,25 +697,38 @@ def periodic_refresh_token(refresh_time: int, _auth: Optional[_Auth]) -> None:
         )
         demon.start()
 
-    def _cancel_background_token_refresh(self) -> None:
+    def _cancel_background_token_refresh(self) -> Optional["asyncio.Task[None]"]:
         """Stop the token refresher, whichever form it took.
 
-        Sets the shutdown event (observed by the sync colour's daemon thread at its next
-        wake-up) and cancels the async colour's refresh task immediately.
+        Sets the shutdown event (ends the sync colour's daemon thread) and cancels the
+        async colour's refresh task. Returns that task when it belongs to the running
+        loop so the caller can await its wind-down; a task on another loop is only asked
+        to cancel (thread-safely) and never awaited, which would hang.
         """
         if self._shutdown_background_event is not None:
             self._shutdown_background_event.set()
-        if self.__token_refresh_task is not None:
-            self.__token_refresh_task.cancel()
-            self.__token_refresh_task = None
+        task, self.__token_refresh_task = self.__token_refresh_task, None
+        if task is None:
+            return None
+        try:
+            running: Optional[asyncio.AbstractEventLoop] = asyncio.get_running_loop()
+        except RuntimeError:
+            running = None
+        loop = task.get_loop()
+        if loop is running:
+            task.cancel()
+            return task
+        if not loop.is_closed():
+            loop.call_soon_threadsafe(task.cancel)
+        return None
 
     @staticmethod
     def __warn_if_token_refresh_died(task: "asyncio.Task[None]") -> None:
         """Surface a refresher death that happened outside the loop body's own handler.
 
-        Nothing ever awaits this task and ``_cancel_background_token_refresh`` drops the
-        reference, so without this callback not even asyncio's "exception was never
-        retrieved" message fires — the client just silently stops refreshing.
+        Nothing awaits this task while it runs (close() only gathers it with
+        return_exceptions=True), so without this callback not even asyncio's "exception
+        was never retrieved" message fires — the client just silently stops refreshing.
         """
         if task.cancelled():
             return
@@ -676,17 +737,14 @@ def __warn_if_token_refresh_died(task: "asyncio.Task[None]") -> None:
             _Warnings.token_refresh_stopped(exc)
 
     async def __periodic_token_refresh_async(
-        self, refresh_time: int, _auth: Optional[_Auth]
+        self, refresh_time: int, _auth: Optional[_Auth], shutdown: Event
     ) -> None:
         """Thread-free equivalent of ``periodic_refresh_token`` for the async colour.
 
         Cancelled by ``close('async')``.
         """
-        while (
-            self._shutdown_background_event is not None
-            and not self._shutdown_background_event.is_set()
-        ):
-            # use refresh token when available
+        failures = 0
+        while not shutdown.is_set():
             await asyncio.sleep(max(refresh_time, 1))
             try:
                 client = self._client
@@ -701,17 +759,16 @@ async def __periodic_token_refresh_async(
                     new_session = await _Auth.aresult(_auth.get_auth_session())
                     client.token = await new_session.fetch_token()
                 refresh_time = client.token.get("expires_in", 60) - 30
+                failures = 0
             except asyncio.CancelledError:
                 # close() cancels this task — cancellation must never be swallowed
                 raise
             except Exception as exc:
-                # retry again after one second, might be an unstable connection.
-                # Catch broadly: an authlib OAuthError (e.g. invalid_grant), a KeyError on
-                # a missing token_endpoint or a malformed IdP body would otherwise kill the
-                # task with nobody watching, and every later request would 401 with nothing
-                # pointing at the token refresh.
-                refresh_time = 1
-                _Warnings.token_refresh_failed(exc)
+                # any failure (an authlib OAuthError such as invalid_grant, a KeyError on a
+                # malformed IdP body, ...) keeps the refresher alive
+                failures += 1
+                refresh_time = _token_refresh_backoff(failures)
+                _Warnings.token_refresh_failed(exc, refresh_time, failures)
 
     def __get_latest_headers(self) -> Dict[str, str]:
         if "authorization" in self._headers:
@@ -740,42 +797,32 @@ def __get_timeout(
         They specify the times depending on how they expect Weaviate to behave. For example, a query might take longer than an insert or vice versa
         but, in either case, the user only cares about how long it takes for a response to be received.
 
-        The one exception is Emscripten/Pyodide, where the connect timeout bounds the whole fetch
-        promise rather than connection setup, so it has to follow the request timeout (see below).
+        Under Emscripten/Pyodide the fetch transport (weaviate-client-web) reads only `read`,
+        so it is the whole-request deadline there; a non-finite value means no deadline.
 
         https://www.python-httpx.org/advanced/timeouts/
         """
+        deadlines = self._deadlines
         timeout = None
         if method == "DELETE" or method == "PATCH" or method == "PUT":
-            timeout = self.timeout_config.insert
+            timeout = deadlines.insert
         elif method == "GET" or method == "HEAD":
-            timeout = self.timeout_config.query
+            timeout = deadlines.query
         elif method == "POST" and is_gql_query:
-            timeout = self.timeout_config.query
+            timeout = deadlines.query
         elif method == "POST" and not is_gql_query:
-            timeout = self.timeout_config.insert
-
-        connect: float = 5.0
-        if sys.platform == "emscripten" and timeout is not None:
-            # Under Pyodide the REST transport is httpx's AsyncJavascriptFetchTransport,
-            # where the connect timeout bounds the WHOLE fetch promise (upload + response)
-            # rather than connection setup only. A fixed 5s there silently caps every
-            # request at ~5s of wall clock and ignores the configured insert/query
-            # timeouts. Never below 5s, and untouched off Emscripten, so native platforms
-            # keep the httpx default for connect/write.
-            connect = max(connect, float(timeout))
+            timeout = deadlines.insert
+
         return Timeout(
-            timeout=connect,
+            timeout=5.0,
             read=timeout,
             pool=self.__connection_config.session_pool_timeout,
         )
 
     def __handle_exceptions(self, e: Exception, error_msg: str) -> None:
-        # httpx raises a bare RuntimeError('Cannot send a request, as the client has been
-        # closed.'); match its message so unrelated RuntimeErrors (e.g. Emscripten's
-        # "can't start new thread") are not rewritten into a misleading 'client is
-        # closed' error.
-        if isinstance(e, RuntimeError) and "client has been closed" in str(e):
+        # httpx raises a bare RuntimeError for a closed client; only rewrite when the
+        # client really is closed, so unrelated RuntimeErrors keep their meaning
+        if isinstance(e, RuntimeError) and (self._client is None or self._client.is_closed):
             raise WeaviateClosedClientError() from e
         if isinstance(e, ConnectError):
             raise WeaviateConnectionError(self.__error_msg_with_detail(error_msg, e)) from e
@@ -848,10 +895,14 @@ def exc(e: Exception) -> None:
     def close(self, colour: executor.Colour) -> executor.Result[None]:
         if self.embedded_db is not None:
             self.embedded_db.stop()
-        self._cancel_background_token_refresh()
+        refresh_task = self._cancel_background_token_refresh()
         if colour == "async":
 
             async def execute() -> None:
+                if refresh_task is not None:
+                    # let the cancellation land before the client goes away; the task's
+                    # own outcome (cancelled, or dead earlier) is not close()'s error
+                    await asyncio.gather(refresh_task, return_exceptions=True)
                 if self._client is not None:
                     assert isinstance(self._client, AsyncClient)
                     await self._client.aclose()
@@ -889,7 +940,7 @@ def resp(res: Response) -> None:
             async def _execute() -> None:
                 try:
                     async with AsyncClient() as client:
-                        res = await client.get(PYPI_PACKAGE_URL, timeout=self.timeout_config.init)
+                        res = await client.get(PYPI_PACKAGE_URL, timeout=self._deadlines.init)
                     return resp(res)
                 except (RequestError, OSError):
                     # ignore any errors related to requests, it is a best-effort warning.
@@ -901,7 +952,7 @@ async def _execute() -> None:
 
         try:
             with Client() as client:
-                res = client.get(PYPI_PACKAGE_URL, timeout=self.timeout_config.init)
+                res = client.get(PYPI_PACKAGE_URL, timeout=self._deadlines.init)
             return resp(res)
         except (RequestError, OSError):
             pass  # ignore any errors related to requests, it is a best-effort warning
@@ -1039,13 +1090,22 @@ def resp(res: Response) -> Optional[Dict[str, Any]]:
 class ConnectionSync(_ConnectionBase):
     """Connection class used to communicate to a weaviate instance."""
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        connection_params: ConnectionParams,
+        auth_client_secret: Optional[AuthCredentials],
+        timeout_config: TimeoutConfig,
+        proxies: Union[str, Proxies, None],
+        trust_env: bool,
+        additional_headers: Optional[Dict[str, Any]],
+        connection_config: ConnectionConfig,
+        embedded_db: Optional[EmbeddedV4] = None,
+        skip_init_checks: bool = False,
+        grpc_config: Optional[GrpcConfig] = None,
+    ) -> None:
         if sys.platform == "emscripten":
-            # Fail at construction with the async-only message; otherwise the first
-            # REST call surfaces an opaque ConnectError long before the grpc-web
-            # shim's own sync guard is reached (wording mirrors the shim's
-            # _ASYNC_ONLY_MESSAGE). Pre-set the attributes __del__ reads so the
-            # never-initialized instance is collected quietly.
+            # fail at construction, before the first REST call surfaces an opaque
+            # ConnectError; pre-set what __del__ reads so the instance is collected quietly
             self._client = None
             self._grpc_channel = None
             raise WeaviateStartUpError(
@@ -1054,7 +1114,18 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
                 "use_async_with_weaviate_cloud / use_async_with_custom, or "
                 "WeaviateAsyncClient) instead."
             )
-        super().__init__(*args, **kwargs)
+        super().__init__(
+            connection_params=connection_params,
+            auth_client_secret=auth_client_secret,
+            timeout_config=timeout_config,
+            proxies=proxies,
+            trust_env=trust_env,
+            additional_headers=additional_headers,
+            connection_config=connection_config,
+            embedded_db=embedded_db,
+            skip_init_checks=skip_init_checks,
+            grpc_config=grpc_config,
+        )
 
     def connect(self, force: bool = False) -> None:
         if self._connected and not force:
@@ -1137,7 +1208,7 @@ def grpc_search(self, request: search_get_pb2.SearchRequest) -> search_get_pb2.S
                 self.grpc_stub.Search,
                 request,
                 metadata=self.grpc_headers(),
-                timeout=self.timeout_config.query,
+                timeout=self._deadlines.query,
             )
             return cast(search_get_pb2.SearchReply, res)
         except RpcError as e:
@@ -1162,7 +1233,7 @@ def grpc_batch_objects(
                 f=self.grpc_stub.BatchObjects,
                 request=request,
                 metadata=self.grpc_headers(),
-                timeout=timeout,
+                timeout=_deadline(timeout),
             )
             res = cast(batch_pb2.BatchObjectsReply, res)
 
@@ -1184,7 +1255,7 @@ def grpc_batch_stream(
             assert self.grpc_stub is not None
             for msg in self.grpc_stub.BatchStream(
                 request_iterator=requests,
-                timeout=self.timeout_config.stream,
+                timeout=self._deadlines.stream,
                 metadata=self.grpc_headers(),
             ):
                 yield msg
@@ -1206,7 +1277,7 @@ def grpc_batch_delete(
                 self.grpc_stub.BatchDelete(
                     request,
                     metadata=self.grpc_headers(),
-                    timeout=self.timeout_config.insert,
+                    timeout=self._deadlines.insert,
                 ),
             )
         except RpcError as e:
@@ -1226,7 +1297,7 @@ def grpc_tenants_get(
                 self.grpc_stub.TenantsGet,
                 request,
                 metadata=self.grpc_headers(),
-                timeout=self.timeout_config.query,
+                timeout=self._deadlines.query,
             )
         except RpcError as e:
             error = cast(Call, e)
@@ -1247,7 +1318,7 @@ def grpc_aggregate(
                 self.grpc_stub.Aggregate,
                 request,
                 metadata=self.grpc_headers(),
-                timeout=self.timeout_config.query,
+                timeout=self._deadlines.query,
             )
             return cast(aggregate_pb2.AggregateReply, res)
         except RpcError as e:
@@ -1348,7 +1419,7 @@ async def grpc_search(
                 self.grpc_stub.Search,
                 request,
                 metadata=self.grpc_headers(),
-                timeout=self.timeout_config.query,
+                timeout=self._deadlines.query,
             )
             return cast(search_get_pb2.SearchReply, res)
         except AioRpcError as e:
@@ -1372,7 +1443,7 @@ async def grpc_batch_objects(
                 f=self.grpc_stub.BatchObjects,
                 request=request,
                 metadata=self.grpc_headers(),
-                timeout=timeout,
+                timeout=_deadline(timeout),
             )
             res = cast(batch_pb2.BatchObjectsReply, res)
 
@@ -1383,7 +1454,7 @@ async def grpc_batch_objects(
         except AioRpcError as e:
             if e.code().name == PERMISSION_DENIED:
                 raise InsufficientPermissionsError(e)
-            raise WeaviateBatchError(str(e)) from e
+            raise WeaviateBatchError(str(e.details())) from e
 
     async def grpc_batch_delete(
         self, request: batch_delete_pb2.BatchDeleteRequest
@@ -1393,7 +1464,7 @@ async def grpc_batch_delete(
             return await self.grpc_stub.BatchDelete(
                 request,
                 metadata=self.grpc_headers(),
-                timeout=self.timeout_config.insert,
+                timeout=self._deadlines.insert,
             )
         except AioRpcError as e:
             if e.code().name == PERMISSION_DENIED:
@@ -1412,7 +1483,7 @@ async def grpc_batch_stream(
                 response_deserializer=batch_pb2.BatchStreamReply.FromString,
             )(
                 request_iterator=requests,
-                timeout=self.timeout_config.stream,
+                timeout=self._deadlines.stream,
                 metadata=self.grpc_headers(),
             ):
                 yield msg
@@ -1467,7 +1538,7 @@ async def grpc_tenants_get(
                 self.grpc_stub.TenantsGet,
                 request,
                 metadata=self.grpc_headers(),
-                timeout=self.timeout_config.query,
+                timeout=self._deadlines.query,
             )
         except AioRpcError as e:
             if e.code().name == PERMISSION_DENIED:
@@ -1487,7 +1558,7 @@ async def grpc_aggregate(
                 self.grpc_stub.Aggregate,
                 request,
                 metadata=self.grpc_headers(),
-                timeout=self.timeout_config.query,
+                timeout=self._deadlines.query,
             )
             return cast(aggregate_pb2.AggregateReply, res)
         except AioRpcError as e:
diff --git a/weaviate/exceptions.py b/weaviate/exceptions.py
index 2024a1258..3f0873a32 100644
--- a/weaviate/exceptions.py
+++ b/weaviate/exceptions.py
@@ -473,5 +473,8 @@ def __init__(self, pb: version.Version, grpc: version.Version) -> None:
         )
 
 
-class _BatchStreamShutdownError(Exception):
-    """Internal exception to signal that the batch stream was shutdown."""
+class _BatchStreamShutdownError(WeaviateBatchStreamError):
+    """Internal exception to signal that the batch stream was shutdown (gRPC ABORTED)."""
+
+    def __init__(self, message: str = "the server aborted the batch stream") -> None:
+        super().__init__(message)
diff --git a/weaviate/proto/v1/__init__.py b/weaviate/proto/v1/__init__.py
index 1ad304222..62b0910f4 100644
--- a/weaviate/proto/v1/__init__.py
+++ b/weaviate/proto/v1/__init__.py
@@ -16,15 +16,10 @@
 
 from weaviate.exceptions import WeaviateProtobufIncompatibility
 
-# Fallback grpcio version used only when grpcio is not installed as a distribution.
-# This happens under Pyodide/Emscripten, where grpcio has no wheel and is excluded
-# via the `sys_platform != "emscripten"` marker in setup.cfg; the grpc module itself
-# is provided there by a pure-Python shim (see the weaviate-client-web package).
-# On every normal install grpcio's metadata is present and the real version is used, so
-# this branch is not taken. Restricted to grpcio AND to Emscripten, so that a broken or
-# partial grpcio install on a normal platform (metadata missing) still surfaces as
-# PackageNotFoundError instead of silently selecting a fallback stub, and so a genuinely
-# missing protobuf (required and pure-Python under Pyodide) is never masked.
+# grpcio version assumed under Pyodide/Emscripten, where grpcio has no wheel (excluded by
+# the `sys_platform != "emscripten"` marker in setup.cfg) and the grpc module is the
+# weaviate-client-web shim. Restricted to grpcio AND Emscripten so a broken grpcio install
+# elsewhere still raises PackageNotFoundError, and a missing protobuf is never masked.
 _GRPCIO_FALLBACK_VERSION = "1.72.1"
 
 def get_version(pkg: str) -> version.Version:
diff --git a/weaviate/warnings.py b/weaviate/warnings.py
index 053bac634..a1dd3180b 100644
--- a/weaviate/warnings.py
+++ b/weaviate/warnings.py
@@ -67,13 +67,11 @@ def auth_cannot_parse_oidc_config(url: str) -> None:
         warnings.warn(message=msg, category=UserWarning, stacklevel=1)
 
     @staticmethod
-    def token_refresh_failed(exc: Exception) -> None:
+    def token_refresh_failed(exc: Exception, retry_in: float, failures: int) -> None:
+        detail = f"{type(exc).__name__}: {exc}" if str(exc) else repr(exc)
         warnings.warn(
-            message=f"""Con001: Could not reach token issuer for the periodic refresh. This client will automatically
-            retry to refresh. If the retry does not succeed, the client will become unauthenticated.
-
-            The cause might be an unstable internet connection or a problem with your authentication provider.
-            Exception: {exc}
+            message=f"""Con001: Token refresh failed ({detail}); retrying in {retry_in}s (consecutive failures: {failures}).
+            The client will become unauthenticated once the current token expires if the refresh keeps failing.
             """,
             category=UserWarning,
             stacklevel=1,