feat(evaluator): give a taskset its own files

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

@@ -27,7 +27,7 @@
     MetricMetadata,
 )
 from nemo_evaluator_sdk.values.common import SecretRef
-from nemo_platform_plugin.refs import ENTITY_REF_PATTERN, parse_entity_ref
+from nemo_platform_plugin.refs import ENTITY_REF_PATTERN, FILESET_REF_PATTERN, parse_entity_ref
 from pydantic import AfterValidator, BaseModel, ConfigDict, Field, RootModel, field_validator
 
 
@@ -253,6 +253,21 @@ def _reject_duplicate_metadata_keys(items: list[MetadataItem]) -> list[MetadataI
 #: A task's metadata: key/value annotations with unique keys (duplicates rejected at validation).
 TaskMetadataList: TypeAlias = Annotated[list[MetadataItem], AfterValidator(_reject_duplicate_metadata_keys)]
 
+#: Where a taskset's own files live: one Files reference, ``workspace/fileset#prefix``. A
+#: grouping's files are a directory, not a set of independently addressed blobs, so one reference
+#: says everything a list of per-file entries would — while making it impossible for the two to
+#: disagree about what the taskset ships.
+#:
+#: Same shape as ``bundle_ref`` and ``archive_ref``, so the fragment is required; here it names the
+#: prefix the files sit under rather than a single file. Use a prefix such as ``#files`` to mean
+#: "the whole of this fileset's file area".
+#:
+#: Pinning is arranged by pointing at a location that is not rewritten — a fileset written once, or
+#: a content-addressed prefix inside a shared one. A reference into a location that *is* later
+#: rewritten resolves to whatever it holds when read, which is the contract the Files service gives
+#: every other consumer.
+TasksetFilesRef: TypeAlias = Annotated[str, Field(pattern=FILESET_REF_PATTERN)]
+
 
 def _reject_duplicate_task_refs(refs: list[TaskRef]) -> list[TaskRef]:
     """A taskset's members are an unordered set expressed as a list; a repeated ref is ambiguous

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

@@ -54,6 +54,9 @@
 from nemo_evaluator.api.fields import (
     TaskRefList as TaskRefList,
 )
+from nemo_evaluator.api.fields import (
+    TasksetFilesRef as TasksetFilesRef,
+)
 from nemo_evaluator.api.fields import (
     TasksetRef as TasksetRef,
 )
@@ -301,6 +304,10 @@ class Taskset(BaseModel):
     tasks: TaskRefList = Field(
         default_factory=list, description="References to the member tasks (set semantics; duplicates rejected)."
     )
+    files_ref: TasksetFilesRef | None = Field(
+        default=None,
+        description="Files reference to the taskset's own files — shared by its members, owned by none.",
+    )
     metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the taskset.")
     revision: int = Field(
         description="Ordinal of the published revision this content corresponds to. Every stored "
@@ -334,6 +341,13 @@ class TasksetInput(BaseModel):
         "Because membership is a set, the stored order is canonical rather than the submitted order: "
         "reordering the same members is not a content change and publishes no revision.",
     )
+    files_ref: TasksetFilesRef | None = Field(
+        default=None,
+        description="Files reference to the taskset's own files — shared by its members, owned by none. "
+        "Upload them to the Files service first and point here ('workspace/fileset#prefix'). The "
+        "reference is part of the taskset's content, so repointing it publishes a revision; pin the "
+        "content by referencing a location that is not rewritten.",
+    )
     metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the taskset.")
     tags: list[str] = Field(
         default_factory=list,

plugins/nemo-evaluator/src/nemo_evaluator/api/service/taskset_service.py

@@ -101,6 +101,7 @@ def _entity_to_taskset(entity: TasksetEntity) -> Taskset:
         project=entity.project,
         description=entity.description,
         tasks=entity.tasks,
+        files_ref=entity.files_ref,
         revision=entity.latest_revision,
         tags=entity.tags,
         metadata=entity.metadata,
@@ -122,6 +123,7 @@ def _revision_to_taskset(head: TasksetEntity, revision: TasksetRevisionEntity) -
         project=head.project,
         description=revision.description,
         tasks=revision.tasks,
+        files_ref=revision.files_ref,
         metadata=revision.metadata,
         revision=revision.revision,
         tags={tag: ordinal for tag, ordinal in head.tags.items() if ordinal == revision.revision},
@@ -265,6 +267,7 @@ async def create_taskset(
             project=project,
             description=taskset_input.description,
             tasks=await self._resolved_content(taskset_input, workspace=workspace),
+            files_ref=taskset_input.files_ref,
             metadata=taskset_input.metadata,
         )
         try:
@@ -312,6 +315,7 @@ async def replace_taskset(
             head.project = project
         head.description = taskset_input.description
         head.tasks = await self._resolved_content(taskset_input, workspace=workspace)
+        head.files_ref = taskset_input.files_ref
         head.metadata = taskset_input.metadata
         # Publish the staged content *without* committing the head first — see the matching comment
         # in ``TaskService.replace_task``. Publishing writes the head itself, so a pre-write would

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

@@ -24,6 +24,7 @@
     TaskDefinition,
     TaskMetadataList,
     TaskRefList,
+    TasksetFilesRef,
 )
 from nemo_evaluator.content_hash import DIGEST_LENGTH, DIGEST_PATTERN
 from nemo_evaluator.shared.metric_bundles.bundles import BundledMetricOutputSpec
@@ -266,6 +267,12 @@ class TasksetEntity(_RevisionedCommon, EntityBase):
     A taskset is a flexible grouping of stored tasks: it holds references to its members
     (``workspace/name``) plus free-form annotations. Membership is a set — order is not significant
     and duplicate references are rejected. Referenced tasks are validated to exist at create time.
+
+    ``files_ref`` points at the grouping's own files: shared by its members, owned by none of
+    them, the way a Harbor dataset ships a metric script beside its tasks. One reference rather
+    than a list of per-file entries — a grouping's files are a directory, and the fileset already
+    knows what is in it. It is a first-class field rather than an annotation because it is
+    content: the revision digest covers it, so repointing publishes a revision.
     """
 
     __entity_type__: ClassVar[str] = "taskset"
@@ -279,6 +286,10 @@ class TasksetEntity(_RevisionedCommon, EntityBase):
         default_factory=list,
         description="References to the member tasks (set semantics; duplicates rejected).",
     )
+    files_ref: TasksetFilesRef | None = Field(
+        default=None,
+        description="Files reference to the taskset's own files (workspace/fileset[#prefix]).",
+    )
     metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the taskset.")
 
 
@@ -371,4 +382,9 @@ class TasksetRevisionEntity(_RevisionCommon, EntityBase):
         default_factory=list,
         description="Digest-pinned references to the member tasks, resolved at publish time.",
     )
+    files_ref: TasksetFilesRef | None = Field(
+        default=None,
+        description="Files reference as published. Unlike a member ref there is nothing to resolve: "
+        "a Files reference already names an exact location.",
+    )
     metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the taskset.")

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

@@ -0,0 +1,119 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""A taskset's own files: one Files reference, shared by its members and owned by none of them."""
+
+from __future__ import annotations
+
+import hashlib
+
+import pytest
+from nemo_evaluator.api.schemas import TaskRef, TasksetInput
+from nemo_evaluator.api.service.taskset_service import TasksetService
+from nemo_platform_plugin.entity_client import NemoEntityNotFoundError
+from pydantic import ValidationError
+
+DIGEST_A = hashlib.sha256(b"files-v1").hexdigest()
+DIGEST_B = hashlib.sha256(b"files-v2").hexdigest()
+REF_A = f"default/harbor-packages#packages/nvidia.ds/{DIGEST_A}"
+REF_B = f"default/harbor-packages#packages/nvidia.ds/{DIGEST_B}"
+
+
+class _FakeTaskService:
+    def __init__(self, existing: set[tuple[str, str]]) -> None:
+        self.existing = existing
+
+    async def get_task(self, workspace: str, name: str) -> object | None:
+        return object() if (workspace, name) in self.existing else None
+
+    async def resolve_revision(self, workspace: str, name: str, fragment: str = "latest") -> str:
+        if (workspace, name) not in self.existing:
+            raise NemoEntityNotFoundError(f"{workspace}/{name} not found")
+        return hashlib.sha256(f"{workspace}/{name}#{fragment}".encode()).hexdigest()
+
+
+@pytest.fixture
+def service(entity_store) -> TasksetService:
+    return TasksetService(entity_store, _FakeTaskService({("default", "task-a")}))
+
+
+def _input(files_ref: str | None = None) -> TasksetInput:
+    return TasksetInput(tasks=[TaskRef("task-a")], files_ref=files_ref)
+
+
+async def test_files_ref_round_trips(service: TasksetService) -> None:
+    created, _ = await service.create_taskset("ts-1", _input(REF_A), workspace="default")
+    assert created.files_ref == REF_A
+
+    got = await service.get_taskset("default", "ts-1")
+    assert got is not None and got.files_ref == REF_A
+
+
+async def test_repointing_the_files_publishes_a_revision(service: TasksetService) -> None:
+    """The reference is content, not annotation: changing where a taskset's files come from
+    changes what the taskset is."""
+    created, _ = await service.create_taskset("ts-1", _input(REF_A), workspace="default")
+    assert created.revision == 1
+
+    replaced, published = await service.replace_taskset("ts-1", _input(REF_B), workspace="default")
+
+    assert published is True
+    assert replaced.revision == 2
+    assert replaced.files_ref == REF_B
+
+
+async def test_republishing_the_same_ref_publishes_nothing(service: TasksetService) -> None:
+    await service.create_taskset("ts-1", _input(REF_A), workspace="default")
+
+    replaced, published = await service.replace_taskset("ts-1", _input(REF_A), workspace="default")
+
+    assert published is False
+    assert replaced.revision == 1
+
+
+async def test_a_pinned_revision_keeps_the_ref_it_was_published_with(service: TasksetService) -> None:
+    """Why the digest covers the reference at all: revision 1 must still name the files it was
+    published with, after the head has been repointed."""
+    await service.create_taskset("ts-1", _input(REF_A), workspace="default")
+    await service.replace_taskset("ts-1", _input(REF_B), workspace="default")
+
+    # Selected by content digest, not ordinal: a non-digest fragment is read as a *tag* name, so
+    # "1" would be a lookup for a tag called "1".
+    revisions = await service.list_revisions("default", "ts-1")
+    first_digest = next(r.content_hash for r in revisions.data if r.revision == 1)
+
+    first = await service.get_taskset("default", "ts-1", revision=first_digest)
+    assert first is not None and first.files_ref == REF_A
+
+    head = await service.get_taskset("default", "ts-1")
+    assert head is not None and head.files_ref == REF_B
+
+
+async def test_the_ref_defaults_to_none(service: TasksetService) -> None:
+    """The field is additive: a taskset that ships no files is exactly as it was before."""
+    created, _ = await service.create_taskset("ts-1", _input(), workspace="default")
+    assert created.files_ref is None
+
+
+async def test_clearing_the_ref_publishes_a_revision(service: TasksetService) -> None:
+    """Dropping a taskset's files is as much a content change as repointing them."""
+    await service.create_taskset("ts-1", _input(REF_A), workspace="default")
+
+    replaced, published = await service.replace_taskset("ts-1", _input(None), workspace="default")
+
+    assert published is True
+    assert replaced.files_ref is None
+
+
+def test_a_plain_prefix_is_accepted() -> None:
+    """The simplest useful form: a fileset plus the prefix its files sit under."""
+    assert TasksetInput(files_ref="default/my-dataset#files").files_ref == "default/my-dataset#files"
+
+
+def test_the_ref_must_be_a_fileset_reference() -> None:
+    """Including the fragment. Same shape as ``bundle_ref``/``archive_ref``, so a bare
+    ``workspace/fileset`` is not a Files reference here and is rejected rather than quietly
+    stored as something no reader can resolve."""
+    for bad in ("not a ref", "", "default/my-dataset", "default/fs#bad path"):
+        with pytest.raises(ValidationError):
+            TasksetInput(files_ref=bad)