[CI/BugFix] Fix Flaky Test for Qwen Omni Perf

[alex-jw-brooks]

Purpose

Fixes the flaky test in the build linked here: #2752 #2389

Streaming requests in vLLM / vLLM Omni follow SSE specification. Since we largely send data, this mostly means that we are sending things like:

b'data: {JSON}\n\n'

Importantly, the space after the : does matter. In our performance script, we are currently .strip() ing all incoming chunks. This is the underlying cause of the erratic CI failures, because higher concurrency in streaming requests can lead to situations like this:

chunk1: b'data: '
chunk2: b'{JSON}\n\n'

When we encounter this case, add_chunks on the handler will add the stripped messages, i.e., giving data:{JSON}\n\n. As a result chunk = message.removeprefix("data: ") later in our script doesn't do anything, and it tries to decode the JSON with the data: in front, which causes the parsing error.

Reproducing it is a bit difficult, but I did log one of the failed requests out and did see the leading data: on it when decoding failed. The best way to repro is likely with a higher concurrency config, e.g., running pytest tests/dfx/perf/scripts/run_benchmark.py -s with the config path set to point to something like below:

[
    {
        "test_name": "test_qwen3_omni_chunk_stress",
        "server_params": {
            "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
            "stage_config_name": "qwen3_omni.yaml",
            "update": {
                "async_chunk": true,
                "stage_args": {
                    "0": {
                        "engine_args.custom_process_next_stage_input_func": "vllm_omni.model_executor.stage_input_processors.qwen3_omni.thinker2talker_async_chunk"
                    },
                    "1": {
                        "engine_args.custom_process_next_stage_input_func": "vllm_omni.model_executor.stage_input_processors.qwen3_omni.talker2code2wav_async_chunk"
                    }
                }
            },
            "delete": {
                "stage_args": {
                    "2": [
                        "custom_process_input_func"
                    ]
                }
            }
        },
        "benchmark_params": [
            {
                "dataset_name": "random",
                "backend": "openai-chat-omni",
                "endpoint": "/v1/chat/completions",
                "num_prompts": 500,
                "max_concurrency": 32,
                "random_input_len": 100,
                "random_output_len": 100,
                "ignore_eos": true,
                "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
                "baseline": {
                    "mean_ttft_ms": 10000,
                    "mean_audio_ttfp_ms": 10000,
                    "mean_audio_rtf": 1.0
                }
            }
        ]
    }
]

CC @tzhouam

[chatgpt-codex-connector]

Codex usage limits have been reached for code reviews. Please check with the admins of this repo to increase the limits by adding credits.
Credits must be used to enable repository wide code reviews.

[amy-why-3459]

#2389
Thank you so much for your fix. I believe your fix will also resolve this issue.

[hsliuustc0106]

BLOCKER scan:

• Correctness: PASS
• Reliability/Safety: PASS
• Breaking Changes: PASS
• Test Coverage: PASS (CI tests verify)
• Documentation: PASS
• Security: PASS

OVERALL: NO BLOCKERS

VERDICT: COMMENT

Good catch on the TCP fragmentation issue. SSE parsing requires exact handling of whitespace - stripping can break the protocol. The comment explaining the issue is clear and helpful.

[yenuo26]

@Gaohan123 @gcanlin @princepride Please help review whether this is ready to be merged.

[Gaohan123]

LGTM. Thanks!