nemo_gym.health.types

View as Markdown

Data contracts shared by rollout-health checks and orchestration.

Module Contents

Classes

NameDescription
CheckInputPersisted or derived evidence required to evaluate a check.
CheckScope-
CheckSpecStable, self-describing health-check contract.
CheckSubject-
FindingEvidence emitted by a check. Checks never emit verdicts.
HealthCheckResult-
RolloutDigest-
_AgentStep-
_CallBindings-
_LineSlice-
_TaskRepeat-
_WorkerInput-

Data

QUALITY_SUMMARY_FILENAME

ROLLOUT_ID_KEY

ROLLOUT_INDEX_KEY

ROLLOUT_VERDICTS_FILENAME

TASK_INDEX_KEY

Verdict

API

class nemo_gym.health.types.CheckInput

Bases: enum.Enum

Persisted or derived evidence required to evaluate a check.

AGENT_TURNS
= 'agent_turns'
BOUND_CALLS
= 'bound_calls'
RECORD
= 'record'
REPEAT_DIGESTS
= 'repeat_digests'
REPEAT_VERDICTS
= 'repeat_verdicts'
TRAJECTORY
= 'trajectory'
class nemo_gym.health.types.CheckScope

Bases: enum.Enum

ROLLOUT
= 'rollout'
RUN
= 'run'
TASK
= 'task'
class nemo_gym.health.types.CheckSpec()

Bases: BaseModel

Stable, self-describing health-check contract.

evaluation_scope
CheckScope
id
str
model_config
= ConfigDict(frozen=True)
reads
frozenset[CheckInput]
subject
CheckSubject
class nemo_gym.health.types.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.health.types.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.health.types.HealthCheckResult()

Bases: BaseModel

rollouts
list[RolloutDigest]
summary
dict[str, Any]
summary_path
Path
verdicts_path
Path
class nemo_gym.health.types.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
successful_model_calls
int = 0
task_index
int | str
transcript_completion_tokens
int = 0
transcript_prompt_tokens
int = 0
unobserved
list[str]
verdict
Verdict
class nemo_gym.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.health.types._CallBindings(
references: tuple[str, ...],
matched_calls: tuple[dict[str, typing.Any], ...],
missing_references: tuple[str, ...],
duplicated_references: tuple[tuple[str, int], ...]
)
Dataclass
complete
bool
duplicated_references
tuple[tuple[str, int], ...]
matched_calls
tuple[dict[str, Any], ...]
missing_references
tuple[str, ...]
observed
bool
references
tuple[str, ...]
class nemo_gym.health.types._LineSlice(
path: str,
offset: int,
length: int,
ordinal: int,
source_index: int,
line_number: int
)
Dataclass
length
int
line_number
int
offset
int
ordinal
int
path
str
source_index
int
class nemo_gym.health.types._TaskRepeat(
rollout_index: int | str,
verdict: nemo_gym.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.health.types._WorkerInput(
line: nemo_gym.health.types._LineSlice,
ignored_checks: frozenset[str]
)
Dataclass
ignored_checks
frozenset[str]
line
_LineSlice
nemo_gym.health.types.QUALITY_SUMMARY_FILENAME = 'quality_summary.json'
nemo_gym.health.types.ROLLOUT_ID_KEY = ROLLOUT_ID_KEY_NAME
nemo_gym.health.types.ROLLOUT_INDEX_KEY = ROLLOUT_INDEX_KEY_NAME
nemo_gym.health.types.ROLLOUT_VERDICTS_FILENAME = 'rollout_verdicts.jsonl'
nemo_gym.health.types.TASK_INDEX_KEY = TASK_INDEX_KEY_NAME
nemo_gym.health.types.Verdict = Literal['healthy', 'unhealthy', 'unobserved']