Address review: narrow multipart change, prove rewinds, always evict

Scope the indexed multipart array names to fs.upload instead of changing
the client's generic array encoding: the endpoint now flattens its own body
with indexed names and asks extract_files for matching file part names, so
load_extensions and any other multipart array keep their existing wire
format.

Prove a multipart file field can be rewound before treating a stale-JWT
failure as retryable. A wrapper can report seekable() while seek() raises,
which rendered the fallback body as an empty part.

Evict a stale direct-to-VM route from the terminal error path too. Retry
eligibility is only consulted when retries remain, so with max_retries=0 a
VM 401/403 previously left the dead route cached and wedged later calls.

Route only logs/stream rather than the whole logs subresource.

Author: tnsardesai

src/kernel/_base_client.py

@@ -587,10 +587,7 @@ def _serialize_multipartform(self, data: Mapping[object, object]) -> dict[str, o
             # TODO: type ignore is required as stringify_items is well typed but we can't be
             # well typed without heavy validation.
             data,  # type: ignore
-            # Indexed names (`files[0][dest_path]`) keep each array entry's fields
-            # grouped together; repeated `files[][dest_path]` parts cannot be
-            # matched back to their file part. `extract_files` uses the same format.
-            array_format="indices",
+            array_format="brackets",
         )
         serialized: dict[str, object] = {}
         for key, value in items:

src/kernel/_client.py

@@ -364,16 +364,28 @@ def _prepare_options(self, options: Any) -> Any:
     def _prepare_request(self, request: httpx.Request) -> None:
         strip_direct_vm_auth(request, cache=self.browser_route_cache)
 
+    def _evict_stale_direct_vm_route(self, response: httpx.Response) -> None:
+        maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
+
     @override
     def _should_retry(self, response: httpx.Response) -> bool:
         if is_stale_direct_vm_auth_response(response):
-            maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
+            self._evict_stale_direct_vm_route(response)
             # The route is evicted either way; only retry when the body can be
             # rebuilt, otherwise the caller sees the original auth failure and a
             # later call goes to the control plane.
             return should_retry_stale_direct_vm_auth(response)
         return super()._should_retry(response)
 
+    @override
+    def _make_status_error_from_response(self, response: httpx.Response) -> APIStatusError:
+        # `_should_retry` never runs when the request has no retries left, so this
+        # is the only place a stale direct-to-VM route gets evicted before the
+        # error surfaces to the caller.
+        if is_stale_direct_vm_auth_response(response):
+            self._evict_stale_direct_vm_route(response)
+        return super()._make_status_error_from_response(response)
+
     @override
     def _process_response(
         self,
@@ -750,16 +762,28 @@ async def _prepare_options(self, options: Any) -> Any:
     async def _prepare_request(self, request: httpx.Request) -> None:
         strip_direct_vm_auth(request, cache=self.browser_route_cache)
 
+    def _evict_stale_direct_vm_route(self, response: httpx.Response) -> None:
+        maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
+
     @override
     def _should_retry(self, response: httpx.Response) -> bool:
         if is_stale_direct_vm_auth_response(response):
-            maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
+            self._evict_stale_direct_vm_route(response)
             # The route is evicted either way; only retry when the body can be
             # rebuilt, otherwise the caller sees the original auth failure and a
             # later call goes to the control plane.
             return should_retry_stale_direct_vm_auth(response)
         return super()._should_retry(response)
 
+    @override
+    def _make_status_error_from_response(self, response: httpx.Response) -> APIStatusError:
+        # `_should_retry` never runs when the request has no retries left, so this
+        # is the only place a stale direct-to-VM route gets evicted before the
+        # error surfaces to the caller.
+        if is_stale_direct_vm_auth_response(response):
+            self._evict_stale_direct_vm_route(response)
+        return super()._make_status_error_from_response(response)
+
     @override
     async def _process_response(
         self,

src/kernel/_utils/_utils.py

@@ -40,17 +40,15 @@ def extract_files(
     query: Mapping[str, object],
     *,
     paths: Sequence[Sequence[str]],
-    array_format: ArrayFormat = "indices",
+    array_format: ArrayFormat = "brackets",
 ) -> list[tuple[str, FileTypes]]:
     """Recursively extract files from the given dictionary based on specified paths.
 
     A path may look like this ['foo', 'files', '<array>', 'data'].
 
     ``array_format`` controls how ``<array>`` segments contribute to the emitted
     field name. Supported values: ``"brackets"`` (``foo[]``), ``"repeat"`` and
-    ``"comma"`` (``foo``), ``"indices"`` (``foo[0]``, ``foo[1]``). Indexed names are
-    the default so that a file part stays associated with the sibling fields of the
-    same array entry, which repeated ``foo[]`` names cannot express.
+    ``"comma"`` (``foo``), ``"indices"`` (``foo[0]``, ``foo[1]``).
 
     Note: this mutates the given dictionary.
     """

src/kernel/lib/browser_routing/routing.py

@@ -52,7 +52,7 @@ def browser_routing_config_from_env() -> BrowserRoutingConfig:
                 "playwright",
                 "process",
                 "fs",
-                "logs",
+                "logs/stream",
             )
         )
     if raw.strip() == "":
@@ -237,11 +237,40 @@ def _multipart_field_is_replayable(field: Any) -> bool:
         return True
     if getattr(file, "closed", False):
         return False
-    if not callable(getattr(file, "seek", None)):
+    return _rewind_succeeds(file)
+
+
+def _rewind_succeeds(file: Any) -> bool:
+    """Whether the file field can actually be rewound for another render.
+
+    `seekable()` is not proof: a wrapper can report True and still raise from
+    `seek()`, which would render the field as an empty part on the retry. The
+    only reliable check is to perform the rewind httpx would perform.
+    """
+    seek = getattr(file, "seek", None)
+    if not callable(seek):
+        return False
+
+    position: object = None
+    tell = getattr(file, "tell", None)
+    if callable(tell):
+        try:
+            position = tell()
+        except Exception:
+            position = None
+
+    try:
+        seek(0)
+    except Exception:
         return False
-    seekable = getattr(file, "seekable", None)
-    # httpx rewinds seekable file fields before rendering them again.
-    return bool(seekable()) if callable(seekable) else True
+
+    if isinstance(position, int) and position > 0:
+        try:
+            seek(position)
+        except Exception:
+            # The field is left rewound, which is where httpx renders it from anyway.
+            pass
+    return True
 
 
 def _session_id_from_browser_delete_path(path: str) -> str | None:

src/kernel/lib/multipart.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import Mapping, Sequence, cast
+
+from .._utils import is_given
+
+__all__ = ["indexed_multipart_body"]
+
+
+def indexed_multipart_body(body: object) -> dict[str, object]:
+    """Flatten a multipart body so that array entries carry their index.
+
+    Endpoints that take an array of objects with a file field need each entry's
+    fields grouped together: `files[0][dest_path]` pairs with the `files[0][file]`
+    part, while repeated `files[][dest_path]` names cannot be matched back to
+    their file. The returned mapping is already flat, so the client's generic
+    multipart serialization passes the names through untouched and every other
+    endpoint keeps its existing encoding.
+    """
+    flattened: dict[str, object] = {}
+    if isinstance(body, Mapping):
+        for key, value in cast(Mapping[object, object], body).items():
+            _flatten(str(key), value, flattened)
+    return flattened
+
+
+def _flatten(key: str, value: object, out: dict[str, object]) -> None:
+    if not is_given(value):
+        return
+
+    if isinstance(value, Mapping):
+        for child_key, child in cast(Mapping[object, object], value).items():
+            _flatten(f"{key}[{child_key}]", child, out)
+        return
+
+    if isinstance(value, (list, tuple)):
+        for index, child in enumerate(cast(Sequence[object], value)):
+            _flatten(f"{key}[{index}]", child, out)
+        return
+
+    out[key] = value

src/kernel/resources/browsers/fs/fs.py

@@ -48,6 +48,7 @@
     async_to_custom_streamed_response_wrapper,
 )
 from ...._base_client import make_request_options
+from ....lib.multipart import indexed_multipart_body
 from ....types.browsers import (
     f_move_params,
     f_upload_params,
@@ -510,14 +511,19 @@ def upload(
             raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")
         extra_headers = {"Accept": "*/*", **(extra_headers or {})}
         body = deepcopy_with_paths({"files": files}, [["files", "<array>", "file"]])
-        extracted_files = extract_files(cast(Mapping[str, object], body), paths=[["files", "<array>", "file"]])
+        # The remote filesystem pairs each file part with the sibling fields of the
+        # same array entry, so both halves of the form use indexed names
+        # (`files[0][file]`, `files[0][dest_path]`).
+        extracted_files = extract_files(
+            cast(Mapping[str, object], body), paths=[["files", "<array>", "file"]], array_format="indices"
+        )
         # It should be noted that the actual Content-Type header that will be
         # sent to the server will contain a `boundary` parameter, e.g.
         # multipart/form-data; boundary=---abc--
         extra_headers["Content-Type"] = "multipart/form-data"
         return self._post(
             path_template("/browsers/{id}/fs/upload", id=id),
-            body=maybe_transform(body, f_upload_params.FUploadParams),
+            body=indexed_multipart_body(maybe_transform(body, f_upload_params.FUploadParams)),
             files=extracted_files,
             options=make_request_options(
                 extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout
@@ -1073,14 +1079,19 @@ async def upload(
             raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")
         extra_headers = {"Accept": "*/*", **(extra_headers or {})}
         body = deepcopy_with_paths({"files": files}, [["files", "<array>", "file"]])
-        extracted_files = extract_files(cast(Mapping[str, object], body), paths=[["files", "<array>", "file"]])
+        # The remote filesystem pairs each file part with the sibling fields of the
+        # same array entry, so both halves of the form use indexed names
+        # (`files[0][file]`, `files[0][dest_path]`).
+        extracted_files = extract_files(
+            cast(Mapping[str, object], body), paths=[["files", "<array>", "file"]], array_format="indices"
+        )
         # It should be noted that the actual Content-Type header that will be
         # sent to the server will contain a `boundary` parameter, e.g.
         # multipart/form-data; boundary=---abc--
         extra_headers["Content-Type"] = "multipart/form-data"
         return await self._post(
             path_template("/browsers/{id}/fs/upload", id=id),
-            body=await async_maybe_transform(body, f_upload_params.FUploadParams),
+            body=indexed_multipart_body(await async_maybe_transform(body, f_upload_params.FUploadParams)),
             files=extracted_files,
             options=make_request_options(
                 extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout