feat(python-sdk): honor stream_idle_timeout per read in AsyncVolume.read_file

The async client bounds each read (response head and every chunk) with
asyncio.wait_for when the caller passes an explicit stream_idle_timeout,
mirroring the JS SDK's streamIdleTimeoutMs watchdog and the envd stream
setup bound in #1558. Explicit values run on the regular transport so
they aren't capped by the streaming transport's 60s idle bound, and 0
disables idle bounding entirely. The sync client keeps ignoring the
parameter — it cannot interrupt a blocking read, so its bound stays on
the streaming transport.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: mishushakov

.changeset/python-pyqwest-rest-transport.md

@@ -11,12 +11,14 @@ API is unchanged — only the transport underneath is swapped — so logging
 event hooks, per-request timeouts, and headers behave as before.
 
 For streamed volume reads (`Volume.read_file(format="stream")`), a stalled
-stream is now bounded by a transport-wide idle read timeout of 60 seconds
-that resets on every chunk (still surfaced as `httpx.ReadTimeout`; matches
-the JS SDK's default stream idle timeout); the per-call
-`stream_idle_timeout` parameter is deprecated and ignored. Passing
-`request_timeout` to a streamed read (and to template uploads) now bounds
-the whole transfer rather than individual socket operations.
+stream is by default bounded by a transport-wide idle read timeout of
+60 seconds that resets on every chunk (still surfaced as
+`httpx.ReadTimeout`; matches the JS SDK's default stream idle timeout).
+`AsyncVolume.read_file` keeps honoring an explicit `stream_idle_timeout`
+per read (including `0` to disable); the sync client ignores it — it cannot
+interrupt a blocking read. Passing `request_timeout` to a streamed read
+(and to template uploads) now bounds the whole transfer rather than
+individual socket operations.
 
 Because pyqwest transports are thread-safe and loop-independent (I/O runs on
 a Rust runtime), the API connection pool is now shared process-wide per

packages/python-sdk/e2b/volume/volume_async.py

@@ -466,9 +466,13 @@ async def read_file(
 
         :param path: Path to the file
         :param format: Format of the file content—`text` by default
-        :param stream_idle_timeout: Deprecated and ignored. A stalled streamed
-            read is bounded by a transport-wide idle read timeout instead
-            (60 seconds), which resets on every chunk.
+        :param stream_idle_timeout: Idle timeout in **seconds** for a streamed
+            read (`format="stream"`)—abort with `httpx.ReadTimeout` if the
+            response head or the next chunk doesn't arrive within this
+            window. Resets on every chunk, so it bounds a stalled stream
+            without limiting total transfer time. Defaults to the
+            transport-wide idle read timeout (60 seconds); pass `0` to
+            disable.
         :param opts: Connection options
 
         :return: File content as string, bytes, or async iterator of bytes
@@ -486,27 +490,38 @@ async def read_file(
             # whole-request deadline that would kill long downloads, so a
             # streamed read is sent with one only when the caller set
             # `request_timeout` explicitly (making it the total-transfer
-            # deadline). A stalled stream is instead bounded by the
-            # transport-wide idle read timeout (see `get_transport`), which
-            # resets on every chunk without limiting total transfer time.
+            # deadline).
             stream_timeout = VolumeConnectionConfig._get_request_timeout(
                 None, opts.get("request_timeout")
             )
-            # The streaming transport carries the idle read timeout; the
-            # regular one must not (it would cut off slow uploads and
-            # responses), so streamed reads get their own client.
-            stream_client = get_volume_api_client(config, for_streaming=True)
+            # By default a stalled stream is bounded by the streaming
+            # transport's idle read timeout (see `get_transport`). An
+            # explicit `stream_idle_timeout` is applied per read with
+            # `wait_for` instead, on the regular transport — so values above
+            # the transport bound aren't capped by it and `0` disables idle
+            # bounding entirely. (The sync client can't interrupt a blocking
+            # read, so only the async flavor honors the per-call value.)
+            stream_client = get_volume_api_client(
+                config, for_streaming=stream_idle_timeout is None
+            )
+
+            async def read_bounded(awaitable):
+                if not stream_idle_timeout:
+                    return await awaitable
+                return await asyncio.wait_for(awaitable, stream_idle_timeout)
 
             async def stream_file() -> AsyncIterator[bytes]:
                 # pyqwest raises the builtin TimeoutError; keep the httpx
                 # exception the streamed-read contract established.
                 try:
-                    async with stream_client.get_async_httpx_client().stream(
+                    stream_cm = stream_client.get_async_httpx_client().stream(
                         method="GET",
                         url=f"/volumecontent/{self._volume_id}/file",
                         params=params,
                         timeout=stream_timeout,
-                    ) as response:
+                    )
+                    response = await read_bounded(stream_cm.__aenter__())
+                    try:
                         if response.status_code == 404:
                             raise NotFoundException(f"Path {path} not found")
 
@@ -519,10 +534,17 @@ async def stream_file() -> AsyncIterator[bytes]:
                             )
                             raise handle_api_exception(api_response, VolumeException)
 
-                        async for chunk in response.aiter_bytes():
+                        chunks = response.aiter_bytes()
+                        while True:
+                            try:
+                                chunk = await read_bounded(chunks.__anext__())
+                            except StopAsyncIteration:
+                                break
                             yield chunk
+                    finally:
+                        await stream_cm.__aexit__(None, None, None)
                 # asyncio.TimeoutError is distinct from the builtin until 3.11,
-                # and the adapter's whole-request deadline raises it.
+                # and wait_for and the adapter's whole-request deadline raise it.
                 except (TimeoutError, asyncio.TimeoutError) as e:
                     raise httpx.ReadTimeout(str(e)) from e
 

packages/python-sdk/e2b/volume/volume_sync.py

@@ -463,9 +463,11 @@ def read_file(
 
         :param path: Path to the file
         :param format: Format of the file content—`text` by default
-        :param stream_idle_timeout: Deprecated and ignored. A stalled streamed
-            read is bounded by a transport-wide idle read timeout instead
-            (60 seconds), which resets on every chunk.
+        :param stream_idle_timeout: Ignored — the sync client cannot
+            interrupt a blocking read. A stalled streamed read is bounded by
+            a transport-wide idle read timeout instead (60 seconds), which
+            resets on every chunk. (`AsyncVolume.read_file` honors this
+            parameter.)
         :param opts: Connection options
 
         :return: File content as string, bytes, or iterator of bytes

packages/python-sdk/tests/test_volume_client.py

@@ -184,8 +184,9 @@ def short_read_timeout(monkeypatch):
 def test_sync_stream_survives_transfers_longer_than_read_timeout(short_read_timeout):
     # The transport read timeout is an idle bound that resets on every chunk:
     # a healthy stream whose total duration exceeds it must complete.
-    # `stream_idle_timeout` is deprecated and ignored — a value shorter than
-    # every chunk gap must not abort the stream.
+    # `stream_idle_timeout` is ignored in the sync client (it cannot
+    # interrupt a blocking read) — a value shorter than every chunk gap must
+    # not abort the stream.
     api_url = _start_volume_file_server([0.15] * 4)
     volume = Volume(volume_id="v1", name="test", token="vol-token")
 
@@ -214,9 +215,7 @@ def test_async_stream_survives_transfers_longer_than_read_timeout(short_read_tim
     volume = AsyncVolume(volume_id="v1", name="test", token="vol-token")
 
     async def run():
-        stream = await volume.read_file(
-            "file.bin", format="stream", stream_idle_timeout=0.01, api_url=api_url
-        )
+        stream = await volume.read_file("file.bin", format="stream", api_url=api_url)
         return b"".join([chunk async for chunk in stream])
 
     assert asyncio.run(run()) == CHUNK * 4
@@ -237,6 +236,59 @@ async def run():
     assert asyncio.run(run()) == [CHUNK]
 
 
+def test_async_explicit_stream_idle_timeout_aborts_stall():
+    # An explicit stream_idle_timeout is honored per read with wait_for
+    # (like the JS SDK's streamIdleTimeoutMs) — no transport rebuild needed.
+    reset_volume_transports()
+    api_url = _start_volume_file_server([0.0, 5.0])
+    volume = AsyncVolume(volume_id="v1", name="test", token="vol-token")
+
+    async def run():
+        stream = await volume.read_file(
+            "file.bin", format="stream", stream_idle_timeout=0.3, api_url=api_url
+        )
+        received = [await stream.__anext__()]
+        with pytest.raises(httpx.ReadTimeout):
+            async for chunk in stream:
+                received.append(chunk)
+        return received
+
+    try:
+        assert asyncio.run(run()) == [CHUNK]
+    finally:
+        reset_volume_transports()
+
+
+def test_async_explicit_stream_idle_timeout_above_transport_bound(short_read_timeout):
+    # An explicit value larger than the transport's idle read timeout must
+    # not be capped by it: explicit values run on the regular transport.
+    api_url = _start_volume_file_server([short_read_timeout * 1.5] * 3)
+    volume = AsyncVolume(volume_id="v1", name="test", token="vol-token")
+
+    async def run():
+        stream = await volume.read_file(
+            "file.bin", format="stream", stream_idle_timeout=5.0, api_url=api_url
+        )
+        return b"".join([chunk async for chunk in stream])
+
+    assert asyncio.run(run()) == CHUNK * 3
+
+
+def test_async_stream_idle_timeout_zero_disables_idle_bound(short_read_timeout):
+    # `stream_idle_timeout=0` disables idle bounding entirely — a stall
+    # longer than the transport's idle read timeout must not abort.
+    api_url = _start_volume_file_server([0.0, short_read_timeout * 3])
+    volume = AsyncVolume(volume_id="v1", name="test", token="vol-token")
+
+    async def run():
+        stream = await volume.read_file(
+            "file.bin", format="stream", stream_idle_timeout=0, api_url=api_url
+        )
+        return b"".join([chunk async for chunk in stream])
+
+    assert asyncio.run(run()) == CHUNK * 2
+
+
 def test_stream_transport_is_separate_from_regular_transport():
     # reqwest's read timer keeps running while a request body is sent and
     # while waiting for the response head, so the idle read timeout lives on