[CI/Build][MiniCPM-o] Derive duplex admission-probe limit from the deploy config (#6678)

Signed-off-by: Deep Shah <deep@socratic.co>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

tests/e2e/features/fullduplex/engine/test_duplex_deploy_config.py

@@ -8,6 +8,7 @@ from dataclasses import FrozenInstanceError
 
 import pytest
 
+from tests.helpers.stage_config import get_deploy_duplex_max_sessions
 from vllm_omni.config.stage_config import (
     DuplexSessionRuntimeConfig,
     load_deploy_config,
@@ -66,3 +67,18 @@ def test_duplex_session_runtime_rejects_non_positive_values(tmp_path, name: str,
 
     with pytest.raises(ValueError, match=rf"duplex_session\.{name} must be positive"):
         load_deploy_config(deploy_path)
+
+
+@pytest.mark.parametrize(
+    ("deploy_yaml", "expected"),
+    [
+        ("minicpmo_4_5.yaml", 4),
+        ("minicpmo_4_5_8x4090.yaml", 1),
+        ("minicpmo_4_5_3gpu_stage1_replicas.yaml", 4),
+    ],
+)
+def test_deploy_duplex_max_sessions_tracks_the_deploy_config(deploy_yaml: str, expected: int) -> None:
+    # Guards the admission probe. A capacity edit, a config that declares none,
+    # or an overlay inheriting one from its base must surface here rather than
+    # as another nightly duplex admission timeout.
+    assert get_deploy_duplex_max_sessions(deploy_yaml) == expected

tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py

@@ -15,10 +15,11 @@ import pytest
 from huggingface_hub import snapshot_download
 
 from tests.helpers.runtime import OmniServerParams, get_model_prefix
-from tests.helpers.stage_config import get_deploy_config_path
+from tests.helpers.stage_config import get_deploy_config_path, get_deploy_duplex_max_sessions
 
 MODEL = "openbmb/MiniCPM-o-4_5"
-DEPLOY_CONFIG = get_deploy_config_path("minicpmo_4_5.yaml")
+DEPLOY_CONFIG_REL = "minicpmo_4_5.yaml"
+DEPLOY_CONFIG = get_deploy_config_path(DEPLOY_CONFIG_REL)
 ASSET_DIR = Path(__file__).resolve().parents[3] / "assets" / "minicpmo_4_5"
 RESPONSE_REQUIRED_WAV = ASSET_DIR / "response_required_16k.wav"
 RESPONSE_REQUIRED_SHA256 = "2e5fd4eb3ee434ce107ee3a0591fa624a33f7683c7462f45fe651c443c9af941"
@@ -39,6 +40,16 @@ SERVER_PARAMS = [
 ]
 
 
+def deploy_max_sessions() -> int:
+    """Concurrent duplex sessions the deploy config under test admits.
+
+    The admission probe has to expect the capacity the server is actually
+    started with. Hardcoding it silently drifts the moment the deploy config
+    changes its capacity, turning that change into an unrelated probe timeout.
+    """
+    return get_deploy_duplex_max_sessions(DEPLOY_CONFIG_REL)
+
+
 def validated_wav(path: Path, expected_sha256: str) -> Path:
     actual_sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
     if actual_sha256 != expected_sha256:

tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py

@@ -15,6 +15,7 @@ import pytest
 from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import (
     SERVER_PARAMS,
     SOFT_INTERRUPT_SHA256,
+    deploy_max_sessions,
     multi_session_args,
     realtime_url,
     resolve_ref_audio,
@@ -46,7 +47,7 @@ def test_duplex_admission_and_expiry_reaper(omni_server, tmp_path: Path) -> None
     args.disconnect_session_index = None
     args.takeover_session_index = None
     args.expire_session_index = 0
-    args.verify_admission_limit = 2
+    args.verify_admission_limit = deploy_max_sessions()
     result = asyncio.run(run_lifecycle_probes(args))
 
     assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2)

tests/helpers/stage_config.py

@@ -11,6 +11,8 @@ from typing import Any
 
 import yaml
 
+from vllm_omni.config.stage_config import load_deploy_config
+
 
 def modify_stage_config(
     yaml_path: str,
@@ -669,6 +671,16 @@ def get_deploy_config_stage(rel_path: str, stage_id: int) -> dict[str, Any]:
     raise KeyError(f"No stage_id={stage_id} in deploy config {rel_path!r}")
 
 
+def get_deploy_duplex_max_sessions(rel_path: str) -> int:
+    """Return the duplex session capacity a deploy yaml admits.
+
+    Loads through ``load_deploy_config`` so ``base_config`` merging and
+    ``DuplexSessionRuntimeConfig`` defaults stay in lockstep with the server
+    instead of being re-implemented here.
+    """
+    return load_deploy_config(get_deploy_config_path(rel_path)).duplex_session.max_sessions
+
+
 def _stage_ids_from_deploy_yaml(stage_config_path: str) -> list[int]:
     """Return ``stage_id`` values from a new-schema deploy YAML (``stages``)."""
     with open(stage_config_path, encoding="utf-8") as f:
@@ -734,6 +746,7 @@ def stage_config_path_for_run_level(stage_config_path: str | None, run_level: st
 __all__ = [
     "get_deploy_config_path",
     "get_deploy_config_stage",
+    "get_deploy_duplex_max_sessions",
     "get_stage_entries",
     "load_stage_ids",
     "load_stage_replica_counts",