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

Classes

Name	Description
`CheckInput`	Persisted or derived evidence required to evaluate a check.
`CheckScope`	-
`CheckSpec`	Stable, self-describing health-check contract.
`CheckSubject`	-
`Finding`	Evidence emitted by a check. Checks never emit verdicts.
`HealthCheckResult`	-
`RolloutDigest`	-

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`	-
`normalize_ignored_checks`	Normalize and validate check IDs supplied by library, CLI, or Hydra config.
`run_health_checks`	Run the RFC’s map/group/reduce pipeline and write both reports.

Data

CHECK_REGISTRY

Verdict

_PROCESS_POOL_CHUNKS_PER_WORKER

_PROCESS_POOL_MAX_CHUNKSIZE

API

class nemo_gym.rollout_health.CheckInput

Bases: enum.Enum

Persisted or derived evidence required to evaluate a check.

AGENT_TURNS

= 'agent_turns'

BOUND_CALLS

= 'bound_calls'

OBSERVED_MODEL_CALLS

= 'observed_model_calls'

OWNED_MODEL_CALLS

= 'owned_model_calls'

RECORD

= 'record'

REPEAT_DIGESTS

= 'repeat_digests'

REPEAT_VERDICTS

= 'repeat_verdicts'

TRAJECTORY

= 'trajectory'

class nemo_gym.rollout_health.CheckScope

Bases: enum.Enum

ROLLOUT

= 'rollout'

RUN

= 'run'

TASK

= 'task'

class nemo_gym.rollout_health.CheckSpec()

Bases: BaseModel

Stable, self-describing health-check contract.

evaluation_scope

CheckScope

str

model_config

= ConfigDict(frozen=True)

reads

frozenset[CheckInput]

subject

CheckSubject

nemo_gym.rollout_health.CheckSpec._require_one_call_binding_view(
    reads: frozenset[nemo_gym.health.types.CheckInput]
) -> frozenset[nemo_gym.health.types.CheckInput]

classmethod

class nemo_gym.rollout_health.CheckSubject

Bases: enum.Enum

AGENT_TURN

= 'agent_turn'

CHECK_EXECUTION

= 'check_execution'

MODEL_CALL

= 'model_call'

RECORD

= 'record'

ROLLOUT

= 'rollout'

TASK

= 'task'

TRAJECTORY_CAPTURE

= 'trajectory_capture'

class nemo_gym.rollout_health.Finding()

Bases: BaseModel

Evidence emitted by a check. Checks never emit verdicts.

check

str

detail

dict[str, Any] = Field(default_factory=dict)

locator

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

subject

dict[str, int | str]

class nemo_gym.rollout_health.HealthCheckResult()

Bases: BaseModel

rollouts

list[RolloutDigest]

summary

dict[str, Any]

summary_path

Path

verdicts_path

Path

class nemo_gym.rollout_health.RolloutDigest()

Bases: BaseModel

capture_completion_tokens

int = 0

capture_observed

bool

capture_prompt_tokens

int = 0

duplicated_calls

int = 0

ended_on_error

bool = False

errors_by_status

dict[str, int] = Field(default_factory=dict)

findings

list[Finding]

model_call_errors

int = 0

model_calls

int = 0

policy_calls_observed

bool = False

rollout_id

str

rollout_index

int | str

stage_index

int | str | None = None

successful_model_calls

int = 0

task_index

int | str

transcript_completion_tokens

int = 0

transcript_prompt_tokens

int = 0

unobserved

list[str]

verdict

Verdict

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], ...]

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.normalize_ignored_checks(
    checks: collections.abc.Sequence[str] | str | None
) -> tuple[str, ...]

Normalize and validate check IDs supplied by library, CLI, or Hydra config.

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.health.checks.CHECK_REGISTRY: tuple[CheckSpec, ...] = (CheckSpec(id='check_execution_error', evaluation_scope=(CheckScope.ROLLOUT), su...

nemo_gym.health.types.Verdict = Literal['healthy', 'unhealthy', 'unobserved']

nemo_gym.rollout_health._PROCESS_POOL_CHUNKS_PER_WORKER = 4

nemo_gym.rollout_health._PROCESS_POOL_MAX_CHUNKSIZE = 128