nemo_gym.rollout_reverification

Module Contents

Classes

Name	Description
`CacheKeysByStatus`	-
`InputRolloutPair`	-
`OutputPaths`	-
`RolloutReverificationConfig`	-
`RolloutReverificationHelper`	-

Functions

Name	Description
`_agent_to_rs_mapping_from_agent_blocks`	-
`_agent_to_rs_mapping_from_resources_only_config`	-
`_build_agent_to_resources_server_mapping`	-
`_build_verify_payload`	-
`_call_aggregate_metrics`	Call /aggregate_metrics on each resource server after rollouts complete.
`_check_reverify_mode`	Query GET /reverify_mode on each unique resource server referenced by agent_to_rs.
`_drop_cache_from_payloads`	-
`_get_rs_names`	-
`_guard_reverify_mode`	Check reverify_mode for every RS in the config before reverification starts.
`_is_judge_failure`	Whether a failures-sidecar row is a judge failure (the only recoverable class).
`_load_cache_keys_by_status`	-
`_load_reverified_results`	Load the full main jsonl (cached + newly re-verified successes), sorted by (task, rollout).
`_parse_output_line`	-
`_parse_output_line_key`	-
`_prepare_output_fpaths`	-
`_prepare_payloads`	-
`_recovery_rollout_predicate`	Build the row filter for `--judge-failed-only` recovery.
`_rollout_verify_debug_summary`	-
`_run_verification_payloads`	-
`_seed_output_with_successes`	Copy the already-successful rollout rows into the output file so the final aggregate covers
`_yield_inputs_and_rollouts_paired`	-
`summarize_cache_usage`	-

Data

JUDGE_FAILED_FAILURE_CLASS

_RECOVERY_TWO_SOURCES_WARNING

API

class nemo_gym.rollout_reverification.CacheKeysByStatus(
    successful_keys: set[tuple[int, int]],
    terminal_keys: set[tuple[int, int]],
    maxed_out_keys: set[tuple[int, int]]
)

Dataclass

maxed_out_keys

set[tuple[int, int]]

successful_keys

set[tuple[int, int]]

terminal_keys

set[tuple[int, int]]

class nemo_gym.rollout_reverification.InputRolloutPair(
    input: typing.Dict[str, typing.Any],
    rollout: typing.Dict[str, typing.Any]
)

Dataclass

input

Dict[str, Any]

rollout

Dict[str, Any]

class nemo_gym.rollout_reverification.OutputPaths(
    output: pathlib.Path,
    failures: pathlib.Path
)

Dataclass

failures

Path

output

Path

class nemo_gym.rollout_reverification.RolloutReverificationConfig()

Bases: BaseNeMoGymCLIConfig

append

bool

disable_aggregation

bool

force

bool

judge_failed_only

bool

limit

Optional[int]

materialized_inputs_jsonl_fpath

str

num_samples_in_parallel

Optional[int]

output_jsonl_fpath

str

overwrite

bool

resume_from_cache

bool

rollouts_jsonl_fpath

str

upload_rollouts_to_wandb

bool

nemo_gym.rollout_reverification.RolloutReverificationConfig._validate_append() -> nemo_gym.rollout_reverification.RolloutReverificationConfig

class nemo_gym.rollout_reverification.RolloutReverificationHelper()

Bases: BaseModel

nemo_gym.rollout_reverification.RolloutReverificationHelper.run_from_config(
    config: nemo_gym.rollout_reverification.RolloutReverificationConfig
) -> typing.List[typing.Dict]

async

nemo_gym.rollout_reverification._agent_to_rs_mapping_from_agent_blocks(
    global_config_dict: typing.Union[typing.Dict[str, typing.Any], omegaconf.DictConfig]
) -> typing.Dict[str, str]

nemo_gym.rollout_reverification._agent_to_rs_mapping_from_resources_only_config(
    global_config_dict: typing.Union[typing.Dict[str, typing.Any], omegaconf.DictConfig]
) -> typing.Dict[str, str]

nemo_gym.rollout_reverification._build_agent_to_resources_server_mapping(
    global_config_dict: typing.Union[typing.Dict[str, typing.Any], omegaconf.DictConfig]
) -> typing.Dict[str, str]

nemo_gym.rollout_reverification._build_verify_payload(
    pair: nemo_gym.rollout_reverification.InputRolloutPair
) -> typing.Dict

nemo_gym.rollout_reverification._call_aggregate_metrics(
    results: typing.List[typing.Dict],
    rows: typing.List[typing.Dict],
    output_fpath: pathlib.Path
) -> typing.Optional[pathlib.Path]

async

Call /aggregate_metrics on each resource server after rollouts complete.

Writes a single _aggregate_metrics.json with one entry per agent (same shape as the old _agent_metrics.json). Returns the file path.

nemo_gym.rollout_reverification._check_reverify_mode(
    server_client: nemo_gym.server_utils.ServerClient,
    agent_to_rs: typing.Dict[str, str]
) -> typing.List[str]

async

Query GET /reverify_mode on each unique resource server referenced by agent_to_rs.

Returns a sorted list of RS names that reported ReverifyMode.UNSUPPORTED or ReverifyMode.UNKNOWN.

nemo_gym.rollout_reverification._drop_cache_from_payloads(
    payloads: typing.List[typing.Dict],
    cache: nemo_gym.rollout_reverification.CacheKeysByStatus
) -> typing.Iterator[typing.Dict]

nemo_gym.rollout_reverification._get_rs_names(
    agent_to_rs: typing.Dict[str, str]
) -> typing.List[str]

nemo_gym.rollout_reverification._guard_reverify_mode(
    config: nemo_gym.rollout_reverification.RolloutReverificationConfig
) -> typing.Optional[str]

async

Check reverify_mode for every RS in the config before reverification starts.

Returns a warning string when the user runs full re-verification (not judge-failed-only) and force=True and at least one RS is UNSUPPORTED or UNKNOWN (caller must print it and apply the unsafe_ output prefix). Raises ConfigError when force=False and at least one RS is UNSUPPORTED or UNKNOWN. Returns None when all RS are STATELESS.

nemo_gym.rollout_reverification._is_judge_failure(
    row: typing.Dict[str, typing.Any]
) -> bool

Whether a failures-sidecar row is a judge failure (the only recoverable class).

nemo_gym.rollout_reverification._load_cache_keys_by_status(
    output_fpaths: nemo_gym.rollout_reverification.OutputPaths
) -> nemo_gym.rollout_reverification.CacheKeysByStatus

nemo_gym.rollout_reverification._load_reverified_results(
    output_fpath: pathlib.Path
) -> typing.Tuple[typing.List[typing.Dict], typing.List[typing.Dict]]

Load the full main jsonl (cached + newly re-verified successes), sorted by (task, rollout).

Returns (results, rows): results are the parsed rows — the source of truth used for both the W&B rollouts export and the aggregate-metrics payload; rows is a minimal {AGENT_REF} projection used only to route each result to its resources server. Read once and reused for both so the file is never read twice.

nemo_gym.rollout_reverification._parse_output_line(
    line: bytes
) -> typing.Dict[str, typing.Any]

nemo_gym.rollout_reverification._parse_output_line_key(
    line: bytes
) -> tuple[int, int] | None

nemo_gym.rollout_reverification._prepare_output_fpaths(
    output_name_prefix: str,
    output_jsonl_fpath: str,
    resume_from_cache: bool,
    overwrite: bool,
    append: bool
) -> nemo_gym.rollout_reverification.OutputPaths

nemo_gym.rollout_reverification._prepare_payloads(
    materialized_inputs_jsonl_fpath: pathlib.Path,
    rollouts_jsonl_fpath: pathlib.Path,
    output_fpaths: nemo_gym.rollout_reverification.OutputPaths,
    resume_from_cache: bool,
    limit: typing.Optional[int] = None,
    rollout_predicate: typing.Optional[typing.Callable[[Dict[str, Any]], bool]] = None
) -> typing.List[typing.Dict]

nemo_gym.rollout_reverification._recovery_rollout_predicate(
    skip_keys: typing.Optional[set[tuple[typing.Any, typing.Any]]] = None
) -> typing.Callable[[Dict[str, Any]], bool]

Build the row filter for --judge-failed-only recovery.

Keeps only judge-class failures, skips any whose (task, rollout) key is in skip_keys (the keys already present in the output: seeded successes + rows a prior recovery already appended) so a key-in-both is never re-verified/duplicated and re-runs are idempotent — this dedup is done here, independent of the resume/cache machinery — and dedups on the key so a sidecar with multiple failure attempts for one key re-verifies it exactly once.

nemo_gym.rollout_reverification._rollout_verify_debug_summary(
    row: typing.Dict[str, typing.Any],
    resources_server_name: str
) -> typing.Dict[str, typing.Any]

nemo_gym.rollout_reverification._run_verification_payloads(
    payloads: typing.List[typing.Dict],
    semaphore: asyncio.Semaphore | contextlib.nullcontext[None] | None = None
) -> typing.Iterator[asyncio.Future]

nemo_gym.rollout_reverification._seed_output_with_successes(
    successes_fpath: pathlib.Path,
    output_fpath: pathlib.Path
) -> set[tuple[int, int]]

Copy the already-successful rollout rows into the output file so the final aggregate covers successes + recovered rows, and return the set of (task, rollout) keys ALREADY PRESENT in the output afterward

nemo_gym.rollout_reverification._yield_inputs_and_rollouts_paired(
    materialized_inputs_jsonl_fpath: pathlib.Path,
    rollouts_jsonl_fpath: pathlib.Path,
    limit: typing.Optional[int] = None,
    rollout_predicate: typing.Optional[typing.Callable[[Dict[str, Any]], bool]] = None
) -> typing.Iterator[nemo_gym.rollout_reverification.InputRolloutPair]

nemo_gym.rollout_reverification.summarize_cache_usage(
    cache: nemo_gym.rollout_reverification.CacheKeysByStatus,
    all_payloads: typing.List[typing.Dict],
    filtered_payloads: typing.List[typing.Dict]
) -> None

nemo_gym.rollout_reverification.JUDGE_FAILED_FAILURE_CLASS = 'judge_failed'

nemo_gym.rollout_reverification._RECOVERY_TWO_SOURCES_WARNING = "WARNING: judge-failed-only recovery merges rewards from TWO sources — the succe...