> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/gym/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/gym/_mcp/server.

# nemo_gym.batch_status

Validated Eval Factory batch manifests and compact, atomic Gym status artifacts.

## Module Contents

### Classes

| Name                                                                | Description                                                                |
| ------------------------------------------------------------------- | -------------------------------------------------------------------------- |
| [`BatchManifest`](#nemo_gym-batch_status-BatchManifest)             | Versioned, secret-free handoff from Eval Factory to Gym.                   |
| [`BatchManifestMember`](#nemo_gym-batch_status-BatchManifestMember) | Eval Factory's expectations for one logical benchmark in a shared Gym run. |
| [`BatchRepeatPolicy`](#nemo_gym-batch_status-BatchRepeatPolicy)     | The repeat shape Gym must observe for every task belonging to one member.  |
| [`BatchStatusTracker`](#nemo_gym-batch_status-BatchStatusTracker)   | Validate one batch manifest and materialize its current per-agent status.  |
| [`ObservedBatchMember`](#nemo_gym-batch_status-ObservedBatchMember) | Dataset and repeat facts derived from materialized Gym input rows.         |

### Functions

| Name                                                                            | Description                                                                              |
| ------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- |
| [`_atomic_write_json`](#nemo_gym-batch_status-_atomic_write_json)               | -                                                                                        |
| [`_canonical_task_bytes`](#nemo_gym-batch_status-_canonical_task_bytes)         | Canonicalize task content independently of routing, repeats, retries, and skills.        |
| [`_read_aggregate_entries`](#nemo_gym-batch_status-_read_aggregate_entries)     | -                                                                                        |
| [`_safe_aggregation_error`](#nemo_gym-batch_status-_safe_aggregation_error)     | Copy only non-secret diagnostics into batch\_status.json.                                |
| [`_seed_from_row`](#nemo_gym-batch_status-_seed_from_row)                       | -                                                                                        |
| [`load_batch_manifest`](#nemo_gym-batch_status-load_batch_manifest)             | Load a manifest and return it with the SHA-256 of the exact input bytes.                 |
| [`observe_materialized_rows`](#nemo_gym-batch_status-observe_materialized_rows) | Derive per-agent dataset fingerprints and repeat policies from full materialized inputs. |
| [`validate_batch_manifest`](#nemo_gym-batch_status-validate_batch_manifest)     | Reject any manifest expectation that disagrees with Gym's materialized inputs.           |

### Data

[`AGGREGATION_ERROR_KEY`](#nemo_gym-batch_status-AGGREGATION_ERROR_KEY)

[`BATCH_STATUS_FNAME`](#nemo_gym-batch_status-BATCH_STATUS_FNAME)

[`BATCH_STATUS_WRITE_INTERVAL_SECONDS`](#nemo_gym-batch_status-BATCH_STATUS_WRITE_INTERVAL_SECONDS)

[`_DATASET_DIGEST_DOMAIN`](#nemo_gym-batch_status-_DATASET_DIGEST_DOMAIN)

[`_FAILURE_CLASS_KEY`](#nemo_gym-batch_status-_FAILURE_CLASS_KEY)

[`_HEX_SHA256`](#nemo_gym-batch_status-_HEX_SHA256)

[`_MISSING`](#nemo_gym-batch_status-_MISSING)

### API

```python
class nemo_gym.batch_status.BatchManifest()
```

**Bases:** `BaseModel`

Versioned, secret-free handoff from Eval Factory to Gym.

**`members`** `Dict[str, BatchManifestMember] = Field(min_length=1)`

---

**`model_config`** `= ConfigDict(extra='forbid')`

---

**`schema_version`** `Literal['1'] = '1'`

---

```python
nemo_gym.batch_status.BatchManifest._validate_members() -> nemo_gym.batch_status.BatchManifest
```

```python
class nemo_gym.batch_status.BatchManifestMember()
```

**Bases:** `BaseModel`

Eval Factory's expectations for one logical benchmark in a shared Gym run.

**`agent_name`** `str`

---

**`dataset_sha256`** `str`

---

**`expected_rollout_count`** `int = Field(ge=1)`

---

**`expected_task_count`** `int = Field(ge=1)`

---

**`metric_keys`** `List[str]`

---

**`model_config`** `= ConfigDict(extra='forbid')`

---

**`repeat_policy`** `BatchRepeatPolicy`

---

**`resolved_recipe_sha256`** `str`

---

**`task_sources`** `List[str] = Field(min_length=1)`

---

```python
nemo_gym.batch_status.BatchManifestMember._validate_agent_name(
    value: str
) -> str
```

classmethod

```python
nemo_gym.batch_status.BatchManifestMember._validate_rollout_count() -> nemo_gym.batch_status.BatchManifestMember
```

```python
nemo_gym.batch_status.BatchManifestMember._validate_sha256(
    value: str
) -> str
```

classmethod

```python
nemo_gym.batch_status.BatchManifestMember._validate_unique_names(
    values: typing.List[str]
) -> typing.List[str]
```

classmethod

```python
class nemo_gym.batch_status.BatchRepeatPolicy()
```

**Bases:** `BaseModel`

The repeat shape Gym must observe for every task belonging to one member.

**`model_config`** `= ConfigDict(extra='forbid')`

---

**`num_repeats`** `int = Field(ge=1)`

---

**`seeded`** `bool`

---

```python
class nemo_gym.batch_status.BatchStatusTracker(
    manifest_fpath: pathlib.Path,
    materialized_rows: typing.Sequence[typing.Mapping[str, typing.Any]],
    write_interval_seconds: typing.Optional[float] = None
)
```

Validate one batch manifest and materialize its current per-agent status.

**`_expected_rollout_identities`** `set[tuple[str, int, int]]`

---

**`_last_progress_count`** `= 0`

---

**`_last_write`** `= 0.0`

---

**`observations`** `= observe_materialized_rows(materialized_rows)`

---

**`status_fpath`** `= manifest_fpath.with_name(BATCH_STATUS_FNAME)`

---

**`write_interval_seconds`**

---

```python
nemo_gym.batch_status.BatchStatusTracker._validated_rollout_identity(
    row: typing.Mapping[str, object],
    record_kind: str,
    row_index: int
) -> tuple[str, int, int]
```

Reject malformed IDs and records belonging to a different materialized batch.

```python
nemo_gym.batch_status.BatchStatusTracker.build_status(
    completed_results: typing.Sequence[typing.Mapping[str, typing.Any]],
    failure_rows: typing.Sequence[typing.Mapping[str, typing.Any]],
    aggregate_metrics_fpath: typing.Optional[pathlib.Path] = None,
    aggregation_deferred: bool = False
) -> typing.Dict[str, typing.Any]
```

Build status from expected identities, raising ConfigError for mismatched result records.

```python
nemo_gym.batch_status.BatchStatusTracker.write_status(
    completed_results: typing.Sequence[typing.Mapping[str, typing.Any]],
    failure_rows: typing.Sequence[typing.Mapping[str, typing.Any]],
    aggregate_metrics_fpath: typing.Optional[pathlib.Path] = None,
    aggregation_deferred: bool = False,
    force: bool = False
) -> bool
```

Atomically refresh status, throttling progress writes unless `force` is true.

```python
class nemo_gym.batch_status.ObservedBatchMember(
    task_sources: typing.List[str],
    dataset_sha256: str,
    task_count: int,
    rollout_count: int,
    repeat_policy: nemo_gym.batch_status.BatchRepeatPolicy
)
```

Dataclass

Dataset and repeat facts derived from materialized Gym input rows.

**`dataset_sha256`** `str`

---

**`repeat_policy`** `BatchRepeatPolicy`

---

**`rollout_count`** `int`

---

**`task_count`** `int`

---

**`task_sources`** `List[str]`

---

```python
nemo_gym.batch_status._atomic_write_json(
    path: pathlib.Path,
    payload: typing.Mapping[str, typing.Any]
) -> None
```

```python
nemo_gym.batch_status._canonical_task_bytes(
    row: typing.Mapping[str, typing.Any]
) -> bytes
```

Canonicalize task content independently of routing, repeats, retries, and skills.

```python
nemo_gym.batch_status._read_aggregate_entries(
    metrics_fpath: typing.Optional[pathlib.Path]
) -> tuple[typing.Dict[str, typing.Dict[str, typing.Any]], typing.Optional[str]]
```

```python
nemo_gym.batch_status._safe_aggregation_error(
    error: typing.Any
) -> typing.Dict[str, typing.Any]
```

Copy only non-secret diagnostics into batch\_status.json.

```python
nemo_gym.batch_status._seed_from_row(
    row: typing.Mapping[str, typing.Any]
) -> typing.Any
```

```python
nemo_gym.batch_status.load_batch_manifest(
    manifest_fpath: pathlib.Path
) -> tuple[nemo_gym.batch_status.BatchManifest, str]
```

Load a manifest and return it with the SHA-256 of the exact input bytes.

```python
nemo_gym.batch_status.observe_materialized_rows(
    rows: typing.Sequence[typing.Mapping[str, typing.Any]]
) -> typing.Dict[str, nemo_gym.batch_status.ObservedBatchMember]
```

Derive per-agent dataset fingerprints and repeat policies from full materialized inputs.

The dataset digest hashes one canonical row per task in its materialized order. Runtime
routing and identity fields, repeat seeds, retry attempts, and run-level skills are excluded;
prompt-expanded inputs and `task_source` remain part of the dataset identity.

```python
nemo_gym.batch_status.validate_batch_manifest(
    manifest: nemo_gym.batch_status.BatchManifest,
    observations: typing.Mapping[str, nemo_gym.batch_status.ObservedBatchMember]
) -> None
```

Reject any manifest expectation that disagrees with Gym's materialized inputs.

```python
nemo_gym.batch_status.AGGREGATION_ERROR_KEY = 'aggregation_error'
```

```python
nemo_gym.batch_status.BATCH_STATUS_FNAME = 'batch_status.json'
```

```python
nemo_gym.batch_status.BATCH_STATUS_WRITE_INTERVAL_SECONDS = 5.0
```

```python
nemo_gym.batch_status._DATASET_DIGEST_DOMAIN = b'nemo-gym-batch-dataset-v1\x00'
```

```python
nemo_gym.batch_status._FAILURE_CLASS_KEY = '_ng_failure_class'
```

```python
nemo_gym.batch_status._HEX_SHA256 = re.compile('^[0-9a-f]{64}$')
```

```python
nemo_gym.batch_status._MISSING = object()
```