diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml index dd4827c27d..e42f4c63ea 100644 --- a/plugins/nemo-evaluator/openapi/openapi.yaml +++ b/plugins/nemo-evaluator/openapi/openapi.yaml @@ -1099,6 +1099,17 @@ paths: default: created_at description: The field to sort by. To sort in decreasing order, use `-` in front of the field name. + - name: include_derived + in: query + required: false + schema: + type: boolean + description: Include derived (task-internal) metrics, which are hidden from + the listing by default. + default: false + title: Include Derived + description: Include derived (task-internal) metrics, which are hidden from + the listing by default. - in: query name: filter style: deepObject @@ -1236,6 +1247,188 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/evaluator/v2/workspaces/{workspace}/tasks: + get: + tags: + - Evaluator Plugin Tasks Routes + summary: List Tasks By Workspace + description: List stored tasks for a specific workspace. + operationId: list_tasks_apis_evaluator_v2_workspaces__workspace__tasks_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: page + in: query + required: false + schema: + type: integer + minimum: 1 + description: Page number. + default: 1 + title: Page + description: Page number. + - name: page_size + in: query + required: false + schema: + type: integer + maximum: 1000 + minimum: 1 + description: Page size. + default: 100 + title: Page Size + description: Page size. + - name: sort + in: query + required: false + schema: + allOf: + - $ref: '#/components/schemas/TaskSort' + description: The field to sort by. To sort in decreasing order, use `-` + in front of the field name. + default: created_at + description: The field to sort by. To sort in decreasing order, use `-` in + front of the field name. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/TaskFilter' + description: Filter tasks by workspace, name, created_at, and updated_at. + responses: + '200': + description: Return stored tasks for a workspace + content: + application/json: + schema: + $ref: '#/components/schemas/TasksPage' + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + /apis/evaluator/v2/workspaces/{workspace}/tasks/{name}: + post: + tags: + - Evaluator Plugin Tasks Routes + summary: Create Task + description: Store a new task, addressed by workspace/name. + operationId: create_task_apis_evaluator_v2_workspaces__workspace__tasks__name__post + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + maxLength: 255 + pattern: ^[\w\-\.]+$ + title: Name + - name: project + in: query + required: false + schema: + description: Optional project to associate with the task. + title: Project + type: string + description: Optional project to associate with the task. + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/TaskInput' + responses: + '201': + description: Store a new task + content: + application/json: + schema: + $ref: '#/components/schemas/Task' + '409': + description: Task already exists + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + get: + tags: + - Evaluator Plugin Tasks Routes + summary: Get Task + description: Get a stored task by workspace and name. + operationId: get_task_apis_evaluator_v2_workspaces__workspace__tasks__name__get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + responses: + '200': + description: Return stored task details + content: + application/json: + schema: + $ref: '#/components/schemas/Task' + '404': + description: Task not found + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + delete: + tags: + - Evaluator Plugin Tasks Routes + summary: Delete Task + description: Delete a stored task by workspace and name. + operationId: delete_task_apis_evaluator_v2_workspaces__workspace__tasks__name__delete + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + responses: + '204': + description: Delete a stored task + '404': + description: Task not found + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' components: schemas: Agent: @@ -1523,11 +1716,9 @@ components: title: Intent description: Human-readable description of the desired agent behavior. inputs: - additionalProperties: true - type: object - title: Inputs - description: What the agent receives or starts from (instruction, seed, - refs). + allOf: + - $ref: '#/components/schemas/TaskInputs' + description: The task's recognized input fields. views: additionalProperties: $ref: '#/components/schemas/SemanticView' @@ -1536,10 +1727,11 @@ components: description: Optional reporting views mapping this task's metric outputs into named semantic scores. metadata: - additionalProperties: true - type: object + items: + $ref: '#/components/schemas/MetadataItem' + type: array title: Metadata - description: Free-form metadata associated with the task. + description: Key/value annotations for the task. metrics: items: anyOf: @@ -1554,7 +1746,6 @@ components: required: - id - intent - - inputs title: AgentEvalTaskInput description: 'Submitter-facing task DTO: metrics may be inline bundles or stored-metric references.' @@ -1569,11 +1760,9 @@ components: title: Intent description: Human-readable description of the desired agent behavior. inputs: - additionalProperties: true - type: object - title: Inputs - description: What the agent receives or starts from (instruction, seed, - refs). + allOf: + - $ref: '#/components/schemas/TaskInputs' + description: The task's recognized input fields. views: additionalProperties: $ref: '#/components/schemas/SemanticView' @@ -1582,10 +1771,11 @@ components: description: Optional reporting views mapping this task's metric outputs into named semantic scores. metadata: - additionalProperties: true - type: object + items: + $ref: '#/components/schemas/MetadataItem' + type: array title: Metadata - description: Free-form metadata associated with the task. + description: Key/value annotations for the task. metrics: items: $ref: '#/components/schemas/MetricInline' @@ -1598,7 +1788,6 @@ components: required: - id - intent - - inputs title: AgentEvalTaskSpec description: 'Canonical task DTO: metrics fully resolved to inline bundles, reconstructed at run time.' @@ -2801,6 +2990,23 @@ components: built-in metric types.' JsonValue: title: JsonValue + MetadataItem: + properties: + key: + type: string + title: Key + description: Annotation key. + value: + type: string + title: Value + description: Annotation value. + additionalProperties: false + type: object + required: + - key + - value + title: MetadataItem + description: A single key/value annotation on a task. Metric: properties: id: @@ -2857,6 +3063,12 @@ components: type: string title: Bundle Ref description: Files reference to the canonical serialized bundle. + derived: + type: boolean + title: Derived + description: True for a content-addressed metric auto-stored from an inline + task metric (excluded from the default metric listing). + default: false created_at: type: string format: date-time @@ -2908,6 +3120,10 @@ components: description: Filter by description. title: Description type: string + derived: + description: Filter by derived flag. + title: Derived + type: boolean created_at: allOf: - $ref: '#/components/schemas/DatetimeFilter' @@ -3733,6 +3949,192 @@ components: type: array title: StringFilter type: object + Task: + properties: + id: + type: string + title: Id + description: Unique identifier for the stored task record. + name: + type: string + title: Name + description: "Task name \u2014 the stable task id, unique within its workspace." + workspace: + type: string + title: Workspace + description: Workspace the task belongs to. + project: + 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. + metadata: + items: + $ref: '#/components/schemas/MetadataItem' + type: array + title: Metadata + description: Key/value annotations for the task. + created_at: + type: string + format: date-time + title: Created At + description: Timestamp the task was created. + updated_at: + type: string + format: date-time + title: Updated At + description: Timestamp the task was last updated. + type: object + required: + - id + - name + - workspace + - intent + - created_at + - updated_at + title: Task + description: "API representation of a stored agent-eval task.\n\nMaps to the\ + \ SDK :class:`~nemo_evaluator_sdk.agent_eval.tasks.AgentEvalTask` \u2014 the\ + \ task's stable\n``id`` is the record ``name`` (unique within its workspace).\ + \ Metrics are stored in their wire form\n(inline bundles and/or references\ + \ to stored metrics); references resolve to inline at run time." + TaskFilter: + additionalProperties: false + description: Filter for task queries (top-level entity columns only; custom-field + filtering is a follow-up). + properties: + workspace: + description: Filter by workspace. + title: Workspace + type: string + name: + description: Filter by name. + title: Name + type: string + created_at: + allOf: + - $ref: '#/components/schemas/DatetimeFilter' + description: Filter by creation date. + updated_at: + allOf: + - $ref: '#/components/schemas/DatetimeFilter' + description: Filter by update date. + title: TaskFilter + 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. + metadata: + items: + $ref: '#/components/schemas/MetadataItem' + type: array + title: Metadata + description: Key/value annotations for the task. + additionalProperties: false + type: object + required: + - intent + 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``\ + \ shape minus server-owned\nfields (id, name, workspace, timestamps)." + TaskInputs: + properties: + instruction: + title: Instruction + description: The agent's instruction (its prompt). Falls back to the task + `intent` when unset. + type: string + additionalProperties: false + type: object + title: TaskInputs + description: '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.' + TaskSort: + type: string + enum: + - name + - -name + - created_at + - -created_at + - updated_at + - -updated_at + title: TaskSort + description: Sort fields for task queries. + TasksPage: + properties: + data: + items: + $ref: '#/components/schemas/Task' + type: array + title: Data + pagination: + allOf: + - $ref: '#/components/schemas/PaginationData' + description: Pagination information. + sort: + title: Sort + description: The field on which the results are sorted. + type: string + filter: + title: Filter + description: Filtering information. + additionalProperties: true + type: object + type: object + required: + - data + title: TasksPage ValidationError: properties: loc: diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/dependencies.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/dependencies.py index 3dfd8c163c..ca2c188a17 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/dependencies.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/dependencies.py @@ -8,6 +8,7 @@ from fastapi import Depends from nemo_evaluator.api.service.metric_service import MetricService from nemo_evaluator.api.service.result_service import ResultService +from nemo_evaluator.api.service.task_service import TaskService from nemo_platform import AsyncNeMoPlatform from nemo_platform_plugin.dependencies import get_entity_client, get_sdk_client from nemo_platform_plugin.entities import EntityClient @@ -26,3 +27,12 @@ def get_result_service( ) -> ResultService: """Provide a ResultService wired to the Entity Store (read-only over result entities).""" return ResultService(entity_client) + + +def get_task_service( + entity_client: EntityClient = Depends(get_entity_client), + metric_service: MetricService = Depends(get_metric_service), +) -> TaskService: + """Provide a TaskService. It uses the MetricService to normalize inline task metrics into + (derived) stored metrics, so a persisted task holds only references.""" + return TaskService(entity_client, metric_service) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py index 982323bfb1..a01e994840 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/schemas.py @@ -7,18 +7,19 @@ from datetime import datetime from enum import StrEnum -from typing import Annotated, Any, Literal +from typing import Annotated, Any, Literal, TypeAlias 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.schema import DatetimeFilter, Filter -from pydantic import BaseModel, ConfigDict, Field, field_validator +from pydantic import AfterValidator, BaseModel, ConfigDict, Field, RootModel, field_validator class DataFilter(Filter): @@ -132,6 +133,26 @@ class MetricInline(BaseModel): payload: MetricPayload = Field(description="Format-specific serialized metric.") +# A reference is ``name`` or ``workspace/name``, each segment using the platform name charset. +# Enforced on the field so empty/malformed refs are rejected at validation rather than during parsing. +_METRIC_REF_PATTERN = r"^[\w\-.]+(/[\w\-.]+)?$" + + +class MetricRef(RootModel[str]): + """Reference to a persisted metric (format: ``workspace/name`` or ``name``).""" + + root: str = Field( + pattern=_METRIC_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 Metric(BaseModel): """API representation of a stored metric. @@ -151,6 +172,11 @@ class Metric(BaseModel): payload_kind: str = Field(description="Payload discriminator of the stored bundle.") payload_digest: str = Field(description="Digest of the stored payload.") bundle_ref: str = Field(description="Files reference to the canonical serialized bundle.") + derived: bool = Field( + default=False, + description="True for a content-addressed metric auto-stored from an inline task metric " + "(excluded from the default metric listing).", + ) created_at: datetime = Field(description="Timestamp the metric was created.") updated_at: datetime = Field(description="Timestamp the metric was last updated.") @@ -173,6 +199,7 @@ class MetricFilter(DataFilter): name: str | None = Field(None, description="Filter by name.") metric_type: str | None = Field(None, description="Filter by metric type.") description: str | None = Field(None, description="Filter by description.") + derived: bool | None = Field(None, description="Filter by derived flag.") created_at: DatetimeFilter | None = Field(None, description="Filter by creation date.") updated_at: DatetimeFilter | None = Field(None, description="Filter by update date.") @@ -218,3 +245,108 @@ class EvaluateResult(_ResultBase): default=None, description="Reference to the dataset evaluated; None for an inline dataset." ) 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. + + Maps to the SDK :class:`~nemo_evaluator_sdk.agent_eval.tasks.AgentEvalTask` — the task's stable + ``id`` is the record ``name`` (unique within its workspace). Metrics are stored in their wire form + (inline bundles and/or references to stored metrics); references resolve to inline at run time. + """ + + id: str = Field(description="Unique identifier for the stored task record.") + 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." + ) + metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") + created_at: datetime = Field(description="Timestamp the task was created.") + updated_at: datetime = Field(description="Timestamp the task was last updated.") + + +class TaskInput(BaseModel): + """Create/replace body for a stored task (the name comes from the path). + + The authorable subset of :class:`Task` — the SDK ``AgentEvalTask`` shape minus server-owned + fields (id, name, workspace, timestamps). + """ + + 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." + ) + metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") + + +class TaskSort(StrEnum): + """Sort fields for task queries.""" + + NAME_ASC = "name" + NAME_DESC = "-name" + CREATED_AT_ASC = "created_at" + CREATED_AT_DESC = "-created_at" + UPDATED_AT_ASC = "updated_at" + UPDATED_AT_DESC = "-updated_at" + + +class TaskFilter(Filter): + """Filter for task queries (top-level entity columns only; custom-field filtering is a follow-up).""" + + workspace: str | None = Field(None, description="Filter by workspace.") + name: str | None = Field(None, description="Filter by name.") + created_at: DatetimeFilter | None = Field(None, description="Filter by creation date.") + updated_at: DatetimeFilter | None = Field(None, description="Filter by update date.") diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/service/metric_service.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/service/metric_service.py index 96e7efde6b..3b85638362 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/service/metric_service.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/service/metric_service.py @@ -20,16 +20,20 @@ from __future__ import annotations +import hashlib +import json import logging from nemo_evaluator.api.schemas import ( Metric, MetricInline, + MetricRef, ) from nemo_evaluator.entities import MetricBundleEntity from nemo_evaluator.metric_storage import delete_bundle_by_ref, store_bundle from nemo_evaluator.shared.metric_bundles.bundles import MetricBundle as RuntimeMetricBundle from nemo_platform import AsyncNeMoPlatform +from nemo_platform_plugin.api.filter import ComparisonOperation, FilterOperator, LogicalOperation from nemo_platform_plugin.entities import ( EntityClient, EntityConflictError, @@ -38,6 +42,15 @@ from nemo_platform_plugin.filter_ops import FilterOperation from nemo_platform_plugin.schema import Page, PaginationData +#: Reserved name prefix for content-addressed derived metrics (auto-stored from inline task metrics). +_DERIVED_METRIC_PREFIX = "derived." + +#: The entity store caps entity names at 63 characters (``^[a-z]...{1,62}...$``). The derived metric's +#: name is ``derived.``; we truncate the (hex) content digest to fit. The retained prefix is +#: far longer than needed to keep content-addressed dedup collision-free (e.g. 55 hex chars ≈ 220 bits). +_MAX_ENTITY_NAME_LENGTH = 63 +_DERIVED_DIGEST_LENGTH = _MAX_ENTITY_NAME_LENGTH - len(_DERIVED_METRIC_PREFIX) + logger = logging.getLogger(__name__) @@ -46,6 +59,21 @@ def _sanitize_for_log(value: object) -> str: return str(value).replace("\r", "").replace("\n", "") +def _and_exclude_derived(filter_operation: FilterOperation | None) -> FilterOperation: + """Combine an optional filter with "derived is not true", so derived metrics stay hidden. + + The filter grammar has no ``$ne``, so this is ``NOT(data.derived == True)`` — which also matches + metrics created before the ``derived`` field existed (the key is simply absent). + """ + not_derived = LogicalOperation( + operator=FilterOperator.NOT, + operations=[ComparisonOperation(field="data.derived", operator=FilterOperator.EQ, value=True)], + ) + if filter_operation is None: + return not_derived + return LogicalOperation(operator=FilterOperator.AND, operations=[filter_operation, not_derived]) + + def _entity_to_schema(entity: MetricBundleEntity) -> Metric: """Convert a stored metric entity to its API representation.""" created_at = entity.created_at @@ -65,6 +93,7 @@ def _entity_to_schema(entity: MetricBundleEntity) -> Metric: payload_kind=entity.payload_kind, payload_digest=entity.payload_digest, bundle_ref=entity.bundle_ref, + derived=entity.derived, created_at=created_at, updated_at=updated_at, ) @@ -77,6 +106,7 @@ def _entity_from_bundle( bundle: RuntimeMetricBundle, bundle_ref: str, project: str | None, + derived: bool = False, ) -> MetricBundleEntity: """Build a stored-metric entity from a bundle and its Files reference.""" return MetricBundleEntity( @@ -91,6 +121,7 @@ def _entity_from_bundle( payload_kind=bundle.payload.kind, payload_digest=bundle.payload.digest, bundle_ref=bundle_ref, + derived=derived, ) @@ -152,6 +183,46 @@ async def create_metric( ) return _entity_to_schema(created) + async def store_derived_metric(self, metric: MetricInline, *, workspace: str) -> MetricRef: + """Store an inline metric as a content-addressed *derived* metric and return a reference to it. + + Used when persisting a task that carries an inline metric: rather than embedding the bundle in + the task entity, we store it like any metric (Files-backed) but mark it ``derived`` (hidden + from the default metric listing) and name it by a digest of its *full* content, so identical + inline metrics across tasks dedupe to one stored bundle. + + The digest covers the whole bundle (metric_type, metadata, outputs, secrets, payload) — not + just ``payload.digest`` — so two metrics that share scoring code but differ in secrets or + output contracts get distinct names and are never silently collapsed onto each other. + """ + runtime_bundle = RuntimeMetricBundle.model_validate_json(metric.model_dump_json()) + canonical = json.dumps(runtime_bundle.model_dump(mode="json"), sort_keys=True, separators=(",", ":")) + content_digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + name = f"{_DERIVED_METRIC_PREFIX}{content_digest[:_DERIVED_DIGEST_LENGTH]}" + ref = MetricRef(f"{workspace}/{name}") + + # Content-addressed: if this exact bundle is already stored, reuse it (dedup). + try: + await self.entity_client.get(MetricBundleEntity, name=name, workspace=workspace) + return ref + except EntityNotFoundError: + pass + + bundle_ref = await store_bundle(self.sdk, workspace, name, runtime_bundle) + entity = _entity_from_bundle( + name=name, workspace=workspace, bundle=runtime_bundle, bundle_ref=bundle_ref, project=None, derived=True + ) + try: + await self.entity_client.create(entity) + except EntityConflictError: + # Raced another writer to the same content-addressed name; theirs is byte-identical, so + # drop the fileset we just uploaded and reuse the existing entry. + await self._discard_bundle(bundle_ref) + except Exception: + await self._discard_bundle(bundle_ref) + raise + return ref + async def get_metric(self, workspace: str, name: str) -> Metric | None: """Get a stored metric by workspace and name.""" try: @@ -167,8 +238,15 @@ async def list_metrics( page_size: int = 100, sort: str | None = None, filter_operation: FilterOperation | None = None, + include_derived: bool = False, ) -> Page[Metric]: - """List stored metrics with filtering and pagination.""" + """List stored metrics with filtering and pagination. + + Derived (task-internal) metrics are excluded unless ``include_derived`` is set — they're + addressable by reference but shouldn't clutter the curated metric listing. + """ + if not include_derived: + filter_operation = _and_exclude_derived(filter_operation) result = await self.entity_client.list( MetricBundleEntity, workspace=workspace, 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 new file mode 100644 index 0000000000..ced3afbe1f --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/service/task_service.py @@ -0,0 +1,140 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""CRUD service for persisted agent-eval task entities. + +A task is stored whole in the entity store (its metrics — inline bundles and/or stored-metric refs — +live in the entity record; there's no separate Files payload, unlike a metric bundle). Stored +``TaskEntity`` rows are mapped to the :class:`Task` API DTO — like ``MetricService`` maps +``MetricBundleEntity`` to ``Metric`` — so the wire contract round-trips cleanly (an ``EntityBase``'s +``id``/``created_at`` are computed and don't deserialize from the entity's own serialized form). +""" + +from __future__ import annotations + +import logging + +from nemo_evaluator.api.schemas import MetricInline, MetricRef, Task, TaskInput +from nemo_evaluator.api.service.metric_service import MetricService +from nemo_evaluator.entities import TaskEntity +from nemo_platform_plugin.entities import EntityClient, EntityConflictError, EntityNotFoundError, PaginationInfo +from nemo_platform_plugin.filter_ops import FilterOperation +from nemo_platform_plugin.schema import Page, PaginationData + +logger = logging.getLogger(__name__) + + +def _sanitize_for_log(value: object) -> str: + return str(value).replace("\r", "").replace("\n", "") + + +def _entity_to_task(entity: TaskEntity) -> Task: + """Map a stored task entity to its API DTO, guarding the persistence timestamps.""" + created_at = entity.created_at + updated_at = entity.updated_at + if created_at is None or updated_at is None: + raise ValueError(f"Stored task '{entity.workspace}/{entity.name}' is missing persistence timestamps") + return Task( + id=entity.id, + name=entity.name, + workspace=entity.workspace, + project=entity.project, + intent=entity.intent, + inputs=entity.inputs, + metrics=entity.metrics, + views=entity.views, + metadata=entity.metadata, + created_at=created_at, + updated_at=updated_at, + ) + + +def _pagination(src: PaginationInfo, current_page_size: int) -> PaginationData: + """Carry the entity-store pagination counts into the API ``Page`` envelope.""" + return PaginationData( + page=src.page, + page_size=src.page_size, + current_page_size=current_page_size, + total_pages=src.total_pages, + total_results=src.total_results, + ) + + +class TaskService: + """Create/get/list/delete for persisted agent-eval task entities, exposed as the ``Task`` DTO.""" + + def __init__(self, entity_client: EntityClient, metric_service: MetricService): + self.entity_client = entity_client + self.metric_service = metric_service + + async def _normalize_metrics(self, metrics: list[MetricRef | MetricInline], *, workspace: str) -> list[MetricRef]: + """Resolve a task's submitted metrics to references — inline metrics are stored as derived + metrics (content-addressed, hidden from the listing) so a persisted task only ever holds refs.""" + refs: list[MetricRef] = [] + for metric in metrics: + if isinstance(metric, MetricRef): + refs.append(metric) + else: + refs.append(await self.metric_service.store_derived_metric(metric, workspace=workspace)) + return refs + + async def create_task( + self, name: str, task_input: TaskInput, *, workspace: str, project: str | None = None + ) -> Task: + """Store a new task (addressed by workspace/name). Raises ``ValueError`` if it already exists.""" + entity = TaskEntity( + name=name, + workspace=workspace, + project=project, + intent=task_input.intent, + inputs=task_input.inputs, + metrics=await self._normalize_metrics(task_input.metrics, workspace=workspace), + views=task_input.views, + metadata=task_input.metadata, + ) + try: + created = await self.entity_client.create(entity) + except EntityConflictError as exc: + raise ValueError(f"Task '{workspace}/{name}' already exists") from exc + logger.info( + "Task created", extra={"workspace": _sanitize_for_log(workspace), "task_name": _sanitize_for_log(name)} + ) + return _entity_to_task(created) + + async def get_task(self, workspace: str, name: str) -> Task | None: + try: + entity = await self.entity_client.get(TaskEntity, workspace=workspace, name=name) + except EntityNotFoundError: + return None + return _entity_to_task(entity) + + async def list_tasks( + self, + *, + workspace: str, + page: int = 1, + page_size: int = 100, + sort: str | None = None, + filter_operation: FilterOperation | None = None, + ) -> Page[Task]: + result = await self.entity_client.list( + TaskEntity, + workspace=workspace, + filter_operation=filter_operation, + sort=sort, + page=page, + page_size=page_size, + ) + data = [_entity_to_task(entity) for entity in result.data] + return Page(data=data, pagination=_pagination(result.pagination, len(data)), sort=sort, filter=None) + + async def delete_task(self, workspace: str, name: str) -> bool: + """Delete a stored task; ``False`` if absent.""" + try: + await self.entity_client.delete(TaskEntity, name, workspace=workspace) + except EntityNotFoundError: + return False + logger.info( + "Task deleted", extra={"workspace": _sanitize_for_log(workspace), "task_name": _sanitize_for_log(name)} + ) + return True diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/v2/metrics.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/v2/metrics.py index 79bbc14026..ee4612d85d 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/api/v2/metrics.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/v2/metrics.py @@ -67,6 +67,10 @@ async def list_metrics( default=MetricSort.CREATED_AT_ASC, description="The field to sort by. To sort in decreasing order, use `-` in front of the field name.", ), + include_derived: bool = Query( + default=False, + description="Include derived (task-internal) metrics, which are hidden from the listing by default.", + ), parsed_filter: ParsedFilter = Depends(make_filter_dep(MetricFilter)), service: MetricService = Depends(get_metric_service), ) -> Page[Metric]: @@ -80,6 +84,7 @@ async def list_metrics( page_size=page_size, sort=sort, filter_operation=parsed_filter.operation, + include_derived=include_derived, ) except Exception: logger.exception(f"Failed to list metrics for workspace {_sanitize_for_log(workspace)}") diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/api/v2/tasks.py b/plugins/nemo-evaluator/src/nemo_evaluator/api/v2/tasks.py new file mode 100644 index 0000000000..fd3119c9de --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/api/v2/tasks.py @@ -0,0 +1,184 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""CRUD routes for stored agent-eval tasks under /apis/evaluator/v2/workspaces/{workspace}/tasks.""" + +from __future__ import annotations + +import logging +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, Path, Query, status +from nemo_evaluator.api.dependencies import get_task_service +from nemo_evaluator.api.schemas import Task, TaskFilter, TaskInput, TaskSort +from nemo_evaluator.api.service.task_service import TaskService +from nemo_evaluator.authz import scope +from nemo_evaluator.entities import MAX_NAME_LENGTH, NAME_PATTERN +from nemo_platform_plugin.api.parsed_filter import ParsedFilter, make_filter_dep +from nemo_platform_plugin.authz import CallerKind, PermissionSet, path_rule, perm +from nemo_platform_plugin.entities import EntityValidationError +from nemo_platform_plugin.jobs.openapi_utils import generate_openapi_extra_params +from nemo_platform_plugin.schema import Page + +logger = logging.getLogger(__name__) + + +class TaskPerms(PermissionSet, namespace="evaluator.tasks"): + """Permissions for the stored-task CRUD collection.""" + + CREATE = perm("Create a stored task") + LIST = perm("List stored tasks") + READ = perm("Read a stored task") + DELETE = perm("Delete a stored task") + + +def _sanitize_for_log(value: object) -> str: + """Prevent log injection by removing line-break/control characters.""" + return str(value).replace("\r", "").replace("\n", "") + + +router = APIRouter() + + +@router.get( + "/tasks", + summary="List Tasks By Workspace", + response_description="Return stored tasks for a workspace", + status_code=status.HTTP_200_OK, + response_model=Page[Task], + response_model_exclude_none=True, + openapi_extra=generate_openapi_extra_params( + filter_schema=TaskFilter, + filter_description="Filter tasks by workspace, name, created_at, and updated_at.", + ), +) +@scope.read +@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[TaskPerms.LIST]) +async def list_tasks( + workspace: str, + page: int = Query(default=1, ge=1, description="Page number."), + page_size: int = Query(default=100, ge=1, le=1000, description="Page size."), + sort: TaskSort = Query( + default=TaskSort.CREATED_AT_ASC, + description="The field to sort by. To sort in decreasing order, use `-` in front of the field name.", + ), + parsed_filter: ParsedFilter = Depends(make_filter_dep(TaskFilter)), + service: TaskService = Depends(get_task_service), +) -> Page[Task]: + """List stored tasks for a specific workspace.""" + # Discard any workspace override in the filter — always scope to the path workspace. + parsed_filter.remove("workspace") + try: + return await service.list_tasks( + workspace=workspace, + page=page, + page_size=page_size, + sort=sort, + filter_operation=parsed_filter.operation, + ) + except Exception: + logger.exception(f"Failed to list tasks for workspace {_sanitize_for_log(workspace)}") + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error") + + +@router.post( + "/tasks/{name}", + summary="Create Task", + response_description="Store a new task", + status_code=status.HTTP_201_CREATED, + responses={status.HTTP_409_CONFLICT: {"description": "Task already exists"}}, +) +@scope.write +@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[TaskPerms.CREATE]) +async def create_task( + workspace: str, + name: Annotated[str, Path(max_length=MAX_NAME_LENGTH, pattern=NAME_PATTERN)], + task: TaskInput, + project: str | None = Query(default=None, description="Optional project to associate with the task."), + service: TaskService = Depends(get_task_service), +) -> Task: + """Store a new task, addressed by workspace/name.""" + safe_workspace = _sanitize_for_log(workspace) + safe_name = _sanitize_for_log(name) + logger.info(f"Creating task: {safe_workspace}/{safe_name}") + try: + return await service.create_task(name, task, workspace=workspace, project=project) + except EntityValidationError as e: + logger.warning(f"Entity store validation error during task creation: {e}") + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e)) + except ValueError as e: + if "already exists" in str(e).lower(): + logger.warning(f"Task already exists: {safe_workspace}/{safe_name}") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"Task with workspace '{workspace}' and name '{name}' already exists", + ) + logger.warning(f"Task creation validation error: {e}") + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid task data") + except HTTPException: + raise + except Exception: + logger.exception("Failed to create task") + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error") + + +@router.get( + "/tasks/{name}", + summary="Get Task", + response_description="Return stored task details", + status_code=status.HTTP_200_OK, + responses={status.HTTP_404_NOT_FOUND: {"description": "Task not found"}}, +) +@scope.read +@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[TaskPerms.READ]) +async def get_task( + workspace: str, + name: str, + service: TaskService = Depends(get_task_service), +) -> Task: + """Get a stored task by workspace and name.""" + logger.debug(f"Getting task: {_sanitize_for_log(workspace)}/{_sanitize_for_log(name)}") + try: + task = await service.get_task(workspace, name) + if not task: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Task not found: {workspace}/{name}", + ) + return task + except HTTPException: + raise + except Exception: + logger.exception(f"Failed to get task {_sanitize_for_log(workspace)}/{_sanitize_for_log(name)}") + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error") + + +@router.delete( + "/tasks/{name}", + summary="Delete Task", + response_description="Delete a stored task", + status_code=status.HTTP_204_NO_CONTENT, + responses={status.HTTP_404_NOT_FOUND: {"description": "Task not found"}}, +) +@scope.write +@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[TaskPerms.DELETE]) +async def delete_task( + workspace: str, + name: str, + service: TaskService = Depends(get_task_service), +): + """Delete a stored task by workspace and name.""" + logger.info(f"Deleting task: {_sanitize_for_log(workspace)}/{_sanitize_for_log(name)}") + try: + deleted = await service.delete_task(workspace, name) + if not deleted: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Task not found: {workspace}/{name}", + ) + return None + except HTTPException: + raise + except Exception: + logger.exception(f"Failed to delete task {_sanitize_for_log(workspace)}/{_sanitize_for_log(name)}") + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error") diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/entities.py b/plugins/nemo-evaluator/src/nemo_evaluator/entities.py index 00415a77c9..d2145ebdf5 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/entities.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/entities.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Stored metric entity for the evaluator plugin. +"""Stored entities for the evaluator plugin (metrics and agent-eval tasks). A :class:`MetricBundleEntity` is the persisted, queryable index for a metric. The full executable :class:`~nemo_evaluator.shared.metric_bundles.bundles.MetricBundle` @@ -9,13 +9,19 @@ service; the entity stores only the lightweight, searchable projection plus a reference (``bundle_ref``) and integrity digest (``payload_digest``) that point back at the canonical copy. + +A :class:`TaskEntity` is the persisted form of an agent-eval task — the SDK +:class:`~nemo_evaluator_sdk.agent_eval.tasks.AgentEvalTask`, addressed by +``workspace/name`` (the task's stable id is the record name) and reusable across runs. """ from __future__ import annotations from typing import ClassVar +from nemo_evaluator.api.schemas import MetricRef, TaskInputs, TaskMetadataList 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 @@ -71,6 +77,12 @@ class MetricBundleEntity(EntityBase): description="Description captured from the bundled metric's metadata.", max_length=MAX_DESCRIPTION_LENGTH, ) + derived: bool = Field( + default=False, + description="True for a content-addressed metric auto-stored from an inline task metric. " + "Derived metrics are excluded from the default metric listing (they're task internals, not " + "curated metrics), but are addressable by reference.", + ) # --- Eval result entities ---------------------------------------------------- @@ -150,3 +162,28 @@ class EvaluateResultEntity(_EvalResultCommon, EntityBase): description="Runtime metric type names applied in the run (e.g. 'exact_match'). Not metric refs: " "by run time the submitted refs are resolved to inline bundles, so the originals aren't available." ) + + +class TaskEntity(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. + """ + + __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.", + ) + metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py index 4eb85f4b2e..b2c715234d 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py @@ -95,10 +95,12 @@ def _to_runtime_task(task: AgentEvalTaskSpec) -> AgentEvalTask: return AgentEvalTask( id=task.id, intent=task.intent, - inputs=task.inputs, + # The runtime task carries plain dicts; the typed DTOs collapse to them — recognized input + # keys only, and the key/value metadata pairs folded into a mapping. + inputs=task.inputs.model_dump(exclude_none=True), metrics=[_runtime_metric(metric) for metric in task.metrics], views=task.views, - metadata=task.metadata, + metadata={item.key: item.value for item in task.metadata}, ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py index f96aae0fa9..dea566c057 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py @@ -18,7 +18,7 @@ # payload kind so MetricBundle payloads round-trip through validation. import nemo_evaluator.shared.metric_bundles.cloudpickle # noqa: F401 import nemo_evaluator.shared.metric_bundles.inline # noqa: F401 -from nemo_evaluator.api.schemas import MetricInline +from nemo_evaluator.api.schemas import MetricInline, TaskInputs, TaskMetadataList from nemo_evaluator.jobs.metric_resolution import to_runtime_bundle, unresolved_model_refs from nemo_evaluator.metric_refs import MetricRefOrInline from nemo_evaluator.shared.metric_bundles.bundles import unbundle_metric @@ -97,12 +97,12 @@ class _AgentEvalTaskCommon(BaseModel): id: str = Field(description="Stable task identifier, unique within the task collection.") intent: str = Field(description="Human-readable description of the desired agent behavior.") - inputs: dict[str, Any] = Field(description="What the agent receives or starts from (instruction, seed, refs).") + inputs: TaskInputs = Field(default_factory=TaskInputs, description="The task's recognized input fields.") views: dict[str, SemanticView] = Field( default_factory=dict, description="Optional reporting views mapping this task's metric outputs into named semantic scores.", ) - metadata: dict[str, Any] = Field(default_factory=dict, description="Free-form metadata associated with the task.") + metadata: TaskMetadataList = Field(default_factory=list, description="Key/value annotations for the task.") class AgentEvalTaskInput(_AgentEvalTaskCommon): diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py b/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py index 66531a2f6e..6c2dd61f1f 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/metric_refs.py @@ -12,33 +12,18 @@ from __future__ import annotations -from typing import Any, TypeAlias +from typing import Any -from nemo_evaluator.api.schemas import MetricInline +# ``MetricRef`` / ``MetricRefOrInline`` are defined in ``api.schemas`` (next to ``MetricInline``) so +# 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 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.entities import EntityNotFoundError -from pydantic import Field, RootModel - -# A reference is ``name`` or ``workspace/name``, each segment using the platform -# name charset. Enforced on the field so empty/malformed refs are rejected at -# validation time rather than during parsing. -_METRIC_REF_PATTERN = r"^[\w\-.]+(/[\w\-.]+)?$" - - -class MetricRef(RootModel[str]): - """Reference to a persisted metric (format: ``workspace/name`` or ``name``).""" - - root: str = Field( - pattern=_METRIC_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. -MetricRefOrInline: TypeAlias = MetricInline | MetricRef def parse_metric_ref(root: str, default_workspace: str) -> tuple[str, str]: diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/metric_resources.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/metric_resources.py index 38a1645359..3db445ce4d 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/metric_resources.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/metric_resources.py @@ -26,13 +26,17 @@ from nemo_platform_plugin.schema import Page -def _list_params(page: int, page_size: int, sort: str | None, metric_type: str | None) -> dict[str, str | int]: +def _list_params( + page: int, page_size: int, sort: str | None, metric_type: str | None, include_derived: bool +) -> dict[str, str | int | bool]: """Build the list query string: paging/sort + the route's ``filter[metric_type]`` trait filter.""" - params: dict[str, str | int] = {"page": page, "page_size": page_size} + params: dict[str, str | int | bool] = {"page": page, "page_size": page_size} if sort is not None: params["sort"] = sort if metric_type is not None: params["filter[metric_type]"] = metric_type + if include_derived: + params["include_derived"] = True return params @@ -109,11 +113,15 @@ def list( page_size: int = 100, sort: str | None = None, metric_type: str | None = None, + include_derived: bool = False, ) -> Page[Metric]: - """List stored metrics in a workspace, optionally filtered by metric type.""" + """List stored metrics in a workspace, optionally filtered by metric type. + + Derived (task-internal) metrics are hidden unless ``include_derived`` is set. + """ response = self._http_client.get( self._collection_url(workspace), - params=_list_params(page, page_size, sort, metric_type), + params=_list_params(page, page_size, sort, metric_type, include_derived), headers=self._headers(), timeout=self._platform.timeout, ) @@ -185,11 +193,15 @@ async def list( page_size: int = 100, sort: str | None = None, metric_type: str | None = None, + include_derived: bool = False, ) -> Page[Metric]: - """List stored metrics in a workspace, optionally filtered by metric type.""" + """List stored metrics in a workspace, optionally filtered by metric type. + + Derived (task-internal) metrics are hidden unless ``include_derived`` is set. + """ response = await self._http_client.get( self._collection_url(workspace), - params=_list_params(page, page_size, sort, metric_type), + params=_list_params(page, page_size, sort, metric_type, include_derived), headers=self._headers(), timeout=self._platform.timeout, ) diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py index 6136fc06f6..5c8dac3213 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/resources.py @@ -29,6 +29,10 @@ EvaluatorAgentEvalResultsResource, EvaluatorEvalResultsResource, ) +from nemo_evaluator.sdk.task_resources import ( + AsyncEvaluatorTasksResource, + EvaluatorTasksResource, +) from nemo_evaluator.sdk.types import ( PluginDatasetInput, RunConfig, @@ -61,6 +65,7 @@ def __init__(self, platform: NeMoPlatform) -> None: self.metrics = EvaluatorMetricsResource(platform) self.agent_eval_results = EvaluatorAgentEvalResultsResource(platform) self.eval_results = EvaluatorEvalResultsResource(platform) + self.tasks = EvaluatorTasksResource(platform) def plugin_status(self) -> dict[str, object]: """Return evaluator plugin health information from the service.""" @@ -229,6 +234,7 @@ def __init__(self, platform: AsyncNeMoPlatform) -> None: self.metrics = AsyncEvaluatorMetricsResource(platform) self.agent_eval_results = AsyncEvaluatorAgentEvalResultsResource(platform) self.eval_results = AsyncEvaluatorEvalResultsResource(platform) + self.tasks = AsyncEvaluatorTasksResource(platform) async def plugin_status(self) -> dict[str, object]: """Return evaluator plugin health information from the service.""" diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/sdk/task_resources.py b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/task_resources.py new file mode 100644 index 0000000000..1973c0f419 --- /dev/null +++ b/plugins/nemo-evaluator/src/nemo_evaluator/sdk/task_resources.py @@ -0,0 +1,141 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""SDK resources for managing stored agent-eval tasks (``client.evaluator.tasks``). + +Thin client over the evaluator service's ``/tasks`` create/get/list/delete API. A task is sent as a +:class:`TaskInput` (its metrics inline and/or as references to stored metrics) and returned as the +:class:`Task` DTO; the service owns persistence in the entity store. +""" + +from __future__ import annotations + +from urllib.parse import quote + +from nemo_evaluator.api.schemas import Task, TaskInput +from nemo_evaluator.sdk import http_utils +from nemo_platform import AsyncNeMoPlatform, NeMoPlatform +from nemo_platform_plugin.schema import Page + + +def _list_params(page: int, page_size: int, sort: str | None) -> dict[str, str | int]: + params: dict[str, str | int] = {"page": page, "page_size": page_size} + if sort is not None: + params["sort"] = sort + return params + + +class EvaluatorTasksResource: + """Sync resource mounted as ``client.evaluator.tasks``.""" + + def __init__(self, platform: NeMoPlatform) -> None: + self._platform = platform + self._http_client = platform._client + + def _headers(self) -> dict[str, str]: + return http_utils.platform_default_headers(self._platform) + + def _collection_url(self, workspace: str | None) -> str: + return http_utils.url(self._platform, "/v2/workspaces/{workspace}/tasks", workspace) + + def _item_url(self, name: str, workspace: str | None) -> str: + return http_utils.url(self._platform, f"/v2/workspaces/{{workspace}}/tasks/{quote(name, safe='')}", workspace) + + def create(self, name: str, *, task: TaskInput, project: str | None = None, workspace: str | None = None) -> Task: + """Store a new task (addressed by workspace/name).""" + response = self._http_client.post( + self._item_url(name, workspace), + json=task.model_dump(mode="json"), + params={"project": project} if project is not None else None, + headers=self._headers(), + timeout=self._platform.timeout, + ) + response.raise_for_status() + return Task.model_validate(response.json()) + + def retrieve(self, name: str, *, workspace: str | None = None) -> Task: + """Get a stored task by name.""" + response = self._http_client.get( + self._item_url(name, workspace), headers=self._headers(), timeout=self._platform.timeout + ) + response.raise_for_status() + return Task.model_validate(response.json()) + + def list( + self, *, workspace: str | None = None, page: int = 1, page_size: int = 100, sort: str | None = None + ) -> Page[Task]: + """List stored tasks in a workspace.""" + response = self._http_client.get( + self._collection_url(workspace), + params=_list_params(page, page_size, sort), + headers=self._headers(), + timeout=self._platform.timeout, + ) + response.raise_for_status() + return Page[Task].model_validate(response.json()) + + def delete(self, name: str, *, workspace: str | None = None) -> None: + """Delete a stored task by name.""" + response = self._http_client.delete( + self._item_url(name, workspace), headers=self._headers(), timeout=self._platform.timeout + ) + response.raise_for_status() + + +class AsyncEvaluatorTasksResource: + """Async resource mounted as ``client.evaluator.tasks``.""" + + def __init__(self, platform: AsyncNeMoPlatform) -> None: + self._platform = platform + self._http_client = platform._client + + def _headers(self) -> dict[str, str]: + return http_utils.platform_default_headers(self._platform) + + def _collection_url(self, workspace: str | None) -> str: + return http_utils.url(self._platform, "/v2/workspaces/{workspace}/tasks", workspace) + + def _item_url(self, name: str, workspace: str | None) -> str: + return http_utils.url(self._platform, f"/v2/workspaces/{{workspace}}/tasks/{quote(name, safe='')}", workspace) + + async def create( + self, name: str, *, task: TaskInput, project: str | None = None, workspace: str | None = None + ) -> Task: + """Store a new task (addressed by workspace/name).""" + response = await self._http_client.post( + self._item_url(name, workspace), + json=task.model_dump(mode="json"), + params={"project": project} if project is not None else None, + headers=self._headers(), + timeout=self._platform.timeout, + ) + response.raise_for_status() + return Task.model_validate(response.json()) + + async def retrieve(self, name: str, *, workspace: str | None = None) -> Task: + """Get a stored task by name.""" + response = await self._http_client.get( + self._item_url(name, workspace), headers=self._headers(), timeout=self._platform.timeout + ) + response.raise_for_status() + return Task.model_validate(response.json()) + + async def list( + self, *, workspace: str | None = None, page: int = 1, page_size: int = 100, sort: str | None = None + ) -> Page[Task]: + """List stored tasks in a workspace.""" + response = await self._http_client.get( + self._collection_url(workspace), + params=_list_params(page, page_size, sort), + headers=self._headers(), + timeout=self._platform.timeout, + ) + response.raise_for_status() + return Page[Task].model_validate(response.json()) + + async def delete(self, name: str, *, workspace: str | None = None) -> None: + """Delete a stored task by name.""" + response = await self._http_client.delete( + self._item_url(name, workspace), headers=self._headers(), timeout=self._platform.timeout + ) + response.raise_for_status() diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/service.py b/plugins/nemo-evaluator/src/nemo_evaluator/service.py index f30890a182..ed4a2083b1 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/service.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/service.py @@ -10,6 +10,7 @@ from fastapi import APIRouter from nemo_evaluator.api.v2 import metrics as metrics_routes from nemo_evaluator.api.v2 import results as results_routes +from nemo_evaluator.api.v2 import tasks as tasks_routes from nemo_evaluator.authz import scope from nemo_evaluator.core import say_hello from nemo_evaluator.jobs.agent_evaluate import AgentEvalJob @@ -101,6 +102,13 @@ async def healthz() -> dict[str, object]: description="Queryable (row) evaluation result records.", prefix="/v2/workspaces/{workspace}", ), + RouterSpec( + # CRUD /apis/evaluator/v2/workspaces/{workspace}/tasks. + router=tasks_routes.router, + tag="Evaluator Plugin Tasks Routes", + description="Stored agent-eval task CRUD routes.", + prefix="/v2/workspaces/{workspace}", + ), ] diff --git a/plugins/nemo-evaluator/tests/api/service/test_metric_service.py b/plugins/nemo-evaluator/tests/api/service/test_metric_service.py index af8bb05467..0537c5b53d 100644 --- a/plugins/nemo-evaluator/tests/api/service/test_metric_service.py +++ b/plugins/nemo-evaluator/tests/api/service/test_metric_service.py @@ -188,3 +188,66 @@ async def test_list_returns_workspace_metrics(service: MetricService) -> None: assert {m.name for m in page.data} == {"a", "b"} assert page.pagination is not None assert page.pagination.total_results == 2 + + +# ---- derived metrics ------------------------------------------------------- + + +async def test_store_derived_metric_names_by_digest_and_marks_derived(service: MetricService) -> None: + from nemo_evaluator.api.service.metric_service import _MAX_ENTITY_NAME_LENGTH + + ref = await service.store_derived_metric(_bundle(), workspace="default") + + workspace, _, name = ref.root.partition("/") + assert workspace == "default" + assert name.startswith("derived.") + # The entity store caps names at 63 chars; the derived name must fit (it 422s otherwise). + assert len(name) <= _MAX_ENTITY_NAME_LENGTH + # Stored entity is flagged derived and Files-backed like any metric. + entity = service.entity_client.entities[("default", name)] + assert entity.derived is True + assert _fileset_of(service, entity.bundle_ref) in service.sdk._store + + +async def test_store_derived_metric_distinguishes_full_contract(service: MetricService) -> None: + # Two metrics with an identical payload but a differing bundle-level field (here: metadata) must + # NOT collapse — addressing on payload.digest alone would have silently rebound one onto the other. + bundle = _bundle() + variant = bundle.model_copy(update={"metadata": bundle.metadata.model_copy(update={"description": "different"})}) + assert bundle.payload.digest == variant.payload.digest # same executable payload... + + first = await service.store_derived_metric(bundle, workspace="default") + second = await service.store_derived_metric(variant, workspace="default") + + assert first.root != second.root # ...but distinct derived metrics, not one silently reused + assert len(service.entity_client.entities) == 2 + + +async def test_store_derived_metric_is_content_addressed_dedup(service: MetricService) -> None: + bundle = _bundle() + + first = await service.store_derived_metric(bundle, workspace="default") + second = await service.store_derived_metric(bundle, workspace="default") + + # Identical content collapses to one stored bundle (same ref, single entity, single fileset). + assert first.root == second.root + assert len(service.entity_client.entities) == 1 + assert len(service.sdk._store) == 1 + + +async def test_list_excludes_derived_by_default(service: MetricService) -> None: + captured: list[object] = [] + original_list = service.entity_client.list + + async def _spy(entity_cls, *, filter_operation=None, **kwargs): + captured.append(filter_operation) + return await original_list(entity_cls, filter_operation=filter_operation, **kwargs) + + service.entity_client.list = _spy + + await service.list_metrics("default") + await service.list_metrics("default", include_derived=True) + + # Default listing injects a filter (NOT derived); include_derived passes none through. + assert captured[0] is not None + assert captured[1] is None diff --git a/plugins/nemo-evaluator/tests/api/service/test_task_service.py b/plugins/nemo-evaluator/tests/api/service/test_task_service.py new file mode 100644 index 0000000000..87998dff0c --- /dev/null +++ b/plugins/nemo-evaluator/tests/api/service/test_task_service.py @@ -0,0 +1,165 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest +from nemo_evaluator.api.schemas import MetricInline, MetricRef, Task, TaskInput +from nemo_evaluator.api.service.task_service import TaskService +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 +from nemo_platform_plugin.entities import ( + EntityBase, + EntityConflictError, + EntityNotFoundError, + ListResponse, + PaginationInfo, +) + + +class _FakeMetricService: + """Records inline-metric normalization so we can assert a task stores refs, not bundles.""" + + def __init__(self) -> None: + self.stored: list[MetricInline] = [] + + async def store_derived_metric(self, metric: MetricInline, *, workspace: str) -> MetricRef: + self.stored.append(metric) + return MetricRef(f"{workspace}/derived.{metric.payload.digest}") + + +def _inline_metric() -> MetricInline: + bundle = bundle_metric( + ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}"), + CloudpickleMetricBundlePackager(), + ) + return MetricInline.model_validate(bundle.model_dump(mode="json")) + + +class _FakeEntityClient: + def __init__(self) -> None: + self.entities: dict[tuple[str, str, str], EntityBase] = {} + + async def create(self, entity): + key = (entity.__entity_type__, entity.workspace, entity.name) + if key in self.entities: + raise EntityConflictError(f"{key} exists") + now = datetime.now(timezone.utc) + entity._id = f"{entity.__entity_type__}-{entity.name}" + entity._created_at = now + entity._updated_at = now + self.entities[key] = entity + return entity + + async def get(self, entity_cls, *, workspace, name): + key = (entity_cls.__entity_type__, workspace, name) + if key not in self.entities: + raise EntityNotFoundError(f"{workspace}/{name} not found") + return self.entities[key] + + async def delete(self, entity_cls, name, *, workspace): + key = (entity_cls.__entity_type__, workspace, name) + if key not in self.entities: + raise EntityNotFoundError(f"{workspace}/{name} not found") + del self.entities[key] + + async def list(self, entity_cls, *, workspace, filter_operation=None, sort=None, page=1, page_size=100): + items = [ + e for (etype, ws, _), e in self.entities.items() if etype == entity_cls.__entity_type__ and ws == workspace + ] + return ListResponse( + data=items, + pagination=PaginationInfo( + page=page, + page_size=page_size, + current_page_size=len(items), + total_pages=1, + total_results=len(items), + ), + ) + + +def _task_input() -> TaskInput: + return TaskInput( + intent="Answer the question.", + inputs={"instruction": "What is 2+2?"}, + metrics=[MetricRef("default/stored-metric")], + metadata=[{"key": "suite", "value": "smoke"}], + ) + + +@pytest.fixture +def metric_service() -> _FakeMetricService: + return _FakeMetricService() + + +@pytest.fixture +def service(metric_service: _FakeMetricService) -> TaskService: + return TaskService(_FakeEntityClient(), metric_service) + + +async def test_create_then_get(service: TaskService) -> None: + created = await service.create_task("task-1", _task_input(), workspace="default") + + 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 created.created_at is not None + + got = await service.get_task("default", "task-1") + assert got is not None and got.name == "task-1" + + +async def test_create_normalizes_inline_metrics_to_refs( + service: TaskService, metric_service: _FakeMetricService +) -> None: + inline = _inline_metric() + task_input = TaskInput( + intent="Answer the question.", + inputs={"instruction": "What is 2+2?"}, + metrics=[MetricRef("default/stored-metric"), inline], + ) + + created = await service.create_task("task-1", task_input, workspace="default") + + # 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}" + + +async def test_create_rejects_duplicate(service: TaskService) -> None: + await service.create_task("task-1", _task_input(), workspace="default") + with pytest.raises(ValueError, match="already exists"): + await service.create_task("task-1", _task_input(), workspace="default") + + +async def test_get_returns_none_when_missing(service: TaskService) -> None: + assert await service.get_task("default", "nope") is None + + +async def test_list_returns_workspace_tasks(service: TaskService) -> None: + await service.create_task("a", _task_input(), workspace="default") + await service.create_task("b", _task_input(), workspace="default") + + page = await service.list_tasks(workspace="default") + + assert {t.name for t in page.data} == {"a", "b"} + assert page.pagination is not None and page.pagination.total_results == 2 + + +async def test_delete(service: TaskService) -> None: + await service.create_task("task-1", _task_input(), workspace="default") + assert await service.delete_task("default", "task-1") is True + assert await service.get_task("default", "task-1") is None + + +async def test_delete_returns_false_when_missing(service: TaskService) -> None: + assert await service.delete_task("default", "nope") is False diff --git a/plugins/nemo-evaluator/tests/api/v2/test_metrics_routes.py b/plugins/nemo-evaluator/tests/api/v2/test_metrics_routes.py index 879ed46845..900a4b118a 100644 --- a/plugins/nemo-evaluator/tests/api/v2/test_metrics_routes.py +++ b/plugins/nemo-evaluator/tests/api/v2/test_metrics_routes.py @@ -177,6 +177,7 @@ def test_metric_filter_translates_custom_fields_to_data_namespace() -> None: assert MetricFilter._get_entity_field_map() == { "metric_type": "data.metric_type", "description": "data.description", + "derived": "data.derived", } op = LogicalOperation( operator=FilterOperator.AND, diff --git a/plugins/nemo-evaluator/tests/api/v2/test_tasks_routes.py b/plugins/nemo-evaluator/tests/api/v2/test_tasks_routes.py new file mode 100644 index 0000000000..b7e8007e1e --- /dev/null +++ b/plugins/nemo-evaluator/tests/api/v2/test_tasks_routes.py @@ -0,0 +1,154 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""HTTP route-level tests for the /tasks CRUD endpoints. + +Drives the real FastAPI router + TaskService through a TestClient with an in-memory entity store. +Covers route wiring, the get_task_service dependency, and status-code mapping (201/204/404/409/422). +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from nemo_evaluator.api.dependencies import get_task_service +from nemo_evaluator.api.schemas import MetricRef, TaskInput +from nemo_evaluator.api.service.task_service import TaskService +from nemo_evaluator.api.v2 import tasks as tasks_routes +from nemo_platform_plugin.entities import ( + EntityBase, + EntityConflictError, + EntityNotFoundError, + ListResponse, + PaginationInfo, +) + + +class _FakeEntityClient: + def __init__(self) -> None: + self.entities: dict[tuple[str, str, str], EntityBase] = {} + + async def create(self, entity): + key = (entity.__entity_type__, entity.workspace, entity.name) + if key in self.entities: + raise EntityConflictError(f"{key} exists") + now = datetime.now(timezone.utc) + entity._id = f"{entity.__entity_type__}-{entity.name}" + entity._created_at = now + entity._updated_at = now + self.entities[key] = entity + return entity + + async def get(self, entity_cls, *, workspace, name): + key = (entity_cls.__entity_type__, workspace, name) + if key not in self.entities: + raise EntityNotFoundError(f"{workspace}/{name} not found") + return self.entities[key] + + async def delete(self, entity_cls, name, *, workspace): + key = (entity_cls.__entity_type__, workspace, name) + if key not in self.entities: + raise EntityNotFoundError(f"{workspace}/{name} not found") + del self.entities[key] + + async def list(self, entity_cls, *, workspace, filter_operation=None, sort=None, page=1, page_size=100): + items = [ + e for (etype, ws, _), e in self.entities.items() if etype == entity_cls.__entity_type__ and ws == workspace + ] + return ListResponse( + data=items, + pagination=PaginationInfo( + page=page, page_size=page_size, current_page_size=len(items), total_pages=1, total_results=len(items) + ), + ) + + +class _FakeMetricService: + """Normalizes inline metrics to derived refs; the route tests submit refs only, so it's unused.""" + + async def store_derived_metric(self, metric, *, workspace: str) -> MetricRef: + return MetricRef(f"{workspace}/derived.{metric.payload.digest}") + + +@pytest.fixture +def client() -> TestClient: + app = FastAPI() + app.include_router(tasks_routes.router, prefix="/v2/workspaces/{workspace}") + service = TaskService(_FakeEntityClient(), _FakeMetricService()) + app.dependency_overrides[get_task_service] = lambda: service + return TestClient(app) + + +def _body() -> dict: + return TaskInput( + intent="Answer the question.", + inputs={"instruction": "What is 2+2?"}, + metrics=[MetricRef("default/stored-metric")], + ).model_dump(mode="json") + + +_BASE = "/v2/workspaces/default/tasks" + + +def test_create_then_get(client: TestClient) -> None: + resp = client.post(f"{_BASE}/task-1", json=_body()) + assert resp.status_code == 201 + assert resp.json()["name"] == "task-1" + + 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 + + +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" + assert client.post(f"{_BASE}/task-1", json=body).status_code == 422 + + +def test_create_rejects_duplicate_metadata_keys(client: TestClient) -> None: + # metadata is a key→value map as a list; duplicate keys are a 422, not a silent last-wins collapse. + body = _body() + body["metadata"] = [{"key": "suite", "value": "smoke"}, {"key": "suite", "value": "regression"}] + assert client.post(f"{_BASE}/task-1", json=body).status_code == 422 + + +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 + + +def test_create_rejects_invalid_name(client: TestClient) -> None: + # NAME_PATTERN forbids slashes/spaces. + assert client.post(f"{_BASE}/bad name", json=_body()).status_code == 422 + + +def test_get_missing_returns_404(client: TestClient) -> None: + assert client.get(f"{_BASE}/nope").status_code == 404 + + +def test_list_returns_created_tasks(client: TestClient) -> None: + client.post(f"{_BASE}/a", json=_body()) + client.post(f"{_BASE}/b", json=_body()) + + resp = client.get(_BASE) + assert resp.status_code == 200 + body = resp.json() + assert {t["name"] for t in body["data"]} == {"a", "b"} + assert body["pagination"]["total_results"] == 2 + + +def test_delete_then_get_404(client: TestClient) -> None: + client.post(f"{_BASE}/task-1", json=_body()) + assert client.delete(f"{_BASE}/task-1").status_code == 204 + assert client.get(f"{_BASE}/task-1").status_code == 404 + + +def test_delete_missing_returns_404(client: TestClient) -> None: + assert client.delete(f"{_BASE}/nope").status_code == 404 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 55d4a93b35..d9fa77b3fc 100644 --- a/plugins/nemo-evaluator/tests/integration/test_agent_evaluate_job.py +++ b/plugins/nemo-evaluator/tests/integration/test_agent_evaluate_job.py @@ -163,7 +163,7 @@ def test_run_local_model_target_scores_a_real_trial(subprocess_platform: str) -> AgentEvalTaskInput( id="ask", intent="Obtain a one-word reply from the model.", - inputs={"prompt": "Reply with the single word DONE and nothing else."}, + inputs={"instruction": "Reply with the single word DONE and nothing else."}, metrics=[_output_contains_metric("DONE")], ) ], @@ -207,7 +207,7 @@ def test_run_local_agent_target_scores_a_real_trial(subprocess_platform: str) -> AgentEvalTaskInput( id="ask", intent="Obtain a one-word reply from the agent.", - inputs={"prompt": "Reply with the single word DONE and nothing else."}, + inputs={"instruction": "Reply with the single word DONE and nothing else."}, metrics=[_output_contains_metric("DONE")], ) ], @@ -395,7 +395,7 @@ def test_submit_model_target_under_auth_forwards_identity_to_igw(auth_subprocess AgentEvalTaskInput( id="ask", intent="Obtain a one-word reply from the model.", - inputs={"prompt": "Reply with the single word DONE and nothing else."}, + inputs={"instruction": "Reply with the single word DONE and nothing else."}, metrics=[_output_contains_metric("DONE")], ) ], diff --git a/plugins/nemo-evaluator/tests/integration/test_task_derived_metrics.py b/plugins/nemo-evaluator/tests/integration/test_task_derived_metrics.py new file mode 100644 index 0000000000..b377e1efe4 --- /dev/null +++ b/plugins/nemo-evaluator/tests/integration/test_task_derived_metrics.py @@ -0,0 +1,106 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration test for task inline-metric normalization through the SDK against a real platform. + +Persisting a task with an inline metric must offload that metric to a *derived* stored metric +(content-addressed, Files-backed) and leave the task holding only a reference. This exercises the +whole path end-to-end against a real entity store + Files service — the part the unit tests fake: + +- an inline task metric becomes a ``default/derived.`` reference on the stored task; +- two tasks carrying byte-identical inline metrics dedupe to the *same* derived metric; +- the derived metric is real (retrievable, ``derived=True``, Files-backed); +- it is hidden from the default ``/metrics`` listing but visible with ``include_derived``. + +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 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 +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" + + +def _unique(prefix: str) -> str: + return f"{prefix}-{uuid.uuid4().hex[:8]}" + + +def _inline_metric(marker: str) -> MetricInline: + """A valid metric whose packaged bytes are unique per ``marker`` (so the derived digest is fresh).""" + bundle = bundle_metric( + # The literal suffix only perturbs the template text — it keeps the metric valid while making + # this run's content (and therefore its content-addressed derived name) distinct from others'. + ExactMatchMetric(reference="{{item.expected}}", candidate="{{item.output}}" + marker), + CloudpickleMetricBundlePackager(), + ) + return MetricInline.model_validate_json(bundle.model_dump_json()) + + +def _task_input(metric: MetricInline) -> TaskInput: + return TaskInput(intent="Answer the question.", inputs={"instruction": "What is 2+2?"}, metrics=[metric]) + + +@pytest.mark.timeout(300) +def test_inline_task_metric_normalizes_to_derived_metric(subprocess_platform: str) -> None: + client = NeMoPlatform(base_url=subprocess_platform, max_retries=2) + client.workspaces.create(name=WORKSPACE, exist_ok=True) + + inline = _inline_metric(_unique("marker")) + task_a = _unique("task-a") + task_b = _unique("task-b") + derived_name: str | None = None + 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 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 + + # The derived metric is a real, Files-backed, flagged metric. + fetched = client.evaluator.metrics.retrieve(derived_name, workspace=WORKSPACE) + assert fetched.derived is True + assert fetched.bundle_ref + + # Hidden from the curated default listing... + default_names = {m.name for m in client.evaluator.metrics.list(workspace=WORKSPACE, page_size=1000).data} + assert derived_name not in default_names + + # ...but addressable when explicitly included, exactly once (content-addressed dedup). + with_derived = client.evaluator.metrics.list(workspace=WORKSPACE, include_derived=True, page_size=1000).data + matching = [m for m in with_derived if m.name == derived_name] + assert len(matching) == 1 + assert matching[0].derived is True + finally: + for name in (task_a, task_b): + try: + client.evaluator.tasks.delete(name, workspace=WORKSPACE) + except Exception: + pass + if derived_name is not None: + try: + client.evaluator.metrics.delete(derived_name, workspace=WORKSPACE) + except Exception: + pass diff --git a/plugins/nemo-evaluator/tests/sdk/test_metric_sdk_resources.py b/plugins/nemo-evaluator/tests/sdk/test_metric_sdk_resources.py index b465225083..4a751c7537 100644 --- a/plugins/nemo-evaluator/tests/sdk/test_metric_sdk_resources.py +++ b/plugins/nemo-evaluator/tests/sdk/test_metric_sdk_resources.py @@ -158,6 +158,20 @@ def test_sync_list_encodes_metric_type_filter_and_sort() -> None: assert params["sort"] == "-created_at" +def test_sync_list_omits_include_derived_unless_requested() -> None: + # Derived (task-internal) metrics are hidden by default: the param is only sent when explicitly set, + # so the default listing matches the route's own default without a redundant query arg. + http_client = MagicMock() + http_client.get.return_value = _response({"data": []}) + resource = EvaluatorMetricsResource(_platform(http_client)) + + resource.list() + assert "include_derived" not in http_client.get.call_args.kwargs["params"] + + resource.list(include_derived=True) + assert http_client.get.call_args.kwargs["params"]["include_derived"] is True + + def test_sync_delete_issues_delete_request() -> None: http_client = MagicMock() http_client.delete.return_value = _response({}) diff --git a/plugins/nemo-evaluator/tests/sdk/test_task_sdk_resources.py b/plugins/nemo-evaluator/tests/sdk/test_task_sdk_resources.py new file mode 100644 index 0000000000..f09f7a99f3 --- /dev/null +++ b/plugins/nemo-evaluator/tests/sdk/test_task_sdk_resources.py @@ -0,0 +1,120 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the client.evaluator.tasks SDK resources (mocked HTTP).""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from nemo_evaluator.api.schemas import MetricRef, Task, TaskInput +from nemo_evaluator.sdk.task_resources import AsyncEvaluatorTasksResource, EvaluatorTasksResource + +_BASE = "http://localhost:8080/apis/evaluator/v2/workspaces/default" + + +def _task_payload(name: str) -> dict[str, Any]: + now = datetime.now(timezone.utc) + return Task( + id=f"task-{name}", + name=name, + workspace="default", + intent="Answer the question.", + inputs={"instruction": "What is 2+2?"}, + metrics=[MetricRef("default/stored-metric")], + created_at=now, + updated_at=now, + ).model_dump(mode="json") + + +def _task_input() -> TaskInput: + return TaskInput(intent="Answer.", inputs={"instruction": "x"}, metrics=[MetricRef("default/stored-metric")]) + + +def _response(payload: Any) -> MagicMock: + response = MagicMock() + response.json.return_value = payload + response.raise_for_status.return_value = None + return response + + +def _platform(http_client: Any) -> MagicMock: + platform = MagicMock() + platform._client = http_client + platform.base_url = "http://localhost:8080" + platform.workspace = "default" + platform.default_headers = {} + platform.timeout = 30 + return platform + + +def test_sync_create_posts_task_input_to_item_url() -> None: + http_client = MagicMock() + http_client.post.return_value = _response(_task_payload("task-1")) + resource = EvaluatorTasksResource(_platform(http_client)) + + result = resource.create("task-1", task=_task_input()) + + 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." + + +def test_sync_retrieve_targets_item_url_and_parses_dto() -> None: + http_client = MagicMock() + http_client.get.return_value = _response(_task_payload("task-1")) + resource = EvaluatorTasksResource(_platform(http_client)) + + result = resource.retrieve("task-1") + + assert isinstance(result, Task) + assert isinstance(result.metrics[0], MetricRef) + assert http_client.get.call_args[0][0] == f"{_BASE}/tasks/task-1" + + +def test_sync_list_parses_page() -> None: + http_client = MagicMock() + http_client.get.return_value = _response( + { + "data": [_task_payload("a"), _task_payload("b")], + "pagination": { + "page": 1, + "page_size": 100, + "current_page_size": 2, + "total_pages": 1, + "total_results": 2, + }, + } + ) + resource = EvaluatorTasksResource(_platform(http_client)) + + page = resource.list(sort="-created_at") + + assert {t.name for t in page.data} == {"a", "b"} + assert http_client.get.call_args[0][0] == f"{_BASE}/tasks" + assert http_client.get.call_args.kwargs["params"]["sort"] == "-created_at" + + +def test_sync_delete_issues_delete_request() -> None: + http_client = MagicMock() + http_client.delete.return_value = _response({}) + resource = EvaluatorTasksResource(_platform(http_client)) + + resource.delete("task-1") + + assert http_client.delete.call_args[0][0] == f"{_BASE}/tasks/task-1" + + +async def test_async_retrieve_parses_dto() -> None: + http_client = MagicMock() + http_client.get = AsyncMock(return_value=_response(_task_payload("task-9"))) + resource = AsyncEvaluatorTasksResource(_platform(http_client)) + + result = await resource.retrieve("task-9") + + assert isinstance(result, Task) + assert result.name == "task-9" + assert http_client.get.call_args[0][0] == f"{_BASE}/tasks/task-9" diff --git a/plugins/nemo-evaluator/tests/test_agent_evaluate.py b/plugins/nemo-evaluator/tests/test_agent_evaluate.py index 8f62e22298..0b3dab9a63 100644 --- a/plugins/nemo-evaluator/tests/test_agent_evaluate.py +++ b/plugins/nemo-evaluator/tests/test_agent_evaluate.py @@ -66,7 +66,7 @@ def _task_spec() -> AgentEvalTaskSpec: return AgentEvalTaskSpec( id="task-1", intent="Answer the question.", - inputs={"prompt": "What is 2+2?"}, + inputs={"instruction": "What is 2+2?"}, metrics=[_inline_metric()], ) @@ -301,7 +301,7 @@ async def test_to_spec_resolves_inline_task_metrics_without_a_platform() -> None AgentEvalTaskInput( id="task-1", intent="Answer the question.", - inputs={"prompt": "What is 2+2?"}, + inputs={"instruction": "What is 2+2?"}, metrics=[_inline_metric()], ) ], @@ -439,7 +439,7 @@ def test_run_local_executes_each_target_type(target: Target, mocker: MockerFixtu input_spec = AgentEvalInputSpec( tasks=[ AgentEvalTaskInput( - id="task-1", intent="Answer.", inputs={"prompt": "What is 2+2?"}, metrics=[_inline_metric()] + id="task-1", intent="Answer.", inputs={"instruction": "What is 2+2?"}, metrics=[_inline_metric()] ) ], target=target, diff --git a/plugins/nemo-evaluator/tests/test_task_entity.py b/plugins/nemo-evaluator/tests/test_task_entity.py new file mode 100644 index 0000000000..86aa76005e --- /dev/null +++ b/plugins/nemo-evaluator/tests/test_task_entity.py @@ -0,0 +1,63 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Serialization round-trip tests for TaskEntity. + +The entity store persists custom fields with ``model_dump(exclude=base, mode="json")`` into a JSON +column and rebuilds them with ``model_validate``. This exercises that round-trip for the fields that +carry non-trivial nested types — the metric references and views. (A persisted task only ever holds +metric refs; inline metrics are normalized to derived stored metrics in the service before storage.) +""" + +from __future__ import annotations + +import json + +from nemo_evaluator.api.schemas import MetricRef +from nemo_evaluator.entities import TaskEntity +from nemo_evaluator_sdk.agent_eval.tasks import SemanticReducer, SemanticView, ViewSignal + + +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")], + ) + }, + metadata=[{"key": "suite", "value": "smoke"}], + ) + + +def _roundtrip(entity: TaskEntity) -> TaskEntity: + data = entity.model_dump(exclude=TaskEntity.__base_fields__, exclude_computed_fields=True, mode="json") + data = json.loads(json.dumps(data)) # prove JSON-serializable (the store uses a JSON column) + return TaskEntity.model_validate({"name": entity.name, "workspace": entity.workspace, **data}) + + +def test_roundtrip_preserves_task_fields() -> None: + entity = _entity() + + restored = _roundtrip(entity) + + assert restored.intent == "Answer the question." + assert restored.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" + # Nested SemanticView survives the JSON column. + assert restored.views == entity.views + + +def test_entity_type_is_task() -> None: + assert TaskEntity.__entity_type__ == "task"