> This page is for version 0.2.1.
> For other versions, use one of these documentation indexes:
> - Main (default): https://docs.nvidia.com/nemo/gym/main/llms.txt
> - 0.7.0: https://docs.nvidia.com/nemo/gym/v0.7.0/llms.txt
> - 0.6.0: https://docs.nvidia.com/nemo/gym/v0.6.0/llms.txt
> - 0.5.1: https://docs.nvidia.com/nemo/gym/v0.5.1/llms.txt
> - 0.5.0: https://docs.nvidia.com/nemo/gym/v0.5.0/llms.txt
> - 0.4.0: https://docs.nvidia.com/nemo/gym/v0.4.0/llms.txt
> - 0.3.0: https://docs.nvidia.com/nemo/gym/v0.3.0/llms.txt
> - 0.2.1: https://docs.nvidia.com/nemo/gym/v0.2.1/llms.txt

> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/gym/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/gym/_mcp/server.

# nemo_gym.models.responses_streaming

Streaming Responses-dialect support shared by every Gym model server.

Blackbox harnesses that speak the OpenAI Responses API over SSE (e.g. the Codex CLI) send
requests the strict `NeMoGymResponseCreateParamsNonStreaming` model would reject: a
`stream: true` flag, client-bookkeeping fields (`client_metadata`, `prompt_cache_key`),
and `namespace` tool specs — functions grouped under a namespace that the model calls back
with separate `namespace` + `name` fields on the `function_call` item. Backends behind a
Gym model server (chat-completions endpoints, vLLM) only understand flat function tools, so
this module provides:

* the request-side sanitizer that flattens namespace tools into plain functions (joined as
  `<namespace>__<name>`), rewrites replayed namespaced calls in the input history the same
  way, and drops the fields the params model does not know;
* the response-side synthesizer that re-emits a complete `NeMoGymResponse` as the minimal
  Responses SSE event sequence streaming clients require (`response.created` ->
  `response.output_item.done` per output item -> the matching completed/incomplete/failed
  terminal event), splitting flattened function-call names back into `namespace` + `name`;
* the keepalive events (`response.created` then repeated `response.in_progress`) that hold
  the stream open while a long backend call runs, when the server opts in.

## Module Contents

### Functions

| Name                                                                                                              | Description                                                                                  |
| ----------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------- |
| [`_delete_loc`](#nemo_gym-models-responses_streaming-_delete_loc)                                                 | Delete the value at a pydantic error loc from a nested dict/list structure.                  |
| [`_input_message_text`](#nemo_gym-models-responses_streaming-_input_message_text)                                 | -                                                                                            |
| [`_is_system_like_message`](#nemo_gym-models-responses_streaming-_is_system_like_message)                         | -                                                                                            |
| [`_sse_event`](#nemo_gym-models-responses_streaming-_sse_event)                                                   | -                                                                                            |
| [`_tool_kept`](#nemo_gym-models-responses_streaming-_tool_kept)                                                   | Whether a tool spec stays in the request; a dropped one is logged with the reason.           |
| [`_tool_valid`](#nemo_gym-models-responses_streaming-_tool_valid)                                                 | Whether a tool spec validates against the params model's tool union; failures are logged.    |
| [`_wire_usage`](#nemo_gym-models-responses_streaming-_wire_usage)                                                 | Return the usage object with integer detail counts, as the Responses wire schema types them. |
| [`flatten_namespace_tools`](#nemo_gym-models-responses_streaming-flatten_namespace_tools)                         | Flatten `namespace` tool specs into plain tools with joined names.                           |
| [`is_responses_lifecycle_chunk`](#nemo_gym-models-responses_streaming-is_responses_lifecycle_chunk)               | Whether `chunk` is exactly one `response.created` or `response.in_progress` SSE event.       |
| [`restore_namespace_tool_calls`](#nemo_gym-models-responses_streaming-restore_namespace_tool_calls)               | Restore the client-visible tool names before capture finalization and SSE output.            |
| [`sanitize_streaming_responses_body`](#nemo_gym-models-responses_streaming-sanitize_streaming_responses_body)     | Map a streaming-dialect request body onto the strict non-streaming params shape.             |
| [`synthesize_responses_failure_sse`](#nemo_gym-models-responses_streaming-synthesize_responses_failure_sse)       | Emit a terminal Responses SSE stream for a backend failure.                                  |
| [`synthesize_responses_keepalive_sse`](#nemo_gym-models-responses_streaming-synthesize_responses_keepalive_sse)   | Emit Responses SSE events that hold a stream open while the backend computes.                |
| [`synthesize_responses_sse`](#nemo_gym-models-responses_streaming-synthesize_responses_sse)                       | Re-emit a complete Responses API response object as an SSE event stream.                     |
| [`validate_streaming_responses_params`](#nemo_gym-models-responses_streaming-validate_streaming_responses_params) | Validate a sanitized streaming-dialect body, pruning fields newer than the params model.     |

### Data

[`LOG`](#nemo_gym-models-responses_streaming-LOG)

[`NamespaceMap`](#nemo_gym-models-responses_streaming-NamespaceMap)

[`_CLIENT_EXECUTED_TOOL_TYPES`](#nemo_gym-models-responses_streaming-_CLIENT_EXECUTED_TOOL_TYPES)

[`_INPUT_ITEM_ADAPTER`](#nemo_gym-models-responses_streaming-_INPUT_ITEM_ADAPTER)

[`_PARAM_FIELDS`](#nemo_gym-models-responses_streaming-_PARAM_FIELDS)

[`_RESPONSES_LIFECYCLE_SSE_PREFIXES`](#nemo_gym-models-responses_streaming-_RESPONSES_LIFECYCLE_SSE_PREFIXES)

[`_TOOL_ADAPTER`](#nemo_gym-models-responses_streaming-_TOOL_ADAPTER)

### API

```python
nemo_gym.models.responses_streaming._delete_loc(
    body: typing.Any,
    loc: tuple
) -> bool
```

Delete the value at a pydantic error loc from a nested dict/list structure.

Returns False when the loc cannot be walked literally (e.g. it contains a union-arm label
rather than a real key), in which case nothing is deleted.

```python
nemo_gym.models.responses_streaming._input_message_text(
    item: dict[str, typing.Any]
) -> str
```

```python
nemo_gym.models.responses_streaming._is_system_like_message(
    item: typing.Any
) -> bool
```

```python
nemo_gym.models.responses_streaming._sse_event(
    payload: dict[str, typing.Any]
) -> str
```

```python
nemo_gym.models.responses_streaming._tool_kept(
    tool: typing.Any,
    drop_custom_tools: bool,
    drop_hosted_tools: bool
) -> bool
```

Whether a tool spec stays in the request; a dropped one is logged with the reason.

`drop_custom_tools` drops a Responses `custom` (free-form) tool. A kept one reaches the
conversion, which under `custom_tool_strategy=passthrough` carries it as a Chat Completions
custom tool — only a backend with free-form tool support accepts the request, and one that
expresses function tools only refuses it whole — and under `function_bridge` lowers it to a
function tool the backend serves.

`drop_hosted_tools` drops every tool type other than `function` and `custom`. A backend
that serves no hosted tools cannot honor such a spec, and offering it to the model produces
calls nothing answers.

```python
nemo_gym.models.responses_streaming._tool_valid(
    tool: typing.Any
) -> bool
```

Whether a tool spec validates against the params model's tool union; failures are logged.

```python
nemo_gym.models.responses_streaming._wire_usage(
    usage: typing.Any
) -> typing.Any
```

Return the usage object with integer detail counts, as the Responses wire schema types them.

Gym keeps an unreported cache or reasoning breakdown as `None` internally to tell unknown
from zero. On the wire those counts are integers, and a strict streaming client (the Codex
CLI) that meets `null` fails to parse `response.completed`, treats the stream as
disconnected, and re-sends the whole request.

```python
nemo_gym.models.responses_streaming.flatten_namespace_tools(
    tools: typing.Any
) -> tuple[list[typing.Any], nemo_gym.models.responses_streaming.NamespaceMap]
```

Flatten `namespace` tool specs into plain tools with joined names.

A sub-tool keeps its own type when it declares one (codex's code-mode
namespace carries a `custom` free-form tool beside function tools);
an untyped sub-tool is a function tool, the namespace spec's only
pre-custom member kind.

```python
nemo_gym.models.responses_streaming.is_responses_lifecycle_chunk(
    chunk: bytes
) -> bool
```

Whether `chunk` is exactly one `response.created` or `response.in_progress` SSE event.

Those events carry no model output. With `stream_keepalive_s` they open and hold a stream
long before generation finishes, so like SSE comments they are liveness, not the start of the
answer. A chunk that also carries another event is output.

```python
nemo_gym.models.responses_streaming.restore_namespace_tool_calls(
    items: list[dict],
    ns_map: nemo_gym.models.responses_streaming.NamespaceMap
) -> list[dict]
```

Restore the client-visible tool names before capture finalization and SSE output.

```python
nemo_gym.models.responses_streaming.sanitize_streaming_responses_body(
    body: dict[str, typing.Any],
    drop_custom_tools: bool = False,
    drop_hosted_tools: bool = False,
    hoist_carrier_custom_tools: bool = False
) -> tuple[dict[str, typing.Any], nemo_gym.models.responses_streaming.NamespaceMap]
```

Map a streaming-dialect request body onto the strict non-streaming params shape.

Returns the cleaned body dict (ready for `NeMoGymResponseCreateParamsNonStreaming`
validation) and the namespace map needed to restore namespaced call names in the
synthesized SSE response. Tool entries that still fail per-entry validation after
flattening are dropped with a warning rather than failing the whole request, since a
harness's exotic tool is better lost than the rollout. `drop_custom_tools` and
`drop_hosted_tools` drop valid tool specs the backend cannot honor (see `_tool_kept`).
`hoist_carrier_custom_tools` hoists `custom` (free-form) entries from an
`additional_tools` carrier item into `tools` instead of dropping them; custom entries
already in `tools` are kept regardless. The serving side enables it when a bridge lowers
custom tools for the backend (`custom_tool_strategy=function_bridge`).

```python
nemo_gym.models.responses_streaming.synthesize_responses_failure_sse(
    message: str,
    code: typing.Optional[str] = 'server_error',
    response_id: typing.Optional[str] = None,
    include_created: bool = True
) -> typing.Iterator[str]
```

Emit a terminal Responses SSE stream for a backend failure.

Once the streaming contract is committed (HTTP 200 + `text/event-stream`), a `responses()`
error can no longer surface as an HTTP 500. Streaming clients (e.g. Codex) expect a terminal
`response.failed` event; without one they see a truncated stream and cannot tell an
application error from a transport failure. Emitting `response.failed` lets the client report
a clean turn failure, and lets model-call capture classify it as an upstream error (its
terminal-SSE table maps `response.failed` to an error) rather than a stream truncation.
A stream that keepalive events already opened passes their `response_id` and
`include_created=False`, so it ends with the same response it announced.

```python
nemo_gym.models.responses_streaming.synthesize_responses_keepalive_sse(
    response_id: str,
    include_created: bool
) -> typing.Iterator[str]
```

Emit Responses SSE events that hold a stream open while the backend computes.

Gym's Responses streaming is buffered: the item events exist only once the whole generation
has returned. A client that bounds the silence between events (the Codex CLI's
`stream_idle_timeout_ms`, 300 s by default) abandons a long turn and re-sends it, and SSE
comments do not reset that timer because the client's event parser drops them. The first
call opens the stream with `response.created`; every call emits `response.in_progress`.
Both name `response_id`, the id of the response the stream ends with, and carry no output,
so clients that track only items and the terminal event ignore them.

```python
nemo_gym.models.responses_streaming.synthesize_responses_sse(
    response_json: dict[str, typing.Any],
    ns_map: typing.Optional[nemo_gym.models.responses_streaming.NamespaceMap] = None,
    include_created: bool = True
) -> typing.Iterator[str]
```

Re-emit a complete Responses API response object as an SSE event stream.

Streaming clients build their view of the turn from `response.output_item.done` events.
The terminal event reflects the response status, retaining partial output, usage and failure
details; `response.created` is included for clients waiting for an acknowledgement, unless
`include_created` is false because keepalive events already opened the stream.

```python
nemo_gym.models.responses_streaming.validate_streaming_responses_params(
    body: dict[str, typing.Any]
) -> nemo_gym.core.responses_api_types.NeMoGymResponseCreateParamsNonStreaming
```

Validate a sanitized streaming-dialect body, pruning fields newer than the params model.

Harness wire formats evolve faster than the pinned OpenAI SDK types (e.g. Codex sending
`reasoning.context`, which the SDK's `Reasoning` model forbids). Any field pydantic flags
as `extra_forbidden` is removed and validation retried, so only errors that cannot be fixed
by dropping an unknown field surface to the client.

```python
nemo_gym.models.responses_streaming.LOG = logging.getLogger(__name__)
```

```python
nemo_gym.models.responses_streaming.NamespaceMap = dict[str, tuple[str, str]]
```

```python
nemo_gym.models.responses_streaming._CLIENT_EXECUTED_TOOL_TYPES = frozenset({'function', 'custom'})
```

```python
nemo_gym.models.responses_streaming._INPUT_ITEM_ADAPTER = TypeAdapter(NeMoGymResponseInputItem)
```

```python
nemo_gym.models.responses_streaming._PARAM_FIELDS = frozenset(NeMoGymResponseCreateParamsNonStreaming.model_fields)
```

```python
nemo_gym.models.responses_streaming._RESPONSES_LIFECYCLE_SSE_PREFIXES = (b'event: response.created\n', b'event: response.in_progress\n')
```

```python
nemo_gym.models.responses_streaming._TOOL_ADAPTER = TypeAdapter(ToolParam)
```