feat(evaluator): let evaluations pin a taskset revision

docs/evaluator/manage-tasks-tasksets.mdx

@@ -290,7 +290,10 @@ input_spec = AgentEvalInputSpec(
 
 When the job runs, the taskset reference is resolved like this:
 
-- Each member is loaded at the **revision pinned in the taskset**, not the task's current tip.
+- The **taskset revision** the ref names is loaded — the current one unless the ref pins a revision
+  (see [Pin the taskset itself](#pin-the-taskset-itself)).
+- Each member of that revision is loaded at the **revision pinned in the taskset**, not the task's
+  current tip.
 - Metric references on those members are hydrated into runnable metrics, the same as for inline tasks.
 - Re-running the same taskset therefore evaluates the same content, even if a member has been
   republished since.
@@ -320,11 +323,42 @@ results flow.
 The inline form remains available for one-off tasks — swap `tasks=TasksetRef(...)` for
 `tasks=[AgentEvalTaskInput(...), ...]`.
 
-<Note>
-A `TasksetRef` names the taskset's current revision; it cannot yet carry a `#<tag-or-digest>`
-fragment of its own. Member content is pinned, so a re-run always grades the same task content — but
-if the taskset itself is replaced, a re-submitted spec expands the new membership.
-</Note>
+### Pin the taskset itself
+
+A bare `TasksetRef` expands the taskset's current revision, so it follows the suite forward every
+time the taskset is republished. Add a `#<tag-or-digest>` fragment to pin the grouping too:
+
+```python
+# The digest lives on the revision, not on the taskset record — revisions come back newest first.
+current = tasksets.list_revisions("geography-suite").data[0]
+
+# Follows the suite forward — the next `replace` is picked up on the next run.
+tracking = TasksetRef("default/geography-suite")
+
+# Frozen: this exact membership and content, regardless of later `replace` calls.
+pinned = TasksetRef(f"default/geography-suite#{current.content_hash}")
+
+# A tag works the same way, and can be moved deliberately when you bless a new suite.
+blessed = TasksetRef("default/geography-suite#blessed")
+```
+
+What each form is stable against:
+
+| | A member task republishes | The taskset is replaced |
+|---|---|---|
+| `TasksetRef("suite")` | unaffected | follows the new revision |
+| `TasksetRef("suite#<digest>")` | unaffected | unaffected |
+
+Every taskset revision pins its members by digest, so **neither** form is disturbed when a member
+task publishes new content on its own — that is the guarantee stored membership buys you.
+
+The two differ on `replace`. A bare ref tracks the taskset's own revisions, and members are
+re-resolved on every write — so a `replace` can change *both* which tasks are named *and* the
+content they resolve to, even when the submitted member names were identical. Pin the taskset when a
+benchmark number has to stay comparable across that.
+
+A fragment that no longer resolves fails the evaluation rather than falling back to the current
+revision.
 
 <Note>
 Stored tasks carry no grader-only `reference` (held-out ground truth): that field lives only on inline
@@ -401,6 +435,7 @@ The SDK resources are a thin client over the Evaluator plugin REST API, mounted
 | Missing or duplicate task reference (taskset) | `422` | Members must exist, and must resolve to distinct tasks. |
 | Reserved or malformed tag name | `422` | `latest` cannot be moved by hand; a digest-shaped tag is refused. |
 | Retrieving or deleting an unknown name or revision | `404` | Applies to both records and revisions. |
+| `TasksetRef` pinning a revision that no longer resolves | job fails | Expansion refuses rather than falling back to the current revision. |
 | `DELETE` on a task a taskset pins | `204` | Not prevented; the taskset's reference dangles and fails on read. |
 
 ## Related Topics

plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py

@@ -151,9 +151,9 @@ class MetricInline(BaseModel):
 # (``#latest``, ``#candidate``) or a full 64-char content digest.
 #
 # Deliberately a sibling of ``_ENTITY_REF_PATTERN`` rather than a widening of it: that constant is
-# shared by ``MetricRef`` and ``TasksetRef``, neither of which has revisions yet, and admitting a
-# fragment there would accept input nothing is built to resolve. ``TasksetRef`` moves onto this
-# pattern when taskset revisions are addressable; ``MetricRef`` when (if) metrics gain revisions.
+# still shared by ``MetricRef``, which has no revisions, and admitting a fragment there would accept
+# input nothing is built to resolve. ``TaskRef`` and ``TasksetRef`` both use this pattern, since both
+# name revisioned records; ``MetricRef`` joins them when (if) metrics gain revisions.
 _SUBENTITY_REF_PATTERN = rf"^[\w\-.]+(/[\w\-.]+)?(#{REF_FRAGMENT_CHARSET})?$"
 
 #: The fragment separator for sub-entity references. Matches the fileset/job ref convention.
@@ -229,15 +229,26 @@ class TaskRef(RootModel[str]):
 
 
 class TasksetRef(RootModel[str]):
-    """Reference to a persisted taskset (format: ``workspace/name`` or ``name``).
+    """Reference to a persisted taskset (format: ``workspace/name`` or ``name``, optionally ``#rev``).
 
     Same shape and charset as :class:`TaskRef`. Lets an evaluation reference a stored taskset in place
     of an inline task list; the taskset's member tasks are loaded and expanded during spec resolution.
+
+    An optional ``#`` fragment pins the taskset revision to expand — a tag or a full content digest,
+    with an absent fragment meaning ``latest``.
+
+    What each form guarantees, precisely. A taskset revision pins its members by digest, so a member
+    task publishing new content never changes what *any* ref expands to. A **bare** ref still tracks
+    the taskset's own revisions, and republishing the taskset re-resolves its members on write — so a
+    ``replace`` can change both which members are named and the content they resolve to, even if the
+    submitted member names were identical. A **pinned** ref is fixed against that too, and is what an
+    evaluation needs to stay comparable across a ``replace``.
     """
 
     root: str = Field(
-        pattern=_ENTITY_REF_PATTERN,
-        description="Reference to a stored taskset (format: workspace/taskset-name, or taskset-name in the job workspace).",
+        pattern=_SUBENTITY_REF_PATTERN,
+        description="Reference to a stored taskset (format: workspace/taskset-name, or taskset-name in the "
+        "job workspace), optionally pinned to a revision with '#<tag-or-digest>'.",
     )
 
 

plugins/nemo-evaluator/src/nemo_evaluator/task_refs.py

@@ -18,8 +18,8 @@
 
 from typing import cast
 
-from nemo_evaluator.api.schemas import TasksetRef, parse_entity_ref, parse_subentity_ref
-from nemo_evaluator.entities import TaskEntity, TaskRevisionEntity, TasksetEntity
+from nemo_evaluator.api.schemas import TasksetRef, parse_subentity_ref
+from nemo_evaluator.entities import TaskEntity, TaskRevisionEntity, TasksetEntity, TasksetRevisionEntity
 from nemo_evaluator.jobs.agent_spec import AgentEvalTaskInput
 from nemo_evaluator.revisions import RevisionNotFoundError, get_revision
 from nemo_platform_plugin.entities import EntityClientProtocol
@@ -49,10 +49,11 @@ def _entity_to_task_input(entity: TaskEntity, revision: TaskRevisionEntity) -> A
     )
 
 
-#: Expanding a taskset reads three entity types through one client — the taskset head, each member
-#: task's head, and the pinned revision of each member. Python has no intersection types, so the
-#: parameter is annotated at one of them and the other two are taken as typed views of the same
-#: object; the concrete client's methods are generic over the entity type and satisfy all three.
+#: Expanding a taskset reads four entity types through one client — the taskset head, its pinned
+#: revision, each member task's head, and the pinned revision of each member. Python has no
+#: intersection types, so the parameter is annotated at one of them and the rest are taken as typed
+#: views of the same object; the concrete client's methods are generic over the entity type and
+#: satisfy all four.
 TasksetStoreProtocol = EntityClientProtocol[TasksetEntity]
 
 
@@ -66,6 +67,12 @@ async def resolve_taskset_ref(
 
     Loading needs only the entity store (metrics stay as refs, resolved downstream), so unlike
     metric-ref resolution this does not require an async SDK / file I/O.
+
+    The ref may pin a taskset revision (``suite#<tag-or-digest>``); an absent fragment means
+    ``latest``. Both paths go through :func:`get_revision` rather than reading the head's own
+    ``tasks``, because a head and its ``latest`` revision are guaranteed to agree and resolving one
+    way for pinned refs and another way for bare ones would make the two drift apart on the next
+    bug. It also buys content verification for the bare case for free.
     """
     if entity_client is None:
         raise ValueError(
@@ -74,8 +81,9 @@ async def resolve_taskset_ref(
         )
     task_store = cast(EntityClientProtocol[TaskEntity], entity_client)
     revision_store = cast(EntityClientProtocol[TaskRevisionEntity], entity_client)
+    taskset_revision_store = cast(EntityClientProtocol[TasksetRevisionEntity], entity_client)
 
-    ref_workspace, name = parse_entity_ref(ref.root, workspace)
+    ref_workspace, name, taskset_fragment = parse_subentity_ref(ref.root, workspace)
     try:
         taskset = await entity_client.get(TasksetEntity, name=name, workspace=ref_workspace)
     except NemoEntityNotFoundError as exc:
@@ -85,12 +93,21 @@ async def resolve_taskset_ref(
             "or pass an inline task list instead."
         ) from exc
 
-    if not taskset.tasks:
+    # Expand the taskset revision the ref names. Members are digest-pinned inside a revision, so a
+    # member republishing on its own never moves this. A bare ref still follows the taskset's own
+    # revisions, and a ``replace`` re-resolves members on write — so it can change both which members
+    # are named and what they resolve to. Only a pinned ref holds both steady.
+    try:
+        taskset_revision = await get_revision(taskset_revision_store, TasksetRevisionEntity, taskset, taskset_fragment)
+    except RevisionNotFoundError as exc:
+        raise ValueError(f"Taskset reference '{ref.root}' names a revision that does not resolve: {exc}") from exc
+
+    if not taskset_revision.tasks:
         raise ValueError(f"Taskset '{ref.root}' has no member tasks; an agent evaluation needs at least one task.")
 
     tasks: list[AgentEvalTaskInput] = []
     seen_ids: set[str] = set()
-    for task_ref in taskset.tasks:
+    for task_ref in taskset_revision.tasks:
         task_workspace, task_name, fragment = parse_subentity_ref(task_ref.root, ref_workspace)
         try:
             entity = await task_store.get(TaskEntity, name=task_name, workspace=task_workspace)

plugins/nemo-evaluator/tests/api/service/test_task_service.py

@@ -250,8 +250,8 @@ async def _delete_also_fails(*args, **kwargs):
         await service.create_task("task-1", _task_input(), workspace="default")
 
 
-async def test_resolve_revision_returns_the_digest_for_a_tag(service: TaskService) -> None:
-    """The hook taskset publishing uses to turn a member's tag into an exact digest."""
+async def test_resolve_revision_defaults_to_the_current_revision(service: TaskService) -> None:
+    """No fragment means ``latest`` — the bare-member case taskset publishing hits most often."""
     await service.create_task("task-1", _task_input(), workspace="default")
 
     digest = await service.resolve_revision("default", "task-1")
@@ -261,6 +261,40 @@ async def test_resolve_revision_returns_the_digest_for_a_tag(service: TaskServic
     assert digest == revisions.data[0].content_hash
 
 
+async def test_resolve_revision_honours_a_tag_naming_an_older_revision(service: TaskService) -> None:
+    """The hook taskset publishing uses to turn a member's tag into an exact digest.
+
+    Needs a task with *more than one* revision and a tag left behind on the older one: against a
+    single-revision task every fragment resolves to the same digest, so the test would pass even if
+    the fragment were ignored entirely.
+    """
+    await service.create_task("task-1", _task_input(), workspace="default")
+    first = (await service.list_revisions("default", "task-1")).data[0].content_hash
+    await service.tag_revision("default", "task-1", "blessed", "latest")
+
+    revised = _task_input()
+    revised.intent = "Answer differently."
+    await service.replace_task("task-1", revised, workspace="default")
+
+    latest = await service.resolve_revision("default", "task-1")
+    blessed = await service.resolve_revision("default", "task-1", "blessed")
+
+    assert latest != first, "the task must actually have moved on, or this proves nothing"
+    assert blessed == first, "a tag must resolve to the revision it names, not the current one"
+
+
+async def test_resolve_revision_round_trips_a_digest_fragment(service: TaskService) -> None:
+    """A member submitted already-pinned must resolve to itself rather than to the head."""
+    await service.create_task("task-1", _task_input(), workspace="default")
+    first = (await service.list_revisions("default", "task-1")).data[0].content_hash
+
+    revised = _task_input()
+    revised.intent = "Answer differently."
+    await service.replace_task("task-1", revised, workspace="default")
+
+    assert await service.resolve_revision("default", "task-1", first) == first
+
+
 async def test_resolve_revision_raises_for_a_missing_task(service: TaskService) -> None:
     """Existence surfaces from resolution itself — taskset publishing relies on this to reject a
     member that does not exist, now that the separate existence check is gone."""

plugins/nemo-evaluator/tests/api/service/test_taskset_service.py

@@ -30,12 +30,19 @@ async def get_task(self, workspace: str, name: str) -> object | None:
     async def resolve_revision(self, workspace: str, name: str, fragment: str = "latest") -> str:
         """A stable per-task digest, so pinned membership is deterministic across a test.
 
-        Raises for an unknown task, matching the real service: resolution fetches the task, so a
-        missing one surfaces here rather than from a separate existence check.
+        The digest covers the **fragment** as well as the task, because the real
+        ``TaskService.resolve_revision`` resolves it: a bare member and a tag-pinned one give
+        different digests whenever the tag names an older revision. A fake that ignored the fragment
+        would return the same digest either way, and a publish path that dropped the fragment
+        entirely would look correct in every assertion here.
         """
         if (workspace, name) not in self.existing:
             raise NemoEntityNotFoundError(f"{workspace}/{name} not found")
-        return hashlib.sha256(f"{workspace}/{name}".encode()).hexdigest()
+        return hashlib.sha256(f"{workspace}/{name}#{fragment}".encode()).hexdigest()
+
+    def digest_for(self, workspace: str, name: str, fragment: str = "latest") -> str:
+        """The digest a test should expect for a member — same derivation as ``resolve_revision``."""
+        return hashlib.sha256(f"{workspace}/{name}#{fragment}".encode()).hexdigest()
 
 
 def _taskset_input() -> TasksetInput:
@@ -73,6 +80,31 @@ async def test_create_then_get(service: TasksetService) -> None:
     assert got is not None and got.name == "ts-1"
 
 
+async def test_a_tag_pinned_member_stores_the_tagged_revision(
+    existing_tasks: set[tuple[str, str]], entity_store
+) -> None:
+    """A member's ``#tag`` must be resolved through the tag, not silently treated as the head.
+
+    This is the whole point of resolving membership on write: the stored ref has to name the
+    revision the tag pointed at *then*, so moving the tag afterwards cannot re-point a published
+    taskset. Dropping the fragment would still produce a digest-shaped ref, so only comparing
+    against the tag-specific digest catches it.
+    """
+    task_service = _FakeTaskService(existing_tasks)
+    service = TasksetService(entity_store, task_service)
+
+    created, _ = await service.create_taskset(
+        "ts-1", TasksetInput(tasks=[TaskRef("task-a#blessed"), TaskRef("task-b")]), workspace="default"
+    )
+
+    stored = {t.root.split("#")[0]: t.root.split("#")[1] for t in created.tasks}
+    assert stored["default/task-a"] == task_service.digest_for("default", "task-a", "blessed")
+    assert stored["default/task-b"] == task_service.digest_for("default", "task-b", "latest")
+    assert stored["default/task-a"] != task_service.digest_for("default", "task-a", "latest"), (
+        "a tag-pinned member must not collapse onto the task's current revision"
+    )
+
+
 async def test_create_validates_missing_task_ref(service: TasksetService) -> None:
     taskset_input = TasksetInput(tasks=[TaskRef("task-a"), TaskRef("nope")])
     with pytest.raises(TaskRefNotFoundError, match="not found"):

plugins/nemo-evaluator/tests/test_subentity_refs.py

@@ -90,10 +90,21 @@ def test_task_ref_rejects_malformed_fragments(ref: str) -> None:
         TaskRef(ref)
 
 
-@pytest.mark.parametrize("ref_type", [MetricRef, TasksetRef])
-def test_sibling_ref_types_still_reject_fragments(ref_type: type) -> None:
-    """The fragment pattern is a sibling, not a widening of the shared constant: metrics and
-    tasksets have no revisions yet, so admitting a fragment would accept input nothing resolves.
-    They move onto it when they gain revisions — deliberately, at that point."""
+def test_metric_ref_still_rejects_fragments() -> None:
+    """The fragment pattern is a sibling, not a widening of the shared constant: metrics have no
+    revisions, so admitting a fragment would accept input nothing resolves. ``MetricRef`` moves onto
+    it when metrics gain revisions — deliberately, at that point."""
     with pytest.raises(ValidationError):
-        ref_type(f"other/thing#{_DIGEST}")
+        MetricRef(f"other/thing#{_DIGEST}")
+
+
+@pytest.mark.parametrize("ref", ["suite", "other/suite", "suite#latest", "suite#blessed", f"other/suite#{_DIGEST}"])
+def test_taskset_ref_accepts_fragments(ref: str) -> None:
+    """Tasksets are revisioned, so a ref may pin the revision to expand — same shape as ``TaskRef``."""
+    assert TasksetRef(ref).root == ref
+
+
+@pytest.mark.parametrize("ref", ["suite#one#two", "suite#bad/frag", "#latest", "suite name#latest"])
+def test_taskset_ref_rejects_malformed_fragments(ref: str) -> None:
+    with pytest.raises(ValidationError):
+        TasksetRef(ref)