prime-runs

pyproject.toml

@@ -36,6 +36,7 @@ dependencies = [
     "openai-agents>=0.8.2",
     "prime-tunnel>=0.1.8",
     "prime-sandboxes>=0.2.37",
+    "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
 

verifiers/v1/cli/dashboard/eval.py

@@ -279,17 +279,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

@@ -18,6 +18,7 @@
     write_config,
 )
 from verifiers.v1.cli.resolve import (
+    config_file_ref,
     extract_id,
     narrow_config,
     plugin_errors,
@@ -68,6 +69,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.toml included, which would clobber the
     # config a resume typically re-runs — would overwrite the previous run.
@@ -146,10 +149,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 not config.rich:
-        from verifiers.v1.utils.platform import push_traces
-
-        push_traces(episodes, config)
     if not config.rich:  # --rich is the whole output; otherwise dump each trace as JSON
         for episode in episodes:
             for trace in episode.traces:

verifiers/v1/cli/eval/runner.py

@@ -4,7 +4,8 @@
 import contextlib
 import logging
 import time
-from typing import cast
+from collections.abc import Awaitable, Iterable
+from typing import TypeVar, cast
 
 from verifiers.v1.cli.dashboard import dashboard
 from verifiers.v1.cli.eval import resume
@@ -18,9 +19,36 @@
 from verifiers.v1.configs.cli.eval import EvalConfig
 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
+
 
 async def run_eval(env: Env, config: EvalConfig) -> list[Episode]:
     logger.info("eval config:\n%s", config.model_dump_json(indent=2))
@@ -76,38 +104,45 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]:
     logger.info("results: %s", out)
 
     write_lock = asyncio.Lock()
+    push_state = PushState() if config.push and config.rich else None
+
+    # Opened before the first rollout so the platform's id is the run's id
+    run = open_run(config, push_state)
+    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])
 
     # Serving resources (shared tool servers, interception) come up once for the
-    # run; plan slots inside so the env's agents borrow them.
-    async with env.serving():
-        planned = [slot for task, n in plan for slot in env.slots(task, n=n)]
-        slots = [RunSlot.finished(episode) for episode in finished] + planned
-        push_state = None
-        if config.push and config.rich:
-            from verifiers.v1.utils.platform import PushState
-
-            push_state = PushState()
-        display = (
-            dashboard(slots, config, start, push=push_state)
-            if config.rich
-            else contextlib.nullcontext()
-        )
-        async with display:
-            results = await asyncio.gather(
-                *(env.run_slot(slot, ctx, semaphore, on_complete) for slot in planned)
+    # run; plan slots inside so the env's agents borrow them. Everything from
+    # bringing those up to tearing them 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 env.serving():
+            planned = [slot for task, n in plan for slot in env.slots(task, n=n)]
+            slots = [RunSlot.finished(episode) for episode in finished] + planned
+            display = (
+                dashboard(slots, config, start, push=push_state)
+                if config.rich
+                else contextlib.nullcontext()
             )
-            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(
+                    env.run_slot(slot, ctx, semaphore, on_complete) 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
 
 
@@ -203,6 +238,12 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]:
         )
         write_lock = asyncio.Lock()
 
+        run = open_run(config)
+        config.run.adopt_id(run.id)
+        for episode in finished:
+            episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name))
+        run.log_episodes(finished)
+
         async def run_unit(payload: dict) -> list[Episode]:
             async with semaphore or contextlib.nullcontext():
                 episode = await client.run(
@@ -213,13 +254,20 @@ async def run_unit(payload: dict) -> list[Episode]:
                 )
             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])
             return [cast(Episode, episode)]
 
         # Each rollout is its own `run` request, dispatched least-busy across workers.
         units = [run_unit(payload) for payload, n in plan for _ in range(n)]
-        results = await asyncio.gather(*units)
-        await client.close()
-        return finished + [record for unit in results for record in unit]
+        try:
+            results = await gather_rollouts(units)
+            await client.close()
+            episodes = finished + [record for unit in results for record in unit]
+            await asyncio.to_thread(finish_run, run, episodes)
+        except BaseException as e:
+            await asyncio.to_thread(abort_run, run, e)
+            raise
+        return episodes
     finally:
         proc.terminate()
         with contextlib.suppress(Exception):

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, default: 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

@@ -40,13 +40,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()