diff --git a/hermes_cli/banner.py b/hermes_cli/banner.py index 62cab7fc26..1a1419c8b0 100644 --- a/hermes_cli/banner.py +++ b/hermes_cli/banner.py @@ -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(): diff --git a/tests/tools/test_startup_latency_regressions.py b/tests/tools/test_startup_latency_regressions.py index c9fb3701ec..77c7c41210 100644 --- a/tests/tools/test_startup_latency_regressions.py +++ b/tests/tools/test_startup_latency_regressions.py @@ -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