ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

Compare commits

base: ooodc:23b7497b12b1a1c5e52b8c6cfc297b3d895dd077
Branches Tags
ooodc:main
..
compare: ooodc:e712fd764589bb8b9ad774dc66a916c935e7f9e5
Branches Tags
ooodc:main

No commits in common. "23b7497b12b1a1c5e52b8c6cfc297b3d895dd077" and "e712fd764589bb8b9ad774dc66a916c935e7f9e5" have entirely different histories.
23b7497b12 ... e712fd7645

10 changed files with 66 additions and 176 deletions
Show Stats Expand all files Collapse all files

65
src/command/handlers/save_session.rs
View File

@ -4,6 +4,8 @@ use crate::command::context::CommandContext;
use crate::command::handler::{CommandHandler, InChatCommandHandler};
use crate::command::response::{CommandError, CommandResponse, MessageKind};
use crate::command::Command;
use crate::config::LLMProviderConfig;
use crate::gateway::agent_prompt_provider::SimpleAgentPromptProvider;
use crate::storage::{SessionRecord, SessionStore};
use crate::agent::AgentError;
use async_trait::async_trait;
@ -27,7 +29,7 @@ pub async fn save_session_to_file(
filepath: Option<String>,
include_all: bool,
store: &SessionStore,
system_prompt_provider: &dyn SystemPromptProvider,
provider_config: &LLMProviderConfig,
) -> Result<PathBuf, String> {
// 获取会话记录
let record = store
@ -49,8 +51,8 @@ pub async fn save_session_to_file(
// 计算用户消息数（用于系统提示词构建）
let user_message_count = messages.iter().filter(|m| m.role == "user").count();
// 构建系统提示词（使用外部传入的提供者）
let system_prompt = build_system_prompt(system_prompt_provider, &record, user_message_count);
// 构建系统提示词
let system_prompt = build_system_prompt(provider_config, &record, user_message_count);
// 生成 Markdown 内容
let markdown = generate_markdown(&record, &system_prompt, &messages);
@ -78,7 +80,7 @@ pub async fn save_session_to_file(
/// 将当前会话内容（系统提示词和消息历史）保存到 Markdown 文件
pub struct SaveSessionCommandHandler {
store: Arc<SessionStore>,
system_prompt_provider: Arc<dyn SystemPromptProvider>,
provider_config: LLMProviderConfig,
}
impl SaveSessionCommandHandler {
@ -86,11 +88,11 @@ impl SaveSessionCommandHandler {
///
/// # Arguments
/// * `store` - 会话存储
/// * `system_prompt_provider` - 系统提示词提供者（负责构建完整的系统提示词）
pub fn new(store: Arc<SessionStore>, system_prompt_provider: Arc<dyn SystemPromptProvider>) -> Self {
/// * `provider_config` - LLM 提供者配置（用于构建系统提示词）
pub fn new(store: Arc<SessionStore>, provider_config: LLMProviderConfig) -> Self {
Self {
store,
system_prompt_provider,
provider_config,
}
}
@ -139,7 +141,7 @@ async fn handle_save_session(
filepath,
include_all,
&*handler.store,
&*handler.system_prompt_provider,
&handler.provider_config,
)
.await
.map_err(|e| CommandError::new("SAVE_ERROR", e))?;
@ -168,10 +170,11 @@ async fn handle_save_session(
/// 构建系统提示词
fn build_system_prompt(
provider: &dyn SystemPromptProvider,
provider_config: &LLMProviderConfig,
record: &SessionRecord,
user_message_count: usize,
) -> Option<SystemPrompt> {
let provider = SimpleAgentPromptProvider::new(provider_config.clone());
let context = SystemPromptContext {
session_id: Some(record.id.clone()),
chat_id: record.chat_id.clone(),
@ -370,15 +373,15 @@ pub fn resolve_filepath(filepath: Option<String>, record: &SessionRecord) -> Pat
/// 用于处理 Feishu/WeChat 等通道中直接输入的 /save 命令
pub struct SaveSessionInChatHandler {
store: Arc<SessionStore>,
system_prompt_provider: Arc<dyn SystemPromptProvider>,
provider_config: LLMProviderConfig,
}
impl SaveSessionInChatHandler {
/// 创建新的 InChat 保存会话命令处理器
pub fn new(store: Arc<SessionStore>, system_prompt_provider: Arc<dyn SystemPromptProvider>) -> Self {
pub fn new(store: Arc<SessionStore>, provider_config: LLMProviderConfig) -> Self {
Self {
store,
system_prompt_provider,
provider_config,
}
}
}
@ -417,7 +420,7 @@ impl InChatCommandHandler for SaveSessionInChatHandler {
filepath,
include_all,
&*self.store,
&*self.system_prompt_provider,
&self.provider_config,
)
.await;
@ -441,6 +444,27 @@ impl InChatCommandHandler for SaveSessionInChatHandler {
mod tests {
use super::*;
use crate::storage::{SessionRecord, SessionStore};
use std::collections::HashMap;
fn test_config() -> LLMProviderConfig {
LLMProviderConfig {
provider_type: "openai".to_string(),
name: "test".to_string(),
base_url: "http://localhost".to_string(),
api_key: "test-key".to_string(),
extra_headers: HashMap::new(),
llm_timeout_secs: 120,
memory_maintenance_timeout_secs: 600,
model_id: "test-model".to_string(),
temperature: Some(0.0),
max_tokens: Some(32),
context_window_tokens: None,
tool_result_max_chars: 20_000,
context_tool_result_trim_chars: 20_000,
model_extra: HashMap::new(),
max_tool_iterations: 1,
}
}
fn create_test_record(id: &str, title: &str) -> SessionRecord {
SessionRecord {
@ -523,23 +547,10 @@ mod tests {
#[test]
fn test_can_handle() {
let store = Arc::new(SessionStore::in_memory().unwrap());
let provider = Arc::new(TestSystemPromptProvider);
let handler = SaveSessionCommandHandler::new(store, provider);
let handler = SaveSessionCommandHandler::new(store, test_config());
assert!(handler.can_handle(&Command::SaveSession { filepath: None, include_all: false }));
assert!(handler.can_handle(&Command::SaveSession { filepath: None, include_all: true }));
assert!(!handler.can_handle(&Command::CreateSession { title: None }));
}
/// 测试用的系统提示词提供者
struct TestSystemPromptProvider;
impl SystemPromptProvider for TestSystemPromptProvider {
fn build(&self, _context: &SystemPromptContext) -> Option<SystemPrompt> {
Some(SystemPrompt {
content: "Test system prompt".to_string(),
context: Some("test".to_string()),
})
}
}
}

54
src/config/mod.rs
View File

@ -314,8 +314,6 @@ pub struct GatewayConfig {
pub agent_prompt_reinject_every: u64,
#[serde(default = "default_max_concurrent_requests")]
pub max_concurrent_requests: usize,
#[serde(default, rename = "session_ttl_hours")]
pub session_ttl_hours: Option<u64>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@ -339,7 +337,6 @@ pub struct SchedulerConfig {
}
pub const BUILTIN_MEMORY_MAINTENANCE_JOB_ID: &str = "builtin.memory_maintenance_daily";
pub const BUILTIN_SESSION_CLEANUP_JOB_ID: &str = "builtin.session_cleanup_hourly";
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
@ -430,40 +427,22 @@ impl SchedulerJobConfig {
impl SchedulerConfig {
pub fn builtin_jobs(time: &TimeConfig) -> Vec<SchedulerJobConfig> {
vec![
SchedulerJobConfig {
id: BUILTIN_MEMORY_MAINTENANCE_JOB_ID.to_string(),
enabled: true,
kind: SchedulerJobKind::InternalEvent,
schedule: Some(SchedulerSchedule::Cron {
expression: "0 */4 * * *".to_string(),
}),
startup_delay_secs: 0,
interval_secs: 0,
target: SchedulerJobTarget::default(),
payload: serde_json::json!({
"event": "memory_maintenance",
"time_zone": time.timezone,
"local_time": "every_4_hours"
}),
},
SchedulerJobConfig {
id: BUILTIN_SESSION_CLEANUP_JOB_ID.to_string(),
enabled: true,
kind: SchedulerJobKind::InternalEvent,
schedule: Some(SchedulerSchedule::Cron {
expression: "0 * * * *".to_string(),
}),
startup_delay_secs: 0,
interval_secs: 0,
target: SchedulerJobTarget::default(),
payload: serde_json::json!({
"event": "session_cleanup",
"time_zone": time.timezone,
"local_time": "every_hour"
}),
},
]
vec![SchedulerJobConfig {
id: BUILTIN_MEMORY_MAINTENANCE_JOB_ID.to_string(),
enabled: true,
kind: SchedulerJobKind::InternalEvent,
schedule: Some(SchedulerSchedule::Cron {
expression: "0 */4 * * *".to_string(),
}),
startup_delay_secs: 0,
interval_secs: 0,
target: SchedulerJobTarget::default(),
payload: serde_json::json!({
"event": "memory_maintenance",
"time_zone": time.timezone,
"local_time": "every_4_hours"
}),
}]
}
pub fn effective_jobs(&self, time: &TimeConfig) -> Vec<SchedulerJobConfig> {
@ -625,7 +604,6 @@ impl Default for GatewayConfig {
chat_history_ttl_hours: Some(4),
agent_prompt_reinject_every: default_agent_prompt_reinject_every(),
max_concurrent_requests: default_max_concurrent_requests(),
session_ttl_hours: Some(24),
}
}
}

4
src/gateway/agent_prompt_provider.rs
View File

@ -11,7 +11,7 @@ use crate::storage::PromptInjectionRepository;
/// 以及动态生成的系统环境信息。
pub struct AgentPromptProvider {
/// 重新注入间隔（用户消息数）
_reinject_every: usize,
reinject_every: usize,
/// LLM 提供者配置（用于生成系统环境信息）
provider_config: LLMProviderConfig,
/// 会话持久化仓库（用于记录注入状态）
@ -26,7 +26,7 @@ impl AgentPromptProvider {
repository: Arc<dyn PromptInjectionRepository>,
) -> Self {
Self {
_reinject_every: reinject_every,
reinject_every,
provider_config,
repository,
}

3
src/gateway/mod.rs
View File

@ -66,8 +66,6 @@ impl GatewayState {
let agent_prompt_reinject_every = config.gateway.agent_prompt_reinject_every;
let show_tool_results = config.gateway.show_tool_results;
let session_ttl_hours = config.gateway.session_ttl_hours;
let skills = Arc::new(SkillRuntime::from_config(config.skills.clone()));
let channel_manager = ChannelManager::new();
let bus = channel_manager.bus();
@ -82,7 +80,6 @@ impl GatewayState {
Arc::new(BusSessionMessageSender::new(bus.clone())),
std::collections::HashSet::new(),
chat_history_ttl_hours,
session_ttl_hours,
)?;
Ok(Self {

16
src/gateway/processor.rs
View File

@ -2,13 +2,11 @@ use std::sync::Arc;
use tokio::sync::Semaphore;
use crate::agent::{AgentError, CompositeSystemPromptProvider};
use crate::agent::AgentError;
use crate::bus::{InboundMessage, MessageBus, OutboundMessage};
use crate::command::handler::InChatCommandRouter;
use crate::command::Command;
use crate::config::LLMProviderConfig;
use crate::gateway::agent_prompt_provider::AgentPromptProvider;
use crate::skills::SkillPromptProvider;
use super::session::{BusToolCallEmitter, SessionManager};
@ -33,19 +31,9 @@ impl InboundProcessor {
// 注册 save_session 处理器
let store = session_manager.store();
let skills = session_manager.skills();
let prompt_repository = session_manager.store().clone();
let system_prompt_provider: Arc<dyn crate::agent::SystemPromptProvider> = Arc::new(CompositeSystemPromptProvider::new(vec![
Box::new(AgentPromptProvider::new(
0, // save_session 不需要 reinject 逻辑
provider_config.clone(),
prompt_repository,
)),
Box::new(SkillPromptProvider::new(skills)),
]));
command_router.register(Box::new(crate::command::handlers::save_session::SaveSessionInChatHandler::new(
store,
system_prompt_provider,
provider_config.clone(),
)));
Self {

5
src/gateway/runtime.rs
View File

@ -30,7 +30,6 @@ pub(crate) fn build_session_manager(
skills: Arc<SkillRuntime>,
disabled_tools: HashSet<String>,
chat_history_ttl_hours: Option<u64>,
session_ttl_hours: Option<u64>,
) -> Result<SessionManager, AgentError> {
build_session_manager_with_sender(
agent_prompt_reinject_every,
@ -42,7 +41,6 @@ pub(crate) fn build_session_manager(
Arc::new(NoopSessionMessageSender),
disabled_tools,
chat_history_ttl_hours,
session_ttl_hours,
)
}
@ -56,7 +54,6 @@ pub(crate) fn build_session_manager_with_sender(
session_message_sender: Arc<dyn SessionMessageSender>,
disabled_tools: HashSet<String>,
chat_history_ttl_hours: Option<u64>,
session_ttl_hours: Option<u64>,
) -> Result<SessionManager, AgentError> {
let store = Arc::new(
SessionStore::new()
@ -104,7 +101,7 @@ pub(crate) fn build_session_manager_with_sender(
skill_events,
chat_history_ttl_hours,
);
let lifecycle = SessionLifecycleService::new(session_factory, session_ttl_hours);
let lifecycle = SessionLifecycleService::new(session_factory);
let cli_sessions = CliSessionService::new(store.clone());
let messages = SessionMessageService::new(lifecycle.clone(), show_tool_results);
let scheduled_tasks = ScheduledAgentTaskService::new(

33
src/gateway/session.rs
View File

@ -387,7 +387,6 @@ impl SessionManager {
skills: Arc<SkillRuntime>,
disabled_tools: std::collections::HashSet<String>,
chat_history_ttl_hours: Option<u64>,
session_ttl_hours: Option<u64>,
) -> Result<Self, AgentError> {
super::runtime::build_session_manager(
agent_prompt_reinject_every,
@ -398,7 +397,6 @@ impl SessionManager {
skills,
disabled_tools,
chat_history_ttl_hours,
session_ttl_hours,
)
}
@ -820,7 +818,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -871,7 +868,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -938,7 +934,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -1022,7 +1017,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -1107,7 +1101,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -1191,7 +1184,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -1257,7 +1249,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -1332,7 +1323,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -1394,7 +1384,6 @@ mod tests {
Arc::new(SkillRuntime::default()),
HashSet::new(),
Some(4),
Some(24),
)
.unwrap();
@ -1800,25 +1789,3 @@ mod tests {
assert!(contents.contains(&"习惯先问方案再要代码".to_string()));
}
}
#[async_trait]
impl crate::scheduler::MaintenanceExecutor for SessionManager {
async fn cleanup_expired_sessions(&self) -> usize {
self.cleanup_expired_sessions().await
}
async fn run_memory_maintenance_for_all_scopes(
&self,
) -> anyhow::Result<Vec<crate::scheduler::MaintenanceRunSummary>> {
match self.run_memory_maintenance_for_all_scopes().await {
Ok(Some(result)) => Ok(vec![crate::scheduler::MaintenanceRunSummary {
scope_key: result.scope_key,
merges: result.output.merges.len(),
conflicts: result.output.conflicts.len(),
low_value: result.output.low_value_ids.len(),
}]),
Ok(None) => Ok(vec![]),
Err(error) => Err(anyhow::anyhow!(error.to_string())),
}
}
}

4
src/gateway/session_lifecycle.rs
View File

@ -14,9 +14,9 @@ pub(crate) struct SessionLifecycleService {
}
impl SessionLifecycleService {
pub(crate) fn new(session_factory: SessionFactory, session_ttl_hours: Option<u64>) -> Self {
pub(crate) fn new(session_factory: SessionFactory) -> Self {
Self {
session_pool: SessionPool::new(session_factory, session_ttl_hours),
session_pool: SessionPool::new(session_factory),
}
}

40
src/gateway/session_pool.rs
View File

@ -14,7 +14,6 @@ use super::session_factory::SessionFactory;
pub(crate) struct SessionPool {
inner: Arc<Mutex<SessionPoolInner>>,
session_factory: SessionFactory,
session_ttl_hours: Option<u64>,
}
struct SessionPoolInner {
@ -23,14 +22,13 @@ struct SessionPoolInner {
}
impl SessionPool {
pub(crate) fn new(session_factory: SessionFactory, session_ttl_hours: Option<u64>) -> Self {
pub(crate) fn new(session_factory: SessionFactory) -> Self {
Self {
inner: Arc::new(Mutex::new(SessionPoolInner {
sessions: HashMap::new(),
session_timestamps: HashMap::new(),
})),
session_factory,
session_ttl_hours,
}
}
@ -72,39 +70,7 @@ impl SessionPool {
}
pub(crate) async fn cleanup_expired_sessions(&self) -> usize {
let ttl_hours = match self.session_ttl_hours {
Some(hours) if hours > 0 => hours,
_ => return 0,
};
let ttl_duration = std::time::Duration::from_secs(ttl_hours * 3600);
let mut inner = self.inner.lock().await;
let now = Instant::now();
let expired_channels: Vec<String> = inner
.session_timestamps
.iter()
.filter_map(|(channel_name, last_active)| {
let elapsed = now.duration_since(*last_active);
if elapsed >= ttl_duration {
tracing::info!(
channel = %channel_name,
elapsed_hours = elapsed.as_secs() / 3600,
ttl_hours = ttl_hours,
"Session expired, removing from memory pool"
);
Some(channel_name.clone())
} else {
None
}
})
.collect();
for channel_name in &expired_channels {
inner.sessions.remove(channel_name);
inner.session_timestamps.remove(channel_name);
}
expired_channels.len()
// Session 级别不再自动清理，返回 0
0
}
}

18
src/gateway/ws.rs
View File

@ -1,5 +1,5 @@
use super::GatewayState;
use crate::agent::{AgentError, CompositeSystemPromptProvider};
use crate::agent::AgentError;
use crate::bus::InboundMessage;
use crate::command::adapter::OutputAdapter;
use crate::command::adapters::websocket::{WebSocketInputAdapter, WebSocketOutputAdapter};
@ -7,9 +7,7 @@ use crate::command::context::CommandContext;
use crate::command::handler::CommandRouter;
use crate::command::handlers::save_session::SaveSessionCommandHandler;
use crate::command::handlers::session::SessionCommandHandler;
use crate::gateway::agent_prompt_provider::AgentPromptProvider;
use crate::protocol::{SessionSummary, WsInbound, WsOutbound, parse_inbound, serialize_outbound};
use crate::skills::SkillPromptProvider;
use axum::extract::State;
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::response::Response;
@ -356,23 +354,11 @@ async fn handle_inbound(
// 获取所需依赖
let store = state.session_manager.store();
let skills = state.session_manager.skills();
let provider_config = state.config.get_provider_config("default")
.map_err(|e| AgentError::Other(e.to_string()))?;
let prompt_repository = state.session_manager.store().clone();
// 构建组合系统提示词提供者（与运行时一致）
let system_prompt_provider: Arc<dyn crate::agent::SystemPromptProvider> = Arc::new(CompositeSystemPromptProvider::new(vec![
Box::new(AgentPromptProvider::new(
0, // save_session 不需要 reinject 逻辑
provider_config.clone(),
prompt_repository,
)),
Box::new(SkillPromptProvider::new(skills)),
]));
// 构建处理器
let handler = SaveSessionCommandHandler::new(store, system_prompt_provider);
let handler = SaveSessionCommandHandler::new(store, provider_config);
let router = {
let mut r = CommandRouter::new();
r.register(Box::new(handler));