diff --git a/cli.py b/cli.py index 0a9760d1cc646..4c538df43d52f 100644 --- a/cli.py +++ b/cli.py @@ -2574,18 +2574,31 @@ def _luminance_from_hex(hex_str: str) -> float | None: return (0.2126 * r + 0.7152 * g + 0.0722 * b) / 255.0 +_DA1_REPLY_RE = re.compile(rb"\x1b\[\?[0-9;]*c") + + def _query_osc11_background() -> str | None: """Ask the terminal for its background color via OSC 11. - Most modern terminals reply with \\x1b]11;rgb:RRRR/GGGG/BBBB\\x1b\\\\ - within a few ms. We wait up to 100ms total before giving up. - Returns "#RRGGBB" or None on timeout / non-tty. + Most modern terminals reply with \x1b]11;rgb:RRRR/GGGG/BBBB\x1b\\ + within a few ms. Returns "#RRGGBB" or None on timeout / non-tty. - Skipped over SSH: the round-trip routinely exceeds our 100ms budget, so a + The OSC 11 query is fenced with a DA1 sentinel (\x1b[c) — the same + pattern the Ink TUI's TerminalQuerier uses. Terminals answer queries + in order and virtually every terminal answers DA1, so seeing the DA1 + reply proves the terminal already ignored our OSC 11 (multiplexers + like herdr answer DA1 in <1ms while swallowing OSC 11). Without the + fence we can only wait out a blind timeout, and a reply that arrives + AFTER we stop listening leaks into prompt_toolkit's stdin as typed + text — the "gibberish ANSI characters" seen inside terminal managers + that relay color queries slowly (herdr, WSL bridges, some tmux + setups). + + Skipped over SSH: the round-trip routinely exceeds our budget, so a late reply lands after prompt_toolkit has grabbed the tty — its payload leaks in as typed text and the BEL terminator reads as Ctrl+G (open - editor), trapping the user in a stray editor. Remote sessions fall back to - COLORFGBG / env hints / the dark default instead. + editor), trapping the user in a stray editor. Remote sessions fall back + to COLORFGBG / env hints / the dark default instead. After the main read + TCSAFLUSH, a short drain window (50 ms) catches late-arriving bytes that slipped past the flush — a race observed on VPS @@ -2608,13 +2621,26 @@ def _query_osc11_background() -> str | None: except Exception: return None try: - sys.stdout.write("\x1b]11;?\x1b\\") + # OSC 11 query + DA1 sentinel fence, in one write so no + # reordering is possible. + sys.stdout.write("\x1b]11;?\x1b\\\x1b[c") sys.stdout.flush() except Exception: return None - # Read up to ~50ms for the response + # Read until the DA1 fence closes — proof the terminal has processed + # everything up to and including our OSC 11, so nothing can arrive + # late and leak into prompt_toolkit's stdin. DA1 is answered by + # effectively every terminal ever made (it predates color), and on + # real terminals the fence closes in single-digit milliseconds + # (herdr: <1ms, xterm/kitty/tmux: <5ms). The 1s deadline is a + # safety net for a hypothetical terminal that ignores DA1 — not a + # window we ever expect to wait out. A slow in-order relay that + # delivers the OSC 11 reply at e.g. 400ms is handled correctly: + # we keep listening until its DA1 reply follows, so the payload is + # consumed here instead of leaking as typed input (the "gibberish + # ANSI characters" seen inside terminal managers). import select - deadline = time.monotonic() + 0.1 + deadline = time.monotonic() + 1.0 buf = b"" while time.monotonic() < deadline: r, _, _ = select.select([fd], [], [], deadline - time.monotonic()) @@ -2627,7 +2653,7 @@ def _query_osc11_background() -> str | None: if not chunk: break buf += chunk - if b"\x1b\\" in buf or b"\x07" in buf: + if _DA1_REPLY_RE.search(buf): break # Parse: \x1b]11;rgb:RRRR/GGGG/BBBB\x1b\\ m = re.search(rb"rgb:([0-9a-fA-F]+)/([0-9a-fA-F]+)/([0-9a-fA-F]+)", buf) diff --git a/tests/cli/test_cli_light_mode.py b/tests/cli/test_cli_light_mode.py index e37f0f20b78d3..0763cc4ff67cb 100644 --- a/tests/cli/test_cli_light_mode.py +++ b/tests/cli/test_cli_light_mode.py @@ -180,14 +180,17 @@ class TestOsc11DrainGuard: """Regression: a late-arriving OSC 11 reply must not leak into prompt_toolkit's input buffer (#40250). - The drain loop in the ``finally`` block of ``_query_osc11_background`` - reads (and discards) any bytes that arrive after TCSAFLUSH completes. + Two layers guard against this: the DA1 fence keeps the main read loop + listening until the terminal proves it has processed our query, and + the drain loop in the ``finally`` block reads (and discards) any + stragglers that slip past TCSAFLUSH. """ - def test_finally_drain_discards_late_bytes(self, cli_mod, monkeypatch): - """Simulate a terminal that sends the OSC 11 reply after the main - read loop's deadline — the drain window must eat it.""" - import io, os, termios, tty as _tty + def test_late_reply_is_consumed_not_leaked(self, cli_mod, monkeypatch): + """Simulate a terminal that sends the OSC 11 reply 150ms after the + query. With the DA1 fence the main loop is still listening, so the + reply is consumed AND used; nothing remains for prompt_toolkit.""" + import os, termios, tty as _tty # Create a pipe pair to fake stdin read_fd, write_fd = os.pipe() @@ -208,28 +211,204 @@ class TestOsc11DrainGuard: for v in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY"): monkeypatch.delenv(v, raising=False) - # Write a delayed OSC 11 reply — the main select() loop will time out - # (nothing in the pipe during the 100ms window), then the drain loop - # should read and discard it. + # Write a delayed OSC 11 reply (then the DA1 fence reply) — the + # fenced main loop must still be listening and consume both. import threading def delayed_write(): import time - time.sleep(0.15) # after the 100ms main deadline - os.write(write_fd, b"\x1b]11;rgb:0c0c/0c0c/0c0c\x1b\\") + time.sleep(0.15) + os.write(write_fd, b"\x1b]11;rgb:0c0c/0c0c/0c0c\x1b\\\x1b[?62;22c") t = threading.Thread(target=delayed_write, daemon=True) t.start() - # The function should return None (no valid response during main window) - # and the drain loop should eat the late bytes. + # The late reply is consumed by the fenced read loop and used. result = cli_mod._query_osc11_background() - assert result is None + assert result == "#0C0C0C" # Verify the pipe is drained — a non-blocking read should return empty import select r, _, _ = select.select([read_fd], [], [], 0) - assert not r, "drain loop should have consumed late OSC 11 bytes" + assert not r, "late OSC 11 bytes must be consumed, not left to leak" os.close(read_fd) os.close(write_fd) + + def test_post_deadline_straggler_is_drained(self, cli_mod, monkeypatch): + """Bytes that arrive after the main loop has already finished (DA1 + answered instantly, reply straggles in during teardown) are eaten + by the post-flush drain window instead of leaking (#40250).""" + import os, termios, tty as _tty + + read_fd, write_fd = os.pipe() + fake_attrs = [0, 0, 0, 0, 0, 0, [b'\x00'] * 32] + monkeypatch.setattr(termios, "tcgetattr", lambda fd: fake_attrs) + monkeypatch.setattr(termios, "tcsetattr", lambda fd, when, attrs: None) + monkeypatch.setattr(_tty, "setcbreak", lambda fd: None) + monkeypatch.setattr(cli_mod.sys.stdin, "isatty", lambda: True, raising=False) + monkeypatch.setattr(cli_mod.sys.stdout, "isatty", lambda: True, raising=False) + monkeypatch.setattr(cli_mod.sys.stdin, "fileno", lambda: read_fd, raising=False) + for v in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY"): + monkeypatch.delenv(v, raising=False) + + # DA1 answered immediately (herdr-style: OSC 11 swallowed) — main + # loop exits fast — then a straggler payload lands during teardown. + os.write(write_fd, b"\x1b[?62;22c") + + import threading + + def straggler(): + import time + time.sleep(0.02) # inside the 50ms drain window + os.write(write_fd, b"\x1b]11;rgb:0c0c/0c0c/0c0c\x1b\\") + + t = threading.Thread(target=straggler, daemon=True) + t.start() + + result = cli_mod._query_osc11_background() + assert result is None # OSC 11 was swallowed; only DA1 answered + + import select + r, _, _ = select.select([read_fd], [], [], 0) + assert not r, "drain loop should have consumed the straggler bytes" + + os.close(read_fd) + os.close(write_fd) + + +# ──────────────────────────────────────────────────────────────────────── +# OSC 11 query — DA1 fence behavior. +# +# The query is fenced with a DA1 sentinel so a terminal manager that +# swallows OSC 11 (herdr) or relays it slowly (SSH bridges, some tmux +# setups) can never leave reply bytes in the tty buffer for +# prompt_toolkit to read as typed input. These tests run the real +# function in a child on a real PTY and play the terminal's role from +# the parent side. + +import os as _os +import sys as _sys + + +_CHILD_SRC = r""" +import sys, os +sys.path.insert(0, os.environ["HERMES_REPO"]) +import cli +bg = cli._query_osc11_background() +print("RESULT:" + repr(bg), flush=True) +# Drain anything left in the tty buffer — must be empty (no leak). +import termios, tty, select, time +fd = sys.stdin.fileno() +old = termios.tcgetattr(fd) +tty.setcbreak(fd) +buf = b"" +deadline = time.monotonic() + 0.6 +while time.monotonic() < deadline: + r, _, _ = select.select([fd], [], [], 0.1) + if r: + buf += os.read(fd, 256) +termios.tcsetattr(fd, termios.TCSADRAIN, old) +print("LEFTOVER:" + repr(buf), flush=True) +""" + + +def _run_osc11_child(reply_fn, repo_root, timeout=8.0): + """Fork a PTY child running _query_osc11_background(). + + reply_fn(query_age_seconds) -> bytes to write once, or None to wait. + Returns (result_line, leftover_line). + """ + import pty + import time as _time + + env = dict(_os.environ, HERMES_REPO=str(repo_root)) + for var in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY"): + env.pop(var, None) + pid, master = pty.fork() + if pid == 0: # child + _os.execvpe(_sys.executable, [_sys.executable, "-c", _CHILD_SRC], env) + _os.set_blocking(master, False) + out = b"" + answered = 0 + query_at = None + t0 = _time.monotonic() + try: + while _time.monotonic() - t0 < timeout: + try: + chunk = _os.read(master, 1024) + if chunk: + out += chunk + except (BlockingIOError, OSError): + pass + # importing cli primes _detect_light_mode() which issues its own + # OSC 11 query before the explicit call — answer each query. + if query_at is None and out.count(b"\x1b]11;?") > answered: + query_at = _time.monotonic() + if query_at is not None: + payload = reply_fn(_time.monotonic() - query_at) + if payload is not None: + _os.write(master, payload) + answered += 1 + query_at = None + if b"LEFTOVER:" in out and out.rstrip().endswith(b"'"): + break + _time.sleep(0.005) + finally: + try: + _os.close(master) + except OSError: + pass + try: + _os.waitpid(pid, 0) + except ChildProcessError: + pass + text = out.decode("utf-8", "replace") + print("child raw output:", repr(text)) # aids debugging on failure + result = leftover = None + for line in text.splitlines(): + # The PTY echoes the query bytes onto the same line as the first + # print, so match anywhere in the line rather than at the start. + if "RESULT:" in line and result is None: + result = line.split("RESULT:", 1)[1] + elif "LEFTOVER:" in line and leftover is None: + leftover = line.split("LEFTOVER:", 1)[1] + return result, leftover + + +@pytest.fixture +def repo_root(): + import pathlib + return pathlib.Path(__file__).resolve().parents[2] + + +@pytest.mark.skipif(_sys.platform == "win32", reason="POSIX PTY test") +class TestOsc11Da1Fence: + def test_herdr_style_da1_only_returns_none_without_leak(self, repo_root): + """Terminal answers DA1 instantly but swallows OSC 11 (herdr).""" + result, leftover = _run_osc11_child( + lambda age: b"\x1b[?62;22c", repo_root + ) + assert result == "None" + assert leftover == "b''" + + def test_slow_inorder_reply_is_consumed_not_leaked(self, repo_root): + """OSC 11 reply arrives at +300ms (past the old 100ms budget), + DA1 right behind it. The fence keeps us listening, so the color + is detected and nothing leaks into the tty buffer.""" + result, leftover = _run_osc11_child( + lambda age: ( + b"\x1b]11;rgb:1e1e/1e1e/2e2e\x1b\\\x1b[?62;22c" + if age > 0.3 else None + ), + repo_root, + ) + assert result == "'#1E1E2E'" + assert leftover == "b''" + + def test_mute_terminal_times_out_clean(self, repo_root): + """Terminal that answers nothing: give up at the safety-net + deadline with no leftovers.""" + result, leftover = _run_osc11_child(lambda age: None, repo_root) + assert result == "None" + assert leftover == "b''"