fix: expand Ink partial clears to prevent ghost content in screenshots
Ink (React CLI framework) clears its output using repeated EL2+CUU1 sequences, one per previously-drawn line. When /clear resets Ink's internal line counter, the next frame only erases a few lines instead of the full previous output. In a real terminal the old content is in scrollback and invisible, but pyte's fixed-size screen retains it, producing ghost content (e.g. duplicated prompts) in SVG screenshots. Added AltScreen.expand_clear_sequences() which detects runs of 3+ EL2+CUU1 pairs that don't reach row 0 and extends them to erase all lines up to the top of the screen. Both DockerExecSession and TerminalSession call this before feeding data to pyte. Also made on_session_end() idempotent (contextlib.suppress KeyError) to prevent a race when close_session() and natural session exit both call it. Added docs/ink-clear-fix.md with root cause analysis, byte-level explanation, and reproduction script.
This commit is contained in:
@@ -129,3 +129,80 @@ class TestAltScreen:
|
||||
stream.feed("\x1b[2J") # ED 2 - erase entire display
|
||||
|
||||
assert all(line.strip() == "" for line in screen.display)
|
||||
|
||||
|
||||
class TestExpandClearSequences:
|
||||
"""Tests for expand_clear_sequences (Ink partial clear fix)."""
|
||||
|
||||
def test_no_clear_sequences(self):
|
||||
"""Data without EL2+CUU1 runs is returned unchanged."""
|
||||
screen = AltScreen(80, 24)
|
||||
data = b"Hello world\r\n"
|
||||
assert screen.expand_clear_sequences(data) == data
|
||||
|
||||
def test_short_clear_not_expanded(self):
|
||||
"""Runs of fewer than 3 EL2+CUU1 pairs are not modified."""
|
||||
screen = AltScreen(80, 24)
|
||||
data = b"\x1b[2K\x1b[1A\x1b[2K\x1b[1A" # 2 pairs
|
||||
assert screen.expand_clear_sequences(data) == data
|
||||
|
||||
def test_full_clear_not_expanded(self):
|
||||
"""A clear that already reaches row 0 is not extended."""
|
||||
screen = AltScreen(80, 24)
|
||||
stream = pyte.ByteStream(screen)
|
||||
# Put cursor at row 5
|
||||
stream.feed(b"\r\n" * 5)
|
||||
assert screen.cursor.y == 5
|
||||
|
||||
# 5-pair clear already covers rows 5 down to 0
|
||||
data = b"\x1b[2K\x1b[1A" * 5
|
||||
result = screen.expand_clear_sequences(data)
|
||||
assert result == data
|
||||
|
||||
def test_partial_clear_is_extended(self):
|
||||
"""A partial clear that doesn't reach row 0 gets extended."""
|
||||
screen = AltScreen(80, 24)
|
||||
stream = pyte.ByteStream(screen)
|
||||
# Draw content to push cursor to row 20
|
||||
for i in range(20):
|
||||
stream.feed(f"Line {i}\r\n".encode())
|
||||
assert screen.cursor.y == 20
|
||||
|
||||
# Only clear 5 lines (should extend to clear all 20)
|
||||
data = b"\x1b[2K\x1b[1A" * 5
|
||||
result = screen.expand_clear_sequences(data)
|
||||
expected_pairs = 20 # extend from 5 to 20
|
||||
assert result.count(b"\x1b[2K\x1b[1A") == expected_pairs
|
||||
|
||||
def test_partial_clear_produces_correct_screen(self):
|
||||
"""Simulates Ink /clear: partial clear + redraw leaves clean screen."""
|
||||
screen = AltScreen(80, 24)
|
||||
stream = pyte.ByteStream(screen)
|
||||
|
||||
# Draw 15 lines of content (Ink frame 1)
|
||||
for i in range(15):
|
||||
stream.feed(f"Old line {i}\r\n".encode())
|
||||
|
||||
# Ink /clear: only clears 5 lines then redraws fresh prompt
|
||||
clear = b"\x1b[2K\x1b[1A" * 5 + b"\x1b[2K\x1b[G"
|
||||
new_content = b"Fresh prompt\r\n"
|
||||
|
||||
expanded = screen.expand_clear_sequences(clear)
|
||||
stream.feed(expanded)
|
||||
stream.feed(new_content)
|
||||
|
||||
# Old content should be gone
|
||||
non_empty = [line.rstrip() for line in screen.display if line.strip()]
|
||||
assert len(non_empty) == 1
|
||||
assert non_empty[0] == "Fresh prompt"
|
||||
|
||||
def test_data_around_clear_preserved(self):
|
||||
"""Text before and after a clear run is preserved."""
|
||||
screen = AltScreen(80, 24)
|
||||
stream = pyte.ByteStream(screen)
|
||||
stream.feed(b"\r\n" * 10)
|
||||
|
||||
data = b"before\x1b[2K\x1b[1A" * 0 + b"before" + b"\x1b[2K\x1b[1A" * 5 + b"after"
|
||||
result = screen.expand_clear_sequences(data)
|
||||
assert result.startswith(b"before")
|
||||
assert result.endswith(b"after")
|
||||
|
||||
@@ -105,6 +105,21 @@ class TestSessionManager:
|
||||
assert session_id not in manager.sessions
|
||||
assert route_key not in manager.routes
|
||||
|
||||
def test_on_session_end_idempotent(self, mock_poller, mock_path, sample_apps):
|
||||
"""Test session end cleanup is idempotent."""
|
||||
manager = SessionManager(mock_poller, mock_path, sample_apps)
|
||||
|
||||
session_id = SessionID("test-session")
|
||||
route_key = RouteKey("test-route")
|
||||
manager.sessions[session_id] = MagicMock()
|
||||
manager.routes[route_key] = session_id
|
||||
|
||||
manager.on_session_end(session_id)
|
||||
manager.on_session_end(session_id)
|
||||
|
||||
assert session_id not in manager.sessions
|
||||
assert route_key not in manager.routes
|
||||
|
||||
def test_on_session_end_nonexistent(self, mock_poller, mock_path, sample_apps):
|
||||
"""Test session end for non-existent session."""
|
||||
manager = SessionManager(mock_poller, mock_path, sample_apps)
|
||||
|
||||
Reference in New Issue
Block a user