ooodc/PicoBot
ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

test(gateway): 定时任务 topic 补齐与消息送达回归测试

Browse Source 
- 新增捕获请求的 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 供测试断言
This commit is contained in:
oudecheng 2026-08-15 18:17:29 +08:00
parent cbe2ff1339
commit 4b93c84447
2 changed files with 353 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

342
src/gateway/session.rs
View File

@ -535,12 +535,15 @@ 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 {
Some(_) => {
tracing::info!(
chat_id = %chat_id,
write_topic_id = ?topic_id,
@ -548,6 +551,10 @@ impl Session {
"Skipping memory history update: message belongs to a different topic"
);
}
None => {
self.add_message(chat_id, message);
}
}
// 更新 topic 的最后活跃时间
if let Some(ref topic_id) = topic_id {
@ -1105,6 +1112,11 @@ impl SessionManager {
self.lifecycle.get(channel_name).await
}
/// 获取定时任务专用 Session（不自动创建）
pub async fn get_scheduler_session(&self, channel_name: &str) -> Option<Arc<Mutex<Session>>> {
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<std::sync::Mutex<Vec<Value>>>)
{
let captured: StdArc<std::sync::Mutex<Vec<Value>>> =
StdArc::new(std::sync::Mutex::new(Vec::new()));
let state = captured.clone();
async fn handle(
axum::extract::State(state): axum::extract::State<StdArc<std::sync::Mutex<Vec<Value>>>>,
Json(body): Json<Value>,
) -> Json<Value> {
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 {

8
src/gateway/session_lifecycle.rs
View File

@ -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<Arc<Mutex<Session>>> {
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;
}