From 2d6ff5b7aae77a51e368eba06cfd7662eca0289c Mon Sep 17 00:00:00 2001
From: Xin Wang <your-email@example.com>
Date: Tue, 14 Jul 2026 11:20:18 +0800
Subject: [PATCH] Add on_client_ready method to BaseBrain and WorkflowBrain

- Implement on_client_ready in BaseBrain to handle client-visible state after the app message channel is ready.
- Extend WorkflowBrain with on_client_ready to replay state that may have been emitted before WebRTC data was ready.
- Update pipeline to call on_client_ready when a client connects.
- Enhance tests to verify the correct behavior of on_client_ready in WorkflowBrain.
---
 backend/services/brains/base.py           |  5 ++++
 backend/services/brains/workflow_brain.py | 13 ++++++++++
 backend/services/pipecat/pipeline.py      |  1 +
 backend/tests/test_brains.py              | 29 +++++++++++++++++++++++
 4 files changed, 48 insertions(+)

diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py
index b9f9036..3f3e3d2 100644
--- a/backend/services/brains/base.py
+++ b/backend/services/brains/base.py
@@ -86,6 +86,9 @@ class BaseBrain:
     async def on_connected(self) -> None:
         """Handle a connected client after the common greeting is queued."""
 
+    async def on_client_ready(self) -> None:
+        """Replay client-visible state after its app message channel is ready."""
+
     def record_user_message(self, content: str) -> None:
         """Observe a committed user message for brain-owned routing state."""
 
@@ -126,6 +129,8 @@ class Brain(Protocol):
 
     async def on_connected(self) -> None: ...
 
+    async def on_client_ready(self) -> None: ...
+
     def record_user_message(self, content: str) -> None: ...
 
     async def on_user_turn_end(self, content: str) -> bool: ...
diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py
index 128e890..50b78df 100644
--- a/backend/services/brains/workflow_brain.py
+++ b/backend/services/brains/workflow_brain.py
@@ -123,6 +123,19 @@ class WorkflowBrain(BaseBrain):
         await self._manager.initialize(node_config)
         logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}")
 
+    async def on_client_ready(self) -> None:
+        """Replay state that may have been emitted before WebRTC data was ready."""
+        current_node = (
+            str(self._manager.current_node)
+            if self._manager and self._manager.current_node
+            else self._engine.start_id
+        )
+        await self._emit_node_active(current_node)
+        await self._emit_variables(
+            reason="client_ready",
+            node_id=current_node,
+        )
+
     def record_user_message(self, content: str) -> None:
         if content and not self._ended:
             self._store.record("user", content)
diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py
index 4593554..4fac104 100644
--- a/backend/services/pipecat/pipeline.py
+++ b/backend/services/pipecat/pipeline.py
@@ -1218,6 +1218,7 @@ async def run_pipeline(
         if greeting and not greeting_transcript_sent:
             greeting_transcript_sent = True
             await queue_transcript("assistant", greeting, time_now_iso8601())
+        await brain.on_client_ready()
 
     @transport.event_handler("on_client_connected")
     async def on_client_connected(_transport, _client):
diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py
index cd43868..507cea6 100644
--- a/backend/tests/test_brains.py
+++ b/backend/tests/test_brains.py
@@ -498,6 +498,25 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
         start_brain._manager = FakeManager()
         await start_brain.on_connected()
         self.assertEqual(start_brain._manager.current_node, "start")
+        queued.clear()
+        await start_brain.on_client_ready()
+        ready_messages = [
+            frame.message
+            for frame in queued
+            if isinstance(frame, OutputTransportMessageUrgentFrame)
+        ]
+        self.assertIn(
+            {"type": "node-active", "nodeId": "start"},
+            ready_messages,
+        )
+        self.assertTrue(
+            any(
+                message.get("type") == "workflow-variables"
+                and message.get("reason") == "client_ready"
+                and message.get("nodeId") == "start"
+                for message in ready_messages
+            )
+        )
 
         agent_brain = WorkflowBrain(
             {
@@ -524,6 +543,16 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
         agent_brain._runtime = runtime
         agent_brain._manager = FakeManager("agent")
         queued.clear()
+        await agent_brain.on_client_ready()
+        self.assertTrue(
+            any(
+                isinstance(frame, OutputTransportMessageUrgentFrame)
+                and frame.message
+                == {"type": "node-active", "nodeId": "agent"}
+                for frame in queued
+            )
+        )
+        queued.clear()
         handled = await agent_brain.on_user_turn_end("请继续回答")
         self.assertTrue(handled)
         self.assertEqual(agent_brain._manager.current_node, "agent")