feat(renderers): native renderer layer + DeepSeek-V4 rollout & cumulative-mode support

rllm-model-gateway/src/rllm_model_gateway/models.py

@@ -123,6 +123,10 @@ class GatewayConfig(BaseModel):
     sync_traces: bool = False
     model: str | None = None  # When set, overrides ``body.model``
     cumulative_token_mode: bool = False
-    # renderers family for the cumulative-mode bridge. Check supported model families
-    # in MODEL_RENDERER_MAP of https://github.com/PrimeIntellect-ai/renderers/blob/main/renderers/base.py
+    # prime-rl renderer family for the gateway-built cumulative-mode renderer.
+    # "auto" matches the tokenizer's HF id against MODEL_RENDERER_MAP; set
+    # explicitly (e.g. "qwen3.5", "glm-5", "deepseek-v3") for local checkpoint
+    # paths. Models without a prime-rl renderer (e.g. DeepSeek-V4) are supported
+    # by injecting a renderer from the rllm side (in-process gateway). Families:
+    # https://github.com/PrimeIntellect-ai/renderers/blob/main/renderers/base.py
     renderer_family: str = "auto"

rllm-model-gateway/src/rllm_model_gateway/proxy.py

@@ -65,6 +65,44 @@ def _strip_logprobs(response: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _tool_call_to_openai(tc: Any, index: int) -> dict[str, Any]:
+    """Convert a renderer ToolCall to OpenAI ``tool_calls`` entry format.
+
+    Handles the shapes ``parse_response`` may return: an object with
+    ``name``/``arguments``, an object with ``function.name``/
+    ``function.arguments``, or a dict. ``arguments`` is JSON-encoded (OpenAI
+    sends it as a string).
+    """
+    fn = getattr(tc, "function", None)
+    if fn is not None:
+        name, arguments = getattr(fn, "name", ""), getattr(fn, "arguments", {})
+    elif isinstance(tc, dict):
+        name, arguments = tc.get("name", ""), tc.get("arguments", {})
+    else:
+        name, arguments = getattr(tc, "name", ""), getattr(tc, "arguments", {})
+    if not isinstance(arguments, str):
+        arguments = json.dumps(arguments)
+    return {"id": f"call_{index}", "type": "function", "function": {"name": name, "arguments": arguments}}
+
+
+def _parsed_response_to_message(parsed: Any) -> dict[str, Any]:
+    """Build an OpenAI assistant message from a renderer ``ParsedResponse``.
+
+    Matches the shape the chat-completions endpoint returns — clean ``content``,
+    plus ``reasoning_content`` and structured ``tool_calls`` when present — so
+    cumulative-mode (token-in) turns are consumed identically to the first
+    (chat-path) turn instead of dumping the raw completion text into ``content``.
+    """
+    message: dict[str, Any] = {"role": "assistant", "content": getattr(parsed, "content", "") or ""}
+    reasoning = getattr(parsed, "reasoning_content", "") or ""
+    if reasoning:
+        message["reasoning_content"] = reasoning
+    tool_calls = getattr(parsed, "tool_calls", None) or []
+    if tool_calls:
+        message["tool_calls"] = [_tool_call_to_openai(tc, i) for i, tc in enumerate(tool_calls)]
+    return message
+
+
 class ReverseProxy:
     """Forward requests to inference workers, capture traces.
 
@@ -154,9 +192,47 @@ async def handle(self, request: Request) -> Response:
             if acc.should_rewrite():
                 messages = request_body.get("messages", [])
                 if not acc.is_cumulative(messages):
-                    # Message history diverged — reset and fall through to
-                    # normal chat path (treated as fresh turn-0).
-                    acc.reset()
+                    kind, idx = acc.divergence(messages)
+                    # Duplicate / retried request (byte-identical to the already-
+                    # folded turn): the agent re-sent the same turn (its prior
+                    # attempt was rejected/failed). Re-sample that turn *in place*
+                    # from its existing prompt and overwrite its trace, instead of
+                    # resetting — so the cumulative chain isn't torn down and the
+                    # trajectory isn't split into two steps. Non-streaming only;
+                    # streaming retries fall through to reset below.
+                    if kind == "duplicate" and not is_stream and acc.prev_prompt_ids:
+                        # Report the prior attempt's outcome so the *cause* of the
+                        # re-send is visible: finish_reason=length → still length;
+                        # completion=0 → empty/failed response (deployment/infra);
+                        # finish_reason=stop with completion>0 → a valid response
+                        # the agent rejected (look agent-side, e.g. parse).
+                        logger.info(
+                            "Re-sampling duplicate request in place (session=%s, turn=%d, %d msgs); "
+                            "prior attempt: finish_reason=%s, completion=%d tok — chain preserved",
+                            session_id, acc.turn_count, acc.message_count,
+                            acc.last_finish_reason, acc.last_completion_len,
+                        )
+                        return await self._handle_cumulative_turn(
+                            request,
+                            request_body,
+                            session_id,
+                            acc,
+                            list(acc.prev_prompt_ids),
+                            originally_requested_logprobs,
+                            is_retry=True,
+                        )
+                    # Otherwise reset and fall through to the normal chat path
+                    # (fresh turn-0). Log *why* so resets can be root-caused.
+                    if kind == "changed":
+                        role = messages[idx].get("role", "?") if idx < len(messages) else "?"
+                        reason = f"prefix not append-only: msg {idx} (role={role}) changed vs verified prefix"
+                    elif kind == "shrunk":
+                        reason = f"history shrank to {idx} msgs (verified prefix was {acc.message_count}); summarization/unwind"
+                    elif kind == "duplicate":
+                        reason = f"duplicate request ({idx} msgs); resetting (streaming retry or no prior prompt to re-sample)"
+                    else:
+                        reason = f"non-cumulative message list (len={len(messages)} <= prefix {acc.message_count})"
+                    acc.reset(reason)
                 else:
                     new_messages = extract_new_messages(messages, acc.message_count)
                     token_ids = None
@@ -171,13 +247,23 @@ async def handle(self, request: Request) -> Response:
                             token_ids,
                             originally_requested_logprobs,
                         )
-                    # No new messages, or the renderer couldn't prove the
-                    # prefix-extension contract (e.g. DefaultRenderer, or an
-                    # assistant message in the new slice). Reset so this turn is
-                    # re-ingested as a fresh turn-0 on the chat path; otherwise
-                    # the stale prefix would drop this turn's completion tokens
-                    # from the next cumulative prompt and break prefix-extension.
-                    acc.reset()
+                    # No bridgeable new messages, or the renderer couldn't prove
+                    # the prefix-extension contract (DefaultRenderer, assistant in
+                    # the new slice, multimodal, a renderer error). Reset so this
+                    # turn is re-ingested as a fresh turn-0 on the chat path;
+                    # otherwise the stale prefix would drop this turn's completion
+                    # tokens from the next cumulative prompt and break extension.
+                    # Log which case it was + the new-slice shape to root-cause.
+                    if not new_messages:
+                        raw_roles = [m.get("role") for m in messages[acc.message_count:]]
+                        reason = f"no bridgeable new messages (raw new-slice roles={raw_roles})"
+                    else:
+                        reason = (
+                            "bridge returned None (new-slice roles="
+                            f"{[m.get('role') for m in new_messages]}, content_types="
+                            f"{[type(m.get('content')).__name__ for m in new_messages]})"
+                        )
+                    acc.reset(reason)
 
         if is_stream:
             return await self._handle_streaming(request, body, request_body, session_id, originally_requested_logprobs)
@@ -244,6 +330,10 @@ async def _handle_non_streaming(
                     if prompt_ids or completion_ids:
                         acc.ingest_turn(prompt_ids, completion_ids)
                         acc.update_prefix(request_body.get("messages", []))
+                        # Remember this turn's trace so a duplicate/retried turn-0
+                        # request can overwrite it instead of appending a new row.
+                        acc.last_trace_id = trace.trace_id
+                        acc.record_outcome((response_body.get("choices") or [{}])[0].get("finish_reason"), len(completion_ids))
 
         # Sanitise response
         needs_strip_vllm = self.strip_vllm
@@ -274,17 +364,23 @@ async def _handle_cumulative_turn(
         acc: TokenAccumulator,
         token_ids: list[int],
         originally_requested_logprobs: bool = False,
+        is_retry: bool = False,
     ) -> Response:
         """Rewrite chat/completions to /v1/completions with pre-tokenized prompt.
 
         ``token_ids`` is the full bridge-extended prompt for this turn, built
         by ``acc.build_next_prompt`` in ``handle()``.
 
+        ``is_retry``: this is a re-sample of an already-folded turn (a duplicate
+        request). The completion is swapped in place (``resample_completion``)
+        and its trace is overwritten, rather than folding a new turn — keeping
+        the cumulative chain intact. Retries are routed non-streaming.
+
         Respects the original stream setting: if the client requested streaming,
         we stream from vLLM and translate completions chunks to chat format in
         real-time.
         """
-        is_stream = request_body.get("stream", False)
+        is_stream = request_body.get("stream", False) and not is_retry
 
         # Construct completions request: forward everything except chat-specific fields
         completions_body = {k: v for k, v in request_body.items() if k not in ("messages", "stream", "stream_options", "tools", "tool_choice")}
@@ -301,6 +397,7 @@ async def _handle_cumulative_turn(
             acc,
             token_ids,
             originally_requested_logprobs,
+            is_retry=is_retry,
         )
 
     async def _handle_cumulative_non_streaming(
@@ -312,6 +409,7 @@ async def _handle_cumulative_non_streaming(
         acc: TokenAccumulator,
         token_ids: list[int],
         originally_requested_logprobs: bool = False,
+        is_retry: bool = False,
     ) -> Response:
         """Non-streaming cumulative turn: send pre-tokenized prompt, return JSON.
 
@@ -355,18 +453,45 @@ async def _handle_cumulative_non_streaming(
         prompt_token_ids = extract_prompt_token_ids(response_body) or token_ids
         completion_token_ids = extract_completion_token_ids(response_body)
 
-        acc.ingest_turn(prompt_token_ids, completion_token_ids)
-        acc.update_prefix(request_body.get("messages", []))
-
-        # Translate to chat format
+        if is_retry:
+            # Duplicate/retried turn: swap the completion in place (don't advance
+            # turn_count/message_count) so the cumulative chain stays intact.
+            acc.resample_completion(completion_token_ids)
+        else:
+            acc.ingest_turn(prompt_token_ids, completion_token_ids)
+            acc.update_prefix(request_body.get("messages", []))
+        # Record this turn's outcome so a later duplicate re-send can report why.
+        acc.record_outcome((response_body.get("choices") or [{}])[0].get("finish_reason"), len(completion_token_ids))
+
+        # Translate completions response → chat format. Parse the completion into
+        # the structured shape the chat endpoint returns (clean content +
+        # reasoning_content + tool_calls) via the renderer, so cumulative turns
+        # are consumed identically to the chat-path first turn. /v1/completions
+        # returns the model's raw text (e.g. "reasoning</think>answer" or raw
+        # tool-call markup for thinking/tool models); dumping that verbatim into
+        # `content` breaks the agent's action/tool parsing on every cumulative
+        # turn. Fall back to the raw text if the renderer can't parse it.
         choices = response_body.get("choices") or []
         if choices:
             first_choice = choices[0]
-            first_choice["message"] = {"role": "assistant", "content": first_choice.pop("text", "")}
+            raw_text = first_choice.pop("text", "")
+            message: dict[str, Any] | None = None
+            if self.renderer is not None and completion_token_ids:
+                try:
+                    parsed = self.renderer.parse_response(list(completion_token_ids))
+                    message = _parsed_response_to_message(parsed)
+                except Exception as err:  # noqa: BLE001 - never break the response on a parse error
+                    logger.warning("cumulative parse_response failed (%s); falling back to raw text", err)
+            first_choice["message"] = message if message is not None else {"role": "assistant", "content": raw_text}
         response_body["object"] = "chat.completion"
 
         if session_id and response_body:
             trace = build_trace_record(session_id, request_body, response_body, latency_ms, weight_version=request.state.weight_version)
+            if is_retry and acc.last_trace_id:
+                # Overwrite the rejected attempt's trace (stores upsert by
+                # trace_id), so the trajectory keeps one row for this turn.
+                trace.trace_id = acc.last_trace_id
+            acc.last_trace_id = trace.trace_id
             await self._persist(trace)
 
         sanitized = response_body