[Bugfix] Keep pipeline event-driven defaults local

Signed-off-by: Sy03 <1370724210@qq.com>

vllm_omni/engine/async_omni_engine.py

@@ -14,7 +14,6 @@ import asyncio
 import concurrent.futures
 import copy
 import json
-import os
 import queue
 import shutil
 import threading
@@ -72,7 +71,6 @@ from vllm_omni.engine.messages import (
     StageSubmissionMessage,
 )
 from vllm_omni.engine.orchestrator import (
-    _EVENT_DRIVEN_ORCH_ENV,
     Orchestrator,
     _event_driven_orch_default_for_pipeline,
 )
@@ -237,12 +235,9 @@ class AsyncOmniEngine:
             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._event_driven_orch_default = _event_driven_orch_default_for_pipeline(
+            pipeline_config.model_type if pipeline_config is not None else None
+        )
         self.duplex_session_config = DuplexSessionRuntimeConfig()
         if deploy_config_path is not None:
             self.duplex_session_config = load_deploy_config(deploy_config_path).duplex_session
@@ -460,6 +455,7 @@ class AsyncOmniEngine:
                 prom_metrics=self._prom_metrics,
                 log_stats=self._log_stats,
                 enable_orch_monitor=self._enable_orch_monitor,
+                event_driven_orch_default=self._event_driven_orch_default,
                 duplex_runtime_extension=duplex_runtime_extension,
                 enable_duplex_control=self._duplex_control_enabled,
                 duplex_session_config=self.duplex_session_config,

vllm_omni/engine/orchestrator.py

@@ -449,6 +449,7 @@ class Orchestrator:
         prom_metrics: Any = None,
         log_stats: bool = False,
         enable_orch_monitor: bool = False,
+        event_driven_orch_default: bool = False,
         duplex_runtime_extension: DuplexRuntimeExtension | None = None,
         enable_duplex_control: bool = False,
         duplex_session_config: DuplexSessionRuntimeConfig | None = None,
@@ -520,7 +521,7 @@ class Orchestrator:
 
         self._shutdown_event = asyncio.Event()
         self._stages_shutdown = False
-        self._event_driven_orch = _event_driven_orch_enabled()
+        self._event_driven_orch = _event_driven_orch_enabled(default=event_driven_orch_default)
 
         # Distributed membership (optional, injected by DistStageRuntime)
         self._membership = membership_controller

vllm_omni/entrypoints/async_omni.py

@@ -943,7 +943,9 @@ class AsyncOmni(EngineClient, OmniBase):
         # event-driven loop (vllm_omni/engine/orchestrator.py).
         from vllm_omni.engine.orchestrator import _event_driven_orch_enabled
 
-        event_driven_drain = _event_driven_orch_enabled() and hasattr(engine, "get_output_blocking_async")
+        event_driven_drain = _event_driven_orch_enabled(
+            default=bool(getattr(engine, "_event_driven_orch_default", False))
+        ) and hasattr(engine, "get_output_blocking_async")
 
         async def _final_output_loop():
             """Background coroutine that dispatches final outputs to request queues."""