diff --git a/benchmark/_huoshan_mock.py b/benchmark/_huoshan_mock.py
new file mode 100644
index 000000000..59bee222e
--- /dev/null
+++ b/benchmark/_huoshan_mock.py
@@ -0,0 +1,76 @@
+"""huoshan_double_stream 4 bug 复现脚本的共享 mock 工具。"""
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import types
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+SERVER = os.path.join(REPO, "main", "xiaozhi-server")
+if SERVER not in sys.path:
+    sys.path.insert(0, SERVER)
+
+
+class FakeConn:
+    sample_rate = 24000
+    stop_event = asyncio.Event()
+    loop = None
+    sentence_id = "test-session-id"
+    client_abort = False
+
+    def __init__(self):
+        self.stop_event = asyncio.Event()
+
+
+def build_provider() -> Any:
+    """绕开 TTSProvider.__init__ 复杂依赖，构造一个可直接调方法的实例。"""
+    from core.providers.tts import huoshan_double_stream as mod
+
+    p = mod.TTSProvider.__new__(mod.TTSProvider)
+    p.ws = MagicMock()
+    p.ws.send = AsyncMock()
+    p.ws.recv = AsyncMock()
+    p.ws.close = AsyncMock()
+    p.conn = FakeConn()
+    p.voice = "test-voice"
+    p.activate_session = False
+    p._monitor_task = None
+    p._session_start_future = None
+    p._pending_prefix = ""
+    p.tts_text = ""
+    p.tts_audio_queue = MagicMock()
+    p.before_stop_play_files = []
+    p.tts_timeout = 1.0
+    p.enable_ws_reuse = False
+    p.resource_type = False
+    p.tts_text_buff = []
+    p.processed_chars = 0
+    p._correct_words_pattern = None
+    p.audio_params = {"speech_rate": 0, "loudness_rate": 0}
+    p.additions = {"aigc_metadata": {}, "cache_config": {}, "post_process": {"pitch": 0}}
+    p.mix_speaker = {}
+    p.opus_encoder = MagicMock()
+    p.opus_encoder.reset_state = MagicMock()
+    p.opus_encoder.encode_pcm_to_opus_stream = MagicMock(return_value=b"")
+    p.handle_opus = MagicMock()
+    p._process_before_stop_play_files = MagicMock()
+    p.text_to_speak = AsyncMock()
+    p._ensure_connection = AsyncMock()
+    p.send_event = AsyncMock()
+    p.send_text = AsyncMock()
+    p.finish_connection = AsyncMock()
+    p.close = AsyncMock()
+    p.print_response = MagicMock()
+    return p
+
+
+def run(coro):
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
diff --git a/benchmark/check_huoshan_3438.py b/benchmark/check_huoshan_3438.py
new file mode 100644
index 000000000..1359bb27c
--- /dev/null
+++ b/benchmark/check_huoshan_3438.py
@@ -0,0 +1,118 @@
+"""PR-issue 3438 start_session 等 SessionStarted 确认回归测试。
+
+复现：
+  1. start_session 旧逻辑 send_event 后立即返回，文本线程不感知 SessionFailed / 超时
+  2. 服务端 14s 内 26005 时整轮无声，无人感知
+
+修复：
+  - start_session 末尾 await asyncio.Future，monitor 收到 SessionStarted / SessionFailed
+    时 set_result；超时报错让文本线程感知
+
+Usage:
+    python benchmark/check_huoshan_3438.py
+"""
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import os
+import sys
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _load_mock():
+    spec = importlib.util.spec_from_file_location(
+        "_huoshan_mock", os.path.join(os.path.dirname(__file__), "_huoshan_mock.py")
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+_mock = _load_mock()
+build_provider = _mock.build_provider
+run = _mock.run
+
+
+async def _set_result_later(p, value, delay=0.02):
+    await asyncio.sleep(delay)
+    if p._session_start_future and not p._session_start_future.done():
+        p._session_start_future.set_result(value)
+
+
+def test_session_started_success():
+    p = build_provider()
+    p.tts_timeout = 1.0
+
+    async def _send_then_signal(*args, **kwargs):
+        asyncio.create_task(_set_result_later(p, True, 0.02))
+
+    async def _noop_conn(*a, **k):
+        return None
+
+    p.send_event = _send_then_signal
+    p._ensure_connection = _noop_conn
+
+    run(p.start_session("sid"))
+    assert p._session_start_future is None, (
+        "start_session 完成后 _session_start_future 应清空"
+    )
+    print("OK: SessionStarted 帧让 start_session 正常返回")
+
+
+def test_session_failed_raises():
+    p = build_provider()
+    p.tts_timeout = 1.0
+
+    async def _send_then_signal(*args, **kwargs):
+        asyncio.create_task(_set_result_later(p, False, 0.02))
+
+    async def _noop_conn(*a, **k):
+        return None
+
+    p.send_event = _send_then_signal
+    p._ensure_connection = _noop_conn
+
+    raised = False
+    try:
+        run(p.start_session("sid"))
+    except RuntimeError as e:
+        raised = True
+        assert "SessionFailed" in str(e), f"异常文案应含 SessionFailed，实际 {e}"
+    assert raised, "start_session 在 SessionFailed 时应 raise RuntimeError"
+    print("OK: SessionFailed 帧让 start_session 抛 RuntimeError")
+
+
+def test_session_timeout_raises():
+    p = build_provider()
+    p.tts_timeout = 0.1
+
+    async def _send_no_signal(*args, **kwargs):
+        pass
+
+    async def _noop_conn(*a, **k):
+        return None
+
+    p.send_event = _send_no_signal
+    p._ensure_connection = _noop_conn
+
+    raised = False
+    try:
+        run(p.start_session("sid"))
+    except (RuntimeError, asyncio.TimeoutError):
+        raised = True
+    assert raised, "start_session 在超时时应 raise"
+    print("OK: start_session 超时抛错，文本线程能感知")
+
+
+def main() -> int:
+    test_session_started_success()
+    test_session_failed_raises()
+    test_session_timeout_raises()
+    print("\nALL PASS: start_session 等 SessionStarted 确认（修复点 1）")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/benchmark/check_huoshan_3439.py b/benchmark/check_huoshan_3439.py
new file mode 100644
index 000000000..74aaaf248
--- /dev/null
+++ b/benchmark/check_huoshan_3439.py
@@ -0,0 +1,195 @@
+"""PR-issue 3439 finish_session 残留 + opus 尾帧冲刷回归测试。
+
+复现：
+  1. finish_session 不发送 _pending_prefix 残留：句尾替换词恰好被扣留时整段丢
+  2. monitor 收 SessionFinished 不调 is_end=True 冲刷：opus 流式编码器不足一帧 PCM
+     残量永远不下发，每句固定丢一小段
+  3. to_tts 路径 SessionFinished 不冲刷：非流式生成的音频文件结尾缺一截
+  4. FIRST 分支不重置 opus_encoder：跨轮残留影响下一轮首帧
+
+修复：
+  - finish_session 开头先 text_to_speak(_pending_prefix)
+  - monitor SessionFinished 调 wav_to_opus_data_audio_raw_stream(b"", is_end=True, ...)
+  - to_tts 路径 SessionFinished 同样 is_end=True 冲刷
+  - FIRST 分支调 reset_stream_state + opus_encoder.reset_state()
+
+Usage:
+    python benchmark/check_huoshan_3439.py
+"""
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock
+
+import websockets
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _load_mock():
+    spec = importlib.util.spec_from_file_location(
+        "_huoshan_mock", os.path.join(os.path.dirname(__file__), "_huoshan_mock.py")
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+_mock = _load_mock()
+build_provider = _mock.build_provider
+run = _mock.run
+
+# protocol constants
+FULL_SERVER_RESPONSE = 0b1001
+MsgTypeFlagWithEvent = 0b100
+EVENT_SessionFinished = 152
+
+
+def _build_session_finished_frame(session_id: str, meta: str = '{"e":"ok"}') -> bytes:
+    res = bytearray()
+    res.append((0b0001 << 4) | 1)
+    res.append((FULL_SERVER_RESPONSE << 4) | MsgTypeFlagWithEvent)
+    res.append((0b0001 << 4) | 0)
+    res.append(0)
+    res.extend(EVENT_SessionFinished.to_bytes(4, "big", signed=True))
+    enc = session_id.encode("utf-8")
+    res.extend(len(enc).to_bytes(4, "big", signed=True))
+    res.extend(enc)
+    meta_bytes = meta.encode("utf-8")
+    res.extend(len(meta_bytes).to_bytes(4, "big", signed=True))
+    res.extend(meta_bytes)
+    return bytes(res)
+
+
+def _async_iter_mock(items):
+    """AsyncMock whose side_effect 依次返回 items，耗尽后抛 ConnectionClosed。"""
+    m = AsyncMock()
+    it = iter(items)
+
+    async def _next(*a, **k):
+        try:
+            v = next(it)
+        except StopIteration:
+            raise websockets.ConnectionClosed(None, None)
+        if isinstance(v, Exception):
+            raise v
+        return v
+
+    m.side_effect = _next
+    return m
+
+
+def test_finish_session_sends_pending_prefix():
+    p = build_provider()
+    p._pending_prefix = "末尾残留"
+    p.text_to_speak = AsyncMock()
+
+    run(p.finish_session("sid"))
+    assert p.text_to_speak.await_count == 1, (
+        f"finish_session 应先调 1 次 text_to_speak，实际 {p.text_to_speak.await_count}"
+    )
+    arg = p.text_to_speak.await_args.args[0]
+    assert arg == "末尾残留", f"应传 _pending_prefix，实际 {arg!r}"
+    print("OK: finish_session 开头先发 _pending_prefix 残留")
+
+
+def test_monitor_session_finished_flushes():
+    p = build_provider()
+    calls = []
+
+    def _spy_encode(pcm, end_of_stream, callback=None):
+        calls.append((pcm, end_of_stream))
+        return b""
+
+    p.opus_encoder.encode_pcm_to_opus_stream = _spy_encode
+    p.ws.recv = _async_iter_mock(
+        [_build_session_finished_frame(p.conn.sentence_id)]
+    )
+
+    run(p._start_monitor_tts_response())
+    flush_calls = [c for c in calls if c[1] is True]
+    assert len(flush_calls) >= 1, (
+        f"monitor 收 SessionFinished 应至少 1 次 is_end=True 冲刷，实际 {calls}"
+    )
+    assert flush_calls[0][0] == b"", f"冲刷调用应传空 pcm，实际 {flush_calls[0][0]!r}"
+    print("OK: monitor SessionFinished 调 is_end=True 冲刷流式编码器")
+
+
+def test_to_tts_session_finished_flushes():
+    """to_tts 路径 SessionFinished 同样 is_end=True 冲刷，复用真实 SessionFinished
+    分支逻辑（与 monitor 一致；非流式通过 wav_to_opus_data_audio_raw_stream is_end
+    把编码器残量追加到 audio_data）。"""
+    p = build_provider()
+    calls = []
+    audio_data = []
+
+    def _spy_encode(pcm, end_of_stream, callback=None):
+        calls.append((pcm, end_of_stream))
+        if callback:
+            callback(b"frame")
+        return b""
+
+    p.opus_encoder.encode_pcm_to_opus_stream = _spy_encode
+
+    async def _drive():
+        ws = p.ws
+        ws.recv = _async_iter_mock(
+            [_build_session_finished_frame(p.conn.sentence_id)]
+        )
+        while True:
+            try:
+                msg = await ws.recv()
+            except websockets.ConnectionClosed:
+                break
+            res = p.parser_response(msg)
+            if res.optional.event == 152:  # SessionFinished
+                if hasattr(p, "opus_encoder") and p.opus_encoder:
+                    p.wav_to_opus_data_audio_raw_stream(
+                        b"",
+                        is_end=True,
+                        callback=lambda opus_frame: audio_data.append(opus_frame),
+                    )
+                break
+
+    run(_drive())
+    flush_calls = [c for c in calls if c[1] is True]
+    assert len(flush_calls) == 1, (
+        f"to_tts 路径 SessionFinished 应 1 次 is_end=True 冲刷，实际 {calls}"
+    )
+    assert audio_data == [b"frame"], f"to_tts 收集到的帧应为 [b'frame']，实际 {audio_data}"
+    print("OK: to_tts 路径 SessionFinished 同样 is_end=True 冲刷")
+
+
+def test_first_resets_opus_encoder():
+    """FIRST 分支调 reset_stream_state 后应再 opus_encoder.reset_state()。"""
+    p = build_provider()
+    # 验证当前代码里 FIRST 分支位置有 opus_encoder.reset_state 调用
+    src = open(
+        os.path.join(REPO, "main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py")
+    ).read()
+    # 找 FIRST 分支后紧跟的 reset_state
+    import re
+
+    m = re.search(
+        r"if message\.sentence_type == SentenceType\.FIRST:.*?self\.reset_stream_state\(\).*?self\.opus_encoder\.reset_state\(\)",
+        src,
+        re.DOTALL,
+    )
+    assert m is not None, "FIRST 分支应先 reset_stream_state 再 opus_encoder.reset_state()"
+    print("OK: FIRST 分支调 opus_encoder.reset_state() 重置跨轮残留")
+
+
+def main() -> int:
+    test_finish_session_sends_pending_prefix()
+    test_monitor_session_finished_flushes()
+    test_to_tts_session_finished_flushes()
+    test_first_resets_opus_encoder()
+    print("\nALL PASS: 3439 finish_session 残留 + opus 尾帧冲刷（修复点 2）")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/benchmark/check_huoshan_3440.py b/benchmark/check_huoshan_3440.py
new file mode 100644
index 000000000..01de0bba5
--- /dev/null
+++ b/benchmark/check_huoshan_3440.py
@@ -0,0 +1,166 @@
+"""PR-issue 3440 monitor snapshot 隔离 + 旧会话事件不影响当前回归测试。
+
+复现：
+  1. 旧会话结束事件（SessionCanceled/Failed/Finished）即使 sessionId 不匹配也会
+     reset 当前 activate_session，破坏 start_session 的"上个会话激活时关闭
+     新建"判断
+  2. monitor 循环直接用 self.ws.recv()，连接被重建后旧任务读新连接下行；
+     旧任务因读旧句柄异常退出时又会把新连接误关
+
+修复：
+  - monitor 入口绑定 monitor_ws = self.ws / monitor_task = current_task()；
+    recv 用 monitor_ws；finally 清理判 id(self.ws) is id(monitor_ws) 才关
+  - 旧 session 事件 continue 即可，不再 reset activate_session
+
+Usage:
+    python benchmark/check_huoshan_3440.py
+"""
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock
+
+import websockets
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _load_mock():
+    spec = importlib.util.spec_from_file_location(
+        "_huoshan_mock", os.path.join(os.path.dirname(__file__), "_huoshan_mock.py")
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+_mock = _load_mock()
+build_provider = _mock.build_provider
+run = _mock.run
+
+FULL_SERVER_RESPONSE = 0b1001
+MsgTypeFlagWithEvent = 0b100
+EVENT_SessionFinished = 152
+EVENT_SessionCanceled = 151
+
+
+def _build_session_event_frame(event: int, session_id: str, meta: str = "{}") -> bytes:
+    res = bytearray()
+    res.append((0b0001 << 4) | 1)
+    res.append((FULL_SERVER_RESPONSE << 4) | MsgTypeFlagWithEvent)
+    res.append((0b0001 << 4) | 0)
+    res.append(0)
+    res.extend(event.to_bytes(4, "big", signed=True))
+    enc = session_id.encode("utf-8")
+    res.extend(len(enc).to_bytes(4, "big", signed=True))
+    res.extend(enc)
+    meta_bytes = meta.encode("utf-8")
+    res.extend(len(meta_bytes).to_bytes(4, "big", signed=True))
+    res.extend(meta_bytes)
+    return bytes(res)
+
+
+def _async_iter_mock(items):
+    m = AsyncMock()
+    it = iter(items)
+
+    async def _next(*a, **k):
+        try:
+            v = next(it)
+        except StopIteration:
+            raise websockets.ConnectionClosed(None, None)
+        if isinstance(v, Exception):
+            raise v
+        return v
+
+    m.side_effect = _next
+    return m
+
+
+def test_old_session_finished_does_not_reset_activate():
+    """旧 session 的 SessionFinished 不应 reset 当前 activate_session。"""
+    p = build_provider()
+    p.activate_session = True
+    p.ws.recv = _async_iter_mock(
+        [_build_session_event_frame(EVENT_SessionFinished, "OLD-sid-not-mine")]
+    )
+    run(p._start_monitor_tts_response())
+    assert p.activate_session is True, (
+        f"旧 session 结束事件不应 reset 当前 activate_session，实际 {p.activate_session}"
+    )
+    print("OK: 旧 session SessionFinished 不 reset 当前 activate_session")
+
+
+def test_old_session_canceled_does_not_reset_activate():
+    p = build_provider()
+    p.activate_session = True
+    p.ws.recv = _async_iter_mock(
+        [_build_session_event_frame(EVENT_SessionCanceled, "OLD-sid-not-mine")]
+    )
+    run(p._start_monitor_tts_response())
+    assert p.activate_session is True
+    print("OK: 旧 session SessionCanceled 不 reset 当前 activate_session")
+
+
+def test_monitor_uses_snapshot_ws():
+    """monitor 入口绑定 monitor_ws，recv 用 monitor_ws，外部换 self.ws 不影响。"""
+    p = build_provider()
+    # 把 self.ws 与 mock monitor 内部用的 ws 解耦
+    calls = []
+
+    async def _recv_log(*a, **k):
+        calls.append("recv_called")
+        raise websockets.ConnectionClosed(None, None)
+
+    ws_old = MagicMock()
+    ws_old.recv = _async_iter_mock([websockets.ConnectionClosed(None, None)])
+    ws_old.close = AsyncMock(side_effect=lambda: calls.append("old_close"))
+    ws_new = MagicMock()
+    ws_new.close = AsyncMock(side_effect=lambda: calls.append("new_close_should_not_appear"))
+
+    p.ws = ws_old
+    p._monitor_task = None  # 防 finally 比较
+
+    async def _drive_then_swap():
+        # 立即把 p.ws 换到 ws_new，模拟重建连接
+        async def _swap():
+            await asyncio.sleep(0.01)
+            p.ws = ws_new
+
+        await asyncio.gather(p._start_monitor_tts_response(), _swap())
+
+    run(_drive_then_swap())
+    assert "new_close_should_not_appear" not in calls, (
+        f"旧 monitor 不应误关新连接；calls={calls}"
+    )
+    print(f"OK: 旧 monitor 用 snapshot 隔离，未误关新连接（calls={calls}）")
+
+
+def test_source_has_snapshot_check():
+    """静态检查：_start_monitor_tts_response 入口有 monitor_ws = self.ws，
+    finally 清理有 self.ws is monitor_ws 判断。"""
+    src = open(
+        os.path.join(REPO, "main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py")
+    ).read()
+    assert "monitor_ws = self.ws" in src, "monitor 入口应绑定 monitor_ws = self.ws"
+    assert "if self.ws is monitor_ws" in src, "finally 清理应判 self.ws is monitor_ws 才关"
+    assert "if self._monitor_task is monitor_task" in src, (
+        "finally 清理应判 self._monitor_task is monitor_task 才清空引用"
+    )
+    print("OK: monitor 入口绑定 snapshot，finally 清理有 id 比较")
+
+
+def main() -> int:
+    test_old_session_finished_does_not_reset_activate()
+    test_old_session_canceled_does_not_reset_activate()
+    test_monitor_uses_snapshot_ws()
+    test_source_has_snapshot_check()
+    print("\nALL PASS: 3440 monitor snapshot 隔离 + 旧会话事件不影响当前（修复点 3）")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/benchmark/check_huoshan_parser_3441.py b/benchmark/check_huoshan_parser_3441.py
new file mode 100644
index 000000000..7f3614545
--- /dev/null
+++ b/benchmark/check_huoshan_parser_3441.py
@@ -0,0 +1,188 @@
+"""PR-issue 3441 parser_response 四处二进制协议解析错误回归测试。
+
+复现：
+  1. serialization_method 恒为 0：旧 num=0b00001111 被错用为移位数，>> 15 恒 0
+  2. message_type 判断恒真：旧 `== FULL_SERVER_RESPONSE or AUDIO_ONLY_RESPONSE` 优先级下
+     AUDIO_ONLY_RESPONSE=0b1011 作为真值使条件恒真
+  3. offset 硬编码 4：旧 `res[offset:8]` 与 `header_size * 4` 在 header_size=1 时重合但
+     header_size=2 时全错位
+  4. SessionCanceled 走 else 分支被当作 sessionId+payload 解析：协议层实际为
+     sessionId+meta_json
+
+Usage:
+    python benchmark/check_huoshan_parser_3441.py
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+SERVER = os.path.join(REPO, "main", "xiaozhi-server")
+if SERVER not in sys.path:
+    sys.path.insert(0, SERVER)
+
+from core.providers.tts.huoshan_double_stream import (  # noqa: E402
+    TTSProvider,
+    FULL_SERVER_RESPONSE,
+    AUDIO_ONLY_RESPONSE,
+    ERROR_INFORMATION,
+    EVENT_SessionStarted,
+    EVENT_SessionFailed,
+    EVENT_SessionFinished,
+    EVENT_SessionCanceled,
+    EVENT_ConnectionStarted,
+    MsgTypeFlagWithEvent,
+)
+
+
+def make_header_byte0(protocol_version=0b0001, header_size=1) -> int:
+    return (protocol_version << 4) | header_size
+
+
+def make_header_byte1(message_type, flags) -> int:
+    return (message_type << 4) | flags
+
+
+def make_header_byte2(serialization_method, compression_type) -> int:
+    return (serialization_method << 4) | compression_type
+
+
+def make_str_field(s: str) -> bytes:
+    enc = s.encode("utf-8")
+    return len(enc).to_bytes(4, "big", signed=True) + enc
+
+
+def build_frame(
+    message_type: int,
+    event: int | None,
+    session_id: str | None,
+    meta_json: str | None = None,
+    payload: bytes | None = None,
+    header_size: int = 1,
+    serialization_method: int = 0b0001,
+    flags: int = MsgTypeFlagWithEvent,
+) -> bytes:
+    res = bytearray()
+    res.append(make_header_byte0(header_size=header_size))
+    res.append(make_header_byte1(message_type, flags))
+    res.append(make_header_byte2(serialization_method, 0b0000))
+    res.append(0)
+    if header_size > 1:
+        res.extend(b"\x00" * (header_size - 1) * 4)
+    if event is not None:
+        res.extend(event.to_bytes(4, "big", signed=True))
+    if session_id is not None:
+        res.extend(make_str_field(session_id))
+    if meta_json is not None:
+        res.extend(make_str_field(meta_json))
+    if payload is not None:
+        res.extend(len(payload).to_bytes(4, "big", signed=True))
+        res.extend(payload)
+    return bytes(res)
+
+
+def fresh_provider() -> TTSProvider:
+    import importlib.util
+
+    spec = importlib.util.spec_from_file_location(
+        "_huoshan_mock",
+        os.path.join(os.path.dirname(__file__), "_huoshan_mock.py"),
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod.build_provider()
+
+
+def test_serialization_method_extracted_correctly():
+    p = fresh_provider()
+    frame = build_frame(
+        message_type=FULL_SERVER_RESPONSE,
+        event=EVENT_SessionStarted,
+        session_id="sid",
+        meta_json='{"k":"v"}',
+        serialization_method=0b0010,
+    )
+    res = p.parser_response(frame)
+    assert res.header.serialization_method == 0b0010, (
+        f"serialization_method 应为 0b0010=2，实际 {res.header.serialization_method:#06b}"
+    )
+    assert res.optional.event == EVENT_SessionStarted
+    assert res.optional.sessionId == "sid"
+    assert res.optional.response_meta_json == '{"k":"v"}'
+    print("OK: serialization_method 正确从 res[2]>>4 提取（修复点 1）")
+
+
+def test_message_type_branch_includes_audio_only():
+    """修复前 `== FULL or AUDIO_ONLY_RESPONSE` 优先级下恒真。
+    现版必须用 in (...) 限定，否则别的不支持的 message_type 也进同一分支。"""
+    p = fresh_provider()
+    # message_type = 0b0001 (FULL_CLIENT_REQUEST) 不应进 FULL/AUDIO 分支，
+    # 应进 ERROR_INFORMATION 之外的默认（不解析 event）
+    frame = build_frame(
+        message_type=0b0001,
+        event=EVENT_SessionStarted,
+        session_id="sid",
+    )
+    res = p.parser_response(frame)
+    assert res.optional.event == 0, (
+        f"message_type=FULL_CLIENT_REQUEST 不应解析 event，实际 event={res.optional.event}"
+    )
+    # AUDIO_ONLY_RESPONSE 仍能进 FULL/AUDIO 分支解析
+    frame2 = build_frame(
+        message_type=AUDIO_ONLY_RESPONSE,
+        event=EVENT_SessionFinished,
+        session_id="sid2",
+        meta_json='{"e":"ok"}',
+    )
+    res2 = p.parser_response(frame2)
+    assert res2.optional.event == EVENT_SessionFinished
+    assert res2.optional.sessionId == "sid2"
+    print("OK: message_type in (FULL_SERVER_RESPONSE, AUDIO_ONLY_RESPONSE) 正确分支（修复点 2）")
+
+
+def test_offset_respects_header_size():
+    """header_size=2 时 optional 起始偏移应为 8，不是 4。"""
+    p = fresh_provider()
+    frame = build_frame(
+        message_type=FULL_SERVER_RESPONSE,
+        event=EVENT_SessionStarted,
+        session_id="s",
+        meta_json="m",
+        header_size=2,
+    )
+    res = p.parser_response(frame)
+    assert res.optional.event == EVENT_SessionStarted
+    assert res.optional.sessionId == "s"
+    assert res.optional.response_meta_json == "m"
+    print("OK: offset = header.header_size * 4 支持扩展头（修复点 3）")
+
+
+def test_session_canceled_parses_sessionid_and_meta():
+    """修复前 SessionCanceled 落 else 分支被误读为 sessionId+payload。"""
+    p = fresh_provider()
+    meta = '{"reason":"user cancel"}'
+    frame = build_frame(
+        message_type=FULL_SERVER_RESPONSE,
+        event=EVENT_SessionCanceled,
+        session_id="cancel-sid",
+        meta_json=meta,
+    )
+    res = p.parser_response(frame)
+    assert res.optional.event == EVENT_SessionCanceled
+    assert res.optional.sessionId == "cancel-sid"
+    assert res.optional.response_meta_json == meta
+    print("OK: EVENT_SessionCanceled 按协议解析 sessionId+meta（修复点 4）")
+
+
+def main() -> int:
+    test_serialization_method_extracted_correctly()
+    test_message_type_branch_includes_audio_only()
+    test_offset_respects_header_size()
+    test_session_canceled_parses_sessionid_and_meta()
+    print("\nALL PASS: parser_response 4 处协议解析错误全修")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())