From 4f5d4c781319eb47a4528a58efbe632a1288ff28 Mon Sep 17 00:00:00 2001
From: guojn1 <gjn16637211661@163.com>
Date: Thu, 24 Sep 2026 12:19:26 +0800
Subject: [PATCH 1/6] [fix][sglang] Enable SGLang top logprobs by default

---
 dingo/common/backend/logprobs.py              | 28 +++++++++++--------
 dingo/common/backend/tests/test_logprobs.py   | 10 +++++--
 dingo/sglang/CLAUDE.md                        |  9 +++---
 .../tests/test_sglang_decode_handler.py       | 19 ++++++-------
 docs/tool-calling/troubleshooting.md          |  9 +++---
 5 files changed, 40 insertions(+), 35 deletions(-)

diff --git a/dingo/common/backend/logprobs.py b/dingo/common/backend/logprobs.py
index ee7dc5926682..cf4fbbbeb663 100644
--- a/dingo/common/backend/logprobs.py
+++ b/dingo/common/backend/logprobs.py
@@ -227,22 +227,28 @@ def extract_prompt_logprobs_from_sglang_meta(
 
 
 _SGLANG_TOP_LOGPROBS_UNSUPPORTED_MSG = (
-    "Dynamo's SGLang backend does not currently support logprobs >= 1 due to "
-    "an O(N) per-position detokenization in the upstream sglang tokenizer "
-    "manager. Use logprobs=0 for chosen-token logprobs, or set "
-    "DYN_SGL_ALLOW_TOP_LOGPROBS=1 to override at your own risk. "
-    "Track the upstream fix at https://github.com/sgl-project/sglang/pull/24447."
+    "SGLang top-k logprobs are disabled by DYN_SGL_ALLOW_TOP_LOGPROBS=0. "
+    "Set DYN_SGL_ALLOW_TOP_LOGPROBS=1 to enable them. SGLang versions without "
+    "batched top-token detokenization may incur extra latency for long outputs. "
+    "See the upstream optimization proposal at "
+    "https://github.com/sgl-project/sglang/pull/24447."
 )
 
 DYN_SGL_ALLOW_TOP_LOGPROBS_ENV = "DYN_SGL_ALLOW_TOP_LOGPROBS"
 
 
 def sglang_top_logprobs_allowed() -> bool:
-    """Read the ``DYN_SGL_ALLOW_TOP_LOGPROBS`` env-var gate."""
-    return os.environ.get(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "").lower() not in (
-        "",
+    """Return whether SGLang top-k logprobs are enabled.
+
+    They are enabled by default so valid OpenAI ``logprobs`` requests work.
+    Set ``DYN_SGL_ALLOW_TOP_LOGPROBS=0`` to restore the opt-out guard on
+    deployments where SGLang's per-position detokenization cost is a concern.
+    """
+    return os.environ.get(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "1").lower() not in (
         "0",
         "false",
+        "no",
+        "off",
     )
 
 
@@ -255,10 +261,8 @@ def build_sglang_logprob_kwargs(
     ``return_logprob`` / ``top_logprobs_num`` / ``logprob_start_len`` kwargs.
 
     Raises ``ValueError`` for ``logprobs >= 1`` when
-    ``allow_top_logprobs`` is ``False``. SGLang's tokenizer manager
-    detokenizes top-k tokens serially (O(N) per generated token), so
-    enabling it without a batched detokenize path degrades latency
-    badly.
+    ``allow_top_logprobs`` is ``False``. This is an opt-out guard for
+    deployments concerned about per-position top-token detokenization cost.
     """
     if not output_options:
         return {}
diff --git a/dingo/common/backend/tests/test_logprobs.py b/dingo/common/backend/tests/test_logprobs.py
index 736d7ac46034..46a41c5cee91 100644
--- a/dingo/common/backend/tests/test_logprobs.py
+++ b/dingo/common/backend/tests/test_logprobs.py
@@ -301,13 +301,13 @@ def test_sglang_kwargs_empty_when_no_options():
 
 
 def test_sglang_kwargs_logprobs_zero_allowed_without_gate():
-    # The default gate forbids logprobs >= 1; logprobs=0 always works.
+    # Chosen-token-only logprobs are available even when top-k is disabled.
     kwargs = build_sglang_logprob_kwargs({"logprobs": 0}, allow_top_logprobs=False)
     assert kwargs == {"return_logprob": True, "top_logprobs_num": 0}
 
 
 def test_sglang_kwargs_top_logprobs_rejected_without_gate():
-    with pytest.raises(ValueError, match="does not currently support logprobs >= 1"):
+    with pytest.raises(ValueError, match="disabled by DYN_SGL_ALLOW_TOP_LOGPROBS=0"):
         build_sglang_logprob_kwargs({"logprobs": 2}, allow_top_logprobs=False)
 
 
@@ -336,12 +336,16 @@ def test_sglang_kwargs_both_set_picks_max():
 
 
 def test_sglang_gate_reads_env(monkeypatch):
+    monkeypatch.delenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, raising=False)
+    assert sglang_top_logprobs_allowed() is True
     monkeypatch.setenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "1")
     assert sglang_top_logprobs_allowed() is True
     monkeypatch.setenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "0")
     assert sglang_top_logprobs_allowed() is False
-    monkeypatch.delenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, raising=False)
+    monkeypatch.setenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, "false")
     assert sglang_top_logprobs_allowed() is False
+    monkeypatch.delenv(DYN_SGL_ALLOW_TOP_LOGPROBS_ENV, raising=False)
+    assert sglang_top_logprobs_allowed() is True
 
 
 # ---------------------------------------------------------------------------
diff --git a/dingo/sglang/CLAUDE.md b/dingo/sglang/CLAUDE.md
index 3bb06b9dcb23..ac4cf46cbf09 100644
--- a/dingo/sglang/CLAUDE.md
+++ b/dingo/sglang/CLAUDE.md
@@ -249,11 +249,10 @@ absolute sequence position where logprob computation starts: `-1` (default) = ou
 only (`len(prompt) - 1`), `0` = from prompt start. We set it to 0 when `prompt_logprobs`
 is requested.
 
-**Top-logprobs gate**: `logprobs >= 1` (or `prompt_logprobs >= 1`) raises `ValueError`
-by default. SGLang's tokenizer manager detokenizes top-k tokens per-position serially,
-causing severe latency degradation (O(N) per generated token). Callers must use
-`logprobs=0` for chosen-token-only logprobs. Set `DYN_SGL_ALLOW_TOP_LOGPROBS=1` to
-override once upstream batches `detokenize_top_logprobs_tokens`.
+**Top-logprobs**: `logprobs >= 1` (or `prompt_logprobs >= 1`) is enabled by default.
+SGLang versions without batched top-token detokenization may incur extra latency
+for long outputs. Set `DYN_SGL_ALLOW_TOP_LOGPROBS=0` to disable top-k logprobs
+on deployments where that cost is a concern.
 
 **Streaming behavior** (`_extract_logprobs`):
 
diff --git a/dingo/sglang/tests/test_sglang_decode_handler.py b/dingo/sglang/tests/test_sglang_decode_handler.py
index 41c6528867b4..4b321fcd8478 100644
--- a/dingo/sglang/tests/test_sglang_decode_handler.py
+++ b/dingo/sglang/tests/test_sglang_decode_handler.py
@@ -436,21 +436,20 @@ def test_build_logprob_kwargs_allows_chosen_token_logprobs(monkeypatch):
     assert kwargs == {"return_logprob": True, "top_logprobs_num": 0}
 
 
-def test_build_logprob_kwargs_rejects_top_logprobs_by_default(monkeypatch):
+def test_build_logprob_kwargs_allows_top_logprobs_by_default(monkeypatch):
     monkeypatch.delenv("DYN_SGL_ALLOW_TOP_LOGPROBS", raising=False)
 
-    with pytest.raises(ValueError, match="does not currently support logprobs >= 1"):
-        DecodeWorkerHandler._build_logprob_kwargs({"output_options": {"logprobs": 1}})
-
-
-def test_build_logprob_kwargs_allows_top_logprobs_with_escape_hatch(monkeypatch):
-    monkeypatch.setenv("DYN_SGL_ALLOW_TOP_LOGPROBS", "1")
-
     kwargs = DecodeWorkerHandler._build_logprob_kwargs(
-        {"output_options": {"logprobs": 2}}
+        {"output_options": {"logprobs": 1}}
     )
+    assert kwargs == {"return_logprob": True, "top_logprobs_num": 1}
 
-    assert kwargs == {"return_logprob": True, "top_logprobs_num": 2}
+
+def test_build_logprob_kwargs_rejects_top_logprobs_when_disabled(monkeypatch):
+    monkeypatch.setenv("DYN_SGL_ALLOW_TOP_LOGPROBS", "0")
+
+    with pytest.raises(ValueError, match="disabled by DYN_SGL_ALLOW_TOP_LOGPROBS=0"):
+        DecodeWorkerHandler._build_logprob_kwargs({"output_options": {"logprobs": 1}})
 
 
 def test_extract_logprobs_formats_top_tokens_as_token_ids():
diff --git a/docs/tool-calling/troubleshooting.md b/docs/tool-calling/troubleshooting.md
index ffc3e22d8c18..ffc444ca25b1 100644
--- a/docs/tool-calling/troubleshooting.md
+++ b/docs/tool-calling/troubleshooting.md
@@ -25,11 +25,10 @@ diagnostic info.
 > the assistant-content channel; the reasoning channel is not surfaced in
 > `logprobs.content`.
 >
-> If the worker is the SGLang backend, `logprobs: true` is rejected by
-> default because SGLang's tokenizer manager detokenizes top-k tokens
-> serially, causing latency degradation. Launch the worker with
-> `DYN_SGL_ALLOW_TOP_LOGPROBS=1` set in the environment to opt in for the
-> duration of the repro request, then unset it afterward. Tracked at
+> SGLang top-k logprobs are enabled by default. SGLang versions without
+> batched top-token detokenization may incur extra latency for long outputs.
+> Set `DYN_SGL_ALLOW_TOP_LOGPROBS=0` on the worker to disable them if needed.
+> The upstream optimization proposal is described at
 > [sgl-project/sglang#24447](https://github.com/sgl-project/sglang/pull/24447).
 
 ## The request

From 51a8e416c44f1a9b3ddff75f8a20a9c065e0bea2 Mon Sep 17 00:00:00 2001
From: guojn1 <gjn16637211661@163.com>
Date: Fri, 25 Sep 2026 22:55:58 +0800
Subject: [PATCH 2/6] fix(frontend): hoist tool parameter  to guided schema
 root for  resolution

---
 dingo/common/backend/tests/test_logprobs.py   |   4 +-
 dingo/frontend/sglang_prepost.py              | 215 +++++++++-
 .../tests/test_sglang_processor_unit.py       | 368 ++++++++++++++++++
 3 files changed, 585 insertions(+), 2 deletions(-)

diff --git a/dingo/common/backend/tests/test_logprobs.py b/dingo/common/backend/tests/test_logprobs.py
index 46a41c5cee91..d07a57d9077c 100644
--- a/dingo/common/backend/tests/test_logprobs.py
+++ b/dingo/common/backend/tests/test_logprobs.py
@@ -660,7 +660,9 @@ def test_parity_sglang_kwargs_rejects_top_logprobs_consistently():
         {"prompt_logprobs": 2},
         {"logprobs": 0, "prompt_logprobs": 3},
     ):
-        with pytest.raises(ValueError, match="does not currently support"):
+        with pytest.raises(
+            ValueError, match="SGLang top-k logprobs are disabled"
+        ):
             build_sglang_logprob_kwargs(opts, allow_top_logprobs=False)
 
 
diff --git a/dingo/frontend/sglang_prepost.py b/dingo/frontend/sglang_prepost.py
index c253b7cfcf5a..a054000d2d68 100644
--- a/dingo/frontend/sglang_prepost.py
+++ b/dingo/frontend/sglang_prepost.py
@@ -735,6 +735,212 @@ def _call_with_optional_parallel_tool_calls(
     return func(*args)
 
 
+def _escape_pointer_token(token: str) -> str:
+    return token.replace("~", "~0").replace("/", "~1")
+
+
+def _unescape_pointer_token(token: str) -> str:
+    return token.replace("~1", "/").replace("~0", "~")
+
+
+# Keywords whose values are literal instance data, not subschemas (JSON
+# Schema spec: const/enum hold instance values, default/examples are
+# annotations carrying instance values).  Reference rewriting must skip
+# them; see _rewrite_def_refs.
+_LITERAL_VALUE_KEYWORDS = frozenset({"const", "enum", "default", "examples"})
+
+
+def _rewrite_def_refs(value: Any, names: dict[str, str]) -> None:
+    """Rewrite ``#/$defs/...`` references in-place according to ``names``.
+
+    Only the leading definition token is remapped; any JSON-pointer suffix
+    below the definition is preserved.
+    """
+    if isinstance(value, dict):
+        ref = value.get("$ref")
+        if isinstance(ref, str) and ref.startswith("#/$defs/"):
+            pointer = ref[len("#/$defs/") :]
+            old_token, separator, suffix = pointer.partition("/")
+            old_name = _unescape_pointer_token(old_token)
+            new_name = names.get(old_name)
+            if new_name is not None and new_name != old_name:
+                value["$ref"] = f"#/$defs/{_escape_pointer_token(new_name)}" + (
+                    f"/{suffix}" if separator else ""
+                )
+        for key, child in value.items():
+            if key in _LITERAL_VALUE_KEYWORDS:
+                # Literal instance data (const/enum/default/examples), not
+                # a subschema: a {"$ref": ...} object here is data the
+                # model must produce and must never be rewritten (review
+                # 20260924 v2).
+                continue
+            _rewrite_def_refs(child, names)
+    elif isinstance(value, list):
+        for child in value:
+            _rewrite_def_refs(child, names)
+
+
+def _namespace_tool_parameter_defs(tools: list[Any]) -> list[Any]:
+    """Isolate each tool's parameter ``$defs`` into a unique namespace.
+
+    SGLang's constraint builders collect all tools' ``$defs`` up front and
+    raise ``ValueError`` when two tools define the same name with different
+    schemas — before any post-builder processing can run.  When two or more
+    tools carry ``$defs``, deep-copy the tools and rename every definition
+    to ``{tool_name}__{def_name}``, rewriting that tool's internal
+    ``#/$defs/...`` references to match.  Generated names are reserved
+    globally up front so they cannot collide with each other (e.g. tool
+    ``a`` defining ``b__X`` vs tool ``a__b`` defining ``X``).  Requests
+    where at most one tool carries ``$defs`` are returned unchanged: no
+    cross-tool conflict is possible and original definition names (and
+    their established behaviour) are preserved.
+    """
+
+    def _params_of(tool: Any) -> Any:
+        fn = getattr(tool, "function", None)
+        return getattr(fn, "parameters", None) if fn is not None else None
+
+    tools_with_defs = [
+        t
+        for t in tools
+        if isinstance(_params_of(t), dict)
+        and isinstance(_params_of(t).get("$defs"), dict)
+    ]
+    if len(tools_with_defs) < 2:
+        return tools
+
+    new_tools = copy.deepcopy(tools)
+
+    # Pass 1: reserve every generated name up front.
+    taken: set[str] = set()
+    plans: list[tuple[dict[str, Any], dict[str, str]]] = []
+    for tool in new_tools:
+        params = _params_of(tool)
+        if not (isinstance(params, dict) and isinstance(params.get("$defs"), dict)):
+            continue
+        tool_name = getattr(tool.function, "name", "") or ""
+        names: dict[str, str] = {}
+        for old_name in params["$defs"]:
+            base = f"{tool_name}__{old_name}"
+            new_name = base
+            suffix = 2
+            while new_name in taken:
+                new_name = f"{base}_{suffix}"
+                suffix += 1
+            taken.add(new_name)
+            names[old_name] = new_name
+        plans.append((params, names))
+
+    # Pass 2: rewrite references and rename the definition keys.
+    for params, names in plans:
+        nested_defs = params.pop("$defs")
+        _rewrite_def_refs(params, names)
+        for definition in nested_defs.values():
+            _rewrite_def_refs(definition, names)
+        params["$defs"] = {names[old]: d for old, d in nested_defs.items()}
+    return new_tools
+
+
+def _hoist_tool_parameter_defs(schema: Any) -> Any:
+    """Hoist tool parameter ``$defs`` to the guided schema's document root.
+
+    SGLang wraps named/required tool parameters in an outer tool-call schema.
+    A parameter schema's local ``#/$defs/...`` references then point at the
+    outer document root, while its ``$defs`` would otherwise remain nested
+    below ``items.properties.parameters``. Move those definitions to the
+    outer root and rename on collision.
+
+    Allocation is done in two passes: every final name is reserved up front
+    — against the existing root ``$defs``, against names allocated earlier
+    in this pass, and against original names not yet processed — and all
+    references are rewritten only after every name is known.  Definitions
+    are never deduplicated by dict equality: two definitions with identical
+    literals can still differ semantically once their own ``$ref``
+    dependencies are resolved (review 20260924 finding 3).
+    """
+    if not isinstance(schema, dict):
+        return schema
+
+    schema = copy.deepcopy(schema)
+    had_root_defs = "$defs" in schema
+    root_defs = schema.get("$defs")
+    if root_defs is None:
+        root_defs = {}
+        schema["$defs"] = root_defs
+    elif not isinstance(root_defs, dict):
+        return schema
+
+    # Pass 1: collect every parameters node carrying $defs, in visit order.
+    entries: list[tuple[dict[str, Any], str, dict[str, Any]]] = []
+
+    def collect_entries(node: Any) -> None:
+        if isinstance(node, list):
+            for child in node:
+                collect_entries(child)
+            return
+        if not isinstance(node, dict):
+            return
+
+        properties = node.get("properties")
+        if isinstance(properties, dict) and "name" in properties:
+            parameters = properties.get("parameters")
+            if isinstance(parameters, dict):
+                name_schema = properties["name"]
+                names = (
+                    name_schema.get("enum", [])
+                    if isinstance(name_schema, dict)
+                    else []
+                )
+                tool_name = names[0] if isinstance(names, list) and names else ""
+                nested_defs = parameters.get("$defs")
+                if isinstance(nested_defs, dict):
+                    entries.append((parameters, tool_name, nested_defs))
+
+        for keyword in ("items", "anyOf", "oneOf", "allOf", "prefixItems"):
+            if keyword in node:
+                collect_entries(node[keyword])
+
+    collect_entries(schema)
+
+    # Pass 2: allocate a unique final name per definition.  A definition
+    # keeps its original name only when no root definition and no earlier
+    # allocation already holds it; otherwise it is renamed with a
+    # ``{tool_name}__`` prefix and a numeric suffix until free.  Because
+    # allocations are sequential and never deduplicated, a renamed
+    # definition can shadow a not-yet-processed original name only by
+    # taking it first — the later original is then renamed instead, and
+    # both keep their own rewritten references.
+    used: set[str] = set(root_defs)
+    allocations: list[tuple[dict[str, Any], dict[str, Any], dict[str, str]]] = []
+    for parameters, tool_name, nested_defs in entries:
+        names: dict[str, str] = {}
+        for old_name in nested_defs:
+            new_name = old_name
+            if new_name in used:
+                base_name = f"{tool_name}__{old_name}" if tool_name else old_name
+                new_name = base_name
+                suffix = 2
+                while new_name in used:
+                    new_name = f"{base_name}_{suffix}"
+                    suffix += 1
+            used.add(new_name)
+            names[old_name] = new_name
+        allocations.append((parameters, nested_defs, names))
+
+    # Pass 3: detach $defs, rewrite references, and write to the root.
+    for parameters, nested_defs, names in allocations:
+        parameters.pop("$defs", None)
+        _rewrite_def_refs(parameters, names)
+        for definition in nested_defs.values():
+            _rewrite_def_refs(definition, names)
+        for old_name, definition in nested_defs.items():
+            root_defs[names[old_name]] = definition
+
+    if not had_root_defs and not root_defs:
+        schema.pop("$defs", None)
+    return schema
+
+
 def build_tool_call_guided_decoding(
     request: dict[str, Any],
     *,
@@ -749,6 +955,13 @@ def build_tool_call_guided_decoding(
     if tool_choice == "none":
         return None
 
+    # Isolate per-tool parameter $defs BEFORE handing tools to SGLang's
+    # constraint builders: their _get_tool_schema_defs() raises ValueError
+    # on same-name/different-schema definitions, which would otherwise
+    # fire before the post-builder hoisting below can run (review 20260924
+    # finding 2).  At most one tool with $defs passes through unchanged.
+    sglang_tools = _namespace_tool_parameter_defs(sglang_tools)
+
     parallel_tool_calls = request.get("parallel_tool_calls")
     constraint: Any = None
 
@@ -788,7 +1001,7 @@ def build_tool_call_guided_decoding(
 
     if isinstance(constraint, tuple) and len(constraint) == 2:
         if constraint[0] == "json_schema":
-            return {"json": constraint[1]}
+            return {"json": _hoist_tool_parameter_defs(constraint[1])}
         if constraint[0] == "structural_tag":
             tag_value = constraint[1]
             # SGLang returns a Pydantic model (LegacyStructuralTagResponseFormat)
diff --git a/dingo/frontend/tests/test_sglang_processor_unit.py b/dingo/frontend/tests/test_sglang_processor_unit.py
index c7ab2ae4bfd3..23dd937bb742 100644
--- a/dingo/frontend/tests/test_sglang_processor_unit.py
+++ b/dingo/frontend/tests/test_sglang_processor_unit.py
@@ -28,6 +28,8 @@
     _flatten_message_content,
     _guided_json_start_chars,
     _guided_output_requires_reasoning,
+    _hoist_tool_parameter_defs,
+    _namespace_tool_parameter_defs,
     _normalize_assistant_tool_call_arguments,
     _normalize_prompt_token_ids,
     _normalize_sglang_parser_name,
@@ -1286,6 +1288,372 @@ def test_required_without_parser_name_still_creates_json_array_parser(self):
         assert isinstance(tcp, JsonArrayParser)
 
 
+# ---------------------------------------------------------------------------
+# _hoist_tool_parameter_defs / _namespace_tool_parameter_defs
+# ---------------------------------------------------------------------------
+
+
+def _outer_tool_schema(*tool_params: tuple[str, dict]) -> dict:
+    """Build an outer tool-call schema shaped like SGLang's named/required
+    wrapper: each tool appears as an anyOf branch whose properties carry a
+    ``name`` enum and a ``parameters`` subschema."""
+    return {
+        "type": "array",
+        "items": {
+            "anyOf": [
+                {
+                    "type": "object",
+                    "properties": {
+                        "name": {"enum": [tool_name]},
+                        "parameters": params,
+                    },
+                }
+                for tool_name, params in tool_params
+            ]
+        },
+    }
+
+
+def _branch_params(schema: dict, tool_name: str) -> dict:
+    for branch in schema["items"]["anyOf"]:
+        if branch["properties"]["name"]["enum"] == [tool_name]:
+            return branch["properties"]["parameters"]
+    raise KeyError(tool_name)
+
+
+class TestHoistToolParameterDefs:  # FRONTEND.3 — $defs hoisting collision safety
+    def test_single_tool_keeps_original_names(self):
+        schema = _outer_tool_schema(
+            (
+                "get_weather",
+                {
+                    "type": "object",
+                    "$defs": {"City": {"type": "string"}},
+                    "properties": {"city": {"$ref": "#/$defs/City"}},
+                },
+            )
+        )
+        result = _hoist_tool_parameter_defs(schema)
+        assert result["$defs"]["City"] == {"type": "string"}
+        params = _branch_params(result, "get_weather")
+        assert "$defs" not in params
+        assert params["properties"]["city"]["$ref"] == "#/$defs/City"
+
+    def test_shared_literal_defs_with_different_dependencies(self):
+        """Defect repro (review 20260924 f3-1): two tools hold literally
+        identical ``Payload`` definitions that reference *different* Leaf
+        schemas.  Dict-equality dedup would keep the shared name and let the
+        later tool's rewritten Payload overwrite the earlier one; every
+        definition must instead get its own name."""
+        a_params = {
+            "type": "object",
+            "$defs": {
+                "Payload": {"$ref": "#/$defs/Leaf"},
+                "Leaf": {"type": "string"},
+            },
+            "properties": {"data": {"$ref": "#/$defs/Payload"}},
+        }
+        b_params = {
+            "type": "object",
+            "$defs": {
+                "Payload": {"$ref": "#/$defs/Leaf"},
+                "Leaf": {"type": "integer"},
+            },
+            "properties": {"data": {"$ref": "#/$defs/Payload"}},
+        }
+        result = _hoist_tool_parameter_defs(
+            _outer_tool_schema(("a", a_params), ("b", b_params))
+        )
+        root = result["$defs"]
+        # a keeps its original names; b is fully renamed.
+        assert root["Leaf"] == {"type": "string"}
+        assert root["b__Leaf"] == {"type": "integer"}
+        assert root["Payload"]["$ref"] == "#/$defs/Leaf"
+        assert root["b__Payload"]["$ref"] == "#/$defs/b__Leaf"
+        # References in each tool's parameters follow their own definitions.
+        assert _branch_params(result, "a")["properties"]["data"]["$ref"] == (
+            "#/$defs/Payload"
+        )
+        assert _branch_params(result, "b")["properties"]["data"]["$ref"] == (
+            "#/$defs/b__Payload"
+        )
+
+    def test_generated_name_does_not_shadow_pending_definition(self):
+        """Defect repro (review 20260924 f3-2): b defines an integer
+        ``Payload`` and a boolean ``b__Payload``.  Renaming b's ``Payload``
+        to ``b__Payload`` must not collide with b's own boolean definition;
+        whichever is allocated later must move aside."""
+        a_params = {
+            "type": "object",
+            "$defs": {"Payload": {"type": "string"}},
+            "properties": {"data": {"$ref": "#/$defs/Payload"}},
+        }
+        b_params = {
+            "type": "object",
+            "$defs": {
+                "Payload": {"type": "integer"},
+                "b__Payload": {"type": "boolean"},
+            },
+            "properties": {
+                "num": {"$ref": "#/$defs/Payload"},
+                "flag": {"$ref": "#/$defs/b__Payload"},
+            },
+        }
+        result = _hoist_tool_parameter_defs(
+            _outer_tool_schema(("a", a_params), ("b", b_params))
+        )
+        root = result["$defs"]
+        assert root["Payload"] == {"type": "string"}
+        # The integer and the boolean must survive under distinct names.
+        types = sorted(
+            d["type"] for name, d in root.items() if name.startswith("b__")
+        )
+        assert types == ["boolean", "integer"]
+        b = _branch_params(result, "b")
+        integer_name = b["properties"]["num"]["$ref"].rsplit("/", 1)[-1]
+        boolean_name = b["properties"]["flag"]["$ref"].rsplit("/", 1)[-1]
+        assert root[integer_name] == {"type": "integer"}
+        assert root[boolean_name] == {"type": "boolean"}
+        assert integer_name != boolean_name
+
+    def test_input_schema_is_not_mutated(self):
+        params = {
+            "type": "object",
+            "$defs": {"City": {"type": "string"}},
+            "properties": {"city": {"$ref": "#/$defs/City"}},
+        }
+        schema = _outer_tool_schema(("get_weather", params))
+        _hoist_tool_parameter_defs(schema)
+        assert "$defs" in params  # original parameters untouched
+
+    def test_const_literal_with_ref_is_not_rewritten(self):
+        """Review 20260924 v2: a {"$ref": ...} object inside const/enum is
+        instance data the model must output, not a schema reference.  When
+        a rename happens (here: the root already holds a builder-merged
+        City, so the nested copy is renamed), literal values must stay
+        byte-identical while schema-position refs are rewritten."""
+        params = {
+            "type": "object",
+            "$defs": {"City": {"type": "string"}},
+            "properties": {
+                "city": {"$ref": "#/$defs/City"},
+                "document": {"const": {"$ref": "#/$defs/City"}},
+            },
+            "required": ["city", "document"],
+        }
+        schema = _outer_tool_schema(("submit", params))
+        schema["$defs"] = {"City": {"type": "string"}}  # builder-merged root
+        result = _hoist_tool_parameter_defs(schema)
+        params_out = _branch_params(result, "submit")
+        # Schema-position ref follows the renamed definition...
+        assert params_out["properties"]["city"]["$ref"] == (
+            "#/$defs/submit__City"
+        )
+        assert result["$defs"]["submit__City"] == {"type": "string"}
+        # ...but the const literal is instance data and must not change.
+        assert params_out["properties"]["document"]["const"] == {
+            "$ref": "#/$defs/City"
+        }
+
+    def test_enum_literals_preserved_but_anyof_refs_rewritten(self):
+        """Guard both directions: enum object literals are instance data
+        (skip), while anyOf arrays hold real subschemas (must rewrite)."""
+        params = {
+            "type": "object",
+            "$defs": {"City": {"type": "string"}},
+            "properties": {
+                "kind": {
+                    "enum": [{"$ref": "#/$defs/City"}, "plain"],
+                },
+                "city": {"anyOf": [{"$ref": "#/$defs/City"}, {"type": "null"}]},
+            },
+        }
+        schema = _outer_tool_schema(("submit", params))
+        schema["$defs"] = {"City": {"type": "string"}}
+        result = _hoist_tool_parameter_defs(schema)
+        params_out = _branch_params(result, "submit")
+        assert params_out["properties"]["kind"]["enum"] == [
+            {"$ref": "#/$defs/City"},
+            "plain",
+        ]
+        assert params_out["properties"]["city"]["anyOf"][0]["$ref"] == (
+            "#/$defs/submit__City"
+        )
+
+
+class TestNamespaceToolParameterDefs:  # FRONTEND.3 — pre-builder $defs isolation
+    @staticmethod
+    def _make_tools(*specs: tuple[str, dict]):
+        return convert_tools(
+            [
+                {
+                    "type": "function",
+                    "function": {
+                        "name": name,
+                        "description": name,
+                        "parameters": params,
+                    },
+                }
+                for name, params in specs
+            ]
+        )
+
+    def test_single_tool_with_defs_passes_through(self):
+        tools = self._make_tools(
+            (
+                "get_weather",
+                {
+                    "type": "object",
+                    "$defs": {"City": {"type": "string"}},
+                    "properties": {"city": {"$ref": "#/$defs/City"}},
+                },
+            )
+        )
+        result = _namespace_tool_parameter_defs(tools)
+        assert result is tools  # untouched: no cross-tool conflict possible
+
+    def test_conflicting_names_are_isolated(self):
+        tools = self._make_tools(
+            (
+                "a",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "string"}},
+                    "properties": {"data": {"$ref": "#/$defs/Payload"}},
+                },
+            ),
+            (
+                "b",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "integer"}},
+                    "properties": {"data": {"$ref": "#/$defs/Payload"}},
+                },
+            ),
+        )
+        result = _namespace_tool_parameter_defs(tools)
+        a, b = result
+        assert set(a.function.parameters["$defs"]) == {"a__Payload"}
+        assert set(b.function.parameters["$defs"]) == {"b__Payload"}
+        assert a.function.parameters["properties"]["data"]["$ref"] == (
+            "#/$defs/a__Payload"
+        )
+        assert b.function.parameters["properties"]["data"]["$ref"] == (
+            "#/$defs/b__Payload"
+        )
+        # Input tools are not mutated (the copy is deep).
+        assert set(tools[0].function.parameters["$defs"]) == {"Payload"}
+        assert tools[0].function.parameters["properties"]["data"]["$ref"] == (
+            "#/$defs/Payload"
+        )
+
+    def test_generated_names_cannot_collide(self):
+        """tool ``a`` defining ``b__X`` and tool ``a__b`` defining ``X``
+        both map to ``a__b__X``; the later one must get a suffix."""
+        tools = self._make_tools(
+            (
+                "a",
+                {
+                    "type": "object",
+                    "$defs": {"b__X": {"type": "string"}},
+                    "properties": {"v": {"$ref": "#/$defs/b__X"}},
+                },
+            ),
+            (
+                "a__b",
+                {
+                    "type": "object",
+                    "$defs": {"X": {"type": "integer"}},
+                    "properties": {"v": {"$ref": "#/$defs/X"}},
+                },
+            ),
+        )
+        result = _namespace_tool_parameter_defs(tools)
+        names_a = set(result[0].function.parameters["$defs"])
+        names_b = set(result[1].function.parameters["$defs"])
+        assert names_a != names_b
+        assert not (names_a & names_b)
+        # References follow each tool's own final name.
+        for tool in result:
+            params = tool.function.parameters
+            final = next(iter(params["$defs"]))
+            assert params["properties"]["v"]["$ref"] == f"#/$defs/{final}"
+
+    def test_default_and_examples_literals_are_preserved(self):
+        """Review 20260924 v2: the namespace stage shares the same rewrite
+        helper, so literal {"$ref": ...} objects in default/examples must
+        also survive namespacing while schema refs are renamed."""
+        tools = self._make_tools(
+            (
+                "a",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "string"}},
+                    "properties": {
+                        "data": {"$ref": "#/$defs/Payload"},
+                        "meta": {
+                            "type": "string",
+                            "default": {"$ref": "#/$defs/Payload"},
+                            "examples": [{"$ref": "#/$defs/Payload"}],
+                        },
+                    },
+                },
+            ),
+            (
+                "b",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "integer"}},
+                    "properties": {"data": {"$ref": "#/$defs/Payload"}},
+                },
+            ),
+        )
+        result = _namespace_tool_parameter_defs(tools)
+        meta = result[0].function.parameters["properties"]["meta"]
+        assert meta["default"] == {"$ref": "#/$defs/Payload"}
+        assert meta["examples"] == [{"$ref": "#/$defs/Payload"}]
+        assert result[0].function.parameters["properties"]["data"]["$ref"] == (
+            "#/$defs/a__Payload"
+        )
+
+    def test_required_multi_tool_conflicting_defs_real_builder(self):
+        """Integration through SGLang's real builder (review 20260924 f2):
+        two tools with same-named, different-content ``$defs`` and
+        tool_choice='required' previously raised ``ValueError: Tool
+        definition 'Payload' has multiple schemas`` before any post-
+        processing could run."""
+        tools = self._make_tools(
+            (
+                "a",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "string"}},
+                    "properties": {"data": {"$ref": "#/$defs/Payload"}},
+                },
+            ),
+            (
+                "b",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "integer"}},
+                    "properties": {"data": {"$ref": "#/$defs/Payload"}},
+                },
+            ),
+        )
+        guided = build_tool_call_guided_decoding(
+            {"tool_choice": "required"},
+            tool_call_parser_name=None,
+            sglang_tools=tools,
+        )
+        assert isinstance(guided, dict) and "json" in guided
+        schema_text = json.dumps(guided["json"])
+        # Both tools' definitions survive under isolated names, and no
+        # reference still points at the un-namespaced original.
+        assert "a__Payload" in schema_text
+        assert "b__Payload" in schema_text
+        assert '"$ref": "#/$defs/Payload"' not in schema_text
+
+
 # ---------------------------------------------------------------------------
 # _parse_json_array_buffer
 # ---------------------------------------------------------------------------

From edf9861f26e2c94c42ea24ac72f764bccf173c45 Mon Sep 17 00:00:00 2001
From: guojn1 <gjn16637211661@163.com>
Date: Sat, 26 Sep 2026 07:08:03 +0800
Subject: [PATCH 3/6] test(frontend): cover JSON-pointer suffix preservation in
 $defs ref rewrites

A $ref below the definition (#/$defs/City/properties/name) must keep
its pointer suffix when the leading definition token is renamed during
hoisting; only the definition name is remapped.  This path
(partition('/') + suffix re-append) had no coverage.  Verified locally
via AST extraction of the shipped functions: the new scenario passes
and all 35 prior checks remain green.  Production code unchanged.
---
 .../tests/test_sglang_processor_unit.py       | 31 +++++++++++++++++++
 1 file changed, 31 insertions(+)

diff --git a/dingo/frontend/tests/test_sglang_processor_unit.py b/dingo/frontend/tests/test_sglang_processor_unit.py
index 23dd937bb742..1a8fb6a6a0f4 100644
--- a/dingo/frontend/tests/test_sglang_processor_unit.py
+++ b/dingo/frontend/tests/test_sglang_processor_unit.py
@@ -1480,6 +1480,37 @@ def test_enum_literals_preserved_but_anyof_refs_rewritten(self):
             "#/$defs/submit__City"
         )
 
+    def test_pointer_suffix_preserved_on_rename(self):
+        """A $ref below the definition (#/$defs/City/properties/name) must
+        keep its JSON-pointer suffix when the leading definition token is
+        renamed; only the definition name is remapped."""
+        params = {
+            "type": "object",
+            "$defs": {
+                "City": {
+                    "type": "object",
+                    "properties": {"name": {"type": "string"}},
+                }
+            },
+            "properties": {
+                "city_name": {"$ref": "#/$defs/City/properties/name"},
+            },
+        }
+        schema = _outer_tool_schema(("submit", params))
+        # Root already holds a builder-merged City → nested copy is renamed.
+        schema["$defs"] = {
+            "City": {"type": "object", "properties": {"name": {"type": "string"}}}
+        }
+        result = _hoist_tool_parameter_defs(schema)
+        params_out = _branch_params(result, "submit")
+        assert params_out["properties"]["city_name"]["$ref"] == (
+            "#/$defs/submit__City/properties/name"
+        )
+        # The renamed definition itself is intact for the suffix to resolve.
+        assert result["$defs"]["submit__City"]["properties"]["name"] == {
+            "type": "string"
+        }
+
 
 class TestNamespaceToolParameterDefs:  # FRONTEND.3 — pre-builder $defs isolation
     @staticmethod

From 95fd0ac77e5ee90ee3926d2fd3b09154f63c828a Mon Sep 17 00:00:00 2001
From: guojn1 <gjn16637211661@163.com>
Date: Sat, 26 Sep 2026 08:52:13 +0800
Subject: [PATCH 4/6] fix(sglang): pass through logprobs data to OpenAI
 response

Frontend SGLang path discarded logprobs from the backend, causing
logprobs:true requests to return null instead of probability data.

- sglang_processor.py: accumulate pending_log_probs/pending_top_logprobs,
  pass them into mapped_response on flush
- sglang_prepost.py: add _build_logprobs to assemble OpenAI-format
  logprobs, replace hardcoded logprobs:None in process_output
- non-streaming path handled by Rust DeltaAggregator automatically
---
 dingo/frontend/sglang_prepost.py   | 91 +++++++++++++++++++++++++++++-
 dingo/frontend/sglang_processor.py | 36 +++++++++++-
 2 files changed, 123 insertions(+), 4 deletions(-)

diff --git a/dingo/frontend/sglang_prepost.py b/dingo/frontend/sglang_prepost.py
index a054000d2d68..6b2807b43fa4 100644
--- a/dingo/frontend/sglang_prepost.py
+++ b/dingo/frontend/sglang_prepost.py
@@ -1392,6 +1392,8 @@ def __init__(
         guided_decoding: dict[str, Any] | None = None,
         response_format_guided_active: bool = False,
         tool_guided_active: bool = False,
+        logprobs_enabled: bool = False,
+        return_tokens_as_token_ids: bool = False,
     ) -> None:
         self.tokenizer = tokenizer
         self.tool_call_parser = tool_call_parser
@@ -1418,6 +1420,8 @@ def __init__(
             response_format_guided_active=response_format_guided_active,
             tool_guided_active=tool_guided_active,
         )
+        self._logprobs_enabled = logprobs_enabled
+        self._return_tokens_as_token_ids = return_tokens_as_token_ids
         self._pending_guided_reasoning_prefix: str | None = (
             ""
             if reasoning_parser is not None and self._guided_json_start_chars
@@ -1710,6 +1714,70 @@ def _parse_reasoning_delta(
         )
         return reasoning_text, normal_text, reasoning_tokens
 
+
+    def _build_logprobs(
+        self,
+        token_ids: list[int],
+        log_probs: list[float] | None,
+        top_logprobs: list[list[dict[str, Any]]] | None,
+    ) -> dict[str, Any] | None:
+        """Build OpenAI-format ``logprobs`` from raw SGLang logprob arrays.
+
+        Returns ``{"content": [...]}`` or ``None`` when no logprobs are
+        available for this chunk.
+        """
+        if not self._logprobs_enabled or log_probs is None:
+            return None
+
+        content: list[dict[str, Any]] = []
+        for i, lp in enumerate(log_probs):
+            tid = token_ids[i] if i < len(token_ids) else 0
+            if self._return_tokens_as_token_ids:
+                token_str = f"token_id:{tid}"
+            else:
+                try:
+                    token_str = self.tokenizer.decode(
+                        [tid], skip_special_tokens=False
+                    )
+                except Exception:
+                    token_str = ""
+            token_bytes = list(token_str.encode("utf-8")) if token_str else None
+
+            top_list: list[dict[str, Any]] = []
+            if top_logprobs and i < len(top_logprobs):
+                for entry in top_logprobs[i]:
+                    top_tid = entry.get("token_id", 0)
+                    if self._return_tokens_as_token_ids:
+                        top_str = f"token_id:{top_tid}"
+                    else:
+                        top_str = entry.get("token", "")
+                        if not top_str:
+                            try:
+                                top_str = self.tokenizer.decode(
+                                    [top_tid], skip_special_tokens=False
+                                )
+                            except Exception:
+                                top_str = ""
+                    top_bytes = list(top_str.encode("utf-8")) if top_str else None
+                    top_list.append(
+                        {
+                            "token": top_str,
+                            "bytes": top_bytes,
+                            "logprob": entry.get("logprob", 0.0),
+                        }
+                    )
+
+            content.append(
+                {
+                    "token": token_str,
+                    "bytes": token_bytes,
+                    "logprob": lp,
+                    "top_logprobs": top_list,
+                }
+            )
+
+        return {"content": content}
+
     def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | None:
         """Process a single engine response chunk into an OpenAI SSE choice dict.
 
@@ -1726,6 +1794,23 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
         if finished:
             token_ids = self._strip_trailing_eos_token_ids(list(token_ids))
 
+        # Extract raw logprobs from the engine response.  The backend
+        # (decode_handler / llm_engine) already sliced SGLang's cumulative
+        # arrays to the new tokens in this chunk, so log_probs and
+        # top_logprobs align 1:1 with the pre-eos-trim token_ids.
+        raw_log_probs = engine_response.get("log_probs")
+        raw_top_logprobs = engine_response.get("top_logprobs")
+
+        # Trim logprobs to match eos-stripped token_ids.
+        if raw_log_probs and len(raw_log_probs) > len(token_ids):
+            raw_log_probs = raw_log_probs[: len(token_ids)]
+        if raw_top_logprobs and len(raw_top_logprobs) > len(token_ids):
+            raw_top_logprobs = raw_top_logprobs[: len(token_ids)]
+
+        logprobs_payload = self._build_logprobs(
+            token_ids, raw_log_probs, raw_top_logprobs
+        )
+
         # A terminal engine chunk commonly contains no token_ids. Still run
         # detokenization so a previously withheld suffix can be flushed.
         delta_text = (
@@ -1743,14 +1828,14 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
                     "index": 0,
                     "delta": {"role": "assistant", "content": delta_text},
                     "finish_reason": finish_reason,
-                    "logprobs": None,
+                    "logprobs": logprobs_payload,
                 }
             elif finish_reason:
                 return {
                     "index": 0,
                     "delta": {},
                     "finish_reason": finish_reason,
-                    "logprobs": None,
+                    "logprobs": logprobs_payload,
                 }
             return None
 
@@ -2069,7 +2154,7 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
                 "index": 0,
                 "delta": delta if has_content else {},
                 "finish_reason": effective_finish,
-                "logprobs": None,
+                "logprobs": logprobs_payload,
             }
 
         return None
diff --git a/dingo/frontend/sglang_processor.py b/dingo/frontend/sglang_processor.py
index ecf58a420969..0230830ea4cb 100644
--- a/dingo/frontend/sglang_processor.py
+++ b/dingo/frontend/sglang_processor.py
@@ -590,6 +590,15 @@ async def _generator_inner(
             guided_decoding=pre.guided_decoding,
             response_format_guided_active=pre.response_format_guided_active,
             tool_guided_active=pre.tool_guided_active,
+            logprobs_enabled=(
+                (dynamo_preproc.get("output_options") or {}).get("logprobs")
+                is not None
+            ),
+            return_tokens_as_token_ids=bool(
+                (dynamo_preproc.get("output_options") or {}).get(
+                    "return_tokens_as_token_ids"
+                )
+            ),
         )
 
         async for item in self._generate_and_stream(
@@ -659,6 +668,17 @@ async def _generator_inner_pool(
                 preproc_result.response_format_guided_active
             ),
             tool_guided_active=preproc_result.tool_guided_active,
+            logprobs_enabled=bool(
+                (preproc_result.dynamo_preproc.get("output_options") or {}).get(
+                    "logprobs"
+                )
+                is not None
+            ),
+            return_tokens_as_token_ids=bool(
+                (preproc_result.dynamo_preproc.get("output_options") or {}).get(
+                    "return_tokens_as_token_ids"
+                )
+            ),
         )
 
         async for item in self._generate_and_stream(
@@ -697,6 +717,8 @@ async def _generate_and_stream(
             # TTFT, then switch to the configured interval.
             pending_token_ids: list[int] = []
             pending_usage: dict[str, Any] | None = None
+            pending_log_probs: list[float] | None = None
+            pending_top_logprobs: list[list[dict[str, Any]]] | None = None
             first_chunk = True
             input_tokens = len(tokens)
             cumulative_output_tokens = 0
@@ -735,8 +757,16 @@ async def _generate_and_stream(
 
                 if usage := engine_response.get("completion_usage"):
                     pending_usage = usage
-                engine_data = engine_response.get("engine_data")
                 pending_token_ids.extend(new_ids)
+                engine_data = engine_response.get("engine_data")
+                if chunk_log_probs := engine_response.get("log_probs"):
+                    if pending_log_probs is None:
+                        pending_log_probs = []
+                    pending_log_probs.extend(chunk_log_probs)
+                if chunk_top_logprobs := engine_response.get("top_logprobs"):
+                    if pending_top_logprobs is None:
+                        pending_top_logprobs = []
+                    pending_top_logprobs.extend(chunk_top_logprobs)
 
                 # Flush on finish or when we've accumulated enough tokens.
                 # First chunk flushes immediately (si=1) to minimize TTFT.
@@ -746,6 +776,8 @@ async def _generate_and_stream(
                     mapped_response = {
                         "token_ids": pending_token_ids,
                         "finish_reason": finish_reason,
+                        "log_probs": pending_log_probs,
+                        "top_logprobs": pending_top_logprobs,
                     }
 
                     if self.debug_perf:
@@ -850,6 +882,8 @@ async def _generate_and_stream(
                         yield envelope
 
                     pending_token_ids = []
+                    pending_log_probs = None
+                    pending_top_logprobs = None
                     pending_usage = None
                     first_chunk = False
         except Unknown:

From b206dac0c82468d7dd982b2b962e64449df4a94f Mon Sep 17 00:00:00 2001
From: guojn1 <gjn16637211661@163.com>
Date: Sat, 26 Sep 2026 09:22:48 +0800
Subject: [PATCH 5/6] fix(sglang): use incremental mode for logprobs extraction
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Dynamo forces incremental_streaming_output=True, so SGLang sends
disjoint logprob arrays per chunk. extract_from_sglang_meta was always
slicing with num_output_logprobs_so_far (cumulative mode), causing
all chunks after the first to return None — only the first chunk's
logprobs survived.

- logprobs.py: add incremental param; when True, skip slicing and
  use output_token_logprobs directly as disjoint entries
- decode_handler.py + llm_engine.py: pass incremental=True
- test_logprobs.py: add incremental mode tests
---
 dingo/common/backend/logprobs.py              | 36 +++++++++++++-----
 dingo/common/backend/tests/test_logprobs.py   | 37 +++++++++++++++++++
 dingo/sglang/llm_engine.py                    |  1 +
 .../request_handlers/llm/decode_handler.py    |  1 +
 4 files changed, 66 insertions(+), 9 deletions(-)

diff --git a/dingo/common/backend/logprobs.py b/dingo/common/backend/logprobs.py
index cf4fbbbeb663..6bab1cf2a68f 100644
--- a/dingo/common/backend/logprobs.py
+++ b/dingo/common/backend/logprobs.py
@@ -302,28 +302,44 @@ def extract_from_sglang_meta(
     num_output_logprobs_so_far: int,
     *,
     return_tokens_as_token_ids: bool = False,
+    incremental: bool = False,
 ) -> tuple[Optional[list[float]], Optional[list[list[dict[str, Any]]]], int]:
     """Extract logprobs from SGLang's ``meta_info`` dict.
 
-    SGLang's ``output_token_logprobs`` / ``output_top_logprobs`` are
-    cumulative across stream chunks even though ``output_ids`` is
-    disjoint — the caller passes the running count to slice the new
-    entries, and the returned third element is the updated count.
+    When ``incremental_streaming_output`` is False (the SGLang default),
+    ``output_token_logprobs`` / ``output_top_logprobs`` are cumulative
+    across stream chunks even though ``output_ids`` is disjoint — the
+    caller passes the running count to slice the new entries, and the
+    returned third element is the updated count.
+
+    When ``incremental_streaming_output`` is True (Dynamo forces this on),
+    SGLang sends only the new token's logprobs in each chunk — the arrays
+    are already disjoint, so no slicing is needed.  The returned third
+    element is ``num_output_logprobs_so_far + len(new_entries)``.
     """
     output_token_logprobs = meta_info.get("output_token_logprobs")
     if not output_token_logprobs:
         return None, None, num_output_logprobs_so_far
 
-    new_logprobs = output_token_logprobs[num_output_logprobs_so_far:]
+    if incremental:
+        new_logprobs = output_token_logprobs
+        new_total = num_output_logprobs_so_far + len(output_token_logprobs)
+    else:
+        new_logprobs = output_token_logprobs[num_output_logprobs_so_far:]
+        new_total = len(output_token_logprobs)
+
     if not new_logprobs:
-        return None, None, num_output_logprobs_so_far
+        return None, None, new_total
 
     log_probs = [float(entry[0]) for entry in new_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[num_output_logprobs_so_far:]
+        if incremental:
+            new_top = output_top
+        else:
+            new_top = output_top[num_output_logprobs_so_far:]
         if new_top:
             top_logprobs = []
             for position_entries in new_top:
@@ -334,7 +350,9 @@ def extract_from_sglang_meta(
                 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]
+                        f"token_id:{tok_id}"
+                        if return_tokens_as_token_ids
+                        else entry[2]
                     )
                     position_list.append(
                         {
@@ -346,4 +364,4 @@ def extract_from_sglang_meta(
                     )
                 top_logprobs.append(position_list)
 
-    return log_probs, top_logprobs, len(output_token_logprobs)
+    return log_probs, top_logprobs, new_total
diff --git a/dingo/common/backend/tests/test_logprobs.py b/dingo/common/backend/tests/test_logprobs.py
index d07a57d9077c..45bfa00ff797 100644
--- a/dingo/common/backend/tests/test_logprobs.py
+++ b/dingo/common/backend/tests/test_logprobs.py
@@ -414,6 +414,43 @@ def test_sglang_extract_returns_offset_unchanged_when_no_new_entries():
     assert new_total == 1
 
 
+def test_sglang_extract_incremental_no_slice():
+    """incremental=True: arrays are already disjoint, no slicing."""
+    meta = {
+        "output_token_logprobs": [(-0.5, 10, "x")],
+        "output_top_logprobs": [[(-0.5, 10, "x"), (-0.8, 11, "y")]],
+    }
+    log_probs, top_logprobs, new_total = extract_from_sglang_meta(
+        meta, 5, incremental=True
+    )
+    assert log_probs == [-0.5]
+    assert top_logprobs == [
+        [
+            {"rank": 1, "token_id": 10, "token": "x", "logprob": -0.5},
+            {"rank": 2, "token_id": 11, "token": "y", "logprob": -0.8},
+        ]
+    ]
+    assert new_total == 6
+
+
+def test_sglang_extract_incremental_empty_array():
+    """incremental=True: empty output_token_logprobs returns None."""
+    meta = {"output_token_logprobs": []}
+    log_probs, _, new_total = extract_from_sglang_meta(meta, 3, incremental=True)
+    assert log_probs is None
+    assert new_total == 3
+
+
+def test_sglang_extract_incremental_no_top_logprobs():
+    """incremental=True: log_probs without top_logprobs."""
+    meta = {"output_token_logprobs": [(-0.1, 1, "a")]}
+    log_probs, top_logprobs, new_total = extract_from_sglang_meta(
+        meta, 0, incremental=True
+    )
+    assert log_probs == [-0.1]
+    assert top_logprobs is None
+    assert new_total == 1
+
 # ---------------------------------------------------------------------------
 # Legacy ↔ unified behavioural parity corner cases.
 #
diff --git a/dingo/sglang/llm_engine.py b/dingo/sglang/llm_engine.py
index b4f0758b04ab..3aa2f5edd6ed 100644
--- a/dingo/sglang/llm_engine.py
+++ b/dingo/sglang/llm_engine.py
@@ -481,6 +481,7 @@ async def generate(
                     meta_info,
                     num_logprobs_per_choice.get(output_idx, 0),
                     return_tokens_as_token_ids=return_tokens_as_token_ids,
+                    incremental=True,
                 )
                 num_logprobs_per_choice[output_idx] = next_total
                 if log_probs is not None:
diff --git a/dingo/sglang/request_handlers/llm/decode_handler.py b/dingo/sglang/request_handlers/llm/decode_handler.py
index 504268f19945..2e80909d8d6d 100644
--- a/dingo/sglang/request_handlers/llm/decode_handler.py
+++ b/dingo/sglang/request_handlers/llm/decode_handler.py
@@ -456,6 +456,7 @@ def _extract_logprobs(
             meta_info,
             num_output_logprobs_so_far,
             return_tokens_as_token_ids=return_tokens_as_token_ids,
+            incremental=True,
         )
 
     async def generate(

From 8c623725a21b89503e9797d8213145dbbdcf599f Mon Sep 17 00:00:00 2001
From: guojn1 <gjn16637211661@163.com>
Date: Mon, 28 Sep 2026 16:00:42 +0800
Subject: [PATCH 6/6] fix(frontend): rewrite refs in schema maps, withhold
 stranded logprobs, dedupe terminal split (review 20260928)

- _rewrite_def_refs skipped keys named const/enum/default/examples even inside properties/$defs maps where they are user-defined field names holding subschemas; their $ref dangled after $defs namespacing. Track schema-map context: skip literal data only on schema nodes, always traverse map values as subschemas (finding 1)
- process_output dropped a chunk logprobs when incremental decode withheld the text (partial UTF-8 character) and returned None; the caller clears pending arrays after flush, so the probabilities were lost. Withhold the records with the text and prepend them to the next emitted choice in token order (finding 2)
- the terminal tool-call split copied choice.logprobs onto both the payload chunk and the finish marker, double-counting terminal tokens for streaming clients and the non-stream DeltaAggregator; the finish-only choice now carries logprobs null (finding 3)
- sync dingo/sglang/CLAUDE.md: with incremental_streaming_output forced on, SGLang logprob arrays arrive per-chunk disjoint

Adds 4 regression tests covering keyword-named properties through namespacing and the real builder, withheld-text logprobs ordering, and terminal-split logprobs sent exactly once.
---
 dingo/frontend/sglang_prepost.py              |  87 ++++++--
 dingo/frontend/sglang_processor.py            |  10 +-
 .../tests/test_sglang_processor_unit.py       | 191 ++++++++++++++++++
 dingo/sglang/CLAUDE.md                        |  20 +-
 4 files changed, 284 insertions(+), 24 deletions(-)

diff --git a/dingo/frontend/sglang_prepost.py b/dingo/frontend/sglang_prepost.py
index 6b2807b43fa4..fa651df726e8 100644
--- a/dingo/frontend/sglang_prepost.py
+++ b/dingo/frontend/sglang_prepost.py
@@ -749,32 +749,58 @@ def _unescape_pointer_token(token: str) -> str:
 # them; see _rewrite_def_refs.
 _LITERAL_VALUE_KEYWORDS = frozenset({"const", "enum", "default", "examples"})
 
+# Keywords whose values are maps from an arbitrary, user-defined name to a
+# subschema.  The map keys are names, not schema keywords, so the
+# literal-value skip must not apply while iterating them: a property named
+# "default" still holds a subschema that may contain a ``$ref`` (review
+# 20260928 finding 1).
+_SCHEMA_MAP_KEYWORDS = frozenset(
+    {
+        "properties",
+        "patternProperties",
+        "$defs",
+        "definitions",
+        "dependentSchemas",
+        "dependencies",
+    }
+)
+
 
-def _rewrite_def_refs(value: Any, names: dict[str, str]) -> None:
+def _rewrite_def_refs(
+    value: Any, names: dict[str, str], *, _inside_schema_map: bool = False
+) -> None:
     """Rewrite ``#/$defs/...`` references in-place according to ``names``.
 
     Only the leading definition token is remapped; any JSON-pointer suffix
     below the definition is preserved.
     """
     if isinstance(value, dict):
-        ref = value.get("$ref")
-        if isinstance(ref, str) and ref.startswith("#/$defs/"):
-            pointer = ref[len("#/$defs/") :]
-            old_token, separator, suffix = pointer.partition("/")
-            old_name = _unescape_pointer_token(old_token)
-            new_name = names.get(old_name)
-            if new_name is not None and new_name != old_name:
-                value["$ref"] = f"#/$defs/{_escape_pointer_token(new_name)}" + (
-                    f"/{suffix}" if separator else ""
-                )
+        if not _inside_schema_map:
+            ref = value.get("$ref")
+            if isinstance(ref, str) and ref.startswith("#/$defs/"):
+                pointer = ref[len("#/$defs/") :]
+                old_token, separator, suffix = pointer.partition("/")
+                old_name = _unescape_pointer_token(old_token)
+                new_name = names.get(old_name)
+                if new_name is not None and new_name != old_name:
+                    value["$ref"] = f"#/$defs/{_escape_pointer_token(new_name)}" + (
+                        f"/{suffix}" if separator else ""
+                    )
         for key, child in value.items():
-            if key in _LITERAL_VALUE_KEYWORDS:
+            if _inside_schema_map:
+                # Map keys are user-defined names; every value is a
+                # subschema and must be traversed whatever it is called.
+                _rewrite_def_refs(child, names)
+            elif key in _LITERAL_VALUE_KEYWORDS:
                 # Literal instance data (const/enum/default/examples), not
                 # a subschema: a {"$ref": ...} object here is data the
                 # model must produce and must never be rewritten (review
                 # 20260924 v2).
                 continue
-            _rewrite_def_refs(child, names)
+            elif key in _SCHEMA_MAP_KEYWORDS:
+                _rewrite_def_refs(child, names, _inside_schema_map=True)
+            else:
+                _rewrite_def_refs(child, names)
     elif isinstance(value, list):
         for child in value:
             _rewrite_def_refs(child, names)
@@ -1434,6 +1460,11 @@ def __init__(
         self._pending_stop_text = ""
 
         self._all_token_ids: list[int] = []
+        # Logprob records built for chunks that emit no visible delta (e.g.
+        # a withheld partial UTF-8 character).  They are prepended to the
+        # next emitted choice so probabilities are never dropped with the
+        # suppressed text (review 20260928 finding 2).
+        self._withheld_logprobs: list[dict[str, Any]] = []
         # Tool call accumulation.  SGLang's streaming parser returns
         # deltas (name in one chunk, argument fragments across subsequent
         # chunks).  However, the base detector processes at most one event
@@ -1778,6 +1809,28 @@ def _build_logprobs(
 
         return {"content": content}
 
+    def _withhold_logprobs_payload(self, payload: dict[str, Any] | None) -> None:
+        """Stash logprob entries when this chunk emits no visible delta.
+
+        A chunk whose text is withheld (unfinished UTF-8 character, stop
+        string prefix, parser buffering) returns ``None``; without stashing,
+        its probabilities would be dropped and the stream would end up with
+        fewer logprob records than completion tokens.
+        """
+        if payload and payload.get("content"):
+            self._withheld_logprobs.extend(payload["content"])
+
+    def _release_logprobs_payload(
+        self, payload: dict[str, Any] | None
+    ) -> dict[str, Any] | None:
+        """Prepend withheld entries so emitted records stay in token order."""
+        if not self._withheld_logprobs:
+            return payload
+        content = self._withheld_logprobs
+        self._withheld_logprobs = []
+        content.extend((payload or {}).get("content") or [])
+        return {"content": content}
+
     def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | None:
         """Process a single engine response chunk into an OpenAI SSE choice dict.
 
@@ -1828,15 +1881,16 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
                     "index": 0,
                     "delta": {"role": "assistant", "content": delta_text},
                     "finish_reason": finish_reason,
-                    "logprobs": logprobs_payload,
+                    "logprobs": self._release_logprobs_payload(logprobs_payload),
                 }
             elif finish_reason:
                 return {
                     "index": 0,
                     "delta": {},
                     "finish_reason": finish_reason,
-                    "logprobs": logprobs_payload,
+                    "logprobs": self._release_logprobs_payload(logprobs_payload),
                 }
+            self._withhold_logprobs_payload(logprobs_payload)
             return None
 
         # -- Reasoning parsing --
@@ -2154,7 +2208,8 @@ def process_output(self, engine_response: dict[str, Any]) -> dict[str, Any] | No
                 "index": 0,
                 "delta": delta if has_content else {},
                 "finish_reason": effective_finish,
-                "logprobs": logprobs_payload,
+                "logprobs": self._release_logprobs_payload(logprobs_payload),
             }
 
+        self._withhold_logprobs_payload(logprobs_payload)
         return None
diff --git a/dingo/frontend/sglang_processor.py b/dingo/frontend/sglang_processor.py
index 0230830ea4cb..00033b1d15ac 100644
--- a/dingo/frontend/sglang_processor.py
+++ b/dingo/frontend/sglang_processor.py
@@ -875,7 +875,15 @@ async def _generate_and_stream(
                             if key not in ("usage", "nvext")
                         }
                         payload["choices"] = [{**choice, "finish_reason": None}]
-                        dynamo_out["choices"] = [{**choice, "delta": {}}]
+                        # The logprobs payload rides with the tool-call
+                        # payload only; duplicating it onto the pure finish
+                        # marker would double-count the terminal tokens when
+                        # a client (or the non-stream DeltaAggregator)
+                        # concatenates logprobs across chunks (review
+                        # 20260928 finding 3).
+                        dynamo_out["choices"] = [
+                            {**choice, "delta": {}, "logprobs": None}
+                        ]
                         yield {**envelope, "data": payload}
                         yield {"_dynamo_annotated": True, "data": dynamo_out}
                     else:
diff --git a/dingo/frontend/tests/test_sglang_processor_unit.py b/dingo/frontend/tests/test_sglang_processor_unit.py
index 1a8fb6a6a0f4..3ef50b39a2aa 100644
--- a/dingo/frontend/tests/test_sglang_processor_unit.py
+++ b/dingo/frontend/tests/test_sglang_processor_unit.py
@@ -1647,6 +1647,70 @@ def test_default_and_examples_literals_are_preserved(self):
             "#/$defs/a__Payload"
         )
 
+    def test_property_named_like_literal_keyword_is_rewritten(self):
+        """Review 20260928 f1: inside a ``properties`` map the keys are
+        user-defined names, so a property literally named
+        const/enum/default/examples still holds a subschema whose ``$ref``
+        must be rewritten alongside every other reference."""
+        for keyword in ("const", "enum", "default", "examples"):
+            tools = self._make_tools(
+                (
+                    "a",
+                    {
+                        "type": "object",
+                        "$defs": {"Payload": {"type": "string"}},
+                        "properties": {keyword: {"$ref": "#/$defs/Payload"}},
+                    },
+                ),
+                (
+                    "b",
+                    {
+                        "type": "object",
+                        "$defs": {"Payload": {"type": "string"}},
+                        "properties": {"data": {"$ref": "#/$defs/Payload"}},
+                    },
+                ),
+            )
+            result = _namespace_tool_parameter_defs(tools)
+            assert set(result[0].function.parameters["$defs"]) == {"a__Payload"}
+            assert result[0].function.parameters["properties"][keyword][
+                "$ref"
+            ] == "#/$defs/a__Payload", keyword
+
+    def test_keyword_named_property_resolves_through_real_builder(self):
+        """Review 20260928 f1 end-to-end: identical-content ``$defs`` keep
+        SGLang's builder from rejecting early, and the final guided schema
+        must not retain a dangling pointer to the pre-namespace name."""
+        tools = self._make_tools(
+            (
+                "a",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "string"}},
+                    "properties": {"default": {"$ref": "#/$defs/Payload"}},
+                    "required": ["default"],
+                },
+            ),
+            (
+                "b",
+                {
+                    "type": "object",
+                    "$defs": {"Payload": {"type": "string"}},
+                    "properties": {"data": {"$ref": "#/$defs/Payload"}},
+                },
+            ),
+        )
+        guided = build_tool_call_guided_decoding(
+            {"tool_choice": "required"},
+            tool_call_parser_name=None,
+            sglang_tools=tools,
+        )
+        assert isinstance(guided, dict) and "json" in guided
+        schema_text = json.dumps(guided["json"])
+        assert "a__Payload" in schema_text
+        assert "b__Payload" in schema_text
+        assert '"$ref": "#/$defs/Payload"' not in schema_text
+
     def test_required_multi_tool_conflicting_defs_real_builder(self):
         """Integration through SGLang's real builder (review 20260924 f2):
         two tools with same-named, different-content ``$defs`` and
@@ -3082,6 +3146,133 @@ def test_fast_path_content_output(self, tokenizer):
         assert choice["logprobs"] is None
 
 
+class TestLogprobsStreaming:  # FRONTEND.6 — logprobs passthrough integrity
+    """Logprob records must survive chunks that emit no visible text, and
+    must not be duplicated when the terminal choice is split in two."""
+
+    class _ByteTokenizer:
+        """One token id per raw byte; an incomplete trailing UTF-8
+        sequence surfaces as a single held-back U+FFFD, like HF byte-level
+        decoders feeding the incremental detokenizer."""
+
+        def decode(
+            self, token_ids: list[int], *, skip_special_tokens: bool
+        ) -> str:
+            text = bytes(token_ids).decode("utf-8", errors="replace")
+            stripped = text.rstrip("�")
+            if len(stripped) != len(text):
+                return stripped + "�"
+            return text
+
+    def test_logprobs_survive_withheld_text(self):
+        """Review 20260928 f2: the bytes of one multi-byte character
+        arriving as separate chunks used to strand every logprob record
+        except the last byte's."""
+        post = SglangStreamingPostProcessor(
+            tokenizer=self._ByteTokenizer(),
+            tool_call_parser=None,
+            reasoning_parser=None,
+            logprobs_enabled=True,
+        )
+        zhong = list("中".encode("utf-8"))
+        assert len(zhong) == 3  # the scenario is one character, three bytes
+
+        text = ""
+        collected: list[float] = []
+        for tid, lp in zip(zhong, (-0.1, -0.2, -0.3)):
+            choice = post.process_output(
+                {"token_ids": [tid], "log_probs": [lp], "finish_reason": None}
+            )
+            if choice is None:
+                continue
+            text += choice["delta"].get("content", "")
+            collected.extend(e["logprob"] for e in choice["logprobs"]["content"])
+
+        assert text == "中"
+        assert collected == [-0.1, -0.2, -0.3]
+
+        # The finish chunk must not re-send already released records.
+        final = post.process_output({"token_ids": [], "finish_reason": "stop"})
+        assert final is not None
+        if final["logprobs"]:
+            collected.extend(e["logprob"] for e in final["logprobs"]["content"])
+        assert collected == [-0.1, -0.2, -0.3]
+
+    def test_terminal_tool_chunk_sends_logprobs_once(self, tokenizer):
+        """Review 20260928 f3: the finish/tool-call split attaches
+        logprobs to the payload choice only, so a client (or the
+        non-stream DeltaAggregator) concatenating records across chunks
+        counts the terminal tokens exactly once."""
+
+        class _TerminalToolCallPost:
+            """Post-processor double: every chunk looks like the terminal
+            tool-call flush carrying one logprob record."""
+
+            def process_output(self, mapped):
+                return {
+                    "index": 0,
+                    "delta": {
+                        "role": "assistant",
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call_1",
+                                "type": "function",
+                                "function": {"name": "f", "arguments": "{}"},
+                            }
+                        ],
+                    },
+                    "finish_reason": "tool_calls",
+                    "logprobs": {
+                        "content": [
+                            {
+                                "token": "}",
+                                "bytes": [125],
+                                "logprob": -0.25,
+                                "top_logprobs": [],
+                            }
+                        ]
+                    },
+                }
+
+        processor = SglangProcessor(
+            tokenizer=tokenizer,
+            routed_engine=FakeRoutedEngine(
+                items=[{"token_ids": [1], "finish_reason": "stop"}]
+            ),
+            tool_call_parser_name=None,
+            reasoning_parser_name=None,
+            eos_token_ids=None,
+        )
+        post = _TerminalToolCallPost()
+
+        async def collect():
+            return [
+                item
+                async for item in processor._generate_and_stream(
+                    "req-logprobs", {"model": "test-model"}, {}, [], post
+                )
+            ]
+
+        envelopes = asyncio.run(collect())
+        data = [e["data"] for e in envelopes if "data" in e]
+        assert len(data) == 2
+
+        payload_choice = data[0]["choices"][0]
+        finish_choice = data[1]["choices"][0]
+        assert payload_choice["delta"].get("tool_calls")
+        assert payload_choice["finish_reason"] is None
+        assert finish_choice["finish_reason"] == "tool_calls"
+        assert finish_choice["delta"] == {}
+
+        collected = []
+        for chunk in data:
+            lp = chunk["choices"][0].get("logprobs")
+            if lp:
+                collected.extend(e["logprob"] for e in lp["content"])
+        assert collected == [-0.25]
+
+
 # ---------------------------------------------------------------------------
 # SglangStreamingPostProcessor: reasoning parsing
 # ---------------------------------------------------------------------------
diff --git a/dingo/sglang/CLAUDE.md b/dingo/sglang/CLAUDE.md
index ac4cf46cbf09..177cd2118418 100644
--- a/dingo/sglang/CLAUDE.md
+++ b/dingo/sglang/CLAUDE.md
@@ -256,10 +256,14 @@ on deployments where that cost is a concern.
 
 **Streaming behavior** (`_extract_logprobs`):
 
-Dynamo forces `stream_output=True` (args.py:374), making `output_ids` disjoint per chunk.
-However, SGLang's `meta_info["output_token_logprobs"]` and `meta_info["output_top_logprobs"]`
-are always **cumulative** — they grow with each chunk. The handler tracks
-`num_output_logprobs_so_far` to slice out only new entries per chunk.
+Dynamo forces `stream_output=True` (args.py:374) and
+`incremental_streaming_output=True` (args.py:512). In incremental streaming
+mode, SGLang's `meta_info["output_token_logprobs"]` and `meta_info["output_top_logprobs"]`
+arrive already sliced to the chunk's new tokens (the upstream tokenizer
+manager splits them by output offset), so the handler passes
+`incremental=True` to `extract_from_sglang_meta()` and forwards them
+directly. Only SGLang's default non-incremental mode sends cumulative
+arrays that must be sliced by `num_output_logprobs_so_far`.
 
 SGLang logprob format: `(logprob, token_id, text_or_None)` tuples.
 Dynamo output format: `log_probs` = list of floats, `top_logprobs` = list of lists of
@@ -307,9 +311,11 @@ text-to-video-diffusion.sh  # 1-2 GPUs - Text-to-video (Wan2.1)
 - **output_modalities default**: Global default is `["text"]`. Image/video diffusion
   workers must override to `["image"]`/`["video"]` or the Rust registration path tries
   to load `config.json` (which doesn't exist for diffusers models).
-- **Cumulative logprobs in streaming**: SGLang's `output_token_logprobs`/`output_top_logprobs`
-  in `meta_info` are cumulative even though `output_ids` are disjoint (stream_output=True).
-  Always slice with an offset, don't assume per-chunk logprobs.
+- **Per-chunk logprobs in incremental streaming**: with Dynamo's forced
+  `incremental_streaming_output=True`, `output_token_logprobs`/`output_top_logprobs`
+  in `meta_info` are already disjoint per chunk — forward them directly.
+  Treating them as cumulative and slicing with `num_output_logprobs_so_far`
+  returns nothing for every chunk after the first.
 - **Zombie GPU processes**: `sgl_diffusion::scheduler` spawns a child process that
   survives parent kill. Always check `nvidia-smi` after teardown.
 - **Session radix cache**: SGLang 0.5.14+ provides session-aware radix ownership