"""sshj.term — raw-ANSI terminal mechanics (leaf). B2 step 2e: moved verbatim from sshj_cli.py (single-file era) — no behavior change. Owns the DSR cursor query, the Term raw-mode wrapper, the ANCHOR viewport state, and the key constants. Reads no theme channels (ms_color lives with the channels it reads). """ import os import re import sys import termios import time # We render with raw ANSI escapes on the PRIMARY screen (like fzf/atuin), # never curses: curses switches to the alternate screen + clears it, which # is what kept hiding the shell feed. We only ever emit line-erase + # column-home sequences, so the lines above the widget (the feed) are # never touched. def _stdout_is_tty(): try: return sys.stdout.isatty() except Exception: return False def _ui_out(): """The UI's output stream: ALWAYS the real terminal. When stdout is captured (the .bashrc wrapper runs sshj inside $(...)), the grid must paint to /dev/tty so the insert command can travel out through the stdout PIPE (the fzf pattern: UI on the terminal, result on stdout). Cached so the DSR query and the paint share one stream. Falls back to stdout where /dev/tty is unavailable.""" global _TTY_OUT if _TTY_OUT is None: try: if sys.stdout.isatty(): _TTY_OUT = sys.stdout else: _TTY_OUT = os.fdopen(os.open("/dev/tty", os.O_WRONLY), "w", buffering=1) except (OSError, Exception): _TTY_OUT = sys.stdout return _TTY_OUT _TTY_OUT = None # The grid's last painted row (1-based), set by render() each frame and # read by main() to tell the relay where to erase a discarded line from. _GRID_BOTTOM_ROW = [None] CSI = "\x1b[" K_UP, K_DOWN, K_PGUP, K_PGDN, K_HOME, K_END, K_ENTER, K_BS, K_ESC, K_TAB = ( "up", "down", "pgup", "pgdn", "home", "end", "enter", "bs", "esc", "tab") K_RIGHT = "right" # right arrow (stages the ALIAS command, two-stage) K_LEFT = "left" # left arrow (closes sshj when the input field is empty) K_CTRL = "ctrl+" # ctrl+letter marker (B11 flag toggles): "ctrl+P" etc. class Term: """Raw-mode TTY handle: cbreak input, real-terminal output, cursor + erase helpers.""" def __init__(self): self.fd = sys.stdin.fileno() self.out = _ui_out() # real terminal, even if stdout is piped self.tty_fd = self.out.fileno() self.old = None self.last_lines = self.last_cols = 0 def enter(self): import tty import select self.old = termios.tcgetattr(self.fd) tty.setcbreak(self.fd) # Drop any input already queued on the tty: when sshj is started by a # readline completion (Tab on "sshj"), that triggering Tab can still # be sitting in the input buffer and would otherwise be read by the # grid as a fresh key (immediately closing it). Same guard for keys # typed just before the command. try: termios.tcflush(self.fd, termios.TCIFLUSH) except Exception: try: while select.select([self.fd], [], [], 0.0)[0]: os.read(self.fd, 256) except Exception: pass self.size() # mouse reporting: SGR extended (wheel + buttons). Wheel up/down is # mapped to selection moves in read_key; other events are swallowed. # (Harmless where the terminal ignores it — e.g. plain WSL.) self.put("\x1b[?1006h") def leave(self): try: self.put("\x1b[?1006l") termios.tcsetattr(self.fd, termios.TCSADRAIN, self.old) except Exception: pass def size(self): import fcntl, struct try: packed = fcntl.ioctl(self.tty_fd, termios.TIOCGWINSZ, b"\0" * 8) lines, cols = struct.unpack("hh", packed[:4]) if lines <= 0 or cols <= 0: raise ValueError except Exception: try: s = os.get_terminal_size() lines, cols = s.lines, s.columns except OSError: lines, cols = 24, 80 self.last_lines, self.last_cols = lines, cols return lines, cols def read_key(self, timeout=0.1): import select if select.select([self.fd], [], [], timeout)[0]: b = os.read(self.fd, 1) else: return None if not b: return None c = b[0] if c != 27: if c in (10, 13): return K_ENTER if c in (127, 8): return K_BS if c == 9: # tab return K_TAB if c == 3: # ctrl-c raise KeyboardInterrupt if 1 <= c < 32: # ctrl+letter (B11 flag toggles): 1=ctrl-a … 26=ctrl-z return K_CTRL + chr(ord("a") + c - 1).upper() if 32 <= c < 127: return chr(c) return None # ESC sequence: read up to a few bytes (arrows/pgup/pgdn/home/end). # The peek timeout is 1ms, not a disambiguation wait: a key # SEQUENCE arrives from the terminal as ONE burst, so by the time # we've read the ESC byte the rest is already in the tty's input # queue — a pending check (nearly-zero timeout) distinguishes # "sequence in progress" from "lone Esc" without the 20ms stall a # select-wait used to add (2026-09-21: Esc felt laggy). 1ms covers # pathological split delivery; it is imperceptible. seq = bytearray([27]) for _ in range(6): if select.select([self.fd], [], [], 0.001)[0]: seq += os.read(self.fd, 1) else: break s = bytes(seq) # mouse events (only arrive while ?1000h/?1006h reporting is on): # SGR: \x1b[<64;col;rowM wheel up / 65 wheel down # legacy:\x1b[?64;col;rowM / X10 \x1b[M... # wheel moves the selection; all other mouse events are swallowed. if len(s) >= 3 and s[1:3] in (b"[<", b"[?"): while not s.endswith((b"M", b"m", b"~")): if select.select([self.fd], [], [], 0.02)[0]: s += os.read(self.fd, 1) else: break m = re.match(rb"\x1b\[= 3: intro, last = s[1], s[-1] if intro in (ord("["), ord("O")): if last == ord("A"): return K_UP if last == ord("B"): return K_DOWN if last == ord("C"): return K_RIGHT if last == ord("D"): return K_LEFT if last == ord("F"): return K_END if last == ord("H"): return K_HOME if intro == ord("["): if last == ord("5") and s == b"\x1b[5~": return K_PGUP if last == ord("6") and s == b"\x1b[6~": return K_PGDN return K_ESC def put(self, s): out = getattr(self, "out", None) if out is None: out = self.out = _ui_out() out.write(s) out.flush() # atuin/ratatui inline-viewport semantics: ratatui's compute_inline_size # starts at backend.get_cursor_position(), which sends a DSR ("\x1b[6n") to # the terminal. We do the same before entering raw mode: the widget then # spawns at the shell's current cursor row (the writing line) and grows # downward. If the terminal doesn't answer, fall back to bottom-anchored. ANCHOR = {"row": None, "col": 0, "captured": False, "top": None, "k": None, "scrolled": 0} def query_cursor(timeout=0.5): """Ask the terminal for the cursor position (DSR). 0-based (row, col) or None. The DSR byte goes to the SAME stream the UI paints to (Term.out) — when stdout is captured by the .bashrc wrapper, that is /dev/tty.""" try: import tty import select fd = sys.stdin.fileno() if not os.isatty(fd): return None old = termios.tcgetattr(fd) try: tty.setcbreak(fd) out = _ui_out() out.write("\x1b[6n") out.flush() data = b"" end = time.time() + timeout while time.time() < end: r, _, _ = select.select([fd], [], [], 0.05) if not r: continue b = os.read(fd, 1) if not b: break data += b if b == b"R" and b"\x1b[" in data: break m = re.search(rb"\[(\d+);(\d+)R", data) if m: return int(m.group(1)) - 1, int(m.group(2)) - 1 finally: termios.tcsetattr(fd, termios.TCSADRAIN, old) except Exception: pass return None def capture_anchor(): """One-shot DSR query of the writing line (or leave it bottom-anchored).""" if not ANCHOR["captured"]: pos = query_cursor() if pos: ANCHOR["row"], ANCHOR["col"] = pos # else: ANCHOR["row"] stays None -> bottom-anchored fallback ANCHOR["captured"] = True def _identity_display_tier(lan, ts, name, avail): """Pick the longest identity tier that fits `avail` cols: `lan · ts name` -> `ts name` -> `name` (ts and name space-joined, matching Identity.display). Drop a WHOLE tier rather than clipping a partial IP (a dangling `…42.66` reads like a broken value). If even the name doesn't fit, keep its tail (the only part a user can act on). Returns "" when nothing meaningful fits.""" tiers = ( " · ".join(p for p in (lan, " ".join(p for p in (ts, name) if p)) if p), " ".join(p for p in (ts, name) if p), name, ) for s in tiers: if s and len(s) <= avail: return s if name: return "…" + name[-max(1, avail - 1):] if avail >= 2 else "" return ""