xiaoski/PicoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(runtime): eliminate unowned persistence tasks

Browse Source
This commit is contained in:
xiaoxixi 2026-07-14 11:50:55 +08:00
parent 59ecb27c06
commit 24d3e26b43
4 changed files with 71 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/agent/context_compressor.rs
View File

@ -536,22 +536,19 @@ impl ContextCompressor {
summary summary
); );
let key = format!("ctx_compressed_{}", uuid::Uuid::new_v4()); let key = format!("ctx_compressed_{}", uuid::Uuid::new_v4());
let mm = self.memory.clone(); if let Err(e) = self
let sid = self.session_id.clone(); .memory
tokio::spawn(async move {
if let Err(e) = mm
.store( .store(
&key, &key,
&timeline_content, &timeline_content,
crate::memory::MemoryCategory::Timeline, crate::memory::MemoryCategory::Timeline,
sid.as_deref(), self.session_id.as_deref(),
Some(0.3), Some(0.3),
) )
.await .await
{ {
tracing::warn!(error = %e, "Failed to store compressed context as timeline"); tracing::warn!(error = %e, "Failed to store compressed context as timeline");
} }
});
// Add summary as a special user message // Add summary as a special user message
new_messages.push(ChatMessage::user(format!( new_messages.push(ChatMessage::user(format!(

7
src/channels/feishu.rs
View File

@ -1252,10 +1252,6 @@ impl FeishuChannel {
forwarded_metadata.insert("feishu.parent_id".to_string(), pid.clone()); forwarded_metadata.insert("feishu.parent_id".to_string(), pid.clone());
} }
// Publish to bus asynchronously
let channel = self.clone();
let bus = bus.clone();
tokio::spawn(async move {
#[cfg(debug_assertions)] #[cfg(debug_assertions)]
tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, content_len = %parsed.content.len(), media_count = %parsed.media.len(), "Publishing message to bus"); tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, content_len = %parsed.content.len(), media_count = %parsed.media.len(), "Publishing message to bus");
let msg = crate::bus::InboundMessage { let msg = crate::bus::InboundMessage {
@ -1268,13 +1264,12 @@ impl FeishuChannel {
metadata: std::collections::HashMap::new(), metadata: std::collections::HashMap::new(),
forwarded_metadata, forwarded_metadata,
}; };
if let Err(e) = channel.handle_and_publish(&bus, &msg).await { if let Err(e) = self.handle_and_publish(&bus, &msg).await {
tracing::error!(error = %e, open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Failed to publish Feishu message to bus"); tracing::error!(error = %e, open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Failed to publish Feishu message to bus");
} else { } else {
#[cfg(debug_assertions)] #[cfg(debug_assertions)]
tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Message published to bus successfully"); tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Message published to bus successfully");
} }
});
} }
Ok(None) => {} Ok(None) => {}
Err(e) => { Err(e) => {

33
src/providers/anthropic.rs
View File

@ -330,24 +330,29 @@ impl LLMProvider for AnthropicProvider {
return Err(format!("API error ({}): {}", status.as_u16(), error_msg).into()); return Err(format!("API error ({}): {}", status.as_u16(), error_msg).into());
} }
let anthropic_resp: AnthropicResponse = serde_json::from_str(&body_text).map_err(|e| { let anthropic_resp: AnthropicResponse = match serde_json::from_str(&body_text) {
Ok(response) => response,
Err(e) => {
let err_msg = format!("decode error: {} | body: {}", e, &body_text); let err_msg = format!("decode error: {} | body: {}", e, &body_text);
if let Some(ref storage) = self.storage { if let Some(ref storage) = self.storage {
let name = self.name.clone();
let model = self.model_id.clone();
let req = req_body_str.clone();
let resp_body = body_text.clone();
let dur = start.elapsed().as_millis() as u64; let dur = start.elapsed().as_millis() as u64;
let err = err_msg.clone(); if let Err(error) = storage
let s = storage.clone(); .append_llm_call(
tokio::spawn(async move { &self.name,
let _ = s &self.model_id,
.append_llm_call(&name, &model, &req, Some(&resp_body), Some(&err), dur) &req_body_str,
.await; Some(&body_text),
}); Some(&err_msg),
dur,
)
.await
{
tracing::warn!("failed to persist LLM call (decode error): {}", error);
} }
err_msg }
})?; return Err(err_msg.into());
}
};
let mut content = String::new(); let mut content = String::new();
let mut reasoning = None; let mut reasoning = None;

30
src/providers/openai.rs
View File

@ -298,27 +298,29 @@ impl LLMProvider for OpenAIProvider {
return Err(error.into()); return Err(error.into());
} }
let openai_resp: OpenAIResponse = serde_json::from_str(&text).map_err(|e| { let openai_resp: OpenAIResponse = match serde_json::from_str(&text) {
Ok(response) => response,
Err(e) => {
let err_msg = format!("decode error: {} | body: {}", e, &text); let err_msg = format!("decode error: {} | body: {}", e, &text);
if let Some(ref storage) = self.storage { if let Some(ref storage) = self.storage {
let name = self.name.clone();
let model = self.model_id.clone();
let req = req_body_str.clone();
let resp = text.clone();
let dur = start.elapsed().as_millis() as u64; let dur = start.elapsed().as_millis() as u64;
let err = err_msg.clone(); if let Err(error) = storage
let s = storage.clone(); .append_llm_call(
tokio::spawn(async move { &self.name,
if let Err(e) = s &self.model_id,
.append_llm_call(&name, &model, &req, Some(&resp), Some(&err), dur) &req_body_str,
Some(&text),
Some(&err_msg),
dur,
)
.await .await
{ {
tracing::warn!("failed to persist LLM call (decode error): {}", e); tracing::warn!("failed to persist LLM call (decode error): {}", error);
} }
});
} }
err_msg return Err(err_msg.into());
})?; }
};
let first_choice = openai_resp let first_choice = openai_resp
.choices .choices