@@ -189,6 +189,7 @@ Also registered with `T5ForConditionalGeneration` (task: `seq2seq`):
 | `moonshine_streaming` | `MoonshineStreamingForConditionalGeneration` | `speech-to-text` | `moonshine-ai/moonshine-streaming-tiny` |
 | `qwen3_asr` | `Qwen3ASRForConditionalGeneration` | `speech-language` | — |
 | `qwen3_forced_aligner` | `Qwen3ASRForConditionalGeneration` | `speech-language` | — |
+| `VibeVoiceForASRStreamingTraining` | `VibeVoiceASRStreamingForConditionalGeneration` | `vibevoice-asr-streaming` | `microsoft/VibeVoice-ASR-Streaming-1.5B`, `microsoft/VibeVoice-ASR-Streaming-7B` |
 
 ### Text-to-Speech
 
 
@@ -112,6 +112,7 @@
 )
 from mobius._configs._world_model import WorldModelConfig
 from mobius._configs.vibevoice import (
+    VibeVoiceASRStreamingConfig,
     VibeVoiceConfig,
     VibeVoiceDiffusionConfig,
     VibeVoiceStreamingConfig,
@@ -184,8 +185,9 @@
     "TTSConfig",
     "VisionConfig",
     "VisionLanguageConfig",
-    "VibeVoiceConfig",
     "VibeVoiceASRConfig",
+    "VibeVoiceASRStreamingConfig",
+    "VibeVoiceConfig",
     "VibeVoiceDiffusionConfig",
     "VibeVoiceStreamingConfig",
     "VibeVoiceStreamingDiffusionConfig",
 
@@ -31,6 +31,7 @@ class VibeVoiceTokenizerConfig:
     rms_norm_eps: float = 1e-5
     layer_scale_init_value: float = 1e-6
     vae_std: float = 0.625
+    std_dist_type: str = "gaussian"
 
     @property
     def hop_length(self) -> int:
@@ -372,3 +373,141 @@ def validate(self) -> None:
             raise ValueError(
                 "VibeVoice Realtime diffusion latent_size must match acoustic vae_dim"
             )
+
+
+def _asr_streaming_tokenizer_config(
+    config, *, default_hidden_size: int
+) -> VibeVoiceTokenizerConfig:
+    """Extract the streaming-ASR encoder schema into the shared causal-tokenizer geometry."""
+    source = _as_attribute_config(config)
+    encoder_ratios = list(getattr(source, "encoder_ratios", [8, 5, 5, 4, 2, 2]))
+    required_values = {
+        "channels": (getattr(source, "channels", 1), 1),
+        "causal": (getattr(source, "causal", True), True),
+        "mixer_layer": (getattr(source, "mixer_layer", "depthwise_conv"), "depthwise_conv"),
+        "conv_norm": (getattr(source, "conv_norm", "none"), "none"),
+        "pad_mode": (getattr(source, "pad_mode", "constant"), "constant"),
+        "disable_last_norm": (getattr(source, "disable_last_norm", True), True),
+        "layernorm": (getattr(source, "layernorm", "RMSNorm"), "RMSNorm"),
+        "layernorm_elementwise_affine": (
+            getattr(source, "layernorm_elementwise_affine", True),
+            True,
+        ),
+        "conv_bias": (getattr(source, "conv_bias", True), True),
+    }
+    unsupported = [
+        f"{name}={actual!r}"
+        for name, (actual, expected) in required_values.items()
+        if actual != expected
+    ]
+    if unsupported:
+        raise ValueError(
+            "Unsupported VibeVoice streaming ASR tokenizer configuration: "
+            + ", ".join(unsupported)
+            + "."
+        )
+    if not encoder_ratios or any(int(ratio) <= 0 for ratio in encoder_ratios):
+        raise ValueError(
+            "VibeVoice streaming ASR encoder_ratios must contain positive integers."
+        )
+    return VibeVoiceTokenizerConfig(
+        channels=int(getattr(source, "channels", 1)),
+        hidden_size=int(getattr(source, "vae_dim", default_hidden_size)),
+        kernel_size=int(getattr(source, "kernel_size", 7)),
+        num_filters=int(getattr(source, "encoder_n_filters", 32)),
+        downsampling_ratios=list(reversed(encoder_ratios)),
+        depths=[
+            int(depth)
+            for depth in str(getattr(source, "encoder_depths", "3-3-3-3-3-3-8")).split("-")
+        ],
+        ffn_expansion=int(getattr(source, "ffn_expansion", 4)),
+        hidden_act=str(getattr(source, "hidden_act", "gelu")),
+        rms_norm_eps=float(getattr(source, "layernorm_eps", 1e-5)),
+        layer_scale_init_value=float(getattr(source, "layer_scale_init_value", 1e-6)),
+        vae_std=float(getattr(source, "fix_std", 0.625)),
+        std_dist_type=str(getattr(source, "std_dist_type", "gaussian")),
+    )
+
+
+@dataclasses.dataclass
+class VibeVoiceASRStreamingConfig(ArchitectureConfig):
+    """Configuration for VibeVoice streaming ASR's causal audio encoders and Qwen2 decoder."""
+
+    acoustic_tokenizer: VibeVoiceTokenizerConfig = dataclasses.field(
+        default_factory=VibeVoiceTokenizerConfig
+    )
+    semantic_tokenizer: VibeVoiceTokenizerConfig = dataclasses.field(
+        default_factory=lambda: VibeVoiceTokenizerConfig(hidden_size=128, std_dist_type="none")
+    )
+    speech_start_token_id: int = 151646
+    speech_end_token_id: int = 151647
+    speech_placeholder_token_id: int = 151648
+    text_chunk_end_token_id: int = 151665
+    compression_ratio: int = 3200
+    sampling_rate: int = 24_000
+    chunk_frames: int = 22
+    lookahead_frames: int = 4
+
+    @classmethod
+    def from_transformers(
+        cls,
+        config,
+        parent_config=None,
+        *,
+        allow_block_fp8_dense_fallback: bool = False,
+    ) -> VibeVoiceASRStreamingConfig:
+        """Extract the public streaming-ASR composite while preserving Qwen2 decoder config."""
+        parent = _as_attribute_config(parent_config or config)
+        decoder = _as_attribute_config(getattr(parent, "decoder_config", config))
+        result = super().from_transformers(
+            decoder,
+            parent_config=parent,
+            allow_block_fp8_dense_fallback=allow_block_fp8_dense_fallback,
+        )
+        acoustic = _asr_streaming_tokenizer_config(
+            getattr(parent, "acoustic_tokenizer_config", None),
+            default_hidden_size=64,
+        )
+        semantic = _asr_streaming_tokenizer_config(
+            getattr(parent, "semantic_tokenizer_config", None),
+            default_hidden_size=128,
+        )
+        if acoustic.std_dist_type != "gaussian" or semantic.std_dist_type != "none":
+            raise ValueError(
+                "VibeVoice streaming ASR requires a gaussian acoustic tokenizer and "
+                "a deterministic semantic tokenizer."
+            )
+        compression_ratio = int(
+            getattr(parent, "speech_tok_compress_ratio", acoustic.hop_length)
+        )
+        if acoustic.hop_length != compression_ratio:
+            raise ValueError(
+                "VibeVoice streaming ASR acoustic tokenizer ratios must match the "
+                "configured speech tokenizer compression ratio."
+            )
+        if semantic.hop_length != acoustic.hop_length:
+            raise ValueError(
+                "VibeVoice streaming ASR acoustic and semantic tokenizer hop lengths must match."
+            )
+        # The composite ASR config controls the tokenizer and connector precision,
+        # while decoder_config independently records Qwen's checkpoint storage dtype.
+        pipeline_dtype = _resolve_dtype_value(getattr(parent, "dtype", None))
+        if pipeline_dtype is None:
+            pipeline_dtype = _resolve_dtype_value(getattr(parent, "torch_dtype", None))
+        return dataclasses.replace(
+            result,
+            model_type="vibevoice_asr_streaming",
+            acoustic_tokenizer=acoustic,
+            semantic_tokenizer=semantic,
+            speech_start_token_id=int(getattr(parent, "object_ref_start_token_id", 151646)),
+            speech_end_token_id=int(getattr(parent, "object_ref_end_token_id", 151647)),
+            speech_placeholder_token_id=int(getattr(parent, "speech_pad_token_id", 151648)),
+            text_chunk_end_token_id=int(getattr(parent, "text_chunk_end_token_id", 151665)),
+            compression_ratio=compression_ratio,
+            sampling_rate=int(getattr(parent, "target_sample_rate", 24_000)),
+            chunk_frames=int(getattr(parent, "chunk_frames", 22)),
+            lookahead_frames=int(getattr(parent, "lookahead_frames", 4)),
+            eos_token_id=getattr(parent, "eos_token_id", 151643),
+            pad_token_id=getattr(parent, "pad_token_id", 151655),
+            dtype=pipeline_dtype or result.dtype,
+        )
@@ -48,6 +48,7 @@
     Qwen4ExpConfig,
     SenseNovaU1Config,
     VibeVoiceASRConfig,
+    VibeVoiceASRStreamingConfig,
     VibeVoiceConfig,
     VibeVoiceStreamingConfig,
     WhisperConfig,
@@ -162,6 +163,7 @@
     SmolLM3CausalLMModel,
     SortformerDiarizationModel,
     VibeVoiceASRForConditionalGeneration,
+    VibeVoiceASRStreamingForConditionalGeneration,
     VibeVoiceForConditionalGeneration,
     VibeVoiceStreamingForConditionalGeneration,
     WhisperForConditionalGeneration,
@@ -231,7 +233,12 @@
 from mobius.models.t5 import T5EncoderModel, T5ForConditionalGeneration
 from mobius.models.talkie import TalkieForCausalLM
 from mobius.models.trocr import TrOCRForConditionalGeneration
-from mobius.models.vibevoice import VIBEVOICE_MODEL_ID, VIBEVOICE_REVISION
+from mobius.models.vibevoice import (
+    VIBEVOICE_ASR_STREAMING_MODEL_ID,
+    VIBEVOICE_ASR_STREAMING_REVISION,
+    VIBEVOICE_MODEL_ID,
+    VIBEVOICE_REVISION,
+)
 from mobius.models.vibevoice_asr import VIBEVOICE_ASR_MODEL_ID, VIBEVOICE_ASR_REVISION
 from mobius.models.vibevoice_streaming import (
     VIBEVOICE_STREAMING_MODEL_ID,
@@ -941,6 +948,15 @@ def _detect_fallback_registration(hf_config) -> ModelRegistration | None:
         family="vibevoice",
         variant="offline-asr",
     ),
+    "VibeVoiceForASRStreamingTraining": ModelRegistration(
+        VibeVoiceASRStreamingForConditionalGeneration,
+        task="vibevoice-asr-streaming",
+        config_class=VibeVoiceASRStreamingConfig,
+        test_model_id=VIBEVOICE_ASR_STREAMING_MODEL_ID,
+        test_revision=VIBEVOICE_ASR_STREAMING_REVISION,
+        family="vibevoice",
+        variant="streaming-asr",
+    ),
     "whisper": ModelRegistration(
         WhisperForConditionalGeneration,
         task="speech-to-text",
 
@@ -769,7 +769,7 @@ def _looks_like_vibevoice_tts(pkg: Any) -> bool:
     } <= names
 
 
-def _looks_like_vibevoice_asr(pkg: Any) -> bool:
+def _looks_like_vibevoice_offline_asr(pkg: Any) -> bool:
     """Detect the offline VibeVoice-ASR dual-encoder component topology."""
     try:
         names = set(pkg.keys())
@@ -784,6 +784,23 @@ def _looks_like_vibevoice_asr(pkg: Any) -> bool:
     } <= names
 
 
+def _looks_like_vibevoice_streaming_asr(pkg: Any) -> bool:
+    """Detect the stateful three-stage VibeVoice streaming-ASR topology."""
+    try:
+        if set(pkg.keys()) != {"audio_encoder", "embedding", "decoder"}:
+            return False
+        audio_inputs = {value.name for value in pkg["audio_encoder"].graph.inputs}
+    except AttributeError:
+        return False
+    return {
+        "speech_tensors",
+        "speech_masks",
+        "is_final_chunk",
+        "past_acoustic_conv.0",
+        "past_semantic_conv.0",
+    } <= audio_inputs
+
+
 def _write_vibevoice_asr_processor_contract(output_dir: str, config: Any) -> str:
     """Materialize the source processor protocol absent from the ASR checkpoint."""
     os.makedirs(output_dir, exist_ok=True)
@@ -1039,6 +1056,43 @@ def write_onnx_genai_config(
             "runtime configuration is claimed."
         )
         return _write_advisory_component_contract(pkg, output_dir, warning=warning)
+    if _looks_like_vibevoice_streaming_asr(pkg):
+        if kv_native_dtype is not None:
+            raise ValueError(
+                "workflow VibeVoice streaming ASR export derives KV and convolution state "
+                "dtypes from ONNX ports; kv_native_dtype overrides are unsupported"
+            )
+        artifacts = _write_text_runtime_assets(output_dir, source, revision=revision)
+        artifacts.update(
+            _copy_runtime_assets(
+                output_dir,
+                source,
+                (
+                    "processor_config.json",
+                    "preprocessor_config.json",
+                    "generation_config.json",
+                ),
+                revision=revision,
+            )
+        )
+        audio_processor_path = _write_hf_audio_processor(output_dir, source, revision=revision)
+        if audio_processor_path is not None:
+            artifacts["audio_processor"] = audio_processor_path
+        artifacts.update(
+            _write_advisory_component_contract(
+                pkg,
+                output_dir,
+                warning=(
+                    "The tested onnx-genai runtime cannot orchestrate VibeVoice streaming "
+                    "ASR's dual causal convolution states, flattened speech-placeholder "
+                    "replacement with arbitrary left-padded attention masks, forced "
+                    "<|text_chunk_end|> control tokens, or host-side hotword and speaker "
+                    "JSON handling. Exact graph and pinned processor contracts are exported "
+                    "without claiming downstream runtime support."
+                ),
+            )
+        )
+        return artifacts
     decoder = pkg.get("decoder") or pkg.get("model")
     decoder_inputs = (
         {value.name for value in decoder.graph.inputs} if decoder is not None else set()
@@ -1376,7 +1430,7 @@ def write_onnx_genai_config(
         )
         return artifacts
 
-    if _looks_like_vibevoice_asr(pkg):
+    if _looks_like_vibevoice_offline_asr(pkg):
         if kv_native_dtype is not None:
             raise ValueError(
                 "VibeVoice-ASR derives KV and convolution state dtypes from ONNX ports; "
 
@@ -178,6 +178,77 @@ def test_vibevoice_asr_writes_processor_and_advisory_contract(monkeypatch, tmp_p
     assert compatibility["runtime_validation_status"] == "unsupported-by-tested-runtime"
 
 
+@pytest.mark.parametrize(
+    ("model_id", "revision"),
+    (
+        ("microsoft/VibeVoice-ASR-Streaming-1.5B", "4262d23d8a539a6530cf64fbd0b1751ef9a30853"),
+        ("microsoft/VibeVoice-ASR-Streaming-7B", "60d858b518b4e19d404af3737f848fc185b30177"),
+    ),
+)
+def test_vibevoice_streaming_asr_writes_pinned_advisory_contract(
+    monkeypatch, tmp_path, model_id, revision
+):
+    from mobius.integrations.onnx_genai import auto_export
+    from mobius.models.vibevoice import (
+        VibeVoiceASRStreamingForConditionalGeneration,
+    )
+    from mobius.models.vibevoice_asr_test import _streaming_config
+    from mobius.tasks import VibeVoiceASRStreamingTask
+
+    config = _streaming_config()
+    package = VibeVoiceASRStreamingTask().build(
+        VibeVoiceASRStreamingForConditionalGeneration(config), config
+    )
+    calls: list[tuple[str, str | None]] = []
+
+    def text_assets(output_dir, source, *, revision=None):
+        calls.append(("text", revision))
+        return {"tokenizer": str(Path(output_dir) / "tokenizer.json")}
+
+    def runtime_assets(output_dir, source, names, *, revision=None):
+        calls.append(("runtime", revision))
+        assert names == (
+            "processor_config.json",
+            "preprocessor_config.json",
+            "generation_config.json",
+        )
+        return {"processor_config": str(Path(output_dir) / "processor_config.json")}
+
+    def audio_processor(output_dir, source, *, revision=None):
+        calls.append(("audio", revision))
+        return str(Path(output_dir) / "audio_processor.json")
+
+    monkeypatch.setattr(auto_export, "_write_text_runtime_assets", text_assets)
+    monkeypatch.setattr(auto_export, "_copy_runtime_assets", runtime_assets)
+    monkeypatch.setattr(auto_export, "_write_hf_audio_processor", audio_processor)
+    monkeypatch.setattr(
+        auto_export,
+        "_write_advisory_component_contract",
+        lambda *args, **kwargs: {
+            "inference_metadata": str(tmp_path / "inference_metadata.yaml")
+        },
+    )
+
+    artifacts = write_onnx_genai_config(
+        package,
+        str(tmp_path),
+        source=model_id,
+        revision=revision,
+    )
+
+    assert calls == [
+        ("text", revision),
+        ("runtime", revision),
+        ("audio", revision),
+    ]
+    assert {
+        "tokenizer",
+        "processor_config",
+        "audio_processor",
+        "inference_metadata",
+    } <= set(artifacts)
+
+
 def _video_diffusion_package() -> ModelPackage:
     latent = ["batch", "frames", 4, "height", "width"]
     transformer = _model(