nemo_gym.models.model_call_capture

Model call capture: the middleware that records each model call a model server handles, and the store it writes to.

Capture is opt-in and off by default. A pure-ASGI middleware records each correlated /v1/responses, /v1/chat/completions, and /v1/messages exchange. It records failed calls too, and writes each exchange to the rollout’s CaptureStore. It forwards bytes downstream unchanged, so it composes with streaming (SSE) responses. Capture is best-effort and never alters the response. A call is correlated by a /ng-rollout/<rollout_id>/v1/… base_url prefix, which the middleware strips before routing.

Module Contents

Classes

Name	Description
`CaptureStore`	Append-only, rollout-keyed JSONL sink for model exchanges.
`ModelCallCaptureConfig`	Run-wide model-call capture settings from Gym’s global config.
`ModelExecutionOutcome`	-
`_CaptureMiddleware`	Pure-ASGI per-rollout capture.

Functions

Name	Description
`_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`	-
`_exception_http_details`	-
`_fail_uncommitted_external_call`	Record a failure when an admitted worker call returns without commit coordinates.
`_headers_content_type`	-
`_is_chat_done`	-
`_parse_sse_events`	Parse an SSE byte stream into its JSON `data:` payloads (best-effort; non-JSON skipped).
`_pending_token`	A filename-safe, dot-free token; the dot is the separator `has_pending` globs on.
`_preserve_capture_prefix_on_redirect`	Keep rollout correlation on root-relative and same-origin redirects.
`_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.
`_stream_report`	The bounded `FailureHeader.STREAM_FIELD` value for a stream’s request log; `None` with nothing to report.
`_unique_request_header`	-
`_validate_rollout_id`	-
`_with_report_field`	-
`_with_stream_report`	`event` with the report added to its data object when it is a terminal Responses event or a Chat chunk.
`install_model_call_capture`	Install model-call capture middleware and return the capture ledger when staging is external.
`make_capture_store`	Build a CaptureStore when observability is enabled; otherwise None.
`note_finish_reason`	Note why a chat backend stopped generating with the call’s execution facts, which capture saves.
`start_model_execution`	Keep adapter-owned execution facts separate from the served response and capture settings.

Data

_CLIENT_SESSION_HEADER

_OBSERVED_PATHS

_PENDING_SUFFIX

_RESPONSES_TERMINAL_EVENTS

_ROLLOUT_PATH_RE

_SSE_KEEPALIVE

_TERMINAL_SSE_LINES

logger

API

class nemo_gym.models.model_call_capture.CaptureStore(
    root: str | pathlib.Path,
    create: bool = True
)

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

_root

= Path(root)

root

Path

nemo_gym.models.model_call_capture.CaptureStore.clear_all_pending(
    rollout_id: str
) -> None

nemo_gym.models.model_call_capture.CaptureStore.clear_pending(
    rollout_id: str,
    model_call_id: str
) -> None

nemo_gym.models.model_call_capture.CaptureStore.has_pending(
    rollout_id: str
) -> bool

True while any claimed write for this rollout has not finished.

The literal dot after the rollout id keeps 1-0 from matching 1-0-retry1.

nemo_gym.models.model_call_capture.CaptureStore.incomplete_path_for(
    rollout_id: str
) -> pathlib.Path

nemo_gym.models.model_call_capture.CaptureStore.is_incomplete(
    rollout_id: str
) -> bool

nemo_gym.models.model_call_capture.CaptureStore.mark_incomplete(
    rollout_id: str
) -> None

nemo_gym.models.model_call_capture.CaptureStore.mark_pending(
    rollout_id: str,
    model_call_id: str
) -> None

Claim a write before it is queued, so a reader can tell ‘not yet’ from ‘never’.

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

nemo_gym.models.model_call_capture.CaptureStore.pending_path_for(
    rollout_id: str,
    model_call_id: str
) -> pathlib.Path

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

nemo_gym.models.model_call_capture.CaptureStore.read_available(
    rollout_id: str
) -> tuple[list[tuple[int, dict[str, typing.Any]]], int]

Read valid exchanges without letting one damaged line hide the rest.

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

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

flock serializes appends to the same rollout across worker processes and threads while allowing independent rollouts to write concurrently. This does blocking file IO + fsync, so callers run it off the event loop (the capture middleware uses asyncio.to_thread).

class nemo_gym.models.model_call_capture.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.models.model_call_capture.ModelCallCaptureConfig.validate_capture_dir() -> nemo_gym.models.model_call_capture.ModelCallCaptureConfig

class nemo_gym.models.model_call_capture.ModelExecutionOutcome

Bases: typing.TypedDict

applied_max_tokens

NotRequired[int]

error_category

NotRequired[str]

finish_reason

NotRequired[str]

local_response_reason

str | None

output_budget_refit

NotRequired[dict[str, int]]

requested_max_tokens

NotRequired[int]

response_source

Literal['upstream', 'local'] | None

upstream_attempted

bool

upstream_status_code

int | None

class nemo_gym.models.model_call_capture._CaptureMiddleware(
    app: typing.Any,
    store: nemo_gym.models.model_call_capture.CaptureStore | None,
    model_server_name: str | None,
    token_store: typing.Any = None,
    configured_sink: typing.Any = None,
    lineage_store: nemo_gym.models.token_capture.protocols.LineageResolver | None = None,
    delta_records: bool = False,
    external_staging: bool = False,
    parent_resolution_mode: nemo_gym.models.token_capture.protocols.ParentResolutionMode = 'fingerprint',
    token_capture_enabled: bool = False,
    assistant_message_header: bytes | None = None,
    non_generating_requests: frozenset[tuple[str, str]] = frozenset()
)

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 it avoids buffering evaluation records, while still persisting external capture failures before terminal events are forwarded.

_assistant_message_header

bytes | None

_capture_ledger

CaptureLedger | None

_lineage_store

LineageResolver | None = lineage_store

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

async

nemo_gym.models.model_call_capture._classify_exception(
    exc: BaseException
) -> str

Normalized error_category for an exception raised while calling the model.

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

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

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

nemo_gym.models.model_call_capture._exception_http_details(
    exc: BaseException
) -> tuple[typing.Optional[int], bytes]

nemo_gym.models.model_call_capture._fail_uncommitted_external_call(
    context: nemo_gym.models.token_capture.sink.CaptureContext | None
) -> None

async

Record a failure when an admitted worker call returns without commit coordinates.

A call the engine refused is skipped: capture_engine_refusal already wrote its definite reason.

nemo_gym.models.model_call_capture._headers_content_type(
    headers: list
) -> bytes

nemo_gym.models.model_call_capture._is_chat_done(
    event: str | bytes
) -> bool

nemo_gym.models.model_call_capture._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.models.model_call_capture._pending_token(
    model_call_id: str
) -> str

A filename-safe, dot-free token; the dot is the separator has_pending globs on.

nemo_gym.models.model_call_capture._preserve_capture_prefix_on_redirect(
    message: dict[str, typing.Any],
    capture_prefix: str,
    request_headers: list[tuple[bytes, bytes]]
) -> dict[str, typing.Any]

Keep rollout correlation on root-relative and same-origin redirects.

nemo_gym.models.model_call_capture._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.models.model_call_capture._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.models.model_call_capture._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.models.model_call_capture._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.models.model_call_capture._record(
    store: nemo_gym.models.model_call_capture.CaptureStore,
    dialect: str,
    model_server_name: typing.Optional[str],
    request_bytes: bytes,
    verifier_invocation_id: typing.Optional[str] = None,
    client_session_id: typing.Optional[str] = None,
    client_assistant_message_id: typing.Optional[str] = None,
    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,
    response_raw: typing.Optional[str] = None,
    execution: typing.Optional[nemo_gym.models.model_call_capture.ModelExecutionOutcome] = None,
    failure: BaseException | None = None
) -> None

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

nemo_gym.models.model_call_capture._stream_report(
    failure_log: typing.Optional[nemo_gym.core.failure_transport.RequestFailureLog],
    error: BaseException | None
) -> typing.Optional[dict]

The bounded FailureHeader.STREAM_FIELD value for a stream’s request log; None with nothing to report.

nemo_gym.models.model_call_capture._unique_request_header(
    headers: list,
    name: bytes
) -> typing.Optional[str]

nemo_gym.models.model_call_capture._validate_rollout_id(
    rollout_id: str
) -> str

nemo_gym.models.model_call_capture._with_report_field(
    payload: dict[str, typing.Any],
    report: typing.Optional[dict]
) -> dict[str, typing.Any]

nemo_gym.models.model_call_capture._with_stream_report(
    event: str | bytes,
    report: typing.Optional[dict]
) -> str | bytes

event with the report added to its data object when it is a terminal Responses event or a Chat chunk.

nemo_gym.models.model_call_capture.install_model_call_capture(
    app: typing.Any,
    config: nemo_gym.models.model_call_capture.ModelCallCaptureConfig,
    model_server_name: str | None = None,
    assistant_message_header: bytes | None = None,
    global_config_dict: typing.Any = None,
    num_workers: int | None = None,
    non_generating_requests: frozenset[tuple[str, str]] = frozenset()
) -> nemo_gym.models.token_capture.protocols.CaptureLedger | None

Install model-call capture middleware and return the capture ledger when staging is external.

Always strip /ng-rollout/<id>/... before routing. Evaluation capture records requests and responses for that path. Non-terminal SSE chunks continue immediately. The terminal event follows the durable evaluation write. Training capture uses /ng-rollout/<id>/training-token-capture/.... That path provides a request-scoped token sink. The model server records token ids from its complete response. Consumers access records through TokenSource.freeze.

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

Build a CaptureStore when observability is enabled; otherwise None.

nemo_gym.models.model_call_capture.note_finish_reason(
    request: fastapi.Request,
    finish_reason: str | None
) -> None

Note why a chat backend stopped generating with the call’s execution facts, which capture saves.

A Responses payload converted from a chat completion cannot say it unless generation was cut short.

nemo_gym.models.model_call_capture.start_model_execution(
    request: fastapi.Request,
    upstream_attempted: bool
) -> nemo_gym.models.model_call_capture.ModelExecutionOutcome

Keep adapter-owned execution facts separate from the served response and capture settings.

nemo_gym.models.model_call_capture._CLIENT_SESSION_HEADER = b'x-session-id'

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

nemo_gym.models.model_call_capture._PENDING_SUFFIX = '.capture.pending'

nemo_gym.models.model_call_capture._RESPONSES_TERMINAL_EVENTS = frozenset({'response.completed', 'response.incomplete', 'response.failed'})

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

nemo_gym.models.model_call_capture._SSE_KEEPALIVE = SSE_KEEPALIVE

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

nemo_gym.models.model_call_capture.logger = logging.getLogger(__name__)