calute.operators.pty

calute.operators.pty#

Persistent PTY session management for operator tooling.

Provides PTYSessionManager, which creates and manages pseudo-terminal (PTY) backed subprocess sessions. Each session is identified by a unique session_id and can be written to, read from, interrupted, or closed independently.

class calute.operators.pty.PTYSession(session_id: str, process: Popen[str], master_fd: int, command: str, workdir: str)[source]#

Bases: object

Live PTY-backed subprocess session.

session_id#

Unique identifier for this session, generated by PTYSessionManager.

Type

str

process#

The underlying subprocess.Popen object that owns the child process.

Type

subprocess.Popen[str]

master_fd#

File descriptor for the master side of the PTY pair. Used for reading output and writing input.

Type

int

command#

The original shell command that was launched.

Type

str

workdir#

Absolute path of the working directory the session was started in.

Type

str

command: str#
master_fd: int#
process: Popen[str]#
session_id: str#
workdir: str#
class calute.operators.pty.PTYSessionManager[source]#

Bases: object

Manage persistent PTY-backed shell sessions.

Maintains a registry of active PTYSession instances keyed by session_id. Sessions can be created, written to, listed, and closed.

_sessions#

Internal mapping from session_id to the corresponding PTYSession.

close(session_id: str) → dict[str, Any][source]#

Terminate and forget a PTY session.

Attempts a graceful SIGTERM first, escalating to SIGKILL if the process does not exit within two seconds. The master file descriptor is closed and the session is removed from the registry.

Parameters

session_id – Identifier of the PTY session to close.

Returns

  • session_id: The session identifier.

  • closed: Always True.

  • exit_code: Final process exit code.

Return type

A dictionary containing

Raises

ValueError – If the session_id is not recognised.

create_session(cmd: str, *, workdir: str | None = None, env: dict[str, str] | None = None, login: bool = True, yield_time_ms: int = 1000, max_output_chars: int = 4000) → dict[str, Any][source]#

Start a PTY session and return its initial output.

Creates a new pseudo-terminal pair, spawns a shell process with the given command, and captures initial output up to the specified limits.

Parameters

  • cmd – Shell command to execute inside the PTY.

  • workdir – Working directory for the new session. When None, the current process working directory is used.

  • env – Additional environment variables merged on top of the current process environment.

  • login – When True, the shell is started with login semantics (-l flag) for bash and zsh so that profile files are sourced.

  • yield_time_ms – Maximum time in milliseconds to wait for initial output before returning.

  • max_output_chars – Maximum number of characters to capture from the initial output.

Returns

  • session_id: Unique identifier for the new session.

  • command: The shell command that was launched.

  • workdir: Resolved working directory path.

  • stdout: Initial output captured from the PTY.

  • running: True if the process is still alive.

  • exit_code: Process exit code, or None if still running.

Return type

A dictionary containing

list_sessions() → list[dict[str, Any]][source]#

Return summaries for current PTY sessions.

Returns

A list of dictionaries, each containing the session_id, original command, workdir, running status, and exit_code of a tracked session.

write(session_id: str, *, chars: str = '', close_stdin: bool = False, interrupt: bool = False, yield_time_ms: int = 1000, max_output_chars: int = 4000) → dict[str, Any][source]#

Write to a PTY session and read back incremental output.

Sends characters, an EOF signal, or an interrupt to the running process and then collects whatever output the process produces within the specified time window.

Parameters

  • session_id – Identifier of the target PTY session.

  • chars – Text to write to the process stdin. Include \n to simulate pressing Enter.

  • close_stdin – When True, send an EOF (Ctrl-D) to the process after writing any provided characters.

  • interrupt – When True, send SIGINT to the process group before reading output.

  • yield_time_ms – Maximum time in milliseconds to wait for output after sending input.

  • max_output_chars – Maximum number of output characters to collect.

Returns

  • session_id: The session identifier.

  • stdout: Incremental output captured since the write.

  • running: True if the process is still alive.

  • exit_code: Process exit code, or None if still running.

Return type

A dictionary containing

Raises

ValueError – If the session_id is not recognised.

Contents