"""sshj.run — the grid interaction loop (render/keys/insert). B2 step 4c: moved verbatim from sshj_cli.py (single-file era) — no behavior change, EXCEPT one path line: the sshj-relay lookup used os.path.dirname(abspath(__file__)), which is the package dir in the single-file era (src) but one level too deep inside the package now, so it is os.path.dirname(os.path.dirname(...)) — the resolved path is unchanged (src/sshj-relay). Owns the two-stage Tab input insertion (TIOCSTI, then the relay fallback) and the run_inline key loop; pick() is the thin entry the composition root (connect.main) calls. Edges: ui (render/incoming_rows/_erase/scroll_feed_up/FLAG_TOGGLES), term (Term/keys/anchor), state (Conns incoming), fuzzy (sort/visible), identity, cmd (ssh_cmd_str preview), jsonio (last-used record). """ import fcntl import os import shlex import shutil import subprocess import sys import termios import time from .cmd import ssh_cmd_str, ssh_cmd_str_typed from .config import (is_single_char, key_action_map, load_config, resolved_keymap) from .fuzzy import SORT_LABELS, sort_hosts, visible_hosts from .identity import Identity from .lastused import load_last_success, record_last_used from .term import (ANCHOR, CSI, K_BS, K_CTRL, K_DOWN, K_END, K_ENTER, K_HOME, K_LEFT, K_PGDN, K_PGUP, K_RIGHT, K_TAB, K_UP, Term, _GRID_BOTTOM_ROW, capture_anchor, query_cursor) from .ui import (FLAG_TOGGLES, KEYMAP, _erase, incoming_rows, render, scroll_feed_up) def _keymap_keys(action): """The default-key set for one KEYMAP action; a dispatch `key in _keymap_keys("up")` is a table lookup, not a re-spelled literal — the hint and the handler share one source of truth (SG1-6). The CONFIG-merged sets are built per-run in run_inline (see `_build_action_sets`), which is what actually dispatches.""" return frozenset(k for k, _alias, act, _label in KEYMAP if act == action and k is not None) def _build_action_sets(kmap): """key -> action over the config-merged keymap (action -> [keys]). Returns (action_of, single_char_actions): `action_of` maps a key to its action (None if unbound); `single_char_actions` is the set of BOUND single-character keys (the input-mode rule consults it: a bound single-char key types instead of acting while the field has content).""" action_of = {} single = set() for act, keys in kmap.items(): for k in keys: action_of[k] = act if is_single_char(k): single.add(k) return action_of, single def _staged_command(h, status, eff_args, form): """The text a stage key places into the input line (two-stage: first press stages, second commits). `form` == "alias" (right arrow): the RUNNABLE command with the host's NAME — an ssh alias becomes exactly `ssh ` (e.g. `ssh phone`), no `--`. `form` == "address" (tab): the command with the NAME swapped for the ADDRESS (user@host:port) — what the footer's right side shows (`ssh [flags] user@host:port`), typed form (no `--` so it's runnable as placed). For a raw/discovered target both forms agree (there is no alias).""" if form == "alias": return ssh_cmd_str_typed(h, status, eff_args) return ssh_cmd_str_typed(h, status, eff_args, resolved=True) def _insert_into_input(text): """Put `text` into the shell's input FIELD (an editable command line), not onto the console — called AFTER the grid has left the terminal (termios restored). TIOCSTI stuffs the characters into the tty's input queue, so the shell's line editor (readline/zle) picks them up exactly as if typed. Returns True on success; False where the kernel refuses TIOCSTI (WSL2 >= 6.2, tmux) — the caller then tries the relay, and only prints the command as a last resort.""" try: import fcntl fd = sys.stdin.fileno() for b in text.encode(): fcntl.ioctl(fd, termios.TIOCSTI, bytes([b])) return True except Exception: return False def _insert_via_relay(text, grid_row=""): """TIOCSTI fallback for kernels that refuse it (WSL2 >= 6.2): run `sshj-relay`, which spawns a child bash whose readline line is prefilled with `text` under the REAL prompt (starship), editable; Enter runs it, Ctrl-C discards. `grid_row` (the grid's last painted row, 1-based) is passed so a discard erases only the child's prompt+line. Returns True when the relay ran and the line was left as the user left it (committed OR abandoned) so the caller does NOT print; False only when the relay is missing or errors, then the caller prints the command as a last resort.""" import shutil relay = shutil.which("sshj-relay") or os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "sshj-relay") if not os.path.exists(relay): return False env = dict(os.environ) if grid_row: env["SSHJ_RELAY_GRID_ROW"] = str(grid_row) try: rc = subprocess.call(f"{shlex.quote(relay)} {shlex.quote(text)}", shell=True, env=env) except Exception: return False return True # any rc: relay handled the line (committed or discarded) def run_inline(hosts, last_used, checker, discovery, conns, extra_args=(), speed_scan=None): """Main loop on the PRIMARY screen. Returns (host, checker, mode, text): mode 'connect' (host chosen), 'quit' (None), or 'insert' (host chosen, text = the command to leave in the shell's input field — tab with the command in the field; main() stuffs it via TIOCSTI after term.leave(), printing it where the kernel refuses TIOCSTI (WSL2/tmux)).""" term = Term() ident = Identity() query, sel = "", 0 drawn = [None, 0, None, None, None, None] # drawn[4] = detail-band height of the last frame (B10): the row # budget below subtracts it so the band never pushes the grid past the # bottom of the screen. tab_cmd = "" # staged command for the shell input (stage-1 sets it; stage-2 commits) stage_form = "alias" # how tab_cmd was staged: "alias" (right: the NAME command, # e.g. `ssh phone`) or "address" (tab: the user@host:port command) toggle_flags = set() # B11: active ctrl-toggled ssh flags (boolean) # config: sort mode + the keymap (defaults merged with the user's # ~/.config/sshj/config.json). TWO views of the same keymap: # * `kmap_syms` — symbols, for the HINT (keymap_hint renders symbols); # * the REAL keymap (resolved_keymap) — for the DISPATCH, which # compares read_key()'s real keys (arrows are \x1b[A, tab \t, …). # The input-mode rule (bound single-char keys type while the field has # content) is built from the real keymap. _cfg = load_config() kmap_syms = _cfg["keymap"] sort_mode = _cfg["sort"] sort_modes = ("last", "last_success", "alpha") action_of, single_char_actions = _build_action_sets( resolved_keymap(kmap_syms)) last_success = load_last_success() def _restage(): """Re-stage a staged command for the current selection (same form) after a move or a flag toggle — a staged command FOLLOWS the selection.""" nonlocal tab_cmd if tab_cmd and rows: tab_cmd = _staged_command(rows[sel], checker.status.get(rows[sel]["alias"]), eff_args, stage_form) # drawn = [per-row contents (or None), footer_row, # (row_lengths, input_idx, chars-before-cursor), _, _, (top,wdt,hgt)] # index 3 is reserved; index 4 unused; index 5 = the geometry the # on-screen contents were painted at (clean-frame fast path + the # resize re-anchor both compare it). input_offset = 1 # rows from `top` to the [source] input row # (updated every frame; the resize branch re-anchors the grid from it) digits = "" # digit buffer (numpick): 1-based index max_extent = 0 # lowest row ever painted (for full erase) resize_pending = False # set on size change; re-anchor next tick cursor_hidden = False # ?25l is out (re-anchor burst) def all_hosts(): return hosts + discovery.snapshot() try: term.enter() hgt, wdt = term.size() # Anchor the grid at the shell's WRITING LINE (its relative # position on screen — the grid opens exactly where the prompt was, # feed visible above it). Scroll the feed up just enough for the # FULL list (up to 12 rows: topbar + header + k + input + footer), # not just a 1-row minimal block — a prompt near the pane bottom # must still get a readable list, and the feed goes into scrollback # (pushed up, never erased). The list then fills whatever rows are # free, per frame; growth from discovery is handled by the ensure-fit # step in the main loop (which scrolls up again if the list needs # more room). Rotation re-anchoring lives in the resize branch. k_open = max(1, min(len(hosts), 12)) H = 4 + k_open + incoming_rows(conns) arow = ANCHOR["row"] if ANCHOR["row"] is not None else hgt - 1 arow = max(0, min(arow, hgt - 1)) # The cursor (writing line, `arow`) sits one row BELOW the shell's # command line — after Enter, the prompt+`sshj` line is at arow-1 # (1-based arow). Anchor the grid AT the writing line so it opens # immediately below the `sshj` line with NO blank row between; the # fresh-paint wipes from `top` (not top-1) so that `sshj` line above # is KEPT (pushed up into the feed). If the full list doesn't fit # below `top`, everything (incl. that line) scrolls up into # scrollback — the "push it up" behavior. top_wanted = arow + 1 # 1-based writing line avail = hgt - top_wanted + 1 S = max(0, H - max(0, avail)) if S: scroll_feed_up(term, S) top = max(1, top_wanted - S) prev_size = (hgt, wdt) # for resize detection (phone rotate) checker.start() if speed_scan is not None: speed_scan.start() discovery.start() conns.start() ident.load_cache() ident.start() try: while True: hgt, wdt = term.size() # resize/zoom (incl. phone rotate): the terminal REFLOWS the # visible rows, so the grid and the writing line have both # MOVED — the old `top` is wrong and any wipe from it # leaves debris stranded (the mangled feed). Rotation bursts # several intermediate sizes, so re-anchoring is DEFERRED # until the size has been stable for one tick: no drawing # during the burst (less flicker), one DSR + wipe at the # final size. See the `resize_pending` block below. if (hgt, wdt) != prev_size: prev_size = (hgt, wdt) resize_pending = True continue # let the size settle; draw nothing if resize_pending: resize_pending = False # A rotate makes the terminal REFLOW the visible rows: # the feed re-wraps (height change), the viewport # scrolls to keep the cursor visible (excess feed -> # scrollback, never erased), and — the invariant this # fix rests on (verified in tmux) — the CURSOR keeps # its CONTENT position through the reflow. Re-anchor: # new_top = dsr_row - W # where (r,c) is the 1-based DSR position and W = the # physical lines (at the NEW width) the grid lines # above the cursor occupy: # W = sum(ceil(L_i / wdt) for grid lines i above the # [source] input line) # + k0 // wdt (input line, part) # k0 = chars before the cursor in the input line, known # exactly from the last frame (the cursor is parked at # the end of the field, never pushed by the terminal). # The feed cancels out: its reflow displaces the grid # by exactly the amount the viewport scroll compensates # (both already baked into the measured cursor row). A # CSI move before the DSR must NOT happen — it would # defeat it. # The cursor is hidden for the whole burst (DSR + wipe # + repaint) so the block cursor doesn't flash on the # grid rows while rebuilding; it's re-shown at the # input-field park below. term.put(f"{CSI}0m\x1b[?25l") cursor_hidden = True pos = query_cursor(timeout=0.5) meta = drawn[2] if pos and meta: r = pos[0] + 1 # 0->1 based lens, iidx, k0 = meta W = 0 for i, L in enumerate(lens): if i < iidx: W += (L + wdt - 1) // wdt # full wrapped line elif i == iidx: W += k0 // wdt # input line, part else: break new_top = max(1, r - W) # if the MINIMAL block doesn't fit below new_top, # scroll the feed up by the deficit (feed -> # scrollback, never erased) and move the writing # line up with it — same adjustment as at init. H = 5 + incoming_rows(conns) S = max(0, H - (hgt - new_top)) if S: scroll_feed_up(term, S) old_top = max(1, top - S) new_top = max(1, new_top - S) else: old_top = top # Wipe the UNION of the old and new grid extents: # the re-anchor can move `top` DOWN as well as up, # and a downward-only wipe from the new top would # strand the previous frame above it (stacked # copies on repeated rotations). wipe_from = max(1, min(old_top, new_top)) term.put(f"{CSI}{wipe_from};1H{CSI}J") top = new_top else: # no prior frame yet or DSR unavailable: fall back # to the whole-screen wipe so the grid at least # repaints cleanly (the feed may keep reflowed # debris in this case). term.put(f"{CSI}H{CSI}J") drawn[0] = None # force a full repaint max_extent = top - 1 rows = visible_hosts( sort_hosts(all_hosts(), sort_mode, last_used, last_success), query) if not rows: sel = 0 rows = all_hosts() if sel >= len(rows): sel = max(0, len(rows) - 1) checker.reprioritize([h["alias"] for h in rows]) # digit pick: typing 3 (no filter) moves the cursor to the # 3rd visible entry; shown in the [ source ] input line. if digits and not query: sel = max(0, min(len(rows) - 1, int(digits) - 1)) # input-field display: staged command (tab or right) > # digit pick > filter if tab_cmd: display = tab_cmd elif digits and not query: display = digits else: display = query # ensure-fit: if the visible list (up to 12) is LONGER than # the room below the grid top, scroll the feed up by the # shortfall and move the grid up with it (feed -> scrollback, # never erased). The grid top can only move up, never down, # so a filter shrinking the list never scrolls back. Fires # again each time discovery appends a host past the current # room — that's the list-growing-fed-up behavior requested. desired = min(len(rows), 12) free = hgt - top - 4 - (drawn[4] or 0) - incoming_rows(conns) if desired > free and top > 1: n = min(desired - free, top - 1) if n > 0: scroll_feed_up(term, n) top = max(1, top - n) max_extent = top - 1 drawn[0] = None # fresh-paint wipe handles residue # the incoming section; cap the list at 12. kk = max(1, min(len(rows), 12, hgt - top - 4 - (drawn[4] or 0) - incoming_rows(conns))) # The grid owns the rows from one above its top down to the # footer (+reserved note row). render() composes wipe + # repaint into ONE buffer and writes it in a single flush # (the terminal sees erase-then-paint as one unit, never a # blank in between); a CLEAN frame (signature unchanged) # emits zero bytes, which is what kills the flicker — the # old code erased the whole screen every 100ms even when # nothing changed, and the user's terminal rendered each # erase as a visible blank flash. The wipe covers one row # above `top` deliberately: it catches the terminal's own # reflow/erase residue at the top edge. # B11: the effective command = CLI passthrough (extra_args) # + the active ctrl-toggled flags. Toggled first (stable # order), then passthrough; ssh places them all before `--`. # render's hint line keys off eff_args (flag keys when any # flag — toggled OR passthrough — is active). eff_args = tuple(sorted(toggle_flags)) + tuple(extra_args) _, footer_row, input_row, cur_col = render( term, rows, sel, checker, last_used, time.time(), ident, hgt, wdt, top, kk, drawn, conns, display, eff_args, speed_scan, kmap=kmap_syms, sort_label=SORT_LABELS[sort_mode]) max_extent = max(max_extent, footer_row) key = term.read_key(0.1) if key is None: continue # the input field = staged command OR filter OR numpick. # INPUT-MODE RULE (2026-09-21): while the field has ANY # content, a BOUND SINGLE-CHARACTER key (an opt-in vim # letter, or a space the user bound) is treated as INPUT, # not a key press — "inputing space into the empty input # field should disable single character keymap keys" # (space is always input; it just makes the field # non-empty). Multi-key sequences (arrows, tab, esc, # ctrl+letter) always act, in either mode. field_empty = not (tab_cmd or query or digits) handled = False moved = False if key.startswith(K_CTRL) and key in FLAG_TOGGLES: # B11: toggle a boolean ssh flag on/off (always acts). fl = FLAG_TOGGLES[key] toggle_flags.discard(fl) if fl in toggle_flags \ else toggle_flags.add(fl) digits = "" _restage() handled = True else: act = action_of.get(key) # INPUT-MODE GATE (2026-09-21): a BOUND SINGLE-CHAR # key is a keypress only while the field is EMPTY; with # any content it types (the letters go into the filter). # Multi-key sequences (arrows, tab, right, enter, esc, # ctrl+letter) always act — their keys are never the # thing you'd search for. This is what makes "type a # space, then j" search instead of move, and why the # default (no single letters bound) always searches. single_bound = act is not None and is_single_char(key) if single_bound and not field_empty: act = None if act in ("up", "down", "page_up", "page_dn", "home", "end") and rows: half = max(1, (hgt // 2) - 2) if act == "up": sel = max(0, sel - 1) elif act == "down": sel = min(len(rows) - 1, sel + 1) elif act == "page_up": sel = max(0, sel - half) elif act == "page_dn": sel = min(len(rows) - 1, sel + half) elif act == "home": sel = 0 else: sel = len(rows) - 1 digits = "" moved = True handled = True elif act == "quit": # esc: cascade clear staged -> filter -> digits, # then flags, then close. if tab_cmd: tab_cmd = "" elif query: query = "" elif digits: digits = "" elif toggle_flags: toggle_flags.clear() else: _erase(term, top, max_extent, hgt, wdt) term.put(f"{CSI}{top};1H") return None, checker, "quit", "", \ frozenset(toggle_flags) handled = True elif act == "left": # left CANCELS: with an EMPTY field it closes sshj # outright (2026-09-21: "pressing left when the # input field is empty should just close sshj # outright"); with content it first clears the # field, then closes — a cancel, never a connect. if tab_cmd or query or digits: tab_cmd, query, digits = "", "", "" _erase(term, top, max_extent, hgt, wdt) term.put(f"{CSI}{top};1H") return None, checker, "quit", "", \ frozenset(toggle_flags) elif act == "connect" and rows: _erase(term, top, max_extent, hgt, wdt) term.put(f"{CSI}{top};1H") return rows[sel], checker, "connect", "", \ frozenset(toggle_flags) elif act in ("stage_alias", "stage_address") and rows: # two-stage: 1st press stages this selection's # command in the [ source ] line (form per action: # right = the NAME/alias command `ssh `; # tab = the ADDRESS command `ssh user@host:port`); # 2nd press COMMITs it to the shell input (the # relay path; the grid's bottom row is recorded for # the discard-erase). Backspace edits in place; # arrows re-stage; esc clears; the grid stays open. if tab_cmd: _GRID_BOTTOM_ROW[0] = top _erase(term, top, max_extent, hgt, wdt) term.put(f"{CSI}{top};1H") return rows[sel], checker, "insert", tab_cmd, \ frozenset(toggle_flags) stage_form = "alias" if act == "stage_alias" \ else "address" tab_cmd = _staged_command(rows[sel], checker.status.get( rows[sel]["alias"]), eff_args, stage_form) digits, query = "", "" handled = True elif act == "sort": # cycle recent -> last success -> alpha; the input # line shows the active mode, the list re-sorts. sort_mode = sort_modes[(sort_modes.index(sort_mode) + 1) % len(sort_modes)] sel = 0 moved = True handled = True elif key == K_BS: if tab_cmd: tab_cmd = tab_cmd[:-1] elif query: query = query[:-1] elif digits: digits = digits[:-1] handled = True if not handled: if len(key) == 1 and key.isdigit(): # numpick (unless editing a staged command) if tab_cmd: tab_cmd += key else: digits = (digits + key)[-3:] handled = True elif key.isprintable(): # INPUT (also the landing for a bound single-char # key typed with a non-empty field — the input-mode # rule). typing while a command is staged starts a # fresh filter (the staged command clears). if tab_cmd: tab_cmd = "" digits = "" query += key handled = True if moved and sel < len(rows): # a staged command FOLLOWS the selection (re-stage for # the newly selected host, in the SAME form it was # staged); backspace edited it, so only re-stage on an # actual move. _restage() finally: checker.stop() finally: if speed_scan is not None: speed_scan.stop() discovery.stop() conns.stop() term.leave() def pick(hosts, last_used, checker, discovery, conns, extra_args=(), speed_scan=None): """Returns (host, checker, mode, text). mode: 'connect' (enter), 'insert' (tab staged the command, then enter: leave the command in the shell's input field; text = the command) or 'quit' (esc/ctrl-c -> None).""" try: size = os.get_terminal_size() hgt, wdt = size.lines, size.columns except OSError: hgt, wdt = 24, 80 if hgt < 8 or wdt < 50: print(f"sshj: terminal too small ({wdt}x{hgt})", file=sys.stderr) sys.exit(1) capture_anchor() # DSR: find the writing line # run_inline places the widget (4 + k + incoming lines) at the writing # line; if it doesn't fit below that line, it scrolls the feed up by # the deficit so the feed stays visible above the grid. It never # scrolls on later frames. try: return run_inline(hosts, last_used, checker, discovery, conns, extra_args, speed_scan) except KeyboardInterrupt: return None, None, "quit", "", frozenset()