[CI/Build] Fix CPU coverage collection with multiple workers

tests/distributed/omni_connectors/test_shm_connector.py

@@ -43,7 +43,12 @@ def test_deadline_receive_does_not_wait_for_writer_lock(self, connector):
         assert connector.get_with_deadline("0", "1", key, deadline=time.monotonic() - 1) is None
         assert connector.get_with_deadline("0", "1", key, deadline=time.monotonic() + 1)[0] == {"value": 7}
 
-    @pytest.mark.parametrize("key", ["test_key_1", "speech-" + uuid.uuid4().hex + "-" + uuid.uuid4().hex + "_0_0"])
+    # Stable IDs keep xdist collection identical despite the per-process random key.
+    @pytest.mark.parametrize(
+        "key",
+        ["test_key_1", "speech-" + uuid.uuid4().hex + "-" + uuid.uuid4().hex + "_0_0"],
+        ids=["simple", "speech"],
+    )
     def test_put_then_get_by_key(self, connector, key):
         data = {"hello": "world", "n": 42}
         ok, size, meta = connector.put("s0", "s1", key, data)

tests/helpers/stage_config.py

@@ -658,8 +658,14 @@ def _materialize_ci_overlay(model_type: str) -> Path:
     if base:
         overlay["base_config"] = str(_DEPLOY_DIR / base)
 
-    with open(out, "w", encoding="utf-8") as f:
-        yaml.safe_dump(overlay, f, sort_keys=False)
+    # Other pytest workers may read this path while regenerating the overlay.
+    fd, tmp_path = tempfile.mkstemp(dir=_CI_GENERATED_DIR, suffix=".yaml")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            yaml.safe_dump(overlay, f, sort_keys=False)
+        os.replace(tmp_path, out)
+    finally:
+        Path(tmp_path).unlink(missing_ok=True)
     return out
 
 

tests/helpers/tests/test_ci_overlay.py

@@ -0,0 +1,36 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""CI overlay publication must not expose incomplete YAML to other workers."""
+
+from pathlib import Path
+
+import pytest
+import yaml
+
+from tests.helpers import stage_config
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_ci_overlay_publishes_complete_document(tmp_path, monkeypatch):
+    monkeypatch.setattr(stage_config, "_CI_GENERATED_DIR", tmp_path)
+    out = tmp_path / "bagel.yaml"
+    previous = "stages: [{stage_id: 99}]\n"
+    out.write_text(previous, encoding="utf-8")
+    dump = yaml.safe_dump
+
+    def check_reader_during_write(data, stream, **kwargs):
+        # Readers must keep seeing the previous document until publication.
+        assert out.read_text(encoding="utf-8") == previous
+        result = dump(data, stream, **kwargs)
+        stream.flush()
+        assert out.read_text(encoding="utf-8") == previous
+        return result
+
+    monkeypatch.setattr(yaml, "safe_dump", check_reader_during_write)
+    assert Path(stage_config.get_deploy_config_path("ci/bagel.yaml")) == out
+    expected = dict(stage_config._CI_OVERLAYS["bagel"])
+    expected["base_config"] = str(stage_config._DEPLOY_DIR / expected["base_config"])
+    assert yaml.safe_load(out.read_text(encoding="utf-8")) == expected
+    assert list(tmp_path.iterdir()) == [out]