fix(tui): survive private-mode CSI sequences in adapter logs

CHANGELOG.md

@@ -46,6 +46,13 @@ breaking changes may land in a minor release.
 
 ### Fixed
 
+- **The TUI no longer crashes on a private-mode CSI sequence in an adapter log.** The gemini
+  CLI's startup burst includes XTMODKEYS `CSI > 4 ; ? m`; the marker byte sat _inside_ the
+  params, so the private-marker strip filter missed it and pyte 0.8.2 raised a `TypeError`
+  that killed the poll worker — and the whole dashboard. The filter now matches a marker
+  anywhere in the params, and any escape sequence pyte still can't parse is dropped instead
+  of propagating (upstream fix exists but was never released — selectel/pyte#202) (#111).
+
 - **An unreadable spec no longer crashes the whole run.** Every spec read-back — the four verify
   gates, the reconcile/sprint/ledger bookkeeping passes, and the generic adapter's Stop poll — raced
   the dev skill's own writes, so a transient `OSError` (a TOCTOU truncation, a lock, an EACCES)

src/bmad_loop/tui/data.py

@@ -354,13 +354,18 @@ def _render_row(row: dict) -> Text:
 
 
 # CSI sequences with a private/secondary marker (< > = ?) are terminal capability
-# negotiation, never display SGR. pyte 0.8.2 ignores the marker and misdispatches
-# them to SGR anyway: e.g. XTMODKEYS `CSI > 4 ; 2 m` (modifyOtherKeys, emitted at
-# session start by Claude Code et al.) is read as SGR 4 = underline-on, leaving the
-# whole log underlined until an exit-time disable a live capture never contains.
-# Strip them before pyte sees them; a legitimate SGR carries no marker, so this can
-# only remove non-display sequences (never printable text or genuine styling).
-_PRIVATE_MARKER_SGR = re.compile(rb"\x1b\[[<>=?][0-9;:]*m")
+# negotiation, never display SGR. pyte 0.8.2 ignores a leading marker and
+# misdispatches them to SGR anyway: e.g. XTMODKEYS `CSI > 4 ; 2 m` (modifyOtherKeys,
+# emitted at session start by Claude Code et al.) is read as SGR 4 = underline-on,
+# leaving the whole log underlined until an exit-time disable a live capture never
+# contains. Worse, a marker byte *inside* the params — gemini's `CSI > 4 ; ? m`,
+# vim9's `CSI ? 4 m` — makes pyte dispatch with private=True to a handler that
+# rejects the kwarg, a TypeError that would kill the poll worker (#111; upstream
+# selectel/pyte#202, fixed in master but never released). Strip them before pyte
+# sees them, matching the marker anywhere in the param bytes; a legitimate SGR
+# carries no marker, so this can only remove non-display sequences (never printable
+# text or genuine styling).
+_PRIVATE_MARKER_SGR = re.compile(rb"\x1b\[[0-9;:<>=?]*[<>=?][0-9;:<>=?]*m")
 # Alternate-screen switch sequences (DECSET/DECRST 1049/1047/47). A CLI fullscreen
 # TUI (Claude Code's fullscreen renderer) switches here and repaints in place; pyte
 # has no altscreen buffer, so the capture collapses to the final frame. Detecting
@@ -400,6 +405,21 @@ def _strip_private_marker_sgr(chunk: bytes) -> tuple[bytes, int]:
     return _PRIVATE_MARKER_SGR.sub(b"", body), held
 
 
+class _TolerantByteStream(pyte.ByteStream):
+    """pyte 0.8.2 dispatches a private-marked CSI (marker byte in the params) with
+    private=True to handlers that don't accept it, raising TypeError (upstream
+    selectel/pyte#202; fixed in master Sep 2025, never released). Any parser
+    exception here would kill the poll worker — and with it the app — so drop the
+    unparseable sequence instead: the base method re-initializes the parser before
+    re-raising, leaving the stream usable for the bytes that follow."""
+
+    def _send_to_parser(self, data: str) -> bool | None:
+        try:
+            return super()._send_to_parser(data)
+        except Exception:
+            return None
+
+
 class _CountingDeque(deque):
     """history.top replacement that counts rows permanently gone above the
     window: maxlen evictions plus the clear() from pyte's reset()/ED-3.
@@ -501,7 +521,7 @@ def _reset_screen(self) -> None:
         self._screen.history = self._screen.history._replace(
             top=_CountingDeque(maxlen=self._history)
         )
-        self._stream = pyte.ByteStream(self._screen)
+        self._stream = _TolerantByteStream(self._screen)
         self._row_cache.clear()
         self._checkpoints: list[tuple[int, int]] = []
         self._render_base = 0

tests/test_tui_data.py

@@ -542,6 +542,81 @@ def test_log_view_strips_private_marker_sgr_split_across_reads(tmp_path):
     assert not any(style.underline for _, _, style in line.spans)
 
 
+def test_log_view_strips_private_marker_mid_params(tmp_path):
+    # gemini's XTMODKEYS reply `CSI > 4 ; ? m` carries the `?` marker *inside* the
+    # params. Unstripped, pyte 0.8.2 dispatches it with private=True and
+    # select_graphic_rendition rejects the kwarg — the TypeError that killed the
+    # whole TUI in #111.
+    path = tmp_path / "task.log"
+    path.write_bytes(b"\x1b[>4;?mhello world\r\n")
+    view = data.LogView(path)
+    assert view.read_new() is True
+    line = view.render()
+    assert "hello world" in line.plain
+    assert not any(style.underline for _, _, style in line.spans)
+
+
+def test_log_view_survives_gemini_startup_preamble(tmp_path):
+    # The exact byte prefix from the #111 traceback: the gemini CLI's terminal
+    # capability negotiation burst, including the crashing `CSI > 4 ; ? m`.
+    path = tmp_path / "task.log"
+    path.write_bytes(
+        b"\x1b[8m\x1b[?u\x1b]11;?\x1b\\\x1b[>q\x1b[>4;?m\x1b[c\x1b[2K\r\x1b[0m" b"ready to work\r\n"
+    )
+    view = data.LogView(path)
+    assert view.read_new() is True
+    assert "ready to work" in view.render().plain
+
+
+def test_log_view_strips_vim9_private_sgr(tmp_path):
+    # `CSI ? 4 m` (vim 9+, upstream selectel/pyte#202): marker in first position
+    # but final `m` — must be stripped, not read as underline or crash pyte.
+    path = tmp_path / "task.log"
+    path.write_bytes(b"\x1b[?4mhello\r\n")
+    view = data.LogView(path)
+    assert view.read_new() is True
+    line = view.render()
+    assert "hello" in line.plain
+    assert not any(style.underline for _, _, style in line.spans)
+
+
+def test_log_view_strips_private_marker_mid_params_split_across_reads(tmp_path):
+    # The #111 sequence straddles two reads; the held-back trailing CSI (whose
+    # char class already admits marker bytes anywhere) lets the filter see it
+    # whole on the next read.
+    path = tmp_path / "task.log"
+    path.write_bytes(b"\x1b[>4;?")
+    view = data.LogView(path)
+    view.read_new()
+    with path.open("ab") as f:
+        f.write(b"mhello\r\n")
+    assert view.read_new() is True
+    line = view.render()
+    assert "hello" in line.plain
+    assert not any(style.underline for _, _, style in line.spans)
+
+
+def test_log_view_survives_unfilterable_private_csi(tmp_path):
+    # Belt-and-braces: a private-marked CSI with a non-`m` final passes the strip
+    # filter deliberately (only marker-SGR is stripped) and crashes raw pyte 0.8.2
+    # (`cursor_position() got an unexpected keyword argument 'private'`). The
+    # tolerant stream drops the sequence instead of killing the poll worker, and
+    # the emulator keeps rendering everything after it.
+    path = tmp_path / "task.log"
+    path.write_bytes(b"\x1b[?1;1Hhello\r\n")
+    view = data.LogView(path)
+    assert view.read_new() is True
+    assert "hello" in view.render().plain
+
+    with path.open("ab") as f:
+        f.write(b"\x1b[31mstill alive\x1b[0m\r\n")
+    assert view.read_new() is True
+    line = view.render()
+    assert "still alive" in line.plain
+    styled = {line.plain[s:e] for s, e, st in line.spans if st.color is not None}
+    assert "still alive" in styled
+
+
 def test_log_view_history_beyond_screen(tmp_path):
     path = tmp_path / "task.log"
     path.write_bytes(b"".join(f"row {i:03d}\r\n".encode() for i in range(1, 81)))