nemo_gym.rollout_health

Deterministic post-run rollout quality verification workflow.

Checks operate only on persisted rollout records and their canonical ng_trajectory evidence. They return evidence; this module derives verdicts and writes reports.

Module Contents

Functions

Name	Description
`_index_jsonl`	-
`_mark_duplicate_identities`	Flag physical records that claim the same logical rollout identity.
`_process_pool_chunksize`	Keep several schedulable chunks per worker without unbounded IPC batches.
`_read_record`	-
`_reduce`	-
`_resolve_rollout_path`	-
`_sort_key`	-
`_task_findings`	-
`_unique_task_repeats`	Collapse duplicate persisted records for task-level repeat semantics.
`_worker`	-
`_write_reports`	-
`format_health_report`	-
`health_check_run_dir`	-
`run_health_checks`	Run the RFC’s map/group/reduce pipeline and write both reports.

Data

_PROCESS_POOL_CHUNKS_PER_WORKER

_PROCESS_POOL_MAX_CHUNKSIZE

__all__

API

nemo_gym.rollout_health._index_jsonl(
    paths: collections.abc.Sequence[pathlib.Path]
) -> list[nemo_gym.health.types._LineSlice]

nemo_gym.rollout_health._mark_duplicate_identities(
    digests: list[nemo_gym.health.types.RolloutDigest],
    ignored_checks: frozenset[str]
) -> None

Flag physical records that claim the same logical rollout identity.

nemo_gym.rollout_health._process_pool_chunksize(
    item_count: int,
    workers: int
) -> int

Keep several schedulable chunks per worker without unbounded IPC batches.

nemo_gym.rollout_health._read_record(
    line: nemo_gym.health.types._LineSlice
) -> tuple[dict[str, typing.Any], str | None]

nemo_gym.rollout_health._reduce(
    digests: list[nemo_gym.health.types.RolloutDigest],
    ignored_checks: frozenset[str]
) -> dict[str, typing.Any]

nemo_gym.rollout_health._resolve_rollout_path(
    run_dir: pathlib.Path,
    rollout_file: str | pathlib.Path | None
) -> pathlib.Path

nemo_gym.rollout_health._sort_key(
    digest: nemo_gym.health.types.RolloutDigest
) -> tuple[tuple[int, typing.Any], tuple[int, typing.Any]]

nemo_gym.rollout_health._task_findings(
    grouped: dict[int | str, list[nemo_gym.health.types._TaskRepeat]],
    ignored_checks: frozenset[str]
) -> tuple[dict[int | str, list[nemo_gym.health.types.Finding]], dict[str, dict[str, int]]]

nemo_gym.rollout_health._unique_task_repeats(
    digests: list[nemo_gym.health.types.RolloutDigest]
) -> list[nemo_gym.health.types._TaskRepeat]

Collapse duplicate persisted records for task-level repeat semantics.

nemo_gym.rollout_health._worker(
    payload: nemo_gym.health.types._WorkerInput
) -> nemo_gym.health.types.RolloutDigest

nemo_gym.rollout_health._write_reports(
    summary: dict[str, typing.Any],
    digests: list[nemo_gym.health.types.RolloutDigest],
    output_dir: pathlib.Path
) -> tuple[pathlib.Path, pathlib.Path]

nemo_gym.rollout_health.format_health_report(
    result: nemo_gym.health.types.HealthCheckResult
) -> str

nemo_gym.rollout_health.health_check_run_dir(
    run_dir: str | pathlib.Path,
    rollout_file: str | pathlib.Path | None = None,
    workers: int | None = None,
    ignored_checks: collections.abc.Sequence[str] = (),
    json_output: bool = False
) -> nemo_gym.health.types.HealthCheckResult

nemo_gym.rollout_health.run_health_checks(
    rollout_paths: pathlib.Path | collections.abc.Sequence[pathlib.Path],
    output_dir: pathlib.Path | None = None,
    workers: int | None = None,
    ignored_checks: collections.abc.Sequence[str] = ()
) -> nemo_gym.health.types.HealthCheckResult

Run the RFC’s map/group/reduce pipeline and write both reports.

nemo_gym.rollout_health._PROCESS_POOL_CHUNKS_PER_WORKER = 4

nemo_gym.rollout_health._PROCESS_POOL_MAX_CHUNKSIZE = 128

nemo_gym.rollout_health.__all__ = ['CHECK_REGISTRY', 'CheckInput', 'CheckScope', 'CheckSpec', 'CheckSubject', 'Fin...