feat(sdk): add sync and async execution stdin helpers (#15)

Author: sid-rl

README-SDK.md

@@ -233,6 +233,23 @@ print("Errors:", result.stderr())
 execution.kill()
 ```
 
+To send input, start the execution with stdin attached:
+
+```python
+execution = devbox.cmd.exec_async("cat", attach_stdin=True)
+execution.send_std_in("first\n")
+execution.send_std_in("second\n")
+execution.close_std_in()
+result = execution.result()
+assert result.stdout() == "first\nsecond\n"
+```
+
+The async SDK exposes the same helpers with `await`. Send nonempty text, finish each
+send before the next, and wait for pending sends before closing stdin. Concurrent
+calls are not serialized. These helpers preserve the generated client's retry policy;
+a retry after an ambiguous connection failure can replay input (not exactly-once
+execution).
+
 **Key methods:**
 
 - `execution.get_state()` - Get current execution state (status, exit_code, etc.)

src/runloop_api_client/sdk/async_execution.py

@@ -9,6 +9,7 @@
 
 from ._types import BaseRequestOptions, LongRequestOptions
 from .._client import AsyncRunloop
+from .._exceptions import RunloopError
 from .async_execution_result import AsyncExecutionResult
 from ..types.devbox_async_execution_detail_view import DevboxAsyncExecutionDetailView
 
@@ -132,6 +133,38 @@ async def get_state(self, **options: Unpack[BaseRequestOptions]) -> DevboxAsyncE
             **options,
         )
 
+    async def send_std_in(self, text: str, **options: Unpack[LongRequestOptions]) -> None:
+        """Send nonempty text to an execution started with ``attach_stdin=True``.
+
+        Complete each send before the next send or ``close_std_in``. Concurrent
+        calls are not ordered. Client retries can replay input after an ambiguous
+        connection failure; delivery is not guaranteed to be exactly once.
+
+        :param text: Text to write to stdin
+        :param options: Request configuration forwarded to the generated client
+        :raises RunloopError: If the API reports unsuccessful delivery
+        """
+        response = await self._client.devboxes.executions.send_std_in(
+            self._execution_id, devbox_id=self._devbox_id, text=text, **options
+        )
+        if not response.success:
+            raise RunloopError(f"Failed to send stdin to execution {self._execution_id}")
+
+    async def close_std_in(self, **options: Unpack[LongRequestOptions]) -> None:
+        """Send EOF after all input has been delivered so the execution can finish.
+
+        Requires ``attach_stdin=True``. Wait for pending sends before closing.
+        This handle does not serialize calls or guarantee exactly-once delivery.
+
+        :param options: Request configuration forwarded to the generated client
+        :raises RunloopError: If the API reports unsuccessful delivery
+        """
+        response = await self._client.devboxes.executions.send_std_in(
+            self._execution_id, devbox_id=self._devbox_id, signal="EOF", **options
+        )
+        if not response.success:
+            raise RunloopError(f"Failed to send stdin to execution {self._execution_id}")
+
     async def kill(self, **options: Unpack[LongRequestOptions]) -> None:
         """Request termination of the running execution.
 

src/runloop_api_client/sdk/execution.py

@@ -9,6 +9,7 @@
 
 from ._types import BaseRequestOptions, LongRequestOptions
 from .._client import Runloop
+from .._exceptions import RunloopError
 from .execution_result import ExecutionResult
 from ..types.devbox_async_execution_detail_view import DevboxAsyncExecutionDetailView
 
@@ -120,6 +121,38 @@ def get_state(self, **options: Unpack[BaseRequestOptions]) -> DevboxAsyncExecuti
             **options,
         )
 
+    def send_std_in(self, text: str, **options: Unpack[LongRequestOptions]) -> None:
+        """Send nonempty text to an execution started with ``attach_stdin=True``.
+
+        Complete each send before the next send or ``close_std_in``. Concurrent
+        calls are not ordered. Client retries can replay input after an ambiguous
+        connection failure; delivery is not guaranteed to be exactly once.
+
+        :param text: Text to write to stdin
+        :param options: Request configuration forwarded to the generated client
+        :raises RunloopError: If the API reports unsuccessful delivery
+        """
+        response = self._client.devboxes.executions.send_std_in(
+            self._execution_id, devbox_id=self._devbox_id, text=text, **options
+        )
+        if not response.success:
+            raise RunloopError(f"Failed to send stdin to execution {self._execution_id}")
+
+    def close_std_in(self, **options: Unpack[LongRequestOptions]) -> None:
+        """Send EOF after all input has been delivered so the execution can finish.
+
+        Requires ``attach_stdin=True``. Wait for pending sends before closing.
+        This handle does not serialize calls or guarantee exactly-once delivery.
+
+        :param options: Request configuration forwarded to the generated client
+        :raises RunloopError: If the API reports unsuccessful delivery
+        """
+        response = self._client.devboxes.executions.send_std_in(
+            self._execution_id, devbox_id=self._devbox_id, signal="EOF", **options
+        )
+        if not response.success:
+            raise RunloopError(f"Failed to send stdin to execution {self._execution_id}")
+
     def kill(self, **options: Unpack[LongRequestOptions]) -> None:
         """Request termination of the running execution.
 

tests/sdk/test_execution_stdin_transport.py

@@ -0,0 +1,86 @@
+"""Stdin wrappers use generated routing and preserve request configuration."""
+
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+
+from runloop_api_client import RunloopError
+from runloop_api_client.sdk import Execution, RunloopSDK, AsyncExecution, AsyncRunloopSDK
+from runloop_api_client.types import DevboxAsyncExecutionDetailView
+
+
+class StdinTransport:
+    def __init__(self, success: bool) -> None:
+        self.success = success
+        self.bodies: list[object] = []
+
+    def __call__(self, request: httpx.Request) -> httpx.Response:
+        assert request.method == "POST"
+        assert request.url.path == "/v1/devboxes/dbx_test/executions/execution_test/send_std_in"
+        assert request.headers["x-request-id"] == "synthetic-idempotency-key"
+        assert request.headers["x-probe"] == "probe"
+        assert request.extensions["timeout"]["read"] == 12
+        self.bodies.append(json.loads(request.content))
+        return httpx.Response(
+            200, json={"devbox_id": "dbx_test", "execution_id": "execution_test", "success": self.success}
+        )
+
+
+@pytest.mark.parametrize("success", [True, False])
+def test_stdin_generated_routing_and_unsuccessful_delivery(success: bool) -> None:
+    transport = StdinTransport(success)
+    with RunloopSDK(
+        bearer_token="synthetic",
+        base_url="https://api.invalid",
+        max_retries=0,
+        http_client=httpx.Client(transport=httpx.MockTransport(transport)),
+    ) as sdk:
+        execution = Execution(
+            sdk.api, "dbx_test", DevboxAsyncExecutionDetailView.model_construct(execution_id="execution_test")
+        )
+        for operation in [
+            lambda: execution.send_std_in(
+                "hello\n", extra_headers={"x-probe": "probe"}, idempotency_key="synthetic-idempotency-key", timeout=12
+            ),
+            lambda: execution.close_std_in(
+                extra_headers={"x-probe": "probe"}, idempotency_key="synthetic-idempotency-key", timeout=12
+            ),
+        ]:
+            if success:
+                assert operation() is None
+            else:
+                with pytest.raises(RunloopError, match="Failed to send stdin"):
+                    operation()
+    assert transport.bodies == [{"text": "hello\n"}, {"signal": "EOF"}]
+
+
+@pytest.mark.parametrize("success", [True, False])
+@pytest.mark.asyncio
+async def test_async_stdin_generated_routing_and_unsuccessful_delivery(success: bool) -> None:
+    transport = StdinTransport(success)
+    async with AsyncRunloopSDK(
+        bearer_token="synthetic",
+        base_url="https://api.invalid",
+        max_retries=0,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(transport)),
+    ) as sdk:
+        execution = AsyncExecution(
+            sdk.api, "dbx_test", DevboxAsyncExecutionDetailView.model_construct(execution_id="execution_test")
+        )
+        for operation in [
+            lambda: execution.send_std_in(
+                "hello\n", extra_headers={"x-probe": "probe"}, idempotency_key="synthetic-idempotency-key", timeout=12
+            ),
+            lambda: execution.close_std_in(
+                extra_headers={"x-probe": "probe"}, idempotency_key="synthetic-idempotency-key", timeout=12
+            ),
+        ]:
+            if success:
+                assert await operation() is None
+            else:
+                with pytest.raises(RunloopError, match="Failed to send stdin"):
+                    await operation()
+    assert transport.bodies == [{"text": "hello\n"}, {"signal": "EOF"}]

tests/smoketests/sdk/test_async_devbox.py

@@ -125,6 +125,22 @@ async def test_exec_with_exit_code(self, shared_devbox: AsyncDevbox) -> None:
         assert result.success is False
         assert await result.stdout(num_lines=1) == ""
 
+    @pytest.mark.timeout(TWO_MINUTE_TIMEOUT)
+    async def test_stdin_ordered_roundtrip(self, shared_devbox: AsyncDevbox) -> None:
+        execution = await shared_devbox.cmd.exec_async("cat", attach_stdin=True)
+        completed = False
+        try:
+            await execution.send_std_in("first\n")
+            await execution.send_std_in("second\n")
+            await execution.close_std_in()
+            result = await execution.result()
+            completed = True
+            assert result.exit_code == 0
+            assert await result.stdout() == "first\nsecond\n"
+        finally:
+            if not completed:
+                await execution.kill()
+
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     async def test_exec_async_command(self, shared_devbox: AsyncDevbox) -> None:
         """Test executing a command asynchronously with exec_async."""

tests/smoketests/sdk/test_devbox.py

@@ -125,6 +125,22 @@ def test_exec_with_exit_code(self, shared_devbox: Devbox) -> None:
         assert result.success is False
         assert "" == result.stdout(num_lines=1)
 
+    @pytest.mark.timeout(TWO_MINUTE_TIMEOUT)
+    def test_stdin_ordered_roundtrip(self, shared_devbox: Devbox) -> None:
+        execution = shared_devbox.cmd.exec_async("cat", attach_stdin=True)
+        completed = False
+        try:
+            execution.send_std_in("first\n")
+            execution.send_std_in("second\n")
+            execution.close_std_in()
+            result = execution.result()
+            completed = True
+            assert result.exit_code == 0
+            assert result.stdout() == "first\nsecond\n"
+        finally:
+            if not completed:
+                execution.kill()
+
     @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
     def test_exec_async_command(self, shared_devbox: Devbox) -> None:
         """Test executing a command asynchronously."""