From 508806a408f702502379838ceda5375c31217246 Mon Sep 17 00:00:00 2001
From: oudecheng <13802883547@139.com>
Date: Sat, 15 Aug 2026 18:16:50 +0800
Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=E5=AE=9A=E6=97=B6=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1=E6=89=A7=E8=A1=8C=E5=89=8D=E8=87=AA=E5=8A=A8=E8=A1=A5?=
 =?UTF-8?q?=E9=BD=90=20topic=EF=BC=8C=E4=BF=AE=E5=A4=8D=20agent=20?=
 =?UTF-8?q?=E7=A9=BA=E5=8E=86=E5=8F=B2=E6=89=A7=E8=A1=8C?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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 路径
---
 src/gateway/execution.rs | 12 +++++++---
 src/gateway/session.rs   | 48 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 57 insertions(+), 3 deletions(-)

diff --git a/src/gateway/execution.rs b/src/gateway/execution.rs
index ff3ea4c..35b9b85 100644
--- a/src/gateway/execution.rs
+++ b/src/gateway/execution.rs
@@ -15,6 +15,7 @@ use tokio::sync::Mutex;
 use super::compaction::schedule_background_history_compaction;
 use super::message_prepare::enrich_user_content_with_media_refs;
 use super::session::Session;
+use super::session_pool::is_scheduler_chat_id;
 use super::wait_coordinator::SessionWaitCoordinator;
 use crate::tools::WaitCoordinator;
 
@@ -442,11 +443,16 @@ impl AgentExecutionService {
         // 获取该 topic 的串行锁（与普通消息路径共享，保证串行执行）
         // 定时任务由调度器触发，无用户消息竞态；在锁前一次性捕获 topic_id，
         // 锁后复用同一值作为 original_topic_id，保证锁键与写入目标一致。
+        //
+        // 关键：若该 chat 尚无 topic（scheduler/ 虚拟会话从不经用户消息分配
+        // topic），必须先补齐，否则后续消息无法进入按 topic 键化的内存历史，
+        // agent 将以空历史执行（任务 prompt 丢失）。
         let (serial_lock, session_store, lock_key, lock_time_topic_id) = {
             let mut session_guard = request.session.lock().await;
-            let tid = session_guard
-                .current_topic(request.chat_id)
-                .map(|s| s.to_string());
+            let tid = match session_guard.current_topic(request.chat_id) {
+                Some(topic_id) => Some(topic_id.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();
             session_guard.ensure_sub_done_channel(&lock_key);
             (
diff --git a/src/gateway/session.rs b/src/gateway/session.rs
index 8392cb5..3879760 100644
--- a/src/gateway/session.rs
+++ b/src/gateway/session.rs
@@ -388,6 +388,54 @@ impl Session {
         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_histories 中独立存储，切换不互斥。
     pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> {