#!/usr/bin/env python3 """sshj-relay — put a command into the shell's INPUT LINE as an editable readline line, on kernels without TIOCSTI (WSL2 >= 6.2, Termux). WHY A CHILD PROCESS: atuin/fzf place a command via a readline CALLBACK (a `bind -x` keybinding that sets READLINE_LINE; the shell then redraws its OWN prompt with the command in it — non-editable prefix, dir/statusline intact, no child). Our Ctrl-x H / sshj+Tab widget path IS that mechanism. The TYPED-DIRECT path can't use it: the shell already ran `sshj` as a plain command, so there is no callback to hand the line back. This relay is that fallback: ONE child bash whose readline line is prefilled with the command. The child draws the REAL prompt — `starship prompt --status 0` emits clean raw ANSI (no PS1 non-printing markers, verified via od) — as the `read -e -p` prompt. So the prefix is the real 2-line prompt (non-editable: readline only edits the text after it), with dir/language/statusline modules intact, and the command sits on the `╰─ ` line. Enter runs it; Ctrl-C discards. Outcomes: ENTER -> child evals the line (ssh runs IN the child; output relayed to the real terminal). relay exits with the command's exit code. ^C -> child killed; from the grid's bottom row (SSHJ_RELAY_GRID_ROW) down is erased, so only the child's prompt+line are removed and the grid above stays; the shell redraws its own prompt next. relay exits 130 (nothing ran). The caller (sshj) sets SSHJ_RELAY_GRID_ROW (1-based) to the grid's last painted row so a discard erases exactly the child's lines, nothing above. """ import fcntl import os import pty import select import shlex import shutil import signal import subprocess import struct import sys import termios import time CSI = "\x1b[" # Child shell: /bin/bash BY CONTRACT (RQ9 rec 4) — a known shell running a # known builtin (`read -e`). No auto-detect of the user's shell: that # reintroduces the fish/zsh prompt-shape problem this design avoids. On # hosts without /bin/bash (Termux: bash at $PREFIX/bin/bash) the relay is # UNAVAILABLE — RQ5 D4: loud exit 2, never the silent 130 no-op. # SSHJ_RELAY_SHELL is a test override; the default is the contract. CHILD_SHELL = os.environ.get("SSHJ_RELAY_SHELL", "/bin/bash") def _child_available(): """Pre-fork guard (RQ5 D4): the child shell must exist + be executable, else the relay can't prefill at all — say so loudly and exit 2 (the caller prints the command; nothing is lost silently).""" if not (os.path.exists(CHILD_SHELL) and os.access(CHILD_SHELL, os.X_OK)): sys.stderr.write( "sshj-relay: no %s (relay child shell missing) — two-stage Tab " "is unavailable; the command is printed instead\n" % CHILD_SHELL) sys.exit(2) def _real_prompt(): """The real prompt as raw ANSI for use as a `read -e -p` prompt. starship emits raw ANSI when it believes stdout is a pipe, but in some environments (tmux) it emits the $PS1 form with literal `\\[`/`\\]` non-printing delimiters even when piped. `read -p` does NOT interpret those (only PS1 does), so strip them — leaving the raw ESC color codes that read -p renders fine. The result is the clean prompt string either way. Empty if starship is missing/fails (blank prefix, not a fake one). """ # SSHJ_STARSHIP overrides (test + exotic installs); then the fast default # path; then a PATH lookup (RQ9 rec 2: stop the silent 5s penalty for # users whose starship lives elsewhere, e.g. /usr/local/bin). sp = os.environ.get("SSHJ_STARSHIP", "") if not sp: sp = os.path.expanduser("~/.local/bin/starship") if not os.path.exists(sp): sp = shutil.which("starship") or "" if not sp or not os.path.exists(sp): return "" try: r = subprocess.run([sp, "prompt", "--status", "0"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, timeout=3) except Exception: return "" p = r.stdout # strip bash PS1 non-printing delimiters (literal backslash + [ / ]) p = p.replace("\\[", "").replace("\\]", "") # drop the leading blank line starship emits before the 2-line prompt # (faithful to the real shell's inter-command gap, but read -p would # print it as an extra blank line). Keep the prompt's TRAILING SPACES # verbatim: with `read -e -p`, the prompt string is the NON-EDITABLE # prefix — starship's `╰─ ` (glyph, dash, space) ends in a space, and # that space is the prompt-to-command separator the user's prompt # already provides. Stripping and re-adding it ourselves (the # 1b9be5a `rstrip` + `prefix = ' '` path) was an engineering-around: # it made the separator EDITABLE text and overrode any prompt that # doesn't end in a space. Procedural fix: never touch it. # rstrip only NEWLINES (subprocess captures starship's final \n) — # trailing SPACES are the prompt's and must survive. p = p.lstrip("\r\n").rstrip("\r\n") if p.strip(): return p return "" def main(): # join with a SINGLE space (argv may be split when called from the # shell, e.g. `sshj-relay ssh -- phone`); the ssh command is a plain # one-liner, so a single space is the correct separator. cmd = " ".join(sys.argv[1:]) if not cmd: sys.exit(2) in_fd = sys.stdin.fileno() out_fd = sys.stdout.fileno() grid_row = os.environ.get("SSHJ_RELAY_GRID_ROW", "") # real stdin -> raw so ^C arrives as a byte (0x03), not SIGINT; the # relay acts on it explicitly. SIGINT ignored defensively. in_attrs = None try: in_attrs = termios.tcgetattr(in_fd) attr = termios.tcgetattr(in_fd) attr[0] = 0 attr[1] = 0 attr[2] &= ~(termios.ICANON | termios.ECHO | termios.ECHOE | termios.ECHOK | termios.ECHONL) attr[3] = 0 termios.tcsetattr(in_fd, termios.TCSANOW, attr) except Exception: in_attrs = None try: signal.signal(signal.SIGINT, signal.SIG_IGN) except Exception: pass prompt = _real_prompt() # child: readline line whose PREFIX is the real prompt; ENTER evals the # line, a cancelled read exits 130 WITHOUT eval (short-circuit). code = ('p=%s; read -e -r -p "$p" __l && eval "$__l" || exit 130' % shlex.quote(prompt)) c_master, c_slave = pty.openpty() # child pty = the real terminal's size (prompt wraps like on screen) try: ws = fcntl.ioctl(out_fd, termios.TIOCGWINSZ, b"\0" * 8) fcntl.ioctl(c_master, termios.TIOCSWINSZ, ws) except Exception: pass _child_available() # RQ5 D4: loud exit 2 before any fork/erase state child = os.fork() if child == 0: try: os.setsid() fcntl.ioctl(c_slave, termios.TIOCSCTTY, 0) except Exception: pass for fd in (0, 1, 2): os.dup2(c_slave, fd) os.close(c_master); os.close(c_slave) os.execve(CHILD_SHELL, [os.path.basename(CHILD_SHELL), "-c", code], dict(os.environ)) os.close(c_slave) # ---- prefill AFTER the child's prompt has rendered. The prompt is # two lines (starship top border, then the `╰─` bottom border that # holds the input), so wait until the bottom-border char arrives in # the child's output, forwarding everything to the real terminal as # it draws. When the prompt is KNOWN EMPTY (no starship / it failed) # there is no glyph to wait for — the deadline would be pure dead # wait (RQ9 rec 1: ~5.5s -> ~0.6s); a short settle is enough. time.sleep(0.4) deadline = time.time() + float( os.environ.get("SSHJ_RELAY_PROMPT_WAIT", "5.0")) # RQ9 rec 3 border_seen = not prompt while time.time() < deadline and not border_seen: r, _, _ = select.select([c_master], [], [], 0.05) if c_master in r: d = os.read(c_master, 65536) if not d: break os.write(out_fd, d) if "╰" in d.decode("utf-8", "replace"): border_seen = True try: if os.waitpid(child, os.WNOHANG)[0] == child: break # child gone; nothing to prefill except ChildProcessError: break time.sleep(0.15) # No separator of our own: _real_prompt() preserves the prompt's # trailing space(s) verbatim, and `read -e -p` renders them as # NON-EDITABLE prefix — the user's own prompt-to-command separator, # intact for however their prompt ends (starship's `╰─ ` ends in a # space; a custom PS1 may not). The command abuts the prompt; what # gap appears is the user's, never a reconstruction by us. (2026-09-21 # procedural fix, replaces the 1b9be5a editable `prefix = ' '`.) try: os.write(c_master, cmd.encode()) except OSError: pass # ---- relay loop: user stdin -> child pty ; child pty -> real terminal alive = True grace = 0.0 discarded = False committed_rc = None try: while alive: try: r, _, _ = select.select([in_fd, c_master], [], [], 0.05) except InterruptedError: continue except OSError: break if in_fd in r: try: d = os.read(in_fd, 65536) except OSError: d = b"" if not d: # stdin closed (the real terminal is gone). Backstop # deadline only — do NOT `continue`: it used to skip the # waitpid/grace check below and loop forever (RQ1). grace = time.time() + 1.0 elif b"\x03" in d or b"\x04" in d: # ^C / ^D: discard. SIGKILL the child and STOP NOW. A # SIGKILLed child's pty master stays select-ready # returning EIO, so the loop must not keep polling it — # `break` and let finally reap the child, erase the # line, and exit 130 (RQ1 root cause). try: os.kill(child, signal.SIGKILL) except (ProcessLookupError, OSError): pass discarded = True alive = False break else: try: os.write(c_master, d) except OSError: grace = time.time() + 0.5 if c_master in r: try: d = os.read(c_master, 65536) except OSError: d = b"" if d: os.write(out_fd, d) else: # child closed its pty (clean EOF, or EIO after it # exited/was killed). No `continue`: fall through so the # waitpid below reaps a dead child and the grace deadline # can end the loop. grace = time.time() + 0.2 if grace and time.time() >= grace: alive = False else: try: wp, st = os.waitpid(child, os.WNOHANG) except ChildProcessError: wp = -1 if wp == child: # 3.8 floor: os.waitstatus_to_exitcode is 3.9+ — the # WIFEXITED/WTERMSIG equivalent (same return contract: # the exit code, or -signal when killed) keeps the relay # runnable on the documented 3.8 (PV3). committed_rc = (os.WEXITSTATUS(st) if os.WIFEXITED(st) else (-os.WTERMSIG(st) if os.WIFSIGNALED(st) else 0)) alive = False finally: try: os.close(c_master) except OSError: pass if in_attrs is not None: try: termios.tcsetattr(in_fd, termios.TCSADRAIN, in_attrs) except Exception: pass try: os.waitpid(child, 0) except (ChildProcessError, OSError): pass if discarded: # remove ONLY the child's prompt + command line: from the # grid's bottom row down (the grid sits just above it). The # shell redraws its own prompt on the next command. try: if grid_row: os.write(out_fd, (f"{CSI}{grid_row};1H{CSI}J").encode()) else: os.write(out_fd, b"\x1b[J") except OSError: pass if committed_rc is not None: sys.exit(committed_rc if 0 <= committed_rc < 256 else 0) sys.exit(130) if __name__ == "__main__": main()