End DCS/OSC strings at any ESC, default the external terminal to Terminal.app

A string with no ST swallowed every later sequence, RIS included, and
Clear Screen could not recover the tab. Any ESC now ends it, as in xterm,
and the sequence it starts is dispatched. The shipped setting needed Kitty.

Also: REP, SCOSC/SCORC, DECOM-relative CPR, `:` subparams and more DECRQM
modes answered; new rows inserted in one edit; controls stripped from
pastes; wide-char mouse columns.
This commit is contained in:
Vaclav Tvrdik
2026-11-01 20:22:56 +01:00
parent 93253aa564
commit 5957e0c718
9 changed files with 264 additions and 55 deletions

View File

@@ -296,11 +296,12 @@ def drive(feed, need_to_render, render, data, tick):
def run_vterm(data, tick):
from vterm.emulator import VtScreen, VtStream
from vterm.render import VtermRenderCommand
from vterm.scheme import get_scheme_manager
from vterm.terminal import Terminal
counter = collections.Counter()
view = CountingView(counter)
terminal = Terminal(view)
terminal = Terminal(view, get_scheme_manager("default"))
terminal.scrollback_size = SCROLLBACK
terminal.process = FakeProcess()
terminal.screen = VtScreen(COLUMNS, ROWS, process=terminal.process, history=SCROLLBACK)

View File

@@ -4,7 +4,7 @@ import _bootstrap # sets sys.path for the imports below
from _bootstrap import FakeProcess
import sublime
from vterm.commands import VtermPasteSelectionCommand, VtermSelectThemeCommand, selection_text
from vterm.commands import VtermPasteSelectionCommand, VtermPasteTextCommand, VtermSelectThemeCommand, selection_text
from vterm.emulator import VtScreen
from vterm.render import CONTINUATION
from vterm.scheme import get_scheme_manager
@@ -245,3 +245,17 @@ view = FakeView("abcdef", [sublime.Region(2, 2)])
assert paste(view) == [], view.commands
print("selection tests passed (sentinel stripping, paste, empty no-op)")
# a paste is a trust boundary: C0 controls bar tab/LF/CR go, so a pasted
# ESC[201~ cannot close the bracket and type the rest into the app
terminal = fake_terminal("default")
terminal.process = terminal.screen._process
terminal.screen.set_mode(2004, private=True)
terminal.view.run_command = lambda name, args=None: None
command = VtermPasteTextCommand.__new__(VtermPasteTextCommand)
command.view = terminal.view
command.run(None, "a\x1b[201~b\x00\tc\nd")
assert terminal.process.written == "\x1b[200~a[201~b\tc\rd\x1b[201~", repr(terminal.process.written)
print("paste tests passed (controls stripped, bracket kept whole)")

View File

@@ -454,8 +454,67 @@ def run_clear_cases():
print("all clear-screen tests passed")
def run_parser_edge_cases():
"""Sequences the stock collector misread: intermediates, `:` subparams, REP, marks."""
for char_by_char in (False, True):
tag = "char-by-char" if char_by_char else "chunked"
# DECSTR (`reset` sends it): `!` is an intermediate, the `p` must not be drawn
proc, screen, stream = make(20, 4)
feed(stream, "\x1b[!pX", char_by_char)
assert screen.display[0].rstrip() == "X", (tag, repr(screen.display[0]))
# `4:3` is curly underline, not SGR 43 (yellow background)
proc, screen, stream = make(20, 4)
feed(stream, "\x1b[4:3mX\x1b[38:2:1:2:3mY", char_by_char)
cell = screen.buffer[0][0]
assert (cell.bg, cell.underscore) == ("default", True), (tag, cell)
assert screen.buffer[0][1].data == "Y", (tag, screen.buffer[0][1])
# REP repeats the last printable; ncurses emits it for runs
proc, screen, stream = make(20, 4)
feed(stream, "a\x1b[3b\x1b[31mb\x1b[b", char_by_char)
assert screen.display[0].rstrip() == "aaaabb", (tag, repr(screen.display[0]))
assert screen.buffer[0][5].fg == "red", (tag, screen.buffer[0][5])
# a Thai vowel sign has combining class 0 but is still a mark: merged, not dropped
proc, screen, stream = make(20, 4)
feed(stream, "\u0e01\u0e31x", char_by_char)
assert screen.buffer[0][0].data == "\u0e01\u0e31", (tag, screen.buffer[0][0])
assert screen.buffer[0][1].data == "x", (tag, screen.buffer[0][1])
# DECRQM: the modes we honour answer set/reset, never "unrecognized"
proc, screen, stream = make()
feed(stream, "\x1b[?2031$p\x1b[?1h\x1b[?1$p\x1b[4$p\x1b[20$p", char_by_char)
assert proc.written == "\x1b[?2031;2$y\x1b[?1;1$y\x1b[4;2$y\x1b[20;2$y", (tag, proc.written)
# RIS on the alternate screen lands back on a cleared primary screen
proc, screen, stream = make(20, 4)
feed(stream, "primary\x1b[?1049halt\x1bc", char_by_char)
assert not screen.alternate_buffer_mode and not screen.savepoints, tag
assert screen.display[0].strip() == "", (tag, repr(screen.display[0]))
feed(stream, "\r\n" * 4, char_by_char)
assert len(screen.history) == 1, (tag, len(screen.history))
# SCOSC/SCORC, and CPR relative to the region under DECOM
proc, screen, stream = make(20, 4)
feed(stream, "ab\x1b[s\x1b[3;3H\x1b[uX\x1b[2;3r\x1b[?6h\x1b[6n", char_by_char)
assert screen.display[0].rstrip() == "abX", (tag, repr(screen.display[0]))
assert proc.written == "\x1b[1;1R", (tag, proc.written)
# a string with no terminator ends at the next ESC, as in xterm; a stray
# ESC P from a binary must not swallow the rest of the session
proc, screen, stream = make(20, 4)
feed(stream, "\x1bPgarbage\x1bcX\x1b]0;title\x1b[1mY", char_by_char)
assert screen.display[0].rstrip() == "XY", (tag, repr(screen.display[0]))
assert screen.title == "title" and screen.buffer[0][1].bold, (tag, screen.title)
print("all parser edge-case tests passed")
run_case(char_by_char=False)
run_case(char_by_char=True)
run_parser_edge_cases()
run_clear_cases()
run_resize_cases()
run_private_csi_cases()

View File

@@ -357,7 +357,13 @@ command = make_command(terminal)
stream.feed("".join("\x1b[31mr{}\x1b[0m\r\n".format(i) for i in range(300)))
command.update_lines(None, terminal)
terminal.screen.dirty.clear()
assert set(command.state.archived_chunks) == {0}, command.state.archived_chunks
# rows new to the view went in as one insert, coloured straight onto their
# chunk keys; only row 0 and the two coloured screen rows have a key of their own
assert set(command.state.archived_chunks) == {0, 1}, command.state.archived_chunks
assert len(terminal.view.regions["vterm@1@vterm.red.default"][1]) == terminal.offset - CHUNK
row_keys = [k for k in terminal.view.regions if k.startswith("vterm#")]
assert len(row_keys) == 3, row_keys
assert terminal.view.text.split("\n")[:3] == ["r0", "r1", "r2"], terminal.view.text[:20]
command.trim_history(None, terminal)
assert "vterm@0@vterm.red.default" not in terminal.view.regions, "trimmed chunk key left behind"
assert 0 not in command.state.archived_chunks
@@ -665,7 +671,15 @@ terminal.view.window_to_text = lambda xy: terminal.view.text_point(2, 4)
mouse_release.run(None, event)
assert terminal.process.written == "\x1b[<0;1;1M\x1b[<32;5;3M\x1b[<0;5;3m", repr(terminal.process.written)
print("click tests passed (first click focuses only; press, drag and release report)")
# the column is counted in cells: a wide char before the point takes two
terminal.process.writes.clear()
terminal.view.text = "\u65e5\u672cx \nr1 \nr2 \nr3 \nr4 \n"
terminal.view.window_to_text = lambda xy: terminal.view.text_point(0, 2)
mouse_press.run(None, event)
mouse_release.run(None, event)
assert terminal.process.written == "\x1b[<0;5;1M\x1b[<0;5;1m", repr(terminal.process.written)
print("click tests passed (first click focuses only; press, drag and release report; wide chars)")
# entering the alt screen folds the scrollback away (kitty hides it); leaving
@@ -751,4 +765,12 @@ 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, orphan filler, reverse)")
# a cell can hold two code points (a mark NFC could not compose): the spans
# that follow are offsets into the row text, so they move by two
line = StaticDefaultDict(Char(" "))
line[0] = Char("q\u0301", "default", "default")
line[1] = Char("x", "blue", "default")
runs = [r for r in segment_buffer_line(line) if r[1] != r[2]]
assert runs == [("q\u0301", 0, 2, "default", "default", ""), ("x", 2, 3, "blue", "default", "")], runs
print("segmenter tests passed (leading gap, wide char, orphan filler, reverse, two-code-point cell)")

View File

@@ -33,12 +33,12 @@
// External terminal for "vterm: Open Terminal in Project Folder".
// %CWD% in parameters is replaced with the project folder.
// "terminal": "/usr/bin/open",
// "parameters": ["-a", "Terminal", "%CWD%"],
"terminal": "/usr/bin/open",
"parameters": ["-a", "Terminal", "%CWD%"],
// kitty instead of Apple Terminal:
"terminal": "/usr/bin/open",
"parameters": ["-a", "Kitty", "%CWD%"],
// "terminal": "/usr/bin/open",
// "parameters": ["-a", "Kitty", "%CWD%"],
"debug": false
}

View File

@@ -40,7 +40,7 @@ FG_AIXTERM = {90 + i: ANSI_COLORS[8 + i] for i in range(8)}
BG_AIXTERM = {100 + i: ANSI_COLORS[8 + i] for i in range(8)}
# DECRQM answers, as wire numbers (pyte stores modes shifted << 5)
REPORTABLE_MODES = (mo.DECAWM >> 5, mo.DECTCEM >> 5, 1000, 1002, 1003, 1004, 1006, 1049, 2004, 2026)
REPORTABLE_MODES = (1, mo.DECAWM >> 5, mo.DECTCEM >> 5, 1000, 1002, 1003, 1004, 1006, 1049, 2004, 2026, 2031)
# DEC private modes, pre-shifted into pyte's internal representation.
# 1: DECCKM, application cursor keys (arrows send SS3)
@@ -153,6 +153,8 @@ class VtScreen(pyte.Screen):
self._char_caches = {}
self._char_cache = {}
self._cache_attrs = None
# last printable drawn, for REP
self._last_char = ""
self.history = deque(maxlen=history)
self.alternate_buffer_mode = False
self._sync_frame_start = 0
@@ -165,11 +167,16 @@ class VtScreen(pyte.Screen):
def reset(self):
"""RIS: reset the screen along with the cursor, scrollback and caches"""
if self.alternate_buffer_mode:
# the primary screen is what a reset clears; leave the alt one behind
self.reset_mode(1049, private=True)
self._default_char = Char(data=" ", fg="default", bg="default", reverse=False)
self._sgr_cache.clear()
self._char_caches.clear()
self._cache_attrs = None
self._last_char = ""
super().reset()
self.savepoints = []
self.cursor = Cursor(0, 0)
self.cursor_style = 0
self.modify_other_keys = 0
@@ -376,6 +383,7 @@ class VtScreen(pyte.Screen):
dirty.add(cursor.y)
cursor.x = x + take
pos = end
self._last_char = data[-1]
return
line = buffer[cursor.y]
@@ -400,6 +408,7 @@ class VtScreen(pyte.Screen):
if cell is None:
cell = cache[char] = attrs._replace(data=char)
line[cursor.x] = cell
self._last_char = char
if char_width == 2 and cursor.x + 1 < columns:
filler = cache.get("")
if filler is None:
@@ -407,7 +416,10 @@ class VtScreen(pyte.Screen):
line[cursor.x + 1] = filler
# a combining mark belongs to the cell before it, which may be the
# last column of the previous row
elif char_width == 0 and unicodedata.combining(char):
# a mark: by category, not combining(), which is 0 for Thai/Indic
# vowel signs. Variation selectors are marks too but stay dropped:
# merged, VS16 would make the cell width 2 and swallow its neighbour
elif char_width == 0 and unicodedata.category(char)[0] == "M" and not "\ufe00" <= char <= "\ufe0f":
# sublime renders decomposed combining chars poorly; merge as NFC
pos = None
last = None
@@ -439,6 +451,11 @@ class VtScreen(pyte.Screen):
self.dirty.add(cursor.y)
def repeat_character(self, count=None):
"""REP: draw the last printable again `count` times."""
if self._last_char:
self.draw(self._last_char * (count or 1))
def index(self):
"""Line feed; at the bottom of the primary screen the top row goes to scrollback."""
# a region spanning the whole screen (CSI 1;<lines> r) is no region
@@ -615,6 +632,8 @@ class VtScreen(pyte.Screen):
elif mode == 6:
x = min(self.cursor.x, self.columns - 1) + 1
y = self.cursor.y + 1
if mo.DECOM in self.mode and self.margins:
y -= self.margins.top
marker = "?" if private else ""
self.write_process_input("\x1b[{}{};{}R".format(marker, y, x))
elif mode == 996 and private:
@@ -632,7 +651,10 @@ class VtScreen(pyte.Screen):
def report_mode_status(self, mode, private):
"""DECRQM: answer 1 set, 2 reset, 0 for modes we do not implement."""
if not private:
self.write_process_input("\x1b[{};0$y".format(mode))
value = 0
if mode in (mo.IRM, mo.LNM):
value = 1 if mode in self.mode else 2
self.write_process_input("\x1b[{};{}$y".format(mode, value))
return
if mode in REPORTABLE_MODES:
value = 1 if (mode << 5) in self.mode else 2
@@ -683,6 +705,9 @@ class VtStream(pyte.Stream):
self.csi = dict(self.csi)
self.csi["S"] = "scroll_up"
self.csi["T"] = "scroll_down"
self.csi["b"] = "repeat_character"
self.csi["s"] = "save_cursor"
self.csi["u"] = "restore_cursor"
self._carry = ""
# param-string -> parsed tuple; SGR strings repeat across a whole frame
self._params_cache = {}
@@ -746,7 +771,8 @@ class VtStream(pyte.Stream):
raw = match.group(1)
params = cache.get(raw)
if params is None:
params = tuple(min(int(p), 9999) if p else 0 for p in raw.split(";"))
# no parameters is (), not (0,): save_cursor takes none
params = tuple(min(int(p), 9999) if p else 0 for p in raw.split(";")) if raw else ()
if len(cache) > 4096:
cache.clear()
cache[raw] = params
@@ -805,24 +831,33 @@ class VtStream(pyte.Stream):
escape_dispatch = create_dispatcher(self.escape)
csi_dispatch = create_dispatcher(self.csi)
# the byte after an ESC that cut a string short: any ESC ends a string,
# as in xterm, or one with no ST would swallow every later sequence
pending = None
while True:
char = yield True
char = (yield True) if pending is None else ESC
# ESC introducer: fold the 7-bit forms onto their C1 equivalents
if char == ESC:
if pending is None:
char = yield None
else:
char, pending = pending, None
if char == "[":
char = CSI_C1
elif char == "]":
char = OSC_C1
elif char in "PX^_":
# consume the payload so it never reaches the screen
prev = ""
while True:
c = yield None
if c == ctrl.ST_C1 or (prev == ESC and c == "\\"):
if c == ESC:
c = yield None
if c != "\\":
pending = c
break
if c == ctrl.ST_C1:
break
prev = c
continue
else:
if char == "#":
@@ -848,13 +883,15 @@ class VtStream(pyte.Stream):
private = False
prefix = ""
intermediate = ""
# inside a `:` subparameter list, which is dropped (4:3 must not read as 43)
sub = False
while True:
char = yield None
if char == "?":
private = True
elif char in "><=":
prefix = char
elif char in " $":
elif " " <= char <= "/":
intermediate = char
elif char in ALLOWED_IN_CSI:
basic_dispatch[char]()
@@ -862,14 +899,16 @@ class VtStream(pyte.Stream):
draw(char)
break
elif char.isdigit():
if not sub:
current += char
elif char == ":":
# SGR subparameters, which we do not support
pass
sub = True
elif char == ";":
params.append(min(int(current or 0), 9999))
current = ""
sub = False
else:
if current or params:
params.append(min(int(current or 0), 9999))
self._dispatch_csi(csi_dispatch, listener, char, params, private, prefix, intermediate)
break
@@ -885,18 +924,19 @@ class VtStream(pyte.Stream):
break
payload = ""
if char == ";" or (char not in OSC_TERMINATORS and char != ESC):
while True:
if char == ";":
char = yield None
while char not in OSC_TERMINATORS:
if char == ESC:
char += yield None
if char in OSC_TERMINATORS:
char = yield None
if char != "\\":
pending = char
# char holds the whole terminator, ST for a cut string
char = ctrl.ST_C0
break
if len(payload) < MAX_OSC:
payload += char
elif char == ESC:
# take the '\' of ST, so char holds the whole terminator
char += yield None
char = yield None
try:
listener.handle_osc(int(code or 0), payload, char)

View File

@@ -5,7 +5,9 @@ The trackpad is not handled here. On macOS it never reaches a mousemap binding
continuous scrolling into discrete wheel events; those arrive as ordinary wheel
notches and take the path below.
"""
import sublime
import sublime_plugin
from wcwidth import wcswidth
from .terminal import Terminal
from .utils import settings
@@ -41,6 +43,7 @@ def _event_to_cell(view, terminal, event, clamp=False):
wheel and a release still belong to the app there, a press does not."""
pt = view.window_to_text((event["x"], event["y"]))
row, col = view.rowcol(pt)
col = wcswidth(view.substr(sublime.Region(pt - col, pt)))
row -= terminal.offset
screen = terminal.screen
if row < 0 or row >= screen.lines:

View File

@@ -1,11 +1,13 @@
""" Rendering the emulator screen into the Sublime view
The view holds real text: scrollback above `terminal.offset`, the live screen
window from `offset` on. Dirty rows are whole-line replaced; colors are
add_regions overlays with scopes served by the dynamic scheme. The ST caret
window from `offset` on. Dirty rows are whole-line replaced, and rows the view
does not hold yet are appended in one insert; colors are add_regions overlays
with scopes served by the dynamic scheme. The ST caret
doubles as the terminal cursor.
"""
import math
from bisect import bisect_left
from functools import lru_cache
import sublime
@@ -88,8 +90,10 @@ def segment_buffer_line(buffer_line):
# 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
data = char.data
run.append(data or " ")
# a cell can hold more than one code point: a mark NFC could not compose
column += len(data) or 1
if attrs is not None:
yield _run(run, start, column, attrs)
@@ -304,23 +308,33 @@ class VtermRenderCommand(sublime_plugin.TextCommand):
# valid for this method only; the trim paths erase rows
self._lastrow = self.view.rowcol(self.view.size())[0]
if dirty_lines:
# history pops newest-first, so it fills the rows below `offset` backwards
history = screen.history
popped = len(history)
terminal.offset += popped
offset = terminal.offset
first = offset - popped
retired = list(history)
history.clear()
for line in range(popped):
buffer_line = history.pop()
self.update_line(edit, offset - line - 1, buffer_line, columns)
# rows the view already holds are rewritten one by one; the rest
# are new and go in as one insert, after those edits moved the end
split = min(popped, self._lastrow + 1 - first)
for i in range(split):
self.update_line(edit, first + i, retired[i], columns)
if split < popped:
self.append_rows(edit, retired[split:], columns, scrollback=True)
# after the loop: a pop edits an earlier point than the one before
# it, so spans measured mid-loop would be stale
self.archive_rows(range(offset - popped, offset))
# after the edits: spans measured mid-loop would be stale
self.archive_rows(range(first, offset))
for line in dirty_lines:
buffer_line = screen.buffer[line]
self.update_line(edit, line + offset, buffer_line, columns)
# screen rows past the view's end go in as one insert too
start = self._lastrow + 1 - offset
cut = bisect_left(dirty_lines, start)
for line in dirty_lines[:cut]:
self.update_line(edit, line + offset, screen.buffer[line], columns)
if cut < len(dirty_lines):
rows = range(start, dirty_lines[-1] + 1)
self.append_rows(edit, [screen.buffer[y] for y in rows], columns, scrollback=False)
def update_line(self, edit, line, buffer_line, columns):
"""Rewrite one view row from a screen row, unless it is unchanged"""
@@ -333,7 +347,17 @@ class VtermRenderCommand(sublime_plugin.TextCommand):
if self.state.row_cache.get(line) == rendered:
return
self.state.row_cache[line] = rendered
text, by_scope = self.row_text(segments, lf)
self.ensure_row(edit, line)
line_region = view.line(view.text_point(line, 0))
self.decolorize_line(line)
view.replace(edit, line_region, text)
# replacing a line cannot move its start
self.colorize_line(line, line_region.begin(), by_scope)
def row_text(self, segments, lf):
"""The view text of a row and its colour spans by scope, as char offsets."""
by_scope = {}
needed = 0
scope_of = self._scheme.scope
@@ -350,13 +374,54 @@ class VtermRenderCommand(sublime_plugin.TextCommand):
if len(text) < needed:
# pad only to the last coloured run, in the same edit as the text
text += " " * (needed - len(text))
return text, by_scope
self.ensure_row(edit, line)
line_region = view.line(view.text_point(line, 0))
self.decolorize_line(line)
view.replace(edit, line_region, text)
# replacing a line cannot move its start
self.colorize_line(line, line_region.begin(), by_scope)
def append_rows(self, edit, buffer_lines, columns, scrollback):
"""Add rows below the view's end in one insert.
Scrollback rows are never rewritten, so their colours go straight onto
the chunk keys with no per-row key for flush_chunk to merge. Screen
rows are coloured per row as usual; a trailing blank run the trim
already erased is still cached as rendered and is left out.
"""
view = self.view
rows = [(line, list(segment_buffer_line(buffer_line)), buffer_line[columns - 1].linefeed)
for line, buffer_line in enumerate(buffer_lines, self._lastrow + 1)]
if not scrollback:
while rows and self.state.row_cache.get(rows[-1][0]) == rows[-1][1:]:
rows.pop()
if not rows:
return
point = view.size() + 1
texts = []
# scrollback: (chunk, scope) -> regions; screen: (line, start, by_scope)
painted = {}
colored = []
trimmed = self.state.trimmed
for line, segments, lf in rows:
self.state.row_cache[line] = (segments, lf)
text, by_scope = self.row_text(segments, lf)
texts.append(text)
if scrollback:
chunk = (line + trimmed) // CHUNK
for scope, spans in by_scope.items():
painted.setdefault((chunk, scope), []).extend(sublime.Region(point + a, point + b) for a, b in spans)
elif by_scope:
colored.append((line, point, by_scope))
point += len(text) + 1
view.insert(edit, view.size(), "\n" + "\n".join(texts))
self._lastrow = rows[-1][0]
for (chunk, scope), regions in painted.items():
key = chunk_key(chunk, scope)
scopes = self.state.archived_chunks.setdefault(chunk, set())
if scope in scopes:
regions = view.get_regions(key) + regions
view.add_regions(key, regions, scope)
scopes.add(scope)
for line, start, by_scope in colored:
self.colorize_line(line, start, by_scope)
def colorize_line(self, line, line_start, by_scope):
"""Lay the colour overlays for one row, one add_regions per scope."""

View File

@@ -2,6 +2,7 @@
import codecs
import logging
import os
import re
import threading
import time
from collections import deque
@@ -40,6 +41,10 @@ IDLE_TICK = 0.2
# a millisecond apart, and drawing between them flashes its showcmd echo
SETTLE = 0.005
# C0 controls bar tab/LF/CR, stripped from text sent as typed: a pasted
# ESC[201~ would end the bracket early and type the rest into the app
CONTROL_CHARS = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
# frames between undo-stack clears: every render edit lands on the view's undo
# stack, which only focus changes clear otherwise, and undo is swallowed anyway
UNDO_CLEAR_FRAMES = 100
@@ -394,7 +399,7 @@ class Terminal:
# escape sequences go through untouched; typed and pasted text has its
# line endings canonicalised for the app's LNM setting
if normalized:
string = string.replace("\r\n", "\n")
string = CONTROL_CHARS.sub("", string).replace("\r\n", "\n")
if mo.LNM in self.screen.mode:
string = string.replace("\n", "\r\n")
else: