[api] Short-circuit log-line decoder after first failure

_decode_pc shells out to PlatformIO via _run_idedata; without a
populated build dir for the device that subprocess fails for every
PC/BT line in a crash dump. Disable decoding after the first
EsphomeError (per logs session) and emit a single user-facing warning
instead of retrying on every line.

Also rename the helper to _LogLineProcessor since it now owns the
per-session decode-enabled state, not just the decode call.

esphome/components/api/client.py

@@ -32,30 +32,47 @@ if TYPE_CHECKING:
 _LOGGER = logging.getLogger(__name__)
 
 
-def _process_stacktrace_line(
-    config: dict[str, Any],
-    raw_line: str,
-    backtrace_state: bool,
-    platform_process_stacktrace: Any | None,
-) -> bool:
-    """Run the stack-trace decoder for a single log line.
+class _LogLineProcessor:
+    """Feeds incoming log lines to the stack-trace decoder.
 
-    on_log runs inside an asyncio protocol callback; if an exception
-    escapes, the loop tears the transport down with "Fatal error:
-    protocol.data_received() call failed." and ReconnectLogic
-    immediately reconnects, the device replays the same crash trace,
-    and we loop forever. Stack-trace decoding requires a populated
-    build dir for the device, which may not exist (e.g. flashed from
-    another machine); log and continue instead of killing the
-    connection.
+    Two responsibilities beyond just calling the decoder:
+    1. Catch EsphomeError. on_log runs inside an asyncio protocol
+       callback; if an exception escapes, the loop tears the transport
+       down with "Fatal error: protocol.data_received() call failed."
+       and ReconnectLogic immediately reconnects, the device replays
+       the same crash trace, and we loop forever.
+    2. Disable decoding after the first failure. _decode_pc shells out
+       to PlatformIO via _run_idedata, which is expensive; a single
+       crash dump can contain many PC/BT lines and we don't want to
+       retry the failing subprocess for each one.
     """
-    try:
-        if platform_process_stacktrace:
-            return platform_process_stacktrace(config, raw_line, backtrace_state)
-        return process_stacktrace(config, raw_line, backtrace_state=backtrace_state)
-    except EsphomeError as exc:
-        _LOGGER.debug("Stack-trace decoding failed: %s", exc)
-        return False
+
+    def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None:
+        self._config = config
+        self._platform_handler = platform_handler
+        self._decode_enabled = True
+        self.backtrace_state = False
+
+    def process_line(self, raw_line: str) -> None:
+        if not self._decode_enabled:
+            return
+        try:
+            if self._platform_handler is not None:
+                self.backtrace_state = self._platform_handler(
+                    self._config, raw_line, self.backtrace_state
+                )
+            else:
+                self.backtrace_state = process_stacktrace(
+                    self._config, raw_line, backtrace_state=self.backtrace_state
+                )
+        except EsphomeError as exc:
+            self._decode_enabled = False
+            self.backtrace_state = False
+            _LOGGER.warning(
+                "Crash trace decoding unavailable (%s). Run "
+                "'esphome compile' for this device to enable PC decoding.",
+                exc,
+            )
 
 
 async def async_run_logs(
@@ -87,7 +104,6 @@ async def async_run_logs(
         addresses=addresses,  # Pass all addresses for automatic retry
     )
     dashboard = CORE.dashboard
-    backtrace_state = False
 
     # Try platform-specific stacktrace handler first, fall back to generic
     platform_process_stacktrace = None
@@ -97,9 +113,10 @@ async def async_run_logs(
     except (AttributeError, ImportError):
         pass
 
+    processor = _LogLineProcessor(config, platform_process_stacktrace)
+
     def on_log(msg: SubscribeLogsResponse) -> None:
         """Handle a new log message."""
-        nonlocal backtrace_state
         time_ = datetime.now()
         message: bytes = msg.message
         text = message.decode("utf8", "backslashreplace")
@@ -110,9 +127,7 @@ async def async_run_logs(
         for parsed_msg in parse_log_message(text, timestamp):
             print(parsed_msg.replace("\033", "\\033") if dashboard else parsed_msg)
         for raw_line in text.splitlines():
-            backtrace_state = _process_stacktrace_line(
-                config, raw_line, backtrace_state, platform_process_stacktrace
-            )
+            processor.process_line(raw_line)
 
     # Safe to fall back to plaintext here only for this diagnostics use
     # case: the stream is one-way from device to client, and this code

tests/unit_tests/components/api/test_client.py

@@ -8,7 +8,7 @@ from esphome.components.api import client as api_client
 from esphome.core import EsphomeError
 
 
-def test_process_stacktrace_line_swallows_esphome_error() -> None:
+def test_decoder_swallows_esphome_error() -> None:
     """A failing stack-trace decode must not propagate.
 
     on_log runs inside an asyncio protocol callback; if EsphomeError
@@ -18,50 +18,68 @@ def test_process_stacktrace_line_swallows_esphome_error() -> None:
     reconnect.
     """
     config = {"esphome": {"name": "test"}}
+    processor = api_client._LogLineProcessor(config, None)
 
     with patch.object(
         api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
     ) as mock_process:
-        result = api_client._process_stacktrace_line(
-            config, "PC: 0x4010496e", True, None
-        )
+        processor.process_line("PC: 0x4010496e")
 
     assert mock_process.called
-    assert result is False
+    assert processor.backtrace_state is False
 
 
-def test_process_stacktrace_line_swallows_platform_handler_error() -> None:
+def test_decoder_swallows_platform_handler_error() -> None:
     """The same protection must apply to the platform-specific handler."""
     config = {"esphome": {"name": "test"}}
 
     def platform_handler(_config, _line, _state):
         raise EsphomeError("no idedata")
 
-    result = api_client._process_stacktrace_line(
-        config, "PC: 0x4010496e", True, platform_handler
-    )
+    processor = api_client._LogLineProcessor(config, platform_handler)
+    processor.process_line("PC: 0x4010496e")
 
-    assert result is False
+    assert processor.backtrace_state is False
 
 
-def test_process_stacktrace_line_returns_handler_result() -> None:
-    """When decoding succeeds, the handler's result is returned unchanged."""
+def test_decoder_short_circuits_after_failure() -> None:
+    """After one failure, subsequent lines must not retry the decoder.
+
+    _decode_pc shells out to PlatformIO; a crash dump can contain many
+    PC/BT lines and retrying the failing subprocess for each one would
+    stall log streaming.
+    """
     config = {"esphome": {"name": "test"}}
+    processor = api_client._LogLineProcessor(config, None)
 
     with patch.object(
-        api_client, "process_stacktrace", return_value=True
+        api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
     ) as mock_process:
-        result = api_client._process_stacktrace_line(
-            config, "PC: 0x4010496e", False, None
-        )
+        processor.process_line("PC: 0x4010496e")
+        processor.process_line("BT0: 0x4010496e")
+        processor.process_line("BT1: 0x401049aa")
 
-    mock_process.assert_called_once_with(
-        config, "PC: 0x4010496e", backtrace_state=False
-    )
-    assert result is True
+    assert mock_process.call_count == 1
 
 
-def test_process_stacktrace_line_uses_platform_handler_when_provided() -> None:
+def test_decoder_threads_backtrace_state() -> None:
+    """When decoding succeeds, backtrace_state is threaded across calls."""
+    config = {"esphome": {"name": "test"}}
+    processor = api_client._LogLineProcessor(config, None)
+
+    with patch.object(
+        api_client, "process_stacktrace", side_effect=[True, False]
+    ) as mock_process:
+        processor.process_line(">>>stack>>>")
+        assert processor.backtrace_state is True
+        processor.process_line("<<<stack<<<")
+        assert processor.backtrace_state is False
+
+    assert mock_process.call_args_list[0].kwargs == {"backtrace_state": False}
+    assert mock_process.call_args_list[1].kwargs == {"backtrace_state": True}
+
+
+def test_decoder_uses_platform_handler_when_provided() -> None:
     """The platform handler is preferred over the generic one."""
     config = {"esphome": {"name": "test"}}
     calls: list[tuple[object, str, bool]] = []
@@ -70,11 +88,11 @@ def test_process_stacktrace_line_uses_platform_handler_when_provided() -> None:
         calls.append((cfg, line, state))
         return True
 
+    processor = api_client._LogLineProcessor(config, platform_handler)
+
     with patch.object(api_client, "process_stacktrace") as mock_generic:
-        result = api_client._process_stacktrace_line(
-            config, "BT0: 0x4010496e", False, platform_handler
-        )
+        processor.process_line("BT0: 0x4010496e")
 
     assert calls == [(config, "BT0: 0x4010496e", False)]
     assert mock_generic.called is False
-    assert result is True
+    assert processor.backtrace_state is True