fix: don't crash printing emoji on a legacy Windows code page (FLYTE-SDK-7M)

src/flyte/cli/_common.py

@@ -476,6 +476,7 @@ def print_url(console: Console, url: str, prefix: str = "➡️  ", of: OutputFo
     """
     Print a URL on a line of its own, soft-wrapped so it stays clickable and copyable.
     """
+    prefix = safe_text(prefix)
     if of in ["table-simple", "json", "json-raw"]:
         console.print(f"{prefix}{url}", highlight=False, soft_wrap=True)
         return
@@ -489,24 +490,74 @@ def get_console() -> Console:
     return Console(color_system="auto", force_terminal=True)
 
 
+def _stdout_can_encode(probe: str) -> bool:
+    """
+    Report whether stdout's encoding can represent `probe`.
+
+    Anything it can't represent raises UnicodeEncodeError at write time, deep inside
+    Rich's renderer, which crashes the command rather than degrading the output.
+    """
+    import sys
+
+    encoding = getattr(sys.stdout, "encoding", None) or ""
+    if encoding.lower().replace("-", "") in ("utf8", "utf16", "utf32"):
+        return True
+    try:
+        probe.encode(encoding)
+    except (UnicodeEncodeError, LookupError):
+        return False
+    return True
+
+
 def safe_spinner(spinner: str = "dots") -> str:
     """
     Pick an ASCII-safe spinner when stdout encoding can't represent the requested
     spinner's characters (e.g. legacy Windows cp1252 consoles can't encode the
     braille characters used by Rich's default "dots" spinner, which crashes
     mid-render with UnicodeEncodeError).
     """
+    # Probe with a representative non-ASCII char from the "dots" spinner.
+    return spinner if _stdout_can_encode("⠙") else "line"
+
+
+# Plain-ASCII stand-ins for the decorative glyphs the CLI prints. Purely ornamental
+# ones map to the empty string; ones that carry meaning keep an ASCII equivalent.
+_ASCII_FALLBACKS = {
+    "\U0001f680": "",  # rocket
+    "\U0001f433": "",  # whale
+    "\u27a1": "->",  # right arrow
+    "\u26a0": "!",  # warning sign
+    "\u274c": "x",  # cross mark
+    "\u2705": "v",  # check mark button
+    "\u2714": "v",  # heavy check mark
+    "\ufe0f": "",  # variation selector-16, trails several of the above
+}
+
+
+def safe_text(text: str) -> str:
+    """
+    Rewrite `text` so every character survives stdout's encoding.
+
+    Legacy Windows consoles run on a regional code page (cp936, cp1252, ...) that has no
+    room for emoji, so printing one raises UnicodeEncodeError from inside Rich's renderer
+    and takes the whole command down. Known decorative glyphs become their ASCII stand-ins
+    and anything else that still won't encode becomes "?", the same substitution Python's
+    own `errors="replace"` would make.
+
+    Returns `text` unchanged whenever stdout can already encode it, which is every UTF-8
+    terminal, so this only ever degrades output that would otherwise have crashed.
+    """
     import sys
 
-    encoding = getattr(sys.stdout, "encoding", None) or ""
-    if encoding.lower().replace("-", "") in ("utf8", "utf16", "utf32"):
-        return spinner
+    if _stdout_can_encode(text):
+        return text
+    for glyph, replacement in _ASCII_FALLBACKS.items():
+        text = text.replace(glyph, replacement)
+    encoding = getattr(sys.stdout, "encoding", None) or "ascii"
     try:
-        # Probe with a representative non-ASCII char from the "dots" spinner.
-        "⠙".encode(encoding)
-    except (UnicodeEncodeError, LookupError):
-        return "line"
-    return spinner
+        return text.encode(encoding, "replace").decode(encoding, "replace")
+    except LookupError:
+        return text.encode("ascii", "replace").decode("ascii")
 
 
 class _StaticStatus:

src/flyte/cli/_devbox.py

@@ -20,6 +20,8 @@
 
 from flyte import _sentry
 
+from ._common import safe_spinner, safe_text
+
 _CONTAINER_NAME = "flyte-devbox"
 _VOLUME_NAME = "flyte-devbox"
 _KUBE_DIR = Path(
@@ -359,7 +361,7 @@ def _launch_devbox_plain(image_name: str, is_dev_mode: bool, steps: list[tuple[s
 
 def _launch_devbox_rich(image_name: str, is_dev_mode: bool, steps: list[tuple[str, str]], gpu: bool = False) -> None:
     with Progress(
-        SpinnerColumn(),
+        SpinnerColumn(spinner_name=safe_spinner()),
         TextColumn("[progress.description]{task.description}"),
         BarColumn(),
         TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
@@ -378,9 +380,11 @@ def _launch_devbox_rich(image_name: str, is_dev_mode: bool, steps: list[tuple[st
     else:
         console.print(
             Panel(
-                "[green bold]Flyte devbox cluster is ready![/green bold]\n\n"
-                "  🚀 UI:             [link=http://localhost:30080/v2]http://localhost:30080/v2[/link]\n"
-                "  🐳 Image Registry: localhost:30000",
+                safe_text(
+                    "[green bold]Flyte devbox cluster is ready![/green bold]\n\n"
+                    "  🚀 UI:             [link=http://localhost:30080/v2]http://localhost:30080/v2[/link]\n"
+                    "  🐳 Image Registry: localhost:30000"
+                ),
                 title="[bold]Flyte Devbox[/bold]",
                 border_style="green",
             )

src/flyte/cli/_run.py

@@ -529,7 +529,7 @@ def invoke(self, ctx: click.Context):
         if not self.run_args.local and self.run_args.copy_style == "all":
             effective_root_dir = Path(self.run_args.root_dir).resolve() if self.run_args.root_dir else Path.cwd()
             if is_home_directory(effective_root_dir):
-                warning = HOME_DIRECTORY_WARNING.format(path=effective_root_dir)
+                warning = common.safe_text(HOME_DIRECTORY_WARNING.format(path=effective_root_dir))
                 common.get_console().print(f"[yellow]Warning: {warning}[/yellow]")
         self._validate_required_params(ctx)
         if self.run_args.tui:

tests/flyte/cli/test_safe_text.py

@@ -0,0 +1,115 @@
+"""Tests for ASCII-safe console text on non-UTF stdout encodings.
+
+Legacy Windows consoles run on a regional code page with no room for emoji, so a
+decorative glyph reaches Rich's renderer and raises UnicodeEncodeError there, taking
+the command down (FLYTE-SDK-7M, a `flyte start devbox` on a cp936 console).
+"""
+
+import io
+from unittest import mock
+
+import pytest
+from rich.console import Console
+from rich.panel import Panel
+
+from flyte._code_bundle._utils import HOME_DIRECTORY_WARNING
+from flyte.cli._common import print_url, safe_text
+
+DEVBOX_READY_PANEL = (
+    "[green bold]Flyte devbox cluster is ready![/green bold]\n\n"
+    "  \U0001f680 UI:             [link=http://localhost:30080/v2]http://localhost:30080/v2[/link]\n"
+    "  \U0001f433 Image Registry: localhost:30000"
+)
+
+
+def _stdout_with_encoding(encoding: str) -> io.TextIOWrapper:
+    return io.TextIOWrapper(io.BytesIO(), encoding=encoding)
+
+
+def _console_on(encoding: str) -> tuple[Console, io.TextIOWrapper]:
+    """A Console writing to a strict stream on `encoding`, like a legacy Windows console."""
+    stream = io.TextIOWrapper(io.BytesIO(), encoding=encoding, errors="strict")
+    return Console(file=stream, width=100, color_system=None), stream
+
+
+def test_utf8_stdout_is_left_alone():
+    with mock.patch("sys.stdout", _stdout_with_encoding("utf-8")):
+        assert safe_text(DEVBOX_READY_PANEL) == DEVBOX_READY_PANEL
+
+
+def test_ascii_only_text_is_left_alone():
+    with mock.patch("sys.stdout", _stdout_with_encoding("cp936")):
+        assert safe_text("Flyte devbox cluster is ready!") == "Flyte devbox cluster is ready!"
+
+
+@pytest.mark.parametrize("encoding", ["cp936", "cp1252", "ascii"])
+def test_emoji_dropped_on_legacy_code_pages(encoding: str):
+    with mock.patch("sys.stdout", _stdout_with_encoding(encoding)):
+        out = safe_text(DEVBOX_READY_PANEL)
+    assert "\U0001f680" not in out
+    assert "\U0001f433" not in out
+    # Only the ornament goes; the message and its Rich markup survive intact.
+    assert "Flyte devbox cluster is ready!" in out
+    assert "[link=http://localhost:30080/v2]" in out
+    assert "Image Registry: localhost:30000" in out
+    out.encode(encoding)  # would raise if anything unencodable were left
+
+
+def test_meaningful_glyphs_keep_an_ascii_stand_in():
+    with mock.patch("sys.stdout", _stdout_with_encoding("cp936")):
+        assert safe_text("➡️  ") == "->  "
+        assert safe_text("⚠️ warning").startswith("! ")
+
+
+def test_unmapped_unencodable_character_becomes_a_question_mark():
+    # Not in the fallback table: it must still not reach the stream unencoded.
+    with mock.patch("sys.stdout", _stdout_with_encoding("cp1252")):
+        assert safe_text("run 中文 done") == "run ?? done"
+
+
+def test_unknown_encoding_name_degrades_to_ascii():
+    class _BogusEncoding:
+        encoding = "not-a-real-codec"
+
+    with mock.patch("sys.stdout", _BogusEncoding()):
+        assert safe_text("\U0001f680 UI") == " UI"
+
+
+def test_missing_encoding_attribute_degrades_to_ascii():
+    class _NoEncoding:
+        encoding = None
+
+    with mock.patch("sys.stdout", _NoEncoding()):
+        assert safe_text("\U0001f680 UI") == " UI"
+
+
+def test_devbox_ready_panel_renders_on_a_cp936_console():
+    """The FLYTE-SDK-7M crash, end to end: Rich writing the panel to a cp936 stream."""
+    console, stream = _console_on("cp936")
+    with mock.patch("sys.stdout", _stdout_with_encoding("cp936")):
+        body = safe_text(DEVBOX_READY_PANEL)
+    console.print(Panel(body, title="[bold]Flyte Devbox[/bold]", border_style="green"))
+    stream.flush()
+    rendered = stream.buffer.getvalue().decode("cp936")
+    assert "Flyte devbox cluster is ready!" in rendered
+
+
+def test_print_url_prefix_renders_on_a_cp936_console():
+    """`flyte run` prints the run URL behind a default arrow-emoji prefix."""
+    console, stream = _console_on("cp936")
+    with mock.patch("sys.stdout", _stdout_with_encoding("cp936")):
+        print_url(console, "https://example.com/run/abc")
+    stream.flush()
+    rendered = stream.buffer.getvalue().decode("cp936")
+    assert "https://example.com/run/abc" in rendered
+    assert rendered.startswith("->")
+
+
+def test_home_directory_warning_renders_on_a_cp936_console():
+    console, stream = _console_on("cp936")
+    with mock.patch("sys.stdout", _stdout_with_encoding("cp936")):
+        warning = safe_text(HOME_DIRECTORY_WARNING.format(path="/home/u"))
+    console.print(f"[yellow]Warning: {warning}[/yellow]")
+    stream.flush()
+    rendered = stream.buffer.getvalue().decode("cp936")
+    assert "Running from your home directory" in rendered