[linj-glitch]

run_examples() now exposes aclose() so direct callers and collection owners can cancel unfinished client work. aclose_and_drain() stops admission before suspending and preserves completed, unconsumed outcomes for deadline consumers such as RL #4482. Completed failures remain observable; consumed payloads are released; outcomes are delivered once.

Concurrent eager consumers now use one queue notification per finished rollout. Drain wakes every parked consumer. A deterministic public-API regression at 1,024 individually arriving outcomes measures 1,024 consumer resumptions, versus 524,800 with the shared event. Tests also cover cancelled notified consumers, multiple parked drain consumers, completed backlogs, exception retrieval, and result lifetime.

Both lifecycle methods wait for local client tasks. They do not acknowledge server/agent cleanup or guarantee sandbox release. Existing sequential use remains compatible; no configuration or benchmark-score behavior changes.

Main c2000f79f6bea21e9b21f7f021d57a06f6082974 is integrated additively. The remaining test import conflict is resolved by retaining signal, sys, and time; #4646's asynchronous artifact reads are preserved. Collector cleanup retains main's timing tasks, SIGTERM restoration, and report, while closing both iterator modes. Its capture-write barrier and reward telemetry are preserved. The queue-based iterator also preserves #4644's consumed-result release; all four new upstream weak-reference cases pass unchanged.

Validation on head f28029f5b359858b09596c45a476fc1dd5b52596:

python3 -m pytest -o addopts= -q \
  tests/unit_tests/test_rollout_collection.py \
  tests/unit_tests/test_server_utils.py \
  tests/unit_tests/telemetry/test_observer.py \
  tests/unit_tests/test_run_timing.py \
  tests/unit_tests/test_base_responses_api_model.py \
  tests/unit_tests/telemetry/test_rollout_reward.py \
  tests/unit_tests/test_rollout_health.py \
  tests/unit_tests/test_rollout_observability.py \
  tests/unit_tests/test_rollout_correlation.py \
  tests/unit_tests/test_token_id_capture.py \
  tests/unit_tests/test_trajectory_builder.py
# 1,244 passed (including all 498 rollout tests)

pre-commit run --all-files
# All applicable hooks passed; no files modified. The guardwords hook retains its catalog-availability guard.

RL #4482 at 778a9d7ee3c8811174745137519748a6b9443271: all 17 native real-Gym interoperability cases passed again against Gym head f28029f5b359858b09596c45a476fc1dd5b52596 after the main integration. The original three performance regressions failed before the queue fix and passed afterward. Disabling drain shutdown also makes the new parked-consumer regression fail. Real HTTP client cancellation remains covered.

Model rollout evidence: N/A for this shared-library lifecycle repair. No training A/B was run. A separate issue is unnecessary because this addresses the P2 feedback on this PR.

#3302 should follow this change and retain both lifecycle methods. RL activation needs a subsequent Gym pin update. Ananth's concrete review requests are addressed. Full mirrored CI and the repository's required approval remain outstanding.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[ananthsub]

Thanks for this. It fills a real gap: a direct caller such as a trainer's straggler cutoff has no way today to withdraw /run requests it already sent, and #3302 does not cover it because run_examples there still returns a map. I checked the mechanism end to end against a real server: cancelling the task closes the connection, the disconnect middleware cancels the server handler within a millisecond, and nothing retries the cancelled request. The new tests fail on the old code.

Two changes I'd like before merge, both inline: the unbounded aclose() should retrieve every task, and the docstrings should say what aclose() actually waits for. Unit and server tests have not run on this head yet (only lint, sign-off and secrets checks ran), so those need to pass too.

flowchart LR
    A["caller: aclose()"] --> B["cancel unfinished rollout tasks"]
    B --> C["gather tasks in this process"]
    C --> D["aclose() returns"]
    B -. "connection closed" .-> E["server: disconnect middleware cancels handler"]
    E -. "after aclose() returns" .-> F["agent cleanup (depends on agent; some keep running)"]

Loading

Smaller points, none blocking:

• Collector cleanup. The collector's finally closes only the bounded iterator. Checking hasattr(rows, "aclose") would cover both. It only matters for library callers with a long-lived event loop, since asyncio.run cancels leftover tasks at shutdown.
• Order with #3302. The two PRs conflict in _run_examples_with_metadata. I'd land this one first; #3302 then keeps its exception-retrieving done callback and has run_outcomes return the closable iterator.
• End-to-end test. The tests mock the post. One test against a real server with the disconnect middleware would pin down the behavior the docstring describes.
• Cancel, not retire. aclose() withdraws requests; it does not stop the attempt everywhere and confirm, which is what the checkpoint stack's retire does (#3882). That is the right primitive to ship here. It does make the open question concrete of whether a retire without checkpointing should exist, which I'll pick up on the failure-handling tracker.

[yuhezhang-ai]

LGTM on e24bc02. Both findings from my review are addressed, and all 496 rollout tests pass locally. Please resolve the remaining import conflict (keeping time, signal, and sys) and rerun the required checks on the resulting commit before merging.

[yuhezhang-ai]

/ok to test f28029f

[yuhezhang-ai]

LGTM on f28029f5. Rechecked after the conflict resolution: both review findings remain fixed, all 498 rollout tests pass locally, the 1,024-row workload produces 1,024 consumer resumptions, and the memory-retention check retains 0/3 payloads after exhaustion and close. The import conflict is resolved. Full CI has been requested and must pass before merge.

[github-actions]

🔄 Merge queue validation started! You can track the progress here: https://github.com/NVIDIA-NeMo/Gym/actions/runs/39008712301