diff --git a/docs/cli/reference.mdx b/docs/cli/reference.mdx index a12d91fadd..106657bc3f 100644 --- a/docs/cli/reference.mdx +++ b/docs/cli/reference.mdx @@ -7115,9 +7115,50 @@ nemo intake traces [OPTIONS] COMMAND [ARGS]... **Commands:** +* `get-metrics`: Get Trace Metrics * `list`: List Traces * `get`: Get Trace +##### nemo intake traces get-metrics + +Get Trace Metrics + +**Usage:** + +```shell +nemo intake traces get-metrics [OPTIONS] +``` + +**Options:** + +* `--workspace` +* `--bucket `: Time bucket granularity. [possible values: total, hour, day, week, month] +* `--timezone`: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + +**Filter Options:** + +* `--filter FILTER_JSON`: Use --filter with JSON for complex/nested queries, or --filter. FIELD options for simple fields. Both can be combined, with field options taking precedence. +JSON-only fields: + started_at: \{gte: str, lte: str} + +Filter the traces the metrics are computed over. Accepts the same fields as the traces list, so agent_name scopes the rollup to one agent. Without a started_at lower bound the rollup covers the last 7 days. +* `--filter.id` +* `--filter.agent-name` +* `--filter.evaluation-id` +* `--filter.evaluation-name` +* `--filter.session-id` +* `--filter.status` +* `--filter.test-case-id` +* `--filter.test-case-name` + +**Help:** + +* `--help, -h`: Show this message and exit. + +**Output Options:** + +* `--output-format, --output, -f `: Output format for an entity. [possible values: json, yaml, raw, code] + ##### nemo intake traces list List Traces @@ -7143,8 +7184,9 @@ nemo intake traces list [OPTIONS] JSON-only fields: started_at: \{gte: str, lte: str} -Filter root-span-backed traces by id, session_id, root status, root span started_at, evaluation_name, and test_case_name. +Filter root-span-backed traces by id, session_id, root status, root span started_at, evaluation_name, test_case_name, and agent_name. * `--filter.id` +* `--filter.agent-name` * `--filter.evaluation-id` * `--filter.evaluation-name` * `--filter.session-id` diff --git a/openapi/ga/individual/platform.openapi.yaml b/openapi/ga/individual/platform.openapi.yaml index 1abfc4f38a..863de164a3 100644 --- a/openapi/ga/individual/platform.openapi.yaml +++ b/openapi/ga/individual/platform.openapi.yaml @@ -4930,7 +4930,7 @@ paths: schema: $ref: '#/components/schemas/TraceFilter' description: Filter root-span-backed traces by id, session_id, root status, - root span started_at, evaluation_name, and test_case_name. + root span started_at, evaluation_name, test_case_name, and agent_name. responses: '200': description: Successful Response @@ -4944,6 +4944,84 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/traces/metrics: + get: + tags: + - Traces + summary: Get Trace Metrics + operationId: get_trace_metrics_apis_intake_v2_workspaces__workspace__traces_metrics_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: bucket + in: query + required: false + schema: + allOf: + - $ref: '#/components/schemas/TraceMetricBucketParam' + description: Time bucket granularity. total collapses the filtered range + into a single row. + default: day + description: Time bucket granularity. total collapses the filtered range into + a single row. + - name: timezone + in: query + required: false + schema: + type: string + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + default: UTC + title: Timezone + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/TraceFilter' + description: Filter the traces the metrics are computed over. Accepts the + same fields as the traces list, so agent_name scopes the rollup to one agent. + Without a started_at lower bound the rollup covers the last 7 days. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/TraceMetrics' + '400': + description: Unsupported query parameter or unknown timezone + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '503': + description: ClickHouse spans storage unavailable + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/traces/{id}: get: tags: @@ -9843,6 +9921,22 @@ components: type: object title: ContextBloatDetectionConfig description: Configuration for context bloat / context manipulation detection. + CostRollupResponse: + properties: + sum: + title: Sum + type: number + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: CostRollupResponse CreateAdapterRequest: properties: name: @@ -14176,6 +14270,25 @@ components: labels must include a `name` to identify what the score measures.' + LatencyRollupResponse: + properties: + mean: + title: Mean + type: number + p50: + title: P50 + type: number + p90: + title: P90 + type: number + p95: + title: P95 + type: number + p99: + title: P99 + type: number + type: object + title: LatencyRollupResponse LinearLayerSpec: properties: name: @@ -18815,6 +18928,23 @@ components: - task title: TaskPrompt description: Configuration for prompts that will be used for a specific task. + TokenRollupResponse: + properties: + sum: + title: Sum + type: integer + minimum: 0.0 + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: TokenRollupResponse ToolCallConfig: properties: tool_call_parser: @@ -18936,6 +19066,12 @@ components: type: string evaluation_context: $ref: '#/components/schemas/EvaluationContext' + agent_name: + title: Agent Name + type: string + agent_version: + title: Agent Version + type: string started_at: type: string format: date-time @@ -18974,6 +19110,20 @@ components: cost_output_usd: title: Cost Output Usd type: number + models: + title: Models + description: Distinct models used across the trace's spans. Omitted in summary + mode. + items: + type: string + type: array + providers: + title: Providers + description: Distinct inference providers used across the trace's spans. + Omitted in summary mode. + items: + type: string + type: array span_count: title: Span Count type: integer @@ -19026,8 +19176,80 @@ components: description: Deprecated alias for test_case_name. Use test_case_name instead. title: Test Case Id type: string + agent_name: + description: Filter by root-span agent name. + title: Agent Name + type: string title: TraceFilter type: object + TraceMetricBucketParam: + type: string + enum: + - total + - hour + - day + - week + - month + title: TraceMetricBucketParam + TraceMetricPointResponse: + properties: + bucket_start: + title: Bucket Start + description: Start of the bucket in the requested timezone. Omitted when + bucket=total. + type: string + format: date-time + run_count: + type: integer + minimum: 0.0 + title: Run Count + description: Agent runs started in this bucket. + failed_run_count: + type: integer + minimum: 0.0 + title: Failed Run Count + description: Runs whose root span ended in error. + input_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + output_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cached_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + total_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cost_usd: + $ref: '#/components/schemas/CostRollupResponse' + latency_ms: + $ref: '#/components/schemas/LatencyRollupResponse' + type: object + required: + - run_count + - failed_run_count + - input_tokens + - output_tokens + - cached_tokens + - total_tokens + - cost_usd + - latency_ms + title: TraceMetricPointResponse + TraceMetrics: + properties: + bucket: + $ref: '#/components/schemas/TraceMetricBucketParam' + timezone: + type: string + title: Timezone + data: + items: + $ref: '#/components/schemas/TraceMetricPointResponse' + type: array + title: Data + type: object + required: + - bucket + - timezone + - data + title: TraceMetrics TraceSortField: type: string enum: diff --git a/openapi/ga/openapi.yaml b/openapi/ga/openapi.yaml index 1abfc4f38a..863de164a3 100644 --- a/openapi/ga/openapi.yaml +++ b/openapi/ga/openapi.yaml @@ -4930,7 +4930,7 @@ paths: schema: $ref: '#/components/schemas/TraceFilter' description: Filter root-span-backed traces by id, session_id, root status, - root span started_at, evaluation_name, and test_case_name. + root span started_at, evaluation_name, test_case_name, and agent_name. responses: '200': description: Successful Response @@ -4944,6 +4944,84 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/traces/metrics: + get: + tags: + - Traces + summary: Get Trace Metrics + operationId: get_trace_metrics_apis_intake_v2_workspaces__workspace__traces_metrics_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: bucket + in: query + required: false + schema: + allOf: + - $ref: '#/components/schemas/TraceMetricBucketParam' + description: Time bucket granularity. total collapses the filtered range + into a single row. + default: day + description: Time bucket granularity. total collapses the filtered range into + a single row. + - name: timezone + in: query + required: false + schema: + type: string + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + default: UTC + title: Timezone + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/TraceFilter' + description: Filter the traces the metrics are computed over. Accepts the + same fields as the traces list, so agent_name scopes the rollup to one agent. + Without a started_at lower bound the rollup covers the last 7 days. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/TraceMetrics' + '400': + description: Unsupported query parameter or unknown timezone + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '503': + description: ClickHouse spans storage unavailable + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/traces/{id}: get: tags: @@ -9843,6 +9921,22 @@ components: type: object title: ContextBloatDetectionConfig description: Configuration for context bloat / context manipulation detection. + CostRollupResponse: + properties: + sum: + title: Sum + type: number + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: CostRollupResponse CreateAdapterRequest: properties: name: @@ -14176,6 +14270,25 @@ components: labels must include a `name` to identify what the score measures.' + LatencyRollupResponse: + properties: + mean: + title: Mean + type: number + p50: + title: P50 + type: number + p90: + title: P90 + type: number + p95: + title: P95 + type: number + p99: + title: P99 + type: number + type: object + title: LatencyRollupResponse LinearLayerSpec: properties: name: @@ -18815,6 +18928,23 @@ components: - task title: TaskPrompt description: Configuration for prompts that will be used for a specific task. + TokenRollupResponse: + properties: + sum: + title: Sum + type: integer + minimum: 0.0 + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: TokenRollupResponse ToolCallConfig: properties: tool_call_parser: @@ -18936,6 +19066,12 @@ components: type: string evaluation_context: $ref: '#/components/schemas/EvaluationContext' + agent_name: + title: Agent Name + type: string + agent_version: + title: Agent Version + type: string started_at: type: string format: date-time @@ -18974,6 +19110,20 @@ components: cost_output_usd: title: Cost Output Usd type: number + models: + title: Models + description: Distinct models used across the trace's spans. Omitted in summary + mode. + items: + type: string + type: array + providers: + title: Providers + description: Distinct inference providers used across the trace's spans. + Omitted in summary mode. + items: + type: string + type: array span_count: title: Span Count type: integer @@ -19026,8 +19176,80 @@ components: description: Deprecated alias for test_case_name. Use test_case_name instead. title: Test Case Id type: string + agent_name: + description: Filter by root-span agent name. + title: Agent Name + type: string title: TraceFilter type: object + TraceMetricBucketParam: + type: string + enum: + - total + - hour + - day + - week + - month + title: TraceMetricBucketParam + TraceMetricPointResponse: + properties: + bucket_start: + title: Bucket Start + description: Start of the bucket in the requested timezone. Omitted when + bucket=total. + type: string + format: date-time + run_count: + type: integer + minimum: 0.0 + title: Run Count + description: Agent runs started in this bucket. + failed_run_count: + type: integer + minimum: 0.0 + title: Failed Run Count + description: Runs whose root span ended in error. + input_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + output_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cached_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + total_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cost_usd: + $ref: '#/components/schemas/CostRollupResponse' + latency_ms: + $ref: '#/components/schemas/LatencyRollupResponse' + type: object + required: + - run_count + - failed_run_count + - input_tokens + - output_tokens + - cached_tokens + - total_tokens + - cost_usd + - latency_ms + title: TraceMetricPointResponse + TraceMetrics: + properties: + bucket: + $ref: '#/components/schemas/TraceMetricBucketParam' + timezone: + type: string + title: Timezone + data: + items: + $ref: '#/components/schemas/TraceMetricPointResponse' + type: array + title: Data + type: object + required: + - bucket + - timezone + - data + title: TraceMetrics TraceSortField: type: string enum: diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index 1abfc4f38a..863de164a3 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -4930,7 +4930,7 @@ paths: schema: $ref: '#/components/schemas/TraceFilter' description: Filter root-span-backed traces by id, session_id, root status, - root span started_at, evaluation_name, and test_case_name. + root span started_at, evaluation_name, test_case_name, and agent_name. responses: '200': description: Successful Response @@ -4944,6 +4944,84 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/traces/metrics: + get: + tags: + - Traces + summary: Get Trace Metrics + operationId: get_trace_metrics_apis_intake_v2_workspaces__workspace__traces_metrics_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: bucket + in: query + required: false + schema: + allOf: + - $ref: '#/components/schemas/TraceMetricBucketParam' + description: Time bucket granularity. total collapses the filtered range + into a single row. + default: day + description: Time bucket granularity. total collapses the filtered range into + a single row. + - name: timezone + in: query + required: false + schema: + type: string + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + default: UTC + title: Timezone + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/TraceFilter' + description: Filter the traces the metrics are computed over. Accepts the + same fields as the traces list, so agent_name scopes the rollup to one agent. + Without a started_at lower bound the rollup covers the last 7 days. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/TraceMetrics' + '400': + description: Unsupported query parameter or unknown timezone + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '503': + description: ClickHouse spans storage unavailable + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/traces/{id}: get: tags: @@ -9843,6 +9921,22 @@ components: type: object title: ContextBloatDetectionConfig description: Configuration for context bloat / context manipulation detection. + CostRollupResponse: + properties: + sum: + title: Sum + type: number + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: CostRollupResponse CreateAdapterRequest: properties: name: @@ -14176,6 +14270,25 @@ components: labels must include a `name` to identify what the score measures.' + LatencyRollupResponse: + properties: + mean: + title: Mean + type: number + p50: + title: P50 + type: number + p90: + title: P90 + type: number + p95: + title: P95 + type: number + p99: + title: P99 + type: number + type: object + title: LatencyRollupResponse LinearLayerSpec: properties: name: @@ -18815,6 +18928,23 @@ components: - task title: TaskPrompt description: Configuration for prompts that will be used for a specific task. + TokenRollupResponse: + properties: + sum: + title: Sum + type: integer + minimum: 0.0 + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: TokenRollupResponse ToolCallConfig: properties: tool_call_parser: @@ -18936,6 +19066,12 @@ components: type: string evaluation_context: $ref: '#/components/schemas/EvaluationContext' + agent_name: + title: Agent Name + type: string + agent_version: + title: Agent Version + type: string started_at: type: string format: date-time @@ -18974,6 +19110,20 @@ components: cost_output_usd: title: Cost Output Usd type: number + models: + title: Models + description: Distinct models used across the trace's spans. Omitted in summary + mode. + items: + type: string + type: array + providers: + title: Providers + description: Distinct inference providers used across the trace's spans. + Omitted in summary mode. + items: + type: string + type: array span_count: title: Span Count type: integer @@ -19026,8 +19176,80 @@ components: description: Deprecated alias for test_case_name. Use test_case_name instead. title: Test Case Id type: string + agent_name: + description: Filter by root-span agent name. + title: Agent Name + type: string title: TraceFilter type: object + TraceMetricBucketParam: + type: string + enum: + - total + - hour + - day + - week + - month + title: TraceMetricBucketParam + TraceMetricPointResponse: + properties: + bucket_start: + title: Bucket Start + description: Start of the bucket in the requested timezone. Omitted when + bucket=total. + type: string + format: date-time + run_count: + type: integer + minimum: 0.0 + title: Run Count + description: Agent runs started in this bucket. + failed_run_count: + type: integer + minimum: 0.0 + title: Failed Run Count + description: Runs whose root span ended in error. + input_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + output_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cached_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + total_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cost_usd: + $ref: '#/components/schemas/CostRollupResponse' + latency_ms: + $ref: '#/components/schemas/LatencyRollupResponse' + type: object + required: + - run_count + - failed_run_count + - input_tokens + - output_tokens + - cached_tokens + - total_tokens + - cost_usd + - latency_ms + title: TraceMetricPointResponse + TraceMetrics: + properties: + bucket: + $ref: '#/components/schemas/TraceMetricBucketParam' + timezone: + type: string + title: Timezone + data: + items: + $ref: '#/components/schemas/TraceMetricPointResponse' + type: array + title: Data + type: object + required: + - bucket + - timezone + - data + title: TraceMetrics TraceSortField: type: string enum: diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/intake/traces.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/intake/traces.py index fc249ff838..43aae9a122 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/intake/traces.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/intake/traces.py @@ -31,6 +31,86 @@ app = create_typer_app(name="traces", help="Manage traces") +@app.command("get-metrics") +@collect_warnings +@handle_errors +def get_metrics_traces( + ctx: typer.Context, + workspace: Annotated[str | None, typer.Option("--workspace")] = None, + bucket: Annotated[ + Literal["total", "hour", "day", "week", "month"] | None, + typer.Option("--bucket", help="Time bucket granularity."), + ] = None, + filter: Annotated[ + str | None, + typer.Option( + "--filter", + metavar="FILTER_JSON", + help="Use --filter with JSON for complex/nested queries, or --filter.FIELD options for simple fields. Both can be combined, with field options taking precedence.\nJSON-only fields:\n started_at: {gte: str, lte: str}\n\nFilter the traces the metrics are computed over. Accepts the same fields as the traces list, so agent_name scopes the rollup to one agent. Without a started_at lower bound the rollup covers the last 7 days.", + rich_help_panel="Filter Options", + ), + ] = None, + filter_id: Annotated[str | None, typer.Option("--filter.id", rich_help_panel="Filter Options")] = None, + filter_agent_name: Annotated[ + str | None, typer.Option("--filter.agent-name", rich_help_panel="Filter Options") + ] = None, + filter_evaluation_id: Annotated[ + str | None, typer.Option("--filter.evaluation-id", rich_help_panel="Filter Options") + ] = None, + filter_evaluation_name: Annotated[ + str | None, typer.Option("--filter.evaluation-name", rich_help_panel="Filter Options") + ] = None, + filter_session_id: Annotated[ + str | None, typer.Option("--filter.session-id", rich_help_panel="Filter Options") + ] = None, + filter_status: Annotated[str | None, typer.Option("--filter.status", rich_help_panel="Filter Options")] = None, + filter_test_case_id: Annotated[ + str | None, typer.Option("--filter.test-case-id", rich_help_panel="Filter Options") + ] = None, + filter_test_case_name: Annotated[ + str | None, typer.Option("--filter.test-case-name", rich_help_panel="Filter Options") + ] = None, + timezone: Annotated[ + str | None, + typer.Option("--timezone", help="IANA timezone the buckets are aligned to, e.g. America/Los_Angeles."), + ] = None, + output_format: EntityOutputFormatOption = None, +) -> None: + """Get Trace Metrics""" + state: CLIContext = ctx.obj + output_format = state.get_output_format(output_format) + + kwargs = build_kwargs( + workspace=workspace, + bucket=bucket, + filter=merge_filter_dict( + filter, + id=filter_id, + agent_name=filter_agent_name, + evaluation_id=filter_evaluation_id, + evaluation_name=filter_evaluation_name, + session_id=filter_session_id, + status=filter_status, + test_case_id=filter_test_case_id, + test_case_name=filter_test_case_name, + ), + timezone=timezone, + ) + if handle_code_generation(["intake", "traces"], "get_metrics", kwargs, output_format, state): + return + + client = state.get_client() + result = client.intake.traces.get_metrics(**kwargs) + + format_output( + result, + is_list=False, + output_format=output_format, + no_truncate=state.get_no_truncate(), + timestamp_format=state.get_timestamp_format(), + ) + + @app.command("list") @collect_warnings @handle_errors @@ -42,11 +122,14 @@ def list_traces( typer.Option( "--filter", metavar="FILTER_JSON", - help="Use --filter with JSON for complex/nested queries, or --filter.FIELD options for simple fields. Both can be combined, with field options taking precedence.\nJSON-only fields:\n started_at: {gte: str, lte: str}\n\nFilter root-span-backed traces by id, session_id, root status, root span started_at, evaluation_name, and test_case_name.", + help="Use --filter with JSON for complex/nested queries, or --filter.FIELD options for simple fields. Both can be combined, with field options taking precedence.\nJSON-only fields:\n started_at: {gte: str, lte: str}\n\nFilter root-span-backed traces by id, session_id, root status, root span started_at, evaluation_name, test_case_name, and agent_name.", rich_help_panel="Filter Options", ), ] = None, filter_id: Annotated[str | None, typer.Option("--filter.id", rich_help_panel="Filter Options")] = None, + filter_agent_name: Annotated[ + str | None, typer.Option("--filter.agent-name", rich_help_panel="Filter Options") + ] = None, filter_evaluation_id: Annotated[ str | None, typer.Option("--filter.evaluation-id", rich_help_panel="Filter Options") ] = None, @@ -99,6 +182,7 @@ def list_traces( filter=merge_filter_dict( filter, id=filter_id, + agent_name=filter_agent_name, evaluation_id=filter_evaluation_id, evaluation_name=filter_evaluation_name, session_id=filter_session_id, diff --git a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml index d85678a56d..473903a76d 100644 --- a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml +++ b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml @@ -4933,7 +4933,7 @@ paths: schema: $ref: '#/components/schemas/TraceFilter' description: Filter root-span-backed traces by id, session_id, root status, - root span started_at, evaluation_name, and test_case_name. + root span started_at, evaluation_name, test_case_name, and agent_name. responses: '200': description: Successful Response @@ -4947,6 +4947,84 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /apis/intake/v2/workspaces/{workspace}/traces/metrics: + get: + tags: + - Traces + summary: Get Trace Metrics + operationId: get_trace_metrics_apis_intake_v2_workspaces__workspace__traces_metrics_get + parameters: + - name: workspace + in: path + required: true + schema: + type: string + title: Workspace + - name: bucket + in: query + required: false + schema: + allOf: + - $ref: '#/components/schemas/TraceMetricBucketParam' + description: Time bucket granularity. total collapses the filtered range + into a single row. + default: day + description: Time bucket granularity. total collapses the filtered range into + a single row. + - name: timezone + in: query + required: false + schema: + type: string + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + default: UTC + title: Timezone + description: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + - in: query + name: filter + style: deepObject + required: false + explode: true + schema: + $ref: '#/components/schemas/TraceFilter' + description: Filter the traces the metrics are computed over. Accepts the + same fields as the traces list, so agent_name scopes the rollup to one agent. + Without a started_at lower bound the rollup covers the last 7 days. + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/TraceMetrics' + '400': + description: Unsupported query parameter or unknown timezone + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '503': + description: ClickHouse spans storage unavailable + content: + application/json: + schema: + type: object + properties: + detail: + type: string + required: + - detail + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /apis/intake/v2/workspaces/{workspace}/traces/{id}: get: tags: @@ -9846,6 +9924,22 @@ components: type: object title: ContextBloatDetectionConfig description: Configuration for context bloat / context manipulation detection. + CostRollupResponse: + properties: + sum: + title: Sum + type: number + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: CostRollupResponse CreateAdapterRequest: properties: name: @@ -14179,6 +14273,25 @@ components: labels must include a `name` to identify what the score measures.' + LatencyRollupResponse: + properties: + mean: + title: Mean + type: number + p50: + title: P50 + type: number + p90: + title: P90 + type: number + p95: + title: P95 + type: number + p99: + title: P99 + type: number + type: object + title: LatencyRollupResponse LinearLayerSpec: properties: name: @@ -18818,6 +18931,23 @@ components: - task title: TaskPrompt description: Configuration for prompts that will be used for a specific task. + TokenRollupResponse: + properties: + sum: + title: Sum + type: integer + minimum: 0.0 + mean: + title: Mean + type: number + p90: + title: P90 + type: number + p99: + title: P99 + type: number + type: object + title: TokenRollupResponse ToolCallConfig: properties: tool_call_parser: @@ -18939,6 +19069,12 @@ components: type: string evaluation_context: $ref: '#/components/schemas/EvaluationContext' + agent_name: + title: Agent Name + type: string + agent_version: + title: Agent Version + type: string started_at: type: string format: date-time @@ -18977,6 +19113,20 @@ components: cost_output_usd: title: Cost Output Usd type: number + models: + title: Models + description: Distinct models used across the trace's spans. Omitted in summary + mode. + items: + type: string + type: array + providers: + title: Providers + description: Distinct inference providers used across the trace's spans. + Omitted in summary mode. + items: + type: string + type: array span_count: title: Span Count type: integer @@ -19029,8 +19179,80 @@ components: description: Deprecated alias for test_case_name. Use test_case_name instead. title: Test Case Id type: string + agent_name: + description: Filter by root-span agent name. + title: Agent Name + type: string title: TraceFilter type: object + TraceMetricBucketParam: + type: string + enum: + - total + - hour + - day + - week + - month + title: TraceMetricBucketParam + TraceMetricPointResponse: + properties: + bucket_start: + title: Bucket Start + description: Start of the bucket in the requested timezone. Omitted when + bucket=total. + type: string + format: date-time + run_count: + type: integer + minimum: 0.0 + title: Run Count + description: Agent runs started in this bucket. + failed_run_count: + type: integer + minimum: 0.0 + title: Failed Run Count + description: Runs whose root span ended in error. + input_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + output_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cached_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + total_tokens: + $ref: '#/components/schemas/TokenRollupResponse' + cost_usd: + $ref: '#/components/schemas/CostRollupResponse' + latency_ms: + $ref: '#/components/schemas/LatencyRollupResponse' + type: object + required: + - run_count + - failed_run_count + - input_tokens + - output_tokens + - cached_tokens + - total_tokens + - cost_usd + - latency_ms + title: TraceMetricPointResponse + TraceMetrics: + properties: + bucket: + $ref: '#/components/schemas/TraceMetricBucketParam' + timezone: + type: string + title: Timezone + data: + items: + $ref: '#/components/schemas/TraceMetricPointResponse' + type: array + title: Data + type: object + required: + - bucket + - timezone + - data + title: TraceMetrics TraceSortField: type: string enum: diff --git a/sdk/python/nemo-platform/.nmpcontext/stainless.yaml b/sdk/python/nemo-platform/.nmpcontext/stainless.yaml index 33b9e81ff2..9473a9c275 100644 --- a/sdk/python/nemo-platform/.nmpcontext/stainless.yaml +++ b/sdk/python/nemo-platform/.nmpcontext/stainless.yaml @@ -904,13 +904,20 @@ resources: retrieve: get /apis/intake/v2/workspaces/{workspace}/sessions/{id} traces: models: + cost_rollup_response: CostRollupResponse + latency_rollup_response: LatencyRollupResponse + token_rollup_response: TokenRollupResponse trace: Trace trace_filter: TraceFilter + trace_metric_bucket_param: TraceMetricBucketParam + trace_metric_point_response: TraceMetricPointResponse + trace_metrics: TraceMetrics trace_sort_field: TraceSortField traces_page: TracesPage methods: list: get /apis/intake/v2/workspaces/{workspace}/traces retrieve: get /apis/intake/v2/workspaces/{workspace}/traces/{id} + get_metrics: get /apis/intake/v2/workspaces/{workspace}/traces/metrics evaluations: standalone_api: true models: diff --git a/sdk/python/nemo-platform/src/nemo_platform/cli/commands/api/intake/traces.py b/sdk/python/nemo-platform/src/nemo_platform/cli/commands/api/intake/traces.py index c7a0c949f3..7790a27313 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/cli/commands/api/intake/traces.py +++ b/sdk/python/nemo-platform/src/nemo_platform/cli/commands/api/intake/traces.py @@ -31,6 +31,86 @@ app = create_typer_app(name="traces", help="Manage traces") +@app.command("get-metrics") +@collect_warnings +@handle_errors +def get_metrics_traces( + ctx: typer.Context, + workspace: Annotated[str | None, typer.Option("--workspace")] = None, + bucket: Annotated[ + Literal["total", "hour", "day", "week", "month"] | None, + typer.Option("--bucket", help="Time bucket granularity."), + ] = None, + filter: Annotated[ + str | None, + typer.Option( + "--filter", + metavar="FILTER_JSON", + help="Use --filter with JSON for complex/nested queries, or --filter.FIELD options for simple fields. Both can be combined, with field options taking precedence.\nJSON-only fields:\n started_at: {gte: str, lte: str}\n\nFilter the traces the metrics are computed over. Accepts the same fields as the traces list, so agent_name scopes the rollup to one agent. Without a started_at lower bound the rollup covers the last 7 days.", + rich_help_panel="Filter Options", + ), + ] = None, + filter_id: Annotated[str | None, typer.Option("--filter.id", rich_help_panel="Filter Options")] = None, + filter_agent_name: Annotated[ + str | None, typer.Option("--filter.agent-name", rich_help_panel="Filter Options") + ] = None, + filter_evaluation_id: Annotated[ + str | None, typer.Option("--filter.evaluation-id", rich_help_panel="Filter Options") + ] = None, + filter_evaluation_name: Annotated[ + str | None, typer.Option("--filter.evaluation-name", rich_help_panel="Filter Options") + ] = None, + filter_session_id: Annotated[ + str | None, typer.Option("--filter.session-id", rich_help_panel="Filter Options") + ] = None, + filter_status: Annotated[str | None, typer.Option("--filter.status", rich_help_panel="Filter Options")] = None, + filter_test_case_id: Annotated[ + str | None, typer.Option("--filter.test-case-id", rich_help_panel="Filter Options") + ] = None, + filter_test_case_name: Annotated[ + str | None, typer.Option("--filter.test-case-name", rich_help_panel="Filter Options") + ] = None, + timezone: Annotated[ + str | None, + typer.Option("--timezone", help="IANA timezone the buckets are aligned to, e.g. America/Los_Angeles."), + ] = None, + output_format: EntityOutputFormatOption = None, +) -> None: + """Get Trace Metrics""" + state: CLIContext = ctx.obj + output_format = state.get_output_format(output_format) + + kwargs = build_kwargs( + workspace=workspace, + bucket=bucket, + filter=merge_filter_dict( + filter, + id=filter_id, + agent_name=filter_agent_name, + evaluation_id=filter_evaluation_id, + evaluation_name=filter_evaluation_name, + session_id=filter_session_id, + status=filter_status, + test_case_id=filter_test_case_id, + test_case_name=filter_test_case_name, + ), + timezone=timezone, + ) + if handle_code_generation(["intake", "traces"], "get_metrics", kwargs, output_format, state): + return + + client = state.get_client() + result = client.intake.traces.get_metrics(**kwargs) + + format_output( + result, + is_list=False, + output_format=output_format, + no_truncate=state.get_no_truncate(), + timestamp_format=state.get_timestamp_format(), + ) + + @app.command("list") @collect_warnings @handle_errors @@ -42,11 +122,14 @@ def list_traces( typer.Option( "--filter", metavar="FILTER_JSON", - help="Use --filter with JSON for complex/nested queries, or --filter.FIELD options for simple fields. Both can be combined, with field options taking precedence.\nJSON-only fields:\n started_at: {gte: str, lte: str}\n\nFilter root-span-backed traces by id, session_id, root status, root span started_at, evaluation_name, and test_case_name.", + help="Use --filter with JSON for complex/nested queries, or --filter.FIELD options for simple fields. Both can be combined, with field options taking precedence.\nJSON-only fields:\n started_at: {gte: str, lte: str}\n\nFilter root-span-backed traces by id, session_id, root status, root span started_at, evaluation_name, test_case_name, and agent_name.", rich_help_panel="Filter Options", ), ] = None, filter_id: Annotated[str | None, typer.Option("--filter.id", rich_help_panel="Filter Options")] = None, + filter_agent_name: Annotated[ + str | None, typer.Option("--filter.agent-name", rich_help_panel="Filter Options") + ] = None, filter_evaluation_id: Annotated[ str | None, typer.Option("--filter.evaluation-id", rich_help_panel="Filter Options") ] = None, @@ -99,6 +182,7 @@ def list_traces( filter=merge_filter_dict( filter, id=filter_id, + agent_name=filter_agent_name, evaluation_id=filter_evaluation_id, evaluation_name=filter_evaluation_name, session_id=filter_session_id, diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/intake/api.md b/sdk/python/nemo-platform/src/nemo_platform/resources/intake/api.md index 6b6dea9a23..adbe2353ea 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/resources/intake/api.md +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/intake/api.md @@ -208,10 +208,22 @@ Methods: Types: ```python -from nemo_platform.types.intake import Trace, TraceFilter, TraceSortField, TracesPage +from nemo_platform.types.intake import ( + CostRollupResponse, + LatencyRollupResponse, + TokenRollupResponse, + Trace, + TraceFilter, + TraceMetricBucketParam, + TraceMetricPointResponse, + TraceMetrics, + TraceSortField, + TracesPage, +) ``` Methods: - client.intake.traces.retrieve(id, \*, workspace, \*\*params) -> Trace - client.intake.traces.list(\*, workspace, \*\*params) -> SyncDefaultPagination[Trace] +- client.intake.traces.get_metrics(\*, workspace, \*\*params) -> TraceMetrics diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/intake/traces.py b/sdk/python/nemo-platform/src/nemo_platform/resources/intake/traces.py index 44de226f6a..0631f8c3c8 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/resources/intake/traces.py +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/intake/traces.py @@ -33,10 +33,18 @@ ) from ...pagination import SyncDefaultPagination, AsyncDefaultPagination from ..._base_client import AsyncPaginator, make_request_options -from ...types.intake import TraceSortField, trace_list_params, trace_retrieve_params +from ...types.intake import ( + TraceSortField, + TraceMetricBucketParam, + trace_list_params, + trace_retrieve_params, + trace_get_metrics_params, +) from ...types.intake.trace import Trace +from ...types.intake.trace_metrics import TraceMetrics from ...types.intake.trace_sort_field import TraceSortField from ...types.intake.trace_filter_param import TraceFilterParam +from ...types.intake.trace_metric_bucket_param import TraceMetricBucketParam __all__ = ["TracesResource", "AsyncTracesResource"] @@ -130,7 +138,7 @@ def list( Args: filter: Filter root-span-backed traces by id, session_id, root status, root span - started_at, evaluation_name, and test_case_name. + started_at, evaluation_name, test_case_name, and agent_name. mode: Response mode. summary returns root-span fields without payloads or rollups; preview adds token, cost, and span-count rollups plus 300-character input/output @@ -174,6 +182,64 @@ def list( model=Trace, ) + def get_metrics( + self, + *, + workspace: str | None = None, + bucket: TraceMetricBucketParam | Omit = omit, + filter: TraceFilterParam | Omit = omit, + timezone: str | 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, + ) -> TraceMetrics: + """Get Trace Metrics + + Args: + bucket: Time bucket granularity. + + total collapses the filtered range into a single row. + + filter: Filter the traces the metrics are computed over. Accepts the same fields as the + traces list, so agent_name scopes the rollup to one agent. Without a started_at + lower bound the rollup covers the last 7 days. + + timezone: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + + 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}") + return self._get( + path_template("/apis/intake/v2/workspaces/{workspace}/traces/metrics", workspace=workspace), + options=make_request_options( + extra_headers=extra_headers, + extra_query=extra_query, + extra_body=extra_body, + timeout=timeout, + query=maybe_transform( + { + "bucket": bucket, + "filter": filter, + "timezone": timezone, + }, + trace_get_metrics_params.TraceGetMetricsParams, + ), + ), + cast_to=TraceMetrics, + ) + class AsyncTracesResource(AsyncAPIResource): @cached_property @@ -264,7 +330,7 @@ def list( Args: filter: Filter root-span-backed traces by id, session_id, root status, root span - started_at, evaluation_name, and test_case_name. + started_at, evaluation_name, test_case_name, and agent_name. mode: Response mode. summary returns root-span fields without payloads or rollups; preview adds token, cost, and span-count rollups plus 300-character input/output @@ -308,6 +374,64 @@ def list( model=Trace, ) + async def get_metrics( + self, + *, + workspace: str | None = None, + bucket: TraceMetricBucketParam | Omit = omit, + filter: TraceFilterParam | Omit = omit, + timezone: str | 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, + ) -> TraceMetrics: + """Get Trace Metrics + + Args: + bucket: Time bucket granularity. + + total collapses the filtered range into a single row. + + filter: Filter the traces the metrics are computed over. Accepts the same fields as the + traces list, so agent_name scopes the rollup to one agent. Without a started_at + lower bound the rollup covers the last 7 days. + + timezone: IANA timezone the buckets are aligned to, e.g. America/Los_Angeles. + + 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}") + return await self._get( + path_template("/apis/intake/v2/workspaces/{workspace}/traces/metrics", workspace=workspace), + options=make_request_options( + extra_headers=extra_headers, + extra_query=extra_query, + extra_body=extra_body, + timeout=timeout, + query=await async_maybe_transform( + { + "bucket": bucket, + "filter": filter, + "timezone": timezone, + }, + trace_get_metrics_params.TraceGetMetricsParams, + ), + ), + cast_to=TraceMetrics, + ) + class TracesResourceWithRawResponse: def __init__(self, traces: TracesResource) -> None: @@ -319,6 +443,9 @@ def __init__(self, traces: TracesResource) -> None: self.list = to_raw_response_wrapper( traces.list, ) + self.get_metrics = to_raw_response_wrapper( + traces.get_metrics, + ) class AsyncTracesResourceWithRawResponse: @@ -331,6 +458,9 @@ def __init__(self, traces: AsyncTracesResource) -> None: self.list = async_to_raw_response_wrapper( traces.list, ) + self.get_metrics = async_to_raw_response_wrapper( + traces.get_metrics, + ) class TracesResourceWithStreamingResponse: @@ -343,6 +473,9 @@ def __init__(self, traces: TracesResource) -> None: self.list = to_streamed_response_wrapper( traces.list, ) + self.get_metrics = to_streamed_response_wrapper( + traces.get_metrics, + ) class AsyncTracesResourceWithStreamingResponse: @@ -355,3 +488,6 @@ def __init__(self, traces: AsyncTracesResource) -> None: self.list = async_to_streamed_response_wrapper( traces.list, ) + self.get_metrics = async_to_streamed_response_wrapper( + traces.get_metrics, + ) diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/__init__.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/__init__.py index 1b2906c26b..d168d51f9c 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/intake/__init__.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/__init__.py @@ -25,6 +25,7 @@ from .spans_page import SpansPage as SpansPage from .span_status import SpanStatus as SpanStatus from .traces_page import TracesPage as TracesPage +from .trace_metrics import TraceMetrics as TraceMetrics from .annotation_kind import AnnotationKind as AnnotationKind from .note_annotation import NoteAnnotation as NoteAnnotation from .span_sort_field import SpanSortField as SpanSortField @@ -41,21 +42,27 @@ from .trace_filter_param import TraceFilterParam as TraceFilterParam from .feedback_annotation import FeedbackAnnotation as FeedbackAnnotation from .metadata_annotation import MetadataAnnotation as MetadataAnnotation +from .cost_rollup_response import CostRollupResponse as CostRollupResponse from .numeric_filter_param import NumericFilterParam as NumericFilterParam from .annotation_sort_field import AnnotationSortField as AnnotationSortField from .note_annotation_param import NoteAnnotationParam as NoteAnnotationParam +from .token_rollup_response import TokenRollupResponse as TokenRollupResponse from .trace_retrieve_params import TraceRetrieveParams as TraceRetrieveParams from .annotation_list_params import AnnotationListParams as AnnotationListParams from .evaluator_results_page import EvaluatorResultsPage as EvaluatorResultsPage from .label_annotation_param import LabelAnnotationParam as LabelAnnotationParam from .annotation_filter_param import AnnotationFilterParam as AnnotationFilterParam +from .latency_rollup_response import LatencyRollupResponse as LatencyRollupResponse from .span_evaluation_context import SpanEvaluationContext as SpanEvaluationContext from .annotation_create_params import AnnotationCreateParams as AnnotationCreateParams from .evaluation_context_param import EvaluationContextParam as EvaluationContextParam +from .trace_get_metrics_params import TraceGetMetricsParams as TraceGetMetricsParams from .feedback_annotation_param import FeedbackAnnotationParam as FeedbackAnnotationParam from .metadata_annotation_param import MetadataAnnotationParam as MetadataAnnotationParam +from .trace_metric_bucket_param import TraceMetricBucketParam as TraceMetricBucketParam from .evaluator_result_data_type import EvaluatorResultDataType as EvaluatorResultDataType from .evaluator_result_sort_field import EvaluatorResultSortField as EvaluatorResultSortField +from .trace_metric_point_response import TraceMetricPointResponse as TraceMetricPointResponse from .evaluator_result_list_params import EvaluatorResultListParams as EvaluatorResultListParams from .evaluator_result_filter_param import EvaluatorResultFilterParam as EvaluatorResultFilterParam from .evaluator_result_create_params import EvaluatorResultCreateParams as EvaluatorResultCreateParams diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/cost_rollup_response.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/cost_rollup_response.py new file mode 100644 index 0000000000..eb850988d6 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/cost_rollup_response.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 typing import Optional + +from ..._models import BaseModel + +__all__ = ["CostRollupResponse"] + + +class CostRollupResponse(BaseModel): + mean: Optional[float] = None + + p90: Optional[float] = None + + p99: Optional[float] = None + + sum: Optional[float] = None diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/latency_rollup_response.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/latency_rollup_response.py new file mode 100644 index 0000000000..ccf0ab9b41 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/latency_rollup_response.py @@ -0,0 +1,34 @@ +# 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 Optional + +from ..._models import BaseModel + +__all__ = ["LatencyRollupResponse"] + + +class LatencyRollupResponse(BaseModel): + mean: Optional[float] = None + + p50: Optional[float] = None + + p90: Optional[float] = None + + p95: Optional[float] = None + + p99: Optional[float] = None diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/token_rollup_response.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/token_rollup_response.py new file mode 100644 index 0000000000..6a9d451634 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/token_rollup_response.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 typing import Optional + +from ..._models import BaseModel + +__all__ = ["TokenRollupResponse"] + + +class TokenRollupResponse(BaseModel): + mean: Optional[float] = None + + p90: Optional[float] = None + + p99: Optional[float] = None + + sum: Optional[int] = None diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace.py index d8bfddf159..b3308ad383 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace.py @@ -15,7 +15,7 @@ # File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. -from typing import Optional +from typing import List, Optional from datetime import datetime from ..._models import BaseModel @@ -36,6 +36,10 @@ class Trace(BaseModel): workspace: str + agent_name: Optional[str] = None + + agent_version: Optional[str] = None + cached_tokens: Optional[int] = None cost_input_usd: Optional[float] = None @@ -64,6 +68,9 @@ class Trace(BaseModel): input_tokens: Optional[int] = None + models: Optional[List[str]] = None + """Distinct models used across the trace's spans. Omitted in summary mode.""" + name: Optional[str] = None output: Optional[str] = None @@ -74,6 +81,12 @@ class Trace(BaseModel): output_tokens: Optional[int] = None + providers: Optional[List[str]] = None + """Distinct inference providers used across the trace's spans. + + Omitted in summary mode. + """ + root_span_id: Optional[str] = None span_count: Optional[int] = None diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_filter_param.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_filter_param.py index c86287ca59..b053f79b98 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_filter_param.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_filter_param.py @@ -29,6 +29,9 @@ class TraceFilterParam(TypedDict, total=False): id: str """Filter by canonical Intake trace id.""" + agent_name: str + """Filter by root-span agent name.""" + evaluation_id: str """Deprecated alias for evaluation_name. Use evaluation_name instead.""" diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_get_metrics_params.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_get_metrics_params.py new file mode 100644 index 0000000000..31a70e7486 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_get_metrics_params.py @@ -0,0 +1,42 @@ +# 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 .trace_filter_param import TraceFilterParam +from .trace_metric_bucket_param import TraceMetricBucketParam + +__all__ = ["TraceGetMetricsParams"] + + +class TraceGetMetricsParams(TypedDict, total=False): + workspace: str + + bucket: TraceMetricBucketParam + """Time bucket granularity. total collapses the filtered range into a single row.""" + + filter: TraceFilterParam + """Filter the traces the metrics are computed over. + + Accepts the same fields as the traces list, so agent_name scopes the rollup to + one agent. Without a started_at lower bound the rollup covers the last 7 days. + """ + + timezone: str + """IANA timezone the buckets are aligned to, e.g. America/Los_Angeles.""" diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_list_params.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_list_params.py index 1f84cc7536..30c442bfc9 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_list_params.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_list_params.py @@ -31,7 +31,7 @@ class TraceListParams(TypedDict, total=False): filter: TraceFilterParam """ Filter root-span-backed traces by id, session_id, root status, root span - started_at, evaluation_name, and test_case_name. + started_at, evaluation_name, test_case_name, and agent_name. """ mode: Literal["summary", "preview", "detailed"] diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metric_bucket_param.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metric_bucket_param.py new file mode 100644 index 0000000000..c7077c11c9 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metric_bucket_param.py @@ -0,0 +1,22 @@ +# 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_extensions import Literal, TypeAlias + +__all__ = ["TraceMetricBucketParam"] + +TraceMetricBucketParam: TypeAlias = Literal["total", "hour", "day", "week", "month"] diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metric_point_response.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metric_point_response.py new file mode 100644 index 0000000000..6efa0677ec --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metric_point_response.py @@ -0,0 +1,49 @@ +# 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 Optional +from datetime import datetime + +from ..._models import BaseModel +from .cost_rollup_response import CostRollupResponse +from .token_rollup_response import TokenRollupResponse +from .latency_rollup_response import LatencyRollupResponse + +__all__ = ["TraceMetricPointResponse"] + + +class TraceMetricPointResponse(BaseModel): + cached_tokens: TokenRollupResponse + + cost_usd: CostRollupResponse + + failed_run_count: int + """Runs whose root span ended in error.""" + + input_tokens: TokenRollupResponse + + latency_ms: LatencyRollupResponse + + output_tokens: TokenRollupResponse + + run_count: int + """Agent runs started in this bucket.""" + + total_tokens: TokenRollupResponse + + bucket_start: Optional[datetime] = None + """Start of the bucket in the requested timezone. Omitted when bucket=total.""" diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metrics.py b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metrics.py new file mode 100644 index 0000000000..bc9d350358 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/types/intake/trace_metrics.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 typing import List + +from ..._models import BaseModel +from .trace_metric_bucket_param import TraceMetricBucketParam +from .trace_metric_point_response import TraceMetricPointResponse + +__all__ = ["TraceMetrics"] + + +class TraceMetrics(BaseModel): + bucket: TraceMetricBucketParam + + data: List[TraceMetricPointResponse] + + timezone: str diff --git a/sdk/python/nemo-platform/tests/api_resources/intake/test_traces.py b/sdk/python/nemo-platform/tests/api_resources/intake/test_traces.py index fdeb46f10d..e80f683bc6 100644 --- a/sdk/python/nemo-platform/tests/api_resources/intake/test_traces.py +++ b/sdk/python/nemo-platform/tests/api_resources/intake/test_traces.py @@ -26,7 +26,10 @@ from nemo_platform import NeMoPlatform, AsyncNeMoPlatform from nemo_platform._utils import parse_datetime from nemo_platform.pagination import SyncDefaultPagination, AsyncDefaultPagination -from nemo_platform.types.intake import Trace +from nemo_platform.types.intake import ( + Trace, + TraceMetrics, +) base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") @@ -111,6 +114,7 @@ def test_method_list_with_all_params(self, client: NeMoPlatform) -> None: workspace="workspace", filter={ "id": "id", + "agent_name": "agent_name", "evaluation_id": "evaluation_id", "evaluation_name": "evaluation_name", "session_id": "session_id", @@ -163,6 +167,72 @@ def test_path_params_list(self, client: NeMoPlatform) -> None: workspace="", ) + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_method_get_metrics(self, client: NeMoPlatform) -> None: + trace = client.intake.traces.get_metrics( + workspace="workspace", + ) + assert_matches_type(TraceMetrics, trace, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_method_get_metrics_with_all_params(self, client: NeMoPlatform) -> None: + trace = client.intake.traces.get_metrics( + workspace="workspace", + bucket="total", + filter={ + "id": "id", + "agent_name": "agent_name", + "evaluation_id": "evaluation_id", + "evaluation_name": "evaluation_name", + "session_id": "session_id", + "started_at": { + "gte": parse_datetime("2019-12-27T18:11:19.117Z"), + "lte": parse_datetime("2019-12-27T18:11:19.117Z"), + }, + "status": "success", + "test_case_id": "test_case_id", + "test_case_name": "test_case_name", + }, + timezone="timezone", + ) + assert_matches_type(TraceMetrics, trace, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_raw_response_get_metrics(self, client: NeMoPlatform) -> None: + response = client.intake.traces.with_raw_response.get_metrics( + workspace="workspace", + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + trace = response.parse() + assert_matches_type(TraceMetrics, trace, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_streaming_response_get_metrics(self, client: NeMoPlatform) -> None: + with client.intake.traces.with_streaming_response.get_metrics( + workspace="workspace", + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + trace = response.parse() + assert_matches_type(TraceMetrics, trace, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + def test_path_params_get_metrics(self, client: NeMoPlatform) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `workspace` but received ''"): + client.intake.traces.with_raw_response.get_metrics( + workspace="", + ) + class TestAsyncTraces: parametrize = pytest.mark.parametrize( @@ -246,6 +316,7 @@ async def test_method_list_with_all_params(self, async_client: AsyncNeMoPlatform workspace="workspace", filter={ "id": "id", + "agent_name": "agent_name", "evaluation_id": "evaluation_id", "evaluation_name": "evaluation_name", "session_id": "session_id", @@ -297,3 +368,69 @@ async def test_path_params_list(self, async_client: AsyncNeMoPlatform) -> None: await async_client.intake.traces.with_raw_response.list( workspace="", ) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_method_get_metrics(self, async_client: AsyncNeMoPlatform) -> None: + trace = await async_client.intake.traces.get_metrics( + workspace="workspace", + ) + assert_matches_type(TraceMetrics, trace, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_method_get_metrics_with_all_params(self, async_client: AsyncNeMoPlatform) -> None: + trace = await async_client.intake.traces.get_metrics( + workspace="workspace", + bucket="total", + filter={ + "id": "id", + "agent_name": "agent_name", + "evaluation_id": "evaluation_id", + "evaluation_name": "evaluation_name", + "session_id": "session_id", + "started_at": { + "gte": parse_datetime("2019-12-27T18:11:19.117Z"), + "lte": parse_datetime("2019-12-27T18:11:19.117Z"), + }, + "status": "success", + "test_case_id": "test_case_id", + "test_case_name": "test_case_name", + }, + timezone="timezone", + ) + assert_matches_type(TraceMetrics, trace, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_raw_response_get_metrics(self, async_client: AsyncNeMoPlatform) -> None: + response = await async_client.intake.traces.with_raw_response.get_metrics( + workspace="workspace", + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + trace = await response.parse() + assert_matches_type(TraceMetrics, trace, path=["response"]) + + @pytest.mark.skip(reason="Mock server tests are disabled") + @parametrize + async def test_streaming_response_get_metrics(self, async_client: AsyncNeMoPlatform) -> None: + async with async_client.intake.traces.with_streaming_response.get_metrics( + workspace="workspace", + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + trace = await response.parse() + assert_matches_type(TraceMetrics, trace, 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_get_metrics(self, async_client: AsyncNeMoPlatform) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `workspace` but received ''"): + await async_client.intake.traces.with_raw_response.get_metrics( + workspace="", + ) diff --git a/sdk/stainless.yaml b/sdk/stainless.yaml index 33b9e81ff2..9473a9c275 100644 --- a/sdk/stainless.yaml +++ b/sdk/stainless.yaml @@ -904,13 +904,20 @@ resources: retrieve: get /apis/intake/v2/workspaces/{workspace}/sessions/{id} traces: models: + cost_rollup_response: CostRollupResponse + latency_rollup_response: LatencyRollupResponse + token_rollup_response: TokenRollupResponse trace: Trace trace_filter: TraceFilter + trace_metric_bucket_param: TraceMetricBucketParam + trace_metric_point_response: TraceMetricPointResponse + trace_metrics: TraceMetrics trace_sort_field: TraceSortField traces_page: TracesPage methods: list: get /apis/intake/v2/workspaces/{workspace}/traces retrieve: get /apis/intake/v2/workspaces/{workspace}/traces/{id} + get_metrics: get /apis/intake/v2/workspaces/{workspace}/traces/metrics evaluations: standalone_api: true models: 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 217f79b36c..6fb4f4fa19 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 @@ -160,6 +160,9 @@ authz: description: "List intake spans" read: description: "Read intake spans" + trace-metrics: + read: + description: "Read time-bucketed intake trace metrics" traces: read: description: "Read intake traces" @@ -315,6 +318,7 @@ authz: - intake.sessions.read - intake.spans.list - intake.spans.read + - intake.trace-metrics.read - intake.traces.read - jobs.list - jobs.read @@ -1128,6 +1132,13 @@ authz: scopes: - intake:read - platform:read + /apis/intake/v2/workspaces/{workspace}/traces/metrics: + get: + permissions: + - intake.trace-metrics.read + scopes: + - intake:read + - platform:read /apis/intake/v2/workspaces/{workspace}/traces: get: permissions: diff --git a/services/intake/src/nmp/intake/repository/clickhouse/trace.py b/services/intake/src/nmp/intake/repository/clickhouse/trace.py index 6f0ee626d1..fd681deb59 100644 --- a/services/intake/src/nmp/intake/repository/clickhouse/trace.py +++ b/services/intake/src/nmp/intake/repository/clickhouse/trace.py @@ -10,13 +10,23 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import datetime, timezone +from math import isfinite from typing import Any from nmp.common.api.common import PaginatedResult from nmp.intake.repository.clickhouse.executor import ClickHouseExecutor, ClickHouseExternalData, ClickHouseQuery from nmp.intake.repository.clickhouse.tables import ClickHouseTable from nmp.intake.repository.trace import TraceRepository -from nmp.intake.spans.domain import IntakeTrace, TraceListFilter, TraceMode +from nmp.intake.spans.domain import ( + CostRollup, + IntakeTrace, + LatencyRollup, + TokenRollup, + TraceListFilter, + TraceMetricBucket, + TraceMetricPoint, + TraceMode, +) from nmp.intake.spans.span_attribute_catalog import SpanAttributeField, spec_for_field from nmp.intake.spans.span_rollups import METRIC_ATTRIBUTE_FIELDS, metric_aggregate_columns from nmp.intake.spans.storage import ( @@ -86,6 +96,7 @@ class _TracePageRef: source_format: str trace_id: str + session_id: str started_at_us: int @classmethod @@ -93,6 +104,7 @@ def from_row(cls, row: Mapping[str, Any]) -> _TracePageRef: return cls( source_format=str(row["source_format"]), trace_id=str(row["id"]), + session_id=str(row["session_id"]), started_at_us=int(row["started_at_us"]), ) @@ -200,6 +212,60 @@ async def get_trace(self, *, workspace: str, trace_id: str, mode: TraceMode) -> ) return result.data[0] if result.data else None + async def trace_metrics( + self, + *, + filters: TraceListFilter, + bucket: TraceMetricBucket, + timezone_name: str, + ) -> list[TraceMetricPoint]: + trace_index_table = self._executor.table(ClickHouseTable.TRACE_INDEX) + spans_table = self._executor.table(ClickHouseTable.SPANS) + roots_sql, parameters = _metric_roots_sql( + trace_index_table=trace_index_table, + filters=filters, + bucket=bucket, + ) + # Reuses the same per-trace span rollup as trace hydration, rescoped from a + # page of trace refs to whatever the filter selected. + rollups_sql, rollup_parameters = _trace_aggregates_sql( + spans_table, + extra_where_sql=( + # session_id engages the spans primary key, trace_id drives the bloom + # filter, and the tuple keeps a trace_id shared across source formats + # from pulling in unrelated spans. + "span_versions.session_id IN (SELECT session_id FROM roots)\n" + " AND span_versions.trace_id IN (SELECT trace_id FROM roots)\n" + " AND (span_versions.source_format, span_versions.trace_id) " + "IN (SELECT source_format, trace_id FROM roots)" + ), + ) + parameters.update(rollup_parameters) + parameters["metrics_timezone"] = timezone_name + + statement = f""" + WITH + roots AS ( + {roots_sql} + ), + rollups AS ( + {rollups_sql} + ) + SELECT + {_metric_select_columns()} + FROM roots + LEFT JOIN rollups + ON roots.workspace = rollups.workspace + AND roots.source_format = rollups.source_format + AND roots.trace_id = rollups.trace_id + GROUP BY bucket_start + ORDER BY bucket_start ASC + """ + rows = await self._executor.fetch_all( + ClickHouseQuery(name="traces.metrics", statement=statement, parameters=parameters) + ) + return [_row_to_metric_point(row, bucket=bucket) for row in rows] + async def latest_trace_started_at_by_group( self, *, @@ -338,6 +404,8 @@ def _trace_select_columns(*, include_aggregates: bool) -> str: "traces.project AS project", "traces.evaluation_name AS evaluation_name", "traces.test_case_name AS test_case_name", + "traces.agent_name AS agent_name", + "traces.agent_version AS agent_version", "traces.started_at AS started_at", "traces.ended_at AS ended_at", "traces.status AS status", @@ -383,6 +451,8 @@ def _trace_index_select_columns(*, mode: TraceMode) -> tuple[str, dict[str, Any] "nullIf(trace_roots.project, '') AS project", "nullIf(trace_roots.evaluation_name, '') AS evaluation_name", "nullIf(trace_roots.test_case_name, '') AS test_case_name", + "nullIf(trace_roots.agent_name, '') AS agent_name", + "nullIf(trace_roots.agent_version, '') AS agent_version", "trace_roots.root_started_at AS started_at", "trace_roots.root_ended_at AS ended_at", "trace_roots.root_status AS status", @@ -391,17 +461,32 @@ def _trace_index_select_columns(*, mode: TraceMode) -> tuple[str, dict[str, Any] return ",\n ".join(columns), text_query_parameters(mode) -def _trace_aggregates_sql(table: str) -> tuple[str, dict[str, Any]]: +# session_id leads the spans sort key after workspace, so restricting it lets the +# primary key prune before the trace-id bloom filter widens the granule set. +_PAGE_TRACE_REFS_WHERE_SQL = ( + "span_versions.session_id IN %(page_session_ids)s\n" + " AND span_versions.trace_id IN %(page_trace_ids)s\n" + " AND (span_versions.source_format, span_versions.trace_id) " + "IN %(page_trace_keys)s" +) + + +def _trace_aggregates_sql( + table: str, + *, + extra_where_sql: str = _PAGE_TRACE_REFS_WHERE_SQL, + extra_select_sql: str = "", +) -> tuple[str, dict[str, Any]]: + """Build the per-trace span rollup shared by trace hydration and metric buckets. + + Both callers must read through ``current_spans_sql``: spans is a + ReplacingMergeTree, so summing raw rows would double-count re-ingested spans. + """ source_alias = "trace_spans" select_columns, parameters = _trace_aggregate_select_columns(source_alias) - current_spans = current_spans_sql( - table, - extra_where_sql=( - "span_versions.trace_id IN %(page_trace_ids)s\n" - " AND (span_versions.source_format, span_versions.trace_id) " - "IN %(page_trace_keys)s" - ), - ) + if extra_select_sql: + select_columns = f"{select_columns},\n {extra_select_sql}" + current_spans = current_spans_sql(table, extra_where_sql=extra_where_sql) query = f""" SELECT {select_columns} @@ -445,6 +530,7 @@ def _trace_page_parameters(refs: Sequence[_TracePageRef]) -> dict[str, object]: return { "page_trace_ids": [ref.trace_id for ref in refs], "page_trace_keys": [ref.trace_key for ref in refs], + "page_session_ids": sorted({ref.session_id for ref in refs}), "page_started_at_min_us": min(started_at_us), "page_started_at_max_us": max(started_at_us), } @@ -464,6 +550,7 @@ def _reconcile_hydrated_page( _TRACE_INDEX_FILTER_COLUMNS = { "evaluation_name": "evaluation_name", "test_case_name": "test_case_name", + "agent_name": "agent_name", } @@ -509,16 +596,18 @@ def current_spans_sql( *, extra_where_sql: str | None = None, ) -> str: + """Read current span versions without materialising wide aggregate states. + + ``spans`` is a ReplacingMergeTree, so ``FINAL`` applies its latest-write-wins + semantics. Callers should scope by ``session_id`` when possible because it + follows ``workspace`` in the table's sorting key. + """ source_alias = "span_versions" columns = [ - *[f"{source_alias}.{column} AS {column}" for column in _CURRENT_SPAN_IDENTITY_COLUMNS], - *[ - f"argMax({source_alias}.{column}, ({source_alias}.event_ts, {source_alias}.is_deleted)) AS {column}" - for column in _CURRENT_SPAN_VALUE_COLUMNS - ], + f"{source_alias}.{column} AS {column}" + for column in (*_CURRENT_SPAN_IDENTITY_COLUMNS, *_CURRENT_SPAN_VALUE_COLUMNS) ] columns_sql = ",\n ".join(columns) - group_by_sql = ", ".join(f"{source_alias}.{column}" for column in _CURRENT_SPAN_IDENTITY_COLUMNS) where_sql = f"{source_alias}.workspace = %(workspace)s" if extra_where_sql is not None: where_sql = f"{where_sql}\n AND {extra_where_sql}" @@ -526,9 +615,8 @@ def current_spans_sql( ( SELECT {columns_sql} - FROM {table} AS {source_alias} + FROM {table} AS {source_alias} FINAL WHERE {where_sql} - GROUP BY {group_by_sql} ) """ @@ -561,6 +649,8 @@ def _row_to_trace(row: dict[str, Any]) -> IntakeTrace: project=row.get("project") or None, evaluation_name=row.get("evaluation_name") or None, test_case_name=row.get("test_case_name") or None, + agent_name=row.get("agent_name") or None, + agent_version=row.get("agent_version") or None, started_at=row["started_at"], ended_at=ended_at, duration_ms=_duration_ms(row["started_at"], ended_at), @@ -580,6 +670,134 @@ def _row_to_trace(row: dict[str, Any]) -> IntakeTrace: ) +# ClickHouse resolves these against the caller's timezone so buckets line up with the +# user's calendar rather than the server's. Week starts Monday (mode 1). +# toStartOfWeek/Month return Date, toStartOfDay returns DateTime. Cast so every bucket +# yields the same type; otherwise some buckets deserialize to naive datetimes and others +# to timezone-aware ones, and chart clients see inconsistent offsets. +_METRIC_BUCKET_EXPRESSIONS = { + "hour": "toStartOfHour(trace_roots.root_started_at, %(metrics_timezone)s)", + "day": "toStartOfDay(trace_roots.root_started_at, %(metrics_timezone)s)", + "week": "toDateTime(toStartOfWeek(trace_roots.root_started_at, 1, %(metrics_timezone)s), %(metrics_timezone)s)", + "month": "toDateTime(toStartOfMonth(trace_roots.root_started_at, %(metrics_timezone)s), %(metrics_timezone)s)", +} + + +def _metric_roots_sql( + *, + trace_index_table: str, + filters: TraceListFilter, + bucket: TraceMetricBucket, +) -> tuple[str, dict[str, Any]]: + """Build the deduplicated root-span CTE the metric buckets group over.""" + + where_sql, parameters = _trace_index_where(filters, qualifier="trace_roots") + query = f""" + SELECT + trace_roots.workspace AS workspace, + trace_roots.source_format AS source_format, + trace_roots.trace_id AS trace_id, + trace_roots.session_id AS session_id, + {_metric_bucket_expression(bucket)} AS bucket_start, + trace_roots.root_status AS root_status, + trace_roots.latency_ms AS latency_ms + FROM {trace_index_table} AS trace_roots FINAL + WHERE {where_sql} + ORDER BY trace_roots.root_started_at ASC, trace_roots.root_span_id ASC + LIMIT 1 BY trace_roots.workspace, trace_roots.source_format, trace_roots.trace_id + """ + return query, parameters + + +_TOKEN_ROLLUP_FIELDS = ("input_tokens", "output_tokens", "cached_tokens", "total_tokens") +_SUMMED_ROLLUP_QUANTILES = (0.9, 0.99) +_LATENCY_QUANTILES = (0.5, 0.9, 0.95, 0.99) + + +def _quantiles_expression(expression: str, quantiles: Sequence[float]) -> str: + # One combined aggregate rather than a quantile state per percentile. + return f"quantiles({', '.join(str(quantile) for quantile in quantiles)})({expression})" + + +def _metric_select_columns() -> str: + columns = [ + "roots.bucket_start AS bucket_start", + "count() AS run_count", + "countIf(roots.root_status = 'error') AS failed_run_count", + ] + for field in (*_TOKEN_ROLLUP_FIELDS, "cost_usd"): + source = f"rollups.{field}" + columns.extend( + ( + f"sum({source}) AS {field}_sum", + f"avg({source}) AS {field}_mean", + f"{_quantiles_expression(source, _SUMMED_ROLLUP_QUANTILES)} AS {field}_quantiles", + ) + ) + columns.extend( + ( + # Percentiles cannot yield a mean, and the design needs one; it is also what + # makes an aggregate latency-per-token ratio derivable client-side. + "avg(roots.latency_ms) AS latency_ms_mean", + f"{_quantiles_expression('roots.latency_ms', _LATENCY_QUANTILES)} AS latency_ms_quantiles", + ) + ) + return ",\n ".join(columns) + + +def _metric_bucket_expression(bucket: TraceMetricBucket) -> str: + if bucket == "total": + # A constant collapses the filtered range into one row; the value is discarded. + return "toDateTime(0)" + try: + return _METRIC_BUCKET_EXPRESSIONS[bucket] + except KeyError: + raise ValueError(f"Unsupported trace metric bucket: {bucket}") from None + + +def _row_to_metric_point(row: dict[str, Any], *, bucket: TraceMetricBucket) -> TraceMetricPoint: + return TraceMetricPoint( + bucket_start=None if bucket == "total" else row["bucket_start"], + run_count=int(row["run_count"]), + # Counts failed *runs* (root status), unlike the span-level error_count on the + # trace rollups, which counts failed spans within one trace. + failed_run_count=int(row["failed_run_count"]), + **{field: _token_rollup(row, field) for field in _TOKEN_ROLLUP_FIELDS}, + cost_usd=CostRollup( + sum=float_or_none(row.get("cost_usd_sum")), + mean=_finite_or_none(row.get("cost_usd_mean")), + **_named_quantiles(row, "cost_usd", _SUMMED_ROLLUP_QUANTILES), + ), + latency_ms=LatencyRollup( + mean=_finite_or_none(row.get("latency_ms_mean")), + **_named_quantiles(row, "latency_ms", _LATENCY_QUANTILES), + ), + ) + + +def _token_rollup(row: dict[str, Any], field: str) -> TokenRollup: + return TokenRollup( + sum=int_or_none(row.get(f"{field}_sum")), + mean=_finite_or_none(row.get(f"{field}_mean")), + **_named_quantiles(row, field, _SUMMED_ROLLUP_QUANTILES), + ) + + +def _named_quantiles(row: dict[str, Any], field: str, quantiles: Sequence[float]) -> dict[str, float | None]: + values: Sequence[Any] = row.get(f"{field}_quantiles") or () + return { + f"p{round(quantile * 100)}": _finite_or_none(values[index]) if index < len(values) else None + for index, quantile in enumerate(quantiles) + } + + +def _finite_or_none(value: Any) -> float | None: + """Drop the NaN ClickHouse yields for an aggregate over an empty bucket.""" + + number = float_or_none(value) + return None if number is None or not isfinite(number) else number + + def _duration_ms(started_at: datetime, ended_at: datetime | None) -> float | None: if ended_at is None: return None diff --git a/services/intake/src/nmp/intake/repository/trace.py b/services/intake/src/nmp/intake/repository/trace.py index 93556c21c5..2934232b0e 100644 --- a/services/intake/src/nmp/intake/repository/trace.py +++ b/services/intake/src/nmp/intake/repository/trace.py @@ -7,7 +7,13 @@ from datetime import datetime from nmp.common.api.common import PaginatedResult -from nmp.intake.spans.domain import IntakeTrace, TraceListFilter, TraceMode +from nmp.intake.spans.domain import ( + IntakeTrace, + TraceListFilter, + TraceMetricBucket, + TraceMetricPoint, + TraceMode, +) class TraceRepository(ABC): @@ -29,6 +35,16 @@ async def list_traces( async def get_trace(self, *, workspace: str, trace_id: str, mode: TraceMode) -> IntakeTrace | None: pass + @abstractmethod + async def trace_metrics( + self, + *, + filters: TraceListFilter, + bucket: TraceMetricBucket, + timezone_name: str, + ) -> list[TraceMetricPoint]: + pass + @abstractmethod async def latest_trace_started_at_by_group( self, diff --git a/services/intake/src/nmp/intake/service.py b/services/intake/src/nmp/intake/service.py index cdf91ec2d0..7b46a094e5 100644 --- a/services/intake/src/nmp/intake/service.py +++ b/services/intake/src/nmp/intake/service.py @@ -20,7 +20,7 @@ ) from nmp.intake.repository.clickhouse.evaluation_rollup import ClickHouseEvaluationRollupRepository from nmp.intake.repository.clickhouse.executor import ClickHouseExecutor -from nmp.intake.spans.api import annotations, evaluator_results, sessions, spans, traces +from nmp.intake.spans.api import annotations, evaluator_results, sessions, spans, trace_metrics, traces from nmp.intake.spans.clickhouse_client import ClickHouseSettings, ClickHouseSpanClient from nmp.intake.spans.ingest import atif, chat_completions, otlp from nmp.intake.spans.ingest import spans as span_ingest @@ -56,6 +56,12 @@ def get_routers(self) -> List[RouterConfig]: """Return routers for the intake service.""" return [ RouterConfig(spans.router, tag="Spans", description="ClickHouse-backed span read endpoints"), + # Must precede traces: /traces/metrics would otherwise bind to /traces/{id}. + RouterConfig( + trace_metrics.router, + tag="Traces", + description="Time-bucketed trace metric rollups", + ), RouterConfig(traces.router, tag="Traces", description="ClickHouse-backed trace summary read endpoints"), RouterConfig(sessions.router, tag="Sessions", description="ClickHouse-backed session detail endpoints"), RouterConfig( diff --git a/services/intake/src/nmp/intake/spans/api/trace_metrics.py b/services/intake/src/nmp/intake/spans/api/trace_metrics.py new file mode 100644 index 0000000000..d595439dd5 --- /dev/null +++ b/services/intake/src/nmp/intake/spans/api/trace_metrics.py @@ -0,0 +1,118 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Read API for time-bucketed Intake trace metrics.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status +from nmp.common.api.parsed_filter import ParsedFilter, make_filter_dep +from nmp.common.api.utils import generate_openapi_extra_params +from nmp.intake.spans.api.dependencies import SpansServiceDep, require_workspace_access +from nmp.intake.spans.api.trace_metrics_schemas import ( + TraceMetricBucketParam, + TraceMetricPointResponse, + TraceMetrics, +) +from nmp.intake.spans.api.traces import _trace_filter +from nmp.intake.spans.api.traces_schemas import TraceFilter +from nmp.intake.spans.domain import TraceListFilter + +router = APIRouter(dependencies=[Depends(require_workspace_access)]) +API_TAG = "Traces" +_ALLOWED_QUERY_PARAMS = frozenset({"bucket", "timezone", "filter"}) +_DEFAULT_WINDOW = timedelta(days=7) +_DETAIL_BODY = { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": {"detail": {"type": "string"}}, + "required": ["detail"], + } + } + }, +} + + +@router.get( + "/v2/workspaces/{workspace}/traces/metrics", + response_model=TraceMetrics, + response_model_exclude_none=True, + tags=[API_TAG], + openapi_extra=generate_openapi_extra_params( + filter_schema=TraceFilter, + filter_description=( + "Filter the traces the metrics are computed over. Accepts the same fields as the traces " + "list, so agent_name scopes the rollup to one agent. Without a started_at " + "lower bound the rollup covers the last 7 days." + ), + ), + responses={ + 400: {"description": "Unsupported query parameter or unknown timezone", **_DETAIL_BODY}, + 503: {"description": "ClickHouse spans storage unavailable", **_DETAIL_BODY}, + }, +) +async def get_trace_metrics( + workspace: str, + request: Request, + service: SpansServiceDep, + bucket: TraceMetricBucketParam = Query( + default=TraceMetricBucketParam.DAY, + description="Time bucket granularity. total collapses the filtered range into a single row.", + ), + timezone: str = Query( + default="UTC", + description="IANA timezone the buckets are aligned to, e.g. America/Los_Angeles.", + ), + parsed: ParsedFilter = Depends(make_filter_dep(TraceFilter)), +) -> TraceMetrics: + _validate_query_params(request) + _validate_timezone(timezone) + filters = _default_started_at_window(_trace_filter(workspace, parsed)) + points = await service.trace_metrics(filters=filters, bucket=bucket.value, timezone_name=timezone) + return TraceMetrics( + bucket=bucket, + timezone=timezone, + data=[TraceMetricPointResponse.from_domain(point) for point in points], + ) + + +def _default_started_at_window(filters: TraceListFilter) -> TraceListFilter: + """Bound an open-ended rollup so a missing filter cannot scan the whole workspace.""" + + if filters.started_at_gte is None: + window_end = filters.started_at_lte or datetime.now(UTC) + filters.started_at_gte = window_end - _DEFAULT_WINDOW + return filters + + +def _validate_query_params(request: Request) -> None: + """Reject anything this endpoint does not implement. + + The shared list validator always permits page/page_size/sort, which this + endpoint has none of; accepting them silently would imply the response is + paginated when it returns every bucket in the filtered range. + """ + unsupported = sorted( + {key for key in request.query_params if key not in _ALLOWED_QUERY_PARAMS and not key.startswith("filter[")} + ) + if unsupported: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Unsupported query parameter(s): {', '.join(unsupported)}", + ) + + +def _validate_timezone(timezone: str) -> None: + # ClickHouse would reject an unknown zone with an opaque server error; fail here instead. + try: + ZoneInfo(timezone) + except (ZoneInfoNotFoundError, ValueError): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Unknown timezone: {timezone}", + ) from None diff --git a/services/intake/src/nmp/intake/spans/api/trace_metrics_schemas.py b/services/intake/src/nmp/intake/spans/api/trace_metrics_schemas.py new file mode 100644 index 0000000000..45de21a019 --- /dev/null +++ b/services/intake/src/nmp/intake/spans/api/trace_metrics_schemas.py @@ -0,0 +1,68 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Pydantic schemas for time-bucketed Intake trace metrics.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum +from typing import Self + +from nmp.intake.spans.domain import TraceMetricPoint +from pydantic import BaseModel, Field + + +class TraceMetricBucketParam(StrEnum): + TOTAL = "total" + HOUR = "hour" + DAY = "day" + WEEK = "week" + MONTH = "month" + + +class TokenRollupResponse(BaseModel): + sum: int | None = Field(default=None, ge=0) + mean: float | None = None + p90: float | None = None + p99: float | None = None + + +class CostRollupResponse(BaseModel): + sum: float | None = None + mean: float | None = None + p90: float | None = None + p99: float | None = None + + +class LatencyRollupResponse(BaseModel): + mean: float | None = None + p50: float | None = None + p90: float | None = None + p95: float | None = None + p99: float | None = None + + +class TraceMetricPointResponse(BaseModel): + bucket_start: datetime | None = Field( + default=None, + description="Start of the bucket in the requested timezone. Omitted when bucket=total.", + ) + run_count: int = Field(ge=0, description="Agent runs started in this bucket.") + failed_run_count: int = Field(ge=0, description="Runs whose root span ended in error.") + input_tokens: TokenRollupResponse + output_tokens: TokenRollupResponse + cached_tokens: TokenRollupResponse + total_tokens: TokenRollupResponse + cost_usd: CostRollupResponse + latency_ms: LatencyRollupResponse + + @classmethod + def from_domain(cls, point: TraceMetricPoint) -> Self: + return cls.model_validate(point, from_attributes=True) + + +class TraceMetrics(BaseModel): + bucket: TraceMetricBucketParam + timezone: str + data: list[TraceMetricPointResponse] diff --git a/services/intake/src/nmp/intake/spans/api/traces.py b/services/intake/src/nmp/intake/spans/api/traces.py index 273cecce95..96ffebf8a9 100644 --- a/services/intake/src/nmp/intake/spans/api/traces.py +++ b/services/intake/src/nmp/intake/spans/api/traces.py @@ -30,6 +30,7 @@ "test_case_name", "evaluation_id", "test_case_id", + "agent_name", } ) TRACE_INDEX_FILTER_ALIASES = { @@ -37,6 +38,7 @@ "evaluation_id": "evaluation_name", "test_case_name": "test_case_name", "test_case_id": "test_case_name", + "agent_name": "agent_name", } @@ -49,7 +51,7 @@ filter_schema=TraceFilter, filter_description=( "Filter root-span-backed traces by id, session_id, root status, root span started_at, " - "evaluation_name, and test_case_name." + "evaluation_name, test_case_name, and agent_name." ), ), ) diff --git a/services/intake/src/nmp/intake/spans/api/traces_schemas.py b/services/intake/src/nmp/intake/spans/api/traces_schemas.py index 8fe8a2e8cf..2d4092b3c2 100644 --- a/services/intake/src/nmp/intake/spans/api/traces_schemas.py +++ b/services/intake/src/nmp/intake/spans/api/traces_schemas.py @@ -46,6 +46,7 @@ class TraceFilter(BaseModel): deprecated=True, description="Deprecated alias for test_case_name. Use test_case_name instead.", ) + agent_name: str | None = Field(default=None, description="Filter by root-span agent name.") class Trace(BaseModel): @@ -69,6 +70,8 @@ class Trace(BaseModel): ), ) evaluation_context: EvaluationContext | None = None + agent_name: str | None = None + agent_version: str | None = None started_at: datetime ended_at: datetime | None = None duration_ms: float | None = None @@ -80,6 +83,14 @@ class Trace(BaseModel): cost_usd: float | None = None cost_input_usd: float | None = None cost_output_usd: float | None = None + models: list[str] | None = Field( + default=None, + description="Distinct models used across the trace's spans. Omitted in summary mode.", + ) + providers: list[str] | None = Field( + default=None, + description="Distinct inference providers used across the trace's spans. Omitted in summary mode.", + ) span_count: int | None = Field(default=None, ge=0) error_count: int | None = Field(default=None, ge=0) @@ -94,6 +105,8 @@ def from_domain(cls, trace: IntakeTrace, *, mode: TraceMode = "detailed") -> Sel input=text_for_mode(trace.input, mode=mode), output=text_for_mode(trace.output, mode=mode), evaluation_context=_evaluation_context(trace), + agent_name=trace.agent_name, + agent_version=trace.agent_version, started_at=trace.started_at, ended_at=trace.ended_at, duration_ms=trace.duration_ms, @@ -105,6 +118,8 @@ def from_domain(cls, trace: IntakeTrace, *, mode: TraceMode = "detailed") -> Sel cost_usd=trace.cost_usd, cost_input_usd=trace.cost_input_usd, cost_output_usd=trace.cost_output_usd, + models=trace.models, + providers=trace.providers, span_count=trace.span_count, error_count=trace.error_count, ) diff --git a/services/intake/src/nmp/intake/spans/clickhouse_migrations.py b/services/intake/src/nmp/intake/spans/clickhouse_migrations.py index 5515ad77c8..ae51424d90 100644 --- a/services/intake/src/nmp/intake/spans/clickhouse_migrations.py +++ b/services/intake/src/nmp/intake/spans/clickhouse_migrations.py @@ -256,6 +256,8 @@ def _create_trace_index_schema(client, settings: ClickHouseMigrationSettings) -> # Ingest writes canonical keys, while the backfill must preserve associations on historical rows. evaluation_name_expr = _coalesced_string_attribute(SpanAttributeField.EVALUATION_NAME) test_case_name_expr = _coalesced_string_attribute(SpanAttributeField.TEST_CASE_NAME) + agent_name_key = spec_for_field(SpanAttributeField.AGENT_NAME).bag_key + agent_version_key = spec_for_field(SpanAttributeField.AGENT_VERSION).bag_key # Note this is logically a single table. CH requires creating an underlying table and then a view that writes to that table. client.command( @@ -276,6 +278,9 @@ def _create_trace_index_schema(client, settings: ClickHouseMigrationSettings) -> evaluation_name String DEFAULT '', test_case_name String DEFAULT '', + agent_name String DEFAULT '', + agent_version String DEFAULT '', + root_started_at DateTime64(6) CODEC(Delta(8), ZSTD(1)), root_ended_at Nullable(DateTime64(6)) CODEC(Delta(8), ZSTD(1)), latency_ms Nullable(Float64), @@ -287,6 +292,7 @@ def _create_trace_index_schema(client, settings: ClickHouseMigrationSettings) -> INDEX idx_session_id session_id TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_evaluation_name evaluation_name TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_test_case_name test_case_name TYPE bloom_filter(0.01) GRANULARITY 1, + INDEX idx_agent_name agent_name TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_root_status root_status TYPE set(4) GRANULARITY 4, INDEX idx_source_format source_format TYPE set(8) GRANULARITY 4 ) @@ -314,6 +320,8 @@ def _create_trace_index_schema(client, settings: ClickHouseMigrationSettings) -> attributes_string['{project_key}'] AS project, {evaluation_name_expr} AS evaluation_name, {test_case_name_expr} AS test_case_name, + attributes_string['{agent_name_key}'] AS agent_name, + attributes_string['{agent_version_key}'] AS agent_version, start_time AS root_started_at, nullIf(end_time, toDateTime64(0, 6)) AS root_ended_at, if(end_time = toDateTime64(0, 6), NULL, dateDiff('millisecond', start_time, end_time)) AS latency_ms, @@ -366,6 +374,11 @@ def _create_trace_index_schema(client, settings: ClickHouseMigrationSettings) -> # and the trace_index columns now describe the name values they store. Rebuild the derived table # and MV with canonical columns while the backfill coalesces canonical and historical bag keys. ("ch_trace_index_0007_nemo_test_case_name", _create_trace_index_schema), + # trace_index gained ``agent_name`` / ``agent_version`` so agent-scoped trace + # listing and metric rollups filter on a real column instead of scanning the spans attribute + # map. Rebuild backfills them from ``spans``, which shares trace_index's retention window, so + # no retained history is lost. A separate revision because 0007 already shipped. + ("ch_trace_index_0008_agent", _create_trace_index_schema), ] CURRENT_SCHEMA_VERSION = _MIGRATIONS[-1][0] diff --git a/services/intake/src/nmp/intake/spans/domain.py b/services/intake/src/nmp/intake/spans/domain.py index 35d84b7e5a..0d3144534b 100644 --- a/services/intake/src/nmp/intake/spans/domain.py +++ b/services/intake/src/nmp/intake/spans/domain.py @@ -98,6 +98,56 @@ class TraceListFilter(BaseModel): started_at_lte: datetime | None = None evaluation_name: str | None = None test_case_name: str | None = None + agent_name: str | None = None + + +TraceMetricBucket = Literal["total", "hour", "day", "week", "month"] + + +class TokenRollup(BaseModel): + """Distribution of a per-run token count across the runs in one bucket.""" + + sum: int | None = Field(default=None, ge=0) + mean: float | None = None + p90: float | None = None + p99: float | None = None + + +class CostRollup(BaseModel): + """Distribution of per-run cost across the runs in one bucket.""" + + sum: float | None = None + mean: float | None = None + p90: float | None = None + p99: float | None = None + + +class LatencyRollup(BaseModel): + """Distribution of end-to-end run duration across the runs in one bucket.""" + + mean: float | None = None + p50: float | None = None + p90: float | None = None + p95: float | None = None + p99: float | None = None + + +class TraceMetricPoint(BaseModel): + """One time bucket of agent-scoped trace metrics. + + ``bucket_start`` is None only for the ``total`` bucket, which collapses the + whole filtered range into a single row. + """ + + bucket_start: datetime | None = None + run_count: int = Field(ge=0) + failed_run_count: int = Field(ge=0) + input_tokens: TokenRollup + output_tokens: TokenRollup + cached_tokens: TokenRollup + total_tokens: TokenRollup + cost_usd: CostRollup + latency_ms: LatencyRollup IntakeResponseMode = Literal["summary", "preview", "detailed"] @@ -117,6 +167,8 @@ class IntakeTrace(BaseModel): project: str | None = None evaluation_name: str | None = None test_case_name: str | None = None + agent_name: str | None = None + agent_version: str | None = None started_at: datetime ended_at: datetime | None = None duration_ms: float | None = None diff --git a/services/intake/src/nmp/intake/spans/service.py b/services/intake/src/nmp/intake/spans/service.py index fb536a79ed..4eaf2448e6 100644 --- a/services/intake/src/nmp/intake/spans/service.py +++ b/services/intake/src/nmp/intake/spans/service.py @@ -26,6 +26,8 @@ SpanListFilter, TraceBatch, TraceListFilter, + TraceMetricBucket, + TraceMetricPoint, TraceMode, ) @@ -142,6 +144,15 @@ async def get_trace(self, *, workspace: str, trace_id: str, mode: TraceMode) -> raise TraceNotFoundError(workspace, trace_id) return trace + async def trace_metrics( + self, + *, + filters: TraceListFilter, + bucket: TraceMetricBucket, + timezone_name: str, + ) -> list[TraceMetricPoint]: + return await self._traces.trace_metrics(filters=filters, bucket=bucket, timezone_name=timezone_name) + async def latest_trace_started_at_by_group( self, *, diff --git a/services/intake/tests/integration/spans/test_clickhouse_bootstrap.py b/services/intake/tests/integration/spans/test_clickhouse_bootstrap.py index 0ba89cb113..332643c06b 100644 --- a/services/intake/tests/integration/spans/test_clickhouse_bootstrap.py +++ b/services/intake/tests/integration/spans/test_clickhouse_bootstrap.py @@ -43,6 +43,7 @@ def test_clickhouse_bootstrap_is_idempotent(clickhouse_client: ClickHouseSpanCli ("ch_trace_index_0005_evaluation_id",), ("ch_trace_index_0006_nemo_evaluation_name",), ("ch_trace_index_0007_nemo_test_case_name",), + ("ch_trace_index_0008_agent",), ] expected_ttl = { ClickHouseTable.SPANS: "TTL toDate(start_time) + toIntervalDay(90)", diff --git a/services/intake/tests/integration/spans/test_otlp_ingest_simple.py b/services/intake/tests/integration/spans/test_otlp_ingest_simple.py index 0083a5600d..3b5f804fc0 100644 --- a/services/intake/tests/integration/spans/test_otlp_ingest_simple.py +++ b/services/intake/tests/integration/spans/test_otlp_ingest_simple.py @@ -157,7 +157,7 @@ def test_otlp_ingest_openinference_session_id(client: TestClient, make_otlp_requ assert "error_message" not in span -def test_otlp_reingest_same_batch_collapses_after_merge( +def test_otlp_reingest_same_batch_deduplicates_before_and_after_merge( client: TestClient, make_otlp_request, clickhouse_client: ClickHouseSpanClient, @@ -172,6 +172,7 @@ def test_otlp_reingest_same_batch_collapses_after_merge( "openinference.span.kind": "LLM", "gen_ai.conversation.id": "conv-idempotent", "gen_ai.response.model": "model-idempotent", + "gen_ai.usage.input_tokens": 123, }, } ] @@ -186,6 +187,16 @@ def test_otlp_reingest_same_batch_collapses_after_merge( assert ingest_response.status_code == 200, ingest_response.text assert ingest_response.json() == {"errors": []} + traces_response = client.get( + "/apis/intake/v2/workspaces/default/traces", + params={"filter[session_id]": "conv-idempotent"}, + ) + assert traces_response.status_code == 200, traces_response.text + trace_payload = traces_response.json() + assert trace_payload["pagination"]["total_results"] == 1 + assert trace_payload["data"][0]["span_count"] == 1 + assert trace_payload["data"][0]["input_tokens"] == 123 + run_async( clickhouse_client.command( f"OPTIMIZE TABLE {qualified_table(clickhouse_client.database, ClickHouseTable.SPANS)} FINAL" diff --git a/services/intake/tests/integration/spans/test_traces_read.py b/services/intake/tests/integration/spans/test_traces_read.py index 6b40da141c..2b307d07a5 100644 --- a/services/intake/tests/integration/spans/test_traces_read.py +++ b/services/intake/tests/integration/spans/test_traces_read.py @@ -106,7 +106,10 @@ def test_traces_read_returns_core_trace_summary(client: TestClient, make_otlp_re assert "test_case_id" not in trace assert "source_format" not in trace assert "project" not in trace - assert "models" not in trace + # models and providers were withheld when this API was designed (#9); ASTD-391's + # details card needs them, so they are now part of the rollup response. + assert trace["models"] == ["gpt-4o-mini"] + assert trace["providers"] == ["openai"] get_response = client.get(f"/apis/intake/v2/workspaces/default/traces/{trace['id']}") assert get_response.status_code == 200, get_response.text @@ -129,6 +132,9 @@ def test_traces_read_returns_core_trace_summary(client: TestClient, make_otlp_re assert "input_tokens" not in summary_trace assert "cost_usd" not in summary_trace assert "span_count" not in summary_trace + # summary skips the span aggregate join, so these have no value to report + assert "models" not in summary_trace + assert "providers" not in summary_trace assert "input" not in summary_trace assert "output" not in summary_trace diff --git a/services/intake/tests/test_atif_v17.py b/services/intake/tests/test_atif_v17.py index 057ba0e1f3..7b478914fa 100644 --- a/services/intake/tests/test_atif_v17.py +++ b/services/intake/tests/test_atif_v17.py @@ -1109,3 +1109,27 @@ def test_atif_v17_observation_on_nested_system_step() -> None: observation = _step_observation(system_step) assert observation is not None assert observation.results[0].content == "is_likely_phishing: true" + + +def test_atif_mapping_stamps_agent_name_on_every_span() -> None: + trajectory = AtifTrajectory( + schema_version="ATIF-v1.7", + session_id="trace-session-id", + agent=AtifAgent( + name="sample-agent", + version="1.0.0", + model_name="provider/sample-model", + ), + steps=[AtifStepUser(step_id=1, source="user", message="solve")], + ) + + spans = trajectory_to_spans( + workspace="default", + trajectory=trajectory, + ingested_at=datetime(2026, 5, 18, tzinfo=timezone.utc), + ) + + # Not just the root, so agent-scoped span queries work too. + assert spans + for span in spans: + assert span.attributes_string["gen_ai.agent.name"] == "sample-agent" diff --git a/services/intake/tests/test_evaluation_session_clickhouse_repository.py b/services/intake/tests/test_evaluation_session_clickhouse_repository.py index 5da836b12a..d60c220a6a 100644 --- a/services/intake/tests/test_evaluation_session_clickhouse_repository.py +++ b/services/intake/tests/test_evaluation_session_clickhouse_repository.py @@ -118,6 +118,8 @@ async def test_list_sessions_maps_rows_and_binds_all_request_values() -> None: assert "'' AS input" in executor.queries[1].statement assert "'' AS output" in executor.queries[1].statement assert "substringUTF8(root_input, 1, %(payload_char_limit)s) AS input" in executor.queries[2].statement + assert "FROM spans AS span_versions FINAL" in executor.queries[2].statement + assert "argMax(span_versions." not in executor.queries[2].statement assert executor.queries[2].parameters["page_session_ids"] == ["session-a"] assert executor.queries[2].parameters["page_trace_ids"] == ["trace-session-a"] assert executor.queries[2].parameters["page_storage_keys"] == [ @@ -151,6 +153,9 @@ async def test_metric_sort_uses_bounded_page_then_restores_hydration_order() -> "evaluation_sessions.metric_sort.page", "evaluation_sessions.hydrate", ] + assert "FROM spans AS span_versions FINAL" in executor.queries[1].statement + assert "argMax(span_versions." not in executor.queries[1].statement + assert "FROM spans AS span_versions FINAL" in executor.queries[2].statement assert executor.queries[2].parameters["page_session_ids"] == ["session-b", "session-a"] diff --git a/services/intake/tests/test_spans_clickhouse_migrations.py b/services/intake/tests/test_spans_clickhouse_migrations.py index b9130af065..8e56a39c7a 100644 --- a/services/intake/tests/test_spans_clickhouse_migrations.py +++ b/services/intake/tests/test_spans_clickhouse_migrations.py @@ -74,6 +74,11 @@ def test_trace_index_schema_is_root_span_projection(): assert "INDEX idx_evaluation_name evaluation_name" in ddl assert "INDEX idx_test_case_name test_case_name" in ddl assert "index_granularity = 256" in ddl + # Agent identity is denormalized so agent-scoped listing and metric rollups filter on a + # column instead of probing the spans attribute map. + assert "attributes_string['{agent_name_key}'] AS agent_name" in ddl + assert "attributes_string['{agent_version_key}'] AS agent_version" in ddl + assert "INDEX idx_agent_name agent_name" in ddl def test_trace_index_mv_keys_match_attribute_catalog(): @@ -84,3 +89,6 @@ def test_trace_index_mv_keys_match_attribute_catalog(): test_case_spec = spec_for_field(SpanAttributeField.TEST_CASE_NAME) assert test_case_spec.bag_key == "nemo.test_case.name" assert test_case_spec.bag_aliases == ("nemo.test_case.id",) + # The MV bakes these keys in at creation time, so a catalog rename needs a new migration. + assert spec_for_field(SpanAttributeField.AGENT_NAME).bag_key == "gen_ai.agent.name" + assert spec_for_field(SpanAttributeField.AGENT_VERSION).bag_key == "agent.version" diff --git a/services/intake/tests/test_trace_metrics_api.py b/services/intake/tests/test_trace_metrics_api.py new file mode 100644 index 0000000000..9f26fd4fa5 --- /dev/null +++ b/services/intake/tests/test_trace_metrics_api.py @@ -0,0 +1,194 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Trace metrics API tests.""" + +from collections.abc import Iterable, Iterator +from datetime import datetime, timedelta, timezone + +import pytest +from fastapi import FastAPI, HTTPException, Request +from fastapi.routing import APIRoute +from nmp.intake.spans.api import trace_metrics, traces +from nmp.intake.spans.api.trace_metrics import ( + _default_started_at_window, + _validate_query_params, + _validate_timezone, +) +from nmp.intake.spans.api.trace_metrics_schemas import ( + TraceMetricBucketParam, + TraceMetricPointResponse, + TraceMetrics, +) +from nmp.intake.spans.domain import ( + CostRollup, + LatencyRollup, + TokenRollup, + TraceListFilter, + TraceMetricPoint, +) +from starlette.routing import Match + + +def test_validate_timezone_accepts_iana_zones() -> None: + _validate_timezone("UTC") + _validate_timezone("America/Los_Angeles") + + +@pytest.mark.parametrize("bad", ["Not/AZone", "", "PST8PDT7"]) +def test_validate_timezone_rejects_unknown_zones(bad: str) -> None: + # ClickHouse would otherwise fail with an opaque server error. + with pytest.raises(HTTPException) as exc: + _validate_timezone(bad) + assert exc.value.status_code == 400 + + +def _metric_point( + *, + run_count: int = 0, + failed_run_count: int = 0, + bucket_start: datetime | None = None, + input_tokens: TokenRollup | None = None, + output_tokens: TokenRollup | None = None, + cached_tokens: TokenRollup | None = None, + total_tokens: TokenRollup | None = None, + cost_usd: CostRollup | None = None, + latency_ms: LatencyRollup | None = None, +) -> TraceMetricPoint: + """Every rollup object is required; name only the ones a test cares about.""" + + return TraceMetricPoint( + bucket_start=bucket_start, + run_count=run_count, + failed_run_count=failed_run_count, + input_tokens=input_tokens or TokenRollup(), + output_tokens=output_tokens or TokenRollup(), + cached_tokens=cached_tokens or TokenRollup(), + total_tokens=total_tokens or TokenRollup(), + cost_usd=cost_usd or CostRollup(), + latency_ms=latency_ms or LatencyRollup(), + ) + + +def test_metric_point_response_round_trips_the_domain_model() -> None: + point = _metric_point( + bucket_start=datetime(2026, 8, 14, tzinfo=timezone.utc), + run_count=3, + failed_run_count=1, + input_tokens=TokenRollup(sum=300, mean=100.0, p90=180.0, p99=190.0), + latency_ms=LatencyRollup(mean=2100.0, p95=4300.0), + ) + + response = TraceMetricPointResponse.from_domain(point) + + assert response.bucket_start == point.bucket_start + assert response.run_count == 3 + assert response.failed_run_count == 1 + assert response.input_tokens.sum == 300 + assert response.input_tokens.mean == 100.0 + assert response.latency_ms.p95 == 4300.0 + # Unset metrics stay null rather than defaulting to zero. + assert response.cost_usd.sum is None + assert response.latency_ms.p50 is None + + +def test_total_bucket_omits_bucket_start_from_the_payload() -> None: + metrics = TraceMetrics( + bucket=TraceMetricBucketParam.TOTAL, + timezone="UTC", + data=[TraceMetricPointResponse.from_domain(_metric_point(run_count=5))], + ) + + payload = metrics.model_dump(exclude_none=True) + + assert payload["bucket"] == "total" + assert "bucket_start" not in payload["data"][0] + assert payload["data"][0]["run_count"] == 5 + + +def test_unspecified_range_defaults_to_the_last_seven_days() -> None: + filters = _default_started_at_window(TraceListFilter(workspace="workspace-a")) + + assert filters.started_at_gte is not None + window = datetime.now(timezone.utc) - filters.started_at_gte + assert timedelta(days=7) <= window < timedelta(days=7, minutes=1) + + +def test_default_window_anchors_to_an_explicit_upper_bound() -> None: + upper = datetime(2026, 8, 14, tzinfo=timezone.utc) + + filters = _default_started_at_window(TraceListFilter(workspace="workspace-a", started_at_lte=upper)) + + assert filters.started_at_gte == upper - timedelta(days=7) + + +def test_default_window_leaves_an_explicit_lower_bound_alone() -> None: + lower = datetime(2025, 1, 1, tzinfo=timezone.utc) + + filters = _default_started_at_window(TraceListFilter(workspace="workspace-a", started_at_gte=lower)) + + assert filters.started_at_gte == lower + + +@pytest.mark.parametrize("param", ["page", "page_size", "sort"]) +def test_rejects_pagination_params_the_endpoint_does_not_implement(param: str) -> None: + # The shared list validator allows these; accepting them would imply the + # response is paginated when it returns every bucket in the filtered range. + request = Request({"type": "http", "query_string": f"{param}=2".encode(), "headers": []}) + + with pytest.raises(HTTPException) as exc: + _validate_query_params(request) + assert exc.value.status_code == 400 + assert param in str(exc.value.detail) + + +@pytest.mark.parametrize("query", [b"", b"bucket=day", b"timezone=UTC", b"filter[agent_name]=x"]) +def test_accepts_supported_params_and_deep_object_filters(query: bytes) -> None: + request = Request({"type": "http", "query_string": query, "headers": []}) + + _validate_query_params(request) + + +def test_metrics_route_is_registered_ahead_of_the_trace_id_route() -> None: + """/traces/metrics must not be swallowed by /traces/{id}.""" + + app = FastAPI() + for router in (trace_metrics.router, traces.router): + app.include_router(router) + scope = { + "type": "http", + "method": "GET", + "path": "/v2/workspaces/workspace-a/traces/metrics", + "path_params": {}, + "root_path": "", + "headers": [], + } + + matched = next(route for route in _api_routes(app.routes) if route.matches(scope)[0] is Match.FULL) + + assert matched.endpoint is trace_metrics.get_trace_metrics + + +def _api_routes(routes: Iterable[object]) -> Iterator[APIRoute]: + """Flatten routes in match order; include_router wraps each router in a proxy.""" + + for route in routes: + if isinstance(route, APIRoute): + yield route + continue + included = getattr(route, "original_router", None) + if included is not None: + yield from _api_routes(included.routes) + + +def test_metrics_route_documents_its_error_responses() -> None: + """400 and 503 are both reachable, so generated clients must know about them.""" + + route = next( + route for route in _api_routes(trace_metrics.router.routes) if route.endpoint is trace_metrics.get_trace_metrics + ) + + assert set(route.responses) == {400, 503} + for status_code in (400, 503): + schema = route.responses[status_code]["content"]["application/json"]["schema"] + assert schema["required"] == ["detail"] diff --git a/services/intake/tests/test_traces_api.py b/services/intake/tests/test_traces_api.py index 1126b975aa..3aa8de79a9 100644 --- a/services/intake/tests/test_traces_api.py +++ b/services/intake/tests/test_traces_api.py @@ -11,8 +11,8 @@ from nmp.common.api.filter import parse_json_filter from nmp.common.api.parsed_filter import ParsedFilter from nmp.intake.spans.api.traces import _trace_filter -from nmp.intake.spans.api.traces_schemas import TraceFilter -from nmp.intake.spans.domain import SpanStatus +from nmp.intake.spans.api.traces_schemas import Trace, TraceFilter +from nmp.intake.spans.domain import IntakeTrace, SpanStatus def test_trace_filter_maps_public_fields_to_repository_filter(): @@ -84,6 +84,19 @@ def test_trace_filter_schema_exposes_canonical_names_and_deprecated_aliases(): assert "experiment_id" not in properties +def test_trace_filter_accepts_agent_name(): + filters = _trace_filter("workspace-a", _parsed_filter({"agent_name": "support-bot"})) + + assert filters.agent_name == "support-bot" + + +def test_trace_filter_schema_exposes_agent_name(): + properties = TraceFilter.model_json_schema()["properties"] + + assert properties["agent_name"]["description"] == "Filter by root-span agent name." + assert "agent_id" not in properties + + def test_trace_filter_applies_no_implicit_time_bound(): filters = _trace_filter("workspace-a", _parsed_filter({"id": "trace-a"})) @@ -94,3 +107,41 @@ def test_trace_filter_applies_no_implicit_time_bound(): def _parsed_filter(value: dict[str, object]) -> ParsedFilter: return ParsedFilter(operation=parse_json_filter(json.dumps(value))) + + +def test_trace_response_exposes_models_and_providers() -> None: + trace = IntakeTrace( + id="trace-a", + workspace="workspace-a", + session_id="session-a", + source_format="atif", + started_at=datetime(2026, 8, 14, tzinfo=timezone.utc), + ingested_at=datetime(2026, 8, 14, tzinfo=timezone.utc), + status=SpanStatus.SUCCESS, + models=["claude-opus-4-6", "qwen3-next-80b"], + providers=["anthropic", "openai"], + ) + + response = Trace.from_domain(trace, mode="detailed") + + assert response.models == ["claude-opus-4-6", "qwen3-next-80b"] + assert response.providers == ["anthropic", "openai"] + + +def test_trace_response_omits_models_when_the_rollup_did_not_run() -> None: + # summary mode skips the span-aggregate join, so these stay unset rather + # than reporting an empty list the caller would read as "no models". + trace = IntakeTrace( + id="trace-a", + workspace="workspace-a", + session_id="session-a", + source_format="atif", + started_at=datetime(2026, 8, 14, tzinfo=timezone.utc), + ingested_at=datetime(2026, 8, 14, tzinfo=timezone.utc), + status=SpanStatus.SUCCESS, + ) + + payload = Trace.from_domain(trace, mode="summary").model_dump(exclude_none=True) + + assert "models" not in payload + assert "providers" not in payload diff --git a/services/intake/tests/test_traces_clickhouse_repository.py b/services/intake/tests/test_traces_clickhouse_repository.py index d7e60defdd..b4c33e63fc 100644 --- a/services/intake/tests/test_traces_clickhouse_repository.py +++ b/services/intake/tests/test_traces_clickhouse_repository.py @@ -206,10 +206,10 @@ async def test_detailed_mode_adds_trace_aggregate_block(): assert "AS trace_spans" in client.queries[2] assert "span_versions.trace_id IN %(page_trace_ids)s" in client.queries[2] assert "(span_versions.source_format, span_versions.trace_id) IN %(page_trace_keys)s" in client.queries[2] - assert "argMax(span_versions.input," not in client.queries[2] - assert "argMax(span_versions.output," not in client.queries[2] - assert "argMax(span_versions.attributes_string," in client.queries[2] - assert "argMax(span_versions.attributes_number," in client.queries[2] + assert "FROM spans AS span_versions FINAL" in client.queries[2] + assert "argMax(span_versions." not in client.queries[2] + assert "span_versions.attributes_string AS attributes_string" in client.queries[2] + assert "span_versions.attributes_number AS attributes_number" in client.queries[2] assert "sumIf" in client.queries[2] assert "groupUniqArrayIf" in client.queries[2] assert "count() AS span_count" in client.queries[2] @@ -402,6 +402,25 @@ async def test_root_filters_use_trace_index_columns(): assert client.parameters[0]["filter_evaluation_name"] == "experiment-a" +@pytest.mark.asyncio +async def test_agent_filter_uses_the_trace_index_column(): + client = _Client() + repository = _repository(client) + + await repository.list_traces( + filters=TraceListFilter(workspace="workspace-a", agent_name="support-bot"), + page=1, + page_size=10, + sort="started_at", + mode="detailed", + ) + + assert "trace_roots.agent_name = %(filter_agent_name)s" in client.queries[0] + # Agent scoping must not fall back to probing the spans attribute map. + assert "candidate_spans" not in client.queries[0] + assert client.parameters[0]["filter_agent_name"] == "support-bot" + + def _trace_row( *, trace_id: str = "trace-a", @@ -439,3 +458,169 @@ def _trace_row( "ingested_at": ingested_at, } return tuple(values[column] for column in TRACE_COLUMNS) + + +_METRIC_COLUMNS = [ + "bucket_start", + "run_count", + "failed_run_count", + *[ + f"{field}_{suffix}" + for field in ("input_tokens", "output_tokens", "cached_tokens", "total_tokens", "cost_usd") + for suffix in ("sum", "mean", "quantiles") + ], + "latency_ms_mean", + "latency_ms_quantiles", +] + + +def _metric_row(*, bucket_start, run_count, failed_run_count, **rollups): + """Build one _METRIC_COLUMNS-shaped row; unnamed rollups come back empty.""" + + row = [bucket_start, run_count, failed_run_count] + for field in ("input_tokens", "output_tokens", "cached_tokens", "total_tokens", "cost_usd"): + row.extend(rollups.get(field, (None, None, []))) + row.extend(rollups.get("latency_ms", (None, []))) + return tuple(row) + + +@pytest.mark.asyncio +async def test_trace_metrics_buckets_by_day_in_the_requested_timezone(): + bucket_start = datetime(2026, 8, 14, tzinfo=timezone.utc) + client = _Client( + [ + _QueryResult( + [ + _metric_row( + bucket_start=bucket_start, + run_count=3, + failed_run_count=1, + input_tokens=(300, 100.0, [180.0, 190.0]), + output_tokens=(150, 50.0, [90.0, 95.0]), + cost_usd=(0.5, 0.16, [0.3, 0.4]), + latency_ms=(2000.0, [1200.0, 3900.0, 4300.0, 4800.0]), + ) + ], + _METRIC_COLUMNS, + ) + ] + ) + repository = _repository(client) + + points = await repository.trace_metrics( + filters=TraceListFilter(workspace="workspace-a", agent_name="support-bot"), + bucket="day", + timezone_name="America/Los_Angeles", + ) + + query = client.queries[0] + assert "toStartOfDay(trace_roots.root_started_at, %(metrics_timezone)s)" in query + assert "trace_roots.agent_name = %(filter_agent_name)s" in query + assert client.parameters[0]["metrics_timezone"] == "America/Los_Angeles" + assert client.parameters[0]["filter_agent_name"] == "support-bot" + + assert len(points) == 1 + point = points[0] + assert point.bucket_start == bucket_start + assert point.run_count == 3 + assert point.failed_run_count == 1 + assert point.input_tokens.sum == 300 + assert point.input_tokens.mean == 100.0 + assert point.input_tokens.p90 == 180.0 + assert point.input_tokens.p99 == 190.0 + assert point.cost_usd.sum == 0.5 + assert point.latency_ms.mean == 2000.0 + assert point.latency_ms.p50 == 1200.0 + assert point.latency_ms.p95 == 4300.0 + assert point.latency_ms.p99 == 4800.0 + # A mean is not derivable from percentiles, so it must come from the query. + assert "avg(roots.latency_ms) AS latency_ms_mean" in query + # One combined aggregate per metric rather than a quantile state per percentile. + assert "quantiles(0.9, 0.99)(rollups.input_tokens) AS input_tokens_quantiles" in query + assert "quantiles(0.5, 0.9, 0.95, 0.99)(roots.latency_ms) AS latency_ms_quantiles" in query + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("bucket", "expected_sql"), + [ + ("hour", "toStartOfHour(trace_roots.root_started_at, %(metrics_timezone)s)"), + ("week", "toDateTime(toStartOfWeek(trace_roots.root_started_at, 1, %(metrics_timezone)s)"), + ("month", "toDateTime(toStartOfMonth(trace_roots.root_started_at, %(metrics_timezone)s)"), + ], +) +async def test_trace_metrics_supports_coarser_buckets(bucket, expected_sql): + client = _Client([_QueryResult([], _METRIC_COLUMNS)]) + repository = _repository(client) + + await repository.trace_metrics( + filters=TraceListFilter(workspace="workspace-a"), + bucket=bucket, + timezone_name="UTC", + ) + + assert expected_sql in client.queries[0] + + +@pytest.mark.asyncio +async def test_trace_metrics_total_bucket_collapses_to_one_row_without_a_start(): + client = _Client( + [ + _QueryResult( + [ + _metric_row( + bucket_start=datetime(1970, 1, 1, tzinfo=timezone.utc), + run_count=5, + failed_run_count=0, + ) + ], + _METRIC_COLUMNS, + ) + ] + ) + repository = _repository(client) + + points = await repository.trace_metrics( + filters=TraceListFilter(workspace="workspace-a"), + bucket="total", + timezone_name="UTC", + ) + + assert "toStartOfDay" not in client.queries[0] + # The constant bucket value is an implementation detail and must not surface. + assert points[0].bucket_start is None + assert points[0].run_count == 5 + # Traces with no token attributes report null rather than a misleading zero. + assert points[0].input_tokens.sum is None + assert points[0].input_tokens.p90 is None + assert points[0].latency_ms.p50 is None + + +@pytest.mark.asyncio +async def test_trace_metrics_reads_tokens_from_final_spans(): + client = _Client([_QueryResult([], _METRIC_COLUMNS)]) + repository = _repository(client) + + await repository.trace_metrics( + filters=TraceListFilter(workspace="workspace-a"), + bucket="day", + timezone_name="UTC", + ) + + query = client.queries[0] + # spans is a ReplacingMergeTree, so the rollup must read through FINAL rather + # than summing raw versions. + assert "FROM spans AS span_versions FINAL" in query + assert "argMax(span_versions." not in query + assert "LEFT JOIN rollups" in query + # session_id leads the spans sort key after workspace, so it must be carried from + # the selected roots for the primary key to prune before the trace-id bloom filter. + assert "span_versions.session_id IN (SELECT session_id FROM roots)" in query + # The rollup CTE must be scoped by source_format too, not just trace_id, so a + # trace_id shared across ingest formats does not drag in unrelated spans. + assert ( + "(span_versions.source_format, span_versions.trace_id) IN (SELECT source_format, trace_id FROM roots)" in query + ) + # Joined on the full trace identity, matching the trace hydration query. + assert "roots.workspace = rollups.workspace" in query + assert "roots.source_format = rollups.source_format" in query