Index

Bus

Bases: Resource

Individual event bus instance for a specific owner (e.g., App or Service).

Source code in src/hassette/bus/bus.py

class Bus(Resource):
    """Individual event bus instance for a specific owner (e.g., App or Service)."""

    bus_service: "BusService"

    sync: BusSyncFacade
    """Synchronous facade for registering listeners from sync code (e.g. ``AppSync`` hooks)."""

    priority: int = 0
    """Priority level for event handlers created by this bus."""

    def __init__(self, hassette: "Hassette", *, priority: int = 0, parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        assert self.parent is not None, "Bus requires a parent Resource for telemetry identity (app_key/source_tier)"
        self.bus_service = self.hassette.bus_service
        self.priority = priority
        self._registered_listeners: dict[tuple[str, int, str, str], Listener] = {}
        self._error_handler: BusErrorHandlerType | None = None
        self.sync = self.add_child(BusSyncFacade, bus=self)

        # Register removal callback so once-fired listeners release their natural key and
        # record cancelled_at, mirroring Scheduler's register_removal_callback pattern.
        # owner_id derives from self.parent; the callback registry key must stay stable across
        # register/deregister even if a test fixture swaps self.parent after construction
        # (production never does — a hot-reload builds a fresh Bus). Freezing the key here keeps
        # on_shutdown's deregister matched to this register; listener removal still uses the live
        # self.owner_id because listeners register under the live owner too.
        self._removal_callback_owner_id = self.owner_id
        self.bus_service.register_removal_callback(self._removal_callback_owner_id, self._on_listener_removed)

    async def on_initialize(self) -> None:
        # Clear before any on() calls so partial-init failures don't leave stale keys.
        self._registered_listeners.clear()
        self._error_handler = None
        mark_ready(self, reason="Bus initialized")

    async def on_shutdown(self) -> None:
        """Cleanup all listeners owned by this bus's owner on shutdown."""
        self.remove_all_listeners()
        self.bus_service.deregister_removal_callback(self._removal_callback_owner_id)

    def _on_listener_removed(self, listener: "Listener") -> None:
        """Callback invoked by BusService when a listener is removed (including once-fire).

        Closes the once-fire gap: the dispatch finally block calls BusService.remove_listener
        directly without going through Bus.remove_listener, leaving the natural key stale and
        cancelled_at unwritten. This callback pops the key and spawns mark_listener_cancelled.

        The spawn guard (key must still be present) prevents a double write when Bus.remove_listener
        already popped the key and spawned mark_listener_cancelled before calling BusService:
        in that path the key is gone by the time this callback runs, so we skip the spawn.

        Accepted trade-off: during shutdown, remove_all_listeners clears _registered_listeners
        before delegating to remove_listeners_by_owner, so a once-listener that fires concurrently
        with teardown finds was_present=False and skips its cancelled_at write. This loses only a
        telemetry write in a narrow teardown window — no routing impact — and is not worth guarding.
        """
        if listener.identity.name is None:
            # Cancel-listeners (create_cancel_listener) are never tracked in _registered_listeners,
            # so there is nothing to pop. Returning early also avoids building a natural key whose
            # name falls back to "" (_listener_natural_key), which could alias a real listener.
            return
        natural_key = self._listener_natural_key(listener)
        was_present = self._registered_listeners.pop(natural_key, None) is not None
        if was_present and listener.db_id is not None:
            self.bus_service.task_bucket.spawn(
                self.bus_service.mark_listener_cancelled(listener.db_id),
                name="bus:mark_listener_cancelled",
            )

    def on_error(self, handler: "BusErrorHandlerType") -> None:
        """Register an app-level error handler for this bus.

        The handler is called when any listener on this bus raises an exception
        (including ``TimeoutError``) and the listener does not have its own
        per-registration error handler.

        This is an app-level fallback — it is resolved at dispatch time, not at listener
        registration time. A later call to ``on_error()`` replaces any previously registered
        handler.

        Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
        app shutdown may be cancelled before they complete. Do not rely on error handlers
        for delivery-critical alerting during system teardown.

        Args:
            handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`.
        """
        self._error_handler = handler

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        """Return the log level from the config for this resource."""
        return self.hassette.config.logging.bus_service

    def add_listener(
        self,
        listener: "Listener",
        *,
        if_exists: IfExistsPolicy = "error",
    ) -> "Coroutine[Any, Any, Subscription]":
        """Add a pre-built listener to the bus.

        This is the direct entry point for callers that construct a ``Listener``
        externally. The normal registration flow (``on_state_change``, ``on()``,
        etc.) goes through ``_on_internal`` instead.

        Args:
            listener: The pre-built listener to add.
            if_exists: Behavior when a listener with the same natural key already exists.
                ``"error"`` (default) raises ``DuplicateListenerError``.
                ``"skip"`` returns a subscription to the existing listener when configs match.
                ``"replace"`` cancels the existing listener and registers the new one.

        Returns:
            A subscription to the added listener (or the existing listener on skip).

        Raises:
            ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
                listeners, including once-listeners; cancel-listeners bypass this path entirely).
            DuplicateListenerError: If the listener's natural key is already registered and
                ``if_exists="error"``.
        """
        # Synchronous validation runs before the handle is constructed (design Edge Cases).
        if not listener.identity.name:
            raise ListenerNameRequiredError(handler_method=listener.identity.handler_name, topic=listener.topic)
        # Cheap path: the pre-built Listener already carries identity.source_location /
        # registration_source; only warning attribution needs the location here.
        source_location = capture_source_location()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._resolve_and_register(listener, if_exists=if_exists),
            owner=self.parent,
            source_location=source_location,
            method_name="add_listener",
        )

    def _resolve_collision(
        self,
        listener: "Listener",
        *,
        if_exists: IfExistsPolicy = "error",
    ) -> "Listener | None":
        """Resolve a potential registration collision using the if_exists policy.

        Mutates the per-bus key registry as a side effect — this is the single point where
        in-session duplicate detection and resolution happens.

        On skip short-circuit (matching existing listener), returns the existing ``Listener``
        so the caller can wrap it in a ``Subscription``. Returns ``None`` on the proceed path
        (no existing, or replace after cancelling the old), meaning the caller should register
        the new listener normally.

        Collision semantics:
        - No existing listener → register the key, return ``None`` (proceed).
        - Existing + ``if_exists="error"`` → raise ``DuplicateListenerError``.
        - Existing + ``if_exists="replace"`` → cancel/remove the existing listener, register
          the new key, return ``None`` (proceed).
        - Existing + ``if_exists="skip"`` and configs match → return the existing listener
          (short-circuit; the caller does NOT register the new one).
        - Existing + ``if_exists="skip"`` and configs differ → raise ``ValueError`` listing
          the changed fields (from ``diff_fields``).

        All listeners — including once-listeners — participate in collision tracking.
        """
        natural_key = self._listener_natural_key(listener)
        existing = self._registered_listeners.get(natural_key)
        if existing is not None:
            if if_exists == "replace":
                self.logger.debug(
                    "Replacing existing listener '%s' on topic '%s' (cancelling old, registering new)",
                    listener.identity.name,
                    listener.topic,
                )
                self.remove_listener(existing)
            elif if_exists == "skip" and existing.config_matches(listener):
                return existing
            elif if_exists == "skip":
                changed_fields = existing.diff_fields(listener)
                msg = (
                    f"A listener named '{listener.identity.name}' on topic '{listener.topic}' already exists "
                    f"but its configuration has changed (changed fields: {', '.join(changed_fields)})"
                )
                if "predicate" in changed_fields:
                    msg += (
                        ". Note: lambda/closure predicates compare by identity — two fresh lambdas with "
                        "identical bodies are not equal; use a named predicate function or if_exists='replace'"
                    )
                raise ValueError(msg)
            else:
                raise DuplicateListenerError(
                    name=listener.identity.name or "",
                    topic=listener.topic,
                    existing_handler=existing.identity.handler_name,
                    duplicate_handler=listener.identity.handler_name,
                )
        self._registered_listeners[natural_key] = listener
        return None

    async def _resolve_and_register(
        self,
        listener: "Listener",
        *,
        if_exists: IfExistsPolicy,
    ) -> Subscription:
        """Resolve a collision then register the listener, returning its Subscription.

        The async body for both add_listener and _on_internal. _resolve_collision mutates the
        per-bus key registry, so it must not run for a never-awaited call — keeping it here, in
        the awaited coroutine, is what prevents that.

        On the skip short-circuit, returns a subscription wrapping the existing listener.
        On the replace path, _resolve_collision has already cancelled the existing listener;
        if the new registration then fails, an ERROR is logged so the now-open gap (no handler
        routed under this key) is observable before the exception propagates.
        """
        # `is_replacing` must be captured BEFORE _resolve_collision runs: on the replace path it
        # pops the existing key, so reading _registered_listeners afterward would always be False.
        natural_key = self._listener_natural_key(listener)
        is_replacing = if_exists == "replace" and natural_key in self._registered_listeners

        existing = self._resolve_collision(listener, if_exists=if_exists)
        if existing is not None:
            # skip short-circuit: return a subscription wrapping the existing listener.
            return Subscription(existing, lambda: self.remove_listener(existing))

        try:
            await self.bus_service.add_listener(listener)
        except Exception:
            # _resolve_collision reserved natural_key before the await; drop it so a failed add
            # doesn't leave a phantom registration that blocks retries with a false collision.
            # Guard on identity: a concurrent replace may have re-pointed the key to a new listener
            # while this task was awaiting, and that newer mapping must not be evicted.
            if self._registered_listeners.get(natural_key) is listener:
                self._registered_listeners.pop(natural_key, None)
            if is_replacing:
                self.logger.error(
                    "Listener '%s' on topic '%s' failed to register after replacing (cancelling) the "
                    "existing listener — no handler is active for this key",
                    listener.identity.name,
                    listener.topic,
                )
            raise
        return Subscription(listener, lambda: self.remove_listener(listener))

    def _listener_natural_key(self, listener: "Listener") -> tuple[str, int, str, str]:
        """Compute the natural key tuple for a listener (for collision tracking).

        CANONICAL form: ``(app_key, instance_index, name, topic)``.
        Matches the SQL unique index defined in 001.sql and the repository upsert ON CONFLICT target.
        """
        return (
            listener.identity.app_key,
            listener.identity.instance_index,
            listener.identity.name or "",
            listener.topic,
        )

    def remove_listener(self, listener: "Listener") -> None:
        """Remove a listener from the bus and persist cancellation to the database.

        Pops the natural key from the in-memory registry, removes the listener from routing
        (via BusService), and — when ``db_id`` is set — spawns ``mark_listener_cancelled``
        on ``bus_service.task_bucket`` so the write survives resource shutdown, mirroring
        ``Scheduler.cancel_job``.

        BusService.remove_listener also fires _on_listener_removed, but that callback only
        spawns mark_listener_cancelled when the key is still present (once-fire path). Because
        this method pops the key first, the callback's spawn is skipped — avoiding a double write.
        """
        natural_key = self._listener_natural_key(listener)
        self._registered_listeners.pop(natural_key, None)
        self.bus_service.remove_listener(listener)
        if listener.db_id is not None:
            self.bus_service.task_bucket.spawn(
                self.bus_service.mark_listener_cancelled(listener.db_id),
                name="bus:mark_listener_cancelled",
            )

    def remove_all_listeners(self) -> None:
        """Remove all listeners owned by this bus's owner."""
        # Pre-clear is load-bearing: it must run before remove_listeners_by_owner so that
        # _on_listener_removed finds was_present=False and skips the cancelled_at spawn for every
        # listener on clean shutdown (shutdown maps to retired_at via reconciliation, not cancelled_at).
        self._registered_listeners.clear()
        self.bus_service.remove_listeners_by_owner(self.owner_id)

    def get_listeners(self) -> list["Listener"]:
        """Get all listeners owned by this bus's owner."""
        return self.bus_service.get_listeners_by_owner(self.owner_id)

    async def emit(self, topic: str, data: object) -> None:
        """Broadcast data to all subscribers of the given topic.

        Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
        If the internal event stream is closed (during shutdown), the event is silently dropped.
        """
        payload = HassettePayload(data=data)
        event = Event(topic=topic, payload=payload)
        await self.hassette.send_event(event)

    def on(
        self,
        *,
        topic: str,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        once: bool = False,
        debounce: float | None = None,
        throttle: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        backpressure: "BackpressurePolicy | str | None" = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to an event topic with optional filtering and modifiers.

        This is the public registration method for raw topic subscriptions. Must be awaited.
        Registration completes before the call returns — ``sub.listener.db_id`` is a valid
        integer immediately on return.

        Args:
            topic: The event topic to listen to.
            handler: The function to call when the event matches.
            where: Optional predicates to filter events. These can be custom callables or predefined predicates from
                `hassette.event_handling.predicates`. They will receive the full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            once: If True, the handler will be called only once and then removed.
            debounce: If set, applies a debounce to the handler.
            throttle: If set, applies a throttle to the handler.
            timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
                None means fall through to the config default.
            timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
            mode: Overlap behavior when a trigger fires while a prior invocation still runs —
                ``"single"``, ``"restart"``, ``"queued"``, or ``"parallel"``. When omitted, the
                effective default is tier-aware: ``parallel`` for framework listeners, ``single``
                for app listeners. Suppressed/dropped counts are live-only diagnostics, reset on
                restart.
            backpressure: Saturation policy when the global dispatch concurrency semaphore is full.
                ``"block"`` (default) waits for a slot; ``"drop_newest"`` skips the event immediately
                and records one drop on the listener. When omitted, the effective default is ``block``.
            name: Required. Stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
                empty string raises ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            if_exists: Behavior when a listener with the same natural key already exists.
                ``"error"`` (default) raises ``DuplicateListenerError``. ``"skip"`` returns the
                existing listener's subscription when the configurations match, and raises
                ``ValueError`` if the configuration has drifted. ``"replace"`` cancels the
                existing listener and registers the new one in its place.

        Returns:
            A subscription object. ``sub.cancel()`` removes the listener.
            ``sub.listener.db_id`` is a valid integer immediately on return.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered and ``if_exists="error"`` (the default).
            ValueError: If ``if_exists="skip"`` and a listener with the same ``(name, topic)``
                exists but with a different configuration (the message lists the changed fields).
        """
        _require_name(name, handler, topic)
        # Eager capture in the public def — user frame is live here (not inside the async body).
        # Returns a 2-tuple — unpack it. Two destinations: guard_await (warning attribution) AND
        # _on_internal (populates ListenerIdentity.source_location / registration_source on the DB record).
        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._on_internal(
                topic=topic,
                handler=handler,
                where=where,
                kwargs=kwargs,
                once=once,
                debounce=debounce,
                throttle=throttle,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                backpressure=backpressure,
                name=name,
                on_error=on_error,
                if_exists=if_exists,
                duration_config=None,
                source_location=source_location,
                registration_source=registration_source,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on",
        )

    async def _on_internal(
        self,
        *,
        topic: str,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        once: bool = False,
        debounce: float | None = None,
        throttle: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        backpressure: "BackpressurePolicy | str | None" = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        duration_config: "DurationConfig | None" = None,
        source_location: str = "",
        registration_source: str | None = None,
    ) -> Subscription:
        """Private registration method carrying the full parameter set.

        ``mode=None`` means "not supplied" — it resolves to the tier-aware default
        (``parallel`` for framework listeners, ``single`` for app listeners). An explicit
        mode always wins.

        Called by on() (with duration_config=None) and by _subscribe() (which
        builds DurationConfig from duration/entity_id when provided).

        source_location and registration_source are captured in the public def (user
        frame live there) and threaded down here to populate ListenerIdentity — the
        capture must NOT be duplicated here or the user frame will be gone by then.

        Builds all sub-structs (ListenerIdentity, ListenerOptions, HandlerInvoker)
        here and calls Listener.create() via the sub-struct path.

        DB registration is awaited inline — the listener's db_id is set and the
        listener is routable before this method returns.
        """
        parent = self.parent
        assert parent is not None
        app_key = parent.app_key
        instance_index = parent.index
        source_tier = parent.source_tier
        assert source_tier in ("app", "framework"), f"Invalid source_tier={source_tier!r} on {parent.class_name}"

        # Tier-aware default: an omitted mode (None) resolves to ``parallel`` for framework
        # listeners — preserving the supervisor/state-cache concurrency — and ``single`` for app
        # listeners. An explicit mode always wins. A raw string is coerced here so an invalid value
        # raises a clear ValueError at registration time.
        if mode is None:
            resolved_mode = ExecutionMode.PARALLEL if source_tier == "framework" else ExecutionMode.SINGLE
        elif isinstance(mode, ExecutionMode):
            resolved_mode = mode
        else:
            try:
                resolved_mode = ExecutionMode(mode)
            except ValueError as exc:
                valid = ", ".join(repr(m.value) for m in ExecutionMode)
                raise ValueError(f"Invalid execution mode {mode!r}; must be one of {valid}") from exc

        handler_name = callable_name(handler)
        short_name = callable_short_name(handler)

        _require_name(name, handler, topic)

        # Resolve instance_name once at registration so the executor hot path reads it off the
        # command instead of traversing app_handler per execution.
        instance_name = parent.instance_name

        identity = ListenerIdentity(
            owner_id=self.owner_id,
            app_key=app_key,
            instance_index=instance_index,
            instance_name=instance_name,
            name=name,
            source_tier=source_tier,
            handler_name=handler_name,
            handler_short_name=short_name,
            source_location=source_location,
            registration_source=registration_source or "",
        )

        # An omitted policy resolves to the flat BLOCK default (no tier-awareness, unlike mode). A raw
        # string is coerced here so an invalid value raises a clear ValueError at registration time —
        # mirroring the mode coercion above and ListenerOptions.__post_init__.
        if backpressure is None:
            resolved_backpressure = BackpressurePolicy.BLOCK
        elif isinstance(backpressure, BackpressurePolicy):
            resolved_backpressure = backpressure
        else:
            try:
                resolved_backpressure = BackpressurePolicy(backpressure)
            except ValueError as exc:
                valid = ", ".join(repr(p.value) for p in BackpressurePolicy)
                raise ValueError(f"Invalid backpressure policy {backpressure!r}; must be one of {valid}") from exc

        options = ListenerOptions(
            once=once,
            debounce=debounce,
            throttle=throttle,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            priority=self.priority,
            mode=resolved_mode,
            backpressure=resolved_backpressure,
        )

        invoker = HandlerInvoker.create(
            task_bucket=self.task_bucket,
            handler=handler,
            kwargs=kwargs,
            options=options,
            error_handler=on_error,
            app_error_handler_resolver=lambda: self._error_handler,
        )

        listener = Listener.create(
            topic=topic,
            identity=identity,
            options=options,
            invoker=invoker,
            where=where,
            duration_config=duration_config,
            logger=self.logger,
        )

        return await self._resolve_and_register(listener, if_exists=if_exists)

    async def _subscribe(
        self,
        *,
        log_label: str,
        topic: str,
        handler: "HandlerType",
        preds: list["Predicate"],
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        log_params: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        entity_id: str | None = None,
        is_attribute_listener: bool = False,
        hold_preds: list["Predicate"] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        source_location: str = "",
        registration_source: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Common subscription tail: log, normalize where, delegate to _on_internal()."""
        if self.logger.isEnabledFor(logging.DEBUG):
            filtered = (
                {k: v for k, v in log_params.items() if v is not None and not isinstance(v, Sentinel)}
                if log_params
                else {}
            )
            params_str = ", ".join(f"{k}='{v}'" for k, v in filtered.items())

            self.logger.debug(
                "Subscribing to %s with %s - being handled by '%s'",
                log_label,
                params_str,
                callable_short_name(handler),
            )

        if where is not None:
            normalized_where = where if callable(where) else P.AllOf.ensure_iterable(where)
            preds.append(normalized_where)
            if hold_preds is not None:
                hold_preds = [*hold_preds, normalized_where]

        # Build DurationConfig when entity_id is provided (for duration or immediate listeners)
        duration_config: DurationConfig | None = None
        if entity_id:
            duration_config = DurationConfig(
                entity_id=entity_id,
                duration=duration,
                immediate=immediate,
                is_attribute_listener=is_attribute_listener,
                hold_predicate=P.AllOf.ensure_iterable(hold_preds) if hold_preds else None,
            )

        return await self._on_internal(
            topic=topic,
            handler=handler,
            where=preds,
            kwargs=kwargs,
            duration_config=duration_config,
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        )

    @staticmethod
    def _normalize_service_where(
        preds: list["Predicate"],
        where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None",
    ) -> None:
        """Normalize on_call_service's Mapping-aware where clause into predicates."""
        if where is None:
            return

        if isinstance(where, Mapping):
            preds.append(P.ServiceDataWhere(where))
        elif callable(where):
            preds.append(where)
        else:
            mappings = [w for w in where if isinstance(w, Mapping)]
            other = [w for w in where if not isinstance(w, Mapping)]

            preds.extend(P.ServiceDataWhere(w) for w in mappings)

            if other:
                preds.append(P.AllOf.ensure_iterable(other))

    def on_state_change(
        self,
        entity_id: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to state changes for a specific entity.

        Must be awaited. Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
                will be used to compare the old and new state values.
            changed_from: A value or callable that will be used to filter state changes *from* this value.
            changed_to: A value or callable that will be used to filter state changes *to* this value.
            where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
                full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
                empty string raises ``ListenerNameRequiredError`` at call time.
            **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
                ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

                ``mode`` controls overlap behavior when a trigger fires while a prior invocation
                is still running: ``"single"`` drops the re-fire (the default for app handlers),
                ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
                (bounded at 10 pending), ``"parallel"`` runs concurrently. When omitted, the
                tier-aware default applies: ``"single"`` for app handlers, ``"parallel"`` for
                framework-internal listeners. An explicit ``mode=`` always wins. Suppressed
                (``single``) and dropped (``queued`` cap) events log at DEBUG only.
                Suppressed/dropped counts are live-only diagnostics, reset on restart.
                See `Execution Modes <https://hassette.dev/core-concepts/bus/execution-modes/>`_.

                ``backpressure`` controls what this listener does when the dispatch concurrency
                semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
                today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
                at the dispatch acquire point (global bus saturation), orthogonal to
                ``mode``/``debounce``/``throttle``.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
            removes the listener from routing.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered on this bus in the current session and ``if_exists="error"``
                (the default).
        """
        # Synchronous validation runs before the handle is constructed (design Edge Cases).
        _require_name(name, handler, f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}")
        if immediate and is_glob(entity_id):
            raise ValueError(
                f"'immediate=True' is not supported with glob patterns. "
                f"entity_id={entity_id!r} contains glob characters."
            )
        if duration is not None and is_glob(entity_id):
            raise ValueError(
                f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
            )

        preds, hold_preds = build_state_preds(
            entity_id, changed=changed, changed_from=changed_from, changed_to=changed_to
        )

        # Eager capture in the public def — user frame is live here.
        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._subscribe(
                log_label=f"entity '{entity_id}'",
                topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
                handler=handler,
                preds=preds,
                where=where,
                kwargs=kwargs,
                log_params={"changed": changed, "changed_from": changed_from, "changed_to": changed_to, "where": where},
                immediate=immediate,
                duration=duration,
                entity_id=entity_id,
                hold_preds=hold_preds if duration is not None else None,
                name=name,
                on_error=on_error,
                source_location=source_location,
                registration_source=registration_source,
                **opts,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on_state_change",
        )

    def on_attribute_change(
        self,
        entity_id: str,
        attr: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to state change events for a specific entity's attribute.

        Must be awaited. Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            attr: The attribute name to filter changes on (e.g., "volume").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
                it will be used to compare the old and new attribute values.
            changed_from: A value or callable that will be used to filter attribute changes *from* this value.
            changed_to: A value or callable that will be used to filter attribute changes *to* this value.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier. Omitting it entirely raises ``TypeError``
                (no default value); passing an empty string raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
                ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

                ``mode`` controls overlap behavior when a trigger fires while a prior invocation
                is still running: ``"single"`` drops the re-fire (the default for app handlers),
                ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
                (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
                counts are live-only diagnostics, reset on restart.

                ``backpressure`` controls what this listener does when the dispatch concurrency
                semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
                today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
                at the dispatch acquire point (global bus saturation), orthogonal to
                ``mode``/``debounce``/``throttle``.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered and ``if_exists="error"`` (the default).
        """
        # Synchronous validation runs before the handle is constructed (design Edge Cases).
        _require_name(name, handler, f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}[{attr}]")
        if immediate and is_glob(entity_id):
            raise ValueError(
                f"'immediate=True' is not supported with glob patterns. "
                f"entity_id={entity_id!r} contains glob characters."
            )
        if duration is not None and is_glob(entity_id):
            raise ValueError(
                f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
            )

        if not changed:
            self.logger.warning(
                (
                    "Handler '%s' - attribute change subscription "
                    "will fire on every change event for '%s' due to 'changed=False'. "
                    "Consider using `on_state_change` with 'changed=False' instead for clarity."
                ),
                callable_short_name(handler),
                entity_id,
            )

        preds, hold_preds = build_attr_preds(
            entity_id, attr, changed=changed, changed_from=changed_from, changed_to=changed_to
        )

        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._subscribe(
                log_label=f"entity '{entity_id}' attribute '{attr}'",
                topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
                handler=handler,
                preds=preds,
                where=where,
                kwargs=kwargs,
                log_params={"changed_from": changed_from, "changed_to": changed_to, "where": where},
                immediate=immediate,
                duration=duration,
                entity_id=entity_id,
                hold_preds=hold_preds if duration is not None else None,
                is_attribute_listener=True,
                name=name,
                on_error=on_error,
                source_location=source_location,
                registration_source=registration_source,
                **opts,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on_attribute_change",
        )

    def on_call_service(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to service call events.

        Must be awaited. Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            domain: The domain to filter service calls (e.g., "light").
            service: The service to filter service calls (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier for this listener. Omitting it entirely
                raises ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
                ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

                ``mode`` controls overlap behavior when a trigger fires while a prior invocation
                is still running: ``"single"`` drops the re-fire (the default for app handlers),
                ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
                (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
                counts are live-only diagnostics, reset on restart.

                ``backpressure`` controls what this listener does when the dispatch concurrency
                semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
                today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
                at the dispatch acquire point (global bus saturation), orthogonal to
                ``mode``/``debounce``/``throttle``.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered and ``if_exists="error"`` (the default).
        """
        _require_name(name, handler, str(Topic.HASS_EVENT_CALL_SERVICE))
        preds: list[Predicate] = []
        if domain is not None:
            preds.append(P.DomainMatches(domain))

        if service is not None:
            preds.append(P.ServiceMatches(service))

        self._normalize_service_where(preds, where)

        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._subscribe(
                log_label="call_service",
                topic=Topic.HASS_EVENT_CALL_SERVICE,
                handler=handler,
                preds=preds,
                where=None,
                kwargs=kwargs,
                log_params={"domain": domain, "service": service, "where": where},
                name=name,
                on_error=on_error,
                source_location=source_location,
                registration_source=registration_source,
                **opts,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on_call_service",
        )

    def on_component_loaded(
        self,
        component: str | None = None,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to component loaded events.

        Must be awaited. Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            component: The component to filter load events (e.g., "light").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        _require_name(name, handler, str(Topic.HASS_EVENT_COMPONENT_LOADED))
        preds: list[Predicate] = []

        if component is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.component"), condition=component))

        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._subscribe(
                log_label="component_loaded",
                topic=Topic.HASS_EVENT_COMPONENT_LOADED,
                handler=handler,
                preds=preds,
                where=where,
                kwargs=kwargs,
                log_params={"component": component, "where": where},
                name=name,
                on_error=on_error,
                source_location=source_location,
                registration_source=registration_source,
                **opts,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on_component_loaded",
        )

    def on_service_registered(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to service registered events.

        Must be awaited. Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            domain: The domain to filter service registrations (e.g., "light").
            service: The service to filter service registrations (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        _require_name(name, handler, str(Topic.HASS_EVENT_SERVICE_REGISTERED))
        preds: list[Predicate] = []

        if domain is not None:
            preds.append(P.DomainMatches(domain))

        if service is not None:
            preds.append(P.ServiceMatches(service))

        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._subscribe(
                log_label="service_registered",
                topic=Topic.HASS_EVENT_SERVICE_REGISTERED,
                handler=handler,
                preds=preds,
                where=where,
                kwargs=kwargs,
                log_params={"domain": domain, "service": service, "where": where},
                name=name,
                on_error=on_error,
                source_location=source_location,
                registration_source=registration_source,
                **opts,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on_service_registered",
        )

    def on_homeassistant_restart(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to Home Assistant restart events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_call_service (the true primary). See design/071.
        return self.on_call_service(
            domain="homeassistant",
            service="restart",
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_homeassistant_start(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to Home Assistant start events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_call_service (the true primary). See design/071.
        return self.on_call_service(
            domain="homeassistant",
            service="start",
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_homeassistant_stop(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to Home Assistant stop events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_call_service (the true primary). See design/071.
        return self.on_call_service(
            domain="homeassistant",
            service="stop",
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_hassette_service_status(
        self,
        status: ResourceStatus | None = None,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to hassette service status events.

        Must be awaited. Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            status: The status to filter events (e.g., ResourceStatus.STARTED).
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
                default value); passing an empty string raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        _require_name(name, handler, str(Topic.HASSETTE_EVENT_SERVICE_STATUS))
        preds: list[Predicate] = []

        if status is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._subscribe(
                log_label="hassette.service_status",
                topic=Topic.HASSETTE_EVENT_SERVICE_STATUS,
                handler=handler,
                preds=preds,
                where=where,
                kwargs=kwargs,
                log_params={"status": status, "where": where},
                name=name,
                on_error=on_error,
                source_location=source_location,
                registration_source=registration_source,
                **opts,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on_hassette_service_status",
        )

    def on_hassette_service_failed(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to hassette service failed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
                default value); passing an empty string raises ``ListenerNameRequiredError``
                at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_hassette_service_status (the true primary). See design/071.
        return self.on_hassette_service_status(
            status=ResourceStatus.FAILED,
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_hassette_service_crashed(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to hassette service crashed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_hassette_service_status (the true primary). See design/071.
        return self.on_hassette_service_status(
            status=ResourceStatus.CRASHED,
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_hassette_service_started(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to hassette service started events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_hassette_service_status (the true primary). See design/071.
        return self.on_hassette_service_status(
            status=ResourceStatus.RUNNING,
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_websocket_connected(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to websocket connected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on() (the true primary). See design/071.
        return self.on(
            topic=Topic.HASSETTE_EVENT_WEBSOCKET_CONNECTED,
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_websocket_disconnected(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to websocket disconnected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on() (the true primary). See design/071.
        return self.on(
            topic=Topic.HASSETTE_EVENT_WEBSOCKET_DISCONNECTED,
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_app_state_changed(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        status: ResourceStatus | None = None,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to app instance state change events.

        Must be awaited. Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            status: Filter events for a specific status.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        _require_name(name, handler, str(Topic.HASSETTE_EVENT_APP_STATE_CHANGED))
        preds: list[Predicate] = []

        if app_key is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.app_key"), condition=app_key))

        if status is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._subscribe(
                log_label="app_state_changed",
                topic=Topic.HASSETTE_EVENT_APP_STATE_CHANGED,
                handler=handler,
                preds=preds,
                where=where,
                kwargs=kwargs,
                log_params={"app_key": app_key, "status": status, "where": where},
                name=name,
                on_error=on_error,
                source_location=source_location,
                registration_source=registration_source,
                **opts,
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="on_app_state_changed",
        )

    def on_app_running(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to app instances reaching RUNNING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_app_state_changed (the true primary). See design/071.
        return self.on_app_state_changed(
            handler=handler,
            app_key=app_key,
            status=ResourceStatus.RUNNING,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

    def on_app_stopping(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> "Coroutine[Any, Any, Subscription]":
        """Subscribe to app instances entering STOPPING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at on_app_state_changed (the true primary). See design/071.
        return self.on_app_state_changed(
            handler=handler,
            app_key=app_key,
            status=ResourceStatus.STOPPING,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )

sync: BusSyncFacade = self.add_child(BusSyncFacade, bus=self) instance-attribute

Synchronous facade for registering listeners from sync code (e.g. AppSync hooks).

priority: int = priority class-attribute instance-attribute

Priority level for event handlers created by this bus.

config_log_level: LOG_LEVEL_TYPE property

Return the log level from the config for this resource.

on_shutdown() -> None async

Cleanup all listeners owned by this bus's owner on shutdown.

Source code in src/hassette/bus/bus.py

async def on_shutdown(self) -> None:
    """Cleanup all listeners owned by this bus's owner on shutdown."""
    self.remove_all_listeners()
    self.bus_service.deregister_removal_callback(self._removal_callback_owner_id)

on_error(handler: BusErrorHandlerType) -> None

Register an app-level error handler for this bus.

The handler is called when any listener on this bus raises an exception (including TimeoutError) and the listener does not have its own per-registration error handler.

This is an app-level fallback — it is resolved at dispatch time, not at listener registration time. A later call to on_error() replaces any previously registered handler.

Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near app shutdown may be cancelled before they complete. Do not rely on error handlers for delivery-critical alerting during system teardown.

Parameters:

Name	Type	Description	Default
handler	BusErrorHandlerType	A sync or async callable that accepts a :class:~hassette.bus.error_context.BusErrorContext.	required

Source code in src/hassette/bus/bus.py

def on_error(self, handler: "BusErrorHandlerType") -> None:
    """Register an app-level error handler for this bus.

    The handler is called when any listener on this bus raises an exception
    (including ``TimeoutError``) and the listener does not have its own
    per-registration error handler.

    This is an app-level fallback — it is resolved at dispatch time, not at listener
    registration time. A later call to ``on_error()`` replaces any previously registered
    handler.

    Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
    app shutdown may be cancelled before they complete. Do not rely on error handlers
    for delivery-critical alerting during system teardown.

    Args:
        handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`.
    """
    self._error_handler = handler

add_listener(listener: Listener, *, if_exists: IfExistsPolicy = 'error') -> Coroutine[Any, Any, Subscription]

Add a pre-built listener to the bus.

This is the direct entry point for callers that construct a Listener externally. The normal registration flow (on_state_change, on(), etc.) goes through _on_internal instead.

Parameters:

Name	Type	Description	Default
listener	Listener	The pre-built listener to add.	required
if_exists	IfExistsPolicy	Behavior when a listener with the same natural key already exists. "error" (default) raises DuplicateListenerError. "skip" returns a subscription to the existing listener when configs match. "replace" cancels the existing listener and registers the new one.	'error'

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription to the added listener (or the existing listener on skip).

Raises:

Type	Description
ListenerNameRequiredError	If the listener has no name (required for all DB-registered listeners, including once-listeners; cancel-listeners bypass this path entirely).
DuplicateListenerError	If the listener's natural key is already registered and if_exists="error".

Source code in src/hassette/bus/bus.py

def add_listener(
    self,
    listener: "Listener",
    *,
    if_exists: IfExistsPolicy = "error",
) -> "Coroutine[Any, Any, Subscription]":
    """Add a pre-built listener to the bus.

    This is the direct entry point for callers that construct a ``Listener``
    externally. The normal registration flow (``on_state_change``, ``on()``,
    etc.) goes through ``_on_internal`` instead.

    Args:
        listener: The pre-built listener to add.
        if_exists: Behavior when a listener with the same natural key already exists.
            ``"error"`` (default) raises ``DuplicateListenerError``.
            ``"skip"`` returns a subscription to the existing listener when configs match.
            ``"replace"`` cancels the existing listener and registers the new one.

    Returns:
        A subscription to the added listener (or the existing listener on skip).

    Raises:
        ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
            listeners, including once-listeners; cancel-listeners bypass this path entirely).
        DuplicateListenerError: If the listener's natural key is already registered and
            ``if_exists="error"``.
    """
    # Synchronous validation runs before the handle is constructed (design Edge Cases).
    if not listener.identity.name:
        raise ListenerNameRequiredError(handler_method=listener.identity.handler_name, topic=listener.topic)
    # Cheap path: the pre-built Listener already carries identity.source_location /
    # registration_source; only warning attribution needs the location here.
    source_location = capture_source_location()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._resolve_and_register(listener, if_exists=if_exists),
        owner=self.parent,
        source_location=source_location,
        method_name="add_listener",
    )

remove_listener(listener: Listener) -> None

Remove a listener from the bus and persist cancellation to the database.

Pops the natural key from the in-memory registry, removes the listener from routing (via BusService), and — when db_id is set — spawns mark_listener_cancelled on bus_service.task_bucket so the write survives resource shutdown, mirroring Scheduler.cancel_job.

BusService.remove_listener also fires _on_listener_removed, but that callback only spawns mark_listener_cancelled when the key is still present (once-fire path). Because this method pops the key first, the callback's spawn is skipped — avoiding a double write.

Source code in src/hassette/bus/bus.py

def remove_listener(self, listener: "Listener") -> None:
    """Remove a listener from the bus and persist cancellation to the database.

    Pops the natural key from the in-memory registry, removes the listener from routing
    (via BusService), and — when ``db_id`` is set — spawns ``mark_listener_cancelled``
    on ``bus_service.task_bucket`` so the write survives resource shutdown, mirroring
    ``Scheduler.cancel_job``.

    BusService.remove_listener also fires _on_listener_removed, but that callback only
    spawns mark_listener_cancelled when the key is still present (once-fire path). Because
    this method pops the key first, the callback's spawn is skipped — avoiding a double write.
    """
    natural_key = self._listener_natural_key(listener)
    self._registered_listeners.pop(natural_key, None)
    self.bus_service.remove_listener(listener)
    if listener.db_id is not None:
        self.bus_service.task_bucket.spawn(
            self.bus_service.mark_listener_cancelled(listener.db_id),
            name="bus:mark_listener_cancelled",
        )

remove_all_listeners() -> None

Remove all listeners owned by this bus's owner.

Source code in src/hassette/bus/bus.py

def remove_all_listeners(self) -> None:
    """Remove all listeners owned by this bus's owner."""
    # Pre-clear is load-bearing: it must run before remove_listeners_by_owner so that
    # _on_listener_removed finds was_present=False and skips the cancelled_at spawn for every
    # listener on clean shutdown (shutdown maps to retired_at via reconciliation, not cancelled_at).
    self._registered_listeners.clear()
    self.bus_service.remove_listeners_by_owner(self.owner_id)

get_listeners() -> list[Listener]

Get all listeners owned by this bus's owner.

Source code in src/hassette/bus/bus.py

def get_listeners(self) -> list["Listener"]:
    """Get all listeners owned by this bus's owner."""
    return self.bus_service.get_listeners_by_owner(self.owner_id)

emit(topic: str, data: object) -> None async

Broadcast data to all subscribers of the given topic.

Subscribers annotated with D.EventData[T] receive data pre-extracted. If the internal event stream is closed (during shutdown), the event is silently dropped.

Source code in src/hassette/bus/bus.py

async def emit(self, topic: str, data: object) -> None:
    """Broadcast data to all subscribers of the given topic.

    Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
    If the internal event stream is closed (during shutdown), the event is silently dropped.
    """
    payload = HassettePayload(data=data)
    event = Event(topic=topic, payload=payload)
    await self.hassette.send_event(event)

on(*, topic: str, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, once: bool = False, debounce: float | None = None, throttle: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, backpressure: BackpressurePolicy | str | None = None, name: str, on_error: BusErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error') -> Coroutine[Any, Any, Subscription]

Subscribe to an event topic with optional filtering and modifiers.

This is the public registration method for raw topic subscriptions. Must be awaited. Registration completes before the call returns — sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
topic	str	The event topic to listen to.	required
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Optional predicates to filter events. These can be custom callables or predefined predicates from hassette.event_handling.predicates. They will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
once	bool	If True, the handler will be called only once and then removed.	False
debounce	float | None	If set, applies a debounce to the handler.	None
throttle	float | None	If set, applies a throttle to the handler.	None
timeout	float | None	Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config. None means fall through to the config default.	None
timeout_disabled	bool	When True, disables timeout enforcement for this listener regardless of config.	False
mode	ExecutionMode | str | None	Overlap behavior when a trigger fires while a prior invocation still runs — "single", "restart", "queued", or "parallel". When omitted, the effective default is tier-aware: parallel for framework listeners, single for app listeners. Suppressed/dropped counts are live-only diagnostics, reset on restart.	None
backpressure	BackpressurePolicy | str | None	Saturation policy when the global dispatch concurrency semaphore is full. "block" (default) waits for a slot; "drop_newest" skips the event immediately and records one drop on the listener. When omitted, the effective default is block.	None
name	str	Required. Stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
if_exists	IfExistsPolicy	Behavior when a listener with the same natural key already exists. "error" (default) raises DuplicateListenerError. "skip" returns the existing listener's subscription when the configurations match, and raises ValueError if the configuration has drifted. "replace" cancels the existing listener and registers the new one in its place.	'error'

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object. sub.cancel() removes the listener.
Coroutine[Any, Any, Subscription]	sub.listener.db_id is a valid integer immediately on return.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered and if_exists="error" (the default).
ValueError	If if_exists="skip" and a listener with the same (name, topic) exists but with a different configuration (the message lists the changed fields).

Source code in src/hassette/bus/bus.py

def on(
    self,
    *,
    topic: str,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    once: bool = False,
    debounce: float | None = None,
    throttle: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    backpressure: "BackpressurePolicy | str | None" = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to an event topic with optional filtering and modifiers.

    This is the public registration method for raw topic subscriptions. Must be awaited.
    Registration completes before the call returns — ``sub.listener.db_id`` is a valid
    integer immediately on return.

    Args:
        topic: The event topic to listen to.
        handler: The function to call when the event matches.
        where: Optional predicates to filter events. These can be custom callables or predefined predicates from
            `hassette.event_handling.predicates`. They will receive the full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        once: If True, the handler will be called only once and then removed.
        debounce: If set, applies a debounce to the handler.
        throttle: If set, applies a throttle to the handler.
        timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
            None means fall through to the config default.
        timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
        mode: Overlap behavior when a trigger fires while a prior invocation still runs —
            ``"single"``, ``"restart"``, ``"queued"``, or ``"parallel"``. When omitted, the
            effective default is tier-aware: ``parallel`` for framework listeners, ``single``
            for app listeners. Suppressed/dropped counts are live-only diagnostics, reset on
            restart.
        backpressure: Saturation policy when the global dispatch concurrency semaphore is full.
            ``"block"`` (default) waits for a slot; ``"drop_newest"`` skips the event immediately
            and records one drop on the listener. When omitted, the effective default is ``block``.
        name: Required. Stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
            empty string raises ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        if_exists: Behavior when a listener with the same natural key already exists.
            ``"error"`` (default) raises ``DuplicateListenerError``. ``"skip"`` returns the
            existing listener's subscription when the configurations match, and raises
            ``ValueError`` if the configuration has drifted. ``"replace"`` cancels the
            existing listener and registers the new one in its place.

    Returns:
        A subscription object. ``sub.cancel()`` removes the listener.
        ``sub.listener.db_id`` is a valid integer immediately on return.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered and ``if_exists="error"`` (the default).
        ValueError: If ``if_exists="skip"`` and a listener with the same ``(name, topic)``
            exists but with a different configuration (the message lists the changed fields).
    """
    _require_name(name, handler, topic)
    # Eager capture in the public def — user frame is live here (not inside the async body).
    # Returns a 2-tuple — unpack it. Two destinations: guard_await (warning attribution) AND
    # _on_internal (populates ListenerIdentity.source_location / registration_source on the DB record).
    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._on_internal(
            topic=topic,
            handler=handler,
            where=where,
            kwargs=kwargs,
            once=once,
            debounce=debounce,
            throttle=throttle,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            backpressure=backpressure,
            name=name,
            on_error=on_error,
            if_exists=if_exists,
            duration_config=None,
            source_location=source_location,
            registration_source=registration_source,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on",
    )

on_state_change(entity_id: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to state changes for a specific entity.

Must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the state changed. If a ComparisonCondition is provided, it will be used to compare the old and new state values.	True
changed_from	ChangeType	A value or callable that will be used to filter state changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter state changes to this value.	NOT_PROVIDED
where	WhereClause	Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options. Accepts once, debounce, throttle, timeout, timeout_disabled, if_exists, mode, and backpressure. mode controls overlap behavior when a trigger fires while a prior invocation is still running: "single" drops the re-fire (the default for app handlers), "restart" cancels and replaces, "queued" serializes in arrival order (bounded at 10 pending), "parallel" runs concurrently. When omitted, the tier-aware default applies: "single" for app handlers, "parallel" for framework-internal listeners. An explicit mode= always wins. Suppressed (single) and dropped (queued cap) events log at DEBUG only. Suppressed/dropped counts are live-only diagnostics, reset on restart. See Execution Modes <https://hassette.dev/core-concepts/bus/execution-modes/>_. backpressure controls what this listener does when the dispatch concurrency semaphore is saturated: "block" (default) waits for a slot, unchanged from today; "drop_newest" skips the event immediately rather than waiting. It gates at the dispatch acquire point (global bus saturation), orthogonal to mode/debounce/throttle.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object. sub.listener.db_id is set immediately. sub.cancel()
Coroutine[Any, Any, Subscription]	removes the listener from routing.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered on this bus in the current session and if_exists="error" (the default).

Source code in src/hassette/bus/bus.py

def on_state_change(
    self,
    entity_id: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to state changes for a specific entity.

    Must be awaited. Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
            will be used to compare the old and new state values.
        changed_from: A value or callable that will be used to filter state changes *from* this value.
        changed_to: A value or callable that will be used to filter state changes *to* this value.
        where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
            full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
            empty string raises ``ListenerNameRequiredError`` at call time.
        **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
            ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

            ``mode`` controls overlap behavior when a trigger fires while a prior invocation
            is still running: ``"single"`` drops the re-fire (the default for app handlers),
            ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
            (bounded at 10 pending), ``"parallel"`` runs concurrently. When omitted, the
            tier-aware default applies: ``"single"`` for app handlers, ``"parallel"`` for
            framework-internal listeners. An explicit ``mode=`` always wins. Suppressed
            (``single``) and dropped (``queued`` cap) events log at DEBUG only.
            Suppressed/dropped counts are live-only diagnostics, reset on restart.
            See `Execution Modes <https://hassette.dev/core-concepts/bus/execution-modes/>`_.

            ``backpressure`` controls what this listener does when the dispatch concurrency
            semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
            today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
            at the dispatch acquire point (global bus saturation), orthogonal to
            ``mode``/``debounce``/``throttle``.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
        removes the listener from routing.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered on this bus in the current session and ``if_exists="error"``
            (the default).
    """
    # Synchronous validation runs before the handle is constructed (design Edge Cases).
    _require_name(name, handler, f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}")
    if immediate and is_glob(entity_id):
        raise ValueError(
            f"'immediate=True' is not supported with glob patterns. "
            f"entity_id={entity_id!r} contains glob characters."
        )
    if duration is not None and is_glob(entity_id):
        raise ValueError(
            f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
        )

    preds, hold_preds = build_state_preds(
        entity_id, changed=changed, changed_from=changed_from, changed_to=changed_to
    )

    # Eager capture in the public def — user frame is live here.
    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._subscribe(
            log_label=f"entity '{entity_id}'",
            topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"changed": changed, "changed_from": changed_from, "changed_to": changed_to, "where": where},
            immediate=immediate,
            duration=duration,
            entity_id=entity_id,
            hold_preds=hold_preds if duration is not None else None,
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on_state_change",
    )

on_attribute_change(entity_id: str, attr: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to state change events for a specific entity's attribute.

Must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
attr	str	The attribute name to filter changes on (e.g., "volume").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the attribute changed. If a ComparisonCondition is provided, it will be used to compare the old and new attribute values.	True
changed_from	ChangeType	A value or callable that will be used to filter attribute changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter attribute changes to this value.	NOT_PROVIDED
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable string identifier. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options. Accepts once, debounce, throttle, timeout, timeout_disabled, if_exists, mode, and backpressure. mode controls overlap behavior when a trigger fires while a prior invocation is still running: "single" drops the re-fire (the default for app handlers), "restart" cancels and replaces, "queued" serializes in arrival order (bounded at 10 pending), "parallel" runs concurrently. Suppressed/dropped counts are live-only diagnostics, reset on restart. backpressure controls what this listener does when the dispatch concurrency semaphore is saturated: "block" (default) waits for a slot, unchanged from today; "drop_newest" skips the event immediately rather than waiting. It gates at the dispatch acquire point (global bus saturation), orthogonal to mode/debounce/throttle.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered and if_exists="error" (the default).

Source code in src/hassette/bus/bus.py

def on_attribute_change(
    self,
    entity_id: str,
    attr: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to state change events for a specific entity's attribute.

    Must be awaited. Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        attr: The attribute name to filter changes on (e.g., "volume").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
            it will be used to compare the old and new attribute values.
        changed_from: A value or callable that will be used to filter attribute changes *from* this value.
        changed_to: A value or callable that will be used to filter attribute changes *to* this value.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier. Omitting it entirely raises ``TypeError``
            (no default value); passing an empty string raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
            ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

            ``mode`` controls overlap behavior when a trigger fires while a prior invocation
            is still running: ``"single"`` drops the re-fire (the default for app handlers),
            ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
            (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
            counts are live-only diagnostics, reset on restart.

            ``backpressure`` controls what this listener does when the dispatch concurrency
            semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
            today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
            at the dispatch acquire point (global bus saturation), orthogonal to
            ``mode``/``debounce``/``throttle``.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered and ``if_exists="error"`` (the default).
    """
    # Synchronous validation runs before the handle is constructed (design Edge Cases).
    _require_name(name, handler, f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}[{attr}]")
    if immediate and is_glob(entity_id):
        raise ValueError(
            f"'immediate=True' is not supported with glob patterns. "
            f"entity_id={entity_id!r} contains glob characters."
        )
    if duration is not None and is_glob(entity_id):
        raise ValueError(
            f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
        )

    if not changed:
        self.logger.warning(
            (
                "Handler '%s' - attribute change subscription "
                "will fire on every change event for '%s' due to 'changed=False'. "
                "Consider using `on_state_change` with 'changed=False' instead for clarity."
            ),
            callable_short_name(handler),
            entity_id,
        )

    preds, hold_preds = build_attr_preds(
        entity_id, attr, changed=changed, changed_from=changed_from, changed_to=changed_to
    )

    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._subscribe(
            log_label=f"entity '{entity_id}' attribute '{attr}'",
            topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"changed_from": changed_from, "changed_to": changed_to, "where": where},
            immediate=immediate,
            duration=duration,
            entity_id=entity_id,
            hold_preds=hold_preds if duration is not None else None,
            is_attribute_listener=True,
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on_attribute_change",
    )

on_call_service(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to service call events.

Must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service calls (e.g., "light").	None
service	str | None	The service to filter service calls (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable string identifier for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options. Accepts once, debounce, throttle, timeout, timeout_disabled, if_exists, mode, and backpressure. mode controls overlap behavior when a trigger fires while a prior invocation is still running: "single" drops the re-fire (the default for app handlers), "restart" cancels and replaces, "queued" serializes in arrival order (bounded at 10 pending), "parallel" runs concurrently. Suppressed/dropped counts are live-only diagnostics, reset on restart. backpressure controls what this listener does when the dispatch concurrency semaphore is saturated: "block" (default) waits for a slot, unchanged from today; "drop_newest" skips the event immediately rather than waiting. It gates at the dispatch acquire point (global bus saturation), orthogonal to mode/debounce/throttle.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered and if_exists="error" (the default).

Source code in src/hassette/bus/bus.py

def on_call_service(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to service call events.

    Must be awaited. Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        domain: The domain to filter service calls (e.g., "light").
        service: The service to filter service calls (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier for this listener. Omitting it entirely
            raises ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
            ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

            ``mode`` controls overlap behavior when a trigger fires while a prior invocation
            is still running: ``"single"`` drops the re-fire (the default for app handlers),
            ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
            (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
            counts are live-only diagnostics, reset on restart.

            ``backpressure`` controls what this listener does when the dispatch concurrency
            semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
            today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
            at the dispatch acquire point (global bus saturation), orthogonal to
            ``mode``/``debounce``/``throttle``.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered and ``if_exists="error"`` (the default).
    """
    _require_name(name, handler, str(Topic.HASS_EVENT_CALL_SERVICE))
    preds: list[Predicate] = []
    if domain is not None:
        preds.append(P.DomainMatches(domain))

    if service is not None:
        preds.append(P.ServiceMatches(service))

    self._normalize_service_where(preds, where)

    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._subscribe(
            log_label="call_service",
            topic=Topic.HASS_EVENT_CALL_SERVICE,
            handler=handler,
            preds=preds,
            where=None,
            kwargs=kwargs,
            log_params={"domain": domain, "service": service, "where": where},
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on_call_service",
    )

on_component_loaded(component: str | None = None, *, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to component loaded events.

Must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
component	str | None	The component to filter load events (e.g., "light").	None
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Stable name for this listener. Required on all DB-registered listeners.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_component_loaded(
    self,
    component: str | None = None,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to component loaded events.

    Must be awaited. Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        component: The component to filter load events (e.g., "light").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    _require_name(name, handler, str(Topic.HASS_EVENT_COMPONENT_LOADED))
    preds: list[Predicate] = []

    if component is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.component"), condition=component))

    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._subscribe(
            log_label="component_loaded",
            topic=Topic.HASS_EVENT_COMPONENT_LOADED,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"component": component, "where": where},
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on_component_loaded",
    )

on_service_registered(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to service registered events.

Must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service registrations (e.g., "light").	None
service	str | None	The service to filter service registrations (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Stable name for this listener. Required on all DB-registered listeners.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_service_registered(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to service registered events.

    Must be awaited. Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        domain: The domain to filter service registrations (e.g., "light").
        service: The service to filter service registrations (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    _require_name(name, handler, str(Topic.HASS_EVENT_SERVICE_REGISTERED))
    preds: list[Predicate] = []

    if domain is not None:
        preds.append(P.DomainMatches(domain))

    if service is not None:
        preds.append(P.ServiceMatches(service))

    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._subscribe(
            log_label="service_registered",
            topic=Topic.HASS_EVENT_SERVICE_REGISTERED,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"domain": domain, "service": service, "where": where},
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on_service_registered",
    )

on_homeassistant_restart(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to Home Assistant restart events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_homeassistant_restart(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to Home Assistant restart events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_call_service (the true primary). See design/071.
    return self.on_call_service(
        domain="homeassistant",
        service="restart",
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_homeassistant_start(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to Home Assistant start events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_homeassistant_start(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to Home Assistant start events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_call_service (the true primary). See design/071.
    return self.on_call_service(
        domain="homeassistant",
        service="start",
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_homeassistant_stop(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to Home Assistant stop events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_homeassistant_stop(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to Home Assistant stop events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_call_service (the true primary). See design/071.
    return self.on_call_service(
        domain="homeassistant",
        service="stop",
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_hassette_service_status(status: ResourceStatus | None = None, *, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to hassette service status events.

Must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
status	ResourceStatus | None	The status to filter events (e.g., ResourceStatus.STARTED).	None
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_hassette_service_status(
    self,
    status: ResourceStatus | None = None,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to hassette service status events.

    Must be awaited. Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        status: The status to filter events (e.g., ResourceStatus.STARTED).
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
            default value); passing an empty string raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    _require_name(name, handler, str(Topic.HASSETTE_EVENT_SERVICE_STATUS))
    preds: list[Predicate] = []

    if status is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._subscribe(
            log_label="hassette.service_status",
            topic=Topic.HASSETTE_EVENT_SERVICE_STATUS,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"status": status, "where": where},
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on_hassette_service_status",
    )

on_hassette_service_failed(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to hassette service failed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_hassette_service_failed(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to hassette service failed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
            default value); passing an empty string raises ``ListenerNameRequiredError``
            at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_hassette_service_status (the true primary). See design/071.
    return self.on_hassette_service_status(
        status=ResourceStatus.FAILED,
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_hassette_service_crashed(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to hassette service crashed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_hassette_service_crashed(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to hassette service crashed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_hassette_service_status (the true primary). See design/071.
    return self.on_hassette_service_status(
        status=ResourceStatus.CRASHED,
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_hassette_service_started(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to hassette service started events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_hassette_service_started(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to hassette service started events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_hassette_service_status (the true primary). See design/071.
    return self.on_hassette_service_status(
        status=ResourceStatus.RUNNING,
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_websocket_connected(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to websocket connected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_websocket_connected(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to websocket connected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on() (the true primary). See design/071.
    return self.on(
        topic=Topic.HASSETTE_EVENT_WEBSOCKET_CONNECTED,
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_websocket_disconnected(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to websocket disconnected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_websocket_disconnected(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to websocket disconnected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on() (the true primary). See design/071.
    return self.on(
        topic=Topic.HASSETTE_EVENT_WEBSOCKET_DISCONNECTED,
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_app_state_changed(*, handler: HandlerType, app_key: str | None = None, status: ResourceStatus | None = None, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to app instance state change events.

Must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
status	ResourceStatus | None	Filter events for a specific status.	None
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Stable name for this listener. Required on all DB-registered listeners.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_app_state_changed(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    status: ResourceStatus | None = None,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to app instance state change events.

    Must be awaited. Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        status: Filter events for a specific status.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    _require_name(name, handler, str(Topic.HASSETTE_EVENT_APP_STATE_CHANGED))
    preds: list[Predicate] = []

    if app_key is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.app_key"), condition=app_key))

    if status is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._subscribe(
            log_label="app_state_changed",
            topic=Topic.HASSETTE_EVENT_APP_STATE_CHANGED,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"app_key": app_key, "status": status, "where": where},
            name=name,
            on_error=on_error,
            source_location=source_location,
            registration_source=registration_source,
            **opts,
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="on_app_state_changed",
    )

on_app_running(*, handler: HandlerType, app_key: str | None = None, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to app instances reaching RUNNING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_app_running(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to app instances reaching RUNNING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_app_state_changed (the true primary). See design/071.
    return self.on_app_state_changed(
        handler=handler,
        app_key=app_key,
        status=ResourceStatus.RUNNING,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

on_app_stopping(*, handler: HandlerType, app_key: str | None = None, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Coroutine[Any, Any, Subscription]

Subscribe to app instances entering STOPPING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Coroutine[Any, Any, Subscription]	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

def on_app_stopping(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> "Coroutine[Any, Any, Subscription]":
    """Subscribe to app instances entering STOPPING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at on_app_state_changed (the true primary). See design/071.
    return self.on_app_state_changed(
        handler=handler,
        app_key=app_key,
        status=ResourceStatus.STOPPING,
        where=where,
        kwargs=kwargs,
        name=name,
        on_error=on_error,
        **opts,
    )

BusErrorContext dataclass

Bases: ErrorContext

Context passed to bus error handlers when a listener raises an exception.

Attributes:

Name	Type	Description
exception	BaseException	The exception that was raised by the listener. Retains its __traceback__ chain — use traceback (the string field) for display. The live traceback pins originating stack frame locals until this context is garbage-collected (bounded by error_handler_timeout_seconds).
traceback	str	Formatted traceback string. Always a non-empty string — the design explicitly requires always-populated tracebacks in the user-facing context, unlike the framework's own log suppression which may suppress them.
topic	str	The topic the listener was registered on.
listener_name	str	The name of the listener function that raised the exception.
event	Event[Any]	The event that was being processed when the exception occurred.

Source code in src/hassette/bus/error_context.py

@dataclass(frozen=True)
class BusErrorContext(ErrorContext):
    """Context passed to bus error handlers when a listener raises an exception.

    Attributes:
        exception: The exception that was raised by the listener. Retains its
            ``__traceback__`` chain — use ``traceback`` (the string field) for display.
            The live traceback pins originating stack frame locals until this context
            is garbage-collected (bounded by ``error_handler_timeout_seconds``).
        traceback: Formatted traceback string. Always a non-empty string — the
            design explicitly requires always-populated tracebacks in the user-facing
            context, unlike the framework's own log suppression which may suppress them.
        topic: The topic the listener was registered on.
        listener_name: The name of the listener function that raised the exception.
        event: The event that was being processed when the exception occurred.
    """

    topic: str
    listener_name: str
    event: "Event[Any]"

    @property
    def domain_label(self) -> str:
        return f"topic={self.topic}, listener={self.listener_name}"

DurationConfig dataclass

Groups duration-hold configuration fields and owns the timer lifecycle; timer is attached via attach_timer().

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class DurationConfig:
    """Groups duration-hold configuration fields and owns the timer lifecycle; timer is attached via attach_timer()."""

    entity_id: str
    """Entity ID this duration listener is tracking. Required — non-empty."""

    duration: float | None = None
    """Duration in seconds the entity must remain in the matching state before the handler fires.
    None for immediate-only or entity_id-only listeners."""

    immediate: bool = False
    """If True, fire the handler immediately with the current entity state on registration."""

    is_attribute_listener: bool = False
    """True when this listener was registered via on_attribute_change."""

    hold_predicate: "Predicate | None" = None
    """State-value predicates only (excludes transition predicates like StateFrom, StateDidChange).
    Used by DurationTimer for cancel evaluation and fire-time recheck. None when not set."""

    _timer: "DurationTimer | None" = field(default=None, init=False)
    """Duration timer. Attached via attach_timer() during BusService registration."""

    def __post_init__(self) -> None:
        if not self.entity_id:
            raise ValueError("'entity_id' must be a non-empty string")
        if self.duration is not None and self.duration <= 0:
            raise ValueError("'duration' must be a positive number")

    @property
    def timer(self) -> "DurationTimer":
        """Return the attached DurationTimer. Asserts it has been attached."""
        assert self._timer is not None, "timer not yet attached — call attach_timer() first"
        return self._timer

    def cancel_timer(self) -> None:
        """Cancel the attached duration timer if present."""
        if self._timer is not None:
            self._timer.cancel()

    def attach_timer(
        self,
        task_bucket: "TaskBucket",
        owner_id: str,
        create_cancel_sub: "Callable[[], Subscription]",
        on_cancel: Callable[[], None] | None = None,
        normalize_cancel_event: "Callable[[Event[Any]], Event[Any]] | None" = None,
    ) -> None:
        """Construct a DurationTimer and store it.

        BusService calls this method during registration, passing the
        cancel-subscription factory and on_cancel callback. Counter
        ownership stays in BusService.
        """
        assert self._timer is None, "timer already attached — call cancel() before re-attaching"
        assert self.duration is not None, "attach_timer() requires a non-None duration"
        self._timer = DurationTimer(
            task_bucket=task_bucket,
            duration=self.duration,
            predicates=self.hold_predicate,
            entity_id=self.entity_id,
            owner_id=owner_id,
            create_cancel_sub=create_cancel_sub,
            on_cancel=on_cancel,
            normalize_cancel_event=normalize_cancel_event,
        )

entity_id: str instance-attribute

Entity ID this duration listener is tracking. Required — non-empty.

duration: float | None = None class-attribute instance-attribute

Duration in seconds the entity must remain in the matching state before the handler fires. None for immediate-only or entity_id-only listeners.

immediate: bool = False class-attribute instance-attribute

If True, fire the handler immediately with the current entity state on registration.

is_attribute_listener: bool = False class-attribute instance-attribute

True when this listener was registered via on_attribute_change.

hold_predicate: Predicate | None = None class-attribute instance-attribute

State-value predicates only (excludes transition predicates like StateFrom, StateDidChange). Used by DurationTimer for cancel evaluation and fire-time recheck. None when not set.

timer: DurationTimer property

Return the attached DurationTimer. Asserts it has been attached.

cancel_timer() -> None

Cancel the attached duration timer if present.

Source code in src/hassette/bus/listeners.py

def cancel_timer(self) -> None:
    """Cancel the attached duration timer if present."""
    if self._timer is not None:
        self._timer.cancel()

attach_timer(task_bucket: TaskBucket, owner_id: str, create_cancel_sub: Callable[[], Subscription], on_cancel: Callable[[], None] | None = None, normalize_cancel_event: Callable[[Event[Any]], Event[Any]] | None = None) -> None

Construct a DurationTimer and store it.

BusService calls this method during registration, passing the cancel-subscription factory and on_cancel callback. Counter ownership stays in BusService.

Source code in src/hassette/bus/listeners.py

def attach_timer(
    self,
    task_bucket: "TaskBucket",
    owner_id: str,
    create_cancel_sub: "Callable[[], Subscription]",
    on_cancel: Callable[[], None] | None = None,
    normalize_cancel_event: "Callable[[Event[Any]], Event[Any]] | None" = None,
) -> None:
    """Construct a DurationTimer and store it.

    BusService calls this method during registration, passing the
    cancel-subscription factory and on_cancel callback. Counter
    ownership stays in BusService.
    """
    assert self._timer is None, "timer already attached — call cancel() before re-attaching"
    assert self.duration is not None, "attach_timer() requires a non-None duration"
    self._timer = DurationTimer(
        task_bucket=task_bucket,
        duration=self.duration,
        predicates=self.hold_predicate,
        entity_id=self.entity_id,
        owner_id=owner_id,
        create_cancel_sub=create_cancel_sub,
        on_cancel=on_cancel,
        normalize_cancel_event=normalize_cancel_event,
    )

HandlerInvoker dataclass

Owns handler invocation, async wrapping, parameter injection, rate limiting, and the once-guard.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class HandlerInvoker:
    """Owns handler invocation, async wrapping, parameter injection, rate limiting, and the once-guard."""

    orig_handler: "HandlerType"
    """Original handler function provided by the user."""

    async_handler: "AsyncHandlerType"
    """Async-wrapped handler function."""

    injector: ParameterInjector
    """Parameter injector for dependency injection."""

    kwargs: Mapping[str, Any] | None
    """Keyword arguments to pass to the handler."""

    error_handler: "BusErrorHandlerType | None"
    """Optional per-listener error handler."""

    app_error_handler_resolver: "Callable[[], BusErrorHandlerType | None] | None"
    """Closure that resolves the app-level error handler at dispatch time."""

    rate_limiter: RateLimiter | None
    """Rate limiter for debounce/throttle. None when no rate limiting is configured."""

    task_bucket: "TaskBucket"
    """TaskBucket used to spawn the cancellable child handler task for non-parallel modes."""

    guard: ExecutionModeGuard
    """Per-listener overlap state machine. Owns the running-task reference and live counters."""

    mode: ExecutionMode
    """Resolved overlap mode. Copied from options so dispatch can branch without reading the guard's
    private state."""

    handler_short_name: str
    """Short handler name, captured for the stall WARNING and debug logs."""

    once: bool = False
    """Whether this invoker fires only once. Intentional copy of ListenerOptions.once —
    dispatch() needs this but cannot back-reference options without a circular dependency."""

    backpressure_dropped: int = 0
    """Count of events dropped at the dispatch acquire gate due to DROP_NEWEST backpressure.
    Incremented only in BusService.dispatch — one writer, on the event loop, no await between
    the locked() check and the increment. Live-only; resets on restart, never persisted."""

    fired: bool = field(default=False, init=False)
    """Guard for once=True: set before the first invocation to prevent double-fire."""

    pending_done: "set[asyncio.Future[None]]" = field(default_factory=set, init=False)
    """Unresolved per-invocation completion futures for non-parallel modes.

    Each ``run_with_mode`` call (single/restart/queued) parks its outer dispatch task on a future
    that resolves when the handler actually runs (or is dropped/released). A queued trigger accepted
    into the deque has no live child until drain time, so its future would hang forever if the
    listener is released first. ``release_guard`` resolves every remaining future here so those outer
    dispatch tasks unwind and ``_dispatch_pending`` settles."""

    @classmethod
    def create(
        cls,
        task_bucket: "TaskBucket",
        handler: "HandlerType",
        kwargs: Mapping[str, Any] | None,
        options: ListenerOptions,
        error_handler: "BusErrorHandlerType | None" = None,
        app_error_handler_resolver: "Callable[[], BusErrorHandlerType | None] | None" = None,
    ) -> "HandlerInvoker":
        """Construct a HandlerInvoker from a handler and options.

        Builds the async wrapper, injector, and rate limiter. Copies options.once.

        Args:
            task_bucket: TaskBucket for async adapter and rate limiter.
            handler: The user-supplied handler callable.
            kwargs: Optional keyword arguments to pass to the handler.
            options: Behavioral options (once, debounce, throttle).
            error_handler: Optional per-listener error handler.
            app_error_handler_resolver: Closure for app-level error handler resolution.
        """
        handler_name = callable_name(handler)
        signature = get_typed_signature(handler)
        async_handler = make_async_handler(handler, task_bucket)
        injector = ParameterInjector(handler_name, signature)

        rate_limiter: RateLimiter | None = None
        if options.debounce is not None or options.throttle is not None:
            rate_limiter = RateLimiter(
                task_bucket=task_bucket,
                debounce=options.debounce,
                throttle=options.throttle,
                handler_name=handler_name,
            )

        return cls(
            orig_handler=handler,
            async_handler=async_handler,
            injector=injector,
            kwargs=kwargs,
            error_handler=error_handler,
            app_error_handler_resolver=app_error_handler_resolver,
            rate_limiter=rate_limiter,
            task_bucket=task_bucket,
            guard=ExecutionModeGuard(options.mode),
            mode=options.mode,
            handler_short_name=callable_short_name(handler),
            once=options.once,
        )

    def mark_fired(self) -> None:
        """Mark this once-invoker as having fired. Called by dispatch() and Listener.cancel()."""
        self.fired = True

    def set_app_error_handler_resolver(self, resolver: "Callable[[], BusErrorHandlerType | None]") -> None:
        """Set the closure that resolves the app-level error handler at dispatch time."""
        self.app_error_handler_resolver = resolver

    async def dispatch(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
        """Apply the once-guard, rate limiter, and overlap mode around the given invoke function.

        Order: once-guard → rate limiter (whether to start) → mode guard (overlap of started
        invocations). BusService builds ``invoke_fn`` (tracked telemetry); HandlerInvoker wraps it.

        Once-guard: if ``once=True`` and the invoker has already fired, returns immediately. Safe
        without a lock — no ``await`` between check-and-set.

        Mode guard: ``parallel`` is a pass-through that awaits ``invoke_fn`` inline (byte-for-byte
        today's behavior, no child task). For ``single``/``restart``/``queued`` the guard receives a
        run-and-track callable that spawns a fresh child task through ``task_bucket``; this method
        then awaits that child so the outer dispatch task stays pending and remains counted by
        ``_dispatch_pending``. A ``restart`` cancellation of the child surfaces as ``CancelledError``
        inside the child only — it is swallowed here so the outer dispatch task does not crash.
        """
        if self.once and self.fired:
            return
        if self.once:
            self.mark_fired()

        if self.rate_limiter:
            await self.rate_limiter.call(lambda: self.run_with_mode(invoke_fn))
        else:
            await self.run_with_mode(invoke_fn)

    async def run_with_mode(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
        """Apply the overlap mode guard to a single started invocation."""
        if self.mode is ExecutionMode.PARALLEL:
            await invoke_fn()
            return

        await run_through_guard(
            guard=self.guard,
            spawn=lambda coro, *, name: self.task_bucket.spawn(coro, name=name),
            pending_done=self.pending_done,
            invoke=invoke_fn,
            warn=self.warn_stalled,
            spawn_name="bus:mode_invocation",
            threshold=STALL_THRESHOLD_SECONDS,
        )

    def warn_stalled(self, threshold: float) -> None:
        """Emit the feature's stall WARNING: a non-parallel handler is still holding its guard."""
        LOGGER.warning(
            "Handler '%s' has held its %s execution-mode guard for over %.0fs and is still running",
            self.handler_short_name,
            self.mode.value,
            threshold,
        )

    def cancel(self) -> None:
        """Cancel any pending rate-limiter tasks."""
        if self.rate_limiter:
            self.rate_limiter.cancel()

    async def release_guard(self) -> None:
        """Release the execution-mode guard: cancel the in-flight task and drop queued factories.

        Called when a listener is cancelled or replaced so no event/listener/app references leak.
        ``parallel`` listeners hold no guard state, so this is a cheap no-op for them.

        Queued triggers still parked in the guard's deque never spawn a child once released, so
        their outer dispatch tasks are parked on ``done`` futures that nothing else will resolve.
        ``drain_pending_done`` resolves every remaining one so those tasks unwind and
        ``_dispatch_pending`` settles.
        """
        await self.guard.release()
        drain_pending_done(self.pending_done)

    async def invoke(self, event: "Event[Any]") -> None:
        """Invoke the handler with dependency injection."""
        kwargs = self.injector.inject_parameters(event, **(self.kwargs or {}))
        await self.async_handler(**kwargs)

orig_handler: HandlerType instance-attribute

Original handler function provided by the user.

async_handler: AsyncHandlerType instance-attribute

Async-wrapped handler function.

injector: ParameterInjector instance-attribute

Parameter injector for dependency injection.

kwargs: Mapping[str, Any] | None instance-attribute

Keyword arguments to pass to the handler.

error_handler: BusErrorHandlerType | None instance-attribute

Optional per-listener error handler.

app_error_handler_resolver: Callable[[], BusErrorHandlerType | None] | None instance-attribute

Closure that resolves the app-level error handler at dispatch time.

rate_limiter: RateLimiter | None instance-attribute

Rate limiter for debounce/throttle. None when no rate limiting is configured.

task_bucket: TaskBucket instance-attribute

TaskBucket used to spawn the cancellable child handler task for non-parallel modes.

guard: ExecutionModeGuard instance-attribute

Per-listener overlap state machine. Owns the running-task reference and live counters.

mode: ExecutionMode instance-attribute

Resolved overlap mode. Copied from options so dispatch can branch without reading the guard's private state.

handler_short_name: str instance-attribute

Short handler name, captured for the stall WARNING and debug logs.

once: bool = False class-attribute instance-attribute

Whether this invoker fires only once. Intentional copy of ListenerOptions.once — dispatch() needs this but cannot back-reference options without a circular dependency.

backpressure_dropped: int = 0 class-attribute instance-attribute

Count of events dropped at the dispatch acquire gate due to DROP_NEWEST backpressure. Incremented only in BusService.dispatch — one writer, on the event loop, no await between the locked() check and the increment. Live-only; resets on restart, never persisted.

fired: bool = field(default=False, init=False) class-attribute instance-attribute

Guard for once=True: set before the first invocation to prevent double-fire.

pending_done: set[asyncio.Future[None]] = field(default_factory=set, init=False) class-attribute instance-attribute

Unresolved per-invocation completion futures for non-parallel modes.

Each run_with_mode call (single/restart/queued) parks its outer dispatch task on a future that resolves when the handler actually runs (or is dropped/released). A queued trigger accepted into the deque has no live child until drain time, so its future would hang forever if the listener is released first. release_guard resolves every remaining future here so those outer dispatch tasks unwind and _dispatch_pending settles.

create(task_bucket: TaskBucket, handler: HandlerType, kwargs: Mapping[str, Any] | None, options: ListenerOptions, error_handler: BusErrorHandlerType | None = None, app_error_handler_resolver: Callable[[], BusErrorHandlerType | None] | None = None) -> HandlerInvoker classmethod

Construct a HandlerInvoker from a handler and options.

Builds the async wrapper, injector, and rate limiter. Copies options.once.

Parameters:

Name	Type	Description	Default
task_bucket	TaskBucket	TaskBucket for async adapter and rate limiter.	required
handler	HandlerType	The user-supplied handler callable.	required
kwargs	Mapping[str, Any] | None	Optional keyword arguments to pass to the handler.	required
options	ListenerOptions	Behavioral options (once, debounce, throttle).	required
error_handler	BusErrorHandlerType | None	Optional per-listener error handler.	None
app_error_handler_resolver	Callable[[], BusErrorHandlerType | None] | None	Closure for app-level error handler resolution.	None

Source code in src/hassette/bus/listeners.py

@classmethod
def create(
    cls,
    task_bucket: "TaskBucket",
    handler: "HandlerType",
    kwargs: Mapping[str, Any] | None,
    options: ListenerOptions,
    error_handler: "BusErrorHandlerType | None" = None,
    app_error_handler_resolver: "Callable[[], BusErrorHandlerType | None] | None" = None,
) -> "HandlerInvoker":
    """Construct a HandlerInvoker from a handler and options.

    Builds the async wrapper, injector, and rate limiter. Copies options.once.

    Args:
        task_bucket: TaskBucket for async adapter and rate limiter.
        handler: The user-supplied handler callable.
        kwargs: Optional keyword arguments to pass to the handler.
        options: Behavioral options (once, debounce, throttle).
        error_handler: Optional per-listener error handler.
        app_error_handler_resolver: Closure for app-level error handler resolution.
    """
    handler_name = callable_name(handler)
    signature = get_typed_signature(handler)
    async_handler = make_async_handler(handler, task_bucket)
    injector = ParameterInjector(handler_name, signature)

    rate_limiter: RateLimiter | None = None
    if options.debounce is not None or options.throttle is not None:
        rate_limiter = RateLimiter(
            task_bucket=task_bucket,
            debounce=options.debounce,
            throttle=options.throttle,
            handler_name=handler_name,
        )

    return cls(
        orig_handler=handler,
        async_handler=async_handler,
        injector=injector,
        kwargs=kwargs,
        error_handler=error_handler,
        app_error_handler_resolver=app_error_handler_resolver,
        rate_limiter=rate_limiter,
        task_bucket=task_bucket,
        guard=ExecutionModeGuard(options.mode),
        mode=options.mode,
        handler_short_name=callable_short_name(handler),
        once=options.once,
    )

mark_fired() -> None

Mark this once-invoker as having fired. Called by dispatch() and Listener.cancel().

Source code in src/hassette/bus/listeners.py

def mark_fired(self) -> None:
    """Mark this once-invoker as having fired. Called by dispatch() and Listener.cancel()."""
    self.fired = True

set_app_error_handler_resolver(resolver: Callable[[], BusErrorHandlerType | None]) -> None

Set the closure that resolves the app-level error handler at dispatch time.

Source code in src/hassette/bus/listeners.py

def set_app_error_handler_resolver(self, resolver: "Callable[[], BusErrorHandlerType | None]") -> None:
    """Set the closure that resolves the app-level error handler at dispatch time."""
    self.app_error_handler_resolver = resolver

dispatch(invoke_fn: Callable[[], Awaitable[None]]) -> None async

Apply the once-guard, rate limiter, and overlap mode around the given invoke function.

Order: once-guard → rate limiter (whether to start) → mode guard (overlap of started invocations). BusService builds invoke_fn (tracked telemetry); HandlerInvoker wraps it.

Once-guard: if once=True and the invoker has already fired, returns immediately. Safe without a lock — no await between check-and-set.

Mode guard: parallel is a pass-through that awaits invoke_fn inline (byte-for-byte today's behavior, no child task). For single/restart/queued the guard receives a run-and-track callable that spawns a fresh child task through task_bucket; this method then awaits that child so the outer dispatch task stays pending and remains counted by _dispatch_pending. A restart cancellation of the child surfaces as CancelledError inside the child only — it is swallowed here so the outer dispatch task does not crash.

Source code in src/hassette/bus/listeners.py

async def dispatch(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
    """Apply the once-guard, rate limiter, and overlap mode around the given invoke function.

    Order: once-guard → rate limiter (whether to start) → mode guard (overlap of started
    invocations). BusService builds ``invoke_fn`` (tracked telemetry); HandlerInvoker wraps it.

    Once-guard: if ``once=True`` and the invoker has already fired, returns immediately. Safe
    without a lock — no ``await`` between check-and-set.

    Mode guard: ``parallel`` is a pass-through that awaits ``invoke_fn`` inline (byte-for-byte
    today's behavior, no child task). For ``single``/``restart``/``queued`` the guard receives a
    run-and-track callable that spawns a fresh child task through ``task_bucket``; this method
    then awaits that child so the outer dispatch task stays pending and remains counted by
    ``_dispatch_pending``. A ``restart`` cancellation of the child surfaces as ``CancelledError``
    inside the child only — it is swallowed here so the outer dispatch task does not crash.
    """
    if self.once and self.fired:
        return
    if self.once:
        self.mark_fired()

    if self.rate_limiter:
        await self.rate_limiter.call(lambda: self.run_with_mode(invoke_fn))
    else:
        await self.run_with_mode(invoke_fn)

run_with_mode(invoke_fn: Callable[[], Awaitable[None]]) -> None async

Apply the overlap mode guard to a single started invocation.

Source code in src/hassette/bus/listeners.py

async def run_with_mode(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
    """Apply the overlap mode guard to a single started invocation."""
    if self.mode is ExecutionMode.PARALLEL:
        await invoke_fn()
        return

    await run_through_guard(
        guard=self.guard,
        spawn=lambda coro, *, name: self.task_bucket.spawn(coro, name=name),
        pending_done=self.pending_done,
        invoke=invoke_fn,
        warn=self.warn_stalled,
        spawn_name="bus:mode_invocation",
        threshold=STALL_THRESHOLD_SECONDS,
    )

warn_stalled(threshold: float) -> None

Emit the feature's stall WARNING: a non-parallel handler is still holding its guard.

Source code in src/hassette/bus/listeners.py

def warn_stalled(self, threshold: float) -> None:
    """Emit the feature's stall WARNING: a non-parallel handler is still holding its guard."""
    LOGGER.warning(
        "Handler '%s' has held its %s execution-mode guard for over %.0fs and is still running",
        self.handler_short_name,
        self.mode.value,
        threshold,
    )

cancel() -> None

Cancel any pending rate-limiter tasks.

Source code in src/hassette/bus/listeners.py

def cancel(self) -> None:
    """Cancel any pending rate-limiter tasks."""
    if self.rate_limiter:
        self.rate_limiter.cancel()

release_guard() -> None async

Release the execution-mode guard: cancel the in-flight task and drop queued factories.

Called when a listener is cancelled or replaced so no event/listener/app references leak. parallel listeners hold no guard state, so this is a cheap no-op for them.

Queued triggers still parked in the guard's deque never spawn a child once released, so their outer dispatch tasks are parked on done futures that nothing else will resolve. drain_pending_done resolves every remaining one so those tasks unwind and _dispatch_pending settles.

Source code in src/hassette/bus/listeners.py

async def release_guard(self) -> None:
    """Release the execution-mode guard: cancel the in-flight task and drop queued factories.

    Called when a listener is cancelled or replaced so no event/listener/app references leak.
    ``parallel`` listeners hold no guard state, so this is a cheap no-op for them.

    Queued triggers still parked in the guard's deque never spawn a child once released, so
    their outer dispatch tasks are parked on ``done`` futures that nothing else will resolve.
    ``drain_pending_done`` resolves every remaining one so those tasks unwind and
    ``_dispatch_pending`` settles.
    """
    await self.guard.release()
    drain_pending_done(self.pending_done)

invoke(event: Event[Any]) -> None async

Invoke the handler with dependency injection.

Source code in src/hassette/bus/listeners.py

async def invoke(self, event: "Event[Any]") -> None:
    """Invoke the handler with dependency injection."""
    kwargs = self.injector.inject_parameters(event, **(self.kwargs or {}))
    await self.async_handler(**kwargs)

Listener dataclass

A listener for events with a specific topic and handler.

Composes four focused sub-structs (identity, invoker, options, duration_config) plus routing fields (topic, predicate) and minimal runtime state.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class Listener:
    """A listener for events with a specific topic and handler.

    Composes four focused sub-structs (identity, invoker, options, duration_config)
    plus routing fields (topic, predicate) and minimal runtime state.
    """

    logger: Logger
    """Logger for the listener."""

    topic: str
    """Topic the listener is subscribed to."""

    predicate: "Predicate | None"
    """Predicate to filter events before invoking the handler."""

    identity: ListenerIdentity
    """Ownership and telemetry identity fields."""

    invoker: HandlerInvoker
    """Handler callable, dispatch engine, and once-guard."""

    options: ListenerOptions
    """Behavioral execution parameters."""

    duration_config: DurationConfig | None
    """Duration-hold configuration and timer. None for non-duration listeners."""

    listener_id: int = field(default_factory=next_id, init=False)
    """Unique identifier for the listener instance."""

    _cancelled: bool = field(default=False, init=False, repr=False)
    """Set by cancel() to signal that a pending add_listener task should skip route insertion."""

    db_id: int | None = field(default=None, init=False)
    """Database row ID for this listener. Set by the executor after persistence; None until then."""

    @property
    def is_cancelled(self) -> bool:
        """Whether this listener has been cancelled. Read-only — use cancel() to set."""
        return self._cancelled

    def mark_registered(self, db_id: int) -> None:
        """Set the database ID after persistence. One-time assignment by BusService."""
        if self.db_id is not None:
            self.logger.warning(
                "Listener %s already registered with db_id=%s, ignoring new db_id=%s",
                self.listener_id,
                self.db_id,
                db_id,
            )
            return
        self.db_id = db_id

    def cancel(self) -> None:
        """Cancel the listener: set the cancelled flag and stop any pending tasks.

        Sets _cancelled flag and calls invoker.mark_fired(). For once=True listeners, mark_fired()
        prevents the handler from starting on any in-flight dispatch task that has not yet checked
        the once-guard. For once=False listeners, mark_fired() has no effect on dispatch (the
        once-guard is not consulted); protection comes from release_guard() cancelling the in-flight
        child task — spawned asynchronously, so there is a small window before it takes effect.
        Also cancels the rate limiter and duration timer, and releases the execution-mode guard
        (cancelling any in-flight handler task and dropping queued factories) so no event/listener
        references leak.

        Terminal operation: the listener must not be reused after this call.
        """
        self._cancelled = True
        self.invoker.mark_fired()
        self.invoker.cancel()
        if self.duration_config is not None:
            self.duration_config.cancel_timer()
        # release_guard is async (it awaits the cancelled task's settling under a lock); cancel()
        # is sync, so spawn the release on the same bucket that runs handler tasks. For ``parallel``
        # listeners this is a cheap no-op; for the others it drops the in-flight task and queue.
        self.invoker.task_bucket.spawn(self.invoker.release_guard(), name="bus:release_guard")

    def config_matches(self, other: "Listener") -> bool:
        """Check whether two listeners represent the same logical configuration.

        Compares handler callable, filter predicate, timing options (once, debounce,
        throttle, timeout, timeout_disabled, priority), the execution mode, handler kwargs,
        per-registration error handler (by identity), and duration configuration scalars.

        Does not compare runtime state: listener_id, db_id, _cancelled, or the
        attached DurationTimer. Lambda/closure predicates and callable conditions
        compare by identity — two fresh lambdas with identical bodies will report
        drift. Use non-lambda predicates or if_exists='replace' to avoid this.
        """
        return (
            self.invoker.orig_handler == other.invoker.orig_handler
            and self.predicate == other.predicate
            and self.options.once == other.options.once
            and self.options.debounce == other.options.debounce
            and self.options.throttle == other.options.throttle
            and self.options.timeout == other.options.timeout
            and self.options.timeout_disabled == other.options.timeout_disabled
            and self.options.priority == other.options.priority
            and self.options.mode == other.options.mode
            and self.options.backpressure == other.options.backpressure
            and self.invoker.kwargs == other.invoker.kwargs
            and self.invoker.error_handler is other.invoker.error_handler
            and _duration_configs_match(self.duration_config, other.duration_config)
        )

    def diff_fields(self, other: "Listener") -> list[str]:
        """Return configuration field names that differ between two listeners.

        Compares the same fields as config_matches(). Returns a stable-ordered list
        of field names (e.g. 'handler', 'predicate', 'once', 'debounce', ...) for use
        in drift error messages.
        """
        changed: list[str] = []
        if self.invoker.orig_handler != other.invoker.orig_handler:
            changed.append("handler")
        if self.predicate != other.predicate:
            changed.append("predicate")
        if self.options.once != other.options.once:
            changed.append("once")
        if self.options.debounce != other.options.debounce:
            changed.append("debounce")
        if self.options.throttle != other.options.throttle:
            changed.append("throttle")
        if self.options.timeout != other.options.timeout:
            changed.append("timeout")
        if self.options.timeout_disabled != other.options.timeout_disabled:
            changed.append("timeout_disabled")
        if self.options.priority != other.options.priority:
            changed.append("priority")
        if self.options.mode != other.options.mode:
            changed.append("mode")
        if self.options.backpressure != other.options.backpressure:
            changed.append("backpressure")
        if self.invoker.kwargs != other.invoker.kwargs:
            changed.append("kwargs")
        if self.invoker.error_handler is not other.invoker.error_handler:
            changed.append("error_handler")
        if not _duration_configs_match(self.duration_config, other.duration_config):
            changed.append("duration_config")
        return changed

    def matches(self, ev: "Event[Any]") -> bool:
        """Check if the event matches the listener's predicate.

        Raises if the predicate raises — the caller (BusService.dispatch) handles
        isolation, telemetry recording, and error-handler routing.
        """
        if self.predicate is None:
            return True
        matched = self.predicate(ev)
        verdict = "matched" if matched else "did not match"
        self.logger.debug("Listener %s %s predicate for event: %s", self, verdict, ev)
        return matched

    def __repr__(self) -> str:
        return f"Listener<{self.identity.owner_id} - {self.identity.handler_short_name}>"

    @classmethod
    def create(
        cls,
        topic: str,
        identity: ListenerIdentity,
        options: ListenerOptions,
        invoker: HandlerInvoker,
        where: WhereClause = None,
        duration_config: DurationConfig | None = None,
        logger: Logger = LOGGER,
    ) -> "Listener":
        """Create a Listener from pre-built sub-structs.

        Cross-concern validation (duration + debounce incompatibility) runs
        here since it spans two sub-structs.
        """
        if duration_config is not None and duration_config.duration is not None:
            if options.debounce is not None:
                raise ValueError("Cannot combine 'duration' with 'debounce'")
            if options.throttle is not None:
                raise ValueError("Cannot combine 'duration' with 'throttle'")

        pred = normalize_where(where)
        return cls(
            logger=logger,
            topic=topic,
            predicate=pred,
            identity=identity,
            invoker=invoker,
            options=options,
            duration_config=duration_config,
        )

    @classmethod
    def create_cancel_listener(
        cls,
        task_bucket: "TaskBucket",
        owner_id: str,
        topic: str,
        handler: "HandlerType",
        predicate: "Predicate | None" = None,
    ) -> "Listener":
        """Create a framework cancel-listener with sensible defaults.

        Produces a listener with source_tier='framework'. No rate limiter,
        no error handler, no duration config.
        """
        handler_name = callable_name(handler)
        short_name = callable_short_name(handler)

        identity = ListenerIdentity(
            owner_id=owner_id,
            handler_name=handler_name,
            handler_short_name=short_name,
            source_tier="framework",
        )

        # Cancel-listeners are source_tier='framework' but bypass _on_internal's tier-aware
        # resolution, so set parallel explicitly to match the framework-tier default. They fire at
        # most once per timer, so parallel is the safe internal default.
        options = ListenerOptions(mode=ExecutionMode.PARALLEL)

        invoker = HandlerInvoker.create(
            task_bucket=task_bucket,
            handler=handler,
            kwargs=None,
            options=options,
            error_handler=None,
        )

        return cls(
            logger=LOGGER,
            topic=topic,
            predicate=predicate,
            identity=identity,
            invoker=invoker,
            options=options,
            duration_config=None,
        )

logger: Logger instance-attribute

Logger for the listener.

topic: str instance-attribute

Topic the listener is subscribed to.

predicate: Predicate | None instance-attribute

Predicate to filter events before invoking the handler.

identity: ListenerIdentity instance-attribute

Ownership and telemetry identity fields.

invoker: HandlerInvoker instance-attribute

Handler callable, dispatch engine, and once-guard.

options: ListenerOptions instance-attribute

Behavioral execution parameters.

duration_config: DurationConfig | None instance-attribute

Duration-hold configuration and timer. None for non-duration listeners.

listener_id: int = field(default_factory=next_id, init=False) class-attribute instance-attribute

Unique identifier for the listener instance.

db_id: int | None = field(default=None, init=False) class-attribute instance-attribute

Database row ID for this listener. Set by the executor after persistence; None until then.

is_cancelled: bool property

Whether this listener has been cancelled. Read-only — use cancel() to set.

mark_registered(db_id: int) -> None

Set the database ID after persistence. One-time assignment by BusService.

Source code in src/hassette/bus/listeners.py

def mark_registered(self, db_id: int) -> None:
    """Set the database ID after persistence. One-time assignment by BusService."""
    if self.db_id is not None:
        self.logger.warning(
            "Listener %s already registered with db_id=%s, ignoring new db_id=%s",
            self.listener_id,
            self.db_id,
            db_id,
        )
        return
    self.db_id = db_id

cancel() -> None

Cancel the listener: set the cancelled flag and stop any pending tasks.

Sets _cancelled flag and calls invoker.mark_fired(). For once=True listeners, mark_fired() prevents the handler from starting on any in-flight dispatch task that has not yet checked the once-guard. For once=False listeners, mark_fired() has no effect on dispatch (the once-guard is not consulted); protection comes from release_guard() cancelling the in-flight child task — spawned asynchronously, so there is a small window before it takes effect. Also cancels the rate limiter and duration timer, and releases the execution-mode guard (cancelling any in-flight handler task and dropping queued factories) so no event/listener references leak.

Terminal operation: the listener must not be reused after this call.

Source code in src/hassette/bus/listeners.py

def cancel(self) -> None:
    """Cancel the listener: set the cancelled flag and stop any pending tasks.

    Sets _cancelled flag and calls invoker.mark_fired(). For once=True listeners, mark_fired()
    prevents the handler from starting on any in-flight dispatch task that has not yet checked
    the once-guard. For once=False listeners, mark_fired() has no effect on dispatch (the
    once-guard is not consulted); protection comes from release_guard() cancelling the in-flight
    child task — spawned asynchronously, so there is a small window before it takes effect.
    Also cancels the rate limiter and duration timer, and releases the execution-mode guard
    (cancelling any in-flight handler task and dropping queued factories) so no event/listener
    references leak.

    Terminal operation: the listener must not be reused after this call.
    """
    self._cancelled = True
    self.invoker.mark_fired()
    self.invoker.cancel()
    if self.duration_config is not None:
        self.duration_config.cancel_timer()
    # release_guard is async (it awaits the cancelled task's settling under a lock); cancel()
    # is sync, so spawn the release on the same bucket that runs handler tasks. For ``parallel``
    # listeners this is a cheap no-op; for the others it drops the in-flight task and queue.
    self.invoker.task_bucket.spawn(self.invoker.release_guard(), name="bus:release_guard")

config_matches(other: Listener) -> bool

Check whether two listeners represent the same logical configuration.

Compares handler callable, filter predicate, timing options (once, debounce, throttle, timeout, timeout_disabled, priority), the execution mode, handler kwargs, per-registration error handler (by identity), and duration configuration scalars.

Does not compare runtime state: listener_id, db_id, _cancelled, or the attached DurationTimer. Lambda/closure predicates and callable conditions compare by identity — two fresh lambdas with identical bodies will report drift. Use non-lambda predicates or if_exists='replace' to avoid this.

Source code in src/hassette/bus/listeners.py

def config_matches(self, other: "Listener") -> bool:
    """Check whether two listeners represent the same logical configuration.

    Compares handler callable, filter predicate, timing options (once, debounce,
    throttle, timeout, timeout_disabled, priority), the execution mode, handler kwargs,
    per-registration error handler (by identity), and duration configuration scalars.

    Does not compare runtime state: listener_id, db_id, _cancelled, or the
    attached DurationTimer. Lambda/closure predicates and callable conditions
    compare by identity — two fresh lambdas with identical bodies will report
    drift. Use non-lambda predicates or if_exists='replace' to avoid this.
    """
    return (
        self.invoker.orig_handler == other.invoker.orig_handler
        and self.predicate == other.predicate
        and self.options.once == other.options.once
        and self.options.debounce == other.options.debounce
        and self.options.throttle == other.options.throttle
        and self.options.timeout == other.options.timeout
        and self.options.timeout_disabled == other.options.timeout_disabled
        and self.options.priority == other.options.priority
        and self.options.mode == other.options.mode
        and self.options.backpressure == other.options.backpressure
        and self.invoker.kwargs == other.invoker.kwargs
        and self.invoker.error_handler is other.invoker.error_handler
        and _duration_configs_match(self.duration_config, other.duration_config)
    )

diff_fields(other: Listener) -> list[str]

Return configuration field names that differ between two listeners.

Compares the same fields as config_matches(). Returns a stable-ordered list of field names (e.g. 'handler', 'predicate', 'once', 'debounce', ...) for use in drift error messages.

Source code in src/hassette/bus/listeners.py

def diff_fields(self, other: "Listener") -> list[str]:
    """Return configuration field names that differ between two listeners.

    Compares the same fields as config_matches(). Returns a stable-ordered list
    of field names (e.g. 'handler', 'predicate', 'once', 'debounce', ...) for use
    in drift error messages.
    """
    changed: list[str] = []
    if self.invoker.orig_handler != other.invoker.orig_handler:
        changed.append("handler")
    if self.predicate != other.predicate:
        changed.append("predicate")
    if self.options.once != other.options.once:
        changed.append("once")
    if self.options.debounce != other.options.debounce:
        changed.append("debounce")
    if self.options.throttle != other.options.throttle:
        changed.append("throttle")
    if self.options.timeout != other.options.timeout:
        changed.append("timeout")
    if self.options.timeout_disabled != other.options.timeout_disabled:
        changed.append("timeout_disabled")
    if self.options.priority != other.options.priority:
        changed.append("priority")
    if self.options.mode != other.options.mode:
        changed.append("mode")
    if self.options.backpressure != other.options.backpressure:
        changed.append("backpressure")
    if self.invoker.kwargs != other.invoker.kwargs:
        changed.append("kwargs")
    if self.invoker.error_handler is not other.invoker.error_handler:
        changed.append("error_handler")
    if not _duration_configs_match(self.duration_config, other.duration_config):
        changed.append("duration_config")
    return changed

matches(ev: Event[Any]) -> bool

Check if the event matches the listener's predicate.

Raises if the predicate raises — the caller (BusService.dispatch) handles isolation, telemetry recording, and error-handler routing.

Source code in src/hassette/bus/listeners.py

def matches(self, ev: "Event[Any]") -> bool:
    """Check if the event matches the listener's predicate.

    Raises if the predicate raises — the caller (BusService.dispatch) handles
    isolation, telemetry recording, and error-handler routing.
    """
    if self.predicate is None:
        return True
    matched = self.predicate(ev)
    verdict = "matched" if matched else "did not match"
    self.logger.debug("Listener %s %s predicate for event: %s", self, verdict, ev)
    return matched

create(topic: str, identity: ListenerIdentity, options: ListenerOptions, invoker: HandlerInvoker, where: WhereClause = None, duration_config: DurationConfig | None = None, logger: Logger = LOGGER) -> Listener classmethod

Create a Listener from pre-built sub-structs.

Cross-concern validation (duration + debounce incompatibility) runs here since it spans two sub-structs.

Source code in src/hassette/bus/listeners.py

@classmethod
def create(
    cls,
    topic: str,
    identity: ListenerIdentity,
    options: ListenerOptions,
    invoker: HandlerInvoker,
    where: WhereClause = None,
    duration_config: DurationConfig | None = None,
    logger: Logger = LOGGER,
) -> "Listener":
    """Create a Listener from pre-built sub-structs.

    Cross-concern validation (duration + debounce incompatibility) runs
    here since it spans two sub-structs.
    """
    if duration_config is not None and duration_config.duration is not None:
        if options.debounce is not None:
            raise ValueError("Cannot combine 'duration' with 'debounce'")
        if options.throttle is not None:
            raise ValueError("Cannot combine 'duration' with 'throttle'")

    pred = normalize_where(where)
    return cls(
        logger=logger,
        topic=topic,
        predicate=pred,
        identity=identity,
        invoker=invoker,
        options=options,
        duration_config=duration_config,
    )

create_cancel_listener(task_bucket: TaskBucket, owner_id: str, topic: str, handler: HandlerType, predicate: Predicate | None = None) -> Listener classmethod

Create a framework cancel-listener with sensible defaults.

Produces a listener with source_tier='framework'. No rate limiter, no error handler, no duration config.

Source code in src/hassette/bus/listeners.py

@classmethod
def create_cancel_listener(
    cls,
    task_bucket: "TaskBucket",
    owner_id: str,
    topic: str,
    handler: "HandlerType",
    predicate: "Predicate | None" = None,
) -> "Listener":
    """Create a framework cancel-listener with sensible defaults.

    Produces a listener with source_tier='framework'. No rate limiter,
    no error handler, no duration config.
    """
    handler_name = callable_name(handler)
    short_name = callable_short_name(handler)

    identity = ListenerIdentity(
        owner_id=owner_id,
        handler_name=handler_name,
        handler_short_name=short_name,
        source_tier="framework",
    )

    # Cancel-listeners are source_tier='framework' but bypass _on_internal's tier-aware
    # resolution, so set parallel explicitly to match the framework-tier default. They fire at
    # most once per timer, so parallel is the safe internal default.
    options = ListenerOptions(mode=ExecutionMode.PARALLEL)

    invoker = HandlerInvoker.create(
        task_bucket=task_bucket,
        handler=handler,
        kwargs=None,
        options=options,
        error_handler=None,
    )

    return cls(
        logger=LOGGER,
        topic=topic,
        predicate=predicate,
        identity=identity,
        invoker=invoker,
        options=options,
        duration_config=None,
    )

ListenerIdentity dataclass

Groups ownership and telemetry fields that identify who registered a listener and where it came from.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class ListenerIdentity:
    """Groups ownership and telemetry fields that identify who registered a listener and where it came from."""

    owner_id: str
    """Unique string identifier for the owner of the listener."""

    handler_name: str
    """Human-readable fully-qualified name for the handler, computed once at creation time."""

    handler_short_name: str
    """Short (last-segment) name for the handler, computed once at creation time."""

    app_key: str = ""
    """Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners."""

    instance_index: int = 0
    """App instance index for DB registration. 0 for non-App owners."""

    instance_name: str | None = None
    """App instance name, precomputed at registration time from the owning app's app_config.instance_name.
    None for framework-tier listeners and non-App owners."""

    name: str | None = None
    """Optional stable name for the listener (the name= escape hatch on Bus.on())."""

    source_tier: SourceTier = "app"
    """Whether this listener originates from a user app or the framework itself."""

    source_location: str = ""
    """Captured source location (file:line) of the user code that registered this listener."""

    registration_source: str = ""
    """Captured source code snippet of the registration call."""

owner_id: str instance-attribute

Unique string identifier for the owner of the listener.

handler_name: str instance-attribute

Human-readable fully-qualified name for the handler, computed once at creation time.

handler_short_name: str instance-attribute

Short (last-segment) name for the handler, computed once at creation time.

app_key: str = '' class-attribute instance-attribute

Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners.

instance_index: int = 0 class-attribute instance-attribute

App instance index for DB registration. 0 for non-App owners.

instance_name: str | None = None class-attribute instance-attribute

App instance name, precomputed at registration time from the owning app's app_config.instance_name. None for framework-tier listeners and non-App owners.

name: str | None = None class-attribute instance-attribute

Optional stable name for the listener (the name= escape hatch on Bus.on()).

source_tier: SourceTier = 'app' class-attribute instance-attribute

Whether this listener originates from a user app or the framework itself.

source_location: str = '' class-attribute instance-attribute

Captured source location (file:line) of the user code that registered this listener.

registration_source: str = '' class-attribute instance-attribute

Captured source code snippet of the registration call.

ListenerOptions dataclass

Behavioral timing parameters (once, debounce, throttle, timeout, priority) with validation.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class ListenerOptions:
    """Behavioral timing parameters (once, debounce, throttle, timeout, priority) with validation."""

    once: bool = False
    """Whether the listener should be removed after one invocation."""

    debounce: float | None = None
    """Debounce delay in seconds. Events reset the timer; handler fires after the quiet period."""

    throttle: float | None = None
    """Throttle interval in seconds. At most one handler execution per window; extras are dropped."""

    timeout: float | None = None
    """Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
    None means fall through to the config default."""

    timeout_disabled: bool = False
    """When True, disables timeout enforcement for this listener regardless of config."""

    priority: int = 0
    """Priority for listener ordering. Higher values run first. Default is 0 for app handlers."""

    mode: ExecutionMode = ExecutionMode.SINGLE
    """Overlap behavior when a trigger fires while a prior invocation still runs.

    ``single`` drops the re-fire, ``restart`` cancels-and-replaces, ``queued`` serializes,
    ``parallel`` runs concurrently. The tier-aware default (framework→parallel, app→single)
    is applied by the registration path, not by this dataclass default. Suppressed/dropped
    counts are live-only diagnostics held on the per-listener guard, reset on restart.
    """

    backpressure: BackpressurePolicy = BackpressurePolicy.BLOCK
    """Saturation policy for this listener when the dispatch concurrency semaphore is at capacity.

    ``block`` (default) waits for a slot, preserving today's behavior unchanged.
    ``drop_newest`` skips the event immediately when the bus is saturated — the handler is not
    invoked and one drop is recorded. Acts at the dispatch acquire gate, orthogonal to
    ``mode``/``debounce``/``throttle`` which act inside the invoker.
    """

    def __post_init__(self) -> None:
        # Coerce a raw string mode (arriving via the Options TypedDict or str ergonomics) into
        # the enum. An unknown value fails coercion — surface it as a clear ValueError.
        if not isinstance(self.mode, ExecutionMode):
            try:
                self.mode = ExecutionMode(self.mode)
            except ValueError as exc:
                valid = ", ".join(repr(m.value) for m in ExecutionMode)
                raise ValueError(f"Invalid execution mode {self.mode!r}; must be one of {valid}") from exc
        # Coerce a raw string backpressure policy the same way.
        if not isinstance(self.backpressure, BackpressurePolicy):
            try:
                self.backpressure = BackpressurePolicy(self.backpressure)
            except ValueError as exc:
                valid = ", ".join(repr(m.value) for m in BackpressurePolicy)
                raise ValueError(f"Invalid backpressure policy {self.backpressure!r}; must be one of {valid}") from exc
        if self.debounce is not None and self.debounce <= 0:
            raise ValueError("'debounce' must be a positive number")
        if self.throttle is not None and self.throttle <= 0:
            raise ValueError("'throttle' must be a positive number")
        if self.debounce is not None and self.throttle is not None:
            raise ValueError("Cannot specify both 'debounce' and 'throttle' parameters")
        if self.once and (self.debounce is not None or self.throttle is not None):
            raise ValueError("Cannot combine 'once=True' with 'debounce' or 'throttle'")
        if self.timeout is not None and (isinstance(self.timeout, bool) or self.timeout <= 0):
            raise ValueError("timeout must be a positive number")
        if self.timeout_disabled and self.timeout is not None:
            raise ValueError("Cannot specify both 'timeout' and 'timeout_disabled=True'")

once: bool = False class-attribute instance-attribute

Whether the listener should be removed after one invocation.

debounce: float | None = None class-attribute instance-attribute

Debounce delay in seconds. Events reset the timer; handler fires after the quiet period.

throttle: float | None = None class-attribute instance-attribute

Throttle interval in seconds. At most one handler execution per window; extras are dropped.

timeout: float | None = None class-attribute instance-attribute

Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config. None means fall through to the config default.

timeout_disabled: bool = False class-attribute instance-attribute

When True, disables timeout enforcement for this listener regardless of config.

priority: int = 0 class-attribute instance-attribute

Priority for listener ordering. Higher values run first. Default is 0 for app handlers.

mode: ExecutionMode = ExecutionMode.SINGLE class-attribute instance-attribute

Overlap behavior when a trigger fires while a prior invocation still runs.

single drops the re-fire, restart cancels-and-replaces, queued serializes, parallel runs concurrently. The tier-aware default (framework→parallel, app→single) is applied by the registration path, not by this dataclass default. Suppressed/dropped counts are live-only diagnostics held on the per-listener guard, reset on restart.

backpressure: BackpressurePolicy = BackpressurePolicy.BLOCK class-attribute instance-attribute

Saturation policy for this listener when the dispatch concurrency semaphore is at capacity.

block (default) waits for a slot, preserving today's behavior unchanged. drop_newest skips the event immediately when the bus is saturated — the handler is not invoked and one drop is recorded. Acts at the dispatch acquire gate, orthogonal to mode/debounce/throttle which act inside the invoker.

Subscription dataclass

A subscription to an event topic with a specific listener key.

This class is used to manage the lifecycle of a listener, allowing it to be cancelled or managed within a context.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class Subscription:
    """A subscription to an event topic with a specific listener key.

    This class is used to manage the lifecycle of a listener, allowing it to be cancelled
    or managed within a context.
    """

    listener: Listener
    """The listener associated with this subscription."""

    unsubscribe: Callable[[], None]
    """Function to call to unsubscribe the listener."""

    def cancel(self) -> None:
        """Cancel the subscription by calling the unsubscribe function."""
        self.unsubscribe()

listener: Listener instance-attribute

The listener associated with this subscription.

unsubscribe: Callable[[], None] instance-attribute

Function to call to unsubscribe the listener.

cancel() -> None

Cancel the subscription by calling the unsubscribe function.

Source code in src/hassette/bus/listeners.py

def cancel(self) -> None:
    """Cancel the subscription by calling the unsubscribe function."""
    self.unsubscribe()

BusSyncFacade

Bases: Resource

Synchronous facade for the event bus.

This class provides synchronous methods that wrap the asynchronous registration methods of the Bus class, allowing listeners to be registered from synchronous code (for example, an AppSync lifecycle hook running in a worker thread).

These methods must not be called from within the event loop; doing so raises a RuntimeError. Use the asynchronous methods on Bus directly when operating within an event loop.

Source code in src/hassette/bus/sync.py

class BusSyncFacade(Resource):
    """Synchronous facade for the event bus.

    This class provides synchronous methods that wrap the asynchronous registration methods of
    the Bus class, allowing listeners to be registered from synchronous code (for example, an
    ``AppSync`` lifecycle hook running in a worker thread).

    These methods must not be called from within the event loop; doing so raises a RuntimeError.
    Use the asynchronous methods on ``Bus`` directly when operating within an event loop.
    """

    _bus: "Bus"

    def __init__(self, hassette: "Hassette", *, bus: "Bus", parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        self._bus = bus

    async def on_initialize(self) -> None:
        mark_ready(self, reason="Synchronous Bus facade initialized")

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        return self.hassette.config.logging.bus_service

    def add_listener(self, listener: "Listener", *, if_exists: IfExistsPolicy = "error") -> Subscription:
        """Add a pre-built listener to the bus.

        This is the direct entry point for callers that construct a ``Listener``
        externally. The normal registration flow (``on_state_change``, ``on()``,
        etc.) goes through ``_on_internal`` instead.

        Args:
            listener: The pre-built listener to add.
            if_exists: Behavior when a listener with the same natural key already exists.
                ``"error"`` (default) raises ``DuplicateListenerError``.
                ``"skip"`` returns a subscription to the existing listener when configs match.
                ``"replace"`` cancels the existing listener and registers the new one.

        Returns:
            A subscription to the added listener (or the existing listener on skip).

        Raises:
            ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
                listeners, including once-listeners; cancel-listeners bypass this path entirely).
            DuplicateListenerError: If the listener's natural key is already registered and
                ``if_exists="error"``.
        """
        return self.task_bucket.run_sync(self._bus.add_listener(listener, if_exists=if_exists))

    def emit(self, topic: str, data: object) -> None:
        """Broadcast data to all subscribers of the given topic.

        Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
        If the internal event stream is closed (during shutdown), the event is silently dropped.
        """
        return self.task_bucket.run_sync(self._bus.emit(topic, data))

    def on(
        self,
        *,
        topic: str,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        once: bool = False,
        debounce: float | None = None,
        throttle: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        backpressure: "BackpressurePolicy | str | None" = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
    ) -> Subscription:
        """Subscribe to an event topic with optional filtering and modifiers.

        This is the public registration method for raw topic subscriptions.
        Registration completes before the call returns — ``sub.listener.db_id`` is a valid
        integer immediately on return.

        Args:
            topic: The event topic to listen to.
            handler: The function to call when the event matches.
            where: Optional predicates to filter events. These can be custom callables or predefined predicates from
                `hassette.event_handling.predicates`. They will receive the full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            once: If True, the handler will be called only once and then removed.
            debounce: If set, applies a debounce to the handler.
            throttle: If set, applies a throttle to the handler.
            timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
                None means fall through to the config default.
            timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
            mode: Overlap behavior when a trigger fires while a prior invocation still runs —
                ``"single"``, ``"restart"``, ``"queued"``, or ``"parallel"``. When omitted, the
                effective default is tier-aware: ``parallel`` for framework listeners, ``single``
                for app listeners. Suppressed/dropped counts are live-only diagnostics, reset on
                restart.
            backpressure: Saturation policy when the global dispatch concurrency semaphore is full.
                ``"block"`` (default) waits for a slot; ``"drop_newest"`` skips the event immediately
                and records one drop on the listener. When omitted, the effective default is ``block``.
            name: Required. Stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
                empty string raises ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            if_exists: Behavior when a listener with the same natural key already exists.
                ``"error"`` (default) raises ``DuplicateListenerError``. ``"skip"`` returns the
                existing listener's subscription when the configurations match, and raises
                ``ValueError`` if the configuration has drifted. ``"replace"`` cancels the
                existing listener and registers the new one in its place.

        Returns:
            A subscription object. ``sub.cancel()`` removes the listener.
            ``sub.listener.db_id`` is a valid integer immediately on return.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered and ``if_exists="error"`` (the default).
            ValueError: If ``if_exists="skip"`` and a listener with the same ``(name, topic)``
                exists but with a different configuration (the message lists the changed fields).
        """
        return self.task_bucket.run_sync(
            self._bus.on(
                topic=topic,
                handler=handler,
                where=where,
                kwargs=kwargs,
                once=once,
                debounce=debounce,
                throttle=throttle,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                backpressure=backpressure,
                name=name,
                on_error=on_error,
                if_exists=if_exists,
            )
        )

    def on_state_change(
        self,
        entity_id: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to state changes for a specific entity.

        Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
                will be used to compare the old and new state values.
            changed_from: A value or callable that will be used to filter state changes *from* this value.
            changed_to: A value or callable that will be used to filter state changes *to* this value.
            where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
                full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
                empty string raises ``ListenerNameRequiredError`` at call time.
            **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
                ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

                ``mode`` controls overlap behavior when a trigger fires while a prior invocation
                is still running: ``"single"`` drops the re-fire (the default for app handlers),
                ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
                (bounded at 10 pending), ``"parallel"`` runs concurrently. When omitted, the
                tier-aware default applies: ``"single"`` for app handlers, ``"parallel"`` for
                framework-internal listeners. An explicit ``mode=`` always wins. Suppressed
                (``single``) and dropped (``queued`` cap) events log at DEBUG only.
                Suppressed/dropped counts are live-only diagnostics, reset on restart.
                See `Execution Modes <https://hassette.dev/core-concepts/bus/execution-modes/>`_.

                ``backpressure`` controls what this listener does when the dispatch concurrency
                semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
                today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
                at the dispatch acquire point (global bus saturation), orthogonal to
                ``mode``/``debounce``/``throttle``.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
            removes the listener from routing.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered on this bus in the current session and ``if_exists="error"``
                (the default).
        """
        return self.task_bucket.run_sync(
            self._bus.on_state_change(
                entity_id,
                handler=handler,
                changed=changed,
                changed_from=changed_from,
                changed_to=changed_to,
                where=where,
                kwargs=kwargs,
                immediate=immediate,
                duration=duration,
                name=name,
                on_error=on_error,
                **opts,
            )
        )

    def on_attribute_change(
        self,
        entity_id: str,
        attr: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to state change events for a specific entity's attribute.

        Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            attr: The attribute name to filter changes on (e.g., "volume").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
                it will be used to compare the old and new attribute values.
            changed_from: A value or callable that will be used to filter attribute changes *from* this value.
            changed_to: A value or callable that will be used to filter attribute changes *to* this value.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier. Omitting it entirely raises ``TypeError``
                (no default value); passing an empty string raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
                ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

                ``mode`` controls overlap behavior when a trigger fires while a prior invocation
                is still running: ``"single"`` drops the re-fire (the default for app handlers),
                ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
                (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
                counts are live-only diagnostics, reset on restart.

                ``backpressure`` controls what this listener does when the dispatch concurrency
                semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
                today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
                at the dispatch acquire point (global bus saturation), orthogonal to
                ``mode``/``debounce``/``throttle``.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered and ``if_exists="error"`` (the default).
        """
        return self.task_bucket.run_sync(
            self._bus.on_attribute_change(
                entity_id,
                attr,
                handler=handler,
                changed=changed,
                changed_from=changed_from,
                changed_to=changed_to,
                where=where,
                kwargs=kwargs,
                immediate=immediate,
                duration=duration,
                name=name,
                on_error=on_error,
                **opts,
            )
        )

    def on_call_service(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to service call events.

        Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            domain: The domain to filter service calls (e.g., "light").
            service: The service to filter service calls (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier for this listener. Omitting it entirely
                raises ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
                ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

                ``mode`` controls overlap behavior when a trigger fires while a prior invocation
                is still running: ``"single"`` drops the re-fire (the default for app handlers),
                ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
                (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
                counts are live-only diagnostics, reset on restart.

                ``backpressure`` controls what this listener does when the dispatch concurrency
                semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
                today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
                at the dispatch acquire point (global bus saturation), orthogonal to
                ``mode``/``debounce``/``throttle``.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered and ``if_exists="error"`` (the default).
        """
        return self.task_bucket.run_sync(
            self._bus.on_call_service(
                domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_component_loaded(
        self,
        component: str | None = None,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to component loaded events.

        Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            component: The component to filter load events (e.g., "light").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_component_loaded(
                component, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_service_registered(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to service registered events.

        Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            domain: The domain to filter service registrations (e.g., "light").
            service: The service to filter service registrations (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_service_registered(
                domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_homeassistant_restart(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant restart events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_homeassistant_restart(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_homeassistant_start(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant start events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_homeassistant_start(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_homeassistant_stop(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant stop events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_homeassistant_stop(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_hassette_service_status(
        self,
        status: ResourceStatus | None = None,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service status events.

        Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            status: The status to filter events (e.g., ResourceStatus.STARTED).
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
                default value); passing an empty string raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_status(
                status, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_hassette_service_failed(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service failed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
                default value); passing an empty string raises ``ListenerNameRequiredError``
                at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_failed(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_hassette_service_crashed(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service crashed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_crashed(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_hassette_service_started(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service started events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_started(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_websocket_connected(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to websocket connected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_websocket_connected(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_websocket_disconnected(
        self,
        *,
        handler: "HandlerType",
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to websocket disconnected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_websocket_disconnected(
                handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_app_state_changed(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        status: ResourceStatus | None = None,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instance state change events.

        Registration completes before the call returns.
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            status: Filter events for a specific status.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_app_state_changed(
                handler=handler,
                app_key=app_key,
                status=status,
                where=where,
                kwargs=kwargs,
                name=name,
                on_error=on_error,
                **opts,
            )
        )

    def on_app_running(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instances reaching RUNNING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_app_running(
                handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_app_stopping(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: WhereClause = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instances entering STOPPING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable name for this listener. Omitting it entirely raises
                ``TypeError`` (no default value); passing an empty string raises
                ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.
            **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return self.task_bucket.run_sync(
            self._bus.on_app_stopping(
                handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_error(self, handler: "BusErrorHandlerType") -> None:
        """Register an app-level error handler for this bus.

        The handler is called when any listener on this bus raises an exception
        (including ``TimeoutError``) and the listener does not have its own
        per-registration error handler.

        This is an app-level fallback — it is resolved at dispatch time, not at listener
        registration time. A later call to ``on_error()`` replaces any previously registered
        handler.

        Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
        app shutdown may be cancelled before they complete. Do not rely on error handlers
        for delivery-critical alerting during system teardown.

        Args:
            handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`.
        """
        return self._bus.on_error(handler)

    def remove_listener(self, listener: "Listener") -> None:
        """Remove a listener from the bus and persist cancellation to the database.

        Pops the natural key from the in-memory registry, removes the listener from routing
        (via BusService), and — when ``db_id`` is set — spawns ``mark_listener_cancelled``
        on ``bus_service.task_bucket`` so the write survives resource shutdown, mirroring
        ``Scheduler.cancel_job``.

        BusService.remove_listener also fires _on_listener_removed, but that callback only
        spawns mark_listener_cancelled when the key is still present (once-fire path). Because
        this method pops the key first, the callback's spawn is skipped — avoiding a double write.
        """
        return self._bus.remove_listener(listener)

    def remove_all_listeners(self) -> None:
        """Remove all listeners owned by this bus's owner."""
        return self._bus.remove_all_listeners()

    def get_listeners(self) -> list["Listener"]:
        """Get all listeners owned by this bus's owner."""
        return self._bus.get_listeners()

add_listener(listener: Listener, *, if_exists: IfExistsPolicy = 'error') -> Subscription

Add a pre-built listener to the bus.

This is the direct entry point for callers that construct a Listener externally. The normal registration flow (on_state_change, on(), etc.) goes through _on_internal instead.

Parameters:

Name	Type	Description	Default
listener	Listener	The pre-built listener to add.	required
if_exists	IfExistsPolicy	Behavior when a listener with the same natural key already exists. "error" (default) raises DuplicateListenerError. "skip" returns a subscription to the existing listener when configs match. "replace" cancels the existing listener and registers the new one.	'error'

Returns:

Type	Description
Subscription	A subscription to the added listener (or the existing listener on skip).

Raises:

Type	Description
ListenerNameRequiredError	If the listener has no name (required for all DB-registered listeners, including once-listeners; cancel-listeners bypass this path entirely).
DuplicateListenerError	If the listener's natural key is already registered and if_exists="error".

Source code in src/hassette/bus/sync.py

def add_listener(self, listener: "Listener", *, if_exists: IfExistsPolicy = "error") -> Subscription:
    """Add a pre-built listener to the bus.

    This is the direct entry point for callers that construct a ``Listener``
    externally. The normal registration flow (``on_state_change``, ``on()``,
    etc.) goes through ``_on_internal`` instead.

    Args:
        listener: The pre-built listener to add.
        if_exists: Behavior when a listener with the same natural key already exists.
            ``"error"`` (default) raises ``DuplicateListenerError``.
            ``"skip"`` returns a subscription to the existing listener when configs match.
            ``"replace"`` cancels the existing listener and registers the new one.

    Returns:
        A subscription to the added listener (or the existing listener on skip).

    Raises:
        ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
            listeners, including once-listeners; cancel-listeners bypass this path entirely).
        DuplicateListenerError: If the listener's natural key is already registered and
            ``if_exists="error"``.
    """
    return self.task_bucket.run_sync(self._bus.add_listener(listener, if_exists=if_exists))

emit(topic: str, data: object) -> None

Broadcast data to all subscribers of the given topic.

Subscribers annotated with D.EventData[T] receive data pre-extracted. If the internal event stream is closed (during shutdown), the event is silently dropped.

Source code in src/hassette/bus/sync.py

def emit(self, topic: str, data: object) -> None:
    """Broadcast data to all subscribers of the given topic.

    Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
    If the internal event stream is closed (during shutdown), the event is silently dropped.
    """
    return self.task_bucket.run_sync(self._bus.emit(topic, data))

on(*, topic: str, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, once: bool = False, debounce: float | None = None, throttle: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, backpressure: BackpressurePolicy | str | None = None, name: str, on_error: BusErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error') -> Subscription

Subscribe to an event topic with optional filtering and modifiers.

This is the public registration method for raw topic subscriptions. Registration completes before the call returns — sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
topic	str	The event topic to listen to.	required
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Optional predicates to filter events. These can be custom callables or predefined predicates from hassette.event_handling.predicates. They will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
once	bool	If True, the handler will be called only once and then removed.	False
debounce	float | None	If set, applies a debounce to the handler.	None
throttle	float | None	If set, applies a throttle to the handler.	None
timeout	float | None	Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config. None means fall through to the config default.	None
timeout_disabled	bool	When True, disables timeout enforcement for this listener regardless of config.	False
mode	ExecutionMode | str | None	Overlap behavior when a trigger fires while a prior invocation still runs — "single", "restart", "queued", or "parallel". When omitted, the effective default is tier-aware: parallel for framework listeners, single for app listeners. Suppressed/dropped counts are live-only diagnostics, reset on restart.	None
backpressure	BackpressurePolicy | str | None	Saturation policy when the global dispatch concurrency semaphore is full. "block" (default) waits for a slot; "drop_newest" skips the event immediately and records one drop on the listener. When omitted, the effective default is block.	None
name	str	Required. Stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
if_exists	IfExistsPolicy	Behavior when a listener with the same natural key already exists. "error" (default) raises DuplicateListenerError. "skip" returns the existing listener's subscription when the configurations match, and raises ValueError if the configuration has drifted. "replace" cancels the existing listener and registers the new one in its place.	'error'

Returns:

Type	Description
Subscription	A subscription object. sub.cancel() removes the listener.
Subscription	sub.listener.db_id is a valid integer immediately on return.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered and if_exists="error" (the default).
ValueError	If if_exists="skip" and a listener with the same (name, topic) exists but with a different configuration (the message lists the changed fields).

Source code in src/hassette/bus/sync.py

def on(
    self,
    *,
    topic: str,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    once: bool = False,
    debounce: float | None = None,
    throttle: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    backpressure: "BackpressurePolicy | str | None" = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
) -> Subscription:
    """Subscribe to an event topic with optional filtering and modifiers.

    This is the public registration method for raw topic subscriptions.
    Registration completes before the call returns — ``sub.listener.db_id`` is a valid
    integer immediately on return.

    Args:
        topic: The event topic to listen to.
        handler: The function to call when the event matches.
        where: Optional predicates to filter events. These can be custom callables or predefined predicates from
            `hassette.event_handling.predicates`. They will receive the full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        once: If True, the handler will be called only once and then removed.
        debounce: If set, applies a debounce to the handler.
        throttle: If set, applies a throttle to the handler.
        timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
            None means fall through to the config default.
        timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
        mode: Overlap behavior when a trigger fires while a prior invocation still runs —
            ``"single"``, ``"restart"``, ``"queued"``, or ``"parallel"``. When omitted, the
            effective default is tier-aware: ``parallel`` for framework listeners, ``single``
            for app listeners. Suppressed/dropped counts are live-only diagnostics, reset on
            restart.
        backpressure: Saturation policy when the global dispatch concurrency semaphore is full.
            ``"block"`` (default) waits for a slot; ``"drop_newest"`` skips the event immediately
            and records one drop on the listener. When omitted, the effective default is ``block``.
        name: Required. Stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
            empty string raises ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        if_exists: Behavior when a listener with the same natural key already exists.
            ``"error"`` (default) raises ``DuplicateListenerError``. ``"skip"`` returns the
            existing listener's subscription when the configurations match, and raises
            ``ValueError`` if the configuration has drifted. ``"replace"`` cancels the
            existing listener and registers the new one in its place.

    Returns:
        A subscription object. ``sub.cancel()`` removes the listener.
        ``sub.listener.db_id`` is a valid integer immediately on return.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered and ``if_exists="error"`` (the default).
        ValueError: If ``if_exists="skip"`` and a listener with the same ``(name, topic)``
            exists but with a different configuration (the message lists the changed fields).
    """
    return self.task_bucket.run_sync(
        self._bus.on(
            topic=topic,
            handler=handler,
            where=where,
            kwargs=kwargs,
            once=once,
            debounce=debounce,
            throttle=throttle,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            backpressure=backpressure,
            name=name,
            on_error=on_error,
            if_exists=if_exists,
        )
    )

on_state_change(entity_id: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to state changes for a specific entity.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the state changed. If a ComparisonCondition is provided, it will be used to compare the old and new state values.	True
changed_from	ChangeType	A value or callable that will be used to filter state changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter state changes to this value.	NOT_PROVIDED
where	WhereClause	Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options. Accepts once, debounce, throttle, timeout, timeout_disabled, if_exists, mode, and backpressure. mode controls overlap behavior when a trigger fires while a prior invocation is still running: "single" drops the re-fire (the default for app handlers), "restart" cancels and replaces, "queued" serializes in arrival order (bounded at 10 pending), "parallel" runs concurrently. When omitted, the tier-aware default applies: "single" for app handlers, "parallel" for framework-internal listeners. An explicit mode= always wins. Suppressed (single) and dropped (queued cap) events log at DEBUG only. Suppressed/dropped counts are live-only diagnostics, reset on restart. See Execution Modes <https://hassette.dev/core-concepts/bus/execution-modes/>_. backpressure controls what this listener does when the dispatch concurrency semaphore is saturated: "block" (default) waits for a slot, unchanged from today; "drop_newest" skips the event immediately rather than waiting. It gates at the dispatch acquire point (global bus saturation), orthogonal to mode/debounce/throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately. sub.cancel()
Subscription	removes the listener from routing.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered on this bus in the current session and if_exists="error" (the default).

Source code in src/hassette/bus/sync.py

def on_state_change(
    self,
    entity_id: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to state changes for a specific entity.

    Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
            will be used to compare the old and new state values.
        changed_from: A value or callable that will be used to filter state changes *from* this value.
        changed_to: A value or callable that will be used to filter state changes *to* this value.
        where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
            full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it entirely raises ``TypeError`` (no default value); passing an
            empty string raises ``ListenerNameRequiredError`` at call time.
        **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
            ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

            ``mode`` controls overlap behavior when a trigger fires while a prior invocation
            is still running: ``"single"`` drops the re-fire (the default for app handlers),
            ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
            (bounded at 10 pending), ``"parallel"`` runs concurrently. When omitted, the
            tier-aware default applies: ``"single"`` for app handlers, ``"parallel"`` for
            framework-internal listeners. An explicit ``mode=`` always wins. Suppressed
            (``single``) and dropped (``queued`` cap) events log at DEBUG only.
            Suppressed/dropped counts are live-only diagnostics, reset on restart.
            See `Execution Modes <https://hassette.dev/core-concepts/bus/execution-modes/>`_.

            ``backpressure`` controls what this listener does when the dispatch concurrency
            semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
            today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
            at the dispatch acquire point (global bus saturation), orthogonal to
            ``mode``/``debounce``/``throttle``.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
        removes the listener from routing.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered on this bus in the current session and ``if_exists="error"``
            (the default).
    """
    return self.task_bucket.run_sync(
        self._bus.on_state_change(
            entity_id,
            handler=handler,
            changed=changed,
            changed_from=changed_from,
            changed_to=changed_to,
            where=where,
            kwargs=kwargs,
            immediate=immediate,
            duration=duration,
            name=name,
            on_error=on_error,
            **opts,
        )
    )

on_attribute_change(entity_id: str, attr: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to state change events for a specific entity's attribute.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
attr	str	The attribute name to filter changes on (e.g., "volume").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the attribute changed. If a ComparisonCondition is provided, it will be used to compare the old and new attribute values.	True
changed_from	ChangeType	A value or callable that will be used to filter attribute changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter attribute changes to this value.	NOT_PROVIDED
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable string identifier. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options. Accepts once, debounce, throttle, timeout, timeout_disabled, if_exists, mode, and backpressure. mode controls overlap behavior when a trigger fires while a prior invocation is still running: "single" drops the re-fire (the default for app handlers), "restart" cancels and replaces, "queued" serializes in arrival order (bounded at 10 pending), "parallel" runs concurrently. Suppressed/dropped counts are live-only diagnostics, reset on restart. backpressure controls what this listener does when the dispatch concurrency semaphore is saturated: "block" (default) waits for a slot, unchanged from today; "drop_newest" skips the event immediately rather than waiting. It gates at the dispatch acquire point (global bus saturation), orthogonal to mode/debounce/throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered and if_exists="error" (the default).

Source code in src/hassette/bus/sync.py

def on_attribute_change(
    self,
    entity_id: str,
    attr: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to state change events for a specific entity's attribute.

    Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        attr: The attribute name to filter changes on (e.g., "volume").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
            it will be used to compare the old and new attribute values.
        changed_from: A value or callable that will be used to filter attribute changes *from* this value.
        changed_to: A value or callable that will be used to filter attribute changes *to* this value.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier. Omitting it entirely raises ``TypeError``
            (no default value); passing an empty string raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
            ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

            ``mode`` controls overlap behavior when a trigger fires while a prior invocation
            is still running: ``"single"`` drops the re-fire (the default for app handlers),
            ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
            (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
            counts are live-only diagnostics, reset on restart.

            ``backpressure`` controls what this listener does when the dispatch concurrency
            semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
            today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
            at the dispatch acquire point (global bus saturation), orthogonal to
            ``mode``/``debounce``/``throttle``.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered and ``if_exists="error"`` (the default).
    """
    return self.task_bucket.run_sync(
        self._bus.on_attribute_change(
            entity_id,
            attr,
            handler=handler,
            changed=changed,
            changed_from=changed_from,
            changed_to=changed_to,
            where=where,
            kwargs=kwargs,
            immediate=immediate,
            duration=duration,
            name=name,
            on_error=on_error,
            **opts,
        )
    )

on_call_service(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to service call events.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service calls (e.g., "light").	None
service	str | None	The service to filter service calls (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable string identifier for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options. Accepts once, debounce, throttle, timeout, timeout_disabled, if_exists, mode, and backpressure. mode controls overlap behavior when a trigger fires while a prior invocation is still running: "single" drops the re-fire (the default for app handlers), "restart" cancels and replaces, "queued" serializes in arrival order (bounded at 10 pending), "parallel" runs concurrently. Suppressed/dropped counts are live-only diagnostics, reset on restart. backpressure controls what this listener does when the dispatch concurrency semaphore is saturated: "block" (default) waits for a slot, unchanged from today; "drop_newest" skips the event immediately rather than waiting. It gates at the dispatch acquire point (global bus saturation), orthogonal to mode/debounce/throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered and if_exists="error" (the default).

Source code in src/hassette/bus/sync.py

def on_call_service(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to service call events.

    Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        domain: The domain to filter service calls (e.g., "light").
        service: The service to filter service calls (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier for this listener. Omitting it entirely
            raises ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        **opts: Additional options. Accepts ``once``, ``debounce``, ``throttle``, ``timeout``,
            ``timeout_disabled``, ``if_exists``, ``mode``, and ``backpressure``.

            ``mode`` controls overlap behavior when a trigger fires while a prior invocation
            is still running: ``"single"`` drops the re-fire (the default for app handlers),
            ``"restart"`` cancels and replaces, ``"queued"`` serializes in arrival order
            (bounded at 10 pending), ``"parallel"`` runs concurrently. Suppressed/dropped
            counts are live-only diagnostics, reset on restart.

            ``backpressure`` controls what this listener does when the dispatch concurrency
            semaphore is saturated: ``"block"`` (default) waits for a slot, unchanged from
            today; ``"drop_newest"`` skips the event immediately rather than waiting. It gates
            at the dispatch acquire point (global bus saturation), orthogonal to
            ``mode``/``debounce``/``throttle``.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered and ``if_exists="error"`` (the default).
    """
    return self.task_bucket.run_sync(
        self._bus.on_call_service(
            domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_component_loaded(component: str | None = None, *, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to component loaded events.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
component	str | None	The component to filter load events (e.g., "light").	None
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Stable name for this listener. Required on all DB-registered listeners.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_component_loaded(
    self,
    component: str | None = None,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to component loaded events.

    Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        component: The component to filter load events (e.g., "light").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_component_loaded(
            component, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_service_registered(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to service registered events.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service registrations (e.g., "light").	None
service	str | None	The service to filter service registrations (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Stable name for this listener. Required on all DB-registered listeners.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_service_registered(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to service registered events.

    Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        domain: The domain to filter service registrations (e.g., "light").
        service: The service to filter service registrations (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_service_registered(
            domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_homeassistant_restart(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to Home Assistant restart events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_homeassistant_restart(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant restart events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_homeassistant_restart(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_homeassistant_start(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to Home Assistant start events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_homeassistant_start(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant start events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_homeassistant_start(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_homeassistant_stop(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to Home Assistant stop events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_homeassistant_stop(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant stop events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_homeassistant_stop(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_hassette_service_status(status: ResourceStatus | None = None, *, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service status events.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
status	ResourceStatus | None	The status to filter events (e.g., ResourceStatus.STARTED).	None
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_status(
    self,
    status: ResourceStatus | None = None,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service status events.

    Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        status: The status to filter events (e.g., ResourceStatus.STARTED).
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
            default value); passing an empty string raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_status(
            status, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_hassette_service_failed(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service failed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_failed(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service failed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it entirely raises ``TypeError`` (no
            default value); passing an empty string raises ``ListenerNameRequiredError``
            at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_failed(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_hassette_service_crashed(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service crashed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_crashed(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service crashed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_crashed(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_hassette_service_started(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service started events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_started(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service started events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_started(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_websocket_connected(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to websocket connected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_websocket_connected(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to websocket connected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_websocket_connected(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_websocket_disconnected(*, handler: HandlerType, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to websocket disconnected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_websocket_disconnected(
    self,
    *,
    handler: "HandlerType",
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to websocket disconnected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_websocket_disconnected(
            handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_app_state_changed(*, handler: HandlerType, app_key: str | None = None, status: ResourceStatus | None = None, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to app instance state change events.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
status	ResourceStatus | None	Filter events for a specific status.	None
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Stable name for this listener. Required on all DB-registered listeners.	required
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_app_state_changed(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    status: ResourceStatus | None = None,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instance state change events.

    Registration completes before the call returns.
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        status: Filter events for a specific status.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_app_state_changed(
            handler=handler,
            app_key=app_key,
            status=status,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )
    )

on_app_running(*, handler: HandlerType, app_key: str | None = None, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to app instances reaching RUNNING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_app_running(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instances reaching RUNNING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_app_running(
            handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_app_stopping(*, handler: HandlerType, app_key: str | None = None, where: WhereClause = None, kwargs: Mapping[str, Any] | None = None, name: str, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to app instances entering STOPPING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	WhereClause	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str	Required. Stable name for this listener. Omitting it entirely raises TypeError (no default value); passing an empty string raises ListenerNameRequiredError at call time.	required
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None
**opts	Unpack[Options]	Additional options like once, debounce, throttle, mode, and backpressure.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_app_stopping(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: WhereClause = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instances entering STOPPING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable name for this listener. Omitting it entirely raises
            ``TypeError`` (no default value); passing an empty string raises
            ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.
        **opts: Additional options like `once`, `debounce`, `throttle`, `mode`, and `backpressure`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return self.task_bucket.run_sync(
        self._bus.on_app_stopping(
            handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_error(handler: BusErrorHandlerType) -> None

Register an app-level error handler for this bus.

The handler is called when any listener on this bus raises an exception (including TimeoutError) and the listener does not have its own per-registration error handler.

This is an app-level fallback — it is resolved at dispatch time, not at listener registration time. A later call to on_error() replaces any previously registered handler.

Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near app shutdown may be cancelled before they complete. Do not rely on error handlers for delivery-critical alerting during system teardown.

Parameters:

Name	Type	Description	Default
handler	BusErrorHandlerType	A sync or async callable that accepts a :class:~hassette.bus.error_context.BusErrorContext.	required

Source code in src/hassette/bus/sync.py

def on_error(self, handler: "BusErrorHandlerType") -> None:
    """Register an app-level error handler for this bus.

    The handler is called when any listener on this bus raises an exception
    (including ``TimeoutError``) and the listener does not have its own
    per-registration error handler.

    This is an app-level fallback — it is resolved at dispatch time, not at listener
    registration time. A later call to ``on_error()`` replaces any previously registered
    handler.

    Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
    app shutdown may be cancelled before they complete. Do not rely on error handlers
    for delivery-critical alerting during system teardown.

    Args:
        handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`.
    """
    return self._bus.on_error(handler)

remove_listener(listener: Listener) -> None

Remove a listener from the bus and persist cancellation to the database.

Pops the natural key from the in-memory registry, removes the listener from routing (via BusService), and — when db_id is set — spawns mark_listener_cancelled on bus_service.task_bucket so the write survives resource shutdown, mirroring Scheduler.cancel_job.

BusService.remove_listener also fires _on_listener_removed, but that callback only spawns mark_listener_cancelled when the key is still present (once-fire path). Because this method pops the key first, the callback's spawn is skipped — avoiding a double write.

Source code in src/hassette/bus/sync.py

def remove_listener(self, listener: "Listener") -> None:
    """Remove a listener from the bus and persist cancellation to the database.

    Pops the natural key from the in-memory registry, removes the listener from routing
    (via BusService), and — when ``db_id`` is set — spawns ``mark_listener_cancelled``
    on ``bus_service.task_bucket`` so the write survives resource shutdown, mirroring
    ``Scheduler.cancel_job``.

    BusService.remove_listener also fires _on_listener_removed, but that callback only
    spawns mark_listener_cancelled when the key is still present (once-fire path). Because
    this method pops the key first, the callback's spawn is skipped — avoiding a double write.
    """
    return self._bus.remove_listener(listener)

remove_all_listeners() -> None

Remove all listeners owned by this bus's owner.

Source code in src/hassette/bus/sync.py

def remove_all_listeners(self) -> None:
    """Remove all listeners owned by this bus's owner."""
    return self._bus.remove_all_listeners()

get_listeners() -> list[Listener]

Get all listeners owned by this bus's owner.

Source code in src/hassette/bus/sync.py

def get_listeners(self) -> list["Listener"]:
    """Get all listeners owned by this bus's owner."""
    return self._bus.get_listeners()