From e0d78c2f9399c5961751980791886af871008a68 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 5 Nov 2026 12:47:36 -0800 Subject: [PATCH 1/2] feat(providers): a live WebSocket check decides each provider's declaration; boat and CoreWeave pass A shared live test starts a stdlib WebSocket server in a sandbox of each remote provider and, through its tunnel, checks a CDP answer, a 10 MB frame and a connection pinged every 20 s after 5 idle minutes, failing when the result doesn't match the provider's WEBSOCKETS declaration. boat and CoreWeave pass it and now declare WEBSOCKETS. Co-Authored-By: Claude Opus 5.5 --- .../providers/sandbox_providers/AGENTS.md | 2 +- .../sandbox_providers/boat/provider.py | 1 + .../sandbox_providers/coreweave/provider.py | 1 + .../sandbox_providers/websockets_live_test.py | 237 ++++++++++++++++++ .../provider_capabilities_test.py | 4 +- 5 files changed, 242 insertions(+), 3 deletions(-) create mode 100644 tst/integration/providers/sandbox_providers/websockets_live_test.py diff --git a/src/agent_env/providers/sandbox_providers/AGENTS.md b/src/agent_env/providers/sandbox_providers/AGENTS.md index e2de947c4..cd7d788b1 100644 --- a/src/agent_env/providers/sandbox_providers/AGENTS.md +++ b/src/agent_env/providers/sandbox_providers/AGENTS.md @@ -86,7 +86,7 @@ sandboxes off this machine, runs containers' images by name, has no private netw | `CONTAINER_ACCEPTS` | `create_container` runs images other than by name: `Accepts.LOADABLE` (tar.gz, registry name or build context), `Accepts.NAME_OR_CONTEXT` | | `SANDBOX_ACCEPTS` | an agent's sandbox runs different images than the default (a provider that implements `create_vm` defaults to `LOADABLE`) | | `PRIVATE_NETWORK` | `create_container(private_network=True)` puts containers on a network they share | -| `WEBSOCKETS` | its sandboxes' exposed ports carry WebSockets both ways, as our own tests or its tunnel's documentation show; a computer with a website browser among its apps, whose CDP address agents reach by WebSocket, refuses a provider without it at preflight, and a chain skips it at deploy | +| `WEBSOCKETS` | its sandboxes' exposed ports carry WebSockets both ways, as `tst/integration/providers/sandbox_providers/websockets_live_test.py` or its tunnel's documentation show; a computer with a website browser among its apps, whose CDP address agents reach by WebSocket, refuses a provider without it at preflight, and a chain skips it at deploy | ## Sandbox handle contract diff --git a/src/agent_env/providers/sandbox_providers/boat/provider.py b/src/agent_env/providers/sandbox_providers/boat/provider.py index a15d316ce..c02c75b81 100644 --- a/src/agent_env/providers/sandbox_providers/boat/provider.py +++ b/src/agent_env/providers/sandbox_providers/boat/provider.py @@ -62,6 +62,7 @@ class BoatSandboxProvider(SandboxProvider): # boat serves hosted ports at https://-.on.boat.dev. EGRESS_HOSTS: ClassVar[tuple[str, ...]] = ("*.on.boat.dev",) + WEBSOCKETS = True _CREATE_ATTEMPTS = 6 _RETRY_BASE_DELAY = 2.0 _RETRY_MAX_DELAY = 60.0 diff --git a/src/agent_env/providers/sandbox_providers/coreweave/provider.py b/src/agent_env/providers/sandbox_providers/coreweave/provider.py index df13857a1..f4ff8d86c 100644 --- a/src/agent_env/providers/sandbox_providers/coreweave/provider.py +++ b/src/agent_env/providers/sandbox_providers/coreweave/provider.py @@ -196,6 +196,7 @@ class CoreWeaveSandboxProvider(SandboxProvider): # The platform pulls a container's image by its name; only a VM loads a tarball or builds a context. CONTAINER_ACCEPTS = Accepts.NAME SANDBOX_ACCEPTS = Accepts.NAME + WEBSOCKETS = True # An exposed port's URL is -.., and a CoreWeave wildcard covers exactly one # label, so no one entry covers every runner. Each parent seen on an exposed port is added here, and every # provider's allowlists include it, so a restricted agent on any provider reaches environments served there. diff --git a/tst/integration/providers/sandbox_providers/websockets_live_test.py b/tst/integration/providers/sandbox_providers/websockets_live_test.py new file mode 100644 index 000000000..8b1d4c292 --- /dev/null +++ b/tst/integration/providers/sandbox_providers/websockets_live_test.py @@ -0,0 +1,237 @@ +"""Whether each remote sandbox provider's ``WEBSOCKETS`` declaration is right: whether its tunnel carries what a +website browser's CDP address needs. Through the tunnel, a CDP command is answered over a WebSocket, a 10 MB frame comes +back intact, and after five idle minutes a connection pinged every 20 s, as agentenv's driver and gateway ping, still +answers. A connection beside it that is never pinged records the tunnel's own idle limit, and doesn't decide. The server +behind the tunnel is a stdlib one this test starts in a sandbox of the provider, a VM where it makes them, so only +the tunnel is under test. + +A case runs when the resolved config builds its provider (the ``remote_sandbox`` capability) and skips otherwise. Each +prints its findings as a ``websockets-check`` JSON line. No test reads or prints a key.""" + +from __future__ import annotations + +import asyncio +import base64 +import json +import os +import time + +import httpx +import pytest +import websockets +from websockets.asyncio.client import connect + +from agent_env.providers.sandbox_providers.sandbox import VmSandbox +from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider +from tst.util.capabilities import skip_without_remote_sandbox + +pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow, pytest.mark.asyncio] + +#: The remote providers, each of which declares whether it carries WebSockets. +REMOTE = ("e2b", "tensorlake", "vercel", "coreweave", "daytona", "boat", "modal", "modal_vm", "boxd", "sail_vm", "tenki", + "agent37") +PORT = 8765 +FRAME_BYTES = 10 * 1024 * 1024 +IDLE_S = 300 +PING_S = 20.0 +GET_VERSION = {"id": 1, "method": "Browser.getVersion"} + +#: A WebSocket server in the standard library alone: it answers ``Browser.getVersion`` as a browser would, echoes every +#: other message, answers pings, and answers a plain GET with 200 for a tunnel's health check. +SERVER = r''' +import base64, hashlib, json, socketserver, struct, sys + +GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + + +def read_message(stream): + """A whole message, its fragments joined: (opcode, payload), or (None, b"") once the connection ends.""" + opcode, parts = None, [] + while True: + head = stream.read(2) + if len(head) < 2: + return None, b"" + fin, code, masked, length = head[0] & 0x80, head[0] & 0x0F, head[1] & 0x80, head[1] & 0x7F + if length == 126: + length = struct.unpack(">H", stream.read(2))[0] + elif length == 127: + length = struct.unpack(">Q", stream.read(8))[0] + mask = stream.read(4) if masked else b"" + payload = stream.read(length) + if mask: + payload = bytes(byte ^ mask[i % 4] for i, byte in enumerate(payload)) + if code >= 8: # a control frame, never fragmented, may come between fragments + return code, payload + opcode = code if code else opcode + parts.append(payload) + if fin: + return opcode, b"".join(parts) + + +def send(conn, opcode, payload): + length = len(payload) + if length < 126: + head = struct.pack(">BB", 0x80 | opcode, length) + elif length < 1 << 16: + head = struct.pack(">BBH", 0x80 | opcode, 126, length) + else: + head = struct.pack(">BBQ", 0x80 | opcode, 127, length) + conn.sendall(head + payload) + + +class Handler(socketserver.BaseRequestHandler): + def handle(self): + conn, data = self.request, b"" + while b"\r\n\r\n" not in data: + chunk = conn.recv(4096) + if not chunk: + return + data += chunk + lines = data.split(b"\r\n\r\n", 1)[0].decode("latin-1").split("\r\n") + headers = {name.strip().lower(): value.strip() for name, _, value in (line.partition(":") for line in lines[1:])} + key = headers.get("sec-websocket-key") + if not key: + conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") + return + accept = base64.b64encode(hashlib.sha1((key + GUID).encode()).digest()).decode() + conn.sendall(("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + "Sec-WebSocket-Accept: " + accept + "\r\n\r\n").encode()) + stream = conn.makefile("rb") + while True: + opcode, payload = read_message(stream) + if opcode is None or opcode == 8: + if opcode == 8: + send(conn, 8, payload[:2]) + return + if opcode == 9: + send(conn, 10, payload) + elif opcode == 1 and json.loads(payload).get("method") == "Browser.getVersion": + answer = {"id": json.loads(payload)["id"], "result": {"product": "agentenv-websockets-check/1"}} + send(conn, 1, json.dumps(answer).encode()) + elif opcode in (1, 2): + send(conn, opcode, payload) + + +class Server(socketserver.ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = True + + +Server(("0.0.0.0", int(sys.argv[1])), Handler).serve_forever() +''' + +PYTHON_IMAGE = "public.ecr.aws/docker/library/python:3.12-alpine" +DIR = "/tmp/websockets-check" + + +def _start_script() -> str: + """Writes the server and starts it detached, in Docker where the sandbox has it, else on its own Python.""" + encoded = base64.b64encode(SERVER.encode()).decode() + return f"""set -e +mkdir -p {DIR} +echo {encoded} | base64 -d > {DIR}/server.py +if command -v docker >/dev/null 2>&1; then + docker run -d --name websockets-check -p {PORT}:{PORT} -v {DIR}:/check {PYTHON_IMAGE} python /check/server.py {PORT} >/dev/null +else + setsid nohup python3 {DIR}/server.py {PORT} > {DIR}/server.log 2>&1 < /dev/null & +fi +""" + + +@pytest.mark.parametrize("name", [pytest.param(name, marks=skip_without_remote_sandbox(name)) for name in REMOTE]) +async def test_a_providers_websockets_declaration_matches_what_its_tunnel_carries(name): + provider = build_sandbox_provider(name) + found: dict = {"provider": name, "declared": provider.WEBSOCKETS} + try: + sandbox = await _sandbox(provider) + try: + await _serve(sandbox) + url = sandbox.tunnel_urls[PORT] + await _until_served(url) + found.update(await _checked("ws" + url.removeprefix("http"))) + finally: + await sandbox.terminate() + finally: + close = getattr(provider, "close", None) + if close is not None: + await close() + found["carries"] = all(found.get(check) is True for check in ("cdp", "frame", "idle_pinged")) + print("websockets-check", json.dumps(found)) + + assert found["carries"] == found["declared"], ( + f"{name} declares WEBSOCKETS = {found['declared']}, but its tunnel " + f"{'carries' if found['carries'] else 'does not carry'} what a website browser's CDP address needs: {found}") + + +async def _sandbox(provider): + """A VM, or on a provider that makes only containers, a container of a stock Python image.""" + try: + return await provider.create_vm(cpu=1, memory=2048, exposed_ports=[PORT], timeout=1800) + except NotImplementedError: + return await provider.create_container(image_name=PYTHON_IMAGE, port=PORT, env={}, cpu=1, memory=2048, + timeout=1800) + + +async def _serve(sandbox) -> None: + if isinstance(sandbox, VmSandbox): + await sandbox.exec_script(_start_script()) + return + exit_code, stdout, stderr = await sandbox.exec_with_output("sh", "-c", _start_script()) + assert exit_code == 0, f"starting the server failed (exit {exit_code}): {stdout[-500:]} {stderr[-500:]}" + + +async def _until_served(url: str, timeout_s: float = 300) -> None: + deadline = time.monotonic() + timeout_s + async with httpx.AsyncClient(timeout=10) as client: + while time.monotonic() < deadline: + try: + if (await client.get(url)).status_code == 200: + return + except httpx.HTTPError: + pass + await asyncio.sleep(3) + raise TimeoutError(f"nothing answered at {url} within {timeout_s:.0f} s") + + +async def _checked(ws_url: str) -> dict: + """What the tunnel at ``ws_url`` carries: a CDP answer, a 10 MB frame back intact, and, after ``IDLE_S`` idle, an + answer on a pinged connection, and how long one never pinged lasted.""" + found: dict = {} + try: + async with connect(ws_url, max_size=None, open_timeout=60, ping_interval=None) as conn: + found["cdp"] = await _version(conn) == "agentenv-websockets-check/1" + frame = os.urandom(FRAME_BYTES) + await conn.send(frame) + found["frame"] = await asyncio.wait_for(conn.recv(), 120) == frame + except (OSError, asyncio.TimeoutError, websockets.WebSocketException) as e: + found.setdefault("cdp", False) + found.setdefault("frame", False) + found["error"] = f"{type(e).__name__}: {e}"[:300] + return found + try: + pinged = await connect(ws_url, max_size=None, open_timeout=60, ping_interval=PING_S, ping_timeout=PING_S) + silent = await connect(ws_url, max_size=None, open_timeout=60, ping_interval=None) + except (OSError, asyncio.TimeoutError, websockets.WebSocketException) as e: + found["idle_pinged"] = False + found["idle_pinged_error"] = f"{type(e).__name__}: {e}"[:300] + return found + started = time.monotonic() + silent_closed = asyncio.ensure_future(silent.wait_closed()) + try: + await asyncio.wait([silent_closed], timeout=IDLE_S) + found["idle_unpinged_s"] = round(time.monotonic() - started) if silent_closed.done() else None + try: + found["idle_pinged"] = await _version(pinged) == "agentenv-websockets-check/1" + except (asyncio.TimeoutError, websockets.WebSocketException) as e: + found["idle_pinged"] = False + found["idle_pinged_error"] = f"{type(e).__name__}: {e}"[:300] + finally: + silent_closed.cancel() + await pinged.close() + await silent.close() + return found + + +async def _version(conn) -> str: + await conn.send(json.dumps(GET_VERSION)) + return json.loads(await asyncio.wait_for(conn.recv(), 60))["result"]["product"] diff --git a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py index 4f21e8f88..cf738e2fc 100644 --- a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py +++ b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py @@ -61,9 +61,9 @@ (TensorlakeSandboxProvider, False, True, LOADABLE, NAME, False, False), (TenkiSandboxProvider, False, True, LOADABLE, NAME, False, True), (BoxdSandboxProvider, False, True, LOADABLE, NAME, False, True), - (CoreWeaveSandboxProvider, False, True, NAME, NAME, False, False), + (CoreWeaveSandboxProvider, False, True, NAME, NAME, False, True), (DaytonaSandboxProvider, False, True, LOADABLE, NAME, False, False), - (BoatSandboxProvider, False, True, LOADABLE, NAME, False, False), + (BoatSandboxProvider, False, True, LOADABLE, NAME, False, True), (Agent37SandboxProvider, False, True, LOADABLE, NAME, False, True), ], ids=["local", "modal", "modal_vm", "e2b", "sail_vm", "tensorlake", "tenki", "boxd", "coreweave", "daytona", "boat", "agent37"]) From 59fb6ecaa8abc37c88658f42dcfc2f41be34c449 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Thu, 5 Nov 2026 13:01:01 -0800 Subject: [PATCH 2/2] fix(providers): the pinged connection waits the full idle window, the unpinged one never decides An unpinged connection that closed early ended the idle wait, so the pinged one was checked before five minutes, and one that failed to open failed the provider. Each is now opened and checked on its own, the unpinged one's close time recorded as it happens, and the test's settings are named. Co-Authored-By: Claude Opus 5.5 --- .../sandbox_providers/websockets_live_test.py | 77 +++++++++++++------ 1 file changed, 53 insertions(+), 24 deletions(-) diff --git a/tst/integration/providers/sandbox_providers/websockets_live_test.py b/tst/integration/providers/sandbox_providers/websockets_live_test.py index 8b1d4c292..677301302 100644 --- a/tst/integration/providers/sandbox_providers/websockets_live_test.py +++ b/tst/integration/providers/sandbox_providers/websockets_live_test.py @@ -35,6 +35,18 @@ IDLE_S = 300 PING_S = 20.0 GET_VERSION = {"id": 1, "method": "Browser.getVersion"} +#: What the server answers ``Browser.getVersion`` with. +PRODUCT = "agentenv-websockets-check/1" +SANDBOX_CPU = 1 +SANDBOX_MEMORY_MB = 2048 +SANDBOX_LIFETIME_S = 1800 +SERVED_WITHIN_S = 300 +SERVED_POLL_S = 3 +HTTP_TIMEOUT_S = 10 +OPEN_TIMEOUT_S = 60 +ANSWER_TIMEOUT_S = 60 +FRAME_TIMEOUT_S = 120 +ERROR_CHARS = 300 #: A WebSocket server in the standard library alone: it answers ``Browser.getVersion`` as a browser would, echoes every #: other message, answers pings, and answers a plain GET with 200 for a tunnel's health check. @@ -106,7 +118,7 @@ def handle(self): if opcode == 9: send(conn, 10, payload) elif opcode == 1 and json.loads(payload).get("method") == "Browser.getVersion": - answer = {"id": json.loads(payload)["id"], "result": {"product": "agentenv-websockets-check/1"}} + answer = {"id": json.loads(payload)["id"], "result": {"product": sys.argv[2]}} send(conn, 1, json.dumps(answer).encode()) elif opcode in (1, 2): send(conn, opcode, payload) @@ -131,9 +143,9 @@ def _start_script() -> str: mkdir -p {DIR} echo {encoded} | base64 -d > {DIR}/server.py if command -v docker >/dev/null 2>&1; then - docker run -d --name websockets-check -p {PORT}:{PORT} -v {DIR}:/check {PYTHON_IMAGE} python /check/server.py {PORT} >/dev/null + docker run -d --name websockets-check -p {PORT}:{PORT} -v {DIR}:/check {PYTHON_IMAGE} python /check/server.py {PORT} {PRODUCT} >/dev/null else - setsid nohup python3 {DIR}/server.py {PORT} > {DIR}/server.log 2>&1 < /dev/null & + setsid nohup python3 {DIR}/server.py {PORT} {PRODUCT} > {DIR}/server.log 2>&1 < /dev/null & fi """ @@ -166,10 +178,11 @@ async def test_a_providers_websockets_declaration_matches_what_its_tunnel_carrie async def _sandbox(provider): """A VM, or on a provider that makes only containers, a container of a stock Python image.""" try: - return await provider.create_vm(cpu=1, memory=2048, exposed_ports=[PORT], timeout=1800) + return await provider.create_vm(cpu=SANDBOX_CPU, memory=SANDBOX_MEMORY_MB, exposed_ports=[PORT], + timeout=SANDBOX_LIFETIME_S) except NotImplementedError: - return await provider.create_container(image_name=PYTHON_IMAGE, port=PORT, env={}, cpu=1, memory=2048, - timeout=1800) + return await provider.create_container(image_name=PYTHON_IMAGE, port=PORT, env={}, cpu=SANDBOX_CPU, + memory=SANDBOX_MEMORY_MB, timeout=SANDBOX_LIFETIME_S) async def _serve(sandbox) -> None: @@ -180,16 +193,16 @@ async def _serve(sandbox) -> None: assert exit_code == 0, f"starting the server failed (exit {exit_code}): {stdout[-500:]} {stderr[-500:]}" -async def _until_served(url: str, timeout_s: float = 300) -> None: +async def _until_served(url: str, timeout_s: float = SERVED_WITHIN_S) -> None: deadline = time.monotonic() + timeout_s - async with httpx.AsyncClient(timeout=10) as client: + async with httpx.AsyncClient(timeout=HTTP_TIMEOUT_S) as client: while time.monotonic() < deadline: try: if (await client.get(url)).status_code == 200: return except httpx.HTTPError: pass - await asyncio.sleep(3) + await asyncio.sleep(SERVED_POLL_S) raise TimeoutError(f"nothing answered at {url} within {timeout_s:.0f} s") @@ -198,40 +211,56 @@ async def _checked(ws_url: str) -> dict: answer on a pinged connection, and how long one never pinged lasted.""" found: dict = {} try: - async with connect(ws_url, max_size=None, open_timeout=60, ping_interval=None) as conn: - found["cdp"] = await _version(conn) == "agentenv-websockets-check/1" + async with connect(ws_url, max_size=None, open_timeout=OPEN_TIMEOUT_S, ping_interval=None) as conn: + found["cdp"] = await _version(conn) == PRODUCT frame = os.urandom(FRAME_BYTES) await conn.send(frame) - found["frame"] = await asyncio.wait_for(conn.recv(), 120) == frame + found["frame"] = await asyncio.wait_for(conn.recv(), FRAME_TIMEOUT_S) == frame except (OSError, asyncio.TimeoutError, websockets.WebSocketException) as e: found.setdefault("cdp", False) found.setdefault("frame", False) - found["error"] = f"{type(e).__name__}: {e}"[:300] + found["error"] = _error(e) return found try: - pinged = await connect(ws_url, max_size=None, open_timeout=60, ping_interval=PING_S, ping_timeout=PING_S) - silent = await connect(ws_url, max_size=None, open_timeout=60, ping_interval=None) + pinged = await connect(ws_url, max_size=None, open_timeout=OPEN_TIMEOUT_S, ping_interval=PING_S, + ping_timeout=PING_S) except (OSError, asyncio.TimeoutError, websockets.WebSocketException) as e: found["idle_pinged"] = False - found["idle_pinged_error"] = f"{type(e).__name__}: {e}"[:300] + found["idle_pinged_error"] = _error(e) return found + try: + silent = await connect(ws_url, max_size=None, open_timeout=OPEN_TIMEOUT_S, ping_interval=None) + except (OSError, asyncio.TimeoutError, websockets.WebSocketException) as e: + silent = None + found["idle_unpinged_error"] = _error(e) started = time.monotonic() - silent_closed = asyncio.ensure_future(silent.wait_closed()) + + async def lasted(conn) -> int: + await conn.wait_closed() + return round(time.monotonic() - started) + + silent_lasted = asyncio.ensure_future(lasted(silent)) if silent is not None else None try: - await asyncio.wait([silent_closed], timeout=IDLE_S) - found["idle_unpinged_s"] = round(time.monotonic() - started) if silent_closed.done() else None + await asyncio.sleep(IDLE_S) + if silent_lasted is not None: + found["idle_unpinged_s"] = silent_lasted.result() if silent_lasted.done() else None try: - found["idle_pinged"] = await _version(pinged) == "agentenv-websockets-check/1" + found["idle_pinged"] = await _version(pinged) == PRODUCT except (asyncio.TimeoutError, websockets.WebSocketException) as e: found["idle_pinged"] = False - found["idle_pinged_error"] = f"{type(e).__name__}: {e}"[:300] + found["idle_pinged_error"] = _error(e) finally: - silent_closed.cancel() + if silent_lasted is not None: + silent_lasted.cancel() + await silent.close() await pinged.close() - await silent.close() return found async def _version(conn) -> str: await conn.send(json.dumps(GET_VERSION)) - return json.loads(await asyncio.wait_for(conn.recv(), 60))["result"]["product"] + return json.loads(await asyncio.wait_for(conn.recv(), ANSWER_TIMEOUT_S))["result"]["product"] + + +def _error(e: BaseException) -> str: + return f"{type(e).__name__}: {e}"[:ERROR_CHARS]