diff --git a/README.md b/README.md index 8f043836..01f3f07a 100644 --- a/README.md +++ b/README.md @@ -295,6 +295,49 @@ sandbox.files.write( ) ``` +### Run commands and collect output + +`sandbox.exec()` and `sandbox.processes.start()` stream command output from the +receiver as soon as execution starts. The SDK collects stdout and stderr in +memory, so a completed result is not limited to the receiver's replay buffer. +This requires a receiver supporting streaming `POST /sandbox/processes`; roll +out the receiver before upgrading the SDK. + +```python +result = sandbox.exec("make test", max_output_bytes=128 * 1024 * 1024) +print(result.stdout, result.stderr, result.exit_code) + +process = sandbox.processes.start("make test") +try: + for event in process.stream(): + if event.type == "stdout": + print(event.data, end="") + result = process.wait() +finally: + process.disconnect() +``` + +The combined output limit defaults to 64 MiB per command and can be adjusted +with `max_output_bytes`. Exceeding it raises `output_limit_exceeded`. A broken +stream, missing output, or receiver truncation raises `incomplete_output`. +These errors include the process ID and do not automatically rerun the command. + +Process streams use a separate 60-second read-idle timeout once response headers +arrive. Output and the receiver's 15-second heartbeats reset this timeout, so quiet +commands can run longer than the client's ordinary HTTP timeout. That ordinary +timeout still applies to connection setup and waiting for response headers. + +`start()` returns after the process starts and collects in the background. +`wait(timeout_sec=...)` limits the local wait; collection continues after a wait +timeout. The timeout passed to `start()` or `exec()` limits command execution. +A local wait timeout raises `TimeoutError` (`asyncio.TimeoutError` in the async API). +`disconnect()` stops collection and leaves the command running. Use `kill()` to +stop it. Reattaching with `get()` can retrieve only retained receiver output; +`wait()` raises if that output has been truncated. + +The async API has the same behavior: await `exec()`, `start()`, `wait()`, and +`disconnect()`, and use `async for` with `stream()`. + ### Resume terminal output after reconnect ```python diff --git a/hyperbrowser/client/managers/async_manager/sandbox.py b/hyperbrowser/client/managers/async_manager/sandbox.py index bf86c74d..fa494461 100644 --- a/hyperbrowser/client/managers/async_manager/sandbox.py +++ b/hyperbrowser/client/managers/async_manager/sandbox.py @@ -307,6 +307,7 @@ async def exec( timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None, run_as: Optional[str] = None, + max_output_bytes: int = 64 * 1024 * 1024, ): return await self.processes.exec( input, @@ -315,6 +316,7 @@ async def exec( timeout_ms=timeout_ms, timeout_sec=timeout_sec, run_as=run_as, + max_output_bytes=max_output_bytes, ) async def get_process(self, process_id: str) -> SandboxProcessHandle: diff --git a/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_processes.py b/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_processes.py index 69d36df5..112920a9 100644 --- a/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_processes.py +++ b/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_processes.py @@ -1,3 +1,4 @@ +import asyncio import base64 from typing import AsyncIterator, Dict, Optional, Union @@ -16,6 +17,11 @@ SandboxProcessStdinParams as SandboxProcessStdinParamsDict, ) from ...sandboxes.shared import _normalize_exec_params +from ...sandboxes.process_output import ( + DEFAULT_MAX_PROCESS_OUTPUT_BYTES, + ProcessOutput, + validate_output_limit, +) from .sandbox_transport import RuntimeTransport DEFAULT_PROCESS_KILL_WAIT_SECONDS = 5.0 @@ -25,6 +31,10 @@ class SandboxProcessHandle: def __init__(self, transport: RuntimeTransport, summary: SandboxProcessSummary): self._transport = transport self._summary = summary + self._output = None + self._collector = None + self._events = None + self._changed = asyncio.Event() @property def id(self) -> str: @@ -50,6 +60,15 @@ async def wait( timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None, ) -> SandboxProcessResult: + if self._collector is not None: + timeout = None + if timeout_sec is not None and timeout_sec > 0: + timeout = timeout_sec + elif timeout_ms is not None and timeout_ms > 0: + timeout = timeout_ms / 1000 + if not self._collector.done(): + await asyncio.wait_for(asyncio.shield(self._collector), timeout) + return self._collected_result() payload = await self._transport.request_json( f"/sandbox/processes/{self.id}/wait", method="POST", @@ -60,6 +79,10 @@ async def wait( headers={"content-type": "application/json"}, ) result = SandboxProcessResult(**payload["result"]) + if result.output_truncated: + raise ProcessOutput(self.id, 0).failure( + "Retained process output is incomplete; collect output from process start" + ) self._summary = SandboxProcessSummary( id=result.id, status=result.status, @@ -133,6 +156,21 @@ async def write_stdin( ) async def stream(self, from_seq: Optional[int] = None) -> AsyncIterator[object]: + if self._output is not None: + index = 0 + while True: + self._changed.clear() + while index < len(self._output.events): + event = self._output.events[index] + index += 1 + if from_seq is None or event.seq >= from_seq: + yield event + if self._collector.done(): + yield SandboxProcessExitEvent( + type="exit", result=self._collected_result() + ) + return + await self._changed.wait() params = {"from_seq": from_seq} if from_seq and from_seq > 0 else None async for event in self._transport.stream_sse( f"/sandbox/processes/{self.id}/stream", @@ -153,6 +191,70 @@ async def stream(self, from_seq: Optional[int] = None) -> AsyncIterator[object]: result=SandboxProcessResult(**data), ) + def _collected_result(self) -> SandboxProcessResult: + if self._output.error is not None: + raise self._output.error + if self._output.result is None: + raise self._output.failure( + "Command stream ended before its completion event" + ) + result = self._output.result + self._summary = self._summary.model_copy( + update={ + "status": result.status, + "exit_code": result.exit_code, + "completed_at": result.completed_at, + } + ) + return result + + async def _collect(self) -> None: + try: + async for event in self._events: + self._output.consume(event) + self._changed.set() + if self._output.result is not None: + return + self._output.error = self._output.failure( + "Command stream ended before its completion event" + ) + except asyncio.CancelledError: + if self._output.result is None: + self._output.error = self._output.failure( + "Command output collection disconnected" + ) + except Exception as error: + self._output.error = ( + self._output.failure(str(error)) + if not hasattr(error, "code") + else error + ) + finally: + try: + await self._events.aclose() + except Exception as error: + if self._output.result is None and self._output.error is None: + self._output.error = self._output.failure(str(error)) + finally: + self._changed.set() + + async def disconnect(self) -> None: + """Stop collecting output; the detached command continues running.""" + if self._collector is not None and not self._collector.done(): + if self._output.result is None and self._output.error is None: + self._output.error = self._output.failure( + "Command output collection disconnected" + ) + self._collector.cancel() + try: + await self._collector + except asyncio.CancelledError: + pass + finally: + # Cancellation may happen before the collector gets its first turn. + await self._events.aclose() + self._changed.set() + async def result(self) -> SandboxProcessResult: return await self.wait() @@ -170,22 +272,21 @@ async def exec( timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None, run_as: Optional[str] = None, + max_output_bytes: int = DEFAULT_MAX_PROCESS_OUTPUT_BYTES, ) -> SandboxProcessResult: - params = _normalize_exec_params( + handle = await self.start( input, cwd=cwd, env=env, timeout_ms=timeout_ms, timeout_sec=timeout_sec, run_as=run_as, + max_output_bytes=max_output_bytes, ) - payload = await self._transport.request_json( - "/sandbox/exec", - method="POST", - json_body=dump_request(params, SandboxExecParams), - headers={"content-type": "application/json"}, - ) - return SandboxProcessResult(**payload["result"]) + try: + return await handle.wait() + finally: + await handle.disconnect() async def start( self, @@ -196,7 +297,9 @@ async def start( timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None, run_as: Optional[str] = None, + max_output_bytes: int = DEFAULT_MAX_PROCESS_OUTPUT_BYTES, ) -> SandboxProcessHandle: + validate_output_limit(max_output_bytes) params = _normalize_exec_params( input, cwd=cwd, @@ -205,16 +308,25 @@ async def start( timeout_sec=timeout_sec, run_as=run_as, ) - payload = await self._transport.request_json( + events = self._transport.stream_sse( "/sandbox/processes", method="POST", json_body=dump_request(params, SandboxExecParams), - headers={"content-type": "application/json"}, - ) - return SandboxProcessHandle( - self._transport, - SandboxProcessSummary(**payload["process"]), ) + try: + started = await events.__anext__() + if started["event"] != "started": + raise RuntimeError("Expected process start event") + handle = SandboxProcessHandle( + self._transport, SandboxProcessSummary(**started["data"]) + ) + except BaseException: + await events.aclose() + raise + handle._events = events + handle._output = ProcessOutput(handle.id, max_output_bytes) + handle._collector = asyncio.create_task(handle._collect()) + return handle async def get(self, process_id: str) -> SandboxProcessHandle: payload = await self._transport.request_json(f"/sandbox/processes/{process_id}") diff --git a/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_transport.py b/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_transport.py index 96e96ee1..ca1ec956 100644 --- a/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_transport.py +++ b/hyperbrowser/client/managers/async_manager/sandboxes/sandbox_transport.py @@ -3,6 +3,7 @@ import httpx +from .....exceptions import HyperbrowserError from .....sandbox_common import ( RuntimeConnection, build_headers, @@ -12,7 +13,11 @@ request_context, resolve_runtime_transport_target, ) -from ...sandboxes.shared import _build_query_path, _is_replayable_http_content +from ...sandboxes.shared import ( + PROCESS_STREAM_IDLE_TIMEOUT_SECONDS, + _build_query_path, + _is_replayable_http_content, +) class RuntimeTransport: @@ -83,9 +88,16 @@ async def stream_bytes( await client.aclose() async def stream_sse( - self, path: str, params: Optional[Dict[str, object]] = None + self, + path: str, + params: Optional[Dict[str, object]] = None, + *, + method: str = "GET", + json_body: Optional[Dict[str, object]] = None, ) -> AsyncIterator[Dict[str, object]]: - client, response = await self._open_stream(path, params=params) + client, response = await self._open_stream( + path, params=params, method=method, json_body=json_body + ) event_name = "message" event_id = None data_lines = [] @@ -115,6 +127,8 @@ def flush_event(): try: async for line in response.aiter_lines(): + # HTTPX < 0.24 includes line terminators in aiter_lines(). + line = line.rstrip("\r\n") if line == "": event = flush_event() if event is not None: @@ -194,18 +208,45 @@ async def _open_stream( *, params: Optional[Dict[str, object]] = None, allow_refresh: bool = True, + method: str = "GET", + json_body: Optional[Dict[str, object]] = None, ): connection = await self._resolve_connection(False) - client, response = await self._send_stream(connection, path, params=params) + client, response = await self._send_stream( + connection, path, params=params, method=method, json_body=json_body + ) if response.status_code == 401 and allow_refresh: await response.aclose() await client.aclose() refreshed = await self._resolve_connection(True) - client, response = await self._send_stream(refreshed, path, params=params) + client, response = await self._send_stream( + refreshed, path, params=params, method=method, json_body=json_body + ) - if not response.is_success: - await response.aread() - ensure_response_ok(response, "runtime") + try: + if not response.is_success: + await response.aread() + ensure_response_ok(response, "runtime") + if method == "POST" and "text/event-stream" not in response.headers.get( + "content-type", "" + ): + raise HyperbrowserError( + "Receiver does not support streaming command start; update the receiver. The command may have started; do not retry it automatically.", + code="streaming_not_supported", + service="runtime", + ) + if "text/event-stream" in response.headers.get("content-type", ""): + # HTTPX passes this timeout extension through to body reads. + # Change it before reading the body, after response headers have + # arrived under the ordinary request timeout. Heartbeats reset + # this idle timeout independently of the command's deadline. + response.request.extensions["timeout"]["read"] = ( + PROCESS_STREAM_IDLE_TIMEOUT_SECONDS + ) + except BaseException: + await response.aclose() + await client.aclose() + raise return client, response async def _open_binary_stream( @@ -320,6 +361,8 @@ async def _send_stream( path: str, *, params: Optional[Dict[str, object]], + method: str = "GET", + json_body: Optional[Dict[str, object]] = None, ): request_path = _build_query_path(path, params) target = resolve_runtime_transport_target( @@ -335,7 +378,9 @@ async def _send_stream( client = httpx.AsyncClient(timeout=self._timeout) try: - request = client.build_request("GET", target.url, headers=headers) + request = client.build_request( + method, target.url, headers=headers, json=json_body + ) response = await client.send(request, stream=True) return client, response except BaseException as error: @@ -344,5 +389,5 @@ async def _send_stream( error, "runtime", "Unknown runtime request error", - request_context("GET", path), + request_context(method, path), ) diff --git a/hyperbrowser/client/managers/sandboxes/process_output.py b/hyperbrowser/client/managers/sandboxes/process_output.py new file mode 100644 index 00000000..ff1524e0 --- /dev/null +++ b/hyperbrowser/client/managers/sandboxes/process_output.py @@ -0,0 +1,88 @@ +"""Collection and validation of the receiver's command event stream.""" + +import base64 +import codecs +from typing import Dict, List, Optional + +from ....exceptions import HyperbrowserError +from ....models.sandbox import SandboxProcessOutputEvent, SandboxProcessResult + +DEFAULT_MAX_PROCESS_OUTPUT_BYTES = 64 * 1024 * 1024 + + +class ProcessOutput: + def __init__(self, process_id: str, max_bytes: int): + self.process_id = process_id + self.max_bytes = max_bytes + self.size = 0 + self.seq = 0 + self.events: List[SandboxProcessOutputEvent] = [] + self.result: Optional[SandboxProcessResult] = None + self.error: Optional[BaseException] = None + self._chunks: Dict[str, List[str]] = {"stdout": [], "stderr": []} + self._decoders = { + stream: codecs.getincrementaldecoder("utf-8")(errors="replace") + for stream in ("stdout", "stderr", "system") + } + + def failure( + self, message: str, code: str = "incomplete_output" + ) -> HyperbrowserError: + return HyperbrowserError( + message, + code=code, + service="runtime", + details={"process_id": self.process_id, "last_seq": self.seq}, + ) + + def consume(self, event) -> None: + kind, data = event["event"], event["data"] + if kind == "output": + if data["seq"] != self.seq + 1: + raise self.failure("Command output contains a sequence gap") + stream = data["stream"] + if stream not in self._decoders: + raise self.failure("Unknown command output stream") + raw = ( + base64.b64decode(data["data"], validate=True) + if data.get("encoding") == "base64" + else data["data"].encode("utf-8") + ) + self.size += len(raw) + if self.size > self.max_bytes: + raise self.failure( + "Command output exceeds max_output_bytes; increase the collection limit or disconnect a detached process", + "output_limit_exceeded", + ) + self.seq = data["seq"] + text = self._decoders[stream].decode(raw) + self._chunks["stdout" if stream == "stdout" else "stderr"].append(text) + self.events.append( + SandboxProcessOutputEvent( + type=stream, + seq=self.seq, + data=text, + timestamp=data["timestamp"], + ) + ) + elif kind == "done": + if data.get("last_seq") != self.seq or data.get("output_truncated", False): + raise self.failure("Receiver reported incomplete command output") + for stream, decoder in self._decoders.items(): + self._chunks["stdout" if stream == "stdout" else "stderr"].append( + decoder.decode(b"", final=True) + ) + result = dict(data) + result["stdout"] = "".join(self._chunks["stdout"]) + result["stderr"] = "".join(self._chunks["stderr"]) + self.result = SandboxProcessResult(**result) + elif kind == "error": + raise self.failure( + str(data.get("error", "Command stream failed")), + data.get("code", "incomplete_output"), + ) + + +def validate_output_limit(value: int) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ValueError("max_output_bytes must be a positive integer") diff --git a/hyperbrowser/client/managers/sandboxes/shared.py b/hyperbrowser/client/managers/sandboxes/shared.py index e4fd4b68..729502e8 100644 --- a/hyperbrowser/client/managers/sandboxes/shared.py +++ b/hyperbrowser/client/managers/sandboxes/shared.py @@ -27,6 +27,8 @@ ) DEFAULT_WATCH_TIMEOUT_MS = 60_000 +# The receiver sends process SSE keepalives every 15 seconds. +PROCESS_STREAM_IDLE_TIMEOUT_SECONDS = 60.0 SHELL_SAFE_TOKEN_PATTERN = re.compile(r"^[A-Za-z0-9_@%+=:,./-]+$") diff --git a/hyperbrowser/client/managers/sync_manager/sandbox.py b/hyperbrowser/client/managers/sync_manager/sandbox.py index 5f2193f0..b78d7111 100644 --- a/hyperbrowser/client/managers/sync_manager/sandbox.py +++ b/hyperbrowser/client/managers/sync_manager/sandbox.py @@ -299,6 +299,7 @@ def exec( timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None, run_as: Optional[str] = None, + max_output_bytes: int = 64 * 1024 * 1024, ): return self.processes.exec( input, @@ -307,6 +308,7 @@ def exec( timeout_ms=timeout_ms, timeout_sec=timeout_sec, run_as=run_as, + max_output_bytes=max_output_bytes, ) def get_process(self, process_id: str) -> SandboxProcessHandle: diff --git a/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_processes.py b/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_processes.py index 3f1c65ba..9c70c56b 100644 --- a/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_processes.py +++ b/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_processes.py @@ -1,4 +1,5 @@ import base64 +import threading from typing import Dict, Optional, Union from ...._request import coerce_request, dump_request @@ -16,6 +17,11 @@ SandboxProcessStdinParams as SandboxProcessStdinParamsDict, ) from ...sandboxes.shared import _normalize_exec_params +from ...sandboxes.process_output import ( + DEFAULT_MAX_PROCESS_OUTPUT_BYTES, + ProcessOutput, + validate_output_limit, +) from .sandbox_transport import RuntimeTransport DEFAULT_PROCESS_KILL_WAIT_SECONDS = 5.0 @@ -25,6 +31,11 @@ class SandboxProcessHandle: def __init__(self, transport: RuntimeTransport, summary: SandboxProcessSummary): self._transport = transport self._summary = summary + self._output = None + self._collector = None + self._events = None + self._changed = threading.Condition() + self._close_stream = None @property def id(self) -> str: @@ -46,6 +57,21 @@ def refresh(self) -> "SandboxProcessHandle": return self def wait(self, timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None): + if self._collector is not None: + timeout = None + if timeout_sec is not None and timeout_sec > 0: + timeout = timeout_sec + elif timeout_ms is not None and timeout_ms > 0: + timeout = timeout_ms / 1000 + with self._changed: + ready = self._changed.wait_for( + lambda: self._output.result is not None + or self._output.error is not None, + timeout, + ) + if not ready: + raise TimeoutError("Timed out waiting for command output") + return self._collected_result() payload = self._transport.request_json( f"/sandbox/processes/{self.id}/wait", method="POST", @@ -56,6 +82,10 @@ def wait(self, timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = No headers={"content-type": "application/json"}, ) result = SandboxProcessResult(**payload["result"]) + if result.output_truncated: + raise ProcessOutput(self.id, 0).failure( + "Retained process output is incomplete; collect output from process start" + ) self._summary = SandboxProcessSummary( id=result.id, status=result.status, @@ -129,6 +159,27 @@ def write_stdin( ) def stream(self, from_seq: Optional[int] = None): + if self._output is not None: + index = 0 + while True: + with self._changed: + events = self._output.events[index:] + index += len(events) + done = ( + self._output.result is not None + or self._output.error is not None + ) + if not events and not done: + self._changed.wait() + continue + for event in events: + if from_seq is None or event.seq >= from_seq: + yield event + if done: + yield SandboxProcessExitEvent( + type="exit", result=self._collected_result() + ) + return params = {"from_seq": from_seq} if from_seq and from_seq > 0 else None for event in self._transport.stream_sse( f"/sandbox/processes/{self.id}/stream", @@ -149,6 +200,67 @@ def stream(self, from_seq: Optional[int] = None): result=SandboxProcessResult(**data), ) + def _collected_result(self) -> SandboxProcessResult: + if self._output.error is not None: + raise self._output.error + if self._output.result is None: + raise self._output.failure( + "Command stream ended before its completion event" + ) + result = self._output.result + self._summary = self._summary.model_copy( + update={ + "status": result.status, + "exit_code": result.exit_code, + "completed_at": result.completed_at, + } + ) + return result + + def _collect(self) -> None: + try: + for event in self._events: + with self._changed: + if self._output.error is not None: + return + self._output.consume(event) + self._changed.notify_all() + if self._output.result is not None: + return + self._output.error = self._output.failure( + "Command stream ended before its completion event" + ) + except Exception as error: + with self._changed: + if self._output.error is None: + self._output.error = ( + self._output.failure(str(error)) + if not hasattr(error, "code") + else error + ) + finally: + try: + self._events.close() + except Exception as error: + with self._changed: + if self._output.result is None and self._output.error is None: + self._output.error = self._output.failure(str(error)) + finally: + with self._changed: + self._changed.notify_all() + + def disconnect(self) -> None: + """Stop collecting output; the detached command continues running.""" + if self._output is not None: + with self._changed: + if self._output.result is None and self._output.error is None: + self._output.error = self._output.failure( + "Command output collection disconnected" + ) + self._changed.notify_all() + if self._close_stream is not None: + self._close_stream() + def result(self) -> SandboxProcessResult: return self.wait() @@ -166,22 +278,21 @@ def exec( timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None, run_as: Optional[str] = None, + max_output_bytes: int = DEFAULT_MAX_PROCESS_OUTPUT_BYTES, ) -> SandboxProcessResult: - params = _normalize_exec_params( + handle = self.start( input, cwd=cwd, env=env, timeout_ms=timeout_ms, timeout_sec=timeout_sec, run_as=run_as, + max_output_bytes=max_output_bytes, ) - payload = self._transport.request_json( - "/sandbox/exec", - method="POST", - json_body=dump_request(params, SandboxExecParams), - headers={"content-type": "application/json"}, - ) - return SandboxProcessResult(**payload["result"]) + try: + return handle.wait() + finally: + handle.disconnect() def start( self, @@ -192,7 +303,9 @@ def start( timeout_ms: Optional[int] = None, timeout_sec: Optional[int] = None, run_as: Optional[str] = None, + max_output_bytes: int = DEFAULT_MAX_PROCESS_OUTPUT_BYTES, ) -> SandboxProcessHandle: + validate_output_limit(max_output_bytes) params = _normalize_exec_params( input, cwd=cwd, @@ -201,16 +314,29 @@ def start( timeout_sec=timeout_sec, run_as=run_as, ) - payload = self._transport.request_json( + close_stream = [] + events = self._transport.stream_sse( "/sandbox/processes", method="POST", json_body=dump_request(params, SandboxExecParams), - headers={"content-type": "application/json"}, - ) - return SandboxProcessHandle( - self._transport, - SandboxProcessSummary(**payload["process"]), + on_open=close_stream.append, ) + try: + started = next(events) + if started["event"] != "started": + raise RuntimeError("Expected process start event") + handle = SandboxProcessHandle( + self._transport, SandboxProcessSummary(**started["data"]) + ) + except BaseException: + events.close() + raise + handle._close_stream = close_stream[0] if close_stream else None + handle._events = events + handle._output = ProcessOutput(handle.id, max_output_bytes) + handle._collector = threading.Thread(target=handle._collect, daemon=True) + handle._collector.start() + return handle def get(self, process_id: str) -> SandboxProcessHandle: payload = self._transport.request_json(f"/sandbox/processes/{process_id}") diff --git a/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_transport.py b/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_transport.py index 74573e50..9ddf9ce6 100644 --- a/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_transport.py +++ b/hyperbrowser/client/managers/sync_manager/sandboxes/sandbox_transport.py @@ -1,8 +1,10 @@ import json -from typing import Any, Dict, Iterator, Optional +import socket +from typing import Any, Callable, Dict, Iterator, Optional import httpx +from .....exceptions import HyperbrowserError from .....sandbox_common import ( RuntimeConnection, build_headers, @@ -12,7 +14,11 @@ request_context, resolve_runtime_transport_target, ) -from ...sandboxes.shared import _build_query_path, _is_replayable_http_content +from ...sandboxes.shared import ( + PROCESS_STREAM_IDLE_TIMEOUT_SECONDS, + _build_query_path, + _is_replayable_http_content, +) class RuntimeTransport: @@ -81,9 +87,33 @@ def stream_bytes( client.close() def stream_sse( - self, path: str, params: Optional[Dict[str, object]] = None + self, + path: str, + params: Optional[Dict[str, object]] = None, + *, + method: str = "GET", + json_body: Optional[Dict[str, object]] = None, + on_open: Optional[Callable] = None, ) -> Iterator[Dict[str, object]]: - client, response = self._open_stream(path, params=params) + client, response = self._open_stream( + path, params=params, method=method, json_body=json_body + ) + if on_open is not None: + + def disconnect(): + # close() alone need not interrupt a recv() on another thread. + # Each stream owns its HTTP client/connection, so shutdown is safe. + network = response.extensions.get("network_stream") + if network is not None: + sock = network.get_extra_info("socket") + if sock is not None: + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + response.close() + + on_open(disconnect) event_name = "message" event_id = None data_lines = [] @@ -113,6 +143,8 @@ def flush_event(): try: for line in response.iter_lines(): + # HTTPX < 0.24 includes line terminators in iter_lines(). + line = line.rstrip("\r\n") if line == "": event = flush_event() if event is not None: @@ -192,18 +224,45 @@ def _open_stream( *, params: Optional[Dict[str, object]] = None, allow_refresh: bool = True, + method: str = "GET", + json_body: Optional[Dict[str, object]] = None, ): connection = self._resolve_connection(False) - client, response = self._send_stream(connection, path, params=params) + client, response = self._send_stream( + connection, path, params=params, method=method, json_body=json_body + ) if response.status_code == 401 and allow_refresh: response.close() client.close() refreshed = self._resolve_connection(True) - client, response = self._send_stream(refreshed, path, params=params) + client, response = self._send_stream( + refreshed, path, params=params, method=method, json_body=json_body + ) - if not response.is_success: - response.read() - ensure_response_ok(response, "runtime") + try: + if not response.is_success: + response.read() + ensure_response_ok(response, "runtime") + if method == "POST" and "text/event-stream" not in response.headers.get( + "content-type", "" + ): + raise HyperbrowserError( + "Receiver does not support streaming command start; update the receiver. The command may have started; do not retry it automatically.", + code="streaming_not_supported", + service="runtime", + ) + if "text/event-stream" in response.headers.get("content-type", ""): + # HTTPX passes this timeout extension through to body reads. + # Change it before reading the body, after response headers have + # arrived under the ordinary request timeout. Heartbeats reset + # this idle timeout independently of the command's deadline. + response.request.extensions["timeout"]["read"] = ( + PROCESS_STREAM_IDLE_TIMEOUT_SECONDS + ) + except BaseException: + response.close() + client.close() + raise return client, response def _open_binary_stream( @@ -318,6 +377,8 @@ def _send_stream( path: str, *, params: Optional[Dict[str, object]], + method: str = "GET", + json_body: Optional[Dict[str, object]] = None, ): request_path = _build_query_path(path, params) target = resolve_runtime_transport_target( @@ -333,7 +394,9 @@ def _send_stream( client = httpx.Client(timeout=self._timeout) try: - request = client.build_request("GET", target.url, headers=headers) + request = client.build_request( + method, target.url, headers=headers, json=json_body + ) response = client.send(request, stream=True) return client, response except BaseException as error: @@ -342,5 +405,5 @@ def _send_stream( error, "runtime", "Unknown runtime request error", - request_context("GET", path), + request_context(method, path), ) diff --git a/hyperbrowser/models/sandbox.py b/hyperbrowser/models/sandbox.py index 18a4d97c..89d0c5ff 100644 --- a/hyperbrowser/models/sandbox.py +++ b/hyperbrowser/models/sandbox.py @@ -577,6 +577,8 @@ class SandboxProcessSummary(SandboxBaseModel): class SandboxProcessResult(SandboxBaseModel): + output_truncated: bool = False + last_seq: Optional[int] = None id: str status: SandboxProcessStatus exit_code: Optional[int] = Field( diff --git a/tests/sandbox/e2e/test_async_process.py b/tests/sandbox/e2e/test_async_process.py index f2d1ab38..910e970d 100644 --- a/tests/sandbox/e2e/test_async_process.py +++ b/tests/sandbox/e2e/test_async_process.py @@ -1,3 +1,5 @@ +import asyncio + import pytest from hyperbrowser.models import SandboxExecParams @@ -101,21 +103,28 @@ async def test_async_sandbox_process_e2e(): assert result.exit_code == 0 assert "result-alias-ok" in result.stdout + noisy_line = "process-replay-window-overflow-" + "x" * 64 noisy_process = await sandbox.processes.start( SandboxExecParams( command="bash", args=[ "-lc", - 'yes "process-replay-window-overflow-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" | head -n 120000', + f"yes {noisy_line} | head -n 120000", ], ) ) result = await noisy_process.result() - assert len(result.stdout) > 3 * 1024 * 1024 + assert result.stdout == (noisy_line + "\n") * 120000 + events = await _collect_process_stream(noisy_process.stream(1)) + assert ( + "".join(event.data for event in events if event.type == "stdout") + == result.stdout + ) + reattached = await sandbox.get_process(noisy_process.id) await expect_hyperbrowser_error_async( "process replay window expired", - lambda: _collect_process_stream(noisy_process.stream(1)), + lambda: _collect_process_stream(reattached.stream(1)), status_code=410, code="replay_window_expired", service="runtime", @@ -126,14 +135,8 @@ async def test_async_sandbox_process_e2e(): timeout_process = await sandbox.processes.start( SandboxExecParams(command="bash", args=["-lc", "sleep 10"]) ) - await expect_hyperbrowser_error_async( - "process wait timeout", - lambda: timeout_process.wait(timeout_ms=100), - status_code=408, - service="runtime", - retryable=False, - message_includes="timed out", - ) + with pytest.raises(asyncio.TimeoutError): + await timeout_process.wait(timeout_ms=100) await timeout_process.signal("TERM") result = await timeout_process.wait(timeout_ms=3000) assert result.status in {"exited", "failed", "killed", "timed_out"} diff --git a/tests/sandbox/e2e/test_process.py b/tests/sandbox/e2e/test_process.py index 3c3f3aea..22ef9ef4 100644 --- a/tests/sandbox/e2e/test_process.py +++ b/tests/sandbox/e2e/test_process.py @@ -1,3 +1,5 @@ +import pytest + from hyperbrowser.models import SandboxExecParams from tests.helpers.config import create_client @@ -97,21 +99,28 @@ def test_sandbox_process_e2e(): assert result.exit_code == 0 assert "result-alias-ok" in result.stdout + noisy_line = "process-replay-window-overflow-" + "x" * 64 noisy_process = sandbox.processes.start( SandboxExecParams( command="bash", args=[ "-lc", - 'yes "process-replay-window-overflow-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" | head -n 120000', + f"yes {noisy_line} | head -n 120000", ], ) ) result = noisy_process.result() - assert len(result.stdout) > 3 * 1024 * 1024 + assert result.stdout == (noisy_line + "\n") * 120000 + events = _collect_process_stream(noisy_process.stream(1)) + assert ( + "".join(event.data for event in events if event.type == "stdout") + == result.stdout + ) + reattached = sandbox.get_process(noisy_process.id) expect_hyperbrowser_error( "process replay window expired", - lambda: _collect_process_stream(noisy_process.stream(1)), + lambda: _collect_process_stream(reattached.stream(1)), status_code=410, code="replay_window_expired", service="runtime", @@ -122,14 +131,8 @@ def test_sandbox_process_e2e(): timeout_process = sandbox.processes.start( SandboxExecParams(command="bash", args=["-lc", "sleep 10"]) ) - expect_hyperbrowser_error( - "process wait timeout", - lambda: timeout_process.wait(timeout_ms=100), - status_code=408, - service="runtime", - retryable=False, - message_includes="timed out", - ) + with pytest.raises(TimeoutError): + timeout_process.wait(timeout_ms=100) timeout_process.signal("TERM") result = timeout_process.wait(timeout_ms=3000) assert result.status in {"exited", "failed", "killed", "timed_out"} diff --git a/tests/test_sandbox_process_collection.py b/tests/test_sandbox_process_collection.py new file mode 100644 index 00000000..1171bf04 --- /dev/null +++ b/tests/test_sandbox_process_collection.py @@ -0,0 +1,234 @@ +import asyncio +import base64 +import threading + +import pytest + +from hyperbrowser.client.managers.async_manager.sandboxes.sandbox_processes import ( + SandboxProcessesApi as AsyncProcesses, +) +from hyperbrowser.client.managers.sync_manager.sandboxes.sandbox_processes import ( + SandboxProcessesApi as SyncProcesses, +) +from hyperbrowser.exceptions import HyperbrowserError + + +def output(seq, data, stream="stdout"): + return { + "event": "output", + "data": { + "seq": seq, + "stream": stream, + "data": base64.b64encode(data).decode(), + "encoding": "base64", + "timestamp": 1, + }, + } + + +def done(seq, **kwargs): + return { + "event": "done", + "data": { + "id": "p1", + "status": "exited", + "exit_code": 7, + "started_at": 1, + "completed_at": 2, + "last_seq": seq, + **kwargs, + }, + } + + +STARTED = { + "event": "started", + "data": { + "id": "p1", + "status": "running", + "command": "test", + "cwd": "/tmp", + "started_at": 1, + }, +} + + +class SyncTransport: + def __init__(self, events): + self.events = events + self.calls = [] + self.closed = threading.Event() + + def stream_sse(self, path, **kwargs): + self.calls.append((path, kwargs)) + kwargs["on_open"](self.closed.set) + try: + yield STARTED + yield from self.events + finally: + self.closed.set() + + +class AsyncTransport: + def __init__(self, events): + self.events = events + self.calls = [] + self.closed = False + self.gate = None + + async def stream_sse(self, path, **kwargs): + self.calls.append((path, kwargs)) + try: + yield STARTED + if self.gate is not None: + await self.gate.wait() + for event in self.events: + yield event + finally: + self.closed = True + + +def large_output(): + # Exceeds the receiver's old 4 MiB replay limit and splits a UTF-8 character. + chunk = b"x" * 32768 + events = [output(i + 1, chunk) for i in range(160)] + events.extend( + [ + output(161, b"\xe2"), + output(162, b"\x82\xac"), + output(163, b"error", "stderr"), + done(163), + ] + ) + return events, "x" * (160 * len(chunk)) + "€" + + +def test_sync_collects_large_output_and_streams_from_same_request(): + events, expected = large_output() + transport = SyncTransport(events) + handle = SyncProcesses(transport).start("test") + result = handle.wait(timeout_sec=5) + assert (result.stdout, result.stderr, result.exit_code) == (expected, "error", 7) + assert handle.status == "exited" + streamed = list(handle.stream()) + assert "".join(e.data for e in streamed if e.type == "stdout") == expected + assert streamed[-1].result == result + handle.disconnect() + assert transport.closed.wait(1) + assert len(transport.calls) == 1 + assert transport.calls[0][1]["method"] == "POST" + + +@pytest.mark.anyio +async def test_async_collects_large_output_and_streams_from_same_request(): + events, expected = large_output() + transport = AsyncTransport(events) + handle = await AsyncProcesses(transport).start("test") + streamed = [event async for event in handle.stream()] + result = await handle.wait() + assert (result.stdout, result.stderr, result.exit_code) == (expected, "error", 7) + assert "".join(e.data for e in streamed if e.type == "stdout") == expected + assert streamed[-1].result == result + assert transport.closed + assert len(transport.calls) == 1 + + +FAILURES = [ + ([output(2, b"gap"), done(2)], 100, "incomplete_output"), + ([output(1, b"no completion")], 100, "incomplete_output"), + ([output(1, b"tail missing"), done(2)], 100, "incomplete_output"), + ([done(0, output_truncated=True)], 100, "incomplete_output"), + ([output(1, b"too much"), done(1)], 4, "output_limit_exceeded"), +] + + +@pytest.mark.parametrize("events,limit,code", FAILURES) +def test_sync_incomplete_output_is_not_success_or_reexecuted(events, limit, code): + transport = SyncTransport(events) + with pytest.raises(HyperbrowserError) as exc: + SyncProcesses(transport).exec("test", max_output_bytes=limit) + assert exc.value.code == code + assert exc.value.details["process_id"] == "p1" + assert not exc.value.retryable + assert transport.closed.wait(1) + assert len(transport.calls) == 1 + + +@pytest.mark.anyio +@pytest.mark.parametrize("events,limit,code", FAILURES) +async def test_async_incomplete_output_is_not_success_or_reexecuted( + events, limit, code +): + transport = AsyncTransport(events) + with pytest.raises(HyperbrowserError) as exc: + await AsyncProcesses(transport).exec("test", max_output_bytes=limit) + assert exc.value.code == code + assert not exc.value.retryable + assert transport.closed + assert len(transport.calls) == 1 + + +@pytest.mark.anyio +async def test_async_wait_timeout_keeps_collector_alive(): + transport = AsyncTransport([output(1, b"later"), done(1)]) + transport.gate = asyncio.Event() + handle = await AsyncProcesses(transport).start("test") + with pytest.raises(asyncio.TimeoutError): + await handle.wait(timeout_ms=1) + assert not transport.closed + transport.gate.set() + assert (await handle.wait()).stdout == "later" + + +@pytest.mark.anyio +@pytest.mark.parametrize("start_collector", [False, True]) +async def test_async_disconnect_closes_stream_without_killing_command(start_collector): + transport = AsyncTransport([]) + transport.gate = asyncio.Event() + handle = await AsyncProcesses(transport).start("test") + if start_collector: + await asyncio.sleep(0) + await handle.disconnect() + assert transport.closed + with pytest.raises(HyperbrowserError, match="disconnected"): + await handle.wait() + assert len(transport.calls) == 1 + + +def test_sync_wait_timeout_and_disconnect_unblock_collector(): + transport = SyncTransport([]) + + def events(): + assert transport.closed.wait(5) + yield done(0) + + transport.events = events() + handle = SyncProcesses(transport).start("test") + with pytest.raises(TimeoutError): + handle.wait(timeout_ms=1) + handle.disconnect() + handle._collector.join(1) + assert not handle._collector.is_alive() + with pytest.raises(HyperbrowserError, match="disconnected"): + handle.wait() + + +@pytest.mark.anyio +async def test_async_exec_cancellation_closes_stream(): + transport = AsyncTransport([]) + transport.gate = asyncio.Event() + task = asyncio.create_task(AsyncProcesses(transport).exec("test")) + while not transport.calls: + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert transport.closed + + +@pytest.mark.parametrize("limit", [0, -1, True, None]) +def test_invalid_collection_limit_rejected_before_start(limit): + transport = SyncTransport([]) + with pytest.raises(ValueError, match="positive integer"): + SyncProcesses(transport).start("test", max_output_bytes=limit) + assert not transport.calls diff --git a/tests/test_sandbox_runtime_transport.py b/tests/test_sandbox_runtime_transport.py index d888d1bf..39c67eb7 100644 --- a/tests/test_sandbox_runtime_transport.py +++ b/tests/test_sandbox_runtime_transport.py @@ -19,6 +19,66 @@ def _connection(token: str) -> RuntimeConnection: ) +def test_sync_streaming_start_does_not_fallback_or_reexecute_on_old_receiver( + monkeypatch, +): + requests = [] + + def respond(request): + requests.append(request) + return httpx.Response(202, json={"process": {"id": "already-started"}}) + + client = httpx.Client + monkeypatch.setattr( + sync_transport_module.httpx, + "Client", + lambda **kwargs: client(transport=httpx.MockTransport(respond), **kwargs), + ) + transport = sync_transport_module.RuntimeTransport(lambda _: _connection("token")) + with pytest.raises(HyperbrowserError) as error: + list( + transport.stream_sse( + "/sandbox/processes", method="POST", json_body={"command": "echo hi"} + ) + ) + assert error.value.code == "streaming_not_supported" + assert not error.value.retryable + assert len(requests) == 1 + assert requests[0].method == "POST" + assert requests[0].headers["accept"] == "text/event-stream" + + +@pytest.mark.anyio +async def test_async_streaming_start_does_not_fallback_or_reexecute_on_old_receiver( + monkeypatch, +): + requests = [] + + def respond(request): + requests.append(request) + return httpx.Response(202, json={"process": {"id": "already-started"}}) + + client = httpx.AsyncClient + monkeypatch.setattr( + async_transport_module.httpx, + "AsyncClient", + lambda **kwargs: client(transport=httpx.MockTransport(respond), **kwargs), + ) + + async def resolve(_): + return _connection("token") + + transport = async_transport_module.RuntimeTransport(resolve) + with pytest.raises(HyperbrowserError) as error: + async for _ in transport.stream_sse( + "/sandbox/processes", method="POST", json_body={"command": "echo hi"} + ): + pass + assert error.value.code == "streaming_not_supported" + assert not error.value.retryable + assert len(requests) == 1 + + def test_sync_runtime_transport_does_not_retry_consumed_stream_body(monkeypatch): calls = [] diff --git a/tests/test_sandbox_stream_timeouts.py b/tests/test_sandbox_stream_timeouts.py new file mode 100644 index 00000000..e5606797 --- /dev/null +++ b/tests/test_sandbox_stream_timeouts.py @@ -0,0 +1,260 @@ +"""Exercise stream deadlines with real sockets, including HTTPX body reads.""" + +import json +import threading +from functools import partial +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import SimpleNamespace + +import anyio +import httpx +import pytest + +from hyperbrowser.client.managers.async_manager.sandboxes import ( + sandbox_transport as async_transport, +) +from hyperbrowser.client.managers.async_manager.sandboxes.sandbox_processes import ( + SandboxProcessesApi as AsyncProcesses, +) +from hyperbrowser.client.managers.sync_manager.sandboxes import ( + sandbox_transport as sync_transport, +) +from hyperbrowser.client.managers.sync_manager.sandboxes.sandbox_processes import ( + SandboxProcessesApi as SyncProcesses, +) +from hyperbrowser.exceptions import HyperbrowserError +from hyperbrowser.sandbox_common import RuntimeConnection + + +@pytest.fixture +def stream_server(): + servers = [] + + def start(respond): + stopped = threading.Event() + requests = [] + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + requests.append((self.command, self.headers.get("Authorization"))) + self.rfile.read(int(self.headers.get("Content-Length", 0))) + try: + respond(self, stopped) + except (BrokenPipeError, ConnectionResetError): + pass # Expected when a timeout closes the client socket. + + do_POST = do_GET + + def headers_for(self, status=200, content_type="text/event-stream"): + self.send_response(status) + self.send_header("Content-Type", content_type) + self.end_headers() + + def event(self, name, data): + self.wfile.write( + ("event: " + name + "\ndata: " + json.dumps(data) + "\n\n").encode() + ) + self.wfile.flush() + + def started(self): + self.headers_for() + self.event( + "started", + { + "id": "p1", + "status": "running", + "command": "quiet", + "cwd": "/tmp", + "started_at": 1, + }, + ) + + def finished(self): + self.event( + "output", + { + "seq": 1, + "stream": "stdout", + "data": "finished", + "timestamp": 2, + }, + ) + self.event( + "done", + { + "id": "p1", + "status": "exited", + "exit_code": 0, + "started_at": 1, + "completed_at": 2, + "last_seq": 1, + }, + ) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread( + target=partial(server.serve_forever, poll_interval=0.02), daemon=True + ) + thread.start() + servers.append((server, thread, stopped)) + return SimpleNamespace( + url="http://127.0.0.1:{}".format(server.server_port), requests=requests + ) + + yield start + for server, thread, stopped in servers: + stopped.set() + server.shutdown() + server.server_close() + thread.join() + + +@pytest.fixture(params=["sync", "async"]) +def client(request, monkeypatch): + asynchronous = request.param == "async" + module = async_transport if asynchronous else sync_transport + + async def call(function, *args, **kwargs): + if asynchronous: + return await function(*args, **kwargs) + return await anyio.to_thread.run_sync(partial(function, *args, **kwargs)) + + def create(server, request_timeout, idle_timeout): + monkeypatch.setattr(module, "PROCESS_STREAM_IDLE_TIMEOUT_SECONDS", idle_timeout) + refreshes = [] + + def resolve(refresh): + refreshes.append(refresh) + return RuntimeConnection( + sandbox_id="test", + base_url=server.url, + token="fresh" if refresh else "old", + ) + + async def async_resolve(refresh): + return resolve(refresh) + + transport = module.RuntimeTransport( + async_resolve if asynchronous else resolve, timeout=request_timeout + ) + api = (AsyncProcesses if asynchronous else SyncProcesses)(transport) + return SimpleNamespace( + api=api, transport=transport, call=call, refreshes=refreshes + ) + + return create + + +@pytest.mark.anyio +async def test_process_stream_accepts_line_terminators( + stream_server, client, monkeypatch +): + # HTTPX 0.23, our minimum supported version, keeps the newline on each line. + iter_lines = httpx.Response.iter_lines + aiter_lines = httpx.Response.aiter_lines + + def legacy_lines(response): + for line in iter_lines(response): + yield line + "\n" + + async def async_legacy_lines(response): + async for line in aiter_lines(response): + yield line + "\n" + + monkeypatch.setattr(httpx.Response, "iter_lines", legacy_lines) + monkeypatch.setattr(httpx.Response, "aiter_lines", async_legacy_lines) + + def respond(handler, stopped): + handler.started() + handler.finished() + + c = client(stream_server(respond), request_timeout=1, idle_timeout=1) + handle = await c.call(c.api.start, "quiet") + try: + result = await c.call(handle.wait, timeout_sec=3) + assert (result.stdout, result.exit_code) == ("finished", 0) + finally: + await c.call(handle.disconnect) + + +@pytest.mark.anyio +@pytest.mark.parametrize("refresh", [False, True]) +async def test_quiet_process_and_heartbeats_outlive_request_timeout( + stream_server, client, refresh +): + def respond(handler, stopped): + if refresh and handler.headers["Authorization"] == "Bearer old": + handler.headers_for(401, "application/json") + handler.wfile.write(b'{"error":"expired"}') + return + handler.started() + # Longer than the ordinary request timeout, shorter than stream idle. + if stopped.wait(0.6): + return + # Total duration exceeds stream idle too: each heartbeat resets it. + for _ in range(4): + handler.event("keepalive", {}) + if stopped.wait(0.25): + return + handler.finished() + + server = stream_server(respond) + c = client(server, request_timeout=0.25, idle_timeout=1.0) + handle = await c.call(c.api.start, "quiet") + try: + result = await c.call(handle.wait, timeout_sec=5) + assert (result.stdout, result.exit_code) == ("finished", 0) + assert c.refreshes == ([False, True] if refresh else [False]) + assert len(server.requests) == (2 if refresh else 1) + finally: + await c.call(handle.disconnect) + + +@pytest.mark.anyio +async def test_missing_heartbeats_fail_without_reexecuting(stream_server, client): + def respond(handler, stopped): + handler.started() + stopped.wait(10) + + server = stream_server(respond) + c = client(server, request_timeout=5, idle_timeout=0.25) + handle = await c.call(c.api.start, "quiet") + try: + with pytest.raises(HyperbrowserError) as exc: + await c.call(handle.wait, timeout_sec=2) + assert exc.value.code == "incomplete_output" + assert exc.value.details["process_id"] == "p1" + assert not exc.value.retryable + assert len(server.requests) == 1 + finally: + await c.call(handle.disconnect) + + +@pytest.mark.anyio +async def test_response_headers_keep_ordinary_request_timeout(stream_server, client): + def respond(handler, stopped): + stopped.wait(2) + + server = stream_server(respond) + c = client(server, request_timeout=0.25, idle_timeout=5) + with pytest.raises(HyperbrowserError) as exc: + await c.call(c.api.start, "quiet") + assert isinstance(exc.value.original_error, httpx.ReadTimeout) + assert len(server.requests) == 1 + + +@pytest.mark.anyio +async def test_json_body_keeps_ordinary_request_timeout(stream_server, client): + def respond(handler, stopped): + handler.headers_for(200, "application/json") + stopped.wait(2) + + server = stream_server(respond) + c = client(server, request_timeout=0.25, idle_timeout=5) + with pytest.raises(HyperbrowserError) as exc: + await c.call(c.transport.request_json, "/sandbox/processes/p1") + assert isinstance(exc.value.original_error, httpx.ReadTimeout) + assert len(server.requests) == 1 diff --git a/tests/test_sandbox_wire_contract.py b/tests/test_sandbox_wire_contract.py index 3063615f..15f8ea82 100644 --- a/tests/test_sandbox_wire_contract.py +++ b/tests/test_sandbox_wire_contract.py @@ -14,6 +14,7 @@ ) from hyperbrowser.client.managers.async_manager.sandboxes.sandbox_processes import ( SandboxProcessesApi as AsyncSandboxProcessesApi, + SandboxProcessHandle as AsyncSandboxProcessHandle, ) from hyperbrowser.client.managers.async_manager.sandboxes.sandbox_terminal import ( SandboxTerminalApi as AsyncSandboxTerminalApi, @@ -27,6 +28,7 @@ ) from hyperbrowser.client.managers.sync_manager.sandboxes.sandbox_processes import ( SandboxProcessesApi, + SandboxProcessHandle, ) from hyperbrowser.client.managers.sync_manager.sandboxes.sandbox_terminal import ( SandboxTerminalApi, @@ -539,6 +541,18 @@ def request_json( return MOVE_FILE_PAYLOAD raise AssertionError(f"Unexpected request path: {path}") + def stream_sse( + self, path, params=None, *, method="GET", json_body=None, on_open=None + ): + self.calls.append( + {"path": path, "method": method, "params": params, "json_body": json_body} + ) + yield {"event": "started", "data": PROCESS_SUMMARY_PAYLOAD["process"]} + yield { + "event": "done", + "data": {**PROCESS_RESULT_PAYLOAD["result"], "last_seq": 0}, + } + def request_bytes(self, path, *, method="GET", params=None, headers=None): self.calls.append( { @@ -597,6 +611,12 @@ async def request_json( headers=headers, ) + async def stream_sse(self, path, params=None, *, method="GET", json_body=None): + for event in super().stream_sse( + path, params, method=method, json_body=json_body + ): + yield event + async def request_bytes(self, path, *, method="GET", params=None, headers=None): return super().request_bytes( path, @@ -1473,6 +1493,9 @@ def test_sync_sandbox_runtime_apis_use_expected_wire_keys(): processes.exec(process_input) handle = processes.start(process_input) handle.wait(timeout_ms=250, timeout_sec=3) + SandboxProcessHandle( + transport, SandboxProcessSummary(**PROCESS_SUMMARY_PAYLOAD["process"]) + ).wait(timeout_ms=250, timeout_sec=3) processes.list( status=["running", "exited"], limit=10, @@ -1655,6 +1678,7 @@ def fake_exec(input, **kwargs): "timeout_ms": None, "timeout_sec": None, "run_as": "root", + "max_output_bytes": 64 * 1024 * 1024, }, ) ] @@ -2018,6 +2042,9 @@ async def test_async_sandbox_runtime_apis_use_expected_wire_keys(): await processes.exec(process_input) handle = await processes.start(process_input) await handle.wait(timeout_ms=250, timeout_sec=3) + await AsyncSandboxProcessHandle( + transport, SandboxProcessSummary(**PROCESS_SUMMARY_PAYLOAD["process"]) + ).wait(timeout_ms=250, timeout_sec=3) await processes.list( status=["running", "exited"], limit=10, diff --git a/tests/test_typed_dict_runtime_parity.py b/tests/test_typed_dict_runtime_parity.py index 3f37ee67..d7336fc9 100644 --- a/tests/test_typed_dict_runtime_parity.py +++ b/tests/test_typed_dict_runtime_parity.py @@ -197,6 +197,14 @@ class RecordingRuntimeTransport: def __init__(self): self.calls = [] + def stream_sse(self, path, *, method="GET", json_body=None, on_open=None): + self.calls.append({"path": path, "method": method, "json_body": json_body}) + yield {"event": "started", "data": PROCESS_SUMMARY_PAYLOAD["process"]} + yield { + "event": "done", + "data": {**PROCESS_RESULT_PAYLOAD["result"], "last_seq": 0}, + } + def request_json( self, path, @@ -221,6 +229,10 @@ def request_json( class AsyncRecordingRuntimeTransport(RecordingRuntimeTransport): + async def stream_sse(self, path, **kwargs): + for event in super().stream_sse(path, **kwargs): + yield event + async def request_json( self, path, diff --git a/tests/typecheck/valid_requests.py b/tests/typecheck/valid_requests.py index 6c2a97ca..3b41d48e 100644 --- a/tests/typecheck/valid_requests.py +++ b/tests/typecheck/valid_requests.py @@ -357,3 +357,21 @@ async def valid_async_requests(client: AsyncHyperbrowser) -> None: await client.sessions.create(LegacyCreateSessionParams(use_proxy=True, region="us")) await client.web.fetch(LegacyFetchParams(url="https://example.com")) + + +def valid_sync_process_collection(client: Hyperbrowser) -> None: + sandbox = client.sandboxes.get("sandbox-id") + sandbox.exec("echo hello", max_output_bytes=1024) + process = sandbox.processes.start({"command": "echo hello"}, max_output_bytes=1024) + process.wait(timeout_sec=10) + process.disconnect() + + +async def valid_async_process_collection(client: AsyncHyperbrowser) -> None: + sandbox = await client.sandboxes.get("sandbox-id") + await sandbox.exec("echo hello", max_output_bytes=1024) + process = await sandbox.processes.start( + {"command": "echo hello"}, max_output_bytes=1024 + ) + await process.wait(timeout_sec=10) + await process.disconnect()