Merge pull request #3678: add MOSS offline diarization sample

Keep Qwen3 behavior and add an explicit whole-recording MOSS HTTP path with tested client installation and documented speaker/completion boundaries.

Signed-off-by: zhifu gao <[email protected]>

.github/workflows/test-moss-adapter.yml

@@ -8,6 +8,8 @@ on:
       - "funasr/bin/_server_app.py"
       - "funasr/bin/server.py"
       - "examples/openai_api/**"
+      - "examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline*"
+      - "tests/test_qwen3_asr_vllm_offline_example.py"
       - "README*.md"
       - "docs/index.rst"
       - "docs/moss_transcribe_diarize*.md"
@@ -23,6 +25,8 @@ on:
       - "funasr/bin/_server_app.py"
       - "funasr/bin/server.py"
       - "examples/openai_api/**"
+      - "examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline*"
+      - "tests/test_qwen3_asr_vllm_offline_example.py"
       - "README*.md"
       - "docs/index.rst"
       - "docs/moss_transcribe_diarize*.md"
@@ -56,7 +60,9 @@ jobs:
           python -m black --check \
             funasr/models/moss_transcribe_diarize/model.py \
             tests/test_moss_transcribe_diarize_model.py \
-            tests/test_moss_transcribe_diarize_docs.py
+            tests/test_moss_transcribe_diarize_docs.py \
+            examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline.py \
+            tests/test_qwen3_asr_vllm_offline_example.py
           python -m py_compile \
             funasr/models/moss_transcribe_diarize/model.py \
             funasr/bin/_server_app.py \
@@ -67,4 +73,5 @@ jobs:
           python -m pytest -q \
             tests/test_moss_transcribe_diarize_model.py \
             tests/test_moss_transcribe_diarize_docs.py \
-            tests/test_server_app_openai_segments.py
+            tests/test_server_app_openai_segments.py \
+            tests/test_qwen3_asr_vllm_offline_example.py

docs/moss_transcribe_diarize.md

@@ -98,6 +98,13 @@ clients keep the exact tagged generation in `raw_text`; in structured mode,
 `raw_text` is the cleaned `text` returned by vLLM and the authoritative speaker
 metadata is in `sentence_info`.
 
+For a runnable CLI that writes speaker-attributed JSON and prints segment times, use
+[`transcribe_vllm_offline.py --engine moss`](../examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline_notes_en.md#moss-offline-transcription-and-anonymous-speakers).
+It sends a complete recording to a separate MOSS vLLM service, without Qwen3 chunking
+or local weight downloads. The example preserves the default Qwen3-only path; it does
+not combine Qwen3 text with MOSS labels. See the linked recipe for client installation,
+authentication, output fields, and long-recording completion limits.
+
 ## FunASR OpenAI-compatible service
 
 The built-in offline HTTP service exposes the same normalized result through

docs/moss_transcribe_diarize_zh.md

@@ -87,6 +87,12 @@ speaker `segments` 直接映射到 `sentence_info`。认证可通过 `vllm_api_k
 此时 `raw_text` 保留原始标签生成；结构化模式下，`raw_text` 是 vLLM 返回的清理后
 `text`，权威说话人信息位于 `sentence_info`。
 
+需要可直接运行、保存说话人 JSON 并打印分段时间的命令行示例，可用
+[`transcribe_vllm_offline.py --engine moss`](../examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline_notes.md#moss-离线转写与匿名说话人)。
+它将整条录音发送给独立的 MOSS vLLM 服务，不经过 Qwen3 切块，也不下载本地权重。
+示例保留默认的 Qwen3 单独转写路径，不混拼 Qwen3 文本与 MOSS 标签；客户端安装、
+认证、输出字段和长音频输出上限见该说明。
+
 ## FunASR OpenAI 兼容服务
 
 内置离线 HTTP 服务通过 `/v1/audio/transcriptions` 返回同一套标准化结果。

examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python3
-"""Transcribe long audio with Qwen3-ASR's native vLLM backend."""
+"""Offline Qwen3-ASR transcription or MOSS HTTP transcription and diarization."""
 
 import argparse
 import json
+import math
+import os
 import subprocess
 import tempfile
 from pathlib import Path
@@ -30,9 +32,7 @@ def transcribe_chunks(model, chunks, sample_rate, language):
         segments.append(
             {
                 "start_ms": round(offset_seconds * 1000),
-                "end_ms": round(
-                    (offset_seconds + len(audio) / sample_rate) * 1000
-                ),
+                "end_ms": round((offset_seconds + len(audio) / sample_rate) * 1000),
                 "text": (_result_value(result, "text", "") or "").strip(),
                 "language": _result_value(result, "language"),
             }
@@ -42,14 +42,23 @@ def transcribe_chunks(model, chunks, sample_rate, language):
 
 def build_parser():
     parser = argparse.ArgumentParser(
-        description="Offline long-audio Qwen3-ASR using its native vLLM backend"
+        description="Offline Qwen3-ASR (native vLLM) or MOSS diarization (HTTP)"
     )
     parser.add_argument("audio", type=Path)
-    parser.add_argument("--model", default="Qwen/Qwen3-ASR-1.7B")
-    parser.add_argument("--language", default=None)
-    parser.add_argument("--chunk-seconds", type=float, default=180.0)
-    parser.add_argument("--max-inference-batch-size", type=int, default=4)
-    parser.add_argument("--gpu-memory-utilization", type=float, default=0.8)
+    parser.add_argument("--engine", choices=("qwen3", "moss"), default="qwen3")
+    qwen = parser.add_argument_group("Qwen3 native vLLM only")
+    qwen.add_argument("--model", default="Qwen/Qwen3-ASR-1.7B")
+    qwen.add_argument("--language", default=None)
+    qwen.add_argument("--chunk-seconds", type=float, default=180.0)
+    qwen.add_argument("--max-inference-batch-size", type=int, default=4)
+    qwen.add_argument("--gpu-memory-utilization", type=float, default=0.8)
+    moss = parser.add_argument_group(
+        "MOSS HTTP only (MOSS_VLLM_API_KEY for authentication)"
+    )
+    moss.add_argument("--vllm-base-url", default=None)
+    moss.add_argument("--served-model", default="moss-transcribe-diarize")
+    moss.add_argument("--request-timeout", type=float, default=600.0)
+    moss.add_argument("--max-completion-tokens", type=int, default=8192)
     parser.add_argument("--output", type=Path, default=None)
     return parser
 
@@ -74,13 +83,70 @@ def _convert_to_mono_wav(audio_path, wav_path):
         raise RuntimeError("ffmpeg is required to normalize the input audio") from exc
 
 
+def run_moss(args):
+    if not args.vllm_base_url or not args.served_model.strip():
+        raise ValueError("MOSS requires --vllm-base-url and a non-empty --served-model")
+    if not math.isfinite(args.request_timeout) or args.request_timeout <= 0:
+        raise ValueError("request_timeout must be finite and positive")
+    if args.max_completion_tokens <= 0:
+        raise ValueError("max_completion_tokens must be positive")
+    if (
+        args.language is not None
+        or args.model != "Qwen/Qwen3-ASR-1.7B"
+        or args.chunk_seconds != 180.0
+        or args.max_inference_batch_size != 4
+        or args.gpu_memory_utilization != 0.8
+    ):
+        raise ValueError("Qwen3 model/language/chunk/GPU options do not apply to MOSS")
+    output = args.output or args.audio.with_suffix(".moss-vllm.json")
+    if output.resolve() == args.audio.resolve() or (
+        output.exists() and output.samefile(args.audio)
+    ):
+        raise ValueError("output must not overwrite the input audio")
+
+    from funasr import AutoModel
+
+    model = AutoModel(
+        model="OpenMOSS-Team/MOSS-Transcribe-Diarize",
+        backend="vllm",
+        device="cpu",
+        vllm_base_url=args.vllm_base_url,
+        vllm_model=args.served_model,
+        vllm_response_format="diarized_json",
+        vllm_timeout=args.request_timeout,
+        vllm_api_key=os.environ.get("MOSS_VLLM_API_KEY", "EMPTY"),
+        disable_update=True,
+        disable_pbar=True,
+    )
+    # Keep one recording in one request so anonymous speaker labels share a scope.
+    results = model.generate(
+        str(args.audio), max_completion_tokens=args.max_completion_tokens
+    )
+    if len(results) != 1:
+        raise RuntimeError("MOSS must return one result for the complete recording")
+    payload = results[0]
+    output.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+    )
+    return output, payload
+
+
 def run(args):
+    if not args.audio.is_file():
+        raise FileNotFoundError(args.audio)
+    if args.engine == "moss":
+        return run_moss(args)
+    if (
+        args.vllm_base_url is not None
+        or args.served_model != "moss-transcribe-diarize"
+        or args.request_timeout != 600.0
+        or args.max_completion_tokens != 8192
+    ):
+        raise ValueError("MOSS HTTP options require --engine moss")
     if args.chunk_seconds <= 0:
         raise ValueError("chunk_seconds must be positive")
     if args.max_inference_batch_size <= 0:
         raise ValueError("max_inference_batch_size must be positive")
-    if not args.audio.is_file():
-        raise FileNotFoundError(args.audio)
 
     import soundfile as sf
     from qwen_asr import Qwen3ASRModel
@@ -123,7 +189,15 @@ def main():
     args = build_parser().parse_args()
     output, payload = run(args)
     print(payload["text"])
-    print(f"Wrote {len(payload['segments'])} segments to {output}")
+    if args.engine == "moss":
+        segments = payload["sentence_info"]
+        for segment in segments:
+            print(
+                f"{segment['start']}..{segment['end']} ms [{segment['spk']}] {segment['text']}"
+            )
+    else:
+        segments = payload["segments"]
+    print(f"Wrote {len(segments)} segments to {output}")
 
 
 if __name__ == "__main__":

examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline_notes.md

@@ -1,6 +1,6 @@
 # Qwen3-ASR 离线长音频 vLLM 示例
 
-`transcribe_vllm_offline.py` 使用 Qwen3-ASR 自带的 `Qwen3ASRModel.LLM` 后端，不经过
+`transcribe_vllm_offline.py` 默认使用 Qwen3-ASR 自带的 `Qwen3ASRModel.LLM` 后端，不经过
 `AutoModelVLLM`。后者服务于 FunASR 自有模型，当前不支持 Qwen3-ASR。
 
 这个示例先用 ffmpeg 将输入统一为 16 kHz 单声道，再复用 qwen-asr 自带的静音边界切块，
@@ -39,3 +39,52 @@ python examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline.py
 该示例受到 [qwen3-asr-service](https://github.com/LanceLRQ/qwen3-asr-service) 的离线
 vLLM 实践以及 [FunASR #3419](https://github.com/modelscope/FunASR/issues/3419) 用户复测
 启发。不同测试集、参考稿对齐方式和模型尺寸的 CER 不能直接横向等同。
+
+## MOSS 离线转写与匿名说话人
+
+需要录音内说话人归属时，显式选择 `--engine moss`。这是第三方
+MOSS-Transcribe-Diarize 的独立 HTTP 路径，不是给 Qwen3-ASR 的文本追加标签。
+客户端复用 FunASR `AutoModel` 的 MOSS 适配器，整条录音只上传一次，不经过上述
+180 秒切块、外部 VAD 或第二个 speaker 模型，也不在客户端下载 MOSS 权重。
+
+先在独立 GPU 环境按 [MOSS vLLM 部署指南](../../../docs/moss_transcribe_diarize.md#vllm)
+启动服务，并确认其返回 `diarized_json`。该指南固定的服务端 vLLM 为 **0.27.1**；
+不要把它安装进上面的 Qwen3 `vllm==0.14.0` 环境。本机只作为 HTTP 客户端时不需要 GPU。
+
+在包含本示例更新的 FunASR 仓库根目录，为客户端建立另一环境。以下以 Linux/Python 3.12
+CPU 客户端为例；FunASR 的安装依赖不自动包含 Torch，因此需要显式安装：
+
+```bash
+python3.12 -m venv .venv-moss-client
+source .venv-moss-client/bin/activate
+python -m pip install "torch==2.10.0" --index-url https://download.pytorch.org/whl/cpu
+python -m pip install -e .
+
+python examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline.py \
+  recording.mp3 \
+  --engine moss \
+  --vllm-base-url http://127.0.0.1:8898/v1 \
+  --served-model moss-transcribe-diarize \
+  --request-timeout 600 \
+  --max-completion-tokens 8192
+```
+
+将服务地址替换为你实际可访问的地址，`--served-model` 必须匹配服务端注册名。
+服务启用认证时，事先通过密钥管理方式设置环境变量 `MOSS_VLLM_API_KEY`，不要把真实密钥
+写进脚本或命令行。此模式不要传 Qwen3 专用的 `--model`、`--language`、切块或 GPU 参数。
+示例来自当前源码，旧版 PyPI 安装中的脚本未必包含此模式。
+
+默认输出 `recording.moss-vllm.json`，也可通过 `--output` 指定其他结果文件：
+
+- `text`：MOSS 服务返回的全文，不与 Qwen3 结果混拼。
+- `sentence_info`：每段的 `start`、`end`、`text`、`spk`；起止时间为毫秒。
+- `timestamp`：对应的 `[start_ms, end_ms]` 列表。
+- `raw_text`：在这个结构化模式下等于服务返回的清理后文本，并非原始带控制标签的生成串。
+
+终端也会打印各段时间和匿名标签，例如 `10..90 ms [S02] ...`。HTTP 失败或畸形段数据会
+报错，不生成新的成功结果，也不覆盖已有结果文件。`S01/S02` 仅在单条录音中表示匿名
+说话人，不是已知人物身份、声纹验证或跨文件身份；这不是实时 WebSocket 路径。
+
+`8192` 只是可调输出上限，不保证 480 秒或更长录音完整转写。需同时核对服务端上下文/
+输出限制、最后一段时间、末尾语音和实际说话人归属。若自行切片，标签不能直接跨片合并；
+本样例不会伪造身份连续性。本机 HTTP 契约测试不代表实际模型的 CER、说话人准确率或容量验证。

examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline_notes_en.md

@@ -1,6 +1,6 @@
 # Offline long-form Qwen3-ASR with vLLM
 
-`transcribe_vllm_offline.py` uses Qwen3-ASR's native `Qwen3ASRModel.LLM` backend. It does
+`transcribe_vllm_offline.py` defaults to Qwen3-ASR's native `Qwen3ASRModel.LLM` backend. It does
 not use `AutoModelVLLM`, which targets FunASR-native models and does not currently support
 Qwen3-ASR.
 
@@ -44,3 +44,61 @@ This example was informed by the offline vLLM work in
 [qwen3-asr-service](https://github.com/LanceLRQ/qwen3-asr-service) and the user evaluation
 in [FunASR #3419](https://github.com/modelscope/FunASR/issues/3419). CER values from
 different datasets, transcript alignment rules, or model sizes are not directly comparable.
+
+## MOSS Offline Transcription and Anonymous Speakers
+
+Select `--engine moss` for within-recording speaker attribution. This is a separate
+HTTP path for the third-party MOSS-Transcribe-Diarize model, not speaker labels attached
+to Qwen3-ASR text. It reuses the FunASR `AutoModel` MOSS adapter and uploads the whole
+recording once: no Qwen3 180-second splitter, external VAD, second speaker model, or
+client-side MOSS weight download.
+
+Start a service in a separate GPU environment using the
+[MOSS vLLM deployment guide](../../../docs/moss_transcribe_diarize.md#vllm), and verify
+that it returns `diarized_json`. That guide pins server vLLM **0.27.1**. Do not install
+it into the Qwen3 `vllm==0.14.0` environment above. The HTTP client itself needs no GPU.
+
+From a FunASR checkout containing this example update, create a separate client environment.
+This recipe targets a Linux/Python 3.12 CPU client. FunASR's installation dependencies
+do not install Torch automatically, so install it explicitly:
+
+```bash
+python3.12 -m venv .venv-moss-client
+source .venv-moss-client/bin/activate
+python -m pip install "torch==2.10.0" --index-url https://download.pytorch.org/whl/cpu
+python -m pip install -e .
+
+python examples/industrial_data_pretraining/qwen3_asr/transcribe_vllm_offline.py \
+  recording.mp3 \
+  --engine moss \
+  --vllm-base-url http://127.0.0.1:8898/v1 \
+  --served-model moss-transcribe-diarize \
+  --request-timeout 600 \
+  --max-completion-tokens 8192
+```
+
+Use your actual reachable service URL and match `--served-model` to its registered
+model name. For authenticated services, set `MOSS_VLLM_API_KEY` through your secret
+management mechanism before running; do not place real keys in source or command-line
+arguments. Do not pass Qwen3-only `--model`, `--language`, chunking, or GPU options in
+this mode. The example is supplied in current source; older PyPI installations may
+not contain this mode.
+
+The default output is `recording.moss-vllm.json`; `--output` selects another result file:
+
+- `text`: the MOSS service transcript, never merged with Qwen3 output.
+- `sentence_info`: segment `start`, `end`, `text`, and `spk`, with times in milliseconds.
+- `timestamp`: the corresponding `[start_ms, end_ms]` pairs.
+- `raw_text`: in this structured mode, the cleaned service text, not the raw tagged generation.
+
+The terminal also prints segment times and anonymous labels, such as `10..90 ms [S02] ...`.
+HTTP errors or malformed segments raise an error without creating a new success result
+or replacing an existing result file. `S01/S02` identify anonymous speakers within one
+recording only, not known people, voice verification, or cross-file identities. This is
+not a realtime WebSocket path.
+
+`8192` is an adjustable completion limit, not a guarantee of complete transcription for
+480-second or longer recordings. Check server context/output limits, the final segment
+time, audible tail, and actual speaker attribution. Labels from independently split
+files cannot simply be joined; this sample does not invent cross-chunk continuity.
+Local HTTP contract tests do not establish model CER, diarization accuracy, or capacity.

tests/test_qwen3_asr_vllm_offline_example.py

@@ -1,6 +1,11 @@
 import importlib.util
+import json
+import sys
 import unittest
 from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -88,5 +93,376 @@ class Qwen3AsrVllmOfflineExampleTest(unittest.TestCase):
         self.assertEqual(args.max_inference_batch_size, 4)
 
 
+@pytest.fixture
+def moss_run(tmp_path, monkeypatch):
+    example = load_example()
+    audio = tmp_path / "meeting.wav"
+    import wave
+
+    with wave.open(str(audio), "wb") as source:
+        source.setnchannels(1)
+        source.setsampwidth(2)
+        source.setframerate(16000)
+        source.writeframes(b"\x00\x00" * 1600)
+    args = example.build_parser().parse_args([str(audio)])
+    args.engine = "moss"
+    args.vllm_base_url = "http://moss.test:8898/v1"
+    args.served_model = "meeting-model"
+    args.request_timeout = 37.5
+    args.max_completion_tokens = 1234
+    response = MagicMock()
+    response.json.return_value = {
+        "text": "hello again",
+        "segments": [
+            {"start": 0.01, "end": 0.04, "text": "hello", "speaker": "S01"},
+            {"start": 0.05, "end": 0.09, "text": "again", "speaker": "S02"},
+        ],
+    }
+    monkeypatch.setenv("MOSS_VLLM_API_KEY", "test-only-key")
+    monkeypatch.setitem(sys.modules, "qwen_asr", None)
+    monkeypatch.setitem(sys.modules, "qwen_asr.inference.utils", None)
+    with patch(
+        "funasr.auto.auto_model.download_model",
+        side_effect=AssertionError("HTTP client must not download weights"),
+    ), patch("requests.Session.post", return_value=response) as post, patch.object(
+        example,
+        "_convert_to_mono_wav",
+        side_effect=AssertionError("No Qwen3 splitter/conversion"),
+    ):
+        yield example, args, response, post
+
+
+def test_moss_uses_actual_adapter_once_and_preserves_speakers(moss_run):
+    example, args, response, post = moss_run
+    output, payload = example.run(args)
+    assert output == args.audio.with_suffix(".moss-vllm.json")
+    assert json.loads(output.read_text()) == payload
+    assert payload["text"] == payload["raw_text"] == "hello again"
+    assert payload["timestamp"] == [[10, 40], [50, 90]]
+    assert [(x["spk"], x["start"], x["end"]) for x in payload["sentence_info"]] == [
+        ("S01", 10, 40),
+        ("S02", 50, 90),
+    ]
+    assert not args.audio.with_suffix(".qwen3-vllm.json").exists()
+    post.assert_called_once()
+    call = post.call_args
+    assert call.args[0] == "http://moss.test:8898/v1/audio/transcriptions"
+    assert call.kwargs["data"] == {
+        "model": "meeting-model",
+        "response_format": "diarized_json",
+        "temperature": "0",
+        "max_completion_tokens": "1234",
+    }
+    assert call.kwargs["timeout"] == 37.5
+    assert call.kwargs["headers"] == {"Authorization": "Bearer test-only-key"}
+    filename, content, mime = call.kwargs["files"]["file"]
+    assert filename == "meeting.wav" and mime in {"audio/wav", "audio/x-wav"}
+    assert content.getvalue() == args.audio.read_bytes()
+    assert "test-only-key" not in output.read_text()
+    response.raise_for_status.assert_called_once()
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        [],
+        {"text": "missing segments"},
+        {
+            "text": "broken",
+            "segments": [{"start": 2, "end": 1, "text": "broken", "speaker": "S01"}],
+        },
+        {"text": "broken", "segments": [{"start": 0, "end": 1, "text": "broken"}]},
+    ],
+)
+def test_moss_malformed_response_does_not_write_success(moss_run, body):
+    example, args, response, post = moss_run
+    response.json.return_value = body
+    args.output = args.audio.parent / "existing.json"
+    args.output.write_text("old result")
+    with pytest.raises(RuntimeError):
+        example.run(args)
+    assert args.output.read_text() == "old result"
+    post.assert_called_once()
+
+
+def test_moss_http_failure_does_not_create_output(moss_run):
+    import requests
+
+    example, args, response, post = moss_run
+    response.raise_for_status.side_effect = requests.HTTPError("503")
+    with pytest.raises(requests.HTTPError):
+        example.run(args)
+    assert not args.audio.with_suffix(".moss-vllm.json").exists()
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("vllm_base_url", None),
+        ("request_timeout", 0),
+        ("request_timeout", float("nan")),
+        ("request_timeout", float("inf")),
+        ("max_completion_tokens", 0),
+        ("served_model", ""),
+        ("language", "Chinese"),
+    ],
+)
+def test_moss_rejects_invalid_configuration_before_request(moss_run, field, value):
+    example, args, _, post = moss_run
+    setattr(args, field, value)
+    with pytest.raises(ValueError):
+        example.run(args)
+    post.assert_not_called()
+
+
+@pytest.mark.parametrize("alias", ["direct", "symlink", "hardlink"])
+def test_moss_rejects_input_as_output_before_request(moss_run, alias):
+    example, args, _, post = moss_run
+    original = args.audio.read_bytes()
+    args.output = args.audio
+    if alias != "direct":
+        args.output = args.audio.parent / "output.json"
+        if alias == "symlink":
+            args.output.symlink_to(args.audio)
+        else:
+            args.output.hardlink_to(args.audio)
+    with pytest.raises(ValueError, match="output"):
+        example.run(args)
+    assert args.audio.read_bytes() == original
+    post.assert_not_called()
+
+
+def test_moss_http_options_require_explicit_engine(moss_run):
+    example, args, _, post = moss_run
+    args.engine = "qwen3"
+    with pytest.raises(ValueError, match="--engine moss"):
+        example.run(args)
+    post.assert_not_called()
+
+
+def test_moss_main_displays_anonymous_speakers(moss_run, monkeypatch, capsys):
+    example, args, _, _ = moss_run
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            str(SCRIPT_PATH),
+            str(args.audio),
+            "--engine",
+            "moss",
+            "--vllm-base-url",
+            args.vllm_base_url,
+        ],
+    )
+    example.main()
+    out = capsys.readouterr().out
+    assert "S01" in out and "S02" in out and "2 segments" in out
+    assert "test-only-key" not in out
+
+
+def test_parser_exposes_explicit_moss_mode_without_changing_qwen_defaults():
+    parser = load_example().build_parser()
+    original = parser.parse_args(["input.wav"])
+    assert original.engine == "qwen3"
+    assert original.model == "Qwen/Qwen3-ASR-1.7B"
+    moss = parser.parse_args(
+        [
+            "input.wav",
+            "--engine",
+            "moss",
+            "--vllm-base-url",
+            "http://localhost:8898/v1",
+            "--served-model",
+            "demo",
+            "--request-timeout",
+            "123",
+            "--max-completion-tokens",
+            "4321",
+        ]
+    )
+    assert (
+        moss.engine,
+        moss.served_model,
+        moss.request_timeout,
+        moss.max_completion_tokens,
+    ) == ("moss", "demo", 123, 4321)
+
+
+def test_default_qwen_run_keeps_native_configuration_and_output(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+    import numpy as np
+    import soundfile as sf
+
+    example = load_example()
+    audio = tmp_path / "source.wav"
+    sf.write(audio, np.zeros(1600), 16000)
+    args = example.build_parser().parse_args([str(audio)])
+    llm = MagicMock(return_value=FakeModel())
+    splitter = MagicMock(return_value=[(np.zeros(1600), 0.0)])
+    monkeypatch.setitem(
+        sys.modules, "qwen_asr", SimpleNamespace(Qwen3ASRModel=SimpleNamespace(LLM=llm))
+    )
+    monkeypatch.setitem(
+        sys.modules,
+        "qwen_asr.inference.utils",
+        SimpleNamespace(split_audio_into_chunks=splitter),
+    )
+    monkeypatch.setattr(
+        example,
+        "_convert_to_mono_wav",
+        lambda src, dest: dest.write_bytes(src.read_bytes()),
+    )
+    output, payload = example.run(args)
+    assert output == audio.with_suffix(".qwen3-vllm.json")
+    assert payload == {
+        "text": "chunk-1",
+        "language": "Chinese",
+        "segments": [
+            {"start_ms": 0, "end_ms": 100, "text": "chunk-1", "language": "Chinese"}
+        ],
+    }
+    llm.assert_called_once_with(
+        model="Qwen/Qwen3-ASR-1.7B",
+        gpu_memory_utilization=0.8,
+        max_inference_batch_size=4,
+    )
+    assert splitter.call_args.kwargs == {"max_chunk_sec": 180.0}
+
+
+@pytest.mark.parametrize("http_status", [200, 503])
+def test_moss_cli_over_real_local_http(tmp_path, http_status):
+    import os
+    import subprocess
+    import threading
+    import wave
+    from email import policy
+    from email.parser import BytesParser
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+
+    audio = tmp_path / "whole-recording.wav"
+    with wave.open(str(audio), "wb") as source:
+        source.setnchannels(1)
+        source.setsampwidth(2)
+        source.setframerate(16000)
+        source.writeframes(b"\x00\x00" * 1600)
+    received = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self):
+            body = self.rfile.read(int(self.headers["Content-Length"]))
+            message = BytesParser(policy=policy.default).parsebytes(
+                ("Content-Type: " + self.headers["Content-Type"] + "\r\n\r\n").encode()
+                + body
+            )
+            fields = {
+                part.get_param("name", header="content-disposition"): part.get_payload(
+                    decode=True
+                )
+                for part in message.iter_parts()
+            }
+            received.append((self.path, self.headers.get("Authorization"), fields))
+            self.send_response(http_status)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(
+                json.dumps(
+                    {
+                        "text": "fixture",
+                        "segments": [
+                            {
+                                "start": 0.01,
+                                "end": 0.09,
+                                "text": "fixture",
+                                "speaker": "S02",
+                            }
+                        ],
+                    }
+                ).encode()
+            )
+
+        def log_message(self, *args):
+            pass
+
+    with HTTPServer(("127.0.0.1", 0), Handler) as server:
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        env = os.environ.copy()
+        env.update(
+            PYTHONPATH=str(ROOT),
+            MOSS_VLLM_API_KEY="http-fixture-key",
+            CUDA_VISIBLE_DEVICES="",
+            HF_HUB_OFFLINE="1",
+            TRANSFORMERS_OFFLINE="1",
+        )
+        try:
+            process = subprocess.run(
+                [
+                    sys.executable,
+                    str(SCRIPT_PATH),
+                    str(audio),
+                    "--engine",
+                    "moss",
+                    "--vllm-base-url",
+                    f"http://127.0.0.1:{server.server_port}/v1",
+                    "--served-model",
+                    "fixture-model",
+                    "--max-completion-tokens",
+                    "256",
+                    "--request-timeout",
+                    "5",
+                ],
+                cwd=tmp_path,
+                env=env,
+                capture_output=True,
+                text=True,
+                timeout=45,
+            )
+        finally:
+            server.shutdown()
+            thread.join(timeout=5)
+        assert not thread.is_alive()
+    assert len(received) == 1
+    path, authorization, fields = received[0]
+    assert path == "/v1/audio/transcriptions"
+    assert authorization == "Bearer http-fixture-key"
+    assert fields["file"] == audio.read_bytes()
+    assert fields["model"] == b"fixture-model"
+    assert fields["response_format"] == b"diarized_json"
+    assert fields["max_completion_tokens"] == b"256"
+    assert "http-fixture-key" not in process.stdout + process.stderr
+    output = audio.with_suffix(".moss-vllm.json")
+    if http_status == 200:
+        assert process.returncode == 0, process.stderr
+        result = json.loads(output.read_text())
+        assert result["timestamp"] == [[10, 90]]
+        assert result["sentence_info"][0]["spk"] == "S02"
+        assert "S02" in process.stdout
+    else:
+        assert process.returncode != 0
+        assert not output.exists()
+
+
+@pytest.mark.parametrize("suffix", ["", "_en"])
+def test_notes_provide_separate_moss_recipe_and_output_contract(suffix):
+    notes = SCRIPT_PATH.with_name(
+        f"transcribe_vllm_offline_notes{suffix}.md"
+    ).read_text()
+    for required in [
+        "--engine moss",
+        "--vllm-base-url",
+        "--served-model",
+        "--max-completion-tokens",
+        "MOSS_VLLM_API_KEY",
+        "sentence_info",
+        "raw_text",
+        ".moss-vllm.json",
+        "0.14.0",
+        "0.27.1",
+        "docs/moss_transcribe_diarize",
+        "8192",
+    ]:
+        assert required in notes
+
+
 if __name__ == "__main__":
     unittest.main()