diff --git a/.github/workflows/ci-check-docs-links.yml b/.github/workflows/ci-check-docs-links.yml index 5080f2ff47..7654ddec0f 100644 --- a/.github/workflows/ci-check-docs-links.yml +++ b/.github/workflows/ci-check-docs-links.yml @@ -61,10 +61,13 @@ jobs: --verbose --no-progress --accept 100..=399 - --exclude '^file://' + --scheme https + --scheme http --exclude 'localhost' --exclude '127\.0\.0\.1' --exclude 'example\.com' + --exclude 'cua\.ai' + --exclude 'platform\.openai\.com' './docs/content/**/*.mdx' token: ${{ secrets.GITHUB_TOKEN }} fail: true diff --git a/.github/workflows/ci-test-python.yml b/.github/workflows/ci-test-python.yml index 20ea758076..a3a55b4ab5 100644 --- a/.github/workflows/ci-test-python.yml +++ b/.github/workflows/ci-test-python.yml @@ -22,6 +22,7 @@ jobs: - computer-server - mcp-server - som + - cua-auto steps: - name: Checkout code diff --git a/README.md b/README.md index ee048f5684..2fbc871393 100644 --- a/README.md +++ b/README.md @@ -166,7 +166,7 @@ lume run macos-sequoia-vanilla:latest | Package | Description | | --------------------------------------------------------------------- | ---------------------------------------------------------- | -| [cuabot](https://cua.ai/docs/cuabot/cuabot) | Multi-agent computer-use sandbox CLI | +| [cuabot](https://docs.trycua.com/cuabot/guide/getting-started/introduction) | Multi-agent computer-use sandbox CLI | | [cua-agent](https://cua.ai/docs/cua/reference/agent-sdk) | AI agent framework for computer-use tasks | | [cua-computer](https://cua.ai/docs/cua/reference/computer-sdk) | SDK for controlling desktop environments | | [cua-computer-server](https://cua.ai/docs/cua/reference/computer-sdk) | Driver for UI interactions and code execution in sandboxes | diff --git a/blog/clawcon-multiplayer.md b/blog/clawcon-multiplayer.md index fe98674e1f..03cda36324 100644 --- a/blog/clawcon-multiplayer.md +++ b/blog/clawcon-multiplayer.md @@ -4,7 +4,7 @@ _Published on February 6, 2026 by Francesco Bonacci and Dillon DuPont_ ClawCon brought over 700 attendees to Frontier Tower, with a waitlist that had people lining up down Market Street, and another 20k tuned into the livestream. It was the first community event for OpenClaw, and we had the 2nd demo session. -We're early OpenClaw contributors and sponsors (we documented [how to deploy OpenClaw to macOS sandboxes with Lume](https://docs.trycua.com/lume/guides/openclaw)), and we genuinely believe computer-use works best as a tool inside more general agentic systems like OpenClaw rather than as standalone screen-takeover agents. +We're early OpenClaw contributors and sponsors (we documented [how to deploy OpenClaw to macOS sandboxes with Lume](https://docs.trycua.com/lume/examples/openclaw)), and we genuinely believe computer-use works best as a tool inside more general agentic systems like OpenClaw rather than as standalone screen-takeover agents. So we deferred our Hacker News launch to ship something big live on stage. diff --git a/docs/content/docs/cua/guide/advanced/interactive-shell.mdx b/docs/content/docs/cua/guide/advanced/interactive-shell.mdx new file mode 100644 index 0000000000..4d34d709cb --- /dev/null +++ b/docs/content/docs/cua/guide/advanced/interactive-shell.mdx @@ -0,0 +1,77 @@ +--- +title: Interactive Shell +description: Open a live PTY session inside a sandbox or local machine from the CLI or Python SDK +--- + +The `cua do shell` command and `computer.pty` API give you an interactive terminal (PTY) session inside any sandbox or local machine — similar to SSH, but without any network setup. + +## CLI + +```bash +# Switch to a target first +cua do switch docker my-container +# or +cua do switch host + +# Open an interactive shell (bash / PowerShell on host) +cua do shell + +# Run a specific program interactively +cua do shell python3 +cua do shell vim /etc/hosts + +# Override terminal dimensions +cua do shell --cols 220 --rows 50 +``` + +The shell is fully interactive — arrow keys, tab completion, and `Ctrl+C` all work. When stdin is piped (non-TTY), the command runs non-interactively and its output is printed normally. + +## Python SDK + +```python +from computer import Computer + +async with Computer(provider_type="docker", name="my-container") as c: + handle = await c.pty.create( + command="bash", + cols=80, + rows=24, + on_data=lambda chunk: print(chunk.decode(errors="replace"), end="", flush=True), + ) + + await handle.send_stdin(b"echo hello\n") + await handle.send_stdin(b"exit\n") + code = await handle.wait() + print(f"exited {code}") +``` + +### PtyHandle methods + +| Method | Description | +|---|---| +| `send_stdin(data: bytes)` | Write bytes to the terminal's stdin | +| `resize(cols, rows)` | Resize the terminal window | +| `kill()` | Kill the session process | +| `disconnect()` | Close the WebSocket without killing the process | +| `wait()` | Block until the session exits; returns the exit code | + +### Reconnecting + +```python +# Disconnect and reconnect later (process keeps running) +await handle.disconnect() + +handle2 = await c.pty.connect( + handle.pid, + on_data=lambda chunk: print(chunk.decode(errors="replace"), end=""), +) +``` + +## How it works + +The PTY stack has four layers: + +1. **`cua-auto`** — spawns a real PTY process (`pty` on Unix, `pywinpty` on Windows) +2. **`computer-server`** — exposes `/pty` REST endpoints and a WebSocket for I/O +3. **`computer`** — Python client (`PtyInterface`) connecting over HTTP + WebSocket +4. **`cua do shell`** — sets your terminal to raw mode and proxies I/O through the WebSocket diff --git a/docs/content/docs/cua/guide/advanced/meta.json b/docs/content/docs/cua/guide/advanced/meta.json index ba5dd84457..879d7ec709 100644 --- a/docs/content/docs/cua/guide/advanced/meta.json +++ b/docs/content/docs/cua/guide/advanced/meta.json @@ -7,6 +7,7 @@ "custom-tools", "sandboxed-python", "local-computer-server", + "interactive-shell", "human-in-the-loop", "vnc-recorder", "demonstration-guided-skills", diff --git a/docs/content/docs/cua/reference/computer-sdk/index.mdx b/docs/content/docs/cua/reference/computer-sdk/index.mdx index a7b64d911a..a8b5c57143 100644 --- a/docs/content/docs/cua/reference/computer-sdk/index.mdx +++ b/docs/content/docs/cua/reference/computer-sdk/index.mdx @@ -29,6 +29,8 @@ Cua Computer Interface for cross-platform computer control. |-------|-------------| | [`Computer`](#computer) | Computer is the main class for interacting with the computer. | | [`VMProviderType`](#vmprovidertype) | Enum of supported VM provider types. | +| [`PtyHandle`](#ptyhandle) | Lightweight handle to a live PTY session on a remote computer-server. | +| [`PtyInterface`](#ptyinterface) | Async HTTP+WebSocket client for the ``/pty`` endpoints on computer-server. | --- @@ -69,6 +71,7 @@ Computer(self, display: Union[Display, Dict[str, int], str] = '1024x768', memory | `shared_directories` | `Any` | | | `use_host_computer_server` | `Any` | | | `interface` | `Any` | Get the computer interface for interacting with the VM. | +| `pty` | `PtyInterface` | Return a :class:`~computer.pty.PtyInterface` for spawning interactive PTY sessions. | | `tracing` | `ComputerTracing` | Get the computer tracing instance for recording sessions. | | `telemetry_enabled` | `bool` | Check if telemetry is enabled for this computer instance. | @@ -404,318 +407,155 @@ Enum of supported VM provider types. --- -## tracing - -Computer tracing functionality for recording sessions. - -This module provides a Computer.tracing API inspired by Playwright's tracing functionality, -allowing users to record computer interactions for debugging, training, and analysis. - ---- +## PtyHandle -## ComputerTracing - -Computer tracing class that records computer interactions and saves them to disk. - -This class provides a flexible API for recording computer sessions with configurable -options for what to record (screenshots, API calls, video, etc.). +Lightweight handle to a live PTY session on a remote computer-server. ### Constructor ```python -ComputerTracing(self, computer_instance) +PtyHandle(self, pid: int, cols: int, rows: int, _iface: 'PtyInterface') -> None ``` ### Attributes | Name | Type | Description | |------|------|-------------| -| `is_tracing` | `bool` | Check if tracing is currently active. | +| `pid` | `int` | | +| `cols` | `int` | | +| `rows` | `int` | | ### Methods -#### ComputerTracing.start - -```python -async def start(self, config: Optional[Dict[str, Any]] = None) -> None -``` - -Start tracing with the specified configuration. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `config` | `Any` | Tracing configuration dict with options: - video: bool - Record video frames (default: False) - screenshots: bool - Record screenshots (default: True) - api_calls: bool - Record API calls and results (default: True) - accessibility_tree: bool - Record accessibility tree snapshots (default: False) - metadata: bool - Record custom metadata (default: True) - name: str - Custom trace name (default: auto-generated) - path: str - Custom trace directory path (default: auto-generated) | - -#### ComputerTracing.stop - -```python -async def stop(self, options: Optional[Dict[str, Any]] = None) -> str -``` - -Stop tracing and save the trace data. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `options` | `Any` | Stop options dict with: - path: str - Custom output path for the trace archive - format: str - Output format ('zip' or 'dir', default: 'zip') | - -**Returns:** str: Path to the saved trace file or directory - -#### ComputerTracing.record_api_call - -```python -async def record_api_call(self, method: str, args: Dict[str, Any], result: Any = None, error: Optional[Exception] = None) -> None -``` - -Record an API call event. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `method` | `Any` | The method name that was called | -| `args` | `Any` | Arguments passed to the method | -| `result` | `Any` | Result returned by the method | -| `error` | `Any` | Exception raised by the method, if any | - -#### ComputerTracing.record_accessibility_tree +#### PtyHandle.send_stdin ```python -async def record_accessibility_tree(self) -> None +async def send_stdin(self, data: bytes) -> None ``` -Record the current accessibility tree if enabled. +Write *data* to the PTY's stdin. -#### ComputerTracing.add_metadata +#### PtyHandle.resize ```python -async def add_metadata(self, key: str, value: Any) -> None +async def resize(self, cols: int, rows: int) -> None ``` -Add custom metadata to the trace. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `key` | `Any` | Metadata key | -| `value` | `Any` | Metadata value | - ---- - -## models - -Models for computer configuration. - ---- - -## BaseVMProvider - -*Inherits from: AsyncContextManager* - -Base interface for VM providers. - -All VM provider implementations must implement this interface. - -### Attributes - -| Name | Type | Description | -|------|------|-------------| -| `provider_type` | `VMProviderType` | Get the provider type. | - -### Methods +Resize the terminal window. -#### BaseVMProvider.get_vm +#### PtyHandle.kill ```python -async def get_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any] +async def kill(self) -> bool ``` -Get VM information by name. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `name` | `Any` | Name of the VM to get information for | -| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | - -**Returns:** Dictionary with VM information including status, IP address, etc. +Kill the PTY session process. -#### BaseVMProvider.list_vms +#### PtyHandle.disconnect ```python -async def list_vms(self) -> ListVMsResponse +async def disconnect(self) -> None ``` -List all available VMs. +Close the WebSocket connection without killing the PTY. -**Returns:** ListVMsResponse: A list of minimal VM objects as defined in `computer.providers.types.MinimalVM`. +The session keeps running on the server; use :meth:`connect` to +re-attach later. -#### BaseVMProvider.run_vm +#### PtyHandle.wait ```python -async def run_vm(self, image: str, name: str, run_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any] +async def wait(self) -> int ``` -Run a VM by name with the given options. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `image` | `Any` | Name/tag of the image to use | -| `name` | `Any` | Name of the VM to run | -| `run_opts` | `Any` | Dictionary of run options (memory, cpu, etc.) | -| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | - -**Returns:** Dictionary with VM run status and information +Block until the PTY session exits and return its exit code. -#### BaseVMProvider.stop_vm +**Raises:** -```python -async def stop_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any] -``` +- `LookupError` - If the session for this handle's pid is not tracked by the interface (e.g. the handle was never connected, or the interface was recreated after a reconnect). -Stop a VM by name. +--- -**Parameters:** +## PtyInterface -| Name | Type | Description | -|------|------|-------------| -| `name` | `Any` | Name of the VM to stop | -| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | +Async HTTP+WebSocket client for the ``/pty`` endpoints on computer-server. -**Returns:** Dictionary with VM stop status and information +Args: + base_url: HTTP base URL of the computer-server, e.g. + ``"http://192.168.64.10:8000"``. + api_key: Optional API key for cloud providers (passed as + ``X-API-Key`` header). + vm_name: Optional VM / container name (passed as + ``X-Container-Name`` header). -#### BaseVMProvider.restart_vm +### Constructor ```python -async def restart_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any] +PtyInterface(self, base_url: str, api_key: Optional[str] = None, vm_name: Optional[str] = None) -> None ``` -Restart a VM by name. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `name` | `Any` | Name of the VM to restart | -| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | - -**Returns:** Dictionary with VM restart status and information +### Methods -#### BaseVMProvider.update_vm +#### PtyInterface.create ```python -async def update_vm(self, name: str, update_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any] +async def create(self, command: Optional[str] = None, cols: int = 80, rows: int = 24, on_data: Optional[Callable[[bytes], None]] = None, cwd: Optional[str] = None, envs: Optional[dict] = None, timeout: int = 60) -> PtyHandle ``` -Update VM configuration. +Spawn a new PTY session on the remote computer-server. **Parameters:** | Name | Type | Description | |------|------|-------------| -| `name` | `Any` | Name of the VM to update | -| `update_opts` | `Any` | Dictionary of update options (memory, cpu, etc.) | -| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | +| `command` | `Any` | Shell command (defaults to ``bash`` on the server side). | +| `cols` | `Any` | Terminal width. | +| `rows` | `Any` | Terminal height. | +| `on_data` | `Any` | Callback invoked with raw bytes whenever the PTY produces output. Called from the asyncio event loop. | +| `cwd` | `Any` | Working directory on the remote host. | +| `envs` | `Any` | Extra environment variables for the remote process. | +| `timeout` | `Any` | HTTP request timeout in seconds. | -**Returns:** Dictionary with VM update status and information +**Returns:** :class:`PtyHandle` for the new session. -#### BaseVMProvider.get_ip +#### PtyInterface.kill ```python -async def get_ip(self, name: str, storage: Optional[str] = None, retry_delay: int = 2) -> str +async def kill(self, pid: int) -> bool ``` -Get the IP address of a VM, waiting indefinitely until it's available. - -**Parameters:** - -| Name | Type | Description | -|------|------|-------------| -| `name` | `Any` | Name of the VM to get the IP for | -| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | -| `retry_delay` | `Any` | Delay between retries in seconds (default: 2) | - -**Returns:** IP address of the VM when it becomes available - ---- - -## Display - -Display configuration. +Kill the remote PTY session *pid*. -### Constructor +#### PtyInterface.resize ```python -Display(self, width: int, height: int) -> None +async def resize(self, pid: int, cols: int, rows: int) -> None ``` -### Attributes - -| Name | Type | Description | -|------|------|-------------| -| `width` | `int` | | -| `height` | `int` | | - ---- - -## Image - -VM image configuration. +Resize the terminal for the remote PTY session *pid*. -### Constructor +#### PtyInterface.send_stdin ```python -Image(self, image: str, tag: str, name: str) -> None +async def send_stdin(self, pid: int, data: bytes) -> None ``` -### Attributes - -| Name | Type | Description | -|------|------|-------------| -| `image` | `str` | | -| `tag` | `str` | | -| `name` | `str` | | - ---- - -## Computer +Write *data* to the remote PTY session's stdin via the WebSocket. -Computer configuration. +If a WebSocket is active the data is sent through it; otherwise falls +back to the HTTP ``/stdin`` endpoint. -### Constructor +#### PtyInterface.connect ```python -Computer(self, image: str, tag: str, name: str, display: Display, memory: str, cpu: str, vm_provider: Optional[BaseVMProvider] = None) -> None +async def connect(self, pid: int, on_data: Optional[Callable[[bytes], None]] = None) -> PtyHandle ``` -### Attributes - -| Name | Type | Description | -|------|------|-------------| -| `image` | `str` | | -| `tag` | `str` | | -| `name` | `str` | | -| `display` | `Display` | | -| `memory` | `str` | | -| `cpu` | `str` | | -| `vm_provider` | `Optional[BaseVMProvider]` | | - -### Methods - -#### Computer.get_ip - -```python -async def get_ip(self) -> Optional[str] -``` +Re-open a WebSocket to an existing PTY session *pid*. -Get the IP address of the VM. +Any previous connection for this pid is cancelled first. +The current terminal dimensions are fetched from the server so that +the returned :class:`PtyHandle` reflects any resizes since creation. --- @@ -1108,6 +948,321 @@ Generate complete source code for a function with all dependencies. --- +## models + +Models for computer configuration. + +--- + +## BaseVMProvider + +*Inherits from: AsyncContextManager* + +Base interface for VM providers. + +All VM provider implementations must implement this interface. + +### Attributes + +| Name | Type | Description | +|------|------|-------------| +| `provider_type` | `VMProviderType` | Get the provider type. | + +### Methods + +#### BaseVMProvider.get_vm + +```python +async def get_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any] +``` + +Get VM information by name. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `name` | `Any` | Name of the VM to get information for | +| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | + +**Returns:** Dictionary with VM information including status, IP address, etc. + +#### BaseVMProvider.list_vms + +```python +async def list_vms(self) -> ListVMsResponse +``` + +List all available VMs. + +**Returns:** ListVMsResponse: A list of minimal VM objects as defined in `computer.providers.types.MinimalVM`. + +#### BaseVMProvider.run_vm + +```python +async def run_vm(self, image: str, name: str, run_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any] +``` + +Run a VM by name with the given options. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `image` | `Any` | Name/tag of the image to use | +| `name` | `Any` | Name of the VM to run | +| `run_opts` | `Any` | Dictionary of run options (memory, cpu, etc.) | +| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | + +**Returns:** Dictionary with VM run status and information + +#### BaseVMProvider.stop_vm + +```python +async def stop_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any] +``` + +Stop a VM by name. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `name` | `Any` | Name of the VM to stop | +| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | + +**Returns:** Dictionary with VM stop status and information + +#### BaseVMProvider.restart_vm + +```python +async def restart_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any] +``` + +Restart a VM by name. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `name` | `Any` | Name of the VM to restart | +| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | + +**Returns:** Dictionary with VM restart status and information + +#### BaseVMProvider.update_vm + +```python +async def update_vm(self, name: str, update_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any] +``` + +Update VM configuration. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `name` | `Any` | Name of the VM to update | +| `update_opts` | `Any` | Dictionary of update options (memory, cpu, etc.) | +| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | + +**Returns:** Dictionary with VM update status and information + +#### BaseVMProvider.get_ip + +```python +async def get_ip(self, name: str, storage: Optional[str] = None, retry_delay: int = 2) -> str +``` + +Get the IP address of a VM, waiting indefinitely until it's available. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `name` | `Any` | Name of the VM to get the IP for | +| `storage` | `Any` | Optional storage path override. If provided, this will be used instead of the provider's default storage path. | +| `retry_delay` | `Any` | Delay between retries in seconds (default: 2) | + +**Returns:** IP address of the VM when it becomes available + +--- + +## Display + +Display configuration. + +### Constructor + +```python +Display(self, width: int, height: int) -> None +``` + +### Attributes + +| Name | Type | Description | +|------|------|-------------| +| `width` | `int` | | +| `height` | `int` | | + +--- + +## Image + +VM image configuration. + +### Constructor + +```python +Image(self, image: str, tag: str, name: str) -> None +``` + +### Attributes + +| Name | Type | Description | +|------|------|-------------| +| `image` | `str` | | +| `tag` | `str` | | +| `name` | `str` | | + +--- + +## Computer + +Computer configuration. + +### Constructor + +```python +Computer(self, image: str, tag: str, name: str, display: Display, memory: str, cpu: str, vm_provider: Optional[BaseVMProvider] = None) -> None +``` + +### Attributes + +| Name | Type | Description | +|------|------|-------------| +| `image` | `str` | | +| `tag` | `str` | | +| `name` | `str` | | +| `display` | `Display` | | +| `memory` | `str` | | +| `cpu` | `str` | | +| `vm_provider` | `Optional[BaseVMProvider]` | | + +### Methods + +#### Computer.get_ip + +```python +async def get_ip(self) -> Optional[str] +``` + +Get the IP address of the VM. + +--- + +## tracing + +Computer tracing functionality for recording sessions. + +This module provides a Computer.tracing API inspired by Playwright's tracing functionality, +allowing users to record computer interactions for debugging, training, and analysis. + +--- + +## ComputerTracing + +Computer tracing class that records computer interactions and saves them to disk. + +This class provides a flexible API for recording computer sessions with configurable +options for what to record (screenshots, API calls, video, etc.). + +### Constructor + +```python +ComputerTracing(self, computer_instance) +``` + +### Attributes + +| Name | Type | Description | +|------|------|-------------| +| `is_tracing` | `bool` | Check if tracing is currently active. | + +### Methods + +#### ComputerTracing.start + +```python +async def start(self, config: Optional[Dict[str, Any]] = None) -> None +``` + +Start tracing with the specified configuration. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `config` | `Any` | Tracing configuration dict with options: - video: bool - Record video frames (default: False) - screenshots: bool - Record screenshots (default: True) - api_calls: bool - Record API calls and results (default: True) - accessibility_tree: bool - Record accessibility tree snapshots (default: False) - metadata: bool - Record custom metadata (default: True) - name: str - Custom trace name (default: auto-generated) - path: str - Custom trace directory path (default: auto-generated) | + +#### ComputerTracing.stop + +```python +async def stop(self, options: Optional[Dict[str, Any]] = None) -> str +``` + +Stop tracing and save the trace data. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `options` | `Any` | Stop options dict with: - path: str - Custom output path for the trace archive - format: str - Output format ('zip' or 'dir', default: 'zip') | + +**Returns:** str: Path to the saved trace file or directory + +#### ComputerTracing.record_api_call + +```python +async def record_api_call(self, method: str, args: Dict[str, Any], result: Any = None, error: Optional[Exception] = None) -> None +``` + +Record an API call event. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `method` | `Any` | The method name that was called | +| `args` | `Any` | Arguments passed to the method | +| `result` | `Any` | Result returned by the method | +| `error` | `Any` | Exception raised by the method, if any | + +#### ComputerTracing.record_accessibility_tree + +```python +async def record_accessibility_tree(self) -> None +``` + +Record the current accessibility tree if enabled. + +#### ComputerTracing.add_metadata + +```python +async def add_metadata(self, key: str, value: Any) -> None +``` + +Add custom metadata to the trace. + +**Parameters:** + +| Name | Type | Description | +|------|------|-------------| +| `key` | `Any` | Metadata key | +| `value` | `Any` | Metadata value | + +--- + ## interface Interface package for Computer SDK. diff --git a/libs/cua-bench/pyproject.toml b/libs/cua-bench/pyproject.toml index 27a16fc672..368a28b407 100644 --- a/libs/cua-bench/pyproject.toml +++ b/libs/cua-bench/pyproject.toml @@ -53,7 +53,7 @@ dependencies = [ "docker>=7.0.0", # CUA SDK "cua-computer>=0.4.19", - "cua-core", + "cua-core>=0.1.18", ] [project.optional-dependencies] diff --git a/libs/cuabot/README.md b/libs/cuabot/README.md index 49a2ebb583..763fc0f003 100644 --- a/libs/cuabot/README.md +++ b/libs/cuabot/README.md @@ -36,5 +36,5 @@ cuabot --click 100 200 # Click at coordinates ## Documentation -- [Getting Started](https://cua.ai/docs/cuabot/cuabot) -- [Installation Guide](https://cua.ai/docs/cuabot/install) +- [Getting Started](https://docs.trycua.com/cuabot/guide/getting-started/introduction) +- [Installation Guide](https://docs.trycua.com/cuabot/guide/getting-started/installation) diff --git a/libs/python/agent/pyproject.toml b/libs/python/agent/pyproject.toml index 4c6856afda..0329ba37c0 100644 --- a/libs/python/agent/pyproject.toml +++ b/libs/python/agent/pyproject.toml @@ -20,7 +20,7 @@ dependencies = [ "rich>=13.7.1", "python-dotenv>=1.0.1", "cua-computer>=0.5.0,<0.6.0", - "cua-core>=0.1.8,<0.2.0", + "cua-core>=0.1.18,<0.2.0", "certifi>=2024.2.2", "litellm>=1.74.12" ] diff --git a/libs/python/computer-server/computer_server/main.py b/libs/python/computer-server/computer_server/main.py index 28d405ecdc..f0eafe0fc2 100644 --- a/libs/python/computer-server/computer_server/main.py +++ b/libs/python/computer-server/computer_server/main.py @@ -1,4 +1,5 @@ import asyncio +import base64 import hashlib import inspect import json @@ -329,6 +330,11 @@ def disconnect(self, websocket: WebSocket): manager = ConnectionManager() auth_manager = AuthenticationManager() +# PTY session manager (lazy-initialised) +from .pty_manager import PtyManager + +pty_manager = PtyManager() + def _resolve_command(command: str) -> str: """Resolve command aliases to their canonical names.""" @@ -642,6 +648,243 @@ async def generate_response(): ) +async def _require_auth( + container_name: Optional[str], + api_key: Optional[str], +) -> None: + """Raise HTTPException(401) when cloud auth is configured and credentials are invalid.""" + server_container_name = os.environ.get("CONTAINER_NAME") + if not server_container_name: + return # local development — no auth required + if not container_name: + raise HTTPException(status_code=401, detail="Container name required") + if not api_key: + raise HTTPException(status_code=401, detail="API key required") + if not await auth_manager.auth(container_name, api_key): + raise HTTPException(status_code=401, detail="Authentication failed") + + +# --------------------------------------------------------------------------- +# PTY endpoints +# --------------------------------------------------------------------------- + + +@app.post("/pty") +async def pty_create( + request: Request, + container_name: Optional[str] = Header(None, alias="X-Container-Name"), + api_key: Optional[str] = Header(None, alias="X-API-Key"), +): + """Spawn a new PTY session. + + Body (JSON, all fields optional): + ``{"command": str, "cols": int, "rows": int, "cwd": str, "envs": dict}`` + + Returns: ``{"pid": int, "cols": int, "rows": int}`` + """ + await _require_auth(container_name, api_key) + try: + body = await request.json() + except Exception: + body = {} + info = await pty_manager.create( + command=body.get("command"), + cols=int(body.get("cols", 80)), + rows=int(body.get("rows", 24)), + cwd=body.get("cwd"), + envs=body.get("envs"), + ) + return info + + +@app.get("/pty/{pid}") +async def pty_info( + pid: int, + container_name: Optional[str] = Header(None, alias="X-Container-Name"), + api_key: Optional[str] = Header(None, alias="X-API-Key"), +): + """Return metadata for PTY session *pid*. + + Returns: ``{"pid": int, "cols": int, "rows": int}`` + """ + await _require_auth(container_name, api_key) + info = pty_manager.get_info(pid) + if info is None: + from fastapi import HTTPException + + raise HTTPException(status_code=404, detail=f"PTY session {pid} not found") + return info + + +@app.delete("/pty/{pid}") +async def pty_kill( + pid: int, + container_name: Optional[str] = Header(None, alias="X-Container-Name"), + api_key: Optional[str] = Header(None, alias="X-API-Key"), +): + """Kill PTY session *pid*. + + Returns: ``{"killed": bool}`` + """ + await _require_auth(container_name, api_key) + killed = await pty_manager.kill(pid) + return {"killed": killed} + + +@app.post("/pty/{pid}/stdin") +async def pty_stdin( + pid: int, + request: Request, + container_name: Optional[str] = Header(None, alias="X-Container-Name"), + api_key: Optional[str] = Header(None, alias="X-API-Key"), +): + """Write data to stdin of PTY session *pid*. + + Body: ``{"data": ""}`` + """ + await _require_auth(container_name, api_key) + body = await request.json() + raw = base64.b64decode(body.get("data", "")) + await pty_manager.send_stdin(pid, raw) + return {"ok": True} + + +@app.post("/pty/{pid}/resize") +async def pty_resize( + pid: int, + request: Request, + container_name: Optional[str] = Header(None, alias="X-Container-Name"), + api_key: Optional[str] = Header(None, alias="X-API-Key"), +): + """Resize the terminal for PTY session *pid*. + + Body: ``{"cols": int, "rows": int}`` + """ + await _require_auth(container_name, api_key) + body = await request.json() + await pty_manager.resize(pid, int(body.get("cols", 80)), int(body.get("rows", 24))) + return {"ok": True} + + +@app.get("/pty/{pid}/stream") +async def pty_stream( + pid: int, + container_name: Optional[str] = Header(None, alias="X-Container-Name"), + api_key: Optional[str] = Header(None, alias="X-API-Key"), +): + """SSE stream for PTY session *pid*. + + Events: + - ``data: {"type": "output", "data": ""}`` + - ``data: {"type": "exit", "code": }`` + """ + await _require_auth(container_name, api_key) + + q = pty_manager.subscribe(pid) + + async def _generate(): + try: + while True: + msg = await q.get() + if msg["type"] == "output": + payload = {"type": "output", "data": base64.b64encode(msg["data"]).decode()} + else: + payload = {"type": "exit", "code": msg.get("code", 0)} + yield f"data: {json.dumps(payload)}\n\n" + if msg["type"] == "exit": + break + finally: + pty_manager.unsubscribe(pid, q) + + return StreamingResponse( + _generate(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}, + ) + + +@app.websocket("/pty/{pid}/ws") +async def pty_ws( + pid: int, + websocket: WebSocket, +): + """WebSocket endpoint for interactive PTY session *pid*. + + Auth (when CONTAINER_NAME is set): pass ``api_key`` and + ``container_name`` as query parameters, e.g. + ``/pty/123/ws?api_key=…&container_name=…``. + + Client → Server messages (JSON): + - ``{"type": "stdin", "data": ""}`` + - ``{"type": "resize", "cols": N, "rows": N}`` + - ``{"type": "disconnect"}`` + + Server → Client messages (JSON): + - ``{"type": "output", "data": ""}`` + - ``{"type": "exit", "code": N}`` + """ + container_name = websocket.query_params.get("container_name") + api_key = websocket.query_params.get("api_key") + try: + await _require_auth(container_name, api_key) + except HTTPException: + await websocket.close(code=1008) # 1008 = Policy Violation + return + + await websocket.accept() + + q = pty_manager.subscribe(pid) + + async def _send_output(): + """Forward PTY output to the WebSocket client.""" + try: + while True: + msg = await q.get() + if msg["type"] == "output": + payload = {"type": "output", "data": base64.b64encode(msg["data"]).decode()} + else: + payload = {"type": "exit", "code": msg.get("code", 0)} + await websocket.send_text(json.dumps(payload)) + if msg["type"] == "exit": + break + except Exception: + pass + finally: + pty_manager.unsubscribe(pid, q) + + output_task = asyncio.create_task(_send_output()) + + try: + while True: + try: + raw = await websocket.receive_text() + except WebSocketDisconnect: + break + except Exception: + break + + try: + msg = json.loads(raw) + except json.JSONDecodeError: + continue + + msg_type = msg.get("type") + if msg_type == "stdin": + data = base64.b64decode(msg.get("data", "")) + await pty_manager.send_stdin(pid, data) + elif msg_type == "resize": + await pty_manager.resize(pid, int(msg.get("cols", 80)), int(msg.get("rows", 24))) + elif msg_type == "disconnect": + break + finally: + output_task.cancel() + pty_manager.unsubscribe(pid, q) + try: + await websocket.close() + except Exception: + pass + + @app.post("/responses") async def agent_response_endpoint( request: Request, diff --git a/libs/python/computer-server/computer_server/pty_manager.py b/libs/python/computer-server/computer_server/pty_manager.py new file mode 100644 index 0000000000..4b18806525 --- /dev/null +++ b/libs/python/computer-server/computer_server/pty_manager.py @@ -0,0 +1,203 @@ +"""Async PTY session manager for computer-server. + +Wraps :class:`cua_auto.terminal.Terminal` and adds asyncio-compatible +output broadcasting via per-consumer :class:`asyncio.Queue` objects. + +Queue items are dicts: + - ``{"type": "output", "data": }`` — terminal output chunk + - ``{"type": "exit", "code": }`` — session terminated (sentinel) +""" + +from __future__ import annotations + +import asyncio +import logging +import threading +from typing import Dict, List, Optional + +logger = logging.getLogger(__name__) + + +class PtyManager: + """Manage PTY sessions with async broadcast queues. + + Usage:: + + mgr = PtyManager() + + # Create a session + info = await mgr.create(command="bash", cols=80, rows=24) + pid = info["pid"] + + # Subscribe to output + q = mgr.subscribe(pid) + async for msg in _drain(q): + ... # msg is {"type": "output", "data": b"..."} or {"type": "exit", "code": 0} + + # Write to stdin + await mgr.send_stdin(pid, b"echo hello\\n") + + # Resize + await mgr.resize(pid, 120, 40) + + # Kill + await mgr.kill(pid) + """ + + def __init__(self) -> None: + # pid → list of subscriber queues + self._queues: Dict[int, List[asyncio.Queue]] = {} + # pid → {"cols": int, "rows": int} + self._sessions: Dict[int, dict] = {} + self._terminal = None + + # ------------------------------------------------------------------ + # Lazy terminal access + # ------------------------------------------------------------------ + + def _get_terminal(self): + if self._terminal is None: + from cua_auto.terminal import Terminal + + self._terminal = Terminal() + return self._terminal + + # ------------------------------------------------------------------ + # Public async API + # ------------------------------------------------------------------ + + async def create( + self, + command: Optional[str] = None, + cols: int = 80, + rows: int = 24, + cwd: Optional[str] = None, + envs: Optional[dict] = None, + ) -> dict: + """Spawn a new PTY session. + + Returns: + ``{"pid": int, "cols": int, "rows": int}`` + """ + loop = asyncio.get_running_loop() + terminal = self._get_terminal() + + # We need pid to route output, but we only know it after create(). + # Use a mutable cell so the callback can look up the pid after creation. + # early_buffer holds chunks that arrive before pid_cell[0] is set (the + # reader thread can fire before asyncio.to_thread returns the session). + pid_cell: List[Optional[int]] = [None] + early_buffer: List[bytes] = [] + + def _on_data(data: bytes) -> None: + pid = pid_cell[0] + if pid is None: + early_buffer.append(data) + return + msg = {"type": "output", "data": data} + for q in list(self._queues.get(pid, [])): + try: + loop.call_soon_threadsafe(q.put_nowait, msg) + except Exception: + pass + + session = await asyncio.to_thread( + terminal.create, + command=command, + cols=cols, + rows=rows, + on_data=_on_data, + cwd=cwd, + envs=envs, + ) + + pid_cell[0] = session.pid + self._queues[session.pid] = [] + self._sessions[session.pid] = {"cols": session.cols, "rows": session.rows} + + # Flush any output that arrived before pid_cell[0] was set. + for chunk in early_buffer: + msg = {"type": "output", "data": chunk} + for q in list(self._queues[session.pid]): + try: + loop.call_soon_threadsafe(q.put_nowait, msg) + except Exception: + pass + + # Watch for process exit in a daemon thread, then broadcast sentinel. + def _watch_exit() -> None: + exit_code = terminal.wait(session.pid) or 0 + sentinel = {"type": "exit", "code": exit_code} + for q in list(self._queues.get(session.pid, [])): + try: + loop.call_soon_threadsafe(q.put_nowait, sentinel) + except Exception: + pass + + threading.Thread(target=_watch_exit, daemon=True, name=f"pty-exit-{session.pid}").start() + + logger.info( + "PTY session created: pid=%d cmd=%r cols=%d rows=%d", session.pid, command, cols, rows + ) + return {"pid": session.pid, "cols": session.cols, "rows": session.rows} + + async def send_stdin(self, pid: int, data: bytes) -> None: + """Write *data* to the stdin of session *pid*.""" + terminal = self._get_terminal() + await asyncio.to_thread(terminal.send_stdin, pid, data) + + async def resize(self, pid: int, cols: int, rows: int) -> None: + """Resize the terminal for session *pid*.""" + terminal = self._get_terminal() + await asyncio.to_thread(terminal.resize, pid, cols, rows) + if pid in self._sessions: + self._sessions[pid]["cols"] = cols + self._sessions[pid]["rows"] = rows + + def get_info(self, pid: int) -> Optional[dict]: + """Return ``{"pid": int, "cols": int, "rows": int}`` for *pid*, or ``None`` if unknown.""" + info = self._sessions.get(pid) + if info is None: + return None + return {"pid": pid, "cols": info["cols"], "rows": info["rows"]} + + async def kill(self, pid: int) -> bool: + """Kill session *pid*. + + Also broadcasts the exit sentinel to all subscribers. + Returns: + ``True`` if the signal was delivered. + """ + terminal = self._get_terminal() + result = await asyncio.to_thread(terminal.kill, pid) + # Broadcast sentinel immediately (the exit watcher will also fire, + # but duplicate sentinels are harmless — consumers stop after the first). + sentinel = {"type": "exit", "code": -1} + for q in list(self._queues.get(pid, [])): + try: + q.put_nowait(sentinel) + except Exception: + pass + return result + + # ------------------------------------------------------------------ + # Queue-based pub/sub + # ------------------------------------------------------------------ + + def subscribe(self, pid: int) -> asyncio.Queue: + """Return a new :class:`asyncio.Queue` that will receive output for *pid*. + + The queue receives dicts with keys ``type`` (``"output"`` or ``"exit"``) + and either ``data`` (bytes) or ``code`` (int). + """ + q: asyncio.Queue = asyncio.Queue() + self._queues.setdefault(pid, []).append(q) + return q + + def unsubscribe(self, pid: int, queue: asyncio.Queue) -> None: + """Remove *queue* from the subscriber list for *pid*.""" + qs = self._queues.get(pid, []) + try: + qs.remove(queue) + except ValueError: + pass diff --git a/libs/python/computer-server/pyproject.toml b/libs/python/computer-server/pyproject.toml index f8f735da99..deee94854c 100644 --- a/libs/python/computer-server/pyproject.toml +++ b/libs/python/computer-server/pyproject.toml @@ -25,6 +25,8 @@ dependencies = [ "pywinctl>=0.4.1", "playwright>=1.40.0", "fastmcp>=2.0,<3", + # PTY support (wraps cua-auto terminal engine) + "cua-auto>=0.1.0", # OS-specific runtime deps "pyobjc-framework-Cocoa>=10.1; sys_platform == 'darwin'", "pyobjc-framework-Quartz>=10.1; sys_platform == 'darwin'", @@ -32,7 +34,7 @@ dependencies = [ "python-xlib>=0.33; sys_platform == 'linux'", "pywin32>=310; sys_platform == 'win32'", "python-certifi-win32; sys_platform == 'win32'", - "cua-core", + "cua-core>=0.1.18", ] [project.optional-dependencies] @@ -82,3 +84,4 @@ api = "python -m computer_server" [tool.uv.sources] cua-core = { workspace = true } +cua-auto = { path = "../cua-auto", editable = true } diff --git a/libs/python/computer/computer/__init__.py b/libs/python/computer/computer/__init__.py index 462f2922c4..a7b8df31ce 100644 --- a/libs/python/computer/computer/__init__.py +++ b/libs/python/computer/computer/__init__.py @@ -44,4 +44,7 @@ # Provider components from .providers.base import VMProviderType -__all__ = ["Computer", "VMProviderType"] +# PTY client +from .pty import PtyHandle, PtyInterface + +__all__ = ["Computer", "PtyHandle", "PtyInterface", "VMProviderType"] diff --git a/libs/python/computer/computer/computer.py b/libs/python/computer/computer/computer.py index 8b1065bebb..1061fb52b5 100644 --- a/libs/python/computer/computer/computer.py +++ b/libs/python/computer/computer/computer.py @@ -286,6 +286,7 @@ def __init__( self._interface = None self._original_interface = None # Keep reference to original interface self._tracing_wrapper = None # Tracing wrapper for interface + self._pty_interface = None # Cached PtyInterface; invalidated on reconnect self.use_host_computer_server = use_host_computer_server # Initialize tracing @@ -676,6 +677,7 @@ async def run(self) -> Optional[str]: async def disconnect(self) -> None: """Disconnect from the computer's WebSocket interface.""" + self._pty_interface = None if self._interface: self._interface.close() @@ -799,6 +801,7 @@ async def restart(self) -> None: self.logger.info(f"Re-initializing interface for {self.os_type} at {ip_address}") from .interface.base import BaseComputerInterface + self._pty_interface = None if ( self.provider_type in (VMProviderType.CLOUD, VMProviderType.CLOUDV2) and self.api_key @@ -1052,6 +1055,35 @@ def interface(self): return result_interface + @property + def pty(self) -> "PtyInterface": + """Return a :class:`~computer.pty.PtyInterface` for spawning interactive PTY sessions. + + The computer must be started (``async with Computer()`` or ``await run()``) + before accessing this property. + + Example:: + + async with Computer(provider_type="docker", name="my-vm") as c: + handle = await c.pty.create(command="bash", cols=80, rows=24, + on_data=lambda d: print(d.decode())) + await handle.send_stdin(b"echo hello\\n") + await handle.send_stdin(b"exit\\n") + await handle.wait() + """ + from .pty import PtyInterface + + if self._interface is None: + raise RuntimeError("Computer not started. Use 'async with Computer()' first.") + if self._pty_interface is None: + protocol = "https" if self.api_key else "http" + port = getattr(self._interface, "_api_port", None) or self.api_port or 8000 + ip = getattr(self._interface, "ip_address", "localhost") + base_url = f"{protocol}://{ip}:{port}" + vm_name = getattr(getattr(self, "config", None), "name", None) or None + self._pty_interface = PtyInterface(base_url, api_key=self.api_key, vm_name=vm_name) + return self._pty_interface + @property def tracing(self) -> ComputerTracing: """Get the computer tracing instance for recording sessions. diff --git a/libs/python/computer/computer/pty.py b/libs/python/computer/computer/pty.py new file mode 100644 index 0000000000..4a8d78ebb1 --- /dev/null +++ b/libs/python/computer/computer/pty.py @@ -0,0 +1,385 @@ +"""e2b-style PTY client for a running Computer instance. + +Usage:: + + async with Computer(provider_type="docker", name="my-container") as c: + handle = await c.pty.create( + command="bash", + cols=80, + rows=24, + on_data=lambda d: print(d.decode(errors="replace"), end="", flush=True), + ) + await handle.send_stdin(b"echo hello\\n") + await handle.send_stdin(b"exit\\n") + code = await handle.wait() + print(f"exited with {code}") +""" + +from __future__ import annotations + +import asyncio +import base64 +import json +import logging +from dataclasses import dataclass, field +from typing import Callable, Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class PtyHandle: + """Lightweight handle to a live PTY session on a remote computer-server.""" + + pid: int + cols: int + rows: int + _iface: "PtyInterface" = field(repr=False) + + async def send_stdin(self, data: bytes) -> None: + """Write *data* to the PTY's stdin.""" + await self._iface.send_stdin(self.pid, data) + + async def resize(self, cols: int, rows: int) -> None: + """Resize the terminal window.""" + await self._iface.resize(self.pid, cols, rows) + + async def kill(self) -> bool: + """Kill the PTY session process.""" + return await self._iface.kill(self.pid) + + async def disconnect(self) -> None: + """Close the WebSocket connection without killing the PTY. + + The session keeps running on the server; use :meth:`connect` to + re-attach later. + """ + await self._iface._disconnect(self.pid) + + async def wait(self) -> int: + """Block until the PTY session exits and return its exit code. + + Raises: + LookupError: If the session for this handle's pid is not tracked + by the interface (e.g. the handle was never connected, or the + interface was recreated after a reconnect). + """ + return await self._iface._wait(self.pid) + + +class PtyInterface: + """Async HTTP+WebSocket client for the ``/pty`` endpoints on computer-server. + + Args: + base_url: HTTP base URL of the computer-server, e.g. + ``"http://192.168.64.10:8000"``. + api_key: Optional API key for cloud providers (passed as + ``X-API-Key`` header). + vm_name: Optional VM / container name (passed as + ``X-Container-Name`` header). + """ + + def __init__( + self, + base_url: str, + api_key: Optional[str] = None, + vm_name: Optional[str] = None, + ) -> None: + self._base_url = base_url.rstrip("/") + self._ws_base = self._base_url.replace("https://", "wss://").replace("http://", "ws://") + self._api_key = api_key + self._vm_name = vm_name + + # pid → (asyncio.Event, exit_code_cell, ws_task) + self._sessions: dict[int, dict] = {} + + # ------------------------------------------------------------------ + # Auth helpers + # ------------------------------------------------------------------ + + def _auth_headers(self) -> dict: + headers = {} + if self._api_key: + headers["X-API-Key"] = self._api_key + if self._vm_name: + headers["X-Container-Name"] = self._vm_name + return headers + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def create( + self, + command: Optional[str] = None, + cols: int = 80, + rows: int = 24, + on_data: Optional[Callable[[bytes], None]] = None, + cwd: Optional[str] = None, + envs: Optional[dict] = None, + timeout: int = 60, + ) -> PtyHandle: + """Spawn a new PTY session on the remote computer-server. + + Args: + command: Shell command (defaults to ``bash`` on the server side). + cols: Terminal width. + rows: Terminal height. + on_data: Callback invoked with raw bytes whenever the PTY produces + output. Called from the asyncio event loop. + cwd: Working directory on the remote host. + envs: Extra environment variables for the remote process. + timeout: HTTP request timeout in seconds. + + Returns: + :class:`PtyHandle` for the new session. + """ + import aiohttp + + body: dict = {"cols": cols, "rows": rows} + if command is not None: + body["command"] = command + if cwd is not None: + body["cwd"] = cwd + if envs is not None: + body["envs"] = envs + + async with aiohttp.ClientSession() as http: + async with http.post( + f"{self._base_url}/pty", + json=body, + headers=self._auth_headers(), + timeout=aiohttp.ClientTimeout(total=timeout), + ) as resp: + resp.raise_for_status() + data = await resp.json() + + pid: int = data["pid"] + logger.debug("PTY session created: pid=%d", pid) + + # Set up session state + exit_event: asyncio.Event = asyncio.Event() + exit_code_cell: list[int] = [0] + self._sessions[pid] = { + "exit_event": exit_event, + "exit_code": exit_code_cell, + "ws_task": None, + } + + # Open WebSocket for bidirectional I/O + ws_task = asyncio.create_task( + self._ws_reader(pid, on_data, exit_event, exit_code_cell), + name=f"pty-ws-{pid}", + ) + self._sessions[pid]["ws_task"] = ws_task + + return PtyHandle(pid=pid, cols=data["cols"], rows=data["rows"], _iface=self) + + async def kill(self, pid: int) -> bool: + """Kill the remote PTY session *pid*.""" + import aiohttp + + async with aiohttp.ClientSession() as http: + async with http.delete( + f"{self._base_url}/pty/{pid}", + headers=self._auth_headers(), + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + resp.raise_for_status() + data = await resp.json() + return bool(data.get("killed")) + + async def resize(self, pid: int, cols: int, rows: int) -> None: + """Resize the terminal for the remote PTY session *pid*.""" + import aiohttp + + async with aiohttp.ClientSession() as http: + async with http.post( + f"{self._base_url}/pty/{pid}/resize", + json={"cols": cols, "rows": rows}, + headers=self._auth_headers(), + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + resp.raise_for_status() + + async def send_stdin(self, pid: int, data: bytes) -> None: + """Write *data* to the remote PTY session's stdin via the WebSocket. + + If a WebSocket is active the data is sent through it; otherwise falls + back to the HTTP ``/stdin`` endpoint. + """ + sess = self._sessions.get(pid) + ws_task = sess.get("ws_task") if sess else None + + if ws_task and not ws_task.done(): + # Route through the WS task via a shared queue + q = sess.get("stdin_queue") + if q is not None: + await q.put(data) + return + + # Fallback: HTTP POST + import aiohttp + + async with aiohttp.ClientSession() as http: + async with http.post( + f"{self._base_url}/pty/{pid}/stdin", + json={"data": base64.b64encode(data).decode()}, + headers=self._auth_headers(), + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + resp.raise_for_status() + + async def connect( + self, + pid: int, + on_data: Optional[Callable[[bytes], None]] = None, + ) -> PtyHandle: + """Re-open a WebSocket to an existing PTY session *pid*. + + Any previous connection for this pid is cancelled first. + The current terminal dimensions are fetched from the server so that + the returned :class:`PtyHandle` reflects any resizes since creation. + """ + import aiohttp + + sess = self._sessions.get(pid) + if sess: + ws_task = sess.get("ws_task") + if ws_task and not ws_task.done(): + ws_task.cancel() + try: + await ws_task + except (asyncio.CancelledError, Exception): + pass + + # Fetch current dimensions from the server. + cols, rows = 80, 24 + try: + async with aiohttp.ClientSession() as http: + async with http.get( + f"{self._base_url}/pty/{pid}", + headers=self._auth_headers(), + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + if resp.status == 200: + info = await resp.json() + cols = int(info.get("cols", cols)) + rows = int(info.get("rows", rows)) + except Exception as exc: + logger.debug("PTY connect: could not fetch dimensions for pid %d: %s", pid, exc) + + exit_event: asyncio.Event = asyncio.Event() + exit_code_cell: list[int] = [0] + self._sessions[pid] = { + "exit_event": exit_event, + "exit_code": exit_code_cell, + "ws_task": None, + } + + ws_task = asyncio.create_task( + self._ws_reader(pid, on_data, exit_event, exit_code_cell), + name=f"pty-ws-reconnect-{pid}", + ) + self._sessions[pid]["ws_task"] = ws_task + + return PtyHandle(pid=pid, cols=cols, rows=rows, _iface=self) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + async def _disconnect(self, pid: int) -> None: + """Cancel the WS reader task without killing the remote process.""" + sess = self._sessions.get(pid) + if not sess: + return + ws_task = sess.get("ws_task") + if ws_task and not ws_task.done(): + ws_task.cancel() + try: + await ws_task + except (asyncio.CancelledError, Exception): + pass + + async def _wait(self, pid: int) -> int: + """Wait for the exit event and return the exit code. + + Raises: + LookupError: If *pid* is not a tracked session. + """ + sess = self._sessions.get(pid) + if not sess: + raise LookupError(f"PTY session {pid} is not tracked by this interface") + await sess["exit_event"].wait() + return sess["exit_code"][0] + + async def _ws_reader( + self, + pid: int, + on_data: Optional[Callable[[bytes], None]], + exit_event: asyncio.Event, + exit_code_cell: list[int], + ) -> None: + """Background coroutine: connect to the WS and forward messages.""" + import aiohttp + + stdin_queue: asyncio.Queue[bytes] = asyncio.Queue() + sess = self._sessions.setdefault(pid, {}) + sess["stdin_queue"] = stdin_queue + + params = {} + if self._api_key: + params["api_key"] = self._api_key + if self._vm_name: + params["container_name"] = self._vm_name + ws_url = f"{self._ws_base}/pty/{pid}/ws" + try: + async with aiohttp.ClientSession() as http: + async with http.ws_connect(ws_url, params=params) as ws: + + async def _write_stdin(): + while True: + data = await stdin_queue.get() + payload = json.dumps( + { + "type": "stdin", + "data": base64.b64encode(data).decode(), + } + ) + await ws.send_str(payload) + + writer_task = asyncio.create_task(_write_stdin()) + + try: + async for raw_msg in ws: + if raw_msg.type == aiohttp.WSMsgType.TEXT: + try: + msg = json.loads(raw_msg.data) + except json.JSONDecodeError: + continue + if msg.get("type") == "output": + chunk = base64.b64decode(msg["data"]) + if on_data is not None: + on_data(chunk) + elif msg.get("type") == "exit": + exit_code_cell[0] = int(msg.get("code", 0)) + break + elif raw_msg.type in ( + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.ERROR, + ): + break + finally: + writer_task.cancel() + try: + await writer_task + except (asyncio.CancelledError, Exception): + pass + + except asyncio.CancelledError: + pass + except Exception as exc: + logger.debug("PTY WS reader error for pid %d: %s", pid, exc) + finally: + exit_event.set() diff --git a/libs/python/computer/pyproject.toml b/libs/python/computer/pyproject.toml index ca2c91e089..175bb7fa8f 100644 --- a/libs/python/computer/pyproject.toml +++ b/libs/python/computer/pyproject.toml @@ -15,7 +15,7 @@ dependencies = [ "websocket-client>=1.8.0", "websockets>=12.0", "aiohttp>=3.9.0", - "cua-core>=0.1.0,<0.2.0", + "cua-core>=0.1.18,<0.2.0", "pydantic>=2.11.1", "mslex>=1.3.0", ] diff --git a/libs/python/core/README.md b/libs/python/core/README.md index 95d3fd2bef..11bdf49306 100644 --- a/libs/python/core/README.md +++ b/libs/python/core/README.md @@ -2,8 +2,8 @@

- - + + Shows my svg
diff --git a/libs/python/cua-auto/cua_auto/__init__.py b/libs/python/cua-auto/cua_auto/__init__.py index 86ac75729e..5a4680458c 100644 --- a/libs/python/cua-auto/cua_auto/__init__.py +++ b/libs/python/cua-auto/cua_auto/__init__.py @@ -20,6 +20,15 @@ __version__ = "0.1.1" -from cua_auto import clipboard, keyboard, mouse, screen, shell, window +# terminal and shell have no display dependency — always safe to import. +from cua_auto import shell, terminal -__all__ = ["mouse", "keyboard", "screen", "window", "clipboard", "shell"] +# These modules require a display server (pynput, PIL, pywinctl, pyperclip). +# Guard them so that headless environments (CI without X, computer-server inside +# a container) can still import cua_auto.terminal / cua_auto.shell without error. +try: + from cua_auto import clipboard, keyboard, mouse, screen, window +except ImportError: + pass + +__all__ = ["mouse", "keyboard", "screen", "window", "clipboard", "shell", "terminal"] diff --git a/libs/python/cua-auto/cua_auto/terminal.py b/libs/python/cua-auto/cua_auto/terminal.py new file mode 100644 index 0000000000..1ab9cfb231 --- /dev/null +++ b/libs/python/cua-auto/cua_auto/terminal.py @@ -0,0 +1,355 @@ +"""Cross-platform PTY (pseudo-terminal) manager. + +Uses the ``pty`` stdlib on Unix/macOS and ``pywinpty`` on Windows. + +Usage:: + + from cua_auto.terminal import terminal + + output = [] + session = terminal.create("bash", on_data=lambda d: output.append(d)) + terminal.send_stdin(session.pid, b"echo hello\\n") + terminal.send_stdin(session.pid, b"exit\\n") + terminal.wait(session.pid) + print(b"".join(output)) +""" + +from __future__ import annotations + +import os +import sys +import threading +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional + + +@dataclass +class PtySession: + """Lightweight handle returned from :meth:`Terminal.create`.""" + + pid: int + cols: int + rows: int + + +class _PtyProcess: + """Internal state holder for one PTY session.""" + + def __init__(self) -> None: + self.process = None # subprocess.Popen (Unix) or None (Windows) + self.master_fd: Optional[int] = None # Unix master side of the PTY + self.winpty_pty = None # winpty.PtyProcess (Windows) + self.on_data: List[Callable[[bytes], None]] = [] + self.reader_thread: Optional[threading.Thread] = None + self._exit_event: threading.Event = threading.Event() + self.exit_code: Optional[int] = None + + +class Terminal: + """Cross-platform PTY manager. + + Each session is keyed by its process PID. All public methods are + thread-safe. + """ + + def __init__(self) -> None: + self._sessions: Dict[int, _PtyProcess] = {} + self._lock = threading.Lock() + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def create( + self, + command: Optional[str] = None, + cols: int = 80, + rows: int = 24, + on_data: Optional[Callable[[bytes], None]] = None, + cwd: Optional[str] = None, + envs: Optional[Dict[str, str]] = None, + ) -> PtySession: + """Spawn a new PTY session. + + Args: + command: Shell command to run. Defaults to ``bash`` on Unix and + ``powershell`` on Windows. + cols: Initial terminal width (columns). + rows: Initial terminal height (rows). + on_data: Callback invoked from the reader thread with raw bytes + whenever the process writes output. + cwd: Working directory for the spawned process. + envs: Additional environment variables (merged into ``os.environ``). + + Returns: + :class:`PtySession` with the PID, cols, and rows. + """ + if sys.platform == "win32": + return self._create_windows(command, cols, rows, on_data, cwd, envs) + return self._create_unix(command, cols, rows, on_data, cwd, envs) + + def send_stdin(self, pid: int, data: bytes) -> None: + """Write *data* to the stdin of session *pid*.""" + with self._lock: + holder = self._sessions.get(pid) + if holder is None: + raise KeyError(f"No PTY session with pid {pid}") + if holder.master_fd is not None: + try: + os.write(holder.master_fd, data) + except OSError: + pass + elif holder.winpty_pty is not None: + holder.winpty_pty.write(data.decode("utf-8", errors="replace")) + + def resize(self, pid: int, cols: int, rows: int) -> None: + """Resize the terminal window for session *pid*.""" + with self._lock: + holder = self._sessions.get(pid) + if holder is None: + return + if holder.master_fd is not None: + import fcntl + import struct + import termios + + try: + fcntl.ioctl( + holder.master_fd, + termios.TIOCSWINSZ, + struct.pack("HHHH", rows, cols, 0, 0), + ) + except OSError: + pass + elif holder.winpty_pty is not None: + holder.winpty_pty.setwinsize(rows, cols) + + def kill(self, pid: int) -> bool: + """Kill the process for session *pid*. + + Returns: + ``True`` if the signal was delivered successfully. + """ + with self._lock: + holder = self._sessions.get(pid) + if holder is None: + return False + if holder.process is not None: + try: + holder.process.kill() + return True + except Exception: + return False + if holder.winpty_pty is not None: + try: + holder.winpty_pty.terminate() + return True + except Exception: + return False + return False + + def wait(self, pid: int, timeout: Optional[float] = None) -> Optional[int]: + """Block until session *pid* exits and return its exit code. + + Args: + pid: Session PID. + timeout: Maximum seconds to wait. ``None`` means wait forever. + + Returns: + Exit code, or ``None`` if the timeout expired or pid is unknown. + """ + with self._lock: + holder = self._sessions.get(pid) + if holder is None: + return None + holder._exit_event.wait(timeout=timeout) + return holder.exit_code + + def connect( + self, + pid: int, + on_data: Callable[[bytes], None], + ) -> PtySession: + """Attach a new *on_data* callback to an existing session. + + The previous callbacks are replaced. Useful for reconnecting an SSE + or WebSocket consumer to an already-running session. + + Returns: + :class:`PtySession` for the existing session (cols/rows default to + 80×24 since we don't track them after creation). + """ + with self._lock: + holder = self._sessions.get(pid) + if holder is None: + raise KeyError(f"No PTY session with pid {pid}") + with self._lock: + holder.on_data = [on_data] + return PtySession(pid=pid, cols=80, rows=24) + + # ------------------------------------------------------------------ + # Platform-specific internals + # ------------------------------------------------------------------ + + def _create_unix( + self, + command: Optional[str], + cols: int, + rows: int, + on_data: Optional[Callable[[bytes], None]], + cwd: Optional[str], + envs: Optional[Dict[str, str]], + ) -> PtySession: + import fcntl + import pty as _pty + import struct + import subprocess + import termios + + cmd_str = command or "bash" + if isinstance(cmd_str, str): + cmd = ["/bin/sh", "-c", cmd_str] + else: + cmd = cmd_str + + master_fd, slave_fd = _pty.openpty() + + # Set initial terminal size + fcntl.ioctl(slave_fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0)) + + env = os.environ.copy() + if envs: + env.update(envs) + env.setdefault("TERM", "xterm-256color") + + proc = subprocess.Popen( + cmd, + stdin=slave_fd, + stdout=slave_fd, + stderr=slave_fd, + close_fds=True, + cwd=cwd, + env=env, + start_new_session=True, + ) + os.close(slave_fd) + + holder = _PtyProcess() + holder.process = proc + holder.master_fd = master_fd + if on_data is not None: + holder.on_data.append(on_data) + + pid = proc.pid + with self._lock: + self._sessions[pid] = holder + + def _reader() -> None: + try: + while True: + try: + data = os.read(master_fd, 4096) + except OSError: + break + if not data: + break + with self._lock: + callbacks = list(holder.on_data) + for cb in callbacks: + try: + cb(data) + except Exception: + pass + finally: + try: + os.close(master_fd) + except OSError: + pass + with self._lock: + holder.master_fd = None + holder.exit_code = proc.wait() + holder._exit_event.set() + + holder.reader_thread = threading.Thread( + target=_reader, daemon=True, name=f"pty-reader-{pid}" + ) + holder.reader_thread.start() + + return PtySession(pid=pid, cols=cols, rows=rows) + + def _create_windows( + self, + command: Optional[str], + cols: int, + rows: int, + on_data: Optional[Callable[[bytes], None]], + cwd: Optional[str], + envs: Optional[Dict[str, str]], + ) -> PtySession: + try: + import winpty # type: ignore[import] + except ImportError as exc: + raise ImportError( + "pywinpty is required for PTY on Windows. " "Install with: pip install pywinpty" + ) from exc + + cmd = command or "powershell" + + env = None + if envs: + env = os.environ.copy() + env.update(envs) + + pty_proc = winpty.PtyProcess.spawn( + cmd, + cwd=cwd, + env=env, + dimensions=(rows, cols), + ) + + holder = _PtyProcess() + holder.winpty_pty = pty_proc + if on_data is not None: + holder.on_data.append(on_data) + + pid: int = pty_proc.pid + with self._lock: + self._sessions[pid] = holder + + def _reader() -> None: + try: + while pty_proc.isalive(): + try: + data = pty_proc.read(4096) + if data: + raw: bytes = ( + data.encode("utf-8", errors="replace") + if isinstance(data, str) + else data + ) + with self._lock: + callbacks = list(holder.on_data) + for cb in callbacks: + try: + cb(raw) + except Exception: + pass + except Exception: + break + finally: + try: + holder.exit_code = pty_proc.wait() + except Exception: + holder.exit_code = -1 + holder._exit_event.set() + + holder.reader_thread = threading.Thread( + target=_reader, daemon=True, name=f"pty-reader-{pid}" + ) + holder.reader_thread.start() + + return PtySession(pid=pid, cols=cols, rows=rows) + + +# Module-level singleton for convenience +terminal = Terminal() diff --git a/libs/python/cua-auto/pyproject.toml b/libs/python/cua-auto/pyproject.toml index a4f15ee7d7..783690360e 100644 --- a/libs/python/cua-auto/pyproject.toml +++ b/libs/python/cua-auto/pyproject.toml @@ -49,8 +49,12 @@ mss = [ windows = [ "pywin32>=306; sys_platform == 'win32'", ] +# PTY support +pty = [ + "pywinpty>=2.0.0; sys_platform == 'win32'", +] all = [ - "cua-auto[mss,windows]", + "cua-auto[mss,windows,pty]", ] dev = [ "pytest>=8.0.0", diff --git a/libs/python/cua-auto/tests/__init__.py b/libs/python/cua-auto/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/libs/python/cua-auto/tests/test_terminal.py b/libs/python/cua-auto/tests/test_terminal.py new file mode 100644 index 0000000000..900317a7e8 --- /dev/null +++ b/libs/python/cua-auto/tests/test_terminal.py @@ -0,0 +1,312 @@ +"""Unit tests for cua_auto.terminal — cross-platform PTY engine. + +Covers Unix PTY paths (echo, spaces, stdin interaction, exit codes, kill, resize). +Windows-only paths are skipped when not running on win32 (and vice-versa). +""" + +from __future__ import annotations + +import sys +import time + +import pytest + +# All tests in this module require a real PTY, so skip on platforms where we +# can't spawn one without the optional dependency. +pytestmark = pytest.mark.skipif( + sys.platform == "win32", + reason="Unix PTY tests — win32 uses pywinpty (tested separately)", +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _run_command(cmd: str, timeout: float = 3.0) -> tuple[bytes, int]: + """Spawn *cmd* in a PTY, collect all output, return (output, exit_code).""" + from cua_auto.terminal import Terminal + + chunks: list[bytes] = [] + t = Terminal() + session = t.create(command=cmd, cols=80, rows=24, on_data=chunks.append) + exit_code = t.wait(session.pid, timeout=timeout) + return b"".join(chunks), exit_code or 0 + + +# --------------------------------------------------------------------------- +# Import / singleton +# --------------------------------------------------------------------------- + + +class TestTerminalImport: + def test_module_importable(self): + import cua_auto.terminal # noqa: F401 + + def test_singleton_exists(self): + from cua_auto.terminal import Terminal, terminal + + assert isinstance(terminal, Terminal) + + def test_pty_session_dataclass(self): + from cua_auto.terminal import PtySession + + s = PtySession(pid=1234, cols=80, rows=24) + assert s.pid == 1234 + assert s.cols == 80 + assert s.rows == 24 + + +# --------------------------------------------------------------------------- +# Basic echo +# --------------------------------------------------------------------------- + + +class TestEchoBasic: + def test_echo_hello(self): + output, code = _run_command("echo hello") + assert b"hello" in output + assert code == 0 + + def test_echo_with_spaces(self): + output, code = _run_command("echo hello world") + assert b"hello world" in output + assert code == 0 + + def test_echo_leading_trailing_spaces(self): + output, code = _run_command("echo ' spaced '") + assert b"spaced" in output + assert code == 0 + + def test_echo_multiple_words(self): + output, code = _run_command("echo foo bar baz") + assert b"foo" in output + assert b"bar" in output + assert b"baz" in output + assert code == 0 + + def test_echo_empty_string(self): + # echo with an empty-ish arg should still exit 0 + output, code = _run_command("echo ''") + assert code == 0 + + def test_echo_numbers(self): + output, code = _run_command("echo 42") + assert b"42" in output + assert code == 0 + + def test_echo_special_chars(self): + output, code = _run_command("echo 'hello-world_test'") + assert b"hello-world_test" in output + assert code == 0 + + +# --------------------------------------------------------------------------- +# Exit codes +# --------------------------------------------------------------------------- + + +class TestExitCodes: + def test_exit_zero(self): + _, code = _run_command("exit 0") + assert code == 0 + + def test_exit_nonzero(self): + _, code = _run_command("exit 1") + assert code == 1 + + def test_exit_42(self): + _, code = _run_command("exit 42") + assert code == 42 + + def test_true_command(self): + _, code = _run_command("true") + assert code == 0 + + def test_false_command(self): + _, code = _run_command("false") + assert code != 0 + + +# --------------------------------------------------------------------------- +# Interactive stdin (send_stdin) +# --------------------------------------------------------------------------- + + +class TestSendStdin: + def test_send_echo_via_stdin(self): + from cua_auto.terminal import Terminal + + chunks: list[bytes] = [] + t = Terminal() + session = t.create(command="bash", cols=80, rows=24, on_data=chunks.append) + + # Give bash a moment to start + time.sleep(0.2) + t.send_stdin(session.pid, b"echo hello_from_stdin\n") + t.send_stdin(session.pid, b"exit\n") + t.wait(session.pid, timeout=3.0) + + output = b"".join(chunks) + assert b"hello_from_stdin" in output + + def test_send_echo_with_spaces_via_stdin(self): + from cua_auto.terminal import Terminal + + chunks: list[bytes] = [] + t = Terminal() + session = t.create(command="bash", cols=80, rows=24, on_data=chunks.append) + + time.sleep(0.2) + t.send_stdin(session.pid, b"echo 'hello world from stdin'\n") + t.send_stdin(session.pid, b"exit\n") + t.wait(session.pid, timeout=3.0) + + output = b"".join(chunks) + assert b"hello world from stdin" in output + + def test_send_multiple_commands(self): + from cua_auto.terminal import Terminal + + chunks: list[bytes] = [] + t = Terminal() + session = t.create(command="bash", cols=80, rows=24, on_data=chunks.append) + + time.sleep(0.2) + t.send_stdin(session.pid, b"echo first\n") + t.send_stdin(session.pid, b"echo second\n") + t.send_stdin(session.pid, b"exit\n") + t.wait(session.pid, timeout=3.0) + + output = b"".join(chunks) + assert b"first" in output + assert b"second" in output + + def test_exit_code_via_stdin(self): + from cua_auto.terminal import Terminal + + t = Terminal() + session = t.create(command="bash", cols=80, rows=24, on_data=lambda _: None) + + time.sleep(0.2) + t.send_stdin(session.pid, b"exit 7\n") + code = t.wait(session.pid, timeout=3.0) + assert code == 7 + + +# --------------------------------------------------------------------------- +# Kill +# --------------------------------------------------------------------------- + + +class TestKill: + def test_kill_running_session(self): + from cua_auto.terminal import Terminal + + t = Terminal() + # sleep for a long time — we'll kill it + session = t.create(command="sleep 60", cols=80, rows=24, on_data=lambda _: None) + + killed = t.kill(session.pid) + assert killed is True + + # After kill, wait should return quickly + code = t.wait(session.pid, timeout=2.0) + assert code is not None # process has exited + + def test_kill_unknown_pid_returns_false(self): + from cua_auto.terminal import Terminal + + t = Terminal() + assert t.kill(9999999) is False + + def test_wait_unknown_pid_returns_none(self): + from cua_auto.terminal import Terminal + + t = Terminal() + assert t.wait(9999999, timeout=0.1) is None + + +# --------------------------------------------------------------------------- +# Resize +# --------------------------------------------------------------------------- + + +class TestResize: + def test_resize_does_not_raise(self): + from cua_auto.terminal import Terminal + + t = Terminal() + session = t.create(command="bash", cols=80, rows=24, on_data=lambda _: None) + + # Resize should not raise + t.resize(session.pid, 120, 40) + t.resize(session.pid, 80, 24) + + t.send_stdin(session.pid, b"exit\n") + t.wait(session.pid, timeout=2.0) + + def test_resize_unknown_pid_is_noop(self): + from cua_auto.terminal import Terminal + + t = Terminal() + # Should not raise even for unknown pids + t.resize(9999999, 80, 24) + + +# --------------------------------------------------------------------------- +# Connect (callback replacement) +# --------------------------------------------------------------------------- + + +class TestConnect: + def test_connect_replaces_callback(self): + from cua_auto.terminal import Terminal + + first_chunks: list[bytes] = [] + second_chunks: list[bytes] = [] + + t = Terminal() + session = t.create(command="bash", cols=80, rows=24, on_data=first_chunks.append) + + time.sleep(0.2) + t.send_stdin(session.pid, b"echo before_connect\n") + time.sleep(0.1) + + # Re-attach with a new callback + t.connect(session.pid, second_chunks.append) + + t.send_stdin(session.pid, b"echo after_connect\n") + t.send_stdin(session.pid, b"exit\n") + t.wait(session.pid, timeout=3.0) + + # "after_connect" should appear in the second callback's data + assert b"after_connect" in b"".join(second_chunks) + + def test_connect_unknown_pid_raises(self): + from cua_auto.terminal import Terminal + + t = Terminal() + with pytest.raises(KeyError): + t.connect(9999999, lambda _: None) + + +# --------------------------------------------------------------------------- +# Singleton convenience +# --------------------------------------------------------------------------- + + +class TestSingletonTerminal: + def test_singleton_echo(self): + from cua_auto.terminal import terminal + + chunks: list[bytes] = [] + session = terminal.create( + command="echo singleton_works", + cols=80, + rows=24, + on_data=chunks.append, + ) + terminal.wait(session.pid, timeout=3.0) + assert b"singleton_works" in b"".join(chunks) diff --git a/libs/python/cua-auto/tests/test_terminal_windows.py b/libs/python/cua-auto/tests/test_terminal_windows.py new file mode 100644 index 0000000000..96780a5883 --- /dev/null +++ b/libs/python/cua-auto/tests/test_terminal_windows.py @@ -0,0 +1,299 @@ +"""Windows-specific PTY tests for cua_auto.terminal. + +Uses pywinpty under the hood via the Terminal._create_windows path. +All tests are skipped on non-Windows platforms. +""" + +from __future__ import annotations + +import sys +import time + +import pytest + +pytestmark = pytest.mark.skipif( + sys.platform != "win32", + reason="Windows PTY tests — requires pywinpty", +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _run_command(cmd: str, timeout: float = 10.0) -> tuple[bytes, int]: + """Spawn *cmd* via PowerShell in a PTY, collect output, return (output, exit_code).""" + from cua_auto.terminal import Terminal + + chunks: list[bytes] = [] + t = Terminal() + # Wrap in powershell -Command so we can use PS syntax + full_cmd = f'powershell -NoProfile -NonInteractive -Command "{cmd}"' + session = t.create(command=full_cmd, cols=80, rows=24, on_data=chunks.append) + exit_code = t.wait(session.pid, timeout=timeout) + return b"".join(chunks), exit_code or 0 + + +# --------------------------------------------------------------------------- +# Import / singleton +# --------------------------------------------------------------------------- + + +class TestTerminalImport: + def test_module_importable(self): + import cua_auto.terminal # noqa: F401 + + def test_singleton_exists(self): + from cua_auto.terminal import Terminal, terminal + + assert isinstance(terminal, Terminal) + + def test_pty_session_dataclass(self): + from cua_auto.terminal import PtySession + + s = PtySession(pid=1234, cols=80, rows=24) + assert s.pid == 1234 + assert s.cols == 80 + assert s.rows == 24 + + +# --------------------------------------------------------------------------- +# Basic echo +# --------------------------------------------------------------------------- + + +class TestEchoBasic: + def test_echo_hello(self): + output, code = _run_command("Write-Output hello") + assert b"hello" in output + assert code == 0 + + def test_echo_with_spaces(self): + output, code = _run_command("Write-Output 'hello world'") + assert b"hello world" in output + assert code == 0 + + def test_echo_numbers(self): + output, code = _run_command("Write-Output 42") + assert b"42" in output + assert code == 0 + + def test_echo_multiple_words(self): + output, code = _run_command("Write-Output 'foo bar baz'") + assert b"foo" in output + assert b"bar" in output + assert b"baz" in output + assert code == 0 + + +# --------------------------------------------------------------------------- +# Exit codes +# --------------------------------------------------------------------------- + + +class TestExitCodes: + def test_exit_zero(self): + _, code = _run_command("exit 0") + assert code == 0 + + def test_exit_nonzero(self): + # pywinpty/ConPTY does not propagate PowerShell's exit code reliably; + # cmd.exe /c exit N does propagate correctly. + from cua_auto.terminal import Terminal + + t = Terminal() + session = t.create(command="cmd /c exit 1", cols=80, rows=24, on_data=lambda _: None) + code = t.wait(session.pid, timeout=5.0) + assert code == 1 + + def test_true_command(self): + _, code = _run_command("$true | Out-Null") + assert code == 0 + + def test_false_exits_nonzero(self): + # pywinpty/ConPTY does not propagate PowerShell's exit code reliably; + # cmd.exe /c exit N does propagate correctly. + from cua_auto.terminal import Terminal + + t = Terminal() + session = t.create(command="cmd /c exit 42", cols=80, rows=24, on_data=lambda _: None) + code = t.wait(session.pid, timeout=5.0) + assert code == 42 + + +# --------------------------------------------------------------------------- +# Interactive stdin (send_stdin) +# --------------------------------------------------------------------------- + + +class TestSendStdin: + def test_send_echo_via_stdin(self): + from cua_auto.terminal import Terminal + + chunks: list[bytes] = [] + t = Terminal() + session = t.create( + command="powershell -NoProfile -NonInteractive", + cols=80, + rows=24, + on_data=chunks.append, + ) + + time.sleep(0.5) + t.send_stdin(session.pid, b"Write-Output hello_from_stdin\r\n") + t.send_stdin(session.pid, b"exit\r\n") + t.wait(session.pid, timeout=10.0) + + output = b"".join(chunks) + assert b"hello_from_stdin" in output + + def test_send_multiple_commands(self): + from cua_auto.terminal import Terminal + + chunks: list[bytes] = [] + t = Terminal() + session = t.create( + command="powershell -NoProfile -NonInteractive", + cols=80, + rows=24, + on_data=chunks.append, + ) + + time.sleep(0.5) + t.send_stdin(session.pid, b"Write-Output first\r\n") + t.send_stdin(session.pid, b"Write-Output second\r\n") + t.send_stdin(session.pid, b"exit\r\n") + t.wait(session.pid, timeout=10.0) + + output = b"".join(chunks) + assert b"first" in output + assert b"second" in output + + +# --------------------------------------------------------------------------- +# Kill +# --------------------------------------------------------------------------- + + +class TestKill: + def test_kill_running_session(self): + from cua_auto.terminal import Terminal + + t = Terminal() + session = t.create( + command="powershell -NoProfile -NonInteractive -Command Start-Sleep 60", + cols=80, + rows=24, + on_data=lambda _: None, + ) + + time.sleep(0.3) + killed = t.kill(session.pid) + assert killed is True + + code = t.wait(session.pid, timeout=5.0) + assert code is not None + + def test_kill_unknown_pid_returns_false(self): + from cua_auto.terminal import Terminal + + t = Terminal() + assert t.kill(9999999) is False + + def test_wait_unknown_pid_returns_none(self): + from cua_auto.terminal import Terminal + + t = Terminal() + assert t.wait(9999999, timeout=0.1) is None + + +# --------------------------------------------------------------------------- +# Resize +# --------------------------------------------------------------------------- + + +class TestResize: + def test_resize_does_not_raise(self): + from cua_auto.terminal import Terminal + + t = Terminal() + session = t.create( + command="powershell -NoProfile -NonInteractive", + cols=80, + rows=24, + on_data=lambda _: None, + ) + + time.sleep(0.3) + t.resize(session.pid, 120, 40) + t.resize(session.pid, 80, 24) + + t.send_stdin(session.pid, b"exit\r\n") + t.wait(session.pid, timeout=5.0) + + def test_resize_unknown_pid_is_noop(self): + from cua_auto.terminal import Terminal + + t = Terminal() + t.resize(9999999, 80, 24) # should not raise + + +# --------------------------------------------------------------------------- +# Connect (callback replacement) +# --------------------------------------------------------------------------- + + +class TestConnect: + def test_connect_replaces_callback(self): + from cua_auto.terminal import Terminal + + first_chunks: list[bytes] = [] + second_chunks: list[bytes] = [] + + t = Terminal() + session = t.create( + command="powershell -NoProfile -NonInteractive", + cols=80, + rows=24, + on_data=first_chunks.append, + ) + + time.sleep(0.5) + t.send_stdin(session.pid, b"Write-Output before_connect\r\n") + time.sleep(0.3) + + t.connect(session.pid, second_chunks.append) + + t.send_stdin(session.pid, b"Write-Output after_connect\r\n") + t.send_stdin(session.pid, b"exit\r\n") + t.wait(session.pid, timeout=10.0) + + assert b"after_connect" in b"".join(second_chunks) + + def test_connect_unknown_pid_raises(self): + from cua_auto.terminal import Terminal + + t = Terminal() + with pytest.raises(KeyError): + t.connect(9999999, lambda _: None) + + +# --------------------------------------------------------------------------- +# Singleton convenience +# --------------------------------------------------------------------------- + + +class TestSingletonTerminal: + def test_singleton_echo(self): + from cua_auto.terminal import terminal + + chunks: list[bytes] = [] + session = terminal.create( + command='powershell -NoProfile -NonInteractive -Command "Write-Output singleton_works"', + cols=80, + rows=24, + on_data=chunks.append, + ) + terminal.wait(session.pid, timeout=10.0) + assert b"singleton_works" in b"".join(chunks) diff --git a/libs/python/cua-cli/cua_cli/commands/do.py b/libs/python/cua-cli/cua_cli/commands/do.py index 2a68ba31cf..d7697cc699 100644 --- a/libs/python/cua-cli/cua_cli/commands/do.py +++ b/libs/python/cua-cli/cua_cli/commands/do.py @@ -1142,15 +1142,15 @@ async def _run() -> int: return run_async(_run()) -def _cmd_shell(args: argparse.Namespace) -> int: - from cua_cli.utils.async_utils import run_async +def _default_shell() -> str: + return "powershell" if sys.platform == "win32" else "bash" - t = _require_target() - if not t: - return 1 - command = " ".join(args.command) - if not command.strip(): +def _cmd_shell_noninteractive(command: str | None, args: argparse.Namespace) -> int: + """Run a non-interactive shell command via run_command (original behaviour).""" + from cua_cli.utils.async_utils import run_async + + if not command or not command.strip(): return _fail("No command provided") async def _run() -> int: @@ -1180,6 +1180,286 @@ async def _run() -> int: return run_async(_run()) +def _shell_host_pty(command: str | None, cols: int | None = None, rows: int | None = None) -> int: + """Interactive PTY session on the local host via cua_auto.terminal.""" + import shutil + import signal + import threading + + try: + import cua_auto.terminal as _term + except ImportError as e: + return _fail(f"cua-auto not installed: {e}") + + _auto_cols, _auto_rows = shutil.get_terminal_size((80, 24)) + cols = cols if cols is not None else _auto_cols + rows = rows if rows is not None else _auto_rows + + def _on_data(data: bytes) -> None: + try: + sys.stdout.buffer.write(data) + sys.stdout.buffer.flush() + except Exception: + pass + + session = _term.terminal.create( + command=command or _default_shell(), + cols=cols, + rows=rows, + on_data=_on_data, + ) + + if sys.platform == "win32": + # Windows: use msvcrt for raw stdin reading + import msvcrt + + def _stdin_loop() -> None: + while True: + try: + ch = msvcrt.getch() + if not ch: + break + _term.terminal.send_stdin(session.pid, ch) + except Exception: + break + + stdin_thread = threading.Thread(target=_stdin_loop, daemon=True) + stdin_thread.start() + exit_code = _term.terminal.wait(session.pid) + + else: + # Unix/macOS: set raw mode, SIGWINCH for resize + import termios + import tty + + old_settings = termios.tcgetattr(sys.stdin.fileno()) + + def _resize(_sig=None, _frame=None) -> None: + c, r = shutil.get_terminal_size((80, 24)) + _term.terminal.resize(session.pid, c, r) + + signal.signal(signal.SIGWINCH, _resize) + tty.setraw(sys.stdin.fileno()) + + def _stdin_loop() -> None: + while True: + try: + data = os.read(sys.stdin.fileno(), 1024) + if not data: + break + _term.terminal.send_stdin(session.pid, data) + except Exception: + break + + stdin_thread = threading.Thread(target=_stdin_loop, daemon=True) + try: + stdin_thread.start() + exit_code = _term.terminal.wait(session.pid) + finally: + try: + termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_settings) + except Exception: + pass + signal.signal(signal.SIGWINCH, signal.SIG_DFL) + + return exit_code or 0 + + +async def _shell_remote_pty( + provider: str, name: str, command: str | None, cols: int | None = None, rows: int | None = None +) -> int: + """Interactive PTY session via WebSocket to a remote computer-server.""" + import asyncio + import shutil + import signal + import threading + + import aiohttp + + _auto_cols, _auto_rows = shutil.get_terminal_size((80, 24)) + cols = cols if cols is not None else _auto_cols + rows = rows if rows is not None else _auto_rows + + try: + api_url = await _get_api_url(provider, name) + except Exception as e: + return _fail(str(e)) + + ws_url = api_url.replace("https://", "wss://").replace("http://", "ws://") + + # Build auth headers (for the initial POST) and query params (for the WS). + headers: dict = {} + ws_params: dict = {} + if provider in ("cloud", "cloudv2"): + from cua_cli.auth.store import get_api_key + + api_key = get_api_key() + if api_key: + headers["X-API-Key"] = api_key + ws_params["api_key"] = api_key + if name: + headers["X-Container-Name"] = name + ws_params["container_name"] = name + + # Create PTY session + try: + async with aiohttp.ClientSession() as http: + async with http.post( + f"{api_url}/pty", + json={"command": command or _default_shell(), "cols": cols, "rows": rows}, + headers={**headers, "Content-Type": "application/json"}, + timeout=aiohttp.ClientTimeout(total=15), + ) as resp: + resp.raise_for_status() + data = await resp.json() + except Exception as e: + return _fail(f"Failed to create PTY session: {e}") + + pid: int = data["pid"] + + exit_code_cell: list[int] = [0] + done_event = asyncio.Event() + + if sys.platform == "win32": + # Windows: use msvcrt for raw stdin + import msvcrt + + async def _run_ws() -> None: + async with aiohttp.ClientSession() as http: + async with http.ws_connect(f"{ws_url}/pty/{pid}/ws", params=ws_params) as ws: + + def _stdin_loop() -> None: + while not done_event.is_set(): + try: + ch = msvcrt.getch() + if ch: + encoded = base64.b64encode(ch).decode() + asyncio.run_coroutine_threadsafe( + ws.send_str(json.dumps({"type": "stdin", "data": encoded})), + asyncio.get_event_loop(), + ) + except Exception: + break + + t = threading.Thread(target=_stdin_loop, daemon=True) + t.start() + + async for raw_msg in ws: + if raw_msg.type == aiohttp.WSMsgType.TEXT: + try: + msg = json.loads(raw_msg.data) + except Exception: + continue + if msg.get("type") == "output": + chunk = base64.b64decode(msg["data"]) + sys.stdout.buffer.write(chunk) + sys.stdout.buffer.flush() + elif msg.get("type") == "exit": + exit_code_cell[0] = int(msg.get("code", 0)) + break + elif raw_msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): + break + done_event.set() + + await _run_ws() + + else: + import termios + import tty + + old_settings = termios.tcgetattr(sys.stdin.fileno()) + tty.setraw(sys.stdin.fileno()) + + loop = asyncio.get_event_loop() + + async def _run_ws() -> None: + async with aiohttp.ClientSession() as http: + async with http.ws_connect(f"{ws_url}/pty/{pid}/ws", params=ws_params) as ws: + + def _resize(_sig=None, _frame=None) -> None: + c, r = shutil.get_terminal_size((80, 24)) + asyncio.run_coroutine_threadsafe( + ws.send_str(json.dumps({"type": "resize", "cols": c, "rows": r})), + loop, + ) + + signal.signal(signal.SIGWINCH, _resize) + + def _stdin_loop() -> None: + while not done_event.is_set(): + try: + data = os.read(sys.stdin.fileno(), 1024) + if not data: + break + encoded = base64.b64encode(data).decode() + asyncio.run_coroutine_threadsafe( + ws.send_str(json.dumps({"type": "stdin", "data": encoded})), + loop, + ) + except Exception: + break + + t = threading.Thread(target=_stdin_loop, daemon=True) + t.start() + + async for raw_msg in ws: + if raw_msg.type == aiohttp.WSMsgType.TEXT: + try: + msg = json.loads(raw_msg.data) + except Exception: + continue + if msg.get("type") == "output": + chunk = base64.b64decode(msg["data"]) + sys.stdout.buffer.write(chunk) + sys.stdout.buffer.flush() + elif msg.get("type") == "exit": + exit_code_cell[0] = int(msg.get("code", 0)) + break + elif raw_msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): + break + + done_event.set() + + try: + await _run_ws() + finally: + try: + termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings) + except Exception: + pass + signal.signal(signal.SIGWINCH, signal.SIG_DFL) + + return exit_code_cell[0] + + +def _cmd_shell(args: argparse.Namespace) -> int: + command_parts = getattr(args, "shell_command", []) + command = " ".join(command_parts).strip() if command_parts else None + + t = _require_target() + if not t: + return 1 + + # Non-interactive (piped / scripted): keep old run_command behaviour + if not sys.stdin.isatty(): + return _cmd_shell_noninteractive(command, args) + + # Interactive PTY mode + state = _load_state() + provider = state["provider"] + name = state.get("name", "") + + cols: int | None = getattr(args, "cols", None) + rows: int | None = getattr(args, "rows", None) + + if provider == "host": + return _shell_host_pty(command, cols=cols, rows=rows) + + from cua_cli.utils.async_utils import run_async + + return run_async(_shell_remote_pty(provider, name, command, cols=cols, rows=rows)) + + def _cmd_open(args: argparse.Namespace) -> int: from cua_cli.utils.async_utils import run_async @@ -1511,8 +1791,13 @@ def register_parser(subparsers: argparse._SubParsersAction) -> None: dr.add_argument("y2", type=int) # shell - sh = sub.add_parser("shell", help="Run a shell command in the VM") - sh.add_argument("command", nargs=argparse.REMAINDER) + sh = sub.add_parser( + "shell", + help="Run a shell command (or open an interactive terminal) in the VM", + ) + sh.add_argument("shell_command", nargs=argparse.REMAINDER) + sh.add_argument("--cols", type=int, default=None, help="Terminal width (default: auto-detect)") + sh.add_argument("--rows", type=int, default=None, help="Terminal height (default: auto-detect)") # open op = sub.add_parser("open", help="Open a file or URL") diff --git a/libs/python/cua-cli/pyproject.toml b/libs/python/cua-cli/pyproject.toml index af44374d75..ba58dbfbb5 100644 --- a/libs/python/cua-cli/pyproject.toml +++ b/libs/python/cua-cli/pyproject.toml @@ -29,9 +29,9 @@ requires-python = ">=3.12,<3.14" dependencies = [ # Core CUA packages "cua-computer>=0.5.0", - "cua-core>=0.1.0", + "cua-core>=0.1.18", # Host automation (used by cua do switch host) - "cua-auto>=0.1.0", + "cua-auto[all]>=0.1.0", # HTTP client "aiohttp>=3.9.0", # CLI output diff --git a/libs/python/som/README.md b/libs/python/som/README.md index 151a57517f..cb53e9c1c7 100644 --- a/libs/python/som/README.md +++ b/libs/python/som/README.md @@ -2,8 +2,8 @@

- - + + Shows my svg
diff --git a/libs/typescript/core/README.md b/libs/typescript/core/README.md index 1b003833ab..53ec38bfc7 100644 --- a/libs/typescript/core/README.md +++ b/libs/typescript/core/README.md @@ -2,8 +2,8 @@

- - + + Shows my svg
diff --git a/scripts/docs-generators/python-sdk.ts b/scripts/docs-generators/python-sdk.ts index 6541ae3681..a92e6dbee0 100644 --- a/scripts/docs-generators/python-sdk.ts +++ b/scripts/docs-generators/python-sdk.ts @@ -198,8 +198,13 @@ async function main() { console.log(` Extracting documentation from ${config.packageDir}...`); let docs: PythonPackage; try { + // Prefer uv run --with griffe python (works cross-platform), fall back to python3 + const pythonCmd = + process.platform === 'win32' + ? `uv run --with griffe python` + : `python3`; const output = execSync( - `python3 "${PYTHON_SCRIPT}" "${packagePath}" "${config.packageName}"`, + `${pythonCmd} "${PYTHON_SCRIPT}" "${packagePath}" "${config.packageName}"`, { encoding: 'utf-8', cwd: ROOT_DIR,