fix(genrm): bound cohort deadlines and failure cleanup

nemo_gym/judge.py

@@ -100,10 +100,21 @@ async def wrapper(*args, **kwargs):
             )
             if body is None:  # verify always has a request body; guard against an opaque 500
                 raise RuntimeError("judge_failsafe: could not locate the verify request body") from e
-            data = body.model_dump() | {
+            error = str(e)[:2000]
+            data = body.model_dump()
+            instance_config = data.get("instance_config")
+            data |= {
                 "reward": 0.0,
+                "mask_sample": True,
+                "failure_kind": "judge_failed",
+                "failure_reason": error,
+                # Keep older NeMo RL consumers working during top-level field migration.
+                "instance_config": {
+                    **(instance_config if isinstance(instance_config, dict) else {}),
+                    "mask_sample": True,
+                },
                 "_ng_failure_class": "judge_failed",
-                "_ng_failure_judge_error": str(e),
+                "_ng_failure_judge_error": error,
             }
             return JSONResponse(content=jsonable_encoder(data))
 

nemo_gym/reward_profile.py

@@ -187,7 +187,7 @@ def rollout_info_from_result(self, result: Dict[str, Any]) -> Dict[str, Any]:
                 rollout_info[k] = v
 
         for k, v in result.items():
-            if k in {TASK_INDEX_KEY_NAME, ROLLOUT_INDEX_KEY_NAME, "reward", "response"}:
+            if k.startswith("_") or k in {"reward", "response"}:
                 continue
             if isinstance(v, bool):
                 rollout_info[k] = int(v)
@@ -425,6 +425,8 @@ def _group_key(row: Dict[str, Any]) -> Any:
                 ROLLOUT_INDEX_KEY_NAME: rollout_idx,
             }
             for k, v in result.items():
+                if k.startswith("_"):
+                    continue
                 if isinstance(v, bool):
                     numeric_result[k] = int(v)
                 elif isinstance(v, (int, float)):

nemo_gym/rollout_collection.py

@@ -940,7 +940,14 @@ class before the wanted classes are picked out. Selecting the other way round wo
             continue
         # Diagnostics stay in the sidecar: an HTTP status is a number, and the aggregator
         # averages every number it is handed.
-        scored = {k: v for k, v in row.items() if not k.startswith("_ng_failure_")}
+        scored = {
+            k: v
+            for k, v in row.items()
+            if not k.startswith("_ng_failure_") and k not in ("failure_kind", "failure_reason")
+        }
+        # This metrics-only copy honors the explicit denominator policy. The original
+        # answer, failure diagnostics, and training mask remain untouched in the sidecar.
+        scored["mask_sample"] = False
         scored.setdefault("reward", 0.0)
         counted.append(scored)
     return counted
@@ -1621,7 +1628,14 @@ async def _run_from_config(self, config: RolloutCollectionConfig) -> Tuple[List[
         results_file.close()
         failures_file.close()
 
-        if input_rows and not persisted_results:
+        # Explicitly counted failures can provide a score even when no rollout
+        # succeeded. Determine eligibility before rejecting an otherwise empty run.
+        counted = _failure_rows_counted_as_zero(
+            [failures_fpath],
+            config.count_failure_classes_as_zero,
+            {(r[TASK_INDEX_KEY_NAME], r[ROLLOUT_INDEX_KEY_NAME]) for r in persisted_results},
+        )
+        if input_rows and not persisted_results and not counted:
             raise RuntimeError(
                 f"None of the {len(input_rows)} dispatched rollouts produced a result "
                 f"{dict(failure_counts)}. Inspect {failures_fpath}; the run has no score to report."
@@ -1639,10 +1653,8 @@ async def _run_from_config(self, config: RolloutCollectionConfig) -> Tuple[List[
         persisted_rows.sort(key=lambda r: (r[TASK_INDEX_KEY_NAME], r[ROLLOUT_INDEX_KEY_NAME]))
         persisted_results.sort(key=lambda r: (r[TASK_INDEX_KEY_NAME], r[ROLLOUT_INDEX_KEY_NAME]))
 
-        # Compute and write aggregate metrics via /aggregate_metrics using only the
-        # rows written to the main rollouts jsonl so runtime aggregation matches
-        # `gym eval aggregate`.
-        counted: List[Dict] = []
+        # Aggregate persisted results plus explicitly counted metrics-only failures,
+        # matching `gym eval aggregate` without changing either rollout artifact.
         if config.disable_aggregation:
             print(
                 "Skipping aggregate-metrics computation because disable_aggregation=True. "
@@ -1651,11 +1663,6 @@ async def _run_from_config(self, config: RolloutCollectionConfig) -> Tuple[List[
             aggregate_metrics_fpath = None
         else:
             print("Computing aggregate metrics")
-            counted[:] = _failure_rows_counted_as_zero(
-                [failures_fpath],
-                config.count_failure_classes_as_zero,
-                {(r[TASK_INDEX_KEY_NAME], r[ROLLOUT_INDEX_KEY_NAME]) for r in persisted_results},
-            )
             if config.count_failure_classes_as_zero:
                 print(
                     f"Counting {len(counted)} failure row(s) as scored zeros: {config.count_failure_classes_as_zero}"