[Refactor] Consolidate stats() in BaseCollector and add WandB monitoring example

[coder-jayp]

Consolidates single-process and multiprocessing collector statistics in BaseCollector.stats(), preserving shared in-flight trajectory counters, non-blocking aggregate snapshots, and the existing per-worker control-channel behavior. Workers return data to the coordinator, where logging happens.

The public stats() API stays the same. The before/after below shows the regression-test scenario: one healthy Ray worker is held inside its first collection call for longer than the stats timeout.

Before: the stats request waits behind collection on the same actor queue, times out, and the healthy worker is omitted from the snapshot.

snapshot = collector.stats(workers="both", timeout=1.0)
snapshot["workers_alive"]  # 0, even though the worker is collecting
"worker_0/frames" in snapshot  # False: no worker snapshot arrived

After: the stats request runs in its own concurrency group and can report progress while collection is still busy.

snapshot = collector.stats(workers="both", timeout=1.0)
snapshot["workers_alive"]  # 1: the worker answered during collection
snapshot["worker_0/frames"]  # 0: its first batch has not finished yet

The new example uses this snapshot API through LoggerMonitor to send aggregate and per-worker progress to W&B from the coordinator:

with LoggerMonitor(logger, poll_interval=0.5) as monitor:
    monitor.watch(
        collector,
        name="ray_collector",
        schedule=Every.counter("frames", 100),
        stats_kwargs={"workers": "both"},
    )
    for batch in collector:
        ...  # Training or other batch processing; monitoring runs separately.

Collection and state-changing calls retain the default actor queue. Remote polling remains timeout-bounded, and caller-supplied actor options and concurrency groups are preserved.

Validation:

• All 11 TestCollectorStats cases pass, including two real-Ray tests with Ray 2.54.0.
• The new regression holds collection behind a gate, verifies stats respond while the gate is closed, then releases it and verifies the collected batch. Restoring the old stats dispatch makes this test fail because workers_alive becomes zero.
• The exact example completes with offline W&B. A separate manual monitor poll verifies persisted aggregate and per-worker frame counters from two live Ray workers.
• Flake8, ufmt, and git diff --check pass.
• The example is registered as runnable CI coverage with offline W&B and an isolated log directory. The manifest/sharding checks and actual Ray/W&B example pass through the CI runner (12 passed).

[pytorch-bot]

🔗 Helpful Links

🧪 See artifacts and rendered test results at hud.pytorch.org/pr/pytorch/rl/4400

• 📄 Preview Python docs built from this PR

Note: Links to docs will display an error until the docs builds have been completed.

This comment was automatically generated by Dr. CI and updates every 15 minutes.

[theap06]

left some comments. overall design lgtm.