From 776424f0de8a69fc2f5da0c4f3cf767f96855f90 Mon Sep 17 00:00:00 2001 From: Zhou Taichang Date: Mon, 27 Jul 2026 16:24:47 +0800 Subject: [PATCH] [Refactor][WIP] Add missing test for modelrunner (G1/N) (#5333) Signed-off-by: Taichang Zhou Co-authored-by: Claude Opus 4.8 (1M context) --- tests/worker/test_capture_replay_contract.py | 79 +++++ .../test_fa3_scheduler_metadata_size.py | 85 ++++++ tests/worker/test_gpu_ar_model_runner.py | 92 ++++++ tests/worker/test_memory_utils.py | 71 +++++ tests/worker/test_payload_span.py | 166 +++++++++++ tests/worker/test_sampler_history_backfill.py | 142 +++++++++ tests/worker/test_worker_base.py | 237 +++++++++++++++ tools/baselines/omni_hash_baseline.py | 276 ++++++++++++++++++ tools/baselines/qwen3_omni_hash.baseline | 48 +++ 9 files changed, 1196 insertions(+) create mode 100644 tests/worker/test_capture_replay_contract.py create mode 100644 tests/worker/test_fa3_scheduler_metadata_size.py create mode 100644 tests/worker/test_memory_utils.py create mode 100644 tests/worker/test_payload_span.py create mode 100644 tests/worker/test_sampler_history_backfill.py create mode 100644 tests/worker/test_worker_base.py create mode 100644 tools/baselines/omni_hash_baseline.py create mode 100644 tools/baselines/qwen3_omni_hash.baseline diff --git a/tests/worker/test_capture_replay_contract.py b/tests/worker/test_capture_replay_contract.py new file mode 100644 index 000000000..504aaa7b0 --- /dev/null +++ b/tests/worker/test_capture_replay_contract.py @@ -0,0 +1,79 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Static coupling check for the ``has_preprocess`` capture/replay buffers. + +For models with ``has_preprocess``, CUDA-graph *capture* (``_dummy_run``) and +*replay* (``_preprocess``) MUST read from the same pre-allocated buffers +(``self.input_ids.gpu`` + ``self.inputs_embeds.gpu``); otherwise replay reads a +buffer capture never wrote and produces garbage. This is a source-level check +(no GPU) that catches a one-sided edit to either branch. + +The check is AST-scoped: it inspects ONLY the body of the ``has_preprocess`` +guarded branch (not a line window), so a buffer read removed from that branch +cannot be masked by an adjacent ``else``/``elif`` that happens to read it. It +matches by attribute chain (``self.input_ids.gpu``), so local variable renames +do not break it. + +Scope: the BASE runner (``OmniGPUModelRunner``) only. The generation runner is +KNOWN to lack the ``has_preprocess`` branch; extend this +check to the generation file when it adds the branch. +""" + +import ast +import inspect +import textwrap + +import pytest + +from vllm_omni.worker.gpu_model_runner import OmniGPUModelRunner + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +_REQUIRED_BUFFERS = {"self.input_ids.gpu", "self.inputs_embeds.gpu"} + + +def _attr_chain(node: ast.AST) -> str | None: + """Reconstruct a dotted ``self.x.y`` chain from an ast.Attribute node.""" + parts: list[str] = [] + cur = node + while isinstance(cur, ast.Attribute): + parts.append(cur.attr) + cur = cur.value + if isinstance(cur, ast.Name): + parts.append(cur.id) + return ".".join(reversed(parts)) + return None + + +def _attr_chains_in(nodes: list[ast.stmt]) -> set[str]: + chains: set[str] = set() + for stmt in nodes: + for inner in ast.walk(stmt): + if isinstance(inner, ast.Attribute): + chain = _attr_chain(inner) + if chain is not None: + chains.add(chain) + return chains + + +def _has_preprocess_branch_reads_both_buffers(method) -> bool: + """True iff SOME ``if has_preprocess:`` branch body reads BOTH buffers. + + Only the guarded branch's own ``body`` is inspected (elif/else excluded), + which is what makes a one-sided buffer removal fail. + """ + src = textwrap.dedent(inspect.getsource(method)) + tree = ast.parse(src) + guarded = [node for node in ast.walk(tree) if isinstance(node, ast.If) and "has_preprocess" in ast.dump(node.test)] + assert guarded, f"{method.__qualname__}: has_preprocess branch removed entirely" + return any(_REQUIRED_BUFFERS <= _attr_chains_in(node.body) for node in guarded) + + +@pytest.mark.parametrize("method_name", ["_dummy_run", "_preprocess"]) +def test_has_preprocess_branch_reads_shared_buffers(method_name): + method = getattr(OmniGPUModelRunner, method_name) + assert _has_preprocess_branch_reads_both_buffers(method), ( + f"{OmniGPUModelRunner.__name__}.{method_name}: the has_preprocess branch no " + "longer reads BOTH self.input_ids.gpu and self.inputs_embeds.gpu — capture " + "and replay would diverge." + ) diff --git a/tests/worker/test_fa3_scheduler_metadata_size.py b/tests/worker/test_fa3_scheduler_metadata_size.py new file mode 100644 index 000000000..5e308bed7 --- /dev/null +++ b/tests/worker/test_fa3_scheduler_metadata_size.py @@ -0,0 +1,85 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Characterization of ``OmniGPUModelRunner.initialize_metadata_builders``. + +Upstream ``FlashAttentionMetadataBuilder`` pre-allocates ``scheduler_metadata`` +for only ``max_num_seqs + 1`` entries, but FA3 with split scheduling +(``max_num_splits > 1``) may need ``max_num_seqs * max_num_splits + 1`` during +CUDA-graph capture. ``OmniGPUModelRunner`` re-sizes the buffer up to that +requirement after upstream init — this test pins that Omni-specific workaround +exactly. + +It drives the Omni runner's OWN method (with the upstream ``super()`` call +stubbed and a fake builder standing in), so a change to the resize is caught and +a failure is unambiguously an Omni-runner regression. Pure CPU — no model, no +CUDA, no flashinfer. +""" + +from types import SimpleNamespace + +import pytest +import torch + +from vllm_omni.worker.gpu_model_runner import OmniGPUModelRunner + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +# The class ``super().initialize_metadata_builders`` dispatches to. +_UPSTREAM_RUNNER = OmniGPUModelRunner.__mro__[1] + + +def _builder(*, prealloc: int | None, max_num_splits: int) -> SimpleNamespace: + sm = None if prealloc is None else torch.zeros(prealloc, dtype=torch.int32) + return SimpleNamespace(scheduler_metadata=sm, max_num_splits=max_num_splits) + + +def _make_runner(monkeypatch, *, max_num_seqs: int, builders: list) -> OmniGPUModelRunner: + # Stub the upstream initialize_metadata_builders; our fake runner already + # carries the attn_groups the Omni override iterates over. + monkeypatch.setattr(_UPSTREAM_RUNNER, "initialize_metadata_builders", lambda self, *a, **k: None) + runner = object.__new__(OmniGPUModelRunner) + runner.scheduler_config = SimpleNamespace(max_num_seqs=max_num_seqs) + runner.cache_config = SimpleNamespace(enable_prefix_caching=False) # skip omni_prefix_cache branch + runner.attn_groups = [[SimpleNamespace(metadata_builders=builders)]] + return runner + + +def test_resizes_up_when_split_scheduling_needs_more(monkeypatch): + max_num_seqs, splits, prealloc = 32, 4, 33 # 33 == upstream default (max_num_seqs + 1) + builder = _builder(prealloc=prealloc, max_num_splits=splits) + runner = _make_runner(monkeypatch, max_num_seqs=max_num_seqs, builders=[builder]) + + OmniGPUModelRunner.initialize_metadata_builders(runner, None, None) + + assert builder.scheduler_metadata.shape[0] == max_num_seqs * splits + 1 + assert builder.scheduler_metadata.dtype == torch.int32 + + +def test_no_resize_when_not_split_scheduling(monkeypatch): + builder = _builder(prealloc=33, max_num_splits=1) # max_num_splits <= 1 -> workaround skipped + runner = _make_runner(monkeypatch, max_num_seqs=32, builders=[builder]) + original = builder.scheduler_metadata + + OmniGPUModelRunner.initialize_metadata_builders(runner, None, None) + + assert builder.scheduler_metadata is original # untouched + + +def test_no_resize_when_prealloc_already_sufficient(monkeypatch): + max_num_seqs, splits = 32, 4 + builder = _builder(prealloc=max_num_seqs * splits + 1, max_num_splits=splits) # already big enough + runner = _make_runner(monkeypatch, max_num_seqs=max_num_seqs, builders=[builder]) + original = builder.scheduler_metadata + + OmniGPUModelRunner.initialize_metadata_builders(runner, None, None) + + assert builder.scheduler_metadata is original + + +def test_no_crash_when_builder_has_no_scheduler_metadata(monkeypatch): + builder = _builder(prealloc=None, max_num_splits=4) # scheduler_metadata is None + runner = _make_runner(monkeypatch, max_num_seqs=32, builders=[builder]) + + OmniGPUModelRunner.initialize_metadata_builders(runner, None, None) + + assert builder.scheduler_metadata is None diff --git a/tests/worker/test_gpu_ar_model_runner.py b/tests/worker/test_gpu_ar_model_runner.py index daab10ba0..a1f885c15 100644 --- a/tests/worker/test_gpu_ar_model_runner.py +++ b/tests/worker/test_gpu_ar_model_runner.py @@ -802,3 +802,95 @@ def test_build_omni_output_falls_back_to_mm_cpu_without_prefix_merge(monkeypatch assert torch.equal(output.inter_stage_outputs[0]["codes.audio"], codes[0:1]) assert torch.equal(output.inter_stage_outputs[1]["codes.audio"], codes[1:2]) assert output.multimodal_outputs is None + + +# --- builder gap-fill: contract corners not covered by the tests above --- +def test_build_omni_output_never_leaks_internal_pooler_output_on_wire(monkeypatch): + """The internal pooler_output feeds full-payload accumulation only; the wire + object's ``pooler_output`` is always None (builder NOTE at gpu_ar:1738).""" + runner = _make_async_output_runner() + + monkeypatch.setattr( + GPUARModelRunner, + "_resolve_pooler_payload_req_ids", + lambda self, req_ids: ("audio", req_ids), + ) + monkeypatch.setattr(GPUARModelRunner, "_should_accumulate_full_payload_output", lambda self: True) + monkeypatch.setattr( + GPUARModelRunner, + "accumulate_full_payload_output", + lambda self, rid, payload, request: None, + ) + monkeypatch.setattr(GPUARModelRunner, "get_omni_connector_output", lambda self: None) + + output = GPUARModelRunner._build_omni_model_runner_output_from_snapshot( + runner, + scheduler_output=SimpleNamespace( + total_num_scheduled_tokens=3, + num_scheduled_tokens={"r1": 1, "r2": 2}, + ), + hidden_states=torch.tensor([[1.0], [2.0], [3.0]]), + staged_hidden_states_cpu=None, + multimodal_outputs={"foo": torch.tensor([10.0, 20.0, 30.0])}, + req_ids_output_copy=["r1", "r2"], + req_id_to_index_output_copy={"r1": 0, "r2": 1}, + valid_sampled_token_ids=[[101], [102]], + logprobs_lists=None, + prompt_logprobs_dict={}, + num_nans_in_logits=None, + kv_connector_output=None, + ec_connector_output=None, + cudagraph_stats=None, + kv_extracted_req_ids=None, + num_scheduled_tokens_np=np.array([1, 2], dtype=np.int32), + query_start_loc_cpu=torch.tensor([0, 1], dtype=torch.long), + ) + + assert output.pooler_output is None + + +def test_build_omni_output_splits_client_mm_from_inter_stage_keys(monkeypatch): + """The inter-stage vs client-mm partition: a client-root key ("audio") lands in + ``multimodal_outputs`` while inter-stage keys ("hidden") stay in + ``inter_stage_outputs`` (partition_flat_payload / _CLIENT_MM_ROOT_KEYS).""" + runner = _make_async_output_runner() + + monkeypatch.setattr( + GPUARModelRunner, + "_resolve_pooler_payload_req_ids", + lambda self, req_ids: ("audio", req_ids), + ) + monkeypatch.setattr(GPUARModelRunner, "_should_accumulate_full_payload_output", lambda self: False) + monkeypatch.setattr(GPUARModelRunner, "get_omni_connector_output", lambda self: None) + + output = GPUARModelRunner._build_omni_model_runner_output_from_snapshot( + runner, + scheduler_output=SimpleNamespace( + total_num_scheduled_tokens=3, + num_scheduled_tokens={"r1": 1, "r2": 2}, + ), + hidden_states=torch.tensor([[1.0], [2.0], [3.0]]), + staged_hidden_states_cpu=None, + # "audio" root is a client mm key; "codes.audio" root ("codes") is inter-stage. + multimodal_outputs={"audio": torch.tensor([10.0, 20.0, 30.0])}, + req_ids_output_copy=["r1", "r2"], + req_id_to_index_output_copy={"r1": 0, "r2": 1}, + valid_sampled_token_ids=[[101], [102]], + logprobs_lists=None, + prompt_logprobs_dict={}, + num_nans_in_logits=None, + kv_connector_output=None, + ec_connector_output=None, + cudagraph_stats=None, + kv_extracted_req_ids=None, + num_scheduled_tokens_np=np.array([1, 2], dtype=np.int32), + query_start_loc_cpu=torch.tensor([0, 1], dtype=torch.long), + ) + + # Client mm key routed to the wire multimodal_outputs, per request. + assert output.multimodal_outputs is not None + assert "audio" in output.multimodal_outputs[0] + # Inter-stage hidden stayed on inter_stage_outputs and did NOT leak to client. + assert output.inter_stage_outputs is not None + assert "hidden" in output.inter_stage_outputs[0] + assert "audio" not in output.inter_stage_outputs[0] diff --git a/tests/worker/test_memory_utils.py b/tests/worker/test_memory_utils.py new file mode 100644 index 000000000..672fa4648 --- /dev/null +++ b/tests/worker/test_memory_utils.py @@ -0,0 +1,71 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Characterization tests for ``vllm_omni.worker.memory_utils``. + +``request_memory_tolerant`` is still live in both workers +(``gpu_ar_worker.py`` / ``gpu_generation_worker.py``); a later change makes its cap loud +but must not change the numeric contract. These tests pin that contract with a +mocked ``MemorySnapshot`` — pure CPU, no GPU / NVML. +""" + +import math +from types import SimpleNamespace + +import pytest + +from vllm_omni.worker.memory_utils import request_memory_tolerant + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +GIB = 1024**3 + + +def _snapshot(total: int, free: int) -> SimpleNamespace: + # Only the fields request_memory_tolerant reads (incl. device_ for the log). + return SimpleNamespace(total_memory=total, free_memory=free, device_="cuda:0") + + +def _cache_config(util: float) -> SimpleNamespace: + return SimpleNamespace(gpu_memory_utilization=util) + + +def test_passes_through_requested_when_free_suffices(): + snap = _snapshot(total=40 * GIB, free=40 * GIB) + cfg = _cache_config(0.5) + + out = request_memory_tolerant(snap, cfg) + + assert out == math.ceil(40 * GIB * 0.5) + + +def test_caps_to_free_when_insufficient(): + cfg = _cache_config(0.9) + requested = math.ceil(40 * GIB * 0.9) + snap = _snapshot(total=40 * GIB, free=10 * GIB) # free < requested + + out = request_memory_tolerant(snap, cfg) + + assert out == 10 * GIB + assert out < requested + + +def test_boundary_free_equals_requested_is_not_capped(): + util = 0.5 + total = 40 * GIB + requested = math.ceil(total * util) + snap = _snapshot(total=total, free=requested) # strict `<`, so no cap at equality + + out = request_memory_tolerant(snap, _cache_config(util)) + + assert out == requested + + +def test_requested_uses_ceil_of_total_times_util(): + # 7 GiB * 0.3333... rounds UP (math.ceil), not down. + total = 7 * GIB + util = 1 / 3 + snap = _snapshot(total=total, free=total) + + out = request_memory_tolerant(snap, _cache_config(util)) + + assert out == math.ceil(total * util) diff --git a/tests/worker/test_payload_span.py b/tests/worker/test_payload_span.py new file mode 100644 index 000000000..5eb4d8319 --- /dev/null +++ b/tests/worker/test_payload_span.py @@ -0,0 +1,166 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Characterization tests for ``vllm_omni.worker.payload_span``. + +These pin the CURRENT behaviour of the thinker-decode span helpers so the later +worker refactors that touch span +plumbing have a regression net. Pure CPU — no model, no CUDA. +""" + +import pytest +import torch + +from vllm_omni.worker.payload_span import ( + get_tensor_span, + get_tensor_span_row, + merge_tensor_spans, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +def _rows(n: int, *, start_val: int = 0, hidden: int = 1) -> torch.Tensor: + """A ``[n, hidden]`` tensor whose first column counts up from ``start_val``. + + ``[n, hidden]`` mirrors the connector async-chunk shape (tokens x hidden). + """ + base = torch.arange(start_val, start_val + n, dtype=torch.float32).unsqueeze(1) + return base.repeat(1, hidden) + + +# --------------------------------------------------------------------------- # +# get_tensor_span # +# --------------------------------------------------------------------------- # +def test_get_tensor_span_valid_returns_triplet(): + tensor = _rows(3, hidden=4) + payload = {"emb": tensor, "start": 5, "end": 8} + + span = get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") + + assert span is not None + got_tensor, start, end = span + assert got_tensor is tensor + assert (start, end) == (5, 8) + + +def test_get_tensor_span_non_tensor_returns_none(): + payload = {"emb": None, "start": 0, "end": 0} + assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None + + +def test_get_tensor_span_non_int_bounds_returns_none(): + tensor = _rows(2) + # start is a float, not an int -> rejected. + payload = {"emb": tensor, "start": 0.0, "end": 2} + assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None + + +def test_get_tensor_span_negative_start_returns_none(): + tensor = _rows(1) + payload = {"emb": tensor, "start": -1, "end": 0} + assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None + + +def test_get_tensor_span_end_before_start_returns_none(): + tensor = _rows(2) + payload = {"emb": tensor, "start": 5, "end": 3} + assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None + + +def test_get_tensor_span_length_mismatch_returns_none(): + tensor = _rows(3) # tensor.shape[0] == 3 + payload = {"emb": tensor, "start": 0, "end": 2} # end - start == 2 + assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None + + +# --------------------------------------------------------------------------- # +# merge_tensor_spans # +# --------------------------------------------------------------------------- # +def test_merge_none_operand_returns_none(): + span = (_rows(2), 0, 2) + assert merge_tensor_spans(None, span) is None + assert merge_tensor_spans(span, None) is None + + +def test_merge_adjacent_spans_concatenates(): + existing = (_rows(3, start_val=0), 0, 3) + incoming = (_rows(2, start_val=3), 3, 5) + + merged = merge_tensor_spans(existing, incoming) + + assert merged is not None + tensor, start, end = merged + assert (start, end) == (0, 5) + assert tensor.shape[0] == 5 + assert tensor[:, 0].tolist() == [0.0, 1.0, 2.0, 3.0, 4.0] + + +def test_merge_overlapping_spans_trims_incoming(): + # existing covers [0, 3); incoming covers [2, 5) -> 1 row of overlap. + existing = (_rows(3, start_val=0), 0, 3) + incoming = (_rows(3, start_val=100), 2, 5) + + merged = merge_tensor_spans(existing, incoming) + + assert merged is not None + tensor, start, end = merged + assert (start, end) == (0, 5) + # First 3 rows from existing, then incoming's tail (overlap=1 row trimmed). + assert tensor[:, 0].tolist() == [0.0, 1.0, 2.0, 101.0, 102.0] + + +def test_merge_fully_overlapping_incoming_keeps_existing(): + # incoming [1, 3) is fully covered by existing [0, 3): overlap >= incoming len. + existing = (_rows(3, start_val=0), 0, 3) + incoming = (_rows(2, start_val=100), 1, 3) + + merged = merge_tensor_spans(existing, incoming) + + assert merged is not None + tensor, start, end = merged + assert (start, end) == (0, 3) + assert tensor is existing[0] # returned unchanged + + +def test_merge_non_contiguous_spans_returns_none(): + # gap between existing end (3) and incoming start (5). + existing = (_rows(3, start_val=0), 0, 3) + incoming = (_rows(2, start_val=100), 5, 7) + + assert merge_tensor_spans(existing, incoming) is None + + +def test_merge_casts_incoming_to_existing_dtype_and_device(): + existing = (_rows(2, start_val=0).to(torch.float32), 0, 2) + incoming = (torch.arange(2, 4, dtype=torch.int64).unsqueeze(1), 2, 4) + + merged = merge_tensor_spans(existing, incoming) + + assert merged is not None + tensor, _, _ = merged + assert tensor.dtype == torch.float32 + assert tensor.device == existing[0].device + + +# --------------------------------------------------------------------------- # +# get_tensor_span_row # +# --------------------------------------------------------------------------- # +def test_get_tensor_span_row_in_range(): + span = (_rows(3, start_val=10), 5, 8) # rows map to indices 5, 6, 7 + row = get_tensor_span_row(span, 6) + assert row is not None + assert row[0].item() == 11.0 # second row (start_val 10 + 1) + + +def test_get_tensor_span_row_none_span_returns_none(): + assert get_tensor_span_row(None, 0) is None + + +def test_get_tensor_span_row_below_start_returns_none(): + span = (_rows(3), 5, 8) + assert get_tensor_span_row(span, 4) is None + + +def test_get_tensor_span_row_at_or_above_end_returns_none(): + span = (_rows(3), 5, 8) + assert get_tensor_span_row(span, 8) is None diff --git a/tests/worker/test_sampler_history_backfill.py b/tests/worker/test_sampler_history_backfill.py new file mode 100644 index 000000000..5f1b520b6 --- /dev/null +++ b/tests/worker/test_sampler_history_backfill.py @@ -0,0 +1,142 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Golden characterization of ``_build_model_sampler_output_token_ids``. + +The base runner (``OmniGPUModelRunner``) and the AR runner +(``GPUARModelRunner``) implement the SAME async-placeholder backfill, but the +AR copy adds a trailing crop step: after backfill it truncates every history at +the first remaining ``-1``, while the base copy KEEPS residual ``-1``s. + +A later change collapses these two copies into one; these goldens pin today's divergence +exactly (crop vs keep) so that merge is provably behaviour-preserving. Pure CPU. +""" + +from types import SimpleNamespace + +import pytest +import torch + +from vllm_omni.worker.gpu_ar_model_runner import GPUARModelRunner +from vllm_omni.worker.gpu_model_runner import OmniGPUModelRunner + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +def _runner( + cls, + *, + req_ids, + req_output_token_ids, + sampled_token_ids_cpu, + prev_req_id_to_index, +): + """Build a fake runner exposing only the ``input_batch`` fields the method reads.""" + runner = object.__new__(cls) + event = SimpleNamespace(synchronize=lambda: None) + runner.input_batch = SimpleNamespace( + req_output_token_ids=req_output_token_ids, + req_ids=req_ids, + sampled_token_ids_cpu=(None if sampled_token_ids_cpu is None else torch.tensor(sampled_token_ids_cpu)), + async_copy_ready_event=event, + prev_req_id_to_index=prev_req_id_to_index, + ) + return runner + + +def _build(cls, **kwargs): + return cls._build_model_sampler_output_token_ids(_runner(cls, **kwargs)) + + +# --------------------------------------------------------------------------- # +# Shared behaviour — base and AR agree # +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner]) +def test_full_backfill_no_residual_placeholder(cls): + # One trailing placeholder, one sampled token -> fully resolved for both. + out = _build( + cls, + req_ids=["r0"], + req_output_token_ids=[[10, 11, -1]], + sampled_token_ids_cpu=[[42]], + prev_req_id_to_index={"r0": 0}, + ) + assert out == [[10, 11, 42]] + + +@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner]) +def test_spec_decode_trailing_pad_in_sampled_is_trimmed(cls): + # sampled row has a trailing -1 pad; only the real prefix [42, 43] is used. + out = _build( + cls, + req_ids=["r0"], + req_output_token_ids=[[10, -1, -1]], + sampled_token_ids_cpu=[[42, 43, -1]], + prev_req_id_to_index={"r0": 0}, + ) + assert out == [[10, 42, 43]] + + +@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner]) +def test_no_trailing_placeholder_is_untouched(cls): + out = _build( + cls, + req_ids=["r0"], + req_output_token_ids=[[10, 11, 12]], + sampled_token_ids_cpu=[[42]], + prev_req_id_to_index={"r0": 0}, + ) + assert out == [[10, 11, 12]] + + +@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner]) +def test_early_return_when_sampled_absent_keeps_placeholders(cls): + # sampled_token_ids_cpu is None -> both copies return BEFORE the crop step, + # so even the AR copy keeps residual -1 here. + out = _build( + cls, + req_ids=["r0"], + req_output_token_ids=[[10, -1]], + sampled_token_ids_cpu=None, + prev_req_id_to_index={"r0": 0}, + ) + assert out == [[10, -1]] + + +# --------------------------------------------------------------------------- # +# Divergence — base KEEPS residual -1, AR CROPS at the first -1 # +# --------------------------------------------------------------------------- # +_RESIDUAL_CASE = dict( + req_ids=["r0"], + req_output_token_ids=[[10, -1, -1]], # two placeholders + sampled_token_ids_cpu=[[42]], # only one sampled token -> one -1 remains + prev_req_id_to_index={"r0": 0}, +) + + +def test_base_keeps_residual_placeholder(): + out = _build(OmniGPUModelRunner, **_RESIDUAL_CASE) + assert out == [[10, 42, -1]] + + +def test_ar_crops_residual_placeholder(): + out = _build(GPUARModelRunner, **_RESIDUAL_CASE) + assert out == [[10, 42]] + + +_ABSENT_FROM_PREV_CASE = dict( + req_ids=["r0"], + req_output_token_ids=[[10, -1]], + sampled_token_ids_cpu=[[42]], + prev_req_id_to_index={}, # r0 not present last step -> no backfill happens +) + + +def test_base_keeps_placeholder_when_request_absent_from_prev_step(): + out = _build(OmniGPUModelRunner, **_ABSENT_FROM_PREV_CASE) + assert out == [[10, -1]] + + +def test_ar_crops_placeholder_when_request_absent_from_prev_step(): + # No backfill occurred, but the AR crop pass still truncates at the -1. + out = _build(GPUARModelRunner, **_ABSENT_FROM_PREV_CASE) + assert out == [[10]] diff --git a/tests/worker/test_worker_base.py b/tests/worker/test_worker_base.py new file mode 100644 index 000000000..d470d86a8 --- /dev/null +++ b/tests/worker/test_worker_base.py @@ -0,0 +1,237 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Characterization tests for ``vllm_omni.worker.base.OmniGPUWorkerBase``. + +Pins the CURRENT behaviour of ``determine_available_memory`` (BOTH the NVML / +process-scoped arm AND the profiling-fallback arm — this base still carries the +NVML branch), plus the +memory-pool / sleep / wake-up plumbing that a later change may touch. + +Workers are constructed via ``object.__new__`` with only the attributes each +method reads; module-level collaborators are monkeypatched. Pure CPU. +""" + +from contextlib import contextmanager, nullcontext +from types import SimpleNamespace + +import pytest +import torch + +import vllm_omni.worker.base as base +from vllm_omni.worker.base import OmniGPUWorkerBase + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +GIB = 1024**3 + + +def _fake_memory_profiling(*, non_torch: int, torch_peak: int): + """A stand-in for ``vllm.utils.mem_utils.memory_profiling`` context manager.""" + + @contextmanager + def _mp(snapshot, weights_memory): # noqa: ARG001 - signature parity only + yield SimpleNamespace( + non_torch_increase=non_torch, + torch_peak_increase=torch_peak, + ) + + return _mp + + +def _make_worker(*, requested_memory: int, kv_cache_memory_bytes: int = 0, model_memory_usage: int = 0): + worker = object.__new__(OmniGPUWorkerBase) + worker.cache_config = SimpleNamespace(kv_cache_memory_bytes=kv_cache_memory_bytes) + worker.model_runner = SimpleNamespace( + profile_run=lambda: None, + model_memory_usage=model_memory_usage, + ) + worker.init_snapshot = object() # unused once memory_profiling is faked + worker.requested_memory = requested_memory + worker.local_rank = 0 + return worker + + +# --------------------------------------------------------------------------- # +# determine_available_memory # +# --------------------------------------------------------------------------- # +def test_determine_available_memory_nvml_arm(monkeypatch): + """NVML available -> available = requested - per-process memory.""" + worker = _make_worker(requested_memory=30 * GIB, model_memory_usage=10 * GIB) + monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=1 * GIB, torch_peak=2 * GIB)) + monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: True) + monkeypatch.setattr(base, "detect_pid_host", lambda: True) + monkeypatch.setattr(base, "get_process_gpu_memory", lambda local_rank: 8 * GIB) + + out = OmniGPUWorkerBase.determine_available_memory(worker) + + # NVML arm ignores profiled usage; uses process_memory directly. + assert out == 30 * GIB - 8 * GIB + + +def test_determine_available_memory_profiling_fallback_arm(monkeypatch): + """NVML unavailable -> available = requested - (weights + peak + non_torch).""" + worker = _make_worker(requested_memory=30 * GIB, model_memory_usage=10 * GIB) + monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=1 * GIB, torch_peak=2 * GIB)) + monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: False) + # detect_pid_host is short-circuited by the `and`, but pin it anyway. + monkeypatch.setattr(base, "detect_pid_host", lambda: True) + + out = OmniGPUWorkerBase.determine_available_memory(worker) + + profiled = 10 * GIB + 2 * GIB + 1 * GIB + assert out == 30 * GIB - profiled + + +def test_determine_available_memory_falls_back_when_not_pid_host(monkeypatch): + """detect_pid_host False also routes to the profiling fallback.""" + worker = _make_worker(requested_memory=20 * GIB, model_memory_usage=5 * GIB) + monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=0, torch_peak=0)) + monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: True) + monkeypatch.setattr(base, "detect_pid_host", lambda: False) + + out = OmniGPUWorkerBase.determine_available_memory(worker) + + assert out == 20 * GIB - 5 * GIB + + +def test_determine_available_memory_clamps_to_zero(monkeypatch): + """Over-subscription clamps the KV budget to 0, never negative.""" + worker = _make_worker(requested_memory=5 * GIB, model_memory_usage=0) + monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=0, torch_peak=0)) + monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: True) + monkeypatch.setattr(base, "detect_pid_host", lambda: True) + monkeypatch.setattr(base, "get_process_gpu_memory", lambda local_rank: 8 * GIB) + + out = OmniGPUWorkerBase.determine_available_memory(worker) + + assert out == 0 + + +def test_determine_available_memory_kv_cache_short_circuit(monkeypatch): + """A pre-set kv_cache_memory_bytes short-circuits profiling and is returned.""" + worker = _make_worker(requested_memory=30 * GIB, kv_cache_memory_bytes=7 * GIB) + profile_calls = [] + worker.model_runner.profile_run = lambda: profile_calls.append(True) + # is_rocm gates only an extra synchronize on the short-circuit path. + monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(is_rocm=lambda: False)) + + out = OmniGPUWorkerBase.determine_available_memory(worker) + + assert out == 7 * GIB + assert profile_calls == [True] # profile_run still runs before returning + + +def test_determine_available_memory_kv_short_circuit_syncs_on_rocm(monkeypatch): + """On ROCm the short-circuit adds an accelerator synchronize (a later change may remove it).""" + worker = _make_worker(requested_memory=30 * GIB, kv_cache_memory_bytes=7 * GIB) + monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(is_rocm=lambda: True)) + synced = [] + monkeypatch.setattr(torch.accelerator, "synchronize", lambda: synced.append(True)) + + out = OmniGPUWorkerBase.determine_available_memory(worker) + + assert out == 7 * GIB + assert synced == [True] # ROCm path synchronizes; non-ROCm path (above) does not + + +# --------------------------------------------------------------------------- # +# _maybe_get_memory_pool_context # +# --------------------------------------------------------------------------- # +def test_memory_pool_context_disabled_returns_nullcontext(): + worker = object.__new__(OmniGPUWorkerBase) + worker.rank = 0 + worker.cache_config = SimpleNamespace(enable_sleep_mode=False) + # No vllm_config attribute -> v1_config_enabled stays False. + + ctx = OmniGPUWorkerBase._maybe_get_memory_pool_context(worker, "weights") + + assert isinstance(ctx, type(nullcontext())) + + +def test_memory_pool_context_enabled_uses_cumem_pool_with_tag(monkeypatch): + import vllm.device_allocator.cumem as cumem_mod + + worker = object.__new__(OmniGPUWorkerBase) + worker.rank = 0 + worker.cache_config = SimpleNamespace(enable_sleep_mode=True) + monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(synchronize=lambda: None)) + + recorded = {} + + class FakeAllocator: + @classmethod + def get_instance(cls): + return cls() + + def use_memory_pool(self, tag): + recorded["tag"] = tag + return "POOL_CTX" + + monkeypatch.setattr(cumem_mod, "CuMemAllocator", FakeAllocator) + + ctx = OmniGPUWorkerBase._maybe_get_memory_pool_context(worker, "weights") + + assert ctx == "POOL_CTX" + assert recorded["tag"] == "weights" + + +# --------------------------------------------------------------------------- # +# sleep / wake_up # +# --------------------------------------------------------------------------- # +def _fake_platform(): + return SimpleNamespace( + get_current_memory_usage=lambda device: 0, + empty_cache=lambda: None, + synchronize=lambda: None, + ) + + +@pytest.mark.parametrize( + ("level", "expected_tags"), + [(1, ("weights",)), (2, tuple())], +) +def test_sleep_offload_tags_by_level(monkeypatch, level, expected_tags): + import vllm.device_allocator.cumem as cumem_mod + + worker = object.__new__(OmniGPUWorkerBase) + worker.rank = 0 + worker.device = "cuda:0" + monkeypatch.setattr(base, "current_omni_platform", _fake_platform()) + + calls = {} + + class FakeAllocator: + @classmethod + def get_instance(cls): + return cls() + + def sleep(self, offload_tags): + calls["offload_tags"] = offload_tags + + monkeypatch.setattr(cumem_mod, "CuMemAllocator", FakeAllocator) + + assert OmniGPUWorkerBase.sleep(worker, level=level) is True + assert calls["offload_tags"] == expected_tags + + +def test_wake_up_forwards_tags_to_allocator(monkeypatch): + import vllm.device_allocator.cumem as cumem_mod + + worker = object.__new__(OmniGPUWorkerBase) + worker.rank = 0 + monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(synchronize=lambda: None)) + + calls = {} + + class FakeAllocator: + @classmethod + def get_instance(cls): + return cls() + + def wake_up(self, tags): + calls["tags"] = tags + + monkeypatch.setattr(cumem_mod, "CuMemAllocator", FakeAllocator) + + assert OmniGPUWorkerBase.wake_up(worker, tags=["weights"]) is True + assert calls["tags"] == ["weights"] diff --git a/tools/baselines/omni_hash_baseline.py b/tools/baselines/omni_hash_baseline.py new file mode 100644 index 000000000..1bb93099b --- /dev/null +++ b/tools/baselines/omni_hash_baseline.py @@ -0,0 +1,276 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Record / verify a deterministic text+audio output hash baseline for an Omni model. + +This is the cheap "output is bit-identical" gate for the behaviour-preserving +worker/model-runner changes: record a baseline on the bench box +BEFORE any code moves, then re-run with ``--check`` after each refactor. + +Determinism: **batch size 1** (one prompt per ``generate`` call) + **fixed seed**. +Thinker and code2wav run greedy (temperature 0); the talker's speech sampler needs +temperature to emit valid codec tokens, so it runs at its natural sampling with a +fixed seed (reproducible via seed + batch-1, not pure-greedy). 3 fixed prompts, run +with async output OFF and ON (6 entries total). Hashes are only meaningful WITHIN a single recorded +environment (driver/cuda/torch/gpu), so the baseline JSON records the env and +``--check`` refuses to compare across a changed env unless ``--force-env``. + +Baseline JSON schema (results keyed by BOTH prompt_id and async_mode so the +matrix is self-identifying):: + + { + "model": "...", + "commit": "", + "env": {"driver": "", "cuda": "", "torch": "...", "gpu": "..."}, + "results": [ + {"prompt_id": 0, "async_mode": false, "text_sha": "...", "wav_sha": "..."}, + ... # exactly len(FIXED_PROMPTS) * 2 entries + ] + } + +The file content is JSON, but committed baselines use a NON-``.json`` extension +(``.baseline``) because the repo ``.gitignore`` excludes ``*.json``; ``--check`` +loads by content, so the extension does not matter to the script. + +Usage:: + + # record (run twice, confirm the two runs are identical before committing) + python tools/baselines/omni_hash_baseline.py --yaml \ + --out tools/baselines/_hash.baseline + # verify after a refactor (same bench box) + python tools/baselines/omni_hash_baseline.py --yaml \ + --check tools/baselines/_hash.baseline +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +import sys + +import torch +from vllm.sampling_params import SamplingParams + +from vllm_omni.entrypoints.omni import Omni + +SEED = 42 +DEFAULT_SYSTEM = ( + "You are Qwen, a virtual human developed by the Qwen Team, Alibaba Group, " + "capable of perceiving auditory and visual inputs, as well as generating " + "text and speech." +) + +# 3 fixed prompts — do NOT edit once a baseline is recorded (prompt_id is positional). +FIXED_PROMPTS = [ + "Explain the system architecture for a scalable audio generation pipeline. Answer in 15 words.", + "Describe how ocean tides work. Answer in 15 words.", + "List three benefits of unit tests. Answer in 15 words.", +] +ASYNC_MODES = (False, True) + + +def _text_prompt(question: str) -> dict: + return { + "prompt": ( + f"<|im_start|>system\n{DEFAULT_SYSTEM}<|im_end|>\n" + f"<|im_start|>user\n{question}<|im_end|>\n" + f"<|im_start|>assistant\n" + ) + } + + +def _sha(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _git_commit() -> str: + try: + return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() + except Exception: + return "unknown" + + +def _driver_version() -> str: + """The actual NVIDIA driver version (NOT torch.version.cuda, which is the build).""" + try: + out = subprocess.check_output(["nvidia-smi", "--query-gpu=driver_version", "--format=csv,noheader"], text=True) + return out.strip().splitlines()[0].strip() + except Exception: + return "unknown" + + +def _env_fingerprint() -> dict: + gpu = torch.cuda.get_device_name(0) if torch.cuda.is_available() else "cpu" + return { + "driver": _driver_version(), + "cuda": str(torch.version.cuda), + "torch": torch.__version__, + "gpu": gpu, + } + + +def _sampling_params() -> list[SamplingParams]: + # thinker + code2wav: greedy. talker: seeded (temperature>0) — its speech sampler + # needs temperature to emit valid codec tokens; the fixed seed keeps it reproducible. + thinker = SamplingParams( + temperature=0.0, top_p=1.0, top_k=-1, max_tokens=128, seed=SEED, detokenize=True, repetition_penalty=1.1 + ) + talker = SamplingParams( + temperature=0.9, top_p=0.8, top_k=40, max_tokens=256, seed=SEED, detokenize=True, repetition_penalty=1.05 + ) + code2wav = SamplingParams( + temperature=0.0, top_p=1.0, top_k=-1, max_tokens=256, seed=SEED, detokenize=True, repetition_penalty=1.1 + ) + return [thinker, talker, code2wav] + + +def _audio_bytes(audio) -> bytes: + """Deterministic bytes for one request's audio, handling chunked (list) output. + + async-chunk mode can emit a list of audio chunks instead of a single tensor; + concatenate them along the last axis in emit order. + """ + if isinstance(audio, (list, tuple)): + chunks = [a for a in audio if a is not None] + if not chunks: + return b"" + audio = torch.cat([torch.as_tensor(a).detach().to(torch.float32).cpu() for a in chunks], dim=-1) + if audio is None: + return b"" + return torch.as_tensor(audio).detach().to(torch.float32).cpu().numpy().tobytes() + + +def _run_once(model: str, yaml_path: str, *, async_mode: bool) -> list[dict]: + """One deterministic pass, batch size 1 per prompt; returns per-prompt hashes.""" + # ``async_chunk`` is the repo's documented async-output toggle for streaming + # (see tests/e2e/accuracy/qwen3_omni). Confirm the exact knob at bench time. + omni = Omni( + model, + stage_configs_path=yaml_path, + stage_init_timeout=1800, + init_timeout=3600, + async_chunk=async_mode, + ) + results: list[dict] = [] + try: + for prompt_id, question in enumerate(FIXED_PROMPTS): + text: str | None = None + wav: bytes = b"" + # Batch size 1: submit a single prompt so scheduling is deterministic. + for step in omni.generate([_text_prompt(question)], _sampling_params(), py_generator=False): + out = step.request_output + if step.final_output_type == "text": + text = out.outputs[0].text # exact text, no strip -> truly bit-identical + elif step.final_output_type == "audio": + wav = _audio_bytes(out.outputs[0].multimodal_output["audio"]) + if text is None: + raise RuntimeError(f"prompt_id={prompt_id} async={async_mode}: produced no text output") + if not wav: + raise RuntimeError(f"prompt_id={prompt_id} async={async_mode}: produced no/empty audio output") + results.append( + { + "prompt_id": prompt_id, + "async_mode": async_mode, + "text_sha": _sha(text.encode("utf-8")), + "wav_sha": _sha(wav), + } + ) + finally: + omni.close() + return results + + +def _record(model: str, yaml_path: str) -> dict: + results: list[dict] = [] + for async_mode in ASYNC_MODES: + results.extend(_run_once(model, yaml_path, async_mode=async_mode)) + return {"model": model, "commit": _git_commit(), "env": _env_fingerprint(), "results": results} + + +def _key(record: dict): + return (record["prompt_id"], record["async_mode"]) + + +def _expected_keys() -> set: + return {(pid, mode) for pid in range(len(FIXED_PROMPTS)) for mode in ASYNC_MODES} + + +def _check(model: str, yaml_path: str, baseline_path: str, *, force_env: bool) -> int: + with open(baseline_path) as fh: + baseline = json.load(fh) + + now = _record(model, yaml_path) + if not force_env and baseline.get("env") != now["env"]: + print( + f"ENV MISMATCH: baseline={baseline.get('env')} now={now['env']} — hashes are " + "env-specific. Re-record on this box or pass --force-env.", + file=sys.stderr, + ) + return 2 + + problems: list[str] = [] + + def _index_unique(records: list[dict], label: str) -> dict: + by_key: dict = {} + for record in records: + key = _key(record) + if key in by_key: + problems.append(f"{label}: duplicate entry for (prompt_id={key[0]}, async={key[1]})") + by_key[key] = record + return by_key + + base_by_key = _index_unique(baseline["results"], "baseline") + now_by_key = _index_unique(now["results"], "current-run") + expected = _expected_keys() + + # Guard against duplicate-collapse: the record count must match the key count. + if len(baseline["results"]) != len(expected): + problems.append(f"baseline has {len(baseline['results'])} records, expected exactly {len(expected)}") + if set(now_by_key) != expected: + problems.append(f"current-run key set {sorted(set(now_by_key))} != expected {sorted(expected)}") + if set(base_by_key) != set(now_by_key): + missing = sorted(set(base_by_key) - set(now_by_key)) + extra = sorted(set(now_by_key) - set(base_by_key)) + problems.append(f"baseline/run key-set mismatch: missing-from-run={missing} extra-in-run={extra}") + for key in sorted(set(base_by_key) & set(now_by_key)): + b, r = base_by_key[key], now_by_key[key] + if (b["text_sha"], b["wav_sha"]) != (r["text_sha"], r["wav_sha"]): + problems.append( + f"HASH MISMATCH (prompt_id={key[0]}, async={key[1]}): " + f"text {b['text_sha'][:8]}!={r['text_sha'][:8]} / wav {b['wav_sha'][:8]}!={r['wav_sha'][:8]}" + ) + + if problems: + for problem in problems: + print(problem, file=sys.stderr) + return 1 + print(f"OK: all {len(expected)} (prompt_id, async_mode) entries match the baseline.") + return 0 + + +def main() -> int: + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("model") + p.add_argument("--yaml", required=True, help="Stage deploy config path") + p.add_argument("--out", help="Write a new baseline JSON to this path") + p.add_argument("--check", metavar="BASELINE_JSON", help="Verify current output against this baseline") + p.add_argument("--force-env", action="store_true", help="Compare hashes even if the recorded env differs") + args = p.parse_args() + + if args.check: + return _check(args.model, args.yaml, args.check, force_env=args.force_env) + + baseline = _record(args.model, args.yaml) + payload = json.dumps(baseline, indent=2) + if args.out: + with open(args.out, "w") as fh: + fh.write(payload) + print(f"Wrote baseline: {args.out}") + else: + print(payload) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/baselines/qwen3_omni_hash.baseline b/tools/baselines/qwen3_omni_hash.baseline new file mode 100644 index 000000000..7fed33016 --- /dev/null +++ b/tools/baselines/qwen3_omni_hash.baseline @@ -0,0 +1,48 @@ +{ + "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct", + "commit": "24eb4ab89cc2ad3f8cdd83f57b66ce7786a0d65c", + "env": { + "driver": "570.133.20", + "cuda": "13.0", + "torch": "2.11.0+cu130", + "gpu": "NVIDIA L20X" + }, + "results": [ + { + "prompt_id": 0, + "async_mode": false, + "text_sha": "e4bf6ed906c0dada879874a9cc09f9ba3a4399149262a831a01fd53e6a8fd043", + "wav_sha": "20a203a149005f948afd485bbea780fa06c109f3689ebb6b3110f51b3237bd35" + }, + { + "prompt_id": 1, + "async_mode": false, + "text_sha": "2e8a6b141a9747bd68aadc2931f278b89712ad7ad7ae12b25b39e193b241bf68", + "wav_sha": "1839a759bb5d8671b6352b2e3110dab5edfa718e38ff073d379176201f1eac8c" + }, + { + "prompt_id": 2, + "async_mode": false, + "text_sha": "e0aa5a513a5162024f35afee3830aeb67f2e378381726adb9d3a66b984aa652d", + "wav_sha": "74338df3e8357bc9d0c05bf141b63c45251b12807de34495097edb07fc7b19f8" + }, + { + "prompt_id": 0, + "async_mode": true, + "text_sha": "e4bf6ed906c0dada879874a9cc09f9ba3a4399149262a831a01fd53e6a8fd043", + "wav_sha": "e09f862f54b553cc7c64e1631b821a768167768812b3f2cd3d0c69d0da991ee7" + }, + { + "prompt_id": 1, + "async_mode": true, + "text_sha": "2e8a6b141a9747bd68aadc2931f278b89712ad7ad7ae12b25b39e193b241bf68", + "wav_sha": "9115d481f62a4b771c4c3d8b89d5c5a01149f002096289ab1da3f034fa388a2b" + }, + { + "prompt_id": 2, + "async_mode": true, + "text_sha": "e0aa5a513a5162024f35afee3830aeb67f2e378381726adb9d3a66b984aa652d", + "wav_sha": "6a5985c9d4ca1596518c291228cd29239e76fcaf310c1cdb5921276b7efce801" + } + ] +}