diff --git a/docs/guide/configuration.md b/docs/guide/configuration.md index 85b049d0..01a37b42 100644 --- a/docs/guide/configuration.md +++ b/docs/guide/configuration.md @@ -93,7 +93,8 @@ root model (top level or `args`) and on an inline `grader` / `extractor`, on every dialect. A derived model (`base:`) uses its root's key. With neither, the `openai_*` dialects read `OPENAI_API_KEY`; the others send -no key, so an external SGLang server started with `--api-key` needs one here. +no key, so an external SGLang or vLLM server started with `--api-key` needs +one here. ```yaml models: @@ -182,7 +183,7 @@ one entry per rollout, in `texts` order, with the same three fields. | Dialect | `n: 1` | `n > 1` | | --- | --- | --- | -| `sglang_native`, `openai_completions`, `openai_chat` | yes | yes | +| `sglang_native`, `vllm_native`, `openai_completions`, `openai_chat` | yes | yes | | `google_genai` | yes | refused | | `openai_responses` | yes | refused (no `n`) | | `anthropic_messages` | refused | refused | @@ -194,8 +195,10 @@ so changing them on a resume reruns inference. With `echo` (`alogprobs(text, echo=True, logprobs=N)` or `infer_args: {echo: true, top_logprobs: N}`), each input token's top-N lands in `ModelOutput.input_top_logprobs` (`{}` for the first); `top_logprobs` keeps -the generated tokens'. Served by `sglang_native` and `openai_completions`. -The reply grows by N entries per input token. +the generated tokens'. Served by `sglang_native`, `vllm_native` and +`openai_completions`. The reply grows by N entries per input token. + +`sglang_native` and `vllm_native` also fill token ids. ## Model Capabilities (`extra`) diff --git a/docs/guide/infer.md b/docs/guide/infer.md index 0a326487..d1b2ea10 100644 --- a/docs/guide/infer.md +++ b/docs/guide/infer.md @@ -257,6 +257,22 @@ task scores input: SGLang's `/v1/completions` cannot. `/generate` has no per-request `seed`, so an explicit one is refused; a managed engine pins `--random-seed` instead. +A vLLM `gen` model keeps `openai_completions` unless it sets `dialect: +vllm_native`, which posts to the same `/v1/completions` with vLLM's own +fields (measured on 0.26.0): + +| Need | `openai_completions` | `vllm_native` | +| --- | --- | --- | +| Input scoring | `echo`, split at `usage.prompt_tokens` | `prompt_logprobs`, checked against `prompt_token_ids` | +| `texts` of a scoring call | prompt + generation | generation only | +| Token ids | none | input and sampled (`return_token_ids`) | +| Structured output | refused | `structured_outputs` | +| Unknown options | `extra_body` | refused, except `min_p`, `repetition_penalty`, `stop_token_ids` | + +It sends one non-streaming POST and offers no reasoning or tool parsers +(they act on chat requests). `suffix` and `max_tokens: 0` are refused before +I/O. Input top-k needs `--max-logprobs` at least the breadth. + ### Model lifecycles A model `sieval run` launches may bound its engine by stage barriers diff --git a/sieval/core/models/dialect_registry.py b/sieval/core/models/dialect_registry.py index b7e37307..46fbc371 100644 --- a/sieval/core/models/dialect_registry.py +++ b/sieval/core/models/dialect_registry.py @@ -54,6 +54,10 @@ CAPABILITY_DECISIONS as SGLANG_NATIVE_CAPABILITY_DECISIONS, ) from .dialects.sglang_native import SglangNativeDialect +from .dialects.vllm_native import ( + CAPABILITY_DECISIONS as VLLM_NATIVE_CAPABILITY_DECISIONS, +) +from .dialects.vllm_native import VllmNativeDialect from .ir import ( ChatInput, ChatMessage, @@ -378,6 +382,10 @@ def _bind_sglang_native(connection: Any, requested_model_id: str) -> Dialect: return SglangNativeDialect(connection, requested_model_id) +def _bind_vllm_native(connection: Any, requested_model_id: str) -> Dialect: + return VllmNativeDialect(connection, requested_model_id) + + def _binder( function: Callable[[Any, str], Dialect], decisions: Mapping[str, DialectCapabilityDecision], @@ -433,6 +441,13 @@ def _binder( SGLANG_NATIVE_CAPABILITY_DECISIONS, ), ), + "vllm_native": _binder( + _bind_vllm_native, + cast( + Mapping[str, DialectCapabilityDecision], + VLLM_NATIVE_CAPABILITY_DECISIONS, + ), + ), } ) @@ -504,7 +519,9 @@ def _registered_decisions( "vllm_native": _spec( "vllm_native", "async_http_json", - request_seed_support=RequestSeedSupport.RESERVED, + decisions=_registered_decisions("vllm_native"), + # Measured on 0.26.0: equal seeds give equal samples at temperature 1. + request_seed_support=RequestSeedSupport.SUPPORTED, input_kinds=("completion",), input_modalities=("text",), ), @@ -628,8 +645,6 @@ def bind_dialect( def _not_implemented_message(dialect_id: str) -> str: - if dialect_id == "vllm_native": - return "dialect 'vllm_native' is reserved and has no executable binder" return f"dialect {dialect_id!r} is reserved and has no executable binder" diff --git a/sieval/core/models/dialects/vllm_native.py b/sieval/core/models/dialects/vllm_native.py new file mode 100644 index 00000000..702df31b --- /dev/null +++ b/sieval/core/models/dialects/vllm_native.py @@ -0,0 +1,873 @@ +"""vLLM native completions dialect for the provider-neutral model IR. + +vLLM's native surface is the fields its ``/v1/completions`` accepts beyond +OpenAI's, posted over raw HTTP (the OpenAI SDK would drop them from the typed +reply): + +- ``prompt_logprobs`` scores the input as its own channel, without ``echo``; +- ``return_token_ids`` gives input and sampled tokens their ids; +- ``structured_outputs`` constrains decoding to a JSON schema or any JSON. + +vLLM ignores an unknown field silently (measured on 0.26.0), so options +outside a probed set are refused rather than forwarded. + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +import json +import math +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from types import MappingProxyType +from typing import cast + +import httpx + +from sieval.core.models._shared import thaw_json_mapping +from sieval.core.models.capabilities import ( + CAPABILITY_KEYS, + Capability, + CapabilityKey, + DialectCapabilityBinding, + DialectCapabilityDecision, + Supported, + Unsupported, +) +from sieval.core.models.connection_factory import AsyncHTTPJSONConnection +from sieval.core.models.dialect import ( + DialectError, + Guarantee, + OutputContract, + OutputContractError, + OutputRule, + PreparedRequest, + RequestAudit, + RuntimePlanView, + WireEvidence, + WireObservation, + active_request_leaves, + validate_choice_logprobs, + validate_input_scoring, + validate_request_invariants, + validate_runtime_binding_plan, + validate_structured_output, + validate_top_logprobs, +) +from sieval.core.models.ir import ( + ChoiceLogprobs, + CompletionInput, + InputScoringResult, + Request, + Response, + StructuredOutput, + TokenLogprob, + TopKEntry, + UsageStats, +) +from sieval.core.types import JSONValue + +from ._http_status import raise_for_provider_status + +_OPERATION = "completions" + +_PARSER_UNSUPPORTED = ( + "/v1/completions takes a raw prompt; vLLM's reasoning and tool parsers " + "act on chat requests only" +) + +_PREFILL_UNSUPPORTED = ( + "assistant prefill is a chat input operation, not a completion one" +) + + +def _decisions() -> Mapping[CapabilityKey, DialectCapabilityDecision]: + decisions: dict[CapabilityKey, DialectCapabilityDecision] = { + "input_scoring": Supported( + DialectCapabilityBinding( + "input_scoring", + request_leaves=("scoring.input_scoring",), + response_channels=("input_scoring",), + ) + ), + "sampled_logprobs": Supported( + DialectCapabilityBinding( + "sampled_logprobs", + request_leaves=("scoring.sampled_logprobs",), + response_channels=("logprobs",), + ) + ), + "top_logprobs": Supported( + DialectCapabilityBinding( + "top_logprobs", + request_leaves=("scoring.top_logprobs",), + response_channels=("top_logprobs",), + ) + ), + "structured_output": Supported( + DialectCapabilityBinding( + "structured_output", + request_leaves=( + "structured_output.format", + "structured_output.schema", + "structured_output.name", + "structured_output.strict", + ), + response_channels=("structured_output",), + ) + ), + "reasoning": Unsupported(_PARSER_UNSUPPORTED), + "function_tools": Unsupported(_PARSER_UNSUPPORTED), + "hosted_tools": Unsupported("vLLM hosts no server-side tools"), + "stateful_session": Unsupported("/v1/completions has no response-id state"), + "opaque_continuation": Unsupported( + "/v1/completions has no opaque continuation channel" + ), + "multimodal_input": Unsupported("this dialect sends text prompts only"), + "prefill": Unsupported(_PREFILL_UNSUPPORTED), + "fim": Unsupported("vLLM refuses suffix: 'suffix is not currently supported'"), + } + if set(decisions) != set(CAPABILITY_KEYS): + raise AssertionError("vLLM native capability row is incomplete") + return MappingProxyType(decisions) + + +CAPABILITY_DECISIONS = _decisions() + + +def _validate_token_logprobs(value: object) -> None: + if not isinstance(value, tuple) or not all( + isinstance(item, TokenLogprob) for item in value + ): + raise OutputContractError("logprobs channel has invalid shape") + + +OUTPUT_CONTRACT = OutputContract( + { + "reasoning": OutputRule(Guarantee.NEVER), + "tool_calls": OutputRule(Guarantee.NEVER), + "server_tool_uses": OutputRule(Guarantee.NEVER), + "structured_output": OutputRule( + Guarantee.PRESENT_OR_ERROR, validate_structured_output + ), + "logprobs": OutputRule(Guarantee.PRESENT_OR_ERROR, _validate_token_logprobs), + "top_logprobs": OutputRule(Guarantee.PRESENT_OR_ERROR, validate_top_logprobs), + "choice_logprobs": OutputRule( + Guarantee.PRESENT_OR_ERROR, validate_choice_logprobs + ), + "input_scoring": OutputRule(Guarantee.PRESENT_OR_ERROR, validate_input_scoring), + "citations": OutputRule(Guarantee.NEVER), + "grounding": OutputRule(Guarantee.NEVER), + "session_id": OutputRule(Guarantee.NEVER), + "usage": OutputRule(Guarantee.BEST_EFFORT), + } +) + + +_CONSUMED_PATHS = frozenset( + { + "input.completion", + "sampling.temperature", + "sampling.top_p", + "sampling.top_k", + "sampling.max_tokens", + "sampling.stop", + "sampling.seed", + "sampling.frequency_penalty", + "sampling.presence_penalty", + "sampling.n", + "scoring.input_scoring", + "scoring.sampled_logprobs", + "scoring.top_logprobs", + "structured_output.format", + "structured_output.schema", + "structured_output.name", + "structured_output.strict", + } +) + +_NOOP_PATHS: Mapping[str, str] = MappingProxyType( + { + # Streamed choices carry no prompt_logprobs (measured on 0.26.0). + "scheduling.stream": "the request is sent as one non-streaming POST", + } +) + +_SCHEMA_NAME_NOOP = ( + "structured_outputs takes no schema name; the name only labels the schema" +) +_STRICT_NOOP = "vLLM's grammar-constrained decoding is always strict" + +#: vLLM sampling parameters with no request leaf, forwarded as top-level +#: fields; each was probed live. +_SAMPLING_OPTION_KEYS = frozenset({"min_p", "repetition_penalty", "stop_token_ids"}) + +#: Option keys that spell a first-class request leaf, or a field this dialect +#: derives from one, and that leaf. +_IR_OWNED_OPTION_KEYS: Mapping[str, str] = MappingProxyType( + { + "prompt": "input.completion", + "suffix": "input.completion.suffix", + "temperature": "sampling.temperature", + "top_p": "sampling.top_p", + "top_k": "sampling.top_k", + "max_tokens": "sampling.max_tokens", + "stop": "sampling.stop", + "seed": "sampling.seed", + "frequency_penalty": "sampling.frequency_penalty", + "presence_penalty": "sampling.presence_penalty", + "n": "sampling.n", + "echo": "scoring.input_scoring", + "prompt_logprobs": "scoring.input_scoring", + "logprobs": "scoring.sampled_logprobs", + "return_token_ids": "scoring.sampled_logprobs", + "structured_outputs": "structured_output.format", + "response_format": "structured_output.format", + "stream": "scheduling.stream", + "stream_options": "scheduling.stream", + } +) + + +@dataclass(frozen=True) +class _CompletionContext: + prompt: str + n: int + input_scoring: bool + sampled_logprobs: bool + top_logprobs: int + structured_output: bool + + +@dataclass(frozen=True) +class _StructuredOutputsVerifier: + """Prove ``structured_outputs`` spells the request's format and schema.""" + + expected: Mapping[str, JSONValue] + + def verify(self, body: Mapping[str, JSONValue]) -> str | None: + actual = body.get("structured_outputs") + if not isinstance(actual, Mapping) or thaw_json_mapping( + cast(Mapping[str, JSONValue], actual), "body.structured_outputs" + ) != thaw_json_mapping(self.expected, "structured_outputs"): + return "body.structured_outputs does not match the request's format" + return None + + +@dataclass(frozen=True) +class _ScoringVerifier: + """Prove a scoring field carries the request's breadth, with token ids. + + The leaf is a flag; its wire field (``prompt_logprobs`` or ``logprobs``) + holds the top-k breadth, so a ``WireObservation`` cannot compare them. + """ + + wire_field: str + breadth: int + + def verify(self, body: Mapping[str, JSONValue]) -> str | None: + actual = body.get(self.wire_field) + if type(actual) is not int or actual != self.breadth: + return f"body.{self.wire_field} does not match the request's top-k breadth" + if body.get("return_token_ids") is not True: + return "body.return_token_ids must be true for a scoring request" + return None + + +@dataclass(frozen=True) +class _LegacyPlan: + """Minimal plan used only by the direct ``arun`` call path.""" + + dialect_id: str = "vllm_native" + available_capabilities: frozenset[str] = frozenset( + key + for key, decision in CAPABILITY_DECISIONS.items() + if isinstance(decision, Supported) + ) + capability_minimums: Mapping[str, Mapping[str, JSONValue]] = field( + default_factory=dict + ) + required_output_channels: frozenset[str] = frozenset() + + +def completions_url(base_url: httpx.URL) -> httpx.URL: + """``/completions`` under the OpenAI-style base (``http://host:8000/v1``).""" + + path = base_url.path.rstrip("/") + return base_url.copy_with(path=f"{path}/{_OPERATION}", query=None) + + +def _structured_outputs(req: Request) -> dict[str, JSONValue] | None: + params = req.structured_output + if params.format is None: + return None + if params.format == "json_object": + return {"json_object": True} + assert params.schema is not None # the IR requires it for json_schema + return {"json": thaw_json_mapping(params.schema, "structured_output.schema")} + + +def _sequence(raw: object, path: str) -> Sequence[object]: + if not isinstance(raw, Sequence) or isinstance(raw, str | bytes): + raise OutputContractError(f"{path} must be a list") + return raw + + +def _mapping(raw: object, path: str) -> Mapping[str, object]: + if not isinstance(raw, Mapping): + raise OutputContractError(f"{path} must be an object") + return cast(Mapping[str, object], raw) + + +def _logprob(raw: object, path: str) -> float: + """A finite or ``-inf`` logprob; vLLM sends ``-inf`` for a masked token.""" + + if isinstance(raw, bool) or not isinstance(raw, int | float): + raise OutputContractError(f"{path} must be numeric") + value = float(raw) + if math.isnan(value) or value == math.inf: + raise OutputContractError(f"{path} must not be NaN or +inf") + return value + + +def _token_id(raw: object, path: str) -> int: + if isinstance(raw, bool) or not isinstance(raw, int) or raw < 0: + raise OutputContractError(f"{path} must be a non-negative integer") + return raw + + +def _text(raw: object, path: str) -> str: + if not isinstance(raw, str): + raise OutputContractError( + f"{path} must be a string; scoring needs detokenized text, so do " + "not launch vLLM with --skip-tokenizer-init" + ) + return raw + + +def _token_ids(raw: object, path: str) -> tuple[int, ...]: + if raw is None: + raise OutputContractError( + f"vLLM returned no {path}; the scoring lift needs return_token_ids, " + "which this server ignored (it predates the field)" + ) + items = _sequence(raw, path) + return tuple( + _token_id(item, f"{path}[{index}]") for index, item in enumerate(items) + ) + + +def _prompt_position( + raw: object, token_id: int, top_k: int, path: str +) -> tuple[TokenLogprob, tuple[TopKEntry, ...]]: + """Lift one ``{token_id: {logprob, rank, decoded_token}}`` position. + + The scored token is the entry keyed by the prompt's own id; ``top_k + 1`` + entries list it beside the top-k, even when a tie gives it a top-k rank. + """ + + entries = _mapping(raw, path) + if len(entries) not in {top_k, top_k + 1}: + raise OutputContractError( + f"{path} lists {len(entries)} entries for a top-{top_k} position" + ) + beside = len(entries) == top_k + 1 + scored: TokenLogprob | None = None + ranked: list[tuple[int, TopKEntry]] = [] + for key, value in entries.items(): + entry_path = f"{path}[{key!r}]" + try: + entry_id = int(key) + except ValueError: + raise OutputContractError(f"{entry_path} key must be a token id") from None + entry = _mapping(value, entry_path) + logprob = _logprob(entry.get("logprob"), f"{entry_path}.logprob") + text = _text(entry.get("decoded_token"), f"{entry_path}.decoded_token") + rank = entry.get("rank") + if isinstance(rank, bool) or not isinstance(rank, int) or rank < 1: + raise OutputContractError(f"{entry_path}.rank must be a positive integer") + if entry_id == token_id: + scored = TokenLogprob(token=text, logprob=logprob, token_id=entry_id) + if not (beside and entry_id == token_id): + ranked.append( + (rank, TopKEntry(token=text, logprob=logprob, token_id=entry_id)) + ) + if scored is None: + raise OutputContractError( + f"{path} does not score the prompt's own token id {token_id}" + ) + ranked.sort(key=lambda item: item[0]) + return scored, tuple(entry for _, entry in ranked) + + +def _first_token_text(prompt: str, scored: Sequence[TokenLogprob]) -> str: + """The unscored first position's text: the prompt prefix the rest leaves. + + vLLM reports ``None`` for position 0, text included; the rest decode + incrementally to the prompt's suffix (measured on CJK, byte-fallback + emoji, leading spaces and newlines). Empty for a BOS outside the prompt, + or when the decoded tokens do not spell it. + """ + + rest = "".join(item.token for item in scored) + if not prompt.endswith(rest): + return "" + return prompt[: len(prompt) - len(rest)] + + +def _input_scoring( + choice: Mapping[str, object], + prompt: str, + prompt_tokens: int | None, + top_k: int, +) -> InputScoringResult: + raw_positions = choice.get("prompt_logprobs") + if raw_positions is None: + raise OutputContractError("vLLM returned no prompt_logprobs") + positions = _sequence(raw_positions, "prompt_logprobs") + token_ids = _token_ids(choice.get("prompt_token_ids"), "prompt_token_ids") + if not positions: + raise OutputContractError("vLLM scored no prompt tokens") + # One position per prompt token, by both the ids and the usage count. + if len(positions) != len(token_ids) or ( + prompt_tokens is not None and len(positions) != prompt_tokens + ): + raise OutputContractError( + f"vLLM returned {len(positions)} prompt_logprobs positions for " + f"{len(token_ids)} prompt token ids and usage.prompt_tokens=" + f"{prompt_tokens}; refusing to score a partial prompt" + ) + if positions[0] is not None: + raise OutputContractError("vLLM scored the first prompt position") + scored: list[TokenLogprob] = [] + top: list[tuple[TopKEntry, ...]] = [()] + for index in range(1, len(positions)): + token, alternatives = _prompt_position( + positions[index], token_ids[index], top_k, f"prompt_logprobs[{index}]" + ) + scored.append(token) + top.append(alternatives) + first = TokenLogprob( + token=_first_token_text(prompt, scored), logprob=None, token_id=token_ids[0] + ) + tokens = (first, *scored) + if top_k == 0: + return InputScoringResult(token_logprobs=tokens) + return InputScoringResult(token_logprobs=tokens, top_logprobs=tuple(top)) + + +def _sampled_top( + raw: object, sampled: Sequence[TokenLogprob], top_k: int +) -> tuple[tuple[TopKEntry, ...], ...]: + """Per-position ``{text: logprob}`` maps, trimmed to the top ``top_k``. + + The map is keyed by decoded text: ``top_k + 1`` entries list the sampled + token beside the top-k, and it is dropped. Tokens that decode alike share + a key; a map that is then short, or holds another value for the sampled + token, is refused. A collision that still leaves ``top_k`` entries is + indistinguishable, since the map carries no rank. + """ + + positions = _sequence(raw, "logprobs.top_logprobs") + if len(positions) != len(sampled): + raise OutputContractError( + f"vLLM returned top logprobs for {len(positions)} of {len(sampled)} " + "sampled tokens" + ) + lifted: list[tuple[TopKEntry, ...]] = [] + for index, (position, token) in enumerate(zip(positions, sampled, strict=True)): + path = f"logprobs.top_logprobs[{index}]" + entries = { + text: _logprob(value, f"{path}[{text!r}]") + for text, value in _mapping(position, path).items() + } + if ( + len(entries) not in {top_k, top_k + 1} + or entries.get(token.token) != token.logprob + ): + raise OutputContractError( + f"{path} cannot be split into the sampled token and its top-{top_k}: " + "two tokens decode to the same text" + ) + if len(entries) == top_k + 1: + del entries[token.token] + ranked = sorted(entries.items(), key=lambda item: -item[1]) + lifted.append(tuple(TopKEntry(token=text, logprob=lp) for text, lp in ranked)) + return tuple(lifted) + + +def _choice_logprobs(choice: Mapping[str, object], top_k: int) -> ChoiceLogprobs: + raw = choice.get("logprobs") + if raw is None: + raise OutputContractError("vLLM returned no logprobs for a sampled choice") + logprobs = _mapping(raw, "logprobs") + texts = _sequence(logprobs.get("tokens"), "logprobs.tokens") + values = _sequence(logprobs.get("token_logprobs"), "logprobs.token_logprobs") + ids = _token_ids(choice.get("token_ids"), "token_ids") + if not len(texts) == len(values) == len(ids): + raise OutputContractError( + f"vLLM returned {len(texts)} sampled tokens, {len(values)} logprobs " + f"and {len(ids)} token ids" + ) + tokens = tuple( + TokenLogprob( + token=_text(text, f"logprobs.tokens[{index}]"), + logprob=_logprob(value, f"logprobs.token_logprobs[{index}]"), + token_id=token_id, + ) + for index, (text, value, token_id) in enumerate( + zip(texts, values, ids, strict=True) + ) + ) + if top_k == 0: + return ChoiceLogprobs(tokens=tokens) + return ChoiceLogprobs( + tokens=tokens, + top_logprobs=_sampled_top(logprobs.get("top_logprobs"), tokens, top_k), + ) + + +def _count(usage: Mapping[str, object], key: str) -> int: + value = usage.get(key) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise OutputContractError(f"vLLM usage.{key} must be a non-negative integer") + return value + + +def _usage(raw: object) -> UsageStats | None: + """Usage from the two counts; ``cached_tokens`` only when vLLM reports it + (``--enable-prompt-tokens-details``).""" + + if raw is None: + return None + usage = _mapping(raw, "usage") + input_tokens = _count(usage, "prompt_tokens") + output_tokens = _count(usage, "completion_tokens") + cached: int | None = None + details = usage.get("prompt_tokens_details") + if isinstance(details, Mapping): + value = cast(Mapping[str, object], details).get("cached_tokens") + if isinstance(value, int) and not isinstance(value, bool) and value >= 0: + cached = value + return UsageStats( + input_tokens=input_tokens, + output_tokens=output_tokens, + total_tokens=input_tokens + output_tokens, + cached_tokens=cached, + ) + + +def _choices(payload: Mapping[str, object], n: int) -> list[Mapping[str, object]]: + by_index: dict[int, Mapping[str, object]] = {} + for position, raw in enumerate(_sequence(payload.get("choices"), "choices")): + choice = _mapping(raw, f"choices[{position}]") + index = choice.get("index") + if isinstance(index, bool) or not isinstance(index, int) or not 0 <= index < n: + raise OutputContractError( + f"vLLM choice index {index!r} is outside the requested range [0, {n})" + ) + if index in by_index: + raise OutputContractError(f"vLLM duplicated choice index {index}") + if not isinstance(choice.get("text"), str): + raise OutputContractError(f"vLLM choice {index} has no text") + by_index[index] = choice + missing = sorted(set(range(n)) - set(by_index)) + if missing: + raise OutputContractError(f"vLLM omitted choice indexes {missing}") + return [by_index[index] for index in range(n)] + + +def _optional_string(payload: Mapping[str, object], key: str) -> str | None: + value = payload.get(key) + if value is not None and not isinstance(value, str): + raise OutputContractError(f"vLLM {key} must be a string") + return value + + +class VllmNativeDialect: + """Executable dialect for vLLM's ``/v1/completions`` with its extensions.""" + + dialect_id = "vllm_native" + connection_family = "async_http_json" + capability_decisions = CAPABILITY_DECISIONS + output_contract = OUTPUT_CONTRACT + CAPABILITIES = frozenset( + { + Capability.Completion, + Capability.InputScoring, + Capability.SampledLogprobs, + Capability.SampledLogprobsWithTokenIds, + Capability.TopKLogprobs, + Capability.StructuredOutput, + } + ) + + def __init__(self, connection: object, requested_model_id: str) -> None: + if not isinstance(connection, AsyncHTTPJSONConnection): + raise TypeError("vllm_native requires AsyncHTTPJSONConnection") + if not requested_model_id: + raise ValueError("requested_model_id must not be empty") + self._connection = connection + self._requested_model_id = requested_model_id + self._url = completions_url(connection.client.base_url) + + @property + def capabilities(self) -> frozenset[Capability]: + """Deprecated enum view of the keyed capability row.""" + + return self.CAPABILITIES + + def validate_request( + self, req: Request, audit: RequestAudit, plan: RuntimePlanView + ) -> None: + """Account for every active leaf before any client operation.""" + + audit.require_matches_request(req) + del plan + structured = req.structured_output + for path in audit.active_paths: + if path == "sampling.max_tokens" and req.sampling.max_tokens == 0: + audit.rejected(path, "vLLM requires max_tokens >= 1") + elif ( + path + in { + "structured_output.schema", + "structured_output.name", + "structured_output.strict", + } + and structured.format != "json_schema" + ): + audit.rejected( + path, "schema, name, and strict require format='json_schema'" + ) + elif path == "structured_output.strict" and structured.strict is False: + audit.rejected( + path, "vLLM cannot relax its grammar-constrained decoding" + ) + elif ( + path in _CONSUMED_PATHS + or path in _NOOP_PATHS + or path == "dialect_options" + ): + continue + elif not path.startswith("dialect_options."): + audit.rejected( + path, "the vLLM native dialect does not support this request leaf" + ) + else: + self._validate_option(req, audit, path) + + @staticmethod + def _validate_option(req: Request, audit: RequestAudit, path: str) -> None: + key = path.removeprefix("dialect_options.") + options = req.dialect_options + assert options is not None + if key in {"prefill", "prefix"}: + audit.rejected(path, _PREFILL_UNSUPPORTED) + elif key in _IR_OWNED_OPTION_KEYS: + audit.rejected(path, f"use the request leaf {_IR_OWNED_OPTION_KEYS[key]}") + elif key not in _SAMPLING_OPTION_KEYS: + audit.rejected( + path, + "unknown vLLM option (vLLM ignores unknown fields silently); " + "forwarded options are " + ", ".join(sorted(_SAMPLING_OPTION_KEYS)), + ) + elif options.values[key] is None: + audit.rejected(path, "a null option has no wire value") + + def prepare(self, req: Request, audit: RequestAudit) -> PreparedRequest: + """Lower one validated request and register exact audit evidence.""" + + audit.require_matches_request(req) + audit.raise_rejections() + if not isinstance(req.input, CompletionInput): + raise DialectError("vllm_native requires CompletionInput") + + def consume(path: str, *wire: WireEvidence) -> None: + audit.consumed(path, *wire) + + body: dict[str, JSONValue] = { + "model": self._requested_model_id, + "prompt": req.input.text, + "stream": False, + } + consume("input.completion", WireObservation(("prompt",))) + + params = req.sampling + simple: dict[str, JSONValue] = { + "temperature": params.temperature, + "top_p": params.top_p, + "top_k": params.top_k, + "max_tokens": params.max_tokens, + "stop": None if params.stop is None else list(params.stop), + "seed": params.seed, + "frequency_penalty": params.frequency_penalty, + "presence_penalty": params.presence_penalty, + } + for name, value in simple.items(): + if value is not None: + body[name] = value + consume(f"sampling.{name}", WireObservation((name,))) + if params.n != 1: + body["n"] = params.n + consume("sampling.n", WireObservation(("n",))) + + scoring = req.scoring + if scoring.input_scoring or scoring.sampled_logprobs: + body["return_token_ids"] = True + if scoring.input_scoring: + body["prompt_logprobs"] = scoring.top_logprobs + consume( + "scoring.input_scoring", + _ScoringVerifier("prompt_logprobs", scoring.top_logprobs), + ) + if scoring.sampled_logprobs: + body["logprobs"] = scoring.top_logprobs + consume( + "scoring.sampled_logprobs", + _ScoringVerifier("logprobs", scoring.top_logprobs), + ) + if scoring.top_logprobs > 0: + # One leaf, two breadth fields: claim the sampled one when sent. + breadth = "logprobs" if scoring.sampled_logprobs else "prompt_logprobs" + consume("scoring.top_logprobs", WireObservation((breadth,))) + + structured = _structured_outputs(req) + if structured is not None: + body["structured_outputs"] = structured + consume("structured_output.format", _StructuredOutputsVerifier(structured)) + if req.structured_output.format == "json_schema": + consume( + "structured_output.schema", + WireObservation(("structured_outputs", "json")), + ) + if "structured_output.name" in audit.active_paths: + audit.noop("structured_output.name", _SCHEMA_NAME_NOOP) + if "structured_output.strict" in audit.active_paths: + audit.noop("structured_output.strict", _STRICT_NOOP) + + for path, reason in _NOOP_PATHS.items(): + if path in audit.active_paths: + audit.noop(path, reason) + + options = req.dialect_options + if options is not None: + if not options.values: + audit.noop( + "dialect_options", "an empty options mapping has no wire semantics" + ) + for key, value in options.values.items(): + # Top-level fields, each consumed against its own body key. + body[key] = value + consume(f"dialect_options.{key}", WireObservation((key,))) + + return PreparedRequest( + operation=_OPERATION, + body=body, + context=_CompletionContext( + prompt=req.input.text, + n=params.n, + input_scoring=scoring.input_scoring, + sampled_logprobs=scoring.sampled_logprobs, + top_logprobs=scoring.top_logprobs, + structured_output=structured is not None, + ), + ) + + async def execute(self, prepared: PreparedRequest) -> Response: + """POST the prepared body to ``/completions`` and lift the reply.""" + + if prepared.operation != _OPERATION: + raise DialectError(f"unexpected vLLM operation {prepared.operation!r}") + context = prepared.context + if not isinstance(context, _CompletionContext): + raise DialectError("prepared vLLM request has invalid context") + body = prepared.thaw_body() + headers = {"Content-Type": "application/json"} + credential = self._connection.credential + if credential is not None: + headers["Authorization"] = f"Bearer {credential}" + response = await self._connection.client.post( + self._url, headers=headers, json=body + ) + raise_for_provider_status(response, "vLLM") + try: + payload = json.loads(response.text) + except ValueError as exc: + raise OutputContractError("vLLM returned invalid JSON") from exc + request_params = { + key: value for key, value in body.items() if key not in {"model", "prompt"} + } + return self._lift(payload, context, request_params) + + def _lift( + self, + raw: object, + context: _CompletionContext, + request_params: Mapping[str, JSONValue], + ) -> Response: + payload = _mapping(raw, "vLLM completions reply") + choices = _choices(payload, context.n) + usage = _usage(payload.get("usage")) + texts = tuple(cast(str, choice["text"]) for choice in choices) + + input_scoring: InputScoringResult | None = None + if context.input_scoring: + input_scoring = _input_scoring( + choices[0], + context.prompt, + None if usage is None else usage.input_tokens, + context.top_logprobs, + ) + + logprobs: tuple[TokenLogprob, ...] | None = None + top_logprobs: tuple[tuple[TopKEntry, ...], ...] | None = None + choice_logprobs: tuple[ChoiceLogprobs, ...] | None = None + if context.sampled_logprobs and context.n > 1: + choice_logprobs = tuple( + _choice_logprobs(choice, context.top_logprobs) for choice in choices + ) + elif context.sampled_logprobs: + sampled = _choice_logprobs(choices[0], context.top_logprobs) + logprobs = sampled.tokens + top_logprobs = sampled.top_logprobs + + structured_output: StructuredOutput | None = None + if context.structured_output: + try: + structured_output = StructuredOutput(json.loads(texts[0])) + except json.JSONDecodeError as exc: + raise OutputContractError( + "structured output is not valid JSON" + ) from exc + + return Response( + texts=texts, + finish_reasons=tuple( + reason if isinstance(reason := choice.get("finish_reason"), str) else "" + for choice in choices + ), + structured_output=structured_output, + logprobs=logprobs, + top_logprobs=top_logprobs, + choice_logprobs=choice_logprobs, + input_scoring=input_scoring, + usage=usage, + request_params=request_params, + response_model=_optional_string(payload, "model"), + system_fingerprint=_optional_string(payload, "system_fingerprint"), + ) + + async def arun(self, req: Request) -> Response: + """Direct compatibility path; ``Model`` uses the split contract.""" + + plan = _LegacyPlan() + validate_request_invariants(req) + validate_runtime_binding_plan(plan, req) + audit = RequestAudit(active_request_leaves(req)) + self.validate_request(req, audit, plan) + audit.raise_rejections() + prepared = self.prepare(req, audit) + audit.finish(prepared) + response = await self.execute(prepared) + self.output_contract.validate(plan, req, response) + return response diff --git a/tests/unit/cli/test_validation.py b/tests/unit/cli/test_validation.py index 2dccc280..76577c34 100644 --- a/tests/unit/cli/test_validation.py +++ b/tests/unit/cli/test_validation.py @@ -620,17 +620,19 @@ def test_retired_sglang_legacy_names_its_replacement(self): for error in result.errors ) - @pytest.mark.parametrize( - ("dialect", "message"), - [ - ("vllm_native", "no executable binder"), - ], - ) - def test_registered_but_inactive_dialect_is_rejected(self, dialect, message): - result = validate_eval_config(self._config(dialect=dialect)) + def test_registered_but_inactive_dialect_is_rejected(self, monkeypatch): + # Every stable id is active now; a binder removed stands in for one + # that is reserved. + import sieval.core.models.dialect_registry as registry + + binders = dict(registry.DIALECT_BINDERS) + del binders["vllm_native"] + monkeypatch.setattr(registry, "DIALECT_BINDERS", binders) + + result = validate_eval_config(self._config(dialect="vllm_native")) assert not result.ok - assert any(message in error for error in result.errors) + assert any("no executable binder" in error for error in result.errors) @pytest.mark.parametrize( ("capabilities", "message"), diff --git a/tests/unit/core/models/dialects/test_vllm_native.py b/tests/unit/core/models/dialects/test_vllm_native.py new file mode 100644 index 00000000..efb20ae5 --- /dev/null +++ b/tests/unit/core/models/dialects/test_vllm_native.py @@ -0,0 +1,963 @@ +"""Tests for the vLLM native completions dialect. + +Requests go through ``VllmNativeDialect.arun`` (the full audit, prepare, +execute and output-contract path) over an ``httpx.MockTransport``. The reply +shapes are vLLM 0.26.0's, recorded from a live server: a ``prompt_logprobs`` +position is keyed by token id and lists the scored token beside its top-k. + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +import json +import math +from typing import Any + +import httpx +import pytest + +from sieval.core.models.capabilities import Supported +from sieval.core.models.connection_factory import AsyncHTTPJSONConnection +from sieval.core.models.dialect import ( + Dialect, + OutputContractError, + PreparedRequest, + RequestAudit, + RequestAuditError, + active_request_leaves, +) +from sieval.core.models.dialect_registry import DIALECT_SPECS, RequestSeedSupport +from sieval.core.models.dialects.vllm_native import ( + CAPABILITY_DECISIONS, + VllmNativeDialect, + _LegacyPlan, + _ScoringVerifier, + _StructuredOutputsVerifier, + completions_url, +) +from sieval.core.models.ir import ( + ChatInput, + ChatMessage, + CompletionInput, + DialectOptions, + Request, + SamplingParams, + SchedulingParams, + ScoringParams, + StructuredOutput, + StructuredOutputParams, + TextPart, + TokenLogprob, + TopKEntry, +) + +_PROMPT = "The capital" + + +class _Server: + """Answer each POST with the next canned reply and keep the requests.""" + + def __init__(self, *replies: object, status: int = 200) -> None: + self.replies = list(replies) + self.status = status + self.requests: list[httpx.Request] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + # Python's encoder, as vLLM's: ``-Infinity`` and ``NaN`` stay literal. + return httpx.Response(self.status, text=json.dumps(self.replies.pop(0))) + + @property + def body(self) -> dict: + return json.loads(self.requests[-1].content) + + +def _dialect( + server: _Server, + *, + base_url: str = "http://host:8000/v1", + credential: str | None = "local", +) -> VllmNativeDialect: + connection = AsyncHTTPJSONConnection( + httpx.AsyncClient(base_url=base_url, transport=httpx.MockTransport(server)), + credential, + ) + return VllmNativeDialect(connection, "Qwen/Qwen2.5-0.5B") + + +def _choice(text: str = " Paris", index: int = 0, **fields: Any) -> dict: + choice = {"index": index, "text": text, "finish_reason": "length"} + choice.update(fields) + return choice + + +def _reply(*choices: dict, prompt_tokens: int = 2, completion_tokens: int = 1) -> dict: + return { + "id": "cmpl-1", + "object": "text_completion", + "model": "Qwen/Qwen2.5-0.5B", + "system_fingerprint": "vllm-0.26.0-0c752b4e", + "choices": list(choices) or [_choice()], + "usage": { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + "prompt_tokens_details": None, + }, + } + + +def _entry(logprob: float, rank: int, text: str | None) -> dict: + return {"logprob": logprob, "rank": rank, "decoded_token": text} + + +#: "The capital" scored with prompt_logprobs=2: " capital" ranks 522, so it is +#: listed before the two entries that rank 1 and 2. +_PROMPT_LOGPROBS = [ + None, + { + "6722": _entry(-9.45, 522, " capital"), + "2701": _entry(-0.32, 1, " following"), + "40389": _entry(-4.95, 2, " Purple"), + }, +] +_PROMPT_IDS = [785, 6722] + +#: The same prompt at breadth 0: the scored token alone. +_SCORED_ONLY = [None, {"6722": _entry(-9.45, 522, " capital")}] + + +#: The sampled " Paris" at rank 1 of a top-2. +_SAMPLED = { + "tokens": [" Paris"], + "token_logprobs": [-0.1], + "top_logprobs": [{" Paris": -0.1, " Lyon": -2.0}], +} + + +def _scored_choice(**fields: Any) -> dict: + return _choice( + prompt_logprobs=_PROMPT_LOGPROBS, + prompt_token_ids=_PROMPT_IDS, + token_ids=[12095], + logprobs=_SAMPLED, + **fields, + ) + + +def _top_k_scoring(**kwargs: Any) -> Request: + """Input scoring with a top-2, which the IR pairs with sampled logprobs.""" + + return _request(input_scoring=True, sampled_logprobs=True, top_logprobs=2, **kwargs) + + +def _request( + text: str = _PROMPT, + *, + sampling: SamplingParams | None = None, + input_scoring: bool = False, + sampled_logprobs: bool = False, + top_logprobs: int = 0, + structured_output: StructuredOutputParams | None = None, + options: dict | None = None, + stream: bool = False, +) -> Request: + return Request( + input=CompletionInput(text), + sampling=sampling or SamplingParams(), + scoring=ScoringParams( + input_scoring=input_scoring, + sampled_logprobs=sampled_logprobs, + top_logprobs=top_logprobs, + ), + structured_output=structured_output or StructuredOutputParams(), + scheduling=SchedulingParams(stream=stream), + dialect_options=( + DialectOptions("vllm_native", options) if options is not None else None + ), + ) + + +async def _run(reply: object, req: Request, **kwargs: Any) -> tuple[Any, _Server]: + server = _Server(reply) + response = await _dialect(server, **kwargs).arun(req) + return response, server + + +class TestDescriptor: + def test_is_an_active_canonical_dialect_with_a_request_seed(self): + spec = DIALECT_SPECS["vllm_native"] + + assert isinstance(_dialect(_Server()), Dialect) + assert spec.connection_family == "async_http_json" + assert spec.request_seed_support is RequestSeedSupport.SUPPORTED + assert spec.engine_level_seed is False + assert spec.input_kinds == ("completion",) + + def test_supports_scoring_and_structured_output_only(self): + supported = { + key + for key, decision in CAPABILITY_DECISIONS.items() + if isinstance(decision, Supported) + } + + assert supported == { + "input_scoring", + "sampled_logprobs", + "top_logprobs", + "structured_output", + } + + def test_requires_an_http_connection_and_model_id(self): + with pytest.raises(TypeError, match="AsyncHTTPJSONConnection"): + VllmNativeDialect(object(), "m") + connection = AsyncHTTPJSONConnection(httpx.AsyncClient(), None) + with pytest.raises(ValueError, match="requested_model_id"): + VllmNativeDialect(connection, "") + + def test_legacy_plan_offers_the_supported_row(self): + assert _LegacyPlan().available_capabilities == { + "input_scoring", + "sampled_logprobs", + "top_logprobs", + "structured_output", + } + + +class TestCompletionsUrl: + @pytest.mark.parametrize( + ("base", "expected"), + [ + ("http://host:8000/v1", "http://host:8000/v1/completions"), + ("http://host:8000/v1/", "http://host:8000/v1/completions"), + ("http://gw/vllm/v1?x=1", "http://gw/vllm/v1/completions"), + ], + ) + def test_completions_sits_under_the_openai_base(self, base, expected): + assert str(completions_url(httpx.URL(base))) == expected + + +class TestLowering: + @pytest.mark.anyio + async def test_posts_the_prompt_with_the_model_and_bearer_auth(self): + response, server = await _run(_reply(), _request("hi")) + + request = server.requests[0] + assert str(request.url) == "http://host:8000/v1/completions" + assert request.headers["Authorization"] == "Bearer local" + assert server.body == { + "model": "Qwen/Qwen2.5-0.5B", + "prompt": "hi", + "stream": False, + } + assert response.texts == (" Paris",) + + @pytest.mark.anyio + async def test_no_credential_sends_no_authorization(self): + _, server = await _run(_reply(), _request(), credential=None) + + assert "Authorization" not in server.requests[0].headers + + @pytest.mark.anyio + async def test_every_sampling_leaf_is_a_top_level_field(self): + sampling = SamplingParams( + temperature=0.5, + top_p=0.9, + top_k=20, + max_tokens=64, + stop=("\n",), + seed=7, + frequency_penalty=0.1, + presence_penalty=0.2, + ) + + _, server = await _run( + _reply(), + _request(sampling=sampling, options={"min_p": 0.05}), + ) + + assert server.body == { + "model": "Qwen/Qwen2.5-0.5B", + "prompt": _PROMPT, + "stream": False, + "temperature": 0.5, + "top_p": 0.9, + "top_k": 20, + "max_tokens": 64, + "stop": ["\n"], + "seed": 7, + "frequency_penalty": 0.1, + "presence_penalty": 0.2, + "min_p": 0.05, + } + + @pytest.mark.anyio + async def test_n_is_sent_only_above_one(self): + reply = _reply(_choice(index=0), _choice(index=1)) + + _, server = await _run(reply, _request(sampling=SamplingParams(n=2))) + + assert server.body["n"] == 2 + + @pytest.mark.anyio + async def test_request_params_are_the_body_without_model_and_prompt(self): + response, _ = await _run( + _reply(), _request("secret", sampling=SamplingParams(seed=1)) + ) + + assert response.request_params == {"stream": False, "seed": 1} + + @pytest.mark.anyio + async def test_stream_is_a_noop_sent_as_one_post(self): + _, server = await _run(_reply(), _request(stream=True)) + + assert len(server.requests) == 1 + assert server.body["stream"] is False + assert "stream_options" not in server.body + + @pytest.mark.anyio + async def test_input_scoring_sends_prompt_logprobs_without_echo(self): + positions = _SCORED_ONLY + choice = _choice(prompt_logprobs=positions, prompt_token_ids=_PROMPT_IDS) + + _, server = await _run(_reply(choice), _request(input_scoring=True)) + + assert server.body["prompt_logprobs"] == 0 + assert server.body["return_token_ids"] is True + assert "echo" not in server.body + assert "logprobs" not in server.body + + @pytest.mark.anyio + async def test_input_top_k_sends_both_breadths(self): + _, server = await _run(_reply(_scored_choice()), _top_k_scoring()) + + assert server.body["prompt_logprobs"] == 2 + assert server.body["logprobs"] == 2 + + @pytest.mark.anyio + async def test_sampled_logprobs_send_logprobs_with_token_ids(self): + choice = _choice( + logprobs={ + "tokens": [" Paris"], + "token_logprobs": [-0.1], + "top_logprobs": [{" Paris": -0.1, " Lyon": -2.0}], + }, + token_ids=[12095], + ) + + _, server = await _run( + _reply(choice), _request(sampled_logprobs=True, top_logprobs=2) + ) + + assert server.body["logprobs"] == 2 + assert server.body["return_token_ids"] is True + assert "prompt_logprobs" not in server.body + + @pytest.mark.anyio + async def test_json_schema_lowers_to_structured_outputs_json(self): + schema = {"type": "object", "properties": {"name": {"type": "string"}}} + params = StructuredOutputParams( + format="json_schema", schema=schema, name="person", strict=True + ) + + response, server = await _run( + _reply(_choice('{"name":"Ann"}')), _request(structured_output=params) + ) + + assert server.body["structured_outputs"] == {"json": schema} + assert "response_format" not in server.body + assert response.structured_output == StructuredOutput({"name": "Ann"}) + + @pytest.mark.anyio + async def test_json_object_lowers_to_structured_outputs_json_object(self): + _, server = await _run( + _reply(_choice("{}")), + _request(structured_output=StructuredOutputParams(format="json_object")), + ) + + assert server.body["structured_outputs"] == {"json_object": True} + + +class TestRejections: + @pytest.mark.anyio + @pytest.mark.parametrize( + ("req", "match"), + [ + (_request(sampling=SamplingParams(max_tokens=0)), "max_tokens >= 1"), + (_request(options={"prefill": "x"}), "prefill"), + (_request(options={"prefix": "x"}), "prefill"), + (_request(options={"guided_json": {}}), "unknown vLLM option"), + (_request(options={"logprobs": 5}), "scoring.sampled_logprobs"), + (_request(options={"prompt_logprobs": 0}), "scoring.input_scoring"), + (_request(options={"seed": 1}), "sampling.seed"), + (_request(options={"response_format": {}}), "structured_output"), + (_request(options={"min_p": None}), "null option"), + ( + _request( + structured_output=StructuredOutputParams( + format="json_schema", schema={}, strict=False + ) + ), + "cannot relax", + ), + ( + _request( + structured_output=StructuredOutputParams( + format="json_object", name="x" + ) + ), + "require format='json_schema'", + ), + ], + ) + async def test_refused_before_io(self, req, match): + server = _Server(_reply()) + + with pytest.raises(RequestAuditError, match=match): + await _dialect(server).arun(req) + + assert server.requests == [] + + @pytest.mark.anyio + async def test_chat_input_is_refused(self): + server = _Server(_reply()) + req = Request(input=ChatInput((ChatMessage("user", (TextPart("x"),)),))) + + with pytest.raises(RequestAuditError, match="does not support"): + await _dialect(server).arun(req) + + assert server.requests == [] + + @pytest.mark.anyio + async def test_suffix_is_refused(self): + from sieval.core.models.dialect import DialectError + + server = _Server(_reply()) + + # The plan refuses it first: fim is unsupported on this row. + with pytest.raises(DialectError, match="unavailable capability 'fim'"): + await _dialect(server).arun(Request(CompletionInput("x", suffix="y"))) + + assert server.requests == [] + + @pytest.mark.anyio + async def test_every_forwarded_option_reaches_the_body(self): + _, server = await _run( + _reply(), + _request( + options={ + "min_p": 0.1, + "repetition_penalty": 1.1, + "stop_token_ids": [13], + } + ), + ) + + assert server.body["min_p"] == 0.1 + assert server.body["repetition_penalty"] == 1.1 + assert server.body["stop_token_ids"] == [13] + + @pytest.mark.anyio + async def test_empty_options_are_a_noop(self): + _, server = await _run(_reply(), _request(options={})) + + assert set(server.body) == {"model", "prompt", "stream"} + + +class TestWireEvidence: + def test_scoring_verifier_checks_breadth_and_token_ids(self): + verifier = _ScoringVerifier("prompt_logprobs", 2) + + assert verifier.verify({"prompt_logprobs": 2, "return_token_ids": True}) is None + assert "breadth" in str( + verifier.verify({"prompt_logprobs": 3, "return_token_ids": True}) + ) + assert "breadth" in str( + verifier.verify({"prompt_logprobs": True, "return_token_ids": True}) + ) + assert "return_token_ids" in str(verifier.verify({"prompt_logprobs": 2})) + + def test_structured_outputs_verifier_compares_the_whole_field(self): + verifier = _StructuredOutputsVerifier({"json_object": True}) + + assert verifier.verify({"structured_outputs": {"json_object": True}}) is None + assert verifier.verify({"structured_outputs": {"json": {}}}) is not None + assert verifier.verify({}) is not None + + def test_finish_rejects_a_tampered_body(self): + dialect = _dialect(_Server()) + req = _request(input_scoring=True) + audit = RequestAudit(active_request_leaves(req)) + dialect.validate_request(req, audit, _LegacyPlan()) + prepared = dialect.prepare(req, audit) + tampered = PreparedRequest( + operation=prepared.operation, + body={**prepared.thaw_body(), "return_token_ids": False}, + context=prepared.context, + ) + + with pytest.raises(RequestAuditError, match="return_token_ids"): + audit.finish(tampered) + + +class TestInputScoring: + @pytest.mark.anyio + async def test_scored_token_and_rank_ordered_top_k(self): + response, _ = await _run(_reply(_scored_choice()), _top_k_scoring()) + + scoring = response.input_scoring + assert scoring.token_logprobs == ( + TokenLogprob(token="The", logprob=None, token_id=785), + TokenLogprob(token=" capital", logprob=-9.45, token_id=6722), + ) + # The scored token ranks 522, below the top 2, so it is not in them. + assert scoring.top_logprobs == ( + (), + ( + TopKEntry(token=" following", logprob=-0.32, token_id=2701), + TopKEntry(token=" Purple", logprob=-4.95, token_id=40389), + ), + ) + + @pytest.mark.anyio + async def test_a_scored_token_inside_the_top_k_is_listed_once(self): + positions = [ + None, + {"6722": _entry(-0.5, 1, " capital"), "11": _entry(-1.0, 2, ",")}, + ] + choice = _scored_choice() + choice["prompt_logprobs"] = positions + + response, _ = await _run(_reply(choice), _top_k_scoring()) + + assert [entry.token for entry in response.input_scoring.top_logprobs[1]] == [ + " capital", + ",", + ] + + @pytest.mark.anyio + async def test_a_rank_tie_beside_the_top_k_stays_out_of_it(self): + # " capital" ties the rank-2 entry but was not selected, so it is + # listed beside the top 2 with rank 2. + positions = [ + None, + { + "6722": _entry(-4.95, 2, " capital"), + "2701": _entry(-0.32, 1, " following"), + "40389": _entry(-4.95, 2, " Purple"), + }, + ] + choice = _scored_choice() + choice["prompt_logprobs"] = positions + + response, _ = await _run(_reply(choice), _top_k_scoring()) + + assert [entry.token for entry in response.input_scoring.top_logprobs[1]] == [ + " following", + " Purple", + ] + + @pytest.mark.anyio + @pytest.mark.parametrize("count", [1, 4]) + async def test_a_position_without_k_or_k_plus_one_entries_is_refused(self, count): + filler = {str(100 + i): _entry(-2.0 - i, 2 + i, f"t{i}") for i in range(3)} + position = {"6722": _entry(-0.5, 1, " capital"), **filler} + choice = _scored_choice() + choice["prompt_logprobs"] = [None, dict(list(position.items())[:count])] + + with pytest.raises(OutputContractError, match=f"lists {count} entries"): + await _run(_reply(choice), _top_k_scoring()) + + @pytest.mark.anyio + async def test_breadth_zero_carries_no_input_top_k(self): + positions = _SCORED_ONLY + choice = _choice(prompt_logprobs=positions, prompt_token_ids=_PROMPT_IDS) + + response, _ = await _run(_reply(choice), _request(input_scoring=True)) + + assert response.input_scoring.top_logprobs is None + # Generated-token scores are not requested, and none are invented. + assert response.logprobs is None + + @pytest.mark.anyio + async def test_reaches_model_output_as_echo_did(self): + """The legacy view puts input scores first, as the echo split did.""" + + from sieval.core.models._legacy_bridge import response_to_model_output + + positions = _SCORED_ONLY + choice = _choice(prompt_logprobs=positions, prompt_token_ids=_PROMPT_IDS) + response, _ = await _run(_reply(choice), _request(input_scoring=True)) + + output = response_to_model_output( + {"model": "m", "api_base": None, "default_params": {}}, response + ) + + assert output.logprobs_tokens == ["The", " capital"] + assert output.logprobs == [None, -9.45] + assert output.usage is not None + assert output.usage["input_tokens"] == 2 + + @pytest.mark.anyio + async def test_first_token_text_is_empty_when_the_rest_does_not_spell_it(self): + positions = [None, {"6722": _entry(-1.0, 1, "")}] + choice = _choice(prompt_logprobs=positions, prompt_token_ids=_PROMPT_IDS) + + response, _ = await _run(_reply(choice), _request(input_scoring=True)) + + assert response.input_scoring.token_logprobs[0].token == "" + + @pytest.mark.anyio + async def test_negative_infinity_is_kept(self): + positions = [None, {"6722": _entry(-math.inf, 9, " capital")}] + choice = _choice(prompt_logprobs=positions, prompt_token_ids=_PROMPT_IDS) + + response, _ = await _run(_reply(choice), _request(input_scoring=True)) + + assert response.input_scoring.token_logprobs[1].logprob == -math.inf + + @pytest.mark.anyio + @pytest.mark.parametrize( + ("choice", "prompt_tokens", "match"), + [ + (_choice(prompt_token_ids=_PROMPT_IDS), 2, "no prompt_logprobs"), + ( + _choice(prompt_logprobs=_PROMPT_LOGPROBS), + 2, + "no prompt_token_ids", + ), + ( + _choice(prompt_logprobs=[], prompt_token_ids=[]), + 0, + "scored no prompt tokens", + ), + # Truncated: fewer positions than the usage count. + ( + _choice(prompt_logprobs=_PROMPT_LOGPROBS, prompt_token_ids=_PROMPT_IDS), + 3, + "partial prompt", + ), + ( + _choice(prompt_logprobs=_PROMPT_LOGPROBS, prompt_token_ids=[785]), + 2, + "partial prompt", + ), + ( + _choice( + prompt_logprobs=[_PROMPT_LOGPROBS[1], _PROMPT_LOGPROBS[1]], + prompt_token_ids=_PROMPT_IDS, + ), + 2, + "first prompt position", + ), + ( + _choice(prompt_logprobs=_SCORED_ONLY, prompt_token_ids=[785, 9]), + 2, + "prompt's own token id 9", + ), + ( + _choice( + prompt_logprobs=[None, {"x": _entry(-1.0, 1, "a")}], + prompt_token_ids=_PROMPT_IDS, + ), + 2, + "key must be a token id", + ), + ( + _choice( + prompt_logprobs=[None, {"6722": _entry(-1.0, 0, " capital")}], + prompt_token_ids=_PROMPT_IDS, + ), + 2, + "rank must be a positive integer", + ), + ( + _choice( + prompt_logprobs=[None, {"6722": _entry(-1.0, 1, None)}], + prompt_token_ids=_PROMPT_IDS, + ), + 2, + "skip-tokenizer-init", + ), + ( + _choice( + prompt_logprobs=[None, {"6722": _entry(math.nan, 1, " c")}], + prompt_token_ids=_PROMPT_IDS, + ), + 2, + "NaN", + ), + ( + _choice(prompt_logprobs=_PROMPT_LOGPROBS, prompt_token_ids=[785, -1]), + 2, + "non-negative integer", + ), + ( + _choice(prompt_logprobs="bad", prompt_token_ids=_PROMPT_IDS), + 2, + "must be a list", + ), + ], + ) + async def test_incomplete_or_malformed_scores_are_refused( + self, choice, prompt_tokens, match + ): + with pytest.raises(OutputContractError, match=match): + await _run( + _reply(choice, prompt_tokens=prompt_tokens), + _request(input_scoring=True), + ) + + @pytest.mark.anyio + async def test_without_usage_the_token_ids_still_bound_the_prompt(self): + choice = _choice(prompt_logprobs=_SCORED_ONLY, prompt_token_ids=_PROMPT_IDS) + reply = _reply(choice) + reply["usage"] = None + + response, _ = await _run(reply, _request(input_scoring=True)) + + assert len(response.input_scoring.token_logprobs) == 2 + assert response.usage is None + + +class TestSampledLogprobs: + @pytest.mark.anyio + async def test_tokens_keep_ids_and_top_k_drops_the_extra_sampled_entry(self): + # " B" was sampled at rank 3, so vLLM lists it beside the top 2. + choice = _choice( + text=" B", + logprobs={ + "tokens": [" B"], + "token_logprobs": [-3.0], + "top_logprobs": [{" A": -0.1, " B": -3.0, " C": -1.0}], + }, + token_ids=[425], + ) + + response, _ = await _run( + _reply(choice), _request(sampled_logprobs=True, top_logprobs=2) + ) + + assert response.logprobs == (TokenLogprob(" B", -3.0, token_id=425),) + assert response.top_logprobs == ( + (TopKEntry(" A", -0.1), TopKEntry(" C", -1.0)), + ) + + @pytest.mark.anyio + @pytest.mark.parametrize( + ("sampled", "top"), + [ + # " B" sampled inside the top 3; the other two decode to "", so + # the map is one short. + (-0.5, {" B": -0.5, "": -0.7}), + # " B" sampled inside the top 3; another decodes alike and + # overwrites its value. + (-0.5, {" B": -0.9, " A": -0.1, " C": -1.0}), + ], + ) + async def test_a_decoded_text_collision_is_refused(self, sampled, top): + choice = _choice( + text=" B", + logprobs={ + "tokens": [" B"], + "token_logprobs": [sampled], + "top_logprobs": [top], + }, + token_ids=[425], + ) + + with pytest.raises(OutputContractError, match="decode to the same text"): + await _run(_reply(choice), _request(sampled_logprobs=True, top_logprobs=3)) + + @pytest.mark.anyio + async def test_breadth_zero_carries_no_top_k(self): + choice = _choice( + logprobs={ + "tokens": [" Paris"], + "token_logprobs": [-0.1], + "top_logprobs": [{" Paris": -0.1}], + }, + token_ids=[12095], + ) + + response, _ = await _run(_reply(choice), _request(sampled_logprobs=True)) + + assert response.top_logprobs is None + assert response.logprobs[0].token_id == 12095 + + @pytest.mark.anyio + async def test_n_samples_lift_per_choice(self): + choices = [ + _choice( + text=text, + index=index, + logprobs={ + "tokens": [text], + "token_logprobs": [-0.5], + "top_logprobs": [{text: -0.5}], + }, + token_ids=[index + 1], + ) + # Out of index order: the lift orders by index, not position. + for index, text in ((1, " b"), (0, " a")) + ] + + response, _ = await _run( + _reply(*choices), + _request( + sampling=SamplingParams(n=2), sampled_logprobs=True, top_logprobs=1 + ), + ) + + assert response.texts == (" a", " b") + assert [c.tokens[0].token_id for c in response.choice_logprobs] == [1, 2] + assert response.logprobs is None + + @pytest.mark.anyio + @pytest.mark.parametrize( + ("choice", "match"), + [ + (_choice(token_ids=[1]), "no logprobs"), + ( + _choice( + logprobs={"tokens": [" a"], "token_logprobs": [-1.0]}, + token_ids=None, + ), + "no token_ids", + ), + ( + _choice( + logprobs={"tokens": [" a", " b"], "token_logprobs": [-1.0]}, + token_ids=[1], + ), + "2 sampled tokens, 1 logprobs", + ), + ( + _choice( + logprobs={ + "tokens": [" a"], + "token_logprobs": [-1.0], + "top_logprobs": [], + }, + token_ids=[1], + ), + "top logprobs for 0 of 1", + ), + ( + _choice( + logprobs={"tokens": [" a"], "token_logprobs": [None]}, + token_ids=[1], + ), + "must be numeric", + ), + ], + ) + async def test_missing_or_misaligned_scores_are_refused(self, choice, match): + with pytest.raises(OutputContractError, match=match): + await _run(_reply(choice), _request(sampled_logprobs=True, top_logprobs=1)) + + +class TestReply: + @pytest.mark.anyio + async def test_model_fingerprint_usage_and_finish_reason(self): + response, _ = await _run(_reply(), _request()) + + assert response.response_model == "Qwen/Qwen2.5-0.5B" + assert response.system_fingerprint == "vllm-0.26.0-0c752b4e" + assert response.finish_reasons == ("length",) + assert response.usage.input_tokens == 2 + assert response.usage.total_tokens == 3 + assert response.usage.cached_tokens is None + + @pytest.mark.anyio + async def test_cached_tokens_are_read_when_reported(self): + reply = _reply() + reply["usage"]["prompt_tokens_details"] = {"cached_tokens": 1} + + response, _ = await _run(reply, _request()) + + assert response.usage.cached_tokens == 1 + + @pytest.mark.anyio + async def test_a_null_finish_reason_reads_as_empty(self): + response, _ = await _run(_reply(_choice(finish_reason=None)), _request()) + + assert response.finish_reasons == ("",) + + @pytest.mark.anyio + async def test_truncated_structured_output_is_refused(self): + params = StructuredOutputParams(format="json_object") + + with pytest.raises(OutputContractError, match="not valid JSON"): + await _run(_reply(_choice('{"name":"')), _request(structured_output=params)) + + @pytest.mark.anyio + @pytest.mark.parametrize( + ("reply", "match"), + [ + ([], "must be an object"), + ({"choices": "x"}, "must be a list"), + ({"choices": [_choice(index=1)]}, "outside the requested range"), + ({"choices": [_choice(), _choice()]}, "duplicated choice index 0"), + ({"choices": []}, r"omitted choice indexes \[0\]"), + ({"choices": [{"index": 0}]}, "has no text"), + ({"choices": [_choice()], "model": 3}, "model must be a string"), + ( + {"choices": [_choice()], "usage": {"prompt_tokens": -1}}, + "usage.prompt_tokens", + ), + ], + ) + async def test_malformed_replies_are_refused(self, reply, match): + with pytest.raises(OutputContractError, match=match): + await _run(reply, _request()) + + @pytest.mark.anyio + async def test_invalid_json_is_refused(self): + def handler(request: httpx.Request) -> httpx.Response: + del request + return httpx.Response(200, text="not json") + + connection = AsyncHTTPJSONConnection( + httpx.AsyncClient( + base_url="http://host:8000/v1", transport=httpx.MockTransport(handler) + ), + None, + ) + + with pytest.raises(OutputContractError, match="invalid JSON"): + await VllmNativeDialect(connection, "m").arun(_request()) + + @pytest.mark.anyio + async def test_http_error_carries_the_server_message(self): + server = _Server( + {"error": {"message": "max_tokens must be at least 1, got 0."}}, + status=400, + ) + + with pytest.raises(httpx.HTTPStatusError, match="vLLM: max_tokens"): + await _dialect(server).arun(_request()) + + +class TestPreparedRequestIntegrity: + @pytest.mark.anyio + async def test_execute_rejects_a_foreign_operation_or_context(self): + from sieval.core.models.dialect import DialectError + + dialect = _dialect(_Server()) + + with pytest.raises(DialectError, match="unexpected vLLM operation"): + await dialect.execute(PreparedRequest("generate", {}, None)) + with pytest.raises(DialectError, match="invalid context"): + await dialect.execute(PreparedRequest("completions", {}, None)) + + def test_prepare_refuses_chat_input(self): + from sieval.core.models.dialect import DialectError + + req = Request(input=ChatInput((ChatMessage("user", (TextPart("x"),)),))) + audit = RequestAudit(active_request_leaves(req)) + + with pytest.raises(DialectError, match="requires CompletionInput"): + _dialect(_Server()).prepare(req, audit) + + def test_capabilities_enum_view(self): + assert _dialect(_Server()).capabilities == VllmNativeDialect.CAPABILITIES diff --git a/tests/unit/core/models/test_dialect_registry.py b/tests/unit/core/models/test_dialect_registry.py index 36568a8d..16be4f61 100644 --- a/tests/unit/core/models/test_dialect_registry.py +++ b/tests/unit/core/models/test_dialect_registry.py @@ -52,6 +52,7 @@ from sieval.core.models.dialects.openai_chat import OpenAIChatDialect from sieval.core.models.dialects.openai_completions import OpenAICompletionsDialect from sieval.core.models.dialects.openai_responses import OpenAIResponsesDialect +from sieval.core.models.dialects.vllm_native import VllmNativeDialect EXPECTED_DIALECTS = ( "openai_chat", @@ -136,7 +137,7 @@ def test_request_seed_support_is_required_and_explicit(self) -> None: "anthropic_messages": RequestSeedSupport.UNSUPPORTED, "google_genai": RequestSeedSupport.SUPPORTED, "sglang_native": RequestSeedSupport.UNSUPPORTED, - "vllm_native": RequestSeedSupport.RESERVED, + "vllm_native": RequestSeedSupport.SUPPORTED, } def test_only_sglang_native_pins_its_seed_on_the_engine(self) -> None: @@ -190,6 +191,7 @@ def test_implemented_dialects_are_active_and_bindable(self) -> None: "anthropic_messages", "google_genai", "sglang_native", + "vllm_native", } assert set(DIALECT_BINDERS) == active assert all(inspect.isfunction(binder) for binder in DIALECT_BINDERS.values()) @@ -255,7 +257,7 @@ def test_bindability_and_compatibility_factory_follow_binder_registry( assert compatibility_factory_for("openai_chat") is None assert dialect_is_bindable("sglang_native") is True assert compatibility_factory_for("sglang_native") is None - assert dialect_is_bindable("vllm_native") is False + assert dialect_is_bindable("vllm_native") is True assert compatibility_factory_for("vllm_native") is None def test_binding_key_must_match_decision_key(self) -> None: @@ -325,18 +327,18 @@ def test_capability_decisions_are_available_only_for_active_dialects(self) -> No class TestDialectBinders: - @pytest.mark.parametrize( - ("dialect_id", "message"), - [ - ("vllm_native", "no executable binder"), - ], - ) def test_reserved_dialects_fail_with_named_error( - self, dialect_id: str, message: str + self, monkeypatch: pytest.MonkeyPatch ) -> None: - with pytest.raises(DialectNotImplemented, match=message): + # Every stable id is active now; a binder removed stands in for one + # that is reserved. + binders = dict(registry.DIALECT_BINDERS) + del binders["vllm_native"] + monkeypatch.setattr(registry, "DIALECT_BINDERS", binders) + + with pytest.raises(DialectNotImplemented, match="no executable binder"): bind_dialect( - dialect_id, + "vllm_native", "model", cast(Any, None), cast(ConnectionPool[Any], None), @@ -351,6 +353,7 @@ def test_reserved_dialects_fail_with_named_error( ("openai_responses", OpenAIResponsesDialect), ("anthropic_messages", AnthropicMessagesDialect), ("google_genai", GoogleGenAIDialect), + ("vllm_native", VllmNativeDialect), ], ) def test_active_binders_construct_expected_dialect( @@ -358,7 +361,7 @@ def test_active_binders_construct_expected_dialect( ) -> None: binder = DIALECT_BINDERS[dialect_id] connection: object = object() - if dialect_id in {"anthropic_messages", "google_genai"}: + if dialect_id in {"anthropic_messages", "google_genai", "vllm_native"}: connection = AsyncHTTPJSONConnection(httpx.AsyncClient(), None) dialect = binder(connection, "requested-model") @@ -374,6 +377,7 @@ def test_active_binders_construct_expected_dialect( ("openai_responses", OpenAIResponsesDialect), ("anthropic_messages", AnthropicMessagesDialect), ("google_genai", GoogleGenAIDialect), + ("vllm_native", VllmNativeDialect), ], ) def test_validated_runtime_plan_binds_active_dialect( diff --git a/tests/unit/core/models/test_reconcile.py b/tests/unit/core/models/test_reconcile.py index 13903bf3..c0e3b960 100644 --- a/tests/unit/core/models/test_reconcile.py +++ b/tests/unit/core/models/test_reconcile.py @@ -513,7 +513,7 @@ def test_hosted_tool_declaration_projects_its_minimum_domain(self) -> None: class TestBindingJoin: - @pytest.mark.parametrize("dialect_id", ["unknown", "vllm_native"]) + @pytest.mark.parametrize("dialect_id", ["unknown", "sglang_legacy"]) def test_unavailable_dialect_becomes_a_binding_diagnostic( self, dialect_id: str ) -> None: