nemo_gym.core.failure_transport

Carry FailureReport data between Gym servers and into Collection’s rows.

Flow for one request a Gym server serves:

The server middleware opens a RequestFailureLog for the request.
While the request runs, observe_failure adds failures that a retry may still recover from (a model attempt), and observe_reply adds the report each reply from another Gym server forwarded.
When the request ends, the middleware merges the request’s own failure (if it raised) with the log and sends the result in the FailureHeader.
The caller’s raise_boundary_failure raises the rebuilt failure with separate HTTP details; raise_for_status keeps the HTTP exception compatibility path. Collection saves the report with FailureReportRow.

Module Contents

Classes

Name	Description
`FailureHeader`	Encode a `FailureReport` into one reply header and decode it on the other side.
`FailureReportRow`	The row keys under which Collection saves a `FailureReport` beside a rollout’s outcome.
`ModelOperationObservation`	A terminal HTTP retry-loop observation, not proof that every operation or response body was observed.
`RequestFailureLog`	The failures observed while one server request is served, other than the request’s own failure.
`StreamFailureReportReader`	Read bounded terminal SSE failure reports without changing the relayed response.

Functions

Name	Description
`_report_keeping_unknown_kinds`	`value` as a `FailureReport`. An event whose kind this Gym does not register (a server pinned to another
`observe_failure`	Record `error` for the request being served, for example a model attempt a retry may replace.
`observe_model_operation`	Record the client’s actual terminal outcome in the existing request observation scope.
`observe_reply`	Record the report a reply from another Gym server forwarded, success or failure.
`reply_failure_report`	`error`’s explicit cause chain for the reply beside its `EpisodeFailure`; `None` without events.
`report_for_current_request`	Project the explicit Gym cause chain, preserving its origin IDs.

Data

LOGGER

UNREGISTERED_KIND_PREFIX

API

class nemo_gym.core.failure_transport.FailureHeader()

Encode a FailureReport into one reply header and decode it on the other side.

aiohttp’s client rejects a reply whose header line (name and value) is over 8190 bytes, and the caller then sees an unreadable error instead of the real status and body. A report whose JSON is over MAX_BYTES is therefore not sent: the header says {"omitted": true}, and every hop above keeps that flag.

MAX_BYTES

int = 4096

NAME

str = 'X-NG-Failure-Report'

STREAM_FIELD

str = 'nemo_gym_failure_report'

nemo_gym.core.failure_transport.FailureHeader.decode(
    headers: collections.abc.Mapping[str, str] | None
) -> nemo_gym.core.failure_records.FailureReport | None

classmethod

The report in headers; None without one. An invalid value is reported as omitted.

nemo_gym.core.failure_transport.FailureHeader.decode_stream_event(
    payload: collections.abc.Mapping[str, typing.Any] | None
) -> nemo_gym.core.failure_records.FailureReport | None

classmethod

The report a terminal stream event carries; None without one. An invalid value is reported as omitted.

nemo_gym.core.failure_transport.FailureHeader.encode(
    report: nemo_gym.core.failure_records.FailureReport
) -> dict[str, str]

classmethod

The reply headers for report; none when it holds nothing.

nemo_gym.core.failure_transport.FailureHeader.stream_value(
    report: nemo_gym.core.failure_records.FailureReport
) -> dict[str, typing.Any] | None

classmethod

The STREAM_FIELD value for report; None when it holds nothing.

class nemo_gym.core.failure_transport.FailureReportRow()

The row keys under which Collection saves a FailureReport beside a rollout’s outcome.

EVENTS

str = '_ng_failure_events'

FAILURE_ID

str = '_ng_failure_id'

KEYS

tuple[str, ...] = (EVENTS, FAILURE_ID, OMITTED)

OMITTED

str = '_ng_failure_events_omitted'

nemo_gym.core.failure_transport.FailureReportRow.fields(
    report: nemo_gym.core.failure_records.FailureReport
) -> dict[str, typing.Any]

classmethod

report as row fields; none for an empty report.

nemo_gym.core.failure_transport.FailureReportRow.merged_into(
    row: dict[str, typing.Any],
    reports: nemo_gym.core.failure_records.FailureReport = ()
) -> dict[str, typing.Any]

classmethod

row with reports merged into its saved report; row’s own failure_id wins.

nemo_gym.core.failure_transport.FailureReportRow.report(
    row: collections.abc.Mapping[str, typing.Any]
) -> nemo_gym.core.failure_records.FailureReport

classmethod

The report saved in row; an invalid one is reported as omitted, never raised.

class nemo_gym.core.failure_transport.ModelOperationObservation()

Bases: BaseModel

A terminal HTTP retry-loop observation, not proof that every operation or response body was observed.

failure_id

str | None = None

final_attempt_id

str | None = None

http_attempts

int | None

model_config

operation_id

str = Field(min_length=1)

outcome

Literal['completed', 'failed']

request_elapsed_ms

float | None

retry_wait_s

float | None

transport_failures

int | None = Field(default=None, ge=0, strict=True)

class nemo_gym.core.failure_transport.RequestFailureLog(
    report: nemo_gym.core.failure_records.FailureReport = FailureReport(),
    model_operations: dict[str, nemo_gym.core.failure_transport.ModelOperationObservation] = dict(),
    dropped_events: int = 0
)

Dataclass

The failures observed while one server request is served, other than the request’s own failure.

The server middleware opens one per request with capture; code serving the request reaches it through current, so a retry loop deep in a client can record an attempt without threading it through calls.

A retry loop can fail without end, so the log keeps the first MAX_EVENTS events and counts the events of every later report in dropped_events; the report is then marked omitted. A report is kept or dropped whole, so every kept event’s causes stay in the log.

MAX_EVENTS

int = 64

_current

ContextVar[RequestFailureLog | None] = ContextVar('ng_request_failure_log', default=None)

dropped_events

int = 0

model_operations

dict[str, ModelOperationObservation] = field(default_factory=dict)

report

FailureReport = field(default_factory=FailureReport)

nemo_gym.core.failure_transport.RequestFailureLog.add(
    report: nemo_gym.core.failure_records.FailureReport
) -> None

Add report’s events as observations; the log never has a failure_id of its own.

nemo_gym.core.failure_transport.RequestFailureLog.capture() -> collections.abc.Iterator[nemo_gym.core.failure_transport.RequestFailureLog]

classmethod

Make a new log current for the request being served.

nemo_gym.core.failure_transport.RequestFailureLog.current() -> nemo_gym.core.failure_transport.RequestFailureLog | None

classmethod

The log of the request being served, or None outside a Gym server request.

nemo_gym.core.failure_transport.RequestFailureLog.reply_report(
    error: BaseException | None
) -> nemo_gym.core.failure_records.FailureReport

The report for this request’s reply: its own failure (error, None on success) and the log.

class nemo_gym.core.failure_transport.StreamFailureReportReader()

Read bounded terminal SSE failure reports without changing the relayed response.

_FIELD

bytes = FailureHeader.STREAM_FIELD.encode()

_KEY

Pattern[bytes]

_KEY_TAIL

int = len(_FIELD) + 12

_MAX_LINE_BYTES

int = 1 << 20

_TERMINAL_TYPES

frozenset[str]

_pending

= b''

report

FailureReport | None = None

nemo_gym.core.failure_transport.StreamFailureReportReader._merge(
    report: nemo_gym.core.failure_records.FailureReport
) -> None

nemo_gym.core.failure_transport.StreamFailureReportReader._read_line(
    line: bytes
) -> None

nemo_gym.core.failure_transport.StreamFailureReportReader.feed(
    chunk: bytes
) -> None

nemo_gym.core.failure_transport.StreamFailureReportReader.feed_text(
    text: str
) -> None

Read an existing capture without copying its entire body or accepting invalid Unicode.

nemo_gym.core.failure_transport.StreamFailureReportReader.finish() -> nemo_gym.core.failure_records.FailureReport | None

Close the body; an unfinished data line is not complete report evidence.

nemo_gym.core.failure_transport.StreamFailureReportReader.observe() -> None

Retain model-call observations without selecting an enclosing episode failure.

nemo_gym.core.failure_transport._report_keeping_unknown_kinds(
    value: typing.Any
) -> nemo_gym.core.failure_records.FailureReport

value as a FailureReport. An event whose kind this Gym does not register (a server pinned to another version) is kept under unregistered:<kind> instead of invalidating the whole report; any other invalid value still raises.

nemo_gym.core.failure_transport.observe_failure(
    error: BaseException
) -> None

Record error for the request being served, for example a model attempt a retry may replace.

nemo_gym.core.failure_transport.observe_model_operation(
    operation_id: str,
    outcome: typing.Literal['completed', 'failed'],
    final_attempt_id: str | None,
    failure_id: str | None = None,
    http_attempts: int | None = None,
    transport_failures: int | None = None,
    retry_wait_s: float | None = None,
    request_elapsed_ms: float | None = None
) -> None

Record the client’s actual terminal outcome in the existing request observation scope.

nemo_gym.core.failure_transport.observe_reply(
    headers: collections.abc.Mapping[str, str] | None
) -> None

Record the report a reply from another Gym server forwarded, success or failure.

nemo_gym.core.failure_transport.reply_failure_report(
    error: BaseException
) -> nemo_gym.core.failure_records.FailureReport | None

error’s explicit cause chain for the reply beside its EpisodeFailure; None without events.

The producer keeps its kind, reason, terminality and stage on the failure; a cause’s kind stays on its event.

nemo_gym.core.failure_transport.report_for_current_request(
    error: BaseException | None
) -> nemo_gym.core.failure_records.FailureReport

Project the explicit Gym cause chain, preserving its origin IDs.

nemo_gym.core.failure_transport.LOGGER = logging.getLogger(__name__)

nemo_gym.core.failure_transport.UNREGISTERED_KIND_PREFIX = 'unregistered:'

The namespace a forwarded event keeps when its kind is not registered here (a peer on another Gym version).