"""sshj.ui — draw the inline grid (segments -> rows -> one frame). B2 step 4b: moved verbatim from sshj_cli.py (single-file era) — no behavior change. Owns the ms_* formatting trio, the column-width constants, the detail band, row_segments, render, incoming_rows and scroll_feed_up. The active theme channels (SEL/C_*/TARGET_ATTR) are module globals this file mutates via apply_theme and reads in render — so the test that asserts a channel swap must read them from THIS module, not from the sshj_cli re-export (which snapshots the values). Edges: cmd (ssh command preview), theme (THEMES + apply_theme data), term (CSI/_identity_display_tier), tailnet (TAILNET_MARK footer), lastused (human_age), jsonio (update_needed). """ import os from .cmd import ssh_cmd_args, ssh_cmd_str from .jsonio import update_needed from .lastused import human_age from .tailnet import TAILNET_MARK from .term import (CSI, K_BS, K_CTRL, K_DOWN, K_END, K_ENTER, K_ESC, K_HOME, K_LEFT, K_PGDN, K_PGUP, K_RIGHT, K_TAB, K_UP, _identity_display_tier) from .theme import THEMES, VER def ms_fmt(ms): """Latency -> '123ms' / '1.2s' (None -> '—').""" if ms is None: return "—" if ms < 1000: return f"{ms:.0f}ms" return f"{ms/1000:.1f}s" # atuin-style speed coloring on the LATENCY column: green fast, red slow, # dim variants in between (muted, not saturated — the target line carries # the up/down health signal). ms None (not probed) -> muted dash. def ms_color(ms): """Latency -> green/amber/red channel (None -> muted dash).""" if ms is None: return C_DIM if ms < 100: return "\x1b[32m" # fast — green if ms < 250: return "\x1b[2;32m" # good — dim green if ms < 500: return "\x1b[2;31m" # slow — dim red return "\x1b[31m" # bad — red def speed_band(bps): """Bandwidth (bytes/s) -> atuin-style compact M/s label. Decimal units (K=1e3) — this is bytes transferred, not bits. None/0 -> '' (the caller omits the leg; a failed transfer shows nothing rather than a lie).""" if not bps: return "" v = float(bps) for div, suf in ((1e9, "G"), (1e6, "M"), (1e3, "K")): if v >= div: x = v / div return f"{x:.1f}{suf}" if x < 10 else f"{x:.0f}{suf}" return f"{v:.0f}K" # ------------------------------ UI ------------------------------ # B15: active theme channels. THEMES (theme.py) holds the palettes; # apply_theme() copies the active one onto these module globals so the # whole draw path reads one place. nord is the default (house style). SEL = "\x1b[1;4;38;2;116;156;213m" # accent (kanagawa blue, bold+underline): selected name ONLY SEL_MARK = "\x1b[1;38;2;116;156;213m" # accent, bold, NO underline: selected row's tailnet mark C_DIM = "\x1b[2m" C_BOLD = "\x1b[1m" C_RESET = "\x1b[0m" C_TEAL = "\x1b[36m" # our-address (identity): atuin guidance teal C_ITALIC = "\x1b[3m" # muted-but-there: stale lazy values C_GRAY = "\x1b[90m" # keymap + source line + footer detail (atuin DarkGray) C_STRIKE = "\x1b[2;9m" # unreachable target: dim + strikethrough C_YELLOW = "\x1b[38;2;116;156;213m" # input line (search term / staged text): the accent # status glyph; the target line carries the health signal, NOT the glyph: # checking -> muted + animated throbber (progress, no health color) # up -> foreground (default text color), filled dot # down -> muted + strikethrough, open dot STATUS_GLYPH = {"up": "\u25cf", "down": "\u25cb"} THROBBER = "\u280b\u2819\u2839\u2838\u283c\u2874\u2866\u2867" # braille spinner TARGET_ATTR = {"up": "", "down": "\x1b[9;90m"} TARGET_DEFAULT = C_GRAY # not yet probed / checking: muted, same as keymap def apply_theme(name): """Copy THEMES[name] onto the module globals every draw reads. Unknown names degrade to nord. Idempotent; safe to call before every run.""" global SEL, SEL_MARK, C_TEAL, C_GRAY, C_STRIKE, C_YELLOW, TARGET_ATTR global TARGET_DEFAULT t = THEMES.get(name) or THEMES["nord"] SEL, SEL_MARK = t["sel"], t.get("sel_mark", t["sel"]) C_TEAL, C_GRAY = t["teal"], t["gray"] C_STRIKE, C_YELLOW = t["strike"], t["yellow"] TARGET_ATTR = {"up": t["target_up"], "down": t["target_down"]} TARGET_DEFAULT = t["target_default"] if t["target_default"] is not None \ else C_GRAY return name # SG1-6 (RQ7 shortlist #6, B4): the grid keymap as ONE table. This is the # single source of truth for every grid binding: the topbar hint is # GENERATED from it (keymap_hint), and run.py's dispatch reads its key # groups — so a new binding lives in exactly one place and can't appear in # the hint but not the handler (or vice versa). `label`/`label_short` are # the human forms (the long hint uses label; the short hint drops # "connect"/"input" for the width-constrained center strip). The flag # toggles are the `action == "flag"` rows; FLAG_TOGGLES is derived from # them so the ctrl-key -> ssh-flag map can't drift from the table either. # B11: the flags are chosen so they (a) are genuinely toggleable (no value # needed) and (b) avoid ctrl-c (SIGINT), ctrl-d (EOF), ctrl-z (suspend) and # the XON/XOFF flow keys. Value flags (-p/-L/-i/-J/-E/-c) are NOT here — # they're the CLI passthrough (`sshj -L 8080:localhost:8080`), which already # renders in the command. KEYMAP = ( (K_UP, None, "up", "move"), (K_DOWN, None, "down", "move"), (K_RIGHT, None, "stage_alias", "insert"), (K_TAB, None, "stage_address", "insert"), (K_LEFT, None, "left", "exit"), (K_PGUP, None, "page_up", "page"), (K_PGDN, None, "page_dn", "page"), (K_HOME, None, "home", "top"), (K_END, None, "end", "bottom"), (K_ENTER, None, "connect", "connect"), (K_ESC, None, "quit", "exit"), (K_CTRL + "R", None, "sort", "sort"), ) # NO single-letter keys are in the default table: every printable character # is free for the search filter (2026-09-21: "currently we can't search for # anything beginning with a vim navigation key"; "make sure we have no # single letters in the default key map"). vim-style single letters (j/k) # are OPT-IN: config.py's "keymap" field merges alternate/extra keys into # these actions ("allow for alternate keys for all input, and allow single # keys like j&k to be set. If someone does that they know and will know # what they did."). run.py applies the input-mode rule: a bound single-char # key that is typed while the input field has any content is treated as # INPUT, not a key press (a leading space therefore disables the bindings). # "type: filter" is the table's FALLBACK binding, not a row: any printable # key not named above (digits included — digits are the numpick branch) # appends to the row filter. It stays `key.isprintable()` in dispatch so # punctuation filtering is byte-identical to the pre-SG1-6 behavior; a NEW # named binding added to this table takes over its key before the fallback # (the dispatch chain checks named sets first), which is exactly the # "hint and handler can't drift" guarantee this table exists for. FLAG_TOGGLES = { K_CTRL + "T": "-t", # force pseudo-terminal K_CTRL + "X": "-X", # X11 forwarding K_CTRL + "Y": "-Y", # trusted X11 forwarding K_CTRL + "N": "-N", # no remote command (forwarding only) K_CTRL + "A": "-A", # allow ssh-agent forwarding K_CTRL + "F": "-f", # go to background after auth K_CTRL + "V": "-v", # verbose } # the two hint widths the topbar center strip folds between. GENERATED from # the (merged) keymap at call time — `keymap_hint(flags_active, kmap)` — so # an opt-in vim binding (j/k in config) shows in the hint without a code # edit (the SG1-6 "hint and handler can't drift" guarantee, now across the # config layer too). Default rendering: arrows. _MOVE_SYM = {"up": "↑", "down": "↓"} def _key_display(k): """Human form of a key for the hint: arrows keep their glyph, ctrl letters get a space, everything else is wrapped in <...>.""" if k in _MOVE_SYM: return _MOVE_SYM[k] if k.startswith("ctrl+"): return "" % k[len("ctrl+")].lower() return "<%s>" % k.lower() def _move_keys_display(kmap): """The move keys for the hint, paired down/up in vim order: opt-in single letters first ('j/k'), then the arrows ('↑/↓'). Default (no letters bound) is just '↑/↓'.""" dk = [k for k in kmap.get("down", []) if k != "down"] uk = [k for k in kmap.get("up", []) if k != "up"] parts = [] if uk and dk and len(uk) == len(dk) \ and all(len(k) == 1 for k in uk + dk): parts.append("/".join(dk + uk)) # vim order: j/k (down/up) if "up" in kmap.get("up", []) and "down" in kmap.get("down", []): parts.append("↑/↓") elif "up" in kmap.get("up", []): parts.append("↑") elif "down" in kmap.get("down", []): parts.append("↓") return "/".join(parts) or "↑/↓" def _default_kmap(): kmap = {} for k, _alias, act, _lbl in KEYMAP: kmap.setdefault(act, []).append(k) return kmap _KEYBINDS_SHORT = None # set below once _move_keys_display exists _FLAG_KEYS_HINT = " : flags" def keymap_hint(flags_active=False, kmap=None): """The topbar keymap hint, generated from the keymap (one source of truth) — `kmap` (action -> [keys], the config-merged map; None = the built-in default) so an opt-in single-letter binding (j/k in config) shows in the hint without a code edit (the SG1-6 "hint and handler can't drift" guarantee, now across the config layer). Returns (long, short). `flags_active` shows the flag keys in place of the plain keymap (the B11 behavior). Each action's PRIMARY key (first in the list) is shown — alternates are for the hands, the hint is for the glance. The sort MODE is NOT here (it lives on the input line, which is always visible); only the sort ACTION is.""" kmap = kmap or _default_kmap() move = _move_keys_display(kmap) def prim(action, default): return _key_display((kmap.get(action) or [default])[0]) right = prim("stage_alias", "right") tab = prim("stage_address", "tab") left = prim("left", "left") sortkey = prim("sort", "ctrl+r") long = (" %s: move, type: filter, : ssh, %s: alias, " "%s: addr, %s: close, %s: sort" % (move, right, tab, left, sortkey)) short = (" %s move · ssh · %s alias · %s addr · %s close" % (move, right, tab, left)) if not flags_active: return long, short letters = "/".join(k[len(K_CTRL):].lower() for k in FLAG_TOGGLES) return long + " : flags" % letters, \ " : flags" % letters # aliases: the DEFAULT-keymap rendering (no config), retained for any # out-of-module reader and the byte-identity test. KEYBINDS, KEYBINDS_SHORT = keymap_hint(False) FLAG_KEYS_HINT = _FLAG_KEYS_HINT def source_label(h): """Where a connection comes from, as a short label for the [ src ] line. Config files → basename; rc/source → ~-basename; history → 'history'.""" if h.get("src") == "hist": return "history" if h.get("src") == "rc": p = h.get("src_path", "") return "~/" + os.path.basename(p) if p and "/" in p else (p or "rc") p = h.get("from", "") if p: return p.replace(os.path.expanduser("~"), "~") return "config" def _flag_breakdown(host, status, extra_args): """The non-default ssh options for a host, as a compact '·'-joined string (the resolved command's flags minus the `ssh` and the `-- target`). Empty when the command carries no options. This is the structured flag view the operator asked to see (B10/B11) — not a repeat of the command string, but its options pulled out.""" args = ssh_cmd_args(host, status, extra_args) try: end = args.index("--") except ValueError: end = len(args) flags = args[1:end] # fold `-opt value` pairs into one token for the compact view out, i = [], 0 while i < len(flags): a = flags[i] if a in ("-o", "-p", "-L", "-R", "-i", "-J", "-c") and i + 1 < len(flags): out.append(a + " " + flags[i + 1]) i += 2 else: out.append(a) i += 1 return " · ".join(out) def detail_band_lines(h, status, extra_args, wdt, speed=None): """atuin-style always-on PREVIEW band (RQ11), placed under the command footer. Content-driven and capped at 3 muted lines; the row already shows host/port/user/age/conns so the band only adds what the row cannot: the non-default flags in play, the tailnet state when this is a tailnet device, and (B14) the measured up/down bandwidth when this host is up. Anti-redundancy: it never re-labels a column. Returns a list of [(text, attr), ...] line-segments (may be empty). `speed` is {"up": bps|None, "down": bps|None} (or None); a down host shows no speed line (nothing meaningful to say about a dead link).""" lines = [] fb = _flag_breakdown(h, status, extra_args) if fb: lines.append([(" flags: ", C_DIM), (fb, C_GRAY)]) if h.get("tailnet"): up = status == "up" state = "up" if up else ("down" if status == "down" else "·") # the mark renders MUTED here too (source tag); the state word # carries the health color (teal when up). lines.append([(" tailnet: ", C_DIM), (TAILNET_MARK + " device ", C_DIM), (state, C_TEAL if up else C_DIM)]) if status == "up" and speed: d = speed_band(speed.get("down")) u = speed_band(speed.get("up")) bits = [] if d: bits.append("↓" + d) if u: bits.append("↑" + u) if bits: lines.append([(" speed: ", C_DIM), (" ".join(bits), C_GRAY)]) # cap 3 lines; clip each to the screen width (muted, one line each) out = [] for segs in lines[:3]: text = "".join(t for t, _ in segs) if len(text) > wdt: segs = [(text[:wdt - 1] + "…", segs[0][1])] out.append(segs) return out def _line_content(segments, width): """The row's visible content (attrs+text, left to right, no positioning/erase sequences) — what the row shows on screen. Used for the per-row dirty comparison: two rows render identically iff these are equal. Returns (content, logical_length).""" parts, x = [], 1 for text, attrs in segments: if not text: continue if x > width: break if len(text) > width - x + 1: text = text[:width - x + 1] # never cross the right edge parts.append((attrs + text + C_RESET) if attrs else text) x += len(text) return "".join(parts), x - 1 def _put_line(term, y1, segments, width, buf=None): """Erase row y1 (1-based) and paint segments [(text, attrs), ...] left-to-right from column 1. The cursor tracks the writes, so each segment's own fixed-width padding IS its column — never re-pad. Returns the number of characters written (the row's logical length). The row is NOT padded out to the screen width: the `CSI 2K` at the start already erases the whole row, and a shorter LOGICAL line is what keeps rotation safe — when the terminal reflows on resize, a row of length L splits into ceil(L/new_width) rows; a full-width padded row always splits, scattering stale text into the feed area. `buf` (optional): when given, the row's escape sequences are APPENDED to it instead of written — render() composes the rows it needs into one buffer and issues a single write.""" content, length = _line_content(segments, width) seq = f"{CSI}{y1};1H{CSI}2K{content}" if buf is None: term.put(seq) else: buf.append(seq) return length def _erase(term, top, footer_row, hgt, wdt): """Blank every row of the widget (incl. the reserved note row).""" for y in range(top, min(footer_row, hgt) + 2): _put_line(term, y, [(" ", None)], wdt) # --------------------------- UI (raw ANSI) ------------------------ def _rjust(text, width): return text.rjust(width) if len(text) <= width else text[-width:] def input_line_segments(wdt, name_col, bracket_col, src, display, sort_label=None): """The `[ source ]` indicator line (atuin's `[ GLOBAL ]`). `[` at the first column, `]` at the right edge of the last-called column, source label centered between them, and the fuzzy search term / digit pick typed at the name column (where host names begin) — that's where the cursor sits and writes. `display` may also hold the whole ssh command (tab: the command staged for the shell's input field). `sort_label` (e.g. "recent" / "last success" / "alpha") shows the active sort mode, muted, right after the `]` — always visible, so the operator can see which order the list is in ( cycles it). It sits at the FAR-RIGHT margin of the line, and only while the field is EMPTY — a staged command / search term (display) owns the line then, and the order is irrelevant mid-edit, so the label yields and the source label keeps its full width. """ pieces = [(1, "[", None), (bracket_col, "]", None)] inner_l, inner_r = 2, bracket_col - 1 # span between [ and ] span = inner_r - inner_l + 1 if src: if len(src) > span: src = src[:max(0, span)] c = inner_l + (span - len(src)) // 2 # centered in the [ ] span if c < inner_l: c = inner_l pieces.append((c, src, None)) if display: pieces.append((name_col, display, C_YELLOW)) elif sort_label: # far-right margin (1 col in from the edge), muted; the command / # search term (display) is at the name column and is shorter than # the gap to the margin, so they never meet. sc = wdt - len(sort_label) if sc < bracket_col + 1: # very narrow: just after the ] sc = bracket_col + 1 pieces.append((sc, sort_label, C_GRAY)) pieces.sort(key=lambda p: p[0]) segs, x = [], 1 for col, text, attr in pieces: if col < x: continue if col > x: segs.append((" " * (col - x), None)) x = col segs.append((text, attr)) x += len(text) if x <= wdt: segs.append((" " * (wdt - x + 1), None)) return segs # Column layout — no column hugs the left. The metric cells (idx/status/ # latency/last-called, each RIGHT-aligned in its cell) are followed by the # name column, which hugs the actual host names (so TARGET scoots up against # it — no fixed dead gap) but is capped at NAME_CAP of the space so one long # name can't crush the target. TARGET then fills whatever width is left, up # to the right screen edge (it also reserves the ⇆N count slot), so the row # still spans the full length and both text columns expand on resize. # W_AGE = 3 digits + 1-char label (s/m/d/m/y) + 1 char right padding. W_GLYPH, W_SPEED, W_AGE = 4, 6, 5 NAME_CAP = 0.5 # name column is at most this fraction of the free space def _underline(sgr): """Append the underline parameter to an existing SGR sequence, keeping its color/params untouched (``'' -> '\\x1b[4m'``). Used for the selected row's ADDRESS text: underline the address, keep its status color as-is (2026-09-21: "underline the selected address too, only the text, keep its color as it is"). The target's right pad and the ⇆N count are separate unstyled segments, so the underline covers the address text only — never the whole field.""" if not sgr: return "\x1b[4m" return (sgr[:-1] + ";4m") if sgr.endswith("m") else sgr def row_segments(h, st, ms, gi, w_idx, is_sel, hw, tgt_w, last_used, now, n_out, tick, frame_cnt_w=0): """One grid row as [(text, attrs), ...]. `tick` drives the checking throbber. The health signal lives in the TARGET line, not the glyph: up -> foreground dot; down -> muted open dot + strikethrough target; check -> muted animated throbber (progress, no health color). Stable-columns contract (SG1-5): the session-count column is FRAME-WIDE (`frame_cnt_w` = the max over all visible rows, computed in `render`), so `⇆1 -> ⇆12` reserves the extra space for EVERY row and the target column's right edge never moves between frames. The dynamic numbers (ms, age) are rjust'd into fixed widths (`W_SPEED`, `W_AGE`); `_rjust` truncates a would-be overflow rather than widen the column. The target text left-anchors its (expanding) column; the session count (⇆N), when present, right-anchors the column's edge.""" if st == "check": glyph, gattr = THROBBER[tick % len(THROBBER)], C_DIM elif st in ("up", "down"): glyph, gattr = STATUS_GLYPH[st], ("" if st == "up" else C_DIM) else: glyph, gattr = " ", None lu = last_used.get(h["alias"]) age = _rjust((human_age(now - lu) if lu else "—"), W_AGE) disp = h.get("disp", h["alias"]) # the tailnet mark owns the name cell's LAST column (when present): # `disp[:hw]` can never truncate it away, even on a narrow terminal. # It is a SEPARATE muted segment (the joiner measures len(text) and # wraps attrs itself — an escape embedded inside the text would break # column tracking). Muted (C_DIM): a source tag, not a health signal. # ALWAYS the single copy: merge_tailnet sets the flag but never bakes # the glyph into disp, so a row can never render it twice. ts = TAILNET_MARK if h.get("tailnet") else "" tgt_attr = TARGET_ATTR.get(st, TARGET_DEFAULT) cnt = "⇆" + str(n_out) if n_out else "" # frame-wide reservation: every row's target ends at the same column, # whether or not THIS row has a session count tmax = max(2, tgt_w - frame_cnt_w) tgt = h["target"] if len(tgt) > tmax: tgt = tgt[:tmax - 1] + "…" # the name cell is THREE segments: leading space + NAME TEXT + right pad. # The accent (underline) covers ONLY the name text — a hyperlink # underlines the word, not the cell (2026-09-20: it used to span the # whole padded cell, which read as an underline on the gaps too). nm = disp[:hw - len(ts)] nm_pad = max(0, hw - len(ts) - len(nm)) segs = [ (f"{gi + 1:>{w_idx}}", C_DIM), (f"{glyph:>{W_GLYPH}}", gattr), (_rjust(ms_fmt(ms), W_SPEED), ms_color(ms)), (age, C_TEAL if lu else C_DIM), (" ", None), (nm, SEL if is_sel else None), (" " * nm_pad, None), # the tailnet mark takes the ACCENT on the selected row (2026-09-21: # "the ts symbol of the selected entry should be the accent color # too, like the name, but not underlined") — SEL_MARK is the accent # without the underline. Other rows keep the muted C_DIM source tag. (ts, (SEL_MARK if is_sel else C_DIM)) if ts else ("", None), (" " * 2, None), (tgt, (_underline(tgt_attr) if is_sel else tgt_attr)), # the selected # row's ADDRESS text is underlined, keeping its status color (the pad # + ⇆N below are separate unstyled segments — text only, not the field) ] if tmax - len(tgt) > 0: segs.append((" " * (tmax - len(tgt)), None)) # unstyled pad if cnt: segs.append((" " * (frame_cnt_w - len(cnt)) + cnt, C_GRAY)) return segs def render(term, rows, sel, checker, last_used, now, ident, hgt, wdt, top, k, drawn, conns, display, extra_args=(), speed_scan=None, kmap=None, sort_label=None): """Paint the widget into rows top.. (1-based) — but ONLY when the frame differs from the last one actually drawn (flicker rule: a full-screen erase + repaint every tick is visible as flicker, so clean frames are not emitted at all; the main loop skips its wipe when the signature is unchanged). Returns (drawn_signature, footer_row, input_row, input_cursor_col); the row below footer is the reserved note row (blanked on exit). `display` is the text shown in the [ source ] input line (search term or digit pick); the signature tracks its length so re-paints stay correct.""" n = len(rows) if n == 0: # No hosts yet (zero ssh config + empty discovery snapshot): one # honest line, no crash — C3's degrade-with-a-message contract # (PV3-3: render used to hit `rows[sel]` with rows=[]). term.put(f"{CSI}{top};1H{CSI}2K") term.put(f"{C_DIM}sshj: no hosts found{C_RESET}") return ("empty", top, top, 1) cnts = conns.counts() w_idx = max(2, len(str(n))) left_w = w_idx + W_GLYPH + W_SPEED + W_AGE + 1 # cols before the name tgt_w = max((len(h["target"]) for h in rows), default=8) any_cnt = any(cnts.get(h["alias"]) for h in rows) cnt_w = (max((len("⇆" + str(cnts.get(h["alias"]))) for h in rows), default=0) + 2) if any_cnt else 0 # name column hugs the actual host names (TARGET then scoots up against # it), capped at NAME_CAP of the free space so one long name can't crush # the target; TARGET fills the remainder up to the right screen edge # (reserving the ⇆N count slot). Clamped so the row never overflows. space = wdt - left_w - 2 names_w = max((len(h.get("disp", h["alias"])) for h in rows), default=0) + 1 hw = max(6, min(names_w, int(space * NAME_CAP))) tgt_w = max(2, space - hw) hdr_prefix = (f"{'':<{left_w}}" f"{'HOST':<{hw}} " f"{'TARGET':<{tgt_w}}").rstrip() up = sum(1 for h in rows if checker.status.get(h["alias"]) == "up") dn = sum(1 for h in rows if checker.status.get(h["alias"]) == "down") sel_status = checker.status.get(rows[sel]["alias"]) if rows else None # note: only the "down" hint — the row throbber carries the checking # state (no "checking N" note) if sel_status == "down": note_txt, note_attr = "down — enter still tries (5s timeout)", "\x1b[31m" else: note_txt, note_attr = "", C_DIM # --- topbar: title (foreground, bold) · keymap (screen-center, gray) -- # · identity (gray) top-right. atuin-style: the title is foreground # colored (bold) and shows the version; red appears ONLY as the # "- UPDATE" flag on its right when an update is available. The keymap # is centered on the whole screen; when it would overlap title/identity # it drops to the short form, then folds (hidden). title = f" sshj v{VER}" tsegs = [(title, C_BOLD)] if update_needed(): tsegs.append((" - UPDATE", C_BOLD + "\x1b[31m")) title_len = sum(len(t) for t, _ in tsegs) # keymap ALWAYS on: centered on the screen; the identity truncates # (full → short → name → …) to make room. Only on screens too narrow # for even title+short-keymap does the keymap fold. (identity is picked # below, once the keymap has set x, so its width is the TRUE room left — # that lets display() drop a whole tier instead of hard-clipping a # partial IP, which read like a broken value.) topbar = tsegs x = title_len kb = None # SG1-6: the hint is GENERATED from the KEYMAP/FLAG_TOGGLES table # (one source of truth) — `keymap_hint` returns the (long, short) pair; # flags active (extra_args non-empty) shows the flag keys in place of # the plain keymap. Fold to the short form when the long won't center. cand_full, cand_short = keymap_hint(bool(extra_args), kmap) for cand in (cand_full, cand_short): kb_x = (wdt - len(cand)) // 2 if kb_x >= title_len + 1: kb = (kb_x, cand) break if kb: kb_x, cand = kb topbar.append((" " * max(0, kb_x - x), None)) topbar.append((cand, C_GRAY)) x = kb_x + len(cand) idtxt = "" id_attr = C_TEAL stale = ident.stale with ident.lock: _lan, _ts, _name = ident.lan, ident.ts, ident.name if _name or _ts: # Pick the longest tier that fits the room left after the keymap; # drop a whole tier (LAN, then ts) rather than clipping a partial IP. idtxt = _identity_display_tier(_lan, _ts, _name, wdt - max(x, 1)) if idtxt: if stale: idtxt = "~" + idtxt id_attr = C_TEAL + C_ITALIC # identity is ALWAYS right-anchored at the screen edge (it follows # the width on expand/contract): drop the LEFT side first. end = wdt tier = idtxt id_start = max(x + 1, end - len(tier) + 1) topbar.append((" " * max(0, id_start - x - 1), None)) topbar.append((tier, id_attr)) x = id_start + len(tier) # --- rows (scroll window keeps the selection visible) -------------- start = max(0, min(sel - (k - 1) // 2, n - k)) any_chk = any(checker.status.get(h["alias"]) == "check" for h in rows) tick = int(now * 6) % len(THROBBER) if any_chk else 0 # 6 fps throbber body = [] for i in range(k): gi = start + i if gi >= n: break h = rows[gi] st = checker.status.get(h["alias"]) body.append(row_segments(h, st, checker.latency.get(h["alias"]), gi, w_idx, gi == sel, hw, tgt_w, last_used, now, cnts.get(h["alias"], 0), tick, cnt_w)) # --- [ source ] input line (between list and status line) ----------- # `[` col 1, `]` at the right edge of the last-called column, source # label centered between, search term / digit / tab-inserted command at # the name column where the cursor sits and writes sh = rows[sel] name_col = (w_idx + W_GLYPH + W_SPEED + W_AGE) + 2 # 1-based name start bracket_col = w_idx + W_GLYPH + W_SPEED + W_AGE # right edge of last-called col # input text starts AT the name column — the same column the host names' # TEXT begins in, so the search term / staged command aligns with the # list's names (2026-09-21: "move the input field to align with the name # column text"). It used to sit at bracket_col+1 (the name cell's 1-char # leading space), one column left of the names. input_text_col = name_col src = source_label(sh) input_line = input_line_segments(wdt, input_text_col, bracket_col, src, display, sort_label) # --- footer: the EXACT ssh commands (gray) · down note (kept) ------- # left = the command with the alias, as it would be typed # right = the same command with the alias swapped for the last outgoing # address (user@host:port), right-aligned at the screen edge — # the anchor of the status line. # note = the "down — enter still tries" hint, between the two (only # when selected is down). # The line NEVER WRAPS: when the resolved command alone is wider than # the screen it is clipped from the left ("…"+tail, still right- # anchored), then the note, then the alias command gives way. # (flag display, e.g. -p/-l/-o/-J, is a future notion: render the # selected command's options as a compact right-side tag column here, # e.g. "⇥ -p 2222 -o C=5", muted, folded when narrow.) cmd_alias = ssh_cmd_str(sh, sel_status, extra_args) cmd_res = ssh_cmd_str(sh, sel_status, extra_args, resolved=True) right = (" → " + cmd_res) if cmd_res != cmd_alias else "" left = " " + cmd_alias mid = (" " + note_txt) if note_txt else "" right_w = len(right) + (1 if right else 0) # text + 1 leading space room = wdt - right_w if right and room < 1: # resolved command alone overflows: clip it, keep its right edge keep = max(2, wdt - 1) right = "…" + right[-(keep - 1):] right_w = len(right) + (1 if right else 0) room = wdt - right_w if room < 0: room = 0 if len(left) + len(mid) > room: mid = "" if len(left) > room: left = "…" + left[-max(0, room - 1):] if room >= 2 else "" footer = [(left, C_GRAY)] if mid: footer.append((mid, note_attr)) used = sum(len(t) for t, _ in footer) if right: gap = wdt - used - len(right) if gap > 0: footer.append((" " * gap, None)) footer.append((right, C_GRAY)) else: footer.append((" " * max(1, wdt - used), None)) # --- detail band (B10/RQ11): atuin-style preview, under the footer. # Content-driven (0-2 muted lines): non-default flags + tailnet state — # only what the row cannot show. Its height is reported via drawn[4] # so the main loop's row budget (kk) stays inside the screen. band = detail_band_lines(sh, sel_status, extra_args, wdt, speed=speed_scan.get(sh["alias"]) if speed_scan else None) lines = [topbar, [(hdr_prefix, C_DIM)]] + body + [input_line, footer] \ + band input_idx = 2 + len(body) # input_line sits just above the footer # --- incoming section (mirrored, strongly muted) -------------------- inc = conns.incoming_list()[:3] if inc: lines.append([(f" ← incoming ({len(conns.incoming_list())})", C_DIM)]) for user, ip, cnt in inc: src = f"{user}@{ip}" if user else ip seg = [(f" {src:<30}", C_DIM), (f"{cnt:>2}", C_DIM), (" " + STATUS_GLYPH["up"], C_DIM)] lines.append(seg) # Per-row CONTENT (what each grid line will show) — the dirty # comparison unit. Comparing content per row (not one big signature) # means an idle frame whose throbber tick advanced by one (or whose # age column aged one second) repaints only that ONE row — no # full-screen erase, no blank flash, and the feed above the grid is # never touched. frame_rows = [] for seg in lines: content, _ = _line_content(seg, wdt) frame_rows.append(content) n_lines = len(frame_rows) # Cursor park: the position where the next frame's input field ends # (unchanged across clean rows). The field starts at input_text_col # (right after the ']'), so the cursor parks there + the text length. cur_col = min(wdt, input_text_col + len(display)) input_row = min(top + input_idx, hgt) # Compare against the last frame we actually drew (drawn[0] holds the # same per-row content list, or None before the first frame / after a # resize re-anchor). if drawn and drawn[0] is not None and drawn[0] == frame_rows \ and drawn[5] == (top, wdt, hgt): # CLEAN frame: on-screen rows already show this content -> emit # NOTHING (zero bytes). This is what kills the flicker: the old # code erased the whole screen + repainted every 100ms, and the # user's terminal rendered each erase as a visible blank flash. return frame_rows, drawn[1], top + input_idx, cur_col # DIRTY frame: repaint ONLY the rows whose content changed (plus the # rows that moved out of the shrunken extent), all in ONE buffer # flushed with a single write, so the terminal composes them as one # unit (no intermediate blank state). A first frame (or resize # re-anchor) repaints everything from `top` and additionally wipes # the row above the grid top (catches the terminal's own # reflow/erase residue at the top edge). fresh = drawn[0] is None or drawn[5] != (top, wdt, hgt) frame = [] if fresh: # Erase from the grid top DOWN. The row above the grid is the # shell's own line (prompt + the executed `sshj` command), which we # KEEP — so the wipe starts AT `top`, not top-1. (WezTerm et al. # treat the cursor as top-left for `J`, so park it at the grid top # first.) The old sequence ended in a stray literal "l" — fixed. frame.append(f"{CSI}{max(1, min(top, hgt))};1H{CSI}J") drawn[5] = (top, wdt, hgt) if not fresh: old = drawn[0] for i in range(max(n_lines, len(old))): new_c = frame_rows[i] if i < n_lines else "" old_c = old[i] if i < len(old) else "" if new_c != old_c: _put_line(term, min(top + i, hgt), lines[i] if i < len(lines) else [("", None)], wdt, buf=frame) else: for i, seg in enumerate(lines): _put_line(term, min(top + i, hgt), seg, wdt, buf=frame) if frame: frame.append(f"\x1b[?25h{CSI}{input_row};{cur_col}H") term.put("".join(frame)) drawn[0] = frame_rows drawn[1] = min(top + len(lines) - 1, hgt) drawn[4] = len(band) # B10: detail-band height (0-2) # drawn[2]: (per-row char counts, input-row index, chars-before-cursor # in the input row) — the resize re-anchor turns the reflowed cursor # position (DSR) back into the grid's new top row from these. lens = [] for seg in lines: _, length = _line_content(seg, wdt) lens.append(length) drawn[2] = (lens, input_idx, cur_col - 1) return frame_rows, drawn[1], input_row, cur_col def incoming_rows(conns): """Number of incoming-connection footer rows to reserve (0..4).""" inc = conns.incoming_list() return (1 + min(3, len(inc))) if inc else 0 def scroll_feed_up(term, n): """Push the screen up by n lines so the top of the (soon drawn) grid lands at the writing line; the feed moves into the scrollback.""" hgt, wdt = term.size() for _ in range(n): term.put(f"{CSI}{hgt};1H\n")