[Container App] Fix az containerapp exec crash on non-ASCII (emoji) terminal output

src/azure-cli/azure/cli/command_modules/containerapp/_ssh_utils.py

@@ -102,10 +102,34 @@ def recv(self, *args, **kwargs):
         return self._socket.recv(*args, **kwargs)
 
 
+def _write_to_terminal(text):
+    # The terminal's encoding (e.g. cp1252 on Windows) may not be able to
+    # represent every character the container emits (emoji, non-Latin scripts).
+    # On a UTF-8 terminal the fast path prints natively; only when the console
+    # codec cannot encode a character do we fall back to a non-failing policy so
+    # the rest of the output is still shown instead of crashing the exec session.
+    try:
+        print(text, end="", flush=True)
+        return
+    except UnicodeEncodeError:
+        pass
+
+    encoding = getattr(sys.stdout, "encoding", None) or SSH_DEFAULT_ENCODING
+    encoded = text.encode(encoding, errors="backslashreplace")
+    buffer = getattr(sys.stdout, "buffer", None)
+    if buffer is not None:
+        buffer.write(encoded)
+        buffer.flush()
+    else:
+        # Stream has no binary buffer (already wrapped / replaced) -> round-trip
+        # through the same codec so the write itself cannot raise.
+        print(encoded.decode(encoding, errors="backslashreplace"), end="", flush=True)
+
+
 def _decode_and_output_to_terminal(connection: WebSocketConnection, response, encodings):
     for i, encoding in enumerate(encodings):
         try:
-            print(response[2:].decode(encoding), end="", flush=True)
+            decoded = response[2:].decode(encoding)
             break
         except UnicodeDecodeError as e:
             if i == len(encodings) - 1:  # ran out of encodings to try
@@ -114,7 +138,11 @@ def _decode_and_output_to_terminal(connection: WebSocketConnection, response, en
                 logger.info("Cluster Control Byte: %s", response[1])
                 logger.info("Hexdump: %s", response[2:].hex())
                 raise CLIInternalError("Failed to decode server data") from e
-            logger.info("Failed to encode with encoding %s", encoding)
+            logger.info("Failed to decode with encoding %s", encoding)
+    else:
+        return  # empty encodings list: nothing to decode or print
+
+    _write_to_terminal(decoded)
 
 
 def read_ssh(connection: WebSocketConnection, response_encodings):

src/azure-cli/azure/cli/command_modules/containerapp/tests/latest/test_ssh_utils.py

@@ -0,0 +1,71 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import io
+import unittest
+from unittest import mock
+
+from azure.cli.command_modules.containerapp import _ssh_utils
+from azure.cli.command_modules.containerapp._ssh_utils import (
+    _write_to_terminal,
+    _decode_and_output_to_terminal,
+)
+
+# Lobster emoji (U+1F99E): valid UTF-8, but has no representation in cp1252.
+EMOJI = "\U0001f99e"
+
+
+class _NarrowStdout:
+    """Mimics a console whose text codec (e.g. cp1252) cannot encode every
+    character. Its ``write`` re-encodes like a real console, so emoji raises
+    ``UnicodeEncodeError``; bytes written via the fallback land in ``buffer``."""
+
+    def __init__(self, encoding="cp1252"):
+        self.encoding = encoding
+        self.buffer = io.BytesIO()
+
+    def write(self, text):
+        text.encode(self.encoding)  # raises UnicodeEncodeError on unencodable chars
+
+    def flush(self):
+        pass
+
+
+class SshUtilsTerminalOutputTest(unittest.TestCase):
+    def test_write_falls_back_when_char_not_encodable(self):
+        text = f"hello {EMOJI} world"
+        stdout = _NarrowStdout(encoding="cp1252")
+
+        with mock.patch.object(_ssh_utils.sys, "stdout", stdout):
+            _write_to_terminal(text)  # must not raise UnicodeEncodeError
+
+        self.assertEqual(stdout.buffer.getvalue(),
+                         text.encode("cp1252", errors="backslashreplace"))
+
+    def test_write_uses_print_fast_path_when_encodable(self):
+        stdout = _NarrowStdout(encoding="utf-8")
+
+        with mock.patch.object(_ssh_utils.sys, "stdout", stdout):
+            _write_to_terminal("plain ascii")
+
+        # Encodable text goes through print(); the fallback buffer stays empty.
+        self.assertEqual(stdout.buffer.getvalue(), b"")
+
+    def test_decode_and_output_handles_emoji_without_disconnecting(self):
+        payload = f"openclaw {EMOJI} v1".encode("utf-8")
+        response = bytes([_ssh_utils.SSH_PROXY_FORWARD,
+                          _ssh_utils.SSH_CLUSTER_STDOUT]) + payload
+        connection = mock.MagicMock()
+        stdout = _NarrowStdout(encoding="cp1252")
+
+        with mock.patch.object(_ssh_utils.sys, "stdout", stdout):
+            _decode_and_output_to_terminal(connection, response, ["utf-8", "latin_1"])
+
+        connection.disconnect.assert_not_called()
+        self.assertIn(b"openclaw", stdout.buffer.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()