fix(runtime): retain HTTP error status and fail promptly on permanent transport failures

- Preserve HTTP status codes alongside error details when Caido returns an HTTPError in caido_api._login_as_guest.

- Distinguish permanent transport and execution failures (ExecTransportError, WorkspaceStopError, exit 126/127) from transient Caido readiness probes in caido_bootstrap._login_as_guest.

- Add unit tests verifying prompt failure on permanent errors and HTTP status retention.
This commit is contained in:
Noor Fatima 2026-09-30 01:28:17 +05:00
parent 367d137f91
commit 1cfa32b114
3 changed files with 152 additions and 3 deletions

View file

@ -28,6 +28,42 @@ _LOGIN_AS_GUEST_BODY = (
'{"query":"mutation LoginAsGuest { loginAsGuest { token { accessToken } } }"}'
)
_PERMANENT_EXIT_CODES = frozenset({
2, # curl: failed to initialize / bad command-line syntax
3, # curl: malformed URL
126, # command invoked cannot execute (permission denied)
127, # command not found (curl is not installed in the sandbox image)
})
def _is_permanent_transport_error(exc: BaseException) -> bool:
"""Return True if the sandbox transport has failed permanently and cannot recover."""
if isinstance(exc, asyncio.CancelledError):
return True
if getattr(exc, "retryable", None) is False:
return True
try:
from agents.sandbox.errors import ExecTransportError, WorkspaceStopError
if isinstance(exc, WorkspaceStopError):
return True
if isinstance(exc, ExecTransportError) and not exc.context.get("retry_safe", False):
return True
except ImportError:
pass
try:
from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore]
if isinstance(exc, (docker_errors.NotFound, docker_errors.APIError)):
return True
except ImportError:
pass
return False
def _extract_access_token(payload: Any) -> str | None:
if not isinstance(payload, dict):
@ -82,7 +118,11 @@ async def _login_as_guest(
f"{container_url}/graphql",
timeout=15,
)
except Exception as exc: # noqa: BLE001
except Exception as exc:
if _is_permanent_transport_error(exc):
raise RuntimeError(
f"loginAsGuest failed permanently: session.exec failed: {exc}"
) from exc
last_err = f"session.exec failed: {exc}"
logger.debug("loginAsGuest attempt %d/%d failed: %s", i, attempts, last_err)
await asyncio.sleep(min(2.0 * i, 8.0))
@ -99,6 +139,10 @@ async def _login_as_guest(
last_err = f"unparseable response: {exc}: {result.stdout!r}"
else:
stderr = _format_stderr(result.stderr)
if result.exit_code in _PERMANENT_EXIT_CODES:
raise RuntimeError(
f"loginAsGuest failed permanently: curl exit {result.exit_code}: {stderr}"
)
last_err = f"curl exit {result.exit_code}: {stderr}"
logger.debug("loginAsGuest attempt %d/%d failed: %s", i, attempts, last_err)
await asyncio.sleep(min(2.0 * i, 8.0))

View file

@ -80,7 +80,12 @@ def _login_as_guest() -> str:
with urllib.request.urlopen(req, timeout=10) as resp: # noqa: S310 # nosec B310
raw_resp = resp.read()
except urllib.error.HTTPError as err:
raw_resp = err.read()
body = err.read()
try:
payload = json.loads(body)
except (json.JSONDecodeError, TypeError):
payload = body.decode("utf-8", errors="replace")[:200].strip() or err.reason
raise RuntimeError(f"Caido loginAsGuest returned HTTP {err.code}: {payload}") from err
except urllib.error.URLError as err:
raise RuntimeError(f"Failed to connect to Caido at {_graphql_url()}: {err}") from err

View file

@ -13,10 +13,12 @@ import sys
import types
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock
import pytest
from agents.sandbox.errors import ExecTransportError, WorkspaceStopError
from strix.runtime.caido_bootstrap import _login_as_guest, bootstrap_caido
from strix.tools.proxy import caido_api
@ -313,7 +315,24 @@ def test_caido_api_login_as_guest_http_500_with_graphql_error(
monkeypatch.setattr(urllib.request, "urlopen", MagicMock(side_effect=http_err))
with pytest.raises(RuntimeError, match="database locked"):
with pytest.raises(RuntimeError, match=r"HTTP 500.*database locked"):
caido_api._login_as_guest()
def test_caido_api_login_as_guest_http_401_non_json_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
http_err = urllib.error.HTTPError(
url="http://127.0.0.1:8080/graphql",
code=401,
msg="Unauthorized",
hdrs={}, # type: ignore[arg-type]
fp=io.BytesIO(b"Unauthorized access"),
)
monkeypatch.setattr(urllib.request, "urlopen", MagicMock(side_effect=http_err))
with pytest.raises(RuntimeError, match=r"HTTP 401.*Unauthorized access"):
caido_api._login_as_guest()
@ -326,3 +345,84 @@ def test_caido_api_login_as_guest_url_error(
with pytest.raises(RuntimeError, match="Failed to connect to Caido"):
caido_api._login_as_guest()
async def test_login_as_guest_fails_promptly_on_permanent_transport_error() -> None:
class _PermanentTransportSession:
def __init__(self) -> None:
self.call_count = 0
async def exec(self, *_args: Any, **_kwargs: Any) -> _FakeExecResult:
self.call_count += 1
raise ExecTransportError(command=["curl"], message="Container gone", retryable=False)
session = _PermanentTransportSession()
with pytest.raises(
RuntimeError, match=r"loginAsGuest failed permanently.*session.exec failed"
):
await _login_as_guest(
session, # type: ignore[arg-type]
container_url="http://container",
attempts=10,
)
assert session.call_count == 1
async def test_login_as_guest_fails_promptly_on_workspace_stop_error() -> None:
class _StoppedSession:
def __init__(self) -> None:
self.call_count = 0
async def exec(self, *_args: Any, **_kwargs: Any) -> _FakeExecResult:
self.call_count += 1
raise WorkspaceStopError(path=Path("/workspace"))
session = _StoppedSession()
with pytest.raises(
RuntimeError, match=r"loginAsGuest failed permanently.*session.exec failed"
):
await _login_as_guest(
session, # type: ignore[arg-type]
container_url="http://container",
attempts=10,
)
assert session.call_count == 1
async def test_login_as_guest_fails_promptly_when_curl_not_found() -> None:
class _CurlNotFoundSession:
def __init__(self) -> None:
self.call_count = 0
async def exec(self, *_args: Any, **_kwargs: Any) -> _FakeExecResult:
self.call_count += 1
return _FakeExecResult("", stderr=b"/bin/sh: curl: not found", exit_code=127)
session = _CurlNotFoundSession()
with pytest.raises(RuntimeError, match="loginAsGuest failed permanently: curl exit 127"):
await _login_as_guest(
session, # type: ignore[arg-type]
container_url="http://container",
attempts=10,
)
assert session.call_count == 1
async def test_login_as_guest_fails_promptly_when_curl_permission_denied() -> None:
class _CurlPermissionDeniedSession:
def __init__(self) -> None:
self.call_count = 0
async def exec(self, *_args: Any, **_kwargs: Any) -> _FakeExecResult:
self.call_count += 1
return _FakeExecResult("", stderr=b"/bin/sh: curl: Permission denied", exit_code=126)
session = _CurlPermissionDeniedSession()
with pytest.raises(RuntimeError, match="loginAsGuest failed permanently: curl exit 126"):
await _login_as_guest(
session, # type: ignore[arg-type]
container_url="http://container",
attempts=10,
)
assert session.call_count == 1