From 0d7eb9cf42f230e1834391767d45f19959cb935d Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Fri, 7 Aug 2026 00:21:23 -0600 Subject: [PATCH 1/8] feat(intake): denormalize agent/model names onto Evaluation entities Add a debounced background worker (EvaluationDenormalizer) that materializes each evaluation's distinct agent_names/agent_versions/model_names from the ClickHouse rollup onto its Evaluation entity, so the list can filter by name against the entity store. Ingest marks (workspace, evaluation_id) dirty; the worker coalesces and drains on an interval, skips no-op writes, and drains gracefully on shutdown. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- services/intake/src/nmp/intake/config.py | 9 + .../src/nmp/intake/entities/experiments.py | 20 ++ .../nmp/intake/experiments/denormalizer.py | 131 +++++++++++++ services/intake/src/nmp/intake/service.py | 21 +++ .../src/nmp/intake/spans/api/dependencies.py | 10 + .../src/nmp/intake/spans/ingest/atif.py | 6 +- .../intake/spans/ingest/chat_completions.py | 6 +- .../tests/test_evaluation_denormalizer.py | 172 ++++++++++++++++++ 8 files changed, 373 insertions(+), 2 deletions(-) create mode 100644 services/intake/src/nmp/intake/experiments/denormalizer.py create mode 100644 services/intake/tests/test_evaluation_denormalizer.py diff --git a/services/intake/src/nmp/intake/config.py b/services/intake/src/nmp/intake/config.py index 0b9967de1c..447e6f9bb7 100644 --- a/services/intake/src/nmp/intake/config.py +++ b/services/intake/src/nmp/intake/config.py @@ -61,3 +61,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=10.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..bff0c752ee --- /dev/null +++ b/services/intake/src/nmp/intake/experiments/denormalizer.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Background worker that denormalizes name fields from ClickHouse onto Evaluation entities. + +Ingest marks ``(workspace, evaluation_id)`` dirty — a cheap, non-blocking set add. A background loop +drains the dirty set on a fixed interval, 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 workspace-wide Evaluations list filter by agent/model name against +the entity store (``$contains``) instead of scanning the ClickHouse session table on every request. + +This is a deliberately narrowed descendant of the closed PR #424 rollup refresher: it denormalizes +only the name fields, not the computed metric rollups. Names are raw observed strings with no formula, +so a change to how any aggregate is computed never invalidates them — the staleness/backfill hazard +that made denormalizing metrics not worth it (ASE-319) does not apply here. Bursts coalesce: many +ingests for one evaluation within an interval collapse to a single recompute, and ingest latency is +never gated on the rollup query. +""" + +from __future__ import annotations + +import asyncio +import logging + +from nmp.common.entities.client import EntityClient, EntityConflictError, EntityNotFoundError +from nmp.intake.entities.experiments import Experiment +from nmp.intake.repository.evaluation_rollup import EvaluationRollup, EvaluationRollupRepository + +logger = logging.getLogger(__name__) + + +class EvaluationDenormalizer: + """Coalesces dirty evaluation ids and refreshes their denormalized name fields on a fixed cadence.""" + + def __init__( + self, + *, + rollup_repository: EvaluationRollupRepository, + entity_client: EntityClient, + interval_seconds: float = 10.0, + ) -> None: + self._rollup_repository = rollup_repository + self._entity_client = entity_client + self._interval_seconds = interval_seconds + self._dirty: set[tuple[str, str]] = set() + self._task: asyncio.Task[None] | None = None + self._stopping = asyncio.Event() + + 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) + + 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 let it finish any in-flight flush — we never cancel mid-flush, so a + # detached batch can't be dropped before it's written. Then a final drain covers the cases the loop + # can't (it saw the stop flag before its first flush, or items were enqueued during the last flush). + self._stopping.set() + if self._task is not None: + await self._task + self._task = None + await self.flush() + + 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 flush(self) -> None: + """Drain the dirty set and write current name fields. Directly callable for deterministic 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._dirty.add((workspace, 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._dirty.add((workspace, evaluation_id)) diff --git a/services/intake/src/nmp/intake/service.py b/services/intake/src/nmp/intake/service.py index 3b650d9ac2..b30b8a402b 100644 --- a/services/intake/src/nmp/intake/service.py +++ b/services/intake/src/nmp/intake/service.py @@ -9,6 +9,9 @@ from nmp.common.service import RouterConfig, Service from nmp.intake.api.v2.experiments import endpoints as experiments from nmp.intake.config import IntakeConfig +from nmp.intake.experiments.denormalizer import EvaluationDenormalizer +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 @@ -26,6 +29,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._ready = False @property @@ -82,12 +87,28 @@ async def on_startup(self) -> None: "clickhouse_database": cfg.clickhouse_config.database, }, ) + # 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 service-owned ClickHouse client.""" self._ready = False + # Stop the refresher first: its final flush still needs the ClickHouse client below. + if self.denormalizer is not None: + await self.denormalizer.stop() + self.denormalizer = None if self.clickhouse_client is not None: await self.clickhouse_client.close() self.clickhouse_client = None 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/tests/test_evaluation_denormalizer.py b/services/intake/tests/test_evaluation_denormalizer.py new file mode 100644 index 0000000000..30f21649f7 --- /dev/null +++ b/services/intake/tests/test_evaluation_denormalizer.py @@ -0,0 +1,172 @@ +# 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 + +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")} From 89c7a49f678835b75acd78648bd6e882956fa493 Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Fri, 7 Aug 2026 00:21:23 -0600 Subject: [PATCH 2/8] feat(intake): filter evaluations by agent name, agent version, or model name Expose agent_name/agent_version/model_name filters on the evaluations list, rewriting each scalar equality into a $contains match over the denormalized list fields (mirroring experiment_id membership). Also self-heal on read: when a read's live rollup names differ from the stored fields, enqueue the evaluation for the denormalizer, backfilling pre-existing data as it is viewed with no migration. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- .../intake/api/v2/experiments/dependencies.py | 10 ++ .../intake/api/v2/experiments/endpoints.py | 71 +++++++++++- .../nmp/intake/api/v2/experiments/schemas.py | 17 +++ .../test_evaluation_denormalizer_self_heal.py | 100 +++++++++++++++++ .../tests/test_evaluation_facet_filter.py | 103 ++++++++++++++++++ 5 files changed, 300 insertions(+), 1 deletion(-) create mode 100644 services/intake/tests/test_evaluation_denormalizer_self_heal.py create mode 100644 services/intake/tests/test_evaluation_facet_filter.py 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/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"] == [] From 9de2546cce65d01ee429ce96b43864866457ed3b Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Fri, 7 Aug 2026 00:21:24 -0600 Subject: [PATCH 3/8] chore(sdk): regenerate OpenAPI and Python SDK for evaluation name filters Regenerated from the new EvaluationFilter fields via make refresh-openapi + make stainless. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- openapi/ga/individual/platform.openapi.yaml | 15 +++++++++++++++ openapi/ga/openapi.yaml | 15 +++++++++++++++ openapi/openapi.yaml | 15 +++++++++++++++ sdk/python/nemo-platform/.nmpcontext/openapi.yaml | 15 +++++++++++++++ .../types/evaluations/evaluation_filter_param.py | 9 +++++++++ .../tests/api_resources/test_evaluations.py | 6 ++++++ 6 files changed, 75 insertions(+) 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, From d9892649783e738f76c65e1de60f68284b3b99b8 Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Fri, 7 Aug 2026 00:21:24 -0600 Subject: [PATCH 4/8] feat(studio): add agent/model name filters to the evaluations list Add free-text agent name, agent version, and model filter inputs to the evaluations list, mapped to the filter[agent_name]/[agent_version]/[model_name] API params. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- .../dataViews/ExperimentDataView/index.tsx | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx b/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx index 58425a9d91..587364d680 100644 --- a/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx +++ b/web/packages/studio/src/components/dataViews/ExperimentDataView/index.tsx @@ -96,6 +96,10 @@ const getEvaluationFilterField = (id: string): string | undefined => { if (id === 'cost_usd') return 'cost_usd.mean'; if (id === 'latency_ms') return 'latency_ms.mean'; 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; @@ -319,12 +323,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', { @@ -351,6 +361,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) => From e084ee3f1574ffd5a62b3b9c4234f0decf844b6b Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:38:50 -0600 Subject: [PATCH 5/8] Use evaluation denormalizer in otlp ingest Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- .../src/nmp/intake/spans/ingest/otlp.py | 19 ++++++++-- services/intake/tests/test_spans_otlp.py | 38 ++++++++++++++++--- 2 files changed, 47 insertions(+), 10 deletions(-) diff --git a/services/intake/src/nmp/intake/spans/ingest/otlp.py b/services/intake/src/nmp/intake/spans/ingest/otlp.py index a84f339eb6..bd18d50130 100644 --- a/services/intake/src/nmp/intake/spans/ingest/otlp.py +++ b/services/intake/src/nmp/intake/spans/ingest/otlp.py @@ -8,7 +8,7 @@ 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, @@ -59,6 +59,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 +80,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: @@ -91,7 +93,7 @@ async def ingest_otlp_traces( } for span in scope_spans.spans: try: - span_domain = _span_to_domain( + span_domain, evaluation_id = _span_to_domain( workspace=workspace, span=span, resource_attributes=resource_attributes, @@ -103,8 +105,15 @@ async def ingest_otlp_traces( errors.append(f"span {span_hex}: {exc}") continue spans.append(span_domain) + 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) @@ -150,7 +159,9 @@ def _span_to_domain( resource_attributes: dict[str, Any], scope_data: dict[str, Any], ingested_at: datetime, -) -> IntakeSpan: +) -> tuple[IntakeSpan, str | None]: + """Build the domain span and surface the evaluation id it's associated with (if any), so the + ingest path can mark that evaluation dirty for facet denormalization.""" trace_id = _required_otlp_id(span.trace_id, field_name="trace_id") trace_id_hex = trace_id.hex() source_span_id = _required_otlp_id(span.span_id, field_name="span_id") @@ -206,7 +217,7 @@ def _span_to_domain( output=_output_payload(attributes, events, kind=kind.value) or "", event_ts=ingested_at, ) - return span_domain + return span_domain, semantic_attributes.evaluation_id def _required_otlp_id(value: Any, *, field_name: str) -> bytes: diff --git a/services/intake/tests/test_spans_otlp.py b/services/intake/tests/test_spans_otlp.py index 007914a8a0..068682d77e 100644 --- a/services/intake/tests/test_spans_otlp.py +++ b/services/intake/tests/test_spans_otlp.py @@ -72,7 +72,7 @@ def test_span_to_domain_filters_resource_raw_attributes(): "user.id": "user-a", } - domain_span = _span_to_domain( + domain_span, _ = _span_to_domain( workspace="default", span=span, resource_attributes=raw_resource_attributes, @@ -88,7 +88,7 @@ def test_span_to_domain_filters_resource_raw_attributes(): def test_span_to_domain_skips_empty_scope_data(): span = _make_span() - domain_span = _span_to_domain( + domain_span, _ = _span_to_domain( workspace="default", span=span, resource_attributes={}, @@ -108,7 +108,7 @@ def test_span_to_domain_does_not_duplicate_model_aliases(): response_model.key = "gen_ai.response.model" response_model.value.string_value = "response-model" - domain_span = _span_to_domain( + domain_span, _ = _span_to_domain( workspace="default", span=span, resource_attributes={}, @@ -130,7 +130,7 @@ def test_span_to_domain_promotes_pydantic_ai_model_messages(): span, "gen_ai.system_instructions", json.dumps([{"type": "text", "content": "You are an analyst."}]) ) - domain_span = _span_to_domain( + domain_span, _ = _span_to_domain( workspace="default", span=span, resource_attributes={"gen_ai.agent.name": "nemo-optimizer-analyst"}, @@ -155,7 +155,7 @@ def test_span_to_domain_promotes_pydantic_ai_agent_run_result(): _add_string_attr(span, "pydantic_ai.all_messages", json.dumps(all_messages)) _add_string_attr(span, "final_result", json.dumps(final_result)) - domain_span = _span_to_domain( + domain_span, _ = _span_to_domain( workspace="default", span=span, resource_attributes={"gen_ai.agent.name": "nemo-optimizer-analyst"}, @@ -174,7 +174,7 @@ def test_span_to_domain_promotes_pydantic_ai_tool_arguments_and_response(): _add_string_attr(span, "gen_ai.tool.call.arguments", '{"code":"await fetch_spans()"}') _add_string_attr(span, "tool_response", '{"return_value":{"count":3}}') - domain_span = _span_to_domain( + domain_span, _ = _span_to_domain( workspace="default", span=span, resource_attributes={"gen_ai.agent.name": "nemo-optimizer-analyst"}, @@ -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_surfaces_evaluation_id() -> None: + # OTLP associates a span with an evaluation via the nemo.experiment.id attribute; _span_to_domain + # surfaces it so the ingest path can mark that evaluation dirty for facet denormalization. + span = _make_span() + _add_string_attr(span, "nemo.experiment.id", "my-eval") + _, evaluation_id = _span_to_domain( + workspace="default", + span=span, + resource_attributes={}, + scope_data={}, + ingested_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + assert evaluation_id == "my-eval" + + +def test_span_to_domain_without_evaluation_attribute_surfaces_no_id() -> None: + _, evaluation_id = _span_to_domain( + workspace="default", + span=_make_span(), + resource_attributes={}, + scope_data={}, + ingested_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + assert evaluation_id is None + + def _make_span( *, trace_id: bytes = DEFAULT_TRACE_ID, From d15c4c0c5d363566d61e03912e032cee1f0e248e Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:16:13 -0600 Subject: [PATCH 6/8] separate generic debounced refresher Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- .../src/nmp/intake/debounced_refresher.py | 86 +++++++++++++++++++ .../nmp/intake/experiments/denormalizer.py | 85 ++++++------------ 2 files changed, 110 insertions(+), 61 deletions(-) create mode 100644 services/intake/src/nmp/intake/debounced_refresher.py diff --git a/services/intake/src/nmp/intake/debounced_refresher.py b/services/intake/src/nmp/intake/debounced_refresher.py new file mode 100644 index 0000000000..3bf00bb8b5 --- /dev/null +++ b/services/intake/src/nmp/intake/debounced_refresher.py @@ -0,0 +1,86 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Generic debounced, coalescing background refresher. + +Owns *how* a background refresh runs — a coalescing dirty set of keys, a periodic drain loop with an +interruptible sleep, and a managed start/stop lifecycle with a graceful final drain — independent of +*what* the work is. Subclasses implement :meth:`_process` with the domain logic and re-queue failures +via :meth:`_enqueue`. + +This has no intake-specific dependencies; it lives here for now but is a candidate to move to +``nmp.common`` once a second consumer needs the same pattern. +""" + +from __future__ import annotations + +import asyncio +import logging +from abc import ABC, abstractmethod +from typing import Generic, TypeVar + +logger = logging.getLogger(__name__) + +KeyT = TypeVar("KeyT") + + +class DebouncedRefresher(ABC, Generic[KeyT]): + """Coalesces dirty keys and processes them in batches on a fixed cadence. + + A burst of :meth:`_enqueue` calls for the same key within an interval collapses to a single batch + entry, and enqueuing never blocks (it's a plain set add), so the hot path (e.g. request handling) + is never gated on the work. Subclasses implement :meth:`_process`. + """ + + def __init__(self, *, interval_seconds: float = 10.0) -> None: + self._interval_seconds = interval_seconds + self._dirty: set[KeyT] = set() + self._task: asyncio.Task[None] | None = None + self._stopping = asyncio.Event() + + def _enqueue(self, key: KeyT) -> None: + """Queue a key for the next batch. Cheap and non-blocking; safe to call from any async context.""" + self._dirty.add(key) + + def pending(self) -> set[KeyT]: + """Return a copy of the currently-queued keys (for observability/tests).""" + return set(self._dirty) + + 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 let it finish any in-flight flush — we never cancel mid-flush, so a + # detached batch can't be dropped before it's written. Then a final drain covers the cases the loop + # can't (it saw the stop flag before its first flush, or items were enqueued during the last flush). + self._stopping.set() + if self._task is not None: + await self._task + self._task = None + await self.flush() + + 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("%s refresh cycle failed", type(self).__name__) + + async def flush(self) -> None: + """Drain the dirty set and hand the batch to :meth:`_process`. Directly callable for tests.""" + if not self._dirty: + return + batch = self._dirty + self._dirty = set() + await self._process(batch) + + @abstractmethod + async def _process(self, batch: set[KeyT]) -> None: + """Process a drained batch of dirty keys. Re-queue any that should retry via :meth:`_enqueue`.""" diff --git a/services/intake/src/nmp/intake/experiments/denormalizer.py b/services/intake/src/nmp/intake/experiments/denormalizer.py index bff0c752ee..4036b05392 100644 --- a/services/intake/src/nmp/intake/experiments/denormalizer.py +++ b/services/intake/src/nmp/intake/experiments/denormalizer.py @@ -1,36 +1,38 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Background worker that denormalizes name fields from ClickHouse onto Evaluation entities. - -Ingest marks ``(workspace, evaluation_id)`` dirty — a cheap, non-blocking set add. A background loop -drains the dirty set on a fixed interval, 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 workspace-wide Evaluations list filter by agent/model name against -the entity store (``$contains``) instead of scanning the ClickHouse session table on every request. - -This is a deliberately narrowed descendant of the closed PR #424 rollup refresher: it denormalizes -only the name fields, not the computed metric rollups. Names are raw observed strings with no formula, -so a change to how any aggregate is computed never invalidates them — the staleness/backfill hazard -that made denormalizing metrics not worth it (ASE-319) does not apply here. Bursts coalesce: many -ingests for one evaluation within an interval collapse to a single recompute, and ingest latency is -never gated on the rollup query. +"""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 workspace-wide Evaluations list +filter by agent/model name against the entity store (``$contains``) instead of scanning the ClickHouse +session table on every request. + +This is a deliberately narrowed descendant of the closed PR #424 rollup refresher: it denormalizes only +the name fields, not the computed metric rollups. Names are raw observed strings with no formula, so a +change to how any aggregate is computed never invalidates them — the staleness/backfill hazard that made +denormalizing metrics not worth it (ASE-319) does not apply here. + +The *how it runs* (the debounced coalescing loop + start/stop lifecycle) lives in the generic +:class:`nmp.intake.debounced_refresher.DebouncedRefresher`; this class carries only the evaluation +business logic — compute the rollup and write the name fields. """ from __future__ import annotations -import asyncio import logging from nmp.common.entities.client import EntityClient, EntityConflictError, EntityNotFoundError +from nmp.intake.debounced_refresher import DebouncedRefresher from nmp.intake.entities.experiments import Experiment from nmp.intake.repository.evaluation_rollup import EvaluationRollup, EvaluationRollupRepository logger = logging.getLogger(__name__) -class EvaluationDenormalizer: - """Coalesces dirty evaluation ids and refreshes their denormalized name fields on a fixed cadence.""" +class EvaluationDenormalizer(DebouncedRefresher[tuple[str, str]]): + """Refreshes each dirty evaluation's denormalized name fields from its ClickHouse rollup.""" def __init__( self, @@ -39,54 +41,15 @@ def __init__( entity_client: EntityClient, interval_seconds: float = 10.0, ) -> None: + super().__init__(interval_seconds=interval_seconds) self._rollup_repository = rollup_repository self._entity_client = entity_client - self._interval_seconds = interval_seconds - self._dirty: set[tuple[str, str]] = set() - self._task: asyncio.Task[None] | None = None - self._stopping = asyncio.Event() 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) - - 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 let it finish any in-flight flush — we never cancel mid-flush, so a - # detached batch can't be dropped before it's written. Then a final drain covers the cases the loop - # can't (it saw the stop flag before its first flush, or items were enqueued during the last flush). - self._stopping.set() - if self._task is not None: - await self._task - self._task = None - await self.flush() + self._enqueue((workspace, evaluation_id)) - 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 flush(self) -> None: - """Drain the dirty set and write current name fields. Directly callable for deterministic tests.""" - if not self._dirty: - return - batch = self._dirty - self._dirty = set() + async def _process(self, batch: set[tuple[str, str]]) -> None: by_workspace: dict[str, list[str]] = {} for workspace, evaluation_id in batch: by_workspace.setdefault(workspace, []).append(evaluation_id) @@ -97,7 +60,7 @@ async def flush(self) -> None: # 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._dirty.add((workspace, evaluation_id)) + 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) @@ -128,4 +91,4 @@ async def _write_names(self, workspace: str, evaluation_id: str, rollup: Evaluat await self._entity_client.update(evaluation) except EntityConflictError: # A concurrent user edit won the optimistic lock; re-queue for the next cycle. - self._dirty.add((workspace, evaluation_id)) + self.mark_dirty(workspace=workspace, evaluation_id=evaluation_id) From c0a2c73cecda4b42991985b9204c77641d0f048b Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Mon, 10 Aug 2026 09:53:42 -0600 Subject: [PATCH 7/8] coderabbit Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- .../src/nmp/intake/debounced_refresher.py | 20 ++++++++++++++++--- .../tests/test_evaluation_denormalizer.py | 14 +++++++++++++ 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/services/intake/src/nmp/intake/debounced_refresher.py b/services/intake/src/nmp/intake/debounced_refresher.py index 3bf00bb8b5..1fdb0d52c6 100644 --- a/services/intake/src/nmp/intake/debounced_refresher.py +++ b/services/intake/src/nmp/intake/debounced_refresher.py @@ -23,6 +23,10 @@ KeyT = TypeVar("KeyT") +# stop()'s final flush can itself re-queue keys (a transient failure/conflict in _process that would +# clear on retry), so it drains up to this many passes; keys still queued afterwards are logged. +_STOP_DRAIN_PASSES = 3 + class DebouncedRefresher(ABC, Generic[KeyT]): """Coalesces dirty keys and processes them in batches on a fixed cadence. @@ -53,13 +57,23 @@ def start(self) -> None: async def stop(self) -> None: # Signal the loop to exit and let it finish any in-flight flush — we never cancel mid-flush, so a - # detached batch can't be dropped before it's written. Then a final drain covers the cases the loop - # can't (it saw the stop flag before its first flush, or items were enqueued during the last flush). + # detached batch can't be dropped before it's written. Then drain what's left in a bounded number + # of passes, since the final flush can re-queue keys itself. self._stopping.set() if self._task is not None: await self._task self._task = None - await self.flush() + for _ in range(_STOP_DRAIN_PASSES): + if not self._dirty: + return + await self.flush() + if self._dirty: + logger.warning( + "%s stopped with %d key(s) still queued after %d drain passes; not retried before shutdown", + type(self).__name__, + len(self._dirty), + _STOP_DRAIN_PASSES, + ) async def _run(self) -> None: while not self._stopping.is_set(): diff --git a/services/intake/tests/test_evaluation_denormalizer.py b/services/intake/tests/test_evaluation_denormalizer.py index 30f21649f7..60d62dddc8 100644 --- a/services/intake/tests/test_evaluation_denormalizer.py +++ b/services/intake/tests/test_evaluation_denormalizer.py @@ -5,6 +5,7 @@ from __future__ import annotations +import asyncio from typing import cast import pytest @@ -170,3 +171,16 @@ async def test_rollup_query_failure_requeues() -> None: 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")} From 92233de2c563751d5647598ebea4a9330f97f799 Mon Sep 17 00:00:00 2001 From: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> Date: Mon, 10 Aug 2026 16:26:00 -0600 Subject: [PATCH 8/8] PR comments Signed-off-by: shanaiabuggy <59746633+shanaiabuggy@users.noreply.github.com> --- .../src/nmp/intake/background_worker.py | 41 +++++++ services/intake/src/nmp/intake/config.py | 2 +- .../src/nmp/intake/debounced_refresher.py | 100 ------------------ .../nmp/intake/experiments/denormalizer.py | 86 +++++++++++---- .../src/nmp/intake/spans/ingest/otlp.py | 15 +-- services/intake/tests/test_spans_otlp.py | 28 ++--- 6 files changed, 132 insertions(+), 140 deletions(-) create mode 100644 services/intake/src/nmp/intake/background_worker.py delete mode 100644 services/intake/src/nmp/intake/debounced_refresher.py 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 38d3748393..08ad16a69f 100644 --- a/services/intake/src/nmp/intake/config.py +++ b/services/intake/src/nmp/intake/config.py @@ -87,7 +87,7 @@ class IntakeConfig(_BaseIntakeConfig): description="Maximum number of trajectory levels accepted for recursive ATIF subagents.", ) denormalization_interval_seconds: float = Field( - default=10.0, + default=60.0, gt=0, description=( "How often the background worker drains the dirty set and denormalizes the distinct " diff --git a/services/intake/src/nmp/intake/debounced_refresher.py b/services/intake/src/nmp/intake/debounced_refresher.py deleted file mode 100644 index 1fdb0d52c6..0000000000 --- a/services/intake/src/nmp/intake/debounced_refresher.py +++ /dev/null @@ -1,100 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -"""Generic debounced, coalescing background refresher. - -Owns *how* a background refresh runs — a coalescing dirty set of keys, a periodic drain loop with an -interruptible sleep, and a managed start/stop lifecycle with a graceful final drain — independent of -*what* the work is. Subclasses implement :meth:`_process` with the domain logic and re-queue failures -via :meth:`_enqueue`. - -This has no intake-specific dependencies; it lives here for now but is a candidate to move to -``nmp.common`` once a second consumer needs the same pattern. -""" - -from __future__ import annotations - -import asyncio -import logging -from abc import ABC, abstractmethod -from typing import Generic, TypeVar - -logger = logging.getLogger(__name__) - -KeyT = TypeVar("KeyT") - -# stop()'s final flush can itself re-queue keys (a transient failure/conflict in _process that would -# clear on retry), so it drains up to this many passes; keys still queued afterwards are logged. -_STOP_DRAIN_PASSES = 3 - - -class DebouncedRefresher(ABC, Generic[KeyT]): - """Coalesces dirty keys and processes them in batches on a fixed cadence. - - A burst of :meth:`_enqueue` calls for the same key within an interval collapses to a single batch - entry, and enqueuing never blocks (it's a plain set add), so the hot path (e.g. request handling) - is never gated on the work. Subclasses implement :meth:`_process`. - """ - - def __init__(self, *, interval_seconds: float = 10.0) -> None: - self._interval_seconds = interval_seconds - self._dirty: set[KeyT] = set() - self._task: asyncio.Task[None] | None = None - self._stopping = asyncio.Event() - - def _enqueue(self, key: KeyT) -> None: - """Queue a key for the next batch. Cheap and non-blocking; safe to call from any async context.""" - self._dirty.add(key) - - def pending(self) -> set[KeyT]: - """Return a copy of the currently-queued keys (for observability/tests).""" - return set(self._dirty) - - 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 let it finish any in-flight flush — we never cancel mid-flush, so a - # detached batch can't be dropped before it's written. Then drain what's left in a bounded number - # of passes, since the final flush can re-queue keys itself. - self._stopping.set() - if self._task is not None: - await self._task - self._task = None - for _ in range(_STOP_DRAIN_PASSES): - if not self._dirty: - return - await self.flush() - if self._dirty: - logger.warning( - "%s stopped with %d key(s) still queued after %d drain passes; not retried before shutdown", - type(self).__name__, - len(self._dirty), - _STOP_DRAIN_PASSES, - ) - - 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("%s refresh cycle failed", type(self).__name__) - - async def flush(self) -> None: - """Drain the dirty set and hand the batch to :meth:`_process`. Directly callable for tests.""" - if not self._dirty: - return - batch = self._dirty - self._dirty = set() - await self._process(batch) - - @abstractmethod - async def _process(self, batch: set[KeyT]) -> None: - """Process a drained batch of dirty keys. Re-queue any that should retry via :meth:`_enqueue`.""" diff --git a/services/intake/src/nmp/intake/experiments/denormalizer.py b/services/intake/src/nmp/intake/experiments/denormalizer.py index 4036b05392..88284f714d 100644 --- a/services/intake/src/nmp/intake/experiments/denormalizer.py +++ b/services/intake/src/nmp/intake/experiments/denormalizer.py @@ -5,51 +5,99 @@ 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 workspace-wide Evaluations list -filter by agent/model name against the entity store (``$contains``) instead of scanning the ClickHouse -session table on every request. - -This is a deliberately narrowed descendant of the closed PR #424 rollup refresher: it denormalizes only -the name fields, not the computed metric rollups. Names are raw observed strings with no formula, so a -change to how any aggregate is computed never invalidates them — the staleness/backfill hazard that made -denormalizing metrics not worth it (ASE-319) does not apply here. - -The *how it runs* (the debounced coalescing loop + start/stop lifecycle) lives in the generic -:class:`nmp.intake.debounced_refresher.DebouncedRefresher`; this class carries only the evaluation -business logic — compute the rollup and write the name fields. +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.debounced_refresher import DebouncedRefresher +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(DebouncedRefresher[tuple[str, str]]): - """Refreshes each dirty evaluation's denormalized name fields from its ClickHouse rollup.""" +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 = 10.0, + interval_seconds: float = 60.0, ) -> None: - super().__init__(interval_seconds=interval_seconds) + 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._enqueue((workspace, evaluation_id)) + 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 _process(self, batch: set[tuple[str, str]]) -> None: + 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) diff --git a/services/intake/src/nmp/intake/spans/ingest/otlp.py b/services/intake/src/nmp/intake/spans/ingest/otlp.py index bd18d50130..9baf83acf5 100644 --- a/services/intake/src/nmp/intake/spans/ingest/otlp.py +++ b/services/intake/src/nmp/intake/spans/ingest/otlp.py @@ -14,7 +14,7 @@ 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 = ( @@ -93,7 +97,7 @@ async def ingest_otlp_traces( } for span in scope_spans.spans: try: - span_domain, evaluation_id = _span_to_domain( + span_domain = _span_to_domain( workspace=workspace, span=span, resource_attributes=resource_attributes, @@ -105,6 +109,7 @@ 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) @@ -159,9 +164,7 @@ def _span_to_domain( resource_attributes: dict[str, Any], scope_data: dict[str, Any], ingested_at: datetime, -) -> tuple[IntakeSpan, str | None]: - """Build the domain span and surface the evaluation id it's associated with (if any), so the - ingest path can mark that evaluation dirty for facet denormalization.""" +) -> IntakeSpan: trace_id = _required_otlp_id(span.trace_id, field_name="trace_id") trace_id_hex = trace_id.hex() source_span_id = _required_otlp_id(span.span_id, field_name="span_id") @@ -217,7 +220,7 @@ def _span_to_domain( output=_output_payload(attributes, events, kind=kind.value) or "", event_ts=ingested_at, ) - return span_domain, semantic_attributes.evaluation_id + return span_domain def _required_otlp_id(value: Any, *, field_name: str) -> bytes: diff --git a/services/intake/tests/test_spans_otlp.py b/services/intake/tests/test_spans_otlp.py index 068682d77e..fdef1edd1b 100644 --- a/services/intake/tests/test_spans_otlp.py +++ b/services/intake/tests/test_spans_otlp.py @@ -72,7 +72,7 @@ def test_span_to_domain_filters_resource_raw_attributes(): "user.id": "user-a", } - domain_span, _ = _span_to_domain( + domain_span = _span_to_domain( workspace="default", span=span, resource_attributes=raw_resource_attributes, @@ -88,7 +88,7 @@ def test_span_to_domain_filters_resource_raw_attributes(): def test_span_to_domain_skips_empty_scope_data(): span = _make_span() - domain_span, _ = _span_to_domain( + domain_span = _span_to_domain( workspace="default", span=span, resource_attributes={}, @@ -108,7 +108,7 @@ def test_span_to_domain_does_not_duplicate_model_aliases(): response_model.key = "gen_ai.response.model" response_model.value.string_value = "response-model" - domain_span, _ = _span_to_domain( + domain_span = _span_to_domain( workspace="default", span=span, resource_attributes={}, @@ -130,7 +130,7 @@ def test_span_to_domain_promotes_pydantic_ai_model_messages(): span, "gen_ai.system_instructions", json.dumps([{"type": "text", "content": "You are an analyst."}]) ) - domain_span, _ = _span_to_domain( + domain_span = _span_to_domain( workspace="default", span=span, resource_attributes={"gen_ai.agent.name": "nemo-optimizer-analyst"}, @@ -155,7 +155,7 @@ def test_span_to_domain_promotes_pydantic_ai_agent_run_result(): _add_string_attr(span, "pydantic_ai.all_messages", json.dumps(all_messages)) _add_string_attr(span, "final_result", json.dumps(final_result)) - domain_span, _ = _span_to_domain( + domain_span = _span_to_domain( workspace="default", span=span, resource_attributes={"gen_ai.agent.name": "nemo-optimizer-analyst"}, @@ -174,7 +174,7 @@ def test_span_to_domain_promotes_pydantic_ai_tool_arguments_and_response(): _add_string_attr(span, "gen_ai.tool.call.arguments", '{"code":"await fetch_spans()"}') _add_string_attr(span, "tool_response", '{"return_value":{"count":3}}') - domain_span, _ = _span_to_domain( + domain_span = _span_to_domain( workspace="default", span=span, resource_attributes={"gen_ai.agent.name": "nemo-optimizer-analyst"}, @@ -186,30 +186,30 @@ 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_surfaces_evaluation_id() -> None: - # OTLP associates a span with an evaluation via the nemo.experiment.id attribute; _span_to_domain - # surfaces it so the ingest path can mark that evaluation dirty for facet denormalization. +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") - _, evaluation_id = _span_to_domain( + domain_span = _span_to_domain( workspace="default", span=span, resource_attributes={}, scope_data={}, ingested_at=datetime(2026, 1, 1, tzinfo=timezone.utc), ) - assert evaluation_id == "my-eval" + assert domain_span.attributes_string.get("nemo.experiment.id") == "my-eval" -def test_span_to_domain_without_evaluation_attribute_surfaces_no_id() -> None: - _, evaluation_id = _span_to_domain( +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 evaluation_id is None + assert domain_span.attributes_string.get("nemo.experiment.id") is None def _make_span(