Dingo router continuous usage sglang 0.5.21

.github/dingo-images.json

@@ -21,12 +21,12 @@
     },
     {
       "framework": "vllm",
-      "enabled": true,
+      "enabled": false,
       "repository": "ai-dingo-vllm"
     },
     {
       "framework": "sglang",
-      "enabled": false,
+      "enabled": true,
       "repository": "ai-dingo-sglang",
       "docker_target": "pre_runtime"
     }

container/context.yaml

@@ -96,16 +96,12 @@ sglang:
     base_image: nvcr.io/nvidia/cuda-dl-base
     runtime_image: lmsysorg/sglang
     base_image_tag: 25.11-cuda13.0-devel-ubuntu24.04
-    runtime_image_tag: glm-5.3-flash
-    # Baseline is the TRUE base of the third-party lmsysorg/sglang image,
-    # nvidia/cuda:13.0.1-cudnn-devel-ubuntu24.04 (Docker Hub; CUDA_VERSION=13.0.1,
-    # cuDNN 9.13.0.50, ubuntu24.04), NOT a self-baseline of lmsysorg/sglang.
-    # lmsysorg/sglang is built FROM this nvidia/cuda cudnn-devel base, so this
-    # attributes everything sglang adds over the clean NVIDIA CUDA+cuDNN floor
-    # (framework, python stack, system libs) rather than hiding it. Per-arch stem;
-    # the licenses stage appends -${TARGETARCH}.cdx.json. (sglang ships no system
-    # GPL ffmpeg/codec dpkgs; the GPL imageio-ffmpeg pip binary is replaced by an
-    # in-tree LGPL ffmpeg in sglang_runtime.Dockerfile — no dpkg purge needed.)
+    runtime_image_tag: v0.5.21
+    # Historical CUDA 13.0.1 compliance corpus; not refreshed for this runtime
+    # update. Upstream v0.5.21 uses CUDA 13.0.3, so refresh this baseline before
+    # publishing compliance artifacts for the new image. Dingo image CI uses
+    # the pre_runtime target and does not publish SBOM/provenance artifacts.
+    # Per-arch stem; the licenses stage appends -${TARGETARCH}.cdx.json.
     baseline_sbom: cuda@8b2705ea
   xpu:
     base_image: intel/deep-learning-essentials

dingo/frontend/sglang_processor.py

@@ -66,7 +66,7 @@ def _cached_tokens_details_from_engine_data(
     """Extract and normalize cached_tokens_details from engine_data.
 
     The decode handler copies sglang's ``meta_info.cached_tokens_details`` into
-    ``engine_data.cached_tokens_details`` on the finish chunk. This normalizes
+    ``engine_data.cached_tokens_details`` on usage-bearing chunks. This normalizes
     the shape so the carrier payload matches sglang's native ``sglext`` field.
     Returns ``None`` when the detail is absent or malformed.
     """
@@ -97,6 +97,15 @@ def _cached_tokens_from_usage(usage: dict[str, Any] | None) -> int | None:
     return cached_tokens if isinstance(cached_tokens, int) else None
 
 
+def _continuous_usage_requested(request: dict[str, Any]) -> bool:
+    options = request.get("stream_options") or {}
+    return bool(
+        request.get("stream")
+        and options.get("include_usage")
+        and options.get("continuous_usage_stats")
+    )
+
+
 def _normalize_eos_token_ids(value: Any) -> list[int]:
     if isinstance(value, int) and not isinstance(value, bool):
         return [value]
@@ -428,6 +437,7 @@ def _build_dynamo_preproc(
                 tool_call_parser is None and reasoning_parser is None
             ),
             "return_tokens_as_token_ids": request.get("return_tokens_as_token_ids"),
+            "continuous_usage_stats": _continuous_usage_requested(request),
         },
         "eos_token_ids": _normalize_eos_token_ids(eos_token_ids),
         "annotations": [],
@@ -705,6 +715,7 @@ async def _generate_and_stream(
         post_proc_total_ms = 0.0
         created_ts = int(time.time())
         stream_interval = self.stream_interval
+        continuous_usage = _continuous_usage_requested(request)
 
         try:
             dynamo_stream = await self.routed_engine.generate(
@@ -771,7 +782,11 @@ async def _generate_and_stream(
                 # Flush on finish or when we've accumulated enough tokens.
                 # First chunk flushes immediately (si=1) to minimize TTFT.
                 flush_threshold = 1 if first_chunk else stream_interval
-                if finish_reason or len(pending_token_ids) >= flush_threshold:
+                if (
+                    finish_reason
+                    or len(pending_token_ids) >= flush_threshold
+                    or (continuous_usage and pending_usage and not pending_token_ids)
+                ):
                     usage_for_metrics = pending_usage
                     mapped_response = {
                         "token_ids": pending_token_ids,
@@ -791,10 +806,10 @@ async def _generate_and_stream(
                         token_count += len(pending_token_ids)
 
                     envelope: dict[str, Any] = {"_dynamo_annotated": True}
-                    if choice:
+                    if choice or (continuous_usage and pending_usage):
                         dynamo_out: dict[str, Any] = {
                             "id": request_id,
-                            "choices": [choice],
+                            "choices": [choice] if choice else [],
                             "created": created_ts,
                             "model": request["model"],
                             "object": "chat.completion.chunk",
@@ -814,7 +829,7 @@ async def _generate_and_stream(
                                 pending_usage["completion_tokens_details"] = {
                                     "reasoning_tokens": reasoning_tokens
                                 }
-                        if pending_usage:
+                        if pending_usage and (finish_reason or continuous_usage):
                             dynamo_out["usage"] = pending_usage
                             pending_usage = None
                         response_nvext: dict[str, Any] = {}
@@ -827,15 +842,9 @@ async def _generate_and_stream(
                         ):
                             response_nvext["engine_data"] = engine_data
 
-                        # Promote SGLang's cached_tokens_details to the
-                        # sglext carrier on the finish chunk. The decode
-                        # handler only populates engine_data.cached_tokens_details
-                        # on the terminal chunk, so this naturally fires once.
-                        # Routed as an internal nvext key that the Rust
-                        # serializer lifts to a top-level `sglext` field,
-                        # matching sglang's native protocol — independent of
-                        # the client's engine_data opt-in.
-                        if finish_reason:
+                        # Preserve the engine's cache breakdown alongside running
+                        # usage, including parser-buffered usage-only chunks.
+                        if finish_reason or continuous_usage:
                             cached_tokens_details = (
                                 _cached_tokens_details_from_engine_data(engine_data)
                             )
@@ -874,6 +883,10 @@ async def _generate_and_stream(
                             for key, value in dynamo_out.items()
                             if key not in ("usage", "nvext")
                         }
+                        if continuous_usage and "usage" in dynamo_out:
+                            payload["usage"] = dynamo_out["usage"]
+                            if "nvext" in dynamo_out:
+                                payload["nvext"] = dynamo_out["nvext"]
                         payload["choices"] = [{**choice, "finish_reason": None}]
                         # The logprobs payload rides with the tool-call
                         # payload only; duplicating it onto the pure finish

dingo/frontend/tests/test_sglang_processor_metrics_unit.py

@@ -233,8 +233,13 @@ async def collect():
 @pytest.mark.parametrize("finish_reason", ["tool_calls", "length"])
 @pytest.mark.parametrize("tool_count", [1, 2])
 @pytest.mark.parametrize("stream_interval", [1, 20])
+@pytest.mark.parametrize("continuous", [False, True])
 def test_terminal_tool_payload_precedes_finish(
-    module_stubs: Any, finish_reason: str, tool_count: int, stream_interval: int
+    module_stubs: Any,
+    finish_reason: str,
+    tool_count: int,
+    stream_interval: int,
+    continuous: bool,
 ) -> None:
     module = _load_processor_module(module_stubs)
     tool_calls = [
@@ -287,7 +292,15 @@ async def collect() -> list[dict[str, Any]]:
             item
             async for item in processor._generate_and_stream(
                 "req-tools",
-                {"model": "glm-5.3", "nvext": {"extra_fields": ["stop_reason"]}},
+                {
+                    "model": "glm-5.3",
+                    "stream": True,
+                    "stream_options": {
+                        "include_usage": True,
+                        "continuous_usage_stats": continuous,
+                    },
+                    "nvext": {"extra_fields": ["stop_reason"]},
+                },
                 {},
                 list(range(10)),
                 Post(),
@@ -299,7 +312,11 @@ async def collect() -> list[dict[str, Any]]:
     data, terminal = [item["data"] for item in items]
     assert data["choices"] == [{**choice, "finish_reason": None}]
     assert terminal["choices"] == [{**choice, "delta": {}}]
-    assert "usage" not in data and "nvext" not in data
+    if continuous:
+        assert data["usage"] == terminal["usage"]
+        assert data["nvext"] == terminal["nvext"]
+    else:
+        assert "usage" not in data and "nvext" not in data
     assert terminal["usage"] == {
         **usage,
         "completion_tokens_details": {"reasoning_tokens": 1},
@@ -323,3 +340,198 @@ async def collect() -> list[dict[str, Any]]:
     assert len(metrics) == 1
     assert metrics[0]["chunk_tokens"] == metrics[0]["output_tokens"] == 3
     assert items[0]["event"] == "llm_metrics"
+
+
+class _ByteTokenizer:
+    def decode(self, ids, **kwargs):
+        return bytes(ids).decode("utf-8", errors="replace")
+
+
+def _usage_chunk(ids, count, *, finish=None, cached=4):
+    usage = {
+        "prompt_tokens": 10,
+        "completion_tokens": count,
+        "total_tokens": 10 + count,
+        "prompt_tokens_details": (
+            {"cached_tokens": cached} if cached is not None else None
+        ),
+    }
+    return {
+        "token_ids": ids,
+        "finish_reason": finish,
+        "completion_usage": usage,
+    }
+
+
+def _usage_processor(module, chunks, *, interval=1):
+    processor = module.SglangProcessor(
+        tokenizer=_ByteTokenizer(),
+        routed_engine=FakeRoutedEngine(items=chunks),
+        tool_call_parser_name=None,
+        reasoning_parser_name=None,
+        eos_token_ids=None,
+    )
+    processor.stream_interval = interval
+    post = module.SglangStreamingPostProcessor(
+        tokenizer=processor.tokenizer,
+        tool_call_parser=None,
+        reasoning_parser=None,
+    )
+    return processor, post
+
+
+def _usage_request(options=None, *, stream=True):
+    request = {"model": "test-model", "stream": stream}
+    if options is not None:
+        request["stream_options"] = options
+    return request
+
+
+_CONTINUOUS_USAGE = {"include_usage": True, "continuous_usage_stats": True}
+
+
+@pytest.mark.parametrize(
+    "options,expected_counts",
+    [
+        (None, [4]),
+        ({"include_usage": True}, [4]),
+        ({"continuous_usage_stats": True}, [4]),
+        (_CONTINUOUS_USAGE, [1, 2, 3, 4]),
+    ],
+)
+def test_usage_survives_buffered_utf8_without_changing_text(
+    module_stubs, options, expected_counts
+):
+    module = _load_processor_module(module_stubs)
+    chunks = [
+        _usage_chunk([byte], index, finish="stop" if index == 4 else None)
+        for index, byte in enumerate("你!".encode("utf-8"), start=1)
+    ]
+    chunks[0]["engine_data"] = {
+        "cached_tokens_details": {"device": 2, "host": 2}
+    }
+    processor, post = _usage_processor(module, chunks)
+
+    async def collect():
+        return [
+            item
+            async for item in processor._generate_and_stream(
+                "req-buffered", _usage_request(options), {}, list(range(10)), post
+            )
+        ]
+
+    events = asyncio.run(collect())
+    data = [event["data"] for event in events if "data" in event]
+    assert "".join(
+        choice["delta"].get("content", "")
+        for item in data
+        for choice in item["choices"]
+    ) == "你!"
+    usages = [item["usage"] for item in data if "usage" in item]
+    assert [usage["completion_tokens"] for usage in usages] == expected_counts
+    assert [usage["total_tokens"] for usage in usages] == [
+        10 + count for count in expected_counts
+    ]
+    assert all(usage["prompt_tokens_details"]["cached_tokens"] == 4 for usage in usages)
+    if options == _CONTINUOUS_USAGE:
+        assert data[0]["choices"] == data[1]["choices"] == []
+        assert data[0]["nvext"][module._INTERNAL_SGLEXT_KEY] == {
+            "cached_tokens_details": {"device": 2, "host": 2}
+        }
+    assert data[-1]["choices"][0]["finish_reason"] == "stop"
+
+
+def test_running_usage_preserves_batching_and_engine_terminal_count(module_stubs):
+    module = _load_processor_module(module_stubs)
+    chunks = [
+        _usage_chunk([byte], index)
+        for index, byte in enumerate(b"ABCDE", start=1)
+    ]
+    # The engine counted a hidden stop token; do not replace its authoritative
+    # completion count with the number of visible token IDs.
+    chunks.append(_usage_chunk([], 6, finish="stop"))
+    processor, post = _usage_processor(module, chunks, interval=3)
+
+    async def collect():
+        return [
+            item
+            async for item in processor._generate_and_stream(
+                "req-batched",
+                _usage_request(_CONTINUOUS_USAGE),
+                {},
+                list(range(10)),
+                post,
+            )
+        ]
+
+    events = asyncio.run(collect())
+    data = [event["data"] for event in events if "data" in event]
+    assert [
+        item["choices"][0]["delta"].get("content", "") for item in data
+    ] == ["A", "BCD", "E"]
+    assert [item["usage"]["completion_tokens"] for item in data] == [1, 4, 6]
+    assert [item["usage"]["total_tokens"] for item in data] == [11, 14, 16]
+    # Metrics describe token delivery, not a sum of cumulative usage snapshots.
+    assert sum(json.loads(event["comment"][0])["chunk_tokens"] for event in events) == 5
+
+
+@pytest.mark.parametrize("cached", [0, 4, None])
+def test_first_content_usage_remains_available_without_draining_stream(
+    module_stubs, cached
+):
+    module = _load_processor_module(module_stubs)
+    chunks = [
+        _usage_chunk([65], 1, cached=cached),
+        _usage_chunk([66], 2, finish="stop", cached=cached),
+    ]
+    processor, post = _usage_processor(module, chunks)
+
+    async def disconnect():
+        stream = processor._generate_and_stream(
+            "req-disconnect",
+            _usage_request(_CONTINUOUS_USAGE),
+            {},
+            list(range(10)),
+            post,
+        )
+        first = await anext(stream)
+        await stream.aclose()
+        return first["data"]
+
+    first = asyncio.run(disconnect())
+    assert first["choices"][0]["delta"]["content"] == "A"
+    assert first["choices"][0]["finish_reason"] is None
+    assert first["usage"] == {
+        "prompt_tokens": 10,
+        "completion_tokens": 1,
+        "total_tokens": 11,
+        "prompt_tokens_details": (
+            {"cached_tokens": cached} if cached is not None else None
+        ),
+    }
+
+
+def test_empty_token_report_emits_usage_without_waiting_for_finish(module_stubs):
+    module = _load_processor_module(module_stubs)
+    processor, post = _usage_processor(
+        module,
+        [_usage_chunk([], 1), _usage_chunk([65], 2, finish="stop")],
+        interval=8,
+    )
+
+    async def first_event():
+        stream = processor._generate_and_stream(
+            "req-no-visible-token",
+            _usage_request(_CONTINUOUS_USAGE),
+            {},
+            list(range(10)),
+            post,
+        )
+        first = await anext(stream)
+        await stream.aclose()
+        return first
+
+    event = asyncio.run(first_event())
+    assert event["data"]["choices"] == []
+    assert event["data"]["usage"]["completion_tokens"] == 1
+    assert json.loads(event["comment"][0])["chunk_tokens"] == 0