Merge pull request #3694 from pipecat-ai/mb/claude-updates

Add observers, error handling, task management, and testing to CLAUDE.md

CLAUDE.md

@@ -63,6 +63,8 @@ Transport Input → Pipeline Source → [Processor1] → [Processor2] → ...
 
 - **RTVI** (`src/pipecat/processors/frameworks/rtvi.py`): Real-Time Voice Interface protocol bridging clients and the pipeline. `RTVIProcessor` handles incoming client messages (text input, audio, function call results). `RTVIObserver` converts pipeline frames to outgoing messages: user/bot speaking events, transcriptions, LLM/TTS lifecycle, function calls, metrics, and audio levels.
 
+- **Observers** (`src/pipecat/observers/`): Monitor frame flow without modifying the pipeline. Passed to `PipelineTask` via the `observers` parameter. Implement `on_process_frame()` and `on_push_frame()` callbacks.
+
 ### Important Patterns
 
 - **Context Aggregation**: `LLMContext` accumulates messages for LLM calls; `UserResponse` aggregates user input
@@ -78,6 +80,10 @@ Transport Input → Pipeline Source → [Processor1] → [Processor2] → ...
 
 - **Events**: Most classes in Pipecat have `BaseObject` as the very base class. `BaseObject` has support for events. Events can run in the background in an async task (default) or synchronously (`sync=True`) if we want immediate action. Synchronous event handlers need to exectue fast.
 
+- **Async Task Management**: Always use `self.create_task(coroutine, name)` instead of raw `asyncio.create_task()`. The `TaskManager` automatically tracks tasks and cleans them up on processor shutdown. Use `await self.cancel_task(task, timeout)` for cancellation.
+
+- **Error Handling**: Use `await self.push_error(msg, exception, fatal)` to push errors upstream. Services should use `fatal=False` (the default) so application code can handle errors and take action (e.g. switch to another service).
+
 ### Key Directories
 
 | Directory                 | Purpose                                            |
@@ -88,6 +94,7 @@ Transport Input → Pipeline Source → [Processor1] → [Processor2] → ...
 | `src/pipecat/services/`   | AI service integrations (60+ providers)            |
 | `src/pipecat/transports/` | Transport layer (Daily, LiveKit, WebSocket, Local) |
 | `src/pipecat/serializers/`| Frame serialization for WebSocket protocols        |
+| `src/pipecat/observers/`  | Pipeline observers for monitoring frame flow       |
 | `src/pipecat/audio/`      | VAD, filters, mixers, turn detection, DTMF         |
 | `src/pipecat/turns/`      | User turn management                               |
 
@@ -138,6 +145,10 @@ When adding a new service:
 6. Add metrics tracking via `MetricsData` if relevant
 7. Follow the pattern of existing services in `src/pipecat/services/`
 
+## Testing
+
+Test utilities live in `src/pipecat/tests/utils.py`. Use `run_test()` to send frames through a pipeline and assert expected output frames in each direction. Use `SleepFrame(sleep=N)` to add delays between frames. 
+
 ## Pull Requests
 
 After creating a PR, use `/changelog <pr_number>` to generate the changelog file and `/pr-description <pr_number>` to update the PR description.