Report who was speaking, and when

[markbackman]

Summary

• Adds SpeakingObserver, reporting a conversation's speaking lifecycle as it happens. Seven kinds, each naming its subject and its layer: user_speech_started / user_speech_stopped (the speech itself, as the detector heard it), user_turn_started / user_turn_stopped (the turn strategy's ruling on that speech, which is what the rest of the pipeline acts on), bot_speech_started / bot_speech_stopped, and interruption.
• Speech is timed to speech, not to the detector. Both VAD frames carry the delay the detector needed to be sure (start_secs / stop_secs), so an interval drawn from arrival times is fat at both ends by exactly that. The moments report timestamp - start_secs and timestamp - stop_secs instead.
• A closing moment names its own start, so an interval reads whole from one record: no window function to pair rows, and a lost record leaves an open interval rather than quietly joining a start to the wrong stop. Duration is timestamp - started_at on the same row, so it isn't carried — a derived field is one more thing that can disagree with itself.
• The user is reported from two sources, and the second earns its place by what it alone holds: speech that never becomes a turn. A cough, a false start, or a pause mid-sentence reaches the microphone and stops there, so it appears in user_speech_* and nowhere else — including in bars drawn from these events, which would otherwise show someone speaking straight through their own pauses. The lag between the two sources is already accounted for by LatencyBreakdown.
• What counts as a turn is left to the consumer. TurnTrackingObserver answers where one turn ends and the next begins — a debounce timer after the bot stops — and that is a policy. A policy that ships inside a record can never be revised; these moments can be grouped again later, differently, over the same history.

A record is a kind, a time, and — where one closes a stretch of speech — the time it began:

{ "kind": "bot_speech_stopped", "timestamp": 1000.5, "started_at": 994.2 }

observer = SpeakingObserver()

@observer.event_handler("on_speech_event")
async def on_speech_event(observer, event):
    logger.info(event.model_dump_json())

Testing

• uv run pytest tests/test_speaking_observer.py — 9 tests, 100% line coverage: backdating against the detector's delay, closing moments naming their own start, voice activity and the strategy reported apart, a barge-in reading as an overlap, a stretch whose start was missed, and non-speaking frames left alone.
• An interruption is broadcast, arriving as two frames with two IDs (broadcast_frame_instance gives each a fresh one), so deduplicating on the ID alone cannot tell them apart. Only the downstream copy is read, and a test builds both siblings to check it.

Context

Part of the observability work for Pipecat Cloud, alongside LatencyBreakdown (#5445), ServiceMetricsObserver (#5607) and StartupTimingObserver (#5600). These moments are what draws a session's timeline — who held the floor, where two voices overlapped — and the silences between them are what the latency breakdown accounts for.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

Files with missing lines	Coverage Δ
src/pipecat/observers/speaking_observer.py	100.00% <100.00%> (ø)

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[markbackman]

More verbose naming, but it's more clear.

The idea is that it's useful to know the actual user speaking timings and the user turn timings. This will give us more flexibility to build out visualizations.

[aconchillo]

elif and remove return None

[aconchillo]

actually don't remove return None I missed the return type

[aconchillo]

LGTM! Very minor comment.