Stop analysis while the collector is disabled

node/src/analyzer.ts

@@ -117,8 +117,11 @@ export class Collector extends FrameProcessor<AudioFrame> {
     this.collectingEnabled = enabled;
   }
 
+  /** True while the collector has collected some audio the analyzer can act on. */
   get initialized(): boolean {
-    return this.hasBufferedAudio && this.nativeCollector !== null;
+    return (
+      this.collectingEnabled && this.hasBufferedAudio && this.nativeCollector !== null
+    );
   }
 
   get currentStreamInfo(): FrameProcessorStreamInfo | null {

node/test/analyzer.test.ts

@@ -223,6 +223,42 @@ describe("Analyzer", () => {
     analyzer.close();
   });
 
+  it("pauses analysis while the collector is disabled", () => {
+    const analyzer = new Analyzer({
+      model: { getId: () => "analysis-test-model" } as never,
+      licenseKey: "test-license",
+      analysisInterval: 0.01,
+    });
+    const events: AnalysisEvent[] = [];
+    analyzer.on("analysisResult", (event) => events.push(event));
+    analyzer.collector.process(makeFrame());
+
+    vi.advanceTimersByTime(10);
+    expect(sdk.analyzers[0]!.analyzeCalls).toBe(1);
+
+    analyzer.collector.setEnabled(false);
+    const frame = makeFrame();
+    expect(analyzer.collector.process(frame)).toBe(frame);
+    vi.advanceTimersByTime(50);
+
+    expect(sdk.collectors[0]!.blocks).toHaveLength(1);
+    expect(sdk.analyzers[0]!.analyzeCalls).toBe(1);
+    expect(events).toHaveLength(1);
+
+    // Re-enabling drops the stale buffer and waits for fresh audio.
+    analyzer.collector.setEnabled(true);
+    expect(sdk.analyzers[0]!.resetCalls).toBe(1);
+    vi.advanceTimersByTime(50);
+    expect(sdk.analyzers[0]!.analyzeCalls).toBe(1);
+
+    analyzer.collector.process(makeFrame());
+    vi.advanceTimersByTime(10);
+    expect(sdk.analyzers[0]!.analyzeCalls).toBe(2);
+    expect(events).toHaveLength(2);
+
+    analyzer.close();
+  });
+
   it("stops the analyzer when RoomIO closes its collector", () => {
     const analyzer = new Analyzer({
       model: { getId: () => "analysis-test-model" } as never,

python/src/livekit/plugins/ai_coustics/analyzer.py

@@ -94,7 +94,9 @@ def enabled(self, value: bool) -> None:
 
     @property
     def initialized(self) -> bool:
-        return self._has_buffered_audio and self._collector is not None
+        """True while the collector is feeding fresh audio the analyzer can act on."""
+
+        return self._enabled and self._has_buffered_audio and self._collector is not None
 
     @property
     def stream_info(self) -> dict[str, str]:

python/tests/test_analyzer.py

@@ -267,6 +267,56 @@ async def test_records_failed_analysis_metrics(
     assert not metric_instruments["score"].measurements
 
 
+@pytest.mark.asyncio
+async def test_disabled_collector_pauses_analysis(
+    fake_sdk: tuple[FakeCollector, FakeNativeAnalyzer, list[int]],
+) -> None:
+    native_collector, native_analyzer, _ = fake_sdk
+    analyzer = Analyzer(
+        model=cast(aic_sdk.Model, FakeModel()),
+        license_key="test-license",
+        analysis_interval=0.01,
+    )
+    events: list[AnalysisEvent] = []
+    analyzer.on("analysis_result", events.append)
+    analyzer.collector._process(make_frame())
+
+    for _ in range(20):
+        if native_analyzer.analyze_calls:
+            break
+        await asyncio.sleep(0.005)
+    assert native_analyzer.analyze_calls >= 1
+
+    analyzer.collector.enabled = False
+    frame = make_frame()
+    assert analyzer.collector._process(frame) is frame
+    assert len(native_collector.blocks) == 1
+
+    # Let an analysis that was already in flight drain, then confirm no new ones start.
+    await asyncio.sleep(0.03)
+    calls_while_disabled = native_analyzer.analyze_calls
+    events_while_disabled = len(events)
+    await asyncio.sleep(0.06)
+    assert native_analyzer.analyze_calls == calls_while_disabled
+    assert len(events) == events_while_disabled
+
+    # Re-enabling drops the stale buffer and waits for fresh audio.
+    analyzer.collector.enabled = True
+    assert native_analyzer.reset_calls == 1
+    await asyncio.sleep(0.06)
+    assert native_analyzer.analyze_calls == calls_while_disabled
+
+    analyzer.collector._process(make_frame())
+    for _ in range(20):
+        if native_analyzer.analyze_calls > calls_while_disabled:
+            break
+        await asyncio.sleep(0.005)
+
+    await asyncio.wait_for(analyzer.aclose(), timeout=1.0)
+    assert native_analyzer.analyze_calls > calls_while_disabled
+    assert len(events) > events_while_disabled
+
+
 @pytest.mark.asyncio
 async def test_room_closing_collector_stops_analyzer(
     fake_sdk: tuple[FakeCollector, FakeNativeAnalyzer, list[int]],