Pins what the failure bound guarantees across parallel environments

Review on #7433 noted that with several environments the recorded failure count can pass
max_num_failures. Measured on the shipped loop: the overshoot is exactly the number of attempts
that end on the step which crosses the bound, minus one, so it is zero for a single environment
and at most num_envs - 1. It cannot be driven to zero -- those attempts have already completed
when the bound is read, and stopping a generator instead leaves env_loop waiting forever for an
action from it, since the loop blocks until all num_envs actions are queued.

So the field's docstring now states the guarantee for parallel runs, and two tests pin both
sides of it: exact when attempts end on separate steps, bounded by num_envs - 1 when they end
together. The test harness gains num_envs and attempts_per_step, defaulting to what it did before.
No behaviour change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019SsU8ziGULaEBbtmvJSQZW

Author: 2047767028-lang

source/isaaclab/isaaclab/envs/mimic_env_cfg.py

@@ -48,6 +48,10 @@ class DataGenConfig:
 
     With the guarantee disabled, generation already stops after :attr:`generation_num_trials`
     attempts and this field is ignored, so setting it cannot cut a fixed-attempt run short.
+
+    The bound is read once per simulation step. Attempts that end on the step that crosses it are
+    already complete, so a run over ``num_envs`` parallel environments can record up to
+    ``num_envs - 1`` failures beyond the bound; it is exact whenever attempts end on separate steps.
     """
 
     seed: int = 1

source/isaaclab_mimic/test/test_generation_failure_cap.py

@@ -31,9 +31,20 @@ class _Fuse(Exception):
     """Raised by the fake environment once the loop has run for longer than any test expects."""
 
 
-def _run(outcomes, *, max_num_failures, generation_num_trials=3, generation_guarantee=True):
+def _run(
+    outcomes,
+    *,
+    max_num_failures,
+    generation_num_trials=3,
+    generation_guarantee=True,
+    num_envs=1,
+    attempts_per_step=1,
+):
     """Run ``env_loop`` over scripted attempt outcomes; return how it ended and the final counters.
 
+    ``attempts_per_step`` is how many of the ``num_envs`` generators finish an attempt on the same
+    step, which is what decides whether the bound is read before or after the extra attempts land.
+
     Returns:
         A tuple ``(how, num_success, num_failures, num_attempts)`` where ``how`` is ``"exited"`` if
         the loop returned on its own and ``"fuse"`` if it was still running after ``FUSE_STEPS``.
@@ -42,25 +53,28 @@ def _run(outcomes, *, max_num_failures, generation_num_trials=3, generation_guar
     loop = asyncio.get_event_loop()
     action_queue: asyncio.Queue = asyncio.Queue()
     reset_queue: asyncio.Queue = asyncio.Queue()
-    action_queue.put_nowait((0, torch.zeros(7)))
+    for env_id in range(num_envs):
+        action_queue.put_nowait((env_id, torch.zeros(7)))
     scripted = iter(outcomes)
     steps = {"n": 0}
 
     def step(actions):
         steps["n"] += 1
         if steps["n"] > FUSE_STEPS:
             raise _Fuse(f"env_loop still running after {FUSE_STEPS} steps")
-        if next(scripted):
-            generation.num_success += 1
-        else:
-            generation.num_failures += 1
-        generation.num_attempts += 1
-        action_queue.put_nowait((0, torch.zeros(7)))
+        for _ in range(attempts_per_step):
+            if next(scripted):
+                generation.num_success += 1
+            else:
+                generation.num_failures += 1
+            generation.num_attempts += 1
+        for env_id in range(num_envs):
+            action_queue.put_nowait((env_id, torch.zeros(7)))
 
     env = SimpleNamespace(
-        num_envs=1,
+        num_envs=num_envs,
         device="cpu",
-        action_space=SimpleNamespace(shape=(1, 7)),
+        action_space=SimpleNamespace(shape=(num_envs, 7)),
         step=step,
         reset=lambda env_ids=None: None,
         close=lambda: None,
@@ -120,3 +134,17 @@ def test_attempt_based_termination_is_unchanged(max_num_failures):
     )
     assert how == "exited"
     assert attempts == 10
+
+
+def test_bound_is_exact_when_attempts_end_on_separate_steps():
+    """Four environments, one attempt landing per step: the bound is read between every attempt."""
+    how, _, fail, _ = _run([False] * 100, max_num_failures=5, num_envs=4, attempts_per_step=1)
+    assert how == "exited"
+    assert fail == 5
+
+
+def test_attempts_ending_together_overshoot_the_bound_by_at_most_num_envs_minus_one():
+    """Attempts that end on the step that crosses the bound are already complete and still count."""
+    how, _, fail, _ = _run([False] * 100, max_num_failures=5, num_envs=4, attempts_per_step=4)
+    assert how == "exited"
+    assert 5 < fail <= 5 + 4 - 1