Class ShellTUI

switchyard/server/shell_tui.py:77–419

Run a child process in a PTY with a live footer drawn below it. The terminal is split into two regions via DECSTBM (scroll-region): * rows 1 … (total - footer_height): the child process's viewport. * rows (total - footer_height + 1) … total: redrawn by a background thread at ``re

class ShellTUI:
    """Run a child process in a PTY with a live footer drawn below it.

    The terminal is split into two regions via DECSTBM (scroll-region):

    * rows 1 … (total - footer_height): the child process's viewport.
    * rows (total - footer_height + 1) … total: redrawn by a background
      thread at ``refresh_hz`` times per second.

    The child inherits a PTY that reports a reduced row count so its own
    TUI doesn't overlap the footer.  SIGWINCH is forwarded to the child
    and triggers an immediate footer repaint.

    Parameters
    ----------
    command:
        Argv for the child process (``[binary, *args]``).
    footer_fn:
        Called with the current terminal column width; returns a list of
        ``(styled_line, visible_width)`` tuples, one per footer row.
    footer_height:
        Callable returning the number of rows reserved at the bottom of the
        terminal. Called on every paint so the height can grow dynamically.
    refresh_hz:
        Footer repaint frequency in Hz.
    env:
        Optional dict of environment variable overrides merged on top of
        ``os.environ`` before ``execvpe``.  ``TERM`` is set to
        ``xterm-256color`` if not already present.
    """

    def __init__(
        self,
        command: list[str],
        footer_fn: Callable[[int], list[tuple[str, int]]],
        footer_height: Callable[[], int] = lambda: 1,
        refresh_hz: float = 2.0,
        env: dict[str, str] | None = None,
    ) -> None:
        self.command = command
        self.footer_fn = footer_fn
        self._footer_height_fn: Callable[[], int] = footer_height
        self.refresh_interval = 1.0 / refresh_hz
        self._env = env
        self._out_lock = threading.Lock()
        self._stop = threading.Event()
        self._master_fd: int | None = None
        # Monotonic time of the most recent child-output write to stdout.
        # Used by ``_maybe_paint_footer`` to debounce paints away from
        # the middle of Ink's multi-write frame bursts. ``-inf`` so the
        # very first paint (after startup, with no traffic yet) is not
        # blocked.
        self._last_activity_ts: float = float("-inf")
        # Monotonic time of the most recent footer paint. Throttled by
        # ``refresh_interval`` so we don't repaint at the polling rate.
        self._last_paint_ts: float = float("-inf")
        # Set by ``_on_winch`` (the SIGWINCH handler) and drained by the main
        # loop, which calls ``_handle_winch`` to perform the deferred resize