diff --git a/openapi/ga/individual/platform.openapi.yaml b/openapi/ga/individual/platform.openapi.yaml index efa7c4b301..8135fb7c79 100644 --- a/openapi/ga/individual/platform.openapi.yaml +++ b/openapi/ga/individual/platform.openapi.yaml @@ -6299,6 +6299,71 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions: + get: + tags: + - Experiments + summary: List Experiment Sessions + operationId: list_experiment_sessions_apis_intake_v2_workspaces__workspace__experiments__name__sessions_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + - name: page + in: query + required: false + schema: + type: integer + minimum: 1 + description: Page number. + default: 1 + title: Page + description: Page number. + - name: page_size + in: query + required: false + schema: + type: integer + maximum: 1000 + minimum: 1 + description: Page size. + default: 100 + title: Page Size + description: Page size. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/ExperimentSessionFilter' + description: Filter sessions by test_case_id and status. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/ExperimentSessionResponsesPage' + '404': + description: Experiment not found + '503': + description: ClickHouse unavailable + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/ingest/atif: post: tags: @@ -15360,6 +15425,122 @@ components: required: - data title: ExperimentResponsesPage + ExperimentSessionFilter: + additionalProperties: false + description: Filter for listing ExperimentSessions. + properties: + test_case_id: + description: Filter by producer-supplied test case id. + title: Test Case Id + type: string + status: + description: Filter by root-span status (success, error, cancelled, unknown). + title: Status + type: string + title: ExperimentSessionFilter + type: object + ExperimentSessionResponse: + properties: + workspace: + type: string + title: Workspace + experiment_name: + type: string + title: Experiment Name + session_id: + type: string + title: Session Id + test_case_id: + title: Test Case Id + description: Producer-supplied test case identifier; null when the producer + did not set one. + type: string + trace_id: + type: string + title: Trace Id + root_span_id: + type: string + title: Root Span Id + started_at: + type: string + format: date-time + title: Started At + ended_at: + title: Ended At + type: string + format: date-time + latency_ms: + title: Latency Ms + type: number + status: + allOf: + - $ref: '#/components/schemas/SpanStatus' + description: 'Root-span status: success, error, cancelled, or unknown.' + input: + title: Input + description: Root-span input text (the query). + type: string + input_tokens: + title: Input Tokens + description: Sum of input tokens across this session's spans. + type: integer + output_tokens: + title: Output Tokens + description: Sum of output tokens across this session's spans. + type: integer + cached_tokens: + title: Cached Tokens + description: Sum of cached tokens across this session's spans. + type: integer + cost_total_usd: + title: Cost Total Usd + description: Sum of cost across this session's spans. + type: number + evaluator_scores: + additionalProperties: + type: number + type: object + title: Evaluator Scores + description: Per-evaluator session-mean score. Includes NUMERIC and BOOLEAN + evaluator results only; text/categorical results are omitted. + type: object + required: + - workspace + - experiment_name + - session_id + - trace_id + - root_span_id + - started_at + - status + title: ExperimentSessionResponse + description: "One ingested session of an Experiment \u2014 a single test case\ + \ execution.\n\nHydrated from ClickHouse at read time by joining ``experiment_sessions``\ + \ with\nthe session's root span (for status, input, tokens, cost) and\n``evaluator_results``\ + \ (for per-evaluator session-mean scores)." + ExperimentSessionResponsesPage: + properties: + data: + items: + $ref: '#/components/schemas/ExperimentSessionResponse' + type: array + title: Data + pagination: + allOf: + - $ref: '#/components/schemas/PaginationData' + description: Pagination information. + sort: + title: Sort + description: The field on which the results are sorted. + type: string + filter: + title: Filter + description: Filtering information. + additionalProperties: true + type: object + type: object + required: + - data + title: ExperimentSessionResponsesPage ExtendedBenchmark: properties: name: diff --git a/openapi/ga/openapi.yaml b/openapi/ga/openapi.yaml index efa7c4b301..8135fb7c79 100644 --- a/openapi/ga/openapi.yaml +++ b/openapi/ga/openapi.yaml @@ -6299,6 +6299,71 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions: + get: + tags: + - Experiments + summary: List Experiment Sessions + operationId: list_experiment_sessions_apis_intake_v2_workspaces__workspace__experiments__name__sessions_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + - name: page + in: query + required: false + schema: + type: integer + minimum: 1 + description: Page number. + default: 1 + title: Page + description: Page number. + - name: page_size + in: query + required: false + schema: + type: integer + maximum: 1000 + minimum: 1 + description: Page size. + default: 100 + title: Page Size + description: Page size. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/ExperimentSessionFilter' + description: Filter sessions by test_case_id and status. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/ExperimentSessionResponsesPage' + '404': + description: Experiment not found + '503': + description: ClickHouse unavailable + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/ingest/atif: post: tags: @@ -15360,6 +15425,122 @@ components: required: - data title: ExperimentResponsesPage + ExperimentSessionFilter: + additionalProperties: false + description: Filter for listing ExperimentSessions. + properties: + test_case_id: + description: Filter by producer-supplied test case id. + title: Test Case Id + type: string + status: + description: Filter by root-span status (success, error, cancelled, unknown). + title: Status + type: string + title: ExperimentSessionFilter + type: object + ExperimentSessionResponse: + properties: + workspace: + type: string + title: Workspace + experiment_name: + type: string + title: Experiment Name + session_id: + type: string + title: Session Id + test_case_id: + title: Test Case Id + description: Producer-supplied test case identifier; null when the producer + did not set one. + type: string + trace_id: + type: string + title: Trace Id + root_span_id: + type: string + title: Root Span Id + started_at: + type: string + format: date-time + title: Started At + ended_at: + title: Ended At + type: string + format: date-time + latency_ms: + title: Latency Ms + type: number + status: + allOf: + - $ref: '#/components/schemas/SpanStatus' + description: 'Root-span status: success, error, cancelled, or unknown.' + input: + title: Input + description: Root-span input text (the query). + type: string + input_tokens: + title: Input Tokens + description: Sum of input tokens across this session's spans. + type: integer + output_tokens: + title: Output Tokens + description: Sum of output tokens across this session's spans. + type: integer + cached_tokens: + title: Cached Tokens + description: Sum of cached tokens across this session's spans. + type: integer + cost_total_usd: + title: Cost Total Usd + description: Sum of cost across this session's spans. + type: number + evaluator_scores: + additionalProperties: + type: number + type: object + title: Evaluator Scores + description: Per-evaluator session-mean score. Includes NUMERIC and BOOLEAN + evaluator results only; text/categorical results are omitted. + type: object + required: + - workspace + - experiment_name + - session_id + - trace_id + - root_span_id + - started_at + - status + title: ExperimentSessionResponse + description: "One ingested session of an Experiment \u2014 a single test case\ + \ execution.\n\nHydrated from ClickHouse at read time by joining ``experiment_sessions``\ + \ with\nthe session's root span (for status, input, tokens, cost) and\n``evaluator_results``\ + \ (for per-evaluator session-mean scores)." + ExperimentSessionResponsesPage: + properties: + data: + items: + $ref: '#/components/schemas/ExperimentSessionResponse' + type: array + title: Data + pagination: + allOf: + - $ref: '#/components/schemas/PaginationData' + description: Pagination information. + sort: + title: Sort + description: The field on which the results are sorted. + type: string + filter: + title: Filter + description: Filtering information. + additionalProperties: true + type: object + type: object + required: + - data + title: ExperimentSessionResponsesPage ExtendedBenchmark: properties: name: diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index efa7c4b301..8135fb7c79 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -6299,6 +6299,71 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions: + get: + tags: + - Experiments + summary: List Experiment Sessions + operationId: list_experiment_sessions_apis_intake_v2_workspaces__workspace__experiments__name__sessions_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + - name: page + in: query + required: false + schema: + type: integer + minimum: 1 + description: Page number. + default: 1 + title: Page + description: Page number. + - name: page_size + in: query + required: false + schema: + type: integer + maximum: 1000 + minimum: 1 + description: Page size. + default: 100 + title: Page Size + description: Page size. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/ExperimentSessionFilter' + description: Filter sessions by test_case_id and status. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/ExperimentSessionResponsesPage' + '404': + description: Experiment not found + '503': + description: ClickHouse unavailable + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/ingest/atif: post: tags: @@ -15360,6 +15425,122 @@ components: required: - data title: ExperimentResponsesPage + ExperimentSessionFilter: + additionalProperties: false + description: Filter for listing ExperimentSessions. + properties: + test_case_id: + description: Filter by producer-supplied test case id. + title: Test Case Id + type: string + status: + description: Filter by root-span status (success, error, cancelled, unknown). + title: Status + type: string + title: ExperimentSessionFilter + type: object + ExperimentSessionResponse: + properties: + workspace: + type: string + title: Workspace + experiment_name: + type: string + title: Experiment Name + session_id: + type: string + title: Session Id + test_case_id: + title: Test Case Id + description: Producer-supplied test case identifier; null when the producer + did not set one. + type: string + trace_id: + type: string + title: Trace Id + root_span_id: + type: string + title: Root Span Id + started_at: + type: string + format: date-time + title: Started At + ended_at: + title: Ended At + type: string + format: date-time + latency_ms: + title: Latency Ms + type: number + status: + allOf: + - $ref: '#/components/schemas/SpanStatus' + description: 'Root-span status: success, error, cancelled, or unknown.' + input: + title: Input + description: Root-span input text (the query). + type: string + input_tokens: + title: Input Tokens + description: Sum of input tokens across this session's spans. + type: integer + output_tokens: + title: Output Tokens + description: Sum of output tokens across this session's spans. + type: integer + cached_tokens: + title: Cached Tokens + description: Sum of cached tokens across this session's spans. + type: integer + cost_total_usd: + title: Cost Total Usd + description: Sum of cost across this session's spans. + type: number + evaluator_scores: + additionalProperties: + type: number + type: object + title: Evaluator Scores + description: Per-evaluator session-mean score. Includes NUMERIC and BOOLEAN + evaluator results only; text/categorical results are omitted. + type: object + required: + - workspace + - experiment_name + - session_id + - trace_id + - root_span_id + - started_at + - status + title: ExperimentSessionResponse + description: "One ingested session of an Experiment \u2014 a single test case\ + \ execution.\n\nHydrated from ClickHouse at read time by joining ``experiment_sessions``\ + \ with\nthe session's root span (for status, input, tokens, cost) and\n``evaluator_results``\ + \ (for per-evaluator session-mean scores)." + ExperimentSessionResponsesPage: + properties: + data: + items: + $ref: '#/components/schemas/ExperimentSessionResponse' + type: array + title: Data + pagination: + allOf: + - $ref: '#/components/schemas/PaginationData' + description: Pagination information. + sort: + title: Sort + description: The field on which the results are sorted. + type: string + filter: + title: Filter + description: Filtering information. + additionalProperties: true + type: object + type: object + required: + - data + title: ExperimentSessionResponsesPage ExtendedBenchmark: properties: name: diff --git a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml index efa7c4b301..8135fb7c79 100644 --- a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml +++ b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml @@ -6299,6 +6299,71 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions: + get: + tags: + - Experiments + summary: List Experiment Sessions + operationId: list_experiment_sessions_apis_intake_v2_workspaces__workspace__experiments__name__sessions_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: name + in: path + required: true + schema: + type: string + title: Name + - name: page + in: query + required: false + schema: + type: integer + minimum: 1 + description: Page number. + default: 1 + title: Page + description: Page number. + - name: page_size + in: query + required: false + schema: + type: integer + maximum: 1000 + minimum: 1 + description: Page size. + default: 100 + title: Page Size + description: Page size. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/ExperimentSessionFilter' + description: Filter sessions by test_case_id and status. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/ExperimentSessionResponsesPage' + '404': + description: Experiment not found + '503': + description: ClickHouse unavailable + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/ingest/atif: post: tags: @@ -15360,6 +15425,122 @@ components: required: - data title: ExperimentResponsesPage + ExperimentSessionFilter: + additionalProperties: false + description: Filter for listing ExperimentSessions. + properties: + test_case_id: + description: Filter by producer-supplied test case id. + title: Test Case Id + type: string + status: + description: Filter by root-span status (success, error, cancelled, unknown). + title: Status + type: string + title: ExperimentSessionFilter + type: object + ExperimentSessionResponse: + properties: + workspace: + type: string + title: Workspace + experiment_name: + type: string + title: Experiment Name + session_id: + type: string + title: Session Id + test_case_id: + title: Test Case Id + description: Producer-supplied test case identifier; null when the producer + did not set one. + type: string + trace_id: + type: string + title: Trace Id + root_span_id: + type: string + title: Root Span Id + started_at: + type: string + format: date-time + title: Started At + ended_at: + title: Ended At + type: string + format: date-time + latency_ms: + title: Latency Ms + type: number + status: + allOf: + - $ref: '#/components/schemas/SpanStatus' + description: 'Root-span status: success, error, cancelled, or unknown.' + input: + title: Input + description: Root-span input text (the query). + type: string + input_tokens: + title: Input Tokens + description: Sum of input tokens across this session's spans. + type: integer + output_tokens: + title: Output Tokens + description: Sum of output tokens across this session's spans. + type: integer + cached_tokens: + title: Cached Tokens + description: Sum of cached tokens across this session's spans. + type: integer + cost_total_usd: + title: Cost Total Usd + description: Sum of cost across this session's spans. + type: number + evaluator_scores: + additionalProperties: + type: number + type: object + title: Evaluator Scores + description: Per-evaluator session-mean score. Includes NUMERIC and BOOLEAN + evaluator results only; text/categorical results are omitted. + type: object + required: + - workspace + - experiment_name + - session_id + - trace_id + - root_span_id + - started_at + - status + title: ExperimentSessionResponse + description: "One ingested session of an Experiment \u2014 a single test case\ + \ execution.\n\nHydrated from ClickHouse at read time by joining ``experiment_sessions``\ + \ with\nthe session's root span (for status, input, tokens, cost) and\n``evaluator_results``\ + \ (for per-evaluator session-mean scores)." + ExperimentSessionResponsesPage: + properties: + data: + items: + $ref: '#/components/schemas/ExperimentSessionResponse' + type: array + title: Data + pagination: + allOf: + - $ref: '#/components/schemas/PaginationData' + description: Pagination information. + sort: + title: Sort + description: The field on which the results are sorted. + type: string + filter: + title: Filter + description: Filtering information. + additionalProperties: true + type: object + type: object + required: + - data + title: ExperimentSessionResponsesPage ExtendedBenchmark: properties: name: diff --git a/sdk/python/nemo-platform/.nmpcontext/stainless.yaml b/sdk/python/nemo-platform/.nmpcontext/stainless.yaml index 71969fca67..2500eed76e 100644 --- a/sdk/python/nemo-platform/.nmpcontext/stainless.yaml +++ b/sdk/python/nemo-platform/.nmpcontext/stainless.yaml @@ -1109,3 +1109,11 @@ resources: retrieve: get /apis/intake/v2/workspaces/{workspace}/experiments/{name} update: put /apis/intake/v2/workspaces/{workspace}/experiments/{name} delete: delete /apis/intake/v2/workspaces/{workspace}/experiments/{name} + subresources: + sessions: + models: + experiment_session_filter: ExperimentSessionFilter + experiment_session_response: ExperimentSessionResponse + experiment_session_responses_page: ExperimentSessionResponsesPage + methods: + list: get /apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/__init__.py b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/__init__.py index 9dbe086588..90ed5b2c22 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/__init__.py +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/__init__.py @@ -15,6 +15,14 @@ # File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. +from .sessions import ( + SessionsResource, + AsyncSessionsResource, + SessionsResourceWithRawResponse, + AsyncSessionsResourceWithRawResponse, + SessionsResourceWithStreamingResponse, + AsyncSessionsResourceWithStreamingResponse, +) from .experiments import ( ExperimentsResource, AsyncExperimentsResource, @@ -25,6 +33,12 @@ ) __all__ = [ + "SessionsResource", + "AsyncSessionsResource", + "SessionsResourceWithRawResponse", + "AsyncSessionsResourceWithRawResponse", + "SessionsResourceWithStreamingResponse", + "AsyncSessionsResourceWithStreamingResponse", "ExperimentsResource", "AsyncExperimentsResource", "ExperimentsResourceWithRawResponse", diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/api.md b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/api.md index 0c72350ab0..57f4bd2b37 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/api.md +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/api.md @@ -19,3 +19,19 @@ Methods: - client.experiments.update(path_name, \*, workspace, \*\*params) -> ExperimentResponse - client.experiments.list(\*, workspace, \*\*params) -> SyncDefaultPagination[ExperimentResponse] - client.experiments.delete(name, \*, workspace) -> None + +## Sessions + +Types: + +```python +from nemo_platform.types.experiments import ( + ExperimentSessionFilter, + ExperimentSessionResponse, + ExperimentSessionResponsesPage, +) +``` + +Methods: + +- client.experiments.sessions.list(name, \*, workspace, \*\*params) -> SyncDefaultPagination[ExperimentSessionResponse] diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/experiments.py b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/experiments.py index ed0a2fd33e..2495e5ea4d 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/experiments.py +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/experiments.py @@ -24,6 +24,14 @@ from ..._types import Body, Omit, Query, Headers, NoneType, NotGiven, omit, not_given from ..._utils import path_template, maybe_transform, async_maybe_transform +from .sessions import ( + SessionsResource, + AsyncSessionsResource, + SessionsResourceWithRawResponse, + AsyncSessionsResourceWithRawResponse, + SessionsResourceWithStreamingResponse, + AsyncSessionsResourceWithStreamingResponse, +) from ..._compat import cached_property from ..._resource import SyncAPIResource, AsyncAPIResource from ..._response import ( @@ -47,6 +55,10 @@ class ExperimentsResource(SyncAPIResource): + @cached_property + def sessions(self) -> SessionsResource: + return SessionsResource(self._client) + @cached_property def with_raw_response(self) -> ExperimentsResourceWithRawResponse: """ @@ -385,6 +397,10 @@ def delete( class AsyncExperimentsResource(AsyncAPIResource): + @cached_property + def sessions(self) -> AsyncSessionsResource: + return AsyncSessionsResource(self._client) + @cached_property def with_raw_response(self) -> AsyncExperimentsResourceWithRawResponse: """ @@ -742,6 +758,10 @@ def __init__(self, experiments: ExperimentsResource) -> None: experiments.delete, ) + @cached_property + def sessions(self) -> SessionsResourceWithRawResponse: + return SessionsResourceWithRawResponse(self._experiments.sessions) + class AsyncExperimentsResourceWithRawResponse: def __init__(self, experiments: AsyncExperimentsResource) -> None: @@ -763,6 +783,10 @@ def __init__(self, experiments: AsyncExperimentsResource) -> None: experiments.delete, ) + @cached_property + def sessions(self) -> AsyncSessionsResourceWithRawResponse: + return AsyncSessionsResourceWithRawResponse(self._experiments.sessions) + class ExperimentsResourceWithStreamingResponse: def __init__(self, experiments: ExperimentsResource) -> None: @@ -784,6 +808,10 @@ def __init__(self, experiments: ExperimentsResource) -> None: experiments.delete, ) + @cached_property + def sessions(self) -> SessionsResourceWithStreamingResponse: + return SessionsResourceWithStreamingResponse(self._experiments.sessions) + class AsyncExperimentsResourceWithStreamingResponse: def __init__(self, experiments: AsyncExperimentsResource) -> None: @@ -804,3 +832,7 @@ def __init__(self, experiments: AsyncExperimentsResource) -> None: self.delete = async_to_streamed_response_wrapper( experiments.delete, ) + + @cached_property + def sessions(self) -> AsyncSessionsResourceWithStreamingResponse: + return AsyncSessionsResourceWithStreamingResponse(self._experiments.sessions) diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/sessions.py b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/sessions.py new file mode 100644 index 0000000000..4ab63ee4d7 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/experiments/sessions.py @@ -0,0 +1,238 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +import httpx + +from ..._types import Body, Omit, Query, Headers, NotGiven, omit, not_given +from ..._utils import path_template, maybe_transform +from ..._compat import cached_property +from ..._resource import SyncAPIResource, AsyncAPIResource +from ..._response import ( + to_raw_response_wrapper, + to_streamed_response_wrapper, + async_to_raw_response_wrapper, + async_to_streamed_response_wrapper, +) +from ...pagination import SyncDefaultPagination, AsyncDefaultPagination +from ..._base_client import AsyncPaginator, make_request_options +from ...types.experiments import session_list_params +from ...types.experiments.experiment_session_response import ExperimentSessionResponse +from ...types.experiments.experiment_session_filter_param import ExperimentSessionFilterParam + +__all__ = ["SessionsResource", "AsyncSessionsResource"] + + +class SessionsResource(SyncAPIResource): + @cached_property + def with_raw_response(self) -> SessionsResourceWithRawResponse: + """ + This property can be used as a prefix for any HTTP method call to return + the raw response object instead of the parsed content. + + For more information, see https://docs.nvidia.com/nemo/microservices/latest/pysdk/index.html#accessing-raw-response-data-e-g-headers + """ + return SessionsResourceWithRawResponse(self) + + @cached_property + def with_streaming_response(self) -> SessionsResourceWithStreamingResponse: + """ + An alternative to `.with_raw_response` that doesn't eagerly read the response body. + + For more information, see https://docs.nvidia.com/nemo/microservices/latest/pysdk/index.html#with_streaming_response + """ + return SessionsResourceWithStreamingResponse(self) + + def list( + self, + name: str, + *, + workspace: str | None = None, + filter: ExperimentSessionFilterParam | Omit = omit, + page: int | Omit = omit, + page_size: int | Omit = omit, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> SyncDefaultPagination[ExperimentSessionResponse]: + """ + List Experiment Sessions + + Args: + filter: Filter sessions by test_case_id and status. + + page: Page number. + + page_size: Page size. + + extra_headers: Send extra headers + + extra_query: Add additional query parameters to the request + + extra_body: Add additional JSON properties to the request + + timeout: Override the client-level default timeout for this request, in seconds + """ + if workspace is None: + workspace = self._client._get_workspace_path_param() + if not workspace: + raise ValueError(f"Expected a non-empty value for `workspace` but received {workspace!r}") + if not name: + raise ValueError(f"Expected a non-empty value for `name` but received {name!r}") + return self._get_api_list( + path_template( + "/apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions", workspace=workspace, name=name + ), + page=SyncDefaultPagination[ExperimentSessionResponse], + options=make_request_options( + extra_headers=extra_headers, + extra_query=extra_query, + extra_body=extra_body, + timeout=timeout, + query=maybe_transform( + { + "filter": filter, + "page": page, + "page_size": page_size, + }, + session_list_params.SessionListParams, + ), + ), + model=ExperimentSessionResponse, + ) + + +class AsyncSessionsResource(AsyncAPIResource): + @cached_property + def with_raw_response(self) -> AsyncSessionsResourceWithRawResponse: + """ + This property can be used as a prefix for any HTTP method call to return + the raw response object instead of the parsed content. + + For more information, see https://docs.nvidia.com/nemo/microservices/latest/pysdk/index.html#accessing-raw-response-data-e-g-headers + """ + return AsyncSessionsResourceWithRawResponse(self) + + @cached_property + def with_streaming_response(self) -> AsyncSessionsResourceWithStreamingResponse: + """ + An alternative to `.with_raw_response` that doesn't eagerly read the response body. + + For more information, see https://docs.nvidia.com/nemo/microservices/latest/pysdk/index.html#with_streaming_response + """ + return AsyncSessionsResourceWithStreamingResponse(self) + + def list( + self, + name: str, + *, + workspace: str | None = None, + filter: ExperimentSessionFilterParam | Omit = omit, + page: int | Omit = omit, + page_size: int | Omit = omit, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> AsyncPaginator[ExperimentSessionResponse, AsyncDefaultPagination[ExperimentSessionResponse]]: + """ + List Experiment Sessions + + Args: + filter: Filter sessions by test_case_id and status. + + page: Page number. + + page_size: Page size. + + extra_headers: Send extra headers + + extra_query: Add additional query parameters to the request + + extra_body: Add additional JSON properties to the request + + timeout: Override the client-level default timeout for this request, in seconds + """ + if workspace is None: + workspace = self._client._get_workspace_path_param() + if not workspace: + raise ValueError(f"Expected a non-empty value for `workspace` but received {workspace!r}") + if not name: + raise ValueError(f"Expected a non-empty value for `name` but received {name!r}") + return self._get_api_list( + path_template( + "/apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions", workspace=workspace, name=name + ), + page=AsyncDefaultPagination[ExperimentSessionResponse], + options=make_request_options( + extra_headers=extra_headers, + extra_query=extra_query, + extra_body=extra_body, + timeout=timeout, + query=maybe_transform( + { + "filter": filter, + "page": page, + "page_size": page_size, + }, + session_list_params.SessionListParams, + ), + ), + model=ExperimentSessionResponse, + ) + + +class SessionsResourceWithRawResponse: + def __init__(self, sessions: SessionsResource) -> None: + self._sessions = sessions + + self.list = to_raw_response_wrapper( + sessions.list, + ) + + +class AsyncSessionsResourceWithRawResponse: + def __init__(self, sessions: AsyncSessionsResource) -> None: + self._sessions = sessions + + self.list = async_to_raw_response_wrapper( + sessions.list, + ) + + +class SessionsResourceWithStreamingResponse: + def __init__(self, sessions: SessionsResource) -> None: + self._sessions = sessions + + self.list = to_streamed_response_wrapper( + sessions.list, + ) + + +class AsyncSessionsResourceWithStreamingResponse: + def __init__(self, sessions: AsyncSessionsResource) -> None: + self._sessions = sessions + + self.list = async_to_streamed_response_wrapper( + sessions.list, + ) diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/experiments/__init__.py b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/__init__.py index 9663ee8371..1fae468515 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/experiments/__init__.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/__init__.py @@ -19,8 +19,12 @@ from .evaluator_aggregate import EvaluatorAggregate as EvaluatorAggregate from .experiment_response import ExperimentResponse as ExperimentResponse +from .session_list_params import SessionListParams as SessionListParams from .experiment_list_params import ExperimentListParams as ExperimentListParams from .experiment_filter_param import ExperimentFilterParam as ExperimentFilterParam from .experiment_create_params import ExperimentCreateParams as ExperimentCreateParams from .experiment_update_params import ExperimentUpdateParams as ExperimentUpdateParams from .experiment_responses_page import ExperimentResponsesPage as ExperimentResponsesPage +from .experiment_session_response import ExperimentSessionResponse as ExperimentSessionResponse +from .experiment_session_filter_param import ExperimentSessionFilterParam as ExperimentSessionFilterParam +from .experiment_session_responses_page import ExperimentSessionResponsesPage as ExperimentSessionResponsesPage diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_filter_param.py b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_filter_param.py new file mode 100644 index 0000000000..579b250f7c --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_filter_param.py @@ -0,0 +1,32 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +from typing_extensions import TypedDict + +__all__ = ["ExperimentSessionFilterParam"] + + +class ExperimentSessionFilterParam(TypedDict, total=False): + """Filter for listing ExperimentSessions.""" + + status: str + """Filter by root-span status (success, error, cancelled, unknown).""" + + test_case_id: str + """Filter by producer-supplied test case id.""" diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_response.py b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_response.py new file mode 100644 index 0000000000..4b15b5a28e --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_response.py @@ -0,0 +1,77 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from typing import Dict, Optional +from datetime import datetime + +from ..._models import BaseModel +from ..intake.span_status import SpanStatus + +__all__ = ["ExperimentSessionResponse"] + + +class ExperimentSessionResponse(BaseModel): + """One ingested session of an Experiment — a single test case execution. + + Hydrated from ClickHouse at read time by joining ``experiment_sessions`` with + the session's root span (for status, input, tokens, cost) and + ``evaluator_results`` (for per-evaluator session-mean scores). + """ + + experiment_name: str + + root_span_id: str + + session_id: str + + started_at: datetime + + status: SpanStatus + """Root-span status: success, error, cancelled, or unknown.""" + + trace_id: str + + workspace: str + + cached_tokens: Optional[int] = None + """Sum of cached tokens across this session's spans.""" + + cost_total_usd: Optional[float] = None + """Sum of cost across this session's spans.""" + + ended_at: Optional[datetime] = None + + evaluator_scores: Optional[Dict[str, float]] = None + """Per-evaluator session-mean score. + + Includes NUMERIC and BOOLEAN evaluator results only; text/categorical results + are omitted. + """ + + input: Optional[str] = None + """Root-span input text (the query).""" + + input_tokens: Optional[int] = None + """Sum of input tokens across this session's spans.""" + + latency_ms: Optional[float] = None + + output_tokens: Optional[int] = None + """Sum of output tokens across this session's spans.""" + + test_case_id: Optional[str] = None + """Producer-supplied test case identifier; null when the producer did not set one.""" diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_responses_page.py b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_responses_page.py new file mode 100644 index 0000000000..476c161ea2 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/experiment_session_responses_page.py @@ -0,0 +1,37 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from typing import Dict, List, Optional + +from ..._models import BaseModel +from ..shared.pagination_data import PaginationData +from .experiment_session_response import ExperimentSessionResponse + +__all__ = ["ExperimentSessionResponsesPage"] + + +class ExperimentSessionResponsesPage(BaseModel): + data: List[ExperimentSessionResponse] + + filter: Optional[Dict[str, object]] = None + """Filtering information.""" + + pagination: Optional[PaginationData] = None + """Pagination information.""" + + sort: Optional[str] = None + """The field on which the results are sorted.""" diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/experiments/session_list_params.py b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/session_list_params.py new file mode 100644 index 0000000000..18a1f7e62e --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/experiments/session_list_params.py @@ -0,0 +1,37 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +from typing_extensions import TypedDict + +from .experiment_session_filter_param import ExperimentSessionFilterParam + +__all__ = ["SessionListParams"] + + +class SessionListParams(TypedDict, total=False): + workspace: str + + filter: ExperimentSessionFilterParam + """Filter sessions by test_case_id and status.""" + + page: int + """Page number.""" + + page_size: int + """Page size.""" diff --git a/sdk/python/nemo-platform/tests/api_resources/experiments/test_sessions.py b/sdk/python/nemo-platform/tests/api_resources/experiments/test_sessions.py new file mode 100644 index 0000000000..b8f2834eae --- /dev/null +++ b/sdk/python/nemo-platform/tests/api_resources/experiments/test_sessions.py @@ -0,0 +1,174 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +import os +from typing import Any, cast + +import pytest + +from tests.utils import assert_matches_type +from nemo_platform import NeMoPlatform, AsyncNeMoPlatform +from nemo_platform.pagination import SyncDefaultPagination, AsyncDefaultPagination +from nemo_platform.types.experiments import ExperimentSessionResponse + +base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") + + +class TestSessions: + parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_method_list(self, client: NeMoPlatform) -> None: + session = client.experiments.sessions.list( + name="name", + workspace="workspace", + ) + assert_matches_type(SyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_method_list_with_all_params(self, client: NeMoPlatform) -> None: + session = client.experiments.sessions.list( + name="name", + workspace="workspace", + filter={ + "status": "status", + "test_case_id": "test_case_id", + }, + page=1, + page_size=1, + ) + assert_matches_type(SyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_raw_response_list(self, client: NeMoPlatform) -> None: + response = client.experiments.sessions.with_raw_response.list( + name="name", + workspace="workspace", + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + session = response.parse() + assert_matches_type(SyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_streaming_response_list(self, client: NeMoPlatform) -> None: + with client.experiments.sessions.with_streaming_response.list( + name="name", + workspace="workspace", + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + session = response.parse() + assert_matches_type(SyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_path_params_list(self, client: NeMoPlatform) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `workspace` but received ''"): + client.experiments.sessions.with_raw_response.list( + name="name", + workspace="", + ) + + with pytest.raises(ValueError, match=r"Expected a non-empty value for `name` but received ''"): + client.experiments.sessions.with_raw_response.list( + name="", + workspace="workspace", + ) + + +class TestAsyncSessions: + parametrize = pytest.mark.parametrize( + "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"] + ) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_method_list(self, async_client: AsyncNeMoPlatform) -> None: + session = await async_client.experiments.sessions.list( + name="name", + workspace="workspace", + ) + assert_matches_type(AsyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_method_list_with_all_params(self, async_client: AsyncNeMoPlatform) -> None: + session = await async_client.experiments.sessions.list( + name="name", + workspace="workspace", + filter={ + "status": "status", + "test_case_id": "test_case_id", + }, + page=1, + page_size=1, + ) + assert_matches_type(AsyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_raw_response_list(self, async_client: AsyncNeMoPlatform) -> None: + response = await async_client.experiments.sessions.with_raw_response.list( + name="name", + workspace="workspace", + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + session = await response.parse() + assert_matches_type(AsyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_streaming_response_list(self, async_client: AsyncNeMoPlatform) -> None: + async with async_client.experiments.sessions.with_streaming_response.list( + name="name", + workspace="workspace", + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + session = await response.parse() + assert_matches_type(AsyncDefaultPagination[ExperimentSessionResponse], session, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_path_params_list(self, async_client: AsyncNeMoPlatform) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `workspace` but received ''"): + await async_client.experiments.sessions.with_raw_response.list( + name="name", + workspace="", + ) + + with pytest.raises(ValueError, match=r"Expected a non-empty value for `name` but received ''"): + await async_client.experiments.sessions.with_raw_response.list( + name="", + workspace="workspace", + ) diff --git a/sdk/stainless.yaml b/sdk/stainless.yaml index 71969fca67..2500eed76e 100644 --- a/sdk/stainless.yaml +++ b/sdk/stainless.yaml @@ -1109,3 +1109,11 @@ resources: retrieve: get /apis/intake/v2/workspaces/{workspace}/experiments/{name} update: put /apis/intake/v2/workspaces/{workspace}/experiments/{name} delete: delete /apis/intake/v2/workspaces/{workspace}/experiments/{name} + subresources: + sessions: + models: + experiment_session_filter: ExperimentSessionFilter + experiment_session_response: ExperimentSessionResponse + experiment_session_responses_page: ExperimentSessionResponsesPage + methods: + list: get /apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions diff --git a/services/core/auth/src/nmp/core/auth/assets/static-authz.yaml b/services/core/auth/src/nmp/core/auth/assets/static-authz.yaml index 1117d2b97d..f4619b7524 100644 --- a/services/core/auth/src/nmp/core/auth/assets/static-authz.yaml +++ b/services/core/auth/src/nmp/core/auth/assets/static-authz.yaml @@ -1260,6 +1260,13 @@ authz: scopes: - intake:write - platform:write + /apis/intake/v2/workspaces/{workspace}/experiments/{name}/sessions: + get: + permissions: + - intake.experiments.read + scopes: + - intake:read + - platform:read /apis/intake/v2/workspaces/{workspace}/ingest/atif: post: permissions: 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 106d4905c8..283395e3b8 100644 --- a/services/intake/src/nmp/intake/api/v2/experiments/endpoints.py +++ b/services/intake/src/nmp/intake/api/v2/experiments/endpoints.py @@ -28,16 +28,28 @@ ExperimentGroupResponse, ExperimentRequest, ExperimentResponse, + ExperimentSessionFilter, + ExperimentSessionResponse, ) from nmp.intake.entities.experiments import Experiment, ExperimentGroup from nmp.intake.spans.api.dependencies import require_workspace_access, validate_list_query_params +from nmp.intake.spans.clickhouse_client import ClickHouseSpanClient +from nmp.intake.spans.domain import SpanStatus from nmp.intake.spans.experiment_rollup_repository import ( ExperimentRollup, ExperimentRollupRepository, ScoreRollup, ) +from nmp.intake.spans.experiment_session_repository import ExperimentSessionRepository +from nmp.intake.spans.storage import make_pagination logger = logging.getLogger(__name__) + + +def _sanitize_for_log(value: str) -> str: + return value.replace("\r", "").replace("\n", "") + + router = APIRouter(dependencies=[Depends(require_workspace_access)]) GROUPS_TAG = "Experiment Groups" @@ -49,26 +61,36 @@ EntityClientDep = Annotated[EntityClient, Depends(get_entity_client)] ExperimentGroupFilterDep = Annotated[ParsedFilter, Depends(make_filter_dep(ExperimentGroupFilter))] ExperimentFilterDep = Annotated[ParsedFilter, Depends(make_filter_dep(ExperimentFilter))] +ExperimentSessionFilterDep = Annotated[ParsedFilter, Depends(make_filter_dep(ExperimentSessionFilter))] -def get_experiment_rollup_repository(request: Request) -> ExperimentRollupRepository | None: - # Rollups are enrichment only. Experiment entity reads should continue when - # ClickHouse is disabled or temporarily unavailable. - service = getattr(request.app.state, "intake_service", None) - if service is None: - service = getattr(request.app.state, "service", None) +def _get_clickhouse_client(request: Request) -> ClickHouseSpanClient | None: + service = getattr(request.app.state, "intake_service", None) or getattr(request.app.state, "service", None) if service is None: return None + return getattr(service, "clickhouse_client", None) - service_client = getattr(service, "clickhouse_client", None) - if service_client is None: - return None - return ExperimentRollupRepository(service_client) + +def get_experiment_rollup_repository(request: Request) -> ExperimentRollupRepository | None: + # Rollups are enrichment only. Experiment entity reads should continue when + # ClickHouse is disabled or temporarily unavailable. + client = _get_clickhouse_client(request) + return ExperimentRollupRepository(client) if client is not None else None ExperimentRollupRepositoryDep = Annotated[ExperimentRollupRepository | None, Depends(get_experiment_rollup_repository)] +def get_experiment_session_repository(request: Request) -> ExperimentSessionRepository | None: + client = _get_clickhouse_client(request) + return ExperimentSessionRepository(client) if client is not None else None + + +ExperimentSessionRepositoryDep = Annotated[ + ExperimentSessionRepository | None, Depends(get_experiment_session_repository) +] + + @router.post( "/v2/workspaces/{workspace}/experiment-groups", response_model=ExperimentGroupResponse, @@ -371,6 +393,86 @@ async def delete_experiment( ) +@router.get( + "/v2/workspaces/{workspace}/experiments/{name}/sessions", + response_model=Page[ExperimentSessionResponse], + tags=[EXPERIMENTS_TAG], + responses={ + 404: {"description": "Experiment not found"}, + 503: {"description": "ClickHouse unavailable"}, + }, + openapi_extra=generate_openapi_extra_params( + filter_schema=ExperimentSessionFilter, + filter_description="Filter sessions by test_case_id and status.", + ), +) +async def list_experiment_sessions( + workspace: str, + name: str, + request: Request, + entity_client: EntityClientDep, + session_repository: ExperimentSessionRepositoryDep, + parsed: ExperimentSessionFilterDep, + page: int = Query(default=1, ge=1, description="Page number."), + page_size: int = Query(default=100, ge=1, le=1000, description="Page size."), +) -> Page[ExperimentSessionResponse]: + validate_list_query_params(request) + await _get_or_404( + entity_client, + Experiment, + workspace=workspace, + name=name, + label="Experiment", + ) + if session_repository is None: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="ClickHouse is unavailable; per-session reads require telemetry storage.", + ) + test_case_id: str | None = parsed.extract("test_case_id") + status_raw: str | None = parsed.extract("status") + try: + status_filter = SpanStatus(status_raw) if status_raw is not None else None + except ValueError: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid status '{status_raw}'. Valid values: {[s.value for s in SpanStatus]}", + ) + try: + result = await session_repository.list_sessions( + workspace=workspace, + experiment_name=name, + status=status_filter, + test_case_id=test_case_id, + page=page, + page_size=page_size, + ) + except Exception as exc: + # Sessions are the response payload (not enrichment), so we can't silently degrade like + # _hydrate_rollups does. Convert backend failures (ClickHouse connection drop, query + # timeout, etc.) to a deterministic 503 instead of letting them bubble as 500s. + logger.exception( + "Per-session read failed for workspace=%s experiment=%s", + _sanitize_for_log(workspace), + _sanitize_for_log(name), + ) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Telemetry store unavailable.", + ) from exc + data = [ExperimentSessionResponse.from_row(row) for row in result.rows] + return Page( + data=data, + pagination=make_pagination( + page=page, + page_size=page_size, + current_page_size=len(data), + total_results=result.total, + ), + filter=parsed.to_response(), + ) + + async def _get_or_404( entity_client: EntityClient, entity_type: type[EntityT], 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 4847cdb879..184ea358e0 100644 --- a/services/intake/src/nmp/intake/api/v2/experiments/schemas.py +++ b/services/intake/src/nmp/intake/api/v2/experiments/schemas.py @@ -14,6 +14,8 @@ from nmp.common.entities.values import DatetimeFilter, Filter from nmp.intake.entities.experiments import Experiment, ExperimentGroup +from nmp.intake.spans.domain import SpanStatus +from nmp.intake.spans.experiment_session_repository import ExperimentSessionRow from pydantic import AnyUrl, BaseModel, ConfigDict, Field @@ -159,3 +161,72 @@ class ExperimentFilter(Filter): default=None, description="Filter experiments by last-updated timestamp; supports `$gte` and `$lte` for ranges.", ) + + +class ExperimentSessionFilter(Filter): + """Filter for listing ExperimentSessions.""" + + test_case_id: str | None = Field(default=None, description="Filter by producer-supplied test case id.") + status: str | None = Field( + default=None, description="Filter by root-span status (success, error, cancelled, unknown)." + ) + + +class ExperimentSessionResponse(BaseModel): + """One ingested session of an Experiment — a single test case execution. + + Hydrated from ClickHouse at read time by joining ``experiment_sessions`` with + the session's root span (for status, input, tokens, cost) and + ``evaluator_results`` (for per-evaluator session-mean scores). + """ + + workspace: str + experiment_name: str + session_id: str + test_case_id: str | None = Field( + default=None, + description="Producer-supplied test case identifier; null when the producer did not set one.", + ) + trace_id: str + root_span_id: str + + started_at: datetime + ended_at: datetime | None = None + latency_ms: float | None = None + + status: SpanStatus = Field(description="Root-span status: success, error, cancelled, or unknown.") + input: str | None = Field(default=None, description="Root-span input text (the query).") + + input_tokens: int | None = Field(default=None, description="Sum of input tokens across this session's spans.") + output_tokens: int | None = Field(default=None, description="Sum of output tokens across this session's spans.") + cached_tokens: int | None = Field(default=None, description="Sum of cached tokens across this session's spans.") + cost_total_usd: float | None = Field(default=None, description="Sum of cost across this session's spans.") + + evaluator_scores: dict[str, float] = Field( + default_factory=dict, + description=( + "Per-evaluator session-mean score. Includes NUMERIC and BOOLEAN evaluator results only; " + "text/categorical results are omitted." + ), + ) + + @classmethod + def from_row(cls, row: ExperimentSessionRow) -> ExperimentSessionResponse: + return cls( + workspace=row.workspace, + experiment_name=row.experiment_name, + session_id=row.session_id, + test_case_id=row.test_case_id, + trace_id=row.trace_id, + root_span_id=row.root_span_id, + started_at=row.started_at, + ended_at=row.ended_at, + latency_ms=row.latency_ms, + status=row.status, + input=row.input, + input_tokens=row.input_tokens, + output_tokens=row.output_tokens, + cached_tokens=row.cached_tokens, + cost_total_usd=row.cost_total_usd, + evaluator_scores=row.evaluator_scores, + ) diff --git a/services/intake/src/nmp/intake/spans/experiment_rollup_repository.py b/services/intake/src/nmp/intake/spans/experiment_rollup_repository.py index 61fae0320a..60bd3e2d24 100644 --- a/services/intake/src/nmp/intake/spans/experiment_rollup_repository.py +++ b/services/intake/src/nmp/intake/spans/experiment_rollup_repository.py @@ -10,7 +10,7 @@ from nmp.intake.spans.clickhouse_client import ClickHouseSpanClient from nmp.intake.spans.span_attribute_catalog import COST_SCALE, SpanAttributeField, spec_for_field -from nmp.intake.spans.storage import result_rows +from nmp.intake.spans.storage import float_or_none, result_rows from nmp.intake.spans.trace_repository import current_spans_sql @@ -72,12 +72,12 @@ async def get_rollups(self, *, workspace: str, experiment_ids: list[str]) -> dic ) ): rollups[row["experiment_id"]].evaluator_scores[row["evaluator_name"]] = ScoreRollup( - sum=_float_or_none(row["sum"]), - mean=_float_or_none(row["mean"]), - median=_float_or_none(row["median"]), - p90=_float_or_none(row["p90"]), - p95=_float_or_none(row["p95"]), - p99=_float_or_none(row["p99"]), + sum=float_or_none(row["sum"]), + mean=float_or_none(row["mean"]), + median=float_or_none(row["median"]), + p90=float_or_none(row["p90"]), + p95=float_or_none(row["p95"]), + p99=float_or_none(row["p99"]), count=int(row["count"]), ) @@ -271,12 +271,12 @@ def _score_rollup(row: dict[str, Any], prefix: str) -> ScoreRollup | None: if count == 0: return None return ScoreRollup( - sum=_float_or_none(row[f"{prefix}_sum"]), - mean=_float_or_none(row[f"{prefix}_mean"]), - median=_float_or_none(row[f"{prefix}_median"]), - p90=_float_or_none(row[f"{prefix}_p90"]), - p95=_float_or_none(row[f"{prefix}_p95"]), - p99=_float_or_none(row[f"{prefix}_p99"]), + sum=float_or_none(row[f"{prefix}_sum"]), + mean=float_or_none(row[f"{prefix}_mean"]), + median=float_or_none(row[f"{prefix}_median"]), + p90=float_or_none(row[f"{prefix}_p90"]), + p95=float_or_none(row[f"{prefix}_p95"]), + p99=float_or_none(row[f"{prefix}_p99"]), count=count, ) @@ -285,9 +285,3 @@ def _string_list(value: Any) -> list[str]: if value is None: return [] return sorted(str(item) for item in value if str(item)) - - -def _float_or_none(value: Any) -> float | None: - if value is None: - return None - return float(value) diff --git a/services/intake/src/nmp/intake/spans/experiment_session_repository.py b/services/intake/src/nmp/intake/spans/experiment_session_repository.py new file mode 100644 index 0000000000..47dc8ecd34 --- /dev/null +++ b/services/intake/src/nmp/intake/spans/experiment_session_repository.py @@ -0,0 +1,378 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""ClickHouse repository for per-session rows of an Experiment. + +Returns one row per ingested session (test case execution), joining +``experiment_sessions`` with the session's root span (status + input text) +and per-session aggregates from all spans (tokens + cost), plus per-evaluator +session-mean scores from ``evaluator_results``. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from nmp.intake.spans.clickhouse_client import ClickHouseSpanClient +from nmp.intake.spans.domain import SpanStatus +from nmp.intake.spans.span_attribute_catalog import COST_SCALE, SpanAttributeField, spec_for_field +from nmp.intake.spans.storage import float_or_none, int_or_none, normalize_span_status, result_rows, str_or_none +from nmp.intake.spans.trace_repository import current_spans_sql + + +@dataclass(frozen=True) +class ExperimentSessionRow: + """One ingested session of an Experiment.""" + + workspace: str + experiment_name: str + session_id: str + test_case_id: str | None + trace_id: str + root_span_id: str + started_at: datetime + ended_at: datetime | None + latency_ms: float | None + status: SpanStatus + input: str | None + input_tokens: int | None + output_tokens: int | None + cached_tokens: int | None + cost_total_usd: float | None + evaluator_scores: dict[str, float] = field(default_factory=dict) + + +@dataclass(frozen=True) +class ExperimentSessionPage: + rows: list[ExperimentSessionRow] + total: int + + +class ExperimentSessionRepository: + def __init__(self, client: ClickHouseSpanClient) -> None: + self._client = client + + async def list_sessions( + self, + *, + workspace: str, + experiment_name: str, + status: SpanStatus | None = None, + test_case_id: str | None = None, + page: int, + page_size: int, + ) -> ExperimentSessionPage: + sessions_table = self._client.table("experiment_sessions") + spans_table = self._client.table("spans") + evaluator_results_table = self._client.table("evaluator_results") + + scoped_filter_sql, scoped_filter_parameters = _scoped_filter(test_case_id=test_case_id) + status_filter_sql, status_filter_parameters = _status_filter(status=status) + + base_parameters: dict[str, Any] = { + "workspace": workspace, + "experiment_name": experiment_name, + "input_tokens_key": spec_for_field(SpanAttributeField.INPUT_TOKENS).bag_key, + "output_tokens_key": spec_for_field(SpanAttributeField.OUTPUT_TOKENS).bag_key, + "cached_tokens_key": spec_for_field(SpanAttributeField.CACHED_TOKENS).bag_key, + "cost_key": spec_for_field(SpanAttributeField.COST_TOTAL_USD).bag_key, + } + + count_sql = _count_sql( + sessions_table=sessions_table, + spans_table=spans_table, + scoped_filter_sql=scoped_filter_sql, + status_filter_sql=status_filter_sql, + include_root_status=status is not None, + ) + count_result = await self._client.query( + count_sql, + parameters={**base_parameters, **scoped_filter_parameters, **status_filter_parameters}, + ) + total = int(count_result.result_rows[0][0]) if count_result.result_rows else 0 + if total == 0: + return ExperimentSessionPage(rows=[], total=0) + + offset = (page - 1) * page_size + list_sql = _list_sql( + sessions_table=sessions_table, + spans_table=spans_table, + evaluator_results_table=evaluator_results_table, + scoped_filter_sql=scoped_filter_sql, + status_filter_sql=status_filter_sql, + ) + list_result = await self._client.query( + list_sql, + parameters={ + **base_parameters, + **scoped_filter_parameters, + **status_filter_parameters, + "limit": page_size, + "offset": offset, + }, + ) + rows = [_row(record) for record in result_rows(list_result)] + return ExperimentSessionPage(rows=rows, total=total) + + +def _scoped_filter(*, test_case_id: str | None) -> tuple[str, dict[str, Any]]: + parameters: dict[str, Any] = {} + if test_case_id is not None: + parameters["test_case_id"] = test_case_id + return "AND test_case_id = %(test_case_id)s", parameters + return "", parameters + + +def _status_filter(*, status: SpanStatus | None) -> tuple[str | None, dict[str, Any]]: + if status is not None: + return "root_span_status = %(status)s", {"status": status.value} + return None, {} + + +def _scoped_sessions_sql(sessions_table: str, *, scoped_filter_sql: str) -> str: + return f""" + SELECT + workspace, + experiment_id, + session_id, + test_case_id, + trace_id, + root_span_id, + start_time, + end_time, + latency_ms + FROM {sessions_table} FINAL + WHERE workspace = %(workspace)s + AND is_deleted = 0 + AND experiment_id = %(experiment_name)s + {scoped_filter_sql} + ORDER BY start_time ASC, root_span_id ASC + LIMIT 1 BY workspace, session_id, experiment_id + """ + + +def _current_root_spans_sql(spans_table: str) -> str: + return current_spans_sql( + spans_table, + extra_where_sql=( + "(span_versions.workspace, span_versions.session_id, span_versions.external_span_id) IN " + "(SELECT workspace, session_id, root_span_id FROM scoped_sessions)" + ), + ) + + +def _rooted_sessions_sql() -> str: + return """ + SELECT + sessions.workspace AS workspace, + sessions.experiment_id AS experiment_id, + sessions.session_id AS session_id, + sessions.test_case_id AS test_case_id, + sessions.trace_id AS trace_id, + sessions.root_span_id AS root_span_id, + sessions.start_time AS start_time, + sessions.end_time AS end_time, + sessions.latency_ms AS latency_ms, + coalesce(root.status, 'unknown') AS root_span_status, + root.input AS input + FROM scoped_sessions AS sessions + LEFT JOIN current_root_spans AS root + ON sessions.workspace = root.workspace + AND sessions.session_id = root.session_id + AND sessions.root_span_id = root.external_span_id + AND root.is_deleted = 0 + """ + + +def _count_sql( + *, + sessions_table: str, + spans_table: str, + scoped_filter_sql: str, + status_filter_sql: str, + include_root_status: bool, +) -> str: + scoped_sessions_sql = _scoped_sessions_sql(sessions_table, scoped_filter_sql=scoped_filter_sql) + if not include_root_status: + return f""" + SELECT count() + FROM ( + {scoped_sessions_sql} + ) AS scoped_sessions + """ + + return f""" + WITH + scoped_sessions AS ( + {scoped_sessions_sql} + ), + current_root_spans AS ( + {_current_root_spans_sql(spans_table)} + ), + rooted_sessions AS ( + {_rooted_sessions_sql()} + ) + SELECT count() + FROM rooted_sessions + {f"WHERE {status_filter_sql}" if status_filter_sql else ""} + """ + + +def _list_sql( + *, + sessions_table: str, + spans_table: str, + evaluator_results_table: str, + scoped_filter_sql: str, + status_filter_sql: str, +) -> str: + return f""" + WITH + scoped_sessions AS ( + {_scoped_sessions_sql(sessions_table, scoped_filter_sql=scoped_filter_sql)} + ), + current_root_spans AS ( + {_current_root_spans_sql(spans_table)} + ), + rooted_sessions AS ( + {_rooted_sessions_sql()} + ), + page_sessions AS ( + SELECT + workspace, + experiment_id, + session_id, + test_case_id, + trace_id, + root_span_id, + start_time, + end_time, + latency_ms, + root_span_status, + input + FROM rooted_sessions + {f"WHERE {status_filter_sql}" if status_filter_sql else ""} + ORDER BY start_time ASC, root_span_id ASC + LIMIT %(limit)s OFFSET %(offset)s + ), + current_page_spans AS ( + { + current_spans_sql( + spans_table, + extra_where_sql=( + "(span_versions.workspace, span_versions.session_id) IN " + "(SELECT workspace, session_id FROM page_sessions)" + ), + ) + } + ), + session_metrics AS ( + SELECT + sessions.workspace AS workspace, + sessions.session_id AS session_id, + {_guarded_sum_sql("input_tokens_key")} AS input_tokens, + {_guarded_sum_sql("output_tokens_key")} AS output_tokens, + {_guarded_sum_sql("cached_tokens_key")} AS cached_tokens, + {_guarded_sum_sql("cost_key", scale=COST_SCALE)} AS cost_total_usd + FROM page_sessions AS sessions + LEFT JOIN current_page_spans AS spans + ON sessions.workspace = spans.workspace + AND sessions.session_id = spans.session_id + AND spans.is_deleted = 0 + GROUP BY sessions.workspace, sessions.session_id + ), + session_scores AS ( + SELECT + workspace, + session_id, + mapFromArrays(groupArray(evaluator_name), groupArray(mean_score)) AS evaluator_scores + FROM ( + SELECT + results.workspace AS workspace, + results.session_id AS session_id, + results.name AS evaluator_name, + avg(results.value) AS mean_score + FROM ( + SELECT workspace, session_id, name, value + FROM {evaluator_results_table} FINAL + WHERE workspace = %(workspace)s + AND (workspace, session_id) IN ( + SELECT workspace, session_id + FROM page_sessions + ) + AND data_type IN ('NUMERIC', 'BOOLEAN') + AND value IS NOT NULL + ) AS results + GROUP BY results.workspace, results.session_id, results.name + ) + GROUP BY workspace, session_id + ) + SELECT + sessions.workspace AS workspace, + sessions.experiment_id AS experiment_id, + sessions.session_id AS session_id, + sessions.test_case_id AS test_case_id, + sessions.trace_id AS trace_id, + sessions.root_span_id AS root_span_id, + sessions.start_time AS start_time, + sessions.end_time AS end_time, + sessions.latency_ms AS latency_ms, + sessions.root_span_status AS root_span_status, + sessions.input AS input, + metrics.input_tokens AS input_tokens, + metrics.output_tokens AS output_tokens, + metrics.cached_tokens AS cached_tokens, + metrics.cost_total_usd AS cost_total_usd, + scores.evaluator_scores AS evaluator_scores + FROM page_sessions AS sessions + LEFT JOIN session_metrics AS metrics + ON sessions.workspace = metrics.workspace + AND sessions.session_id = metrics.session_id + LEFT JOIN session_scores AS scores + ON sessions.workspace = scores.workspace + AND sessions.session_id = scores.session_id + ORDER BY sessions.start_time ASC, sessions.root_span_id ASC + """ + + +def _guarded_sum_sql(parameter_name: str, *, scale: int = 1) -> str: + key = f"%({parameter_name})s" + sum_expr = f"sumIf(spans.attributes_number[{key}], has(mapKeys(spans.attributes_number), {key}))" + if scale != 1: + sum_expr = f"{sum_expr} / {scale}" + return f""" + if( + countIf(has(mapKeys(spans.attributes_number), {key})) = 0, + NULL, + {sum_expr} + ) + """ + + +def _row(record: dict[str, Any]) -> ExperimentSessionRow: + return ExperimentSessionRow( + workspace=record["workspace"], + experiment_name=record["experiment_id"], + session_id=record["session_id"], + test_case_id=str_or_none(record["test_case_id"]), + trace_id=record["trace_id"], + root_span_id=record["root_span_id"], + started_at=record["start_time"], + ended_at=record["end_time"], + latency_ms=float_or_none(record["latency_ms"]), + status=normalize_span_status(record["root_span_status"]), + input=str_or_none(record["input"]), + input_tokens=int_or_none(record["input_tokens"]), + output_tokens=int_or_none(record["output_tokens"]), + cached_tokens=int_or_none(record["cached_tokens"]), + cost_total_usd=float_or_none(record["cost_total_usd"]), + evaluator_scores=_score_map(record.get("evaluator_scores")), + ) + + +def _score_map(value: Any) -> dict[str, float]: + if value is None: + return {} + return {str(key): float(score) for key, score in dict(value).items()} diff --git a/services/intake/src/nmp/intake/spans/storage.py b/services/intake/src/nmp/intake/spans/storage.py index 473f6c67eb..d50e2b28b6 100644 --- a/services/intake/src/nmp/intake/spans/storage.py +++ b/services/intake/src/nmp/intake/spans/storage.py @@ -90,3 +90,22 @@ def dict_to_row(row: dict[str, Any], columns: Sequence[str]) -> list[Any]: def result_rows(result: Any) -> list[dict[str, Any]]: return [dict(zip(result.column_names, row, strict=True)) for row in result.result_rows] + + +def float_or_none(value: Any) -> float | None: + if value is None: + return None + return float(value) + + +def int_or_none(value: Any) -> int | None: + if value is None: + return None + return int(value) + + +def str_or_none(value: Any) -> str | None: + # ClickHouse String columns use "" as the null sentinel; treat it the same as SQL NULL. + if value is None or value == "": + return None + return str(value) diff --git a/services/intake/src/nmp/intake/spans/trace_repository.py b/services/intake/src/nmp/intake/spans/trace_repository.py index af226c9511..f5f4f41062 100644 --- a/services/intake/src/nmp/intake/spans/trace_repository.py +++ b/services/intake/src/nmp/intake/spans/trace_repository.py @@ -14,7 +14,7 @@ from nmp.intake.spans.span_attribute_bags import SpanAttributeBags from nmp.intake.spans.span_attribute_catalog import COST_SCALE, SpanAttributeField, spec_for_field, where_clause from nmp.intake.spans.span_semantic_attributes import SpanSemanticAttributes -from nmp.intake.spans.storage import make_pagination, normalize_span_status, result_rows +from nmp.intake.spans.storage import float_or_none, int_or_none, make_pagination, normalize_span_status, result_rows TRACE_SORT_COLUMNS = { "started_at": "started_at", @@ -476,17 +476,17 @@ def _row_to_trace(row: dict[str, Any]) -> IntakeTrace: duration_ms=_duration_ms(row["started_at"], ended_at), ingested_at=row["ingested_at"], status=normalize_span_status(row.get("status")), - input_tokens=_int_or_none(row.get("input_tokens")), - output_tokens=_int_or_none(row.get("output_tokens")), - cached_tokens=_int_or_none(row.get("cached_tokens")), - total_tokens=_int_or_none(row.get("total_tokens")), - cost_usd=_float_or_none(row.get("cost_usd")), - cost_input_usd=_float_or_none(row.get("cost_input_usd")), - cost_output_usd=_float_or_none(row.get("cost_output_usd")), + input_tokens=int_or_none(row.get("input_tokens")), + output_tokens=int_or_none(row.get("output_tokens")), + cached_tokens=int_or_none(row.get("cached_tokens")), + total_tokens=int_or_none(row.get("total_tokens")), + cost_usd=float_or_none(row.get("cost_usd")), + cost_input_usd=float_or_none(row.get("cost_input_usd")), + cost_output_usd=float_or_none(row.get("cost_output_usd")), models=_string_list_or_none(row.get("models")), providers=_string_list_or_none(row.get("providers")), - span_count=_int_or_none(row.get("span_count")), - error_count=_int_or_none(row.get("error_count")), + span_count=int_or_none(row.get("span_count")), + error_count=int_or_none(row.get("error_count")), ) @@ -524,18 +524,6 @@ def _none_if_zero_datetime(value: Any) -> datetime | None: return value -def _int_or_none(value: Any) -> int | None: - if value is None: - return None - return int(value) - - -def _float_or_none(value: Any) -> float | None: - if value is None: - return None - return float(value) - - def _string_list_or_none(value: Any) -> list[str] | None: if value is None: return None diff --git a/services/intake/tests/integration/spans/test_experiment_sessions.py b/services/intake/tests/integration/spans/test_experiment_sessions.py new file mode 100644 index 0000000000..8a11dc3449 --- /dev/null +++ b/services/intake/tests/integration/spans/test_experiment_sessions.py @@ -0,0 +1,261 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration tests for the per-session experiment endpoint.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +ATIF_INGEST = "/apis/intake/v2/workspaces/default/ingest/atif" +EXPERIMENTS = "/apis/intake/v2/workspaces/default/experiments" + + +def test_list_experiment_sessions_returns_joined_session_rows(client: TestClient) -> None: + experiment_name = "sessions-exp" + created = client.post( + EXPERIMENTS, + json={ + "name": experiment_name, + "agent_name": "sample-agent", + "agent_version": "1.0.0", + "dataset_name": "sessions-dataset", + "dataset_version": "v1", + }, + ) + assert created.status_code == 201, created.text + + started_at = datetime.now(timezone.utc).replace(microsecond=0) + seeds = [ + ("case-a", 1.0, 0.05, 1000, 100, 10, 0), + ("case-b", 0.5, 0.10, 2000, 200, 20, 5), + ("case-c", 0.0, 0.20, 3000, 300, 30, 10), + ] + for index, ( + test_case_id, + score, + cost_usd, + latency_ms, + prompt_tokens, + completion_tokens, + offset_seconds, + ) in enumerate(seeds): + response = client.post( + ATIF_INGEST, + json=_atif_body( + started_at=started_at, + experiment_name=experiment_name, + test_case_id=test_case_id, + score=score, + cost_usd=cost_usd, + latency_ms=latency_ms, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + offset_seconds=offset_seconds, + run_id=f"run-{index}", + ), + ) + assert response.status_code == 201, response.text + + listed = client.get(f"{EXPERIMENTS}/{experiment_name}/sessions") + assert listed.status_code == 200, listed.text + body = listed.json() + assert body["pagination"]["total_results"] == 3 + assert len(body["data"]) == 3 + + rows_by_case = {row["test_case_id"]: row for row in body["data"]} + assert set(rows_by_case) == {"case-a", "case-b", "case-c"} + + case_a = rows_by_case["case-a"] + assert case_a["experiment_name"] == experiment_name + assert case_a["session_id"] + assert case_a["trace_id"] + assert case_a["root_span_id"] + assert case_a["latency_ms"] == pytest.approx(1000.0) + assert case_a["input_tokens"] == 100 + assert case_a["output_tokens"] == 10 + assert case_a["cost_total_usd"] == pytest.approx(0.05) + assert case_a["evaluator_scores"] == {"harbor.verifier": pytest.approx(1.0)} + assert case_a["status"] in {"success", "unknown"} + + paged = client.get(f"{EXPERIMENTS}/{experiment_name}/sessions", params={"page": 2, "page_size": 1}) + assert paged.status_code == 200, paged.text + paged_body = paged.json() + assert paged_body["pagination"]["total_results"] == 3 + assert len(paged_body["data"]) == 1 + assert paged_body["data"][0]["test_case_id"] == "case-b" + assert paged_body["data"][0]["evaluator_scores"] == {"harbor.verifier": pytest.approx(0.5)} + + +def test_list_experiment_sessions_filter_by_test_case(client: TestClient) -> None: + experiment_name = "sessions-filter-exp" + created = client.post( + EXPERIMENTS, + json={ + "name": experiment_name, + "agent_name": "sample-agent", + "agent_version": "1.0.0", + "dataset_name": "sessions-dataset", + "dataset_version": "v1", + }, + ) + assert created.status_code == 201, created.text + + started_at = datetime.now(timezone.utc).replace(microsecond=0) + for index, test_case_id in enumerate(["alpha", "beta"]): + response = client.post( + ATIF_INGEST, + json=_atif_body( + started_at=started_at, + experiment_name=experiment_name, + test_case_id=test_case_id, + score=1.0, + cost_usd=0.01, + latency_ms=100, + prompt_tokens=50, + completion_tokens=5, + offset_seconds=index * 5, + run_id=f"run-{index}", + ), + ) + assert response.status_code == 201, response.text + + filtered = client.get( + f"{EXPERIMENTS}/{experiment_name}/sessions", + params={"filter[test_case_id]": "alpha"}, + ) + assert filtered.status_code == 200, filtered.text + body = filtered.json() + assert body["pagination"]["total_results"] == 1 + assert len(body["data"]) == 1 + assert body["data"][0]["test_case_id"] == "alpha" + + +def test_list_experiment_sessions_filter_by_status(client: TestClient) -> None: + # ATIF ingest doesn't expose explicit per-span status, so all seeded sessions land with the + # default root-span status. This test verifies the filter is wired through (no SQL break and + # mismatched filters return zero) rather than per-status seeding. + experiment_name = "sessions-status-exp" + created = client.post( + EXPERIMENTS, + json={ + "name": experiment_name, + "agent_name": "sample-agent", + "agent_version": "1.0.0", + "dataset_name": "sessions-dataset", + "dataset_version": "v1", + }, + ) + assert created.status_code == 201, created.text + + response = client.post( + ATIF_INGEST, + json=_atif_body( + started_at=datetime.now(timezone.utc).replace(microsecond=0), + experiment_name=experiment_name, + test_case_id="case-1", + score=1.0, + cost_usd=0.01, + latency_ms=100, + prompt_tokens=50, + completion_tokens=5, + offset_seconds=0, + run_id="run-1", + ), + ) + assert response.status_code == 201, response.text + + listed = client.get(f"{EXPERIMENTS}/{experiment_name}/sessions") + assert listed.status_code == 200, listed.text + seeded_status = listed.json()["data"][0]["status"] + + matching = client.get( + f"{EXPERIMENTS}/{experiment_name}/sessions", + params={"filter[status]": seeded_status}, + ) + assert matching.status_code == 200, matching.text + assert matching.json()["pagination"]["total_results"] == 1 + + other_status = "error" if seeded_status != "error" else "cancelled" + mismatched = client.get( + f"{EXPERIMENTS}/{experiment_name}/sessions", + params={"filter[status]": other_status}, + ) + assert mismatched.status_code == 200, mismatched.text + assert mismatched.json()["pagination"]["total_results"] == 0 + + +def test_list_experiment_sessions_returns_404_for_unknown_experiment(client: TestClient) -> None: + response = client.get(f"{EXPERIMENTS}/does-not-exist/sessions") + assert response.status_code == 404, response.text + + +def test_list_experiment_sessions_rejects_unknown_query_param(client: TestClient) -> None: + response = client.get( + f"{EXPERIMENTS}/does-not-exist/sessions", + params={"test_caseid": "case-1"}, + ) + assert response.status_code == 400, response.text + assert response.json()["detail"] == "Unsupported query parameter(s): test_caseid" + + +def _atif_body( + *, + started_at: datetime, + experiment_name: str, + test_case_id: str, + score: float, + cost_usd: float, + latency_ms: int, + prompt_tokens: int, + completion_tokens: int, + offset_seconds: int, + run_id: str, +) -> dict[str, Any]: + session_started_at = started_at + timedelta(seconds=offset_seconds) + finished_at = session_started_at + timedelta(milliseconds=latency_ms) + session_id = f"{experiment_name}-{run_id}-{test_case_id}" + return { + "schema_version": "ATIF-v1.7", + "session_id": session_id, + "experiment_context": { + "experiment_id": experiment_name, + "test_case_id": test_case_id, + }, + "extra": { + "task_id": test_case_id, + "task_name": test_case_id, + "verifier": { + "started_at": _iso(session_started_at), + "finished_at": _iso(finished_at), + }, + "verifier_result": {"rewards": {"reward": score}}, + }, + "agent": { + "name": "sample-agent", + "version": "1.0.0", + "model_name": "provider/sample-model", + }, + "steps": [ + { + "step_id": 1, + "timestamp": _iso(session_started_at), + "source": "agent", + "model_name": "provider/sample-model", + "message": f"solved {test_case_id}", + "metrics": { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "cost_usd": cost_usd, + }, + } + ], + } + + +def _iso(value: datetime) -> str: + return value.isoformat().replace("+00:00", "Z")