diff --git a/crates/core/src/codec/anthropic.rs b/crates/core/src/codec/anthropic.rs index 837e6333c..b7f8b84b7 100644 --- a/crates/core/src/codec/anthropic.rs +++ b/crates/core/src/codec/anthropic.rs @@ -283,6 +283,75 @@ fn encode_anthropic_tools(tools: &[ToolDefinition]) -> Vec { .collect() } +fn anthropic_text_message(content_blocks: Option<&[Json]>) -> Option { + let text_parts: Vec<&str> = content_blocks + .map(|blocks| blocks.iter().filter_map(anthropic_text_block).collect()) + .unwrap_or_default(); + + (!text_parts.is_empty()).then(|| MessageContent::Text(text_parts.join("\n"))) +} + +fn anthropic_text_block(block: &Json) -> Option<&str> { + if block.get("type")?.as_str()? != "text" { + return None; + } + block.get("text")?.as_str() +} + +fn anthropic_tool_calls(content_blocks: Option<&[Json]>) -> Option> { + let tool_calls: Vec = content_blocks + .map(|blocks| { + blocks + .iter() + .filter_map(anthropic_tool_call_block) + .collect() + }) + .unwrap_or_default(); + + (!tool_calls.is_empty()).then_some(tool_calls) +} + +fn anthropic_tool_call_block(block: &Json) -> Option { + if block.get("type")?.as_str()? != "tool_use" { + return None; + } + Some(ResponseToolCall { + id: block.get("id")?.as_str()?.to_string(), + name: block.get("name")?.as_str()?.to_string(), + // CRITICAL: input is already parsed JSON -- clone directly. + arguments: block.get("input")?.clone(), + }) +} + +fn anthropic_usage( + raw_usage: Option, + model_for_pricing: Option<&str>, +) -> Option { + let model_provider = infer_model_provider("anthropic", model_for_pricing); + raw_usage.map(|u| { + let prompt = u.input_tokens; + let completion = u.output_tokens; + let mut usage = Usage { + prompt_tokens: prompt, + completion_tokens: completion, + // Anthropic does not supply total_tokens; compute it. + total_tokens: match (prompt, completion) { + (Some(p), Some(c)) => Some(p + c), + _ => None, + }, + cache_read_tokens: u.cache_read_input_tokens, + cache_write_tokens: u.cache_creation_input_tokens, + cost: provider_reported_cost(u.provider_cost, u.cost), + }; + if usage.cost.is_none() { + usage.cost = model_for_pricing.and_then(|model| { + estimate_cost_for_provider(model_provider.as_deref(), model, &usage) + }); + } + usage + }) +} + // --------------------------------------------------------------------------- // LlmResponseCodec implementation // --------------------------------------------------------------------------- @@ -292,91 +361,16 @@ impl LlmResponseCodec for AnthropicMessagesCodec { let raw: RawAnthropicResponse = serde_json::from_value(response.clone()) .map_err(|e| FlowError::Internal(format!("Anthropic Messages response decode: {e}")))?; - // Process content blocks. - let content_blocks = raw.content.as_ref(); - - // Extract text from all "text" blocks, concatenated with newline. - let text_parts: Vec<&str> = content_blocks - .map(|blocks| { - blocks - .iter() - .filter_map(|block| { - let block_type = block.get("type")?.as_str()?; - if block_type == "text" { - block.get("text")?.as_str() - } else { - None - } - }) - .collect() - }) - .unwrap_or_default(); - - let message = if text_parts.is_empty() { - None - } else { - Some(MessageContent::Text(text_parts.join("\n"))) - }; - + let content_blocks = raw.content.as_deref(); + let message = anthropic_text_message(content_blocks); // Extract tool_use blocks (only "tool_use" type, NOT mcp_tool_use or server_tool_use). - let tool_calls: Vec = content_blocks - .map(|blocks| { - blocks - .iter() - .filter_map(|block| { - let block_type = block.get("type")?.as_str()?; - if block_type == "tool_use" { - let id = block.get("id")?.as_str()?.to_string(); - let name = block.get("name")?.as_str()?.to_string(); - // CRITICAL: input is already parsed JSON -- clone directly. - let arguments = block.get("input")?.clone(); - Some(ResponseToolCall { - id, - name, - arguments, - }) - } else { - None - } - }) - .collect() - }) - .unwrap_or_default(); - - let tool_calls = if tool_calls.is_empty() { - None - } else { - Some(tool_calls) - }; + let tool_calls = anthropic_tool_calls(content_blocks); // Map stop_reason to FinishReason. let finish_reason = raw.stop_reason.as_deref().map(map_anthropic_stop_reason); // Map usage. - let model_for_pricing = raw.model.as_deref(); - let model_provider = infer_model_provider("anthropic", model_for_pricing); - let usage = raw.usage.map(|u| { - let prompt = u.input_tokens; - let completion = u.output_tokens; - let mut usage = Usage { - prompt_tokens: prompt, - completion_tokens: completion, - // Anthropic does not supply total_tokens; compute it. - total_tokens: match (prompt, completion) { - (Some(p), Some(c)) => Some(p + c), - _ => None, - }, - cache_read_tokens: u.cache_read_input_tokens, - cache_write_tokens: u.cache_creation_input_tokens, - cost: provider_reported_cost(u.provider_cost, u.cost), - }; - if usage.cost.is_none() { - usage.cost = model_for_pricing.and_then(|model| { - estimate_cost_for_provider(model_provider.as_deref(), model, &usage) - }); - } - usage - }); + let usage = anthropic_usage(raw.usage, raw.model.as_deref()); // Build API-specific fields: all content blocks + stop_sequence. let api_specific_content_blocks = raw.content.clone(); diff --git a/crates/core/src/observability/atif.rs b/crates/core/src/observability/atif.rs index 830ecbb65..4d4d1f31c 100644 --- a/crates/core/src/observability/atif.rs +++ b/crates/core/src/observability/atif.rs @@ -440,38 +440,52 @@ fn unwrap_llm_request(input: &Json) -> Json { /// Tool-call-only responses use an empty string message and keep the full /// response under `Step.extra.llm_response`. fn extract_llm_response_message(output: &Json) -> Json { - if let Some(obj) = output.as_object() { - if let Some(content) = non_null_object_field(obj, "content") { - if let Some(message) = anthropic_messages_content_message(output, &content) { - return message; - } - return atif_content_value(&content); - } - if let Some(content) = obj - .get("assistant_message") - .and_then(Json::as_object) - .and_then(|assistant| non_null_object_field(assistant, "content")) - { - return atif_content_value(&content); - } - if let Some(content) = raw_response_message_field(output, "content") - && !content.is_null() - { - return atif_content_value(content); - } - if let Some(answer) = non_null_object_field(obj, "answer") { - return atif_content_value(&answer); - } - if let Some(content) = openai_responses_output_message(output) { - return content; - } - if tool_call_array(output).is_some() { - return empty_message(); - } + let Some(obj) = output.as_object() else { + return atif_content_value(output); + }; + + if let Some(message) = extract_object_llm_response_message(output, obj) { + return message; } + atif_content_value(output) } +fn extract_object_llm_response_message( + output: &Json, + obj: &serde_json::Map, +) -> Option { + if let Some(content) = non_null_object_field(obj, "content") { + return Some(extract_content_message(output, &content)); + } + if let Some(content) = assistant_message_content(obj) { + return Some(atif_content_value(&content)); + } + if let Some(content) = raw_response_message_field(output, "content") + && !content.is_null() + { + return Some(atif_content_value(content)); + } + if let Some(answer) = non_null_object_field(obj, "answer") { + return Some(atif_content_value(&answer)); + } + if let Some(content) = openai_responses_output_message(output) { + return Some(content); + } + tool_call_array(output).map(|_| empty_message()) +} + +fn extract_content_message(output: &Json, content: &Json) -> Json { + anthropic_messages_content_message(output, content) + .unwrap_or_else(|| atif_content_value(content)) +} + +fn assistant_message_content(obj: &serde_json::Map) -> Option { + obj.get("assistant_message") + .and_then(Json::as_object) + .and_then(|assistant| non_null_object_field(assistant, "content")) +} + fn non_null_object_field(obj: &serde_json::Map, key: &str) -> Option { obj.get(key).filter(|value| !value.is_null()).cloned() } @@ -1163,48 +1177,51 @@ fn event_extra(event: &Event) -> Json { /// Always returns `Some(AtifFinalMetrics)` with `total_steps` set. Each token /// or cost total is populated only when at least one step provides that field. fn compute_final_metrics(steps: &[AtifStep]) -> Option { - let mut total_prompt: u64 = 0; - let mut total_completion: u64 = 0; - let mut total_cached: u64 = 0; - let mut total_cost: f64 = 0.0; - let mut has_prompt = false; - let mut has_completion = false; - let mut has_cached = false; - let mut has_cost = false; - - for step in steps { - if let Some(m) = &step.metrics { - if let Some(prompt_tokens) = m.prompt_tokens { - has_prompt = true; - total_prompt += prompt_tokens; - } - if let Some(completion_tokens) = m.completion_tokens { - has_completion = true; - total_completion += completion_tokens; - } - if let Some(cached_tokens) = m.cached_tokens { - has_cached = true; - total_cached += cached_tokens; - } - if let Some(cost) = m.cost_usd { - has_cost = true; - total_cost += cost; - } + let mut totals = FinalMetricsTotals::default(); + for metrics in steps.iter().filter_map(|step| step.metrics.as_ref()) { + totals.add(metrics); + } + Some(totals.into_final_metrics(steps.len())) +} + +#[derive(Default)] +struct FinalMetricsTotals { + prompt_tokens: Option, + completion_tokens: Option, + cached_tokens: Option, + cost_usd: Option, +} + +impl FinalMetricsTotals { + fn add(&mut self, metrics: &AtifMetrics) { + add_u64_total(&mut self.prompt_tokens, metrics.prompt_tokens); + add_u64_total(&mut self.completion_tokens, metrics.completion_tokens); + add_u64_total(&mut self.cached_tokens, metrics.cached_tokens); + add_f64_total(&mut self.cost_usd, metrics.cost_usd); + } + + fn into_final_metrics(self, step_count: usize) -> AtifFinalMetrics { + AtifFinalMetrics { + total_prompt_tokens: self.prompt_tokens, + total_completion_tokens: self.completion_tokens, + total_cached_tokens: self.cached_tokens, + total_cost_usd: self.cost_usd, + total_steps: Some(step_count as u64), + extra: None, } } +} - Some(AtifFinalMetrics { - total_prompt_tokens: if has_prompt { Some(total_prompt) } else { None }, - total_completion_tokens: if has_completion { - Some(total_completion) - } else { - None - }, - total_cached_tokens: if has_cached { Some(total_cached) } else { None }, - total_cost_usd: if has_cost { Some(total_cost) } else { None }, - total_steps: Some(steps.len() as u64), - extra: None, - }) +fn add_u64_total(total: &mut Option, value: Option) { + if let Some(value) = value { + *total = Some(total.unwrap_or(0) + value); + } +} + +fn add_f64_total(total: &mut Option, value: Option) { + if let Some(value) = value { + *total = Some(total.unwrap_or(0.0) + value); + } } // --------------------------------------------------------------------------- diff --git a/crates/core/src/observability/atof.rs b/crates/core/src/observability/atof.rs index b307cc289..221c10ceb 100644 --- a/crates/core/src/observability/atof.rs +++ b/crates/core/src/observability/atof.rs @@ -778,40 +778,16 @@ async fn run_ndjson_endpoint( config: AtofEndpointConfig, mut rx: tokio::sync::mpsc::UnboundedReceiver, ) { - let client = match reqwest::Client::builder() - .connect_timeout(Duration::from_millis(config.timeout_millis)) - .default_headers(match build_header_map(&config.headers) { - Ok(headers) => headers, - Err(error) => { - eprintln!("nemo_relay: ATOF endpoint[{index}] disabled: {error}"); - drain_closed(rx).await; - return; - } - }) - .build() - { + let client = match build_ndjson_client(&config) { Ok(client) => client, Err(error) => { - eprintln!("nemo_relay: ATOF endpoint[{index}] client build failed: {error}"); + eprintln!("nemo_relay: ATOF endpoint[{index}] {error}"); drain_closed(rx).await; return; } }; - let (body_tx, body_rx) = tokio::sync::mpsc::unbounded_channel::(); - let body_stream = stream::unfold(body_rx, |mut body_rx| async { - loop { - match body_rx.recv().await? { - NdjsonBodyMessage::Event(bytes) => { - return Some((Ok::<_, std::io::Error>(bytes), body_rx)); - } - NdjsonBodyMessage::Flush(done) => { - let _ = done.send(()); - } - } - } - }); - let body = reqwest::Body::wrap_stream(body_stream); + let (body_tx, body) = ndjson_body_channel(); let request = tokio::spawn(async move { client .post(config.url) @@ -824,46 +800,100 @@ async fn run_ndjson_endpoint( while let Some(message) = rx.recv().await { match message { - EndpointMessage::Event(raw_json) => { - if let Err(error) = body_tx.send(NdjsonBodyMessage::Event( - format!("{raw_json}\n").into_bytes(), - )) { - eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON send failed: {error}"); - } - } - EndpointMessage::Flush(done) => { - if let Err(error) = body_tx.send(NdjsonBodyMessage::Flush(done)) { - eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON flush failed: {error}"); - error.0.acknowledge_if_flush(); - } - } + EndpointMessage::Event(raw_json) => send_ndjson_event(index, &body_tx, raw_json), + EndpointMessage::Flush(done) => send_ndjson_flush(index, &body_tx, done), EndpointMessage::Close(done) => { drop(body_tx); - match tokio::time::timeout(close_timeout, request).await { - Ok(Ok(Ok(response))) if response.status().is_success() => {} - Ok(Ok(Ok(response))) => eprintln!( - "nemo_relay: ATOF endpoint[{index}] NDJSON HTTP status {}", - response.status() - ), - Ok(Ok(Err(error))) => { - eprintln!( - "nemo_relay: ATOF endpoint[{index}] NDJSON upload failed: {error}" - ) - } - Ok(Err(error)) => { - eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON task failed: {error}") - } - Err(_) => { - eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON close timed out") - } - } - let _ = done.send(()); + finish_ndjson_upload(index, request, close_timeout, done).await; return; } } } } +#[cfg(all(feature = "atof-streaming", not(target_arch = "wasm32")))] +fn build_ndjson_client( + config: &AtofEndpointConfig, +) -> std::result::Result { + let headers = + build_header_map(&config.headers).map_err(|error| format!("disabled: {error}"))?; + reqwest::Client::builder() + .connect_timeout(Duration::from_millis(config.timeout_millis)) + .default_headers(headers) + .build() + .map_err(|error| format!("client build failed: {error}")) +} + +#[cfg(all(feature = "atof-streaming", not(target_arch = "wasm32")))] +fn ndjson_body_channel() -> ( + tokio::sync::mpsc::UnboundedSender, + reqwest::Body, +) { + let (body_tx, body_rx) = tokio::sync::mpsc::unbounded_channel::(); + let body_stream = stream::unfold(body_rx, |mut body_rx| async { + loop { + match body_rx.recv().await? { + NdjsonBodyMessage::Event(bytes) => { + return Some((Ok::<_, std::io::Error>(bytes), body_rx)); + } + NdjsonBodyMessage::Flush(done) => { + let _ = done.send(()); + } + } + } + }); + (body_tx, reqwest::Body::wrap_stream(body_stream)) +} + +#[cfg(all(feature = "atof-streaming", not(target_arch = "wasm32")))] +fn send_ndjson_event( + index: usize, + body_tx: &tokio::sync::mpsc::UnboundedSender, + raw_json: String, +) { + if let Err(error) = body_tx.send(NdjsonBodyMessage::Event( + format!("{raw_json}\n").into_bytes(), + )) { + eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON send failed: {error}"); + } +} + +#[cfg(all(feature = "atof-streaming", not(target_arch = "wasm32")))] +fn send_ndjson_flush( + index: usize, + body_tx: &tokio::sync::mpsc::UnboundedSender, + done: std_mpsc::Sender<()>, +) { + if let Err(error) = body_tx.send(NdjsonBodyMessage::Flush(done)) { + eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON flush failed: {error}"); + error.0.acknowledge_if_flush(); + } +} + +#[cfg(all(feature = "atof-streaming", not(target_arch = "wasm32")))] +async fn finish_ndjson_upload( + index: usize, + request: tokio::task::JoinHandle>, + close_timeout: Duration, + done: std_mpsc::Sender<()>, +) { + match tokio::time::timeout(close_timeout, request).await { + Ok(Ok(Ok(response))) if response.status().is_success() => {} + Ok(Ok(Ok(response))) => eprintln!( + "nemo_relay: ATOF endpoint[{index}] NDJSON HTTP status {}", + response.status() + ), + Ok(Ok(Err(error))) => { + eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON upload failed: {error}") + } + Ok(Err(error)) => { + eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON task failed: {error}") + } + Err(_) => eprintln!("nemo_relay: ATOF endpoint[{index}] NDJSON close timed out"), + } + let _ = done.send(()); +} + #[cfg(all(feature = "atof-streaming", not(target_arch = "wasm32")))] async fn drain_closed(mut rx: tokio::sync::mpsc::UnboundedReceiver) { while let Some(message) = rx.recv().await { diff --git a/crates/core/src/observability/openinference.rs b/crates/core/src/observability/openinference.rs index 98b6b2397..bd25f00b3 100644 --- a/crates/core/src/observability/openinference.rs +++ b/crates/core/src/observability/openinference.rs @@ -729,28 +729,8 @@ fn end_attributes(event: &Event) -> Vec { .annotated_response() .and_then(|response| response.usage.as_ref()) .or(fallback_usage.as_ref()); - if is_llm && let Some(usage) = usage { - if let Some(v) = usage.prompt_tokens { - attributes.push(KeyValue::new(oi::llm::token_count::PROMPT, v as i64)); - } - if let Some(v) = usage.completion_tokens { - attributes.push(KeyValue::new(oi::llm::token_count::COMPLETION, v as i64)); - } - if let Some(v) = usage.total_tokens { - attributes.push(KeyValue::new(oi::llm::token_count::TOTAL, v as i64)); - } - if let Some(v) = usage.cache_read_tokens { - attributes.push(KeyValue::new( - oi::llm::token_count::prompt_details::CACHE_READ, - v as i64, - )); - } - if let Some(v) = usage.cache_write_tokens { - attributes.push(KeyValue::new( - oi::llm::token_count::prompt_details::CACHE_WRITE, - v as i64, - )); - } + if is_llm { + push_llm_usage_attributes(&mut attributes, usage); } if is_llm && let Some(cost_total) = cost_total_from_llm_event(event, fallback_usage.as_ref()) { attributes.push(KeyValue::new(oi::llm::cost::TOTAL, cost_total)); @@ -761,6 +741,33 @@ fn end_attributes(event: &Event) -> Vec { attributes } +fn push_llm_usage_attributes(attributes: &mut Vec, usage: Option<&Usage>) { + let Some(usage) = usage else { + return; + }; + if let Some(v) = usage.prompt_tokens { + attributes.push(KeyValue::new(oi::llm::token_count::PROMPT, v as i64)); + } + if let Some(v) = usage.completion_tokens { + attributes.push(KeyValue::new(oi::llm::token_count::COMPLETION, v as i64)); + } + if let Some(v) = usage.total_tokens { + attributes.push(KeyValue::new(oi::llm::token_count::TOTAL, v as i64)); + } + if let Some(v) = usage.cache_read_tokens { + attributes.push(KeyValue::new( + oi::llm::token_count::prompt_details::CACHE_READ, + v as i64, + )); + } + if let Some(v) = usage.cache_write_tokens { + attributes.push(KeyValue::new( + oi::llm::token_count::prompt_details::CACHE_WRITE, + v as i64, + )); + } +} + fn push_llm_request_attributes(attributes: &mut Vec, event: &Event) { if let Some(request) = event.annotated_request() { push_annotated_request_attributes(attributes, request); diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 2e5b7c1ef..8c45ebb3c 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -1408,9 +1408,23 @@ fn validate_observability_plugin_config( }; let mut diagnostics = vec![]; + validate_top_level_observability_fields(&mut diagnostics, &config.policy, plugin_config); + validate_version(&mut diagnostics, &config.policy, config.version); + validate_policy_fields(&mut diagnostics, &config.policy, plugin_config); + validate_observability_section_fields(&mut diagnostics, &config.policy, plugin_config); + validate_observability_section_values(&mut diagnostics, &config); + + diagnostics +} + +fn validate_top_level_observability_fields( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + plugin_config: &Map, +) { validate_unknown_fields( - &mut diagnostics, - &config.policy, + diagnostics, + policy, Some(OBSERVABILITY_PLUGIN_KIND.to_string()), plugin_config, &[ @@ -1422,12 +1436,16 @@ fn validate_observability_plugin_config( "policy", ], ); +} - validate_version(&mut diagnostics, &config.policy, config.version); - validate_policy_fields(&mut diagnostics, &config.policy, plugin_config); +fn validate_observability_section_fields( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + plugin_config: &Map, +) { validate_section_fields( - &mut diagnostics, - &config.policy, + diagnostics, + policy, plugin_config, "atof", &[ @@ -1439,8 +1457,8 @@ fn validate_observability_plugin_config( ], ); validate_section_fields( - &mut diagnostics, - &config.policy, + diagnostics, + policy, plugin_config, "atif", &[ @@ -1456,8 +1474,8 @@ fn validate_observability_plugin_config( ], ); validate_section_fields( - &mut diagnostics, - &config.policy, + diagnostics, + policy, plugin_config, "opentelemetry", &[ @@ -1474,8 +1492,8 @@ fn validate_observability_plugin_config( ], ); validate_section_fields( - &mut diagnostics, - &config.policy, + diagnostics, + policy, plugin_config, "openinference", &[ @@ -1491,98 +1509,219 @@ fn validate_observability_plugin_config( "timeout_millis", ], ); +} +fn validate_observability_section_values( + diagnostics: &mut Vec, + config: &ObservabilityConfig, +) { if let Some(section) = &config.atof { - validate_atof_values(&mut diagnostics, &config.policy, section); - #[cfg(target_arch = "wasm32")] - if section.enabled { - push_policy_diag( - &mut diagnostics, - config.policy.unsupported_value, - "observability.unsupported_value", - Some("atof".to_string()), - Some("enabled".to_string()), - "ATOF file export is not supported on WebAssembly".to_string(), - ); - } - #[cfg(target_arch = "wasm32")] - if section.enabled && !section.endpoints.is_empty() { - push_policy_diag( - &mut diagnostics, - config.policy.unsupported_value, - "observability.unsupported_value", - Some("atof".to_string()), - Some("endpoints".to_string()), - "ATOF streaming endpoints are not supported on WebAssembly".to_string(), - ); - } - #[cfg(all(not(feature = "atof-streaming"), not(target_arch = "wasm32")))] - if section.enabled && !section.endpoints.is_empty() { - push_policy_diag( - &mut diagnostics, - config.policy.unsupported_value, - "observability.unsupported_value", - Some("atof".to_string()), - Some("endpoints".to_string()), - "ATOF streaming endpoints are not enabled in this build".to_string(), - ); - } + validate_atof_section(diagnostics, &config.policy, section); } if let Some(section) = &config.atif { - validate_atif_values(&mut diagnostics, &config.policy, section); - #[cfg(target_arch = "wasm32")] - if section.enabled { - push_policy_diag( - &mut diagnostics, - config.policy.unsupported_value, - "observability.unsupported_value", - Some("atif".to_string()), - Some("enabled".to_string()), - "ATIF file export is not supported on WebAssembly".to_string(), - ); - } - #[cfg(not(all(feature = "object-store", not(target_arch = "wasm32"))))] - if !section.storage.is_empty() { - push_policy_diag( - &mut diagnostics, - config.policy.unsupported_value, - "observability.feature_disabled", - Some("atif".to_string()), - Some("storage".to_string()), - "ATIF storage support is not enabled in this build".to_string(), - ); - } + validate_atif_section(diagnostics, &config.policy, section); } if let Some(section) = &config.opentelemetry { - validate_otlp_values(&mut diagnostics, &config.policy, "opentelemetry", section); - #[cfg(not(feature = "otel"))] - if section.enabled { - push_policy_diag( - &mut diagnostics, - config.policy.unsupported_value, - "observability.feature_disabled", - Some("opentelemetry".to_string()), - Some("enabled".to_string()), - "OpenTelemetry support is not enabled in this build".to_string(), - ); - } + validate_opentelemetry_section(diagnostics, &config.policy, section); } if let Some(section) = &config.openinference { - validate_otlp_values(&mut diagnostics, &config.policy, "openinference", section); - #[cfg(not(feature = "openinference"))] - if section.enabled { - push_policy_diag( - &mut diagnostics, - config.policy.unsupported_value, - "observability.feature_disabled", - Some("openinference".to_string()), - Some("enabled".to_string()), - "OpenInference support is not enabled in this build".to_string(), - ); - } + validate_openinference_section(diagnostics, &config.policy, section); } +} - diagnostics +fn validate_atof_section( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtofSectionConfig, +) { + validate_atof_values(diagnostics, policy, section); + validate_atof_feature_support(diagnostics, policy, section); +} + +#[cfg(target_arch = "wasm32")] +fn validate_atof_feature_support( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtofSectionConfig, +) { + if section.enabled { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some("enabled".to_string()), + "ATOF file export is not supported on WebAssembly".to_string(), + ); + } + if section.enabled && !section.endpoints.is_empty() { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some("endpoints".to_string()), + "ATOF streaming endpoints are not supported on WebAssembly".to_string(), + ); + } +} + +#[cfg(all(not(feature = "atof-streaming"), not(target_arch = "wasm32")))] +fn validate_atof_feature_support( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtofSectionConfig, +) { + if section.enabled && !section.endpoints.is_empty() { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some("endpoints".to_string()), + "ATOF streaming endpoints are not enabled in this build".to_string(), + ); + } +} + +#[cfg(all(feature = "atof-streaming", not(target_arch = "wasm32")))] +fn validate_atof_feature_support( + _diagnostics: &mut Vec, + _policy: &ConfigPolicy, + _section: &AtofSectionConfig, +) { +} + +fn validate_atif_section( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtifSectionConfig, +) { + validate_atif_values(diagnostics, policy, section); + validate_atif_file_export_support(diagnostics, policy, section); + validate_atif_storage_support(diagnostics, policy, section); +} + +#[cfg(target_arch = "wasm32")] +fn validate_atif_file_export_support( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtifSectionConfig, +) { + if section.enabled { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atif".to_string()), + Some("enabled".to_string()), + "ATIF file export is not supported on WebAssembly".to_string(), + ); + } +} + +#[cfg(not(target_arch = "wasm32"))] +fn validate_atif_file_export_support( + _diagnostics: &mut Vec, + _policy: &ConfigPolicy, + _section: &AtifSectionConfig, +) { +} + +#[cfg(not(all(feature = "object-store", not(target_arch = "wasm32"))))] +fn validate_atif_storage_support( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &AtifSectionConfig, +) { + if section.enabled && !section.storage.is_empty() { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.feature_disabled", + Some("atif".to_string()), + Some("storage".to_string()), + "ATIF storage support is not enabled in this build".to_string(), + ); + } +} + +#[cfg(all(feature = "object-store", not(target_arch = "wasm32")))] +fn validate_atif_storage_support( + _diagnostics: &mut Vec, + _policy: &ConfigPolicy, + _section: &AtifSectionConfig, +) { +} + +fn validate_opentelemetry_section( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &OtlpSectionConfig, +) { + validate_otlp_values(diagnostics, policy, "opentelemetry", section); + validate_opentelemetry_feature_support(diagnostics, policy, section); +} + +#[cfg(not(feature = "otel"))] +fn validate_opentelemetry_feature_support( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &OtlpSectionConfig, +) { + if section.enabled { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.feature_disabled", + Some("opentelemetry".to_string()), + Some("enabled".to_string()), + "OpenTelemetry support is not enabled in this build".to_string(), + ); + } +} + +#[cfg(feature = "otel")] +fn validate_opentelemetry_feature_support( + _diagnostics: &mut Vec, + _policy: &ConfigPolicy, + _section: &OtlpSectionConfig, +) { +} + +fn validate_openinference_section( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &OtlpSectionConfig, +) { + validate_otlp_values(diagnostics, policy, "openinference", section); + validate_openinference_feature_support(diagnostics, policy, section); +} + +#[cfg(not(feature = "openinference"))] +fn validate_openinference_feature_support( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + section: &OtlpSectionConfig, +) { + if section.enabled { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.feature_disabled", + Some("openinference".to_string()), + Some("enabled".to_string()), + "OpenInference support is not enabled in this build".to_string(), + ); + } +} + +#[cfg(feature = "openinference")] +fn validate_openinference_feature_support( + _diagnostics: &mut Vec, + _policy: &ConfigPolicy, + _section: &OtlpSectionConfig, +) { } fn validate_version(diagnostics: &mut Vec, policy: &ConfigPolicy, version: u32) { diff --git a/crates/core/src/plugins/nemo_guardrails/component.rs b/crates/core/src/plugins/nemo_guardrails/component.rs index 7b7b968b1..6c8e52330 100644 --- a/crates/core/src/plugins/nemo_guardrails/component.rs +++ b/crates/core/src/plugins/nemo_guardrails/component.rs @@ -611,123 +611,112 @@ fn validate_non_empty_strings( policy: &ConfigPolicy, config: &NeMoGuardrailsConfig, ) { - if let Some(config_path) = &config.config_path - && config_path.trim().is_empty() - { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("config_path".to_string()), - "config_path must not be empty".to_string(), - ); - } - - if let Some(config_yaml) = &config.config_yaml - && config_yaml.trim().is_empty() - { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("config_yaml".to_string()), - "config_yaml must not be empty".to_string(), - ); - } - - if let Some(colang_content) = &config.colang_content - && colang_content.trim().is_empty() - { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("colang_content".to_string()), - "colang_content must not be empty".to_string(), - ); - } + validate_optional_non_empty_string( + diagnostics, + policy, + "config_path", + config.config_path.as_deref(), + "config_path must not be empty", + ); + validate_optional_non_empty_string( + diagnostics, + policy, + "config_yaml", + config.config_yaml.as_deref(), + "config_yaml must not be empty", + ); + validate_optional_non_empty_string( + diagnostics, + policy, + "colang_content", + config.colang_content.as_deref(), + "colang_content must not be empty", + ); if let Some(remote) = &config.remote { - if let Some(endpoint) = &remote.endpoint - && endpoint.trim().is_empty() - { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("remote.endpoint".to_string()), - "remote.endpoint must not be empty".to_string(), - ); - } - if let Some(config_id) = &remote.config_id - && config_id.trim().is_empty() - { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("remote.config_id".to_string()), - "remote.config_id must not be empty".to_string(), - ); - } - for (index, config_id) in remote.config_ids.iter().enumerate() { - if config_id.trim().is_empty() { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some(format!("remote.config_ids[{index}]")), - "remote.config_ids entries must not be empty".to_string(), - ); - } - } + validate_remote_non_empty_strings(diagnostics, policy, remote); } - if let Some(local) = &config.local - && let Some(python_module) = &local.python_module - && python_module.trim().is_empty() - { - push_policy_diag( - diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("local.python_module".to_string()), - "local.python_module must not be empty".to_string(), - ); + if let Some(local) = &config.local { + validate_local_non_empty_strings(diagnostics, policy, local); } +} - if let Some(local) = &config.local - && let Some(python_executable) = &local.python_executable - && python_executable.trim().is_empty() - { - push_policy_diag( +fn validate_remote_non_empty_strings( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + remote: &RemoteBackendConfig, +) { + validate_optional_non_empty_string( + diagnostics, + policy, + "remote.endpoint", + remote.endpoint.as_deref(), + "remote.endpoint must not be empty", + ); + validate_optional_non_empty_string( + diagnostics, + policy, + "remote.config_id", + remote.config_id.as_deref(), + "remote.config_id must not be empty", + ); + for (index, config_id) in remote.config_ids.iter().enumerate() { + validate_optional_non_empty_string( diagnostics, - policy.unsupported_value, - "nemo_guardrails.unsupported_value", - Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("local.python_executable".to_string()), - "local.python_executable must not be empty".to_string(), + policy, + format!("remote.config_ids[{index}]"), + Some(config_id.as_str()), + "remote.config_ids entries must not be empty", ); } +} - if let Some(local) = &config.local - && let Some(python_path) = &local.python_path - && python_path.trim().is_empty() +fn validate_local_non_empty_strings( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + local: &LocalBackendConfig, +) { + validate_optional_non_empty_string( + diagnostics, + policy, + "local.python_module", + local.python_module.as_deref(), + "local.python_module must not be empty", + ); + validate_optional_non_empty_string( + diagnostics, + policy, + "local.python_executable", + local.python_executable.as_deref(), + "local.python_executable must not be empty", + ); + validate_optional_non_empty_string( + diagnostics, + policy, + "local.python_path", + local.python_path.as_deref(), + "local.python_path must not be empty", + ); +} + +fn validate_optional_non_empty_string( + diagnostics: &mut Vec, + policy: &ConfigPolicy, + field: impl Into, + value: Option<&str>, + message: &str, +) { + if let Some(value) = value + && value.trim().is_empty() { push_policy_diag( diagnostics, policy.unsupported_value, "nemo_guardrails.unsupported_value", Some(NEMO_GUARDRAILS_PLUGIN_KIND.to_string()), - Some("local.python_path".to_string()), - "local.python_path must not be empty".to_string(), + Some(field.into()), + message.to_string(), ); } } diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index e0bbf18e3..584d9edba 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -1327,6 +1327,26 @@ fn atif_storage_unknown_backend_type_is_rejected() { ); } +#[test] +fn disabled_atif_storage_config_does_not_report_feature_disabled() { + let report = validate_plugin_config(&plugin_config(json!({ + "atif": { + "enabled": false, + "filename_template": "trajectory-{session_id}.json", + "storage": [{"type": "s3", "bucket": "configured-but-disabled"}] + } + }))); + + assert!( + !report.diagnostics.iter().any(|diag| { + diag.code == "observability.feature_disabled" + && diag.field.as_deref() == Some("storage") + }), + "disabled ATIF storage should not report feature-disabled diagnostics: {:?}", + report.diagnostics + ); +} + #[test] fn atif_storage_empty_bucket_is_rejected() { let report = validate_plugin_config(&plugin_config(json!({