feat(experiments): filter evaluations by agent name, agent version, or model name

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)]

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,
     *,

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:

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

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