diff --git a/openapi/ga/individual/platform.openapi.yaml b/openapi/ga/individual/platform.openapi.yaml index 5376947a7e..ae79f72678 100644 --- a/openapi/ga/individual/platform.openapi.yaml +++ b/openapi/ga/individual/platform.openapi.yaml @@ -10846,6 +10846,21 @@ components: additionalProperties: type: string type: object + agent_name: + description: Filter evaluations that observed this agent name in any ingested + session. + title: Agent Name + type: string + agent_version: + description: Filter evaluations that observed this agent version in any + ingested session. + title: Agent Version + type: string + model_name: + description: Filter evaluations that observed this model name in any ingested + session. + title: Model Name + type: string run_count: allOf: - $ref: '#/components/schemas/NumberFilter' diff --git a/openapi/ga/openapi.yaml b/openapi/ga/openapi.yaml index 5376947a7e..ae79f72678 100644 --- a/openapi/ga/openapi.yaml +++ b/openapi/ga/openapi.yaml @@ -10846,6 +10846,21 @@ components: additionalProperties: type: string type: object + agent_name: + description: Filter evaluations that observed this agent name in any ingested + session. + title: Agent Name + type: string + agent_version: + description: Filter evaluations that observed this agent version in any + ingested session. + title: Agent Version + type: string + model_name: + description: Filter evaluations that observed this model name in any ingested + session. + title: Model Name + type: string run_count: allOf: - $ref: '#/components/schemas/NumberFilter' diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index 5376947a7e..ae79f72678 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -10846,6 +10846,21 @@ components: additionalProperties: type: string type: object + agent_name: + description: Filter evaluations that observed this agent name in any ingested + session. + title: Agent Name + type: string + agent_version: + description: Filter evaluations that observed this agent version in any + ingested session. + title: Agent Version + type: string + model_name: + description: Filter evaluations that observed this model name in any ingested + session. + title: Model Name + type: string run_count: allOf: - $ref: '#/components/schemas/NumberFilter' diff --git a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml index 5376947a7e..ae79f72678 100644 --- a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml +++ b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml @@ -10846,6 +10846,21 @@ components: additionalProperties: type: string type: object + agent_name: + description: Filter evaluations that observed this agent name in any ingested + session. + title: Agent Name + type: string + agent_version: + description: Filter evaluations that observed this agent version in any + ingested session. + title: Agent Version + type: string + model_name: + description: Filter evaluations that observed this model name in any ingested + session. + title: Model Name + type: string run_count: allOf: - $ref: '#/components/schemas/NumberFilter' diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/evaluations/evaluation_filter_param.py b/sdk/python/nemo-platform/src/nemo_platform/types/evaluations/evaluation_filter_param.py index 80f4e5e88b..88d81b4c53 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/evaluations/evaluation_filter_param.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/evaluations/evaluation_filter_param.py @@ -30,6 +30,12 @@ class EvaluationFilterParam(TypedDict, total=False): """Filter for listing Evaluations.""" + agent_name: str + """Filter evaluations that observed this agent name in any ingested session.""" + + agent_version: str + """Filter evaluations that observed this agent version in any ingested session.""" + cost_usd: MetricStatFiltersParam """Numeric range filters keyed by rollup aggregate stat. @@ -102,6 +108,9 @@ class EvaluationFilterParam(TypedDict, total=False): filter[metadata.model]=claude-opus-4-8. """ + model_name: str + """Filter evaluations that observed this model name in any ingested session.""" + name: str """Filter evaluations by name.""" diff --git a/sdk/python/nemo-platform/tests/api_resources/test_evaluations.py b/sdk/python/nemo-platform/tests/api_resources/test_evaluations.py index 1623e172d3..699b3ff41a 100644 --- a/sdk/python/nemo-platform/tests/api_resources/test_evaluations.py +++ b/sdk/python/nemo-platform/tests/api_resources/test_evaluations.py @@ -253,6 +253,8 @@ def test_method_list_with_all_params(self, client: NeMoPlatform) -> None: evaluation = client.evaluations.list( workspace="workspace", filter={ + "agent_name": "agent_name", + "agent_version": "agent_version", "cost_usd": { "count": { "eq": 0, @@ -420,6 +422,7 @@ def test_method_list_with_all_params(self, client: NeMoPlatform) -> None: }, }, "metadata": {"foo": "string"}, + "model_name": "model_name", "name": "name", "run_count": { "eq": 0, @@ -971,6 +974,8 @@ async def test_method_list_with_all_params(self, async_client: AsyncNeMoPlatform evaluation = await async_client.evaluations.list( workspace="workspace", filter={ + "agent_name": "agent_name", + "agent_version": "agent_version", "cost_usd": { "count": { "eq": 0, @@ -1138,6 +1143,7 @@ async def test_method_list_with_all_params(self, async_client: AsyncNeMoPlatform }, }, "metadata": {"foo": "string"}, + "model_name": "model_name", "name": "name", "run_count": { "eq": 0, diff --git a/services/intake/src/nmp/intake/api/v2/experiments/dependencies.py b/services/intake/src/nmp/intake/api/v2/experiments/dependencies.py index d56db74624..be6d5ee569 100644 --- a/services/intake/src/nmp/intake/api/v2/experiments/dependencies.py +++ b/services/intake/src/nmp/intake/api/v2/experiments/dependencies.py @@ -8,6 +8,7 @@ from fastapi import Depends, Request from nmp.common.entities.client import EntityClient from nmp.common.service.dependencies import get_entity_client +from nmp.intake.experiments.denormalizer import EvaluationDenormalizer from nmp.intake.experiments.read_service import EvaluationReadService from nmp.intake.repository.clickhouse.evaluation_rollup import ClickHouseEvaluationRollupRepository from nmp.intake.repository.clickhouse.evaluation_session import ClickHouseEvaluationSessionRepository @@ -59,3 +60,12 @@ def get_evaluation_read_service( EvaluationReadServiceDep = Annotated[EvaluationReadService, Depends(get_evaluation_read_service)] + + +def get_denormalizer(request: Request) -> EvaluationDenormalizer | None: + """Reach the service-owned denormalizer from the request (absent if startup didn't create one).""" + service = getattr(request.app.state, "intake_service", None) or getattr(request.app.state, "service", None) + return getattr(service, "denormalizer", None) if service is not None else None + + +DenormalizerDep = Annotated[EvaluationDenormalizer | None, Depends(get_denormalizer)] diff --git a/services/intake/src/nmp/intake/api/v2/experiments/endpoints.py b/services/intake/src/nmp/intake/api/v2/experiments/endpoints.py index 4e6ed36fde..3156890652 100644 --- a/services/intake/src/nmp/intake/api/v2/experiments/endpoints.py +++ b/services/intake/src/nmp/intake/api/v2/experiments/endpoints.py @@ -24,7 +24,11 @@ from nmp.common.api.parsed_filter import ParsedFilter, make_filter_dep from nmp.common.api.utils import generate_openapi_extra_params from nmp.common.entities.client import EntityClient, EntityConflictError, EntityNotFoundError -from nmp.intake.api.v2.experiments.dependencies import EntityClientDep, EvaluationReadServiceDep +from nmp.intake.api.v2.experiments.dependencies import ( + DenormalizerDep, + EntityClientDep, + EvaluationReadServiceDep, +) from nmp.intake.api.v2.experiments.schemas import ( EvaluationFilter, EvaluationPatchRequest, @@ -44,6 +48,7 @@ # layer uses; only the entity's own field names (e.g. parent_experiment_id) reference Experiment directly. from nmp.intake.entities.experiments import Experiment as Evaluation from nmp.intake.entities.experiments import ExperimentGroup +from nmp.intake.experiments.denormalizer import EvaluationDenormalizer from nmp.intake.experiments.read_service import ( EvaluationNotFoundError, EvaluationRead, @@ -379,6 +384,7 @@ async def list_evaluations( request: Request, read_service: EvaluationReadServiceDep, parsed: EvaluationFilterDep, + denormalizer: DenormalizerDep, page: int = Query(default=1, ge=1, description="Page number."), page_size: int = Query(default=100, ge=1, le=1000, description="Page size."), sort: str | None = Query( @@ -415,6 +421,9 @@ async def list_evaluations( # Translate the exposed `experiment_group_id` filter into a membership match over `experiment_ids` # (plus the legacy scalar), so listing a group returns every evaluation that belongs to it. entity_operation = _rewrite_group_filter(entity_operation) + # Turn the scalar agent_name/agent_version/model_name params into $contains matches on the + # denormalized list facets, so a workspace-wide list can filter by name against the entity store. + entity_operation = _rewrite_facet_filters(entity_operation) # Compute-on-read: fetch the whole (entity-filtered) group, hydrate every rollup, then filter, sort, # and paginate in memory so a single request can sort/filter by a ClickHouse metric that lives # outside the entity store. Bounded to hundreds of evaluations per group (see _MAX_GROUP_EVALUATIONS). @@ -442,6 +451,7 @@ async def list_evaluations( "filter (e.g. experiment_group_id)." ), ) from exc + _enqueue_stale_denormalization(denormalizer, workspace=workspace, reads=result.evaluations) responses = [_to_evaluation_response(evaluation) for evaluation in result.evaluations] # A metric-backed sort or filter is meaningless without rollups: if hydration was skipped (ClickHouse # disabled or down) every metric value would be unset, so a metric sort would silently collapse to @@ -480,11 +490,13 @@ async def get_evaluation( workspace: str, name: str, read_service: EvaluationReadServiceDep, + denormalizer: DenormalizerDep, ) -> EvaluationResponse: try: evaluation = await read_service.get_evaluation(workspace=workspace, name=name) except EvaluationNotFoundError as exc: raise _evaluation_not_found_http_error(exc) from exc + _enqueue_stale_denormalization(denormalizer, workspace=workspace, reads=[evaluation]) return _to_evaluation_response(evaluation) @@ -1004,6 +1016,34 @@ def _rewrite_group_filter(operation: FilterOperation | None) -> FilterOperation return operation +# Denormalized list facets on the Evaluation entity whose scalar filter param means "list contains value". +_FACET_CONTAINS_FIELDS = frozenset({"data.agent_names", "data.agent_versions", "data.model_names"}) + + +def _rewrite_facet_filters(operation: FilterOperation | None) -> FilterOperation | None: + """Rewrite an equality on a denormalized name-facet field into a ``$contains`` membership match. + + The user-facing params (``agent_name``/``agent_version``/``model_name``) are scalars that parse to an + equality, but each is stored as a list of distinct observed names (``agent_names``/``agent_versions``/ + ``model_names``). "Matches this name" therefore means "the list contains it", mirroring how + ``experiment_id`` matches membership in ``experiment_ids``. + """ + if operation is None: + return None + if isinstance(operation, ComparisonOperation): + if operation.field in _FACET_CONTAINS_FIELDS and operation.operator == FilterOperator.EQ: + return ComparisonOperation(operator=FilterOperator.CONTAINS, field=operation.field, value=operation.value) + return operation + if isinstance(operation, LogicalOperation): + return LogicalOperation( + operator=operation.operator, + operations=[ + rewritten for op in operation.operations if (rewritten := _rewrite_facet_filters(op)) is not None + ], + ) + return operation + + def _apply_is_deleted_filter(parsed: ParsedFilter) -> None: """Append an ``is_deleted`` clause so list endpoints hide soft-deleted rows by default. @@ -1319,6 +1359,35 @@ def _to_evaluation_response(evaluation: EvaluationRead) -> EvaluationResponse: return response +def _enqueue_stale_denormalization( + denormalizer: EvaluationDenormalizer | None, + *, + workspace: str, + reads: list[EvaluationRead], +) -> None: + """Self-heal the denormalized name facets on read. + + When a read's live rollup names differ from the entity's stored facets, queue the evaluation for + the refresher. This backfills evaluations that were ingested before the facets existed (and corrects + any drift) the first time they're read, with no separate migration to run on each instance — the + live rollup was already fetched to build this response, so the comparison is free and the write is + deferred to the background worker (which re-checks and skips no-ops). + """ + if denormalizer is None: + return + for read in reads: + rollup = read.rollup + if rollup is None: + continue + entity = read.entity + if ( + entity.agent_names != rollup.agent_names + or entity.agent_versions != rollup.agent_versions + or entity.model_names != rollup.model_names + ): + denormalizer.mark_dirty(workspace=workspace, evaluation_id=entity.name) + + async def _evaluation_response_with_rollup( read_service: EvaluationReadServiceDep, *, diff --git a/services/intake/src/nmp/intake/api/v2/experiments/schemas.py b/services/intake/src/nmp/intake/api/v2/experiments/schemas.py index b1e6bd77ff..dfac0afbaf 100644 --- a/services/intake/src/nmp/intake/api/v2/experiments/schemas.py +++ b/services/intake/src/nmp/intake/api/v2/experiments/schemas.py @@ -368,6 +368,23 @@ class EvaluationFilter(Filter): default=None, description="Filter by a metadata key/value pair, e.g. filter[metadata.model]=claude-opus-4-8.", ) + # Name-facet filters. Each is a scalar that matches evaluations whose denormalized list facet + # *contains* the value (the endpoint rewrites the parsed equality into a $contains membership match, + # like experiment_id). These are entity-store predicates (workspace-scoped, indexed prefix), so they + # filter the whole workspace without touching ClickHouse. The facets are refreshed after ingest, so a + # just-ingested name can lag by up to denormalization_interval_seconds. + agent_name: Annotated[str | None, map_entity_field("data.agent_names")] = Field( + default=None, + description="Filter evaluations that observed this agent name in any ingested session.", + ) + agent_version: Annotated[str | None, map_entity_field("data.agent_versions")] = Field( + default=None, + description="Filter evaluations that observed this agent version in any ingested session.", + ) + model_name: Annotated[str | None, map_entity_field("data.model_names")] = Field( + default=None, + description="Filter evaluations that observed this model name in any ingested session.", + ) # Rollup-metric filters. These live in ClickHouse, not the entity store, so they're declared as # self-mapping namespaces (the path is left untranslated) and applied in the application layer # after rollup hydration rather than forwarded to Postgres. Stat sub-paths mirror the sort grammar: diff --git a/services/intake/src/nmp/intake/background_worker.py b/services/intake/src/nmp/intake/background_worker.py new file mode 100644 index 0000000000..f24f26db98 --- /dev/null +++ b/services/intake/src/nmp/intake/background_worker.py @@ -0,0 +1,41 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""A background task with a managed start/stop lifecycle. + +Runs a subclass's ``_run()`` coroutine as a single asyncio ``Task``: ``start()`` launches it and +``stop()`` signals it (via ``self._stopping``) and awaits its exit. It never cancels mid-run, so an +in-flight iteration finishes before ``stop()`` returns. Subclasses own what the loop actually does. + +Intake-agnostic; a candidate to move to ``nmp.common`` if another service wants the same lifecycle. +""" + +from __future__ import annotations + +import asyncio +from abc import ABC, abstractmethod + + +class BackgroundWorker(ABC): + """Runs ``_run()`` as a single asyncio task with a start/stop lifecycle.""" + + def __init__(self) -> None: + self._task: asyncio.Task[None] | None = None + self._stopping = asyncio.Event() + + def start(self) -> None: + if self._task is None: + self._stopping.clear() + self._task = asyncio.create_task(self._run()) + + async def stop(self) -> None: + # Signal the loop to exit and await it — we never cancel mid-run, so an in-flight iteration + # finishes before stop() returns. + self._stopping.set() + if self._task is not None: + await self._task + self._task = None + + @abstractmethod + async def _run(self) -> None: + """The worker loop. Must return promptly once ``self._stopping`` is set.""" diff --git a/services/intake/src/nmp/intake/config.py b/services/intake/src/nmp/intake/config.py index 46a5647674..08ad16a69f 100644 --- a/services/intake/src/nmp/intake/config.py +++ b/services/intake/src/nmp/intake/config.py @@ -86,3 +86,12 @@ class IntakeConfig(_BaseIntakeConfig): le=MAX_ATIF_MAX_SUBAGENT_DEPTH, description="Maximum number of trajectory levels accepted for recursive ATIF subagents.", ) + denormalization_interval_seconds: float = Field( + default=60.0, + gt=0, + description=( + "How often the background worker drains the dirty set and denormalizes the distinct " + "agent/model name fields from ClickHouse onto Evaluation entities. Bounds how stale those " + "fields can be after ingest; ingest never blocks on it." + ), + ) diff --git a/services/intake/src/nmp/intake/entities/experiments.py b/services/intake/src/nmp/intake/entities/experiments.py index fc72db8603..c22bd4f588 100644 --- a/services/intake/src/nmp/intake/entities/experiments.py +++ b/services/intake/src/nmp/intake/entities/experiments.py @@ -204,3 +204,23 @@ def _coerce_metadata(cls, value: Any) -> Any: "Pin state is workspace-shared: every user with workspace access sees the same pinned set." ), ) + + # System-managed denormalized name fields. These mirror the distinct name sets that the ClickHouse + # rollup already derives from this evaluation's sessions; a background refresher writes them here + # (see nmp.intake.experiments.denormalizer) so the workspace-wide Evaluations list can filter by + # agent/model name against the entity store ($contains) instead of scanning the session table. Not + # accepted on the create/update body (the request schemas omit them); default empty until refreshed. + # Unlike computed rollups, these are raw observed strings with no formula, so they never need a + # backfill/recompute when aggregation logic changes — only when a new name is ingested. + agent_names: list[str] = Field( + default_factory=list, + description="System-managed: distinct agent names observed across this evaluation's ingested sessions.", + ) + agent_versions: list[str] = Field( + default_factory=list, + description="System-managed: distinct agent versions observed across this evaluation's ingested sessions.", + ) + model_names: list[str] = Field( + default_factory=list, + description="System-managed: distinct model names observed across this evaluation's ingested sessions.", + ) diff --git a/services/intake/src/nmp/intake/experiments/denormalizer.py b/services/intake/src/nmp/intake/experiments/denormalizer.py new file mode 100644 index 0000000000..88284f714d --- /dev/null +++ b/services/intake/src/nmp/intake/experiments/denormalizer.py @@ -0,0 +1,142 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Denormalizes agent/model name fields from ClickHouse onto Evaluation entities. + +Ingest marks ``(workspace, evaluation_id)`` dirty; a background loop recomputes each touched +evaluation's rollup and writes the distinct ``agent_names``/``agent_versions``/``model_names`` sets onto +the (system-managed) fields of its Evaluation entity. That lets the Evaluations list filter by +agent/model name against the entity store (``$contains``) instead of scanning the ClickHouse session +table on every request. + +Only the name fields are denormalized, never the computed metric rollups: names are raw observed +strings with no formula, so a change to how an aggregate is computed can never invalidate them. + +The start/stop lifecycle comes from :class:`nmp.intake.background_worker.BackgroundWorker`; the +debouncing (a coalescing dirty set drained on a fixed cadence) lives here because it is specific to how +this worker refreshes evaluations. +""" + +from __future__ import annotations + +import asyncio +import logging + +from nmp.common.entities.client import EntityClient, EntityConflictError, EntityNotFoundError +from nmp.intake.background_worker import BackgroundWorker +from nmp.intake.entities.experiments import Experiment +from nmp.intake.repository.evaluation_rollup import EvaluationRollup, EvaluationRollupRepository + +logger = logging.getLogger(__name__) + +# stop()'s final flush can itself re-queue keys (a transient failure/conflict that would clear on +# retry), so it drains up to this many passes; anything still queued afterwards is logged. +_STOP_DRAIN_PASSES = 3 + + +class EvaluationDenormalizer(BackgroundWorker): + """Coalesces dirty evaluation ids and refreshes their denormalized name fields on a fixed cadence. + + A burst of :meth:`mark_dirty` calls for one evaluation within an interval collapses to a single + refresh, and marking is a plain, non-blocking set add — so the ingest and read hot paths are never + gated on the refresh work. + """ + + def __init__( + self, + *, + rollup_repository: EvaluationRollupRepository, + entity_client: EntityClient, + interval_seconds: float = 60.0, + ) -> None: + super().__init__() + self._rollup_repository = rollup_repository + self._entity_client = entity_client + self._interval_seconds = interval_seconds + self._dirty: set[tuple[str, str]] = set() + + def mark_dirty(self, *, workspace: str, evaluation_id: str) -> None: + """Queue an evaluation for refresh. Cheap and non-blocking; safe to call from the ingest path.""" + self._dirty.add((workspace, evaluation_id)) + + def pending(self) -> set[tuple[str, str]]: + """Return a copy of the currently-queued ``(workspace, evaluation_id)`` pairs (observability/tests).""" + return set(self._dirty) + + async def _run(self) -> None: + while not self._stopping.is_set(): + try: + # Interruptible sleep: wakes early when stop() sets the event so shutdown is prompt. + await asyncio.wait_for(self._stopping.wait(), timeout=self._interval_seconds) + except asyncio.TimeoutError: + pass # interval elapsed; time for a periodic flush + try: + await self.flush() + except Exception: + logger.exception("Evaluation denormalization cycle failed") + + async def stop(self) -> None: + await super().stop() + # The loop may have exited before a final flush, and a flush can itself re-queue keys (a + # transient failure/conflict), so drain what's left — bounded, so a persistent failure can't + # hang shutdown. + for _ in range(_STOP_DRAIN_PASSES): + if not self._dirty: + return + await self.flush() + if self._dirty: + logger.warning( + "Evaluation denormalizer stopped with %d evaluation(s) still queued after %d drain " + "passes; not retried before shutdown", + len(self._dirty), + _STOP_DRAIN_PASSES, + ) + + async def flush(self) -> None: + """Drain the dirty set and refresh each evaluation's name fields. Directly callable for tests.""" + if not self._dirty: + return + batch = self._dirty + self._dirty = set() + by_workspace: dict[str, list[str]] = {} + for workspace, evaluation_id in batch: + by_workspace.setdefault(workspace, []).append(evaluation_id) + for workspace, evaluation_ids in by_workspace.items(): + try: + await self._refresh_workspace(workspace, evaluation_ids) + except Exception: + # Re-queue the whole workspace batch for the next cycle (e.g. ClickHouse unavailable). + logger.exception("Failed to refresh evaluation names for workspace %s; re-queuing", workspace) + for evaluation_id in evaluation_ids: + self.mark_dirty(workspace=workspace, evaluation_id=evaluation_id) + + async def _refresh_workspace(self, workspace: str, evaluation_ids: list[str]) -> None: + rollups = await self._rollup_repository.get_rollups(workspace=workspace, evaluation_ids=evaluation_ids) + for evaluation_id in evaluation_ids: + rollup = rollups.get(evaluation_id) + if rollup is None: + continue + await self._write_names(workspace, evaluation_id, rollup) + + async def _write_names(self, workspace: str, evaluation_id: str, rollup: EvaluationRollup) -> None: + try: + evaluation = await self._entity_client.get(Experiment, name=evaluation_id, workspace=workspace) + except EntityNotFoundError: + # Deleted between ingest and refresh; nothing to update. + return + # Skip the write when nothing changed, so a burst of re-ingests that add no new names doesn't + # churn the entity store (and doesn't lose an optimistic-lock race for no reason). + if ( + evaluation.agent_names == rollup.agent_names + and evaluation.agent_versions == rollup.agent_versions + and evaluation.model_names == rollup.model_names + ): + return + evaluation.agent_names = rollup.agent_names + evaluation.agent_versions = rollup.agent_versions + evaluation.model_names = rollup.model_names + try: + await self._entity_client.update(evaluation) + except EntityConflictError: + # A concurrent user edit won the optimistic lock; re-queue for the next cycle. + self.mark_dirty(workspace=workspace, evaluation_id=evaluation_id) diff --git a/services/intake/src/nmp/intake/service.py b/services/intake/src/nmp/intake/service.py index 69608e47ee..4e38aadbb3 100644 --- a/services/intake/src/nmp/intake/service.py +++ b/services/intake/src/nmp/intake/service.py @@ -11,12 +11,15 @@ from nmp.common.service import RouterConfig, Service from nmp.intake.api.v2.experiments import endpoints as experiments from nmp.intake.config import IntakeConfig, should_provision_local_clickhouse +from nmp.intake.experiments.denormalizer import EvaluationDenormalizer from nmp.intake.local_clickhouse import ( DockerUnavailableError, LocalClickHouseProvisioningError, reconcile_local_clickhouse, stop_local_clickhouse, ) +from nmp.intake.repository.clickhouse.evaluation_rollup import ClickHouseEvaluationRollupRepository +from nmp.intake.repository.clickhouse.executor import ClickHouseExecutor from nmp.intake.spans.api import annotations, evaluator_results, sessions, spans, traces from nmp.intake.spans.clickhouse_client import ClickHouseSettings, ClickHouseSpanClient from nmp.intake.spans.ingest import atif, chat_completions, otlp @@ -34,6 +37,8 @@ def __init__(self): super().__init__(name="intake", module_name="nmp.intake") # The client is owned by the service lifecycle; it is absent before startup and after shutdown. self.clickhouse_client: ClickHouseSpanClient | None = None + # Background worker that denormalizes agent/model name fields onto Evaluation entities. + self.denormalizer: EvaluationDenormalizer | None = None self._local_clickhouse_data_dir: Path | None = None self._owns_local_clickhouse = False self._ready = False @@ -109,12 +114,28 @@ async def on_startup(self) -> None: ) self.clickhouse_client = ClickHouseSpanClient(settings) + # Start the background denormalizer. It needs a service-principal entity client (no request + # context) to write onto Evaluation entities; skip it if the entity client can't be built. + entity_client = self.dependency_provider.get_entity_client(as_service=self.name) + if entity_client is not None: + self.denormalizer = EvaluationDenormalizer( + rollup_repository=ClickHouseEvaluationRollupRepository(ClickHouseExecutor(self.clickhouse_client)), + entity_client=entity_client, + interval_seconds=cfg.denormalization_interval_seconds, + ) + self.denormalizer.start() + else: + logger.warning("Entity client unavailable; evaluation denormalizer not started") self._ready = True async def on_shutdown(self) -> None: """Close the client and stop the managed local ClickHouse container.""" self._ready = False + # Stop the denormalizer first: its final flush still needs the ClickHouse client below. + if self.denormalizer is not None: + await self.denormalizer.stop() + self.denormalizer = None try: if self.clickhouse_client is not None: await self.clickhouse_client.close() diff --git a/services/intake/src/nmp/intake/spans/api/dependencies.py b/services/intake/src/nmp/intake/spans/api/dependencies.py index 512b3c1ec9..261762af28 100644 --- a/services/intake/src/nmp/intake/spans/api/dependencies.py +++ b/services/intake/src/nmp/intake/spans/api/dependencies.py @@ -8,6 +8,7 @@ from fastapi import Depends, HTTPException, Request, status from nemo_platform import AsyncNeMoPlatform from nmp.common.service.dependencies import get_sdk_client +from nmp.intake.experiments.denormalizer import EvaluationDenormalizer from nmp.intake.repository.annotations import AnnotationsRepository from nmp.intake.repository.clickhouse.annotations import ClickHouseAnnotationsRepository from nmp.intake.repository.clickhouse.evaluator_results import ClickHouseEvaluatorResultsRepository @@ -105,3 +106,12 @@ def get_spans_service( SpansServiceDep = Annotated[IntakeSpansService, Depends(get_spans_service)] + + +def get_denormalizer(request: Request) -> EvaluationDenormalizer | None: + """Reach the service-owned denormalizer from the request (absent if startup didn't create one).""" + service = getattr(request.app.state, "intake_service", None) or getattr(request.app.state, "service", None) + return getattr(service, "denormalizer", None) if service is not None else None + + +DenormalizerDep = Annotated[EvaluationDenormalizer | None, Depends(get_denormalizer)] diff --git a/services/intake/src/nmp/intake/spans/ingest/atif.py b/services/intake/src/nmp/intake/spans/ingest/atif.py index 8c306c4ff9..72e7d369b7 100644 --- a/services/intake/src/nmp/intake/spans/ingest/atif.py +++ b/services/intake/src/nmp/intake/spans/ingest/atif.py @@ -11,7 +11,7 @@ from nmp.common.entities.client import EntityClient from nmp.common.service.dependencies import get_entity_client from nmp.intake.config import IntakeConfig -from nmp.intake.spans.api.dependencies import SpansServiceDep, require_workspace_access +from nmp.intake.spans.api.dependencies import DenormalizerDep, SpansServiceDep, require_workspace_access from nmp.intake.spans.domain import TraceBatch from nmp.intake.spans.ingest.atif_domain import ( AtifAgent, @@ -109,6 +109,7 @@ async def ingest_atif( request: Request, service: SpansServiceDep, entity_client: EntityClientDep, + denormalizer: DenormalizerDep, ) -> Response: await validate_evaluation_context( workspace=workspace, @@ -135,4 +136,7 @@ async def ingest_atif( max_subagent_depth=max_subagent_depth, ) await service.ingest_batch(TraceBatch(spans=spans, evaluator_results=evaluator_results)) + context = body.evaluation_context + if denormalizer is not None and context is not None and context.evaluation_id: + denormalizer.mark_dirty(workspace=workspace, evaluation_id=context.evaluation_id) return Response(status_code=status.HTTP_201_CREATED) diff --git a/services/intake/src/nmp/intake/spans/ingest/chat_completions.py b/services/intake/src/nmp/intake/spans/ingest/chat_completions.py index 1929075d90..e461310647 100644 --- a/services/intake/src/nmp/intake/spans/ingest/chat_completions.py +++ b/services/intake/src/nmp/intake/spans/ingest/chat_completions.py @@ -19,7 +19,7 @@ from fastapi import APIRouter, Depends, status from nmp.common.entities.client import EntityClient from nmp.common.service.dependencies import get_entity_client -from nmp.intake.spans.api.dependencies import SpansServiceDep, require_workspace_access +from nmp.intake.spans.api.dependencies import DenormalizerDep, SpansServiceDep, require_workspace_access from nmp.intake.spans.domain import IntakeSpan, SpanKind, SpanStatus, TraceBatch from nmp.intake.spans.ingest.evaluation_context import EvaluationContext from nmp.intake.spans.ingest.evaluation_context_validation import validate_evaluation_context @@ -151,6 +151,7 @@ async def ingest_chat_completion( body: ChatCompletionsIngestRequest, service: SpansServiceDep, entity_client: EntityClientDep, + denormalizer: DenormalizerDep, ) -> ChatCompletionsIngestResponse: await validate_evaluation_context( workspace=workspace, @@ -160,6 +161,9 @@ async def ingest_chat_completion( ingested_at = utc_now() span = _chat_completion_to_span(workspace=workspace, body=body, ingested_at=ingested_at) await service.ingest_batch(TraceBatch(spans=[span])) + context = body.evaluation_context + if denormalizer is not None and context is not None and context.evaluation_id: + denormalizer.mark_dirty(workspace=workspace, evaluation_id=context.evaluation_id) return ChatCompletionsIngestResponse( session_id=span.session_id, span_id=span.external_span_id, diff --git a/services/intake/src/nmp/intake/spans/ingest/otlp.py b/services/intake/src/nmp/intake/spans/ingest/otlp.py index a84f339eb6..9baf83acf5 100644 --- a/services/intake/src/nmp/intake/spans/ingest/otlp.py +++ b/services/intake/src/nmp/intake/spans/ingest/otlp.py @@ -8,13 +8,13 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Request, status from nmp.intake.config import IntakeConfig -from nmp.intake.spans.api.dependencies import SpansServiceDep, require_workspace_access +from nmp.intake.spans.api.dependencies import DenormalizerDep, SpansServiceDep, require_workspace_access from nmp.intake.spans.domain import ( IntakeSpan, SpanStatus, TraceBatch, ) -from nmp.intake.spans.span_attribute_catalog import SpanAttributeField +from nmp.intake.spans.span_attribute_catalog import SpanAttributeField, spec_for_field from nmp.intake.spans.span_semantic_attributes import SpanSemanticAttributes from nmp.intake.spans.storage import json_dumps_preserve, normalize_span_kind, stable_id, utc_now from pydantic import BaseModel, Field @@ -22,6 +22,10 @@ router = APIRouter(dependencies=[Depends(require_workspace_access)]) API_TAG = "Ingest" +# Bag key the evaluation id lands under on a built span (from the attribute catalog). The ingest loop +# reads it back off each span to learn which evaluations a batch touched, for the denormalizer. +_EVALUATION_ID_BAG_KEY = spec_for_field(SpanAttributeField.EVALUATION_ID).bag_key + # Ordered by precedence. Keep direct input.value/output.value first so existing # OpenInference/LangChain payloads continue to win over framework-specific fallbacks. OTLP_INPUT_PAYLOAD_ATTRIBUTE_KEYS = ( @@ -59,6 +63,7 @@ async def ingest_otlp_traces( workspace: str, request: Request, service: SpansServiceDep, + denormalizer: DenormalizerDep, content_type: str = Header(default="application/octet-stream"), content_length: int | None = Header(default=None), ) -> IngestResponse: @@ -79,6 +84,7 @@ async def ingest_otlp_traces( export_request = _parse_export_request(body) ingested_at = utc_now() spans: list[IntakeSpan] = [] + evaluation_ids: set[str] = set() errors: list[str] = [] for resource_spans in export_request.resource_spans: @@ -103,8 +109,16 @@ async def ingest_otlp_traces( errors.append(f"span {span_hex}: {exc}") continue spans.append(span_domain) + evaluation_id = span_domain.attributes_string.get(_EVALUATION_ID_BAG_KEY) + if evaluation_id: + evaluation_ids.add(evaluation_id) await service.ingest_batch(TraceBatch(spans=spans)) + # A single OTLP batch can carry spans for several evaluations (associated via the + # nemo.experiment.id attribute), so refresh the denormalized name facets for every one it touched. + if denormalizer is not None: + for evaluation_id in evaluation_ids: + denormalizer.mark_dirty(workspace=workspace, evaluation_id=evaluation_id) return IngestResponse(errors=errors) diff --git a/services/intake/tests/test_evaluation_denormalizer.py b/services/intake/tests/test_evaluation_denormalizer.py new file mode 100644 index 0000000000..60d62dddc8 --- /dev/null +++ b/services/intake/tests/test_evaluation_denormalizer.py @@ -0,0 +1,186 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the debounced evaluation name-facet refresher.""" + +from __future__ import annotations + +import asyncio +from typing import cast + +import pytest +from nmp.common.entities.client import EntityClient, EntityConflictError, EntityNotFoundError +from nmp.intake.entities.experiments import Experiment +from nmp.intake.experiments.denormalizer import EvaluationDenormalizer +from nmp.intake.repository.evaluation_rollup import EvaluationRollup, EvaluationRollupRepository + + +def _sample_rollup(evaluation_id: str) -> EvaluationRollup: + return EvaluationRollup( + evaluation_id=evaluation_id, + model_names=["provider/model-a", "provider/model-b"], + agent_names=["agent-x"], + agent_versions=["1.0", "1.1"], + ) + + +class _FakeRollupRepo: + def __init__(self, *, error: Exception | None = None) -> None: + self.calls: list[tuple[str, list[str]]] = [] + self._error = error + + async def get_rollups(self, *, workspace: str, evaluation_ids: list[str]) -> dict[str, EvaluationRollup]: + self.calls.append((workspace, list(evaluation_ids))) + if self._error is not None: + raise self._error + return {evaluation_id: _sample_rollup(evaluation_id) for evaluation_id in evaluation_ids} + + +class _FakeEntityClient: + def __init__( + self, + *, + get_error: Exception | None = None, + update_error: Exception | None = None, + existing: Experiment | None = None, + ) -> None: + self.updated: list[Experiment] = [] + self.get_calls: list[tuple[str, str]] = [] + self._get_error = get_error + self._update_error = update_error + self._existing = existing + + async def get(self, entity_type: type[Experiment], *, name: str, workspace: str) -> Experiment: + self.get_calls.append((name, workspace)) + if self._get_error is not None: + raise self._get_error + if self._existing is not None: + return self._existing + return Experiment(name=name, workspace=workspace, experiment_ids=["grp"], dataset_name="ds") + + async def update(self, entity: Experiment) -> Experiment: + if self._update_error is not None: + raise self._update_error + self.updated.append(entity) + return entity + + +def _refresher(repo: _FakeRollupRepo, entity_client: _FakeEntityClient) -> EvaluationDenormalizer: + return EvaluationDenormalizer( + rollup_repository=cast(EvaluationRollupRepository, repo), + entity_client=cast(EntityClient, entity_client), + interval_seconds=999, + ) + + +@pytest.mark.asyncio +async def test_flush_writes_denormalized_facets() -> None: + repo = _FakeRollupRepo() + entity_client = _FakeEntityClient() + refresher = _refresher(repo, entity_client) + + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + refresher.mark_dirty(workspace="default", evaluation_id="eval-b") + await refresher.flush() + + # One batched rollup query for the workspace, covering both dirty evaluations. + assert len(repo.calls) == 1 + workspace, ids = repo.calls[0] + assert workspace == "default" + assert set(ids) == {"eval-a", "eval-b"} + + # Each evaluation got its name facets written, and the dirty set is drained. + assert {entity.name for entity in entity_client.updated} == {"eval-a", "eval-b"} + written = entity_client.updated[0] + assert written.agent_names == ["agent-x"] + assert written.agent_versions == ["1.0", "1.1"] + assert written.model_names == ["provider/model-a", "provider/model-b"] + assert refresher.pending() == set() + + +@pytest.mark.asyncio +async def test_unchanged_facets_skip_write() -> None: + # The stored entity already carries exactly the rollup's names -> no update, no wasted write. + existing = Experiment( + name="eval-a", + workspace="default", + experiment_ids=["grp"], + dataset_name="ds", + agent_names=["agent-x"], + agent_versions=["1.0", "1.1"], + model_names=["provider/model-a", "provider/model-b"], + ) + entity_client = _FakeEntityClient(existing=existing) + refresher = _refresher(_FakeRollupRepo(), entity_client) + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + await refresher.flush() + assert entity_client.updated == [] + assert refresher.pending() == set() + + +def test_mark_dirty_dedupes() -> None: + refresher = _refresher(_FakeRollupRepo(), _FakeEntityClient()) + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + assert refresher.pending() == {("default", "eval-a")} + + +@pytest.mark.asyncio +async def test_flush_noop_when_clean() -> None: + repo = _FakeRollupRepo() + await _refresher(repo, _FakeEntityClient()).flush() + assert repo.calls == [] + + +@pytest.mark.asyncio +async def test_missing_evaluation_is_skipped() -> None: + entity_client = _FakeEntityClient(get_error=EntityNotFoundError("gone")) + refresher = _refresher(_FakeRollupRepo(), entity_client) + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + await refresher.flush() + assert entity_client.updated == [] + assert refresher.pending() == set() # not re-queued; the evaluation no longer exists + + +@pytest.mark.asyncio +async def test_update_conflict_requeues() -> None: + entity_client = _FakeEntityClient(update_error=EntityConflictError("version mismatch")) + refresher = _refresher(_FakeRollupRepo(), entity_client) + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + await refresher.flush() + # A concurrent edit won the optimistic lock; the evaluation is re-queued for the next cycle. + assert refresher.pending() == {("default", "eval-a")} + + +@pytest.mark.asyncio +async def test_stop_flushes_pending_without_loss() -> None: + entity_client = _FakeEntityClient() + refresher = _refresher(_FakeRollupRepo(), entity_client) + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + refresher.start() + # stop() signals the loop to exit and lets it run a final drain — no mid-flush cancellation. + await refresher.stop() + assert {entity.name for entity in entity_client.updated} == {"eval-a"} + assert refresher.pending() == set() + + +@pytest.mark.asyncio +async def test_rollup_query_failure_requeues() -> None: + repo = _FakeRollupRepo(error=RuntimeError("clickhouse down")) + refresher = _refresher(repo, _FakeEntityClient()) + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + await refresher.flush() + assert refresher.pending() == {("default", "eval-a")} + + +@pytest.mark.asyncio +async def test_stop_bounded_drain_does_not_hang_on_persistent_requeue() -> None: + # A persistent optimistic-lock conflict re-queues the evaluation on every flush. stop() must + # terminate (bounded drain) rather than loop forever, leaving the key queued. + entity_client = _FakeEntityClient(update_error=EntityConflictError("version mismatch")) + refresher = _refresher(_FakeRollupRepo(), entity_client) + refresher.mark_dirty(workspace="default", evaluation_id="eval-a") + refresher.start() + # wait_for turns a (regressed) infinite drain into a failure instead of a hung test. + await asyncio.wait_for(refresher.stop(), timeout=5) + assert refresher.pending() == {("default", "eval-a")} diff --git a/services/intake/tests/test_evaluation_denormalizer_self_heal.py b/services/intake/tests/test_evaluation_denormalizer_self_heal.py new file mode 100644 index 0000000000..3be821b954 --- /dev/null +++ b/services/intake/tests/test_evaluation_denormalizer_self_heal.py @@ -0,0 +1,100 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Self-heal of denormalized name facets on read. + +Reading an evaluation compares its live rollup names against the entity's stored facets and queues a +refresh when they differ. This backfills evaluations ingested before the facets existed the first time +they're read, with no per-instance migration to run. +""" + +from __future__ import annotations + +from typing import cast + +from nmp.intake.api.v2.experiments.endpoints import _enqueue_stale_denormalization +from nmp.intake.entities.experiments import Experiment +from nmp.intake.experiments.denormalizer import EvaluationDenormalizer +from nmp.intake.experiments.read_service import EvaluationRead +from nmp.intake.repository.evaluation_rollup import EvaluationRollup + + +class _CapturingRefresher: + def __init__(self) -> None: + self.marked: list[tuple[str, str]] = [] + + def mark_dirty(self, *, workspace: str, evaluation_id: str) -> None: + self.marked.append((workspace, evaluation_id)) + + +def _entity( + name: str, + *, + agent_names: list[str] | None = None, + agent_versions: list[str] | None = None, + model_names: list[str] | None = None, +) -> Experiment: + return Experiment( + name=name, + workspace="default", + experiment_ids=["grp"], + dataset_name="ds", + agent_names=agent_names or [], + agent_versions=agent_versions or [], + model_names=model_names or [], + ) + + +def _rollup( + name: str, *, agent_names: list[str], agent_versions: list[str], model_names: list[str] +) -> EvaluationRollup: + return EvaluationRollup( + evaluation_id=name, + agent_names=agent_names, + agent_versions=agent_versions, + model_names=model_names, + ) + + +def _heal(reads: list[EvaluationRead]) -> list[tuple[str, str]]: + refresher = _CapturingRefresher() + _enqueue_stale_denormalization(cast(EvaluationDenormalizer, refresher), workspace="default", reads=reads) + return refresher.marked + + +def test_enqueues_when_stored_facets_are_empty_but_rollup_has_names() -> None: + read = EvaluationRead( + entity=_entity("eval-a"), + rollup=_rollup("eval-a", agent_names=["agent-x"], agent_versions=["1.0"], model_names=["m"]), + ) + assert _heal([read]) == [("default", "eval-a")] + + +def test_does_not_enqueue_when_facets_match() -> None: + names = dict(agent_names=["agent-x"], agent_versions=["1.0"], model_names=["m"]) + read = EvaluationRead(entity=_entity("eval-a", **names), rollup=_rollup("eval-a", **names)) + assert _heal([read]) == [] + + +def test_enqueues_on_drift_in_any_single_field() -> None: + # Agent names/versions match, but the model set drifted -> still a discrepancy. + read = EvaluationRead( + entity=_entity("eval-a", agent_names=["agent-x"], agent_versions=["1.0"], model_names=["m-old"]), + rollup=_rollup("eval-a", agent_names=["agent-x"], agent_versions=["1.0"], model_names=["m-old", "m-new"]), + ) + assert _heal([read]) == [("default", "eval-a")] + + +def test_skips_when_rollup_missing() -> None: + # No live rollup (e.g. ClickHouse unavailable) -> nothing to compare against, leave facets as-is. + read = EvaluationRead(entity=_entity("eval-a"), rollup=None) + assert _heal([read]) == [] + + +def test_no_refresher_is_a_noop() -> None: + read = EvaluationRead( + entity=_entity("eval-a"), + rollup=_rollup("eval-a", agent_names=["agent-x"], agent_versions=["1.0"], model_names=["m"]), + ) + # Must not raise when the refresher is absent (startup didn't create one). + _enqueue_stale_denormalization(None, workspace="default", reads=[read]) diff --git a/services/intake/tests/test_evaluation_facet_filter.py b/services/intake/tests/test_evaluation_facet_filter.py new file mode 100644 index 0000000000..4e8ec47568 --- /dev/null +++ b/services/intake/tests/test_evaluation_facet_filter.py @@ -0,0 +1,103 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Filtering the workspace Evaluations list by the denormalized agent/model name facets. + +The scalar ``agent_name``/``agent_version``/``model_name`` params match evaluations whose stored +list facet *contains* the value; the endpoint rewrites the parsed equality into a ``$contains`` +membership match against the entity store (mirroring ``experiment_id`` over ``experiment_ids``). +""" + +from __future__ import annotations + +import pytest +from nmp.common.api.filter import ComparisonOperation, FilterOperator, LogicalOperation +from nmp.intake.api.v2.experiments.dependencies import get_evaluation_rollup_repository +from nmp.intake.api.v2.experiments.endpoints import _rewrite_facet_filters +from nmp.intake.entities.experiments import Experiment +from nmp.intake.service import IntakeService +from nmp.testing import ClientContext, create_test_client + +EXPERIMENTS = "/apis/intake/v2/workspaces/default/experiments" +EVALUATIONS = "/apis/intake/v2/workspaces/default/evaluations" + +_FACET_FIELDS = ("data.agent_names", "data.agent_versions", "data.model_names") + + +def _eq(field: str, value: str) -> ComparisonOperation: + return ComparisonOperation(operator=FilterOperator.EQ, field=field, value=value) + + +def test_rewrite_converts_facet_equality_to_contains() -> None: + for field in _FACET_FIELDS: + rewritten = _rewrite_facet_filters(_eq(field, "x")) + assert isinstance(rewritten, ComparisonOperation) + assert rewritten.operator == FilterOperator.CONTAINS + assert rewritten.field == field + assert rewritten.value == "x" + + +def test_rewrite_leaves_non_facet_and_non_eq_untouched() -> None: + # A different field keeps its equality. + assert _rewrite_facet_filters(_eq("data.name", "x")).operator == FilterOperator.EQ + # A non-equality on a facet field is left alone (only the scalar-param equality is a membership match). + already = ComparisonOperation(operator=FilterOperator.CONTAINS, field="data.agent_names", value="x") + assert _rewrite_facet_filters(already).operator == FilterOperator.CONTAINS + + +def test_rewrite_recurses_through_logical_tree() -> None: + op = LogicalOperation( + operator=FilterOperator.AND, + operations=[_eq("data.agent_names", "a"), _eq("data.name", "n")], + ) + rewritten = _rewrite_facet_filters(op) + assert isinstance(rewritten, LogicalOperation) + by_field = {o.field: o.operator for o in rewritten.operations} + assert by_field["data.agent_names"] == FilterOperator.CONTAINS + assert by_field["data.name"] == FilterOperator.EQ + + +def test_rewrite_none_passthrough() -> None: + assert _rewrite_facet_filters(None) is None + + +@pytest.mark.asyncio +async def test_filter_by_agent_name_returns_only_matching_evaluations() -> None: + with create_test_client( + IntakeService, + client_type=ClientContext, + dependency_overrides={get_evaluation_rollup_repository: lambda: None}, + ) as ctx: + tc = ctx.test_client + group = tc.post(EXPERIMENTS, json={"name": "facet-grp"}).json() + for name in ("eval-x", "eval-y"): + created = tc.post( + EVALUATIONS, + json={"name": name, "experiment_group_id": group["id"], "dataset_name": "ds"}, + ) + assert created.status_code == 201, created.text + + # Seed the system-managed facets directly (what the background refresher would write after ingest). + for name, agents, models in ( + ("eval-x", ["agent-x"], ["provider/model-a"]), + ("eval-y", ["agent-y"], ["provider/model-b"]), + ): + entity = await ctx.entity_client.get(Experiment, name=name, workspace="default") + entity.agent_names = agents + entity.model_names = models + await ctx.entity_client.update(entity) + + # Filter by an agent name: only the evaluation whose facet list contains it comes back. + resp = tc.get(EVALUATIONS, params={"filter[agent_name]": "agent-x"}) + assert resp.status_code == 200, resp.text + assert {e["name"] for e in resp.json()["data"]} == {"eval-x"} + + # A model-name filter is independent and scopes to the other evaluation. + resp = tc.get(EVALUATIONS, params={"filter[model_name]": "provider/model-b"}) + assert resp.status_code == 200, resp.text + assert {e["name"] for e in resp.json()["data"]} == {"eval-y"} + + # A name nobody observed matches nothing. + resp = tc.get(EVALUATIONS, params={"filter[agent_name]": "agent-z"}) + assert resp.status_code == 200, resp.text + assert resp.json()["data"] == [] diff --git a/services/intake/tests/test_spans_otlp.py b/services/intake/tests/test_spans_otlp.py index 007914a8a0..fdef1edd1b 100644 --- a/services/intake/tests/test_spans_otlp.py +++ b/services/intake/tests/test_spans_otlp.py @@ -186,6 +186,32 @@ def test_span_to_domain_promotes_pydantic_ai_tool_arguments_and_response(): assert json.loads(domain_span.output) == {"return_value": {"count": 3}} +def test_span_to_domain_carries_evaluation_id_attribute() -> None: + # OTLP associates a span with an evaluation via the nemo.experiment.id attribute; the ingest path + # reads it off the built span to mark that evaluation dirty for facet denormalization. + span = _make_span() + _add_string_attr(span, "nemo.experiment.id", "my-eval") + domain_span = _span_to_domain( + workspace="default", + span=span, + resource_attributes={}, + scope_data={}, + ingested_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + assert domain_span.attributes_string.get("nemo.experiment.id") == "my-eval" + + +def test_span_to_domain_without_evaluation_attribute_has_no_id() -> None: + domain_span = _span_to_domain( + workspace="default", + span=_make_span(), + resource_attributes={}, + scope_data={}, + ingested_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + assert domain_span.attributes_string.get("nemo.experiment.id") is None + + def _make_span( *, trace_id: bytes = DEFAULT_TRACE_ID, diff --git a/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx b/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx index 4c65fdc7f7..969a1c900d 100644 --- a/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx +++ b/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx @@ -99,6 +99,10 @@ const getEvaluationFilterField = (id: string): string | undefined => { if (id === 'latency_ms') return 'latency_ms.mean'; if (id === 'end_to_end_latency_ms') return 'latency_ms.sum'; if (id === 'tokens') return 'tokens.mean'; + // The list columns hold the plural name facets; the API filter params are singular contains-matches. + if (id === 'agent_names') return 'agent_name'; + if (id === 'agent_versions') return 'agent_version'; + if (id === 'model_names') return 'model_name'; const evaluatorMatch = id.match(/^evaluator-(.+)$/); if (evaluatorMatch) return `evaluators.${evaluatorMatch[1]}.mean`; return undefined; @@ -322,12 +326,18 @@ export const ExperimentDataView: FC = ({ group, paretoV id: 'agent_names', header: 'Agent Names', enableSorting: false, + meta: { + filter: { type: 'text', label: 'Agent Name', placeholder: 'Filter by Agent Name' }, + }, cell: ({ getValue }) => {getValue() || '-'}, }), accessor((original) => original.agent_versions?.join(', '), { id: 'agent_versions', header: 'Agent Versions', enableSorting: false, + meta: { + filter: { type: 'text', label: 'Agent Version', placeholder: 'Filter by Agent Version' }, + }, cell: ({ getValue }) => {getValue() || '-'}, }), accessor('dataset_name', { @@ -354,6 +364,9 @@ export const ExperimentDataView: FC = ({ group, paretoV id: 'model_names', header: 'Models', enableSorting: false, + meta: { + filter: { type: 'text', label: 'Model', placeholder: 'Filter by Model' }, + }, cell: ({ getValue }) => {getValue() || '-'}, }), ...metadataKeys.map((key) =>