Interruptions During `tts_say` Result in Dangling Tasks

[mitchelldehaven-caprx]

The tts_say action can be interrupted which can result in dangling tasks. The culprit seems to be that tts_say tasks essentially block other actions, if they are interrupted the current action counter doesn't track that.

Class ActionManager:
    def __init__(self, task: PipelineTask, flow_manager: "FlowManager"):
        ...
        @task.event_handler("on_frame_reached_downstream")
        async def on_frame_reached_downstream(task, frame):
            if isinstance(frame, FunctionActionFrame):
                # Run function action
                await frame.function(frame.action, flow_manager)
                self._decrement_ongoing_actions_count()
            elif isinstance(frame, BotStoppedSpeakingFrame):
                # Execute deferred post-actions if the bot's turn is over.
                # A BotStoppedSpeakingFrame only indicates that the bot's turn is over if there are
                # no ongoing actions (otherwise one of those actions may have been responsible for it).
                if self._ongoing_actions_count == 0:
                    await self._execute_deferred_post_actions()
            elif isinstance(frame, ActionFinishedFrame):
                # Handle action finished
                self._decrement_ongoing_actions_count()

If an interruption occurs while the tts_say is currently running, the associated ActionFnishedFrame will never hit downstream, which means the ongoing action count gets out of sync.

    async def _handle_tts_action(self, action: dict) -> None:
        """Built-in handler for TTS actions.

        Args:
            action: Action configuration containing 'text' to speak.
        """
        text = action.get("text")
        if not text:
            logger.error("TTS action missing 'text' field")
            return

        try:
            # Mark that we're starting the action
            self._increment_ongoing_actions_count()

            # Queue the action frame
            await self._task.queue_frame(TTSSpeakFrame(text=text))

            # Queue a frame marking the end of the action
            await self._task.queue_frame(ActionFinishedFrame()) <-- this won't make it downstream when an interruption occurs and its still queue
        except Exception as e:
            self._decrement_ongoing_actions_count()
            logger.error(f"TTS error: {e}")

I can make a PR if this is deemed to be an issue worth solving, the work around being extending the class and adding support for handling interruption frames in the downstream listener. I'm not sure if this extends to affecting other actions, but the only other post I saw in the issues about dangling tasks was for an MCP server function call which I suspect could also be the same issue.

[markbackman]

Tagging @kompfner to take a look as he worked on this initially.

[mitchelldehaven-caprx]

I can get a replication script up, but simply running a pipeline and flow with a tts_say pre_action followed up a brief wait and interruption frame, then closing the pipeline down should replicate it.