Skip to content

refactor: reduce SonarQube complexity findings #243

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rapids-bot merged 3 commits into from
Jun 8, 2026
+618 −422
Merged

refactor: reduce SonarQube complexity findings #243

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fa0b710
refactor: reduce SonarQube complexity findings
willkill07 Jun 7, 2026
6e2d9b4
Merge branch 'main' into wkk_sonarqube-complexity-refactors
willkill07 Jun 7, 2026
e0d4a80
fix: gate disabled ATIF storage diagnostics
willkill07 Jun 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 73 additions & 79 deletions crates/core/src/codec/anthropic.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,6 +283,75 @@ fn encode_anthropic_tools(tools: &[ToolDefinition]) -> Vec<Json> {
.collect()
}

fn anthropic_text_message(content_blocks: Option<&[Json]>) -> Option<MessageContent> {
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<Vec<ResponseToolCall>> {
let tool_calls: Vec<ResponseToolCall> = 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<ResponseToolCall> {
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<RawAnthropicUsage>,
model_for_pricing: Option<&str>,
) -> Option<Usage> {
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
// ---------------------------------------------------------------------------
Expand All @@ -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<ResponseToolCall> = 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();
Expand Down
151 changes: 84 additions & 67 deletions crates/core/src/observability/atif.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Json>,
) -> Option<Json> {
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<String, Json>) -> Option<Json> {
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<String, Json>, key: &str) -> Option<Json> {
obj.get(key).filter(|value| !value.is_null()).cloned()
}
Expand Down Expand Up @@ -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<AtifFinalMetrics> {
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<u64>,
completion_tokens: Option<u64>,
cached_tokens: Option<u64>,
cost_usd: Option<f64>,
}

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<u64>, value: Option<u64>) {
if let Some(value) = value {
*total = Some(total.unwrap_or(0) + value);
}
}

fn add_f64_total(total: &mut Option<f64>, value: Option<f64>) {
if let Some(value) = value {
*total = Some(total.unwrap_or(0.0) + value);
}
}

// ---------------------------------------------------------------------------
Expand Down
Loading
Loading