feat(evaluator): persist agent-eval tasks as entities (AALGO-307)

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)

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.")

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.<digest>``; 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,