> This page is for version 0.7.0. > For other versions, use one of these documentation indexes: > - Main (default): https://docs.nvidia.com/nemo/gym/main/llms.txt > - 0.7.0: https://docs.nvidia.com/nemo/gym/v0.7.0/llms.txt > - 0.6.0: https://docs.nvidia.com/nemo/gym/v0.6.0/llms.txt > - 0.5.1: https://docs.nvidia.com/nemo/gym/v0.5.1/llms.txt > - 0.5.0: https://docs.nvidia.com/nemo/gym/v0.5.0/llms.txt > - 0.4.0: https://docs.nvidia.com/nemo/gym/v0.4.0/llms.txt > - 0.3.0: https://docs.nvidia.com/nemo/gym/v0.3.0/llms.txt > - 0.2.1: https://docs.nvidia.com/nemo/gym/v0.2.1/llms.txt > 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.rollouts.rollout_collection ## Module Contents ### Classes | Name | Description | | ------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------- | | [`E2ERolloutCollectionConfig`](#nemo_gym-rollouts-rollout_collection-E2ERolloutCollectionConfig) | Spin up all necessary servers and perform a batch of rollout collection using each dataset inside the provided configs. | | [`RolloutAggregationConfig`](#nemo_gym-rollouts-rollout_collection-RolloutAggregationConfig) | Aggregate metrics across rollout shards produced by `gym eval run --no-serve +disable_aggregation=true`. | | [`RolloutAggregationHelper`](#nemo_gym-rollouts-rollout_collection-RolloutAggregationHelper) | - | | [`RolloutCollectionConfig`](#nemo_gym-rollouts-rollout_collection-RolloutCollectionConfig) | Perform a batch of rollout collection. | | [`RolloutCollectionHelper`](#nemo_gym-rollouts-rollout_collection-RolloutCollectionHelper) | Collect rollouts by sending tasks to a running Gym deployment. | | [`RunTimingReport`](#nemo_gym-rollouts-rollout_collection-RunTimingReport) | Health table for a collection whose rollouts have a per-rollout wall-clock limit. | | [`SharedRolloutCollectionConfig`](#nemo_gym-rollouts-rollout_collection-SharedRolloutCollectionConfig) | - | | [`_BoundedCompletionIterator`](#nemo_gym-rollouts-rollout_collection-_BoundedCompletionIterator) | Completion-order iterator with a bounded set of resident asyncio tasks. | | [`_CollectedOutcome`](#nemo_gym-rollouts-rollout_collection-_CollectedOutcome) | A finished dispatch as collection stores it, with the facts the collector's loop still reads. | | [`_CompletedRollout`](#nemo_gym-rollouts-rollout_collection-_CompletedRollout) | A finished `/run` dispatch, with timing carried alongside (not inside) the raw result. | | [`_EagerCompletionIterator`](#nemo_gym-rollouts-rollout_collection-_EagerCompletionIterator) | Start every rollout in input order on first use, yielding completion-order awaitables. | | [`_EagerCompletionQueue`](#nemo_gym-rollouts-rollout_collection-_EagerCompletionQueue) | Own undelivered outcomes independently of the iterator and its handed-out awaitables. | | [`_RowResultIterator`](#nemo_gym-rollouts-rollout_collection-_RowResultIterator) | Map completion metadata to `(row, result)` for the direct collection APIs. | | [`_RunRecordWriter`](#nemo_gym-rollouts-rollout_collection-_RunRecordWriter) | Writes finished rollouts into a run folder the way `ng_collect_rollouts` lays it out, one shard at a time. | ### Functions | Name | Description | | ------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------- | | [`__getattr__`](#nemo_gym-rollouts-rollout_collection-__getattr__) | - | | [`_agent_request_failure_row`](#nemo_gym-rollouts-rollout_collection-_agent_request_failure_row) | One sidecar row for a `/run` call that came back without a result. | | [`_aggregation_error_entry`](#nemo_gym-rollouts-rollout_collection-_aggregation_error_entry) | Build the metrics-file placeholder for one agent whose aggregation failed. | | [`_batch_status_row`](#nemo_gym-rollouts-rollout_collection-_batch_status_row) | Keep status identities, failure classes and kinds without retaining rollout payloads. | | [`_collected_outcome`](#nemo_gym-rollouts-rollout_collection-_collected_outcome) | Build the record `ng_collect_rollouts` stores for a finished dispatch, success or failure. | | [`_collection_failure_report`](#nemo_gym-rollouts-rollout_collection-_collection_failure_report) | Collection's own failure for a failed `/run` dispatch, caused by the dispatch error, with the report the | | [`_count_lines`](#nemo_gym-rollouts-rollout_collection-_count_lines) | - | | [`_counted_failure_row`](#nemo_gym-rollouts-rollout_collection-_counted_failure_row) | The metric-input copy of a sidecar row counted as zero. | | [`_counted_failure_rows`](#nemo_gym-rollouts-rollout_collection-_counted_failure_rows) | - | | [`_counts_as_zero`](#nemo_gym-rollouts-rollout_collection-_counts_as_zero) | Whether a rollout's latest sidecar row enters the metrics as a zero: its class is wanted, or it is marked. | | [`_coverage_report`](#nemo_gym-rollouts-rollout_collection-_coverage_report) | State how much of the input the score covers, for the runs where it is not all of it. | | [`_credentials_rejected_stop`](#nemo_gym-rollouts-rollout_collection-_credentials_rejected_stop) | The end of a run whose provider (sandbox, or the grading API) refused its credentials: no attempt succeeds | | [`_current_span_is_valid`](#nemo_gym-rollouts-rollout_collection-_current_span_is_valid) | - | | [`_dataset_admitted_agents`](#nemo_gym-rollouts-rollout_collection-_dataset_admitted_agents) | `(dataset name, allowed_agents)` for each dataset entry that declares the dispatched rows with a non-empty | | [`_datasets_declaring_taskset`](#nemo_gym-rollouts-rollout_collection-_datasets_declaring_taskset) | The entries that declare the rows of `taskset`, on whichever agent block declares them: an entry whose | | [`_declared_datasets`](#nemo_gym-rollouts-rollout_collection-_declared_datasets) | The `datasets` entries one agent block declares; a list that is missing, a placeholder or not a list | | [`_dispatch_drained_result`](#nemo_gym-rollouts-rollout_collection-_dispatch_drained_result) | The result Gym builds for a row the dispatch budget never started. | | [`_dispatch_span`](#nemo_gym-rollouts-rollout_collection-_dispatch_span) | One rollout's dispatch as the root of its own trace, linked to the run's `job` span. | | [`_enrich_returned_result`](#nemo_gym-rollouts-rollout_collection-_enrich_returned_result) | `enrich_run_result` for a result `run_examples` returns, keeping its public shape. | | [`_environment_server_for_agent`](#nemo_gym-rollouts-rollout_collection-_environment_server_for_agent) | Return the one environment server that fronts an agent. | | [`_environment_server_for_config_row`](#nemo_gym-rollouts-rollout_collection-_environment_server_for_config_row) | Pick the environment server a row is dispatched to, or None for today's agent path. | | [`_environment_servers_by_agent`](#nemo_gym-rollouts-rollout_collection-_environment_servers_by_agent) | Map each agent name to the environment servers that front it. | | [`_episode_record`](#nemo_gym-rollouts-rollout_collection-_episode_record) | Turn a `BaseEpisodeResponse` into the rollout record the collector stores. | | [`_episode_reply_violation`](#nemo_gym-rollouts-rollout_collection-_episode_reply_violation) | The typed failure for a reply that is neither an episode result nor a failure. | | [`_episode_request_body`](#nemo_gym-rollouts-rollout_collection-_episode_request_body) | - | | [`_expand_input_glob`](#nemo_gym-rollouts-rollout_collection-_expand_input_glob) | Expand a glob-or-comma-separated-globs string into a sorted, deduplicated list of paths. | | [`_failure_compatibility_row`](#nemo_gym-rollouts-rollout_collection-_failure_compatibility_row) | Keep existing sidecar routing keys and diagnostics alongside the canonical failure. | | [`_failure_diagnostics`](#nemo_gym-rollouts-rollout_collection-_failure_diagnostics) | Preserve producer diagnostics outside the bounded canonical failure record. | | [`_failure_export_metrics`](#nemo_gym-rollouts-rollout_collection-_failure_export_metrics) | Failure and retry counts for exporters, read from the failures sidecars. | | [`_failure_outcome`](#nemo_gym-rollouts-rollout_collection-_failure_outcome) | - | | [`_failure_rows_counted_as_zero`](#nemo_gym-rollouts-rollout_collection-_failure_rows_counted_as_zero) | Sidecar rows the caller opted to count in the metrics denominator. | | [`_fill_task_fields`](#nemo_gym-rollouts-rollout_collection-_fill_task_fields) | Give the rows the metric input adds, counted failures and imputed zeros, their task's dataset fields. | | [`_final_zero_failure_rows`](#nemo_gym-rollouts-rollout_collection-_final_zero_failure_rows) | The latest failure rows that settle their rollout as a scored zero: final, and marked by their producer. | | [`_identity_key`](#nemo_gym-rollouts-rollout_collection-_identity_key) | A rollout's key across runs, which each number their tasks from 0. | | [`_is_collector_key`](#nemo_gym-rollouts-rollout_collection-_is_collector_key) | Keys rollout collection writes itself; an Environment Server result must not use them. | | [`_is_episode_response`](#nemo_gym-rollouts-rollout_collection-_is_episode_response) | True for a `BaseEpisodeResponse`-shaped reply: object identities plus a `result` or `failure` key. | | [`_judge_failure_response`](#nemo_gym-rollouts-rollout_collection-_judge_failure_response) | Legacy judge_failsafe carries real generation that can be judged again. | | [`_latest_failure_rows`](#nemo_gym-rollouts-rollout_collection-_latest_failure_rows) | The last failed attempt recorded for each rollout across the failures sidecars. | | [`_masking_step_metrics`](#nemo_gym-rollouts-rollout_collection-_masking_step_metrics) | In-progress view of what a run is losing to its environment rather than its policy. | | [`_matches_failure_class`](#nemo_gym-rollouts-rollout_collection-_matches_failure_class) | Match the canonical class or the original legacy producer's routing class. | | [`_materialized_taskset`](#nemo_gym-rollouts-rollout_collection-_materialized_taskset) | - | | [`_metric_failure_kind`](#nemo_gym-rollouts-rollout_collection-_metric_failure_kind) | The kind as a bounded label: registered or namespaced names pass, anything else is one bucket. | | [`_metrics_group`](#nemo_gym-rollouts-rollout_collection-_metrics_group) | The environment server a rollout is scored under, as `_call_aggregate_metrics` groups it. | | [`_missing_rollout_rows_counted_as_zero`](#nemo_gym-rollouts-rollout_collection-_missing_rollout_rows_counted_as_zero) | Materialized rollouts that produced no row anywhere, counted as zeros. | | [`_names_its_task`](#nemo_gym-rollouts-rollout_collection-_names_its_task) | - | | [`_native_episode_record`](#nemo_gym-rollouts-rollout_collection-_native_episode_record) | Validate the dispatched episode identity before translating its outcome. | | [`_normalize_health_check_ignored_checks`](#nemo_gym-rollouts-rollout_collection-_normalize_health_check_ignored_checks) | - | | [`_normalize_rollout_outcome`](#nemo_gym-rollouts-rollout_collection-_normalize_rollout_outcome) | - | | [`_print_on_sigterm`](#nemo_gym-rollouts-rollout_collection-_print_on_sigterm) | Print render() when Slurm's time limit sends SIGTERM, then terminate as before. | | [`_protocol_violation_stop`](#nemo_gym-rollouts-rollout_collection-_protocol_violation_stop) | The end of a run whose environment server has answered every request so far with a reply the collector cannot | | [`_quota_exhausted_stop`](#nemo_gym-rollouts-rollout_collection-_quota_exhausted_stop) | The end of a run whose grading provider's quota is exhausted: nothing can be graded until it resets. | | [`_raise_for_aggregation_errors`](#nemo_gym-rollouts-rollout_collection-_raise_for_aggregation_errors) | Signal failure after partial metrics and any batch status have been saved. | | [`_read_jsonl`](#nemo_gym-rollouts-rollout_collection-_read_jsonl) | Parse a whole JSONL artifact, letting other tasks on the event loop run between rows. | | [`_record_policy_model_server`](#nemo_gym-rollouts-rollout_collection-_record_policy_model_server) | Name the agent's configured policy server on the record (landed or failed), so offline timing needs no | | [`_record_rollout_health`](#nemo_gym-rollouts-rollout_collection-_record_rollout_health) | Evaluate the trajectory health checks on a finished rollout and count the verdict and flagged checks. | | [`_record_rollout_outcome`](#nemo_gym-rollouts-rollout_collection-_record_rollout_outcome) | Count a persisted attempt, including its numeric reward when scored. | | [`_record_rollout_shape`](#nemo_gym-rollouts-rollout_collection-_record_rollout_shape) | Record the size of one completed rollout, while metrics are exporting. | | [`_reflect_enrichment`](#nemo_gym-rollouts-rollout_collection-_reflect_enrichment) | Add a stored record's enrichment and identity keys to the reply `run_examples` returns in its own shape. | | [`_refresh_shard_status_aggregation`](#nemo_gym-rollouts-rollout_collection-_refresh_shard_status_aggregation) | Mark each shard's run status with the outcome of the aggregation that merged it, keeping its counts. | | [`_reject_attempt_qualified_rollout_ids`](#nemo_gym-rollouts-rollout_collection-_reject_attempt_qualified_rollout_ids) | Refuse dataset rows whose explicit rollout ID names an attempt. | | [`_reply_failure_kind`](#nemo_gym-rollouts-rollout_collection-_reply_failure_kind) | The kind of the outermost failure in the chain an agent's failed reply selected that is not a routing kind. | | [`_reply_failure_report`](#nemo_gym-rollouts-rollout_collection-_reply_failure_report) | The `FailureReport` an environment reply carries beside its outcome; an invalid one is omitted. | | [`_restore_sigterm`](#nemo_gym-rollouts-rollout_collection-_restore_sigterm) | - | | [`_rollout_order_key`](#nemo_gym-rollouts-rollout_collection-_rollout_order_key) | Task then repeat: metrics that read a task's repeats positionally need that order. | | [`_rollout_request_debug_summary`](#nemo_gym-rollouts-rollout_collection-_rollout_request_debug_summary) | - | | [`_round_robin_by_agent`](#nemo_gym-rollouts-rollout_collection-_round_robin_by_agent) | Interleave resolved rows by environment server or agent, preserving each input order. | | [`_routing_identity`](#nemo_gym-rollouts-rollout_collection-_routing_identity) | The agent a rollout ran on, or for a materialized taskset row, which names none, its environment server. | | [`_rows_of_first_tasks`](#nemo_gym-rollouts-rollout_collection-_rows_of_first_tasks) | Yield the rows of the first `limit` tasks. Identical rows are one task, as in the task index. | | [`_sample_server_queues`](#nemo_gym-rollouts-rollout_collection-_sample_server_queues) | Poll every vLLM server's /metrics and record requests waiting and running, summed over servers. | | [`_sanitize_legacy_result`](#nemo_gym-rollouts-rollout_collection-_sanitize_legacy_result) | Discard collector bookkeeping while preserving declared legacy evidence. | | [`_shard_record`](#nemo_gym-rollouts-rollout_collection-_shard_record) | The record `ng_collect_rollouts` would store for `completed`, for a `run_examples` call recording a shard. | | [`_shard_rollout_id`](#nemo_gym-rollouts-rollout_collection-_shard_rollout_id) | The rollout id a recorded shard dispatches `row` under: the row's own id, prefixed with the shard's run id. | | [`_span_task_id`](#nemo_gym-rollouts-rollout_collection-_span_task_id) | - | | [`_stamp_collector_fields`](#nemo_gym-rollouts-rollout_collection-_stamp_collector_fields) | Stamp the row's identity and routing fields onto the record collection stores, as every reader expects. | | [`_status_failure_kind`](#nemo_gym-rollouts-rollout_collection-_status_failure_kind) | The failure kind a sidecar row carries: top-level, the legacy diagnostic key, or the typed record's. | | [`_sum_vllm_queue`](#nemo_gym-rollouts-rollout_collection-_sum_vllm_queue) | - | | [`_task_data_provenance`](#nemo_gym-rollouts-rollout_collection-_task_data_provenance) | The `PROVENANCE_KEYS` of a materialized row's `task_data`: provenance stamps for the record. | | [`_truncated_body`](#nemo_gym-rollouts-rollout_collection-_truncated_body) | Decode at most the kept prefix, so a huge error page is never decoded in full. | | [`enrich_run_result`](#nemo_gym-rollouts-rollout_collection-enrich_run_result) | Attach a finished rollout's collected evidence to its `/run` result, in place. | | [`is_non_superseding_rollout_row`](#nemo_gym-rollouts-rollout_collection-is_non_superseding_rollout_row) | Whether readers skip a sidecar row before they pick a rollout's latest attempt. | | [`is_omitted_rollout_row`](#nemo_gym-rollouts-rollout_collection-is_omitted_rollout_row) | Whether a failures-sidecar row is a `record_omitted_rollouts` record of a rollout that produced no result. | | [`is_unstarted_rollout_row`](#nemo_gym-rollouts-rollout_collection-is_unstarted_rollout_row) | Whether a sidecar row records a rollout the dispatch budget never started: no execution, so no history. | | [`loads_jsonl_line`](#nemo_gym-rollouts-rollout_collection-loads_jsonl_line) | Parse one JSONL line, raising a clean `ConfigError` (naming file + line) on malformed JSON. | ### Data [`AGENT_REQUEST_FAILED_FAILURE_CLASS`](#nemo_gym-rollouts-rollout_collection-AGENT_REQUEST_FAILED_FAILURE_CLASS) [`AGENT_RUN_ERROR_FAILURE_CLASS`](#nemo_gym-rollouts-rollout_collection-AGENT_RUN_ERROR_FAILURE_CLASS) [`ENVIRONMENT_SERVER_FAILURE_CLASS`](#nemo_gym-rollouts-rollout_collection-ENVIRONMENT_SERVER_FAILURE_CLASS) [`NG_DATASET_DIGEST_KEY`](#nemo_gym-rollouts-rollout_collection-NG_DATASET_DIGEST_KEY) [`NG_DIGEST_KEY`](#nemo_gym-rollouts-rollout_collection-NG_DIGEST_KEY) [`NG_ELAPSED_KEY`](#nemo_gym-rollouts-rollout_collection-NG_ELAPSED_KEY) [`NG_ENVIRONMENT_SERVER_KEY`](#nemo_gym-rollouts-rollout_collection-NG_ENVIRONMENT_SERVER_KEY) [`NG_FAILURE_COUNTS_AS_ZERO_KEY`](#nemo_gym-rollouts-rollout_collection-NG_FAILURE_COUNTS_AS_ZERO_KEY) [`NG_FAILURE_JUDGE_ERROR_KEY`](#nemo_gym-rollouts-rollout_collection-NG_FAILURE_JUDGE_ERROR_KEY) [`NG_FAILURE_LEGACY_CLASS_KEY`](#nemo_gym-rollouts-rollout_collection-NG_FAILURE_LEGACY_CLASS_KEY) [`NG_FAILURE_MAX_ATTEMPTS_KEY`](#nemo_gym-rollouts-rollout_collection-NG_FAILURE_MAX_ATTEMPTS_KEY) [`NG_FAILURE_REPORT_KEYS`](#nemo_gym-rollouts-rollout_collection-NG_FAILURE_REPORT_KEYS) [`NG_OMITTED_ROLLOUT_KEY`](#nemo_gym-rollouts-rollout_collection-NG_OMITTED_ROLLOUT_KEY) [`NG_POLICY_MODEL_SERVER_KEY`](#nemo_gym-rollouts-rollout_collection-NG_POLICY_MODEL_SERVER_KEY) [`NG_RESULT_TYPE_KEY`](#nemo_gym-rollouts-rollout_collection-NG_RESULT_TYPE_KEY) [`NG_TASK_ID_KEY`](#nemo_gym-rollouts-rollout_collection-NG_TASK_ID_KEY) [`NG_TERMINAL_KEY`](#nemo_gym-rollouts-rollout_collection-NG_TERMINAL_KEY) [`NG_UNROUTED_RUN_ERROR_KEY`](#nemo_gym-rollouts-rollout_collection-NG_UNROUTED_RUN_ERROR_KEY) [`PROVENANCE_KEYS`](#nemo_gym-rollouts-rollout_collection-PROVENANCE_KEYS) [`ROLLOUT_REPEAT_ATTRIBUTE`](#nemo_gym-rollouts-rollout_collection-ROLLOUT_REPEAT_ATTRIBUTE) [`STATUS_FAILURE_KIND_KEY`](#nemo_gym-rollouts-rollout_collection-STATUS_FAILURE_KIND_KEY) [`STATUS_MODEL_FINAL_KEY`](#nemo_gym-rollouts-rollout_collection-STATUS_MODEL_FINAL_KEY) [`TASK_ID_ATTRIBUTE`](#nemo_gym-rollouts-rollout_collection-TASK_ID_ATTRIBUTE) [`_ENRICHMENT_KEYS`](#nemo_gym-rollouts-rollout_collection-_ENRICHMENT_KEYS) [`_LEGACY_RESULT_FIELDS`](#nemo_gym-rollouts-rollout_collection-_LEGACY_RESULT_FIELDS) [`_MAX_FAILURE_BODY_CHARS`](#nemo_gym-rollouts-rollout_collection-_MAX_FAILURE_BODY_CHARS) [`_NO_RESULT_FAILURE_CLASSES`](#nemo_gym-rollouts-rollout_collection-_NO_RESULT_FAILURE_CLASSES) [`_NO_SIGTERM_HOOK`](#nemo_gym-rollouts-rollout_collection-_NO_SIGTERM_HOOK) [`_OWNED_JOB_SPAN_CONTEXT`](#nemo_gym-rollouts-rollout_collection-_OWNED_JOB_SPAN_CONTEXT) [`_READ_JSONL_YIELD_INTERVAL_S`](#nemo_gym-rollouts-rollout_collection-_READ_JSONL_YIELD_INTERVAL_S) [`_RESERVED_SHARD_ENDINGS`](#nemo_gym-rollouts-rollout_collection-_RESERVED_SHARD_ENDINGS) [`_ROUTING_FAILURE_KINDS`](#nemo_gym-rollouts-rollout_collection-_ROUTING_FAILURE_KINDS) [`_RUN_FAILURE_ERRORS`](#nemo_gym-rollouts-rollout_collection-_RUN_FAILURE_ERRORS) [`_SERVER_DID_NOT_RUN_STATUSES`](#nemo_gym-rollouts-rollout_collection-_SERVER_DID_NOT_RUN_STATUSES) [`_SHARD_IDENTITY_KEYS`](#nemo_gym-rollouts-rollout_collection-_SHARD_IDENTITY_KEYS) [`_SHARD_LABEL`](#nemo_gym-rollouts-rollout_collection-_SHARD_LABEL) [`_failures_path_for`](#nemo_gym-rollouts-rollout_collection-_failures_path_for) [`_moved_cli_entry_points`](#nemo_gym-rollouts-rollout_collection-_moved_cli_entry_points) [`_moved_names`](#nemo_gym-rollouts-rollout_collection-_moved_names) [`get_max_rollout_attempts`](#nemo_gym-rollouts-rollout_collection-get_max_rollout_attempts) [`logger`](#nemo_gym-rollouts-rollout_collection-logger) ### API ```python class nemo_gym.rollouts.rollout_collection.E2ERolloutCollectionConfig() ``` **Bases:** [SharedRolloutCollectionConfig](#nemo_gym-rollouts-rollout_collection-SharedRolloutCollectionConfig) Spin up all necessary servers and perform a batch of rollout collection using each dataset inside the provided configs. Examples: ```python gym eval run +output_jsonl_fpath=weather_rollouts.jsonl +num_samples_in_parallel=10 ``` **`reuse_existing_data_preparation`** `bool = False` --- **`split`** `Union[Literal['train'], Literal['validation'], Literal['benchmark']]` --- ```python nemo_gym.rollouts.rollout_collection.E2ERolloutCollectionConfig._reject_example_split( data ) ``` classmethod ```python nemo_gym.rollouts.rollout_collection.E2ERolloutCollectionConfig._reject_input_jsonl_fpath( data ) ``` classmethod ```python class nemo_gym.rollouts.rollout_collection.RolloutAggregationConfig() ``` **Bases:** [RunHealthConfig](/nemo/gym/nemo-gym/nemo_gym/rollouts/health/types#nemo_gym-rollouts-health-types-RunHealthConfig), [BaseNeMoGymCLIConfig](/nemo/gym/nemo-gym/nemo_gym/config/config_types#nemo_gym-config-config_types-BaseNeMoGymCLIConfig) Aggregate metrics across rollout shards produced by `gym eval run --no-serve +disable_aggregation=true`. Reads every JSONL file matching `input_glob`, computes aggregate metrics by POSTing to each agent server's `/aggregate_metrics` endpoint over the global union of records, and writes a single `_aggregate_metrics.json` next to the rollouts. By default also concatenates all shards into `output_jsonl_fpath`. Examples: ```python gym eval aggregate "+config_paths=[benchmarks/aime24/config.yaml,responses_api_models/vllm_model/configs/vllm_model.yaml]" +input_glob='results/rollouts-rs*-chunk*.jsonl' +output_jsonl_fpath=results/rollouts.jsonl ``` **`batch_manifest_fpath`** `Optional[str]` --- **`count_failure_classes_as_zero`** `List[str]` --- **`count_missing_rollouts_as_zero`** `bool` --- **`disable_health_check`** `bool` --- **`health_check_ignored_checks`** `List[str]` --- **`health_check_workers`** `Optional[int]` --- **`input_glob`** `str` --- **`materialized_inputs_jsonl_fpath`** `Optional[str]` --- **`merge_shards`** `bool` --- **`output_jsonl_fpath`** `str` --- **`retry_terminal_timeouts`** `bool` --- ```python nemo_gym.rollouts.rollout_collection.RolloutAggregationConfig._validate_health_check_ignored_checks( value ) ``` classmethod ```python class nemo_gym.rollouts.rollout_collection.RolloutAggregationHelper() ``` **Bases:** `BaseModel` ```python nemo_gym.rollouts.rollout_collection.RolloutAggregationHelper.run_from_config( config: nemo_gym.rollouts.rollout_collection.RolloutAggregationConfig ) -> typing.Optional[pathlib.Path] ``` async ```python class nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig() ``` **Bases:** [SharedRolloutCollectionConfig](#nemo_gym-rollouts-rollout_collection-SharedRolloutCollectionConfig) Perform a batch of rollout collection. Examples: ```python gym eval run --no-serve +agent_name=example_single_tool_call_simple_agent +input_jsonl_fpath=weather_query.jsonl +output_jsonl_fpath=weather_rollouts.jsonl +limit=100 +num_repeats=4 +num_samples_in_parallel=10 ``` **`agent_map`** `Optional[Dict[str, str]]` --- **`agent_name`** `Optional[str]` --- **`allow_unsafe_resume`** `bool` --- **`dispatch_all_started_within_s`** `Optional[float]` --- **`dispatch_budget_s`** `Optional[float]` --- **`dispatch_connect_grace_s`** `float` --- **`dispatch_longest_first`** `bool` --- **`dispatch_start_report_min_fraction`** `float` --- **`dispatch_start_report_within_s`** `Optional[float]` --- **`drain_margin_s`** `Optional[float]` --- **`fan_out`** `Optional[Dict[str, List[str]]]` --- **`input_jsonl_fpath`** `str` --- **`interleave_repeats`** `bool` --- **`limit`** `Optional[int]` --- **`long_rollout_max_fraction`** `float` --- **`long_rollout_s`** `Optional[float]` --- **`materialized_jsonl_fpath`** `Path` --- **`num_repeats`** `Union[int, Dict[str, int]]` --- **`num_repeats_add_seed`** `Union[bool, Dict[str, bool]]` --- **`prompt_config`** `Optional[str]` --- **`record_omitted_rollouts`** `bool` --- **`resubmit`** `bool` --- **`resume_from_cache`** `bool` --- **`retry_terminal_timeouts`** `bool` --- **`server_queue_max_waiting_ratio`** `float` --- **`server_queue_metrics_urls`** `Optional[List[str]]` --- **`server_queue_sample_interval_s`** `float` --- **`skills`** `Optional[SkillsConfig]` --- **`timed_out_max_fraction`** `float` --- **`timing_summary_interval_s`** `Optional[float]` --- ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig._coerce_null_num_repeats( v ) ``` classmethod ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig._fold_agent_name_into_agent_map() -> nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ``` ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig._resubmit_resumes() -> nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ``` ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig._validate_dispatch_concurrency() -> nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ``` ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig._validate_fan_out() -> nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ``` ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig._validate_num_repeats() -> nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ``` ```python class nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper() ``` **Bases:** `BaseModel` Collect rollouts by sending tasks to a running Gym deployment. Each rollout batch fetches the deployment's configuration from the head server. A caller that runs many batches against one deployment can set `reuse_server_client=True`. The helper then fetches the configuration once. Build a new helper after that deployment's configuration changes. **`_cached_head_server_config`** `Optional[BaseServerConfig] = PrivateAttr(default=None)` --- **`_cached_server_client`** `Optional[ServerClient] = PrivateAttr(default=None)` --- **`_collection_started`** `bool = PrivateAttr(default=False)` --- **`_landed_keys`** `set = PrivateAttr(default_factory=set)` --- **`_store`** `RolloutStore | None = PrivateAttr(default=None)` --- **`_termination_requested`** `Callable[[], bool] | None = PrivateAttr(default=None)` --- **`reuse_server_client`** `bool` --- ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._agent_name_for_row( row: dict[str, typing.Any], global_config_dict: omegaconf.DictConfig ) -> str | None ``` staticmethod ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._call_aggregate_metrics( results: typing.List[typing.Dict], rows: typing.List[typing.Dict], output_fpath: pathlib.Path, raise_on_error: bool = True ) -> typing.Optional[pathlib.Path] ``` async Call /aggregate_metrics on the environment server each rollout ran through. Rows are grouped by the environment server stamped on them at preprocessing (`_ng_environment_server`); a row without the stamp is grouped by the environment server that fronts its agent, as before, so the identity decided at dispatch is the one aggregation uses, across shards and resumed runs alike. Writes a single \_aggregate_metrics.json with one entry per environment server (same shape as the old \_agent_metrics.json, plus the server name). Returns the file path. Failed requests retain an entry with empty metric collections and `aggregation_error`. By default, re-raise the first original exception after saving partial metrics, preserving its type and traceback for direct callers such as custom evaluation drivers. Collection and aggregation entrypoints pass `raise_on_error=False` to write batch status first, then call `_raise_for_aggregation_errors` for a summary error. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._dispatch_name( row: dict[str, typing.Any] ) -> str ``` staticmethod ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._load_from_cache( config: nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig, retain_results_in_memory: bool = True, success_keys: typing.Optional[set] = None ) -> typing.Tuple[typing.List[typing.Dict], typing.List[typing.Dict], typing.List[typing.Dict], typing.List[typing.List[bytes]]] ``` ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._preprocess_raw_rows( raw_rows: typing.List[typing.Tuple[int, str, typing.Dict]], config: nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ) -> typing.List[typing.Dict] ``` staticmethod ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._preprocess_rows_from_config( config: nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ) -> typing.List[typing.Dict] ``` ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._run_examples_with_metadata( examples: typing.List[typing.Dict], head_server_config: typing.Optional[nemo_gym.config.config_types.BaseServerConfig] = None, semaphore: typing.Optional[asyncio.Semaphore] = None, route_failures_to_sidecar: bool = False, environment_server_name: str | None = None, max_resident_tasks: typing.Optional[int] = None, typed_outcomes: bool = False, validate_results: bool = False, on_dispatch: typing.Optional[typing.Callable[[Dict], None]] = None, on_dispatch_batch: typing.Optional[typing.Callable[[List[Dict]], None]] = None, interleave_by_agent: bool = False, dispatch_budget_s: typing.Optional[float] = None, drain_margin_s: typing.Optional[float] = None, latency_tracker: typing.Optional[nemo_gym.rollouts.rollout_perf.DispatchLatencyTracker] = None, retry_requests: bool = True, _max_num_tries: typing.Optional[int] = None, run_timing: typing.Optional[nemo_gym.rollouts.rollout_collection.RunTimingReport] = None, record_unrouted_failures: bool = False, _connect_grace_s: typing.Optional[float] = None ) -> typing.Iterator[asyncio.Future] ``` Internal dispatch shared by `run_examples` and Gym's own collection paths. `record_unrouted_failures` returns a failed `/run` that `route_failures_to_sidecar` would route as a failure row whose `run_error` holds the error, so the collection writer can keep evidence of it before it raises the error. Other callers keep receiving the exception. When `max_resident_tasks` is set, at most that many rollout tasks are admitted at once. When unset, all examples are scheduled as before. The collection owner closes either iterator on completion, cancellation or error. `dispatch_budget_s` stops starting rows that many seconds after this call, and `drain_margin_s` stops sooner for rows that would not have time to finish. A row drained this way resolves to a Gym-built `_dispatch_drained_result`. Identical contract to `run_examples`, but each future resolves to a `_CompletedRollout` that carries `rollout_latency_ms` alongside the raw `/run` result instead of inside it, so internal-only timing never has to be smuggled through (and stripped back out of) a dict that a direct caller of `run_examples` could also observe. `interleave_by_agent` is enabled by the full evaluation path. Lower-level callers retain the order they supplied. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._run_from_config( config: nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ) -> typing.Tuple[typing.List[typing.Dict]] ``` async ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._run_locked_from_config( config: nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ) -> typing.Tuple[typing.List[typing.Dict]] ``` async ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._stamp_environment_server_agent_refs( examples: list[dict[str, typing.Any]], global_config_dict: omegaconf.DictConfig ) -> None ``` classmethod Stamp compatibility-routed rows with the bound agent. These rows never reach `resolve_task_sources`, so without this they carry no `agent_ref` and results, aggregate metrics and reward profiling lose the agent they ran on. A row that already names an agent is left alone; the name is validated against the environment server by `_validate_environment_servers`. Materialized tasks are skipped: they carry no agent by design, and their result projection is not the legacy shape this key belongs to. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._validate_agent_names( examples: typing.List[typing.Dict], global_config_dict: omegaconf.DictConfig ) -> None ``` staticmethod Fail before any dispatch when a row names an agent absent from the running config. Without this, the first bad row dies mid-collection with a raw omegaconf ConfigKeyError after valid rows have already been dispatched. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._validate_agent_pairings( examples: typing.List[typing.Dict], global_config_dict: omegaconf.DictConfig ) -> None ``` staticmethod Fail before dispatch when a row points to agent incompatible with the resources server it runs on. The verifier's `allowed_agents` is read from the row's `task_source` and from the resources server the agent binds; the dataset entry that declares the rows may also name the harnesses it admits (a Harbor dataset's `[gym] allowed_agents`, carried into its generated `datasets:` entry, whose rows route by taskset to an environment server that binds no resources server). That list is checked against the agent's own type, so a run config edited to another harness after `gym eval run` admitted one is refused here as well. A materialized row is matched to its entry by taskset, wherever the entry is declared; a flat row meets every entry the agent declares (see `_dataset_admitted_agents`). ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper._validate_environment_servers( examples: list[dict], global_config_dict: omegaconf.DictConfig ) -> None ``` classmethod ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper.preprocess_examples( examples: typing.List[typing.Dict], agent_map: typing.Optional[typing.Dict[str, str]] = None, fan_out: typing.Optional[typing.Dict[str, typing.List[str]]] = None, num_repeats: typing.Union[int, typing.Dict[str, int]] = 1, num_repeats_add_seed: typing.Union[bool, typing.Dict[str, bool]] = False, global_config_dict: typing.Optional[omegaconf.DictConfig] = None, task_ids: typing.Optional[collections.abc.Sequence[str]] = None ) -> typing.List[typing.Dict] ``` Apply run-level routing and repetition to caller-held rows. Public entry point for direct `run_examples` callers (e.g. trainer integrations that drive dispatch themselves): `run_examples` resolves task_sources and validates agent names, but `agent_map`, `fan_out` and `num_repeats` are applied only during preprocessing. Call this first, then pass the returned rows to `run_examples`. Pass `global_config_dict` (the merged config) to also resolve task_source-only rows to their agents here; leave it None to defer that to `run_examples`, which does it against the head server's config. Pass `task_ids` to keep only those tasks, selected by the id stored in each row (see `nemo_gym.data.task_selection.select_tasks`), each task once before repetition. Input rows are not mutated; the expanded, stamped copies are returned. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper.resolve_task_sources( examples: typing.List[typing.Dict], global_config_dict: omegaconf.DictConfig ) -> None ``` staticmethod Stamp an agent_ref onto every row that carries only a task_source. task_source names the config instance that declared the row's dataset. Resolution is `resolve_dataset_agent` — the same rules benchmark discovery uses, so dispatch can never disagree with the listing. Conflicting `agent:` pins across one instance's datasets are a hard error (rows carry only the instance name), as are unknown/non-routable instances; +agent_map is the disambiguator. Rows that already have an agent_ref are left untouched, so this is a no-op on legacy datasets and on already-resolved (materialized) rows. Runs before any dispatch. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper.run_examples( examples: typing.List[typing.Dict], head_server_config: typing.Optional[nemo_gym.config.config_types.BaseServerConfig] = None, semaphore: typing.Optional[asyncio.Semaphore] = None, route_failures_to_sidecar: bool = False, environment_server_name: str | None = None, max_resident_tasks: typing.Optional[int] = None, dispatch_budget_s: typing.Optional[float] = None, drain_margin_s: typing.Optional[float] = None, latency_tracker: typing.Optional[nemo_gym.rollouts.rollout_perf.DispatchLatencyTracker] = None, retry_requests: bool = True, enrich_results: bool = False, include_model_call_payloads: bool = True, delete_model_call_capture: bool = False, record_output_dir: typing.Optional[pathlib.Path | str] = None, record_shard: typing.Optional[str] = None, record_omitted_rollouts: bool = False ) -> nemo_gym.rollouts.rollout_collection._RowResultIterator ``` We provide this function as a lower level interface for running rollout collection. Rows are dispatched as given: task_sources are resolved and agent names validated here, but run-level knobs (`agent_map`, `fan_out`, `num_repeats`) are NOT applied — call `preprocess_examples` first if you need them. `route_failures_to_sidecar` makes a failed `/run` a failure row instead of an exception that ends every rollout still in flight. It defaults off because those rollouts then leave the score. `max_resident_tasks` limits admitted tasks and therefore concurrent requests, even when `semaphore` allows more. Admission starts when the first returned awaitable is awaited. None schedules all examples up front, as with `asyncio.as_completed`. To stop early, `await` the returned iterator's `aclose()`: it cancels unfinished rollouts and waits for their client tasks in this process to end. Cancelling an in-flight `/run` closes its connection; a server with `ClientDisconnectCancellationMiddleware` subsequently cancels its handler. Server/agent cleanup is not awaited or acknowledged, depends on the agent, and may continue after return. Sandbox release is not guaranteed. Handed-out awaitables may return completed results or raise their exceptions, including `CancelledError`; bounded awaitables first awaited after closure raise `CancelledError`. The error propagates in the awaiting caller's task even without cancelling that task, and `asyncio.gather` and `TaskGroup` treat it as cancellation. Without it, abandoned rollouts keep running to completion in the background. To preserve completed but unconsumed work when stopping, use `await completions.aclose_and_drain()` instead. It stops admission before suspending, cancels and awaits unfinished client tasks, and returns settled `(row, result)` futures, including failures. Consume those futures and any outcome already claimed by an outstanding completion waiter; an unclaimed waiter raises `CancelledError`. Repeated drains return an empty list. This also works with bounded admission without starting more rows, and has the same server/agent cleanup limits as `aclose()`. Use `finally` to close the iterator if collection stops early: ```python completions = helper.run_examples(examples) try: for future in completions: row, result = await future if should_stop(row, result): break finally: await completions.aclose() ``` `dispatch_budget_s` stops starting rows that many seconds after this call, and `drain_margin_s` stops sooner for rows that would not have time to finish. Rows start in the order given, once the returned iterator is first consumed. A row that ran resolves to exactly the `(row, result)` pair Gym's own `/run` endpoint returned — no Gym-private fields are added to `result`. A row that produced no `/run` result resolves to a Gym-built `result` carrying Gym-private fields instead: a row drained by the dispatch budget gets `_ng_failure_class="cancelled"` with the `_ng_dispatch_drained` and `_ng_no_persist` markers, and a failed `/run` under `route_failures_to_sidecar` gets a failure row with `_ng_failure_*` fields. `retry_requests=False` disables automatic HTTP replay of a mutating `/run`. Connections that were never established can still be retried safely. It does not convert failures to successful result rows. `enrich_results` runs `enrich_run_result` on each result before it is returned, so the caller receives what `ng_collect_rollouts` records: the row's rollout identity, the captured model calls (`ng_model_call_capture`, with request and response bodies unless `include_model_call_payloads` is false), the standardized `ng_trajectory` and `ng_perf`. Model calls are captured only when the global config enables `observability_enabled`. `delete_model_call_capture` removes each rollout's capture file after it is merged, and needs the payloads kept. With `enrich_results` the rows are dispatched as collection dispatches them, on copies that name this call's run: results are validated, and under `route_failures_to_sidecar` a failed `/run` comes back as the typed `RolloutFailureRecord`, as `run_outcomes` returns it; rows without a rollout identity keep the plain dispatch. With `enrich_results` off, results are exactly what `/run` returned. `record_output_dir` with `record_shard` also writes each rollout to that run folder through `_RunRecordWriter`, as `ng_collect_rollouts` would: `rollouts-.jsonl`, its failures sidecar with the typed failure rows collection writes, the rows as the shard's materialized inputs, its run manifest and its saved progress for `gym eval status` (omitted-rollout rows with `record_omitted_rollouts`). It implies `enrich_results`, dispatches as collection does (typed failure records under `route_failures_to_sidecar`, validated results) on copies of the rows that name the shard's run, and the Store, the run health check and `gym eval aggregate` read the shard like an eval run. Each copy's rollout id is the row's own id (its `_ng_rollout_id`, or `-`) prefixed with the shard's run id, so the shard's model calls are captured under keys of its own; a caller that looks a capture up reads its key from the returned row (`maybe_rollout_id_from_run_body`). `gym eval run` with `+input_jsonl_fpath=/rollouts-_materialized_inputs.jsonl +output_jsonl_fpath=/rollouts-.jsonl +resume_from_cache=true` resumes a shard and runs only what it still lacks. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper.run_from_config( config: nemo_gym.rollouts.rollout_collection.RolloutCollectionConfig ) -> typing.Tuple[typing.List[typing.Dict]] ``` async Collect rollouts for a whole config. Wrapped in the run-scoped `job` span. This is the driver side of an evaluation run and the outermost span Gym produces. Rollouts are not its descendants: each dispatch opens its own trace, linked to this span (see `_dispatch_span`), so a run's traces stay bounded by one rollout each while `job` keeps the run's own lifecycle. `job` is in the `default` preset and not in `per_rollout`. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper.run_outcomes( examples: typing.List[typing.Dict], head_server_config: typing.Optional[nemo_gym.config.config_types.BaseServerConfig] = None, semaphore: typing.Optional[asyncio.Semaphore] = None, run_id: str | None = None ) -> nemo_gym.rollouts.rollout_collection._RowResultIterator ``` Yield `(row, completed_dict | RolloutFailureRecord)` in completion order. Expected HTTP, JSON and result-shape failures become bounded records. Programming errors and cancellation still propagate. Call `preprocess_examples` first, or provide stable explicit rollout ids. This does not persist or retry work; a direct caller owns those decisions. Supply `run_id` to associate direct calls with a run, overriding any row stamp. Without it, existing row stamps are preserved; unstamped rows share a new invocation ID. Returned rows are invocation-owned copies; input rows are not modified. `run_examples` retains its behavior. On early stopping, await the iterator's `aclose()` to cancel and join unfinished client tasks, or `aclose_and_drain()` to also retrieve settled, unconsumed outcomes. Both have the same server/agent cleanup limitations documented by `run_examples`. ```python nemo_gym.rollouts.rollout_collection.RolloutCollectionHelper.setup_server_client( head_server_config: typing.Optional[nemo_gym.config.config_types.BaseServerConfig] = None ) -> nemo_gym.client.server_client.ServerClient ``` Return a client for the head server, reusing the cached one when the helper opted in. Without opt-in, every call fetches the configuration from the head server again. An opted-in helper keeps one client and fetches again only when the head config differs from the cached one. Build a new helper after the configuration behind the same head changes. ```python class nemo_gym.rollouts.rollout_collection.RunTimingReport( total: int, start_within_s: float, start_min_fraction: float, all_started_within_s: typing.Optional[float], long_rollout_s: typing.Optional[float], long_rollout_max_fraction: float, timed_out_max_fraction: float, max_waiting_ratio: typing.Optional[float] = None ) ``` Health table for a collection whose rollouts have a per-rollout wall-clock limit. A per-rollout limit (e.g. BrowseComp's `rollout_timeout_s`) makes the score depend on the deployment as well as the model: if the servers are too few or too slow, rollouts start late, queue for the GPU, or run into the limit, and are scored 0 for reasons unrelated to the model. This table shows whether a run stayed clear of that, so a score from a starved run is not mistaken for a model result. Each row compares this run with a healthy bound: * start coverage: rollouts started within `start_within_s` and all started by `all_started_within_s` (late starts eat into a fixed job allocation); * rollouts ended by the wall-clock limit (above the bound, the limit is shaping the score); * rollouts longer than `long_rollout_s` (an early sign that the limit is too tight); * summed waiting / running requests on the model servers, when their URLs are given (requests queue instead of being served). Diagnostic only: it prints a warning per unhealthy row and never stops the run. **`_all_started_within_s`** --- **`_durations`** `List[float] = []` --- **`_peak_ratio`** `= 0.0` --- **`_queue_samples`** `= 0` --- **`_running_sum`** `= 0.0` --- **`_start_offsets`** `List[float] = []` --- **`_t0`** `= time.monotonic()` --- **`_timed_out`** `= 0` --- **`_waiting_sum`** `= 0.0` --- ```python nemo_gym.rollouts.rollout_collection.RunTimingReport.record_queue( waiting: float, running: float ) -> None ``` One sample of requests waiting and running, summed over every server. ```python nemo_gym.rollouts.rollout_collection.RunTimingReport.record_result( result: collections.abc.Mapping[str, typing.Any], seconds: float ) -> None ``` ```python nemo_gym.rollouts.rollout_collection.RunTimingReport.record_start( now: float ) -> None ``` ```python nemo_gym.rollouts.rollout_collection.RunTimingReport.table() -> str ``` ```python class nemo_gym.rollouts.rollout_collection.SharedRolloutCollectionConfig() ``` **Bases:** [RemovedRolloutOptionsMixin](/nemo/gym/nemo-gym/nemo_gym/config/config_types#nemo_gym-config-config_types-RemovedRolloutOptionsMixin), [UploadRolloutsConfigMixin](/nemo/gym/nemo-gym/nemo_gym/config/config_types#nemo_gym-config-config_types-UploadRolloutsConfigMixin), [RunHealthConfig](/nemo/gym/nemo-gym/nemo_gym/rollouts/health/types#nemo_gym-rollouts-health-types-RunHealthConfig), [BaseNeMoGymCLIConfig](/nemo/gym/nemo-gym/nemo_gym/config/config_types#nemo_gym-config-config_types-BaseNeMoGymCLIConfig) **`batch_manifest_fpath`** `Optional[str]` --- **`count_failure_classes_as_zero`** `List[str]` --- **`count_missing_rollouts_as_zero`** `bool` --- **`disable_aggregation`** `bool` --- **`disable_health_check`** `bool` --- **`environment_protocol_violation_stop_after`** `int` --- **`environment_routing_mode`** `Literal['agent', 'legacy', 'taskset']` --- **`environment_server_name`** `str | None` --- **`environment_server_routes`** `dict[str, str]` --- **`health_check_ignored_checks`** `List[str]` --- **`health_check_workers`** `Optional[int]` --- **`max_resident_rollout_tasks`** `Optional[int]` --- **`num_samples_in_parallel`** `Optional[int]` --- **`output_jsonl_fpath`** `str` --- **`require_complete`** `bool` --- **`responses_create_params`** `Dict[str, Any]` --- **`retain_results_in_memory`** `bool` --- **`rollout_collection_driver`** `Optional[str]` --- **`route_failures_to_sidecar`** `bool` --- ```python nemo_gym.rollouts.rollout_collection.SharedRolloutCollectionConfig._validate_health_check_ignored_checks( value ) ``` classmethod ```python nemo_gym.rollouts.rollout_collection.SharedRolloutCollectionConfig.check_completion( expected: int, results: typing.List[typing.Dict[str, typing.Any]], completed: int | None = None, intentionally_omitted: int = 0, retryable: bool, made_progress: bool, final_zero_rows: collections.abc.Sequence[collections.abc.Mapping[str, typing.Any]] = () ) -> None ``` Raise the strict run's rejection, if any; see `completion_rejection` for the arguments. ```python nemo_gym.rollouts.rollout_collection.SharedRolloutCollectionConfig.completion_rejection( expected: int, results: typing.List[typing.Dict[str, typing.Any]], completed: int | None = None, intentionally_omitted: int = 0, retryable: bool, made_progress: bool, final_zero_rows: collections.abc.Sequence[collections.abc.Mapping[str, typing.Any]] = () ) -> nemo_gym.rollouts.rollout_recovery.IncompleteEvaluationError | None ``` The error that rejects unfinished work, or None; omissions alongside completed tasks are allowed. Collection decides this before it computes the aggregate, so a rejected run publishes no score. Drivers must supply whether any tasks can still be retried and whether this invocation saved outcomes that made progress (not just interruptions). Assuming either can cause endless automatic restarts when a driver has no work left or makes no progress. `final_zero_rows` are failure rows that settle their rollout as a scored zero (see `_final_zero_failure_rows`): resume does not run them again, so they are completed samples. A caller that passes `completed` has already counted them. ```python nemo_gym.rollouts.rollout_collection.SharedRolloutCollectionConfig.validate_environment_routing() -> nemo_gym.rollouts.rollout_collection.SharedRolloutCollectionConfig ``` ```python class nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator( awaitables: typing.Iterator, max_resident_tasks: int, total: int ) ``` Completion-order iterator with a bounded set of resident asyncio tasks. **`_awaitables`** `= iter(awaitables)` --- **`_lock`** `= asyncio.Lock()` --- **`_pending`** `set[Task] = set()` --- **`_progress`** --- **`_ready`** `list[Task] = []` --- **`_resident_task_count`** `int` --- ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator.__iter__() ``` ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator.__next__() ``` ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator._admit() -> bool ``` ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator._close( drain: bool ) -> list[asyncio.Future] ``` async ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator._fill() -> None ``` ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator._next_completed() ``` async ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator.aclose() -> None ``` async ```python nemo_gym.rollouts.rollout_collection._BoundedCompletionIterator.aclose_and_drain() -> list[asyncio.Future] ``` async ```python class nemo_gym.rollouts.rollout_collection._CollectedOutcome( record: typing.Dict[str, typing.Any], structured_failure: bool, diagnostics: typing.Dict[str, typing.Any], metric_result: typing.Dict[str, typing.Any], verification_response: typing.Optional[typing.Dict[str, typing.Any]] ) ``` Dataclass A finished dispatch as collection stores it, with the facts the collector's loop still reads. **`diagnostics`** `Dict[str, Any]` --- **`metric_result`** `Dict[str, Any]` --- **`record`** `Dict[str, Any]` --- **`structured_failure`** `bool` --- **`verification_response`** `Optional[Dict[str, Any]]` --- ```python class nemo_gym.rollouts.rollout_collection._CompletedRollout( row: typing.Dict[str, typing.Any], result: typing.Dict[str, typing.Any] | nemo_gym.rollouts.rollout_outcomes.RolloutFailureRecord, rollout_latency_ms: typing.Optional[float], verification_response: typing.Optional[typing.Dict[str, typing.Any]] = None, diagnostics: typing.Optional[typing.Dict[str, typing.Any]] = None, environment_server: typing.Optional[str] = None, environment_server_type: typing.Optional[str] = None, elapsed_s: typing.Optional[float] = None, dispatch_span_context: typing.Optional[opentelemetry.trace.SpanContext] = None, failure_report: typing.Optional[nemo_gym.core.failure_records.FailureReport] = None, run_error: typing.Optional[BaseException] = None ) ``` Dataclass A finished `/run` dispatch, with timing carried alongside (not inside) the raw result. **`diagnostics`** `Optional[Dict[str, Any]] = None` --- **`dispatch_span_context`** `Optional[SpanContext] = None` --- **`elapsed_s`** `Optional[float] = None` --- **`environment_server`** `Optional[str] = None` --- **`environment_server_type`** `Optional[str] = None` --- **`failure_report`** `Optional[FailureReport] = None` --- **`result`** `Dict[str, Any] | RolloutFailureRecord` --- **`rollout_latency_ms`** `Optional[float]` --- **`row`** `Dict[str, Any]` --- **`run_error`** `Optional[BaseException] = None` --- **`verification_response`** `Optional[Dict[str, Any]] = None` --- ```python class nemo_gym.rollouts.rollout_collection._EagerCompletionIterator( awaitables: typing.Iterator, total: int ) ``` Start every rollout in input order on first use, yielding completion-order awaitables. **`_queue`** `Optional[_EagerCompletionQueue] = None` --- **`_tasks`** `set[Future] = set()` --- ```python nemo_gym.rollouts.rollout_collection._EagerCompletionIterator.__iter__() ``` ```python nemo_gym.rollouts.rollout_collection._EagerCompletionIterator.__next__() ``` ```python nemo_gym.rollouts.rollout_collection._EagerCompletionIterator._close( drain: bool ) -> list[asyncio.Future] ``` async ```python nemo_gym.rollouts.rollout_collection._EagerCompletionIterator._on_done( queue: nemo_gym.rollouts.rollout_collection._EagerCompletionQueue, task: asyncio.Future ) -> None ``` ```python nemo_gym.rollouts.rollout_collection._EagerCompletionIterator.aclose() -> None ``` async ```python nemo_gym.rollouts.rollout_collection._EagerCompletionIterator.aclose_and_drain() -> list[asyncio.Future] ``` async ```python class nemo_gym.rollouts.rollout_collection._EagerCompletionQueue( progress ) ``` Own undelivered outcomes independently of the iterator and its handed-out awaitables. **`ready`** `Queue[Future] = asyncio.Queue()` --- ```python nemo_gym.rollouts.rollout_collection._EagerCompletionQueue.drain() -> list[asyncio.Future] ``` ```python nemo_gym.rollouts.rollout_collection._EagerCompletionQueue.get() ``` async ```python nemo_gym.rollouts.rollout_collection._EagerCompletionQueue.put( task: asyncio.Future ) -> None ``` ```python class nemo_gym.rollouts.rollout_collection._RowResultIterator( completions: typing.Iterator[asyncio.Future], enrich: typing.Optional[typing.Callable[[_CompletedRollout], typing.Awaitable[typing.Any]]] = None, on_close: typing.Optional[typing.Callable[[], None]] = None ) ``` Map completion metadata to `(row, result)` for the direct collection APIs. A thin map over the internal completion iterator that keeps its `aclose()` reachable, so a direct caller can stop collecting and cancel the rollouts still in flight. ```python nemo_gym.rollouts.rollout_collection._RowResultIterator.__iter__() ``` ```python nemo_gym.rollouts.rollout_collection._RowResultIterator.__next__() ``` ```python nemo_gym.rollouts.rollout_collection._RowResultIterator._finish() -> None ``` Run the close hook once, after the outcomes a close still delivers have been enriched. ```python nemo_gym.rollouts.rollout_collection._RowResultIterator._without_metadata( future: asyncio.Future ) -> typing.Tuple[typing.Dict, typing.Dict | nemo_gym.rollouts.rollout_outcomes.RolloutFailureRecord] ``` async ```python nemo_gym.rollouts.rollout_collection._RowResultIterator.aclose() -> None ``` async Cancel unfinished rollouts and await the end of their client tasks in this process. Cancelling an in-flight `/run` request closes its connection. A server using `ClientDisconnectCancellationMiddleware` subsequently cancels its handler. This method does not wait for or acknowledge server/agent cleanup, which depends on the agent and may continue after return; it does not guarantee sandbox release or retire an attempt. A handed-out awaitable may return an already completed result or raise its exception, including `CancelledError` for cancelled work. In bounded mode, an awaitable first awaited after closure raises `CancelledError` even if a result was ready. This error propagates in the awaiting caller's task without a call to that task's `cancel()`; `asyncio.gather` and `TaskGroup` treat it as cancellation. Iteration stops after closure. Safe to call again after closure completes, or before first use (no rollouts start). ```python nemo_gym.rollouts.rollout_collection._RowResultIterator.aclose_and_drain() -> list[asyncio.Future[typing.Tuple[typing.Dict, typing.Dict | nemo_gym.rollouts.rollout_outcomes.RolloutFailureRecord]]] ``` async Stop admission, cancel unfinished client tasks, and preserve completed outcomes. Before its first suspension, this operation stops admitting rollouts and snapshots completed outcomes not yet claimed by a completion awaitable. It then cancels and awaits unfinished client tasks, and returns settled futures yielding `(row, result)` or raising the original exception. No new rollouts start while draining, including in bounded mode. Outcomes are transferred once; repeated calls return an empty list. A handed-out awaitable that already claimed an outcome retains it. Other issued awaitables raise `CancelledError`. Callers with an outstanding waiter must also await that waiter and consume its outcome if it was not cancelled. The returned futures have their exceptions retrieved even if abandoned; awaiting them still raises. An outcome already cancelled at the snapshot is returned as a cancelled future and raises `CancelledError` when awaited. Tasks cancelled by this operation are excluded. Calling this before first use starts nothing. Calling it after `aclose()` returns nothing. Like `aclose()`, it waits only for local client tasks, not server/agent cleanup, and does not guarantee sandbox release. ```python class nemo_gym.rollouts.rollout_collection._RunRecordWriter( record_output_dir: pathlib.Path | str, shard: str, record_omitted_rollouts: bool = False, rows: typing.Optional[typing.List[typing.Dict[str, typing.Any]]] = None, run_id: typing.Optional[str] = None, status_rows: typing.Optional[typing.List[typing.Dict[str, typing.Any]]] = None, global_config: typing.Any = None ) ``` Writes finished rollouts into a run folder the way `ng_collect_rollouts` lays it out, one shard at a time. Private: `run_examples(record_output_dir=..., record_shard=...)` is the entry point, and it hands this writer the records the collector's own builder produced. A caller writing rows of its own would bypass that builder. A shard is `/rollouts-.jsonl` with its failures sidecar and, when the rows are given, its materialized inputs, a verified run manifest with every attempt reserved, and its saved progress (`rollouts-_batch_status.json`) beside it, so a folder of shards is what `gym eval aggregate`, the run health check, the Store (including resume and recovery) and EvalRDL's reader already read. The manifest is the one `gym eval run` computes when it resumes the shard (`+input_jsonl_fpath=/rollouts-_materialized_inputs.jsonl +output_jsonl_fpath=/rollouts-.jsonl +resume_from_cache=true`), so that resume runs only what the shard still lacks. The caller picks the shard label (for example a training step and worker); Gym owns the file names. A result goes to the rollouts file, a failed one to the failures sidecar as the typed failure row collection writes, with its attempt cap, and an interrupted or never-started try to the sidecar as an omitted-rollout row only when `record_omitted_rollouts` is set. Opening a shard replaces what an earlier run wrote under the same label, so a replayed step does not duplicate rows. **`_generation`** `= self.rollouts_fpath.stat().st_ino` --- **`_lock_fpath`** --- **`_status`** `Optional[RunStatusTracker] = RunStatusTracker(self.status_fpath, status_rows)` --- **`_status_agent_refs`** `Dict[Tuple[int, int], Dict]` --- **`_status_completed`** `List[Dict] = []` --- **`_status_expected`** `= len(rows) if rows is not None else None` --- **`_status_failures`** `List[Dict] = []` --- **`_status_written`** `= 0` --- **`_writer_token`** `= secrets.token_hex(16)` --- **`failures_fpath`** `= failures_path_for(self.rollouts_fpath)` --- **`manifest`** `= self._resumable_manifest(rows, global_config)` --- **`manifest_fpath`** `= manifest_path_for(self.rollouts_fpath)` --- **`materialized_fpath`** `= materialized_path_for(self.rollouts_fpath)` --- **`rollouts_fpath`** --- **`run_id`** `= run_id or uuid4().hex` --- **`status_fpath`** `= batch_status_path_for(self.rollouts_fpath)` --- ```python nemo_gym.rollouts.rollout_collection._RunRecordWriter._locked() -> typing.Iterator[typing.Any] ``` Hold the shard's lock across processes, so opening and each write never interleave with another's. The lock file is never unlinked or replaced, so every writer of a label locks the same object; it also holds the token of the writer that opened the shard last. ```python nemo_gym.rollouts.rollout_collection._RunRecordWriter._resumable_manifest( rows: typing.List[typing.Dict[str, typing.Any]], global_config: typing.Any ) -> nemo_gym.rollouts.rollout_recovery.RunManifest ``` The manifest `gym eval run` computes when it resumes this shard, with every row's attempt reserved. That resume names the shard's materialized inputs as its input and the shard as its output, and leaves the other collection settings at their defaults: the same source file, the same rows and, with the same servers, the same configuration. Collection numbers each task's rollouts 0, 1, ... in input order, so rows numbered otherwise are recorded too, and that resume refuses them as other materialized inputs. A configuration this call cannot take an identity from still records the shard, whose manifest then verifies its own inputs only, so a resume refuses it and says why. ```python nemo_gym.rollouts.rollout_collection._RunRecordWriter.flush_status() -> None ``` Write the shard's status as it stands, for a caller that stops consuming before every row is in. ```python nemo_gym.rollouts.rollout_collection._RunRecordWriter.write( result: collections.abc.Mapping[str, typing.Any] ) -> typing.Optional[pathlib.Path] ``` Append one finished rollout's record; returns the file written, or None for an omitted row not kept. **Raises:** * `RuntimeError`: Another writer opened the same shard since this one did. ```python nemo_gym.rollouts.rollout_collection.__getattr__( name: str ) -> typing.Any ``` ```python nemo_gym.rollouts.rollout_collection._agent_request_failure_row( exc: BaseException, status: typing.Optional[int], headers: typing.Any = None ) -> typing.Dict[str, typing.Any] ``` One sidecar row for a `/run` call that came back without a result. No reward and no response: an infrastructure failure is not a verifier score of zero, and a placeholder would read as real generation data to token capture, aggregation and trainers. The class says whether the rollout ran. A NeMo Gym server answers 500 when its own handler raises, so any status it answered with means the rollout ran and broke, which is also how a model server rejecting the model's own output arrives here. A gateway status, or no reply to take a status from, says nothing about the rollout. Neither class carries a reward; an evaluation that wants the first counted names it in `count_failure_classes_as_zero`. ```python nemo_gym.rollouts.rollout_collection._aggregation_error_entry( agent_name: str, exc: BaseException, response_status: typing.Optional[int] = None ) -> typing.Dict[str, typing.Any] ``` Build the metrics-file placeholder for one agent whose aggregation failed. ```python nemo_gym.rollouts.rollout_collection._batch_status_row( result: typing.Dict, agent_refs_by_key: typing.Dict[typing.Tuple[int, int], typing.Dict] ) -> typing.Dict ``` Keep status identities, failure classes and kinds without retaining rollout payloads. ```python nemo_gym.rollouts.rollout_collection._collected_outcome( completed: nemo_gym.rollouts.rollout_collection._CompletedRollout, result: typing.Dict[str, typing.Any] | nemo_gym.rollouts.rollout_outcomes.RolloutFailureRecord ) -> nemo_gym.rollouts.rollout_collection._CollectedOutcome ``` Build the record `ng_collect_rollouts` stores for a finished dispatch, success or failure. `result` is `completed.result` after the caller's episode translation and legacy sanitizing. A failure, typed or marked, becomes the failures-sidecar row: the typed record's compatibility fields beside the producer's diagnostics and the judge reply it kept, then the identity and routing stamps and the causal failure report. `_RunRecordWriter` builds its rows here as well, so an RL shard reads like an eval run. ```python nemo_gym.rollouts.rollout_collection._collection_failure_report( exc: BaseException, failure_class: str, headers: typing.Any ) -> nemo_gym.core.failure_records.FailureReport ``` Collection's own failure for a failed `/run` dispatch, caused by the dispatch error, with the report the reply forwarded. Routing classes remain outcome metadata; this diagnostic has a registered collection kind. ```python nemo_gym.rollouts.rollout_collection._count_lines( path: pathlib.Path ) -> int ``` ```python nemo_gym.rollouts.rollout_collection._counted_failure_row( row: typing.Dict[str, typing.Any] ) -> typing.Dict[str, typing.Any] ``` The metric-input copy of a sidecar row counted as zero. ```python nemo_gym.rollouts.rollout_collection._counted_failure_rows( rows: typing.List[typing.Dict], failure_classes: typing.List[str] ) -> typing.List[typing.Dict] ``` ```python nemo_gym.rollouts.rollout_collection._counts_as_zero( row: collections.abc.Mapping[str, typing.Any], wanted: set ) -> bool ``` Whether a rollout's latest sidecar row enters the metrics as a zero: its class is wanted, or it is marked. ```python nemo_gym.rollouts.rollout_collection._coverage_report( expected: int, scored: int, failure_counts: collections.Counter, failures_hint: typing.Any, imputed: int = 0 ) -> str ``` State how much of the input the score covers, for the runs where it is not all of it. Silence here is what makes a partial run look complete, so this reports against the materialized input rather than the rollouts one hop happened to dispatch, and names the rollouts that are in the score only as imputed zeros. ```python nemo_gym.rollouts.rollout_collection._credentials_rejected_stop( failures: collections.abc.Sequence[collections.abc.Mapping[str, typing.Any]] ) -> nemo_gym.rollouts.rollout_recovery.IncompleteEvaluationError | None ``` The end of a run whose provider (sandbox, or the grading API) refused its credentials: no attempt succeeds until they are fixed. ```python nemo_gym.rollouts.rollout_collection._current_span_is_valid() -> bool ``` ```python nemo_gym.rollouts.rollout_collection._dataset_admitted_agents( global_config_dict: omegaconf.DictConfig, agent_config: omegaconf.DictConfig, taskset: str | None ) -> list[tuple[str, list[str]]] ``` `(dataset name, allowed_agents)` for each dataset entry that declares the dispatched rows with a non-empty `allowed_agents`. A materialized row names its dataset through its taskset, so only the entry that declares that taskset counts (`_datasets_declaring_taskset`), not every dataset the executing agent happens to declare: a second, unrelated entry on the agent cannot refuse the rows. The lookup covers every agent block, not the executing agent's alone: a Harbor dataset's entry sits on the agent block `gym eval run` generated for it, and a run config edited to route the taskset's environment server to another agent block afterwards still has to meet the dataset's list. A flat row routed to an agent directly names no dataset, so every dataset that agent declares counts. So do the executing agent's entries that set no `taskset` when no entry anywhere declares the rows' taskset: a hand-written entry may list pre-materialized rows under a name of its own (`collate_samples` accepts such rows in any entry), and those rows are still that agent's dataset, so they are checked rather than none; an entry that sets a `taskset` materializes and routes its rows as that taskset, so one naming another taskset does not hold these rows. A bare string is one entry, as `allowed_agents_for` reads a resources server's. ```python nemo_gym.rollouts.rollout_collection._datasets_declaring_taskset( global_config_dict: omegaconf.DictConfig, taskset: str ) -> list[omegaconf.DictConfig] ``` The entries that declare the rows of `taskset`, on whichever agent block declares them: an entry whose `taskset` is that name, or whose `name` is when it sets none (a Harbor dataset's generated entry, named after its taskset and routed by it). ```python nemo_gym.rollouts.rollout_collection._declared_datasets( agent_config: omegaconf.DictConfig ) -> list[omegaconf.DictConfig] ``` The `datasets` entries one agent block declares; a list that is missing, a placeholder or not a list declares nothing. ```python nemo_gym.rollouts.rollout_collection._dispatch_drained_result( remaining_s: float, margin_s: typing.Optional[float] ) -> typing.Dict[str, typing.Any] ``` The result Gym builds for a row the dispatch budget never started. No row is written anywhere: absence is the resume signal, so the task is re-dispatched intact next allocation instead of being started and killed part-way through. `_ng_dispatch_drained` keeps it out of capture, token finalization, progress metrics and the rollouts upload, since nothing ran. ```python nemo_gym.rollouts.rollout_collection._dispatch_span( row: collections.abc.Mapping[str, typing.Any], dispatch_name: str, server_name: str ) -> typing.Iterator[opentelemetry.trace.SpanContext] ``` One rollout's dispatch as the root of its own trace, linked to the run's `job` span. The `/run` call made inside carries this span as `traceparent`, so the agent's `gym.rollout` span and everything below it join the new trace instead of nesting under `gym.job`. The link, not a parent edge, ties the trace back to the job; both carry the run id as a resource attribute. The identity comes from the collector *row*, not from the posted body: a materialized episode body carries it only as `episode_id`. `dispatch_name` is the value the rollout counters record under the same key, so spans and metrics join on it; `server_name` is the server the call goes to, which differs for a row routed by agent. Only a job the driver opened as its own root is split; under an embedder's trace the dispatch keeps nesting (see `linked_root_span`). Gate on the `rollout` group and the trace exporter before calling. Yields the span's context, which the driver keeps to attach `gym.rollout.health` later. ```python nemo_gym.rollouts.rollout_collection._enrich_returned_result( row: dict[str, typing.Any], result: dict[str, typing.Any], rollout_latency_ms: typing.Optional[float], options: typing.Any = {} ) -> dict[str, typing.Any] ``` async `enrich_run_result` for a result `run_examples` returns, keeping its public shape. An Environment Server reply to a materialized task is a `BaseEpisodeResponse` whose evidence is nested; it is enriched through the record the collector stores, and the enrichment keys are added to the reply's top level. Returns the enriched record. ```python nemo_gym.rollouts.rollout_collection._environment_server_for_agent( agent_name: str, servers_by_agent: collections.abc.Mapping[str, list[str]] ) -> str ``` Return the one environment server that fronts an agent. A row routed by its agent cannot choose between several environment servers. Several servers may still front one agent when every row names its server directly. Materialized tasksets name their environment server through `environment_server_routes`. ```python nemo_gym.rollouts.rollout_collection._environment_server_for_config_row( row: collections.abc.Mapping[str, typing.Any], config: typing.Any ) -> str | None ``` Pick the environment server a row is dispatched to, or None for today's agent path. A materialized task (`task_id.taskset` plus `task_input`) always routes by its taskset: it is already an episode request and no agent-server `/run` accepts it. A flat row follows `environment_routing_mode`: `agent` keeps today's routing (its agent's environment server is resolved at dispatch), `legacy` sends every flat row to `environment_server_name`, and `taskset` refuses flat rows so a run of materialized rows cannot silently pick up legacy input. One batch may therefore hold both kinds of rows in `agent` and `legacy` mode. The chosen server is stamped on the row as `_ng_environment_server` and travels with it through the materialized input file, retries, and results. ```python nemo_gym.rollouts.rollout_collection._environment_servers_by_agent( global_config_dict: omegaconf.DictConfig ) -> dict[str, list[str]] ``` Map each agent name to the environment servers that front it. ```python nemo_gym.rollouts.rollout_collection._episode_record( response: typing.Dict[str, typing.Any] ) -> typing.Dict[str, typing.Any] ``` Turn a `BaseEpisodeResponse` into the rollout record the collector stores. A handled `failure` becomes a failures-sidecar row, so resume retries a non-terminal one and never a terminal one. A `result` is stored as the Environment Server returned it; the collector adds only its own `_ng_*` keys, so any Environment Server type can be collected without the collector knowing its result fields. Every Environment Server type is scored the same way: through the result's top-level `reward`, with optional top-level `reward_components`. A reward nested elsewhere in the result is stored as data, and a result without a top-level `reward` is unscored. A failure's `failure_kind` is kept under that name, next to the `_ng_failure_*` diagnostics: it is the key `_record_rollout_outcome` reads for the metric label and `_counted_failure_row` drops. ```python nemo_gym.rollouts.rollout_collection._episode_reply_violation( response: typing.Any ) -> nemo_gym.core.typed_exceptions.NgFailureEvent ``` The typed failure for a reply that is neither an episode result nor a failure. It names what the reply held (its type and top-level keys, never values), so a producer that answers an episode request with another shape is found from the failure record alone. ```python nemo_gym.rollouts.rollout_collection._episode_request_body( row: collections.abc.Mapping[str, typing.Any] ) -> dict[str, typing.Any] ``` ```python nemo_gym.rollouts.rollout_collection._expand_input_glob( input_glob: str ) -> typing.List[str] ``` Expand a glob-or-comma-separated-globs string into a sorted, deduplicated list of paths. **Examples:** ```python 'results/rollouts.jsonl' -> ['results/rollouts.jsonl'] (if it exists) 'a/*.jsonl, b/*.jsonl' -> matches of both patterns, deduplicated ``` ```python nemo_gym.rollouts.rollout_collection._failure_compatibility_row( failure: nemo_gym.rollouts.rollout_outcomes.RolloutFailureRecord, verification_response: typing.Optional[typing.Dict] = None ) -> typing.Dict ``` Keep existing sidecar routing keys and diagnostics alongside the canonical failure. Remove these aliases when row-oriented consumers and #3180 producers migrate. The canonical record never carries generation. The sidecar envelope retains actual judge input for existing --judge-failed-only readers, without a reward. Other legacy response placeholders are discarded. ```python nemo_gym.rollouts.rollout_collection._failure_diagnostics( result: typing.Any ) -> typing.Dict ``` Preserve producer diagnostics outside the bounded canonical failure record. ```python nemo_gym.rollouts.rollout_collection._failure_export_metrics( failures_fpaths: typing.List[pathlib.Path], identity: typing.Callable[[Mapping[str, Any]], typing.Any], landed: set, histories: collections.abc.Sequence[nemo_gym.rollouts.rollout_store.RolloutStore] = () ) -> typing.Dict[str, int] ``` Failure and retry counts for exporters, read from the failures sidecars. The sidecars hold every failed attempt of the run, across resumes, so these are run totals rather than one job's. `identity` keys a sidecar row the way `landed` keys the rollouts that have a row. `coverage/failed/` counts the rollouts with no row, by the class of their last attempt. `retries/attempts` counts failed attempts, `retries/rollouts_retried` the rollouts dispatched more than once, and `retries/at_cap` the rollouts with no row that used all their attempts. Each rollout is held to the cap its collection recorded on its last attempt (`_ng_failure_max_attempts`), or the current cap for a row written before the cap was recorded. `retries/max_attempts` is that cap, and is left out when the rollouts were collected with different caps. A `record_omitted_rollouts` row is no failed attempt; an interrupted one is still the rollout's latest execution, so an older failure's class is not counted. ```python nemo_gym.rollouts.rollout_collection._failure_outcome( row: typing.Dict, failure: typing.Dict, stage: str ) -> nemo_gym.rollouts.rollout_outcomes.RolloutFailureRecord ``` ```python nemo_gym.rollouts.rollout_collection._failure_rows_counted_as_zero( failures_fpaths: typing.List[pathlib.Path], failure_classes: typing.List[str], scored_keys: set ) -> typing.List[typing.Dict[str, typing.Any]] ``` Sidecar rows the caller opted to count in the metrics denominator. The last attempt of a rollout is the one that stands, so it is selected across every failure class before the wanted classes are picked out. Selecting the other way round would let a stale attempt be counted after a later one landed in a class the caller did not ask for. A row that already carries a `reward` is counted as it stands. A row that carries none records that no rollout happened, so it is counted as a zero here and only here: the score enters the metric input, never the sidecar or the rollouts jsonl, which keeps the artifacts free of a verdict no verifier gave. A rollout that also succeeded is never counted. A row its producer marked with `_ng_failure_counts_as_zero` is counted whatever its class. ```python nemo_gym.rollouts.rollout_collection._fill_task_fields( added_rows: typing.List[typing.Dict[str, typing.Any]], real_rows: typing.List[typing.Dict[str, typing.Any]], materialized_fpaths: typing.List[pathlib.Path], group: typing.Callable[[Mapping[str, Any]], typing.Optional[str]] ) -> None ``` Give the rows the metric input adds, counted failures and imputed zeros, their task's dataset fields. Rows are ordered by repeat, so an added row can come first in its task, and metric hooks read task-level fields such as a subset label or a weight from a task's first rollout. The fields come from the task's materialized row, and only those that real rows of the same group carry: the aggregator averages every number it is handed, so a field no real row reports would become a metric of its own. A field the verifier computes is not in the materialized row, so the added row lacks it. ```python nemo_gym.rollouts.rollout_collection._final_zero_failure_rows( failures_fpaths: typing.List[pathlib.Path], scored_keys: set, retry_terminal_timeouts: bool ) -> typing.List[typing.Dict[str, typing.Any]] ``` The latest failure rows that settle their rollout as a scored zero: final, and marked by their producer. Resume does not run such a rollout again and the metrics count it as 0, so its work is done. A row counted only because `count_failure_classes_as_zero` names its class is not settled: that option scores the metrics, and resume may still run the rollout. ```python nemo_gym.rollouts.rollout_collection._identity_key( row: collections.abc.Mapping[str, typing.Any], group: typing.Callable[[Mapping[str, Any]], typing.Optional[str]] ) -> tuple ``` A rollout's key across runs, which each number their tasks from 0. ```python nemo_gym.rollouts.rollout_collection._is_collector_key( key: str ) -> bool ``` Keys rollout collection writes itself; an Environment Server result must not use them. `ng_trajectory` is not one of them: it is the producer's slot, where an Environment Server forwards the harness's own turns from its session close (as an agent's `/run` result does), and the collector replaces it with the canonical projection built from those turns. ```python nemo_gym.rollouts.rollout_collection._is_episode_response( result: typing.Any ) -> bool ``` True for a `BaseEpisodeResponse`-shaped reply: object identities plus a `result` or `failure` key. The collector only applies this to a row it dispatched as an episode request (`_materialized_taskset(row)`), so an agent's verify response that echoes identity fields is left alone. ```python nemo_gym.rollouts.rollout_collection._judge_failure_response( result: typing.Any ) -> typing.Optional[typing.Dict] ``` Legacy judge_failsafe carries real generation that can be judged again. ```python nemo_gym.rollouts.rollout_collection._latest_failure_rows( failures_fpaths: typing.List[pathlib.Path] ) -> typing.Dict[typing.Tuple[typing.Any, typing.Any], typing.Dict[str, typing.Any]] ``` The last failed attempt recorded for each rollout across the failures sidecars. An interrupted attempt that `record_omitted_rollouts` recorded is the rollout's latest execution but no failure: it hides any older failure, so the rollout is left out here and is missing. A row the dispatch budget never started, and the evidence of a failed /run with routing off, are skipped (`is_non_superseding_rollout_row`). ```python nemo_gym.rollouts.rollout_collection._masking_step_metrics( agent_name: str, scored: collections.Counter, dropped: collections.Counter ) -> typing.Dict[str, float] ``` In-progress view of what a run is losing to its environment rather than its policy. `scored` covers persisted rollouts only, split into the unmasked ones (`count`, `reward`) and the masked ones; `dropped` counts what never reached the main output at all. `reward_unmasked` averages over the unmasked rollouts alone, so the gap against the existing `reward` series is the score lost to infrastructure. Failed and omitted attempts are reported as counts, never folded into a quality average. Empty until something is actually masked or dropped, so a healthy run exports exactly what it exported before. The final numbers come from `/aggregate_metrics`; this is the progress view while the run is still going. ```python nemo_gym.rollouts.rollout_collection._matches_failure_class( row: collections.abc.Mapping[str, typing.Any], failure_classes: collections.abc.Collection[str] ) -> bool ``` Match the canonical class or the original legacy producer's routing class. ```python nemo_gym.rollouts.rollout_collection._materialized_taskset( row: collections.abc.Mapping[str, typing.Any] ) -> str | None ``` ```python nemo_gym.rollouts.rollout_collection._metric_failure_kind( failure_kind: typing.Any ) -> typing.Optional[str] ``` The kind as a bounded label: registered or namespaced names pass, anything else is one bucket. ```python nemo_gym.rollouts.rollout_collection._metrics_group( row: collections.abc.Mapping[str, typing.Any], servers_by_agent: typing.Callable[[], collections.abc.Mapping[str, list[str]]] ) -> typing.Optional[str] ``` The environment server a rollout is scored under, as `_call_aggregate_metrics` groups it. The row's stamp, else the one server that fronts its agent, looked up only for an unstamped row: a materialized row routed by its agent has no stamp while its result has one, and both must land in the same group. Runs of one agent behind two servers stay apart this way. An agent behind no server or several stays its own group; `_call_aggregate_metrics` rejects such a row. ```python nemo_gym.rollouts.rollout_collection._missing_rollout_rows_counted_as_zero( materialized_fpaths: typing.List[pathlib.Path], failures_fpaths: typing.List[pathlib.Path], scored_keys: set, history: typing.Optional[nemo_gym.rollouts.rollout_store.RolloutStore] = None ) -> typing.List[typing.Dict[str, typing.Any]] ``` Materialized rollouts that produced no row anywhere, counted as zeros. The failure classes reach a rollout that failed and said so. This reaches the one that never got that far -- killed mid-flight, or dispatched and lost -- which leaves nothing in the rollouts jsonl and nothing in the sidecar. Without it such a rollout leaves the denominator as well as the numerator, so the score reads higher the more of the run went missing. The sidecar is read here rather than trusted from the caller. A failure whose class the caller left out of `count_failure_classes_as_zero` is absent from the scored keys, and counting it here would score the very rollouts that selection excluded -- silently turning the selection into a no-op. The zero carries the rollout's identity: its agent, and its environment server stamp when the row has one, which a materialized taskset row needs because it names no agent. `_fill_task_fields` adds the task's dataset fields. A row that names neither cannot reach any server's metrics, so it is warned about rather than counted. The score enters the metric input and nothing else, the same way a counted failure row does. ```python nemo_gym.rollouts.rollout_collection._names_its_task( row: dict[str, typing.Any] ) -> bool ``` ```python nemo_gym.rollouts.rollout_collection._native_episode_record( row: typing.Dict, response: typing.Any ) -> typing.Dict ``` Validate the dispatched episode identity before translating its outcome. ```python nemo_gym.rollouts.rollout_collection._normalize_health_check_ignored_checks( value ) -> typing.List[str] ``` ```python nemo_gym.rollouts.rollout_collection._normalize_rollout_outcome( row: typing.Dict, result: typing.Any ) -> typing.Dict | nemo_gym.rollouts.rollout_outcomes.RolloutFailureRecord ``` ```python nemo_gym.rollouts.rollout_collection._print_on_sigterm( render: typing.Callable[[], str] ) -> typing.Any ``` Print render() when Slurm's time limit sends SIGTERM, then terminate as before. Python's default SIGTERM action exits without unwinding, so nothing printed in a `finally` survives a time-limit kill. Returns the previous handler for `_restore_sigterm`, or `_NO_SIGTERM_HOOK` when no hook could be installed. ```python nemo_gym.rollouts.rollout_collection._protocol_violation_stop( server: str, answered: int ) -> nemo_gym.rollouts.rollout_recovery.IncompleteEvaluationError ``` The end of a run whose environment server has answered every request so far with a reply the collector cannot use: the producer is wrong, and no attempt can change that. ```python nemo_gym.rollouts.rollout_collection._quota_exhausted_stop( failures: collections.abc.Sequence[collections.abc.Mapping[str, typing.Any]] ) -> nemo_gym.rollouts.rollout_recovery.IncompleteEvaluationError | None ``` The end of a run whose grading provider's quota is exhausted: nothing can be graded until it resets. ```python nemo_gym.rollouts.rollout_collection._raise_for_aggregation_errors( metrics_fpath: typing.Optional[pathlib.Path] ) -> None ``` Signal failure after partial metrics and any batch status have been saved. ```python nemo_gym.rollouts.rollout_collection._read_jsonl( path: pathlib.Path ) -> typing.List[typing.Dict] ``` async Parse a whole JSONL artifact, letting other tasks on the event loop run between rows. Parsing stays on the event loop thread instead of a worker thread. The loop thread then reuses the memory it freed during collection for these rows. A worker thread allocates from its own malloc arena, so the process would hold both. ```python nemo_gym.rollouts.rollout_collection._record_policy_model_server( result: dict[str, typing.Any], global_config: typing.Any, agent_name: str | None ) -> None ``` Name the agent's configured policy server on the record (landed or failed), so offline timing needs no live config and a capture joined later from the capture dir is still partitioned by role. ```python nemo_gym.rollouts.rollout_collection._record_rollout_health( result: typing.Dict[str, typing.Any], outcome: str, ignored_checks: frozenset[str], dispatch_span_context: typing.Optional[opentelemetry.trace.SpanContext] = None ) -> None ``` Evaluate the trajectory health checks on a finished rollout and count the verdict and flagged checks. With a `dispatch_span_context`, the verdict and the flagged checks also go on a `gym.rollout.health` span under that dispatch span. Gated on telemetry only, not on the post-run report switch. Never raises: telemetry must not fail a rollout. ```python nemo_gym.rollouts.rollout_collection._record_rollout_outcome( result: typing.Dict[str, typing.Any], outcome: str, dispatch_name: str ) -> None ``` Count a persisted attempt, including its numeric reward when scored. ```python nemo_gym.rollouts.rollout_collection._record_rollout_shape( row: collections.abc.Mapping[str, typing.Any], result: collections.abc.Mapping[str, typing.Any] ) -> None ``` Record the size of one completed rollout, while metrics are exporting. Turns and tool calls come from the rollout's `ng_perf` when rollout observability produced it. Without it, tool calls are counted from the final response's `function_call` items, and turns are not recorded. ```python nemo_gym.rollouts.rollout_collection._reflect_enrichment( result: typing.Any, record: typing.Dict[str, typing.Any] ) -> None ``` Add a stored record's enrichment and identity keys to the reply `run_examples` returns in its own shape. ```python nemo_gym.rollouts.rollout_collection._refresh_shard_status_aggregation( input_paths: typing.List[str], aggregate_metrics_fpath: typing.Optional[pathlib.Path] ) -> None ``` Mark each shard's run status with the outcome of the aggregation that merged it, keeping its counts. EvalRDL sums the shards' status files, so the merged output gets no status file of its own. ```python nemo_gym.rollouts.rollout_collection._reject_attempt_qualified_rollout_ids( rows: collections.abc.Iterable[collections.abc.Mapping[str, typing.Any]], fpath: pathlib.Path ) -> None ``` Refuse dataset rows whose explicit rollout ID names an attempt. The suffix `-a` is reserved: `"-a"` always means attempt N of rollout ``, which is how a training controller names a restored attempt. A resume retry stamps the next attempt on a row, and an ID that already names one would then fail mid-run, so the row is refused when the input loads. ```python nemo_gym.rollouts.rollout_collection._reply_failure_kind( headers: typing.Any ) -> typing.Optional[str] ``` The kind of the outermost failure in the chain an agent's failed reply selected that is not a routing kind. `None` without a report, with an omitted or broken chain, or when every failure on it only routes. ```python nemo_gym.rollouts.rollout_collection._reply_failure_report( response: collections.abc.Mapping[str, typing.Any] ) -> tuple[nemo_gym.core.failure_records.FailureReport, ...] ``` The `FailureReport` an environment reply carries beside its outcome; an invalid one is omitted. ```python nemo_gym.rollouts.rollout_collection._restore_sigterm( previous: typing.Any ) -> None ``` ```python nemo_gym.rollouts.rollout_collection._rollout_order_key( row: typing.Dict[str, typing.Any] ) -> tuple ``` Task then repeat: metrics that read a task's repeats positionally need that order. ```python nemo_gym.rollouts.rollout_collection._rollout_request_debug_summary( row: typing.Dict[str, typing.Any] ) -> typing.Dict[str, typing.Any] ``` ```python nemo_gym.rollouts.rollout_collection._round_robin_by_agent( examples: typing.List[typing.Dict] ) -> typing.List[typing.Dict] ``` Interleave resolved rows by environment server or agent, preserving each input order. Agent order is the order in which each name first appears. Rows are returned unchanged, so task/rollout indexes, repeat seeds, and the identities used by resume remain intact. ```python nemo_gym.rollouts.rollout_collection._routing_identity( row: collections.abc.Mapping[str, typing.Any] ) -> typing.Optional[str] ``` The agent a rollout ran on, or for a materialized taskset row, which names none, its environment server. A materialized row, its result and its sidecar row all name the same one. ```python nemo_gym.rollouts.rollout_collection._rows_of_first_tasks( numbered_rows: collections.abc.Iterable[tuple[int, str]], limit: int ) -> typing.Iterator[tuple[int, str]] ``` Yield the rows of the first `limit` tasks. Identical rows are one task, as in the task index. ```python nemo_gym.rollouts.rollout_collection._sample_server_queues( report: nemo_gym.rollouts.rollout_collection.RunTimingReport, base_urls: typing.List[str], interval_s: float ) -> None ``` async Poll every vLLM server's /metrics and record requests waiting and running, summed over servers. ```python nemo_gym.rollouts.rollout_collection._sanitize_legacy_result( result: typing.Dict ) -> None ``` Discard collector bookkeeping while preserving declared legacy evidence. ```python nemo_gym.rollouts.rollout_collection._shard_record( completed: nemo_gym.rollouts.rollout_collection._CompletedRollout, run_id: str, global_config: typing.Any, agent_name: str | None ) -> typing.Dict[str, typing.Any] ``` The record `ng_collect_rollouts` would store for `completed`, for a `run_examples` call recording a shard. A try that left no result (interrupted, or never started) is a marker, not an outcome: it keeps its identity and the writer turns it into an omitted-rollout row. Anything else goes through `_collected_outcome`, so a failed `/run` is saved as the typed failure row collection saves, never as the marker the dispatcher returned. Every record names the shard's run, as collection stamps its own, so the Store checks it against the shard's manifest. ```python nemo_gym.rollouts.rollout_collection._shard_rollout_id( row: collections.abc.Mapping[str, typing.Any], run_id: str ) -> str ``` The rollout id a recorded shard dispatches `row` under: the row's own id, prefixed with the shard's run id. Model calls are captured under the capture key, which is this id with the attempt suffix, and capture files are kept by default. Without the prefix, two shards that dispatch the same task and rollout (a returned row fed back, or indices numbered again each step) share a key, and the later shard's records carry the earlier shard's calls. A row an earlier shard returned carries that shard's prefix, which is replaced rather than nested. ```python nemo_gym.rollouts.rollout_collection._span_task_id( row: collections.abc.Mapping[str, typing.Any] ) -> typing.Optional[str] ``` ```python nemo_gym.rollouts.rollout_collection._stamp_collector_fields( completed: nemo_gym.rollouts.rollout_collection._CompletedRollout, result: dict[str, typing.Any] ) -> None ``` Stamp the row's identity and routing fields onto the record collection stores, as every reader expects. ```python nemo_gym.rollouts.rollout_collection._status_failure_kind( row: collections.abc.Mapping[str, typing.Any] ) -> typing.Optional[str] ``` The failure kind a sidecar row carries: top-level, the legacy diagnostic key, or the typed record's. ```python nemo_gym.rollouts.rollout_collection._sum_vllm_queue( metrics_text: str ) -> typing.Tuple[float, float] ``` ```python nemo_gym.rollouts.rollout_collection._task_data_provenance( row: collections.abc.Mapping[str, typing.Any] ) -> typing.Dict[str, typing.Any] ``` The `PROVENANCE_KEYS` of a materialized row's `task_data`: provenance stamps for the record. Preparation writes them (a Harbor row carries the task folder digest as `_ng_digest` and the dataset digest as `_ng_dataset_digest`) so that every results row names the inputs it ran, without the materialized-inputs sidecar. An Environment Server never returns them (`_ng_*` is the collector's namespace), so the collector copies them from the row onto the record. Only the allowlisted keys are copied: `task_data` comes from the dataset, and the collector reads other `_ng_*` keys (`_ng_no_persist`, `_ng_failure_class`, `_ng_result_type`, ...) that a success record does not carry, so a dataset could otherwise drop, reroute or relabel its own results. ```python nemo_gym.rollouts.rollout_collection._truncated_body( body: typing.Optional[bytes] ) -> typing.Optional[str] ``` Decode at most the kept prefix, so a huge error page is never decoded in full. ```python nemo_gym.rollouts.rollout_collection.enrich_run_result( row: dict[str, typing.Any], result: dict[str, typing.Any], capture_dirs: list[pathlib.Path], observability_enabled: bool, include_payloads: bool = True, rollout_latency_ms: typing.Optional[float] = None, delete_capture: bool = False ) -> dict[str, typing.Any] ``` async Attach a finished rollout's collected evidence to its `/run` result, in place. These are the per-rollout steps `ng_collect_rollouts` runs, shared so a framework that dispatches through `run_examples` records the same thing: wait for the model server's in-flight capture writes for this rollout, merge its captured model calls (`ng_model_call_capture`) and join the agent's observations to them, project the standardized `ng_trajectory`, and assemble `ng_perf` when observability is enabled. The row's rollout identity is copied onto the result where the result lacks it, because the capture key is read from the record. Harness output and reward are never modified. A row that names no task (no task id field, task index or trajectory identity) gets no `ng_trajectory` rather than a guessed task id. `include_payloads=False` keeps call metadata only; `delete_capture` removes the rollout's capture file once it has been merged, so it requires the payloads to be kept. ```python nemo_gym.rollouts.rollout_collection.is_non_superseding_rollout_row( row: collections.abc.Mapping[str, typing.Any] ) -> bool ``` Whether readers skip a sidecar row before they pick a rollout's latest attempt. A row never started is no execution. A failed /run kept as evidence with routing off is no new outcome: the attempt it records stands no more than before it was kept, so an older failure still stands and a later attempt still replaces it. An interrupted `record_omitted_rollouts` row is neither and still hides older failures. ```python nemo_gym.rollouts.rollout_collection.is_omitted_rollout_row( row: collections.abc.Mapping[str, typing.Any] ) -> bool ``` Whether a failures-sidecar row is a `record_omitted_rollouts` record of a rollout that produced no result. ```python nemo_gym.rollouts.rollout_collection.is_unstarted_rollout_row( row: collections.abc.Mapping[str, typing.Any] ) -> bool ``` Whether a sidecar row records a rollout the dispatch budget never started: no execution, so no history. ```python nemo_gym.rollouts.rollout_collection.loads_jsonl_line( raw, fpath, line_no: int ) ``` Parse one JSONL line, raising a clean `ConfigError` (naming file + line) on malformed JSON. ```python nemo_gym.rollouts.rollout_collection.AGENT_REQUEST_FAILED_FAILURE_CLASS = 'agent_request_failed' ``` ```python nemo_gym.rollouts.rollout_collection.AGENT_RUN_ERROR_FAILURE_CLASS = 'agent_run_error' ``` ```python nemo_gym.rollouts.rollout_collection.ENVIRONMENT_SERVER_FAILURE_CLASS = ENVIRONMENT_SERVER_FAILED ``` ```python nemo_gym.rollouts.rollout_collection.NG_DATASET_DIGEST_KEY = '_ng_dataset_digest' ``` ```python nemo_gym.rollouts.rollout_collection.NG_DIGEST_KEY = '_ng_digest' ``` ```python nemo_gym.rollouts.rollout_collection.NG_ELAPSED_KEY = 'elapsed_seconds' ``` ```python nemo_gym.rollouts.rollout_collection.NG_ENVIRONMENT_SERVER_KEY = ENVIRONMENT_SERVER_STAMP_KEY_NAME ``` ```python nemo_gym.rollouts.rollout_collection.NG_FAILURE_COUNTS_AS_ZERO_KEY = '_ng_failure_counts_as_zero' ``` ```python nemo_gym.rollouts.rollout_collection.NG_FAILURE_JUDGE_ERROR_KEY = '_ng_failure_judge_error' ``` ```python nemo_gym.rollouts.rollout_collection.NG_FAILURE_LEGACY_CLASS_KEY = '_ng_failure_legacy_class' ``` ```python nemo_gym.rollouts.rollout_collection.NG_FAILURE_MAX_ATTEMPTS_KEY = '_ng_failure_max_attempts' ``` ```python nemo_gym.rollouts.rollout_collection.NG_FAILURE_REPORT_KEYS = ('_ng_failure_events', '_ng_failure_id', '_ng_failure_events_omitted') ``` ```python nemo_gym.rollouts.rollout_collection.NG_OMITTED_ROLLOUT_KEY = '_ng_omitted_rollout' ``` ```python nemo_gym.rollouts.rollout_collection.NG_POLICY_MODEL_SERVER_KEY = 'ng_policy_model_server' ``` ```python nemo_gym.rollouts.rollout_collection.NG_RESULT_TYPE_KEY = '_ng_result_type' ``` ```python nemo_gym.rollouts.rollout_collection.NG_TASK_ID_KEY = '_ng_task_id' ``` ```python nemo_gym.rollouts.rollout_collection.NG_TERMINAL_KEY = '_ng_failure_terminal' ``` ```python nemo_gym.rollouts.rollout_collection.NG_UNROUTED_RUN_ERROR_KEY = '_ng_unrouted_run_error' ``` ```python nemo_gym.rollouts.rollout_collection.PROVENANCE_KEYS = (NG_DIGEST_KEY, NG_DATASET_DIGEST_KEY) ``` ```python nemo_gym.rollouts.rollout_collection.ROLLOUT_REPEAT_ATTRIBUTE = 'nemo.gym.rollout.repeat' ``` ```python nemo_gym.rollouts.rollout_collection.STATUS_FAILURE_KIND_KEY = 'failure_kind' ``` ```python nemo_gym.rollouts.rollout_collection.STATUS_MODEL_FINAL_KEY = '_ng_model_final' ``` ```python nemo_gym.rollouts.rollout_collection.TASK_ID_ATTRIBUTE = 'nemo.gym.task.id' ``` ```python nemo_gym.rollouts.rollout_collection._ENRICHMENT_KEYS = ('ng_model_call_capture', NG_TRAJECTORY_KEY, NG_PERF_KEY) ``` ```python nemo_gym.rollouts.rollout_collection._LEGACY_RESULT_FIELDS = frozenset({NG_FAILURE_CLASS_KEY, NG_NO_PERSIST_KEY, NG_TERMINAL_KEY, '_ng_failur... ``` ```python nemo_gym.rollouts.rollout_collection._MAX_FAILURE_BODY_CHARS = 2000 ``` ```python nemo_gym.rollouts.rollout_collection._NO_RESULT_FAILURE_CLASSES = frozenset({AGENT_REQUEST_FAILED_FAILURE_CLASS, AGENT_RUN_ERROR_FAILURE_CLASS, EN... ``` ```python nemo_gym.rollouts.rollout_collection._NO_SIGTERM_HOOK = object() ``` ```python nemo_gym.rollouts.rollout_collection._OWNED_JOB_SPAN_CONTEXT: ContextVar[Optional[SpanContext]] = ContextVar('nemo_gym_owned_job_span_context', default=None) ``` ```python nemo_gym.rollouts.rollout_collection._READ_JSONL_YIELD_INTERVAL_S = 0.01 ``` ```python nemo_gym.rollouts.rollout_collection._RESERVED_SHARD_ENDINGS = ('_failures', '_materialized_inputs', '_attempts') ``` ```python nemo_gym.rollouts.rollout_collection._ROUTING_FAILURE_KINDS = frozenset({AGENT_ROUTE_FAILED, COLLECTION_DISPATCH_FAILED, AGENT_RUN_ERROR_FAILU... ``` ```python nemo_gym.rollouts.rollout_collection._RUN_FAILURE_ERRORS = (ClientError, orjson.JSONDecodeError, TimeoutError) ``` ```python nemo_gym.rollouts.rollout_collection._SERVER_DID_NOT_RUN_STATUSES = frozenset({429, 502, 503, 504}) ``` ```python nemo_gym.rollouts.rollout_collection._SHARD_IDENTITY_KEYS = (ROLLOUT_ID_KEY_NAME, TASK_INDEX_KEY_NAME, ROLLOUT_INDEX_KEY_NAME, ATTEMPT_INDEX... ``` ```python nemo_gym.rollouts.rollout_collection._SHARD_LABEL = re.compile('[A-Za-z0-9][A-Za-z0-9_.-]*') ``` ```python nemo_gym.rollouts.rollout_collection._failures_path_for = failures_path_for ``` ```python nemo_gym.rollouts.rollout_collection._moved_cli_entry_points = moved_attr_getter(__name__, {'collect_rollouts': 'nemo_gym.cli.eval', 'aggregate... ``` ```python nemo_gym.rollouts.rollout_collection._moved_names = moved_names_getattr(__name__) ``` ```python nemo_gym.rollouts.rollout_collection.get_max_rollout_attempts = _get_max_rollout_attempts ``` ```python nemo_gym.rollouts.rollout_collection.logger = logging.getLogger(__name__) ```