From b408e5a51115ce92b0c0a9fa6e7935fc8dfabb97 Mon Sep 17 00:00:00 2001
From: Qiyuan Chen <chenqiyuan1012@gmail.com>
Date: Wed, 4 Nov 2026 15:59:00 +0800
Subject: [PATCH 1/4] feat(perf): add reusable RL iteration accounting and
 persistent metrics

---
 src/twinkle/infra/_ray/ray_helper.py |  15 ++
 src/twinkle/model/base.py            |   7 +
 src/twinkle/perf/__init__.py         |  13 +-
 src/twinkle/perf/rl.py               | 317 +++++++++++++++++++++++++++
 src/twinkle/perf/rl_metrics.py       | 151 +++++++++++++
 tests/perf/test_rl_metrics.py        |  69 ++++++
 tests/perf/test_rl_recorder.py       | 213 ++++++++++++++++++
 7 files changed, 784 insertions(+), 1 deletion(-)
 create mode 100644 src/twinkle/perf/rl.py
 create mode 100644 src/twinkle/perf/rl_metrics.py
 create mode 100644 tests/perf/test_rl_metrics.py
 create mode 100644 tests/perf/test_rl_recorder.py

diff --git a/src/twinkle/infra/_ray/ray_helper.py b/src/twinkle/infra/_ray/ray_helper.py
index ed3e45c54..a73481088 100644
--- a/src/twinkle/infra/_ray/ray_helper.py
+++ b/src/twinkle/infra/_ray/ray_helper.py
@@ -68,6 +68,21 @@ def done(self) -> bool:
         """Whether every referenced task has finished (Future protocol alias of :meth:`ready`)."""
         return self.ready()
 
+    def wait(self):
+        """Wait for worker completion without pulling tensor payloads to the driver.
+
+        Downstream consumers still resolve the original references and propagate
+        worker exceptions. This is a measurement barrier, not result collection.
+        """
+        import ray
+
+        refs = [future for future in self._futures if isinstance(future, ray.ObjectRef)]
+        if refs:
+            _, pending = ray.wait(refs, num_returns=len(refs), timeout=self._timeout, fetch_local=False)
+            if pending:
+                raise TimeoutError(f'{self._method} did not complete within {self._timeout} seconds')
+        return self
+
     def ready(self) -> bool:
         """Whether every referenced task has finished, WITHOUT fetching any payload.
 
diff --git a/src/twinkle/model/base.py b/src/twinkle/model/base.py
index 18eebd11a..c5e40a2e6 100644
--- a/src/twinkle/model/base.py
+++ b/src/twinkle/model/base.py
@@ -297,6 +297,13 @@ def apply_patch(self, patch_cls: Union[Patch, Type[Patch], str], **kwargs) -> No
     def add_metric(self, metric_cls: Union[Metric, str], is_training: Optional[bool] = None, **kwargs) -> None:
         ...
 
+    @remote_function(dispatch='all', collect='none', lazy_collect=False)
+    def performance_memory(self) -> dict[str, float]:
+        """Per-rank allocator lifetime peaks; call after the actor update."""
+        from twinkle.perf.rl_metrics import accelerator_memory_metrics
+
+        return accelerator_memory_metrics()
+
     @abstractmethod
     def calculate_metric(self, is_training: bool, **kwargs) -> Dict[str, Any]:
         ...
diff --git a/src/twinkle/perf/__init__.py b/src/twinkle/perf/__init__.py
index 4b2c2eeae..ce77e3e8f 100644
--- a/src/twinkle/perf/__init__.py
+++ b/src/twinkle/perf/__init__.py
@@ -1,7 +1,7 @@
 # Copyright (c) ModelScope Contributors. All rights reserved.
 """twinkle.perf -- a general-purpose performance observability component.
 
-Four independent, low-intrusion parts:
+Five independent, low-intrusion parts:
 
 * :class:`TimeRecorder` / :func:`timed` / :func:`span` -- a synchronous- and
   async-agnostic wall-clock span recorder that nests into an overall timing tree
@@ -21,6 +21,10 @@
   Chrome Trace rendering and a text DAG / lane report. Consumed by whoever holds
   the window's events; the recorder itself only appends.
 
+* :class:`RLPerformanceRecorder` -- explicit rollout-batch accounting, phase timing,
+  durable JSONL records and final summaries. Accepts lengths, integer update counts
+  and plain observation dictionaries, independently of any training recipe.
+
 The recorder and profiler are disabled/no-op by default, so importing perf never
 changes existing behaviour; the counter registry is always live by design (a few
 float additions per step). None of the parts imports twinkle/swift business
@@ -37,11 +41,18 @@
 from .counters import CounterRegistry, default_counters
 from .module_profiler import ModuleProfiler, activation_stats, build_module_profiler, param_stats
 from .recorder import TimeRecorder, default_recorder, disable_perf, enable_perf, span, timed
+from .rl import RLPerformanceRecorder, reduce_memory, sample_lengths
+from .rl_metrics import JsonlMetricWriter, compute_rl_performance_metrics
 from .stats import describe_shape, primary_tensor, tensor_moments, tensor_numel
 from .timeline import chrome_trace, dag_report
 
 __all__ = [
     'Perf',
+    'RLPerformanceRecorder',
+    'JsonlMetricWriter',
+    'compute_rl_performance_metrics',
+    'reduce_memory',
+    'sample_lengths',
     'TimeRecorder',
     'default_recorder',
     'timed',
diff --git a/src/twinkle/perf/rl.py b/src/twinkle/perf/rl.py
new file mode 100644
index 000000000..a2f22e058
--- /dev/null
+++ b/src/twinkle/perf/rl.py
@@ -0,0 +1,317 @@
+# Copyright (c) ModelScope Contributors. All rights reserved.
+"""Reusable RL iteration accounting, phase timing and persistent metrics.
+
+The iteration axis is independent of optimizer updates. Never average per-step
+throughputs to compare a run: sum tokens / sum seconds / physical allocation.
+"""
+
+import logging
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from pathlib import Path
+from time import perf_counter
+
+from .rl_metrics import JsonlMetricWriter, compute_rl_performance_metrics
+
+
+def sample_lengths(sample):
+    """Unpadded initial prompt and full response, including tool observations.
+
+    Twinkle stores shifted labels/masks. Policy tokens identify the first response
+    position; subsequent masked tool tokens belong to the full response span too.
+    """
+    encoded = sample.encoded
+    ids = list(encoded["input_ids"])
+    labels = list(encoded["labels"])
+    if len(labels) != len(ids):
+        raise ValueError("performance metrics require one shifted label per input token")
+    labels = labels[-1:] + labels[:-1]
+    mask = encoded.get("completion_mask")
+    if mask is not None:
+        mask = list(mask)
+        if len(mask) != len(ids):
+            raise ValueError("completion mask and input_ids must have equal lengths")
+        mask = mask[-1:] + mask[:-1]
+    attention = encoded.get("attention_mask")
+    attention = list(attention) if attention is not None else [1] * len(ids)
+    if len(attention) != len(ids):
+        raise ValueError("performance metrics require a one-dimensional attention mask")
+    positions = [i for i, label in enumerate(labels) if label != -100 and (mask is None or mask[i]) and attention[i]]
+    first = positions[0] if positions else len(ids)
+    prompt = sum(bool(x) for x in attention[:first])
+    response = sum(bool(x) for x in attention[first:])
+    return prompt, response
+
+
+def reduce_memory(rows):
+    """Reduce allocator peaks by rank maximum and other memory observations by mean."""
+    rows = [rows] if isinstance(rows, dict) else rows
+    keys = set().union(*(row.keys() for row in rows))
+    result = {}
+    for key in keys:
+        values = [row[key] for row in rows if key in row]
+        result[key] = max(values) if "max" in key else sum(values) / len(values)
+    return result
+
+
+class RLPerformanceRecorder:
+    def __init__(
+        self,
+        output_dir,
+        device_groups,
+        *,
+        mode="synchronous",
+        run_id=None,
+        parameter_sync_step=1,
+        clock=perf_counter,
+        logger=None,
+        metadata=None,
+        max_forward_workers=3,
+        track_batch_wait=False,
+        memory_prefix=None,
+    ):
+        self.writer = JsonlMetricWriter(Path(output_dir) / "performance.jsonl", run_id=run_id)
+        self.summary_writer = JsonlMetricWriter(
+            Path(output_dir) / "performance_summary.jsonl", run_id=self.writer.run_id
+        )
+        self.wall_clock = clock
+        self.excluded_seconds = 0.0
+        self.clock = lambda: self.wall_clock() - self.excluded_seconds
+        self.groups = device_groups
+        self.n_gpus = len({rank for _, ranks in device_groups for rank in ranks})
+        if not self.n_gpus:
+            raise ValueError("missing physical device allocation for RL performance metrics")
+        self.mode = mode
+        self.logger = logger or logging.getLogger(__name__)
+        self.metadata = dict(metadata or {})
+        self.max_forward_workers = max_forward_workers
+        self.track_batch_wait = track_batch_wait
+        self.memory_prefix = memory_prefix
+        self.iteration = 0
+        self.active = False
+        self.prompt_lengths = None
+        self.response_lengths = None
+        self.truncated = None
+        self.assistant_turns = None
+        self.memory = None
+        self._publish_time = 0.0
+        self._publishes = 0
+        self.parameter_sync_step = parameter_sync_step
+        self.total_tokens = 0
+        self.total_seconds = 0.0
+        self._forward_executor = None
+        self._forward_observers = []
+
+    def begin(self, optimizer_step):
+        self.timings = {}
+        self.start_step = optimizer_step
+        self.prompt_lengths = None
+        self.response_lengths = None
+        self.truncated = None
+        self.assistant_turns = None
+        self.memory = None
+        self.sample_visits = 0
+        self.trained_indices = set()
+        self.started = self.clock()
+        self.wall_started = self.wall_clock()
+        self.excluded_at_start = self.excluded_seconds
+        self.active = True
+
+    @contextmanager
+    def exclude_logging(self):
+        """Exclude reporting I/O from logical time while retaining full wall time."""
+        started = self.wall_clock()
+        try:
+            yield
+        finally:
+            self.excluded_seconds += self.wall_clock() - started
+
+    @contextmanager
+    def stage(self, name):
+        started = self.clock()
+        try:
+            yield
+        finally:
+            self.timings[name] = self.timings.get(name, 0.0) + self.clock() - started
+
+    def call(self, name, fn, *args, collect_result=False, **kwargs):
+        with self.stage(name):
+            result = fn(*args, **kwargs)
+            # Complete the actual remote work; do not time only RPC submission.
+            if getattr(result, "_is_lazy_collect", False):
+                if collect_result:
+                    # These small side-effect results have no downstream tensor
+                    # consumer to surface failures. Waiting alone is insufficient.
+                    result = result.result()
+                else:
+                    result.wait()
+            return result
+
+    def launch_forward(self, name, fn, *args, **kwargs):
+        """Observe independent RPCs without serializing auxiliary model lanes.
+
+        Each span is submission-to-completion wall time, including queueing. They
+        may overlap and must not be added to reconstruct iteration time.
+        """
+        started = self.clock()
+        result = fn(*args, **kwargs)
+        if getattr(result, "_is_lazy_collect", False):
+            if self._forward_executor is None:
+                self._forward_executor = ThreadPoolExecutor(
+                    max_workers=self.max_forward_workers, thread_name_prefix="rl-metric"
+                )
+
+            def observe():
+                result.wait()  # no payload transfer; consumer propagates task errors
+                return self.clock() - started
+
+            self._forward_observers.append((name, self._forward_executor.submit(observe)))
+        else:
+            self.timings[name] = self.timings.get(name, 0.0) + self.clock() - started
+        return result
+
+    def join_forwards(self):
+        """The advantage consumer already depends on every auxiliary lane."""
+        pending, self._forward_observers = self._forward_observers, []
+        for name, observer in pending:
+            self.timings[name] = self.timings.get(name, 0.0) + observer.result()
+
+    def received(self, samples):
+        """Adapt encoded rollout samples; callers may instead supply lengths directly."""
+        lengths = [sample_lengths(sample) for sample in samples]
+        turns = [getattr(sample, "rollout_infos", {}).get("turns") for sample in samples]
+        self.received_lengths(
+            [prompt for prompt, _ in lengths],
+            [response for _, response in lengths],
+            truncated=[bool(getattr(sample, "truncated", False)) for sample in samples],
+            assistant_turns=turns if turns and all(isinstance(n, (int, float)) for n in turns) else None,
+        )
+
+    def received_lengths(self, prompt_lengths, response_lengths, *, truncated=None, assistant_turns=None):
+        """Register a rollout batch without depending on any training/sample class."""
+        prompts, responses = list(prompt_lengths), list(response_lengths)
+        if not prompts or len(prompts) != len(responses):
+            raise ValueError("expected a nonempty batch of paired prompt/response lengths")
+        flags = [False] * len(prompts) if truncated is None else list(truncated)
+        turns = None if assistant_turns is None else list(assistant_turns)
+        if len(flags) != len(prompts) or (turns is not None and len(turns) != len(prompts)):
+            raise ValueError("per-sample observations must match the batch length")
+        self.prompt_lengths, self.response_lengths = prompts, responses
+        self.truncated, self.assistant_turns = flags, turns
+        if self.track_batch_wait:
+            self.timings["gen"] = self.clock() - self.started
+
+    def record_training(self, sample_indices):
+        """Count visits across replay while retaining unique original-batch indices."""
+        indices = list(sample_indices)
+        if self.prompt_lengths is None or any(not 0 <= index < len(self.prompt_lengths) for index in indices):
+            raise ValueError("training indices must refer to the registered rollout batch")
+        self.sample_visits += len(indices)
+        self.trained_indices.update(indices)
+
+    def finish(
+        self, optimizer_step, *, stream_stats=None, resource_metrics=None, memory=None, terminal=False, publications=0
+    ):
+        if not self.active or self.prompt_lengths is None:
+            return
+        ended = self.clock()
+        wall_elapsed = self.wall_clock() - self.wall_started
+        self.active = False
+        timings = dict(self.timings, step=ended - self.started)
+        extra = {}
+        if stream_stats is not None:
+            stats = dict(stream_stats)
+            if "publish_time" in stats:
+                publish_time = stats["publish_time"]
+                timings["update_weights"] = publish_time - self._publish_time
+                self._publish_time = publish_time
+            if "publishes" in stats:
+                publications = stats["publishes"] - self._publishes
+                self._publishes = stats["publishes"]
+            extra.update(
+                {
+                    f"stream/{key}": value
+                    for key, value in stats.items()
+                    if isinstance(value, (int, float))
+                    and key not in ("total_time", "final_version", "dropped_incomplete")
+                }
+            )
+        extra.update(resource_metrics or {})
+        metrics = compute_rl_performance_metrics(self.prompt_lengths, self.response_lengths, timings, self.n_gpus)
+        metrics.update(extra)
+        turns = self.assistant_turns
+        if turns is not None and all(isinstance(n, (int, float)) for n in turns):
+            # Count assistant generations, independently of user/tool message counts.
+            metrics.update(
+                {
+                    "rollout/assistant_turns/mean": sum(turns) / len(turns),
+                    "rollout/assistant_turns/min": min(turns),
+                    "rollout/assistant_turns/max": max(turns),
+                }
+            )
+        metrics.update(memory or {})
+        if self.memory_prefix:
+            metrics.update({f"{self.memory_prefix}/{key}": value for key, value in (memory or {}).items()})
+        metrics.update(
+            {
+                "training/optimizer_step_start": self.start_step,
+                "training/weight_publications_this_iteration": publications,
+                "training/optimizer_step": optimizer_step,
+                "training/optimizer_steps_this_iteration": optimizer_step - self.start_step,
+                "training/sample_visits": self.sample_visits,
+                "training/unique_samples_trained": len(self.trained_indices),
+                "training/samples_not_trained": len(self.prompt_lengths) - len(self.trained_indices),
+                "response/truncated_ratio": sum(bool(flag) for flag in self.truncated) / len(self.truncated),
+                "perf/gpu_seconds": timings["step"] * self.n_gpus,
+                "observability/iteration_wall_seconds": wall_elapsed,
+                "observability/excluded_logging_seconds": self.excluded_seconds - self.excluded_at_start,
+            }
+        )
+        self.iteration += 1
+        self.writer.write(
+            self.iteration,
+            metrics,
+            schema_version=1,
+            step_unit="rl_iteration",
+            iteration_scope="one_consumed_rollout_batch",
+            parameter_sync_step=self.parameter_sync_step,
+            parameter_sync_step_unit="optimizer_update",
+            mode=self.mode,
+            device_groups=self.groups,
+            token_scope="initial_prompt_plus_full_response_including_tool_observations",
+            terminal=terminal,
+            **self.metadata,
+            unavailable_metrics=["perf/mfu/actor"],
+            unsupported_mfu_reason="no validated model FLOPs and device peak FLOPs estimator",
+        )
+        self.total_tokens += metrics["perf/total_num_tokens"]
+        self.total_seconds += timings["step"]
+        self.logger.info(
+            "[perf] iteration=%s step=%s seconds=%.3f tokens=%s tok/s/GPU=%.3f",
+            self.iteration,
+            optimizer_step,
+            timings["step"],
+            metrics["perf/total_num_tokens"],
+            metrics["perf/throughput"],
+        )
+
+    def write_summary(self, optimizer_step, *, stream_stats=None, status="completed"):
+        """Call after driver teardown: terminal drops/final_version now exist."""
+        data = {
+            "completed_iterations": self.iteration,
+            "optimizer_step": optimizer_step,
+            "total_num_tokens": self.total_tokens,
+            "sum_iteration_seconds": self.total_seconds,
+            "allocated_gpus": self.n_gpus,
+            "throughput": (self.total_tokens / (self.total_seconds * self.n_gpus) if self.total_seconds else None),
+        }
+        if stream_stats is not None:
+            data.update({f"stream/{k}": v for k, v in stream_stats.items() if isinstance(v, (int, float))})
+        self.summary_writer.write(self.iteration, data, step_unit="run_summary", status=status)
+
+    def close(self):
+        self.active = False
+        self.writer.close()
+        self.summary_writer.close()
+        if self._forward_executor is not None:
+            self._forward_executor.shutdown(wait=False, cancel_futures=True)
diff --git a/src/twinkle/perf/rl_metrics.py b/src/twinkle/perf/rl_metrics.py
new file mode 100644
index 000000000..d210bd827
--- /dev/null
+++ b/src/twinkle/perf/rl_metrics.py
@@ -0,0 +1,151 @@
+# Copyright (c) ModelScope Contributors. All rights reserved.
+"""RL iteration metrics with explicit token, time and allocation scopes.
+
+This module is deliberately independent of torch/Ray. Counts are unpadded sequence
+lengths, once per trajectory in an RL batch, NOT once per optimizer update/replay.
+Timing is supplied by the orchestration layer; a formula cannot fix a mismatched
+measurement boundary. Callers supply explicit iteration and phase boundaries.
+"""
+
+from __future__ import annotations
+import math
+import threading
+import time
+import uuid
+from pathlib import Path
+
+import json
+
+
+def compute_rl_performance_metrics(
+    prompt_lengths, response_lengths, timing_raw, n_gpus, *, prompt_width=None, response_width=None
+):
+    """Throughput, lengths and per-token timing with explicit denominators.
+
+    Ragged batches use the batch maximum as padded width. ``clip_ratio`` therefore
+    counts rows equal to that width, not rows that hit the configured generation
+    cap. Explicit widths also support fixed-width batches.
+    """
+    prompts, responses = list(prompt_lengths), list(response_lengths)
+    if len(prompts) != len(responses) or not prompts:
+        raise ValueError("expected a nonempty batch of paired prompt/response lengths")
+    if any(not isinstance(n, int) or isinstance(n, bool) or n < 0 for n in prompts + responses):
+        raise ValueError("token lengths must be nonnegative integers")
+    elapsed = timing_raw["step"]
+    if not isinstance(n_gpus, int) or isinstance(n_gpus, bool) or n_gpus < 1:
+        raise ValueError("n_gpus must count allocated physical devices, including rollout")
+    if elapsed <= 0 or any(not math.isfinite(t) or t < 0 for t in timing_raw.values()):
+        raise ValueError("timings must be finite and nonnegative, with positive step time")
+    p, r = sum(prompts), sum(responses)
+    total = p + r
+    out = {
+        "perf/total_num_tokens": total,
+        "perf/time_per_step": elapsed,
+        "perf/throughput": total / (elapsed * n_gpus),
+        "perf/num_gpus": n_gpus,
+        "perf/num_samples": len(prompts),
+        "perf/num_prompt_tokens": p,
+        "perf/num_response_tokens": r,
+        "perf/throughput_total": total / elapsed,
+        "perf/response_tokens_per_second": r / elapsed,
+        "perf/samples_per_second": len(prompts) / elapsed,
+        **{f"timing_s/{name}": t for name, t in timing_raw.items()},
+    }
+    denominators = {"gen": r, **dict.fromkeys(("ref", "values", "adv", "update_critic", "update_actor"), total)}
+    for name, count in denominators.items():
+        if name in timing_raw:
+            out[f"timing_per_token_ms/{name}"] = timing_raw[name] * 1000 / count if count else 0.0
+    for name, lengths, width in (
+        ("prompt_length", prompts, prompt_width),
+        ("response_length", responses, response_width),
+    ):
+        width = max(lengths) if width is None else width
+        if width < max(lengths):
+            raise ValueError("padded width cannot be smaller than an actual sequence")
+        out.update(
+            {
+                f"{name}/mean": sum(lengths) / len(lengths),
+                f"{name}/min": min(lengths),
+                f"{name}/max": max(lengths),
+                f"{name}/clip_ratio": sum(n == width for n in lengths) / len(lengths),
+            }
+        )
+    active = [n for n in responses if n > 0]
+    width = max(responses) if response_width is None else response_width
+    out["response/aborted_ratio"] = 1 - len(active) / len(responses)
+    out.update(
+        {
+            "response_length_non_aborted/mean": sum(active) / len(active) if active else float("nan"),
+            "response_length_non_aborted/min": min(active) if active else float("nan"),
+            "response_length_non_aborted/max": max(active) if active else float("nan"),
+            "response_length_non_aborted/clip_ratio": sum(n == width for n in active) / len(active)
+            if active
+            else float("nan"),
+        }
+    )
+    return out
+
+
+def _json_value(value):
+    # Nonfinite numbers become JSON null, never invalid NaN literals.
+    if isinstance(value, float) and not math.isfinite(value):
+        return None
+    if isinstance(value, dict):
+        return {str(k): _json_value(v) for k, v in value.items()}
+    if isinstance(value, (tuple, list)):
+        return [_json_value(v) for v in value]
+    return value
+
+
+class JsonlMetricWriter:
+    """Incremental, append-only local output with a {step, data} envelope.
+
+    A new run_id on every construction disambiguates resumed/repeated runs in the
+    same directory. Each record is flushed; this does not claim power-loss fsync.
+    """
+
+    def __init__(self, path, *, run_id=None):
+        self.path = Path(path)
+        self.run_id = run_id or uuid.uuid4().hex
+        self._stream = None
+        self._lock = threading.Lock()
+
+    def write(self, step, data, **metadata):
+        row = _json_value(dict(step=step, data=data, run_id=self.run_id, timestamp=time.time(), **metadata))
+        line = json.dumps(row, ensure_ascii=False, allow_nan=False) + "\n"
+        with self._lock:
+            if self._stream is None:
+                self.path.parent.mkdir(parents=True, exist_ok=True)
+                self._stream = self.path.open("a", encoding="utf-8")
+            self._stream.write(line)
+            self._stream.flush()
+
+    def close(self):
+        with self._lock:
+            if self._stream is not None:
+                self._stream.close()
+                self._stream = None
+
+
+def accelerator_memory_metrics():
+    """Per-worker allocator high-water marks and host-used RAM.
+
+    These are lifetime peaks (no reset), not device-wide memory usage. CPU RAM is
+    host-wide virtual_memory().used, NOT process RSS. Unsupported counters are
+    absent, not zero. This helper launches no accelerator work.
+    """
+    import psutil
+    import torch
+
+    from twinkle import Platform
+
+    device = getattr(torch, Platform.device_prefix(), None)
+    result = {"perf/cpu_memory_used_gb": psutil.virtual_memory().used / 1024**3}
+    if device is not None and getattr(device, "is_available", lambda: False)():
+        # The update must have actually finished, not just queued optimizer kernels.
+        device.synchronize()
+        for name in ("max_memory_allocated", "max_memory_reserved"):
+            fn = getattr(device, name, None)
+            if fn is not None:
+                result[f"perf/{name}_gb"] = fn() / 1024**3
+    return result
diff --git a/tests/perf/test_rl_metrics.py b/tests/perf/test_rl_metrics.py
new file mode 100644
index 000000000..eb6a3fb2e
--- /dev/null
+++ b/tests/perf/test_rl_metrics.py
@@ -0,0 +1,69 @@
+"""Numerical contracts for rollout accounting, padding and durable records."""
+
+import json
+import math
+
+import pytest
+
+from twinkle.perf import JsonlMetricWriter, compute_rl_performance_metrics
+
+
+@pytest.mark.parametrize("widths,clip,non_aborted_clip", [((None, None), 1 / 3, 0.5), ((12, 8), 0.0, 0.0)])
+def test_variable_lengths_and_explicit_padded_width(widths, clip, non_aborted_clip):
+    timings = dict(step=7.25, gen=3.5, ref=0.25, old_log_prob=0.4, adv=0.01, update_actor=2, update_weights=0.5)
+    actual = compute_rl_performance_metrics(
+        [3, 6, 4], [0, 2, 5], timings, 8, prompt_width=widths[0], response_width=widths[1]
+    )
+    expected = {
+        "perf/total_num_tokens": 20,
+        "perf/num_prompt_tokens": 13,
+        "perf/num_response_tokens": 7,
+        "perf/num_samples": 3,
+        "perf/num_gpus": 8,
+        "perf/time_per_step": 7.25,
+        "perf/throughput": 0.3448275862068966,
+        "timing_per_token_ms/gen": 500,
+        "timing_per_token_ms/ref": 12.5,
+        "timing_per_token_ms/adv": 0.5,
+        "timing_per_token_ms/update_actor": 100,
+        "prompt_length/mean": 13 / 3,
+        "prompt_length/min": 3,
+        "prompt_length/max": 6,
+        "prompt_length/clip_ratio": clip,
+        "response_length/mean": 7 / 3,
+        "response_length/min": 0,
+        "response_length/max": 5,
+        "response_length/clip_ratio": clip,
+        "response/aborted_ratio": 1 / 3,
+        "response_length_non_aborted/mean": 3.5,
+        "response_length_non_aborted/min": 2,
+        "response_length_non_aborted/max": 5,
+        "response_length_non_aborted/clip_ratio": non_aborted_clip,
+    }
+    for key, value in expected.items():
+        assert actual[key] == pytest.approx(value), key
+    assert "timing_per_token_ms/old_log_prob" not in actual
+
+
+def test_all_empty_responses_have_no_non_aborted_population():
+    actual = compute_rl_performance_metrics([2, 3], [0, 0], {"step": 7.25, "gen": 3.5, "update_actor": 2}, 8)
+    assert actual["perf/total_num_tokens"] == 5
+    assert actual["perf/throughput"] == pytest.approx(0.08620689655172414)
+    assert actual["response/aborted_ratio"] == 1
+    assert actual["response_length/clip_ratio"] == 1
+    assert actual["timing_per_token_ms/gen"] == 0
+    assert actual["timing_per_token_ms/update_actor"] == 400
+    assert math.isnan(actual["response_length_non_aborted/mean"])
+
+
+def test_invalid_allocation_or_timing_fails_instead_of_faking_throughput():
+    for n_gpus, seconds in [(0, 1), (2, 0), (2, float("nan"))]:
+        with pytest.raises(ValueError):
+            compute_rl_performance_metrics([3], [2], {"step": seconds}, n_gpus)
+
+
+def test_jsonl_flush_and_nonfinite(tmp_path):
+    writer = JsonlMetricWriter(tmp_path / "metrics.jsonl")
+    writer.write(1, {"good": 1.5, "nan": float("nan")})
+    assert json.loads(writer.path.read_text())["data"] == {"good": 1.5, "nan": None}
+    writer.close()
diff --git a/tests/perf/test_rl_recorder.py b/tests/perf/test_rl_recorder.py
new file mode 100644
index 000000000..4f580292f
--- /dev/null
+++ b/tests/perf/test_rl_recorder.py
@@ -0,0 +1,213 @@
+"""Framework-independent rollout iteration recording and future observation."""
+
+import json
+from types import SimpleNamespace as NS
+
+import pytest
+
+from twinkle.perf import RLPerformanceRecorder, reduce_memory, sample_lengths
+
+
+def test_plain_lengths_and_update_numbers_need_no_training_objects(tmp_path):
+    now = [0.0]
+    perf = RLPerformanceRecorder(tmp_path, [("train", [0, 1]), ("shared", [0])], clock=lambda: now[0])
+    perf.begin(4)
+    perf.received_lengths([3, 4], [2, 3], truncated=[False, True], assistant_turns=[1, 2])
+    perf.record_training([0, 1, 0, 1])
+    now[0] = 2.0
+    perf.finish(6, publications=1, resource_metrics={"resource/replicas": 2})
+    perf.write_summary(6)
+    perf.close()
+    row = json.loads((tmp_path / "performance.jsonl").read_text())
+    assert row["data"]["perf/throughput"] == 3
+    assert row["data"]["training/sample_visits"] == 4
+    assert row["data"]["training/unique_samples_trained"] == 2
+    assert row["data"]["training/optimizer_steps_this_iteration"] == 2
+    assert row["data"]["training/weight_publications_this_iteration"] == 1
+    assert row["data"]["response/truncated_ratio"] == 0.5
+    assert row["data"]["resource/replicas"] == 2
+
+
+def test_plain_snapshot_publications_use_actual_counter_deltas(tmp_path):
+    now = [0.0]
+    perf = RLPerformanceRecorder(tmp_path, [("train", [0])], clock=lambda: now[0], track_batch_wait=True)
+    for step, publish_time, publications in [(1, 0.5, 1), (2, 0.5, 1)]:
+        perf.begin(step - 1)
+        now[0] += 1
+        perf.received_lengths([3], [2])
+        now[0] += 1
+        perf.finish(step, stream_stats={"publish_time": publish_time, "publishes": publications})
+    perf.close()
+    rows = [json.loads(line)["data"] for line in (tmp_path / "performance.jsonl").read_text().splitlines()]
+    assert [row["training/weight_publications_this_iteration"] for row in rows] == [1, 0]
+    assert [row["timing_s/update_weights"] for row in rows] == [0.5, 0]
+    assert all(row["timing_s/gen"] == 1 for row in rows)
+
+
+def sample(prompt=3, response=2, *, tool=0):
+    ids = list(range(prompt + response + tool))
+    labels = [-100] * prompt + ids[prompt : prompt + response] + [-100] * tool
+    return NS(encoded={"input_ids": ids, "labels": labels[1:] + labels[:1]}, truncated=False)
+
+
+def test_lengths_include_tool_observations_but_not_padding():
+    s = sample(tool=4)
+    assert sample_lengths(s) == (3, 6)
+    s.encoded["input_ids"] += [0, 0]
+    # Output-order labels for 3 prompt, 2 assistant, 4 observation and 2 padding tokens.
+    labels = [-100] * 3 + [3, 4] + [-100] * 6
+    s.encoded["labels"] = labels[1:] + labels[:1]
+    s.encoded["attention_mask"] = [1] * 9 + [0, 0]
+    assert sample_lengths(s) == (3, 6)
+    assert sample_lengths(sample(response=0)) == (3, 0)
+
+
+def test_iteration_counts_tokens_once_across_optimizer_replay_and_flushes(tmp_path):
+    now = [10.0]
+    perf = RLPerformanceRecorder(
+        tmp_path, [("trainer", [0, 1]), ("shared", [0]), ("rollout", [2, 3])], mode="colocate", clock=lambda: now[0]
+    )
+    perf.begin(7)
+    perf.received([sample(), sample(prompt=4, response=3)])
+    perf.sample_visits = 8  # four epochs over two samples
+    perf.trained_indices = {0, 1}
+    now[0] = 12.0
+    perf.finish(11)
+    row = json.loads((tmp_path / "performance.jsonl").read_text())  # before close
+    assert row["step_unit"] == "rl_iteration"
+    assert row["data"]["perf/total_num_tokens"] == 12
+    assert row["data"]["perf/throughput"] == 1.5  # 12 tokens / 2s / 4 physical GPUs
+    assert row["data"]["training/optimizer_steps_this_iteration"] == 4
+    assert row["data"]["training/sample_visits"] == 8
+    perf.close()
+
+
+def test_remote_stage_waits_without_collecting_payload(tmp_path):
+    now = [0.0]
+    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate", clock=lambda: now[0])
+    perf.begin(0)
+
+    class Future:
+        _is_lazy_collect = True
+
+        def wait(self):
+            now[0] += 3.0
+
+        def __call__(self):
+            pytest.fail("must not transfer tensor payloads to the driver")
+
+    future = Future()
+    assert perf.call("ref", lambda: future) is future
+    assert perf.timings["ref"] == 3.0
+
+
+def test_memory_reduction_is_rank_max_for_peaks_host_mean_for_ram():
+    result = reduce_memory(
+        [
+            {"perf/max_memory_allocated_gb": 3, "perf/cpu_memory_used_gb": 8},
+            {"perf/max_memory_allocated_gb": 5, "perf/cpu_memory_used_gb": 10},
+        ]
+    )
+    assert result == {"perf/max_memory_allocated_gb": 5, "perf/cpu_memory_used_gb": 9}
+
+
+def test_side_effect_future_failure_is_not_a_successful_update(tmp_path):
+    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate")
+    perf.begin(0)
+
+    class FailedUpdate:
+        _is_lazy_collect = True
+
+        def wait(self):
+            return self
+
+        def result(self):
+            raise RuntimeError("optimizer failed")
+
+    with pytest.raises(RuntimeError, match="optimizer failed"):
+        perf.call("optimizer", FailedUpdate, collect_result=True)
+
+
+def test_reporter_time_excluded_but_full_wall_time_is_saved(tmp_path):
+    now = [0.0]
+    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate", clock=lambda: now[0])
+    perf.begin(0)
+    perf.received([sample()])
+    now[0] = 2.0
+    with perf.exclude_logging():
+        now[0] = 5.0
+    now[0] = 7.0
+    perf.finish(1)
+    data = json.loads((tmp_path / "performance.jsonl").read_text())["data"]
+    assert data["perf/time_per_step"] == 4.0
+    assert data["observability/iteration_wall_seconds"] == 7.0
+    assert data["observability/excluded_logging_seconds"] == 3.0
+    perf.close()
+
+
+def test_auxiliary_forwards_keep_original_parallel_submission(tmp_path):
+    import threading
+
+    ready = threading.Event()
+    launches = []
+    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate")
+    perf.begin(0)
+
+    class Forward:
+        _is_lazy_collect = True
+
+        def wait(self):
+            assert ready.wait(2), "forwards were serialized before all lanes could launch"
+
+        def __call__(self):
+            pytest.fail("do not collect tensor payloads")
+
+    def submit(name):
+        launches.append(name)
+        if len(launches) == 3:
+            ready.set()
+        return Forward()
+
+    try:
+        for name in ("ref", "teacher", "old_log_prob"):
+            perf.launch_forward(name, submit, name)
+        perf.join_forwards()
+        assert set(perf.timings) == {"ref", "teacher", "old_log_prob"}
+        assert all(value > 0 for value in perf.timings.values())
+    finally:
+        ready.set()
+        perf.close()
+
+
+def test_summary_records_terminal_driver_counters_after_teardown(tmp_path):
+    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="adaptive", parameter_sync_step=2)
+    perf.begin(0)
+    perf.received([sample()])
+    stats = {"publish_time": 0.0, "publishes": 0, "dropped_incomplete": 0, "total_time": 0.0, "final_version": 0}
+    perf.finish(1, stream_stats=stats)
+    stats.update(dropped_incomplete=3, total_time=12.0, final_version=2)
+    perf.write_summary(1, stream_stats=stats, status="loop_completed")
+    summary = json.loads((tmp_path / "performance_summary.jsonl").read_text())
+    row = json.loads((tmp_path / "performance.jsonl").read_text())
+    assert summary["run_id"] == row["run_id"]
+    assert summary["data"]["stream/dropped_incomplete"] == 3
+    assert summary["data"]["stream/final_version"] == 2
+    assert summary["data"]["stream/total_time"] == 12.0
+    assert "stream/total_time" not in row["data"]
+    assert "stream/dropped_incomplete" not in row["data"]
+    assert row["parameter_sync_step_unit"] == "optimizer_update"
+    assert row["iteration_scope"] == "one_consumed_rollout_batch"
+    perf.close()
+
+
+def test_assistant_turn_count_has_an_explicit_unit(tmp_path):
+    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate")
+    perf.begin(0)
+    s = sample()
+    s.rollout_infos = {"turns": 1}  # Count assistant generations independently of user/tool messages.
+    perf.received([s])
+    perf.finish(1)
+    data = json.loads((tmp_path / "performance.jsonl").read_text())["data"]
+    assert data["rollout/assistant_turns/mean"] == 1
+    assert "num_turns/mean" not in data
+    perf.close()

From 0db48d7240b1a3731bb83a19feee9180b13f2788 Mon Sep 17 00:00:00 2001
From: Qiyuan Chen <chenqiyuan1012@gmail.com>
Date: Wed, 4 Nov 2026 15:59:12 +0800
Subject: [PATCH 2/4] feat(async-rl): export iteration and resource performance
 observations

---
 .../async_rl/cluster_adapter.py               | 12 ++++
 .../async_rl/streaming_driver.py              | 31 +++++++-
 .../test_iteration_observers.py               | 71 +++++++++++++++++++
 3 files changed, 113 insertions(+), 1 deletion(-)
 create mode 100644 tests/twinkle_agentic/test_iteration_observers.py

diff --git a/src/twinkle_agentic/async_rl/cluster_adapter.py b/src/twinkle_agentic/async_rl/cluster_adapter.py
index f676ca8b0..840cdc86b 100644
--- a/src/twinkle_agentic/async_rl/cluster_adapter.py
+++ b/src/twinkle_agentic/async_rl/cluster_adapter.py
@@ -967,6 +967,18 @@ def shutdown(self) -> None:
 
     # --- observation / decision ------------------------------------------------------------------------
 
+    def performance_snapshot(self) -> dict[str, float]:
+        """Commit-boundary resource counters; allocated GPU budget stays fixed."""
+        return {
+            'resource/shared_enabled': int(self._shared_enabled),
+            'resource/in_generation': int(self.in_generation),
+            'resource/flex_sampler_replicas': len(self._flex_workers),
+            'resource/free_flex_groups': len(self._free_flex_groups),
+            'resource/change_pending': int(self.change_pending),
+            'resource/last_change_step': self._last_change_step,
+            'resource/handover_seconds_total': self._counters.get('handover'),
+        }
+
     def observe(self, *, step: int, driver_stats: DriverStats) -> ResourceWindow:
         """Fold one batch boundary's stat deltas into the decision window (the commit boundary).
 
diff --git a/src/twinkle_agentic/async_rl/streaming_driver.py b/src/twinkle_agentic/async_rl/streaming_driver.py
index b35ccbaba..17e677ea9 100644
--- a/src/twinkle_agentic/async_rl/streaming_driver.py
+++ b/src/twinkle_agentic/async_rl/streaming_driver.py
@@ -7,7 +7,7 @@
 import threading
 import time
 from collections.abc import Callable, Iterable, Iterator, Sequence
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
 
 from .types import PartitionAdmission, RolloutPolicy
@@ -186,7 +186,13 @@ def __init__(self,
                  adapter_name: Optional[str] = None,
                  max_steps: Optional[int] = None,
                  resource_adapter: Optional[ClusterAdapter] = None,
+                 on_iteration_start: Optional[Callable[[], None]] = None,
+                 on_iteration_end: Optional[Callable[[bool], None]] = None,
                  sleep: Optional[Callable[[float], None]] = None) -> None:
+        # Observers run on the driver thread. End follows ACTUAL publication,
+        # including the terminal case where the budget deliberately skips it.
+        self.on_iteration_start = on_iteration_start
+        self.on_iteration_end = on_iteration_end
         # Fail loudly on a mis-wired seam: a non-callable injected here would otherwise surface deep
         # inside the run loop (or silently skip a lifecycle step), far from the construction that caused it.
         # The group-barrier algorithm hooks: PRM scoring is the remote ``score`` shape's own channel,
@@ -499,6 +505,8 @@ def run(self) -> None:
         # and the sampler generates from it, never an unpinned base model (None for in-place; a no-op
         # without version control, where the version is pinned at 0 and nothing publishes).
         self._register_initial_policy()
+        if self.on_iteration_start is not None:
+            self.on_iteration_start()
         # A second run() on the same driver must not inherit the previous run's fold snapshots.
         self.stats.last_pull_idle_ratio = 0.0
         self._train_at_pull = 0.0
@@ -534,9 +542,12 @@ def run(self) -> None:
         resolved_groups: set = set()
         regen_queue: List[Tuple[Any, str]] = []
         regen_counts: Dict[str, int] = {}
+        iteration_pending = False
         try:
             while True:
                 if exhausted or self._aborted or self._default_reached_max():
+                    if iteration_pending and self.on_iteration_end is not None:
+                        self.on_iteration_end(True)
                     break
 
                 # (1) Publish phase: run an EXTERNALLY-requested publication (see request_publish). The driver
@@ -569,6 +580,13 @@ def run(self) -> None:
                         # references (runs AFTER the stale scan, which released the dropped pins).
                         self._prune()
 
+                if iteration_pending:
+                    iteration_pending = False
+                    if self.on_iteration_end is not None:
+                        self.on_iteration_end(False)
+                    if self.on_iteration_start is not None:
+                        self.on_iteration_start()
+
                 # (2) Admission: open a fresh window while the staleness gate is open, then admit
                 #     per-sample until the stream ends or the backpressure cap is reached.
                 if not stream_done or regen_queue:
@@ -716,6 +734,7 @@ def run(self) -> None:
                         self._reap_scores(records, blocking=True)
                     _t = time.perf_counter()
                     self._train_pull(records, ready, group_counts)
+                    iteration_pending = not self._iterating or self._iteration_committed
                     train_now = time.perf_counter()
                     # Fold the bubble ratio at the pull boundary, where the pairing is exact: the
                     # train delta is the PREVIOUS pull's training (already accumulated into the
@@ -1077,6 +1096,7 @@ def _bridge_consume(self, records: List[BufferRecord]) -> int:
         """
         batch = [record.sample for record in records]
         self._committed_steps = 0
+        self._iteration_committed = False
         self._ack_event.clear()
         self._batch_queue.put(batch)
         # A single wait() would deadlock against abort(): abort's lone set() can land before this
@@ -1090,6 +1110,14 @@ def _bridge_consume(self, records: List[BufferRecord]) -> int:
 
     # --- iterator regime (``for batch in driver:``) ------------------------------------------------------
 
+    def performance_snapshot(self) -> dict:
+        """Export plain observations without exposing driver internals to recorders."""
+        adapter = self._resource_adapter
+        return {
+            'stream_stats': asdict(self.stats),
+            'resource_metrics': adapter.performance_snapshot() if adapter is not None else {},
+        }
+
     def __iter__(self) -> 'StreamingDriver':
         """Drive the blocking :meth:`run` on a background thread and yield assembled, scored batches.
 
@@ -1184,6 +1212,7 @@ def commit(self, optimizer_steps: Optional[int] = 1) -> None:
         """
         # None means the caller omitted the count. An explicit 0 does not consume budget.
         self._committed_steps = 1 if optimizer_steps is None else int(optimizer_steps)
+        self._iteration_committed = True
         self._budget_steps += self._committed_steps
         self._ack_event.set()
 
diff --git a/tests/twinkle_agentic/test_iteration_observers.py b/tests/twinkle_agentic/test_iteration_observers.py
new file mode 100644
index 000000000..003728d03
--- /dev/null
+++ b/tests/twinkle_agentic/test_iteration_observers.py
@@ -0,0 +1,71 @@
+"""Observe actual post-commit publication, including terminal and error paths."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from twinkle_agentic.async_rl.streaming_driver import StreamingDriver
+
+
+def make_driver(**kwargs):
+    return StreamingDriver(
+        prompt_stream=list(range(12)),
+        max_staleness=0,
+        num_generations=1,
+        groups_per_partition=1,
+        buffer_depth=4,
+        need_rows=4,
+        group_pull=True,
+        version_control=True,
+        max_steps=2,
+        submit=lambda ref, index, policy: (ref.ordinal, index),
+        poll=lambda handles: list(handles),
+        collect=lambda handle: SimpleNamespace(),
+        consume=None,
+        **kwargs,
+    )
+
+
+def test_observer_runs_after_real_publication_and_before_next_batch():
+    events = []
+    driver = make_driver(on_iteration_start=lambda: events.append("start"))
+    driver.on_iteration_end = lambda terminal: events.append(("end", terminal, driver.stats.publishes))
+    try:
+        for samples in driver:
+            events.append("train")
+            driver.request_publish(lambda window: events.append("publish"))
+            driver.commit(1)
+    finally:
+        driver.abort()
+    assert events == ["start", "train", "publish", ("end", False, 1), "start", "train", ("end", True, 1)]
+
+
+def test_failed_uncommitted_batch_is_not_a_completed_iteration():
+    completions = []
+    driver = make_driver(on_iteration_end=lambda terminal: completions.append(terminal))
+    next(iter(driver))
+    driver.abort()  # recipe threw before commit
+    assert completions == []
+
+
+def test_observer_failure_propagates_to_recipe():
+    def failed(terminal):
+        raise ValueError("cannot persist iteration")
+
+    driver = make_driver(on_iteration_end=failed)
+    try:
+        with pytest.raises(ValueError, match="cannot persist"):
+            for samples in driver:
+                driver.commit(1)
+    finally:
+        driver.abort()
+
+
+def test_performance_snapshot_is_plain_detached_data():
+    driver = make_driver()
+    driver.stats.publishes = 2
+    snapshot = driver.performance_snapshot()
+    assert snapshot["stream_stats"]["publishes"] == 2
+    assert snapshot["resource_metrics"] == {}
+    snapshot["stream_stats"]["publishes"] = 99
+    assert driver.stats.publishes == 2

From 95c04776c09eaa626171b3c6001e078d3b51e94c Mon Sep 17 00:00:00 2001
From: Qiyuan Chen <chenqiyuan1012@gmail.com>
Date: Wed, 4 Nov 2026 15:59:23 +0800
Subject: [PATCH 3/4] perf(vllm): initialize independent rollout engines
 concurrently

---
 .../sampler/vllm_sampler/vllm_engine.py       | 10 ++++------
 .../test_posix_file_lock.py}                  | 20 +++++++++----------
 2 files changed, 14 insertions(+), 16 deletions(-)
 rename tests/{sampler/test_vllm_startup_lock.py => utils/test_posix_file_lock.py} (58%)

diff --git a/src/twinkle/sampler/vllm_sampler/vllm_engine.py b/src/twinkle/sampler/vllm_sampler/vllm_engine.py
index 304e57049..849b26f55 100644
--- a/src/twinkle/sampler/vllm_sampler/vllm_engine.py
+++ b/src/twinkle/sampler/vllm_sampler/vllm_engine.py
@@ -14,7 +14,6 @@
 from twinkle.sampler.base_engine import BaseSamplerEngine
 from twinkle.utils import Platform
 from twinkle.utils.framework import Torch
-from twinkle.utils.parallel import PosixFileLock
 from twinkle.utils.zmq_utils import configure_zmq_socket, get_timeout_s_from_env
 
 logger = get_logger()
@@ -403,11 +402,10 @@ def _create_engine(self):
         engine_args = AsyncEngineArgs(**filtered_engine_config)
         vllm_config = engine_args.create_engine_config(usage_context=UsageContext.OPENAI_API_SERVER)
 
-        with PosixFileLock('/tmp/twinkle-vllm-engine-init.lock'):
-            engine = AsyncLLM.from_vllm_config(
-                vllm_config=vllm_config,
-                usage_context=UsageContext.OPENAI_API_SERVER,
-            )
+        engine = AsyncLLM.from_vllm_config(
+            vllm_config=vllm_config,
+            usage_context=UsageContext.OPENAI_API_SERVER,
+        )
 
         logger.info(f'VLLMEngine initialized: model={self.model_id}')
         return engine
diff --git a/tests/sampler/test_vllm_startup_lock.py b/tests/utils/test_posix_file_lock.py
similarity index 58%
rename from tests/sampler/test_vllm_startup_lock.py
rename to tests/utils/test_posix_file_lock.py
index 06e5dc47c..876013014 100644
--- a/tests/sampler/test_vllm_startup_lock.py
+++ b/tests/utils/test_posix_file_lock.py
@@ -5,30 +5,30 @@
 from twinkle.utils.parallel import PosixFileLock
 
 
-def _hold_startup_lock(lock_path: str, acquired, release) -> None:
+def _hold_lock(lock_path: str, acquired, release) -> None:
     with PosixFileLock(lock_path):
         acquired.set()
         if not release.wait(timeout=30):
-            raise TimeoutError('test did not release vLLM startup lock')
+            raise TimeoutError("test did not release file lock")
 
 
-def _acquire_startup_lock(lock_path: str, started, acquired) -> None:
+def _acquire_lock(lock_path: str, started, acquired) -> None:
     started.set()
     with PosixFileLock(lock_path):
         acquired.set()
 
 
-@pytest.mark.skipif(os.name == 'nt', reason='vLLM startup lock requires fcntl')
-def test_vllm_engine_startup_is_serialized(tmp_path):
-    """Concurrent local sampler actors must not race vLLM's free-port probe."""
-    context = multiprocessing.get_context('spawn')
-    lock_path = str(tmp_path / 'vllm-engine-init.lock')
+@pytest.mark.skipif(os.name == "nt", reason="POSIX file lock requires fcntl")
+def test_posix_file_lock_serializes_processes(tmp_path):
+    """A held advisory lock excludes an independent process until released."""
+    context = multiprocessing.get_context("spawn")
+    lock_path = str(tmp_path / "process.lock")
     first_acquired = context.Event()
     release_first = context.Event()
     second_started = context.Event()
     second_acquired = context.Event()
-    first = context.Process(target=_hold_startup_lock, args=(lock_path, first_acquired, release_first))
-    second = context.Process(target=_acquire_startup_lock, args=(lock_path, second_started, second_acquired))
+    first = context.Process(target=_hold_lock, args=(lock_path, first_acquired, release_first))
+    second = context.Process(target=_acquire_lock, args=(lock_path, second_started, second_acquired))
 
     try:
         first.start()

From 160c12be82162b6a72ba16dbf007cab453cced6e Mon Sep 17 00:00:00 2001
From: Qiyuan Chen <chenqiyuan1012@gmail.com>
Date: Fri, 6 Nov 2026 12:03:39 +0800
Subject: [PATCH 4/4] refactor(metrics): reuse native reporting for performance
 observations

Keep raw timing and memory collection in perf, move throughput calculations into the Metric protocol, and reuse MetricsReporter for append-only records and terminal totals. Remove the RL recorder, duplicate JSONL writer and unused remote wait API.

Co-authored-by: Codex
---
 src/twinkle/infra/_ray/ray_helper.py          |  15 -
 src/twinkle/metric/__init__.py                |   2 +-
 src/twinkle/metric/reporting.py               |  23 +-
 src/twinkle/metric/train_metric.py            | 116 +++++++
 src/twinkle/model/base.py                     |   8 +-
 src/twinkle/perf/__init__.py                  |  16 +-
 src/twinkle/perf/recorder.py                  |   9 +-
 src/twinkle/perf/rl.py                        | 317 ------------------
 src/twinkle/perf/rl_metrics.py                | 151 ---------
 src/twinkle/perf/stats.py                     |  30 +-
 tests/metric/test_reporting.py                |  69 ++++
 .../test_throughput.py}                       |  45 ++-
 tests/perf/test_observations.py               |  36 ++
 tests/perf/test_rl_recorder.py                | 213 ------------
 14 files changed, 313 insertions(+), 737 deletions(-)
 delete mode 100644 src/twinkle/perf/rl.py
 delete mode 100644 src/twinkle/perf/rl_metrics.py
 create mode 100644 tests/metric/test_reporting.py
 rename tests/{perf/test_rl_metrics.py => metric/test_throughput.py} (58%)
 create mode 100644 tests/perf/test_observations.py
 delete mode 100644 tests/perf/test_rl_recorder.py

diff --git a/src/twinkle/infra/_ray/ray_helper.py b/src/twinkle/infra/_ray/ray_helper.py
index a73481088..ed3e45c54 100644
--- a/src/twinkle/infra/_ray/ray_helper.py
+++ b/src/twinkle/infra/_ray/ray_helper.py
@@ -68,21 +68,6 @@ def done(self) -> bool:
         """Whether every referenced task has finished (Future protocol alias of :meth:`ready`)."""
         return self.ready()
 
-    def wait(self):
-        """Wait for worker completion without pulling tensor payloads to the driver.
-
-        Downstream consumers still resolve the original references and propagate
-        worker exceptions. This is a measurement barrier, not result collection.
-        """
-        import ray
-
-        refs = [future for future in self._futures if isinstance(future, ray.ObjectRef)]
-        if refs:
-            _, pending = ray.wait(refs, num_returns=len(refs), timeout=self._timeout, fetch_local=False)
-            if pending:
-                raise TimeoutError(f'{self._method} did not complete within {self._timeout} seconds')
-        return self
-
     def ready(self) -> bool:
         """Whether every referenced task has finished, WITHOUT fetching any payload.
 
diff --git a/src/twinkle/metric/__init__.py b/src/twinkle/metric/__init__.py
index 3ef151464..a493c356d 100644
--- a/src/twinkle/metric/__init__.py
+++ b/src/twinkle/metric/__init__.py
@@ -10,5 +10,5 @@
 from .loss import LossMetric
 from .ppo import PPOValueMetric
 from .reporting import MetricsReporter, create_metrics_reporter
-from .train_metric import TrainMetric
+from .train_metric import ThroughputMetric, TrainMetric, memory_metrics
 from .types import MetricRecord
diff --git a/src/twinkle/metric/reporting.py b/src/twinkle/metric/reporting.py
index de7f9265e..cda64d017 100644
--- a/src/twinkle/metric/reporting.py
+++ b/src/twinkle/metric/reporting.py
@@ -2,8 +2,6 @@
 """Asynchronous JSONL and SwanLab metric reporting."""
 
 from __future__ import annotations
-
-import json
 import logging
 import math
 import os
@@ -16,6 +14,8 @@
 from pathlib import Path
 from typing import Any
 
+import json
+
 from .types import MetricRecord
 
 logger = logging.getLogger(__name__)
@@ -119,6 +119,16 @@ def add(self, record: MetricRecord) -> None:
         if record.stage == 'run':
             self.run_status = record.status
             self.result = {**record.values, **record.attributes}
+            # A producer may report terminal totals instead of individual sample/group events.
+            if record.context_key:
+                for name, counter in (
+                    ('rollout_groups', self.context_rollout_groups),
+                    ('rollout_samples', self.context_rollout_samples),
+                    ('trained_samples', self.context_trained_samples),
+                ):
+                    total = _finite_number(record.values.get(name))
+                    if total is not None:
+                        counter[record.context_key] = int(total)
 
     def as_dict(self, backend_health: Mapping[str, Any]) -> dict[str, Any]:
         wall_time = _finite_number(self.result.get('wall_time_s'))
@@ -325,10 +335,10 @@ def _close_sink(self) -> None:
 
 class _JSONLBackend(_QueuedBackend):
 
-    def __init__(self, path: str | Path, **kwargs: Any):
+    def __init__(self, path: str | Path, *, append: bool = False, **kwargs: Any):
         self.path = Path(path)
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        self._stream = self.path.open('w', encoding='utf-8')
+        self._stream = self.path.open('a' if append else 'w', encoding='utf-8')
         super().__init__('jsonl', **kwargs)
 
     def _write_batch(self, batch: Sequence[dict[str, Any]]) -> None:
@@ -465,7 +475,7 @@ def _normalize(self, record: MetricRecord, sequence: int) -> MetricRecord:
         for name, value in record.values.items():
             number = _finite_number(value)
             if number is None:
-                non_numeric[name] = value
+                non_numeric[name] = None if isinstance(value, float) and not math.isfinite(value) else value
             else:
                 values[name] = number
         attributes = dict(record.attributes)
@@ -523,7 +533,8 @@ def create_metrics_reporter(config: Mapping[str, Any] | None, *, run_id: str) ->
     }
     if bool(jsonl_config.get('enabled', True)):
         try:
-            backends.append(_JSONLBackend(jsonl_config['path'], **backend_defaults))
+            backends.append(_JSONLBackend(
+                jsonl_config['path'], append=bool(jsonl_config.get('append', False)), **backend_defaults))
         except Exception as exc:
             backend_errors['jsonl'] = exc
             logger.warning('JSONL metrics backend could not start: %s', exc)
diff --git a/src/twinkle/metric/train_metric.py b/src/twinkle/metric/train_metric.py
index 8d785c38b..8610973cb 100644
--- a/src/twinkle/metric/train_metric.py
+++ b/src/twinkle/metric/train_metric.py
@@ -60,3 +60,119 @@ def calculate(self):
             results['speed'] = f'{speed:.2f} iters/s'
         self.reset()
         return results
+
+def memory_metrics(rows):
+    """Convert per-worker byte counters and reduce peaks by max, host usage by mean."""
+    rows = [rows] if isinstance(rows, dict) else rows
+    fields = {
+        "max_memory_allocated_bytes": "max_memory_allocated_gb",
+        "max_memory_reserved_bytes": "max_memory_reserved_gb",
+        "cpu_memory_used_bytes": "cpu_memory_used_gb",
+    }
+    result = {}
+    for raw, name in fields.items():
+        values = [row[raw] / 1024**3 for row in rows if raw in row]
+        if values:
+            result[f"perf/{name}"] = max(values) if raw.startswith("max_") else sum(values) / len(values)
+    return result
+
+
+class ThroughputMetric(Metric):
+    """Accumulate explicit token counts and phase seconds; calculate ratios of sums.
+
+    The producer defines window boundaries and supplies the physical device budget.
+    No timing, loop state, task waiting or file output occurs inside this metric.
+    """
+
+    def __init__(self, device_mesh=None, process_group=None, **kwargs):
+        super().__init__(device_mesh, process_group, **kwargs)
+        self.reset()
+
+    def reset(self):
+        from collections import Counter
+
+        self.prompts = Counter()
+        self.responses = Counter()
+        self.timings = {}
+        self.num_devices = None
+
+    def accumulate(
+        self, inputs=None, outputs=None, *, prompt_lengths, response_lengths, timings, num_devices, **kwargs
+    ):
+        import math
+
+        prompts, responses = list(prompt_lengths), list(response_lengths)
+        if not prompts or len(prompts) != len(responses):
+            raise ValueError("expected a nonempty batch of paired prompt/response lengths")
+        if any(not isinstance(n, int) or isinstance(n, bool) or n < 0 for n in prompts + responses):
+            raise ValueError("token lengths must be nonnegative integers")
+        if not isinstance(num_devices, int) or isinstance(num_devices, bool) or num_devices < 1:
+            raise ValueError("num_devices must count allocated physical devices")
+        if self.num_devices is not None and self.num_devices != num_devices:
+            raise ValueError("reset the metric when the physical allocation changes")
+        if timings.get("step", 0) <= 0 or any(not math.isfinite(t) or t < 0 for t in timings.values()):
+            raise ValueError("timings must be finite and nonnegative, with positive step time")
+        self.num_devices = num_devices
+        self.prompts.update(prompts)
+        self.responses.update(responses)
+        for name, seconds in timings.items():
+            self.timings[name] = self.timings.get(name, 0.0) + seconds
+
+    def calculate(self, *, prompt_width=None, response_width=None):
+        if not self.prompts:
+            return {}
+        prompts, responses = self.prompts, self.responses
+        count = sum(prompts.values())
+        p = sum(length * frequency for length, frequency in prompts.items())
+        r = sum(length * frequency for length, frequency in responses.items())
+        total, elapsed = p + r, self.timings["step"]
+        result = {
+            "perf/total_num_tokens": total,
+            "perf/num_prompt_tokens": p,
+            "perf/num_response_tokens": r,
+            "perf/num_samples": count,
+            "perf/num_gpus": self.num_devices,
+            "perf/time_per_step": elapsed,
+            "perf/throughput": total / elapsed / self.num_devices,
+            "perf/throughput_total": total / elapsed,
+            "perf/response_tokens_per_second": r / elapsed,
+            "perf/samples_per_second": count / elapsed,
+            **{f"timing_s/{name}": seconds for name, seconds in self.timings.items()},
+        }
+        denominators = {
+            "gen": r,
+            **dict.fromkeys(("ref", "values", "adv", "adv_pipeline", "update_critic", "update_actor"), total),
+        }
+        for name, tokens in denominators.items():
+            if name in self.timings:
+                result[f"timing_per_token_ms/{name}"] = self.timings[name] * 1000 / tokens if tokens else 0.0
+        for name, lengths, width in (
+            ("prompt_length", prompts, prompt_width),
+            ("response_length", responses, response_width),
+        ):
+            width = max(lengths) if width is None else width
+            if width < max(lengths):
+                raise ValueError("padded width cannot be smaller than an actual sequence")
+            result.update(
+                {
+                    f"{name}/mean": sum(n * frequency for n, frequency in lengths.items()) / count,
+                    f"{name}/min": min(lengths),
+                    f"{name}/max": max(lengths),
+                    f"{name}/clip_ratio": lengths.get(width, 0) / count,
+                }
+            )
+        active = {n: frequency for n, frequency in responses.items() if n > 0}
+        active_count = sum(active.values())
+        width = max(responses) if response_width is None else response_width
+        result.update(
+            {
+                "response/aborted_ratio": 1 - active_count / count,
+                "response_length_non_aborted/mean": r / active_count if active_count else float("nan"),
+                "response_length_non_aborted/min": min(active) if active else float("nan"),
+                "response_length_non_aborted/max": max(active) if active else float("nan"),
+                "response_length_non_aborted/clip_ratio": active.get(width, 0) / active_count
+                if active_count
+                else float("nan"),
+            }
+        )
+        return result
diff --git a/src/twinkle/model/base.py b/src/twinkle/model/base.py
index c5e40a2e6..37bc898c6 100644
--- a/src/twinkle/model/base.py
+++ b/src/twinkle/model/base.py
@@ -298,11 +298,11 @@ def add_metric(self, metric_cls: Union[Metric, str], is_training: Optional[bool]
         ...
 
     @remote_function(dispatch='all', collect='none', lazy_collect=False)
-    def performance_memory(self) -> dict[str, float]:
-        """Per-rank allocator lifetime peaks; call after the actor update."""
-        from twinkle.perf.rl_metrics import accelerator_memory_metrics
+    def memory_stats(self, *, synchronize: bool = False) -> dict[str, int]:
+        """Raw per-rank allocator peaks and host RAM in bytes; no cross-rank reduction."""
+        from twinkle.perf.stats import memory_stats
 
-        return accelerator_memory_metrics()
+        return memory_stats(synchronize=synchronize)
 
     @abstractmethod
     def calculate_metric(self, is_training: bool, **kwargs) -> Dict[str, Any]:
diff --git a/src/twinkle/perf/__init__.py b/src/twinkle/perf/__init__.py
index ce77e3e8f..4b22a2bc7 100644
--- a/src/twinkle/perf/__init__.py
+++ b/src/twinkle/perf/__init__.py
@@ -1,7 +1,7 @@
 # Copyright (c) ModelScope Contributors. All rights reserved.
 """twinkle.perf -- a general-purpose performance observability component.
 
-Five independent, low-intrusion parts:
+Four independent, low-intrusion parts:
 
 * :class:`TimeRecorder` / :func:`timed` / :func:`span` -- a synchronous- and
   async-agnostic wall-clock span recorder that nests into an overall timing tree
@@ -21,10 +21,6 @@
   Chrome Trace rendering and a text DAG / lane report. Consumed by whoever holds
   the window's events; the recorder itself only appends.
 
-* :class:`RLPerformanceRecorder` -- explicit rollout-batch accounting, phase timing,
-  durable JSONL records and final summaries. Accepts lengths, integer update counts
-  and plain observation dictionaries, independently of any training recipe.
-
 The recorder and profiler are disabled/no-op by default, so importing perf never
 changes existing behaviour; the counter registry is always live by design (a few
 float additions per step). None of the parts imports twinkle/swift business
@@ -41,18 +37,11 @@
 from .counters import CounterRegistry, default_counters
 from .module_profiler import ModuleProfiler, activation_stats, build_module_profiler, param_stats
 from .recorder import TimeRecorder, default_recorder, disable_perf, enable_perf, span, timed
-from .rl import RLPerformanceRecorder, reduce_memory, sample_lengths
-from .rl_metrics import JsonlMetricWriter, compute_rl_performance_metrics
-from .stats import describe_shape, primary_tensor, tensor_moments, tensor_numel
+from .stats import describe_shape, memory_stats, primary_tensor, tensor_moments, tensor_numel
 from .timeline import chrome_trace, dag_report
 
 __all__ = [
     'Perf',
-    'RLPerformanceRecorder',
-    'JsonlMetricWriter',
-    'compute_rl_performance_metrics',
-    'reduce_memory',
-    'sample_lengths',
     'TimeRecorder',
     'default_recorder',
     'timed',
@@ -71,4 +60,5 @@
     'tensor_moments',
     'tensor_numel',
     'describe_shape',
+    'memory_stats',
 ]
diff --git a/src/twinkle/perf/recorder.py b/src/twinkle/perf/recorder.py
index 05fe1cf11..53243885d 100644
--- a/src/twinkle/perf/recorder.py
+++ b/src/twinkle/perf/recorder.py
@@ -57,13 +57,16 @@ def __init__(self, name: str, parent: Optional['_Span'], start_ns: int, tid: int
 class TimeRecorder:
     """Aggregate per-name span timings into a summary and a metric-ready dict.
 
+    ``clock_ns`` optionally supplies a monotonic nanosecond clock for replay/tests.
+
     Enable/disable is checked at CALL time (not decoration time), so decorating a
     function once and toggling the recorder later behaves correctly, and a disabled
     recorder takes a no-op fast path with no timing or allocation.
     """
 
-    def __init__(self, *, enabled: bool = True) -> None:
+    def __init__(self, *, enabled: bool = True, clock_ns: Callable[[], int] | None = None) -> None:
         self._enabled = enabled
+        self._clock_ns = clock_ns or time.perf_counter_ns
         self._lock = threading.Lock()
         # name -> [count, total_ns, self_ns]
         self._stats: Dict[str, list] = {}
@@ -133,7 +136,7 @@ def _span(self, name: str) -> Generator[None, None, None]:
         partial duration); the exception is NOT suppressed.
         """
         parent = self._current.get()
-        current = _Span(name, parent, time.perf_counter_ns(), threading.get_ident())
+        current = _Span(name, parent, self._clock_ns(), threading.get_ident())
         token = self._current.set(current)
         try:
             yield
@@ -141,7 +144,7 @@ def _span(self, name: str) -> Generator[None, None, None]:
             # Reset BEFORE recording so a raising body cannot leave a stale span on
             # the chain for subsequent code in this context.
             self._current.reset(token)
-            self._record(current, time.perf_counter_ns())
+            self._record(current, self._clock_ns())
 
     def _record(self, current: _Span, end_ns: int) -> None:
         total_ns = end_ns - current.start_ns
diff --git a/src/twinkle/perf/rl.py b/src/twinkle/perf/rl.py
deleted file mode 100644
index a2f22e058..000000000
--- a/src/twinkle/perf/rl.py
+++ /dev/null
@@ -1,317 +0,0 @@
-# Copyright (c) ModelScope Contributors. All rights reserved.
-"""Reusable RL iteration accounting, phase timing and persistent metrics.
-
-The iteration axis is independent of optimizer updates. Never average per-step
-throughputs to compare a run: sum tokens / sum seconds / physical allocation.
-"""
-
-import logging
-from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
-from pathlib import Path
-from time import perf_counter
-
-from .rl_metrics import JsonlMetricWriter, compute_rl_performance_metrics
-
-
-def sample_lengths(sample):
-    """Unpadded initial prompt and full response, including tool observations.
-
-    Twinkle stores shifted labels/masks. Policy tokens identify the first response
-    position; subsequent masked tool tokens belong to the full response span too.
-    """
-    encoded = sample.encoded
-    ids = list(encoded["input_ids"])
-    labels = list(encoded["labels"])
-    if len(labels) != len(ids):
-        raise ValueError("performance metrics require one shifted label per input token")
-    labels = labels[-1:] + labels[:-1]
-    mask = encoded.get("completion_mask")
-    if mask is not None:
-        mask = list(mask)
-        if len(mask) != len(ids):
-            raise ValueError("completion mask and input_ids must have equal lengths")
-        mask = mask[-1:] + mask[:-1]
-    attention = encoded.get("attention_mask")
-    attention = list(attention) if attention is not None else [1] * len(ids)
-    if len(attention) != len(ids):
-        raise ValueError("performance metrics require a one-dimensional attention mask")
-    positions = [i for i, label in enumerate(labels) if label != -100 and (mask is None or mask[i]) and attention[i]]
-    first = positions[0] if positions else len(ids)
-    prompt = sum(bool(x) for x in attention[:first])
-    response = sum(bool(x) for x in attention[first:])
-    return prompt, response
-
-
-def reduce_memory(rows):
-    """Reduce allocator peaks by rank maximum and other memory observations by mean."""
-    rows = [rows] if isinstance(rows, dict) else rows
-    keys = set().union(*(row.keys() for row in rows))
-    result = {}
-    for key in keys:
-        values = [row[key] for row in rows if key in row]
-        result[key] = max(values) if "max" in key else sum(values) / len(values)
-    return result
-
-
-class RLPerformanceRecorder:
-    def __init__(
-        self,
-        output_dir,
-        device_groups,
-        *,
-        mode="synchronous",
-        run_id=None,
-        parameter_sync_step=1,
-        clock=perf_counter,
-        logger=None,
-        metadata=None,
-        max_forward_workers=3,
-        track_batch_wait=False,
-        memory_prefix=None,
-    ):
-        self.writer = JsonlMetricWriter(Path(output_dir) / "performance.jsonl", run_id=run_id)
-        self.summary_writer = JsonlMetricWriter(
-            Path(output_dir) / "performance_summary.jsonl", run_id=self.writer.run_id
-        )
-        self.wall_clock = clock
-        self.excluded_seconds = 0.0
-        self.clock = lambda: self.wall_clock() - self.excluded_seconds
-        self.groups = device_groups
-        self.n_gpus = len({rank for _, ranks in device_groups for rank in ranks})
-        if not self.n_gpus:
-            raise ValueError("missing physical device allocation for RL performance metrics")
-        self.mode = mode
-        self.logger = logger or logging.getLogger(__name__)
-        self.metadata = dict(metadata or {})
-        self.max_forward_workers = max_forward_workers
-        self.track_batch_wait = track_batch_wait
-        self.memory_prefix = memory_prefix
-        self.iteration = 0
-        self.active = False
-        self.prompt_lengths = None
-        self.response_lengths = None
-        self.truncated = None
-        self.assistant_turns = None
-        self.memory = None
-        self._publish_time = 0.0
-        self._publishes = 0
-        self.parameter_sync_step = parameter_sync_step
-        self.total_tokens = 0
-        self.total_seconds = 0.0
-        self._forward_executor = None
-        self._forward_observers = []
-
-    def begin(self, optimizer_step):
-        self.timings = {}
-        self.start_step = optimizer_step
-        self.prompt_lengths = None
-        self.response_lengths = None
-        self.truncated = None
-        self.assistant_turns = None
-        self.memory = None
-        self.sample_visits = 0
-        self.trained_indices = set()
-        self.started = self.clock()
-        self.wall_started = self.wall_clock()
-        self.excluded_at_start = self.excluded_seconds
-        self.active = True
-
-    @contextmanager
-    def exclude_logging(self):
-        """Exclude reporting I/O from logical time while retaining full wall time."""
-        started = self.wall_clock()
-        try:
-            yield
-        finally:
-            self.excluded_seconds += self.wall_clock() - started
-
-    @contextmanager
-    def stage(self, name):
-        started = self.clock()
-        try:
-            yield
-        finally:
-            self.timings[name] = self.timings.get(name, 0.0) + self.clock() - started
-
-    def call(self, name, fn, *args, collect_result=False, **kwargs):
-        with self.stage(name):
-            result = fn(*args, **kwargs)
-            # Complete the actual remote work; do not time only RPC submission.
-            if getattr(result, "_is_lazy_collect", False):
-                if collect_result:
-                    # These small side-effect results have no downstream tensor
-                    # consumer to surface failures. Waiting alone is insufficient.
-                    result = result.result()
-                else:
-                    result.wait()
-            return result
-
-    def launch_forward(self, name, fn, *args, **kwargs):
-        """Observe independent RPCs without serializing auxiliary model lanes.
-
-        Each span is submission-to-completion wall time, including queueing. They
-        may overlap and must not be added to reconstruct iteration time.
-        """
-        started = self.clock()
-        result = fn(*args, **kwargs)
-        if getattr(result, "_is_lazy_collect", False):
-            if self._forward_executor is None:
-                self._forward_executor = ThreadPoolExecutor(
-                    max_workers=self.max_forward_workers, thread_name_prefix="rl-metric"
-                )
-
-            def observe():
-                result.wait()  # no payload transfer; consumer propagates task errors
-                return self.clock() - started
-
-            self._forward_observers.append((name, self._forward_executor.submit(observe)))
-        else:
-            self.timings[name] = self.timings.get(name, 0.0) + self.clock() - started
-        return result
-
-    def join_forwards(self):
-        """The advantage consumer already depends on every auxiliary lane."""
-        pending, self._forward_observers = self._forward_observers, []
-        for name, observer in pending:
-            self.timings[name] = self.timings.get(name, 0.0) + observer.result()
-
-    def received(self, samples):
-        """Adapt encoded rollout samples; callers may instead supply lengths directly."""
-        lengths = [sample_lengths(sample) for sample in samples]
-        turns = [getattr(sample, "rollout_infos", {}).get("turns") for sample in samples]
-        self.received_lengths(
-            [prompt for prompt, _ in lengths],
-            [response for _, response in lengths],
-            truncated=[bool(getattr(sample, "truncated", False)) for sample in samples],
-            assistant_turns=turns if turns and all(isinstance(n, (int, float)) for n in turns) else None,
-        )
-
-    def received_lengths(self, prompt_lengths, response_lengths, *, truncated=None, assistant_turns=None):
-        """Register a rollout batch without depending on any training/sample class."""
-        prompts, responses = list(prompt_lengths), list(response_lengths)
-        if not prompts or len(prompts) != len(responses):
-            raise ValueError("expected a nonempty batch of paired prompt/response lengths")
-        flags = [False] * len(prompts) if truncated is None else list(truncated)
-        turns = None if assistant_turns is None else list(assistant_turns)
-        if len(flags) != len(prompts) or (turns is not None and len(turns) != len(prompts)):
-            raise ValueError("per-sample observations must match the batch length")
-        self.prompt_lengths, self.response_lengths = prompts, responses
-        self.truncated, self.assistant_turns = flags, turns
-        if self.track_batch_wait:
-            self.timings["gen"] = self.clock() - self.started
-
-    def record_training(self, sample_indices):
-        """Count visits across replay while retaining unique original-batch indices."""
-        indices = list(sample_indices)
-        if self.prompt_lengths is None or any(not 0 <= index < len(self.prompt_lengths) for index in indices):
-            raise ValueError("training indices must refer to the registered rollout batch")
-        self.sample_visits += len(indices)
-        self.trained_indices.update(indices)
-
-    def finish(
-        self, optimizer_step, *, stream_stats=None, resource_metrics=None, memory=None, terminal=False, publications=0
-    ):
-        if not self.active or self.prompt_lengths is None:
-            return
-        ended = self.clock()
-        wall_elapsed = self.wall_clock() - self.wall_started
-        self.active = False
-        timings = dict(self.timings, step=ended - self.started)
-        extra = {}
-        if stream_stats is not None:
-            stats = dict(stream_stats)
-            if "publish_time" in stats:
-                publish_time = stats["publish_time"]
-                timings["update_weights"] = publish_time - self._publish_time
-                self._publish_time = publish_time
-            if "publishes" in stats:
-                publications = stats["publishes"] - self._publishes
-                self._publishes = stats["publishes"]
-            extra.update(
-                {
-                    f"stream/{key}": value
-                    for key, value in stats.items()
-                    if isinstance(value, (int, float))
-                    and key not in ("total_time", "final_version", "dropped_incomplete")
-                }
-            )
-        extra.update(resource_metrics or {})
-        metrics = compute_rl_performance_metrics(self.prompt_lengths, self.response_lengths, timings, self.n_gpus)
-        metrics.update(extra)
-        turns = self.assistant_turns
-        if turns is not None and all(isinstance(n, (int, float)) for n in turns):
-            # Count assistant generations, independently of user/tool message counts.
-            metrics.update(
-                {
-                    "rollout/assistant_turns/mean": sum(turns) / len(turns),
-                    "rollout/assistant_turns/min": min(turns),
-                    "rollout/assistant_turns/max": max(turns),
-                }
-            )
-        metrics.update(memory or {})
-        if self.memory_prefix:
-            metrics.update({f"{self.memory_prefix}/{key}": value for key, value in (memory or {}).items()})
-        metrics.update(
-            {
-                "training/optimizer_step_start": self.start_step,
-                "training/weight_publications_this_iteration": publications,
-                "training/optimizer_step": optimizer_step,
-                "training/optimizer_steps_this_iteration": optimizer_step - self.start_step,
-                "training/sample_visits": self.sample_visits,
-                "training/unique_samples_trained": len(self.trained_indices),
-                "training/samples_not_trained": len(self.prompt_lengths) - len(self.trained_indices),
-                "response/truncated_ratio": sum(bool(flag) for flag in self.truncated) / len(self.truncated),
-                "perf/gpu_seconds": timings["step"] * self.n_gpus,
-                "observability/iteration_wall_seconds": wall_elapsed,
-                "observability/excluded_logging_seconds": self.excluded_seconds - self.excluded_at_start,
-            }
-        )
-        self.iteration += 1
-        self.writer.write(
-            self.iteration,
-            metrics,
-            schema_version=1,
-            step_unit="rl_iteration",
-            iteration_scope="one_consumed_rollout_batch",
-            parameter_sync_step=self.parameter_sync_step,
-            parameter_sync_step_unit="optimizer_update",
-            mode=self.mode,
-            device_groups=self.groups,
-            token_scope="initial_prompt_plus_full_response_including_tool_observations",
-            terminal=terminal,
-            **self.metadata,
-            unavailable_metrics=["perf/mfu/actor"],
-            unsupported_mfu_reason="no validated model FLOPs and device peak FLOPs estimator",
-        )
-        self.total_tokens += metrics["perf/total_num_tokens"]
-        self.total_seconds += timings["step"]
-        self.logger.info(
-            "[perf] iteration=%s step=%s seconds=%.3f tokens=%s tok/s/GPU=%.3f",
-            self.iteration,
-            optimizer_step,
-            timings["step"],
-            metrics["perf/total_num_tokens"],
-            metrics["perf/throughput"],
-        )
-
-    def write_summary(self, optimizer_step, *, stream_stats=None, status="completed"):
-        """Call after driver teardown: terminal drops/final_version now exist."""
-        data = {
-            "completed_iterations": self.iteration,
-            "optimizer_step": optimizer_step,
-            "total_num_tokens": self.total_tokens,
-            "sum_iteration_seconds": self.total_seconds,
-            "allocated_gpus": self.n_gpus,
-            "throughput": (self.total_tokens / (self.total_seconds * self.n_gpus) if self.total_seconds else None),
-        }
-        if stream_stats is not None:
-            data.update({f"stream/{k}": v for k, v in stream_stats.items() if isinstance(v, (int, float))})
-        self.summary_writer.write(self.iteration, data, step_unit="run_summary", status=status)
-
-    def close(self):
-        self.active = False
-        self.writer.close()
-        self.summary_writer.close()
-        if self._forward_executor is not None:
-            self._forward_executor.shutdown(wait=False, cancel_futures=True)
diff --git a/src/twinkle/perf/rl_metrics.py b/src/twinkle/perf/rl_metrics.py
deleted file mode 100644
index d210bd827..000000000
--- a/src/twinkle/perf/rl_metrics.py
+++ /dev/null
@@ -1,151 +0,0 @@
-# Copyright (c) ModelScope Contributors. All rights reserved.
-"""RL iteration metrics with explicit token, time and allocation scopes.
-
-This module is deliberately independent of torch/Ray. Counts are unpadded sequence
-lengths, once per trajectory in an RL batch, NOT once per optimizer update/replay.
-Timing is supplied by the orchestration layer; a formula cannot fix a mismatched
-measurement boundary. Callers supply explicit iteration and phase boundaries.
-"""
-
-from __future__ import annotations
-import math
-import threading
-import time
-import uuid
-from pathlib import Path
-
-import json
-
-
-def compute_rl_performance_metrics(
-    prompt_lengths, response_lengths, timing_raw, n_gpus, *, prompt_width=None, response_width=None
-):
-    """Throughput, lengths and per-token timing with explicit denominators.
-
-    Ragged batches use the batch maximum as padded width. ``clip_ratio`` therefore
-    counts rows equal to that width, not rows that hit the configured generation
-    cap. Explicit widths also support fixed-width batches.
-    """
-    prompts, responses = list(prompt_lengths), list(response_lengths)
-    if len(prompts) != len(responses) or not prompts:
-        raise ValueError("expected a nonempty batch of paired prompt/response lengths")
-    if any(not isinstance(n, int) or isinstance(n, bool) or n < 0 for n in prompts + responses):
-        raise ValueError("token lengths must be nonnegative integers")
-    elapsed = timing_raw["step"]
-    if not isinstance(n_gpus, int) or isinstance(n_gpus, bool) or n_gpus < 1:
-        raise ValueError("n_gpus must count allocated physical devices, including rollout")
-    if elapsed <= 0 or any(not math.isfinite(t) or t < 0 for t in timing_raw.values()):
-        raise ValueError("timings must be finite and nonnegative, with positive step time")
-    p, r = sum(prompts), sum(responses)
-    total = p + r
-    out = {
-        "perf/total_num_tokens": total,
-        "perf/time_per_step": elapsed,
-        "perf/throughput": total / (elapsed * n_gpus),
-        "perf/num_gpus": n_gpus,
-        "perf/num_samples": len(prompts),
-        "perf/num_prompt_tokens": p,
-        "perf/num_response_tokens": r,
-        "perf/throughput_total": total / elapsed,
-        "perf/response_tokens_per_second": r / elapsed,
-        "perf/samples_per_second": len(prompts) / elapsed,
-        **{f"timing_s/{name}": t for name, t in timing_raw.items()},
-    }
-    denominators = {"gen": r, **dict.fromkeys(("ref", "values", "adv", "update_critic", "update_actor"), total)}
-    for name, count in denominators.items():
-        if name in timing_raw:
-            out[f"timing_per_token_ms/{name}"] = timing_raw[name] * 1000 / count if count else 0.0
-    for name, lengths, width in (
-        ("prompt_length", prompts, prompt_width),
-        ("response_length", responses, response_width),
-    ):
-        width = max(lengths) if width is None else width
-        if width < max(lengths):
-            raise ValueError("padded width cannot be smaller than an actual sequence")
-        out.update(
-            {
-                f"{name}/mean": sum(lengths) / len(lengths),
-                f"{name}/min": min(lengths),
-                f"{name}/max": max(lengths),
-                f"{name}/clip_ratio": sum(n == width for n in lengths) / len(lengths),
-            }
-        )
-    active = [n for n in responses if n > 0]
-    width = max(responses) if response_width is None else response_width
-    out["response/aborted_ratio"] = 1 - len(active) / len(responses)
-    out.update(
-        {
-            "response_length_non_aborted/mean": sum(active) / len(active) if active else float("nan"),
-            "response_length_non_aborted/min": min(active) if active else float("nan"),
-            "response_length_non_aborted/max": max(active) if active else float("nan"),
-            "response_length_non_aborted/clip_ratio": sum(n == width for n in active) / len(active)
-            if active
-            else float("nan"),
-        }
-    )
-    return out
-
-
-def _json_value(value):
-    # Nonfinite numbers become JSON null, never invalid NaN literals.
-    if isinstance(value, float) and not math.isfinite(value):
-        return None
-    if isinstance(value, dict):
-        return {str(k): _json_value(v) for k, v in value.items()}
-    if isinstance(value, (tuple, list)):
-        return [_json_value(v) for v in value]
-    return value
-
-
-class JsonlMetricWriter:
-    """Incremental, append-only local output with a {step, data} envelope.
-
-    A new run_id on every construction disambiguates resumed/repeated runs in the
-    same directory. Each record is flushed; this does not claim power-loss fsync.
-    """
-
-    def __init__(self, path, *, run_id=None):
-        self.path = Path(path)
-        self.run_id = run_id or uuid.uuid4().hex
-        self._stream = None
-        self._lock = threading.Lock()
-
-    def write(self, step, data, **metadata):
-        row = _json_value(dict(step=step, data=data, run_id=self.run_id, timestamp=time.time(), **metadata))
-        line = json.dumps(row, ensure_ascii=False, allow_nan=False) + "\n"
-        with self._lock:
-            if self._stream is None:
-                self.path.parent.mkdir(parents=True, exist_ok=True)
-                self._stream = self.path.open("a", encoding="utf-8")
-            self._stream.write(line)
-            self._stream.flush()
-
-    def close(self):
-        with self._lock:
-            if self._stream is not None:
-                self._stream.close()
-                self._stream = None
-
-
-def accelerator_memory_metrics():
-    """Per-worker allocator high-water marks and host-used RAM.
-
-    These are lifetime peaks (no reset), not device-wide memory usage. CPU RAM is
-    host-wide virtual_memory().used, NOT process RSS. Unsupported counters are
-    absent, not zero. This helper launches no accelerator work.
-    """
-    import psutil
-    import torch
-
-    from twinkle import Platform
-
-    device = getattr(torch, Platform.device_prefix(), None)
-    result = {"perf/cpu_memory_used_gb": psutil.virtual_memory().used / 1024**3}
-    if device is not None and getattr(device, "is_available", lambda: False)():
-        # The update must have actually finished, not just queued optimizer kernels.
-        device.synchronize()
-        for name in ("max_memory_allocated", "max_memory_reserved"):
-            fn = getattr(device, name, None)
-            if fn is not None:
-                result[f"perf/{name}_gb"] = fn() / 1024**3
-    return result
diff --git a/src/twinkle/perf/stats.py b/src/twinkle/perf/stats.py
index 2d4cbd458..a11cb91a1 100644
--- a/src/twinkle/perf/stats.py
+++ b/src/twinkle/perf/stats.py
@@ -1,7 +1,7 @@
 # Copyright (c) ModelScope Contributors. All rights reserved.
-"""Tensor statistics helpers for the module profiler.
+"""Raw tensor and process observations for performance instrumentation.
 
-These are plain functions over ``torch.Tensor`` (and the nested containers a
+Tensor helpers are plain functions over ``torch.Tensor`` (and nested containers a
 forward may return) so both the built-in :func:`activation_stats` /
 :func:`param_stats` collectors and user collectors can reuse them. Every
 reduction ends in ``.item()``, which forces a device sync -- that is
@@ -16,7 +16,7 @@
 
 import torch
 
-__all__ = ['iter_tensors', 'primary_tensor', 'tensor_moments', 'tensor_numel', 'describe_shape']
+__all__ = ['iter_tensors', 'primary_tensor', 'tensor_moments', 'tensor_numel', 'describe_shape', 'memory_stats']
 
 
 def iter_tensors(obj: Any) -> Iterator[torch.Tensor]:
@@ -84,3 +84,27 @@ def describe_shape(obj: Any) -> str:
     if len(tensors) == 1:
         return 'x'.join(str(dim) for dim in tensors[0].shape)
     return '(' + ', '.join('x'.join(str(dim) for dim in t.shape) for t in tensors) + ')'
+
+
+def memory_stats(*, synchronize: bool = False) -> dict[str, int]:
+    """Raw per-process allocator peaks and host-wide used RAM, in bytes.
+
+    Counters are lifetime peaks, not step peaks. Host usage is not process RSS.
+    Reading never resets counters. Device synchronization is an explicit opt-in
+    measurement barrier, disabled by default.
+    Unsupported counters are absent rather than invented zeroes.
+    """
+    import psutil
+
+    from twinkle import Platform
+
+    result = {"cpu_memory_used_bytes": psutil.virtual_memory().used}
+    device = getattr(torch, Platform.device_prefix(), None)
+    if device is not None and getattr(device, "is_available", lambda: False)():
+        if synchronize:
+            device.synchronize()
+        for name in ("max_memory_allocated", "max_memory_reserved"):
+            counter = getattr(device, name, None)
+            if counter is not None:
+                result[name + "_bytes"] = counter()
+    return result
diff --git a/tests/metric/test_reporting.py b/tests/metric/test_reporting.py
new file mode 100644
index 000000000..ee3b43787
--- /dev/null
+++ b/tests/metric/test_reporting.py
@@ -0,0 +1,69 @@
+"""Existing metric output supports resumed runs and raw observation attributes."""
+
+import json
+from concurrent.futures import ThreadPoolExecutor
+
+from twinkle.metric import MetricRecord, create_metrics_reporter
+
+
+def reporter(path, run_id):
+    return create_metrics_reporter({"jsonl": {
+        "path": path, "summary_path": path.with_name("summary.json"),
+        "append": True, "batch_size": 1,
+    }}, run_id=run_id)
+
+
+def test_existing_reporter_flushes_raw_observations_and_nonfinite_values(tmp_path):
+    path = tmp_path / "metrics.jsonl"
+    output = reporter(path, "run-1")
+    output.record(MetricRecord(
+        stage="train", optimizer_step=2, values={"tokens": 19, "loss": float("nan")},
+        attributes={"memory_by_rank": [{"allocated_bytes": 4096}], "timing_spans": [(1, 2, 3)]},
+    ))
+    output.flush()
+    row = json.loads(path.read_text())
+    assert row["values"] == {"tokens": 19}
+    assert row["attributes"]["non_numeric_values"]["loss"] is None
+    assert row["attributes"]["memory_by_rank"] == [{"allocated_bytes": 4096}]
+    assert row["attributes"]["timing_spans"] == [[1, 2, 3]]
+    assert row["run_id"] == "run-1" and row["optimizer_step"] == 2
+    output.close()
+
+
+def test_append_does_not_overwrite_previous_run(tmp_path):
+    path = tmp_path / "metrics.jsonl"
+    for run_id in ("first", "second"):
+        output = reporter(path, run_id)
+        output.record(MetricRecord(stage="train", optimizer_step=1, values={"loss": 0.5}))
+        output.close()
+    rows = [json.loads(line) for line in path.read_text().splitlines()]
+    assert [row["run_id"] for row in rows] == ["first", "second"]
+
+
+def test_existing_reporter_orders_concurrent_records_and_flushes_on_close(tmp_path):
+    path = tmp_path / "metrics.jsonl"
+    output = reporter(path, "run-1")
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        list(pool.map(lambda step: output.record(MetricRecord(stage="train", optimizer_step=step, values={"x": step})), range(32)))
+    output.close()
+    rows = [json.loads(line) for line in path.read_text().splitlines()]
+    assert [row["sequence"] for row in rows] == list(range(1, 33))
+    assert sorted(row["optimizer_step"] for row in rows) == list(range(32))
+    assert output.health()["dropped_records"] == 0
+
+
+def test_terminal_totals_are_used_when_producer_reports_batch_scopes(tmp_path):
+    output = reporter(tmp_path / "metrics.jsonl", "run-1")
+    output.record(MetricRecord(stage="train", context_key="run-1", optimizer_step=1, values={"loss": 0.5}))
+    output.record(MetricRecord(
+        stage="rollout", context_key="run-1", values={"sample_count": 64}, attributes={"scope": "iteration"},
+    ))
+    output.record(MetricRecord(
+        stage="run", context_key="run-1", values={"rollout_groups": 8, "rollout_samples": 64, "trained_samples": 128},
+    ))
+    output.close()
+    summary = json.loads((tmp_path / "summary.json").read_text())
+    assert summary["train_steps"] == 1
+    assert summary["rollout_groups"] == 8
+    assert summary["rollout_samples"] == 64
+    assert summary["trained_samples"] == 128
diff --git a/tests/perf/test_rl_metrics.py b/tests/metric/test_throughput.py
similarity index 58%
rename from tests/perf/test_rl_metrics.py
rename to tests/metric/test_throughput.py
index eb6a3fb2e..5d7d1b037 100644
--- a/tests/perf/test_rl_metrics.py
+++ b/tests/metric/test_throughput.py
@@ -1,17 +1,29 @@
 """Numerical contracts for rollout accounting, padding and durable records."""
 
-import json
 import math
 
 import pytest
+from twinkle.metric import ThroughputMetric, memory_metrics
 
-from twinkle.perf import JsonlMetricWriter, compute_rl_performance_metrics
+
+def compute_performance_metrics(prompts, responses, timings, devices, **widths):
+    metric = ThroughputMetric()
+    metric.accumulate(prompt_lengths=prompts, response_lengths=responses, timings=timings, num_devices=devices)
+    return metric.calculate(**widths)
 
 
 @pytest.mark.parametrize("widths,clip,non_aborted_clip", [((None, None), 1 / 3, 0.5), ((12, 8), 0.0, 0.0)])
 def test_variable_lengths_and_explicit_padded_width(widths, clip, non_aborted_clip):
-    timings = dict(step=7.25, gen=3.5, ref=0.25, old_log_prob=0.4, adv=0.01, update_actor=2, update_weights=0.5)
-    actual = compute_rl_performance_metrics(
+    timings = {
+        "step": 7.25,
+        "gen": 3.5,
+        "ref": 0.25,
+        "old_log_prob": 0.4,
+        "adv": 0.01,
+        "update_actor": 2,
+        "update_weights": 0.5,
+    }
+    actual = compute_performance_metrics(
         [3, 6, 4], [0, 2, 5], timings, 8, prompt_width=widths[0], response_width=widths[1]
     )
     expected = {
@@ -46,7 +58,7 @@ def test_variable_lengths_and_explicit_padded_width(widths, clip, non_aborted_cl
 
 
 def test_all_empty_responses_have_no_non_aborted_population():
-    actual = compute_rl_performance_metrics([2, 3], [0, 0], {"step": 7.25, "gen": 3.5, "update_actor": 2}, 8)
+    actual = compute_performance_metrics([2, 3], [0, 0], {"step": 7.25, "gen": 3.5, "update_actor": 2}, 8)
     assert actual["perf/total_num_tokens"] == 5
     assert actual["perf/throughput"] == pytest.approx(0.08620689655172414)
     assert actual["response/aborted_ratio"] == 1
@@ -59,11 +71,22 @@ def test_all_empty_responses_have_no_non_aborted_population():
 def test_invalid_allocation_or_timing_fails_instead_of_faking_throughput():
     for n_gpus, seconds in [(0, 1), (2, 0), (2, float("nan"))]:
         with pytest.raises(ValueError):
-            compute_rl_performance_metrics([3], [2], {"step": seconds}, n_gpus)
+            compute_performance_metrics([3], [2], {"step": seconds}, n_gpus)
+
+
+def test_window_throughput_uses_sum_tokens_over_sum_seconds():
+    metric = ThroughputMetric()
+    metric.accumulate(prompt_lengths=[2], response_lengths=[8], timings={"step": 1}, num_devices=2)
+    metric.accumulate(prompt_lengths=[2], response_lengths=[18], timings={"step": 4}, num_devices=2)
+    assert metric.calculate()["perf/throughput"] == 3  # 30 / 5 / 2, not mean(5, 2.5)
+    metric.reset()
+    assert metric.calculate() == {}
 
 
-def test_jsonl_flush_and_nonfinite(tmp_path):
-    writer = JsonlMetricWriter(tmp_path / "metrics.jsonl")
-    writer.write(1, {"good": 1.5, "nan": float("nan")})
-    assert json.loads(writer.path.read_text())["data"] == {"good": 1.5, "nan": None}
-    writer.close()
+def test_memory_reduction_uses_rank_max_for_peaks_and_mean_for_host_ram():
+    assert memory_metrics(
+        [
+            {"max_memory_allocated_bytes": 3 * 1024**3, "cpu_memory_used_bytes": 8 * 1024**3},
+            {"max_memory_allocated_bytes": 5 * 1024**3, "cpu_memory_used_bytes": 10 * 1024**3},
+        ]
+    ) == {"perf/max_memory_allocated_gb": 5, "perf/cpu_memory_used_gb": 9}
diff --git a/tests/perf/test_observations.py b/tests/perf/test_observations.py
new file mode 100644
index 000000000..b0ae0b99b
--- /dev/null
+++ b/tests/perf/test_observations.py
@@ -0,0 +1,36 @@
+"""Atomic observations carry units and stay independent of recipe policy."""
+
+from types import SimpleNamespace
+
+import torch
+
+from twinkle import Platform
+from twinkle.perf.stats import memory_stats
+
+
+def test_memory_returns_raw_bytes_without_implicit_sync_or_reset(monkeypatch):
+    import psutil
+
+    actions = []
+    monkeypatch.setattr(Platform, "device_prefix", staticmethod(lambda: "cuda"))
+    monkeypatch.setattr(psutil, "virtual_memory", lambda: SimpleNamespace(used=123456))
+    monkeypatch.setattr(torch, "cuda", SimpleNamespace(
+        is_available=lambda: True,
+        synchronize=lambda: actions.append("sync"),
+        max_memory_allocated=lambda: 8192,
+        max_memory_reserved=lambda: 16384,
+    ))
+    assert memory_stats() == {
+        "cpu_memory_used_bytes": 123456,
+        "max_memory_allocated_bytes": 8192,
+        "max_memory_reserved_bytes": 16384,
+    }
+    assert actions == []
+    assert memory_stats(synchronize=True)["max_memory_allocated_bytes"] == 8192
+    assert actions == ["sync"]
+
+
+def test_unavailable_device_counters_are_absent(monkeypatch):
+    monkeypatch.setattr(Platform, "device_prefix", staticmethod(lambda: "cuda"))
+    monkeypatch.setattr(torch, "cuda", SimpleNamespace(is_available=lambda: False))
+    assert set(memory_stats()) == {"cpu_memory_used_bytes"}
diff --git a/tests/perf/test_rl_recorder.py b/tests/perf/test_rl_recorder.py
deleted file mode 100644
index 4f580292f..000000000
--- a/tests/perf/test_rl_recorder.py
+++ /dev/null
@@ -1,213 +0,0 @@
-"""Framework-independent rollout iteration recording and future observation."""
-
-import json
-from types import SimpleNamespace as NS
-
-import pytest
-
-from twinkle.perf import RLPerformanceRecorder, reduce_memory, sample_lengths
-
-
-def test_plain_lengths_and_update_numbers_need_no_training_objects(tmp_path):
-    now = [0.0]
-    perf = RLPerformanceRecorder(tmp_path, [("train", [0, 1]), ("shared", [0])], clock=lambda: now[0])
-    perf.begin(4)
-    perf.received_lengths([3, 4], [2, 3], truncated=[False, True], assistant_turns=[1, 2])
-    perf.record_training([0, 1, 0, 1])
-    now[0] = 2.0
-    perf.finish(6, publications=1, resource_metrics={"resource/replicas": 2})
-    perf.write_summary(6)
-    perf.close()
-    row = json.loads((tmp_path / "performance.jsonl").read_text())
-    assert row["data"]["perf/throughput"] == 3
-    assert row["data"]["training/sample_visits"] == 4
-    assert row["data"]["training/unique_samples_trained"] == 2
-    assert row["data"]["training/optimizer_steps_this_iteration"] == 2
-    assert row["data"]["training/weight_publications_this_iteration"] == 1
-    assert row["data"]["response/truncated_ratio"] == 0.5
-    assert row["data"]["resource/replicas"] == 2
-
-
-def test_plain_snapshot_publications_use_actual_counter_deltas(tmp_path):
-    now = [0.0]
-    perf = RLPerformanceRecorder(tmp_path, [("train", [0])], clock=lambda: now[0], track_batch_wait=True)
-    for step, publish_time, publications in [(1, 0.5, 1), (2, 0.5, 1)]:
-        perf.begin(step - 1)
-        now[0] += 1
-        perf.received_lengths([3], [2])
-        now[0] += 1
-        perf.finish(step, stream_stats={"publish_time": publish_time, "publishes": publications})
-    perf.close()
-    rows = [json.loads(line)["data"] for line in (tmp_path / "performance.jsonl").read_text().splitlines()]
-    assert [row["training/weight_publications_this_iteration"] for row in rows] == [1, 0]
-    assert [row["timing_s/update_weights"] for row in rows] == [0.5, 0]
-    assert all(row["timing_s/gen"] == 1 for row in rows)
-
-
-def sample(prompt=3, response=2, *, tool=0):
-    ids = list(range(prompt + response + tool))
-    labels = [-100] * prompt + ids[prompt : prompt + response] + [-100] * tool
-    return NS(encoded={"input_ids": ids, "labels": labels[1:] + labels[:1]}, truncated=False)
-
-
-def test_lengths_include_tool_observations_but_not_padding():
-    s = sample(tool=4)
-    assert sample_lengths(s) == (3, 6)
-    s.encoded["input_ids"] += [0, 0]
-    # Output-order labels for 3 prompt, 2 assistant, 4 observation and 2 padding tokens.
-    labels = [-100] * 3 + [3, 4] + [-100] * 6
-    s.encoded["labels"] = labels[1:] + labels[:1]
-    s.encoded["attention_mask"] = [1] * 9 + [0, 0]
-    assert sample_lengths(s) == (3, 6)
-    assert sample_lengths(sample(response=0)) == (3, 0)
-
-
-def test_iteration_counts_tokens_once_across_optimizer_replay_and_flushes(tmp_path):
-    now = [10.0]
-    perf = RLPerformanceRecorder(
-        tmp_path, [("trainer", [0, 1]), ("shared", [0]), ("rollout", [2, 3])], mode="colocate", clock=lambda: now[0]
-    )
-    perf.begin(7)
-    perf.received([sample(), sample(prompt=4, response=3)])
-    perf.sample_visits = 8  # four epochs over two samples
-    perf.trained_indices = {0, 1}
-    now[0] = 12.0
-    perf.finish(11)
-    row = json.loads((tmp_path / "performance.jsonl").read_text())  # before close
-    assert row["step_unit"] == "rl_iteration"
-    assert row["data"]["perf/total_num_tokens"] == 12
-    assert row["data"]["perf/throughput"] == 1.5  # 12 tokens / 2s / 4 physical GPUs
-    assert row["data"]["training/optimizer_steps_this_iteration"] == 4
-    assert row["data"]["training/sample_visits"] == 8
-    perf.close()
-
-
-def test_remote_stage_waits_without_collecting_payload(tmp_path):
-    now = [0.0]
-    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate", clock=lambda: now[0])
-    perf.begin(0)
-
-    class Future:
-        _is_lazy_collect = True
-
-        def wait(self):
-            now[0] += 3.0
-
-        def __call__(self):
-            pytest.fail("must not transfer tensor payloads to the driver")
-
-    future = Future()
-    assert perf.call("ref", lambda: future) is future
-    assert perf.timings["ref"] == 3.0
-
-
-def test_memory_reduction_is_rank_max_for_peaks_host_mean_for_ram():
-    result = reduce_memory(
-        [
-            {"perf/max_memory_allocated_gb": 3, "perf/cpu_memory_used_gb": 8},
-            {"perf/max_memory_allocated_gb": 5, "perf/cpu_memory_used_gb": 10},
-        ]
-    )
-    assert result == {"perf/max_memory_allocated_gb": 5, "perf/cpu_memory_used_gb": 9}
-
-
-def test_side_effect_future_failure_is_not_a_successful_update(tmp_path):
-    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate")
-    perf.begin(0)
-
-    class FailedUpdate:
-        _is_lazy_collect = True
-
-        def wait(self):
-            return self
-
-        def result(self):
-            raise RuntimeError("optimizer failed")
-
-    with pytest.raises(RuntimeError, match="optimizer failed"):
-        perf.call("optimizer", FailedUpdate, collect_result=True)
-
-
-def test_reporter_time_excluded_but_full_wall_time_is_saved(tmp_path):
-    now = [0.0]
-    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate", clock=lambda: now[0])
-    perf.begin(0)
-    perf.received([sample()])
-    now[0] = 2.0
-    with perf.exclude_logging():
-        now[0] = 5.0
-    now[0] = 7.0
-    perf.finish(1)
-    data = json.loads((tmp_path / "performance.jsonl").read_text())["data"]
-    assert data["perf/time_per_step"] == 4.0
-    assert data["observability/iteration_wall_seconds"] == 7.0
-    assert data["observability/excluded_logging_seconds"] == 3.0
-    perf.close()
-
-
-def test_auxiliary_forwards_keep_original_parallel_submission(tmp_path):
-    import threading
-
-    ready = threading.Event()
-    launches = []
-    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate")
-    perf.begin(0)
-
-    class Forward:
-        _is_lazy_collect = True
-
-        def wait(self):
-            assert ready.wait(2), "forwards were serialized before all lanes could launch"
-
-        def __call__(self):
-            pytest.fail("do not collect tensor payloads")
-
-    def submit(name):
-        launches.append(name)
-        if len(launches) == 3:
-            ready.set()
-        return Forward()
-
-    try:
-        for name in ("ref", "teacher", "old_log_prob"):
-            perf.launch_forward(name, submit, name)
-        perf.join_forwards()
-        assert set(perf.timings) == {"ref", "teacher", "old_log_prob"}
-        assert all(value > 0 for value in perf.timings.values())
-    finally:
-        ready.set()
-        perf.close()
-
-
-def test_summary_records_terminal_driver_counters_after_teardown(tmp_path):
-    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="adaptive", parameter_sync_step=2)
-    perf.begin(0)
-    perf.received([sample()])
-    stats = {"publish_time": 0.0, "publishes": 0, "dropped_incomplete": 0, "total_time": 0.0, "final_version": 0}
-    perf.finish(1, stream_stats=stats)
-    stats.update(dropped_incomplete=3, total_time=12.0, final_version=2)
-    perf.write_summary(1, stream_stats=stats, status="loop_completed")
-    summary = json.loads((tmp_path / "performance_summary.jsonl").read_text())
-    row = json.loads((tmp_path / "performance.jsonl").read_text())
-    assert summary["run_id"] == row["run_id"]
-    assert summary["data"]["stream/dropped_incomplete"] == 3
-    assert summary["data"]["stream/final_version"] == 2
-    assert summary["data"]["stream/total_time"] == 12.0
-    assert "stream/total_time" not in row["data"]
-    assert "stream/dropped_incomplete" not in row["data"]
-    assert row["parameter_sync_step_unit"] == "optimizer_update"
-    assert row["iteration_scope"] == "one_consumed_rollout_batch"
-    perf.close()
-
-
-def test_assistant_turn_count_has_an_explicit_unit(tmp_path):
-    perf = RLPerformanceRecorder(tmp_path, [("trainer", [0])], mode="colocate")
-    perf.begin(0)
-    s = sample()
-    s.rollout_infos = {"turns": 1}  # Count assistant generations independently of user/tool messages.
-    perf.received([s])
-    perf.finish(1)
-    data = json.loads((tmp_path / "performance.jsonl").read_text())["data"]
-    assert data["rollout/assistant_turns/mean"] == 1
-    assert "num_turns/mean" not in data
-    perf.close()