Merge branch 'main' into pk/service-settings-refactor
This commit is contained in:
kompfner
@@ -5,6 +5,7 @@
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
@@ -23,6 +24,13 @@ except ImportError:
|
||||
AIC_FILTER_MODULE = "pipecat.audio.filters.aic_filter"
|
||||
|
||||
|
||||
def _model_manager_ref_count(manager, key: str) -> int:
|
||||
"""Test helper: return reference count for a cache key (reads internal cache)."""
|
||||
with manager._lock:
|
||||
entry = manager._cache.get(key)
|
||||
return entry[1] if entry else 0
|
||||
|
||||
|
||||
class MockProcessor:
|
||||
"""A lightweight mock for AIC ProcessorAsync that mimics real behavior."""
|
||||
|
||||
@@ -99,10 +107,11 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Import AICFilter after confirming aic_sdk is available."""
|
||||
from pipecat.audio.filters.aic_filter import AICFilter
|
||||
from pipecat.audio.filters.aic_filter import AICFilter, AICModelManager
|
||||
from pipecat.frames.frames import FilterEnableFrame
|
||||
|
||||
cls.AICFilter = AICFilter
|
||||
cls.AICModelManager = AICModelManager
|
||||
cls.FilterEnableFrame = FilterEnableFrame
|
||||
|
||||
def setUp(self):
|
||||
@@ -122,13 +131,13 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
async def _start_filter_with_mocks(self, filter_instance, sample_rate=16000):
|
||||
"""Start a filter with mocked SDK components."""
|
||||
cache_key = "test-cache-key"
|
||||
with (
|
||||
patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.AICModelManager") as mock_manager_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor),
|
||||
):
|
||||
mock_model_cls.from_file.return_value = self.mock_model
|
||||
mock_model_cls.download_async = AsyncMock(return_value="/tmp/model")
|
||||
mock_manager_cls.acquire = AsyncMock(return_value=(self.mock_model, cache_key))
|
||||
mock_config_cls.optimal.return_value = MagicMock()
|
||||
await filter_instance.start(sample_rate)
|
||||
|
||||
@@ -171,37 +180,44 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase):
|
||||
filter_instance = self._create_filter_with_mocks(model_id=None, model_path=model_path)
|
||||
|
||||
with (
|
||||
patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.AICModelManager") as mock_manager_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor),
|
||||
):
|
||||
mock_model_cls.from_file.return_value = self.mock_model
|
||||
mock_manager_cls.acquire = AsyncMock(
|
||||
return_value=(self.mock_model, "path:/tmp/test.aicmodel")
|
||||
)
|
||||
mock_config_cls.optimal.return_value = MagicMock()
|
||||
|
||||
await filter_instance.start(16000)
|
||||
|
||||
mock_model_cls.from_file.assert_called_once_with(str(model_path))
|
||||
mock_manager_cls.acquire.assert_called_once()
|
||||
call_kw = mock_manager_cls.acquire.call_args[1]
|
||||
self.assertEqual(call_kw["model_path"], model_path)
|
||||
self.assertIsNone(call_kw["model_id"])
|
||||
self.assertTrue(filter_instance._aic_ready)
|
||||
self.assertEqual(filter_instance._sample_rate, 16000)
|
||||
self.assertEqual(filter_instance._frames_per_block, 160)
|
||||
|
||||
async def test_start_with_model_id_downloads(self):
|
||||
"""Test starting filter with model_id triggers download."""
|
||||
"""Test starting filter with model_id uses manager (download happens in manager)."""
|
||||
filter_instance = self._create_filter_with_mocks()
|
||||
|
||||
with (
|
||||
patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.AICModelManager") as mock_manager_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor),
|
||||
):
|
||||
mock_model_cls.from_file.return_value = self.mock_model
|
||||
mock_model_cls.download_async = AsyncMock(return_value="/tmp/model")
|
||||
mock_manager_cls.acquire = AsyncMock(
|
||||
return_value=(self.mock_model, "id:test-model:/custom/cache")
|
||||
)
|
||||
mock_config_cls.optimal.return_value = MagicMock()
|
||||
|
||||
await filter_instance.start(16000)
|
||||
|
||||
mock_model_cls.download_async.assert_called_once()
|
||||
mock_model_cls.from_file.assert_called_once()
|
||||
mock_manager_cls.acquire.assert_called_once()
|
||||
call_kw = mock_manager_cls.acquire.call_args[1]
|
||||
self.assertEqual(call_kw["model_id"], "test-model")
|
||||
self.assertTrue(filter_instance._aic_ready)
|
||||
|
||||
async def test_start_creates_processor(self):
|
||||
@@ -209,14 +225,13 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase):
|
||||
filter_instance = self._create_filter_with_mocks()
|
||||
|
||||
with (
|
||||
patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.AICModelManager") as mock_manager_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls,
|
||||
patch(
|
||||
f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor
|
||||
) as mock_processor_cls,
|
||||
):
|
||||
mock_model_cls.from_file.return_value = self.mock_model
|
||||
mock_model_cls.download_async = AsyncMock(return_value="/tmp/model")
|
||||
mock_manager_cls.acquire = AsyncMock(return_value=(self.mock_model, "test-cache-key"))
|
||||
mock_config_cls.optimal.return_value = MagicMock()
|
||||
|
||||
await filter_instance.start(16000)
|
||||
@@ -241,17 +256,21 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase):
|
||||
self.assertEqual(bypass_params[-1][1], 0.0)
|
||||
|
||||
async def test_stop_cleans_up_resources(self):
|
||||
"""Test that stop properly cleans up resources."""
|
||||
"""Test that stop properly cleans up resources and releases model reference."""
|
||||
filter_instance = self._create_filter_with_mocks()
|
||||
await self._start_filter_with_mocks(filter_instance)
|
||||
cache_key = filter_instance._model_cache_key
|
||||
|
||||
await filter_instance.stop()
|
||||
with patch(f"{AIC_FILTER_MODULE}.AICModelManager.release") as mock_release:
|
||||
await filter_instance.stop()
|
||||
|
||||
mock_release.assert_called_once_with(cache_key)
|
||||
self.assertTrue(self.mock_processor.processor_ctx.reset_called)
|
||||
self.assertIsNone(filter_instance._processor)
|
||||
self.assertIsNone(filter_instance._processor_ctx)
|
||||
self.assertIsNone(filter_instance._vad_ctx)
|
||||
self.assertIsNone(filter_instance._model)
|
||||
self.assertIsNone(filter_instance._model_cache_key)
|
||||
self.assertFalse(filter_instance._aic_ready)
|
||||
|
||||
async def test_stop_without_start(self):
|
||||
@@ -261,6 +280,177 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase):
|
||||
# Should not raise
|
||||
await filter_instance.stop()
|
||||
|
||||
async def test_model_manager_reference_count(self):
|
||||
"""Test that AICModelManager reference count increments and decrements correctly."""
|
||||
model_path = Path("/tmp/refcount-test.aicmodel")
|
||||
mock_model = MockModel()
|
||||
manager = self.AICModelManager
|
||||
|
||||
with patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls:
|
||||
mock_model_cls.from_file.return_value = mock_model
|
||||
|
||||
# Acquire first reference
|
||||
model1, key = await manager.acquire(model_path=model_path)
|
||||
self.assertEqual(model1, mock_model)
|
||||
self.assertEqual(_model_manager_ref_count(manager, key), 1)
|
||||
|
||||
# Acquire second reference (same key, cached)
|
||||
model2, key2 = await manager.acquire(model_path=model_path)
|
||||
self.assertIs(model2, model1)
|
||||
self.assertEqual(key2, key)
|
||||
self.assertEqual(_model_manager_ref_count(manager, key), 2)
|
||||
|
||||
# Release one reference
|
||||
manager.release(key)
|
||||
self.assertEqual(_model_manager_ref_count(manager, key), 1)
|
||||
|
||||
# Release last reference (model evicted from cache)
|
||||
manager.release(key)
|
||||
self.assertEqual(_model_manager_ref_count(manager, key), 0)
|
||||
|
||||
async def test_model_manager_concurrent_load_deduplication(self):
|
||||
"""Test that concurrent acquire calls for the same key share a single load task."""
|
||||
model_path = Path("/tmp/concurrent-load-test.aicmodel")
|
||||
mock_model = MockModel()
|
||||
manager = self.AICModelManager
|
||||
load_count = 0
|
||||
|
||||
def from_file_once(path):
|
||||
nonlocal load_count
|
||||
load_count += 1
|
||||
time.sleep(0.02) # yield so other acquire callers can hit _loading and await same task
|
||||
return mock_model
|
||||
|
||||
with patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls:
|
||||
mock_model_cls.from_file.side_effect = from_file_once
|
||||
|
||||
# Start several acquire calls concurrently before any completes
|
||||
results = await asyncio.gather(
|
||||
manager.acquire(model_path=model_path),
|
||||
manager.acquire(model_path=model_path),
|
||||
manager.acquire(model_path=model_path),
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
load_count, 1, "Model.from_file should be called once for concurrent callers"
|
||||
)
|
||||
model1, key1 = results[0]
|
||||
model2, key2 = results[1]
|
||||
model3, key3 = results[2]
|
||||
self.assertIs(model1, mock_model)
|
||||
self.assertIs(model2, mock_model)
|
||||
self.assertIs(model3, mock_model)
|
||||
self.assertEqual(key1, key2)
|
||||
self.assertEqual(key2, key3)
|
||||
self.assertEqual(_model_manager_ref_count(manager, key1), 3)
|
||||
|
||||
# Release all references
|
||||
manager.release(key1)
|
||||
manager.release(key1)
|
||||
manager.release(key1)
|
||||
self.assertEqual(_model_manager_ref_count(manager, key1), 0)
|
||||
|
||||
async def test_load_model_from_file_invalid_args_raises(self):
|
||||
"""Test _load_model_from_file defensive else: raises ValueError."""
|
||||
manager = self.AICModelManager
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
await manager._load_model_from_file(
|
||||
"key",
|
||||
model_path=None,
|
||||
model_id=None,
|
||||
model_download_dir=None,
|
||||
)
|
||||
self.assertIn("Unexpected", str(ctx.exception))
|
||||
|
||||
async def test_model_manager_acquire_by_model_id_hits_download_path(self):
|
||||
"""Test acquire with model_id runs download path in _load_model_from_file."""
|
||||
model_id = "test-model-id"
|
||||
model_download_dir = Path("/tmp/aic-downloads")
|
||||
mock_model = MockModel()
|
||||
manager = self.AICModelManager
|
||||
|
||||
with patch(f"{AIC_FILTER_MODULE}.Model") as mock_model_cls:
|
||||
mock_model_cls.download_async = AsyncMock(
|
||||
return_value="/tmp/aic-downloads/model.aicmodel"
|
||||
)
|
||||
mock_model_cls.from_file.return_value = mock_model
|
||||
|
||||
model, key = await manager.acquire(
|
||||
model_id=model_id,
|
||||
model_download_dir=model_download_dir,
|
||||
)
|
||||
|
||||
mock_model_cls.download_async.assert_called_once()
|
||||
mock_model_cls.from_file.assert_called_once_with("/tmp/aic-downloads/model.aicmodel")
|
||||
self.assertIs(model, mock_model)
|
||||
self.assertEqual(_model_manager_ref_count(manager, key), 1)
|
||||
manager.release(key)
|
||||
|
||||
def test_get_cache_key_invalid_raises(self):
|
||||
"""Test _get_cache_key raises ValueError for invalid args."""
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self.AICModelManager._get_cache_key(model_path=None, model_id=None)
|
||||
self.assertIn("model_path", str(ctx.exception))
|
||||
|
||||
with self.assertRaises(ValueError) as ctx2:
|
||||
self.AICModelManager._get_cache_key(
|
||||
model_path=None,
|
||||
model_id="x",
|
||||
model_download_dir=None,
|
||||
)
|
||||
self.assertIn("model_download_dir", str(ctx2.exception))
|
||||
|
||||
async def test_start_processor_init_failure(self):
|
||||
"""Test start() when ProcessorAsync raises: exception logged, _aic_ready False."""
|
||||
filter_instance = self._create_filter_with_mocks()
|
||||
|
||||
with (
|
||||
patch(f"{AIC_FILTER_MODULE}.AICModelManager") as mock_manager_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls,
|
||||
patch(
|
||||
f"{AIC_FILTER_MODULE}.ProcessorAsync",
|
||||
side_effect=RuntimeError("SDK init failed"),
|
||||
),
|
||||
):
|
||||
mock_manager_cls.acquire = AsyncMock(return_value=(self.mock_model, "test-key"))
|
||||
mock_config_cls.optimal.return_value = MagicMock()
|
||||
|
||||
await filter_instance.start(16000)
|
||||
|
||||
self.assertIsNone(filter_instance._processor)
|
||||
self.assertFalse(filter_instance._aic_ready)
|
||||
|
||||
async def test_start_parameter_fixed_error_logged(self):
|
||||
"""Test start() when set_parameter raises ParameterFixedError: logged, no raise."""
|
||||
filter_instance = self._create_filter_with_mocks()
|
||||
self.mock_processor.processor_ctx.set_parameter = MagicMock(
|
||||
side_effect=aic_sdk.ParameterFixedError("fixed")
|
||||
)
|
||||
|
||||
with (
|
||||
patch(f"{AIC_FILTER_MODULE}.AICModelManager") as mock_manager_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorConfig") as mock_config_cls,
|
||||
patch(f"{AIC_FILTER_MODULE}.ProcessorAsync", return_value=self.mock_processor),
|
||||
):
|
||||
mock_manager_cls.acquire = AsyncMock(return_value=(self.mock_model, "test-key"))
|
||||
mock_config_cls.optimal.return_value = MagicMock()
|
||||
|
||||
await filter_instance.start(16000)
|
||||
|
||||
self.assertTrue(filter_instance._aic_ready)
|
||||
|
||||
async def test_process_frame_set_parameter_exception_logged(self):
|
||||
"""Test process_frame when set_parameter raises: exception logged, no raise."""
|
||||
filter_instance = self._create_filter_with_mocks()
|
||||
await self._start_filter_with_mocks(filter_instance)
|
||||
filter_instance._processor_ctx.set_parameter = MagicMock(
|
||||
side_effect=ValueError("param error")
|
||||
)
|
||||
|
||||
await filter_instance.process_frame(self.FilterEnableFrame(enable=True))
|
||||
|
||||
self.assertFalse(filter_instance._bypass)
|
||||
|
||||
async def test_process_frame_enable(self):
|
||||
"""Test processing FilterEnableFrame to enable filtering."""
|
||||
filter_instance = self._create_filter_with_mocks()
|
||||
|
||||
@@ -24,7 +24,9 @@ from pipecat.frames.frames import (
|
||||
LLMThoughtEndFrame,
|
||||
LLMThoughtStartFrame,
|
||||
LLMThoughtTextFrame,
|
||||
StartFrame,
|
||||
TranscriptionFrame,
|
||||
UserMuteStartedFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
VADUserStartedSpeakingFrame,
|
||||
@@ -40,7 +42,11 @@ from pipecat.processors.aggregators.llm_response_universal import (
|
||||
LLMUserAggregatorParams,
|
||||
)
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
from pipecat.turns.user_mute import FirstSpeechUserMuteStrategy, FunctionCallUserMuteStrategy
|
||||
from pipecat.turns.user_mute import (
|
||||
FirstSpeechUserMuteStrategy,
|
||||
FunctionCallUserMuteStrategy,
|
||||
MuteUntilFirstBotCompleteUserMuteStrategy,
|
||||
)
|
||||
from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy
|
||||
from pipecat.turns.user_turn_strategies import UserTurnStrategies
|
||||
|
||||
@@ -386,6 +392,42 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
self.assertIsNone(strategy) # strategy is None for end/cancel
|
||||
self.assertEqual(message.content, "Hello!")
|
||||
|
||||
async def test_start_frame_before_mute_event(self):
|
||||
"""StartFrame must reach downstream before mute events are broadcast.
|
||||
|
||||
With MuteUntilFirstBotCompleteUserMuteStrategy, the mute logic should
|
||||
not run on control frames (StartFrame, EndFrame, CancelFrame). This
|
||||
ensures StartFrame reaches downstream processors before
|
||||
UserMuteStartedFrame is broadcast.
|
||||
|
||||
The default TurnAnalyzerUserTurnStopStrategy broadcasts a
|
||||
SpeechControlParamsFrame when it processes StartFrame, which gets
|
||||
re-queued to the aggregator. That non-control frame legitimately
|
||||
triggers the mute state change, so UserMuteStartedFrame follows
|
||||
StartFrame — but crucially, after it.
|
||||
"""
|
||||
context = LLMContext()
|
||||
|
||||
user_aggregator = LLMUserAggregator(
|
||||
context,
|
||||
params=LLMUserAggregatorParams(
|
||||
user_mute_strategies=[MuteUntilFirstBotCompleteUserMuteStrategy()],
|
||||
),
|
||||
)
|
||||
|
||||
pipeline = Pipeline([user_aggregator])
|
||||
|
||||
# run_test internally sends StartFrame via PipelineRunner. With
|
||||
# ignore_start=False we can verify ordering: StartFrame must arrive
|
||||
# before UserMuteStartedFrame. Before the fix, UserMuteStartedFrame
|
||||
# was broadcast before StartFrame reached downstream processors.
|
||||
(down_frames, _) = await run_test(
|
||||
pipeline,
|
||||
frames_to_send=[],
|
||||
expected_down_frames=[StartFrame, UserMuteStartedFrame],
|
||||
ignore_start=False,
|
||||
)
|
||||
|
||||
|
||||
class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_empty(self):
|
||||
|
||||
@@ -25,7 +25,6 @@ from pipecat.frames.frames import (
|
||||
from pipecat.pipeline.pipeline import Pipeline
|
||||
from pipecat.processors.filters.identity_filter import IdentityFilter
|
||||
from pipecat.processors.frame_processor import (
|
||||
INTERRUPTION_COMPLETION_TIMEOUT,
|
||||
FrameDirection,
|
||||
FrameProcessor,
|
||||
)
|
||||
@@ -521,7 +520,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
|
||||
# Complete after the timeout so the warning fires
|
||||
# but the test doesn't hang.
|
||||
async def delayed_complete():
|
||||
await asyncio.sleep(INTERRUPTION_COMPLETION_TIMEOUT + 1.0)
|
||||
await asyncio.sleep(1.0)
|
||||
frame.complete()
|
||||
|
||||
asyncio.create_task(delayed_complete())
|
||||
@@ -532,7 +531,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
if isinstance(frame, TextFrame):
|
||||
await self.push_interruption_task_frame_and_wait()
|
||||
await self.push_interruption_task_frame_and_wait(timeout=0.5)
|
||||
await self.push_frame(OutputTransportMessageUrgentFrame(message="done"))
|
||||
else:
|
||||
await self.push_frame(frame, direction)
|
||||
|
||||
@@ -223,3 +223,77 @@ async def test_openai_llm_emits_error_frame_on_exception():
|
||||
assert "Error during completion" in pushed_errors[0]["error_msg"]
|
||||
assert "API Error" in pushed_errors[0]["error_msg"]
|
||||
assert isinstance(pushed_errors[0]["exception"], RuntimeError)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_openai_llm_async_iterator_closed_on_stream_end():
|
||||
"""Test that the async iterator is explicitly closed after stream consumption.
|
||||
|
||||
This prevents uvloop's broken asyncgen finalizer from firing on Python 3.12+
|
||||
when async generators are garbage-collected without explicit cleanup.
|
||||
See MagicStack/uvloop#699.
|
||||
"""
|
||||
with patch.object(OpenAILLMService, "create_client"):
|
||||
service = OpenAILLMService(model="gpt-4")
|
||||
service._client = AsyncMock()
|
||||
|
||||
# Track if the iterator's aclose was called
|
||||
iterator_aclosed = False
|
||||
stream_closed = False
|
||||
|
||||
class MockAsyncIterator:
|
||||
"""Mock async iterator that tracks aclose() calls."""
|
||||
|
||||
def __init__(self):
|
||||
self.iteration_count = 0
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
self.iteration_count += 1
|
||||
if self.iteration_count > 2:
|
||||
raise StopAsyncIteration()
|
||||
# Return a minimal chunk
|
||||
mock_chunk = AsyncMock()
|
||||
mock_chunk.usage = None
|
||||
mock_chunk.model = None
|
||||
mock_chunk.choices = []
|
||||
return mock_chunk
|
||||
|
||||
async def aclose(self):
|
||||
nonlocal iterator_aclosed
|
||||
iterator_aclosed = True
|
||||
|
||||
class MockAsyncStream:
|
||||
"""Mock stream whose __aiter__ returns a separate iterator object."""
|
||||
|
||||
def __init__(self, iterator):
|
||||
self._iterator = iterator
|
||||
|
||||
def __aiter__(self):
|
||||
return self._iterator
|
||||
|
||||
async def close(self):
|
||||
nonlocal stream_closed
|
||||
stream_closed = True
|
||||
|
||||
mock_iterator = MockAsyncIterator()
|
||||
mock_stream = MockAsyncStream(mock_iterator)
|
||||
|
||||
service._stream_chat_completions_specific_context = AsyncMock(return_value=mock_stream)
|
||||
service._stream_chat_completions_universal_context = AsyncMock(return_value=mock_stream)
|
||||
service.start_ttfb_metrics = AsyncMock()
|
||||
service.stop_ttfb_metrics = AsyncMock()
|
||||
service.start_llm_usage_metrics = AsyncMock()
|
||||
|
||||
context = LLMContext(
|
||||
messages=[{"role": "user", "content": "Hello"}],
|
||||
)
|
||||
|
||||
await service._process_context(context)
|
||||
|
||||
# Verify the iterator was explicitly closed (prevents uvloop crash)
|
||||
assert iterator_aclosed, "Async iterator should be explicitly closed"
|
||||
# Verify the stream was also closed (releases HTTP resources)
|
||||
assert stream_closed, "Stream should be closed to release HTTP resources"
|
||||
|
||||
127
tests/test_tracing_context.py
Normal file
127
tests/test_tracing_context.py
Normal file
@@ -0,0 +1,127 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import unittest
|
||||
|
||||
try:
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
|
||||
HAS_OPENTELEMETRY = True
|
||||
except ImportError:
|
||||
HAS_OPENTELEMETRY = False
|
||||
|
||||
from pipecat.utils.tracing.tracing_context import TracingContext
|
||||
|
||||
|
||||
@unittest.skipUnless(HAS_OPENTELEMETRY, "opentelemetry not installed")
|
||||
class TestTracingContext(unittest.TestCase):
|
||||
"""Tests for TracingContext."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Set up a tracer provider for generating span contexts."""
|
||||
cls._provider = TracerProvider()
|
||||
cls._tracer = cls._provider.get_tracer("test")
|
||||
|
||||
def test_initial_state_is_empty(self):
|
||||
"""Test that a new TracingContext starts with no context set."""
|
||||
ctx = TracingContext()
|
||||
self.assertIsNone(ctx.get_conversation_context())
|
||||
self.assertIsNone(ctx.get_turn_context())
|
||||
self.assertIsNone(ctx.conversation_id)
|
||||
|
||||
def test_set_and_get_conversation_context(self):
|
||||
"""Test setting and retrieving conversation context."""
|
||||
ctx = TracingContext()
|
||||
span = self._tracer.start_span("conv")
|
||||
span_context = span.get_span_context()
|
||||
|
||||
ctx.set_conversation_context(span_context, "conv-123")
|
||||
|
||||
self.assertIsNotNone(ctx.get_conversation_context())
|
||||
self.assertEqual(ctx.conversation_id, "conv-123")
|
||||
span.end()
|
||||
|
||||
def test_clear_conversation_context(self):
|
||||
"""Test clearing conversation context by passing None."""
|
||||
ctx = TracingContext()
|
||||
span = self._tracer.start_span("conv")
|
||||
|
||||
ctx.set_conversation_context(span.get_span_context(), "conv-123")
|
||||
self.assertIsNotNone(ctx.get_conversation_context())
|
||||
|
||||
ctx.set_conversation_context(None)
|
||||
self.assertIsNone(ctx.get_conversation_context())
|
||||
self.assertIsNone(ctx.conversation_id)
|
||||
span.end()
|
||||
|
||||
def test_set_and_get_turn_context(self):
|
||||
"""Test setting and retrieving turn context."""
|
||||
ctx = TracingContext()
|
||||
span = self._tracer.start_span("turn")
|
||||
span_context = span.get_span_context()
|
||||
|
||||
ctx.set_turn_context(span_context)
|
||||
|
||||
self.assertIsNotNone(ctx.get_turn_context())
|
||||
span.end()
|
||||
|
||||
def test_clear_turn_context(self):
|
||||
"""Test clearing turn context by passing None."""
|
||||
ctx = TracingContext()
|
||||
span = self._tracer.start_span("turn")
|
||||
|
||||
ctx.set_turn_context(span.get_span_context())
|
||||
self.assertIsNotNone(ctx.get_turn_context())
|
||||
|
||||
ctx.set_turn_context(None)
|
||||
self.assertIsNone(ctx.get_turn_context())
|
||||
span.end()
|
||||
|
||||
def test_generate_conversation_id(self):
|
||||
"""Test that generated conversation IDs are unique UUIDs."""
|
||||
id1 = TracingContext.generate_conversation_id()
|
||||
id2 = TracingContext.generate_conversation_id()
|
||||
self.assertIsInstance(id1, str)
|
||||
self.assertNotEqual(id1, id2)
|
||||
|
||||
def test_instances_are_isolated(self):
|
||||
"""Test that two TracingContext instances do not share state."""
|
||||
ctx_a = TracingContext()
|
||||
ctx_b = TracingContext()
|
||||
|
||||
span = self._tracer.start_span("turn")
|
||||
|
||||
ctx_a.set_turn_context(span.get_span_context())
|
||||
ctx_a.set_conversation_context(span.get_span_context(), "conv-a")
|
||||
|
||||
# ctx_b should still be empty
|
||||
self.assertIsNone(ctx_b.get_turn_context())
|
||||
self.assertIsNone(ctx_b.get_conversation_context())
|
||||
self.assertIsNone(ctx_b.conversation_id)
|
||||
span.end()
|
||||
|
||||
def test_conversation_and_turn_are_independent(self):
|
||||
"""Test that clearing turn context does not affect conversation context."""
|
||||
ctx = TracingContext()
|
||||
conv_span = self._tracer.start_span("conv")
|
||||
turn_span = self._tracer.start_span("turn")
|
||||
|
||||
ctx.set_conversation_context(conv_span.get_span_context(), "conv-1")
|
||||
ctx.set_turn_context(turn_span.get_span_context())
|
||||
|
||||
# Clear turn but conversation should remain
|
||||
ctx.set_turn_context(None)
|
||||
self.assertIsNone(ctx.get_turn_context())
|
||||
self.assertIsNotNone(ctx.get_conversation_context())
|
||||
self.assertEqual(ctx.conversation_id, "conv-1")
|
||||
|
||||
conv_span.end()
|
||||
turn_span.end()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
505
tests/test_turn_trace_observer.py
Normal file
505
tests/test_turn_trace_observer.py
Normal file
@@ -0,0 +1,505 @@
|
||||
#
|
||||
# Copyright (c) 2024-2026, Daily
|
||||
#
|
||||
# SPDX-License-Identifier: BSD 2-Clause License
|
||||
#
|
||||
|
||||
import asyncio
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
try:
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult
|
||||
|
||||
HAS_OPENTELEMETRY = True
|
||||
except ImportError:
|
||||
HAS_OPENTELEMETRY = False
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
)
|
||||
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
|
||||
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
|
||||
from pipecat.processors.filters.identity_filter import IdentityFilter
|
||||
from pipecat.tests.utils import SleepFrame, run_test
|
||||
from pipecat.utils.tracing.tracing_context import TracingContext
|
||||
from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
|
||||
|
||||
if HAS_OPENTELEMETRY:
|
||||
|
||||
class _InMemorySpanExporter(SpanExporter):
|
||||
"""Simple in-memory span exporter for testing."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the exporter."""
|
||||
self._spans = []
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def export(self, spans):
|
||||
"""Export spans to memory."""
|
||||
with self._lock:
|
||||
self._spans.extend(spans)
|
||||
return SpanExportResult.SUCCESS
|
||||
|
||||
def get_finished_spans(self):
|
||||
"""Return collected spans."""
|
||||
with self._lock:
|
||||
return list(self._spans)
|
||||
|
||||
def clear(self):
|
||||
"""Clear collected spans."""
|
||||
with self._lock:
|
||||
self._spans.clear()
|
||||
|
||||
|
||||
@unittest.skipUnless(HAS_OPENTELEMETRY, "opentelemetry not installed")
|
||||
class TestTurnTraceObserver(unittest.IsolatedAsyncioTestCase):
|
||||
"""Tests for TurnTraceObserver."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up a fresh provider and exporter for each test.
|
||||
|
||||
We create a dedicated TracerProvider per test and inject its tracer
|
||||
directly into the observer, avoiding the global provider singleton.
|
||||
"""
|
||||
self._exporter = _InMemorySpanExporter()
|
||||
self._provider = TracerProvider()
|
||||
self._provider.add_span_processor(SimpleSpanProcessor(self._exporter))
|
||||
self._tracer = self._provider.get_tracer("pipecat.turn")
|
||||
|
||||
def tearDown(self):
|
||||
"""Shut down the provider to flush spans."""
|
||||
self._provider.shutdown()
|
||||
|
||||
def _create_observers(self, conversation_id=None, tracing_context=None):
|
||||
"""Create a standard set of turn/trace observers.
|
||||
|
||||
Args:
|
||||
conversation_id: Optional conversation ID.
|
||||
tracing_context: Optional TracingContext instance.
|
||||
|
||||
Returns:
|
||||
Tuple of (turn_tracker, latency_tracker, trace_observer, tracing_context).
|
||||
"""
|
||||
tracing_context = tracing_context or TracingContext()
|
||||
turn_tracker = TurnTrackingObserver(turn_end_timeout_secs=0.2)
|
||||
latency_tracker = UserBotLatencyObserver()
|
||||
trace_observer = TurnTraceObserver(
|
||||
turn_tracker,
|
||||
latency_tracker=latency_tracker,
|
||||
conversation_id=conversation_id,
|
||||
tracing_context=tracing_context,
|
||||
)
|
||||
# Inject the test tracer so spans go to our in-memory exporter
|
||||
trace_observer._tracer = self._tracer
|
||||
return turn_tracker, latency_tracker, trace_observer, tracing_context
|
||||
|
||||
def _all_observers(self, trace_observer):
|
||||
"""Return the list of observers needed for run_test."""
|
||||
return [trace_observer._turn_tracker, trace_observer._latency_tracker, trace_observer]
|
||||
|
||||
def _get_spans_by_name(self, name):
|
||||
"""Return finished spans with the given name."""
|
||||
return [s for s in self._exporter.get_finished_spans() if s.name == name]
|
||||
|
||||
async def test_conversation_span_created_on_start_frame(self):
|
||||
"""Test that a conversation span is created when StartFrame is observed."""
|
||||
_, _, trace_observer, _ = self._create_observers(conversation_id="test-conv")
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=self._all_observers(trace_observer),
|
||||
)
|
||||
|
||||
# End conversation to flush the conversation span (normally done by PipelineTask._cleanup)
|
||||
trace_observer.end_conversation_tracing()
|
||||
|
||||
conv_spans = self._get_spans_by_name("conversation")
|
||||
self.assertEqual(len(conv_spans), 1)
|
||||
self.assertEqual(conv_spans[0].attributes["conversation.id"], "test-conv")
|
||||
self.assertEqual(conv_spans[0].attributes["conversation.type"], "voice")
|
||||
|
||||
async def test_turn_spans_created_for_each_turn(self):
|
||||
"""Test that a turn span is created for each conversation turn."""
|
||||
_, _, trace_observer, _ = self._create_observers()
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
# Turn 1
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.05),
|
||||
# Turn 2
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=self._all_observers(trace_observer),
|
||||
)
|
||||
|
||||
turn_spans = self._get_spans_by_name("turn")
|
||||
self.assertEqual(len(turn_spans), 2)
|
||||
turn_numbers = {s.attributes["turn.number"] for s in turn_spans}
|
||||
self.assertEqual(turn_numbers, {1, 2})
|
||||
|
||||
async def test_turn_spans_are_children_of_conversation(self):
|
||||
"""Test that turn spans are parented under the conversation span."""
|
||||
_, _, trace_observer, _ = self._create_observers()
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=self._all_observers(trace_observer),
|
||||
)
|
||||
|
||||
# End conversation to flush the conversation span
|
||||
trace_observer.end_conversation_tracing()
|
||||
|
||||
conv_spans = self._get_spans_by_name("conversation")
|
||||
turn_spans = self._get_spans_by_name("turn")
|
||||
self.assertEqual(len(conv_spans), 1)
|
||||
self.assertEqual(len(turn_spans), 1)
|
||||
|
||||
# Turn span's parent should be the conversation span
|
||||
conv_span_id = conv_spans[0].context.span_id
|
||||
turn_parent_id = turn_spans[0].parent.span_id
|
||||
self.assertEqual(turn_parent_id, conv_span_id)
|
||||
|
||||
async def test_interrupted_turn_marked(self):
|
||||
"""Test that an interrupted turn span has was_interrupted=True."""
|
||||
_, _, trace_observer, _ = self._create_observers()
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
# User interrupts
|
||||
UserStartedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=self._all_observers(trace_observer),
|
||||
)
|
||||
|
||||
# End conversation to flush remaining spans
|
||||
trace_observer.end_conversation_tracing()
|
||||
|
||||
turn_spans = self._get_spans_by_name("turn")
|
||||
self.assertGreaterEqual(len(turn_spans), 1)
|
||||
# First turn should be interrupted
|
||||
interrupted_turns = [s for s in turn_spans if s.attributes.get("turn.was_interrupted")]
|
||||
self.assertGreaterEqual(len(interrupted_turns), 1)
|
||||
|
||||
async def test_tracing_context_updated_during_turn(self):
|
||||
"""Test that TracingContext is populated during a turn and cleared after."""
|
||||
tracing_ctx = TracingContext()
|
||||
_, _, trace_observer, _ = self._create_observers(tracing_context=tracing_ctx)
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=self._all_observers(trace_observer),
|
||||
)
|
||||
|
||||
# After the turn ends, turn context should be cleared
|
||||
self.assertIsNone(tracing_ctx.get_turn_context())
|
||||
|
||||
async def test_tracing_context_cleared_after_conversation_end(self):
|
||||
"""Test that TracingContext is cleared when conversation tracing ends."""
|
||||
tracing_ctx = TracingContext()
|
||||
_, _, trace_observer, _ = self._create_observers(tracing_context=tracing_ctx)
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=self._all_observers(trace_observer),
|
||||
)
|
||||
|
||||
# Manually end conversation tracing (as PipelineTask._cleanup does)
|
||||
trace_observer.end_conversation_tracing()
|
||||
|
||||
self.assertIsNone(tracing_ctx.get_conversation_context())
|
||||
self.assertIsNone(tracing_ctx.get_turn_context())
|
||||
self.assertIsNone(tracing_ctx.conversation_id)
|
||||
|
||||
async def test_additional_span_attributes(self):
|
||||
"""Test that additional span attributes are added to the conversation span."""
|
||||
extra_attrs = {"deployment.id": "abc-123", "customer.tier": "premium"}
|
||||
tracing_ctx = TracingContext()
|
||||
turn_tracker = TurnTrackingObserver(turn_end_timeout_secs=0.2)
|
||||
latency_tracker = UserBotLatencyObserver()
|
||||
trace_observer = TurnTraceObserver(
|
||||
turn_tracker,
|
||||
latency_tracker=latency_tracker,
|
||||
additional_span_attributes=extra_attrs,
|
||||
tracing_context=tracing_ctx,
|
||||
)
|
||||
trace_observer._tracer = self._tracer
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=[turn_tracker, latency_tracker, trace_observer],
|
||||
)
|
||||
|
||||
# End conversation to flush the conversation span
|
||||
trace_observer.end_conversation_tracing()
|
||||
|
||||
conv_spans = self._get_spans_by_name("conversation")
|
||||
self.assertEqual(len(conv_spans), 1)
|
||||
self.assertEqual(conv_spans[0].attributes["deployment.id"], "abc-123")
|
||||
self.assertEqual(conv_spans[0].attributes["customer.tier"], "premium")
|
||||
|
||||
async def test_concurrent_pipelines_are_isolated(self):
|
||||
"""Test that two pipelines with separate TracingContexts don't interfere."""
|
||||
tracing_ctx_a = TracingContext()
|
||||
tracing_ctx_b = TracingContext()
|
||||
|
||||
_, _, trace_observer_a, _ = self._create_observers(
|
||||
conversation_id="conv-a", tracing_context=tracing_ctx_a
|
||||
)
|
||||
_, _, trace_observer_b, _ = self._create_observers(
|
||||
conversation_id="conv-b", tracing_context=tracing_ctx_b
|
||||
)
|
||||
|
||||
processor_a = IdentityFilter()
|
||||
processor_b = IdentityFilter()
|
||||
|
||||
frames = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
# Run both pipelines concurrently
|
||||
await asyncio.gather(
|
||||
run_test(
|
||||
processor_a,
|
||||
frames_to_send=frames,
|
||||
expected_down_frames=expected,
|
||||
observers=self._all_observers(trace_observer_a),
|
||||
),
|
||||
run_test(
|
||||
processor_b,
|
||||
frames_to_send=frames,
|
||||
expected_down_frames=expected,
|
||||
observers=self._all_observers(trace_observer_b),
|
||||
),
|
||||
)
|
||||
|
||||
# End both conversations to flush spans
|
||||
trace_observer_a.end_conversation_tracing()
|
||||
trace_observer_b.end_conversation_tracing()
|
||||
|
||||
# Each TracingContext should have its own conversation ID
|
||||
conv_spans = self._get_spans_by_name("conversation")
|
||||
conv_ids = {s.attributes["conversation.id"] for s in conv_spans}
|
||||
self.assertEqual(conv_ids, {"conv-a", "conv-b"})
|
||||
|
||||
# Turn spans should be children of their own conversation span, not cross-linked
|
||||
turn_spans = self._get_spans_by_name("turn")
|
||||
conv_span_map = {s.context.span_id: s.attributes["conversation.id"] for s in conv_spans}
|
||||
for turn_span in turn_spans:
|
||||
parent_id = turn_span.parent.span_id
|
||||
turn_conv_id = turn_span.attributes["conversation.id"]
|
||||
parent_conv_id = conv_span_map[parent_id]
|
||||
self.assertEqual(
|
||||
turn_conv_id,
|
||||
parent_conv_id,
|
||||
f"Turn span for {turn_conv_id} parented under {parent_conv_id}",
|
||||
)
|
||||
|
||||
async def test_end_conversation_closes_active_turn(self):
|
||||
"""Test that end_conversation_tracing closes any active turn span."""
|
||||
_, _, trace_observer, _ = self._create_observers()
|
||||
|
||||
# Manually start conversation and a turn
|
||||
trace_observer.start_conversation_tracing("conv-end-test")
|
||||
await trace_observer._handle_turn_started(1)
|
||||
|
||||
self.assertIsNotNone(trace_observer._current_span)
|
||||
self.assertIsNotNone(trace_observer._conversation_span)
|
||||
|
||||
# End conversation — should close both turn and conversation
|
||||
trace_observer.end_conversation_tracing()
|
||||
|
||||
self.assertIsNone(trace_observer._current_span)
|
||||
self.assertIsNone(trace_observer._conversation_span)
|
||||
|
||||
# Check span attributes
|
||||
turn_spans = self._get_spans_by_name("turn")
|
||||
self.assertEqual(len(turn_spans), 1)
|
||||
self.assertTrue(turn_spans[0].attributes["turn.was_interrupted"])
|
||||
self.assertTrue(turn_spans[0].attributes["turn.ended_by_conversation_end"])
|
||||
|
||||
async def test_conversation_id_auto_generated(self):
|
||||
"""Test that a conversation ID is auto-generated when none is provided."""
|
||||
_, _, trace_observer, _ = self._create_observers(conversation_id=None)
|
||||
processor = IdentityFilter()
|
||||
|
||||
frames_to_send = [
|
||||
UserStartedSpeakingFrame(),
|
||||
UserStoppedSpeakingFrame(),
|
||||
BotStartedSpeakingFrame(),
|
||||
BotStoppedSpeakingFrame(),
|
||||
SleepFrame(sleep=0.4),
|
||||
]
|
||||
|
||||
expected_down_frames = [
|
||||
UserStartedSpeakingFrame,
|
||||
UserStoppedSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
]
|
||||
|
||||
await run_test(
|
||||
processor,
|
||||
frames_to_send=frames_to_send,
|
||||
expected_down_frames=expected_down_frames,
|
||||
observers=self._all_observers(trace_observer),
|
||||
)
|
||||
|
||||
# End conversation to flush the conversation span
|
||||
trace_observer.end_conversation_tracing()
|
||||
|
||||
conv_spans = self._get_spans_by_name("conversation")
|
||||
self.assertEqual(len(conv_spans), 1)
|
||||
# Should have an auto-generated UUID as conversation.id
|
||||
conv_id = conv_spans[0].attributes["conversation.id"]
|
||||
self.assertIsNotNone(conv_id)
|
||||
self.assertGreater(len(conv_id), 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -6,12 +6,14 @@
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
import unittest.mock
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
BotSpeakingFrame,
|
||||
BotStartedSpeakingFrame,
|
||||
BotStoppedSpeakingFrame,
|
||||
FunctionCallResultFrame,
|
||||
FunctionCallsStartedFrame,
|
||||
UserSpeakingFrame,
|
||||
UserIdleTimeoutUpdateFrame,
|
||||
UserStartedSpeakingFrame,
|
||||
)
|
||||
from pipecat.turns.user_idle_controller import UserIdleController
|
||||
@@ -25,8 +27,8 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
self.task_manager = TaskManager()
|
||||
self.task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop()))
|
||||
|
||||
async def test_basic_idle_detection(self):
|
||||
"""Test that idle event is triggered after timeout when no activity."""
|
||||
async def test_idle_after_bot_stops_speaking(self):
|
||||
"""Test that idle event fires after BotStoppedSpeakingFrame + timeout."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -37,18 +39,16 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# Wait for idle timeout
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertTrue(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_user_speaking_resets_idle_timer(self):
|
||||
"""Test that continuous UserSpeakingFrame frames reset the idle timer."""
|
||||
async def test_user_speaking_cancels_timer(self):
|
||||
"""Test that UserStartedSpeakingFrame cancels the idle timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -59,20 +59,18 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
|
||||
# Send UserSpeakingFrame continuously to reset timer
|
||||
for _ in range(5):
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.5) # 50% of timeout period
|
||||
await controller.process_frame(UserSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_bot_speaking_resets_idle_timer(self):
|
||||
"""Test that BotSpeakingFrame frames reset the idle timer."""
|
||||
async def test_bot_speaking_cancels_timer(self):
|
||||
"""Test that BotStartedSpeakingFrame cancels the idle timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -83,102 +81,61 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_no_idle_before_bot_speaks(self):
|
||||
"""Test that idle does not fire if no BotStoppedSpeakingFrame is received."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Wait without any frames
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_interruption_no_false_trigger(self):
|
||||
"""Test that BotStoppedSpeakingFrame during a user turn does not start the timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# User starts speaking (interruption)
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
# Bot stops speaking due to interruption
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# Bot speaking should reset timer
|
||||
for _ in range(5):
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.6) # 60% of timeout
|
||||
await controller.process_frame(BotSpeakingFrame())
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_function_call_prevents_idle(self):
|
||||
"""Test that function calls in progress prevent idle event."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
|
||||
# Start function call
|
||||
await controller.process_frame(FunctionCallsStartedFrame(function_calls=[]))
|
||||
|
||||
# Wait longer than idle timeout
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
# Should not trigger idle because function call is in progress
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
# Complete function call
|
||||
await controller.process_frame(
|
||||
FunctionCallResultFrame(
|
||||
function_name="test",
|
||||
tool_call_id="123",
|
||||
arguments={},
|
||||
result=None,
|
||||
run_llm=False,
|
||||
)
|
||||
)
|
||||
|
||||
# Now idle should trigger
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertTrue(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_no_idle_before_conversation_starts(self):
|
||||
"""Test that idle monitoring doesn't start before first conversation activity."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Wait without starting conversation
|
||||
# Wait - timer should NOT have started because user turn is in progress
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_idle_starts_with_bot_speaking(self):
|
||||
"""Test that monitoring starts with BotSpeakingFrame, not just user speech."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start conversation with bot speaking
|
||||
await controller.process_frame(BotSpeakingFrame())
|
||||
|
||||
# Wait for idle timeout
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertTrue(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_multiple_idle_events(self):
|
||||
"""Test that idle event can trigger multiple times."""
|
||||
async def test_idle_cycle(self):
|
||||
"""Test that idle fires, then can fire again after another bot speaking cycle."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
@@ -189,29 +146,175 @@ class TestUserIdleController(unittest.IsolatedAsyncioTestCase):
|
||||
nonlocal idle_count
|
||||
idle_count += 1
|
||||
|
||||
# Start conversation
|
||||
await controller.process_frame(UserStartedSpeakingFrame())
|
||||
|
||||
# First idle
|
||||
# First cycle: bot stops → idle fires
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
first_count = idle_count
|
||||
self.assertGreaterEqual(first_count, 1)
|
||||
self.assertEqual(idle_count, 1)
|
||||
|
||||
# Second idle
|
||||
# Second cycle: bot starts → bot stops → idle fires again
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
second_count = idle_count
|
||||
self.assertGreater(second_count, first_count)
|
||||
self.assertEqual(idle_count, 2)
|
||||
|
||||
# User activity resets timer
|
||||
await controller.process_frame(UserSpeakingFrame())
|
||||
await controller.cleanup()
|
||||
|
||||
# Give a moment for the timer to reset
|
||||
await asyncio.sleep(0.1)
|
||||
async def test_cleanup_cancels_timer(self):
|
||||
"""Test that cleanup cancels a pending idle timer."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
await controller.cleanup()
|
||||
|
||||
# Third idle
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
third_count = idle_count
|
||||
self.assertGreater(third_count, second_count)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
async def test_function_call_cancels_timer(self):
|
||||
"""Test normal ordering: BotStopped starts timer, FunctionCallsStarted cancels it."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Bot finishes speaking, timer starts
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
# Function call starts shortly after, cancels the timer
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
await controller.process_frame(
|
||||
FunctionCallsStartedFrame(function_calls=[unittest.mock.Mock()])
|
||||
)
|
||||
|
||||
# Wait longer than timeout — should not fire
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_function_call_suppresses_timer(self):
|
||||
"""Test race condition: FunctionCallsStarted arrives before BotStopped.
|
||||
|
||||
A race condition can cause FunctionCallsStarted to arrive before
|
||||
BotStoppedSpeaking. The counter guard prevents the timer from starting
|
||||
while a function call is in progress.
|
||||
"""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# LLM emits function call and "let me check" concurrently
|
||||
await controller.process_frame(
|
||||
FunctionCallsStartedFrame(function_calls=[unittest.mock.Mock()])
|
||||
)
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# Wait longer than timeout — should not fire (function call in progress)
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
# Function call completes, bot speaks result
|
||||
await controller.process_frame(
|
||||
FunctionCallResultFrame(
|
||||
function_name="test", tool_call_id="123", arguments={}, result="ok"
|
||||
)
|
||||
)
|
||||
await controller.process_frame(BotStartedSpeakingFrame())
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
|
||||
# Now the timer should start and fire
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertTrue(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_disabled_by_default(self):
|
||||
"""Test that timeout=0 means idle detection is disabled."""
|
||||
controller = UserIdleController()
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_enable_via_frame(self):
|
||||
"""Test enabling idle detection at runtime via UserIdleTimeoutUpdateFrame."""
|
||||
controller = UserIdleController()
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Initially disabled — no idle fires
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
# Enable idle detection
|
||||
await controller.process_frame(UserIdleTimeoutUpdateFrame(timeout=USER_IDLE_TIMEOUT))
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertTrue(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
async def test_disable_via_frame(self):
|
||||
"""Test disabling idle detection at runtime via UserIdleTimeoutUpdateFrame."""
|
||||
controller = UserIdleController(user_idle_timeout=USER_IDLE_TIMEOUT)
|
||||
await controller.setup(self.task_manager)
|
||||
|
||||
idle_triggered = False
|
||||
|
||||
@controller.event_handler("on_user_turn_idle")
|
||||
async def on_user_turn_idle(controller):
|
||||
nonlocal idle_triggered
|
||||
idle_triggered = True
|
||||
|
||||
# Start the timer
|
||||
await controller.process_frame(BotStoppedSpeakingFrame())
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT * 0.3)
|
||||
|
||||
# Disable — should cancel running timer
|
||||
await controller.process_frame(UserIdleTimeoutUpdateFrame(timeout=0))
|
||||
|
||||
await asyncio.sleep(USER_IDLE_TIMEOUT + 0.1)
|
||||
|
||||
self.assertFalse(idle_triggered)
|
||||
|
||||
await controller.cleanup()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user