From 3bc140d29214c8410267771f24b988da59864ebf Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 16:56:16 +0200 Subject: [PATCH 1/5] fix(server): reject failed SGLang generation terminals --- .../sie_server/adapters/sglang/generation.py | 65 ++++++----- .../tests/adapters/test_sglang_generation.py | 101 ++++++++++++++++-- 2 files changed, 129 insertions(+), 37 deletions(-) diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 6dce05dbb..7c308be59 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -42,6 +42,8 @@ FinishReason, GenerationAdapter, GenerationChunk, + GenerationError, + GenerationInvalidRequestError, GenerationResult, ) from sie_server.adapters._spec import AdapterSpec @@ -156,17 +158,27 @@ def _encode_image_data(images: list[ImageInput] | None) -> list[str] | None: return encoded -def _raise_for_sglang_event_error(event: Any) -> None: - """Propagate an error carried inside SGLang's HTTP-200 SSE stream.""" - if not isinstance(event, dict) or "error" not in event: - return - error = event["error"] - message = error.get("message") if isinstance(error, dict) else error - if not isinstance(message, str) or not message.strip(): - message = "unknown in-band SGLang error" - message = message.strip()[:500] - logger.error("SGLang /generate in-band error: %s", message) - raise RuntimeError(f"SGLang /generate error: {message}") +def _raise_for_sglang_event_error(event: Any, *, grammar: GrammarSpec | None = None, terminal: bool = False) -> None: + """Reject upstream failures before consuming their text or usage.""" + if not isinstance(event, dict): + raise GenerationError("SGLang /generate returned an invalid event") + if "error" in event: + raise GenerationError("SGLang /generate returned an in-band error") + meta = event.get("meta_info") + finish = meta.get("finish_reason") if isinstance(meta, dict) else None + kind = finish.get("type") if isinstance(finish, dict) else finish + if kind == "abort": + if isinstance(finish, dict) and finish.get("status_code") == 400: + if grammar is not None: + raise GenerationInvalidRequestError( + "grammar", f"The generation backend rejected the requested {grammar.kind} grammar" + ) + raise GenerationInvalidRequestError("prompt", "The generation backend rejected the request") + raise GenerationError("SGLang /generate aborted the request") + if kind is not None and kind not in ("stop", "length"): + raise GenerationError("SGLang /generate returned an unsupported finish reason") + if (terminal or event.get("finished")) and kind is None: + raise GenerationError("SGLang /generate completed without a finish reason") def _tail_file(path: str, *, max_lines: int = 200) -> str: @@ -1257,7 +1269,7 @@ async def generate( event = json.loads(line) except json.JSONDecodeError: continue - _raise_for_sglang_event_error(event) + _raise_for_sglang_event_error(event, grammar=grammar) idx = int(event.get("index", 0)) cumulative = event.get("text", "") if not isinstance(cumulative, str): @@ -1378,6 +1390,8 @@ async def generate( # tolerate a single dict defensively. if isinstance(results, dict): results = [results] + for result in results: + _raise_for_sglang_event_error(result, grammar=grammar, terminal=True) if rank: # Highest cumulative token-logprob first; keep the top return_count. results = sorted(results, key=_cumulative_logprob, reverse=True)[:return_count] @@ -1438,7 +1452,7 @@ async def generate( candidates.append( { "text": r.get("text", "") if isinstance(r, dict) else "", - "finish_reason": fr_type if isinstance(fr_type, str) else "stop", + "finish_reason": fr_type, "logprobs": cand_logprobs, } ) @@ -1506,13 +1520,8 @@ async def generate( try: async with client.stream("POST", f"{self._server_url}/generate", json=body) as response: if response.status_code != 200: - # Drain a bit of the body for diagnostics, then raise. - body_preview = await response.aread() - logger.error( - "SGLang /generate stream error %d: %s", - response.status_code, - body_preview[:500], - ) + await response.aread() + logger.error("SGLang /generate stream error %d", response.status_code) response.raise_for_status() last_cumulative_text = "" @@ -1550,10 +1559,10 @@ async def generate( try: event = json.loads(line) except json.JSONDecodeError: - logger.warning("SGLang stream: skipping non-JSON line: %s", line[:200]) + logger.warning("SGLang stream: skipping non-JSON line") continue - _raise_for_sglang_event_error(event) + _raise_for_sglang_event_error(event, grammar=grammar) chunk = _chunk_from_sglang_event( event, previous_cumulative_text=last_cumulative_text, @@ -1815,6 +1824,7 @@ def _chunk_from_sglang_event( OpenAI ``ChatCompletionTokenLogprob`` shape for the tokens introduced by *this* event. """ + _raise_for_sglang_event_error(event) if not isinstance(event, dict): return None cumulative = event.get("text", "") @@ -1897,11 +1907,7 @@ def _chunk_from_sglang_event( chunk_logprobs = tuple(built) if is_terminal: - finish_reason: FinishReason - if raw_finish in ("stop", "length", "cancelled", "error"): - finish_reason = raw_finish # type: ignore[assignment] - else: - finish_reason = "stop" + finish_reason = cast("FinishReason", raw_finish) prompt_tokens = meta.get("prompt_tokens") if isinstance(meta, dict) else None completion_tokens = meta.get("completion_tokens") if isinstance(meta, dict) else None return GenerationChunk( @@ -1976,6 +1982,7 @@ def _parse_sglang_generate_response(result: Any) -> GenerationResult: msg = f"SGLang /generate returned unexpected shape: {type(result).__name__}" raise RuntimeError(msg) + _raise_for_sglang_event_error(result, terminal=True) text = result.get("text", "") if not isinstance(text, str): msg = "SGLang /generate response missing 'text'" @@ -1988,11 +1995,11 @@ def _parse_sglang_generate_response(result: Any) -> GenerationResult: raw_finish = meta.get("finish_reason") if isinstance(raw_finish, dict): raw_finish = raw_finish.get("type") - finish_reason = raw_finish if raw_finish in ("stop", "length") else "stop" + finish_reason = cast("FinishReason", raw_finish) return GenerationResult( text=text, - finish_reason=finish_reason, # type: ignore[arg-type] + finish_reason=finish_reason, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, ) diff --git a/packages/sie_server/tests/adapters/test_sglang_generation.py b/packages/sie_server/tests/adapters/test_sglang_generation.py index 681dcc966..d7ba777c5 100644 --- a/packages/sie_server/tests/adapters/test_sglang_generation.py +++ b/packages/sie_server/tests/adapters/test_sglang_generation.py @@ -22,7 +22,13 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from sie_server.adapters._generation_base import GenerationChunk, collect_generation, suppress_thinking_blocks +from sie_server.adapters._generation_base import ( + GenerationChunk, + GenerationError, + GenerationInvalidRequestError, + collect_generation, + suppress_thinking_blocks, +) from sie_server.adapters.sglang import _server from sie_server.adapters.sglang.cuda13 import SGLangStrictThinkingAdapter from sie_server.adapters.sglang.generation import ( @@ -412,7 +418,7 @@ async def _collect() -> None: with pytest.raises( RuntimeError, - match="SGLang /generate error: vision processor rejected image", + match="SGLang /generate returned an in-band error", ): asyncio.run(_collect()) @@ -1583,12 +1589,9 @@ def test_parse_response_with_list_shape() -> None: assert result.completion_tokens == 1 -def test_parse_response_missing_meta_defaults_to_stop() -> None: - result = _parse_sglang_generate_response({"text": "xyz"}) - assert result.text == "xyz" - assert result.finish_reason == "stop" - assert result.prompt_tokens == 0 - assert result.completion_tokens == 0 +def test_parse_response_missing_meta_is_an_error() -> None: + with pytest.raises(GenerationError, match="without a finish reason"): + _parse_sglang_generate_response({"text": "xyz"}) def test_chunk_translator_surfaces_logprobs_3tuple() -> None: @@ -2336,3 +2339,85 @@ def process_mm_data(self, input_text, images=None, videos=None, audios=None, **k assert completed.returncode == 0, completed.stderr assert "Error in sitecustomize" not in completed.stderr assert completed.stdout.strip() == "mm-process-config-ready" + + +@pytest.mark.parametrize( + ("n", "stream", "best_of"), [(1, True, None), (2, True, None), (2, False, None), (1, False, 2)] +) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_backend_grammar_abort_never_emits_placeholder_or_usage( + mock_async_client: MagicMock, adapter, n: int, stream: bool, best_of: int | None +) -> None: + schema = { + "type": "object", + "properties": {"value": {"type": ["string", "null"]}}, + "required": ["value"], + "additionalProperties": False, + } + abort = { + "text": "[]", + "meta_info": { + "prompt_tokens": 1, + "completion_tokens": 1, + "finish_reason": { + "type": "abort", + "status_code": 400, + "message": "Invalid grammar request: private-schema-content", + }, + }, + } + client = _make_client_with_stream(_FakeStreamingResponse(["data: " + json.dumps(abort)])) + response = MagicMock() + response.json.return_value = [ + {"text": "valid", "meta_info": {"finish_reason": {"type": "stop"}}}, + abort, + ] + client.post = AsyncMock(return_value=response) + mock_async_client.return_value = client + adapter._server_url = "http://localhost:30005" + chunks = [] + + async def consume() -> None: + async for chunk in adapter.generate( + prompt="Extract the optional value.", + max_new_tokens=64, + n=n, + stream=stream, + best_of=best_of, + grammar=GrammarSpec(kind="json_schema", value=schema), + ): + chunks.append(chunk) + + with pytest.raises(GenerationInvalidRequestError, match="rejected the requested json_schema grammar") as exc: + asyncio.run(consume()) + assert exc.value.code == "invalid_request" + assert exc.value.param == "grammar" + assert "private-schema-content" not in str(exc.value) + assert chunks == [] + + +@pytest.mark.parametrize( + "finish", ["abort", "error", "cancelled", {"type": "abort", "status_code": 500}, {"type": "unknown"}] +) +def test_chunk_translator_rejects_failed_or_unknown_terminal(finish: Any) -> None: + with pytest.raises(GenerationError): + _chunk_from_sglang_event( + {"text": "[]", "meta_info": {"finish_reason": finish, "prompt_tokens": 1, "completion_tokens": 1}}, + previous_cumulative_text="", + first_yield_done=False, + ) + + +@pytest.mark.parametrize("meta", [None, {}, []]) +def test_chunk_translator_rejects_terminal_without_reason(meta: Any) -> None: + with pytest.raises(GenerationError, match="without a finish reason"): + _chunk_from_sglang_event( + {"text": "[]", "finished": True, "meta_info": meta}, + previous_cumulative_text="", + first_yield_done=False, + ) + + +def test_legacy_parser_rejects_backend_abort() -> None: + with pytest.raises(GenerationError, match="aborted"): + _parse_sglang_generate_response({"text": "[]", "meta_info": {"finish_reason": {"type": "abort"}}}) From 2be0155f8ef1ff0eda201739e87fca1a0ed5d6e4 Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 17:01:07 +0200 Subject: [PATCH 2/5] fix(server): reject malformed backend finish metadata --- .../src/sie_server/adapters/sglang/generation.py | 2 ++ .../tests/adapters/test_sglang_generation.py | 10 ++++++++++ 2 files changed, 12 insertions(+) diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 7c308be59..523bfd088 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -167,6 +167,8 @@ def _raise_for_sglang_event_error(event: Any, *, grammar: GrammarSpec | None = N meta = event.get("meta_info") finish = meta.get("finish_reason") if isinstance(meta, dict) else None kind = finish.get("type") if isinstance(finish, dict) else finish + if finish is not None and kind is None: + raise GenerationError("SGLang /generate returned a malformed finish reason") if kind == "abort": if isinstance(finish, dict) and finish.get("status_code") == 400: if grammar is not None: diff --git a/packages/sie_server/tests/adapters/test_sglang_generation.py b/packages/sie_server/tests/adapters/test_sglang_generation.py index d7ba777c5..9f57f008c 100644 --- a/packages/sie_server/tests/adapters/test_sglang_generation.py +++ b/packages/sie_server/tests/adapters/test_sglang_generation.py @@ -2421,3 +2421,13 @@ def test_chunk_translator_rejects_terminal_without_reason(meta: Any) -> None: def test_legacy_parser_rejects_backend_abort() -> None: with pytest.raises(GenerationError, match="aborted"): _parse_sglang_generate_response({"text": "[]", "meta_info": {"finish_reason": {"type": "abort"}}}) + + +@pytest.mark.parametrize("finish", [{}, {"type": None}]) +def test_chunk_translator_rejects_malformed_nonnull_finish_metadata(finish: Any) -> None: + with pytest.raises(GenerationError, match="malformed finish reason"): + _chunk_from_sglang_event( + {"text": "[]", "meta_info": {"finish_reason": finish, "prompt_tokens": 1, "completion_tokens": 1}}, + previous_cumulative_text="", + first_yield_done=False, + ) From 56efe5fbe2abf18f30f3d19d9feb14cbe04b6ee4 Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 17:07:46 +0200 Subject: [PATCH 3/5] fix(server): narrow validated terminal result type --- .../sie_server/src/sie_server/adapters/sglang/generation.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 523bfd088..78148f5ec 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -34,7 +34,7 @@ import uuid from collections.abc import AsyncIterator from pathlib import Path -from typing import Any, cast +from typing import Any, Literal, cast import httpx @@ -1909,7 +1909,7 @@ def _chunk_from_sglang_event( chunk_logprobs = tuple(built) if is_terminal: - finish_reason = cast("FinishReason", raw_finish) + finish_reason = cast("Literal['stop', 'length']", raw_finish) prompt_tokens = meta.get("prompt_tokens") if isinstance(meta, dict) else None completion_tokens = meta.get("completion_tokens") if isinstance(meta, dict) else None return GenerationChunk( @@ -1997,7 +1997,7 @@ def _parse_sglang_generate_response(result: Any) -> GenerationResult: raw_finish = meta.get("finish_reason") if isinstance(raw_finish, dict): raw_finish = raw_finish.get("type") - finish_reason = cast("FinishReason", raw_finish) + finish_reason = cast("Literal['stop', 'length']", raw_finish) return GenerationResult( text=text, From afaab574e6d035f399fce98029281e9c2f2ef507 Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 17:10:23 +0200 Subject: [PATCH 4/5] fix(server): require complete generation candidates --- .../sie_server/adapters/sglang/generation.py | 15 ++++- .../tests/adapters/test_sglang_generation.py | 66 +++++++++++++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 78148f5ec..544a38bab 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -1247,6 +1247,7 @@ async def generate( sbody["top_logprobs_num"] = top_logprobs sclient = await self._get_or_create_http_client() last_text: dict[int, str] = {} + completed_candidates: set[int] = set() # Per-candidate logprob watermark: SGLang's # ``meta_info.output_token_logprobs`` is a per-candidate cumulative # list growing across events for that index. Slicing @@ -1272,7 +1273,11 @@ async def generate( except json.JSONDecodeError: continue _raise_for_sglang_event_error(event, grammar=grammar) - idx = int(event.get("index", 0)) + idx = event.get("index", 0) + if not isinstance(idx, int) or isinstance(idx, bool) or not 0 <= idx < return_count: + raise GenerationError("SGLang /generate returned an invalid candidate index") + if idx in completed_candidates: + raise GenerationError("SGLang /generate returned an event after the candidate terminal") cumulative = event.get("text", "") if not isinstance(cumulative, str): cumulative = last_text.get(idx, "") @@ -1284,6 +1289,8 @@ async def generate( fr = meta.get("finish_reason") fr_type = fr.get("type") if isinstance(fr, dict) else fr candidate_done = fr_type is not None + if candidate_done: + completed_candidates.add(idx) if candidate_done and isinstance(meta.get("completion_tokens"), int): total_completion += meta["completion_tokens"] # Per-candidate logprob slice — same shape conversion as the @@ -1346,6 +1353,8 @@ async def generate( choice_index=idx, logprobs=chunk_logprobs, ) + if len(completed_candidates) != return_count: + raise GenerationError("SGLang /generate ended before all candidates completed") # Single global terminal closes the multi-candidate stream (carries # aggregate usage). Each candidate already received its own # ``finish_reason`` on the per-choice completion chunk above; this @@ -1392,8 +1401,12 @@ async def generate( # tolerate a single dict defensively. if isinstance(results, dict): results = [results] + if not isinstance(results, list): + raise GenerationError("SGLang /generate returned an invalid candidate list") for result in results: _raise_for_sglang_event_error(result, grammar=grammar, terminal=True) + if len(results) != gen_count: + raise GenerationError("SGLang /generate returned an incorrect candidate count") if rank: # Highest cumulative token-logprob first; keep the top return_count. results = sorted(results, key=_cumulative_logprob, reverse=True)[:return_count] diff --git a/packages/sie_server/tests/adapters/test_sglang_generation.py b/packages/sie_server/tests/adapters/test_sglang_generation.py index 9f57f008c..1699bd0d1 100644 --- a/packages/sie_server/tests/adapters/test_sglang_generation.py +++ b/packages/sie_server/tests/adapters/test_sglang_generation.py @@ -1885,6 +1885,17 @@ def test_generate_n_gt_one_non_streaming_omits_logprobs_when_not_requested( }, }, ] + sglang_results.append( + { + "text": "b", + "meta_info": { + "finish_reason": {"type": "stop"}, + "completion_tokens": 1, + "prompt_tokens": 3, + "output_token_logprobs": [[-1.0, 2, "b"]], + }, + } + ) resp = MagicMock() resp.json = MagicMock(return_value=sglang_results) resp.raise_for_status = MagicMock() @@ -2431,3 +2442,58 @@ def test_chunk_translator_rejects_malformed_nonnull_finish_metadata(finish: Any) previous_cumulative_text="", first_yield_done=False, ) + + +@pytest.mark.parametrize("indexes", [[], [0], [0, 0], [0, 2], [0, True], [0, "1"]]) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_streaming_candidates_require_exact_distinct_terminals(mock_async_client: MagicMock, adapter, indexes) -> None: + events = [ + { + "index": index, + "text": "value", + "meta_info": { + "finish_reason": {"type": "stop"}, + "prompt_tokens": 4, + "completion_tokens": 1, + }, + } + for index in indexes + ] + mock_async_client.return_value = _make_client_with_stream( + _FakeStreamingResponse([*("data: " + json.dumps(event) for event in events), "data: [DONE]"]) + ) + adapter._server_url = "http://localhost:30005" + chunks = [] + + async def consume() -> None: + async for chunk in adapter.generate(prompt="Two values", max_new_tokens=8, n=2, stream=True): + chunks.append(chunk) + + with pytest.raises(GenerationError): + asyncio.run(consume()) + assert not any(chunk.done for chunk in chunks) + assert all(chunk.prompt_tokens is None and chunk.completion_tokens is None for chunk in chunks) + + +@pytest.mark.parametrize(("count", "n", "best_of"), [(0, 2, None), (1, 2, None), (3, 2, None), (2, 1, 3)]) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_buffered_candidates_require_exact_count_before_ranking( + mock_async_client: MagicMock, adapter, count: int, n: int, best_of: int | None +) -> None: + response = MagicMock() + response.json.return_value = [ + {"text": "value", "meta_info": {"finish_reason": {"type": "stop"}}} for _ in range(count) + ] + client = _make_client_with_stream(_FakeStreamingResponse([])) + client.post = AsyncMock(return_value=response) + mock_async_client.return_value = client + adapter._server_url = "http://localhost:30005" + chunks = [] + + async def consume() -> None: + async for chunk in adapter.generate(prompt="Two values", max_new_tokens=8, n=n, best_of=best_of): + chunks.append(chunk) + + with pytest.raises(GenerationError, match="incorrect candidate count"): + asyncio.run(consume()) + assert chunks == [] From c0aca4438f1fe1359a8a0ef704ef592e49dd193f Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 17:16:45 +0200 Subject: [PATCH 5/5] fix(generation): require explicit streamed candidate indexes --- .../sie_server/src/sie_server/adapters/sglang/generation.py | 2 +- packages/sie_server/tests/adapters/test_sglang_generation.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 544a38bab..426797c9c 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -1273,7 +1273,7 @@ async def generate( except json.JSONDecodeError: continue _raise_for_sglang_event_error(event, grammar=grammar) - idx = event.get("index", 0) + idx = event.get("index") if not isinstance(idx, int) or isinstance(idx, bool) or not 0 <= idx < return_count: raise GenerationError("SGLang /generate returned an invalid candidate index") if idx in completed_candidates: diff --git a/packages/sie_server/tests/adapters/test_sglang_generation.py b/packages/sie_server/tests/adapters/test_sglang_generation.py index 1699bd0d1..559c98964 100644 --- a/packages/sie_server/tests/adapters/test_sglang_generation.py +++ b/packages/sie_server/tests/adapters/test_sglang_generation.py @@ -2444,12 +2444,12 @@ def test_chunk_translator_rejects_malformed_nonnull_finish_metadata(finish: Any) ) -@pytest.mark.parametrize("indexes", [[], [0], [0, 0], [0, 2], [0, True], [0, "1"]]) +@pytest.mark.parametrize("indexes", [[], [0], [0, 0], [0, 2], [0, True], [0, "1"], [1, None]]) @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") def test_streaming_candidates_require_exact_distinct_terminals(mock_async_client: MagicMock, adapter, indexes) -> None: events = [ { - "index": index, + **({"index": index} if index is not None else {}), "text": "value", "meta_info": { "finish_reason": {"type": "stop"},