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_valid`	Whether a tool spec validates against the params model’s tool union; failures are logged.
`flatten_namespace_tools`	Flatten `namespace` tool specs into plain function tools with joined names.
`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

NAMESPACE_TOOL_DELIMITER

NamespaceMap

_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_valid(
    tool: typing.Any
) -> bool

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

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

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.NAMESPACE_TOOL_DELIMITER = '__'

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

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)