nemo_gym.rollouts.rollout_store

Persist evaluation outcomes and reserve execution identities for its controller.

The controller holds run_lock across preparation, optional checkpoint restoration, collection and reporting. Results and sidecars contain outcomes; the manifest owns execution numbers. There is no dispatch journal or second outcome commit.

Module Contents

Classes

Name	Description
`RolloutStore`	Saved outcomes shared by the evaluation controller and offline readers.

Functions

Name	Description
`_record_output_write`	Measure the persisted outcome bytes, including their JSONL newline.
`raw_outcomes_are_selected`	Whether an existing row-oriented reader can read this run without filtering.

Data

_REJUDGE_DROPPED_PREFIXES

_REJUDGE_GRADING_KEYS

logger

API

class nemo_gym.rollouts.rollout_store.RolloutStore(
    output: pathlib.Path,
    state: nemo_gym.rollouts.rollout_records.RolloutRecords,
    read_only: bool = False
)

Saved outcomes shared by the evaluation controller and offline readers.

_allocated

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

attempt_cap

int

Every rollout’s attempt cap: the saved max_rollout_attempts plus the recorded extension.

attempt_extension

int

Attempts an explicit recovery added to every rollout’s saved cap (--extend-attempts).

manifest

= state.manifest

outcomes_recorded

int = 0

progress_outcomes_recorded

int = 0

started_fresh

bool = False

nemo_gym.rollouts.rollout_store.RolloutStore.__enter__() -> nemo_gym.rollouts.rollout_store.RolloutStore

nemo_gym.rollouts.rollout_store.RolloutStore.__exit__(
    exc_type,
    exc_value,
    traceback
)

nemo_gym.rollouts.rollout_store.RolloutStore._require_open() -> None

nemo_gym.rollouts.rollout_store.RolloutStore._unverified_manifest(
    output: pathlib.Path
) -> nemo_gym.rollouts.rollout_recovery.RunManifest

staticmethod

nemo_gym.rollouts.rollout_store.RolloutStore._write_pending(
    pending: dict
) -> None

nemo_gym.rollouts.rollout_store.RolloutStore.admit_rejudgment(
    failure: dict
) -> nemo_gym.rollouts.rollout_recovery.PendingRejudge

Durably admit a judge-only re-judgment of the selected judge failure before its judge call.

The caller holds run_lock. Only the manifest changes: the failure stays selected with its debit, and no model attempt is reserved. failure may also be a selected saved result whose grading is re-verified from its saved answer: a failed re-verification of it saves nothing, so it spends no failure attempt and needs none.

nemo_gym.rollouts.rollout_store.RolloutStore.allocate_attempt(
    row: dict
) -> int

Durably reserve the next execution number before dispatch or restore.

The controller passes the returned number to checkpoint restore; restore must not increment it. A failed restore reserves a new number before a fresh execution. Reservations do not consume the failure retry budget.

nemo_gym.rollouts.rollout_store.RolloutStore.allocate_attempts(
    rows: list[dict]
) -> None

Reserve one admitted batch atomically before any of its requests start.

nemo_gym.rollouts.rollout_store.RolloutStore.attempt_count(
    row: dict
) -> int

nemo_gym.rollouts.rollout_store.RolloutStore.check_rejudgment(
    failure: dict,
    result: dict
) -> dict | None

Return the row that would settle failure (None when already saved); raise when it cannot settle it.

nemo_gym.rollouts.rollout_store.RolloutStore.close_rejudgment(
    identity: str,
    operation_id: str
) -> None

Close an admission once the selected outcome carries its operation id (the outcome is already fsynced).

nemo_gym.rollouts.rollout_store.RolloutStore.coverage() -> dict

nemo_gym.rollouts.rollout_store.RolloutStore.disposition(
    row: dict
) -> str

Return the newest allocated attempt’s disposition for this rollout.

nemo_gym.rollouts.rollout_store.RolloutStore.failure_export_metrics() -> dict[str, int]

Export saved failure totals with current outcome selection and retry budgets.

Failed-attempt totals retain history across resubmissions. Budget exhaustion uses only eligible failures since the most recent reset and the manifest’s saved cap, rather than the exporting process’s configuration.

nemo_gym.rollouts.rollout_store.RolloutStore.failures() -> list[dict]

Latest failure payloads, including terminal skips classified as omitted.

nemo_gym.rollouts.rollout_store.RolloutStore.inputs_for(
    results: list[dict]
) -> list[dict]

nemo_gym.rollouts.rollout_store.RolloutStore.open_for_rejudgment(
    output: pathlib.Path
) -> nemo_gym.rollouts.rollout_store.RolloutStore

classmethod

Load a managed run so judge-only re-judgments can settle its judge failures; the caller holds run_lock.

nemo_gym.rollouts.rollout_store.RolloutStore.pending(
    max_attempts: int | None = None,
    retry_terminal_timeouts: bool | None = None,
    dispatch_longest_first: bool = False
) -> list[dict]

Select unfinished work, retaining the saved timeout policy unless explicitly changed.

The controller holds the writer lock while updating a run’s policy. A read-only store can preview an explicit override without changing files.

nemo_gym.rollouts.rollout_store.RolloutStore.read(
    output: pathlib.Path,
    import_legacy: bool = True,
    retry_terminal_timeouts: bool = False,
    skip_keys: frozenset[str] = frozenset()
) -> nemo_gym.rollouts.rollout_store.RolloutStore | None

classmethod

Read selected outcomes without modifying files or acquiring a writer lock.

Managed runs use their saved timeout policy. retry_terminal_timeouts applies only to legacy artifacts without a manifest. skip_keys (see RolloutRecords.load) is for a selection that reads no trajectory: its records read back without those values.

nemo_gym.rollouts.rollout_store.RolloutStore.record_dispatch(
    row: dict
) -> None

Reserve once for this controller’s request; no dispatch event is stored.

nemo_gym.rollouts.rollout_store.RolloutStore.record_dispatches(
    rows: list[dict]
) -> None

Reserve requests admitted together with one durable manifest write.

nemo_gym.rollouts.rollout_store.RolloutStore.record_omission(
    row: dict,
    reason: str
) -> None

Save intentional omissions separately from scored results and failures.

nemo_gym.rollouts.rollout_store.RolloutStore.record_outcome(
    result: dict,
    sync: bool = False
) -> None

Append the outcome once; a complete JSONL row is the recovery record.

nemo_gym.rollouts.rollout_store.RolloutStore.record_rejudgment(
    failure: dict,
    result: dict
) -> bool

Settle the selected judge failure failure with its re-judged result for the same attempt.

The result keeps the failure’s run, rollout and attempt identity and its saved answer, so no model attempt is added and the failure stays in the sidecar with its debit. Returns False when that settlement is already saved.

nemo_gym.rollouts.rollout_store.RolloutStore.record_unstarted(
    row: dict
) -> None

Save record_omitted_rollouts evidence for a row the dispatch budget never started.

It is not an outcome: the history skips it on load, and resume still runs the rollout.

nemo_gym.rollouts.rollout_store.RolloutStore.replayable_identities() -> set[str]

The logical rollouts this Store would run again under some attempt cap: its replay rule without the cap.

A recovery that asks for more attempts reads which exhausted rollouts the extension would make pending.

nemo_gym.rollouts.rollout_store.RolloutStore.rerun_approved_identities() -> set[str]

The logical rollouts whose selected result is a masked one an operator approved to run again (--rerun-masked-without-artifact): each counts one failed attempt and a later attempt supersedes it.

nemo_gym.rollouts.rollout_store.RolloutStore.selected(
    disposition: str
) -> list[dict]

Load selected payloads; use selected_records for metadata-only access.

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

Map selected rollout/attempt identities to their original file slices.

nemo_gym.rollouts.rollout_store.RolloutStore.start_or_resume(
    output: pathlib.Path,
    prepare_inputs: collections.abc.Callable[[], tuple[list[dict], nemo_gym.rollouts.rollout_recovery.RunManifest]],
    resume: bool,
    allow_unsafe: bool = False,
    resubmit: bool = False
) -> nemo_gym.rollouts.rollout_store.RolloutStore

classmethod

Prepare or validate a run while its controller holds the run lock.

nemo_gym.rollouts.rollout_store.RolloutStore.unresolved_rejudgments() -> list[str]

Identities with an admitted re-judgment whose outcome is not saved: whether its judge ran is unknown.

nemo_gym.rollouts.rollout_store.RolloutStore.withdraw_rejudgment(
    identity: str,
    operation_id: str,
    unsaved_judge_failure: bool = False
) -> None

Withdraw the admission of a saved result’s re-verification that answered without a grade.

The verifier answered, so its outcome is known; nothing is saved and the saved result stays selected as it was. Only a saved-result admission can be withdrawn: a judge failure’s re-judgment saves its failed outcome. With unsaved_judge_failure the caller saves no outcome for this judge failure’s re-judgment (its verifier answered without a grade or a judge failure, or its pass raised before it answered), so its admission is withdrawn too: the failure stays selected, and a later pass judges its saved answer again.

nemo_gym.rollouts.rollout_store.RolloutStore.write_coverage(
    reporting: int = {}
) -> None

nemo_gym.rollouts.rollout_store._record_output_write(
    started: float,
    raw: bytes
) -> None

Measure the persisted outcome bytes, including their JSONL newline.

nemo_gym.rollouts.rollout_store.raw_outcomes_are_selected(
    path: pathlib.Path,
    artifacts: tuple[pathlib.Path, ...] | None = None
) -> bool

Whether an existing row-oriented reader can read this run without filtering.

A manifest alone does not require a different reader. Superseded rows, including answers superseded by interrupted attempts, do. Validate the whole run, but compare only the artifacts the caller actually reads when supplied.

nemo_gym.rollouts.rollout_store._REJUDGE_DROPPED_PREFIXES = ('_ng_failure', '_ng_rejudge')

nemo_gym.rollouts.rollout_store._REJUDGE_GRADING_KEYS = frozenset({'reward', 'mask_sample', 'evaluation_completed', 'failure_kind', 'fai...

nemo_gym.rollouts.rollout_store.logger = logging.getLogger(__name__)