[release] v0.128.8

api/oss/src/apis/fastapi/queries/exceptions.py

@@ -0,0 +1,21 @@
+from functools import wraps
+
+from fastapi import HTTPException
+
+from oss.src.core.tracing.dtos import FilteringException
+
+
+def handle_filtering_exceptions():
+    """Answer 400 when a saved query holds a filter that cannot be applied."""
+
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            try:
+                return await func(*args, **kwargs)
+            except FilteringException as e:
+                raise HTTPException(status_code=400, detail=str(e)) from e
+
+        return wrapper
+
+    return decorator

api/oss/src/apis/fastapi/queries/router.py

@@ -15,6 +15,7 @@
 )
 from oss.src.core.git.utils import build_retrieval_info
 from oss.src.core.queries.dtos import QueryRevision, QueryRevisionCommit
+from oss.src.apis.fastapi.queries.exceptions import handle_filtering_exceptions
 from oss.src.core.queries.service import (
     QueriesService,
     SimpleQueriesService,
@@ -726,6 +727,7 @@ async def fork_query_variant(
 
     @intercept_exceptions()
     @handle_git_exceptions()
+    @handle_filtering_exceptions()
     async def create_query_revision(
         self,
         request: Request,
@@ -798,6 +800,7 @@ async def fetch_query_revision(
         return query_revision_response
 
     @intercept_exceptions()
+    @handle_filtering_exceptions()
     async def edit_query_revision(
         self,
         request: Request,
@@ -933,6 +936,7 @@ async def query_query_revisions(
         return query_revisions_response
 
     @intercept_exceptions()
+    @handle_filtering_exceptions()
     async def commit_query_revision(
         self,
         request: Request,
@@ -1183,6 +1187,7 @@ def __init__(
     # SIMPLE QUERIES -----------------------------------------------------------
 
     @intercept_exceptions()
+    @handle_filtering_exceptions()
     async def create_simple_query(
         self,
         request: Request,
@@ -1243,6 +1248,7 @@ async def fetch_simple_query(
         return simple_query_response
 
     @intercept_exceptions()
+    @handle_filtering_exceptions()
     async def edit_simple_query(
         self,
         request: Request,

api/oss/src/core/evaluations/runtime/sources.py

@@ -15,14 +15,40 @@
 from oss.src.core.tracing.service import TracingService
 from oss.src.core.shared.dtos import Reference
 from oss.src.core.tracing.dtos import (
+    ComparisonOperator,
+    Condition,
+    Fields,
     Filtering,
+    TraceType,
     Windowing,
     Formatting,
     Format,
     Focus,
     TracingQuery,
     LogicalOperator,
 )
+from oss.src.utils.logging import get_module_logger
+
+log = get_module_logger(__name__)
+
+
+def _only_invocations(filtering: Filtering) -> Filtering:
+    """The live query filter under an outer AND on invocation traces.
+
+    A live run must never read evaluator (annotation) traces back, also when the
+    user's filter has an OR that matches them.
+    """
+    return Filtering(
+        operator=LogicalOperator.AND,
+        conditions=[
+            Condition(
+                field=Fields.TRACE_TYPE.value,
+                operator=ComparisonOperator.IS,
+                value=TraceType.INVOCATION.value,
+            ),
+            filtering,
+        ],
+    )
 
 
 def _extract_root_span(trace: Any) -> Optional[Any]:
@@ -32,11 +58,12 @@ def _extract_root_span(trace: Any) -> Optional[Any]:
     if not isinstance(spans, dict) or not spans:
         return None
 
-    for span in spans.values():
-        if isinstance(span, list):
-            continue
-        if _extract_span_id(span):
-            return span
+    # Same-name sibling roots arrive as a list. Any root carries the link that
+    # types the evaluator trace `annotation`, so a list must not leave it unlinked.
+    for entry in spans.values():
+        for span in entry if isinstance(entry, list) else [entry]:
+            if _extract_span_id(span):
+                return span
 
     return None
 
@@ -97,6 +124,34 @@ def _trace_source_item(
     )
 
 
+def _linked_trace_source_items(
+    *,
+    traces: List[Any],
+    step_key: str,
+) -> List[ResolvedSourceItem]:
+    """Query source items for the traces that have a root span id.
+
+    The root span id becomes the evaluator's `invocation` link, and that link types
+    the evaluator trace `annotation`. A trace without one would give an evaluator
+    trace typed `invocation`, which the next live tick reads back as a new trace.
+    Such a trace also has no outputs to evaluate, so it is skipped.
+    """
+    items: List[ResolvedSourceItem] = []
+    for trace in traces:
+        if not trace or not trace.trace_id:
+            continue
+        item = _trace_source_item(trace=trace, step_key=step_key)
+        if not item.span_id:
+            log.warning(
+                "[EVAL] Skipping a query trace with no root span id.",
+                trace_id=str(item.trace_id),
+                step_key=step_key,
+            )
+            continue
+        items.append(item)
+    return items
+
+
 class SourceResolutionError(Exception):
     """An input step does not carry exactly one recognized source reference."""
 
@@ -487,6 +542,9 @@ async def resolve_live_query_traces(
                 elif query_windowing:
                     windowing.rate = query_windowing.rate
 
+            if not use_windowing:
+                filtering = _only_invocations(filtering)
+
             query_traces[query_step_key] = (
                 await self.tracing_service.query_traces(
                     project_id=project_id,
@@ -552,10 +610,9 @@ async def resolve_query_source_items(
         )
 
         return {
-            query_step_key: [
-                _trace_source_item(trace=trace, step_key=query_step_key)
-                for trace in traces
-                if trace and trace.trace_id
-            ]
+            query_step_key: _linked_trace_source_items(
+                traces=traces,
+                step_key=query_step_key,
+            )
             for query_step_key, traces in query_traces.items()
         }

api/oss/src/core/evaluations/service.py

@@ -82,6 +82,7 @@
 from oss.src.core.tracing.dtos import (
     TracingQuery,
     Filtering,
+    FilteringException,
     Condition,
     ListOperator,
     MetricSpec,
@@ -297,7 +298,7 @@ async def refresh_runs(
                     run=run,
                 ):
                     log.warning(
-                        "[LIVE] Closing invalid live run (null data or non-invocation trace_type).",
+                        "[LIVE] Closing invalid live run (null data, non-invocation trace_type, or a filter that cannot be applied).",
                         project_id=project_id,
                         run_id=run.id,
                     )
@@ -381,7 +382,8 @@ async def _is_live_run_valid(
         project_id: UUID,
         run: EvaluationRun,
     ) -> bool:
-        """Every query step must reference a revision with data targeting invocation traces."""
+        """Every query step must reference a revision with data targeting invocation
+        traces and a filter that the tracing query can apply."""
         if not run.data or not run.data.steps:
             return False
 
@@ -407,6 +409,18 @@ async def _is_live_run_valid(
             if not _is_invocation_query(query_revision.data):
                 return False
 
+            try:
+                QueriesService.validate_query_revision_filtering(query_revision.data)
+            except FilteringException as e:
+                log.warning(
+                    "[LIVE] Live query filter cannot be applied.",
+                    project_id=project_id,
+                    run_id=run.id,
+                    query_revision_id=query_revision_id,
+                    error=str(e),
+                )
+                return False
+
         return True
 
     async def _close_live_run(

api/oss/src/core/queries/service.py

@@ -23,6 +23,7 @@
     Focus,
     Format,
 )
+from oss.src.core.tracing.utils.filtering import validate_filtering
 from oss.src.core.tracing.utils.parsing import (
     parse_trace_id_from_uuid,
     parse_span_id_from_uuid,
@@ -89,7 +90,17 @@ def __init__(
         self.tracing_service = tracing_service
 
     @staticmethod
+    def validate_query_revision_filtering(
+        data: Optional[QueryRevisionData],
+    ) -> None:
+        """Raise `FilteringException` for a filter the tracing query cannot apply,
+        also for an unknown field."""
+        if data is not None:
+            validate_filtering(data.filtering)
+
+    @classmethod
     def _sanitize_persisted_query_revision_data(
+        cls,
         data: Optional[Any],
     ) -> Optional[Dict[str, Any]]:
         """Persist only canonical query revision fields."""
@@ -101,6 +112,8 @@ def _sanitize_persisted_query_revision_data(
         except Exception:
             return None
 
+        cls.validate_query_revision_filtering(parsed)
+
         persisted_data: Dict[str, Any] = {}
         if parsed.formatting is not None:
             persisted_data["formatting"] = parsed.formatting
@@ -1060,6 +1073,8 @@ async def create(
         #
         query_id: Optional[UUID] = None,
     ) -> Optional[SimpleQuery]:
+        QueriesService.validate_query_revision_filtering(simple_query_create.data)
+
         # ----------------------------------------------------------------------
         # Query
         # ----------------------------------------------------------------------
@@ -1305,6 +1320,8 @@ async def edit(
         if str(query_id) != str(simple_query_edit.id):
             return None
 
+        QueriesService.validate_query_revision_filtering(simple_query_edit.data)
+
         # ----------------------------------------------------------------------
         # Query
         # ----------------------------------------------------------------------

api/oss/src/core/tracing/dtos.py

@@ -97,6 +97,7 @@ class Fields(str, Enum):
     SESSION_ID = "session_id"
     USER_ID = "user_id"
     AGENT_ID = "agent_id"
+    HAS_ANNOTATION = "has_annotation"
 
 
 class LogicalOperator(str, Enum):