ci: run focused vLLM 0.30.0 weekly baseline

Signed-off-by: tzhouam <tzhouam@connect.ust.hk>

.buildkite/cuda/rebase-pipeline.yml

@@ -1,35 +1,46 @@
+# Focused weekly comparison against unchanged Omni main and official vLLM 0.30.0.
 steps:
-  - label: ":docker: Build image"
+- label: ':docker: Build image'
-    key: image-build
+  key: image-build
-    commands:
+  commands:
-      - "bash .buildkite/cuda/scripts/build-ci-image.sh"
+  - bash .buildkite/cuda/scripts/build-ci-image.sh
-    agents:
+  agents:
-      queue: "cpu_queue_premerge"
+    queue: cpu_queue_premerge
-
+- label: Upload selected weekly baseline
-  # L2 Test
+  depends_on: image-build
-  - label: "Upload Ready Pipeline"
+  key: upload-weekly-pipeline
-    depends_on: image-build
+  commands:
-    key: upload-ready-pipeline
+  - |
-    commands:
+    python3 - <<'PYBASELINE'
-      - python3 .buildkite/common/scripts/upload_pipeline.py --upload --all .buildkite/cuda/test-ready.yml
+    import tempfile
-    agents:
+    from pathlib import Path
-      queue: "cpu_queue_premerge"
+    import yaml
-
+    labels = set(['TTS · Audex · CosyVoice · Qwen3-TTS · L4', 'TTS · H100', 'Diffusion · Bagel · H100 · Single-GPU', 'Diffusion · SenseNova-U1 · H100 · Single-GPU', 'Diffusion · DreamZero · H100 · 2-GPU', 'Diffusion · H100 · 4-GPU', 'Diffusion · MAGI-2 Preview Serving Smoke · H100 · 4-GPU'])
-  # L3 Test
+    source = yaml.safe_load(Path('.buildkite/cuda/test-weekly.yml').read_text())
-  - label: "Upload Merge Pipeline"
+    selected = set()
-    depends_on: image-build
+    guard = 'python3 -c \'import vllm, torch, flashinfer, vllm._C_stable_libtorch; assert vllm.__version__ == \'"\'"\'0.30.0\'"\'"\'; assert torch.cuda.is_available(); print(\'"\'"\'Verified baseline runtime:\'"\'"\', vllm.__version__, torch.__version__)\''
-    key: upload-merge-pipeline
+    def select(steps):
-    commands:
+        kept = []
-      - python3 .buildkite/common/scripts/upload_pipeline.py --upload --all .buildkite/cuda/test-merge.yml
+        for step in steps:
-    agents:
+            if not isinstance(step, dict):
-      queue: "cpu_queue_premerge"
+                continue
-
+            if 'steps' in step:
-  # L4 Test
+                children = select(step['steps'])
-  - label: "Upload Nightly Pipeline"
+                if children:
-    depends_on: image-build
+                    kept.append(dict(step, steps=children))
-    key: upload-nightly-pipeline
+            elif step.get('label') in labels:
-    if: build.env("NIGHTLY") == "1"
+                item = dict(step)
-    commands:
+                commands = item['commands']
-      - python3 .buildkite/common/scripts/upload_pipeline.py --upload --all .buildkite/cuda/test-nightly.yml
+                item['commands'] = [guard] + ([commands] if isinstance(commands, str) else commands)
-    agents:
+                kept.append(item)
-      queue: "cpu_queue_premerge"
+                selected.add(item['label'])
+        return kept
+    source['steps'] = select(source['steps'])
+    assert selected == labels, (selected, labels)
+    path = Path(tempfile.gettempdir()) / 'vllm-omni-selected-weekly-baseline.yml'
+    path.write_text(yaml.safe_dump(source, sort_keys=False, allow_unicode=True))
+    print('Selected baseline suites:', sorted(selected), flush=True)
+    PYBASELINE
+    python3 .buildkite/common/scripts/upload_pipeline.py --upload --all /tmp/vllm-omni-selected-weekly-baseline.yml
+  agents:
+    queue: cpu_queue_premerge

docker/Dockerfile.ci

@@ -36,7 +36,8 @@ RUN apt-get update && \
 # for CI
 COPY pyproject.toml setup.py ./
 COPY requirements/ requirements/
-ENV SETUPTOOLS_SCM_PRETEND_VERSION=0.0.0
+# Keep source-built dependency versions independent of the Omni CI version.
+ENV VLLM_OMNI_VERSION_OVERRIDE=0.0.0
 # Editable install: deps + project metadata are cached in this layer, while
 # the package itself is linked to APP_DIR (populated later by COPY --link).
 # Do NOT use a non-editable install here — that copies an empty stub into

tests/helpers/runtime.py

@@ -659,14 +659,6 @@ class OmniRunner:
         self.seed = seed
         self._prompt_len_estimate_cache: dict[str, Any] = {}
         self.omni: Any = None
-        self._runner_process = psutil.Process()
-        self._initial_child_ids: set[tuple[int, float]] = set()
-        self._owned_engine_processes: dict[int, float] = {}
-        for child in self._runner_process.children(recursive=True):
-            try:
-                self._initial_child_ids.add((child.pid, child.create_time()))
-            except psutil.Error:
-                pass
         try:
             from vllm_omni.entrypoints.omni import Omni
 
@@ -678,14 +670,12 @@ class OmniRunner:
                 deploy_config=deploy_config,
                 **kwargs,
             )
-            self._remember_engine_processes(startup=True)
             startup_s = time.perf_counter() - startup_t0
             if log_stats:
                 print(f"OmniRunner startup took {startup_s:.3f}s (model={model_name})", flush=True)
         except BaseException:
             # ``with OmniRunner(...)`` never reaches ``__enter__``/``__exit__``
             # when construction fails after worker processes have started.
-            self._remember_engine_processes(startup=True)
             self.__exit__(None, None, None)
             raise
 
@@ -909,51 +899,18 @@ class OmniRunner:
     def stop_profile(self, stages: list[int] | None = None) -> list[Any]:
         return self.omni.stop_profile(stages=stages)
 
-    def _remember_engine_processes(self, *, startup: bool = False) -> None:
-        """Track startup workers, then only descendants of those workers.
-
-        Existing children may belong to another runner or a shared transcriber.
-        Record identities before close() can reparent surviving workers.
-        """
-        try:
-            if startup:
-                candidates = self._runner_process.children(recursive=True)
-            else:
-                candidates = []
-                for pid, created in list(self._owned_engine_processes.items()):
-                    try:
-                        proc = psutil.Process(pid)
-                        if proc.create_time() == created:
-                            candidates.extend(proc.children(recursive=True))
-                    except psutil.Error:
-                        pass
-            for proc in candidates:
-                try:
-                    identity = (proc.pid, proc.create_time())
-                    if identity in self._initial_child_ids:
-                        continue
-                    if startup and any(
-                        (parent.pid, parent.create_time()) in self._initial_child_ids for parent in proc.parents()
-                    ):
-                        continue
-                    cmdline = " ".join(proc.cmdline()).lower()
-                    if "enginecore" in cmdline or "enginecore" in proc.name().lower():
-                        self._owned_engine_processes[identity[0]] = identity[1]
-                except psutil.Error:
-                    pass
-        except psutil.Error as error:
-            print(f"Warning: could not inspect runner workers: {error}")
-
     def _cleanup_process(self):
         try:
+            keywords = ["enginecore"]
             matched = []
-            for pid, created in self._owned_engine_processes.items():
+            for proc in psutil.process_iter(["pid", "name", "cmdline", "username"]):
                 try:
-                    proc = psutil.Process(pid)
+                    cmdline = " ".join(proc.cmdline()).lower() if proc.cmdline() else ""
-                    if proc.create_time() == created:
+                    name = proc.name().lower()
-                        print(f"Found owned vllm process: PID={proc.pid}")
+                    if any(k in cmdline for k in keywords) or any(k in name for k in keywords):
+                        print(f"Found vllm process: PID={proc.pid}, cmd={cmdline[:100]}")
                         matched.append(proc)
-                except psutil.Error:
+                except (psutil.NoSuchProcess, psutil.AccessDenied):
                     pass
             for proc in matched:
                 try:
@@ -981,14 +938,11 @@ class OmniRunner:
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        self._remember_engine_processes()
+        omni = getattr(self, "omni", None)
-        try:
+        if omni is not None and hasattr(omni, "close"):
-            omni = getattr(self, "omni", None)
+            omni.close()
-            if omni is not None and hasattr(omni, "close"):
+        self._cleanup_process()
-                omni.close()
+        cleanup_test_environment()
-        finally:
-            self._cleanup_process()
-            cleanup_test_environment()
 
 
 # ---------------------------------------------------------------------------

tests/helpers/tests/test_runtime_omni_runner.py

@@ -1,15 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
-"""Tests for OmniRunner startup rollback and worker ownership."""
+"""Tests for OmniRunner startup rollback."""
 
 from __future__ import annotations
 
-import subprocess
-import sys
-from collections.abc import Generator
-
-import psutil
 import pytest
 
 from tests.helpers.runtime import OmniRunner
@@ -43,114 +38,3 @@ def test_runner_init_rolls_back_on_omni_startup_failure(monkeypatch: pytest.Monk
     assert cleaned.count("process") == 1
     # Once at the start of ``__init__``, once from the constructor rollback.
     assert cleaned.count("env") == 2
-
-
-@pytest.fixture
-def live_workers() -> Generator[list[subprocess.Popen[bytes]], None, None]:
-    workers: list[subprocess.Popen[bytes]] = []
-    try:
-        yield workers
-    finally:
-        for worker in workers:
-            if worker.poll() is None:
-                worker.kill()
-            worker.wait(timeout=10)
-            if worker.stdout is not None:
-                worker.stdout.close()
-
-
-def _start_engine_worker(
-    workers: list[subprocess.Popen[bytes]], *, ignore_terminate: bool = False
-) -> subprocess.Popen[bytes]:
-    script = "import signal, time; "
-    if ignore_terminate:
-        script += "signal.signal(signal.SIGTERM, signal.SIG_IGN); "
-    script += "print('ready', flush=True); time.sleep(90)"
-    worker = subprocess.Popen([sys.executable, "-c", script, "enginecore"], stdout=subprocess.PIPE)
-    workers.append(worker)
-    assert worker.stdout is not None
-    assert worker.stdout.readline() == b"ready\n"
-    return worker
-
-
-@pytest.mark.parametrize("close_raises", [False, True])
-def test_runner_cleanup_preserves_unrelated_engine_workers(
-    monkeypatch: pytest.MonkeyPatch,
-    live_workers: list[subprocess.Popen[bytes]],
-    close_raises: bool,
-) -> None:
-    existing = _start_engine_worker(live_workers)
-    owned: list[subprocess.Popen[bytes]] = []
-    late_unrelated: list[subprocess.Popen[bytes]] = []
-    monkeypatch.setattr("tests.helpers.runtime.cleanup_test_environment", lambda: None)
-
-    class _OwnedOmni:
-        def __init__(self, *args: object, **kwargs: object) -> None:
-            owned.append(_start_engine_worker(live_workers))
-
-        def close(self) -> None:
-            late_unrelated.append(_start_engine_worker(live_workers))
-            if close_raises:
-                raise RuntimeError("close failed")
-
-    monkeypatch.setattr("vllm_omni.entrypoints.omni.Omni", _OwnedOmni)
-    if close_raises:
-        with pytest.raises(RuntimeError, match="close failed"), OmniRunner("fake-model"):
-            pass
-    else:
-        with OmniRunner("fake-model"):
-            pass
-
-    assert owned[0].poll() is not None
-    assert existing.poll() is None
-    assert late_unrelated[0].poll() is None
-
-
-def test_runner_startup_failure_cleans_only_new_workers(
-    monkeypatch: pytest.MonkeyPatch, live_workers: list[subprocess.Popen[bytes]]
-) -> None:
-    unrelated = _start_engine_worker(live_workers)
-    owned: list[subprocess.Popen[bytes]] = []
-    monkeypatch.setattr("tests.helpers.runtime.cleanup_test_environment", lambda: None)
-
-    class _FailingOmni:
-        def __init__(self, *args: object, **kwargs: object) -> None:
-            owned.append(_start_engine_worker(live_workers))
-            raise RuntimeError("startup failed")
-
-    monkeypatch.setattr("vllm_omni.entrypoints.omni.Omni", _FailingOmni)
-    with pytest.raises(RuntimeError, match="startup failed"):
-        OmniRunner("fake-model")
-
-    assert owned[0].poll() is not None
-    assert unrelated.poll() is None
-
-
-def test_runner_cleanup_rejects_stale_process_identity(live_workers: list[subprocess.Popen[bytes]]) -> None:
-    unrelated = _start_engine_worker(live_workers)
-    runner = object.__new__(OmniRunner)
-    runner._owned_engine_processes = {unrelated.pid: psutil.Process(unrelated.pid).create_time() + 1}
-
-    runner._cleanup_process()
-
-    assert unrelated.poll() is None
-
-
-def test_runner_cleanup_kills_owned_worker_ignoring_terminate(
-    monkeypatch: pytest.MonkeyPatch, live_workers: list[subprocess.Popen[bytes]]
-) -> None:
-    owned: list[subprocess.Popen[bytes]] = []
-    monkeypatch.setattr("tests.helpers.runtime.cleanup_test_environment", lambda: None)
-
-    class _StubbornOmni:
-        def __init__(self, *args: object, **kwargs: object) -> None:
-            owned.append(_start_engine_worker(live_workers, ignore_terminate=True))
-
-        def close(self) -> None:
-            pass
-
-    monkeypatch.setattr("vllm_omni.entrypoints.omni.Omni", _StubbornOmni)
-    with OmniRunner("fake-model"):
-        assert owned[0].poll() is None
-
-    assert owned[0].poll() is not None