Merge pull request #3889 from ai-coustics/goedev/aic-voice-focus-and-memoryview-fix

AIC Voice Focus version update & concurrency safety issue on audio buffer.

changelog/3889.changed.md

@@ -0,0 +1,3 @@
+- Support for Voice Focus 2.0 models.
+  - Updated `aic-sdk` to `~=2.1.0` to support Voice Focus 2.0 models.
+  - Cleaned unused `ParameterFixedError` exception handling in `AICFilter` parameter setup. 

changelog/3889.fixed.md

@@ -0,0 +1 @@
+- Fixed `BufferError: Existing exports of data: object cannot be re-sized` in `AICFilter` caused by holding a `memoryview` on the mutable audio buffer across async yield points.

examples/foundational/07zd-interruptible-aicoustics.py

@@ -40,7 +40,7 @@ def _create_aic_filter() -> AICFilter:
 
     return AICFilter(
         license_key=license_key,
-        model_id="quail-vf-l-16khz",
+        model_id="quail-vf-2.0-l-16khz",
     )
 
 

pyproject.toml

@@ -52,7 +52,7 @@ Issues = "https://github.com/pipecat-ai/pipecat/issues"
 Changelog = "https://github.com/pipecat-ai/pipecat/blob/main/CHANGELOG.md"
 
 [project.optional-dependencies]
-aic = [ "aic-sdk~=2.0.1" ]
+aic = [ "aic-sdk~=2.1.0" ]
 anthropic = [ "anthropic~=0.49.0" ]
 assemblyai = [ "pipecat-ai[websockets-base]" ]
 asyncai = [ "pipecat-ai[websockets-base]" ]

src/pipecat/audio/filters/aic_filter.py

@@ -375,12 +375,7 @@ class AICFilter(BaseAudioFilter):
         self._vad_ctx = self._processor.get_vad_context()
 
         # Apply initial parameters
-        try:
-            self._processor_ctx.set_parameter(
-                ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0
-            )
-        except ParameterFixedError as e:
-            logger.error(f"AIC parameter update failed: {e}")
+        self._processor_ctx.set_parameter(ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0)
 
         # Log processor information
         logger.debug(f"ai-coustics filter started:")
@@ -389,7 +384,8 @@ class AICFilter(BaseAudioFilter):
         logger.debug(f"  Frames per chunk: {self._frames_per_block}")
         logger.debug(f"  Optimal sample rate: {self._model.get_optimal_sample_rate()} Hz")
         logger.debug(
-            f"  Optimal number of frames for {self._sample_rate} Hz: {self._model.get_optimal_num_frames(self._sample_rate)}"
+            f"  Optimal number of frames for {self._sample_rate} Hz: "
+            f"{self._model.get_optimal_num_frames(self._sample_rate)}"
         )
         logger.debug(
             f"  Output delay: {self._processor_ctx.get_output_delay()} samples "
@@ -458,13 +454,16 @@ class AICFilter(BaseAudioFilter):
         if num_blocks == 0:
             return b""
 
-        filtered_chunks: List[bytes] = []
-        mv = memoryview(self._audio_buffer)
         block_size = self._frames_per_block * self._bytes_per_sample
+        total_size = num_blocks * block_size
+        blocks_data = bytes(self._audio_buffer[:total_size])
+        self._audio_buffer = self._audio_buffer[total_size:]
+
+        filtered_chunks: List[bytes] = []
 
         for i in range(num_blocks):
             start = i * block_size
-            block_i16 = np.frombuffer(mv[start : start + block_size], dtype=self._dtype)
+            block_i16 = np.frombuffer(blocks_data[start : start + block_size], dtype=self._dtype)
 
             # Reuse input buffer, in-place divide
             np.copyto(self._in_f32[0], block_i16)
@@ -479,5 +478,4 @@ class AICFilter(BaseAudioFilter):
 
             filtered_chunks.append(self._out_i16.tobytes())
 
-        self._audio_buffer = self._audio_buffer[num_blocks * block_size :]
         return b"".join(filtered_chunks)

tests/test_aic_filter.py

@@ -632,6 +632,84 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase):
         for result in results:
             self.assertIsInstance(result, bytes)
 
+    async def test_concurrent_filter_no_buffer_resize_error(self):
+        """Regression: concurrent filter() must not raise BufferError.
+
+        When process_async yields to the event loop, a second filter() call
+        runs and calls _audio_buffer.extend().  If the first call still holds
+        a memoryview on the bytearray, extend() raises:
+
+            BufferError: Existing exports of data: object cannot be re-sized
+
+        The fix snapshots the needed data into immutable bytes and trims the
+        buffer *before* any await, so no memoryview is held across yield
+        points.
+        """
+        filter_instance = self._create_filter_with_mocks()
+
+        # Make process_async yield to the event loop so concurrent filter()
+        # calls can interleave and attempt _audio_buffer.extend().
+        async def yielding_process_async(audio_array):
+            await asyncio.sleep(0)
+            return audio_array.copy()
+
+        self.mock_processor.process_async = yielding_process_async
+        await self._start_filter_with_mocks(filter_instance)
+
+        samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16)
+        input_audio = samples.tobytes()
+
+        async def filter_audio():
+            return await filter_instance.filter(input_audio)
+
+        # 20 concurrent calls to reliably trigger the interleaving.
+        tasks = [filter_audio() for _ in range(20)]
+        results = await asyncio.gather(*tasks)
+
+        for result in results:
+            self.assertIsInstance(result, bytes)
+
+    async def test_stop_during_filter_no_buffer_resize_error(self):
+        """Regression: stop() during filter() must not raise BufferError.
+
+        When filter() holds a memoryview on _audio_buffer across an await
+        (process_async), a concurrent stop() that calls
+        _audio_buffer.clear() raises:
+
+            BufferError: Existing exports of data: object cannot be re-sized
+
+        This is the exact error path reported in production (line 414).
+        The fix removes the memoryview by snapshotting data into immutable
+        bytes before any await.
+        """
+        filter_instance = self._create_filter_with_mocks()
+
+        # Gate so stop() waits until filter() is inside process_async.
+        processing_started = asyncio.Event()
+
+        async def yielding_process_async(audio_array):
+            processing_started.set()
+            await asyncio.sleep(0)  # yield — stop() runs here
+            return audio_array.copy()
+
+        self.mock_processor.process_async = yielding_process_async
+        await self._start_filter_with_mocks(filter_instance)
+
+        # Exactly one complete frame so the loop runs once.
+        samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16)
+        input_audio = samples.tobytes()
+
+        async def stop_after_filter_enters_process_async():
+            await processing_started.wait()
+            await filter_instance.stop()
+
+        # filter() enters process_async → yields → stop() calls clear()
+        filter_result, _ = await asyncio.gather(
+            filter_instance.filter(input_audio),
+            stop_after_filter_enters_process_async(),
+        )
+        self.assertIsInstance(filter_result, bytes)
+
     async def test_buffer_cleared_on_stop(self):
         """Test that audio buffer is cleared when stopping."""
         filter_instance = self._create_filter_with_mocks()