diff --git a/tests/tools/test_file_write_surrogate_roundtrip.py b/tests/tools/test_file_write_surrogate_roundtrip.py
new file mode 100644
index 0000000000000..4d267e91ef7e5
--- /dev/null
+++ b/tests/tools/test_file_write_surrogate_roundtrip.py
@@ -0,0 +1,79 @@
+"""Surrogate-safe stdin piping for the local execution environment (#79178).
+
+These tests exercise the REAL `_pipe_stdin` writer thread against a real
+subprocess — no mocks. They pin the round-trip byte contract (utf-8 +
+surrogateescape is the inverse of the decode that produced the content) and
+the always-close / error-capture guarantees of the writer thread. Later
+tasks in this plan append propagation and write_file tests to this file.
+"""
+import shlex
+import subprocess
+
+import pytest
+
+from tools.environments.base import _pipe_stdin
+
+
+def _cat_to_file_proc(out_path):
+    """A real child that copies its stdin to a file, byte for byte."""
+    return subprocess.Popen(
+        ["bash", "-c", f"cat > {shlex.quote(str(out_path))}"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+    )
+
+
+def _wait_or_kill(proc, timeout=5):
+    """wait() with a bounded timeout; kill on timeout so a hung child never
+    leaks into the next test."""
+    try:
+        proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        raise
+
+
+class TestPipeStdinSurrogates:
+    def test_roundtrips_surrogateescape_bytes(self, tmp_path):
+        out = tmp_path / "out.bin"
+        proc = _cat_to_file_proc(out)
+        content = b"\xff\x00\xfe".decode("utf-8", "surrogateescape")
+        try:
+            _pipe_stdin(proc, content)
+            _wait_or_kill(proc)
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+        assert proc.returncode == 0
+        assert out.read_bytes() == b"\xff\x00\xfe"
+        assert proc._hermes_stdin_errors == []
+
+    def test_unencodable_surrogate_captures_error_and_closes_stdin(self, tmp_path):
+        out = tmp_path / "out.bin"
+        proc = _cat_to_file_proc(out)
+        try:
+            _pipe_stdin(proc, "\ud800")  # outside the surrogateescape round-trip range
+            _wait_or_kill(proc)  # child MUST exit promptly — stdin closed in finally
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+        assert proc.returncode == 0  # child saw EOF and exited cleanly
+        assert proc._hermes_stdin_errors  # the encode failure was captured
+        assert isinstance(proc._hermes_stdin_errors[0], UnicodeEncodeError)
+
+    def test_normal_content_unchanged(self, tmp_path):
+        out = tmp_path / "out.bin"
+        proc = _cat_to_file_proc(out)
+        try:
+            _pipe_stdin(proc, "hello\nworld\n")
+            _wait_or_kill(proc)
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+        assert proc.returncode == 0
+        assert out.read_bytes() == b"hello\nworld\n"
+        assert proc._hermes_stdin_errors == []
diff --git a/tools/environments/base.py b/tools/environments/base.py
index 64e7fb67968a7..7b0d0279e3480 100644
--- a/tools/environments/base.py
+++ b/tools/environments/base.py
@@ -307,21 +307,47 @@ def _pipe_stdin(proc: subprocess.Popen, data: str) -> None:
     newline translation entirely on every platform.  No behaviour change
     on POSIX — the byte sequence is identical to what text-mode would
     produce there.
+
+    Encoding uses ``errors="surrogateescape"`` — the exact inverse of the
+    surrogateescape decode, so original bytes are restored.  For
+    surrogate-free strings it is byte-identical to strict UTF-8.
+    Surrogates outside the round-trip range U+DC80–U+DCFF raise and are
+    recorded on ``proc._hermes_stdin_errors`` while stdin is still closed
+    in ``finally`` so the child sees EOF instead of hanging;
+    ``_wait_for_process`` reads the recorded error and surfaces it as
+    ``stdin_error`` on the result.
     """
 
+    errors: list[BaseException] = []
+    proc._hermes_stdin_errors = errors
+
     def _write():
+        if proc.stdin is None:
+            errors.append(RuntimeError("process stdin unavailable"))
+            return
+        # Resolve the target BEFORE encoding: a failed encode must still
+        # reach the finally-close, or the child hangs on EOF forever.
+        target = getattr(proc.stdin, "buffer", proc.stdin)
         try:
-            # proc.stdin is a TextIOWrapper when text=True was set on the
-            # Popen.  Its ``.buffer`` attribute is the raw BufferedWriter
-            # that bypasses newline translation.  When Popen was created
-            # in byte mode, proc.stdin is already a BufferedWriter with
-            # no ``.buffer`` attribute — fall back to .write() directly.
-            raw = data.encode("utf-8") if isinstance(data, str) else data
-            target = getattr(proc.stdin, "buffer", proc.stdin)
-            target.write(raw)
-            target.close()
+            raw = data.encode("utf-8", "surrogateescape") if isinstance(data, str) else data
+            written = target.write(raw)
+            if written != len(raw):
+                # Buffered writers normally complete or raise; a short write
+                # is a real failure and must be surfaced, not swallowed.
+                raise RuntimeError(f"short stdin write: {written} of {len(raw)} bytes")
         except (BrokenPipeError, OSError):
-            pass
+            pass  # child closed stdin early — normal
+        except Exception as exc:
+            # Only reachable with surrogates outside the surrogateescape
+            # round-trip range (e.g. a literal U+D800). Record it so
+            # _wait_for_process can surface it instead of a silent false
+            # success.
+            errors.append(exc)
+        finally:
+            try:
+                target.close()
+            except Exception:
+                pass
 
     threading.Thread(target=_write, daemon=True).start()