From 4b93c844476d5fbce08c9214601cff7cd8e82c74 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Sat, 15 Aug 2026 18:17:29 +0800 Subject: [PATCH] =?UTF-8?q?test(gateway):=20=E5=AE=9A=E6=97=B6=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=20topic=20=E8=A1=A5=E9=BD=90=E4=B8=8E=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E9=80=81=E8=BE=BE=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增捕获请求的 mock OpenAI server，直接断言任务 prompt 进入 LLM 输入 - 静默任务按生产默认 fresh_session=true 连续执行两轮，验证清空-重建时序下 prompt 仍送达、topic 自动创建并持久化、系统提示词含送达提示 - ensure_topic_for_chat 幂等性与全新 chat 默认 topic 创建测试 - append_persisted_message 无 topic 回退 chat_id 键内存历史测试 - SessionManager/SessionLifecycleService 暴露 get_scheduler_session 供测试断言 --- src/gateway/session.rs | 354 ++++++++++++++++++++++++++++++- src/gateway/session_lifecycle.rs | 8 + 2 files changed, 353 insertions(+), 9 deletions(-) diff --git a/src/gateway/session.rs b/src/gateway/session.rs index 3879760..880d238 100644 --- a/src/gateway/session.rs +++ b/src/gateway/session.rs @@ -535,18 +535,25 @@ impl Session { // 只有当写入的 topic 匹配当前活跃 topic 时才更新内存历史。 // 当用户已切换到新 topic 时，旧 topic 的排队消息不应污染新 topic 的内存历史。 + // 完全无 topic 时回退到以 chat_id 为键的内存历史（与调用方 + // history_key = topic.unwrap_or(chat_id) 的约定一致），避免无 topic + // 路径的消息只落库、不进内存，导致 agent 拿到空历史。 let current_chat_topic = self.history.chat_topic(chat_id); - if topic_id.as_deref() == current_chat_topic { - if let Some(ref tid) = topic_id { + match topic_id.as_deref() { + Some(tid) if Some(tid) == current_chat_topic => { self.add_message(tid, message); } - } else { - tracing::info!( - chat_id = %chat_id, - write_topic_id = ?topic_id, - current_topic_id = ?current_chat_topic, - "Skipping memory history update: message belongs to a different topic" - ); + Some(_) => { + tracing::info!( + chat_id = %chat_id, + write_topic_id = ?topic_id, + current_topic_id = ?current_chat_topic, + "Skipping memory history update: message belongs to a different topic" + ); + } + None => { + self.add_message(chat_id, message); + } } // 更新 topic 的最后活跃时间 @@ -1105,6 +1112,11 @@ impl SessionManager { self.lifecycle.get(channel_name).await } + /// 获取定时任务专用 Session（不自动创建） + pub async fn get_scheduler_session(&self, channel_name: &str) -> Option>> { + self.lifecycle.get_scheduler_session(channel_name).await + } + /// 获取指定 chat 的当前话题（确保 session 存在，自动从数据库恢复） pub async fn get_current_topic( &self, @@ -1480,6 +1492,55 @@ mod tests { format!("http://{}", address) } + /// 捕获所有 LLM 请求体的 mock server，用于断言"哪些消息真正进入了 LLM 输入" + async fn start_mock_openai_server_capturing() -> (String, StdArc>>) + { + let captured: StdArc>> = + StdArc::new(std::sync::Mutex::new(Vec::new())); + let state = captured.clone(); + + async fn handle( + axum::extract::State(state): axum::extract::State>>>, + Json(body): Json, + ) -> Json { + if let Ok(mut guard) = state.lock() { + guard.push(body.clone()); + } + let model = body + .get("model") + .and_then(|value| value.as_str()) + .unwrap_or("unknown-model"); + + Json(json!({ + "id": "mock-response", + "model": model, + "choices": [ + { + "message": { + "content": "任务已完成", + "tool_calls": [] + } + } + ], + "usage": { + "prompt_tokens": 1, + "completion_tokens": 1, + "total_tokens": 2 + } + })) + } + + let app = Router::new() + .route("/chat/completions", post(handle)) + .with_state(state); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (format!("http://{}", address), captured) + } + async fn start_mock_openai_504_server() -> String { async fn handle() -> (StatusCode, &'static str) { (StatusCode::GATEWAY_TIMEOUT, "stream timeout") @@ -1761,6 +1822,281 @@ mod tests { assert!(scheduled_prompt.content.contains("你是邮箱待办同步助手。")); } + /// 回归：静默定时任务（scheduler/ 虚拟会话）必须—— + /// 1. 自动创建 topic，任务 prompt 进入内存历史并送达 LLM（修复前为空历史执行）； + /// 2. 持久化的定时系统提示词包含 send_session_message 主动送达提示。 + #[tokio::test] + async fn test_run_silent_agent_task_passes_prompt_and_creates_topic() { + let (base_url, captured) = start_mock_openai_server_capturing().await; + let provider_config = LLMProviderConfig { + provider_type: "openai".to_string(), + name: "default-provider".to_string(), + base_url, + api_key: "test-key".to_string(), + extra_headers: HashMap::new(), + model_id: "default-model".to_string(), + temperature: Some(0.0), + max_tokens: Some(32), + context_window_tokens: None, + model_extra: HashMap::new(), + max_tool_iterations: 1, + llm_timeout_secs: 30, + memory_maintenance_timeout_secs: 600, + max_retries: 3, + tool_result_max_chars: 100_000, + context_tool_result_trim_chars: 100_000, + max_images_in_context: 1, + max_image_age_rounds: 10, + }; + + let session_manager = SessionManager::new( + 100, + false, + "Asia/Shanghai".to_string(), + provider_config.clone(), + HashMap::from([("default".to_string(), provider_config)]), + Arc::new(SkillRuntime::default()), + HashSet::new(), + crate::config::TaskConfig::default(), + crate::config::SubagentsConfig::default(), + test_maintenance_config(), + Some(24), + crate::mcp::McpConfig::default(), + ) + .unwrap(); + + let outbound = session_manager + .run_silent_agent_task( + "test-channel", + "scheduler/silent-delivery-check", + Some("oc_notification_target"), + "检查网关连通性并用 send_session_message 通知用户", + // fresh_session=true 是 scheduler 生产路径的默认值：每次运行清空历史， + // 必须保证清空后任务 prompt 仍进入内存历史 + ScheduledAgentTaskOptions { + fresh_session: true, + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!(outbound.len(), 1); + + // 第二次运行（fresh_session 再次清空历史）prompt 仍必须送达 LLM + session_manager + .run_silent_agent_task( + "test-channel", + "scheduler/silent-delivery-check", + Some("oc_notification_target"), + "第二轮执行：检查网关连通性", + ScheduledAgentTaskOptions { + fresh_session: true, + ..Default::default() + }, + ) + .await + .unwrap(); + + // 1. 两次运行都应发起 LLM 请求，且各自的任务 prompt 出现在 user 消息中（非空历史执行） + let requests = captured.lock().unwrap().clone(); + assert!(requests.len() >= 2, "两次定时任务应各发起至少一次 LLM 请求"); + for expected in [ + "检查网关连通性并用 send_session_message 通知用户", + "第二轮执行：检查网关连通性", + ] { + let prompt_delivered = requests.iter().any(|request| { + request + .get("messages") + .and_then(|value| value.as_array()) + .map(|messages| { + messages.iter().any(|message| { + message.get("role").and_then(|role| role.as_str()) == Some("user") + && message + .get("content") + .and_then(|content| content.as_str()) + .is_some_and(|content| content.contains(expected)) + }) + }) + .unwrap_or(false) + }); + assert!( + prompt_delivered, + "定时任务 prompt “{}” 必须进入 LLM 输入，实际请求：{:?}", + expected, requests + ); + } + + // 2. scheduler/ 虚拟会话自动创建 topic，且消息挂在该 topic 下 + let scheduler_session = session_manager + .get_scheduler_session("test-channel") + .await + .expect("scheduler session should exist"); + let guard = scheduler_session.lock().await; + assert!( + guard + .current_topic("scheduler/silent-delivery-check") + .is_some(), + "定时任务应为虚拟会话设置当前 topic" + ); + let session_id = guard.persistent_session_id("scheduler/silent-delivery-check"); + let topics = guard.session_store().list_topics(&session_id).unwrap(); + assert!(!topics.is_empty(), "scheduler 虚拟会话应自动创建 topic"); + + let messages = guard + .store() + .load_messages_for_topic(&topics[0].id, Some(&session_id)) + .unwrap(); + assert!( + messages + .iter() + .any(|message| message.role == "user" && message.content.contains("检查网关连通性")), + "任务 prompt 应持久化到 topic" + ); + + // 3. 静默任务的系统提示词包含主动送达提示 + let scheduled_prompt = messages + .iter() + .find(|message| message.has_system_context(SYSTEM_CONTEXT_SCHEDULED_PROMPT)) + .expect("missing scheduled system prompt"); + assert!( + scheduled_prompt.content.contains("send_session_message"), + "静默任务系统提示词应包含 send_session_message 送达提示" + ); + } + + /// 回归：定时任务运行后 chat 自动拥有 topic；ensure_topic_for_chat 幂等， + /// 对全新 chat 会创建并持久化默认 topic。 + #[tokio::test] + async fn test_ensure_topic_for_chat_creates_and_is_idempotent() { + let base_url = start_mock_openai_server().await; + let provider_config = LLMProviderConfig { + provider_type: "openai".to_string(), + name: "default-provider".to_string(), + base_url, + api_key: "test-key".to_string(), + extra_headers: HashMap::new(), + model_id: "default-model".to_string(), + temperature: Some(0.0), + max_tokens: Some(32), + context_window_tokens: None, + model_extra: HashMap::new(), + max_tool_iterations: 1, + llm_timeout_secs: 30, + memory_maintenance_timeout_secs: 600, + max_retries: 3, + tool_result_max_chars: 100_000, + context_tool_result_trim_chars: 100_000, + max_images_in_context: 1, + max_image_age_rounds: 10, + }; + + let session_manager = SessionManager::new( + 100, + false, + "Asia/Shanghai".to_string(), + provider_config.clone(), + HashMap::from([("default".to_string(), provider_config)]), + Arc::new(SkillRuntime::default()), + HashSet::new(), + crate::config::TaskConfig::default(), + crate::config::SubagentsConfig::default(), + test_maintenance_config(), + Some(24), + crate::mcp::McpConfig::default(), + ) + .unwrap(); + + session_manager + .run_scheduled_agent_task( + "test-channel", + "chat-topic-auto", + "执行任务A", + ScheduledAgentTaskOptions::default(), + ) + .await + .unwrap(); + + let session = session_manager.get("test-channel").await.unwrap(); + let mut guard = session.lock().await; + + let tid1 = guard + .current_topic("chat-topic-auto") + .map(str::to_string) + .expect("定时任务运行后 chat 应自动拥有 topic"); + let tid2 = guard.ensure_topic_for_chat("chat-topic-auto").unwrap(); + assert_eq!(tid1, tid2, "ensure_topic_for_chat 应幂等返回当前 topic"); + + let tid3 = guard.ensure_topic_for_chat("chat-brand-new").unwrap(); + let session_id = guard.persistent_session_id("chat-brand-new"); + let topics = guard.session_store().list_topics(&session_id).unwrap(); + assert!( + topics.iter().any(|topic| topic.id == tid3), + "对无历史的 chat 应创建并持久化默认 topic" + ); + assert!(guard.current_topic("chat-brand-new").is_some()); + } + + /// 回归：无 topic 时 append_persisted_message 应回退到 chat_id 键的内存历史， + /// 避免消息只落库、不进内存，导致 agent 拿到空历史。 + #[tokio::test] + async fn test_append_persisted_message_without_topic_updates_chat_keyed_history() { + let base_url = start_mock_openai_server().await; + let provider_config = LLMProviderConfig { + provider_type: "openai".to_string(), + name: "default-provider".to_string(), + base_url, + api_key: "test-key".to_string(), + extra_headers: HashMap::new(), + model_id: "default-model".to_string(), + temperature: Some(0.0), + max_tokens: Some(32), + context_window_tokens: None, + model_extra: HashMap::new(), + max_tool_iterations: 1, + llm_timeout_secs: 30, + memory_maintenance_timeout_secs: 600, + max_retries: 3, + tool_result_max_chars: 100_000, + context_tool_result_trim_chars: 100_000, + max_images_in_context: 1, + max_image_age_rounds: 10, + }; + + let session_manager = SessionManager::new( + 100, + false, + "Asia/Shanghai".to_string(), + provider_config.clone(), + HashMap::from([("default".to_string(), provider_config)]), + Arc::new(SkillRuntime::default()), + HashSet::new(), + crate::config::TaskConfig::default(), + crate::config::SubagentsConfig::default(), + test_maintenance_config(), + Some(24), + crate::mcp::McpConfig::default(), + ) + .unwrap(); + + session_manager + .ensure_session("test-channel") + .await + .unwrap(); + let session = session_manager.get("test-channel").await.unwrap(); + let mut guard = session.lock().await; + guard.ensure_persistent_session("chat-no-topic").unwrap(); + + guard + .append_persisted_message("chat-no-topic", None, ChatMessage::user("hello-no-topic")) + .unwrap(); + + let history = guard + .get_history("chat-no-topic") + .expect("无 topic 消息应进入以 chat_id 为键的内存历史"); + assert_eq!(history.len(), 1); + assert_eq!(history[0].content, "hello-no-topic"); + } + /// 测试专用的 MemoryMaintenanceConfig，降低 min_memories_to_keep 以便于单条记忆测试 fn test_maintenance_config() -> crate::config::MemoryMaintenanceConfig { crate::config::MemoryMaintenanceConfig { diff --git a/src/gateway/session_lifecycle.rs b/src/gateway/session_lifecycle.rs index f2314a7..92bfa55 100644 --- a/src/gateway/session_lifecycle.rs +++ b/src/gateway/session_lifecycle.rs @@ -28,6 +28,14 @@ impl SessionLifecycleService { self.session_pool.get(channel_name).await } + /// 获取定时任务专用 Session（不自动创建） + pub(crate) async fn get_scheduler_session( + &self, + channel_name: &str, + ) -> Option>> { + self.session_pool.get_scheduler_session(channel_name).await + } + pub(crate) async fn touch(&self, channel_name: &str) { self.session_pool.touch(channel_name).await; }