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

refactor(session): make cron stateless, split session lock, eliminate deadlock

Browse Source 
- handle_cron_message: bypass Session entirely, use create_cron_agent() for direct LLM call; cron LLM delivers results via send_message tool
- handle_message: split session lock into Phase1(prepare)/Phase2(LLM unlocked)/Phase3(persist), allowing send_message to lock freely
- send_message: restore lock().await (remove try_lock workaround)
- scheduler: remove outbound publish (LLM handles delivery via send_message)
This commit is contained in:
xiaoski 2026-05-13 10:40:58 +08:00
parent 99a57a816a
commit ebbf7e4036
3 changed files with 85 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/gateway/mod.rs
View File

@ -228,7 +228,6 @@ impl GatewayState {
let sched = Arc::new(Scheduler::new(
self.storage.clone(),
self.session_manager.clone(),
self.channel_manager.bus(),
scheduler_config,
));
tokio::spawn(async move {

32
src/scheduler/mod.rs
View File

@ -4,11 +4,12 @@ use std::sync::Arc;
use std::time::Instant;
use tokio::time;
use crate::bus::MessageBus;
use crate::config::SchedulerConfig;
use crate::session::session::HandleResult;
use crate::session::SessionManager;
use crate::storage::{JobRun, ScheduledJob, Storage};
use crate::storage::ScheduledJob;
use crate::storage::Storage;
use crate::storage::JobRun;
pub use types::Schedule;
@ -53,7 +54,6 @@ fn now_ms() -> i64 {
pub struct Scheduler {
storage: Arc<Storage>,
session_manager: Arc<SessionManager>,
bus: Arc<MessageBus>,
config: SchedulerConfig,
}
@ -61,13 +61,11 @@ impl Scheduler {
pub fn new(
storage: Arc<Storage>,
session_manager: Arc<SessionManager>,
bus: Arc<MessageBus>,
config: SchedulerConfig,
) -> Self {
Self {
storage,
session_manager,
bus,
config,
}
}
@ -136,18 +134,6 @@ impl Scheduler {
match result {
Ok(HandleResult::AgentResponse(output)) => {
let outbound = crate::bus::OutboundMessage {
channel: job.channel.clone(),
chat_id: job.chat_id.clone(),
content: output.clone(),
reply_to: None,
media: vec![],
metadata: std::collections::HashMap::new(),
};
if let Err(e) = self.bus.publish_outbound(outbound).await {
tracing::warn!(job_id = %job.id, "scheduler: failed to publish outbound: {}", e);
}
let output_truncated = if output.len() > 8000 {
format!("{}...[truncated]", &output[..output.ceil_char_boundary(8000)])
} else {
@ -180,18 +166,6 @@ impl Scheduler {
);
}
Ok(HandleResult::CommandOutput(output)) => {
let outbound = crate::bus::OutboundMessage {
channel: job.channel.clone(),
chat_id: job.chat_id.clone(),
content: output.clone(),
reply_to: None,
media: vec![],
metadata: std::collections::HashMap::new(),
};
if let Err(e) = self.bus.publish_outbound(outbound).await {
tracing::warn!(job_id = %job.id, "scheduler: failed to publish outbound: {}", e);
}
let run = JobRun {
id: 0,
job_id: job.id.clone(),

207
src/session/session.rs
View File

@ -839,6 +839,19 @@ impl SessionManager {
self.tools.clone()
}
/// 为定时任务创建一个无 session 绑定的 AgentLoop
pub fn create_cron_agent(&self) -> Result<AgentLoop, AgentError> {
let provider = create_provider(self.provider_config.clone())
.map_err(|e| AgentError::Other(format!("failed to create cron provider: {}", e)))?;
Ok(AgentLoop::with_provider_and_tools(
Arc::from(provider),
self.tools.clone(),
self.provider_config.max_tool_iterations,
self.provider_config.model_id.clone(),
self.provider_config.workspace_dir.clone(),
).with_context_window(self.provider_config.token_limit))
}
/// 获取所有可用的斜杠命令
pub fn get_slash_commands(&self) -> &[SlashCommand] {
SLASH_COMMANDS
@ -1320,7 +1333,8 @@ impl SessionManager {
});
}
let response: String = {
// Phase 1: prepare data under session lock
let (agent, history, system_prompt) = {
let mut session_guard = session.lock().await;
let media_refs: Vec<String> = media.iter().map(|m| m.path.clone()).collect();
@ -1335,10 +1349,8 @@ impl SessionManager {
let history = session_guard.get_history().to_vec();
// Build skills prompt
let skills_prompt = self.skills_loader.build_skills_prompt();
// Fetch memory context
let memory_context = match self.memory_manager.recall(content, 5, Some(crate::memory::MemoryCategory::Knowledge), None).await {
Ok(entries) if !entries.is_empty() => {
Some(entries.iter()
@ -1353,9 +1365,6 @@ impl SessionManager {
_ => None,
};
// Build combined system prompt and inject at position 0 AFTER compression.
// This ensures AgentLoop.process() sees a system message without it participating
// in context compression (system prompt is dynamic and should not be persisted).
let system_prompt = session_guard.build_system_prompt(&skills_prompt, memory_context.as_deref());
let result = session_guard.compressor
@ -1371,7 +1380,6 @@ impl SessionManager {
history.insert(0, ChatMessage::system(system_prompt.clone()));
// Persist consolidation state
let now = chrono::Utc::now().timestamp_millis();
session_guard.last_consolidated_at = Some(now);
if let Err(e) = session_guard.persist_session_meta().await {
@ -1379,13 +1387,17 @@ impl SessionManager {
}
let agent = session_guard.create_agent_with_notify(notify_tx)?;
(agent, history, system_prompt)
}; // session lock released — send_message can now lock freely
// Try LLM call; on context overflow, re-compress with tighter window and retry once.
// Phase 2: LLM call (no session lock held)
let result = match agent.process(history).await {
Ok(r) => r,
Err(AgentError::LlmError(ref msg))
if is_context_overflow_error(msg) =>
{
let retry_history = {
let mut session_guard = session.lock().await;
let new_window = crate::agent::ContextCompressor::parse_context_limit_from_error(msg)
.unwrap_or(session_guard.compressor_threshold());
tracing::warn!(
@ -1403,28 +1415,30 @@ impl SessionManager {
}
}
let mut retry = retry_result.history;
retry.insert(0, ChatMessage::system(system_prompt));
agent.process(retry).await
retry.insert(0, ChatMessage::system(system_prompt.clone()));
retry
}; // lock released again for retry
agent.process(retry_history).await
.inspect_err(|e| {
tracing::error!(error = %e, "Agent retry after context compression failed");
})?
}
Err(e) => {
tracing::error!(
error = %e,
elapsed = ?"LLM call in handle_message failed",
"Agent processing error — propagating to caller"
);
tracing::error!(error = %e, "Agent processing error — propagating to caller");
return Err(e);
},
};
// Phase 3: persist results under session lock
let response: String = {
let mut session_guard = session.lock().await;
for msg in result.emitted_messages {
session_guard.add_message(msg, true).await
.map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
}
// Check if we need to generate a title (after 10 user messages)
if session_guard.should_generate_title()
&& let Err(e) = session_guard.generate_title().await {
tracing::warn!("failed to generate title: {}", e);
@ -1447,9 +1461,10 @@ impl SessionManager {
/// Handle a message triggered by a scheduled cron job.
///
/// This is similar to `handle_message`, but the user message is created with
/// `SourceKind::ExternalTrigger` source metadata so that the cron job identity
/// is preserved in the conversation history and database.
/// Runs in a stateless manner: no session creation, no history persistence.
/// The cron system prompt instructs the LLM to deliver results via the
/// `send_message` tool, which handles both delivery and history writing
/// on the target session.
pub async fn handle_cron_message(
&self,
channel: &str,
@ -1458,133 +1473,45 @@ impl SessionManager {
job_id: &str,
job_name: &str,
) -> Result<HandleResult, AgentError> {
use crate::bus::{MessageSource, SourceKind};
let unified_id = self.resolve_dialog_id(channel, chat_id).await?;
tracing::debug!(unified_id = %unified_id, job_id = %job_id, "handle_cron_message resolved");
let session = self.get_or_create_session(&unified_id).await?;
CURRENT_SOURCE_SESSION.scope(Some(unified_id.to_string()), async {
let (notify_tx, mut notify_rx) = tokio::sync::mpsc::unbounded_channel();
{
use std::collections::HashMap;
use crate::bus::OutboundMessage;
let bus = self.bus.clone();
let ch = channel.to_string();
let cid = chat_id.to_string();
tokio::spawn(async move {
while let Some(notif) = notify_rx.recv().await {
let mut metadata = HashMap::new();
metadata.insert("_type".to_string(), "notification".to_string());
let outbound = OutboundMessage {
channel: ch.clone(),
chat_id: cid.clone(),
content: notif,
reply_to: None,
media: vec![],
metadata,
};
let _ = bus.publish_outbound(outbound).await;
}
});
}
let response: String = {
let mut session_guard = session.lock().await;
let source = MessageSource {
kind: SourceKind::ExternalTrigger,
from_channel: Some(channel.to_string()),
from_session: None,
from_user_id: None,
system_name: Some(job_name.to_string()),
task_id: Some(job_id.to_string()),
};
let user_message = session_guard.create_user_message_with_source(prompt, vec![], source);
session_guard.add_message(user_message, true).await
.map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
let history = session_guard.get_history().to_vec();
let skills_prompt = self.skills_loader.build_skills_prompt();
let system_prompt = session_guard.build_system_prompt(&skills_prompt, None);
let cron_context = format!(
"\n\n## 定时任务执行\n\n\
你正在执行定时任务「{}」({})。\n\
目标渠道: {}:{}\n\n\
定时任务执行规则：\n\
- 这不是聊天对话，没有人会回复你，不要等待用户输入\n\
- 你的职责是根据任务指令直接生成要发送的消息内容\n\
- 只输出最终消息，不要输出中间思考过程或分析\n\
- 系统会自动将你的回复推送到目标渠道，不要使用 send_message 工具\n\
- 你的最终回复就是推送给用户的消息原文",
job_name, job_id, channel, chat_id
);
let full_system_prompt = format!("{}{}", system_prompt, cron_context);
// Inject system prompt AFTER compression so it doesn't participate
// in context compression (system prompt is dynamic and should not be persisted).
let mut history = session_guard.compressor
.compress_if_needed(history)
let base_prompt = build_system_prompt(
&self.provider_config.workspace_dir,
&self.provider_config.model_id,
&self.tools,
Some(&format!("cron:{}:{}", job_name, job_id)),
None,
false,
);
let cron_context = format!(
"## 定时任务执行\n\n\
你正在执行定时任务「{job_name}」({job_id})。\n\
目标渠道: {channel}:{chat_id}\n\n\
规则:\n\
- 这不是聊天对话，没有用户会直接看到你的输出\n\
- 你必须使用 send_message 工具将最终结果发送到目标渠道\n\
- send_message 格式: target_chat_id=\"{channel}:{chat_id}\", content=\"消息内容\"\n\
- 可以调用其他工具收集信息、处理任务，但最终消息必须通过 send_message 发送\n\
- 只输出最终消息内容，不要输出中间思考过程或分析！"
);
let full_system_prompt = format!("{}\n\n{}\n\n{}", base_prompt, skills_prompt, cron_context);
let history = vec![
ChatMessage::system(full_system_prompt),
ChatMessage::user(prompt),
];
let agent = self.create_cron_agent()?;
let source_session = format!("cron:{}", job_name);
let result = CURRENT_SOURCE_SESSION.scope(Some(source_session), async {
agent.process(history).await
})
.await
.inspect_err(|e| {
tracing::warn!(error = %e, "Context compression failed in handle_cron_message");
})?
.history;
history.insert(0, ChatMessage::system(full_system_prompt));
let agent = session_guard.create_agent_with_notify(notify_tx)?;
let result = agent.process(history).await
.inspect_err(|e| {
tracing::error!(error = %e, "Agent processing error in handle_cron_message");
tracing::error!(error = %e, job_id = %job_id, "Cron agent processing error");
})?;
for msg in result.emitted_messages {
session_guard.add_message(msg, true).await
.map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
}
if session_guard.should_generate_title()
&& let Err(e) = session_guard.generate_title().await {
tracing::warn!("failed to generate title: {}", e);
}
let raw_response = result.final_response.content;
let prefix = format!(
"[message from cron:{}({})]\n",
job_name, job_id
);
let prefixed_response = format!("{}{}", prefix, raw_response);
let source = MessageSource {
kind: SourceKind::CrossChannel,
from_channel: Some("cron".to_string()),
from_session: Some(format!("{}:{}", job_name, job_id)),
from_user_id: None,
system_name: Some(job_name.to_string()),
task_id: Some(job_id.to_string()),
};
let msg = ChatMessage::assistant_with_source(prefixed_response.clone(), source);
session_guard.add_message(msg, true).await
.map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;
prefixed_response
};
#[cfg(debug_assertions)]
tracing::debug!(
channel = %channel,
chat_id = %chat_id,
job_id = %job_id,
response_len = %response.len(),
"Cron agent response received"
);
Ok(HandleResult::AgentResponse(response))
}).await
Ok(HandleResult::AgentResponse(result.final_response.content))
}
pub async fn clear_session_history(&self, unified_id: &UnifiedSessionId) -> Result<(), AgentError> {