diff --git a/docs/_scripts/lint_python_snippets.py b/docs/_scripts/lint_python_snippets.py index 73fae2263e..a8e4b6e798 100644 --- a/docs/_scripts/lint_python_snippets.py +++ b/docs/_scripts/lint_python_snippets.py @@ -51,7 +51,9 @@ "", "", } -DEFAULT_IGNORED_TY_RULES = ("possibly-unbound-attribute",) +# ``possibly-unbound-attribute`` was renamed upstream; passing the old name makes ty emit +# ``warning[unknown-rule]``, which fails this check for every doc regardless of its snippets. +DEFAULT_IGNORED_TY_RULES = ("possibly-missing-attribute",) @dataclass(frozen=True) diff --git a/docs/evaluator/manage-tasks-tasksets.mdx b/docs/evaluator/manage-tasks-tasksets.mdx index 32c1be6ce5..a3f20e4c90 100644 --- a/docs/evaluator/manage-tasks-tasksets.mdx +++ b/docs/evaluator/manage-tasks-tasksets.mdx @@ -73,34 +73,69 @@ Reference the stored metric with a `MetricRef` (`workspace/name`, or a bare `nam the task's workspace). The service returns the stored `Task`. ```python -from nemo_evaluator.api.schemas import MetadataItem, MetricRef, TaskInput, TaskInputs +from nemo_evaluator.api.schemas import EvaluatorTaskDefinition, MetadataItem, MetricRef, TaskInput, TaskInputs task = TaskInput( - intent="Answer the user's geography question with the capital city.", - inputs=TaskInputs(instruction="What is the capital of France?"), - metrics=[MetricRef("default/answer-exact-match")], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the user's geography question with the capital city.", + inputs=TaskInputs(instruction="What is the capital of France?"), + metrics=[MetricRef("default/answer-exact-match")], + ), metadata=[MetadataItem(key="suite", value="geography")], ) stored = tasks.create("capital-of-france", task=task) -print(stored.id, stored.metrics) +print(stored.id, stored.spec.metrics) ``` ### `TaskInput` fields +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `spec` | `TaskDefinition` | Yes | The task's content, discriminated by `kind` — see below. | +| `metadata` | `list[MetadataItem]` | No | Key/value annotations. Keys must be unique. | +| `tags` | `list[str]` | No | Tags to point at the revision this request publishes. `latest` is always applied server-side. | + +### Task kinds + +A task is an evaluation unit; its `kind` says which runner executes it. There are two: + +- `evaluator` — the task's content is fields you author, scored by platform metrics. +- `harbor` — the task's content is a packaged directory of files, scored by Harbor's own reward. + +Both are stored as the same record type, so a taskset can group them and you manage every evaluation +unit in one place regardless of which runner executes it. + +`EvaluatorTaskDefinition` (`kind="evaluator"`): + | Field | Type | Required | Description | |-------|------|----------|-------------| | `intent` | `str` | Yes | Human-readable description of the desired agent behavior. | | `inputs` | `TaskInputs` | No | The task's recognized input fields. `instruction` is the agent's prompt; it falls back to `intent` when unset. | +| `reference` | `dict[str, Any]` | No | Grader-only ground truth (held-out tests, expected outputs, rubric data). Surfaced to metrics but never seeded into the agent's workspace or shown to the agent. Held out from the *agent*, not from the API. | | `metrics` | `list[MetricRefOrInline]` | No | The metrics that score the task, as `MetricRef` references (`workspace/name`) to stored metrics. Pre-built inline metric bundles (`MetricInline`) are also accepted and are normalized to stored metrics on create. | | `views` | `dict[str, SemanticView]` | No | Optional reporting views mapping metric outputs into named semantic scores. | -| `metadata` | `list[MetadataItem]` | No | Key/value annotations. Keys must be unique. | -| `tags` | `list[str]` | No | Tags to point at the revision this request publishes. `latest` is always applied server-side. | + +`HarborTaskDefinition` (`kind="harbor"`): + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `archive_ref` | `str` | Yes | Files reference to the task's packaged directory (`workspace/fileset#path`). One fileset per task, so a task shared by several tasksets is stored once. | +| `archive_digest` | `str` | Yes | Content hash Harbor computed over the task directory. | +| `instruction` | `str` | No | The task's instruction text, when it has one. | +| `config` | `dict` | No | Harbor's own task configuration (verifier, agent, environment, steps), stored as published. | + + +Storing a Harbor task is supported; **running one from storage is not yet**. A taskset may group both +kinds, but expanding a `harbor` member is rejected with `422` before the run starts, whatever target +you submit against. Harbor evaluations continue to run through the existing dataset-driven path. + A stored task holds **metric references only**. Any inline metric bundle you pass on create is stored as a content-addressed *derived* metric, and the task record is normalized to reference it. This is -why `stored.metrics` always comes back as a list of `MetricRef` references. +why `stored.spec.metrics` always comes back as a list of `MetricRef` references. ### Retrieve, list, and delete @@ -108,12 +143,12 @@ why `stored.metrics` always comes back as a list of `MetricRef` references. ```python # Retrieve one task by name (its current content) task = tasks.retrieve("capital-of-france") -print(task.revision, task.tags) # e.g. 1 {'latest': 1} +print(task.spec.kind, task.revision, task.tags) # e.g. evaluator 1 {'latest': 1} # List tasks in the workspace (paginated) page = tasks.list(page=1, page_size=100, sort="-created_at") for item in page.data: - print(item.name, item.intent) + print(item.name, item.spec.kind) # Delete a task (this also removes all of its revisions) tasks.delete("capital-of-france") @@ -131,9 +166,12 @@ no existence check. ```python revised_task = TaskInput( - intent="Answer the user's geography question with the capital city.", - inputs=TaskInputs(instruction="Name the capital city of France."), - metrics=[MetricRef("default/answer-exact-match")], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the user's geography question with the capital city.", + inputs=TaskInputs(instruction="Name the capital city of France."), + metrics=[MetricRef("default/answer-exact-match")], + ), metadata=[MetadataItem(key="suite", value="geography")], ) @@ -169,7 +207,7 @@ original = tasks.retrieve("capital-of-france", revision=digest) # revision 1, a current = tasks.retrieve("capital-of-france") # revision 2, the current content assert original.revision == 1 and current.revision == 2 -assert original.inputs.instruction != current.inputs.instruction +assert original.spec.inputs.instruction != current.spec.inputs.instruction ``` ### Tag a revision @@ -361,9 +399,10 @@ A fragment that no longer resolves fails the evaluation rather than falling back revision. -Stored tasks carry no grader-only `reference` (held-out ground truth): that field lives only on inline -`AgentEvalTaskInput`. Taskset-driven tasks therefore run with an empty `reference`, so use a taskset -when your metrics score the agent's output directly rather than against per-task held-out data. +A member's grader-only `reference` (held-out ground truth) is loaded from the pinned revision along +with the rest of its content, so a taskset-driven run grades against the ground truth that revision +fixed. Because `reference` is covered by the revision digest, changing it publishes a new revision — +a pin fixes the grading, not just the prompt. ## Async usage diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/refs.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/refs.py index 2c7c555f15..f43586ab29 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/refs.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/refs.py @@ -93,6 +93,14 @@ class LocalDir(StrRef): __cli_metavar__: ClassVar[str | None] = "PATH" +#: Regex form of the shape :func:`parse_entity_ref` accepts: ``name`` or ``workspace/name``, each +#: segment using the platform name charset. Pydantic fields that hold a reference declare +#: ``pattern=ENTITY_REF_PATTERN`` so a malformed ref is rejected at validation rather than surfacing +#: as a confusing failure during parsing; :func:`parse_entity_ref` then only has to split. Kept +#: beside the parser so the two cannot drift apart. +ENTITY_REF_PATTERN = r"^[\w\-.]+(/[\w\-.]+)?$" + + class FilesetRef(StrRef): """A NeMo Platform fileset reference (``"name"`` or ``"workspace/name"``). @@ -104,6 +112,14 @@ class FilesetRef(StrRef): __cli_metavar__: ClassVar[str | None] = "FILESET_REF" +#: A reference to a *file inside* a fileset: ``workspace/fileset#path/inside.ext``. Unlike +#: :data:`ENTITY_REF_PATTERN` the workspace is mandatory (a stored reference must be unambiguous +#: wherever it is later read from), and the ``#`` fragment is a file path, so it admits ``/`` and +#: ``.``. Declared as a field pattern so a malformed reference is rejected when it is stored rather +#: than surfacing as a download failure mid-run. +FILESET_REF_PATTERN = r"^[\w\-.]+/[\w\-.]+#[\w\-./]+$" + + # Documentary union alias — the wire shape is still ``str``. The # ``_spec_flags`` generator collapses this to a single ``--output`` flag # of type ``str``; the disambiguation between the two arms happens in @@ -182,6 +198,8 @@ def parse_entity_ref(identifier: str, default_workspace: str | None = None) -> P __all__ = [ + "ENTITY_REF_PATTERN", + "FILESET_REF_PATTERN", "EndpointURL", "FilesetRef", "LocalDir", diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml index 366017a676..eb4a8a2c21 100644 --- a/plugins/nemo-evaluator/openapi/openapi.yaml +++ b/plugins/nemo-evaluator/openapi/openapi.yaml @@ -3461,6 +3461,64 @@ components: title: EvaluateSpec description: Canonical SDK evaluation spec with platform model and metric references resolved. + EvaluatorTaskDefinition: + properties: + kind: + type: string + const: evaluator + title: Kind + description: Task kind discriminator. + intent: + type: string + title: Intent + description: Human-readable description of the desired agent behavior. + inputs: + allOf: + - $ref: '#/components/schemas/TaskInputs' + description: The task's recognized input fields. + metrics: + items: + anyOf: + - $ref: '#/components/schemas/MetricInline' + - $ref: '#/components/schemas/MetricRef' + type: array + title: Metrics + description: "Metrics that score this task \u2014 stored-metric references,\ + \ and inline bundles on create (normalized to derived stored metrics before\ + \ the task is persisted)." + reference: + additionalProperties: true + type: object + title: Reference + description: 'Grader-only ground truth (held-out tests, expected outputs, + rubric data). Surfaced to metrics but never seeded into the agent''s workspace + or shown to the agent, so a metric can grade against artifacts the agent + cannot influence. Held out from the *agent*, not from the API: anyone + who can read the task can read this.' + views: + additionalProperties: + $ref: '#/components/schemas/SemanticView' + type: object + title: Views + description: Optional reporting views mapping metric outputs into named + semantic scores. + additionalProperties: false + type: object + required: + - kind + - intent + title: EvaluatorTaskDefinition + description: "What the agent should do, and how the platform scores it.\n\n\ + ``metrics`` accepts inline bundles on the way in and holds references once\ + \ stored: the service\noffloads an inline metric to a content-addressed *derived*\ + \ metric on create, so a persisted task\nonly ever names metrics it does not\ + \ own. That narrowing is a service invariant rather than a\ntype-level one\ + \ \u2014 a single model keeps the API surface small, at the cost of this field\ + \ being\nwider than what a stored task actually contains.\n\nEvery field here\ + \ is covered by the revision digest, ``reference`` included: it decides what\ + \ a\nmetric grades against, so two revisions that score differently must not\ + \ share a digest. Pinning\na revision therefore fixes the grading, not just\ + \ the prompt." EvidenceDescriptor: anyOf: - required: @@ -3764,6 +3822,60 @@ components: injected from the job''s storage at run time; only the harness-selection and run knobs live here.' + HarborTaskDefinition: + properties: + kind: + type: string + const: harbor + title: Kind + description: Task kind discriminator. + archive_ref: + type: string + pattern: ^[\w\-.]+/[\w\-.]+#[\w\-./]+$ + title: Archive Ref + description: 'Files reference to the task''s packaged directory (format: + workspace/fileset#path).' + archive_digest: + type: string + maxLength: 64 + minLength: 64 + pattern: ^[0-9a-f]{64}$ + title: Archive Digest + description: "Content hash Harbor computed over the task directory. This\ + \ is the authoritative identity of a Harbor task's content \u2014 every\ + \ file, including task.toml." + instruction: + title: Instruction + description: The task's instruction text, when it has one (multi-step tasks + may not). + type: string + config: + additionalProperties: true + type: object + title: Config + description: "Harbor's own task configuration (verifier, agent, environment,\ + \ steps), as published. A queryable projection of task.toml \u2014 inspect\ + \ a task's verifier without downloading the archive. Opaque here: Harbor\ + \ owns this schema." + additionalProperties: false + type: object + required: + - kind + - archive_ref + - archive_digest + title: HarborTaskDefinition + description: "A reference to the task's packaged files, plus a projection of\ + \ Harbor's own config.\n\nHarbor identifies a task by a *directory* \u2014\ + \ ``task.toml``, an instruction, an environment \u2014 so\nwhat is stored\ + \ is a reference to that directory's archive in the Files service, not the\ + \ files\nthemselves. One fileset per task, so a task shared by several tasksets\ + \ is stored once. The\narchive is materialized back into ``//``\ + \ at run time, which is the layout\nHarbor's own discovery expects.\n\nWhich\ + \ agent runs the task is *not* stored here. That comes from the run's target\n\ + (``HarborRunnerTarget``), so the same stored task can be evaluated against\ + \ different agents.\nHarbor's own ``[agent]`` block \u2014 carried inside\ + \ ``config`` \u2014 configures how the agent *phase*\nruns (timeout, user,\ + \ network policy), not which agent it is." HelloResponse: properties: message: @@ -5026,29 +5138,18 @@ components: title: Project description: The project associated with this task. type: string - intent: - type: string - title: Intent - description: Human-readable description of the desired agent behavior. - inputs: - allOf: - - $ref: '#/components/schemas/TaskInputs' - description: The task's recognized input fields. - metrics: - items: - $ref: '#/components/schemas/MetricRef' - type: array - title: Metrics - description: References to the metrics that score this task; inline metrics - submitted on create are normalized to (derived) stored metrics, so a stored - task holds refs only. - views: - additionalProperties: - $ref: '#/components/schemas/SemanticView' - type: object - title: Views - description: Optional reporting views mapping metric outputs into named - semantic scores. + spec: + oneOf: + - $ref: '#/components/schemas/EvaluatorTaskDefinition' + - $ref: '#/components/schemas/HarborTaskDefinition' + title: Spec + description: The task's content, discriminated by which runner executes + it. + discriminator: + propertyName: kind + mapping: + evaluator: '#/components/schemas/EvaluatorTaskDefinition' + harbor: '#/components/schemas/HarborTaskDefinition' metadata: items: $ref: '#/components/schemas/MetadataItem' @@ -5085,7 +5186,7 @@ components: - id - name - workspace - - intent + - spec - revision - created_at - updated_at @@ -5120,30 +5221,18 @@ components: type: object TaskInput: properties: - intent: - type: string - title: Intent - description: Human-readable description of the desired agent behavior. - inputs: - allOf: - - $ref: '#/components/schemas/TaskInputs' - description: The task's recognized input fields. - metrics: - items: - anyOf: - - $ref: '#/components/schemas/MetricInline' - - $ref: '#/components/schemas/MetricRef' - type: array - title: Metrics - description: "Metrics that score this task \u2014 inline bundles and/or\ - \ stored-metric refs." - views: - additionalProperties: - $ref: '#/components/schemas/SemanticView' - type: object - title: Views - description: Optional reporting views mapping metric outputs into named - semantic scores. + spec: + oneOf: + - $ref: '#/components/schemas/EvaluatorTaskDefinition' + - $ref: '#/components/schemas/HarborTaskDefinition' + title: Spec + description: The task's content, discriminated by which runner executes + it. + discriminator: + propertyName: kind + mapping: + evaluator: '#/components/schemas/EvaluatorTaskDefinition' + harbor: '#/components/schemas/HarborTaskDefinition' metadata: items: $ref: '#/components/schemas/MetadataItem' @@ -5160,7 +5249,7 @@ components: additionalProperties: false type: object required: - - intent + - spec title: TaskInput description: "Create/replace body for a stored task (the name comes from the\ \ path).\n\nThe authorable subset of :class:`Task` \u2014 the SDK ``AgentEvalTask``\ diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/fields.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/fields.py new file mode 100644 index 0000000000..2e682c9baa --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/fields.py @@ -0,0 +1,297 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared field types for the evaluator API: entity references and metric payloads. + +Split out of :mod:`nemo_evaluator.api.schemas` so the per-kind task definitions can use these +without importing the module that composes them into DTOs — the definitions are imported *by* +``schemas``, so they cannot import from it. + +What counts as a ``workspace/name`` reference is **not** decided here: the shape +(:data:`~nemo_platform_plugin.refs.ENTITY_REF_PATTERN`) and the parser +(:func:`~nemo_platform_plugin.refs.parse_entity_ref`) are the platform's, shared with every other +plugin. This module only adds what is specific to a *revisioned* evaluator entity — the ``#fragment`` +that selects a revision. + +Everything here is re-exported from ``schemas`` for callers that already import it from there. +""" + +from __future__ import annotations + +import re +from typing import Annotated, Any, Literal, TypeAlias + +from nemo_evaluator.content_hash import DIGEST_PATTERN +from nemo_evaluator.shared.metric_bundles.bundles import ( + BundledMetricOutputSpec, + MetricMetadata, +) +from nemo_evaluator_sdk.values.common import SecretRef +from nemo_platform_plugin.refs import ENTITY_REF_PATTERN, parse_entity_ref +from pydantic import AfterValidator, BaseModel, ConfigDict, Field, RootModel, field_validator + + +class CloudpickleMetricPayload(BaseModel): + """Wire schema for a cloudpickle-serialized metric payload. + + Mirrors the runtime ``CloudpickleMetricPayload`` so the API contract is + explicit in the OpenAPI spec. The runtime bundle model serializes payloads + polymorphically (typed as an abstract base), which renders as an opaque + object in the spec; this concrete DTO documents the actual fields. + """ + + model_config = ConfigDict(extra="forbid", ser_json_bytes="base64", val_json_bytes="base64") + + kind: Literal["cloudpickle"] = Field(description="Payload format discriminator.") + python_version: str = Field(description="Python version the metric was pickled with (must match at execution).") + cloudpickle_version: str = Field(description="cloudpickle version used to serialize the metric.") + pickle_protocol: int = Field(description="Pickle protocol used.") + blob: bytes = Field(description="Base64-encoded cloudpickled metric object.") + digest: str | None = Field( + default=None, + description="SHA-256 digest of the payload bytes. Informational; recomputed server-side.", + ) + + +class InlineMetricPayload(BaseModel): + """Wire schema for an inline (config-serialized) metric payload. + + Mirrors the runtime ``InlineMetricPayload``. The metric is stored as its own + JSON configuration and reconstructed from the metric type union at execution, + so no code is shipped or executed on load. Used for platform-recognized + built-in metric types. + """ + + model_config = ConfigDict(extra="forbid") + + kind: Literal["inline"] = Field(description="Payload format discriminator.") + metric: dict[str, Any] = Field( + description="JSON-serialized built-in metric configuration, discriminated by its own `type`." + ) + digest: str | None = Field( + default=None, + description="SHA-256 digest of the canonical metric JSON. Informational; recomputed server-side.", + ) + + @field_validator("metric") + @classmethod + def _metric_must_declare_type(cls, value: dict[str, Any]) -> dict[str, Any]: + """Reject payloads without a metric ``type`` discriminator at the API boundary. + + The metric body stays an open object (the concrete shape is validated when + the bundle is hydrated against the metric type union), but a non-empty + ``type`` is required so malformed payloads fail fast rather than at execution. + """ + metric_type = value.get("type") + if not isinstance(metric_type, str) or not metric_type: + raise ValueError("inline metric payload must include a non-empty 'type'") + return value + + +# Discriminated on ``kind`` so additional payload formats can join the union +# without changing the field type. +MetricPayload = Annotated[CloudpickleMetricPayload | InlineMetricPayload, Field(discriminator="kind")] + + +class MetricInline(BaseModel): + """An executable metric submitted to the platform. + + Carries the bundled metric — type, metadata, output contracts, secret + references, and a format-specific payload — used both as the create-request + body and as an inline metric in an evaluation job. + """ + + model_config = ConfigDict(extra="forbid") + + bundle_kind: Literal["metric-bundle"] = "metric-bundle" + bundle_format_version: Literal["v1"] = "v1" + metric_type: str = Field(min_length=1, description="Runtime metric type name.") + metadata: MetricMetadata = Field(default_factory=MetricMetadata, description="User-facing metric metadata.") + outputs: list[BundledMetricOutputSpec] = Field(min_length=1, description="The metric's output contracts.") + secrets: dict[str, SecretRef] = Field( + default_factory=dict, description="Secret references required to execute the metric." + ) + payload: MetricPayload = Field(description="Format-specific serialized metric.") + + +#: The charset a ``#fragment`` may use. Exported because anything that *mints* a fragment — notably +#: revision tag names — has to be constrained by it: a value outside this set can be stored happily +#: and then never appear in a reference, which is a silent dead end rather than an error. +REF_FRAGMENT_CHARSET = r"[\w\-.]+" + +# A *sub-entity* reference adds an optional ``#fragment`` to the platform's ``ENTITY_REF_PATTERN``, +# which is the standard way of addressing something contained within an entity (filesets address a +# contained file the same way: ``workspace/fileset#path``). For a revisioned entity the fragment +# selects a revision — either a tag (``#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 +# 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. +# +# The base alternation is spliced in from the shared constant (minus its anchors) so the two shapes +# cannot drift: widening what counts as a ``workspace/name`` widens both at once. +_SUBENTITY_REF_PATTERN = rf"^{ENTITY_REF_PATTERN.removeprefix('^').removesuffix('$')}(#{REF_FRAGMENT_CHARSET})?$" +#: The fragment separator for sub-entity references. Matches the fileset/job ref convention. +REF_FRAGMENT_SEPARATOR = "#" +#: The tag applied to every publish and used when a ref carries no fragment. +LATEST_TAG = "latest" + + +def parse_subentity_ref(root: str, default_workspace: str) -> tuple[str, str, str]: + """Split a reference into ``(workspace, name, fragment)``. + + The ``workspace/name`` split is delegated to the platform's :func:`~nemo_platform_plugin.refs. + parse_entity_ref`; this only adds the revision fragment on top, so evaluator refs and every other + plugin's refs agree on what a ``workspace/name`` is. Callers that don't care about revisions + discard the third element — that, rather than a second parser, is how a pinned ref is read + unpinned. + + An absent fragment resolves to :data:`LATEST_TAG` — a bare ``workspace/name`` means "the current + revision", never "unpinned". The fragment is returned verbatim: it may be a tag or a content + digest, and telling them apart is resolution's job, not parsing's. + """ + base, separator, fragment = root.partition(REF_FRAGMENT_SEPARATOR) + parsed = parse_entity_ref(base, default_workspace) + return parsed.workspace, parsed.name, fragment if separator and fragment else LATEST_TAG + + +class MetricRef(RootModel[str]): + """Reference to a persisted metric (format: ``workspace/name`` or ``name``).""" + + root: str = Field( + pattern=ENTITY_REF_PATTERN, + description="Reference to a stored metric (format: workspace/metric-name, or metric-name in the job workspace).", + ) + + +#: A wire metric is either an inline bundle DTO or a reference to a stored metric. Lives here (next to +#: ``MetricInline``) rather than in ``metric_refs`` so entity/DTO modules can use it without importing +#: the ref-resolution logic (which depends on ``entities`` and would cycle); ``metric_refs`` re-exports. +MetricRefOrInline: TypeAlias = MetricInline | MetricRef + + +class TaskRef(RootModel[str]): + """Reference to a persisted task (format: ``workspace/name``, ``name``, or either with a + ``#revision`` fragment). + + A taskset points at its member tasks by reference (there are no inline tasks), so a stored + taskset only ever holds refs. Unlike :class:`MetricRef`, a task ref may address a specific + revision via the platform's standard ``#`` sub-entity fragment. + + The fragment is optional *on input* and means :data:`LATEST_TAG` when absent — a bare + ``workspace/name`` is "the current revision", not "unpinned". It may name a tag or a content + digest. Anything **persisted** as a published snapshot must carry a resolved digest: tags move, + and a stored tag fragment would silently re-point published membership. + """ + + root: str = Field( + pattern=_SUBENTITY_REF_PATTERN, + description="Reference to a stored task (format: workspace/task-name, or task-name in the " + "taskset workspace), optionally pinned to a revision with '#'.", + ) + + +class TasksetRef(RootModel[str]): + """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=_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 '#'.", + ) + + +class TaskInputs(BaseModel): + """A task's recognized input fields. + + ``extra="forbid"``: only the field below is accepted. ``instruction`` is the agent's prompt; the + runtime falls back to the task ``intent`` when it is unset. + """ + + model_config = ConfigDict(extra="forbid") + + instruction: str | None = Field( + default=None, description="The agent's instruction (its prompt). Falls back to the task `intent` when unset." + ) + + +class MetadataItem(BaseModel): + """A single key/value annotation on a task.""" + + model_config = ConfigDict(extra="forbid") + + key: str = Field(description="Annotation key.") + value: str = Field(description="Annotation value.") + + +def _reject_duplicate_metadata_keys(items: list[MetadataItem]) -> list[MetadataItem]: + """Metadata is a key→value map expressed as a list; duplicate keys would silently collapse (e.g. + when folded into a mapping for the runtime), so reject them at validation rather than lose data.""" + seen: set[str] = set() + for item in items: + if item.key in seen: + raise ValueError(f"duplicate metadata key: {item.key!r}") + seen.add(item.key) + return items + + +#: A task's metadata: key/value annotations with unique keys (duplicates rejected at validation). +TaskMetadataList: TypeAlias = Annotated[list[MetadataItem], AfterValidator(_reject_duplicate_metadata_keys)] + + +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 + (it can't mean anything more than membership), so reject duplicates at validation.""" + seen: set[str] = set() + for ref in refs: + if ref.root in seen: + raise ValueError(f"duplicate task reference: {ref.root!r}") + seen.add(ref.root) + return refs + + +#: A list of task references with set semantics (order not significant, duplicates rejected). +TaskRefList: TypeAlias = Annotated[list[TaskRef], AfterValidator(_reject_duplicate_task_refs)] +#: Shape of a content digest in a ref fragment: full-length lowercase hex, never truncated. +_DIGEST_FRAGMENT_PATTERN = re.compile(DIGEST_PATTERN) + + +def _require_pinned_task_refs(refs: list[TaskRef]) -> list[TaskRef]: + """Every member of a *published* taskset revision must name an exact content digest. + + Enforced on the field rather than in the publish path so it cannot be bypassed by any other + writer. A ref that is bare (``workspace/name``) or tag-pinned (``#latest``, ``#candidate``) + resolves through a mutable pointer: the moment that tag moves, the published revision's + membership silently changes under it, and a "reproducible" dataset stops being reproducible. + Tags are resolution *inputs*, resolved to digests at publish time; only digests persist. + """ + for ref in refs: + _, _, fragment = parse_subentity_ref(ref.root, "") + if not _DIGEST_FRAGMENT_PATTERN.match(fragment): + raise ValueError( + f"task reference {ref.root!r} is not pinned to a content digest: a published taskset " + f"revision must reference an exact revision (got fragment {fragment!r}). Tags move; " + "resolve them to a digest before persisting." + ) + return refs + + +#: Member refs of a published taskset revision: set semantics *and* every ref digest-pinned. +PinnedTaskRefList: TypeAlias = Annotated[ + list[TaskRef], AfterValidator(_reject_duplicate_task_refs), AfterValidator(_require_pinned_task_refs) +] diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py index 22c894bf77..b6b3c5017f 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py @@ -5,23 +5,79 @@ from __future__ import annotations -import re from datetime import datetime from enum import StrEnum -from typing import Annotated, Any, Literal, TypeAlias +from typing import Annotated, TypeAlias -from nemo_evaluator.content_hash import DIGEST_PATTERN +from nemo_evaluator.api.fields import ( + LATEST_TAG as LATEST_TAG, +) +from nemo_evaluator.api.fields import ( + REF_FRAGMENT_CHARSET as REF_FRAGMENT_CHARSET, +) +from nemo_evaluator.api.fields import ( + REF_FRAGMENT_SEPARATOR as REF_FRAGMENT_SEPARATOR, +) +from nemo_evaluator.api.fields import ( + CloudpickleMetricPayload as CloudpickleMetricPayload, +) +from nemo_evaluator.api.fields import ( + InlineMetricPayload as InlineMetricPayload, +) +from nemo_evaluator.api.fields import ( + MetadataItem as MetadataItem, +) +from nemo_evaluator.api.fields import ( + MetricInline as MetricInline, +) +from nemo_evaluator.api.fields import ( + MetricPayload as MetricPayload, +) +from nemo_evaluator.api.fields import ( + MetricRef as MetricRef, +) +from nemo_evaluator.api.fields import ( + MetricRefOrInline as MetricRefOrInline, +) +from nemo_evaluator.api.fields import ( + PinnedTaskRefList as PinnedTaskRefList, +) +from nemo_evaluator.api.fields import ( + TaskInputs as TaskInputs, +) +from nemo_evaluator.api.fields import ( + TaskMetadataList as TaskMetadataList, +) +from nemo_evaluator.api.fields import ( + TaskRef as TaskRef, +) +from nemo_evaluator.api.fields import ( + TaskRefList as TaskRefList, +) +from nemo_evaluator.api.fields import ( + TasksetRef as TasksetRef, +) +from nemo_evaluator.api.fields import ( + parse_subentity_ref as parse_subentity_ref, +) +from nemo_evaluator.api.task_definitions.evaluator import EvaluatorTaskDefinition as EvaluatorTaskDefinition +from nemo_evaluator.api.task_definitions.harbor import HarborTaskDefinition as HarborTaskDefinition from nemo_evaluator.shared.metric_bundles.bundles import ( BundledMetricOutputSpec, - MetricMetadata, ) -from nemo_evaluator_sdk.agent_eval.tasks import SemanticView from nemo_evaluator_sdk.values.common import SecretRef from nemo_evaluator_sdk.values.results import AggregatedMetricResult from nemo_platform_plugin.api.filter import ComparisonOperation, FilterOperation, LogicalOperation from nemo_platform_plugin.api.parsed_filter import ENTITY_BASE_FIELDS +from nemo_platform_plugin.refs import ( + FILESET_REF_PATTERN as FILESET_REF_PATTERN, +) from nemo_platform_plugin.schema import DatetimeFilter, Filter -from pydantic import AfterValidator, BaseModel, ConfigDict, Field, RootModel, field_validator +from pydantic import BaseModel, ConfigDict, Field + +#: A stored task's content, discriminated by which runner executes it. Widen with more members as +#: runners land — the same way ``AgentRunnerTarget`` does on the target side. +TaskDefinition: TypeAlias = Annotated[EvaluatorTaskDefinition | HarborTaskDefinition, Field(discriminator="kind")] class DataFilter(Filter): @@ -52,206 +108,6 @@ def _walk(op: FilterOperation) -> FilterOperation: return _walk(operation) -class CloudpickleMetricPayload(BaseModel): - """Wire schema for a cloudpickle-serialized metric payload. - - Mirrors the runtime ``CloudpickleMetricPayload`` so the API contract is - explicit in the OpenAPI spec. The runtime bundle model serializes payloads - polymorphically (typed as an abstract base), which renders as an opaque - object in the spec; this concrete DTO documents the actual fields. - """ - - model_config = ConfigDict(extra="forbid", ser_json_bytes="base64", val_json_bytes="base64") - - kind: Literal["cloudpickle"] = Field(description="Payload format discriminator.") - python_version: str = Field(description="Python version the metric was pickled with (must match at execution).") - cloudpickle_version: str = Field(description="cloudpickle version used to serialize the metric.") - pickle_protocol: int = Field(description="Pickle protocol used.") - blob: bytes = Field(description="Base64-encoded cloudpickled metric object.") - digest: str | None = Field( - default=None, - description="SHA-256 digest of the payload bytes. Informational; recomputed server-side.", - ) - - -class InlineMetricPayload(BaseModel): - """Wire schema for an inline (config-serialized) metric payload. - - Mirrors the runtime ``InlineMetricPayload``. The metric is stored as its own - JSON configuration and reconstructed from the metric type union at execution, - so no code is shipped or executed on load. Used for platform-recognized - built-in metric types. - """ - - model_config = ConfigDict(extra="forbid") - - kind: Literal["inline"] = Field(description="Payload format discriminator.") - metric: dict[str, Any] = Field( - description="JSON-serialized built-in metric configuration, discriminated by its own `type`." - ) - digest: str | None = Field( - default=None, - description="SHA-256 digest of the canonical metric JSON. Informational; recomputed server-side.", - ) - - @field_validator("metric") - @classmethod - def _metric_must_declare_type(cls, value: dict[str, Any]) -> dict[str, Any]: - """Reject payloads without a metric ``type`` discriminator at the API boundary. - - The metric body stays an open object (the concrete shape is validated when - the bundle is hydrated against the metric type union), but a non-empty - ``type`` is required so malformed payloads fail fast rather than at execution. - """ - metric_type = value.get("type") - if not isinstance(metric_type, str) or not metric_type: - raise ValueError("inline metric payload must include a non-empty 'type'") - return value - - -# Discriminated on ``kind`` so additional payload formats can join the union -# without changing the field type. -MetricPayload = Annotated[CloudpickleMetricPayload | InlineMetricPayload, Field(discriminator="kind")] - - -class MetricInline(BaseModel): - """An executable metric submitted to the platform. - - Carries the bundled metric — type, metadata, output contracts, secret - references, and a format-specific payload — used both as the create-request - body and as an inline metric in an evaluation job. - """ - - model_config = ConfigDict(extra="forbid") - - bundle_kind: Literal["metric-bundle"] = "metric-bundle" - bundle_format_version: Literal["v1"] = "v1" - metric_type: str = Field(min_length=1, description="Runtime metric type name.") - metadata: MetricMetadata = Field(default_factory=MetricMetadata, description="User-facing metric metadata.") - outputs: list[BundledMetricOutputSpec] = Field(min_length=1, description="The metric's output contracts.") - secrets: dict[str, SecretRef] = Field( - default_factory=dict, description="Secret references required to execute the metric." - ) - payload: MetricPayload = Field(description="Format-specific serialized metric.") - - -# An entity reference is ``name`` or ``workspace/name``, each segment using the platform name charset. -# Shared by every ``workspace/name`` reference type (metrics, tasks). Enforced on the field so -# empty/malformed refs are rejected at validation rather than during parsing. -_ENTITY_REF_PATTERN = r"^[\w\-.]+(/[\w\-.]+)?$" - -#: The charset a ``#fragment`` may use. Exported because anything that *mints* a fragment — notably -#: revision tag names — has to be constrained by it: a value outside this set can be stored happily -#: and then never appear in a reference, which is a silent dead end rather than an error. -REF_FRAGMENT_CHARSET = r"[\w\-.]+" - -# A *sub-entity* reference adds an optional ``#fragment``, the platform's standard way of addressing -# something contained within an entity (filesets address a contained file the same way: -# ``workspace/fileset#path``). For a revisioned entity the fragment selects a revision — either a tag -# (``#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 -# 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. -REF_FRAGMENT_SEPARATOR = "#" - -#: The tag applied to every publish and used when a ref carries no fragment. -LATEST_TAG = "latest" - - -def parse_entity_ref(root: str, default_workspace: str) -> tuple[str, str]: - """Split a validated ``workspace/name`` (or bare ``name``) reference into ``(workspace, name)``. - - The ``workspace/name`` vs bare-``name`` shape is guaranteed by the field's ``_ENTITY_REF_PATTERN``, - so this only needs to split. Shared by every reference type (metrics, tasks); lives here — next to - the pattern, with no entity dependency — so ref-owning modules can reuse it without cycling. - - Any ``#fragment`` is stripped before splitting, so callers that don't care about revisions keep - working unchanged against a pinned ref. Use :func:`parse_subentity_ref` to read the fragment. - """ - base, _, _ = root.partition(REF_FRAGMENT_SEPARATOR) - workspace, separator, name = base.partition("/") - if separator: - return workspace, name - return default_workspace, base - - -def parse_subentity_ref(root: str, default_workspace: str) -> tuple[str, str, str]: - """Split a reference into ``(workspace, name, fragment)``. - - An absent fragment resolves to :data:`LATEST_TAG` — a bare ``workspace/name`` means "the current - revision", never "unpinned". The fragment is returned verbatim: it may be a tag or a content - digest, and telling them apart is resolution's job, not parsing's. - """ - base, separator, fragment = root.partition(REF_FRAGMENT_SEPARATOR) - workspace, name = parse_entity_ref(base, default_workspace) - return workspace, name, fragment if separator and fragment else LATEST_TAG - - -class MetricRef(RootModel[str]): - """Reference to a persisted metric (format: ``workspace/name`` or ``name``).""" - - root: str = Field( - pattern=_ENTITY_REF_PATTERN, - description="Reference to a stored metric (format: workspace/metric-name, or metric-name in the job workspace).", - ) - - -#: A wire metric is either an inline bundle DTO or a reference to a stored metric. Lives here (next to -#: ``MetricInline``) rather than in ``metric_refs`` so entity/DTO modules can use it without importing -#: the ref-resolution logic (which depends on ``entities`` and would cycle); ``metric_refs`` re-exports. -MetricRefOrInline: TypeAlias = MetricInline | MetricRef - - -class TaskRef(RootModel[str]): - """Reference to a persisted task (format: ``workspace/name``, ``name``, or either with a - ``#revision`` fragment). - - A taskset points at its member tasks by reference (there are no inline tasks), so a stored - taskset only ever holds refs. Unlike :class:`MetricRef`, a task ref may address a specific - revision via the platform's standard ``#`` sub-entity fragment. - - The fragment is optional *on input* and means :data:`LATEST_TAG` when absent — a bare - ``workspace/name`` is "the current revision", not "unpinned". It may name a tag or a content - digest. Anything **persisted** as a published snapshot must carry a resolved digest: tags move, - and a stored tag fragment would silently re-point published membership. - """ - - root: str = Field( - pattern=_SUBENTITY_REF_PATTERN, - description="Reference to a stored task (format: workspace/task-name, or task-name in the " - "taskset workspace), optionally pinned to a revision with '#'.", - ) - - -class TasksetRef(RootModel[str]): - """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=_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 '#'.", - ) - - class Metric(BaseModel): """API representation of a stored metric. @@ -346,44 +202,6 @@ class EvaluateResult(_ResultBase): metric_types: list[str] = Field(description="Runtime metric type names applied in the run.") -class TaskInputs(BaseModel): - """A task's recognized input fields. - - ``extra="forbid"``: only the field below is accepted. ``instruction`` is the agent's prompt; the - runtime falls back to the task ``intent`` when it is unset. - """ - - model_config = ConfigDict(extra="forbid") - - instruction: str | None = Field( - default=None, description="The agent's instruction (its prompt). Falls back to the task `intent` when unset." - ) - - -class MetadataItem(BaseModel): - """A single key/value annotation on a task.""" - - model_config = ConfigDict(extra="forbid") - - key: str = Field(description="Annotation key.") - value: str = Field(description="Annotation value.") - - -def _reject_duplicate_metadata_keys(items: list[MetadataItem]) -> list[MetadataItem]: - """Metadata is a key→value map expressed as a list; duplicate keys would silently collapse (e.g. - when folded into a mapping for the runtime), so reject them at validation rather than lose data.""" - seen: set[str] = set() - for item in items: - if item.key in seen: - raise ValueError(f"duplicate metadata key: {item.key!r}") - seen.add(item.key) - return items - - -#: A task's metadata: key/value annotations with unique keys (duplicates rejected at validation). -TaskMetadataList: TypeAlias = Annotated[list[MetadataItem], AfterValidator(_reject_duplicate_metadata_keys)] - - class Task(BaseModel): """API representation of a stored agent-eval task. @@ -396,16 +214,7 @@ class Task(BaseModel): name: str = Field(description="Task name — the stable task id, unique within its workspace.") workspace: str = Field(description="Workspace the task belongs to.") project: str | None = Field(default=None, description="The project associated with this task.") - intent: str = Field(description="Human-readable description of the desired agent behavior.") - inputs: TaskInputs = Field(default_factory=TaskInputs, description="The task's recognized input fields.") - metrics: list[MetricRef] = Field( - default_factory=list, - description="References to the metrics that score this task; inline metrics submitted on create " - "are normalized to (derived) stored metrics, so a stored task holds refs only.", - ) - views: dict[str, SemanticView] = Field( - default_factory=dict, description="Optional reporting views mapping metric outputs into named semantic scores." - ) + spec: TaskDefinition = Field(description="The task's content, discriminated by which runner executes it.") metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") revision: int = Field( description="Ordinal of the published revision this content corresponds to. Every stored task " @@ -430,14 +239,7 @@ class TaskInput(BaseModel): model_config = ConfigDict(extra="forbid") - intent: str = Field(description="Human-readable description of the desired agent behavior.") - inputs: TaskInputs = Field(default_factory=TaskInputs, description="The task's recognized input fields.") - metrics: list[MetricRefOrInline] = Field( - default_factory=list, description="Metrics that score this task — inline bundles and/or stored-metric refs." - ) - views: dict[str, SemanticView] = Field( - default_factory=dict, description="Optional reporting views mapping metric outputs into named semantic scores." - ) + spec: TaskDefinition = Field(description="The task's content, discriminated by which runner executes it.") metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") tags: list[str] = Field( default_factory=list, @@ -484,50 +286,6 @@ class TaskFilter(Filter): updated_at: DatetimeFilter | None = Field(None, description="Filter by update date.") -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 - (it can't mean anything more than membership), so reject duplicates at validation.""" - seen: set[str] = set() - for ref in refs: - if ref.root in seen: - raise ValueError(f"duplicate task reference: {ref.root!r}") - seen.add(ref.root) - return refs - - -#: A list of task references with set semantics (order not significant, duplicates rejected). -TaskRefList: TypeAlias = Annotated[list[TaskRef], AfterValidator(_reject_duplicate_task_refs)] - -#: Shape of a content digest in a ref fragment: full-length lowercase hex, never truncated. -_DIGEST_FRAGMENT_PATTERN = re.compile(DIGEST_PATTERN) - - -def _require_pinned_task_refs(refs: list[TaskRef]) -> list[TaskRef]: - """Every member of a *published* taskset revision must name an exact content digest. - - Enforced on the field rather than in the publish path so it cannot be bypassed by any other - writer. A ref that is bare (``workspace/name``) or tag-pinned (``#latest``, ``#candidate``) - resolves through a mutable pointer: the moment that tag moves, the published revision's - membership silently changes under it, and a "reproducible" dataset stops being reproducible. - Tags are resolution *inputs*, resolved to digests at publish time; only digests persist. - """ - for ref in refs: - _, _, fragment = parse_subentity_ref(ref.root, "") - if not _DIGEST_FRAGMENT_PATTERN.match(fragment): - raise ValueError( - f"task reference {ref.root!r} is not pinned to a content digest: a published taskset " - f"revision must reference an exact revision (got fragment {fragment!r}). Tags move; " - "resolve them to a digest before persisting." - ) - return refs - - -#: Member refs of a published taskset revision: set semantics *and* every ref digest-pinned. -PinnedTaskRefList: TypeAlias = Annotated[ - list[TaskRef], AfterValidator(_reject_duplicate_task_refs), AfterValidator(_require_pinned_task_refs) -] - - class Taskset(BaseModel): """API representation of a stored taskset — a flexible grouping of tasks with metadata. diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/service/task_service.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/service/task_service.py index 171bda0fae..2b44f76778 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/service/task_service.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/service/task_service.py @@ -17,14 +17,16 @@ from nemo_evaluator.api.schemas import ( LATEST_TAG, + HarborTaskDefinition, MetricInline, MetricRef, Revision, Task, + TaskDefinition, TaskInput, - parse_entity_ref, ) from nemo_evaluator.entities import TaskEntity, TaskRevisionEntity +from nemo_evaluator.metric_refs import parse_metric_ref from nemo_evaluator.revisions import ( apply_tag, get_revision, @@ -68,10 +70,7 @@ def _entity_to_task(entity: TaskEntity) -> Task: name=entity.name, workspace=entity.workspace, project=entity.project, - intent=entity.intent, - inputs=entity.inputs, - metrics=entity.metrics, - views=entity.views, + spec=entity.spec, metadata=entity.metadata, revision=entity.latest_revision, tags=entity.tags, @@ -95,10 +94,7 @@ def _revision_to_task(head: TaskEntity, revision: TaskRevisionEntity) -> Task: name=head.name, workspace=head.workspace, project=head.project, - intent=revision.intent, - inputs=revision.inputs, - metrics=revision.metrics, - views=revision.views, + spec=revision.spec, metadata=revision.metadata, revision=revision.revision, tags={tag: ordinal for tag, ordinal in head.tags.items() if ordinal == revision.revision}, @@ -158,7 +154,7 @@ async def _normalize_metrics(self, metrics: list[MetricRef | MetricInline], *, w refs: list[MetricRef] = [] for metric in metrics: if isinstance(metric, MetricRef): - ref_workspace, name = parse_entity_ref(metric.root, workspace) + ref_workspace, name = parse_metric_ref(metric.root, workspace) if await self.metric_service.get_metric(ref_workspace, name) is None: raise MetricRefNotFoundError( f"Metric reference '{metric.root}' not found. " @@ -170,12 +166,21 @@ async def _normalize_metrics(self, metrics: list[MetricRef | MetricInline], *, w refs.append(await self.metric_service.store_derived_metric(metric, workspace=workspace)) return refs + async def _normalize_spec(self, spec: TaskDefinition, *, workspace: str) -> TaskDefinition: + """Narrow a submitted spec to its stored form. + + Only the agent-eval variant changes: its inline metrics are offloaded to derived stored + metrics so a persisted task holds references only. A Harbor spec is already in stored form — + its archive was uploaded before the task was submitted. + """ + if isinstance(spec, HarborTaskDefinition): + return spec + # Same model in and out — only ``metrics`` narrows, from possibly-inline to references. + return spec.model_copy(update={"metrics": await self._normalize_metrics(spec.metrics, workspace=workspace)}) + async def _apply_content(self, entity: TaskEntity, task_input: TaskInput, *, workspace: str) -> TaskEntity: """Overwrite a head record's content from a request body (leaving revision pointers alone).""" - entity.intent = task_input.intent - entity.inputs = task_input.inputs - entity.metrics = await self._normalize_metrics(task_input.metrics, workspace=workspace) - entity.views = task_input.views + entity.spec = await self._normalize_spec(task_input.spec, workspace=workspace) entity.metadata = task_input.metadata return entity @@ -188,10 +193,13 @@ async def create_task( where ``published`` is always ``True`` here — a fresh task always cuts a revision. Use :meth:`replace_task` to publish a further revision of an existing task. """ - entity = await self._apply_content( - TaskEntity(name=name, workspace=workspace, project=project, intent=task_input.intent), - task_input, + # Normalize once: ``_apply_content`` would offload the same inline metrics a second time. + entity = TaskEntity( + name=name, workspace=workspace, + project=project, + spec=await self._normalize_spec(task_input.spec, workspace=workspace), + metadata=task_input.metadata, ) try: created = await self.entity_client.create(entity) 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 359fea7a24..b6fe457ca1 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 @@ -26,7 +26,6 @@ TaskRef, Taskset, TasksetInput, - parse_entity_ref, parse_subentity_ref, ) from nemo_evaluator.entities import TasksetEntity, TasksetRevisionEntity @@ -184,10 +183,14 @@ def _reject_duplicate_members(self, tasks: list[TaskRef], *, workspace: str) -> The field validator only catches byte-identical refs; this catches refs that differ in form but resolve to the same ``(workspace, name)`` — e.g. ``task-a`` and ``default/task-a`` in the ``default`` workspace. + + The revision fragment is deliberately discarded: two refs naming the same task at different + revisions are still the same member, and a taskset holding both would expand that task twice. """ seen: set[tuple[str, str]] = set() for ref in tasks: - resolved = parse_entity_ref(ref.root, workspace) + ref_workspace, name, _ = parse_subentity_ref(ref.root, workspace) + resolved = (ref_workspace, name) if resolved in seen: raise DuplicateTaskRefError( f"Task reference '{ref.root}' resolves to '{resolved[0]}/{resolved[1]}', already in this taskset" diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/task_definitions/evaluator.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/task_definitions/evaluator.py new file mode 100644 index 0000000000..584231cc64 --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/task_definitions/evaluator.py @@ -0,0 +1,49 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""The built-in task kind: an agent scored by platform metrics.""" + +from __future__ import annotations + +from typing import Any, Literal + +from nemo_evaluator.api.fields import MetricRefOrInline, TaskInputs +from nemo_evaluator_sdk.agent_eval.tasks import SemanticView +from pydantic import BaseModel, ConfigDict, Field + + +class EvaluatorTaskDefinition(BaseModel): + """What the agent should do, and how the platform scores it. + + ``metrics`` accepts inline bundles on the way in and holds references once stored: the service + offloads an inline metric to a content-addressed *derived* metric on create, so a persisted task + only ever names metrics it does not own. That narrowing is a service invariant rather than a + type-level one — a single model keeps the API surface small, at the cost of this field being + wider than what a stored task actually contains. + + Every field here is covered by the revision digest, ``reference`` included: it decides what a + metric grades against, so two revisions that score differently must not share a digest. Pinning + a revision therefore fixes the grading, not just the prompt. + """ + + model_config = ConfigDict(extra="forbid") + + kind: Literal["evaluator"] = Field(description="Task kind discriminator.") + intent: str = Field(description="Human-readable description of the desired agent behavior.") + inputs: TaskInputs = Field(default_factory=TaskInputs, description="The task's recognized input fields.") + metrics: list[MetricRefOrInline] = Field( + default_factory=list, + description="Metrics that score this task — stored-metric references, and inline bundles on " + "create (normalized to derived stored metrics before the task is persisted).", + ) + reference: dict[str, Any] = Field( + default_factory=dict, + description="Grader-only ground truth (held-out tests, expected outputs, rubric data). Surfaced to " + "metrics but never seeded into the agent's workspace or shown to the agent, so a metric can grade " + "against artifacts the agent cannot influence. Held out from the *agent*, not from the API: anyone " + "who can read the task can read this.", + ) + views: dict[str, SemanticView] = Field( + default_factory=dict, + description="Optional reporting views mapping metric outputs into named semantic scores.", + ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/task_definitions/harbor.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/task_definitions/harbor.py new file mode 100644 index 0000000000..6bc71e7144 --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/task_definitions/harbor.py @@ -0,0 +1,58 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""The Harbor task kind: a packaged task directory, run and scored by Harbor.""" + +from __future__ import annotations + +from typing import Any, Literal + +from nemo_evaluator.content_hash import DIGEST_LENGTH, DIGEST_PATTERN +from nemo_platform_plugin.refs import FILESET_REF_PATTERN +from pydantic import BaseModel, ConfigDict, Field + + +class HarborTaskDefinition(BaseModel): + """A reference to the task's packaged files, plus a projection of Harbor's own config. + + Harbor identifies a task by a *directory* — ``task.toml``, an instruction, an environment — so + what is stored is a reference to that directory's archive in the Files service, not the files + themselves. One fileset per task, so a task shared by several tasksets is stored once. The + archive is materialized back into ``//`` at run time, which is the layout + Harbor's own discovery expects. + + Which agent runs the task is *not* stored here. That comes from the run's target + (``HarborRunnerTarget``), so the same stored task can be evaluated against different agents. + Harbor's own ``[agent]`` block — carried inside ``config`` — configures how the agent *phase* + runs (timeout, user, network policy), not which agent it is. + """ + + model_config = ConfigDict(extra="forbid") + + kind: Literal["harbor"] = Field(description="Task kind discriminator.") + archive_ref: str = Field( + pattern=FILESET_REF_PATTERN, + description="Files reference to the task's packaged directory (format: workspace/fileset#path).", + ) + archive_digest: str = Field( + description="Content hash Harbor computed over the task directory. This is the authoritative " + "identity of a Harbor task's content — every file, including task.toml.", + min_length=DIGEST_LENGTH, + max_length=DIGEST_LENGTH, + pattern=DIGEST_PATTERN, + ) + instruction: str | None = Field( + default=None, description="The task's instruction text, when it has one (multi-step tasks may not)." + ) + # Excluded from the revision digest (see ``_DERIVED_SPEC_FIELDS`` in ``entities``). Safe only + # because this is never an execution input: Harbor reads the real ``task.toml`` out of the + # materialized archive, and ``archive_digest`` already covers every file in that directory. + # Hashing the projection too would add no coverage, and would make revision history sensitive to + # Harbor's serialization — a release that reordered keys would cut a revision for byte-identical + # files. Anything here that becomes a genuine execution or grading input must be digested. + config: dict[str, Any] = Field( + default_factory=dict, + description="Harbor's own task configuration (verifier, agent, environment, steps), as published. " + "A queryable projection of task.toml — inspect a task's verifier without downloading the " + "archive. Opaque here: Harbor owns this schema.", + ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/content_hash.py b/plugins/nemo-evaluator/src/nemo_evaluator/content_hash.py index 78dd55e9b8..df245c0470 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/content_hash.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/content_hash.py @@ -35,7 +35,8 @@ import hashlib import json -from collections.abc import Set +from collections.abc import Mapping, Set +from typing import Any from nemo_platform_plugin.entities import EntityBase @@ -46,7 +47,16 @@ DIGEST_PATTERN = r"^[0-9a-f]{64}$" -def canonical_payload(entity: EntityBase, *, exclude: Set[str] | None = None) -> str: +def _as_exclude_map(exclude: Set[str] | Mapping[str, Any] | None) -> dict[str, Any]: + """Normalize either accepted ``exclude`` form to pydantic's dict form.""" + if exclude is None: + return {} + if isinstance(exclude, Mapping): + return {str(name): nested for name, nested in exclude.items()} + return {str(name): True for name in exclude} + + +def canonical_payload(entity: EntityBase, *, exclude: Set[str] | Mapping[str, Any] | None = None) -> str: """Return the canonical JSON serialization that :func:`content_hash` digests. Exposed separately because it is the actual compatibility contract: if this string changes @@ -78,12 +88,15 @@ def canonical_payload(entity: EntityBase, *, exclude: Set[str] | None = None) -> their own revision/tag bookkeeping here — a revision's digest must not cover the revision index that was assigned *because of* that digest. """ - excluded = set(entity.__base_fields__) | set(exclude or ()) + # Pydantic's dict form lets a caller exclude a *nested* field (``{"spec": {"config"}}``), which + # a flat set cannot express. Both forms are accepted so simple cases stay simple. + excluded: dict[str, Any] = {str(name): True for name in entity.__base_fields__} + excluded.update(_as_exclude_map(exclude)) payload = entity.model_dump(exclude=excluded, exclude_computed_fields=True, mode="json") return json.dumps(payload, sort_keys=True, separators=(",", ":")) -def content_hash(entity: EntityBase, *, exclude: Set[str] | None = None) -> str: +def content_hash(entity: EntityBase, *, exclude: Set[str] | Mapping[str, Any] | None = None) -> str: """Return the full 64-char lowercase hex SHA-256 digest of an entity's content. See :func:`canonical_payload` for what is and is not included. diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/entities.py b/plugins/nemo-evaluator/src/nemo_evaluator/entities.py index cd2404db77..a32c66ec1f 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/entities.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/entities.py @@ -20,15 +20,13 @@ from typing import ClassVar from nemo_evaluator.api.schemas import ( - MetricRef, PinnedTaskRefList, - TaskInputs, + TaskDefinition, TaskMetadataList, TaskRefList, ) from nemo_evaluator.content_hash import DIGEST_LENGTH, DIGEST_PATTERN from nemo_evaluator.shared.metric_bundles.bundles import BundledMetricOutputSpec -from nemo_evaluator_sdk.agent_eval.tasks import SemanticView from nemo_evaluator_sdk.values.common import SecretRef from nemo_evaluator_sdk.values.results import AggregatedMetricResult from nemo_platform_plugin.entities import EntityBase @@ -52,6 +50,39 @@ #: neither itself nor the ordinal that was assigned because of it. REVISION_SELF_FIELDS = frozenset({"content_hash", "revision"}) +#: Spec fields excluded from the revision digest. +#: +#: The rule for what belongs *in* the digest: any field that affects the output of a task's +#: execution or the mechanism used to grade it. A field may only be excluded if it is a derived +#: view of content the digest already covers by another route. +#: +#: ``HarborTaskDefinition.config`` qualifies. It is a projection of ``task.toml``, which lives +#: inside the archive, and Harbor reads the real ``task.toml`` out of the materialized archive at +#: run time — this copy is never an execution input, only a queryable convenience. ``archive_digest`` +#: is authoritative over every file in that directory including ``task.toml``, so a config change +#: that actually alters execution or grading already moves the digest. Hashing the projection too +#: would add no coverage and would make revision history sensitive to Harbor's serialization: a +#: release that reordered keys or emitted a new defaulted field would cut a revision for +#: byte-identical files. +#: +#: That makes ``archive_digest`` load-bearing. If a Harbor field ever becomes an execution input in +#: its own right — read from the stored record rather than from the archive — it must be digested. +_DERIVED_SPEC_FIELDS = {"config"} + +#: What a *head* record excludes when digesting: its revision pointers, plus derived spec fields. +#: Nested form, because the derived fields live inside ``spec``. +REVISION_POINTER_EXCLUDE: dict[str, object] = { + **dict.fromkeys(REVISION_POINTER_FIELDS, True), + "spec": set(_DERIVED_SPEC_FIELDS), +} + +#: The mirror for a *revision* record. Both must exclude the same derived fields, or the head and +#: its revision would digest differently and publish-time dedup would never fire. +REVISION_SELF_EXCLUDE: dict[str, object] = { + **dict.fromkeys(REVISION_SELF_FIELDS, True), + "spec": set(_DERIVED_SPEC_FIELDS), +} + class MetricBundleEntity(EntityBase): """Persisted index for a stored metric, addressed by workspace/name. @@ -209,27 +240,23 @@ class _RevisionedCommon(BaseModel): class TaskEntity(_RevisionedCommon, EntityBase): - """Persisted, queryable agent-eval task, addressed by workspace/name. - - Maps to the SDK :class:`~nemo_evaluator_sdk.agent_eval.tasks.AgentEvalTask`: the task's stable - ``id`` is the record ``name``, and ``metrics`` are stored in their wire form (inline bundles - and/or references to stored metrics) so a task can reference curated metrics or carry its own; - references resolve to inline runtime metrics when the task is run. + """Persisted, queryable task, addressed by workspace/name. + + A task is an evaluation unit; ``spec`` says what it is and which runner executes it. Both kinds + live in one record type so a user manages every evaluation unit in one place, and so a taskset + can group them without caring how each one runs — the same way ``AgentRunnerTarget`` already + treats codex/fabric/harbor as members of one union on the target side. + + Content is nested under ``spec`` rather than flattened with nullable per-kind fields, so each + variant's required fields stay genuinely required and the revision digest covers the spec as one + unit. An agent-eval task's ``metrics`` are stored as references (inline metrics submitted on + create are normalized to derived stored metrics); a Harbor task's files live in a fileset, and + the spec holds a reference to them. """ __entity_type__: ClassVar[str] = "task" - intent: str = Field(description="Human-readable description of the desired agent behavior.") - inputs: TaskInputs = Field(default_factory=TaskInputs, description="The task's recognized input fields.") - metrics: list[MetricRef] = Field( - default_factory=list, - description="References to the metrics that score this task. Inline metrics submitted with the " - "task are normalized to (derived) stored metrics, so a persisted task only ever holds refs.", - ) - views: dict[str, SemanticView] = Field( - default_factory=dict, - description="Optional reporting views mapping this task's metric outputs into named semantic scores.", - ) + spec: TaskDefinition = Field(description="The task's content, discriminated by which runner executes it.") metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") @@ -320,16 +347,7 @@ class TaskRevisionEntity(_RevisionCommon, EntityBase): __entity_type__: ClassVar[str] = "task_revision" - intent: str = Field(description="Human-readable description of the desired agent behavior.") - inputs: TaskInputs = Field(default_factory=TaskInputs, description="The task's recognized input fields.") - metrics: list[MetricRef] = Field( - default_factory=list, - description="References to the metrics that score this task, as of this revision.", - ) - views: dict[str, SemanticView] = Field( - default_factory=dict, - description="Reporting views mapping this task's metric outputs into named semantic scores.", - ) + spec: TaskDefinition = Field(description="The task's content as of this revision.") metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py b/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py index e46f6901f7..b9d922b75c 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py @@ -16,21 +16,25 @@ # entity/DTO modules can reference them without importing this module's entities-dependent resolution # logic (which would create an import cycle). Imported here for use below and re-exported for the # existing ``nemo_evaluator.metric_refs`` import sites. -from nemo_evaluator.api.schemas import MetricRef, MetricRefOrInline, parse_entity_ref +from nemo_evaluator.api.schemas import MetricRef, MetricRefOrInline from nemo_evaluator.entities import MetricBundleEntity from nemo_evaluator.metric_storage import load_bundle from nemo_evaluator.shared.metric_bundles.bundles import MetricBundle from nemo_platform import AsyncNeMoPlatform from nemo_platform_plugin.entity_client import NemoEntityGetterProtocol, NemoEntityNotFoundError +from nemo_platform_plugin.refs import parse_entity_ref def parse_metric_ref(root: str, default_workspace: str) -> tuple[str, str]: """Split a validated metric reference into ``(workspace, name)``. - Thin alias over the shared :func:`~nemo_evaluator.api.schemas.parse_entity_ref` (all - ``workspace/name`` refs split identically); kept for the existing ``metric_refs`` call sites. + Thin alias over the platform's :func:`~nemo_platform_plugin.refs.parse_entity_ref` (all + ``workspace/name`` refs split identically); kept for the existing ``metric_refs`` call sites, + which want a tuple. A metric ref carries no ``#fragment`` — metrics are not revisioned — so the + plain entity parser is the right one here. """ - return parse_entity_ref(root, default_workspace) + parsed = parse_entity_ref(root, default_workspace) + return parsed.workspace, parsed.name async def resolve_metric_ref( diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/revisions.py b/plugins/nemo-evaluator/src/nemo_evaluator/revisions.py index 5b02566018..40555bc483 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/revisions.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/revisions.py @@ -32,8 +32,9 @@ from nemo_evaluator.api.schemas import LATEST_TAG, REF_FRAGMENT_CHARSET from nemo_evaluator.content_hash import DIGEST_PATTERN, content_hash from nemo_evaluator.entities import ( + REVISION_POINTER_EXCLUDE, REVISION_POINTER_FIELDS, - REVISION_SELF_FIELDS, + REVISION_SELF_EXCLUDE, TaskEntity, TaskRevisionEntity, TasksetEntity, @@ -117,7 +118,7 @@ def head_digest(head: EntityBase) -> str: The exclusion is what makes this comparable to a revision's own digest: pointers describe *which* content is current, not what the content is. """ - return content_hash(head, exclude=REVISION_POINTER_FIELDS) + return content_hash(head, exclude=REVISION_POINTER_EXCLUDE) def validate_tag_name(tag: str) -> str: @@ -293,7 +294,7 @@ def _verify_content(head: TaskEntity | TasksetEntity, revision: TaskRevisionEnti write to one. This turns that convention into something detectable rather than something the reader has to assume. """ - actual = content_hash(revision, exclude=REVISION_SELF_FIELDS) + actual = content_hash(revision, exclude=REVISION_SELF_EXCLUDE) if actual != revision.content_hash: raise RevisionContentMismatchError( f"revision {revision.revision} of '{head.workspace}/{head.name}' does not match its " @@ -394,6 +395,8 @@ async def publish_revision( return current, head, False ordinal = head.latest_revision + 1 + # Copy *all* content, including fields the digest excludes: a revision stores the full + # published spec, and only its hash ignores the derived parts. content = head.model_dump(exclude=set(REVISION_POINTER_FIELDS) | set(head.__base_fields__), mode="json") revision = revision_type( name=revision_name(ordinal), diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/task_refs.py b/plugins/nemo-evaluator/src/nemo_evaluator/task_refs.py index d7788799bd..1fccec90e8 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/task_refs.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/task_refs.py @@ -18,7 +18,7 @@ from typing import cast -from nemo_evaluator.api.schemas import TasksetRef, parse_subentity_ref +from nemo_evaluator.api.schemas import EvaluatorTaskDefinition, 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 @@ -26,6 +26,19 @@ from nemo_platform_plugin.entity_client import NemoEntityNotFoundError +class UnsupportedTaskKindError(ValueError): + """A taskset member's runner kind cannot be executed by the requested target. + + A taskset may group tasks of different kinds — that is the point of managing every evaluation + unit in one place — but a single run has one target, so expansion is where a mismatch surfaces. + + Raised during ``to_spec``, which the job submit path wraps: any exception there becomes a 422 + carrying this message (``_apply_transformer`` in ``nemo_platform_plugin.jobs.api_factory``). It + subclasses ``ValueError`` for local callers that catch it deliberately, not to obtain that + mapping — the mapping is a catch-all and would apply to any exception type. + """ + + def _entity_to_task_input(entity: TaskEntity, revision: TaskRevisionEntity) -> AgentEvalTaskInput: """Project a stored task's *published revision* onto the submitter-facing inline task DTO. @@ -36,15 +49,32 @@ def _entity_to_task_input(entity: TaskEntity, revision: TaskRevisionEntity) -> A A stored task holds metric *references* (inline metrics were normalized to derived stored metrics on create); those resolve to inline bundles in the shared metric-ref pass that runs - after expansion. A stored task carries no grader-only ``reference`` (the entity has no such - field), so taskset-driven tasks run with an empty one. + after expansion. The grader-only ``reference`` comes from the revision too, so a taskset-driven + run grades against the ground truth that revision pinned — held-out data is not the privilege of + inline submissions. """ + spec = revision.spec + if not isinstance(spec, EvaluatorTaskDefinition): + # A Harbor task's content is a *directory of files*, not fields — the runner needs the + # archive materialized on disk, which this pure projection cannot do. Rejecting here means a + # mismatched taskset fails before the run rather than silently evaluating an empty task. + # + # Deliberately does *not* suggest picking a different target: no target can run a stored + # task of this kind yet, so pointing at one would send the reader in circles. Storing the + # kind landed ahead of the execution bridge (AALGO-481). + raise UnsupportedTaskKindError( + f"Task '{entity.workspace}/{entity.name}' is a {spec.kind!r} task. Running a stored " + f"{spec.kind!r} task is not supported yet — no target can execute one, so this taskset " + "cannot be evaluated until that lands. Remove the member, or submit an " + "'evaluator'-kind taskset." + ) return AgentEvalTaskInput( id=entity.name, - intent=revision.intent, - inputs=revision.inputs, - metrics=list(revision.metrics), - views=revision.views, + intent=spec.intent, + inputs=spec.inputs, + reference=dict(spec.reference), + metrics=list(spec.metrics), + views=spec.views, metadata=revision.metadata, ) diff --git a/plugins/nemo-evaluator/tests/api/service/test_task_service.py b/plugins/nemo-evaluator/tests/api/service/test_task_service.py index 46b117f50d..b8cfa3ebcf 100644 --- a/plugins/nemo-evaluator/tests/api/service/test_task_service.py +++ b/plugins/nemo-evaluator/tests/api/service/test_task_service.py @@ -4,7 +4,16 @@ from __future__ import annotations import pytest -from nemo_evaluator.api.schemas import MetadataItem, MetricInline, MetricRef, Task, TaskInput, TaskInputs +from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, + HarborTaskDefinition, + MetadataItem, + MetricInline, + MetricRef, + Task, + TaskInput, + TaskInputs, +) from nemo_evaluator.api.service.task_service import MetricRefNotFoundError, TaskService from nemo_evaluator.shared.metric_bundles.bundles import bundle_metric from nemo_evaluator.shared.metric_bundles.cloudpickle import CloudpickleMetricBundlePackager @@ -13,10 +22,17 @@ class _FakeMetricService: - """Records inline-metric normalization so we can assert a task stores refs, not bundles.""" + """Records both metric-service entry points. + + ``stored`` covers inline-metric normalization, so a test can assert a task stores refs rather + than bundles. ``looked_up`` covers ref validation — recorded separately because "this task never + touched the metric service" is a claim about *both* calls, and asserting only on ``stored`` + would leave a lookup-only path silently passing. + """ def __init__(self, existing: set[tuple[str, str]] | None = None) -> None: self.stored: list[MetricInline] = [] + self.looked_up: list[tuple[str, str]] = [] self.existing = existing if existing is not None else {("default", "stored-metric")} async def store_derived_metric(self, metric: MetricInline, *, workspace: str) -> MetricRef: @@ -24,6 +40,7 @@ async def store_derived_metric(self, metric: MetricInline, *, workspace: str) -> return MetricRef(f"{workspace}/derived.{metric.payload.digest}") async def get_metric(self, workspace: str, name: str) -> object | None: + self.looked_up.append((workspace, name)) return object() if (workspace, name) in self.existing else None @@ -35,11 +52,37 @@ def _inline_metric() -> MetricInline: return MetricInline.model_validate(bundle.model_dump(mode="json")) +def _evaluator_spec(task: Task) -> EvaluatorTaskDefinition: + """Narrow ``Task.spec`` to the evaluator variant before reading a variant-specific field. + + ``spec`` is a discriminated union, so a test that reads ``intent``/``metrics``/``reference`` has + to say which kind it expects. Asserting it rather than assuming it means a change that routed + the wrong variant here fails on the kind, not with an ``AttributeError`` mid-assertion. + """ + assert isinstance(task.spec, EvaluatorTaskDefinition) + return task.spec + + +def _harbor_spec(task: Task) -> HarborTaskDefinition: + """The Harbor half of :func:`_evaluator_spec`.""" + assert isinstance(task.spec, HarborTaskDefinition) + return task.spec + + +def _ref(metric: MetricRef | MetricInline) -> MetricRef: + """Narrow a stored task's metric to a reference — inline bundles are offloaded on create.""" + assert isinstance(metric, MetricRef) + return metric + + def _task_input() -> TaskInput: return TaskInput( - intent="Answer the question.", - inputs=TaskInputs(instruction="What is 2+2?"), - metrics=[MetricRef("default/stored-metric")], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the question.", + inputs=TaskInputs(instruction="What is 2+2?"), + metrics=[MetricRef("default/stored-metric")], + ), metadata=[MetadataItem(key="suite", value="smoke")], ) @@ -60,8 +103,8 @@ async def test_create_then_get(service: TaskService) -> None: assert isinstance(created, Task) assert created.name == "task-1" assert created.id == "task-task-1" - assert created.intent == "Answer the question." - assert isinstance(created.metrics[0], MetricRef) + assert _evaluator_spec(created).intent == "Answer the question." + assert isinstance(_evaluator_spec(created).metrics[0], MetricRef) assert created.created_at is not None got = await service.get_task("default", "task-1") @@ -73,9 +116,12 @@ async def test_create_normalizes_inline_metrics_to_refs( ) -> None: inline = _inline_metric() task_input = TaskInput( - intent="Answer the question.", - inputs=TaskInputs(instruction="What is 2+2?"), - metrics=[MetricRef("default/stored-metric"), inline], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the question.", + inputs=TaskInputs(instruction="What is 2+2?"), + metrics=[MetricRef("default/stored-metric"), inline], + ) ) created, _ = await service.create_task("task-1", task_input, workspace="default") @@ -83,22 +129,55 @@ async def test_create_normalizes_inline_metrics_to_refs( # The inline metric was offloaded to the metric service (stored as a derived metric)... assert metric_service.stored == [inline] # ...and the persisted task holds only refs — the passthrough ref plus the derived one. - assert all(isinstance(m, MetricRef) for m in created.metrics) - assert created.metrics[0].root == "default/stored-metric" - assert created.metrics[1].root == f"default/derived.{inline.payload.digest}" + assert all(isinstance(m, MetricRef) for m in _evaluator_spec(created).metrics) + assert _ref(_evaluator_spec(created).metrics[0]).root == "default/stored-metric" + assert _ref(_evaluator_spec(created).metrics[1]).root == f"default/derived.{inline.payload.digest}" + + +async def test_create_preserves_grader_only_reference(service: TaskService) -> None: + """Normalization narrows ``metrics`` and must leave the rest of the spec alone. + + ``_normalize_spec`` rebuilds the spec with ``model_copy(update=...)``, so a field it does not + name rides along untouched — this pins that, since silently dropping ground truth would leave + metrics grading against nothing while the run still reported a score. + """ + reference = {"expected": "Paris", "held_out_tests": ["test_capital.py"]} + task_input = TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the question.", + inputs=TaskInputs(instruction="What is the capital of France?"), + reference=reference, + metrics=[_inline_metric()], + ) + ) + + created, _ = await service.create_task("task-1", task_input, workspace="default") + + assert _evaluator_spec(created).reference == reference, "normalizing metrics must not disturb the reference" + got = await service.get_task("default", "task-1") + assert got is not None and _evaluator_spec(got).reference == reference async def test_create_rejects_missing_metric_ref(service: TaskService) -> None: - task_input = TaskInput(intent="x", inputs=TaskInputs(instruction="?"), metrics=[MetricRef("default/nope")]) + task_input = TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", intent="x", inputs=TaskInputs(instruction="?"), metrics=[MetricRef("default/nope")] + ) + ) with pytest.raises(MetricRefNotFoundError, match="not found"): await service.create_task("task-1", task_input, workspace="default") async def test_create_canonicalizes_bare_metric_ref(service: TaskService) -> None: # A bare "stored-metric" ref resolves against the task workspace and is persisted as "default/stored-metric". - task_input = TaskInput(intent="x", inputs=TaskInputs(instruction="?"), metrics=[MetricRef("stored-metric")]) + task_input = TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", intent="x", inputs=TaskInputs(instruction="?"), metrics=[MetricRef("stored-metric")] + ) + ) created, _ = await service.create_task("task-1", task_input, workspace="default") - assert created.metrics[0].root == "default/stored-metric" + assert _ref(_evaluator_spec(created).metrics[0]).root == "default/stored-metric" async def test_create_rejects_duplicate(service: TaskService) -> None: @@ -195,7 +274,12 @@ async def _boom(entity): entity_store.create = _boom changed = TaskInput( - intent="Rewritten.", inputs=TaskInputs(instruction="?"), metrics=[MetricRef("default/stored-metric")] + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Rewritten.", + inputs=TaskInputs(instruction="?"), + metrics=[MetricRef("default/stored-metric")], + ) ) with pytest.raises(RuntimeError): await service.replace_task("task-1", changed, workspace="default") @@ -203,7 +287,7 @@ async def _boom(entity): entity_store.create = real_create.__get__(entity_store) head = await service.get_task("default", "task-1") assert head is not None - assert head.intent == "Answer the question.", "the head must still hold the last published content" + assert _evaluator_spec(head).intent == "Answer the question.", "the head must still hold the last published content" async def test_tag_revision_returns_none_for_a_missing_task(service: TaskService) -> None: @@ -273,7 +357,7 @@ async def test_resolve_revision_honours_a_tag_naming_an_older_revision(service: await service.tag_revision("default", "task-1", "blessed", "latest") revised = _task_input() - revised.intent = "Answer differently." + revised.spec.intent = "Answer differently." await service.replace_task("task-1", revised, workspace="default") latest = await service.resolve_revision("default", "task-1") @@ -289,7 +373,7 @@ async def test_resolve_revision_round_trips_a_digest_fragment(service: TaskServi first = (await service.list_revisions("default", "task-1")).data[0].content_hash revised = _task_input() - revised.intent = "Answer differently." + revised.spec.intent = "Answer differently." await service.replace_task("task-1", revised, workspace="default") assert await service.resolve_revision("default", "task-1", first) == first @@ -300,3 +384,134 @@ async def test_resolve_revision_raises_for_a_missing_task(service: TaskService) member that does not exist, now that the separate existence check is gone.""" with pytest.raises(NemoEntityNotFoundError): await service.resolve_revision("default", "nope") + + +# --- Harbor-kind tasks -------------------------------------------------------- + + +def _harbor_input(digest: str = "a" * 64) -> TaskInput: + return TaskInput( + spec=HarborTaskDefinition( + kind="harbor", + archive_ref="default/harbor-tasks#packages/org-name/abc/dist.tar.gz", + archive_digest=digest, + instruction="Fix the failing test.", + config={"verifier": {"type": "pytest"}}, + ), + metadata=[MetadataItem(key="suite", value="swe")], + ) + + +async def test_stores_a_harbor_task(service: TaskService) -> None: + """Both kinds live in one record type, so a user manages every evaluation unit in one place.""" + created, published = await service.create_task("fix-test", _harbor_input(), workspace="default") + + assert published + assert created.spec.kind == "harbor" + assert created.spec.archive_ref.endswith("dist.tar.gz") + assert created.spec.config == {"verifier": {"type": "pytest"}} + + +async def test_harbor_task_publishes_revisions_like_any_other(service: TaskService) -> None: + await service.create_task("fix-test", _harbor_input(), workspace="default") + + same, published_again = await service.replace_task("fix-test", _harbor_input(), workspace="default") + assert not published_again, "identical content must not cut a revision" + + changed, published = await service.replace_task("fix-test", _harbor_input(digest="b" * 64), workspace="default") + assert published and changed.revision == 2 + + +async def test_a_harbor_task_never_reaches_the_metric_service( + service: TaskService, metric_service: _FakeMetricService +) -> None: + """Metric normalization is agent-eval-specific: a Harbor task is scored by Harbor's own reward, + and its spec arrives already in stored form. + + Both entry points, not just the write: a Harbor spec must not be validated against stored + metrics either, so ``_normalize_spec`` has to short-circuit before ref resolution rather than + merely find nothing to offload. + """ + await service.create_task("fix-test", _harbor_input(), workspace="default") + assert metric_service.stored == [] + assert metric_service.looked_up == [] + + +async def test_kinds_with_matching_metadata_do_not_share_a_digest(service: TaskService) -> None: + """The revision digest covers the whole spec, so two kinds cannot collide on content.""" + harbor, _ = await service.create_task("a", _harbor_input(), workspace="default") + agent, _ = await service.create_task("b", _task_input(), workspace="default") + + harbor_revisions = await service.list_revisions("default", "a") + agent_revisions = await service.list_revisions("default", "b") + assert harbor_revisions is not None and agent_revisions is not None + assert harbor_revisions.data[0].content_hash != agent_revisions.data[0].content_hash + + +async def test_harbor_config_is_stored_but_not_hashed(service: TaskService) -> None: + """`config` is a projection of task.toml, which lives inside the archive — a real change moves + `archive_digest`. Hashing the projection too would make our history sensitive to Harbor's + serialization: a release that reordered keys would cut a revision for byte-identical files.""" + await service.create_task("fix-test", _harbor_input(), workspace="default") + + reserialized = TaskInput( + spec=HarborTaskDefinition( + kind="harbor", + archive_ref="default/harbor-tasks#packages/org-name/abc/dist.tar.gz", + archive_digest="a" * 64, + instruction="Fix the failing test.", + config={"verifier": {"type": "pytest"}, "added_by_a_new_harbor_release": True}, + ), + metadata=[MetadataItem(key="suite", value="swe")], + ) + same, published = await service.replace_task("fix-test", reserialized, workspace="default") + + assert not published, "a config-only change must not cut a revision" + assert same.revision == 1 + + # ...but the new config is *persisted*, so the queryable projection stays current. Re-read + # rather than trusting the returned object: `replace_task` builds its result from the head it + # already mutated in memory, so asserting on `same` would pass even if nothing were written. + # On this path the write is a lone `entity_client.update` whose comment justifies it by + # `project` alone — drop it as a redundant round trip and only a re-read notices. + refetched = await service.get_task("default", "fix-test") + assert refetched is not None + assert _harbor_spec(refetched).config["added_by_a_new_harbor_release"] is True + + +async def test_reference_only_change_publishes_a_revision(service: TaskService) -> None: + """The mirror of the Harbor ``config`` case, and the reason the two differ. + + ``config`` is excluded because it is a projection of content ``archive_digest`` already covers. + ``reference`` is nothing of the sort: it is the ground truth a metric grades against, so a task + whose reference changed scores differently and must be a distinct revision. Deduping it onto the + old digest would let a pinned taskset silently re-grade. + """ + + def _graded(expected: str) -> TaskInput: + return TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the question.", + inputs=TaskInputs(instruction="What is the capital of France?"), + reference={"expected": expected}, + metrics=[MetricRef("default/stored-metric")], + ) + ) + + await service.create_task("capital", _graded("Paris"), workspace="default") + + same, published_again = await service.replace_task("capital", _graded("Paris"), workspace="default") + assert not published_again and same.revision == 1, "identical content must still dedup" + + changed, published = await service.replace_task("capital", _graded("Lyon"), workspace="default") + assert published, "changing the ground truth must cut a new revision" + assert changed.revision == 2 + assert _evaluator_spec(changed).reference == {"expected": "Lyon"} + + +async def test_a_real_archive_change_does_cut_a_revision(service: TaskService) -> None: + """The flip side: `archive_digest` is the authoritative identity, so it must still move.""" + await service.create_task("fix-test", _harbor_input(), workspace="default") + changed, published = await service.replace_task("fix-test", _harbor_input(digest="b" * 64), workspace="default") + assert published and changed.revision == 2 diff --git a/plugins/nemo-evaluator/tests/api/v2/test_tasks_routes.py b/plugins/nemo-evaluator/tests/api/v2/test_tasks_routes.py index d8a71339d7..c7b3861471 100644 --- a/plugins/nemo-evaluator/tests/api/v2/test_tasks_routes.py +++ b/plugins/nemo-evaluator/tests/api/v2/test_tasks_routes.py @@ -13,7 +13,14 @@ from fastapi import FastAPI from fastapi.testclient import TestClient from nemo_evaluator.api.dependencies import get_task_service -from nemo_evaluator.api.schemas import MetricInline, MetricRef, TaskInput, TaskInputs +from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, + HarborTaskDefinition, + MetricInline, + MetricRef, + TaskInput, + TaskInputs, +) from nemo_evaluator.api.service.task_service import TaskService from nemo_evaluator.api.v2 import tasks as tasks_routes from nemo_platform_plugin.entity_client import NemoEntityConflictError @@ -41,9 +48,12 @@ def client(entity_store) -> TestClient: def _body(*, intent: str = "Answer the question.", tags: list[str] | None = None) -> dict: return TaskInput( - intent=intent, - inputs=TaskInputs(instruction="What is 2+2?"), - metrics=[MetricRef("default/stored-metric")], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent=intent, + inputs=TaskInputs(instruction="What is 2+2?"), + metrics=[MetricRef("default/stored-metric")], + ), tags=tags or [], ).model_dump(mode="json") @@ -59,14 +69,14 @@ def test_create_then_get(client: TestClient) -> None: got = client.get(f"{_BASE}/task-1") assert got.status_code == 200 body = got.json() - assert body["intent"] == "Answer the question." - assert body["metrics"] == ["default/stored-metric"] # MetricRef serializes to a bare string + assert body["spec"]["intent"] == "Answer the question." + assert body["spec"]["metrics"] == ["default/stored-metric"] # MetricRef serializes to a bare string def test_create_rejects_unrecognized_input_key(client: TestClient) -> None: # inputs is a strict TaskInputs (extra="forbid") — an unknown key is a 422, not silently stored. body = _body() - body["inputs"]["expected"] = "4" + body["spec"]["inputs"]["expected"] = "4" assert client.post(f"{_BASE}/task-1", json=body).status_code == 422 @@ -79,10 +89,29 @@ def test_create_rejects_duplicate_metadata_keys(client: TestClient) -> None: def test_create_missing_metric_ref_returns_422(client: TestClient) -> None: body = _body() - body["metrics"] = ["default/missing-metric"] + body["spec"]["metrics"] = ["default/missing-metric"] assert client.post(f"{_BASE}/task-1", json=body).status_code == 422 +@pytest.mark.parametrize("method", ["post", "put"]) +def test_write_without_a_spec_kind_returns_422(client: TestClient, method: str) -> None: + """``spec`` is a discriminated union, so a raw body that omits ``kind`` has no variant to + validate against. ``kind`` is therefore required in both definitions — a schema that defaulted + it would tell a generated client it may be omitted, and every such request would 422.""" + body = _body() + del body["spec"]["kind"] + response = getattr(client, method)(f"{_BASE}/task-1", json=body) + assert response.status_code == 422 + assert response.json()["detail"][0]["type"] == "union_tag_not_found" + + +@pytest.mark.parametrize("definition", [EvaluatorTaskDefinition, HarborTaskDefinition]) +def test_published_schema_requires_the_kind_discriminator(definition: type) -> None: + """The generated spec has to agree with the validator above: a default on ``kind`` would leave + it out of ``required``, and a client generated from that spec would omit it.""" + assert "kind" in definition.model_json_schema()["required"] + + def test_create_duplicate_returns_409(client: TestClient) -> None: assert client.post(f"{_BASE}/task-1", json=_body()).status_code == 201 assert client.post(f"{_BASE}/task-1", json=_body()).status_code == 409 @@ -185,7 +214,7 @@ def test_get_returns_the_current_revision(client: TestClient) -> None: client.put(f"{_BASE}/task-1", json=_body(intent="Do something else.")) got = client.get(f"{_BASE}/task-1").json() assert got["revision"] == 2 - assert got["intent"] == "Do something else." + assert got["spec"]["intent"] == "Do something else." # --- Reading and tagging a specific revision --------------------------------- @@ -214,9 +243,9 @@ def test_get_by_digest_returns_the_published_content(client: TestClient) -> None client.put(f"{_BASE}/task-1", json=_body(intent="Newer.")) pinned = client.get(f"{_BASE}/task-1/revisions/{digest}").json() - assert pinned["intent"] == first["intent"] + assert pinned["spec"]["intent"] == first["spec"]["intent"] assert pinned["revision"] == 1 - assert client.get(f"{_BASE}/task-1").json()["intent"] == "Newer." + assert client.get(f"{_BASE}/task-1").json()["spec"]["intent"] == "Newer." def test_get_by_tag_resolves(client: TestClient) -> None: @@ -271,3 +300,24 @@ async def _stale(entity, *, original_name=None): entity_store.update = _stale assert client.put(f"{_BASE}/task-1", json=_body(intent="Newer.")).status_code == 409 + + +def test_list_includes_harbor_tasks(client: TestClient) -> None: + """Both kinds are one record type, so the listing must serialize either.""" + client.post(f"{_BASE}/evaluator-task", json=_body()) + client.post( + f"{_BASE}/harbor-task", + json=TaskInput( + spec=HarborTaskDefinition( + kind="harbor", archive_ref="default/harbor#packages/o-n/abc/dist.tar.gz", archive_digest="a" * 64 + ) + ).model_dump(mode="json"), + ) + + response = client.get(_BASE) + + assert response.status_code == 200 + assert {t["name"]: t["spec"]["kind"] for t in response.json()["data"]} == { + "evaluator-task": "evaluator", + "harbor-task": "harbor", + } diff --git a/plugins/nemo-evaluator/tests/integration/test_agent_evaluate_job.py b/plugins/nemo-evaluator/tests/integration/test_agent_evaluate_job.py index 0bb2892a33..05156539d7 100644 --- a/plugins/nemo-evaluator/tests/integration/test_agent_evaluate_job.py +++ b/plugins/nemo-evaluator/tests/integration/test_agent_evaluate_job.py @@ -37,6 +37,7 @@ import httpx import pytest from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, MetricInline, TaskInput, TaskInputs, @@ -554,9 +555,12 @@ def test_submit_over_taskset_ref_resolves_and_scores(subprocess_platform: str) - client.evaluator.tasks.create( name, task=TaskInput( - intent="Obtain a one-word reply from the model.", - inputs=TaskInputs(instruction="Reply with the single word DONE and nothing else."), - metrics=[MetricRef(f"{WORKSPACE}/{metric_name}")], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Obtain a one-word reply from the model.", + inputs=TaskInputs(instruction="Reply with the single word DONE and nothing else."), + metrics=[MetricRef(f"{WORKSPACE}/{metric_name}")], + ) ), ) taskset_name = _unique("done-suite") diff --git a/plugins/nemo-evaluator/tests/integration/test_docs_manage_tasks_tasksets.py b/plugins/nemo-evaluator/tests/integration/test_docs_manage_tasks_tasksets.py new file mode 100644 index 0000000000..e4b5ca842f --- /dev/null +++ b/plugins/nemo-evaluator/tests/integration/test_docs_manage_tasks_tasksets.py @@ -0,0 +1,205 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""The ``Manage Tasks & Tasksets`` doc walkthrough, executed against a real platform. + +``make docs-check-python-snippets`` type-checks the doc's snippets, which catches a snippet that +names a field that no longer exists — but not one that type-checks and then fails at run time, and +not a documented *output* that no longer matches. Both happened: the task model moved its content +under a discriminated ``spec``, and the revision snippets in this doc kept the old flat shape +through review because nothing executed them. + +So this walks the doc top to bottom, in order, doing what it says and asserting the results it +claims. It deliberately mirrors the doc's own code rather than being written as an idiomatic test — +when it fails, the fix is usually the doc. + +Pure CRUD (no codex/IGW), so it only needs the host subprocess backend. Shares the evaluator-plugin +integration opt-in (``RUN_AGENT_EVAL_INTEGRATION``) and the session-scoped ``subprocess_platform``. +""" + +from __future__ import annotations + +import os +import uuid + +import pytest +from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, + MetadataItem, + MetricRef, + TaskInput, + TaskInputs, + TaskRef, + TasksetInput, + TasksetRef, +) +from nemo_evaluator_sdk import ExactMatchMetric +from nemo_platform import NeMoPlatform + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not os.environ.get("RUN_AGENT_EVAL_INTEGRATION"), + reason="opt-in; set RUN_AGENT_EVAL_INTEGRATION=1 to run (spins real nemo services platforms)", + ), +] + +WORKSPACE = "default" + + +@pytest.fixture +def doc_client(subprocess_platform: str) -> NeMoPlatform: + """The doc's own ``Initialize the SDK`` snippet, with the base URL the fixture provides. + + ``workspace=`` on the constructor is part of what is being checked: every later snippet omits a + per-call workspace and relies on this default. + """ + client = NeMoPlatform(base_url=subprocess_platform, workspace=WORKSPACE, max_retries=2) + client.workspaces.create(name=WORKSPACE, exist_ok=True) + return client + + +def _unique(prefix: str) -> str: + """Names are per-test so a reused platform can't leak state between them.""" + return f"{prefix}-{uuid.uuid4().hex[:8]}" + + +@pytest.mark.timeout(300) +def test_the_manage_tasks_walkthrough(doc_client: NeMoPlatform) -> None: + """``Manage Tasks`` through ``Tag a revision`` — create, read, publish, pin, tag.""" + client = doc_client + tasks = client.evaluator.tasks + task_name = _unique("capital-of-france") + metric_name = _unique("answer-exact-match") + + client.evaluator.metrics.create( + metric_name, + metric=ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}"), + ) + + task = TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the user's geography question with the capital city.", + inputs=TaskInputs(instruction="What is the capital of France?"), + metrics=[MetricRef(f"{WORKSPACE}/{metric_name}")], + ), + metadata=[MetadataItem(key="suite", value="geography")], + ) + + stored = tasks.create(task_name, task=task) + # The doc prints `stored.id, stored.spec.metrics` and states that a stored task holds metric + # *references* only. + assert stored.id + assert [ref.root for ref in stored.spec.metrics] == [f"{WORKSPACE}/{metric_name}"] + + # "Retrieve, list, and delete" — the doc's comment claims `evaluator 1 {'latest': 1}`. + retrieved = tasks.retrieve(task_name) + assert (retrieved.spec.kind, retrieved.revision, retrieved.tags) == ("evaluator", 1, {"latest": 1}) + + page = tasks.list(page=1, page_size=100, sort="-created_at") + assert (task_name, "evaluator") in [(item.name, item.spec.kind) for item in page.data] + + # "Publish a new revision" — the doc's comment claims revision 2. + revised_task = TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the user's geography question with the capital city.", + inputs=TaskInputs(instruction="Name the capital city of France."), + metrics=[MetricRef(f"{WORKSPACE}/{metric_name}")], + ), + metadata=[MetadataItem(key="suite", value="geography")], + ) + updated = tasks.replace(task_name, task=revised_task) + assert updated.revision == 2 + + # The doc's idempotence Note: re-submitting identical content publishes nothing. + assert tasks.replace(task_name, task=revised_task).revision == 2 + + # "Read a specific revision" — a pinned read returns what was published, not what is current. + revisions = tasks.list_revisions(task_name) + digest = next(revision.content_hash for revision in revisions.data if revision.revision == 1) + + original = tasks.retrieve(task_name, revision=digest) + current = tasks.retrieve(task_name) + assert original.revision == 1 and current.revision == 2 + assert original.spec.inputs.instruction != current.spec.inputs.instruction + + # "Tag a revision", including the documented `ValueError` when both selectors are passed. + tasks.tag(task_name, tag="blessed", revision=digest) + blessed = tasks.retrieve(task_name, tag="blessed") + assert blessed.revision == 1 + with pytest.raises(ValueError): + tasks.retrieve(task_name, revision=digest, tag="blessed") + + tasks.delete(task_name) + + +@pytest.mark.timeout(300) +def test_the_manage_tasksets_walkthrough(doc_client: NeMoPlatform) -> None: + """``Manage Tasksets`` and ``Pin the taskset itself`` — membership pinning is the claim.""" + client = doc_client + tasks = client.evaluator.tasks + tasksets = client.evaluator.tasksets + france, japan = _unique("capital-of-france"), _unique("capital-of-japan") + suite = _unique("geography-suite") + + for name, city in ((france, "France"), (japan, "Japan")): + tasks.create( + name, + task=TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the user's geography question with the capital city.", + inputs=TaskInputs(instruction=f"What is the capital of {city}?"), + ) + ), + ) + + taskset = TasksetInput( + description="Geography questions for smoke-testing the agent.", + tasks=[TaskRef(f"{WORKSPACE}/{france}"), TaskRef(f"{WORKSPACE}/{japan}")], + ) + stored = tasksets.create(suite, taskset=taskset) + + # The doc's central claim: a bare member ref is stored resolved to `workspace/name#`. + assert all("#" in ref.root for ref in stored.tasks) + assert {ref.root.split("#")[0] for ref in stored.tasks} == {f"{WORKSPACE}/{france}", f"{WORKSPACE}/{japan}"} + + page = tasksets.list(page=1, page_size=100, sort="name") + assert suite in [item.name for item in page.data] + assert tasksets.retrieve(suite).description == "Geography questions for smoke-testing the agent." + + # The doc's Note: member *order* is not part of a taskset's identity, so reordering the same + # members publishes nothing. + reordered = TasksetInput( + description="Geography questions for smoke-testing the agent.", + tasks=[TaskRef(f"{WORKSPACE}/{japan}"), TaskRef(f"{WORKSPACE}/{france}")], + ) + assert tasksets.replace(suite, taskset=reordered).revision == 1 + + # ...but re-resolving after a member republishes genuinely differs, so it does cut a revision. + tasks.replace( + france, + task=TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the user's geography question with the capital city.", + inputs=TaskInputs(instruction="Name the capital city of France."), + ) + ), + ) + assert tasksets.replace(suite, taskset=taskset).revision == 2 + + # "Pin the taskset itself" — both ref forms are accepted by the field. + current = tasksets.list_revisions(suite).data[0] + assert current.revision == 2 # revisions come back newest-first, as the doc's comment says + assert TasksetRef(f"{WORKSPACE}/{suite}").root + assert TasksetRef(f"{WORKSPACE}/{suite}#{current.content_hash}").root + + # Deleting a taskset does not delete its member tasks. + tasksets.delete(suite) + assert tasks.retrieve(france).name == france + + for name in (france, japan): + tasks.delete(name) diff --git a/plugins/nemo-evaluator/tests/integration/test_task_derived_metrics.py b/plugins/nemo-evaluator/tests/integration/test_task_derived_metrics.py index b377e1efe4..8b9d8106e9 100644 --- a/plugins/nemo-evaluator/tests/integration/test_task_derived_metrics.py +++ b/plugins/nemo-evaluator/tests/integration/test_task_derived_metrics.py @@ -22,7 +22,7 @@ import uuid import pytest -from nemo_evaluator.api.schemas import MetricInline, TaskInput +from nemo_evaluator.api.schemas import EvaluatorTaskDefinition, MetricInline, TaskInput from nemo_evaluator.shared.metric_bundles.bundles import bundle_metric from nemo_evaluator.shared.metric_bundles.cloudpickle import CloudpickleMetricBundlePackager from nemo_evaluator_sdk.metrics.exact_match import ExactMatchMetric @@ -55,7 +55,11 @@ def _inline_metric(marker: str) -> MetricInline: def _task_input(metric: MetricInline) -> TaskInput: - return TaskInput(intent="Answer the question.", inputs={"instruction": "What is 2+2?"}, metrics=[metric]) + return TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", intent="Answer the question.", inputs={"instruction": "What is 2+2?"}, metrics=[metric] + ) + ) @pytest.mark.timeout(300) @@ -70,14 +74,16 @@ def test_inline_task_metric_normalizes_to_derived_metric(subprocess_platform: st try: # The inline metric is offloaded: the stored task holds a single derived reference, not a bundle. created_a = client.evaluator.tasks.create(task_a, task=_task_input(inline), workspace=WORKSPACE) - assert len(created_a.metrics) == 1 - derived_ref = created_a.metrics[0].root + assert isinstance(created_a.spec, EvaluatorTaskDefinition) + assert len(created_a.spec.metrics) == 1 + derived_ref = created_a.spec.metrics[0].root assert derived_ref.startswith(f"{WORKSPACE}/derived.") derived_name = derived_ref.split("/", 1)[1] # A second task with byte-identical inline content dedupes to the same derived metric. created_b = client.evaluator.tasks.create(task_b, task=_task_input(inline), workspace=WORKSPACE) - assert created_b.metrics[0].root == derived_ref + assert isinstance(created_b.spec, EvaluatorTaskDefinition) + assert created_b.spec.metrics[0].root == derived_ref # The derived metric is a real, Files-backed, flagged metric. fetched = client.evaluator.metrics.retrieve(derived_name, workspace=WORKSPACE) diff --git a/plugins/nemo-evaluator/tests/integration/test_task_revisions.py b/plugins/nemo-evaluator/tests/integration/test_task_revisions.py index 4e814251ff..bbe7f34d9c 100644 --- a/plugins/nemo-evaluator/tests/integration/test_task_revisions.py +++ b/plugins/nemo-evaluator/tests/integration/test_task_revisions.py @@ -27,7 +27,12 @@ import uuid import pytest -from nemo_evaluator.api.schemas import TaskInput, TasksetInput +from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, + HarborTaskDefinition, + TaskInput, + TasksetInput, +) from nemo_platform import NeMoPlatform pytestmark = [ @@ -46,7 +51,10 @@ def _unique(prefix: str) -> str: def _task_input(intent: str = "Answer the question.", *, tags: list[str] | None = None) -> TaskInput: - return TaskInput(intent=intent, inputs={"instruction": "What is 2+2?"}, tags=tags or []) + return TaskInput( + spec=EvaluatorTaskDefinition(kind="evaluator", intent=intent, inputs={"instruction": "What is 2+2?"}), + tags=tags or [], + ) def _client(base_url: str) -> NeMoPlatform: @@ -71,9 +79,12 @@ def test_publish_and_read_a_pinned_revision(subprocess_platform: str) -> None: assert replaced.revision == 2 pinned = client.evaluator.tasks.retrieve(name, revision=first_digest, workspace=WORKSPACE) - assert pinned.intent == "First." + assert isinstance(pinned.spec, EvaluatorTaskDefinition) + assert pinned.spec.intent == "First." assert pinned.revision == 1 - assert client.evaluator.tasks.retrieve(name, workspace=WORKSPACE).intent == "Second." + current = client.evaluator.tasks.retrieve(name, workspace=WORKSPACE) + assert isinstance(current.spec, EvaluatorTaskDefinition) + assert current.spec.intent == "Second." finally: client.evaluator.tasks.delete(name, workspace=WORKSPACE) @@ -165,7 +176,9 @@ def test_tagging_an_older_revision_leaves_latest_alone(subprocess_platform: str) assert tagged.tags["blessed"] == 1 assert tagged.tags["latest"] == 2, "latest is machine-managed and must not follow a manual tag" - assert client.evaluator.tasks.retrieve(name, tag="blessed", workspace=WORKSPACE).intent == "First." + blessed = client.evaluator.tasks.retrieve(name, tag="blessed", workspace=WORKSPACE) + assert isinstance(blessed.spec, EvaluatorTaskDefinition) + assert blessed.spec.intent == "First." finally: client.evaluator.tasks.delete(name, workspace=WORKSPACE) @@ -214,10 +227,9 @@ def test_taskset_membership_is_pinned_and_stays_pinned(subprocess_platform: str) client.evaluator.tasks.replace(task_name, task=_task_input("Updated."), workspace=WORKSPACE) assert client.evaluator.tasksets.retrieve(set_name, workspace=WORKSPACE).tasks[0].root == member - assert ( - client.evaluator.tasks.retrieve(task_name, revision=pinned_digest, workspace=WORKSPACE).intent - == "Original." - ) + pinned_task = client.evaluator.tasks.retrieve(task_name, revision=pinned_digest, workspace=WORKSPACE) + assert isinstance(pinned_task.spec, EvaluatorTaskDefinition) + assert pinned_task.spec.intent == "Original." finally: client.evaluator.tasksets.delete(set_name, workspace=WORKSPACE) client.evaluator.tasks.delete(task_name, workspace=WORKSPACE) @@ -247,3 +259,80 @@ def test_republishing_a_taskset_after_a_member_moves_cuts_a_revision(subprocess_ finally: client.evaluator.tasksets.delete(set_name, workspace=WORKSPACE) client.evaluator.tasks.delete(task_name, workspace=WORKSPACE) + + +# --- Harbor-kind tasks -------------------------------------------------------- + + +def _harbor_input(digest: str = "a" * 64, *, config: dict | None = None) -> TaskInput: + return TaskInput( + spec=HarborTaskDefinition( + kind="harbor", + archive_ref="default/harbor-tasks#packages/org-name/abc/dist.tar.gz", + archive_digest=digest, + instruction="Fix the failing test.", + config=config if config is not None else {"verifier": {"type": "pytest"}}, + ) + ) + + +@pytest.mark.timeout(300) +def test_harbor_and_evaluator_tasks_coexist(subprocess_platform: str) -> None: + """Both kinds are one record type, so they list together and a taskset can group them — the + point of managing every evaluation unit in one place.""" + client = _client(subprocess_platform) + harbor_name, evaluator_name = _unique("harbor"), _unique("evaluator") + try: + harbor = client.evaluator.tasks.create(harbor_name, task=_harbor_input(), workspace=WORKSPACE) + evaluator = client.evaluator.tasks.create(evaluator_name, task=_task_input(), workspace=WORKSPACE) + + assert harbor.spec.kind == "harbor" + assert evaluator.spec.kind == "evaluator" + + listed = {t.name: t.spec.kind for t in client.evaluator.tasks.list(workspace=WORKSPACE, page_size=1000).data} + assert listed[harbor_name] == "harbor" + assert listed[evaluator_name] == "evaluator" + finally: + client.evaluator.tasks.delete(harbor_name, workspace=WORKSPACE) + client.evaluator.tasks.delete(evaluator_name, workspace=WORKSPACE) + + +@pytest.mark.timeout(300) +def test_harbor_task_round_trips_through_the_store(subprocess_platform: str) -> None: + """The discriminated union has to survive the entity store's JSON column, which is the one + thing a unit test against an in-memory fake cannot confirm.""" + client = _client(subprocess_platform) + name = _unique("harbor") + try: + client.evaluator.tasks.create(name, task=_harbor_input(), workspace=WORKSPACE) + + fetched = client.evaluator.tasks.retrieve(name, workspace=WORKSPACE) + assert isinstance(fetched.spec, HarborTaskDefinition) + assert fetched.spec.kind == "harbor" + assert fetched.spec.archive_digest == "a" * 64 + assert fetched.spec.config == {"verifier": {"type": "pytest"}} + assert fetched.spec.instruction == "Fix the failing test." + finally: + client.evaluator.tasks.delete(name, workspace=WORKSPACE) + + +@pytest.mark.timeout(300) +def test_harbor_config_changes_do_not_cut_a_revision(subprocess_platform: str) -> None: + """`config` is excluded from the digest because it is a projection of task.toml inside the + archive. Confirmed end-to-end, since the exclusion is applied where the digest is computed.""" + client = _client(subprocess_platform) + name = _unique("harbor") + try: + client.evaluator.tasks.create(name, task=_harbor_input(), workspace=WORKSPACE) + + same = client.evaluator.tasks.replace( + name, task=_harbor_input(config={"verifier": {"type": "pytest"}, "new_field": 1}), workspace=WORKSPACE + ) + assert same.revision == 1, "a config-only change must not publish" + assert isinstance(same.spec, HarborTaskDefinition) + assert same.spec.config["new_field"] == 1 + + moved = client.evaluator.tasks.replace(name, task=_harbor_input(digest="b" * 64), workspace=WORKSPACE) + assert moved.revision == 2, "an archive change must publish" + finally: + client.evaluator.tasks.delete(name, workspace=WORKSPACE) diff --git a/plugins/nemo-evaluator/tests/sdk/test_task_sdk_resources.py b/plugins/nemo-evaluator/tests/sdk/test_task_sdk_resources.py index 6334460ece..e9e80fdc0e 100644 --- a/plugins/nemo-evaluator/tests/sdk/test_task_sdk_resources.py +++ b/plugins/nemo-evaluator/tests/sdk/test_task_sdk_resources.py @@ -10,7 +10,13 @@ from unittest.mock import AsyncMock, MagicMock import pytest -from nemo_evaluator.api.schemas import MetricRef, Revision, Task, TaskInput +from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, + MetricRef, + Revision, + Task, + TaskInput, +) from nemo_evaluator.sdk.task_resources import AsyncEvaluatorTasksResource, EvaluatorTasksResource _BASE = "http://localhost:8080/apis/evaluator/v2/workspaces/default" @@ -19,12 +25,15 @@ def _task_payload(name: str) -> dict[str, Any]: now = datetime.now(timezone.utc) return Task( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the question.", + inputs={"instruction": "What is 2+2?"}, + metrics=[MetricRef("default/stored-metric")], + ), id=f"task-{name}", name=name, workspace="default", - intent="Answer the question.", - inputs={"instruction": "What is 2+2?"}, - metrics=[MetricRef("default/stored-metric")], revision=1, tags={"latest": 1}, created_at=now, @@ -33,7 +42,14 @@ def _task_payload(name: str) -> dict[str, Any]: def _task_input() -> TaskInput: - return TaskInput(intent="Answer.", inputs={"instruction": "x"}, metrics=[MetricRef("default/stored-metric")]) + return TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer.", + inputs={"instruction": "x"}, + metrics=[MetricRef("default/stored-metric")], + ) + ) def _response(payload: Any) -> MagicMock: @@ -63,7 +79,7 @@ def test_sync_create_posts_task_input_to_item_url() -> None: assert isinstance(result, Task) assert result.name == "task-1" assert http_client.post.call_args[0][0] == f"{_BASE}/tasks/task-1" - assert http_client.post.call_args.kwargs["json"]["intent"] == "Answer." + assert http_client.post.call_args.kwargs["json"]["spec"]["intent"] == "Answer." def test_sync_retrieve_targets_item_url_and_parses_dto() -> None: @@ -74,7 +90,8 @@ def test_sync_retrieve_targets_item_url_and_parses_dto() -> None: result = resource.retrieve("task-1") assert isinstance(result, Task) - assert isinstance(result.metrics[0], MetricRef) + assert isinstance(result.spec, EvaluatorTaskDefinition) + assert isinstance(result.spec.metrics[0], MetricRef) assert http_client.get.call_args[0][0] == f"{_BASE}/tasks/task-1" @@ -143,7 +160,7 @@ def test_sync_replace_puts_task_input_to_item_url() -> None: result = resource.replace("task-1", task=_task_input()) assert http_client.put.call_args.args[0] == f"{_BASE}/tasks/task-1" - assert http_client.put.call_args.kwargs["json"]["intent"] == "Answer." + assert http_client.put.call_args.kwargs["json"]["spec"]["intent"] == "Answer." assert isinstance(result, Task) diff --git a/plugins/nemo-evaluator/tests/test_content_hash.py b/plugins/nemo-evaluator/tests/test_content_hash.py index 6dd082ec48..77311b0638 100644 --- a/plugins/nemo-evaluator/tests/test_content_hash.py +++ b/plugins/nemo-evaluator/tests/test_content_hash.py @@ -14,11 +14,19 @@ import hashlib import json import re -from typing import ClassVar - -from nemo_evaluator.api.schemas import MetadataItem, MetricRef, TaskInputs, TaskRef +from typing import Any, ClassVar + +from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, + HarborTaskDefinition, + MetadataItem, + MetricRef, + TaskInputs, + TaskRef, +) from nemo_evaluator.content_hash import DIGEST_PATTERN, canonical_payload, content_hash from nemo_evaluator.entities import TaskEntity, TasksetEntity +from nemo_evaluator.revisions import head_digest from nemo_evaluator_sdk.agent_eval.tasks import SemanticReducer, SemanticView, ViewSignal from nemo_platform_plugin.entities import EntityBase from pydantic import Field @@ -39,22 +47,40 @@ def _task( project: str | None = None, intent: str = "Answer the question.", inputs: TaskInputs | None = None, + reference: dict[str, Any] | None = None, metrics: list[MetricRef] | None = None, views: dict[str, SemanticView] | None = None, metadata: list[MetadataItem] | None = None, ) -> TaskEntity: return TaskEntity( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent=intent, + inputs=inputs if inputs is not None else TaskInputs(instruction="What is 2+2?"), + reference=reference if reference is not None else {}, + metrics=metrics if metrics is not None else [MetricRef("default/stored-metric")], + views=views if views is not None else _DEFAULT_VIEWS, + ), name=name, workspace=workspace, project=project, - intent=intent, - inputs=inputs if inputs is not None else TaskInputs(instruction="What is 2+2?"), - metrics=metrics if metrics is not None else [MetricRef("default/stored-metric")], - views=views if views is not None else _DEFAULT_VIEWS, metadata=metadata if metadata is not None else _DEFAULT_METADATA, ) +def _harbor_task(*, config: dict[str, Any] | None = None, archive_digest: str = "a" * 64) -> TaskEntity: + return TaskEntity( + spec=HarborTaskDefinition( + kind="harbor", + archive_ref="default/harbor#packages/o-n/abc/dist.tar.gz", + archive_digest=archive_digest, + config=config if config is not None else {}, + ), + name="harbor-1", + workspace="default", + ) + + # --- Shape ------------------------------------------------------------------- @@ -147,6 +173,18 @@ def test_metric_ref_order_changes_digest() -> None: assert content_hash(a) != content_hash(b) +def test_grader_only_reference_changes_digest() -> None: + """``reference`` decides what a metric grades *against*, so it is task content. + + Two revisions that score the same output differently must not share a digest — otherwise + publish-time dedup would collapse them and a pin would no longer fix the grading. This is the + general rule for the digest: it covers anything affecting a task's execution output or the + mechanism used to grade it. + """ + assert content_hash(_task(reference={"expected": "Paris"})) != content_hash(_task()) + assert content_hash(_task(reference={"expected": "Paris"})) != content_hash(_task(reference={"expected": "Lyon"})) + + def test_nested_view_change_changes_digest() -> None: """Nested sub-models participate; a change buried in a view must not be invisible.""" changed = _task( @@ -192,6 +230,35 @@ def test_int_and_float_render_distinctly() -> None: ) +# --- Harbor: the one deliberate exclusion ------------------------------------ + + +def test_harbor_config_does_not_change_digest() -> None: + """``config`` is a *projection* of ``task.toml``, never an execution input. + + Harbor reads the real ``task.toml`` out of the materialized archive at run time, so this copy + affects neither execution nor grading. Hashing it would buy no coverage and would make revision + history sensitive to Harbor's serialization — a release that reordered keys or emitted a new + defaulted field would cut a revision for byte-identical files. + + Exercised through ``head_digest`` rather than ``content_hash``: the exclusion lives in + ``REVISION_POINTER_EXCLUDE``, not in the hashing primitive. + """ + plain = _harbor_task() + configured = _harbor_task(config={"verifier": {"type": "pytest"}, "agent": {"timeout": 600}}) + assert head_digest(plain) == head_digest(configured) + + +def test_harbor_archive_digest_changes_digest() -> None: + """The invariant that makes excluding ``config`` safe. + + ``archive_digest`` is authoritative over every file in the task directory, ``task.toml`` + included — so a config change that genuinely alters execution or grading moves *this* field and + is covered. If this ever stopped holding, excluding ``config`` would become a real gap. + """ + assert head_digest(_harbor_task()) != head_digest(_harbor_task(archive_digest="b" * 64)) + + # --- Tasksets ---------------------------------------------------------------- diff --git a/plugins/nemo-evaluator/tests/test_revision_entity.py b/plugins/nemo-evaluator/tests/test_revision_entity.py index ca1dffb4bf..ed21bb7db5 100644 --- a/plugins/nemo-evaluator/tests/test_revision_entity.py +++ b/plugins/nemo-evaluator/tests/test_revision_entity.py @@ -14,11 +14,11 @@ import re import pytest -from nemo_evaluator.api.schemas import MetadataItem, MetricRef, TaskInputs, TaskRef +from nemo_evaluator.api.schemas import EvaluatorTaskDefinition, MetadataItem, MetricRef, TaskInputs, TaskRef from nemo_evaluator.content_hash import content_hash from nemo_evaluator.entities import ( - REVISION_POINTER_FIELDS, - REVISION_SELF_FIELDS, + REVISION_POINTER_EXCLUDE, + REVISION_SELF_EXCLUDE, TaskEntity, TaskRevisionEntity, TasksetEntity, @@ -41,11 +41,9 @@ def _task_head(*, intent: str = _INTENT, latest_revision: int = 0, tags: dict[str, int] | None = None) -> TaskEntity: return TaskEntity( + spec=EvaluatorTaskDefinition(kind="evaluator", intent=intent, inputs=_INPUTS, metrics=_METRICS), name="task-1", workspace="default", - intent=intent, - inputs=_INPUTS, - metrics=_METRICS, metadata=_ANNOTATIONS, latest_revision=latest_revision, tags=tags or {}, @@ -54,13 +52,11 @@ def _task_head(*, intent: str = _INTENT, latest_revision: int = 0, tags: dict[st def _task_revision(*, intent: str = _INTENT, revision: int = 1, digest: str = _DIGEST) -> TaskRevisionEntity: return TaskRevisionEntity( + spec=EvaluatorTaskDefinition(kind="evaluator", intent=intent, inputs=_INPUTS, metrics=_METRICS), name=f"rev.{revision}", workspace="default", content_hash=digest, revision=revision, - intent=intent, - inputs=_INPUTS, - metrics=_METRICS, metadata=_ANNOTATIONS, ) @@ -89,43 +85,43 @@ def _taskset_revision(*, members: list[TaskRef] | None = None) -> TasksetRevisio def test_task_head_and_revision_digests_agree() -> None: - assert content_hash(_task_head(), exclude=REVISION_POINTER_FIELDS) == content_hash( - _task_revision(), exclude=REVISION_SELF_FIELDS + assert content_hash(_task_head(), exclude=REVISION_POINTER_EXCLUDE) == content_hash( + _task_revision(), exclude=REVISION_SELF_EXCLUDE ) def test_taskset_head_and_revision_digests_agree() -> None: - assert content_hash(_taskset_head(), exclude=REVISION_POINTER_FIELDS) == content_hash( - _taskset_revision(), exclude=REVISION_SELF_FIELDS + assert content_hash(_taskset_head(), exclude=REVISION_POINTER_EXCLUDE) == content_hash( + _taskset_revision(), exclude=REVISION_SELF_EXCLUDE ) def test_moving_a_tag_does_not_change_the_head_digest() -> None: """Tags are pointers, not content. If they were digested, every retag would fork history.""" tagged = _task_head(latest_revision=7, tags={"latest": 7, "candidate": 3}) - assert content_hash(_task_head(), exclude=REVISION_POINTER_FIELDS) == content_hash( - tagged, exclude=REVISION_POINTER_FIELDS + assert content_hash(_task_head(), exclude=REVISION_POINTER_EXCLUDE) == content_hash( + tagged, exclude=REVISION_POINTER_EXCLUDE ) def test_ordinal_does_not_change_the_revision_digest() -> None: """Two revisions of identical content digest identically regardless of when they were cut.""" - assert content_hash(_task_revision(revision=1), exclude=REVISION_SELF_FIELDS) == content_hash( - _task_revision(revision=9), exclude=REVISION_SELF_FIELDS + assert content_hash(_task_revision(revision=1), exclude=REVISION_SELF_EXCLUDE) == content_hash( + _task_revision(revision=9), exclude=REVISION_SELF_EXCLUDE ) def test_content_change_changes_the_revision_digest() -> None: - assert content_hash(_task_revision(), exclude=REVISION_SELF_FIELDS) != content_hash( - _task_revision(intent="Do something else."), exclude=REVISION_SELF_FIELDS + assert content_hash(_task_revision(), exclude=REVISION_SELF_EXCLUDE) != content_hash( + _task_revision(intent="Do something else."), exclude=REVISION_SELF_EXCLUDE ) def test_membership_change_changes_the_taskset_revision_digest() -> None: """A published dataset's identity is its membership — including which revision of each member.""" repinned = _taskset_revision(members=[TaskRef(f"default/task-a#{_OTHER_DIGEST}")]) - assert content_hash(_taskset_revision(), exclude=REVISION_SELF_FIELDS) != content_hash( - repinned, exclude=REVISION_SELF_FIELDS + assert content_hash(_taskset_revision(), exclude=REVISION_SELF_EXCLUDE) != content_hash( + repinned, exclude=REVISION_SELF_EXCLUDE ) diff --git a/plugins/nemo-evaluator/tests/test_revisions.py b/plugins/nemo-evaluator/tests/test_revisions.py index b263389dfe..42d9dcdc7b 100644 --- a/plugins/nemo-evaluator/tests/test_revisions.py +++ b/plugins/nemo-evaluator/tests/test_revisions.py @@ -16,7 +16,7 @@ from typing import TypeVar import pytest -from nemo_evaluator.api.schemas import LATEST_TAG, MetricRef, TaskInputs, TaskRef +from nemo_evaluator.api.schemas import LATEST_TAG, EvaluatorTaskDefinition, MetricRef, TaskInputs, TaskRef from nemo_evaluator.entities import TaskEntity, TaskRevisionEntity from nemo_evaluator.revisions import ( RevisionConflictError, @@ -202,11 +202,14 @@ def concurrent_head_write(self, head: EntityBase, *, tags: dict[str, int]) -> No def _head(store: FakeStore, *, intent: str = "Answer the question.") -> TaskEntity: head = TaskEntity( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent=intent, + inputs=TaskInputs(instruction="What is 2+2?"), + metrics=[MetricRef("default/stored-metric")], + ), name="task-1", workspace="default", - intent=intent, - inputs=TaskInputs(instruction="What is 2+2?"), - metrics=[MetricRef("default/stored-metric")], ) head._id = "head-1" head._db_version = 0 @@ -223,11 +226,14 @@ def _head(store: FakeStore, *, intent: str = "Answer the question.") -> TaskEnti def _head_named(store: FakeStore, name: str) -> TaskEntity: """A second record with content identical to :func:`_head`'s — same digest, different parent.""" head = TaskEntity( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the question.", + inputs=TaskInputs(instruction="What is 2+2?"), + metrics=[MetricRef("default/stored-metric")], + ), name=name, workspace="default", - intent="Answer the question.", - inputs=TaskInputs(instruction="What is 2+2?"), - metrics=[MetricRef("default/stored-metric")], ) head._id = f"head-{name}" head._db_version = 0 @@ -307,7 +313,7 @@ async def test_changed_content_allocates_the_next_ordinal() -> None: store = FakeStore() head = _head(store) first, _ = await _publish(store, head) - head.intent = "Do something else." + head.spec.intent = "Do something else." second, created = await _publish(store, head) assert created assert second.revision == 2 @@ -325,7 +331,7 @@ async def test_contended_ordinal_is_retried() -> None: store = FakeStore() head = _head(store) await _publish(store, head) - head.intent = "Changed." + head.spec.intent = "Changed." store.contend_ordinals = {2} revision, created = await _publish(store, head) assert created @@ -346,7 +352,7 @@ async def test_identical_contended_publish_adopts_the_winners_revision() -> None head = _head(store) await _publish(store, head) - head.intent = "Changed." + head.spec.intent = "Changed." store.contend_identically = {2} revision, created = await _publish(store, head) @@ -372,7 +378,7 @@ async def test_contended_publish_of_different_content_still_allocates_a_new_ordi head = _head(store) await _publish(store, head) - head.intent = "Changed." + head.spec.intent = "Changed." store.contend_ordinals = {2} # winner publishes *different* content revision, created = await _publish(store, head) @@ -400,7 +406,7 @@ async def test_publishing_recovers_when_a_revision_exists_but_the_head_never_adv assert isinstance(stored, TaskEntity) stored.latest_revision, stored.tags = 0, {} - head.intent = "Changed." + head.spec.intent = "Changed." revision, created = await _publish(store, head) assert created @@ -422,7 +428,7 @@ async def test_losing_the_head_race_does_not_leave_the_head_on_an_older_revision a = await store.get(TaskEntity, name="task-1", workspace="default") b = await store.get(TaskEntity, name="task-1", workspace="default") - a.intent, b.intent = "A's content.", "B's content." + a.spec.intent, b.spec.intent = "A's content.", "B's content." async def b_publishes() -> None: await publish_revision(store, store, b, TaskRevisionEntity) @@ -435,13 +441,13 @@ async def b_publishes() -> None: latest = await get_revision(store, TaskRevisionEntity, stored, LATEST_TAG) assert stored.tags[LATEST_TAG] == 3 - assert stored.intent == latest.intent == "B's content." + assert stored.spec.intent == latest.spec.intent == "B's content." assert stored.latest_revision == latest.revision, "the reported revision must describe the content served" # A's publish is not lost — it is a real revision, still resolvable by digest. assert a_revision.revision == 2 pinned = await get_revision(store, TaskRevisionEntity, stored, a_revision.content_hash) - assert pinned.intent == "A's content." + assert pinned.spec.intent == "A's content." @pytest.mark.asyncio @@ -464,7 +470,7 @@ async def test_latest_revision_never_rewinds() -> None: store = FakeStore() head = _head(store) first, _ = await _publish(store, head) - head.intent = "Changed." + head.spec.intent = "Changed." await _publish(store, head) head = await store.get(TaskEntity, "task-1", workspace="default") @@ -534,7 +540,7 @@ async def test_user_tags_may_be_moved_backwards() -> None: store = FakeStore() head = _head(store) older, _ = await _publish(store, head, tags={"blessed"}) - head.intent = "Newer content." + head.spec.intent = "Newer content." await _publish(store, head) head = await store.get(TaskEntity, "task-1", workspace="default") @@ -556,10 +562,10 @@ async def test_reverting_to_earlier_content_publishes_a_new_revision() -> None: store = FakeStore() head = _head(store) await _publish(store, head) # rev.1: "Answer the question." - head.intent = "Changed." + head.spec.intent = "Changed." await _publish(store, head) # rev.2 - head.intent = "Answer the question." # back to rev.1's content + head.spec.intent = "Answer the question." # back to rev.1's content revision, created = await _publish(store, head) assert created, "a revert is a publish, not a no-op" @@ -567,7 +573,7 @@ async def test_reverting_to_earlier_content_publishes_a_new_revision() -> None: assert head.tags[LATEST_TAG] == 3 latest = await get_revision(store, TaskRevisionEntity, head, LATEST_TAG) - assert latest.intent == head.intent, "the head and #latest must describe the same content" + assert latest.spec.intent == head.spec.intent, "the head and #latest must describe the same content" @pytest.mark.asyncio @@ -578,9 +584,9 @@ async def test_a_digest_shared_by_two_revisions_resolves_to_the_newer_one() -> N store = FakeStore() head = _head(store) first, _ = await _publish(store, head) # rev.1 - head.intent = "Changed." + head.spec.intent = "Changed." await _publish(store, head) # rev.2 - head.intent = "Answer the question." + head.spec.intent = "Answer the question." third, _ = await _publish(store, head) # rev.3, same digest as rev.1 assert third.content_hash == first.content_hash @@ -610,7 +616,7 @@ async def test_resolves_latest_by_default() -> None: store = FakeStore() head = _head(store) await _publish(store, head) - head.intent = "Changed." + head.spec.intent = "Changed." second, _ = await _publish(store, head) assert (await get_revision(store, TaskRevisionEntity, head)).content_hash == second.content_hash @@ -620,7 +626,7 @@ async def test_resolves_a_digest_to_its_revision() -> None: store = FakeStore() head = _head(store) first, _ = await _publish(store, head) - head.intent = "Changed." + head.spec.intent = "Changed." await _publish(store, head) resolved = await get_revision(store, TaskRevisionEntity, head, first.content_hash) assert resolved.revision == 1 @@ -702,7 +708,7 @@ async def test_reading_a_revision_whose_content_was_tampered_with_is_refused() - stored = store.records[store._key(TaskRevisionEntity, revision_name(1), "default", head.id)] assert isinstance(stored, TaskRevisionEntity) - stored.intent = "Tampered with after publication." + stored.spec.intent = "Tampered with after publication." with pytest.raises(RevisionContentMismatchError, match="does not match its recorded digest"): await get_revision(store, TaskRevisionEntity, head, LATEST_TAG) @@ -718,7 +724,7 @@ async def test_a_digest_pinned_read_is_verified_too() -> None: stored = store.records[store._key(TaskRevisionEntity, revision_name(1), "default", head.id)] assert isinstance(stored, TaskRevisionEntity) - stored.intent = "Tampered with after publication." + stored.spec.intent = "Tampered with after publication." with pytest.raises(RevisionContentMismatchError): await get_revision(store, TaskRevisionEntity, head, revision.content_hash) diff --git a/plugins/nemo-evaluator/tests/test_skill_examples.py b/plugins/nemo-evaluator/tests/test_skill_examples.py index f59e67bc1b..49f48cba94 100644 --- a/plugins/nemo-evaluator/tests/test_skill_examples.py +++ b/plugins/nemo-evaluator/tests/test_skill_examples.py @@ -351,8 +351,8 @@ def test_skill_routes_dataset_examples_to_references() -> None: assert "references/execution.md#getting-job-results" in skill assert "references/resources.md#store-a-metric-task-and-taskset" in skill assert "references/resources.md#query-persisted-results" in skill - assert "result = Evaluator().run_sync(" not in skill - assert "job = client.evaluator.submit(" not in skill + assert "result = Evaluator().run_dataset_sync(" not in skill + assert "job = client.evaluator.evaluate_dataset(" not in skill def test_skill_links_to_evaluation_shape_guidance() -> None: @@ -394,7 +394,7 @@ def test_execution_pairs_python_examples_with_cli_when_supported() -> None: assert cli_blocks >= python_blocks submit_block = reference.split("**Platform Python SDK**", 1)[1].split("```python", 1)[1].split("```", 1)[0] - assert "metric=ExactMatchMetric(" in submit_block + assert "metrics=[ExactMatchMetric(" in submit_block assert "dataset=[" in submit_block @@ -445,13 +445,20 @@ def test_multiple_metric_platform_submission_uses_cli() -> None: assert "nemo evaluator evaluate submit --spec-file multi-metric.json" in section -def test_resources_show_inline_task_before_held_out_reference_guidance() -> None: +def test_resources_show_a_stored_task_carrying_held_out_reference() -> None: + """Held-out ground truth belongs on a *stored* task, so it survives taskset expansion. + + The skill used to steer users to an inline ``AgentEvalTaskInput`` because the stored spec had no + ``reference`` field. It has one now, and routing them back to inline would cost them tasksets + and revision pinning for no reason. + """ reference = (_repo_root() / "skills/nemo-evaluator-plugin/references/resources.md").read_text(encoding="utf-8") - example_position = reference.index("inline_task = AgentEvalTaskInput(") + example_position = reference.index('"capital-france-graded"') guidance_position = reference.index("Stored tasks keep metric references.") assert example_position < guidance_position assert 'reference={"expected": "Paris"}' in reference + assert "EvaluatorTaskDefinition(" in reference def test_agent_evaluation_shows_how_to_retrieve_stored_trials() -> None: @@ -492,9 +499,9 @@ def test_authored_skill_guidance_uses_submit_for_plugin_jobs() -> None: assert "is being retired" in normalized_skill assert "`nemo_evaluator_sdk.Evaluator`" in normalized_skill - assert "Evaluator().run_sync(" in guidance + assert "Evaluator().run_dataset_sync(" in guidance assert "AgentEvaluator().run(" in guidance - assert "client.evaluator.submit(" in guidance + assert "client.evaluator.evaluate_dataset(" in guidance assert "nemo evaluator evaluate submit" in guidance assert "nemo evaluator agent-evaluate submit" in guidance diff --git a/plugins/nemo-evaluator/tests/test_subentity_refs.py b/plugins/nemo-evaluator/tests/test_subentity_refs.py index 5beab88b1d..b4bc02fd05 100644 --- a/plugins/nemo-evaluator/tests/test_subentity_refs.py +++ b/plugins/nemo-evaluator/tests/test_subentity_refs.py @@ -5,21 +5,23 @@ A revision is addressed with the platform's standard ``#`` fragment — the same convention filesets use for a contained file (``workspace/fileset#path``). These tests pin two things: that an absent -fragment means ``latest`` rather than "unpinned", and that existing fragment-unaware callers keep -working against a pinned ref (``parse_entity_ref`` strips it). +fragment means ``latest`` rather than "unpinned", and that a fragment-unaware caller reading a +pinned ref still lands on the right task rather than on one literally named ``task-a#``. """ from __future__ import annotations +import re + import pytest from nemo_evaluator.api.schemas import ( LATEST_TAG, MetricRef, TaskRef, TasksetRef, - parse_entity_ref, parse_subentity_ref, ) +from nemo_platform_plugin.refs import ENTITY_REF_PATTERN, parse_entity_ref from pydantic import ValidationError _DIGEST = "a" * 64 @@ -60,20 +62,37 @@ def test_fragment_is_returned_verbatim() -> None: assert fragment == _DIGEST -# --- Backward compatibility -------------------------------------------------- +# --- Composition with the platform's entity parser --------------------------- -def test_parse_entity_ref_strips_the_fragment() -> None: - """Fragment-unaware callers (metric resolution, taskset member existence checks) keep working - against a pinned ref instead of trying to look up a task literally named 'task-a#'.""" - assert parse_entity_ref(f"other/task-a#{_DIGEST}", "default") == ("other", "task-a") - assert parse_entity_ref("task-a#latest", "default") == ("default", "task-a") +def test_dropping_the_fragment_recovers_the_plain_entity_ref() -> None: + """Fragment-unaware callers (taskset member existence checks) read a pinned ref by discarding the + third element, rather than through a second parser that strips ``#`` itself. Keeping one parser + is what stops evaluator refs and platform refs drifting on what a ``workspace/name`` is.""" + assert parse_subentity_ref(f"other/task-a#{_DIGEST}", "default")[:2] == ("other", "task-a") + assert parse_subentity_ref("task-a#latest", "default")[:2] == ("default", "task-a") def test_pinned_and_bare_refs_resolve_to_the_same_task() -> None: """The property taskset duplicate-detection relies on: two refs differing only by fragment are the same member, and must not both be admitted.""" - assert parse_entity_ref(f"task-a#{_DIGEST}", "default") == parse_entity_ref("task-a", "default") + assert parse_subentity_ref(f"task-a#{_DIGEST}", "default")[:2] == parse_subentity_ref("task-a", "default")[:2] + + +def test_the_base_split_is_the_platform_parser() -> None: + """Not an implementation detail worth pinning for its own sake — it is the guarantee that a + reference means the same thing to the evaluator as it does to every other plugin.""" + parsed = parse_entity_ref("other/task-a", "default") + assert parse_subentity_ref("other/task-a", "default")[:2] == (parsed.workspace, parsed.name) + + +def test_subentity_pattern_is_the_entity_pattern_plus_a_fragment() -> None: + """The evaluator's ref shape is derived from the platform constant, so widening what counts as a + ``workspace/name`` widens both at once instead of leaving one behind.""" + assert TaskRef.model_fields["root"].metadata # the pattern is declared on the field + for bare in ("task-a", "other/task-a"): + assert re.fullmatch(ENTITY_REF_PATTERN, bare) + assert TaskRef(bare).root == bare # --- Field validation -------------------------------------------------------- diff --git a/plugins/nemo-evaluator/tests/test_task_entity.py b/plugins/nemo-evaluator/tests/test_task_entity.py index 86aa76005e..0a88acf5ca 100644 --- a/plugins/nemo-evaluator/tests/test_task_entity.py +++ b/plugins/nemo-evaluator/tests/test_task_entity.py @@ -13,7 +13,7 @@ import json -from nemo_evaluator.api.schemas import MetricRef +from nemo_evaluator.api.schemas import EvaluatorTaskDefinition, MetricRef from nemo_evaluator.entities import TaskEntity from nemo_evaluator_sdk.agent_eval.tasks import SemanticReducer, SemanticView, ViewSignal @@ -22,16 +22,19 @@ def _entity() -> TaskEntity: return TaskEntity( name="task-1", workspace="default", - intent="Answer the question.", - inputs={"instruction": "What is 2+2?"}, - # A persisted task holds metric references only — a workspace-qualified ref and a bare name. - metrics=[MetricRef("default/stored-metric"), MetricRef("derived.abc123")], - views={ - "correctness": SemanticView( - reducer=SemanticReducer.SINGLE, - signals=[ViewSignal(metric="exact-match", output="score")], - ) - }, + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Answer the question.", + inputs={"instruction": "What is 2+2?"}, + # A persisted task holds metric references only — a workspace-qualified ref and a bare name. + metrics=[MetricRef("default/stored-metric"), MetricRef("derived.abc123")], + views={ + "correctness": SemanticView( + reducer=SemanticReducer.SINGLE, + signals=[ViewSignal(metric="exact-match", output="score")], + ) + }, + ), metadata=[{"key": "suite", "value": "smoke"}], ) @@ -47,16 +50,16 @@ def test_roundtrip_preserves_task_fields() -> None: restored = _roundtrip(entity) - assert restored.intent == "Answer the question." - assert restored.inputs.instruction == "What is 2+2?" + assert restored.spec.intent == "Answer the question." + assert restored.spec.inputs.instruction == "What is 2+2?" assert [(m.key, m.value) for m in restored.metadata] == [("suite", "smoke")] # Metric refs survive as RootModel strings. - assert isinstance(restored.metrics[0], MetricRef) - assert restored.metrics[0].root == "default/stored-metric" - assert isinstance(restored.metrics[1], MetricRef) - assert restored.metrics[1].root == "derived.abc123" + assert isinstance(restored.spec.metrics[0], MetricRef) + assert restored.spec.metrics[0].root == "default/stored-metric" + assert isinstance(restored.spec.metrics[1], MetricRef) + assert restored.spec.metrics[1].root == "derived.abc123" # Nested SemanticView survives the JSON column. - assert restored.views == entity.views + assert restored.spec.views == entity.spec.views def test_entity_type_is_task() -> None: diff --git a/plugins/nemo-evaluator/tests/test_task_refs.py b/plugins/nemo-evaluator/tests/test_task_refs.py index 1b794c9749..27fb0a33e2 100644 --- a/plugins/nemo-evaluator/tests/test_task_refs.py +++ b/plugins/nemo-evaluator/tests/test_task_refs.py @@ -9,6 +9,8 @@ import pytest from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, + HarborTaskDefinition, MetadataItem, MetricRef, TaskInputs, @@ -19,7 +21,11 @@ from nemo_evaluator.entities import TaskEntity, TaskRevisionEntity, TasksetEntity, TasksetRevisionEntity from nemo_evaluator.jobs.agent_spec import AgentEvalTaskInput from nemo_evaluator.revisions import apply_tag, get_revision, head_digest, is_digest, publish_revision -from nemo_evaluator.task_refs import resolve_agent_eval_tasks, resolve_taskset_ref +from nemo_evaluator.task_refs import ( + UnsupportedTaskKindError, + resolve_agent_eval_tasks, + resolve_taskset_ref, +) from nemo_platform_plugin.entities import EntityBase from nemo_platform_plugin.entity_client import NemoEntityNotFoundError from pydantic import ValidationError @@ -29,11 +35,14 @@ def _task(name: str, *, workspace: str = "default", metric: str = "default/m") -> TaskEntity: return TaskEntity( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent=f"Do {name}.", + inputs=TaskInputs(instruction=f"instruction for {name}"), + metrics=[MetricRef(metric)], + ), name=name, workspace=workspace, - intent=f"Do {name}.", - inputs=TaskInputs(instruction=f"instruction for {name}"), - metrics=[MetricRef(metric)], metadata=[MetadataItem(key="suite", value="geo")], ) @@ -117,10 +126,46 @@ async def test_resolves_taskset_members_to_inline_task_inputs(entity_store) -> N assert tasks[0].metrics == [MetricRef("default/m")] assert tasks[0].intent == "Do capital-of-france." assert tasks[0].inputs.instruction == "instruction for capital-of-france" - # A stored task carries no grader-only reference. + # A task stored without ground truth expands to an empty reference, not a missing one. assert tasks[0].reference == {} +async def test_grader_only_reference_survives_taskset_expansion(entity_store) -> None: + """Held-out ground truth must not be the privilege of inline submissions. + + Expansion projects a stored task onto the inline DTO field by field, so a field added to the + stored spec and forgotten here silently becomes empty at run time — the agent is then graded + against nothing, and the run still reports a score. That is the failure this guards. + """ + task = _task("fix-bug") + task.spec.reference = {"expected": "Paris", "held_out_tests": ["test_capital.py"]} + client = await _store(entity_store, task, _taskset("geo", ["default/fix-bug"])) + + tasks = await resolve_taskset_ref(TasksetRef("default/geo"), workspace="default", entity_client=client) + + assert tasks[0].reference == {"expected": "Paris", "held_out_tests": ["test_capital.py"]} + + +async def test_expansion_returns_the_pinned_reference_not_the_current_one(entity_store) -> None: + """``reference`` is digest-covered, so republishing it cuts a revision the old pin excludes. + + A pin that honoured new ground truth would silently re-grade a "reproducible" dataset. + """ + task = _task("fix-bug") + task.spec.reference = {"expected": "Paris"} + client = await _store(entity_store, task) + pinned_digest = head_digest(task) + await _create_published(client, _taskset("geo", [f"default/fix-bug#{pinned_digest}"])) + + task.spec.reference = {"expected": "Lyon"} + await client.update(task) + await publish_revision(client, client, task, TaskRevisionEntity) + + tasks = await resolve_taskset_ref(TasksetRef("default/geo"), workspace="default", entity_client=client) + + assert tasks[0].reference == {"expected": "Paris"}, "expansion must return the pinned ground truth" + + async def test_bare_member_ref_resolves_against_taskset_workspace(entity_store) -> None: client = await _store(entity_store, _task("t1", workspace="team"), _taskset("ts", ["t1"], workspace="team")) @@ -191,7 +236,7 @@ async def test_expansion_uses_the_pinned_revision_not_current_content(entity_sto await _create_published(client, _taskset("geo", [f"default/capital-of-france#{pinned_digest}"])) # The member publishes newer content after the taskset was pinned. - task.intent = "Something else entirely." + task.spec.intent = "Something else entirely." await client.update(task) await publish_revision(client, client, task, TaskRevisionEntity) @@ -216,7 +261,7 @@ async def test_a_tag_pinned_member_stores_the_tagged_revision_not_the_head(entit # ``apply_tag`` hands back — tagging bumps the record's version, so the original object is stale. stored_task = await client.get(TaskEntity, name="capital-of-france", workspace="default") tagged = await apply_tag(client, client, TaskRevisionEntity, stored_task, "blessed", "latest") - tagged.intent = "Something else entirely." + tagged.spec.intent = "Something else entirely." await client.update(tagged) await publish_revision(client, client, tagged, TaskRevisionEntity) @@ -322,3 +367,21 @@ async def test_expansion_fails_loudly_when_a_pin_no_longer_resolves(entity_store with pytest.raises(ValueError, match="no longer resolves"): await resolve_taskset_ref(TasksetRef("default/geo"), workspace="default", entity_client=client) + + +async def test_expansion_rejects_a_task_whose_runner_the_target_cannot_run(entity_store) -> None: + """A Harbor task's content is a directory of files, not fields. Projecting it onto an inline + agent-eval task would silently produce a task with no intent and no metrics — an evaluation that + runs and scores nothing. Refused instead, before the run starts.""" + harbor_task = TaskEntity( + name="fix-test", + workspace="default", + spec=HarborTaskDefinition( + kind="harbor", archive_ref="default/harbor#packages/o-n/abc/dist.tar.gz", archive_digest="a" * 64 + ), + ) + client = await _store(entity_store, harbor_task) + await _create_published(client, _taskset("mixed", [f"default/fix-test#{head_digest(harbor_task)}"])) + + with pytest.raises(UnsupportedTaskKindError, match="harbor"): + await resolve_taskset_ref(TasksetRef("default/mixed"), workspace="default", entity_client=client) diff --git a/skills/nemo-evaluator-plugin/SKILL.md b/skills/nemo-evaluator-plugin/SKILL.md index 248abf92d0..3f5624a360 100644 --- a/skills/nemo-evaluator-plugin/SKILL.md +++ b/skills/nemo-evaluator-plugin/SKILL.md @@ -46,14 +46,16 @@ metric for a rubric, RAG workflow, or tool-calling evaluation. | Need | Interface | | --- | --- | | Fast metric iteration without NeMo Platform | `nemo_evaluator_sdk.Evaluator` | -| Dataset-driven platform job | `client.evaluator.submit(...)` or `nemo evaluator evaluate submit` | +| Dataset-driven platform job | `client.evaluator.evaluate_dataset(...)` or `nemo evaluator evaluate submit` | | Multiple inline/stored metric refs in one job | `nemo evaluator evaluate submit` with an `EvaluateInputSpec` | | Task-driven platform job | `nemo evaluator agent-evaluate submit` | | Reusable platform definitions and result indexes | `client.evaluator.metrics`, `.tasks`, `.tasksets`, `.eval_results`, `.agent_eval_results` | -Default to `submit` for every plugin evaluation. The plugin's local execution -path — `client.evaluator.run()` and the `nemo evaluator ... run` CLI verb — is -being retired, so do not build on it even though `--help` still lists it. For +Default to `evaluate_dataset` for dataset-driven plugin evaluations, and to +`nemo evaluator agent-evaluate submit` for task-driven ones. The plugin's local +execution path is being retired: `client.evaluator.run()` has been removed, and +the `nemo evaluator ... run` CLI verb should not be built on even though +`--help` still lists it. For fast metric iteration without the platform, use the standalone `nemo_evaluator_sdk.Evaluator` instead. @@ -65,7 +67,7 @@ result queries. ## Limitations - `api_key_secret` is an environment-variable name standalone but a NeMo - Platform secret name on `submit`. See [API Auth](references/api-auth.md). + Platform secret name on `evaluate_dataset`. See [API Auth](references/api-auth.md). - HTTP 409 from a submission often means a referenced platform secret is missing, not a duplicate job. Read the response body. - `intent` is grader metadata and is never shown to the agent; only `inputs` diff --git a/skills/nemo-evaluator-plugin/assets/examples/plugin_sdk_examples.py b/skills/nemo-evaluator-plugin/assets/examples/plugin_sdk_examples.py index b5e127d60d..8482b2a597 100644 --- a/skills/nemo-evaluator-plugin/assets/examples/plugin_sdk_examples.py +++ b/skills/nemo-evaluator-plugin/assets/examples/plugin_sdk_examples.py @@ -28,11 +28,13 @@ def evaluate_standalone() -> Any: """Evaluate one deterministic metric in process.""" from nemo_evaluator_sdk import Evaluator, ExactMatchMetric - return Evaluator().run_sync( - metrics=ExactMatchMetric( - reference="{{item.expected}}", - candidate="{{item.output}}", - ), + return Evaluator().run_dataset_sync( + metrics=[ + ExactMatchMetric( + reference="{{item.expected}}", + candidate="{{item.output}}", + ) + ], dataset=[ {"expected": "Paris", "output": "Paris"}, {"expected": "Paris", "output": "London"}, @@ -44,11 +46,13 @@ def submit_and_collect(client: Any, output_dir: Path) -> tuple[Any, Path]: """Submit one metric, wait for completion, and retrieve its artifacts.""" from nemo_evaluator_sdk import ExactMatchMetric - job = client.evaluator.submit( - metric=ExactMatchMetric( - reference="{{item.expected}}", - candidate="{{item.output}}", - ), + job = client.evaluator.evaluate_dataset( + metrics=[ + ExactMatchMetric( + reference="{{item.expected}}", + candidate="{{item.output}}", + ) + ], dataset=[{"expected": "Paris", "output": "Paris"}], ) job.wait_until_done() @@ -58,6 +62,7 @@ def submit_and_collect(client: Any, output_dir: Path) -> tuple[Any, Path]: def store_resources(client: Any) -> None: """Store one metric, task, and taskset.""" from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, MetricRef, TaskInput, TaskInputs, @@ -72,14 +77,17 @@ def store_resources(client: Any) -> None: client.evaluator.tasks.create( "capital-france", task=TaskInput( - intent="Name the capital of France.", - inputs=TaskInputs(instruction="What is the capital of France?"), - metrics=[MetricRef("default/answer-exact")], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Name the capital of France.", + inputs=TaskInputs(instruction="What is the capital of France?"), + metrics=[MetricRef("answer-exact")], + ), ), ) client.evaluator.tasksets.create( "geography", - taskset=TasksetInput(tasks=[TaskRef("default/capital-france")]), + taskset=TasksetInput(tasks=[TaskRef("capital-france")]), ) diff --git a/skills/nemo-evaluator-plugin/references/api-auth.md b/skills/nemo-evaluator-plugin/references/api-auth.md index e49701938b..9e4fec5395 100644 --- a/skills/nemo-evaluator-plugin/references/api-auth.md +++ b/skills/nemo-evaluator-plugin/references/api-auth.md @@ -10,7 +10,7 @@ durable platform job. | Execution | `api_key_secret` resolves to | | --- | --- | | Standalone SDK | Environment-variable name in the calling process, such as `NVIDIA_API_KEY` | -| Plugin `submit` | NeMo Platform secret name in the target workspace, such as `nvidia-api-key` | +| Plugin `evaluate_dataset` | NeMo Platform secret name in the target workspace, such as `nvidia-api-key` | A remote job cannot read the submitting shell's environment variables. Before submitting, verify the `api_key_secret` is in the list of secrets: diff --git a/skills/nemo-evaluator-plugin/references/execution.md b/skills/nemo-evaluator-plugin/references/execution.md index df3d4db52d..01c6dc3ae7 100644 --- a/skills/nemo-evaluator-plugin/references/execution.md +++ b/skills/nemo-evaluator-plugin/references/execution.md @@ -15,11 +15,11 @@ Use the standalone SDK for the fastest in-process metric loop: ```python from nemo_evaluator_sdk import Evaluator, ExactMatchMetric -result = Evaluator().run_sync( - metrics=ExactMatchMetric( +result = Evaluator().run_dataset_sync( + metrics=[ExactMatchMetric( reference="{{item.expected}}", candidate="{{item.output}}", - ), + )], dataset=[ {"expected": "Paris", "output": "Paris"}, {"expected": "Paris", "output": "London"}, @@ -40,23 +40,23 @@ uv run nemo evaluator evaluate submit \ **Platform Python SDK** -Use `client.evaluator.submit` for execution through the installed nemo-evaluator-plugin: +Use `client.evaluator.evaluate_dataset` for execution through the installed nemo-evaluator-plugin: ```python from nemo_evaluator_sdk import ExactMatchMetric, RunConfig from nemo_platform import NeMoPlatform client = NeMoPlatform(base_url="http://localhost:8080", workspace="default") -job = client.evaluator.submit( - metric=ExactMatchMetric( +job = client.evaluator.evaluate_dataset( + metrics=[ExactMatchMetric( reference="{{item.expected}}", candidate="{{item.output}}", - ), + )], dataset=[ {"expected": "Paris", "output": "Paris"}, {"expected": "Paris", "output": "London"}, ], - config=RunConfig(parallelism=2), + params=RunConfig(parallelism=2), ) job.wait_until_done() result = job.get_result() @@ -108,14 +108,14 @@ target = Model( api_key_secret=SecretRef(root="nvidia-api-key"), ) -job = client.evaluator.submit( - metric=ExactMatchMetric(reference="{{item.expected}}"), +job = client.evaluator.evaluate_dataset( + metrics=[ExactMatchMetric(reference="{{item.expected}}")], dataset=[{"question": "Capital of France?", "expected": "Paris"}], target=target, prompt_template={ "messages": [{"role": "user", "content": "{{item.question}}"}], }, - config=RunConfigOnlineModel(parallelism=2), + params=RunConfigOnlineModel(parallelism=2), ) job.wait_until_done() result = job.get_result() @@ -178,10 +178,10 @@ Use field_mapping when a metric or online prompt uses canonical evaluator fields **Platform SDK** ```python -job = client.evaluator.submit( - metric=metric, +job = client.evaluator.evaluate_dataset( + metrics=[metric], dataset=dataset, - config=config, + params=config, target=target, prompt_template=prompt_template, ) @@ -222,7 +222,7 @@ operations. - Always wait for terminal completion. A metric can report 100 percent progress before the platform finishes publishing result artifacts. -- `submit` accepts a concrete `Model` or `ModelRef`; the platform resolves model +- `evaluate_dataset` accepts a concrete `Model` or `ModelRef`; the platform resolves model references in the target workspace. ## Multiple metrics @@ -262,8 +262,8 @@ metric submitted to a service, opt in explicitly: ```python from nemo_evaluator.shared.metric_bundles.hybrid import HybridMetricBundlePackager -job = client.evaluator.submit( - metric=custom_metric, +job = client.evaluator.evaluate_dataset( + metrics=[custom_metric], dataset=rows, metric_bundle_packager=HybridMetricBundlePackager(), ) diff --git a/skills/nemo-evaluator-plugin/references/metric-selection.md b/skills/nemo-evaluator-plugin/references/metric-selection.md index 98aa4d8c90..dd79a84686 100644 --- a/skills/nemo-evaluator-plugin/references/metric-selection.md +++ b/skills/nemo-evaluator-plugin/references/metric-selection.md @@ -81,7 +81,7 @@ SDK — pass a metric sequence in one call: ```python from nemo_evaluator_sdk import Evaluator -result = Evaluator().run_sync(metrics=[accuracy, style], dataset=rows) +result = Evaluator().run_dataset_sync(metrics=[accuracy, style], dataset=rows) ``` Platform job — put multiple stored metrics on the job spec: @@ -92,5 +92,5 @@ uv run nemo evaluator evaluate submit --spec \ ``` Each `metrics` entry may be an inline metric bundle, a stored `MetricRef`, or -a mix of both. The high-level `client.evaluator.submit` helper still accepts -only one runtime metric per call. +a mix of both. The high-level `client.evaluator.evaluate_dataset` helper takes a +list of runtime metrics and returns the job handle. diff --git a/skills/nemo-evaluator-plugin/references/resources.md b/skills/nemo-evaluator-plugin/references/resources.md index 826b02794e..0c3465d978 100644 --- a/skills/nemo-evaluator-plugin/references/resources.md +++ b/skills/nemo-evaluator-plugin/references/resources.md @@ -20,6 +20,7 @@ new versioned name. ```python from nemo_evaluator.api.schemas import ( + EvaluatorTaskDefinition, MetricRef, TaskInput, TaskInputs, @@ -43,9 +44,12 @@ client.evaluator.metrics.create( client.evaluator.tasks.create( "capital-france", task=TaskInput( - intent="Name the capital of France.", - inputs=TaskInputs(instruction="What is the capital of France?"), - metrics=[MetricRef("default/answer-exact")], + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Name the capital of France.", + inputs=TaskInputs(instruction="What is the capital of France?"), + metrics=[MetricRef("answer-exact")], + ), ), ) @@ -53,17 +57,16 @@ client.evaluator.tasksets.create( "geography", taskset=TasksetInput( description="Geography smoke tasks.", - tasks=[TaskRef("default/capital-france")], + tasks=[TaskRef("capital-france")], ), ) ``` -For a task that needs held-out ground truth invisible to the agent, keep the reference on an -inline `AgentEvalTaskInput` and use a metric that reads it: +For a task that needs held-out ground truth invisible to the agent, put it in `reference` and +use a metric that reads it. This works on a stored task, so it survives into taskset-driven runs: ```python -from nemo_evaluator.api.schemas import MetricRef, TaskInputs -from nemo_evaluator.jobs.agent_spec import AgentEvalTaskInput +from nemo_evaluator.api.schemas import EvaluatorTaskDefinition, MetricRef, TaskInput, TaskInputs from nemo_evaluator_sdk import ExactMatchMetric client.evaluator.metrics.create( @@ -74,19 +77,28 @@ client.evaluator.metrics.create( ), ) -inline_task = AgentEvalTaskInput( - id="capital-france", - intent="Name the capital of France.", - inputs=TaskInputs(instruction="What is the capital of France?"), - reference={"expected": "Paris"}, - metrics=[MetricRef("default/answer-from-reference")], +client.evaluator.tasks.create( + "capital-france-graded", + task=TaskInput( + spec=EvaluatorTaskDefinition( + kind="evaluator", + intent="Name the capital of France.", + inputs=TaskInputs(instruction="What is the capital of France?"), + reference={"expected": "Paris"}, + metrics=[MetricRef("answer-from-reference")], + ), + ), ) ``` +`reference` is surfaced to metrics but never seeded into the agent's workspace or shown to the +agent, so a metric can grade against artifacts the agent cannot edit. It is held out from the +*agent*, not from the API — anyone who can read the task can read it. It is covered by the revision +digest, so changing ground truth publishes a new revision. + Stored tasks keep metric references. Inline task metrics are normalized into -content-addressed derived metrics. The stored-task example uses an output-only -metric because stored tasks do not carry the grader-only `reference` field; use -an inline `AgentEvalTaskInput` when held-out per-task data is required. +content-addressed derived metrics. The same `reference` field is available on an inline +`AgentEvalTaskInput` for one-off submissions. ## Retrieve, list, and delete diff --git a/skills/nemo-evaluator-plugin/references/troubleshooting.md b/skills/nemo-evaluator-plugin/references/troubleshooting.md index 98a08bd85e..663139917c 100644 --- a/skills/nemo-evaluator-plugin/references/troubleshooting.md +++ b/skills/nemo-evaluator-plugin/references/troubleshooting.md @@ -17,7 +17,7 @@ nemo evaluator agent-evaluate explain | Symptom | Likely cause | Fix | | --- | --- | --- | | `No such command 'evaluation'` | The legacy generated CLI group is not the plugin surface | Use `nemo evaluator ...` | -| Guidance or `--help` references a local plugin `run` verb | That execution path is being retired | Use `submit`, or the standalone SDK for local iteration | +| Guidance or `--help` references a local plugin `run` verb | That execution path is being retired | Use `evaluate_dataset`, or the standalone SDK for local iteration | | Agent-eval metric fails every trial with a missing template key | The metric uses the dataset-driven `item.*` context in a task-driven run | Use `inputs.*`, `reference.*`, `task.*`, `trial.*`, or `sample.output_text` | | Spec validation error | Fields do not match the current job schema | Run the matching `explain` command and validate against the spec class before submission | | Dataset row has missing fields | Jinja templates or `field_mapping` do not match row keys | Inspect one row and every referenced template before rerunning | @@ -26,7 +26,7 @@ nemo evaluator agent-evaluate explain | Built-in metric bundle contains cloudpickle | A legacy or explicit packager was used | Regenerate with `InlineMetricBundlePackager` or the current default | | `cloudpickle metric payload was created with Python ...` (HTTP 422) | The bundle was created with a different Python major/minor runtime | For a built-in metric, regenerate the checked inline JSON spec; for an intentional custom metric, recreate the bundle with the worker's Python major/minor version | | Custom metric submission rejects the default packager | Shipping custom code requires explicit opt-in | Pass `HybridMetricBundlePackager()` (preferred) or `CloudpickleMetricBundlePackager()` | -| `ModelRef` fails with the standalone SDK | Model references are resolved by the platform submission path | Use a concrete `Model` with the standalone SDK or use `submit` with `ModelRef` | +| `ModelRef` fails with the standalone SDK | Model references are resolved by the platform submission path | Use a concrete `Model` with the standalone SDK or use `evaluate_dataset` with `ModelRef` | | Fileset evaluation cannot load data | The reference, fragment, or workspace is wrong | Verify the `FilesetRef` and access it through the same workspace | | Result download fails while progress shows 100% | Metric progress finished before the platform job finalized artifacts | Call `job.wait_until_done()` before `get_result()` or `download_artifacts()` | | Agent-eval rejects the spec | Both or neither of `target` and `trials` were provided | Provide exactly one |