Session Management (Ops)

Unified, config-driven session management for tmux and containers (Docker/Podman).

kstlib.ops wraps tmux and Docker/Podman behind a single API to start, stop, attach, monitor, and collect logs from persistent processes. All session definitions live in kstlib.conf.yml, making the config file the single source of truth.

Container interoperability: the list, status, logs, attach, and stop commands work with any container running on the host, not only those created by kstlib. The start command creates new containers from an image definition.

Tip

Pair this reference with Ops for the feature guide.

Quick overview

  • SessionManager is the main facade for session lifecycle management

  • TmuxRunner provides direct tmux session control

  • ContainerRunner provides direct Podman/Docker container control

  • SessionConfig and SessionStatus are the core data models

  • Configuration follows the standard priority chain: constructor args > kstlib.conf.yml > defaults

Configuration cascade

The module consults the loaded config for session definitions:

ops:
  default_backend: tmux
  tmux_binary: tmux
  container_runtime: null  # auto-detect (podman or docker)

  sessions:
    astro:
      backend: tmux
      command: "python -m astro.bot"
      working_dir: "/opt/astro"

    astro-prod:
      backend: container
      image: "astro-bot:latest"
      volumes:
        - "./data:/app/data"

Load a session from config:

from kstlib.ops import SessionManager

session = SessionManager.from_config("astro")

Note

Sessions defined in config are visible in kstlib ops list with state DEFINED even before they are started. The container_runtime field accepts null for auto-detection (checks podman first, then docker). CLI arguments override config values when both are provided.

Usage patterns

Basic tmux session

from kstlib.ops import SessionManager

session = SessionManager("dev", backend="tmux")
session.start("python app.py")
session.attach()  # Ctrl+B D to detach
session.stop()

Container with volumes

from kstlib.ops import SessionManager

session = SessionManager(
    "prod",
    backend="container",
    image="app:latest",
    volumes=["./data:/app/data"],
    log_volume="./logs:/app/logs",
)
session.start()

Session lifecycle

from kstlib.ops import SessionManager

session = SessionManager.from_config("astro")

# Check state
if session.exists():
    status = session.status()
    print(f"State: {status.state.value}, PID: {status.pid}")

# Logs
logs = session.logs(lines=100)

# Running check
if session.is_running():
    session.stop(graceful=True, timeout=10)

Low-level runner access

from kstlib.ops import TmuxRunner, ContainerRunner, SessionConfig, BackendType

# Direct tmux
tmux = TmuxRunner()
config = SessionConfig(name="test", backend=BackendType.TMUX, command="bash")
tmux.start(config)
tmux.send_keys("test", "echo hello")
tmux.stop("test")

# Direct container
container = ContainerRunner(runtime="podman")
config = SessionConfig(
    name="test",
    backend=BackendType.CONTAINER,
    image="python:3.10-slim",
)
container.start(config)
container.exec("test", "pip list")
container.stop("test")

Module reference

Session management with tmux and container backends.

This module provides config-driven session management for running persistent processes like trading bots. It supports two backends:

  • tmux: For local development and backtesting with detach/attach

  • container: For production with Podman/Docker and persistent logs

The module is designed around the principle of backend abstraction, allowing the same code to work with either tmux sessions or containers.

Examples

Local development with tmux:

>>> from kstlib.ops import SessionManager
>>> session = SessionManager("astro", backend="tmux")
>>> session.start("python -m astro.bot")  
>>> session.attach()  # tmux attach-session -t astro

Production with containers:

>>> session = SessionManager(
...     "astro",
...     backend="container",
...     image="astro-bot:latest",
... )
>>> session.start()  
>>> session.attach()  # podman attach astro

Config-driven usage:

>>> session = SessionManager.from_config("astro")  
>>> session.start()

Note

The attach() method uses os.execvp and replaces the current process. It does not return on success.

class kstlib.ops.AbstractRunner(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the interface for session runners.

All session runners (TmuxRunner, ContainerRunner) must implement this protocol to ensure consistent behavior across backends.

The protocol defines core operations for session lifecycle management: - start: Create and start a new session - stop: Stop a running session - attach: Attach terminal to a session (replaces current process) - status: Get current session status - logs: Retrieve session logs - exists: Check if a session exists - list_sessions: List all sessions managed by this runner

Examples

>>> def run_session(runner: AbstractRunner, config: SessionConfig) -> None:
...     status = runner.start(config)
...     print(f"Session {status.name} started with PID {status.pid}")
start(self, config: 'SessionConfig') → 'SessionStatus' -> SessionStatus[source]

Create and start a new session.

Parameters:

config (SessionConfig) – Session configuration including name, command, etc.

Returns:

SessionStatus with current state after starting.

Raises:
Return type:

SessionStatus

stop(self, name: 'str', *, graceful: 'bool' = True, timeout: 'int' = 10) → 'bool' -> bool[source]

Stop a running session.

Parameters:

  • name (str) – Session name to stop.

  • graceful (bool) – If True, send SIGTERM first and wait for graceful shutdown. If False, send SIGKILL immediately.

  • timeout (int) – Seconds to wait for graceful shutdown before forcing.

Returns:

True if the session was stopped, False if it was not running.

Raises:
Return type:

bool

attach(self, name: 'str') → 'None' -> None[source]

Attach terminal to a running session.

This method replaces the current process with the attach command using os.execvp. It does not return on success.

Parameters:

name (str) – Session name to attach to.

Raises:

Note

This method uses os.execvp and does not return on success. The calling process is replaced by the attach command.

status(self, name: 'str') → 'SessionStatus' -> SessionStatus[source]

Get the current status of a session.

Parameters:

name (str) – Session name to query.

Returns:

SessionStatus with current state information.

Raises:

SessionNotFoundError – If the session does not exist.

Return type:

SessionStatus

logs(self, name: 'str', lines: 'int' = 100) → 'str' -> str[source]

Retrieve recent logs from a session.

Parameters:

  • name (str) – Session name to get logs from.

  • lines (int) – Number of lines to retrieve (default 100).

Returns:

String containing the log output with ANSI codes preserved.

Raises:

SessionNotFoundError – If the session does not exist.

Return type:

str

exists(self, name: 'str') → 'bool' -> bool[source]

Check if a session with the given name exists.

Parameters:

name (str) – Session name to check.

Returns:

True if the session exists, False otherwise.

Return type:

bool

list_sessions(self) → 'list[SessionStatus]' -> list[SessionStatus][source]

List all sessions managed by this runner.

Returns:

List of SessionStatus for all sessions. Empty list if no sessions exist.

Return type:

list[SessionStatus]

__init__(self, *args, **kwargs)
exception kstlib.ops.BackendNotFoundError[source]

Bases: OpsError, FileNotFoundError

Backend binary (tmux, podman, docker) not found in PATH.

Raised when the required backend binary is not installed or not accessible from the current PATH.

class kstlib.ops.BackendType(value)[source]

Bases: str, Enum

Backend type for session management.

TMUX

Use tmux for session management (dev/local).

CONTAINER

Use Podman/Docker for session management (prod).

TMUX = 'tmux'
CONTAINER = 'container'
__format__(self, format_spec)

Returns format using actual value type unless __str__ has been overridden.

class kstlib.ops.ContainerRunner(runtime=None)[source]

Bases: object

Container runner for Podman/Docker containers.

Manages containers for running persistent processes with pseudo-terminal support for TUI applications.

Parameters:

runtime (str | None) – Container runtime to use (“podman” or “docker”).

runtime

The container runtime name.

binary

The validated runtime binary path.

Examples

>>> runner = ContainerRunner()  # Uses podman by default  
>>> runner = ContainerRunner(runtime="docker")  
>>> config = SessionConfig(
...     name="app",
...     backend=BackendType.CONTAINER,
...     image="python:3.10-slim",
... )
>>> status = runner.start(config)
__init__(self, runtime: 'str | None' = None) → 'None' -> None[source]

Initialize ContainerRunner.

Parameters:

runtime (str | None) – Container runtime (“podman”, “docker”, or None for auto-detect). Auto-detection tries podman first, then docker.

property runtime: str

Return the configured runtime name.

property binary: str

Return validated container runtime binary path.

Raises:

ContainerRuntimeNotFoundError – If runtime is not installed.

start(self, config: 'SessionConfig') → 'SessionStatus' -> SessionStatus[source]

Create and start a new container.

Parameters:

config (SessionConfig) – Session configuration with image and options.

Returns:

SessionStatus with state information.

Raises:
Return type:

SessionStatus

stop(self, name: 'str', *, graceful: 'bool' = True, timeout: 'int' = 10) → 'bool' -> bool[source]

Stop a running container.

Parameters:

  • name (str) – Container name to stop.

  • graceful (bool) – If True, use stop with timeout. If False, use kill.

  • timeout (int) – Seconds to wait for graceful shutdown.

Returns:

True if stopped, False if not running.

Raises:
Return type:

bool

attach(self, name: 'str') → 'None' -> None[source]

Attach to a running container.

This method replaces the current process with container attach. It does not return on success.

Parameters:

name (str) – Container name to attach to.

Raises:

Note

Use Ctrl+P Ctrl+Q to detach from the container.

status(self, name: 'str') → 'SessionStatus' -> SessionStatus[source]

Get status of a container.

Parameters:

name (str) – Container name to query.

Returns:

SessionStatus with current state.

Raises:

SessionNotFoundError – If container doesn’t exist.

Return type:

SessionStatus

logs(self, name: 'str', lines: 'int' = 100) → 'str' -> str[source]

Retrieve recent logs from a container.

Parameters:

  • name (str) – Container name to get logs from.

  • lines (int) – Number of lines to retrieve.

Returns:

String with log output (ANSI codes preserved).

Raises:

SessionNotFoundError – If container doesn’t exist.

Return type:

str

exists(self, name: 'str') → 'bool' -> bool[source]

Check if a container with the given name exists.

Parameters:

name (str) – Container name to check.

Returns:

True if container exists, False otherwise.

Return type:

bool

list_sessions(self) → 'list[SessionStatus]' -> list[SessionStatus][source]

List all containers.

Returns:

List of SessionStatus for all containers.

Return type:

list[SessionStatus]

exec(self, name: 'str', command: 'str', *, interactive: 'bool' = False) → 'subprocess.CompletedProcess[str]' -> subprocess.CompletedProcess[str][source]

Execute a command in a running container.

Parameters:

  • name (str) – Container name.

  • command (str) – Command to execute.

  • interactive (bool) – If True, use -it flags.

Returns:

CompletedProcess with stdout/stderr.

Raises:

SessionNotFoundError – If container doesn’t exist.

Return type:

CompletedProcess[str]

exception kstlib.ops.ContainerRuntimeNotFoundError[source]

Bases: BackendNotFoundError

Container runtime (podman or docker) not found in PATH.

Install podman or docker to use the container backend: - Podman: https://podman.io/getting-started/installation - Docker: https://docs.docker.com/get-docker/

exception kstlib.ops.OpsError[source]

Bases: KstlibError

Base exception for all ops module errors.

All ops-specific exceptions inherit from this class, allowing for easy catching of any ops error.

exception kstlib.ops.SessionAmbiguousError(name, backends)[source]

Bases: SessionError

Session exists in multiple backends.

Raised when auto-detection finds a session in both tmux and container backends, requiring explicit backend specification.

__init__(self, name: 'str', backends: 'list[str]') → 'None' -> None[source]

Initialize SessionAmbiguousError.

Parameters:

  • name (str) – The session name that exists in multiple backends.

  • backends (list[str]) – List of backend names where the session was found.

exception kstlib.ops.SessionAttachError(name, backend, reason)[source]

Bases: SessionError

Failed to attach to session or container.

Raised when the backend command to attach to a session fails. This can happen if the session is not running or if the terminal is not interactive.

__init__(self, name: 'str', backend: 'str', reason: 'str') → 'None' -> None[source]

Initialize SessionAttachError.

Parameters:

  • name (str) – The session name that failed to attach.

  • backend (str) – The backend type (tmux, container).

  • reason (str) – The reason for the failure.

class kstlib.ops.SessionConfig(name, backend=BackendType.TMUX, command=None, working_dir=None, env=<factory>, image=None, volumes=<factory>, ports=<factory>, runtime=None, log_volume=None)[source]

Bases: object

Configuration for creating a session.

This dataclass holds all configuration options for both tmux and container backends. Options that are not applicable to the selected backend are ignored.

name

Unique session name (required).

Type:

str

backend

Backend type (tmux or container).

Type:

kstlib.ops.models.BackendType

command

Command to run in the session (tmux) or container.

Type:

str | None

working_dir

Working directory for the session.

Type:

str | None

env

Environment variables to set.

Type:

dict[str, str]

image

Container image to use (container backend only).

Type:

str | None

volumes

Volume mounts in “host:container” format.

Type:

list[str]

ports

Port mappings in “host:container” format.

Type:

list[str]

runtime

Container runtime to use (“podman” or “docker”).

Type:

str | None

log_volume

Log volume mount for persistence (auto-mounted).

Type:

str | None

Examples

>>> config = SessionConfig(
...     name="astro",
...     backend=BackendType.TMUX,
...     command="python -m astro.bot",
... )

>>> config = SessionConfig(
...     name="astro-prod",
...     backend=BackendType.CONTAINER,
...     image="astro-bot:latest",
...     volumes=["./data:/app/data"],
...     log_volume="./logs:/app/logs",
... )
name: str
backend: BackendType
command: str | None
working_dir: str | None
env: dict[str, str]
image: str | None
volumes: list[str]
ports: list[str]
runtime: str | None
log_volume: str | None
__post_init__(self) → 'None' -> None[source]

Validate configuration values.

Raises:

ValueError – If any configuration value is invalid.

__init__(self, name: 'str', backend: 'BackendType' = <BackendType.TMUX: 'tmux'>, command: 'str | None' = None, working_dir: 'str | None' = None, env: 'dict[str, str]' = <factory>, image: 'str | None' = None, volumes: 'list[str]' = <factory>, ports: 'list[str]' = <factory>, runtime: 'str | None' = None, log_volume: 'str | None' = None) → None -> None
exception kstlib.ops.SessionConfigError[source]

Bases: OpsError

Configuration error for session management.

Raised when session configuration is invalid or missing required fields.

exception kstlib.ops.SessionError[source]

Bases: OpsError

Base exception for session-related errors.

All session operation exceptions inherit from this class.

exception kstlib.ops.SessionExistsError(name, backend)[source]

Bases: SessionError

Session or container with this name already exists.

Raised when attempting to create a session with a name that is already in use by another session or container.

__init__(self, name: 'str', backend: 'str') → 'None' -> None[source]

Initialize SessionExistsError.

Parameters:

  • name (str) – The session name that already exists.

  • backend (str) – The backend type (tmux, container).

class kstlib.ops.SessionManager(name, *, backend=BackendType.TMUX, **kwargs)[source]

Bases: object

Config-driven session manager with backend abstraction.

Provides a unified interface for managing sessions across different backends (tmux, container). The backend can be specified directly or loaded from kstlib.conf.yml configuration.

Parameters:

  • name (str) – Unique session name.

  • backend (str | BackendType) – Backend type (“tmux” or “container”).

  • **kwargs (Any) – Backend-specific options (image, volumes, ports, etc.).

name

The session name.

backend

The backend type being used.

config

The full session configuration.

Examples

>>> # Direct instantiation with tmux
>>> session = SessionManager("dev", backend="tmux")
>>> session.start("python app.py")  
>>> session.attach()  

>>> # Direct instantiation with container
>>> session = SessionManager(
...     "prod",
...     backend="container",
...     image="app:latest",
...     volumes=["./data:/app/data"],
... )
>>> session.start()  

>>> # From config file (recommended)
>>> session = SessionManager.from_config("astro")
__init__(self, name: 'str', *, backend: 'str | BackendType' = <BackendType.TMUX: 'tmux'>, **kwargs: 'Any') → 'None' -> None[source]

Initialize SessionManager.

Parameters:

  • name (str) – Unique session name.

  • backend (str | BackendType) – Backend type (“tmux” or “container”).

  • **kwargs (Any) – Backend-specific options.

Raises:

SessionConfigError – If configuration is invalid.

property name: str

Return the session name.

property backend: BackendType

Return the backend type.

property config: SessionConfig

Return the session configuration.

classmethod from_config(name: 'str') → 'SessionManager' -> SessionManager[source]

Create SessionManager from kstlib configuration.

Loads session configuration from kstlib.conf.yml under the ops.sessions.{name} key.

Parameters:

name (str) – Session name to load from config.

Returns:

SessionManager configured from the config file.

Raises:

SessionConfigError – If session not found in config.

Return type:

SessionManager

Example

Config file (kstlib.conf.yml):

ops:
  sessions:
    astro:
      backend: tmux
      command: "python -m astro.bot"
      working_dir: "/opt/astro"

Usage:

>>> session = SessionManager.from_config("astro")
start(self, command: 'str | None' = None, **kwargs: 'Any') → 'SessionStatus' -> SessionStatus[source]

Start the session.

Parameters:

  • command (str | None) – Command to run (overrides config).

  • **kwargs (Any) – Additional options to override config.

Returns:

SessionStatus with current state.

Raises:
Return type:

SessionStatus

stop(self, *, graceful: 'bool' = True, timeout: 'int' = 10) → 'bool' -> bool[source]

Stop the session.

Parameters:

  • graceful (bool) – If True, attempt graceful shutdown first.

  • timeout (int) – Seconds to wait for graceful shutdown.

Returns:

True if stopped successfully.

Raises:
Return type:

bool

attach(self) → 'None' -> None[source]

Attach to the session.

This method replaces the current process. It does not return on success.

Raises:
status(self) → 'SessionStatus' -> SessionStatus[source]

Get current session status.

Returns:

SessionStatus with current state.

Raises:

SessionNotFoundError – If session does not exist.

Return type:

SessionStatus

logs(self, lines: 'int' = 100) → 'str' -> str[source]

Get recent session logs.

Parameters:

lines (int) – Number of lines to retrieve.

Returns:

Log output as string (ANSI codes preserved).

Raises:

SessionNotFoundError – If session does not exist.

Return type:

str

exists(self) → 'bool' -> bool[source]

Check if the session exists.

Returns:

True if session exists, False otherwise.

Return type:

bool

is_running(self) → 'bool' -> bool[source]

Check if the session is currently running.

Returns:

True if running, False otherwise.

Return type:

bool

exception kstlib.ops.SessionNotFoundError(name, backend)[source]

Bases: SessionError

Session or container not found.

Raised when attempting to access a session that does not exist.

__init__(self, name: 'str', backend: 'str') → 'None' -> None[source]

Initialize SessionNotFoundError.

Parameters:

  • name (str) – The session name that was not found.

  • backend (str) – The backend type (tmux, container).

exception kstlib.ops.SessionStartError(name, backend, reason)[source]

Bases: SessionError

Failed to start session or container.

Raised when the backend command to create a new session fails.

__init__(self, name: 'str', backend: 'str', reason: 'str') → 'None' -> None[source]

Initialize SessionStartError.

Parameters:

  • name (str) – The session name that failed to start.

  • backend (str) – The backend type (tmux, container).

  • reason (str) – The reason for the failure.

class kstlib.ops.SessionState(value)[source]

Bases: str, Enum

State of a session or container.

RUNNING

Session is active and running.

STOPPED

Session was stopped gracefully.

EXITED

Container exited (with exit code).

DEFINED

Session exists in config but has not been started.

UNKNOWN

State cannot be determined.

RUNNING = 'running'
STOPPED = 'stopped'
EXITED = 'exited'
DEFINED = 'defined'
UNKNOWN = 'unknown'
__format__(self, format_spec)

Returns format using actual value type unless __str__ has been overridden.

class kstlib.ops.SessionStatus(name, state, backend, pid=None, created_at=None, window_count=0, socket_name=None, image=None, exit_code=None)[source]

Bases: object

Current status of a session or container.

This dataclass holds the runtime status information for a session, including state, PID, creation time, and backend-specific details.

name

Session name.

Type:

str

state

Current state (running, stopped, exited, unknown).

Type:

kstlib.ops.models.SessionState

backend

Backend type used for this session.

Type:

kstlib.ops.models.BackendType

pid

Process ID (tmux server PID or container main PID).

Type:

int | None

created_at

ISO timestamp when the session was created.

Type:

str | None

window_count

Number of tmux windows (tmux backend only).

Type:

int

socket_name

tmux socket name for custom sockets (tmux backend only).

Type:

str | None

image

Container image name (container backend only).

Type:

str | None

exit_code

Container exit code if exited (container backend only).

Type:

int | None

Examples

>>> status = SessionStatus(
...     name="astro",
...     state=SessionState.RUNNING,
...     backend=BackendType.TMUX,
...     pid=12345,
...     window_count=1,
... )

>>> status = SessionStatus(
...     name="astro-prod",
...     state=SessionState.RUNNING,
...     backend=BackendType.CONTAINER,
...     pid=67890,
...     image="astro-bot:latest",
... )
name: str
state: SessionState
backend: BackendType
pid: int | None
created_at: str | None
window_count: int
socket_name: str | None
image: str | None
exit_code: int | None
__init__(self, name: 'str', state: 'SessionState', backend: 'BackendType', pid: 'int | None' = None, created_at: 'str | None' = None, window_count: 'int' = 0, socket_name: 'str | None' = None, image: 'str | None' = None, exit_code: 'int | None' = None) → None -> None
exception kstlib.ops.SessionStopError(name, backend, reason)[source]

Bases: SessionError

Failed to stop session or container.

Raised when the backend command to stop a session fails.

__init__(self, name: 'str', backend: 'str', reason: 'str') → 'None' -> None[source]

Initialize SessionStopError.

Parameters:

  • name (str) – The session name that failed to stop.

  • backend (str) – The backend type (tmux, container).

  • reason (str) – The reason for the failure.

exception kstlib.ops.TmuxNotFoundError[source]

Bases: BackendNotFoundError

tmux binary not found in PATH.

Install tmux to use the tmux backend: - macOS: brew install tmux - Ubuntu/Debian: apt install tmux - Windows: Use WSL2 with tmux installed

class kstlib.ops.TmuxRunner(binary='tmux', socket_name=None)[source]

Bases: object

tmux session runner for local development.

Manages tmux sessions for running persistent processes with detach/attach capability. Supports custom sockets via the -L flag for multi-instance setups (e.g. multiple bots on a single host).

Parameters:

  • binary (str) – Path or name of the tmux binary.

  • socket_name (str | None) – Custom tmux socket name (-L flag).

binary

The tmux binary path (validated on first use).

Examples

>>> runner = TmuxRunner()  
>>> config = SessionConfig(name="bot", command="python bot.py")
>>> status = runner.start(config)  
>>> runner.attach("bot")
__init__(self, binary: 'str' = 'tmux', socket_name: 'str | None' = None) → 'None' -> None[source]

Initialize TmuxRunner.

Parameters:

  • binary (str) – Path or name of the tmux binary.

  • socket_name (str | None) – Custom tmux socket name (-L flag). If None, uses the default tmux socket.

property binary: str

Return validated tmux binary path.

Raises:

TmuxNotFoundError – If tmux is not installed.

start(self, config: 'SessionConfig') → 'SessionStatus' -> SessionStatus[source]

Create and start a new tmux session.

Parameters:

config (SessionConfig) – Session configuration.

Returns:

SessionStatus with state information.

Raises:
Return type:

SessionStatus

Note

The exists() check is subject to TOCTOU but provides a clear error message. tmux itself will reject duplicate session names.

stop(self, name: 'str', *, graceful: 'bool' = True, timeout: 'int' = 10) → 'bool' -> bool[source]

Stop a tmux session.

Parameters:

  • name (str) – Session name to stop.

  • graceful (bool) – If True, send C-c first, then kill if needed.

  • timeout (int) – Unused for tmux (interface compliance with AbstractRunner).

Returns:

True if stopped, False if not running.

Raises:
Return type:

bool

attach(self, name: 'str') → 'None' -> None[source]

Attach to a tmux session.

This method replaces the current process with tmux attach. It does not return on success.

Parameters:

name (str) – Session name to attach to.

Raises:
status(self, name: 'str') → 'SessionStatus' -> SessionStatus[source]

Get status of a tmux session.

Parameters:

name (str) – Session name to query.

Returns:

SessionStatus with current state.

Raises:

SessionNotFoundError – If session doesn’t exist.

Return type:

SessionStatus

logs(self, name: 'str', lines: 'int' = 100) → 'str' -> str[source]

Capture recent output from a tmux session.

Parameters:

  • name (str) – Session name to get logs from.

  • lines (int) – Number of lines to capture.

Returns:

String with captured output (ANSI codes preserved).

Raises:

SessionNotFoundError – If session doesn’t exist.

Return type:

str

exists(self, name: 'str') → 'bool' -> bool[source]

Check if a tmux session exists.

Parameters:

name (str) – Session name to check.

Returns:

True if session exists, False otherwise.

Return type:

bool

list_sessions(self) → 'list[SessionStatus]' -> list[SessionStatus][source]

List all tmux sessions.

Returns:

List of SessionStatus for all sessions.

Return type:

list[SessionStatus]

send_keys(self, name: 'str', keys: 'str', *, enter: 'bool' = True) → 'None' -> None[source]

Send keys to a tmux session.

Parameters:

  • name (str) – Session name to send keys to.

  • keys (str) – Keys or text to send.

  • enter (bool) – If True, send Enter key after the text.

Raises:

SessionNotFoundError – If session doesn’t exist.

kstlib.ops.auto_detect_backend(name: 'str', *, socket_name: 'str | None' = None) → 'BackendType | None' -> BackendType | None[source]

Auto-detect which backend a session exists in.

Checks both tmux and container backends to find where a session with the given name exists. Skips backends that are not available (binary not found).

Parameters:

  • name (str) – Session name to search for.

  • socket_name (str | None) – Custom tmux socket name to check.

Returns:

BackendType if found in exactly one backend, None if not found.

Raises:

SessionAmbiguousError – If session exists in multiple backends.

Return type:

BackendType | None

Examples

>>> # Session exists in tmux only
>>> backend = auto_detect_backend("mybot")  
>>> backend == BackendType.TMUX  
True

>>> # Session not found
>>> auto_detect_backend("nonexistent") is None  
True

Exceptions

Specialized exceptions raised by the kstlib.ops module.

Exception hierarchy:

KstlibError
    OpsError (base for all ops errors)
        BackendNotFoundError (binary not in PATH)
            TmuxNotFoundError
            ContainerRuntimeNotFoundError
        SessionError (session-related errors)
            SessionExistsError
            SessionNotFoundError
            SessionStartError
            SessionAttachError
            SessionStopError
            SessionAmbiguousError
exception kstlib.ops.exceptions.BackendNotFoundError[source]

Bases: OpsError, FileNotFoundError

Backend binary (tmux, podman, docker) not found in PATH.

Raised when the required backend binary is not installed or not accessible from the current PATH.

exception kstlib.ops.exceptions.ContainerRuntimeNotFoundError[source]

Bases: BackendNotFoundError

Container runtime (podman or docker) not found in PATH.

Install podman or docker to use the container backend: - Podman: https://podman.io/getting-started/installation - Docker: https://docs.docker.com/get-docker/

exception kstlib.ops.exceptions.OpsError[source]

Bases: KstlibError

Base exception for all ops module errors.

All ops-specific exceptions inherit from this class, allowing for easy catching of any ops error.

exception kstlib.ops.exceptions.SessionAmbiguousError(name, backends)[source]

Bases: SessionError

Session exists in multiple backends.

Raised when auto-detection finds a session in both tmux and container backends, requiring explicit backend specification.

__init__(self, name: 'str', backends: 'list[str]') → 'None' -> None[source]

Initialize SessionAmbiguousError.

Parameters:

  • name (str) – The session name that exists in multiple backends.

  • backends (list[str]) – List of backend names where the session was found.

exception kstlib.ops.exceptions.SessionAttachError(name, backend, reason)[source]

Bases: SessionError

Failed to attach to session or container.

Raised when the backend command to attach to a session fails. This can happen if the session is not running or if the terminal is not interactive.

__init__(self, name: 'str', backend: 'str', reason: 'str') → 'None' -> None[source]

Initialize SessionAttachError.

Parameters:

  • name (str) – The session name that failed to attach.

  • backend (str) – The backend type (tmux, container).

  • reason (str) – The reason for the failure.

exception kstlib.ops.exceptions.SessionError[source]

Bases: OpsError

Base exception for session-related errors.

All session operation exceptions inherit from this class.

exception kstlib.ops.exceptions.SessionExistsError(name, backend)[source]

Bases: SessionError

Session or container with this name already exists.

Raised when attempting to create a session with a name that is already in use by another session or container.

__init__(self, name: 'str', backend: 'str') → 'None' -> None[source]

Initialize SessionExistsError.

Parameters:

  • name (str) – The session name that already exists.

  • backend (str) – The backend type (tmux, container).

exception kstlib.ops.exceptions.SessionNotFoundError(name, backend)[source]

Bases: SessionError

Session or container not found.

Raised when attempting to access a session that does not exist.

__init__(self, name: 'str', backend: 'str') → 'None' -> None[source]

Initialize SessionNotFoundError.

Parameters:

  • name (str) – The session name that was not found.

  • backend (str) – The backend type (tmux, container).

exception kstlib.ops.exceptions.SessionStartError(name, backend, reason)[source]

Bases: SessionError

Failed to start session or container.

Raised when the backend command to create a new session fails.

__init__(self, name: 'str', backend: 'str', reason: 'str') → 'None' -> None[source]

Initialize SessionStartError.

Parameters:

  • name (str) – The session name that failed to start.

  • backend (str) – The backend type (tmux, container).

  • reason (str) – The reason for the failure.

exception kstlib.ops.exceptions.SessionStopError(name, backend, reason)[source]

Bases: SessionError

Failed to stop session or container.

Raised when the backend command to stop a session fails.

__init__(self, name: 'str', backend: 'str', reason: 'str') → 'None' -> None[source]

Initialize SessionStopError.

Parameters:

  • name (str) – The session name that failed to stop.

  • backend (str) – The backend type (tmux, container).

  • reason (str) – The reason for the failure.

exception kstlib.ops.exceptions.TmuxNotFoundError[source]

Bases: BackendNotFoundError

tmux binary not found in PATH.

Install tmux to use the tmux backend: - macOS: brew install tmux - Ubuntu/Debian: apt install tmux - Windows: Use WSL2 with tmux installed

Models

Data models for the kstlib.ops module.

This module defines the core data structures used by the ops module:

  • BackendType: Enum for backend selection (tmux, container)

  • SessionState: Enum for session state (running, stopped, exited, unknown)

  • SessionConfig: Configuration for creating a session

  • SessionStatus: Current status of a session

class kstlib.ops.models.BackendType(value)[source]

Bases: str, Enum

Backend type for session management.

TMUX

Use tmux for session management (dev/local).

CONTAINER

Use Podman/Docker for session management (prod).

TMUX = 'tmux'
CONTAINER = 'container'
__format__(self, format_spec)

Returns format using actual value type unless __str__ has been overridden.

class kstlib.ops.models.SessionConfig(name, backend=BackendType.TMUX, command=None, working_dir=None, env=<factory>, image=None, volumes=<factory>, ports=<factory>, runtime=None, log_volume=None)[source]

Bases: object

Configuration for creating a session.

This dataclass holds all configuration options for both tmux and container backends. Options that are not applicable to the selected backend are ignored.

name

Unique session name (required).

Type:

str

backend

Backend type (tmux or container).

Type:

kstlib.ops.models.BackendType

command

Command to run in the session (tmux) or container.

Type:

str | None

working_dir

Working directory for the session.

Type:

str | None

env

Environment variables to set.

Type:

dict[str, str]

image

Container image to use (container backend only).

Type:

str | None

volumes

Volume mounts in “host:container” format.

Type:

list[str]

ports

Port mappings in “host:container” format.

Type:

list[str]

runtime

Container runtime to use (“podman” or “docker”).

Type:

str | None

log_volume

Log volume mount for persistence (auto-mounted).

Type:

str | None

Examples

>>> config = SessionConfig(
...     name="astro",
...     backend=BackendType.TMUX,
...     command="python -m astro.bot",
... )

>>> config = SessionConfig(
...     name="astro-prod",
...     backend=BackendType.CONTAINER,
...     image="astro-bot:latest",
...     volumes=["./data:/app/data"],
...     log_volume="./logs:/app/logs",
... )
name: str
backend: BackendType
command: str | None
working_dir: str | None
env: dict[str, str]
image: str | None
volumes: list[str]
ports: list[str]
runtime: str | None
log_volume: str | None
__post_init__(self) → 'None' -> None[source]

Validate configuration values.

Raises:

ValueError – If any configuration value is invalid.

__init__(self, name: 'str', backend: 'BackendType' = <BackendType.TMUX: 'tmux'>, command: 'str | None' = None, working_dir: 'str | None' = None, env: 'dict[str, str]' = <factory>, image: 'str | None' = None, volumes: 'list[str]' = <factory>, ports: 'list[str]' = <factory>, runtime: 'str | None' = None, log_volume: 'str | None' = None) → None -> None
class kstlib.ops.models.SessionState(value)[source]

Bases: str, Enum

State of a session or container.

RUNNING

Session is active and running.

STOPPED

Session was stopped gracefully.

EXITED

Container exited (with exit code).

DEFINED

Session exists in config but has not been started.

UNKNOWN

State cannot be determined.

RUNNING = 'running'
STOPPED = 'stopped'
EXITED = 'exited'
DEFINED = 'defined'
UNKNOWN = 'unknown'
__format__(self, format_spec)

Returns format using actual value type unless __str__ has been overridden.

class kstlib.ops.models.SessionStatus(name, state, backend, pid=None, created_at=None, window_count=0, socket_name=None, image=None, exit_code=None)[source]

Bases: object

Current status of a session or container.

This dataclass holds the runtime status information for a session, including state, PID, creation time, and backend-specific details.

name

Session name.

Type:

str

state

Current state (running, stopped, exited, unknown).

Type:

kstlib.ops.models.SessionState

backend

Backend type used for this session.

Type:

kstlib.ops.models.BackendType

pid

Process ID (tmux server PID or container main PID).

Type:

int | None

created_at

ISO timestamp when the session was created.

Type:

str | None

window_count

Number of tmux windows (tmux backend only).

Type:

int

socket_name

tmux socket name for custom sockets (tmux backend only).

Type:

str | None

image

Container image name (container backend only).

Type:

str | None

exit_code

Container exit code if exited (container backend only).

Type:

int | None

Examples

>>> status = SessionStatus(
...     name="astro",
...     state=SessionState.RUNNING,
...     backend=BackendType.TMUX,
...     pid=12345,
...     window_count=1,
... )

>>> status = SessionStatus(
...     name="astro-prod",
...     state=SessionState.RUNNING,
...     backend=BackendType.CONTAINER,
...     pid=67890,
...     image="astro-bot:latest",
... )
name: str
state: SessionState
backend: BackendType
pid: int | None
created_at: str | None
window_count: int
socket_name: str | None
image: str | None
exit_code: int | None
__init__(self, name: 'str', state: 'SessionState', backend: 'BackendType', pid: 'int | None' = None, created_at: 'str | None' = None, window_count: 'int' = 0, socket_name: 'str | None' = None, image: 'str | None' = None, exit_code: 'int | None' = None) → None -> None

TmuxRunner

tmux session runner for local development.

This module provides the TmuxRunner class for managing tmux sessions, enabling detach/attach workflows for local development and backtesting.

Features: - Create named sessions with custom commands and working directories - Attach to running sessions (replaces current process) - Capture session logs with ANSI codes preserved - List all sessions with status information

Example

>>> from kstlib.ops import SessionConfig, BackendType
>>> from kstlib.ops.tmux import TmuxRunner
>>> runner = TmuxRunner()  
>>> config = SessionConfig(  
...     name="dev",
...     backend=BackendType.TMUX,
...     command="python app.py",
... )
>>> status = runner.start(config)  
>>> runner.attach("dev")  # Replaces process
class kstlib.ops.tmux.TmuxRunner(binary='tmux', socket_name=None)[source]

Bases: object

tmux session runner for local development.

Manages tmux sessions for running persistent processes with detach/attach capability. Supports custom sockets via the -L flag for multi-instance setups (e.g. multiple bots on a single host).

Parameters:

  • binary (str) – Path or name of the tmux binary.

  • socket_name (str | None) – Custom tmux socket name (-L flag).

binary

The tmux binary path (validated on first use).

Examples

>>> runner = TmuxRunner()  
>>> config = SessionConfig(name="bot", command="python bot.py")
>>> status = runner.start(config)  
>>> runner.attach("bot")
__init__(self, binary: 'str' = 'tmux', socket_name: 'str | None' = None) → 'None' -> None[source]

Initialize TmuxRunner.

Parameters:

  • binary (str) – Path or name of the tmux binary.

  • socket_name (str | None) – Custom tmux socket name (-L flag). If None, uses the default tmux socket.

property binary: str

Return validated tmux binary path.

Raises:

TmuxNotFoundError – If tmux is not installed.

start(self, config: 'SessionConfig') → 'SessionStatus' -> SessionStatus[source]

Create and start a new tmux session.

Parameters:

config (SessionConfig) – Session configuration.

Returns:

SessionStatus with state information.

Raises:
Return type:

SessionStatus

Note

The exists() check is subject to TOCTOU but provides a clear error message. tmux itself will reject duplicate session names.

stop(self, name: 'str', *, graceful: 'bool' = True, timeout: 'int' = 10) → 'bool' -> bool[source]

Stop a tmux session.

Parameters:

  • name (str) – Session name to stop.

  • graceful (bool) – If True, send C-c first, then kill if needed.

  • timeout (int) – Unused for tmux (interface compliance with AbstractRunner).

Returns:

True if stopped, False if not running.

Raises:
Return type:

bool

attach(self, name: 'str') → 'None' -> None[source]

Attach to a tmux session.

This method replaces the current process with tmux attach. It does not return on success.

Parameters:

name (str) – Session name to attach to.

Raises:
status(self, name: 'str') → 'SessionStatus' -> SessionStatus[source]

Get status of a tmux session.

Parameters:

name (str) – Session name to query.

Returns:

SessionStatus with current state.

Raises:

SessionNotFoundError – If session doesn’t exist.

Return type:

SessionStatus

logs(self, name: 'str', lines: 'int' = 100) → 'str' -> str[source]

Capture recent output from a tmux session.

Parameters:

  • name (str) – Session name to get logs from.

  • lines (int) – Number of lines to capture.

Returns:

String with captured output (ANSI codes preserved).

Raises:

SessionNotFoundError – If session doesn’t exist.

Return type:

str

exists(self, name: 'str') → 'bool' -> bool[source]

Check if a tmux session exists.

Parameters:

name (str) – Session name to check.

Returns:

True if session exists, False otherwise.

Return type:

bool

list_sessions(self) → 'list[SessionStatus]' -> list[SessionStatus][source]

List all tmux sessions.

Returns:

List of SessionStatus for all sessions.

Return type:

list[SessionStatus]

send_keys(self, name: 'str', keys: 'str', *, enter: 'bool' = True) → 'None' -> None[source]

Send keys to a tmux session.

Parameters:

  • name (str) – Session name to send keys to.

  • keys (str) – Keys or text to send.

  • enter (bool) – If True, send Enter key after the text.

Raises:

SessionNotFoundError – If session doesn’t exist.

kstlib.ops.tmux.discover_tmux_sockets() → 'list[str]' -> list[str][source]

Discover non-default tmux socket names.

Scans the tmux socket directory (/tmp/tmux-{uid}/) for socket files other than default. Only works on Unix systems.

Returns:

List of custom socket names found.

Return type:

list[str]

ContainerRunner

Container session runner for production environments.

This module provides the ContainerRunner class for managing Podman/Docker containers, enabling persistent processes with pseudo-terminal support.

Features: - Create named containers with custom images and volumes - Attach to running containers (replaces current process) - Retrieve container logs with ANSI codes preserved - Support for both Podman and Docker runtimes - Automatic log volume mounting for post-mortem analysis

Example

>>> from kstlib.ops import SessionConfig, BackendType
>>> from kstlib.ops.container import ContainerRunner
>>> runner = ContainerRunner(runtime="podman")  
>>> config = SessionConfig(  
...     name="bot",
...     backend=BackendType.CONTAINER,
...     image="bot:latest",
...     volumes=["./data:/app/data"],
... )
>>> status = runner.start(config)  
>>> runner.attach("bot")  # Replaces process
class kstlib.ops.container.ContainerRunner(runtime=None)[source]

Bases: object

Container runner for Podman/Docker containers.

Manages containers for running persistent processes with pseudo-terminal support for TUI applications.

Parameters:

runtime (str | None) – Container runtime to use (“podman” or “docker”).

runtime

The container runtime name.

binary

The validated runtime binary path.

Examples

>>> runner = ContainerRunner()  # Uses podman by default  
>>> runner = ContainerRunner(runtime="docker")  
>>> config = SessionConfig(
...     name="app",
...     backend=BackendType.CONTAINER,
...     image="python:3.10-slim",
... )
>>> status = runner.start(config)
__init__(self, runtime: 'str | None' = None) → 'None' -> None[source]

Initialize ContainerRunner.

Parameters:

runtime (str | None) – Container runtime (“podman”, “docker”, or None for auto-detect). Auto-detection tries podman first, then docker.

property runtime: str

Return the configured runtime name.

property binary: str

Return validated container runtime binary path.

Raises:

ContainerRuntimeNotFoundError – If runtime is not installed.

start(self, config: 'SessionConfig') → 'SessionStatus' -> SessionStatus[source]

Create and start a new container.

Parameters:

config (SessionConfig) – Session configuration with image and options.

Returns:

SessionStatus with state information.

Raises:
Return type:

SessionStatus

stop(self, name: 'str', *, graceful: 'bool' = True, timeout: 'int' = 10) → 'bool' -> bool[source]

Stop a running container.

Parameters:

  • name (str) – Container name to stop.

  • graceful (bool) – If True, use stop with timeout. If False, use kill.

  • timeout (int) – Seconds to wait for graceful shutdown.

Returns:

True if stopped, False if not running.

Raises:
Return type:

bool

attach(self, name: 'str') → 'None' -> None[source]

Attach to a running container.

This method replaces the current process with container attach. It does not return on success.

Parameters:

name (str) – Container name to attach to.

Raises:

Note

Use Ctrl+P Ctrl+Q to detach from the container.

status(self, name: 'str') → 'SessionStatus' -> SessionStatus[source]

Get status of a container.

Parameters:

name (str) – Container name to query.

Returns:

SessionStatus with current state.

Raises:

SessionNotFoundError – If container doesn’t exist.

Return type:

SessionStatus

logs(self, name: 'str', lines: 'int' = 100) → 'str' -> str[source]

Retrieve recent logs from a container.

Parameters:

  • name (str) – Container name to get logs from.

  • lines (int) – Number of lines to retrieve.

Returns:

String with log output (ANSI codes preserved).

Raises:

SessionNotFoundError – If container doesn’t exist.

Return type:

str

exists(self, name: 'str') → 'bool' -> bool[source]

Check if a container with the given name exists.

Parameters:

name (str) – Container name to check.

Returns:

True if container exists, False otherwise.

Return type:

bool

list_sessions(self) → 'list[SessionStatus]' -> list[SessionStatus][source]

List all containers.

Returns:

List of SessionStatus for all containers.

Return type:

list[SessionStatus]

exec(self, name: 'str', command: 'str', *, interactive: 'bool' = False) → 'subprocess.CompletedProcess[str]' -> subprocess.CompletedProcess[str][source]

Execute a command in a running container.

Parameters:

  • name (str) – Container name.

  • command (str) – Command to execute.

  • interactive (bool) – If True, use -it flags.

Returns:

CompletedProcess with stdout/stderr.

Raises:

SessionNotFoundError – If container doesn’t exist.

Return type:

CompletedProcess[str]

SessionManager

Session manager facade for unified backend access.

This module provides the SessionManager class, a config-driven facade that abstracts the underlying backend (tmux or container) and provides a unified interface for session management.

Features: - Automatic backend selection based on configuration - Config-driven session creation from kstlib.conf.yml - Unified API for start, stop, attach, status, and logs - Support for both tmux and container backends

Example

>>> from kstlib.ops import SessionManager
>>> # Local dev with tmux
>>> session = SessionManager("dev", backend="tmux")
>>> session.start("python -m app")  
>>> session.attach()  

>>> # From config file
>>> session = SessionManager.from_config("astro")  
>>> session.start()
exception kstlib.ops.manager.SessionConfigError[source]

Bases: OpsError

Configuration error for session management.

Raised when session configuration is invalid or missing required fields.

class kstlib.ops.manager.SessionManager(name, *, backend=BackendType.TMUX, **kwargs)[source]

Bases: object

Config-driven session manager with backend abstraction.

Provides a unified interface for managing sessions across different backends (tmux, container). The backend can be specified directly or loaded from kstlib.conf.yml configuration.

Parameters:

  • name (str) – Unique session name.

  • backend (str | BackendType) – Backend type (“tmux” or “container”).

  • **kwargs (Any) – Backend-specific options (image, volumes, ports, etc.).

name

The session name.

backend

The backend type being used.

config

The full session configuration.

Examples

>>> # Direct instantiation with tmux
>>> session = SessionManager("dev", backend="tmux")
>>> session.start("python app.py")  
>>> session.attach()  

>>> # Direct instantiation with container
>>> session = SessionManager(
...     "prod",
...     backend="container",
...     image="app:latest",
...     volumes=["./data:/app/data"],
... )
>>> session.start()  

>>> # From config file (recommended)
>>> session = SessionManager.from_config("astro")
__init__(self, name: 'str', *, backend: 'str | BackendType' = <BackendType.TMUX: 'tmux'>, **kwargs: 'Any') → 'None' -> None[source]

Initialize SessionManager.

Parameters:

  • name (str) – Unique session name.

  • backend (str | BackendType) – Backend type (“tmux” or “container”).

  • **kwargs (Any) – Backend-specific options.

Raises:

SessionConfigError – If configuration is invalid.

property name: str

Return the session name.

property backend: BackendType

Return the backend type.

property config: SessionConfig

Return the session configuration.

classmethod from_config(name: 'str') → 'SessionManager' -> SessionManager[source]

Create SessionManager from kstlib configuration.

Loads session configuration from kstlib.conf.yml under the ops.sessions.{name} key.

Parameters:

name (str) – Session name to load from config.

Returns:

SessionManager configured from the config file.

Raises:

SessionConfigError – If session not found in config.

Return type:

SessionManager

Example

Config file (kstlib.conf.yml):

ops:
  sessions:
    astro:
      backend: tmux
      command: "python -m astro.bot"
      working_dir: "/opt/astro"

Usage:

>>> session = SessionManager.from_config("astro")
start(self, command: 'str | None' = None, **kwargs: 'Any') → 'SessionStatus' -> SessionStatus[source]

Start the session.

Parameters:

  • command (str | None) – Command to run (overrides config).

  • **kwargs (Any) – Additional options to override config.

Returns:

SessionStatus with current state.

Raises:
Return type:

SessionStatus

stop(self, *, graceful: 'bool' = True, timeout: 'int' = 10) → 'bool' -> bool[source]

Stop the session.

Parameters:

  • graceful (bool) – If True, attempt graceful shutdown first.

  • timeout (int) – Seconds to wait for graceful shutdown.

Returns:

True if stopped successfully.

Raises:
Return type:

bool

attach(self) → 'None' -> None[source]

Attach to the session.

This method replaces the current process. It does not return on success.

Raises:
status(self) → 'SessionStatus' -> SessionStatus[source]

Get current session status.

Returns:

SessionStatus with current state.

Raises:

SessionNotFoundError – If session does not exist.

Return type:

SessionStatus

logs(self, lines: 'int' = 100) → 'str' -> str[source]

Get recent session logs.

Parameters:

lines (int) – Number of lines to retrieve.

Returns:

Log output as string (ANSI codes preserved).

Raises:

SessionNotFoundError – If session does not exist.

Return type:

str

exists(self) → 'bool' -> bool[source]

Check if the session exists.

Returns:

True if session exists, False otherwise.

Return type:

bool

is_running(self) → 'bool' -> bool[source]

Check if the session is currently running.

Returns:

True if running, False otherwise.

Return type:

bool

kstlib.ops.manager.auto_detect_backend(name: 'str', *, socket_name: 'str | None' = None) → 'BackendType | None' -> BackendType | None[source]

Auto-detect which backend a session exists in.

Checks both tmux and container backends to find where a session with the given name exists. Skips backends that are not available (binary not found).

Parameters:

  • name (str) – Session name to search for.

  • socket_name (str | None) – Custom tmux socket name to check.

Returns:

BackendType if found in exactly one backend, None if not found.

Raises:

SessionAmbiguousError – If session exists in multiple backends.

Return type:

BackendType | None

Examples

>>> # Session exists in tmux only
>>> backend = auto_detect_backend("mybot")  
>>> backend == BackendType.TMUX  
True

>>> # Session not found
>>> auto_detect_backend("nonexistent") is None  
True