feat(evaluator): add AgentEvalTaskset and filesystem evidence handle

Add the AgentEvalTaskset value type (a named, validated collection of
agent-eval tasks with unique task ids) and extend LocalFilesystemEvidence
with read_bytes, list(pattern), diff, and run_verifier. run_verifier executes
against a throwaway overlay copy so stored evidence can never be mutated.

Includes the FilesystemEntry / FilesystemDiff / CommandResult value types and
example-only tests_pass / no_test_cheating metrics that score over filesystem
evidence.

Vendored mirror synced via make vendor.

Signed-off-by: Arpit Singh (SW-CLOUD) <arpsingh@nvidia.com>

Author: arpitsardhana

packages/nemo_evaluator_sdk/examples/run_agent_eval/example_metrics.py

@@ -0,0 +1,89 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Reference metrics-over-evidence for this example (not SDK API).
+
+These show how to score from the SDK's filesystem evidence handle instead of a
+stamped verifier reward:
+
+* :class:`TestsPassMetric` runs a command against ``final_state`` filesystem
+  evidence (in a throwaway overlay) and scores on exit 0.
+* :class:`NoTestCheatingMetric` diffs ``initial_state`` against ``final_state``
+  and fails if the agent touched protected (e.g. test) paths.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
+
+
+class TestsPassMetric:
+    """Score ``True`` when a verifier command exits 0 against final-state evidence."""
+
+    def __init__(
+        self,
+        command: Sequence[str],
+        *,
+        evidence_name: str = "final_state",
+        cwd: str = ".",
+        timeout_s: float = 300.0,
+    ) -> None:
+        self._command = list(command)
+        self._evidence_name = evidence_name
+        self._cwd = cwd
+        self._timeout_s = timeout_s
+
+    @property
+    def type(self) -> str:
+        return "tests_pass"
+
+    def output_spec(self) -> list[MetricOutputSpec]:
+        return [MetricOutputSpec.boolean("tests_pass")]
+
+    async def compute_scores(self, input: MetricInput) -> MetricResult:
+        passed = False
+        evidence = input.candidate.evidence
+        if evidence is not None and evidence.get(self._evidence_name) is not None:
+            handle = await evidence.filesystem(self._evidence_name)
+            result = await handle.run_verifier(self._command, cwd=self._cwd, timeout_s=self._timeout_s)
+            passed = result.ok
+        return MetricResult(outputs=[MetricOutput(name="tests_pass", value=passed)])
+
+
+class NoTestCheatingMetric:
+    """Score ``False`` when the agent modified or deleted protected paths."""
+
+    def __init__(
+        self,
+        *,
+        protected: Sequence[str] = ("tests/",),
+        change_types: Sequence[str] = ("modified", "deleted"),
+        initial_name: str = "initial_state",
+        final_name: str = "final_state",
+    ) -> None:
+        self._protected = tuple(protected)
+        self._change_types = set(change_types)
+        self._initial_name = initial_name
+        self._final_name = final_name
+
+    @property
+    def type(self) -> str:
+        return "no_test_cheating"
+
+    def output_spec(self) -> list[MetricOutputSpec]:
+        return [MetricOutputSpec.boolean("no_test_cheating")]
+
+    async def compute_scores(self, input: MetricInput) -> MetricResult:
+        evidence = input.candidate.evidence
+        clean = True
+        if evidence is not None and evidence.get(self._initial_name) and evidence.get(self._final_name):
+            initial = await evidence.filesystem(self._initial_name)
+            final = await evidence.filesystem(self._final_name)
+            diff = await initial.diff(final)
+            violations = [
+                entry for prefix in self._protected for entry in diff.changed(prefix=prefix, kinds=self._change_types)
+            ]
+            clean = not violations
+        return MetricResult(outputs=[MetricOutput(name="no_test_cheating", value=clean)])

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/tasks.py

@@ -130,6 +130,31 @@ def _validate_metric_references(self) -> AgentEvalTask:
         return self
 
 
+class AgentEvalTaskset(BaseModel):
+    """Named collection of tasks; ``id``/``name`` sugar around a task list."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    id: str = Field(description="Stable taskset identifier.")
+    name: str | None = Field(default=None, description="Human-readable taskset name.")
+    tasks: list[AgentEvalTask] = Field(description="Tasks in this set; task ids must be unique.")
+
+    @field_validator("id")
+    @classmethod
+    def _id_must_not_be_empty(cls, value: str) -> str:
+        if not value:
+            raise ValueError("taskset id must not be empty")
+        return value
+
+    @model_validator(mode="after")
+    def _task_ids_unique(self) -> AgentEvalTaskset:
+        ids = [task.id for task in self.tasks]
+        duplicates = sorted({task_id for task_id in ids if ids.count(task_id) > 1})
+        if duplicates:
+            raise ValueError(f"duplicate taskset task ids: {duplicates}")
+        return self
+
+
 class AgentEvalRunConfig(BaseModel):
     """Configuration for a standalone agent-eval run."""
 

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/values/__init__.py

@@ -10,7 +10,14 @@
     InputSchema,
 )
 from nemo_evaluator_sdk.values.datasets import DatasetInput, DatasetRows
-from nemo_evaluator_sdk.values.evidence import CandidateEvidence, EvidenceDescriptor, LocalFilesystemEvidence
+from nemo_evaluator_sdk.values.evidence import (
+    CandidateEvidence,
+    CommandResult,
+    EvidenceDescriptor,
+    FilesystemDiff,
+    FilesystemEntry,
+    LocalFilesystemEvidence,
+)
 from nemo_evaluator_sdk.values.metrics import (
     BLEU,
     F1,
@@ -98,7 +105,10 @@
     "BooleanValue",
     "CandidateEvidence",
     "CandidateOutput",
+    "CommandResult",
     "ContinuousScore",
+    "FilesystemDiff",
+    "FilesystemEntry",
     "DatasetRow",
     "DatasetRows",
     "DefaultAggregateFieldName",

packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/values/evidence.py

@@ -6,12 +6,65 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
+import shutil
+import tempfile
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 from urllib.parse import urlparse
 
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
 
+# ``LocalFilesystemEvidence.list`` (the documented evidence API) shadows the
+# builtin inside that class body; alias it for annotations there.
+_StrList = list[str]
+
+
+class FilesystemEntry(BaseModel):
+    """One path that differs between two filesystem snapshots."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    path: str
+    change_type: Literal["added", "modified", "deleted"]
+
+
+class FilesystemDiff(BaseModel):
+    """Set of paths that changed between two filesystem snapshots."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    entries: list[FilesystemEntry] = Field(default_factory=list)
+
+    def changed(
+        self,
+        *,
+        prefix: str | None = None,
+        kinds: set[str] | None = None,
+    ) -> list[FilesystemEntry]:
+        """Return entries optionally filtered by path prefix and change kind."""
+        return [
+            entry
+            for entry in self.entries
+            if (prefix is None or entry.path.startswith(prefix)) and (kinds is None or entry.change_type in kinds)
+        ]
+
+
+class CommandResult(BaseModel):
+    """Outcome of running a verifier command against filesystem evidence."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    exit_code: int
+    stdout: str = ""
+    stderr: str = ""
+    timed_out: bool = False
+
+    @property
+    def ok(self) -> bool:
+        """Whether the command exited 0 without timing out."""
+        return self.exit_code == 0 and not self.timed_out
+
 
 class LocalFilesystemEvidence:
     """Constrained local filesystem handle for metric evidence access."""
@@ -42,17 +95,95 @@ async def read_text(self, relative_path: str | Path, *, encoding: str = "utf-8")
         path = self.path(relative_path)
         return await asyncio.to_thread(path.read_text, encoding=encoding)
 
-    async def iter_paths(self, relative_path: str | Path = ".", *, recursive: bool = False) -> list[str]:
+    async def iter_paths(self, relative_path: str | Path = ".", *, recursive: bool = False) -> _StrList:
         """Return stable relative path names under the evidence root."""
         base = self.path(relative_path)
         return await asyncio.to_thread(self._iter_paths_sync, base, recursive)
 
-    def _iter_paths_sync(self, base: Path, recursive: bool) -> list[str]:
+    def _iter_paths_sync(self, base: Path, recursive: bool) -> _StrList:
         if base.is_file():
             return [base.relative_to(self._root).as_posix()]
         iterator = base.rglob("*") if recursive else base.iterdir()
         return sorted(path.relative_to(self._root).as_posix() for path in iterator)
 
+    async def read_bytes(self, relative_path: str | Path) -> bytes:
+        """Read a binary file under the evidence root."""
+        path = self.path(relative_path)
+        return await asyncio.to_thread(path.read_bytes)
+
+    async def list(self, pattern: str = "**/*") -> _StrList:
+        """Return relative posix paths of files matching ``pattern`` (files only)."""
+        return await asyncio.to_thread(self._list_sync, pattern)
+
+    def _list_sync(self, pattern: str) -> _StrList:
+        return sorted(path.relative_to(self._root).as_posix() for path in self._root.glob(pattern) if path.is_file())
+
+    async def diff(self, other: LocalFilesystemEvidence) -> FilesystemDiff:
+        """Diff this snapshot (before) against ``other`` (after) by file content hash."""
+        return await asyncio.to_thread(self._diff_sync, other)
+
+    def _diff_sync(self, other: LocalFilesystemEvidence) -> FilesystemDiff:
+        before = self._hashes()
+        after = other._hashes()
+        entries = [FilesystemEntry(path=path, change_type="added") for path in sorted(after.keys() - before.keys())]
+        entries += [FilesystemEntry(path=path, change_type="deleted") for path in sorted(before.keys() - after.keys())]
+        entries += [
+            FilesystemEntry(path=path, change_type="modified")
+            for path in sorted(before.keys() & after.keys())
+            if before[path] != after[path]
+        ]
+        return FilesystemDiff(entries=entries)
+
+    def _hashes(self) -> dict[str, str]:
+        hashes: dict[str, str] = {}
+        for rel in self._list_sync("**/*"):
+            hashes[rel] = hashlib.sha256((self._root / rel).read_bytes()).hexdigest()
+        return hashes
+
+    async def run_verifier(
+        self,
+        command: _StrList,
+        *,
+        cwd: str = ".",
+        timeout_s: float | None = None,
+    ) -> CommandResult:
+        """Run ``command`` (no shell) against a throwaway copy of the evidence.
+
+        The evidence is copied to a temp overlay so the command can never mutate
+        stored evidence (pytest caches, build artifacts, ...). ``command`` is a
+        list passed straight to exec — no shell parsing, so no injection surface.
+        """
+        overlay = Path(tempfile.mkdtemp(prefix="evidence-verify-")).resolve()
+        try:
+            workdir = (overlay / cwd).resolve()
+            if workdir != overlay and overlay not in workdir.parents:
+                raise ValueError(f"verifier cwd {cwd!r} resolves outside evidence overlay")
+            await asyncio.to_thread(shutil.copytree, self._root, overlay, dirs_exist_ok=True)
+            return await self._exec(command, workdir, timeout_s)
+        finally:
+            await asyncio.to_thread(shutil.rmtree, overlay, True)
+
+    @staticmethod
+    async def _exec(command: _StrList, cwd: Path, timeout_s: float | None) -> CommandResult:
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            cwd=str(cwd),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        try:
+            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout_s)
+        except TimeoutError:
+            # wait_for cancels communicate() but leaves the child running; kill it.
+            process.kill()
+            await process.wait()
+            return CommandResult(exit_code=-1, timed_out=True)
+        return CommandResult(
+            exit_code=process.returncode if process.returncode is not None else -1,
+            stdout=stdout.decode(errors="replace"),
+            stderr=stderr.decode(errors="replace"),
+        )
+
 
 class EvidenceDescriptor(BaseModel):
     """Descriptor for a candidate trace, source, or artifact."""

packages/nemo_evaluator_sdk/tests/agent_eval/test_evidence.py

@@ -5,7 +5,11 @@
 
 import pytest
 from nemo_evaluator_sdk.execution.samples import build_metric_input
-from nemo_evaluator_sdk.values.evidence import CandidateEvidence, EvidenceDescriptor
+from nemo_evaluator_sdk.values.evidence import (
+    CandidateEvidence,
+    EvidenceDescriptor,
+    LocalFilesystemEvidence,
+)
 
 
 def test_metric_input_preserves_candidate_evidence_out_of_metadata() -> None:
@@ -53,3 +57,52 @@ async def test_candidate_evidence_filesystem_access_is_lazy_and_cached(tmp_path:
         handle.path("../outside")
     with pytest.raises(ValueError, match="only supports local filesystem"):
         await evidence.filesystem("remote_state")
+
+
+@pytest.mark.asyncio
+async def test_filesystem_read_bytes_list_and_diff(tmp_path: Path) -> None:
+    before = tmp_path / "before"
+    after = tmp_path / "after"
+    for root in (before, after):
+        (root / "src").mkdir(parents=True)
+    (before / "keep.txt").write_text("same", encoding="utf-8")
+    (before / "src" / "mod.py").write_text("old", encoding="utf-8")
+    (before / "gone.txt").write_text("bye", encoding="utf-8")
+    (after / "keep.txt").write_text("same", encoding="utf-8")
+    (after / "src" / "mod.py").write_text("new", encoding="utf-8")
+    (after / "added.txt").write_text("hi", encoding="utf-8")
+
+    initial = LocalFilesystemEvidence(before)
+    final = LocalFilesystemEvidence(after)
+
+    assert await final.read_bytes("added.txt") == b"hi"
+    assert await final.list("**/*.py") == ["src/mod.py"]
+
+    diff = await initial.diff(final)
+    assert {(entry.path, entry.change_type) for entry in diff.entries} == {
+        ("added.txt", "added"),
+        ("gone.txt", "deleted"),
+        ("src/mod.py", "modified"),
+    }
+    assert [entry.path for entry in diff.changed(prefix="src/", kinds={"modified"})] == ["src/mod.py"]
+
+
+@pytest.mark.asyncio
+async def test_run_verifier_uses_overlay_and_reports_outcome(tmp_path: Path) -> None:
+    root = tmp_path / "workspace"
+    root.mkdir()
+    (root / "answer.txt").write_text("42", encoding="utf-8")
+    handle = LocalFilesystemEvidence(root)
+
+    ok = await handle.run_verifier(["cat", "answer.txt"])
+    assert ok.ok and ok.exit_code == 0 and ok.stdout.strip() == "42"
+
+    failed = await handle.run_verifier(["false"])
+    assert not failed.ok and failed.exit_code != 0
+
+    timed_out = await handle.run_verifier(["sleep", "5"], timeout_s=0.2)
+    assert timed_out.timed_out and not timed_out.ok
+
+    # The verifier ran in a throwaway copy, so the stored evidence is untouched.
+    await handle.run_verifier(["sh", "-c", "echo cheat > sneaked.txt"])
+    assert await handle.list("**/*") == ["answer.txt"]