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,28 @@ 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. Say so and keep the URL, rather than reporting a failed push and
+        # hiding the run it did produce.
         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.done:
+        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,9 @@ 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 — uploaded verbatim with the
+    # run, so its Config tab shows what someone actually wrote.
+    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 +150,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,6 +4,7 @@
 import contextlib
 import logging
 import time
+from typing import Any
 
 from verifiers.v1.cli.dashboard import dashboard
 from verifiers.v1.cli.eval import resume
@@ -18,10 +19,17 @@
 from verifiers.v1.env import Env, RunSlot
 from verifiers.v1.episode import Episode
 from verifiers.v1.trace import EvalRunInfo
+from verifiers.v1.utils.platform import PushState, abort_run, finish_run, open_run
 
 logger = logging.getLogger(__name__)
 
 
+def record_run(episode: "Episode[Any, Any, Any]", config: EvalConfig) -> None:
+    """Stamp the run onto an episode's traces — the id the platform knows it by."""
+    for trace in episode.traces:
+        trace.record_run(EvalRunInfo(id=config.run.id, name=config.run.name))
+
+
 async def run_eval(env: Env, config: EvalConfig) -> list[Episode]:
     logger.info("eval config:\n%s", config.model_dump_json(indent=2))
     taskset = env.taskset
@@ -70,39 +78,51 @@ 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:
+    # every trace is stamped with it once, at rollout time, and nothing is
+    # re-stamped or rewritten afterwards.
+    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 — otherwise the platform would hold half a run.
+    for episode in finished:
+        record_run(episode, config)
+    run.log_traces(finished)
 
     async def on_complete(episode: Episode) -> None:
-        for trace in episode.traces:
-            trace.record_run(EvalRunInfo(id=config.run.id, name=config.run.name))
+        record_run(episode, config)
         await append_episode(out, episode, write_lock)
+        # A queue put, but a bounded one: handing it to a thread keeps a full
+        # queue from stalling every other rollout (and freezing the dashboard).
+        await asyncio.to_thread(run.log_traces, [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 asyncio.gather(
+                    *(
+                        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
 
 
@@ -196,6 +216,13 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]:
             asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None
         )
         write_lock = asyncio.Lock()
+        # Same contract as the in-process runner: the run opens before the first
+        # rollout, so its id is the one every trace carries.
+        run = open_run(config)
+        config.run.adopt_id(run.id)
+        for episode in finished:
+            record_run(episode, config)
+        run.log_traces(finished)
 
         async def run_unit(payload: dict) -> list[Episode]:
             async with semaphore or contextlib.nullcontext():
@@ -205,16 +232,22 @@ async def run_unit(payload: dict) -> list[Episode]:
                     sampling=config.sampling,
                     **payload,
                 )
-            for trace in episode.traces:
-                trace.record_run(EvalRunInfo(id=config.run.id, name=config.run.name))
+            record_run(episode, config)
             await append_episode(out, episode, write_lock)
+            await asyncio.to_thread(run.log_traces, [episode])
             return [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 asyncio.gather(*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,22 @@ 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.
+
+    Only the root form counts: `--env @ env.toml` configures one block, whereas
+    `@ eval.toml` *is* the run's config, which is what gets recorded on the run —
+    so a `@` right after a flag is somebody else's file. Several root files merge
+    into one config that no single path describes, so that case records nothing
+    rather than half of it."""
+    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,36 @@ 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.
+    # 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 never reaches
+    # the platform. Private so it stays out of the saved config and its digest —
+    # two runs of the same config differ by id, and resume compares configs.
     _id: str = PrivateAttr(default_factory=lambda: str(uuid4()))
 
+    _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 — one id, minted in one place,
+        never re-stamped afterwards."""
+        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()