Skip to content

Commit dc63846

Browse files
ZhiXiao-Linclaude
ZhiXiao-Lin
and
claude
committed
fix: remove all tracing spans from entire async call chain
The previous fix only removed spans from agent.rs, but spans in llm/anthropic.rs, llm/openai.rs, and tools/registry.rs also made their futures !Send with tracing 0.1.41+ (PhantomNotSend in Span). Since these are called from execute_loop which is spawned via tokio::spawn and JoinSet::spawn, the entire call chain must be Send-compatible. Remove all info_span!/Instrument usage from: - llm/anthropic.rs: complete() and complete_streaming() - llm/openai.rs: complete() and complete_streaming() - tools/registry.rs: execute() All tracing::info!/warn! event logging is preserved. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 9c3d58a commit dc63846

3 files changed

Lines changed: 323 additions & 362 deletions

File tree

‎core/src/llm/anthropic.rs‎

Lines changed: 147 additions & 165 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@ use futures::StreamExt;
1010
use serde::Deserialize;
1111
use std::sync::Arc;
1212
use tokio::sync::mpsc;
13-
use tracing::Instrument;
1413

1514
/// Default max tokens for LLM responses
1615
pub(crate) const DEFAULT_MAX_TOKENS: usize = 8192;
@@ -99,17 +98,7 @@ impl LlmClient for AnthropicClient {
9998
system: Option<&str>,
10099
tools: &[ToolDefinition],
101100
) -> Result<LlmResponse> {
102-
let span = tracing::info_span!(
103-
"a3s.llm.completion",
104-
"a3s.llm.provider" = "anthropic",
105-
"a3s.llm.model" = %self.model,
106-
"a3s.llm.streaming" = false,
107-
"a3s.llm.prompt_tokens" = tracing::field::Empty,
108-
"a3s.llm.completion_tokens" = tracing::field::Empty,
109-
"a3s.llm.total_tokens" = tracing::field::Empty,
110-
"a3s.llm.stop_reason" = tracing::field::Empty,
111-
);
112-
async {
101+
{
113102
let request_body = self.build_request(messages, system, tools);
114103
let url = format!("{}/v1/messages", self.base_url);
115104

@@ -192,8 +181,6 @@ impl LlmClient for AnthropicClient {
192181

193182
Ok(llm_response)
194183
}
195-
.instrument(span)
196-
.await
197184
}
198185

199186
async fn complete_streaming(
@@ -202,116 +189,114 @@ impl LlmClient for AnthropicClient {
202189
system: Option<&str>,
203190
tools: &[ToolDefinition],
204191
) -> Result<mpsc::Receiver<StreamEvent>> {
205-
let span = tracing::info_span!(
206-
"a3s.llm.completion",
207-
"a3s.llm.provider" = "anthropic",
208-
"a3s.llm.model" = %self.model,
209-
"a3s.llm.streaming" = true,
210-
"a3s.llm.prompt_tokens" = tracing::field::Empty,
211-
"a3s.llm.completion_tokens" = tracing::field::Empty,
212-
"a3s.llm.total_tokens" = tracing::field::Empty,
213-
"a3s.llm.stop_reason" = tracing::field::Empty,
214-
);
215-
async {
216-
let mut request_body = self.build_request(messages, system, tools);
217-
request_body["stream"] = serde_json::json!(true);
218-
219-
let url = format!("{}/v1/messages", self.base_url);
220-
221-
let headers = vec![
222-
("x-api-key", self.api_key.expose()),
223-
("anthropic-version", "2023-06-01"),
224-
];
225-
226-
let streaming_resp = crate::retry::with_retry(&self.retry_config, |_attempt| {
227-
let http = &self.http;
228-
let url = &url;
229-
let headers = headers.clone();
230-
let request_body = &request_body;
231-
async move {
232-
match http.post_streaming(url, headers, request_body).await {
233-
Ok(resp) => {
234-
let status = reqwest::StatusCode::from_u16(resp.status)
235-
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
236-
if status.is_success() {
237-
AttemptOutcome::Success(resp)
238-
} else {
239-
let retry_after = resp
240-
.retry_after
241-
.as_deref()
242-
.and_then(|v| RetryConfig::parse_retry_after(Some(v)));
243-
if self.retry_config.is_retryable_status(status) {
244-
AttemptOutcome::Retryable {
245-
status,
246-
body: resp.error_body,
247-
retry_after,
248-
}
192+
{
193+
let mut request_body = self.build_request(messages, system, tools);
194+
request_body["stream"] = serde_json::json!(true);
195+
196+
let url = format!("{}/v1/messages", self.base_url);
197+
198+
let headers = vec![
199+
("x-api-key", self.api_key.expose()),
200+
("anthropic-version", "2023-06-01"),
201+
];
202+
203+
let streaming_resp = crate::retry::with_retry(&self.retry_config, |_attempt| {
204+
let http = &self.http;
205+
let url = &url;
206+
let headers = headers.clone();
207+
let request_body = &request_body;
208+
async move {
209+
match http.post_streaming(url, headers, request_body).await {
210+
Ok(resp) => {
211+
let status = reqwest::StatusCode::from_u16(resp.status)
212+
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
213+
if status.is_success() {
214+
AttemptOutcome::Success(resp)
249215
} else {
250-
AttemptOutcome::Fatal(anyhow::anyhow!(
251-
"Anthropic API error at {} ({}): {}", url, status, resp.error_body
252-
))
216+
let retry_after = resp
217+
.retry_after
218+
.as_deref()
219+
.and_then(|v| RetryConfig::parse_retry_after(Some(v)));
220+
if self.retry_config.is_retryable_status(status) {
221+
AttemptOutcome::Retryable {
222+
status,
223+
body: resp.error_body,
224+
retry_after,
225+
}
226+
} else {
227+
AttemptOutcome::Fatal(anyhow::anyhow!(
228+
"Anthropic API error at {} ({}): {}",
229+
url,
230+
status,
231+
resp.error_body
232+
))
233+
}
253234
}
254235
}
236+
Err(e) => AttemptOutcome::Fatal(anyhow::anyhow!(
237+
"Failed to send streaming request: {}",
238+
e
239+
)),
255240
}
256-
Err(e) => AttemptOutcome::Fatal(anyhow::anyhow!(
257-
"Failed to send streaming request: {}", e
258-
)),
259241
}
260-
}
261-
})
262-
.await?;
263-
264-
let (tx, rx) = mpsc::channel(100);
265-
266-
let mut stream = streaming_resp.byte_stream;
267-
tokio::spawn(async move {
268-
let mut buffer = String::new();
269-
let mut content_blocks: Vec<ContentBlock> = Vec::new();
270-
let mut current_tool_id = String::new();
271-
let mut current_tool_name = String::new();
272-
let mut current_tool_input = String::new();
273-
let mut usage = TokenUsage::default();
274-
let mut stop_reason = None;
275-
276-
while let Some(chunk_result) = stream.next().await {
277-
let chunk = match chunk_result {
278-
Ok(c) => c,
279-
Err(e) => {
280-
tracing::error!("Stream error: {}", e);
281-
break;
282-
}
283-
};
242+
})
243+
.await?;
284244

285-
buffer.push_str(&String::from_utf8_lossy(&chunk));
245+
let (tx, rx) = mpsc::channel(100);
246+
247+
let mut stream = streaming_resp.byte_stream;
248+
tokio::spawn(async move {
249+
let mut buffer = String::new();
250+
let mut content_blocks: Vec<ContentBlock> = Vec::new();
251+
let mut current_tool_id = String::new();
252+
let mut current_tool_name = String::new();
253+
let mut current_tool_input = String::new();
254+
let mut usage = TokenUsage::default();
255+
let mut stop_reason = None;
256+
257+
while let Some(chunk_result) = stream.next().await {
258+
let chunk = match chunk_result {
259+
Ok(c) => c,
260+
Err(e) => {
261+
tracing::error!("Stream error: {}", e);
262+
break;
263+
}
264+
};
286265

287-
while let Some(event_end) = buffer.find("\n\n") {
288-
let event_data: String = buffer.drain(..event_end).collect();
289-
buffer.drain(..2);
266+
buffer.push_str(&String::from_utf8_lossy(&chunk));
290267

291-
for line in event_data.lines() {
292-
if let Some(data) = line.strip_prefix("data: ") {
293-
if data == "[DONE]" {
294-
continue;
295-
}
268+
while let Some(event_end) = buffer.find("\n\n") {
269+
let event_data: String = buffer.drain(..event_end).collect();
270+
buffer.drain(..2);
296271

297-
if let Ok(event) = serde_json::from_str::<AnthropicStreamEvent>(data) {
298-
match event {
299-
AnthropicStreamEvent::ContentBlockStart {
300-
index: _,
301-
content_block,
302-
} => match content_block {
303-
AnthropicContentBlock::Text { .. } => {}
304-
AnthropicContentBlock::ToolUse { id, name, .. } => {
305-
current_tool_id = id.clone();
306-
current_tool_name = name.clone();
307-
current_tool_input.clear();
308-
let _ = tx
309-
.send(StreamEvent::ToolUseStart { id, name })
310-
.await;
311-
}
312-
},
313-
AnthropicStreamEvent::ContentBlockDelta { index: _, delta } => {
314-
match delta {
272+
for line in event_data.lines() {
273+
if let Some(data) = line.strip_prefix("data: ") {
274+
if data == "[DONE]" {
275+
continue;
276+
}
277+
278+
if let Ok(event) =
279+
serde_json::from_str::<AnthropicStreamEvent>(data)
280+
{
281+
match event {
282+
AnthropicStreamEvent::ContentBlockStart {
283+
index: _,
284+
content_block,
285+
} => match content_block {
286+
AnthropicContentBlock::Text { .. } => {}
287+
AnthropicContentBlock::ToolUse { id, name, .. } => {
288+
current_tool_id = id.clone();
289+
current_tool_name = name.clone();
290+
current_tool_input.clear();
291+
let _ = tx
292+
.send(StreamEvent::ToolUseStart { id, name })
293+
.await;
294+
}
295+
},
296+
AnthropicStreamEvent::ContentBlockDelta {
297+
index: _,
298+
delta,
299+
} => match delta {
315300
AnthropicDelta::TextDelta { text } => {
316301
let _ = tx.send(StreamEvent::TextDelta(text)).await;
317302
}
@@ -323,11 +308,10 @@ impl LlmClient for AnthropicClient {
323308
))
324309
.await;
325310
}
326-
}
327-
}
328-
AnthropicStreamEvent::ContentBlockStop { index: _ } => {
329-
if !current_tool_id.is_empty() {
330-
let input: serde_json::Value =
311+
},
312+
AnthropicStreamEvent::ContentBlockStop { index: _ } => {
313+
if !current_tool_id.is_empty() {
314+
let input: serde_json::Value =
331315
serde_json::from_str(&current_tool_input)
332316
.unwrap_or_else(|e| {
333317
tracing::warn!(
@@ -341,60 +325,58 @@ impl LlmClient for AnthropicClient {
341325
)
342326
})
343327
});
344-
content_blocks.push(ContentBlock::ToolUse {
345-
id: current_tool_id.clone(),
346-
name: current_tool_name.clone(),
347-
input,
348-
});
349-
current_tool_id.clear();
350-
current_tool_name.clear();
351-
current_tool_input.clear();
328+
content_blocks.push(ContentBlock::ToolUse {
329+
id: current_tool_id.clone(),
330+
name: current_tool_name.clone(),
331+
input,
332+
});
333+
current_tool_id.clear();
334+
current_tool_name.clear();
335+
current_tool_input.clear();
336+
}
352337
}
338+
AnthropicStreamEvent::MessageStart { message } => {
339+
usage.prompt_tokens = message.usage.input_tokens;
340+
}
341+
AnthropicStreamEvent::MessageDelta {
342+
delta,
343+
usage: msg_usage,
344+
} => {
345+
stop_reason = Some(delta.stop_reason);
346+
usage.completion_tokens = msg_usage.output_tokens;
347+
usage.total_tokens =
348+
usage.prompt_tokens + usage.completion_tokens;
349+
}
350+
AnthropicStreamEvent::MessageStop => {
351+
crate::telemetry::record_llm_usage(
352+
usage.prompt_tokens,
353+
usage.completion_tokens,
354+
usage.total_tokens,
355+
stop_reason.as_deref(),
356+
);
357+
358+
let response = LlmResponse {
359+
message: Message {
360+
role: "assistant".to_string(),
361+
content: std::mem::take(&mut content_blocks),
362+
reasoning_content: None,
363+
},
364+
usage: usage.clone(),
365+
stop_reason: stop_reason.clone(),
366+
};
367+
let _ = tx.send(StreamEvent::Done(response)).await;
368+
}
369+
_ => {}
353370
}
354-
AnthropicStreamEvent::MessageStart { message } => {
355-
usage.prompt_tokens = message.usage.input_tokens;
356-
}
357-
AnthropicStreamEvent::MessageDelta {
358-
delta,
359-
usage: msg_usage,
360-
} => {
361-
stop_reason = Some(delta.stop_reason);
362-
usage.completion_tokens = msg_usage.output_tokens;
363-
usage.total_tokens =
364-
usage.prompt_tokens + usage.completion_tokens;
365-
}
366-
AnthropicStreamEvent::MessageStop => {
367-
crate::telemetry::record_llm_usage(
368-
usage.prompt_tokens,
369-
usage.completion_tokens,
370-
usage.total_tokens,
371-
stop_reason.as_deref(),
372-
);
373-
374-
let response = LlmResponse {
375-
message: Message {
376-
role: "assistant".to_string(),
377-
content: std::mem::take(&mut content_blocks),
378-
reasoning_content: None,
379-
},
380-
usage: usage.clone(),
381-
stop_reason: stop_reason.clone(),
382-
};
383-
let _ = tx.send(StreamEvent::Done(response)).await;
384-
}
385-
_ => {}
386371
}
387372
}
388373
}
389374
}
390375
}
391-
}
392-
});
376+
});
393377

394-
Ok(rx)
378+
Ok(rx)
395379
}
396-
.instrument(span)
397-
.await
398380
}
399381
}
400382

0 commit comments

Comments
 (0)