fix(actions): set ActionManager._ongoing_actions_finished_event at init

``ActionManager.__init__`` set ``_ongoing_actions_count = 0`` and built
``_ongoing_actions_finished_event = asyncio.Event()``, which defaults to
the clear state. This violates the natural invariant *count=0 ↔ event
set*: an immediate ``await event.wait()`` blocks forever even though no
action is in flight.

The counter helpers (``_increment_ongoing_actions_count`` /
``_decrement_ongoing_actions_count``) flip the event correctly on
transition, so the only buggy state was the initial one. Set the event
at init time to restore the invariant; behaviour from the first action
onward is unchanged.

Reproducer: ``await ActionManager(worker, mgr)._ongoing_actions_finished_event.wait()``
hung forever pre-fix; returns immediately post-fix. Added a regression
test pinning both ``count == 0`` and ``event.is_set()`` at init plus a
``wait_for(..., timeout=0.1)`` that exercises the path.

Author: Maurotb

changelog/287.fixed.md

@@ -0,0 +1,5 @@
+- Fixed `ActionManager._ongoing_actions_finished_event` starting in the cleared
+  state at init while `_ongoing_actions_count` started at 0, violating the
+  invariant *count=0 ↔ event set*. An immediate `await event.wait()` would
+  block forever even though no action was in flight. The event is now set at
+  init time; the counter helpers keep it consistent on increment/decrement.

src/pipecat_flows/actions.py

@@ -96,7 +96,14 @@ def __init__(self, worker: PipelineWorker, flow_manager: "FlowManager"):
         self._worker = worker
         self._flow_manager = flow_manager
         self._ongoing_actions_count = 0
+        # Invariant: ``_ongoing_actions_count == 0`` ↔ event SET.
+        # ``asyncio.Event()`` defaults to clear; combined with count=0 this
+        # violates the invariant — an immediate ``await event.wait()`` blocks
+        # forever even though no action is in flight. Setting the event here
+        # restores the invariant at init time; the counter helpers keep it
+        # consistent on increment/decrement.
         self._ongoing_actions_finished_event = asyncio.Event()
+        self._ongoing_actions_finished_event.set()
         self._deferred_post_actions: list[ActionConfig] = []
         self._showed_deprecation_warning_for_legacy_action_handler = False
 

tests/test_actions.py

@@ -75,6 +75,24 @@ async def test_initialization(self):
         self.assertIn("tts_say", self.action_manager._action_handlers)
         self.assertIn("end_conversation", self.action_manager._action_handlers)
 
+    async def test_ongoing_actions_finished_event_is_set_at_init(self):
+        """``_ongoing_actions_finished_event`` must be set whenever
+        ``_ongoing_actions_count == 0``. At init the count is 0, so the
+        event must be set — an immediate ``await event.wait()`` must
+        return without blocking.
+
+        Regression: ``asyncio.Event()`` defaults to clear; combined with
+        ``count=0`` this violated the invariant and made an immediate
+        wait block forever even though no action was in flight.
+        """
+        self.assertEqual(self.action_manager._ongoing_actions_count, 0)
+        self.assertTrue(self.action_manager._ongoing_actions_finished_event.is_set())
+        # The wait must return immediately, not hang.
+        await asyncio.wait_for(
+            self.action_manager._ongoing_actions_finished_event.wait(),
+            timeout=0.1,
+        )
+
     async def test_tts_action(self):
         """Test basic TTS action execution."""
         action = {"type": "tts_say", "text": "Hello"}