-- SPDX-FileCopyrightText: © 2022—2026 Vladimir Zorin -- SPDX-License-Identifier: LicenseRef-OWL-1.0-or-later -- Licensed under OWL v1.0+. See LICENSE. --[==[ Core terminal I/O library for cursor control, colors, ANSI styles, and keyboard input. Supports Kitty keyboard protocol, text sizing protocol, terminal mode switching, and modifier key encoding/decoding. ]==] local core = require("term.core") local buffer = require("string.buffer") local std = require("std") --[[ See https://en.wikipedia.org/wiki/ANSI_escape_code for info on ANSI escape codes, text colors and attributes. ]] -- DEC private mode numbers (set with CSI ? Pn h, reset with CSI ? Pn l). local DEC_MODE_CURSOR = 25 -- DECTCEM: cursor visibility local DEC_MODE_SYNC = 2026 -- Synchronized output (batch writes atomically) local DEC_MODE_BRACKETED_PASTE = 2004 -- Clipboard paste framed with ESC[200~/ESC[201~ local dec_set = function(mode) return "\027[?" .. mode .. "h" end local dec_reset = function(mode) return "\027[?" .. mode .. "l" end -- Synchronized output mode (DEC private mode 2026) batches all writes -- between START and END into a single atomic screen update, preventing tearing. -- https://gist.github.com/christianparpart/d8a62cc1ab659194337d73e399004036 local SYNC_START = dec_set(DEC_MODE_SYNC) local SYNC_END = dec_reset(DEC_MODE_SYNC) -- Cap bracketed-paste size (defense in depth against a TTY that never -- sends the closing ESC[201~). local MAX_PASTE_BYTES = 10 * 1024 * 1024 local ESC_BYTE = 27 -- A bracketed paste arrives as a burst, but its chunks (and the trailing -- ESC[201~) can be split by gaps larger than the raw-mode read timeout -- (VMIN=0/VTIME=1, ~100ms): a slow X clipboard, a large paste, or system load. -- For a blocking stdio source such a gap surfaces as a nil read that is NOT -- end-of-paste, so we retry up to this many consecutive idle reads (~2s at -- VTIME=1) before giving up — long enough to bridge any real gap, bounded so a -- TTY that drops ESC[201~ can't hang the loop forever. local MAX_PASTE_IDLE_READS = 20 ---! Execute a function inside a synchronized output block ---@ sync(`fn`{.func}) --[===[ Wraps `fn` in synchronized output mode escapes — all terminal writes performed by `fn` are batched and flushed as a single atomic screen update when the block ends. If `fn` raises an error, the synchronized block is still properly closed before the error is re-raised. ]===] local sync = function(fn) io.write(SYNC_START) local ok, err = xpcall(fn, debug.traceback) io.write(SYNC_END) io.flush() if not ok then error(err, 2) end end ---! Hide the terminal cursor ---@ hide_cursor() local hide_cursor = function() io.write(dec_reset(DEC_MODE_CURSOR)) io.flush() end ---! Show the terminal cursor ---@ show_cursor() local show_cursor = function() io.write(dec_set(DEC_MODE_CURSOR)) io.flush() end ---! Move cursor to an absolute line and column position ---@ go(`line`{.num}, `col`{.num}) local go = function(l, c) if l and c then io.write("\027[", l, ";", c, "H") io.flush() end end ---! Write a string to stdout and flush ---@ write(`s`{.str}) local write = function(s) io.write(s) io.flush() end ---! Move cursor to a position and write a string ---@ write_at(`line`{.num}, `col`{.num}, `s`{.str}) local write_at = function(l, c, s) go(l, c) io.write(s) io.flush() end local MOVEMENTS = { up = "\027[{count}A", down = "\027[{count}B", right = "\027[{count}C", left = "\027[{count}D", line_down = "\027[{count}E", line_up = "\027[{count}F", column = "\027[{count}G", } ---! Move cursor in a relative direction ---@ move(`direction`{.str}, `count`{.num .opt}) --[===[ Direction can be `"up"`, `"down"`, `"left"`, `"right"`, `"line_down"`, `"line_up"`, or `"column"`. Count defaults to 1. ]===] local move = function(direction, count) if MOVEMENTS[direction] then if count == nil then count = 1 else count = math.floor(tonumber(count) or 0) if count <= 0 then return end end local move = MOVEMENTS[direction]:gsub("{count}", count) io.write(move) io.flush() end end ---! Clear the entire screen ---@ clear() local clear = function() io.write("\027[2J") io.flush() end ---! Clear part or all of the current line ---@ clear_line(`mode`{.num .opt}) --[===[ Mode 0 (default) clears from cursor to end, 1 from cursor to beginning, 2 the entire line. Cursor position does not change. ]===] local clear_line = function(mode) mode = mode or 0 -- 0 == from cursor to the end of the line -- 1 == from cursor to the beginning of the line -- 2 == the whole line -- Cursor position does not change! write("\027[" .. mode .. "K") end local COLORS = { black = 30, red = 31, green = 32, yellow = 33, blue = 34, magenta = 35, cyan = 36, white = 37, reset = 39, } local STYLES = { reset = 0, bold = 1, dim = 2, italic = 3, underlined = 4, slow_blink = 5, rapid_blink = 6, inverted = 7, exp = 8, double_underlined = 21, normal = 22, not_italic = 23, not_underlined = 24, not_blinking = 25, not_inverted = 27, } ---! Build an ANSI style escape sequence from style names ---@ style(`...`{.str}) -> `ansi`{.str} --[===[ Accepts style names such as `"bold"`, `"italic"`, `"underlined"`, `"dim"`, `"inverted"`, `"reset"`. Returns the combined ANSI escape sequence, or an empty string if no valid styles are given. ]===] local style = function(...) local args = { ... } local style_ansi = "\27[" for _, v in ipairs(args) do if STYLES[v] then style_ansi = style_ansi .. STYLES[v] .. ";" end end if style_ansi == "\27[" then style_ansi = "" else style_ansi = style_ansi:sub(1, -2) .. "m" end return style_ansi end ---! Apply ANSI styles to the terminal immediately ---@ set_style(`...`{.str}) local set_style = function(...) local ansi = style(...) if ansi ~= "" then write(ansi) end end ---! Build an ANSI color escape sequence for foreground and background ---@ color(`fg`{.str .opt}, `bg`{.str .opt}) -> `ansi`{.str} --[===[ Color values can be a named color string (e.g. `"red"`), a 256-color index number, or an RGB table `{r, g, b}`. ]===] local color = function(fg, bg) local fg_ansi = "" local bg_ansi = "" if fg then if type(fg) == "string" and COLORS[fg] then fg_ansi = "\027[" .. COLORS[fg] .. "m" elseif type(fg) == "number" then fg_ansi = "\027[38;5;" .. fg .. "m" else fg_ansi = "\027[38;2;" .. fg[1] .. ";" .. fg[2] .. ";" .. fg[3] .. "m" end end if bg then if type(bg) == "string" and COLORS[bg] then bg_ansi = "\027[" .. COLORS[bg] + 10 .. "m" elseif type(bg) == "number" then bg_ansi = "\027[48;5;" .. bg .. "m" else bg_ansi = "\027[48;2;" .. bg[1] .. ";" .. bg[2] .. ";" .. bg[3] .. "m" end end return fg_ansi .. bg_ansi end ---! Apply foreground and background colors immediately ---@ set_color(`fg`{.str .opt}, `bg`{.str .opt}) local set_color = function(fg, bg) local ansi = color(fg, bg) if ansi ~= "" then write(ansi) end end ---! Query the terminal for the current cursor position ---@ cursor_position() -> `line`{.num .or_nil}, `col`{.num .or_nil} --[===[ Sends a CPR request (`ESC [ 6 n`) and reads the response (`ESC [ row ; col R`). Requires a TTY; typically called while the terminal is in raw mode. Returns nil on parse failure or if the terminal does not respond with the expected sequence. ]===] local cursor_position = function() write("\027[6n") local response = "" -- Read ESC local c = io.read(1) if not c or string.byte(c) ~= 27 then return nil end -- Read [ c = io.read(1) if not c or c ~= "[" then return nil end -- Read until 'R' terminator repeat c = io.read(1) if c and c ~= "R" then response = response .. c end until not c or c == "R" if response and c == "R" then local line = tonumber(response:match("^(%d+)")) local column = tonumber(response:match("^%d+;(%d+)")) return line, column end return nil end ---! Set the terminal window title ---@ title(`str`{.str}) local title = function(str) write("\027]0;" .. str .. string.char(7)) end ---! Send a desktop notification via Kitty's notification protocol ---@ kitty_notify(`title`{.str .opt}, `body`{.str .opt}) local kitty_notify = function(title, body) local id = "i=" .. tostring(os.time()) .. ":" local start = "\027]99;" .. id local ending = "\027\\" if title and body then local out = start .. "d=0:p=title;" .. title .. ending out = out .. start .. "d=1:p=body;" .. body .. ending write(out) elseif title or body then local msg = title or body local out = start .. "d=1:p=body;" .. msg .. ending write(out) end end --[[ Kitty Text Sizing Protocol support See: https://sw.kovidgoyal.net/kitty/text-sizing-protocol/ The text sizing protocol allows rendering text in multiple terminal cells by scaling the font size. This is useful for creating visual hierarchy, superscripts, subscripts, and other text effects. ]] -- Preset configurations for common text sizing scenarios local TS_PRESETS = { triple = { s = 3 }, double = { s = 2 }, big = { s = 2, n = 6, d = 9, w = 2, v = 2, h = 0 }, superscript = { n = 1, d = 2, v = 0, w = 1 }, -- Half-size, top-aligned, 2 chars/cell subscript = { n = 1, d = 2, v = 1, w = 1 }, -- Half-size, bottom-aligned, 2 chars/cell half = { n = 1, d = 2, w = 1 }, -- Half-size, 2 chars/cell compact = { n = 1, d = 2, v = 2, w = 1 }, -- Half-size, centered, 2 chars/cell } -- Validate and normalize numeric text sizing fields. -- Returns a table with validated s, w, n, d, v, h fields (nil when invalid). local validate_ts_opts = function(opts) if not opts then return nil end local r = {} if opts.s and opts.s >= 1 and opts.s <= 7 then r.s = math.floor(opts.s) end if opts.w and opts.w >= 0 and opts.w <= 7 then r.w = math.floor(opts.w) end if opts.n and opts.d and opts.n >= 0 and opts.n <= 15 and opts.d >= 0 and opts.d <= 15 then if opts.d > opts.n then r.n = math.floor(opts.n) r.d = math.floor(opts.d) end end if opts.v and opts.v >= 0 and opts.v <= 2 then r.v = math.floor(opts.v) end if opts.h and opts.h >= 0 and opts.h <= 2 then r.h = math.floor(opts.h) end if not next(r) then return nil end return r end -- Build a colon-separated metadata string from validated ts fields. -- If w_override is provided, it is used instead of ts.w. local build_ts_meta = function(ts, w_override) if not ts then return nil end local metadata = {} if ts.s then table.insert(metadata, "s=" .. ts.s) end if ts.n and ts.d then table.insert(metadata, "n=" .. ts.n) table.insert(metadata, "d=" .. ts.d) end if ts.v then table.insert(metadata, "v=" .. ts.v) end if ts.h then table.insert(metadata, "h=" .. ts.h) end local w = w_override or ts.w if w then table.insert(metadata, "w=" .. w) end if #metadata == 0 then return nil end return table.concat(metadata, ":") end -- Generate text sizing escape sequence -- opts table can contain: s (scale 1-7), w (width 0-7), n (numerator 0-15), -- d (denominator 0-15), v (vertical align 0-2), h (horizontal align 0-2) -- -- For fractional scaling with explicit w, text is automatically chunked so that -- each chunk fits within the specified cell width. Per the Kitty text sizing protocol, -- fractional scaling does NOT affect the number of cells - only the rendered font size. -- To fit multiple characters per cell, we emit separate escape sequences for each chunk. ---! Apply Kitty text sizing protocol to text ---@ text_size(`text`{.str}, `opts`{.tbl}) -> `sized`{.str} --[===[ The `opts` table may contain: `s` (scale 1--7), `w` (width 0--7), `n`/`d` (fractional numerator/denominator), `v` (vertical align), `h` (horizontal align). When fractional scaling with explicit width is used, text is automatically chunked. ]===] local text_size = function(text, opts) if not text or text == "" or not opts then return text or "" end local ts = validate_ts_opts(opts) if not ts then return text end local scale = ts.s local width = ts.w local numerator = ts.n local denominator = ts.d -- Determine if we need to chunk text for fractional scaling -- Chunking is needed when: fractional scaling is active AND explicit w is specified -- The formula: chars_per_cell = w * d / n (how many characters fit in w cells) local needs_chunking = numerator and denominator and width and width > 0 and numerator > 0 if needs_chunking then -- Calculate how many display columns fit per chunk -- For n=1, d=2, w=1: target_width = 1 * 2 / 1 = 2 display columns per chunk local target_width = math.floor(width * denominator / numerator) if target_width < 1 then target_width = 1 end local meta_str = build_ts_meta(ts) -- Chunk the text based on display width local result = {} local chunk = "" local chunk_width = 0 local i = 1 local text_len = std.utf.len(text) while i <= text_len do local char = std.utf.sub(text, i, i) local char_width = std.utf.display_len(char) -- Check if adding this character would exceed target width if chunk_width + char_width > target_width and chunk ~= "" then -- Emit current chunk table.insert(result, "\027]66;" .. meta_str .. ";" .. chunk .. "\027\\") chunk = "" chunk_width = 0 end -- Add character to current chunk chunk = chunk .. char chunk_width = chunk_width + char_width -- If chunk is exactly at target width, emit it if chunk_width >= target_width then table.insert(result, "\027]66;" .. meta_str .. ";" .. chunk .. "\027\\") chunk = "" chunk_width = 0 end i = i + 1 end -- Emit any remaining characters in the last chunk if chunk ~= "" then table.insert(result, "\027]66;" .. meta_str .. ";" .. chunk .. "\027\\") end return table.concat(result, "") else -- No chunking needed - emit single escape sequence local meta_str = build_ts_meta(ts) if not meta_str then return text end return "\027]66;" .. meta_str .. ";" .. text .. "\027\\" end end ---! Calculate text-sizing chunk boundaries without generating escape sequences ---@ text_size_chunks(`text`{.str}, `opts`{.tbl}) -> `chunks`{.tbl .or_nil}, `meta_str`{.str .or_nil} --[===[ Helper used by `tss:apply_sized()`. Accepts the same `opts` table as `text_size()`. Three return cases: - Chunking needed (`n`, `d`, and `w` all set): a table of `{start, stop, meta_str}` chunk descriptors (1-based UTF-8 character indices) plus the `meta_str` string. - Single wrap (no fractional chunking): `nil`, `meta_str`. - No text sizing applicable: `nil`, `nil`. ]===] local text_size_chunks = function(text, opts) if not text or text == "" or not opts then return nil, nil end local ts = validate_ts_opts(opts) if not ts then return nil, nil end local meta_str = build_ts_meta(ts) if not meta_str then return nil, nil end local width = ts.w local numerator = ts.n local denominator = ts.d -- Check if chunking is needed local needs_chunking = numerator and denominator and width and width > 0 and numerator > 0 if not needs_chunking then return nil, meta_str end -- Calculate chunk boundaries local target_width = math.floor(width * denominator / numerator) if target_width < 1 then target_width = 1 end local chunks = {} local chunk_start = 1 local chunk_width = 0 local i = 1 local text_len = std.utf.len(text) while i <= text_len do local char = std.utf.sub(text, i, i) local char_width = std.utf.display_len(char) -- Check if adding this character would exceed target width if chunk_width + char_width > target_width and chunk_start <= i - 1 then -- Record current chunk table.insert(chunks, { start = chunk_start, stop = i - 1, meta_str = meta_str, }) chunk_start = i chunk_width = 0 end chunk_width = chunk_width + char_width -- If chunk is exactly at target width, finalize it if chunk_width >= target_width then table.insert(chunks, { start = chunk_start, stop = i, meta_str = meta_str, }) chunk_start = i + 1 chunk_width = 0 end i = i + 1 end -- Handle any remaining characters if chunk_start <= text_len then table.insert(chunks, { start = chunk_start, stop = text_len, meta_str = meta_str, }) end return chunks, meta_str end --[[ We use and support [kitty keyboard protocol](https://sw.kovidgoyal.net/kitty/keyboard-protocol/) first and foremost. Legacy protocol will be supported eventually, though. `kkbp` in function names stands for Kitty KeyBoard Protocol. ]] ---! Check whether the terminal supports the Kitty keyboard protocol ---@ has_kkbp() -> `supported`{.bool} --[===[ Sends two probes simultaneously: a KKBP flags query (`CSI ? u`) and a Primary Device Attributes request (`CSI c`). Whichever response arrives first determines the result: `CSI ? u` → supported; `CSI ? ... c` → not supported. ]===] local has_kkbp = function() write("\027[?u\027[c") local buf = buffer.new() local esc_received = false local bracket_received = false repeat local c = io.read(1) if c then if not esc_received and string.byte(c) == 27 then esc_received = true elseif esc_received and not bracket_received and c == "[" then bracket_received = true elseif bracket_received then buf:put(c) end end until not c or c == "u" or c == "c" local answer = buf:get() -- kkbp response format: ? flags u (where flags can be empty or digits) -- DA response format: ? number ; number ... c if answer:match("^%?%d*u$") then return true end return false end --[[ `has_ts` checks for text sizing protocol support using CPR method. IMPORTANT: This function must be called while the terminal is in raw mode. It reads terminal responses via io.read() which requires raw mode to work correctly. If called in normal (cooked) mode, the function will hang waiting for input that never arrives in the expected format. Detection mechanism from the spec: 1. Send CR + CPR + OSC66(w=2) + CPR + OSC66(s=2) + CPR 2. Wait for three CPR responses 3. Compare cursor positions: - All same: no support - 2nd moved 2 cells: width support - 3rd moved another 2 cells: full support (scale) Returns: false | "width" | true ]] ---! Detect text sizing protocol support via cursor position reports ---@ has_ts() -> `support`{.bool} --[===[ Must be called in raw mode. Returns `true` for full support, `"width"` for width-only support, or `false` for no support. ]===] local ts_cache = nil local ts_combined_cache = nil local ts_width_mode_cache = nil local supports_ts_cache = nil local has_ts = function() if ts_cache ~= nil then return ts_cache end -- Save current cursor position write("\r") -- Carriage return to start of line local l1, c1 = cursor_position() if not l1 or not c1 then ts_cache = false return false end -- Test width support: send w=2 with a space write("\027]66;w=2; \027\\") local l2, c2 = cursor_position() -- Test scale support: send s=2 with a space write("\027]66;s=2; \027\\") local l3, c3 = cursor_position() -- Clean up write("\r\027[K") if not l2 or not c2 or not l3 or not c3 then ts_cache = false return false end -- Analyze cursor movements local width_support = (c2 - c1) == 2 local scale_support = (c3 - c2) == 2 if scale_support then ts_cache = true return true -- Full text sizing support elseif width_support then ts_cache = "width" return "width" -- Only width parameter supported else ts_cache = false return false -- No support end end ---! Check how the terminal handles combined scale and width text sizing ---@ has_ts_combined() -> `combined`{.bool} --[===[ Must be called in raw mode. Sends `OSC 66 ; s=2:w=2 ; ` then reads the resulting cursor position. A 4-cell advance means the terminal uses combined semantics (`s` × `w` cells); a 2-cell advance means width-only semantics. Results are cached after the first call. ]===] local has_ts_combined = function() if ts_combined_cache ~= nil then return ts_combined_cache end write("\r") local l1, c1 = cursor_position() if not l1 or not c1 then ts_combined_cache = false return false end write("\027]66;s=2:w=2; \027\\") local l2, c2 = cursor_position() write("\r\027[K") if not l2 or not c2 then ts_combined_cache = false return false end ts_combined_cache = (c2 - c1) == 4 return ts_combined_cache end ---! Derive and cache text-sizing width mode from detection results ---@ init_ts_width_mode() --[===[ Calls `has_ts()` and, when full support is detected, `has_ts_combined()` (both require raw mode). Caches the results consumed by `ts_width_mode()` and `supports_ts()`. Must be called once during terminal capability detection before any text-sized rendering begins. ]===] local init_ts_width_mode = function() local ts_support = has_ts() supports_ts_cache = (ts_support ~= false) if ts_support == "width" then ts_width_mode_cache = "w_only" elseif ts_support == true and not has_ts_combined() then ts_width_mode_cache = "w_only" else ts_width_mode_cache = "combined" end std.utf.set_ts_width_mode(ts_width_mode_cache) end ---! Return cached text-sizing width mode ---@ ts_width_mode() -> `mode`{.str} --[===[ Returns `"w_only"` when the `w` parameter and fractional `n`/`d` scaling are interpreted independently by the terminal, or `"combined"` when the terminal applies Kitty-native semantics where `w` multiplies the scale factor. Defaults to `"combined"` if `init_ts_width_mode()` has not been called yet. ]===] local ts_width_mode = function() return ts_width_mode_cache or "combined" end ---! Return cached text-sizing support boolean ---@ supports_ts() -> `supported`{.bool} local supports_ts = function() if supports_ts_cache == nil then return true end return supports_ts_cache end ---! Enable the Kitty keyboard protocol with progressive enhancement level 15 ---@ enable_kkbp() --[===[ Pushes a KKBP stack entry and sets progressive enhancement flags to 15 (1 = disambiguate escape codes, 2 = report event types, 4 = report alternate keys, 8 = report all keys as escape codes). Flag 16 (report associated text) is intentionally excluded. The `term.input` module assumes this exact enhancement level. ]===] local enable_kkbp = function() write("\027[>1u") write("\027[=15;1u") end ---! Disable the Kitty keyboard protocol and restore default key handling ---@ disable_kkbp() local disable_kkbp = function() write("\027[ `win_w`{.num .or_nil}, `win_h`{.num}, `cell_w`{.num}, `cell_h`{.num} --[===[ Sends `CSI 14 t` (window pixel size) and `CSI 16 t` (cell pixel size) and reads both xterm-style responses. Returns `win_w`, `win_h`, `cell_w`, `cell_h` (all in pixels). Returns nil if either response does not arrive within `timeout_ms` (default 100 ms) or the terminal does not support pixel dimension queries. ]===] local get_pixel_dimensions = function(timeout_ms) timeout_ms = timeout_ms or 100 -- Query both window size (14t) and cell size (16t) write("\027[14t\027[16t") local buf = buffer.new() local win_w, win_h, cell_w, cell_h local responses_received = 0 -- Count consecutive nil returns from io.read(1). Each nil represents -- ~100 ms of real time (one VTIME cycle in raw mode), so we can -- approximate a wall-clock timeout without os.clock() (which measures -- CPU time and barely advances during I/O blocking). local nil_limit = math.max(1, math.floor(timeout_ms / 100)) local nil_count = 0 while responses_received < 2 do local c = io.read(1) if not c then nil_count = nil_count + 1 if nil_count >= nil_limit then break end else nil_count = 0 buf:put(c) local data = buf:tostring() -- Look for complete responses: ESC [ 4 ; h ; w t or ESC [ 6 ; h ; w t local response_type, height, width = data:match("\027%[([46]);(%d+);(%d+)t") if response_type then if response_type == "4" then win_h = tonumber(height) win_w = tonumber(width) responses_received = responses_received + 1 elseif response_type == "6" then cell_h = tonumber(height) cell_w = tonumber(width) responses_received = responses_received + 1 end -- Resetting the buffer assumes the two responses don't arrive -- interleaved; in practice xterm writes them back-to-back. buf:reset() end end end if win_w and cell_w then return win_w, win_h, cell_w, cell_h end return nil end ---! Restore the terminal to sane (cooked) mode ---@ set_sane_mode() local set_sane_mode = function() core.set_sane_mode() end ---! Switch the terminal to raw mode if not already active ---@ set_raw_mode() local set_raw_mode = function() if not core.raw_mode() then core.set_raw_mode() end end ---! Switch between the main and alternate terminal screen buffers ---@ switch_screen(`scr`{.str .opt}) --[===[ Pass `"main"` (default) to switch to the main screen, or `"alt"` for the alternate screen. ]===] local switch_screen = function(scr) scr = scr or "main" if scr == "main" then io.write("\027[?47l") else io.write("\027[?47h") end io.flush() end ---! Enter the alternate screen with KKBP and bracketed paste enabled ---@ alt_screen(`opts`{.tbl .opt}) -> `state`{.tbl} --[===[ Saves the cursor position, switches to the alt screen, enables KKBP and bracketed paste, and hides the cursor. Returns a state object with a `state:done()` method that restores everything. `opts.query_cursor` (default true): when false, skips the `ESC[6n` cursor-position query. The query reads stdin through buffered stdio, so callers that then read input via a raw `lev.fd_reader` must pass `false` or those reads will miss buffer-swallowed keystrokes. ]===] local alt_screen = function(opts) opts = opts or {} set_raw_mode() -- The CPR query (`cursor_position`) reads stdin through buffered stdio. -- Callers that subsequently read input via a raw `lev.fd_reader` (e.g. -- `term.gfx.loop`) must skip it: stdio would buffer-swallow keystrokes -- the raw read(2) never sees. Pass `query_cursor = false` to opt out. local l, c if opts.query_cursor ~= false then l, c = cursor_position() end switch_screen("alt") enable_kkbp() enable_bracketed_paste() hide_cursor() return { l = l, c = c, done = function(self) disable_bracketed_paste() disable_kkbp() switch_screen("main") if self.l and self.c then go(self.l, self.c) end show_cursor() set_sane_mode() end, } end --[[ Our terminal input relies on the [Kitty's keyboard protocol](https://sw.kovidgoyal.net/kitty/keyboard-protocol/), `kkbp` prefix in variable/function names stands for `Kitty KeyBoard Protocol`. Initially this module was built around the legacy input protocol, but it is really not good enough for any serious terminal application, and it seemed too much of a hassle to maintain both implementations. Yet it'd be nice to have a minimal implementation of the legacy one as a fallback, so there is a chance it will be added in the future... ]] -- local KKBP_CODES = { ["57441"] = "LEFT_SHIFT", ["57442"] = "LEFT_CTRL", ["57443"] = "LEFT_ALT", ["57444"] = "LEFT_SUPER", ["57447"] = "RIGHT_SHIFT", ["57448"] = "RIGHT_CTRL", ["57449"] = "RIGHT_ALT", ["57450"] = "RIGHT_SUPER", ["57358"] = "CAPS_LOCK", ["57359"] = "SCROLL_LOCK", ["57360"] = "NUM_LOCK", ["57361"] = "PRINT_SCREEN", ["57362"] = "PAUSE", ["57363"] = "MENU", ["57428"] = "MEDIA_PLAY", ["57429"] = "MEDIA_PAUSE", ["57430"] = "MEDIA_PLAY_PAUSE", ["57431"] = "MEDIA_REVERSE", ["57432"] = "MEDIA_STOP", ["57433"] = "MEDIA_FAST_FORWARD", ["57434"] = "MEDIA_REWIND", ["57435"] = "MEDIA_TRACK_NEXT", ["57436"] = "MEDIA_TRACK_PREVIOUS", ["57437"] = "MEDIA_RECORD", ["57438"] = "VOLUME_DOWN", ["57439"] = "VOLUME_UP", ["57440"] = "VOLUME_MUTE", ["57451"] = "RIGHT_HYPER", ["57399"] = "KP_0", ["57400"] = "KP_1", ["57401"] = "KP_2", ["57402"] = "KP_3", ["57403"] = "KP_4", ["57404"] = "KP_5", ["57405"] = "KP_6", ["57406"] = "KP_7", ["57407"] = "KP_8", ["57408"] = "KP_9", ["57409"] = "KP_DECIMAL", ["57410"] = "KP_DIVIDE", ["57411"] = "KP_MULTIPLY", ["57412"] = "KP_SUBTRACT", ["57413"] = "KP_ADD", ["57414"] = "KP_ENTER", ["57415"] = "KP_EQUAL", ["57416"] = "KP_SEPARATOR", ["57417"] = "KP_LEFT", ["57418"] = "KP_RIGHT", ["57419"] = "KP_UP", ["57420"] = "KP_DOWN", ["57421"] = "KP_PAGE_UP", ["57422"] = "KP_PAGE_DOWN", ["57423"] = "KP_HOME", ["57424"] = "KP_END", ["57425"] = "KP_INSERT", ["57426"] = "KP_DELETE", ["D"] = "LEFT", ["C"] = "RIGHT", ["A"] = "UP", ["B"] = "DOWN", ["H"] = "HOME", ["F"] = "END", ["P"] = "F1", ["Q"] = "F2", ["R"] = "F3", ["S"] = "F4", ["57365"] = "F5", ["57366"] = "F6", ["57367"] = "F7", ["57368"] = "F8", ["57369"] = "F9", ["57370"] = "F10", ["57371"] = "F11", ["57372"] = "F12", ["E"] = "KP_BEGIN", ["9"] = "TAB", ["13"] = "ENTER", ["127"] = "BACKSPACE", ["27"] = "ESC", } -- VT220-style tilde-terminated sequences use numeric codes that overlap -- with ASCII control codes in KKBP_CODES (e.g. "13~" is F3 but "13" alone -- is ENTER). When a CSI sequence ends in `~`, resolve its codepoint via -- this table first so F3/F5+ and nav keys aren't mis-parsed. local KKBP_CODES_LEGACY = { ["2"] = "INSERT", ["3"] = "DELETE", ["5"] = "PAGE_UP", ["6"] = "PAGE_DOWN", ["7"] = "HOME", ["8"] = "END", ["11"] = "F1", ["12"] = "F2", ["13"] = "F3", ["14"] = "F4", ["15"] = "F5", ["17"] = "F6", ["18"] = "F7", ["19"] = "F8", ["20"] = "F9", ["21"] = "F10", ["23"] = "F11", ["24"] = "F12", } -- Bare modifier / lock / meta keys reported by KKBP flag 8 that should -- never produce application-level events. Values match KKBP_CODES above. local MODIFIER_ONLY_KEYS = { LEFT_SHIFT = true, RIGHT_SHIFT = true, LEFT_CTRL = true, RIGHT_CTRL = true, LEFT_ALT = true, RIGHT_ALT = true, LEFT_SUPER = true, RIGHT_SUPER = true, RIGHT_HYPER = true, CAPS_LOCK = true, SCROLL_LOCK = true, NUM_LOCK = true, PRINT_SCREEN = true, PAUSE = true, MENU = true, } -- In KKBP, the terminal reports modifiers as bitmask + 1. -- mods <= 2 means "no modifier" (1) or "SHIFT only" (2) -- treat key as printable. local MAX_PRINTABLE_MODS = 2 -- Unicode BMP Private Use Area: KKBP uses this range to encode non-printable keys. local PUA_START = 0xE000 local PUA_END = 0xF8FF -- US QWERTY Shift map: unshifted key → shifted key for non-letter keys. -- Used to reconstruct the US layout character when Shift is held with -- a non-Latin keyboard layout (where `base` only reports the unshifted key). local US_SHIFT_MAP = { ["`"] = "~", ["1"] = "!", ["2"] = "@", ["3"] = "#", ["4"] = "$", ["5"] = "%", ["6"] = "^", ["7"] = "&", ["8"] = "*", ["9"] = "(", ["0"] = ")", ["-"] = "_", ["="] = "+", ["["] = "{", ["]"] = "}", ["\\"] = "|", [";"] = ":", ["'"] = '"', [","] = "<", ["."] = ">", ["/"] = "?", } --[[ We are assuming that every key press is reported as an escape sequence (PE enum set to 15), which simplifies parsing a lot. ]] ---! Read a keyboard event via the Kitty keyboard protocol ---@ get() -> `key`{.str .or_nil}, `mods`{.num}, `event`{.num}, `shifted`{.str .or_nil}, `base`{.str .or_nil}, `is_paste`{.bool} --[===[ Returns the key name or character, modifier bitmask, event type (1=press, 2=repeat, 3=release), shifted key, and base key. Handles bracketed paste: the pasted text is returned as `key` with a trailing `is_paste` flag set true (and the other fields nil), so callers can insert it as literal text instead of dispatching it as a keypress. ]===] -- Byte-source abstraction: the KKBP/escape parser pulls input bytes -- through read1() instead of calling io.read directly. The default stdio -- source preserves the historical blocking io.read behavior, so get() is -- unchanged for all existing callers. A lev source (see read_key) reads -- cooperatively on the event loop without blocking other tasks. Either way -- the parser reassembles multi-byte sequences itself; a source only needs to -- yield single bytes and set transient_eof when a nil read means "not yet" -- rather than "closed". local stdio_source = { -- nil from io.read here is a raw-mode read timeout ("no data right now"), -- not a closed stream; the paste loop retries such reads (see -- transient_eof) instead of treating them as end-of-paste. transient_eof = true, read1 = function() return io.read(1) end, } -- Consume a string-terminated control sequence (APC/DCS/OSC/PM/SOS) up to its -- ST (ESC \) or, for OSC, a BEL. Returns false if the source ran dry mid-way. local skip_string_sequence = function(source) while true do local c = source.read1() if not c then return false end local b = string.byte(c) if b == 7 then -- BEL terminates OSC return true end if b == 27 then -- possible ST: ESC \ local n = source.read1() if not n then return false end if n == "\\" then return true end end end end local parse_key = function(source) local stop_chars = "[ABCDEFHPQRSu~]" -- Skip terminal *replies* interleaved on the input stream (graphics APC -- responses, OSC/DCS answers, SS3, …). Under KKBP every real key is a CSI -- sequence, so an ESC that is not CSI is a reply, not a keypress: consume it -- whole and keep looking for the next key, rather than bailing — a bail -- makes a cooperative reader (gfx loop, combo watcher) treat the reply as -- EOF and stop reading. nil is returned only when the source is exhausted -- (reader closed/cancelled). while true do local first = source.read1() if not first then return nil end if string.byte(first) ~= 27 then -- With PE flag 15 and bracketed paste mode, this shouldn't happen -- for normal input, but handle it gracefully just in case return first end local second = source.read1() if not second then return nil end if second == "[" then break -- CSI: a real key (or a harmless CSI reply, parsed below) end -- ESC + non-CSI: a terminal reply. Consume it and keep looking. if second == "_" or second == "P" or second == "]" or second == "^" or second == "X" then if not skip_string_sequence(source) then return nil end elseif second == "O" then if not source.read1() then -- SS3: one final byte return nil end end -- else ESC + a lone byte (already consumed): loop for the next key end local buf = buffer.new() repeat local c = source.read1() if c then buf:put(c) end until not c or c:match(stop_chars) local seq = buf:get() -- Handle bracketed paste mode (ESC[200~ starts paste, ESC[201~ ends it) if seq == "200~" then local paste_buf = buffer.new() local paste_bytes = 0 local put = function(s) if s then paste_buf:put(s) paste_bytes = paste_bytes + #s end end -- Read one byte, bridging transient idle reads. For a blocking stdio -- source a nil read is a raw-mode timeout mid-burst, so retry up to -- MAX_PASTE_IDLE_READS before reporting nil; for a cooperative (lev) -- source nil means closed/cancelled, so stop at once. Returning nil -- here is the only real end-of-stream while we're inside a paste. local paste_read1 = function() if not source.transient_eof then return source.read1() end for _ = 1, MAX_PASTE_IDLE_READS do local c = source.read1() if c then return c end end return nil end -- A byte already read that still needs main-loop handling (e.g. the byte -- that broke a partial end-marker match, which may itself be an ESC that -- starts the real marker). Carried over rather than re-read. local pending local next_byte = function() if pending ~= nil then local b = pending pending = nil return b end return paste_read1() end -- "201~" follows the ESC[ that opens the closing marker. Match it one -- byte at a time so a non-matching byte isn't swallowed: on the first -- mismatch, the matched prefix becomes literal content and the offending -- byte is re-processed by the main loop. This avoids over-reading past an -- embedded "ESC[" inside the pasted text. local END_TAIL = "[201~" while true do local p = next_byte() if not p then break -- stream closed/cancelled, or no ESC[201~ within the bound end if string.byte(p) == ESC_BYTE then local matched = "" local closed = true for i = 1, #END_TAIL do local c = paste_read1() if c == END_TAIL:sub(i, i) then matched = matched .. c else -- False alarm: emit ESC + the matched prefix as content and -- re-handle the byte that broke the match (nil = stream end). put("\027" .. matched) pending = c closed = false break end end if closed then break -- full ESC[201~ -> end of paste end else put(p) end if paste_bytes >= MAX_PASTE_BYTES then break end end -- The trailing `true` flags this tuple as a paste (not a keypress) for -- callers that must insert it as literal text rather than dispatch it. return paste_buf:get(), nil, nil, nil, nil, true end local modifiers = "1" local event = "1" local codepoint, shifted, base -- Let's handle the weird exception (`1;mod:eventCODEPOINT`) fisrt if seq:match("^1;") then modifiers, event = seq:match("^1;(%d+):([123])") if not event then modifiers = seq:match("^1;(%d+)") event = "1" end codepoint = seq:match("(%u)$") else codepoint = seq:match("^[^:;u~]+") if seq:match(";") then modifiers = seq:match("^[^;]+;(%d+)") or "1" event = seq:match("^[^;]+;%d+:([123])") or "1" end end local alternate_block = seq:match("^[^:;]+:([%d:]+)") if alternate_block then if alternate_block:match("^:") then base = alternate_block:match("^:(.*)") else shifted = alternate_block:match("^(%d+)") if alternate_block:match(":") then base = alternate_block:match("^[^:]+:(.*)") end end end if shifted then shifted = std.utf.char(tonumber(shifted)) end if base then base = std.utf.char(tonumber(base)) end if seq:sub(-1) == "~" and KKBP_CODES_LEGACY[codepoint] then codepoint = KKBP_CODES_LEGACY[codepoint] elseif KKBP_CODES[codepoint] then codepoint = KKBP_CODES[codepoint] else local cp = tonumber(codepoint) if cp then -- KKBP uses Unicode Private Use Area for encoding non-printable keys, -- and our KKBP_CODES mapping does not include all of them. if cp >= PUA_START and cp <= PUA_END then codepoint = "TBD" else codepoint = std.utf.char(cp) end end end modifiers = tonumber(modifiers) or 0 return codepoint, modifiers, tonumber(event), shifted, base end local get = function() return parse_key(stdio_source) end ---! Cooperatively read one key/sequence from a lev fd reader ---@ read_key(`reader`{.tbl}, `opts`{.tbl .opt}) -> ... (same tuple as get()) --[===[ Like get(), but pulls bytes from a `lev.fd_reader` (typically over stdin, fd 0) instead of blocking on io.read. Yields to the event loop until a full key/sequence arrives, so other tasks keep running while waiting. `opts` is forwarded to the reader's read() (e.g. `{ cancel = token }`); when the read is cancelled/closed it returns nil. KKBP sequences split across multiple reads are reassembled correctly. ]===] local read_key = function(reader, opts) local source = { read1 = function() local data = reader:read(1, opts) if not data then return nil end return data end, } return parse_key(source) end -- KKBP modifier bit positions (shared by mods_to_string and string_to_mods) local MODIFIER_BITS = { { 1, "SHIFT" }, { 2, "ALT" }, { 4, "CTRL" }, { 8, "SUPER" }, { 16, "HYPER" }, { 32, "META" }, { 64, "CAPS_LOCK" }, { 128, "NUM_LOCK" }, } -- Reverse lookup: name → bitmask (derived at module load) local MODIFIER_NAMES = {} for _, pair in ipairs(MODIFIER_BITS) do MODIFIER_NAMES[pair[2]] = pair[1] end ---! Convert a modifier bitmask to a human-readable string ---@ mods_to_string(`mods`{.num}) -> `combination`{.str} --[===[ Maps a KKBP modifier bitmask to a `"+"` separated string of modifier names. Bit values: SHIFT=1, ALT=2, CTRL=4, SUPER=8, HYPER=16, META=32, CAPS_LOCK=64, NUM_LOCK=128. Example: `mods_to_string(5)` → `"SHIFT+CTRL"` ]===] local mods_to_string = function(mods) local combination = {} for i = 1, #MODIFIER_BITS do local mask = MODIFIER_BITS[i][1] local name = MODIFIER_BITS[i][2] if bit.band(mods, mask) ~= 0 then table.insert(combination, name) end end return table.concat(combination, "+") end ---! Convert a modifier string to a bitmask ---@ string_to_mods(`combination`{.str}) -> `mods`{.num} --[===[ Parses a `"+"` separated modifier name string and returns the corresponding KKBP bitmask. Unknown names are silently ignored. Example: `string_to_mods("SHIFT+CTRL")` → `5` ]===] local string_to_mods = function(combination) local byte = 0 for modifier in string.gmatch(combination, "%w+") do if MODIFIER_NAMES[modifier] then byte = bit.bor(byte, MODIFIER_NAMES[modifier]) end end return byte end ---! Read a keyboard event and return a simplified key description ---@ simple_get() -> `key`{.str .or_nil}, `base_key`{.str .or_nil} --[===[ Wraps `get()` to return either a single character for printable keys or a shortcut string like `"CTRL+c"` for modified keys. Ignores key release events. The optional second return value `base_key` is the US QWERTY equivalent of the physical key, derived from the KKBP `base` field. It is non-nil only when it differs from the primary key (i.e. when a non-Latin keyboard layout is active). Callers that need layout-independent command dispatch (e.g. editor normal mode) can use `base_key` instead of `key`. ]===] local simple_get = function() local cp, mods, event, shifted, base, is_paste = get() if not cp then return nil end -- A bracketed paste: surface it as a distinct third return so callers can -- insert it as literal text. This must come before the `not mods` check -- below, which a paste also satisfies but which equally matches a legacy -- raw key byte on non-KKBP terminals. if is_paste then return cp, nil, true end -- Handle case where get() returns just a character (non-KKBP input) if not mods then return cp end -- Ignore key release events (event == 3) if event == 3 then return nil end -- Ignore bare modifier / lock key events (reported by KKBP flag 8) if MODIFIER_ONLY_KEYS[cp] then return nil end if mods <= MAX_PRINTABLE_MODS and std.utf.len(cp) < 2 then if shifted then cp = shifted end -- Compute effective US-layout key from KKBP base local effective_base if base and base ~= cp then if mods == 2 then -- Shift only effective_base = US_SHIFT_MAP[base] or base:upper() else effective_base = base end end return cp, effective_base end if base then cp = base end local mod_string = mods_to_string(mods - 1) local shortcut = mod_string .. "+" if mod_string == "" then shortcut = cp else shortcut = shortcut .. cp end return shortcut end return { -- Synchronized output SYNC_START = SYNC_START, SYNC_END = SYNC_END, sync = sync, -- Constants TS_PRESETS = TS_PRESETS, COLORS = COLORS, kitty_notify = kitty_notify, title = title, go = go, move = move, clear_line = clear_line, cursor_position = cursor_position, write = write, write_at = write_at, clear = clear, hide_cursor = hide_cursor, show_cursor = show_cursor, style = style, color = color, set_style = set_style, set_color = set_color, set_raw_mode = set_raw_mode, set_sane_mode = set_sane_mode, is_tty = core.is_tty, resized = core.resized, raw_mode = core.raw_mode, window_size = core.get_window_size, get_pixel_dimensions = get_pixel_dimensions, switch_screen = switch_screen, alt_screen = alt_screen, has_kkbp = has_kkbp, enable_kkbp = enable_kkbp, disable_kkbp = disable_kkbp, enable_bracketed_paste = enable_bracketed_paste, disable_bracketed_paste = disable_bracketed_paste, get = get, read_key = read_key, simple_get = simple_get, mods_to_string = mods_to_string, string_to_mods = string_to_mods, -- Text sizing protocol validate_ts_opts = validate_ts_opts, build_ts_meta = build_ts_meta, text_size = text_size, text_size_chunks = text_size_chunks, has_ts = has_ts, has_ts_combined = has_ts_combined, init_ts_width_mode = init_ts_width_mode, ts_width_mode = ts_width_mode, supports_ts = supports_ts, -- Cancel (SIGINT) API install_cancel_handler = core.install_cancel_handler, remove_cancel_handler = core.remove_cancel_handler, check_cancel = core.check_cancel, }