fix(gateway): 定时任务执行前自动补齐 topic,修复 agent 空历史执行

scheduler/ 虚拟会话从不经用户消息分配 topic,导致定时任务的系统提示词与任务 prompt 只落库、不进内存历史(append_persisted_message 在 topic=None 时跳过内存写入),agent 以空历史执行,任务实际未运行、消息无法送达会话。

- Session 新增 ensure_topic_for_chat:内存无 topic 时从 DB 恢复最近活跃 topic,无则自动创建(先确保持久会话以规避外键约束)

- prepare_and_execute_scheduled_task 捕获 topic 前调用 ensure,保证 original_topic_id 恒为 Some

- append_persisted_message 无 topic 时回退到 chat_id 键内存历史,兜底所有无 topic 路径
This commit is contained in:
oudecheng 2026-08-15 18:16:50 +08:00
parent 73c25e5a20
commit 508806a408
2 changed files with 57 additions and 3 deletions

View File

@ -15,6 +15,7 @@ use tokio::sync::Mutex;
use super::compaction::schedule_background_history_compaction; use super::compaction::schedule_background_history_compaction;
use super::message_prepare::enrich_user_content_with_media_refs; use super::message_prepare::enrich_user_content_with_media_refs;
use super::session::Session; use super::session::Session;
use super::session_pool::is_scheduler_chat_id;
use super::wait_coordinator::SessionWaitCoordinator; use super::wait_coordinator::SessionWaitCoordinator;
use crate::tools::WaitCoordinator; use crate::tools::WaitCoordinator;
@ -442,11 +443,16 @@ impl AgentExecutionService {
// 获取该 topic 的串行锁(与普通消息路径共享,保证串行执行) // 获取该 topic 的串行锁(与普通消息路径共享,保证串行执行)
// 定时任务由调度器触发,无用户消息竞态;在锁前一次性捕获 topic_id // 定时任务由调度器触发,无用户消息竞态;在锁前一次性捕获 topic_id
// 锁后复用同一值作为 original_topic_id保证锁键与写入目标一致。 // 锁后复用同一值作为 original_topic_id保证锁键与写入目标一致。
//
// 关键:若该 chat 尚无 topicscheduler/ 虚拟会话从不经用户消息分配
// topic必须先补齐否则后续消息无法进入按 topic 键化的内存历史,
// agent 将以空历史执行(任务 prompt 丢失)。
let (serial_lock, session_store, lock_key, lock_time_topic_id) = { let (serial_lock, session_store, lock_key, lock_time_topic_id) = {
let mut session_guard = request.session.lock().await; let mut session_guard = request.session.lock().await;
let tid = session_guard let tid = match session_guard.current_topic(request.chat_id) {
.current_topic(request.chat_id) Some(topic_id) => Some(topic_id.to_string()),
.map(|s| s.to_string()); None => Some(session_guard.ensure_topic_for_chat(request.chat_id)?),
};
let lock_key = tid.as_deref().unwrap_or(request.chat_id).to_string(); let lock_key = tid.as_deref().unwrap_or(request.chat_id).to_string();
session_guard.ensure_sub_done_channel(&lock_key); session_guard.ensure_sub_done_channel(&lock_key);
( (

View File

@ -388,6 +388,54 @@ impl Session {
self.history.chat_topic(chat_id) self.history.chat_topic(chat_id)
} }
/// 确保指定 chat 存在当前话题,返回话题 ID。
///
/// 内存中无当前话题时,先从数据库恢复最近活跃的话题;数据库中也没有则
/// 自动创建默认话题。定时任务路径(尤其 scheduler/ 虚拟会话)不经过用户
/// 消息的 topic 分配流程,若不补齐话题,消息将无法进入按 topic 键化的
/// 内存历史,导致 agent 以空历史执行。
pub fn ensure_topic_for_chat(&mut self, chat_id: &str) -> Result<String, AgentError> {
if let Some(topic_id) = self.history.chat_topic(chat_id) {
return Ok(topic_id.to_string());
}
// create_topic 依赖 sessions 行存在(外键约束),先确保持久会话已建立
self.ensure_persistent_session(chat_id)?;
let session_id = self.persistent_session_id(chat_id);
let topics = self
.store
.list_topics(&session_id)
.map_err(|e| AgentError::Other(format!("Failed to list topics: {}", e)))?;
if let Some(latest_topic) = topics.first() {
let topic_id = latest_topic.id.clone();
self.history.set_chat_topic(chat_id, topic_id.clone());
tracing::info!(
chat_id = %chat_id,
topic_id = %topic_id,
"Restored current topic from database"
);
return Ok(topic_id);
}
let title = format!("话题 {}", chrono::Local::now().format("%m/%d %H:%M"));
let topic = self
.store
.create_topic(&session_id, &title, None)
.map_err(|e| {
AgentError::Other(format!("Failed to auto-create default topic: {}", e))
})?;
self.history.set_chat_topic(chat_id, topic.id.clone());
tracing::info!(
chat_id = %chat_id,
topic_id = %topic.id,
session_id = %session_id,
"Auto-created default topic for chat"
);
Ok(topic.id)
}
/// 切换话题 - 设置当前 topic 并加载新话题的历史到内存 /// 切换话题 - 设置当前 topic 并加载新话题的历史到内存
/// 不同 topic 的历史在 topic_histories 中独立存储,切换不互斥。 /// 不同 topic 的历史在 topic_histories 中独立存储,切换不互斥。
pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> { pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> {