prime-runs

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
     "openai>=2.54.0,<3.0.0",
     "prime-tunnel>=0.1.8",
     "prime-sandboxes>=0.2.39",
+    "prime-runs>=0.1.1",
     "pydantic>=2.12.3",
     "rich>=11.0.0",
     "tenacity>=8.5.0",
@@ -146,6 +147,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,7 @@ def _eval_config(
         output_dir=output_dir.parent,
         run={"dir": output_dir.name},
         model=CI_MODEL,
+        push=False,
     )
 
 

verifiers/v1/cli/dashboard/eval.py

@@ -281,17 +281,25 @@ 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` line under the rollouts: dim with the run's URL while it streams,
+    white once pushed, red when it failed. `None` 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 what 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 push.incomplete and push.url:
+        # Closed out, but the uploader lost records: what landed is there, say what didn't.
+        line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold")
+        line.append(f"  incomplete: {push.incomplete}", style="red")
+    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")
     return Group(Rule(style="dim"), line)
 
 

verifiers/v1/cli/eval/main.py

@@ -134,10 +134,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,35 @@
 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,
+    log_episodes,
+    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 cancels the rest and waits for them
+    to unwind, so nothing keeps uploading into a run the caller is already closing.
+    Not a `TaskGroup`: that wraps errors in an `ExceptionGroup`, and `main` would no
+    longer see 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: wait for every task, not just the first to cancel.
+        await asyncio.gather(*tasks, return_exceptions=True)
+        raise
+
+
 RunSlotFn = Callable[[RunSlot], Awaitable[Episode]]
 OnComplete = Callable[[Episode], Awaitable[None]]
 
@@ -203,47 +229,56 @@ 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()
+
+    # Opened before the first rollout so every episode streams as it lands.
+    run = open_run(config, push_state, num_examples=len(tasks))
+    # Resumed rollouts are part of this run too.
+    log_episodes(run, 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(log_episodes, run, [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)]
+    # The run is closed out whatever breaks, backend setup and teardown included.
+    # Once `finish_run` has started, the outcome is its: a Ctrl-C then cancels only
+    # this task's await, not the worker, which completes the run regardless (the
+    # SDK serialises `finish()` calls and the first one reports the status). So an
+    # interrupt from that point on is not an abort.
+    closing = False
+    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.
+                closing = True
+                await asyncio.to_thread(finish_run, run, episodes, push_state)
+    except BaseException as e:
+        if not closing:
+            await asyncio.to_thread(abort_run, run, e, push_state)
+        raise
     return episodes

verifiers/v1/configs/cli/eval.py

@@ -48,13 +48,23 @@ 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()))
+    _id: str | None = PrivateAttr(default=None)
 
     @property
     def id(self) -> str:
+        """The run's one id, assigned by `open_run` from the prime-runs handle: the
+        platform's evaluation id online, the SDK's local id otherwise. The run mints
+        none of its own, so the run dir, every trace and the dashboard agree."""
+        if self._id is None:
+            raise RuntimeError("the run has no id until `open_run` has opened it")
         return self._id
 
+    def assign_id(self, run_id: str) -> None:
+        """Called once by `open_run`, before the first rollout."""
+        if self._id is not None and self._id != run_id:
+            raise RuntimeError(f"the run already has id {self._id!r}")
+        self._id = run_id
+
 
 class EvalConfig(BaseConfig):
     env: SerializeAsAny[EnvConfig] = SingleAgentEnvConfig()