feat: default telemetry subresource to direct-VM routing + extra fallback tests

Address adversarial review notes:
- Land the spec's stated default allowlist ["curl", "telemetry"] so the
  telemetry stream (the motivating GET /telemetry consumer) is routed to the
  VM and eligible for control-plane fallback out of the box.
- Add coverage for 503/504 (parametrized), async gone-404 fallback, and async
  POST non-fallback parity.

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

Author: rgarcia

src/kernel/lib/browser_routing/routing.py

@@ -41,7 +41,7 @@ class BrowserRoutingConfig:
 def browser_routing_config_from_env() -> BrowserRoutingConfig:
     raw = os.environ.get("KERNEL_BROWSER_ROUTING_SUBRESOURCES")
     if raw is None:
-        return BrowserRoutingConfig(subresources=("curl",))
+        return BrowserRoutingConfig(subresources=("curl", "telemetry"))
     if raw.strip() == "":
         return BrowserRoutingConfig()
 

tests/test_browser_routing.py

@@ -313,9 +313,9 @@ def test_browser_route_from_browser_requires_base_url_and_jwt() -> None:
     assert browser_route_from_browser({**_fake_browser(), "cdp_ws_url": None}) is None
 
 
-def test_browser_routing_config_from_env_defaults_to_curl(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_browser_routing_config_from_env_defaults_to_curl_and_telemetry(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.delenv("KERNEL_BROWSER_ROUTING_SUBRESOURCES", raising=False)
-    assert browser_routing_config_from_env().subresources == ("curl",)
+    assert browser_routing_config_from_env().subresources == ("curl", "telemetry")
 
 
 def test_browser_routing_config_from_env_empty_string_disables_routing(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -495,3 +495,63 @@ async def test_async_dead_vm_502_falls_back_to_control_plane(monkeypatch: pytest
     cp_request = cast(httpx.Request, cast(Any, cp_route.calls[0]).request)
     assert cp_request.headers.get("Authorization") == f"Bearer {api_key}"
     assert cp_request.url.params.get("jwt") is None
+
+
+@pytest.mark.parametrize("status", [503, 504])
+@respx.mock
+def test_dead_vm_503_504_fall_back_to_control_plane(monkeypatch: pytest.MonkeyPatch, status: int) -> None:
+    # 503/504 share the unreachable-status path with 502; lock in both explicitly.
+    monkeypatch.setenv("KERNEL_BROWSER_ROUTING_SUBRESOURCES", "process")
+    vm_route = respx.get(_VM_STATUS).mock(return_value=httpx.Response(status, text="unavailable"))
+    cp_route = respx.get(_CP_STATUS).mock(return_value=httpx.Response(200, json={"state": "exited"}))
+    with Kernel(base_url=base_url, api_key=api_key, max_retries=0, _strict_response_validation=True) as client:
+        _cache_browser(client)
+        out = client.browsers.process.status("p-1", id="sess-1")
+
+    assert vm_route.called
+    assert cp_route.called
+    assert out.state == "exited"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_async_gone_signal_404_falls_back_to_control_plane(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("KERNEL_BROWSER_ROUTING_SUBRESOURCES", "process")
+    vm_route = respx.get(_VM_STATUS).mock(return_value=httpx.Response(404, json=_GONE_BODY, headers=_GONE_HEADERS))
+    cp_route = respx.get(_CP_STATUS).mock(return_value=httpx.Response(200, json={"state": "exited"}))
+    async with AsyncKernel(base_url=base_url, api_key=api_key, max_retries=0, _strict_response_validation=True) as client:
+        route = browser_route_from_browser(_fake_browser())
+        assert route is not None
+        client.browser_route_cache.set(route)
+        out = await client.browsers.process.status("p-1", id="sess-1")
+
+    assert vm_route.called
+    assert cp_route.called
+    assert out.state == "exited"
+    cp_request = cast(httpx.Request, cast(Any, cp_route.calls[0]).request)
+    assert cp_request.headers.get("Authorization") == f"Bearer {api_key}"
+    assert cp_request.url.params.get("jwt") is None
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_async_post_to_dead_vm_does_not_fall_back(monkeypatch: pytest.MonkeyPatch) -> None:
+    from kernel import APIStatusError
+
+    monkeypatch.setenv("KERNEL_BROWSER_ROUTING_SUBRESOURCES", "process")
+    vm_route = respx.post("http://browser-session.test/browser/kernel/process/exec").mock(
+        return_value=httpx.Response(502, text="Upstream not available")
+    )
+    cp_route = respx.post(f"{base_url}/browsers/sess-1/process/exec").mock(
+        return_value=httpx.Response(200, json={"exit_code": 0, "stdout_b64": "", "stderr_b64": ""})
+    )
+    async with AsyncKernel(base_url=base_url, api_key=api_key, max_retries=0, _strict_response_validation=True) as client:
+        route = browser_route_from_browser(_fake_browser())
+        assert route is not None
+        client.browser_route_cache.set(route)
+        with pytest.raises(APIStatusError):
+            await client.browsers.process.exec("sess-1", command="echo", args=["hi"])
+
+    assert vm_route.called
+    # POST is side-effecting; must never be re-issued to the control plane (async parity).
+    assert not cp_route.called