Separate Assist pipeline controller and processor (#182219)

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

homeassistant/components/assist_pipeline/run.py

@@ -0,0 +1,454 @@
+"""Run controller for Assist pipelines."""
+
+import asyncio
+from collections.abc import AsyncIterable
+from dataclasses import dataclass, field
+import logging
+from pathlib import Path
+from queue import Empty, Queue
+from threading import Thread
+import time
+from typing import Any, Protocol, override
+import wave
+
+from homeassistant.components import stt
+from homeassistant.core import Context, HomeAssistant, callback
+from homeassistant.helpers import chat_session
+from homeassistant.util import ulid as ulid_util
+from homeassistant.util.limited_size_dict import LimitedSizeDict
+
+from .const import (
+    CONF_DEBUG_RECORDING_DIR,
+    DATA_CONFIG,
+    DATA_LAST_WAKE_UP,
+    SAMPLE_CHANNELS,
+    SAMPLE_RATE,
+    SAMPLE_WIDTH,
+    WAKE_WORD_COOLDOWN,
+)
+from .default_pipeline import _DefaultPipelineProcessor
+from .error import (
+    DuplicateWakeUpDetectedError,
+    InvalidPipelineStagesError,
+    PipelineError,
+)
+from .models import (
+    PIPELINE_STAGE_ORDER,
+    AudioSettings,
+    Pipeline,
+    PipelineEvent,
+    PipelineEventCallback,
+    PipelineEventType,
+    PipelineStage,
+    WakeWordSettings,
+)
+from .runtime import KEY_ASSIST_PIPELINE, PipelineRunDebug
+
+_LOGGER = logging.getLogger(__name__)
+
+STORED_PIPELINE_RUNS = 10
+
+
+class _PipelineResponseAudio(Protocol):
+    """Response audio metadata exposed by a pipeline processor."""
+
+    @property
+    def token(self) -> str:
+        """Return the response audio token."""
+
+    @property
+    def url(self) -> str:
+        """Return the response audio URL."""
+
+    @property
+    def content_type(self) -> str:
+        """Return the response audio content type."""
+
+
+class _PipelineProcessor(Protocol):
+    """Implementation boundary for processing a pipeline run."""
+
+    @property
+    def response_audio(self) -> _PipelineResponseAudio | None:
+        """Return the response audio stream."""
+
+    @property
+    def supports_streaming_response(self) -> bool | None:
+        """Return whether response audio can be streamed."""
+
+    async def async_validate(self, request: _PipelineProcessorRequest) -> None:
+        """Validate pipeline input and prepare processing resources."""
+
+    async def async_execute(self, request: _PipelineProcessorRequest) -> None:
+        """Process pipeline input."""
+
+    def invalidate(self) -> None:
+        """Invalidate active processing."""
+
+    def cleanup(self) -> None:
+        """Clean up resources after a pipeline error."""
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class _PipelineProcessorRequest:
+    """Input passed across the private pipeline processor boundary."""
+
+    session: chat_session.ChatSession
+    stt_metadata: stt.SpeechMetadata | None = None
+    stt_stream: AsyncIterable[bytes] | None = None
+    wake_word_phrase: str | None = None
+    intent_input: str | None = None
+    tts_input: str | None = None
+    conversation_extra_system_prompt: str | None = None
+    device_id: str | None = None
+    satellite_id: str | None = None
+
+
+@dataclass
+class PipelineRun:
+    """Home Assistant controller for a pipeline run."""
+
+    hass: HomeAssistant
+    context: Context
+    pipeline: Pipeline
+    start_stage: PipelineStage
+    end_stage: PipelineStage
+    event_callback: PipelineEventCallback
+    language: str = None  # type: ignore[assignment]
+    runner_data: Any | None = None
+    tts_audio_output: str | dict[str, Any] | None = None
+    wake_word_settings: WakeWordSettings | None = None
+    audio_settings: AudioSettings = field(default_factory=AudioSettings)
+
+    id: str = field(default_factory=ulid_util.ulid_now)
+    debug_recording_thread: Thread | None = None
+    """Thread that records audio to debug_recording_dir."""
+    debug_recording_queue: Queue[str | bytes | None] | None = None
+    """Queue to communicate with the debug recording thread."""
+    _device_id: str | None = None
+    _satellite_id: str | None = None
+    _processor: _PipelineProcessor = field(init=False, repr=False)
+    _registered: bool = field(init=False, default=False, repr=False)
+    _started: bool = field(init=False, default=False, repr=False)
+    _ended: bool = field(init=False, default=False, repr=False)
+
+    def __post_init__(self) -> None:
+        """Initialize the pipeline controller."""
+        self.language = self.pipeline.language or self.hass.config.language
+
+        if PIPELINE_STAGE_ORDER.index(self.end_stage) < PIPELINE_STAGE_ORDER.index(
+            self.start_stage
+        ):
+            raise InvalidPipelineStagesError(self.start_stage, self.end_stage)
+
+        self._processor = _create_pipeline_processor(self)
+        pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE]
+        if self.pipeline.id not in pipeline_data.pipeline_debug:
+            pipeline_data.pipeline_debug[self.pipeline.id] = LimitedSizeDict(
+                size_limit=STORED_PIPELINE_RUNS
+            )
+        pipeline_data.pipeline_debug[self.pipeline.id][self.id] = PipelineRunDebug()
+        pipeline_data.pipeline_runs.add_run(self)
+        self._registered = True
+
+    @override
+    def __eq__(self, other: object) -> bool:
+        """Compare pipeline runs by id."""
+        if isinstance(other, PipelineRun):
+            return self.id == other.id
+        return False
+
+    @property
+    def device_id(self) -> str | None:
+        """Return the device associated with the run."""
+        return self._device_id
+
+    @property
+    def satellite_id(self) -> str | None:
+        """Return the satellite associated with the run."""
+        return self._satellite_id
+
+    @callback
+    def process_event(self, event: PipelineEvent) -> None:
+        """Log an event and call the listener."""
+        self.event_callback(event)
+        pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE]
+        if self.id not in pipeline_data.pipeline_debug[self.pipeline.id]:
+            return
+        pipeline_data.pipeline_debug[self.pipeline.id][self.id].events.append(event)
+
+    def start(
+        self, conversation_id: str, device_id: str | None, satellite_id: str | None
+    ) -> None:
+        """Emit the run-start event."""
+        if self._started:
+            raise RuntimeError("Pipeline run has already started")
+
+        self._started = True
+        self._device_id = device_id
+        self._satellite_id = satellite_id
+        if self.start_stage in (PipelineStage.WAKE_WORD, PipelineStage.STT):
+            self._start_debug_recording_thread()
+
+        data: dict[str, Any] = {
+            "pipeline": self.pipeline.id,
+            "language": self.language,
+            "conversation_id": conversation_id,
+        }
+        if satellite_id is not None:
+            data["satellite_id"] = satellite_id
+        if self.runner_data is not None:
+            data["runner_data"] = self.runner_data
+        if (response_audio := self._processor.response_audio) is not None:
+            data["tts_output"] = {
+                "token": response_audio.token,
+                "url": response_audio.url,
+                "mime_type": response_audio.content_type,
+                "stream_response": self._processor.supports_streaming_response,
+            }
+        self.process_event(PipelineEvent(PipelineEventType.RUN_START, data))
+
+    async def end(self) -> None:
+        """Emit the run-end event."""
+        if self._ended:
+            return
+
+        if not self._started:
+            self._ended = True
+            self._unregister()
+            return
+
+        self._ended = True
+        try:
+            self.capture_audio(None)
+            await self._stop_debug_recording_thread()
+            self.process_event(PipelineEvent(PipelineEventType.RUN_END))
+        finally:
+            self._unregister()
+
+    async def async_validate(self, pipeline_input: PipelineInput) -> None:
+        """Validate input and prepare the pipeline processor."""
+        request = pipeline_input.create_processor_request()
+        self._set_request_identity(request)
+        try:
+            await self._processor.async_validate(request)
+        except BaseException:
+            self._cleanup_failed_processor()
+            self._unregister()
+            raise
+
+    async def async_execute(
+        self, pipeline_input: PipelineInput, *, validate: bool = False
+    ) -> None:
+        """Run the pipeline processor with the Home Assistant lifecycle."""
+        request = pipeline_input.create_processor_request()
+        validation_error: PipelineError | None = None
+        self._set_request_identity(request)
+
+        try:
+            if validate:
+                try:
+                    await self._processor.async_validate(request)
+                except PipelineError as err:
+                    validation_error = err
+
+            self.start(
+                conversation_id=request.session.conversation_id,
+                device_id=request.device_id,
+                satellite_id=request.satellite_id,
+            )
+            await self._async_process(request, validation_error)
+        except PipelineError as err:
+            self._cleanup_failed_processor()
+            self.process_event(
+                PipelineEvent(
+                    PipelineEventType.ERROR,
+                    {"code": err.code, "message": err.message},
+                )
+            )
+        except BaseException:
+            self._cleanup_failed_processor()
+            raise
+        finally:
+            await self.end()
+
+    async def _async_process(
+        self,
+        request: _PipelineProcessorRequest,
+        validation_error: PipelineError | None,
+    ) -> None:
+        """Execute a previously validated processor request."""
+        if validation_error is not None:
+            raise validation_error
+        await self._processor.async_execute(request)
+
+    @callback
+    def _set_request_identity(self, request: _PipelineProcessorRequest) -> None:
+        """Set identity used by controller-provided processor services."""
+        self._device_id = request.device_id
+        self._satellite_id = request.satellite_id
+
+    @callback
+    def _cleanup_failed_processor(self) -> None:
+        """Clean up processor resources after an error."""
+        self._processor.cleanup()
+
+    @callback
+    def _unregister(self) -> None:
+        """Remove this run from the active run registry."""
+        if self._registered:
+            self.hass.data[KEY_ASSIST_PIPELINE].pipeline_runs.remove_run(self)
+            self._registered = False
+
+    @callback
+    def invalidate(self) -> None:
+        """Invalidate active processing for this run."""
+        self._processor.invalidate()
+
+    @callback
+    def accept_wake_word(self, wake_word_phrase: str) -> None:
+        """Apply Home Assistant's duplicate wake-up policy."""
+        last_wake_up = self.hass.data[DATA_LAST_WAKE_UP].get(wake_word_phrase)
+        if (
+            last_wake_up is not None
+            and (time.monotonic() - last_wake_up) < WAKE_WORD_COOLDOWN
+        ):
+            _LOGGER.debug("Duplicate wake-up detected for %s", wake_word_phrase)
+            raise DuplicateWakeUpDetectedError(wake_word_phrase)
+        self.hass.data[DATA_LAST_WAKE_UP][wake_word_phrase] = time.monotonic()
+
+    @callback
+    def capture_audio(self, audio_bytes: bytes | None) -> None:
+        """Forward an audio chunk to Home Assistant capture mechanisms."""
+        if self.debug_recording_queue is not None:
+            self.debug_recording_queue.put_nowait(audio_bytes)
+
+        if self._device_id is None:
+            return
+        audio_queue = self.hass.data[KEY_ASSIST_PIPELINE].device_audio_queues.get(
+            self._device_id
+        )
+        if audio_queue is None:
+            return
+        try:
+            audio_queue.queue.put_nowait(audio_bytes)
+        except asyncio.QueueFull:
+            audio_queue.overflow = True
+            _LOGGER.warning("Audio queue full for device %s", self._device_id)
+
+    @callback
+    def start_debug_recording(self, name: str) -> None:
+        """Start a new debug WAV file if recording is active."""
+        if self.debug_recording_queue is not None:
+            self.debug_recording_queue.put_nowait(name)
+
+    def _start_debug_recording_thread(self) -> None:
+        """Start the debug recording thread if configured."""
+        assert self.debug_recording_thread is None
+        if debug_recording_dir := self.hass.data[DATA_CONFIG].get(
+            CONF_DEBUG_RECORDING_DIR
+        ):
+            if self._device_id is None:
+                run_recording_dir = (
+                    Path(debug_recording_dir)
+                    / self.pipeline.name
+                    / str(time.monotonic_ns())
+                )
+            else:
+                run_recording_dir = (
+                    Path(debug_recording_dir)
+                    / self._device_id
+                    / self.pipeline.name
+                    / str(time.monotonic_ns())
+                )
+            self.debug_recording_queue = Queue()
+            self.debug_recording_thread = Thread(
+                target=_pipeline_debug_recording_thread_proc,
+                args=(run_recording_dir, self.debug_recording_queue),
+                daemon=True,
+            )
+            self.debug_recording_thread.start()
+
+    async def _stop_debug_recording_thread(self) -> None:
+        """Stop the debug recording thread."""
+        if self.debug_recording_thread is None or self.debug_recording_queue is None:
+            return
+        await self.hass.async_add_executor_job(self.debug_recording_thread.join)
+        self.debug_recording_queue = None
+        self.debug_recording_thread = None
+
+
+@dataclass(kw_only=True)
+class PipelineInput:
+    """Input to a pipeline run."""
+
+    run: PipelineRun
+    session: chat_session.ChatSession
+    stt_metadata: stt.SpeechMetadata | None = None
+    stt_stream: AsyncIterable[bytes] | None = None
+    wake_word_phrase: str | None = None
+    intent_input: str | None = None
+    tts_input: str | None = None
+    conversation_extra_system_prompt: str | None = None
+    device_id: str | None = None
+    satellite_id: str | None = None
+
+    async def execute(self, validate: bool = False) -> None:
+        """Run pipeline."""
+        await self.run.async_execute(self, validate=validate)
+
+    async def validate(self) -> None:
+        """Validate pipeline input against start stage."""
+        await self.run.async_validate(self)
+
+    def create_processor_request(self) -> _PipelineProcessorRequest:
+        """Create the private request passed to the pipeline processor."""
+        return _PipelineProcessorRequest(
+            session=self.session,
+            stt_metadata=self.stt_metadata,
+            stt_stream=self.stt_stream,
+            wake_word_phrase=self.wake_word_phrase,
+            intent_input=self.intent_input,
+            tts_input=self.tts_input,
+            conversation_extra_system_prompt=self.conversation_extra_system_prompt,
+            device_id=self.device_id,
+            satellite_id=self.satellite_id,
+        )
+
+
+def _create_pipeline_processor(run: PipelineRun) -> _PipelineProcessor:
+    """Create the default pipeline processor."""
+    return _DefaultPipelineProcessor(run)
+
+
+def _pipeline_debug_recording_thread_proc(
+    run_recording_dir: Path,
+    queue: Queue[str | bytes | None],
+    message_timeout: float = 5,
+) -> None:
+    """Write pipeline audio to debug WAV files."""
+    wav_writer: wave.Wave_write | None = None
+    try:
+        _LOGGER.debug("Saving wake/stt audio to %s", run_recording_dir)
+        run_recording_dir.mkdir(parents=True, exist_ok=True)
+        while True:
+            message = queue.get(timeout=message_timeout)
+            if message is None:
+                break
+            if isinstance(message, str):
+                if wav_writer is not None:
+                    wav_writer.close()
+                wav_path = run_recording_dir / f"{message}.wav"
+                wav_writer = wave.open(str(wav_path), "wb")
+                wav_writer.setframerate(SAMPLE_RATE)
+                wav_writer.setsampwidth(SAMPLE_WIDTH)
+                wav_writer.setnchannels(SAMPLE_CHANNELS)
+            elif isinstance(message, bytes) and wav_writer is not None:
+                wav_writer.writeframes(message)
+    except Empty:
+        pass
+    except Exception:
+        _LOGGER.exception("Unexpected error in debug recording thread")
+    finally:
+        if wav_writer is not None:
+            wav_writer.close()

homeassistant/components/assist_pipeline/runtime.py

@@ -14,7 +14,8 @@ from .const import DOMAIN
 from .models import PipelineEvent
 
 if TYPE_CHECKING:
-    from .pipeline import PipelineRun, PipelineStorageCollection
+    from .pipeline import PipelineStorageCollection
+    from .run import PipelineRun
 
 
 class PipelineRuns:
@@ -45,7 +46,7 @@ class PipelineRuns:
         if pipeline_runs := self._pipeline_runs.get(item_id):
             # Create a temporary list in case the list is modified while we iterate
             for pipeline_run in list(pipeline_runs.values()):
-                pipeline_run.abort_wake_word_detection = True
+                pipeline_run.invalidate()
 
 
 @dataclass(slots=True)

tests/components/assist_pipeline/test_init.py

@@ -672,7 +672,7 @@ async def test_pipeline_saved_audio_empty_queue(
 
         # Wrap original function to time out immediately
         _pipeline_debug_recording_thread_proc = (
-            assist_pipeline.pipeline._pipeline_debug_recording_thread_proc
+            assist_pipeline.run._pipeline_debug_recording_thread_proc
         )
 
         def proc_wrapper(run_recording_dir, queue):
@@ -686,7 +686,7 @@ async def test_pipeline_saved_audio_empty_queue(
             )
 
         with patch(
-            "homeassistant.components.assist_pipeline.pipeline._pipeline_debug_recording_thread_proc",
+            "homeassistant.components.assist_pipeline.run._pipeline_debug_recording_thread_proc",
             proc_wrapper,
         ):
             await assist_pipeline.async_pipeline_from_audio_stream(

tests/components/assist_pipeline/test_pipeline.py

@@ -1,6 +1,7 @@
 """Websocket tests for Voice Assistant integration."""
 
 from collections.abc import AsyncGenerator, Generator
+from dataclasses import FrozenInstanceError
 from pathlib import Path
 from typing import Any
 from unittest.mock import ANY, AsyncMock, Mock, patch
@@ -25,6 +26,10 @@ from homeassistant.components.assist_pipeline.const import (
     DATA_CONFIG,
     DOMAIN,
 )
+from homeassistant.components.assist_pipeline.default_pipeline import (
+    _async_local_fallback_intent_filter,
+    _DefaultPipelineProcessor,
+)
 from homeassistant.components.assist_pipeline.pipeline import (
     STORAGE_KEY,
     STORAGE_VERSION,
@@ -34,12 +39,12 @@ from homeassistant.components.assist_pipeline.pipeline import (
     PipelineEventType,
     PipelineStorageCollection,
     PipelineStore,
-    _async_local_fallback_intent_filter,
     async_create_default_pipeline,
     async_get_pipeline,
     async_get_pipelines,
     async_update_pipeline,
 )
+from homeassistant.components.assist_pipeline.run import _PipelineProcessorRequest
 from homeassistant.components.llm import LLMTools
 from homeassistant.const import ATTR_FRIENDLY_NAME, MATCH_ALL
 from homeassistant.core import Context, HomeAssistant
@@ -818,6 +823,55 @@ def test_pipeline_run_equality(hass: HomeAssistant, init_components) -> None:
     assert run_1 != 1234
 
 
+async def test_pipeline_run_delegates_to_processor(
+    hass: HomeAssistant,
+    init_components: None,
+    mock_chat_session: chat_session.ChatSession,
+) -> None:
+    """Test that the run controller delegates processing and owns the lifecycle."""
+    events: list[assist_pipeline.PipelineEvent] = []
+    processor = Mock(
+        response_audio=None,
+        supports_streaming_response=False,
+        async_validate=AsyncMock(),
+        async_execute=AsyncMock(),
+        invalidate=Mock(),
+        cleanup=Mock(),
+    )
+
+    with patch(
+        "homeassistant.components.assist_pipeline.run._create_pipeline_processor",
+        return_value=processor,
+    ):
+        pipeline_input = assist_pipeline.pipeline.PipelineInput(
+            intent_input="test input",
+            session=mock_chat_session,
+            run=assist_pipeline.pipeline.PipelineRun(
+                hass,
+                context=Context(),
+                pipeline=assist_pipeline.pipeline.async_get_pipeline(hass),
+                start_stage=assist_pipeline.PipelineStage.INTENT,
+                end_stage=assist_pipeline.PipelineStage.INTENT,
+                event_callback=events.append,
+            ),
+        )
+
+    await pipeline_input.execute(validate=True)
+
+    validate_request = processor.async_validate.await_args.args[0]
+    execute_request = processor.async_execute.await_args.args[0]
+    assert isinstance(validate_request, _PipelineProcessorRequest)
+    assert validate_request == execute_request
+    assert validate_request.session is pipeline_input.session
+    assert validate_request.intent_input == pipeline_input.intent_input
+    with pytest.raises(FrozenInstanceError):
+        validate_request.intent_input = "changed"
+    assert [event.type for event in events] == [
+        PipelineEventType.RUN_START,
+        PipelineEventType.RUN_END,
+    ]
+
+
 async def test_text_only_run_does_not_start_debug_recording_thread(
     hass: HomeAssistant,
     init_components,
@@ -882,16 +936,13 @@ async def test_tts_audio_output(
     await pipeline_input.validate()
 
     # Verify TTS audio settings
-    assert pipeline_input.run.tts_stream.options is not None
-    assert pipeline_input.run.tts_stream.options.get(tts.ATTR_PREFERRED_FORMAT) == "wav"
-    assert (
-        pipeline_input.run.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_RATE)
-        == 16000
-    )
-    assert (
-        pipeline_input.run.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_CHANNELS)
-        == 1
-    )
+    processor = pipeline_input.run._processor
+    assert isinstance(processor, _DefaultPipelineProcessor)
+    assert processor.tts_stream is not None
+    assert processor.tts_stream.options is not None
+    assert processor.tts_stream.options.get(tts.ATTR_PREFERRED_FORMAT) == "wav"
+    assert processor.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_RATE) == 16000
+    assert processor.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_CHANNELS) == 1
 
     with patch.object(mock_tts_entity, "get_tts_audio") as mock_get_tts_audio:
         await pipeline_input.execute()
@@ -1089,13 +1140,12 @@ async def test_sentence_trigger_overrides_conversation_agent(
             start_stage=assist_pipeline.PipelineStage.INTENT,
             end_stage=assist_pipeline.PipelineStage.INTENT,
             event_callback=events.append,
-            intent_agent="test-agent",  # not the default agent
         ),
     )
 
     # Ensure prepare succeeds
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info",
         return_value=conversation.AgentInfo(
             id="test-agent",
             name="Test Agent",
@@ -1105,7 +1155,7 @@ async def test_sentence_trigger_overrides_conversation_agent(
         await pipeline_input.validate()
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse"
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse"
     ) as mock_async_converse:
         await pipeline_input.execute()
 
@@ -1178,7 +1228,7 @@ async def test_prefer_local_intents(
 
     # Ensure prepare succeeds
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info",
         return_value=conversation.AgentInfo(
             id="test-agent",
             name="Test Agent",
@@ -1188,7 +1238,7 @@ async def test_prefer_local_intents(
         await pipeline_input.validate()
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse"
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse"
     ) as mock_async_converse:
         await pipeline_input.execute()
 
@@ -1247,7 +1297,7 @@ async def test_intent_continue_conversation(
 
     # Ensure prepare succeeds
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info",
         return_value=conversation.AgentInfo(
             id="test-agent",
             name="Test Agent",
@@ -1260,7 +1310,7 @@ async def test_intent_continue_conversation(
     response.async_set_speech("For how long?")
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse",
         return_value=conversation.ConversationResult(
             response=response,
             conversation_id=mock_chat_session.conversation_id,
@@ -1322,7 +1372,7 @@ async def test_intent_continue_conversation(
 
     # Ensure prepare succeeds
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info",
         return_value=conversation.AgentInfo(
             id="test-agent",
             name="Test Agent",
@@ -1338,7 +1388,7 @@ async def test_intent_continue_conversation(
     response.async_set_speech("Timer set for 20 minutes")
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse",
         return_value=conversation.ConversationResult(
             response=response,
             conversation_id=mock_chat_session.conversation_id,
@@ -1410,7 +1460,7 @@ async def test_stt_language_used_instead_of_conversation_language(
     await pipeline_input.validate()
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse",
         return_value=conversation.ConversationResult(
             intent.IntentResponse(pipeline.language)
         ),
@@ -1486,7 +1536,7 @@ async def test_tts_language_used_instead_of_conversation_language(
     await pipeline_input.validate()
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse",
         return_value=conversation.ConversationResult(
             intent.IntentResponse(pipeline.language)
         ),
@@ -1562,7 +1612,7 @@ async def test_pipeline_language_used_instead_of_conversation_language(
     await pipeline_input.validate()
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse",
         return_value=conversation.ConversationResult(
             intent.IntentResponse(pipeline.language)
         ),
@@ -1748,7 +1798,7 @@ async def test_chat_log_tts_streaming(
     mock_tts_entity.async_supports_streaming_input = Mock(return_value=True)
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info",
+        "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info",
         return_value=conversation.AgentInfo(
             id="test-agent",
             name="Test Agent",
@@ -1828,7 +1878,7 @@ async def test_chat_log_tts_streaming(
             return_value=LLMTools(tools=[mock_tool]),
         ),
         patch(
-            "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse",
+            "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse",
             mock_converse,
         ),
     ):
@@ -1913,7 +1963,7 @@ async def test_acknowledge(
         await pipeline_input.execute()
 
     with patch(
-        "homeassistant.components.assist_pipeline.PipelineRun.text_to_speech"
+        "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.text_to_speech"
     ) as text_to_speech:
 
         def _reset() -> None:
@@ -2108,7 +2158,7 @@ async def test_acknowledge_child_device_inherits_area(
         await pipeline_input.execute()
 
     with patch(
-        "homeassistant.components.assist_pipeline.PipelineRun.text_to_speech"
+        "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.text_to_speech"
     ) as text_to_speech:
         await _run("turn on light 1")
 
@@ -2177,7 +2227,7 @@ async def test_acknowledge_other_agents(
 
     with (
         patch(
-            "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info",
+            "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info",
             return_value=conversation.AgentInfo(
                 id="test-agent",
                 name="Test Agent",
@@ -2185,16 +2235,16 @@ async def test_acknowledge_other_agents(
             ),
         ),
         patch(
-            "homeassistant.components.assist_pipeline.PipelineRun.prepare_text_to_speech"
+            "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.prepare_text_to_speech"
         ),
         patch(
-            "homeassistant.components.assist_pipeline.PipelineRun.text_to_speech"
+            "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.text_to_speech"
         ) as text_to_speech,
         patch(
             "homeassistant.components.conversation.async_converse", return_value=None
         ) as async_converse,
         patch(
-            "homeassistant.components.assist_pipeline.PipelineRun._get_all_targets_in_satellite_area"
+            "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor._get_all_targets_in_satellite_area"
         ) as get_all_targets_in_satellite_area,
     ):
         pipeline_input = assist_pipeline.pipeline.PipelineInput(
@@ -2266,7 +2316,7 @@ async def test_stt_vad_enabled_based_on_audio_processing(
     # VAD should be used
     with (
         patch(
-            "homeassistant.components.assist_pipeline.pipeline.VoiceCommandSegmenter"
+            "homeassistant.components.assist_pipeline.default_pipeline.VoiceCommandSegmenter"
         ) as mock_vad,
         patch(
             "homeassistant.components.stt.async_get_speech_to_text_engine",
@@ -2317,7 +2367,7 @@ async def test_stt_vad_enabled_based_on_audio_processing(
     # VAD should NOT be used
     with (
         patch(
-            "homeassistant.components.assist_pipeline.pipeline.VoiceCommandSegmenter"
+            "homeassistant.components.assist_pipeline.default_pipeline.VoiceCommandSegmenter"
         ) as mock_vad,
         patch(
             "homeassistant.components.stt.async_get_speech_to_text_engine",

tests/components/assist_satellite/test_entity.py

@@ -199,7 +199,7 @@ async def test_pipeline_validation_error_ends_pipeline(
     )
 
     with patch(
-        "homeassistant.components.assist_pipeline.pipeline.PipelineRun.prepare_speech_to_text"
+        "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.prepare_speech_to_text"
     ):
         await entity.async_accept_pipeline_from_satellite(
             object(),  # type: ignore[arg-type]
@@ -927,7 +927,7 @@ async def test_ask_question(
     )
 
     async def speech_to_text(self, *args, **kwargs):
-        self.process_event(
+        self.host.process_event(
             PipelineEvent(
                 PipelineEventType.STT_END, {"stt_output": {"text": response_text}}
             )
@@ -950,10 +950,10 @@ async def test_ask_question(
         audio_stream = object()
         with (
             patch(
-                "homeassistant.components.assist_pipeline.pipeline.PipelineRun.prepare_speech_to_text"
+                "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.prepare_speech_to_text"
             ),
             patch(
-                "homeassistant.components.assist_pipeline.pipeline.PipelineRun.speech_to_text",
+                "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.speech_to_text",
                 speech_to_text,
             ),
         ):