[luhengsw]

feat(rl): stream live training metrics over motrix.metrics/v1

Why

Until now, training metrics were only available after the fact as TensorBoard event files. An external
dashboard couldn't show progress, loss or reward during a run, or tell that a run had died. With this
PR, a training process pushes its metrics to any ingest service compatible with motrix.metrics/v1
while it runs.

It is off by default and turns on only when MOTRIX_METRICS_URL is set. Once enabled, every record
goes both to run_dir/metrics.jsonl and, in batches, to the service. A slow, unreachable or erroring
service never blocks training and never makes it fail. The terminal panel and TensorBoard are
unchanged.

Design: wiki/design/live-metrics.md. Plan:
wiki/plan/live-metrics.md

Usage

export MOTRIX_METRICS_URL=http://metrics.example:8080
export MOTRIX_METRICS_TOKEN=...            # optional; only ever sent in the Authorization header
export MOTRIX_METRICS_RUN_ID=my-run        # optional; derived from run_dir by default
export MOTRIX_METRICS_PROJECT=g1-locomotion
export MOTRIX_METRICS_GROUP=ablation-lr
export MOTRIX_METRICS_META='{"owner": "lh"}'

Without MOTRIX_METRICS_URL the logger is disabled: it creates no file, starts no thread, and every
call is a no-op. play never builds a logger.

Changes

Control plane (motrix_rl)

• New motrix_rl/metrics.py, which uses only the stdlib, so there are no new dependencies:
  • MetricsLogger has run(), scalars(), event() and close().
    • Each record gets one seq and goes to both sinks, so the jsonl can later backfill the service
      record for record, deduplicated by (writer, seq).
    • seq starts at the wall clock rather than 0. A restarted container can reuse the same hostname
      and pid, and the service would otherwise drop the new records as duplicates.
  • JsonlSink appends to metrics.jsonl and warns once on IO errors.
  • HttpSink uses a daemon thread and a bounded queue. It sends gzip batches to POST {URL}/api/v1/ingest, up to 500 records per batch or every 2s.
    • Full queue: it drops only the oldest scalars, never run or event records.
    • Retries: network errors, 408, 429 and 5xx are retried with exponential backoff, and 503
      honours Retry-After. Any other 4xx drops the batch with a single warning.
    • Heartbeats: it sends them while idle, so a long torch.compile isn't marked as lost.
  • NaN and Inf are encoded as null, and the first non-finite value for each key also emits a
    nonfinite event. numpy and torch values are converted recursively.
• TrainerHandle.train() reports the run lifecycle:
  • It sends running, then one of finished, failed, or killed (on Ctrl+C).
  • A failed record carries a reason: Type: message, capped at 2KB, plus the paths of any
    async_errors/*.log files.
  • Exceptions still propagate unchanged.
• The running record carries the metadata.json fields plus family (rl.offpolicy /
  rl.onpolicy), total_steps, num_envs, the resolved config and motrixlab_version.
• New AgentProvider.total_env_steps(config, num_envs) hook, used for progress reporting.
• TrainerContext.metrics passes the logger to providers. When live metrics are enabled,
  metadata.json records live.run_id.

Providers

All providers emit scalars under shared canonical keys, aligned on axes.env_steps. TensorBoard tags
are unchanged.

• FastSAC (sync and async):
  • The canonical keys are the existing TensorBoard tags.
  • In async mode only the parent process holds the logger, and it emits worker_crashed when a worker
    exits abnormally.
  • It also emits the run_started, warmup_done, checkpoint_saved and resumed events, plus
    sys/* resource metrics.
• SKRL (torch and jax PPO): a new live_metrics.py maps native tags to canonical keys. Unmapped
  tags are kept as skrl/<slug>.
• RSLRL: a new live_logger.py does the same mapping. Unmapped tags are kept as rslrl/<slug>.

The full key mapping and the step / env_steps conventions are in §3 of the design doc.

Fixes found along the way

• RSLRL: the env wrappers (wrap_np / wrap_torch) now forward reward terms and env metrics via
  extras["log"]. Previously they never reached TensorBoard.
• SKRL: reward totals are now tracked per instance.
• FastSAC sync: the TensorBoard writer is now closed when training fails.

Tests

• motrix_rl/tests/test_metrics.py covers the logger and both sinks: the disabled state, seq, NaN
  handling, the full-queue drop policy, retry and backoff, and close timeouts.
• motrix_rl/tests/fixtures/metrics_v1/*.json are golden fixtures that lock the protocol format,
  including the pipeline-side pending / failed records.
• test_rl_contracts.py covers the TrainerHandle lifecycle: finished, failed, killed and
  SystemExit(0).
• test_fastsac_live_metrics.py, test_skrl_live_metrics.py and test_rslrl_live_logger.py cover
  each provider's key mapping, step axes and events.
• Regression cases were added to test_fastsac_sync_trainer.py, test_skrl_ppo.py and
  test_rslrl_sim_backend.py.

Compatibility

• The task_config schema is unchanged, and existing runs' task_config.yaml and metadata.json
  still load.
• There is no new logging.backend value; live metrics are an additional output.
• With the env vars unset, behavior is identical to before.

Non-goals

This PR does not add a wandb sink, media upload, eval record production, or the server itself. The
protocol reserves the eval and media kinds for later.

[Copilot]

🔵 Needs a closer look

It is a large cross-cutting feature touching training lifecycle, a background daemon HTTP sink with concurrency/retry semantics, and async process orchestration across all three providers, which warrants human review despite thorough tests and no defects found.

0 open findings

What changed in this PR

This PR adds opt-in live streaming of training metrics over the motrix.metrics/v1 protocol. When MOTRIX_METRICS_URL is set, each training run writes every record to run_dir/metrics.jsonl and pushes the same records in gzip batches to an ingest service from a background daemon thread; a slow/unreachable/erroring service never blocks or fails training. With the env var unset the logger is a pure no-op (no file, no thread). The feature spans the motrix_rl control plane plus all three providers (FastSAC sync/async, SKRL torch/jax, RSLRL), each mapping its native TensorBoard tags onto shared canonical keys aligned on axes.env_steps. TensorBoard output and the terminal panel are unchanged.

Changes:

• New stdlib-only motrix_rl/metrics.py (MetricsLogger, JsonlSink, HttpSink) with bounded queue, scalar-drop policy, retry/backoff and heartbeats; TrainerHandle.train() emits the run lifecycle (running/finished/killed/failed) and AgentProvider.total_env_steps() feeds progress.
• Per-provider live emission: FastSAC live_metrics.system_scalars + sync/async scalar/event wiring, SKRL live_metrics.py tag mapping, RSLRL live_logger.LiveLogger; plus incidental correctness fixes (SKRL per-instance reward totals, RSLRL extras["log"] forwarding, FastSAC sync TB writer closed in finally).
• Extensive new/updated tests and golden protocol fixtures; design + plan docs under wiki/.

File	Description
motrix_rl/​src/​motrix_rl/​metrics.py	New core protocol logger and two sinks
motrix_rl/​src/​motrix_rl/​contracts.py	TrainerHandle.train() lifecycle, _failure_reason, metrics/run_info fields, total_env_steps hook
motrix_rl/​src/​motrix_rl/​runner.py	Build logger from env, annotate run, assemble running fields
motrix_rl/​src/​motrix_rl/​frameworks.py	Thread metrics through trainer context/handle
motrix_rl/​src/​motrix_rl/​runs.py	Extract task_config_snapshot reused by the running record
motrix_rl/​tests/​test_metrics.py	New comprehensive sink/logger tests
motrix_rl/​tests/​test_rl_contracts.py	TrainerHandle finished/killed/failed coverage
motrix_rl/​tests/​test_runner.py	Live-metrics-on-train vs disabled-on-play coverage
motrix_rl/​tests/​fixtures/​metrics_v1/​*.json	Golden protocol fixtures (pipeline + motrixlab)
motrix_rl_skrl/​src/​.../​live_metrics.py	New SKRL tag→canonical-key mapping
motrix_rl_skrl/​src/​.../​ppo.py	Attach live metrics, emit run_started/final checkpoint
motrix_rl_skrl/​src/​.../​torch/​train/​ppo.py	Per-instance reward totals; keyword-only write_tracking_data/write_checkpoint overrides (match skrl 2.1)
motrix_rl_skrl/​src/​.../​jax/​train/​ppo.py	Same overrides for JAX PPO
motrix_rl_skrl/​src/​.../​framework.py	total_env_steps
motrix_rl_skrl/​tests/​test_skrl_*.py	Mapping, isolation, lifecycle tests
motrix_rl_rslrl/​src/​.../​torch/​train/​live_logger.py	New RSL-RL LiveLogger subclass
motrix_rl_rslrl/​src/​.../​torch/​wrap_torch.py, wrap_np.py	Forward reward terms/metrics via extras["log"] as 0-dim tensors
motrix_rl_rslrl/​src/​.../​torch/​train/​ppo.py	Attach LiveLogger when enabled
motrix_rl_rslrl/​src/​.../​framework.py	total_env_steps
motrix_rl_rslrl/​tests/​test_rslrl_*.py	Logger mapping, RND, rank, real-runner tests
motrix_rl_builtin/​src/​.../​fastsac/​live_metrics.py	New system_scalars sys/* keys
motrix_rl_builtin/​src/​.../​fastsac/​sync/​train.py	Scalars/events + TB writer close in finally
motrix_rl_builtin/​src/​.../​fastsac/​async_impl/​train.py	Parent-only scalars/events, checkpoint notices, worker-crash reporting
motrix_rl_builtin/​src/​.../​fastsac/​async_impl/​worker.py	Per-save checkpoint notice via panel queue
motrix_rl_builtin/​src/​.../​fastsac/​framework.py	total_env_steps
motrix_rl_builtin/​tests/​*.py	Sync/async live-metrics tests, thread-backed spawn mocks
wiki/​design/​live-metrics.md, wiki/​plan/​live-metrics.md, *index.md	Design/plan docs and index entries

🧠 Review effort: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[wlgys8]

静态检查了一下 live metrics 对训练 hot path 的额外成本，建议重点关注以下几点（未跑 benchmark，暂不量化吞吐影响）：

1. RSLRL：开启 live 后每个 PPO iteration 重复聚合。 LiveLogger 先遍历 ep_extras，把每步标量转到 logger device，再执行 cat → mean → item，随后 super().log() 对同一窗口再次搬运、拼接和归约。对于 CPU 指标，CUDA logger 下 live 额外增加约「指标数 × rollout 步数」次小粒度 CPU→GPU 转移，并按指标回读均值；action_std 的均值和回读也重复。建议 live 与原生日志复用一次聚合后的 host 标量，而不是两端各算一次。

2. SKRL：开启 live 后，在每个 tracking write 日志点重复窗口均值。 live 对要发送的 tags 执行 float(np.mean(values))，随后原生 write_tracking_data 再聚合同一窗口。Torch 通常已是 host floats，主要是额外 CPU 遍历/临时数组/均值；JAX device scalars 还涉及设备值取回。min/max companion tags 已跳过。频率是 logging.interval，不是每个梯度 update。也建议复用原生聚合结果。

3. HTTP 异步，但 JSONL 仍同步。 MetricsLogger._emit 在调用线程持锁执行 JSON 序列化、file.write 和每条记录的 flush，之后才 HTTP 入队。因此 RSLRL 每个 iteration、SKRL/FastSAC sync 的日志点仍会等待本地 IO；run_dir 在慢盘或网络文件系统上时会直接阻塞训练。建议文件写出也后台化、批量 flush，并测试慢文件 IO 下的行为。

另需区分「开启 live 的增量」和「本 PR 相对基线的增量」：RSLRL wrapper 新增的每步 reward/metrics 归约和标量分配不受 live 开关影响，关闭 live 也执行；训练时原生日志会消费，但 play/headless 视频路径返回的 infos 未被使用，建议关闭该场景的诊断采集。这些 Torch reward 均值保留在设备上，不应误判为新增每步 GPU→CPU 同步。

FastSAC sync 正常日志路径复用已有 host scalars，没有为 live 再做一次 loss 设备归约；async 常规上报在父进程，未进入 collector/learner 每步计算路径。不过 checkpoint 通知的 put(timeout=5.0) 在队列满时会等待，且 live 关闭时也存在，属于低频背压风险。

建议优先消除 RSLRL 的重复聚合、无消费者路径的逐步诊断，以及训练线程上的同步 JSONL IO。

[luhengsw]

结论：开启 live metrics（只开 jsonl、只开 http 或两者都开）后，训练吞吐的变化都在两次运行之间的正常波动范围内。即使输出出了故障也是如此，包括：

• 磁盘慢、卡死、写入报错；
• ingest 服务响应慢、一直不应答、拒绝连接。

每次记录时，live 让训练线程多花约 10–20 µs；作为对比，同一次调用里 TensorBoard 本身就要约 400–550 µs。

为什么没有额外开销

手段	对训练线程的效果
统一的日志接口 TrainingLogger：训练器每个日志窗口只组装一份 scalars，同一份值同时写 TensorBoard 和发 live	不会重复归约，也不会重复从设备取数；每个窗口仍只有一次 host 同步，和本 PR 之前相同
训练线程只负责入队：MetricsLogger._emit 只做数值清洗、分配 seq，再放进两个内存队列	训练线程上没有文件 IO、网络 IO、JSON 序列化或 gzip
每个 sink 有自己的 daemon 线程：JsonlSink 和 TensorBoard 的 EventFileWriter 一样异步写出，每 2 s 批量写一次、flush 一次；HttpSink 每 2 s 或攒满 500 条发一批 gzip	磁盘或服务变慢、卡死，只会拖住它自己的线程
队列有上限，而且永不阻塞：队列满时丢最旧的 scalars，run 和 event 记录保留（TensorBoard 的做法是阻塞调用方）	即使 sink 一直恢复不了，也不会反压训练；内存有上限
close() 是异步的：只表示"没有后续数据了"，每个 sink 按自己的节奏收尾（请求超时 5 s，重试最多持续 60 s）	训练结束时不等待 live metrics
关闭时零成本：没有设置 MOTRIX_METRICS_* 环境变量时 logger 处于 disabled 状态，不建文件、不起线程；只发给 live 的采样（sys/*）只在开启 live 时才计算；FastSAC async 的 checkpoint 通知也只在开启 live 时发送，用的是 put_nowait	不用 live metrics 的训练行为不变

测试方法

bench/bench_live_metrics.py 在每个场景下跑同一个合成训练循环：

• 每步约 2 ms 的负载；
• 每次调用 TrainingLogger.scalars 写 40 个 key；
• 所有场景都开着 TensorBoard，因此和 off 的差值就是 live 本身的开销；
• ingest 服务放在独立进程里运行，不和训练进程争 GIL。

off, control 是把 off 原样再跑一遍，它的数字就是噪声基线。吞吐取多次重复的中位数。

两种负载：

• numpy：BLAS 计算期间释放 GIL，接近 torch 训练步的情况。
• python：训练线程全程持有 GIL，这是对后台 sink 线程最不利的情况。

--log-every 1 表示每次迭代都记录一次，远比真实训练频繁，属于最坏情况；--log-every 50 更接近实际使用。

python bench/bench_live_metrics.py --repeats 7                      # numpy，每次迭代都记录
python bench/bench_live_metrics.py --workload python --repeats 5    # 持有 GIL，每次迭代都记录
python bench/bench_live_metrics.py --log-every 50 --repeats 15      # numpy，每 50 次迭代记录一次

测试环境：AMD Ryzen 7 9700X（8 核 16 线程），Python 3.10.19，torch 2.7.0+cu128，机器空闲。

结果

开销 = 循环吞吐中位数相对 off 的变化，正数表示变慢。

场景	numpy，每次迭代	python（持有 GIL），每次迭代	numpy，每 50 次迭代
off, control（噪声基线）	-0.53%	+0.56%	+0.50%
jsonl	-0.67%	+1.64%	-0.46%
jsonl，慢盘（每次写入卡 0.5 s）	-1.33%	-0.47%	-0.24%
jsonl，磁盘卡死（写入永不返回）	-2.58%	-0.83%	-1.48%
jsonl，写入报错（每次写入都失败）	-1.35%	-1.42%	+0.60%
http	-0.54%	-1.37%	+0.39%
http，慢服务（0.5 s 才应答）	+0.37%	-1.11%	+0.14%
http，无响应（接受连接但永不应答）	-0.26%	-1.89%	+0.29%
http，拒绝连接	-0.31%	-0.90%	-1.29%
jsonl + http	+1.04%	+0.19%	-0.62%
磁盘卡死 + 服务无响应	+0.66%	-0.95%	-0.63%
范围	-2.6% … +1.0%	-1.9% … +1.6%	-1.5% … +0.6%

所有场景都在约 ±2.6% 以内，正负方向是随机的：正常输出并不比故障输出更快。这说明这些差异是测量噪声，而不是 live 的开销。

训练线程上每次 scalars 调用的耗时（p50）：

负载	off	开启 live	差值
numpy，每次迭代	519 µs	529–534 µs	+10–15 µs
numpy，每 50 次迭代	561 µs	573–578 µs	+12–18 µs
python（持有 GIL），每次迭代	389 µs	395–445 µs	+6–56 µs（上限部分是在等 GIL）

p99 的变化幅度和 p50 相同。

送达情况：

• 正常的 jsonl 和 http 收到了全部记录（2000 / 2000 和 40 / 40）。
• 出故障的输出丢弃了记录，但没有影响训练。
• "慢盘"场景只写入了部分记录（例如 1721 / 2000），这是 bench 本身的设置造成的：故障场景结束时 bench 不等待 sink 写完就开始统计。

说明与限制

• 单次调用偶尔变慢，关闭 live 时也会出现。 单次最慢的调用出现在"磁盘卡死 + 服务无响应"场景，耗时 160 ms。这是一次 gen-2 垃圾回收恰好发生在这次调用里。用 gc.callbacks 记录后，关闭 live 时最长的 gen-2 停顿是 107 ms，开启 live 时是 113 ms，回收次数相同，与 live metrics 无关。
• GIL。 sink 线程 JSON 序列化一批 500 条记录时，会持有 GIL 约 3 ms。这部分已经包含在上面的数字里。
• sink 卡死时的内存占用。 每条排队中的记录约 1.6 KiB（40 个 key）。由于队列有上限，jsonl 最多占约 156 MiB（10 万条），http 最多约 16 MiB（1 万条）。
• 进程退出。 sink 跑在 daemon 线程里，解释器退出时会直接杀掉它们。为此加了一个 atexit 钩子，最多等待 5 s，让训练完马上退出的脚本也能把最后几条记录（包括终态）发出去。正常情况下只需几毫秒；只有输出出故障时才会等满 5 s，而且每个进程只发生一次，发生在训练结束之后。

[luhengsw]

另： 去掉了rslrl/skrl的logger-metrics实现等后续在整理

[wlgys8]

rebase一下

[luhengsw]

rebase一下

complete