nemo_gym.rollouts.rollout_records

Index saved rollout outcomes for collection and offline aggregation.

Results and failure sidecars are the source of truth. Execution numbers are reserved in the run manifest; no dispatch or outcome event journal is written.

Module Contents

Classes

Name	Description
`RolloutRecord`	A persisted JSONL record, loaded only when its contents are requested.
`RolloutRecords`	-
`_IncompleteRow`	-
`_Outcome`	-
`_TopLevelProjector`	Copy one JSON object, replacing the values of some top-level keys with null, in bounded memory.

Functions

Name	Description
`_has_eligible_interruption_cause`	Whether the saved interruption names a cause resume may act on: a request never sent or refused at admission.
`_holds_saved_answer`	Use the same retained-generation facts as collection and recovery.
`_indexed_records`	Scan one record at a time, retaining its original byte coordinates.
`_read_row`	Read one JSONL row from the file position in chunks; a large row is parsed without its skipped values.
`coverage_path_for`	-
`is_interruption_failure`	Identify failures that spend neither the failure budget nor retry progress.
`is_rejudgment_failure_of`	Whether `row` records that re-judging the saved answer of judge failure `failure` failed again.
`is_rejudgment_of`	Whether `result` re-scores the saved answer of `failure` without a new model attempt.
`journal_path_for`	-
`logical_rollout_id`	-
`materialized_path_for`	-
`prepare_append`	Repair only an unterminated tail; preserve every complete history record.
`read_records`	Read committed JSONL records; an incomplete final write is not a record.
`rejudge_predecessor_id`	The id a re-judgment names its predecessor by: a failure’s `_ng_failure_id`, else, for a saved result whose
`rejudgment_admission_closed_by`	Whether outcome is the saved result of admission pending: operation, predecessor id and row hash, identity,
`resolve_rollout_owner`	Find a run’s main output when given its failure sidecar or saved inputs.
`resolve_rollout_path`	Resolve an output without losing its companions.

Data

COUNTS_AS_ZERO_ROW_KEY

PROJECTED_ROW_BYTES

REJUDGED_FAILURE_ID_KEY

REJUDGED_OBSERVATIONS_KEY

REJUDGE_OPERATION_ID_KEY

RESULT_PREDECESSOR_PREFIX

API

class nemo_gym.rollouts.rollout_records.RolloutRecord(
    path: pathlib.Path,
    offset: int,
    length: int,
    line_number: int = 0,
    legacy_attempt_index: int | None = None,
    file_identity: tuple[int, int] | None = None,
    expected_identity: tuple[str | None, str, int] | None = None,
    skip_keys: frozenset[str] = frozenset()
)

Dataclass

A persisted JSONL record, loaded only when its contents are requested.

Legacy migration may assign an attempt index without rewriting the source. Offsets and lengths are bytes, including any final newline.

expected_identity

tuple[str | None, str, int] | None = None

file_identity

tuple[int, int] | None = None

legacy_attempt_index

int | None = None

length

int

line_number

int = 0

offset

int

path

Path

skip_keys

frozenset[str] = frozenset()

nemo_gym.rollouts.rollout_records.RolloutRecord._read(
    file: typing.BinaryIO
) -> dict

nemo_gym.rollouts.rollout_records.RolloutRecord.read() -> dict

Read this record with its effective legacy attempt identity.

class nemo_gym.rollouts.rollout_records.RolloutRecords(
    manifest: nemo_gym.rollouts.rollout_recovery.RunManifest,
    expected: list[dict],
    classification: nemo_gym.core.failure_kinds.FailureClassificationConfig | None = None
)

attempt_cap

int

Every rollout’s attempt cap: the run’s saved cap plus the extension an explicit recovery recorded.

attempt_counts

Counter = Counter()

attempt_extension

= 0

expected

= {}

latest

dict[str, int]

omitted

set[tuple[str, int]] = set()

overridden_terminal_counts

Counter = Counter()

payloads

dict[tuple[str, int], _Outcome] = {}

replay_blocked

set[str] = set()

rerun_masked

frozenset[tuple[int, int, int]] = frozenset()

retry_terminal_timeouts

= manifest.retry_terminal_timeouts

unresolved_dispatches

set[str] = set()

nemo_gym.rollouts.rollout_records.RolloutRecords._final(
    identity: str
) -> bool

nemo_gym.rollouts.rollout_records.RolloutRecords._key(
    row: dict
) -> tuple[str, int]

nemo_gym.rollouts.rollout_records.RolloutRecords._payload(
    row: dict,
    record: nemo_gym.rollouts.rollout_records.RolloutRecord,
    legacy: bool = False,
    count_failure: bool = True
) -> None

nemo_gym.rollouts.rollout_records.RolloutRecords._retryable(
    identity: str
) -> bool

nemo_gym.rollouts.rollout_records.RolloutRecords.allocation_end_charges() -> list[dict]

The failure rows that charge a rollout for each later end of an allocation it was in flight in.

A reserved attempt with no saved outcome was in flight when the allocation that made it ended: the writer that loads the Store holds run_lock, so no other allocation of the run is alive. The first such end of an identity re-runs for free, and every later one spends an attempt of its cap: the earliest reservation without an outcome stays as it is and each later one gets a non-terminal rollout_allocation_ended failure. Only an identity a resume would run again is charged; a finished, settled or blocked one is never touched. The count is derived from the manifest’s reservations each time, so a saved charge is never charged again.

nemo_gym.rollouts.rollout_records.RolloutRecords.check_outcome(
    row: dict,
    legacy: bool = False
) -> tuple[str, int]

Validate without mutation, before a writer appends the payload.

nemo_gym.rollouts.rollout_records.RolloutRecords.check_run_identity(
    row: dict,
    legacy: bool = False
) -> None

nemo_gym.rollouts.rollout_records.RolloutRecords.coverage() -> dict

nemo_gym.rollouts.rollout_records.RolloutRecords.disposition(
    identity: str
) -> str

nemo_gym.rollouts.rollout_records.RolloutRecords.exhausted_count(
    max_attempts: int
) -> int

nemo_gym.rollouts.rollout_records.RolloutRecords.failure_count(
    identity: str
) -> int

nemo_gym.rollouts.rollout_records.RolloutRecords.load(
    output: pathlib.Path,
    manifest: nemo_gym.rollouts.rollout_recovery.RunManifest,
    import_legacy: bool = False,
    outcome_limits: dict[pathlib.Path, int] | None = None,
    skip_keys: frozenset[str] = frozenset()
) -> nemo_gym.rollouts.rollout_records.RolloutRecords

classmethod

Index outcomes, optionally restricted to prefixes captured before the manifest read.

skip_keys names top-level row values that no selection rule reads; large rows are parsed without them.

nemo_gym.rollouts.rollout_records.RolloutRecords.outcome(
    row: dict,
    record: nemo_gym.rollouts.rollout_records.RolloutRecord
) -> None

Index an outcome after its JSONL record has been flushed.

nemo_gym.rollouts.rollout_records.RolloutRecords.pending(
    max_attempts: int
) -> list[dict]

nemo_gym.rollouts.rollout_records.RolloutRecords.rerun_approved_identities() -> set[str]

The logical rollouts whose selected result is a masked one an operator approved to run again.

nemo_gym.rollouts.rollout_records.RolloutRecords.selected(
    disposition: str
) -> list[dict]

nemo_gym.rollouts.rollout_records.RolloutRecords.selected_records(
    disposition: str
) -> dict[tuple[str, int], nemo_gym.rollouts.rollout_records.RolloutRecord]

Select byte locations using attempt metadata, without loading payloads.

nemo_gym.rollouts.rollout_records.RolloutRecords.stopped_count(
    max_attempts: int
) -> int

Failed rollouts resume does not run again although they are neither final nor at their cap: a kept answer or a judge failure (left to verification recovery) and an unclassified execution (left to recovery).

class nemo_gym.rollouts.rollout_records._IncompleteRow(
    consumed: int
)

Exception

Bases: Exception

class nemo_gym.rollouts.rollout_records._Outcome(
    record: nemo_gym.rollouts.rollout_records.RolloutRecord,
    failure_class: str | None,
    legacy_failure_class: str | None,
    terminal: bool,
    masked: bool,
    has_reward: bool,
    elapsed_seconds: float | None,
    final_zero: bool = False,
    counted_zero_kind: str | None = None,
    interruption_cause_unknown: bool = False,
    saved_answer: bool = False,
    unclassified_execution_failure: bool = False,
    masked_infrastructure: bool = False,
    rerun_approved: bool = False
)

Dataclass

counted_zero_kind

str | None = None

elapsed_seconds

float | None

failure_class

str | None

final_zero

bool = False

has_reward

bool

interruption_cause_unknown

bool = False

legacy_failure_class

str | None

masked

bool

masked_infrastructure

bool = False

record

RolloutRecord

rerun_approved

bool = False

saved_answer

bool = False

terminal

bool

unclassified_execution_failure

bool = False

class nemo_gym.rollouts.rollout_records._TopLevelProjector(
    skip_keys: frozenset[str]
)

Copy one JSON object, replacing the values of some top-level keys with null, in bounded memory.

Bytes are fed in chunks; a skipped value is scanned for its end but never kept. Only top-level keys are matched, so the same name nested anywhere else is copied.

_depth

= 0

_key

bytearray | None = None

_last_key

bytes | None = None

_parts

list[bytes] = []

_skip

= {(key.encode()) for key in skip_keys}

nemo_gym.rollouts.rollout_records._TopLevelProjector.feed(
    chunk: bytes
) -> None

nemo_gym.rollouts.rollout_records._TopLevelProjector.result() -> bytes

nemo_gym.rollouts.rollout_records._has_eligible_interruption_cause(
    row: dict
) -> bool

Whether the saved interruption names a cause resume may act on: a request never sent or refused at admission.

nemo_gym.rollouts.rollout_records._holds_saved_answer(
    row: dict
) -> bool

Use the same retained-generation facts as collection and recovery.

nemo_gym.rollouts.rollout_records._indexed_records(
    path: pathlib.Path,
    end_offset: int | None = None,
    skip_keys: frozenset[str] = frozenset()
) -> collections.abc.Iterator[tuple[nemo_gym.rollouts.rollout_records.RolloutRecord, dict]]

Scan one record at a time, retaining its original byte coordinates.

With skip_keys a row larger than PROJECTED_ROW_BYTES is read in chunks and parsed without those top-level values, so a read-only selection never holds a whole large row.

nemo_gym.rollouts.rollout_records._read_row(
    file: typing.BinaryIO,
    limit: int | None,
    skip_keys: frozenset[str]
) -> tuple[dict | None, int]

Read one JSONL row from the file position in chunks; a large row is parsed without its skipped values.

Returns the row (None for a blank line, or for an incomplete final row) and the bytes consumed.

nemo_gym.rollouts.rollout_records.coverage_path_for(
    output: pathlib.Path
) -> pathlib.Path

nemo_gym.rollouts.rollout_records.is_interruption_failure(
    row: dict
) -> bool

Identify failures that spend neither the failure budget nor retry progress.

nemo_gym.rollouts.rollout_records.is_rejudgment_failure_of(
    failure: dict,
    row: dict
) -> bool

Whether row records that re-judging the saved answer of judge failure failure failed again.

nemo_gym.rollouts.rollout_records.is_rejudgment_of(
    failure: dict,
    result: dict
) -> bool

Whether result re-scores the saved answer of failure without a new model attempt.

failure is a judge failure, or a saved result whose grading is re-verified from its saved answer (a result masked by a judge or verify-transport failure, or one with no recorded grading). Which saved results are re-verified is the caller’s selection; this checks only that result binds the same saved answer and attempt.

nemo_gym.rollouts.rollout_records.journal_path_for(
    output: pathlib.Path
) -> pathlib.Path

nemo_gym.rollouts.rollout_records.logical_rollout_id(
    row: dict
) -> str

nemo_gym.rollouts.rollout_records.materialized_path_for(
    output: pathlib.Path
) -> pathlib.Path

nemo_gym.rollouts.rollout_records.prepare_append(
    path: pathlib.Path
) -> None

Repair only an unterminated tail; preserve every complete history record.

nemo_gym.rollouts.rollout_records.read_records(
    path: pathlib.Path
) -> collections.abc.Iterator[dict]

Read committed JSONL records; an incomplete final write is not a record.

nemo_gym.rollouts.rollout_records.rejudge_predecessor_id(
    row: dict
) -> str | None

The id a re-judgment names its predecessor by: a failure’s _ng_failure_id, else, for a saved result whose grading is re-verified from its saved answer, the digest of the whole saved row.

nemo_gym.rollouts.rollout_records.rejudgment_admission_closed_by(
    pending,
    outcome: dict,
    identity: str,
    failures_path: pathlib.Path,
    results_path: pathlib.Path | None = None
) -> bool

Whether outcome is the saved result of admission pending: operation, predecessor id and row hash, identity, attempt and admission time all match. Cold readers and the close use this one check.

A failure predecessor is found in failures_path; a saved-result predecessor in results_path.

nemo_gym.rollouts.rollout_records.resolve_rollout_owner(
    path: pathlib.Path
) -> pathlib.Path

Find a run’s main output when given its failure sidecar or saved inputs.

An output with its own history owns that history regardless of its name; strip the suffix only when the parent run has a manifest or journal.

nemo_gym.rollouts.rollout_records.resolve_rollout_path(
    output: pathlib.Path,
    read_only: bool = False
) -> pathlib.Path

Resolve an output without losing its companions.

Readers may retain an unambiguous legacy alias layout. Recovery writers require all history beside the resolved output, so aliases cannot split a run. Legacy-only reverification may reuse reader resolution after excluding history. Derived coverage reports do not establish ownership of a run.

nemo_gym.rollouts.rollout_records.COUNTS_AS_ZERO_ROW_KEY = '_ng_failure_counts_as_zero'

nemo_gym.rollouts.rollout_records.PROJECTED_ROW_BYTES = 64 << 20

nemo_gym.rollouts.rollout_records.REJUDGED_FAILURE_ID_KEY = '_ng_rejudged_failure_id'

nemo_gym.rollouts.rollout_records.REJUDGED_OBSERVATIONS_KEY = '_ng_rejudged_failure_observations'

nemo_gym.rollouts.rollout_records.REJUDGE_OPERATION_ID_KEY = '_ng_rejudge_operation_id'

nemo_gym.rollouts.rollout_records.RESULT_PREDECESSOR_PREFIX = 'result-sha256:'

nemo_gym.rollouts.rollout_records.RUN_ID_KEY = '_ng_run_id'

nemo_gym.rollouts.rollout_records._CHUNK_BYTES = 1 << 20

nemo_gym.rollouts.rollout_records._IN_STRING = re.compile(b'["\\\\]')

nemo_gym.rollouts.rollout_records._NESTED = re.compile(b'["{}\\[\\]]')

nemo_gym.rollouts.rollout_records._NESTED_RUN = re.compile(b'[^"\\\\\\[\\]{}]*(?:"[^"\\\\]*(?:\\\\.[^"\\\\]*)*"[^"\\\\\\[\\]{}]*...

nemo_gym.rollouts.rollout_records._NON_SPACE = re.compile(b'[^ \\t\\r\\n]')

nemo_gym.rollouts.rollout_records._STRUCTURE = re.compile(b'["{}\\[\\],:]')