[core] Lift the log line processor into esphome/stacktrace.py (#18076)

esphome/__main__.py

@@ -58,6 +58,7 @@ from esphome.core import CORE, EsphomeError, coroutine
 from esphome.enum import StrEnum
 from esphome.helpers import get_bool_env, indent, is_ip_address
 from esphome.log import AnsiFore, color, setup_log
+from esphome.stacktrace import LogLineProcessor
 from esphome.types import ConfigType
 from esphome.upload_targets import PortType, get_port_type
 from esphome.util import (
@@ -631,11 +632,9 @@ def run_miniterm(config: ConfigType, port: str, args) -> int:
         return 1
     _LOGGER.info("Starting log output from %s with baud rate %s", port, baud_rate)
 
-    # Stacktrace analysis is optional; platform_hooks owns resolution
+    # Decoder resolution, crash isolation, and disable-after-failure
-    # and the user-facing messages.
+    # all live in LogLineProcessor, shared with the API log path.
-    process_stacktrace = platform_hooks.get_stacktrace_handler(CORE.target_platform)
+    processor = LogLineProcessor(config, CORE.target_platform)
-
-    backtrace_state = False
     ser = serial.Serial()
     ser.baudrate = baud_rate
     ser.port = port
@@ -675,11 +674,7 @@ def run_miniterm(config: ConfigType, port: str, args) -> int:
                                 "utf8", "backslashreplace"
                             )
                             safe_print(parser.parse_line(line, time_str))
-
+                            processor.process_line(line)
-                            if process_stacktrace is not None:
-                                backtrace_state = process_stacktrace(
-                                    config, line, backtrace_state
-                                )
                     except serial.SerialException:
                         _LOGGER.error("Serial port closed!")
                         return 0

esphome/api_client.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import asyncio
 from contextlib import suppress
 from datetime import datetime
-import importlib
 import logging
 from typing import TYPE_CHECKING, Any
 import warnings
@@ -18,6 +17,7 @@ with warnings.catch_warnings():
 
 from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_PORT, __version__
 from esphome.core import CORE
+from esphome.stacktrace import LogLineProcessor
 from esphome.util import safe_print
 
 if TYPE_CHECKING:
@@ -29,50 +29,6 @@ if TYPE_CHECKING:
 _LOGGER = logging.getLogger(__name__)
 
 
-class _LogLineProcessor:
-    """Feeds incoming log lines to the stack-trace decoder.
-
-    Two responsibilities beyond just calling the decoder:
-    1. Catch everything the decoder can raise. aioesphomeapi isolates
-       exceptions raised by log handlers, so an escaping one no longer
-       kills the session, but it does log a full traceback per line. A
-       crash dump carries a PC line plus one per backtrace frame, so the
-       tracebacks bury the dump the user is trying to read. Decoding is a
-       diagnostic nicety; nothing it raises is worth that noise.
-    2. Disable decoding after the first failure. _decode_pc shells out to
-       the toolchain to resolve addr2line, 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. This only works if every
-       failure is caught, which is why 1 is not narrowed to EsphomeError.
-    """
-
-    def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None:
-        self._config = config
-        self._platform_handler = platform_handler
-        self._decode_enabled = platform_handler is not None
-        self.backtrace_state = False
-
-    def process_line(self, raw_line: str) -> None:
-        if not self._decode_enabled:
-            return
-        try:
-            self.backtrace_state = self._platform_handler(
-                self._config, raw_line, self.backtrace_state
-            )
-        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-except
-            self._decode_enabled = False
-            self.backtrace_state = False
-            # _run_idedata raises EsphomeError with no message; fall back
-            # to a generic explanation when str(exc) is empty.
-            detail = str(exc) or "build artifacts not found locally"
-            _LOGGER.debug("Stack-trace decoding failed", exc_info=True)
-            _LOGGER.warning(
-                "Crash trace decoding unavailable: %s. "
-                "Run 'esphome compile' for this device to enable PC decoding.",
-                detail,
-            )
-
-
 async def async_run_logs(
     config: dict[str, Any],
     addresses: list[str],
@@ -100,21 +56,9 @@ async def async_run_logs(
         provide_time=False,
     )
 
-    # Try platform-specific stacktrace handler first, fall back to generic
+    # Decoder resolution, crash isolation, and disable-after-failure
-    platform_process_stacktrace = None
+    # all live in LogLineProcessor, shared with the serial log path.
-    try:
+    processor = LogLineProcessor(config, CORE.target_platform)
-        module = importlib.import_module("esphome.components." + CORE.target_platform)
-        platform_process_stacktrace = module.process_stacktrace
-    except (AttributeError, ImportError):
-        # Distinguish "platform has no analyzer" from a genuinely broken
-        # platform package when debugging.
-        _LOGGER.debug("Stacktrace analyzer lookup failed", exc_info=True)
-        _LOGGER.info(
-            'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".',
-            CORE.target_platform,
-        )
-
-    processor = _LogLineProcessor(config, platform_process_stacktrace)
 
     def on_log(msg: SubscribeLogsResponse) -> None:
         """Handle a new log message."""

esphome/platform_hooks.py

@@ -10,10 +10,9 @@ imports each platform package and fails when they drift.
 
 The compile-path ``run_compile`` hook is deliberately not registered:
 compiling imports the platform package regardless, so its probe in
-``__main__.py`` stays eager. The serial log path resolves
+``__main__.py`` stays eager. Both log paths resolve
-``process_stacktrace`` through get_stacktrace_handler below; the network
+``process_stacktrace`` through ``esphome.stacktrace.LogLineProcessor``,
-log client's probe in ``esphome/api_client.py`` still uses the old
+which uses get_stacktrace_handler below.
-importlib pattern and is converted separately.
 """
 
 from __future__ import annotations

esphome/platformio/toolchain.py

@@ -6,7 +6,7 @@ from pathlib import Path
 import re
 import shutil
 import sys
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import platformdirs
 
@@ -364,18 +364,24 @@ def run_compile(config, verbose):
 def _run_idedata(config):
     args = ["-t", "idedata"]
     stdout = run_platformio_cli_run(config, False, *args, capture_stdout=True)
+    if not isinstance(stdout, str):
+        # run_external_process returns 1 instead of captured output when
+        # launching platformio raised; see the error it logged above.
+        raise EsphomeError("Could not launch platformio to get idedata")
     match = re.search(r'{\s*".*}', stdout)
     if match is None:
-        _LOGGER.error("Could not match idedata, please report this error")
+        # A run that launches but fails emits its build error instead of
+        # idedata; the logged stdout is the useful part, not a bug report.
+        _LOGGER.error("Could not find idedata in the platformio output")
         _LOGGER.error("Stdout: %s", stdout)
-        raise EsphomeError
+        raise EsphomeError("PlatformIO did not report idedata")
 
     try:
         return json.loads(match.group())
-    except ValueError:
+    except ValueError as err:
         _LOGGER.exception("Could not parse idedata")
         _LOGGER.error("Stdout: %s", stdout)
-        raise
+        raise EsphomeError("Could not parse idedata from platformio") from err
 
 
 def _load_idedata(config):
@@ -419,9 +425,27 @@ class IDEData:
     def __init__(self, raw):
         self.raw = raw
 
+    def _require(self, *keys: str) -> Any:
+        """Read a nested key, classifying a miss as an environment error.
+
+        A stale or truncated cached idedata JSON is the user's build
+        tree, not a bug; recompiling regenerates it. The message names
+        the key so a platformio schema change stays diagnosable.
+        """
+        value = self.raw
+        # TypeError covers a key that is null instead of absent.
+        try:
+            for key in keys:
+                value = value[key]
+        except (KeyError, TypeError) as err:
+            raise EsphomeError(
+                f"Cached idedata is incomplete (missing {'.'.join(keys)})"
+            ) from err
+        return value
+
     @property
     def firmware_elf_path(self) -> Path:
-        return Path(self.raw["prog_path"])
+        return Path(self._require("prog_path"))
 
     @property
     def firmware_bin_path(self) -> Path:
@@ -429,15 +453,22 @@ class IDEData:
 
     @property
     def extra_flash_images(self) -> list[FlashImage]:
-        return [
+        try:
-            FlashImage(path=Path(entry["path"]), offset=entry["offset"])
+            return [
-            for entry in self.raw["extra"]["flash_images"]
+                FlashImage(path=Path(entry["path"]), offset=entry["offset"])
-        ]
+                for entry in self._require("extra", "flash_images")
+            ]
+        except (KeyError, TypeError) as err:
+            # Covers entries missing path/offset and a null or non-list
+            # flash_images value alike.
+            raise EsphomeError(
+                "Cached idedata is incomplete (malformed extra.flash_images)"
+            ) from err
 
     @property
     def cc_path(self) -> str:
         # For example /Users/<USER>/.platformio/packages/toolchain-xtensa32/bin/xtensa-esp32-elf-gcc
-        return self.raw["cc_path"]
+        return self._require("cc_path")
 
     @property
     def addr2line_path(self) -> str:

esphome/stacktrace.py

@@ -0,0 +1,105 @@
+"""Stack-trace decoding for streamed device log lines.
+
+Shared by the serial (run_miniterm) and network (api_client) log paths.
+Deliberately light: importing this module must not pull in aioesphomeapi
+or any platform package.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from esphome import platform_hooks
+from esphome.core import EsphomeError
+from esphome.types import ConfigType
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    # The contract every platform's process_stacktrace implements.
+    StacktraceHandler = Callable[[ConfigType, str, bool], bool]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class LogLineProcessor:
+    """Feeds incoming log lines to the stack-trace decoder.
+
+    Two responsibilities beyond just calling the decoder:
+    1. Catch everything the decoder can raise. aioesphomeapi isolates
+       exceptions raised by log handlers, so an escaping one no longer
+       kills the session, but it does log a full traceback per line. A
+       crash dump carries a PC line plus one per backtrace frame, so the
+       tracebacks bury the dump the user is trying to read. Decoding is a
+       diagnostic nicety; nothing it raises is worth that noise.
+    2. Disable decoding for the rest of the session after a failure.
+       _decode_pc shells out to the toolchain to resolve addr2line,
+       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. This only works if every failure is caught, which is why 1
+       is not narrowed to EsphomeError. The latch is deliberately one
+       way: nothing a decode failure depends on heals by itself within
+       a session, the warning names the fix, and a fresh ``esphome
+       logs`` run picks it up; retrying mid-session would block the
+       stream with a failing subprocess instead.
+    """
+
+    def __init__(self, config: ConfigType, platform: str) -> None:
+        self._config = config
+        self._platform = platform
+        self._platform_handler: StacktraceHandler | None
+        try:
+            self._platform_handler = platform_hooks.get_stacktrace_handler(platform)
+        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-except
+            # Total containment includes resolution: a platform package
+            # broken in an unanticipated way must not kill the session.
+            # Name the cause; the full traceback only exists at debug.
+            _LOGGER.debug("Stacktrace analyzer resolution failed", exc_info=True)
+            _LOGGER.warning(
+                'Stacktrace analysis is unavailable: analyzer for target platform "%s" could not be loaded: %s',
+                platform,
+                f"{type(exc).__name__}: {exc}",
+            )
+            self._platform_handler = None
+        self._decode_enabled = self._platform_handler is not None
+        self.backtrace_state = False
+
+    def process_line(self, raw_line: str) -> None:
+        if not self._decode_enabled:
+            return
+        self._feed(raw_line)
+
+    def _feed(self, raw_line: str) -> None:
+        try:
+            self.backtrace_state = self._platform_handler(
+                self._config, raw_line, self.backtrace_state
+            )
+        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-except
+            self._decode_enabled = False
+            self.backtrace_state = False
+            _LOGGER.debug("Stack-trace decoding failed", exc_info=True)
+            if isinstance(exc, (EsphomeError, OSError)):
+                # The environment branch: idedata and build tree failures
+                # get the remediation hint. The fallback string is
+                # defensive; the in-tree raise sites all carry a message
+                # now, but a bare EsphomeError must not render as parens.
+                _LOGGER.warning(
+                    "Crash trace decoding unavailable: %s. "
+                    "Run 'esphome compile' for this device to enable PC decoding.",
+                    str(exc) or "build artifacts not found locally",
+                )
+            else:
+                # A decoder bug is ESPHome's problem, not the user's;
+                # don't send them to recompile a healthy build. Always
+                # name the type: a bare KeyError message reads like a
+                # raised string in the paste a bug report needs.
+                detail = type(exc).__name__
+                if msg := str(exc):
+                    detail = f"{detail}: {msg}"
+                _LOGGER.warning(
+                    'Crash trace decoding disabled: decoder for "%s" raised %s '
+                    "(this is a bug; run with -v for the traceback)",
+                    self._platform,
+                    detail,
+                )

tests/unit_tests/analyze_memory/test_build_artifacts.py

@@ -143,8 +143,8 @@ def test_cc_path_from_cxx(cxx_path: str, expected: str) -> None:
 def test_native_idedata_resolves_toolchain_tools() -> None:
     """The binutils paths are derived from the native ESP-IDF cc_path.
 
-    Without cc_path, IDEData.objdump_path raises KeyError and the memory
+    Without cc_path, IDEData.objdump_path raises EsphomeError and the
-    analysis silently degrades to no component or symbol detail.
+    memory analysis silently degrades to no component or symbol detail.
     """
     idedata = IDEData(
         {

tests/unit_tests/test_api_client.py

@@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, Mock, patch
 import pytest
 
 from esphome import api_client
-from esphome.components import esp32
 from esphome.const import (
     CONF_ENCRYPTION,
     CONF_KEY,
@@ -16,7 +15,7 @@ from esphome.const import (
     KEY_CORE,
     KEY_TARGET_PLATFORM,
 )
-from esphome.core import CORE, EsphomeError
+from esphome.core import CORE
 
 
 def test_component_shim_reexports_runtime_client() -> None:
@@ -29,135 +28,6 @@ def test_component_shim_reexports_runtime_client() -> None:
     assert api.CONF_ENCRYPTION is CONF_ENCRYPTION
 
 
-def test_decoder_swallows_esphome_error() -> None:
-    """A failing stack-trace decode must not propagate.
-
-    aioesphomeapi isolates exceptions raised by log handlers, so an
-    escaping one logs a full traceback for every line it fires on rather
-    than being reported once as an unavailable decoder.
-    """
-    config = {"esphome": {"name": "test"}}
-
-    with patch.object(
-        esp32, "process_stacktrace", side_effect=EsphomeError("no idedata")
-    ) as mock_process:
-        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
-        processor.process_line("PC: 0x4010496e")
-
-    assert mock_process.called
-    assert processor.backtrace_state is False
-
-
-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")
-
-    processor = api_client._LogLineProcessor(config, platform_handler)
-    processor.process_line("PC: 0x4010496e")
-
-    assert processor.backtrace_state is False
-
-
-def test_decoder_swallows_non_esphome_error() -> None:
-    """Decoding failures that aren't EsphomeError must be contained too.
-
-    A missing build directory surfaces as FileNotFoundError from the toolchain
-    subprocess. aioesphomeapi isolates it, so the session survives, but it logs
-    a traceback for every PC/BT line and decoding is never disabled, which
-    buries the crash dump the user is trying to read.
-    """
-    config = {"esphome": {"name": "test"}}
-
-    with patch.object(
-        esp32,
-        "process_stacktrace",
-        side_effect=FileNotFoundError(
-            2, "No such file or directory", "/build/ol/build"
-        ),
-    ) as mock_process:
-        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
-        processor.process_line("PC: 0x4010496e")
-        processor.process_line("BT0: 0x4010496e")
-
-    # Disabled after the first failure rather than retried per backtrace line.
-    assert mock_process.call_count == 1
-    assert processor.backtrace_state is False
-
-
-def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None:
-    """_run_idedata raises EsphomeError with no message; the warning
-    must show a useful explanation rather than empty parens.
-    """
-    config = {"esphome": {"name": "test"}}
-
-    with patch.object(esp32, "process_stacktrace", side_effect=EsphomeError()):
-        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
-        processor.process_line("PC: 0x4010496e")
-
-    warnings = [r.message for r in caplog.records if r.levelname == "WARNING"]
-    assert any("build artifacts not found locally" in m for m in warnings)
-    assert not any("()" in m for m in warnings)
-
-
-def test_decoder_short_circuits_after_failure() -> None:
-    """After one failure, subsequent lines must not retry the decoder.
-
-    _decode_pc shells out to the toolchain; 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"}}
-
-    with patch.object(
-        esp32, "process_stacktrace", side_effect=EsphomeError("no idedata")
-    ) as mock_process:
-        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
-        processor.process_line("PC: 0x4010496e")
-        processor.process_line("BT0: 0x4010496e")
-        processor.process_line("BT1: 0x401049aa")
-
-    assert mock_process.call_count == 1
-
-
-def test_decoder_threads_backtrace_state() -> None:
-    """When decoding succeeds, backtrace_state is threaded across calls."""
-    config = {"esphome": {"name": "test"}}
-
-    with patch.object(
-        esp32, "process_stacktrace", side_effect=[True, False]
-    ) as mock_process:
-        processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
-        processor.process_line(">>>stack>>>")
-        assert processor.backtrace_state is True
-        processor.process_line("<<<stack<<<")
-        assert processor.backtrace_state is False
-
-    assert not mock_process.call_args_list[0].args[-1]
-    assert mock_process.call_args_list[1].args[-1]
-
-
-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]] = []
-
-    def platform_handler(cfg, line, state):
-        calls.append((cfg, line, state))
-        return True
-
-    processor = api_client._LogLineProcessor(config, platform_handler)
-
-    with patch.object(esp32, "process_stacktrace") as mock_generic:
-        processor.process_line("BT0: 0x4010496e")
-
-    assert calls == [(config, "BT0: 0x4010496e", False)]
-    assert mock_generic.called is False
-    assert processor.backtrace_state is True
-
-
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("extra_config", "expected_deep_sleep"),
@@ -194,6 +64,7 @@ async def test_async_run_logs_full_flow(caplog) -> None:
     stop() cleanup in the finally block.
     """
     caplog.set_level("INFO", logger="esphome.api_client")
+    caplog.set_level("INFO", logger="esphome.platform_hooks")
     CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "host"}
     config = {
         "esphome": {"name": "test"},
@@ -233,7 +104,7 @@ async def test_async_run_logs_full_flow(caplog) -> None:
     assert mock_client.call_args.kwargs["noise_psk"] == "psk123"
     assert mock_client.call_args.kwargs["addresses"] == ["1.2.3.4", "5.6.7.8"]
     assert "1.2.3.4 or 5.6.7.8" in caplog.text
-    # host has no stacktrace analyzer; the fallback message is logged.
+    # host has no stacktrace analyzer; the notice fires at session start.
     assert "Stacktrace analysis is unavailable" in caplog.text
     # The log message was printed with a timestamp prefix.
     assert any("hello world" in line for line in printed)

tests/unit_tests/test_lazy_imports.py

@@ -37,16 +37,21 @@ HEAVY_MODULES = (
 # existence guard and the leak check must watch the same list.
 FAST_PATH_HEAVY_MODULES = HEAVY_MODULES + ("esphome.components.esp32",)
 
+# Heavy only for modules that must not know about the API transport;
+# in the existence guard so a rename can't silently no-op its check.
+API_HEAVY_MODULES = ("aioesphomeapi",)
 
-def _leaked_heavy_modules(module: str) -> str:
+
+def _leaked_heavy_modules(module: str, extra: tuple[str, ...] = ()) -> str:
     """Import ``module`` in a subprocess and report the heavy modules it pulled.
 
     Any ``esphome.components.*`` package counts as heavy: executing a
     component package drags in codegen/validation machinery by design.
+    ``extra`` adds modules that are heavy for this caller specifically.
     """
     check = (
         f"import sys; import {module}; "
-        f"leaked = [m for m in {HEAVY_MODULES!r} if m in sys.modules]; "
+        f"leaked = [m for m in {HEAVY_MODULES + extra!r} if m in sys.modules]; "
         "leaked += [m for m in sys.modules if m.startswith('esphome.components.')]; "
         "print(','.join(leaked))"
     )
@@ -72,7 +77,7 @@ def test_main_module_does_not_import_heavy_modules() -> None:
 
 def test_watched_heavy_modules_exist() -> None:
     """A renamed heavy module would silently disable the leak checks."""
-    for module in FAST_PATH_HEAVY_MODULES:
+    for module in FAST_PATH_HEAVY_MODULES + API_HEAVY_MODULES:
         assert importlib.util.find_spec(module) is not None, (
             f"{module} no longer resolves; update the heavy-module lists"
         )
@@ -145,6 +150,21 @@ def test_api_client_does_not_import_heavy_modules() -> None:
     )
 
 
+def test_stacktrace_does_not_import_heavy_modules() -> None:
+    """``esphome.stacktrace`` guards its own docstring's contract.
+
+    Both log paths construct a LogLineProcessor before streaming
+    starts; importing the module must not pull in aioesphomeapi or
+    any platform package.
+    """
+    leaked = _leaked_heavy_modules("esphome.stacktrace", extra=API_HEAVY_MODULES)
+    assert not leaked, (
+        f"esphome.stacktrace imports heavy modules at top level: {leaked}. "
+        "The logs fast path skips validation; importing the validation "
+        "stack anyway defeats the validated-config cache."
+    )
+
+
 def test_espidf_toolchain_does_not_import_heavy_modules() -> None:
     """The esp-idf upload path must not pull the esp32 package back in.
 

tests/unit_tests/test_main.py

@@ -5923,6 +5923,38 @@ def test_run_miniterm_backtrace_state_maintained() -> None:
     assert backtrace_states[3][1] is True
 
 
+def test_run_miniterm_decoder_failure_keeps_streaming(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A decoder exception must not kill serial streaming.
+
+    This is the serial path's gain from sharing LogLineProcessor: before
+    the lift a decoder exception propagated out of the read loop.
+    """
+    chunk = b"PC: 0x4010496e\r\nBT0: 0x4010496e\r\nstill streaming\r\n"
+    mock_serial = MockSerial([chunk, MOCK_SERIAL_END])
+
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
+    config = {
+        CONF_LOGGER: {
+            CONF_BAUD_RATE: 115200,
+            "deassert_rts_dtr": False,
+        }
+    }
+    args = MockArgs()
+
+    decoder = Mock(side_effect=EsphomeError("no idedata"))
+    with (
+        patch("serial.Serial", return_value=mock_serial),
+        patch.object(esp32, "process_stacktrace", decoder),
+    ):
+        run_miniterm(config, "/dev/ttyUSB0", args)
+
+    # The failure is contained and latched; streaming continued to EOF.
+    assert decoder.call_count == 1
+    assert "Crash trace decoding unavailable" in caplog.text
+
+
 def test_run_miniterm_handles_empty_reads(
     capfd: CaptureFixture[str],
 ) -> None:

tests/unit_tests/test_platformio_toolchain.py

@@ -278,15 +278,56 @@ def test_run_idedata_raises_on_no_json(
 def test_run_idedata_raises_on_invalid_json(
     setup_core: Path, mock_run_platformio_cli_run: Mock
 ) -> None:
-    """Test _run_idedata raises on malformed JSON."""
+    """Malformed JSON is the environment (garbage stdout), so it must
+    surface as EsphomeError and get the recompile hint downstream.
+    """
     config = {"name": "test"}
     mock_run_platformio_cli_run.return_value = '{"invalid": json"}'
 
-    # The ValueError from json.loads is re-raised
+    with pytest.raises(EsphomeError):
-    with pytest.raises(ValueError):
         toolchain._run_idedata(config)
 
 
+def test_run_idedata_raises_on_launch_failure(
+    setup_core: Path, mock_run_platformio_cli_run: Mock
+) -> None:
+    """A failed platformio launch returns its exit code as an int; that
+    must surface as EsphomeError, not a TypeError from re.search.
+    """
+    config = {"name": "test"}
+    mock_run_platformio_cli_run.return_value = 1
+
+    with pytest.raises(EsphomeError):
+        toolchain._run_idedata(config)
+
+
+def test_idedata_missing_prog_path_raises_esphome_error(setup_core: Path) -> None:
+    """A stale cached idedata JSON without prog_path is the build tree's
+    fault; it must surface as EsphomeError, not a KeyError.
+    """
+    with pytest.raises(EsphomeError):
+        _ = toolchain.IDEData({}).firmware_elf_path
+
+
+def test_idedata_missing_flash_image_field_raises_esphome_error(
+    setup_core: Path,
+) -> None:
+    """A cached idedata whose flash image entries lost a field must
+    classify as an environment error too, not a raw KeyError.
+    """
+    idedata = toolchain.IDEData({"extra": {"flash_images": [{"offset": "0x1000"}]}})
+    with pytest.raises(EsphomeError):
+        _ = idedata.extra_flash_images
+
+
+def test_idedata_null_section_raises_esphome_error(setup_core: Path) -> None:
+    """A section that is null instead of absent must classify the same
+    as a missing key instead of escaping as TypeError.
+    """
+    with pytest.raises(EsphomeError):
+        _ = toolchain.IDEData({"extra": None}).extra_flash_images
+
+
 def test_run_platformio_cli_sets_environment_variables(
     setup_core: Path, mock_run_external_process: Mock
 ) -> None:

tests/unit_tests/test_stacktrace.py

@@ -0,0 +1,151 @@
+"""Tests for esphome.stacktrace."""
+
+from __future__ import annotations
+
+from unittest.mock import Mock, patch
+
+from esphome import stacktrace
+from esphome.const import PLATFORM_BK72XX, PLATFORM_ESP32, PLATFORM_ESP8266
+from esphome.core import EsphomeError
+
+CONFIG = {"esphome": {"name": "test"}}
+
+
+def _run(
+    handler,
+    platform: str = PLATFORM_ESP32,
+    lines: tuple[str, ...] = ("PC: 0x4010496e",),
+) -> stacktrace.LogLineProcessor:
+    """Processor with the resolver stubbed, fed the given lines."""
+    with patch.object(
+        stacktrace.platform_hooks, "get_stacktrace_handler", return_value=handler
+    ):
+        processor = stacktrace.LogLineProcessor(CONFIG, platform)
+    for line in lines:
+        processor.process_line(line)
+    return processor
+
+
+def _fed(handler) -> list[str]:
+    return [call.args[1] for call in handler.call_args_list]
+
+
+def _warnings(caplog) -> list[str]:
+    return [r.message for r in caplog.records if r.levelname == "WARNING"]
+
+
+def test_decoder_contains_failures_and_short_circuits() -> None:
+    """One decode failure is contained and never retried.
+
+    aioesphomeapi isolates exceptions raised by log handlers, so an
+    escaping one logs a full traceback for every line it fires on; and
+    _decode_pc shells out to the toolchain, so retrying it per backtrace
+    line would stall streaming.
+    """
+    handler = Mock(side_effect=EsphomeError("no idedata"))
+    processor = _run(
+        handler, lines=("PC: 0x4010496e", "BT0: 0x4010496e", "BT1: 0x401049aa")
+    )
+
+    assert handler.call_count == 1
+    assert processor.backtrace_state is False
+
+
+def test_resolution_failure_is_contained(caplog) -> None:
+    """A platform package broken in an unanticipated way must not kill
+    the session; decoding degrades with a warning like any other failure.
+    """
+    with patch.object(
+        stacktrace.platform_hooks,
+        "get_stacktrace_handler",
+        side_effect=RuntimeError("boom"),
+    ):
+        processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_ESP32)
+        processor.process_line("PC: 0x4010496e")
+
+    assert processor.backtrace_state is False
+    assert any("could not be loaded" in m for m in _warnings(caplog))
+
+
+def test_decoder_swallows_os_error_with_remediation_hint(caplog) -> None:
+    """Decoding failures that aren't EsphomeError must be contained too.
+
+    A missing build directory surfaces as an OSError; that is the
+    user's environment, not a decoder bug, so it disables decoding
+    like an EsphomeError does and keeps the recompile hint.
+    """
+    handler = Mock(
+        side_effect=FileNotFoundError(2, "No such file or directory", "/build")
+    )
+    processor = _run(handler, lines=("PC: 0x4010496e", "BT0: 0x4010496e"))
+
+    assert handler.call_count == 1
+    assert processor.backtrace_state is False
+    warnings = _warnings(caplog)
+    assert any("esphome compile" in m for m in warnings)
+    assert not any("this is a bug" in m for m in warnings)
+
+
+def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None:
+    """A message-less EsphomeError must show a useful explanation.
+
+    Defensive: the in-tree idedata raise sites all carry a message now,
+    but a bare EsphomeError from elsewhere must not render as parens.
+    """
+    _run(Mock(side_effect=EsphomeError()))
+
+    warnings = _warnings(caplog)
+    assert any("build artifacts not found locally" in m for m in warnings)
+    assert not any("()" in m for m in warnings)
+
+
+def test_decoder_bug_with_empty_message_names_the_type(caplog) -> None:
+    """A zero-message decoder bug must not masquerade as missing artifacts.
+
+    The recompile hint is only right for EsphomeError from _run_idedata;
+    anything else is ESPHome's own bug and says so instead of sending
+    the user down a dead-end remediation path.
+    """
+    _run(Mock(side_effect=IndexError()))
+
+    warnings = _warnings(caplog)
+    assert any("IndexError" in m and "this is a bug" in m for m in warnings)
+    assert not any("esphome compile" in m for m in warnings)
+
+
+def test_decoder_bug_warning_keeps_the_type_with_a_message(caplog) -> None:
+    """The type must survive a non-empty message; a bare KeyError message
+    like 'prog_path' reads as a raised string in a bug report paste.
+    """
+    _run(Mock(side_effect=KeyError("prog_path")))
+
+    warnings = _warnings(caplog)
+    assert any("KeyError: 'prog_path'" in m for m in warnings)
+
+
+def test_state_threads_between_lines() -> None:
+    """backtrace_state carries from one decoded line to the next."""
+    handler = Mock(side_effect=[True, True])
+    processor = _run(
+        handler,
+        platform=PLATFORM_ESP8266,
+        lines=(">>>stack>>>", "3ffffe10: 40201234 3ffe8410 00000000 40201000"),
+    )
+
+    assert _fed(handler) == [
+        ">>>stack>>>",
+        "3ffffe10: 40201234 3ffe8410 00000000 40201000",
+    ]
+    assert handler.call_args_list[0].args[2] is False
+    assert handler.call_args_list[1].args[2] is True
+    assert processor.backtrace_state is True
+
+
+def test_no_analyzer_disables_decoding(caplog) -> None:
+    """Platforms without an analyzer report at session start and stay quiet."""
+    caplog.set_level("INFO", logger="esphome.platform_hooks")
+    processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_BK72XX)
+    processor.process_line("PC: 0x40104960")
+
+    assert "Stacktrace analysis is unavailable" in caplog.text
+    assert processor.backtrace_state is False