Route browser fs and logs endpoints directly to the VM

Add fs and logs to the default direct-to-VM subresource prefixes so
filesystem operations and log streaming use the cached browser base_url
and JWT instead of the control plane.

Serialize multipart array entries with indexed names so a file part stays
associated with the sibling fields of its array entry, which repeated
`files[][file]` names cannot express.

Only retry a stale direct-to-VM auth failure on the control plane when the
request body can be rebuilt byte for byte; a streamed body is consumed by
the direct attempt, so retrying would send a truncated body. The stale
route is evicted either way.

Author: tnsardesai

src/kernel/_base_client.py

@@ -587,7 +587,10 @@ 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
-            array_format="brackets",
+            # 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",
         )
         serialized: dict[str, object] = {}
         for key, value in items:

src/kernel/_client.py

@@ -41,6 +41,7 @@
     strip_direct_vm_auth,
     rewrite_direct_vm_options,
     browser_routing_config_from_env,
+    is_stale_direct_vm_auth_response,
     should_retry_stale_direct_vm_auth,
     maybe_evict_browser_route_from_response,
     maybe_populate_browser_route_cache_from_response,
@@ -365,9 +366,12 @@ def _prepare_request(self, request: httpx.Request) -> None:
 
     @override
     def _should_retry(self, response: httpx.Response) -> bool:
-        if should_retry_stale_direct_vm_auth(response):
+        if is_stale_direct_vm_auth_response(response):
             maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
-            return True
+            # 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
@@ -748,9 +752,12 @@ async def _prepare_request(self, request: httpx.Request) -> None:
 
     @override
     def _should_retry(self, response: httpx.Response) -> bool:
-        if should_retry_stale_direct_vm_auth(response):
+        if is_stale_direct_vm_auth_response(response):
             maybe_evict_browser_route_from_response(response, cache=self.browser_route_cache)
-            return True
+            # 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

src/kernel/_utils/_utils.py

@@ -40,15 +40,17 @@ def extract_files(
     query: Mapping[str, object],
     *,
     paths: Sequence[Sequence[str]],
-    array_format: ArrayFormat = "brackets",
+    array_format: ArrayFormat = "indices",
 ) -> 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]``).
+    ``"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.
 
     Note: this mutates the given dictionary.
     """

src/kernel/lib/browser_routing/routing.py

@@ -44,7 +44,17 @@ def browser_routing_config_from_env() -> BrowserRoutingConfig:
         # Path prefixes eligible for direct-to-VM routing. "telemetry/stream" is
         # the live SSE endpoint (VM); "telemetry/events" is a historical read
         # served by the control plane (S2) and must NOT be here.
-        return BrowserRoutingConfig(subresources=("curl", "telemetry/stream", "computer", "playwright", "process"))
+        return BrowserRoutingConfig(
+            subresources=(
+                "curl",
+                "telemetry/stream",
+                "computer",
+                "playwright",
+                "process",
+                "fs",
+                "logs",
+            )
+        )
     if raw.strip() == "":
         return BrowserRoutingConfig()
 
@@ -189,7 +199,49 @@ def is_stale_direct_vm_auth_response(response: httpx.Response) -> bool:
 
 
 def should_retry_stale_direct_vm_auth(response: httpx.Response) -> bool:
-    return is_stale_direct_vm_auth_response(response)
+    """Whether a stale direct-to-VM auth failure can be retried on the control plane.
+
+    A retry rebuilds the request from the original options, so it is only safe when
+    the body can be serialized again byte for byte. Streamed bodies (e.g. a file
+    object passed to fs.write_file) are consumed by the direct request, so retrying
+    would send a truncated or empty body to the control plane.
+    """
+    if not is_stale_direct_vm_auth_response(response):
+        return False
+    return direct_vm_request_body_is_replayable(response.request)
+
+
+def direct_vm_request_body_is_replayable(request: httpx.Request) -> bool:
+    try:
+        _ = request.content
+    except httpx.RequestNotRead:
+        pass
+    else:
+        # httpx already buffered the body, so rebuilding it yields the same bytes.
+        return True
+
+    # httpx encodes multipart bodies as a stream of fields it re-renders per attempt.
+    fields = getattr(request.stream, "fields", None)
+    if fields is None:
+        # A streamed body (file object, iterator or async iterator) cannot be replayed.
+        return False
+    return all(_multipart_field_is_replayable(field) for field in cast("list[Any]", fields))
+
+
+def _multipart_field_is_replayable(field: Any) -> bool:
+    file = getattr(field, "file", None)
+    if file is None:
+        # A data field renders from an in-memory value.
+        return True
+    if isinstance(file, (bytes, str)):
+        return True
+    if getattr(file, "closed", False):
+        return False
+    if not callable(getattr(file, "seek", None)):
+        return False
+    seekable = getattr(file, "seekable", None)
+    # httpx rewinds seekable file fields before rendering them again.
+    return bool(seekable()) if callable(seekable) else True
 
 
 def _session_id_from_browser_delete_path(path: str) -> str | None: