fix(preview): count the app's own 403/404 as routed

.changeset/preview-wait-app-404.md

@@ -0,0 +1,5 @@
+---
+"neevai": patch
+---
+
+`get_url` / `get_port_url` no longer wait out the whole timeout when the app itself answers 403 or 404 at its root, as an API-only or sign-in-gated server does. Those answers now count as routed; only a 403/404 from before the URL is routed keeps the wait going.

src/neevai/resources/sandboxes.py

@@ -34,15 +34,25 @@
 DEFAULT_PORT_WAIT_TIMEOUT_MS = 60_000
 DEFAULT_PORT_POLL_INTERVAL_MS = 2_000
 
+# Set by the edge proxy on a response that came from the sandbox, never on one it answered itself.
+_UPSTREAM_HEADER = "x-envoy-upstream-service-time"
+
+
+def _routed(resp: httpx.Response) -> bool:
+    """True for any status but the edge's own 403/404; an app's own 403/404 carries the mark."""
+    if _UPSTREAM_HEADER in resp.headers:
+        return True
+    return resp.status_code not in (403, 404)
+
 
 def _preview_url_reachable(url: str, timeout_ms: float) -> bool:
-    """Probes a preview URL; True once it is reachable (stops returning a 403/404)."""
+    """Probes a preview URL; True once it is reachable (see :func:`_routed`)."""
     try:
         resp = httpx.get(url, follow_redirects=False, timeout=timeout_ms / 1000.0)
     except (httpx.HTTPError, httpx.InvalidURL):
         # Connection error, DNS failure, timeout, or a malformed URL — not routable yet.
         return False
-    return resp.status_code not in (403, 404)
+    return _routed(resp)
 
 
 async def _apreview_url_reachable(url: str, timeout_ms: float) -> bool:
@@ -52,7 +62,7 @@ async def _apreview_url_reachable(url: str, timeout_ms: float) -> bool:
             resp = await client.get(url, follow_redirects=False, timeout=timeout_ms / 1000.0)
     except (httpx.HTTPError, httpx.InvalidURL):
         return False
-    return resp.status_code not in (403, 404)
+    return _routed(resp)
 
 
 def _validate_wait_args(timeout_ms: int, poll_interval_ms: int) -> None:

tests/test_ports.py

@@ -1,9 +1,13 @@
 """Tests for sandbox preview ports: expose / list / revoke / get_url."""
 
+import httpx
 import pytest
 
 from neevai.client import AsyncNeevAI, NeevAI
 from neevai.errors import NeevAIError
+from neevai.resources.sandboxes import _apreview_url_reachable, _preview_url_reachable
+
+APP = {"x-envoy-upstream-service-time": "3"}  # the edge's mark on a response the app produced
 
 
 def _client(mock_transport) -> NeevAI:
@@ -87,6 +91,31 @@ def test_get_url_raises_when_not_routable_before_timeout(mock_transport, monkeyp
     client.close()
 
 
+@pytest.mark.parametrize(
+    ("status", "headers", "routed"),
+    [
+        (404, {}, False),  # the edge, before the URL is routed
+        (403, {}, False),
+        (404, APP, True),  # an API-only app with nothing at its root
+        (403, APP, True),  # a sign-in-gated root
+        (200, {}, True),
+        (503, {}, True),  # the edge, routed but nothing listening yet
+    ],
+)
+def test_preview_url_reachable_tells_the_app_from_the_edge(monkeypatch, status, headers, routed):
+    monkeypatch.setattr(httpx, "get", lambda *a, **k: httpx.Response(status, headers=headers))
+    assert _preview_url_reachable("https://p.example", 1000) is routed
+
+
+@pytest.mark.asyncio
+async def test_async_preview_url_reachable_counts_the_apps_own_404(monkeypatch):
+    async def get(self, *a, **k):
+        return httpx.Response(404, headers=APP)
+
+    monkeypatch.setattr(httpx.AsyncClient, "get", get)
+    assert await _apreview_url_reachable("https://p.example", 1000) is True
+
+
 def test_get_url_rejects_non_positive_timeout(mock_transport):
     client = _client(mock_transport)
     sb = _new_sandbox(client)