From 45aa902c18d4b2ee2250765b5cc038d8ba0bdc2e Mon Sep 17 00:00:00 2001
From: Theophilus Chinomona <executus.ahli@gmail.com>
Date: Wed, 5 Aug 2026 10:51:29 +0200
Subject: [PATCH] fix(process_registry): surrogateescape-safe PTY stdin writes
 (#79178)

---
 ...process_registry_write_stdin_surrogates.py | 38 +++++++++++++++++++
 tools/process_registry.py                     |  4 +-
 2 files changed, 41 insertions(+), 1 deletion(-)
 create mode 100644 tests/tools/test_process_registry_write_stdin_surrogates.py

diff --git a/tests/tools/test_process_registry_write_stdin_surrogates.py b/tests/tools/test_process_registry_write_stdin_surrogates.py
new file mode 100644
index 0000000000000..811323d70c7ee
--- /dev/null
+++ b/tests/tools/test_process_registry_write_stdin_surrogates.py
@@ -0,0 +1,38 @@
+"""Sibling regression test for #79178: background-PTY stdin must round-trip
+surrogateescape content instead of crashing on the strict UTF-8 encode."""
+import shlex
+import time
+
+import pytest
+
+from tools.process_registry import ProcessRegistry
+
+
+def test_write_stdin_pty_surrogateescape_roundtrip(tmp_path):
+    registry = ProcessRegistry()
+    out = tmp_path / "out.bin"
+    script = tmp_path / "read_stdin.py"
+    # readline(): a PTY never delivers EOF, so read one line (canonical mode
+    # delivers it after the newline we send).
+    script.write_text(
+        f"import sys\nopen({str(out)!r}, 'wb').write(sys.stdin.buffer.readline())\n"
+    )
+    session = registry.spawn_local(
+        f"python3 {shlex.quote(str(script))}",
+        cwd=str(tmp_path),
+        use_pty=True,
+    )
+    if session._pty is None:
+        registry.kill_process(session.id)
+        pytest.skip("ptyprocess not available; PTY path not exercised")
+    try:
+        result = registry.write_stdin(
+            session.id, b"\xff".decode("utf-8", "surrogateescape") + "\n"
+        )
+        assert result["status"] == "ok", result
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline and not out.exists():
+            time.sleep(0.05)
+        assert out.read_bytes() == b"\xff\n"
+    finally:
+        registry.kill_process(session.id)
diff --git a/tools/process_registry.py b/tools/process_registry.py
index a99c71dd15a31..f3ffe03466d90 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -2120,7 +2120,9 @@ class ProcessRegistry:
                 if _IS_WINDOWS:
                     pty_data = data.decode("utf-8") if isinstance(data, bytes) else str(data)
                 else:
-                    pty_data = data.encode("utf-8") if isinstance(data, str) else data
+                    # surrogateescape: a PTY is a byte stream — round-trip the
+                    # original bytes instead of crashing on surrogate content.
+                    pty_data = data.encode("utf-8", "surrogateescape") if isinstance(data, str) else data
                 session._pty.write(pty_data)
                 return {"status": "ok", "bytes_written": len(data)}
             except Exception as e: