[core] Show the last line when output stops without a newline (#18265)

esphome/espidf/runner.py

@@ -90,6 +90,7 @@ def main() -> int:
         sys.path.pop(0)
     # ---- end sys.path fix-up -----------------------------------------------
 
+    import contextlib
     import os
     from pathlib import Path
     import re
@@ -179,6 +180,44 @@ def main() -> int:
         def flush(self) -> None:
             self._stream.flush()
 
+        def _emit(self, line: str) -> None:
+            if self._filter_pattern is not None:
+                stripped = ansi_escape.sub("", line).rstrip()
+                if self._filter_pattern.match(stripped) is not None:
+                    return
+            self._stream.write(line)
+
+        def drain(self) -> None:
+            """Write out a held-back line that never got its terminator.
+
+            idf.py and CMake do not always end their last line with a
+            newline, and a build that dies part way through can stop mid
+            line. Without this the user is left staring at a build that
+            ended with no explanation.
+            """
+            if not self._line_buffer:
+                return
+            line, self._line_buffer = self._line_buffer, ""
+            try:
+                # Add the terminator the line never got, so whatever ESPHome
+                # prints next does not run onto the same line.
+                self._emit(line + "\n")
+                self._stream.flush()
+            except (OSError, ValueError) as err:
+                # We are called from cleanup, so raising would replace the
+                # build's real exit code. Saying so must not raise either:
+                # under the dashboard our stdout and stderr are the same
+                # pipe, so whatever broke the write has most likely broken
+                # the report, and ``sys.__stderr__`` is None on some
+                # interpreters. Carry the line along; it is usually the
+                # message saying why the build failed.
+                if (real_stderr := sys.__stderr__) is not None:
+                    with contextlib.suppress(OSError, ValueError):
+                        print(
+                            f"Could not write out remaining output ({err}): {line}",
+                            file=real_stderr,
+                        )
+
         def write(self, data) -> int:
             # Text streams normally hand us ``str``; decode in case
             # somebody writes bytes directly.
@@ -186,7 +225,8 @@ def main() -> int:
                 data = data.decode(errors="replace")
 
             if self._filter_pattern is None:
-                self._stream.write(data)
+                # Nothing to match against, so no need to wait for a full line.
+                self._emit(data)
             else:
                 self._line_buffer += data
                 for line in self._line_buffer.splitlines(keepends=True):
@@ -195,11 +235,7 @@ def main() -> int:
                         self._line_buffer = line
                         break
                     self._line_buffer = ""
-
-                    stripped = ansi_escape.sub("", line).rstrip()
-                    if self._filter_pattern.match(stripped) is not None:
-                        continue
-                    self._stream.write(line)
+                    self._emit(line)
 
             # We tell idf.py it is talking to a terminal, so it sends progress
             # bars and cursor moves. Our own stdout is usually a pipe, which is
@@ -222,8 +258,8 @@ def main() -> int:
     is_verbose = any(arg in ("-v", "--verbose") for arg in sys.argv[2:])
     filter_lines = None if is_verbose else FILTER_IDF_LINES or None
 
-    sys.stdout = _FilteringTTYStream(sys.stdout, filter_lines)  # type: ignore[assignment]
-    sys.stderr = _FilteringTTYStream(sys.stderr, filter_lines)  # type: ignore[assignment]
+    stdout_shim = sys.stdout = _FilteringTTYStream(sys.stdout, filter_lines)  # type: ignore[assignment]
+    stderr_shim = sys.stderr = _FilteringTTYStream(sys.stderr, filter_lines)  # type: ignore[assignment]
 
     # Shift argv so the target script sees its own path as argv[0] and
     # its own arguments starting at argv[1]. runpy.run_path does not
@@ -241,8 +277,19 @@ def main() -> int:
 
     # If idf.py calls sys.exit(), SystemExit propagates out of run_path
     # and carries the exit code back to our caller. For normal returns,
-    # fall through and exit with 0.
-    runpy.run_path(script_path, run_name="__main__")
+    # fall through and exit with 0. Either way the streams get a chance to
+    # release a last line that never got its terminator. Drain the shims we
+    # made rather than sys.stdout, which the script is free to replace, and
+    # report instead of raising so cleanup cannot bury the real exit code.
+    try:
+        runpy.run_path(script_path, run_name="__main__")
+    finally:
+        # Drain stderr from a finally so a surprise from the first one cannot
+        # strand the second.
+        try:
+            stdout_shim.drain()
+        finally:
+            stderr_shim.drain()
     return 0
 
 

esphome/platformio/runner.py

@@ -179,12 +179,24 @@ def main() -> int:
     is_verbose = any(arg in ("-v", "--verbose") for arg in sys.argv[1:])
     filter_lines = None if is_verbose else FILTER_PLATFORMIO_LINES
 
-    sys.stdout = RedirectText(sys.stdout, filter_lines=filter_lines)
-    sys.stderr = RedirectText(sys.stderr, filter_lines=filter_lines)
+    stdout_redirect = sys.stdout = RedirectText(sys.stdout, filter_lines=filter_lines)
+    stderr_redirect = sys.stderr = RedirectText(sys.stderr, filter_lines=filter_lines)
 
     import platformio.__main__
 
-    return platformio.__main__.main() or 0
+    # PlatformIO exits through ``sys.exit``, so drain from a finally to give
+    # a last line without a terminator a chance to reach the user. Drain the
+    # wrappers we made rather than sys.stdout, which PlatformIO is free to
+    # replace while it runs.
+    try:
+        return platformio.__main__.main() or 0
+    finally:
+        # Drain stderr from a finally so a surprise from the first one cannot
+        # strand the second.
+        try:
+            stdout_redirect.drain()
+        finally:
+            stderr_redirect.drain()
 
 
 if __name__ == "__main__":

esphome/util.py

@@ -174,6 +174,51 @@ class RedirectText:
             s = s.replace("\033", "\\033")
         self._out.write(s)
 
+    def _emit_line(self, line: str) -> None:
+        line_without_ansi = ANSI_ESCAPE.sub("", line)
+        line_without_end = line_without_ansi.rstrip()
+        if (
+            self._filter_pattern is not None
+            and self._filter_pattern.match(line_without_end) is not None
+        ):
+            # Filter pattern matched, ignore the line
+            return
+
+        self._write_color_replace(line)
+        # Check for flash size error and provide helpful guidance
+        if (
+            "Error: The program size" in line
+            and "is greater than maximum allowed" in line
+            and (help_msg := get_esp32_arduino_flash_error_help())
+        ):
+            self._write_color_replace(help_msg)
+        for callback in self._line_callbacks:
+            if msg := callback(line_without_end):
+                self._write_color_replace(msg)
+
+    def drain(self) -> None:
+        """Write out a held-back line that never got its terminator.
+
+        A tool that dies part way through a line, or ends its output without
+        a final newline, would otherwise have that text sit in the buffer
+        and never reach the user.
+        """
+        if not self._line_buffer:
+            return
+        line, self._line_buffer = self._line_buffer, ""
+        try:
+            # Add the terminator the line never got, so whatever ESPHome
+            # prints next does not run onto the same line.
+            self._emit_line(line + "\n")
+            self._out.flush()
+        except (OSError, ValueError) as err:
+            # Every caller drains from a cleanup path, where the command's
+            # real result is already on its way out; raising here would
+            # replace it with an unrelated traceback. Carry the line into
+            # the warning, since the stream we were told to write it to is
+            # the one that just failed.
+            _LOGGER.warning("Could not write out remaining output (%s): %s", err, line)
+
     def write(self, s: str | bytes) -> int:
         # s is usually a str already (self._out is of type TextIOWrapper)
         # However, s is sometimes also a bytes object in python3. Let's make sure it's a
@@ -192,27 +237,7 @@ class RedirectText:
                     self._line_buffer = line
                     break
                 self._line_buffer = ""
-
-                line_without_ansi = ANSI_ESCAPE.sub("", line)
-                line_without_end = line_without_ansi.rstrip()
-                if (
-                    self._filter_pattern is not None
-                    and self._filter_pattern.match(line_without_end) is not None
-                ):
-                    # Filter pattern matched, ignore the line
-                    continue
-
-                self._write_color_replace(line)
-                # Check for flash size error and provide helpful guidance
-                if (
-                    "Error: The program size" in line
-                    and "is greater than maximum allowed" in line
-                    and (help_msg := get_esp32_arduino_flash_error_help())
-                ):
-                    self._write_color_replace(help_msg)
-                for callback in self._line_callbacks:
-                    if msg := callback(line_without_end):
-                        self._write_color_replace(msg)
+                self._emit_line(line)
         else:
             self._write_color_replace(s)
 
@@ -261,11 +286,11 @@ def run_external_command(
     _LOGGER.debug("Running:  %s", full_cmd)
 
     orig_stdout = sys.stdout
-    sys.stdout = RedirectText(
+    stdout_redirect = sys.stdout = RedirectText(
         sys.stdout, filter_lines=filter_lines, line_callbacks=line_callbacks
     )
     orig_stderr = sys.stderr
-    sys.stderr = RedirectText(
+    stderr_redirect = sys.stderr = RedirectText(
         sys.stderr, filter_lines=filter_lines, line_callbacks=line_callbacks
     )
 
@@ -291,6 +316,18 @@ def run_external_command(
         sys.stdout = orig_stdout
         sys.stderr = orig_stderr
 
+        # Release a last line that never got its terminator. This runs after
+        # the real streams are back, and uses the wrappers we made rather
+        # than whatever the command left in sys.stdout, so it cannot strand
+        # them. With capture_stdout the stdout wrapper was never written to,
+        # so draining it does nothing. Drain stderr from a finally so a
+        # surprise from the first one cannot strand the second; a real bug
+        # still propagates, it just does not take the other line with it.
+        try:
+            stdout_redirect.drain()
+        finally:
+            stderr_redirect.drain()
+
     if capture_stdout:
         return cap_stdout.getvalue()
 

tests/unit_tests/fixtures/espidf/closing_probe.py

@@ -0,0 +1,11 @@
+"""Leave a partial line behind and then close the stream under the runner.
+
+Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. Draining
+cannot work here; the point is that the failure is reported rather than
+raised out of the runner's cleanup, where it would bury the exit code.
+"""
+
+import sys
+
+sys.stdout.write("partial before close")
+sys.stdout.close()

tests/unit_tests/fixtures/espidf/crashing_probe.py

@@ -0,0 +1,11 @@
+"""Die part way through a line, the way a build that blows up does.
+
+Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. The
+message has no trailing newline, so the runner's shim is holding it when
+the process exits; nothing else will ever come to release it.
+"""
+
+import sys
+
+sys.stdout.write("FATAL: ld returned 1 exit status")
+sys.exit(2)

tests/unit_tests/fixtures/espidf/partial_noise_probe.py

@@ -0,0 +1,10 @@
+"""End on an unterminated line that the filter is supposed to drop.
+
+Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py, to
+check that releasing a held-back line still applies the filter.
+"""
+
+import sys
+
+sys.stdout.write("Compiling main.cpp\n")
+sys.stdout.write("Project build complete.")

tests/unit_tests/test_espidf_runner.py

@@ -19,10 +19,10 @@ from esphome.espidf import runner
 FIRST_LINE_TIMEOUT = 10.0
 
 
-def _run_main(
+def _prepare_main(
     monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str
 ) -> tuple[io.BytesIO, io.TextIOWrapper]:
-    """Run ``runner.main()`` in-process against a buffered fake stdout.
+    """Point ``runner.main()`` at *probe* with a buffered fake stdout.
 
     ``main`` rewrites ``sys.path``, ``sys.argv``, both std streams and
     ``os.get_terminal_size``; every one of those is monkeypatched so it is
@@ -39,6 +39,14 @@ def _run_main(
     monkeypatch.setattr(sys, "stderr", stream)
     monkeypatch.setattr(os, "get_terminal_size", os.get_terminal_size)
 
+    return buf, stream
+
+
+def _run_main(
+    monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str
+) -> tuple[io.BytesIO, io.TextIOWrapper]:
+    """Run ``runner.main()`` against *probe* and expect a clean exit."""
+    buf, stream = _prepare_main(monkeypatch, probe, *args)
     assert runner.main() == 0
     return buf, stream
 
@@ -59,8 +67,73 @@ def test_main_filters_noise_and_flushes_each_write(
     # Matched by FILTER_IDF_LINES, so they never leave the runner.
     assert "Project build complete." not in output
     assert "-- Component paths:" not in output
-    # Held back because no terminator arrived.
-    assert "still going" not in output
+    # Held back until the end because no terminator arrived.
+    assert output.endswith("still going\n")
+
+
+def test_main_drains_a_partial_line_when_the_build_dies(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """A build that stops mid line must still show that line.
+
+    This is the whole point of draining: the message explaining why the
+    build failed is exactly the one most likely to arrive without a
+    trailing newline.
+    """
+    buf, _stream = _prepare_main(
+        monkeypatch, fixture_path / "espidf" / "crashing_probe.py"
+    )
+
+    with pytest.raises(SystemExit) as excinfo:
+        runner.main()
+
+    assert excinfo.value.code == 2
+    assert buf.getvalue().decode("utf-8") == "FATAL: ld returned 1 exit status\n"
+
+
+def test_main_reports_rather_than_raises_when_draining_fails(
+    monkeypatch: pytest.MonkeyPatch,
+    fixture_path: Path,
+    capfd: pytest.CaptureFixture[str],
+) -> None:
+    """A stream that closed under us must not crash the runner's cleanup.
+
+    The drain runs from a ``finally``, so an exception there would replace
+    whatever exit code the build was carrying back.
+    """
+    _prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py")
+
+    assert runner.main() == 0
+    reported = capfd.readouterr().err
+    assert "Could not write out remaining output" in reported
+    # The held line has to come along; the stream it was meant for is gone.
+    assert "partial before close" in reported
+
+
+def test_main_survives_a_drain_failure_with_nowhere_to_report_it(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """With no real stderr to report to, cleanup still must not raise.
+
+    ``sys.__stderr__`` is None on some interpreters, and ``print(file=None)``
+    falls back to ``sys.stdout``, which here is the shim wrapping the stream
+    that just failed.
+    """
+    monkeypatch.setattr(sys, "__stderr__", None)
+    _prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py")
+
+    assert runner.main() == 0
+
+
+def test_main_still_filters_a_drained_partial_line(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """Releasing a held line does not smuggle noise past the filter."""
+    buf, _stream = _run_main(
+        monkeypatch, fixture_path / "espidf" / "partial_noise_probe.py"
+    )
+
+    assert buf.getvalue().decode("utf-8") == "Compiling main.cpp\n"
 
 
 def test_main_keeps_everything_in_verbose_mode(

tests/unit_tests/test_platformio_runner.py

@@ -0,0 +1,93 @@
+"""Tests for esphome.platformio.runner."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+import io
+import sys
+from types import ModuleType
+
+import pytest
+
+from esphome.platformio import runner
+
+
+def _prepare_main(
+    monkeypatch: pytest.MonkeyPatch, pio_main: Callable[[], int]
+) -> io.BytesIO:
+    """Point ``runner.main()`` at a fake PlatformIO with a fake stdout.
+
+    The real ``main`` patches PlatformIO internals and then hands control to
+    it; both are stubbed out so only the stream wrapping is exercised. The
+    fake stdout is block buffered like a pipe, so the caller can see what
+    actually left the wrapper.
+    """
+    buf = io.BytesIO()
+    stream = io.TextIOWrapper(buf, encoding="utf-8", newline="\n", line_buffering=False)
+
+    monkeypatch.setattr(sys, "argv", ["pio", "run"])
+    monkeypatch.setattr(sys, "stdout", stream)
+    monkeypatch.setattr(sys, "stderr", stream)
+    monkeypatch.setattr(runner, "patch_structhash", lambda: None)
+    monkeypatch.setattr(runner, "patch_file_downloader", lambda: None)
+
+    platformio = ModuleType("platformio")
+    platformio_main = ModuleType("platformio.__main__")
+    platformio_main.main = pio_main  # type: ignore[attr-defined]
+    platformio.__main__ = platformio_main  # type: ignore[attr-defined]
+    monkeypatch.setitem(sys.modules, "platformio", platformio)
+    monkeypatch.setitem(sys.modules, "platformio.__main__", platformio_main)
+
+    return buf
+
+
+def test_main_drains_a_partial_line_on_a_clean_run(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A build ending mid line still shows that line."""
+
+    def pio_main() -> int:
+        print("Linking .pioenvs/firmware.elf\n", end="")
+        print("Building took 12.4 seconds", end="")
+        return 0
+
+    buf = _prepare_main(monkeypatch, pio_main)
+
+    assert runner.main() == 0
+    assert buf.getvalue().decode("utf-8") == (
+        "Linking .pioenvs/firmware.elf\nBuilding took 12.4 seconds\n"
+    )
+
+
+def test_main_drains_when_platformio_exits_early(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Leaving through ``sys.exit`` still drains, because it runs in a finally."""
+
+    def pio_main() -> int:
+        print("*** [.pioenvs/firmware.elf] Error 1", end="")
+        sys.exit(1)
+
+    buf = _prepare_main(monkeypatch, pio_main)
+
+    with pytest.raises(SystemExit) as excinfo:
+        runner.main()
+
+    assert excinfo.value.code == 1
+    assert buf.getvalue().decode("utf-8") == "*** [.pioenvs/firmware.elf] Error 1\n"
+
+
+def test_main_still_filters_a_drained_partial_line(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Releasing a held line does not smuggle noise past the filter."""
+
+    def pio_main() -> int:
+        # Matches FILTER_PLATFORMIO_LINES, and arrives without a terminator.
+        print("Verbose mode can be enabled via `-v, --verbose` option", end="")
+        return 0
+
+    buf = _prepare_main(monkeypatch, pio_main)
+
+    assert runner.main() == 0
+    assert buf.getvalue() == b""

tests/unit_tests/test_util.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 import io
+import logging
 from pathlib import Path
 import subprocess
 import sys
@@ -442,6 +443,69 @@ def test_redirect_text_flushes_so_piped_output_streams() -> None:
     assert buf.getvalue() == b"Writing at 0x00010000 (50%)\r"
 
 
+def test_redirect_text_drain_releases_held_partial_line() -> None:
+    """A last line with no terminator must still reach the user.
+
+    A tool that dies part way through a line leaves that text in the buffer,
+    and it is usually the message saying what went wrong.
+    """
+    redirect, buf = _make_redirect(filter_lines=["ignore me"])
+    redirect.write("FATAL: ld returned 1 exit status")
+
+    # Still held: no terminator has arrived.
+    assert buf.getvalue() == ""
+
+    redirect.drain()
+
+    assert buf.getvalue() == "FATAL: ld returned 1 exit status\n"
+
+
+def test_redirect_text_drain_still_applies_the_filter() -> None:
+    """Releasing a held line does not smuggle noise past the filter."""
+    redirect, buf = _make_redirect(filter_lines=["Verbose mode can be enabled"])
+    redirect.write("Verbose mode can be enabled")
+
+    redirect.drain()
+
+    assert buf.getvalue() == ""
+
+
+def test_redirect_text_drain_is_a_no_op_when_nothing_is_held() -> None:
+    """Draining twice, or with an empty buffer, writes nothing extra."""
+    redirect, buf = _make_redirect(filter_lines=["ignore me"])
+    redirect.write("complete line\n")
+
+    redirect.drain()
+    redirect.drain()
+
+    assert buf.getvalue() == "complete line\n"
+
+
+def test_redirect_text_adds_flash_size_help(monkeypatch: pytest.MonkeyPatch) -> None:
+    """An out-of-flash error gets the how-to-fix note appended."""
+    monkeypatch.setattr(
+        util, "get_esp32_arduino_flash_error_help", lambda: "TIP: switch to esp-idf\n"
+    )
+    redirect, buf = _make_redirect(filter_lines=["ignore me"])
+
+    redirect.write("Error: The program size is greater than maximum allowed\n")
+
+    assert "Error: The program size" in buf.getvalue()
+    assert "TIP: switch to esp-idf" in buf.getvalue()
+
+
+def test_redirect_text_skips_flash_size_help_on_other_platforms(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The note is ESP32-with-Arduino only, so elsewhere the line stands alone."""
+    monkeypatch.setattr(util, "get_esp32_arduino_flash_error_help", lambda: None)
+    redirect, buf = _make_redirect(filter_lines=["ignore me"])
+
+    redirect.write("Error: The program size is greater than maximum allowed\n")
+
+    assert buf.getvalue() == "Error: The program size is greater than maximum allowed\n"
+
+
 def test_redirect_text_callback_called_on_matching_line() -> None:
     """Test that a line callback is called and its output is written."""
     results: list[str] = []
@@ -571,6 +635,140 @@ def test_run_external_command_line_callbacks(capsys: pytest.CaptureFixture) -> N
     assert "CALLBACK FIRED" in captured.out
 
 
+def test_run_external_command_drains_partial_line(
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """A command that stops mid line still shows that line.
+
+    esptool runs in-process here, so a message it writes without a trailing
+    newline would otherwise be dropped when the streams are put back.
+    """
+
+    def fake_main() -> int:
+        print("A fatal error occurred: no serial data", end="")
+        return 1
+
+    rc = util.run_external_command(fake_main, "fake", filter_lines=["ignore me"])
+
+    assert rc == 1
+    assert "A fatal error occurred: no serial data" in capsys.readouterr().out
+
+
+def test_run_external_command_drains_on_early_exit(
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """The drain also happens when the command exits through ``sys.exit``."""
+
+    def fake_main() -> int:
+        print("Fatal: bailing out", end="")
+        sys.exit(3)
+
+    rc = util.run_external_command(fake_main, "fake", filter_lines=["ignore me"])
+
+    assert rc == 3
+    assert "Fatal: bailing out" in capsys.readouterr().out
+
+
+def test_run_external_command_capture_stdout_has_nothing_to_drain() -> None:
+    """With ``capture_stdout`` there is nothing held to write out.
+
+    The stdout wrapper still gets built, but ``sys.stdout`` is replaced by
+    the capture buffer right after, so the wrapper never sees a write and
+    draining it does nothing.
+    """
+
+    def fake_main() -> int:
+        print("captured output", end="")
+        return 0
+
+    out = util.run_external_command(
+        fake_main, "fake", capture_stdout=True, filter_lines=["ignore me"]
+    )
+
+    assert out == "captured output"
+
+
+def test_run_external_command_survives_a_command_that_swaps_stdout(
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """Draining must not depend on what the command left in ``sys.stdout``.
+
+    A command is free to replace the stream; reaching for ``drain`` on
+    whatever it left there would raise from the cleanup path and bury the
+    real exit code.
+    """
+
+    def fake_main() -> int:
+        print("before the swap", end="")
+        sys.stdout = io.StringIO()
+        sys.exit(7)
+
+    rc = util.run_external_command(fake_main, "fake", filter_lines=["ignore me"])
+
+    assert rc == 7
+    assert "before the swap" in capsys.readouterr().out
+
+
+def test_drain_reports_the_lost_line_instead_of_raising(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A broken stream during cleanup is reported, not raised.
+
+    The warning carries the held text, because the stream we were asked to
+    write it to is the one that just failed.
+    """
+    caplog.set_level(logging.WARNING, logger=util.__name__)
+    out = MagicMock()
+    out.write.side_effect = BrokenPipeError("pipe is gone")
+    redirect = util.RedirectText(out, filter_lines=["ignore me"])
+    redirect.write("FATAL: ld returned 1 exit status")
+
+    redirect.drain()
+
+    assert "pipe is gone" in caplog.text
+    assert "FATAL: ld returned 1 exit status" in caplog.text
+
+
+def test_drain_lets_other_errors_through() -> None:
+    """Only an unusable stream is tolerated; a bug still has to be visible."""
+
+    def broken_callback(line: str) -> str | None:
+        raise TypeError("a line callback is broken")
+
+    redirect, _buf = _make_redirect(line_callbacks=[broken_callback])
+    redirect.write("a line with no terminator")
+
+    with pytest.raises(TypeError):
+        redirect.drain()
+
+
+def test_run_external_command_drains_stderr_even_if_stdout_drain_raises(
+    capsys: pytest.CaptureFixture,
+) -> None:
+    """One stream failing must not strand the other's held line.
+
+    ``drain`` deliberately lets anything that is not a stream error through,
+    so a broken line callback would otherwise skip the stderr drain and take
+    that line down with it.
+    """
+
+    def broken_on_stdout(line: str) -> str | None:
+        if "stdout" in line:
+            raise TypeError("a line callback is broken")
+        return None
+
+    def fake_main() -> int:
+        print("stdout partial", end="")
+        print("stderr FATAL: the real reason", end="", file=sys.stderr)
+        return 0
+
+    with pytest.raises(TypeError):
+        util.run_external_command(fake_main, "fake", line_callbacks=[broken_on_stdout])
+
+    # The bug still surfaces, but stderr's held line was written first.
+    assert "stderr FATAL: the real reason" in capsys.readouterr().err
+
+
 def test_run_external_process_line_callbacks() -> None:
     """Test that run_external_process passes line_callbacks to RedirectText."""
     results: list[str] = []