diff --git a/dingo/frontend/README.md b/dingo/frontend/README.md
index b31e31da684a..15883997c8d3 100644
--- a/dingo/frontend/README.md
+++ b/dingo/frontend/README.md
@@ -6,3 +6,36 @@
The API gateway for serving LLM inference requests with OpenAI-compatible HTTP and KServe gRPC endpoints.
See [docs/components/frontend/](../../../../docs/components/frontend/) for documentation.
+
+## SGLang chat generation limits and stops
+
+The SGLang chat processor defaults to 8192 generated tokens when both
+`max_completion_tokens` and `max_tokens` are omitted. An explicit
+`max_completion_tokens` takes precedence over `max_tokens`. The limit includes
+reasoning and answer tokens; reaching it returns `finish_reason: "length"` and
+can leave the answer empty if all tokens were spent reasoning.
+
+For the exact model ID `glm-5.3` (case-insensitive, also repository-qualified
+IDs such as `zai-org/GLM-5.3`), negative `frequency_penalty` values are accepted
+but projected to `0.0` for sampling. This compatibility policy targets normal
+completion, as observed with negative values on the official GLM-5.3 API; it
+does not assert how that API implements the parameter. Nonnegative values and
+other models retain their existing behavior. Arbitrary serving aliases do not
+activate this policy. Reasoning remains enabled, and the generation limit
+remains a fallback rather than a guarantee that every prompt will finish.
+
+With an active reasoning parser, string `stop` sequences apply after reasoning
+separation, to the non-reasoning output (including tool syntax). A match stops
+worker generation and returns `finish_reason: "stop"` (or `"tool_calls"` when
+applicable). Stops split across chunks are supported; `include_stop_str_in_output`
+controls whether the matched string is retained. EOS and `stop_token_ids` remain
+worker-side conditions. Without a reasoning parser, including requests that set
+`separate_reasoning: false`, string stops retain their worker-side behavior.
+
+When no reasoning parser is explicitly configured on the frontend or in worker
+metadata, the frontend selects `glm45` for GLM-5.3 / GLM-5.3-Flash if their chat
+template opens a thinking block. Qwen3 continues to require explicit `qwen3`
+reasoning parser configuration.
+Detection uses the discovered model's `config.json`, so serving aliases do not
+affect it. Explicit parser settings take precedence. Unknown model types with
+thinking templates produce a warning requesting explicit parser configuration.
diff --git a/dingo/frontend/sglang_prepost.py b/dingo/frontend/sglang_prepost.py
index fa651df726e8..33f0db5b4699 100644
--- a/dingo/frontend/sglang_prepost.py
+++ b/dingo/frontend/sglang_prepost.py
@@ -1415,6 +1415,8 @@ def __init__(
reasoning_parser_name: str | None = None,
eos_token_ids: list[int] | None = None,
stop_strings: set[str] | None = None,
+ include_stop_str_in_output: bool = False,
+ min_tokens: int = 0,
guided_decoding: dict[str, Any] | None = None,
response_format_guided_active: bool = False,
tool_guided_active: bool = False,
@@ -1458,6 +1460,10 @@ def __init__(
self._eos_token_ids = set(eos_token_ids or [])
self._stop_strings = {stop for stop in (stop_strings or set()) if stop}
self._pending_stop_text = ""
+ self._include_stop_str_in_output = include_stop_str_in_output
+ self._min_tokens = min_tokens
+ self._generated_token_count = 0
+ self.matched_stop: str | None = None
self._all_token_ids: list[int] = []
# Logprob records built for chunks that emit no visible delta (e.g.
@@ -1663,6 +1669,32 @@ def _filter_stop_string_delta(self, text: str, finish_reason: str | None) -> str
return text[:-pending_len]
return text
+ def _filter_answer_stop(self, text: str, finish_reason: str | None) -> str:
+ """Match strings after reasoning separation, including split chunks."""
+ text = self._pending_stop_text + text
+ self._pending_stop_text = ""
+ if self._generated_token_count < self._min_tokens:
+ return text
+ matches = [
+ (pos, -len(stop), stop)
+ for stop in self._stop_strings
+ if (pos := text.find(stop)) >= 0
+ ]
+ if matches:
+ pos, _, self.matched_stop = min(matches)
+ end = (
+ pos + len(self.matched_stop)
+ if self._include_stop_str_in_output
+ else pos
+ )
+ return text[:end]
+ if not finish_reason:
+ pending_len = _trailing_stop_prefix_len(text, self._stop_strings)
+ if pending_len:
+ self._pending_stop_text = text[-pending_len:]
+ return text[:-pending_len]
+ return text
+
def _parse_reasoning_delta(
self,
delta_text: str,
@@ -1840,8 +1872,11 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
Returns:
OpenAI choice dict or ``None`` if nothing to emit yet.
"""
+ if self.matched_stop is not None:
+ return None
raw_ids = engine_response.get("token_ids")
token_ids = raw_ids if isinstance(raw_ids, list) else list(raw_ids or [])
+ self._generated_token_count += len(token_ids)
finish_reason = engine_response.get("finish_reason")
finished = finish_reason is not None
if finished:
@@ -1871,7 +1906,8 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
if token_ids or finished
else ""
)
- delta_text = self._filter_stop_string_delta(delta_text, finish_reason)
+ if self.reasoning_parser is None:
+ delta_text = self._filter_stop_string_delta(delta_text, finish_reason)
if self._is_kimi_k3 and delta_text:
self._kimi_k3_raw_text_parts.append(delta_text)
@@ -1903,6 +1939,10 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
len(token_ids),
)
self._reasoning_token_count += reasoning_tokens
+ if self.reasoning_parser is not None:
+ normal_text = self._filter_answer_stop(normal_text, finish_reason)
+ if self.matched_stop is not None:
+ finish_reason = "stop"
if self._is_kimi_k3:
reasoning_text = _strip_kimi_k3_control_markers(reasoning_text) or None
@@ -2155,6 +2195,7 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
if (
finish_reason
and self._is_kimi_k3
+ and self.matched_stop is None
and not self._saw_normal_output
and not self._tool_call_names
):
diff --git a/dingo/frontend/sglang_processor.py b/dingo/frontend/sglang_processor.py
index faf3e1811253..32c980909070 100644
--- a/dingo/frontend/sglang_processor.py
+++ b/dingo/frontend/sglang_processor.py
@@ -51,6 +51,8 @@
logger = logging.getLogger(__name__)
DEFAULT_SGLANG_STREAM_INTERVAL: int = 1
+# Bound reasoning and answer tokens together when the client omits both limits.
+DEFAULT_SGLANG_MAX_TOKENS: int = 8192
# Internal carrier key mirroring the Rust constant in
# lib/llm/src/protocols/openai/chat_completions.rs (INTERNAL_SGLEXT_KEY).
@@ -121,8 +123,10 @@ def _normalize_eos_token_ids(value: Any) -> list[int]:
return []
-def _request_stop_strings(request: dict[str, Any]) -> set[str]:
- if request.get("include_stop_str_in_output") is True:
+def _request_stop_strings(
+ request: dict[str, Any], *, match_visible: bool = False
+) -> set[str]:
+ if request.get("include_stop_str_in_output") is True and not match_visible:
return set()
stop = request.get("stop")
if isinstance(stop, str):
@@ -242,6 +246,29 @@ def _runtime_config_parser_name(
return value if isinstance(value, str) and value else None
+def _infer_reasoning_parser_name(source_path: str, chat_template: str | None) -> str | None:
+ """Infer known think-tag parsers from the discovered model, not its alias."""
+ if not detect_force_reasoning_from_template(chat_template):
+ return None
+ try:
+ with open(os.path.join(source_path, "config.json"), encoding="utf-8") as f:
+ model_config = json.load(f)
+ except (OSError, ValueError):
+ logger.warning("Cannot infer reasoning parser from %s/config.json", source_path)
+ return None
+ parser = {
+ "glm_moe_dsa": "glm45",
+ "glm5_next": "glm45",
+ }.get(model_config.get("model_type"))
+ if parser is None:
+ logger.warning(
+ "Thinking template detected for model_type=%r without a reasoning parser; "
+ "configure --dyn-reasoning-parser to separate reasoning and answer stops",
+ model_config.get("model_type"),
+ )
+ return parser
+
+
def _unsupported_n_message(n: int) -> str:
return f"Unsupported value: 'n={n}'. " "This endpoint currently supports only n=1."
@@ -376,7 +403,24 @@ def _build_dynamo_preproc(
require_reasoning: bool = False,
) -> dict[str, Any]:
"""Build the Dynamo preprocessed request dict from request fields."""
- max_tokens = request.get("max_completion_tokens") or request.get("max_tokens")
+ max_tokens = request.get("max_completion_tokens")
+ if max_tokens is None:
+ max_tokens = request.get("max_tokens")
+ if max_tokens is None:
+ max_tokens = DEFAULT_SGLANG_MAX_TOKENS
+
+ frequency_penalty = request.get("frequency_penalty", 0.0)
+ # GLM-5.3 compatibility policy: accept negative values without rewarding
+ # repetition. Official API probes complete normally with -2; this is our
+ # compatibility choice, not a claim about the provider's implementation.
+ # Match the exact model ID (also accepting a repository-qualified ID),
+ # rather than applying the policy to every model using the GLM parser.
+ if (
+ model_name.rsplit("/", 1)[-1].lower() == "glm-5.3"
+ and isinstance(frequency_penalty, (int, float))
+ and frequency_penalty < 0
+ ):
+ frequency_penalty = 0.0
stop = request.get("stop")
stop_token_ids = request.get("stop_token_ids", [])
@@ -390,6 +434,12 @@ def _build_dynamo_preproc(
elif stop is None:
stop = []
+ # The worker sees the unsplit reasoning/answer stream. String stops belong
+ # to the frontend when a reasoning parser is active; token stops and EOS
+ # remain backend-owned.
+ if reasoning_parser is not None:
+ stop = []
+
# Handle logprobs
logprobs_val = None
logprobs = request.get("logprobs")
@@ -415,7 +465,7 @@ def _build_dynamo_preproc(
"sampling_options": {
"n": request.get("n", 1),
"presence_penalty": request.get("presence_penalty", 0.0),
- "frequency_penalty": request.get("frequency_penalty", 0.0),
+ "frequency_penalty": frequency_penalty,
"repetition_penalty": request.get("repetition_penalty", 1.0),
"temperature": request.get("temperature", 1.0),
"top_p": request.get("top_p", 1.0),
@@ -596,7 +646,11 @@ async def _generator_inner(
tool_call_parser_name=self.tool_call_parser_name,
reasoning_parser_name=self.reasoning_parser_name,
eos_token_ids=self.eos_token_ids,
- stop_strings=_request_stop_strings(request),
+ stop_strings=_request_stop_strings(
+ request, match_visible=pre.reasoning_parser is not None
+ ),
+ include_stop_str_in_output=request.get("include_stop_str_in_output", False),
+ min_tokens=request.get("min_tokens", 0),
guided_decoding=pre.guided_decoding,
response_format_guided_active=pre.response_format_guided_active,
tool_guided_active=pre.tool_guided_active,
@@ -670,7 +724,11 @@ async def _generator_inner_pool(
tool_call_parser_name=self.tool_call_parser_name,
reasoning_parser_name=self.reasoning_parser_name,
eos_token_ids=self.eos_token_ids,
- stop_strings=_request_stop_strings(request),
+ stop_strings=_request_stop_strings(
+ request, match_visible=reasoning_parser is not None
+ ),
+ include_stop_str_in_output=request.get("include_stop_str_in_output", False),
+ min_tokens=request.get("min_tokens", 0),
guided_decoding=preproc_result.dynamo_preproc.get(
"sampling_options", {}
).get("guided_decoding"),
@@ -716,6 +774,8 @@ async def _generate_and_stream(
created_ts = int(time.time())
stream_interval = self.stream_interval
continuous_usage = _continuous_usage_requested(request)
+ dynamo_stream = None
+ terminal_items: list[dict[str, Any]] = []
try:
dynamo_stream = await self.routed_engine.generate(
@@ -799,6 +859,19 @@ async def _generate_and_stream(
t_pp0 = time.monotonic()
choice = post.process_output(mapped_response)
+ local_stop = getattr(post, "matched_stop", None)
+ if local_stop is not None:
+ finish_reason = "stop"
+ stop_reason = local_stop
+ # Workers normally attach usage to their terminal
+ # chunk. A frontend stop happens before that chunk.
+ pending_usage = dict(pending_usage or {})
+ prompt_tokens = pending_usage.get("prompt_tokens", input_tokens)
+ pending_usage.update(
+ prompt_tokens=prompt_tokens,
+ completion_tokens=cumulative_output_tokens,
+ total_tokens=prompt_tokens + cumulative_output_tokens,
+ )
if self.debug_perf:
t_pp1 = time.monotonic()
@@ -897,16 +970,35 @@ async def _generate_and_stream(
dynamo_out["choices"] = [
{**choice, "delta": {}, "logprobs": None}
]
- yield {**envelope, "data": payload}
- yield {"_dynamo_annotated": True, "data": dynamo_out}
+ outgoing = [
+ {**envelope, "data": payload},
+ {"_dynamo_annotated": True, "data": dynamo_out},
+ ]
else:
- yield envelope
+ outgoing = [envelope]
+
+ if local_stop is not None:
+ terminal_items = outgoing
+ break
+ for item in outgoing:
+ yield item
pending_token_ids = []
pending_log_probs = None
pending_top_logprobs = None
pending_usage = None
first_chunk = False
+
+ # Break releases the async-for iterator reference. Dropping the
+ # native stream closes its receiver; RoutedEngine signals only
+ # the child generation context, leaving HTTP completion intact.
+ if dynamo_stream is not None:
+ close = getattr(dynamo_stream, "aclose", None)
+ if close is not None:
+ await close()
+ dynamo_stream = None
+ for item in terminal_items:
+ yield item
except Unknown:
raise
except Exception as e:
@@ -915,6 +1007,11 @@ async def _generate_and_stream(
f"Error generating response for request {request_id}: {e}"
) from e
finally:
+ if dynamo_stream is not None:
+ close = getattr(dynamo_stream, "aclose", None)
+ if close is not None:
+ await close()
+ dynamo_stream = None
if self.debug_perf and token_count > 0:
logger.info(
"[perf] sglang stream done: request=%s tokens=%d "
@@ -999,6 +1096,9 @@ async def chat_engine_factory(
reasoning_parser_name = (
self.reasoning_parser_name
or _runtime_config_parser_name(mdc, "reasoning_parser")
+ or _infer_reasoning_parser_name(
+ local_dir, getattr(tokenizer, "chat_template", None)
+ )
)
if tool_call_parser_name:
diff --git a/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py b/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py
new file mode 100644
index 000000000000..212012f92e15
--- /dev/null
+++ b/dingo/frontend/tests/test_sglang_reasoning_stops_unit.py
@@ -0,0 +1,364 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""CPU-only regression tests using SGLang's actual GLM reasoning parser."""
+
+import asyncio
+import json
+from types import SimpleNamespace
+
+import pytest
+from _routed_engine_fakes import FakeRoutedEngine, FakeRoutedItem
+from sglang.srt.parser.reasoning_parser import ReasoningParser
+
+import dingo.frontend.sglang_processor as processor_module
+from dingo.frontend.sglang_prepost import SglangStreamingPostProcessor
+from dingo.frontend.sglang_processor import (
+ DEFAULT_SGLANG_MAX_TOKENS,
+ SglangProcessor,
+ _build_dynamo_preproc,
+ _infer_reasoning_parser_name,
+ _request_stop_strings,
+)
+
+pytestmark = [pytest.mark.unit, pytest.mark.sglang, pytest.mark.gpu_0]
+
+
+@pytest.mark.parametrize(
+ "model_type,expected",
+ [
+ ("glm_moe_dsa", "glm45"),
+ ("glm5_next", "glm45"),
+ ("qwen3", None),
+ ("qwen3_moe", None),
+ ("qwen3_next", None),
+ ("unknown", None),
+ ],
+)
+def test_infer_reasoning_parser_requires_known_model_and_thinking_template(
+ tmp_path, model_type, expected
+):
+ (tmp_path / "config.json").write_text(json.dumps({"model_type": model_type}))
+ assert (
+ _infer_reasoning_parser_name(str(tmp_path), "<|assistant|>") == expected
+ )
+ assert _infer_reasoning_parser_name(str(tmp_path), "plain assistant") is None
+
+
+def test_infer_reasoning_parser_missing_config(tmp_path):
+ assert _infer_reasoning_parser_name(str(tmp_path), "<|assistant|>") is None
+
+
+@pytest.mark.parametrize(
+ "model_type,expected,source",
+ [
+ ("glm5_next", "glm45", "inferred"),
+ ("glm_moe_dsa", "glm45", "inferred"),
+ ("glm5_next", "glm45", "frontend"),
+ ("glm5_next", "glm45", "worker"),
+ ("qwen3", "qwen3", "frontend"),
+ ("qwen3", "qwen3", "worker"),
+ ],
+)
+def test_factory_selects_parser_and_preserves_reasoning_before_stop(
+ tmp_path, monkeypatch, model_type, expected, source
+):
+ # Exercise discovery -> factory -> parser -> preprocessing/streaming.
+ # No model weights or tokenizer downloads are needed.
+ (tmp_path / "config.json").write_text(json.dumps({"model_type": model_type}))
+ tokenizer = TextTokenizer()
+ tokenizer.chat_template = "<|assistant|>"
+ monkeypatch.setattr(processor_module, "_load_tokenizer", lambda *_: tokenizer)
+ monkeypatch.setattr(
+ processor_module, "_resolve_model_eos_token_ids", lambda **_: []
+ )
+ monkeypatch.setattr(
+ processor_module, "PythonAsyncEngine", lambda generator, _: generator
+ )
+ configured = "qwen3"
+ selected = expected if source == "inferred" else configured
+ factory = processor_module.SglangEngineFactory(
+ SimpleNamespace(
+ trust_remote_code=False,
+ preprocess_workers=0,
+ exclude_tools_when_tool_choice_none=True,
+ ),
+ reasoning_parser_name=configured if source == "frontend" else None,
+ )
+ mdc = SimpleNamespace(
+ model_type=lambda: SimpleNamespace(supports_chat=lambda: True),
+ local_dir=lambda: str(tmp_path),
+ name=lambda: "custom-serving-alias",
+ runtime_config=lambda: (
+ {"reasoning_parser": configured} if source == "worker" else {}
+ ),
+ )
+ engine = FakeRoutedEngine(
+ [
+ {"token_ids": list(map(ord, "thinking\nmore thinking\n"))},
+ {"token_ids": list(map(ord, "ALPHA\nBETA"))},
+ ]
+ )
+
+ async def run():
+ generator = await factory.chat_engine_factory(None, mdc, engine)
+ processor = generator.__self__
+ assert processor.reasoning_parser_name == selected
+ request = {"model": "custom-serving-alias", "stop": ["\n"]}
+ parser = ReasoningParser(
+ model_type=processor.reasoning_parser_name,
+ stream_reasoning=True,
+ force_reasoning=True,
+ )
+ preproc = _build_dynamo_preproc(
+ request, [1], request["model"], None, reasoning_parser=parser
+ )
+ assert preproc["stop_conditions"]["stop"] == []
+ post = SglangStreamingPostProcessor(
+ tokenizer=tokenizer,
+ tool_call_parser=None,
+ reasoning_parser=parser,
+ stop_strings={"\n"},
+ )
+ return [
+ x
+ async for x in processor._generate_and_stream(
+ "test", request, preproc, [1], post
+ )
+ ]
+
+ items = asyncio.run(run())
+ choices = [c for item in items for c in item.get("data", {}).get("choices", [])]
+ assert (
+ "".join(c["delta"].get("reasoning_content", "") for c in choices)
+ == "thinking\nmore thinking\n"
+ )
+ assert "".join(c["delta"].get("content", "") for c in choices) == "ALPHA"
+ assert choices[-1]["finish_reason"] == "stop"
+
+
+class TextTokenizer:
+ def decode(self, token_ids, *, skip_special_tokens):
+ return "".join(chr(token_id) for token_id in token_ids)
+
+
+def make_post(stops, **kwargs):
+ return SglangStreamingPostProcessor(
+ tokenizer=TextTokenizer(),
+ tool_call_parser=None,
+ reasoning_parser=ReasoningParser(
+ model_type="glm45", stream_reasoning=True, force_reasoning=True
+ ),
+ stop_strings=set(stops),
+ **kwargs,
+ )
+
+
+def feed(post, text, finish=None):
+ return post.process_output(
+ {"token_ids": list(map(ord, text)), "finish_reason": finish}
+ )
+
+
+@pytest.mark.parametrize(
+ "limits,expected",
+ [
+ ({}, DEFAULT_SGLANG_MAX_TOKENS),
+ ({"max_tokens": 256}, 256),
+ ({"max_tokens": 256, "max_completion_tokens": 512}, 512),
+ ({"max_tokens": 256, "max_completion_tokens": 0}, 0),
+ (
+ {"max_tokens": None, "max_completion_tokens": None},
+ DEFAULT_SGLANG_MAX_TOKENS,
+ ),
+ ],
+)
+def test_glm_negative_penalty_is_neutral_with_finite_limit(limits, expected):
+ result = _build_dynamo_preproc(
+ {"frequency_penalty": -2, **limits}, [1], "glm-5.3", None
+ )
+ assert result["sampling_options"]["frequency_penalty"] == 0.0
+ assert result["stop_conditions"]["max_tokens"] == expected
+
+
+@pytest.mark.parametrize("model", ["glm-5.3", "GLM-5.3", "zai-org/GLM-5.3"])
+@pytest.mark.parametrize("penalty", [-2, -0.5, 0, 0.5, 2])
+def test_glm_frequency_penalty_compatibility(model, penalty):
+ request = {"model": model, "frequency_penalty": penalty}
+ result = _build_dynamo_preproc(request, [1], model, None, require_reasoning=True)
+ assert result["sampling_options"]["frequency_penalty"] == max(0, penalty)
+ assert result["model"] == model
+ assert result["require_reasoning"] is True
+ assert request == {"model": model, "frequency_penalty": penalty}
+
+
+@pytest.mark.parametrize(
+ "model", ["test-model", "glm-5.2", "glm-5.3-flash", "my-glm-5.3"]
+)
+def test_other_models_keep_negative_frequency_penalty(model):
+ result = _build_dynamo_preproc(
+ {"frequency_penalty": -2}, [1], model, None, reasoning_parser=object()
+ )
+ assert result["sampling_options"]["frequency_penalty"] == -2
+
+
+def test_message_field_does_not_become_sampling_option():
+ result = _build_dynamo_preproc(
+ {"messages": [{"role": "user", "content": "hi", "frequency_penalty": -2}]},
+ [1],
+ "glm-5.3",
+ None,
+ )
+ assert result["sampling_options"]["frequency_penalty"] == 0.0
+
+
+def test_only_string_stops_move_to_reasoning_frontend():
+ request = {"stop": ["\n"], "stop_token_ids": [2]}
+ result = _build_dynamo_preproc(
+ request, [1], "glm-5.3", [3], reasoning_parser=object()
+ )
+ assert result["stop_conditions"]["stop"] == []
+ assert result["stop_conditions"]["stop_token_ids"] == [2]
+ assert result["eos_token_ids"] == [3]
+ assert request["stop"] == ["\n"]
+ plain = _build_dynamo_preproc(request, [1], "glm-5.3", [3])
+ assert plain["stop_conditions"]["stop"] == ["\n"]
+
+
+@pytest.mark.parametrize("include", [False, True])
+def test_reasoning_newline_does_not_stop_answer_newline_does(include):
+ request = {"stop": "\n", "include_stop_str_in_output": include}
+ post = make_post(
+ _request_stop_strings(request, match_visible=True),
+ include_stop_str_in_output=include,
+ )
+ thought = feed(post, "thinking\nmore thinking\n")
+ assert thought["delta"]["reasoning_content"] == "thinking\nmore thinking\n"
+ assert thought["finish_reason"] is None
+ answer = feed(post, "ALPHA\nBETA")
+ assert answer["delta"]["content"] == ("ALPHA\n" if include else "ALPHA")
+ assert answer["finish_reason"] == "stop"
+ assert post.matched_stop == "\n"
+ assert feed(post, "must not appear") is None
+
+
+@pytest.mark.parametrize("split", range(1, 5))
+def test_stop_across_chunks(split):
+ post = make_post([""])
+ feed(post, "thinking\n")
+ first = feed(post, "hello"[: 5 + split])
+ second = feed(post, ""[split:] + "ignored")
+ assert first["delta"]["content"] == "hello"
+ assert first["finish_reason"] is None
+ assert second["finish_reason"] == "stop"
+ assert not second["delta"].get("content")
+
+
+def test_partial_stop_flushed_on_length_and_reasoning_only_length():
+ post = make_post([""])
+ feed(post, "thinking")
+ feed(post, "helloanswerSTOPdiscardEND")
+ assert end["delta"]["content"] == "answer"
+ assert post.matched_stop == "STOP"
+ post = make_post(["\n"], min_tokens=150)
+ assert feed(post, "thinkinga\n")["finish_reason"] is None
+ end = feed(post, "x" * 160 + "\nignored")
+ assert end["finish_reason"] == "stop"
+
+
+@pytest.mark.parametrize("interval", [1, 20])
+@pytest.mark.parametrize("continuous", [False, True])
+@pytest.mark.parametrize("native_stream", [False, True])
+def test_local_stop_closes_generation_and_emits_terminal_usage(
+ interval, continuous, native_stream
+):
+ class Engine:
+ closed = False
+ consumed = 0
+
+ async def generate(self, request, **kwargs):
+ texts = ["thinking\n", "ALPHA\nBETAignored", "never read"]
+
+ # Native AsyncResponseStream has no aclose method: dropping its
+ # last reference closes the receiver and stops child generation.
+ engine = self
+
+ class NativeStream:
+ def __aiter__(self):
+ return self
+
+ async def __anext__(self):
+ if engine.consumed == len(texts):
+ raise StopAsyncIteration
+ text = texts[engine.consumed]
+ engine.consumed += 1
+ return FakeRoutedItem({"token_ids": list(map(ord, text))})
+
+ def __del__(self):
+ engine.closed = True
+
+ if native_stream:
+ return NativeStream()
+
+ async def stream():
+ try:
+ for text in texts:
+ self.consumed += 1
+ yield FakeRoutedItem({"token_ids": list(map(ord, text))})
+ finally:
+ self.closed = True
+
+ return stream()
+
+ engine = Engine()
+ processor = SglangProcessor(
+ tokenizer=TextTokenizer(),
+ routed_engine=engine,
+ tool_call_parser_name=None,
+ reasoning_parser_name="glm45",
+ eos_token_ids=None,
+ stream_interval=interval,
+ )
+
+ async def collect():
+ items = []
+ async for item in processor._generate_and_stream(
+ "test",
+ {
+ "model": "glm-5.3",
+ "nvext": {"extra_fields": ["stop_reason"]},
+ "stream_options": {"continuous_usage_stats": continuous},
+ },
+ {},
+ [1, 2],
+ make_post(["\n"]),
+ ):
+ items.append(item)
+ choices = item.get("data", {}).get("choices", [])
+ if choices and choices[0]["finish_reason"]:
+ assert engine.closed
+ return items
+
+ items = asyncio.run(collect())
+ assert engine.consumed == 2
+ terminal = items[-1]["data"]
+ assert terminal["choices"][0]["finish_reason"] == "stop"
+ assert terminal["choices"][0]["delta"]["content"] == "ALPHA"
+ generated = len("thinking\nALPHA\nBETAignored")
+ assert terminal["usage"]["completion_tokens"] == generated
+ assert terminal["usage"]["total_tokens"] == generated + 2
+ assert terminal["nvext"]["stop_reason"] == "\n"
+ assert json.loads(items[-1]["comment"][0])["output_tokens"] == generated