vaclavt/vterm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

small refactoring

Browse Source
This commit is contained in:
Vaclav Tvrdik
2026-09-08 12:48:52 +02:00
parent f672bff6a0
commit eafe53b2fb
9 changed files with 118 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
Architecture.md
View File

@@ -338,8 +338,9 @@ handle): `auto_close`, `scrollback_history_size`, `honor_erase_scrollback`,
`parameters`, `debug`. `page_scroll_modifier` is the modifier that turns PgUp/PgDn into a
page scroll of the view instead of a key for the app. `terminal` and
`parameters` name the external terminal spawned by *vterm: Open Terminal in Project
Folder* (`%CWD%` in `parameters` becomes the project folder). `honor_erase_scrollback` is off, so ED 3
blanks the screen without dropping the scrollback: pi and Claude Code repaint by erasing
Folder* (`%CWD%` in `parameters` becomes the project folder). ED 3 never touches the visible screen (as in
xterm), and `honor_erase_scrollback` is off, so it does not drop the scrollback either:
pi and Claude Code repaint by erasing
saved lines, which would take the history with them (`cmd+shift+k` still empties the tab). Changing `theme` fires `SchemeManager.reload`, which rebuilds the
palette and restyles open terminals in place.

28
tests/test_emulator.py
View File

@@ -203,6 +203,24 @@ def run_resize_cases():
feed(stream, "\x1b[?1049l", False)
assert max(screen.buffer[0]) < 8, sorted(screen.buffer[0])
# a shrink clamps the cursor (pyte does it via restore_cursor); without
# the clamp draw() writes to rows above or below the screen
proc, screen, stream = make(10, 8)
feed(stream, "\r\n".join("line{}".format(i) for i in range(8)), False)
feed(stream, "\x1b[1;1H", False) # full screen, cursor at home
screen.resize(4, 10)
assert screen.cursor.y == 0, screen.cursor.y
proc, screen, stream = make(10, 8)
feed(stream, "top\x1b[7;1H", False) # cursor on a blank row below content
screen.resize(4, 10)
assert screen.cursor.y == 3, screen.cursor.y
proc, screen, stream = make(20, 4)
feed(stream, "x" * 15, False)
screen.resize(4, 10)
assert screen.cursor.x == 10, screen.cursor.x
# mode 2031 is how a live theme change reaches the app
proc, screen, stream = make()
assert not screen.theme_notify_enabled()
@@ -381,8 +399,7 @@ def run_clear_cases():
assert len(screen.history) > 0, "history suppressed for too long"
# ED 3 is app-driven: it clears scrollback and fires the clear callback,
# and (as in pyte/Terminus) also blanks the screen. Both agents pair it
# with 2J and repaint, so the render side must not wipe or nudge again.
# but (as in xterm, unlike pyte) leaves the visible screen alone
cleared = []
proc = FakeProcess()
screen = VtScreen(10, 4, process=proc, clear_callback=lambda: cleared.append(1))
@@ -391,9 +408,10 @@ def run_clear_cases():
assert len(screen.history) > 0
feed(stream, "\x1b[3J", False)
assert len(screen.history) == 0 and cleared, (len(screen.history), cleared)
assert screen.buffer[0][0].data == "a", screen.buffer[0][0]
# with erase_scrollback off, ED 3 still blanks the screen but the history
# and the view survive: agents repaint with it and would take both
# with erase_scrollback off, ED 3 is a no-op: agents repaint with it and
# would take the history and the view with them
cleared = []
proc = FakeProcess()
screen = VtScreen(10, 4, process=proc, erase_scrollback=False,
@@ -405,7 +423,7 @@ def run_clear_cases():
del cleared[:] # the screen's own reset() fired one at construction
feed(stream, "\x1b[3J", False)
assert len(screen.history) == kept and not cleared, (len(screen.history), cleared)
assert screen.buffer[0][0].data == " ", screen.buffer[0][0]
assert screen.buffer[0][0].data == "a", screen.buffer[0][0]
print("all clear-screen tests passed")

2
tests/test_imports.py
View File

@@ -81,7 +81,7 @@ for entry in load_sublime_json("vterm.sublime-commands") + keymap:
# every pending_* name mentioned in the modules must be a real Terminal
# attribute, so a renamed flag cannot leave a writer pointing at nothing
flags = {f for f in vars(Terminal(None)) if f.startswith("pending_")}
assert flags == {"pending_clear", "pending_clear_screen"}, flags
assert flags == {"pending_clear", "pending_clear_screen", "pending_render"}, flags
for module in (commands, render, terminal_module):
with open(module.__file__, encoding="utf-8") as f:
# (?<!_) skips the private _pending_to_send_string write-queue flag

13
tests/test_render.py
View File

@@ -595,10 +595,21 @@ line[2] = Char("x", "blue", "default")
runs = [r for r in segment_buffer_line(line) if r[1] != r[2]]
assert runs == [("漢", 0, 1, "red", "default", ""), ("x", 1, 2, "blue", "default", "")], runs
# an orphaned wide-char filler (its wide half was overwritten with a narrow
# char) still occupies a column: rendered as a space so later spans stay aligned
line = StaticDefaultDict(Char(" "))
line[0] = Char("a", "default", "default")
line[1] = Char("", "red", "default")
line[2] = Char("x", "blue", "default")
runs = [r for r in segment_buffer_line(line) if r[1] != r[2]]
assert runs == [("a", 0, 1, "default", "default", ""),
(" ", 1, 2, "red", "default", ""),
("x", 2, 3, "blue", "default", "")], runs
# reverse video swaps the pair, and default becomes reverse_default
line = StaticDefaultDict(Char(" "))
line[0] = Char("r", "red", "default", False, False, False, False, True)
runs = [r for r in segment_buffer_line(line) if r[1] != r[2]]
assert runs == [("r", 0, 1, "reverse_default", "red", "")], runs
print("segmenter tests passed (leading gap, wide char, reverse)")
print("segmenter tests passed (leading gap, wide char, orphan filler, reverse)")

42
vterm/emulator.py
View File

@@ -44,6 +44,8 @@ BG_AIXTERM = {100 + i: ANSI_COLORS[8 + i] for i in range(8)}
REPORTABLE_MODES = (mo.DECAWM >> 5, mo.DECTCEM >> 5, 1000, 1002, 1003, 1004, 1006, 1049, 2004, 2026)
# DEC private modes, pre-shifted into pyte's internal representation.
# 1: DECCKM, application cursor keys (arrows send SS3); pyte has no name for it
APP_CURSOR_MODE = 1 << 5
# 2026: synchronized output, the app is mid-frame and we should not draw
SYNC_MODE = 2026 << 5
# 1049: alternate screen, a full-screen app owns the viewport
@@ -91,7 +93,7 @@ def quantize_256(r, g, b):
return 16 + 36 * level(r) + 6 * level(g) + level(b)
# linefeed marks a hard wrap, so copy can re-join soft-wrapped lines
# linefeed marks an autowrap (soft wrap), so copy can re-join the halves
Char = namedtuple(
"Char",
["data", "fg", "bg", "bold", "italics", "underscore", "strikethrough", "reverse", "linefeed"],
@@ -227,6 +229,10 @@ class VtScreen(pyte.Screen):
line.pop(x, None)
self.lines, self.columns = lines, columns
# pyte clamps via restore_cursor; without it a shrink leaves the cursor
# negative or past the new edge. x == columns keeps a pending wrap
self.cursor.x = min(self.cursor.x, columns)
self.cursor.y = min(max(self.cursor.y, 0), lines - 1)
self.set_margins()
self.tabstops = set(range(8, self.columns, 8))
@@ -364,21 +370,14 @@ class VtScreen(pyte.Screen):
elif char_width > 0:
cursor.x -= char_width
if char_width == 1:
if char_width == 1 or char_width == 2:
if insert_mode:
self.insert_characters(char_width)
cell = cache.get(char)
if cell is None:
cell = cache[char] = cursor.attrs._replace(data=char)
line[cursor.x] = cell
elif char_width == 2:
if insert_mode:
self.insert_characters(char_width)
cell = cache.get(char)
if cell is None:
cell = cache[char] = cursor.attrs._replace(data=char)
line[cursor.x] = cell
if cursor.x + 1 < columns:
if char_width == 2 and cursor.x + 1 < columns:
filler = cache.get("")
if filler is None:
filler = cache[""] = cursor.attrs._replace(data="")
@@ -430,11 +429,18 @@ class VtScreen(pyte.Screen):
if not self.alternate_buffer_mode and (how == 2 or (how == 0 and self.cursor.x == 0 and self.cursor.y == 0)):
self.push_lines_into_history()
if how == 3:
# xterm's ED 3 erases the scrollback only; the screen stays
if self._erase_scrollback:
self.history.clear()
self._clear_callback()
return
if how == 0:
interval = range(self.cursor.y + 1, self.lines)
elif how == 1:
interval = range(self.cursor.y)
elif how == 2 or how == 3:
elif how == 2:
interval = range(self.lines)
else:
return
@@ -451,10 +457,6 @@ class VtScreen(pyte.Screen):
if how == 0 or how == 1:
self.erase_in_line(how)
if how == 3 and self._erase_scrollback:
self.history.clear()
self._clear_callback()
def select_graphic_rendition(self, *attrs, **kwargs):
"""Apply an SGR sequence to the cursor attributes.
@@ -490,9 +492,9 @@ class VtScreen(pyte.Screen):
if attr[1:] != "blink":
replace[attr[1:]] = attr.startswith("+")
elif attr in FG_AIXTERM:
replace.update(fg=FG_AIXTERM[attr])
replace["fg"] = FG_AIXTERM[attr]
elif attr in BG_AIXTERM:
replace.update(bg=BG_AIXTERM[attr])
replace["bg"] = BG_AIXTERM[attr]
elif attr in (g.FG_256, g.BG_256):
key = "fg" if attr == g.FG_256 else "bg"
try:
@@ -592,7 +594,11 @@ class VtScreen(pyte.Screen):
self.write_process_input("\x1b[{}{};{}R".format(marker, y, x))
elif mode == 996 and private:
# color-scheme report, pi asks with a 2s timeout
self.write_process_input("\x1b[?997;{}n".format(1 if self.theme_is_dark else 2))
self.write_process_input(self.theme_report())
def theme_report(self):
"""DSR 997 report: the palette is dark (1) or light (2). Also sent on a live theme change."""
return "\x1b[?997;{}n".format(1 if self.theme_is_dark else 2)
def report_xtversion(self):
"""XTVERSION: naming ourselves is what unlocks ?2026 for apps on a generic TERM."""

6
vterm/listeners.py
View File

@@ -25,8 +25,10 @@ class VtermCoreEventListener(sublime_plugin.EventListener):
terminal.focused = True
terminal.send_focus(True)
view.run_command("vterm_clear_undo_stack")
# catch up on the frames, title and caret skipped while in the background
view.run_command("vterm_render")
# catch up on the frames, title and caret skipped while in the
# background — via the renderer tick: a render here would race the
# renderer thread, which owns the emulator
terminal.pending_render = True
view.run_command("vterm_show_cursor", {"scroll": False})
elif view.settings().get("vterm_view") and not view.settings().get("vterm_view.finished"):
# the plugin host restarted; this tab's terminal is unreachable

2
vterm/mouse.py
View File

@@ -39,7 +39,7 @@ class VtermMouseEventListener(sublime_plugin.EventListener):
if not terminal.screen.mouse_reporting_enabled():
# leave selection, caret and scrolling to sublime
return
if name == "drag_select" and args and "event" in args and args.get("event", {}).get("button") == 1 and len(args) == 1:
if name == "drag_select" and args and args.get("event", {}).get("button") == 1 and len(args) == 1:
return ("vterm_mouse_click", {"event": args["event"]})

25
vterm/render.py
View File

@@ -49,6 +49,11 @@ STYLE_FLAGS = (("bold", "b"), ("italics", "i"), ("underscore", "u"))
ATTRS = slice(1, 8)
def is_default_run(fg, bg, style):
"""True for a run that needs no colour overlay."""
return fg == "default" and bg == "default" and not style
def segment_buffer_line(buffer_line):
"""Yield (text, start, end, fg, bg, style) runs of a screen row.
@@ -82,7 +87,9 @@ def segment_buffer_line(buffer_line):
start = column
attrs = char[ATTRS]
run.append(char.data)
# an orphaned wide-char filler (its first half was overwritten) still
# occupies a column; render a space so later spans stay aligned
run.append(char.data or " ")
column += 1
if attrs is not None:
@@ -283,8 +290,7 @@ class VtermRenderCommand(sublime_plugin.TextCommand):
for line in range(popped):
buffer_line = history.pop()
lf = buffer_line[columns - 1].linefeed
self.update_line(edit, offset - line - 1, buffer_line, lf)
self.update_line(edit, offset - line - 1, buffer_line, columns)
# after the loop: a pop edits an earlier point than the one before
# it, so spans measured mid-loop would be stale
@@ -292,17 +298,18 @@ class VtermRenderCommand(sublime_plugin.TextCommand):
for line in dirty_lines:
buffer_line = screen.buffer[line]
lf = buffer_line[columns - 1].linefeed
self.update_line(edit, line + offset, buffer_line, lf)
self.update_line(edit, line + offset, buffer_line, columns)
def update_line(self, edit, line, buffer_line, lf):
def update_line(self, edit, line, buffer_line, columns):
"""Rewrite one view row from a screen row, unless it is unchanged.
Args:
line: absolute view row.
lf: the row ended in a real newline, not a soft wrap.
columns: screen width; the last cell's linefeed flag marks a soft
wrap, which gets the CONTINUATION sentinel.
"""
view = self.view
lf = buffer_line[columns - 1].linefeed
segments = list(segment_buffer_line(buffer_line))
# pyte's dirty set is a repaint hint, not a diff: most dirty rows come
# back identical, and rewriting one costs an edit plus its region churn
@@ -314,7 +321,7 @@ class VtermRenderCommand(sublime_plugin.TextCommand):
by_scope = {}
needed = 0
for _, start, end, fg, bg, style in segments:
if fg == "default" and bg == "default" and not style:
if is_default_run(fg, bg, style):
continue
needed = max(needed, end)
by_scope.setdefault(scheme_manager.scope(fg, bg, style), []).append((start, end))
@@ -392,7 +399,7 @@ class VtermRenderCommand(sublime_plugin.TextCommand):
segments = self._row_cache[row][0]
start = view.line(view.text_point(row, 0)).begin()
for _, a, b, fg, bg, style in segments:
if fg == "default" and bg == "default" and not style:
if is_default_run(fg, bg, style):
continue
scope = scheme_manager.scope(fg, bg, style)
grouped.setdefault(scope, []).append(sublime.Region(start + a, start + b))

54
vterm/terminal.py
View File

@@ -9,8 +9,9 @@ from queue import Empty, Queue
import sublime
from ptyprocess import PtyProcess
from pyte import modes as mo
from .emulator import VtScreen, VtStream
from .emulator import APP_CURSOR_MODE, VtScreen, VtStream
from .keys import get_key_code
from .scheme import scheme_manager
from .utils import intermission, responsive, settings
@@ -93,7 +94,7 @@ class Terminal:
self.default_title = ""
self.offset = 0
self.viewport_y = 0
self.scrollback_size = 5000
self.scrollback_size = 50000
self._hosted = True
self._hosted_checked = 0
self.focused = True
@@ -104,10 +105,15 @@ class Terminal:
self.applied_alt_screen = False
self._strings = Queue()
self._pending_to_send_string = False
# guards the queue/flag handoff between send_string and the drain thread
self._send_lock = threading.Lock()
self._wheel_captured = False
self._scroll_pending = False
self.pending_clear = False
self.pending_clear_screen = False
# a forced frame (set on activation); rendering stays on the renderer
# thread, which owns the emulator
self.pending_render = False
# a UTF-8 sequence split across read chunks must not be dropped
self._decoder = codecs.getincrementaldecoder("utf-8")("replace")
@@ -192,7 +198,7 @@ class Terminal:
screen.default_bg_color = scheme_manager.background.lstrip("#")
screen.theme_is_dark = scheme_manager.is_dark()
if screen.theme_notify_enabled():
self.send_string("\x1b[?997;{}n".format(1 if screen.theme_is_dark else 2), normalized=False)
self.send_string(screen.theme_report(), normalized=False)
def kill(self):
"""Terminate the child and drop this terminal from the registry."""
@@ -218,8 +224,9 @@ class Terminal:
def _need_to_render(self):
"""Whether anything changed since the last frame: dirty rows, a clear, or the cursor."""
screen = self.screen
if self.pending_clear or self.pending_clear_screen:
# a pending clear must render even when the app is idle
if self.pending_render or self.pending_clear or self.pending_clear_screen:
# a pending clear or forced frame must render even when the app is idle
self.pending_render = False
return True
cursor = (screen.cursor.x, screen.cursor.y, screen.cursor_hidden(), screen.cursor_style)
if screen.dirty or cursor != self._cached_cursor:
@@ -329,8 +336,8 @@ class Terminal:
def send_key(self, **kwargs):
"""Encode a key name plus modifiers for the app's current modes and send it."""
kwargs["application_mode"] = (1 << 5) in self.screen.mode
kwargs["new_line_mode"] = 20 in self.screen.mode
kwargs["application_mode"] = APP_CURSOR_MODE in self.screen.mode
kwargs["new_line_mode"] = mo.LNM in self.screen.mode
self.send_string(get_key_code(**kwargs), normalized=False)
def send_string(self, string, normalized=True):
@@ -342,22 +349,24 @@ class Terminal:
"""
if normalized:
string = string.replace("\r\n", "\n")
if 20 in self.screen.mode:
if mo.LNM in self.screen.mode:
string = string.replace("\n", "\r\n")
else:
string = string.replace("\n", "\r")
# chunked so a big paste cannot overrun the pty buffer; the blocking
# pty write is the backpressure, no pacing sleep needed
no_queue = not self._pending_to_send_string
if no_queue and len(string) <= 4096:
self.process.write(string)
else:
for i in range(0, len(string), 4096):
self._strings.put(string[i : i + 4096])
if no_queue:
self._pending_to_send_string = True
threading.Thread(target=self._process_send_string, daemon=True).start()
with self._send_lock:
no_queue = not self._pending_to_send_string
if not (no_queue and len(string) <= 4096):
for i in range(0, len(string), 4096):
self._strings.put(string[i : i + 4096])
if no_queue:
self._pending_to_send_string = True
threading.Thread(target=self._process_send_string, daemon=True).start()
return
# outside the lock: the pty write can block on backpressure
self.process.write(string)
def _process_send_string(self):
"""Drain the queued paste chunks; the blocking pty write paces them."""
@@ -365,10 +374,15 @@ class Terminal:
try:
# linger so a burst of sends does not respawn the thread
string = self._strings.get(timeout=0.2)
self.process.write(string)
except Empty:
self._pending_to_send_string = False
return
# under the lock: a chunk enqueued after the timeout but before
# the flag drops would otherwise strand until the next send
with self._send_lock:
if self._strings.empty():
self._pending_to_send_string = False
return
continue
self.process.write(string)
def _sync_wheel_flag(self):
"""Publish whether the app owns the wheel, for the mousemap context."""