From 34ee87893fdfd4ee2cc3e079610439190f25968f Mon Sep 17 00:00:00 2001 From: Sandy Chapman Date: Tue, 11 Aug 2026 12:05:59 -0300 Subject: [PATCH] feat(evaluator): give a taskset its own files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A grouping can need files that every member uses and none owns — a shared metric script, a fixture, a schema. A Harbor dataset ships exactly this beside its tasks. There was nowhere to put them, so the only option was to encode a JSON blob in `metadata`, which is unqueryable, untyped, and invisible to anything that does not know the convention. `files_ref` is one Files reference for the whole grouping, the same shape as `bundle_ref` and `archive_ref`. A taskset's files are a directory, not a set of independently addressed blobs: the fileset already knows what it contains, so a list of per-file references would carry no information it does not hold and would let the two disagree about what the taskset ships. It is a first-class field rather than an annotation because it is content — the revision digest covers it, so repointing or clearing it publishes a revision and a pinned revision keeps naming the files it was published with. Pinning the bytes is arranged by referencing 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. The alternative — a per-file digest stored beside each reference — buys pinning the entity store cannot enforce anyway, and duplicates state the Files service owns. The revision entity carries the reference too, but needs no resolution step: unlike a member ref, which is resolved from a possibly-tag-pinned form to an exact digest at publish time, a Files reference already names an exact location. The field is additive and defaults to unset, so an existing taskset is unchanged and its revision digest does not move. Signed-off-by: Sandy Chapman --- plugins/nemo-evaluator/openapi/openapi.yaml | 15 +++ .../src/nemo_evaluator/api/fields.py | 17 ++- .../src/nemo_evaluator/api/schemas.py | 14 +++ .../api/service/taskset_service.py | 4 + .../src/nemo_evaluator/entities.py | 16 +++ .../tests/api/service/test_taskset_files.py | 119 ++++++++++++++++++ 6 files changed, 184 insertions(+), 1 deletion(-) create mode 100644 plugins/nemo-evaluator/tests/api/service/test_taskset_files.py diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml index d9d602b24a..8df3a434a6 100644 --- a/plugins/nemo-evaluator/openapi/openapi.yaml +++ b/plugins/nemo-evaluator/openapi/openapi.yaml @@ -5346,6 +5346,12 @@ components: type: array title: Tasks description: References to the member tasks (set semantics; duplicates rejected). + files_ref: + title: Files Ref + description: "Files reference to the taskset's own files \u2014 shared by\ + \ its members, owned by none." + type: string + pattern: ^[\w\-.]+/[\w\-.]+#[\w\-./]+$ metadata: items: $ref: '#/components/schemas/MetadataItem' @@ -5429,6 +5435,15 @@ components: change underneath you when a member republishes. 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: + title: Files Ref + description: "Files reference to the taskset's own files \u2014 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." + type: string + pattern: ^[\w\-.]+/[\w\-.]+#[\w\-./]+$ metadata: items: $ref: '#/components/schemas/MetadataItem' diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/fields.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/fields.py index 2e682c9baa..8da4eefa6d 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/fields.py +++ b/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 diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py index b6b3c5017f..b7ecf1f2fc 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py +++ b/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, diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/service/taskset_service.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/service/taskset_service.py index b6fe457ca1..c3e6964246 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/service/taskset_service.py +++ b/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 diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/entities.py b/plugins/nemo-evaluator/src/nemo_evaluator/entities.py index a32c66ec1f..c7547c62e8 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/entities.py +++ b/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.") diff --git a/plugins/nemo-evaluator/tests/api/service/test_taskset_files.py b/plugins/nemo-evaluator/tests/api/service/test_taskset_files.py new file mode 100644 index 0000000000..46f413682d --- /dev/null +++ b/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)