feat: one prime monitor, sft online evals on the training run

docs/eval.md

@@ -130,7 +130,7 @@ Each source's epoch streams as an evaluation on the Prime platform. The launcher
 uv run eval gsm8k -n 32 -r 4 -c 8
 ```
 
-The monitor opens one platform evaluation per env when its epoch starts, streams every episode to it as it lands, and closes it with the epoch's aggregates. The run leaves its platform identity in `monitors/prime/run.json` (the training run's id and URL, or one entry per eval epoch), and the dashboard's top bar shows a `View on Prime` link from the moment the evaluation opens: a direct link for a training run or a single evaluation, a menu when a run has several.
+The monitor opens one Prime evaluation per env when its epoch starts, streams every episode to it as it lands, and closes it with the epoch's aggregates. The run leaves its Prime identity in `monitors/prime/run.json` (one entry per evaluation), and the dashboard's top bar shows a `View on Prime` link from the moment the evaluation opens: a direct link for a training run or a single evaluation, a menu when a run has several.
 
 ## Metrics
 

docs/training.md

@@ -491,7 +491,7 @@ Name the Prime run in TOML:
 name = "my-experiment"
 ```
 
-On `sft` the monitor registers the run with the dataset-batched training fields (`max_steps`, `batch_size`, `seq_len`) and streams the trainer's per-step metrics and validation losses; SFT has no rollouts, so no episodes are uploaded.
+The process that supervises the launch registers the run before the components start, and every component logs to it. That is the `rl` or `sft` launcher, also inside a single-node SLURM job, or the run owner that wraps a multi-node SLURM job's step. The run's id is also the run identity (`PRL_RUN_ID`) and the W&B run id. On `sft` the run carries the dataset-batched training fields (`max_steps`, `batch_size`, `seq_len`) and the trainer's per-step metrics and validation losses. Online evals log to the same run, like an `rl` run's: the online-eval process uploads its eval metrics and every eval episode. The components only flush their uploads; the process that registered the run finishes it once every component exited. A launch that fails marks the run failed, and so does a SLURM job that SLURM cancels or stops at its time limit.
 
 The platform run needs `PRIME_API_KEY` (`prime login` or the env var) and a team (`PRIME_TEAM_ID`, or the team selected with `prime login`) enabled for external runs. Like the W&B monitor, a run without a key, or one whose registration fails (no team, a team outside the allowlist), fails at launch: authenticate, select a team with `prime switch`, or pass `--no-monitors.prime`. `PRIME_RUNS_MODE=disabled` keeps the monitor configured but opens no platform run; `RUN_ID=<id>` attaches to an external run a launcher already created instead of registering a new one. Currently internal-only.
 

packages/prime-rl-configs/src/prime_rl/configs/eval.py

@@ -4,7 +4,7 @@
 import verifiers.v1 as vf
 from pydantic import AliasChoices, Field, model_validator
 
-from prime_rl.configs.monitors import EvalMonitorsConfig
+from prime_rl.configs.monitors import MonitorsConfig
 from prime_rl.configs.orchestrator import ConcurrencyConfig, EvalSourcesConfig, ScheduledEvalConfig
 from prime_rl.configs.shared import (
     ClientConfig,
@@ -108,7 +108,7 @@ class EvalConfig(ServedEvalConfig):
 
     log: LogConfig = LogConfig()
 
-    monitors: EvalMonitorsConfig = EvalMonitorsConfig()
+    monitors: MonitorsConfig = MonitorsConfig()
     """Metric monitors (``monitors.wandb``, ``monitors.file``, ``monitors.prime``)."""
 
     @property
@@ -172,14 +172,9 @@ class SFTOnlineEvalConfig(ScheduledEvalConfig, ServedEvalConfig):
 
     log: LogConfig = LogConfig()
 
-    monitors: EvalMonitorsConfig = EvalMonitorsConfig()
-    """Metric monitors (``monitors.wandb``, ``monitors.file``, ``monitors.prime``).
-
-    ``EvalMonitorsConfig``, not the base ``MonitorsConfig``: the launcher
-    converts the trainer's ``[monitors.prime]`` into a
-    ``PrimeEvalMonitorConfig`` for the online-eval process, and the base
-    type would drop (or forbid) the prime block at the eval.json
-    dump/re-parse boundary."""
+    monitors: MonitorsConfig = MonitorsConfig()
+    """Metric monitors (``monitors.wandb``, ``monitors.file``, ``monitors.prime``): the
+    trainer's, so the process logs to the trainer's W&B and Prime runs."""
 
     @model_validator(mode="after")
     def auto_setup_broadcasts_dir(self):

packages/prime-rl-configs/src/prime_rl/configs/monitors.py

@@ -43,14 +43,9 @@ class FileMonitorConfig(BaseConfig):
     not the records, so this only trades record size against overlay precision."""
 
 
-class PrimeTrainMonitorConfig(BaseConfig):
+class PrimeMonitorConfig(BaseConfig):
     name: str | None = None
-    """Run name shown on the platform. Inherits ``run.name`` when unset."""
-
-
-class PrimeEvalMonitorConfig(BaseConfig):
-    name: str | None = None
-    """Evaluation name shown on the platform. Inherits ``run.name`` when unset."""
+    """Run or evaluation name shown on Prime. Inherits ``run.name`` when unset."""
 
 
 class MonitorsConfig(BaseConfig):
@@ -60,14 +55,7 @@ class MonitorsConfig(BaseConfig):
     file: FileMonitorConfig | None = FileMonitorConfig()
     """Log metrics and episode traces to the run's output directory. On by default; disable with ``--no-monitors.file``."""
 
-
-class TrainMonitorsConfig(MonitorsConfig):
-    prime: PrimeTrainMonitorConfig | None = PrimeTrainMonitorConfig()
-    """Log metrics and episodes to the Prime Intellect platform. On by default and
-    needs a Prime API key; disable with ``--no-monitors.prime``."""
-
-
-class EvalMonitorsConfig(MonitorsConfig):
-    prime: PrimeEvalMonitorConfig | None = PrimeEvalMonitorConfig()
-    """Upload every eval source's epoch as an evaluation on the Prime Intellect platform. On by
-    default and needs a Prime API key; disable with ``--no-monitors.prime``."""
+    prime: PrimeMonitorConfig | None = PrimeMonitorConfig()
+    """Log metrics and episodes to the Prime Intellect platform: a training run's processes
+    log to one platform run, and a standalone eval uploads each source's epoch as an
+    evaluation. On by default and needs a Prime API key; disable with ``--no-monitors.prime``."""

packages/prime-rl-configs/src/prime_rl/configs/orchestrator.py

@@ -14,7 +14,7 @@
     GRPOAlgoConfig,
 )
 from prime_rl.configs.diff import Frozen, Live
-from prime_rl.configs.monitors import TrainMonitorsConfig
+from prime_rl.configs.monitors import MonitorsConfig
 from prime_rl.configs.shared import (
     BaseModelConfig,
     ControlConfig,
@@ -500,7 +500,7 @@ class OrchestratorConfig(BaseConfig):
     env_vars: EnvVars = {}
     """Extra environment variables for the orchestrator process(es). Merged on top of the launcher defaults."""
 
-    monitors: TrainMonitorsConfig = TrainMonitorsConfig()
+    monitors: MonitorsConfig = MonitorsConfig()
     """Metric monitors (``monitors.wandb``, ``monitors.file``, ``monitors.prime``)."""
 
     collect_inference_metrics: bool = True

packages/prime-rl-configs/src/prime_rl/configs/rl.py

@@ -7,7 +7,7 @@
 
 from prime_rl.configs.inference import InferenceConfig
 from prime_rl.configs.inference import WeightBroadcastConfig as InferenceWeightBroadcastConfig
-from prime_rl.configs.monitors import FileMonitorConfig, PrimeTrainMonitorConfig
+from prime_rl.configs.monitors import FileMonitorConfig, PrimeMonitorConfig
 from prime_rl.configs.orchestrator import OrchestratorConfig, TrainSamplingConfig
 from prime_rl.configs.shared import (
     BaseMultiNodeDeploymentConfig,
@@ -79,7 +79,7 @@ class SharedMonitorsConfig(BaseConfig):
     file: FileMonitorConfig | None = None
     """Shared local JSONL metric sink. If set, enables ``<output_dir>/monitors/file/metrics.jsonl`` on both trainer and orchestrator."""
 
-    prime: PrimeTrainMonitorConfig | None = None
+    prime: PrimeMonitorConfig | None = None
     """Prime platform monitor. Propagated to the orchestrator only — the trainer has no platform integration."""
 
 
@@ -342,6 +342,16 @@ def require_equal(name: str, values: dict[str, Any]) -> None:
 
         # The orchestrator queries the inference server it is paired with; without one, it
         # must match the trainer whose weights the external server receives.
+        # The components log to one run per monitor, so a monitor is on in all or none.
+        for name in ("wandb", "file"):
+            require_equal(
+                f"monitors.{name}",
+                {
+                    f"{component}.monitors.{name}": getattr(config.monitors, name) is not None
+                    for component, config in (("trainer", trainer), ("orchestrator", orchestrator))
+                },
+            )
+
         if inference is not None:
             require_equal(
                 "deployment.gpus_per_node",

packages/prime-rl-configs/src/prime_rl/configs/sft.py

@@ -11,7 +11,7 @@
 from prime_rl.configs.eval import SFTOnlineEvalConfig
 from prime_rl.configs.inference import InferenceConfig
 from prime_rl.configs.inference import WeightBroadcastConfig as InferenceWeightBroadcastConfig
-from prime_rl.configs.monitors import TrainMonitorsConfig
+from prime_rl.configs.monitors import MonitorsConfig
 from prime_rl.configs.shared import (
     BaseMultiNodeDeploymentConfig,
     BaseSingleNodeDeploymentConfig,
@@ -217,7 +217,7 @@ class SFTConfig(BaseTrainerConfig):
     resume: ResumeConfig | None = None
     """Resume the run from a checkpoint (point at it with the previous run's ``run.name``). Without ``[ckpt]`` the run loads the checkpoint but saves no new ones. If None, does not resume."""
 
-    monitors: TrainMonitorsConfig = TrainMonitorsConfig()
+    monitors: MonitorsConfig = MonitorsConfig()
     """Metric monitors (``monitors.wandb``, ``monitors.file``, ``monitors.prime``)."""
 
     run: RunConfig = Field(default_factory=RunConfig)

packages/prime-rl-configs/src/prime_rl/utils/config.py

@@ -15,7 +15,7 @@ def default_output_dir() -> Path:
     return Path(os.environ.get("PRL_OUTPUT_DIR", "outputs"))
 
 
-def dump_resolved_config(config: BaseModel, exclude: set[str] | None = None) -> dict:
+def dump_resolved_config(config: BaseModel, exclude: set[str] | dict[str, Any] | None = None) -> dict:
     """Dump a resolved config for a machine-written JSON artifact.
 
     Resolved configs are written as JSON, not TOML: JSON keeps nulls, so explicit None

skills/training/monitor-run/SKILL.md

@@ -217,7 +217,7 @@ stream records, newest winning.
 wc -l {run_dir}/monitors/file/traces/stream.index.jsonl
 zstd -dcf {run_dir}/monitors/file/traces/stream/* | jq '.traces[].rewards'
 zstd -dcf {run_dir}/monitors/file/traces/stream/* | jq 'select(.ok | not) | {id, env: .env.id, errors}'
-jq '{trace_id, info}' {run_dir}/monitors/file/traces/annotations/orch.index.jsonl
+jq '{trace_id, info}' {run_dir}/monitors/file/traces/annotations/orchestrator.index.jsonl
 ```
 
 The batches consumed by the trainer are shipped over ZMQ by default, so nothing binary is written. With `rollout_transport.type = "filesystem"` they land at `{run_dir}/batches/step_{n}/rank_<rank>.bin` (one packed micro-batch file per trainer DP rank).

src/prime_rl/dashboard/static/app.js

@@ -461,17 +461,17 @@ const INFERENCE_PANELS = [
 ];
 
 const TRAINER_KEY_RE = /^(perf|optim|loss|entropy|system|mismatch_kl|kl_ent_ratio|is_masked|masked_|unmasked_|max_vio|routing_|ref_kl|val)[/_]?/;
-const ORCH_KEY_RE = /^(train|batch|staleness|curriculum|eval)\//;
+const ORCHESTRATOR_KEY_RE = /^(train|batch|staleness|curriculum|eval)\//;
 
 function rowProducer(row, meta) {
   if (typeof row.producer === "string") return row.producer; // stamped by FileMonitor
   // legacy files predate the producer stamp: infer from key namespaces
   if (meta?.type === "sft") return "trainer";
   for (const key of Object.keys(row)) {
     if (TRAINER_KEY_RE.test(key)) return "trainer";
-    if (ORCH_KEY_RE.test(key)) return "orch";
+    if (ORCHESTRATOR_KEY_RE.test(key)) return "orchestrator";
   }
-  return Object.keys(row).some((k) => k.startsWith("progress/")) ? "orch" : "trainer";
+  return Object.keys(row).some((k) => k.startsWith("progress/")) ? "orchestrator" : "trainer";
 }
 
 /* store = {byKey, timeKeys, timeZero} — the primary run's is state.metrics,

src/prime_rl/entrypoints/dashboard.py

@@ -26,12 +26,14 @@
 import urllib.request
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Callable
+from typing import TYPE_CHECKING, Callable
 
-from prime_rl.utils.links import PRIME_EVAL_LINKS, PRIME_LINKS, WANDB_LINKS
 from prime_rl.utils.logger import get_logger
 from prime_rl.utils.pathing import CACHE_DIR
 
+if TYPE_CHECKING:
+    from prime_rl.configs.monitors import MonitorsConfig
+
 STATE_DIR = CACHE_DIR / "dashboard"
 DAEMON_FILE = STATE_DIR / "daemon.json"
 DIRS_FILE = STATE_DIR / "dirs.json"
@@ -118,7 +120,7 @@ def ensure_dashboard(output_dir: Path, logger) -> str | None:
     return None
 
 
-LINK_NAMES = {PRIME_LINKS: "Prime", PRIME_EVAL_LINKS: "Prime", WANDB_LINKS: "W&B"}
+LINK_NAMES = {"prime": "Prime", "wandb": "W&B"}
 
 
 def log_run_links(dashboard_url: str | None, links: Callable[[], dict[str, dict[str, str]]] | None = None) -> None:
@@ -145,21 +147,19 @@ def show(name: str, url: str, suffix: str) -> None:
 
 
 def run_links(
-    run_dir: Path, attempt_log_dir: Path, monitors: dict[str, object | None], stop: Callable[[], bool]
+    run_dir: Path, attempt_log_dir: Path, monitors: "MonitorsConfig", stop: Callable[[], bool]
 ) -> Callable[[], dict[str, dict[str, str]]] | None:
-    """The ``links`` argument of ``log_run_links``: waits for every configured monitor
-    in ``monitors`` (links-record key -> its config, None when off) that can open a
-    page. A monitor turned off by its mode variable (``$PRIME_RUNS_MODE``, an offline
+    """The ``links`` argument of ``log_run_links``: waits for every monitor in
+    ``monitors`` that is on and can open a page, keyed by its config field. A monitor turned off by its mode variable (``$PRIME_RUNS_MODE``, an offline
     or disabled ``$WANDB_MODE``) opens none; one without credentials fails its
     component, which ends the wait through ``stop``."""
     from prime_rl.utils.links import wait_for_links
 
-    def opens_pages(key: str) -> bool:
-        if key == WANDB_LINKS:
-            return os.environ.get("WANDB_MODE") not in ("offline", "disabled")
-        return os.environ.get("PRIME_RUNS_MODE", "online") == "online"
-
-    waited = [key for key, config in monitors.items() if config is not None and opens_pages(key)]
+    opens_pages = {
+        "prime": monitors.prime is not None and (os.environ.get("PRIME_RUNS_MODE") or "online") == "online",
+        "wandb": monitors.wandb is not None and os.environ.get("WANDB_MODE") not in ("offline", "disabled"),
+    }
+    waited = [key for key, opens in opens_pages.items() if opens]
     if not waited:
         return None
     return functools.partial(wait_for_links, run_dir, attempt_log_dir, waited, stop)

src/prime_rl/entrypoints/eval.py

@@ -121,7 +121,6 @@ def main():
     sys.argv = [sys.argv[0], *argv]
 
     from prime_rl.entrypoints.dashboard import ensure_dashboard, log_run_links, run_links
-    from prime_rl.utils.links import PRIME_EVAL_LINKS, WANDB_LINKS
     from prime_rl.utils.logger import setup_logger
     from prime_rl.utils.pathing import (
         format_config_message,
@@ -195,8 +194,7 @@ def sigterm_handler(signum, frame):
     # Each source's Prime evaluation opens once the eval counts its tasks, so the
     # banner waits for the links next to the running eval.
     eval_done = threading.Event()
-    monitors = {PRIME_EVAL_LINKS: config.monitors.prime, WANDB_LINKS: config.monitors.wandb}
-    links = run_links(config.run_dir, log_dir, monitors, eval_done.is_set)
+    links = run_links(config.run_dir, log_dir, config.monitors, eval_done.is_set)
     banner = threading.Thread(target=log_run_links, args=(dashboard_url, links), daemon=True)
     banner.start()
     try:

src/prime_rl/entrypoints/rl.py

@@ -5,6 +5,7 @@
 import sys
 import time
 from pathlib import Path
+from typing import Any
 from urllib.parse import urlparse
 
 from prime_rl.configs.algorithm import FrozenModelConfig
@@ -14,7 +15,6 @@
 from prime_rl.entrypoints.dashboard import ensure_dashboard, log_run_links, run_links
 from prime_rl.entrypoints.inference import mooncake_offloads, vllm_overrides_fragment, write_pd_configs
 from prime_rl.utils.config import cli, dump_resolved_config, dump_rl_components
-from prime_rl.utils.links import PRIME_LINKS, WANDB_LINKS
 from prime_rl.utils.logger import get_logger, setup_logger
 from prime_rl.utils.pathing import (
     clean_future_steps,
@@ -68,7 +68,7 @@ def env_server_names(config: RLConfig, split: str) -> list[str]:
     return [source.resolved_name for source_split, source in env_servers(config) if source_split == split]
 
 
-def write_config(config: RLConfig, output_dir: Path, exclude: set[str] | None = None) -> None:
+def write_config(config: RLConfig, output_dir: Path, exclude: set[str] | dict[str, Any] | None = None) -> None:
     """Write resolved config to disk, excluding launcher-only fields."""
     output_dir.mkdir(parents=True, exist_ok=True)
     with open(output_dir / RL_CONFIG, "w") as f:
@@ -333,8 +333,8 @@ def rl_local(config: RLConfig):
         )
 
         logger.success("Launcher complete")
-        monitors = {PRIME_LINKS: config.orchestrator.monitors.prime, WANDB_LINKS: config.orchestrator.monitors.wandb}
-        log_run_links(dashboard_url, run_links(config.run_dir, log_dir, monitors, lambda: bool(processes.error_queue)))
+        links = run_links(config.run_dir, log_dir, config.orchestrator.monitors, lambda: bool(processes.error_queue))
+        log_run_links(dashboard_url, links)
 
         # Trainer and orchestrator completion is the successful stop condition.
         processes.wait("trainer", "orchestrator")
@@ -522,8 +522,12 @@ def rl_slurm(config: RLConfig):
             env_names={"train": train_env_names, "eval": eval_env_names},
         )
     else:
+        # The job's run owner registers the Prime run with the whole config; nothing runs it.
+        # It keeps [slurm], which multi-node needs, so it drops the copy in [inference].
+        write_config(config, config_dir, exclude={"dry_run": True, "clean": True, "inference": {"slurm"}})
         write_subconfigs(config, config_dir)
-        logger.info(f"Configs:\n{format_config_message(config_dir, 'rl', rl_config_components(config, config_dir))}")
+        components = [("RL", config_dir / RL_CONFIG), *rl_config_components(config, config_dir)]
+        logger.info(f"Configs:\n{format_config_message(config_dir, 'rl', components)}")
 
         train_env_names = env_server_names(config, "train")
         eval_env_names = env_server_names(config, "eval")
@@ -560,7 +564,6 @@ def rl_slurm(config: RLConfig):
 
 
 def rl(config: RLConfig):
-    # TODO: fetch the run id from the Prime SDK once runs are registered there.
     ckpt_output_dir = config.ckpt.output_dir if config.ckpt else None
     prepare_run(config, ckpt_output_dir)
 
@@ -591,9 +594,14 @@ def rl(config: RLConfig):
 
         pre_download_model(config.trainer.model.name, skip_weights=config.trainer.model.debug.random_init)
 
+    from prime_rl.monitors.prime import launch_run
+
+    # On SLURM the job registers the run: the in-job `rl` (single node) or the
+    # batch script's run owner (multi node), so that a live process owns its status.
     if config.slurm is not None:
         rl_slurm(config)
-    else:
+        return
+    with launch_run(config, None if config.dry_run else config.orchestrator.monitors.prime):
         rl_local(config)