Make TriggerNotTriggeredAction a regular function

homeassistant/components/automation/__init__.py

@@ -976,6 +976,51 @@ class AutomationEntity(BaseAutomationEntity, RestoreEntity):
             return None
         return await self.async_trigger(run_variables, context, skip_condition)
 
+    @callback
+    def _handle_not_triggered(
+        self,
+        run_variables: dict[str, Any],
+        info: trigger_helper.NotTriggeredInfo,
+        context: Context | None = None,
+    ) -> None:
+        """Record a trace for a trigger that evaluated a change but did not fire.
+
+        This is the diagnostic sibling of async_trigger: a trigger calls it - in
+        certain interesting cases - when it does not run the action, so the user
+        can see in the trace why the automation was not triggered.
+        """
+        if not self._is_enabled:
+            return
+
+        # Create a new context referring to the old context.
+        parent_id = None if context is None else context.id
+        trigger_context = Context(parent_id=parent_id)
+
+        with trace_automation(
+            self.hass,
+            self.unique_id,
+            self.raw_config,
+            self._blueprint_inputs,
+            trigger_context,
+            self._trace_config,
+            not_triggered=True,
+        ) as automation_trace:
+            automation_trace.set_trace(trace_get())
+
+            trigger_description = run_variables.get("trigger", {}).get("description")
+            automation_trace.set_trigger_description(trigger_description)
+
+            # Record the trigger and its diagnostics as the trigger step.
+            if "idx" in run_variables.get("trigger", {}):
+                trigger_path = f"trigger/{run_variables['trigger']['idx']}"
+            else:
+                trigger_path = "trigger"
+            trace_element = TraceElement(run_variables, trigger_path)
+            trace_element.set_result(**info.as_dict())
+            trace_append_element(trace_element)
+
+            script_execution_set("not_triggered")
+
     async def _async_attach_triggers(
         self, home_assistant_start: bool
     ) -> Callable[[], None] | None:
@@ -1004,6 +1049,7 @@ class AutomationEntity(BaseAutomationEntity, RestoreEntity):
             self._log_callback,
             home_assistant_start,
             variables,
+            did_not_trigger=self._handle_not_triggered,
         )
 
 

homeassistant/components/automation/trace.py

@@ -26,10 +26,13 @@ class AutomationTrace(ActionTrace):
         config: ConfigType | None,
         blueprint_inputs: ConfigType | None,
         context: Context,
+        *,
+        not_triggered: bool = False,
     ) -> None:
         """Container for automation trace."""
         super().__init__(item_id, config, blueprint_inputs, context)
         self._trigger_description: str | None = None
+        self.not_triggered = not_triggered
 
     def set_trigger_description(self, trigger: str) -> None:
         """Set trigger description."""
@@ -53,9 +56,13 @@ def trace_automation(
     blueprint_inputs: ConfigType | None,
     context: Context,
     trace_config: ConfigType,
+    *,
+    not_triggered: bool = False,
 ) -> Generator[AutomationTrace]:
     """Trace action execution of automation with automation_id."""
-    trace = AutomationTrace(automation_id, config, blueprint_inputs, context)
+    trace = AutomationTrace(
+        automation_id, config, blueprint_inputs, context, not_triggered=not_triggered
+    )
     async_store_trace(hass, trace, trace_config[CONF_STORED_TRACES])
 
     try:

homeassistant/components/calendar/trigger.py

@@ -27,7 +27,12 @@ from homeassistant.helpers.event import (
     async_track_time_interval,
 )
 from homeassistant.helpers.target import TargetEntityChangeTracker, TargetSelection
-from homeassistant.helpers.trigger import Trigger, TriggerActionRunner, TriggerConfig
+from homeassistant.helpers.trigger import (
+    Trigger,
+    TriggerActionRunner,
+    TriggerConfig,
+    TriggerNotTriggeredReporter,
+)
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.util import dt as dt_util
 
@@ -393,7 +398,9 @@ class SingleEntityEventTrigger(Trigger):
         self._options = config.options
 
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach a trigger."""
 
@@ -444,7 +451,9 @@ class EventTrigger(Trigger):
         self._options = config.options
 
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach a trigger."""
 

homeassistant/components/timer/trigger.py

@@ -20,6 +20,7 @@ from homeassistant.helpers.trigger import (
     Trigger,
     TriggerActionRunner,
     TriggerConfig,
+    TriggerNotTriggeredReporter,
     make_entity_target_state_trigger,
 )
 from homeassistant.helpers.typing import ConfigType
@@ -66,7 +67,9 @@ class TimeRemainingTrigger(Trigger):
 
     @override
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach the trigger to an action runner."""
         scheduled: dict[str, CALLBACK_TYPE] = {}

homeassistant/components/todo/trigger.py

@@ -16,7 +16,12 @@ from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.entity_component import EntityComponent
 from homeassistant.helpers.target import TargetEntityChangeTracker, TargetSelection
-from homeassistant.helpers.trigger import Trigger, TriggerActionRunner, TriggerConfig
+from homeassistant.helpers.trigger import (
+    Trigger,
+    TriggerActionRunner,
+    TriggerConfig,
+    TriggerNotTriggeredReporter,
+)
 from homeassistant.helpers.typing import ConfigType
 
 from . import TodoItem, TodoListEntity
@@ -140,7 +145,9 @@ class ItemTriggerBase(Trigger, abc.ABC):
         self._target = config.target
 
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach a trigger."""
 

homeassistant/components/trace/__init__.py

@@ -58,8 +58,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         try:
             await store.async_save(
                 {
-                    key: list(traces.values())
-                    for key, traces in hass.data[DATA_TRACE].items()
+                    key: list(trace_bucket.all_traces())
+                    for key, trace_bucket in hass.data[DATA_TRACE].items()
                 }
             )
         except HomeAssistantError as exc:

homeassistant/components/trace/models.py

@@ -2,6 +2,8 @@
 
 import abc
 from collections import deque
+from collections.abc import Iterator
+from dataclasses import dataclass
 import datetime as dt
 from typing import Any
 
@@ -16,7 +18,7 @@ from homeassistant.helpers.trace import (
 from homeassistant.util import dt as dt_util, uuid as uuid_util
 from homeassistant.util.limited_size_dict import LimitedSizeDict
 
-type TraceData = dict[str, LimitedSizeDict[str, BaseTrace]]
+type TraceData = dict[str, TraceBuckets]
 
 
 class BaseTrace(abc.ABC):
@@ -25,6 +27,9 @@ class BaseTrace(abc.ABC):
     context: Context
     key: str
     run_id: str
+    # True for traces recording that a trigger evaluated a relevant change but
+    # did not fire. These are counted separately from actual runs.
+    not_triggered: bool = False
 
     def as_dict(self) -> dict[str, Any]:
         """Return an dictionary version of this ActionTrace for saving."""
@@ -42,6 +47,27 @@ class BaseTrace(abc.ABC):
         """Return a brief dictionary version of this ActionTrace."""
 
 
+@dataclass(slots=True)
+class TraceBuckets:
+    """The run and not-triggered traces for a single script or automation.
+
+    Not-triggered traces (a trigger evaluated a change but did not fire) are
+    counted and size-limited separately so they never evict actual run traces.
+    """
+
+    runs: LimitedSizeDict[str, BaseTrace]
+    not_triggered: LimitedSizeDict[str, BaseTrace]
+
+    def bucket(self, not_triggered: bool) -> LimitedSizeDict[str, BaseTrace]:
+        """Return the bucket holding traces of the requested kind."""
+        return self.not_triggered if not_triggered else self.runs
+
+    def all_traces(self) -> Iterator[BaseTrace]:
+        """Yield all traces, runs first then not-triggered."""
+        yield from self.runs.values()
+        yield from self.not_triggered.values()
+
+
 class ActionTrace(BaseTrace):
     """Base container for a script or automation trace."""
 
@@ -123,7 +149,7 @@ class ActionTrace(BaseTrace):
             last_step = list(self._trace)[-1]
         domain, item_id = self.key.split(".", 1)
 
-        result = {
+        result: dict[str, Any] = {
             "last_step": last_step,
             "run_id": self.run_id,
             "state": self._state,
@@ -135,6 +161,8 @@ class ActionTrace(BaseTrace):
             "domain": domain,
             "item_id": item_id,
         }
+        if self.not_triggered:
+            result["not_triggered"] = True
         if self._error is not None:
             result["error"] = str(self._error)
 
@@ -159,6 +187,7 @@ class RestoredTrace(BaseTrace):
         self.context = context
         self.key = f"{extended_dict['domain']}.{extended_dict['item_id']}"
         self.run_id = extended_dict["run_id"]
+        self.not_triggered = short_dict.get("not_triggered", False)
         self._dict = extended_dict
         self._short_dict = short_dict
 

homeassistant/components/trace/util.py

@@ -9,7 +9,7 @@ from homeassistant.exceptions import HomeAssistantError
 from homeassistant.util.limited_size_dict import LimitedSizeDict
 
 from .const import DATA_TRACE, DATA_TRACE_STORE, DATA_TRACES_RESTORED
-from .models import ActionTrace, BaseTrace, RestoredTrace, TraceData
+from .models import ActionTrace, BaseTrace, RestoredTrace, TraceBuckets, TraceData
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -21,7 +21,9 @@ async def async_get_trace(
     # Restore saved traces if not done
     await async_restore_traces(hass)
 
-    return hass.data[DATA_TRACE][key][run_id].as_extended_dict()
+    trace_bucket = hass.data[DATA_TRACE][key]
+    trace = trace_bucket.runs.get(run_id) or trace_bucket.not_triggered[run_id]
+    return trace.as_extended_dict()
 
 
 async def async_list_contexts(
@@ -31,7 +33,7 @@ async def async_list_contexts(
     # Restore saved traces if not done
     await async_restore_traces(hass)
 
-    values: Mapping[str, LimitedSizeDict[str, BaseTrace] | None] | TraceData
+    values: Mapping[str, TraceBuckets | None] | TraceData
     if key is not None:
         values = {key: hass.data[DATA_TRACE].get(key)}
     else:
@@ -44,16 +46,16 @@ async def async_list_contexts(
 
     return {
         trace.context.id: _trace_id(trace.run_id, key)
-        for key, traces in values.items()
-        if traces is not None
-        for trace in traces.values()
+        for key, trace_bucket in values.items()
+        if trace_bucket is not None
+        for trace in trace_bucket.all_traces()
     }
 
 
 def _get_debug_traces(hass: HomeAssistant, key: str) -> list[dict[str, Any]]:
     """Return a serializable list of debug traces for a script or automation."""
-    if traces_for_key := hass.data[DATA_TRACE].get(key):
-        return [trace.as_short_dict() for trace in traces_for_key.values()]
+    if trace_bucket := hass.data[DATA_TRACE].get(key):
+        return [trace.as_short_dict() for trace in trace_bucket.all_traces()]
     return []
 
 
@@ -79,14 +81,23 @@ async def async_list_traces(
 def async_store_trace(
     hass: HomeAssistant, trace: ActionTrace, stored_traces: int
 ) -> None:
-    """Store a trace if its key is valid."""
+    """Store a trace if its key is valid.
+
+    Run traces and not-triggered traces are kept in separate, independently
+    size-limited buckets so a flood of not-triggered traces never evicts runs.
+    """
     if key := trace.key:
         traces = hass.data[DATA_TRACE]
         if key not in traces:
-            traces[key] = LimitedSizeDict(size_limit=stored_traces)
-        else:
-            traces[key].size_limit = stored_traces
-        traces[key][trace.run_id] = trace
+            traces[key] = TraceBuckets(
+                runs=LimitedSizeDict(size_limit=stored_traces),
+                not_triggered=LimitedSizeDict(size_limit=stored_traces),
+            )
+        trace_bucket = traces[key]
+        trace_bucket.runs.size_limit = stored_traces
+        trace_bucket.not_triggered.size_limit = stored_traces
+        bucket = trace_bucket.bucket(trace.not_triggered)
+        bucket[trace.run_id] = trace
 
 
 def _async_store_restored_trace(hass: HomeAssistant, trace: RestoredTrace) -> None:
@@ -94,9 +105,12 @@ def _async_store_restored_trace(hass: HomeAssistant, trace: RestoredTrace) -> No
     key = trace.key
     traces = hass.data[DATA_TRACE]
     if key not in traces:
-        traces[key] = LimitedSizeDict()
-    traces[key][trace.run_id] = trace
-    traces[key].move_to_end(trace.run_id, last=False)
+        traces[key] = TraceBuckets(
+            runs=LimitedSizeDict(), not_triggered=LimitedSizeDict()
+        )
+    bucket = traces[key].bucket(trace.not_triggered)
+    bucket[trace.run_id] = trace
+    bucket.move_to_end(trace.run_id, last=False)
 
 
 async def async_restore_traces(hass: HomeAssistant) -> None:
@@ -116,17 +130,18 @@ async def async_restore_traces(hass: HomeAssistant) -> None:
     for key, traces in restored_traces.items():
         # Add stored traces in reversed order to prioritize the newest traces
         for json_trace in reversed(traces):
-            if (
-                (stored_traces := hass.data[DATA_TRACE].get(key))
-                and stored_traces.size_limit is not None
-                and len(stored_traces) >= stored_traces.size_limit
-            ):
-                break
-
             try:
                 trace = RestoredTrace(json_trace)
             # Catch any exception to not blow up if the stored trace is invalid
             except Exception:
                 _LOGGER.exception("Failed to restore trace")
                 continue
+
+            # Runs and not-triggered traces are capped independently, so check
+            # the bucket this trace belongs to rather than breaking the loop.
+            if (trace_bucket := hass.data[DATA_TRACE].get(key)) is not None:
+                bucket = trace_bucket.bucket(trace.not_triggered)
+                if bucket.size_limit is not None and len(bucket) >= bucket.size_limit:
+                    continue
+
             _async_store_restored_trace(hass, trace)

homeassistant/components/zone/trigger.py

@@ -39,6 +39,7 @@ from homeassistant.helpers.trigger import (
     Trigger,
     TriggerActionRunner,
     TriggerConfig,
+    TriggerNotTriggeredReporter,
 )
 from homeassistant.helpers.typing import ConfigType
 
@@ -127,7 +128,9 @@ class LegacyZoneTrigger(Trigger):
         self._options = config.options
 
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Listen for state changes based on configuration."""
         entity_id: list[str] = self._options[CONF_ENTITY_ID]

homeassistant/components/zwave_js/triggers/event.py

@@ -20,7 +20,12 @@ from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv, device_registry as dr
 from homeassistant.helpers.automation import move_top_level_schema_fields_to_options
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
-from homeassistant.helpers.trigger import Trigger, TriggerActionRunner, TriggerConfig
+from homeassistant.helpers.trigger import (
+    Trigger,
+    TriggerActionRunner,
+    TriggerConfig,
+    TriggerNotTriggeredReporter,
+)
 from homeassistant.helpers.typing import ConfigType
 
 from ..const import (
@@ -168,7 +173,9 @@ class EventTrigger(Trigger):
         self._options = config.options
 
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach a trigger."""
         dev_reg = dr.async_get(self._hass)

homeassistant/components/zwave_js/triggers/value_updated.py

@@ -14,7 +14,12 @@ from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv, device_registry as dr
 from homeassistant.helpers.automation import move_top_level_schema_fields_to_options
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
-from homeassistant.helpers.trigger import Trigger, TriggerActionRunner, TriggerConfig
+from homeassistant.helpers.trigger import (
+    Trigger,
+    TriggerActionRunner,
+    TriggerConfig,
+    TriggerNotTriggeredReporter,
+)
 from homeassistant.helpers.typing import ConfigType
 
 from ..config_validation import VALUE_SCHEMA
@@ -225,7 +230,9 @@ class ValueUpdatedTrigger(Trigger):
         self._options = config.options
 
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach a trigger."""
         return await async_attach_trigger(self._hass, self._options, run_action)

homeassistant/helpers/trigger.py

@@ -4,6 +4,7 @@ import abc
 import asyncio
 from collections import defaultdict
 from collections.abc import Callable, Coroutine, Iterable, Mapping
+from contextvars import copy_context
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 import functools
@@ -302,8 +303,15 @@ class Trigger(abc.ABC):
         self,
         action: TriggerAction,
         action_payload_builder: TriggerActionPayloadBuilder,
+        *,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
-        """Attach the trigger to an action."""
+        """Attach the trigger to an action.
+
+        The optional ``did_not_trigger`` reporter is the sibling of the action
+        runner: triggers may call it - in certain interesting cases - when they
+        evaluate a relevant change but decide not to fire.
+        """
 
         @callback
         def run_action(
@@ -316,11 +324,13 @@ class Trigger(abc.ABC):
             payload = action_payload_builder(extra_trigger_payload, description)
             return self._hass.async_create_task(action(payload, context))
 
-        return await self.async_attach_runner(run_action)
+        return await self.async_attach_runner(run_action, did_not_trigger)
 
     @abc.abstractmethod
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach the trigger to an action runner."""
 
@@ -527,7 +537,9 @@ class EntityTriggerBase(Trigger):
 
     @override
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach the trigger to an action runner."""
 
@@ -577,11 +589,25 @@ class EntityTriggerBase(Trigger):
             if not from_state or not to_state:
                 return
 
+            @callback
+            def report_not_triggered(reason: str, **data: Any) -> None:
+                """Report why this state change did not fire the trigger."""
+                if did_not_trigger is None:
+                    return
+                did_not_trigger(
+                    NotTriggeredInfo(reason=reason, data=data), event.context
+                )
+
             # The trigger should never fire if the new state is excluded
             # or not a target state.
             if to_state.state in self._excluded_states or not self.is_valid_state(
                 to_state
             ):
+                report_not_triggered(
+                    "new_state_not_a_match",
+                    entity_id=entity_id,
+                    to_state=to_state.state,
+                )
                 return
 
             # The trigger should never fire if the origin state is excluded
@@ -590,6 +616,12 @@ class EntityTriggerBase(Trigger):
                 from_state.state in self._excluded_from_states
                 or not self.is_valid_transition(from_state, to_state)
             ):
+                report_not_triggered(
+                    "transition_not_a_match",
+                    entity_id=entity_id,
+                    from_state=from_state.state,
+                    to_state=to_state.state,
+                )
                 return
 
             # Count against the targeted entity states as of this event, not
@@ -603,6 +635,9 @@ class EntityTriggerBase(Trigger):
                     target_state_change_data.targeted_entity_states,
                 )
                 if matches != included:
+                    report_not_triggered(
+                        "not_all_targets_matched", matches=matches, included=included
+                    )
                     return
             elif behavior == BEHAVIOR_FIRST:
                 # Note: It's enough to test for exactly 1 match here because if there
@@ -613,6 +648,9 @@ class EntityTriggerBase(Trigger):
                     target_state_change_data.targeted_entity_states,
                 )
                 if matches != 1:
+                    report_not_triggered(
+                        "behavior_first_not_satisfied", matches=matches
+                    )
                     return
 
             @callback
@@ -1205,6 +1243,27 @@ class TriggerConfig:
     options: dict[str, Any] | None = None
 
 
+@dataclass(slots=True, frozen=True)
+class NotTriggeredInfo:
+    """Diagnostics describing why a trigger evaluated a change but did not fire.
+
+    Passed by a trigger to its ``did_not_trigger`` reporter, the sibling of the
+    action runner that is called - in certain interesting cases - when the
+    trigger does not fire. ``reason`` is a stable, machine-readable code; the
+    optional ``data`` carries the evaluated context for the trace.
+    """
+
+    reason: str
+    data: Mapping[str, Any] | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        """Return a JSON-serializable dict for storing in a trace."""
+        result: dict[str, Any] = {"reason": self.reason}
+        if self.data is not None:
+            result["data"] = dict(self.data)
+        return result
+
+
 class TriggerActionRunner(Protocol):
     """Protocol type for the trigger action runner helper callback."""
 
@@ -1222,6 +1281,39 @@ class TriggerActionRunner(Protocol):
         """
 
 
+class TriggerNotTriggeredReporter(Protocol):
+    """Protocol type for the did_not_trigger reporter passed to a trigger runner.
+
+    A trigger calls this to report that it evaluated a relevant change but
+    decided not to fire, supplying diagnostics for tracing.
+    """
+
+    @callback
+    def __call__(
+        self,
+        info: NotTriggeredInfo,
+        context: Context | None = None,
+    ) -> None:
+        """Report that the trigger did not fire."""
+
+
+class TriggerNotTriggeredAction(Protocol):
+    """Protocol type for the did_not_trigger consumer callback.
+
+    Sibling of the action callback. Invoked - instead of the action - when a
+    trigger evaluated a relevant change but reported it did not fire.
+    """
+
+    @callback
+    def __call__(
+        self,
+        run_variables: dict[str, Any],
+        info: NotTriggeredInfo,
+        context: Context | None = None,
+    ) -> None:
+        """Define did_not_trigger consumer callback type."""
+
+
 class TriggerActionPayloadBuilder(Protocol):
     """Protocol type for the trigger action payload builder."""
 
@@ -1506,6 +1598,7 @@ async def _async_attach_trigger_cls(
     conf: ConfigType,
     action: Callable,
     trigger_info: TriggerInfo,
+    did_not_trigger: TriggerNotTriggeredAction | None = None,
 ) -> CALLBACK_TYPE:
     """Initialize a new Trigger class and attach it."""
 
@@ -1526,6 +1619,26 @@ async def _async_attach_trigger_cls(
             payload.update(trigger_variables.async_render(hass, payload))
         return payload
 
+    report_not_triggered: TriggerNotTriggeredReporter | None = None
+    if did_not_trigger is not None:
+        not_triggered_action = did_not_trigger
+
+        @callback
+        def report_not_triggered(
+            info: NotTriggeredInfo, context: Context | None = None
+        ) -> None:
+            """Forward a did-not-fire report to the consumer."""
+            run_variables = {
+                "trigger": {
+                    **trigger_info["trigger_data"],
+                    CONF_PLATFORM: trigger_key,
+                }
+            }
+            # The consumer records a trace using the trace context variables.
+            # Run it in a copied context so it does not disturb the trace of the
+            # run that produced this state change (e.g. a chained automation).
+            copy_context().run(not_triggered_action, run_variables, info, context)
+
     # Wrap sync action so that it is always async.
     # This simplifies the Trigger action runner interface by
     # always returning a coroutine, removing the need for
@@ -1564,7 +1677,9 @@ async def _async_attach_trigger_cls(
             options=conf.get(CONF_OPTIONS),
         ),
     )
-    return await trigger.async_attach_action(action, action_payload_builder)
+    return await trigger.async_attach_action(
+        action, action_payload_builder, did_not_trigger=report_not_triggered
+    )
 
 
 async def async_initialize_triggers(
@@ -1576,8 +1691,14 @@ async def async_initialize_triggers(
     log_cb: Callable,
     home_assistant_start: bool = False,
     variables: TemplateVarsType = None,
+    did_not_trigger: TriggerNotTriggeredAction | None = None,
 ) -> CALLBACK_TYPE | None:
-    """Initialize triggers."""
+    """Initialize triggers.
+
+    The optional ``did_not_trigger`` consumer is the sibling of ``action``,
+    invoked - for new-style triggers that support it - when a trigger evaluates
+    a relevant change but reports it did not fire. Old-style triggers ignore it.
+    """
     triggers: list[asyncio.Task[CALLBACK_TYPE]] = []
     for idx, conf in enumerate(trigger_config):
         # Skip triggers that are not enabled
@@ -1613,7 +1734,7 @@ async def async_initialize_triggers(
             )
             trigger_cls = trigger_descriptors[relative_trigger_key]
             coro = _async_attach_trigger_cls(
-                hass, trigger_cls, trigger_key, conf, action, info
+                hass, trigger_cls, trigger_key, conf, action, info, did_not_trigger
             )
         else:
             action_wrapper = _trigger_action_wrapper(hass, action, conf)

tests/components/automation/test_init.py

@@ -32,8 +32,10 @@ from homeassistant.const import (
     STATE_UNAVAILABLE,
 )
 from homeassistant.core import (
+    CALLBACK_TYPE,
     Context,
     CoreState,
+    Event,
     HomeAssistant,
     ServiceCall,
     State,
@@ -55,16 +57,26 @@ from homeassistant.helpers.script import (
     SCRIPT_MODE_SINGLE,
     _async_stop_scripts_at_shutdown,
 )
+from homeassistant.helpers.trigger import (
+    NotTriggeredInfo,
+    Trigger,
+    TriggerActionRunner,
+    TriggerNotTriggeredReporter,
+)
+from homeassistant.helpers.typing import ConfigType
 from homeassistant.setup import async_setup_component
 from homeassistant.util import dt as dt_util, yaml as yaml_util
 
 from tests.common import (
     MockConfigEntry,
+    MockModule,
     MockUser,
     assert_setup_component,
     async_capture_events,
     async_fire_time_changed,
     async_mock_service,
+    mock_integration,
+    mock_platform,
     mock_restore_cache,
 )
 from tests.components.logbook.common import MockRow, mock_humanify
@@ -4214,3 +4226,249 @@ async def test_automation_changed_entity_id(
     hass.bus.async_fire("test_event")
     await hass.async_block_till_done()
     assert len(calls) == 2
+
+
+class _MockDiagnosticTrigger(Trigger):
+    """A new-style trigger that fires on demand and otherwise reports why not."""
+
+    @classmethod
+    async def async_validate_config(
+        cls, hass: HomeAssistant, config: ConfigType
+    ) -> ConfigType:
+        """Validate config."""
+        return config
+
+    async def async_attach_runner(
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
+    ) -> CALLBACK_TYPE:
+        """Fire on `mock_diag_event` with `fire`, else report not-triggered."""
+
+        @callback
+        def handle_event(event: Event) -> None:
+            if event.data.get("fire"):
+                run_action({"extra": "fired"}, "mock fired", event.context)
+            elif did_not_trigger is not None:
+                did_not_trigger(
+                    NotTriggeredInfo(reason="mock_reason", data={"x": 1}),
+                    event.context,
+                )
+
+        return self._hass.bus.async_listen("mock_diag_event", handle_event)
+
+
+async def _setup_diagnostic_automation(
+    hass: HomeAssistant, stored_traces: int | None = None
+) -> list[Any]:
+    """Set up an automation driven by the mock diagnostic trigger."""
+
+    async def async_get_triggers(
+        hass: HomeAssistant,
+    ) -> dict[str, type[Trigger]]:
+        return {"diag": _MockDiagnosticTrigger}
+
+    mock_integration(hass, MockModule("test"))
+    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
+    calls = async_mock_service(hass, "test", "automation")
+
+    config: dict[str, Any] = {
+        "id": "diag_auto",
+        "trigger": {"platform": "test.diag"},
+        "action": {"service": "test.automation"},
+    }
+    if stored_traces is not None:
+        config["trace"] = {"stored_traces": stored_traces}
+
+    assert await async_setup_component(
+        hass, automation.DOMAIN, {automation.DOMAIN: config}
+    )
+    await hass.async_block_till_done()
+    return calls
+
+
+async def test_automation_records_not_triggered_trace(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+    """A non-firing evaluation records a not-triggered trace with diagnostics."""
+    calls = await _setup_diagnostic_automation(hass)
+    client = await hass_ws_client()
+    msg_id = 0
+
+    def next_id() -> int:
+        nonlocal msg_id
+        msg_id += 1
+        return msg_id
+
+    hass.bus.async_fire("mock_diag_event", {"fire": False})
+    await hass.async_block_till_done()
+
+    # The action did not run, but a not-triggered trace was recorded.
+    assert len(calls) == 0
+    await client.send_json(
+        {
+            "id": next_id(),
+            "type": "trace/list",
+            "domain": "automation",
+            "item_id": "diag_auto",
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    traces = response["result"]
+    assert len(traces) == 1
+    assert traces[0]["not_triggered"] is True
+    assert traces[0]["script_execution"] == "not_triggered"
+
+    await client.send_json(
+        {
+            "id": next_id(),
+            "type": "trace/get",
+            "domain": "automation",
+            "item_id": "diag_auto",
+            "run_id": traces[0]["run_id"],
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    trace = response["result"]
+    assert trace["trace"]["trigger/0"][0]["result"] == {
+        "reason": "mock_reason",
+        "data": {"x": 1},
+    }
+
+
+async def test_not_triggered_traces_counted_separately(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+    """Not-triggered traces are capped independently from run traces."""
+    calls = await _setup_diagnostic_automation(hass, stored_traces=2)
+    client = await hass_ws_client()
+    msg_id = 0
+
+    def next_id() -> int:
+        nonlocal msg_id
+        msg_id += 1
+        return msg_id
+
+    # Fire more non-firing and firing evaluations than the stored-traces limit.
+    for _ in range(3):
+        hass.bus.async_fire("mock_diag_event", {"fire": False})
+        await hass.async_block_till_done()
+    for _ in range(3):
+        hass.bus.async_fire("mock_diag_event", {"fire": True})
+        await hass.async_block_till_done()
+
+    assert len(calls) == 3
+    await client.send_json(
+        {
+            "id": next_id(),
+            "type": "trace/list",
+            "domain": "automation",
+            "item_id": "diag_auto",
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    traces = response["result"]
+    not_triggered = [trace for trace in traces if trace.get("not_triggered")]
+    runs = [trace for trace in traces if not trace.get("not_triggered")]
+    # Each bucket is capped at 2 independently; neither evicts the other.
+    assert len(not_triggered) == 2
+    assert len(runs) == 2
+
+
+async def test_not_triggered_trace_isolated_from_chained_run(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+    """A not-triggered trace must not corrupt the trace of the run that caused it.
+
+    Automation "parent" fires the event that automation "child"'s trigger
+    evaluates and declines. The child's not-triggered handler runs inline in the
+    parent's context; without a copied context its ``trace_clear()`` would
+    redirect the parent's remaining action steps into the child's trace.
+    """
+
+    async def async_get_triggers(
+        hass: HomeAssistant,
+    ) -> dict[str, type[Trigger]]:
+        return {"diag": _MockDiagnosticTrigger}
+
+    mock_integration(hass, MockModule("test"))
+    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
+
+    assert await async_setup_component(
+        hass,
+        automation.DOMAIN,
+        {
+            automation.DOMAIN: [
+                {
+                    "id": "parent",
+                    "trigger": {"platform": "event", "event_type": "chain_start"},
+                    "action": [
+                        # The child trigger evaluates this and declines to fire.
+                        {"event": "mock_diag_event"},
+                        # The parent's own step after the chained evaluation.
+                        {"event": "chain_marker"},
+                    ],
+                },
+                {
+                    "id": "child",
+                    "trigger": {"platform": "test.diag"},
+                    "action": {"event": "child_ran"},
+                },
+            ]
+        },
+    )
+    await hass.async_block_till_done()
+
+    child_ran = async_capture_events(hass, "child_ran")
+    hass.bus.async_fire("chain_start")
+    await hass.async_block_till_done()
+
+    # The child trigger evaluated the event but did not fire.
+    assert len(child_ran) == 0
+
+    client = await hass_ws_client()
+    msg_id = 0
+
+    def next_id() -> int:
+        nonlocal msg_id
+        msg_id += 1
+        return msg_id
+
+    async def _get_only_trace(item_id: str) -> dict[str, Any]:
+        await client.send_json(
+            {
+                "id": next_id(),
+                "type": "trace/list",
+                "domain": "automation",
+                "item_id": item_id,
+            }
+        )
+        response = await client.receive_json()
+        assert response["success"]
+        assert len(response["result"]) == 1
+        run_id = response["result"][0]["run_id"]
+        await client.send_json(
+            {
+                "id": next_id(),
+                "type": "trace/get",
+                "domain": "automation",
+                "item_id": item_id,
+                "run_id": run_id,
+            }
+        )
+        response = await client.receive_json()
+        assert response["success"]
+        return response["result"]
+
+    # The parent keeps all of its own steps; action/1 must not leak away into
+    # the child's trace. This fails if the context is not copied.
+    parent_trace = await _get_only_trace("parent")
+    assert set(parent_trace["trace"]) == {"trigger/0", "action/0", "action/1"}
+
+    # The child's not-triggered trace holds only its own trigger step.
+    child_trace = await _get_only_trace("child")
+    assert child_trace["not_triggered"] is True
+    assert set(child_trace["trace"]) == {"trigger/0"}

tests/components/trace/test_websocket_api.py

@@ -1,7 +1,7 @@
 """Test Trace websocket API."""
 
 import asyncio
-from collections import defaultdict
+from collections import defaultdict, deque
 import json
 from typing import Any
 from unittest.mock import patch
@@ -9,9 +9,12 @@ from unittest.mock import patch
 import pytest
 from pytest_unordered import unordered
 
-from homeassistant.components.trace.const import DEFAULT_STORED_TRACES
+from homeassistant.components.trace import ActionTrace
+from homeassistant.components.trace.const import DATA_TRACE, DEFAULT_STORED_TRACES
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import Context, CoreState, HomeAssistant, callback
+from homeassistant.helpers.json import ExtendedJSONEncoder
+from homeassistant.helpers.trace import TraceElement
 from homeassistant.helpers.typing import UNDEFINED
 from homeassistant.setup import async_setup_component
 from homeassistant.util.uuid import random_uuid_hex
@@ -1660,3 +1663,83 @@ async def test_trace_blueprint_automation(
     assert trace["script_execution"] == "error"
     assert trace["item_id"] == "sun"
     assert trace.get("trigger", UNDEFINED) == "event 'blueprint_event'"
+
+
+class _DiagnosticActionTrace(ActionTrace):
+    """Automation-domain trace used to exercise not-triggered serialization."""
+
+    _domain = "automation"
+
+
+def _serialize_trace(not_triggered: bool, reason: str) -> dict[str, Any]:
+    """Build a trace serialized the way the trace Store persists it to disk."""
+    trace = _DiagnosticActionTrace("diag", {"id": "diag"}, None, Context())
+    trace.not_triggered = not_triggered
+    element = TraceElement({"trigger": {"idx": "0"}}, "trigger/0")
+    element.set_result(reason=reason)
+    trace.set_trace({"trigger/0": deque([element])})
+    trace.finished()
+    return json.loads(json.dumps(trace.as_dict(), cls=ExtendedJSONEncoder))
+
+
+async def test_not_triggered_trace_serialization(
+    hass: HomeAssistant,
+    hass_storage: dict[str, Any],
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Not-triggered traces serialize with their flag and restore to their bucket."""
+    run = _serialize_trace(False, "ran")
+    not_triggered = _serialize_trace(True, "mock_reason")
+
+    # Serialization: only the not-triggered trace carries the flag, and its
+    # diagnostics survive in the extended dict.
+    assert "not_triggered" not in run["short_dict"]
+    assert not_triggered["short_dict"]["not_triggered"] is True
+    assert not_triggered["extended_dict"]["trace"]["trigger/0"][0]["result"] == {
+        "reason": "mock_reason"
+    }
+
+    hass_storage["trace.saved_traces"] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": "trace.saved_traces",
+        "data": {"automation.diag": [run, not_triggered]},
+    }
+    assert await async_setup_component(hass, "trace", {})
+    client = await hass_ws_client()
+
+    # Restore (lazily, on the first query) and confirm the flag round-trips.
+    await client.send_json(
+        {"id": 1, "type": "trace/list", "domain": "automation", "item_id": "diag"}
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    assert {
+        trace["run_id"]: trace.get("not_triggered", False)
+        for trace in response["result"]
+    } == {
+        run["short_dict"]["run_id"]: False,
+        not_triggered["short_dict"]["run_id"]: True,
+    }
+
+    # Restored traces are routed back into their separate buckets.
+    trace_bucket = hass.data[DATA_TRACE]["automation.diag"]
+    assert list(trace_bucket.runs) == [run["short_dict"]["run_id"]]
+    assert list(trace_bucket.not_triggered) == [not_triggered["short_dict"]["run_id"]]
+
+    # The restored not-triggered trace keeps its diagnostics.
+    await client.send_json(
+        {
+            "id": 2,
+            "type": "trace/get",
+            "domain": "automation",
+            "item_id": "diag",
+            "run_id": not_triggered["short_dict"]["run_id"],
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    assert response["result"]["not_triggered"] is True
+    assert response["result"]["trace"]["trigger/0"][0]["result"] == {
+        "reason": "mock_reason"
+    }

tests/helpers/test_trigger.py

@@ -69,11 +69,13 @@ from homeassistant.helpers.trigger import (
     EntityNumericalStateChangedTriggerWithUnitBase,
     EntityNumericalStateCrossedThresholdTriggerWithUnitBase,
     EntityTriggerBase,
+    NotTriggeredInfo,
     PluggableAction,
     StatelessEntityTriggerBase,
     Trigger,
     TriggerActionRunner,
     TriggerConfig,
+    TriggerNotTriggeredReporter,
     _async_get_trigger_platform,
     async_initialize_triggers,
     async_validate_trigger_config,
@@ -156,7 +158,9 @@ class _MockTrigger(Trigger):
         self._options = config.options or {}
 
     async def async_attach_runner(
-        self, run_action: TriggerActionRunner
+        self,
+        run_action: TriggerActionRunner,
+        did_not_trigger: TriggerNotTriggeredReporter | None = None,
     ) -> CALLBACK_TYPE:
         """Attach the trigger to a bus event."""
         raw_template = self._options.get("option_template")
@@ -809,7 +813,9 @@ async def test_platform_multiple_triggers(
         """Mock trigger 1."""
 
         async def async_attach_runner(
-            self, run_action: TriggerActionRunner
+            self,
+            run_action: TriggerActionRunner,
+            did_not_trigger: TriggerNotTriggeredReporter | None = None,
         ) -> CALLBACK_TYPE:
             """Attach a trigger."""
             run_action({"extra": "test_trigger_1"}, "trigger 1 desc")
@@ -818,7 +824,9 @@ async def test_platform_multiple_triggers(
         """Mock trigger 2."""
 
         async def async_attach_runner(
-            self, run_action: TriggerActionRunner
+            self,
+            run_action: TriggerActionRunner,
+            did_not_trigger: TriggerNotTriggeredReporter | None = None,
         ) -> CALLBACK_TYPE:
             """Attach a trigger."""
             run_action({"extra": "test_trigger_2"}, "trigger 2 desc")
@@ -968,7 +976,9 @@ async def test_get_trigger_platform_registers_triggers(
         """Mock trigger."""
 
         async def async_attach_runner(
-            self, run_action: TriggerActionRunner
+            self,
+            run_action: TriggerActionRunner,
+            did_not_trigger: TriggerNotTriggeredReporter | None = None,
         ) -> CALLBACK_TYPE:
             return lambda: None
 
@@ -3962,6 +3972,132 @@ async def _arm_off_to_on_trigger(
     )
 
 
+async def _arm_off_to_on_trigger_with_diagnostics(
+    hass: HomeAssistant,
+    entity_ids: list[str],
+    behavior: str,
+    calls: list[dict[str, Any]],
+    reports: list[tuple[dict[str, Any], NotTriggeredInfo]],
+) -> CALLBACK_TYPE:
+    """Set up _OffToOnTrigger with both an action and a did_not_trigger reporter."""
+
+    async def async_get_triggers(
+        hass: HomeAssistant,
+    ) -> dict[str, type[Trigger]]:
+        return {"off_to_on": _OffToOnTrigger}
+
+    mock_integration(hass, MockModule("test"))
+    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
+
+    trigger_config = {
+        CONF_PLATFORM: "test.off_to_on",
+        CONF_TARGET: {CONF_ENTITY_ID: entity_ids},
+        CONF_OPTIONS: {ATTR_BEHAVIOR: behavior},
+    }
+    log = logging.getLogger(__name__)
+
+    @callback
+    def action(run_variables: dict[str, Any], context: Context | None = None) -> None:
+        calls.append(run_variables["trigger"])
+
+    @callback
+    def did_not_trigger(
+        run_variables: dict[str, Any],
+        info: NotTriggeredInfo,
+        context: Context | None = None,
+    ) -> None:
+        reports.append((run_variables["trigger"], info))
+
+    validated_config = await async_validate_trigger_config(hass, [trigger_config])
+    return await async_initialize_triggers(
+        hass,
+        validated_config,
+        action,
+        domain="test",
+        name="test_off_to_on",
+        log_cb=log.log,
+        did_not_trigger=did_not_trigger,
+    )
+
+
+async def test_entity_trigger_reports_did_not_trigger(hass: HomeAssistant) -> None:
+    """An entity trigger reports diagnostics for changes that do not fire it."""
+    entity_id = "test.entity"
+    hass.states.async_set(entity_id, STATE_OFF)
+    await hass.async_block_till_done()
+
+    calls: list[dict[str, Any]] = []
+    reports: list[tuple[dict[str, Any], NotTriggeredInfo]] = []
+    unsub = await _arm_off_to_on_trigger_with_diagnostics(
+        hass, [entity_id], BEHAVIOR_EACH, calls, reports
+    )
+
+    # A matching change fires the action and reports nothing.
+    hass.states.async_set(entity_id, STATE_ON)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert reports == []
+
+    # An invalid transition (already on) does not fire: "transition_not_a_match".
+    hass.states.async_set(entity_id, STATE_ON, {"brightness": 100})
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert len(reports) == 1
+    trigger_payload, info = reports[0]
+    assert trigger_payload[CONF_PLATFORM] == "test.off_to_on"
+    assert info.reason == "transition_not_a_match"
+    assert info.data == {
+        "entity_id": entity_id,
+        "from_state": STATE_ON,
+        "to_state": STATE_ON,
+    }
+
+    # A non-target new state does not fire: "new_state_not_a_match".
+    hass.states.async_set(entity_id, STATE_OFF)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert len(reports) == 2
+    _trigger_payload, info = reports[1]
+    assert info.reason == "new_state_not_a_match"
+    assert info.data == {"entity_id": entity_id, "to_state": STATE_OFF}
+
+    unsub()
+
+
+async def test_entity_trigger_reports_did_not_trigger_behavior_all(
+    hass: HomeAssistant,
+) -> None:
+    """Behavior 'all' reports when not every targeted entity matches."""
+    entity_1 = "test.entity_1"
+    entity_2 = "test.entity_2"
+    hass.states.async_set(entity_1, STATE_OFF)
+    hass.states.async_set(entity_2, STATE_OFF)
+    await hass.async_block_till_done()
+
+    calls: list[dict[str, Any]] = []
+    reports: list[tuple[dict[str, Any], NotTriggeredInfo]] = []
+    unsub = await _arm_off_to_on_trigger_with_diagnostics(
+        hass, [entity_1, entity_2], BEHAVIOR_ALL, calls, reports
+    )
+
+    # Only one of the two targets is on, so the trigger does not fire.
+    hass.states.async_set(entity_1, STATE_ON)
+    await hass.async_block_till_done()
+    assert calls == []
+    assert len(reports) == 1
+    _trigger_payload, info = reports[0]
+    assert info.reason == "not_all_targets_matched"
+    assert info.data == {"matches": 1, "included": 2}
+
+    # Now both are on: the trigger fires and reports nothing further.
+    hass.states.async_set(entity_2, STATE_ON)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert len(reports) == 1
+
+    unsub()
+
+
 def _set_or_remove_state(
     hass: HomeAssistant, entity_id: str, state: str | None
 ) -> None: