chore(release/1.5.0): cherry-pick #13640, #14317, #14525, #12856, #13951

.github/codeowners/areas.yaml

@@ -464,6 +464,7 @@ areas:
   github_team: '@ai-dynamo/dynamo-rl-codeowners'
   path_globs:
   - components/src/dynamo/common/rl/
+  - examples/rl/
   - lib/rl/
   - tests/rl/
 - label: performance

CODEOWNERS

@@ -392,6 +392,7 @@
 
 # === rl  (@ai-dynamo/dynamo-rl-codeowners) ===
 /lib/rl/                                                                                                @ai-dynamo/dynamo-rl-codeowners
+/examples/rl/                                                                                           @ai-dynamo/dynamo-rl-codeowners
 
 # === performance  (@ai-dynamo/dynamo-performance-codeowners) ===
 /recipes/                                                                                               @ai-dynamo/dynamo-performance-codeowners

components/src/dynamo/common/backend/logprobs.py

@@ -314,52 +314,42 @@ def _parse(name: str, value: Any) -> Optional[int]:
 def extract_from_sglang_meta(
     meta_info: dict[str, Any],
     *,
-    num_output_tokens_in_chunk: Optional[int] = None,
     return_tokens_as_token_ids: bool = False,
 ) -> tuple[Optional[list[float]], Optional[list[list[dict[str, Any]]]]]:
     """Extract logprobs from SGLang's ``meta_info`` dict.
 
-    The pinned SGLang release and its N-1 predecessor align
-    ``output_token_logprobs`` and ``output_top_logprobs`` with each incremental
-    ``output_ids`` chunk. When provided, ``num_output_tokens_in_chunk`` keeps
-    malformed trailing metadata out of Dynamo's response.
+    Dynamo enables SGLang's incremental streaming output, so
+    ``output_token_logprobs`` and ``output_top_logprobs`` already align with
+    the current disjoint ``output_ids`` chunk and can be forwarded directly.
     """
     output_token_logprobs = meta_info.get("output_token_logprobs")
     if not output_token_logprobs:
         return None, None
 
-    new_logprobs = output_token_logprobs
-    if num_output_tokens_in_chunk is not None:
-        new_logprobs = new_logprobs[:num_output_tokens_in_chunk]
-    if not new_logprobs:
-        return None, None
-
-    log_probs = [float(entry[0]) for entry in new_logprobs]
+    log_probs = [float(entry[0]) for entry in output_token_logprobs]
 
     top_logprobs: Optional[list[list[dict[str, Any]]]] = None
     output_top = meta_info.get("output_top_logprobs")
     if output_top:
-        new_top = output_top[: len(new_logprobs)]
-        if new_top:
-            top_logprobs = []
-            for position_entries in new_top:
-                if position_entries is None:
-                    top_logprobs.append([])
-                    continue
-                position_list: list[dict[str, Any]] = []
-                for rank_idx, entry in enumerate(position_entries):
-                    tok_id = entry[1]
-                    token_str = (
-                        f"token_id:{tok_id}" if return_tokens_as_token_ids else entry[2]
-                    )
-                    position_list.append(
-                        {
-                            "rank": rank_idx + 1,
-                            "token_id": tok_id,
-                            "token": token_str,
-                            "logprob": float(entry[0]),
-                        }
-                    )
-                top_logprobs.append(position_list)
+        top_logprobs = []
+        for position_entries in output_top:
+            if position_entries is None:
+                top_logprobs.append([])
+                continue
+            position_list: list[dict[str, Any]] = []
+            for rank_idx, entry in enumerate(position_entries):
+                tok_id = entry[1]
+                token_str = (
+                    f"token_id:{tok_id}" if return_tokens_as_token_ids else entry[2]
+                )
+                position_list.append(
+                    {
+                        "rank": rank_idx + 1,
+                        "token_id": tok_id,
+                        "token": token_str,
+                        "logprob": float(entry[0]),
+                    }
+                )
+            top_logprobs.append(position_list)
 
     return log_probs, top_logprobs

components/src/dynamo/common/backend/tests/test_logprobs.py

@@ -405,9 +405,7 @@ def test_sglang_extract_supports_incremental_streaming_metadata():
         "output_token_logprobs": [(-0.2, 2, "b")],
         "output_top_logprobs": [[(-0.2, 2, "b"), (-1.2, 20, "B")]],
     }
-    log_probs, top_logprobs = extract_from_sglang_meta(
-        meta, num_output_tokens_in_chunk=1
-    )
+    log_probs, top_logprobs = extract_from_sglang_meta(meta)
     assert log_probs == [-0.2]
     assert top_logprobs == [
         [
@@ -456,14 +454,6 @@ def test_sglang_extract_none_top_position_becomes_empty_list():
     ]
 
 
-def test_sglang_extract_clamps_metadata_to_output_chunk():
-    meta = {
-        "output_token_logprobs": [(-0.1, 1, "a"), (-0.2, 2, "b")],
-    }
-    log_probs, _ = extract_from_sglang_meta(meta, num_output_tokens_in_chunk=1)
-    assert log_probs == [-0.1]
-
-
 # ---------------------------------------------------------------------------
 # Active handler adapters. These tests import the static methods on each
 # backend's handler class and compare them with the shared helper. They catch

components/src/dynamo/frontend/sglang_prepost.py

@@ -986,6 +986,7 @@ def __init__(
         eos_token_ids: list[int] | None = None,
         prompt_token_ids: list[int] | None = None,
         stop_strings: set[str] | None = None,
+        stop_token_ids: set[int] | None = None,
     ) -> None:
         self.tokenizer = tokenizer
         self.tool_call_parser = tool_call_parser
@@ -1007,7 +1008,11 @@ def __init__(
         self._pending_guided_reasoning_parts: list[str] | None = (
             [] if self._is_json_array_parser and reasoning_parser is not None else None
         )
+        # The frontend owns text shaping. Keep every stop token in the raw
+        # engine stream, but exclude a matched stop suffix when decoding
+        # user-visible text.
         self._eos_token_ids = set(eos_token_ids or [])
+        self._request_stop_token_ids = set(stop_token_ids or [])
         self._stop_strings = stop_strings or set()
         self._pending_stop_text = ""
         self._locally_finished = False
@@ -1036,11 +1041,35 @@ def __init__(
         # Full text accumulator for robust finish-time re-parse.
         self._tool_text_parts: list[str] = []
 
-    def _strip_trailing_eos_token_ids(self, token_ids: list[int]) -> list[int]:
-        if not self._eos_token_ids:
+    def _strip_matched_stop_token_ids(
+        self, token_ids: list[int], stop_reason: Any
+    ) -> list[int]:
+        """Remove only the engine-reported token-stop suffix from display IDs."""
+        if not token_ids:
             return token_ids
-        while token_ids and token_ids[-1] in self._eos_token_ids:
-            token_ids.pop()
+
+        known_stop_ids = self._eos_token_ids | self._request_stop_token_ids
+        if isinstance(stop_reason, int) and not isinstance(stop_reason, bool):
+            matched_ids = [stop_reason] if stop_reason in known_stop_ids else []
+        elif (
+            isinstance(stop_reason, list)
+            and stop_reason
+            and all(
+                isinstance(token_id, int)
+                and not isinstance(token_id, bool)
+                and token_id in known_stop_ids
+                for token_id in stop_reason
+            )
+        ):
+            matched_ids = stop_reason
+        elif stop_reason is None and token_ids[-1] in self._eos_token_ids:
+            # Model EOS is intentionally omitted from the public stop_reason.
+            matched_ids = [token_ids[-1]]
+        else:
+            matched_ids = []
+
+        if matched_ids and token_ids[-len(matched_ids) :] == matched_ids:
+            del token_ids[-len(matched_ids) :]
         return token_ids
 
     def _tool_call_id(self, name: str, index: int) -> str:
@@ -1364,9 +1393,12 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
         stop_reason = engine_response.get("stop_reason")
         log_probs = engine_response.get("log_probs")
         top_logprobs = engine_response.get("top_logprobs")
-        if finish_reason is not None:
+        stop_terminated = engine_response.get(
+            "stop_terminated", finish_reason == "stop"
+        )
+        if stop_terminated:
             raw_token_count = len(token_ids)
-            token_ids = self._strip_trailing_eos_token_ids(list(token_ids))
+            token_ids = self._strip_matched_stop_token_ids(list(token_ids), stop_reason)
             retained_token_count = len(token_ids)
             if log_probs is not None and len(log_probs) == raw_token_count:
                 log_probs = log_probs[:retained_token_count]

components/src/dynamo/frontend/sglang_processor.py

@@ -146,6 +146,17 @@ def _request_stop_strings(request: dict[str, Any]) -> set[str]:
     return set()
 
 
+def _request_stop_token_ids(request: dict[str, Any]) -> list[int]:
+    """Merge ``stop_token_ids`` with the legacy integer-valued ``stop`` form."""
+    values: list[Any] = list(request.get("stop_token_ids") or [])
+    stop = request.get("stop")
+    if isinstance(stop, list) and all(
+        isinstance(item, int) and not isinstance(item, bool) for item in stop
+    ):
+        values.extend(stop)
+    return _normalize_eos_token_ids(values)
+
+
 def _tokenizer_eos_token_ids(tokenizer: Any) -> list[int]:
     eos_token_ids = _normalize_eos_token_ids(getattr(tokenizer, "eos_token_ids", None))
     if eos_token_ids:
@@ -378,13 +389,12 @@ def _build_dynamo_preproc(
     max_tokens = request.get("max_completion_tokens") or request.get("max_tokens")
 
     stop = request.get("stop")
-    stop_token_ids = request.get("stop_token_ids", [])
+    stop_token_ids = _request_stop_token_ids(request)
     if isinstance(stop, str):
         stop = [stop]
     elif isinstance(stop, list) and all(
         isinstance(item, int) and not isinstance(item, bool) for item in stop
     ):
-        stop_token_ids = [*stop_token_ids, *stop]
         stop = []
     elif stop is None:
         stop = []
@@ -612,6 +622,7 @@ async def _generator_inner(
             eos_token_ids=self.eos_token_ids,
             prompt_token_ids=pre.prompt_token_ids,
             stop_strings=_request_stop_strings(request),
+            stop_token_ids=set(_request_stop_token_ids(request)),
         )
 
         async for item in self._generate_and_stream(
@@ -672,6 +683,7 @@ async def _generator_inner_pool(
             eos_token_ids=self.eos_token_ids,
             prompt_token_ids=preproc_result.prompt_token_ids,
             stop_strings=_request_stop_strings(request),
+            stop_token_ids=set(_request_stop_token_ids(request)),
         )
 
         async for item in self._generate_and_stream(
@@ -727,6 +739,7 @@ def flush_pending(
                 *,
                 finish_reason: str | None,
                 stop_reason: Any | None,
+                stop_terminated: bool,
                 engine_data: Any | None,
             ) -> dict[str, Any]:
                 nonlocal pending_token_ids
@@ -742,6 +755,7 @@ def flush_pending(
                     "token_ids": pending_token_ids,
                     "finish_reason": finish_reason,
                     "stop_reason": stop_reason,
+                    "stop_terminated": stop_terminated,
                 }
                 if pending_log_probs is not None:
                     mapped_response["log_probs"] = pending_log_probs
@@ -864,6 +878,7 @@ def flush_pending(
                         envelope = flush_pending(
                             finish_reason=None,
                             stop_reason=None,
+                            stop_terminated=False,
                             engine_data=None,
                         )
                         yield envelope
@@ -872,8 +887,10 @@ def flush_pending(
 
                 chunk_tokens = len(new_ids)
                 cumulative_output_tokens += chunk_tokens
-                finish_reason = _map_finish_reason(engine_response.get("finish_reason"))
+                raw_finish_reason = engine_response.get("finish_reason")
+                finish_reason = _map_finish_reason(raw_finish_reason)
                 stop_reason = engine_response.get("stop_reason")
+                stop_terminated = raw_finish_reason in {"eos", "stop"}
 
                 if usage := engine_response.get("completion_usage"):
                     pending_usage = usage
@@ -898,6 +915,7 @@ def flush_pending(
                     envelope = flush_pending(
                         finish_reason=finish_reason,
                         stop_reason=stop_reason,
+                        stop_terminated=stop_terminated,
                         engine_data=engine_data,
                     )
                     yield envelope