fix: escape undecodable subprocess output bytes and deflake graceful shutdown test

src/openjd/adaptor_runtime/process/_logging_subprocess.py

@@ -69,8 +69,12 @@ def __init__(
             # error handling a single undecodable byte raises UnicodeDecodeError in the
             # stream-reader thread, killing it. Once the reader thread is gone the pipe is
             # no longer drained, the subprocess blocks on a full stdout pipe, and the job
-            # deadlocks. Replacing undecodable bytes keeps the reader alive.
-            errors="replace",
+            # deadlocks. Escaping undecodable bytes (e.g. b"\xc7" -> "\\xc7") keeps the
+            # reader alive while preserving the original byte values in the logs, which
+            # helps identify the codepage the subprocess is emitting.
+            # Note: encoding/errors also apply to stdin, so text written to the child's
+            # stdin would have unencodable characters escaped silently rather than raising.
+            errors="backslashreplace",
             cwd=startup_directory,
         )
         if OSName.is_windows():  # pragma: is-posix

test/openjd/adaptor_runtime/integ/process/test_integration_logging_subprocess.py

@@ -174,13 +174,77 @@ def test_non_utf8_output_does_not_kill_reader(self, caplog):
         # WHEN
         p = LoggingSubprocess(args=[sys.executable, "-c", script])
         p.wait()
-        p._cleanup_io_threads()
 
         # THEN
         # "after" is only logged if the reader thread survived the undecodable byte.
         assert "after" in caplog.text
-        # The undecodable byte is replaced rather than raising.
-        assert "�" in caplog.text
+        # The undecodable byte is escaped rather than raising.
+        assert "bad \\xc7 byte" in caplog.text
+
+    @pytest.mark.timeout(10)
+    @pytest.mark.parametrize(
+        argnames=("raw_bytes", "expected_escaped"),
+        argvalues=[
+            # 0xc7 is "Ç" in cp1252 (observed from 3dsmaxbatch.exe on stdout).
+            (b"bad \xc7 byte", "bad \\xc7 byte"),
+            # 0xff is never valid anywhere in UTF-8.
+            (b"bad \xff byte", "bad \\xff byte"),
+            # Consecutive invalid bytes must each be escaped separately.
+            (b"bad \xc7\xff bytes", "bad \\xc7\\xff bytes"),
+            # cp1252-encoded "Çé" — an invalid two-byte run in UTF-8.
+            (b"bad \xc7\xe9 text", "bad \\xc7\\xe9 text"),
+            # A truncated UTF-8 multi-byte sequence (0xe4 0xbd is an incomplete
+            # 3-byte sequence) followed by valid ASCII.
+            (b"truncated \xe4\xbd then ok", "truncated \\xe4\\xbd then ok"),
+        ],
+        ids=["cp1252-byte", "invalid-byte", "consecutive-invalid", "cp1252-text", "truncated-utf8"],
+    )
+    def test_non_utf8_output_is_escaped(self, caplog, raw_bytes: bytes, expected_escaped: str):
+        """
+        Undecodable bytes in subprocess output must be escaped with backslashreplace
+        (e.g. b"\xc7" -> "\\xc7") so the original byte values are preserved in the logs.
+        """
+        # GIVEN
+        caplog.set_level(_STDOUT_LEVEL)
+        script = (
+            "import sys; "
+            f"sys.stdout.buffer.write({raw_bytes + b'!'!r}); "
+            "sys.stdout.buffer.flush()"
+        )
+
+        # WHEN
+        p = LoggingSubprocess(args=[sys.executable, "-c", script])
+        p.wait()
+
+        # THEN
+        # The trailing "!" proves the full line was logged, not truncated at the bad byte.
+        assert expected_escaped + "!" in caplog.text
+        # The replacement character must not appear; the byte value must be preserved.
+        assert "�" not in caplog.text
+
+    @pytest.mark.timeout(10)
+    def test_valid_utf8_is_not_escaped(self, caplog):
+        """
+        Valid multi-byte UTF-8 sequences must pass through unmodified — escaping applies
+        only to genuinely invalid sequences, even when a multi-byte character could be
+        split across internal read chunk boundaries.
+        """
+        # GIVEN
+        caplog.set_level(_STDOUT_LEVEL)
+        message = "héllo wörld Ç 星期五"
+        script = (
+            "import sys; "
+            f"sys.stdout.buffer.write({message.encode('utf-8')!r} + b'\\n'); "
+            "sys.stdout.buffer.flush()"
+        )
+
+        # WHEN
+        p = LoggingSubprocess(args=[sys.executable, "-c", script])
+        p.wait()
+
+        # THEN
+        assert message in caplog.text
+        assert "\\x" not in caplog.text
 
     def test_executable_not_found(self):
         """When calling LoggingSubprocess with a missing executable, FileNotFoundError will be raised"""

test/openjd/adaptor_runtime/unit/process/test_logging_subprocess.py

@@ -69,7 +69,7 @@ def test_process_creation(self, mock_popen: mock.Mock, mock_stream_logger: mock.
         popen_params = dict(
             args=args,
             encoding="utf-8",
-            errors="replace",
+            errors="backslashreplace",
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
@@ -483,7 +483,7 @@ def test_startup_directory_default(self, mock_popen_autospec: mock.Mock):
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             encoding="utf-8",
-            errors="replace",
+            errors="backslashreplace",
             cwd=None,
         )
         if OSName.is_windows():
@@ -506,7 +506,7 @@ def test_start_directory(self, mock_popen_autospec: mock.Mock):
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             encoding="utf-8",
-            errors="replace",
+            errors="backslashreplace",
             cwd="startup_dir",
         )
         if OSName.is_windows():

test/openjd/adaptor_runtime_client/integ/fake_client.py

@@ -37,6 +37,9 @@ def run(self):
 
 def run_client():
     test_client = FakeClient("1234")
+    # Signal handler registration (in the main thread) happens in the constructor above.
+    # Tests wait for this line before sending signals to avoid a startup race.
+    print("client ready", flush=True)
     test_client.run()
 
 

test/openjd/adaptor_runtime_client/integ/test_integration_client_interface.py

@@ -33,8 +33,13 @@ def test_graceful_shutdown(self) -> None:
             popen_params.update(creationflags=_subprocess.CREATE_NEW_PROCESS_GROUP)  # type: ignore[attr-defined]
         client_subprocess = _subprocess.Popen(**popen_params)
 
-        # To avoid a race condition, giving some extra time for the logging subprocess to start.
-        _sleep(0.5 if OSName.is_posix() else 4)
+        # Wait for the client to report that it has started (and registered its signal
+        # handler) before sending the signal. A fixed sleep is racy on slow CI hosts: if the
+        # signal arrives before the handler is registered, the default handler kills the
+        # process and the test fails with empty output.
+        assert client_subprocess.stdout is not None
+        ready_line = client_subprocess.stdout.readline()
+        assert "client ready" in ready_line
         signal_type: signal.Signals
         if OSName.is_windows():
             signal_type = signal.CTRL_BREAK_EVENT  # type: ignore[attr-defined]