0f60b02ede
1. 删除缓存诊断日志 4 处（openai.rs）：排查已完结，字段已补齐，每次 LLM 调用都刷屏。 2. 删除 AgentPromptProvider 模型配置日志（agent_prompt_provider.rs）：排查遗留，每次请求刷。 3. AgentFactory 创建日志 info→debug（agent_factory.rs）：每轮对话刷，降级保留供偶发排查。 4. Calling tool / Tool calls detected info→debug（agent_loop.rs）：每次工具调用都刷，全量参数打印开销大，tracing 在未启用级别时不评估字段。 5. LoadTaskMessages 3 条日志 info→debug（load_task_messages.rs）：每次前端 trigger 都刷，一次 3 条。 6. 旧结果诊断日志 info→debug（execution.rs）：每条消息首次 agent 迭代都触发。 7. pending subagents 日志 info→debug（processor.rs）：子代理运行期高频率重复触发。 8. Updating current_session_id 仅变化时打（ws.rs）：old==new 时每条消息都刷，无信息量。 编译通过，测试全绿。
1259 lines
52 KiB
Rust
1259 lines
52 KiB
Rust
use super::GatewayState;
|
||
use crate::agent::AgentError;
|
||
use crate::bus::{InboundMessage, MediaItem};
|
||
use crate::command::adapter::{InputAdapter, OutputAdapter};
|
||
use crate::command::adapters::websocket::{WebSocketInputAdapter, WebSocketOutputAdapter};
|
||
use crate::command::context::CommandContext;
|
||
use crate::command::handler::CommandRouter;
|
||
use crate::command::handlers::delete_topic::DeleteTopicCommandHandler;
|
||
use crate::command::handlers::get_current::GetCurrentSessionCommandHandler;
|
||
use crate::command::handlers::help::HelpCommandHandler;
|
||
use crate::command::handlers::list_channels::ListChannelsCommandHandler;
|
||
use crate::command::handlers::list_memories::ListMemoriesCommandHandler;
|
||
use crate::command::handlers::list_scheduler_jobs::ListSchedulerJobsCommandHandler;
|
||
use crate::command::handlers::list_sessions::ListSessionsCommandHandler;
|
||
use crate::command::handlers::list_sessions_by_channel::ListSessionsByChannelCommandHandler;
|
||
use crate::command::handlers::list_skills::ListSkillsCommandHandler;
|
||
use crate::command::handlers::list_todos::ListTodosCommandHandler;
|
||
use crate::command::handlers::list_topics::ListTopicsCommandHandler;
|
||
use crate::command::handlers::load_chat_messages::LoadChatMessagesCommandHandler;
|
||
use crate::command::handlers::load_task_messages::LoadTaskMessagesCommandHandler;
|
||
use crate::command::handlers::load_topic::LoadTopicCommandHandler;
|
||
use crate::command::handlers::memory_crud::MemoryCrudCommandHandler;
|
||
use crate::command::handlers::rename_topic::RenameTopicCommandHandler;
|
||
use crate::command::handlers::save_session::SaveSessionCommandHandler;
|
||
use crate::command::handlers::save_topic::SaveTopicCommandHandler;
|
||
use crate::command::handlers::session::SessionCommandHandler;
|
||
use crate::command::handlers::stop_execution::StopExecutionCommandHandler;
|
||
use crate::command::handlers::switch_topic::SwitchTopicCommandHandler;
|
||
use crate::gateway::agent_factory::build_system_prompt_provider;
|
||
use crate::protocol::{MediaSummary, WsInbound, WsOutbound, parse_inbound, serialize_outbound};
|
||
use crate::storage::persistent_session_id;
|
||
use crate::tools::task::repository::TaskRepository;
|
||
use crate::utils::current_timestamp;
|
||
use axum::extract::Query;
|
||
use axum::extract::State;
|
||
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
|
||
use axum::http::StatusCode;
|
||
use axum::response::{IntoResponse, Response};
|
||
use base64::{Engine as _, engine::general_purpose::STANDARD};
|
||
use futures_util::{SinkExt, StreamExt};
|
||
use std::collections::HashMap;
|
||
use std::path::PathBuf;
|
||
use std::sync::Arc;
|
||
use tokio::sync::mpsc;
|
||
use tokio_util::sync::CancellationToken;
|
||
|
||
const WS_CHANNEL_NAME: &str = "websocket";
|
||
|
||
/// Default media directory for WebSocket uploads
|
||
fn default_ws_media_dir() -> PathBuf {
|
||
let home = crate::platform::picobot_home_dir();
|
||
home.join(".picobot").join("media").join("ws")
|
||
}
|
||
|
||
/// Build a unique filename for media upload
|
||
fn build_media_filename(media_type: &str, file_name: Option<&str>) -> String {
|
||
if let Some(file_name) = file_name {
|
||
let sanitized: String = file_name
|
||
.chars()
|
||
.map(|ch| match ch {
|
||
'/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_',
|
||
_ => ch,
|
||
})
|
||
.collect();
|
||
if !sanitized.trim().is_empty() {
|
||
return format!("{}_{}", uuid::Uuid::new_v4(), sanitized);
|
||
}
|
||
}
|
||
format!("{}_{}", media_type, uuid::Uuid::new_v4())
|
||
}
|
||
|
||
/// Process attachments with base64 content: save to local file and return MediaItem with correct path
|
||
/// Keeps content_base64 for frontend display/download
|
||
fn process_attachments_with_base64(
|
||
attachments: Vec<MediaSummary>,
|
||
) -> Result<Vec<MediaItem>, AgentError> {
|
||
if attachments.is_empty() {
|
||
return Ok(Vec::new());
|
||
}
|
||
|
||
let media_dir = default_ws_media_dir();
|
||
std::fs::create_dir_all(&media_dir)
|
||
.map_err(|error| AgentError::Other(format!("Failed to create media dir: {}", error)))?;
|
||
|
||
attachments
|
||
.into_iter()
|
||
.map(|att| {
|
||
// If content_base64 exists, save to file and update path
|
||
if let Some(base64_content) = &att.content_base64 {
|
||
let decoded = STANDARD.decode(base64_content).map_err(|error| {
|
||
AgentError::Other(format!("Failed to decode base64: {}", error))
|
||
})?;
|
||
|
||
let filename = build_media_filename(&att.media_type, att.file_name.as_deref());
|
||
let file_path = media_dir.join(&filename);
|
||
|
||
std::fs::write(&file_path, decoded).map_err(|error| {
|
||
AgentError::Other(format!("Failed to write media file: {}", error))
|
||
})?;
|
||
|
||
tracing::info!(
|
||
filename = %filename,
|
||
media_type = %att.media_type,
|
||
file_path = %file_path.to_string_lossy(),
|
||
"Saved WebSocket media to local file"
|
||
);
|
||
|
||
Ok(MediaItem {
|
||
path: file_path.to_string_lossy().to_string(),
|
||
media_type: att.media_type,
|
||
mime_type: att.mime_type,
|
||
original_key: None,
|
||
// Keep content_base64 for frontend display/download
|
||
content_base64: att.content_base64,
|
||
file_name: att.file_name,
|
||
})
|
||
} else {
|
||
// No base64 content, keep original path (should already be valid)
|
||
Ok(MediaItem {
|
||
path: att.path,
|
||
media_type: att.media_type,
|
||
mime_type: att.mime_type,
|
||
original_key: None,
|
||
content_base64: None,
|
||
file_name: att.file_name,
|
||
})
|
||
}
|
||
})
|
||
.collect()
|
||
}
|
||
|
||
#[derive(serde::Deserialize)]
|
||
pub struct WsAuthQuery {
|
||
/// 可选的认证 token（浏览器原生 WebSocket 不支持自定义 header，通过 query param 传递）
|
||
pub token: Option<String>,
|
||
}
|
||
|
||
pub async fn ws_handler(
|
||
ws: WebSocketUpgrade,
|
||
State(state): State<Arc<GatewayState>>,
|
||
Query(query): Query<WsAuthQuery>,
|
||
auth_cfg: Option<axum::Extension<crate::gateway::auth::AuthConfig>>,
|
||
) -> Response {
|
||
// 若启用了认证（auth_cfg 存在且 token 已配置），校验 query param 中的 token
|
||
if let Some(axum::Extension(cfg)) = auth_cfg {
|
||
if let Some(ref expected) = cfg.token {
|
||
let provided = query.token.as_deref();
|
||
if !crate::gateway::auth::token_matches(provided, &Some(expected.clone())) {
|
||
tracing::warn!("WebSocket connection rejected: missing or invalid token");
|
||
return (StatusCode::UNAUTHORIZED, "missing or invalid token").into_response();
|
||
}
|
||
}
|
||
}
|
||
|
||
ws.on_upgrade(|socket| async {
|
||
handle_socket(socket, state).await;
|
||
})
|
||
}
|
||
|
||
async fn handle_socket(ws: WebSocket, state: Arc<GatewayState>) {
|
||
let (sender, receiver) = mpsc::channel::<WsOutbound>(100);
|
||
|
||
let cli_sessions = state.session_manager.cli_sessions();
|
||
let store = state.session_manager.store();
|
||
|
||
// 1. 查询 websocket 和 cli 两个通道的 Sessions（兼容旧版本 cli 通道创建的会话）
|
||
let mut websocket_sessions = store.list_sessions("websocket", false).unwrap_or_default();
|
||
let cli_channel_sessions = store.list_sessions("cli", false).unwrap_or_default();
|
||
websocket_sessions.extend(cli_channel_sessions);
|
||
websocket_sessions.sort_by_key(|s| -(s.last_active_at));
|
||
|
||
// 2. 如果没有，自动创建一个默认 Session
|
||
let initial_record = if websocket_sessions.is_empty() {
|
||
match cli_sessions.create_with_channel("websocket", Some("默认会话")) {
|
||
Ok(record) => record,
|
||
Err(e) => {
|
||
tracing::error!(error = %e, "Failed to create initial WebSocket session");
|
||
return;
|
||
}
|
||
}
|
||
} else {
|
||
// 使用最新的 Session
|
||
websocket_sessions[0].clone()
|
||
};
|
||
|
||
let runtime_session_id = uuid::Uuid::new_v4().to_string();
|
||
// 清理数据库中可能已被污染的多重前缀 session_id（幂等处理）
|
||
let mut current_session_id = persistent_session_id(WS_CHANNEL_NAME, &initial_record.id);
|
||
let mut current_topic_id: Option<String> = None;
|
||
let shutdown_token = state
|
||
.channel_manager
|
||
.websocket_channel()
|
||
.register_connection(
|
||
current_session_id.clone(),
|
||
runtime_session_id.clone(),
|
||
sender.clone(),
|
||
)
|
||
.await;
|
||
tracing::info!(runtime_session_id = %runtime_session_id, session_id = %current_session_id, "WebSocket session established");
|
||
|
||
let _ = sender
|
||
.send(WsOutbound::SessionEstablished {
|
||
session_id: current_session_id.clone(),
|
||
})
|
||
.await;
|
||
|
||
// 连接建立后立即发送通道列表（合并 websocket + ChannelManager 动态通道）
|
||
let channels = state.channel_manager.build_channel_list().await;
|
||
let _ = sender.send(WsOutbound::ChannelList { channels }).await;
|
||
|
||
// 3. 发送合并后的 Session 列表（已在上面合并了 websocket + cli 通道）
|
||
// 如果刚创建了新会话，确保它也在列表中
|
||
let has_initial = websocket_sessions.iter().any(|s| s.id == initial_record.id);
|
||
if !has_initial {
|
||
websocket_sessions.push(initial_record);
|
||
websocket_sessions.sort_by_key(|s| -(s.last_active_at));
|
||
}
|
||
|
||
tracing::info!("Sending {} sessions to client", websocket_sessions.len());
|
||
for s in &websocket_sessions {
|
||
tracing::info!(" - {}: {} (channel: {})", s.id, s.title, s.channel_name);
|
||
}
|
||
|
||
let session_summaries: Vec<crate::protocol::SessionSummary> = websocket_sessions
|
||
.into_iter()
|
||
.map(|s| crate::protocol::SessionSummary {
|
||
session_id: s.id,
|
||
title: s.title,
|
||
channel_name: s.channel_name,
|
||
chat_id: s.chat_id,
|
||
message_count: s.message_count,
|
||
last_active_at: s.last_active_at,
|
||
archived_at: s.archived_at,
|
||
})
|
||
.collect();
|
||
|
||
let _ = sender
|
||
.send(WsOutbound::SessionList {
|
||
sessions: session_summaries,
|
||
current_session_id: Some(current_session_id.clone()),
|
||
channel_name: Some("websocket".to_string()),
|
||
})
|
||
.await;
|
||
|
||
let (mut ws_sender, mut ws_receiver) = ws.split();
|
||
|
||
let mut receiver = receiver;
|
||
let session_id_for_sender = runtime_session_id.clone();
|
||
// 每连接独立的关闭信号：writer 超时/错误时 cancel，通知主 loop 退出
|
||
// 不能用共享的 shutdown_token——那是 CliChannel 级别的，cancel 会关闭所有连接
|
||
let writer_closed = CancellationToken::new();
|
||
let writer_closed_clone = writer_closed.clone();
|
||
tokio::spawn(async move {
|
||
while let Some(msg) = receiver.recv().await {
|
||
if let Ok(text) = serialize_outbound(&msg) {
|
||
let send_result = tokio::time::timeout(
|
||
std::time::Duration::from_secs(30),
|
||
ws_sender.send(WsMessage::Text(text.into())),
|
||
)
|
||
.await;
|
||
match send_result {
|
||
Ok(Ok(())) => {}
|
||
Ok(Err(_)) => {
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(session_id = %session_id_for_sender, "WebSocket send error");
|
||
writer_closed_clone.cancel();
|
||
break;
|
||
}
|
||
Err(_) => {
|
||
tracing::warn!(
|
||
session_id = %session_id_for_sender,
|
||
"WebSocket send timed out after 30s, closing connection"
|
||
);
|
||
writer_closed_clone.cancel();
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
});
|
||
|
||
loop {
|
||
tokio::select! {
|
||
// 监听全局 shutdown 信号（来自 CliChannel::stop()）
|
||
_ = shutdown_token.cancelled() => {
|
||
tracing::info!(session_id = %current_session_id, "WebSocket shutdown signal received, closing connection");
|
||
break;
|
||
}
|
||
// 监听 writer 退出信号（writer 超时或错误，仅关闭当前连接）
|
||
_ = writer_closed.cancelled() => {
|
||
tracing::info!(session_id = %current_session_id, "WebSocket writer closed, shutting down connection");
|
||
break;
|
||
}
|
||
// 监听 WebSocket 消息
|
||
msg = ws_receiver.next() => {
|
||
let Some(msg) = msg else {
|
||
break;
|
||
};
|
||
match msg {
|
||
Ok(WsMessage::Text(text)) => {
|
||
let text = text.to_string();
|
||
match parse_inbound(&text) {
|
||
Ok(inbound) => {
|
||
if let Err(e) = handle_inbound(
|
||
&state,
|
||
&sender,
|
||
&runtime_session_id,
|
||
&mut current_session_id,
|
||
&mut current_topic_id,
|
||
inbound,
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!(error = %e, session_id = %current_session_id, "Failed to handle inbound message");
|
||
let _ = sender
|
||
.send(WsOutbound::Error {
|
||
timestamp: Some(crate::protocol::now_timestamp()),
|
||
code:"SESSION_ERROR".to_string(),
|
||
message: e.to_string(),
|
||
subagent_task_id: None,
|
||
})
|
||
.await;
|
||
}
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, "Failed to parse inbound message");
|
||
let _ = sender
|
||
.send(WsOutbound::Error {
|
||
timestamp: Some(crate::protocol::now_timestamp()),
|
||
code:"PARSE_ERROR".to_string(),
|
||
message: e.to_string(),
|
||
subagent_task_id: None,
|
||
})
|
||
.await;
|
||
}
|
||
}
|
||
}
|
||
Ok(WsMessage::Close(_)) | Err(_) => {
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(session_id = %runtime_session_id, "WebSocket closed");
|
||
break;
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
state
|
||
.channel_manager
|
||
.websocket_channel()
|
||
.unregister_connection(&runtime_session_id)
|
||
.await;
|
||
tracing::info!(session_id = %runtime_session_id, current_session_id = %current_session_id, "CLI session ended");
|
||
}
|
||
|
||
async fn handle_inbound(
|
||
state: &Arc<GatewayState>,
|
||
sender: &mpsc::Sender<WsOutbound>,
|
||
runtime_session_id: &str,
|
||
current_session_id: &mut String,
|
||
current_topic_id: &mut Option<String>,
|
||
inbound: WsInbound,
|
||
) -> Result<(), AgentError> {
|
||
match inbound {
|
||
WsInbound::Message {
|
||
content,
|
||
attachments,
|
||
chat_id,
|
||
sender_id,
|
||
..
|
||
} => {
|
||
let chat_id = chat_id.unwrap_or_else(|| current_session_id.clone());
|
||
let sender_id = resolve_ws_sender_id(sender_id.as_deref(), runtime_session_id);
|
||
|
||
let _ = state
|
||
.channel_manager
|
||
.websocket_channel()
|
||
.register_connection(
|
||
chat_id.clone(),
|
||
runtime_session_id.to_string(),
|
||
sender.clone(),
|
||
)
|
||
.await;
|
||
|
||
// Process attachments: save base64 content to local files and build MediaItems with correct paths
|
||
let media = process_attachments_with_base64(attachments)?;
|
||
|
||
state
|
||
.bus
|
||
.publish_inbound(InboundMessage {
|
||
channel: WS_CHANNEL_NAME.to_string(),
|
||
sender_id,
|
||
chat_id,
|
||
content,
|
||
timestamp: current_timestamp(),
|
||
media,
|
||
metadata: HashMap::new(),
|
||
forwarded_metadata: HashMap::new(),
|
||
trace_id: crate::observability::tracing_ctx::new_trace_id(),
|
||
})
|
||
.await
|
||
.map_err(|error| AgentError::Other(error.to_string()))?;
|
||
|
||
Ok(())
|
||
}
|
||
WsInbound::Command { payload } => {
|
||
// 使用 Command 系统处理命令
|
||
let input_adapter = WebSocketInputAdapter::new();
|
||
let output_adapter = WebSocketOutputAdapter::new();
|
||
|
||
// 解析命令
|
||
let adapter_ctx = crate::command::context::AdapterContext::new("websocket")
|
||
.with_session_id(current_session_id.as_str());
|
||
|
||
let cmd = match input_adapter.try_parse(&payload, adapter_ctx) {
|
||
Ok(Some(cmd)) => cmd,
|
||
Ok(None) => {
|
||
// 不是命令，返回错误
|
||
let _ = sender
|
||
.send(WsOutbound::Error {
|
||
timestamp: Some(crate::protocol::now_timestamp()),
|
||
code: "INVALID_COMMAND".to_string(),
|
||
message: "Invalid command payload".to_string(),
|
||
subagent_task_id: None,
|
||
})
|
||
.await;
|
||
return Ok(());
|
||
}
|
||
Err(e) => {
|
||
let _ = sender
|
||
.send(WsOutbound::Error {
|
||
timestamp: Some(crate::protocol::now_timestamp()),
|
||
code: "PARSE_ERROR".to_string(),
|
||
message: e.to_string(),
|
||
subagent_task_id: None,
|
||
})
|
||
.await;
|
||
return Ok(());
|
||
}
|
||
};
|
||
|
||
// 创建命令路由器
|
||
let store = state.session_manager.store();
|
||
let skills = state.session_manager.skills();
|
||
let skills_for_handler = skills.clone();
|
||
let provider_config = state
|
||
.config
|
||
.read()
|
||
.await
|
||
.get_provider_config("default")
|
||
.map_err(|e| AgentError::Other(e.to_string()))?;
|
||
let prompt_repository = state.session_manager.store().clone();
|
||
|
||
// 与 AgentFactory::create 共享同一构建逻辑，确保 /save、/save-session、
|
||
// /current 保存/展示的系统提示词与 LLM 实际接收的完全一致
|
||
let system_prompt_provider = build_system_prompt_provider(
|
||
0, // 命令侧不需要 reinject 逻辑
|
||
provider_config.clone(),
|
||
prompt_repository,
|
||
skills,
|
||
state.session_manager.experts(),
|
||
state.session_manager.subagent_runtime(),
|
||
);
|
||
|
||
let mut router = CommandRouter::new();
|
||
// 注册 Session 处理器
|
||
let session_handler = SessionCommandHandler::new(store.clone())
|
||
.with_session_manager(state.session_manager.clone());
|
||
router.register(Box::new(session_handler));
|
||
// 注册 list_sessions 处理器
|
||
router.register(Box::new(ListSessionsCommandHandler::new(store.clone())));
|
||
// 注册 list_sessions_by_channel 处理器
|
||
router.register(Box::new(ListSessionsByChannelCommandHandler::new(
|
||
store.clone(),
|
||
)));
|
||
// 注册 list_channels 处理器
|
||
router.register(Box::new(ListChannelsCommandHandler::new(Arc::new(
|
||
state.channel_manager.clone(),
|
||
))));
|
||
// 注册 list_topics 处理器
|
||
router.register(Box::new(ListTopicsCommandHandler::new(store.clone())));
|
||
// 注册 switch_topic 处理器
|
||
let switch_handler = SwitchTopicCommandHandler::new(store.clone())
|
||
.with_session_manager(state.session_manager.clone());
|
||
router.register(Box::new(switch_handler));
|
||
// 注册 get_current 处理器
|
||
router.register(Box::new(
|
||
GetCurrentSessionCommandHandler::new(store.clone())
|
||
.with_system_prompt_provider(system_prompt_provider.clone()),
|
||
));
|
||
// 注册 load_topic 处理器
|
||
router.register(Box::new(LoadTopicCommandHandler::new(store.clone())));
|
||
// 注册 load_task_messages 处理器
|
||
router.register(Box::new(LoadTaskMessagesCommandHandler::new(
|
||
state.task_repository.clone(),
|
||
store.clone(),
|
||
)));
|
||
router.register(Box::new(SaveSessionCommandHandler::new(
|
||
store.clone(),
|
||
state.task_repository.clone(),
|
||
system_prompt_provider.clone(),
|
||
)));
|
||
// 注册 save_topic 处理器
|
||
router.register(Box::new(SaveTopicCommandHandler::new(
|
||
store.clone(),
|
||
state.task_repository.clone(),
|
||
system_prompt_provider.clone(),
|
||
)));
|
||
// 注册 delete_topic 处理器
|
||
router.register(Box::new(
|
||
DeleteTopicCommandHandler::new(store.clone())
|
||
.with_session_manager(state.session_manager.clone()),
|
||
));
|
||
// 注册 rename_topic 处理器
|
||
router.register(Box::new(RenameTopicCommandHandler::new(store.clone())));
|
||
// 注册 help 处理器
|
||
let metadata = router.metadata_arc();
|
||
router.register(Box::new(HelpCommandHandler::new(metadata)));
|
||
// 注册 list_scheduler_jobs 处理器
|
||
router.register(Box::new(ListSchedulerJobsCommandHandler::new(
|
||
store.clone(),
|
||
)));
|
||
// 注册 list_memories 处理器
|
||
router.register(Box::new(ListMemoriesCommandHandler::new(store.clone())));
|
||
// 注册 list_skills 处理器
|
||
router.register(Box::new(ListSkillsCommandHandler::new(skills_for_handler)));
|
||
// 注册 list_todos 处理器
|
||
router.register(Box::new(ListTodosCommandHandler::new(store.clone())));
|
||
// 注册 memory_crud 处理器
|
||
router.register(Box::new(MemoryCrudCommandHandler::new(store.clone())));
|
||
// 注册 load_chat_messages 处理器
|
||
router.register(Box::new(LoadChatMessagesCommandHandler::new()));
|
||
// 注册 stop_execution 处理器
|
||
router.register(Box::new(StopExecutionCommandHandler::new(
|
||
state.cancel_manager.clone(),
|
||
state.session_manager.clone(),
|
||
state.subagent_executor.clone(),
|
||
)));
|
||
|
||
// 构建命令上下文
|
||
tracing::debug!(
|
||
current_session_id = %current_session_id,
|
||
current_topic_id = ?current_topic_id,
|
||
"Building CommandContext for WebSocket command"
|
||
);
|
||
let mut cmd_ctx = CommandContext::new("websocket", "websocket")
|
||
.with_session_id(current_session_id.as_str())
|
||
.with_chat_id(current_session_id.as_str());
|
||
// 只在有 topic_id 时才设置
|
||
if let Some(ref topic_id) = *current_topic_id {
|
||
cmd_ctx = cmd_ctx.with_topic_id(topic_id.as_str());
|
||
}
|
||
|
||
// 执行命令
|
||
let response = router.dispatch_with_response(cmd, cmd_ctx).await;
|
||
|
||
// 处理响应
|
||
if response.success {
|
||
// 更新当前会话 ID（如果是创建会话）；仅在变化时记录日志
|
||
if let Some(session_id) = response.metadata.get("session_id") {
|
||
if session_id != current_session_id {
|
||
tracing::info!(
|
||
old_session_id = %current_session_id,
|
||
new_session_id = %session_id,
|
||
"Updating current_session_id"
|
||
);
|
||
}
|
||
*current_session_id = session_id.clone();
|
||
let _ = state
|
||
.channel_manager
|
||
.websocket_channel()
|
||
.register_connection(
|
||
session_id.clone(),
|
||
runtime_session_id.to_string(),
|
||
sender.clone(),
|
||
)
|
||
.await;
|
||
}
|
||
// 更新当前话题 ID（如果是创建话题或切换话题）
|
||
if let Some(topic_id) = response.metadata.get("topic_id") {
|
||
tracing::info!(
|
||
old_topic_id = ?current_topic_id,
|
||
new_topic_id = %topic_id,
|
||
"Updating current_topic_id"
|
||
);
|
||
*current_topic_id = Some(topic_id.clone());
|
||
|
||
// 加载并发送该话题的历史消息
|
||
if let Err(e) = send_topic_history(
|
||
&store,
|
||
current_session_id,
|
||
topic_id,
|
||
sender,
|
||
&state.task_repository,
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!(error = %e, topic_id = %topic_id, "Failed to send topic history");
|
||
}
|
||
}
|
||
// 加载子智能体任务消息
|
||
if let Some(task_session_id) = response.metadata.get("task_session_id") {
|
||
// 提前提取 task_id，用于给历史消息打标记
|
||
let task_id = response
|
||
.metadata
|
||
.get("task_id")
|
||
.cloned()
|
||
.unwrap_or_default();
|
||
if let Err(e) = send_task_messages(
|
||
&store,
|
||
task_session_id,
|
||
sender,
|
||
Some(task_id.clone()),
|
||
Some(&state.task_repository),
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!(error = %e, task_session_id = %task_session_id, "Failed to send task messages");
|
||
}
|
||
|
||
// 发送 TaskMessagesLoaded 元数据
|
||
let description = response
|
||
.metadata
|
||
.get("task_description")
|
||
.cloned()
|
||
.unwrap_or_default();
|
||
let subagent_type = response
|
||
.metadata
|
||
.get("task_subagent_type")
|
||
.cloned()
|
||
.unwrap_or_default();
|
||
let status = response
|
||
.metadata
|
||
.get("task_status")
|
||
.cloned()
|
||
.unwrap_or_default();
|
||
let summary = response.metadata.get("task_summary").cloned();
|
||
let token_stats = response.metadata.get("task_token_stats").and_then(|json| {
|
||
serde_json::from_str::<crate::protocol::TopicTokenStats>(json).ok()
|
||
});
|
||
|
||
let _ = sender
|
||
.send(WsOutbound::TaskMessagesLoaded {
|
||
task_id,
|
||
description,
|
||
subagent_type,
|
||
status,
|
||
summary,
|
||
token_stats,
|
||
})
|
||
.await;
|
||
}
|
||
|
||
// 处理定时任务列表
|
||
if let Some(jobs_json) = response.metadata.get("scheduler_jobs") {
|
||
if let Ok(jobs) =
|
||
serde_json::from_str::<Vec<crate::protocol::SchedulerJobSummary>>(jobs_json)
|
||
{
|
||
let _ = sender.send(WsOutbound::SchedulerJobList { jobs }).await;
|
||
}
|
||
}
|
||
|
||
// 处理技能列表
|
||
if let Some(skills_json) = response.metadata.get("skills") {
|
||
if let Ok(skills) =
|
||
serde_json::from_str::<Vec<crate::protocol::SkillSummary>>(skills_json)
|
||
{
|
||
let _ = sender.send(WsOutbound::SkillList { skills }).await;
|
||
}
|
||
}
|
||
|
||
// 处理 Todo 列表
|
||
if let Some(todos_json) = response.metadata.get("todos") {
|
||
if let Ok(todos) =
|
||
serde_json::from_str::<Vec<crate::protocol::TodoItemSummary>>(todos_json)
|
||
{
|
||
let scope_key = response
|
||
.metadata
|
||
.get("todos_scope_key")
|
||
.cloned()
|
||
.unwrap_or_default();
|
||
tracing::debug!(todo_count = todos.len(), %scope_key, "list_todos command response");
|
||
let _ = sender.send(WsOutbound::TodoList { todos, scope_key }).await;
|
||
}
|
||
}
|
||
|
||
// 处理记忆列表
|
||
if let Some(memories_json) = response.metadata.get("memories") {
|
||
if let Ok(memories) =
|
||
serde_json::from_str::<Vec<crate::protocol::MemorySummary>>(memories_json)
|
||
{
|
||
let _ = sender.send(WsOutbound::MemoryList { memories }).await;
|
||
}
|
||
}
|
||
|
||
// 记忆 CRUD 后自动刷新列表
|
||
if response.metadata.get("memory_updated").map(|v| v.as_str()) == Some("true") {
|
||
if let Ok(records) =
|
||
store.list_memories_for_scope("user", crate::storage::GLOBAL_SCOPE_KEY)
|
||
{
|
||
let memories: Vec<crate::protocol::MemorySummary> = records
|
||
.into_iter()
|
||
.filter(|m| m.namespace != "_meta")
|
||
.map(|m| crate::protocol::MemorySummary {
|
||
id: m.id,
|
||
namespace: m.namespace,
|
||
memory_key: m.memory_key,
|
||
content: m.content,
|
||
created_at: m.created_at,
|
||
updated_at: m.updated_at,
|
||
})
|
||
.collect();
|
||
let _ = sender.send(WsOutbound::MemoryList { memories }).await;
|
||
}
|
||
}
|
||
|
||
// 处理加载聊天消息请求
|
||
if let Some(load_chat_id) = response.metadata.get("load_chat_id") {
|
||
let load_chat_channel = response
|
||
.metadata
|
||
.get("load_chat_channel")
|
||
.cloned()
|
||
.unwrap_or_default();
|
||
// session_id = "{channel}:{chat_id}" (cli channel 例外)
|
||
let session_id =
|
||
crate::storage::persistent_session_id(&load_chat_channel, load_chat_id);
|
||
if let Err(e) =
|
||
send_task_messages(&store, &session_id, sender, None, None).await
|
||
{
|
||
tracing::warn!(
|
||
error = %e,
|
||
channel = %load_chat_channel,
|
||
chat_id = %load_chat_id,
|
||
session_id = %session_id,
|
||
"Failed to send chat messages"
|
||
);
|
||
}
|
||
}
|
||
|
||
if current_topic_id.is_none() {
|
||
if let Some(topics_json) = response.metadata.get("topics") {
|
||
match serde_json::from_str::<Vec<crate::protocol::TopicSummary>>(
|
||
topics_json,
|
||
) {
|
||
Ok(topics) => {
|
||
if let Some(first_topic) = topics.first() {
|
||
let topic_id = first_topic.topic_id.clone();
|
||
*current_topic_id = Some(topic_id.clone());
|
||
if let Err(e) = send_topic_history(
|
||
&store,
|
||
current_session_id,
|
||
&topic_id,
|
||
sender,
|
||
&state.task_repository,
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!(error = %e, topic_id = %topic_id, "Failed to send initial topic history");
|
||
}
|
||
}
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, "Failed to parse topics metadata for initial history");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
} else if let Some(ref error) = response.error {
|
||
tracing::warn!(
|
||
error_code = %error.code,
|
||
error_message = %error.message,
|
||
"Command failed"
|
||
);
|
||
}
|
||
|
||
// 适配并发送响应
|
||
let outbounds = output_adapter.adapt(response);
|
||
for msg in outbounds {
|
||
let _ = sender.send(msg).await;
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
WsInbound::Ping => {
|
||
let _ = sender.send(WsOutbound::Pong).await;
|
||
Ok(())
|
||
}
|
||
}
|
||
}
|
||
|
||
fn resolve_ws_sender_id(sender_id: Option<&str>, runtime_session_id: &str) -> String {
|
||
sender_id
|
||
.map(str::trim)
|
||
.filter(|sender_id| !sender_id.is_empty())
|
||
.map(ToOwned::to_owned)
|
||
.unwrap_or_else(|| runtime_session_id.to_string())
|
||
}
|
||
|
||
/// 加载并发送话题历史消息
|
||
async fn send_topic_history(
|
||
store: &Arc<crate::storage::SessionStore>,
|
||
session_id: &str,
|
||
topic_id: &str,
|
||
sender: &mpsc::Sender<WsOutbound>,
|
||
task_repository: &Arc<dyn TaskRepository>,
|
||
) -> Result<(), Box<dyn std::error::Error>> {
|
||
// 加载话题消息，按 session_id 过滤，避免混入子智能体消息
|
||
let messages = store.load_messages_for_topic_full(topic_id, Some(session_id))?;
|
||
|
||
tracing::info!(topic_id = %topic_id, message_count = messages.len(), "Sending topic history");
|
||
|
||
// 收集已有 tool_result 的 tool_call_id 集合，用于判断任务是否已有结果
|
||
let mut tool_call_ids_with_results: std::collections::HashSet<String> =
|
||
std::collections::HashSet::new();
|
||
for msg in &messages {
|
||
if msg.role == "tool" {
|
||
if let Some(ref tcid) = msg.tool_call_id {
|
||
tool_call_ids_with_results.insert(tcid.clone());
|
||
}
|
||
}
|
||
}
|
||
|
||
// 将消息转换为 WsOutbound 并发送
|
||
for msg in messages {
|
||
for outbound in chat_message_to_ws_outbound(&msg) {
|
||
let _ = sender.send(outbound).await;
|
||
}
|
||
}
|
||
|
||
// 查询该话题下所有子智能体任务，补发 TaskStarted 事件
|
||
// 解决页面刷新后 navigateToTaskId 丢失的问题
|
||
let tasks = match task_repository.list_tasks_for_topic(topic_id).await {
|
||
Ok(tasks) => tasks,
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, topic_id = %topic_id, "Failed to list tasks for topic");
|
||
return Ok(());
|
||
}
|
||
};
|
||
|
||
for task in tasks {
|
||
// 判断是否需要补发 TaskStarted：
|
||
// - 如果该任务的 tool_call_id 已有对应的 tool_result，前端会显示结果，不需要补发
|
||
// - 否则（Running 状态或已完成但结果未进入历史），补发 TaskStarted 以便前端显示"查看实时进度"
|
||
let has_tool_result = task
|
||
.tool_call_id
|
||
.as_ref()
|
||
.map(|tcid| tool_call_ids_with_results.contains(tcid))
|
||
.unwrap_or(false);
|
||
if has_tool_result {
|
||
continue;
|
||
}
|
||
|
||
// 判断是否为孙智能体：parent_session_id 以 "sub:" 开头表示父会话是子智能体
|
||
let parent_task_id = extract_parent_task_id(&task);
|
||
|
||
tracing::info!(
|
||
task_id = %task.id,
|
||
description = %task.description,
|
||
parent_task_id = ?parent_task_id,
|
||
tool_call_id = ?task.tool_call_id,
|
||
state = ?task.state,
|
||
"Re-sending TaskStarted for task without tool_result after topic history load"
|
||
);
|
||
let _ = sender
|
||
.send(WsOutbound::TaskStarted {
|
||
task_id: task.id.clone(),
|
||
description: task.description.clone(),
|
||
subagent_type: task.subagent_type.clone(),
|
||
topic_id: Some(topic_id.to_string()),
|
||
parent_task_id,
|
||
tool_call_id: task.tool_call_id.clone(),
|
||
})
|
||
.await;
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// 加载并发送子智能体任务的历史消息
|
||
async fn send_task_messages(
|
||
store: &Arc<crate::storage::SessionStore>,
|
||
session_id: &str,
|
||
sender: &mpsc::Sender<WsOutbound>,
|
||
subagent_task_id: Option<String>,
|
||
task_repository: Option<&Arc<dyn TaskRepository>>,
|
||
) -> Result<(), Box<dyn std::error::Error>> {
|
||
let messages = store.load_messages(session_id)?;
|
||
|
||
tracing::info!(session_id = %session_id, message_count = messages.len(), "Sending task messages");
|
||
|
||
// 收集已有 tool_result 的 tool_call_id 集合，用于判断子任务是否已有结果
|
||
let mut tool_call_ids_with_results: std::collections::HashSet<String> =
|
||
std::collections::HashSet::new();
|
||
for msg in &messages {
|
||
if msg.role == "tool" {
|
||
if let Some(ref tcid) = msg.tool_call_id {
|
||
tool_call_ids_with_results.insert(tcid.clone());
|
||
}
|
||
}
|
||
}
|
||
|
||
for msg in messages {
|
||
let mut outbounds = chat_message_to_ws_outbound(&msg);
|
||
if let Some(ref task_id) = subagent_task_id {
|
||
for ob in &mut outbounds {
|
||
set_subagent_task_id(ob, task_id);
|
||
}
|
||
}
|
||
for outbound in outbounds {
|
||
let _ = sender.send(outbound).await;
|
||
}
|
||
}
|
||
|
||
// 补发子任务（孙智能体）的 TaskStarted 事件
|
||
// 解决重新进入子智能体视图后 navigateToTaskId 丢失的问题
|
||
if let (Some(repo), Some(parent_task_id)) = (task_repository, &subagent_task_id) {
|
||
match repo.list_tasks_for_session(session_id).await {
|
||
Ok(child_tasks) => {
|
||
for child in child_tasks {
|
||
// 如果该子任务的 tool_call_id 已有对应的 tool_result，前端会显示结果，不需要补发
|
||
let has_tool_result = child
|
||
.tool_call_id
|
||
.as_ref()
|
||
.map(|tcid| tool_call_ids_with_results.contains(tcid))
|
||
.unwrap_or(false);
|
||
if has_tool_result {
|
||
continue;
|
||
}
|
||
|
||
tracing::info!(
|
||
child_task_id = %child.id,
|
||
parent_task_id = %parent_task_id,
|
||
tool_call_id = ?child.tool_call_id,
|
||
state = ?child.state,
|
||
"Re-sending TaskStarted for child task without tool_result after sub-agent view re-enter"
|
||
);
|
||
let _ = sender
|
||
.send(WsOutbound::TaskStarted {
|
||
task_id: child.id.clone(),
|
||
description: child.description.clone(),
|
||
subagent_type: child.subagent_type.clone(),
|
||
topic_id: child.parent_topic_id.clone(),
|
||
parent_task_id: Some(parent_task_id.clone()),
|
||
tool_call_id: child.tool_call_id.clone(),
|
||
})
|
||
.await;
|
||
}
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, session_id = %session_id, "Failed to list child tasks for resend");
|
||
}
|
||
}
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// 给 WsOutbound 消息注入 subagent_task_id（仅对有该字段的变体生效）
|
||
fn set_subagent_task_id(outbound: &mut WsOutbound, task_id: &str) {
|
||
match outbound {
|
||
WsOutbound::AssistantResponse {
|
||
subagent_task_id, ..
|
||
}
|
||
| WsOutbound::ToolCall {
|
||
subagent_task_id, ..
|
||
}
|
||
| WsOutbound::ToolResult {
|
||
subagent_task_id, ..
|
||
}
|
||
| WsOutbound::ToolPending {
|
||
subagent_task_id, ..
|
||
}
|
||
| WsOutbound::StreamDelta {
|
||
subagent_task_id, ..
|
||
}
|
||
| WsOutbound::StreamEnd {
|
||
subagent_task_id, ..
|
||
} => {
|
||
*subagent_task_id = Some(task_id.to_string());
|
||
}
|
||
_ => {} // 其他变体没有 subagent_task_id 字段
|
||
}
|
||
}
|
||
|
||
/// 从 TaskSession 中提取父任务 ID（仅孙智能体有值）。
|
||
/// 孙智能体的 parent_session_id 格式为 "sub:{grandparent_session}:task:{parent_task_uuid}"，
|
||
/// 从中提取 "task:{parent_task_uuid}" 作为 parent_task_id。
|
||
fn extract_parent_task_id(task: &crate::tools::task::types::TaskSession) -> Option<String> {
|
||
let parent = &task.parent_session_id;
|
||
// 仅当父会话是子智能体会话时才提取（格式: "sub:...:task:{uuid}"）
|
||
if parent.starts_with("sub:") {
|
||
if let Some(pos) = parent.find(":task:") {
|
||
return Some(parent[pos + 1..].to_string()); // "task:{uuid}"
|
||
}
|
||
}
|
||
None
|
||
}
|
||
|
||
/// 将 ChatMessage 转换为 WsOutbound 列表
|
||
fn chat_message_to_ws_outbound(msg: &crate::bus::ChatMessage) -> Vec<WsOutbound> {
|
||
use crate::bus::message::ToolMessageState;
|
||
|
||
// Helper function to strip media_refs_json from content
|
||
fn strip_media_refs_json(content: &str) -> String {
|
||
// Remove the media_refs_json suffix if present
|
||
if let Some(pos) = content.find("\n\nmedia_refs_json:") {
|
||
content[..pos].to_string()
|
||
} else {
|
||
content.to_string()
|
||
}
|
||
}
|
||
|
||
// Build attachments from media_refs, reading file content for base64
|
||
let attachments: Vec<MediaSummary> = msg
|
||
.media_refs
|
||
.iter()
|
||
.filter_map(|path| {
|
||
// Try to read file and encode as base64
|
||
let file_content = std::fs::read(path).ok()?;
|
||
let base64_content = STANDARD.encode(&file_content);
|
||
|
||
// Guess mime type from path
|
||
let mime_type = mime_guess::from_path(path)
|
||
.first_raw()
|
||
.map(ToOwned::to_owned);
|
||
|
||
// Determine media type from mime type
|
||
let media_type = mime_type
|
||
.as_ref()
|
||
.map(|m| {
|
||
if m.starts_with("image/") {
|
||
"image"
|
||
} else if m.starts_with("audio/") {
|
||
"audio"
|
||
} else if m.starts_with("video/") {
|
||
"video"
|
||
} else {
|
||
"file"
|
||
}
|
||
})
|
||
.unwrap_or("file");
|
||
|
||
// Get file name from path
|
||
let file_name = std::path::Path::new(path)
|
||
.file_name()
|
||
.and_then(|name| name.to_str())
|
||
.map(ToOwned::to_owned);
|
||
|
||
Some(MediaSummary {
|
||
path: path.clone(),
|
||
media_type: media_type.to_string(),
|
||
mime_type,
|
||
content_base64: Some(base64_content),
|
||
file_name,
|
||
})
|
||
})
|
||
.collect();
|
||
|
||
match msg.role.as_str() {
|
||
"assistant" => {
|
||
if let Some(tool_calls) = &msg.tool_calls {
|
||
let mut outbound = Vec::new();
|
||
let has_content_or_reasoning =
|
||
!msg.content.trim().is_empty() || msg.reasoning_content.is_some();
|
||
if has_content_or_reasoning {
|
||
outbound.push(WsOutbound::AssistantResponse {
|
||
id: msg.id.clone(),
|
||
content: msg.content.clone(),
|
||
role: msg.role.clone(),
|
||
attachments: Vec::new(),
|
||
subagent_task_id: None,
|
||
topic_id: None,
|
||
timestamp: Some(msg.timestamp / 1000),
|
||
reasoning_content: msg.reasoning_content.clone(),
|
||
user_message_id: None,
|
||
});
|
||
}
|
||
// AssistantResponse 已携带 reasoning 时，ToolCall 不再重复
|
||
let tc_reasoning = if has_content_or_reasoning {
|
||
None
|
||
} else {
|
||
msg.reasoning_content.clone()
|
||
};
|
||
for tool_call in tool_calls {
|
||
outbound.push(WsOutbound::ToolCall {
|
||
id: tool_call.id.clone(),
|
||
tool_call_id: tool_call.id.clone(),
|
||
tool_name: tool_call.name.clone(),
|
||
arguments: tool_call.arguments.clone(),
|
||
content: format!("{}\nargs: {}", tool_call.name, tool_call.arguments),
|
||
role: msg.role.clone(),
|
||
subagent_task_id: None,
|
||
topic_id: None,
|
||
timestamp: Some(msg.timestamp / 1000),
|
||
reasoning_content: tc_reasoning.clone(),
|
||
user_message_id: None,
|
||
});
|
||
}
|
||
outbound
|
||
} else {
|
||
// 普通助手消息
|
||
vec![WsOutbound::AssistantResponse {
|
||
id: msg.id.clone(),
|
||
content: msg.content.clone(),
|
||
role: msg.role.clone(),
|
||
attachments: Vec::new(),
|
||
subagent_task_id: None,
|
||
topic_id: None,
|
||
timestamp: Some(msg.timestamp / 1000),
|
||
reasoning_content: msg.reasoning_content.clone(),
|
||
user_message_id: None,
|
||
}]
|
||
}
|
||
}
|
||
"tool" => {
|
||
let tool_state = msg
|
||
.tool_state
|
||
.as_ref()
|
||
.unwrap_or(&ToolMessageState::Completed);
|
||
match tool_state {
|
||
ToolMessageState::Completed => vec![WsOutbound::ToolResult {
|
||
id: msg
|
||
.tool_call_id
|
||
.clone()
|
||
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
|
||
tool_call_id: msg.tool_call_id.clone().unwrap_or_default(),
|
||
tool_name: msg.tool_name.clone().unwrap_or_default(),
|
||
content: msg.content.clone(),
|
||
role: msg.role.clone(),
|
||
subagent_task_id: None,
|
||
topic_id: None,
|
||
duration_ms: msg.tool_duration_ms,
|
||
timestamp: Some(msg.timestamp / 1000),
|
||
}],
|
||
ToolMessageState::PendingUserAction => vec![WsOutbound::ToolPending {
|
||
id: msg
|
||
.tool_call_id
|
||
.clone()
|
||
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
|
||
tool_call_id: msg.tool_call_id.clone().unwrap_or_default(),
|
||
tool_name: msg.tool_name.clone().unwrap_or_default(),
|
||
content: msg.content.clone(),
|
||
role: msg.role.clone(),
|
||
resume_hint: "完成外部操作后，直接发一条继续消息即可。".to_string(),
|
||
subagent_task_id: None,
|
||
topic_id: None,
|
||
timestamp: Some(msg.timestamp / 1000),
|
||
}],
|
||
}
|
||
}
|
||
"user" => vec![WsOutbound::AssistantResponse {
|
||
id: msg.id.clone(),
|
||
content: strip_media_refs_json(&msg.content),
|
||
role: msg.role.clone(),
|
||
attachments,
|
||
subagent_task_id: None,
|
||
topic_id: None,
|
||
timestamp: Some(msg.timestamp / 1000),
|
||
reasoning_content: None,
|
||
user_message_id: None,
|
||
}],
|
||
_ => Vec::new(),
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::{build_media_filename, process_attachments_with_base64, resolve_ws_sender_id};
|
||
use crate::protocol::MediaSummary;
|
||
use base64::{Engine as _, engine::general_purpose::STANDARD};
|
||
|
||
#[test]
|
||
fn test_resolve_ws_sender_id_prefers_inbound_sender() {
|
||
assert_eq!(
|
||
resolve_ws_sender_id(Some("user-42"), "runtime-1"),
|
||
"user-42"
|
||
);
|
||
assert_eq!(
|
||
resolve_ws_sender_id(Some(" user-42 "), "runtime-1"),
|
||
"user-42"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn test_resolve_ws_sender_id_falls_back_to_runtime_session_id() {
|
||
assert_eq!(resolve_ws_sender_id(None, "runtime-1"), "runtime-1");
|
||
assert_eq!(resolve_ws_sender_id(Some(" "), "runtime-1"), "runtime-1");
|
||
}
|
||
|
||
#[test]
|
||
fn test_build_media_filename_preserves_original_name() {
|
||
let filename = build_media_filename("image", Some("photo.png"));
|
||
assert!(filename.ends_with("_photo.png"));
|
||
// UUID is 36 chars, plus underscore and original name
|
||
assert!(filename.len() >= 36 + 1 + "photo.png".len());
|
||
}
|
||
|
||
#[test]
|
||
fn test_build_media_filename_generates_default_when_no_name() {
|
||
let filename = build_media_filename("image", None);
|
||
assert!(filename.starts_with("image_"));
|
||
}
|
||
|
||
#[test]
|
||
fn test_process_attachments_with_base64_saves_to_file() {
|
||
let test_content = "test image content";
|
||
let base64_content = STANDARD.encode(test_content.as_bytes());
|
||
|
||
let attachments = vec![MediaSummary {
|
||
path: "test_image.png".to_string(),
|
||
media_type: "image".to_string(),
|
||
mime_type: Some("image/png".to_string()),
|
||
content_base64: Some(base64_content.clone()),
|
||
file_name: Some("test_image.png".to_string()),
|
||
}];
|
||
|
||
let result = process_attachments_with_base64(attachments).unwrap();
|
||
|
||
// Verify path is now a full path
|
||
assert!(result[0].path.contains(".picobot"));
|
||
assert!(result[0].path.contains("media"));
|
||
assert!(result[0].path.contains("ws"));
|
||
|
||
// Verify content_base64 is kept for frontend display
|
||
assert!(result[0].content_base64.is_some());
|
||
assert_eq!(result[0].content_base64.as_ref().unwrap(), &base64_content);
|
||
|
||
// Verify file was actually written
|
||
let file_content = std::fs::read(&result[0].path).unwrap();
|
||
assert_eq!(file_content, test_content.as_bytes());
|
||
|
||
// Cleanup
|
||
std::fs::remove_file(&result[0].path).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn test_process_attachments_without_base64_keeps_original_path() {
|
||
let attachments = vec![MediaSummary {
|
||
path: "/existing/path/image.jpg".to_string(),
|
||
media_type: "image".to_string(),
|
||
mime_type: Some("image/jpeg".to_string()),
|
||
content_base64: None,
|
||
file_name: Some("image.jpg".to_string()),
|
||
}];
|
||
|
||
let result = process_attachments_with_base64(attachments).unwrap();
|
||
|
||
// Path should remain unchanged
|
||
assert_eq!(result[0].path, "/existing/path/image.jpg");
|
||
}
|
||
|
||
#[test]
|
||
fn test_process_empty_attachments_returns_empty_vec() {
|
||
let result = process_attachments_with_base64(Vec::new()).unwrap();
|
||
assert!(result.is_empty());
|
||
}
|
||
}
|