[core] Stop dropping complete lines behind an unfinished one (#18279)

esphome/espidf/runner.py

@@ -144,12 +144,14 @@ def main() -> int:
 
         * ``isatty()`` unconditionally returns True, tricking downstream
           code into emitting TTY-format output.
-        * Input is split on ``\\n`` / ``\\r`` via
-          ``str.splitlines(keepends=True)`` and any complete line whose
+        * Input is split with ``str.splitlines(keepends=True)``, which
+          breaks on more than ``\\n`` and ``\\r``; form feed and a few
+          other control characters count too. Any piece whose
           ANSI-stripped, right-stripped form matches one of
           ``filter_lines`` is dropped.
-        * Incomplete trailing chunks are held in a buffer until a
-          terminator arrives.
+        * Only the final piece can still be waiting for more text, so
+          that one is held until a ``\\n`` or ``\\r`` arrives. A piece
+          that ended on one of the other breaks goes out as it is.
 
         Mirrors the matching semantics of ``esphome.util.RedirectText``
         so filter patterns behave identically in both the PlatformIO
@@ -228,13 +230,22 @@ def main() -> int:
                 # 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):
-                    if "\n" not in line and "\r" not in line:
-                        # Incomplete — hold until we see a terminator.
-                        self._line_buffer = line
-                        break
+                lines = (self._line_buffer + data).splitlines(keepends=True)
+                # Every piece but the last ends with something
+                # ``str.splitlines`` treats as a break, so only the last one
+                # can still be waiting for more text. Hold that one, write
+                # out the rest.
+                #
+                # Some of those breaks are not line endings to us, a form
+                # feed for one, so a piece can go out without ending in a
+                # newline. That beats what we did before, which was to stop
+                # at the first such piece and drop every complete line
+                # behind it.
+                if lines and not lines[-1].endswith(("\n", "\r")):
+                    self._line_buffer = lines.pop()
+                else:
                     self._line_buffer = ""
+                for line in lines:
                     self._emit(line)
 
             # We tell idf.py it is talking to a terminal, so it sends progress

esphome/util.py

@@ -229,14 +229,21 @@ class RedirectText:
             s = s.decode()
 
         if self._filter_pattern is not None or self._line_callbacks:
-            self._line_buffer += s
-            lines = self._line_buffer.splitlines(True)
-            for line in lines:
-                if "\n" not in line and "\r" not in line:
-                    # Not a complete line, set line buffer
-                    self._line_buffer = line
-                    break
+            lines = (self._line_buffer + s).splitlines(True)
+            # Every piece but the last ends with something
+            # ``str.splitlines`` treats as a break, so only the last one can
+            # still be waiting for more text. Hold that one, write out the
+            # rest.
+            #
+            # Some of those breaks are not line endings to us, a form feed
+            # for one, so a piece can go out without ending in a newline.
+            # That beats what we did before, which was to stop at the first
+            # such piece and drop every complete line behind it.
+            if lines and not lines[-1].endswith(("\n", "\r")):
+                self._line_buffer = lines.pop()
+            else:
                 self._line_buffer = ""
+            for line in lines:
                 self._emit_line(line)
         else:
             self._write_color_replace(s)

tests/unit_tests/fixtures/espidf/formfeed_probe.py

@@ -0,0 +1,12 @@
+"""Write a form feed part way through the output.
+
+Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. A form
+feed is not a line terminator here, so everything written must still come
+out, including the complete lines that follow it.
+"""
+
+import sys
+
+sys.stdout.write("Compiling main.cpp\n")
+sys.stdout.write("page one\x0cpage two\n")
+sys.stdout.write("[2/9] Building C object\n")

tests/unit_tests/test_espidf_runner.py

@@ -71,6 +71,17 @@ def test_main_filters_noise_and_flushes_each_write(
     assert output.endswith("still going\n")
 
 
+def test_main_keeps_output_after_a_form_feed(
+    monkeypatch: pytest.MonkeyPatch, fixture_path: Path
+) -> None:
+    """A form feed is text, not a line break, so nothing after it is lost."""
+    buf, _stream = _run_main(monkeypatch, fixture_path / "espidf" / "formfeed_probe.py")
+
+    assert buf.getvalue().decode("utf-8") == (
+        "Compiling main.cpp\npage one\x0cpage two\n[2/9] Building C object\n"
+    )
+
+
 def test_main_drains_a_partial_line_when_the_build_dies(
     monkeypatch: pytest.MonkeyPatch, fixture_path: Path
 ) -> None:

tests/unit_tests/test_util.py

@@ -443,6 +443,42 @@ def test_redirect_text_flushes_so_piped_output_streams() -> None:
     assert buf.getvalue() == b"Writing at 0x00010000 (50%)\r"
 
 
+@pytest.mark.parametrize(
+    "break_char",
+    ["\x0c", "\x0b", "\x1c", "\x1d", "\x1e", "\x85", "\u2028", "\u2029"],
+    ids=["formfeed", "vtab", "fs", "gs", "rs", "nel", "lsep", "psep"],
+)
+def test_redirect_text_keeps_output_after_an_exotic_break_character(
+    break_char: str,
+) -> None:
+    r"""Only ``\n`` and ``\r`` end a line; the rest is ordinary text.
+
+    ``str.splitlines`` treats all of these as line breaks. Splitting on them
+    used to strand the fragment in the buffer and drop every complete line
+    that came after it, which for a form feed in toolchain output meant
+    losing the rest of the build log.
+    """
+    redirect, buf = _make_redirect(filter_lines=["ignore me"])
+
+    redirect.write(f"first{break_char}second\nthird\n")
+
+    assert buf.getvalue() == f"first{break_char}second\nthird\n"
+
+
+def test_redirect_text_treats_crlf_as_one_terminator() -> None:
+    r"""``\r\n``, a lone ``\r`` and a lone ``\n`` each end exactly one line."""
+    redirect, buf = _make_redirect(filter_lines=["ignore me"])
+
+    redirect.write("one\r\ntwo\rthree\nfour")
+
+    # "four" has no terminator yet, so it is held back.
+    assert buf.getvalue() == "one\r\ntwo\rthree\n"
+
+    redirect.drain()
+
+    assert buf.getvalue() == "one\r\ntwo\rthree\nfour\n"
+
+
 def test_redirect_text_drain_releases_held_partial_line() -> None:
     """A last line with no terminator must still reach the user.