Files
wehub-resource-sync 91e75e620b
CI: cua-driver distro-compat matrix / debian:12 (glibc 2.36) (push) Has been cancelled
CI: SPDX Headers / Check SPDX headers (warn-only) (push) Has been cancelled
CD: Docs MCP Server / build (linux/amd64) (push) Has been cancelled
CD: Docs MCP Server / build (linux/arm64) (push) Has been cancelled
CD: Docs MCP Server / merge (push) Has been cancelled
CI: cua-driver distro-compat matrix / Resolve release version (push) Has been cancelled
CI: cua-driver distro-compat matrix / fedora:41 (glibc 2.40) (push) Has been cancelled
CI: cua-driver distro-compat matrix / rockylinux:9 (glibc 2.34) (push) Has been cancelled
CI: cua-driver distro-compat matrix / ubuntu:22.04 (glibc 2.35) (push) Has been cancelled
CI: cua-driver distro-compat matrix / ubuntu:24.04 (glibc 2.39) (push) Has been cancelled
CI: cua-driver distro-compat matrix / Distro compat summary (push) Has been cancelled
CI: Rust Linux unit / Rust Linux unit and compile (push) Has been cancelled
CI: Rust Windows unit / Rust Windows unit and compile (push) Has been cancelled
CI: Nix Linux Rust source / Nix / compositor build (push) Has been cancelled
CI: Nix Linux Rust source / Nix / driver package (push) Has been cancelled
CI: Nix Linux Rust source / Nix / Rust unit tests (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:03:19 +08:00

31 lines
982 B
Python

"""Terminal (PTY) interface, backed by a Transport."""
from __future__ import annotations
from typing import Optional
from cua_sandbox.transport.base import Transport
class Terminal:
"""PTY terminal sessions."""
def __init__(self, transport: Transport):
self._t = transport
async def create(self, command: Optional[str] = None, cols: int = 80, rows: int = 24) -> dict:
"""Create a new PTY session. Returns {"pid": int, "cols": int, "rows": int}."""
return await self._t.pty_create(command=command, cols=cols, rows=rows)
async def send_input(self, pid: int, data: str) -> None:
"""Send input to a PTY session."""
await self._t.pty_send(pid, data)
async def info(self, pid: int) -> Optional[dict]:
"""Return session info or None if gone."""
return await self._t.pty_info(pid)
async def close(self, pid: int) -> bool:
"""Kill a PTY session."""
return await self._t.pty_kill(pid)