prime-runs

pyproject.toml

@@ -36,6 +36,7 @@ dependencies = [
     "openai-agents>=0.8.2",
     "prime-tunnel>=0.1.8",
     "prime-sandboxes>=0.2.39",
+    "prime-runs>=0.1.0",
     "pydantic>=2.12.3",
     "requests",
     "rich>=11.0.0",
@@ -163,6 +164,8 @@ url = "https://pypi.org/simple"
 default = true
 
 [tool.uv.sources]
+# TEMPORARY: prime-runs is not on PyPI yet
+prime-runs = { git = "https://github.com/PrimeIntellect-ai/prime.git", subdirectory = "packages/prime-runs", branch = "feature/prime-runs-sdk" }
 compact = { path = "environments/compact", editable = true }
 glossary = { path = "environments/glossary", editable = true }
 deepwiki = { path = "environments/deepwiki", editable = true }
@@ -192,6 +195,7 @@ ty = "2026-07-28T00:00:00Z"
 # PrimeIntellect-published on PyPI (trusted publisher)
 prime-tunnel = false
 prime-sandboxes = false
+prime-runs = false
 prime-pydantic-config = false
 renderers = false
 

tests/v1/conftest.py

@@ -188,6 +188,9 @@ def _eval_config(
         output_dir=output_dir.parent,
         run={"dir": output_dir.name},
         model=CI_MODEL,
+        # `run_eval` opens a platform run before the first rollout when `push` is
+        # on and an API key is present (it is, in CI); the E2Es must stay local.
+        push=False,
     )
 
 

verifiers/v1/cli/dashboard/eval.py

@@ -281,17 +281,27 @@ def Overview(config: EvalConfig) -> Table:
 
 
 def _push_footer(push: "PushState | None") -> Group | None:
-    """The `--push` status line under the rollouts, shown once the run finishes and the upload
-    begins: dim `Pushing traces...` while it runs, then white `Traces pushed (<url>)` or red
-    `Trace push failed (<err>)`. `None` (no line) until the upload starts and when `--push` is off."""
+    """The `--push` status line under the rollouts. The run opens before the first rollout and
+    its traces stream up as they land, so the line carries the run's URL for the whole eval:
+    dim `Pushing traces (<url>)` while rollouts are still going, then white `Traces pushed
+    (<url>)` — with anything that degraded along the way appended in yellow — or red `Trace
+    push failed (<err>)` when there is no run to show. `None` (no line) when `--push` is off
+    or the run stayed local."""
     if push is None or not push.started:
         return None
-    if not push.done:
-        line = Text("Pushing traces...", style="dim")
-    elif push.url:
+    if push.error and push.url:
+        # The run exists and holds everything that streamed up; only closing it out
+        # failed.
         line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold")
-    else:
+        line.append(f"  not closed out: {push.error}", style="red")
+    elif push.error:
         line = Text(f"Trace push failed ({push.error})", style="red", overflow="fold")
+    elif not push.finished:
+        line = Text(f"Pushing traces ({push.url})", style="dim", overflow="fold")
+    else:
+        line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold")
+        if push.warning:  # pushed, but not all of it - say what went wrong
+            line.append(f"  {push.warning}", style="yellow")
     return Group(Rule(style="dim"), line)
 
 

verifiers/v1/cli/eval/main.py

@@ -17,6 +17,7 @@
     write_config,
 )
 from verifiers.v1.cli.resolve import (
+    config_file_ref,
     extract_id,
     narrow_config,
     plugin_errors,
@@ -67,6 +68,8 @@ def main(argv: list[str] | None = None) -> None:
             *argv,
         ]  # let prime-pydantic-config render help/errors
         config = cli(config_type)
+    # The `@ eval.toml` this run was launched from.
+    config.run.record_source(config_file_ref(argv))
     # A named run directory is re-entered only by `--resume` or wiped by `--clean`: any
     # other write into it — the dry-run config included, which would clobber the
     # config a resume typically re-runs — would overwrite the previous run.
@@ -136,10 +139,6 @@ def main(argv: list[str] | None = None) -> None:
         # Graceful cleanup has already run (each rollout's `finally`); partial results are on
         # disk. Exit on the conventional Ctrl-C code without a traceback.
         raise SystemExit(130)
-    if config.push and config.rich is None:
-        from verifiers.v1.utils.platform import push_traces
-
-        push_traces(episodes, config)
     if (
         config.rich is None
     ):  # --rich is the whole output; otherwise dump each trace as JSON

verifiers/v1/cli/eval/runner.py

@@ -13,8 +13,8 @@
 import contextlib
 import logging
 import time
-from collections.abc import AsyncIterator, Awaitable, Callable
-from typing import cast
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
+from typing import TypeVar, cast
 
 from verifiers.v1.cli.dashboard import dashboard
 from verifiers.v1.cli.eval import resume
@@ -30,9 +30,37 @@
 from verifiers.v1.configs.serve import ServeConfig
 from verifiers.v1.env import Env, RunSlot
 from verifiers.v1.episode import Episode, EvalRunInfo
+from verifiers.v1.utils.platform import PushState, abort_run, finish_run, open_run
 
 logger = logging.getLogger(__name__)
 
+T = TypeVar("T")
+
+
+async def gather_rollouts(rollouts: Iterable[Awaitable[T]]) -> list[T]:
+    """`asyncio.gather`, but one rollout failing stops the others too.
+
+    Plain `gather` raises the first error and leaves the rest running. They then
+    keep going while the caller is already handling that error — still uploading
+    to a run it has just closed, still using an env it is tearing down.
+    Cancelling them here, and waiting for each one to finish unwinding, keeps
+    those two things from overlapping.
+
+    The error is re-raised exactly as it arrived, which is why this is not an
+    `asyncio.TaskGroup`: a TaskGroup wraps everything in an `ExceptionGroup`, and
+    `main` would stop recognizing a `KeyboardInterrupt` as Ctrl-C."""
+    tasks = [asyncio.ensure_future(rollout) for rollout in rollouts]
+    try:
+        return await asyncio.gather(*tasks)
+    except BaseException:
+        for task in tasks:
+            task.cancel()
+        # return_exceptions so this waits for all of them; without it the first
+        # cancellation would raise and the rest would be left running again.
+        await asyncio.gather(*tasks, return_exceptions=True)
+        raise
+
+
 RunSlotFn = Callable[[RunSlot], Awaitable[Episode]]
 OnComplete = Callable[[Episode], Awaitable[None]]
 
@@ -203,47 +231,57 @@ async def run_eval(config: EvalConfig) -> list[Episode]:
         asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None
     )
     write_lock = asyncio.Lock()
+    push_state = PushState() if config.push and config.rich is not None else None
+
+    # Opened before the first rollout so the platform's id is the run's id. A
+    # run that stays local keeps its own uuid rather than the SDK's placeholder.
+    run = open_run(config, push_state, num_examples=len(tasks))
+    if run.mode == "online":
+        config.run.adopt_id(run.id)
+    # A resume's kept rollouts are part of this run too, so they carry its id and
+    # go up with the rest
+    for episode in finished:
+        episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name))
+    run.log_episodes(finished)
 
     async def on_complete(episode: Episode) -> None:
         episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name))
         await append_episode(out, episode, write_lock)
+        await asyncio.to_thread(run.log_episodes, [episode])
 
     backend = (
         _in_process(env, config, semaphore, on_complete)
         if env is not None
         else _server(config, config.serve, semaphore, on_complete)
     )
-    async with backend as run_slot:
-        # The display slots: in-process ones are the env's own (it fills their live
-        # traces); a served rollout's is a client-side stand-in its worker never sees.
-        planned = [
-            slot
-            for task, n in plan
-            for slot in (
-                env.slots(task, n)
-                if env is not None
-                else [RunSlot(task) for _ in range(n)]
+    # Everything from bringing the backend up (serving resources, or the worker
+    # pool) to tearing it down is inside the try: a run that was opened is closed
+    # out whatever breaks, so none of them sits at running.
+    try:
+        async with backend as run_slot:
+            # The display slots: in-process ones are the env's own (it fills their live
+            # traces); a served rollout's is a client-side stand-in its worker never sees.
+            planned = [
+                slot
+                for task, n in plan
+                for slot in (
+                    env.slots(task, n)
+                    if env is not None
+                    else [RunSlot(task) for _ in range(n)]
+                )
+            ]
+            slots = [RunSlot.finished(episode) for episode in finished] + planned
+            display = (
+                dashboard(slots, config, start, push=push_state)
+                if config.rich is not None
+                else contextlib.nullcontext()
             )
-        ]
-        slots = [RunSlot.finished(episode) for episode in finished] + planned
-        push_state = None
-        if config.push and config.rich is not None:
-            from verifiers.v1.utils.platform import PushState
-
-            push_state = PushState()
-        display = (
-            dashboard(slots, config, start, push=push_state)
-            if config.rich is not None
-            else contextlib.nullcontext()
-        )
-        async with display:
-            results = await asyncio.gather(*(run_slot(slot) for slot in planned))
-            episodes = finished + list(results)
-            if (
-                push_state is not None
-            ):  # upload off the event loop so the view keeps refreshing
-                from verifiers.v1.utils.platform import push_traces
-
-                push_state.started = True
-                await asyncio.to_thread(push_traces, episodes, config, push_state)
+            async with display:
+                results = await gather_rollouts(run_slot(slot) for slot in planned)
+                episodes = finished + list(results)
+                # Drain and close out off the event loop so the view keeps refreshing.
+                await asyncio.to_thread(finish_run, run, episodes, push_state)
+    except BaseException as e:
+        await asyncio.to_thread(abort_run, run, e, push_state)
+        raise
     return episodes

verifiers/v1/cli/resolve.py

@@ -46,6 +46,16 @@ def references_config_file(argv: list[str]) -> bool:
     return any(arg.startswith("@") for arg in argv)
 
 
+def config_file_ref(argv: list[str]) -> str | None:
+    """The file a run was launched from — the root-level `@ <path>` or None."""
+    paths = [
+        argv[i + 1]
+        for i, arg in enumerate(argv)
+        if arg == "@" and i + 1 < len(argv) and not (i and argv[i - 1].startswith("--"))
+    ]
+    return paths[0] if len(paths) == 1 else None
+
+
 def extract_id(argv: list[str], field: str) -> str:
     """The chosen `<field>.id` from `--<field>.id <x>` (or `=<x>`) on the CLI, before
     the typed parse (the positional taskset shorthand is applied upstream). Two

verifiers/v1/configs/cli/eval.py

@@ -48,13 +48,33 @@ class RunConfig(BaseConfig):
     """Run directory name — the run writes to `output_dir / dir`. Defaults to `run.name`;
     set it only when the directory should differ from the display name."""
 
-    # TODO: fetch the id from the Prime SDK once runs are registered there.
     _id: str = PrivateAttr(default_factory=lambda: str(uuid4()))
+    """The platform's run id once `prime_runs.init()` has opened the run (see
+    `adopt_id`); the local uuid until then, and for a run that stays local."""
+
+    _source: str | None = PrivateAttr(default=None)
+    """The `@ file.toml` this run was launched from, recorded by the CLI."""
 
     @property
     def id(self) -> str:
         return self._id
 
+    @property
+    def source(self) -> str | None:
+        return self._source
+
+    def adopt_id(self, run_id: str) -> None:
+        """Take the platform's run id as this run's id.
+
+        Called once, before the first rollout, so that every trace is stamped
+        with the id the platform knows the run by."""
+        self._id = run_id
+
+    def record_source(self, path: str | None) -> None:
+        """Remember the config file the run was launched from, so it can be
+        uploaded verbatim with the run."""
+        self._source = path
+
 
 class EvalConfig(BaseConfig):
     env: SerializeAsAny[EnvConfig] = SingleAgentEnvConfig()