diff --git a/sdk/python/README.md b/sdk/python/README.md index 96e7b3f5c..d5ec99683 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -206,14 +206,20 @@ if the bootstrap has been sitting for a while. ## Status -Phase 1 of CORE-58 — the hello-world closed loop: `Sandbox` / -`AsyncSandbox` create/connect/list, `kill`/`pause`/`info` (`pause` and -the paused-sandbox reconnect path are wire-complete but reject with an -unimplemented error until the daemon's CORE-21 lands), `commands.run` -(foreground result + background handle with streamed output, -`wait_for_exit`, `kill`), and whole-file `files` read/write. Deferred: -PTY, `ports`, `wait_for_port`/`wait_for_log`, stdin, filesystem path -verbs (stat/list/mkdir/...), `Template` statics, `events()`, -`set_lifecycle`, the capabilities handshake, and the SDK-side default -idle-reaping policy (design decision 4 — applied once the daemon -enforces the lifecycle knobs). +Phase 2a of CORE-58, on both the sync and async surfaces. Shipped: + +| Surface | Notes | +| --- | --- | +| `Sandbox` create/connect/list, `kill`/`pause`/`info` | pause/resume are live daemon-side (CORE-21); data-plane calls auto-resume a paused sandbox | +| `commands.run` | foreground result + background handle; `stdin=str\|bytes` (write-then-close) or `stdin=True` (keep open); `pty=PtySize(cols, rows)` | +| `CommandHandle` | streamed `output` with transparent offset-resume across stream death (bounded retries → `ConnectionLostError`), `wait_for_exit`, `kill`, `write_stdin`/`close_stdin`/`stdin_status` (offset-idempotent), `resize` | +| `commands.get(id)` | re-attach a handle by execution id (stdin cursor seeded from the daemon) | +| `sandbox.events()` | typed lifecycle events, keepalives filtered; sync reads genuinely block; a mid-stream drop is `ConnectionLostError` | +| `sandbox.set_lifecycle()` | tri-state: omitted (`UNCHANGED`) = unchanged, `None` = restore default (as on `create`), value = replace | +| `arcbox.capabilities()` | daemon handshake (version/protocol/features/nested virt), cached per client | +| `files` | whole-file read/write | + +Deferred to phase 2b (the daemon answers Unimplemented today): +`commands.list()` (ListExecutions), `ports` + `wait_for_port`, +filesystem path verbs (stat/list/mkdir/...), `Template` statics, and the +SDK-side default idle-reaping policy (design decision 4). diff --git a/sdk/python/scripts/gen_sync.py b/sdk/python/scripts/gen_sync.py index befe26db2..06ad6dc79 100644 --- a/sdk/python/scripts/gen_sync.py +++ b/sdk/python/scripts/gen_sync.py @@ -37,6 +37,7 @@ "AsyncCommands": "Commands", "AsyncCommandHandle": "CommandHandle", "AsyncOutputStream": "OutputStream", + "AsyncEventStream": "EventStream", "AsyncFiles": "Files", "AsyncConnectClient": "ConnectClient", "AsyncServerStream": "ServerStream", diff --git a/sdk/python/src/arcbox/__init__.py b/sdk/python/src/arcbox/__init__.py index 566a582ed..77ea0fcd8 100644 --- a/sdk/python/src/arcbox/__init__.py +++ b/sdk/python/src/arcbox/__init__.py @@ -12,22 +12,30 @@ from arcbox._async._client import AsyncConnectClient from arcbox._async.commands import AsyncCommandHandle, AsyncCommands, AsyncOutputStream from arcbox._async.files import AsyncFiles -from arcbox._async.sandbox import AsyncArcBox, AsyncSandbox +from arcbox._async.sandbox import AsyncArcBox, AsyncEventStream, AsyncSandbox from arcbox._connection import Connection from arcbox._sync._client import ConnectClient from arcbox._sync.commands import CommandHandle, Commands, OutputStream from arcbox._sync.files import Files -from arcbox._sync.sandbox import ArcBox, Sandbox +from arcbox._sync.sandbox import ArcBox, EventStream, Sandbox from arcbox._types import ( MAX_FILE_BYTES, + UNCHANGED, + Capabilities, CommandResult, IdlePolicy, + NestedVirtCapability, OutputChannel, OutputChunk, + PtySize, + SandboxEvent, + SandboxEventKind, SandboxInfo, SandboxState, SandboxSummary, SignalName, + StdinStatus, + Unchanged, ) try: @@ -39,27 +47,37 @@ __all__ = [ "MAX_FILE_BYTES", + "UNCHANGED", "ArcBox", "AsyncArcBox", "AsyncCommandHandle", "AsyncCommands", "AsyncConnectClient", + "AsyncEventStream", "AsyncFiles", "AsyncOutputStream", "AsyncSandbox", + "Capabilities", "CommandHandle", "CommandResult", "Commands", "ConnectClient", "Connection", + "EventStream", "Files", "IdlePolicy", + "NestedVirtCapability", "OutputChannel", "OutputChunk", "OutputStream", + "PtySize", "Sandbox", + "SandboxEvent", + "SandboxEventKind", "SandboxInfo", "SandboxState", "SandboxSummary", "SignalName", + "StdinStatus", + "Unchanged", ] diff --git a/sdk/python/src/arcbox/_async/_client.py b/sdk/python/src/arcbox/_async/_client.py index cc04842b6..3830bd80f 100644 --- a/sdk/python/src/arcbox/_async/_client.py +++ b/sdk/python/src/arcbox/_async/_client.py @@ -23,7 +23,7 @@ end_stream_error, unary_error, ) -from arcbox.errors import ArcBoxError, InvalidArgumentError +from arcbox.errors import ArcBoxError, ConnectionLostError, InvalidArgumentError if TYPE_CHECKING: from collections.abc import AsyncIterator, Iterable, Mapping @@ -171,8 +171,9 @@ async def client_stream( if not ended: # Without the terminal frame a truncated body is # indistinguishable from success — never report one as - # completed (WriteFile rides this path). - raise ArcBoxError("the stream ended without an EndStreamResponse") + # completed (WriteFile rides this path). Truncation means the + # connection died mid-body: a stream-death error. + raise ConnectionLostError("the stream ended without an EndStreamResponse") if message is None: raise ArcBoxError("the stream ended without a response message") return message @@ -252,4 +253,6 @@ async def _messages(self) -> AsyncIterator[M]: message = self._response_type() message.ParseFromString(payload) yield message - raise ArcBoxError("the stream ended without an EndStreamResponse") + # Truncation means the connection died mid-body: a stream-death + # error, which resumable consumers treat as retryable. + raise ConnectionLostError("the stream ended without an EndStreamResponse") diff --git a/sdk/python/src/arcbox/_async/commands.py b/sdk/python/src/arcbox/_async/commands.py index 68f620108..9e0a0f6b2 100644 --- a/sdk/python/src/arcbox/_async/commands.py +++ b/sdk/python/src/arcbox/_async/commands.py @@ -6,8 +6,10 @@ import math import time import uuid +from contextlib import aclosing, suppress from typing import TYPE_CHECKING, Literal, overload +import httpx from google.protobuf import empty_pb2 from arcbox._boundary import wrap_errors @@ -16,15 +18,22 @@ SIGNAL_VALUES, CommandResult, OutputChunk, + StdinStatus, command_result_from_execution, ) -from arcbox.errors import InvalidArgumentError, TimeoutError +from arcbox.errors import ( + ArcBoxError, + ConnectionFailedError, + ConnectionLostError, + InvalidArgumentError, + TimeoutError, +) if TYPE_CHECKING: from collections.abc import AsyncGenerator, AsyncIterator, Mapping, Sequence from types import TracebackType - from arcbox._types import OutputChannel, SignalName + from arcbox._types import OutputChannel, PtySize, SignalName from ._client import AsyncConnectClient, AsyncServerStream @@ -34,6 +43,11 @@ #: surfaces as an error instead of an infinite silent wait. _WAIT_SLICE_SECONDS = 30 +#: Consecutive dead re-attach dials tolerated before an output stream +#: surfaces the stream-death error. Delivered output resets the budget, +#: so a long-lived stream survives any number of isolated drops. +_MAX_ATTACH_RETRIES = 3 + def _normalize_cmd(cmd: str | Sequence[str]) -> list[str]: """``str`` is sugar for a shell command; a sequence is argv, executed @@ -89,11 +103,23 @@ class AsyncCommandHandle: decoupled from this object: dropping the handle never kills the process.""" - def __init__(self, client: AsyncConnectClient, sandbox_id: str, command_id: str) -> None: + def __init__( + self, + client: AsyncConnectClient, + sandbox_id: str, + command_id: str, + *, + stdin_offset: int | None = 0, + ) -> None: self._client = client self._sandbox_id = sandbox_id #: Execution id, unique within the sandbox; addressable across clients. self.command_id = command_id + # The next stdin write offset — advanced only on a successful + # write, so a retried write lands at the same offset and the + # daemon's deduplication makes it idempotent. None = unknown (a + # re-attached handle); resynced lazily via GetStdinStatus. + self._stdin_offset = stdin_offset @property def output(self) -> AsyncOutputStream: @@ -101,20 +127,81 @@ def output(self) -> AsyncOutputStream: earliest byte the daemon still retains (8 MiB per channel); replayed buffered output comes first, then live output follows; the stream ends when the process exits (deterministic - termination — never silence). When breaking out early, iterate - via the context-manager form (see :class:`AsyncOutputStream`).""" + termination — never silence). A transport drop mid-stream + re-attaches transparently from the last delivered offsets (see + :class:`arcbox.errors.ConnectionLostError` for the + exhausted-retries case). When breaking out early, iterate via + the context-manager form (see :class:`AsyncOutputStream`).""" return AsyncOutputStream(self._stream_output()) async def _stream_output(self) -> AsyncGenerator[OutputChunk]: - with wrap_errors("commands.output"): - async with self._attach() as stream: - async for event in stream: - kind = event.WhichOneof("event") - if kind == "output": - chunk = event.output - yield OutputChunk(_channel_name(chunk.channel), chunk.data) - elif kind == "exited": - return + # aclosing releases the inner attach loop (and its HTTP stream) + # deterministically on early exit — breaking out of an async-for + # alone would leave it to generator finalization. + async with aclosing(self._attach_events("commands.output")) as events: + async for event in events: + kind = event.WhichOneof("event") + if kind == "output": + chunk = event.output + yield OutputChunk(_channel_name(chunk.channel), chunk.data) + elif kind == "exited": + return + + async def _attach_events(self, operation: str) -> AsyncGenerator[process_pb2.ExecutionEvent]: + """The resumable attach loop shared by :attr:`output` and the + result collection: streams execution events, tracking the byte + offset each channel has delivered. When the transport drops + mid-stream, it re-attaches from those offsets — the daemon + replays nothing already delivered — so the consumer sees one + seamless, gapless stream. Only consecutive dead dials count + against the retry budget (delivered output resets it); once + exhausted, the stream-death + :class:`arcbox.errors.ConnectionLostError` carries the last + transport failure. Daemon-typed stream errors are never + retried.""" + with wrap_errors(operation): + stdout_offset = 0 + stderr_offset = 0 + failures = 0 + while True: + try: + async with self._attach(stdout_offset, stderr_offset) as stream: + async for event in stream: + kind = event.WhichOneof("event") + if kind == "output": + chunk = event.output + after = chunk.offset + len(chunk.data) + if chunk.channel == process_pb2.STDIO_CHANNEL_STDERR: + if after > stderr_offset: + stderr_offset = after + failures = 0 + elif after > stdout_offset: + stdout_offset = after + failures = 0 + yield event + if kind == "exited": + return + # A clean server-side end without an exited frame: + # nothing more is coming (the daemon closed the record). + return + # ConnectionFailedError covers the drop's OTHER wire shape: + # the daemon losing its upstream agent stream ends the HTTP + # stream cleanly with a Connect `unavailable` error frame, + # decoded into ConnectionFailedError (ConnectionLostError — + # raw truncation — is its subclass). Daemon-typed errors + # map to other classes and are never retried. + except (httpx.HTTPError, ConnectionFailedError) as exc: + failures += 1 + if failures > _MAX_ATTACH_RETRIES: + raise ConnectionLostError( + "the output stream died and could not be re-attached " + "within the retry budget", + context={ + "command_id": self.command_id, + "retries": str(_MAX_ATTACH_RETRIES), + }, + operation=operation, + ) from exc async def wait_for_exit(self, timeout: float | None = None) -> CommandResult: """Wait until the command exits and return its result (server-side @@ -180,14 +267,97 @@ async def kill(self, signal: SignalName = "SIGTERM") -> None: empty_pb2.Empty, ) - def _attach(self) -> AsyncServerStream[process_pb2.ExecutionEvent]: + async def write_stdin(self, data: str | bytes) -> None: + """Write bytes (or a UTF-8 string) to the command's stdin. + Requires a run started with ``stdin=True`` or a PTY. + + Writes are offset-idempotent: the handle tracks the stdin cursor + and advances it only on success, so retrying a failed or lost + write *with the same data* is safe — the daemon deduplicates + bytes below its accepted count and never double-feeds the + process. Issue writes sequentially; the handle tracks a single + cursor.""" + payload = data.encode() if isinstance(data, str) else data + with wrap_errors("commands.write_stdin"): + offset = await self._stdin_cursor() + status = await self._client.unary( + _PROCESS + "WriteStdin", + process_pb2.WriteStdinRequest( + sandbox_id=self._sandbox_id, + execution_id=self.command_id, + offset=offset, + data=payload, + eof=False, + ), + process_pb2.StdinStatus, + ) + self._stdin_offset = int(status.bytes_written) + + async def close_stdin(self) -> None: + """Close the command's stdin (EOF). Rejected for PTY commands — + write Ctrl-D (``"\\x04"``) instead.""" + with wrap_errors("commands.close_stdin"): + offset = await self._stdin_cursor() + await self._client.unary( + _PROCESS + "WriteStdin", + process_pb2.WriteStdinRequest( + sandbox_id=self._sandbox_id, + execution_id=self.command_id, + offset=offset, + data=b"", + eof=True, + ), + process_pb2.StdinStatus, + ) + + async def stdin_status(self) -> StdinStatus: + """The daemon's stdin acceptance state — the recovery point after + a lost write response. Also resyncs the handle's write cursor.""" + with wrap_errors("commands.stdin_status"): + status = await self._fetch_stdin_status() + return StdinStatus(bytes_written=int(status.bytes_written), closed=status.closed) + + async def resize(self, cols: int, rows: int) -> None: + """Resize a PTY command's terminal.""" + with wrap_errors("commands.resize"): + await self._client.unary( + _PROCESS + "ResizeExecutionTty", + process_pb2.ResizeExecutionTtyRequest( + sandbox_id=self._sandbox_id, + execution_id=self.command_id, + size=process_pb2.TerminalSize(width=cols, height=rows), + ), + empty_pb2.Empty, + ) + + async def _stdin_cursor(self) -> int: + """The tracked stdin cursor, resynced from the daemon when unknown.""" + offset = self._stdin_offset + if offset is None: + offset = int((await self._fetch_stdin_status()).bytes_written) + return offset + + async def _fetch_stdin_status(self) -> process_pb2.StdinStatus: + status = await self._client.unary( + _PROCESS + "GetStdinStatus", + process_pb2.GetStdinStatusRequest( + sandbox_id=self._sandbox_id, execution_id=self.command_id + ), + process_pb2.StdinStatus, + ) + self._stdin_offset = int(status.bytes_written) + return status + + def _attach( + self, stdout_offset: int, stderr_offset: int + ) -> AsyncServerStream[process_pb2.ExecutionEvent]: return self._client.stream( _PROCESS + "AttachExecution", process_pb2.AttachExecutionRequest( sandbox_id=self._sandbox_id, execution_id=self.command_id, - stdout_offset=0, - stderr_offset=0, + stdout_offset=stdout_offset, + stderr_offset=stderr_offset, ), process_pb2.ExecutionEvent, ) @@ -203,8 +373,8 @@ async def _collect_result(self, execution: process_pb2.Execution) -> CommandResu next_stdout = 0 next_stderr = 0 truncated = False - async with self._attach() as stream: - async for event in stream: + async with aclosing(self._attach_events("commands.wait_for_exit")) as events: + async for event in events: kind = event.WhichOneof("event") if kind == "output": chunk = event.output @@ -245,6 +415,8 @@ async def run( timeout: float | None = None, check: bool = False, background: Literal[False] = False, + pty: PtySize | None = None, + stdin: str | bytes | bool | None = None, ) -> CommandResult: ... @overload @@ -257,6 +429,8 @@ async def run( user: str | None = None, timeout: float | None = None, background: Literal[True], + pty: PtySize | None = None, + stdin: str | bytes | bool | None = None, ) -> AsyncCommandHandle: ... async def run( @@ -269,6 +443,8 @@ async def run( timeout: float | None = None, check: bool = False, background: bool = False, + pty: PtySize | None = None, + stdin: str | bytes | bool | None = None, ) -> CommandResult | AsyncCommandHandle: """Run a command. Foreground (default): returns the complete :class:`CommandResult` once the process exits — non-zero exit is @@ -278,19 +454,75 @@ async def run( ``timeout`` kills the whole process group after that many seconds; expiry surfaces as signal death in the result - (exit-as-data), not as a raised error.""" + (exit-as-data), not as a raised error. + + ``pty`` allocates a pseudo-terminal of that size: output then + arrives merged on the ``"pty"`` channel and stdin stays open — + end input by writing Ctrl-D (``"\\x04"``) via ``write_stdin``. + + ``stdin``: a string (UTF-8) or bytes is written and then closed + before the run resolves (subprocess semantics); ``True`` keeps + stdin open for manual ``write_stdin``/``close_stdin`` — + background runs only. ``None``/``False``: the process starts + with stdin already at EOF.""" if background and check: raise InvalidArgumentError( "check=True applies to foreground runs; call " ".wait_for_exit() and .expect() on the handle instead", operation="commands.run", ) - handle = await self._start(cmd, cwd=cwd, env=env, user=user, timeout=timeout) + if stdin is True and not background: + raise InvalidArgumentError( + "stdin=True keeps stdin open for the handle and requires " + "background=True; pass str or bytes to feed a foreground run", + operation="commands.run", + ) + stdin_data = stdin.encode() if isinstance(stdin, str) else stdin + if isinstance(stdin_data, bytes) and pty is not None: + raise InvalidArgumentError( + "a PTY's stdin cannot be closed after a one-shot write; use " + 'background=True with write_stdin, ending input with Ctrl-D ("\\x04")', + operation="commands.run", + ) + handle = await self._start( + cmd, + cwd=cwd, + env=env, + user=user, + timeout=timeout, + pty=pty, + stdin_open=stdin is True or isinstance(stdin_data, bytes), + ) + if isinstance(stdin_data, bytes): + await self._feed_stdin(handle, stdin_data) if background: return handle result = await handle.wait_for_exit() return result.expect() if check else result + async def get(self, command_id: str) -> AsyncCommandHandle: + """Re-attach to an execution by id — from another process, or + after losing the handle. Verifies the execution exists (an + unknown id is a typed not-found error) and seeds the handle's + stdin cursor from the daemon's accepted count so later writes + resume without a gap.""" + with wrap_errors("commands.get"): + execution = await self._client.unary( + _PROCESS + "WaitExecution", + process_pb2.WaitExecutionRequest( + sandbox_id=self._sandbox_id, + execution_id=command_id, + timeout_seconds=0, + ), + process_pb2.Execution, + ) + stdin_offset = ( + int(execution.stdin.bytes_written) if execution.HasField("stdin") else None + ) + return AsyncCommandHandle( + self._client, self._sandbox_id, execution.id, stdin_offset=stdin_offset + ) + async def _start( self, cmd: str | Sequence[str], @@ -299,24 +531,64 @@ async def _start( env: Mapping[str, str] | None, user: str | None, timeout: float | None, + pty: PtySize | None, + stdin_open: bool, ) -> AsyncCommandHandle: with wrap_errors("commands.run"): # The execution id is minted client-side: a lost response # leaves an addressable execution, and retries are idempotent # by contract. execution_id = str(uuid.uuid4()) + request = process_pb2.StartExecutionRequest( + sandbox_id=self._sandbox_id, + execution_id=execution_id, + cmd=_normalize_cmd(cmd), + env=dict(env) if env else {}, + working_dir=cwd or "", + user=user or "", + timeout_seconds=0 if timeout is None else math.ceil(timeout), + # A PTY inherently keeps stdin open (EOF is not + # expressible on a terminal); otherwise stdin stays open + # exactly when the caller feeds or drives it. + stdin=pty is not None or stdin_open, + tty=pty is not None, + ) + if pty is not None: + request.tty_size.width = pty.cols + request.tty_size.height = pty.rows execution = await self._client.unary( - _PROCESS + "StartExecution", - process_pb2.StartExecutionRequest( - sandbox_id=self._sandbox_id, - execution_id=execution_id, - cmd=_normalize_cmd(cmd), - env=dict(env) if env else {}, - working_dir=cwd or "", - user=user or "", - timeout_seconds=0 if timeout is None else math.ceil(timeout), - stdin=False, - ), - process_pb2.Execution, + _PROCESS + "StartExecution", request, process_pb2.Execution ) return AsyncCommandHandle(self._client, self._sandbox_id, execution.id) + + async def _feed_stdin(self, handle: AsyncCommandHandle, data: bytes) -> None: + """Write-then-close the one-shot stdin payload. A process is free + to exit without consuming its stdin (`subprocess` semantics): + when the feed fails but the execution has already exited, the + exit result is the truth and the failed feed is noise — the + write merely raced the exit. Any other failure is real and + surfaces.""" + try: + await handle.write_stdin(data) + await handle.close_stdin() + except ArcBoxError: + state: process_pb2.Execution | None = None + # If the poll itself fails, the original feed error stands. + with suppress(Exception): + state = await self._client.unary( + _PROCESS + "WaitExecution", + process_pb2.WaitExecutionRequest( + sandbox_id=self._sandbox_id, + execution_id=handle.command_id, + timeout_seconds=0, + ), + process_pb2.Execution, + ) + if state is None or state.state != process_pb2.EXECUTION_STATE_EXITED: + # The caller gets no handle out of a thrown run(), so a + # still-running process (cat waiting on input) would + # keep the sandbox RUNNING with no way to reach it. + # Best-effort kill; the feed error is the one to surface. + with suppress(Exception): + await handle.kill("SIGKILL") + raise diff --git a/sdk/python/src/arcbox/_async/sandbox.py b/sdk/python/src/arcbox/_async/sandbox.py index e27b29830..ed6384bf2 100644 --- a/sdk/python/src/arcbox/_async/sandbox.py +++ b/sdk/python/src/arcbox/_async/sandbox.py @@ -19,8 +19,14 @@ from arcbox._boundary import wrap_errors from arcbox._gen import sandbox_pb2 from arcbox._types import ( + UNCHANGED, + Capabilities, + SandboxEvent, SandboxInfo, SandboxSummary, + Unchanged, + capabilities_from_proto, + sandbox_event_from_proto, sandbox_info_from_proto, sandbox_state_from_proto, sandbox_state_to_proto, @@ -28,6 +34,8 @@ ) from arcbox.errors import ( ArcBoxError, + ConnectionFailedError, + ConnectionLostError, InvalidArgumentError, NotFoundError, RequestTimeoutError, @@ -40,7 +48,7 @@ from .files import AsyncFiles if TYPE_CHECKING: - from collections.abc import AsyncIterator, Mapping, Sequence + from collections.abc import AsyncGenerator, AsyncIterator, Mapping, Sequence from types import TracebackType from arcbox._connection import Connection @@ -63,11 +71,10 @@ sandbox_pb2.SANDBOX_EVENT_KIND_REMOVED, ) -#: How often ``connect`` re-inspects a PAUSING sandbox. -#: ``SANDBOX_EVENT_KIND_PAUSED`` names this edge in the proto, but the -#: daemon does not emit it yet (Pause/Resume are CORE-21 stubs), so the -#: checkpoint is polled out instead. Once it is emitted, this poll -#: should become an event wait. +#: How often ``connect`` re-inspects a PAUSING sandbox. The daemon +#: emits ``SANDBOX_EVENT_KIND_PAUSED`` on this edge (CORE-21), but the +#: settle poll predates it and remains the simple, robust route — a +#: poll-to-event-wait conversion is a candidate cleanup, not a bug. _PAUSE_SETTLE_POLL_SECONDS = 0.5 #: Default overall deadline for ``connect`` in seconds — generous @@ -110,11 +117,31 @@ class AsyncArcBox: def __init__(self, connection: Connection | None = None) -> None: self._client = AsyncConnectClient(connection) + self._capabilities: Capabilities | None = None async def aclose(self) -> None: """Close the SDK-owned HTTP client (no-op for an injected one).""" await self._client.aclose() + async def capabilities(self) -> Capabilities: + """What the daemon can do: version, sandbox protocol level, + feature flags, and whether nested virtualization is available. + Answered host-side (works before any sandbox exists) and cached + for the life of this client — a failed fetch is not cached, so + the next call retries. The SDK does not gate on it: the daemon + fails fast on its own (a ``CapabilityError`` from ``create``); + this is the inspectable version of the same answer.""" + if self._capabilities is None: + with wrap_errors("arcbox.capabilities"): + self._capabilities = capabilities_from_proto( + await self._client.unary( + _SANDBOX + "GetCapabilities", + sandbox_pb2.GetCapabilitiesRequest(), + sandbox_pb2.GetCapabilitiesResponse, + ) + ) + return self._capabilities + async def __aenter__(self) -> AsyncArcBox: return self @@ -435,6 +462,37 @@ def check(state: int, error: str) -> bool: ) +class AsyncEventStream: + """A sandbox's lifecycle events, iterable one typed event at a time. + + The iterator ends when the daemon ends the stream. When exiting + early (``break``), iterate inside the context-manager form — + ``async with sandbox.events() as stream`` in the async flavor, + ``with`` in the sync one — so the subscription closes at the break + instead of whenever the generator finalizer runs.""" + + def __init__(self, events: AsyncGenerator[SandboxEvent]) -> None: + self._events = events + + def __aiter__(self) -> AsyncIterator[SandboxEvent]: + return self._events + + async def aclose(self) -> None: + """Cancel the subscription without consuming the rest.""" + await self._events.aclose() + + async def __aenter__(self) -> AsyncEventStream: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + await self.aclose() + + class AsyncSandbox: """A handle to one sandbox.""" @@ -553,16 +611,85 @@ async def kill(self) -> None: empty_pb2.Empty, ) + async def set_lifecycle( + self, + *, + ttl: float | Unchanged | None = UNCHANGED, + idle_timeout: float | Unchanged | None = UNCHANGED, + on_idle: IdlePolicy | Unchanged | None = UNCHANGED, + ) -> None: + """Replace lifecycle deadlines. Each knob is tri-state: omitted + (the :data:`arcbox.UNCHANGED` default) leaves it as it is; + ``None`` restores the daemon default (no TTL / no idle detection + / the default idle action) — the same meaning ``None`` has on + ``create``; a value replaces it. ``ttl`` re-arms the hard cap + this many seconds from NOW — calling repeatedly keeps a busy + sandbox alive; ``idle_timeout`` re-arms a live idle timer. Works + in any non-terminal state, including paused.""" + with wrap_errors("sandbox.set_lifecycle"): + request = sandbox_pb2.SetLifecycleRequest(id=self.id) + if not isinstance(ttl, Unchanged): + request.ttl_seconds = _seconds_to_wire(ttl) + if not isinstance(idle_timeout, Unchanged): + request.idle_timeout_seconds = _seconds_to_wire(idle_timeout) + if not isinstance(on_idle, Unchanged): + request.on_idle = ( + sandbox_pb2.IDLE_ACTION_UNSPECIFIED + if on_idle is None + else sandbox_pb2.IDLE_ACTION_KILL + if on_idle == "kill" + else sandbox_pb2.IDLE_ACTION_PAUSE + ) + await self._client.unary(_SANDBOX + "SetLifecycle", request, empty_pb2.Empty) + + def events(self) -> AsyncEventStream: + """Subscribe to this sandbox's lifecycle events, yielded as + typed :class:`SandboxEvent` values (keepalive frames are + filtered out). The iterator ends when the daemon ends the + stream; closing the stream (or its context) cancels the + subscription. A transport drop mid-stream is surfaced as + :class:`arcbox.errors.ConnectionLostError` — re-subscribing is + the caller's decision, since missed events cannot be replayed.""" + return AsyncEventStream(self._stream_events()) + + async def _stream_events(self) -> AsyncGenerator[SandboxEvent]: + with wrap_errors("sandbox.events"): + entered = False + try: + async with self._client.stream( + _SANDBOX + "Events", + sandbox_pb2.SandboxEventsRequest(sandbox_id=self.id), + sandbox_pb2.WatchEventsResponse, + ) as stream: + entered = True + async for frame in stream: + if frame.WhichOneof("payload") == "event": + yield sandbox_event_from_proto(frame.event) + # ConnectionFailedError covers the drop's OTHER wire shape: + # the daemon losing its upstream event source ends the HTTP + # stream cleanly with a Connect `unavailable` error frame, + # decoded into ConnectionFailedError. Daemon-typed errors map + # to other classes and keep them. + except (httpx.HTTPError, ConnectionFailedError) as exc: + if not entered or isinstance(exc, ConnectionLostError): + # Before entry: a dial failure — the daemon was never + # reached; wrap_errors maps it to + # ConnectionFailedError. ConnectionLostError is + # already the stream-death error. + raise + raise ConnectionLostError( + "the event stream died", + context={"id": self.id}, + operation="sandbox.events", + ) from exc + async def pause(self) -> None: """Checkpoint the sandbox to disk under the same id and release its runtime resources. Resume happens on the next ``connect`` (or transparently, daemon-side, on the next data-plane call). Trades RAM for disk: a paused sandbox keeps paying - ``storage_bytes``. - - Requires daemon-side CORE-21: the current local daemon serves - Pause/Resume as contract-only stubs, so this raises an - unimplemented :class:`ArcBoxError` until that lands.""" + ``storage_bytes``. Requires a quiescent sandbox (READY — no + running command).""" with wrap_errors("sandbox.pause"): # No per-request deadline: checkpointing takes as long as it # takes. diff --git a/sdk/python/src/arcbox/_sync/_client.py b/sdk/python/src/arcbox/_sync/_client.py index 5731937f7..74b5a19e4 100644 --- a/sdk/python/src/arcbox/_sync/_client.py +++ b/sdk/python/src/arcbox/_sync/_client.py @@ -24,7 +24,7 @@ end_stream_error, unary_error, ) -from arcbox.errors import ArcBoxError, InvalidArgumentError +from arcbox.errors import ArcBoxError, ConnectionLostError, InvalidArgumentError if TYPE_CHECKING: from collections.abc import Iterable, Iterator, Mapping @@ -170,8 +170,9 @@ def client_stream(self, path: str, requests: Iterable[Message], response_type: t if not ended: # Without the terminal frame a truncated body is # indistinguishable from success — never report one as - # completed (WriteFile rides this path). - raise ArcBoxError("the stream ended without an EndStreamResponse") + # completed (WriteFile rides this path). Truncation means the + # connection died mid-body: a stream-death error. + raise ConnectionLostError("the stream ended without an EndStreamResponse") if message is None: raise ArcBoxError("the stream ended without a response message") return message @@ -251,4 +252,6 @@ def _messages(self) -> Iterator[M]: message = self._response_type() message.ParseFromString(payload) yield message - raise ArcBoxError("the stream ended without an EndStreamResponse") + # Truncation means the connection died mid-body: a stream-death + # error, which resumable consumers treat as retryable. + raise ConnectionLostError("the stream ended without an EndStreamResponse") diff --git a/sdk/python/src/arcbox/_sync/commands.py b/sdk/python/src/arcbox/_sync/commands.py index 5020cd5cc..180ddf24b 100644 --- a/sdk/python/src/arcbox/_sync/commands.py +++ b/sdk/python/src/arcbox/_sync/commands.py @@ -6,8 +6,10 @@ import math import time import uuid +from contextlib import closing, suppress from typing import TYPE_CHECKING, Literal, overload +import httpx from google.protobuf import empty_pb2 from arcbox._boundary import wrap_errors @@ -16,15 +18,22 @@ SIGNAL_VALUES, CommandResult, OutputChunk, + StdinStatus, command_result_from_execution, ) -from arcbox.errors import InvalidArgumentError, TimeoutError +from arcbox.errors import ( + ArcBoxError, + ConnectionFailedError, + ConnectionLostError, + InvalidArgumentError, + TimeoutError, +) if TYPE_CHECKING: from collections.abc import Generator, Iterator, Mapping, Sequence from types import TracebackType - from arcbox._types import OutputChannel, SignalName + from arcbox._types import OutputChannel, PtySize, SignalName from ._client import ConnectClient, ServerStream @@ -34,6 +43,11 @@ #: surfaces as an error instead of an infinite silent wait. _WAIT_SLICE_SECONDS = 30 +#: Consecutive dead re-attach dials tolerated before an output stream +#: surfaces the stream-death error. Delivered output resets the budget, +#: so a long-lived stream survives any number of isolated drops. +_MAX_ATTACH_RETRIES = 3 + def _normalize_cmd(cmd: str | Sequence[str]) -> list[str]: """``str`` is sugar for a shell command; a sequence is argv, executed @@ -89,11 +103,23 @@ class CommandHandle: decoupled from this object: dropping the handle never kills the process.""" - def __init__(self, client: ConnectClient, sandbox_id: str, command_id: str) -> None: + def __init__( + self, + client: ConnectClient, + sandbox_id: str, + command_id: str, + *, + stdin_offset: int | None = 0, + ) -> None: self._client = client self._sandbox_id = sandbox_id #: Execution id, unique within the sandbox; addressable across clients. self.command_id = command_id + # The next stdin write offset — advanced only on a successful + # write, so a retried write lands at the same offset and the + # daemon's deduplication makes it idempotent. None = unknown (a + # re-attached handle); resynced lazily via GetStdinStatus. + self._stdin_offset = stdin_offset @property def output(self) -> OutputStream: @@ -101,13 +127,19 @@ def output(self) -> OutputStream: earliest byte the daemon still retains (8 MiB per channel); replayed buffered output comes first, then live output follows; the stream ends when the process exits (deterministic - termination — never silence). When breaking out early, iterate - via the context-manager form (see :class:`AsyncOutputStream`).""" + termination — never silence). A transport drop mid-stream + re-attaches transparently from the last delivered offsets (see + :class:`arcbox.errors.ConnectionLostError` for the + exhausted-retries case). When breaking out early, iterate via + the context-manager form (see :class:`AsyncOutputStream`).""" return OutputStream(self._stream_output()) def _stream_output(self) -> Generator[OutputChunk]: - with wrap_errors("commands.output"), self._attach() as stream: - for event in stream: + # aclosing releases the inner attach loop (and its HTTP stream) + # deterministically on early exit — breaking out of an async-for + # alone would leave it to generator finalization. + with closing(self._attach_events("commands.output")) as events: + for event in events: kind = event.WhichOneof("event") if kind == "output": chunk = event.output @@ -115,6 +147,62 @@ def _stream_output(self) -> Generator[OutputChunk]: elif kind == "exited": return + def _attach_events(self, operation: str) -> Generator[process_pb2.ExecutionEvent]: + """The resumable attach loop shared by :attr:`output` and the + result collection: streams execution events, tracking the byte + offset each channel has delivered. When the transport drops + mid-stream, it re-attaches from those offsets — the daemon + replays nothing already delivered — so the consumer sees one + seamless, gapless stream. Only consecutive dead dials count + against the retry budget (delivered output resets it); once + exhausted, the stream-death + :class:`arcbox.errors.ConnectionLostError` carries the last + transport failure. Daemon-typed stream errors are never + retried.""" + with wrap_errors(operation): + stdout_offset = 0 + stderr_offset = 0 + failures = 0 + while True: + try: + with self._attach(stdout_offset, stderr_offset) as stream: + for event in stream: + kind = event.WhichOneof("event") + if kind == "output": + chunk = event.output + after = chunk.offset + len(chunk.data) + if chunk.channel == process_pb2.STDIO_CHANNEL_STDERR: + if after > stderr_offset: + stderr_offset = after + failures = 0 + elif after > stdout_offset: + stdout_offset = after + failures = 0 + yield event + if kind == "exited": + return + # A clean server-side end without an exited frame: + # nothing more is coming (the daemon closed the record). + return + # ConnectionFailedError covers the drop's OTHER wire shape: + # the daemon losing its upstream agent stream ends the HTTP + # stream cleanly with a Connect `unavailable` error frame, + # decoded into ConnectionFailedError (ConnectionLostError — + # raw truncation — is its subclass). Daemon-typed errors + # map to other classes and are never retried. + except (httpx.HTTPError, ConnectionFailedError) as exc: + failures += 1 + if failures > _MAX_ATTACH_RETRIES: + raise ConnectionLostError( + "the output stream died and could not be re-attached " + "within the retry budget", + context={ + "command_id": self.command_id, + "retries": str(_MAX_ATTACH_RETRIES), + }, + operation=operation, + ) from exc + def wait_for_exit(self, timeout: float | None = None) -> CommandResult: """Wait until the command exits and return its result (server-side long-poll; no client-side spinning). ``timeout`` bounds the WAIT, @@ -179,14 +267,97 @@ def kill(self, signal: SignalName = "SIGTERM") -> None: empty_pb2.Empty, ) - def _attach(self) -> ServerStream[process_pb2.ExecutionEvent]: + def write_stdin(self, data: str | bytes) -> None: + """Write bytes (or a UTF-8 string) to the command's stdin. + Requires a run started with ``stdin=True`` or a PTY. + + Writes are offset-idempotent: the handle tracks the stdin cursor + and advances it only on success, so retrying a failed or lost + write *with the same data* is safe — the daemon deduplicates + bytes below its accepted count and never double-feeds the + process. Issue writes sequentially; the handle tracks a single + cursor.""" + payload = data.encode() if isinstance(data, str) else data + with wrap_errors("commands.write_stdin"): + offset = self._stdin_cursor() + status = self._client.unary( + _PROCESS + "WriteStdin", + process_pb2.WriteStdinRequest( + sandbox_id=self._sandbox_id, + execution_id=self.command_id, + offset=offset, + data=payload, + eof=False, + ), + process_pb2.StdinStatus, + ) + self._stdin_offset = int(status.bytes_written) + + def close_stdin(self) -> None: + """Close the command's stdin (EOF). Rejected for PTY commands — + write Ctrl-D (``"\\x04"``) instead.""" + with wrap_errors("commands.close_stdin"): + offset = self._stdin_cursor() + self._client.unary( + _PROCESS + "WriteStdin", + process_pb2.WriteStdinRequest( + sandbox_id=self._sandbox_id, + execution_id=self.command_id, + offset=offset, + data=b"", + eof=True, + ), + process_pb2.StdinStatus, + ) + + def stdin_status(self) -> StdinStatus: + """The daemon's stdin acceptance state — the recovery point after + a lost write response. Also resyncs the handle's write cursor.""" + with wrap_errors("commands.stdin_status"): + status = self._fetch_stdin_status() + return StdinStatus(bytes_written=int(status.bytes_written), closed=status.closed) + + def resize(self, cols: int, rows: int) -> None: + """Resize a PTY command's terminal.""" + with wrap_errors("commands.resize"): + self._client.unary( + _PROCESS + "ResizeExecutionTty", + process_pb2.ResizeExecutionTtyRequest( + sandbox_id=self._sandbox_id, + execution_id=self.command_id, + size=process_pb2.TerminalSize(width=cols, height=rows), + ), + empty_pb2.Empty, + ) + + def _stdin_cursor(self) -> int: + """The tracked stdin cursor, resynced from the daemon when unknown.""" + offset = self._stdin_offset + if offset is None: + offset = int((self._fetch_stdin_status()).bytes_written) + return offset + + def _fetch_stdin_status(self) -> process_pb2.StdinStatus: + status = self._client.unary( + _PROCESS + "GetStdinStatus", + process_pb2.GetStdinStatusRequest( + sandbox_id=self._sandbox_id, execution_id=self.command_id + ), + process_pb2.StdinStatus, + ) + self._stdin_offset = int(status.bytes_written) + return status + + def _attach( + self, stdout_offset: int, stderr_offset: int + ) -> ServerStream[process_pb2.ExecutionEvent]: return self._client.stream( _PROCESS + "AttachExecution", process_pb2.AttachExecutionRequest( sandbox_id=self._sandbox_id, execution_id=self.command_id, - stdout_offset=0, - stderr_offset=0, + stdout_offset=stdout_offset, + stderr_offset=stderr_offset, ), process_pb2.ExecutionEvent, ) @@ -202,8 +373,8 @@ def _collect_result(self, execution: process_pb2.Execution) -> CommandResult: next_stdout = 0 next_stderr = 0 truncated = False - with self._attach() as stream: - for event in stream: + with closing(self._attach_events("commands.wait_for_exit")) as events: + for event in events: kind = event.WhichOneof("event") if kind == "output": chunk = event.output @@ -244,6 +415,8 @@ def run( timeout: float | None = None, check: bool = False, background: Literal[False] = False, + pty: PtySize | None = None, + stdin: str | bytes | bool | None = None, ) -> CommandResult: ... @overload @@ -256,6 +429,8 @@ def run( user: str | None = None, timeout: float | None = None, background: Literal[True], + pty: PtySize | None = None, + stdin: str | bytes | bool | None = None, ) -> CommandHandle: ... def run( @@ -268,6 +443,8 @@ def run( timeout: float | None = None, check: bool = False, background: bool = False, + pty: PtySize | None = None, + stdin: str | bytes | bool | None = None, ) -> CommandResult | CommandHandle: """Run a command. Foreground (default): returns the complete :class:`CommandResult` once the process exits — non-zero exit is @@ -277,19 +454,75 @@ def run( ``timeout`` kills the whole process group after that many seconds; expiry surfaces as signal death in the result - (exit-as-data), not as a raised error.""" + (exit-as-data), not as a raised error. + + ``pty`` allocates a pseudo-terminal of that size: output then + arrives merged on the ``"pty"`` channel and stdin stays open — + end input by writing Ctrl-D (``"\\x04"``) via ``write_stdin``. + + ``stdin``: a string (UTF-8) or bytes is written and then closed + before the run resolves (subprocess semantics); ``True`` keeps + stdin open for manual ``write_stdin``/``close_stdin`` — + background runs only. ``None``/``False``: the process starts + with stdin already at EOF.""" if background and check: raise InvalidArgumentError( "check=True applies to foreground runs; call " ".wait_for_exit() and .expect() on the handle instead", operation="commands.run", ) - handle = self._start(cmd, cwd=cwd, env=env, user=user, timeout=timeout) + if stdin is True and not background: + raise InvalidArgumentError( + "stdin=True keeps stdin open for the handle and requires " + "background=True; pass str or bytes to feed a foreground run", + operation="commands.run", + ) + stdin_data = stdin.encode() if isinstance(stdin, str) else stdin + if isinstance(stdin_data, bytes) and pty is not None: + raise InvalidArgumentError( + "a PTY's stdin cannot be closed after a one-shot write; use " + 'background=True with write_stdin, ending input with Ctrl-D ("\\x04")', + operation="commands.run", + ) + handle = self._start( + cmd, + cwd=cwd, + env=env, + user=user, + timeout=timeout, + pty=pty, + stdin_open=stdin is True or isinstance(stdin_data, bytes), + ) + if isinstance(stdin_data, bytes): + self._feed_stdin(handle, stdin_data) if background: return handle result = handle.wait_for_exit() return result.expect() if check else result + def get(self, command_id: str) -> CommandHandle: + """Re-attach to an execution by id — from another process, or + after losing the handle. Verifies the execution exists (an + unknown id is a typed not-found error) and seeds the handle's + stdin cursor from the daemon's accepted count so later writes + resume without a gap.""" + with wrap_errors("commands.get"): + execution = self._client.unary( + _PROCESS + "WaitExecution", + process_pb2.WaitExecutionRequest( + sandbox_id=self._sandbox_id, + execution_id=command_id, + timeout_seconds=0, + ), + process_pb2.Execution, + ) + stdin_offset = ( + int(execution.stdin.bytes_written) if execution.HasField("stdin") else None + ) + return CommandHandle( + self._client, self._sandbox_id, execution.id, stdin_offset=stdin_offset + ) + def _start( self, cmd: str | Sequence[str], @@ -298,24 +531,64 @@ def _start( env: Mapping[str, str] | None, user: str | None, timeout: float | None, + pty: PtySize | None, + stdin_open: bool, ) -> CommandHandle: with wrap_errors("commands.run"): # The execution id is minted client-side: a lost response # leaves an addressable execution, and retries are idempotent # by contract. execution_id = str(uuid.uuid4()) + request = process_pb2.StartExecutionRequest( + sandbox_id=self._sandbox_id, + execution_id=execution_id, + cmd=_normalize_cmd(cmd), + env=dict(env) if env else {}, + working_dir=cwd or "", + user=user or "", + timeout_seconds=0 if timeout is None else math.ceil(timeout), + # A PTY inherently keeps stdin open (EOF is not + # expressible on a terminal); otherwise stdin stays open + # exactly when the caller feeds or drives it. + stdin=pty is not None or stdin_open, + tty=pty is not None, + ) + if pty is not None: + request.tty_size.width = pty.cols + request.tty_size.height = pty.rows execution = self._client.unary( - _PROCESS + "StartExecution", - process_pb2.StartExecutionRequest( - sandbox_id=self._sandbox_id, - execution_id=execution_id, - cmd=_normalize_cmd(cmd), - env=dict(env) if env else {}, - working_dir=cwd or "", - user=user or "", - timeout_seconds=0 if timeout is None else math.ceil(timeout), - stdin=False, - ), - process_pb2.Execution, + _PROCESS + "StartExecution", request, process_pb2.Execution ) return CommandHandle(self._client, self._sandbox_id, execution.id) + + def _feed_stdin(self, handle: CommandHandle, data: bytes) -> None: + """Write-then-close the one-shot stdin payload. A process is free + to exit without consuming its stdin (`subprocess` semantics): + when the feed fails but the execution has already exited, the + exit result is the truth and the failed feed is noise — the + write merely raced the exit. Any other failure is real and + surfaces.""" + try: + handle.write_stdin(data) + handle.close_stdin() + except ArcBoxError: + state: process_pb2.Execution | None = None + # If the poll itself fails, the original feed error stands. + with suppress(Exception): + state = self._client.unary( + _PROCESS + "WaitExecution", + process_pb2.WaitExecutionRequest( + sandbox_id=self._sandbox_id, + execution_id=handle.command_id, + timeout_seconds=0, + ), + process_pb2.Execution, + ) + if state is None or state.state != process_pb2.EXECUTION_STATE_EXITED: + # The caller gets no handle out of a thrown run(), so a + # still-running process (cat waiting on input) would + # keep the sandbox RUNNING with no way to reach it. + # Best-effort kill; the feed error is the one to surface. + with suppress(Exception): + handle.kill("SIGKILL") + raise diff --git a/sdk/python/src/arcbox/_sync/sandbox.py b/sdk/python/src/arcbox/_sync/sandbox.py index 7358c9d8d..bbc99d5cf 100644 --- a/sdk/python/src/arcbox/_sync/sandbox.py +++ b/sdk/python/src/arcbox/_sync/sandbox.py @@ -19,8 +19,14 @@ from arcbox._boundary import wrap_errors from arcbox._gen import sandbox_pb2 from arcbox._types import ( + UNCHANGED, + Capabilities, + SandboxEvent, SandboxInfo, SandboxSummary, + Unchanged, + capabilities_from_proto, + sandbox_event_from_proto, sandbox_info_from_proto, sandbox_state_from_proto, sandbox_state_to_proto, @@ -28,6 +34,8 @@ ) from arcbox.errors import ( ArcBoxError, + ConnectionFailedError, + ConnectionLostError, InvalidArgumentError, NotFoundError, RequestTimeoutError, @@ -40,7 +48,7 @@ from .files import Files if TYPE_CHECKING: - from collections.abc import Iterator, Mapping, Sequence + from collections.abc import Generator, Iterator, Mapping, Sequence from types import TracebackType from arcbox._connection import Connection @@ -63,11 +71,10 @@ sandbox_pb2.SANDBOX_EVENT_KIND_REMOVED, ) -#: How often ``connect`` re-inspects a PAUSING sandbox. -#: ``SANDBOX_EVENT_KIND_PAUSED`` names this edge in the proto, but the -#: daemon does not emit it yet (Pause/Resume are CORE-21 stubs), so the -#: checkpoint is polled out instead. Once it is emitted, this poll -#: should become an event wait. +#: How often ``connect`` re-inspects a PAUSING sandbox. The daemon +#: emits ``SANDBOX_EVENT_KIND_PAUSED`` on this edge (CORE-21), but the +#: settle poll predates it and remains the simple, robust route — a +#: poll-to-event-wait conversion is a candidate cleanup, not a bug. _PAUSE_SETTLE_POLL_SECONDS = 0.5 #: Default overall deadline for ``connect`` in seconds — generous @@ -110,11 +117,31 @@ class ArcBox: def __init__(self, connection: Connection | None = None) -> None: self._client = ConnectClient(connection) + self._capabilities: Capabilities | None = None def close(self) -> None: """Close the SDK-owned HTTP client (no-op for an injected one).""" self._client.close() + def capabilities(self) -> Capabilities: + """What the daemon can do: version, sandbox protocol level, + feature flags, and whether nested virtualization is available. + Answered host-side (works before any sandbox exists) and cached + for the life of this client — a failed fetch is not cached, so + the next call retries. The SDK does not gate on it: the daemon + fails fast on its own (a ``CapabilityError`` from ``create``); + this is the inspectable version of the same answer.""" + if self._capabilities is None: + with wrap_errors("arcbox.capabilities"): + self._capabilities = capabilities_from_proto( + self._client.unary( + _SANDBOX + "GetCapabilities", + sandbox_pb2.GetCapabilitiesRequest(), + sandbox_pb2.GetCapabilitiesResponse, + ) + ) + return self._capabilities + def __enter__(self) -> ArcBox: return self @@ -433,6 +460,37 @@ def check(state: int, error: str) -> bool: ) +class EventStream: + """A sandbox's lifecycle events, iterable one typed event at a time. + + The iterator ends when the daemon ends the stream. When exiting + early (``break``), iterate inside the context-manager form — + ``async with sandbox.events() as stream`` in the async flavor, + ``with`` in the sync one — so the subscription closes at the break + instead of whenever the generator finalizer runs.""" + + def __init__(self, events: Generator[SandboxEvent]) -> None: + self._events = events + + def __iter__(self) -> Iterator[SandboxEvent]: + return self._events + + def close(self) -> None: + """Cancel the subscription without consuming the rest.""" + self._events.close() + + def __enter__(self) -> EventStream: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + self.close() + + class Sandbox: """A handle to one sandbox.""" @@ -551,16 +609,85 @@ def kill(self) -> None: empty_pb2.Empty, ) + def set_lifecycle( + self, + *, + ttl: float | Unchanged | None = UNCHANGED, + idle_timeout: float | Unchanged | None = UNCHANGED, + on_idle: IdlePolicy | Unchanged | None = UNCHANGED, + ) -> None: + """Replace lifecycle deadlines. Each knob is tri-state: omitted + (the :data:`arcbox.UNCHANGED` default) leaves it as it is; + ``None`` restores the daemon default (no TTL / no idle detection + / the default idle action) — the same meaning ``None`` has on + ``create``; a value replaces it. ``ttl`` re-arms the hard cap + this many seconds from NOW — calling repeatedly keeps a busy + sandbox alive; ``idle_timeout`` re-arms a live idle timer. Works + in any non-terminal state, including paused.""" + with wrap_errors("sandbox.set_lifecycle"): + request = sandbox_pb2.SetLifecycleRequest(id=self.id) + if not isinstance(ttl, Unchanged): + request.ttl_seconds = _seconds_to_wire(ttl) + if not isinstance(idle_timeout, Unchanged): + request.idle_timeout_seconds = _seconds_to_wire(idle_timeout) + if not isinstance(on_idle, Unchanged): + request.on_idle = ( + sandbox_pb2.IDLE_ACTION_UNSPECIFIED + if on_idle is None + else sandbox_pb2.IDLE_ACTION_KILL + if on_idle == "kill" + else sandbox_pb2.IDLE_ACTION_PAUSE + ) + self._client.unary(_SANDBOX + "SetLifecycle", request, empty_pb2.Empty) + + def events(self) -> EventStream: + """Subscribe to this sandbox's lifecycle events, yielded as + typed :class:`SandboxEvent` values (keepalive frames are + filtered out). The iterator ends when the daemon ends the + stream; closing the stream (or its context) cancels the + subscription. A transport drop mid-stream is surfaced as + :class:`arcbox.errors.ConnectionLostError` — re-subscribing is + the caller's decision, since missed events cannot be replayed.""" + return EventStream(self._stream_events()) + + def _stream_events(self) -> Generator[SandboxEvent]: + with wrap_errors("sandbox.events"): + entered = False + try: + with self._client.stream( + _SANDBOX + "Events", + sandbox_pb2.SandboxEventsRequest(sandbox_id=self.id), + sandbox_pb2.WatchEventsResponse, + ) as stream: + entered = True + for frame in stream: + if frame.WhichOneof("payload") == "event": + yield sandbox_event_from_proto(frame.event) + # ConnectionFailedError covers the drop's OTHER wire shape: + # the daemon losing its upstream event source ends the HTTP + # stream cleanly with a Connect `unavailable` error frame, + # decoded into ConnectionFailedError. Daemon-typed errors map + # to other classes and keep them. + except (httpx.HTTPError, ConnectionFailedError) as exc: + if not entered or isinstance(exc, ConnectionLostError): + # Before entry: a dial failure — the daemon was never + # reached; wrap_errors maps it to + # ConnectionFailedError. ConnectionLostError is + # already the stream-death error. + raise + raise ConnectionLostError( + "the event stream died", + context={"id": self.id}, + operation="sandbox.events", + ) from exc + def pause(self) -> None: """Checkpoint the sandbox to disk under the same id and release its runtime resources. Resume happens on the next ``connect`` (or transparently, daemon-side, on the next data-plane call). Trades RAM for disk: a paused sandbox keeps paying - ``storage_bytes``. - - Requires daemon-side CORE-21: the current local daemon serves - Pause/Resume as contract-only stubs, so this raises an - unimplemented :class:`ArcBoxError` until that lands.""" + ``storage_bytes``. Requires a quiescent sandbox (READY — no + running command).""" with wrap_errors("sandbox.pause"): # No per-request deadline: checkpointing takes as long as it # takes. diff --git a/sdk/python/src/arcbox/_types.py b/sdk/python/src/arcbox/_types.py index 4416f4d80..45ac4b459 100644 --- a/sdk/python/src/arcbox/_types.py +++ b/sdk/python/src/arcbox/_types.py @@ -12,7 +12,7 @@ from dataclasses import dataclass, field from datetime import datetime, timezone -from typing import TYPE_CHECKING, Literal +from typing import TYPE_CHECKING, Final, Literal, final from arcbox._gen import process_pb2, sandbox_pb2 from arcbox.errors import ArcBoxError, CommandFailedError, SandboxDiedError @@ -36,6 +36,42 @@ #: What the daemon does when the idle timeout expires. IdlePolicy = Literal["kill", "pause"] +#: Kind of a sandbox lifecycle event. "idle" fires when an execution +#: exits and the sandbox returns to ready; "pausing"/"resumed" carry a +#: "reason" attribute distinguishing client calls from automation +#: ("idle_timeout" / "auto_resume"). +SandboxEventKind = Literal[ + "created", + "ready", + "running", + "idle", + "stopping", + "stopped", + "failed", + "removed", + "pausing", + "paused", + "resumed", + "unknown", +] + + +@final +class Unchanged: + """Sentinel type for ``set_lifecycle``: leave this knob as it is. + + ``UNCHANGED`` is its only instance and the default for every knob, + so passing it explicitly (e.g. from a conditional update) is + equivalent to omitting the argument. + """ + + def __repr__(self) -> str: + return "UNCHANGED" + + +#: The single :class:`Unchanged` sentinel. +UNCHANGED: Final = Unchanged() + #: Signals deliverable to a command's process group. SignalName = Literal["SIGTERM", "SIGKILL", "SIGINT", "SIGHUP", "SIGQUIT"] @@ -62,6 +98,27 @@ class OutputChunk: data: bytes +@dataclass(frozen=True) +class PtySize: + """Terminal geometry for a PTY command.""" + + #: Terminal width in columns. + cols: int + #: Terminal height in rows. + rows: int + + +@dataclass(frozen=True) +class StdinStatus: + """Stdin acceptance state of a command, as reported by the daemon.""" + + #: Bytes accepted and forwarded so far — the offset the next stdin + #: write starts at. + bytes_written: int + #: Whether stdin has been closed. + closed: bool + + @dataclass(frozen=True) class CommandResult: """A finished command. Non-zero exit is data, not an exception — @@ -117,6 +174,43 @@ class SandboxInfo: storage_bytes: int = 0 +@dataclass(frozen=True) +class SandboxEvent: + """One sandbox lifecycle event, as delivered by ``sandbox.events()``.""" + + sandbox_id: str + kind: SandboxEventKind + #: When it happened (daemon clock). + time: datetime | None = None + #: Per-kind context: "exit_code"/"signal" on "idle", "error" on + #: "failed", "reason" on "pausing"/"resumed". + attributes: dict[str, str] = field(default_factory=dict[str, str]) + + +@dataclass(frozen=True) +class NestedVirtCapability: + """Nested-virtualization support on this host.""" + + #: True when sandboxes can run (M3+ hardware, VZ backend). + supported: bool + #: The daemon's authoritative reason, when unsupported. + reason: str | None = None + + +@dataclass(frozen=True) +class Capabilities: + """What the daemon can do — the ``arcbox.capabilities()`` handshake.""" + + #: Daemon version string (informational). + daemon_version: str + #: Sandbox API protocol level. + protocol: int + #: Append-only named feature flags (e.g. "pause_resume"). + features: list[str] + #: Whether this host can run sandboxes at all. + nested_virt: NestedVirtCapability + + @dataclass(frozen=True) class SandboxSummary: """One row of a sandbox listing.""" @@ -208,6 +302,43 @@ def sandbox_summary_from_proto(summary: sandbox_pb2.SandboxSummary) -> SandboxSu ) +_EVENT_KIND_NAMES: dict[int, SandboxEventKind] = { + sandbox_pb2.SANDBOX_EVENT_KIND_CREATED: "created", + sandbox_pb2.SANDBOX_EVENT_KIND_READY: "ready", + sandbox_pb2.SANDBOX_EVENT_KIND_RUNNING: "running", + sandbox_pb2.SANDBOX_EVENT_KIND_IDLE: "idle", + sandbox_pb2.SANDBOX_EVENT_KIND_STOPPING: "stopping", + sandbox_pb2.SANDBOX_EVENT_KIND_STOPPED: "stopped", + sandbox_pb2.SANDBOX_EVENT_KIND_FAILED: "failed", + sandbox_pb2.SANDBOX_EVENT_KIND_REMOVED: "removed", + sandbox_pb2.SANDBOX_EVENT_KIND_PAUSING: "pausing", + sandbox_pb2.SANDBOX_EVENT_KIND_PAUSED: "paused", + sandbox_pb2.SANDBOX_EVENT_KIND_RESUMED: "resumed", +} + + +def sandbox_event_from_proto(event: sandbox_pb2.SandboxEvent) -> SandboxEvent: + """Map one Events frame to the public DTO ("unknown" for kinds this + SDK predates).""" + return SandboxEvent( + sandbox_id=event.sandbox_id, + kind=_EVENT_KIND_NAMES.get(event.kind, "unknown"), + time=_optional_time(event.time, event.HasField("time")), + attributes=dict(event.attributes), + ) + + +def capabilities_from_proto(response: sandbox_pb2.GetCapabilitiesResponse) -> Capabilities: + """Map the GetCapabilities response to the public DTO.""" + nested = response.nested_virt + return Capabilities( + daemon_version=response.daemon_version, + protocol=response.protocol, + features=list(response.features), + nested_virt=NestedVirtCapability(supported=nested.supported, reason=nested.reason or None), + ) + + def signal_display_name(value: int) -> str: """Map a POSIX signal number to its conventional name.""" try: diff --git a/sdk/python/src/arcbox/errors.py b/sdk/python/src/arcbox/errors.py index c7852ff58..5338b3f1f 100644 --- a/sdk/python/src/arcbox/errors.py +++ b/sdk/python/src/arcbox/errors.py @@ -33,6 +33,7 @@ "CommandNotFoundError", "CommandTimeoutError", "ConnectionFailedError", + "ConnectionLostError", "FileNotFoundError", "FileTooLargeError", "InvalidArgumentError", @@ -85,6 +86,12 @@ class ConnectionFailedError(ArcBoxError): """The daemon is unreachable (socket missing, connection refused, ...).""" +class ConnectionLostError(ConnectionFailedError): + """A live stream died mid-flow — and, where the SDK re-attaches + (command output), could not be re-established within the retry + budget. ``__cause__`` carries the underlying transport failure.""" + + class AuthenticationError(ArcBoxError): """Authentication is required or was rejected. Reserved for the remote tier (CORE-63).""" diff --git a/sdk/python/tests/test_e2e.py b/sdk/python/tests/test_e2e.py index 4785f1d5b..2cd35fe8b 100644 --- a/sdk/python/tests/test_e2e.py +++ b/sdk/python/tests/test_e2e.py @@ -5,11 +5,11 @@ Connection resolution applies, so ARCBOX_SOCKET / ARCBOX_DATA_DIR point the loop at a dev daemon. -This is the design doc's 20-line hello world minus the parts outside -phase 1's surface: ports.expose and wait_for_port are deferred, so the -background command is observed through its output stream and -wait_for_exit instead of a port probe. Both surfaces run the same -scenario — the sync flavor exercises the generated tree end to end. +This is the design doc's 20-line hello world plus the phase 2a surface: +PTY, stdin, re-attach, set_lifecycle, capabilities, and events. Still +outside scope: ports.expose and wait_for_port (2b). The sync flavor +exercises the generated tree end to end — including genuinely blocking +event-stream reads. """ from __future__ import annotations @@ -18,7 +18,7 @@ import pytest -from arcbox import AsyncSandbox, Sandbox +from arcbox import ArcBox, AsyncSandbox, PtySize, Sandbox pytestmark = pytest.mark.skipif( os.environ.get("ARCBOX_SDK_E2E") != "1", @@ -26,6 +26,17 @@ ) +def test_capabilities_handshake() -> None: + # This suite only runs on sandbox-capable hosts, so nested_virt must + # report supported — the same answer create() relies on. + with ArcBox() as box: + caps = box.capabilities() + assert caps.protocol >= 1 + assert caps.daemon_version != "" + assert "pause_resume" in caps.features + assert caps.nested_virt.supported is True + + def test_sync_hello_world() -> None: # Built-in minimal template (busybox) — no image pull involved. sandbox = Sandbox.create("", ttl=300) @@ -55,12 +66,70 @@ def test_sync_hello_world() -> None: assert killed.signal == "SIGKILL" assert killed.exit_code == 137 + # stdin: foreground write-then-close (subprocess semantics). + echoed = sandbox.commands.run(["/bin/cat"], stdin="hello stdin\n") + assert echoed.expect().stdout == "hello stdin\n" + + # stdin: a background handle drives offset-idempotent writes. + cat_bg = sandbox.commands.run(["/bin/cat"], background=True, stdin=True) + cat_bg.write_stdin("first ") + cat_bg.write_stdin("second\n") + cat_bg.close_stdin() + assert cat_bg.wait_for_exit(30).stdout == "first second\n" + + # commands.get: re-attach by id; the retained output replays. + again = sandbox.commands.get(cat_bg.command_id) + assert again.wait_for_exit(30).stdout == "first second\n" + + # PTY: stty reads the allocated terminal's geometry (rows cols), + # and output arrives merged (a pty run's stdout carries it all). + tty = sandbox.commands.run("stty size", pty=PtySize(cols=120, rows=40)) + assert "40 120" in tty.expect().stdout + + # PTY resize: the running terminal observes the new geometry. + resized = sandbox.commands.run( + "sleep 2; stty size", pty=PtySize(cols=80, rows=24), background=True + ) + resized.resize(200, 50) + assert "50 200" in resized.wait_for_exit(30).stdout + + # set_lifecycle tri-state: re-arm the TTL, then remove it (None), + # then re-arm again so the sandbox cannot outlive a crash here. + sandbox.set_lifecycle(ttl=600) + assert sandbox.info().ttl_deadline is not None + sandbox.set_lifecycle(ttl=None) + assert sandbox.info().ttl_deadline is None + sandbox.set_lifecycle(ttl=300) + # info() is always fresh. assert sandbox.info().state in ("ready", "running") finally: sandbox.kill() +def test_events_observe_the_idle_auto_pause() -> None: + # A short idle timeout with the PAUSE policy: the daemon must emit + # PAUSING (reason idle_timeout) then PAUSED on the events stream. + # The sync flavor genuinely blocks on each read — no polling. + sandbox = Sandbox.create("", ttl=300, idle_timeout=4, on_idle="pause") + try: + kinds: list[str] = [] + reason = "" + with sandbox.events() as stream: + for event in stream: + kinds.append(event.kind) + if event.kind == "pausing": + reason = event.attributes.get("reason", "") + if event.kind == "paused": + break + assert "pausing" in kinds + assert kinds[-1] == "paused" + assert reason == "idle_timeout" + assert sandbox.info().state == "paused" + finally: + sandbox.kill() + + @pytest.mark.anyio async def test_async_hello_world() -> None: sandbox = await AsyncSandbox.create("", ttl=300) diff --git a/sdk/python/tests/test_lifecycle.py b/sdk/python/tests/test_lifecycle.py new file mode 100644 index 000000000..5daf09e76 --- /dev/null +++ b/sdk/python/tests/test_lifecycle.py @@ -0,0 +1,249 @@ +"""events(), set_lifecycle(), and the capabilities handshake against a +mock daemon. + +set_lifecycle's tri-state is the contract that matters: an omitted knob +must be ABSENT on the wire (unchanged), None must be an explicit +zero/UNSPECIFIED (restore the default — the same meaning None has on +create), and a value must replace. Getting presence wrong silently +rewrites deadlines the caller never touched. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import httpx +import pytest +from google.protobuf import empty_pb2 + +if TYPE_CHECKING: + from collections.abc import Callable + +from arcbox import UNCHANGED, ArcBox, AsyncArcBox, AsyncSandbox, Connection, Sandbox +from arcbox._async._client import AsyncConnectClient +from arcbox._envelope import FLAG_END_STREAM, encode_envelope +from arcbox._gen import sandbox_pb2 +from arcbox._sync._client import ConnectClient +from arcbox.errors import ConnectionLostError + +if TYPE_CHECKING: + from google.protobuf.message import Message + + +def proto_response(message: Message) -> httpx.Response: + return httpx.Response( + 200, + content=message.SerializeToString(), + headers={"content-type": "application/proto"}, + ) + + +def event_frame(kind: sandbox_pb2.SandboxEventKind, **attributes: str) -> bytes: + frame = sandbox_pb2.WatchEventsResponse() + frame.event.sandbox_id = "sb-1" + frame.event.kind = kind + for key, value in attributes.items(): + frame.event.attributes[key] = value + return encode_envelope(0, frame.SerializeToString()) + + +def keepalive_frame() -> bytes: + frame = sandbox_pb2.WatchEventsResponse() + frame.keep_alive.SetInParent() + return encode_envelope(0, frame.SerializeToString()) + + +def events_response(frames: list[bytes], truncated: bool = False) -> httpx.Response: + body = b"".join(frames) + if not truncated: + body += encode_envelope(FLAG_END_STREAM, b"{}") + return httpx.Response(200, content=body, headers={"content-type": "application/connect+proto"}) + + +def sync_sandbox(handler: Callable[[httpx.Request], httpx.Response]) -> Sandbox: + http = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://arcbox") + return Sandbox(ConnectClient(Connection(http_client=http)), "sb-1") + + +class TestEvents: + def test_yields_typed_events_and_filters_keepalives(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/Events") + return events_response( + [ + keepalive_frame(), + event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_CREATED), + event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_READY), + keepalive_frame(), + event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_PAUSING, reason="idle_timeout"), + event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_PAUSED), + ] + ) + + sandbox = sync_sandbox(handler) + kinds: list[str] = [] + reasons: list[str | None] = [] + with sandbox.events() as stream: + for event in stream: + kinds.append(event.kind) + if event.kind == "pausing": + reasons.append(event.attributes.get("reason")) + assert kinds == ["created", "ready", "pausing", "paused"] + assert reasons == ["idle_timeout"] + + def test_early_exit_via_the_context_closes_the_subscription(self) -> None: + def handler(_request: httpx.Request) -> httpx.Response: + return events_response([event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_READY)] * 5) + + sandbox = sync_sandbox(handler) + with sandbox.events() as stream: + for event in stream: + assert event.kind == "ready" + break + with pytest.raises(StopIteration): + next(iter(stream)) + + def test_a_mid_stream_drop_is_the_stream_death_error(self) -> None: + def handler(_request: httpx.Request) -> httpx.Response: + return events_response( + [event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_READY)], truncated=True + ) + + sandbox = sync_sandbox(handler) + with pytest.raises(ConnectionLostError): + for _event in sandbox.events(): + pass + + def test_a_server_signaled_unavailable_end_frame_is_also_stream_death(self) -> None: + # The drop's other wire shape: the daemon loses its upstream + # event source and ends the HTTP stream CLEANLY with a Connect + # `unavailable` error frame. + def handler(_request: httpx.Request) -> httpx.Response: + end = b'{"error": {"code": "unavailable", "message": "event source lost"}}' + body = event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_READY) + encode_envelope( + FLAG_END_STREAM, end + ) + return httpx.Response( + 200, content=body, headers={"content-type": "application/connect+proto"} + ) + + sandbox = sync_sandbox(handler) + with pytest.raises(ConnectionLostError): + for _event in sandbox.events(): + pass + + +class LifecycleProbe: + def __init__(self) -> None: + self.requests: list[sandbox_pb2.SetLifecycleRequest] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/SetLifecycle") + self.requests.append(sandbox_pb2.SetLifecycleRequest.FromString(request.content)) + return proto_response(empty_pb2.Empty()) + + +class TestSetLifecycleTriState: + def test_omitted_knobs_are_absent_on_the_wire(self) -> None: + probe = LifecycleProbe() + sync_sandbox(probe).set_lifecycle() + req = probe.requests[0] + assert not req.HasField("ttl_seconds") + assert not req.HasField("idle_timeout_seconds") + assert not req.HasField("on_idle") + + def test_values_replace(self) -> None: + probe = LifecycleProbe() + sync_sandbox(probe).set_lifecycle(ttl=5, idle_timeout=30) + req = probe.requests[0] + assert req.ttl_seconds == 5 + assert req.idle_timeout_seconds == 30 + assert not req.HasField("on_idle") + + def test_none_restores_the_default_as_explicit_zero(self) -> None: + probe = LifecycleProbe() + sync_sandbox(probe).set_lifecycle(ttl=None, on_idle=None) + req = probe.requests[0] + assert req.HasField("ttl_seconds") + assert req.ttl_seconds == 0 + assert not req.HasField("idle_timeout_seconds") + assert req.HasField("on_idle") + assert req.on_idle == sandbox_pb2.IDLE_ACTION_UNSPECIFIED + + def test_explicit_unchanged_equals_omission(self) -> None: + probe = LifecycleProbe() + sync_sandbox(probe).set_lifecycle(ttl=UNCHANGED, on_idle="pause") + req = probe.requests[0] + assert not req.HasField("ttl_seconds") + assert req.on_idle == sandbox_pb2.IDLE_ACTION_PAUSE + + +class CapsDaemon: + def __init__(self, fail_first: bool = False) -> None: + self.calls = 0 + self.fail_first = fail_first + + def __call__(self, request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/GetCapabilities") + self.calls += 1 + if self.fail_first and self.calls == 1: + return httpx.Response(503, content=b'{"code": "unavailable", "message": "starting up"}') + response = sandbox_pb2.GetCapabilitiesResponse( + daemon_version="0.9.0", + protocol=1, + features=["pause_resume", "auto_resume"], + ) + response.nested_virt.supported = False + response.nested_virt.reason = "requires M3 or newer" + return proto_response(response) + + +class TestCapabilities: + def test_maps_the_handshake_and_caches_it_per_client(self) -> None: + daemon = CapsDaemon() + http = httpx.Client(transport=httpx.MockTransport(daemon), base_url="http://arcbox") + box = ArcBox(Connection(http_client=http)) + caps = box.capabilities() + assert caps.daemon_version == "0.9.0" + assert caps.protocol == 1 + assert caps.features == ["pause_resume", "auto_resume"] + assert caps.nested_virt.supported is False + assert caps.nested_virt.reason == "requires M3 or newer" + assert box.capabilities() is caps + assert daemon.calls == 1 + + def test_a_failed_fetch_is_not_cached(self) -> None: + daemon = CapsDaemon(fail_first=True) + http = httpx.Client(transport=httpx.MockTransport(daemon), base_url="http://arcbox") + box = ArcBox(Connection(http_client=http)) + with pytest.raises(Exception, match="starting up"): + box.capabilities() + assert box.capabilities().protocol == 1 + assert daemon.calls == 2 + + +@pytest.mark.anyio +async def test_async_tree_runs_the_same_surface() -> None: + probe = LifecycleProbe() + http = httpx.AsyncClient(transport=httpx.MockTransport(probe), base_url="http://arcbox") + sandbox = AsyncSandbox(AsyncConnectClient(Connection(http_client=http)), "sb-1") + await sandbox.set_lifecycle(ttl=None) + assert probe.requests[0].HasField("ttl_seconds") + + daemon = CapsDaemon() + http2 = httpx.AsyncClient(transport=httpx.MockTransport(daemon), base_url="http://arcbox") + box = AsyncArcBox(Connection(http_client=http2)) + assert (await box.capabilities()).protocol == 1 + + def events_handler(_request: httpx.Request) -> httpx.Response: + return events_response([event_frame(sandbox_pb2.SANDBOX_EVENT_KIND_READY)]) + + http3 = httpx.AsyncClient( + transport=httpx.MockTransport(events_handler), base_url="http://arcbox" + ) + watching = AsyncSandbox(AsyncConnectClient(Connection(http_client=http3)), "sb-1") + kinds: list[str] = [] + async with watching.events() as stream: + async for event in stream: + kinds.append(event.kind) + assert kinds == ["ready"] diff --git a/sdk/python/tests/test_reattach.py b/sdk/python/tests/test_reattach.py new file mode 100644 index 000000000..b2aff819e --- /dev/null +++ b/sdk/python/tests/test_reattach.py @@ -0,0 +1,262 @@ +"""Offset-resume across stream death against a mock daemon. + +The contract under test: when an attach stream drops mid-flow, the +handle re-attaches from the last DELIVERED per-channel offsets and the +consumer sees one seamless, gapless stream — the SDK's whole reason for +offset-addressed output. A truncated streaming body (no terminal +EndStreamResponse frame) IS the drop: the connection died mid-body. +Retries are bounded by consecutive dead dials; delivered output resets +the budget. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +import httpx +import pytest + +from arcbox import AsyncSandbox, Connection, Sandbox +from arcbox._async._client import AsyncConnectClient +from arcbox._envelope import FLAG_END_STREAM, EnvelopeDecoder, encode_envelope +from arcbox._gen import process_pb2 +from arcbox._sync._client import ConnectClient +from arcbox.errors import ConnectionLostError, NotFoundError + +if TYPE_CHECKING: + from google.protobuf.message import Message + + +def proto_response(message: Message) -> httpx.Response: + return httpx.Response( + 200, + content=message.SerializeToString(), + headers={"content-type": "application/proto"}, + ) + + +def stream_response(frames: list[bytes], truncated: bool) -> httpx.Response: + body = b"".join(frames) + if not truncated: + body += encode_envelope(FLAG_END_STREAM, b"{}") + return httpx.Response(200, content=body, headers={"content-type": "application/connect+proto"}) + + +def exited_execution() -> process_pb2.Execution: + execution = process_pb2.Execution(id="cmd", state=process_pb2.EXECUTION_STATE_EXITED) + execution.exit_status.code = 0 + return execution + + +@dataclass +class Chunk: + channel: process_pb2.StdioChannel + offset: int + text: bytes + + +@dataclass +class FlakyDaemon: + """Serves AttachExecution from a chunk script, truncating the body + after ``die_after[n]`` chunks on the n-th attach (die forever once + the script runs out). Replays only chunks at or past the requested + offset, like the daemon.""" + + chunks: list[Chunk] + die_after: list[int] + attaches: list[process_pb2.AttachExecutionRequest] = field( + default_factory=list[process_pb2.AttachExecutionRequest] + ) + + def __call__(self, request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/StartExecution"): + req = process_pb2.StartExecutionRequest.FromString(request.content) + running = process_pb2.EXECUTION_STATE_RUNNING + return proto_response(process_pb2.Execution(id=req.execution_id, state=running)) + if path.endswith("/WaitExecution"): + return proto_response(exited_execution()) + if path.endswith("/AttachExecution"): + _flags, payload = EnvelopeDecoder().feed(request.content)[0] + attach = process_pb2.AttachExecutionRequest.FromString(payload) + call = len(self.attaches) + self.attaches.append(attach) + budget = self.die_after[call] if call < len(self.die_after) else None + frames: list[bytes] = [] + sent = 0 + for chunk in self.chunks: + is_stderr = chunk.channel == process_pb2.STDIO_CHANNEL_STDERR + from_offset = attach.stderr_offset if is_stderr else attach.stdout_offset + if chunk.offset < from_offset: + continue + if budget is not None and sent >= budget: + return stream_response(frames, truncated=True) + event = process_pb2.ExecutionEvent() + event.output.channel = chunk.channel + event.output.offset = chunk.offset + event.output.data = chunk.text + frames.append(encode_envelope(0, event.SerializeToString())) + sent += 1 + if budget is not None: + return stream_response(frames, truncated=True) + exited = process_pb2.ExecutionEvent() + exited.exited.execution.CopyFrom(exited_execution()) + frames.append(encode_envelope(0, exited.SerializeToString())) + return stream_response(frames, truncated=False) + return httpx.Response(404, content=b"unhandled: " + path.encode()) + + +def script() -> list[Chunk]: + return [ + Chunk(process_pb2.STDIO_CHANNEL_STDOUT, 0, b"hel"), + Chunk(process_pb2.STDIO_CHANNEL_STDERR, 0, b"warn"), + Chunk(process_pb2.STDIO_CHANNEL_STDOUT, 3, b"lo "), + Chunk(process_pb2.STDIO_CHANNEL_STDOUT, 6, b"world"), + ] + + +def sync_sandbox(daemon: FlakyDaemon) -> Sandbox: + http = httpx.Client(transport=httpx.MockTransport(daemon), base_url="http://arcbox") + return Sandbox(ConnectClient(Connection(http_client=http)), "sb-1") + + +def test_the_output_iterator_reattaches_from_the_delivered_offsets() -> None: + # First attach dies after two chunks (stdout "hel" + stderr "warn"). + daemon = FlakyDaemon(script(), [2]) + handle = sync_sandbox(daemon).commands.run("emit", background=True) + stdout = b"" + stderr = b"" + for chunk in handle.output: + if chunk.channel == "stderr": + stderr += chunk.data + else: + stdout += chunk.data + # Seamless and gapless despite the mid-stream death. + assert stdout == b"hello world" + assert stderr == b"warn" + assert len(daemon.attaches) == 2 + # The re-attach resumed exactly at the delivered high-water marks. + assert daemon.attaches[1].stdout_offset == 3 + assert daemon.attaches[1].stderr_offset == 4 + + +def test_repeated_drops_survive_while_each_dial_delivers_output() -> None: + # Every attach dies after one delivered chunk; progress resets the + # retry budget each time. + daemon = FlakyDaemon(script(), [1, 1, 1, 1]) + handle = sync_sandbox(daemon).commands.run("emit", background=True) + stdout = b"".join(c.data for c in handle.output if c.channel != "stderr") + assert stdout == b"hello world" + assert len(daemon.attaches) == 5 + + +def test_wait_for_exit_result_collection_resumes_through_the_same_loop() -> None: + daemon = FlakyDaemon(script(), [3]) + handle = sync_sandbox(daemon).commands.run("emit", background=True) + result = handle.wait_for_exit(5) + assert result.stdout == "hello world" + assert result.stderr == "warn" + # The resumed chunks were contiguous — no false truncation flag. + assert result.truncated is False + + +def test_bounded_retries_exhaust_into_the_stream_death_error() -> None: + # Every dial dies before delivering anything. + daemon = FlakyDaemon(script(), [0] * 8) + handle = sync_sandbox(daemon).commands.run("emit", background=True) + with pytest.raises(ConnectionLostError) as exc_info: + for _chunk in handle.output: + pass + assert exc_info.value.context["retries"] == "3" + # The initial dial plus three re-dials. + assert len(daemon.attaches) == 4 + + +def test_a_server_signaled_unavailable_end_frame_is_retried() -> None: + # The drop's other wire shape: the daemon loses its upstream agent + # stream and ends the HTTP stream CLEANLY with a Connect + # `unavailable` error frame. That must resume like raw truncation. + attaches: list[process_pb2.AttachExecutionRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/StartExecution"): + req = process_pb2.StartExecutionRequest.FromString(request.content) + running = process_pb2.EXECUTION_STATE_RUNNING + return proto_response(process_pb2.Execution(id=req.execution_id, state=running)) + if path.endswith("/AttachExecution"): + _flags, payload = EnvelopeDecoder().feed(request.content)[0] + attach = process_pb2.AttachExecutionRequest.FromString(payload) + attaches.append(attach) + if len(attaches) == 1: + event = process_pb2.ExecutionEvent() + event.output.channel = process_pb2.STDIO_CHANNEL_STDOUT + event.output.offset = 0 + event.output.data = b"hel" + end = b'{"error": {"code": "unavailable", "message": "agent stream lost"}}' + body = encode_envelope(0, event.SerializeToString()) + encode_envelope( + FLAG_END_STREAM, end + ) + return httpx.Response( + 200, content=body, headers={"content-type": "application/connect+proto"} + ) + event = process_pb2.ExecutionEvent() + event.output.channel = process_pb2.STDIO_CHANNEL_STDOUT + event.output.offset = 3 + event.output.data = b"lo" + exited = process_pb2.ExecutionEvent() + exited.exited.execution.CopyFrom(exited_execution()) + frames = [ + encode_envelope(0, event.SerializeToString()), + encode_envelope(0, exited.SerializeToString()), + ] + return stream_response(frames, truncated=False) + return httpx.Response(404) + + http = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://arcbox") + sandbox = Sandbox(ConnectClient(Connection(http_client=http)), "sb-1") + handle = sandbox.commands.run("emit", background=True) + stdout = b"".join(chunk.data for chunk in handle.output) + assert stdout == b"hello" + assert attaches[1].stdout_offset == 3 + + +def test_a_daemon_typed_stream_error_is_surfaced_never_retried() -> None: + attaches = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal attaches + path = request.url.path + if path.endswith("/WaitExecution"): + return proto_response(exited_execution()) + if path.endswith("/AttachExecution"): + attaches += 1 + end = b'{"error": {"code": "not_found", "message": "no such execution"}}' + body = encode_envelope(FLAG_END_STREAM, end) + return httpx.Response( + 200, content=body, headers={"content-type": "application/connect+proto"} + ) + return httpx.Response(404) + + http = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://arcbox") + sandbox = Sandbox(ConnectClient(Connection(http_client=http)), "sb-1") + handle = sandbox.commands.get("cmd") + with pytest.raises(NotFoundError): + for _chunk in handle.output: + pass + assert attaches == 1 + + +@pytest.mark.anyio +async def test_async_tree_resumes_the_same_way() -> None: + daemon = FlakyDaemon(script(), [2]) + http = httpx.AsyncClient(transport=httpx.MockTransport(daemon), base_url="http://arcbox") + sandbox = AsyncSandbox(AsyncConnectClient(Connection(http_client=http)), "sb-1") + handle = await sandbox.commands.run("emit", background=True) + stdout = b"" + async for chunk in handle.output: + if chunk.channel != "stderr": + stdout += chunk.data + assert stdout == b"hello world" + assert daemon.attaches[1].stdout_offset == 3 diff --git a/sdk/python/tests/test_stdin_pty.py b/sdk/python/tests/test_stdin_pty.py new file mode 100644 index 000000000..1a9d988ec --- /dev/null +++ b/sdk/python/tests/test_stdin_pty.py @@ -0,0 +1,240 @@ +"""PTY and stdin against a mock daemon. + +The invariant that matters: stdin writes are offset-idempotent. The +handle advances its cursor only on a successful response, so a retry of +a lost write lands at the SAME offset and the daemon's deduplication +swallows the duplicate — never a double feed. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import httpx +import pytest +from google.protobuf import empty_pb2 + +from arcbox import AsyncSandbox, Connection, PtySize, Sandbox +from arcbox._async._client import AsyncConnectClient +from arcbox._envelope import FLAG_END_STREAM, encode_envelope +from arcbox._gen import process_pb2 +from arcbox._sync._client import ConnectClient +from arcbox.errors import ConnectionFailedError, InvalidArgumentError, SandboxStateError + +if TYPE_CHECKING: + from google.protobuf.message import Message + + +def proto_response(message: Message) -> httpx.Response: + return httpx.Response( + 200, + content=message.SerializeToString(), + headers={"content-type": "application/proto"}, + ) + + +class MockDaemon: + """A process-service double tracking stdin acceptance like the guest.""" + + def __init__(self) -> None: + self.starts: list[process_pb2.StartExecutionRequest] = [] + self.writes: list[process_pb2.WriteStdinRequest] = [] + self.resizes: list[process_pb2.TerminalSize] = [] + self.signals: list[int] = [] + self.accepted = 0 + self.closed = False + #: Accept the next write, then fail its response (a lost response). + self.fail_next_write_response = False + #: Execution state served by WaitExecution polls. + self.wait_state: process_pb2.ExecutionState = process_pb2.EXECUTION_STATE_EXITED + #: Reject writes with this status code without accepting anything. + self.reject_writes: int | None = None + + def __call__(self, request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/StartExecution"): + req = process_pb2.StartExecutionRequest.FromString(request.content) + self.starts.append(req) + running = process_pb2.EXECUTION_STATE_RUNNING + return proto_response(process_pb2.Execution(id=req.execution_id, state=running)) + if path.endswith("/WriteStdin"): + write = process_pb2.WriteStdinRequest.FromString(request.content) + self.writes.append(write) + if self.reject_writes is not None: + body = b'{"code": "failed_precondition", "message": "execution has exited"}' + return httpx.Response(self.reject_writes, content=body) + if write.offset > self.accepted: + return httpx.Response(416, content=b'{"message": "stdin gap"}') + # Deduplicate bytes below the accepted count (the guest contract). + fresh = write.offset + len(write.data) - self.accepted + if fresh > 0: + self.accepted += fresh + if write.eof: + self.closed = True + if self.fail_next_write_response: + self.fail_next_write_response = False + raise httpx.ConnectError("connection reset", request=request) + return proto_response( + process_pb2.StdinStatus(bytes_written=self.accepted, closed=self.closed) + ) + if path.endswith("/GetStdinStatus"): + return proto_response( + process_pb2.StdinStatus(bytes_written=self.accepted, closed=self.closed) + ) + if path.endswith("/SignalExecution"): + signal = process_pb2.SignalExecutionRequest.FromString(request.content) + self.signals.append(signal.signal) + return proto_response(empty_pb2.Empty()) + if path.endswith("/ResizeExecutionTty"): + resize = process_pb2.ResizeExecutionTtyRequest.FromString(request.content) + self.resizes.append(resize.size) + return proto_response(empty_pb2.Empty()) + if path.endswith("/WaitExecution"): + req = process_pb2.WaitExecutionRequest.FromString(request.content) + execution = process_pb2.Execution(id=req.execution_id, state=self.wait_state) + if self.wait_state == process_pb2.EXECUTION_STATE_EXITED: + execution.exit_status.code = 0 + execution.stdin.bytes_written = self.accepted + execution.stdin.closed = self.closed + return proto_response(execution) + if path.endswith("/AttachExecution"): + exited = process_pb2.ExecutionEvent() + exited.exited.execution.state = process_pb2.EXECUTION_STATE_EXITED + exited.exited.execution.exit_status.code = 0 + body = encode_envelope(0, exited.SerializeToString()) + encode_envelope( + FLAG_END_STREAM, b"{}" + ) + return httpx.Response( + 200, content=body, headers={"content-type": "application/connect+proto"} + ) + return httpx.Response(404, content=b"unhandled: " + path.encode()) + + +def sync_sandbox(daemon: MockDaemon) -> Sandbox: + http = httpx.Client(transport=httpx.MockTransport(daemon), base_url="http://arcbox") + return Sandbox(ConnectClient(Connection(http_client=http)), "sb-1") + + +class TestPty: + def test_pty_starts_a_tty_execution_with_stdin_open(self) -> None: + daemon = MockDaemon() + sync_sandbox(daemon).commands.run("stty size", pty=PtySize(cols=120, rows=40)) + start = daemon.starts[0] + assert start.tty is True + assert (start.tty_size.width, start.tty_size.height) == (120, 40) + assert start.stdin is True + + def test_a_plain_run_keeps_stdin_at_eof_without_a_tty(self) -> None: + daemon = MockDaemon() + sync_sandbox(daemon).commands.run("true") + start = daemon.starts[0] + assert start.tty is False + assert start.stdin is False + + def test_resize_sends_the_new_geometry(self) -> None: + daemon = MockDaemon() + handle = sync_sandbox(daemon).commands.run( + "sh", pty=PtySize(cols=80, rows=24), background=True + ) + handle.resize(200, 50) + assert (daemon.resizes[0].width, daemon.resizes[0].height) == (200, 50) + + def test_pty_with_one_shot_stdin_is_rejected_before_any_rpc(self) -> None: + daemon = MockDaemon() + with pytest.raises(InvalidArgumentError): + sync_sandbox(daemon).commands.run("cat", pty=PtySize(cols=80, rows=24), stdin="x") + assert daemon.starts == [] + + +class TestStdin: + def test_cursor_tracks_across_writes_and_close_lands_at_it(self) -> None: + daemon = MockDaemon() + handle = sync_sandbox(daemon).commands.run("cat", background=True, stdin=True) + handle.write_stdin("hello ") + handle.write_stdin("world\n") + handle.close_stdin() + assert [w.offset for w in daemon.writes] == [0, 6, 12] + assert daemon.writes[2].eof is True + assert daemon.accepted == 12 + assert daemon.closed is True + + def test_a_retried_lost_write_never_double_feeds(self) -> None: + daemon = MockDaemon() + handle = sync_sandbox(daemon).commands.run("cat", background=True, stdin=True) + handle.write_stdin("hello ") + # The daemon accepts the write but the response is lost. + daemon.fail_next_write_response = True + with pytest.raises(ConnectionFailedError): + handle.write_stdin("world") + assert daemon.accepted == 11 + # The cursor did not advance, so the retry lands at the same + # offset and the daemon deduplicates it — the process saw the + # bytes once. + handle.write_stdin("world") + assert daemon.writes[-1].offset == 6 + assert daemon.accepted == 11 + + def test_stdin_status_reports_acceptance_and_resyncs(self) -> None: + daemon = MockDaemon() + handle = sync_sandbox(daemon).commands.run("cat", background=True, stdin=True) + handle.write_stdin("abc") + status = handle.stdin_status() + assert status.bytes_written == 3 + assert status.closed is False + + def test_foreground_stdin_is_written_then_closed_before_the_wait(self) -> None: + daemon = MockDaemon() + result = sync_sandbox(daemon).commands.run("cat", stdin="fed\n") + assert result.exit_code == 0 + assert daemon.starts[0].stdin is True + assert [w.eof for w in daemon.writes] == [False, True] + assert daemon.closed is True + + def test_a_feed_racing_an_early_exit_is_noise_not_a_failure(self) -> None: + daemon = MockDaemon() + daemon.reject_writes = 412 + daemon.wait_state = process_pb2.EXECUTION_STATE_EXITED + result = sync_sandbox(daemon).commands.run("true", stdin="unread") + assert result.exit_code == 0 + + def test_a_feed_failure_with_the_process_still_running_is_real(self) -> None: + daemon = MockDaemon() + daemon.reject_writes = 412 + daemon.wait_state = process_pb2.EXECUTION_STATE_RUNNING + with pytest.raises(SandboxStateError): + sync_sandbox(daemon).commands.run("cat", stdin="data") + # The caller gets no handle out of a raised run(): without the + # best-effort kill, cat would wait on stdin forever and keep the + # sandbox RUNNING. + assert daemon.signals == [process_pb2.SIGNAL_SIGKILL] + + def test_foreground_stdin_true_is_rejected_before_any_rpc(self) -> None: + daemon = MockDaemon() + with pytest.raises(InvalidArgumentError): + sync_sandbox(daemon).commands.run("cat", stdin=True) + assert daemon.starts == [] + + +class TestGet: + def test_get_reattaches_and_seeds_the_stdin_cursor(self) -> None: + daemon = MockDaemon() + daemon.accepted = 7 + daemon.wait_state = process_pb2.EXECUTION_STATE_RUNNING + handle = sync_sandbox(daemon).commands.get("exec-9") + assert handle.command_id == "exec-9" + handle.write_stdin("more") + assert daemon.writes[0].offset == 7 + + +@pytest.mark.anyio +async def test_async_tree_runs_the_same_stdin_loop() -> None: + daemon = MockDaemon() + http = httpx.AsyncClient(transport=httpx.MockTransport(daemon), base_url="http://arcbox") + sandbox = AsyncSandbox(AsyncConnectClient(Connection(http_client=http)), "sb-1") + handle = await sandbox.commands.run("cat", background=True, stdin=True) + await handle.write_stdin("ab") + await handle.close_stdin() + assert [w.offset for w in daemon.writes] == [0, 2] + assert daemon.closed is True + again = await sandbox.commands.get(handle.command_id) + assert again.command_id == handle.command_id diff --git a/sdk/python/tests/test_sync_parity.py b/sdk/python/tests/test_sync_parity.py index 429091124..950c429e4 100644 --- a/sdk/python/tests/test_sync_parity.py +++ b/sdk/python/tests/test_sync_parity.py @@ -24,12 +24,14 @@ AsyncCommandHandle, AsyncCommands, AsyncConnectClient, + AsyncEventStream, AsyncFiles, AsyncOutputStream, AsyncSandbox, CommandHandle, Commands, ConnectClient, + EventStream, Files, OutputStream, Sandbox, @@ -43,6 +45,7 @@ (AsyncCommands, Commands), (AsyncCommandHandle, CommandHandle), (AsyncOutputStream, OutputStream), + (AsyncEventStream, EventStream), (AsyncFiles, Files), (AsyncConnectClient, ConnectClient), ] diff --git a/sdk/typescript/README.md b/sdk/typescript/README.md index eb62b6012..0df484e70 100644 --- a/sdk/typescript/README.md +++ b/sdk/typescript/README.md @@ -124,12 +124,19 @@ package that already exists on the registry): ## Status -Phase 1 of CORE-58 — the hello-world closed loop: `Sandbox` -create/connect/list, `kill`/`pause`/`info` (`pause` and the -paused-sandbox reconnect path are wire-complete but reject with -Unimplemented until the daemon's CORE-21 lands — the daemon serves -Pause/Resume as contract-only stubs today), `commands.run` (foreground -result + background handle with streamed output, `waitForExit`, `kill`), -and whole-file `files` read/write. Deferred: PTY, `ports`, -`waitForPort`/`waitForLog`, filesystem path verbs (stat/list/mkdir/…), -`Template` statics, `events()`, `setLifecycle`, capabilities handshake. +Phase 2a of CORE-58. Shipped: + +| Surface | Notes | +| --- | --- | +| `Sandbox` create/connect/list, `kill`/`pause`/`info` | pause/resume are live daemon-side (CORE-21); data-plane calls auto-resume a paused sandbox | +| `commands.run` | foreground result + background handle; `stdin: string\|bytes` (write-then-close) or `stdin: true` (keep open); `pty: {cols, rows}` | +| `CommandHandle` | streamed `output` with transparent offset-resume across stream death (bounded retries → `ConnectionLostError`), `waitForExit`, `kill`, `writeStdin`/`closeStdin`/`stdinStatus` (offset-idempotent), `resize` | +| `commands.get(id)` | re-attach a handle by execution id (stdin cursor seeded from the daemon) | +| `sandbox.events()` | typed lifecycle events, keepalives filtered; a mid-stream drop is `ConnectionLostError` | +| `sandbox.setLifecycle()` | tri-state: omitted = unchanged, `null` = restore default, value = replace | +| `arcbox.capabilities()` | daemon handshake (version/protocol/features/nested virt), cached per client | +| `files` | whole-file read/write | + +Deferred to phase 2b (the daemon answers Unimplemented today): +`commands.list()` (ListExecutions), `ports` + `waitForPort`, filesystem +path verbs (stat/list/mkdir/…), `Template` statics. diff --git a/sdk/typescript/src/commands.ts b/sdk/typescript/src/commands.ts index 52f88cde9..6d0f98cbc 100644 --- a/sdk/typescript/src/commands.ts +++ b/sdk/typescript/src/commands.ts @@ -6,11 +6,17 @@ import { createClient } from "@connectrpc/connect"; import { ArcBoxError, CommandFailedError, + ConnectionFailedError, + ConnectionLostError, + InvalidArgumentError, SandboxDiedError, TimeoutError, toArcBoxError, } from "./errors"; -import type { Execution } from "./gen/arcbox/sandbox/v1/process_pb"; +import type { + Execution, + ExecutionEvent, +} from "./gen/arcbox/sandbox/v1/process_pb"; import { ExecutionState, SandboxProcessService, @@ -36,6 +42,33 @@ const SIGNAL_VALUES: Record = { SIGTERM: Signal.SIGTERM, }; +/** + * Consecutive dead re-attach dials tolerated before an output stream + * surfaces the stream-death error. Delivered output resets the budget, + * so a long-lived stream survives any number of isolated drops. + */ +const MAX_ATTACH_RETRIES = 3; +const MAX_ATTACH_RETRIES_LABEL = String(MAX_ATTACH_RETRIES); + +/** Terminal geometry for a PTY command. */ +export interface PtySize { + /** Terminal width in columns. */ + cols: number; + /** Terminal height in rows. */ + rows: number; +} + +/** Stdin acceptance state of a command, as reported by the daemon. */ +export interface StdinStatus { + /** + * Bytes accepted and forwarded so far — the offset the next stdin + * write starts at. + */ + bytesWritten: number; + /** Whether stdin has been closed. */ + closed: boolean; +} + /** Options for {@link Commands.run}. */ export interface RunOptions { /** Working directory (default: rootfs default). */ @@ -51,6 +84,23 @@ export interface RunOptions { timeoutMs?: number; /** Return a {@link CommandHandle} immediately instead of waiting for exit. */ background?: boolean; + /** + * Allocate a pseudo-terminal of this size. Output then arrives merged + * on the `"pty"` channel — stdout and stderr are indistinguishable + * once a terminal is allocated — and stdin stays open: end input by + * writing Ctrl-D (`"\x04"`) via {@link CommandHandle.writeStdin}; + * {@link CommandHandle.closeStdin} is rejected for PTY commands. + */ + pty?: PtySize; + /** + * Feed the command's stdin. A string (UTF-8) or bytes is written and + * then closed before the run resolves (subprocess semantics). `true` + * keeps stdin open for manual {@link CommandHandle.writeStdin} / + * {@link CommandHandle.closeStdin} — background runs only, since a + * foreground run cannot write while it waits. Unset: the process + * starts with stdin already at EOF. + */ + stdin?: string | Uint8Array | boolean; } /** One chunk of command output. */ @@ -169,17 +219,26 @@ export class CommandHandle { readonly #ctx: ClientContext; readonly #client: ProcessClient; readonly #sandboxId: string; + /** + * The next stdin write offset — advanced only on a successful write, + * so a retried write lands at the same offset and the daemon's + * deduplication makes it idempotent. `undefined` = unknown (a + * re-attached handle); resynced lazily via GetStdinStatus. + */ + #stdinOffset: bigint | undefined; constructor( ctx: ClientContext, client: ProcessClient, sandboxId: string, commandId: string, + stdinOffset: bigint | undefined = 0n, ) { this.#ctx = ctx; this.#client = client; this.#sandboxId = sandboxId; this.commandId = commandId; + this.#stdinOffset = stdinOffset; } /** @@ -187,7 +246,9 @@ export class CommandHandle { * earliest byte the daemon still retains (8 MiB per channel); replayed * buffered output comes first, then live output follows; the stream * ends when the process exits (deterministic termination — never - * silence). + * silence). A transport drop mid-stream re-attaches transparently + * from the last delivered offsets (see {@link ConnectionLostError} + * for the exhausted-retries case). */ get output(): AsyncIterable { return this.#streamOutput(); @@ -195,7 +256,7 @@ export class CommandHandle { async *#streamOutput(): AsyncGenerator { try { - for await (const event of this.#attach()) { + for await (const event of this.#attachEvents("commands.output")) { if (event.event.case === "output") { const chunk = event.event.value; yield { @@ -265,6 +326,107 @@ export class CommandHandle { } } + /** + * Write bytes (or a UTF-8 string) to the command's stdin. Requires a + * run started with `stdin: true` or a PTY. + * + * Writes are offset-idempotent: the handle tracks the stdin cursor and + * advances it only on success, so retrying a failed or lost write + * *with the same data* is safe — the daemon deduplicates bytes below + * its accepted count and never double-feeds the process. Issue writes + * sequentially; the handle tracks a single cursor. + */ + async writeStdin(data: string | Uint8Array): Promise { + const bytes = + typeof data === "string" ? new TextEncoder().encode(data) : data; + try { + const offset = await this.#stdinCursor(); + const status = await this.#client.writeStdin( + { + sandboxId: this.#sandboxId, + executionId: this.commandId, + offset, + data: bytes, + eof: false, + }, + unaryOptions(this.#ctx), + ); + this.#stdinOffset = status.bytesWritten; + } catch (error) { + throw toArcBoxError(error, "commands.writeStdin"); + } + } + + /** + * Close the command's stdin (EOF). Rejected for PTY commands — write + * Ctrl-D (`"\x04"`) instead. + */ + async closeStdin(): Promise { + try { + const offset = await this.#stdinCursor(); + await this.#client.writeStdin( + { + sandboxId: this.#sandboxId, + executionId: this.commandId, + offset, + data: new Uint8Array(), + eof: true, + }, + unaryOptions(this.#ctx), + ); + } catch (error) { + throw toArcBoxError(error, "commands.closeStdin"); + } + } + + /** + * The daemon's stdin acceptance state — the recovery point after a + * lost write response. Also resyncs the handle's write cursor. + */ + async stdinStatus(): Promise { + try { + const status = await this.#client.getStdinStatus( + { sandboxId: this.#sandboxId, executionId: this.commandId }, + unaryOptions(this.#ctx), + ); + this.#stdinOffset = status.bytesWritten; + return { + bytesWritten: Number(status.bytesWritten), + closed: status.closed, + }; + } catch (error) { + throw toArcBoxError(error, "commands.stdinStatus"); + } + } + + /** Resize a PTY command's terminal. */ + async resize(cols: number, rows: number): Promise { + try { + await this.#client.resizeExecutionTty( + { + sandboxId: this.#sandboxId, + executionId: this.commandId, + size: { width: cols, height: rows }, + }, + unaryOptions(this.#ctx), + ); + } catch (error) { + throw toArcBoxError(error, "commands.resize"); + } + } + + /** The tracked stdin cursor, resynced from the daemon when unknown. */ + async #stdinCursor(): Promise { + if (this.#stdinOffset === undefined) { + const status = await this.#client.getStdinStatus( + { sandboxId: this.#sandboxId, executionId: this.commandId }, + unaryOptions(this.#ctx), + ); + this.#stdinOffset = status.bytesWritten; + } + return this.#stdinOffset; + } + /** Deliver a signal to the whole process group (default SIGTERM). */ async kill(signal: SignalName = "SIGTERM"): Promise { try { @@ -281,13 +443,72 @@ export class CommandHandle { } } - #attach() { - return this.#client.attachExecution({ - sandboxId: this.#sandboxId, - executionId: this.commandId, - stdoutOffset: 0n, - stderrOffset: 0n, - }); + /** + * The resumable attach loop shared by {@link output} and the result + * collection: streams execution events, tracking the byte offset each + * channel has delivered. When the transport drops mid-stream, it + * re-attaches from those offsets — the daemon replays nothing already + * delivered — so the consumer sees one seamless, gapless stream. Only + * consecutive dead dials count against the retry budget (delivered + * output resets it); once exhausted, the stream-death + * {@link ConnectionLostError} carries the last transport failure. + */ + async *#attachEvents(operation: string): AsyncGenerator { + let stdoutOffset = 0n; + let stderrOffset = 0n; + let failures = 0; + for (;;) { + try { + // eslint-disable-next-line no-await-in-loop -- sequential by design: each re-attach resumes where the last stream died + for await (const event of this.#client.attachExecution({ + sandboxId: this.#sandboxId, + executionId: this.commandId, + stdoutOffset, + stderrOffset, + })) { + if (event.event.case === "output") { + const chunk = event.event.value; + const after = chunk.offset + BigInt(chunk.data.byteLength); + if (chunk.channel === StdioChannel.STDERR) { + if (after > stderrOffset) { + stderrOffset = after; + failures = 0; + } + } else if (after > stdoutOffset) { + stdoutOffset = after; + failures = 0; + } + } + yield event; + if (event.event.case === "exited") { + return; + } + } + // A clean server-side end without an exited frame: nothing more + // is coming (the daemon closed the record). + return; + } catch (error) { + const mapped = toArcBoxError(error, operation); + if (!(mapped instanceof ConnectionFailedError)) { + throw mapped; + } + failures += 1; + if (failures > MAX_ATTACH_RETRIES) { + throw new ConnectionLostError( + "the output stream died and could not be re-attached within " + + "the retry budget", + { + operation, + cause: error, + context: { + commandId: this.commandId, + retries: MAX_ATTACH_RETRIES_LABEL, + }, + }, + ); + } + } + } } /** @@ -303,7 +524,7 @@ export class CommandHandle { let nextStdout = 0n; let nextStderr = 0n; let truncated = false; - for await (const event of this.#attach()) { + for await (const event of this.#attachEvents("commands.waitForExit")) { if (event.event.case === "output") { const chunk = event.event.value; const isStderr = chunk.channel === StdioChannel.STDERR; @@ -365,13 +586,65 @@ export class Commands { cmd: string | string[], opts: RunOptions = {}, ): Promise { - const handle = await this.#start(cmd, opts); + const stdinData = + typeof opts.stdin === "string" + ? new TextEncoder().encode(opts.stdin) + : opts.stdin instanceof Uint8Array + ? opts.stdin + : undefined; + if (opts.stdin === true && opts.background !== true) { + throw new InvalidArgumentError( + "stdin: true keeps stdin open for the handle and requires " + + "background: true; pass a string or bytes to feed a foreground run", + { operation: "commands.run" }, + ); + } + if (stdinData !== undefined && opts.pty !== undefined) { + throw new InvalidArgumentError( + "a PTY's stdin cannot be closed after a one-shot write; use " + + "background: true with writeStdin, ending input with Ctrl-D (0x04)", + { operation: "commands.run" }, + ); + } + const handle = await this.#start(cmd, opts, stdinData !== undefined); + if (stdinData !== undefined) { + await this.#feedStdin(handle, stdinData); + } return opts.background === true ? handle : handle.waitForExit(); } + /** + * Re-attach to an execution by id — from another process, or after + * losing the handle. Verifies the execution exists (an unknown id is a + * typed not-found error) and seeds the handle's stdin cursor from the + * daemon's accepted count so later writes resume without a gap. + */ + async get(commandId: string): Promise { + try { + const execution = await this.#client.waitExecution( + { + sandboxId: this.#sandboxId, + executionId: commandId, + timeoutSeconds: 0, + }, + unaryOptions(this.#ctx), + ); + return new CommandHandle( + this.#ctx, + this.#client, + this.#sandboxId, + execution.id, + execution.stdin?.bytesWritten, + ); + } catch (error) { + throw toArcBoxError(error, "commands.get"); + } + } + async #start( cmd: string | string[], opts: RunOptions, + feedsStdin: boolean, ): Promise { try { // The execution id is minted client-side: a lost response leaves an @@ -387,7 +660,14 @@ export class Commands { user: opts.user ?? "", timeoutSeconds: opts.timeoutMs === undefined ? 0 : Math.ceil(opts.timeoutMs / 1000), - stdin: false, + // A PTY inherently keeps stdin open (EOF is not expressible on + // a terminal); otherwise stdin stays open exactly when the + // caller feeds or drives it. + stdin: opts.pty !== undefined || opts.stdin === true || feedsStdin, + tty: opts.pty !== undefined, + ...(opts.pty !== undefined && { + ttySize: { width: opts.pty.cols, height: opts.pty.rows }, + }), }, unaryOptions(this.#ctx), ); @@ -401,6 +681,47 @@ export class Commands { throw toArcBoxError(error, "commands.run"); } } + + /** + * Write-then-close the one-shot stdin payload. A process is free to + * exit without consuming its stdin (`subprocess` semantics): when the + * feed fails but the execution has already exited, the exit result is + * the truth and the failed feed is noise — the write merely raced the + * exit. Any other failure is real and surfaces. + */ + async #feedStdin(handle: CommandHandle, data: Uint8Array): Promise { + try { + await handle.writeStdin(data); + await handle.closeStdin(); + } catch (error) { + const mapped = toArcBoxError(error, "commands.run"); + let state: Execution | undefined; + try { + state = await this.#client.waitExecution( + { + sandboxId: this.#sandboxId, + executionId: handle.commandId, + timeoutSeconds: 0, + }, + unaryOptions(this.#ctx), + ); + } catch { + // The poll itself failed; the original feed error stands. + } + if (state?.state !== ExecutionState.EXITED) { + // The caller gets no handle out of a thrown run(), so a + // still-running process (cat waiting on input) would keep the + // sandbox RUNNING with no way to reach it. Best-effort kill; + // the feed error is the one to surface. + try { + await handle.kill("SIGKILL"); + } catch { + // Best-effort only. + } + throw mapped; + } + } + } } function channelName(channel: StdioChannel): "stdout" | "pty" { diff --git a/sdk/typescript/src/errors.ts b/sdk/typescript/src/errors.ts index 3388a2b8d..bc4893ecb 100644 --- a/sdk/typescript/src/errors.ts +++ b/sdk/typescript/src/errors.ts @@ -59,6 +59,15 @@ export class ConnectionFailedError extends ArcBoxError { name = "ConnectionFailedError"; } +/** + * A live stream died mid-flow — and, where the SDK re-attaches + * (command output), could not be re-established within the retry + * budget. The `cause` chain carries the underlying transport failure. + */ +export class ConnectionLostError extends ConnectionFailedError { + name = "ConnectionLostError"; +} + /** Authentication is required or was rejected. Reserved for the remote tier (CORE-63). */ export class AuthenticationError extends ArcBoxError { name = "AuthenticationError"; @@ -259,13 +268,23 @@ function classForConnectCode( /** * Whether the cause chain bottoms out in a connection-level syscall * failure. connect-node maps ECONNREFUSED to Code.Unavailable but leaves - * ENOENT — the missing-socket shape of "daemon not running" — as Unknown, - * so both are detected here directly. + * ENOENT — the missing-socket shape of "daemon not running" — as + * Unknown, so both are detected here directly. Mid-stream teardown + * shapes (ECONNRESET, EPIPE, ECONNABORTED) are the same family: the + * daemon stopped answering. */ function isConnectionRefused(reason: unknown): boolean { + const codes = new Set([ + "ENOENT", + "ECONNREFUSED", + "ENOTSOCK", + "ECONNRESET", + "ECONNABORTED", + "EPIPE", + ]); for (let cursor = reason; typeof cursor === "object" && cursor !== null; ) { const code = (cursor as { code?: unknown }).code; - if (code === "ENOENT" || code === "ECONNREFUSED" || code === "ENOTSOCK") { + if (typeof code === "string" && codes.has(code)) { return true; } cursor = (cursor as { cause?: unknown }).cause; diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts index 9ce06d3fa..1719cbf15 100644 --- a/sdk/typescript/src/index.ts +++ b/sdk/typescript/src/index.ts @@ -11,11 +11,18 @@ export { ArcBox, Sandbox } from "./sandbox"; export type { ConnectSandboxOptions, CreateSandboxOptions, + LifecycleUpdate, ListSandboxesOptions, } from "./sandbox"; export { CommandHandle, CommandResult, Commands } from "./commands"; -export type { CommandOutput, RunOptions, SignalName } from "./commands"; +export type { + CommandOutput, + PtySize, + RunOptions, + SignalName, + StdinStatus, +} from "./commands"; export { Files, MAX_FILE_BYTES } from "./files"; export type { WriteOptions } from "./files"; @@ -23,7 +30,11 @@ export type { WriteOptions } from "./files"; export type { ConnectionOptions } from "./connection"; export type { + Capabilities, IdlePolicy, + NestedVirtCapability, + SandboxEvent, + SandboxEventKind, SandboxInfo, SandboxState, SandboxSummary, @@ -37,6 +48,7 @@ export { CommandNotFoundError, CommandTimeoutError, ConnectionFailedError, + ConnectionLostError, FileNotFoundError, FileTooLargeError, InvalidArgumentError, diff --git a/sdk/typescript/src/sandbox.ts b/sdk/typescript/src/sandbox.ts index 15c02161a..9a4bb9b65 100644 --- a/sdk/typescript/src/sandbox.ts +++ b/sdk/typescript/src/sandbox.ts @@ -8,6 +8,8 @@ import { Commands } from "./commands"; import type { ConnectionOptions } from "./connection"; import { ArcBoxError, + ConnectionFailedError, + ConnectionLostError, InvalidArgumentError, NotFoundError, SandboxStateError, @@ -25,8 +27,17 @@ import { } from "./gen/arcbox/sandbox/v1/sandbox_pb"; import type { ClientContext } from "./transport"; import { createClientContext, unaryOptions } from "./transport"; -import type { SandboxInfo, SandboxState, SandboxSummary } from "./types"; +import type { + Capabilities, + IdlePolicy, + SandboxEvent, + SandboxInfo, + SandboxState, + SandboxSummary, +} from "./types"; import { + capabilitiesFromProto, + sandboxEventFromProto, sandboxInfoFromProto, sandboxStateToProto, sandboxSummaryFromProto, @@ -90,14 +101,38 @@ export interface ListSandboxesOptions { connection?: ConnectionOptions; } +/** + * A lifecycle-deadline update for {@link Sandbox.setLifecycle}. Each + * knob is tri-state: **omitted** (or `undefined`) leaves it unchanged; + * **`null`** restores the daemon default (no TTL / no idle detection / + * the default idle action); a **value** replaces it. + */ +export interface LifecycleUpdate { + /** + * Replace the hard maximum lifetime: expire this long from NOW — + * calling repeatedly keeps a busy sandbox alive (E2B timeout + * semantics). `null` removes the limit. + */ + ttlMs?: number | null; + /** + * Replace the idle window, re-arming a live timer. `null` disables + * idle detection. + */ + idleTimeoutMs?: number | null; + /** + * Replace what the daemon does when the idle timeout expires. + * `null` restores the daemon default (currently `"kill"`). + */ + onIdle?: IdlePolicy | null; +} + type SandboxClient = Client; /** - * How often {@link ArcBox.connect} re-inspects a PAUSING sandbox. - * `SANDBOX_EVENT_KIND_PAUSED` names this edge in the proto, but the - * daemon does not emit it yet (Pause/Resume are CORE-21 stubs), so the - * checkpoint is polled out instead. Once it is emitted, this poll - * should become an event wait. + * How often {@link ArcBox.connect} re-inspects a PAUSING sandbox. The + * daemon emits `SANDBOX_EVENT_KIND_PAUSED` on this edge (CORE-21), but + * the settle poll predates it and remains the simple, robust route — a + * poll-to-event-wait conversion is a candidate cleanup, not a bug. */ const PAUSE_SETTLE_POLL_MS = 500; @@ -123,12 +158,36 @@ function withSignal( export class ArcBox { readonly #ctx: ClientContext; readonly #client: SandboxClient; + #capabilities?: Promise; constructor(options: ConnectionOptions = {}) { this.#ctx = createClientContext(options); this.#client = createClient(SandboxService, this.#ctx.transport); } + /** + * What the daemon can do: version, sandbox protocol level, feature + * flags, and whether nested virtualization is available. Answered + * host-side (works before any sandbox exists) and cached for the life + * of this client — a failed fetch is not cached, so the next call + * retries. The SDK does not gate on it: the daemon fails fast on its + * own (a `CapabilityError` from `create`); this is the inspectable + * version of the same answer. + */ + capabilities(): Promise { + this.#capabilities ??= this.#fetchCapabilities().catch((error: unknown) => { + this.#capabilities = undefined; + throw toArcBoxError(error, "arcbox.capabilities"); + }); + return this.#capabilities; + } + + async #fetchCapabilities(): Promise { + return capabilitiesFromProto( + await this.#client.getCapabilities({}, unaryOptions(this.#ctx)), + ); + } + /** * Create a sandbox and (by default) wait until it is READY. * @@ -527,11 +586,8 @@ export class Sandbox { * Checkpoint the sandbox to disk under the same id and release its * runtime resources. Resume happens on the next {@link Sandbox.connect} * (or transparently, daemon-side, on the next data-plane call). Trades - * RAM for disk: a paused sandbox keeps paying `storageBytes`. - * - * Requires daemon-side CORE-21: the current local daemon serves - * Pause/Resume as contract-only stubs, so this rejects with an - * Unimplemented {@link ArcBoxError} until that lands. + * RAM for disk: a paused sandbox keeps paying `storageBytes`. Requires + * a quiescent sandbox (READY — no running command). */ async pause(): Promise { try { @@ -542,6 +598,84 @@ export class Sandbox { } } + /** + * Replace lifecycle deadlines. Each knob is tri-state (see + * {@link LifecycleUpdate}): omitted = unchanged, `null` = restore the + * daemon default, a value = replace. `ttlMs` re-arms the hard cap + * from NOW; `idleTimeoutMs` re-arms a live idle timer. Works in any + * non-terminal state, including paused. + */ + async setLifecycle(update: LifecycleUpdate): Promise { + try { + await this.#client.setLifecycle( + { + id: this.id, + ...(update.ttlMs !== undefined && { + ttlSeconds: update.ttlMs === null ? 0 : secondsFromMs(update.ttlMs), + }), + ...(update.idleTimeoutMs !== undefined && { + idleTimeoutSeconds: + update.idleTimeoutMs === null + ? 0 + : secondsFromMs(update.idleTimeoutMs), + }), + ...(update.onIdle !== undefined && { + onIdle: + update.onIdle === null + ? IdleAction.UNSPECIFIED + : update.onIdle === "kill" + ? IdleAction.KILL + : IdleAction.PAUSE, + }), + }, + unaryOptions(this.#ctx), + ); + } catch (error) { + throw toArcBoxError(error, "sandbox.setLifecycle"); + } + } + + /** + * Subscribe to this sandbox's lifecycle events, yielded as typed + * {@link SandboxEvent}s (keepalive frames are filtered out). The + * iterator ends when the daemon ends the stream; breaking out of the + * loop cancels the subscription. A transport drop mid-stream is + * surfaced as {@link ConnectionLostError} — re-subscribing is the + * caller's decision, since missed events cannot be replayed. + */ + events(): AsyncIterable { + return this.#streamEvents(); + } + + async *#streamEvents(): AsyncGenerator { + let delivered = false; + try { + for await (const frame of this.#client.events({ sandboxId: this.id })) { + delivered = true; + if (frame.payload.case === "event") { + yield sandboxEventFromProto(frame.payload.value); + } + } + } catch (error) { + const mapped = toArcBoxError(error, "sandbox.events"); + // A connection failure after frames flowed is a mid-stream drop + // (the stream-death error); before any frame it is an unreachable + // daemon, reported as such. + if ( + delivered && + mapped instanceof ConnectionFailedError && + !(mapped instanceof ConnectionLostError) + ) { + throw new ConnectionLostError("the event stream died", { + operation: "sandbox.events", + cause: error, + context: { id: this.id }, + }); + } + throw mapped; + } + } + /** * `await using` disposal: kill the sandbox, so a leaked handle never * leaks a VM. Swallows only "already gone" — the whole NotFoundError diff --git a/sdk/typescript/src/types.ts b/sdk/typescript/src/types.ts index b5954927f..20ad0d0c1 100644 --- a/sdk/typescript/src/types.ts +++ b/sdk/typescript/src/types.ts @@ -1,11 +1,14 @@ import { timestampDate } from "@bufbuild/protobuf/wkt"; import type { + GetCapabilitiesResponse, + SandboxEvent as SandboxEventProto, SandboxInfo as SandboxInfoProto, SandboxSummary as SandboxSummaryProto, } from "./gen/arcbox/sandbox/v1/sandbox_pb"; import { IdleAction, + SandboxEventKind as SandboxEventKindProto, SandboxState as SandboxStateProto, } from "./gen/arcbox/sandbox/v1/sandbox_pb"; @@ -52,6 +55,59 @@ export interface SandboxInfo { storageBytes: number; } +/** + * Kind of a sandbox lifecycle event. `"idle"` fires when an execution + * exits and the sandbox returns to ready; `"pausing"`/`"resumed"` carry + * a `reason` attribute distinguishing client calls from automation + * (`idle_timeout` / `auto_resume`). + */ +export type SandboxEventKind = + | "created" + | "ready" + | "running" + | "idle" + | "stopping" + | "stopped" + | "failed" + | "removed" + | "pausing" + | "paused" + | "resumed" + | "unknown"; + +/** One sandbox lifecycle event, as delivered by {@link Sandbox.events}. */ +export interface SandboxEvent { + sandboxId: string; + kind: SandboxEventKind; + /** When it happened (daemon clock). */ + time?: Date; + /** + * Per-kind context: `exit_code`/`signal` on `"idle"`, `error` on + * `"failed"`, `reason` on `"pausing"`/`"resumed"`. + */ + attributes: Record; +} + +/** Nested-virtualization support on this host. */ +export interface NestedVirtCapability { + /** True when sandboxes can run (M3+ hardware, VZ backend). */ + supported: boolean; + /** The daemon's authoritative reason, when unsupported. */ + reason?: string; +} + +/** What the daemon can do — the {@link ArcBox.capabilities} handshake. */ +export interface Capabilities { + /** Daemon version string (informational). */ + daemonVersion: string; + /** Sandbox API protocol level. */ + protocol: number; + /** Append-only named feature flags (e.g. "pause_resume"). */ + features: string[]; + /** Whether this host can run sandboxes at all. */ + nestedVirt: NestedVirtCapability; +} + /** One row of a sandbox listing. */ export interface SandboxSummary { id: string; @@ -170,3 +226,49 @@ function assignIfSet( target[key] = value; } } + +const EVENT_KIND_NAMES: Partial< + Record +> = { + [SandboxEventKindProto.CREATED]: "created", + [SandboxEventKindProto.READY]: "ready", + [SandboxEventKindProto.RUNNING]: "running", + [SandboxEventKindProto.IDLE]: "idle", + [SandboxEventKindProto.STOPPING]: "stopping", + [SandboxEventKindProto.STOPPED]: "stopped", + [SandboxEventKindProto.FAILED]: "failed", + [SandboxEventKindProto.REMOVED]: "removed", + [SandboxEventKindProto.PAUSING]: "pausing", + [SandboxEventKindProto.PAUSED]: "paused", + [SandboxEventKindProto.RESUMED]: "resumed", +}; + +/** Map one Events frame to the public DTO ("unknown" for kinds this SDK predates). */ +export function sandboxEventFromProto(event: SandboxEventProto): SandboxEvent { + const out: SandboxEvent = { + sandboxId: event.sandboxId, + kind: EVENT_KIND_NAMES[event.kind] ?? "unknown", + attributes: event.attributes, + }; + assignIfSet(out, "time", optionalDate(event.time)); + return out; +} + +/** Map the GetCapabilities response to the public DTO. */ +export function capabilitiesFromProto( + response: GetCapabilitiesResponse, +): Capabilities { + const nestedVirt: NestedVirtCapability = { + supported: response.nestedVirt?.supported ?? false, + }; + const reason = response.nestedVirt?.reason; + if (reason !== undefined && reason !== "") { + nestedVirt.reason = reason; + } + return { + daemonVersion: response.daemonVersion, + protocol: response.protocol, + features: response.features, + nestedVirt, + }; +} diff --git a/sdk/typescript/test/e2e.test.ts b/sdk/typescript/test/e2e.test.ts index 316da3433..51bdb0efa 100644 --- a/sdk/typescript/test/e2e.test.ts +++ b/sdk/typescript/test/e2e.test.ts @@ -5,17 +5,26 @@ // Connection resolution applies, so ARCBOX_SOCKET / ARCBOX_DATA_DIR // point the loop at a dev daemon. // -// This is the design doc's 20-line hello world minus the parts outside -// phase 1's surface: ports.expose and waitForPort are deferred, so the -// background command is observed through its output stream and -// waitForExit instead of a port probe. +// This is the design doc's 20-line hello world plus the phase 2a +// surface: PTY, stdin, re-attach, setLifecycle, capabilities, and +// events. Still outside scope: ports.expose and waitForPort (2b). import { describe, expect, it } from "vitest"; -import { Sandbox } from "../src/index"; +import { ArcBox, Sandbox } from "../src/index"; const enabled = process.env.ARCBOX_SDK_E2E === "1"; describe.skipIf(!enabled)("hello world against a live daemon", () => { + it("answers the capabilities handshake", async () => { + // This suite only runs on sandbox-capable hosts, so nested_virt + // must report supported — the same answer create() relies on. + const caps = await new ArcBox().capabilities(); + expect(caps.protocol).toBeGreaterThanOrEqual(1); + expect(caps.daemonVersion).not.toBe(""); + expect(caps.features).toContain("pause_resume"); + expect(caps.nestedVirt.supported).toBe(true); + }, 60000); + it("creates a sandbox, moves files, runs commands, and cleans up", async () => { // Built-in minimal template (busybox) — no image pull involved. const sandbox = await Sandbox.create("", { ttlMs: 300000 }); @@ -60,6 +69,49 @@ describe.skipIf(!enabled)("hello world against a live daemon", () => { expect(killed.signal).toBe("SIGKILL"); expect(killed.exitCode).toBe(137); + // stdin: foreground write-then-close (subprocess semantics). + const echoed = await sandbox.commands.run(["/bin/cat"], { + stdin: "hello stdin\n", + }); + expect(echoed.expect().stdout).toBe("hello stdin\n"); + + // stdin: a background handle drives offset-idempotent writes. + const catBg = await sandbox.commands.run(["/bin/cat"], { + background: true, + stdin: true, + }); + await catBg.writeStdin("first "); + await catBg.writeStdin("second\n"); + await catBg.closeStdin(); + expect((await catBg.waitForExit(30000)).stdout).toBe("first second\n"); + + // commands.get: re-attach by id; the retained output replays. + const again = await sandbox.commands.get(catBg.commandId); + expect((await again.waitForExit(30000)).stdout).toBe("first second\n"); + + // PTY: stty reads the allocated terminal's geometry (rows cols), + // and output arrives merged (a pty run's stdout carries it all). + const tty = await sandbox.commands.run("stty size", { + pty: { cols: 120, rows: 40 }, + }); + expect(tty.expect().stdout).toContain("40 120"); + + // PTY resize: the running terminal observes the new geometry. + const resized = await sandbox.commands.run("sleep 2; stty size", { + pty: { cols: 80, rows: 24 }, + background: true, + }); + await resized.resize(200, 50); + expect((await resized.waitForExit(30000)).stdout).toContain("50 200"); + + // setLifecycle tri-state: re-arm the TTL, then remove it (null), + // then re-arm again so the sandbox cannot outlive a crash here. + await sandbox.setLifecycle({ ttlMs: 600000 }); + expect((await sandbox.info()).ttlDeadline).toBeDefined(); + await sandbox.setLifecycle({ ttlMs: null }); + expect((await sandbox.info()).ttlDeadline).toBeUndefined(); + await sandbox.setLifecycle({ ttlMs: 300000 }); + // info() is always fresh. const info = await sandbox.info(); expect(["ready", "running"]).toContain(info.state); @@ -67,4 +119,33 @@ describe.skipIf(!enabled)("hello world against a live daemon", () => { await sandbox.kill(); } }, 300000); + + it("events() observes the idle auto-pause", async () => { + // A short idle timeout with the PAUSE policy: the daemon must emit + // PAUSING (reason idle_timeout) then PAUSED on the events stream. + const sandbox = await Sandbox.create("", { + ttlMs: 300000, + idleTimeoutMs: 4000, + onIdle: "pause", + }); + try { + const kinds: string[] = []; + let reason = ""; + for await (const event of sandbox.events()) { + kinds.push(event.kind); + if (event.kind === "pausing") { + reason = event.attributes.reason ?? ""; + } + if (event.kind === "paused") { + break; + } + } + expect(kinds).toContain("pausing"); + expect(kinds.at(-1)).toBe("paused"); + expect(reason).toBe("idle_timeout"); + expect((await sandbox.info()).state).toBe("paused"); + } finally { + await sandbox.kill(); + } + }, 300000); }); diff --git a/sdk/typescript/test/lifecycle.test.ts b/sdk/typescript/test/lifecycle.test.ts new file mode 100644 index 000000000..fdb63b554 --- /dev/null +++ b/sdk/typescript/test/lifecycle.test.ts @@ -0,0 +1,213 @@ +// events(), setLifecycle(), and the capabilities handshake against a +// mock daemon. +// +// setLifecycle's tri-state is the contract that matters: an omitted +// knob must be ABSENT on the wire (unchanged), null must be an explicit +// zero/UNSPECIFIED (restore the default), and a value must replace. +// Getting presence wrong silently rewrites deadlines the caller never +// touched. + +import { create } from "@bufbuild/protobuf"; +import { EmptySchema } from "@bufbuild/protobuf/wkt"; +import type { Transport } from "@connectrpc/connect"; +import { Code, ConnectError, createRouterTransport } from "@connectrpc/connect"; +import { describe, expect, it } from "vitest"; + +import { ConnectionLostError, NotFoundError } from "../src/errors"; +import type { SetLifecycleRequest } from "../src/gen/arcbox/sandbox/v1/sandbox_pb"; +import { + GetCapabilitiesResponseSchema, + IdleAction, + KeepAliveSchema, + SandboxEventKind, + SandboxEventSchema, + SandboxService, + WatchEventsResponseSchema, +} from "../src/gen/arcbox/sandbox/v1/sandbox_pb"; +import { ArcBox, Sandbox } from "../src/sandbox"; + +function eventFrame( + kind: SandboxEventKind, + attributes: Record = {}, +) { + return create(WatchEventsResponseSchema, { + payload: { + case: "event", + value: create(SandboxEventSchema, { + sandboxId: "sb-1", + kind, + attributes, + }), + }, + }); +} + +const keepAlive = create(WatchEventsResponseSchema, { + payload: { case: "keepAlive", value: create(KeepAliveSchema) }, +}); + +/** A Sandbox handle on a mock transport (no lifecycle routing involved). */ +function sandboxOn(transport: Transport): Sandbox { + return new Sandbox({ transport }, "sb-1"); +} + +describe("sandbox.events", () => { + it("yields typed events and filters keepalives; a clean end ends the loop", async () => { + const mock = createRouterTransport(({ service }) => { + service(SandboxService, { + // eslint-disable-next-line @typescript-eslint/require-await -- a server-streaming handler is an async generator + async *events(req) { + expect(req.sandboxId).toBe("sb-1"); + yield keepAlive; + yield eventFrame(SandboxEventKind.CREATED); + yield eventFrame(SandboxEventKind.READY); + yield keepAlive; + yield eventFrame(SandboxEventKind.PAUSING, { + reason: "idle_timeout", + }); + yield eventFrame(SandboxEventKind.PAUSED); + }, + }); + }); + const sandbox = sandboxOn(mock); + const seen: string[] = []; + let reason = ""; + for await (const event of sandbox.events()) { + seen.push(event.kind); + if (event.kind === "pausing") { + reason = event.attributes.reason ?? ""; + } + } + expect(seen).toEqual(["created", "ready", "pausing", "paused"]); + expect(reason).toBe("idle_timeout"); + }); + + it("a drop after frames flowed is the stream-death error", async () => { + const mock = createRouterTransport(({ service }) => { + service(SandboxService, { + // eslint-disable-next-line @typescript-eslint/require-await -- a server-streaming handler is an async generator + async *events() { + yield eventFrame(SandboxEventKind.READY); + throw new ConnectError("stream reset", Code.Unavailable); + }, + }); + }); + const sandbox = sandboxOn(mock); + const consume = async () => { + for await (const event of sandbox.events()) { + void event; + } + }; + await expect(consume()).rejects.toBeInstanceOf(ConnectionLostError); + }); + + it("a daemon-typed stream error keeps its own class", async () => { + const mock = createRouterTransport(({ service }) => { + service(SandboxService, { + // eslint-disable-next-line @typescript-eslint/require-await -- a server-streaming handler is an async generator + async *events() { + yield eventFrame(SandboxEventKind.READY); + throw new ConnectError("gone", Code.NotFound); + }, + }); + }); + const sandbox = sandboxOn(mock); + const consume = async () => { + for await (const event of sandbox.events()) { + void event; + } + }; + await expect(consume()).rejects.toBeInstanceOf(NotFoundError); + }); +}); + +function lifecycleProbe() { + const requests: SetLifecycleRequest[] = []; + const mock = createRouterTransport(({ service }) => { + service(SandboxService, { + setLifecycle(req) { + requests.push(req); + return create(EmptySchema); + }, + }); + }); + return { sandbox: sandboxOn(mock), requests }; +} + +describe("sandbox.setLifecycle tri-state", () => { + it("an empty update sends every knob absent (all unchanged)", async () => { + const { sandbox, requests } = lifecycleProbe(); + await sandbox.setLifecycle({}); + expect(requests[0]?.ttlSeconds).toBeUndefined(); + expect(requests[0]?.idleTimeoutSeconds).toBeUndefined(); + expect(requests[0]?.onIdle).toBeUndefined(); + }); + + it("values replace: ttl re-arms from now, idle window swaps", async () => { + const { sandbox, requests } = lifecycleProbe(); + await sandbox.setLifecycle({ ttlMs: 5000, idleTimeoutMs: 30000 }); + expect(requests[0]?.ttlSeconds).toBe(5); + expect(requests[0]?.idleTimeoutSeconds).toBe(30); + expect(requests[0]?.onIdle).toBeUndefined(); + }); + + it("null restores the default: explicit zero / UNSPECIFIED on the wire", async () => { + const { sandbox, requests } = lifecycleProbe(); + await sandbox.setLifecycle({ ttlMs: null, onIdle: null }); + expect(requests[0]?.ttlSeconds).toBe(0); + expect(requests[0]?.idleTimeoutSeconds).toBeUndefined(); + expect(requests[0]?.onIdle).toBe(IdleAction.UNSPECIFIED); + }); + + it("onIdle maps the policy names", async () => { + const { sandbox, requests } = lifecycleProbe(); + await sandbox.setLifecycle({ onIdle: "pause" }); + expect(requests[0]?.onIdle).toBe(IdleAction.PAUSE); + await sandbox.setLifecycle({ onIdle: "kill" }); + expect(requests[1]?.onIdle).toBe(IdleAction.KILL); + }); +}); + +function capsTransport(counter: { calls: number }, failFirst = false) { + return createRouterTransport(({ service }) => { + service(SandboxService, { + getCapabilities() { + counter.calls += 1; + if (failFirst && counter.calls === 1) { + throw new ConnectError("starting up", Code.Unavailable); + } + return create(GetCapabilitiesResponseSchema, { + daemonVersion: "0.9.0", + protocol: 1, + features: ["pause_resume", "auto_resume"], + nestedVirt: { supported: false, reason: "requires M3 or newer" }, + }); + }, + }); + }); +} + +describe("arcbox.capabilities", () => { + it("maps the handshake and caches it per client", async () => { + const counter = { calls: 0 }; + const box = new ArcBox({ transport: capsTransport(counter) }); + const caps = await box.capabilities(); + expect(caps).toEqual({ + daemonVersion: "0.9.0", + protocol: 1, + features: ["pause_resume", "auto_resume"], + nestedVirt: { supported: false, reason: "requires M3 or newer" }, + }); + await box.capabilities(); + expect(counter.calls).toBe(1); + }); + + it("does not cache a failed fetch", async () => { + const counter = { calls: 0 }; + const box = new ArcBox({ transport: capsTransport(counter, true) }); + await expect(box.capabilities()).rejects.toThrow("starting up"); + const caps = await box.capabilities(); + expect(caps.protocol).toBe(1); + expect(counter.calls).toBe(2); + }); +}); diff --git a/sdk/typescript/test/reattach.test.ts b/sdk/typescript/test/reattach.test.ts new file mode 100644 index 000000000..d36711e3a --- /dev/null +++ b/sdk/typescript/test/reattach.test.ts @@ -0,0 +1,211 @@ +// Offset-resume across stream death against a mock daemon. +// +// The contract under test: when an attach stream drops mid-flow, the +// handle re-attaches from the last DELIVERED per-channel offsets and the +// consumer sees one seamless, gapless stream — the SDK's whole reason +// for offset-addressed output. Retries are bounded by consecutive dead +// dials; delivered output resets the budget. + +import { create } from "@bufbuild/protobuf"; +import type { Transport } from "@connectrpc/connect"; +import { Code, ConnectError, createRouterTransport } from "@connectrpc/connect"; +import { describe, expect, it } from "vitest"; + +import { Commands } from "../src/commands"; +import { ConnectionLostError, NotFoundError } from "../src/errors"; +import type { AttachExecutionRequest } from "../src/gen/arcbox/sandbox/v1/process_pb"; +import { + ExecutionEventSchema, + ExecutionSchema, + ExecutionState, + SandboxProcessService, + StdioChannel, +} from "../src/gen/arcbox/sandbox/v1/process_pb"; + +interface Chunk { + channel: StdioChannel; + offset: bigint; + text: string; +} + +const exited = create(ExecutionSchema, { + id: "cmd", + state: ExecutionState.EXITED, + exitStatus: { status: { case: "code", value: 0 } }, +}); + +/** + * Serves AttachExecution from a chunk script, killing the stream after + * `dieAfter[n]` chunks on the n-th attach (die forever once the script + * runs out). Replays only chunks at or past the requested offset, like + * the daemon. + */ +class FlakyDaemon { + attaches: AttachExecutionRequest[] = []; + + constructor( + public chunks: Chunk[], + public dieAfter: number[], + ) {} + + transport(): Transport { + // Locals captured by the handlers below; the arrays are shared refs. + const { attaches, chunks, dieAfter } = this; + return createRouterTransport(({ service }) => { + service(SandboxProcessService, { + startExecution: (req) => + create(ExecutionSchema, { + id: req.executionId, + state: ExecutionState.RUNNING, + }), + waitExecution: () => exited, + // eslint-disable-next-line @typescript-eslint/require-await -- a server-streaming handler is an async generator; this double serves from memory + async *attachExecution(req) { + const call = attaches.length; + attaches.push(req); + const budget = dieAfter[call]; + let sent = 0; + for (let i = 0, len = chunks.length; i < len; i++) { + const chunk = chunks[i]; + if (chunk === undefined) { + continue; + } + const isStderr = chunk.channel === StdioChannel.STDERR; + const from = isStderr ? req.stderrOffset : req.stdoutOffset; + if (chunk.offset < from) { + continue; + } + if (budget !== undefined && sent >= budget) { + throw new ConnectError("stream reset", Code.Unavailable); + } + yield create(ExecutionEventSchema, { + event: { + case: "output", + value: { + channel: chunk.channel, + offset: chunk.offset, + data: new TextEncoder().encode(chunk.text), + }, + }, + }); + sent += 1; + } + if (budget !== undefined) { + throw new ConnectError("stream reset", Code.Unavailable); + } + yield create(ExecutionEventSchema, { + event: { case: "exited", value: { execution: exited } }, + }); + }, + }); + }); + } + + commands(): Commands { + return new Commands({ transport: this.transport() }, "sb-1"); + } +} + +const script: Chunk[] = [ + { channel: StdioChannel.STDOUT, offset: 0n, text: "hel" }, + { channel: StdioChannel.STDERR, offset: 0n, text: "warn" }, + { channel: StdioChannel.STDOUT, offset: 3n, text: "lo " }, + { channel: StdioChannel.STDOUT, offset: 6n, text: "world" }, +]; + +describe("offset-resume", () => { + it("the output iterator re-attaches from the delivered offsets", async () => { + // First attach dies after two chunks (stdout "hel" + stderr "warn"). + const daemon = new FlakyDaemon(script, [2]); + const handle = await daemon.commands().run("emit", { background: true }); + let stdout = ""; + let stderr = ""; + for await (const chunk of handle.output) { + const text = new TextDecoder().decode(chunk.data); + if (chunk.channel === "stderr") { + stderr += text; + } else { + stdout += text; + } + } + // Seamless and gapless despite the mid-stream death. + expect(stdout).toBe("hello world"); + expect(stderr).toBe("warn"); + expect(daemon.attaches).toHaveLength(2); + // The re-attach resumed exactly at the delivered high-water marks. + expect(daemon.attaches[1]?.stdoutOffset).toBe(3n); + expect(daemon.attaches[1]?.stderrOffset).toBe(4n); + }); + + it("survives repeated drops as long as each dial delivers output", async () => { + // Every attach dies after one delivered chunk; four dials complete + // the script. Progress resets the retry budget each time. + const daemon = new FlakyDaemon(script, [1, 1, 1, 1]); + const handle = await daemon.commands().run("emit", { background: true }); + let stdout = ""; + for await (const chunk of handle.output) { + if (chunk.channel !== "stderr") { + stdout += new TextDecoder().decode(chunk.data); + } + } + expect(stdout).toBe("hello world"); + expect(daemon.attaches).toHaveLength(5); + }); + + it("waitForExit's result collection resumes through the same loop", async () => { + const daemon = new FlakyDaemon(script, [3]); + const handle = await daemon.commands().run("emit", { background: true }); + const result = await handle.waitForExit(1000); + expect(result.stdout).toBe("hello world"); + expect(result.stderr).toBe("warn"); + // The resumed chunks were contiguous — no false truncation flag. + expect(result.truncated).toBe(false); + }); + + it("bounded retries: dead dials exhaust into ConnectionLostError", async () => { + // Every dial dies before delivering anything. + const daemon = new FlakyDaemon(script, [0, 0, 0, 0, 0, 0, 0, 0]); + const handle = await daemon.commands().run("emit", { background: true }); + const consume = async () => { + for await (const chunk of handle.output) { + void chunk; + } + }; + await expect(consume()).rejects.toBeInstanceOf(ConnectionLostError); + // The initial dial plus MAX_ATTACH_RETRIES re-dials. + expect(daemon.attaches).toHaveLength(4); + }); + + it("a daemon-typed stream error is surfaced, never retried", async () => { + let attaches = 0; + const transport = createRouterTransport(({ service }) => { + service(SandboxProcessService, { + waitExecution: () => exited, + // eslint-disable-next-line @typescript-eslint/require-await -- a server-streaming handler is an async generator + async *attachExecution() { + attaches += 1; + yield create(ExecutionEventSchema, { + event: { + case: "output", + value: { + channel: StdioChannel.STDOUT, + offset: 0n, + data: new Uint8Array([120]), + }, + }, + }); + throw new ConnectError("no such execution", Code.NotFound); + }, + }); + }); + const commands = new Commands({ transport }, "sb-1"); + const handle = await commands.get("cmd"); + const consume = async () => { + for await (const chunk of handle.output) { + void chunk; + } + }; + await expect(consume()).rejects.toBeInstanceOf(NotFoundError); + expect(attaches).toBe(1); + }); +}); diff --git a/sdk/typescript/test/stdin-pty.test.ts b/sdk/typescript/test/stdin-pty.test.ts new file mode 100644 index 000000000..4b6196ea1 --- /dev/null +++ b/sdk/typescript/test/stdin-pty.test.ts @@ -0,0 +1,284 @@ +// PTY and stdin against a mock daemon. +// +// The invariant that matters: stdin writes are offset-idempotent. The +// handle advances its cursor only on a successful response, so a retry +// of a lost write lands at the SAME offset and the daemon's +// deduplication swallows the duplicate — never a double feed. + +import { create } from "@bufbuild/protobuf"; +import { EmptySchema } from "@bufbuild/protobuf/wkt"; +import type { Transport } from "@connectrpc/connect"; +import { Code, ConnectError, createRouterTransport } from "@connectrpc/connect"; +import { describe, expect, it } from "vitest"; + +import { Commands } from "../src/commands"; +import { ConnectionFailedError, InvalidArgumentError } from "../src/errors"; +import type { + StartExecutionRequest, + TerminalSize, + WriteStdinRequest, +} from "../src/gen/arcbox/sandbox/v1/process_pb"; +import { + ExecutionSchema, + ExecutionState, + SandboxProcessService, + Signal, + StdinStatusSchema, +} from "../src/gen/arcbox/sandbox/v1/process_pb"; +import type { ClientContext } from "../src/transport"; + +/** A process-service double tracking stdin acceptance like the guest. */ +class MockDaemon { + starts: StartExecutionRequest[] = []; + writes: WriteStdinRequest[] = []; + resizes: TerminalSize[] = []; + accepted = 0n; + closed = false; + /** Accept the next write, then fail its response (a lost response). */ + failNextWriteResponse = false; + /** Execution state served by WaitExecution polls. */ + waitState = ExecutionState.EXITED; + /** Throw this from WriteStdin without accepting anything. */ + rejectWrites?: ConnectError; + + /** Signals delivered via SignalExecution. */ + signals: number[] = []; + + transport(): Transport { + return createRouterTransport(({ service }) => { + service(SandboxProcessService, { + startExecution: (req) => { + this.starts.push(req); + return create(ExecutionSchema, { + id: req.executionId, + state: ExecutionState.RUNNING, + }); + }, + signalExecution: (req) => { + this.signals.push(req.signal); + return create(EmptySchema); + }, + writeStdin: (req) => { + this.writes.push(req); + if (this.rejectWrites !== undefined) { + throw this.rejectWrites; + } + if (req.offset > this.accepted) { + throw new ConnectError("stdin gap", Code.OutOfRange); + } + // Deduplicate bytes below the accepted count (the guest contract). + const fresh = + req.offset + BigInt(req.data.byteLength) - this.accepted; + if (fresh > 0n) { + this.accepted += fresh; + } + if (req.eof) { + this.closed = true; + } + if (this.failNextWriteResponse) { + this.failNextWriteResponse = false; + throw new ConnectError("connection reset", Code.Unavailable); + } + return create(StdinStatusSchema, { + bytesWritten: this.accepted, + closed: this.closed, + }); + }, + getStdinStatus: () => + create(StdinStatusSchema, { + bytesWritten: this.accepted, + closed: this.closed, + }), + resizeExecutionTty: (req) => { + if (req.size !== undefined) { + this.resizes.push(req.size); + } + return create(EmptySchema); + }, + waitExecution: (req) => + create(ExecutionSchema, { + id: req.executionId, + state: this.waitState, + ...(this.waitState === ExecutionState.EXITED && { + exitStatus: { status: { case: "code" as const, value: 0 } }, + }), + }), + // eslint-disable-next-line @typescript-eslint/require-await -- a server-streaming handler is an async generator; this double serves from memory + async *attachExecution() { + yield { + event: { + case: "exited" as const, + value: { + execution: create(ExecutionSchema, { + state: ExecutionState.EXITED, + exitStatus: { status: { case: "code" as const, value: 0 } }, + }), + }, + }, + }; + }, + }); + }); + } + + commands(): Commands { + const ctx: ClientContext = { transport: this.transport() }; + return new Commands(ctx, "sb-1"); + } +} + +describe("PTY", () => { + it("run(pty) starts a TTY execution of that geometry with stdin open", async () => { + const daemon = new MockDaemon(); + await daemon.commands().run("stty size", { pty: { cols: 120, rows: 40 } }); + const start = daemon.starts[0]; + expect(start?.tty).toBe(true); + expect(start?.ttySize).toMatchObject({ width: 120, height: 40 }); + expect(start?.stdin).toBe(true); + }); + + it("a plain run keeps stdin at EOF and allocates no TTY", async () => { + const daemon = new MockDaemon(); + await daemon.commands().run("true"); + const start = daemon.starts[0]; + expect(start?.tty).toBe(false); + expect(start?.stdin).toBe(false); + }); + + it("resize() sends the new geometry", async () => { + const daemon = new MockDaemon(); + const handle = await daemon + .commands() + .run("sh", { pty: { cols: 80, rows: 24 }, background: true }); + await handle.resize(200, 50); + expect(daemon.resizes[0]).toMatchObject({ width: 200, height: 50 }); + }); + + it("rejects pty combined with one-shot stdin data before any RPC", async () => { + const daemon = new MockDaemon(); + await expect( + daemon.commands().run("cat", { pty: { cols: 80, rows: 24 }, stdin: "x" }), + ).rejects.toBeInstanceOf(InvalidArgumentError); + expect(daemon.starts).toHaveLength(0); + }); +}); + +describe("stdin", () => { + it("tracks the write cursor across writes and closes at it", async () => { + const daemon = new MockDaemon(); + const handle = await daemon + .commands() + .run("cat", { background: true, stdin: true }); + await handle.writeStdin("hello "); + await handle.writeStdin("world\n"); + await handle.closeStdin(); + expect(daemon.writes.map((w) => w.offset)).toEqual([0n, 6n, 12n]); + expect(daemon.writes[2]?.eof).toBe(true); + expect(daemon.accepted).toBe(12n); + expect(daemon.closed).toBe(true); + }); + + it("a retried lost write never double-feeds (offset idempotency)", async () => { + const daemon = new MockDaemon(); + const handle = await daemon + .commands() + .run("cat", { background: true, stdin: true }); + await handle.writeStdin("hello "); + // The daemon accepts the write but the response is lost. + daemon.failNextWriteResponse = true; + await expect(handle.writeStdin("world")).rejects.toBeInstanceOf( + ConnectionFailedError, + ); + expect(daemon.accepted).toBe(11n); + // The cursor did not advance, so the retry lands at the same offset + // and the daemon deduplicates it — the process saw the bytes once. + await handle.writeStdin("world"); + expect(daemon.writes.at(-1)?.offset).toBe(6n); + expect(daemon.accepted).toBe(11n); + }); + + it("stdinStatus() reports the daemon's acceptance state and resyncs", async () => { + const daemon = new MockDaemon(); + const handle = await daemon + .commands() + .run("cat", { background: true, stdin: true }); + await handle.writeStdin("abc"); + const status = await handle.stdinStatus(); + expect(status).toEqual({ bytesWritten: 3, closed: false }); + }); + + it("foreground stdin data is written then closed before the wait", async () => { + const daemon = new MockDaemon(); + const result = await daemon.commands().run("cat", { stdin: "fed\n" }); + expect(result.exitCode).toBe(0); + expect(daemon.starts[0]?.stdin).toBe(true); + expect(daemon.writes).toHaveLength(2); + expect(daemon.writes[0]?.eof).toBe(false); + expect(daemon.writes[1]?.eof).toBe(true); + expect(daemon.closed).toBe(true); + }); + + it("a feed racing the process's early exit is noise, not a failure", async () => { + const daemon = new MockDaemon(); + daemon.rejectWrites = new ConnectError( + "execution has exited", + Code.FailedPrecondition, + ); + daemon.waitState = ExecutionState.EXITED; + const result = await daemon.commands().run("true", { stdin: "unread" }); + expect(result.exitCode).toBe(0); + }); + + it("a feed failure with the process still running is real, and the leaked process is killed", async () => { + const daemon = new MockDaemon(); + daemon.rejectWrites = new ConnectError( + "stdin is already closed", + Code.FailedPrecondition, + ); + daemon.waitState = ExecutionState.RUNNING; + await expect( + daemon.commands().run("cat", { stdin: "data" }), + ).rejects.toThrow("stdin is already closed"); + // The caller gets no handle out of a thrown run(): without the + // best-effort kill, cat would wait on stdin forever and keep the + // sandbox RUNNING. + expect(daemon.signals).toEqual([Signal.SIGKILL]); + }); + + it("rejects stdin: true on a foreground run before any RPC", async () => { + const daemon = new MockDaemon(); + await expect( + daemon.commands().run("cat", { stdin: true }), + ).rejects.toBeInstanceOf(InvalidArgumentError); + expect(daemon.starts).toHaveLength(0); + }); +}); + +describe("commands.get", () => { + it("re-attaches by id and seeds the stdin cursor from the daemon", async () => { + const daemon = new MockDaemon(); + daemon.accepted = 7n; + daemon.waitState = ExecutionState.RUNNING; + const seeded = create(ExecutionSchema, { + id: "exec-9", + state: ExecutionState.RUNNING, + stdin: { bytesWritten: 7n, closed: false }, + }); + const transport = createRouterTransport(({ service }) => { + service(SandboxProcessService, { + waitExecution: () => seeded, + writeStdin(req) { + daemon.writes.push(req); + return create(StdinStatusSchema, { + bytesWritten: req.offset + BigInt(req.data.byteLength), + }); + }, + }); + }); + const commands = new Commands({ transport }, "sb-1"); + const handle = await commands.get("exec-9"); + expect(handle.commandId).toBe("exec-9"); + await handle.writeStdin("more"); + expect(daemon.writes[0]?.offset).toBe(7n); + }); +});