From f0c902038c0a3867b2bec78e1dd09c5e2a336473 Mon Sep 17 00:00:00 2001 From: Matthias Geier Date: Mon, 14 Sep 2026 15:32:13 +0200 Subject: [PATCH 1/2] Stop analysis while the collector is disabled Collector.setEnabled(False) only stopped buffering; the analysis loop gated on `initialized`, which stayed true because neither the buffered audio flag nor the native collector handle changed on disable. The analyzer kept running inference on the frozen buffer every interval, emitting analysis results with incrementing sequence numbers and recording metrics for audio that was no longer being collected. Gate `initialized` on the enabled flag in both runtimes. Re-enabling already resets the analyzer and drops the stale buffer, matching how Processor and VAD handle their own enable toggles, so analysis resumes only once fresh audio arrives. Co-Authored-By: Claude Opus 5 (1M context) --- node/src/analyzer.ts | 5 +- node/test/analyzer.test.ts | 36 +++++++++++++ .../livekit/plugins/ai_coustics/analyzer.py | 4 +- python/tests/test_analyzer.py | 50 +++++++++++++++++++ 4 files changed, 93 insertions(+), 2 deletions(-) diff --git a/node/src/analyzer.ts b/node/src/analyzer.ts index 8b40be1..ced5a37 100644 --- a/node/src/analyzer.ts +++ b/node/src/analyzer.ts @@ -117,8 +117,11 @@ export class Collector extends FrameProcessor { this.collectingEnabled = enabled; } + /** True while the collector is feeding fresh 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 { diff --git a/node/test/analyzer.test.ts b/node/test/analyzer.test.ts index 03cde46..42e0d08 100644 --- a/node/test/analyzer.test.ts +++ b/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, diff --git a/python/src/livekit/plugins/ai_coustics/analyzer.py b/python/src/livekit/plugins/ai_coustics/analyzer.py index d225130..094bebd 100644 --- a/python/src/livekit/plugins/ai_coustics/analyzer.py +++ b/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]: diff --git a/python/tests/test_analyzer.py b/python/tests/test_analyzer.py index 94e3f33..f759f8d 100644 --- a/python/tests/test_analyzer.py +++ b/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]], From 748fa98d69ae28b740e43686c05c08430eb50b36 Mon Sep 17 00:00:00 2001 From: Matthias Geier Date: Mon, 14 Sep 2026 16:48:19 +0200 Subject: [PATCH 2/2] Apply batched suggestions from code review Co-authored-by: Matthias Geier --- node/src/analyzer.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/node/src/analyzer.ts b/node/src/analyzer.ts index ced5a37..5a470e5 100644 --- a/node/src/analyzer.ts +++ b/node/src/analyzer.ts @@ -117,7 +117,7 @@ export class Collector extends FrameProcessor { this.collectingEnabled = enabled; } - /** True while the collector is feeding fresh audio the analyzer can act on. */ + /** True while the collector has collected some audio the analyzer can act on. */ get initialized(): boolean { return ( this.collectingEnabled && this.hasBufferedAudio && this.nativeCollector !== null