refactor: 服务层抽象下沉、异步锁全栈迁移、客户端韧性加固与移动端适配

- 服务层拆分：删除 api/helpers.rs，新增 citation/note/session/pipeline/paper/vision 独立服务模块 - Agent 工具精简：paper_content+paper_outline 合并为 paper.rs，图片分析逻辑下沉至 services/vision - 并发模型升级：std::sync::{Mutex,RwLock} → tokio::sync::{Mutex,RwLock}，消除 async 上下文中的阻塞风险 - 客户端加固：HTTP 客户端统一超时配置、ADS 429 / arXiv 503 自动重试、构造函数返回 Result - 启动安全：全局 panic hook 日志化、空密码拒绝启动、向量表维度不匹配需显式确认 - CLI 扩展：构建完整 AppState 复用服务层，新增 Content/Outline/Citations/Search/Process 子命令 - 前端：移动端汉堡菜单、侧栏滑出面板、引用星系触屏手势（单指拖拽/双指缩放）
2026-06-30 19:26:01 +08:00
parent c5fd5b0d66
commit f885c0a4a8
90 changed files with 5184 additions and 3916 deletions
@@ -65,6 +65,7 @@ impl CompactionCircuitBreaker {

    /// 设置可注入的时间源（仅用于测试）。
    #[cfg(test)]
+    #[allow(dead_code)]
    fn with_time_source(mut self, f: Box<dyn Fn() -> Instant + Send>) -> Self {
        self.time_source = Some(f);
        self
@@ -512,10 +512,10 @@ mod tests {
    #[test]
    fn test_backoff_delay() {
        let d0 = backoff_delay(0, None);
-        assert!(d0 >= 500 && d0 <= 700);
+        assert!((500..=700).contains(&d0));

        let d3 = backoff_delay(3, None);
-        assert!(d3 >= 4000 && d3 <= 5000);
+        assert!((4000..=5000).contains(&d3));

        let d10 = backoff_delay(10, None);
        assert!(d10 <= 40_000);
@@ -413,37 +413,7 @@ pub async fn execute_parallel(

                    // 存储待处理的权限请求
                    {
-                        let mut perms = match app_state.pending_permissions.lock() {
-                            Ok(p) => p,
-                            Err(e) => {
-                                warn!("[Executor] 权限系统内部错误: {}", e);
-                                let err_output =
-                                    format!("权限系统内部错误，工具 {} 被拒绝", prep.tool_name);
-                                let _ = tx.send(AgentStreamEvent::ToolResult {
-                                    tool_call_id: prep.tool_call_id.clone(),
-                                    name: prep.tool_name.clone(),
-                                    output: err_output.clone(),
-                                    is_error: true,
-                                    metadata: serde_json::json!({}),
-                                    step,
-                                });
-                                let err_msg =
-                                    ChatMessage::tool_result(&prep.tool_call_id, &err_output);
-                                save_tool_message_sync(db, &sid, turn_index, step, &err_msg);
-                                tool_messages.push(ToolResultMessage {
-                                    chat_message: err_msg,
-                                    was_error: true,
-                                });
-                                // 内部错误 → 记录拒绝追踪
-                                if let Some(dt) = denial_tracker {
-                                    if let Ok(mut tracker) = dt.lock() {
-                                        tracker.record_denial();
-                                    }
-                                }
-                                denied_indices.insert(i);
-                                continue;
-                            }
-                        };
+                        let mut perms = app_state.pending_permissions.lock().await;
                        perms.insert(
                            perm_id.clone(),
                            PendingPermission {
@@ -461,9 +431,7 @@ pub async fn execute_parallel(
                    let perm_result = tokio::time::timeout(timeout_dur, resp_rx).await;

                    // 清理待处理的权限请求
-                    if let Ok(mut perms) = app_state.pending_permissions.lock() {
-                        perms.remove(&perm_id);
-                    }
+                    app_state.pending_permissions.lock().await.remove(&perm_id);

                    match perm_result {
                        Ok(Ok(response)) if response.allowed => {
@@ -570,11 +538,10 @@ pub async fn execute_parallel(
    let cancel_handle = tokio::spawn(async move {
        loop {
            tokio::time::sleep(std::time::Duration::from_millis(250)).await;
-            if let Ok(locked) = app_state_ref.cancelled_runs.lock() {
-                if locked.contains(&sid_ref) {
-                    cancel_flag.store(true, Ordering::SeqCst);
-                    return;
-                }
+            let locked = app_state_ref.cancelled_runs.lock().await;
+            if locked.contains(&sid_ref) {
+                cancel_flag.store(true, Ordering::SeqCst);
+                return;
            }
        }
    });
@@ -14,7 +14,6 @@

 use lru::LruCache;
 use std::num::NonZeroUsize;
-use std::path::Path;
 use tracing::{info, warn};

 /// 缓存条目最大数量
@@ -82,35 +81,27 @@ impl FileStateCache {
        }
    }

-    /// 规范化路径 key（确保一致性）
+    /// 规范化路径 key（确保一致性）—— 仅做字符串规范化，不做磁盘 I/O
    fn normalize_key(path: &str) -> String {
-        // 去除尾随斜杠，规范化重复斜杠
-        let p = Path::new(path);
-        // 尝试 canonicalize（跟随符号链接），失败则用简单的字符串规范化
-        match p.canonicalize() {
-            Ok(canon) => canon.to_string_lossy().to_string(),
-            Err(_) => {
-                // 简单规范化：折叠重复的 /
-                let mut result = String::with_capacity(path.len());
-                let mut prev_slash = false;
-                for ch in path.chars() {
-                    if ch == '/' || ch == '\\' {
-                        if !prev_slash {
-                            result.push('/');
-                            prev_slash = true;
-                        }
-                    } else {
-                        result.push(ch);
-                        prev_slash = false;
-                    }
+        // 简单规范化：折叠重复的 / 和 \
+        let mut result = String::with_capacity(path.len());
+        let mut prev_slash = false;
+        for ch in path.chars() {
+            if ch == '/' || ch == '\\' {
+                if !prev_slash {
+                    result.push('/');
+                    prev_slash = true;
                }
-                // 去除尾随 /
-                if result.ends_with('/') && result.len() > 1 {
-                    result.pop();
-                }
-                result
+            } else {
+                result.push(ch);
+                prev_slash = false;
            }
        }
+        // 去除尾随 /
+        if result.ends_with('/') && result.len() > 1 {
+            result.pop();
+        }
+        result
    }

    /// 获取缓存的条目，返回 Some(&FileState) 若存在
@@ -74,7 +74,7 @@ pub struct AgentRuntime {
    /// 后台任务通知队列（支持 bg_task_run/bg_task_check）
    bg_notification_queue: Arc<BgNotificationQueue>,
    /// 指标采集 hook 的数据引用（供 API 查询）
-    metrics_data: Arc<std::sync::Mutex<super::hooks::MetricsData>>,
+    metrics_data: Arc<tokio::sync::Mutex<super::hooks::MetricsData>>,
    /// 压缩熔断器（跨 turn 共享，防止无限压缩循环）
    compaction_breaker: Arc<std::sync::Mutex<circuit_breaker::CompactionCircuitBreaker>>,
    /// 权限检查器
@@ -111,7 +111,7 @@ impl AgentRuntime {
        apply_mode_config(&mut config, mode);

        let queue = Arc::new(BgNotificationQueue::new());
-        let metrics_data = Arc::new(std::sync::Mutex::new(super::hooks::MetricsData::default()));
+        let metrics_data = Arc::new(tokio::sync::Mutex::new(super::hooks::MetricsData::default()));
        let permission_checker = Arc::new(permission::PermissionChecker::from_config(&config));
        let denial_tracker = Arc::new(std::sync::Mutex::new(denial_tracker::DenialTracker::new(
            config.denial_max_consecutive,
@@ -182,7 +182,7 @@ impl AgentRuntime {
        apply_mode_config(&mut config, mode);

        let queue = Arc::new(BgNotificationQueue::new());
-        let metrics_data = Arc::new(std::sync::Mutex::new(super::hooks::MetricsData::default()));
+        let metrics_data = Arc::new(tokio::sync::Mutex::new(super::hooks::MetricsData::default()));
        let permission_checker = Arc::new(permission::PermissionChecker::from_config(&config));
        let denial_tracker = Arc::new(std::sync::Mutex::new(denial_tracker::DenialTracker::new(
            config.denial_max_consecutive,
@@ -240,8 +240,7 @@ impl AgentRuntime {
    /// 返回当前运行指标快照（锁异常时返回默认值）
    pub fn get_metrics(&self) -> super::hooks::MetricsData {
        self.metrics_data
-            .lock()
-            .ok()
+            .try_lock()
            .map(|m| m.clone())
            .unwrap_or_default()
    }
@@ -493,11 +492,10 @@ impl AgentRuntime {

        // 注册当前会话的权限检查器（如不存在则从全局配置初始化）
        {
-            if let Ok(mut checkers) = self.app_state.session_permission_checkers.write() {
-                checkers
-                    .entry(sid.clone())
-                    .or_insert_with(|| (*self.permission_checker).clone());
-            }
+            let mut checkers = self.app_state.session_permission_checkers.write().await;
+            checkers
+                .entry(sid.clone())
+                .or_insert_with(|| (*self.permission_checker).clone());
        }

        let tool_defs = self.tool_registry.definitions();
@@ -525,11 +523,8 @@ impl AgentRuntime {

            // 检查用户取消
            let is_cancelled = {
-                if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() {
-                    cancelled.remove(sid)
-                } else {
-                    false
-                }
+                let mut cancelled = self.app_state.cancelled_runs.lock().await;
+                cancelled.remove(sid)
            };

            if is_cancelled {
@@ -874,8 +869,9 @@ impl AgentRuntime {
                self.app_state
                    .session_permission_checkers
                    .read()
-                    .ok()
-                    .and_then(|checkers| checkers.get(sid).cloned())
+                    .await
+                    .get(sid)
+                    .cloned()
            };
            let exec_result = executor::execute_parallel(
                &prepared_calls,
@@ -948,9 +944,8 @@ impl AgentRuntime {
            }

            if exec_result.was_cancelled {
-                if let Ok(mut locked) = self.app_state.cancelled_runs.lock() {
-                    locked.remove(sid);
-                }
+                let mut locked = self.app_state.cancelled_runs.lock().await;
+                locked.remove(sid);
                warn!(
                    "[AgentRuntime] 工具执行期间被用户手动中止，会话 ID: {}",
                    sid
@@ -1018,9 +1013,8 @@ impl AgentRuntime {
        match output.status {
            StreamStatus::Success => return Some(output),
            StreamStatus::Cancelled => {
-                if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() {
-                    cancelled.remove(session_id);
-                }
+                let mut cancelled = self.app_state.cancelled_runs.lock().await;
+                cancelled.remove(session_id);
                warn!(
                    "[AgentRuntime] 流式调用期间被用户手动中止，会话 ID: {}",
                    session_id
@@ -1098,14 +1092,13 @@ impl AgentRuntime {
                }

                // 检查用户取消
-                if let Ok(cancelled) = self.app_state.cancelled_runs.lock() {
-                    if cancelled.contains(session_id) {
-                        warn!("[AgentRuntime] 退避重试期间被用户取消");
-                        let _ = tx.send(AgentStreamEvent::Error {
-                            message: "用户已手动中止执行。".to_string(),
-                        });
-                        return None;
-                    }
+                let cancelled = self.app_state.cancelled_runs.lock().await;
+                if cancelled.contains(session_id) {
+                    warn!("[AgentRuntime] 退避重试期间被用户取消");
+                    let _ = tx.send(AgentStreamEvent::Error {
+                        message: "用户已手动中止执行。".to_string(),
+                    });
+                    return None;
                }

                // 重试 LLM 调用
@@ -1127,9 +1120,8 @@ impl AgentRuntime {
                        return Some(retry_output);
                    }
                    StreamStatus::Cancelled => {
-                        if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() {
-                            cancelled.remove(session_id);
-                        }
+                        let mut cancelled = self.app_state.cancelled_runs.lock().await;
+                        cancelled.remove(session_id);
                        return None;
                    }
                    StreamStatus::Error(_) => {
@@ -1237,9 +1229,8 @@ impl AgentRuntime {
                    return Some(retry_output);
                }
                StreamStatus::Cancelled => {
-                    if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() {
-                        cancelled.remove(session_id);
-                    }
+                    let mut cancelled = self.app_state.cancelled_runs.lock().await;
+                    cancelled.remove(session_id);
                    warn!("[AgentRuntime] 恢复期间被用户中止");
                    let _ = tx.send(AgentStreamEvent::Error {
                        message: "用户已手动中止执行。".to_string(),
@@ -1341,7 +1332,7 @@ impl AgentRuntime {
        if let Some(skills) = self
            .app_state
            .skill_registry
-            .read()
+            .try_read()
            .ok()
            .and_then(|r| r.build_reminder())
        {
@@ -102,7 +102,8 @@ impl ToolPartitioner {
 #[cfg(test)]
 mod tests {
    use std::path::PathBuf;
-    use std::sync::{Arc, RwLock};
+    use std::sync::Arc;
+    use tokio::sync::RwLock;

    use super::*;
    use crate::agent::skills::SkillRegistry;
@@ -1017,7 +1017,8 @@ mod tests {
    #[tokio::test]
    async fn test_create_new_session() {
        let db = setup_db_with_mode().await;
-        let llm = LlmClient::new("key".into(), "http://localhost".into(), "test-model".into());
+        let llm =
+            LlmClient::new("key".into(), "http://localhost".into(), "test-model".into()).unwrap();

        let info = create_or_resume_session(&db, None, &llm, "deep-research")
            .await
@@ -1040,7 +1041,8 @@ mod tests {
    #[tokio::test]
    async fn test_resume_existing_session() {
        let db = setup_db_with_mode().await;
-        let llm = LlmClient::new("key".into(), "http://localhost".into(), "test-model".into());
+        let llm =
+            LlmClient::new("key".into(), "http://localhost".into(), "test-model".into()).unwrap();

        // 先创建一个会话
        let created = create_or_resume_session(&db, None, &llm, "literature-reader")
@@ -1067,7 +1069,8 @@ mod tests {
    #[tokio::test]
    async fn test_resume_nonexistent_session_errors() {
        let db = setup_db_with_mode().await;
-        let llm = LlmClient::new("key".into(), "http://localhost".into(), "test-model".into());
+        let llm =
+            LlmClient::new("key".into(), "http://localhost".into(), "test-model".into()).unwrap();

        let result =
            create_or_resume_session(&db, Some("nonexistent-id".into()), &llm, "default").await;
@@ -43,7 +43,7 @@ pub async fn process_llm_stream(
    tx: &mpsc::UnboundedSender<AgentStreamEvent>,
    step: usize,
    session_id: &str,
-    cancelled_runs: Arc<std::sync::Mutex<std::collections::HashSet<String>>>,
+    cancelled_runs: Arc<tokio::sync::Mutex<std::collections::HashSet<String>>>,
    enable_thinking: bool,
 ) -> StreamOutput {
    // 1. 发起 LLM 流式调用
@@ -76,10 +76,9 @@ pub async fn process_llm_stream(
    let cancel_fut = async {
        loop {
            tokio::time::sleep(std::time::Duration::from_millis(250)).await;
-            if let Ok(cancelled) = cancelled_runs.lock() {
-                if cancelled.contains(&sid) {
-                    return;
-                }
+            let cancelled = cancelled_runs.lock().await;
+            if cancelled.contains(&sid) {
+                return;
            }
        }
    };
@@ -549,8 +549,6 @@ mod tests {
    /// 构建测试用的 ToolContext。
    async fn make_test_tool_context() -> ToolContext {
        use crate::agent::memory::MemoryManager;
-        use crate::agent::runtime::file_cache::FileStateCache;
-        use crate::agent::runtime::permission::PermissionChecker;
        use crate::agent::skills::SkillRegistry;
        use crate::api::AppState;
        use crate::clients::ads::AdsClient;
@@ -563,16 +561,17 @@ mod tests {
        use crate::services::translation::Dictionary;
        use crate::Config;
        use std::path::PathBuf;
-        use std::sync::{Arc, Mutex, RwLock};
+        use std::sync::Arc;

        let pool = sqlx::SqlitePool::connect("sqlite::memory:").await.unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();

        let config = Config::from_env();
-        let llm = LlmClient::new("tk".into(), "http://localhost".into(), "m".into());
-        let embedding = EmbeddingClient::new("tk".into(), "http://localhost".into(), "e".into());
-        let ads = AdsClient::new("tk".into());
-        let arxiv = ArxivClient::new();
+        let llm = LlmClient::new("tk".into(), "http://localhost".into(), "m".into()).unwrap();
+        let embedding =
+            EmbeddingClient::new("tk".into(), "http://localhost".into(), "e".into()).unwrap();
+        let ads = AdsClient::new("tk".into()).unwrap();
+        let arxiv = ArxivClient::new().unwrap();
        let qiniu = QiniuClient::new(
            "ak".into(),
            "sk".into(),
@@ -593,20 +592,27 @@ mod tests {
            vision_llm: None,
            embedding,
            downloader: Downloader::new().expect("downloader"),
+            http_client: reqwest::Client::new(),
            harvest_status: Arc::new(tokio::sync::Mutex::new(MetaSyncStatus::default())),
            batch_status: Arc::new(tokio::sync::Mutex::new(AssetBatchStatus::default())),
            active_bibcode: Arc::new(tokio::sync::Mutex::new(None)),
-            cancelled_runs: Arc::new(Mutex::new(std::collections::HashSet::new())),
-            skill_registry: Arc::new(RwLock::new(SkillRegistry::new(PathBuf::from("/tmp/sk")))),
-            pending_questions: Arc::new(Mutex::new(std::collections::HashMap::new())),
-            pending_permissions: Arc::new(Mutex::new(std::collections::HashMap::new())),
-            session_permission_checkers: Arc::new(RwLock::new(std::collections::HashMap::new())),
+            cancelled_runs: Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())),
+            skill_registry: Arc::new(tokio::sync::RwLock::new(SkillRegistry::new(PathBuf::from(
+                "/tmp/sk",
+            )))),
+            pending_questions: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
+            pending_permissions: Arc::new(
+                tokio::sync::Mutex::new(std::collections::HashMap::new()),
+            ),
+            session_permission_checkers: Arc::new(tokio::sync::RwLock::new(
+                std::collections::HashMap::new(),
+            )),
            sse_broadcast: None,
            memory_manager: Arc::new(tokio::sync::Mutex::new(MemoryManager::new(PathBuf::from(
                "/tmp/test_mem",
            )))),
-            sessions: Arc::new(Mutex::new(std::collections::HashMap::new())),
-            login_rate_limiter: Arc::new(Mutex::new(std::collections::HashMap::new())),
+            sessions: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
+            login_rate_limiter: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
        });

        ToolContext::new(app_state)
@@ -614,7 +620,7 @@ mod tests {

    /// 创建包含指定 Mock 工具的 StreamingToolExecutor。
    async fn make_executor(tools: Vec<MockAgentTool>) -> StreamingToolExecutor {
-        let skill_registry = Arc::new(std::sync::RwLock::new(
+        let skill_registry = Arc::new(tokio::sync::RwLock::new(
            crate::agent::skills::SkillRegistry::new(std::path::PathBuf::from("/tmp/sk")),
        ));
        let mut registry = ToolRegistry::new(skill_registry);