FunASR/tests/test_realtime_ws_service.py

import asyncio
import importlib.util
import json
import subprocess
import sys
import threading
import time
import types
from pathlib import Path

import numpy as np
import pytest
import torch


REPO_ROOT = Path(__file__).resolve().parents[1]
EXAMPLE_SERVICE_PATH = (
    REPO_ROOT / "examples" / "industrial_data_pretraining" / "fun_asr_nano" / "serve_realtime_ws.py"
)
PACKAGE_SERVICE_PATH = REPO_ROOT / "funasr" / "bin" / "realtime_ws.py"
SERVICE_PATH = PACKAGE_SERVICE_PATH


def test_realtime_ws_service_is_packaged():
    assert PACKAGE_SERVICE_PATH.is_file()


def test_example_realtime_entrypoint_delegates_to_packaged_cli():
    result = subprocess.run(
        [sys.executable, str(EXAMPLE_SERVICE_PATH), "--help"],
        capture_output=True,
        text=True,
        check=False,
    )

    assert result.returncode == 0, result.stderr
    assert "--endpoint-mode" in result.stdout


def load_service_module():
    websockets_stub = types.ModuleType("websockets")
    websockets_stub.serve = lambda *args, **kwargs: None
    exceptions_stub = types.ModuleType("websockets.exceptions")
    exceptions_stub.ConnectionClosed = type("ConnectionClosed", (Exception,), {})

    previous_websockets = sys.modules.get("websockets")
    previous_exceptions = sys.modules.get("websockets.exceptions")
    sys.modules["websockets"] = websockets_stub
    sys.modules["websockets.exceptions"] = exceptions_stub

    module_name = "serve_realtime_ws_under_test"
    sys.modules.pop(module_name, None)
    spec = importlib.util.spec_from_file_location(module_name, SERVICE_PATH)
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    try:
        spec.loader.exec_module(module)
    finally:
        if previous_websockets is None:
            sys.modules.pop("websockets", None)
        else:
            sys.modules["websockets"] = previous_websockets
        if previous_exceptions is None:
            sys.modules.pop("websockets.exceptions", None)
        else:
            sys.modules["websockets.exceptions"] = previous_exceptions
    return module


def test_handler_catches_unexpected_errors_without_top_level_exceptions(
    monkeypatch, caplog
):
    module = load_service_module()

    class FailingWebSocket:
        remote_address = ("127.0.0.1", 12345)

        def __aiter__(self):
            return self

        async def __anext__(self):
            raise RuntimeError("boom")

    class Session:
        def __init__(self, *args, **kwargs):
            pass

    monkeypatch.setattr(
        module, "load_models", lambda args: (object(), {}, object(), None)
    )
    monkeypatch.setattr(module, "DynamicStreamingVAD", lambda model: object())
    monkeypatch.setattr(
        module, "create_speaker_tracker", lambda model, args: None
    )
    monkeypatch.setattr(module, "RealtimeASRSession", Session)
    args = types.SimpleNamespace(
        device="cpu",
        decode_interval=0.48,
        partial_window_sec=15.0,
    )

    asyncio.run(module.handle_client(FailingWebSocket(), args))

    assert "boom" in caplog.text


def test_cli_defaults_disable_speaker_and_bound_partial_window():
    module = load_service_module()

    assert hasattr(module, "build_arg_parser")
    args = module.build_arg_parser().parse_args([])

    assert args.enable_spk is False
    assert args.partial_window_sec == 8.0
    assert args.endpoint_mode == "server"
    assert args.log_session_stats_interval == 0.0
    assert args.decode_batch_wait_ms == 10.0
    assert args.decode_max_batch_size == 16
    assert args.log_decode_profile is False
    assert args.vad_device == "cpu"
    assert args.vad_ncpu == 1


def test_cli_accepts_client_endpoint_mode():
    module = load_service_module()

    args = module.build_arg_parser().parse_args(["--endpoint-mode", "client"])

    assert args.endpoint_mode == "client"


def test_cli_accepts_float32_dtype_alias():
    module = load_service_module()

    args = module.build_arg_parser().parse_args(["--dtype", "float32"])

    assert args.dtype == "fp32"


def test_websocket_keepalive_kwargs_are_configurable():
    module = load_service_module()

    args = module.build_arg_parser().parse_args(
        [
            "--ws-ping-interval", "35",
            "--ws-ping-timeout", "70",
            "--ws-close-timeout", "12",
            "--ws-max-size", "12345",
        ]
    )

    assert module.build_websocket_serve_kwargs(args) == {
        "max_size": 12345,
        "ping_interval": 35.0,
        "ping_timeout": 70.0,
        "close_timeout": 12.0,
    }


def test_websocket_keepalive_kwargs_can_disable_ping():
    module = load_service_module()

    args = module.build_arg_parser().parse_args(
        [
            "--ws-ping-interval", "0",
            "--ws-ping-timeout", "0",
        ]
    )

    kwargs = module.build_websocket_serve_kwargs(args)
    assert kwargs["ping_interval"] is None
    assert kwargs["ping_timeout"] is None


def test_websocket_defaults_keep_pings_without_a_compute_timeout():
    module = load_service_module()

    args = module.build_arg_parser().parse_args([])

    kwargs = module.build_websocket_serve_kwargs(args)
    assert kwargs["ping_interval"] == 20.0
    assert kwargs["ping_timeout"] is None


def test_cli_accepts_session_stats_interval():
    module = load_service_module()

    args = module.build_arg_parser().parse_args(
        ["--log-session-stats-interval", "30"]
    )

    assert args.log_session_stats_interval == 30.0


def test_create_speaker_tracker_skips_spk_when_disabled():
    module = load_service_module()
    args = types.SimpleNamespace(enable_spk=False, device="cuda:0")

    assert module.create_speaker_tracker(object(), args) is None


def test_partial_decode_audio_uses_recent_window():
    module = load_service_module()

    class DummyVad:
        current_speech_start = 0

        def reset(self):
            pass

    sample_rate = 16000
    session = module.RealtimeASRSession(
        vllm_engine=object(),
        asr_kwargs={},
        vad=DummyVad(),
        spk_tracker=None,
        sample_rate=sample_rate,
        chunk_ms=960,
        partial_window_sec=2.0,
    )
    session.audio_buffer = np.arange(sample_rate * 5, dtype=np.float32)
    session.total_samples = len(session.audio_buffer)

    audio, start_ms = session.get_partial_decode_audio()

    assert start_ms == 3000
    np.testing.assert_array_equal(audio, session.audio_buffer[-sample_rate * 2:])


def test_partial_decode_audio_keeps_short_current_segment():
    module = load_service_module()

    class DummyVad:
        current_speech_start = 1500

        def reset(self):
            pass

    sample_rate = 16000
    session = module.RealtimeASRSession(
        vllm_engine=object(),
        asr_kwargs={},
        vad=DummyVad(),
        spk_tracker=None,
        sample_rate=sample_rate,
        chunk_ms=960,
        partial_window_sec=10.0,
    )
    session.audio_buffer = np.arange(sample_rate * 5, dtype=np.float32)
    session.total_samples = len(session.audio_buffer)

    audio, start_ms = session.get_partial_decode_audio()
    assert start_ms == 1500
    np.testing.assert_array_equal(audio, session.audio_buffer[int(sample_rate * 1.5):])


def test_empty_partial_response_preserves_zero_speech_start():
    module = load_service_module()

    class DummyVad:
        current_speech_start = 0

        def reset(self):
            pass

    sample_rate = 16000
    session = module.RealtimeASRSession(
        vllm_engine=object(),
        asr_kwargs={},
        vad=DummyVad(),
        spk_tracker=None,
        sample_rate=sample_rate,
        chunk_ms=960,
    )
    session.audio_buffer = np.zeros(sample_rate, dtype=np.float32)
    session.total_samples = len(session.audio_buffer)
    session.last_partial_text = ""

    response = session._build_response(is_final=False)

    assert response["partial_start_ms"] == 0


class SilentVad:
    current_speech_start = None

    def feed(self, audio, is_final=False):
        return []

    def reset(self):
        pass


class SegmentVad(SilentVad):
    def __init__(self, sample_rate, segment_end_sample):
        self.sample_rate = sample_rate
        self.segment_end_sample = segment_end_sample
        self.total_samples = 0

    def feed(self, audio, is_final=False):
        self.total_samples += len(audio)
        if self.total_samples == self.segment_end_sample:
            end_ms = int(self.total_samples * 1000 / self.sample_rate)
            return [[end_ms - 1000, end_ms]]
        return []


class DummyTokenizer:
    def encode(self, text):
        return []

    def decode(self, token_ids, skip_special_tokens=True):
        return ""


class DummyEngine:
    def __init__(self):
        self._engine = types.SimpleNamespace(tokenizer=DummyTokenizer())
        self.input_lengths = []
        self.generate_kwargs = []

    def generate(self, inputs, **kwargs):
        self.input_lengths.extend(len(audio) for audio in inputs)
        self.generate_kwargs.append(kwargs)
        return [{"text": "hello"}]


class FixedTextEngine(DummyEngine):
    def __init__(self, text):
        super().__init__()
        self.text = text

    def generate(self, inputs, **kwargs):
        self.input_lengths.extend(len(audio) for audio in inputs)
        self.generate_kwargs.append(kwargs)
        return [{"text": self.text}]


class FailingTextEngine(DummyEngine):
    def generate(self, inputs, **kwargs):
        raise RuntimeError("synthetic final decode failure")


class ScriptedTextEngine(DummyEngine):
    def __init__(self, outputs):
        super().__init__()
        self.outputs = list(outputs)

    def generate(self, inputs, **kwargs):
        self.input_lengths.extend(len(audio) for audio in inputs)
        self.generate_kwargs.append(kwargs)
        output = self.outputs.pop(0)
        if isinstance(output, Exception):
            raise output
        return [{"text": output}]


class ControllableVad:
    def __init__(self, sample_rate=16000):
        self.sample_rate = sample_rate
        self.current_speech_start = 0
        self.total_samples = 0
        self.lock_next = False

    def feed(self, audio, is_final=False):
        self.total_samples += len(audio)
        if not self.lock_next:
            return []
        self.lock_next = False
        self.current_speech_start = None
        return [[0, int(self.total_samples * 1000 / self.sample_rate)]]

    def start_utterance(self, start_ms):
        if self.current_speech_start is None:
            self.current_speech_start = start_ms

    def reset(self):
        self.current_speech_start = None


class PreviousBoundaryVad(ControllableVad):
    def feed(self, audio, is_final=False):
        self.total_samples += len(audio)
        if not self.lock_next:
            return []
        self.lock_next = False
        self.current_speech_start = None
        end_samples = self.total_samples - len(audio)
        return [[0, int(end_samples * 1000 / self.sample_rate)]]


class FinalFlushVad:
    def __init__(self, sample_rate=16000):
        self.sample_rate = sample_rate
        self.current_speech_start = None
        self.total_samples = 0
        self.finalize_calls = 0

    def feed(self, audio, is_final=False):
        assert is_final is False
        self.total_samples += len(audio)
        return []

    def finalize(self):
        self.finalize_calls += 1
        return [[0, int(self.total_samples * 1000 / self.sample_rate)]]

    def reset(self):
        self.current_speech_start = None


class RecordingSpeakerTracker:
    def __init__(self):
        self.assigned = []
        self.finalize_calls = 0

    def assign_streaming(self, audio, start_sec, end_sec, sentence):
        sentence["spk"] = 7
        self.assigned.append((len(audio), start_sec, end_sec))

    def finalize(self, sentences):
        self.finalize_calls += 1
        return sentences

    def reset(self):
        pass


class OneShotSegmentVad:
    def __init__(self, sample_rate=16000):
        self.sample_rate = sample_rate
        self.current_speech_start = 0
        self.emitted = False

    def feed(self, audio, is_final=False):
        if self.emitted:
            return []
        self.emitted = True
        self.current_speech_start = None
        return [[0, int(len(audio) * 1000 / self.sample_rate)]]

    def reset(self):
        self.current_speech_start = None


def make_one_shot_segment_session(
    module, final_text, partial_text, partial_start_ms=0, engine=None
):
    sample_rate = 16000
    session = module.RealtimeASRSession(
        vllm_engine=engine or FixedTextEngine(final_text),
        asr_kwargs={},
        vad=OneShotSegmentVad(sample_rate),
        sample_rate=sample_rate,
        chunk_ms=960,
    )
    session.last_partial_text = partial_text
    session.last_partial_start_ms = partial_start_ms
    session.last_partial_end_ms = 2000
    session.segment_partial_text = partial_text
    session.segment_partial_start_ms = partial_start_ms
    session.segment_partial_end_ms = 2000
    session.segment_partial_stable_count = 2
    session.last_partial_eligible = True
    session.add_audio(np.zeros(2 * sample_rate, dtype=np.int16).tobytes())
    return session


def add_pcm(session, duration_ms, sample_rate=16000):
    samples = int(duration_ms * sample_rate / 1000)
    session.add_audio(np.zeros(samples, dtype=np.int16).tobytes())


def decode_partial(session, duration_ms):
    add_pcm(session, duration_ms, session.sample_rate)
    assert session.should_decode()
    return session.decode(is_final=False)


def test_completed_segment_keeps_aligned_partial_when_final_is_truncated_prefix():
    module = load_service_module()
    partial = (
        "当然可以。目前我们已经完成了第一阶段的调查研究工作，并收集了所有必要的数据。"
        "接下来我们将进入分析阶段，预计在下周完成。"
    )

    session = make_one_shot_segment_session(module, "当然可以。", partial)

    assert session.locked_sentences == [
        {"text": partial, "start": 0, "end": 2000}
    ]
    assert session.last_partial_text == ""


def make_reported_long_segment_reconciliation_session(module, partial, partial_end_ms):
    session = module.RealtimeASRSession(
        FixedTextEngine(""),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=960,
    )
    session.last_partial_text = partial
    session.last_partial_start_ms = 2920
    session.last_partial_end_ms = partial_end_ms
    session.last_partial_eligible = True
    session.segment_partial_observation_count = 2
    session.segment_best_partial_text = partial
    session.segment_best_partial_start_ms = 2920
    session.segment_best_partial_end_ms = partial_end_ms
    session.segment_best_partial_observation_count = 2
    return session


def test_long_segment_recovers_when_final_regresses_after_shared_prefix():
    module = load_service_module()
    partial = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "哎，OK OK，好嘞，胡总各位长官。"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=13500
    )

    text = session._reconcile_completed_segment_text(
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎。好，拜",
        [2920, 14420],
        decode_succeeded=True,
    )

    assert text == partial


def test_long_segment_recovers_from_catastrophically_short_final():
    module = load_service_module()
    partial = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "哎，O K O K，好嘞，胡总各位长官。"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=13500
    )

    text = session._reconcile_completed_segment_text(
        "哇，嘿",
        [2920, 14000],
        decode_succeeded=True,
    )

    assert text == partial


def test_long_segment_recovers_from_single_character_final():
    module = load_service_module()
    partial = (
        "哎，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "胡总，各位长官。"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=35328
    )

    text = session._reconcile_completed_segment_text(
        "哎",
        [2920, 36260],
        decode_succeeded=True,
    )

    assert text == partial


def test_long_segment_accepts_partial_slightly_past_vad_end():
    module = load_service_module()
    partial = (
        "哎，不好意思，上午好，高雄交通比较混乱，影响人有点。"
        "哎，哎，胡总，各位长官，大家好。"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=126976
    )
    session.last_partial_start_ms = 114430
    session.segment_best_partial_start_ms = 114430

    text = session._reconcile_completed_segment_text(
        "哎",
        [114430, 126740],
        decode_succeeded=True,
    )

    assert text == partial


def test_long_segment_rejects_partial_far_past_vad_end():
    module = load_service_module()
    partial = (
        "哎，不好意思，上午好，高雄交通比较混乱，影响人有点。"
        "哎，哎，胡总，各位长官，大家好。"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=128800
    )
    session.last_partial_start_ms = 114430
    session.segment_best_partial_start_ms = 114430

    text = session._reconcile_completed_segment_text(
        "哎",
        [114430, 126740],
        decode_succeeded=True,
    )

    assert text == "哎"


def test_long_segment_recovers_when_an_early_partial_correction_hides_tail_regression():
    module = load_service_module()
    partial = (
        "哇，哎，不好意思，上午杭高雄的交通比较混乱一点，我们不晓得我们有点。"
        "哎，好嘞，胡总各位长官。"
    )
    final = (
        "哇，不好意思，上午高雄的交通比较混乱一点，我们不晓得我们有点。"
        "哎，好嘞，"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=14336
    )

    text = session._reconcile_completed_segment_text(
        final,
        [2920, 14420],
        decode_succeeded=True,
    )

    assert text == partial


def test_long_segment_keeps_distinct_short_final_without_prefix_alignment():
    module = load_service_module()
    partial = "会议取消，稍后另行通知，请各位等待后续的完整安排和确认消息。"
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=13500
    )

    text = session._reconcile_completed_segment_text(
        "项目批准",
        [2920, 14000],
        decode_succeeded=True,
    )

    assert text == "项目批准"


def test_long_segment_does_not_trust_a_single_regressed_partial_observation():
    module = load_service_module()
    partial = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "哎，OK OK，好嘞，胡总各位长官。"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, partial, partial_end_ms=13500
    )
    session.segment_partial_observation_count = 1
    final = "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎。好，拜"

    text = session._reconcile_completed_segment_text(
        final,
        [2920, 14420],
        decode_succeeded=True,
    )

    assert text == final


def test_long_segment_can_recover_the_best_earlier_partial():
    module = load_service_module()
    best_partial = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎，"
        "OK OK OK。"
    )
    final = "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎。好，拜"
    session = make_reported_long_segment_reconciliation_session(
        module, final, partial_end_ms=13500
    )
    session.segment_partial_observation_count = 10
    session.segment_best_partial_text = best_partial
    session.segment_best_partial_start_ms = 2920
    session.segment_best_partial_end_ms = 12500
    session.segment_best_partial_observation_count = 10

    text = session._reconcile_completed_segment_text(
        final,
        [2920, 14420],
        decode_succeeded=True,
    )

    assert text == best_partial


def test_long_segment_recovers_from_merged_window_history_when_final_is_short():
    module = load_service_module()
    session = module.RealtimeASRSession(
        FixedTextEngine(""),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=960,
        partial_window_sec=8.0,
    )
    session.total_samples = 8000 * 16
    session._record_partial_text(
        "今天会议讨论项目进度以及后续计划安排", start_ms=0
    )
    session.total_samples = 12000 * 16
    session._record_partial_text(
        "后续计划安排需要各部门继续确认资源", start_ms=4000
    )
    session.total_samples = 16000 * 16
    session._record_partial_text(
        "继续确认资源之后再安排下周执行时间", start_ms=8000
    )

    text = session._reconcile_completed_segment_text(
        "今天会议", [0, 16000], decode_succeeded=True
    )

    assert text == (
        "今天会议讨论项目进度以及后续计划安排"
        "需要各部门继续确认资源之后再安排下周执行时间"
    )


def test_unaligned_merged_history_does_not_shadow_valid_best_partial():
    module = load_service_module()
    best_partial = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "哎，O K O K，好嘞，胡总各位长官。"
    )
    session = make_reported_long_segment_reconciliation_session(
        module, best_partial, partial_end_ms=13500
    )
    session.segment_partial_text = best_partial + "这段更长但没有覆盖语音段起点"
    session.segment_partial_start_ms = 4000
    session.segment_partial_end_ms = 13500
    session.segment_partial_observation_count = 4
    session.segment_partial_stable_count = 1

    text = session._reconcile_completed_segment_text(
        "哇，嘿", [2920, 14000], decode_succeeded=True
    )

    assert text == best_partial


def test_long_segment_keeps_clean_history_after_a_hallucinated_partial():
    module = load_service_module()
    best_partial = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "哎，OK OK，好嘞，胡总各位长官。"
    )
    final = "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎。好，拜"
    session = make_reported_long_segment_reconciliation_session(
        module, best_partial, partial_end_ms=13500
    )
    session.segment_best_partial_observation_count = 8

    session.last_partial_text = final
    session.last_partial_start_ms = 2920
    session.last_partial_end_ms = 13500
    session.last_partial_eligible = False
    session._reset_partial_history(preserve_best=True)

    text = session._reconcile_completed_segment_text(
        final,
        [2920, 14420],
        decode_succeeded=True,
    )

    assert text == best_partial


def test_retry_extension_requires_a_longer_aligned_clean_transcript():
    module = load_service_module()
    fallback = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎，"
        "OK OK OK。"
    )
    extension = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "哎，OK OK，好嘞，胡总各位长官。"
    )

    assert module._is_supported_transcript_extension(extension, fallback) is True
    assert module._is_supported_transcript_extension("完全不同的较长结果文本", fallback) is False
    assert module._is_supported_transcript_extension("哇，不好意思。", fallback) is False


def test_completed_segment_retries_only_after_a_supported_regression():
    module = load_service_module()
    fallback = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎，"
        "OK OK OK。"
    )
    extension = (
        "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点。"
        "哎，OK OK，好嘞，胡总各位长官。"
    )
    engine = ScriptedTextEngine(
        [
            "哇，不好意思，上午高雄交通比较混乱一点，我们不晓得我们有点哎。好，拜",
            extension,
        ]
    )
    session = make_reported_long_segment_reconciliation_session(
        module, fallback, partial_end_ms=13500
    )
    session.vllm_engine = engine
    session.total_samples = int(14.42 * session.sample_rate)
    session.audio_buffer = np.zeros(session.total_samples, dtype=np.float32)

    text = session._decode_segment([2920, 14420])

    assert text == extension
    assert len(engine.generate_kwargs) == 2


def test_completed_segment_does_not_retry_a_normal_final():
    module = load_service_module()
    final = "感谢各位今天参加会议，后续安排将在明天正式通知。"
    engine = ScriptedTextEngine([final])
    session = make_reported_long_segment_reconciliation_session(
        module, "感谢各位今天参加会以。", partial_end_ms=13500
    )
    session.vllm_engine = engine
    session.total_samples = int(14.42 * session.sample_rate)
    session.audio_buffer = np.zeros(session.total_samples, dtype=np.float32)

    text = session._decode_segment([2920, 14420])

    assert text == final
    assert len(engine.generate_kwargs) == 1


def test_completed_segment_keeps_clean_aligned_partial_when_final_hallucinates():
    module = load_service_module()
    partial = "不好意思，上午杭高架交通比较混乱一点，我们不晓得我们有点耽误。"
    repeated = "好，拜拜。" * 80

    session = make_one_shot_segment_session(module, repeated, partial)

    assert session.locked_sentences == [
        {"text": partial, "start": 0, "end": 2000}
    ]
    assert session.last_partial_text == ""


def test_completed_segment_keeps_aligned_partial_when_final_decode_fails():
    module = load_service_module()
    partial = "最终解码异常时，已经稳定显示的内容不应整句丢失。"

    session = make_one_shot_segment_session(
        module,
        "",
        partial,
        engine=FailingTextEngine(),
    )

    assert session.locked_sentences == [
        {"text": partial, "start": 0, "end": 2000}
    ]
    assert session.last_partial_text == ""


def test_completed_segment_keeps_distinct_final_correction_and_clears_partial():
    module = load_service_module()

    session = make_one_shot_segment_session(
        module,
        "感谢各位今天参加会议。",
        "感谢各位今天参加会以。",
    )

    assert session.locked_sentences == [
        {"text": "感谢各位今天参加会议。", "start": 0, "end": 2000}
    ]
    assert session.last_partial_text == ""


def test_completed_segment_does_not_reuse_partial_from_another_window():
    module = load_service_module()
    stale_partial = "当然可以。目前我们已经完成了第一阶段的调查研究工作。"

    session = make_one_shot_segment_session(
        module,
        "当然可以。",
        stale_partial,
        partial_start_ms=5000,
    )

    assert session.locked_sentences == [
        {"text": "当然可以。", "start": 0, "end": 2000}
    ]
    assert session.last_partial_text == ""


def test_hallucination_repair_handles_repeated_phrases_with_punctuation():
    module = load_service_module()
    prefix = "不好意思，上午杭高架交通比较混乱一点。"
    text = prefix + "好，拜拜。" * 80

    repaired, hallucinated = module.detect_and_fix_hallucination(text)

    assert hallucinated is True
    assert repaired == prefix + "好，拜拜。"


def test_server_stop_under_one_chunk_keeps_partial_when_final_decode_fails():
    module = load_service_module()
    engine = ScriptedTextEngine(
        ["已经稳定显示的短句。", RuntimeError("synthetic final decode failure")]
    )
    session = module.RealtimeASRSession(
        engine,
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=960,
    )

    partial = decode_partial(session, 600)
    assert partial["partial"] == "已经稳定显示的短句。"
    assert len(engine.generate_kwargs) == 1
    result = session.decode(is_final=True)

    assert len(engine.generate_kwargs) == 2
    assert engine.outputs == []
    assert result["sentences"] == [
        {"text": "已经稳定显示的短句。", "start": 0, "end": 600}
    ]
    assert session.last_partial_text == ""


def test_successful_empty_final_does_not_restore_old_partial():
    module = load_service_module()
    partial = "会议取消。稍后另行通知。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, partial, ""]),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=500,
    )

    decode_partial(session, 500)
    decode_partial(session, 500)
    result = session.decode(is_final=True)

    assert result["sentences"] == []
    assert session.last_partial_text == ""


def test_short_dense_final_can_remove_a_speculative_partial_suffix():
    module = load_service_module()
    partial = "会议取消。稍后另行通知。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, partial, "会议取消。"]),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=500,
    )

    decode_partial(session, 500)
    decode_partial(session, 500)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "会议取消。", "start": 0, "end": 1000}
    ]


def test_successful_final_can_remove_suffix_after_two_growing_partials():
    module = load_service_module()
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            [
                "会议取消。稍后另行通知。",
                "会议取消。稍后另行通知，请等待后续安排。",
                "会议取消。",
            ]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=1000,
        endpoint_mode="client",
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "会议取消。", "start": 0, "end": 2000}
    ]


def test_regressed_partial_resets_confidence_before_final_correction():
    module = load_service_module()
    partial = "会议取消。稍后另行通知，请等待后续安排。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, partial, "会议取消。", "会议取消。"]),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=1000,
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    decode_partial(session, 1000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "会议取消。", "start": 0, "end": 3000}
    ]


def test_vad_lock_keeps_stable_full_partial_and_assigns_speaker():
    module = load_service_module()
    partial = (
        "当然可以。目前我们已经完成了第一阶段的调查研究工作，并收集了所有必要的数据。"
        "接下来我们将进入分析阶段。"
    )
    vad = PreviousBoundaryVad()
    speaker = RecordingSpeakerTracker()
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, partial, "当然可以。"]),
        {},
        vad,
        spk_tracker=speaker,
        sample_rate=16000,
        chunk_ms=1000,
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    vad.lock_next = True
    add_pcm(session, 1000)

    assert session.locked_sentences == [
        {"text": partial, "start": 0, "end": 2000, "spk": 7}
    ]
    assert speaker.assigned == [(32000, 0.0, 2.0)]
    assert speaker.finalize_calls == 0
    assert session.last_partial_text == ""


def test_client_commit_uses_the_same_stable_partial_reconciliation():
    module = load_service_module()
    partial = (
        "当然可以。目前我们已经完成了第一阶段的调查研究工作，并收集了所有必要的数据。"
        "接下来我们将进入分析阶段。"
    )
    speaker = RecordingSpeakerTracker()
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, partial, "当然可以。"]),
        {},
        ControllableVad(),
        spk_tracker=speaker,
        sample_rate=16000,
        chunk_ms=1000,
        endpoint_mode="client",
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": partial, "start": 0, "end": 2000, "spk": 7}
    ]
    assert speaker.finalize_calls == 1
    assert session.last_partial_text == ""


def test_shifted_partial_without_full_segment_history_is_not_reused():
    module = load_service_module()
    partial = "当然可以。目前我们已经完成了全部调查研究工作。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, "当然可以。"]),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=960,
        partial_window_sec=1.5,
    )

    decode_partial(session, 2000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "当然可以。", "start": 0, "end": 2000}
    ]


def test_partial_missing_the_segment_tail_is_not_reused():
    module = load_service_module()
    partial = "当然可以。目前我们已经完成了全部调查研究工作，并进入分析阶段。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, partial, "当然可以。"]),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=1000,
        endpoint_mode="client",
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    add_pcm(session, 1000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "当然可以。", "start": 0, "end": 3000}
    ]


def test_nonoverlapping_partial_invalidates_old_full_segment_history():
    module = load_service_module()
    partial = "当然可以。目前我们已经完成了全部调查研究工作，并进入分析阶段。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            [
                partial,
                partial,
                "完全无关的新窗口内容",
                "仍然无关的后续窗口内容",
                "当然可以。",
            ]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=500,
        partial_window_sec=1.0,
        endpoint_mode="client",
    )

    decode_partial(session, 500)
    decode_partial(session, 500)
    decode_partial(session, 500)
    decode_partial(session, 500)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "当然可以。", "start": 0, "end": 2000}
    ]


def test_hallucinated_partial_is_not_locked_when_final_decode_fails():
    module = load_service_module()
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            ["好，拜拜。" * 80, RuntimeError("synthetic final decode failure")]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=500,
        endpoint_mode="client",
    )

    partial = decode_partial(session, 500)
    assert partial["partial"] == "好，拜拜。"
    result = session.decode(is_final=True)

    assert result["sentences"] == []


def test_hallucinated_partial_breaks_stable_partial_continuity():
    module = load_service_module()
    partial = "会议取消。稍后另行通知，请等待后续安排。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            [partial, "好，拜拜。" * 80, partial, "会议取消。"]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=1000,
        endpoint_mode="client",
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    decode_partial(session, 1000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "会议取消。", "start": 0, "end": 3000}
    ]


def test_partial_decode_error_breaks_stable_partial_continuity():
    module = load_service_module()
    partial = "会议取消。稍后另行通知，请等待后续安排。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            [partial, RuntimeError("synthetic partial failure"), partial, "会议取消。"]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=1000,
        endpoint_mode="client",
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    decode_partial(session, 1000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "会议取消。", "start": 0, "end": 3000}
    ]


def test_empty_partial_breaks_stable_partial_continuity():
    module = load_service_module()
    partial = "会议取消。稍后另行通知，请等待后续安排。"
    session = module.RealtimeASRSession(
        ScriptedTextEngine([partial, "", partial, "会议取消。"]),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=1000,
        endpoint_mode="client",
    )

    decode_partial(session, 1000)
    decode_partial(session, 1000)
    decode_partial(session, 1000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "会议取消。", "start": 0, "end": 3000}
    ]


def test_partial_overlap_cannot_bridge_an_audio_coverage_gap():
    module = load_service_module()
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            ["ABCDEFGH", "EFGHIJKLMNOP", "好，拜拜。" * 80]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=500,
        partial_window_sec=1.0,
        endpoint_mode="client",
    )

    decode_partial(session, 500)
    decode_partial(session, 2000)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "好，拜拜。", "start": 0, "end": 2500}
    ]


def test_partial_overlap_cannot_bridge_a_100ms_audio_gap():
    module = load_service_module()
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            ["ABCDEFGH", "EFGHIJKLMNOP", "好，拜拜。" * 80]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=500,
        partial_window_sec=0.5,
        endpoint_mode="client",
    )

    decode_partial(session, 500)
    decode_partial(session, 600)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "好，拜拜。", "start": 0, "end": 1100}
    ]


def test_partial_must_start_exactly_at_segment_start_for_fallback():
    module = load_service_module()
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            ["ABCDEFGHIJKLMNOP", "ABCDEFGHIJKLMNOP", "ABCD"]
        ),
        {},
        ControllableVad(),
        sample_rate=16000,
        chunk_ms=500,
        partial_window_sec=1.9,
        endpoint_mode="client",
    )

    decode_partial(session, 2000)
    session.decode(is_final=False)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "ABCD", "start": 0, "end": 2000}
    ]


def test_server_stop_assigns_speaker_to_active_segment():
    module = load_service_module()
    speaker = RecordingSpeakerTracker()
    session = module.RealtimeASRSession(
        FixedTextEngine("已经完成的短句。"),
        {},
        ControllableVad(),
        spk_tracker=speaker,
        sample_rate=16000,
        chunk_ms=960,
    )

    add_pcm(session, 600)
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {"text": "已经完成的短句。", "start": 0, "end": 600, "spk": 7}
    ]
    assert speaker.assigned == [(9600, 0.0, 0.6)]
    assert speaker.finalize_calls == 1


def test_server_stop_flushes_vad_before_decoding_final_segment():
    module = load_service_module()
    vad = FinalFlushVad()
    speaker = RecordingSpeakerTracker()
    engine = FixedTextEngine("由 VAD final flush 确认的短句。")
    session = module.RealtimeASRSession(
        engine,
        {},
        vad,
        spk_tracker=speaker,
        sample_rate=16000,
        chunk_ms=960,
    )

    add_pcm(session, 600)
    result = session.decode(is_final=True)

    assert vad.finalize_calls == 1
    assert len(engine.generate_kwargs) == 1
    assert result["sentences"] == [
        {
            "text": "由 VAD final flush 确认的短句。",
            "start": 0,
            "end": 600,
            "spk": 7,
        }
    ]
    assert speaker.finalize_calls == 1


def test_server_stop_finalizes_speakers_after_vad_already_locked():
    module = load_service_module()
    vad = ControllableVad()
    speaker = RecordingSpeakerTracker()
    session = module.RealtimeASRSession(
        FixedTextEngine("已经由 VAD 锁定的句子。"),
        {},
        vad,
        spk_tracker=speaker,
        sample_rate=16000,
        chunk_ms=960,
    )

    vad.lock_next = True
    add_pcm(session, 600)
    assert speaker.finalize_calls == 0
    result = session.decode(is_final=True)

    assert result["sentences"] == [
        {
            "text": "已经由 VAD 锁定的句子。",
            "start": 0,
            "end": 600,
            "spk": 7,
        }
    ]
    assert speaker.finalize_calls == 1


def test_overlapping_partial_windows_preserve_long_segment_on_final_regression():
    module = load_service_module()
    vad = PreviousBoundaryVad()
    session = module.RealtimeASRSession(
        ScriptedTextEngine(
            [
                "ABCD",
                "ABCDEFG",
                "ABCDEFGHI",
                "ABCDEFGHIJ",
                "DEFGHIJKLMNO",
                "好，拜拜。" * 80,
            ]
        ),
        {},
        vad,
        sample_rate=16000,
        chunk_ms=500,
        partial_window_sec=2.0,
    )

    for _ in range(5):
        decode_partial(session, 500)
    vad.lock_next = True
    add_pcm(session, 500)

    assert session.locked_sentences == [
        {"text": "ABCDEFGHIJKLMNO", "start": 0, "end": 2500}
    ]
    assert session.last_partial_text == ""


def test_hallucination_repair_stops_before_the_next_opening_quote():
    module = load_service_module()
    prefix = "前缀："

    repaired, hallucinated = module.detect_and_fix_hallucination(
        prefix + "“好”" * 80
    )

    assert hallucinated is True
    assert repaired == prefix + "“好”"


def test_overlap_merge_preserves_casefold_expansion_source_positions():
    module = load_service_module()

    assert module._merge_overlapping_transcripts(
        "Die Straße", "Straße ist frei"
    ) == "Die Straße ist frei"


def test_overlap_merge_preserves_new_punctuation_after_overlap():
    module = load_service_module()

    assert module._merge_overlapping_transcripts(
        "今天天气", "今天天气，真好"
    ) == "今天天气，真好"


def test_hallucination_repair_keeps_one_ascii_quoted_phrase():
    module = load_service_module()

    repaired, hallucinated = module.detect_and_fix_hallucination(
        "前缀：" + '"good"' * 80
    )

    assert hallucinated is True
    assert repaired == '前缀："good"'


def test_hallucination_repair_distinguishes_apostrophe_from_ascii_quote():
    module = load_service_module()

    repaired, hallucinated = module.detect_and_fix_hallucination(
        "don't " + "'good'" * 80
    )

    assert hallucinated is True
    assert repaired == "don't 'good'"


def test_overlap_merge_rejects_casefold_split_inside_source_character():
    module = load_service_module()

    assert module._merge_overlapping_transcripts("abcs", "abcßXYZ") is None


def test_client_endpoint_mode_skips_fsmn_vad_model(monkeypatch):
    module = load_service_module()
    auto_model_calls = []

    import funasr

    monkeypatch.setattr(
        funasr,
        "AutoModel",
        lambda model, **kwargs: auto_model_calls.append(model) or object(),
    )
    vllm_stub = types.ModuleType("funasr.auto.auto_model_vllm")
    vllm_stub.AutoModelVLLM = lambda **kwargs: object()
    monkeypatch.setitem(sys.modules, "funasr.auto.auto_model_vllm", vllm_stub)

    args = types.SimpleNamespace(
        model="FunAudioLLM/Fun-ASR-Nano-2512",
        hub="ms",
        device="cpu",
        dtype="fp32",
        tensor_parallel_size=1,
        gpu_memory_utilization=0.8,
        max_model_len=2048,
        hotword_file="",
        language=None,
        enable_spk=False,
        endpoint_mode="client",
    )

    _, _, vad_model, _ = module.load_models(args)

    assert vad_model is None
    assert auto_model_calls == []


def make_client_endpoint_session(module):
    assert hasattr(module, "ClientEndpointVAD")
    return module.RealtimeASRSession(
        vllm_engine=DummyEngine(),
        asr_kwargs={},
        vad=module.ClientEndpointVAD(),
        sample_rate=16000,
        chunk_ms=960,
        endpoint_mode="client",
    )


def test_client_endpoint_mode_emits_partial_at_first_decode_threshold():
    module = load_service_module()
    session = make_client_endpoint_session(module)
    audio = np.zeros(int(0.48 * session.sample_rate), dtype=np.int16).tobytes()

    session.add_audio(audio)
    response = session.decode(is_final=False)

    assert session.should_decode() is False
    assert response["partial"] == "hello"
    assert response["partial_start_ms"] == 0
    assert response["is_final"] is False
    assert session.vllm_engine.input_lengths == [int(0.48 * session.sample_rate)]


def test_postprocess_hotwords_correct_final_text_without_model_hotword_bias():
    module = load_service_module()
    session = make_client_endpoint_session(module)
    session.asr_kwargs = {
        "postprocess_hotwords": {"hello": "Tool"},
        "return_postprocess_hotword_matches": True,
    }

    session.add_audio(np.zeros(int(0.4 * session.sample_rate), dtype=np.int16).tobytes())
    result = session.commit()

    assert result["sentences"] == [{"text": "Tool", "start": 0, "end": 400}]
    assert session.vllm_engine.generate_kwargs[-1].get("hotwords") is None


def test_client_commits_short_utterances_once_with_monotonic_timestamps():
    module = load_service_module()
    session = make_client_endpoint_session(module)

    session.add_audio(np.zeros(int(0.4 * session.sample_rate), dtype=np.int16).tobytes())
    first = session.commit()
    session.add_audio(np.zeros(int(0.3 * session.sample_rate), dtype=np.int16).tobytes())
    second = session.commit()

    assert first == {
        "sentences": [{"text": "hello", "start": 0, "end": 400}],
        "partial": "",
        "partial_start_ms": 0,
        "duration_ms": 400,
        "is_final": True,
    }
    assert second == {
        "sentences": [{"text": "hello", "start": 400, "end": 700}],
        "partial": "",
        "partial_start_ms": 0,
        "duration_ms": 700,
        "is_final": True,
    }
    assert session.vllm_engine.input_lengths == [6400, 4800]
    assert len(session.audio_buffer) == 0
    assert session.audio_buffer_start_sample == session.total_samples


def test_handler_commit_returns_one_final_and_reuses_connection(monkeypatch):
    module = load_service_module()
    dynamic_vad_calls = []
    client_vad_calls = []
    received_audio = []

    class ProtocolSession:
        def __init__(self, *args, **kwargs):
            self.is_active = False
            self.total_samples = 0
            self.audio_buffer_start_sample = 0
            self.audio_buffer = np.array([], dtype=np.float32)
            self.commit_count = 0

        def reset(self):
            self.total_samples = 0
            self.audio_buffer_start_sample = 0
            self.audio_buffer = np.array([], dtype=np.float32)

        def add_audio(self, message):
            received_audio.append(message)
            self.total_samples += 1
            self.audio_buffer = np.ones(1, dtype=np.float32)

        def should_decode(self):
            return False

        def commit(self):
            self.commit_count += 1
            self.audio_buffer = np.array([], dtype=np.float32)
            self.audio_buffer_start_sample = self.total_samples
            return {"is_final": True, "sequence": self.commit_count}

        def decode(self, is_final=False):
            return self.commit()

    class FakeWebSocket:
        remote_address = ("127.0.0.1", 12345)

        def __init__(self):
            self.messages = iter(
                [
                    "START",
                    b"first",
                    "COMMIT",
                    b"second",
                    "COMMIT",
                    "STOP",
                    b"after-stop",
                ]
            )
            self.sent = []

        def __aiter__(self):
            return self

        async def __anext__(self):
            try:
                return next(self.messages)
            except StopIteration as error:
                raise StopAsyncIteration from error

        async def send(self, message):
            self.sent.append(json.loads(message))

    monkeypatch.setattr(module, "load_models", lambda args: (object(), {}, None, None))
    monkeypatch.setattr(
        module,
        "DynamicStreamingVAD",
        lambda model: dynamic_vad_calls.append(model) or object(),
    )
    monkeypatch.setattr(
        module,
        "ClientEndpointVAD",
        lambda: client_vad_calls.append(True) or object(),
        raising=False,
    )
    monkeypatch.setattr(module, "create_speaker_tracker", lambda model, args: None)
    monkeypatch.setattr(module, "RealtimeASRSession", ProtocolSession)
    websocket = FakeWebSocket()
    args = types.SimpleNamespace(
        device="cpu",
        decode_interval=0.48,
        partial_window_sec=15.0,
        endpoint_mode="client",
    )

    asyncio.run(module.handle_client(websocket, args))

    finals = [message for message in websocket.sent if message.get("is_final")]
    events = [message.get("event") for message in websocket.sent if message.get("event")]
    assert finals == [
        {"is_final": True, "sequence": 1},
        {"is_final": True, "sequence": 2},
    ]
    assert events == ["started", "stopped"]
    assert dynamic_vad_calls == []
    assert client_vad_calls == [True]
    assert received_audio == [b"first", b"second"]


def test_handler_accepts_postprocess_hotwords_without_model_hotword_bias(monkeypatch):
    module = load_service_module()
    session_kwargs = []

    class ProtocolSession:
        def __init__(self, vllm_engine, asr_kwargs, *args, **kwargs):
            self.is_active = False
            self.asr_kwargs = asr_kwargs
            session_kwargs.append(dict(asr_kwargs))

        def reset(self):
            pass

        def commit(self):
            return {"is_final": True, "asr_kwargs": dict(self.asr_kwargs)}

    class FakeWebSocket:
        remote_address = ("127.0.0.1", 12345)

        def __init__(self):
            self.messages = iter(
                [
                    "START",
                    "POSTPROCESS_HOTWORDS:hello=>Tool,哈囉=>客製化",
                    "COMMIT",
                ]
            )
            self.sent = []

        def __aiter__(self):
            return self

        async def __anext__(self):
            try:
                return next(self.messages)
            except StopIteration as error:
                raise StopAsyncIteration from error

        async def send(self, message):
            self.sent.append(json.loads(message))

    monkeypatch.setattr(module, "load_models", lambda args: (object(), {}, None, None))
    monkeypatch.setattr(module, "ClientEndpointVAD", lambda: object(), raising=False)
    monkeypatch.setattr(module, "create_speaker_tracker", lambda model, args: None)
    monkeypatch.setattr(module, "RealtimeASRSession", ProtocolSession)
    websocket = FakeWebSocket()
    args = types.SimpleNamespace(
        decode_interval=0.48,
        partial_window_sec=15.0,
        endpoint_mode="client",
        log_session_stats_interval=0.0,
    )

    asyncio.run(module.handle_client(websocket, args))

    assert session_kwargs == [{}]
    assert websocket.sent[1] == {
        "event": "postprocess_hotwords_set",
        "postprocess_hotwords": {"hello": "Tool", "哈囉": "客製化"},
    }
    final = [message for message in websocket.sent if message.get("is_final")][0]
    assert final["asr_kwargs"] == {
        "postprocess_hotwords": {"hello": "Tool", "哈囉": "客製化"},
        "postprocess_hotword_fuzzy": False,
        "return_postprocess_hotword_matches": True,
    }
    assert "hotwords" not in final["asr_kwargs"]


def test_load_models_reads_postprocess_hotword_file(monkeypatch, tmp_path):
    module = load_service_module()
    module._vllm_engine = None
    module._asr_kwargs = None
    module._vad_model = None
    module._spk_model = None
    hotword_file = tmp_path / "postprocess_hotwords.txt"
    hotword_file.write_text("hello=>Tool\n哈囉=>客製化\n", encoding="utf-8")
    auto_model_calls = []

    import funasr

    monkeypatch.setattr(
        funasr,
        "AutoModel",
        lambda model, **kwargs: auto_model_calls.append(model) or object(),
    )
    vllm_stub = types.ModuleType("funasr.auto.auto_model_vllm")
    vllm_stub.AutoModelVLLM = lambda **kwargs: object()
    monkeypatch.setitem(sys.modules, "funasr.auto.auto_model_vllm", vllm_stub)

    args = types.SimpleNamespace(
        model="FunAudioLLM/Fun-ASR-Nano-2512",
        hub="ms",
        device="cpu",
        dtype="fp32",
        tensor_parallel_size=1,
        gpu_memory_utilization=0.8,
        max_model_len=2048,
        hotword_file="",
        postprocess_hotword_file=str(hotword_file),
        language=None,
        enable_spk=False,
        endpoint_mode="client",
    )

    _, asr_kwargs, _, _ = module.load_models(args)

    assert asr_kwargs == {
        "postprocess_hotword_file": str(hotword_file),
        "postprocess_hotword_fuzzy": False,
        "return_postprocess_hotword_matches": True,
    }
    assert auto_model_calls == []


def test_two_hour_session_keeps_audio_bounded_and_duration_absolute():
    module = load_service_module()
    sample_rate = 10
    lookback_seconds = 5
    session = module.RealtimeASRSession(
        vllm_engine=DummyEngine(),
        asr_kwargs={},
        vad=SilentVad(),
        sample_rate=sample_rate,
        chunk_ms=1000,
        audio_lookback_sec=lookback_seconds,
    )
    one_second = np.zeros(sample_rate, dtype=np.int16).tobytes()

    for _ in range(2 * 60 * 60):
        session.add_audio(one_second)

    assert session.total_samples == 2 * 60 * 60 * sample_rate
    assert len(session.audio_buffer) <= lookback_seconds * sample_rate
    assert session.audio_buffer_start_sample == session.total_samples - len(session.audio_buffer)
    assert session._build_response(is_final=False)["duration_ms"] == 2 * 60 * 60 * 1000


def test_session_stats_report_bounded_long_session_state():
    module = load_service_module()
    sample_rate = 10
    session = module.RealtimeASRSession(
        vllm_engine=DummyEngine(),
        asr_kwargs={},
        vad=SilentVad(),
        sample_rate=sample_rate,
        chunk_ms=1000,
        audio_lookback_sec=5,
    )
    for _ in range(20):
        session.add_audio(np.zeros(sample_rate, dtype=np.int16).tobytes())
    session.locked_sentences = [{"text": "done", "start": 0, "end": 1000}]

    stats = session.session_stats()

    assert stats["duration_ms"] == 20000
    assert stats["total_samples"] == 200
    assert stats["audio_buffer_samples"] <= 50
    assert stats["audio_buffer_start_ms"] == 15000
    assert stats["locked_sentences"] == 1
    assert stats["partial_active"] is False


def test_completed_segment_uses_absolute_offsets_after_audio_compaction():
    module = load_service_module()
    sample_rate = 16000
    vad = SegmentVad(sample_rate=sample_rate, segment_end_sample=11 * sample_rate)
    engine = DummyEngine()
    session = module.RealtimeASRSession(
        vllm_engine=engine,
        asr_kwargs={},
        vad=vad,
        sample_rate=sample_rate,
        chunk_ms=1000,
        audio_lookback_sec=2,
    )
    one_second = np.zeros(sample_rate, dtype=np.int16).tobytes()

    for _ in range(11):
        session.add_audio(one_second)

    assert engine.input_lengths == [sample_rate]
    assert session.locked_sentences == [{"text": "hello", "start": 10000, "end": 11000}]
    assert session._build_response(is_final=False)["duration_ms"] == 11000
    assert session.audio_buffer_start_sample > 0


def test_final_decode_releases_audio_without_resetting_absolute_duration():
    module = load_service_module()
    sample_rate = 10
    for sample_count in (5, 100):
        session = module.RealtimeASRSession(
            vllm_engine=DummyEngine(),
            asr_kwargs={},
            vad=SilentVad(),
            sample_rate=sample_rate,
            chunk_ms=1000,
            audio_lookback_sec=5,
        )
        session.add_audio(np.zeros(sample_count, dtype=np.int16).tobytes())

        response = session.decode(is_final=True)

        assert response["duration_ms"] == sample_count * 100
        assert len(session.audio_buffer) == 0
        assert session.audio_buffer_start_sample == session.total_samples


def test_speaker_history_and_identity_state_have_hard_limits(monkeypatch):
    utils_stub = types.ModuleType("funasr.models.campplus.utils")
    utils_stub.sv_chunk = lambda segments: segments

    def postprocess(segments, vad_segments, labels, embeddings, return_spk_center=False):
        output = [[segment[0], segment[1], int(label)] for segment, label in zip(segments, labels)]
        centers = torch.stack(
            [embeddings[labels == label].mean(0) for label in sorted(set(labels.tolist()))]
        )
        return (output, centers) if return_spk_center else output

    def distribute_spk(sentences, speaker_segments):
        for sentence in sentences:
            sentence["spk"] = int(speaker_segments[-1][2])
        return sentences

    utils_stub.postprocess = postprocess
    utils_stub.distribute_spk = distribute_spk
    cluster_stub = types.ModuleType("funasr.models.campplus.cluster_backend")

    class FakeClusterBackend:
        def __init__(self, merge_thr):
            pass

        def to(self, device):
            return self

        def __call__(self, embeddings, oracle_num=None):
            return np.zeros(len(embeddings), dtype=np.int64)

    cluster_stub.ClusterBackend = FakeClusterBackend
    monkeypatch.setitem(sys.modules, "funasr.models.campplus.utils", utils_stub)
    monkeypatch.setitem(sys.modules, "funasr.models.campplus.cluster_backend", cluster_stub)

    module = load_service_module()

    class FakeSpeakerModel:
        def generate(self, input, **kwargs):
            return [{"spk_embedding": torch.tensor([[1.0, 0.0]])} for _ in input]

    tracker = module.HybridSpeakerTracker(
        spk_model=FakeSpeakerModel(),
        device="cpu",
        max_history_chunks=3,
        max_speakers=2,
    )
    tracker.sv_chunk = lambda segments: [
        [segments[0][0], segments[0][1], segments[0][2]]
    ]
    tracker.cluster_backend = lambda embeddings, oracle_num=None: np.zeros(
        len(embeddings), dtype=np.int64
    )

    for index in range(6):
        sentence = {"text": f"segment {index}", "start": index * 1000, "end": (index + 1) * 1000}
        tracker.assign_streaming(
            np.ones(16000, dtype=np.float32),
            index,
            index + 1,
            sentence,
        )
        assert sentence["spk"] == 0

    assert len(tracker.all_chunks) == 3
    assert len(tracker.all_embeddings) == 3
    assert all(len(chunk) == 2 for chunk in tracker.all_chunks)

    first = tracker._map_cluster_centers(
        torch.tensor([[1.0, 0.0], [0.0, 1.0]]), update=True
    )
    swapped = tracker._map_cluster_centers(
        torch.tensor([[0.0, 1.0], [1.0, 0.0]]), update=True
    )
    capped = tracker._map_cluster_centers(
        torch.tensor([[-1.0, 0.0]]), update=True
    )

    assert first == [0, 1]
    assert swapped == [1, 0]
    assert capped[0] in {0, 1}
    assert len(tracker.speaker_centers) == 2
    assert tracker.all_chunks.maxlen == 3
    assert tracker.all_embeddings.maxlen == 3

    finalized = tracker.finalize(
        [{"text": "abcdef", "start": 0, "end": 6000, "spk": 1}]
    )
    assert finalized == [
        {"text": "abc", "start": 0, "end": 3000, "spk": 1},
        {"text": "def", "start": 3000, "end": 6000, "spk": 0},
    ]


def test_realtime_batching_engine_combines_compatible_concurrent_requests():
    module = load_service_module()

    class RecordingEngine:
        def __init__(self):
            self.calls = []
            self._engine = object()

        def generate(self, inputs, **kwargs):
            self.calls.append((list(inputs), dict(kwargs)))
            return [{"text": str(int(item[0]))} for item in inputs]

    engine = RecordingEngine()
    batching_engine = module.RealtimeBatchingEngine(
        engine, batch_wait_ms=50, max_batch_size=8
    )
    assert batching_engine._engine is engine._engine
    barrier = threading.Barrier(3)
    results = [None, None]

    def generate(index, values):
        barrier.wait()
        results[index] = batching_engine.generate(
            [np.array([value], dtype=np.float32) for value in values],
            language="zh",
        )

    threads = [
        threading.Thread(target=generate, args=(0, [1, 2])),
        threading.Thread(target=generate, args=(1, [3])),
    ]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=2)

    assert all(not thread.is_alive() for thread in threads)
    assert [[item["text"] for item in result] for result in results] == [
        ["1", "2"],
        ["3"],
    ]
    assert len(engine.calls) == 1
    assert len(engine.calls[0][0]) == 3
    assert engine.calls[0][1] == {"language": "zh"}


def test_realtime_batching_engine_logs_opt_in_decode_profile(caplog):
    module = load_service_module()

    class RecordingEngine:
        def generate(self, inputs, **kwargs):
            return [{"text": "ok"} for _ in inputs]

    caplog.set_level("INFO")
    batching_engine = module.RealtimeBatchingEngine(
        RecordingEngine(), batch_wait_ms=0, max_batch_size=8, log_profile=True
    )

    result = batching_engine.generate(
        [
            np.zeros(16000, dtype=np.float32),
            np.zeros(32000, dtype=np.float32),
        ]
    )

    assert result == [{"text": "ok"}, {"text": "ok"}]
    assert "Realtime decode profile:" in caplog.text
    assert "requests=1" in caplog.text
    assert "samples=2" in caplog.text
    assert "audio_sec_total=3.000" in caplog.text
    assert "audio_sec_min=1.000" in caplog.text
    assert "audio_sec_max=2.000" in caplog.text
    assert "queue_ms_p50=" in caplog.text
    assert "queue_ms_max=" in caplog.text
    assert "engine_ms=" in caplog.text


def test_realtime_batching_engine_keeps_incompatible_options_separate():
    module = load_service_module()

    class RecordingEngine:
        def __init__(self):
            self.calls = []

        def generate(self, inputs, **kwargs):
            self.calls.append((list(inputs), dict(kwargs)))
            return [{"text": kwargs["language"]} for _ in inputs]

    engine = RecordingEngine()
    batching_engine = module.RealtimeBatchingEngine(
        engine, batch_wait_ms=50, max_batch_size=8
    )
    barrier = threading.Barrier(3)
    results = [None, None]

    def generate(index, language):
        barrier.wait()
        results[index] = batching_engine.generate(
            [np.array([index], dtype=np.float32)], language=language
        )

    threads = [
        threading.Thread(target=generate, args=(0, "zh")),
        threading.Thread(target=generate, args=(1, "en")),
    ]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=2)

    assert all(not thread.is_alive() for thread in threads)
    assert [result[0]["text"] for result in results] == ["zh", "en"]
    assert len(engine.calls) == 2


def test_realtime_batching_engine_propagates_engine_errors():
    module = load_service_module()

    class FailingEngine:
        def generate(self, inputs, **kwargs):
            raise RuntimeError("synthetic batch failure")

    batching_engine = module.RealtimeBatchingEngine(
        FailingEngine(), batch_wait_ms=0, max_batch_size=8
    )

    with pytest.raises(RuntimeError, match="synthetic batch failure"):
        batching_engine.generate([np.array([1], dtype=np.float32)])


def test_realtime_batching_engine_recovers_from_request_grouping_error():
    module = load_service_module()

    class ExplodingHash:
        def __hash__(self):
            raise RuntimeError("synthetic grouping failure")

    class RecordingEngine:
        def generate(self, inputs, **kwargs):
            return [{"text": "ok"} for _ in inputs]

    batching_engine = module.RealtimeBatchingEngine(
        RecordingEngine(), batch_wait_ms=10, max_batch_size=8
    )
    outcomes = {}

    def generate(name, marker):
        try:
            outcomes[name] = batching_engine.generate(
                [np.array([1], dtype=np.float32)], marker=marker
            )
        except Exception as error:
            outcomes[name] = error

    bad_thread = threading.Thread(
        target=generate, args=("bad", ExplodingHash()), daemon=True
    )
    bad_thread.start()
    bad_thread.join(timeout=2)
    good_thread = threading.Thread(target=generate, args=("good", "safe"), daemon=True)
    good_thread.start()
    good_thread.join(timeout=2)

    assert not bad_thread.is_alive()
    assert not good_thread.is_alive()
    assert isinstance(outcomes["bad"], RuntimeError)
    assert "synthetic grouping failure" in str(outcomes["bad"])
    assert outcomes["good"] == [{"text": "ok"}]


def test_realtime_batching_engine_recovers_from_result_distribution_error():
    module = load_service_module()

    class ExplodingResult(dict):
        def get(self, key, default=None):
            raise RuntimeError("synthetic distribution failure")

    class SelectiveEngine:
        def generate(self, inputs, **kwargs):
            if int(inputs[0][0]) < 0:
                return [ExplodingResult(text="bad")]
            return [{"key": None, "text": "ok"}]

    batching_engine = module.RealtimeBatchingEngine(
        SelectiveEngine(), batch_wait_ms=0, max_batch_size=8
    )

    with pytest.raises(RuntimeError, match="synthetic distribution failure"):
        batching_engine.generate([np.array([-1], dtype=np.float32)])

    assert batching_engine.generate([np.array([1], dtype=np.float32)]) == [
        {"key": None, "text": "ok"}
    ]


def test_realtime_batching_engine_enforces_max_batch_size_between_requests():
    module = load_service_module()

    class RecordingEngine:
        def __init__(self):
            self.calls = []

        def generate(self, inputs, **kwargs):
            self.calls.append(list(inputs))
            return [{"text": str(int(item[0]))} for item in inputs]

    engine = RecordingEngine()
    batching_engine = module.RealtimeBatchingEngine(
        engine, batch_wait_ms=50, max_batch_size=3
    )
    barrier = threading.Barrier(3)
    results = [None, None]

    def generate(index, values):
        barrier.wait()
        results[index] = batching_engine.generate(
            [np.array([value], dtype=np.float32) for value in values]
        )

    threads = [
        threading.Thread(target=generate, args=(0, [1, 2])),
        threading.Thread(target=generate, args=(1, [3, 4])),
    ]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=2)

    assert all(not thread.is_alive() for thread in threads)
    assert [[item["text"] for item in result] for result in results] == [
        ["1", "2"],
        ["3", "4"],
    ]
    assert len(engine.calls) == 2
    assert all(len(call) <= 3 for call in engine.calls)


def test_realtime_batching_engine_rejects_request_larger_than_max_batch_size():
    module = load_service_module()

    class RecordingEngine:
        def generate(self, inputs, **kwargs):
            raise AssertionError("oversized request must not reach the engine")

    batching_engine = module.RealtimeBatchingEngine(
        RecordingEngine(), batch_wait_ms=0, max_batch_size=1
    )

    with pytest.raises(ValueError, match="2 inputs.*maximum is 1"):
        batching_engine.generate(
            [np.array([1], dtype=np.float32), np.array([2], dtype=np.float32)]
        )


def test_realtime_batching_engine_isolates_a_bad_request_after_batch_failure():
    module = load_service_module()

    class SelectiveEngine:
        def generate(self, inputs, **kwargs):
            values = [int(item[0]) for item in inputs]
            if len(values) > 1:
                raise RuntimeError("synthetic batch failure")
            if values[0] < 0:
                raise ValueError("bad audio")
            return [{"key": "sample_0", "text": str(values[0])}]

    batching_engine = module.RealtimeBatchingEngine(
        SelectiveEngine(), batch_wait_ms=50, max_batch_size=8
    )
    barrier = threading.Barrier(3)
    outcomes = [None, None]

    def generate(index, value):
        barrier.wait()
        try:
            outcomes[index] = batching_engine.generate(
                [np.array([value], dtype=np.float32)]
            )
        except Exception as error:
            outcomes[index] = error

    threads = [
        threading.Thread(target=generate, args=(0, -1)),
        threading.Thread(target=generate, args=(1, 7)),
    ]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=2)

    assert all(not thread.is_alive() for thread in threads)
    assert isinstance(outcomes[0], ValueError)
    assert str(outcomes[0]) == "bad audio"
    assert outcomes[1] == [{"key": "sample_0", "text": "7"}]


def test_thread_safe_generate_model_serializes_shared_model_calls():
    module = load_service_module()

    class RecordingModel:
        def __init__(self):
            self.active = 0
            self.max_active = 0
            self.lock = threading.Lock()

        def generate(self, value):
            with self.lock:
                self.active += 1
                self.max_active = max(self.max_active, self.active)
            try:
                time.sleep(0.05)
                return value
            finally:
                with self.lock:
                    self.active -= 1

    model = RecordingModel()
    shared_model = module.ThreadSafeGenerateModel(model)
    threads = [
        threading.Thread(target=shared_model.generate, args=(index,))
        for index in range(2)
    ]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=1)

    assert all(not thread.is_alive() for thread in threads)
    assert model.max_active == 1
    assert shared_model.active == 0


def test_create_vad_shares_weights_but_copies_runtime_config(monkeypatch):
    module = load_service_module()

    class NestedVadModel:
        def __init__(self):
            self.encoder = object()
            self.vad_opts = types.SimpleNamespace(max_end_silence_time=800)

    class FakeAutoModel:
        _IMMUTABLE_KWARGS_KEYS = frozenset({"frontend"})

        def __init__(self):
            self.model = NestedVadModel()
            self.frontend = object()
            self.kwargs = {"frontend": self.frontend, "runtime": {"items": []}}
            self._base_kwargs_map = {"kwargs": dict(self.kwargs)}

    monkeypatch.setattr(module, "DynamicStreamingVAD", lambda value: value)
    model = FakeAutoModel()
    args = types.SimpleNamespace(endpoint_mode="server")

    first = module.create_vad(model, args)
    second = module.create_vad(model, args)

    assert first is not model
    assert second is not model
    assert first is not second
    assert first.model is not model.model
    assert second.model is not model.model
    assert first.model is not second.model
    assert first.model.encoder is model.model.encoder
    assert second.model.encoder is model.model.encoder
    assert first.model.vad_opts is not second.model.vad_opts
    assert first.kwargs is not second.kwargs
    assert first.kwargs["frontend"] is model.frontend
    first.kwargs["runtime"]["items"].append("first")
    assert second.kwargs["runtime"]["items"] == []


def test_create_vad_isolates_concurrent_nested_model_runtime_state(monkeypatch):
    module = load_service_module()
    barrier = threading.Barrier(2)

    class NestedVadModel:
        def __init__(self):
            self.encoder = object()
            self.vad_opts = types.SimpleNamespace(max_end_silence_time=800)

        def initialize(self, value):
            self.vad_opts.max_end_silence_time = value
            barrier.wait()
            return self.vad_opts.max_end_silence_time

    class FakeAutoModel:
        _IMMUTABLE_KWARGS_KEYS = frozenset()

        def __init__(self):
            self.model = NestedVadModel()
            self.kwargs = {}

        def generate(self, value):
            return self.model.initialize(value)

    monkeypatch.setattr(module, "DynamicStreamingVAD", lambda value: value)
    model = FakeAutoModel()
    args = types.SimpleNamespace(endpoint_mode="server")
    sessions = [module.create_vad(model, args), module.create_vad(model, args)]
    results = [None, None]
    threads = [
        threading.Thread(
            target=lambda index=index: results.__setitem__(
                index, sessions[index].generate(index + 1)
            )
        )
        for index in range(2)
    ]

    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(timeout=2)

    assert all(not thread.is_alive() for thread in threads)
    assert results == [1, 2]


def test_handler_keeps_event_loop_responsive_and_runs_connections_concurrently(
    monkeypatch,
):
    module = load_service_module()

    class BlockingSession:
        active_workers = 0
        max_active_workers = 0
        worker_lock = threading.Lock()

        def __init__(self, *args, **kwargs):
            self.is_active = False
            self.audio_buffer = np.zeros(1, dtype=np.float32)

        def reset(self):
            pass

        def add_audio(self, message):
            with self.worker_lock:
                type(self).active_workers += 1
                type(self).max_active_workers = max(
                    type(self).max_active_workers,
                    type(self).active_workers,
                )
            try:
                time.sleep(0.2)
            finally:
                with self.worker_lock:
                    type(self).active_workers -= 1

        def should_decode(self):
            return False

    class FakeWebSocket:
        remote_address = ("127.0.0.1", 12345)

        def __init__(self):
            self.messages = iter(["START", b"audio"])
            self.sent = []

        def __aiter__(self):
            return self

        async def __anext__(self):
            try:
                return next(self.messages)
            except StopIteration as error:
                raise StopAsyncIteration from error

        async def send(self, message):
            self.sent.append(message)

    monkeypatch.setattr(
        module, "load_models", lambda args: (object(), {}, object(), None)
    )
    monkeypatch.setattr(module, "DynamicStreamingVAD", lambda model: object())
    monkeypatch.setattr(module, "create_speaker_tracker", lambda model, args: None)
    monkeypatch.setattr(module, "RealtimeASRSession", BlockingSession)

    async def exercise_handler():
        stop = False
        gaps = []

        async def ticker():
            previous = asyncio.get_running_loop().time()
            while not stop:
                await asyncio.sleep(0.005)
                current = asyncio.get_running_loop().time()
                gaps.append(current - previous)
                previous = current

        ticker_task = asyncio.create_task(ticker())
        await asyncio.sleep(0.01)
        args = types.SimpleNamespace(device="cpu", decode_interval=0.48, partial_window_sec=15.0)
        await asyncio.gather(
            module.handle_client(FakeWebSocket(), args),
            module.handle_client(FakeWebSocket(), args),
        )
        stop = True
        await ticker_task
        return gaps

    gaps = asyncio.run(exercise_handler())

    assert gaps
    assert max(gaps) < 0.08
    assert BlockingSession.max_active_workers == 2


def test_handler_logs_session_stats_on_configured_interval(monkeypatch):
    module = load_service_module()
    logged = []

    class StatsSession:
        def __init__(self, *args, **kwargs):
            self.is_active = False
            self.audio_buffer = np.zeros(1, dtype=np.float32)
            self.total_samples = 0
            self.audio_buffer_start_sample = 0

        def reset(self):
            pass

        def add_audio(self, message):
            self.total_samples += 1

        def should_decode(self):
            return False

        def decode(self, is_final=False):
            assert is_final is True
            return {"text": "", "sentences": []}

        def session_stats(self):
            return {
                "duration_ms": self.total_samples,
                "audio_buffer_samples": 1,
                "audio_buffer_start_ms": 0,
                "locked_sentences": 0,
                "partial_active": False,
                "speaker_history_chunks": 0,
                "speaker_history_embeddings": 0,
                "speaker_centers": 0,
            }

    class FakeWebSocket:
        remote_address = ("127.0.0.1", 12345)

        def __init__(self):
            self.messages = iter(["START", b"a", b"b", "STOP"])
            self.sent = []

        def __aiter__(self):
            return self

        async def __anext__(self):
            try:
                return next(self.messages)
            except StopIteration as error:
                raise StopAsyncIteration from error

        async def send(self, message):
            self.sent.append(json.loads(message))

    clock = iter([0.0, 0.0, 2.0])
    monkeypatch.setattr(module, "load_models", lambda args: (object(), {}, object(), None))
    monkeypatch.setattr(module, "DynamicStreamingVAD", lambda model: object())
    monkeypatch.setattr(module, "create_speaker_tracker", lambda model, args: None)
    monkeypatch.setattr(module, "RealtimeASRSession", StatsSession)
    monkeypatch.setattr(
        module,
        "log_session_stats",
        lambda session: logged.append(session.session_stats()),
    )
    monkeypatch.setattr(module.time, "time", lambda: next(clock, 2.0))
    args = types.SimpleNamespace(
        device="cpu",
        decode_interval=60.0,
        partial_window_sec=15.0,
        endpoint_mode="server",
        log_session_stats_interval=1.0,
    )

    websocket = FakeWebSocket()
    asyncio.run(module.handle_client(websocket, args))

    assert logged == [
        {
            "duration_ms": 2,
            "audio_buffer_samples": 1,
            "audio_buffer_start_ms": 0,
            "locked_sentences": 0,
            "partial_active": False,
            "speaker_history_chunks": 0,
            "speaker_history_embeddings": 0,
            "speaker_centers": 0,
        }
    ]
    assert websocket.sent[-1] == {"event": "stopped"}