fix: route deferred update notice through cprint to avoid ANSI garbling
The deferred update notice ("N commits behind") was rendered via
console.print which writes ANSI escapes directly to stdout. Under
patch_stdout (active in the interactive CLI), the StdoutProxy strips
ESC bytes, leaving visible [1;33m...[0m artifacts.
Fix: render Rich markup to an ANSI string via a captured Console, then
route through cprint (which uses prompt_toolkit ANSI parser) to
bypass the StdoutProxy mangling.
This is the same class of bug as #2262 (fixed by #2448 for agent._print_fn
and display.py), but the _defer_update_notice callsite in banner.py was
not covered by that fix.
Fixes #83969
This commit is contained in:
+17
-4
@@ -491,11 +491,25 @@ def _format_update_notice(behind: int) -> str:
|
||||
_deferred_update_notice_started = False
|
||||
|
||||
|
||||
def _render_markup_to_ansi(markup: str) -> str:
|
||||
"""Rich markup → ANSI string, for output that must go through prompt_toolkit's renderer.
|
||||
|
||||
Under ``patch_stdout`` (the interactive CLI), a plain ``Console.print`` writes ESC bytes into
|
||||
the StdoutProxy, which sanitizes them into visible ``?[1;33m…`` artifacts (#83969).
|
||||
"""
|
||||
from io import StringIO
|
||||
from rich.console import Console as _Console
|
||||
buf = StringIO()
|
||||
_Console(file=buf, force_terminal=True, color_system="truecolor", highlight=False).print(markup)
|
||||
return buf.getvalue().rstrip("\n")
|
||||
|
||||
|
||||
def _defer_update_notice(console: "Console", max_wait: float = 30.0) -> None:
|
||||
"""Print the update warning once the prefetched check completes (at most once per process).
|
||||
|
||||
Used when the banner rendered before the update prefetch finished so startup never blocks on
|
||||
git/network.
|
||||
git/network. The notice lands after prompt_toolkit owns the terminal, so it is routed through
|
||||
``cprint`` (prompt_toolkit's renderer prints above a running application from any thread).
|
||||
"""
|
||||
global _deferred_update_notice_started
|
||||
if _deferred_update_notice_started:
|
||||
@@ -504,7 +518,7 @@ def _defer_update_notice(console: "Console", max_wait: float = 30.0) -> None:
|
||||
|
||||
def _wait_and_print() -> None:
|
||||
if _update_check_done.wait(timeout=max_wait) and _update_result:
|
||||
console.print(_format_update_notice(_update_result))
|
||||
cprint(_render_markup_to_ansi(_format_update_notice(_update_result)))
|
||||
_daemon("update-notice", _wait_and_print) # never break the session over an update notice
|
||||
|
||||
|
||||
@@ -863,8 +877,7 @@ def build_welcome_banner(
|
||||
right_lines.append(f"[dim {dim}]{' · '.join(summary_parts)}[/]")
|
||||
# Update check — NEVER block the banner on it: the prefetch does git/network work that rarely
|
||||
# finishes before render, so a blocking wait adds its full timeout to every startup. If not
|
||||
# ready, a daemon thread prints the same notice above the prompt when it lands
|
||||
# (prompt_toolkit's patch_stdout renders late prints safely).
|
||||
# ready, a daemon thread prints the same notice above the prompt when it lands.
|
||||
def _update_line():
|
||||
behind = get_update_result(timeout=0.05)
|
||||
if behind is None and not _update_check_done.is_set():
|
||||
|
||||
@@ -175,37 +175,41 @@ class TestBannerUpdateCheckNonBlocking:
|
||||
|
||||
printed = []
|
||||
|
||||
class _Console:
|
||||
def print(self, msg, *a, **k):
|
||||
printed.append(msg)
|
||||
def _fake_cprint(text):
|
||||
printed.append(text)
|
||||
|
||||
done = threading.Event()
|
||||
with patch.object(banner, "_update_check_done", done), \
|
||||
patch.object(banner, "_update_result", None), \
|
||||
patch.object(banner, "_deferred_update_notice_started", False):
|
||||
banner._defer_update_notice(_Console(), max_wait=5.0)
|
||||
patch.object(banner, "_deferred_update_notice_started", False), \
|
||||
patch.object(banner, "cprint", _fake_cprint):
|
||||
banner._defer_update_notice(None, max_wait=5.0)
|
||||
banner._update_result = 3
|
||||
done.set()
|
||||
deadline = time.time() + 5
|
||||
while not printed and time.time() < deadline:
|
||||
time.sleep(0.02)
|
||||
assert printed, "deferred update notice never printed"
|
||||
assert "3 commits behind" in printed[0]
|
||||
# cprint receives ANSI-rendered text (ESC escapes + visible text),
|
||||
# so check that the visible payload contains the expected message.
|
||||
import re
|
||||
visible = re.sub(r"\x1b\[[0-9;]*m", "", printed[0])
|
||||
assert "3 commits behind" in visible
|
||||
|
||||
def test_deferred_notice_silent_when_up_to_date(self):
|
||||
import hermes_cli.banner as banner
|
||||
|
||||
printed = []
|
||||
|
||||
class _Console:
|
||||
def print(self, msg, *a, **k):
|
||||
printed.append(msg)
|
||||
def _fake_cprint(text):
|
||||
printed.append(text)
|
||||
|
||||
done = threading.Event()
|
||||
with patch.object(banner, "_update_check_done", done), \
|
||||
patch.object(banner, "_update_result", 0), \
|
||||
patch.object(banner, "_deferred_update_notice_started", False):
|
||||
banner._defer_update_notice(_Console(), max_wait=2.0)
|
||||
patch.object(banner, "_deferred_update_notice_started", False), \
|
||||
patch.object(banner, "cprint", _fake_cprint):
|
||||
banner._defer_update_notice(None, max_wait=2.0)
|
||||
done.set()
|
||||
time.sleep(0.3)
|
||||
assert not printed
|
||||
|
||||
Reference in New Issue
Block a user