fix(session): /new 后仍停留在旧对话的问题
问题原因：/new 创建新 session 后，客户端下次发消息仍带着旧的 dialog_id，导致服务端找到旧 session。 解决方案：在 SessionManager 中新增 current_sessions 跟踪 每个 channel:chat_id 的当前活跃 session： - create_session / get_or_create_session 时更新 current_sessions - switch_dialog / delete_dialog 时同步更新 current_sessions - handle_message 无 dialog_id 时优先使用 current_sessions
This commit is contained in:
parent
7fac68b51c
commit
e235268133
@ -562,6 +562,8 @@ struct SessionManagerInner {
|
||||
sessions: HashMap<String, Arc<Mutex<Session>>>,
|
||||
session_timestamps: HashMap<String, Instant>,
|
||||
session_ttl: Duration,
|
||||
/// Current active session per channel:chat_id
|
||||
current_sessions: HashMap<String, String>,
|
||||
}
|
||||
|
||||
fn create_default_tools(skills_loader: Arc<SkillsLoader>) -> ToolRegistry {
|
||||
@ -662,6 +664,7 @@ impl SessionManager {
|
||||
sessions: HashMap::new(),
|
||||
session_timestamps: HashMap::new(),
|
||||
session_ttl: Duration::from_secs(session_ttl_hours * 3600),
|
||||
current_sessions: HashMap::new(),
|
||||
})),
|
||||
provider_config,
|
||||
tools,
|
||||
@ -876,7 +879,10 @@ impl SessionManager {
|
||||
let arc = Arc::new(Mutex::new(session));
|
||||
let inner = &mut *self.inner.lock().await;
|
||||
inner.sessions.insert(session_id_str.clone(), arc.clone());
|
||||
inner.session_timestamps.insert(session_id_str, Instant::now());
|
||||
inner.session_timestamps.insert(session_id_str.clone(), Instant::now());
|
||||
// Set as current session for this channel:chat_id
|
||||
let chat_scope = format!("{}:{}", channel, chat_id);
|
||||
inner.current_sessions.insert(chat_scope, session_id_str);
|
||||
|
||||
Ok((unified_id, title))
|
||||
}
|
||||
@ -904,7 +910,10 @@ impl SessionManager {
|
||||
|
||||
let arc = Arc::new(Mutex::new(session));
|
||||
inner.sessions.insert(session_id_str.clone(), arc.clone());
|
||||
inner.session_timestamps.insert(session_id_str, Instant::now());
|
||||
inner.session_timestamps.insert(session_id_str.clone(), Instant::now());
|
||||
// Set as current session
|
||||
let chat_scope = format!("{}:{}", unified_id.channel, unified_id.chat_id);
|
||||
inner.current_sessions.insert(chat_scope, session_id_str);
|
||||
return Ok(arc);
|
||||
}
|
||||
Err(_) => {
|
||||
@ -926,7 +935,10 @@ impl SessionManager {
|
||||
|
||||
let arc = Arc::new(Mutex::new(session));
|
||||
inner.sessions.insert(session_id_str.clone(), arc.clone());
|
||||
inner.session_timestamps.insert(session_id_str, Instant::now());
|
||||
inner.session_timestamps.insert(session_id_str.clone(), Instant::now());
|
||||
// Set as current session
|
||||
let chat_scope = format!("{}:{}", unified_id.channel, unified_id.chat_id);
|
||||
inner.current_sessions.insert(chat_scope, session_id_str);
|
||||
Ok(arc)
|
||||
}
|
||||
|
||||
@ -956,6 +968,10 @@ impl SessionManager {
|
||||
let unified_id = UnifiedSessionId::new(channel, chat_id, dialog_id);
|
||||
// Ensure session is loaded into memory
|
||||
self.get_or_create_session(&unified_id).await?;
|
||||
// Update current session tracking
|
||||
let mut inner = self.inner.lock().await;
|
||||
let chat_scope = format!("{}:{}", channel, chat_id);
|
||||
inner.current_sessions.insert(chat_scope, unified_id.to_string());
|
||||
Ok(unified_id)
|
||||
}
|
||||
|
||||
@ -999,10 +1015,12 @@ impl SessionManager {
|
||||
self.storage.soft_delete_session(&session_id_str).await
|
||||
.map_err(|e| AgentError::Other(format!("failed to delete dialog: {}", e)))?;
|
||||
|
||||
// Remove from memory
|
||||
// Remove from memory and current sessions
|
||||
let mut inner = self.inner.lock().await;
|
||||
inner.sessions.remove(&session_id_str);
|
||||
inner.session_timestamps.remove(&session_id_str);
|
||||
let chat_scope = format!("{}:{}", session_id.channel, session_id.chat_id);
|
||||
inner.current_sessions.remove(&chat_scope);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@ -1025,20 +1043,54 @@ impl SessionManager {
|
||||
content: &str,
|
||||
media: Vec<crate::bus::MediaItem>,
|
||||
) -> Result<HandleResult, AgentError> {
|
||||
// Determine dialog_id: if not provided, find most recent active or create new
|
||||
// Determine dialog_id: if not provided, use current session or find active or create new
|
||||
let unified_id = if let Some(did) = dialog_id {
|
||||
UnifiedSessionId::new(channel, chat_id, did)
|
||||
} else {
|
||||
// Find active session from Storage
|
||||
let ttl_millis = self.inner.lock().await.session_ttl.as_millis() as i64;
|
||||
match self.storage.find_active_session(channel, chat_id, ttl_millis).await {
|
||||
Ok(Some(meta)) => {
|
||||
UnifiedSessionId::new(channel, chat_id, &meta.dialog_id)
|
||||
// Check if we have a current session tracked for this channel:chat_id
|
||||
let chat_scope = format!("{}:{}", channel, chat_id);
|
||||
let current_session_id = {
|
||||
let inner = self.inner.lock().await;
|
||||
inner.current_sessions.get(&chat_scope).cloned()
|
||||
};
|
||||
if let Some(current_id) = current_session_id {
|
||||
// Verify current session still exists in Storage
|
||||
match self.storage.get_session(¤t_id).await {
|
||||
Ok(meta) => {
|
||||
// Current session still valid
|
||||
let parts: Vec<&str> = current_id.split(':').collect();
|
||||
if parts.len() == 3 {
|
||||
UnifiedSessionId::new(channel, chat_id, parts[2])
|
||||
} else {
|
||||
// Malformed, fallback to find or create
|
||||
let ttl_millis = self.inner.lock().await.session_ttl.as_millis() as i64;
|
||||
match self.storage.find_active_session(channel, chat_id, ttl_millis).await {
|
||||
Ok(Some(m)) => UnifiedSessionId::new(channel, chat_id, &m.dialog_id),
|
||||
_ => {
|
||||
let (new_id, _) = self.create_session(channel, chat_id, None, String::new()).await?;
|
||||
new_id
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
// Current session no longer exists, create new
|
||||
let (new_id, _) = self.create_session(channel, chat_id, None, String::new()).await?;
|
||||
new_id
|
||||
}
|
||||
}
|
||||
Ok(None) | Err(_) => {
|
||||
// Create new session
|
||||
let (new_id, _) = self.create_session(channel, chat_id, None, String::new()).await?;
|
||||
new_id
|
||||
} else {
|
||||
// No current session tracked, find active or create new
|
||||
let ttl_millis = self.inner.lock().await.session_ttl.as_millis() as i64;
|
||||
match self.storage.find_active_session(channel, chat_id, ttl_millis).await {
|
||||
Ok(Some(meta)) => {
|
||||
UnifiedSessionId::new(channel, chat_id, &meta.dialog_id)
|
||||
}
|
||||
Ok(None) | Err(_) => {
|
||||
// Create new session
|
||||
let (new_id, _) = self.create_session(channel, chat_id, None, String::new()).await?;
|
||||
new_id
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user