nemo_gym.rollouts.health.types

Data contracts shared by rollout-health checks and orchestration.

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`	-
`RunHealthConfig`	The health-check settings every command that judges a run shares: one place for names and defaults.
`RunHealthLimits`	The run-level failure limits, as a share of expected rollouts; `None` is no limit, the default.
`_AgentStep`	-
`_CallBindings`	-
`_LineSlice`	-
`_TaskRepeat`	-
`_WorkerInput`	-

Data

CALL_BINDING_INPUTS

API

class nemo_gym.rollouts.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'

RUN_ACCOUNTING

= 'run_accounting'

TRAJECTORY

= 'trajectory'

class nemo_gym.rollouts.rollout_health.CheckScope

Bases: enum.Enum

ROLLOUT

= 'rollout'

RUN

= 'run'

TASK

= 'task'

class nemo_gym.rollouts.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.rollouts.rollout_health.CheckSpec._require_one_call_binding_view(
    reads: frozenset[nemo_gym.rollouts.health.types.CheckInput]
) -> frozenset[nemo_gym.rollouts.health.types.CheckInput]

classmethod

class nemo_gym.rollouts.rollout_health.CheckSubject

Bases: enum.Enum

AGENT_TURN

= 'agent_turn'

CHECK_EXECUTION

= 'check_execution'

MODEL_CALL

= 'model_call'

RECORD

= 'record'

ROLLOUT

= 'rollout'

RUN

= 'run'

TASK

= 'task'

TRAJECTORY_CAPTURE

= 'trajectory_capture'

class nemo_gym.rollouts.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.rollouts.rollout_health.HealthCheckResult()

Bases: BaseModel

rollouts

list[RolloutDigest]

summary

dict[str, Any]

summary_path

Path

verdicts_path

Path

class nemo_gym.rollouts.rollout_health.RolloutDigest()

Bases: BaseModel

capture_completion_tokens

int = 0

capture_observed

bool

capture_prompt_tokens

int = 0

context_window_exceeded

bool | None = None

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_call_errors_usage_unknown

int = 0

model_calls

int = 0

policy_calls_observed

bool = False

responses

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

rollout_id

str

rollout_index

int | str

scored

bool = False

stage_index

int | str | None = None

successful_model_calls

int = 0

task_index

int | str

timing

RolloutTiming | None = None

transcript_completion_tokens

int = 0

transcript_prompt_tokens

int = 0

unobserved

list[str]

upstream_recovery_observations

dict[str, int] | None = None

verdict

Verdict

verification_skipped

bool | None = None

class nemo_gym.rollouts.health.types.RunHealthConfig()

Bases: BaseModel

The health-check settings every command that judges a run shares: one place for names and defaults.

health_check_max_final_failure_fraction

float | None

health_check_max_not_final_failure_fraction

float | None

run_health_limits

RunHealthLimits

class nemo_gym.rollouts.health.types.RunHealthLimits()

Bases: BaseModel

The run-level failure limits, as a share of expected rollouts; None is no limit, the default.

max_final_failure_fraction

float | None = Field(default=None, ge=0, le=1)

max_not_final_failure_fraction

float | None = Field(default=None, ge=0, le=1)

model_config

= ConfigDict(frozen=True)

nemo_gym.rollouts.health.types.RunHealthLimits.summary(
    source: nemo_gym.rollouts.health.types.LimitsSource | collections.abc.Mapping[str, nemo_gym.rollouts.health.types.LimitsSource]
) -> dict[str, typing.Any]

The limits as written to quality_summary.json, so the verdict can be reproduced.

source maps each limit to where it came from; a single source applies to every limit.

class nemo_gym.rollouts.health.types._AgentStep(
    locator: dict[str, int | str],
    has_message: bool,
    has_tool_calls: bool,
    model_call_refs: tuple[str, ...]
)

Dataclass

has_message

bool

has_model_activity

bool

has_tool_calls

bool

locator

dict[str, int | str]

model_call_refs

tuple[str, ...]

class nemo_gym.rollouts.health.types._CallBindings(
    references: tuple[str, ...],
    matched_calls: tuple[dict[str, typing.Any], ...],
    missing_references: tuple[str, ...],
    duplicated_references: tuple[tuple[str, int], ...],
    resolutions: collections.abc.Mapping[tuple[str, str, str, str], tuple[dict[str, typing.Any], ...]] = dict()
)

Dataclass

complete

bool

duplicated_references

tuple[tuple[str, int], ...]

matched_calls

tuple[dict[str, Any], ...]

missing_references

tuple[str, ...]

observed

bool

references

tuple[str, ...]

resolutions

Mapping[tuple[str, str, str, str], tuple[dict[str, Any], ...]] = field(default_factory=dict)

class nemo_gym.rollouts.health.types._LineSlice(
    path: str,
    offset: int,
    length: int,
    ordinal: int,
    source_index: int,
    line_number: int,
    file_identity: tuple[int, int],
    expected_identity: tuple[str | None, str, int] | None = None,
    legacy_attempt_index: int | None = None
)

Dataclass

expected_identity

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

file_identity

tuple[int, int]

legacy_attempt_index

int | None = None

length

int

line_number

int

offset

int

ordinal

int

path

str

source_index

int

class nemo_gym.rollouts.health.types._TaskRepeat(
    rollout_index: int | str,
    verdict: nemo_gym.rollouts.health.types.Verdict,
    policy_calls_observed: bool,
    successful_model_calls: int
)

Dataclass

policy_calls_observed

bool

rollout_index

int | str

successful_model_calls

int

verdict

Verdict

class nemo_gym.rollouts.health.types._WorkerInput(
    line: nemo_gym.rollouts.health.types._LineSlice,
    ignored_checks: frozenset[str]
)

Dataclass

ignored_checks

frozenset[str]

line

_LineSlice

nemo_gym.rollouts.health.types.CALL_BINDING_INPUTS = frozenset({CheckInput.BOUND_CALLS, CheckInput.OWNED_MODEL_CALLS})

nemo_gym.rollouts.health.types.LimitsSource = Literal['config', 'persisted', 'override']

nemo_gym.rollouts.health.types.ROLLOUT_ID_KEY = ROLLOUT_ID_KEY_NAME

nemo_gym.rollouts.health.types.ROLLOUT_INDEX_KEY = ROLLOUT_INDEX_KEY_NAME

nemo_gym.rollouts.health.types.TASK_INDEX_KEY = TASK_INDEX_KEY_NAME

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

nemo_gym.rollouts.health.types.__getattr__ = moved_names_getattr(__name__)