feat(adapters): request-rewriting interceptors

fern/versions/latest/pages/model-server/adapters-rewrites.mdx

@@ -0,0 +1,74 @@
+---
+title: "Request-Rewriting Interceptors"
+description: "Interceptors that mutate the outbound request body before it reaches the upstream."
+position: 6
+---
+
+Interceptors that mutate the outbound request body before it reaches the upstream. Run in the REQUEST stage.
+
+| Name | Stage | Purpose |
+|------|-------|---------|
+| `drop_params` | request | Remove named parameters from the outbound body. |
+| `payload_modifier` | request | Add, remove, and rename body fields. |
+| `system_message` | request | Inject a system message (prepend / append / replace). |
+| `consolidate_system` | request | Merge displaced system messages into one at position 0. |
+| `modify_tools` | request | Strip or add properties on `tools[].function.parameters`. |
+| `turn_counter` | request | Per-session turn budget; raises `GracefulError` (→ 429) on exhaustion. |
+
+### `drop_params`
+
+```yaml
+- name: drop_params
+  config:
+    params: ["top_k", "frequency_penalty"]
+```
+
+### `payload_modifier`
+
+```yaml
+- name: payload_modifier
+  config:
+    params_to_remove: ["secret_field"]
+    params_to_add: { "max_completion_tokens": 4096 }
+    params_to_rename: { "old_name": "new_name" }
+```
+
+### `system_message`
+
+```yaml
+- name: system_message
+  config:
+    system_message: "You are a careful assistant."
+    strategy: prepend   # one of: prepend | append | replace
+```
+
+### `consolidate_system`
+
+```yaml
+- name: consolidate_system
+  config:
+    separator: "\n\n"
+```
+
+### `modify_tools`
+
+```yaml
+- name: modify_tools
+  config:
+    strip_properties: ["internal_flag"]
+    add_properties:
+      reasoning:
+        type: string
+        description: "model's chain-of-thought"
+```
+
+### `turn_counter`
+
+```yaml
+- name: turn_counter
+  config:
+    every: 1            # log on every Nth turn
+    max_turns: 20       # null disables the budget
+```
+
+The session key is taken from `ctx.extra["session_id"]` (set by the middleware when the path matches `/s/<hex>/...`); otherwise a body-hash fallback is used.

nemo_gym/adapters/interceptors/consolidate_system.py

@@ -0,0 +1,100 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Merge all system messages into one at position 0.
+
+Models like Qwen3 reject system messages that appear mid-conversation
+(``System message must be at the beginning``).
+"""
+
+from __future__ import annotations
+
+import logging
+
+from nemo_gym.adapters.types import AdapterRequest, RequestInterceptor
+
+
+logger = logging.getLogger(__name__)
+
+
+def _content_to_str(content) -> str:
+    """Extract plain text from either a string or OpenAI list-format content."""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts = []
+        for item in content:
+            if isinstance(item, dict):
+                parts.append(item.get("text", ""))
+            elif isinstance(item, str):
+                parts.append(item)
+        return "\n".join(parts)
+    return str(content) if content else ""
+
+
+class Interceptor(RequestInterceptor):
+    def __init__(self, *, separator: str = "\n\n") -> None:
+        self._sep = separator
+        self._fix_count = 0
+
+    async def intercept_request(self, req: AdapterRequest) -> AdapterRequest:
+        messages: list[dict] = req.body.get("messages", [])
+        if not messages:
+            return req
+
+        system_indices: list[int] = []
+        system_parts: list[str] = []
+        non_system: list[dict] = []
+        for i, msg in enumerate(messages):
+            if msg.get("role") == "system":
+                system_indices.append(i)
+                text = _content_to_str(msg.get("content", ""))
+                if text:
+                    system_parts.append(text)
+            else:
+                non_system.append(msg)
+
+        if len(system_indices) <= 1 and (not system_indices or system_indices[0] == 0):
+            return req
+
+        self._fix_count += 1
+        session = req.ctx.extra.get("session_id", "?")
+
+        role_seq = [m.get("role", "?") for m in messages]
+        displaced = [i for i in system_indices if i > 0]
+        diag_parts: list[str] = []
+        for idx in displaced:
+            lo = max(0, idx - 2)
+            hi = min(len(role_seq), idx + 3)
+            window = " ".join(f"[{j}]={role_seq[j]}" for j in range(lo, hi))
+            content_preview = _content_to_str(messages[idx].get("content", ""))[:300]
+            diag_parts.append(f"system@{idx} window=({window}) content_preview={content_preview!r}")
+
+        logger.warning(
+            "consolidate_system: fix #%d session=%s n_msgs=%d system_at=%s n_system=%d roles_head=%s | %s",
+            self._fix_count,
+            session,
+            len(messages),
+            system_indices,
+            len(system_parts),
+            role_seq[:8],
+            " | ".join(diag_parts) if diag_parts else "system missing from pos 0",
+        )
+
+        merged: list[dict] = []
+        if system_parts:
+            merged.append({"role": "system", "content": self._sep.join(system_parts)})
+        merged.extend(non_system)
+        req.body["messages"] = merged
+        return req

nemo_gym/adapters/interceptors/drop_params.py

@@ -0,0 +1,38 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from __future__ import annotations
+
+import logging
+
+from nemo_gym.adapters.types import AdapterRequest, RequestInterceptor
+
+
+logger = logging.getLogger(__name__)
+
+
+class Interceptor(RequestInterceptor):
+    def __init__(self, *, params: list[str]) -> None:
+        self._params = set(params)
+        self._logged_once = False
+
+    async def intercept_request(self, req: AdapterRequest) -> AdapterRequest:
+        for p in self._params:
+            req.body.pop(p, None)
+
+        if not self._logged_once:
+            logger.info("drop_params: removing %s", sorted(self._params))
+            self._logged_once = True
+
+        return req

nemo_gym/adapters/interceptors/modify_tools.py

@@ -0,0 +1,82 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from __future__ import annotations
+
+import copy
+import logging
+from typing import Any
+
+from nemo_gym.adapters.types import AdapterRequest, RequestInterceptor
+
+
+logger = logging.getLogger(__name__)
+
+
+def _apply_modifications(
+    tools: list[dict[str, Any]],
+    strip: frozenset[str],
+    add: dict[str, dict[str, Any]],
+) -> int:
+    count = 0
+    for tool in tools:
+        fn = tool.get("function") or tool
+        params = fn.get("parameters") or {}
+        props = params.get("properties")
+        if not isinstance(props, dict):
+            continue
+
+        req: list[str] | None = params.get("required")
+
+        for field in strip:
+            if field in props:
+                del props[field]
+                count += 1
+            if isinstance(req, list) and field in req:
+                req.remove(field)
+
+        for field, schema in add.items():
+            if field not in props:
+                props[field] = copy.deepcopy(schema)
+                count += 1
+    return count
+
+
+class Interceptor(RequestInterceptor):
+    def __init__(
+        self,
+        *,
+        strip_properties: list[str] | None = None,
+        add_properties: dict[str, dict[str, Any]] | None = None,
+    ) -> None:
+        self._strip = frozenset(strip_properties or [])
+        self._add: dict[str, dict[str, Any]] = add_properties or {}
+        self._logged_once = False
+
+        if not self._strip and not self._add:
+            logger.warning("modify_tools: no modifications configured — interceptor is a no-op")
+
+    async def intercept_request(self, req: AdapterRequest) -> AdapterRequest:
+        tools = req.body.get("tools")
+        if tools:
+            n = _apply_modifications(tools, self._strip, self._add)
+            if n and not self._logged_once:
+                logger.info(
+                    "modify_tools: applied %d change(s) (strip=%s, add=%s)",
+                    n,
+                    sorted(self._strip),
+                    sorted(self._add),
+                )
+                self._logged_once = True
+        return req

nemo_gym/adapters/interceptors/payload_modifier.py

@@ -0,0 +1,74 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from nemo_gym.adapters.types import AdapterRequest, RequestInterceptor
+
+
+logger = logging.getLogger(__name__)
+
+
+def _remove_keys(obj: Any, keys: set[str]) -> Any:
+    if isinstance(obj, dict):
+        return {k: _remove_keys(v, keys) for k, v in obj.items() if k not in keys}
+    if isinstance(obj, list):
+        return [_remove_keys(item, keys) for item in obj]
+    return obj
+
+
+def _rename_keys(obj: Any, mapping: dict[str, str]) -> Any:
+    if isinstance(obj, dict):
+        return {mapping.get(k, k): _rename_keys(v, mapping) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_rename_keys(item, mapping) for item in obj]
+    return obj
+
+
+class Interceptor(RequestInterceptor):
+    def __init__(
+        self,
+        *,
+        params_to_remove: list[str] | None = None,
+        params_to_add: dict[str, Any] | None = None,
+        params_to_rename: dict[str, str] | None = None,
+    ) -> None:
+        self._remove = set(params_to_remove or [])
+        self._add: dict[str, Any] = params_to_add or {}
+        self._rename: dict[str, str] = params_to_rename or {}
+        self._logged_once = False
+
+    async def intercept_request(self, req: AdapterRequest) -> AdapterRequest:
+        if self._remove:
+            req.body = _remove_keys(req.body, self._remove)
+
+        if self._rename:
+            req.body = _rename_keys(req.body, self._rename)
+
+        if self._add:
+            req.body.update(self._add)
+
+        if not self._logged_once:
+            logger.info(
+                "payload_modifier: remove=%s, add=%s, rename=%s",
+                sorted(self._remove),
+                sorted(self._add),
+                sorted(self._rename),
+            )
+            self._logged_once = True
+
+        return req