diff --git a/docs/configuration/environment_variables.md b/docs/configuration/environment_variables.md index 3a37b52d2..88e2dd567 100644 --- a/docs/configuration/environment_variables.md +++ b/docs/configuration/environment_variables.md @@ -83,7 +83,7 @@ depends on the installed kernels and model path. | `SPEAKER_SAMPLES_DIR` | Filesystem path; default `~/.cache/vllm-omni/speakers` | Speech server; read when speaker storage initializes | Environment-only setting. The directory is created; filesystem errors propagate. | Stable | | `SPEAKER_MAX_UPLOADED` | Integer; default `1000` | Speech server; read when speaker storage initializes | Environment-only setting. A non-integer logs a warning and uses `1000`; range is not otherwise validated. | Stable | | `VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT` | Float seconds; default `600` | Diffusion engine async-output wait in `step_streaming`; resolved per call on the request path, not at import | Environment-only setting. A non-float or `<=0` value warns once and uses the default. | Experimental | -| `VLLM_OMNI_EVENT_DRIVEN_ORCH` | `1`, `true`, `yes` or `on` enables; default `0` (off) | Orchestration loop and the serving-side final-output drain; read once when the `Orchestrator` is constructed | Environment-only setting. Values are stripped and case-normalized; any unrecognized value leaves the legacy poll loop selected. | Experimental | +| `VLLM_OMNI_EVENT_DRIVEN_ORCH` | `1`, `true`, `yes` or `on` enables; default `0` (off), except Qwen3-TTS defaults to on | Orchestration loop and the serving-side final-output drain; read once when the `Orchestrator` is constructed | Environment-only setting. An explicit value always wins. Values are stripped and case-normalized; any unrecognized value leaves the legacy poll loop selected. | Experimental | | `VLLM_OMNI_INPUT_WAIT_TIMEOUT_S` | Float seconds; default `600`; `<=0` disables | Full-payload input coordinator, not async-chunk transfer; read when the scheduler module imports in each worker | Environment-only setting. A non-float logs a warning and uses `600`. | Stable operational control | | `VLLM_OMNI_ORCH_MONITOR_PATH` | Filesystem path; default `/vllm_omni_orch_monitor_.json` | Orchestrator monitor enabled by `--enable-orch-monitor`; read when the monitor is created | Environment-only path override. Parent directories are created; write errors are logged. | Diagnostic | | `VLLM_OMNI_VIDEO_SYNC_TIMEOUT` | Float seconds; default `600` | Synchronous Videos API; read when the API server module imports | Environment-only setting. A non-float raises `ValueError` during import. | Experimental | diff --git a/docs/design/module/engine_orchestration.md b/docs/design/module/engine_orchestration.md index d4dbf51a2..a930b4008 100644 --- a/docs/design/module/engine_orchestration.md +++ b/docs/design/module/engine_orchestration.md @@ -72,7 +72,7 @@ the boundary affected by the in-flight stage client/process refactor in responsibilities proposed only by that PR are not current contracts. The orchestration loop also has an opt-in event-driven mode -(`VLLM_OMNI_EVENT_DRIVEN_ORCH=1`, default off) proposed in +(`VLLM_OMNI_EVENT_DRIVEN_ORCH=1`, default off except for Qwen3-TTS) proposed in [#5221](https://github.com/vllm-project/vllm-omni/pull/5221). It changes poll cadence only: the routing, ordering, and terminal-state contracts below hold identically on both loops. diff --git a/tests/engine/test_orchestrator_event_driven.py b/tests/engine/test_orchestrator_event_driven.py index 284dde55d..abfa05933 100644 --- a/tests/engine/test_orchestrator_event_driven.py +++ b/tests/engine/test_orchestrator_event_driven.py @@ -19,7 +19,10 @@ import pytest from vllm_omni.engine.async_omni_engine import AsyncOmniEngine from vllm_omni.engine.messages import ShutdownRequestMessage -from vllm_omni.engine.orchestrator import _event_driven_orch_enabled +from vllm_omni.engine.orchestrator import ( + _event_driven_orch_default_for_pipeline, + _event_driven_orch_enabled, +) from . import test_orchestrator as legacy from . import test_orchestrator_error_handling as legacy_errors @@ -125,6 +128,19 @@ def test_flag_parsing(monkeypatch) -> None: assert _event_driven_orch_enabled() is False +def test_only_qwen3_tts_has_a_pipeline_default() -> None: + assert _event_driven_orch_default_for_pipeline("qwen3_tts") is True + for model_type in (None, "qwen3_omni_moe", "minicpmo_4_5", "moss_tts_delay"): + assert _event_driven_orch_default_for_pipeline(model_type) is False + + +def test_explicit_event_driven_value_overrides_pipeline_default(monkeypatch) -> None: + monkeypatch.delenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", raising=False) + assert _event_driven_orch_enabled(default=True) is True + monkeypatch.setenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", "0") + assert _event_driven_orch_enabled(default=True) is False + + def test_default_is_legacy_loop(monkeypatch) -> None: """Without the env flag, the harness runs the legacy poll loop.""" monkeypatch.delenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", raising=False) diff --git a/vllm_omni/engine/async_omni_engine.py b/vllm_omni/engine/async_omni_engine.py index 6850e6b29..7e7314585 100644 --- a/vllm_omni/engine/async_omni_engine.py +++ b/vllm_omni/engine/async_omni_engine.py @@ -13,6 +13,8 @@ from __future__ import annotations import asyncio import concurrent.futures import copy +import json +import os import queue import shutil import threading @@ -69,7 +71,11 @@ from vllm_omni.engine.messages import ( OutputMessage, StageSubmissionMessage, ) -from vllm_omni.engine.orchestrator import Orchestrator +from vllm_omni.engine.orchestrator import ( + _EVENT_DRIVEN_ORCH_ENV, + Orchestrator, + _event_driven_orch_default_for_pipeline, +) from vllm_omni.engine.rpc_result_router import CorrelatedRpcClient from vllm_omni.engine.serialization import deserialize_additional_information from vllm_omni.engine.stage_client import StageClient @@ -212,6 +218,40 @@ class AsyncOmniEngine: # retaining the richer resolver result for pipeline-wide settings. # Overrides of that private seam fall back to the factory below. deploy_config_path = kwargs.get("deploy_config") + # ``trust_remote_code`` is tri-state (bool | None): ``None`` means "not + # specified" so stage-config resolution can defer to the deploy yaml's + # per-stage value (see ``with_trust_remote_code_override``). The + # restriction path below loads the top-level HF config via vLLM's + # ``get_config``, which needs a real bool, so collapse ``None`` to the + # default ``False`` here (#5495). + pipeline_config = StageConfigFactory.get_pipeline_config( + model=model, + trust_remote_code=bool(trust_remote_code), + deploy_config_path=deploy_config_path, + ) + self.endpoint_restrictions = pipeline_config.endpoint_restrictions if pipeline_config is not None else () + self._duplex_runtime_extension_path = ( + pipeline_config.duplex_runtime_extension if pipeline_config is not None else None + ) + self.duplex_serving_adapter_path = ( + pipeline_config.duplex_serving_adapter if pipeline_config is not None else None + ) + self._duplex_control_enabled = bool(pipeline_config and pipeline_config.duplex_control_enabled) + if ( + _EVENT_DRIVEN_ORCH_ENV not in os.environ + and pipeline_config is not None + and _event_driven_orch_default_for_pipeline(pipeline_config.model_type) + ): + os.environ[_EVENT_DRIVEN_ORCH_ENV] = "1" + self.duplex_session_config = DuplexSessionRuntimeConfig() + if deploy_config_path is not None: + self.duplex_session_config = load_deploy_config(deploy_config_path).duplex_session + + # Tri-state: None means "not specified" — the deploy yaml's per-stage + # trust_remote_code stays in effect. An explicit True/False here is a + # global override (precedence: caller > deploy yaml > default False); + # the merge rule lives in with_trust_remote_code_override. + kwargs = with_trust_remote_code_override(kwargs, trust_remote_code) self._config_resolution: OmniConfigResolution | None = None self.config_path, self.stage_configs = self._resolve_stage_configs( model, diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index 498a9dc89..fadbd4698 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -96,7 +96,7 @@ def cleanup_request_artifact_dirs(artifact_dirs: set[str] | list[str]) -> None: # 1 ms poll cadence to event-driven wakeups: one reader task per live LLM stage # replica awaits `client.get_output_async()` directly — the same pattern vLLM's # own AsyncLLM output handler uses — and feeds a single serial dispatch queue. -# Default off; the legacy poll loop remains the fallback. +# Default is off except for pipelines with an explicit validated default. _EVENT_DRIVEN_ORCH_ENV = "VLLM_OMNI_EVENT_DRIVEN_ORCH" # How often the event-driven loop reconciles its reader-task set against @@ -104,8 +104,16 @@ _EVENT_DRIVEN_ORCH_ENV = "VLLM_OMNI_EVENT_DRIVEN_ORCH" _ORCH_READER_RECONCILE_INTERVAL_S = 0.5 -def _event_driven_orch_enabled() -> bool: - return os.environ.get(_EVENT_DRIVEN_ORCH_ENV, "0").strip().lower() in ("1", "true", "yes", "on") +def _event_driven_orch_enabled(*, default: bool = False) -> bool: + value = os.environ.get(_EVENT_DRIVEN_ORCH_ENV) + if value is None: + return default + return value.strip().lower() in ("1", "true", "yes", "on") + + +def _event_driven_orch_default_for_pipeline(pipeline_model_type: str | None) -> bool: + """Return whether a pipeline has a validated event-driven default.""" + return pipeline_model_type == "qwen3_tts" def _build_terminal_empty_output(