nemo_gym.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 -> response.completed), splitting flattened function-call names back into namespace + name on the way out.

Module Contents

Functions

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

Data

LOG

NamespaceMap

_CLIENT_EXECUTED_TOOL_TYPES

_INPUT_ITEM_ADAPTER

_PARAM_FIELDS

_TOOL_ADAPTER

API

nemo_gym.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.

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

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

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

nemo_gym.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. The conversion carries it as a Chat Completions custom tool, which only a backend with free-form tool support accepts; a backend that expresses function tools only would refuse the whole request because of it.

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.

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

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

nemo_gym.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.

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

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

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

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

nemo_gym.responses_streaming.sanitize_streaming_responses_body(
    body: dict[str, typing.Any],
    drop_custom_tools: bool = False,
    drop_hosted_tools: bool = False
) -> tuple[dict[str, typing.Any], nemo_gym.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).

nemo_gym.responses_streaming.synthesize_responses_failure_sse(
    message: str,
    code: str = 'server_error'
) -> 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.

nemo_gym.responses_streaming.synthesize_responses_sse(
    response_json: dict[str, typing.Any],
    ns_map: typing.Optional[nemo_gym.responses_streaming.NamespaceMap] = None
) -> 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 and treat response.completed (which carries the response id and usage) as the terminal event, so those two are the required minimum; response.created is included for clients that wait for an acknowledgement before reading items.

nemo_gym.responses_streaming.validate_streaming_responses_params(
    body: dict[str, typing.Any]
) -> nemo_gym.openai_utils.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.

nemo_gym.responses_streaming.LOG = logging.getLogger(__name__)

nemo_gym.responses_streaming.NamespaceMap = dict[str, tuple[str, str]]

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

nemo_gym.responses_streaming._INPUT_ITEM_ADAPTER = TypeAdapter(NeMoGymResponseInputItem)

nemo_gym.responses_streaming._PARAM_FIELDS = frozenset(NeMoGymResponseCreateParamsNonStreaming.model_fields)

nemo_gym.responses_streaming._TOOL_ADAPTER = TypeAdapter(ToolParam)