From 11a9150266d9e0a8492531c3dfa75e00c3e6a91f Mon Sep 17 00:00:00 2001 From: Will Killian Date: Thu, 14 May 2026 08:30:06 -0400 Subject: [PATCH] fix: address release 0.2 sonar issues Signed-off-by: Will Killian --- crates/cli/src/doctor.rs | 70 ++++---- crates/cli/src/plugins.rs | 137 ++++++++++----- crates/core/src/codec/openai_responses.rs | 194 +++++++++++++--------- go/nemo_flow/observability_plugin_test.go | 20 ++- 4 files changed, 263 insertions(+), 158 deletions(-) diff --git a/crates/cli/src/doctor.rs b/crates/cli/src/doctor.rs index 1eded232a..c21f149b2 100644 --- a/crates/cli/src/doctor.rs +++ b/crates/cli/src/doctor.rs @@ -525,41 +525,55 @@ async fn collect_observability(gateway: &GatewayConfig) -> Vec { async fn collect_observability_component_checks(checks: &mut Vec, config: &Value) { for section in ["atof", "atif"] { - if section_enabled(config, section) { - let label = if section == "atof" { - "ATOF dir" - } else { - "ATIF dir" - }; - match section_output_directory(config, section) { - Some(path) => checks.push(check_directory(label, &path)), - None => checks.push(Check { - name: label, - status: Status::Info, - details: "enabled; using runtime default output directory".into(), - }), - } + if let Some(check) = observability_file_exporter_check(config, section) { + checks.push(check); } } for section in ["opentelemetry", "openinference"] { - if section_enabled(config, section) { - let label = if section == "opentelemetry" { - "OpenTelemetry endpoint" - } else { - "OpenInference endpoint" - }; - match section_endpoint(config, section) { - Some(endpoint) => checks.push(probe_http_named(label, &endpoint).await), - None => checks.push(Check { - name: label, - status: Status::Info, - details: "enabled; using exporter default endpoint".into(), - }), - } + if let Some(check) = observability_http_exporter_check(config, section).await { + checks.push(check); } } } +fn observability_file_exporter_check(config: &Value, section: &str) -> Option { + if !section_enabled(config, section) { + return None; + } + let label = if section == "atof" { + "ATOF dir" + } else { + "ATIF dir" + }; + Some(match section_output_directory(config, section) { + Some(path) => check_directory(label, &path), + None => Check { + name: label, + status: Status::Info, + details: "enabled; using runtime default output directory".into(), + }, + }) +} + +async fn observability_http_exporter_check(config: &Value, section: &str) -> Option { + if !section_enabled(config, section) { + return None; + } + let label = if section == "opentelemetry" { + "OpenTelemetry endpoint" + } else { + "OpenInference endpoint" + }; + Some(match section_endpoint(config, section) { + Some(endpoint) => probe_http_named(label, &endpoint).await, + None => Check { + name: label, + status: Status::Info, + details: "enabled; using exporter default endpoint".into(), + }, + }) +} + fn observability_component_config(plugin_value: &Value) -> Option<&Value> { plugin_value .get("components") diff --git a/crates/cli/src/plugins.rs b/crates/cli/src/plugins.rs index d92258c32..3def100e4 100644 --- a/crates/cli/src/plugins.rs +++ b/crates/cli/src/plugins.rs @@ -463,28 +463,7 @@ fn edit_section( .ok_or_else(|| CliError::Config(format!("{} is not an editable section", section.name)))? .fields; loop { - let mut items = Vec::new(); - if section_has_enabled_toggle(section) { - let enabled = section_enabled(config, section).unwrap_or(false); - items.push(MenuItem::new(format!( - "Toggle section [{}]", - status_label(enabled) - ))); - } - for field in fields { - let configured = section_field_configured(config, section, *field)?; - items.push(MenuItem::new(format!( - "{} = {}", - configured_label(configured, field.name), - section_field_value(config, section, field.name)? - .map(|value| display_field_value(section, *field, &value)) - .or_else(|| default_field_value(section, *field) - .map(|value| format!("{} (default)", display_value(&value)))) - .unwrap_or_else(|| "(default)".to_string()) - ))); - } - items.push(MenuItem::new(shortcut_label("Reset section", "r"))); - items.push(MenuItem::new(shortcut_label("Back", "q"))); + let items = section_menu_items(config, section, fields)?; let selection = prompt_menu(theme, section.name, &items, 0)?; let selection = match selection { MenuResponse::Selected(selection) => selection, @@ -493,14 +472,7 @@ fn edit_section( continue; } MenuResponse::Shortcut(MenuShortcut::Reset, selected) => { - if reset_selected_field(config, section, fields, selected)? { - continue; - } - let reset_section_index = - usize::from(section_has_enabled_toggle(section)) + fields.len(); - if selected == reset_section_index { - reset_section(config, section); - } + reset_selected_item(config, section, fields, selected)?; continue; } MenuResponse::Shortcut(MenuShortcut::Clear, selected) => { @@ -516,24 +488,91 @@ fn edit_section( } MenuResponse::Cancel => return Ok(()), }; - let mut index = selection; - if section_has_enabled_toggle(section) { - if index == 0 { - toggle_section(config, section); - continue; - } - index -= 1; - } - if index < fields.len() { - edit_field(theme, config, section, &fields[index])?; - } else if index == fields.len() { - reset_section(config, section); - } else { + if !edit_selected_section_item(theme, config, section, fields, selection)? { return Ok(()); } } } +fn section_menu_items( + config: &ObservabilityConfig, + section: EditorFieldSpec, + fields: &[EditorFieldSpec], +) -> Result, CliError> { + let mut items = Vec::new(); + if section_has_enabled_toggle(section) { + let enabled = section_enabled(config, section).unwrap_or(false); + items.push(MenuItem::new(format!( + "Toggle section [{}]", + status_label(enabled) + ))); + } + for field in fields { + items.push(section_field_menu_item(config, section, *field)?); + } + items.push(MenuItem::new(shortcut_label("Reset section", "r"))); + items.push(MenuItem::new(shortcut_label("Back", "q"))); + Ok(items) +} + +fn section_field_menu_item( + config: &ObservabilityConfig, + section: EditorFieldSpec, + field: EditorFieldSpec, +) -> Result { + let configured = section_field_configured(config, section, field)?; + let value = section_field_value(config, section, field.name)? + .map(|value| display_field_value(section, field, &value)) + .or_else(|| { + default_field_value(section, field) + .map(|value| format!("{} (default)", display_value(&value))) + }) + .unwrap_or_else(|| "(default)".to_string()); + Ok(MenuItem::new(format!( + "{} = {}", + configured_label(configured, field.name), + value + ))) +} + +fn reset_selected_item( + config: &mut ObservabilityConfig, + section: EditorFieldSpec, + fields: &[EditorFieldSpec], + selected: usize, +) -> Result<(), CliError> { + if reset_selected_field(config, section, fields, selected)? { + return Ok(()); + } + if selected == reset_section_index(section, fields) { + reset_section(config, section); + } + Ok(()) +} + +fn edit_selected_section_item( + theme: &ColorfulTheme, + config: &mut ObservabilityConfig, + section: EditorFieldSpec, + fields: &[EditorFieldSpec], + selection: usize, +) -> Result { + if section_has_enabled_toggle(section) && selection == 0 { + toggle_section(config, section); + return Ok(true); + } + let index = selected_field_index(section, selection); + if let Some(field) = fields.get(index) { + edit_field(theme, config, section, field)?; + return Ok(true); + } + if index == fields.len() { + reset_section(config, section); + return Ok(true); + } + Ok(false) +} + fn edit_field( theme: &ColorfulTheme, config: &mut ObservabilityConfig, @@ -755,6 +794,18 @@ fn reset_selected_field( Ok(true) } +fn selected_field_index(section: EditorFieldSpec, selected: usize) -> usize { + let mut index = selected; + if section_has_enabled_toggle(section) { + index -= 1; + } + index +} + +fn reset_section_index(section: EditorFieldSpec, fields: &[EditorFieldSpec]) -> usize { + usize::from(section_has_enabled_toggle(section)) + fields.len() +} + fn section_has_enabled_toggle(section: EditorFieldSpec) -> bool { section.name != POLICY_SECTION && section diff --git a/crates/core/src/codec/openai_responses.rs b/crates/core/src/codec/openai_responses.rs index be82e8db0..8d57951ad 100644 --- a/crates/core/src/codec/openai_responses.rs +++ b/crates/core/src/codec/openai_responses.rs @@ -295,6 +295,118 @@ fn overlay_generation_params(obj: &mut serde_json::Map, params: &G } } +fn encode_openai_responses_input( + obj: &mut serde_json::Map, + annotated: &AnnotatedLlmRequest, +) -> Result<()> { + let (system_text, input_messages) = split_system_and_input_messages(&annotated.messages); + set_or_remove_string(obj, "instructions", system_text); + if let Some(raw_input_items) = annotated.extra.get(UNPARSED_INPUT_ITEMS_KEY) { + obj.insert("input".into(), raw_input_items.clone()); + } else { + insert_serialized(obj, "input", &input_messages, "input")?; + } + Ok(()) +} + +fn encode_openai_responses_tools( + obj: &mut serde_json::Map, + annotated: &AnnotatedLlmRequest, +) -> Result<()> { + if let Some(ref tools) = annotated.tools { + insert_serialized(obj, "tools", tools, "tools")?; + } + if let Some(ref tool_choice) = annotated.tool_choice { + insert_serialized(obj, "tool_choice", tool_choice, "tool_choice")?; + } + Ok(()) +} + +fn overlay_openai_responses_fields( + obj: &mut serde_json::Map, + annotated: &AnnotatedLlmRequest, +) { + if let Some(ref model) = annotated.model { + obj.insert("model".into(), Json::String(model.clone())); + } + overlay_openai_responses_json_fields(obj, annotated); + overlay_openai_responses_string_fields(obj, annotated); + overlay_openai_responses_bool_fields(obj, annotated); + overlay_openai_responses_u64_fields(obj, annotated); +} + +fn overlay_openai_responses_json_fields( + obj: &mut serde_json::Map, + annotated: &AnnotatedLlmRequest, +) { + for (key, value) in [ + ("truncation", &annotated.truncation), + ("reasoning", &annotated.reasoning), + ("include", &annotated.include), + ("metadata", &annotated.metadata), + ] { + if let Some(value) = value { + obj.insert(key.into(), value.clone()); + } + } +} + +fn overlay_openai_responses_string_fields( + obj: &mut serde_json::Map, + annotated: &AnnotatedLlmRequest, +) { + for (key, value) in [ + ("previous_response_id", &annotated.previous_response_id), + ("user", &annotated.user), + ("service_tier", &annotated.service_tier), + ] { + if let Some(value) = value { + obj.insert(key.into(), Json::String(value.clone())); + } + } +} + +fn overlay_openai_responses_bool_fields( + obj: &mut serde_json::Map, + annotated: &AnnotatedLlmRequest, +) { + for (key, value) in [ + ("store", annotated.store), + ("parallel_tool_calls", annotated.parallel_tool_calls), + ("stream", annotated.stream), + ] { + if let Some(value) = value { + obj.insert(key.into(), Json::Bool(value)); + } + } +} + +fn overlay_openai_responses_u64_fields( + obj: &mut serde_json::Map, + annotated: &AnnotatedLlmRequest, +) { + for (key, value) in [ + ("max_output_tokens", annotated.max_output_tokens), + ("max_tool_calls", annotated.max_tool_calls), + ("top_logprobs", annotated.top_logprobs), + ] { + if let Some(value) = value { + obj.insert(key.into(), Json::from(value)); + } + } +} + +fn merge_openai_responses_extra_fields( + obj: &mut serde_json::Map, + extra: &serde_json::Map, +) { + for (k, v) in extra { + if k != UNPARSED_INPUT_ITEMS_KEY { + obj.insert(k.clone(), v.clone()); + } + } +} + fn decode_openai_or_anthropic_tool_choice(value: &Json) -> Option { if let Ok(parsed) = serde_json::from_value::(value.clone()) { return Some(parsed); @@ -516,87 +628,13 @@ impl LlmCodec for OpenAIResponsesCodec { .as_object_mut() .ok_or_else(|| FlowError::Internal("original content is not an object".into()))?; - let (system_text, input_messages) = split_system_and_input_messages(&annotated.messages); - set_or_remove_string(obj, "instructions", system_text); - if let Some(raw_input_items) = annotated.extra.get(UNPARSED_INPUT_ITEMS_KEY) { - obj.insert("input".into(), raw_input_items.clone()); - } else { - insert_serialized(obj, "input", &input_messages, "input")?; - } - - // Overlay model if present. - if let Some(ref model) = annotated.model { - obj.insert("model".into(), Json::String(model.clone())); - } - - // Overlay generation params. + encode_openai_responses_input(obj, annotated)?; if let Some(ref params) = annotated.params { overlay_generation_params(obj, params); } - - // Overlay tools if present. - if let Some(ref tools) = annotated.tools { - insert_serialized(obj, "tools", tools, "tools")?; - } - - // Overlay tool_choice if present. - if let Some(ref tool_choice) = annotated.tool_choice { - insert_serialized(obj, "tool_choice", tool_choice, "tool_choice")?; - } - - if let Some(store) = annotated.store { - obj.insert("store".into(), Json::Bool(store)); - } - if let Some(ref previous_response_id) = annotated.previous_response_id { - obj.insert( - "previous_response_id".into(), - Json::String(previous_response_id.clone()), - ); - } - if let Some(ref truncation) = annotated.truncation { - obj.insert("truncation".into(), truncation.clone()); - } - if let Some(ref reasoning) = annotated.reasoning { - obj.insert("reasoning".into(), reasoning.clone()); - } - if let Some(ref include) = annotated.include { - obj.insert("include".into(), include.clone()); - } - if let Some(ref user) = annotated.user { - obj.insert("user".into(), Json::String(user.clone())); - } - if let Some(ref metadata) = annotated.metadata { - obj.insert("metadata".into(), metadata.clone()); - } - if let Some(ref service_tier) = annotated.service_tier { - obj.insert("service_tier".into(), Json::String(service_tier.clone())); - } - if let Some(parallel_tool_calls) = annotated.parallel_tool_calls { - obj.insert( - "parallel_tool_calls".into(), - Json::Bool(parallel_tool_calls), - ); - } - if let Some(max_output_tokens) = annotated.max_output_tokens { - obj.insert("max_output_tokens".into(), Json::from(max_output_tokens)); - } - if let Some(max_tool_calls) = annotated.max_tool_calls { - obj.insert("max_tool_calls".into(), Json::from(max_tool_calls)); - } - if let Some(top_logprobs) = annotated.top_logprobs { - obj.insert("top_logprobs".into(), Json::from(top_logprobs)); - } - if let Some(stream) = annotated.stream { - obj.insert("stream".into(), Json::Bool(stream)); - } - - // Merge extra fields back. - for (k, v) in &annotated.extra { - if k == UNPARSED_INPUT_ITEMS_KEY { - continue; - } - obj.insert(k.clone(), v.clone()); - } + encode_openai_responses_tools(obj, annotated)?; + overlay_openai_responses_fields(obj, annotated); + merge_openai_responses_extra_fields(obj, &annotated.extra); Ok(LlmRequest { headers: original.headers.clone(), diff --git a/go/nemo_flow/observability_plugin_test.go b/go/nemo_flow/observability_plugin_test.go index 34222fcb8..8369db837 100644 --- a/go/nemo_flow/observability_plugin_test.go +++ b/go/nemo_flow/observability_plugin_test.go @@ -18,6 +18,8 @@ const ( FirstAgentName = "go-first-agent" NestedAgentName = "go-nested-agent" SecondAgentName = "go-second-agent" + fatalErrorFormat = "%s: %v" + failedSuffix = " failed" ) func TestObservabilityConfigHelpers(t *testing.T) { @@ -50,7 +52,7 @@ func TestObservabilityConfigHelpers(t *testing.T) { func TestObservabilityPluginAtofAndAtifFiles(t *testing.T) { if err := ClearPluginConfiguration(); err != nil { - t.Fatalf("%s: %v", ClearPluginConfigurationFailed, err) + t.Fatalf(fatalErrorFormat, ClearPluginConfigurationFailed, err) } t.Cleanup(func() { requireNoError(t, ClearPluginConfiguration(), ClearPluginConfigurationFailed) @@ -80,7 +82,7 @@ func TestObservabilityPluginAtofAndAtifFiles(t *testing.T) { t.Fatalf("unexpected diagnostics: %#v", report.Diagnostics) } if _, err := InitializePlugins(PluginConfig{Version: 1, Components: []PluginComponentSpec{ObservabilityComponent(config)}}); err != nil { - t.Fatalf("%s: %v", InitializePluginsFailed, err) + t.Fatalf(fatalErrorFormat, InitializePluginsFailed, err) } handle, err := PushScope("go-observability-agent", ScopeTypeAgent, WithInput(json.RawMessage(`{"agent":true}`))) @@ -94,7 +96,7 @@ func TestObservabilityPluginAtofAndAtifFiles(t *testing.T) { t.Fatalf("PopScope failed: %v", err) } if err := ClearPluginConfiguration(); err != nil { - t.Fatalf("%s: %v", ClearPluginConfigurationFailed, err) + t.Fatalf(fatalErrorFormat, ClearPluginConfigurationFailed, err) } jsonl := string(mustReadFile(t, filepath.Join(dir, eventsJSONLFilename))) @@ -183,7 +185,7 @@ func TestObservabilityPluginListKindIsAutomatic(t *testing.T) { func TestObservabilityAtifOpenAgentFlushesOnClear(t *testing.T) { if err := ClearPluginConfiguration(); err != nil { - t.Fatalf("%s: %v", ClearPluginConfigurationFailed, err) + t.Fatalf(fatalErrorFormat, ClearPluginConfigurationFailed, err) } t.Cleanup(func() { requireNoError(t, ClearPluginConfiguration(), ClearPluginConfigurationFailed) @@ -195,14 +197,14 @@ func TestObservabilityAtifOpenAgentFlushesOnClear(t *testing.T) { atif.OutputDirectory = dir config.Atif = &atif if _, err := InitializePlugins(PluginConfig{Version: 1, Components: []PluginComponentSpec{ObservabilityComponent(config)}}); err != nil { - t.Fatalf("%s: %v", InitializePluginsFailed, err) + t.Fatalf(fatalErrorFormat, InitializePluginsFailed, err) } handle, err := PushScope("go-open-agent", ScopeTypeAgent) if err != nil { t.Fatalf("PushScope failed: %v", err) } if err := ClearPluginConfiguration(); err != nil { - t.Fatalf("%s: %v", ClearPluginConfigurationFailed, err) + t.Fatalf(fatalErrorFormat, ClearPluginConfigurationFailed, err) } path := filepath.Join(dir, "nemo-flow-atif-"+handle.UUID()+".json") if _, err := os.Stat(path); err != nil { @@ -241,14 +243,14 @@ func EmitAgentTrajectory(t *testing.T, Label string, Name string) *ScopeHandle { func EmitAgentStart(t *testing.T, Label string, Name string) *ScopeHandle { t.Helper() Handle, Err := PushScope(Name, ScopeTypeAgent, WithInput(json.RawMessage(`{"agent":"`+Label+`"}`))) - requireNoError(t, Err, "PushScope "+Label+" failed") - requireNoError(t, EmitEvent("go-"+Label+"-mark", WithEventParent(Handle), WithEventData(json.RawMessage(`{"agent":"`+Label+`"}`))), "EmitEvent "+Label+" failed") + requireNoError(t, Err, "PushScope "+Label+failedSuffix) + requireNoError(t, EmitEvent("go-"+Label+"-mark", WithEventParent(Handle), WithEventData(json.RawMessage(`{"agent":"`+Label+`"}`))), "EmitEvent "+Label+failedSuffix) return Handle } func EmitAgentEnd(t *testing.T, Label string, Handle *ScopeHandle) { t.Helper() - requireNoError(t, PopScope(Handle, WithOutput(json.RawMessage(`{"done":true}`))), "PopScope "+Label+" failed") + requireNoError(t, PopScope(Handle, WithOutput(json.RawMessage(`{"done":true}`))), "PopScope "+Label+failedSuffix) } func TrajectoryFilePath(Dir string, Handle *ScopeHandle) string {