Clean up and fix translation key

homeassistant/bootstrap.py

@@ -515,7 +515,7 @@ async def async_from_config_dict(
         issue_registry.async_create_issue(
             hass,
             core.DOMAIN,
-            f"python_version_{required_python_version}",
+            "python_version",
             is_fixable=False,
             severity=issue_registry.IssueSeverity.WARNING,
             breaks_in_ha_version=REQUIRED_NEXT_PYTHON_HA_RELEASE,

homeassistant/components/assist_pipeline/pipeline.py

@@ -31,6 +31,7 @@ from homeassistant.components.tts import (
 )
 from homeassistant.core import Context, HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import intent
 from homeassistant.helpers.collection import (
     CHANGE_UPDATED,
     CollectionError,
@@ -109,6 +110,7 @@ PIPELINE_FIELDS: VolDictType = {
     vol.Required("tts_voice"): vol.Any(str, None),
     vol.Required("wake_word_entity"): vol.Any(str, None),
     vol.Required("wake_word_id"): vol.Any(str, None),
+    vol.Optional("prefer_local_intents"): bool,
 }
 
 STORED_PIPELINE_RUNS = 10
@@ -322,6 +324,7 @@ async def async_update_pipeline(
     tts_voice: str | None | UndefinedType = UNDEFINED,
     wake_word_entity: str | None | UndefinedType = UNDEFINED,
     wake_word_id: str | None | UndefinedType = UNDEFINED,
+    prefer_local_intents: bool | UndefinedType = UNDEFINED,
 ) -> None:
     """Update a pipeline."""
     pipeline_data: PipelineData = hass.data[DOMAIN]
@@ -345,6 +348,7 @@ async def async_update_pipeline(
                 ("tts_voice", tts_voice),
                 ("wake_word_entity", wake_word_entity),
                 ("wake_word_id", wake_word_id),
+                ("prefer_local_intents", prefer_local_intents),
             )
             if val is not UNDEFINED
         }
@@ -398,6 +402,7 @@ class Pipeline:
     tts_voice: str | None
     wake_word_entity: str | None
     wake_word_id: str | None
+    prefer_local_intents: bool = False
 
     id: str = field(default_factory=ulid_util.ulid_now)
 
@@ -421,6 +426,7 @@ class Pipeline:
             tts_voice=data["tts_voice"],
             wake_word_entity=data["wake_word_entity"],
             wake_word_id=data["wake_word_id"],
+            prefer_local_intents=data.get("prefer_local_intents", False),
         )
 
     def to_json(self) -> dict[str, Any]:
@@ -438,6 +444,7 @@ class Pipeline:
             "tts_voice": self.tts_voice,
             "wake_word_entity": self.wake_word_entity,
             "wake_word_id": self.wake_word_id,
+            "prefer_local_intents": self.prefer_local_intents,
         }
 
 
@@ -1016,15 +1023,58 @@ class PipelineRun:
         )
 
         try:
-            conversation_result = await conversation.async_converse(
-                hass=self.hass,
+            user_input = conversation.ConversationInput(
                 text=intent_input,
+                context=self.context,
                 conversation_id=conversation_id,
                 device_id=device_id,
-                context=self.context,
-                language=self.pipeline.conversation_language,
+                language=self.pipeline.language,
                 agent_id=self.intent_agent,
             )
+
+            # Sentence triggers override conversation agent
+            if (
+                trigger_response_text
+                := await conversation.async_handle_sentence_triggers(
+                    self.hass, user_input
+                )
+            ):
+                # Sentence trigger matched
+                trigger_response = intent.IntentResponse(
+                    self.pipeline.conversation_language
+                )
+                trigger_response.async_set_speech(trigger_response_text)
+                conversation_result = conversation.ConversationResult(
+                    response=trigger_response,
+                    conversation_id=user_input.conversation_id,
+                )
+            # Try local intents first, if preferred.
+            # Skip this step if the default agent is already used.
+            elif (
+                self.pipeline.prefer_local_intents
+                and (user_input.agent_id != conversation.HOME_ASSISTANT_AGENT)
+                and (
+                    intent_response := await conversation.async_handle_intents(
+                        self.hass, user_input
+                    )
+                )
+            ):
+                # Local intent matched
+                conversation_result = conversation.ConversationResult(
+                    response=intent_response,
+                    conversation_id=user_input.conversation_id,
+                )
+            else:
+                # Fall back to pipeline conversation agent
+                conversation_result = await conversation.async_converse(
+                    hass=self.hass,
+                    text=user_input.text,
+                    conversation_id=user_input.conversation_id,
+                    device_id=user_input.device_id,
+                    context=user_input.context,
+                    language=user_input.language,
+                    agent_id=user_input.agent_id,
+                )
         except Exception as src_error:
             _LOGGER.exception("Unexpected error during intent recognition")
             raise IntentRecognitionError(

homeassistant/components/cloud/assist_pipeline.py

@@ -1,6 +1,7 @@
 """Handle Cloud assist pipelines."""
 
 import asyncio
+from typing import Any
 
 from homeassistant.components.assist_pipeline import (
     async_create_default_pipeline,
@@ -98,7 +99,7 @@ async def async_migrate_cloud_pipeline_engine(
     # is an after dependency of cloud
     await async_setup_pipeline_store(hass)
 
-    kwargs: dict[str, str] = {pipeline_attribute: engine_id}
+    kwargs: dict[str, Any] = {pipeline_attribute: engine_id}
     pipelines = async_get_pipelines(hass)
     for pipeline in pipelines:
         if getattr(pipeline, pipeline_attribute) == DOMAIN:

homeassistant/components/conversation/__init__.py

@@ -44,7 +44,7 @@ from .const import (
     SERVICE_RELOAD,
     ConversationEntityFeature,
 )
-from .default_agent import async_setup_default_agent
+from .default_agent import DefaultAgent, async_setup_default_agent
 from .entity import ConversationEntity
 from .http import async_setup as async_setup_conversation_http
 from .models import AbstractConversationAgent, ConversationInput, ConversationResult
@@ -207,6 +207,32 @@ async def async_prepare_agent(
     await agent.async_prepare(language)
 
 
+async def async_handle_sentence_triggers(
+    hass: HomeAssistant, user_input: ConversationInput
+) -> str | None:
+    """Try to match input against sentence triggers and return response text.
+
+    Returns None if no match occurred.
+    """
+    default_agent = async_get_agent(hass)
+    assert isinstance(default_agent, DefaultAgent)
+
+    return await default_agent.async_handle_sentence_triggers(user_input)
+
+
+async def async_handle_intents(
+    hass: HomeAssistant, user_input: ConversationInput
+) -> intent.IntentResponse | None:
+    """Try to match input against registered intents and return response.
+
+    Returns None if no match occurred.
+    """
+    default_agent = async_get_agent(hass)
+    assert isinstance(default_agent, DefaultAgent)
+
+    return await default_agent.async_handle_intents(user_input)
+
+
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Register the process service."""
     entity_component = EntityComponent[ConversationEntity](_LOGGER, DOMAIN, hass)

homeassistant/components/conversation/default_agent.py

@@ -214,10 +214,15 @@ class DefaultAgent(ConversationEntity):
         ]
 
     async def async_recognize(
-        self, user_input: ConversationInput
+        self,
+        user_input: ConversationInput,
+        strict_intents_only: bool = False,
+        match_sentence_triggers: bool = True,
     ) -> RecognizeResult | SentenceTriggerResult | None:
         """Recognize intent from user input."""
-        if trigger_result := await self._match_triggers(user_input.text):
+        if (match_sentence_triggers) and (
+            trigger_result := await self._match_triggers(user_input.text)
+        ):
             return trigger_result
 
         language = user_input.language or self.hass.config.language
@@ -240,6 +245,7 @@ class DefaultAgent(ConversationEntity):
             slot_lists,
             intent_context,
             language,
+            strict_intents_only,
         )
 
         _LOGGER.debug(
@@ -251,56 +257,33 @@ class DefaultAgent(ConversationEntity):
 
     async def async_process(self, user_input: ConversationInput) -> ConversationResult:
         """Process a sentence."""
-        language = user_input.language or self.hass.config.language
-        conversation_id = None  # Not supported
-
         result = await self.async_recognize(user_input)
 
         # Check if a trigger matched
         if isinstance(result, SentenceTriggerResult):
-            # Gather callback responses in parallel
-            trigger_callbacks = [
-                self._trigger_sentences[trigger_id].callback(
-                    result.sentence, trigger_result, user_input.device_id
-                )
-                for trigger_id, trigger_result in result.matched_triggers.items()
-            ]
-
-            # Use first non-empty result as response.
-            #
-            # There may be multiple copies of a trigger running when editing in
-            # the UI, so it's critical that we filter out empty responses here.
-            response_text: str | None = None
-            response_set_by_trigger = False
-            for trigger_future in asyncio.as_completed(trigger_callbacks):
-                trigger_response = await trigger_future
-                if trigger_response is None:
-                    continue
-
-                response_text = trigger_response
-                response_set_by_trigger = True
-                break
+            # Process callbacks and get response
+            response_text = await self._handle_trigger_result(result, user_input)
 
             # Convert to conversation result
-            response = intent.IntentResponse(language=language)
+            response = intent.IntentResponse(
+                language=user_input.language or self.hass.config.language
+            )
             response.response_type = intent.IntentResponseType.ACTION_DONE
-
-            if response_set_by_trigger:
-                # Response was explicitly set to empty
-                response_text = response_text or ""
-            elif not response_text:
-                # Use translated acknowledgment for pipeline language
-                translations = await translation.async_get_translations(
-                    self.hass, language, DOMAIN, [DOMAIN]
-                )
-                response_text = translations.get(
-                    f"component.{DOMAIN}.conversation.agent.done", "Done"
-                )
-
             response.async_set_speech(response_text)
 
             return ConversationResult(response=response)
 
+        return await self._async_process_intent_result(result, user_input)
+
+    async def _async_process_intent_result(
+        self,
+        result: RecognizeResult | None,
+        user_input: ConversationInput,
+    ) -> ConversationResult:
+        """Process user input with intents."""
+        language = user_input.language or self.hass.config.language
+        conversation_id = None  # Not supported
+
         # Intent match or failure
         lang_intents = await self.async_get_or_load_intents(language)
 
@@ -436,6 +419,7 @@ class DefaultAgent(ConversationEntity):
         slot_lists: dict[str, SlotList],
         intent_context: dict[str, Any] | None,
         language: str,
+        strict_intents_only: bool,
     ) -> RecognizeResult | None:
         """Search intents for a match to user input."""
         strict_result = self._recognize_strict(
@@ -446,6 +430,9 @@ class DefaultAgent(ConversationEntity):
             # Successful strict match
             return strict_result
 
+        if strict_intents_only:
+            return None
+
         # Try again with all entities (including unexposed)
         entity_registry = er.async_get(self.hass)
         all_entity_names: list[tuple[str, str, dict[str, Any]]] = []
@@ -1098,6 +1085,82 @@ class DefaultAgent(ConversationEntity):
 
         return SentenceTriggerResult(sentence, matched_template, matched_triggers)
 
+    async def _handle_trigger_result(
+        self, result: SentenceTriggerResult, user_input: ConversationInput
+    ) -> str:
+        """Run sentence trigger callbacks and return response text."""
+
+        # Gather callback responses in parallel
+        trigger_callbacks = [
+            self._trigger_sentences[trigger_id].callback(
+                user_input.text, trigger_result, user_input.device_id
+            )
+            for trigger_id, trigger_result in result.matched_triggers.items()
+        ]
+
+        # Use first non-empty result as response.
+        #
+        # There may be multiple copies of a trigger running when editing in
+        # the UI, so it's critical that we filter out empty responses here.
+        response_text = ""
+        response_set_by_trigger = False
+        for trigger_future in asyncio.as_completed(trigger_callbacks):
+            trigger_response = await trigger_future
+            if trigger_response is None:
+                continue
+
+            response_text = trigger_response
+            response_set_by_trigger = True
+            break
+
+        if response_set_by_trigger:
+            # Response was explicitly set to empty
+            response_text = response_text or ""
+        elif not response_text:
+            # Use translated acknowledgment for pipeline language
+            language = user_input.language or self.hass.config.language
+            translations = await translation.async_get_translations(
+                self.hass, language, DOMAIN, [DOMAIN]
+            )
+            response_text = translations.get(
+                f"component.{DOMAIN}.conversation.agent.done", "Done"
+            )
+
+        return response_text
+
+    async def async_handle_sentence_triggers(
+        self, user_input: ConversationInput
+    ) -> str | None:
+        """Try to input sentence against sentence triggers and return response text.
+
+        Returns None if no match occurred.
+        """
+        if trigger_result := await self._match_triggers(user_input.text):
+            return await self._handle_trigger_result(trigger_result, user_input)
+
+        return None
+
+    async def async_handle_intents(
+        self,
+        user_input: ConversationInput,
+    ) -> intent.IntentResponse | None:
+        """Try to match sentence against registered intents and return response.
+
+        Only performs strict matching with exposed entities and exact wording.
+        Returns None if no match occurred.
+        """
+        result = await self.async_recognize(
+            user_input, strict_intents_only=True, match_sentence_triggers=False
+        )
+        if not isinstance(result, RecognizeResult):
+            # No error message on failed match
+            return None
+
+        conversation_result = await self._async_process_intent_result(
+            result, user_input
+        )
+        return conversation_result.response
+
 
 def _make_error_result(
     language: str,
@@ -1108,7 +1171,6 @@ def _make_error_result(
     """Create conversation result with error code and text."""
     response = intent.IntentResponse(language=language)
     response.async_set_error(error_code, response_text)
-
     return ConversationResult(response, conversation_id)
 
 

homeassistant/components/smarty/strings.json

@@ -28,10 +28,6 @@
     "deprecated_yaml_import_issue_auth_error": {
       "title": "YAML import failed due to an authentication error",
       "description": "Configuring {integration_title} using YAML is being removed but there was an authentication error while importing your existing configuration.\nSetup will not proceed.\n\nVerify that your {integration_title} is operating correctly and restart Home Assistant to attempt the import again.\n\nAlternatively, you may remove the `{domain}` configuration from your configuration.yaml entirely, restart Home Assistant, and add the {integration_title} integration manually."
-    },
-    "deprecated_yaml_import_issue_cannot_connect": {
-      "title": "YAML import failed due to a connection error",
-      "description": "Configuring {integration_title} using YAML is being removed but there was a connect error while importing your existing configuration.\nSetup will not proceed.\n\nVerify that your {integration_title} is operating correctly and restart Home Assistant to attempt the import again.\n\nAlternatively, you may remove the `{domain}` configuration from your configuration.yaml entirely, restart Home Assistant, and add the {integration_title} integration manually."
     }
   },
   "entity": {

tests/components/acaia/snapshots/test_button.ambr

@@ -1,5 +1,5 @@
 # serializer version: 1
-# name: test_buttons[button.lunar_ddeeff_reset_timer-entry]
+# name: test_buttons[entry_button_reset_timer]
   EntityRegistryEntrySnapshot({
     'aliases': set({
     }),
@@ -32,20 +32,7 @@
     'unit_of_measurement': None,
   })
 # ---
-# name: test_buttons[button.lunar_ddeeff_reset_timer-state]
-  StateSnapshot({
-    'attributes': ReadOnlyDict({
-      'friendly_name': 'LUNAR-DDEEFF Reset timer',
-    }),
-    'context': <ANY>,
-    'entity_id': 'button.lunar_ddeeff_reset_timer',
-    'last_changed': <ANY>,
-    'last_reported': <ANY>,
-    'last_updated': <ANY>,
-    'state': 'unknown',
-  })
-# ---
-# name: test_buttons[button.lunar_ddeeff_start_stop_timer-entry]
+# name: test_buttons[entry_button_start_stop_timer]
   EntityRegistryEntrySnapshot({
     'aliases': set({
     }),
@@ -78,20 +65,7 @@
     'unit_of_measurement': None,
   })
 # ---
-# name: test_buttons[button.lunar_ddeeff_start_stop_timer-state]
-  StateSnapshot({
-    'attributes': ReadOnlyDict({
-      'friendly_name': 'LUNAR-DDEEFF Start/stop timer',
-    }),
-    'context': <ANY>,
-    'entity_id': 'button.lunar_ddeeff_start_stop_timer',
-    'last_changed': <ANY>,
-    'last_reported': <ANY>,
-    'last_updated': <ANY>,
-    'state': 'unknown',
-  })
-# ---
-# name: test_buttons[button.lunar_ddeeff_tare-entry]
+# name: test_buttons[entry_button_tare]
   EntityRegistryEntrySnapshot({
     'aliases': set({
     }),
@@ -124,7 +98,33 @@
     'unit_of_measurement': None,
   })
 # ---
-# name: test_buttons[button.lunar_ddeeff_tare-state]
+# name: test_buttons[state_button_reset_timer]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'LUNAR-DDEEFF Reset timer',
+    }),
+    'context': <ANY>,
+    'entity_id': 'button.lunar_ddeeff_reset_timer',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'unknown',
+  })
+# ---
+# name: test_buttons[state_button_start_stop_timer]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'friendly_name': 'LUNAR-DDEEFF Start/stop timer',
+    }),
+    'context': <ANY>,
+    'entity_id': 'button.lunar_ddeeff_start_stop_timer',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'unknown',
+  })
+# ---
+# name: test_buttons[state_button_tare]
   StateSnapshot({
     'attributes': ReadOnlyDict({
       'friendly_name': 'LUNAR-DDEEFF Tare',

tests/components/acaia/test_button.py

@@ -1,24 +1,21 @@
 """Tests for the acaia buttons."""
 
 from datetime import timedelta
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 from freezegun.api import FrozenDateTimeFactory
+import pytest
 from syrupy import SnapshotAssertion
 
 from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
-from homeassistant.const import (
-    ATTR_ENTITY_ID,
-    STATE_UNAVAILABLE,
-    STATE_UNKNOWN,
-    Platform,
-)
+from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE, STATE_UNKNOWN
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 
-from . import setup_integration
+from tests.common import async_fire_time_changed
+
+pytestmark = pytest.mark.usefixtures("init_integration")
 
-from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
 
 BUTTONS = (
     "tare",
@@ -31,25 +28,24 @@ async def test_buttons(
     hass: HomeAssistant,
     entity_registry: er.EntityRegistry,
     snapshot: SnapshotAssertion,
-    mock_scale: MagicMock,
-    mock_config_entry: MockConfigEntry,
 ) -> None:
     """Test the acaia buttons."""
+    for button in BUTTONS:
+        state = hass.states.get(f"button.lunar_ddeeff_{button}")
+        assert state
+        assert state == snapshot(name=f"state_button_{button}")
 
-    with patch("homeassistant.components.acaia.PLATFORMS", [Platform.BUTTON]):
-        await setup_integration(hass, mock_config_entry)
-    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+        entry = entity_registry.async_get(state.entity_id)
+        assert entry
+        assert entry == snapshot(name=f"entry_button_{button}")
 
 
 async def test_button_presses(
     hass: HomeAssistant,
     mock_scale: MagicMock,
-    mock_config_entry: MockConfigEntry,
 ) -> None:
     """Test the acaia button presses."""
 
-    await setup_integration(hass, mock_config_entry)
-
     for button in BUTTONS:
         await hass.services.async_call(
             BUTTON_DOMAIN,
@@ -67,13 +63,10 @@ async def test_button_presses(
 async def test_buttons_unavailable_on_disconnected_scale(
     hass: HomeAssistant,
     mock_scale: MagicMock,
-    mock_config_entry: MockConfigEntry,
     freezer: FrozenDateTimeFactory,
 ) -> None:
     """Test the acaia buttons are unavailable when the scale is disconnected."""
 
-    await setup_integration(hass, mock_config_entry)
-
     for button in BUTTONS:
         state = hass.states.get(f"button.lunar_ddeeff_{button}")
         assert state

tests/components/assist_pipeline/snapshots/test_init.ambr

@@ -139,7 +139,7 @@
             'data': dict({
               'code': 'no_intent_match',
             }),
-            'language': 'en-US',
+            'language': 'en',
             'response_type': 'error',
             'speech': dict({
               'plain': dict({
@@ -228,7 +228,7 @@
             'data': dict({
               'code': 'no_intent_match',
             }),
-            'language': 'en-US',
+            'language': 'en',
             'response_type': 'error',
             'speech': dict({
               'plain': dict({

tests/components/assist_pipeline/test_init.py

@@ -11,13 +11,20 @@ import wave
 import pytest
 from syrupy.assertion import SnapshotAssertion
 
-from homeassistant.components import assist_pipeline, media_source, stt, tts
+from homeassistant.components import (
+    assist_pipeline,
+    conversation,
+    media_source,
+    stt,
+    tts,
+)
 from homeassistant.components.assist_pipeline.const import (
     BYTES_PER_CHUNK,
     CONF_DEBUG_RECORDING_DIR,
     DOMAIN,
 )
 from homeassistant.core import Context, HomeAssistant
+from homeassistant.helpers import intent
 from homeassistant.setup import async_setup_component
 
 from .conftest import (
@@ -927,3 +934,148 @@ async def test_tts_dict_preferred_format(
         assert int(options.get(tts.ATTR_PREFERRED_SAMPLE_RATE)) == 48000
         assert int(options.get(tts.ATTR_PREFERRED_SAMPLE_CHANNELS)) == 2
         assert int(options.get(tts.ATTR_PREFERRED_SAMPLE_BYTES)) == 2
+
+
+async def test_sentence_trigger_overrides_conversation_agent(
+    hass: HomeAssistant,
+    init_components,
+    pipeline_data: assist_pipeline.pipeline.PipelineData,
+) -> None:
+    """Test that sentence triggers are checked before the conversation agent."""
+    assert await async_setup_component(
+        hass,
+        "automation",
+        {
+            "automation": {
+                "trigger": {
+                    "platform": "conversation",
+                    "command": [
+                        "test trigger sentence",
+                    ],
+                },
+                "action": {
+                    "set_conversation_response": "test trigger response",
+                },
+            }
+        },
+    )
+
+    events: list[assist_pipeline.PipelineEvent] = []
+
+    pipeline_store = pipeline_data.pipeline_store
+    pipeline_id = pipeline_store.async_get_preferred_item()
+    pipeline = assist_pipeline.pipeline.async_get_pipeline(hass, pipeline_id)
+
+    pipeline_input = assist_pipeline.pipeline.PipelineInput(
+        intent_input="test trigger sentence",
+        run=assist_pipeline.pipeline.PipelineRun(
+            hass,
+            context=Context(),
+            pipeline=pipeline,
+            start_stage=assist_pipeline.PipelineStage.INTENT,
+            end_stage=assist_pipeline.PipelineStage.INTENT,
+            event_callback=events.append,
+        ),
+    )
+    await pipeline_input.validate()
+
+    with patch(
+        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse"
+    ) as mock_async_converse:
+        await pipeline_input.execute()
+
+        # Sentence trigger should have been handled
+        mock_async_converse.assert_not_called()
+
+        # Verify sentence trigger response
+        intent_end_event = next(
+            (
+                e
+                for e in events
+                if e.type == assist_pipeline.PipelineEventType.INTENT_END
+            ),
+            None,
+        )
+        assert (intent_end_event is not None) and intent_end_event.data
+        assert (
+            intent_end_event.data["intent_output"]["response"]["speech"]["plain"][
+                "speech"
+            ]
+            == "test trigger response"
+        )
+
+
+async def test_prefer_local_intents(
+    hass: HomeAssistant,
+    init_components,
+    pipeline_data: assist_pipeline.pipeline.PipelineData,
+) -> None:
+    """Test that the default agent is checked first when local intents are preferred."""
+    events: list[assist_pipeline.PipelineEvent] = []
+
+    # Reuse custom sentences in test config
+    class OrderBeerIntentHandler(intent.IntentHandler):
+        intent_type = "OrderBeer"
+
+        async def async_handle(
+            self, intent_obj: intent.Intent
+        ) -> intent.IntentResponse:
+            response = intent_obj.create_response()
+            response.async_set_speech("Order confirmed")
+            return response
+
+    handler = OrderBeerIntentHandler()
+    intent.async_register(hass, handler)
+
+    # Fake a test agent and prefer local intents
+    pipeline_store = pipeline_data.pipeline_store
+    pipeline_id = pipeline_store.async_get_preferred_item()
+    pipeline = assist_pipeline.pipeline.async_get_pipeline(hass, pipeline_id)
+    await assist_pipeline.pipeline.async_update_pipeline(
+        hass, pipeline, conversation_engine="test-agent", prefer_local_intents=True
+    )
+    pipeline = assist_pipeline.pipeline.async_get_pipeline(hass, pipeline_id)
+
+    pipeline_input = assist_pipeline.pipeline.PipelineInput(
+        intent_input="I'd like to order a stout please",
+        run=assist_pipeline.pipeline.PipelineRun(
+            hass,
+            context=Context(),
+            pipeline=pipeline,
+            start_stage=assist_pipeline.PipelineStage.INTENT,
+            end_stage=assist_pipeline.PipelineStage.INTENT,
+            event_callback=events.append,
+        ),
+    )
+
+    # Ensure prepare succeeds
+    with patch(
+        "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info",
+        return_value=conversation.AgentInfo(id="test-agent", name="Test Agent"),
+    ):
+        await pipeline_input.validate()
+
+    with patch(
+        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse"
+    ) as mock_async_converse:
+        await pipeline_input.execute()
+
+        # Test agent should not have been called
+        mock_async_converse.assert_not_called()
+
+        # Verify local intent response
+        intent_end_event = next(
+            (
+                e
+                for e in events
+                if e.type == assist_pipeline.PipelineEventType.INTENT_END
+            ),
+            None,
+        )
+        assert (intent_end_event is not None) and intent_end_event.data
+        assert (
+            intent_end_event.data["intent_output"]["response"]["speech"]["plain"][
+                "speech"
+            ]
+            == "Order confirmed"
+        )

tests/components/assist_pipeline/test_pipeline.py

@@ -574,6 +574,7 @@ async def test_update_pipeline(
         "tts_voice": "test_voice",
         "wake_word_entity": "wake_work.test_1",
         "wake_word_id": "wake_word_id_1",
+        "prefer_local_intents": False,
     }
 
     await async_update_pipeline(
@@ -617,6 +618,7 @@ async def test_update_pipeline(
         "tts_voice": "test_voice",
         "wake_word_entity": "wake_work.test_1",
         "wake_word_id": "wake_word_id_1",
+        "prefer_local_intents": False,
     }
 
 

tests/components/assist_pipeline/test_websocket.py

@@ -974,6 +974,7 @@ async def test_add_pipeline(
             "tts_voice": "Arnold Schwarzenegger",
             "wake_word_entity": "wakeword_entity_1",
             "wake_word_id": "wakeword_id_1",
+            "prefer_local_intents": True,
         }
     )
     msg = await client.receive_json()
@@ -991,6 +992,7 @@ async def test_add_pipeline(
         "tts_voice": "Arnold Schwarzenegger",
         "wake_word_entity": "wakeword_entity_1",
         "wake_word_id": "wakeword_id_1",
+        "prefer_local_intents": True,
     }
 
     assert len(pipeline_store.data) == 2
@@ -1008,6 +1010,7 @@ async def test_add_pipeline(
         tts_voice="Arnold Schwarzenegger",
         wake_word_entity="wakeword_entity_1",
         wake_word_id="wakeword_id_1",
+        prefer_local_intents=True,
     )
 
     await client.send_json_auto_id(
@@ -1195,6 +1198,7 @@ async def test_get_pipeline(
         "tts_voice": "james_earl_jones",
         "wake_word_entity": None,
         "wake_word_id": None,
+        "prefer_local_intents": False,
     }
 
     # Get conversation agent as pipeline
@@ -1220,6 +1224,7 @@ async def test_get_pipeline(
         "tts_voice": "james_earl_jones",
         "wake_word_entity": None,
         "wake_word_id": None,
+        "prefer_local_intents": False,
     }
 
     await client.send_json_auto_id(
@@ -1249,6 +1254,7 @@ async def test_get_pipeline(
             "tts_voice": "Arnold Schwarzenegger",
             "wake_word_entity": "wakeword_entity_1",
             "wake_word_id": "wakeword_id_1",
+            "prefer_local_intents": False,
         }
     )
     msg = await client.receive_json()
@@ -1277,6 +1283,7 @@ async def test_get_pipeline(
         "tts_voice": "Arnold Schwarzenegger",
         "wake_word_entity": "wakeword_entity_1",
         "wake_word_id": "wakeword_id_1",
+        "prefer_local_intents": False,
     }
 
 
@@ -1304,6 +1311,7 @@ async def test_list_pipelines(
                 "tts_voice": "james_earl_jones",
                 "wake_word_entity": None,
                 "wake_word_id": None,
+                "prefer_local_intents": False,
             }
         ],
         "preferred_pipeline": ANY,
@@ -1395,6 +1403,7 @@ async def test_update_pipeline(
         "tts_voice": "new_tts_voice",
         "wake_word_entity": "new_wakeword_entity",
         "wake_word_id": "new_wakeword_id",
+        "prefer_local_intents": False,
     }
 
     assert len(pipeline_store.data) == 2
@@ -1446,6 +1455,7 @@ async def test_update_pipeline(
         "tts_voice": None,
         "wake_word_entity": None,
         "wake_word_id": None,
+        "prefer_local_intents": False,
     }
 
     pipeline = pipeline_store.data[pipeline_id]

tests/components/cloud/__init__.py

@@ -35,6 +35,7 @@ PIPELINE_DATA = {
             "tts_voice": "Arnold Schwarzenegger",
             "wake_word_entity": None,
             "wake_word_id": None,
+            "prefer_local_intents": False,
         },
         {
             "conversation_engine": "conversation_engine_2",
@@ -49,6 +50,7 @@ PIPELINE_DATA = {
             "tts_voice": "The Voice",
             "wake_word_entity": None,
             "wake_word_id": None,
+            "prefer_local_intents": False,
         },
         {
             "conversation_engine": "conversation_engine_3",
@@ -63,6 +65,7 @@ PIPELINE_DATA = {
             "tts_voice": None,
             "wake_word_entity": None,
             "wake_word_id": None,
+            "prefer_local_intents": False,
         },
     ],
     "preferred_item": "01GX8ZWBAQYWNB1XV3EXEZ75DY",

tests/components/conftest.py

@@ -26,12 +26,7 @@ from homeassistant.config_entries import (
 )
 from homeassistant.const import STATE_OFF, STATE_ON
 from homeassistant.core import HomeAssistant
-from homeassistant.data_entry_flow import (
-    FlowContext,
-    FlowHandler,
-    FlowManager,
-    FlowResultType,
-)
+from homeassistant.data_entry_flow import FlowHandler, FlowManager, FlowResultType
 from homeassistant.helpers.translation import async_get_translations
 
 if TYPE_CHECKING:
@@ -562,12 +557,12 @@ def _validate_translation_placeholders(
             description_placeholders is None
             or placeholder not in description_placeholders
         ):
-            ignore_translations[full_key] = (
+            pytest.fail(
                 f"Description not found for placeholder `{placeholder}` in {full_key}"
             )
 
 
-async def _validate_translation(
+async def _ensure_translation_exists(
     hass: HomeAssistant,
     ignore_translations: dict[str, StoreInfo],
     category: str,
@@ -593,7 +588,7 @@ async def _validate_translation(
         ignore_translations[full_key] = "used"
         return
 
-    ignore_translations[full_key] = (
+    pytest.fail(
         f"Translation not found for {component}: `{category}.{key}`. "
         f"Please add to homeassistant/components/{component}/strings.json"
     )
@@ -609,106 +604,84 @@ def ignore_translations() -> str | list[str]:
     return []
 
 
-async def _check_config_flow_result_translations(
-    manager: FlowManager,
-    flow: FlowHandler,
-    result: FlowResult[FlowContext, str],
-    ignore_translations: dict[str, str],
-) -> None:
-    if isinstance(manager, ConfigEntriesFlowManager):
-        category = "config"
-        integration = flow.handler
-    elif isinstance(manager, OptionsFlowManager):
-        category = "options"
-        integration = flow.hass.config_entries.async_get_entry(flow.handler).domain
-    else:
-        return
-
-    # Check if this flow has been seen before
-    # Gets set to False on first run, and to True on subsequent runs
-    setattr(flow, "__flow_seen_before", hasattr(flow, "__flow_seen_before"))
-
-    if result["type"] is FlowResultType.FORM:
-        if step_id := result.get("step_id"):
-            # neither title nor description are required
-            # - title defaults to integration name
-            # - description is optional
-            for header in ("title", "description"):
-                await _validate_translation(
-                    flow.hass,
-                    ignore_translations,
-                    category,
-                    integration,
-                    f"step.{step_id}.{header}",
-                    result["description_placeholders"],
-                    translation_required=False,
-                )
-        if errors := result.get("errors"):
-            for error in errors.values():
-                await _validate_translation(
-                    flow.hass,
-                    ignore_translations,
-                    category,
-                    integration,
-                    f"error.{error}",
-                    result["description_placeholders"],
-                )
-        return
-
-    if result["type"] is FlowResultType.ABORT:
-        # We don't need translations for a discovery flow which immediately
-        # aborts, since such flows won't be seen by users
-        if not flow.__flow_seen_before and flow.source in DISCOVERY_SOURCES:
-            return
-        await _validate_translation(
-            flow.hass,
-            ignore_translations,
-            category,
-            integration,
-            f"abort.{result["reason"]}",
-            result["description_placeholders"],
-        )
-
-
 @pytest.fixture(autouse=True)
-def check_translations(ignore_translations: str | list[str]) -> Generator[None]:
-    """Check that translation requirements are met.
-
-    Current checks:
-    - data entry flow results (ConfigFlow/OptionsFlow)
-    """
+def check_config_translations(ignore_translations: str | list[str]) -> Generator[None]:
+    """Ensure config_flow translations are available."""
     if not isinstance(ignore_translations, list):
         ignore_translations = [ignore_translations]
 
     _ignore_translations = {k: "unused" for k in ignore_translations}
+    _original = FlowManager._async_handle_step
 
-    # Keep reference to original functions
-    _original_flow_manager_async_handle_step = FlowManager._async_handle_step
-
-    # Prepare override functions
-    async def _flow_manager_async_handle_step(
+    async def _async_handle_step(
         self: FlowManager, flow: FlowHandler, *args
     ) -> FlowResult:
-        result = await _original_flow_manager_async_handle_step(self, flow, *args)
-        await _check_config_flow_result_translations(
-            self, flow, result, _ignore_translations
-        )
+        result = await _original(self, flow, *args)
+        if isinstance(self, ConfigEntriesFlowManager):
+            category = "config"
+            component = flow.handler
+        elif isinstance(self, OptionsFlowManager):
+            category = "options"
+            component = flow.hass.config_entries.async_get_entry(flow.handler).domain
+        else:
+            return result
+
+        # Check if this flow has been seen before
+        # Gets set to False on first run, and to True on subsequent runs
+        setattr(flow, "__flow_seen_before", hasattr(flow, "__flow_seen_before"))
+
+        if result["type"] is FlowResultType.FORM:
+            if step_id := result.get("step_id"):
+                # neither title nor description are required
+                # - title defaults to integration name
+                # - description is optional
+                for header in ("title", "description"):
+                    await _ensure_translation_exists(
+                        flow.hass,
+                        _ignore_translations,
+                        category,
+                        component,
+                        f"step.{step_id}.{header}",
+                        result["description_placeholders"],
+                        translation_required=False,
+                    )
+            if errors := result.get("errors"):
+                for error in errors.values():
+                    await _ensure_translation_exists(
+                        flow.hass,
+                        _ignore_translations,
+                        category,
+                        component,
+                        f"error.{error}",
+                        result["description_placeholders"],
+                    )
+            return result
+
+        if result["type"] is FlowResultType.ABORT:
+            # We don't need translations for a discovery flow which immediately
+            # aborts, since such flows won't be seen by users
+            if not flow.__flow_seen_before and flow.source in DISCOVERY_SOURCES:
+                return result
+            await _ensure_translation_exists(
+                flow.hass,
+                _ignore_translations,
+                category,
+                component,
+                f"abort.{result["reason"]}",
+                result["description_placeholders"],
+            )
+
         return result
 
-    # Use override functions
     with patch(
         "homeassistant.data_entry_flow.FlowManager._async_handle_step",
-        _flow_manager_async_handle_step,
+        _async_handle_step,
     ):
         yield
 
-    # Run final checks
     unused_ignore = [k for k, v in _ignore_translations.items() if v == "unused"]
     if unused_ignore:
         pytest.fail(
             f"Unused ignore translations: {', '.join(unused_ignore)}. "
             "Please remove them from the ignore_translations fixture."
         )
-    for description in _ignore_translations.values():
-        if description not in {"used", "unused"}:
-            pytest.fail(description)

tests/components/conversation/test_init.py

@@ -8,10 +8,15 @@ from syrupy.assertion import SnapshotAssertion
 import voluptuous as vol
 
 from homeassistant.components import conversation
-from homeassistant.components.conversation import default_agent
+from homeassistant.components.conversation import (
+    ConversationInput,
+    async_handle_intents,
+    async_handle_sentence_triggers,
+    default_agent,
+)
 from homeassistant.components.conversation.const import DATA_DEFAULT_ENTITY
 from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
-from homeassistant.core import HomeAssistant
+from homeassistant.core import Context, HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import intent
 from homeassistant.setup import async_setup_component
@@ -229,3 +234,93 @@ async def test_prepare_agent(
         await conversation.async_prepare_agent(hass, agent_id, "en")
 
     assert len(mock_prepare.mock_calls) == 1
+
+
+async def test_async_handle_sentence_triggers(hass: HomeAssistant) -> None:
+    """Test handling sentence triggers with async_handle_sentence_triggers."""
+    assert await async_setup_component(hass, "homeassistant", {})
+    assert await async_setup_component(hass, "conversation", {})
+
+    response_template = "response {{ trigger.device_id }}"
+    assert await async_setup_component(
+        hass,
+        "automation",
+        {
+            "automation": {
+                "trigger": {
+                    "platform": "conversation",
+                    "command": ["my trigger"],
+                },
+                "action": {
+                    "set_conversation_response": response_template,
+                },
+            }
+        },
+    )
+
+    # Device id will be available in response template
+    device_id = "1234"
+    expected_response = f"response {device_id}"
+    actual_response = await async_handle_sentence_triggers(
+        hass,
+        ConversationInput(
+            text="my trigger",
+            context=Context(),
+            conversation_id=None,
+            device_id=device_id,
+            language=hass.config.language,
+        ),
+    )
+    assert actual_response == expected_response
+
+
+async def test_async_handle_intents(hass: HomeAssistant) -> None:
+    """Test handling registered intents with async_handle_intents."""
+    assert await async_setup_component(hass, "homeassistant", {})
+    assert await async_setup_component(hass, "conversation", {})
+
+    # Reuse custom sentences in test config to trigger default agent.
+    class OrderBeerIntentHandler(intent.IntentHandler):
+        intent_type = "OrderBeer"
+
+        def __init__(self) -> None:
+            super().__init__()
+            self.was_handled = False
+
+        async def async_handle(
+            self, intent_obj: intent.Intent
+        ) -> intent.IntentResponse:
+            self.was_handled = True
+            return intent_obj.create_response()
+
+    handler = OrderBeerIntentHandler()
+    intent.async_register(hass, handler)
+
+    # Registered intent will be handled
+    result = await async_handle_intents(
+        hass,
+        ConversationInput(
+            text="I'd like to order a stout",
+            context=Context(),
+            conversation_id=None,
+            device_id=None,
+            language=hass.config.language,
+        ),
+    )
+    assert result is not None
+    assert result.intent is not None
+    assert result.intent.intent_type == handler.intent_type
+    assert handler.was_handled
+
+    # No error messages, just None as a result
+    result = await async_handle_intents(
+        hass,
+        ConversationInput(
+            text="this sentence does not exist",
+            context=Context(),
+            conversation_id=None,
+            device_id=None,
+            language=hass.config.language,
+        ),
+    )
+    assert result is None