feat: control-plane fallback for direct-VM browser routing

Add a transparent control-plane fallback to the existing direct-to-VM
routing. When a GET/HEAD that was routed to a browser VM (allowlisted
subresource + cached route) fails because the VM is unreachable or the
session is gone, the original request is re-issued once to the control
plane (original URL, Authorization restored, no jwt param).

Triggers: connection/transport error, HTTP 502/503/504, or the clean
gone signal (404 + X-Kernel-Upstream: gone + body code browser_gone).
Never falls back for POST/PUT/PATCH/DELETE, ordinary live-VM 4xx, or
non-routed requests. Exactly one fallback attempt; the SDK's own retry,
cache eviction, and stream handling are left intact.

Implemented by overriding request() in Kernel/AsyncKernel to wrap the
base loop: eligibility is computed from the pre-rewrite options, and a
ContextVar suppresses re-routing during the single fallback re-issue.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rgarcia

src/kernel/_client.py

@@ -4,7 +4,8 @@
 
 import os
 from typing import TYPE_CHECKING, Any, Dict, Type, Mapping, cast
-from typing_extensions import Self, Literal, override
+from contextvars import ContextVar
+from typing_extensions import Self, Literal, overload, override
 
 import httpx
 
@@ -25,22 +26,26 @@
     is_mapping_t,
     get_async_library,
 )
-from ._compat import cached_property
+from ._compat import model_copy, cached_property
 from ._models import FinalRequestOptions
 from ._version import __version__
 from ._streaming import Stream as Stream, AsyncStream as AsyncStream
-from ._exceptions import KernelError, APIStatusError
+from ._exceptions import KernelError, APIStatusError, APIConnectionError
 from ._base_client import (
     DEFAULT_MAX_RETRIES,
     SyncAPIClient,
     AsyncAPIClient,
+    _StreamT,
+    _AsyncStreamT,
 )
 from .lib.browser_routing.routing import (
     BrowserRouteCache,
     BrowserRoutingConfig,
     strip_direct_vm_auth,
     rewrite_direct_vm_options,
+    is_vm_unreachable_response,
     browser_routing_config_from_env,
+    should_fallback_to_control_plane,
     maybe_evict_browser_route_from_response,
     maybe_populate_browser_route_cache_from_response,
 )
@@ -92,6 +97,13 @@
     "development": "https://localhost:3001/",
 }
 
+# Set (per thread / per async task) only during a control-plane fallback attempt
+# so `_prepare_options` skips direct-VM rewriting for that single re-issued request.
+# A ContextVar (rather than an instance attribute) keeps the flag local to the
+# in-flight request and avoids interfering with other concurrent requests sharing
+# the same client.
+_disable_browser_routing: ContextVar[bool] = ContextVar("kernel_disable_browser_routing", default=False)
+
 
 class Kernel(SyncAPIClient):
     # client options
@@ -307,12 +319,98 @@ def default_headers(self) -> dict[str, str | Omit]:
     @override
     def _prepare_options(self, options: Any) -> Any:
         options = cast(Any, super()._prepare_options(options))
+        # During a control-plane fallback attempt we deliberately skip direct-VM
+        # rewriting so the original request is re-issued to the control plane.
+        if _disable_browser_routing.get():
+            return options
         return rewrite_direct_vm_options(options, cache=self.browser_route_cache, config=self._browser_routing)
 
     @override
     def _prepare_request(self, request: httpx.Request) -> None:
         strip_direct_vm_auth(request, cache=self.browser_route_cache)
 
+    @overload
+    def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: Literal[True],
+        stream_cls: Type[_StreamT],
+    ) -> _StreamT: ...
+
+    @overload
+    def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: Literal[False] = False,
+    ) -> ResponseT: ...
+
+    @overload
+    def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: bool = False,
+        stream_cls: Type[_StreamT] | None = None,
+    ) -> ResponseT | _StreamT: ...
+
+    @override
+    def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: bool = False,
+        stream_cls: type[_StreamT] | None = None,
+    ) -> ResponseT | _StreamT:
+        # Capture the ORIGINAL, pre-rewrite control-plane options. `request`
+        # rewrites a routed request to the VM internally (via `_prepare_options`),
+        # so we must decide fallback eligibility from these untouched options.
+        original_options = model_copy(options)
+        eligible = _disable_browser_routing.get() is False and should_fallback_to_control_plane(
+            original_options, cache=self.browser_route_cache, config=self._browser_routing
+        )
+
+        try:
+            return super().request(cast_to, options, stream=stream, stream_cls=stream_cls)
+        except APIStatusError as err:
+            # Direct-VM attempt completed but the VM is unreachable/gone. The SDK's
+            # own retry logic has already run and given up by the time we get here.
+            if eligible and is_vm_unreachable_response(err.response):
+                return self._control_plane_fallback(
+                    cast_to, original_options, stream=stream, stream_cls=stream_cls
+                )
+            raise
+        except APIConnectionError:
+            # Connection/network/transport error (incl. timeouts) talking to the VM.
+            if eligible:
+                return self._control_plane_fallback(
+                    cast_to, original_options, stream=stream, stream_cls=stream_cls
+                )
+            raise
+
+    def _control_plane_fallback(
+        self,
+        cast_to: Type[ResponseT],
+        original_options: FinalRequestOptions,
+        *,
+        stream: bool,
+        stream_cls: type[_StreamT] | None,
+    ) -> ResponseT | _StreamT:
+        # Re-issue the ORIGINAL request to the control plane exactly once: original
+        # URL, Authorization restored (default auth headers are re-applied because
+        # routing is disabled, so `strip_direct_vm_auth` is a no-op), no jwt param.
+        # The flag prevents re-routing to the VM during this single attempt.
+        token = _disable_browser_routing.set(True)
+        try:
+            return super().request(cast_to, model_copy(original_options), stream=stream, stream_cls=stream_cls)
+        finally:
+            _disable_browser_routing.reset(token)
+
     @override
     def _process_response(
         self,
@@ -638,12 +736,100 @@ def default_headers(self) -> dict[str, str | Omit]:
     @override
     async def _prepare_options(self, options: Any) -> Any:
         options = cast(Any, await super()._prepare_options(options))
+        # During a control-plane fallback attempt we deliberately skip direct-VM
+        # rewriting so the original request is re-issued to the control plane.
+        if _disable_browser_routing.get():
+            return options
         return rewrite_direct_vm_options(options, cache=self.browser_route_cache, config=self._browser_routing)
 
     @override
     async def _prepare_request(self, request: httpx.Request) -> None:
         strip_direct_vm_auth(request, cache=self.browser_route_cache)
 
+    @overload
+    async def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: Literal[False] = False,
+    ) -> ResponseT: ...
+
+    @overload
+    async def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: Literal[True],
+        stream_cls: type[_AsyncStreamT],
+    ) -> _AsyncStreamT: ...
+
+    @overload
+    async def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: bool = False,
+        stream_cls: type[_AsyncStreamT] | None = None,
+    ) -> ResponseT | _AsyncStreamT: ...
+
+    @override
+    async def request(
+        self,
+        cast_to: Type[ResponseT],
+        options: FinalRequestOptions,
+        *,
+        stream: bool = False,
+        stream_cls: type[_AsyncStreamT] | None = None,
+    ) -> ResponseT | _AsyncStreamT:
+        # Capture the ORIGINAL, pre-rewrite control-plane options. `request`
+        # rewrites a routed request to the VM internally (via `_prepare_options`),
+        # so we must decide fallback eligibility from these untouched options.
+        original_options = model_copy(options)
+        eligible = _disable_browser_routing.get() is False and should_fallback_to_control_plane(
+            original_options, cache=self.browser_route_cache, config=self._browser_routing
+        )
+
+        try:
+            return await super().request(cast_to, options, stream=stream, stream_cls=stream_cls)
+        except APIStatusError as err:
+            # Direct-VM attempt completed but the VM is unreachable/gone. The SDK's
+            # own retry logic has already run and given up by the time we get here.
+            if eligible and is_vm_unreachable_response(err.response):
+                return await self._control_plane_fallback(
+                    cast_to, original_options, stream=stream, stream_cls=stream_cls
+                )
+            raise
+        except APIConnectionError:
+            # Connection/network/transport error (incl. timeouts) talking to the VM.
+            if eligible:
+                return await self._control_plane_fallback(
+                    cast_to, original_options, stream=stream, stream_cls=stream_cls
+                )
+            raise
+
+    async def _control_plane_fallback(
+        self,
+        cast_to: Type[ResponseT],
+        original_options: FinalRequestOptions,
+        *,
+        stream: bool,
+        stream_cls: type[_AsyncStreamT] | None,
+    ) -> ResponseT | _AsyncStreamT:
+        # Re-issue the ORIGINAL request to the control plane exactly once: original
+        # URL, Authorization restored (default auth headers are re-applied because
+        # routing is disabled, so `strip_direct_vm_auth` is a no-op), no jwt param.
+        # The flag prevents re-routing to the VM during this single attempt.
+        token = _disable_browser_routing.set(True)
+        try:
+            return await super().request(
+                cast_to, model_copy(original_options), stream=stream, stream_cls=stream_cls
+            )
+        finally:
+            _disable_browser_routing.reset(token)
+
     @override
     async def _process_response(
         self,

src/kernel/lib/browser_routing/routing.py

@@ -216,6 +216,73 @@ def rewrite_direct_vm_options(
     return rewritten
 
 
+# HTTP methods that are safe to transparently re-issue. Only idempotent reads may
+# fall back to the control plane; re-running a POST/PUT/PATCH/DELETE could
+# double-execute a side effect (e.g. a routed `curl`), so those never fall back.
+_FALLBACK_IDEMPOTENT_METHODS = frozenset({"GET", "HEAD"})
+
+# The clean "browser is gone" signal emitted by the VM proxy: a 404 carrying this
+# header value plus a JSON body with this code. 404 mirrors the control plane's
+# not_found semantics, so falling back to the control plane is safe.
+_GONE_UPSTREAM_HEADER = "x-kernel-upstream"
+_GONE_UPSTREAM_VALUE = "gone"
+_GONE_BODY_CODE = "browser_gone"
+
+# Transport-level / "VM unreachable" status codes. A live VM never returns these
+# for a real request, so they (along with the gone signal and connection errors)
+# indicate the session's VM is unreachable or gone.
+_UNREACHABLE_STATUS_CODES = frozenset({502, 503, 504})
+
+
+def should_fallback_to_control_plane(options: FinalRequestOptions, *, cache: BrowserRouteCache, config: BrowserRoutingConfig) -> bool:
+    """Return True if ``options`` (the ORIGINAL, pre-rewrite control-plane options)
+    describes a request that was actually routed to a browser VM AND is safe to
+    transparently re-issue to the control plane on failure.
+
+    Eligibility requires all of: an allowlisted browser subresource, a cached
+    route for the session, and an idempotent HTTP method (GET/HEAD).
+    """
+    if options.method.upper() not in _FALLBACK_IDEMPOTENT_METHODS:
+        return False
+
+    match = match_direct_vm_path(options.url)
+    if match is None:
+        return False
+
+    session_id, subresource, _suffix = match
+    if subresource not in set(config.subresources):
+        return False
+
+    return cache.get(session_id) is not None
+
+
+def is_vm_unreachable_response(response: httpx.Response) -> bool:
+    """Return True if ``response`` from the direct-VM attempt indicates the VM is
+    unreachable or the session is gone.
+
+    Triggers on 502/503/504 (today's dead-VM signal) or the clean gone marker
+    (404 + ``X-Kernel-Upstream: gone`` + JSON body code ``browser_gone``). An
+    ordinary 4xx from a live VM (400/401/403, or a non-marker 404) does NOT
+    trigger.
+    """
+    status = response.status_code
+    if status in _UNREACHABLE_STATUS_CODES:
+        return True
+
+    # Clean "gone" signal: 404 + header marker + JSON body code. Require the
+    # header so we never confuse a live VM's ordinary 404 with a gone session.
+    if status == 404 and response.headers.get(_GONE_UPSTREAM_HEADER, "").strip().lower() == _GONE_UPSTREAM_VALUE:
+        try:
+            body = response.json()
+        except Exception:
+            return False
+        if not isinstance(body, Mapping):
+            return False
+        return cast(Mapping[str, object], body).get("code") == _GONE_BODY_CODE
+
+    return False
+
+
 def strip_direct_vm_auth(request: httpx.Request, *, cache: BrowserRouteCache) -> None:
     raw = str(request.url)
     for route in cache.values():