fix(client): FileSet upload retries and timeout propagation

packages/filesets/src/filesets/filesystem/filesystem.py

@@ -358,24 +358,24 @@ def _ensure_async(client: FilesClient | AsyncFilesClient) -> AsyncFilesClient:
 
         import httpx
 
+        # A timeout lives in two layers, so mirror each from its own source: the
+        # transport carries the client-level default, and ``_timeout`` the
+        # per-request override that ``send`` puts on every request. Leave the
+        # transport's unset and httpx falls back to its own 5s, which a multi-GB
+        # upload blows through waiting for the server to commit the body to storage.
         asgi_app = getattr(client._http, "asgi_app", None)
-        http_client = (
-            httpx.AsyncClient(
-                transport=httpx.ASGITransport(app=asgi_app),
-                base_url=client.base_url,
-                headers=dict(client._default_headers) if client._default_headers else None,
-            )
-            if asgi_app is not None
-            else httpx.AsyncClient(
-                base_url=client.base_url,
-                headers=dict(client._default_headers) if client._default_headers else None,
-            )
+        http_client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=asgi_app) if asgi_app is not None else None,
+            base_url=client.base_url,
+            headers=dict(client._default_headers) if client._default_headers else None,
+            timeout=client._http.timeout,
         )
         return AsyncFilesClient(
             base_url=client.base_url,
             workspace=client.workspace,
             auth=client._auth,
             default_headers=client._default_headers or None,
+            timeout=client._timeout,
             retry=client._retry,
             http_client=http_client,
             url_resolver=client._url_resolver,

packages/filesets/tests/test_filesystem_client.py

@@ -0,0 +1,80 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Client construction inside FilesetFileSystem.
+
+Uploads and downloads run on the async client built by ``_ensure_async``, not on
+the sync client the caller configured. Anything that client fails to carry over
+is silently dropped from every transfer.
+"""
+
+from __future__ import annotations
+
+import httpx
+from filesets.filesystem.filesystem import FilesetFileSystem
+from nemo_platform_plugin.client.types import RetryPolicy
+from nemo_platform_plugin.files.client import FilesClient
+
+BASE = "http://test:8000"
+UPLOAD_TIMEOUT = httpx.Timeout(30.0, write=10 * 60, read=5 * 60)
+
+
+def _sync_client(*, timeout: httpx.Timeout) -> FilesClient:
+    http_client = httpx.Client(
+        transport=httpx.MockTransport(lambda request: httpx.Response(200, request=request)),
+        timeout=timeout,
+    )
+    return FilesClient(
+        base_url=BASE,
+        workspace="default",
+        http_client=http_client,
+        retry=RetryPolicy(max_retries=2),
+    )
+
+
+def test_ensure_async_carries_transport_timeout() -> None:
+    """With no override to carry, the transport's own timeout is what governs.
+
+    Without this the new AsyncClient falls back to httpx's 5s default.
+    """
+    async_client = FilesetFileSystem._ensure_async(_sync_client(timeout=httpx.Timeout(60.0)))
+
+    assert async_client._timeout is None
+    assert async_client._http.timeout == httpx.Timeout(60.0)
+    assert async_client._http.timeout != httpx.Timeout(5.0)
+
+
+def test_ensure_async_carries_per_request_timeout_override() -> None:
+    """An override goes out on every request, so it governs regardless of the transport."""
+    client = _sync_client(timeout=httpx.Timeout(60.0)).with_options(timeout=UPLOAD_TIMEOUT)
+
+    async_client = FilesetFileSystem._ensure_async(client)
+
+    assert async_client._timeout == UPLOAD_TIMEOUT
+    # Each layer is copied from its own counterpart, so the transport keeps the
+    # client-level default it had on the sync side rather than the override.
+    assert async_client._http.timeout == httpx.Timeout(60.0)
+
+
+def test_ensure_async_preserves_workspace_and_retry() -> None:
+    client = _sync_client(timeout=httpx.Timeout(60.0))
+
+    async_client = FilesetFileSystem._ensure_async(client)
+
+    assert async_client.workspace == "default"
+    assert async_client.retry == RetryPolicy(max_retries=2)
+
+
+def test_upload_timeout_survives_the_whole_client_chain() -> None:
+    """End to end: an SDK-level timeout override reaches the client that transfers."""
+    from nemo_platform import NeMoPlatform
+
+    http_client = httpx.Client(
+        transport=httpx.MockTransport(lambda request: httpx.Response(200, request=request)),
+        timeout=httpx.Timeout(60.0),
+    )
+    platform = NeMoPlatform(base_url=BASE, workspace="default", http_client=http_client)
+
+    fs = platform.with_options(timeout=UPLOAD_TIMEOUT).files.fsspec
+
+    assert fs._client._timeout == UPLOAD_TIMEOUT

packages/nemo_platform_plugin/src/nemo_platform_plugin/client/adapter.py

@@ -69,23 +69,39 @@ def client_from_platform(
         respect_retry_after_headers=True,
     )
     url_resolver = _url_resolver_from_platform(platform)
+
+    # Carry the platform's timeout across as a per-request override. The shared
+    # httpx client keeps whatever timeout it was built with, so a caller's
+    # ``platform.with_options(timeout=...)`` would otherwise be silently dropped
+    # on the way to the typed client — the httpx client it hands over is the
+    # *same* object, with the *original* timeout still on it.
+    timeout = platform.timeout
+    if timeout is None:
+        # ``None`` on the platform means "no timeout at all", but the typed
+        # client reads None as "defer to the transport". Say the same thing in
+        # the form httpx itself uses, so the override survives.
+        timeout = httpx.Timeout(None)
+
     if isinstance(platform, AsyncNeMoPlatform):
         if not issubclass(client_cls, AsyncNemoClient):
             raise TypeError("AsyncNeMoPlatform requires an AsyncNemoClient class")
         return client_cls(
             base_url=str(platform.base_url).rstrip("/"),
             workspace=platform.workspace,
             default_headers=headers or None,
+            timeout=timeout,
             retry=retry,
             http_client=platform._client,
             url_resolver=url_resolver,
         )
+
     if not issubclass(client_cls, NemoClient):
         raise TypeError("NeMoPlatform requires a NemoClient class")
     return client_cls(
         base_url=str(platform.base_url).rstrip("/"),
         workspace=platform.workspace,
         default_headers=headers or None,
+        timeout=timeout,
         retry=retry,
         http_client=platform._client,
         url_resolver=url_resolver,