nemo_gym.base_responses_api_model

Model server base classes and per-rollout model-call capture.

Every Gym model server derives from SimpleResponsesAPIModel, which wires the three model dialects (/v1/responses, /v1/chat/completions, /v1/messages) and installs the model-call capture middleware.

Capture is opt-in, off by default. A pure-ASGI middleware records correlated /v1/responses, /v1/chat/completions, and /v1/messages exchanges — including failed calls — into a per-rollout CaptureStore, forwarding bytes downstream unchanged so it composes with streaming (SSE) responses. Best-effort; never alters the response. Correlation is carried by a /ng-rollout/<rollout_id>/v1/… base_url prefix, which is stripped before routing.

Module Contents

Classes

Name	Description
`BaseResponsesAPIModel`	-
`BaseResponsesAPIModelConfig`	-
`CaptureStore`	Append-only, rollout-keyed JSONL sink for model exchanges.
`ModelCallCaptureConfig`	Run-wide model-call capture settings from Gym’s global config.
`ModelCallRecord`	Observability record derived from one captured model-server exchange.
`SimpleResponsesAPIModel`	-
`_CaptureMiddleware`	Pure-ASGI per-rollout capture.

Functions

Name	Description
`_as_arguments`	-
`_cache_signal`	Cache hit/miss + cached-token count, from usage cache fields (OpenAI / Anthropic).
`_classify_exception`	Normalized error_category for an exception raised while calling the model.
`_classify_status`	Normalized error_category from an HTTP status (None when < 400).
`_consume_terminal_sse_event`	-
`_headers_content_type`	-
`_parse_sse_events`	Parse an SSE byte stream into its JSON `data:` payloads (best-effort; non-JSON skipped).
`_reconstruct_anthropic_sse`	Rebuild a complete Anthropic Messages response from its streamed events.
`_reconstruct_chat_sse`	Rebuild a Chat Completions response from streamed chunks.
`_reconstruct_responses_sse`	Rebuild a Responses API response: the terminal envelope carries the full response object.
`_reconstruct_streamed_response`	Best-effort: reassemble a final response object from a streamed (SSE) body, by dialect.
`_record`	Append one exchange (success or failure). Best-effort: never raises.
`_store_for_rollout`	-
`_tool_calls_and_reasoning`	Structured tool calls (name, arguments, call_id) and reasoning text, across all three shapes.
`_validate_responses_params`	Validate a /v1/responses body dict, surfacing failures as FastAPI’s standard 422.
`_validate_rollout_id`	-
`aggregate_model_call_metrics`	Aggregate model-call metrics for one rollout id.
`aggregate_model_call_records`	Aggregate token and latency values from model-call records.
`build_model_call_record`	Map one captured exchange and its transport metadata into an observability record.
`clear_model_call_captures_for_rollouts`	Remove stale per-rollout capture files for these records before dispatch.
`extract_token_stats`	Normalize token totals across Responses, Chat Completions, and Anthropic Messages usage.
`install_model_call_capture`	Install model-call capture middleware.
`make_capture_store`	Build a CaptureStore when observability is enabled; otherwise None.
`maybe_rollout_id_from_run_body`	Per-rollout model-call capture id from a run-request’s task/rollout indices.
`merge_model_call_capture_into_record`	Attach captured model-call observability data to a rollout record in place.
`model_call_capture_dirs_from_config`	Return the single run-wide capture directory when capture is enabled.
`read_model_call_records`	Read captured exchanges in durable append order.

Data

_ANTHROPIC_CONVERTER

_OBSERVED_PATHS

_ROLLOUT_PATH_RE

_TERMINAL_SSE_LINES

logger

API

class nemo_gym.base_responses_api_model.BaseResponsesAPIModel()

Bases: BaseServer

config

BaseResponsesAPIModelConfig

class nemo_gym.base_responses_api_model.BaseResponsesAPIModelConfig()

Bases: BaseRunServerInstanceConfig

class nemo_gym.base_responses_api_model.CaptureStore(
    root: str | pathlib.Path
)

Append-only, rollout-keyed JSONL sink for model exchanges.

_lock

= threading.Lock()

_root

= Path(root)

root

Path

nemo_gym.base_responses_api_model.CaptureStore.path_for(
    rollout_id: str
) -> pathlib.Path

nemo_gym.base_responses_api_model.CaptureStore.read(
    rollout_id: str
) -> list[dict[str, typing.Any]]

nemo_gym.base_responses_api_model.CaptureStore.record(
    rollout_id: str,
    exchange: dict[str, typing.Any]
) -> None

Append one exchange and fsync (durable across a killed box).

flock serializes appends across worker processes (a model server may run with num_workers > 1, where the in-process lock can’t coordinate); the in-process lock serializes threads. This does blocking file IO + fsync, so callers run it off the event loop (the capture middleware offloads it via asyncio.to_thread).

class nemo_gym.base_responses_api_model.ModelCallCaptureConfig()

Bases: BaseModel

Run-wide model-call capture settings from Gym’s global config.

model_call_capture_dir

Optional[Path] = None

observability_enabled

bool = False

nemo_gym.base_responses_api_model.ModelCallCaptureConfig.validate_capture_dir() -> nemo_gym.base_responses_api_model.ModelCallCaptureConfig

class nemo_gym.base_responses_api_model.ModelCallRecord()

Bases: BaseModel

Observability record derived from one captured model-server exchange.

cache_creation_tokens

Optional[int] = None

cache_hit

Optional[bool] = None

cached_tokens

Optional[int] = None

call_index

int

completed_at

Optional[float] = None

dialect

Optional[str] = None

error_category

Optional[str] = None

latency_total_ms

Optional[float] = None

latency_ttft_ms

Optional[float] = None

model_call_id

Optional[str] = None

model_ref

Optional[ModelServerRef] = None

reasoning_content

Optional[str] = None

request

Optional[dict[str, Any]] = None

response

Optional[dict[str, Any]] = None

started_at

Optional[float] = None

status_code

Optional[int] = None

tokens_in

Optional[int] = None

tokens_out

Optional[int] = None

tokens_reasoning

Optional[int] = None

tokens_total

Optional[int] = None

tool_calls

list[dict[str, Any]] = Field(default_factory=list)

class nemo_gym.base_responses_api_model.SimpleResponsesAPIModel()

Bases: BaseResponsesAPIModel, SimpleServer

nemo_gym.base_responses_api_model.SimpleResponsesAPIModel._invoke_responses(
    request: fastapi.Request,
    params: nemo_gym.openai_utils.NeMoGymResponseCreateParamsNonStreaming
) -> nemo_gym.openai_utils.NeMoGymResponse

async

nemo_gym.base_responses_api_model.SimpleResponsesAPIModel.chat_completions(
    body: nemo_gym.openai_utils.NeMoGymChatCompletionCreateParamsNonStreaming = Body()
) -> nemo_gym.openai_utils.NeMoGymChatCompletion

asyncabstract

nemo_gym.base_responses_api_model.SimpleResponsesAPIModel.messages(
    request: fastapi.Request,
    body: dict = Body()
)

async

Default Anthropic Messages <-> Responses mapping shared by every Gym model server.

Translates the inbound Anthropic Messages request to the Responses API, delegates to this server’s own responses() (so it reuses whatever backend the server has), and maps the result back to an Anthropic Messages response. When the client requested stream: true (the Claude Code CLI always does), the complete response is re-emitted as a synthesized Anthropic SSE event stream. Servers may override this for native Messages handling.

nemo_gym.base_responses_api_model.SimpleResponsesAPIModel.responses(
    body: nemo_gym.openai_utils.NeMoGymResponseCreateParamsNonStreaming = Body()
) -> nemo_gym.openai_utils.NeMoGymResponse

asyncabstract

nemo_gym.base_responses_api_model.SimpleResponsesAPIModel.responses_dispatch(
    request: fastapi.Request,
    body: dict = Body()
)

async

Default /v1/responses entrypoint shared by every Gym model server.

A plain JSON request validates strictly against NeMoGymResponseCreateParamsNonStreaming and delegates to this server’s own responses(), preserving the historical non-streaming behavior. When the client requests stream: true (blackbox Responses-over-SSE harnesses like the Codex CLI always do), the request is first sanitized from the streaming wire dialect (extra bookkeeping fields, namespace tool specs — see nemo_gym.responses_streaming), delegated to the same responses(), and the complete response is re-emitted as a synthesized Responses SSE event stream. A responses() failure on this path is turned into a terminal response.failed event rather than an HTTP 500 (bad-request validation still fails eagerly, before the stream is committed).

nemo_gym.base_responses_api_model.SimpleResponsesAPIModel.setup_webserver() -> fastapi.FastAPI

class nemo_gym.base_responses_api_model._CaptureMiddleware(
    app: typing.Any,
    store: typing.Optional[nemo_gym.base_responses_api_model.CaptureStore],
    model_server_name: typing.Optional[str]
)

Pure-ASGI per-rollout capture.

Always strips an optional /ng-rollout/<id> path prefix before routing (used as the capture key) so the prefix is a stable routing feature independent of capture. When store is set it buffers the request body and a copy of the response while forwarding both downstream unchanged, so it composes with streaming (SSE) responses — it never consumes or rewraps the stream. SSE chunks are forwarded immediately except for the terminal event, which is released after the capture is durable. Every chunk is also buffered for post-hoc reassembly, so a very long stream is held in memory until it completes. When store is None (capture disabled) it strips the prefix and forwards only.

nemo_gym.base_responses_api_model._CaptureMiddleware.__call__(
    scope: dict[str, typing.Any],
    receive: typing.Any,
    send: typing.Any
) -> None

async

nemo_gym.base_responses_api_model._as_arguments(
    arguments: typing.Any
) -> dict[str, typing.Any]

nemo_gym.base_responses_api_model._cache_signal(
    usage: typing.Any
) -> tuple[typing.Optional[bool], typing.Optional[int]]

Cache hit/miss + cached-token count, from usage cache fields (OpenAI / Anthropic).

nemo_gym.base_responses_api_model._classify_exception(
    exc: BaseException
) -> str

Normalized error_category for an exception raised while calling the model.

nemo_gym.base_responses_api_model._classify_status(
    status_code: int
) -> typing.Optional[str]

Normalized error_category from an HTTP status (None when < 400).

nemo_gym.base_responses_api_model._consume_terminal_sse_event(
    buffer: bytearray,
    dialect: str
) -> typing.Optional[str]

nemo_gym.base_responses_api_model._headers_content_type(
    headers: list
) -> bytes

nemo_gym.base_responses_api_model._parse_sse_events(
    raw: bytes
) -> list[dict[str, typing.Any]]

Parse an SSE byte stream into its JSON data: payloads (best-effort; non-JSON skipped).

nemo_gym.base_responses_api_model._reconstruct_anthropic_sse(
    events: list[dict[str, typing.Any]]
) -> typing.Optional[dict[str, typing.Any]]

Rebuild a complete Anthropic Messages response from its streamed events.

nemo_gym.base_responses_api_model._reconstruct_chat_sse(
    events: list[dict[str, typing.Any]]
) -> typing.Optional[dict[str, typing.Any]]

Rebuild a Chat Completions response from streamed chunks.

nemo_gym.base_responses_api_model._reconstruct_responses_sse(
    events: list[dict[str, typing.Any]]
) -> typing.Optional[dict[str, typing.Any]]

Rebuild a Responses API response: the terminal envelope carries the full response object.

nemo_gym.base_responses_api_model._reconstruct_streamed_response(
    raw: bytes,
    dialect: str
) -> typing.Optional[dict[str, typing.Any]]

Best-effort: reassemble a final response object from a streamed (SSE) body, by dialect.

nemo_gym.base_responses_api_model._record(
    store: nemo_gym.base_responses_api_model.CaptureStore,
    dialect: str,
    model_server_name: typing.Optional[str],
    request_bytes: bytes,
    rollout_id: str,
    model_call_id: str,
    started_at: float,
    completed_at: float,
    response_body: typing.Any,
    status_code: typing.Optional[int],
    error_category: typing.Optional[str],
    latency_ms: float,
    ttft_ms: typing.Optional[float] = None
) -> None

Append one exchange (success or failure). Best-effort: never raises.

nemo_gym.base_responses_api_model._store_for_rollout(
    rollout_id: str,
    capture_dirs: list[pathlib.Path]
) -> typing.Optional[nemo_gym.base_responses_api_model.CaptureStore]

nemo_gym.base_responses_api_model._tool_calls_and_reasoning(
    response: dict[str, typing.Any]
) -> tuple[list[dict[str, typing.Any]], typing.Optional[str]]

Structured tool calls (name, arguments, call_id) and reasoning text, across all three shapes.

nemo_gym.base_responses_api_model._validate_responses_params(
    body: dict
) -> nemo_gym.openai_utils.NeMoGymResponseCreateParamsNonStreaming

Validate a /v1/responses body dict, surfacing failures as FastAPI’s standard 422.

nemo_gym.base_responses_api_model._validate_rollout_id(
    rollout_id: str
) -> str

nemo_gym.base_responses_api_model.aggregate_model_call_metrics(
    store: nemo_gym.base_responses_api_model.CaptureStore,
    rollout_id: str
) -> dict[str, typing.Any]

Aggregate model-call metrics for one rollout id.

nemo_gym.base_responses_api_model.aggregate_model_call_records(
    calls: list[nemo_gym.base_responses_api_model.ModelCallRecord]
) -> dict[str, typing.Any]

Aggregate token and latency values from model-call records.

nemo_gym.base_responses_api_model.build_model_call_record(
    exchange: dict[str, typing.Any],
    call_index: int
) -> nemo_gym.base_responses_api_model.ModelCallRecord

Map one captured exchange and its transport metadata into an observability record.

nemo_gym.base_responses_api_model.clear_model_call_captures_for_rollouts(
    records: list[typing.Any],
    capture_dirs: list[pathlib.Path]
) -> None

Remove stale per-rollout capture files for these records before dispatch.

Capture files are keyed by a deterministic rollout id (task-rollout-attempt), so without this a fresh run or a kill-shaped retry would append onto the previous attempt’s capture for the same id. The caller passes only rows about to be dispatched, after assigning any retry suffix.

nemo_gym.base_responses_api_model.extract_token_stats(
    usage: typing.Any
) -> dict[str, typing.Optional[int]]

Normalize token totals across Responses, Chat Completions, and Anthropic Messages usage.

For native Anthropic /v1/messages with prompt caching, input_tokens is only the uncached remainder, so cache-read + cache-creation tokens are folded into tokens_in to reflect the true prompt size (and cache-creation is surfaced separately as cache_creation_tokens). OpenAI / Responses usage already includes cached tokens in input_tokens / prompt_tokens (where cached_tokens is a subset), so it is left untouched — no double counting.

tokens_in is a prompt-size metric, not a cost proxy: providers price cache-read (~0.1x) and cache-creation (~1.25x) differently from base input, so cost-accurate consumers should weight cached_tokens and cache_creation_tokens separately rather than summing tokens_in.

nemo_gym.base_responses_api_model.install_model_call_capture(
    app: typing.Any,
    config: nemo_gym.base_responses_api_model.ModelCallCaptureConfig,
    model_server_name: typing.Optional[str] = None
) -> None

Install model-call capture middleware.

Always installed so the /ng-rollout/<id> correlation prefix is stripped before routing regardless of whether capture is enabled (otherwise a default gym eval would 404 on every prefixed model call). When capture is enabled the middleware additionally records each observed call’s request + response into a rollout-keyed CaptureStore while forwarding bytes downstream unchanged (non-terminal SSE chunks are forwarded as they arrive; the terminal event follows the durable capture write).

nemo_gym.base_responses_api_model.make_capture_store(
    config: nemo_gym.base_responses_api_model.ModelCallCaptureConfig
) -> typing.Optional[nemo_gym.base_responses_api_model.CaptureStore]

Build a CaptureStore when observability is enabled; otherwise None.

nemo_gym.base_responses_api_model.maybe_rollout_id_from_run_body(
    body: pydantic.BaseModel | typing.Mapping[str, typing.Any] | None
) -> typing.Optional[str]

Per-rollout model-call capture id from a run-request’s task/rollout indices.

Reads the canonical row keys (_ng_task_index / _ng_rollout_index) that rollout_collection ships to an agent’s /run. When a resume re-dispatch attempt is present (_ng_attempt_index > 0), an -a<n> suffix is appended so a retry’s captured model calls stay separable from the prior attempt; the first attempt (0) keeps the bare <task>-<rollout> key for backward compatibility.

nemo_gym.base_responses_api_model.merge_model_call_capture_into_record(
    record: dict[str, typing.Any],
    capture_dirs: list[pathlib.Path],
    include_payloads: bool = False
) -> dict[str, typing.Any]

Attach captured model-call observability data to a rollout record in place.

Keyed by the rollout id derived from the record’s task/rollout/attempt indices, so the attached shape is identical for every agent harness. Adds ng_model_call_capture = {rollout_id, metrics, calls} where calls are derived observability records. Raw request and response payloads remain in the capture store unless include_payloads is true. No-op when no capture exists. The harness output and reward are not modified.

nemo_gym.base_responses_api_model.model_call_capture_dirs_from_config(
    global_config_dict: typing.Any
) -> list[pathlib.Path]

Return the single run-wide capture directory when capture is enabled.

nemo_gym.base_responses_api_model.read_model_call_records(
    store: nemo_gym.base_responses_api_model.CaptureStore,
    rollout_id: str
) -> list[nemo_gym.base_responses_api_model.ModelCallRecord]

Read captured exchanges in durable append order.

nemo_gym.base_responses_api_model._ANTHROPIC_CONVERTER = AnthropicConverter()

nemo_gym.base_responses_api_model._OBSERVED_PATHS = {'/v1/responses': 'responses', '/v1/chat/completions': 'chat', '/v1/messages': '...

nemo_gym.base_responses_api_model._ROLLOUT_PATH_RE = re.compile(f'^/{re.escape(ROLLOUT_PATH_PREFIX)}/(?P<rollout_id>[^/]+)(?P<rest>/....

nemo_gym.base_responses_api_model._TERMINAL_SSE_LINES: dict[str, dict[bytes, str]] = {'responses': {b'event: response.completed': 'complete', b'event: response.incom...

nemo_gym.base_responses_api_model.logger = logging.getLogger(__name__)