diff --git a/.gitignore b/.gitignore index d6798ab0a1..8aa73fc269 100644 --- a/.gitignore +++ b/.gitignore @@ -73,6 +73,4 @@ tests/ui-testing/playwright-tests/utils/auth/user.json /.claude/settings.json /.claude/settings.local.json /.claude/worktrees/ - -__pytest__/ -__pycache__/ +.screenshots/ diff --git a/src/api/http/src/handler/http/router/mod.rs b/src/api/http/src/handler/http/router/mod.rs index 48b02f0869..f94c9489eb 100644 --- a/src/api/http/src/handler/http/router/mod.rs +++ b/src/api/http/src/handler/http/router/mod.rs @@ -1232,6 +1232,10 @@ pub fn service_routes() -> Router { put(experiments::set_experiment_baseline) .delete(experiments::clear_experiment_baseline), ) + .route( + "/{org_id}/experiments/{experiment_id}/rows", + get(experiments::list_experiment_result_rows), + ) .route( "/{org_id}/experiments/{experiment_id}/rows/{row_id}", get(experiments::get_experiment_row), diff --git a/src/api/http/src/handler/http/router/openapi.rs b/src/api/http/src/handler/http/router/openapi.rs index 8f45f8a09b..3ce6f1a4e5 100644 --- a/src/api/http/src/handler/http/router/openapi.rs +++ b/src/api/http/src/handler/http/router/openapi.rs @@ -535,6 +535,7 @@ pub struct ApiDoc; openobserve_api_management::request::experiments::list_experiments, openobserve_api_management::request::experiments::compare_experiments, openobserve_api_management::request::experiments::get_experiment, + openobserve_api_management::request::experiments::list_experiment_result_rows, openobserve_api_management::request::experiments::get_experiment_row, openobserve_api_management::request::experiments::retry_experiment_slot, openobserve_api_management::request::experiments::cancel_experiment, diff --git a/src/api/management/src/models/experiment_comparison.rs b/src/api/management/src/models/experiment_comparison.rs index 28b9693c57..2871271138 100644 --- a/src/api/management/src/models/experiment_comparison.rs +++ b/src/api/management/src/models/experiment_comparison.rs @@ -15,6 +15,9 @@ //! HTTP boundary models for Experiment comparisons. +use std::collections::{BTreeMap, BTreeSet, HashMap}; + +use config::meta::self_reporting::llm_experiments::ExperimentExecutionRecord; pub use domain::DEFAULT_COMPARISON_THRESHOLD; use openobserve_core::llm_evaluations::{ experiment_comparison as domain, experiment_results::ScoringStatus, @@ -115,6 +118,15 @@ pub struct ExperimentComparisonRowBody { pub candidate_row_id: Option, pub bucket: ExperimentComparisonBucketBody, pub dimensions: Vec, + pub trials: Vec, +} + +#[derive(Clone, Debug, PartialEq, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ExperimentComparisonTrialBody { + pub trial_index: u32, + pub baseline_output: Option, + pub candidate_output: Option, } #[derive(Clone, Debug, Default, PartialEq, Serialize, ToSchema)] @@ -314,12 +326,88 @@ impl From for ExperimentComparisonResponseBody { candidate_row_id: row.candidate_row_id, bucket: row.bucket.into(), dimensions: row.dimensions.into_iter().map(Into::into).collect(), + trials: Vec::new(), }) .collect(), } } } +impl ExperimentComparisonResponseBody { + pub fn with_trial_outputs( + mut self, + baseline_executions: &[ExperimentExecutionRecord], + candidate_executions: &[ExperimentExecutionRecord], + ) -> Self { + add_trial_outputs(&mut self.rows, baseline_executions, candidate_executions); + self + } +} + +fn add_trial_outputs( + rows: &mut [ExperimentComparisonRowBody], + baseline_executions: &[ExperimentExecutionRecord], + candidate_executions: &[ExperimentExecutionRecord], +) { + let baseline_outputs = execution_outputs_by_row(baseline_executions); + let candidate_outputs = execution_outputs_by_row(candidate_executions); + + for row in rows { + let baseline = row + .baseline_row_id + .as_deref() + .and_then(|row_id| baseline_outputs.get(row_id)); + let candidate = row + .candidate_row_id + .as_deref() + .and_then(|row_id| candidate_outputs.get(row_id)); + let trial_indexes = baseline + .into_iter() + .flat_map(|outputs| outputs.keys()) + .chain(candidate.into_iter().flat_map(|outputs| outputs.keys())) + .copied() + .collect::>(); + + row.trials = trial_indexes + .into_iter() + .map(|trial_index| ExperimentComparisonTrialBody { + trial_index, + baseline_output: trial_output(baseline, trial_index), + candidate_output: trial_output(candidate, trial_index), + }) + .collect(); + } +} + +fn execution_outputs_by_row( + executions: &[ExperimentExecutionRecord], +) -> HashMap<&str, BTreeMap)>> { + let mut outputs = HashMap::<_, BTreeMap<_, _>>::new(); + for execution in executions { + let trials = outputs.entry(execution.row_id.as_str()).or_default(); + let should_replace = trials + .get(&execution.trial_index) + .is_none_or(|(timestamp, _)| execution._timestamp >= *timestamp); + if should_replace { + trials.insert( + execution.trial_index, + (execution._timestamp, execution.output.as_ref()), + ); + } + } + outputs +} + +fn trial_output( + outputs: Option<&BTreeMap)>>, + trial_index: u32, +) -> Option { + outputs + .and_then(|outputs| outputs.get(&trial_index)) + .and_then(|(_, output)| *output) + .cloned() +} + #[cfg(test)] mod tests { use serde_json::json; @@ -335,6 +423,7 @@ mod tests { candidate_row_id: Some("candidate-row".to_string()), bucket: ExperimentComparisonBucketBody::Unchanged, dimensions: vec![], + trials: vec![], }; let value = serde_json::to_value(body).unwrap(); @@ -405,4 +494,64 @@ mod tests { assert_eq!(value["baselineLabel"], serde_json::Value::Null); assert_eq!(value["candidateLabel"], serde_json::Value::Null); } + + #[test] + fn comparison_rows_expose_latest_baseline_and_candidate_output_per_trial() { + let mut rows = vec![ExperimentComparisonRowBody { + logical_id: "row-1".to_string(), + input: json!({"question": "What changed?"}), + baseline_row_id: Some("baseline-row".to_string()), + candidate_row_id: Some("candidate-row".to_string()), + bucket: ExperimentComparisonBucketBody::Unchanged, + dimensions: vec![], + trials: vec![], + }]; + let baseline = vec![ + execution("baseline-row", 0, 10, json!("old baseline")), + execution("baseline-row", 0, 20, json!("baseline")), + execution("baseline-row", 1, 30, json!("baseline second")), + ]; + let candidate = vec![ + execution("candidate-row", 0, 20, json!("candidate")), + execution("candidate-row", 2, 30, json!("candidate third")), + ]; + + add_trial_outputs(&mut rows, &baseline, &candidate); + + assert_eq!( + serde_json::to_value(&rows[0].trials).unwrap(), + json!([ + { + "trialIndex": 0, + "baselineOutput": "baseline", + "candidateOutput": "candidate" + }, + { + "trialIndex": 1, + "baselineOutput": "baseline second", + "candidateOutput": null + }, + { + "trialIndex": 2, + "baselineOutput": null, + "candidateOutput": "candidate third" + } + ]) + ); + } + + fn execution( + row_id: &str, + trial_index: u32, + timestamp: i64, + output: Value, + ) -> ExperimentExecutionRecord { + ExperimentExecutionRecord { + row_id: row_id.to_string(), + trial_index, + output: Some(output), + _timestamp: timestamp, + ..Default::default() + } + } } diff --git a/src/api/management/src/models/experiments.rs b/src/api/management/src/models/experiments.rs index 04df66e40d..dd4aa9ebf2 100644 --- a/src/api/management/src/models/experiments.rs +++ b/src/api/management/src/models/experiments.rs @@ -27,7 +27,8 @@ use openobserve_core::llm_evaluations::{ experiment_results::{ ExperimentAggregateSummary, ExperimentClientScoreSummary, ExperimentProgress, ExperimentResultScore, ExperimentResultScoreStatus, ExperimentResultSlot, - ExperimentResultTaskStatus, ExperimentScoreSummary, ExperimentSkipSummary, ScoringStatus, + ExperimentResultTaskStatus, ExperimentScoreSummary, ExperimentSkipSummary, + ExperimentSlotStatus, ScoringStatus, }, experiments::{ CloneExperimentOverrides, CreateExperiment, CreateExperimentResult, Experiment, @@ -40,6 +41,59 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use utoipa::{IntoParams, ToSchema}; +#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum ExperimentResultRowSortBody { + #[default] + Dataset, + DispersionDesc, +} + +#[derive(Debug, Deserialize, IntoParams)] +#[into_params(parameter_in = Query)] +#[serde(rename_all = "camelCase")] +pub struct ExperimentResultRowPageQuery { + pub page: Option, + pub page_size: Option, + pub sort: Option, + pub high_dispersion_only: Option, +} + +/// One pinned Dataset case with every trial reduced to list-surface aggregates. +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ExperimentResultRowBody { + /// Zero-based position in deterministic pinned-snapshot row order. + pub row_index: usize, + pub row_id: String, + pub logical_id: String, + pub input: Value, + pub expected_output: Option, + pub trial_count: usize, + pub status: ExperimentSlotStatusBody, + /// Present only for a single-trial row; multi-trial outputs belong in drill-down. + pub output: Option, + pub score_summaries: Vec, + pub p50_latency_ms: Option, + pub dispersion: Option, +} + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ExperimentResultRowPaginationBody { + pub page: usize, + pub page_size: usize, + pub total_rows: usize, + pub has_more: bool, +} + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ExperimentResultRowPageResponseBody { + pub rows: Vec, + pub pagination: ExperimentResultRowPaginationBody, +} + #[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] #[serde(rename_all = "snake_case")] pub enum DatasetItemSourceBody { @@ -1083,6 +1137,8 @@ pub struct ExperimentResultSlotBody { pub trial_index: u32, pub input: Value, pub expected_output: Option, + /// The one field a list surface needs; task and score statuses remain for drill-down. + pub status: ExperimentSlotStatusBody, pub task_status: ExperimentResultTaskStatusBody, pub execution: Option, pub scores: Vec, @@ -1110,6 +1166,18 @@ pub enum ExperimentResultTaskStatusBody { Error, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum ExperimentSlotStatusBody { + Pending, + Running, + Scoring, + Completed, + Skipped, + TaskFailed, + ScoreFailed, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] #[serde(rename_all = "snake_case")] pub enum ExperimentResultScoreStatusBody { @@ -1182,6 +1250,21 @@ pub struct ExperimentAggregateSummaryBody { pub incomplete: bool, pub incomplete_task_slots: u64, pub incomplete_score_dimensions: u64, + pub error_task_slots: u64, +} + +impl From for ExperimentSlotStatusBody { + fn from(value: ExperimentSlotStatus) -> Self { + match value { + ExperimentSlotStatus::Pending => Self::Pending, + ExperimentSlotStatus::Running => Self::Running, + ExperimentSlotStatus::Scoring => Self::Scoring, + ExperimentSlotStatus::Completed => Self::Completed, + ExperimentSlotStatus::Skipped => Self::Skipped, + ExperimentSlotStatus::TaskFailed => Self::TaskFailed, + ExperimentSlotStatus::ScoreFailed => Self::ScoreFailed, + } + } } impl From for ExperimentResultTaskStatusBody { @@ -1229,6 +1312,7 @@ impl From for ExperimentResultSlotBody { trial_index: value.trial_index, input: value.input, expected_output: value.expected_output, + status: value.status.into(), task_status: value.task_status.into(), execution: value .execution @@ -1305,6 +1389,7 @@ impl From for ExperimentAggregateSummaryBody { incomplete: value.incomplete, incomplete_task_slots: value.incomplete_task_slots, incomplete_score_dimensions: value.incomplete_score_dimensions, + error_task_slots: value.error_task_slots, } } } diff --git a/src/api/management/src/models/playground.rs b/src/api/management/src/models/playground.rs index 01cab66434..1dcaf33547 100644 --- a/src/api/management/src/models/playground.rs +++ b/src/api/management/src/models/playground.rs @@ -57,6 +57,14 @@ impl From for PlaygroundSnapshotResponseBody { // Run // --------------------------------------------------------------------------- +#[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct PlaygroundToolCallBody { + pub id: String, + pub name: String, + pub arguments: String, +} + #[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct PlaygroundMessageBody { @@ -65,6 +73,12 @@ pub struct PlaygroundMessageBody { /// tool-result messages survive. #[serde(default)] pub content: Option, + #[serde(default)] + pub reasoning_content: Option, + #[serde(default)] + pub tool_call_id: Option, + #[serde(default)] + pub tool_calls: Vec, } #[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] diff --git a/src/api/management/src/request/experiments.rs b/src/api/management/src/request/experiments.rs index 68d7b35e6c..69ec22286f 100644 --- a/src/api/management/src/request/experiments.rs +++ b/src/api/management/src/request/experiments.rs @@ -13,7 +13,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use axum::{ extract::{Path, Query}, @@ -25,10 +25,10 @@ use openobserve_core::{ auth::{UserEmail, is_ofga_object_visible}, llm_evaluations::{ datasets, experiment_baseline, experiment_deletion, - experiment_dispersion::{self, NormalizationSpans}, + experiment_dispersion::{self, NormalizationSpans, RowDispersion}, experiment_ingest::{self, IngestError}, - experiment_results, - experiments::{self, ExperimentError}, + experiment_results::{self, ExperimentResultSlot, ExperimentSlotStatus}, + experiments::{self, ExperimentError, PinnedExperimentScorer}, remote_tasks, }, }; @@ -44,12 +44,14 @@ use crate::{ CloneExperimentRequestBody, CreateExperimentRequestBody, CreateExperimentResponseBody, ExperimentBaselineResponseBody, ExperimentDetailQuery, ExperimentDetailResponseBody, ExperimentDispersionSummaryBody, ExperimentPreviewQuery, ExperimentPreviewResponseBody, - ExperimentResponseBody, ExperimentResultPaginationBody, ExperimentResultsResponseBody, - ExperimentRowDetailResponseBody, ExperimentRowNavigationBody, - ExperimentRowSnapshotBody, ExperimentScoreSummaryBody, ExperimentSlotPageQuery, - ExperimentSlotPageResponseBody, ExperimentTaskBody, ListExperimentsResponseBody, - RetryExperimentSlotRequestBody, SubmitExperimentRecordsRequestBody, - SubmitExperimentRecordsResponseBody, + ExperimentResponseBody, ExperimentResultPaginationBody, ExperimentResultRowBody, + ExperimentResultRowPageQuery, ExperimentResultRowPageResponseBody, + ExperimentResultRowPaginationBody, ExperimentResultRowSortBody, + ExperimentResultsResponseBody, ExperimentRowDetailResponseBody, + ExperimentRowNavigationBody, ExperimentRowSnapshotBody, ExperimentScoreSummaryBody, + ExperimentSlotPageQuery, ExperimentSlotPageResponseBody, ExperimentTaskBody, + ListExperimentsResponseBody, RetryExperimentSlotRequestBody, + SubmitExperimentRecordsRequestBody, SubmitExperimentRecordsResponseBody, }, }, }; @@ -168,6 +170,151 @@ fn score_summary_bodies( .collect() } +fn experiment_result_row_status(slots: &[ExperimentResultSlot]) -> ExperimentSlotStatus { + for status in [ + ExperimentSlotStatus::TaskFailed, + ExperimentSlotStatus::ScoreFailed, + ExperimentSlotStatus::Running, + ExperimentSlotStatus::Scoring, + ExperimentSlotStatus::Pending, + ExperimentSlotStatus::Completed, + ExperimentSlotStatus::Skipped, + ] { + if slots.iter().any(|slot| slot.status == status) { + return status; + } + } + ExperimentSlotStatus::Pending +} + +fn group_result_slots(slots: Vec) -> Vec> { + let mut rows = Vec::>::new(); + for slot in slots { + if rows + .last() + .and_then(|row| row.first()) + .is_none_or(|first| first.row_id != slot.row_id) + { + rows.push(Vec::new()); + } + rows.last_mut() + .expect("the current result row was initialized") + .push(slot); + } + rows +} + +fn experiment_result_rows( + slots: Vec, + scorers: &[PinnedExperimentScorer], + dispersions: Vec, + scorer_definitions: &[infra::table::scorers::Scorer], + score_configs: &[infra::table::score_configs::ScoreConfig], +) -> Vec { + let mut dispersions = dispersions + .into_iter() + .map(|dispersion| (dispersion.row_id.clone(), dispersion)) + .collect::>(); + group_result_slots(slots) + .into_iter() + .enumerate() + .filter_map(|(row_index, trials)| { + let first = trials.first()?; + let executions = trials + .iter() + .filter_map(|trial| trial.execution.clone()) + .collect::>(); + let scores = trials + .iter() + .flat_map(|trial| { + trial + .scores + .iter() + .filter_map(|score| score.score.clone()) + .chain(trial.client_scores.iter().cloned()) + }) + .collect::>(); + let summary = + experiment_results::row_result_summary(trials.len(), scorers, &executions, &scores); + let aggregate = experiment_results::aggregate_summary( + &executions, + &summary.task_progress, + &summary.scoring_progress, + ); + let output = (trials.len() == 1) + .then(|| { + first + .execution + .as_ref() + .and_then(|record| record.output.clone()) + }) + .flatten(); + Some(ExperimentResultRowBody { + row_index, + row_id: first.row_id.clone(), + logical_id: first.logical_id.clone(), + input: first.input.clone(), + expected_output: first.expected_output.clone(), + trial_count: trials.len(), + status: experiment_result_row_status(&trials).into(), + output, + score_summaries: score_summary_bodies( + summary.score_summaries, + scorer_definitions, + score_configs, + ), + p50_latency_ms: aggregate.p50_latency_ms, + dispersion: dispersions.remove(&first.row_id).map(Into::into), + }) + }) + .collect() +} + +fn paginate_experiment_result_rows( + mut rows: Vec, + page: usize, + page_size: usize, + sort: ExperimentResultRowSortBody, + high_dispersion_only: bool, +) -> ExperimentResultRowPageResponseBody { + if high_dispersion_only { + rows.retain(|row| { + row.dispersion + .as_ref() + .is_some_and(|dispersion| dispersion.high) + }); + } + if matches!(sort, ExperimentResultRowSortBody::DispersionDesc) { + rows.sort_by(|left, right| { + let left_value = left + .dispersion + .as_ref() + .and_then(|dispersion| dispersion.max_normalized) + .unwrap_or(-1.0); + let right_value = right + .dispersion + .as_ref() + .and_then(|dispersion| dispersion.max_normalized) + .unwrap_or(-1.0); + right_value + .total_cmp(&left_value) + .then_with(|| left.row_index.cmp(&right.row_index)) + }); + } + let total_rows = rows.len(); + let start = page.saturating_sub(1).saturating_mul(page_size); + let rows = rows.into_iter().skip(start).take(page_size).collect(); + ExperimentResultRowPageResponseBody { + rows, + pagination: ExperimentResultRowPaginationBody { + page, + page_size, + total_rows, + has_more: start.saturating_add(page_size) < total_rows, + }, + } +} + /// Scoring Status of one Experiment, derived from the same evidence and the /// same rules the results page uses. async fn derive_scoring_status( @@ -681,29 +828,138 @@ pub async fn compare_experiments( } }; let policy = ComparisonPolicy::from_configs(&score_configs); - MetaHttpResponse::json(ExperimentComparisonResponseBody::from(compare_experiments( - CompareExperimentsInput { - baseline_id: baseline.id, - candidate_id: candidate.id, - dataset_id: baseline.dataset_id, - threshold, - policy: &policy, - scoring: ComparisonScoringState { - baseline: baseline_scoring, - candidate: candidate_scoring, - }, - baseline: ComparisonEvidence { - slots: &baseline_slots, - executions: &baseline_results.executions, - scores: &baseline_results.scores, - }, - candidate: ComparisonEvidence { - slots: &candidate_slots, - executions: &candidate_results.executions, - scores: &candidate_results.scores, - }, + let comparison = compare_experiments(CompareExperimentsInput { + baseline_id: baseline.id, + candidate_id: candidate.id, + dataset_id: baseline.dataset_id, + threshold, + policy: &policy, + scoring: ComparisonScoringState { + baseline: baseline_scoring, + candidate: candidate_scoring, }, - ))) + baseline: ComparisonEvidence { + slots: &baseline_slots, + executions: &baseline_results.executions, + scores: &baseline_results.scores, + }, + candidate: ComparisonEvidence { + slots: &candidate_slots, + executions: &candidate_results.executions, + scores: &candidate_results.scores, + }, + }); + let response = ExperimentComparisonResponseBody::from(comparison) + .with_trial_outputs(&baseline_results.executions, &candidate_results.executions); + MetaHttpResponse::json(response) +} + +/// List pinned Dataset cases with trial-level evidence reduced to row aggregates. +#[utoipa::path( + get, + path = "/{org_id}/experiments/{experiment_id}/rows", + context_path = "/api", + tag = "Experiments", + operation_id = "ListExperimentResultRows", + security(("Authorization" = [])), + params( + ("org_id" = String, Path, description = "Organization name"), + ("experiment_id" = String, Path, description = "Experiment ID"), + ExperimentResultRowPageQuery, + ), + responses( + (status = 200, description = "Pinned result rows", body = ExperimentResultRowPageResponseBody), + (status = 400, description = "Invalid pagination"), + (status = 403, description = "Experiment is not accessible"), + (status = 404, description = "Experiment not found"), + ) +)] +pub async fn list_experiment_result_rows( + Path((org_id, experiment_id)): Path<(String, String)>, + Headers(user): Headers, + Query(query): Query, +) -> Response { + if let Err(response) = + require_experiment_visibility(&org_id, &experiment_id, &user.user_id, "GET").await + { + return response; + } + let page = query.page.unwrap_or(1); + let page_size = query + .page_size + .unwrap_or(openobserve_core::llm_evaluations::experiment_runner::DEFAULT_RESULT_PAGE_SIZE); + if page == 0 + || page_size == 0 + || page_size > openobserve_core::llm_evaluations::experiment_runner::MAX_RESULT_PAGE_SIZE + { + return MetaHttpResponse::bad_request("Invalid Experiment row pagination"); + } + let experiment = match experiments::get(&org_id, &experiment_id).await { + Ok(experiment) => experiment, + Err(error) => return experiment_error_response(error), + }; + if let Err(error) = + openobserve_core::self_reporting::llm_scores_schema::ensure_llm_scores_stream_initialized( + &org_id, + ) + .await + { + log::error!("[Experiment] failed to initialize score stream for {experiment_id}: {error}"); + return MetaHttpResponse::internal_error("Failed to load Experiment rows"); + } + let slots = match experiments::slot_set_existing(&org_id, &experiment).await { + Ok(slots) => slots, + Err(error) => return experiment_error_response(error), + }; + let evidence = match openobserve_core::llm_evaluations::experiment_runner::results(&experiment) + .await + { + Ok(evidence) => evidence, + Err(error) => { + log::error!("[Experiment] failed to load row evidence for {experiment_id}: {error}"); + return MetaHttpResponse::internal_error("Failed to load Experiment rows"); + } + }; + let score_configs = match infra::table::score_configs::get_all_by_org(&org_id).await { + Ok(configs) => configs, + Err(error) => { + log::error!("[Experiment] failed to load Score Configs for {experiment_id}: {error}"); + return MetaHttpResponse::internal_error("Failed to load Experiment rows"); + } + }; + let scorer_definitions = match experiments::scorer_definitions(&org_id, &experiment).await { + Ok(definitions) => definitions, + Err(error) => { + log::error!( + "[Experiment] failed to load Scorer definitions for {experiment_id}: {error}" + ); + return MetaHttpResponse::internal_error("Failed to load Experiment rows"); + } + }; + let scorers = experiment.scorers.clone(); + let dispersions = experiment_dispersion::row_dispersions( + &experiments::row_keys(&slots), + &evidence.scores, + &scorers, + &NormalizationSpans::from_configs(&score_configs), + ); + let result_slots = + experiment_results::result_slots(slots, &evidence.executions, &evidence.scores, &scorers); + let rows = experiment_result_rows( + result_slots, + &scorers, + dispersions, + &scorer_definitions, + &score_configs, + ); + let response = paginate_experiment_result_rows( + rows, + page, + page_size, + query.sort.unwrap_or_default(), + query.high_dispersion_only.unwrap_or(false), + ); + MetaHttpResponse::json(response) } /// Load one pinned dataset row and every trial's current execution and score evidence. @@ -1235,6 +1491,91 @@ pub async fn finalize_experiment( mod tests { use super::*; + fn result_slot( + row_id: &str, + trial_index: u32, + status: ExperimentSlotStatus, + ) -> ExperimentResultSlot { + ExperimentResultSlot { + row_id: row_id.to_string(), + logical_id: format!("case-{row_id}"), + trial_index, + input: serde_json::json!("question"), + expected_output: None, + status, + task_status: + openobserve_core::llm_evaluations::experiment_results::ExperimentResultTaskStatus::Pending, + execution: None, + scores: Vec::new(), + client_scores: Vec::new(), + } + } + + fn result_row(row_index: usize, max_normalized: f64, high: bool) -> ExperimentResultRowBody { + let row_id = format!("row-{row_index}"); + ExperimentResultRowBody { + row_index, + row_id: row_id.clone(), + logical_id: format!("case-{row_index}"), + input: serde_json::json!("question"), + expected_output: None, + trial_count: 2, + status: crate::models::experiments::ExperimentSlotStatusBody::Completed, + output: None, + score_summaries: Vec::new(), + p50_latency_ms: None, + dispersion: Some(crate::models::experiments::ExperimentRowDispersionBody { + row_id, + logical_id: format!("case-{row_index}"), + dimensions: Vec::new(), + max_normalized: Some(max_normalized), + high, + outlier_trial_index: None, + }), + } + } + + #[test] + fn result_rows_group_trials_and_surface_the_most_actionable_status() { + let rows = group_result_slots(vec![ + result_slot("one", 0, ExperimentSlotStatus::Completed), + result_slot("one", 1, ExperimentSlotStatus::TaskFailed), + result_slot("two", 0, ExperimentSlotStatus::Pending), + ]); + + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].len(), 2); + assert!(matches!( + experiment_result_row_status(&rows[0]), + ExperimentSlotStatus::TaskFailed + )); + } + + #[test] + fn result_row_pagination_orders_and_filters_by_dispersion() { + let rows = vec![result_row(0, 0.1, false), result_row(1, 0.8, true)]; + let sorted = paginate_experiment_result_rows( + rows, + 1, + 100, + ExperimentResultRowSortBody::DispersionDesc, + false, + ); + + assert_eq!(sorted.rows[0].row_index, 1); + assert_eq!(sorted.pagination.total_rows, 2); + + let filtered = paginate_experiment_result_rows( + sorted.rows, + 1, + 100, + ExperimentResultRowSortBody::Dataset, + true, + ); + assert_eq!(filtered.rows.len(), 1); + assert_eq!(filtered.rows[0].row_index, 1); + } + #[test] fn score_summary_uses_the_pinned_score_config_name_before_scores_exist() { let summaries = vec![experiment_results::ExperimentScoreSummary { diff --git a/src/api/management/src/request/playground.rs b/src/api/management/src/request/playground.rs index fdc09b38f5..55e13dd65e 100644 --- a/src/api/management/src/request/playground.rs +++ b/src/api/management/src/request/playground.rs @@ -14,6 +14,8 @@ //! help a person improve. Sharing a snapshot is the only endpoint here that //! stores anything. +use std::collections::HashSet; + use axum::{ body::Body, extract::{Path, rejection::JsonRejection}, @@ -29,8 +31,8 @@ use openobserve_core::{ experiment_runner::render_prompt, playground::{self, PlaygroundError}, provider::{ - PreparedProvider, ProviderChatMessage, ProviderChatParams, ProviderStreamStartError, - RawProviderConfig, + PreparedProvider, ProviderChatMessage, ProviderChatParams, ProviderChatToolCall, + ProviderStreamStartError, RawProviderConfig, }, provider_stream::ProviderChatDelta, sync_scoring::{self, SyncScoringError}, @@ -314,7 +316,7 @@ fn render_messages( column: &PlaygroundColumnBody, input: &Value, ) -> Result, Response> { - column + let messages = column .messages .iter() .map(|message| { @@ -333,9 +335,85 @@ fn render_messages( Ok(ProviderChatMessage { role: message.role.clone(), content, + reasoning_content: message.reasoning_content.clone(), + tool_call_id: message.tool_call_id.clone(), + tool_calls: message + .tool_calls + .iter() + .map(|call| ProviderChatToolCall { + id: call.id.clone(), + name: call.name.clone(), + arguments: call.arguments.clone(), + }) + .collect(), }) }) - .collect() + .collect::, Response>>()?; + validate_tool_messages(&messages)?; + Ok(messages) +} + +fn validate_tool_messages(messages: &[ProviderChatMessage]) -> Result<(), Response> { + let mut call_ids = HashSet::new(); + let mut result_ids = HashSet::new(); + for message in messages { + if message.reasoning_content.is_some() && message.role != "assistant" { + return Err(MetaHttpResponse::bad_request( + "Only assistant messages may contain reasoningContent", + )); + } + if !message.tool_calls.is_empty() { + if message.role != "assistant" { + return Err(MetaHttpResponse::bad_request( + "Only assistant messages may contain tool calls", + )); + } + for call in &message.tool_calls { + if call.id.trim().is_empty() || call.name.trim().is_empty() { + return Err(MetaHttpResponse::bad_request( + "Tool calls require a non-empty id and name", + )); + } + if !serde_json::from_str::(&call.arguments) + .is_ok_and(|value| value.is_object()) + { + return Err(MetaHttpResponse::bad_request(format!( + "Tool call '{}' arguments must be a JSON object", + call.id + ))); + } + if !call_ids.insert(call.id.as_str()) { + return Err(MetaHttpResponse::bad_request(format!( + "Tool call id '{}' is duplicated", + call.id + ))); + } + } + } + + if message.role == "tool" { + let Some(call_id) = message.tool_call_id.as_deref() else { + return Err(MetaHttpResponse::bad_request( + "Tool result messages require toolCallId", + )); + }; + if !call_ids.contains(call_id) { + return Err(MetaHttpResponse::bad_request(format!( + "Tool result references unknown call id '{call_id}'" + ))); + } + if !result_ids.insert(call_id) { + return Err(MetaHttpResponse::bad_request(format!( + "Tool call id '{call_id}' has more than one result" + ))); + } + } else if message.tool_call_id.is_some() { + return Err(MetaHttpResponse::bad_request( + "Only tool result messages may contain toolCallId", + )); + } + } + Ok(()) } /// Split the column's params into the fields the provider names explicitly and @@ -461,7 +539,15 @@ pub async fn run_playground_cell( "type": "rendered", "messages": messages .iter() - .map(|message| json!({"role": message.role, "content": message.content})) + .map(|message| { + json!({ + "role": message.role, + "content": message.content, + "reasoningContent": message.reasoning_content, + "toolCallId": message.tool_call_id, + "toolCalls": message.tool_calls, + }) + }) .collect::>(), }); @@ -484,6 +570,7 @@ pub async fn run_playground_cell( "type": "done", "model": result.model_used, "latencyMs": result.latency_ms, + "reasoningContent": result.reasoning_content, "usage": { "promptTokens": result.prompt_tokens, "completionTokens": result.completion_tokens, @@ -571,7 +658,7 @@ pub async fn score_playground_cell( #[cfg(test)] mod tests { use super::*; - use crate::models::playground::PlaygroundMessageBody; + use crate::models::playground::{PlaygroundMessageBody, PlaygroundToolCallBody}; fn column(params: Value) -> PlaygroundColumnBody { PlaygroundColumnBody { @@ -612,10 +699,16 @@ mod tests { PlaygroundMessageBody { role: "user".to_string(), content: Some(json!("Hello {{name}}")), + reasoning_content: None, + tool_call_id: None, + tool_calls: Vec::new(), }, PlaygroundMessageBody { role: "user".to_string(), content: Some(structured.clone()), + reasoning_content: None, + tool_call_id: None, + tool_calls: Vec::new(), }, ]; @@ -623,4 +716,38 @@ mod tests { assert_eq!(rendered[0].content, json!("Hello Ada")); assert_eq!(rendered[1].content, structured); } + + #[test] + fn tool_call_metadata_survives_message_rendering() { + let mut column = column(json!({})); + column.messages = vec![ + PlaygroundMessageBody { + role: "assistant".to_string(), + content: Some(json!("")), + reasoning_content: Some("I should look up this order.".to_string()), + tool_call_id: None, + tool_calls: vec![PlaygroundToolCallBody { + id: "call_1".to_string(), + name: "lookup_order".to_string(), + arguments: r#"{"order_id":"123"}"#.to_string(), + }], + }, + PlaygroundMessageBody { + role: "tool".to_string(), + content: Some(json!({"status": "shipped"})), + reasoning_content: None, + tool_call_id: Some("call_1".to_string()), + tool_calls: Vec::new(), + }, + ]; + + let rendered = render_messages(&column, &Value::Null).unwrap(); + assert_eq!(rendered[0].tool_calls[0].id, "call_1"); + assert_eq!(rendered[0].tool_calls[0].name, "lookup_order"); + assert_eq!( + rendered[0].reasoning_content.as_deref(), + Some("I should look up this order.") + ); + assert_eq!(rendered[1].tool_call_id.as_deref(), Some("call_1")); + } } diff --git a/src/api/search/src/traces/schema_compat.rs b/src/api/search/src/traces/schema_compat.rs index eb074dcd59..dee0a16288 100644 --- a/src/api/search/src/traces/schema_compat.rs +++ b/src/api/search/src/traces/schema_compat.rs @@ -19,7 +19,9 @@ //! //! - **New (`gen_ai_*`)** — emitted since PR #11626 ("refactor llm attribute trace"). Tokens, cost, //! model, user, session id all use the OTEL Gen-AI spec names: `gen_ai_usage_input_tokens`, -//! `gen_ai_response_model`, `user.id` (queried as `user.id`), `gen_ai_conversation_id`, etc. +//! `gen_ai_response_model`, `user.id`, `gen_ai_conversation_id`, etc. Ingestion flattens every +//! attribute key with `flatten::format_key` (dot → underscore), so the spec's `user.id` is stored +//! — and must be queried — as the `user_id` column. //! - **Legacy (`_o2_llm`)** — pre-PR #11626. Tokens live under //! `llm_usage_tokens_{input,output,total}`, cost under `llm_usage_cost_total`, model under //! `llm_model_name`, and user/session ids under `llm_user_id`/`llm_session_id`. @@ -39,7 +41,6 @@ use config::meta::{ REQUIRED_GEN_AI_FIELDS, REQUIRED_LLM_FIELDS, }, }; -use openobserve_core::traces::otel::attributes::OtelAttributes; /// Column names that vary between the new and legacy LLM schemas. /// @@ -55,10 +56,14 @@ pub(super) struct LlmColumns { impl LlmColumns { /// Column layout for the current (`gen_ai_*`) schema. + /// + /// These are stored column names, not raw OTEL attribute keys: ingestion + /// flattens `user.id` to `user_id` (see `GEN_AI_SCHEMA_FIELDS` in + /// `db/schema.rs`, which provisions exactly these columns). pub(super) const fn current() -> Self { Self { session_id: "gen_ai_conversation_id", - user_id: OtelAttributes::USER_ID, + user_id: "user_id", } } @@ -241,6 +246,7 @@ pub(super) fn validate_llm_schema( #[cfg(test)] mod tests { use arrow_schema::{DataType, Field, Schema}; + use openobserve_core::traces::otel::attributes::OtelAttributes; use super::*; @@ -287,10 +293,10 @@ mod tests { fn current_layout_uses_gen_ai_columns() { let cols = LlmColumns::current(); assert_eq!(cols.session_id, "gen_ai_conversation_id"); - // user.id is the OTEL attribute name — the search layer resolves it - // to the flattened column. - assert_eq!(cols.user_id, OtelAttributes::USER_ID); - assert_eq!(cols.user_id, "user.id"); + // The stored column is the Gen-AI spec's `user.id` attribute after the + // dot → underscore flattening ingestion applies to every key. + assert_eq!(cols.user_id, "user_id"); + assert_eq!(cols.user_id, OtelAttributes::USER_ID.replace('.', "_")); } #[test] diff --git a/src/api/search/src/traces/session.rs b/src/api/search/src/traces/session.rs index fa16c40543..c3d83c38c0 100644 --- a/src/api/search/src/traces/session.rs +++ b/src/api/search/src/traces/session.rs @@ -934,6 +934,7 @@ fn build_session_trace_details_sql( service_key_expr: &str, trace_id_predicate: &str, ) -> String { + let user_id_col = validated.columns.user_id; let (root_service_name_expr, root_operation_name_expr) = if has_ref_parent_id { ( "max(CASE WHEN reference_parent_span_id IS NULL OR reference_parent_span_id = '' THEN service_name END)", @@ -950,7 +951,9 @@ fn build_session_trace_details_sql( {root_service_name_expr} AS root_service_name, \ {root_operation_name_expr} AS root_operation_name, \ first_value(service_name ORDER BY {TIMESTAMP_COL_NAME} ASC) AS first_service_name, \ - first_value(operation_name ORDER BY {TIMESTAMP_COL_NAME} ASC) AS first_operation_name" + first_value(operation_name ORDER BY {TIMESTAMP_COL_NAME} ASC) AS first_operation_name, \ + array_agg(DISTINCT {user_id_col}) \ + FILTER (WHERE {user_id_col} IS NOT NULL AND {user_id_col} != '') AS user_ids" ); if validated.has_gen_ai { @@ -1054,6 +1057,20 @@ fn build_session_trace_response_item( item: &json::Value, ) -> Option<(String, i64, SessionTraceResponseItem)> { let tid = item.get("trace_id")?.as_str()?.to_string(); + let mut user_ids = item + .get("user_ids") + .and_then(json::Value::as_array) + .map(|values| { + values + .iter() + .filter_map(json::Value::as_str) + .filter(|value| !value.is_empty()) + .map(String::from) + .collect::>() + }) + .unwrap_or_default(); + user_ids.sort(); + user_ids.dedup(); let trace_start_time = json::get_int_value(item.get("trace_start_time").unwrap_or_default()); let trace_end_time = json::get_int_value(item.get("trace_end_time").unwrap_or_default()); let span_count = json::get_int_value(item.get("span_count").unwrap_or_default()); @@ -1110,6 +1127,7 @@ fn build_session_trace_response_item( let hit = SessionTraceResponseItem { trace_id: tid.clone(), + user_ids, start_time: trace_start_time, end_time: trace_end_time, duration: computed_duration, @@ -1253,13 +1271,7 @@ fn build_latest_sessions_sql( validated: &super::schema_compat::ValidatedLlmSchema, ) -> String { let session_id_col = validated.columns.session_id; - // Trace ingestion normalizes OTEL `user.id` to the stored `user_id` - // column. Keep the legacy schema's explicit `llm_user_id` name. - let user_id_col = if validated.has_gen_ai { - "user_id" - } else { - validated.columns.user_id - }; + let user_id_col = validated.columns.user_id; let session_ids_sql = session_ids .iter() .map(|session_id| format!("'{}'", session_id.replace('\'', "''"))) @@ -1419,6 +1431,7 @@ fn normalize_latest_session_hits( #[derive(Debug, Serialize)] struct SessionTraceResponseItem { trace_id: String, + user_ids: Vec, start_time: i64, end_time: i64, duration: i64, @@ -1581,6 +1594,7 @@ mod tests { assert!( sql.contains("sum(CASE WHEN span_status = 'ERROR' THEN 1 ELSE 0 END) AS error_count") ); + assert!(sql.contains("array_agg(DISTINCT user_id)")); assert!(!sql.contains("gen_ai_conversation_id")); } @@ -1598,6 +1612,7 @@ mod tests { "root_operation_name": "gen_ai.chat.completions", "first_service_name": "o2-ai", "first_operation_name": "fallback-op", + "user_ids": ["user-b", "user-a", "user-b", ""], "gen_ai_usage_details_input": 10, "gen_ai_usage_details_output": 20, "gen_ai_usage_details_total": 30, @@ -1610,10 +1625,52 @@ mod tests { assert_eq!(service_count, 1); assert_eq!(hit.spans, [3, 1]); + assert_eq!( + hit.user_ids, + vec!["user-a".to_string(), "user-b".to_string()] + ); assert_eq!(hit.service_name[0].service_name, "o2-ai"); assert_eq!(hit.models, vec!["claude-sonnet-4-6".to_string()]); } + #[test] + fn gen_ai_session_sql_uses_flattened_user_id_column() { + // Ingestion flattens the Gen-AI `user.id` attribute to a `user_id` + // column, so the SQL must never emit the dotted attribute name — + // DataFusion would read `user.id` as `.`. + let validated = super::super::schema_compat::ValidatedLlmSchema::fallback(true); + let details_sql = build_session_trace_details_sql( + "default", + &validated, + false, + "service_name", + "\"trace_id\" IN ('trace-1')", + ); + let sessions_sql = + build_latest_sessions_sql("default", &["session-1".to_string()], &validated); + + for sql in [&details_sql, &sessions_sql] { + assert!(sql.contains("array_agg(DISTINCT user_id)")); + assert!(sql.contains("user_id IS NOT NULL AND user_id != ''")); + assert!(!sql.contains("user.id")); + } + } + + #[test] + fn legacy_session_trace_details_sql_uses_legacy_user_id_column() { + let validated = super::super::schema_compat::ValidatedLlmSchema::fallback(false); + let sql = build_session_trace_details_sql( + "default", + &validated, + false, + "service_name", + "\"trace_id\" IN ('trace-1')", + ); + + assert!(sql.contains("array_agg(DISTINCT llm_user_id)")); + assert!(!sql.contains("array_agg(DISTINCT user_id)")); + } + #[test] fn shared_trace_ids_from_hits_preserves_exact_ids_and_deduplicates() { let hits = vec![ diff --git a/web/src/composables/useFullscreenBox.spec.ts b/web/src/composables/useFullscreenBox.spec.ts new file mode 100644 index 0000000000..d5a75f2256 --- /dev/null +++ b/web/src/composables/useFullscreenBox.spec.ts @@ -0,0 +1,124 @@ +// Copyright 2026 OpenObserve Inc. + +// @vitest-environment jsdom + +import { describe, expect, it, vi, beforeEach, afterEach } from "vitest"; +import { defineComponent, nextTick } from "vue"; +import { mount } from "@vue/test-utils"; +import { useFullscreenBox } from "./useFullscreenBox"; + +// jsdom implements neither the Fullscreen API nor a settable +// `document.fullscreenElement` — both are stood up manually so `toggle()`'s +// real call into `Element.requestFullscreen()` has something to hit, and so a +// test can simulate the browser firing `fullscreenchange` (e.g. on Escape). +let currentFullscreenElement: Element | null = null; + +function fireFullscreenChange(target: Element | null) { + currentFullscreenElement = target; + document.dispatchEvent(new Event("fullscreenchange")); +} + +beforeEach(() => { + currentFullscreenElement = null; + Object.defineProperty(document, "fullscreenElement", { + configurable: true, + get: () => currentFullscreenElement, + }); + Element.prototype.requestFullscreen = vi.fn(function (this: Element) { + fireFullscreenChange(this); + return Promise.resolve(); + }); + document.exitFullscreen = vi.fn(() => { + fireFullscreenChange(null); + return Promise.resolve(); + }); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +function mountHarness() { + let exposed: ReturnType; + const wrapper = mount( + defineComponent({ + setup() { + exposed = useFullscreenBox(); + return () => null; + }, + }), + ); + return { + wrapper, + get exposed() { + return exposed; + }, + }; +} + +describe("useFullscreenBox", () => { + it("starts with nothing fullscreen", () => { + const { exposed } = mountHarness(); + expect(exposed.fullscreenEl.value).toBeNull(); + }); + + it("requests fullscreen on the given element and reflects it once the browser confirms", async () => { + const { exposed } = mountHarness(); + const box = document.createElement("div"); + + exposed.toggle(box); + await nextTick(); + + expect(box.requestFullscreen).toHaveBeenCalled(); + expect(exposed.fullscreenEl.value).toBe(box); + }); + + it("a second box's toggle replaces the first — only one element is ever fullscreen", async () => { + const { exposed } = mountHarness(); + const boxA = document.createElement("div"); + const boxB = document.createElement("div"); + + exposed.toggle(boxA); + await nextTick(); + expect(exposed.fullscreenEl.value).toBe(boxA); + + // The real browser call is exitFullscreen() then requestFullscreen() on + // the new target, not a direct swap — simulate that sequence. + await document.exitFullscreen(); + exposed.toggle(boxB); + await nextTick(); + + expect(exposed.fullscreenEl.value).toBe(boxB); + }); + + it("clears when the browser exits fullscreen outside any click — e.g. Escape", async () => { + const { exposed } = mountHarness(); + const box = document.createElement("div"); + exposed.toggle(box); + await nextTick(); + expect(exposed.fullscreenEl.value).toBe(box); + + fireFullscreenChange(null); + await nextTick(); + + expect(exposed.fullscreenEl.value).toBeNull(); + }); + + it("does nothing for a null element rather than throwing", () => { + const { exposed } = mountHarness(); + expect(() => exposed.toggle(null)).not.toThrow(); + expect(exposed.fullscreenEl.value).toBeNull(); + }); + + it("stops listening once the owning component unmounts", async () => { + const { wrapper, exposed } = mountHarness(); + wrapper.unmount(); + + fireFullscreenChange(document.createElement("div")); + await nextTick(); + + // No error from a listener touching a torn-down component, and the last + // value it saw stays put — there is nothing left to update it. + expect(exposed.fullscreenEl.value).toBeNull(); + }); +}); diff --git a/web/src/composables/useFullscreenBox.ts b/web/src/composables/useFullscreenBox.ts new file mode 100644 index 0000000000..b3e5cb6c13 --- /dev/null +++ b/web/src/composables/useFullscreenBox.ts @@ -0,0 +1,51 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +import { onBeforeUnmount, onMounted, ref } from "vue"; +import { toggleFullscreen } from "@/utils/dom"; + +/** + * Fullscreen state for a page/drawer with several independently-expandable + * content boxes, sharing ONE `fullscreenEl` ref rather than one boolean per + * box — the Fullscreen API only ever has a single fullscreen element at a + * time, so a per-box boolean could desync from reality (two boxes reading + * "fullscreen" at once, or none, after an Escape-key exit browsers fire + * outside any click handler). + * + * const { fullscreenEl, toggle } = useFullscreenBox(); + *
+ *