[Config] Disable Async Chunk for Unsupported Pipelines (#5099)

Signed-off-by: Alex Brooks <albrooks@redhat.com>

tests/config/test_config_factory.py

@@ -1405,6 +1405,21 @@ class TestDeployConfigLoading:
         with pytest.raises(ValueError, match=r"stage_args.*PipelineConfig.*stages"):
             load_deploy_config(deploy_path)
 
+    @pytest.mark.parametrize(
+        ("yaml_val", "expected"),
+        [
+            ('"2"', 2),
+            ('""', 0),
+            ("null", 0),
+        ],
+    )
+    def test_coerces_active_stream_window_to_int(self, tmp_path, yaml_val, expected):
+        """Ensure active stream window coerces str to int."""
+        deploy_path = tmp_path / "deploy.yaml"
+        deploy_path.write_text(f"active_stream_window: {yaml_val}\n", encoding="utf-8")
+        deploy = load_deploy_config(deploy_path)
+        assert deploy.active_stream_window == expected
+
     @pytest.mark.parametrize(
         ("filename", "max_sessions"),
         [
@@ -3511,14 +3526,10 @@ class TestSentinelDefaultPrecedence:
         assert stage1.sync_process_input_func is not None
         assert stage1.sync_process_input_func.endswith("thinker2talker_token_only")
 
-        # async_chunk=True must now be rejected: removing the fake hook means
-        # there is no next-stage input processor for the validator to accept.
-        # (Positive consequence -- users can't accidentally enable async_chunk
-        # on an arch that doesn't actually support it.)
-        import pytest as _pytest
-
-        with _pytest.raises(ValueError, match="async_chunk=True"):
-            merge_pipeline_deploy(pipeline, DeployConfig(async_chunk=True))
+        # ensure merging the pipeline deploy with async chunk set to True disables it,
+        # since currently it is not supported for this pipeline.
+        stage_configs = merge_pipeline_deploy(pipeline, DeployConfig(async_chunk=True))
+        assert all([not stg_cfg.yaml_engine_args["async_chunk"] for stg_cfg in stage_configs])
 
         # async_chunk=False merges cleanly and stage-0 yaml_engine_args carries
         # no spurious full-payload hook.
@@ -3687,3 +3698,71 @@ class TestObjectStorageConfigResolution:
 
         matching = "s3://any-bucket/my-cosyvoice3-model"
         assert StageConfigFactory.try_infer_model_type(model=matching, trust_remote_code=False) == "cosyvoice3"
+
+
+class TestAsyncChunkDefaults:
+    def test_async_chunk_auto_disabled_without_processor(self):
+        """Ensure a multi-stage model that doesn't support async chunk turns it off by default."""
+        pipeline = PipelineConfig(
+            model_type="test_no_async",
+            model_arch="TestNoAsync",
+            stages=(
+                StagePipelineConfig(
+                    stage_id=0,
+                    model_stage="ar",
+                    execution_type=StageExecutionType.LLM_AR,
+                    final_output=True,
+                ),
+                StagePipelineConfig(
+                    stage_id=1,
+                    model_stage="generation",
+                    execution_type=StageExecutionType.LLM_GENERATION,
+                    input_sources=(0,),
+                ),
+            ),
+        )
+
+        deploy = DeployConfig()
+        # async chunk should not try to default to True in this case,
+        # since doing so will just raise a ValueError in validation.
+        merge_pipeline_deploy(pipeline, deploy)
+        assert not deploy.async_chunk
+
+    def test_async_chunk_auto_disabled_when_yaml_omits_key(self, tmp_path):
+        """Ensure a multi-stage model that doesn't support async chunk turns it off by default
+        when a deploy config is provided that doesn't explicitly set it."""
+
+        deploy_path = tmp_path / "no_async_chunk.yaml"
+        deploy_path.write_text(
+            """
+stages:
+  - stage_id: 0
+    devices: "0"
+  - stage_id: 1
+    devices: "0"
+""",
+            encoding="utf-8",
+        )
+        deploy = load_deploy_config(deploy_path)
+
+        pipeline = PipelineConfig(
+            model_type="test_no_async",
+            model_arch="TestNoAsync",
+            stages=(
+                StagePipelineConfig(
+                    stage_id=0,
+                    model_stage="ar",
+                    execution_type=StageExecutionType.LLM_AR,
+                    final_output=True,
+                ),
+                StagePipelineConfig(
+                    stage_id=1,
+                    model_stage="generation",
+                    execution_type=StageExecutionType.LLM_GENERATION,
+                    input_sources=(0,),
+                ),
+            ),
+        )
+
+        merge_pipeline_deploy(pipeline, deploy)
+        assert not deploy.async_chunk

tests/config/test_omni_config.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import logging
 import warnings
 from dataclasses import fields, replace
 from inspect import Parameter, signature
@@ -54,6 +55,7 @@ from vllm_omni.config.stage_config import (
     load_deploy_config,
     merge_pipeline_deploy,
     resolve_deploy_yaml,
+    update_deploy_config_async_chunk_enabled,
 )
 from vllm_omni.diffusion.diffusion_kv.config import DiffusionKVCacheMode
 from vllm_omni.engine.stage_engine_startup import _serialize_stage_config
@@ -64,22 +66,27 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 _DEPLOY_DIR = Path(__file__).parents[2] / "vllm_omni" / "deploy"
 
 
-@pytest.mark.parametrize("async_chunk", [False, True])
-def test_native_kv_transfer_requires_completed_ar_stage(async_chunk):
-    from types import SimpleNamespace
-
-    pipeline = SimpleNamespace(stages=(), model_type="test")
+@pytest.mark.parametrize("async_chunk", [None, False, True])
+def test_native_kv_transfer_disables_async_chunk(async_chunk):
+    """Ensure that a multistage pipeline that supports async chunk disables it if we have a kv transfer config."""
+    pipeline = PipelineConfig(
+        model_type="test",
+        stages=(
+            # async_chunk_process_next_stage_input_func doesn't matter, it just needs to exist for this test.
+            StagePipelineConfig(stage_id=0, model_stage="ar", async_chunk_process_next_stage_input_func="foo.bar"),
+            StagePipelineConfig(stage_id=1, model_stage="dit", input_sources=(0,), final_output=True),
+        ),
+    )
+    kv_transfer_config = {"kv_connector": "MooncakeConnector"}
     deploy = DeployConfig(
         async_chunk=async_chunk,
         stages=[
-            StageDeployConfig(stage_id=0, engine_extras={"kv_transfer_config": {"kv_connector": "MooncakeConnector"}})
+            StageDeployConfig(stage_id=0, engine_extras={"kv_transfer_config": kv_transfer_config}),
+            StageDeployConfig(stage_id=1, engine_extras={"kv_transfer_config": kv_transfer_config}),
         ],
     )
-    if async_chunk:
-        with pytest.raises(ValueError, match="requires async_chunk=False"):
-            omni_config_module._validate_async_chunk_support(pipeline, deploy)
-    else:
-        omni_config_module._validate_async_chunk_support(pipeline, deploy)
+    update_deploy_config_async_chunk_enabled(pipeline, deploy)
+    assert not deploy.async_chunk
 
 
 @pytest.fixture(autouse=True)
@@ -780,6 +787,33 @@ def test_from_pipeline_config_dispatches_async_chunk_processors_without_mutating
     assert pipeline.get_stage(1).custom_process_input_func is None
 
 
+def test_from_pipeline_config_async_chunk_none_preserves_yaml_default():
+    """Ensure passing None for async chunk in when creating from pipeline keeps yaml defaults."""
+    config = _from_pipeline_key("qwen3_tts", cli_overrides={"async_chunk": None})
+    talker_cfg = config.stage_by_id(0)
+    custom_proc = talker_cfg.custom_process_next_stage_input_func
+    assert custom_proc is not None
+    assert custom_proc.endswith("talker2code2wav_async_chunk")
+
+
+def test_from_pipeline_config_warns_when_single_stage_async_chunk_enabled(caplog: pytest.LogCaptureFixture):
+    """Ensure that we get a warning indicating async chunk is disabled if requested for single stage models."""
+    pipeline = PipelineConfig(
+        model_type="single_stage",
+        stages=(StagePipelineConfig(stage_id=0, model_stage="stage", final_output=True),),
+    )
+
+    with caplog.at_level(logging.WARNING, logger="vllm_omni.config.stage_config"):
+        config = VllmOmniConfig.from_pipeline_config(pipeline, cli_overrides={"async_chunk": True})
+
+    assert config.stage_by_id(0).connector_config.async_chunk is False
+    assert any(
+        record.levelno == logging.WARNING
+        and "async chunk is inapplicable to single stage models" in record.getMessage()
+        for record in caplog.records
+    )
+
+
 def test_joyai_code2wav_waits_for_full_payload():
     config = _from_pipeline_key("joyai_vl_interaction")
     talker = config.stage_by_id(1)
@@ -2285,3 +2319,31 @@ def test_platform_stage_overlay_rejects_invalid_model_runner():
     )
     with pytest.raises(ValueError, match="model_runner must be 'v1' or 'v2'"):
         _apply_platform_overrides(deploy, platform="cuda")
+
+
+def test_async_chunk_auto_disabled_without_processor():
+    """Ensure a multi-stage model that doesn't support async chunk turns it off by default."""
+    pipeline = PipelineConfig(
+        model_type="test_no_async",
+        model_arch="TestNoAsync",
+        stages=(
+            StagePipelineConfig(
+                stage_id=0,
+                model_stage="ar",
+                execution_type=StageExecutionType.LLM_AR,
+                final_output=True,
+            ),
+            StagePipelineConfig(
+                stage_id=1,
+                model_stage="generation",
+                execution_type=StageExecutionType.LLM_GENERATION,
+                input_sources=(0,),
+            ),
+        ),
+    )
+
+    deploy = DeployConfig()
+    # async chunk should not try to default to True in this case,
+    # since doing so will just raise a ValueError in validation.
+    merge_pipeline_deploy(pipeline, deploy)
+    assert not deploy.async_chunk

vllm_omni/config/config_factory.py

@@ -514,7 +514,7 @@ class StageConfigFactory:
 
         pipeline_cfg, deploy_cfg = inject_forced_aligner_stage(pipeline_cfg, deploy_cfg, cli_overrides)
 
-        stages = merge_pipeline_deploy(pipeline_cfg, deploy_cfg, cli_overrides)
+        stages = merge_pipeline_deploy(pipeline_cfg, deploy_cfg)
 
         # Overlay declarative parallel strategies (opt-in) before CLI overrides.
         applied = cls._apply_strategy_specs(stages, strategy_specs)

vllm_omni/config/omni_config.py

@@ -58,8 +58,8 @@ from vllm_omni.config.stage_config import (
     reconcile_diffusion_attention_overrides,
     resolve_stage_async_chunk,
     resolve_stage_model_runner,
+    update_deploy_config_async_chunk_enabled,
     validate_native_mrv2_session,
-    validate_stage_async_chunk_edges,
 )
 from vllm_omni.diffusion.diffusion_kv.config import DiffusionKVCacheMode
 
@@ -336,23 +336,6 @@ def _first_defined(*values: Any) -> Any:
     return None
 
 
-def _validate_async_chunk_support(pipeline: PipelineConfig, deploy: DeployConfig) -> None:
-    has_inter_stage_edges = any(stage.input_sources for stage in pipeline.stages)
-    if deploy.async_chunk and any(stage.engine_extras.get("kv_transfer_config") for stage in deploy.stages):
-        raise ValueError("Native AR-to-DiT KV transfer requires async_chunk=False.")
-    if (
-        deploy.async_chunk
-        and has_inter_stage_edges
-        and not any(stage.async_chunk_process_next_stage_input_func for stage in pipeline.stages)
-    ):
-        raise ValueError(
-            f"Pipeline {pipeline.model_type!r} has async_chunk=True in deploy but no stage "
-            "declares a dedicated async-chunk next-stage processor "
-            "(``async_chunk_process_next_stage_input_func``). "
-            "Either set async_chunk=False or implement an async-chunk producer on the pipeline."
-        )
-
-
 def _resolve_execution_mode(execution_type: StageExecutionType) -> tuple[StageType, str | None]:
     try:
         return _EXECUTION_TYPE_TO_STAGE_WORKER[execution_type]
@@ -2245,23 +2228,21 @@ class VllmOmniConfig:
             deploy_config_path,
         )
 
-        if cli_overrides.get("async_chunk") is not None:
-            deploy.async_chunk = bool(cli_overrides["async_chunk"])
         for name in _PIPELINE_DEPLOY_CLI_FIELDS:
             if cli_overrides.get(name) is not None:
                 setattr(deploy, name, _copy_value(cli_overrides[name]))
 
         deploy = _apply_platform_overrides(deploy)
-        if len(pipeline_cfg.stages) <= 1:
-            deploy.async_chunk = False
-        _validate_async_chunk_support(pipeline_cfg, deploy)
-        validate_stage_async_chunk_edges(pipeline_cfg, deploy)
+
+        cli_async_chunk = cli_overrides.get("async_chunk")
+        if cli_async_chunk is not None:
+            deploy.async_chunk = bool(cli_async_chunk)
 
         strategy_result = None
         if strategy_specs:
             from vllm_omni.config.composable_parallel import apply_strategy_specs
 
-            strategy_stages = merge_pipeline_deploy(pipeline_cfg, copy.deepcopy(deploy), {})
+            strategy_stages = merge_pipeline_deploy(pipeline_cfg, copy.deepcopy(deploy))
             strategy_result = apply_strategy_specs(strategy_stages, strategy_specs)
             strategy_overrides: dict[str, Any] = {}
             axis_fields = {
@@ -2313,6 +2294,7 @@ class VllmOmniConfig:
                 cli_overrides["omni_lb_policy"] = strategy_result.omni_lb_policy
 
         deploy_by_id = {stage.stage_id: stage for stage in deploy.stages}
+        update_deploy_config_async_chunk_enabled(pipeline_cfg, deploy)
         model = cli_overrides.get("model")
 
         stage_configs = tuple(

vllm_omni/config/stage_config.py

@@ -581,7 +581,8 @@ class DeployConfig:
     individual ``StageDeployConfig`` entries under ``stages:``.
     """
 
-    async_chunk: bool = True
+    # Async chunk's default depends on whether or not the pipeline supports it.
+    async_chunk: bool | None = None
     session_mode: str = "turn"
     model_runner: Literal["v1", "v2"] = "v1"
     # Stage-1 active stream slots; 0 preserves legacy all-stream cycling.
@@ -627,6 +628,22 @@ _STAGE_RESERVED_KEYS = frozenset(
     }
 )
 
+# Pipeline-wide DeployConfig fields that are propagated to every stage's
+# engine args during merge. These live at top level of the deploy YAML.
+PIPELINE_WIDE_ENGINE_FIELDS: tuple[str, ...] = (
+    "trust_remote_code",
+    "distributed_executor_backend",
+    "dtype",
+    "quantization",
+    "enable_prefix_caching",
+    "enable_chunked_prefill",
+    "data_parallel_size",
+    "pipeline_parallel_size",
+    "active_stream_window",
+    "custom_voice_dir",
+)
+
+
 # Fields on StageDeployConfig that are populated from engine_args dict
 _STAGE_DEPLOY_FIELDS = {f.name: f for f in fields(StageDeployConfig) if f.name not in _STAGE_RESERVED_KEYS}
 
@@ -639,7 +656,7 @@ def deploy_runtime_override_keys() -> frozenset[str]:
     They must remain overridable even if they are also modeled on
     ``OrchestratorArgs`` for top-level CLI parsing.
     """
-    return frozenset(_STAGE_DEPLOY_FIELDS) | frozenset(_PIPELINE_WIDE_ENGINE_FIELDS)
+    return frozenset(_STAGE_DEPLOY_FIELDS) | frozenset(PIPELINE_WIDE_ENGINE_FIELDS)
 
 
 def _parse_stage_deploy(stage_data: dict[str, Any]) -> StageDeployConfig:
@@ -827,13 +844,13 @@ def load_deploy_config(path: str | Path) -> DeployConfig:
     if not isinstance(raw_dict.get("cuda_mps", False), bool):
         raise ValueError("cuda_mps must be a boolean")
 
+    # TODO (Alex): Clean this up, we should not have fallback values here
     kwargs: dict[str, Any] = {
         "speech_cache": SpeechCacheConfig(**speech_cache),
         "cuda_mps": raw_dict.get("cuda_mps", False),
-        "async_chunk": raw_dict.get("async_chunk", True),
+        "async_chunk": raw_dict.get("async_chunk"),
         "session_mode": raw_dict.get("session_mode", "turn"),
         "model_runner": model_runner,
-        "active_stream_window": int(raw_dict.get("active_stream_window", 0) or 0),
         "duplex_session": DuplexSessionRuntimeConfig(**(raw_dict.get("duplex_session") or {})),
         "connectors": raw_dict.get("connectors", None),
         "edges": raw_dict.get("edges", None),
@@ -843,19 +860,14 @@ def load_deploy_config(path: str | Path) -> DeployConfig:
     }
     # Pipeline-wide engine settings: only set if explicitly present in YAML
     # so the DeployConfig dataclass defaults take effect otherwise.
-    for name in (
-        "trust_remote_code",
-        "distributed_executor_backend",
-        "dtype",
-        "quantization",
-        "enable_prefix_caching",
-        "enable_chunked_prefill",
-        "data_parallel_size",
-        "pipeline_parallel_size",
-        "custom_voice_dir",
-    ):
+    for name in PIPELINE_WIDE_ENGINE_FIELDS:
         if name in raw_dict:
-            kwargs[name] = raw_dict[name]
+            value = raw_dict[name]
+            # NOTE: This is arguably bad behavior since it's a one-off for
+            # only this field. We should consider deprecating this coercion.
+            if name == "active_stream_window":
+                value = int(value or 0)
+            kwargs[name] = value
     return DeployConfig(**kwargs)
 
 
@@ -997,24 +1009,6 @@ def resolve_stage_async_chunk(deploy: DeployConfig, stage: StageDeployConfig | N
     return deploy.async_chunk and (stage is None or stage.async_chunk is not False)
 
 
-def validate_stage_async_chunk_edges(pipeline: PipelineConfig, deploy: DeployConfig) -> None:
-    stages = {stage.stage_id: stage for stage in pipeline.stages}
-    deploy_by_id = {stage.stage_id: stage for stage in deploy.stages}
-    for consumer in pipeline.stages:
-        for source_id in consumer.input_sources:
-            producer = stages[source_id]
-            if not producer.async_chunk_process_next_stage_input_func:
-                continue
-            producer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(source_id))
-            consumer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(consumer.stage_id))
-            if producer_async != consumer_async:
-                raise ValueError(
-                    f"Pipeline {pipeline.model_type!r} has incompatible async_chunk settings on "
-                    f"connector edge {source_id} -> {consumer.stage_id}. "
-                    "Set the same async_chunk mode on both stages or disable pipeline-wide async_chunk."
-                )
-
-
 def _select_processor_funcs(
     ps: StagePipelineConfig,
     async_chunk: bool,
@@ -1029,23 +1023,6 @@ def _select_processor_funcs(
     return input_proc, next_stage_proc
 
 
-# Pipeline-wide DeployConfig fields that are propagated to every stage's
-# engine args during merge. These live at top level of the deploy YAML.
-_PIPELINE_WIDE_ENGINE_FIELDS: tuple[str, ...] = (
-    "trust_remote_code",
-    "distributed_executor_backend",
-    "dtype",
-    "quantization",
-    "enable_prefix_caching",
-    "enable_chunked_prefill",
-    "data_parallel_size",
-    "pipeline_parallel_size",
-    "active_stream_window",
-    "custom_voice_dir",
-)
-PIPELINE_WIDE_ENGINE_FIELDS = _PIPELINE_WIDE_ENGINE_FIELDS
-
-
 def _build_engine_args(
     ps: StagePipelineConfig,
     ds: StageDeployConfig | None,
@@ -1083,7 +1060,7 @@ def _build_engine_args(
         engine_args["stage_output_payload_keys"] = tuple(ps.stage_output_payload_keys)
 
     # Pipeline-wide top-level DeployConfig settings, applied to every stage.
-    for name in _PIPELINE_WIDE_ENGINE_FIELDS:
+    for name in PIPELINE_WIDE_ENGINE_FIELDS:
         value = getattr(deploy, name)
         if value is not None:
             engine_args[name] = value
@@ -1174,51 +1151,95 @@ def _build_extras(
     return extras
 
 
+def update_deploy_config_async_chunk_enabled(
+    pipeline: PipelineConfig,
+    deploy: DeployConfig,
+) -> None:
+    """Set the pipeline wide async chunk value & validate it."""
+    deploy.async_chunk = _resolve_pipeline_async_chunk_enabled(pipeline, deploy)
+    validate_stage_async_chunk_edges(pipeline, deploy)
+
+
+def _resolve_pipeline_async_chunk_enabled(
+    pipeline: PipelineConfig,
+    deploy: DeployConfig,
+) -> bool:
+    """Given the pipeline config and deploy config, determine the value of async_chunk.
+    Note that this currently assumes the cli override has been externally handled.
+
+    The default is True if the model actually supports async chunk and is multistage,
+    and False otherwise. If the user tried to enable async chunk through the deploy
+    config, but it's inapplicable or unsupported, it will be disabled with a warning.
+    """
+    # Single stage should never use async chunk
+    if len(pipeline.stages) <= 1:
+        if deploy.async_chunk:
+            logger.warning(
+                "Deploy config set async_chunk=True, but async chunk is inapplicable "
+                "to single stage models. As such, it will be disabled."
+            )
+        return False
+
+    has_inter_stage_edges = any(stage.input_sources for stage in pipeline.stages)
+    has_next_stage_inps = any(stage.async_chunk_process_next_stage_input_func for stage in pipeline.stages)
+    has_kv_transfer_cfg = any(stage.engine_extras.get("kv_transfer_config") for stage in deploy.stages)
+    supports_async_chunk = has_inter_stage_edges and has_next_stage_inps
+
+    # If async chunk was set, make sure it's supported if
+    # requested; otherwise warn and disable it.
+    if deploy.async_chunk is not None:
+        if deploy.async_chunk:
+            if not supports_async_chunk:
+                logger.warning(
+                    "Deploy config set async_chunk=True, but the pipeline config does not support it; disabling it."
+                )
+                return False
+            if has_kv_transfer_cfg:
+                logger.warning("Native AR-to-DiT KV transfer requires async_chunk=False; async chunk will be disabled.")
+                return False
+        return deploy.async_chunk
+
+    return supports_async_chunk and not has_kv_transfer_cfg
+
+
+def validate_stage_async_chunk_edges(pipeline: PipelineConfig, deploy: DeployConfig) -> None:
+    stages = {stage.stage_id: stage for stage in pipeline.stages}
+    deploy_by_id = {stage.stage_id: stage for stage in deploy.stages}
+    for consumer in pipeline.stages:
+        for source_id in consumer.input_sources:
+            producer = stages[source_id]
+            if not producer.async_chunk_process_next_stage_input_func:
+                continue
+            producer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(source_id))
+            consumer_async = resolve_stage_async_chunk(deploy, deploy_by_id.get(consumer.stage_id))
+            if producer_async != consumer_async:
+                raise ValueError(
+                    f"Pipeline {pipeline.model_type!r} has incompatible async_chunk settings on "
+                    f"connector edge {source_id} -> {consumer.stage_id}. "
+                    "Set the same async_chunk mode on both stages or disable pipeline-wide async_chunk."
+                )
+
+
 def merge_pipeline_deploy(
     pipeline: PipelineConfig,
     deploy: DeployConfig,
-    cli_overrides: dict[str, Any] | None = None,
 ) -> list[StageConfig]:
     """Merge pipeline + deploy + platform overrides → list[StageConfig]."""
-    if cli_overrides is None:
-        cli_overrides = {}
-
     deploy = _apply_platform_overrides(deploy)
     deploy_by_id = {s.stage_id: s for s in deploy.stages}
 
-    # async_chunk is irrelevant for single-stage pipelines, so we always disable it
-    if len(pipeline.stages) <= 1:
-        deploy.async_chunk = False
+    # NOTE: The CLI override is applied to the Deployconfig for async chunk prior
+    # to this point. We are in the process of better organizing the creation of the
+    # DeployConfig/PipelineConfig, and this path will be removed with the incorporation
+    # of the OmniConfig.
+    update_deploy_config_async_chunk_enabled(pipeline, deploy)
 
-    # async_chunk only applies to multi-stage pipelines: a pipeline with no
-    # consumer stages (every stage has empty input_sources) has no inter-stage
-    # edges, so async_chunk is a no-op and we skip the check entirely.
-    # For pipelines that DO have inter-stage edges, require a dedicated per-step
-    # async producer (``async_chunk_process_next_stage_input_func``).
-    # ``custom_process_next_stage_input_func`` is the full-payload / connector-path
-    # producer and does NOT imply async_chunk support — pipelines like qwen2_5_omni
-    # and covo_audio have it but removed their consumer-side ``custom_process_input_func``
-    # because they don't support async_chunk, so accepting them here would silently
-    # miswire the consumer stage instead of raising a clear error.
-    _has_inter_stage_edges = any(ps.input_sources for ps in pipeline.stages)
-    if (
-        deploy.async_chunk
-        and _has_inter_stage_edges
-        and not any(ps.async_chunk_process_next_stage_input_func for ps in pipeline.stages)
-    ):
-        raise ValueError(
-            f"Pipeline {pipeline.model_type!r} has async_chunk=True in deploy but no stage "
-            "declares a dedicated async-chunk next-stage processor "
-            "(``async_chunk_process_next_stage_input_func``). "
-            "Either set async_chunk=False or implement an async-chunk producer on the pipeline."
-        )
-
-    validate_stage_async_chunk_edges(pipeline, deploy)
     result: list[StageConfig] = []
     for ps in pipeline.stages:
         ds = deploy_by_id.get(ps.stage_id)
         stage_type, worker_type = _resolve_execution_mode(ps.execution_type)
-        input_proc, next_stage_proc = _select_processor_funcs(ps, resolve_stage_async_chunk(deploy, ds))
+        stage_async_chunk = resolve_stage_async_chunk(deploy, ds)
+        input_proc, next_stage_proc = _select_processor_funcs(ps, stage_async_chunk)
         engine_args = _build_engine_args(ps, ds, pipeline, deploy, next_stage_proc)
         # Downstream stages may share a multimodal wrapper class without owning
         # an encoder. Do not make vLLM profile dummy multimodal inputs for them.