feat(evaluator): expose per-task attempts with trial identity

docs/evaluator/agent-eval/reading-results.mdx

@@ -37,6 +37,52 @@ result = await AgentEvaluator().run(tasks=..., target=...)
 
 - **`summary.metric_coverage`** — per metric output, how many trials were `total` / `scored` / failed /
   missing, so you can tell a low mean from low coverage.
+- **`summary.task_metric_values`** — per task, the individual trial values behind those means, keyed
+  `<metric.type>.<output>`. Each record carries the `trial_id` that produced it and its metric `value`, so
+  you can answer "which tasks were flaky, and on which trial?" without regrouping `result.scores`
+  yourself:
+
+  ```python
+  for task_id, by_output in result.summary.task_metric_values.items():
+      # .get: keys are per task, so a task scored by a different metric simply has none.
+      print(task_id, [(a.trial_id, a.value) for a in by_output.get("reward.score", [])])
+  ```
+
+  A `value` of `None` is a trial that died before scoring — a trial that did not pass. A trial
+  whose *metric* failed is absent entirely, because that leaves it unmeasured rather than
+  unsuccessful. Look up by `trial_id` rather than by position: the two rules above mean lists for
+  different outputs of one task need not be the same length.
+
+  Values keep the type the metric produced them in — a count stays an `int`, a flag stays a `bool`,
+  and a judge's verdict stays a `str`. Each record's `value_type` says which it is (`number`,
+  `label` or `missing`), which is what tells a real `NaN` apart from a label that reads `"NaN"`, since
+  strict JSON has no NaN literal and both travel as strings. Before doing arithmetic, project with
+  `numeric_metric_values`, which drops labels and keeps a dead trial's `None`:
+
+  ```python
+  from nemo_evaluator_sdk.agent_eval.results import numeric_metric_values
+
+  records = result.summary.task_metric_values["task-47"]["reward.score"]
+  scores = numeric_metric_values(records)   # [1.0, None, 0.0]
+  ```
+
+- **`summary.task_outcomes(metric_name=None)`** — the same data as models rather than nested dicts,
+  sorted by task then metric, each naming its own `task_id` and `metric_name`. Pass a
+  `"<metric.type>.<output>"` to narrow to one metric, which is what a report over a single metric
+  wants:
+
+  ```python
+  for per_task in result.summary.task_outcomes("reward.score"):
+      for outcome in per_task.outcomes:
+          values = numeric_metric_values(outcome.trials)
+          print(per_task.task_id, outcome.metric_name, values)
+  ```
+
+  - When you narrow, a task the metric never measured is **dropped** — it was scored by a different
+    metric, so listing it would invent missing coverage.
+  - A task that declared the metric but produced no usable value keeps its entry with an empty
+    `trials` list, because there the coverage really is missing.
+  - Unfiltered, every task is returned.
 - **`summary.task_count`**, **`summary.trial_count`**, **`summary.score_count`**.
 
 ### Per-metric scores

packages/nemo_evaluator_sdk/examples/gym/README.md

@@ -57,7 +57,20 @@ Run bundle (run.json, trials.jsonl, scores.jsonl, report.html): /var/folders/...
 
 ## Read the results
 
-`inspect_results.py` is the companion to the above: it reads a bundle and shows how to reach each kind of result — headline aggregates, `pass@k`, per-task outcomes, and the runner's own imported numbers. Its accessors (`aggregate`, `per_task_outcomes`) are written to be lifted into your own code, and everything it shows also works on the in-memory `AgentEvalResult` that `AgentEvaluator().run(...)` returns — reading a bundle just makes it runnable without a live run.
+`inspect_results.py` reads `summary.json` and shows each result layer: run aggregates from
+`summary.scores`, ordered per-task values from `summary.task_outcomes("<metric_type>.<output>")`, and
+runner-owned aggregates under `runner.gym.*`. That accessor returns models rather than nested dicts —
+each row names its own `task_id` and `metric_name` — so the example needs no per-task accessor of its
+own. A `null` value is a trial that failed before scoring, while an empty `trials` list means the
+metric produced no usable measurement; a task the metric never measured is not returned at all.
+
+Each record names the trial that produced it, so `trial_id` — not list position — is what joins two
+outputs of the same task, or looks up `trials.jsonl`. A trial whose metric failed is absent
+rather than `null`, so two lists for one task need not be the same length.
+
+Values keep the metric's own type (a count stays an int, a judge's verdict stays a label), and each
+carries a `value_type` of `number` / `label` / `missing`. Use `numeric_metric_values` before doing
+arithmetic — it drops labels and keeps a dead trial's `null`.
 
 No bundle is checked in; the run above produces one. Give it a stable `--output-dir` and point the reader at the same path:
 

packages/nemo_evaluator_sdk/examples/gym/inspect_results.py

@@ -7,10 +7,11 @@
 how to get at each kind of result — headline aggregates, ``pass@k``, per-task outcomes, and the
 runner's own aggregations.
 
-The helpers below (:func:`aggregate`, :func:`per_task_outcomes`) are written to be lifted directly
-into your own code. Everything shown here also works on the in-memory ``AgentEvalResult`` returned by
-``AgentEvaluator().run(...)`` — reading from a bundle just makes the example runnable without a live
-run.
+Per-task results are read through the SDK's own typed view, ``summary.task_outcomes(metric_name)``,
+rather than by walking the nested ``summary.task_metric_values`` dict here — that is the accessor to
+lift into your own code. Everything shown here also works on the in-memory ``AgentEvalResult``
+returned by ``AgentEvaluator().run(...)`` — reading from a bundle just makes the example runnable
+without a live run.
 
 There is no bundle checked into the repo — ``run_gym_eval.py`` makes one. It writes to a fresh
 temporary directory by default, so give it an explicit ``--output-dir`` and point this script at the
@@ -29,19 +30,27 @@
 
 import argparse
 import json
-from collections.abc import Sequence
 from pathlib import Path
 
-from nemo_evaluator_sdk.agent_eval.results import AgentEvalSummary
-from nemo_evaluator_sdk.agent_eval.scores import AgentEvalScoreStatus, AgentEvalTaskScore, is_trial_failure
+from nemo_evaluator_sdk.agent_eval.results import (
+    AgentEvalSummary,
+    PerTaskOutcomes,
+    numeric_metric_values,
+)
 from nemo_evaluator_sdk.values.results import AggregateScalarScore, AggregateScore
+from pydantic import ValidationError
 
-#: Value at which an attempt counts as a pass, matching the SDK's pass@k definition (full credit).
+#: Value at which a trial counts as a pass, matching the SDK's pass@k definition (full credit).
 PASS_VALUE = 1.0
 
 #: Namespace the Gym runner's own aggregations are imported under, so they never collide with ours.
 RUNNER_PREFIX = "runner.gym."
 
+
+class BundleFormatError(Exception):
+    """A run bundle this script cannot read (wrong directory, or written by an older SDK)."""
+
+
 # --------------------------------------------------------------------------------------------------
 # Accessors — lift these into your own code.
 # --------------------------------------------------------------------------------------------------
@@ -60,55 +69,43 @@ def aggregate(summary: AgentEvalSummary, name: str) -> AggregateScore:
     raise KeyError(f"no aggregate named {name!r}; available: {available}")
 
 
-def per_task_outcomes(
-    scores: Sequence[AgentEvalTaskScore],
-    *,
-    metric_type: str,
-    output_name: str,
-) -> dict[str, list[float | None]]:
-    """Group per-trial score values by task: ``task_id -> [value per attempt]``, ``None`` if it died.
-
-    A run with ``num_repeats=R`` produces R trials per task, and the scores are a flat
-    task x trial x metric list — so answering "which tasks failed?" means grouping them yourself.
-
-    Failed trials are kept, as ``None``. Dropping them would show a task that passed once and crashed
-    once as solved, and disagrees with how the SDK computes pass@k (a dead rollout is an attempt that
-    did not pass). A failed *metric* is dropped instead: it leaves the attempt unmeasured rather than
-    unsuccessful, so counting it against the agent would turn a judge timeout into a failure.
-    """
-    by_task: dict[str, list[float | None]] = {}
-    for score in scores:
-        if score.metric_type != metric_type:
-            continue
-        if is_trial_failure(score):
-            by_task.setdefault(score.task_id, []).append(None)
-            continue
-        if score.status == AgentEvalScoreStatus.FAILED:
-            continue
-        for output in score.outputs:
-            if output.name == output_name and isinstance(output.value, int | float):
-                by_task.setdefault(score.task_id, []).append(float(output.value))
-    return by_task
-
-
 # --------------------------------------------------------------------------------------------------
 # Bundle loading (see the run.json manifest for the full artifact list).
 # --------------------------------------------------------------------------------------------------
 
 
-def load_bundle(bundle: Path) -> tuple[AgentEvalSummary, list[AgentEvalTaskScore]]:
-    """Hydrate the pieces of a persisted run bundle used below.
+def load_bundle(bundle: Path) -> AgentEvalSummary:
+    """Load the persisted summary, including native and runner aggregates and per-task values.
 
-    A runner's own numbers need no separate file: they are imported into ``summary.scores`` under
-    ``runner.<name>.``, so one load covers both.
+    Every way a bundle can be unreadable raises :class:`BundleFormatError` and nothing else, so
+    :func:`main` turns all of them into an exit code rather than a traceback. In particular it
+    rejects a bundle written before ``task_metric_values`` existed rather than reading one: the
+    field defaults to empty, so an older bundle would otherwise load cleanly and simply show no
+    per-task section — the reader would conclude the run had no per-task outcomes rather than that
+    this script cannot see them.
     """
-    summary = AgentEvalSummary.model_validate(json.loads((bundle / "summary.json").read_text(encoding="utf-8")))
-    scores = [
-        AgentEvalTaskScore.model_validate(json.loads(line))
-        for line in (bundle / "scores.jsonl").read_text(encoding="utf-8").splitlines()
-        if line.strip()
-    ]
-    return summary, scores
+    summary_path = bundle / "summary.json"
+    if not summary_path.exists():
+        raise BundleFormatError(f"{bundle} is not a run bundle (no summary.json). Run run_gym_eval.py first.")
+    try:
+        payload = json.loads(summary_path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise BundleFormatError(f"{summary_path} is not readable JSON: {exc}") from exc
+    # Before the membership test below, which raises TypeError on a root that is not a container and
+    # matches a substring on a bare JSON string.
+    if not isinstance(payload, dict):
+        raise BundleFormatError(f"{summary_path} is not a JSON object (found {type(payload).__name__}).")
+    # Checked explicitly because `model_validate` would *not* catch this: the field defaults to an
+    # empty dict, so an older bundle loads cleanly and simply shows no per-task section.
+    if "task_metric_values" not in payload:
+        raise BundleFormatError(
+            f"{summary_path} predates summary.task_metric_values, which this script reads "
+            "per-task outcomes from. Re-run the eval to produce a current bundle."
+        )
+    try:
+        return AgentEvalSummary.model_validate(payload)
+    except ValidationError as exc:
+        raise BundleFormatError(f"{summary_path} is not a valid run summary: {exc}") from exc
 
 
 # --------------------------------------------------------------------------------------------------
@@ -138,29 +135,42 @@ def show_aggregates(summary: AgentEvalSummary) -> None:
     print(f"\n  {summary.task_count} tasks · {summary.trial_count} trials · {summary.score_count} scores")
 
 
-def show_per_task(by_task: dict[str, list[float | None]]) -> None:
+def show_per_task(outcomes: list[PerTaskOutcomes]) -> None:
     """Per-task outcomes: which tasks were solved, and how consistently.
 
-    An attempt passes on full credit (``>= PASS_VALUE``), matching how the SDK computes pass@k. A
-    ``None`` is a trial that died: it counts as an attempt and never as a pass, so a task that passed
+    Takes the SDK's typed view, so every row already knows its own task and metric and no dict has
+    to be re-keyed here. Already sorted by task, hence no ``sorted()``.
+
+    A trial passes on full credit (``>= PASS_VALUE``), matching how the SDK computes pass@k. A
+    ``None`` is a trial that died: it counts toward ``n`` and never as a pass, so a task that passed
     once and crashed once reads as flaky rather than solved.
     """
-    print("\nPer-task outcomes (attempt values; an attempt passes at full credit)")
-    solved = flaky = failed = 0
-    for task_id, values in sorted(by_task.items()):
-        passes = sum(1 for value in values if value is not None and value >= PASS_VALUE)
-        if passes == len(values):
-            verdict, marker = "solved", "+"
-            solved += 1
-        elif passes:
-            verdict, marker = f"flaky ({passes}/{len(values)})", "~"
-            flaky += 1
-        else:
-            verdict, marker = "failed", "-"
-            failed += 1
-        attempts = ", ".join("died" if value is None else f"{value:g}" for value in values)
-        print(f"  {marker} {task_id[:16]}…  [{attempts}]  {verdict}")
-    print(f"\n  {solved} solved · {flaky} flaky · {failed} failed")
+    print("\nPer-task outcomes (trial values; a trial passes at full credit)")
+    solved = flaky = failed = unmeasured = 0
+    for per_task in outcomes:
+        for outcome in per_task.outcomes:
+            # Projected to floats only here, where the work is genuinely arithmetic: `>=` and `:g`
+            # both raise on a judge's label, and numeric_metric_values drops those while keeping a
+            # dead trial's None. Everything above reads the records as they were recorded.
+            values = numeric_metric_values(outcome.trials)
+            if not values:
+                verdict, marker = "unmeasured", "?"
+                unmeasured += 1
+                shown = ""
+            else:
+                passes = sum(1 for value in values if value is not None and value >= PASS_VALUE)
+                if passes == len(values):
+                    verdict, marker = "solved", "+"
+                    solved += 1
+                elif passes:
+                    verdict, marker = f"flaky ({passes}/{len(values)})", "~"
+                    flaky += 1
+                else:
+                    verdict, marker = "failed", "-"
+                    failed += 1
+                shown = ", ".join("died" if value is None else f"{value:g}" for value in values)
+            print(f"  {marker} {per_task.task_id[:16]}…  [{shown}]  {verdict}")
+    print(f"\n  {solved} solved · {flaky} flaky · {failed} failed · {unmeasured} unmeasured")
 
 
 def show_runner_aggregations(summary: AgentEvalSummary) -> None:
@@ -203,15 +213,17 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
 
 def main(argv: list[str] | None = None) -> int:
     args = _parse_args(argv)
-    if not (args.bundle / "summary.json").exists():
-        raise SystemExit(f"{args.bundle} is not a run bundle (no summary.json). Run run_gym_eval.py first.")
-
-    summary, scores = load_bundle(args.bundle)
+    # The CLI boundary is where a bad bundle becomes an exit code; the accessors above just raise.
+    try:
+        summary = load_bundle(args.bundle)
+    except BundleFormatError as exc:
+        raise SystemExit(str(exc)) from exc
 
     show_aggregates(summary)
-    by_task = per_task_outcomes(scores, metric_type=args.metric_type, output_name=args.output_name)
-    if by_task:
-        show_per_task(by_task)
+    # Empty when no task was measured by this metric at all -- typically a wrong --metric-type.
+    outcomes = summary.task_outcomes(f"{args.metric_type}.{args.output_name}")
+    if outcomes:
+        show_per_task(outcomes)
     show_runner_aggregations(summary)
 
     print(f"\nFull report: {args.bundle / 'report.html'}")

packages/nemo_evaluator_sdk/examples/gym/run_gym_eval.py

@@ -46,6 +46,14 @@
 from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig
 
 
+def _non_negative_int(raw: str) -> int:
+    """argparse type for ``--limit``: a negative value would silently slice tasks off the END."""
+    value = int(raw)
+    if value < 0:
+        raise argparse.ArgumentTypeError(f"must be non-negative, got {value}")
+    return value
+
+
 def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
     parser.add_argument(
@@ -68,6 +76,9 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         help="`inference_provider` speaks OpenAI-compatible chat; `openai_model` uses the Responses API.",
     )
     parser.add_argument("--num-repeats", type=int, default=2, help="Attempts per task (each becomes a trial).")
+    parser.add_argument(
+        "--limit", type=_non_negative_int, default=None, help="Run only the first N tasks (handy for smoke runs)."
+    )
     parser.add_argument(
         "--output-dir",
         type=Path,
@@ -115,6 +126,10 @@ async def _main(args: argparse.Namespace) -> int:
     dataset = args.dataset or _packaged_dataset(args.resources_server)
     tasks = discover_gym_tasks(dataset)
     print(f"discovered {len(tasks)} tasks from {dataset}")
+    if args.limit is not None:
+        # The runner materializes Gym's input from these tasks, so this bounds the rollout too.
+        tasks = tasks[: args.limit]
+        print(f"limited to {len(tasks)} task(s)")
 
     runner = GymAgentTaskRunner(
         config=GymRuntimeConfig(

packages/nemo_evaluator_sdk/examples/harbor/run_harbor_example.py

@@ -21,6 +21,7 @@
 Run it as a module from the repository root::
 
     python -m packages.nemo_evaluator_sdk.examples.harbor.run_harbor_example --mode native
+    python -m packages.nemo_evaluator_sdk.examples.harbor.run_harbor_example --mode native --n-attempts 2
     python -m packages.nemo_evaluator_sdk.examples.harbor.run_harbor_example --mode optimizer
 """
 
@@ -45,9 +46,17 @@
 HELLO_WORLD_DATASET_DIR = Path(__file__).resolve().parent / "hello_world_dataset"
 
 
-async def _main(mode: str, jobs_dir: Path) -> None:
+async def _main(mode: str, jobs_dir: Path, *, n_attempts: int, job_name: str | None) -> None:
     # The entire caller-side plumbing: a config and one call.
-    config = HarborRuntimeConfig(jobs_dir=jobs_dir, agent_name="oracle")
+    # n_attempts>1 runs the same verifier criteria per attempt so summary can emit pass@k.
+    config = HarborRuntimeConfig(
+        jobs_dir=jobs_dir,
+        job_name=job_name,
+        agent_name="oracle",
+        n_attempts=n_attempts,
+        n_concurrent_trials=1,
+        quiet=False,
+    )
     result = await run_harbor_eval(config, HELLO_WORLD_DATASET_DIR)
 
     if mode == "optimizer":
@@ -78,5 +87,16 @@ async def _main(mode: str, jobs_dir: Path) -> None:
         default=Path(__file__).resolve().parent / "harbor-example-output",
         help="Directory Harbor writes its job results into.",
     )
+    parser.add_argument(
+        "--n-attempts",
+        type=int,
+        default=1,
+        help="Harbor trials per task (same verifier criteria each attempt; enables pass@k when >1).",
+    )
+    parser.add_argument(
+        "--job-name",
+        default=None,
+        help="Pin a stable Harbor job name to reuse the job-dir cache across debug runs.",
+    )
     args = parser.parse_args()
-    asyncio.run(_main(args.mode, args.jobs_dir))
+    asyncio.run(_main(args.mode, args.jobs_dir, n_attempts=args.n_attempts, job_name=args.job_name))